142 lines
4.9 KiB
Python
142 lines
4.9 KiB
Python
"""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()
|