this may not be shown

server/runtime.py

@@ -6,6 +6,7 @@ import threading
 import time
 from dataclasses import dataclass
 from datetime import datetime
+from pathlib import Path
 from queue import Empty, Queue
 from typing import Any, Callable
 
@@ -54,6 +55,7 @@ class PipelineManager:
         self._thread: threading.Thread | None = None
         self._start_thread: threading.Thread | None = None
         self._sinks: list[Sink] = []
+        self._session_id = 0
 
         self._source_factory: SourceFactory | None = None
         self._sink_factory: SinkFactory | None = None
@@ -62,6 +64,7 @@ class PipelineManager:
         self._source_spec: str = "0"
         self.running = False
         self.starting = False
+        self.stopping = False
         self.paused = False
         self.last_error: str | None = None
 
@@ -78,15 +81,17 @@ class PipelineManager:
             self._pipeline_factory = pipeline_factory
             self._source_spec = initial_source
 
-    def _close_sinks(self):
-        for sink in self._sinks:
+    def _close_sinks(self, sinks: list[Sink] | None = None):
+        sink_list = self._sinks if sinks is None else sinks
+        for sink in sink_list:
             try:
                 sink.close()
             except Exception:
                 logger.exception("Error closing sink: %s", sink.__class__.__name__)
-        self._sinks = []
+        if sinks is None:
+            self._sinks = []
 
-    def _run_wrapper(self):
+    def _run_wrapper(self, session_id: int):
         global pipeline, source
         assert pipeline is not None and source is not None
         try:
@@ -97,11 +102,14 @@ class PipelineManager:
             logger.exception("Pipeline crashed")
         finally:
             with self._lock:
-                self.running = False
-                # Only auto-close sinks when pipeline exits by itself
-                self._close_sinks()
-                pipeline = None
-                source = None
+                if self._session_id == session_id:
+                    self.running = False
+                    self.stopping = False
+                    self.starting = False
+                    self.paused = False
+                    self._close_sinks()
+                    pipeline = None
+                    source = None
 
     def start(self, source_spec: str | None = None) -> tuple[bool, str]:
         global pipeline, source, live_frame
@@ -109,6 +117,9 @@ class PipelineManager:
             if self.starting:
                 return False, "Pipeline is starting"
 
+            if self.stopping:
+                return False, "Pipeline is stopping"
+
             if self.running:
                 return False, "Pipeline already running"
 
@@ -151,9 +162,12 @@ class PipelineManager:
             self._sinks = sinks_obj
             pipeline = pipeline_obj
             live_frame = None
+            self._session_id += 1
+            session_id = self._session_id
 
             self._thread = threading.Thread(
                 target=self._run_wrapper,
+                args=(session_id,),
                 daemon=True,
                 name="pipeline",
             )
@@ -189,8 +203,13 @@ class PipelineManager:
     def stop(self) -> tuple[bool, str]:
         global pipeline, source
         with self._lock:
+            if self.stopping:
+                return False, "Pipeline is stopping"
+
+            self.stopping = True
             thread = self._thread
             source_obj = source
+            stop_session_id = self._session_id
 
         if source_obj is not None:
             try:
@@ -202,8 +221,12 @@ class PipelineManager:
             thread.join(timeout=3.0)
 
         with self._lock:
+            if self._session_id != stop_session_id:
+                self.stopping = False
+                return True, "Pipeline stop request completed"
             self.starting = False
             self.running = False
+            self.stopping = False
             self.paused = False
             self._close_sinks()
             pipeline = None
@@ -256,6 +279,7 @@ class PipelineManager:
             return {
                 "running": self.running,
                 "starting": self.starting,
+                "stopping": self.stopping,
                 "paused": self.paused,
                 "source": self._source_spec,
                 "last_error": self.last_error,
@@ -287,9 +311,27 @@ class ClipJob:
             "started_at": self.started_at,
             "finished_at": self.finished_at,
             "error": self.error,
+            "outputs": clip_job_outputs(self.job_id),
         }
 
 
+def clip_job_outputs(job_id: str) -> dict[str, Any]:
+    base_local = Path("static") / "clip_jobs" / job_id
+    events_local = base_local / "events"
+
+    event_clips: list[str] = []
+    if events_local.exists():
+        for event_file in sorted(events_local.glob("event_*.mp4")):
+            event_clips.append(f"/static/clip_jobs/{job_id}/events/{event_file.name}")
+
+    return {
+        "processed_video": f"/static/clip_jobs/{job_id}/processed.mp4",
+        "alerts_log": f"/static/clip_jobs/{job_id}/alerts.jsonl",
+        "events_dir": f"/static/clip_jobs/{job_id}/events/",
+        "event_clips": event_clips,
+    }
+
+
 class ClipJobManager:
     """Simple queued clip processor with a single worker."""