bggpipe web: the whole pipeline as a local web app

A dashboard at / joins the review page (now at /review): drag-and-drop
photo upload (re-uploading a photo drops its raw cache so extract
re-reads it), per-stage status cards fed by /api/pipeline (counts and
key NAMES only — never values), and run buttons that execute stages
one-at-a-time in a background JobRunner with captured output streamed
to the page. The real upload sits behind a confirmation, defaults to
dry-run at the API layer, and stays disabled while stub data is
present. The CLI is unchanged and shares all state with the web UI.
python-multipart joins the deps for the upload endpoint; RunBody lives
at module scope because postponed annotations keep FastAPI from
resolving function-local models.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

src/bggpipe/jobs.py

@@ -0,0 +1,101 @@
+"""One-at-a-time background execution of pipeline stages for the web UI.
+
+The stages share the data/ artifacts, so running two concurrently is
+unsupported everywhere in the pipeline — the runner enforces it with a
+single slot. Stage output (typer.echo goes to stdout) is captured by
+swapping sys.stdout for the job's duration; that swap is process-wide,
+which is safe here only because the runner holds the single slot and the
+web server itself never writes to stdout (uvicorn logs on stderr).
+"""
+
+from __future__ import annotations
+
+import io
+import threading
+import time
+from collections.abc import Callable
+from contextlib import redirect_stdout
+
+import typer
+
+
+class _LineBuffer(io.TextIOBase):
+    def __init__(self) -> None:
+        self._lock = threading.Lock()
+        self._lines: list[str] = []
+        self._partial = ""
+
+    def write(self, text: str) -> int:
+        with self._lock:
+            self._partial += text
+            *complete, self._partial = self._partial.split("\n")
+            self._lines.extend(complete)
+        return len(text)
+
+    def lines(self) -> list[str]:
+        with self._lock:
+            return self._lines + ([self._partial] if self._partial else [])
+
+
+class JobRunner:
+    """At most one running job; finished state persists until the next start."""
+
+    def __init__(self) -> None:
+        self._lock = threading.Lock()
+        self._thread: threading.Thread | None = None
+        self._stage = ""
+        self._status = "idle"  # idle | running | done | failed
+        self._buffer = _LineBuffer()
+        self._error = ""
+        self._started = 0.0
+        self._finished = 0.0
+
+    def start(self, stage: str, fn: Callable[[], object]) -> bool:
+        """Begin a job; False when one is already running."""
+        with self._lock:
+            if self._status == "running":
+                return False
+            self._stage = stage
+            self._status = "running"
+            self._buffer = _LineBuffer()
+            self._error = ""
+            self._started = time.time()
+            self._finished = 0.0
+            self._thread = threading.Thread(target=self._run, args=(fn,), daemon=True)
+            self._thread.start()
+            return True
+
+    def _run(self, fn: Callable[[], object]) -> None:
+        status, error = "done", ""
+        try:
+            with redirect_stdout(self._buffer):
+                fn()
+        except typer.Exit as exc:
+            if exc.exit_code:
+                status, error = "failed", f"exited with code {exc.exit_code}"
+        except SystemExit as exc:
+            if exc.code:
+                status, error = "failed", f"exited with code {exc.code}"
+        except Exception as exc:  # surfaced in the UI, never swallowed
+            status, error = "failed", f"{type(exc).__name__}: {exc}"
+        with self._lock:
+            self._status = status
+            self._error = error
+            self._finished = time.time()
+
+    def snapshot(self) -> dict:
+        with self._lock:
+            return {
+                "stage": self._stage,
+                "status": self._status,
+                "log": self._buffer.lines()[-200:],
+                "error": self._error,
+                "started": self._started,
+                "finished": self._finished,
+            }
+
+    def wait(self, timeout: float = 10.0) -> None:
+        """Test hook: block until the current job's thread finishes."""
+        thread = self._thread
+        if thread is not None:
+            thread.join(timeout)