fix(events): 拒绝规范事件键的冲突重用

同一 event_key 再次写入时逐字段核对 schema、类型、应用、根 Trace、来源、sequence 和 payload；只有完全相同的幂等重放才复用 event_id，冲突事件稳定失败关闭。

补充相同 key 不同 payload 的对抗测试，避免业务投影视图静默接受被篡改的事件。

cyber_agent/application/events.py

@@ -157,6 +157,26 @@ class RunEventService:
                 None,
             )
             if existing is not None:
+                expected = {
+                    "schema_version": CURRENT_RUN_EVENT_SCHEMA_VERSION,
+                    "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,
+                }
+                actual = {
+                    key: existing.get(key)
+                    for key in expected
+                }
+                if actual != expected:
+                    raise ValueError(
+                        f"run_event key was reused with conflicting payload: {event_key}"
+                    )
                 event_id = int(existing["event_id"])
             else:
                 event_id = await self.store.append_event(

tests/test_application_events.py

@@ -188,6 +188,24 @@ class ApplicationEventsTest(unittest.IsolatedAsyncioTestCase):
         self.assertEqual(unknown_type_id, self.projector.cursor)
         self.assertNotIn("unknown-type", self.projector.view)
 
+    async def test_same_event_key_cannot_hide_a_conflicting_payload(self):
+        service = self.runner.event_service
+        await service.emit(
+            source_trace_id=self.trace.trace_id,
+            event_type="task.progress.revised",
+            event_key="stable-operation",
+            effective_at_sequence=1,
+            payload={"revision": 1},
+        )
+        with self.assertRaisesRegex(ValueError, "conflicting payload"):
+            await service.emit(
+                source_trace_id=self.trace.trace_id,
+                event_type="task.progress.revised",
+                event_key="stable-operation",
+                effective_at_sequence=1,
+                payload={"revision": 2},
+            )
+
     async def test_event_append_failure_preserves_state_and_reconcile_repairs_it(self):
         service = self.runner.event_service
         original_append = self.store.append_event