From a453d9e6ba5b764e42784aaf6ffc72c295cf1f04 Mon Sep 17 00:00:00 2001 From: Jerry Yan <792602257@qq.com> Date: Mon, 6 Jul 2026 23:23:44 +0800 Subject: [PATCH] multi agent --- agents/__init__.py | 1 + agents/collab_runner.py | 184 +++++++++++++++++ agents/config.py | 41 ++++ agents/models.py | 108 ++++++++++ agents/observer.py | 55 +++++ agents/reflector.py | 146 +++++++++++++ agents/verifier.py | 146 +++++++++++++ openspec/changes/multi-agent-runtime/tasks.md | 50 ++--- pyproject.toml | 1 + tests/test_agents_collab_runner.py | 191 ++++++++++++++++++ tests/test_agents_e2e.py | 141 +++++++++++++ tests/test_agents_integration.py | 71 +++++++ tests/test_agents_models.py | 77 +++++++ tests/test_agents_no_coupling.py | 36 ++++ tests/test_agents_observer.py | 85 ++++++++ tests/test_agents_reflector.py | 136 +++++++++++++ tests/test_agents_verifier.py | 137 +++++++++++++ tests/test_smoke.py | 1 + 18 files changed, 1582 insertions(+), 25 deletions(-) create mode 100644 agents/__init__.py create mode 100644 agents/collab_runner.py create mode 100644 agents/config.py create mode 100644 agents/models.py create mode 100644 agents/observer.py create mode 100644 agents/reflector.py create mode 100644 agents/verifier.py create mode 100644 tests/test_agents_collab_runner.py create mode 100644 tests/test_agents_e2e.py create mode 100644 tests/test_agents_integration.py create mode 100644 tests/test_agents_models.py create mode 100644 tests/test_agents_no_coupling.py create mode 100644 tests/test_agents_observer.py create mode 100644 tests/test_agents_reflector.py create mode 100644 tests/test_agents_verifier.py diff --git a/agents/__init__.py b/agents/__init__.py new file mode 100644 index 0000000..9d48db4 --- /dev/null +++ b/agents/__init__.py @@ -0,0 +1 @@ +from __future__ import annotations diff --git a/agents/collab_runner.py b/agents/collab_runner.py new file mode 100644 index 0000000..b23f057 --- /dev/null +++ b/agents/collab_runner.py @@ -0,0 +1,184 @@ +from __future__ import annotations + +import logging +from dataclasses import dataclass, replace + +from agents.config import CollaborationConfig, load_config +from agents.models import Observation, VerificationVerdict +from agents.observer import Observer +from agents.reflector import Reflector +from agents.verifier import Verifier +from core.models import Scene, Task, utc_now +from runtime.context import TaskContext +from runtime.executor import Executor +from runtime.planner import PlannedStep, Planner +from runtime.task import TaskRunner, TaskRunnerConfig +from semantic.llm_client import AnthropicSemanticClient + +logger = logging.getLogger(__name__) + + +@dataclass +class CollaborativeTaskRunnerConfig: + max_steps: int = 20 + max_recovery_attempts: int = 3 + + +class CollaborativeTaskRunner: + def __init__( + self, + *, + planner: Planner | None = None, + executor: Executor | None = None, + task_runner: TaskRunner | None = None, + observer: Observer | None = None, + verifier: Verifier | None = None, + reflector: Reflector | None = None, + config: CollaborativeTaskRunnerConfig | None = None, + collaboration_config: CollaborationConfig | None = None, + llm_client: AnthropicSemanticClient | None = None, + ) -> None: + self.planner = planner or Planner() + self.executor = executor or Executor() + self.task_runner = task_runner + self.observer = observer or Observer() + self.verifier = verifier or Verifier(client=llm_client) + self.reflector = reflector or Reflector(client=llm_client) + self.config = config or CollaborativeTaskRunnerConfig() + self.collaboration_config = collaboration_config or load_config() + + def run(self, task: Task) -> Task: + if not self.collaboration_config.enabled: + return self._run_plain(task) + return self._run_collaborative(task) + + def _run_plain(self, task: Task) -> Task: + runner = self.task_runner or TaskRunner( + planner=self.planner, + executor=self.executor, + config=TaskRunnerConfig(max_steps=self.config.max_steps), + ) + return runner.run(task) + + def _run_collaborative(self, task: Task) -> Task: + context = TaskContext(task_id=task.id, goal=task.goal) + task.status = "running" # type: ignore[assignment] + task.updated_at = utc_now() + recovery_attempts = 0 + + for _ in range(self.config.max_steps): + scene = self._observe_scene(task.device_id) + context.add_scene(scene) + + pre_observation = self.observer.observe( + scene=scene, + world=context.world, + ) + + steps = self.planner.plan( + goal=task.goal, + scene=scene, + context=context, + ) + if not steps or self.planner.goal_reached( + goal=task.goal, + scene=scene, + context=context, + ): + task.status = "completed" # type: ignore[assignment] + task.updated_at = utc_now() + task.completed_at = utc_now() + return task + + for step in steps: + executable_step = self._step_for_device(step, task.device_id) + result = self.executor.execute(executable_step, context=context) + context.add_step_result(result) + + post_scene = self._observe_scene(task.device_id) + post_observation = self.observer.observe( + scene=post_scene, + world=context.world, + ) + + verdict = self.verifier.verify( + pre_observation=pre_observation, + post_observation=post_observation, + planned_step=step, + step_result=result, + ) + + if verdict.result == "achieved": + continue + + if recovery_attempts >= self.collaboration_config.max_recovery_attempts: + task.status = "failed" # type: ignore[assignment] + task.updated_at = utc_now() + task.completed_at = utc_now() + task.failure_reason = ( + f"Reflection recovery ceiling reached " + f"({self.collaboration_config.max_recovery_attempts} attempts)" + ) + return task + + outcome = self.reflector.reflect( + observation=post_observation, + planned_step=step, + step_result=result, + verdict=verdict, + ) + recovery_attempts += 1 + + if outcome.replan: + break + + if outcome.action is not None: + recovery_step = PlannedStep( + action=outcome.action.action, + description=outcome.action.description, + args=outcome.action.args, + ) + recovery_result = self.executor.execute( + self._step_for_device(recovery_step, task.device_id), + context=context, + ) + context.add_step_result(recovery_result) + if not recovery_result.success: + task.status = "failed" # type: ignore[assignment] + task.updated_at = utc_now() + task.completed_at = utc_now() + task.failure_reason = ( + f"Recovery action failed: {recovery_result.error}" + ) + return task + + task.status = "failed" # type: ignore[assignment] + task.updated_at = utc_now() + task.completed_at = utc_now() + task.failure_reason = f"max steps exceeded: {self.config.max_steps}" + return task + + def _observe_scene(self, device_id: str) -> Scene: + from tools.describe_screen import describe_screen + + return describe_screen(device_id) + + def _step_for_device(self, step: PlannedStep, device_id: str) -> PlannedStep: + device_scoped_actions = { + "take_screenshot", + "screenshot", + "tap", + "swipe", + "input_text", + "launch_app", + "terminate_app", + "get_ui_tree", + "ui_tree", + "describe_screen", + "describe_screen_semantic", + "find_text_on_screen", + "find_icon_on_screen", + } + if step.action not in device_scoped_actions or "device_id" in step.args: + return step + return replace(step, args={**step.args, "device_id": device_id}) diff --git a/agents/config.py b/agents/config.py new file mode 100644 index 0000000..e89749d --- /dev/null +++ b/agents/config.py @@ -0,0 +1,41 @@ +from __future__ import annotations + +import os +from collections.abc import Mapping +from dataclasses import dataclass + +ENABLED_ENV = "MULTI_AGENT_COLLABORATION_ENABLED" +MAX_RECOVERY_ENV = "MULTI_AGENT_MAX_RECOVERY_ATTEMPTS" +DEFAULT_MAX_RECOVERY_ATTEMPTS = 3 + + +@dataclass(frozen=True) +class CollaborationConfig: + enabled: bool = False + max_recovery_attempts: int = DEFAULT_MAX_RECOVERY_ATTEMPTS + + +def load_config(env: Mapping[str, str] | None = None) -> CollaborationConfig: + values = env or os.environ + return CollaborationConfig( + enabled=_parse_bool(values.get(ENABLED_ENV), default=False), + max_recovery_attempts=_parse_int( + values.get(MAX_RECOVERY_ENV), default=DEFAULT_MAX_RECOVERY_ATTEMPTS + ), + ) + + +def _parse_bool(value: str | None, *, default: bool) -> bool: + if value is None: + return default + return value.strip().lower() in {"1", "true", "yes", "on", "enabled"} + + +def _parse_int(value: str | None, *, default: int) -> int: + if value is None: + return default + try: + result = int(value) + except ValueError: + return default + return result if result > 0 else default diff --git a/agents/models.py b/agents/models.py new file mode 100644 index 0000000..b787b63 --- /dev/null +++ b/agents/models.py @@ -0,0 +1,108 @@ +from __future__ import annotations + +from dataclasses import dataclass, field +from typing import Any, Literal + + +@dataclass +class Observation: + scene_summary: str + semantic_page: str | None = None + semantic_intents: list[str] = field(default_factory=list) + world_app: str | None = None + world_page: str | None = None + world_variables: dict[str, Any] = field(default_factory=dict) + raw_scene: dict[str, Any] | None = None + + def to_dict(self) -> dict[str, Any]: + return { + "scene_summary": self.scene_summary, + "semantic_page": self.semantic_page, + "semantic_intents": list(self.semantic_intents), + "world_app": self.world_app, + "world_page": self.world_page, + "world_variables": dict(self.world_variables), + "raw_scene": self.raw_scene, + } + + @classmethod + def from_dict(cls, data: dict[str, Any]) -> Observation: + return cls( + scene_summary=str(data["scene_summary"]), + semantic_page=data.get("semantic_page"), + semantic_intents=list(data.get("semantic_intents", [])), + world_app=data.get("world_app"), + world_page=data.get("world_page"), + world_variables=dict(data.get("world_variables", {})), + raw_scene=data.get("raw_scene"), + ) + + +VerdictResult = Literal["achieved", "not_achieved"] + + +@dataclass +class VerificationVerdict: + result: VerdictResult + reasoning: str + + def to_dict(self) -> dict[str, Any]: + return { + "result": self.result, + "reasoning": self.reasoning, + } + + @classmethod + def from_dict(cls, data: dict[str, Any]) -> VerificationVerdict: + result = data["result"] + if result not in ("achieved", "not_achieved"): + raise ValueError(f"invalid verdict result: {result}") + return cls( + result=result, + reasoning=str(data["reasoning"]), + ) + + +@dataclass +class ReflectionAction: + action: str + description: str + args: dict[str, Any] = field(default_factory=dict) + + def to_dict(self) -> dict[str, Any]: + return { + "action": self.action, + "description": self.description, + "args": dict(self.args), + } + + @classmethod + def from_dict(cls, data: dict[str, Any]) -> ReflectionAction: + return cls( + action=str(data["action"]), + description=str(data["description"]), + args=dict(data.get("args", {})), + ) + + +@dataclass +class ReflectionOutcome: + replan: bool = False + action: ReflectionAction | None = None + reasoning: str = "" + + def to_dict(self) -> dict[str, Any]: + return { + "replan": self.replan, + "action": self.action.to_dict() if self.action else None, + "reasoning": self.reasoning, + } + + @classmethod + def from_dict(cls, data: dict[str, Any]) -> ReflectionOutcome: + action_data = data.get("action") + return cls( + replan=bool(data.get("replan", False)), + action=ReflectionAction.from_dict(action_data) if action_data else None, + reasoning=str(data.get("reasoning", "")), + ) diff --git a/agents/observer.py b/agents/observer.py new file mode 100644 index 0000000..1f9da95 --- /dev/null +++ b/agents/observer.py @@ -0,0 +1,55 @@ +from __future__ import annotations + +from typing import TYPE_CHECKING, Any + +from agents.models import Observation +from core.models import Scene, Step +from runtime.executor import StepResult +from runtime.planner import PlannedStep + +if TYPE_CHECKING: + from semantic.models import SemanticScene + from world.models import WorldState + + +class Observer: + def observe( + self, + *, + scene: Scene, + step: PlannedStep | None = None, + result: StepResult | None = None, + semantic_scene: SemanticScene | None = None, + world: WorldState | None = None, + ) -> Observation: + scene_summary = self._build_scene_summary(scene) + semantic_page = None + semantic_intents: list[str] = [] + if semantic_scene is not None: + semantic_page = semantic_scene.page + semantic_intents = list(semantic_scene.intents) + + world_app = None + world_page = None + world_variables: dict[str, Any] = {} + if world is not None: + world_app = world.current_app + world_page = world.current_page + world_variables = dict(world.variables) + + return Observation( + scene_summary=scene_summary, + semantic_page=semantic_page, + semantic_intents=semantic_intents, + world_app=world_app, + world_page=world_page, + world_variables=world_variables, + raw_scene=scene.to_dict(), + ) + + def _build_scene_summary(self, scene: Scene) -> str: + parts = [f"Screen {scene.width}x{scene.height}"] + for element in scene.elements[:10]: + text = element.text or element.type + parts.append(f" [{element.type}] {text}") + return "\n".join(parts) diff --git a/agents/reflector.py b/agents/reflector.py new file mode 100644 index 0000000..90ba1ca --- /dev/null +++ b/agents/reflector.py @@ -0,0 +1,146 @@ +from __future__ import annotations + +import json +from typing import Any + +from agents.models import Observation, ReflectionAction, ReflectionOutcome, VerificationVerdict +from runtime.executor import StepResult +from runtime.planner import PlannedStep +from semantic.llm_client import AnthropicSemanticClient, EnrichmentUnavailable + +REFLECTOR_SYSTEM_PROMPT = """You are a reflection agent. A step failed to achieve its intended effect despite mechanical success. Your job is to propose either: +1. A distinct recovery action (different from the failed step) that might fix the issue +2. A replan request if no bounded corrective action is identifiable + +You must NEVER simply re-issue the identical failed step. + +Respond with JSON: {"replan": bool, "action": null or {"action": "...", "description": "...", "args": {...}}, "reasoning": "explanation"}""" + +REFLECTOR_JSON_SCHEMA: dict[str, Any] = { + "type": "object", + "additionalProperties": False, + "required": ["replan", "reasoning"], + "properties": { + "replan": {"type": "boolean"}, + "action": { + "type": ["object", "null"], + "properties": { + "action": {"type": "string"}, + "description": {"type": "string"}, + "args": {"type": "object"}, + }, + }, + "reasoning": {"type": "string", "minLength": 1}, + }, +} + + +class Reflector: + def __init__(self, *, client: AnthropicSemanticClient | None = None) -> None: + self._client = client + + def reflect( + self, + *, + observation: Observation, + planned_step: PlannedStep, + step_result: StepResult, + verdict: VerificationVerdict, + ) -> ReflectionOutcome: + if self._client is None: + return self._fallback_reflect( + observation=observation, + planned_step=planned_step, + step_result=step_result, + verdict=verdict, + ) + + try: + return self._llm_reflect( + observation=observation, + planned_step=planned_step, + step_result=step_result, + verdict=verdict, + ) + except EnrichmentUnavailable: + return self._fallback_reflect( + observation=observation, + planned_step=planned_step, + step_result=step_result, + verdict=verdict, + ) + + def _llm_reflect( + self, + *, + observation: Observation, + planned_step: PlannedStep, + step_result: StepResult, + verdict: VerificationVerdict, + ) -> ReflectionOutcome: + user_prompt = json.dumps( + { + "observation": observation.to_dict(), + "planned_step": { + "action": planned_step.action, + "description": planned_step.description, + "args": planned_step.args, + "expected_text": planned_step.expected_text, + }, + "step_result": step_result.to_dict(), + "verdict": verdict.to_dict(), + }, + ensure_ascii=False, + ) + response = self._client._create_message( # type: ignore[union-attr] + {"prompt": user_prompt, "system": REFLECTOR_SYSTEM_PROMPT}, + timeout=10.0, + ) + payload = _extract_outcome(response) + outcome = ReflectionOutcome.from_dict(payload) + if not outcome.replan and outcome.action is not None: + if outcome.action.action == planned_step.action and outcome.action.args == planned_step.args: + return ReflectionOutcome( + replan=True, + reasoning="Reflector proposed identical action; requesting replan instead.", + ) + return outcome + + def _fallback_reflect( + self, + *, + observation: Observation, + planned_step: PlannedStep, + step_result: StepResult, + verdict: VerificationVerdict, + ) -> ReflectionOutcome: + return ReflectionOutcome( + replan=True, + reasoning=f"Step '{planned_step.description}' failed semantically; replanning (fallback).", + ) + + +def _extract_outcome(response: Any) -> dict[str, Any]: + if isinstance(response, dict) and "replan" in response: + return response + content = getattr(response, "content", None) + if isinstance(content, list): + for block in content: + text = getattr(block, "text", None) + if isinstance(text, str): + try: + decoded = json.loads(text) + if isinstance(decoded, dict) and "replan" in decoded: + return decoded + except json.JSONDecodeError: + continue + if isinstance(block, dict) and "replan" in block: + return block + if isinstance(content, str): + try: + decoded = json.loads(content) + if isinstance(decoded, dict) and "replan" in decoded: + return decoded + except json.JSONDecodeError: + pass + raise ValueError("could not extract reflection outcome from LLM response") diff --git a/agents/verifier.py b/agents/verifier.py new file mode 100644 index 0000000..f7ef5c7 --- /dev/null +++ b/agents/verifier.py @@ -0,0 +1,146 @@ +from __future__ import annotations + +import json +from typing import Any + +from agents.models import Observation, VerificationVerdict +from runtime.executor import StepResult +from runtime.planner import PlannedStep +from semantic.llm_client import AnthropicSemanticClient, EnrichmentUnavailable + +VERIFIER_SYSTEM_PROMPT = """You are a verification agent. Your job is to determine whether a planned step achieved its intended effect. + +You will receive: +1. A pre-step observation of the device state +2. A post-step observation of the device state +3. The planned step (action, description, expected outcome) +4. The step result (whether the tool call succeeded mechanically) + +Analyze whether the intended effect of the step actually occurred by comparing the pre and post observations against the step's stated intent. + +Respond with JSON: {"result": "achieved" or "not_achieved", "reasoning": "explanation"}""" + +VERIFIER_JSON_SCHEMA: dict[str, Any] = { + "type": "object", + "additionalProperties": False, + "required": ["result", "reasoning"], + "properties": { + "result": {"type": "string", "enum": ["achieved", "not_achieved"]}, + "reasoning": {"type": "string", "minLength": 1}, + }, +} + + +class Verifier: + def __init__(self, *, client: AnthropicSemanticClient | None = None) -> None: + self._client = client + + def verify( + self, + *, + pre_observation: Observation, + post_observation: Observation, + planned_step: PlannedStep, + step_result: StepResult, + ) -> VerificationVerdict: + if not step_result.success: + return VerificationVerdict( + result="not_achieved", + reasoning=f"Step failed mechanically: {step_result.error}", + ) + + if self._client is None: + return self._fallback_verify( + pre_observation=pre_observation, + post_observation=post_observation, + planned_step=planned_step, + step_result=step_result, + ) + + try: + return self._llm_verify( + pre_observation=pre_observation, + post_observation=post_observation, + planned_step=planned_step, + step_result=step_result, + ) + except EnrichmentUnavailable: + return self._fallback_verify( + pre_observation=pre_observation, + post_observation=post_observation, + planned_step=planned_step, + step_result=step_result, + ) + + def _llm_verify( + self, + *, + pre_observation: Observation, + post_observation: Observation, + planned_step: PlannedStep, + step_result: StepResult, + ) -> VerificationVerdict: + user_prompt = json.dumps( + { + "pre_observation": pre_observation.to_dict(), + "post_observation": post_observation.to_dict(), + "planned_step": { + "action": planned_step.action, + "description": planned_step.description, + "args": planned_step.args, + "expected_text": planned_step.expected_text, + }, + "step_result": step_result.to_dict(), + }, + ensure_ascii=False, + ) + response = self._client._create_message( # type: ignore[union-attr] + {"prompt": user_prompt, "system": VERIFIER_SYSTEM_PROMPT}, + timeout=10.0, + ) + payload = _extract_verdict(response) + return VerificationVerdict.from_dict(payload) + + def _fallback_verify( + self, + *, + pre_observation: Observation, + post_observation: Observation, + planned_step: PlannedStep, + step_result: StepResult, + ) -> VerificationVerdict: + if pre_observation.scene_summary == post_observation.scene_summary: + return VerificationVerdict( + result="not_achieved", + reasoning="Scene unchanged after step execution (fallback heuristic).", + ) + return VerificationVerdict( + result="achieved", + reasoning="Scene changed after step execution (fallback heuristic).", + ) + + +def _extract_verdict(response: Any) -> dict[str, Any]: + if isinstance(response, dict) and "result" in response: + return response + content = getattr(response, "content", None) + if isinstance(content, list): + for block in content: + text = getattr(block, "text", None) + if isinstance(text, str): + try: + decoded = json.loads(text) + if isinstance(decoded, dict) and "result" in decoded: + return decoded + except json.JSONDecodeError: + continue + if isinstance(block, dict) and "result" in block: + return block + if isinstance(content, str): + try: + decoded = json.loads(content) + if isinstance(decoded, dict) and "result" in decoded: + return decoded + except json.JSONDecodeError: + pass + raise ValueError("could not extract verification verdict from LLM response") diff --git a/openspec/changes/multi-agent-runtime/tasks.md b/openspec/changes/multi-agent-runtime/tasks.md index 4a6df4f..d10f051 100644 --- a/openspec/changes/multi-agent-runtime/tasks.md +++ b/openspec/changes/multi-agent-runtime/tasks.md @@ -1,48 +1,48 @@ ## 1. Package scaffolding -- [ ] 1.1 Create the `agents/` package (`__init__.py`, `models.py`, `observer.py`, `verifier.py`, `reflector.py`, `collab_runner.py`, `config.py`) -- [ ] 1.2 Add an `agents*` entry to `[tool.setuptools.packages.find].include` in `pyproject.toml` (no new third-party dependency) -- [ ] 1.3 Add collaboration configuration: enabled/disabled flag (default disabled) and max-reflection-recovery-attempts ceiling, sourced from a single place `agents/config.py` reads from -- [ ] 1.4 Extend the project's smoke test (that imports every package) to import `agents` +- [x] 1.1 Create the `agents/` package (`__init__.py`, `models.py`, `observer.py`, `verifier.py`, `reflector.py`, `collab_runner.py`, `config.py`) +- [x] 1.2 Add an `agents*` entry to `[tool.setuptools.packages.find].include` in `pyproject.toml` (no new third-party dependency) +- [x] 1.3 Add collaboration configuration: enabled/disabled flag (default disabled) and max-reflection-recovery-attempts ceiling, sourced from a single place `agents/config.py` reads from +- [x] 1.4 Extend the project's smoke test (that imports every package) to import `agents` ## 2. Handoff protocol data models (capability: multi-agent-collaboration) -- [ ] 2.1 Implement `agents/models.py`: `Observation`, `VerificationVerdict`, `ReflectionOutcome`, and `ReflectionAction` dataclasses with `to_dict()`/`from_dict()`, mirroring the style of `core/models.py` -- [ ] 2.2 Write unit tests for each dataclass round-tripping through `to_dict()`/`from_dict()` +- [x] 2.1 Implement `agents/models.py`: `Observation`, `VerificationVerdict`, `ReflectionOutcome`, and `ReflectionAction` dataclasses with `to_dict()`/`from_dict()`, mirroring the style of `core/models.py` +- [x] 2.2 Write unit tests for each dataclass round-tripping through `to_dict()`/`from_dict()` ## 3. Observer role (capability: multi-agent-collaboration) -- [ ] 3.1 Implement `agents/observer.py`: `Observer.observe(...) -> Observation`, perceiving current device state via `SemanticScene`/`WorldState` when available -- [ ] 3.2 Make `Observer.observe(...)` work when `SemanticScene` and/or `WorldState` are `None`, falling back to the raw `Scene`/`PlannedStep`/`StepResult` -- [ ] 3.3 Write unit tests for `Observer.observe(...)` covering: both `SemanticScene`/`WorldState` present, both absent, and each present individually +- [x] 3.1 Implement `agents/observer.py`: `Observer.observe(...) -> Observation`, perceiving current device state via `SemanticScene`/`WorldState` when available +- [x] 3.2 Make `Observer.observe(...)` work when `SemanticScene` and/or `WorldState` are `None`, falling back to the raw `Scene`/`PlannedStep`/`StepResult` +- [x] 3.3 Write unit tests for `Observer.observe(...)` covering: both `SemanticScene`/`WorldState` present, both absent, and each present individually ## 4. Verifier role (capability: multi-agent-collaboration) -- [ ] 4.1 Implement `agents/verifier.py`: `Verifier.verify(pre_observation, post_observation, planned_step, step_result) -> VerificationVerdict`, comparing pre-step and post-step `Observation`s against the `PlannedStep`'s stated intent -- [ ] 4.2 Wire `Verifier` to reuse `semantic/llm_client.py`'s existing LLM client abstraction (injectable/mockable) rather than a new client -- [ ] 4.3 Write unit tests for `Verifier.verify(...)` against a fake client covering: verified-achieved, verified-not-achieved, and client-failure degrade cases +- [x] 4.1 Implement `agents/verifier.py`: `Verifier.verify(pre_observation, post_observation, planned_step, step_result) -> VerificationVerdict`, comparing pre-step and post-step `Observation`s against the `PlannedStep`'s stated intent +- [x] 4.2 Wire `Verifier` to reuse `semantic/llm_client.py`'s existing LLM client abstraction (injectable/mockable) rather than a new client +- [x] 4.3 Write unit tests for `Verifier.verify(...)` against a fake client covering: verified-achieved, verified-not-achieved, and client-failure degrade cases ## 5. Reflector role (capability: multi-agent-collaboration) -- [ ] 5.1 Implement `agents/reflector.py`: `Reflector.reflect(observation, planned_step, step_result, verdict) -> ReflectionOutcome`, invoked only on a Verifier-flagged not-achieved verdict -- [ ] 5.2 Ensure `Reflector.reflect(...)` never returns an outcome that simply re-issues the identical failed `PlannedStep`; it returns either a distinct `ReflectionAction` or a replan request -- [ ] 5.3 Write unit tests for `Reflector.reflect(...)` covering: recovery-action outcome, replan-request outcome, and client-failure degrade case +- [x] 5.1 Implement `agents/reflector.py`: `Reflector.reflect(observation, planned_step, step_result, verdict) -> ReflectionOutcome`, invoked only on a Verifier-flagged not-achieved verdict +- [x] 5.2 Ensure `Reflector.reflect(...)` never returns an outcome that simply re-issues the identical failed `PlannedStep`; it returns either a distinct `ReflectionAction` or a replan request +- [x] 5.3 Write unit tests for `Reflector.reflect(...)` covering: recovery-action outcome, replan-request outcome, and client-failure degrade case ## 6. CollaborativeTaskRunner and handoff loop (capability: multi-agent-collaboration) -- [ ] 6.1 Implement `agents/collab_runner.py`: `CollaborativeTaskRunner`, composing the existing `Planner`, `Executor`, and `runtime/task.py`'s `TaskRunner` strictly by import -- [ ] 6.2 Implement the handoff loop: Observer → Planner → Executor → Verifier → (Reflector only on a Verifier-flagged failure) → back to Observer -- [ ] 6.3 Implement the reflection-recovery ceiling: a configurable max-attempts counter, independent of and layered on top of `Executor.max_retries`, that stops further reflection-driven recovery once exhausted and surfaces the failure -- [ ] 6.4 Write unit tests for `CollaborativeTaskRunner` covering: a task that completes without any Verifier-flagged failure, a task recovered via one Reflector-proposed action, and a task that exhausts the reflection-recovery ceiling +- [x] 6.1 Implement `agents/collab_runner.py`: `CollaborativeTaskRunner`, composing the existing `Planner`, `Executor`, and `runtime/task.py`'s `TaskRunner` strictly by import +- [x] 6.2 Implement the handoff loop: Observer → Planner → Executor → Verifier → (Reflector only on a Verifier-flagged failure) → back to Observer +- [x] 6.3 Implement the reflection-recovery ceiling: a configurable max-attempts counter, independent of and layered on top of `Executor.max_retries`, that stops further reflection-driven recovery once exhausted and surfaces the failure +- [x] 6.4 Write unit tests for `CollaborativeTaskRunner` covering: a task that completes without any Verifier-flagged failure, a task recovered via one Reflector-proposed action, and a task that exhausts the reflection-recovery ceiling ## 7. Config wiring and opt-in behavior (capability: multi-agent-collaboration) -- [ ] 7.1 Confirm collaboration is disabled by default: constructing/running a task without explicitly enabling it behaves exactly as plain `TaskRunner.run()` today -- [ ] 7.2 Write a unit test asserting `runtime/planner.py`'s `Planner`, `runtime/executor.py`'s `Executor`, and `runtime/task.py`'s `TaskRunner` are unmodified/unaffected by this change (no accidental coupling introduced from `agents/`) +- [x] 7.1 Confirm collaboration is disabled by default: constructing/running a task without explicitly enabling it behaves exactly as plain `TaskRunner.run()` today +- [x] 7.2 Write a unit test asserting `runtime/planner.py`'s `Planner`, `runtime/executor.py`'s `Executor`, and `runtime/task.py`'s `TaskRunner` are unmodified/unaffected by this change (no accidental coupling introduced from `agents/`) ## 8. End-to-end validation -- [ ] 8.1 Write an end-to-end test simulating a full collaborative task run against a mocked `Driver`/`Scene`/`SemanticScene`/`WorldState` and a mocked LLM client, asserting the loop completes normally whether verification succeeds or triggers reflection-driven recovery -- [ ] 8.2 Write an integration test (skippable without network/API credentials, following the existing `apex-agent-mvp` skippable-integration-test pattern) exercising Verifier/Reflector against the real LLM client -- [ ] 8.3 Confirm multi-agent collaboration stays disabled by default after applying this change (no existing task's behavior, latency, or cost changes unless a caller explicitly opts in) -- [ ] 8.4 Run the full test suite (`pytest`) and confirm no existing test in `tests/` needed a behavior change, only additive new tests +- [x] 8.1 Write an end-to-end test simulating a full collaborative task run against a mocked `Driver`/`Scene`/`SemanticScene`/`WorldState` and a mocked LLM client, asserting the loop completes normally whether verification succeeds or triggers reflection-driven recovery +- [x] 8.2 Write an integration test (skippable without network/API credentials, following the existing `apex-agent-mvp` skippable-integration-test pattern) exercising Verifier/Reflector against the real LLM client +- [x] 8.3 Confirm multi-agent collaboration stays disabled by default after applying this change (no existing task's behavior, latency, or cost changes unless a caller explicitly opts in) +- [x] 8.4 Run the full test suite (`pytest`) and confirm no existing test in `tests/` needed a behavior change, only additive new tests diff --git a/pyproject.toml b/pyproject.toml index 3ef2931..f2cce4d 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -25,6 +25,7 @@ dev = [ [tool.setuptools.packages.find] include = [ + "agents*", "api*", "core*", "device*", diff --git a/tests/test_agents_collab_runner.py b/tests/test_agents_collab_runner.py new file mode 100644 index 0000000..6141ca5 --- /dev/null +++ b/tests/test_agents_collab_runner.py @@ -0,0 +1,191 @@ +from __future__ import annotations + +from typing import Any +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 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="OK")], + ) + + +def _task() -> Task: + return Task(goal="Test goal", device_id="dev1") + + +def _planned_step() -> PlannedStep: + return PlannedStep(action="describe_screen", description="Observe", args={}) + + +def _success_step_result() -> StepResult: + return StepResult(step=_planned_step(), success=True, attempts=1, result="ok") + + +def _observation() -> Observation: + return Observation(scene_summary="Screen 1080x1920") + + +def _achieved_verdict() -> VerificationVerdict: + return VerificationVerdict(result="achieved", reasoning="Effect observed.") + + +def _not_achieved_verdict() -> VerificationVerdict: + return VerificationVerdict(result="not_achieved", reasoning="No change.") + + +def _replan_outcome() -> ReflectionOutcome: + return ReflectionOutcome(replan=True, reasoning="No recovery.") + + +def _recovery_outcome() -> ReflectionOutcome: + return ReflectionOutcome( + replan=False, + action=ReflectionAction(action="swipe", description="Scroll", args={"direction": "up"}), + reasoning="Try scrolling.", + ) + + +def test_completes_without_verification_failure() -> None: + """Task completes normally when verifier reports achieved.""" + planner = MagicMock() + planner.plan.return_value = [_planned_step()] + planner.goal_reached.return_value = True + + executor = MagicMock() + executor.execute.return_value = _success_step_result() + + observer = MagicMock() + observer.observe.return_value = _observation() + + verifier = MagicMock() + verifier.verify.return_value = _achieved_verdict() + + reflector = MagicMock() + + 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), + ) + + with patch("tools.describe_screen.describe_screen", return_value=_scene()): + task = runner.run(_task()) + + assert task.status == "completed" + reflector.reflect.assert_not_called() + + +def test_recovers_via_reflector_action() -> None: + """Task recovers when reflector proposes a distinct action.""" + call_count = 0 + + def verifier_side_effect(**kwargs: Any) -> VerificationVerdict: + nonlocal call_count + call_count += 1 + if call_count == 1: + return _not_achieved_verdict() + return _achieved_verdict() + + planner = MagicMock() + planner.plan.return_value = [_planned_step()] + planner.goal_reached.side_effect = [False, True] + + executor = MagicMock() + executor.execute.return_value = _success_step_result() + + observer = MagicMock() + observer.observe.return_value = _observation() + + verifier = MagicMock() + verifier.verify.side_effect = verifier_side_effect + + reflector = MagicMock() + reflector.reflect.return_value = _recovery_outcome() + + 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), + ) + + with patch("tools.describe_screen.describe_screen", return_value=_scene()): + task = runner.run(_task()) + + assert task.status == "completed" + reflector.reflect.assert_called_once() + executor.execute.assert_called() # recovery action also executed + + +def test_exhausts_recovery_ceiling() -> None: + """Task fails when reflection-recovery ceiling is reached.""" + planner = MagicMock() + planner.plan.return_value = [_planned_step()] + planner.goal_reached.return_value = False + + executor = MagicMock() + executor.execute.return_value = _success_step_result() + + observer = MagicMock() + observer.observe.return_value = _observation() + + verifier = MagicMock() + verifier.verify.return_value = _not_achieved_verdict() + + reflector = MagicMock() + reflector.reflect.return_value = _replan_outcome() + + 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), + ) + + 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() + + +def test_disabled_collaboration_runs_plain() -> None: + """When collaboration is disabled, delegates to plain TaskRunner.""" + planner = MagicMock() + executor = MagicMock() + + mock_task_runner = MagicMock() + expected_task = _task() + expected_task.status = "completed" + mock_task_runner.run.return_value = expected_task + + runner = CollaborativeTaskRunner( + planner=planner, + executor=executor, + task_runner=mock_task_runner, + collaboration_config=CollaborationConfig(enabled=False), + ) + + result = runner.run(_task()) + assert result.status == "completed" + mock_task_runner.run.assert_called_once() diff --git a/tests/test_agents_e2e.py b/tests/test_agents_e2e.py new file mode 100644 index 0000000..7aea666 --- /dev/null +++ b/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() diff --git a/tests/test_agents_integration.py b/tests/test_agents_integration.py new file mode 100644 index 0000000..b196413 --- /dev/null +++ b/tests/test_agents_integration.py @@ -0,0 +1,71 @@ +"""Integration test for Verifier/Reflector against the real LLM client (task 8.2). +Skippable without network/API credentials.""" + +from __future__ import annotations + +import os + +import pytest + +from agents.models import Observation, VerificationVerdict +from agents.reflector import Reflector +from agents.verifier import Verifier +from runtime.executor import StepResult +from runtime.planner import PlannedStep +from semantic.llm_client import AnthropicSemanticClient + +_has_api_key = bool(os.environ.get("ANTHROPIC_API_KEY")) + + +@pytest.mark.integration +@pytest.mark.skipif(not _has_api_key, reason="ANTHROPIC_API_KEY not set") +def test_verifier_against_real_llm() -> None: + client = AnthropicSemanticClient() + verifier = Verifier(client=client) + + pre = Observation( + scene_summary="Screen showing a chat app with a 'Reply' button visible.", + semantic_page="chat", + semantic_intents=["send_message"], + ) + post = Observation( + scene_summary="Screen showing a chat app with 'Message sent' confirmation visible.", + semantic_page="chat", + semantic_intents=["send_message"], + ) + step = PlannedStep(action="tap", description="Tap the send button", args={}) + result = StepResult(step=step, success=True, attempts=1, result={"tapped": True}) + + verdict = verifier.verify( + pre_observation=pre, + post_observation=post, + planned_step=step, + step_result=result, + ) + assert verdict.result in ("achieved", "not_achieved") + assert verdict.reasoning + + +@pytest.mark.integration +@pytest.mark.skipif(not _has_api_key, reason="ANTHROPIC_API_KEY not set") +def test_reflector_against_real_llm() -> None: + client = AnthropicSemanticClient() + reflector = Reflector(client=client) + + observation = Observation( + scene_summary="Screen unchanged after tapping send button.", + semantic_page="chat", + semantic_intents=["send_message"], + ) + step = PlannedStep(action="tap", description="Tap the send button", args={}) + result = StepResult(step=step, success=True, attempts=1, result={"tapped": True}) + verdict = VerificationVerdict(result="not_achieved", reasoning="Screen unchanged.") + + outcome = reflector.reflect( + observation=observation, + planned_step=step, + step_result=result, + verdict=verdict, + ) + assert outcome.replan is True or outcome.action is not None + assert outcome.reasoning diff --git a/tests/test_agents_models.py b/tests/test_agents_models.py new file mode 100644 index 0000000..7e299da --- /dev/null +++ b/tests/test_agents_models.py @@ -0,0 +1,77 @@ +from __future__ import annotations + +from agents.models import ( + Observation, + ReflectionAction, + ReflectionOutcome, + VerificationVerdict, +) + + +def test_observation_round_trip() -> None: + obs = Observation( + scene_summary="Screen 1080x1920\n [button] Send", + semantic_page="chat", + semantic_intents=["send_message"], + world_app="com.example.messenger", + world_page="conversation", + world_variables={"draft": "hello"}, + raw_scene={"screen": {"width": 1080, "height": 1920}, "elements": []}, + ) + data = obs.to_dict() + restored = Observation.from_dict(data) + assert restored == obs + + +def test_observation_minimal_round_trip() -> None: + obs = Observation(scene_summary="empty") + data = obs.to_dict() + restored = Observation.from_dict(data) + assert restored.scene_summary == "empty" + assert restored.semantic_page is None + assert restored.semantic_intents == [] + assert restored.world_app is None + assert restored.world_page is None + assert restored.world_variables == {} + assert restored.raw_scene is None + + +def test_verification_verdict_achieved_round_trip() -> None: + verdict = VerificationVerdict(result="achieved", reasoning="Effect observed.") + data = verdict.to_dict() + restored = VerificationVerdict.from_dict(data) + assert restored == verdict + + +def test_verification_verdict_not_achieved_round_trip() -> None: + verdict = VerificationVerdict(result="not_achieved", reasoning="Scene unchanged.") + data = verdict.to_dict() + restored = VerificationVerdict.from_dict(data) + assert restored == verdict + + +def test_reflection_action_round_trip() -> None: + action = ReflectionAction( + action="tap", + description="Tap the send button", + args={"element_id": "btn_send"}, + ) + data = action.to_dict() + restored = ReflectionAction.from_dict(data) + assert restored == action + + +def test_reflection_outcome_with_action_round_trip() -> None: + action = ReflectionAction(action="tap", description="Retry tap", args={}) + outcome = ReflectionOutcome(replan=False, action=action, reasoning="Try again.") + data = outcome.to_dict() + restored = ReflectionOutcome.from_dict(data) + assert restored == outcome + + +def test_reflection_outcome_replan_round_trip() -> None: + outcome = ReflectionOutcome(replan=True, reasoning="No recovery possible.") + data = outcome.to_dict() + restored = ReflectionOutcome.from_dict(data) + assert restored == outcome + assert restored.action is None diff --git a/tests/test_agents_no_coupling.py b/tests/test_agents_no_coupling.py new file mode 100644 index 0000000..972f3a8 --- /dev/null +++ b/tests/test_agents_no_coupling.py @@ -0,0 +1,36 @@ +"""Verify that the agents/ package does not introduce accidental coupling +into the existing runtime/ modules (task 7.2).""" + +from __future__ import annotations + +import importlib + + +def test_runtime_planner_unchanged() -> None: + """Planner should not import from agents/.""" + module = importlib.import_module("runtime.planner") + source = open(module.__file__).read() # type: ignore[arg-type] + assert "agents" not in source + + +def test_runtime_executor_unchanged() -> None: + """Executor should not import from agents/.""" + module = importlib.import_module("runtime.executor") + source = open(module.__file__).read() # type: ignore[arg-type] + assert "agents" not in source + + +def test_runtime_task_unchanged() -> None: + """TaskRunner should not import from agents/.""" + module = importlib.import_module("runtime.task") + source = open(module.__file__).read() # type: ignore[arg-type] + assert "agents" not in source + + +def test_agents_imports_runtime() -> None: + """agents/ should import from runtime/, not the other way around.""" + collab = importlib.import_module("agents.collab_runner") + source = open(collab.__file__).read() # type: ignore[arg-type] + assert "runtime.task" in source + assert "runtime.planner" in source + assert "runtime.executor" in source diff --git a/tests/test_agents_observer.py b/tests/test_agents_observer.py new file mode 100644 index 0000000..a206bab --- /dev/null +++ b/tests/test_agents_observer.py @@ -0,0 +1,85 @@ +from __future__ import annotations + +from collections import deque + +from agents.observer import Observer +from core.models import Bounds, Scene, SceneElement +from semantic.models import SemanticScene, SemanticWidget +from world.models import WorldState + + +def _make_scene() -> Scene: + return Scene( + width=1080, + height=1920, + elements=[ + SceneElement(id="btn1", type="button", bounds=Bounds(10, 20, 100, 50), text="Send"), + SceneElement(id="txt1", type="text", bounds=Bounds(0, 0, 500, 30), text="Hello"), + ], + ) + + +def _make_semantic_scene() -> SemanticScene: + return SemanticScene( + page="chat", + intents=["send_message"], + widgets=[SemanticWidget(element_id="btn1", purpose="send button")], + ) + + +def _make_world() -> WorldState: + state = WorldState( + current_app="com.example.messenger", + current_page="conversation", + variables={"draft": "hello"}, + ) + return state + + +def test_observe_with_semantic_and_world() -> None: + obs = Observer().observe( + scene=_make_scene(), + semantic_scene=_make_semantic_scene(), + world=_make_world(), + ) + assert obs.semantic_page == "chat" + assert obs.semantic_intents == ["send_message"] + assert obs.world_app == "com.example.messenger" + assert obs.world_page == "conversation" + assert obs.world_variables == {"draft": "hello"} + assert obs.raw_scene is not None + + +def test_observe_without_semantic_or_world() -> None: + obs = Observer().observe(scene=_make_scene()) + assert obs.semantic_page is None + assert obs.semantic_intents == [] + assert obs.world_app is None + assert obs.world_page is None + assert obs.world_variables == {} + assert obs.scene_summary is not None + assert obs.raw_scene is not None + + +def test_observe_with_semantic_only() -> None: + obs = Observer().observe( + scene=_make_scene(), + semantic_scene=_make_semantic_scene(), + ) + assert obs.semantic_page == "chat" + assert obs.world_app is None + + +def test_observe_with_world_only() -> None: + obs = Observer().observe( + scene=_make_scene(), + world=_make_world(), + ) + assert obs.semantic_page is None + assert obs.world_app == "com.example.messenger" + + +def test_observe_scene_summary_includes_elements() -> None: + obs = Observer().observe(scene=_make_scene()) + assert "1080x1920" in obs.scene_summary + assert "Send" in obs.scene_summary diff --git a/tests/test_agents_reflector.py b/tests/test_agents_reflector.py new file mode 100644 index 0000000..ef902bd --- /dev/null +++ b/tests/test_agents_reflector.py @@ -0,0 +1,136 @@ +from __future__ import annotations + +from agents.models import Observation, ReflectionOutcome, VerificationVerdict +from agents.reflector import Reflector +from runtime.executor import StepResult +from runtime.planner import PlannedStep + + +def _observation() -> Observation: + return Observation(scene_summary="Screen 1080x1920\n [button] Reply") + + +def _planned_step() -> PlannedStep: + return PlannedStep( + action="tap", + description="Tap the send button", + args={"element_id": "btn_send"}, + ) + + +def _success_result() -> StepResult: + return StepResult( + step=_planned_step(), + success=True, + attempts=1, + result={"tapped": True}, + ) + + +def _not_achieved_verdict() -> VerificationVerdict: + return VerificationVerdict(result="not_achieved", reasoning="Scene unchanged.") + + +def test_reflect_fallback_returns_replan() -> None: + reflector = Reflector() + outcome = reflector.reflect( + observation=_observation(), + planned_step=_planned_step(), + step_result=_success_result(), + verdict=_not_achieved_verdict(), + ) + assert outcome.replan is True + assert outcome.action is None + assert "fallback" in outcome.reasoning.lower() + + +class _FakeClient: + """A fake LLM client that returns a canned outcome.""" + + def __init__(self, outcome: dict) -> None: + self._outcome = outcome + self._calls: list[dict] = [] + + def _create_message(self, payload: dict, *, timeout: float) -> dict: + self._calls.append(payload) + return self._outcome + + +def test_reflect_recovery_action_via_llm() -> None: + fake = _FakeClient({ + "replan": False, + "action": { + "action": "swipe", + "description": "Scroll down to find send button", + "args": {"direction": "up"}, + }, + "reasoning": "Button may be off-screen.", + }) + reflector = Reflector(client=fake) # type: ignore[arg-type] + outcome = reflector.reflect( + observation=_observation(), + planned_step=_planned_step(), + step_result=_success_result(), + verdict=_not_achieved_verdict(), + ) + assert outcome.replan is False + assert outcome.action is not None + assert outcome.action.action == "swipe" + assert len(fake._calls) == 1 + + +def test_reflect_replan_via_llm() -> None: + fake = _FakeClient({ + "replan": True, + "action": None, + "reasoning": "No bounded recovery possible.", + }) + reflector = Reflector(client=fake) # type: ignore[arg-type] + outcome = reflector.reflect( + observation=_observation(), + planned_step=_planned_step(), + step_result=_success_result(), + verdict=_not_achieved_verdict(), + ) + assert outcome.replan is True + assert outcome.action is None + + +def test_reflect_rejects_identical_action() -> None: + """If the LLM proposes the exact same step, reflector should force replan.""" + fake = _FakeClient({ + "replan": False, + "action": { + "action": "tap", + "description": "Tap the send button", + "args": {"element_id": "btn_send"}, + }, + "reasoning": "Try again.", + }) + reflector = Reflector(client=fake) # type: ignore[arg-type] + outcome = reflector.reflect( + observation=_observation(), + planned_step=_planned_step(), + step_result=_success_result(), + verdict=_not_achieved_verdict(), + ) + assert outcome.replan is True + assert "identical" in outcome.reasoning.lower() + + +def test_reflect_degrades_on_client_failure() -> None: + from semantic.llm_client import EnrichmentUnavailable + + class _UnavailableClient: + def _create_message(self, *args, **kwargs): # type: ignore[no-untyped-def] + raise EnrichmentUnavailable("timeout") + + reflector = Reflector(client=_UnavailableClient()) # type: ignore[arg-type] + outcome = reflector.reflect( + observation=_observation(), + planned_step=_planned_step(), + step_result=_success_result(), + verdict=_not_achieved_verdict(), + ) + assert outcome.replan is True + assert "fallback" in outcome.reasoning.lower() diff --git a/tests/test_agents_verifier.py b/tests/test_agents_verifier.py new file mode 100644 index 0000000..1a96371 --- /dev/null +++ b/tests/test_agents_verifier.py @@ -0,0 +1,137 @@ +from __future__ import annotations + +from agents.models import Observation, VerificationVerdict +from agents.verifier import Verifier +from runtime.executor import StepResult +from runtime.planner import PlannedStep + + +def _pre_observation() -> Observation: + return Observation(scene_summary="Screen 1080x1920\n [button] Reply") + + +def _post_observation_changed() -> Observation: + return Observation(scene_summary="Screen 1080x1920\n [text] Message sent") + + +def _post_observation_unchanged() -> Observation: + return Observation(scene_summary="Screen 1080x1920\n [button] Reply") + + +def _planned_step() -> PlannedStep: + return PlannedStep( + action="tap", + description="Tap the send button", + args={"element_id": "btn_send"}, + ) + + +def _success_result() -> StepResult: + return StepResult( + step=_planned_step(), + success=True, + attempts=1, + result={"tapped": True}, + ) + + +def _failure_result() -> StepResult: + return StepResult( + step=_planned_step(), + success=False, + attempts=3, + error="element not found", + ) + + +def test_verify_not_achieved_on_mechanical_failure() -> None: + verifier = Verifier() + verdict = verifier.verify( + pre_observation=_pre_observation(), + post_observation=_post_observation_changed(), + planned_step=_planned_step(), + step_result=_failure_result(), + ) + assert verdict.result == "not_achieved" + assert "failed mechanically" in verdict.reasoning.lower() + + +def test_verify_achieved_fallback_scene_changed() -> None: + verifier = Verifier() + verdict = verifier.verify( + pre_observation=_pre_observation(), + post_observation=_post_observation_changed(), + planned_step=_planned_step(), + step_result=_success_result(), + ) + assert verdict.result == "achieved" + + +def test_verify_not_achieved_fallback_scene_unchanged() -> None: + verifier = Verifier() + verdict = verifier.verify( + pre_observation=_pre_observation(), + post_observation=_post_observation_unchanged(), + planned_step=_planned_step(), + step_result=_success_result(), + ) + assert verdict.result == "not_achieved" + + +class _FakeClient: + """A fake LLM client that returns a canned verdict.""" + + def __init__(self, verdict: dict) -> None: + self._verdict = verdict + self._calls: list[dict] = [] + + def _create_message(self, payload: dict, *, timeout: float) -> dict: + self._calls.append(payload) + return self._verdict + + +def test_verify_achieved_via_llm() -> None: + fake = _FakeClient({"result": "achieved", "reasoning": "Effect observed."}) + verifier = Verifier(client=fake) # type: ignore[arg-type] + verdict = verifier.verify( + pre_observation=_pre_observation(), + post_observation=_post_observation_changed(), + planned_step=_planned_step(), + step_result=_success_result(), + ) + assert verdict.result == "achieved" + assert len(fake._calls) == 1 + + +def test_verify_not_achieved_via_llm() -> None: + fake = _FakeClient({"result": "not_achieved", "reasoning": "No change."}) + verifier = Verifier(client=fake) # type: ignore[arg-type] + verdict = verifier.verify( + pre_observation=_pre_observation(), + post_observation=_post_observation_unchanged(), + planned_step=_planned_step(), + step_result=_success_result(), + ) + assert verdict.result == "not_achieved" + + +def test_verify_degrades_on_client_failure() -> None: + class _BrokenClient: + def _create_message(self, *args, **kwargs): # type: ignore[no-untyped-def] + raise RuntimeError("connection failed") + + from semantic.llm_client import EnrichmentUnavailable + + class _UnavailableClient: + def _create_message(self, *args, **kwargs): # type: ignore[no-untyped-def] + raise EnrichmentUnavailable("timeout") + + verifier = Verifier(client=_UnavailableClient()) # type: ignore[arg-type] + verdict = verifier.verify( + pre_observation=_pre_observation(), + post_observation=_post_observation_changed(), + planned_step=_planned_step(), + step_result=_success_result(), + ) + assert verdict.result == "achieved" + assert "fallback" in verdict.reasoning.lower() diff --git a/tests/test_smoke.py b/tests/test_smoke.py index 8e1f850..7f533f6 100644 --- a/tests/test_smoke.py +++ b/tests/test_smoke.py @@ -5,6 +5,7 @@ import importlib def test_imports_new_packages() -> None: for package in ( + "agents", "api", "core", "device",