编排：隔离批量任务的预留与执行异常

dispatch_tasks 按输入索引收集每个 Task 的独立结果，TaskConflict、预留异常和运行期异常不再中断同批任务。Ledger 提交后的事件与 Goal 投影改为非破坏性处理，并修复预留失败误伤旧 running Attempt 的竞态。增加 sibling 隔离、顺序保持、投影失败和事件失败回归测试。

agent/agent/orchestration/coordinator.py

@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 import asyncio
+import logging
 from dataclasses import asdict
 from typing import Any, Callable, Dict, Iterable, List, Optional, Sequence, Tuple
 from uuid import uuid4
@@ -43,6 +44,9 @@ from .state_machine import transition
 from .store import TaskStoreNotFound
 
 
+logger = logging.getLogger(__name__)
+
+
 TERMINAL_TASK_STATUSES = {
     TaskStatus.COMPLETED,
     TaskStatus.CANCELLED,
@@ -132,7 +136,19 @@ class TaskCoordinator:
 
     async def _emit(self, root_trace_id: str, event_type: str, payload: Dict[str, Any]) -> None:
         if self.event_sink:
-            await self.event_sink.emit(root_trace_id, event_type, payload)
+            try:
+                await self.event_sink.emit(root_trace_id, event_type, payload)
+            except Exception as exc:
+                # Events are an observational projection.  Once the Ledger
+                # commit succeeds, an EventSink outage must not make callers
+                # believe the state mutation failed or retry it blindly.
+                logger.warning(
+                    "Orchestration event emission failed after Ledger commit "
+                    "(%s, %s): %s",
+                    root_trace_id,
+                    event_type,
+                    exc,
+                )
 
     async def create_tasks(
         self,
@@ -259,25 +275,146 @@ class TaskCoordinator:
 
         # Reserve every task before starting any worker so concurrent duplicate
         # dispatches fail deterministically.
-        reservations: List[Tuple[str, str]] = []
+        reservations: List[Tuple[int, str, str]] = []
+        early_results: Dict[int, TaskCycleResult] = {}
         for index, (task_id, preset) in enumerate(zip(task_ids, presets)):
             key = f"{idempotency_key}:{index}" if idempotency_key else None
-            attempt = await self._create_attempt(root_trace_id, task_id, preset, key)
-            reservations.append((task_id, attempt["attempt_id"]))
+            try:
+                attempt = await self._create_attempt(root_trace_id, task_id, preset, key)
+                reservations.append((index, task_id, attempt["attempt_id"]))
+            except TaskConflict as exc:
+                # A conflict belongs to this item only; previously reserved
+                # tasks must still run and later items must still be attempted.
+                early_results[index] = await self._cycle_error_result(
+                    root_trace_id, task_id, None, f"Dispatch conflict: {exc}"
+                )
+            except Exception as exc:
+                # _create_attempt contains all post-commit adapter failures.
+                # Therefore an exception here occurred before a new Attempt
+                # was safely returned.  Never guess by "latest Attempt": that
+                # could corrupt a pre-existing run owned by another dispatch.
+                attempt_id = None
+                early_results[index] = await self._cycle_error_result(
+                    root_trace_id, task_id, attempt_id,
+                    f"Attempt reservation failed: {type(exc).__name__}: {exc}",
+                )
 
         semaphore = asyncio.Semaphore(self.config.max_parallel_tasks)
 
-        async def run_one(task_id: str, attempt_id: str) -> TaskCycleResult:
+        async def run_one(index: int, task_id: str, attempt_id: str) -> Tuple[int, TaskCycleResult]:
             async with semaphore:
-                return await self._run_task_cycle(root_trace_id, task_id, attempt_id)
-
-        results = list(await asyncio.gather(*(run_one(t, a) for t, a in reservations)))
+                try:
+                    result = await self._run_task_cycle(root_trace_id, task_id, attempt_id)
+                except Exception as exc:
+                    error = f"Task cycle failed: {type(exc).__name__}: {exc}"
+                    await self._recover_cycle_failure(
+                        root_trace_id, task_id, attempt_id, error
+                    )
+                    result = await self._cycle_error_result(
+                        root_trace_id, task_id, attempt_id, error
+                    )
+                return index, result
+
+        completed = await asyncio.gather(
+            *(run_one(index, task_id, attempt_id) for index, task_id, attempt_id in reservations)
+        )
+        indexed_results = dict(early_results)
+        indexed_results.update(completed)
+        results = [indexed_results[index] for index in range(len(task_ids))]
         if idempotency_key:
             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
 
+    async def _recover_cycle_failure(
+        self,
+        root_trace_id: str,
+        task_id: str,
+        attempt_id: str,
+        error: str,
+    ) -> None:
+        """Best-effort normalization for an unexpected per-task failure."""
+
+        def mutate(ledger: TaskLedger) -> Dict[str, Any]:
+            task = _task(ledger, task_id)
+            attempt = ledger.attempts.get(attempt_id)
+            if attempt and attempt.status == AttemptStatus.RUNNING:
+                attempt.status = AttemptStatus.FAILED
+                attempt.error = error
+                attempt.updated_at = utc_now()
+
+            validation_id = None
+            for candidate_id in reversed(task.validation_ids):
+                candidate = ledger.validations[candidate_id]
+                if candidate.attempt_id == attempt_id:
+                    validation_id = candidate_id
+                    if candidate.status in {
+                        ValidationRunStatus.PENDING,
+                        ValidationRunStatus.RUNNING,
+                    }:
+                        candidate.status = ValidationRunStatus.ERROR
+                        candidate.error = error
+                        candidate.updated_at = utc_now()
+                    break
+
+            if task.status in {
+                TaskStatus.RUNNING,
+                TaskStatus.AWAITING_VALIDATION,
+                TaskStatus.VALIDATING,
+            }:
+                transition(task.status, TaskStatus.NEEDS_REPLAN)
+                task.status = TaskStatus.NEEDS_REPLAN
+                task.updated_at = utc_now()
+            return {
+                "task_id": task_id,
+                "attempt_id": attempt_id,
+                "validation_id": validation_id,
+                "error": error,
+            }
+
+        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
+
+    async def _cycle_error_result(
+        self,
+        root_trace_id: str,
+        task_id: str,
+        attempt_id: Optional[str],
+        error: str,
+    ) -> TaskCycleResult:
+        try:
+            ledger = await self.task_store.load(root_trace_id)
+            task = ledger.tasks.get(task_id)
+            status = task.status if task else TaskStatus.NEEDS_REPLAN
+            validation_id = None
+            validation = None
+            if task and attempt_id:
+                for candidate_id in reversed(task.validation_ids):
+                    candidate = ledger.validations[candidate_id]
+                    if candidate.attempt_id == attempt_id:
+                        validation_id = candidate_id
+                        validation = candidate
+                        break
+        except Exception:
+            status = TaskStatus.NEEDS_REPLAN
+            validation_id = None
+            validation = None
+        return TaskCycleResult(
+            task_id=task_id,
+            task_status=status,
+            attempt_id=attempt_id,
+            validation_id=validation_id,
+            validation=validation,
+            error=error,
+        )
+
     async def _create_attempt(
         self,
         root_trace_id: str,
@@ -328,7 +465,17 @@ class TaskCoordinator:
             }
 
         result = await self._mutate(root_trace_id, "attempt_created", mutate, idempotency_key)
-        await self.project_goal_state(root_trace_id, task_id)
+        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,
+            )
         return result
 
     async def _run_task_cycle(

agent/tests/test_coordinator_integration.py

@@ -81,14 +81,20 @@ class FakeExecutor:
         return ValidatorRunResult(context["validator_trace_id"], "completed")
 
 
-async def make_coordinator(tmp_path, executor):
-    trace_store = FileSystemTraceStore(str(tmp_path))
+async def make_coordinator(
+    tmp_path,
+    executor,
+    artifact_store=None,
+    trace_store=None,
+    task_store=None,
+):
+    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 = FileSystemTaskStore(str(tmp_path))
+    task_store = task_store or FileSystemTaskStore(str(tmp_path))
     coordinator = TaskCoordinator(
         task_store,
-        FileSystemArtifactStore(str(tmp_path)),
+        artifact_store or FileSystemArtifactStore(str(tmp_path)),
         trace_store,
         OrchestrationConfig(max_parallel_tasks=4),
         TraceEventSink(str(tmp_path)),
@@ -399,3 +405,165 @@ async def test_real_runner_local_executor_creates_independent_terminal_traces(tm
     assert worker.agent_role == "worker" and worker.result_summary
     assert validator.agent_role == "validator" and validator.result_summary
     assert worker.trace_id != validator.trace_id
+
+
+class OneShotGetFailureArtifactStore(FileSystemArtifactStore):
+    def __init__(self, base_path):
+        super().__init__(base_path)
+        self.get_calls = 0
+
+    def for_root(self, root_trace_id):
+        self.root_trace_id = root_trace_id
+        return self
+
+    async def get(self, snapshot_id):
+        self.get_calls += 1
+        if self.get_calls == 2:
+            raise RuntimeError("injected artifact read failure")
+        return await super().get(snapshot_id)
+
+
+class OneShotGoalProjectionFailureStore(FileSystemTraceStore):
+    def __init__(self, base_path):
+        super().__init__(base_path)
+        self.fail_next_goal_update = True
+
+    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 OneShotLoadFailureTaskStore(FileSystemTaskStore):
+    def __init__(self, base_path):
+        super().__init__(base_path)
+        self.fail_next_load = False
+
+    async def load(self, root_trace_id):
+        if self.fail_next_load:
+            self.fail_next_load = False
+            raise RuntimeError("injected reservation load failure")
+        return await super().load(root_trace_id)
+
+
+class FailingEventSink:
+    async def emit(self, root_trace_id, event_type, payload):
+        raise RuntimeError("injected event sink failure")
+
+
+@pytest.mark.asyncio
+async def test_parallel_unexpected_branch_error_does_not_cancel_siblings(tmp_path):
+    executor = FakeExecutor([ValidationVerdict.PASSED] * 4, delay=0.01)
+    artifact_store = OneShotGetFailureArtifactStore(str(tmp_path))
+    coordinator, store, _ = await make_coordinator(
+        tmp_path, executor, artifact_store=artifact_store
+    )
+    task_ids = [await create_task(coordinator, f"isolated-{index}") for index in range(4)]
+    results = await coordinator.dispatch_tasks("root", task_ids)
+
+    assert [result.task_id for result in results] == task_ids
+    failures = [result for result in results if result.error]
+    assert len(failures) == 1
+    assert "artifact read failure" in failures[0].error
+    assert failures[0].task_status == TaskStatus.NEEDS_REPLAN
+    successes = [result for result in results if not result.error]
+    assert len(successes) == 3
+    assert all(result.task_status == TaskStatus.AWAITING_DECISION for result in successes)
+    ledger = await store.load("root")
+    assert all(
+        ledger.tasks[task_id].status not in {TaskStatus.RUNNING, TaskStatus.VALIDATING}
+        for task_id in task_ids
+    )
+
+
+@pytest.mark.asyncio
+async def test_batch_reservation_conflict_does_not_strand_other_tasks(tmp_path):
+    executor = FakeExecutor([ValidationVerdict.PASSED, ValidationVerdict.PASSED])
+    coordinator, store, _ = await make_coordinator(tmp_path, executor)
+    task_ids = [await create_task(coordinator, f"reserve-{index}") for index in range(3)]
+    existing = await coordinator._create_attempt(
+        "root", task_ids[1], "worker", "existing-reservation"
+    )
+
+    results = await coordinator.dispatch_tasks("root", task_ids)
+    assert [result.task_id for result in results] == task_ids
+    assert results[0].task_status == TaskStatus.AWAITING_DECISION
+    assert results[1].attempt_id is None
+    assert "Dispatch conflict" in results[1].error
+    assert results[2].task_status == TaskStatus.AWAITING_DECISION
+
+    ledger = await store.load("root")
+    assert ledger.tasks[task_ids[0]].status == TaskStatus.AWAITING_DECISION
+    assert ledger.tasks[task_ids[2]].status == TaskStatus.AWAITING_DECISION
+    assert ledger.attempts[existing["attempt_id"]].status.value == "running"
+
+
+@pytest.mark.asyncio
+async def test_goal_projection_failure_does_not_abort_committed_attempt(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")
+
+    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"
+
+
+@pytest.mark.asyncio
+async def test_reservation_load_failure_does_not_corrupt_existing_attempt(tmp_path):
+    executor = FakeExecutor([ValidationVerdict.PASSED])
+    task_store = OneShotLoadFailureTaskStore(str(tmp_path))
+    coordinator, store, _ = await make_coordinator(
+        tmp_path,
+        executor,
+        task_store=task_store,
+    )
+    running_task = await create_task(coordinator, "already-running")
+    existing = await coordinator._create_attempt(
+        "root", running_task, "worker", "existing-running-attempt"
+    )
+    sibling_task = await create_task(coordinator, "unrelated-sibling")
+
+    task_store.fail_next_load = True
+    results = await coordinator.dispatch_tasks(
+        "root",
+        [running_task, sibling_task],
+    )
+
+    assert [result.task_id for result in results] == [running_task, sibling_task]
+    assert results[0].attempt_id is None
+    assert results[0].task_status == TaskStatus.RUNNING
+    assert "reservation load failure" in results[0].error
+    assert results[1].task_status == TaskStatus.AWAITING_DECISION
+    ledger = await store.load("root")
+    assert ledger.tasks[running_task].status == TaskStatus.RUNNING
+    assert ledger.attempts[existing["attempt_id"]].status.value == "running"
+
+
+@pytest.mark.asyncio
+async def test_event_sink_failure_does_not_rollback_committed_ledger(tmp_path):
+    executor = FakeExecutor([])
+    coordinator, store, _ = await make_coordinator(tmp_path, executor)
+    coordinator.event_sink = FailingEventSink()
+
+    task_id = await create_task(coordinator, "event-outage")
+
+    ledger = await store.load("root")
+    assert ledger.tasks[task_id].status == TaskStatus.PENDING