multi agent

tests/test_agents_e2e.py

@@ -0,0 +1,141 @@
+"""End-to-end tests for multi-agent collaboration (task 8.1)."""
+
+from __future__ import annotations
+
+from unittest.mock import MagicMock, patch
+
+from agents.collab_runner import CollaborativeTaskRunner, CollaborativeTaskRunnerConfig
+from agents.config import CollaborationConfig
+from agents.models import Observation, ReflectionAction, ReflectionOutcome, VerificationVerdict
+from agents.observer import Observer
+from agents.reflector import Reflector
+from agents.verifier import Verifier
+from core.models import Bounds, Scene, SceneElement, Task
+from runtime.executor import StepResult
+from runtime.planner import PlannedStep
+
+
+def _scene() -> Scene:
+    return Scene(
+        width=1080,
+        height=1920,
+        elements=[
+            SceneElement(id="btn1", type="button", bounds=Bounds(10, 20, 100, 50), text="Send"),
+        ],
+    )
+
+
+def _task() -> Task:
+    return Task(goal="Send a message", device_id="dev1")
+
+
+def test_full_loop_verification_succeeds() -> None:
+    """Full collaborative loop: plan → execute → verify (achieved) → complete."""
+    planner = MagicMock()
+    step = PlannedStep(action="tap", description="Tap send", args={"element_id": "btn1"})
+    planner.plan.return_value = [step]
+    planner.goal_reached.side_effect = [False, True]
+
+    executor = MagicMock()
+    result = StepResult(step=step, success=True, attempts=1, result={"tapped": True})
+    executor.execute.return_value = result
+
+    runner = CollaborativeTaskRunner(
+        planner=planner,
+        executor=executor,
+        observer=Observer(),
+        verifier=Verifier(),
+        reflector=Reflector(),
+        config=CollaborativeTaskRunnerConfig(max_steps=5),
+        collaboration_config=CollaborationConfig(enabled=True, max_recovery_attempts=3),
+    )
+
+    scene = _scene()
+    with patch("tools.describe_screen.describe_screen", return_value=scene):
+        task = runner.run(_task())
+
+    assert task.status == "completed"
+
+
+def test_full_loop_with_reflection_recovery() -> None:
+    """Full loop: plan → execute → verify (not achieved) → reflect (recovery) → execute recovery → verify (achieved)."""
+    planner = MagicMock()
+    step = PlannedStep(action="tap", description="Tap send", args={"element_id": "btn1"})
+    planner.plan.return_value = [step]
+    planner.goal_reached.side_effect = [False, False, True]
+
+    executor = MagicMock()
+    success = StepResult(step=step, success=True, attempts=1, result={"tapped": True})
+    executor.execute.return_value = success
+
+    # First verify: not achieved; second verify: achieved
+    verify_call_count = 0
+
+    def verifier_verify(**kwargs):
+        nonlocal verify_call_count
+        verify_call_count += 1
+        if verify_call_count == 1:
+            return VerificationVerdict(result="not_achieved", reasoning="No change.")
+        return VerificationVerdict(result="achieved", reasoning="Done.")
+
+    verifier = MagicMock()
+    verifier.verify.side_effect = verifier_verify
+
+    reflector = MagicMock()
+    reflector.reflect.return_value = ReflectionOutcome(
+        replan=False,
+        action=ReflectionAction(action="swipe", description="Scroll", args={"direction": "up"}),
+        reasoning="Try scrolling.",
+    )
+
+    runner = CollaborativeTaskRunner(
+        planner=planner,
+        executor=executor,
+        observer=Observer(),
+        verifier=verifier,
+        reflector=reflector,
+        config=CollaborativeTaskRunnerConfig(max_steps=5),
+        collaboration_config=CollaborationConfig(enabled=True, max_recovery_attempts=3),
+    )
+
+    scene = _scene()
+    with patch("tools.describe_screen.describe_screen", return_value=scene):
+        task = runner.run(_task())
+
+    assert task.status == "completed"
+    reflector.reflect.assert_called_once()
+
+
+def test_full_loop_exhausts_ceiling() -> None:
+    """Full loop fails when recovery ceiling is exhausted."""
+    planner = MagicMock()
+    step = PlannedStep(action="tap", description="Tap send", args={"element_id": "btn1"})
+    planner.plan.return_value = [step]
+    planner.goal_reached.return_value = False
+
+    executor = MagicMock()
+    success = StepResult(step=step, success=True, attempts=1, result={"tapped": True})
+    executor.execute.return_value = success
+
+    verifier = MagicMock()
+    verifier.verify.return_value = VerificationVerdict(result="not_achieved", reasoning="No change.")
+
+    reflector = MagicMock()
+    reflector.reflect.return_value = ReflectionOutcome(replan=True, reasoning="Replan.")
+
+    runner = CollaborativeTaskRunner(
+        planner=planner,
+        executor=executor,
+        observer=Observer(),
+        verifier=verifier,
+        reflector=reflector,
+        config=CollaborativeTaskRunnerConfig(max_steps=10, max_recovery_attempts=2),
+        collaboration_config=CollaborationConfig(enabled=True, max_recovery_attempts=2),
+    )
+
+    scene = _scene()
+    with patch("tools.describe_screen.describe_screen", return_value=scene):
+        task = runner.run(_task())
+
+    assert task.status == "failed"
+    assert "ceiling" in (task.failure_reason or "").lower()