fix(events): 准确统计启动对账实际补发事件

保留 emit 返回 event_id 的兼容接口，内部同时返回 created 标记；reconcile_root 只在规范事件首次追加时增加补发计数。

新增连续两次对账回归：第一次修复缺失事件，第二次稳定返回零且不产生重复投影。

cyber_agent/application/events.py

@@ -129,7 +129,7 @@ class RunEventService:
             _ROOT_LOCKS[key] = lock
         return lock
 
-    async def emit(
+    async def _emit_record(
         self,
         *,
         source_trace_id: str,
@@ -138,7 +138,7 @@ class RunEventService:
         effective_at_sequence: int,
         payload: dict[str, Any],
         project: bool = True,
-    ) -> int:
+    ) -> tuple[int, bool]:
         source = await self.store.get_trace(source_trace_id)
         if source is None:
             raise ValueError(f"Trace not found: {source_trace_id}")
@@ -178,6 +178,7 @@ class RunEventService:
                         f"run_event key was reused with conflicting payload: {event_key}"
                     )
                 event_id = int(existing["event_id"])
+                created = False
             else:
                 event_id = await self.store.append_event(
                     root_trace_id,
@@ -195,10 +196,19 @@ class RunEventService:
                         "payload": payload,
                     },
                 )
+                created = True
         if project:
             await self.try_pump(root_trace_id)
+        return event_id, created
+
+    async def emit(self, **kwargs: Any) -> int:
+        event_id, _created = await self._emit_record(**kwargs)
         return event_id
 
+    async def _emit_missing(self, **kwargs: Any) -> int:
+        _event_id, created = await self._emit_record(**kwargs)
+        return int(created)
+
     async def emit_after_commit(self, **kwargs: Any) -> None:
         """Never roll back protocol state because journaling/projection failed."""
         try:
@@ -227,7 +237,7 @@ class RunEventService:
         for trace in traces:
             state = ensure_task_protocol(trace.context)
             for revision in state.get("task_progress_revisions", []):
-                emitted += bool(await self.emit(
+                emitted += await self._emit_missing(
                     source_trace_id=trace.trace_id,
                     event_type="task.progress.revised",
                     event_key=(
@@ -237,10 +247,10 @@ class RunEventService:
                     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(
+                emitted += await self._emit_missing(
                     source_trace_id=trace.trace_id,
                     event_type="validation.completed",
                     event_key=(
@@ -255,10 +265,10 @@ class RunEventService:
                         "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(
+                emitted += await self._emit_missing(
                     source_trace_id=trace.trace_id,
                     event_type="task.review.recorded",
                     event_key=(
@@ -268,23 +278,23 @@ class RunEventService:
                     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(
+                emitted += await self._emit_missing(
                     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(
+                emitted += await self._emit_missing(
                     source_trace_id=trace.trace_id,
                     event_type="run.rewound",
                     event_key=(
@@ -296,12 +306,12 @@ class RunEventService:
                     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(
+            emitted += await self._emit_missing(
                 source_trace_id=candidate.owner_trace_id,
                 event_type="candidate.version_registered",
                 event_key=(
@@ -311,10 +321,10 @@ class RunEventService:
                 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(
+            emitted += await self._emit_missing(
                 source_trace_id=lifecycle.source_trace_id or candidate.owner_trace_id,
                 event_type="candidate.lifecycle_changed",
                 event_key=(
@@ -324,10 +334,10 @@ class RunEventService:
                 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(
+            emitted += await self._emit_missing(
                 source_trace_id=candidate["owner_trace_id"],
                 event_type="validation.completed",
                 event_key=(
@@ -341,6 +351,6 @@ class RunEventService:
                     "validation_result": raw["validation_result"],
                 },
                 project=False,
-            ))
+            )
         await self.try_pump(root_trace_id)
         return emitted

tests/test_application_events.py

@@ -231,7 +231,12 @@ class ApplicationEventsTest(unittest.IsolatedAsyncioTestCase):
             item.get("event_key", "").endswith(":2") for item in events
         ))
 
-        await self.runtime.reconcile_events(self.trace.trace_id)
+        repaired_count = await self.runtime.reconcile_events(self.trace.trace_id)
+        self.assertGreater(repaired_count, 0)
+        self.assertEqual(
+            0,
+            await self.runtime.reconcile_events(self.trace.trace_id),
+        )
         events = await self.store.get_events(self.trace.trace_id)
         repaired = [
             RunEvent.model_validate({