feat(events): 增加规范运行事件对账与幂等投影

在根 Trace 现有 events.jsonl 中追加版本化 run_event，覆盖 TaskProgress、候选版本与生命周期、验证完成、父级审核和回溯。所有事件均在权威状态落盘后写入，以稳定 event_key 去重；日志失败不回滚 Agent 协议状态，恢复时可从 TaskProtocol 与 candidate ledger 补发缺失事件。

新增按 event_id cursor 严格顺序推进的 RunEventPump。Projector 必须在业务事务中同时完成幂等写入和 cursor 推进；N 失败时不投影 N+1，未知 schema 与普通 Trace 事件不破坏顺序，应用恢复和 Run 完成时都会尝试继续投影。

新增事件 ID 缺口、投影器写业务数据后失败、重复重放、未知 schema 与 append_event 故障恢复测试。

cyber_agent/application/__init__.py

@@ -36,6 +36,8 @@ _LAZY_EXPORTS = {
     "QualityCheckProvider": ("cyber_agent.application.quality", "QualityCheckProvider"),
     "ValidationSubject": ("cyber_agent.application.quality", "ValidationSubject"),
     "TaskContextProvider": ("cyber_agent.application.ports", "TaskContextProvider"),
+    "RunEvent": ("cyber_agent.application.events", "RunEvent"),
+    "RunEventProjector": ("cyber_agent.application.ports", "RunEventProjector"),
 }
 
 
@@ -75,6 +77,8 @@ __all__ = [
     "QualityCheckProvider",
     "RoleRunLimits",
     "RunLimits",
+    "RunEvent",
+    "RunEventProjector",
     "SkillSpec",
     "TaskContextProvider",
     "ToolSet",

cyber_agent/application/candidate_service.py

@@ -42,11 +42,13 @@ class CandidateService:
         application_binding: Any,
         repository: CandidateRepository,
         artifact_resolver: ArtifactResolver,
+        event_service: Any = None,
     ) -> None:
         self.store = store
         self.binding = application_binding
         self.repository = repository
         self.artifact_resolver = artifact_resolver
+        self.event_service = event_service
 
     @classmethod
     def _lock_for(cls, root_trace_id: str) -> asyncio.Lock:
@@ -195,6 +197,7 @@ class CandidateService:
                 root_trace_id,
                 completed.model_dump(mode="json"),
             )
+            await self._emit_candidate_registered(candidate_ref)
             return candidate_ref
 
     async def validate_report_refs(
@@ -298,6 +301,24 @@ class CandidateService:
                 root.trace_id,
                 updated.model_dump(mode="json"),
             )
+            if self.event_service is not None:
+                await self.event_service.emit_after_commit(
+                    source_trace_id=record.candidate_ref.owner_trace_id,
+                    event_type="validation.completed",
+                    event_key=(
+                        "validation.completed:candidate:"
+                        f"{record.candidate_ref.candidate_id}:"
+                        f"{record.candidate_ref.revision}:{record.plan_hash}"
+                    ),
+                    effective_at_sequence=record.validated_at_sequence,
+                    payload={
+                        "subject_type": "candidate",
+                        "candidate_ref": record.candidate_ref.model_dump(
+                            mode="json"
+                        ),
+                        "validation_result": record.validation_result,
+                    },
+                )
 
     async def apply_review_actions(
         self,
@@ -465,6 +486,7 @@ class CandidateService:
                 root_trace_id,
                 ledger.model_dump(mode="json"),
             )
+            await self._emit_lifecycle(record, root_trace_id)
             return ledger, record
 
         pending = CandidateLifecycleRecord(
@@ -483,6 +505,7 @@ class CandidateService:
                 root_trace_id,
                 ledger.model_dump(mode="json"),
             )
+            await self._emit_lifecycle(pending, root_trace_id)
         try:
             receipt = CandidateAdoptionReceipt.model_validate(
                 await self.repository.commit_adoption(CandidateAdoptionRequest(
@@ -511,6 +534,7 @@ class CandidateService:
                 root_trace_id,
                 ledger.model_dump(mode="json"),
             )
+            await self._emit_lifecycle(failed, root_trace_id)
             raise
         adopted = CandidateLifecycleRecord(
             operation_id=operation_id,
@@ -528,8 +552,64 @@ class CandidateService:
             root_trace_id,
             ledger.model_dump(mode="json"),
         )
+        await self._emit_lifecycle(adopted, root_trace_id)
         return ledger, adopted
 
+    async def _emit_candidate_registered(self, candidate: CandidateRef) -> None:
+        if self.event_service is None:
+            return
+        await self.event_service.emit_after_commit(
+            source_trace_id=candidate.owner_trace_id,
+            event_type="candidate.version_registered",
+            event_key=(
+                f"candidate.version_registered:{candidate.candidate_id}:"
+                f"{candidate.revision}"
+            ),
+            effective_at_sequence=candidate.created_at_sequence,
+            payload={"candidate_ref": candidate.model_dump(mode="json")},
+        )
+        await self.event_service.emit_after_commit(
+            source_trace_id=candidate.owner_trace_id,
+            event_type="candidate.lifecycle_changed",
+            event_key=(
+                f"candidate.lifecycle_changed:candidate:"
+                f"{candidate.owner_trace_id}:{candidate.created_at_sequence}:proposed"
+            ),
+            effective_at_sequence=candidate.created_at_sequence,
+            payload={"lifecycle": CandidateLifecycleRecord(
+                operation_id=(
+                    f"candidate:{candidate.owner_trace_id}:"
+                    f"{candidate.created_at_sequence}"
+                ),
+                candidate=CandidatePointer(
+                    candidate_id=candidate.candidate_id,
+                    revision=candidate.revision,
+                ),
+                state="proposed",
+                effective_at_sequence=candidate.created_at_sequence,
+                source_trace_id=candidate.owner_trace_id,
+            ).model_dump(mode="json")},
+        )
+
+    async def _emit_lifecycle(
+        self,
+        lifecycle: CandidateLifecycleRecord,
+        root_trace_id: str,
+    ) -> None:
+        del root_trace_id
+        if self.event_service is None or lifecycle.source_trace_id is None:
+            return
+        await self.event_service.emit_after_commit(
+            source_trace_id=lifecycle.source_trace_id,
+            event_type="candidate.lifecycle_changed",
+            event_key=(
+                f"candidate.lifecycle_changed:{lifecycle.operation_id}:"
+                f"{lifecycle.state}"
+            ),
+            effective_at_sequence=lifecycle.effective_at_sequence,
+            payload={"lifecycle": lifecycle.model_dump(mode="json")},
+        )
+
     @staticmethod
     def _finish_review_operation(
         ledger: CandidateLedger,
@@ -678,6 +758,7 @@ class CandidateService:
             ),
             state="proposed",
             effective_at_sequence=candidate.created_at_sequence,
+            source_trace_id=candidate.owner_trace_id,
         ))
         return CandidateLedger(
             candidates=(*ledger.candidates, candidate),

cyber_agent/application/events.py

@@ -0,0 +1,306 @@
+"""Canonical root event journal and ordered, failure-isolated projection pump."""
+
+from __future__ import annotations
+
+import asyncio
+import logging
+from typing import Any
+from weakref import WeakValueDictionary
+
+from pydantic import Field
+
+from cyber_agent.application.candidate import CandidateLedger
+from cyber_agent.application.models import ApplicationModel, ApplicationRef
+from cyber_agent.application.ports import RunEventProjector
+from cyber_agent.core.task_protocol import ensure_task_protocol
+
+
+logger = logging.getLogger(__name__)
+_ROOT_LOCKS: "WeakValueDictionary[tuple[int, str], asyncio.Lock]" = (
+    WeakValueDictionary()
+)
+
+
+class RunEvent(ApplicationModel):
+    event_id: int = Field(gt=0)
+    schema_version: int = Field(ge=1)
+    event_type: str = Field(min_length=1, max_length=200)
+    event_key: str = Field(min_length=1, max_length=500)
+    application_ref: ApplicationRef
+    root_trace_id: str = Field(min_length=1)
+    source_trace_id: str = Field(min_length=1)
+    effective_at_sequence: int = Field(ge=0)
+    payload: dict[str, Any]
+
+
+class RunEventPump:
+    """Project canonical events in journal order without touching Agent state."""
+
+    def __init__(
+        self,
+        *,
+        store: Any,
+        application_ref: ApplicationRef,
+        projector: RunEventProjector | None,
+    ) -> None:
+        self.store = store
+        self.application_ref = application_ref
+        self.projector = projector
+
+    async def pump(self, root_trace_id: str) -> int:
+        if self.projector is None:
+            return 0
+        cursor = await self.projector.load_cursor(
+            self.application_ref,
+            root_trace_id,
+        )
+        projected = 0
+        for raw in await self.store.get_events(root_trace_id, cursor):
+            if raw.get("event") != "run_event":
+                continue
+            event = RunEvent.model_validate({
+                key: raw[key]
+                for key in RunEvent.model_fields
+                if key in raw
+            })
+            if (
+                event.application_ref != self.application_ref
+                or event.root_trace_id != root_trace_id
+            ):
+                raise ValueError("run_event application/root identity mismatch")
+            await self.projector.project(event, cursor)
+            next_cursor = await self.projector.load_cursor(
+                self.application_ref,
+                root_trace_id,
+            )
+            if next_cursor != event.event_id:
+                raise ValueError(
+                    "RunEventProjector did not atomically advance its cursor"
+                )
+            cursor = next_cursor
+            projected += 1
+        return projected
+
+
+class RunEventService:
+    """Append idempotent canonical events after authoritative state commits."""
+
+    def __init__(
+        self,
+        *,
+        store: Any,
+        application_ref: ApplicationRef,
+        projector: RunEventProjector | None = None,
+    ) -> None:
+        self.store = store
+        self.application_ref = application_ref
+        self.pump = RunEventPump(
+            store=store,
+            application_ref=application_ref,
+            projector=projector,
+        )
+
+    @staticmethod
+    def _lock(root_trace_id: str) -> asyncio.Lock:
+        key = (id(asyncio.get_running_loop()), root_trace_id)
+        lock = _ROOT_LOCKS.get(key)
+        if lock is None:
+            lock = asyncio.Lock()
+            _ROOT_LOCKS[key] = lock
+        return lock
+
+    async def emit(
+        self,
+        *,
+        source_trace_id: str,
+        event_type: str,
+        event_key: str,
+        effective_at_sequence: int,
+        payload: dict[str, Any],
+        project: bool = True,
+    ) -> int:
+        source = await self.store.get_trace(source_trace_id)
+        if source is None:
+            raise ValueError(f"Trace not found: {source_trace_id}")
+        root_trace_id = source.context.get("root_trace_id") or source.trace_id
+        if source.context.get("application_ref") != self.application_ref.model_dump(
+            mode="json"
+        ):
+            raise ValueError("run_event source application binding mismatch")
+        async with self._lock(root_trace_id):
+            existing = next(
+                (
+                    item for item in await self.store.get_events(root_trace_id)
+                    if item.get("event") == "run_event"
+                    and item.get("event_key") == event_key
+                ),
+                None,
+            )
+            if existing is not None:
+                event_id = int(existing["event_id"])
+            else:
+                event_id = await self.store.append_event(
+                    root_trace_id,
+                    "run_event",
+                    {
+                        "schema_version": 1,
+                        "event_type": event_type,
+                        "event_key": event_key,
+                        "application_ref": self.application_ref.model_dump(
+                            mode="json"
+                        ),
+                        "root_trace_id": root_trace_id,
+                        "source_trace_id": source_trace_id,
+                        "effective_at_sequence": effective_at_sequence,
+                        "payload": payload,
+                    },
+                )
+        if project:
+            await self.try_pump(root_trace_id)
+        return event_id
+
+    async def emit_after_commit(self, **kwargs: Any) -> None:
+        """Never roll back protocol state because journaling/projection failed."""
+        try:
+            await self.emit(**kwargs)
+        except Exception:
+            logger.exception("Failed to emit/project canonical run event")
+
+    async def try_pump(self, root_trace_id: str) -> None:
+        try:
+            await self.pump.pump(root_trace_id)
+        except Exception:
+            logger.exception(
+                "RunEventProjector stopped at its current cursor for root %s",
+                root_trace_id,
+            )
+
+    async def reconcile_root(self, root_trace_id: str) -> int:
+        """Rebuild missing canonical events from current authoritative records."""
+        traces = [
+            item for item in await self.store.list_traces(limit=10_000)
+            if (item.context.get("root_trace_id") or item.trace_id) == root_trace_id
+            and item.context.get("application_ref")
+            == self.application_ref.model_dump(mode="json")
+        ]
+        emitted = 0
+        for trace in traces:
+            state = ensure_task_protocol(trace.context)
+            for revision in state.get("task_progress_revisions", []):
+                emitted += bool(await self.emit(
+                    source_trace_id=trace.trace_id,
+                    event_type="task.progress.revised",
+                    event_key=(
+                        f"task.progress.revised:{trace.trace_id}:"
+                        f"{revision['revision']}"
+                    ),
+                    effective_at_sequence=revision["effective_at_sequence"],
+                    payload={"revision": revision},
+                    project=False,
+                ))
+            cache = state.get("task_report_validation")
+            if isinstance(cache, dict) and cache.get("aggregate_result"):
+                emitted += bool(await self.emit(
+                    source_trace_id=trace.trace_id,
+                    event_type="validation.completed",
+                    event_key=(
+                        f"validation.completed:trace:{trace.trace_id}:"
+                        f"{cache['plan_hash']}"
+                    ),
+                    effective_at_sequence=int(
+                        cache.get("validated_at_sequence", 0) or 0
+                    ),
+                    payload={
+                        "subject_type": "trace",
+                        "validation_result": cache["aggregate_result"],
+                    },
+                    project=False,
+                ))
+            for review in state.get("reviews", []):
+                sequence = int(review.get("reviewed_at_sequence", 0) or 0)
+                emitted += bool(await self.emit(
+                    source_trace_id=trace.trace_id,
+                    event_type="task.review.recorded",
+                    event_key=(
+                        f"task.review.recorded:{trace.trace_id}:"
+                        f"{review.get('child_trace_id')}:{sequence}"
+                    ),
+                    effective_at_sequence=sequence,
+                    payload={"task_review": review},
+                    project=False,
+                ))
+            for operation in trace.context.get("run_event_operations", []):
+                if operation.get("event_type") != "run.rewound":
+                    continue
+                emitted += bool(await self.emit(
+                    source_trace_id=trace.trace_id,
+                    event_type="run.rewound",
+                    event_key=f"run.rewound:{operation['operation_id']}",
+                    effective_at_sequence=operation["effective_at_sequence"],
+                    payload=operation["payload"],
+                    project=False,
+                ))
+            for raw in await self.store.get_events(trace.trace_id):
+                if raw.get("event") != "rewind":
+                    continue
+                sequence = int(raw.get("after_sequence", 0) or 0)
+                emitted += bool(await self.emit(
+                    source_trace_id=trace.trace_id,
+                    event_type="run.rewound",
+                    event_key=(
+                        f"run.rewound:{raw['operation_id']}"
+                        if raw.get("operation_id")
+                        else f"run.rewound:{trace.trace_id}:"
+                        f"{raw['event_id']}:{sequence}"
+                    ),
+                    effective_at_sequence=sequence,
+                    payload={"after_sequence": sequence},
+                    project=False,
+                ))
+        ledger = CandidateLedger.model_validate(
+            await self.store.get_candidate_ledger(root_trace_id) or {}
+        )
+        for candidate in ledger.candidates:
+            emitted += bool(await self.emit(
+                source_trace_id=candidate.owner_trace_id,
+                event_type="candidate.version_registered",
+                event_key=(
+                    f"candidate.version_registered:{candidate.candidate_id}:"
+                    f"{candidate.revision}"
+                ),
+                effective_at_sequence=candidate.created_at_sequence,
+                payload={"candidate_ref": candidate.model_dump(mode="json")},
+                project=False,
+            ))
+        for lifecycle in ledger.lifecycle:
+            candidate = ledger.candidate(lifecycle.candidate)
+            emitted += bool(await self.emit(
+                source_trace_id=lifecycle.source_trace_id or candidate.owner_trace_id,
+                event_type="candidate.lifecycle_changed",
+                event_key=(
+                    f"candidate.lifecycle_changed:{lifecycle.operation_id}:"
+                    f"{lifecycle.state}"
+                ),
+                effective_at_sequence=lifecycle.effective_at_sequence,
+                payload={"lifecycle": lifecycle.model_dump(mode="json")},
+                project=False,
+            ))
+        for raw in ledger.validations:
+            candidate = raw["candidate_ref"]
+            emitted += bool(await self.emit(
+                source_trace_id=candidate["owner_trace_id"],
+                event_type="validation.completed",
+                event_key=(
+                    f"validation.completed:candidate:{candidate['candidate_id']}:"
+                    f"{candidate['revision']}:{raw['plan_hash']}"
+                ),
+                effective_at_sequence=raw["validated_at_sequence"],
+                payload={
+                    "subject_type": "candidate",
+                    "candidate_ref": candidate,
+                    "validation_result": raw["validation_result"],
+                },
+                project=False,
+            ))
+        await self.try_pump(root_trace_id)
+        return emitted

cyber_agent/application/ports.py

@@ -55,3 +55,15 @@ class TaskContextProvider(Protocol):
         request: ContextRequest,
         descriptor: ContextResourceDescriptor,
     ) -> ContextMaterial: ...
+
+
+class RunEventProjector(Protocol):
+    """Application projection port; cursor movement is part of its transaction."""
+
+    async def load_cursor(
+        self,
+        application_ref: ApplicationRef,
+        root_trace_id: str,
+    ) -> int: ...
+
+    async def project(self, event: Any, expected_cursor: int) -> None: ...

cyber_agent/application/runtime.py

@@ -232,6 +232,13 @@ class ApplicationRuntime:
 
     def build_runner(self, binding: ApplicationBinding):
         from cyber_agent.core.runner import AgentRunner
+        from cyber_agent.application.events import RunEventService
+
+        event_service = RunEventService(
+            store=self.trace_store,
+            application_ref=binding.application_ref,
+            projector=binding.services.event_projector,
+        )
 
         candidate_service = None
         if binding.services.candidate_repository is not None:
@@ -246,6 +253,7 @@ class ApplicationRuntime:
                 application_binding=binding,
                 repository=binding.services.candidate_repository,
                 artifact_resolver=binding.services.artifact_resolver,
+                event_service=event_service,
             )
 
         return AgentRunner(
@@ -258,8 +266,33 @@ class ApplicationRuntime:
             application_binding=binding,
             context_provider=binding.services.context_provider,
             candidate_service=candidate_service,
+            event_service=event_service,
         )
 
+    async def reconcile_events(self, root_trace_id: str) -> int:
+        trace = await self.trace_store.get_trace(root_trace_id)
+        if trace is None:
+            raise ValueError(f"Trace not found: {root_trace_id}")
+        binding = self.registry.resolve_ref(trace.context.get("application_ref"))
+        from cyber_agent.application.events import RunEventService
+
+        return await RunEventService(
+            store=self.trace_store,
+            application_ref=binding.application_ref,
+            projector=binding.services.event_projector,
+        ).reconcile_root(root_trace_id)
+
+    async def reconcile_all_events(self) -> int:
+        roots = {
+            item.context.get("root_trace_id") or item.trace_id
+            for item in await self.trace_store.list_traces(limit=10_000)
+            if item.context.get("application_ref") is not None
+        }
+        total = 0
+        for root_trace_id in sorted(roots):
+            total += await self.reconcile_events(root_trace_id)
+        return total
+
     def new_run(
         self,
         application_id: str,
@@ -365,6 +398,9 @@ class ApplicationRuntime:
                 "Application snapshot identity does not match Trace metadata"
             )
         runner._validate_application_snapshot(snapshot, trace)
+        await self.reconcile_events(
+            trace.context.get("root_trace_id") or trace.trace_id
+        )
         return runner, config
 
     @staticmethod

cyber_agent/core/runner.py

@@ -340,6 +340,7 @@ class AgentRunner:
         application_binding: Any = None,
         context_provider: Any = None,
         candidate_service: Any = None,
+        event_service: Any = None,
     ):
         """
         初始化 AgentRunner
@@ -373,6 +374,7 @@ class AgentRunner:
         self.application_binding = application_binding
         self.context_provider = context_provider
         self.candidate_service = candidate_service
+        self.event_service = event_service
         self.stdin_check: Optional[Callable] = None  # 由外部设置，用于子 agent 执行期间检查 stdin
         self._cancel_events: Dict[str, asyncio.Event] = {}  # trace_id → cancel event
         self._recursive_active_traces: Dict[str, asyncio.Event] = {}
@@ -394,7 +396,7 @@ class AgentRunner:
             ResourceBudgetController(trace_store) if trace_store else None
         )
         self.task_protocol_service = (
-            TaskProtocolService(trace_store) if trace_store else None
+            TaskProtocolService(trace_store, event_service) if trace_store else None
         )
 
     # ===== 核心公开方法 =====
@@ -984,6 +986,20 @@ class AgentRunner:
                 evaluated_trace_id,
                 context=fresh.context,
             )
+            if self.event_service is not None:
+                await self.event_service.emit_after_commit(
+                    source_trace_id=evaluated_trace_id,
+                    event_type="validation.completed",
+                    event_key=(
+                        f"validation.completed:trace:{evaluated_trace_id}:"
+                        f"{plan.plan_hash}"
+                    ),
+                    effective_at_sequence=plan.evaluated_head_sequence,
+                    payload={
+                        "subject_type": "trace",
+                        "validation_result": run.result.model_dump(mode="json"),
+                    },
+                )
             return run
         finally:
             if lineage_event is not None:
@@ -4144,6 +4160,10 @@ class AgentRunner:
                 head_sequence=head_seq,
                 completed_at=datetime.now(),
             )
+            if self.event_service is not None:
+                await self.event_service.try_pump(
+                    trace.context.get("root_trace_id") or trace_id
+                )
             trace_obj = await self.trace_store.get_trace(trace_id)
             if trace_obj:
                 yield trace_obj
@@ -4228,6 +4248,16 @@ class AgentRunner:
         if self.candidate_service is not None:
             await self.candidate_service.assert_rewind_allowed(trace_id, cutoff)
 
+        rewind_operation_id = (
+            f"rewind:{trace_id}:"
+            f"{max((item.sequence for item in all_messages), default=0)}:{cutoff}"
+        )
+        rewind_payload: dict[str, Any] = {
+            "operation_id": rewind_operation_id,
+            "after_sequence": cutoff,
+            "head_sequence": cutoff,
+        }
+
         # 3. 快照并重建 GoalTree
         if goal_tree:
             # 获取截断点消息的 created_at 作为时间界限
@@ -4239,12 +4269,7 @@ class AgentRunner:
 
             cutoff_time = cutoff_msg.created_at if cutoff_msg else datetime.now()
 
-            # 快照到 events（含 head_sequence 供前端感知分支切换）
-            await self.trace_store.append_event(trace_id, "rewind", {
-                "after_sequence": cutoff,
-                "head_sequence": cutoff,
-                "goal_tree_snapshot": goal_tree.to_dict(),
-            })
+            rewind_payload["goal_tree_snapshot"] = goal_tree.to_dict()
 
             # 按时间重建干净的 GoalTree
             new_tree = goal_tree.rebuild_for_rewind(cutoff_time)
@@ -4331,6 +4356,19 @@ class AgentRunner:
             )
             state["protocol_correction_attempts"] = 0
             prune_context_access(trace.context, cutoff)
+            operations = trace.context.setdefault("run_event_operations", [])
+            if not any(
+                item.get("operation_id") == rewind_operation_id
+                for item in operations
+                if isinstance(item, dict)
+            ):
+                operations.append({
+                    "operation_id": rewind_operation_id,
+                    "event_type": "run.rewound",
+                    "effective_at_sequence": cutoff,
+                    "payload": {"after_sequence": cutoff},
+                })
+                del operations[:-64]
             await self.trace_store.update_trace(trace_id, context=trace.context)
 
             projected_statuses: dict[str, str] = {}
@@ -4350,6 +4388,15 @@ class AgentRunner:
 
         # 4. 更新 head_sequence 到 rewind 点
         await self.trace_store.update_trace(trace_id, head_sequence=cutoff)
+        await self.trace_store.append_event(trace_id, "rewind", rewind_payload)
+        if self.event_service is not None:
+            await self.event_service.emit_after_commit(
+                source_trace_id=trace_id,
+                event_type="run.rewound",
+                event_key=f"run.rewound:{rewind_operation_id}",
+                effective_at_sequence=cutoff,
+                payload={"after_sequence": cutoff},
+            )
 
         # 5. 返回 next sequence（全局递增，不复用）
         max_seq = max((m.sequence for m in all_messages), default=0)
@@ -5436,6 +5483,7 @@ class AgentRunner:
             "store": self.trace_store,
             "task_protocol_service": self.task_protocol_service,
             "candidate_service": self.candidate_service,
+            "event_service": self.event_service,
             "trace_id": trace_id,
             "goal_id": goal_id,
             "runner": self,

cyber_agent/core/task_protocol_service.py

@@ -30,8 +30,9 @@ class TaskProtocolStateError(RuntimeError):
 class TaskProtocolService:
     """将协议领域纯函数与 TraceStore 持久化组合起来。"""
 
-    def __init__(self, trace_store: TraceStore) -> None:
+    def __init__(self, trace_store: TraceStore, event_service=None) -> None:
         self.trace_store = trace_store
+        self.event_service = event_service
 
     def _lock(self, trace_id: str) -> asyncio.Lock:
         key = (asyncio.get_running_loop(), trace_id)
@@ -111,4 +112,14 @@ class TaskProtocolService:
                 ),
             )
             await self.trace_store.update_trace(trace_id, context=next_context)
+            if self.event_service is not None:
+                await self.event_service.emit_after_commit(
+                    source_trace_id=trace_id,
+                    event_type="task.progress.revised",
+                    event_key=(
+                        f"task.progress.revised:{trace_id}:{revision.revision}"
+                    ),
+                    effective_at_sequence=revision.effective_at_sequence,
+                    payload={"revision": revision.model_dump(mode="json")},
+                )
             return revision

cyber_agent/tools/builtin/task_protocol.py

@@ -511,6 +511,18 @@ async def review_task_result(
         state["pending_replans"].clear()
 
     await store.update_trace(parent_trace_id, context=parent.context)
+    event_service = context.get("event_service")
+    if event_service is not None:
+        await event_service.emit_after_commit(
+            source_trace_id=parent_trace_id,
+            event_type="task.review.recorded",
+            event_key=(
+                f"task.review.recorded:{parent_trace_id}:"
+                f"{child_trace_id}:{reviewed_at_sequence}"
+            ),
+            effective_at_sequence=reviewed_at_sequence,
+            payload={"task_review": review_record},
+        )
 
     goal_id = origin_goal_id
     if goal_id:

tests/test_application_events.py

@@ -0,0 +1,193 @@
+import tempfile
+import unittest
+from unittest.mock import patch
+
+from cyber_agent.application import (
+    AgentApplication,
+    AgentRole,
+    ApplicationRegistry,
+    ApplicationRuntime,
+    ApplicationServices,
+    ProviderRef,
+)
+from cyber_agent.application.events import RunEvent
+from cyber_agent.core.task_protocol import TaskProgress
+from cyber_agent.tools.registry import ToolRegistry
+from cyber_agent.trace.store import FileSystemTraceStore
+
+
+async def unused_llm(**kwargs):
+    raise AssertionError(f"LLM must not run: {kwargs}")
+
+
+class TransactionalProjector:
+    def __init__(self):
+        self.cursor = 0
+        self.view = {}
+        self.calls = []
+        self.fail_key = None
+        self.failed_once = False
+
+    async def load_cursor(self, application_ref, root_trace_id):
+        del application_ref, root_trace_id
+        return self.cursor
+
+    async def project(self, event, expected_cursor):
+        if expected_cursor != self.cursor:
+            raise ValueError("stale projector cursor")
+        self.calls.append(event.event_key)
+        self.view.setdefault(event.event_key, event.payload)
+        if event.event_key == self.fail_key and not self.failed_once:
+            self.failed_once = True
+            raise RuntimeError("failed after business data write")
+        self.cursor = event.event_id
+
+
+class ApplicationEventsTest(unittest.IsolatedAsyncioTestCase):
+    async def asyncSetUp(self):
+        self.temp = tempfile.TemporaryDirectory()
+        self.store = FileSystemTraceStore(self.temp.name)
+        self.projector = TransactionalProjector()
+        application = AgentApplication(
+            application_id="events.reference",
+            application_version="1",
+            root_role="writer",
+            roles=(AgentRole(
+                role_id="writer",
+                model="fake",
+                system_prompt="write",
+            ),),
+            event_projector_ref=ProviderRef(
+                provider_id="projector",
+                provider_version="1",
+            ),
+        )
+        self.registry = ApplicationRegistry()
+        self.binding = self.registry.register(
+            application,
+            ApplicationServices(
+                tool_registry=ToolRegistry(),
+                event_projector=self.projector,
+            ),
+        )
+        self.runtime = ApplicationRuntime(
+            registry=self.registry,
+            trace_store=self.store,
+            llm_call=unused_llm,
+        )
+        self.runner, config = self.runtime.new_run(
+            "events.reference",
+            "1",
+            uid="user-1",
+            root_task_anchor={
+                "objective": "write",
+                "completion_criteria": ["complete"],
+                "constraints": [],
+            },
+        )
+        with patch.dict("os.environ", {"AGENT_MODE": "recursive"}, clear=False):
+            self.trace, _goal, _sequence = await self.runner._prepare_new_trace(
+                [{"role": "user", "content": "start"}],
+                config,
+            )
+
+    async def asyncTearDown(self):
+        self.temp.cleanup()
+
+    async def test_projection_stops_in_order_and_replay_is_idempotent(self):
+        service = self.runner.event_service
+        await service.emit(
+            source_trace_id=self.trace.trace_id,
+            event_type="test.first",
+            event_key="first",
+            effective_at_sequence=1,
+            payload={"value": 1},
+        )
+        self.projector.fail_key = "second"
+        await service.emit(
+            source_trace_id=self.trace.trace_id,
+            event_type="test.second",
+            event_key="second",
+            effective_at_sequence=2,
+            payload={"value": 2},
+        )
+        self.assertEqual({"first", "second"}, set(self.projector.view))
+        self.projector.failed_once = False
+        await service.emit(
+            source_trace_id=self.trace.trace_id,
+            event_type="test.third",
+            event_key="third",
+            effective_at_sequence=3,
+            payload={"value": 3},
+        )
+        self.assertEqual({"first", "second"}, set(self.projector.view))
+        self.assertNotIn("third", self.projector.view)
+
+        self.projector.fail_key = None
+        await service.pump.pump(self.trace.trace_id)
+        self.assertEqual({"first", "second", "third"}, set(self.projector.view))
+        self.assertEqual(1, list(self.projector.view).count("second"))
+        self.assertEqual(3, self.projector.calls.count("second"))
+
+    async def test_unknown_schema_and_event_id_gap_are_safely_projected(self):
+        await self.store.append_event(
+            self.trace.trace_id,
+            "ordinary",
+            {"ignored": True},
+        )
+        event_id = await self.store.append_event(
+            self.trace.trace_id,
+            "run_event",
+            {
+                "schema_version": 99,
+                "event_type": "future.event",
+                "event_key": "future",
+                "application_ref": self.binding.application_ref.model_dump(
+                    mode="json"
+                ),
+                "root_trace_id": self.trace.trace_id,
+                "source_trace_id": self.trace.trace_id,
+                "effective_at_sequence": 1,
+                "payload": {"future": True},
+            },
+        )
+        await self.runner.event_service.pump.pump(self.trace.trace_id)
+        self.assertEqual(event_id, self.projector.cursor)
+        self.assertEqual({"future": True}, self.projector.view["future"])
+
+    async def test_event_append_failure_preserves_state_and_reconcile_repairs_it(self):
+        service = self.runner.event_service
+        original_append = self.store.append_event
+
+        async def fail_run_event(trace_id, event_type, payload):
+            if event_type == "run_event":
+                raise OSError("journal unavailable")
+            return await original_append(trace_id, event_type, payload)
+
+        with patch.object(self.store, "append_event", side_effect=fail_run_event):
+            revision = await self.runner.task_protocol_service.update_progress(
+                self.trace.trace_id,
+                expected_revision=1,
+                progress=TaskProgress(
+                    phase="investigating",
+                    decision_rationale="verify one fact before writing",
+                ),
+                effective_at_sequence=2,
+            )
+        self.assertEqual(2, revision.revision)
+        events = await self.store.get_events(self.trace.trace_id)
+        self.assertFalse(any(
+            item.get("event_key", "").endswith(":2") for item in events
+        ))
+
+        await self.runtime.reconcile_events(self.trace.trace_id)
+        events = await self.store.get_events(self.trace.trace_id)
+        repaired = [
+            RunEvent.model_validate({
+                key: item[key] for key in RunEvent.model_fields if key in item
+            })
+            for item in events
+            if item.get("event_key", "").endswith(":2")
+        ]
+        self.assertEqual(1, len(repaired))
+        self.assertEqual("task.progress.revised", repaired[0].event_type)