fix(events): 未知规范事件只推进投影游标

RunEventPump 仅把 schema_version=1 且属于已知规范类型的事件交给业务 Projector；未来 schema 或未知类型改走独立 advance_cursor 事务，完全不暴露未知 payload 或改写业务视图。

扩展 Projector 端口和参考实现，增加未来 schema、当前 schema 未知类型及 event_id 缺口回归，确认游标前进后后续已知事件仍可严格顺序投影。

cyber_agent/application/events.py

@@ -16,6 +16,15 @@ from cyber_agent.core.task_protocol import ensure_task_protocol
 
 
 logger = logging.getLogger(__name__)
+CURRENT_RUN_EVENT_SCHEMA_VERSION = 1
+SUPPORTED_RUN_EVENT_TYPES = frozenset({
+    "task.progress.revised",
+    "candidate.version_registered",
+    "candidate.lifecycle_changed",
+    "validation.completed",
+    "task.review.recorded",
+    "run.rewound",
+})
 _ROOT_LOCKS: "WeakValueDictionary[tuple[int, str], asyncio.Lock]" = (
     WeakValueDictionary()
 )
@@ -68,7 +77,18 @@ class RunEventPump:
                 or event.root_trace_id != root_trace_id
             ):
                 raise ValueError("run_event application/root identity mismatch")
-            await self.projector.project(event, cursor)
+            if (
+                event.schema_version != CURRENT_RUN_EVENT_SCHEMA_VERSION
+                or event.event_type not in SUPPORTED_RUN_EVENT_TYPES
+            ):
+                await self.projector.advance_cursor(
+                    self.application_ref,
+                    root_trace_id,
+                    event.event_id,
+                    cursor,
+                )
+            else:
+                await self.projector.project(event, cursor)
             next_cursor = await self.projector.load_cursor(
                 self.application_ref,
                 root_trace_id,
@@ -143,7 +163,7 @@ class RunEventService:
                     root_trace_id,
                     "run_event",
                     {
-                        "schema_version": 1,
+                        "schema_version": CURRENT_RUN_EVENT_SCHEMA_VERSION,
                         "event_type": event_type,
                         "event_key": event_key,
                         "application_ref": self.application_ref.model_dump(

cyber_agent/application/ports.py

@@ -67,3 +67,13 @@ class RunEventProjector(Protocol):
     ) -> int: ...
 
     async def project(self, event: Any, expected_cursor: int) -> None: ...
+
+    async def advance_cursor(
+        self,
+        application_ref: ApplicationRef,
+        root_trace_id: str,
+        event_id: int,
+        expected_cursor: int,
+    ) -> None:
+        """Atomically skip an unsupported event without updating business views."""
+        ...

examples/application_reference/providers/projector.py

@@ -14,3 +14,15 @@ class ReferenceProjector:
             raise ValueError("stale projection cursor")
         self.rows.setdefault(event.event_key, event.payload)
         self.cursor = event.event_id
+
+    async def advance_cursor(
+        self,
+        application_ref,
+        root_trace_id,
+        event_id,
+        expected_cursor,
+    ):
+        del application_ref, root_trace_id
+        if expected_cursor != self.cursor:
+            raise ValueError("stale projection cursor")
+        self.cursor = event_id

tests/test_application_events.py

@@ -42,6 +42,18 @@ class TransactionalProjector:
             raise RuntimeError("failed after business data write")
         self.cursor = event.event_id
 
+    async def advance_cursor(
+        self,
+        application_ref,
+        root_trace_id,
+        event_id,
+        expected_cursor,
+    ):
+        del application_ref, root_trace_id
+        if expected_cursor != self.cursor:
+            raise ValueError("stale projector cursor")
+        self.cursor = event_id
+
 
 class ApplicationEventsTest(unittest.IsolatedAsyncioTestCase):
     async def asyncSetUp(self):
@@ -98,7 +110,7 @@ class ApplicationEventsTest(unittest.IsolatedAsyncioTestCase):
         service = self.runner.event_service
         await service.emit(
             source_trace_id=self.trace.trace_id,
-            event_type="test.first",
+            event_type="task.progress.revised",
             event_key="first",
             effective_at_sequence=1,
             payload={"value": 1},
@@ -106,7 +118,7 @@ class ApplicationEventsTest(unittest.IsolatedAsyncioTestCase):
         self.projector.fail_key = "second"
         await service.emit(
             source_trace_id=self.trace.trace_id,
-            event_type="test.second",
+            event_type="candidate.version_registered",
             event_key="second",
             effective_at_sequence=2,
             payload={"value": 2},
@@ -115,7 +127,7 @@ class ApplicationEventsTest(unittest.IsolatedAsyncioTestCase):
         self.projector.failed_once = False
         await service.emit(
             source_trace_id=self.trace.trace_id,
-            event_type="test.third",
+            event_type="validation.completed",
             event_key="third",
             effective_at_sequence=3,
             payload={"value": 3},
@@ -129,7 +141,7 @@ class ApplicationEventsTest(unittest.IsolatedAsyncioTestCase):
         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):
+    async def test_unknown_schema_and_event_id_gap_advance_without_projection(self):
         await self.store.append_event(
             self.trace.trace_id,
             "ordinary",
@@ -153,7 +165,28 @@ class ApplicationEventsTest(unittest.IsolatedAsyncioTestCase):
         )
         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"])
+        self.assertNotIn("future", self.projector.view)
+        self.assertNotIn({"future": True}, self.projector.view.values())
+
+        unknown_type_id = await self.store.append_event(
+            self.trace.trace_id,
+            "run_event",
+            {
+                "schema_version": 1,
+                "event_type": "unknown.current-schema-event",
+                "event_key": "unknown-type",
+                "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": 2,
+                "payload": {"must_not_reach_view": True},
+            },
+        )
+        await self.runner.event_service.pump.pump(self.trace.trace_id)
+        self.assertEqual(unknown_type_id, self.projector.cursor)
+        self.assertNotIn("unknown-type", self.projector.view)
 
     async def test_event_append_failure_preserves_state_and_reconcile_repairs_it(self):
         service = self.runner.event_service