q792602257
49ad589c2d
tools/ is a Hexagonal inner layer that must never depend on LLM concerns (ADR 0002), but describe_screen_semantic.py imported semantic.enricher, which pulls in the Anthropic client by default. Relocated the wrapper to runtime/, which is where LLM-dependent code is allowed to live; updated the tool registry and all test imports accordingly. No behavior change. openspec: semantic-scene capability, archived change semantic-scene-runtime
135 lines
4.0 KiB
Python
135 lines
4.0 KiB
Python
from __future__ import annotations
|
|
|
|
from typing import Any
|
|
|
|
import pytest
|
|
|
|
from core.models import Bounds, Scene, SceneElement, Task
|
|
from device.manager import DeviceManager
|
|
from runtime.describe_screen_semantic import describe_screen_semantic
|
|
from runtime.executor import Executor, ExecutorConfig
|
|
from runtime.planner import PlannedStep, Planner
|
|
from runtime.task import TaskRunner, TaskRunnerConfig
|
|
from semantic.config import SemanticConfig
|
|
from semantic.llm_client import EnrichmentUnavailable
|
|
from tests.fakes import FakeDriver, PNG_10X20
|
|
|
|
|
|
class FakePerceptionProvider:
|
|
def __init__(self, scene: Scene) -> None:
|
|
self.scene = scene
|
|
|
|
def build_scene(self, screenshot: bytes, raw_tree: Any) -> Scene:
|
|
return self.scene
|
|
|
|
|
|
class FakeClient:
|
|
def __init__(
|
|
self,
|
|
*,
|
|
response: dict[str, Any] | None = None,
|
|
error: Exception | None = None,
|
|
) -> None:
|
|
self.response = response
|
|
self.error = error
|
|
|
|
def enrich(self, scene_json: dict[str, Any], *, timeout: float) -> dict[str, Any]:
|
|
if self.error:
|
|
raise self.error
|
|
assert self.response is not None
|
|
return self.response
|
|
|
|
|
|
class ObserveSemanticPlanner(Planner):
|
|
def plan(self, *, goal: str, scene: Scene, context: object) -> list[PlannedStep]:
|
|
if context.step_results: # type: ignore[attr-defined]
|
|
return []
|
|
return [
|
|
PlannedStep(
|
|
action="describe_screen_semantic",
|
|
description=f"Observe current screen for goal: {goal}",
|
|
)
|
|
]
|
|
|
|
def goal_reached(self, *, goal: str, scene: Scene, context: object) -> bool:
|
|
return bool(context.step_results) and all( # type: ignore[attr-defined]
|
|
result.success for result in context.step_results # type: ignore[attr-defined]
|
|
)
|
|
|
|
|
|
def _scene() -> Scene:
|
|
return Scene(
|
|
width=10,
|
|
height=20,
|
|
elements=[
|
|
SceneElement(
|
|
id="send",
|
|
type="button",
|
|
text="Send",
|
|
bounds=Bounds(1, 2, 3, 4),
|
|
)
|
|
],
|
|
)
|
|
|
|
|
|
def _connected_manager() -> DeviceManager:
|
|
manager = DeviceManager()
|
|
manager.register_device("phone", FakeDriver)
|
|
manager.connect("phone", max_retries=1)
|
|
return manager
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"client,expected_semantic",
|
|
[
|
|
(
|
|
FakeClient(
|
|
response={
|
|
"page": "Chat",
|
|
"intents": ["send a message"],
|
|
"widgets": [{"element_id": "send", "purpose": "send message"}],
|
|
}
|
|
),
|
|
True,
|
|
),
|
|
(FakeClient(error=EnrichmentUnavailable("forced failure")), False),
|
|
],
|
|
)
|
|
def test_task_loop_completes_when_describe_screen_semantic_succeeds_or_degrades(
|
|
client: FakeClient,
|
|
expected_semantic: bool,
|
|
) -> None:
|
|
scene = _scene()
|
|
manager = _connected_manager()
|
|
provider = FakePerceptionProvider(scene)
|
|
tool_results: list[dict[str, object]] = []
|
|
|
|
def semantic_tool(device_id: str | None = None) -> dict[str, object]:
|
|
result = describe_screen_semantic(
|
|
device_id,
|
|
manager=manager,
|
|
perception_provider=provider,
|
|
client=client,
|
|
semantic_config=SemanticConfig(enabled=True),
|
|
)
|
|
tool_results.append(result)
|
|
return result
|
|
|
|
runner = TaskRunner(
|
|
planner=ObserveSemanticPlanner(),
|
|
executor=Executor(
|
|
tools={"describe_screen_semantic": semantic_tool},
|
|
config=ExecutorConfig(max_retries=1, backoff_seconds=0),
|
|
),
|
|
config=TaskRunnerConfig(max_steps=3),
|
|
observer=lambda device_id: scene,
|
|
screenshot_provider=lambda device_id: PNG_10X20,
|
|
)
|
|
task = Task(goal="inspect current screen", device_id="phone")
|
|
|
|
result = runner.run(task)
|
|
|
|
assert result.status == "completed"
|
|
assert len(tool_results) == 1
|
|
assert (tool_results[0]["semantic_scene"] is not None) is expected_semantic
|