重构(编排): 拆分任务图决策与目标投影

保持 TaskCoordinator 全部公开方法和状态转换语义不变，将任务拓扑、Planner 决策校验和 Goal 投影拆为内部协作组件，并整理执行器与后台操作的边界。

agent/agent/orchestration/_decision_engine.py

@@ -0,0 +1,245 @@
+"""Pure Planner-decision rules for the explicit orchestration aggregate."""
+
+from __future__ import annotations
+
+from dataclasses import asdict
+from typing import Any, Callable, Dict, Optional
+
+from ._task_graph import ACTIVE_TASK_STATUSES, TERMINAL_TASK_STATUSES, TaskGraph
+from .errors import TaskConflict
+from .models import (
+    DecisionAction,
+    PlannerDecision,
+    TaskLedger,
+    TaskRecord,
+    TaskStatus,
+    ValidationReport,
+    ValidationRunStatus,
+    ValidationVerdict,
+    json_values,
+    new_id,
+    utc_now,
+)
+from .state_machine import transition
+
+
+class DecisionEngine:
+    """Apply a Planner decision to a caller-owned in-memory ledger."""
+
+    def __init__(
+        self,
+        task_graph: TaskGraph,
+        max_repair_continuations: Callable[[], int],
+    ) -> None:
+        self._task_graph = task_graph
+        self._max_repair_continuations = max_repair_continuations
+
+    def apply(
+        self,
+        ledger: TaskLedger,
+        *,
+        task_id: str,
+        validation_id: Optional[str],
+        action: DecisionAction,
+        payload: Dict[str, Any],
+    ) -> Dict[str, Any]:
+        task = self._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()
+        if not reason and action != DecisionAction.UNBLOCK:
+            raise ValueError("Planner decision reason is required")
+
+        if action == DecisionAction.ACCEPT:
+            self.guard_accept(task, validation, ledger)
+            target = TaskStatus.COMPLETED
+        elif action == DecisionAction.REPAIR:
+            self.guard_replan(task, validation, ledger)
+            count_key = str(task.current_spec_version)
+            used = task.repair_count_by_version.get(count_key, 0)
+            if used >= self._max_repair_continuations():
+                raise TaskConflict(
+                    "Repair continuation limit reached; retry, revise, split, or block"
+                )
+            task.repair_count_by_version[count_key] = used + 1
+            target = TaskStatus.PENDING
+        elif action == DecisionAction.RETRY:
+            self._require_replan_state(task, "Retry")
+            target = TaskStatus.PENDING
+        elif action == DecisionAction.REVISE:
+            self._require_replan_state(task, "Revise")
+            objective = payload.get("objective", task.current_spec.objective)
+            version = task.current_spec_version + 1
+            draft = {
+                "objective": objective,
+                "acceptance_criteria": payload.get(
+                    "acceptance_criteria",
+                    [
+                        json_values(asdict(item))
+                        for item in task.current_spec.acceptance_criteria
+                    ],
+                ),
+                "context_refs": payload.get(
+                    "context_refs",
+                    task.current_spec.context_refs,
+                ),
+            }
+            task.specs.append(TaskGraph.task_spec_from_draft(draft, version=version))
+            task.current_spec_version = version
+            target = TaskStatus.PENDING
+        elif action == DecisionAction.SPLIT:
+            self._require_replan_state(task, "Split")
+            drafts = payload.get("tasks") or []
+            if not drafts:
+                raise ValueError("Split requires payload.tasks")
+            child_ids = self._task_graph.create_child_records(ledger, task, drafts)
+            payload["child_task_ids"] = child_ids
+            target = TaskStatus.WAITING_CHILDREN
+        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 TaskGraph.nonterminal_descendants(ledger, task)
+            ):
+                raise TaskConflict("Task with active descendants cannot be blocked")
+            task.blocked_reason = reason
+            target = TaskStatus.BLOCKED
+        elif action == DecisionAction.UNBLOCK:
+            if task.status != TaskStatus.BLOCKED:
+                raise TaskConflict("Only a blocked task can be unblocked")
+            task.blocked_reason = None
+            target = TaskStatus.NEEDS_REPLAN
+        elif action == DecisionAction.CANCEL:
+            self._guard_replace_or_cancel(ledger, task, is_root, "cancelled")
+            target = TaskStatus.CANCELLED
+        elif action == DecisionAction.SUPERSEDE:
+            self._guard_replace_or_cancel(ledger, task, is_root, "superseded")
+            replacement = payload.get("replacement") or {}
+            replacement_ids = self._task_graph.create_sibling_records(
+                ledger,
+                task,
+                [replacement],
+            )
+            task.superseded_by = replacement_ids[0]
+            payload["replacement_task_id"] = replacement_ids[0]
+            target = TaskStatus.SUPERSEDED
+        elif action == DecisionAction.REVALIDATE:
+            raise ValueError("Use revalidate_attempt() / validate_attempt tool")
+        else:
+            raise ValueError(f"Unsupported decision action: {action.value}")
+
+        transition(before, target)
+        task.status = target
+        task.updated_at = utc_now()
+        decision = PlannerDecision(
+            decision_id=new_id(),
+            task_id=task_id,
+            action=action,
+            reason=reason,
+            from_status=before,
+            to_status=target,
+            attempt_id=attempt_id,
+            validation_id=validation_id,
+            payload=payload,
+        )
+        ledger.decisions[decision.decision_id] = decision
+        task.decision_ids.append(decision.decision_id)
+        TaskGraph.update_parent_after_child(ledger, task)
+        return {
+            "task_id": task_id,
+            "decision_id": decision.decision_id,
+            "action": action.value,
+            "status": target.value,
+            "payload": payload,
+        }
+
+    @staticmethod
+    def guard_accept(
+        task: TaskRecord,
+        validation: Optional[ValidationReport],
+        ledger: TaskLedger,
+    ) -> None:
+        if task.status != TaskStatus.AWAITING_DECISION:
+            raise TaskConflict("Accept requires awaiting_decision")
+        if not validation or validation.status != ValidationRunStatus.COMPLETED:
+            raise TaskConflict("Accept requires a completed validation")
+        if validation.spec_version != task.current_spec_version:
+            raise TaskConflict("Validation belongs to an obsolete TaskSpec version")
+        if validation.verdict != ValidationVerdict.PASSED:
+            raise TaskConflict("Only a passed validation can be accepted")
+        if not task.attempt_ids or validation.attempt_id != task.attempt_ids[-1]:
+            raise TaskConflict("Validation does not belong to the current attempt")
+        attempt = ledger.attempts[validation.attempt_id]
+        if validation.snapshot_id != attempt.snapshot_id:
+            raise TaskConflict("Validation does not belong to the current snapshot")
+        if TaskGraph.nonterminal_descendants(ledger, task):
+            raise TaskConflict(
+                "Task cannot be accepted while descendants are non-terminal"
+            )
+
+    @staticmethod
+    def guard_replan(
+        task: TaskRecord,
+        validation: Optional[ValidationReport],
+        ledger: TaskLedger,
+    ) -> None:
+        if task.status != TaskStatus.AWAITING_DECISION:
+            raise TaskConflict("Repair requires awaiting_decision")
+        if not validation or validation.status != ValidationRunStatus.COMPLETED:
+            raise TaskConflict("Repair requires a completed validation")
+        if validation.spec_version != task.current_spec_version:
+            raise TaskConflict(
+                "Repair validation belongs to an obsolete TaskSpec version"
+            )
+        if not task.attempt_ids or validation.attempt_id != task.attempt_ids[-1]:
+            raise TaskConflict(
+                "Repair validation does not belong to the current attempt"
+            )
+        attempt = ledger.attempts[validation.attempt_id]
+        if validation.snapshot_id != attempt.snapshot_id:
+            raise TaskConflict(
+                "Repair validation does not belong to the current snapshot"
+            )
+        if validation.verdict == ValidationVerdict.PASSED:
+            raise TaskConflict("Passed work should be accepted, not repaired")
+
+    @staticmethod
+    def _task(ledger: TaskLedger, task_id: str) -> TaskRecord:
+        try:
+            return ledger.tasks[task_id]
+        except KeyError as exc:
+            raise ValueError(f"Task not found: {task_id}") from exc
+
+    @staticmethod
+    def _require_replan_state(task: TaskRecord, action: str) -> None:
+        if task.status not in (
+            TaskStatus.AWAITING_DECISION,
+            TaskStatus.NEEDS_REPLAN,
+        ):
+            raise TaskConflict(f"{action} requires awaiting_decision or needs_replan")
+
+    @staticmethod
+    def _guard_replace_or_cancel(
+        ledger: TaskLedger,
+        task: TaskRecord,
+        is_root: bool,
+        action: str,
+    ) -> None:
+        if is_root:
+            raise TaskConflict(f"The root task cannot be {action}")
+        if task.status in TERMINAL_TASK_STATUSES:
+            raise TaskConflict("Task is already terminal")
+        if TaskGraph.nonterminal_descendants(ledger, task):
+            raise TaskConflict(f"Task with non-terminal descendants cannot be {action}")
+
+
+__all__ = ["DecisionEngine"]

agent/agent/orchestration/_goal_projection.py

@@ -0,0 +1,175 @@
+"""Best-effort TaskLedger-to-GoalTree compatibility projection."""
+
+from __future__ import annotations
+
+import logging
+from typing import Any, Awaitable, Callable, Dict, Optional
+
+from agent.trace.goal_models import GoalTree
+
+from .models import TaskLedger, TaskRecord, TaskStatus
+from .protocols import TaskStore
+
+
+logger = logging.getLogger(__name__)
+
+
+GOAL_STATUS_BY_TASK_STATUS = {
+    TaskStatus.PENDING: "pending",
+    TaskStatus.RUNNING: "in_progress",
+    TaskStatus.AWAITING_VALIDATION: "in_progress",
+    TaskStatus.VALIDATING: "in_progress",
+    TaskStatus.AWAITING_DECISION: "in_progress",
+    TaskStatus.NEEDS_REPLAN: "in_progress",
+    TaskStatus.WAITING_CHILDREN: "in_progress",
+    TaskStatus.BLOCKED: "in_progress",
+    TaskStatus.COMPLETED: "completed",
+    TaskStatus.CANCELLED: "abandoned",
+    TaskStatus.SUPERSEDED: "abandoned",
+}
+
+
+class GoalProjection:
+    """Project authoritative task state into the legacy GoalTree view.
+
+    The projection never commits a ledger directly. Linking a projected Goal
+    back to its Task is delegated to the coordinator's mutation callback so the
+    coordinator remains the single persistence boundary.
+    """
+
+    def __init__(
+        self,
+        *,
+        task_store: TaskStore,
+        get_trace_store: Callable[[], Any],
+        mutate: Callable[..., Awaitable[Dict[str, Any]]],
+    ) -> None:
+        self._task_store = task_store
+        self._get_trace_store = get_trace_store
+        self._mutate = mutate
+
+    async def project_state(self, root_trace_id: str, task_id: str) -> None:
+        trace_store = self._get_trace_store()
+        if not trace_store:
+            return
+        ledger = await self._task_store.load(root_trace_id)
+        task = _task(ledger, task_id)
+        if not task.goal_id:
+            return
+        summary = None
+        if task.decision_ids:
+            summary = ledger.decisions[task.decision_ids[-1]].reason
+        await trace_store.update_goal(
+            root_trace_id,
+            task.goal_id,
+            cascade_completion=False,
+            status=GOAL_STATUS_BY_TASK_STATUS[task.status],
+            summary=summary,
+        )
+
+    async def project_compatibility(
+        self,
+        root_trace_id: str,
+        task_id: str,
+        *,
+        ensure: bool = False,
+        after_task_id: Optional[str] = None,
+    ) -> None:
+        """Run projection without invalidating an authoritative Ledger commit."""
+
+        if not self._get_trace_store():
+            return
+        try:
+            if ensure:
+                await self.ensure(
+                    root_trace_id,
+                    task_id,
+                    after_task_id=after_task_id,
+                )
+            await self.project_state(root_trace_id, task_id)
+        except Exception as exc:
+            self.log_failure(root_trace_id, task_id, exc)
+
+    @staticmethod
+    def log_failure(
+        root_trace_id: str,
+        task_id: str,
+        error: Exception,
+    ) -> None:
+        logger.warning(
+            "Goal compatibility projection failed after Ledger commit "
+            "(%s, %s): %s",
+            root_trace_id,
+            task_id,
+            error,
+        )
+
+    async def ensure(
+        self,
+        root_trace_id: str,
+        task_id: str,
+        after_task_id: Optional[str] = None,
+    ) -> None:
+        trace_store = self._get_trace_store()
+        if not trace_store:
+            return
+        ledger = await self._task_store.load(root_trace_id)
+        task = _task(ledger, task_id)
+        tree = await trace_store.get_goal_tree(root_trace_id)
+        if tree is None:
+            tree = GoalTree(mission=ledger.root_objective)
+        if task.goal_id and tree.find(task.goal_id):
+            return
+        parent_goal_id = None
+        if task.parent_task_id:
+            parent = ledger.tasks[task.parent_task_id]
+            if not parent.goal_id:
+                await self.ensure(root_trace_id, parent.task_id)
+                ledger = await self._task_store.load(root_trace_id)
+                parent = ledger.tasks[task.parent_task_id]
+                tree = await trace_store.get_goal_tree(root_trace_id) or tree
+            parent_goal_id = parent.goal_id
+        after_goal_id = None
+        if after_task_id and after_task_id in ledger.tasks:
+            after_goal_id = ledger.tasks[after_task_id].goal_id
+        if after_goal_id and tree.find(after_goal_id):
+            goal = tree.add_goals_after(
+                after_goal_id,
+                descriptions=[task.current_spec.objective],
+                reasons=["TaskLedger compatibility projection"],
+            )[0]
+        else:
+            goal = tree.add_goals(
+                descriptions=[task.current_spec.objective],
+                reasons=["TaskLedger compatibility projection"],
+                parent_id=parent_goal_id,
+            )[0]
+        await trace_store.update_goal_tree(root_trace_id, tree)
+
+        def link(current: TaskLedger) -> Dict[str, Any]:
+            current_task = _task(current, task_id)
+            if current_task.goal_id is None:
+                current_task.goal_id = goal.id
+            return {"task_id": task_id, "goal_id": current_task.goal_id}
+
+        await self._mutate(root_trace_id, "goal_projection_linked", link)
+
+    async def reconcile(self, root_trace_id: str) -> Dict[str, Any]:
+        ledger = await self._task_store.load(root_trace_id)
+        for task_id in ledger.tasks:
+            await self.ensure(root_trace_id, task_id)
+            await self.project_state(root_trace_id, task_id)
+        return {
+            "root_trace_id": root_trace_id,
+            "reconciled_tasks": len(ledger.tasks),
+        }
+
+
+def _task(ledger: TaskLedger, task_id: str) -> TaskRecord:
+    try:
+        return ledger.tasks[task_id]
+    except KeyError as exc:
+        raise ValueError(f"Task not found: {task_id}") from exc
+
+
+__all__ = ["GOAL_STATUS_BY_TASK_STATUS", "GoalProjection"]

agent/agent/orchestration/_task_graph.py

@@ -0,0 +1,353 @@
+"""Pure task-tree operations used by :mod:`agent.orchestration.coordinator`.
+
+The task ledger remains owned and committed by ``TaskCoordinator``.  This
+module only mutates the in-memory aggregate handed to it; it never performs
+I/O, acquires locks, or commits state on its own.
+"""
+
+from __future__ import annotations
+
+from dataclasses import asdict
+from typing import Any, Dict, List, Optional, Sequence, Tuple
+
+from .errors import OrchestrationError, TaskConflict
+from .models import (
+    DecisionAction,
+    PlannerDecision,
+    TaskAttempt,
+    TaskLedger,
+    TaskRecord,
+    TaskSpec,
+    TaskStatus,
+    ValidationReport,
+    ValidationRunStatus,
+    ValidationVerdict,
+    json_values,
+    new_id,
+    utc_now,
+)
+from .state_machine import transition
+
+
+TERMINAL_TASK_STATUSES = {
+    TaskStatus.COMPLETED,
+    TaskStatus.CANCELLED,
+    TaskStatus.SUPERSEDED,
+}
+
+ACTIVE_TASK_STATUSES = {
+    TaskStatus.RUNNING,
+    TaskStatus.AWAITING_VALIDATION,
+    TaskStatus.VALIDATING,
+}
+
+
+def path_key(display_path: str) -> Tuple[int, ...]:
+    """Return a stable numeric sort key for a dotted task display path."""
+
+    try:
+        return tuple(int(part) for part in display_path.split("."))
+    except ValueError:
+        return (10**9,)
+
+
+class TaskGraph:
+    """Maintain task hierarchy and accepted child-result bindings.
+
+    All methods operate on a caller-owned ``TaskLedger``.  Keeping persistence
+    out of this component preserves ``TaskCoordinator`` as the sole writer.
+    """
+
+    @staticmethod
+    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]] = []
+        if attempt is None:
+            decision_ids = TaskGraph.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 = TaskGraph.accepted_child_decision_ids(ledger, task)
+        if tuple(decision_ids) != expected_decision_ids:
+            raise OrchestrationError(
+                "Attempt child decision bindings do not match direct children in "
+                "display_path order"
+            )
+        direct_children = set(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]
+            child_path = path_key(child.display_path)
+            if prior_path is not None and child_path < prior_path:
+                raise OrchestrationError(
+                    "Attempt child decision bindings are not in display_path order"
+                )
+            prior_path = child_path
+            bound_decision, child_attempt, validation = TaskGraph.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": json_values(asdict(child.current_spec)),
+                    "attempt_id": child_attempt.attempt_id,
+                    "snapshot_id": child_attempt.snapshot_id,
+                    "submission": json_values(asdict(child_attempt.submission)),
+                    "validation": {
+                        "validation_id": validation.validation_id,
+                        "summary": validation.summary,
+                        "evidence_refs": json_values(
+                            asdict(validation)["evidence_refs"]
+                        ),
+                    },
+                }
+            )
+        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 = TaskGraph.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:
+            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,
+        parent: TaskRecord,
+        drafts: Sequence[Dict[str, Any]],
+    ) -> List[str]:
+        return self.create_records(ledger, drafts, parent.task_id, parent.display_path)
+
+    def create_sibling_records(
+        self,
+        ledger: TaskLedger,
+        task: TaskRecord,
+        drafts: Sequence[Dict[str, Any]],
+    ) -> List[str]:
+        parent_path = (
+            ledger.tasks[task.parent_task_id].display_path
+            if task.parent_task_id
+            else ""
+        )
+        return self.create_records(ledger, drafts, task.parent_task_id, parent_path)
+
+    @staticmethod
+    def add_task_record(
+        ledger: TaskLedger,
+        draft: Dict[str, Any],
+        parent_task_id: Optional[str],
+        display_path: str,
+    ) -> str:
+        task_id = new_id()
+        spec = TaskGraph.task_spec_from_draft(draft, version=1)
+        ledger.tasks[task_id] = TaskRecord(
+            task_id=task_id,
+            goal_id=None,
+            parent_task_id=parent_task_id,
+            display_path=display_path,
+            specs=[spec],
+        )
+        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,
+        drafts: Sequence[Dict[str, Any]],
+        parent_task_id: Optional[str],
+        parent_path: str,
+    ) -> List[str]:
+        siblings = sorted(
+            (
+                task
+                for task in ledger.tasks.values()
+                if task.parent_task_id == parent_task_id
+            ),
+            key=lambda item: path_key(item.display_path),
+        )
+        created: List[str] = []
+        for offset, draft in enumerate(drafts, start=1):
+            index = len(siblings) + offset
+            display_path = f"{parent_path}.{index}" if parent_path else str(index)
+            task_id = self.add_task_record(
+                ledger,
+                draft,
+                parent_task_id,
+                display_path,
+            )
+            if parent_task_id:
+                ledger.tasks[parent_task_id].child_task_ids.append(task_id)
+            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[child_id] for child_id in parent.child_task_ids]
+        if children and all(
+            candidate.status in TERMINAL_TASK_STATUSES for candidate in children
+        ):
+            transition(parent.status, TaskStatus.NEEDS_REPLAN)
+            parent.status = TaskStatus.NEEDS_REPLAN
+        parent.updated_at = utc_now()
+
+    @staticmethod
+    def display_path(
+        ledger: TaskLedger,
+        parent_task_id: Optional[str],
+        index: int,
+    ) -> str:
+        if not parent_task_id:
+            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):
+            TaskGraph.rebase_task_path(
+                ledger,
+                ledger.tasks[child_id],
+                f"{display_path}.{index}",
+            )
+
+
+__all__ = [
+    "ACTIVE_TASK_STATUSES",
+    "TERMINAL_TASK_STATUSES",
+    "TaskGraph",
+    "path_key",
+]

agent/agent/orchestration/coordinator.py

@@ -11,8 +11,6 @@ from dataclasses import asdict, is_dataclass
 from typing import Any, Callable, Dict, Iterable, List, Optional, Sequence, Tuple
 from uuid import uuid4
 
-from agent.trace.goal_models import GoalTree
-
 from .config import OrchestrationConfig
 from .models import (
     AgentRole,
@@ -55,7 +53,7 @@ from .protocols import (
 from .state_machine import transition
 from .store import RevisionConflict, TaskStoreNotFound
 from .errors import OrchestrationError, TaskConflict
-from .operations import OperationController, stage_timeout
+from .operations import OperationController, _failure_stats, stage_timeout
 from .evidence import (
     EvidenceBudgetExceeded,
     EvidenceOwnershipError,
@@ -72,24 +70,18 @@ from .validation_policy import (
     ValidationContext,
     ValidationPolicy,
 )
+from ._goal_projection import GoalProjection
+from ._decision_engine import DecisionEngine
+from ._task_graph import (
+    TERMINAL_TASK_STATUSES,
+    TaskGraph,
+    path_key as _path_key,
+)
 
 
 logger = logging.getLogger(__name__)
 
 
-TERMINAL_TASK_STATUSES = {
-    TaskStatus.COMPLETED,
-    TaskStatus.CANCELLED,
-    TaskStatus.SUPERSEDED,
-}
-
-ACTIVE_TASK_STATUSES = {
-    TaskStatus.RUNNING,
-    TaskStatus.AWAITING_VALIDATION,
-    TaskStatus.VALIDATING,
-}
-
-
 class TaskCoordinator:
     """Coordinates Task -> Attempt -> Validation -> Planner Decision.
 
@@ -121,6 +113,16 @@ class TaskCoordinator:
         self.deterministic_validator = deterministic_validator
         self.evidence_provider = evidence_provider
         self._locks: Dict[str, asyncio.Lock] = {}
+        self._task_graph = TaskGraph()
+        self._decision_engine = DecisionEngine(
+            self._task_graph,
+            max_repair_continuations=lambda: self.config.max_repair_continuations,
+        )
+        self._goal_projection = GoalProjection(
+            task_store=self.task_store,
+            get_trace_store=lambda: self.trace_store,
+            mutate=self._mutate,
+        )
         self._operations = OperationController(
             task_store=self.task_store,
             mutate=self._mutate,
@@ -2239,126 +2241,13 @@ class TaskCoordinator:
         payload = dict(payload or {})
 
         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()
-            if not reason and action not in (DecisionAction.UNBLOCK,):
-                raise ValueError("Planner decision reason is required")
-
-            if action == DecisionAction.ACCEPT:
-                self._guard_accept(task, validation, ledger)
-                target = TaskStatus.COMPLETED
-            elif action == DecisionAction.REPAIR:
-                self._guard_replan_decision(task, validation, ledger)
-                count_key = str(task.current_spec_version)
-                used = task.repair_count_by_version.get(count_key, 0)
-                if used >= self.config.max_repair_continuations:
-                    raise TaskConflict("Repair continuation limit reached; retry, revise, split, or block")
-                task.repair_count_by_version[count_key] = used + 1
-                target = TaskStatus.PENDING
-            elif action == DecisionAction.RETRY:
-                if task.status not in (TaskStatus.AWAITING_DECISION, TaskStatus.NEEDS_REPLAN):
-                    raise TaskConflict("Retry requires awaiting_decision or needs_replan")
-                target = TaskStatus.PENDING
-            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 = (
-                    payload["objective"]
-                    if "objective" in payload
-                    else task.current_spec.objective
-                )
-                version = task.current_spec_version + 1
-                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:
-                if task.status not in (TaskStatus.AWAITING_DECISION, TaskStatus.NEEDS_REPLAN):
-                    raise TaskConflict("Split requires awaiting_decision or needs_replan")
-                drafts = payload.get("tasks") or []
-                if not drafts:
-                    raise ValueError("Split requires payload.tasks")
-                child_ids = self._create_child_records(ledger, task, drafts)
-                payload["child_task_ids"] = child_ids
-                target = TaskStatus.WAITING_CHILDREN
-            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:
-                if task.status != TaskStatus.BLOCKED:
-                    raise TaskConflict("Only a blocked task can be unblocked")
-                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]
-                payload["replacement_task_id"] = replacement_ids[0]
-                target = TaskStatus.SUPERSEDED
-            elif action == DecisionAction.REVALIDATE:
-                raise ValueError("Use revalidate_attempt() / validate_attempt tool")
-            else:
-                raise ValueError(f"Unsupported decision action: {action.value}")
-
-            transition(before, target)
-            task.status = target
-            task.updated_at = utc_now()
-            decision = PlannerDecision(
-                decision_id=new_id(),
+            return self._decision_engine.apply(
+                ledger,
                 task_id=task_id,
-                action=action,
-                reason=reason,
-                from_status=before,
-                to_status=target,
-                attempt_id=attempt_id,
                 validation_id=validation_id,
+                action=action,
                 payload=payload,
             )
-            ledger.decisions[decision.decision_id] = decision
-            task.decision_ids.append(decision.decision_id)
-            self._update_parent_after_child(ledger, task)
-            return {
-                "task_id": task_id,
-                "decision_id": decision.decision_id,
-                "action": action.value,
-                "status": target.value,
-                "payload": payload,
-            }
 
         result = await self._mutate(
             root_trace_id,
@@ -2398,21 +2287,7 @@ class TaskCoordinator:
         validation: Optional[ValidationReport],
         ledger: TaskLedger,
     ) -> None:
-        if task.status != TaskStatus.AWAITING_DECISION:
-            raise TaskConflict("Accept requires awaiting_decision")
-        if not validation or validation.status != ValidationRunStatus.COMPLETED:
-            raise TaskConflict("Accept requires a completed validation")
-        if validation.spec_version != task.current_spec_version:
-            raise TaskConflict("Validation belongs to an obsolete TaskSpec version")
-        if validation.verdict != ValidationVerdict.PASSED:
-            raise TaskConflict("Only a passed validation can be accepted")
-        if not task.attempt_ids or validation.attempt_id != task.attempt_ids[-1]:
-            raise TaskConflict("Validation does not belong to the current attempt")
-        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")
+        DecisionEngine.guard_accept(task, validation, ledger)
 
     @staticmethod
     def _guard_replan_decision(
@@ -2420,18 +2295,7 @@ class TaskCoordinator:
         validation: Optional[ValidationReport],
         ledger: TaskLedger,
     ) -> None:
-        if task.status != TaskStatus.AWAITING_DECISION:
-            raise TaskConflict("Repair requires awaiting_decision")
-        if not validation or validation.status != ValidationRunStatus.COMPLETED:
-            raise TaskConflict("Repair requires a completed validation")
-        if validation.spec_version != task.current_spec_version:
-            raise TaskConflict("Repair validation belongs to an obsolete TaskSpec version")
-        if not task.attempt_ids or validation.attempt_id != task.attempt_ids[-1]:
-            raise TaskConflict("Repair validation does not belong to the current attempt")
-        if validation.snapshot_id != ledger.attempts[validation.attempt_id].snapshot_id:
-            raise TaskConflict("Repair validation does not belong to the current snapshot")
-        if validation.verdict == ValidationVerdict.PASSED:
-            raise TaskConflict("Passed work should be accepted, not repaired")
+        DecisionEngine.guard_replan(task, validation, ledger)
 
     async def revalidate_attempt(
         self,
@@ -2762,145 +2626,32 @@ class TaskCoordinator:
         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]] = []
-        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
-        )
-        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": 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,
-                    "evidence_refs": _plain(asdict(validation)["evidence_refs"]),
-                },
-            })
-        return results
+        return TaskGraph.accepted_child_results(ledger, task, attempt)
 
     @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)
+        return TaskGraph.accepted_child_decision_ids(ledger, task)
 
     @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
+        return TaskGraph.root_task(ledger)
 
     @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
+        return TaskGraph.accepted_result_binding(ledger, task)
 
     @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
+        return TaskGraph.nonterminal_descendants(ledger, task)
 
     def _create_child_records(
         self,
@@ -2908,7 +2659,7 @@ class TaskCoordinator:
         parent: TaskRecord,
         drafts: Sequence[Dict[str, Any]],
     ) -> List[str]:
-        return self._create_records(ledger, drafts, parent.task_id, parent.display_path)
+        return self._task_graph.create_child_records(ledger, parent, drafts)
 
     def _create_sibling_records(
         self,
@@ -2916,8 +2667,7 @@ class TaskCoordinator:
         task: TaskRecord,
         drafts: Sequence[Dict[str, Any]],
     ) -> List[str]:
-        parent_path = ledger.tasks[task.parent_task_id].display_path if task.parent_task_id else ""
-        return self._create_records(ledger, drafts, task.parent_task_id, parent_path)
+        return self._task_graph.create_sibling_records(ledger, task, drafts)
 
     @staticmethod
     def _add_task_record(
@@ -2926,27 +2676,16 @@ class TaskCoordinator:
         parent_task_id: Optional[str],
         display_path: str,
     ) -> str:
-        task_id = new_id()
-        spec = TaskCoordinator._task_spec_from_draft(draft, version=1)
-        ledger.tasks[task_id] = TaskRecord(
-            task_id=task_id,
-            goal_id=None,
-            parent_task_id=parent_task_id,
-            display_path=display_path,
-            specs=[spec],
+        return TaskGraph.add_task_record(
+            ledger,
+            draft,
+            parent_task_id,
+            display_path,
         )
-        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", ()),
-        })
+        return TaskGraph.task_spec_from_draft(draft, version=version)
 
     def _create_records(
         self,
@@ -2955,76 +2694,23 @@ class TaskCoordinator:
         parent_task_id: Optional[str],
         parent_path: str,
     ) -> List[str]:
-        siblings = sorted(
-            (t for t in ledger.tasks.values() if t.parent_task_id == parent_task_id),
-            key=lambda item: _path_key(item.display_path),
+        return self._task_graph.create_records(
+            ledger,
+            drafts,
+            parent_task_id,
+            parent_path,
         )
-        created: List[str] = []
-        for offset, draft in enumerate(drafts, start=1):
-            index = len(siblings) + offset
-            display_path = f"{parent_path}.{index}" if parent_path else str(index)
-            task_id = self._add_task_record(
-                ledger,
-                draft,
-                parent_task_id,
-                display_path,
-            )
-            if parent_task_id:
-                ledger.tasks[parent_task_id].child_task_ids.append(task_id)
-            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()
+        TaskGraph.mark_parent_waiting(parent)
 
     @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 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()
+        TaskGraph.update_parent_after_child(ledger, child)
 
     async def project_goal_state(self, root_trace_id: str, task_id: str) -> None:
-        if not self.trace_store:
-            return
-        ledger = await self.task_store.load(root_trace_id)
-        task = _task(ledger, task_id)
-        if not task.goal_id:
-            return
-        status_map = {
-            TaskStatus.PENDING: "pending",
-            TaskStatus.RUNNING: "in_progress",
-            TaskStatus.AWAITING_VALIDATION: "in_progress",
-            TaskStatus.VALIDATING: "in_progress",
-            TaskStatus.AWAITING_DECISION: "in_progress",
-            TaskStatus.NEEDS_REPLAN: "in_progress",
-            TaskStatus.WAITING_CHILDREN: "in_progress",
-            TaskStatus.BLOCKED: "in_progress",
-            TaskStatus.COMPLETED: "completed",
-            TaskStatus.CANCELLED: "abandoned",
-            TaskStatus.SUPERSEDED: "abandoned",
-        }
-        summary = None
-        if task.decision_ids:
-            summary = ledger.decisions[task.decision_ids[-1]].reason
-        await self.trace_store.update_goal(
-            root_trace_id,
-            task.goal_id,
-            cascade_completion=False,
-            status=status_map[task.status],
-            summary=summary,
-        )
+        await self._goal_projection.project_state(root_trace_id, task_id)
 
     async def _project_goal_compatibility(
         self,
@@ -3034,25 +2720,12 @@ class TaskCoordinator:
         ensure: bool = False,
         after_task_id: Optional[str] = None,
     ) -> None:
-        """Best-effort GoalTree projection after an authoritative commit.
-
-        Automatic projection must never turn a successful Ledger mutation into
-        an apparent command failure. Explicit reconcile_goal_tree() deliberately
-        keeps using the strict helpers so operators can observe repair failures.
-        """
-
-        if not self.trace_store:
-            return
-        try:
-            if ensure:
-                await self._ensure_goal_projection(
-                    root_trace_id,
-                    task_id,
-                    after_task_id=after_task_id,
-                )
-            await self.project_goal_state(root_trace_id, task_id)
-        except Exception as exc:
-            self._log_projection_failure(root_trace_id, task_id, exc)
+        await self._goal_projection.project_compatibility(
+            root_trace_id,
+            task_id,
+            ensure=ensure,
+            after_task_id=after_task_id,
+        )
 
     @staticmethod
     def _log_projection_failure(
@@ -3060,13 +2733,7 @@ class TaskCoordinator:
         task_id: str,
         error: Exception,
     ) -> None:
-        logger.warning(
-            "Goal compatibility projection failed after Ledger commit "
-            "(%s, %s): %s",
-            root_trace_id,
-            task_id,
-            error,
-        )
+        GoalProjection.log_failure(root_trace_id, task_id, error)
 
     async def _ensure_goal_projection(
         self,
@@ -3074,55 +2741,14 @@ class TaskCoordinator:
         task_id: str,
         after_task_id: Optional[str] = None,
     ) -> None:
-        if not self.trace_store:
-            return
-        ledger = await self.task_store.load(root_trace_id)
-        task = _task(ledger, task_id)
-        tree = await self.trace_store.get_goal_tree(root_trace_id)
-        if tree is None:
-            tree = GoalTree(mission=ledger.root_objective)
-        if task.goal_id and tree.find(task.goal_id):
-            return
-        parent_goal_id = None
-        if task.parent_task_id:
-            parent = ledger.tasks[task.parent_task_id]
-            if not parent.goal_id:
-                await self._ensure_goal_projection(root_trace_id, parent.task_id)
-                ledger = await self.task_store.load(root_trace_id)
-                parent = ledger.tasks[task.parent_task_id]
-                tree = await self.trace_store.get_goal_tree(root_trace_id) or tree
-            parent_goal_id = parent.goal_id
-        after_goal_id = None
-        if after_task_id and after_task_id in ledger.tasks:
-            after_goal_id = ledger.tasks[after_task_id].goal_id
-        if after_goal_id and tree.find(after_goal_id):
-            goal = tree.add_goals_after(
-                after_goal_id,
-                descriptions=[task.current_spec.objective],
-                reasons=["TaskLedger compatibility projection"],
-            )[0]
-        else:
-            goal = tree.add_goals(
-                descriptions=[task.current_spec.objective],
-                reasons=["TaskLedger compatibility projection"],
-                parent_id=parent_goal_id,
-            )[0]
-        await self.trace_store.update_goal_tree(root_trace_id, tree)
-
-        def link(current: TaskLedger) -> Dict[str, Any]:
-            current_task = _task(current, task_id)
-            if current_task.goal_id is None:
-                current_task.goal_id = goal.id
-            return {"task_id": task_id, "goal_id": current_task.goal_id}
-
-        await self._mutate(root_trace_id, "goal_projection_linked", link)
+        await self._goal_projection.ensure(
+            root_trace_id,
+            task_id,
+            after_task_id=after_task_id,
+        )
 
     async def reconcile_goal_tree(self, root_trace_id: str) -> Dict[str, Any]:
-        ledger = await self.task_store.load(root_trace_id)
-        for task_id in ledger.tasks:
-            await self._ensure_goal_projection(root_trace_id, task_id)
-            await self.project_goal_state(root_trace_id, task_id)
-        return {"root_trace_id": root_trace_id, "reconciled_tasks": len(ledger.tasks)}
+        return await self._goal_projection.reconcile(root_trace_id)
 
     async def _tasks_result(self, root_trace_id: str, task_ids: Iterable[str]) -> Dict[str, Any]:
         ledger = await self.task_store.load(root_trace_id)
@@ -3142,9 +2768,7 @@ class TaskCoordinator:
 
     @staticmethod
     def _display_path(ledger: TaskLedger, parent_task_id: Optional[str], index: int) -> str:
-        if not parent_task_id:
-            return str(index)
-        return f"{ledger.tasks[parent_task_id].display_path}.{index}"
+        return TaskGraph.display_path(ledger, parent_task_id, index)
 
     @staticmethod
     def _rebase_task_path(
@@ -3152,13 +2776,7 @@ class TaskCoordinator:
         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}",
-            )
+        TaskGraph.rebase_task_path(ledger, task, display_path)
 
 
 def _task(ledger: TaskLedger, task_id: str) -> TaskRecord:
@@ -3275,24 +2893,4 @@ def _failure_code(run_status: str, *, protocol_failure: bool) -> FailureCode:
     }.get(run_status, FailureCode.PROTOCOL_VIOLATION)
 
 
-def _failure_stats(
-    stats: Optional[ExecutionStats],
-    fallback: FailureCode,
-    *,
-    override: bool = False,
-) -> ExecutionStats:
-    if stats is None:
-        return ExecutionStats(failure_code=fallback)
-    if stats.failure_code is not None and not override:
-        return stats
-    return replace(stats, failure_code=fallback)
-
-
-def _path_key(display_path: str) -> Tuple[int, ...]:
-    try:
-        return tuple(int(part) for part in display_path.split("."))
-    except ValueError:
-        return (10**9,)
-
-
 __all__ = ["TaskCoordinator", "OrchestrationError", "TaskConflict"]

agent/agent/orchestration/executor.py

@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 import json
+import logging
 from collections.abc import Mapping
 from copy import deepcopy
 from dataclasses import replace
@@ -11,7 +12,7 @@ from types import MappingProxyType
 from typing import Any, Dict, Iterable, Optional, Tuple
 
 from agent.core.runner import RunConfig
-from agent.tools.builtin.knowledge import KnowledgeConfig
+from agent.core.knowledge_config import KnowledgeConfig
 
 from .models import AgentRole, CompletionPolicy, ExecutionStats, FailureCode
 from .protocols import ValidatorRunResult, WorkerRunResult
@@ -23,6 +24,9 @@ from .run_config import (
 )
 
 
+logger = logging.getLogger(__name__)
+
+
 class LocalAgentExecutor:
     """Run workers and validators with the same Runner in isolated traces."""
 
@@ -298,6 +302,7 @@ class LocalAgentExecutor:
         try:
             return await trace_store.get_trace(trace_id)
         except Exception:
+            logger.debug("Unable to reload local sub-trace %s", trace_id, exc_info=True)
             return None
 
     @staticmethod

agent/agent/orchestration/operations.py

@@ -785,10 +785,12 @@ def stage_timeout(
 def _failure_stats(
     stats: Optional[ExecutionStats],
     fallback: FailureCode,
+    *,
+    override: bool = False,
 ) -> ExecutionStats:
     if stats is None:
         return ExecutionStats(failure_code=fallback)
-    if stats.failure_code is not None:
+    if stats.failure_code is not None and not override:
         return stats
     return replace(stats, failure_code=fallback)