fix: decouple realtime alerts from snapshots

ai/events/clip_worker.py

@@ -404,6 +404,16 @@ def save_clip_to_mp4(task):
         raise RuntimeError(f"encoded clip is empty: {output_path}")
     return output_path
 
+def _stable_event_id(metadata):
+    if not isinstance(metadata, dict):
+        return None
+    event_id = metadata.get("eventId") or metadata.get("event_id")
+    if event_id is None:
+        return None
+    event_id = str(event_id).strip()
+    return event_id or None
+
+
 #ClipWriterWorker는 태스크를 큐에서 꺼내어 관리하는 작업 스레드의 역할만 담당하며, 실제 비디오 파일로 인코딩하여 기록하는 연산은 save_clip_to_mp4 함수가 처리하는 구조
 class ClipWriterWorker:
     def __init__(self, task_queue, uploader=upload_clip, publisher=None, mqtt_event_topic=None, snapshot_uploader=upload_snapshot_jpeg):
@@ -435,7 +445,7 @@ def _run(self):
                 try:
                     upload_result = self.uploader(output_path, task.metadata)
                     meta = task.metadata or {}
-                    event_id = meta.get("eventId") or meta.get("event_id") or meta.get("evidenceId")
+                    event_id = _stable_event_id(meta)
                     clip_key = upload_result.get("s3_key") if upload_result else None
                     clip_ok = bool(upload_result and upload_result.get("uploaded") and upload_result.get("url"))
                     duration = len(task.frames) / float(task.fps or 30.0)
@@ -485,12 +495,7 @@ def _run(self):
                     if upload_result and upload_result.get("uploaded") and upload_result.get("url") and self.publisher:
                         s3_url = upload_result["url"]
                         meta = task.metadata or {}
-                        # Prefer stable incident eventId from metadata; never use timestamp as id.
-                        event_id = (
-                            meta.get("eventId")
-                            or meta.get("event_id")
-                            or meta.get("evidenceId")
-                        )
+                        event_id = _stable_event_id(meta)
                         if not event_id:
                             print(
                                 f"[clip-worker] skip MQTT re-publish: missing eventId for camera={task.camera_id}",

scripts/serve_ai_overlay.py

@@ -38,7 +38,6 @@
     update_prediction_counts,
     update_tracking_summary,
     lifecycle_payload_kwargs,
-    snapshot_assist_meta_from_event_payload,
 )
 from ai.inference.rtsp_runtime import (
     log_detection_stage,
@@ -1061,23 +1060,6 @@ def _process_frame_impl(
             try:
                 event_publish_result = _publish_event(publisher, payload, topic_settings["event_topic"])
                 _record_publish_outcome(summary, "events_publish", event_publish_result, attempted=True)
-                # Single VLM snapshot assist hook (never blocks alert loop)
-                try:
-                    from ai.snapshot_assist_upload import submit_frame_snapshot_async
-                    assist_meta = snapshot_assist_meta_from_event_payload(
-                        payload,
-                        track_id=track_id,
-                        prediction=track_prediction,
-                        emit_decision=emit_decision,
-                    )
-                    submit_frame_snapshot_async(
-                        event_id=str(payload.get("eventId") or ""),
-                        camera_login_id=str(stream_id),
-                        frame=getattr(frame_packet, "frame", None),
-                        **assist_meta,
-                    )
-                except Exception as snap_exc:
-                    print(f"[snapshot-assist][warn] non-fatal upload schedule failed: {snap_exc}", flush=True)
             except Exception as exc:
                 _record_publish_outcome(summary, "events_publish", attempted=True)
                 print(f"[ai-worker][error] failed to publish event payload for camera={stream_id}: {exc}", file=sys.stderr, flush=True)
@@ -1099,11 +1081,6 @@ def _process_frame_impl(
                 "confidence": payload.get("confidence"),
                 "faint_prob": payload.get("faint_prob") or payload.get("faintProb"),
                 "consecutive_count": payload.get("consecutive_count") or payload.get("consecutiveCount"),
-                # Primary snapshot key must survive clip re-publish so back can attach Snapshot.
-                "snapshot_object_key": payload.get("snapshot_object_key") or payload.get("snapshotObjectKey"),
-                "snapshotObjectKey": payload.get("snapshotObjectKey") or payload.get("snapshot_object_key"),
-                "snapshot_url": payload.get("snapshot_url") or payload.get("snapshotUrl"),
-                "snapshotUrl": payload.get("snapshotUrl") or payload.get("snapshot_url"),
             }
 
             clip_triggered = state.clip_buffer.trigger_event(

tests/test_event_clip.py

@@ -3,10 +3,11 @@
 import time
 import unittest
 from pathlib import Path
+from unittest.mock import patch
 
 import numpy as np
 
-from ai.events.clip_worker import ClipWriterWorker, enqueue_event_clip, save_clip_to_mp4
+from ai.events.clip_worker import ClipWriterWorker, _stable_event_id, enqueue_event_clip, save_clip_to_mp4
 from ai.events.event_clip import CircularFrameBuffer, EventClipBuffer, EventClipTask
 
 
@@ -82,9 +83,14 @@ def test_clip_worker_prefers_event_id_not_timestamp(self):
             "confidence": 0.88,
             "faint_prob": 0.88,
         }
-        event_id = meta.get("eventId") or meta.get("event_id") or meta.get("evidenceId")
+        event_id = _stable_event_id(meta)
         self.assertEqual(event_id, "stable-evt-1")
         self.assertNotEqual(event_id, meta["event_timestamp"])
+
+    def test_clip_worker_rejects_evidence_id_as_event_id_fallback(self):
+        self.assertIsNone(_stable_event_id({"evidenceId": "derived-id", "event_timestamp": 1_700_000_000.0}))
+        self.assertIsNone(_stable_event_id({"eventId": "   "}))
+        self.assertEqual(_stable_event_id({"event_id": " stable-evt-2 "}), "stable-evt-2")
     def test_clip_worker_passes_snapshot_object_key_from_metadata(self):
         meta = {
             "eventId": "stable-evt-snap",
@@ -213,6 +219,99 @@ def test_save_clip_to_mp4_keeps_legacy_glob_with_evidence_suffix(self):
         self.assertTrue(output_path.name.startswith("Fall_cam_01_"))
         self.assertIn("evidence-cam_01-7-2000", output_path.name)
 
+    def test_worker_publishes_clip_and_thumbnail_with_stable_event_id(self):
+        published = []
+
+        class Publisher:
+            def publish_event(self, payload):
+                published.append(payload)
+
+        with tempfile.TemporaryDirectory() as temp_dir:
+            output_path = Path(temp_dir) / "clip.mp4"
+            output_path.write_bytes(b"clip")
+            task_queue = queue.Queue(maxsize=1)
+            worker = ClipWriterWorker(
+                task_queue,
+                uploader=lambda path, metadata: {
+                    "uploaded": True,
+                    "s3_key": "clips/stable-evt.mp4",
+                    "url": "https://example.test/clips/stable-evt.mp4",
+                },
+                publisher=Publisher(),
+                snapshot_uploader=lambda jpeg, event_id, metadata: {
+                    "uploaded": True,
+                    "s3_key": f"snapshots/{event_id}.jpg",
+                },
+            )
+            task = EventClipTask(
+                event_type="Fall",
+                camera_id="cam_01",
+                frames=[dummy_frame(7)],
+                fps=10.0,
+                output_dir=temp_dir,
+                metadata={
+                    "eventId": "stable-evt",
+                    "event_timestamp": "2026-07-17T00:00:00Z",
+                    "event_frame_index": 0,
+                },
+            )
+            with patch("ai.events.clip_worker.save_clip_to_mp4", return_value=output_path):
+                worker.start()
+                self.assertTrue(enqueue_event_clip(task_queue, task))
+                task_queue.join()
+                worker.stop()
+
+        self.assertEqual(len(published), 1)
+        self.assertEqual(published[0]["eventId"], "stable-evt")
+        self.assertEqual(published[0]["clip_object_key"], "clips/stable-evt.mp4")
+        self.assertEqual(published[0]["snapshot_object_key"], "snapshots/stable-evt.jpg")
+
+    def test_worker_snapshot_failure_still_publishes_clip_attachment(self):
+        published = []
+
+        class Publisher:
+            def publish_event(self, payload):
+                published.append(payload)
+
+        def failing_snapshot_upload(*_args, **_kwargs):
+            raise RuntimeError("S3 put_object failed")
+
+        with tempfile.TemporaryDirectory() as temp_dir:
+            output_path = Path(temp_dir) / "clip.mp4"
+            output_path.write_bytes(b"clip")
+            task_queue = queue.Queue(maxsize=1)
+            worker = ClipWriterWorker(
+                task_queue,
+                uploader=lambda path, metadata: {
+                    "uploaded": True,
+                    "s3_key": "clips/stable-evt.mp4",
+                    "url": "https://example.test/clips/stable-evt.mp4",
+                },
+                publisher=Publisher(),
+                snapshot_uploader=failing_snapshot_upload,
+            )
+            task = EventClipTask(
+                event_type="Fall",
+                camera_id="cam_01",
+                frames=[dummy_frame(7)],
+                fps=10.0,
+                output_dir=temp_dir,
+                metadata={
+                    "eventId": "stable-evt",
+                    "event_timestamp": "2026-07-17T00:00:00Z",
+                    "event_frame_index": 0,
+                },
+            )
+            with patch("ai.events.clip_worker.save_clip_to_mp4", return_value=output_path):
+                worker.start()
+                self.assertTrue(enqueue_event_clip(task_queue, task))
+                task_queue.join()
+                worker.stop()
+
+        self.assertEqual(len(published), 1)
+        self.assertEqual(published[0]["eventId"], "stable-evt")
+        self.assertEqual(published[0]["clip_object_key"], "clips/stable-evt.mp4")
+        self.assertNotIn("snapshot_object_key", published[0])
     def test_worker_failure_does_not_stop_worker_thread(self):
         with tempfile.TemporaryDirectory() as temp_dir:
             task_queue = queue.Queue(maxsize=2)

tests/test_primary_snapshot_decouple.py

@@ -12,9 +12,12 @@
 
 from __future__ import annotations
 
+import ast
 import os
+import time
 import unittest
-from unittest.mock import patch
+from pathlib import Path
+from unittest.mock import Mock, patch
 
 import numpy as np
 
@@ -35,6 +38,54 @@ def _dummy_frame():
     return np.zeros((8, 8, 3), dtype="uint8")
 
 
+class RealtimeEventPathTest(unittest.TestCase):
+    _SCRIPTS = (
+        Path(__file__).parents[1] / "scripts" / "serve_ai_overlay.py",
+        Path(__file__).parents[1] / "scripts" / "run_rtsp_inference.py",
+    )
+
+    def test_realtime_scripts_do_not_call_snapshot_or_vlm_network_hooks(self):
+        forbidden_calls = {
+            "attach_primary_snapshot_if_enabled",
+            "submit_frame_snapshot_async",
+            "submit_snapshot",
+            "upload_snapshot_jpeg",
+            "enqueue_vlm",
+        }
+        for script in self._SCRIPTS:
+            tree = ast.parse(script.read_text(encoding="utf-8"), filename=str(script))
+            called_names = set()
+            for node in ast.walk(tree):
+                if not isinstance(node, ast.Call):
+                    continue
+                if isinstance(node.func, ast.Name):
+                    called_names.add(node.func.id)
+                elif isinstance(node.func, ast.Attribute):
+                    called_names.add(node.func.attr)
+            self.assertTrue(forbidden_calls.isdisjoint(called_names), f"{script.name}: {forbidden_calls & called_names}")
+
+    def test_slow_s3_put_object_is_not_reached_before_initial_mqtt_publish(self):
+        from scripts.serve_ai_overlay import _publish_event
+
+        s3_client = Mock()
+        s3_client.put_object.side_effect = lambda **_kwargs: time.sleep(0.2)
+        published = []
+
+        class Publisher:
+            def enqueue_event(self, payload, topic):
+                published.append((payload, topic))
+                return True
+
+        started = time.perf_counter()
+        result = _publish_event(Publisher(), {"eventId": "evt-now"}, "safety/events")
+        elapsed = time.perf_counter() - started
+
+        self.assertTrue(result)
+        self.assertEqual(published, [({"eventId": "evt-now"}, "safety/events")])
+        s3_client.put_object.assert_not_called()
+        self.assertLess(elapsed, 0.1)
+
+
 class StorageBucketConfigTest(unittest.TestCase):
     def test_legacy_bucket_name_remains_supported(self):
         with patch.dict(os.environ, {"S3_BUCKET_NAME": "legacy-bucket"}, clear=True):