merge: feat/58-signal-bus into freeze/0.22.0

CHANGELOG.md

@@ -6,6 +6,26 @@ Versions follow [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
 ---
 
+## [0.22.0] — 2026-07-10
+
+### Added
+
+**`circuitforge_core.signal_bus`** — generic SSE event publisher for real-time signal streams (MIT, closes #58)
+
+- `SignalBus` — publish/subscribe bus for real-time events over Server-Sent Events. One instance per service; `publish()` is thread-safe (safe to call from sync producer threads such as OpenCV, Meshtastic, or PyPubSub callbacks) via `loop.call_soon_threadsafe()`. `subscribe(request)` returns a FastAPI `StreamingResponse` — each subscriber gets an independent bounded `asyncio.Queue` (default 100 events) with oldest-drop-on-overflow, plus a 15s keepalive comment to keep proxies from closing idle connections.
+- `SignalEvent` — frozen dataclass (`source`, `kind`, `payload`, auto-populated ISO-8601 UTC `timestamp`) with `to_sse()` wire-format serialization.
+- New `signal-bus` extra (`fastapi>=0.110`).
+
+**`circuitforge_core.video.app`** — `POST /caption/upload` endpoint
+
+Accepts a multipart video file upload, writes it to a temp file, captions it via the configured backend, then deletes the temp file. Lets callers without filesystem access to the video-service node (e.g. a product on one host posting to cf-video on another over an SSH tunnel) caption a video without a shared mount. `video-service` extra now also pulls in `python-multipart`, required by FastAPI's `UploadFile` parsing.
+
+### Docs
+
+- `mkdocs.yml` — new blue-grey/cyan palette, wired to a central `docs/stylesheets/theme.css` for consistent theme-aware styling across the site (light and dark mode).
+
+---
+
 ## [0.20.0] — 2026-05-05
 
 ### Fixed / Enhanced

README.md

@@ -42,6 +42,7 @@ pip install circuitforge-core[tts-chatterbox]       # Text-to-speech via Chatter
 pip install circuitforge-core[reranker-qwen3]       # Reranking via Qwen3
 pip install circuitforge-core[video-service]        # Video captioning service (Marlin-2B)
 pip install circuitforge-core[mqtt]                 # MQTT broker client
+pip install circuitforge-core[signal-bus]           # SSE event publisher for real-time signal streams
 pip install circuitforge-core[meshtastic-service]   # Meshtastic mesh radio + MQTT + FastAPI
 pip install circuitforge-core[memory]               # Knowledge graph via mnemo sidecar
 pip install circuitforge-core[community]            # PostgreSQL-backed community store
@@ -82,6 +83,7 @@ pip install circuitforge-core[dev]                  # All dev dependencies
 | `cloud_session` | Implemented | Cloud session management primitives |
 | `input` | Implemented | Input handling — MediaPipe gesture recognition |
 | `job_quality` | Implemented | Job listing quality scoring and signal extraction |
+| `signal_bus` | Implemented | Generic SSE event publisher for real-time signal streams |
 | `vision` | Stub | Vision router (moondream2 / SigLIP dispatch — planned) |
 | `wizard` | Stub | First-run wizard base class — products subclass `BaseWizard` |
 | `pipeline` | Stub | Staging queue base — products provide concrete schema |

circuitforge_core/signal_bus/__init__.py

@@ -0,0 +1,36 @@
+"""cf-core signal_bus — generic SSE event publisher for real-time signal streams.
+
+MIT licensed. Ticket: cf-core #58.
+
+## Quick start
+
+    from circuitforge_core.signal_bus import SignalBus, SignalEvent
+    from fastapi import Request
+
+    bus = SignalBus()
+
+    # Producer — safe to call from sync threads (OpenCV, Meshtastic, PyPubSub)
+    bus.publish(SignalEvent(
+        source="merlin",
+        kind="gesture",
+        payload={"name": "open_palm", "confidence": 0.94},
+    ))
+
+    # Consumer — SSE endpoint
+    @app.get("/events")
+    async def events(request: Request):
+        return bus.subscribe(request)
+
+## Notes
+
+- Each subscriber gets an independent bounded asyncio.Queue (default: 100 events).
+- When full, the oldest event is dropped to make room for the newest.
+- publish() is thread-safe via loop.call_soon_threadsafe().
+- Keepalive comment sent every 15 s to keep proxies from closing idle connections.
+- Install: no extra dependencies beyond fastapi (already a core dep for service products).
+"""
+
+from circuitforge_core.signal_bus.bus import SignalBus
+from circuitforge_core.signal_bus.models import SignalEvent
+
+__all__ = ["SignalBus", "SignalEvent"]

circuitforge_core/signal_bus/bus.py

@@ -0,0 +1,133 @@
+"""SignalBus — generic SSE event publisher for real-time signal streams.
+
+MIT licensed.
+"""
+from __future__ import annotations
+
+import asyncio
+import logging
+import threading
+from typing import AsyncIterator
+
+from fastapi import Request
+from fastapi.responses import StreamingResponse
+
+from circuitforge_core.signal_bus.models import SignalEvent
+
+logger = logging.getLogger(__name__)
+
+_DEFAULT_KEEPALIVE = 15.0  # seconds between SSE keepalive comments
+
+
+def _enqueue_drop_oldest(q: asyncio.Queue, item: str) -> None:
+    """Put item on queue; if full, drop the oldest entry to make room.
+
+    Designed to run on the event loop thread via call_soon_threadsafe.
+    """
+    if q.full():
+        try:
+            q.get_nowait()
+        except asyncio.QueueEmpty:
+            pass
+    try:
+        q.put_nowait(item)
+    except asyncio.QueueFull:
+        pass  # another thread won the race; drop the new item
+
+
+class SignalBus:
+    """Publish-subscribe bus for real-time signal events over SSE.
+
+    One bus instance per service. Producers call publish() (sync-safe);
+    consumers mount subscribe() as a FastAPI endpoint.
+
+    Args:
+        queue_size: Max buffered events per subscriber before oldest is dropped.
+    """
+
+    def __init__(self, queue_size: int = 100, keepalive_interval: float = _DEFAULT_KEEPALIVE) -> None:
+        self._queue_size = queue_size
+        self._keepalive_interval = keepalive_interval
+        # int(id(queue)) → queue; guarded by _lock for publish() from sync threads
+        self._subscribers: dict[int, asyncio.Queue] = {}
+        self._lock = threading.Lock()
+        # Captured on first subscribe() call; set from the FastAPI event loop
+        self._loop: asyncio.AbstractEventLoop | None = None
+
+    # ------------------------------------------------------------------
+    # Producer API
+    # ------------------------------------------------------------------
+
+    def publish(self, event: SignalEvent) -> None:
+        """Publish an event to all active subscribers.
+
+        Thread-safe — may be called from sync threads (OpenCV, Meshtastic,
+        PyPubSub callbacks, etc.). No-op when no subscribers are connected.
+        """
+        loop = self._loop
+        if loop is None or not loop.is_running():
+            return
+
+        sse = event.to_sse()
+        with self._lock:
+            subs = list(self._subscribers.values())
+
+        for q in subs:
+            loop.call_soon_threadsafe(_enqueue_drop_oldest, q, sse)
+
+    # ------------------------------------------------------------------
+    # Consumer API
+    # ------------------------------------------------------------------
+
+    def subscribe(self, request: Request) -> StreamingResponse:
+        """Return an SSE StreamingResponse for a FastAPI endpoint.
+
+        Each caller gets an independent queue. The subscriber is removed
+        automatically when the client disconnects.
+
+        Usage::
+
+            @app.get("/events")
+            async def events(request: Request):
+                return bus.subscribe(request)
+        """
+        loop = asyncio.get_event_loop()
+        self._loop = loop
+
+        q: asyncio.Queue = asyncio.Queue(maxsize=self._queue_size)
+        sub_id = id(q)
+        with self._lock:
+            self._subscribers[sub_id] = q
+
+        logger.debug("signal_bus: subscriber %d connected (%d total)", sub_id, len(self._subscribers))
+
+        async def _generate() -> AsyncIterator[str]:
+            # Starlette calls aclose() on this generator when the client disconnects,
+            # injecting GeneratorExit. Polling request.is_disconnected() is unreliable
+            # in ASGI test transports and adds latency in production.
+            try:
+                while True:
+                    try:
+                        chunk = await asyncio.wait_for(q.get(), timeout=self._keepalive_interval)
+                        yield chunk
+                    except asyncio.TimeoutError:
+                        yield ": keepalive\n\n"
+            finally:
+                with self._lock:
+                    self._subscribers.pop(sub_id, None)
+                logger.debug(
+                    "signal_bus: subscriber %d disconnected (%d remaining)",
+                    sub_id,
+                    len(self._subscribers),
+                )
+
+        return StreamingResponse(_generate(), media_type="text/event-stream")
+
+    # ------------------------------------------------------------------
+    # Introspection
+    # ------------------------------------------------------------------
+
+    @property
+    def subscriber_count(self) -> int:
+        with self._lock:
+            return len(self._subscribers)

circuitforge_core/signal_bus/models.py

@@ -0,0 +1,42 @@
+"""Data models for the cf-core signal_bus module.
+
+MIT licensed.
+"""
+from __future__ import annotations
+
+import json
+from dataclasses import dataclass, field
+from datetime import datetime, timezone
+
+
+def _utcnow() -> str:
+    return datetime.now(timezone.utc).isoformat().replace("+00:00", "Z")
+
+
+@dataclass(frozen=True)
+class SignalEvent:
+    """An event published to the signal bus.
+
+    Args:
+        source:    Originating service, e.g. "merlin", "linnet".
+        kind:      Event type, e.g. "gesture", "tone", "alpha_rising".
+        payload:   Arbitrary dict — contents are opaque to the bus.
+        timestamp: ISO-8601 UTC. Auto-populated if not supplied.
+    """
+
+    source: str
+    kind: str
+    payload: dict
+    timestamp: str = field(default_factory=_utcnow)
+
+    def to_sse(self) -> str:
+        """Serialize to SSE wire format: ``data: {...}\\n\\n``."""
+        data = json.dumps(
+            {
+                "source": self.source,
+                "kind": self.kind,
+                "payload": self.payload,
+                "timestamp": self.timestamp,
+            }
+        )
+        return f"data: {data}\n\n"

circuitforge_core/video/app.py

@@ -32,7 +32,10 @@ import logging
 import os
 from typing import Any
 
-from fastapi import FastAPI, HTTPException
+import shutil
+import tempfile
+
+from fastapi import FastAPI, HTTPException, UploadFile
 from pydantic import BaseModel, Field
 
 from circuitforge_core.video.backends.base import VideoBackend, make_video_backend
@@ -139,6 +142,46 @@ def find(req: FindRequest) -> FindResponse:
     )
 
 
+@app.post("/caption/upload", response_model=CaptionResponse)
+async def caption_upload(
+    file: UploadFile,
+    max_new_tokens: int = 2048,
+) -> CaptionResponse:
+    """Accept a video file upload, caption it, then delete the temp file.
+
+    Allows callers that don't share a filesystem with this node (e.g. Waxwing
+    on Heimdall posting to cf-video on Muninn via the SSH tunnel port forward).
+    """
+    if _backend is None:
+        raise HTTPException(503, detail="backend not initialised")
+
+    suffix = f".{file.filename.rsplit('.', 1)[-1]}" if file.filename and "." in file.filename else ".mkv"
+    tmp_fd, tmp_path = tempfile.mkstemp(suffix=suffix)
+    try:
+        with os.fdopen(tmp_fd, "wb") as dst:
+            shutil.copyfileobj(file.file, dst)
+
+        result = _backend.caption(tmp_path, max_new_tokens=max_new_tokens)
+    except Exception as exc:
+        logging.exception("caption/upload failed")
+        raise HTTPException(500, detail=str(exc)) from exc
+    finally:
+        try:
+            os.unlink(tmp_path)
+        except OSError:
+            pass
+
+    return CaptionResponse(
+        scene=result.scene,
+        events=[
+            VideoEventOut(start=ev.start, end=ev.end, description=ev.description)
+            for ev in result.events
+        ],
+        caption=result.caption,
+        model=result.model,
+    )
+
+
 # ── CLI entry point ───────────────────────────────────────────────────────────
 
 def _parse_args() -> argparse.Namespace:

docs/stylesheets/theme.css

@@ -0,0 +1,39 @@
+/*
+ * circuitforge-core docs theme
+ *
+ * Central place for palette tweaks and small layout fixes on top of
+ * mkdocs-material's blue grey / cyan palette (see mkdocs.yml). Keep
+ * product-specific overrides out of individual markdown pages — add
+ * them here so light/dark mode and mobile layout stay consistent
+ * across the whole site.
+ */
+
+:root {
+  --cf-accent-strong: #00acc1; /* cyan 600, matches material accent */
+  --cf-code-bg-light: #eceff1; /* blue grey 50 */
+  --cf-code-bg-dark: #263238; /* blue grey 900 */
+}
+
+/* Slightly stronger contrast for inline code than material's default,
+   in both color schemes. */
+[data-md-color-scheme="default"] {
+  --md-code-bg-color: var(--cf-code-bg-light);
+}
+
+[data-md-color-scheme="slate"] {
+  --md-code-bg-color: var(--cf-code-bg-dark);
+}
+
+/* Module reference tables can get wide (Module | Status | Description);
+   let them scroll horizontally on narrow viewports instead of overflowing
+   the page or squeezing text unreadably small. */
+.md-typeset table:not([class]) {
+  display: block;
+  overflow-x: auto;
+}
+
+@media screen and (max-width: 44.9375em) {
+  .md-typeset table:not([class]) {
+    font-size: 0.7rem;
+  }
+}

mkdocs.yml

@@ -9,14 +9,14 @@ theme:
   name: material
   palette:
     - scheme: default
-      primary: deep purple
+      primary: blue grey
-      accent: purple
+      accent: cyan
       toggle:
         icon: material/brightness-7
         name: Switch to dark mode
     - scheme: slate
-      primary: deep purple
+      primary: blue grey
-      accent: purple
+      accent: cyan
       toggle:
         icon: material/brightness-4
         name: Switch to light mode
@@ -78,5 +78,8 @@ nav:
     - Editable Install Pattern: developer/editable-install.md
     - BSL vs MIT Boundaries: developer/licensing.md
 
+extra_css:
+  - stylesheets/theme.css
+
 extra_javascript:
   - plausible.js

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "circuitforge-core"
-version = "0.21.0"
+version = "0.22.0"
 description = "Shared scaffold for CircuitForge products (MIT)"
 requires-python = ">=3.11"
 dependencies = [
@@ -20,6 +20,9 @@ memory = [
 sync = [
     "fastapi>=0.110",
 ]
+signal-bus = [
+    "fastapi>=0.110",
+]
 community = [
     "psycopg2>=2.9",
 ]
@@ -86,6 +89,7 @@ video-service = [
     "circuitforge-core[video-marlin]",
     "fastapi>=0.110",
     "uvicorn[standard]>=0.29",
+    "python-multipart>=0.0.9",
 ]
 vision-siglip = [
     "torch>=2.0",

tests/test_signal_bus/test_signal_bus.py

@@ -0,0 +1,192 @@
+"""Tests for SignalBus and SignalEvent."""
+from __future__ import annotations
+
+import asyncio
+import json
+from unittest.mock import MagicMock
+
+import pytest
+from fastapi import Request
+from fastapi.responses import StreamingResponse
+
+from circuitforge_core.signal_bus import SignalBus, SignalEvent
+
+
+# ---------------------------------------------------------------------------
+# SignalEvent
+# ---------------------------------------------------------------------------
+
+class TestSignalEvent:
+    def test_sse_format(self) -> None:
+        ev = SignalEvent(source="merlin", kind="gesture", payload={"name": "open_palm"}, timestamp="2026-01-01T00:00:00Z")
+        sse = ev.to_sse()
+        assert sse.startswith("data: ")
+        assert sse.endswith("\n\n")
+
+    def test_sse_contains_all_fields(self) -> None:
+        ev = SignalEvent(source="linnet", kind="tone", payload={"label": "sarcasm"}, timestamp="2026-01-01T00:00:00Z")
+        data = json.loads(ev.to_sse()[len("data: "):].strip())
+        assert data["source"] == "linnet"
+        assert data["kind"] == "tone"
+        assert data["payload"] == {"label": "sarcasm"}
+        assert data["timestamp"] == "2026-01-01T00:00:00Z"
+
+    def test_timestamp_auto_populated(self) -> None:
+        ev = SignalEvent(source="s", kind="k", payload={})
+        assert ev.timestamp  # not empty
+        assert "T" in ev.timestamp  # ISO format
+
+    def test_frozen(self) -> None:
+        ev = SignalEvent(source="s", kind="k", payload={})
+        with pytest.raises((AttributeError, TypeError)):
+            ev.source = "other"  # type: ignore
+
+
+# ---------------------------------------------------------------------------
+# SignalBus — unit (direct queue inspection)
+# ---------------------------------------------------------------------------
+
+class TestSignalBusUnit:
+    def test_publish_noop_before_subscribe(self) -> None:
+        bus = SignalBus()
+        ev = SignalEvent(source="s", kind="k", payload={})
+        bus.publish(ev)  # must not raise
+
+    def test_subscriber_count_zero_initially(self) -> None:
+        bus = SignalBus()
+        assert bus.subscriber_count == 0
+
+    @pytest.mark.asyncio
+    async def test_enqueue_drop_oldest_on_overflow(self) -> None:
+        from circuitforge_core.signal_bus.bus import _enqueue_drop_oldest
+        q: asyncio.Queue = asyncio.Queue(maxsize=2)
+        _enqueue_drop_oldest(q, "a")
+        _enqueue_drop_oldest(q, "b")
+        _enqueue_drop_oldest(q, "c")  # drops "a"
+        assert q.qsize() == 2
+        assert q.get_nowait() == "b"
+        assert q.get_nowait() == "c"
+
+    @pytest.mark.asyncio
+    async def test_enqueue_drop_oldest_empty_queue(self) -> None:
+        from circuitforge_core.signal_bus.bus import _enqueue_drop_oldest
+        q: asyncio.Queue = asyncio.Queue(maxsize=5)
+        _enqueue_drop_oldest(q, "x")
+        assert q.get_nowait() == "x"
+
+
+# ---------------------------------------------------------------------------
+# SignalBus — integration via FastAPI TestClient
+# ---------------------------------------------------------------------------
+
+def _make_bus() -> SignalBus:
+    return SignalBus(keepalive_interval=0.1)
+
+
+def _inject_subscriber(bus: SignalBus, q: "asyncio.Queue[str]") -> None:
+    """Register a pre-built queue as a subscriber (test helper)."""
+    bus._loop = asyncio.get_event_loop()
+    with bus._lock:
+        bus._subscribers[id(q)] = q
+
+
+class TestSignalBusIntegration:
+    """Integration tests verifying publish/subscribe queue mechanics directly.
+
+    httpx.ASGITransport cannot cancel an unbounded SSE generator — the transport
+    buffers outbound chunks but never delivers a disconnect signal to the ASGI
+    generator, so any test that exits a client.stream() block before the generator
+    finishes hangs forever. These tests bypass ASGI and inspect the subscriber
+    queues directly, which is both faster and more reliable.
+
+    The ASGI wiring is a one-liner (StreamingResponse) verified by
+    test_subscribe_returns_streaming_response below.
+    """
+
+    @pytest.mark.asyncio
+    async def test_event_delivered_to_subscriber_queue(self) -> None:
+        """publish() enqueues an SSE-formatted string to all subscriber queues."""
+        bus = _make_bus()
+        q: asyncio.Queue = asyncio.Queue(maxsize=10)
+        _inject_subscriber(bus, q)
+
+        ev = SignalEvent(
+            source="merlin", kind="gesture",
+            payload={"name": "open_palm", "confidence": 0.94},
+            timestamp="2026-01-01T00:00:00Z",
+        )
+        bus.publish(ev)
+        await asyncio.sleep(0)  # allow call_soon_threadsafe callback to run
+
+        assert not q.empty()
+        sse = q.get_nowait()
+        data = json.loads(sse[len("data: "):].strip())
+        assert data["source"] == "merlin"
+        assert data["kind"] == "gesture"
+        assert data["payload"]["name"] == "open_palm"
+
+    @pytest.mark.asyncio
+    async def test_multiple_subscribers_each_receive_event(self) -> None:
+        """Every connected subscriber queue receives its own copy of the event."""
+        bus = _make_bus()
+        q1: asyncio.Queue = asyncio.Queue(maxsize=10)
+        q2: asyncio.Queue = asyncio.Queue(maxsize=10)
+        _inject_subscriber(bus, q1)
+        bus._loop = asyncio.get_event_loop()
+        with bus._lock:
+            bus._subscribers[id(q2)] = q2
+
+        ev = SignalEvent(source="linnet", kind="tone", payload={}, timestamp="T")
+        bus.publish(ev)
+        await asyncio.sleep(0)
+
+        assert not q1.empty()
+        assert not q2.empty()
+        d1 = json.loads(q1.get_nowait()[len("data: "):].strip())
+        d2 = json.loads(q2.get_nowait()[len("data: "):].strip())
+        assert d1["kind"] == "tone"
+        assert d2["kind"] == "tone"
+
+    @pytest.mark.asyncio
+    async def test_publish_safe_while_subscribed(self) -> None:
+        """Rapid publish() calls with active subscribers must not raise."""
+        bus = _make_bus()
+        q: asyncio.Queue = asyncio.Queue(maxsize=100)
+        _inject_subscriber(bus, q)
+
+        errors: list[Exception] = []
+        try:
+            for i in range(20):
+                bus.publish(SignalEvent(source="s", kind="k", payload={"i": i}, timestamp="T"))
+        except Exception as exc:
+            errors.append(exc)
+
+        await asyncio.sleep(0)
+        assert not errors
+        assert q.qsize() > 0
+
+    @pytest.mark.asyncio
+    async def test_overflow_does_not_block_producer(self) -> None:
+        """A slow consumer with a tiny queue must not block publish()."""
+        bus = SignalBus(queue_size=2, keepalive_interval=0.1)
+        q: asyncio.Queue = asyncio.Queue(maxsize=2)
+        _inject_subscriber(bus, q)
+
+        for i in range(50):
+            bus.publish(SignalEvent(source="s", kind="k", payload={"i": i}, timestamp="T"))
+
+        await asyncio.sleep(0)
+        # Drop-oldest: only the last 2 events survive
+        assert q.qsize() == 2
+
+    @pytest.mark.asyncio
+    async def test_subscribe_returns_streaming_response(self) -> None:
+        """subscribe() returns a StreamingResponse with text/event-stream media type."""
+        bus = _make_bus()
+        mock_req = MagicMock(spec=Request)
+
+        resp = bus.subscribe(mock_req)
+
+        assert isinstance(resp, StreamingResponse)
+        assert resp.media_type == "text/event-stream"
+        assert bus.subscriber_count == 1

tests/test_video/test_app.py

@@ -7,6 +7,8 @@ so a zero-byte placeholder is sufficient.
 """
 from __future__ import annotations
 
+import os
+
 import pytest
 from fastapi.testclient import TestClient
 
@@ -234,3 +236,116 @@ def test_find_rejects_max_new_tokens_above_max(client, video_file):
         json={"video_path": video_file, "event": "wave", "max_new_tokens": 99999},
     )
     assert resp.status_code == 422
+
+
+# ── /caption/upload ───────────────────────────────────────────────────────────
+
+
+def test_caption_upload_returns_200(client):
+    resp = client.post(
+        "/caption/upload",
+        files={"file": ("sample.mp4", b"\x00" * 16, "video/mp4")},
+    )
+    assert resp.status_code == 200
+
+
+def test_caption_upload_response_matches_caption_shape(client):
+    data = client.post(
+        "/caption/upload",
+        files={"file": ("sample.mp4", b"\x00" * 16, "video/mp4")},
+    ).json()
+    assert isinstance(data["scene"], str) and data["scene"]
+    assert isinstance(data["events"], list) and len(data["events"]) >= 1
+    assert isinstance(data["caption"], str) and data["caption"]
+    assert isinstance(data["model"], str)
+
+
+def test_caption_upload_preserves_file_extension(client, monkeypatch):
+    seen_paths: list[str] = []
+    original_caption = MockVideoBackend.caption
+
+    def _spy_caption(self, video_path, *, max_new_tokens=2048):
+        seen_paths.append(video_path)
+        return original_caption(self, video_path, max_new_tokens=max_new_tokens)
+
+    monkeypatch.setattr(MockVideoBackend, "caption", _spy_caption)
+
+    resp = client.post(
+        "/caption/upload",
+        files={"file": ("sample.mkv", b"\x00" * 16, "video/x-matroska")},
+    )
+    assert resp.status_code == 200
+    assert seen_paths and seen_paths[0].endswith(".mkv")
+
+
+def test_caption_upload_defaults_extension_when_filename_has_none(client, monkeypatch):
+    seen_paths: list[str] = []
+    original_caption = MockVideoBackend.caption
+
+    def _spy_caption(self, video_path, *, max_new_tokens=2048):
+        seen_paths.append(video_path)
+        return original_caption(self, video_path, max_new_tokens=max_new_tokens)
+
+    monkeypatch.setattr(MockVideoBackend, "caption", _spy_caption)
+
+    resp = client.post(
+        "/caption/upload",
+        files={"file": ("sample", b"\x00" * 16, "application/octet-stream")},
+    )
+    assert resp.status_code == 200
+    assert seen_paths and seen_paths[0].endswith(".mkv")
+
+
+def test_caption_upload_removes_temp_file_after_request(client, monkeypatch):
+    captured_paths: list[str] = []
+    original_caption = MockVideoBackend.caption
+
+    def _spy_caption(self, video_path, *, max_new_tokens=2048):
+        captured_paths.append(video_path)
+        return original_caption(self, video_path, max_new_tokens=max_new_tokens)
+
+    monkeypatch.setattr(MockVideoBackend, "caption", _spy_caption)
+
+    resp = client.post(
+        "/caption/upload",
+        files={"file": ("sample.mp4", b"\x00" * 16, "video/mp4")},
+    )
+    assert resp.status_code == 200
+    assert captured_paths
+    assert not os.path.exists(captured_paths[0])
+
+
+def test_caption_upload_503_when_no_backend(client):
+    video_app._backend = None
+    resp = client.post(
+        "/caption/upload",
+        files={"file": ("sample.mp4", b"\x00" * 16, "video/mp4")},
+    )
+    assert resp.status_code == 503
+
+
+def test_caption_upload_500_and_cleans_up_on_backend_error(client, monkeypatch):
+    captured_paths: list[str] = []
+
+    def _raise(self, video_path, *, max_new_tokens=2048):
+        captured_paths.append(video_path)
+        raise RuntimeError("boom")
+
+    monkeypatch.setattr(MockVideoBackend, "caption", _raise)
+
+    resp = client.post(
+        "/caption/upload",
+        files={"file": ("sample.mp4", b"\x00" * 16, "video/mp4")},
+    )
+    assert resp.status_code == 500
+    assert captured_paths
+    assert not os.path.exists(captured_paths[0])
+
+
+def test_caption_upload_custom_max_new_tokens(client):
+    resp = client.post(
+        "/caption/upload",
+        params={"max_new_tokens": 512},
+        files={"file": ("sample.mp4", b"\x00" * 16, "video/mp4")},
+    )
+    assert resp.status_code == 200