docs: update mkdocs configuration

# Conflicts:
#	CHANGELOG.md
#	README.md
#	pyproject.toml

CHANGELOG.md

@@ -6,6 +6,67 @@ 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).
+
+**`circuitforge_core.hardware.model_vram_estimate`** — model-to-hardware VRAM fit check (closes #64)
+
+Answers "can this hardware run model X at quantization level Y?" — the missing capability noted against `cf_core.hardware`, which detects available VRAM but had no way to cross-reference model requirements. Queries the HuggingFace Hub API for parameter count (`safetensors.total`) and architecture (`config.json`: `num_hidden_layers`, `hidden_size`, `num_attention_heads`, `num_key_value_heads`), then applies the standard formula: `vram_gb = params * bytes_per_param(quant) + kv_cache_gb(ctx_len, arch) + overhead_gb`. Reference algorithm from [LLMcalc](https://github.com/Raskoll2/LLMcalc) (unlicensed upstream — algorithm reference only, no code copied, no dependency added).
+
+- `model_vram_estimate(hf_model_id, quant_level, *, ctx_len=4096, available_vram_mb=None, overhead_gb=0.6, timeout=10.0) -> VramEstimate`
+- Supports common quant levels: `fp32`, `fp16`/`bf16`, `int8`/`q8`/`q8_0`, `q6_k`, `q5_k_m`/`q5_0`, `int4`/`q4`/`q4_k_m`/`q4_0`, `q3_k_m`, `q2_k`.
+- KV cache sizing accounts for GQA (`num_key_value_heads`); falls back to 0 GB when the model's `config.json` lacks standard architecture fields, rather than failing the whole estimate — weights dominate VRAM use regardless.
+- Raises `ModelVramLookupError` on HF Hub API failures or missing safetensors metadata; raises `ValueError` for unrecognized quant levels.
+- Application points noted in the ticket: Avocet preflight (verify a checkpoint fits before benchmarking), cf-orch worker assignment (match model to GPU by VRAM fit), Peregrine/Kiwi onboarding wizard ("your GPU has X GB — here are models that will run well").
+
+**`circuitforge_core.retry`** — standard retry/backoff wrapper over `backon` (closes #65)
+
+Standardizes retry/backoff behavior for cf-core modules and products that make external calls, replacing ad-hoc per-product retry loops. Wraps [backon](https://github.com/Llucs/backon) (MIT, zero stdlib dependencies) rather than reimplementing retry logic — chosen over `tenacity`/`backoff` per the eval in #65 for its native async support, built-in circuit breaker/hedging primitives, and a process-wide enable/disable toggle that's ideal for tests.
+
+- `on_exception(exception, *, max_tries=3, max_time=30.0, **backon_kwargs)` — decorator with CF's standard defaults (full-jitter exponential backoff); extra kwargs forward to `backon.on_exception` (e.g. `sleep=` for tests, `on_backoff=` for logging hooks).
+- `retry(target, *args, exception=Exception, max_tries=3, max_time=30.0, **kwargs)` — call an already-defined callable with the same retry behavior, for use without decorator syntax.
+- `disable_retries()` / `enable_retries()` — re-exported `backon` context managers for scoping retry-disable to a test or block.
+- `disable_retries_globally()` / `enable_retries_globally()` — re-exported `backon.disable()`/`enable()` process-wide toggles, for a test-suite-level fixture.
+- New core dependency: `backon>=4.0`.
+- Scope note: only the standalone wrapper module ships in this PR. Wiring it into `llm`, `affiliates`, `reranker`, and `activitypub` (the call sites identified in #65) is left as follow-up work per module, rather than retrofitting `LLMRouter`'s existing fallback-chain error handling in the same change.
+
+**`circuitforge_core.task_bridge`** — shared data contract for pushing tasks into an external scheduler (closes #66)
+
+Kiwi is the pilot consumer, pushing pantry expiry-alert tasks into Ashley Venn's Focus Flow scheduler (AGPL-3.0, external project). To keep AGPL and BSL code from ever sharing a process or artifact, the task data contract lives here in cf-core (MIT); each product implements its own exporter/receiver. Design spec: `circuitforge-plans/shared/superpowers/specs/2026-07-05-focus-flow-task-bridge-design.md`.
+
+- `models.py` — `ExternalTask` frozen dataclass: `schema_version`, `source_product`, `external_id`, `title`, `notes`, `due_at` (ISO 8601 UTC), `kind` (always `"flexible"` in v1 — external sources can never inject urgency into another product's UX), `status` (`"active"` / `"cancelled"`). Validates required fields and `kind`/`status` values in `__post_init__`.
+- `client.py` — `push_tasks(endpoint, token, tasks)`: thin `httpx` wrapper POSTing a batch as `{"tasks": [...]}` with a bearer token. Raises `TaskBridgeError` on transport failure or non-2xx response. No transport server, no auth/token generation — that lives on the receiving side.
+- New `task-bridge` extra (`httpx>=0.27`).
+- 19 tests: schema/validation/serialization (no network) plus a contract test against a real local `http.server` instance standing in for Focus Flow's importer, verifying the client emits a conformant payload.
+
+**`circuitforge_core.tasks.dispatch_task` / `get_task_status`** — generic caller/args task dispatch (closes #67)
+
+Pagepiper (and potentially other products copying its pattern) imported `dispatch_task(caller, args) -> task_id` / `get_task_status(task_id) -> dict` from `circuitforge_core.tasks`, expecting a `"product/task_name"` + kwargs-dict interface — neither function existed anywhere, so every call silently hit an `except Exception` fallback to local `BackgroundTasks`, with no visible error. This is a different API shape from the existing VRAM-budgeted `TaskScheduler` (keyed by `task_id`/`job_id`/`params` against a specific SQLite `background_tasks` table), so it's a new module rather than a `TaskScheduler` wrapper.
+
+- `register_task_runner(caller, fn)` — register a runnable under a name (e.g. `"pagepiper/ingest_pdf"`) once at product startup.
+- `dispatch_task(caller, args)` — runs the registered runnable as `fn(**args)` on a background thread, returns a `task_id` immediately. Raises `LookupError` if `caller` isn't registered — products with an `except Exception: ...` fallback (like pagepiper's `_dispatch_ingest`) keep working unchanged.
+- `get_task_status(task_id)` — returns `{"status": "queued"|"running"|"complete"|"error", "progress": int|None, "error": str|None}`. Raises `KeyError` for an unknown `task_id`.
+- `reset_dispatch_registry()` — test teardown only.
+- **Scope note:** this is the free-tier, in-process, single-node implementation — no cross-node distribution. Routing through the `circuitforge-orch` coordinator (BSL, separate package) would need a new generic task-dispatch endpoint on that coordinator, which doesn't exist today (`circuitforge_orch.client.CFOrchClient` only exposes model/service allocation, not a generic caller/args job queue); that's out of scope for this cf-core-only PR and tracked as follow-up. Consuming products (pagepiper) additionally need to call `register_task_runner()` at startup to benefit — not done here, since that's product-side work in a separate repo.
+- 10 tests: unregistered lookup, task_id uniqueness, args passed as kwargs, success/error status transitions, unregister, unknown task_id, status snapshot immutability.
+
+---
+
 ## [0.20.0] — 2026-05-05
 
 ### Fixed / Enhanced

README.md

@@ -42,8 +42,10 @@ 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[task-bridge]          # Push tasks into an external scheduler (e.g. Focus Flow)
 pip install circuitforge-core[community]            # PostgreSQL-backed community store
 pip install circuitforge-core[manage]               # cf-manage CLI (Typer)
 pip install circuitforge-core[dev]                  # All dev dependencies
@@ -63,7 +65,7 @@ pip install circuitforge-core[dev]                  # All dev dependencies
 | `documents` | Implemented | PDF, DOCX, and image OCR ingestion into `StructuredDocument` |
 | `affiliates` | Implemented | Affiliate URL wrapping with per-user opt-out and env-var fallback |
 | `preferences` | Implemented | User preference store — local YAML with pluggable backend; dot-path get/set |
-| `tasks` | Implemented | VRAM-aware LLM task scheduler; shared slot manager across services |
+| `tasks` | Implemented | VRAM-aware LLM task scheduler; shared slot manager across services; generic caller/args `dispatch_task`/`get_task_status` |
 | `manage` | Implemented | Cross-platform product process manager (Docker and native modes) |
 | `resources` | Implemented | VRAM allocation, eviction engine, GPU profile registry |
 | `text` | Implemented | Text utilities (normalize, chunk, truncate) + local LLM inference service (GGUF/transformers/VLM/classifier backends, multimodal content-block API) |
@@ -82,6 +84,9 @@ 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 |
+| `retry` | Implemented | Standard retry/backoff wrapper over `backon` for external-call modules |
+| `task_bridge` | Implemented | Shared data contract + push client for external task schedulers (e.g. Focus Flow) |
 | `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/hardware/__init__.py

@@ -16,6 +16,7 @@ from .detect import detect_hardware, detect_hardware_json
 from .generator import generate_profile
 from .models import HardwareSpec, LLMBackendConfig, LLMConfig
 from .tiers import VRAM_TIERS, VramTier, select_tier
+from .vram_estimate import ModelVramLookupError, VramEstimate, model_vram_estimate
 
 __all__ = [
     "detect_hardware",
@@ -27,4 +28,7 @@ __all__ = [
     "VRAM_TIERS",
     "VramTier",
     "select_tier",
+    "ModelVramLookupError",
+    "VramEstimate",
+    "model_vram_estimate",
 ]

circuitforge_core/hardware/vram_estimate.py

@@ -0,0 +1,185 @@
+# circuitforge_core/hardware/vram_estimate.py
+"""
+Model VRAM fit estimation — cf-core #64.
+
+`cf_core.hardware` can detect available VRAM but has no way to answer "can this
+hardware run model X at quantization level Y?". This module closes that gap by
+querying the HuggingFace Hub API for parameter count and architecture, then
+applying the standard VRAM estimation formula:
+
+    vram_gb = params * bytes_per_param(quant) + kv_cache_gb(ctx_len, arch) + overhead_gb
+
+The formula is the reference algorithm used by LLMcalc
+(https://github.com/Raskoll2/LLMcalc, no license — algorithm reference only,
+not a dependency). No LLMcalc code is copied here.
+"""
+from __future__ import annotations
+
+from dataclasses import dataclass
+
+import requests
+
+_HF_API_MODEL_URL = "https://huggingface.co/api/models/{model_id}"
+_HF_CONFIG_URL = "https://huggingface.co/{model_id}/resolve/main/config.json"
+
+_DEFAULT_OVERHEAD_GB = 0.6  # CUDA context + activation buffers, rough constant
+_KV_CACHE_DTYPE_BYTES = 2  # KV cache is stored fp16 in the common case
+
+# Bits per parameter for common quantization levels (weights-only, excludes KV cache).
+_QUANT_BITS: dict[str, float] = {
+    "fp32": 32.0,
+    "fp16": 16.0,
+    "bf16": 16.0,
+    "int8": 8.0,
+    "q8": 8.0,
+    "q8_0": 8.0,
+    "q6_k": 6.0,
+    "q5_k_m": 5.0,
+    "q5_0": 5.0,
+    "int4": 4.0,
+    "q4": 4.0,
+    "q4_k_m": 4.0,
+    "q4_0": 4.0,
+    "q3_k_m": 3.0,
+    "q2_k": 2.0,
+}
+
+
+class ModelVramLookupError(RuntimeError):
+    """Raised when the HuggingFace Hub API can't supply data needed to estimate VRAM."""
+
+
+@dataclass(frozen=True)
+class VramEstimate:
+    """Result of a `model_vram_estimate()` call."""
+
+    hf_model_id: str
+    quant_level: str
+    params_billions: float
+    weights_gb: float
+    kv_cache_gb: float
+    overhead_gb: float
+    total_vram_gb: float
+    fits: bool | None  # None when available_vram_mb wasn't supplied
+
+
+def _quant_bits(quant_level: str) -> float:
+    key = quant_level.strip().lower()
+    try:
+        return _QUANT_BITS[key]
+    except KeyError:
+        raise ValueError(
+            f"Unknown quant_level {quant_level!r}. Known levels: {sorted(_QUANT_BITS)}"
+        ) from None
+
+
+def _fetch_param_count(hf_model_id: str, *, timeout: float) -> float:
+    """Return total parameter count via the HF Hub API's safetensors metadata."""
+    url = _HF_API_MODEL_URL.format(model_id=hf_model_id)
+    try:
+        resp = requests.get(url, params={"expand": ["safetensors"]}, timeout=timeout)
+    except requests.RequestException as exc:
+        raise ModelVramLookupError(
+            f"HF Hub API request failed for {hf_model_id!r}: {exc}"
+        ) from exc
+
+    if resp.status_code != 200:
+        raise ModelVramLookupError(
+            f"HF Hub API returned {resp.status_code} for {hf_model_id!r}"
+        )
+
+    data = resp.json()
+    total = (data.get("safetensors") or {}).get("total")
+    if not total:
+        raise ModelVramLookupError(
+            f"{hf_model_id!r} has no safetensors parameter metadata on the HF Hub "
+            "(model may not publish safetensors weights)"
+        )
+    return float(total)
+
+
+def _fetch_arch_config(hf_model_id: str, *, timeout: float) -> dict:
+    """Return the model's config.json (architecture fields used for KV cache sizing)."""
+    url = _HF_CONFIG_URL.format(model_id=hf_model_id)
+    try:
+        resp = requests.get(url, timeout=timeout)
+    except requests.RequestException as exc:
+        raise ModelVramLookupError(
+            f"config.json fetch failed for {hf_model_id!r}: {exc}"
+        ) from exc
+
+    if resp.status_code != 200:
+        raise ModelVramLookupError(
+            f"config.json unavailable for {hf_model_id!r} (HTTP {resp.status_code})"
+        )
+    return resp.json()
+
+
+def _kv_cache_gb(config: dict, ctx_len: int) -> float:
+    """Estimate KV cache size in GB from architecture fields, 0.0 if unavailable."""
+    num_layers = config.get("num_hidden_layers")
+    hidden_size = config.get("hidden_size")
+    num_heads = config.get("num_attention_heads")
+    num_kv_heads = config.get("num_key_value_heads", num_heads)
+
+    if not (num_layers and hidden_size and num_heads):
+        # Non-standard config (missing architecture fields) — skip the KV
+        # estimate rather than fail the whole call. Weights dominate VRAM use.
+        return 0.0
+
+    head_dim = hidden_size / num_heads
+    bytes_total = 2 * num_layers * num_kv_heads * head_dim * ctx_len * _KV_CACHE_DTYPE_BYTES
+    return bytes_total / 1e9
+
+
+def model_vram_estimate(
+    hf_model_id: str,
+    quant_level: str,
+    *,
+    ctx_len: int = 4096,
+    available_vram_mb: int | None = None,
+    overhead_gb: float = _DEFAULT_OVERHEAD_GB,
+    timeout: float = 10.0,
+) -> VramEstimate:
+    """
+    Estimate VRAM required to run `hf_model_id` at `quant_level`.
+
+    Args:
+        hf_model_id: HuggingFace model repo ID, e.g. "Qwen/Qwen2.5-7B-Instruct".
+        quant_level: One of the known quant levels (see `_QUANT_BITS`), e.g. "q4_k_m".
+        ctx_len: Context length used for KV cache sizing.
+        available_vram_mb: If given, populates `VramEstimate.fits`.
+        overhead_gb: Fixed overhead for CUDA context / activation buffers.
+        timeout: Per-request timeout in seconds for HF Hub API calls.
+
+    Raises:
+        ModelVramLookupError: HF Hub API request failed or returned unusable data.
+        ValueError: `quant_level` isn't a recognized quantization level.
+    """
+    bits = _quant_bits(quant_level)
+    params = _fetch_param_count(hf_model_id, timeout=timeout)
+    weights_gb = (params * bits / 8) / 1e9
+
+    try:
+        config = _fetch_arch_config(hf_model_id, timeout=timeout)
+        kv_gb = _kv_cache_gb(config, ctx_len)
+    except ModelVramLookupError:
+        # Architecture lookup is best-effort — weights_gb alone is still useful.
+        kv_gb = 0.0
+
+    total_gb = weights_gb + kv_gb + overhead_gb
+
+    fits = None
+    if available_vram_mb is not None:
+        fits = total_gb <= (available_vram_mb / 1024)
+
+    return VramEstimate(
+        hf_model_id=hf_model_id,
+        quant_level=quant_level,
+        params_billions=params / 1e9,
+        weights_gb=weights_gb,
+        kv_cache_gb=kv_gb,
+        overhead_gb=overhead_gb,
+        total_vram_gb=total_gb,
+        fits=fits,
+    )

circuitforge_core/retry.py

@@ -0,0 +1,93 @@
+# circuitforge_core/retry.py
+"""
+circuitforge_core.retry — thin standard-retry wrapper over backon (cf-core #65).
+
+Standardizes retry/backoff behavior across cf-core modules and products that
+make external calls (LLM endpoints, affiliate links, reranker APIs, federated
+ActivityPub delivery). Wraps `backon` (MIT, zero stdlib dependencies) rather
+than reimplementing retry logic, and rather than each module hand-rolling its
+own ad-hoc retry loop.
+
+Typical usage::
+
+    from circuitforge_core.retry import on_exception
+    import requests
+
+    @on_exception(requests.RequestException, max_tries=3)
+    def fetch(url):
+        return requests.get(url, timeout=5)
+
+Tests should disable retries globally rather than waiting out real backoff
+delays::
+
+    from circuitforge_core.retry import disable_retries
+
+    def test_fetch_raises_immediately_on_failure():
+        with disable_retries():
+            with pytest.raises(requests.RequestException):
+                fetch("http://unreachable")
+"""
+from __future__ import annotations
+
+from collections.abc import Callable
+from typing import TypeVar
+
+import backon
+from backon import disable as disable_retries_globally
+from backon import disable_retries, enable_retries
+from backon import enable as enable_retries_globally
+
+T = TypeVar("T")
+
+__all__ = [
+    "on_exception",
+    "retry",
+    "disable_retries",
+    "enable_retries",
+    "disable_retries_globally",
+    "enable_retries_globally",
+]
+
+_DEFAULT_MAX_TRIES = 3
+_DEFAULT_MAX_TIME = 30.0
+
+
+def on_exception(
+    exception: type[Exception] | tuple[type[Exception], ...] = Exception,
+    *,
+    max_tries: int = _DEFAULT_MAX_TRIES,
+    max_time: float | None = _DEFAULT_MAX_TIME,
+    **backon_kwargs,
+) -> Callable[[Callable[..., T]], Callable[..., T]]:
+    """
+    Decorator: retry a function on `exception` using full-jitter exponential
+    backoff — CF's standard defaults for external API calls.
+
+    Extra keyword arguments are forwarded to `backon.on_exception` (e.g.
+    `on_backoff=` for a logging callback, `sleep=` to override the sleep
+    function in tests).
+    """
+    return backon.on_exception(
+        backon.expo(),
+        exception,
+        max_tries=max_tries,
+        max_time=max_time,
+        logger="circuitforge_core.retry",
+        **backon_kwargs,
+    )
+
+
+def retry(
+    target: Callable[..., T],
+    *args,
+    exception: type[Exception] | tuple[type[Exception], ...] = Exception,
+    max_tries: int = _DEFAULT_MAX_TRIES,
+    max_time: float | None = _DEFAULT_MAX_TIME,
+    **kwargs,
+) -> T:
+    """
+    Call `target(*args, **kwargs)` with CF's standard retry/backoff behavior,
+    for wrapping an already-defined callable without decorator syntax.
+    """
+    wrapped = on_exception(exception, max_tries=max_tries, max_time=max_time)(target)
+    return wrapped(*args, **kwargs)

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/task_bridge/__init__.py

@@ -0,0 +1,33 @@
+# circuitforge_core/task_bridge/__init__.py
+"""
+task_bridge — shared MIT data contract for pushing tasks from a CF product
+into an external scheduler (pilot consumer: Kiwi -> Focus Flow, AGPL-3.0).
+
+Pure data contract + reusable push helper. No transport server, no auth/token
+generation logic (that lives on the receiving side), no product-specific
+behavior — keeps AGPL and BSL code from ever sharing a process or artifact.
+
+Typical usage::
+
+    from circuitforge_core.task_bridge import ExternalTask, push_tasks
+
+    task = ExternalTask(
+        source_product="kiwi",
+        external_id="kiwi:item:1234",
+        title="Use up milk",
+        due_at="2026-07-07T00:00:00Z",
+        notes="Opened 2026-07-01",
+    )
+    push_tasks("http://127.0.0.1:8512/import/tasks", token, [task])
+
+Design spec: circuitforge-plans/shared/superpowers/specs/2026-07-05-focus-flow-task-bridge-design.md
+"""
+from .client import TaskBridgeError, push_tasks
+from .models import SCHEMA_VERSION, ExternalTask
+
+__all__ = [
+    "ExternalTask",
+    "SCHEMA_VERSION",
+    "TaskBridgeError",
+    "push_tasks",
+]

circuitforge_core/task_bridge/client.py

@@ -0,0 +1,53 @@
+"""Push client for the cf-core task_bridge module.
+
+MIT licensed. Thin httpx wrapper any CF product reuses to push a batch of
+ExternalTask records to a configured local HTTP endpoint. No transport
+server, no auth/token generation logic — that lives on the receiving side
+(e.g. Focus Flow's importer). See:
+circuitforge-plans/shared/superpowers/specs/2026-07-05-focus-flow-task-bridge-design.md
+"""
+from __future__ import annotations
+
+from collections.abc import Sequence
+
+import httpx
+
+from circuitforge_core.task_bridge.models import ExternalTask
+
+
+class TaskBridgeError(RuntimeError):
+    """Raised when pushing tasks to the configured endpoint fails."""
+
+
+def push_tasks(
+    endpoint: str,
+    token: str,
+    tasks: Sequence[ExternalTask],
+    *,
+    timeout: float = 10.0,
+) -> httpx.Response:
+    """
+    POST a batch of ExternalTask records to `endpoint` as `{"tasks": [...]}`,
+    authenticated with a bearer token.
+
+    Raises:
+        TaskBridgeError: the request failed to send, or the endpoint returned
+            a non-2xx status.
+    """
+    payload = {"tasks": [t.to_dict() for t in tasks]}
+    try:
+        resp = httpx.post(
+            endpoint,
+            json=payload,
+            headers={"Authorization": f"Bearer {token}"},
+            timeout=timeout,
+        )
+    except httpx.HTTPError as exc:
+        raise TaskBridgeError(f"task_bridge push to {endpoint!r} failed: {exc}") from exc
+
+    if resp.status_code >= 300:
+        raise TaskBridgeError(
+            f"task_bridge push to {endpoint!r} returned {resp.status_code}: {resp.text}"
+        )
+
+    return resp

circuitforge_core/task_bridge/models.py

@@ -0,0 +1,59 @@
+"""Data models for the cf-core task_bridge module.
+
+MIT licensed. Ticket: cf-core #66. Design spec:
+circuitforge-plans/shared/superpowers/specs/2026-07-05-focus-flow-task-bridge-design.md
+"""
+from __future__ import annotations
+
+from dataclasses import asdict, dataclass
+from typing import Any, Literal
+
+SCHEMA_VERSION = 1
+
+# v1 external tasks are always "flexible" — Focus Flow's calm task kind.
+# External sources never set "inflexible" / "critical" / "locked" / "surprise",
+# so no CF product can inject urgency into another product's UX.
+TaskKind = Literal["flexible"]
+TaskStatus = Literal["active", "cancelled"]
+
+_VALID_STATUSES: frozenset[str] = frozenset({"active", "cancelled"})
+
+
+@dataclass(frozen=True)
+class ExternalTask:
+    """
+    A task pushed from a CF product into an external scheduler (e.g. Focus Flow).
+
+    Pure data contract — no transport, no auth, no product-specific behavior.
+    `external_id` must be stable and idempotent per source item (e.g.
+    "kiwi:item:1234") so repeated pushes upsert rather than duplicate.
+    """
+
+    source_product: str
+    external_id: str
+    title: str
+    due_at: str  # ISO 8601 UTC, e.g. "2026-07-07T00:00:00Z"
+    notes: str | None = None
+    kind: TaskKind = "flexible"
+    status: TaskStatus = "active"
+    schema_version: int = SCHEMA_VERSION
+
+    def __post_init__(self) -> None:
+        if not self.source_product:
+            raise ValueError("source_product must not be empty")
+        if not self.external_id:
+            raise ValueError("external_id must not be empty")
+        if not self.title:
+            raise ValueError("title must not be empty")
+        if self.kind != "flexible":
+            raise ValueError(
+                f"kind must be 'flexible' in schema v{SCHEMA_VERSION}, got {self.kind!r}"
+            )
+        if self.status not in _VALID_STATUSES:
+            raise ValueError(
+                f"status must be one of {sorted(_VALID_STATUSES)}, got {self.status!r}"
+            )
+
+    def to_dict(self) -> dict[str, Any]:
+        """Serialize to the wire format expected by an importer's HTTP API."""
+        return asdict(self)

circuitforge_core/tasks/__init__.py

@@ -1,3 +1,10 @@
+from circuitforge_core.tasks.dispatch import (
+    dispatch_task,
+    get_task_status,
+    register_task_runner,
+    reset_dispatch_registry,
+    unregister_task_runner,
+)
 from circuitforge_core.tasks.scheduler import (
     TaskScheduler,
     LocalScheduler,
@@ -12,4 +19,9 @@ __all__ = [
     "detect_available_vram_gb",
     "get_scheduler",
     "reset_scheduler",
+    "dispatch_task",
+    "get_task_status",
+    "register_task_runner",
+    "unregister_task_runner",
+    "reset_dispatch_registry",
 ]

circuitforge_core/tasks/dispatch.py

@@ -0,0 +1,138 @@
+# circuitforge_core/tasks/dispatch.py
+"""
+Generic caller/args task dispatch — cf-core #67.
+
+Products (e.g. pagepiper) were already importing `dispatch_task`/
+`get_task_status` from `circuitforge_core.tasks` expecting a
+"product/task_name" + kwargs-dict interface, distinct from the VRAM-budgeted
+`TaskScheduler` in `scheduler.py` (which is keyed by task_id/job_id/params
+against a specific SQLite `background_tasks` table). Neither function
+existed, so every call silently hit the `except Exception` fallback.
+
+This is the free-tier (in-process, single-node) implementation: a runnable
+is registered under a name once at product startup, then dispatched by that
+name any number of times. There is no coordinator here — cross-node
+distribution via the separate circuitforge-orch package would need a new
+generic task-dispatch endpoint on that coordinator; this module doesn't
+attempt that, and is written so a future coordinator-backed implementation
+can slot in behind the same two function signatures.
+
+Usage::
+
+    from circuitforge_core.tasks import register_task_runner, dispatch_task, get_task_status
+
+    register_task_runner("pagepiper/ingest_pdf", run_ingest_pdf)
+
+    task_id = dispatch_task("pagepiper/ingest_pdf", {"doc_id": "...", "file_path": "..."})
+    get_task_status(task_id)  # {"status": "running", "progress": 0, "error": None}
+"""
+from __future__ import annotations
+
+import logging
+import threading
+from collections.abc import Callable
+from concurrent.futures import ThreadPoolExecutor
+from typing import Any
+from uuid import uuid4
+
+logger = logging.getLogger(__name__)
+
+_registry_lock = threading.Lock()
+_task_runners: dict[str, Callable[..., None]] = {}
+
+_status_lock = threading.Lock()
+_task_status: dict[str, dict[str, Any]] = {}
+
+_executor_lock = threading.Lock()
+_executor: ThreadPoolExecutor | None = None
+
+
+def _get_executor() -> ThreadPoolExecutor:
+    global _executor
+    with _executor_lock:
+        if _executor is None:
+            _executor = ThreadPoolExecutor(thread_name_prefix="cf-task-dispatch")
+        return _executor
+
+
+def register_task_runner(caller: str, fn: Callable[..., None]) -> None:
+    """
+    Register a runnable under `caller` (e.g. "pagepiper/ingest_pdf").
+
+    Must be called once at product startup before dispatch_task() is used
+    with the same `caller` string. `fn` is called as `fn(**args)`.
+    """
+    with _registry_lock:
+        _task_runners[caller] = fn
+
+
+def unregister_task_runner(caller: str) -> None:
+    """Remove a registration. TEST TEARDOWN ONLY."""
+    with _registry_lock:
+        _task_runners.pop(caller, None)
+
+
+def dispatch_task(caller: str, args: dict[str, Any]) -> str:
+    """
+    Dispatch the runnable registered under `caller` with `args` as kwargs,
+    running it on a background thread. Returns immediately with a task_id.
+
+    Raises LookupError if no runnable is registered for `caller` — callers
+    that assume this always routes through a coordinator (the pattern
+    pagepiper's `_dispatch_ingest` uses) should catch and fall back, exactly
+    as they already do for the ImportError this used to raise.
+    """
+    with _registry_lock:
+        fn = _task_runners.get(caller)
+    if fn is None:
+        raise LookupError(
+            f"No task runner registered for {caller!r}. "
+            "Call register_task_runner() before dispatch_task()."
+        )
+
+    task_id = str(uuid4())
+    with _status_lock:
+        _task_status[task_id] = {"status": "queued", "progress": 0, "error": None}
+
+    def _run() -> None:
+        with _status_lock:
+            _task_status[task_id] = {"status": "running", "progress": 0, "error": None}
+        try:
+            fn(**args)
+        except Exception as exc:
+            logger.exception("Task %s (%s) failed", task_id, caller)
+            with _status_lock:
+                _task_status[task_id] = {"status": "error", "progress": None, "error": str(exc)}
+        else:
+            with _status_lock:
+                _task_status[task_id] = {"status": "complete", "progress": 100, "error": None}
+
+    _get_executor().submit(_run)
+    return task_id
+
+
+def get_task_status(task_id: str) -> dict[str, Any]:
+    """
+    Return `{"status", "progress", "error"}` for `task_id`.
+
+    Raises KeyError if `task_id` is unknown — never dispatched, or the
+    process restarted since (status is in-memory only, not persisted).
+    """
+    with _status_lock:
+        status = _task_status.get(task_id)
+    if status is None:
+        raise KeyError(f"Unknown task_id: {task_id!r}")
+    return dict(status)
+
+
+def reset_dispatch_registry() -> None:
+    """Clear all registrations and status, and shut down the executor. TEST TEARDOWN ONLY."""
+    global _executor
+    with _registry_lock:
+        _task_runners.clear()
+    with _status_lock:
+        _task_status.clear()
+    with _executor_lock:
+        if _executor is not None:
+            _executor.shutdown(wait=True)
+            _executor = None

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/modules/hardware.md

@@ -49,3 +49,18 @@ Profile selection rules:
 ## HardwareProfile
 
 The `HardwareProfile` dataclass is written to `compose.override.yml` by `preflight.py` at product startup, making GPU capabilities available to Docker Compose without hardcoding.
+
+## Model VRAM fit estimation
+
+`model_vram_estimate()` answers "can this hardware run model X at quantization level Y?" by querying the HuggingFace Hub API for parameter count and architecture, then applying the standard VRAM formula (weights + KV cache + overhead). Reference algorithm from [LLMcalc](https://github.com/Raskoll2/LLMcalc) — no code copied, since LLMcalc has no published license.
+
+```python
+from circuitforge_core.hardware import model_vram_estimate
+
+est = model_vram_estimate("Qwen/Qwen2.5-7B-Instruct", "q4_k_m", available_vram_mb=8_000)
+print(est.total_vram_gb, est.fits)  # e.g. 5.2 True
+```
+
+Use cases: Avocet preflight (verify a checkpoint fits before benchmarking), cf-orch worker assignment (match model to GPU by VRAM fit), and onboarding wizards ("your GPU has X GB — here are models that will run well").
+
+Raises `ModelVramLookupError` if the HF Hub API request fails or the model has no safetensors metadata; raises `ValueError` for an unrecognized `quant_level`.

docs/modules/task_bridge.md

@@ -0,0 +1,44 @@
+# task_bridge
+
+Shared MIT data contract for pushing tasks from a CF product into an external scheduler. Pilot consumer: Kiwi's pantry expiry alerts pushed into [Focus Flow](https://git.opensourcesolarpunk.com/Circuit-Forge/focus-flow), Ashley Venn's ADHD-first scheduler (AGPL-3.0, external project).
+
+## Why this lives in cf-core
+
+Focus Flow is AGPL-3.0; CF's AI features are BSL 1.1. To avoid any question of AGPL reach extending into CF's licensed code, the two sides never share a process or artifact:
+
+- The **data contract** (this module) is MIT, a pure data definition with no product-specific logic.
+- Each **product's exporter/receiver** lives in that product's own repo, under its own licensing.
+- The two sides talk only over local HTTP, never a shared library or process.
+
+Design spec: `circuitforge-plans/shared/superpowers/specs/2026-07-05-focus-flow-task-bridge-design.md`.
+
+## Usage
+
+```python
+from circuitforge_core.task_bridge import ExternalTask, push_tasks
+
+task = ExternalTask(
+    source_product="kiwi",
+    external_id="kiwi:item:1234",   # stable, idempotent per source item — upserts, not duplicates
+    title="Use up milk",
+    due_at="2026-07-07T00:00:00Z",  # ISO 8601 UTC
+    notes="Opened 2026-07-01",
+)
+
+push_tasks("http://127.0.0.1:8512/import/tasks", token, [task])
+```
+
+`kind` is always `"flexible"` in schema v1 — Focus Flow's calm task kind. External sources can never set `"inflexible"`, `"critical"`, `"locked"`, or `"surprise"`, so no CF product can inject urgency into another product's UX.
+
+`status` is `"active"` (default) or `"cancelled"` — push a `"cancelled"` record when the source item is deleted or edited away before the external scheduler marks it complete.
+
+## API
+
+- `ExternalTask(source_product, external_id, title, due_at, notes=None, kind="flexible", status="active")` — frozen dataclass; validates required fields and `kind`/`status` in `__post_init__`. `.to_dict()` serializes to the wire format.
+- `push_tasks(endpoint, token, tasks, *, timeout=10.0)` — POSTs `{"tasks": [...]}` with `Authorization: Bearer <token>`. Raises `TaskBridgeError` on transport failure or non-2xx response.
+
+## What this module does *not* do
+
+No transport server, no auth/token generation or pairing logic, no product-specific behavior (diffing, callback handling, inventory updates). Those live entirely on the consuming product's side — see Kiwi's `focus_flow` service for the reference implementation.
+
+Install: `pip install circuitforge-core[task-bridge]` (pulls in `httpx`).

docs/modules/tasks.md

@@ -76,3 +76,22 @@ def get_scheduler():
 ```
 
 Always re-export `reset_scheduler` from the shim so the FastAPI lifespan can import it from one place.
+
+## Generic caller/args dispatch
+
+Separate from the VRAM-budgeted scheduler above: `dispatch_task()`/`get_task_status()` are a generic, product-agnostic pair for running a named callable in the background and polling its status, keyed by a `"product/task_name"` string rather than the scheduler's `task_id`/`job_id`/SQLite-table shape.
+
+```python
+from circuitforge_core.tasks import register_task_runner, dispatch_task, get_task_status
+
+# Once, at product startup:
+register_task_runner("pagepiper/ingest_pdf", run_ingest_pdf)
+
+# Anywhere a task needs dispatching:
+task_id = dispatch_task("pagepiper/ingest_pdf", {"doc_id": "...", "file_path": "..."})
+get_task_status(task_id)  # {"status": "running", "progress": 0, "error": None}
+```
+
+Free-tier, in-process, single-node — runs on a background thread pool, no cross-node distribution. `dispatch_task()` raises `LookupError` for an unregistered `caller`, so products with an `except Exception: ...` fallback to local execution keep working unchanged if they forget to register a runner.
+
+`register_task_runner`/`dispatch_task`/`get_task_status` route through `circuitforge_core.tasks.dispatch` internally — nothing shared with `scheduler.py`'s VRAM-aware queue.

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
-      accent: purple
+      primary: blue grey
+      accent: cyan
       toggle:
         icon: material/brightness-7
         name: Switch to dark mode
     - scheme: slate
-      primary: deep purple
-      accent: purple
+      primary: blue grey
+      accent: cyan
       toggle:
         icon: material/brightness-4
         name: Switch to light mode
@@ -71,6 +71,7 @@ nav:
     - stt: modules/stt.md
     - tts: modules/tts.md
     - pipeline: modules/pipeline.md
+    - task_bridge: modules/task_bridge.md
     - vision: modules/vision.md
     - wizard: modules/wizard.md
   - Developer Guide:
@@ -78,6 +79,10 @@ nav:
     - Editable Install Pattern: developer/editable-install.md
     - BSL vs MIT Boundaries: developer/licensing.md
     - "@circuitforge/display (Vue package)": developer/display-package.md
+  - All CF Docs: https://docs.circuitforge.tech
+
+extra_css:
+  - stylesheets/theme.css
 
 extra_javascript:
   - plausible.js

pyproject.toml

@@ -4,13 +4,14 @@ 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 = [
     "pyyaml>=6.0",
     "requests>=2.31",
     "openai>=1.0",
+    "backon>=4.0",
 ]
 
 [project.optional-dependencies]
@@ -20,6 +21,12 @@ memory = [
 sync = [
     "fastapi>=0.110",
 ]
+signal-bus = [
+    "fastapi>=0.110",
+]
+task-bridge = [
+    "httpx>=0.27",
+]
 community = [
     "psycopg2>=2.9",
 ]
@@ -86,6 +93,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_hardware/test_vram_estimate.py

@@ -0,0 +1,169 @@
+"""Tests for circuitforge_core.hardware.vram_estimate (cf-core #64)."""
+from __future__ import annotations
+
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from circuitforge_core.hardware.vram_estimate import (
+    ModelVramLookupError,
+    model_vram_estimate,
+)
+
+_QWEN_CONFIG = {
+    "num_hidden_layers": 28,
+    "hidden_size": 3584,
+    "num_attention_heads": 28,
+    "num_key_value_heads": 4,
+}
+
+
+def _api_response(status_code=200, safetensors_total=7_000_000_000):
+    resp = MagicMock()
+    resp.status_code = status_code
+    resp.json.return_value = {"safetensors": {"total": safetensors_total}} if safetensors_total else {}
+    return resp
+
+
+def _config_response(status_code=200, config=None):
+    resp = MagicMock()
+    resp.status_code = status_code
+    resp.json.return_value = config if config is not None else _QWEN_CONFIG
+    return resp
+
+
+class TestModelVramEstimate:
+    def test_estimates_weights_gb_for_fp16(self):
+        with patch(
+            "circuitforge_core.hardware.vram_estimate.requests.get",
+            side_effect=[_api_response(), _config_response()],
+        ):
+            est = model_vram_estimate("Qwen/Qwen2.5-7B-Instruct", "fp16")
+
+        assert est.params_billions == pytest.approx(7.0)
+        # 7e9 params * 16 bits / 8 bits-per-byte / 1e9 = 14 GB
+        assert est.weights_gb == pytest.approx(14.0)
+
+    def test_lower_bit_quant_uses_less_vram(self):
+        with patch(
+            "circuitforge_core.hardware.vram_estimate.requests.get",
+            side_effect=[_api_response(), _config_response()],
+        ):
+            fp16 = model_vram_estimate("Qwen/Qwen2.5-7B-Instruct", "fp16")
+        with patch(
+            "circuitforge_core.hardware.vram_estimate.requests.get",
+            side_effect=[_api_response(), _config_response()],
+        ):
+            q4 = model_vram_estimate("Qwen/Qwen2.5-7B-Instruct", "q4_k_m")
+
+        assert q4.weights_gb < fp16.weights_gb
+
+    def test_kv_cache_included_when_config_available(self):
+        with patch(
+            "circuitforge_core.hardware.vram_estimate.requests.get",
+            side_effect=[_api_response(), _config_response()],
+        ):
+            est = model_vram_estimate("Qwen/Qwen2.5-7B-Instruct", "fp16", ctx_len=8192)
+
+        assert est.kv_cache_gb > 0.0
+        assert est.total_vram_gb == pytest.approx(
+            est.weights_gb + est.kv_cache_gb + est.overhead_gb
+        )
+
+    def test_kv_cache_zero_when_config_missing_arch_fields(self):
+        with patch(
+            "circuitforge_core.hardware.vram_estimate.requests.get",
+            side_effect=[_api_response(), _config_response(config={"some_other_field": 1})],
+        ):
+            est = model_vram_estimate("weird/model", "fp16")
+
+        assert est.kv_cache_gb == 0.0
+
+    def test_kv_cache_zero_when_config_fetch_fails(self):
+        with patch(
+            "circuitforge_core.hardware.vram_estimate.requests.get",
+            side_effect=[_api_response(), _config_response(status_code=404)],
+        ):
+            est = model_vram_estimate("Qwen/Qwen2.5-7B-Instruct", "fp16")
+
+        assert est.kv_cache_gb == 0.0
+
+    def test_fits_true_when_vram_sufficient(self):
+        with patch(
+            "circuitforge_core.hardware.vram_estimate.requests.get",
+            side_effect=[_api_response(), _config_response()],
+        ):
+            est = model_vram_estimate(
+                "Qwen/Qwen2.5-7B-Instruct", "q4_k_m", available_vram_mb=24_000
+            )
+
+        assert est.fits is True
+
+    def test_fits_false_when_vram_insufficient(self):
+        with patch(
+            "circuitforge_core.hardware.vram_estimate.requests.get",
+            side_effect=[_api_response(), _config_response()],
+        ):
+            est = model_vram_estimate(
+                "Qwen/Qwen2.5-7B-Instruct", "fp32", available_vram_mb=4_000
+            )
+
+        assert est.fits is False
+
+    def test_fits_none_when_available_vram_not_supplied(self):
+        with patch(
+            "circuitforge_core.hardware.vram_estimate.requests.get",
+            side_effect=[_api_response(), _config_response()],
+        ):
+            est = model_vram_estimate("Qwen/Qwen2.5-7B-Instruct", "fp16")
+
+        assert est.fits is None
+
+    def test_unknown_quant_level_raises_value_error(self):
+        with pytest.raises(ValueError):
+            model_vram_estimate("Qwen/Qwen2.5-7B-Instruct", "not-a-real-quant")
+
+    def test_quant_level_case_insensitive(self):
+        with patch(
+            "circuitforge_core.hardware.vram_estimate.requests.get",
+            side_effect=[_api_response(), _config_response()],
+        ):
+            est = model_vram_estimate("Qwen/Qwen2.5-7B-Instruct", "FP16")
+
+        assert est.weights_gb == pytest.approx(14.0)
+
+    def test_raises_on_non_200_from_hf_api(self):
+        with patch(
+            "circuitforge_core.hardware.vram_estimate.requests.get",
+            return_value=_api_response(status_code=404),
+        ):
+            with pytest.raises(ModelVramLookupError):
+                model_vram_estimate("nonexistent/model", "fp16")
+
+    def test_raises_on_missing_safetensors_metadata(self):
+        with patch(
+            "circuitforge_core.hardware.vram_estimate.requests.get",
+            return_value=_api_response(safetensors_total=None),
+        ):
+            with pytest.raises(ModelVramLookupError):
+                model_vram_estimate("gguf-only/model", "fp16")
+
+    def test_raises_on_request_exception(self):
+        import requests
+
+        with patch(
+            "circuitforge_core.hardware.vram_estimate.requests.get",
+            side_effect=requests.ConnectionError("no network"),
+        ):
+            with pytest.raises(ModelVramLookupError):
+                model_vram_estimate("Qwen/Qwen2.5-7B-Instruct", "fp16")
+
+    def test_result_includes_hf_model_id_and_quant_level(self):
+        with patch(
+            "circuitforge_core.hardware.vram_estimate.requests.get",
+            side_effect=[_api_response(), _config_response()],
+        ):
+            est = model_vram_estimate("Qwen/Qwen2.5-7B-Instruct", "q4_k_m")
+
+        assert est.hf_model_id == "Qwen/Qwen2.5-7B-Instruct"
+        assert est.quant_level == "q4_k_m"

tests/test_retry/test_retry.py

@@ -0,0 +1,114 @@
+"""Tests for circuitforge_core.retry (cf-core #65)."""
+from __future__ import annotations
+
+from unittest.mock import patch
+
+import pytest
+
+from circuitforge_core.retry import (
+    disable_retries,
+    enable_retries_globally,
+    on_exception,
+    retry,
+)
+
+
+class FlakyError(Exception):
+    pass
+
+
+def _flaky_after(n_failures):
+    """Return a function that raises FlakyError n_failures times, then succeeds."""
+    state = {"calls": 0}
+
+    def fn():
+        state["calls"] += 1
+        if state["calls"] <= n_failures:
+            raise FlakyError(f"attempt {state['calls']}")
+        return state["calls"]
+
+    fn.state = state
+    return fn
+
+
+@pytest.fixture(autouse=True)
+def _ensure_retries_enabled():
+    """Some tests disable retries; always restore the global state after."""
+    yield
+    enable_retries_globally()
+
+
+class TestOnException:
+    def test_succeeds_without_retry_when_no_exception(self):
+        calls = {"n": 0}
+
+        @on_exception(FlakyError, max_tries=3)
+        def fn():
+            calls["n"] += 1
+            return "ok"
+
+        assert fn() == "ok"
+        assert calls["n"] == 1
+
+    def test_retries_until_success(self):
+        fn = _flaky_after(2)
+        wrapped = on_exception(FlakyError, max_tries=5, sleep=lambda s: None)(fn)
+
+        assert wrapped() == 3
+        assert fn.state["calls"] == 3
+
+    def test_gives_up_after_max_tries(self):
+        fn = _flaky_after(10)
+        wrapped = on_exception(FlakyError, max_tries=3, sleep=lambda s: None)(fn)
+
+        with pytest.raises(FlakyError):
+            wrapped()
+        assert fn.state["calls"] == 3
+
+    def test_only_retries_specified_exception_type(self):
+        @on_exception(FlakyError, max_tries=3, sleep=lambda s: None)
+        def fn():
+            raise ValueError("not a FlakyError")
+
+        with pytest.raises(ValueError):
+            fn()
+
+    def test_disable_retries_context_manager_calls_once(self):
+        fn = _flaky_after(10)
+        wrapped = on_exception(FlakyError, max_tries=5, sleep=lambda s: None)(fn)
+
+        with disable_retries():
+            with pytest.raises(FlakyError):
+                wrapped()
+
+        assert fn.state["calls"] == 1
+
+    def test_retries_resume_after_disable_retries_context_exits(self):
+        fn = _flaky_after(2)
+        wrapped = on_exception(FlakyError, max_tries=5, sleep=lambda s: None)(fn)
+
+        with disable_retries():
+            pass  # just exercise enter/exit without calling wrapped()
+
+        assert wrapped() == 3
+
+
+class TestRetry:
+    def test_retry_calls_target_with_args_and_kwargs(self):
+        def fn(a, b, *, c):
+            return a + b + c
+
+        assert retry(fn, 1, 2, c=3, max_tries=1) == 6
+
+    def test_retry_retries_on_failure(self):
+        fn = _flaky_after(2)
+
+        with patch("time.sleep"):
+            assert retry(fn, max_tries=5) == 3
+        assert fn.state["calls"] == 3
+
+    def test_retry_raises_after_exhausting_max_tries(self):
+        fn = _flaky_after(10)
+
+        with patch("time.sleep"), pytest.raises(FlakyError):
+            retry(fn, exception=FlakyError, max_tries=2)

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_task_bridge/test_client.py

@@ -0,0 +1,149 @@
+"""Tests for circuitforge_core.task_bridge.client (cf-core #66)."""
+from __future__ import annotations
+
+import json
+import threading
+from contextlib import contextmanager
+from http.server import BaseHTTPRequestHandler, HTTPServer
+
+import httpx
+import pytest
+
+from circuitforge_core.task_bridge.client import TaskBridgeError, push_tasks
+from circuitforge_core.task_bridge.models import ExternalTask
+
+
+def _task(**overrides) -> ExternalTask:
+    fields = {
+        "source_product": "kiwi",
+        "external_id": "kiwi:item:1234",
+        "title": "Use up milk",
+        "due_at": "2026-07-07T00:00:00Z",
+    }
+    fields.update(overrides)
+    return ExternalTask(**fields)
+
+
+@contextmanager
+def _mocked_httpx_post(handler):
+    """Patch httpx.post to route through an httpx.MockTransport handler."""
+    transport = httpx.MockTransport(handler)
+
+    def fake_post(url, *, json=None, headers=None, timeout=None):
+        with httpx.Client(transport=transport) as client:
+            return client.post(url, json=json, headers=headers, timeout=timeout)
+
+    import circuitforge_core.task_bridge.client as client_module
+
+    original = client_module.httpx.post
+    client_module.httpx.post = fake_post
+    try:
+        yield
+    finally:
+        client_module.httpx.post = original
+
+
+class TestPushTasksUnit:
+    def test_sends_conformant_payload(self):
+        captured = {}
+
+        def handler(request: httpx.Request) -> httpx.Response:
+            captured["body"] = json.loads(request.content)
+            captured["auth"] = request.headers.get("authorization")
+            return httpx.Response(200, json={"accepted": 1})
+
+        with _mocked_httpx_post(handler):
+            resp = push_tasks(
+                "http://127.0.0.1:9999/import/tasks", "tok123", [_task()]
+            )
+
+        assert resp.status_code == 200
+        assert captured["auth"] == "Bearer tok123"
+        assert captured["body"] == {"tasks": [_task().to_dict()]}
+
+    def test_pushes_multiple_tasks_in_one_batch(self):
+        captured = {}
+
+        def handler(request: httpx.Request) -> httpx.Response:
+            captured["body"] = json.loads(request.content)
+            return httpx.Response(200, json={"accepted": 2})
+
+        tasks = [_task(external_id="kiwi:item:1"), _task(external_id="kiwi:item:2")]
+        with _mocked_httpx_post(handler):
+            push_tasks("http://127.0.0.1:9999/import/tasks", "tok", tasks)
+
+        assert len(captured["body"]["tasks"]) == 2
+
+    def test_raises_on_non_2xx_response(self):
+        def handler(request: httpx.Request) -> httpx.Response:
+            return httpx.Response(401, text="bad token")
+
+        with _mocked_httpx_post(handler):
+            with pytest.raises(TaskBridgeError):
+                push_tasks("http://127.0.0.1:9999/import/tasks", "bad", [_task()])
+
+    def test_raises_on_transport_error(self):
+        def handler(request: httpx.Request) -> httpx.Response:
+            raise httpx.ConnectError("connection refused")
+
+        with _mocked_httpx_post(handler):
+            with pytest.raises(TaskBridgeError):
+                push_tasks("http://127.0.0.1:9999/import/tasks", "tok", [_task()])
+
+    def test_empty_task_list_sends_empty_batch(self):
+        captured = {}
+
+        def handler(request: httpx.Request) -> httpx.Response:
+            captured["body"] = json.loads(request.content)
+            return httpx.Response(200, json={"accepted": 0})
+
+        with _mocked_httpx_post(handler):
+            push_tasks("http://127.0.0.1:9999/import/tasks", "tok", [])
+
+        assert captured["body"] == {"tasks": []}
+
+
+class _FakeImporterHandler(BaseHTTPRequestHandler):
+    """A minimal stand-in for Focus Flow's import API."""
+
+    received: list[dict] = []
+
+    def do_POST(self):
+        length = int(self.headers["Content-Length"])
+        body = json.loads(self.rfile.read(length))
+        _FakeImporterHandler.received.append(
+            {"body": body, "auth": self.headers.get("Authorization")}
+        )
+        self.send_response(200)
+        self.send_header("Content-Type", "application/json")
+        self.end_headers()
+        self.wfile.write(json.dumps({"accepted": len(body["tasks"])}).encode())
+
+    def log_message(self, format, *args):
+        pass  # silence test output
+
+
+class TestPushTasksContract:
+    """Real local HTTP server standing in for Focus Flow's importer."""
+
+    def test_client_emits_conformant_payload_over_real_socket(self):
+        _FakeImporterHandler.received = []
+        server = HTTPServer(("127.0.0.1", 0), _FakeImporterHandler)
+        port = server.server_address[1]
+        thread = threading.Thread(target=server.handle_request, daemon=True)
+        thread.start()
+        try:
+            task = _task(notes="Opened 2026-07-01")
+            resp = push_tasks(
+                f"http://127.0.0.1:{port}/import/tasks", "pairing-token-abc", [task]
+            )
+            thread.join(timeout=5)
+
+            assert resp.status_code == 200
+            assert resp.json() == {"accepted": 1}
+            assert len(_FakeImporterHandler.received) == 1
+            received = _FakeImporterHandler.received[0]
+            assert received["auth"] == "Bearer pairing-token-abc"
+            assert received["body"] == {"tasks": [task.to_dict()]}
+        finally:
+            server.server_close()

tests/test_task_bridge/test_models.py

@@ -0,0 +1,83 @@
+"""Tests for circuitforge_core.task_bridge.models (cf-core #66). No network."""
+from __future__ import annotations
+
+import pytest
+
+from circuitforge_core.task_bridge.models import SCHEMA_VERSION, ExternalTask
+
+
+def _task(**overrides) -> ExternalTask:
+    fields = {
+        "source_product": "kiwi",
+        "external_id": "kiwi:item:1234",
+        "title": "Use up milk",
+        "due_at": "2026-07-07T00:00:00Z",
+    }
+    fields.update(overrides)
+    return ExternalTask(**fields)
+
+
+class TestExternalTaskDefaults:
+    def test_default_kind_is_flexible(self):
+        assert _task().kind == "flexible"
+
+    def test_default_status_is_active(self):
+        assert _task().status == "active"
+
+    def test_default_schema_version(self):
+        assert _task().schema_version == SCHEMA_VERSION == 1
+
+    def test_notes_defaults_to_none(self):
+        assert _task().notes is None
+
+    def test_frozen(self):
+        task = _task()
+        with pytest.raises((AttributeError, TypeError)):
+            task.title = "other"  # type: ignore
+
+
+class TestExternalTaskValidation:
+    def test_rejects_empty_source_product(self):
+        with pytest.raises(ValueError):
+            _task(source_product="")
+
+    def test_rejects_empty_external_id(self):
+        with pytest.raises(ValueError):
+            _task(external_id="")
+
+    def test_rejects_empty_title(self):
+        with pytest.raises(ValueError):
+            _task(title="")
+
+    def test_rejects_non_flexible_kind(self):
+        with pytest.raises(ValueError):
+            _task(kind="critical")
+
+    def test_rejects_unknown_status(self):
+        with pytest.raises(ValueError):
+            _task(status="done")
+
+    def test_accepts_cancelled_status(self):
+        assert _task(status="cancelled").status == "cancelled"
+
+
+class TestExternalTaskToDict:
+    def test_to_dict_contains_all_fields(self):
+        task = _task(notes="Opened 2026-07-01")
+        d = task.to_dict()
+        assert d == {
+            "source_product": "kiwi",
+            "external_id": "kiwi:item:1234",
+            "title": "Use up milk",
+            "due_at": "2026-07-07T00:00:00Z",
+            "notes": "Opened 2026-07-01",
+            "kind": "flexible",
+            "status": "active",
+            "schema_version": 1,
+        }
+
+    def test_to_dict_is_json_serializable(self):
+        import json
+
+        task = _task()
+        json.dumps(task.to_dict())  # must not raise

tests/test_tasks/test_dispatch.py

@@ -0,0 +1,116 @@
+"""Tests for circuitforge_core.tasks.dispatch (cf-core #67)."""
+from __future__ import annotations
+
+import threading
+import time
+
+import pytest
+
+from circuitforge_core.tasks.dispatch import (
+    dispatch_task,
+    get_task_status,
+    register_task_runner,
+    reset_dispatch_registry,
+    unregister_task_runner,
+)
+
+
+@pytest.fixture(autouse=True)
+def _clean_registry():
+    reset_dispatch_registry()
+    yield
+    reset_dispatch_registry()
+
+
+def _wait_for_status(task_id, target_statuses, timeout=2.0):
+    deadline = time.monotonic() + timeout
+    while time.monotonic() < deadline:
+        status = get_task_status(task_id)
+        if status["status"] in target_statuses:
+            return status
+        time.sleep(0.01)
+    raise AssertionError(f"task {task_id} did not reach {target_statuses} in time")
+
+
+class TestDispatchTask:
+    def test_raises_lookup_error_when_unregistered(self):
+        with pytest.raises(LookupError):
+            dispatch_task("pagepiper/ingest_pdf", {"doc_id": "1"})
+
+    def test_returns_a_task_id_string(self):
+        register_task_runner("test/noop", lambda **kwargs: None)
+        task_id = dispatch_task("test/noop", {})
+        assert isinstance(task_id, str) and task_id
+
+    def test_two_dispatches_get_different_task_ids(self):
+        register_task_runner("test/noop", lambda **kwargs: None)
+        id1 = dispatch_task("test/noop", {})
+        id2 = dispatch_task("test/noop", {})
+        assert id1 != id2
+
+    def test_task_runs_registered_fn_with_args_as_kwargs(self):
+        received = {}
+        event = threading.Event()
+
+        def runner(*, doc_id, file_path):
+            received["doc_id"] = doc_id
+            received["file_path"] = file_path
+            event.set()
+
+        register_task_runner("pagepiper/ingest_pdf", runner)
+        task_id = dispatch_task(
+            "pagepiper/ingest_pdf", {"doc_id": "abc", "file_path": "/tmp/x.pdf"}
+        )
+        assert event.wait(timeout=2.0)
+        assert received == {"doc_id": "abc", "file_path": "/tmp/x.pdf"}
+        _wait_for_status(task_id, {"complete"})
+
+    def test_task_status_reaches_complete_on_success(self):
+        register_task_runner("test/ok", lambda **kwargs: None)
+        task_id = dispatch_task("test/ok", {})
+        status = _wait_for_status(task_id, {"complete", "error"})
+        assert status == {"status": "complete", "progress": 100, "error": None}
+
+    def test_task_status_reaches_error_on_exception(self):
+        def failing(**kwargs):
+            raise ValueError("boom")
+
+        register_task_runner("test/fail", failing)
+        task_id = dispatch_task("test/fail", {})
+        status = _wait_for_status(task_id, {"complete", "error"})
+        assert status["status"] == "error"
+        assert status["error"] == "boom"
+
+    def test_unregister_removes_runner(self):
+        register_task_runner("test/temp", lambda **kwargs: None)
+        unregister_task_runner("test/temp")
+        with pytest.raises(LookupError):
+            dispatch_task("test/temp", {})
+
+
+class TestGetTaskStatus:
+    def test_raises_key_error_for_unknown_task_id(self):
+        with pytest.raises(KeyError):
+            get_task_status("not-a-real-task-id")
+
+    def test_status_immediately_after_dispatch_is_queued_or_running(self):
+        gate = threading.Event()
+
+        def slow(**kwargs):
+            gate.wait(timeout=2.0)
+
+        register_task_runner("test/slow", slow)
+        try:
+            task_id = dispatch_task("test/slow", {})
+            status = get_task_status(task_id)
+            assert status["status"] in ("queued", "running")
+        finally:
+            gate.set()
+
+    def test_returns_a_copy_not_the_internal_dict(self):
+        register_task_runner("test/ok", lambda **kwargs: None)
+        task_id = dispatch_task("test/ok", {})
+        _wait_for_status(task_id, {"complete"})
+        status = get_task_status(task_id)
+        status["status"] = "mutated"
+        assert get_task_status(task_id)["status"] == "complete"

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