feat: preserve planner context across task steps
Tests / Test apps.device-host-agent.tests.test_mcp_token.test_load_or_create_concurrent_calls_do_not_corrupt failed

This commit is contained in:
showtan001
2026-08-30 22:59:19 +08:00
parent dd8df33910
commit 60ee157e97
11 changed files with 268 additions and 95 deletions
+57 -53
View File
@@ -8,6 +8,7 @@ from core.errors import TaskFailedError
from core.models import Bounds, Scene, SceneElement
from runtime.ai_planner import AIPlanner
from runtime.context import TaskContext
from runtime.executor import StepResult
from runtime.planner_config import PlannerConfig
from runtime.tool_calling_client import ToolCallDecision
from runtime.tool_specs import ALL_TOOL_SPECS
@@ -26,6 +27,7 @@ class FakeToolCallingClient:
screenshot: bytes | None,
tools: list[Any],
timeout: float,
history: list[dict[str, Any]] | None = None,
) -> ToolCallDecision:
self.calls.append(
{
@@ -34,6 +36,7 @@ class FakeToolCallingClient:
"screenshot": screenshot,
"tools": tools,
"timeout": timeout,
"history": list(history) if history is not None else None,
}
)
return self.decision
@@ -242,63 +245,64 @@ def test_ai_planner_propagates_rationale_and_thinking_to_planned_step() -> None:
assert steps[0].expected_outcome == "The account settings page is visible."
def test_history_summary_returns_compact_format() -> None:
from collections import deque
from runtime.ai_planner import _history_summary
from world.models import WorldEvent, WorldState
state = WorldState(
history=deque(
[
WorldEvent(
action="tap",
success=True,
rationale="Opened settings.",
arguments={"x": 1, "y": 2},
purpose="Open settings.",
expected_outcome="Settings is visible.",
page="Home",
),
WorldEvent(
action="swipe",
success=False,
rationale=None,
arguments={"start_y": 700, "end_y": 200},
page="Settings",
),
]
def test_ai_planner_carries_completed_turn_into_the_next_llm_call() -> None:
client = FakeToolCallingClient(
ToolCallDecision(
tool_name="tap",
arguments={"x": 1, "y": 2},
text_output="Opening the send control.",
purpose="Open the send control.",
expected_outcome="The composer is focused.",
)
)
planner = AIPlanner(client=client)
context = _context()
summary = _history_summary(state)
first_step = planner.plan(goal=context.goal, scene=_scene(), context=context)[0]
context.add_step_result(
StepResult(
step=first_step,
success=True,
attempts=1,
result={"ok": True},
)
)
planner.plan(goal=context.goal, scene=_scene(), context=context)
assert summary == [
assert client.calls[0]["history"] == []
history = client.calls[1]["history"]
assert history is not None
assert history[0]["tool_name"] == "tap"
assert history[0]["arguments"] == {
"x": 1,
"y": 2,
"purpose": "Open the send control.",
"expected_outcome": "The composer is focused.",
}
assert history[0]["tool_result"]["success"] is True
assert history[0]["tool_result"]["result"] == {"ok": True}
def test_ai_planner_limits_history_sent_to_the_llm() -> None:
client = FakeToolCallingClient(
ToolCallDecision(tool_name="tap", arguments={"x": 1, "y": 2})
)
planner = AIPlanner(client=client, config=PlannerConfig(history_max_turns=2))
context = _context()
context.planner_history.extend(
{
"page": "Home",
"action": "tap",
"arguments": {"x": 1, "y": 2},
"rationale": "Opened settings.",
"purpose": "Open settings.",
"expected_outcome": "Settings is visible.",
"success": True,
},
{
"page": "Settings",
"action": "swipe",
"arguments": {"start_y": 700, "end_y": 200},
"user_prompt": f"turn-{index}",
"tool_name": "tap",
"arguments": {},
"rationale": None,
"purpose": None,
"expected_outcome": None,
"success": False,
},
"tool_result": {"success": True},
}
for index in range(3)
)
planner.plan(goal=context.goal, scene=_scene(), context=context)
assert [turn["user_prompt"] for turn in client.calls[0]["history"]] == [
"turn-1",
"turn-2",
]
# Must not contain scene element data
for entry in summary:
assert "scene_summary" not in entry
assert "elements" not in entry
def test_history_summary_returns_empty_for_none_world() -> None:
from runtime.ai_planner import _history_summary
assert _history_summary(None) == []
+9 -2
View File
@@ -9,7 +9,6 @@ def test_planner_user_prompt_includes_time_zone_and_configured_device_type() ->
prompt = planner_user_prompt(
goal="open settings",
scene_json={"screen": {"width": 1, "height": 1}, "elements": []},
history_summary=[],
device_platform="ios",
now=datetime(
2026,
@@ -35,8 +34,16 @@ def test_planner_user_prompt_uses_scene_platform_when_context_is_unavailable() -
"elements": [],
"app": {"platform": "android"},
},
history_summary=[],
now=datetime(2026, 7, 16, tzinfo=timezone.utc),
)
assert "Device type: android" in prompt
def test_planner_user_prompt_does_not_duplicate_conversation_history() -> None:
prompt = planner_user_prompt(
goal="open settings",
scene_json={"screen": {"width": 1, "height": 1}, "elements": []},
)
assert "Recent history" not in prompt