this may not be shown

.gitignore

@@ -16,7 +16,7 @@ wheels/
 alerts.jsonl
 output.jsonl
 output.mp4
-static/alerts/
+static/
 screenshots/
 events/
 logs/

README.md

@@ -63,6 +63,9 @@ The server runs the detection pipeline in the background while serving a REST AP
 # Default: webcam 0, classic pipeline
 uv run server.py
 
+# Package entrypoint (same server)
+uv run -m server
+
 # Use stage-based pipeline with a specific camera
 uv run server.py --staged --source 2
 
@@ -76,6 +79,14 @@ uv run server.py --host 0.0.0.0 --port 9000
 uvicorn server:app --host 0.0.0.0 --port 8000
 ```
 
+### Live + Clip Control (Server Runtime)
+
+The server now runs a lifecycle manager for the live pipeline and a queued clip-job worker:
+
+- **Live manager**: start/stop/pause/resume/switch source at runtime.
+- **Clip jobs**: submit files to a queue, process one-at-a-time, track status.
+- **Resource policy**: clip jobs can temporarily pause live processing and resume after job completion.
+
 On startup the server prints the LAN address for the mobile app:
 
 ```
@@ -122,7 +133,7 @@ uv run main.py --source videos/test.mp4 --debug
 ### Check Available Webcams
 
 ```bash
-uv run -m tools.webcam_check
+uv run -m scripts.webcam_check
 ```
 
 ### CLI Arguments
@@ -181,6 +192,10 @@ The system generates multiple artifacts during execution:
 ```text
 gp/
 ├── server.py                # FastAPI server (runs pipeline + REST API)
+├── server/                  # Modular server runtime + routes
+│   ├── runtime.py           # PipelineManager + ClipJobManager
+│   ├── routes.py            # API routers (/live/*, /clips/*, auth/history)
+│   └── __main__.py          # `uv run -m server` entrypoint
 ├── main.py                  # CLI entry point (standalone pipeline)
 ├── pyproject.toml           # Project config (uv/pip)
 ├── models/                  # Directory for .pt model weights
@@ -228,8 +243,9 @@ gp/
 │       ├── geometry.py      # Box expansion, zone filtering
 │       └── drawing.py       # Visualization helpers
 │
-├── tools/
+├── scripts/
-│   └── webcam_check.py      # Utility to list available webcams
+│   ├── webcam_check.py      # Utility to list available webcams
+│   └── live_stress_test.py  # Stress test for /live/* and queue responsiveness
 │
 └── SmokeDetectionApp/       # Android companion app
     └── app/src/main/java/com/example/smokedetection/
@@ -289,17 +305,56 @@ pipeline = SmokingDetectionPipeline(..., config=config)
 The server (`server.py`) exposes these endpoints:
 
 | Method | Endpoint              | Description                                     |
-| :----- | :--------------- | :---------------------------------- |
+| :----- | :-------------------- | :---------------------------------------------- |
 | GET    | `/ping`               | Health check (verify connectivity)              |
 | POST   | `/register`           | Create a new user account                       |
 | POST   | `/login`              | Authenticate and get user ID                    |
-| GET    | `/history`       | Get detection records for a user    |
+| GET    | `/history`            | Get detection records (optionally by `user_id`) |
-| POST   | `/clear_history` | Delete all records for a user       |
+| POST   | `/clear_history`      | Delete detection records (optional `user_id`)   |
 | GET    | `/snapshot`           | Latest camera frame as JPEG                     |
 | GET    | `/video_feed`         | MJPEG stream with detection overlay             |
+| GET    | `/live/status`        | Live pipeline state (`running/paused/source`)   |
+| POST   | `/live/start`         | Start live pipeline (optional `source`)         |
+| POST   | `/live/stop`          | Stop live pipeline                              |
+| POST   | `/live/pause`         | Pause live pipeline                             |
+| POST   | `/live/resume`        | Resume paused live pipeline                     |
+| POST   | `/live/switch_source` | Switch active webcam/RTSP source                |
+| POST   | `/clips/submit`       | Queue a clip job (upload file or local path)    |
+| GET    | `/clips`              | List recent clip jobs + queue stats             |
+| GET    | `/clips/{job_id}`     | Get status/details for a clip job               |
 
 Interactive docs at **<http://localhost:8000/docs>** (Swagger UI).
 
+### Quick Test Calls
+
+PowerShell examples:
+
+```powershell
+$base='http://127.0.0.1:8000'
+
+# Live controls
+Invoke-WebRequest -UseBasicParsing -Uri ($base+'/live/status')
+Invoke-WebRequest -UseBasicParsing -Method Post -Uri ($base+'/live/switch_source') -Body @{source='2'}
+Invoke-WebRequest -UseBasicParsing -Method Post -Uri ($base+'/live/pause')
+Invoke-WebRequest -UseBasicParsing -Method Post -Uri ($base+'/live/resume')
+
+# Queue clip from local path
+Invoke-WebRequest -UseBasicParsing -Method Post -Uri ($base+'/clips/submit') -Body @{path='test_media/139474-772542450_medium.mp4'}
+
+# List jobs and inspect one job
+Invoke-WebRequest -UseBasicParsing -Uri ($base+'/clips')
+Invoke-WebRequest -UseBasicParsing -Uri ($base+'/clips/<job_id>')
+```
+
+curl examples:
+
+```bash
+curl http://127.0.0.1:8000/live/status
+curl -X POST -F "source=2" http://127.0.0.1:8000/live/switch_source
+curl -X POST -F "path=test_media/139474-772542450_medium.mp4" http://127.0.0.1:8000/clips/submit
+curl http://127.0.0.1:8000/clips
+```
+
 ## 📱 Android App
 
 The companion app (`SmokeDetectionApp/`) connects to the server for auth, alert history, and live camera feed.

pipeline/sinks/__init__.py

@@ -1,7 +1,7 @@
 """Sinks for pipeline output handling."""
 
 from pipeline.sinks.alerts import FileAlertStore
-from pipeline.sinks.database import DatabaseSink
+from pipeline.sinks.database import DatabaseSink, DatabaseSmartEventSink
 from pipeline.sinks.display import WindowDisplaySink
 from pipeline.sinks.events import SmartEventSink
 from pipeline.sinks.metrics import MetricsSink
@@ -10,6 +10,7 @@ from pipeline.sinks.video import VideoFileStore
 
 __all__ = [
     "DatabaseSink",
+    "DatabaseSmartEventSink",
     "FileAlertStore",
     "MetricsSink",
     "ScreenshotSink",

pipeline/sinks/database.py

@@ -8,6 +8,7 @@ import cv2
 
 from db.repository import Repository
 from pipeline.ops.drawing import draw_alert_banner, draw_detections
+from pipeline.sinks.events import SmartEventSink
 from pipeline.types import Frame
 
 logger = logging.getLogger(__name__)
@@ -72,3 +73,52 @@ class DatabaseSink:
 
     def close(self):
         pass  # DB connection is managed by the repository
+
+
+class DatabaseSmartEventSink(SmartEventSink):
+    """Records event clips and stores clip path in DB as alert evidence."""
+
+    def __init__(
+        self,
+        db_repo: Repository,
+        fps: float,
+        user_id: int = 1,
+        camera_source: str = "Pipeline",
+        output_dir: str = "static/events",
+        pre_event_seconds: float = 2,
+        post_event_seconds: float = 3,
+    ):
+        super().__init__(
+            output_dir=output_dir,
+            fps=fps,
+            pre_event_seconds=pre_event_seconds,
+            post_event_seconds=post_event_seconds,
+        )
+        self.db_repo = db_repo
+        self.user_id = user_id
+        self.camera_source = camera_source
+
+    def _trigger_event(self, trigger_frame: Frame):
+        was_recording = self.recording
+        super()._trigger_event(trigger_frame)
+
+        # Save one DB row per newly started event clip
+        if was_recording or not self.recording or self.current_event_id is None:
+            return
+
+        ticket = trigger_frame.ticket
+        if not ticket:
+            return
+
+        filename = f"event_{self.current_event_id}.mp4"
+        web_path = f"/static/events/{filename}"
+        try:
+            self.db_repo.save_detection(
+                user_id=self.user_id,
+                image_path=web_path,
+                confidence=ticket.confidence,
+                camera_source=self.camera_source,
+            )
+            logger.info("Saved event alert to DB: %s", web_path)
+        except Exception as e:
+            logger.error("Failed to save event alert to DB: %s", e)

reports/live_stress_latest.json

@@ -0,0 +1,622 @@
+{
+  "started_at": "2026-03-01T19:28:11",
+  "base_url": "http://127.0.0.1:8000",
+  "config": {
+    "cycles": 20,
+    "request_timeout": 8.0,
+    "sleep_ms": 120,
+    "pause_every": 3,
+    "pause_sleep_ms": 120,
+    "source_a": "0",
+    "source_b": "2"
+  },
+  "summary": {
+    "total": 75,
+    "ok": 75,
+    "err": 0,
+    "avg_ms": 286.4,
+    "max_ms": 1051,
+    "over_500ms": 16,
+    "over_1000ms": 12
+  },
+  "errors": [],
+  "tail": [
+    {
+      "name": "cycle18-pause",
+      "ok": true,
+      "duration_ms": 1003,
+      "status_code": 200,
+      "body": "{\"status\":\"success\",\"message\":\"Pipeline paused\",\"live\":{\"running\":false,\"starting\":false,\"paused\":true,\"source\":\"2\",\"last_error\":null,\"thread_alive\":false}}"
+    },
+    {
+      "name": "cycle18-resume",
+      "ok": true,
+      "duration_ms": 357,
+      "status_code": 200,
+      "body": "{\"status\":\"success\",\"message\":\"Pipeline resumed\",\"live\":{\"running\":true,\"starting\":false,\"paused\":false,\"source\":\"2\",\"last_error\":null,\"thread_alive\":true}}"
+    },
+    {
+      "name": "cycle18-status-after",
+      "ok": true,
+      "duration_ms": 1,
+      "status_code": 200,
+      "body": "{\"status\":\"success\",\"live\":{\"running\":true,\"starting\":false,\"paused\":false,\"source\":\"2\",\"last_error\":null,\"thread_alive\":true}}"
+    },
+    {
+      "name": "cycle19-status-before",
+      "ok": true,
+      "duration_ms": 1,
+      "status_code": 200,
+      "body": "{\"status\":\"success\",\"live\":{\"running\":true,\"starting\":false,\"paused\":false,\"source\":\"2\",\"last_error\":null,\"thread_alive\":true}}"
+    },
+    {
+      "name": "cycle19-switch-0",
+      "ok": true,
+      "duration_ms": 1001,
+      "status_code": 200,
+      "body": "{\"status\":\"success\",\"message\":\"Pipeline started (source=0)\",\"live\":{\"running\":true,\"starting\":false,\"paused\":false,\"source\":\"0\",\"last_error\":null,\"thread_alive\":true}}"
+    },
+    {
+      "name": "cycle19-status-after",
+      "ok": true,
+      "duration_ms": 18,
+      "status_code": 200,
+      "body": "{\"status\":\"success\",\"live\":{\"running\":true,\"starting\":false,\"paused\":false,\"source\":\"0\",\"last_error\":null,\"thread_alive\":true}}"
+    },
+    {
+      "name": "cycle20-status-before",
+      "ok": true,
+      "duration_ms": 2,
+      "status_code": 200,
+      "body": "{\"status\":\"success\",\"live\":{\"running\":true,\"starting\":false,\"paused\":false,\"source\":\"0\",\"last_error\":null,\"thread_alive\":true}}"
+    },
+    {
+      "name": "cycle20-switch-2",
+      "ok": true,
+      "duration_ms": 401,
+      "status_code": 200,
+      "body": "{\"status\":\"success\",\"message\":\"Pipeline started (source=2)\",\"live\":{\"running\":true,\"starting\":false,\"paused\":false,\"source\":\"2\",\"last_error\":null,\"thread_alive\":true}}"
+    },
+    {
+      "name": "cycle20-status-after",
+      "ok": true,
+      "duration_ms": 2,
+      "status_code": 200,
+      "body": "{\"status\":\"success\",\"live\":{\"running\":true,\"starting\":false,\"paused\":false,\"source\":\"2\",\"last_error\":null,\"thread_alive\":true}}"
+    },
+    {
+      "name": "final-status",
+      "ok": true,
+      "duration_ms": 1,
+      "status_code": 200,
+      "body": "{\"status\":\"success\",\"live\":{\"running\":true,\"starting\":false,\"paused\":false,\"source\":\"2\",\"last_error\":null,\"thread_alive\":true}}"
+    }
+  ],
+  "all_results": [
+    {
+      "name": "warmup-status",
+      "ok": true,
+      "duration_ms": 23,
+      "status_code": 200,
+      "body": "{\"status\":\"success\",\"live\":{\"running\":true,\"starting\":false,\"paused\":false,\"source\":\"2\",\"last_error\":null,\"thread_alive\":true}}"
+    },
+    {
+      "name": "warmup-start",
+      "ok": true,
+      "duration_ms": 1,
+      "status_code": 200,
+      "body": "{\"status\":\"fail\",\"message\":\"Pipeline already running\",\"live\":{\"running\":true,\"starting\":false,\"paused\":false,\"source\":\"2\",\"last_error\":null,\"thread_alive\":true}}"
+    },
+    {
+      "name": "cycle1-status-before",
+      "ok": true,
+      "duration_ms": 1,
+      "status_code": 200,
+      "body": "{\"status\":\"success\",\"live\":{\"running\":true,\"starting\":false,\"paused\":false,\"source\":\"2\",\"last_error\":null,\"thread_alive\":true}}"
+    },
+    {
+      "name": "cycle1-switch-0",
+      "ok": true,
+      "duration_ms": 275,
+      "status_code": 200,
+      "body": "{\"status\":\"success\",\"message\":\"Pipeline started (source=0)\",\"live\":{\"running\":true,\"starting\":false,\"paused\":false,\"source\":\"0\",\"last_error\":null,\"thread_alive\":true}}"
+    },
+    {
+      "name": "cycle1-status-after",
+      "ok": true,
+      "duration_ms": 1,
+      "status_code": 200,
+      "body": "{\"status\":\"success\",\"live\":{\"running\":true,\"starting\":false,\"paused\":false,\"source\":\"0\",\"last_error\":null,\"thread_alive\":true}}"
+    },
+    {
+      "name": "cycle2-status-before",
+      "ok": true,
+      "duration_ms": 2,
+      "status_code": 200,
+      "body": "{\"status\":\"success\",\"live\":{\"running\":true,\"starting\":false,\"paused\":false,\"source\":\"0\",\"last_error\":null,\"thread_alive\":true}}"
+    },
+    {
+      "name": "cycle2-switch-2",
+      "ok": true,
+      "duration_ms": 520,
+      "status_code": 200,
+      "body": "{\"status\":\"success\",\"message\":\"Pipeline started (source=2)\",\"live\":{\"running\":true,\"starting\":false,\"paused\":false,\"source\":\"2\",\"last_error\":null,\"thread_alive\":true}}"
+    },
+    {
+      "name": "cycle2-status-after",
+      "ok": true,
+      "duration_ms": 24,
+      "status_code": 200,
+      "body": "{\"status\":\"success\",\"live\":{\"running\":true,\"starting\":false,\"paused\":false,\"source\":\"2\",\"last_error\":null,\"thread_alive\":true}}"
+    },
+    {
+      "name": "cycle3-status-before",
+      "ok": true,
+      "duration_ms": 2,
+      "status_code": 200,
+      "body": "{\"status\":\"success\",\"live\":{\"running\":true,\"starting\":false,\"paused\":false,\"source\":\"2\",\"last_error\":null,\"thread_alive\":true}}"
+    },
+    {
+      "name": "cycle3-switch-0",
+      "ok": true,
+      "duration_ms": 1051,
+      "status_code": 200,
+      "body": "{\"status\":\"success\",\"message\":\"Pipeline started (source=0)\",\"live\":{\"running\":true,\"starting\":false,\"paused\":false,\"source\":\"0\",\"last_error\":null,\"thread_alive\":true}}"
+    },
+    {
+      "name": "cycle3-pause",
+      "ok": true,
+      "duration_ms": 1003,
+      "status_code": 200,
+      "body": "{\"status\":\"success\",\"message\":\"Pipeline paused\",\"live\":{\"running\":false,\"starting\":false,\"paused\":true,\"source\":\"0\",\"last_error\":null,\"thread_alive\":false}}"
+    },
+    {
+      "name": "cycle3-resume",
+      "ok": true,
+      "duration_ms": 151,
+      "status_code": 200,
+      "body": "{\"status\":\"success\",\"message\":\"Pipeline resumed\",\"live\":{\"running\":true,\"starting\":false,\"paused\":false,\"source\":\"0\",\"last_error\":null,\"thread_alive\":true}}"
+    },
+    {
+      "name": "cycle3-status-after",
+      "ok": true,
+      "duration_ms": 2,
+      "status_code": 200,
+      "body": "{\"status\":\"success\",\"live\":{\"running\":true,\"starting\":false,\"paused\":false,\"source\":\"0\",\"last_error\":null,\"thread_alive\":true}}"
+    },
+    {
+      "name": "cycle4-status-before",
+      "ok": true,
+      "duration_ms": 2,
+      "status_code": 200,
+      "body": "{\"status\":\"success\",\"live\":{\"running\":true,\"starting\":false,\"paused\":false,\"source\":\"0\",\"last_error\":null,\"thread_alive\":true}}"
+    },
+    {
+      "name": "cycle4-switch-2",
+      "ok": true,
+      "duration_ms": 494,
+      "status_code": 200,
+      "body": "{\"status\":\"success\",\"message\":\"Pipeline started (source=2)\",\"live\":{\"running\":true,\"starting\":false,\"paused\":false,\"source\":\"2\",\"last_error\":null,\"thread_alive\":true}}"
+    },
+    {
+      "name": "cycle4-status-after",
+      "ok": true,
+      "duration_ms": 1,
+      "status_code": 200,
+      "body": "{\"status\":\"success\",\"live\":{\"running\":true,\"starting\":false,\"paused\":false,\"source\":\"2\",\"last_error\":null,\"thread_alive\":true}}"
+    },
+    {
+      "name": "cycle5-status-before",
+      "ok": true,
+      "duration_ms": 2,
+      "status_code": 200,
+      "body": "{\"status\":\"success\",\"live\":{\"running\":true,\"starting\":false,\"paused\":false,\"source\":\"2\",\"last_error\":null,\"thread_alive\":true}}"
+    },
+    {
+      "name": "cycle5-switch-0",
+      "ok": true,
+      "duration_ms": 1024,
+      "status_code": 200,
+      "body": "{\"status\":\"success\",\"message\":\"Pipeline started (source=0)\",\"live\":{\"running\":true,\"starting\":false,\"paused\":false,\"source\":\"0\",\"last_error\":null,\"thread_alive\":true}}"
+    },
+    {
+      "name": "cycle5-status-after",
+      "ok": true,
+      "duration_ms": 1,
+      "status_code": 200,
+      "body": "{\"status\":\"success\",\"live\":{\"running\":true,\"starting\":false,\"paused\":false,\"source\":\"0\",\"last_error\":null,\"thread_alive\":true}}"
+    },
+    {
+      "name": "cycle6-status-before",
+      "ok": true,
+      "duration_ms": 1,
+      "status_code": 200,
+      "body": "{\"status\":\"success\",\"live\":{\"running\":true,\"starting\":false,\"paused\":false,\"source\":\"0\",\"last_error\":null,\"thread_alive\":true}}"
+    },
+    {
+      "name": "cycle6-switch-2",
+      "ok": true,
+      "duration_ms": 451,
+      "status_code": 200,
+      "body": "{\"status\":\"success\",\"message\":\"Pipeline started (source=2)\",\"live\":{\"running\":true,\"starting\":false,\"paused\":false,\"source\":\"2\",\"last_error\":null,\"thread_alive\":true}}"
+    },
+    {
+      "name": "cycle6-pause",
+      "ok": true,
+      "duration_ms": 1006,
+      "status_code": 200,
+      "body": "{\"status\":\"success\",\"message\":\"Pipeline paused\",\"live\":{\"running\":false,\"starting\":false,\"paused\":true,\"source\":\"2\",\"last_error\":null,\"thread_alive\":false}}"
+    },
+    {
+      "name": "cycle6-resume",
+      "ok": true,
+      "duration_ms": 341,
+      "status_code": 200,
+      "body": "{\"status\":\"success\",\"message\":\"Pipeline resumed\",\"live\":{\"running\":true,\"starting\":false,\"paused\":false,\"source\":\"2\",\"last_error\":null,\"thread_alive\":true}}"
+    },
+    {
+      "name": "cycle6-status-after",
+      "ok": true,
+      "duration_ms": 1,
+      "status_code": 200,
+      "body": "{\"status\":\"success\",\"live\":{\"running\":true,\"starting\":false,\"paused\":false,\"source\":\"2\",\"last_error\":null,\"thread_alive\":true}}"
+    },
+    {
+      "name": "cycle7-status-before",
+      "ok": true,
+      "duration_ms": 25,
+      "status_code": 200,
+      "body": "{\"status\":\"success\",\"live\":{\"running\":true,\"starting\":false,\"paused\":false,\"source\":\"2\",\"last_error\":null,\"thread_alive\":true}}"
+    },
+    {
+      "name": "cycle7-switch-0",
+      "ok": true,
+      "duration_ms": 998,
+      "status_code": 200,
+      "body": "{\"status\":\"success\",\"message\":\"Pipeline started (source=0)\",\"live\":{\"running\":true,\"starting\":false,\"paused\":false,\"source\":\"0\",\"last_error\":null,\"thread_alive\":true}}"
+    },
+    {
+      "name": "cycle7-status-after",
+      "ok": true,
+      "duration_ms": 21,
+      "status_code": 200,
+      "body": "{\"status\":\"success\",\"live\":{\"running\":true,\"starting\":false,\"paused\":false,\"source\":\"0\",\"last_error\":null,\"thread_alive\":true}}"
+    },
+    {
+      "name": "cycle8-status-before",
+      "ok": true,
+      "duration_ms": 2,
+      "status_code": 200,
+      "body": "{\"status\":\"success\",\"live\":{\"running\":true,\"starting\":false,\"paused\":false,\"source\":\"0\",\"last_error\":null,\"thread_alive\":true}}"
+    },
+    {
+      "name": "cycle8-switch-2",
+      "ok": true,
+      "duration_ms": 424,
+      "status_code": 200,
+      "body": "{\"status\":\"success\",\"message\":\"Pipeline started (source=2)\",\"live\":{\"running\":true,\"starting\":false,\"paused\":false,\"source\":\"2\",\"last_error\":null,\"thread_alive\":true}}"
+    },
+    {
+      "name": "cycle8-status-after",
+      "ok": true,
+      "duration_ms": 1,
+      "status_code": 200,
+      "body": "{\"status\":\"success\",\"live\":{\"running\":true,\"starting\":false,\"paused\":false,\"source\":\"2\",\"last_error\":null,\"thread_alive\":true}}"
+    },
+    {
+      "name": "cycle9-status-before",
+      "ok": true,
+      "duration_ms": 17,
+      "status_code": 200,
+      "body": "{\"status\":\"success\",\"live\":{\"running\":true,\"starting\":false,\"paused\":false,\"source\":\"2\",\"last_error\":null,\"thread_alive\":true}}"
+    },
+    {
+      "name": "cycle9-switch-0",
+      "ok": true,
+      "duration_ms": 1034,
+      "status_code": 200,
+      "body": "{\"status\":\"success\",\"message\":\"Pipeline started (source=0)\",\"live\":{\"running\":true,\"starting\":false,\"paused\":false,\"source\":\"0\",\"last_error\":null,\"thread_alive\":true}}"
+    },
+    {
+      "name": "cycle9-pause",
+      "ok": true,
+      "duration_ms": 1013,
+      "status_code": 200,
+      "body": "{\"status\":\"success\",\"message\":\"Pipeline paused\",\"live\":{\"running\":false,\"starting\":false,\"paused\":true,\"source\":\"0\",\"last_error\":null,\"thread_alive\":false}}"
+    },
+    {
+      "name": "cycle9-resume",
+      "ok": true,
+      "duration_ms": 150,
+      "status_code": 200,
+      "body": "{\"status\":\"success\",\"message\":\"Pipeline resumed\",\"live\":{\"running\":true,\"starting\":false,\"paused\":false,\"source\":\"0\",\"last_error\":null,\"thread_alive\":true}}"
+    },
+    {
+      "name": "cycle9-status-after",
+      "ok": true,
+      "duration_ms": 17,
+      "status_code": 200,
+      "body": "{\"status\":\"success\",\"live\":{\"running\":true,\"starting\":false,\"paused\":false,\"source\":\"0\",\"last_error\":null,\"thread_alive\":true}}"
+    },
+    {
+      "name": "cycle10-status-before",
+      "ok": true,
+      "duration_ms": 16,
+      "status_code": 200,
+      "body": "{\"status\":\"success\",\"live\":{\"running\":true,\"starting\":false,\"paused\":false,\"source\":\"0\",\"last_error\":null,\"thread_alive\":true}}"
+    },
+    {
+      "name": "cycle10-switch-2",
+      "ok": true,
+      "duration_ms": 389,
+      "status_code": 200,
+      "body": "{\"status\":\"success\",\"message\":\"Pipeline started (source=2)\",\"live\":{\"running\":true,\"starting\":false,\"paused\":false,\"source\":\"2\",\"last_error\":null,\"thread_alive\":true}}"
+    },
+    {
+      "name": "cycle10-status-after",
+      "ok": true,
+      "duration_ms": 1,
+      "status_code": 200,
+      "body": "{\"status\":\"success\",\"live\":{\"running\":true,\"starting\":false,\"paused\":false,\"source\":\"2\",\"last_error\":null,\"thread_alive\":true}}"
+    },
+    {
+      "name": "cycle11-status-before",
+      "ok": true,
+      "duration_ms": 1,
+      "status_code": 200,
+      "body": "{\"status\":\"success\",\"live\":{\"running\":true,\"starting\":false,\"paused\":false,\"source\":\"2\",\"last_error\":null,\"thread_alive\":true}}"
+    },
+    {
+      "name": "cycle11-switch-0",
+      "ok": true,
+      "duration_ms": 1013,
+      "status_code": 200,
+      "body": "{\"status\":\"success\",\"message\":\"Pipeline started (source=0)\",\"live\":{\"running\":true,\"starting\":false,\"paused\":false,\"source\":\"0\",\"last_error\":null,\"thread_alive\":true}}"
+    },
+    {
+      "name": "cycle11-status-after",
+      "ok": true,
+      "duration_ms": 21,
+      "status_code": 200,
+      "body": "{\"status\":\"success\",\"live\":{\"running\":true,\"starting\":false,\"paused\":false,\"source\":\"0\",\"last_error\":null,\"thread_alive\":true}}"
+    },
+    {
+      "name": "cycle12-status-before",
+      "ok": true,
+      "duration_ms": 1,
+      "status_code": 200,
+      "body": "{\"status\":\"success\",\"live\":{\"running\":true,\"starting\":false,\"paused\":false,\"source\":\"0\",\"last_error\":null,\"thread_alive\":true}}"
+    },
+    {
+      "name": "cycle12-switch-2",
+      "ok": true,
+      "duration_ms": 461,
+      "status_code": 200,
+      "body": "{\"status\":\"success\",\"message\":\"Pipeline started (source=2)\",\"live\":{\"running\":true,\"starting\":false,\"paused\":false,\"source\":\"2\",\"last_error\":null,\"thread_alive\":true}}"
+    },
+    {
+      "name": "cycle12-pause",
+      "ok": true,
+      "duration_ms": 1004,
+      "status_code": 200,
+      "body": "{\"status\":\"success\",\"message\":\"Pipeline paused\",\"live\":{\"running\":false,\"starting\":false,\"paused\":true,\"source\":\"2\",\"last_error\":null,\"thread_alive\":false}}"
+    },
+    {
+      "name": "cycle12-resume",
+      "ok": true,
+      "duration_ms": 340,
+      "status_code": 200,
+      "body": "{\"status\":\"success\",\"message\":\"Pipeline resumed\",\"live\":{\"running\":true,\"starting\":false,\"paused\":false,\"source\":\"2\",\"last_error\":null,\"thread_alive\":true}}"
+    },
+    {
+      "name": "cycle12-status-after",
+      "ok": true,
+      "duration_ms": 1,
+      "status_code": 200,
+      "body": "{\"status\":\"success\",\"live\":{\"running\":true,\"starting\":false,\"paused\":false,\"source\":\"2\",\"last_error\":null,\"thread_alive\":true}}"
+    },
+    {
+      "name": "cycle13-status-before",
+      "ok": true,
+      "duration_ms": 1,
+      "status_code": 200,
+      "body": "{\"status\":\"success\",\"live\":{\"running\":true,\"starting\":false,\"paused\":false,\"source\":\"2\",\"last_error\":null,\"thread_alive\":true}}"
+    },
+    {
+      "name": "cycle13-switch-0",
+      "ok": true,
+      "duration_ms": 999,
+      "status_code": 200,
+      "body": "{\"status\":\"success\",\"message\":\"Pipeline started (source=0)\",\"live\":{\"running\":true,\"starting\":false,\"paused\":false,\"source\":\"0\",\"last_error\":null,\"thread_alive\":true}}"
+    },
+    {
+      "name": "cycle13-status-after",
+      "ok": true,
+      "duration_ms": 14,
+      "status_code": 200,
+      "body": "{\"status\":\"success\",\"live\":{\"running\":true,\"starting\":false,\"paused\":false,\"source\":\"0\",\"last_error\":null,\"thread_alive\":true}}"
+    },
+    {
+      "name": "cycle14-status-before",
+      "ok": true,
+      "duration_ms": 1,
+      "status_code": 200,
+      "body": "{\"status\":\"success\",\"live\":{\"running\":true,\"starting\":false,\"paused\":false,\"source\":\"0\",\"last_error\":null,\"thread_alive\":true}}"
+    },
+    {
+      "name": "cycle14-switch-2",
+      "ok": true,
+      "duration_ms": 400,
+      "status_code": 200,
+      "body": "{\"status\":\"success\",\"message\":\"Pipeline started (source=2)\",\"live\":{\"running\":true,\"starting\":false,\"paused\":false,\"source\":\"2\",\"last_error\":null,\"thread_alive\":true}}"
+    },
+    {
+      "name": "cycle14-status-after",
+      "ok": true,
+      "duration_ms": 1,
+      "status_code": 200,
+      "body": "{\"status\":\"success\",\"live\":{\"running\":true,\"starting\":false,\"paused\":false,\"source\":\"2\",\"last_error\":null,\"thread_alive\":true}}"
+    },
+    {
+      "name": "cycle15-status-before",
+      "ok": true,
+      "duration_ms": 1,
+      "status_code": 200,
+      "body": "{\"status\":\"success\",\"live\":{\"running\":true,\"starting\":false,\"paused\":false,\"source\":\"2\",\"last_error\":null,\"thread_alive\":true}}"
+    },
+    {
+      "name": "cycle15-switch-0",
+      "ok": true,
+      "duration_ms": 1017,
+      "status_code": 200,
+      "body": "{\"status\":\"success\",\"message\":\"Pipeline started (source=0)\",\"live\":{\"running\":true,\"starting\":false,\"paused\":false,\"source\":\"0\",\"last_error\":null,\"thread_alive\":true}}"
+    },
+    {
+      "name": "cycle15-pause",
+      "ok": true,
+      "duration_ms": 1004,
+      "status_code": 200,
+      "body": "{\"status\":\"success\",\"message\":\"Pipeline paused\",\"live\":{\"running\":false,\"starting\":false,\"paused\":true,\"source\":\"0\",\"last_error\":null,\"thread_alive\":false}}"
+    },
+    {
+      "name": "cycle15-resume",
+      "ok": true,
+      "duration_ms": 130,
+      "status_code": 200,
+      "body": "{\"status\":\"success\",\"message\":\"Pipeline resumed\",\"live\":{\"running\":true,\"starting\":false,\"paused\":false,\"source\":\"0\",\"last_error\":null,\"thread_alive\":true}}"
+    },
+    {
+      "name": "cycle15-status-after",
+      "ok": true,
+      "duration_ms": 22,
+      "status_code": 200,
+      "body": "{\"status\":\"success\",\"live\":{\"running\":true,\"starting\":false,\"paused\":false,\"source\":\"0\",\"last_error\":null,\"thread_alive\":true}}"
+    },
+    {
+      "name": "cycle16-status-before",
+      "ok": true,
+      "duration_ms": 2,
+      "status_code": 200,
+      "body": "{\"status\":\"success\",\"live\":{\"running\":true,\"starting\":false,\"paused\":false,\"source\":\"0\",\"last_error\":null,\"thread_alive\":true}}"
+    },
+    {
+      "name": "cycle16-switch-2",
+      "ok": true,
+      "duration_ms": 382,
+      "status_code": 200,
+      "body": "{\"status\":\"success\",\"message\":\"Pipeline started (source=2)\",\"live\":{\"running\":true,\"starting\":false,\"paused\":false,\"source\":\"2\",\"last_error\":null,\"thread_alive\":true}}"
+    },
+    {
+      "name": "cycle16-status-after",
+      "ok": true,
+      "duration_ms": 26,
+      "status_code": 200,
+      "body": "{\"status\":\"success\",\"live\":{\"running\":true,\"starting\":false,\"paused\":false,\"source\":\"2\",\"last_error\":null,\"thread_alive\":true}}"
+    },
+    {
+      "name": "cycle17-status-before",
+      "ok": true,
+      "duration_ms": 2,
+      "status_code": 200,
+      "body": "{\"status\":\"success\",\"live\":{\"running\":true,\"starting\":false,\"paused\":false,\"source\":\"2\",\"last_error\":null,\"thread_alive\":true}}"
+    },
+    {
+      "name": "cycle17-switch-0",
+      "ok": true,
+      "duration_ms": 977,
+      "status_code": 200,
+      "body": "{\"status\":\"success\",\"message\":\"Pipeline started (source=0)\",\"live\":{\"running\":true,\"starting\":false,\"paused\":false,\"source\":\"0\",\"last_error\":null,\"thread_alive\":true}}"
+    },
+    {
+      "name": "cycle17-status-after",
+      "ok": true,
+      "duration_ms": 14,
+      "status_code": 200,
+      "body": "{\"status\":\"success\",\"live\":{\"running\":true,\"starting\":false,\"paused\":false,\"source\":\"0\",\"last_error\":null,\"thread_alive\":true}}"
+    },
+    {
+      "name": "cycle18-status-before",
+      "ok": true,
+      "duration_ms": 15,
+      "status_code": 200,
+      "body": "{\"status\":\"success\",\"live\":{\"running\":true,\"starting\":false,\"paused\":false,\"source\":\"0\",\"last_error\":null,\"thread_alive\":true}}"
+    },
+    {
+      "name": "cycle18-switch-2",
+      "ok": true,
+      "duration_ms": 353,
+      "status_code": 200,
+      "body": "{\"status\":\"success\",\"message\":\"Pipeline started (source=2)\",\"live\":{\"running\":true,\"starting\":false,\"paused\":false,\"source\":\"2\",\"last_error\":null,\"thread_alive\":true}}"
+    },
+    {
+      "name": "cycle18-pause",
+      "ok": true,
+      "duration_ms": 1003,
+      "status_code": 200,
+      "body": "{\"status\":\"success\",\"message\":\"Pipeline paused\",\"live\":{\"running\":false,\"starting\":false,\"paused\":true,\"source\":\"2\",\"last_error\":null,\"thread_alive\":false}}"
+    },
+    {
+      "name": "cycle18-resume",
+      "ok": true,
+      "duration_ms": 357,
+      "status_code": 200,
+      "body": "{\"status\":\"success\",\"message\":\"Pipeline resumed\",\"live\":{\"running\":true,\"starting\":false,\"paused\":false,\"source\":\"2\",\"last_error\":null,\"thread_alive\":true}}"
+    },
+    {
+      "name": "cycle18-status-after",
+      "ok": true,
+      "duration_ms": 1,
+      "status_code": 200,
+      "body": "{\"status\":\"success\",\"live\":{\"running\":true,\"starting\":false,\"paused\":false,\"source\":\"2\",\"last_error\":null,\"thread_alive\":true}}"
+    },
+    {
+      "name": "cycle19-status-before",
+      "ok": true,
+      "duration_ms": 1,
+      "status_code": 200,
+      "body": "{\"status\":\"success\",\"live\":{\"running\":true,\"starting\":false,\"paused\":false,\"source\":\"2\",\"last_error\":null,\"thread_alive\":true}}"
+    },
+    {
+      "name": "cycle19-switch-0",
+      "ok": true,
+      "duration_ms": 1001,
+      "status_code": 200,
+      "body": "{\"status\":\"success\",\"message\":\"Pipeline started (source=0)\",\"live\":{\"running\":true,\"starting\":false,\"paused\":false,\"source\":\"0\",\"last_error\":null,\"thread_alive\":true}}"
+    },
+    {
+      "name": "cycle19-status-after",
+      "ok": true,
+      "duration_ms": 18,
+      "status_code": 200,
+      "body": "{\"status\":\"success\",\"live\":{\"running\":true,\"starting\":false,\"paused\":false,\"source\":\"0\",\"last_error\":null,\"thread_alive\":true}}"
+    },
+    {
+      "name": "cycle20-status-before",
+      "ok": true,
+      "duration_ms": 2,
+      "status_code": 200,
+      "body": "{\"status\":\"success\",\"live\":{\"running\":true,\"starting\":false,\"paused\":false,\"source\":\"0\",\"last_error\":null,\"thread_alive\":true}}"
+    },
+    {
+      "name": "cycle20-switch-2",
+      "ok": true,
+      "duration_ms": 401,
+      "status_code": 200,
+      "body": "{\"status\":\"success\",\"message\":\"Pipeline started (source=2)\",\"live\":{\"running\":true,\"starting\":false,\"paused\":false,\"source\":\"2\",\"last_error\":null,\"thread_alive\":true}}"
+    },
+    {
+      "name": "cycle20-status-after",
+      "ok": true,
+      "duration_ms": 2,
+      "status_code": 200,
+      "body": "{\"status\":\"success\",\"live\":{\"running\":true,\"starting\":false,\"paused\":false,\"source\":\"2\",\"last_error\":null,\"thread_alive\":true}}"
+    },
+    {
+      "name": "final-status",
+      "ok": true,
+      "duration_ms": 1,
+      "status_code": 200,
+      "body": "{\"status\":\"success\",\"live\":{\"running\":true,\"starting\":false,\"paused\":false,\"source\":\"2\",\"last_error\":null,\"thread_alive\":true}}"
+    }
+  ]
+}

scripts/live_stress_test.py

@@ -0,0 +1,265 @@
+"""Stress-test live control endpoints for the smoking detection server.
+
+Usage example:
+    python scripts/live_stress_test.py --cycles 20 --request-timeout 8 --sleep-ms 120
+"""
+
+from __future__ import annotations
+
+import argparse
+import json
+import time
+from dataclasses import asdict, dataclass
+from datetime import datetime
+from pathlib import Path
+from typing import Any
+from urllib.error import HTTPError, URLError
+from urllib.parse import urlencode
+from urllib.request import Request, urlopen
+
+
+@dataclass
+class StepResult:
+    name: str
+    ok: bool
+    duration_ms: int
+    status_code: int | None
+    body: str
+
+
+def _call(
+    base_url: str,
+    name: str,
+    method: str,
+    path: str,
+    timeout_seconds: float,
+    form: dict[str, Any] | None = None,
+) -> StepResult:
+    url = f"{base_url}{path}"
+    data = None
+    headers: dict[str, str] = {}
+
+    if form is not None:
+        data = urlencode(form).encode("utf-8")
+        headers["Content-Type"] = "application/x-www-form-urlencoded"
+
+    request = Request(url=url, data=data, method=method, headers=headers)
+
+    start = time.perf_counter()
+    try:
+        with urlopen(request, timeout=timeout_seconds) as response:
+            body = response.read().decode("utf-8", errors="replace")
+            status = getattr(response, "status", None)
+        duration_ms = int((time.perf_counter() - start) * 1000)
+        return StepResult(
+            name=name, ok=True, duration_ms=duration_ms, status_code=status, body=body
+        )
+    except HTTPError as err:
+        duration_ms = int((time.perf_counter() - start) * 1000)
+        try:
+            body = err.read().decode("utf-8", errors="replace")
+        except Exception:
+            body = str(err)
+        return StepResult(
+            name=name,
+            ok=False,
+            duration_ms=duration_ms,
+            status_code=err.code,
+            body=body,
+        )
+    except URLError as err:
+        duration_ms = int((time.perf_counter() - start) * 1000)
+        return StepResult(
+            name=name,
+            ok=False,
+            duration_ms=duration_ms,
+            status_code=None,
+            body=str(err.reason),
+        )
+    except Exception as err:
+        duration_ms = int((time.perf_counter() - start) * 1000)
+        return StepResult(
+            name=name,
+            ok=False,
+            duration_ms=duration_ms,
+            status_code=None,
+            body=str(err),
+        )
+
+
+def _summarize(results: list[StepResult]) -> dict[str, Any]:
+    ok_results = [r for r in results if r.ok]
+    err_results = [r for r in results if not r.ok]
+    durations = [r.duration_ms for r in ok_results]
+
+    return {
+        "total": len(results),
+        "ok": len(ok_results),
+        "err": len(err_results),
+        "avg_ms": round(sum(durations) / len(durations), 1) if durations else 0,
+        "max_ms": max(durations) if durations else 0,
+        "over_500ms": sum(1 for d in durations if d > 500),
+        "over_1000ms": sum(1 for d in durations if d > 1000),
+    }
+
+
+def run_stress(args: argparse.Namespace) -> tuple[list[StepResult], dict[str, Any]]:
+    results: list[StepResult] = []
+
+    def add(result: StepResult) -> None:
+        results.append(result)
+        if args.verbose or not result.ok:
+            status = "OK" if result.ok else "ERR"
+            print(
+                f"{status}|{result.name}|{result.duration_ms}ms|"
+                f"status={result.status_code}|{result.body[:180]}"
+            )
+
+    add(
+        _call(
+            args.base_url, "warmup-status", "GET", "/live/status", args.request_timeout
+        )
+    )
+    add(
+        _call(
+            args.base_url,
+            "warmup-start",
+            "POST",
+            "/live/start",
+            args.request_timeout,
+            {"source": args.source_b},
+        )
+    )
+
+    for cycle in range(1, args.cycles + 1):
+        add(
+            _call(
+                args.base_url,
+                f"cycle{cycle}-status-before",
+                "GET",
+                "/live/status",
+                args.request_timeout,
+            )
+        )
+
+        src = args.source_b if cycle % 2 == 0 else args.source_a
+        add(
+            _call(
+                args.base_url,
+                f"cycle{cycle}-switch-{src}",
+                "POST",
+                "/live/switch_source",
+                args.request_timeout,
+                {"source": src},
+            )
+        )
+
+        if args.pause_every > 0 and cycle % args.pause_every == 0:
+            add(
+                _call(
+                    args.base_url,
+                    f"cycle{cycle}-pause",
+                    "POST",
+                    "/live/pause",
+                    args.request_timeout,
+                )
+            )
+            time.sleep(args.pause_sleep_ms / 1000.0)
+            add(
+                _call(
+                    args.base_url,
+                    f"cycle{cycle}-resume",
+                    "POST",
+                    "/live/resume",
+                    args.request_timeout,
+                )
+            )
+
+        add(
+            _call(
+                args.base_url,
+                f"cycle{cycle}-status-after",
+                "GET",
+                "/live/status",
+                args.request_timeout,
+            )
+        )
+
+        time.sleep(args.sleep_ms / 1000.0)
+
+    add(
+        _call(
+            args.base_url, "final-status", "GET", "/live/status", args.request_timeout
+        )
+    )
+
+    summary = _summarize(results)
+    return results, summary
+
+
+def parse_args() -> argparse.Namespace:
+    parser = argparse.ArgumentParser(description="Stress test /live/* endpoints")
+    parser.add_argument("--base-url", default="http://127.0.0.1:8000")
+    parser.add_argument("--cycles", type=int, default=20)
+    parser.add_argument("--request-timeout", type=float, default=8.0)
+    parser.add_argument("--sleep-ms", type=int, default=120)
+    parser.add_argument("--pause-every", type=int, default=3)
+    parser.add_argument("--pause-sleep-ms", type=int, default=120)
+    parser.add_argument("--source-a", default="0")
+    parser.add_argument("--source-b", default="2")
+    parser.add_argument("--out", default="reports/live_stress_latest.json")
+    parser.add_argument("--verbose", action="store_true")
+    return parser.parse_args()
+
+
+def main() -> int:
+    args = parse_args()
+    started_at = datetime.now().isoformat(timespec="seconds")
+
+    results, summary = run_stress(args)
+
+    report = {
+        "started_at": started_at,
+        "base_url": args.base_url,
+        "config": {
+            "cycles": args.cycles,
+            "request_timeout": args.request_timeout,
+            "sleep_ms": args.sleep_ms,
+            "pause_every": args.pause_every,
+            "pause_sleep_ms": args.pause_sleep_ms,
+            "source_a": args.source_a,
+            "source_b": args.source_b,
+        },
+        "summary": summary,
+        "errors": [asdict(r) for r in results if not r.ok],
+        "tail": [asdict(r) for r in results[-10:]],
+        "all_results": [asdict(r) for r in results],
+    }
+
+    out_path = Path(args.out)
+    out_path.parent.mkdir(parents=True, exist_ok=True)
+    out_path.write_text(json.dumps(report, indent=2), encoding="utf-8")
+
+    print(
+        "SUMMARY|"
+        f"total={summary['total']}|ok={summary['ok']}|err={summary['err']}|"
+        f"avg_ms={summary['avg_ms']}|max_ms={summary['max_ms']}|"
+        f"over_500ms={summary['over_500ms']}|over_1000ms={summary['over_1000ms']}"
+    )
+    print(f"REPORT|{out_path.as_posix()}")
+
+    if summary["err"] > 0:
+        print("ERRORS_BEGIN")
+        for err in report["errors"][:8]:
+            print(
+                f"ERR|{err['name']}|{err['duration_ms']}ms|"
+                f"status={err['status_code']}|{err['body'][:180]}"
+            )
+        print("ERRORS_END")
+        return 1
+
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

server.py

@@ -15,122 +15,27 @@ Usage:
 import argparse
 import logging
 import os
-import socket
-import threading
-import time
 from contextlib import asynccontextmanager
 from pathlib import Path
 
-import cv2
+from fastapi import FastAPI
-from fastapi import FastAPI, Form, Response
-from fastapi.responses import StreamingResponse
 from fastapi.staticfiles import StaticFiles
 
-from db import SQLiteRepository
+import server.runtime as rt
-from db.repository import Repository
 from pipeline.config import PipelineConfig
 from pipeline.core import SmokingDetectionPipeline
 from pipeline.models import YoloCigaretteDetector, YoloClassifier, YoloPersonDetector
 from pipeline.protocols import Sink
-from pipeline.sinks import FileAlertStore, MetricsSink, SmartEventSink
+from pipeline.sinks import FileAlertStore, MetricsSink
+from pipeline.sinks.database import DatabaseSmartEventSink
+from pipeline.sinks.video import VideoFileStore
+from pipeline.sources.file import FileStreamSource
 from pipeline.sources.stream import ThreadedStreamSource
 from pipeline.staged_core import StageBasedPipeline
-from pipeline.types import Frame
+from server.routes import router as api_router
 
 logger = logging.getLogger(__name__)
 
-# ---------------------------------------------------------------------------
-# Global state (set during lifespan)
-# ---------------------------------------------------------------------------
-db: Repository = SQLiteRepository()  # default: data/smoke.db
-pipeline = None  # set at startup
-source: ThreadedStreamSource | None = None  # set at startup
-live_frame: Frame | None = None  # latest frame for MJPEG / snapshot
-
-
-# ---------------------------------------------------------------------------
-# Sink that exposes frames to the web layer
-# ---------------------------------------------------------------------------
-class LiveFrameSink(Sink):
-    """Captures the latest frame so API endpoints can stream it."""
-
-    def handle(self, frame: Frame):
-        global live_frame
-        live_frame = frame
-
-    def close(self):
-        pass
-
-
-class DatabaseSmartEventSink(SmartEventSink):
-    """Records event clips and stores clip path in DB as alert evidence."""
-
-    def __init__(
-        self,
-        db_repo: Repository,
-        fps: float,
-        user_id: int = 1,
-        camera_source: str = "Pipeline",
-        output_dir: str = "static/events",
-    ):
-        super().__init__(output_dir=output_dir, fps=fps)
-        self.db_repo = db_repo
-        self.user_id = user_id
-        self.camera_source = camera_source
-
-    def _trigger_event(self, trigger_frame: Frame):
-        was_recording = self.recording
-        super()._trigger_event(trigger_frame)
-
-        # Save one DB row per newly started event clip
-        if was_recording or not self.recording or self.current_event_id is None:
-            return
-
-        ticket = trigger_frame.ticket
-        if not ticket:
-            return
-
-        filename = f"event_{self.current_event_id}.mp4"
-        web_path = f"/static/events/{filename}"
-        try:
-            self.db_repo.save_detection(
-                user_id=self.user_id,
-                image_path=web_path,
-                confidence=ticket.confidence,
-                camera_source=self.camera_source,
-            )
-            logger.info("Saved event alert to DB: %s", web_path)
-        except Exception as e:
-            logger.error("Failed to save event alert to DB: %s", e)
-
-
-def _get_lan_address() -> tuple[str, str]:
-    """Return (ip, hostname) for the primary LAN interface."""
-    ip = "127.0.0.1"
-    try:
-        # Open a UDP socket to an external IP (doesn't actually send anything)
-        # to figure out which local NIC the OS would route through.
-        s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
-        s.connect(("8.8.8.8", 80))
-        ip = s.getsockname()[0]
-        s.close()
-    except Exception:
-        pass
-    hostname = socket.getfqdn()
-    return ip, hostname
-
-
-# ---------------------------------------------------------------------------
-# Pipeline runner (background thread)
-# ---------------------------------------------------------------------------
-def _run_pipeline():
-    """Blocking call: runs in a daemon thread."""
-    assert pipeline is not None and source is not None
-    try:
-        pipeline.process_stream(source)
-    except Exception:  # noqa: BLE001
-        logger.exception("Pipeline crashed")
-
 
 # ---------------------------------------------------------------------------
 # CLI args  (parsed before uvicorn when run as __main__)
@@ -174,8 +79,6 @@ _args = argparse.Namespace(
 # ---------------------------------------------------------------------------
 @asynccontextmanager
 async def app_lifespan(_app: FastAPI):
-    global pipeline, source  # noqa: PLW0603
-
     logging.basicConfig(
         level=logging.DEBUG if _args.debug else logging.INFO,
         format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
@@ -185,7 +88,7 @@ async def app_lifespan(_app: FastAPI):
 
     # 1. Database
     logger.info("Initializing database...")
-    db.initialize()
+    rt.db.initialize()
 
     # 2. Models
     logger.info("Loading models...")
@@ -194,32 +97,33 @@ async def app_lifespan(_app: FastAPI):
     verifier = YoloClassifier(_args.verify_model)
     logger.info("Models loaded.")
 
-    # 3. Source (webcam / RTSP)
+    # 3. Configure live pipeline manager
-    cam = int(_args.source) if _args.source.isdigit() else _args.source
+    def source_factory(source_spec: str) -> ThreadedStreamSource:
-    source = ThreadedStreamSource(cam)
+        cam = int(source_spec) if source_spec.isdigit() else source_spec
-    logger.info("Camera source ready: %s (%.1f FPS)", _args.source, source.fps)
+        return ThreadedStreamSource(cam)
 
-    # 4. Sinks
+    def sink_factory(src: ThreadedStreamSource) -> list[Sink]:
-    sinks: list[Sink] = [
+        return [
-        LiveFrameSink(),
+            rt.LiveFrameSink(),
-        DatabaseSmartEventSink(db, fps=source.fps, output_dir="static/events"),
+            DatabaseSmartEventSink(rt.db, fps=src.fps, output_dir="static/events"),
             MetricsSink(log_interval=200),
             FileAlertStore("alerts.jsonl"),
         ]
 
-    # 5. Pipeline config
+    def pipeline_factory(src: ThreadedStreamSource, sinks: list[Sink]):
         config = PipelineConfig(
             idle_interval=0.5,
             active_interval=0.0,
-        cooldown_frames=int(source.fps * 2),
+            cooldown_frames=int(src.fps * 2),
         )
 
-    # 6. Build pipeline
         if _args.staged or _args.vlm:
             vlm_key = _args.vlm_key or os.environ.get(
-            "OPENAI_API_KEY" if _args.vlm_provider == "openai" else "ANTHROPIC_API_KEY"
+                "OPENAI_API_KEY"
+                if _args.vlm_provider == "openai"
+                else "ANTHROPIC_API_KEY"
             )
-        pipeline = StageBasedPipeline(
+            return StageBasedPipeline(
                 person_model=person_model,
                 cigarette_model=cig_model,
                 verifier_model=verifier,
@@ -230,8 +134,8 @@ async def app_lifespan(_app: FastAPI):
                 sinks=sinks,
                 config=config,
             )
-    else:
+
-        pipeline = SmokingDetectionPipeline(
+        return SmokingDetectionPipeline(
             person_model=person_model,
             cigarette_model=cig_model,
             verifier_model=verifier,
@@ -239,13 +143,78 @@ async def app_lifespan(_app: FastAPI):
             config=config,
         )
 
-    # 7. Launch pipeline in background thread
+    rt.manager.configure(
-    t = threading.Thread(target=_run_pipeline, daemon=True, name="pipeline")
+        source_factory=source_factory,
-    t.start()
+        sink_factory=sink_factory,
-    logger.info("Pipeline running in background.")
+        pipeline_factory=pipeline_factory,
+        initial_source=_args.source,
+    )
+
+    def clip_source_factory(path: str) -> FileStreamSource:
+        return FileStreamSource(path)
+
+    def clip_sink_factory(src: FileStreamSource, job_id: str) -> list[Sink]:
+        job_dir = Path("static/clip_jobs") / job_id
+        events_dir = job_dir / "events"
+        job_dir.mkdir(parents=True, exist_ok=True)
+        events_dir.mkdir(parents=True, exist_ok=True)
+        return [
+            VideoFileStore(str(job_dir / "processed.mp4"), fps=src.fps),
+            DatabaseSmartEventSink(
+                rt.db,
+                fps=src.fps,
+                camera_source=f"ClipJob:{job_id}",
+                output_dir=str(events_dir),
+            ),
+            MetricsSink(log_interval=200),
+            FileAlertStore(str(job_dir / "alerts.jsonl")),
+        ]
+
+    def clip_pipeline_factory(src: FileStreamSource, sinks: list[Sink]):
+        config = PipelineConfig(
+            idle_interval=0.5,
+            active_interval=0.0,
+            cooldown_frames=max(1, int(src.fps * 2)),
+        )
+
+        if _args.staged or _args.vlm:
+            vlm_key = _args.vlm_key or os.environ.get(
+                "OPENAI_API_KEY"
+                if _args.vlm_provider == "openai"
+                else "ANTHROPIC_API_KEY"
+            )
+            return StageBasedPipeline(
+                person_model=person_model,
+                cigarette_model=cig_model,
+                verifier_model=verifier,
+                vlm_api_key=vlm_key if _args.vlm else None,
+                vlm_provider=_args.vlm_provider,
+                vlm_model=_args.vlm_model,
+                use_vlm=_args.vlm,
+                sinks=sinks,
+                config=config,
+            )
+
+        return SmokingDetectionPipeline(
+            person_model=person_model,
+            cigarette_model=cig_model,
+            verifier_model=verifier,
+            sinks=sinks,
+            config=config,
+        )
+
+    rt.clip_jobs.configure(
+        source_factory=clip_source_factory,
+        sink_factory=clip_sink_factory,
+        pipeline_factory=clip_pipeline_factory,
+    )
+    ok, msg = rt.manager.start_async()
+    if not ok:
+        raise RuntimeError(msg)
+    logger.info("%s (source=%s)", msg, _args.source)
 
     # Show real addresses
-    ip, hostname = _get_lan_address()
+    ip, hostname = rt.get_lan_address()
     port = _args.port if hasattr(_args, "port") else 8000
     logger.info("=" * 50)
     logger.info("  Server ready")
@@ -259,10 +228,9 @@ async def app_lifespan(_app: FastAPI):
 
     # Shutdown
     logger.info("Shutting down...")
-    source.stop()
+    rt.manager.stop()
-    for s in sinks:
+    rt.clip_jobs.shutdown()
-        s.close()
+    rt.db.close()
-    db.close()
     logger.info("Goodbye.")
 
 
@@ -270,111 +238,15 @@ async def app_lifespan(_app: FastAPI):
 # FastAPI app
 # ---------------------------------------------------------------------------
 app = FastAPI(title="Smoking Detection", lifespan=app_lifespan)  # type: ignore
+app.include_router(api_router)
 
 # Serve static assets (alert images, CSS/JS, etc.)
 Path("static/alerts").mkdir(parents=True, exist_ok=True)
 Path("static/events").mkdir(parents=True, exist_ok=True)
+Path("static/clip_jobs").mkdir(parents=True, exist_ok=True)
 app.mount("/static", StaticFiles(directory="static"), name="static")
 
 
-# -- Auth ------------------------------------------------------------------
-
-
-@app.get("/ping")
-def ping():
-    """Health check for the mobile app to verify connectivity."""
-    return {"status": "ok", "name": "SmokingDetection"}
-
-
-@app.post("/register")
-def register(username: str = Form(...), password: str = Form(...)):
-    try:
-        if db.get_user_by_username(username):
-            return {"status": "fail", "message": "Username already taken"}
-        new_id = db.create_user(username, password)
-        return {"status": "success", "user_id": new_id, "message": "Account created"}
-    except Exception as e:
-        return {"status": "error", "message": str(e)}
-
-
-@app.post("/login")
-def login(username: str = Form(...), password: str = Form(...)):
-    try:
-        user = db.get_user_by_credentials(username, password)
-        if user:
-            return {"status": "success", "user_id": user.id, "username": user.username}
-        return {"status": "fail", "message": "Invalid username or password"}
-    except Exception as e:
-        return {"status": "error", "message": str(e)}
-
-
-# -- Detection history -----------------------------------------------------
-
-
-@app.get("/history")
-def get_history(user_id: int | None = None):
-    try:
-        records = db.get_detections(user_id)
-        return [
-            {
-                "id": r.id,
-                "user_id": r.user_id,
-                "image_path": r.image_path,
-                "confidence": r.confidence,
-                "timestamp": r.timestamp,
-                "camera_source": r.camera_source,
-            }
-            for r in records
-        ]
-    except Exception:
-        return []
-
-
-@app.post("/clear_history")
-def clear_history():
-    try:
-        deleted = db.clear_detections()
-        logger.info("Cleared %d records", deleted)
-        return {"status": "success", "message": "History cleared"}
-    except Exception as e:
-        return {"status": "error", "message": str(e)}
-
-
-# -- Video / Snapshot ------------------------------------------------------
-
-
-@app.get("/snapshot")
-def snapshot():
-    """Return the latest frame as a JPEG."""
-    if live_frame is not None:
-        _, jpg = cv2.imencode(".jpg", live_frame.data)
-        return Response(content=jpg.tobytes(), media_type="image/jpeg")
-    return Response(status_code=404)
-
-
-@app.get("/video_feed")
-def video_feed():
-    """MJPEG stream - drop into an <img> tag."""
-
-    def generate():
-        while True:
-            if live_frame is not None:
-                from pipeline.ops.drawing import draw_detections
-
-                img = live_frame.data.copy()
-                draw_detections(img, live_frame.detections)
-                _, jpg = cv2.imencode(".jpg", img)
-                yield (
-                    b"--frame\r\n"
-                    b"Content-Type: image/jpeg\r\n\r\n" + jpg.tobytes() + b"\r\n"
-                )
-            time.sleep(0.033)  # ~30 FPS cap
-
-    return StreamingResponse(
-        generate(), media_type="multipart/x-mixed-replace;boundary=frame"
-    )
-
-
 # ---------------------------------------------------------------------------
 # Entry-point
 # ---------------------------------------------------------------------------

server/__main__.py

@@ -0,0 +1,9 @@
+"""Package entrypoint so `python -m server` / `uv run -m server` works."""
+
+import runpy
+from pathlib import Path
+
+# Execute top-level server.py as __main__
+runpy.run_path(
+    str(Path(__file__).resolve().parent.parent / "server.py"), run_name="__main__"
+)

server/routes.py

@@ -0,0 +1,237 @@
+"""API route definitions for the smoking detection server."""
+
+import shutil
+import time
+from pathlib import Path
+
+import cv2
+from fastapi import APIRouter, File, Form, Response, UploadFile
+from fastapi.responses import StreamingResponse
+
+import server.runtime as rt
+
+router = APIRouter()
+
+
+@router.get("/ping")
+def ping():
+    """Health check for the mobile app to verify connectivity."""
+    return {"status": "ok", "name": "SmokingDetection"}
+
+
+@router.post("/register")
+def register(username: str = Form(...), password: str = Form(...)):
+    try:
+        if rt.db.get_user_by_username(username):
+            return {"status": "fail", "message": "Username already taken"}
+        new_id = rt.db.create_user(username, password)
+        return {"status": "success", "user_id": new_id, "message": "Account created"}
+    except Exception as e:
+        return {"status": "error", "message": str(e)}
+
+
+@router.post("/login")
+def login(username: str = Form(...), password: str = Form(...)):
+    try:
+        user = rt.db.get_user_by_credentials(username, password)
+        if user:
+            return {"status": "success", "user_id": user.id, "username": user.username}
+        return {"status": "fail", "message": "Invalid username or password"}
+    except Exception as e:
+        return {"status": "error", "message": str(e)}
+
+
+@router.get("/history")
+def get_history(user_id: int | None = None):
+    try:
+        records = rt.db.get_detections(user_id)
+        return [
+            {
+                "id": r.id,
+                "user_id": r.user_id,
+                "image_path": r.image_path,
+                "confidence": r.confidence,
+                "timestamp": r.timestamp,
+                "camera_source": r.camera_source,
+            }
+            for r in records
+        ]
+    except Exception:
+        return []
+
+
+@router.post("/clear_history")
+def clear_history(user_id: int | None = Form(default=None)):
+    try:
+        deleted = rt.db.clear_detections(user_id)
+        return {
+            "status": "success",
+            "message": "History cleared",
+            "deleted": deleted,
+            "user_id": user_id,
+        }
+    except Exception as e:
+        return {"status": "error", "message": str(e)}
+
+
+@router.get("/live/status")
+def live_status():
+    """Get current live pipeline status."""
+    return {"status": "success", "live": rt.manager.status()}
+
+
+@router.post("/live/start")
+def live_start(source: str | None = Form(default=None)):
+    """Start live pipeline (optionally with source override)."""
+    ok, message = rt.manager.start(source)
+    return {
+        "status": "success" if ok else "fail",
+        "message": message,
+        "live": rt.manager.status(),
+    }
+
+
+@router.post("/live/stop")
+def live_stop():
+    """Stop live pipeline."""
+    ok, message = rt.manager.stop()
+    return {
+        "status": "success" if ok else "fail",
+        "message": message,
+        "live": rt.manager.status(),
+    }
+
+
+@router.post("/live/pause")
+def live_pause():
+    """Pause live pipeline (stops source, resumable)."""
+    ok, message = rt.manager.pause()
+    return {
+        "status": "success" if ok else "fail",
+        "message": message,
+        "live": rt.manager.status(),
+    }
+
+
+@router.post("/live/resume")
+def live_resume():
+    """Resume paused live pipeline."""
+    ok, message = rt.manager.resume()
+    return {
+        "status": "success" if ok else "fail",
+        "message": message,
+        "live": rt.manager.status(),
+    }
+
+
+@router.post("/live/switch_source")
+def live_switch_source(source: str = Form(...)):
+    """Switch active live source (webcam index or RTSP URL)."""
+    ok, message = rt.manager.switch_source(source)
+    return {
+        "status": "success" if ok else "fail",
+        "message": message,
+        "live": rt.manager.status(),
+    }
+
+
+@router.post("/clips/submit")
+async def clips_submit(
+    file: UploadFile | None = File(default=None),
+    path: str | None = Form(default=None),
+):
+    """Queue a clip processing job from uploaded file or existing local path."""
+    if file is None and not path:
+        return {
+            "status": "fail",
+            "message": "Provide either an uploaded file or a local file path",
+        }
+
+    clip_input_path = ""
+    source_name = None
+
+    try:
+        if file is not None:
+            uploads_dir = Path("media/clip_jobs/uploads")
+            uploads_dir.mkdir(parents=True, exist_ok=True)
+
+            source_name = file.filename or "upload.mp4"
+            suffix = Path(source_name).suffix or ".mp4"
+            safe_name = f"clip_{int(time.time() * 1000)}{suffix}"
+            saved_path = uploads_dir / safe_name
+
+            with saved_path.open("wb") as out:
+                shutil.copyfileobj(file.file, out)
+
+            clip_input_path = str(saved_path)
+        else:
+            local_path = Path(path or "")
+            if not local_path.exists() or not local_path.is_file():
+                return {
+                    "status": "fail",
+                    "message": f"File not found: {local_path}",
+                }
+            clip_input_path = str(local_path)
+            source_name = local_path.name
+    finally:
+        if file is not None:
+            await file.close()
+
+    job = rt.clip_jobs.submit(clip_input_path, source_name)
+    return {
+        "status": "success",
+        "message": "Clip job queued",
+        "job": job,
+        "queue": rt.clip_jobs.status(),
+    }
+
+
+@router.get("/clips")
+def clips_list(limit: int = 20):
+    """List recent clip jobs."""
+    return {
+        "status": "success",
+        "queue": rt.clip_jobs.status(),
+        "jobs": rt.clip_jobs.list(limit=max(1, min(limit, 200))),
+    }
+
+
+@router.get("/clips/{job_id}")
+def clips_get(job_id: str):
+    """Get one clip job state."""
+    job = rt.clip_jobs.get(job_id)
+    if not job:
+        return {"status": "fail", "message": "Job not found", "job_id": job_id}
+    return {"status": "success", "job": job}
+
+
+@router.get("/snapshot")
+def snapshot():
+    """Return the latest frame as a JPEG."""
+    if rt.live_frame is not None:
+        _, jpg = cv2.imencode(".jpg", rt.live_frame.data)
+        return Response(content=jpg.tobytes(), media_type="image/jpeg")
+    return Response(status_code=404)
+
+
+@router.get("/video_feed")
+def video_feed():
+    """MJPEG stream - drop into an <img> tag."""
+
+    def generate():
+        while True:
+            if rt.live_frame is not None:
+                from pipeline.ops.drawing import draw_detections
+
+                img = rt.live_frame.data.copy()
+                draw_detections(img, rt.live_frame.detections)
+                _, jpg = cv2.imencode(".jpg", img)
+                yield (
+                    b"--frame\r\n"
+                    b"Content-Type: image/jpeg\r\n\r\n" + jpg.tobytes() + b"\r\n"
+                )
+            time.sleep(0.033)
+
+    return StreamingResponse(
+        generate(), media_type="multipart/x-mixed-replace;boundary=frame"
+    )

server/runtime.py

@@ -0,0 +1,538 @@
+"""Runtime state and utility helpers for the FastAPI server."""
+
+import logging
+import socket
+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
+
+from db import SQLiteRepository
+from db.repository import Repository
+from pipeline.protocols import Sink
+from pipeline.sources.file import FileStreamSource
+from pipeline.sources.stream import ThreadedStreamSource
+from pipeline.types import Frame
+
+logger = logging.getLogger(__name__)
+
+# Global runtime state
+# (kept as module-level singletons for current single-instance server design)
+db: Repository = SQLiteRepository()
+pipeline = None
+source: ThreadedStreamSource | None = None
+live_frame: Frame | None = None
+
+
+SourceFactory = Callable[[str], ThreadedStreamSource]
+SinkFactory = Callable[[ThreadedStreamSource], list[Sink]]
+PipelineFactory = Callable[[ThreadedStreamSource, list[Sink]], Any]
+
+ClipSourceFactory = Callable[[str], FileStreamSource]
+ClipSinkFactory = Callable[[FileStreamSource, str], list[Sink]]
+ClipPipelineFactory = Callable[[FileStreamSource, list[Sink]], Any]
+
+
+class LiveFrameSink(Sink):
+    """Captures the latest frame so API endpoints can stream it."""
+
+    def handle(self, frame: Frame):
+        global live_frame
+        live_frame = frame
+
+    def close(self):
+        pass
+
+
+class PipelineManager:
+    """Controls the live pipeline lifecycle (start/stop/pause/resume/switch)."""
+
+    def __init__(self):
+        self._lock = threading.RLock()
+        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
+        self._pipeline_factory: PipelineFactory | None = None
+
+        self._source_spec: str = "0"
+        self.running = False
+        self.starting = False
+        self.stopping = False
+        self.paused = False
+        self.last_error: str | None = None
+
+    def configure(
+        self,
+        source_factory: SourceFactory,
+        sink_factory: SinkFactory,
+        pipeline_factory: PipelineFactory,
+        initial_source: str = "0",
+    ):
+        with self._lock:
+            self._source_factory = source_factory
+            self._sink_factory = sink_factory
+            self._pipeline_factory = pipeline_factory
+            self._source_spec = initial_source
+
+    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__)
+        if sinks is None:
+            self._sinks = []
+
+    def _run_wrapper(self, session_id: int):
+        global pipeline, source
+        assert pipeline is not None and source is not None
+        try:
+            pipeline.process_stream(source)
+        except Exception as e:
+            with self._lock:
+                self.last_error = str(e)
+            logger.exception("Pipeline crashed")
+        finally:
+            with self._lock:
+                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
+        with self._lock:
+            if self.starting:
+                return False, "Pipeline is starting"
+
+            if self.stopping:
+                return False, "Pipeline is stopping"
+
+            if self.running:
+                return False, "Pipeline already running"
+
+            if source_spec is not None:
+                self._source_spec = source_spec
+
+            if (
+                not self._source_factory
+                or not self._sink_factory
+                or not self._pipeline_factory
+            ):
+                return False, "PipelineManager not configured"
+
+            source_spec_local = self._source_spec
+            source_factory = self._source_factory
+            sink_factory = self._sink_factory
+            pipeline_factory = self._pipeline_factory
+            self.starting = True
+
+        try:
+            source_obj = source_factory(source_spec_local)
+            sinks_obj = sink_factory(source_obj)
+            pipeline_obj = pipeline_factory(source_obj, sinks_obj)
+        except Exception as e:
+            with self._lock:
+                self.starting = False
+                self.last_error = str(e)
+            return False, f"Failed to build pipeline: {e}"
+
+        with self._lock:
+            if self.running:
+                self.starting = False
+                try:
+                    source_obj.stop()
+                except Exception:
+                    logger.exception("Error stopping source after race")
+                return False, "Pipeline already running"
+
+            source = source_obj
+            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",
+            )
+            self._thread.start()
+            self.starting = False
+            self.running = True
+            self.paused = False
+            self.last_error = None
+            return True, f"Pipeline started (source={self._source_spec})"
+
+    def start_async(self, source_spec: str | None = None) -> tuple[bool, str]:
+        with self._lock:
+            if self.starting:
+                return False, "Pipeline is starting"
+            if self.running:
+                return False, "Pipeline already running"
+
+            def _runner():
+                ok, msg = self.start(source_spec)
+                if ok:
+                    logger.info(msg)
+                else:
+                    logger.warning(msg)
+
+            self._start_thread = threading.Thread(
+                target=_runner,
+                daemon=True,
+                name="pipeline-start",
+            )
+            self._start_thread.start()
+            return True, "Pipeline start requested"
+
+    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:
+                source_obj.stop()
+            except Exception:
+                logger.exception("Error stopping source")
+
+        if thread and thread.is_alive():
+            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
+            source = None
+            return True, "Pipeline stopped"
+
+    def pause(self) -> tuple[bool, str]:
+        with self._lock:
+            if not self.running:
+                return False, "Pipeline is not running"
+
+        ok, msg = self.stop()
+        if ok:
+            with self._lock:
+                self.paused = True
+            return True, "Pipeline paused"
+        return False, msg
+
+    def resume(self) -> tuple[bool, str]:
+        with self._lock:
+            if self.running:
+                return False, "Pipeline already running"
+            if not self.paused:
+                return False, "Pipeline is not paused"
+
+        ok, msg = self.start()
+        if ok:
+            with self._lock:
+                self.paused = False
+            return True, "Pipeline resumed"
+        return False, msg
+
+    def switch_source(self, source_spec: str) -> tuple[bool, str]:
+        with self._lock:
+            self._source_spec = source_spec
+            running = self.running
+            paused = self.paused
+
+        if running:
+            self.stop()
+            return self.start(source_spec)
+
+        if paused:
+            return True, f"Source set to {source_spec} (will apply on resume)"
+
+        return self.start(source_spec)
+
+    def status(self) -> dict[str, Any]:
+        with self._lock:
+            return {
+                "running": self.running,
+                "starting": self.starting,
+                "stopping": self.stopping,
+                "paused": self.paused,
+                "source": self._source_spec,
+                "last_error": self.last_error,
+                "thread_alive": bool(self._thread and self._thread.is_alive()),
+            }
+
+
+manager = PipelineManager()
+
+
+@dataclass
+class ClipJob:
+    job_id: str
+    input_path: str
+    submitted_at: str
+    source_name: str | None = None
+    status: str = "queued"  # queued | running | done | failed
+    started_at: str | None = None
+    finished_at: str | None = None
+    error: str | None = None
+
+    def to_dict(self) -> dict[str, Any]:
+        return {
+            "job_id": self.job_id,
+            "input_path": self.input_path,
+            "source_name": self.source_name,
+            "status": self.status,
+            "submitted_at": self.submitted_at,
+            "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."""
+
+    def __init__(self):
+        self._lock = threading.RLock()
+        self._queue: Queue[str] = Queue()
+        self._jobs: dict[str, ClipJob] = {}
+        self._ordered_ids: list[str] = []
+
+        self._source_factory: ClipSourceFactory | None = None
+        self._sink_factory: ClipSinkFactory | None = None
+        self._pipeline_factory: ClipPipelineFactory | None = None
+
+        self._shutdown = False
+        self._worker = threading.Thread(
+            target=self._worker_loop,
+            daemon=True,
+            name="clip-worker",
+        )
+        self._worker.start()
+
+    def configure(
+        self,
+        source_factory: ClipSourceFactory,
+        sink_factory: ClipSinkFactory,
+        pipeline_factory: ClipPipelineFactory,
+    ):
+        with self._lock:
+            self._source_factory = source_factory
+            self._sink_factory = sink_factory
+            self._pipeline_factory = pipeline_factory
+
+    def submit(self, input_path: str, source_name: str | None = None) -> dict[str, Any]:
+        now = datetime.now().isoformat(timespec="seconds")
+        job_id = f"job_{int(time.time() * 1000)}"
+        job = ClipJob(
+            job_id=job_id,
+            input_path=input_path,
+            source_name=source_name,
+            submitted_at=now,
+        )
+
+        with self._lock:
+            self._jobs[job_id] = job
+            self._ordered_ids.append(job_id)
+            self._queue.put(job_id)
+            return job.to_dict()
+
+    def get(self, job_id: str) -> dict[str, Any] | None:
+        with self._lock:
+            job = self._jobs.get(job_id)
+            return job.to_dict() if job else None
+
+    def list(self, limit: int = 20) -> list[dict[str, Any]]:
+        with self._lock:
+            ids = self._ordered_ids[-limit:]
+            ids.reverse()
+            return [self._jobs[job_id].to_dict() for job_id in ids]
+
+    def status(self) -> dict[str, Any]:
+        with self._lock:
+            queued = sum(1 for job in self._jobs.values() if job.status == "queued")
+            running = sum(1 for job in self._jobs.values() if job.status == "running")
+            done = sum(1 for job in self._jobs.values() if job.status == "done")
+            failed = sum(1 for job in self._jobs.values() if job.status == "failed")
+            return {
+                "queued": queued,
+                "running": running,
+                "done": done,
+                "failed": failed,
+                "worker_alive": self._worker.is_alive(),
+            }
+
+    def shutdown(self):
+        with self._lock:
+            self._shutdown = True
+
+        if self._worker.is_alive():
+            self._worker.join(timeout=2.0)
+
+    def _set_job_state(
+        self,
+        job_id: str,
+        status: str,
+        *,
+        started_at: str | None = None,
+        finished_at: str | None = None,
+        error: str | None = None,
+    ):
+        with self._lock:
+            job = self._jobs.get(job_id)
+            if not job:
+                return
+            job.status = status
+            if started_at is not None:
+                job.started_at = started_at
+            if finished_at is not None:
+                job.finished_at = finished_at
+            if error is not None:
+                job.error = error
+
+    def _worker_loop(self):
+        while True:
+            with self._lock:
+                if self._shutdown:
+                    return
+
+            try:
+                job_id = self._queue.get(timeout=0.5)
+            except Empty:
+                continue
+
+            self._process_job(job_id)
+            self._queue.task_done()
+
+    def _process_job(self, job_id: str):
+        with self._lock:
+            job = self._jobs.get(job_id)
+            source_factory = self._source_factory
+            sink_factory = self._sink_factory
+            pipeline_factory = self._pipeline_factory
+
+        if not job:
+            return
+
+        if not source_factory or not sink_factory or not pipeline_factory:
+            self._set_job_state(job_id, "failed", error="ClipJobManager not configured")
+            return
+
+        self._set_job_state(
+            job_id,
+            "running",
+            started_at=datetime.now().isoformat(timespec="seconds"),
+            error=None,
+        )
+
+        paused_live = False
+        try:
+            live_status = manager.status()
+            if live_status.get("running"):
+                ok, _ = manager.pause()
+                paused_live = ok
+
+            src = source_factory(job.input_path)
+            sinks = sink_factory(src, job_id)
+            pipe = pipeline_factory(src, sinks)
+
+            try:
+                pipe.process_stream(src)
+            finally:
+                try:
+                    src.stop()
+                except Exception:
+                    logger.exception("Failed to stop clip source for %s", job_id)
+
+                for sink in sinks:
+                    try:
+                        sink.close()
+                    except Exception:
+                        logger.exception("Failed to close clip sink for %s", job_id)
+
+            self._set_job_state(
+                job_id,
+                "done",
+                finished_at=datetime.now().isoformat(timespec="seconds"),
+            )
+            logger.info("Clip job completed: %s", job_id)
+        except Exception as e:
+            logger.exception("Clip job failed: %s", job_id)
+            self._set_job_state(
+                job_id,
+                "failed",
+                finished_at=datetime.now().isoformat(timespec="seconds"),
+                error=str(e),
+            )
+        finally:
+            if paused_live:
+                manager.resume()
+
+
+clip_jobs = ClipJobManager()
+
+
+def get_lan_address() -> tuple[str, str]:
+    """Return (ip, hostname) for the primary LAN interface."""
+    ip = "127.0.0.1"
+    try:
+        probe = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
+        probe.connect(("8.8.8.8", 80))
+        ip = probe.getsockname()[0]
+        probe.close()
+    except Exception:
+        pass
+    hostname = socket.getfqdn()
+    return ip, hostname
+
+
+def run_pipeline():
+    """Blocking call used by the background pipeline thread."""
+    assert pipeline is not None and source is not None
+    try:
+        pipeline.process_stream(source)
+    except Exception:
+        logger.exception("Pipeline crashed")