feat(orchestration): add root mission lifecycle

agent/agent/core/prompts/orchestration.py

@@ -8,6 +8,9 @@ dispatch_tasks. Worker completion is only an attempt submission. A task is
 complete only after an independent passed validation and your task_decide
 action=accept. For failed, inconclusive, or validation errors, explicitly
 repair, retry, revise, split, revalidate, block, cancel, or supersede.
+Start or resume by calling task_plan operation=inspect. The returned Root Task
+is the mission: top-level tasks are its children, and the run is complete only
+after the Root itself has passed validation and you accept it.
 """.strip()
 
 WORKER_ROLE_CONTRACT = """

agent/agent/orchestration/api_v2.py

@@ -224,6 +224,7 @@ def create_orchestration_router(coordinator: TaskCoordinator) -> Any:
             ledger = await coordinator.task_store.load(root_trace_id)
             return TaskList(
                 revision=ledger.revision,
+                root_task_id=ledger.root_task_id,
                 items=[
                     TaskView.from_domain(item)
                     for item in sorted(

agent/agent/orchestration/coordinator.py

@@ -15,7 +15,6 @@ from agent.trace.goal_models import GoalTree
 
 from .config import OrchestrationConfig
 from .models import (
-    AcceptanceCriterion,
     AgentRole,
     ArtifactRef,
     AttemptStatus,
@@ -84,6 +83,12 @@ TERMINAL_TASK_STATUSES = {
     TaskStatus.SUPERSEDED,
 }
 
+ACTIVE_TASK_STATUSES = {
+    TaskStatus.RUNNING,
+    TaskStatus.AWAITING_VALIDATION,
+    TaskStatus.VALIDATING,
+}
+
 
 class TaskCoordinator:
     """Coordinates Task -> Attempt -> Validation -> Planner Decision.
@@ -141,28 +146,135 @@ class TaskCoordinator:
     def _idem(root_trace_id: str, key: Optional[str]) -> Optional[str]:
         return f"{root_trace_id}:{key}" if key else None
 
-    async def ensure_ledger(self, root_trace_id: str, mission: str) -> TaskLedger:
+    async def ensure_ledger(
+        self,
+        root_trace_id: str,
+        root_task_spec: Optional[Dict[str, Any]] = None,
+        *,
+        allow_create: bool = True,
+    ) -> TaskLedger:
         async with self._lock(root_trace_id):
             for attempt_number in range(8):
                 try:
-                    return await self.task_store.load(root_trace_id)
+                    ledger = await self.task_store.load(root_trace_id)
+                    if ledger.root_trace_id != root_trace_id:
+                        raise OrchestrationError("Task ledger root trace identity is invalid")
+                    self._root_task(ledger)
+                    return ledger
                 except TaskStoreNotFound:
-                    ledger = TaskLedger(root_trace_id=root_trace_id, mission=mission)
+                    if not allow_create:
+                        raise TaskConflict(
+                            "Existing explicit Planner Trace has no Root Task ledger; "
+                            "rebuild the development trace"
+                        )
+                    if root_task_spec is None:
+                        raise ValueError(
+                            "root_task_spec is required for a new explicit-validation Planner trace"
+                        )
+                    if not isinstance(root_task_spec, dict):
+                        raise ValueError("root_task_spec must be an object")
+                    unknown = set(root_task_spec) - {
+                        "objective", "acceptance_criteria", "context_refs",
+                    }
+                    if unknown:
+                        raise ValueError(
+                            f"root_task_spec contains unsupported fields: {sorted(unknown)}"
+                        )
+                    spec = self._task_spec_from_draft(root_task_spec, version=1)
+                    root_task_id = new_id()
+                    root = TaskRecord(
+                        task_id=root_task_id,
+                        goal_id=None,
+                        parent_task_id=None,
+                        display_path="0",
+                        specs=[spec],
+                        status=TaskStatus.NEEDS_REPLAN,
+                    )
+                    ledger = TaskLedger(
+                        root_trace_id=root_trace_id,
+                        mission=spec.objective,
+                        root_task_id=root_task_id,
+                        tasks={root_task_id: root},
+                        focused_task_id=root_task_id,
+                    )
                     try:
                         committed = await self.task_store.commit(
                             ledger,
                             expected_revision=-1,
-                            event=EventDraft("ledger_created", {"mission": mission}),
+                            event=EventDraft(
+                                "ledger_created",
+                                {
+                                    "mission": spec.objective,
+                                    "root_task_id": root_task_id,
+                                },
+                            ),
                         )
                     except RevisionConflict:
                         if attempt_number == 7:
                             raise
                         await asyncio.sleep(_cas_backoff(attempt_number))
                         continue
-                    await self._emit(root_trace_id, "ledger_created", {"mission": mission})
+                    await self._emit(
+                        root_trace_id,
+                        "ledger_created",
+                        {"mission": spec.objective, "root_task_id": root_task_id},
+                    )
                     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."""
+
+        ledger = await self.task_store.load(root_trace_id)
+        root = self._root_task(ledger)
+        nonterminal_descendants = self._nonterminal_descendants(ledger, root)
+        if root.status == TaskStatus.COMPLETED and nonterminal_descendants:
+            raise OrchestrationError(
+                "Completed root task has non-terminal descendants"
+            )
+        result_summary = None
+        if root.status == TaskStatus.COMPLETED:
+            _, attempt, _ = self._accepted_result_binding(ledger, root)
+            result_summary = attempt.submission.summary
+        pending = [
+            {
+                "task_id": task.task_id,
+                "display_path": task.display_path,
+                "status": task.status.value,
+                "objective": task.current_spec.objective,
+            }
+            for task in sorted(ledger.tasks.values(), key=lambda item: _path_key(item.display_path))
+            if task.status not in TERMINAL_TASK_STATUSES
+        ]
+        return {
+            "root_task_id": root.task_id,
+            "status": root.status.value,
+            "blocked_reason": root.blocked_reason,
+            "result_summary": result_summary,
+            "pending_tasks": pending,
+        }
+
+    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)
+        return {
+            "root_task_id": ledger.root_task_id,
+            "focused_task_id": ledger.focused_task_id,
+            "tasks": [
+                {
+                    "task_id": task.task_id,
+                    "parent_task_id": task.parent_task_id,
+                    "display_path": task.display_path,
+                    "status": task.status.value,
+                    "blocked_reason": task.blocked_reason,
+                    "current_spec": _plain(asdict(task.current_spec)),
+                }
+                for task in sorted(
+                    ledger.tasks.values(), key=lambda item: _path_key(item.display_path)
+                )
+            ],
+        }
+
     async def _mutate(
         self,
         root_trace_id: str,
@@ -262,13 +374,27 @@ class TaskCoordinator:
             after_task_id = placement.get("after_task_id")
             if after_task_id:
                 target = _task(ledger, after_task_id)
+                if target.task_id == ledger.root_task_id:
+                    raise ValueError("The root task cannot have siblings")
                 if parent_task_id is not None and target.parent_task_id != parent_task_id:
                     raise ValueError("after_task_id must be a sibling under parent_task_id")
                 effective_parent_id = target.parent_task_id
             else:
-                effective_parent_id = parent_task_id
+                effective_parent_id = parent_task_id or ledger.root_task_id
             if effective_parent_id and effective_parent_id not in ledger.tasks:
                 raise ValueError(f"Parent task not found: {effective_parent_id}")
+            if effective_parent_id is None:
+                raise OrchestrationError("Task ledger has no root task")
+            parent = ledger.tasks[effective_parent_id]
+            if parent.status not in {
+                TaskStatus.PENDING,
+                TaskStatus.NEEDS_REPLAN,
+                TaskStatus.WAITING_CHILDREN,
+            }:
+                raise TaskConflict(
+                    f"Task {effective_parent_id} cannot receive children from "
+                    f"{parent.status.value}"
+                )
             siblings = sorted(
                 (t for t in ledger.tasks.values() if t.parent_task_id == effective_parent_id),
                 key=lambda item: _path_key(item.display_path),
@@ -277,7 +403,11 @@ class TaskCoordinator:
                 target_index = next(i for i, sibling in enumerate(siblings) if sibling.task_id == after_task_id)
                 start = target_index + 2
                 for index, sibling in enumerate(siblings[target_index + 1:], start=start + len(drafts)):
-                    sibling.display_path = self._display_path(ledger, effective_parent_id, index)
+                    self._rebase_task_path(
+                        ledger,
+                        sibling,
+                        self._display_path(ledger, effective_parent_id, index),
+                    )
             else:
                 start = len(siblings) + 1
             parent_path = ledger.tasks[effective_parent_id].display_path if effective_parent_id else ""
@@ -299,6 +429,7 @@ class TaskCoordinator:
                     else:
                         children.append(task_id)
                 created.append(task_id)
+            self._mark_parent_waiting(parent)
             if placement and placement.get("focus") and created:
                 ledger.focused_task_id = created[0]
             return {
@@ -894,6 +1025,10 @@ class TaskCoordinator:
             task = _task(ledger, task_id)
             if task.status not in (TaskStatus.PENDING, TaskStatus.NEEDS_REPLAN):
                 raise TaskConflict(f"Task {task_id} cannot dispatch from {task.status.value}")
+            if self._nonterminal_descendants(ledger, task):
+                raise TaskConflict(
+                    f"Task {task_id} cannot dispatch while descendants are non-terminal"
+                )
             active = [
                 ledger.attempts[x] for x in task.attempt_ids
                 if ledger.attempts[x].status == AttemptStatus.RUNNING
@@ -1017,6 +1152,7 @@ class TaskCoordinator:
                     if attempt.execution_mode == "repair"
                     else None
                 ),
+                "accepted_child_results": self._accepted_child_results(ledger, task),
                 "operation_id": operation_id,
                 "execution_epoch": execution_epoch,
                 "deadline": deadline,
@@ -2040,6 +2176,7 @@ class TaskCoordinator:
         def mutate(ledger: TaskLedger) -> Dict[str, Any]:
             task = _task(ledger, task_id)
             before = task.status
+            is_root = task.task_id == ledger.root_task_id
             validation = ledger.validations.get(validation_id) if validation_id else None
             attempt_id = validation.attempt_id if validation else (task.attempt_ids[-1] if task.attempt_ids else None)
             reason = str(payload.get("reason", "")).strip()
@@ -2064,19 +2201,24 @@ class TaskCoordinator:
             elif action == DecisionAction.REVISE:
                 if task.status not in (TaskStatus.AWAITING_DECISION, TaskStatus.NEEDS_REPLAN):
                     raise TaskConflict("Revise requires awaiting_decision or needs_replan")
-                objective = str(payload.get("objective") or task.current_spec.objective).strip()
-                criteria_data = payload.get("acceptance_criteria")
-                criteria = (
-                    [AcceptanceCriterion.from_dict(x) for x in criteria_data]
-                    if criteria_data is not None else list(task.current_spec.acceptance_criteria)
+                objective = (
+                    payload["objective"]
+                    if "objective" in payload
+                    else task.current_spec.objective
                 )
                 version = task.current_spec_version + 1
-                task.specs.append(TaskSpec(
-                    version=version,
-                    objective=objective,
-                    acceptance_criteria=criteria,
-                    context_refs=list(payload.get("context_refs", task.current_spec.context_refs)),
-                ))
+                draft = {
+                    "objective": objective,
+                    "acceptance_criteria": (
+                        payload["acceptance_criteria"]
+                        if "acceptance_criteria" in payload
+                        else [_plain(asdict(x)) for x in task.current_spec.acceptance_criteria]
+                    ),
+                    "context_refs": payload.get(
+                        "context_refs", task.current_spec.context_refs
+                    ),
+                }
+                task.specs.append(self._task_spec_from_draft(draft, version=version))
                 task.current_spec_version = version
                 target = TaskStatus.PENDING
             elif action == DecisionAction.SPLIT:
@@ -2091,6 +2233,11 @@ class TaskCoordinator:
             elif action == DecisionAction.BLOCK:
                 if task.status in TERMINAL_TASK_STATUSES:
                     raise TaskConflict("Terminal task cannot be blocked")
+                if any(
+                    child.status in ACTIVE_TASK_STATUSES
+                    for child in self._nonterminal_descendants(ledger, task)
+                ):
+                    raise TaskConflict("Task with active descendants cannot be blocked")
                 task.blocked_reason = reason
                 target = TaskStatus.BLOCKED
             elif action == DecisionAction.UNBLOCK:
@@ -2099,12 +2246,20 @@ class TaskCoordinator:
                 task.blocked_reason = None
                 target = TaskStatus.NEEDS_REPLAN
             elif action == DecisionAction.CANCEL:
+                if is_root:
+                    raise TaskConflict("The root task cannot be cancelled")
                 if task.status in TERMINAL_TASK_STATUSES:
                     raise TaskConflict("Task is already terminal")
+                if self._nonterminal_descendants(ledger, task):
+                    raise TaskConflict("Task with non-terminal descendants cannot be cancelled")
                 target = TaskStatus.CANCELLED
             elif action == DecisionAction.SUPERSEDE:
+                if is_root:
+                    raise TaskConflict("The root task cannot be superseded")
                 if task.status in TERMINAL_TASK_STATUSES:
                     raise TaskConflict("Task is already terminal")
+                if self._nonterminal_descendants(ledger, task):
+                    raise TaskConflict("Task with non-terminal descendants cannot be superseded")
                 replacement = payload.get("replacement") or {}
                 replacement_ids = self._create_sibling_records(ledger, task, [replacement])
                 task.superseded_by = replacement_ids[0]
@@ -2191,6 +2346,8 @@ class TaskCoordinator:
         attempt = ledger.attempts[validation.attempt_id]
         if validation.snapshot_id != attempt.snapshot_id:
             raise TaskConflict("Validation does not belong to the current snapshot")
+        if TaskCoordinator._nonterminal_descendants(ledger, task):
+            raise TaskConflict("Task cannot be accepted while descendants are non-terminal")
 
     @staticmethod
     def _guard_replan_decision(
@@ -2518,6 +2675,100 @@ class TaskCoordinator:
             "recommendation": report.recommendation,
         }
 
+    @staticmethod
+    def _accepted_child_results(
+        ledger: TaskLedger,
+        task: TaskRecord,
+    ) -> 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),
+        )
+        for child in children:
+            if child.status != TaskStatus.COMPLETED:
+                continue
+            _, attempt, validation = TaskCoordinator._accepted_result_binding(
+                ledger, child
+            )
+            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)),
+                "validation": {
+                    "validation_id": validation.validation_id,
+                    "summary": validation.summary,
+                    "evidence_refs": _plain(asdict(validation)["evidence_refs"]),
+                },
+            })
+        return results
+
+    @staticmethod
+    def _root_task(ledger: TaskLedger) -> TaskRecord:
+        if not ledger.root_task_id:
+            raise TaskConflict(
+                "Rootless task ledger is incompatible with mission execution; "
+                "rebuild the development trace"
+            )
+        root = ledger.tasks.get(ledger.root_task_id)
+        top_level = [task.task_id for task in ledger.tasks.values() if task.parent_task_id is None]
+        if not root or root.parent_task_id is not None or top_level != [root.task_id]:
+            raise OrchestrationError("Task ledger Root Task structure is invalid")
+        return root
+
+    @staticmethod
+    def _accepted_result_binding(
+        ledger: TaskLedger,
+        task: TaskRecord,
+    ) -> Tuple[PlannerDecision, TaskAttempt, ValidationReport]:
+        if task.status != TaskStatus.COMPLETED or not task.decision_ids:
+            raise OrchestrationError(f"Task {task.task_id} has no accepted result")
+        decision = ledger.decisions.get(task.decision_ids[-1])
+        if (
+            not decision
+            or decision.task_id != task.task_id
+            or decision.action != DecisionAction.ACCEPT
+            or not decision.attempt_id
+            or not decision.validation_id
+        ):
+            raise OrchestrationError(f"Task {task.task_id} has an invalid accept decision")
+        attempt = ledger.attempts.get(decision.attempt_id)
+        validation = ledger.validations.get(decision.validation_id)
+        if (
+            not attempt
+            or attempt.task_id != task.task_id
+            or not attempt.submission
+            or not attempt.snapshot_id
+            or attempt.spec_version != task.current_spec_version
+            or not validation
+            or validation.task_id != task.task_id
+            or validation.attempt_id != attempt.attempt_id
+            or validation.snapshot_id != attempt.snapshot_id
+            or validation.spec_version != task.current_spec_version
+            or validation.status != ValidationRunStatus.COMPLETED
+            or validation.verdict != ValidationVerdict.PASSED
+        ):
+            raise OrchestrationError(f"Task {task.task_id} has an invalid accepted result binding")
+        return decision, attempt, validation
+
+    @staticmethod
+    def _nonterminal_descendants(
+        ledger: TaskLedger,
+        task: TaskRecord,
+    ) -> List[TaskRecord]:
+        descendants: List[TaskRecord] = []
+        pending_ids = list(task.child_task_ids)
+        while pending_ids:
+            child = ledger.tasks[pending_ids.pop()]
+            pending_ids.extend(child.child_task_ids)
+            if child.status not in TERMINAL_TASK_STATUSES:
+                descendants.append(child)
+        return descendants
+
     def _create_child_records(
         self,
         ledger: TaskLedger,
@@ -2542,19 +2793,8 @@ class TaskCoordinator:
         parent_task_id: Optional[str],
         display_path: str,
     ) -> str:
-        objective = str(draft.get("objective", "")).strip()
-        if not objective:
-            raise ValueError("Task objective is required")
         task_id = new_id()
-        spec = TaskSpec(
-            version=1,
-            objective=objective,
-            acceptance_criteria=[
-                AcceptanceCriterion.from_dict(item)
-                for item in draft.get("acceptance_criteria", [])
-            ],
-            context_refs=list(draft.get("context_refs", [])),
-        )
+        spec = TaskCoordinator._task_spec_from_draft(draft, version=1)
         ledger.tasks[task_id] = TaskRecord(
             task_id=task_id,
             goal_id=None,
@@ -2564,6 +2804,17 @@ class TaskCoordinator:
         )
         return task_id
 
+    @staticmethod
+    def _task_spec_from_draft(draft: Dict[str, Any], *, version: int) -> TaskSpec:
+        if not isinstance(draft, dict):
+            raise ValueError("Task draft must be an object")
+        return TaskSpec.from_dict({
+            "version": version,
+            "objective": draft.get("objective", ""),
+            "acceptance_criteria": draft.get("acceptance_criteria", []),
+            "context_refs": draft.get("context_refs", ()),
+        })
+
     def _create_records(
         self,
         ledger: TaskLedger,
@@ -2590,22 +2841,25 @@ class TaskCoordinator:
             created.append(task_id)
         return created
 
+    @staticmethod
+    def _mark_parent_waiting(parent: TaskRecord) -> None:
+        if parent.status == TaskStatus.WAITING_CHILDREN:
+            return
+        transition(parent.status, TaskStatus.WAITING_CHILDREN)
+        parent.status = TaskStatus.WAITING_CHILDREN
+        parent.updated_at = utc_now()
+
     @staticmethod
     def _update_parent_after_child(ledger: TaskLedger, child: TaskRecord) -> None:
         if not child.parent_task_id or child.status not in TERMINAL_TASK_STATUSES:
             return
         parent = ledger.tasks[child.parent_task_id]
+        if parent.status != TaskStatus.WAITING_CHILDREN:
+            return
         children = [ledger.tasks[x] for x in parent.child_task_ids]
-        if child.status in (TaskStatus.CANCELLED, TaskStatus.SUPERSEDED):
-            if parent.status == TaskStatus.WAITING_CHILDREN:
-                transition(parent.status, TaskStatus.NEEDS_REPLAN)
-                parent.status = TaskStatus.NEEDS_REPLAN
-        elif children and all(x.status == TaskStatus.COMPLETED for x in children):
-            if parent.status == TaskStatus.WAITING_CHILDREN:
-                transition(parent.status, TaskStatus.NEEDS_REPLAN)
-                parent.status = TaskStatus.NEEDS_REPLAN
-        elif any(x.status in {TaskStatus.RUNNING, TaskStatus.VALIDATING, TaskStatus.BLOCKED, TaskStatus.PENDING} for x in children):
-            parent.status = TaskStatus.WAITING_CHILDREN
+        if children and all(x.status in TERMINAL_TASK_STATUSES for x in children):
+            transition(parent.status, TaskStatus.NEEDS_REPLAN)
+            parent.status = TaskStatus.NEEDS_REPLAN
         parent.updated_at = utc_now()
 
     async def project_goal_state(self, root_trace_id: str, task_id: str) -> None:
@@ -2759,6 +3013,20 @@ class TaskCoordinator:
             return str(index)
         return f"{ledger.tasks[parent_task_id].display_path}.{index}"
 
+    @staticmethod
+    def _rebase_task_path(
+        ledger: TaskLedger,
+        task: TaskRecord,
+        display_path: str,
+    ) -> None:
+        task.display_path = display_path
+        for index, child_id in enumerate(task.child_task_ids, start=1):
+            TaskCoordinator._rebase_task_path(
+                ledger,
+                ledger.tasks[child_id],
+                f"{display_path}.{index}",
+            )
+
 
 def _task(ledger: TaskLedger, task_id: str) -> TaskRecord:
     try:

agent/agent/orchestration/executor.py

@@ -53,6 +53,7 @@ class LocalAgentExecutor:
             "task_spec": context["task_spec"],
             "attempt_id": context["attempt_id"],
             "repair_feedback": context.get("repair_feedback"),
+            "accepted_child_results": context.get("accepted_child_results", []),
             "instruction": "Execute this TaskSpec and finish with submit_attempt.",
         }
         result = await self._run_role(

agent/agent/orchestration/models.py

@@ -249,12 +249,30 @@ class AcceptanceCriterion:
     description: str
     hard: bool = True
 
+    def __post_init__(self) -> None:
+        if not isinstance(self.criterion_id, str) or not self.criterion_id.strip():
+            raise ValueError("AcceptanceCriterion.criterion_id must be a non-empty string")
+        if not isinstance(self.description, str) or not self.description.strip():
+            raise ValueError("AcceptanceCriterion.description must be a non-empty string")
+        if not isinstance(self.hard, bool):
+            raise ValueError("AcceptanceCriterion.hard must be a boolean")
+        object.__setattr__(self, "criterion_id", self.criterion_id.strip())
+        object.__setattr__(self, "description", self.description.strip())
+
     @classmethod
     def from_dict(cls, data: Dict[str, Any]) -> "AcceptanceCriterion":
+        if not isinstance(data, dict):
+            raise ValueError("AcceptanceCriterion must be an object")
+        if "criterion_id" in data:
+            criterion_id = data["criterion_id"]
+        elif "id" in data:
+            criterion_id = data["id"]
+        else:
+            criterion_id = new_id()
         return cls(
-            criterion_id=str(data.get("criterion_id") or data.get("id") or new_id()),
-            description=str(data.get("description", "")),
-            hard=bool(data.get("hard", data.get("required", True))),
+            criterion_id=criterion_id,
+            description=data.get("description", ""),
+            hard=data.get("hard", data.get("required", True)),
         )
 
 
@@ -262,17 +280,59 @@ class AcceptanceCriterion:
 class TaskSpec:
     version: int
     objective: str
-    acceptance_criteria: List[AcceptanceCriterion] = field(default_factory=list)
-    context_refs: List[str] = field(default_factory=list)
+    acceptance_criteria: Tuple[AcceptanceCriterion, ...] = field(default_factory=tuple)
+    context_refs: Tuple[str, ...] = field(default_factory=tuple)
     created_at: str = field(default_factory=utc_now)
 
+    def __post_init__(self) -> None:
+        if (
+            isinstance(self.version, bool)
+            or not isinstance(self.version, int)
+            or self.version <= 0
+        ):
+            raise ValueError("TaskSpec.version must be a positive integer")
+        if not isinstance(self.objective, str) or not self.objective.strip():
+            raise ValueError("TaskSpec.objective must be a non-empty string")
+        if isinstance(self.context_refs, (str, bytes, dict)):
+            raise ValueError("TaskSpec.context_refs must be a collection of strings")
+        try:
+            acceptance_criteria = tuple(self.acceptance_criteria)
+            context_refs = tuple(self.context_refs)
+        except TypeError as exc:
+            raise ValueError("TaskSpec collections must be iterable") from exc
+        if not acceptance_criteria:
+            raise ValueError("TaskSpec.acceptance_criteria must contain at least one criterion")
+        if any(not isinstance(item, AcceptanceCriterion) for item in acceptance_criteria):
+            raise ValueError(
+                "TaskSpec.acceptance_criteria must contain AcceptanceCriterion values"
+            )
+        criterion_ids = [item.criterion_id for item in acceptance_criteria]
+        if len(set(criterion_ids)) != len(criterion_ids):
+            raise ValueError("TaskSpec acceptance criterion IDs must be unique")
+        if any(not isinstance(item, str) or not item.strip() for item in context_refs):
+            raise ValueError("TaskSpec.context_refs must contain non-empty strings")
+        object.__setattr__(self, "objective", self.objective.strip())
+        object.__setattr__(self, "acceptance_criteria", acceptance_criteria)
+        object.__setattr__(self, "context_refs", tuple(item.strip() for item in context_refs))
+
     @classmethod
     def from_dict(cls, data: Dict[str, Any]) -> "TaskSpec":
+        if not isinstance(data, dict):
+            raise ValueError("TaskSpec must be an object")
+        criteria_data = data.get("acceptance_criteria", [])
+        if isinstance(criteria_data, (str, bytes, dict)) or criteria_data is None:
+            raise ValueError("TaskSpec.acceptance_criteria must be a collection")
+        try:
+            acceptance_criteria = tuple(
+                AcceptanceCriterion.from_dict(item) for item in criteria_data
+            )
+        except TypeError as exc:
+            raise ValueError("TaskSpec.acceptance_criteria must be a collection") from exc
         return cls(
-            version=int(data.get("version", 1)),
-            objective=str(data.get("objective", "")),
-            acceptance_criteria=[AcceptanceCriterion.from_dict(x) for x in data.get("acceptance_criteria", [])],
-            context_refs=list(data.get("context_refs", [])),
+            version=data.get("version", 1),
+            objective=data.get("objective", ""),
+            acceptance_criteria=acceptance_criteria,
+            context_refs=data.get("context_refs", ()),
             created_at=data.get("created_at", utc_now()),
         )
 
@@ -677,6 +737,7 @@ class EventPage:
 class TaskLedger:
     root_trace_id: str
     mission: str
+    root_task_id: Optional[str] = None
     revision: int = -1
     tasks: Dict[str, TaskRecord] = field(default_factory=dict)
     attempts: Dict[str, TaskAttempt] = field(default_factory=dict)
@@ -697,6 +758,7 @@ class TaskLedger:
         return cls(
             root_trace_id=data["root_trace_id"],
             mission=data.get("mission", ""),
+            root_task_id=data.get("root_task_id"),
             revision=int(data.get("revision", -1)),
             tasks={k: TaskRecord.from_dict(v) for k, v in data.get("tasks", {}).items()},
             attempts={k: TaskAttempt.from_dict(v) for k, v in data.get("attempts", {}).items()},

agent/agent/orchestration/state_machine.py

@@ -12,12 +12,12 @@ class InvalidTaskTransition(ValueError):
 
 
 LEGAL_TRANSITIONS: Dict[TaskStatus, FrozenSet[TaskStatus]] = {
-    TaskStatus.PENDING: frozenset({TaskStatus.RUNNING, TaskStatus.BLOCKED, TaskStatus.CANCELLED, TaskStatus.SUPERSEDED}),
+    TaskStatus.PENDING: frozenset({TaskStatus.RUNNING, TaskStatus.WAITING_CHILDREN, TaskStatus.BLOCKED, TaskStatus.CANCELLED, TaskStatus.SUPERSEDED}),
     TaskStatus.RUNNING: frozenset({TaskStatus.AWAITING_VALIDATION, TaskStatus.NEEDS_REPLAN, TaskStatus.BLOCKED, TaskStatus.CANCELLED, TaskStatus.SUPERSEDED}),
     TaskStatus.AWAITING_VALIDATION: frozenset({TaskStatus.VALIDATING, TaskStatus.NEEDS_REPLAN, TaskStatus.BLOCKED, TaskStatus.CANCELLED, TaskStatus.SUPERSEDED}),
     TaskStatus.VALIDATING: frozenset({TaskStatus.AWAITING_DECISION, TaskStatus.NEEDS_REPLAN, TaskStatus.BLOCKED, TaskStatus.CANCELLED, TaskStatus.SUPERSEDED}),
     TaskStatus.AWAITING_DECISION: frozenset({TaskStatus.COMPLETED, TaskStatus.PENDING, TaskStatus.WAITING_CHILDREN, TaskStatus.BLOCKED, TaskStatus.CANCELLED, TaskStatus.SUPERSEDED}),
-    TaskStatus.NEEDS_REPLAN: frozenset({TaskStatus.PENDING, TaskStatus.VALIDATING, TaskStatus.WAITING_CHILDREN, TaskStatus.BLOCKED, TaskStatus.CANCELLED, TaskStatus.SUPERSEDED}),
+    TaskStatus.NEEDS_REPLAN: frozenset({TaskStatus.PENDING, TaskStatus.RUNNING, TaskStatus.VALIDATING, TaskStatus.WAITING_CHILDREN, TaskStatus.BLOCKED, TaskStatus.CANCELLED, TaskStatus.SUPERSEDED}),
     TaskStatus.WAITING_CHILDREN: frozenset({TaskStatus.NEEDS_REPLAN, TaskStatus.BLOCKED, TaskStatus.CANCELLED, TaskStatus.SUPERSEDED}),
     TaskStatus.BLOCKED: frozenset({TaskStatus.NEEDS_REPLAN, TaskStatus.CANCELLED, TaskStatus.SUPERSEDED}),
     TaskStatus.COMPLETED: frozenset(),

agent/agent/orchestration/wire.py

@@ -215,6 +215,7 @@ class ValidationView(WireModel):
 
 class TaskList(WireModel):
     revision: int
+    root_task_id: Optional[str] = None
     items: List[TaskView]
 
 

agent/agent/tools/builtin/orchestration.py

@@ -36,7 +36,9 @@ async def task_plan(
     context = context or {}
     coordinator = _coordinator(context)
     root_trace_id = context["root_trace_id"]
-    if operation == "focus":
+    if operation == "inspect":
+        result = await coordinator.inspect_tasks(root_trace_id)
+    elif operation == "focus":
         if not focus_task_id:
             raise ValueError("focus_task_id is required")
         result = await coordinator.focus_task(root_trace_id, focus_task_id)
@@ -52,7 +54,7 @@ async def task_plan(
         if focus_task_id:
             await coordinator.focus_task(root_trace_id, focus_task_id)
     else:
-        raise ValueError("operation must be create or focus")
+        raise ValueError("operation must be create, focus, or inspect")
     return json.dumps(result, ensure_ascii=False, indent=2)
 
 

agent/tests/test_coordinator_integration.py

@@ -25,6 +25,14 @@ from agent.core.runner import AgentRunner
 from agent.orchestration.wiring import wire_orchestration
 
 
+ROOT_TASK_SPEC = {
+    "objective": "mission",
+    "acceptance_criteria": [
+        {"criterion_id": "mission-done", "description": "mission is complete", "hard": True}
+    ],
+}
+
+
 class FakeExecutor:
     def __init__(self, verdicts, submit_worker=True, submit_validator=True, delay=0):
         self.verdicts = deque(verdicts)
@@ -107,7 +115,7 @@ async def make_coordinator(
         executor,
     )
     executor.coordinator = coordinator
-    await coordinator.ensure_ledger("root", "mission")
+    await coordinator.ensure_ledger("root", ROOT_TASK_SPEC)
     return coordinator, task_store, trace_store
 
 
@@ -294,7 +302,7 @@ async def test_block_unblock_cancel_and_supersede(tmp_path):
     replacement_spec = ledger.tasks[replacement_id].current_spec
     assert replacement_spec.objective == "replacement"
     assert replacement_spec.acceptance_criteria[0].criterion_id == "replacement-c1"
-    assert replacement_spec.context_refs == ["replacement-context"]
+    assert replacement_spec.context_refs == ("replacement-context",)
 
 
 @pytest.mark.asyncio
@@ -305,21 +313,33 @@ async def test_insert_after_keeps_stable_ids_and_updates_display_order(tmp_path)
     third = await create_task(coordinator, "third")
     inserted = await coordinator.create_tasks(
         "root",
-        [{"objective": "second", "context_refs": ["second-context"]}],
+        [{
+            "objective": "second",
+            "acceptance_criteria": [
+                {"criterion_id": "second-c1", "description": "second passes"}
+            ],
+            "context_refs": ["second-context"],
+        }],
         placement={"after_task_id": first, "focus": True},
         idempotency_key="insert",
     )
     second = inserted["tasks"][0]["task_id"]
     ledger = await store.load("root")
-    assert ledger.tasks[first].display_path == "1"
-    assert ledger.tasks[second].display_path == "2"
-    assert ledger.tasks[third].display_path == "3"
-    assert ledger.tasks[second].current_spec.context_refs == ["second-context"]
+    assert ledger.tasks[first].display_path == "0.1"
+    assert ledger.tasks[second].display_path == "0.2"
+    assert ledger.tasks[third].display_path == "0.3"
+    assert ledger.tasks[second].current_spec.context_refs == ("second-context",)
     assert ledger.focused_task_id == second
     assert len({first, second, third}) == 3
     repeated = await coordinator.create_tasks(
         "root",
-        [{"objective": "second", "context_refs": ["second-context"]}],
+        [{
+            "objective": "second",
+            "acceptance_criteria": [
+                {"criterion_id": "second-c1", "description": "second passes"}
+            ],
+            "context_refs": ["second-context"],
+        }],
         placement={"after_task_id": first, "focus": True},
         idempotency_key="insert",
     )
@@ -354,10 +374,10 @@ async def test_split_children_do_not_auto_complete_parent(tmp_path):
     child_ids = decision["payload"]["child_task_ids"]
     ledger = await store.load("root")
     assert ledger.tasks[parent_id].child_task_ids == child_ids
-    assert [ledger.tasks[x].display_path for x in child_ids] == ["1.1", "1.2"]
+    assert [ledger.tasks[x].display_path for x in child_ids] == ["0.1.1", "0.1.2"]
     assert [ledger.tasks[x].current_spec.context_refs for x in child_ids] == [
-        ["child-one-context"],
-        ["child-two-context"],
+        ("child-one-context",),
+        ("child-two-context",),
     ]
     assert [
         ledger.tasks[x].current_spec.acceptance_criteria[0].criterion_id
@@ -435,7 +455,7 @@ async def test_real_runner_local_executor_creates_independent_terminal_traces(tm
         FileSystemTaskStore(str(tmp_path)),
         FileSystemArtifactStore(str(tmp_path)),
     )
-    await coordinator.ensure_ledger("root", "mission")
+    await coordinator.ensure_ledger("root", ROOT_TASK_SPEC)
     task_id = await create_task(coordinator)
     cycle = (await coordinator.dispatch_tasks("root", [task_id]))[0]
     ledger = await coordinator.task_store.load("root")
@@ -626,7 +646,7 @@ async def test_task_creation_survives_goal_tree_projection_failure_and_reconcile
 
     result = await coordinator.reconcile_goal_tree("root")
     ledger = await store.load("root")
-    assert result["reconciled_tasks"] == 1
+    assert result["reconciled_tasks"] == 2
     assert ledger.tasks[task_id].goal_id is not None
 
 

agent/tests/test_orchestration_v2_api.py

@@ -9,6 +9,7 @@ from agent.orchestration.api_v2 import create_orchestration_router
 from agent.orchestration.client_v2 import OrchestrationAPIError, OrchestrationClient
 from agent.orchestration.coordinator import TaskConflict
 from agent.orchestration.models import (
+    AcceptanceCriterion,
     BackgroundOperation,
     EventPage,
     OperationKind,
@@ -84,7 +85,18 @@ def api_fixture():
         goal_id=None,
         parent_task_id=None,
         display_path="1",
-        specs=[TaskSpec(version=1, objective="test")],
+        specs=[
+            TaskSpec(
+                version=1,
+                objective="test",
+                acceptance_criteria=[
+                    AcceptanceCriterion(
+                        criterion_id="done",
+                        description="The task is complete",
+                    )
+                ],
+            )
+        ],
         status=TaskStatus.PENDING,
     )
     attempt = TaskAttempt(
@@ -111,7 +123,12 @@ def api_fixture():
         request_fingerprint="private-fingerprint",
         task_ids=[task.task_id],
     )
-    ledger = TaskLedger(root_trace_id="root", mission="test", revision=3)
+    ledger = TaskLedger(
+        root_trace_id="root",
+        mission="test",
+        root_task_id=task.task_id,
+        revision=3,
+    )
     ledger.tasks[task.task_id] = task
     ledger.attempts[attempt.attempt_id] = attempt
     ledger.validations[validation.validation_id] = validation
@@ -147,7 +164,9 @@ async def test_router_start_queries_and_problem_details(api_fixture):
     assert "request_fingerprint" not in response.json()
     assert coordinator.calls[0][-1]["idempotency_key"] == "dispatch-once"
 
-    assert (await http.get("/api/v2/roots/root/tasks")).json()["revision"] == 3
+    task_list = (await http.get("/api/v2/roots/root/tasks")).json()
+    assert task_list["revision"] == 3
+    assert task_list["root_task_id"] == "task-1"
     assert (await http.get("/api/v2/roots/root/tasks/task-1")).json()["task_id"] == "task-1"
     assert (await http.get("/api/v2/roots/root/attempts")).json()["items"][0]["attempt_id"] == "attempt-1"
     assert (await http.get("/api/v2/roots/root/attempts/attempt-1")).json()["attempt_id"] == "attempt-1"
@@ -238,7 +257,9 @@ async def test_async_client_round_trip_and_watch(api_fixture):
     assert (await client.poll("root", "operation-1")).operation_id == "operation-1"
     assert (await client.stop("root", "operation-1")).status == "stopped"
     assert (await client.resume("root", "operation-1")).status == "pending"
-    assert (await client.list_tasks("root")).items[0].task_id == "task-1"
+    task_list = await client.list_tasks("root")
+    assert task_list.root_task_id == "task-1"
+    assert task_list.items[0].task_id == "task-1"
     assert (await client.get_task("root", "task-1")).task_id == "task-1"
     assert (await client.list_attempts("root")).items[0].attempt_id == "attempt-1"
     assert (await client.get_attempt("root", "attempt-1")).attempt_id == "attempt-1"

agent/tests/test_orchestration_validation_policy.py

@@ -13,6 +13,7 @@ from agent.orchestration.evidence import (
     EvidenceQuery,
 )
 from agent.orchestration.models import (
+    AcceptanceCriterion,
     AgentRole,
     ArtifactRef,
     ArtifactSnapshot,
@@ -41,7 +42,11 @@ def make_context(root: str = "root") -> ValidationContext:
         goal_id=None,
         parent_task_id=None,
         display_path="1",
-        specs=[TaskSpec(version=1, objective="validate")],
+        specs=[TaskSpec(
+            version=1,
+            objective="validate",
+            acceptance_criteria=[AcceptanceCriterion("criterion", "validate result")],
+        )],
     )
     attempt = TaskAttempt(
         attempt_id="attempt-1",