重构(编排): 让显式任务账本摆脱 GoalTree 投影

删除 TaskLedger 到 GoalTree 的兼容投影、状态同步和修复入口，显式模式只保留 TaskLedger 这一份权威任务状态。

移除持久化 goal_id，同时保留仅构造期的旧 Host 参数兼容；新增紧凑的 TaskLedger 只读摘要，为后续上下文和可视化消费提供统一数据源。

agent/agent/orchestration/_goal_projection.py

@@ -1,175 +0,0 @@
-"""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_context.py

@@ -0,0 +1,42 @@
+"""Compact, non-authoritative context view of an explicit TaskLedger."""
+
+from __future__ import annotations
+
+from .models import TaskLedger
+from ._task_graph import path_key
+
+
+def render_task_context(ledger: TaskLedger) -> str:
+    """Render current task state for Planner refresh and context compression."""
+
+    tasks = sorted(ledger.tasks.values(), key=lambda task: path_key(task.display_path))
+    focused = ledger.tasks.get(ledger.focused_task_id or "")
+    lines = [
+        "## Current Task Plan",
+        "",
+        f"Root: {_one_line(ledger.root_objective)}",
+        "Focused: " + (
+            f"{focused.display_path} [{focused.status.value}] {_one_line(focused.current_spec.objective)}"
+            if focused
+            else "none"
+        ),
+        "",
+        "Tasks:",
+    ]
+    for task in tasks:
+        line = (
+            f"- {task.display_path} [{task.status.value}] "
+            f"{_one_line(task.current_spec.objective)}"
+        )
+        if task.blocked_reason:
+            line += f" (blocked: {_one_line(task.blocked_reason)})"
+        lines.append(line)
+    return "\n".join(lines)
+
+
+def _one_line(value: object, limit: int = 300) -> str:
+    text = " ".join(str(value).split())
+    return text if len(text) <= limit else f"{text[:limit - 1]}…"
+
+
+__all__ = ["render_task_context"]

agent/agent/orchestration/_task_graph.py

@@ -247,7 +247,6 @@ class TaskGraph:
         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],

agent/agent/orchestration/coordinator.py

@@ -70,8 +70,8 @@ from .validation_policy import (
     ValidationContext,
     ValidationPolicy,
 )
-from ._goal_projection import GoalProjection
 from ._decision_engine import DecisionEngine
+from ._task_context import render_task_context
 from ._task_graph import (
     TERMINAL_TASK_STATUSES,
     TaskGraph,
@@ -118,11 +118,6 @@ class TaskCoordinator:
             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,
@@ -186,7 +181,6 @@ class TaskCoordinator:
                     root_task_id = new_id()
                     root = TaskRecord(
                         task_id=root_task_id,
-                        goal_id=None,
                         parent_task_id=None,
                         display_path="0",
                         specs=[spec],
@@ -278,6 +272,13 @@ class TaskCoordinator:
             ],
         }
 
+    async def task_context(self, root_trace_id: str) -> str:
+        """Return a compact read model of the authoritative TaskLedger."""
+
+        ledger = await self.task_store.load(root_trace_id)
+        self._root_task(ledger)
+        return render_task_context(ledger)
+
     async def _mutate(
         self,
         root_trace_id: str,
@@ -462,15 +463,6 @@ class TaskCoordinator:
                 "placement": placement,
             },
         )
-        after_task_id = result.get("after_task_id")
-        for task_id in result["task_ids"]:
-            await self._project_goal_compatibility(
-                root_trace_id,
-                task_id,
-                ensure=True,
-                after_task_id=after_task_id,
-            )
-            after_task_id = task_id
         return await self._tasks_result(root_trace_id, result["task_ids"])
 
     async def focus_task(self, root_trace_id: str, task_id: str) -> Dict[str, Any]:
@@ -479,19 +471,7 @@ class TaskCoordinator:
             ledger.focused_task_id = task_id
             return {"task_id": task_id, "status": task.status.value}
 
-        result = await self._mutate(root_trace_id, "task_focused", mutate)
-        if self.trace_store:
-            try:
-                ledger = await self.task_store.load(root_trace_id)
-                task = ledger.tasks[task_id]
-                if task.goal_id:
-                    tree = await self.trace_store.get_goal_tree(root_trace_id)
-                    if tree and tree.find(task.goal_id):
-                        tree.focus(task.goal_id)
-                        await self.trace_store.update_goal_tree(root_trace_id, tree)
-            except Exception as exc:
-                self._log_projection_failure(root_trace_id, task_id, exc)
-        return result
+        return await self._mutate(root_trace_id, "task_focused", mutate)
 
     async def dispatch_tasks(
         self,
@@ -981,7 +961,6 @@ class TaskCoordinator:
             # persistent Store outage cannot be repaired in-process (V2), but
             # must not cancel sibling Tasks in this batch.
             return
-        await self._project_goal_compatibility(root_trace_id, task_id)
 
     async def _cycle_error_result(
         self,
@@ -1101,7 +1080,6 @@ class TaskCoordinator:
             },
             operation_id=operation_id,
         )
-        await self._project_goal_compatibility(root_trace_id, task_id)
         return result
 
     async def advance_cycle(
@@ -1756,7 +1734,6 @@ class TaskCoordinator:
             idempotency_payload=submission_payload,
             operation_id=attempt.operation_id,
         )
-        await self._project_goal_compatibility(root_trace_id, task_id)
         return result
 
     async def _start_validation(
@@ -1818,7 +1795,6 @@ class TaskCoordinator:
             mutate,
             operation_id=operation_id,
         )
-        await self._project_goal_compatibility(root_trace_id, task_id)
         return result["validation_id"]
 
     async def _plan_validation(
@@ -1957,7 +1933,6 @@ class TaskCoordinator:
             idempotency_payload=report_payload,
             operation_id=validation.operation_id,
         )
-        await self._project_goal_compatibility(root_trace_id, task_id)
         return result
 
     async def query_evidence(
@@ -2332,24 +2307,6 @@ class TaskCoordinator:
                 "payload": payload,
             },
         )
-        affected = [task_id]
-        affected.extend(result.get("payload", {}).get("child_task_ids", []))
-        replacement = result.get("payload", {}).get("replacement_task_id")
-        if replacement:
-            affected.append(replacement)
-        try:
-            ledger = await self.task_store.load(root_trace_id)
-            parent_id = ledger.tasks[task_id].parent_task_id
-            if parent_id:
-                affected.append(parent_id)
-        except Exception as exc:
-            self._log_projection_failure(root_trace_id, task_id, exc)
-        for affected_task in dict.fromkeys(affected):
-            await self._project_goal_compatibility(
-                root_trace_id,
-                affected_task,
-                ensure=True,
-            )
         return result
 
     @staticmethod
@@ -2455,7 +2412,6 @@ class TaskCoordinator:
             operation_id=operation_id,
         )
         validation_id = result["validation_id"]
-        await self._project_goal_compatibility(root_trace_id, task_id)
         return await self.advance_validation(
             root_trace_id,
             task_id,
@@ -2634,7 +2590,6 @@ class TaskCoordinator:
             return {"task_id": task_id, "attempt_id": attempt_id, "error": error}
 
         await self._mutate(root_trace_id, "worker_failed", mutate)
-        await self._project_goal_compatibility(root_trace_id, task_id)
 
     async def _mark_validation_error(
         self,
@@ -2676,7 +2631,6 @@ class TaskCoordinator:
             return {"task_id": task_id, "validation_id": validation_id, "error": error}
 
         await self._mutate(root_trace_id, "validation_error", mutate)
-        await self._project_goal_compatibility(root_trace_id, task_id)
 
     @staticmethod
     def _repair_feedback(ledger: TaskLedger, task: TaskRecord) -> Optional[Dict[str, Any]]:
@@ -2780,54 +2734,13 @@ class TaskCoordinator:
     def _update_parent_after_child(ledger: TaskLedger, child: TaskRecord) -> None:
         TaskGraph.update_parent_after_child(ledger, child)
 
-    async def project_goal_state(self, root_trace_id: str, task_id: str) -> None:
-        await self._goal_projection.project_state(root_trace_id, task_id)
-
-    async def _project_goal_compatibility(
-        self,
-        root_trace_id: str,
-        task_id: str,
-        *,
-        ensure: bool = False,
-        after_task_id: Optional[str] = None,
-    ) -> None:
-        await self._goal_projection.project_compatibility(
-            root_trace_id,
-            task_id,
-            ensure=ensure,
-            after_task_id=after_task_id,
-        )
-
-    @staticmethod
-    def _log_projection_failure(
-        root_trace_id: str,
-        task_id: str,
-        error: Exception,
-    ) -> None:
-        GoalProjection.log_failure(root_trace_id, task_id, error)
-
-    async def _ensure_goal_projection(
-        self,
-        root_trace_id: str,
-        task_id: str,
-        after_task_id: Optional[str] = None,
-    ) -> None:
-        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]:
-        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)
         return {
             "tasks": [
                 {
                     "task_id": task_id,
-                    "goal_id": ledger.tasks[task_id].goal_id,
+                    "goal_id": None,
                     "display_path": ledger.tasks[task_id].display_path,
                     "status": ledger.tasks[task_id].status.value,
                     "spec_version": ledger.tasks[task_id].current_spec_version,

agent/agent/orchestration/models.py

@@ -1,13 +1,12 @@
 """Domain models for explicit task execution and independent validation.
 
-The orchestration ledger is intentionally independent from GoalTree and the
-runtime implementation.  It is the source of truth for explicit-validation
-runs; GoalTree is only a compatibility projection.
+The orchestration ledger is the complete source of truth for
+``explicit_validation`` runs and has no dependency on legacy GoalTree state.
 """
 
 from __future__ import annotations
 
-from dataclasses import asdict, dataclass, field
+from dataclasses import InitVar, asdict, dataclass, field
 from datetime import datetime, timezone
 from enum import Enum
 from math import isfinite
@@ -331,10 +330,12 @@ class TaskSpec:
 @dataclass
 class TaskRecord:
     task_id: str
-    goal_id: Optional[str]
     parent_task_id: Optional[str]
     display_path: str
     specs: List[TaskSpec]
+    # Constructor-only compatibility for older Hosts. It is never stored in
+    # the record or persisted in TaskLedger JSON.
+    goal_id: InitVar[Optional[str]] = None
     current_spec_version: int = 1
     status: TaskStatus = TaskStatus.PENDING
     child_task_ids: List[str] = field(default_factory=list)
@@ -358,7 +359,6 @@ class TaskRecord:
     def from_dict(cls, data: Dict[str, Any]) -> "TaskRecord":
         return cls(
             task_id=data["task_id"],
-            goal_id=data.get("goal_id"),
             parent_task_id=data.get("parent_task_id"),
             display_path=data.get("display_path", ""),
             specs=[TaskSpec.from_dict(x) for x in data.get("specs", [])],

agent/agent/orchestration/wire.py

@@ -132,6 +132,8 @@ class ExecutionStatsView(WireModel):
 
 class TaskView(WireModel):
     task_id: str
+    # Deprecated compatibility field. Explicit ledgers no longer persist or
+    # project GoalTree identities, so new responses always expose ``None``.
     goal_id: Optional[str] = None
     parent_task_id: Optional[str] = None
     display_path: str