feat(agent): freeze root child input causality

agent/agent/core/runner.py

@@ -2142,12 +2142,15 @@ class AgentRunner:
 
     async def _get_mission_completion(self, root_trace_id: str) -> Dict[str, Any]:
         """Read the coordinator-owned mission completion snapshot."""
-        method = getattr(self.task_coordinator, "mission_completion", None)
+        method = getattr(self.task_coordinator, "root_completion", None)
         if method is None:
-            raise RuntimeError("TaskCoordinator does not provide mission_completion()")
+            # Compatibility with Coordinator stubs and older integrations.
+            method = getattr(self.task_coordinator, "mission_completion", None)
+        if method is None:
+            raise RuntimeError("TaskCoordinator does not provide root_completion()")
         snapshot = await method(root_trace_id)
         if not isinstance(snapshot, dict) or "status" not in snapshot:
-            raise RuntimeError("mission_completion() returned an invalid snapshot")
+            raise RuntimeError("root_completion() returned an invalid snapshot")
         return snapshot
 
     async def _restore_resume_identity(self, config: RunConfig) -> Optional[Trace]:

agent/agent/orchestration/coordinator.py

@@ -222,8 +222,8 @@ class TaskCoordinator:
                     return committed.ledger
             raise AssertionError("unreachable")
 
-    async def mission_completion(self, root_trace_id: str) -> Dict[str, Any]:
-        """Return the derived mission state without persisting a second state machine."""
+    async def root_completion(self, root_trace_id: str) -> Dict[str, Any]:
+        """Return Root Task completion without persisting a second state machine."""
 
         ledger = await self.task_store.load(root_trace_id)
         root = self._root_task(ledger)
@@ -248,12 +248,18 @@ class TaskCoordinator:
         ]
         return {
             "root_task_id": root.task_id,
+            "root_objective": ledger.root_objective,
             "status": root.status.value,
             "blocked_reason": root.blocked_reason,
             "result_summary": result_summary,
             "pending_tasks": pending,
         }
 
+    async def mission_completion(self, root_trace_id: str) -> Dict[str, Any]:
+        """Compatibility alias for callers using the pre-Root terminology."""
+
+        return await self.root_completion(root_trace_id)
+
     async def inspect_tasks(self, root_trace_id: str) -> Dict[str, Any]:
         ledger = await self.task_store.load(root_trace_id)
         self._root_task(ledger)
@@ -1053,6 +1059,9 @@ class TaskCoordinator:
                 worker_trace_id=worker_trace_id,
                 worker_preset=worker_preset,
                 execution_mode=execution_mode,
+                accepted_child_decision_ids=self._accepted_child_decision_ids(
+                    ledger, task
+                ),
                 continue_from_trace_id=continue_trace_id,
                 operation_id=operation_id,
                 execution_epoch=execution_epoch,
@@ -1133,6 +1142,53 @@ class TaskCoordinator:
             ledger = await self.task_store.load(root_trace_id)
             task = ledger.tasks[task_id]
             attempt = ledger.attempts[attempt_id]
+            if attempt.accepted_child_decision_ids is None:
+                accepted_child_results: List[Dict[str, Any]] = []
+                worker_result = WorkerRunResult(
+                    trace_id=attempt.worker_trace_id,
+                    status="failed",
+                    error=(
+                        "Legacy running attempt has no frozen child input binding"
+                    ),
+                    execution_stats=ExecutionStats(
+                        failure_code=FailureCode.PROTOCOL_VIOLATION
+                    ),
+                )
+            else:
+                try:
+                    accepted_child_results = self._accepted_child_results(
+                        ledger, task, attempt
+                    )
+                    for child_result in accepted_child_results:
+                        snapshot = await self.artifact_store.get(
+                            root_trace_id, child_result["snapshot_id"]
+                        )
+                        if snapshot.attempt_id != child_result["attempt_id"]:
+                            raise OrchestrationError(
+                                "Accepted child snapshot does not belong to its attempt"
+                            )
+                        child_attempt = ledger.attempts[child_result["attempt_id"]]
+                        if (
+                            snapshot.snapshot_id != child_attempt.snapshot_id
+                            or snapshot.artifact_refs
+                            != child_attempt.submission.artifact_refs  # type: ignore[union-attr]
+                            or snapshot.evidence_refs
+                            != child_attempt.submission.evidence_refs  # type: ignore[union-attr]
+                        ):
+                            raise OrchestrationError(
+                                "Accepted child snapshot does not match its submission"
+                            )
+                except Exception as exc:
+                    worker_result = WorkerRunResult(
+                        trace_id=attempt.worker_trace_id,
+                        status="failed",
+                        error=f"Invalid frozen child input binding: {exc}",
+                        execution_stats=ExecutionStats(
+                            failure_code=FailureCode.PROTOCOL_VIOLATION
+                        ),
+                    )
+                else:
+                    worker_result = None
             worker_context = {
                 "root_trace_id": root_trace_id,
                 "task_id": task_id,
@@ -1152,35 +1208,36 @@ class TaskCoordinator:
                     if attempt.execution_mode == "repair"
                     else None
                 ),
-                "accepted_child_results": self._accepted_child_results(ledger, task),
+                "accepted_child_results": accepted_child_results,
                 "operation_id": operation_id,
                 "execution_epoch": execution_epoch,
                 "deadline": deadline,
             }
-            try:
-                worker_result = await asyncio.wait_for(
-                    self.executor.run_worker(worker_context),  # type: ignore[union-attr]
-                    timeout=stage_timeout(
-                        deadline,
-                        self.config.worker_timeout_seconds,
-                    ),
-                )
-            except asyncio.TimeoutError:
-                worker_result = WorkerRunResult(
-                    trace_id=attempt.worker_trace_id,
-                    status="expired",
-                    error="Worker execution timed out",
-                    execution_stats=ExecutionStats(failure_code=FailureCode.TIMEOUT),
-                )
-            except Exception as exc:
-                worker_result = WorkerRunResult(
-                    trace_id=attempt.worker_trace_id,
-                    status="failed",
-                    error=f"Worker executor error: {exc}",
-                    execution_stats=ExecutionStats(
-                        failure_code=FailureCode.EXECUTOR_ERROR
-                    ),
-                )
+            if worker_result is None:
+                try:
+                    worker_result = await asyncio.wait_for(
+                        self.executor.run_worker(worker_context),  # type: ignore[union-attr]
+                        timeout=stage_timeout(
+                            deadline,
+                            self.config.worker_timeout_seconds,
+                        ),
+                    )
+                except asyncio.TimeoutError:
+                    worker_result = WorkerRunResult(
+                        trace_id=attempt.worker_trace_id,
+                        status="expired",
+                        error="Worker execution timed out",
+                        execution_stats=ExecutionStats(failure_code=FailureCode.TIMEOUT),
+                    )
+                except Exception as exc:
+                    worker_result = WorkerRunResult(
+                        trace_id=attempt.worker_trace_id,
+                        status="failed",
+                        error=f"Worker executor error: {exc}",
+                        execution_stats=ExecutionStats(
+                            failure_code=FailureCode.EXECUTOR_ERROR
+                        ),
+                    )
         else:
             worker_result = None
 
@@ -2679,26 +2736,60 @@ class TaskCoordinator:
     def _accepted_child_results(
         ledger: TaskLedger,
         task: TaskRecord,
+        attempt: Optional[TaskAttempt] = None,
     ) -> List[Dict[str, Any]]:
         """Build immutable, accepted direct-child inputs for a parent Worker."""
 
         results: List[Dict[str, Any]] = []
-        children = sorted(
-            (ledger.tasks[child_id] for child_id in task.child_task_ids),
-            key=lambda child: _path_key(child.display_path),
+        if attempt is None:
+            decision_ids = TaskCoordinator._accepted_child_decision_ids(ledger, task)
+        elif attempt.accepted_child_decision_ids is None:
+            raise OrchestrationError(
+                "Attempt has no frozen accepted child decision binding"
+            )
+        else:
+            decision_ids = attempt.accepted_child_decision_ids
+        if len(set(decision_ids)) != len(decision_ids):
+            raise OrchestrationError("Attempt child decision bindings are not unique")
+        expected_decision_ids = TaskCoordinator._accepted_child_decision_ids(
+            ledger, task
         )
-        for child in children:
-            if child.status != TaskStatus.COMPLETED:
-                continue
-            _, attempt, validation = TaskCoordinator._accepted_result_binding(
-                ledger, child
+        if tuple(decision_ids) != expected_decision_ids:
+            raise OrchestrationError(
+                "Attempt child decision bindings do not match direct children in display_path order"
+            )
+        direct_children = {child_id for child_id in task.child_task_ids}
+        prior_path: Optional[Tuple[int, ...]] = None
+        for decision_id in decision_ids:
+            decision = ledger.decisions.get(decision_id)
+            if (
+                decision is None
+                or decision.task_id not in direct_children
+                or decision.action != DecisionAction.ACCEPT
+            ):
+                raise OrchestrationError(
+                    "Attempt child decision binding is not a direct-child ACCEPT"
+                )
+            child = ledger.tasks[decision.task_id]
+            path = _path_key(child.display_path)
+            if prior_path is not None and path < prior_path:
+                raise OrchestrationError(
+                    "Attempt child decision bindings are not in display_path order"
+                )
+            prior_path = path
+            bound_decision, child_attempt, validation = (
+                TaskCoordinator._accepted_result_binding(ledger, child)
             )
+            if bound_decision.decision_id != decision_id:
+                raise OrchestrationError(
+                    "Attempt child decision binding is not the child's accepted result"
+                )
             results.append({
                 "task_id": child.task_id,
                 "task_spec": _plain(asdict(child.current_spec)),
-                "attempt_id": attempt.attempt_id,
-                "snapshot_id": attempt.snapshot_id,
-                "submission": _plain(asdict(attempt.submission)),
+                "attempt_id": child_attempt.attempt_id,
+                "snapshot_id": child_attempt.snapshot_id,
+                "submission": _plain(asdict(child_attempt.submission)),
                 "validation": {
                     "validation_id": validation.validation_id,
                     "summary": validation.summary,
@@ -2707,6 +2798,24 @@ class TaskCoordinator:
             })
         return results
 
+    @staticmethod
+    def _accepted_child_decision_ids(
+        ledger: TaskLedger,
+        task: TaskRecord,
+    ) -> Tuple[str, ...]:
+        decision_ids: List[str] = []
+        children = sorted(
+            (ledger.tasks[child_id] for child_id in task.child_task_ids),
+            key=lambda child: _path_key(child.display_path),
+        )
+        for child in children:
+            if child.status == TaskStatus.COMPLETED:
+                decision, _attempt, _validation = (
+                    TaskCoordinator._accepted_result_binding(ledger, child)
+                )
+                decision_ids.append(decision.decision_id)
+        return tuple(decision_ids)
+
     @staticmethod
     def _root_task(ledger: TaskLedger) -> TaskRecord:
         if not ledger.root_task_id:
@@ -2947,7 +3056,7 @@ class TaskCoordinator:
         task = _task(ledger, task_id)
         tree = await self.trace_store.get_goal_tree(root_trace_id)
         if tree is None:
-            tree = GoalTree(mission=ledger.mission)
+            tree = GoalTree(mission=ledger.root_objective)
         if task.goal_id and tree.find(task.goal_id):
             return
         parent_goal_id = None

agent/agent/orchestration/models.py

@@ -458,6 +458,9 @@ class TaskAttempt:
     worker_trace_id: str
     worker_preset: str
     execution_mode: str
+    # None is reserved for attempts written before child-input binding existed.
+    # New attempts always persist a tuple, including () when there are no inputs.
+    accepted_child_decision_ids: Optional[Tuple[str, ...]] = None
     status: AttemptStatus = AttemptStatus.RUNNING
     operation_id: Optional[str] = None
     execution_epoch: int = 0
@@ -475,6 +478,15 @@ class TaskAttempt:
     def duration_ms(self) -> Optional[int]:
         return _duration_ms(self.started_at, self.completed_at)
 
+    def __post_init__(self) -> None:
+        if self.accepted_child_decision_ids is not None:
+            values = tuple(self.accepted_child_decision_ids)
+            if any(not isinstance(item, str) or not item for item in values):
+                raise ValueError(
+                    "TaskAttempt.accepted_child_decision_ids must contain non-empty strings"
+                )
+            object.__setattr__(self, "accepted_child_decision_ids", values)
+
     @classmethod
     def from_dict(cls, data: Dict[str, Any]) -> "TaskAttempt":
         submission = data.get("submission")
@@ -485,6 +497,11 @@ class TaskAttempt:
             worker_trace_id=data["worker_trace_id"],
             worker_preset=data.get("worker_preset", "worker"),
             execution_mode=data.get("execution_mode", "new"),
+            accepted_child_decision_ids=(
+                tuple(data["accepted_child_decision_ids"])
+                if data.get("accepted_child_decision_ids") is not None
+                else None
+            ),
             status=AttemptStatus(data.get("status", AttemptStatus.RUNNING.value)),
             operation_id=data.get("operation_id"),
             execution_epoch=int(data.get("execution_epoch", 0)),
@@ -750,8 +767,20 @@ class TaskLedger:
     created_at: str = field(default_factory=utc_now)
     updated_at: str = field(default_factory=utc_now)
 
+    @property
+    def root_objective(self) -> str:
+        """Current Root Task objective, with a rootless legacy fallback."""
+
+        if self.root_task_id:
+            return self.tasks[self.root_task_id].current_spec.objective
+        return self.mission
+
     def to_dict(self) -> Dict[str, Any]:
-        return json_values(asdict(self))
+        data = json_values(asdict(self))
+        # Keep the historical JSON field normalized without creating a second
+        # persisted source of truth for the Root Task objective.
+        data["mission"] = self.root_objective
+        return data
 
     @classmethod
     def from_dict(cls, data: Dict[str, Any]) -> "TaskLedger":

agent/agent/orchestration/store.py

@@ -123,6 +123,7 @@ class FileSystemTaskStore:
             new_revision = expected_revision + 1
             updated_at = utc_now()
             data = ledger.to_dict()
+            ledger.mission = ledger.root_objective
             data["revision"] = new_revision
             data["updated_at"] = updated_at
             events = self._read_events(raw)

agent/tests/test_attempt_input_binding.py

@@ -0,0 +1,92 @@
+import pytest
+
+from agent.orchestration.coordinator import OrchestrationError
+from agent.orchestration.models import (
+    DecisionAction,
+    FailureCode,
+    TaskStatus,
+    ValidationVerdict,
+)
+
+from test_coordinator_integration import FakeExecutor, create_task, make_coordinator
+
+
+async def _accept_child(coordinator, task_id, key):
+    cycle = (await coordinator.dispatch_tasks("root", [task_id]))[0]
+    await coordinator.decide_task(
+        "root",
+        task_id,
+        cycle.validation_id,
+        DecisionAction.ACCEPT,
+        {"reason": "accepted"},
+        key,
+    )
+    return cycle
+
+
+@pytest.mark.asyncio
+async def test_new_attempt_distinguishes_empty_and_ordered_child_bindings(tmp_path):
+    executor = FakeExecutor([ValidationVerdict.PASSED, ValidationVerdict.PASSED])
+    coordinator, store, _ = await make_coordinator(tmp_path, executor)
+    first = await create_task(coordinator, "first")
+    second = await create_task(coordinator, "second")
+    await _accept_child(coordinator, second, "accept-second")
+    await _accept_child(coordinator, first, "accept-first")
+
+    ledger = await store.load("root")
+    root = ledger.tasks[ledger.root_task_id]
+    reserved = await coordinator._create_attempt("root", root.task_id, "worker", None)
+    frozen = (await store.load("root")).attempts[reserved["attempt_id"]]
+    assert frozen.accepted_child_decision_ids == (
+        ledger.tasks[first].decision_ids[-1],
+        ledger.tasks[second].decision_ids[-1],
+    )
+    first_attempt = ledger.attempts[ledger.tasks[first].attempt_ids[-1]]
+    assert first_attempt.accepted_child_decision_ids == ()
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("corruption", ["duplicate", "reverse", "cross", "non_accept"])
+async def test_frozen_child_decision_binding_rejects_corruption(tmp_path, corruption):
+    executor = FakeExecutor([ValidationVerdict.PASSED, ValidationVerdict.PASSED])
+    coordinator, store, _ = await make_coordinator(tmp_path, executor)
+    first = await create_task(coordinator, "first")
+    second = await create_task(coordinator, "second")
+    await _accept_child(coordinator, first, "accept-first")
+    await _accept_child(coordinator, second, "accept-second")
+    ledger = await store.load("root")
+    root = ledger.tasks[ledger.root_task_id]
+    reserved = await coordinator._create_attempt("root", root.task_id, "worker", None)
+    ledger = await store.load("root")
+    attempt = ledger.attempts[reserved["attempt_id"]]
+    ids = list(attempt.accepted_child_decision_ids)
+    if corruption == "duplicate":
+        attempt.accepted_child_decision_ids = (ids[0], ids[0])
+    elif corruption == "reverse":
+        attempt.accepted_child_decision_ids = tuple(reversed(ids))
+    elif corruption == "cross":
+        attempt.accepted_child_decision_ids = ("missing-decision",)
+    else:
+        decision = ledger.decisions[ids[0]]
+        object.__setattr__(decision, "action", DecisionAction.RETRY)
+    with pytest.raises(OrchestrationError):
+        coordinator._accepted_child_results(ledger, root, attempt)
+
+
+@pytest.mark.asyncio
+async def test_legacy_running_attempt_never_guesses_child_inputs(tmp_path):
+    executor = FakeExecutor([])
+    coordinator, store, _ = await make_coordinator(tmp_path, executor)
+    root_id = (await store.load("root")).root_task_id
+    reserved = await coordinator._create_attempt("root", root_id, "worker", None)
+    ledger = await store.load("root")
+    ledger.attempts[reserved["attempt_id"]].accepted_child_decision_ids = None
+    await store.commit(ledger, expected_revision=ledger.revision)
+
+    result = await coordinator.advance_cycle("root", root_id, reserved["attempt_id"])
+    recovered = await store.load("root")
+    attempt = recovered.attempts[reserved["attempt_id"]]
+    assert result.error == "Legacy running attempt has no frozen child input binding"
+    assert executor.worker_calls == 0
+    assert recovered.tasks[root_id].status == TaskStatus.NEEDS_REPLAN
+    assert attempt.execution_stats.failure_code == FailureCode.PROTOCOL_VIOLATION

agent/tests/test_mission_root.py

@@ -41,6 +41,45 @@ async def test_new_ledger_has_one_stable_root_and_inspect_exposes_it(tmp_path):
     ] == "mission-done"
 
 
+@pytest.mark.asyncio
+async def test_root_revise_drives_root_objective_mission_and_goal_rebuild(tmp_path):
+    coordinator, store, trace_store = await make_coordinator(
+        tmp_path, FakeExecutor([])
+    )
+    ledger = await store.load("root")
+    root_id = ledger.root_task_id
+    assert ledger.root_objective == "mission"
+
+    await coordinator.decide_task(
+        "root",
+        root_id,
+        None,
+        DecisionAction.REVISE,
+        {"reason": "clarify", "objective": "revised mission"},
+        "revise-root-objective",
+    )
+    revised = await store.load("root")
+    assert revised.root_objective == "revised mission"
+    assert revised.mission == "revised mission"
+
+    goal_file = trace_store._get_goal_file("root")
+    goal_file.unlink()
+    await coordinator._ensure_goal_projection("root", root_id)
+    rebuilt = await trace_store.get_goal_tree("root")
+    assert rebuilt is not None
+    assert rebuilt.mission == "revised mission"
+
+
+@pytest.mark.asyncio
+async def test_rootless_legacy_ledger_falls_back_to_mission(tmp_path):
+    store = FileSystemTaskStore(str(tmp_path))
+    ledger = TaskLedger(root_trace_id="legacy-rootless", mission="legacy mission")
+    assert ledger.root_objective == "legacy mission"
+    committed = await store.commit(ledger, expected_revision=-1)
+    assert committed.ledger.mission == "legacy mission"
+    assert (await store.load("legacy-rootless")).root_objective == "legacy mission"
+
+
 @pytest.mark.asyncio
 async def test_new_ledger_requires_root_spec_and_rootless_policy_is_explicit(tmp_path):
     store = FileSystemTaskStore(str(tmp_path))