加固 Framework V1 编排闭环

Agent框架任务规划与独立验证技术方案.md

@@ -115,7 +115,7 @@ Host 传入的 context 会先合并，但 role、task_id、attempt_id、validati
 
 `agent/agent/orchestration/models.py` 已定义 Task、TaskSpec、Attempt、ArtifactSnapshot、ValidationReport 和 PlannerDecision。`TaskCoordinator` 是唯一 Ledger 写入入口。
 
-GoalTree 仍保留任务树展示和焦点能力，但 explicit 模式不向 Planner 暴露旧 `goal` 写入工具。Coordinator 先更新 Ledger，再投影 Goal 状态；Ledger 仍是唯一状态真相，后续可用 `reconcile_goal_tree()` 修复不一致。V1 已隔离 Attempt 创建后的投影异常，但其他投影点失败仍可能表现为“Ledger 已提交、当前工具调用报错”；统一 best-effort 隔离属于后续加固项。
+GoalTree 仍保留任务树展示和焦点能力，但 explicit 模式不向 Planner 暴露旧 `goal` 写入工具。Coordinator 先更新 Ledger，再以 best-effort 方式投影 Goal 状态；Ledger 仍是唯一状态真相。所有自动投影失败都只记录警告，不会把已提交的命令表现成失败；后续可用严格执行的 `reconcile_goal_tree()` 修复不一致。
 
 ### 3.4 Worker 和 Validator 通过 LocalAgentExecutor 运行
 
@@ -133,7 +133,7 @@ V1 explicit 只支持本地 Sub-Trace，`remote_*` Worker 或 Validator 会被
 
 `dispatch_tasks` 由 Coordinator 直接为每个 Task 保留独立 Attempt，再以 `asyncio.Semaphore` 限制并行度。每个分支都运行自己的 Worker→Validator 周期，按输入 task_ids 顺序返回 TaskCycleResult。
 
-TaskConflict、单个 Task 预留异常和单个执行周期内的 Artifact/框架异常会按 Task 归一化，不会中断同批 sibling。但所有分支返回后，批次级幂等结果写入失败仍会使整次工具调用报错，持续 Store 故障也不保证能返回逐 Task 结果。V1 不生成“整批成功”的聚合业务状态。
+TaskConflict、单个 Task 预留异常和单个执行周期内的 Artifact/框架异常会按 Task 归一化，不会中断同批 sibling。所有分支返回后，即使批次级幂等结果缓存写入失败，调用仍返回已持久化的逐 Task 结果；同一幂等请求重试时复用原 Attempt/Validation，不重复运行 Agent，并再次尝试保存整批结果。批量幂等键绑定原始 task_ids 和 worker presets，错误复用于另一批请求时会被拒绝。持续 Store 故障仍不保证恢复尚未提交的状态。V1 不生成“整批成功”的聚合业务状态。
 
 ### 3.6 端口、默认适配器和兼容边界已明确
 
@@ -183,11 +183,11 @@ V1 默认每个 TaskSpec 最多一次 repair 续用。“是否属于局部小
 
 ### 4.6 Ledger 真相与观测投影边界
 
-Ledger commit 成功后不会因观测失败而回滚，因此 Ledger 始终是真相。EventSink 异常和 Attempt 创建后的 Goal 投影异常已被捕获为警告；其他 Goal 投影调用点还没有统一隔离，可能让已成功的 Ledger commit 对外表现为当前工具调用失败。Attempt 异常处理使用明确 attempt_id，不会按“最新 Attempt”猜测并误伤其他正在运行的 Attempt。
+Ledger commit 成功后不会因观测失败而回滚，因此 Ledger 始终是真相。EventSink 异常以及 Task 创建、焦点、Attempt、Validation、Decision 和失败恢复后的自动 Goal 投影异常都被捕获为警告；只有显式 `reconcile_goal_tree()` 保留严格失败语义。Attempt 异常处理使用明确 attempt_id，不会按“最新 Attempt”猜测并误伤其他正在运行的 Attempt。
 
 ### 4.7 核心回归测试已建立
 
-当前 `agent/tests` 共有 45 个测试用例，覆盖状态机、Ledger/Snapshot、terminal tool、角色权限、Worker→Validator→Decision 闭环、repair/retry/revise/split/revalidate、并行异常隔离、Goal 投影、legacy 真实执行和 Host 解耦。
+当前 `agent/tests` 共有 49 个测试用例，覆盖状态机、Ledger/Snapshot、terminal tool、角色权限、Worker→Validator→Decision 闭环、repair/retry/revise/split/revalidate、并行异常隔离、Goal 投影、批次结果持久化故障、幂等请求绑定与并发重放、legacy 真实执行和 Host 解耦。
 
 ## 五、目标架构
 
@@ -896,7 +896,7 @@ Worker → Validator → Planner Decision
 
 V1 当前使用 `await` 模式跑完工具内的同步闭环，没有引入分布式队列。
 
-多个互不依赖的 Task 由 Coordinator 使用 Semaphore 并发调用 LocalAgentExecutor，不复用 legacy `agent()` 的多分支聚合语义。Task 预留和执行周期中的分支异常会逐 Task 隔离，其他 Task 继续执行，成功返回时顺序与输入一致。批次级 `dispatch_completed` 幂等写入不在分支 catch 内；该写入失败或持续 Store 故障仍可使整次工具调用报错。
+多个互不依赖的 Task 由 Coordinator 使用 Semaphore 并发调用 LocalAgentExecutor，不复用 legacy `agent()` 的多分支聚合语义。Task 预留和执行周期中的分支异常会逐 Task 隔离，其他 Task 继续执行，成功返回时顺序与输入一致。批次级 `dispatch_completed` 幂等缓存写入失败时返回已完成的逐 Task 结果；相同幂等请求重试会复用原 Attempt/Validation 并重新保存整批结果，不重复运行 Agent。
 
 ### 17.2 后续增加后台任务接口
 
@@ -988,8 +988,8 @@ agent/agent/orchestration/
 - 提供单进程幂等键，防止同一请求重复创建 Attempt、Validation 或 Decision；
 - orchestration 内核不再动态 import examples 或业务包；
 - `legacy_auto` 与 `explicit_validation` 双模式；
-- 建立 45 个当前回归用例，orchestration 覆盖率 90.07%；
-- 批量 Task 在预留和执行周期内按分支隔离异常，正常返回时顺序与输入一致；批次幂等写入失败的恢复仍属 V2；
+- 建立 49 个当前回归用例；
+- 批量 Task 在预留和执行周期内按分支隔离异常，正常返回时顺序与输入一致；批次幂等缓存写入失败时仍返回结果，重试不重复执行 Agent；
 - 明确 V1 限制：仅本地 Sub-Trace、同步工具调用、单进程文件存储、无新 Task HTTP API。
 
 ### 20.2 Framework V2：通用扩展接口与可靠恢复

agent/agent/core/runner.py

@@ -17,7 +17,6 @@ Agent Runner - Agent 执行引擎
 import asyncio
 import json
 import logging
-import os
 import uuid
 from dataclasses import dataclass, field
 from datetime import datetime
@@ -30,8 +29,6 @@ from agent.trace.compaction import (
     CompressionConfig,
     compress_completed_goals,
     estimate_tokens,
-    needs_level2_compression,
-    build_compression_prompt,
 )
 from agent.skill.models import Skill
 from agent.skill.skill_loader import load_skills_from_dir
@@ -50,13 +47,9 @@ from agent.orchestration.policy import (
 from agent.core.prompts import (
     DEFAULT_SYSTEM_PREFIX,
     TRUNCATION_HINT,
-    TOOL_INTERRUPTED_MESSAGE,
     AGENT_INTERRUPTED_SUMMARY,
-    AGENT_CONTINUE_HINT_TEMPLATE,
     TASK_NAME_GENERATION_SYSTEM_PROMPT,
     TASK_NAME_FALLBACK,
-    SUMMARY_HEADER_TEMPLATE,
-    build_summary_header,
     build_tool_interrupted_message,
     build_agent_continue_hint,
     ROLE_CONTRACTS,
@@ -1878,7 +1871,6 @@ class AgentRunner:
     
                         # 跟踪上传的知识（通过 upload_knowledge）
                         if tool_name == "upload_knowledge" and isinstance(tool_result, dict):
-                            metadata = tool_result.get("metadata", {})
                             # upload_knowledge 返回的是统计信息，不是单个 knowledge_id
                             # 这里只记录上传动作，不跟踪具体 ID
                             self.log.info(f"[Knowledge Tracking] 知识已上传到 Knowledge Manager")
@@ -2781,8 +2773,7 @@ class AgentRunner:
             if not base64_url.startswith("data:"):
                 return None
 
-            header, data = base64_url.split(",", 1)
-            media_type = header.split(";")[0].split(":")[1]  # image/png
+            _header, data = base64_url.split(",", 1)
 
             # 解码图片
             img_data = base64.b64decode(data)

agent/agent/docs/orchestration-v1.md

@@ -12,7 +12,7 @@ Planner -> Task -> Worker Attempt -> Artifact Snapshot
 - `legacy_auto`：默认值。保留原有 `agent`、`evaluate`、`goal` 和 Goal 父节点级联语义。
 - `explicit_validation`：启用 V1。只能配合带 `planner`、`worker`、`validator` role 的 preset 使用，且必须先装配 `TaskCoordinator`。
 
-显式模式的状态由 `TaskLedger` 决定。`GoalTree` 只用于兼容展示；二者不一致时可调用 `TaskCoordinator.reconcile_goal_tree()` 修复 Goal 投影。
+显式模式的状态由 `TaskLedger` 决定。`GoalTree` 只用于兼容展示；Ledger 提交后的自动 Goal 投影全部是 best-effort，投影失败只记录警告，不会把已经成功的命令表现成失败。二者不一致时可调用 `TaskCoordinator.reconcile_goal_tree()` 显式修复，修复本身失败时仍会向调用方报错。
 
 ## 最小装配
 
@@ -79,6 +79,7 @@ Worker 只能使用 `submit_attempt` 正式交付。Validator 只能使用 `subm
 - `failed`、`inconclusive` 没有 override 接受入口。
 - 同一 TaskSpec 最多 repair 一次；repair 产生新 Attempt 和 Snapshot，但续用原 Worker Trace。retry、revise 使用新 Worker Trace。
 - 子 Task 全部完成后，父 Task 进入 `needs_replan`，不会自动完成。
+- 带幂等键的批量 dispatch 在逐 Task 周期完成后，即使整批结果缓存写入失败，也会返回已持久化的逐 Task 结果；重试会复用原 Attempt/Validation，不会重复执行 Agent。幂等键绑定原始 Task 列表和 Worker preset，不能串用于另一批请求。
 
 ## 存储与扩展点
 

agent/agent/orchestration/coordinator.py

@@ -187,25 +187,13 @@ class TaskCoordinator:
             parent_path = ledger.tasks[effective_parent_id].display_path if effective_parent_id else ""
             created: List[str] = []
             for offset, draft in enumerate(drafts):
-                objective = str(draft.get("objective", "")).strip()
-                if not objective:
-                    raise ValueError("Task objective is required")
-                task_id = new_id()
-                criteria = [AcceptanceCriterion.from_dict(x) for x in draft.get("acceptance_criteria", [])]
-                spec = TaskSpec(
-                    version=1,
-                    objective=objective,
-                    acceptance_criteria=criteria,
-                    context_refs=list(draft.get("context_refs", [])),
-                )
                 index = start + offset
                 display_path = f"{parent_path}.{index}" if parent_path else str(index)
-                ledger.tasks[task_id] = TaskRecord(
-                    task_id=task_id,
-                    goal_id=None,
-                    parent_task_id=effective_parent_id,
-                    display_path=display_path,
-                    specs=[spec],
+                task_id = self._add_task_record(
+                    ledger,
+                    draft,
+                    effective_parent_id,
+                    display_path,
                 )
                 if effective_parent_id:
                     children = ledger.tasks[effective_parent_id].child_task_ids
@@ -226,7 +214,12 @@ class TaskCoordinator:
         result = await self._mutate(root_trace_id, "tasks_created", mutate, idempotency_key)
         after_task_id = result.get("after_task_id")
         for task_id in result["task_ids"]:
-            await self._ensure_goal_projection(root_trace_id, task_id, after_task_id=after_task_id)
+            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"])
 
@@ -237,13 +230,17 @@ class TaskCoordinator:
             return {"task_id": task_id, "status": task.status.value}
 
         result = await self._mutate(root_trace_id, "task_focused", mutate)
-        ledger = await self.task_store.load(root_trace_id)
-        task = ledger.tasks[task_id]
-        if self.trace_store and 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)
+        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
 
     async def dispatch_tasks(
@@ -271,16 +268,47 @@ class TaskCoordinator:
         if full_idem:
             existing = (await self.task_store.load(root_trace_id)).idempotency_results.get(full_idem)
             if existing and "results" in existing:
+                recorded_task_ids = existing.get("task_ids")
+                recorded_presets = existing.get("worker_presets")
+                if recorded_task_ids is not None and list(recorded_task_ids) != list(task_ids):
+                    raise TaskConflict("Idempotency key is bound to different task_ids")
+                if recorded_presets is not None and list(recorded_presets) != presets:
+                    raise TaskConflict("Idempotency key is bound to different worker_presets")
                 return [TaskCycleResult.from_dict(x) for x in existing["results"]]
 
         # Reserve every task before starting any worker so concurrent duplicate
         # dispatches fail deterministically.
+        dispatch_owner = str(uuid4())
         reservations: List[Tuple[int, str, str]] = []
         early_results: Dict[int, TaskCycleResult] = {}
+        replayed_indices = set()
         for index, (task_id, preset) in enumerate(zip(task_ids, presets)):
             key = f"{idempotency_key}:{index}" if idempotency_key else None
             try:
-                attempt = await self._create_attempt(root_trace_id, task_id, preset, key)
+                attempt = await self._create_attempt(
+                    root_trace_id,
+                    task_id,
+                    preset,
+                    key,
+                    reservation_owner=dispatch_owner,
+                )
+                if attempt.get("task_id") != task_id:
+                    raise TaskConflict(
+                        "Dispatch item idempotency key is bound to a different task"
+                    )
+                recorded_preset = attempt.get("worker_preset")
+                if recorded_preset is not None and recorded_preset != preset:
+                    raise TaskConflict(
+                        "Dispatch item idempotency key is bound to a different worker preset"
+                    )
+                if key and attempt.get("reservation_owner") != dispatch_owner:
+                    replayed_indices.add(index)
+                    early_results[index] = await self._cycle_result_from_ledger(
+                        root_trace_id,
+                        task_id,
+                        attempt.get("attempt_id"),
+                    )
+                    continue
                 reservations.append((index, task_id, attempt["attempt_id"]))
             except TaskConflict as exc:
                 # A conflict belongs to this item only; previously reserved
@@ -321,12 +349,79 @@ class TaskCoordinator:
         indexed_results = dict(early_results)
         indexed_results.update(completed)
         results = [indexed_results[index] for index in range(len(task_ids))]
-        if idempotency_key:
+        replay_in_progress = any(
+            early_results[index].task_status in {
+                TaskStatus.RUNNING,
+                TaskStatus.AWAITING_VALIDATION,
+                TaskStatus.VALIDATING,
+            }
+            for index in replayed_indices
+        )
+        if idempotency_key and not replay_in_progress:
             def remember(ledger: TaskLedger) -> Dict[str, Any]:
-                return {"results": [x.to_dict() for x in results]}
-            await self._mutate(root_trace_id, "dispatch_completed", remember, idempotency_key)
+                return {
+                    "results": [x.to_dict() for x in results],
+                    "task_ids": list(task_ids),
+                    "worker_presets": presets,
+                }
+            try:
+                await self._mutate(
+                    root_trace_id,
+                    "dispatch_completed",
+                    remember,
+                    idempotency_key,
+                )
+            except Exception as exc:
+                # Every Task cycle has already reached its own durable state.
+                # Returning those branch results is safer than making callers
+                # retry completed Workers because only the aggregate cache
+                # failed. A retry replays per-item reservations and can persist
+                # the aggregate result without executing the Agents again.
+                logger.warning(
+                    "Dispatch result persistence failed after Task cycles "
+                    "completed (%s, %s): %s",
+                    root_trace_id,
+                    idempotency_key,
+                    exc,
+                )
         return results
 
+    async def _cycle_result_from_ledger(
+        self,
+        root_trace_id: str,
+        task_id: str,
+        attempt_id: Optional[str],
+    ) -> TaskCycleResult:
+        """Rebuild an idempotent dispatch result without rerunning Agents."""
+
+        ledger = await self.task_store.load(root_trace_id)
+        task = _task(ledger, task_id)
+        attempt = ledger.attempts.get(attempt_id) if attempt_id else None
+        validation = None
+        validation_id = None
+        if attempt_id:
+            for candidate_id in reversed(task.validation_ids):
+                candidate = ledger.validations[candidate_id]
+                if candidate.attempt_id == attempt_id:
+                    validation = candidate
+                    validation_id = candidate_id
+                    break
+        error = validation.error if validation else (attempt.error if attempt else None)
+        if task.status in {
+            TaskStatus.RUNNING,
+            TaskStatus.AWAITING_VALIDATION,
+            TaskStatus.VALIDATING,
+        }:
+            error = error or "Dispatch with this idempotency key is still in progress"
+        return TaskCycleResult(
+            task_id=task_id,
+            task_status=task.status,
+            attempt_id=attempt_id,
+            validation_id=validation_id,
+            validation=validation,
+            error=error,
+        )
+
     async def _recover_cycle_failure(
         self,
         root_trace_id: str,
@@ -375,12 +470,12 @@ class TaskCoordinator:
 
         try:
             await self._mutate(root_trace_id, "task_cycle_failed", mutate)
-            await self.project_goal_state(root_trace_id, task_id)
         except Exception:
             # The original adapter failure is the useful branch result.  A
             # 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,
@@ -421,6 +516,7 @@ class TaskCoordinator:
         task_id: str,
         worker_preset: str,
         idempotency_key: Optional[str],
+        reservation_owner: Optional[str] = None,
     ) -> Dict[str, Any]:
         def mutate(ledger: TaskLedger) -> Dict[str, Any]:
             task = _task(ledger, task_id)
@@ -461,21 +557,13 @@ class TaskCoordinator:
                 "task_id": task_id,
                 "attempt_id": attempt.attempt_id,
                 "worker_trace_id": worker_trace_id,
+                "worker_preset": worker_preset,
                 "execution_mode": execution_mode,
+                "reservation_owner": reservation_owner,
             }
 
         result = await self._mutate(root_trace_id, "attempt_created", mutate, idempotency_key)
-        try:
-            await self.project_goal_state(root_trace_id, task_id)
-        except Exception as exc:
-            # GoalTree is a compatibility projection.  The committed Attempt
-            # remains authoritative and can execute; reconcile_goal_tree can
-            # repair the projection later.
-            logger.warning(
-                "Goal projection failed after Attempt %s was committed: %s",
-                result["attempt_id"],
-                exc,
-            )
+        await self._project_goal_compatibility(root_trace_id, task_id)
         return result
 
     async def _run_task_cycle(
@@ -624,7 +712,7 @@ class TaskCoordinator:
             }
 
         result = await self._mutate(root_trace_id, "attempt_submitted", mutate, tool_call_id)
-        await self.project_goal_state(root_trace_id, task_id)
+        await self._project_goal_compatibility(root_trace_id, task_id)
         return result
 
     async def _start_validation(self, root_trace_id: str, task_id: str, attempt_id: str) -> str:
@@ -651,7 +739,7 @@ class TaskCoordinator:
             return {"validation_id": validation.validation_id, "task_id": task_id}
 
         result = await self._mutate(root_trace_id, "validation_started", mutate)
-        await self.project_goal_state(root_trace_id, task_id)
+        await self._project_goal_compatibility(root_trace_id, task_id)
         return result["validation_id"]
 
     async def submit_validation(
@@ -714,7 +802,7 @@ class TaskCoordinator:
             }
 
         result = await self._mutate(root_trace_id, "validation_submitted", mutate, tool_call_id)
-        await self.project_goal_state(root_trace_id, task_id)
+        await self._project_goal_compatibility(root_trace_id, task_id)
         return result
 
     @staticmethod
@@ -851,18 +939,24 @@ class TaskCoordinator:
             }
 
         result = await self._mutate(root_trace_id, "planner_decision", mutate, idempotency_key)
-        ledger = await self.task_store.load(root_trace_id)
         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)
-        parent_id = ledger.tasks[task_id].parent_task_id
-        if parent_id:
-            affected.append(parent_id)
+        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._ensure_goal_projection(root_trace_id, affected_task)
-            await self.project_goal_state(root_trace_id, affected_task)
+            await self._project_goal_compatibility(
+                root_trace_id,
+                affected_task,
+                ensure=True,
+            )
         return result
 
     @staticmethod
@@ -940,6 +1034,7 @@ class TaskCoordinator:
 
         result = await self._mutate(root_trace_id, "validation_restarted", mutate, idempotency_key)
         validation_id = result["validation_id"]
+        await self._project_goal_compatibility(root_trace_id, task_id)
         ledger = await self.task_store.load(root_trace_id)
         task = ledger.tasks[task_id]
         attempt = ledger.attempts[attempt_id]
@@ -1039,7 +1134,7 @@ 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_state(root_trace_id, task_id)
+        await self._project_goal_compatibility(root_trace_id, task_id)
 
     async def _mark_validation_error(
         self,
@@ -1065,7 +1160,7 @@ 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_state(root_trace_id, task_id)
+        await self._project_goal_compatibility(root_trace_id, task_id)
 
     @staticmethod
     def _repair_feedback(ledger: TaskLedger, task: TaskRecord) -> Optional[Dict[str, Any]]:
@@ -1098,7 +1193,36 @@ class TaskCoordinator:
         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:
+        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", [])),
+        )
+        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
+
     def _create_records(
+        self,
         ledger: TaskLedger,
         drafts: Sequence[Dict[str, Any]],
         parent_task_id: Optional[str],
@@ -1110,21 +1234,13 @@ class TaskCoordinator:
         )
         created: List[str] = []
         for offset, draft in enumerate(drafts, start=1):
-            objective = str(draft.get("objective", "")).strip()
-            if not objective:
-                raise ValueError("New task objective is required")
-            task_id = new_id()
             index = len(siblings) + offset
             display_path = f"{parent_path}.{index}" if parent_path else str(index)
-            spec = TaskSpec(
-                version=1,
-                objective=objective,
-                acceptance_criteria=[AcceptanceCriterion.from_dict(x) for x in draft.get("acceptance_criteria", [])],
-                context_refs=list(draft.get("context_refs", [])),
-            )
-            ledger.tasks[task_id] = TaskRecord(
-                task_id=task_id, goal_id=None, parent_task_id=parent_task_id,
-                display_path=display_path, specs=[spec],
+            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)
@@ -1180,6 +1296,48 @@ class TaskCoordinator:
             summary=summary,
         )
 
+    async def _project_goal_compatibility(
+        self,
+        root_trace_id: str,
+        task_id: str,
+        *,
+        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)
+
+    @staticmethod
+    def _log_projection_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_goal_projection(
         self,
         root_trace_id: str,

agent/agent/trace/store.py

@@ -492,8 +492,6 @@ class FileSystemTraceStore:
 
         return affected
 
-        return affected
-
     async def get_message(self, message_id: str) -> Optional[Message]:
         """获取 Message（扫描所有 trace）"""
         for trace_dir in self.base_path.iterdir():

agent/tests/test_coordinator_integration.py

@@ -32,8 +32,11 @@ class FakeExecutor:
         self.coordinator = None
         self.active = 0
         self.max_active = 0
+        self.worker_calls = 0
+        self.validator_calls = 0
 
     async def run_worker(self, context):
+        self.worker_calls += 1
         self.active += 1
         self.max_active = max(self.max_active, self.active)
         if self.delay:
@@ -57,6 +60,7 @@ class FakeExecutor:
             self.active -= 1
 
     async def run_validator(self, context):
+        self.validator_calls += 1
         verdict = self.verdicts.popleft() if self.verdicts else ValidationVerdict.PASSED
         if self.submit_validator:
             criteria = [
@@ -269,13 +273,26 @@ async def test_block_unblock_cancel_and_supersede(tmp_path):
     old_id = await create_task(coordinator, "old")
     result = await coordinator.decide_task(
         "root", old_id, None, DecisionAction.SUPERSEDE,
-        {"reason": "replace spec", "replacement": {"objective": "replacement"}},
+        {
+            "reason": "replace spec",
+            "replacement": {
+                "objective": "replacement",
+                "acceptance_criteria": [
+                    {"criterion_id": "replacement-c1", "description": "replacement passes"}
+                ],
+                "context_refs": ["replacement-context"],
+            },
+        },
         "supersede-1",
     )
     replacement_id = result["payload"]["replacement_task_id"]
     ledger = await store.load("root")
     assert ledger.tasks[old_id].status == TaskStatus.SUPERSEDED
     assert ledger.tasks[replacement_id].status == TaskStatus.PENDING
+    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"]
 
 
 @pytest.mark.asyncio
@@ -286,8 +303,8 @@ 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"}],
-        placement={"after_task_id": first},
+        [{"objective": "second", "context_refs": ["second-context"]}],
+        placement={"after_task_id": first, "focus": True},
         idempotency_key="insert",
     )
     second = inserted["tasks"][0]["task_id"]
@@ -295,11 +312,13 @@ async def test_insert_after_keeps_stable_ids_and_updates_display_order(tmp_path)
     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.focused_task_id == second
     assert len({first, second, third}) == 3
     repeated = await coordinator.create_tasks(
         "root",
-        [{"objective": "second"}],
-        placement={"after_task_id": first},
+        [{"objective": "second", "context_refs": ["second-context"]}],
+        placement={"after_task_id": first, "focus": True},
         idempotency_key="insert",
     )
     assert repeated["tasks"][0]["task_id"] == second
@@ -316,13 +335,32 @@ async def test_split_children_do_not_auto_complete_parent(tmp_path):
         {
             "reason": "split work",
             "tasks": [
-                {"objective": "child one", "acceptance_criteria": [{"criterion_id": "c1", "description": "pass"}]},
-                {"objective": "child two", "acceptance_criteria": [{"criterion_id": "c1", "description": "pass"}]},
+                {
+                    "objective": "child one",
+                    "acceptance_criteria": [{"criterion_id": "c1", "description": "pass"}],
+                    "context_refs": ["child-one-context"],
+                },
+                {
+                    "objective": "child two",
+                    "acceptance_criteria": [{"criterion_id": "c2", "description": "pass"}],
+                    "context_refs": ["child-two-context"],
+                },
             ],
         },
         "split-1",
     )
     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].current_spec.context_refs for x in child_ids] == [
+        ["child-one-context"],
+        ["child-two-context"],
+    ]
+    assert [
+        ledger.tasks[x].current_spec.acceptance_criteria[0].criterion_id
+        for x in child_ids
+    ] == ["c1", "c2"]
     cycles = await coordinator.dispatch_tasks("root", child_ids)
     for child_id, cycle in zip(child_ids, cycles):
         await coordinator.decide_task(
@@ -426,7 +464,7 @@ class OneShotGetFailureArtifactStore(FileSystemArtifactStore):
 class OneShotGoalProjectionFailureStore(FileSystemTraceStore):
     def __init__(self, base_path):
         super().__init__(base_path)
-        self.fail_next_goal_update = True
+        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:
@@ -440,6 +478,37 @@ class OneShotGoalProjectionFailureStore(FileSystemTraceStore):
         )
 
 
+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)
+        self.fail_next_dispatch_completion = True
+
+    async def commit(self, ledger, expected_revision, idempotency_key=None):
+        if (
+            self.fail_next_dispatch_completion
+            and idempotency_key == "root:batch"
+        ):
+            self.fail_next_dispatch_completion = False
+            raise RuntimeError("injected dispatch completion persistence failure")
+        return await super().commit(
+            ledger,
+            expected_revision,
+            idempotency_key=idempotency_key,
+        )
+
+
 class OneShotLoadFailureTaskStore(FileSystemTaskStore):
     def __init__(self, base_path):
         super().__init__(base_path)
@@ -514,6 +583,7 @@ async def test_goal_projection_failure_does_not_abort_committed_attempt(tmp_path
         trace_store=trace_store,
     )
     task_id = await create_task(coordinator, "reservation-projection-failure")
+    trace_store.fail_next_goal_update = True
 
     result = (await coordinator.dispatch_tasks("root", [task_id]))[0]
 
@@ -526,6 +596,136 @@ async def test_goal_projection_failure_does_not_abort_committed_attempt(tmp_path
     assert ledger.attempts[result.attempt_id].status.value == "submitted"
 
 
+@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")
+
+    ledger = await store.load("root")
+    assert ledger.tasks[task_id].status == TaskStatus.PENDING
+    assert ledger.tasks[task_id].goal_id is None
+
+    result = await coordinator.reconcile_goal_tree("root")
+    ledger = await store.load("root")
+    assert result["reconciled_tasks"] == 1
+    assert ledger.tasks[task_id].goal_id is not None
+
+
+@pytest.mark.asyncio
+async def test_dispatch_returns_results_when_batch_result_persistence_fails(tmp_path):
+    executor = FakeExecutor([ValidationVerdict.PASSED, ValidationVerdict.PASSED])
+    task_store = FailingDispatchCompletionTaskStore(str(tmp_path))
+    coordinator, store, _ = await make_coordinator(
+        tmp_path,
+        executor,
+        task_store=task_store,
+    )
+    task_ids = [await create_task(coordinator, f"batch-{index}") for index in range(2)]
+
+    first = await coordinator.dispatch_tasks(
+        "root",
+        task_ids,
+        idempotency_key="batch",
+    )
+
+    assert [result.task_status for result in first] == [
+        TaskStatus.AWAITING_DECISION,
+        TaskStatus.AWAITING_DECISION,
+    ]
+    assert executor.worker_calls == 2
+    assert executor.validator_calls == 2
+
+    repeated = await coordinator.dispatch_tasks(
+        "root",
+        task_ids,
+        idempotency_key="batch",
+    )
+
+    assert [result.attempt_id for result in repeated] == [
+        result.attempt_id for result in first
+    ]
+    assert executor.worker_calls == 2
+    assert executor.validator_calls == 2
+    ledger = await store.load("root")
+    assert all(len(ledger.tasks[task_id].attempt_ids) == 1 for task_id in task_ids)
+    assert all(len(ledger.tasks[task_id].validation_ids) == 1 for task_id in task_ids)
+
+
+@pytest.mark.asyncio
+async def test_dispatch_idempotency_key_is_bound_to_tasks_and_presets(tmp_path):
+    executor = FakeExecutor([ValidationVerdict.PASSED])
+    coordinator, _, _ = await make_coordinator(tmp_path, executor)
+    first_task = await create_task(coordinator, "first-binding")
+    second_task = await create_task(coordinator, "second-binding")
+
+    await coordinator.dispatch_tasks(
+        "root",
+        [first_task],
+        worker_presets=["worker"],
+        idempotency_key="bound-batch",
+    )
+
+    with pytest.raises(TaskConflict, match="different task_ids"):
+        await coordinator.dispatch_tasks(
+            "root",
+            [second_task],
+            worker_presets=["worker"],
+            idempotency_key="bound-batch",
+        )
+    with pytest.raises(TaskConflict, match="different worker_presets"):
+        await coordinator.dispatch_tasks(
+            "root",
+            [first_task],
+            worker_presets=["alternate-worker"],
+            idempotency_key="bound-batch",
+        )
+
+    assert executor.worker_calls == 1
+    assert executor.validator_calls == 1
+
+
+@pytest.mark.asyncio
+async def test_concurrent_dispatch_replay_does_not_duplicate_agent_execution(tmp_path):
+    executor = FakeExecutor([ValidationVerdict.PASSED], delay=0.02)
+    coordinator, store, _ = await make_coordinator(tmp_path, executor)
+    task_id = await create_task(coordinator, "concurrent-idempotency")
+
+    first, replay = await asyncio.gather(
+        coordinator.dispatch_tasks("root", [task_id], idempotency_key="concurrent"),
+        coordinator.dispatch_tasks("root", [task_id], idempotency_key="concurrent"),
+    )
+
+    assert executor.worker_calls == 1
+    assert executor.validator_calls == 1
+    assert any(
+        result[0].task_status == TaskStatus.AWAITING_DECISION
+        for result in (first, replay)
+    )
+    assert any(
+        result[0].error
+        and "still in progress" in result[0].error
+        for result in (first, replay)
+    )
+
+    final = await coordinator.dispatch_tasks(
+        "root",
+        [task_id],
+        idempotency_key="concurrent",
+    )
+    ledger = await store.load("root")
+    assert final[0].task_status == TaskStatus.AWAITING_DECISION
+    assert len(ledger.tasks[task_id].attempt_ids) == 1
+    assert len(ledger.tasks[task_id].validation_ids) == 1
+
+
 @pytest.mark.asyncio
 async def test_reservation_load_failure_does_not_corrupt_existing_attempt(tmp_path):
     executor = FakeExecutor([ValidationVerdict.PASSED])