138 lines
4.2 KiB
Python
138 lines
4.2 KiB
Python
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()
|