测试(编排): 用 TaskLedger 生命周期替代 Goal 投影断言

删除投影故障与重建测试，改为验证显式任务从创建、派发到等待决策的完整过程中不会产生 GoalTree。

补充 TaskLedger 当前计划摘要断言，并同步 Root 修订、领域模型和 V2 接口测试，使测试语义与单一权威状态一致。

agent/tests/test_coordinator_integration.py

@@ -18,7 +18,6 @@ from agent.orchestration.models import (
 )
 from agent.orchestration.protocols import ValidatorRunResult, WorkerRunResult
 from agent.orchestration.store import FileSystemArtifactStore, FileSystemTaskStore, TraceEventSink
-from agent.trace.goal_models import GoalTree
 from agent.trace.models import Trace
 from agent.trace.store import FileSystemTraceStore
 from agent.core.runner import AgentRunner
@@ -104,7 +103,6 @@ async def make_coordinator(
 ):
     trace_store = trace_store or FileSystemTraceStore(str(tmp_path))
     await trace_store.create_trace(Trace(trace_id="root", mode="agent", task="mission", agent_role="planner"))
-    await trace_store.update_goal_tree("root", GoalTree(mission="mission"))
     task_store = task_store or FileSystemTaskStore(str(tmp_path))
     coordinator = TaskCoordinator(
         task_store,
@@ -448,7 +446,6 @@ async def test_real_runner_local_executor_creates_independent_terminal_traces(tm
 
     trace_store = FileSystemTraceStore(str(tmp_path))
     await trace_store.create_trace(Trace(trace_id="root", mode="agent", task="mission", agent_role="planner"))
-    await trace_store.update_goal_tree("root", GoalTree(mission="mission"))
     runner = AgentRunner(trace_store=trace_store, llm_call=fake_llm)
     coordinator = wire_orchestration(
         runner,
@@ -483,35 +480,6 @@ class OneShotGetFailureArtifactStore(FileSystemArtifactStore):
         return await super().get(root_trace_id, snapshot_id)
 
 
-class OneShotGoalProjectionFailureStore(FileSystemTraceStore):
-    def __init__(self, base_path):
-        super().__init__(base_path)
-        self.fail_next_goal_update = False
-
-    async def update_goal(self, trace_id, goal_id, cascade_completion=True, **updates):
-        if self.fail_next_goal_update:
-            self.fail_next_goal_update = False
-            raise RuntimeError("injected goal projection failure")
-        return await super().update_goal(
-            trace_id,
-            goal_id,
-            cascade_completion=cascade_completion,
-            **updates,
-        )
-
-
-class OneShotGoalTreeProjectionFailureStore(FileSystemTraceStore):
-    def __init__(self, base_path):
-        super().__init__(base_path)
-        self.fail_next_goal_tree_update = False
-
-    async def update_goal_tree(self, trace_id, goal_tree):
-        if self.fail_next_goal_tree_update:
-            self.fail_next_goal_tree_update = False
-            raise RuntimeError("injected goal tree projection failure")
-        return await super().update_goal_tree(trace_id, goal_tree)
-
-
 class FailingDispatchCompletionTaskStore(FileSystemTaskStore):
     def __init__(self, base_path):
         super().__init__(base_path)
@@ -605,49 +573,28 @@ async def test_batch_reservation_conflict_does_not_strand_other_tasks(tmp_path):
 
 
 @pytest.mark.asyncio
-async def test_goal_projection_failure_does_not_abort_committed_attempt(tmp_path):
+async def test_explicit_lifecycle_never_creates_goal_tree(tmp_path):
     executor = FakeExecutor([ValidationVerdict.PASSED])
-    trace_store = OneShotGoalProjectionFailureStore(str(tmp_path))
-    coordinator, store, _ = await make_coordinator(
-        tmp_path,
-        executor,
-        trace_store=trace_store,
-    )
-    task_id = await create_task(coordinator, "reservation-projection-failure")
-    trace_store.fail_next_goal_update = True
+    coordinator, store, trace_store = await make_coordinator(tmp_path, executor)
+    task_id = await create_task(coordinator, "task-ledger-only")
 
     result = (await coordinator.dispatch_tasks("root", [task_id]))[0]
 
-    assert result.task_id == task_id
     assert result.task_status == TaskStatus.AWAITING_DECISION
-    assert result.attempt_id is not None
-    assert result.error is None
-    ledger = await store.load("root")
-    assert ledger.tasks[task_id].status == TaskStatus.AWAITING_DECISION
-    assert ledger.attempts[result.attempt_id].status.value == "submitted"
+    assert (await store.load("root")).tasks[task_id].status == TaskStatus.AWAITING_DECISION
+    assert await trace_store.get_goal_tree("root") is None
 
 
 @pytest.mark.asyncio
-async def test_task_creation_survives_goal_tree_projection_failure_and_reconciles(tmp_path):
-    executor = FakeExecutor([])
-    trace_store = OneShotGoalTreeProjectionFailureStore(str(tmp_path))
-    coordinator, store, _ = await make_coordinator(
-        tmp_path,
-        executor,
-        trace_store=trace_store,
-    )
-    trace_store.fail_next_goal_tree_update = True
-
-    task_id = await create_task(coordinator, "projection-outage")
+async def test_task_context_is_rendered_from_ledger(tmp_path):
+    coordinator, _, _ = await make_coordinator(tmp_path, FakeExecutor([]))
+    await create_task(coordinator, "draft the answer")
 
-    ledger = await store.load("root")
-    assert ledger.tasks[task_id].status == TaskStatus.PENDING
-    assert ledger.tasks[task_id].goal_id is None
+    context = await coordinator.task_context("root")
 
-    result = await coordinator.reconcile_goal_tree("root")
-    ledger = await store.load("root")
-    assert result["reconciled_tasks"] == 2
-    assert ledger.tasks[task_id].goal_id is not None
+    assert "## Current Task Plan" in context
+    assert "- 0 [waiting_children] mission" in context
+    assert "- 0.1 [pending] draft the answer" in context
 
 
 @pytest.mark.asyncio

agent/tests/test_goal_compatibility.py

@@ -5,7 +5,7 @@ from agent.trace.models import Trace
 from agent.trace.store import FileSystemTraceStore
 
 
-def test_goal_tree_legacy_cascade_and_explicit_no_cascade():
+def test_goal_tree_legacy_cascade_and_optional_no_cascade():
     legacy = GoalTree(mission="m")
     parent = legacy.add_goals(["parent"])[0]
     children = legacy.add_goals(["a", "b"], parent_id=parent.id)
@@ -13,12 +13,12 @@ def test_goal_tree_legacy_cascade_and_explicit_no_cascade():
     legacy.complete(children[1].id, "b")
     assert legacy.find(parent.id).status == "completed"
 
-    explicit = GoalTree(mission="m")
-    parent = explicit.add_goals(["parent"])[0]
-    children = explicit.add_goals(["a", "b"], parent_id=parent.id)
-    explicit.complete(children[0].id, "a", cascade_completion=False)
-    explicit.complete(children[1].id, "b", cascade_completion=False)
-    assert explicit.find(parent.id).status == "pending"
+    non_cascading = GoalTree(mission="m")
+    parent = non_cascading.add_goals(["parent"])[0]
+    children = non_cascading.add_goals(["a", "b"], parent_id=parent.id)
+    non_cascading.complete(children[0].id, "a", cascade_completion=False)
+    non_cascading.complete(children[1].id, "b", cascade_completion=False)
+    assert non_cascading.find(parent.id).status == "pending"
 
 
 @pytest.mark.asyncio

agent/tests/test_mission_root.py

@@ -42,8 +42,8 @@ async def test_new_ledger_has_one_stable_root_and_inspect_exposes_it(tmp_path):
 
 
 @pytest.mark.asyncio
-async def test_root_revise_drives_root_objective_mission_and_goal_rebuild(tmp_path):
-    coordinator, store, trace_store = await make_coordinator(
+async def test_root_revise_drives_root_objective_and_mission(tmp_path):
+    coordinator, store, _ = await make_coordinator(
         tmp_path, FakeExecutor([])
     )
     ledger = await store.load("root")
@@ -62,13 +62,6 @@ async def test_root_revise_drives_root_objective_mission_and_goal_rebuild(tmp_pa
     assert revised.root_objective == "revised mission"
     assert revised.mission == "revised mission"
 
-    goal_file = trace_store._get_goal_file("root")
-    goal_file.unlink()
-    await coordinator._ensure_goal_projection("root", root_id)
-    rebuilt = await trace_store.get_goal_tree("root")
-    assert rebuilt is not None
-    assert rebuilt.mission == "revised mission"
-
 
 @pytest.mark.asyncio
 async def test_rootless_legacy_ledger_falls_back_to_mission(tmp_path):

agent/tests/test_orchestration_edges.py

@@ -144,7 +144,7 @@ async def test_decision_and_revalidation_guards(tmp_path):
 
 def test_validation_report_requires_every_hard_criterion():
     task = TaskRecord(
-        task_id="t", goal_id=None, parent_task_id=None, display_path="1",
+        task_id="t", parent_task_id=None, display_path="1",
         specs=[TaskSpec(
             version=1, objective="o",
             acceptance_criteria=[AcceptanceCriterion("c1", "required", hard=True)],

agent/tests/test_orchestration_v2_api.py

@@ -111,7 +111,6 @@ class StubArtifactStore:
 def api_fixture():
     task = TaskRecord(
         task_id="task-1",
-        goal_id=None,
         parent_task_id=None,
         display_path="1",
         specs=[

agent/tests/test_orchestration_validation_policy.py

@@ -47,7 +47,6 @@ from test_coordinator_integration import (
 def make_context(root: str = "root") -> ValidationContext:
     task = TaskRecord(
         task_id="task-1",
-        goal_id=None,
         parent_task_id=None,
         display_path="1",
         specs=[