tf

you immediately see what subject materials im reading and where im at

.env.example

@@ -0,0 +1,5 @@
+# what the helly
+DB_USER=smoke_admin
+DB_PASSWORD=123456
+DB_HOST=localhost
+DB_NAME=smoke_detection

.gitattributes

@@ -3,3 +3,4 @@
 *.jsonl filter=lfs diff=lfs merge=lfs -text
 *.png filter=lfs diff=lfs merge=lfs -text
 *.jpg filter=lfs diff=lfs merge=lfs -text
+*.db filter=lfs diff=lfs merge=lfs -text

.gitignore

@@ -9,4 +9,15 @@ wheels/
 # Virtual environments
 .venv
 
-Group_project_*
+# Local sqlite
+*.db*
+
+# artifacts/logs i forgor what are the paths uhhhhhhh
+alerts.jsonl
+output.jsonl
+output.mp4
+static/
+screenshots/
+events/
+logs/
+media/

.gitmodules

@@ -0,0 +1,3 @@
+[submodule "SmokeDetectionApp"]
+	path = SmokeDetectionApp
+	url = https://github.com/ldlda/SmokeDihtectionApp.git

.vscode/settings.json

@@ -6,5 +6,6 @@
     "**/cache/": true
   },
   "pylint.args": ["--generated-members=cv2.*", "--extension-pkg-whitelist=cv2"],
+  "java.configuration.updateBuildConfiguration": "interactive"
   // "editor.fontFamily": "'JetbrainsMono Nerd Font', Consolas, 'Courier New', monospace"
 }

README.md

@@ -7,19 +7,16 @@ A robust, multi-stage computer vision pipeline designed to detect smoking events
 The pipeline operates in a "Fail-Fast" cascade with **early exit** optimization:
 
 1. **Broad Phase (Person Detection)**:
-
    - Detects humans in the frame.
    - _Optimization_: If no people are found, processing stops for that frame.
 
 2. **Narrow Phase (Cigarette Candidate Detection)**:
-
    - Crops the detected person (with configurable margin for context).
    - Runs a specialized YOLO detector to find small objects (cigarettes/lighters).
    - _Early Exit_: If confidence ≥ 0.8, skips verification phase entirely.
    - _Filter_: Rejects candidates smaller than 10x10 pixels.
 
 3. **Verification Phase (Classification)**:
-
    - Crops the candidate object with padding.
    - Runs a dedicated Binary Classifier to verify "Smoking" vs "Not Smoking".
    - _Output_: Only high-confidence events are flagged as alerts.
@@ -34,6 +31,8 @@ The pipeline operates in a "Fail-Fast" cascade with **early exit** optimization:
   - **Active Mode (Max FPS)**: Switches to full speed when activity is detected.
   - **Cooldown**: Stays active for 2 seconds after people leave frame.
 - **Non-Maximum Suppression (NMS)**: Cleans up overlapping boxes per-class.
+- **Weighted Boxes Fusion (WBF)**: Alternative to NMS that merges overlapping boxes for better localization.
+- **Inference Budget Caps**: Limits expensive model calls (e.g., max 3 person crops, max 5 verifications).
 - **Smart Event Recording**: Captures 2s before + 3s after each smoking event as separate clips.
 - **Live Display**: Real-time visualization window with 'q' to quit.
 - **Metrics Tracking**: FPS, latency, detection counts with summary on exit.
@@ -56,7 +55,63 @@ cd gp
 uv sync
 ```
 
-### Running the Pipeline
+### Running the Server
+
+The server runs the detection pipeline in the background while serving a REST API for the mobile app.
+
+```bash
+# 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
+
+# RTSP camera
+uv run server.py --source rtsp://admin:[email protected]:554/stream
+
+# Custom host/port
+uv run server.py --host 0.0.0.0 --port 9000
+
+# Via uvicorn directly (uses env vars for config)
+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:
+
+```
+==================================================
+  Server ready
+  Local:    http://127.0.0.1:8000
+  Network:  http://192.168.1.42:8000
+  Hostname: http://lda.local:8000
+==================================================
+```
+
+Configuration can also be set via environment variables:
+
+| Variable        | Default                    | Description          |
+| :-------------- | :------------------------- | :------------------- |
+| `CAMERA_SOURCE` | `0`                        | Webcam index or RTSP |
+| `PERSON_MODEL`  | `models/human-det-best.pt` | Person model path    |
+| `CIG_MODEL`     | `models/cig-det-best.pt`   | Cigarette model path |
+| `VERIFY_MODEL`  | `models/cig-cls-best.pt`   | Classifier path      |
+| `STAGED`        | `false`                    | Use stage pipeline   |
+| `USE_VLM`       | `false`                    | Enable VLM stage     |
+| `VLM_API_KEY`   |                            | API key for VLM      |
+| `VLM_PROVIDER`  | `openai`                   | `openai`/`anthropic` |
+
+### Running the Pipeline CLI (standalone)
 
 ```bash
 # Run on a local video file
@@ -78,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
@@ -105,51 +160,54 @@ uv run -m tools.webcam_check
 The system generates multiple artifacts during execution:
 
 1. **Visual Debug (`output.mp4`)**:
-
    - A video file with bounding boxes drawn for all stages.
    - **Blue**: Detected Person.
    - **Yellow**: Cigarette Candidate.
    - **Red**: Confirmed Smoking Event.
 
 2. **Frame Metadata (`output.jsonl`)**:
-
    - A synchronized JSON Lines file containing raw detection data for every frame.
 
 3. **Alert Log (`alerts.jsonl`)**:
-
    - A structured log file containing **only** confirmed smoking events.
 
 4. **Event Clips (`events/event_<timestamp>.mp4`)**:
-
    - Smart clips containing only the smoking incidents.
    - Includes 2 seconds of pre-event context and 3 seconds post-event.
    - Automatically extends if smoking continues.
 
 5. **Screenshots (`screenshots/alert_<frame>_<conf>.jpg`)**:
-
    - Annotated JPG images saved when alerts fire.
    - Useful for evidence/audit trails without full video.
 
 6. **Application Logs (`logs/pipeline.log`)**:
-
    - Rotating log files with detailed debug information.
    - Includes thread names, function calls, and timing data.
 
 7. **Metrics Summary** (printed on exit):
-
    - Total frames, average FPS, latency, detection counts by label.
 
 ## 📂 Project Structure
 
 ```text
 gp/
-├── main.py                  # CLI Entry point
+├── 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
 ├── logs/                    # Rotating application logs
 ├── events/                  # Smart event clips (auto-created)
 ├── screenshots/             # Alert evidence images
 │
+├── db/                      # Database abstraction layer
+│   ├── repository.py        # Repository protocol + User/DetectionRecord
+│   ├── sqlite.py            # SQLiteRepository (zero-setup default)
+│   └── mysql.py             # MySQLRepository (optional)
+│
 ├── pipeline/
 │   ├── config.py            # PipelineConfig dataclass (all thresholds)
 │   ├── core.py              # SmokingDetectionPipeline (classic, fixed stages)
@@ -166,6 +224,7 @@ gp/
 │   │
 │   ├── sinks/               # Output handlers
 │   │   ├── alerts.py        # FileAlertStore (JSONL)
+│   │   ├── database.py      # DatabaseSink (saves alerts to DB)
 │   │   ├── display.py       # WindowDisplaySink (live preview)
 │   │   ├── events.py        # SmartEventSink (clip recorder)
 │   │   ├── metrics.py       # MetricsSink (FPS, latency, counts)
@@ -179,12 +238,25 @@ gp/
 │   │   └── api.py           # APIPersonDetector, VLMVerifier (cloud stages)
 │   │
 │   └── ops/                 # Algorithms & utilities
-│       ├── nms.py           # Non-Maximum Suppression
+│       ├── nms.py           # NMS + Weighted Boxes Fusion (WBF)
+│       ├── selection.py     # Budget-aware box selection (top-K, merge)
 │       ├── 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/
+        ├── LoginActivity     # Auth screen + hidden server config
+        ├── MainActivity      # Dashboard with camera preview
+        ├── HistoryActivity   # Alert history list
+        ├── StreamActivity    # Live MJPEG viewer
+        ├── ViewEvidenceActivity # Full-screen alert image
+        ├── SmokeAlert        # Foreground polling service
+        ├── HistoryAdapter    # RecyclerView adapter
+        └── ApiClient         # Dynamic server URL + HTTP helpers
 ```
 
 ## ⚙️ Configuration
@@ -207,6 +279,12 @@ class PipelineConfig:
     idle_interval: float = 0.5   # 2 FPS when idle
     active_interval: float = 0.0 # Max FPS when active
     cooldown_frames: int = 60    # Stay active after detection
+
+    # Inference budget caps
+    max_person_crops: int = 3       # Max cig model runs (0 = unlimited)
+    person_priority: str = "area"   # "area", "confidence", "area_conf"
+    merge_people: bool = False      # Merge overlapping people into regions
+    max_cig_verifications: int = 5  # Max verifier runs (0 = unlimited)
 ```
 
 Pass a custom config to the pipeline:
@@ -222,6 +300,69 @@ config = PipelineConfig(
 pipeline = SmokingDetectionPipeline(..., config=config)
 ```
 
+## API Endpoints
+
+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 (optionally by `user_id`) |
+| 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.
+
+- **Dynamic server URL**: Tap the "Smoking Alert" title on the login screen **5 times** to configure the server address. The URL is persisted and verified via `/ping` before saving.
+- **Background polling**: `SmokeAlert` foreground service checks `/history` every 5 seconds and pushes a notification on new detections.
+- **Edge-to-edge**: All activities draw behind system bars with proper inset handling.
+
 ## 🔧 Performance Tuning
 
 For RTSP streams, the `ThreadedStreamSource` uses a background thread to prevent frame buildup and handles reconnection automatically.
@@ -232,6 +373,26 @@ Key tuning parameters:
 - `active_interval`: Set to 0.0 for max detection rate
 - `cig_high_confidence`: Lower for faster alerts, higher for fewer false positives
 
+### Inference Budget
+
+When using lightweight (nano) models that produce many false positives, cap expensive downstream calls:
+
+```python
+config = PipelineConfig(
+    max_person_crops=3,        # Only process 3 people max per frame
+    person_priority="area",    # Prioritize largest (closest to camera)
+    merge_people=True,         # Merge overlapping people into regions
+    merge_people_iou=0.3,      # IoU threshold for merging
+    max_cig_verifications=5,   # Only verify top 5 cigarette candidates
+)
+```
+
+| Priority     | Strategy                                            |
+| ------------ | --------------------------------------------------- |
+| `area`       | Largest boxes first (closer to camera, more detail) |
+| `confidence` | Highest scoring boxes first                         |
+| `area_conf`  | Balanced: area × confidence                         |
+
 ## 🔌 Pluggable Stage System
 
 The `--staged` flag enables the modular stage pipeline, allowing you to mix local models with cloud APIs:
@@ -239,7 +400,7 @@ The `--staged` flag enables the modular stage pipeline, allowing you to mix loca
 ### Stage Types
 
 | Stage                    | Description              | Use Case                 |
-|:------|:------------|:---------|
+| :----------------------- | :----------------------- | :----------------------- |
 | `LocalPersonDetector`    | YOLO person detection    | Default, GPU/CPU         |
 | `LocalCigaretteDetector` | YOLO cigarette detection | Default, GPU/CPU         |
 | `LocalVerifier`          | YOLO classifier          | Default, GPU/CPU         |

db/__init__.py

@@ -0,0 +1,18 @@
+"""Database abstraction layer.
+
+Provides a Repository protocol and concrete implementations for SQLite and MySQL.
+The server doesn't care which one it talks to.
+"""
+
+from db.repository import Repository
+from db.sqlite import SQLiteRepository
+
+__all__ = ["Repository", "SQLiteRepository"]
+
+# Optional MySQL import (requires mysql-connector-python)
+try:
+    from db.mysql import MySQLRepository
+
+    __all__.append("MySQLRepository")
+except ImportError:
+    pass

db/mysql.py

@@ -0,0 +1,138 @@
+"""MySQL implementation of the Repository."""
+
+import os
+from typing import Optional
+
+import mysql.connector  # type: ignore[import-untyped]
+
+from db.repository import DetectionRecord, User
+
+
+class MySQLRepository:
+    """MySQL-backed repository. Needs a running MySQL server + .env config."""
+
+    def __init__(
+        self,
+        user: Optional[str] = None,
+        password: Optional[str] = None,
+        host: Optional[str] = None,
+        database: Optional[str] = None,
+    ):
+        self.config = {
+            "user": user or os.getenv("DB_USER", "smoke_admin"),
+            "password": password or os.getenv("DB_PASSWORD", "123456"),
+            "host": host or os.getenv("DB_HOST", "localhost"),
+            "database": database or os.getenv("DB_NAME", "smoke_detection"),
+        }
+
+    def _connect(self):
+        return mysql.connector.connect(**self.config)
+
+    def initialize(self) -> None:
+        conn = self._connect()
+        cursor = conn.cursor()
+        cursor.execute(
+            """CREATE TABLE IF NOT EXISTS users (
+                id INT AUTO_INCREMENT PRIMARY KEY,
+                username VARCHAR(50) UNIQUE NOT NULL,
+                password VARCHAR(255) NOT NULL
+            )"""
+        )
+        cursor.execute(
+            """CREATE TABLE IF NOT EXISTS detections (
+                id INT AUTO_INCREMENT PRIMARY KEY,
+                user_id INT,
+                image_path VARCHAR(255) NOT NULL,
+                confidence FLOAT,
+                timestamp TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
+                camera_source VARCHAR(50) DEFAULT 'Laptop Webcam',
+                FOREIGN KEY (user_id) REFERENCES users(id) ON DELETE CASCADE
+            )"""
+        )
+        cursor.execute(
+            "INSERT IGNORE INTO users (username, password) VALUES ('admin', '123456')"
+        )
+        conn.commit()
+        conn.close()
+
+    # --- Users ---
+
+    def get_user_by_username(self, username: str) -> Optional[User]:
+        conn = self._connect()
+        cursor = conn.cursor(dictionary=True)
+        cursor.execute("SELECT * FROM users WHERE username = %s", (username,))
+        row = cursor.fetchone()
+        conn.close()
+        return User(**row) if row else None
+
+    def get_user_by_credentials(self, username: str, password: str) -> Optional[User]:
+        conn = self._connect()
+        cursor = conn.cursor(dictionary=True)
+        cursor.execute(
+            "SELECT * FROM users WHERE username = %s AND password = %s",
+            (username, password),
+        )
+        row = cursor.fetchone()
+        conn.close()
+        return User(**row) if row else None
+
+    def create_user(self, username: str, password: str) -> int:
+        conn = self._connect()
+        cursor = conn.cursor()
+        cursor.execute(
+            "INSERT INTO users (username, password) VALUES (%s, %s)",
+            (username, password),
+        )
+        conn.commit()
+        new_id = cursor.lastrowid
+        conn.close()
+        return new_id  # type: ignore[return-value]
+
+    # --- Detections ---
+
+    def save_detection(
+        self,
+        user_id: int,
+        image_path: str,
+        confidence: float,
+        camera_source: str = "Laptop Webcam",
+    ) -> int:
+        conn = self._connect()
+        cursor = conn.cursor()
+        cursor.execute(
+            "INSERT INTO detections (user_id, image_path, confidence, camera_source) VALUES (%s, %s, %s, %s)",
+            (user_id, image_path, confidence, camera_source),
+        )
+        conn.commit()
+        new_id = cursor.lastrowid
+        conn.close()
+        return new_id  # type: ignore[return-value]
+
+    def get_detections(self, user_id: Optional[int] = None) -> list[DetectionRecord]:
+        conn = self._connect()
+        cursor = conn.cursor(dictionary=True)
+        if user_id is not None:
+            cursor.execute(
+                "SELECT * FROM detections WHERE user_id = %s ORDER BY timestamp DESC",
+                (user_id,),
+            )
+        else:
+            cursor.execute("SELECT * FROM detections ORDER BY timestamp DESC")
+        rows = cursor.fetchall()
+        conn.close()
+        return [DetectionRecord(**r) for r in rows]
+
+    def clear_detections(self, user_id: Optional[int] = None) -> int:
+        conn = self._connect()
+        cursor = conn.cursor()
+        if user_id is None:
+            cursor.execute("DELETE FROM detections")
+        else:
+            cursor.execute("DELETE FROM detections WHERE user_id = %s", (user_id,))
+        conn.commit()
+        count = cursor.rowcount
+        conn.close()
+        return count
+
+    def close(self) -> None:
+        pass  # MySQL uses per-call connections

db/repository.py

@@ -0,0 +1,68 @@
+"""Repository protocol - the interface all DB backends must implement."""
+
+from dataclasses import dataclass
+from typing import Optional, Protocol, runtime_checkable
+
+
+@dataclass
+class User:
+    id: int
+    username: str
+    password: str  # plaintext for now (legacy compat) - hash this later
+
+
+@dataclass
+class DetectionRecord:
+    id: int
+    user_id: int
+    image_path: str
+    confidence: float
+    timestamp: str
+    camera_source: str
+
+
+@runtime_checkable
+class Repository(Protocol):
+    """Database operations for the smoking detection server."""
+
+    def initialize(self) -> None:
+        """Create tables if they don't exist."""
+        ...
+
+    # --- Users ---
+
+    def get_user_by_username(self, username: str) -> Optional[User]:
+        """Find a user by username. Returns None if not found."""
+        ...
+
+    def get_user_by_credentials(self, username: str, password: str) -> Optional[User]:
+        """Find a user by username + password. Returns None if invalid."""
+        ...
+
+    def create_user(self, username: str, password: str) -> int:
+        """Create a new user. Returns the new user ID."""
+        ...
+
+    # --- Detections ---
+
+    def save_detection(
+        self,
+        user_id: int,
+        image_path: str,
+        confidence: float,
+        camera_source: str = "Laptop Webcam",
+    ) -> int:
+        """Save a detection record. Returns the new record ID."""
+        ...
+
+    def get_detections(self, user_id: Optional[int] = None) -> list[DetectionRecord]:
+        """Get detection history, optionally filtered by user. Newest first."""
+        ...
+
+    def clear_detections(self, user_id: Optional[int] = None) -> int:
+        """Delete detections, optionally filtered by user. Returns rows deleted."""
+        ...
+
+    def close(self) -> None:
+        """Close the database connection."""
+        ...

db/sqlite.py

@@ -0,0 +1,126 @@
+"""SQLite implementation of the Repository."""
+
+import sqlite3
+from pathlib import Path
+from typing import Optional
+
+from db.repository import DetectionRecord, User
+
+# Schema that works with SQLite (no MySQL-specific syntax)
+SCHEMA = """\
+CREATE TABLE IF NOT EXISTS users (
+    id INTEGER PRIMARY KEY AUTOINCREMENT,
+    username TEXT UNIQUE NOT NULL,
+    password TEXT NOT NULL
+);
+
+CREATE TABLE IF NOT EXISTS detections (
+    id INTEGER PRIMARY KEY AUTOINCREMENT,
+    user_id INTEGER,
+    image_path TEXT NOT NULL,
+    confidence REAL,
+    timestamp TEXT DEFAULT (datetime('now', 'localtime')),
+    camera_source TEXT DEFAULT 'Laptop Webcam',
+    FOREIGN KEY (user_id) REFERENCES users(id) ON DELETE CASCADE
+);
+
+INSERT OR IGNORE INTO users (username, password) VALUES ('admin', '123456');
+"""
+
+
+class SQLiteRepository:
+    """SQLite-backed repository. Zero setup, file-based."""
+
+    def __init__(self, db_path: str | Path = "data/smoke.db"):
+        self.db_path = Path(db_path)
+        self.db_path.parent.mkdir(parents=True, exist_ok=True)
+        self._conn: Optional[sqlite3.Connection] = None
+
+    def _get_conn(self) -> sqlite3.Connection:
+        if self._conn is None:
+            self._conn = sqlite3.connect(str(self.db_path), check_same_thread=False)
+            self._conn.row_factory = sqlite3.Row
+            self._conn.execute("PRAGMA journal_mode=WAL")
+            self._conn.execute("PRAGMA foreign_keys=ON")
+        return self._conn
+
+    def initialize(self) -> None:
+        conn = self._get_conn()
+        conn.executescript(SCHEMA)
+        conn.commit()
+
+    # --- Users ---
+
+    def get_user_by_username(self, username: str) -> Optional[User]:
+        row = (
+            self._get_conn()
+            .execute("SELECT * FROM users WHERE username = ?", (username,))
+            .fetchone()
+        )
+        return User(**dict(row)) if row else None
+
+    def get_user_by_credentials(self, username: str, password: str) -> Optional[User]:
+        row = (
+            self._get_conn()
+            .execute(
+                "SELECT * FROM users WHERE username = ? AND password = ?",
+                (username, password),
+            )
+            .fetchone()
+        )
+        return User(**dict(row)) if row else None
+
+    def create_user(self, username: str, password: str) -> int:
+        conn = self._get_conn()
+        cursor = conn.execute(
+            "INSERT INTO users (username, password) VALUES (?, ?)",
+            (username, password),
+        )
+        conn.commit()
+        return cursor.lastrowid  # type: ignore[return-value]
+
+    # --- Detections ---
+
+    def save_detection(
+        self,
+        user_id: int,
+        image_path: str,
+        confidence: float,
+        camera_source: str = "Laptop Webcam",
+    ) -> int:
+        conn = self._get_conn()
+        cursor = conn.execute(
+            "INSERT INTO detections (user_id, image_path, confidence, camera_source) VALUES (?, ?, ?, ?)",
+            (user_id, image_path, confidence, camera_source),
+        )
+        conn.commit()
+        return cursor.lastrowid  # type: ignore[return-value]
+
+    def get_detections(self, user_id: Optional[int] = None) -> list[DetectionRecord]:
+        conn = self._get_conn()
+        if user_id is not None:
+            rows = conn.execute(
+                "SELECT * FROM detections WHERE user_id = ? ORDER BY timestamp DESC",
+                (user_id,),
+            ).fetchall()
+        else:
+            rows = conn.execute(
+                "SELECT * FROM detections ORDER BY timestamp DESC"
+            ).fetchall()
+        return [DetectionRecord(**dict(r)) for r in rows]
+
+    def clear_detections(self, user_id: Optional[int] = None) -> int:
+        conn = self._get_conn()
+        if user_id is None:
+            cursor = conn.execute("DELETE FROM detections")
+        else:
+            cursor = conn.execute(
+                "DELETE FROM detections WHERE user_id = ?", (user_id,)
+            )
+        conn.commit()
+        return cursor.rowcount
+
+    def close(self) -> None:
+        if self._conn:
+            self._conn.close()
+            self._conn = None

pipeline/config.py

@@ -2,6 +2,8 @@
 
 from dataclasses import dataclass
 
+from pipeline.ops.selection import SelectPriority
+
 
 @dataclass
 class PipelineConfig:
@@ -22,16 +24,30 @@ class PipelineConfig:
     min_cig_size: int = 10  # Minimum cig box dimension
 
     # Person crop expansion
-    person_margin: float = 0.2  # 20% expansion around person box
+    person_margin: float = 1.0  # 100% expansion around person box
 
     # Verification crop padding
-    cig_crop_padding: float = 0.5  # 50% padding around cig for context
+    cig_crop_padding: float = 1.2  # 120% padding around cig for context
 
     # Adaptive sampling
     idle_interval: float = 0.5  # Seconds between inference when idle
     active_interval: float = 0.0  # Seconds between inference when active
     cooldown_frames: int = 60  # Frames to stay active after last detection
 
+    # === Inference Budget Caps ===
+    # Limits expensive model calls when there are too many detections
+
+    # Max person crops to run cig detection on (0 = unlimited)
+    max_person_crops: int = 3
+    # How to prioritize which people to process: "area", "confidence", "area_conf"
+    person_priority: SelectPriority = "area"
+    # Merge overlapping people into regions instead of processing separately
+    merge_people: bool = False
+    merge_people_iou: float = 0.3
+
+    # Max cigarette candidates to run verifier on (0 = unlimited)
+    max_cig_verifications: int = 5
+
 
 # Default config instance
 DEFAULT_CONFIG = PipelineConfig()

pipeline/ops/__init__.py

@@ -17,12 +17,23 @@ from pipeline.ops.geometry import (
     point_in_box,
     point_in_polygon,
 )
-from pipeline.ops.nms import nms, nms_by_class
+from pipeline.ops.nms import nms, nms_by_class, wbf, wbf_by_class
+from pipeline.ops.selection import (
+    merge_overlapping,
+    select_for_verification,
+    select_top_k,
+)
 
 __all__ = [
-    # NMS
+    # NMS / Box Fusion
     "nms",
     "nms_by_class",
+    "wbf",
+    "wbf_by_class",
+    # Selection / Budget
+    "select_top_k",
+    "select_for_verification",
+    "merge_overlapping",
     # Geometry
     "expand_box",
     "point_in_box",

pipeline/ops/nms.py

@@ -1,4 +1,4 @@
-"""Non-Maximum Suppression algorithms for detection filtering."""
+"""Box filtering algorithms: NMS and Weighted Boxes Fusion."""
 
 from typing import List
 
@@ -59,3 +59,119 @@ def nms_by_class(
         result.extend(nms(dets, iou_threshold))
 
     return result
+
+
+def wbf(
+    detections: List[Detection],
+    iou_threshold: float = 0.5,
+    skip_threshold: float = 0.0,
+) -> List[Detection]:
+    """
+    Weighted Boxes Fusion: Merge overlapping boxes instead of suppressing.
+
+    Unlike NMS which picks the best box and discards overlapping ones,
+    WBF fuses overlapping boxes by averaging their coordinates weighted
+    by confidence. This often gives better localization.
+
+    Args:
+        detections: List of Detection objects
+        iou_threshold: Boxes with IoU > threshold are fused together
+        skip_threshold: Ignore boxes with confidence below this
+
+    Returns:
+        Fused list of detections
+
+    Reference:
+        https://arxiv.org/abs/1910.13302
+    """
+    if not detections:
+        return []
+
+    # Filter low confidence
+    dets = [d for d in detections if d.score >= skip_threshold]
+    if not dets:
+        return []
+
+    # Sort by confidence (highest first)
+    dets = sorted(dets, key=lambda d: d.score, reverse=True)
+
+    # Clusters of fused boxes
+    clusters: List[List[Detection]] = []
+
+    for det in dets:
+        matched_cluster = None
+
+        # Find a cluster this box belongs to
+        for cluster in clusters:
+            # Check IoU with the fused box (first element is always the fused result)
+            if cluster[0].box.iou(det.box) >= iou_threshold:
+                matched_cluster = cluster
+                break
+
+        if matched_cluster is not None:
+            # Add to existing cluster and recompute fused box
+            matched_cluster.append(det)
+            _update_fused_box(matched_cluster)
+        else:
+            # Start new cluster
+            clusters.append([det])
+
+    # Return the fused box from each cluster (first element)
+    return [cluster[0] for cluster in clusters]
+
+
+def _update_fused_box(cluster: List[Detection]):
+    """
+    Recompute the fused box for a cluster.
+    The fused box is stored as cluster[0].
+    """
+    if len(cluster) == 1:
+        return
+
+    # Weighted average of coordinates
+    total_weight = sum(d.score for d in cluster)
+
+    x1 = sum(d.box.x1 * d.score for d in cluster) / total_weight
+    y1 = sum(d.box.y1 * d.score for d in cluster) / total_weight
+    x2 = sum(d.box.x2 * d.score for d in cluster) / total_weight
+    y2 = sum(d.box.y2 * d.score for d in cluster) / total_weight
+
+    # Fused confidence: average boosted by number of boxes
+    # More boxes agreeing = higher confidence (capped at 1.0)
+    avg_conf = total_weight / len(cluster)
+    fused_conf = min(1.0, avg_conf * (1 + 0.1 * (len(cluster) - 1)))
+
+    # Update the first element with fused values
+    fused = cluster[0]
+    fused.box.x1 = int(x1)
+    fused.box.y1 = int(y1)
+    fused.box.x2 = int(x2)
+    fused.box.y2 = int(y2)
+    fused.box.conf = fused_conf
+    fused.score = fused_conf
+    fused.metadata["wbf_count"] = len(cluster)
+
+
+def wbf_by_class(
+    detections: List[Detection],
+    iou_threshold: float = 0.5,
+    skip_threshold: float = 0.0,
+) -> List[Detection]:
+    """
+    Run WBF separately for each class label.
+    This prevents boxes of different classes from being fused together.
+    """
+    if not detections:
+        return []
+
+    # Group by label
+    by_label: dict[str, List[Detection]] = {}
+    for det in detections:
+        by_label.setdefault(det.label, []).append(det)
+
+    # Run WBF per class
+    result = []
+    for dets in by_label.values():
+        result.extend(wbf(dets, iou_threshold, skip_threshold))
+
+    return result

pipeline/ops/selection.py

@@ -0,0 +1,174 @@
+"""Budget-aware box selection for capping expensive inference calls."""
+
+from typing import List, Literal
+
+from pipeline.types import BoundingBox, Detection
+
+type SelectPriority = Literal["area", "confidence", "area_conf"]
+
+
+def select_top_k(
+    detections: List[Detection],
+    k: int,
+    diversity_iou: float = 0.5,
+    priority: SelectPriority = "area",
+) -> List[Detection]:
+    """
+    Select top K detections with spatial diversity.
+
+    Picks the K "best" boxes while ensuring they don't overlap too much.
+    This caps expensive downstream inference (e.g., max 3 person crops → cig model).
+
+    Args:
+        detections: Candidate detections
+        k: Maximum number to select
+        diversity_iou: Reject candidates overlapping selected boxes above this IoU
+        priority: How to rank candidates:
+            - "area": Larger boxes first (closer to camera, more detail)
+            - "confidence": Higher confidence first
+            - "area_conf": area * confidence (balanced)
+
+    Returns:
+        Up to K diverse, high-priority detections
+    """
+    if not detections or k <= 0:
+        return []
+
+    # Score each detection
+    def score(det: Detection) -> float:
+        box = det.box
+        # area = (box.x2 - box.x1) * (box.y2 - box.y1)
+        area = box.area
+        if priority == "area":
+            return area
+        elif priority == "confidence":
+            return det.score
+        else:  # area_conf
+            return area * det.score
+
+    # Sort by priority (highest first)
+    ranked = sorted(detections, key=score, reverse=True)
+
+    selected = []
+    for det in ranked:
+        if len(selected) >= k:
+            break
+
+        # Check diversity: don't select if overlaps too much with already selected
+        dominated = False
+        for sel in selected:
+            if det.box.iou(sel.box) > diversity_iou:
+                dominated = True
+                break
+
+        if not dominated:
+            selected.append(det)
+
+    return selected
+
+
+def merge_overlapping(
+    detections: List[Detection],
+    iou_threshold: float = 0.3,
+) -> List[Detection]:
+    """
+    Merge overlapping boxes into larger regions.
+
+    Instead of processing each overlapping person separately, merge them
+    into one bigger crop. Good when people are clustered together.
+
+    Args:
+        detections: List of detections (e.g., people)
+        iou_threshold: Merge boxes with IoU above this
+
+    Returns:
+        Merged detections (fewer, larger boxes)
+    """
+    if not detections:
+        return []
+
+    # Sort by confidence
+    dets = sorted(detections, key=lambda d: d.score, reverse=True)
+
+    clusters: List[List[Detection]] = []
+
+    for det in dets:
+        merged = False
+        for cluster in clusters:
+            # Check if this det overlaps with any box in cluster
+            for member in cluster:
+                if det.box.iou(member.box) >= iou_threshold:
+                    cluster.append(det)
+                    merged = True
+                    break
+            if merged:
+                break
+
+        if not merged:
+            clusters.append([det])
+
+    # Create merged boxes from clusters
+    result = []
+    for cluster in clusters:
+        merged_det = _merge_cluster(cluster)
+        result.append(merged_det)
+
+    return result
+
+
+def _merge_cluster(cluster: List[Detection]) -> Detection:
+    """Merge a cluster of detections into one encompassing box."""
+    if len(cluster) == 1:
+        return cluster[0]
+
+    # Bounding box that contains all
+    x1 = min(d.box.x1 for d in cluster)
+    y1 = min(d.box.y1 for d in cluster)
+    x2 = max(d.box.x2 for d in cluster)
+    y2 = max(d.box.y2 for d in cluster)
+
+    # Confidence: max of cluster (could also use average)
+    conf = max(d.score for d in cluster)
+
+    # Use label from highest confidence detection
+    best = max(cluster, key=lambda d: d.score)
+
+    return Detection(
+        box=BoundingBox(x1=x1, y1=y1, x2=x2, y2=y2, conf=conf),
+        score=conf,
+        label=best.label,
+        metadata={
+            "merged_count": len(cluster),
+            "original_boxes": [
+                (d.box.x1, d.box.y1, d.box.x2, d.box.y2) for d in cluster
+            ],
+        },
+    )
+
+
+def select_for_verification(
+    detections: List[Detection],
+    k: int,
+    min_confidence: float = 0.0,
+) -> List[Detection]:
+    """
+    Select top K cigarette candidates for expensive verification.
+
+    Simple confidence-based selection for the verifier stage.
+    Higher confidence cigs are more likely real → verify those first.
+
+    Args:
+        detections: Cigarette candidates
+        k: Max verifications to run
+        min_confidence: Skip candidates below this
+
+    Returns:
+        Top K candidates by confidence
+    """
+    if not detections or k <= 0:
+        return []
+
+    filtered = [d for d in detections if d.score >= min_confidence]
+    ranked = sorted(filtered, key=lambda d: d.score, reverse=True)
+
+    return ranked[:k]

pipeline/sinks/__init__.py

@@ -1,6 +1,7 @@
 """Sinks for pipeline output handling."""
 
 from pipeline.sinks.alerts import FileAlertStore
+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
@@ -8,6 +9,8 @@ from pipeline.sinks.screenshots import ScreenshotSink
 from pipeline.sinks.video import VideoFileStore
 
 __all__ = [
+    "DatabaseSink",
+    "DatabaseSmartEventSink",
     "FileAlertStore",
     "MetricsSink",
     "ScreenshotSink",

pipeline/sinks/database.py

@@ -0,0 +1,124 @@
+"""Database sink - saves detection alerts to the database."""
+
+import logging
+import time
+from pathlib import Path
+
+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__)
+
+
+class DatabaseSink:
+    """
+    Pipeline sink that saves smoking alerts to a database.
+
+    On ticket (alert), saves an annotated screenshot and inserts a detection record.
+    """
+
+    def __init__(
+        self,
+        db: Repository,
+        user_id: int = 1,
+        image_dir: str = "static/alerts",
+        camera_source: str = "Pipeline",
+    ):
+        self.db = db
+        self.user_id = user_id
+        self.camera_source = camera_source
+        self.image_dir = Path(image_dir)
+        self.image_dir.mkdir(parents=True, exist_ok=True)
+        self.count = 0
+
+        logger.info("DatabaseSink initialized (user_id=%d, dir=%s)", user_id, image_dir)
+
+    def handle(self, frame: Frame):
+        if not frame.ticket:
+            return
+
+        self.count += 1
+        ticket = frame.ticket
+
+        # Save annotated image
+        filename = f"alert_{int(time.time())}_{self.count}.jpg"
+        filepath = self.image_dir / filename
+
+        img = frame.data.copy()
+        draw_detections(img, frame.detections)
+        draw_alert_banner(img, f"SMOKING {ticket.confidence:.0%}")
+        cv2.imwrite(str(filepath), img)
+
+        # Insert into database
+        web_path = f"/static/alerts/{filename}"
+        try:
+            self.db.save_detection(
+                user_id=self.user_id,
+                image_path=web_path,
+                confidence=ticket.confidence,
+                camera_source=self.camera_source,
+            )
+            logger.info(
+                "Alert #%d saved to DB: %s (%.2f)",
+                self.count,
+                filename,
+                ticket.confidence,
+            )
+        except Exception as e:
+            logger.error("Failed to save alert to DB: %s", e)
+
+    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)

pipeline/sources/file.py

@@ -3,6 +3,7 @@
 import glob
 import logging
 import time
+from pathlib import Path
 from typing import Iterator
 
 import cv2
@@ -19,9 +20,10 @@ class FileStreamSource(StreamSource):
     def __init__(self, path: str):
         self.path = path
         self._fps = 30.0
+        self._image_exts = {".jpg", ".jpeg", ".png", ".bmp", ".webp"}
 
         # Probe for FPS if it's a single file
-        if "*" not in self.path:
+        if "*" not in self.path and not self._is_image_path(self.path):
             try:
                 cap = cv2.VideoCapture(self.path)
                 if cap.isOpened():
@@ -30,6 +32,16 @@ class FileStreamSource(StreamSource):
             except Exception as e:
                 logger.warning("Could not probe FPS from %s: %s", path, e)
 
+    def _is_image_path(self, value: str) -> bool:
+        return Path(value).suffix.lower() in self._image_exts
+
+    def _iter_image_files(self, directory: Path) -> list[str]:
+        files: list[str] = []
+        for ext in self._image_exts:
+            files.extend(glob.glob(str(directory / f"*{ext}")))
+            files.extend(glob.glob(str(directory / f"*{ext.upper()}")))
+        return sorted(set(files))
+
     @property
     def fps(self) -> float:
         return self._fps
@@ -45,6 +57,28 @@ class FileStreamSource(StreamSource):
                     yield Frame(timestamp=time.time(), data=img, frame_id=frame_id)
             return
 
+        file_path = Path(self.path)
+
+        # Handle single image file
+        if self._is_image_path(self.path):
+            img = cv2.imread(self.path)
+            if img is None:
+                logger.warning("Could not read image: %s", self.path)
+                return
+            logger.info("Loaded single image: %s", self.path)
+            yield Frame(timestamp=time.time(), data=img, frame_id=0)
+            return
+
+        # Handle directory of images
+        if file_path.exists() and file_path.is_dir():
+            files = self._iter_image_files(file_path)
+            logger.info("Loading %d images from directory: %s", len(files), self.path)
+            for frame_id, file in enumerate(files):
+                img = cv2.imread(file)
+                if img is not None:
+                    yield Frame(timestamp=time.time(), data=img, frame_id=frame_id)
+            return
+
         logger.info("Opening video file: %s", self.path)
         cap = cv2.VideoCapture(self.path)
         frame_id = 0

pipeline/stages/api.py

@@ -149,6 +149,7 @@ class APIPersonDetector(APIStageBase):
         return resp.json()
 
 
+# haiku 4.5 is multimodal but i fear its going to ping 65 million images per second (we need cache control? who are we? why cant we upload video?)
 class VLMVerifier(APIStageBase):
     """
     Vision-Language Model verification stage.

pipeline/stages/local.py

@@ -5,7 +5,12 @@ import time
 from typing import List, Optional
 
 from pipeline.config import DEFAULT_CONFIG, PipelineConfig
-from pipeline.ops import nms_by_class
+from pipeline.ops import (
+    merge_overlapping,
+    nms_by_class,
+    select_for_verification,
+    select_top_k,
+)
 from pipeline.protocols import Model, Verifier
 from pipeline.stages.protocol import StageContext, StageResult
 from pipeline.types import BoundingBox, Detection, Ticket
@@ -66,10 +71,39 @@ class LocalCigaretteDetector:
         cfg = self.cfg
         h_img, w_img = ctx.frame_data.shape[:2]
 
+        # === Budget: Limit how many people we process ===
+        people_to_process = ctx.people
+
+        if cfg.merge_people:
+            # Merge overlapping people into regions
+            people_to_process = merge_overlapping(
+                people_to_process, cfg.merge_people_iou
+            )
+            if len(people_to_process) < len(ctx.people):
+                logger.debug(
+                    "Merged %d people into %d regions",
+                    len(ctx.people),
+                    len(people_to_process),
+                )
+
+        if cfg.max_person_crops > 0 and len(people_to_process) > cfg.max_person_crops:
+            # Select top K diverse people by priority
+            people_to_process = select_top_k(
+                people_to_process,
+                k=cfg.max_person_crops,
+                priority=cfg.person_priority,
+            )
+            logger.debug(
+                "Budget: Processing %d/%d people (priority=%s)",
+                len(people_to_process),
+                len(ctx.people),
+                cfg.person_priority,
+            )
+
         all_cigs: List[Detection] = []
         early_hit = None
 
-        for person in ctx.people:
+        for person in people_to_process:
             # Expand person box for crop
             crop_box = self._expand_box(person.box, cfg.person_margin, w_img, h_img)
             ex1, ey1, ex2, ey2 = crop_box
@@ -195,12 +229,28 @@ class LocalVerifier:
         cfg = self.cfg
         h_img, w_img = ctx.frame_data.shape[:2]
 
+        # === Budget: Limit how many cigs we verify ===
+        cigs_to_verify = ctx.cigarettes
+
+        if (
+            cfg.max_cig_verifications > 0
+            and len(cigs_to_verify) > cfg.max_cig_verifications
+        ):
+            # Select top K by confidence (most likely real → verify first)
+            cigs_to_verify = select_for_verification(
+                cigs_to_verify,
+                k=cfg.max_cig_verifications,
+                min_confidence=cfg.cig_min_confidence,
+            )
+            logger.debug(
+                "Budget: Verifying %d/%d cigarettes",
+                len(cigs_to_verify),
+                len(ctx.cigarettes),
+            )
+
         confirmed: list[tuple[Detection, float]] = []
 
-        for cig_det in ctx.cigarettes:
+        for cig_det in cigs_to_verify:
-            offset = cig_det.metadata.get("crop_offset", (0, 0))
-            ex1, ey1 = offset
-
             # Extract padded crop for classifier
             crop = self._extract_cig_crop(ctx.frame_data, cig_det, w_img, h_img)
             if crop.size == 0:

pipeline/stages/runner.py

@@ -60,7 +60,7 @@ class StagePipeline:
                     stage.name,
                     ctx.stop_reason,
                 )
-                continue
+                continue  # break?
 
             try:
                 result = stage.process(ctx)

pipeline/types.py

@@ -10,7 +10,7 @@ class BoundingBox:
     y1: int
     x2: int
     y2: int
-    conf: float = 0.0
+    conf: float = 0.0  # legacy
 
     @property
     def area(self):

pyproject.toml

@@ -5,11 +5,17 @@ description = "Add your description here"
 readme = "README.md"
 requires-python = ">=3.13"
 dependencies = [
+    "dotenv>=0.9.9",
+    "fastapi>=0.133.0",
+    "mysql-connector-python>=9.6.0",
     "numpy>=2.4.0",
     "opencv-python>=4.11.0.86",
+    "python-dotenv>=1.2.1",
+    "python-multipart>=0.0.22",
     "torch>=2.9.1",
     "torchvision>=0.24.1",
     "ultralytics>=8.3.248",
+    "uvicorn[standard]>=0.41.0",
 ]
 
 [[tool.uv.index]]

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

@@ -0,0 +1,260 @@
+"""
+Smoking Detection Server
+========================
+
+FastAPI server that runs the full detection pipeline on a live webcam feed
+and exposes REST endpoints for the frontend (auth, history, video stream).
+
+Usage:
+    uvicorn server:app --host 0.0.0.0 --port 8000
+    python server.py                          # dev mode
+    python server.py --staged                 # use stage-based pipeline
+    python server.py --source rtsp://...      # RTSP camera
+"""
+
+import argparse
+import logging
+import os
+from contextlib import asynccontextmanager
+from pathlib import Path
+
+from fastapi import FastAPI
+from fastapi.staticfiles import StaticFiles
+
+import server.runtime as rt
+from pipeline.config import PipelineConfig
+from pipeline.core import SmokingDetectionPipeline
+from pipeline.models import YoloCigaretteDetector, YoloClassifier, YoloPersonDetector
+from pipeline.protocols import Sink, StreamSource
+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 server.routes import router as api_router
+
+logger = logging.getLogger(__name__)
+
+
+# ---------------------------------------------------------------------------
+# CLI args  (parsed before uvicorn when run as __main__)
+# ---------------------------------------------------------------------------
+def parse_args():
+    p = argparse.ArgumentParser(description="Smoking Detection Server")
+    p.add_argument(
+        "--source", type=str, default="0", help="Webcam index or RTSP URL (default: 0)"
+    )
+    p.add_argument("--person-model", default="models/human-det-best.pt")
+    p.add_argument("--cig-model", default="models/cig-det-best.pt")
+    p.add_argument("--verify-model", default="models/cig-cls-best.pt")
+    p.add_argument("--staged", action="store_true", help="Use stage-based pipeline")
+    p.add_argument("--vlm", action="store_true")
+    p.add_argument("--vlm-key", default=None)
+    p.add_argument("--vlm-provider", default="openai", choices=["openai", "anthropic"])
+    p.add_argument("--vlm-model", default="gpt-4-vision-preview")
+    p.add_argument("--host", default="0.0.0.0")
+    p.add_argument("--port", type=int, default=8000)
+    p.add_argument("--debug", action="store_true")
+    return p.parse_args()
+
+
+# Module-level defaults (overridden by __main__ block or env)
+_args = argparse.Namespace(
+    source=os.environ.get("CAMERA_SOURCE", "0"),
+    person_model=os.environ.get("PERSON_MODEL", "models/human-det-best.pt"),
+    cig_model=os.environ.get("CIG_MODEL", "models/cig-det-best.pt"),
+    verify_model=os.environ.get("VERIFY_MODEL", "models/cig-cls-best.pt"),
+    staged=os.environ.get("STAGED", "").lower() in ("1", "true", "yes"),
+    vlm=os.environ.get("USE_VLM", "").lower() in ("1", "true", "yes"),
+    vlm_key=os.environ.get("VLM_API_KEY"),
+    vlm_provider=os.environ.get("VLM_PROVIDER", "openai"),
+    vlm_model=os.environ.get("VLM_MODEL", "gpt-4-vision-preview"),
+    debug=False,
+)
+
+
+# ---------------------------------------------------------------------------
+# Lifespan - startup & shutdown
+# ---------------------------------------------------------------------------
+@asynccontextmanager
+async def app_lifespan(_app: FastAPI):
+    logging.basicConfig(
+        level=logging.DEBUG if _args.debug else logging.INFO,
+        format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
+        datefmt="%H:%M:%S",
+    )
+    logging.getLogger("ultralytics").setLevel(logging.WARNING)
+
+    # 1. Database
+    logger.info("Initializing database...")
+    rt.db.initialize()
+
+    # 2. Models
+    logger.info("Loading models...")
+    person_model = YoloPersonDetector(_args.person_model)
+    cig_model = YoloCigaretteDetector(_args.cig_model)
+    verifier = YoloClassifier(_args.verify_model)
+    logger.info("Models loaded.")
+
+    # 3. Configure live pipeline manager
+    def source_factory(source_spec: str) -> StreamSource:
+        if source_spec.lower() in {"phone", "android", "mobile"}:
+            rt.phone_source.reset()
+            return rt.phone_source
+        cam = int(source_spec) if source_spec.isdigit() else source_spec
+        return ThreadedStreamSource(cam)
+
+    def sink_factory(src: StreamSource) -> list[Sink]:
+        return [
+            rt.LiveFrameSink(),
+            DatabaseSmartEventSink(rt.db, fps=src.fps, output_dir="static/events"),
+            MetricsSink(log_interval=200),
+            FileAlertStore("alerts.jsonl"),
+        ]
+
+    def pipeline_factory(src: StreamSource, sinks: list[Sink]):
+        config = PipelineConfig(
+            idle_interval=0.5,
+            active_interval=0.0,
+            cooldown_frames=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.manager.configure(
+        source_factory=source_factory,
+        sink_factory=sink_factory,
+        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 = rt.get_lan_address()
+    port = _args.port if hasattr(_args, "port") else 8000
+    logger.info("=" * 50)
+    logger.info("  Server ready")
+    logger.info("  Local:    http://127.0.0.1:%d", port)
+    logger.info("  Network:  http://%s:%d", ip, port)
+    if hostname and hostname != ip:
+        logger.info("  Hostname: http://%s:%d", hostname, port)
+    logger.info("=" * 50)
+
+    yield  # --- app serves requests ---
+
+    # Shutdown
+    logger.info("Shutting down...")
+    rt.manager.stop()
+    rt.clip_jobs.shutdown()
+    rt.db.close()
+    logger.info("Goodbye.")
+
+
+# ---------------------------------------------------------------------------
+# 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")
+
+
+# ---------------------------------------------------------------------------
+# Entry-point
+# ---------------------------------------------------------------------------
+if __name__ == "__main__":
+    import uvicorn
+
+    _args = parse_args()
+    uvicorn.run(app, host=_args.host, port=_args.port)

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,371 @@
+"""API route definitions for the smoking detection server."""
+
+import asyncio
+import shutil
+import time
+from pathlib import Path
+
+import cv2
+from fastapi import (
+    APIRouter,
+    File,
+    Form,
+    Response,
+    UploadFile,
+    WebSocket,
+    WebSocketDisconnect,
+)
+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.websocket("/ws/updates")
+async def ws_updates(websocket: WebSocket, history_limit: int = 10):
+    await websocket.accept()
+    clamped_limit = max(1, min(history_limit, 50))
+
+    try:
+        while True:
+            try:
+                records = rt.db.get_detections(None)[:clamped_limit]
+                history = [
+                    {
+                        "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:
+                history = []
+
+            await websocket.send_json(
+                {
+                    "type": "snapshot",
+                    "live": rt.manager.status(),
+                    "history": history,
+                    "history_count": len(history),
+                    "server_time": int(time.time()),
+                }
+            )
+            await asyncio.sleep(1.0)
+    except WebSocketDisconnect:
+        return
+
+
+@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/start_phone")
+def live_start_phone():
+    """Start live pipeline using phone frame push source."""
+    ok, message = rt.manager.switch_source("phone")
+    return {
+        "status": "success" if ok else "fail",
+        "message": message,
+        "live": rt.manager.status(),
+        "hint": "Push JPEG frames to /live/ws/push_frames",
+    }
+
+
+@router.get("/live/source_history")
+def live_source_history(limit: int = 10):
+    return {
+        "status": "success",
+        "current_source": rt.manager.status().get("source"),
+        "history": rt.manager.source_history(limit=limit),
+    }
+
+
+@router.post("/live/switch_previous")
+def live_switch_previous():
+    ok, message = rt.manager.switch_previous()
+    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("/live/push_frame")
+async def live_push_frame(frame: UploadFile = File(...)):
+    """Push one JPEG frame from phone/app into live source queue."""
+    try:
+        data = await frame.read()
+    finally:
+        await frame.close()
+
+    if not data:
+        return {"status": "fail", "message": "Empty frame payload"}
+
+    ok, message = rt.push_live_frame_bytes(data)
+    return {
+        "status": "success" if ok else "fail",
+        "message": message,
+        "live": rt.manager.status(),
+    }
+
+
+@router.websocket("/live/ws/push_frames")
+async def live_ws_push_frames(websocket: WebSocket):
+    await websocket.accept()
+    accepted = 0
+    dropped = 0
+
+    try:
+        while True:
+            message = await websocket.receive()
+            if message.get("type") == "websocket.disconnect":
+                break
+
+            payload = message.get("bytes")
+            if payload is None:
+                text_payload = message.get("text")
+                if text_payload == "ping":
+                    await websocket.send_json({"type": "pong"})
+                continue
+
+            ok, _ = rt.push_live_frame_bytes(payload)
+            if ok:
+                accepted += 1
+            else:
+                dropped += 1
+
+            if (accepted + dropped) % 60 == 0:
+                await websocket.send_json(
+                    {
+                        "type": "stats",
+                        "accepted": accepted,
+                        "dropped": dropped,
+                        "live": rt.manager.status(),
+                    }
+                )
+    except WebSocketDisconnect:
+        return
+
+
+@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,650 @@
+"""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
+
+import cv2
+import numpy as np
+
+from db import SQLiteRepository
+from db.repository import Repository
+from pipeline.protocols import Sink, StreamSource
+from pipeline.sources.file import FileStreamSource
+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: StreamSource | None = None
+live_frame: Frame | None = None
+
+
+SourceFactory = Callable[[str], StreamSource]
+SinkFactory = Callable[[StreamSource], list[Sink]]
+PipelineFactory = Callable[[StreamSource, 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 PushFrameSource(StreamSource):
+    """Frame source that receives JPEG frames pushed via API."""
+
+    def __init__(self, fps: float = 15.0, max_queue: int = 120):
+        self._fps = fps
+        self._queue: Queue[Frame] = Queue(maxsize=max_queue)
+        self._running = True
+        self._frame_id = 0
+        self._lock = threading.Lock()
+
+    @property
+    def fps(self) -> float:
+        return self._fps
+
+    def reset(self):
+        with self._lock:
+            self._running = True
+            self._frame_id = 0
+            while not self._queue.empty():
+                try:
+                    self._queue.get_nowait()
+                    self._queue.task_done()
+                except Empty:
+                    break
+
+    def push_jpeg(self, data: bytes) -> tuple[bool, str]:
+        if not self._running:
+            return False, "phone source not running"
+
+        arr = np.frombuffer(data, dtype=np.uint8)
+        image = cv2.imdecode(arr, cv2.IMREAD_COLOR)
+        if image is None:
+            return False, "invalid image payload"
+
+        with self._lock:
+            frame = Frame(timestamp=time.time(), data=image, frame_id=self._frame_id)
+            self._frame_id += 1
+
+        if self._queue.full():
+            try:
+                self._queue.get_nowait()
+                self._queue.task_done()
+            except Empty:
+                pass
+
+        self._queue.put(frame)
+        return True, "frame pushed"
+
+    def stream(self):
+        while self._running:
+            try:
+                frame = self._queue.get(timeout=0.5)
+            except Empty:
+                continue
+            try:
+                yield frame
+            finally:
+                self._queue.task_done()
+
+    def stop(self):
+        self._running = False
+
+
+phone_source = PushFrameSource()
+
+
+def push_live_frame_bytes(data: bytes) -> tuple[bool, str]:
+    return phone_source.push_jpeg(data)
+
+
+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._source_history: list[dict[str, Any]] = []
+        self.running = False
+        self.starting = False
+        self.stopping = False
+        self.paused = False
+        self.last_error: str | None = None
+
+    def _record_source_locked(self, source_spec: str):
+        if (
+            self._source_history
+            and self._source_history[-1].get("source") == source_spec
+        ):
+            return
+        self._source_history.append(
+            {
+                "source": source_spec,
+                "timestamp": datetime.now().isoformat(timespec="seconds"),
+            }
+        )
+        if len(self._source_history) > 50:
+            self._source_history = self._source_history[-50:]
+
+    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
+            self._record_source_locked(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
+            self._record_source_locked(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 source_history(self, limit: int = 10) -> list[dict[str, Any]]:
+        with self._lock:
+            clamped = max(1, min(limit, 50))
+            items = self._source_history[-clamped:]
+            return list(reversed(items))
+
+    def switch_previous(self) -> tuple[bool, str]:
+        with self._lock:
+            current = self._source_spec
+            previous = None
+            for item in reversed(self._source_history):
+                source_spec = item.get("source")
+                if source_spec and source_spec != current:
+                    previous = source_spec
+                    break
+
+        if previous is None:
+            return False, "No previous source in history"
+
+        return self.switch_source(previous)
+
+    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,
+                "source_history": self.source_history(limit=10),
+                "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")