137 lines
4.1 KiB
Python
137 lines
4.1 KiB
Python
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()
|