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

@@ -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(
+        return self.task_runner.run(task)
-            planner=self.planner,
-            executor=self.executor,
-            config=TaskRunnerConfig(max_steps=self.config.max_steps),
-        )
-        return 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]
+        world_handle: TaskWorldView | None = self.task_runner._start_world_view(task.id)
-        task.updated_at = utc_now()
+        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]
+                return self.task_runner._complete_task(task)
-                task.updated_at = utc_now()
-                task.completed_at = utc_now()
-                return 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]
+                    return self._fail(
-                    task.updated_at = utc_now()
+                        task,
-                    task.completed_at = utc_now()
-                    task.failure_reason = (
                         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]
+                        return self._fail(
-                        task.updated_at = utc_now()
+                            task, f"Recovery action failed: {recovery_result.error}"
-                        task.completed_at = utc_now()
-                        task.failure_reason = (
-                            f"Recovery action failed: {recovery_result.error}"
                         )
-                        return task
 
-        task.status = "failed"  # type: ignore[assignment]
+                    recovery_scene = self._observe_scene(task.device_id)
-        task.updated_at = utc_now()
+                    pre_observation = self.observer.observe(
-        task.completed_at = utc_now()
+                        scene=recovery_scene,
-        task.failure_reason = f"max steps exceeded: {self.config.max_steps}"
+                        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:

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()