From 031b929067632d6920d5dbdbf058dbce418cfd59 Mon Sep 17 00:00:00 2001 From: Jerry Yan <792602257@qq.com> Date: Tue, 7 Jul 2026 08:30:57 +0800 Subject: [PATCH] fix(multi-agent-collaboration): compose TaskRunner bookkeeping instead of forking it CollaborativeTaskRunner hand-rolled its own step loop instead of composing runtime/task.py's TaskRunner as design.md D3 requires, silently dropping Timeline recording, WorldModel wiring, TaskMetadataStore sync, and the on_task_succeeded/skill-synthesis hook. Now calls TaskRunner's shared _start_world_view/ _record_step_result helpers for that bookkeeping. Also fixes pre_observation being reused stale across steps in a multi-step plan instead of refreshing to the prior step's post-observation. openspec: multi-agent-collaboration capability, archived change multi-agent-runtime --- agents/collab_runner.py | 78 +++++++++++------- tests/test_agents_collab_runner.py | 128 +++++++++++++++++++++++++++++ 2 files changed, 176 insertions(+), 30 deletions(-) diff --git a/agents/collab_runner.py b/agents/collab_runner.py index b23f057..b3458a5 100644 --- a/agents/collab_runner.py +++ b/agents/collab_runner.py @@ -8,12 +8,13 @@ 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 core.models import Scene, Task 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 +from world.model import TaskWorldView logger = logging.getLogger(__name__) @@ -40,12 +41,19 @@ class CollaborativeTaskRunner: ) -> 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() + # Shared with `_run_plain`: also holds the Timeline/TaskMetadataStore/ + # WorldModel/on_task_succeeded wiring that `_run_collaborative` reuses + # for its own per-step bookkeeping, so the two paths cannot drift apart. + self.task_runner = task_runner or TaskRunner( + planner=self.planner, + executor=self.executor, + config=TaskRunnerConfig(max_steps=self.config.max_steps), + ) def run(self, task: Task) -> Task: if not self.collaboration_config.enabled: @@ -53,17 +61,14 @@ class CollaborativeTaskRunner: 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) + return self.task_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() + world_handle: TaskWorldView | None = self.task_runner._start_world_view(task.id) + if world_handle is not None: + context.world = world_handle.state + self.task_runner._update_task(task, status="running") recovery_attempts = 0 for _ in range(self.config.max_steps): @@ -85,15 +90,15 @@ class CollaborativeTaskRunner: scene=scene, context=context, ): - task.status = "completed" # type: ignore[assignment] - task.updated_at = utc_now() - task.completed_at = utc_now() - return task + return self.task_runner._complete_task(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) + self.task_runner._record_step_result( + world_handle, context, task, scene, step, result + ) post_scene = self._observe_scene(task.device_id) post_observation = self.observer.observe( @@ -108,18 +113,21 @@ class CollaborativeTaskRunner: step_result=result, ) + # Refresh pre_observation for the *next* step in this plan (or + # the next outer-loop iteration) so it always reflects the + # most recent known device state, never the stale + # pre-whole-plan observation. + pre_observation = post_observation + 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 = ( + return self._fail( + task, f"Reflection recovery ceiling reached " - f"({self.collaboration_config.max_recovery_attempts} attempts)" + f"({self.collaboration_config.max_recovery_attempts} attempts)", ) - return task outcome = self.reflector.reflect( observation=post_observation, @@ -143,19 +151,29 @@ class CollaborativeTaskRunner: context=context, ) context.add_step_result(recovery_result) + self.task_runner._record_step_result( + world_handle, context, task, post_scene, recovery_step, 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 self._fail( + task, 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}" + recovery_scene = self._observe_scene(task.device_id) + pre_observation = self.observer.observe( + scene=recovery_scene, + world=context.world, + ) + + return self._fail(task, f"max steps exceeded: {self.config.max_steps}") + + def _fail(self, task: Task, reason: str) -> Task: + self.task_runner._update_task( + task, + status="failed", + completed=True, + failure_reason=reason, + ) return task def _observe_scene(self, device_id: str) -> Scene: diff --git a/tests/test_agents_collab_runner.py b/tests/test_agents_collab_runner.py index 6141ca5..d129ad5 100644 --- a/tests/test_agents_collab_runner.py +++ b/tests/test_agents_collab_runner.py @@ -9,6 +9,11 @@ from agents.models import Observation, ReflectionAction, ReflectionOutcome, Veri from core.models import Bounds, Scene, SceneElement, Task from runtime.executor import StepResult from runtime.planner import PlannedStep +from runtime.task import TaskRunner, TaskRunnerConfig +from storage.artifact_store import ArtifactStore +from storage.task_metadata import TaskMetadataStore +from storage.timeline import Timeline +from world.config import WorldConfig def _scene() -> Scene: @@ -169,6 +174,129 @@ def test_exhausts_recovery_ceiling() -> None: assert "ceiling" in (task.failure_reason or "").lower() +def test_collaborative_run_shares_bookkeeping_with_task_runner(tmp_path) -> None: + """A completed collaborative task records Timeline/TaskMetadataStore/ + on_task_succeeded exactly like a plain TaskRunner.run() would for the + same scripted single-step scenario, and populates context.world along + the way (verified indirectly via what the Observer is called with).""" + 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.return_value = _achieved_verdict() + + reflector = MagicMock() + + metadata = TaskMetadataStore(tmp_path / "tasks.sqlite3") + timeline = Timeline(ArtifactStore(tmp_path / "history")) + on_task_succeeded = MagicMock() + + shared_task_runner = TaskRunner( + metadata_store=metadata, + timeline=timeline, + on_task_succeeded=on_task_succeeded, + world_config=WorldConfig(enabled=True), + config=TaskRunnerConfig(max_steps=5), + ) + + runner = CollaborativeTaskRunner( + planner=planner, + executor=executor, + observer=observer, + verifier=verifier, + reflector=reflector, + task_runner=shared_task_runner, + config=CollaborativeTaskRunnerConfig(max_steps=5), + collaboration_config=CollaborationConfig(enabled=True, max_recovery_attempts=3), + ) + + task = _task() + metadata.create_task(task) + + with patch("tools.describe_screen.describe_screen", return_value=_scene()): + result = runner.run(task) + + assert result.status == "completed" + assert len(timeline.read(task.id)) == 1 + assert metadata.get_task(task.id)["status"] == "completed" + on_task_succeeded.assert_called_once_with(task.id, task.goal, timeline) + + # context.world was populated (not None) throughout the run: every + # Observer.observe call received a non-None `world` kwarg. + assert observer.observe.call_args_list + for call in observer.observe.call_args_list: + assert call.kwargs["world"] is not None + + +def test_second_step_verified_against_post_step_one_observation() -> None: + """When a single Planner.plan() call returns a 2-element plan, the + second step's Verifier.verify() call uses the first step's post + observation as its pre_observation, not the stale pre-plan observation.""" + step_one = _planned_step() + step_two = PlannedStep(action="tap", description="tap next", args={"x": 1, "y": 2}) + + planner = MagicMock() + planner.plan.return_value = [step_one, step_two] + planner.goal_reached.side_effect = [False, True] + + executor = MagicMock() + executor.execute.return_value = _success_step_result() + + obs_pre_plan = Observation(scene_summary="obs0-pre-plan") + obs_after_step_one = Observation(scene_summary="obs1-after-step-one") + obs_after_step_two = Observation(scene_summary="obs2-after-step-two") + obs_next_outer_iter = Observation(scene_summary="obs3-next-outer-iter") + + observer = MagicMock() + observer.observe.side_effect = [ + obs_pre_plan, + obs_after_step_one, + obs_after_step_two, + obs_next_outer_iter, + ] + + 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() + assert verifier.verify.call_count == 2 + + first_call_kwargs = verifier.verify.call_args_list[0].kwargs + second_call_kwargs = verifier.verify.call_args_list[1].kwargs + + assert first_call_kwargs["pre_observation"] is obs_pre_plan + assert first_call_kwargs["post_observation"] is obs_after_step_one + + # The bug being guarded against: the second step must not be verified + # against the stale pre-whole-plan observation. + assert second_call_kwargs["pre_observation"] is obs_after_step_one + assert second_call_kwargs["pre_observation"] is not obs_pre_plan + assert second_call_kwargs["post_observation"] is obs_after_step_two + + def test_disabled_collaboration_runs_plain() -> None: """When collaboration is disabled, delegates to plain TaskRunner.""" planner = MagicMock()