feat(ui): queue position, elapsed time, filename, CPU-mode notice (#46)

Co-authored-by: Dirk Riemann <ditori@gmail.com>
Co-committed-by: Dirk Riemann <ditori@gmail.com>

AGENTS.md

@@ -4,7 +4,11 @@ Async, on-prem, LLM-powered structured information extraction microservice. Give
 
 Designed to be used by other on-prem services (e.g. mammon) as a reliable fallback / second opinion for format-specific deterministic parsers.
 
-Status: MVP deployed (2026-04-18) at `http://192.168.68.42:8994` — LAN only. Full reference spec at `docs/spec-core-pipeline.md`; MVP spec at `docs/superpowers/specs/2026-04-18-ix-mvp-design.md`; deploy runbook at `docs/deployment.md`.
+Status: MVP deployed (2026-04-18) at `http://192.168.68.42:8994` — LAN only. Browser UI at `http://192.168.68.42:8994/ui`. Full reference spec at `docs/spec-core-pipeline.md`; MVP spec at `docs/superpowers/specs/2026-04-18-ix-mvp-design.md`; deploy runbook at `docs/deployment.md`.
+
+Use cases: the built-in registry lives in `src/ix/use_cases/__init__.py` (`bank_statement_header` for MVP). Callers without a registered entry can ship an ad-hoc schema inline via `RequestIX.use_case_inline` (see README "Ad-hoc use cases"); the pipeline builds the Pydantic classes on the fly per request. The `/ui` page exposes this as a "custom" option so non-engineering users can experiment without a deploy.
+
+UX notes: the `/ui` job page surfaces queue position + elapsed MM:SS on each poll, renders the client-provided filename (stored via `FileRef.display_name`, optional metadata — the pipeline ignores it for execution), and shows a CPU-mode notice when `/healthz` reports `ocr_gpu: false`.
 
 ## Guiding Principles
 

README.md

@@ -4,7 +4,11 @@ Async, on-prem, LLM-powered structured information extraction microservice.
 
 Given a document (PDF, image, text) and a named *use case*, ix returns a structured JSON result whose shape matches the use-case schema — together with per-field provenance (OCR segment IDs, bounding boxes, cross-OCR agreement flags) that let the caller decide how much to trust each extracted value.
 
-**Status:** MVP deployed. Live on the home LAN at `http://192.168.68.42:8994`.
+**Status:** MVP deployed. Live on the home LAN at `http://192.168.68.42:8994` (REST API + browser UI at `/ui`).
+
+## Web UI
+
+A minimal browser UI lives at [`http://192.168.68.42:8994/ui`](http://192.168.68.42:8994/ui): drop a PDF, pick a registered use case or define one inline, submit, see the pretty-printed result. HTMX polls the job status every 2 s until the pipeline finishes. LAN-only, no auth.
 
 - Full reference spec: [`docs/spec-core-pipeline.md`](docs/spec-core-pipeline.md) (aspirational; MVP is a strict subset)
 - **MVP design:** [`docs/superpowers/specs/2026-04-18-ix-mvp-design.md`](docs/superpowers/specs/2026-04-18-ix-mvp-design.md)
@@ -40,6 +44,28 @@ curl -X POST http://192.168.68.42:8994/jobs \
 
 Poll `GET /jobs/{job_id}` until `status` is `done` or `error`. Optionally pass `callback_url` to receive a webhook on completion (one-shot, no retry; polling stays authoritative).
 
+### Ad-hoc use cases
+
+For one-offs where a registered use case doesn't exist yet, ship the schema inline:
+
+```jsonc
+{
+  "use_case": "adhoc-invoice",        // free-form label (logs/metrics only)
+  "use_case_inline": {
+    "use_case_name": "Invoice totals",
+    "system_prompt": "Extract vendor and total amount.",
+    "fields": [
+      {"name": "vendor", "type": "str", "required": true},
+      {"name": "total",  "type": "decimal"},
+      {"name": "currency", "type": "str", "choices": ["USD", "EUR", "CHF"]}
+    ]
+  },
+  // ...ix_client_id, request_id, context...
+}
+```
+
+When `use_case_inline` is set, the pipeline builds the response schema on the fly and skips the registry. Supported types: `str`, `int`, `float`, `decimal`, `date`, `datetime`, `bool`. `choices` is only allowed on `str` fields. Precedence: inline wins over `use_case` when both are present.
+
 Full REST surface + provenance response shape documented in the MVP design spec.
 
 ## Running locally
@@ -50,6 +76,15 @@ uv run pytest tests/unit -v                    # hermetic unit + integration sui
 IX_TEST_OLLAMA=1 uv run pytest tests/live -v    # needs LAN access to Ollama + GPU
 ```
 
+### UI queue + progress UX
+
+The `/ui` job page polls `GET /ui/jobs/{id}/fragment` every 2 s and surfaces:
+
+- **Queue position** while pending: "Queue position: N ahead — M jobs total in flight (single worker)" so it's obvious a new submission is waiting on an earlier job rather than stuck. "About to start" when the worker has just freed up.
+- **Elapsed time** while running ("Running for MM:SS") and on finish ("Finished in MM:SS").
+- **Original filename** — the UI stashes the client-provided upload name in `FileRef.display_name` so the browser shows `your_statement.pdf` instead of the on-disk UUID.
+- **CPU-mode notice** when `/healthz` reports `ocr_gpu: false` (the Surya OCR client observed `torch.cuda.is_available() == False`): a collapsed `<details>` pointing at the deployment runbook.
+
 ## Deploying
 
 ```bash

docker-compose.yml

@@ -10,6 +10,8 @@
 # The GPU reservation block matches immich-ml / the shape Docker Compose
 # expects for GPU allocation on this host.
 
+name: infoxtractor
+
 services:
   infoxtractor:
     build: .

docs/superpowers/specs/2026-04-18-ix-mvp-design.md

@@ -85,6 +85,7 @@ class FileRef(BaseModel):
     url: str                                 # http(s):// or file://
     headers: dict[str, str] = {}             # e.g. {"Authorization": "Token …"}
     max_bytes: Optional[int] = None          # per-file override; defaults to IX_FILE_MAX_BYTES
+    display_name: Optional[str] = None       # UI-only metadata; client-provided filename for display (pipeline ignores)
 
 class Options(BaseModel):
     ocr: OCROptions = OCROptions()
@@ -108,6 +109,25 @@ class ProvenanceOptions(BaseModel):
 
 **Dropped from spec (no-ops under MVP):** `OCROptions.computer_vision_scaling_factor`, `include_page_tags` (always on), `GenAIOptions.use_vision`/`vision_scaling_factor`/`vision_detail`/`reasoning_effort`, `ProvenanceOptions.granularity`/`include_bounding_boxes`/`source_type`/`min_confidence`, `RequestIX.version`.
 
+**Ad-hoc use cases (post-MVP add-on).** `RequestIX` carries an optional `use_case_inline: InlineUseCase | None = None`. When set, the pipeline builds the `(Request, Response)` Pydantic class pair on the fly from that inline definition and **skips the registry lookup entirely** — the `use_case` field becomes a free-form label (still required for metrics / logging). Inline definitions look like:
+
+```python
+class UseCaseFieldDef(BaseModel):
+    name: str                               # valid Python identifier
+    type: Literal["str", "int", "float", "decimal", "date", "datetime", "bool"]
+    required: bool = False
+    description: str | None = None
+    choices: list[str] | None = None        # str-typed fields only; builds Literal[*choices]
+
+class InlineUseCase(BaseModel):
+    use_case_name: str
+    system_prompt: str
+    default_model: str | None = None
+    fields: list[UseCaseFieldDef]
+```
+
+Precedence: `use_case_inline` wins when both are set. Structural errors (dup field name, invalid identifier, `choices` on a non-str type, empty fields list) raise `IX_001_001` (same code as registry miss). The builder lives in `ix.use_cases.inline.build_use_case_classes` and returns fresh classes per call — the pipeline never caches them.
+
 ### ResponseIX
 
 Identical to spec §2.2 except `FieldProvenance` gains two fields:
@@ -206,14 +226,15 @@ Callers that prefer direct SQL (the `pg_queue_adapter` contract): insert a row w
 | `POST` | `/jobs` | Body = `RequestIX` (+ optional `callback_url`). → `201 {job_id, ix_id, status: "pending"}`. Idempotent on `(ix_client_id, request_id)` — same pair returns the existing `job_id` with `200`. |
 | `GET`  | `/jobs/{job_id}` | → full `Job`. Source of truth regardless of submission path or callback outcome. |
 | `GET`  | `/jobs?client_id=…&request_id=…` | Lookup-by-correlation (caller idempotency helper). The pair is UNIQUE in the table → at most one match. Returns the job or `404`. |
-| `GET`  | `/healthz` | `{postgres, ollama, ocr}`. See below for semantics. Used by `infrastructure` monitoring dashboard. |
+| `GET`  | `/healthz` | `{postgres, ollama, ocr, ocr_gpu}`. See below for semantics. Used by `infrastructure` monitoring dashboard. `ocr_gpu` is additive metadata (not part of the gate). |
 | `GET`  | `/metrics` | Counters over the last 24 hours: `jobs_pending`, `jobs_running`, `jobs_done_24h`, `jobs_error_24h`, per-use-case avg seconds over the same window. Plain JSON, no Prometheus format for MVP. |
 
 **`/healthz` semantics:**
 - `postgres`: `SELECT 1` on the job store pool; `ok` iff the query returns within 2 s.
 - `ollama`: `GET {IX_OLLAMA_URL}/api/tags` within 5 s; `ok` iff reachable AND the default model (`IX_DEFAULT_MODEL`) is listed in the tags response; `degraded` iff reachable but the model is missing (ops action: run `ollama pull <model>` on the host); `fail` on any other error.
 - `ocr`: `SuryaOCRClient.selfcheck()` — returns `ok` iff CUDA is available and the Surya text-recognition model is loaded into GPU memory at process start. `fail` on any error.
-- Overall HTTP status: `200` iff all three are `ok`; `503` otherwise. The monitoring dashboard only surfaces `200`/`non-200`.
+- `ocr_gpu`: `true | false | null`. Additive metadata: reports whether the OCR client observed `torch.cuda.is_available() == True` at first warm-up. `null` means not yet probed (fresh process, fake client, etc.). The UI reads this to surface a CPU-mode slowdown notice; never part of the 200/503 gate.
+- Overall HTTP status: `200` iff all three core statuses (`postgres`, `ollama`, `ocr`) are `ok`; `503` otherwise. `ocr_gpu` does not affect the gate. The monitoring dashboard only surfaces `200`/`non-200`.
 
 **Callback delivery** (when `callback_url` is set): one POST of the full `Job` body, 10 s timeout. 2xx → `callback_status='delivered'`. Anything else → `'failed'`. No retry. Callers always have `GET /jobs/{id}` as the authoritative fallback.
 

pyproject.toml

@@ -29,6 +29,14 @@ dependencies = [
     "pillow>=10.2,<11.0",
     "python-magic>=0.4.27",
     "python-dateutil>=2.9",
+
+    # UI (HTMX + Jinja2 templates served from /ui). Both arrive as transitive
+    # deps via FastAPI/Starlette already, but we pin explicitly so the import
+    # surface is owned by us. python-multipart backs FastAPI's `Form()` /
+    # `UploadFile` parsing — required by `/ui/jobs` submissions.
+    "jinja2>=3.1",
+    "aiofiles>=24.1",
+    "python-multipart>=0.0.12",
 ]
 
 [project.optional-dependencies]

src/ix/adapters/rest/routes.py

@@ -18,7 +18,7 @@ from __future__ import annotations
 
 import asyncio
 from collections.abc import Callable
-from dataclasses import dataclass
+from dataclasses import dataclass, field
 from datetime import UTC, datetime, timedelta
 from typing import Annotated, Literal
 from uuid import UUID
@@ -44,10 +44,15 @@ class Probes:
     keeping them sync lets tests pass plain lambdas. Real probes that need
     async work run the call through ``asyncio.run_in_executor`` inside the
     callable (Chunk 4).
+
+    ``ocr_gpu`` is additive metadata for the UI (not a health gate): returns
+    ``True`` iff the OCR client reports CUDA is available, ``False`` for
+    explicit CPU-mode, ``None`` if unknown (fake client, not yet warmed up).
     """
 
     ollama: Callable[[], Literal["ok", "degraded", "fail"]]
     ocr: Callable[[], Literal["ok", "fail"]]
+    ocr_gpu: Callable[[], bool | None] = field(default=lambda: None)
 
 
 def get_session_factory_dep() -> async_sessionmaker[AsyncSession]:
@@ -163,8 +168,16 @@ async def healthz(
     except Exception:
         ocr_state = "fail"
 
+    try:
+        ocr_gpu_state: bool | None = probes.ocr_gpu()
+    except Exception:
+        ocr_gpu_state = None
+
     body = HealthStatus(
-        postgres=postgres_state, ollama=ollama_state, ocr=ocr_state
+        postgres=postgres_state,
+        ollama=ollama_state,
+        ocr=ocr_state,
+        ocr_gpu=ocr_gpu_state,
     )
     if postgres_state != "ok" or ollama_state != "ok" or ocr_state != "ok":
         response.status_code = 503

src/ix/adapters/rest/schemas.py

@@ -28,9 +28,15 @@ class HealthStatus(BaseModel):
     """Body of GET /healthz.
 
     Each field reports per-subsystem state. Overall HTTP status is 200 iff
-    every field is ``"ok"`` (spec §5). ``ollama`` can be ``"degraded"``
+    every of the three core status keys is ``"ok"`` (spec §5). ``ollama`` can
-    when the backend is reachable but the default model isn't pulled —
+    be ``"degraded"`` when the backend is reachable but the default model
-    monitoring surfaces that as non-200.
+    isn't pulled — monitoring surfaces that as non-200.
+
+    ``ocr_gpu`` is additive metadata, not part of the health gate: it reports
+    whether the Surya OCR client observed ``torch.cuda.is_available() == True``
+    on first warm-up. ``None`` means we haven't probed yet (fresh process,
+    fake client, or warm_up hasn't happened). The UI reads this to surface a
+    CPU-mode slowdown warning to users.
     """
 
     model_config = ConfigDict(extra="forbid")
@@ -38,6 +44,7 @@ class HealthStatus(BaseModel):
     postgres: Literal["ok", "fail"]
     ollama: Literal["ok", "degraded", "fail"]
     ocr: Literal["ok", "fail"]
+    ocr_gpu: bool | None = None
 
 
 class MetricsResponse(BaseModel):

src/ix/app.py

@@ -24,6 +24,7 @@ from contextlib import asynccontextmanager, suppress
 from typing import Literal
 
 from fastapi import FastAPI
+from fastapi.staticfiles import StaticFiles
 
 from ix.adapters.rest.routes import Probes, get_probes
 from ix.adapters.rest.routes import router as rest_router
@@ -38,6 +39,8 @@ from ix.pipeline.pipeline import Pipeline
 from ix.pipeline.reliability_step import ReliabilityStep
 from ix.pipeline.response_handler_step import ResponseHandlerStep
 from ix.pipeline.setup_step import SetupStep
+from ix.ui import build_router as build_ui_router
+from ix.ui.routes import STATIC_DIR as UI_STATIC_DIR
 
 
 def build_pipeline(
@@ -105,6 +108,20 @@ def _make_ocr_probe(ocr: OCRClient) -> Callable[[], Literal["ok", "fail"]]:
     return probe
 
 
+def _make_ocr_gpu_probe(ocr: OCRClient) -> Callable[[], bool | None]:
+    """Adapter: read the OCR client's recorded ``gpu_available`` attribute.
+
+    The attribute is set by :meth:`SuryaOCRClient.warm_up` on first load.
+    Returns ``None`` when the client has no such attribute (e.g. FakeOCRClient
+    in test mode) or warm_up hasn't happened yet. Never raises.
+    """
+
+    def probe() -> bool | None:
+        return getattr(ocr, "gpu_available", None)
+
+    return probe
+
+
 def _run_async_sync(make_coro, *, fallback: str) -> str:  # type: ignore[no-untyped-def]
     """Run ``make_coro()`` on a fresh loop in a thread; return its result.
 
@@ -164,6 +181,7 @@ def create_app(*, spawn_worker: bool = True) -> FastAPI:
             lambda: Probes(
                 ollama=_make_ollama_probe(genai_client, cfg),
                 ocr=_make_ocr_probe(ocr_client),
+                ocr_gpu=_make_ocr_gpu_probe(ocr_client),
             ),
         )
 
@@ -202,6 +220,16 @@ def create_app(*, spawn_worker: bool = True) -> FastAPI:
 
     app = FastAPI(lifespan=lifespan, title="infoxtractor", version="0.1.0")
     app.include_router(rest_router)
+    # Browser UI — additive, never touches the REST paths above.
+    app.include_router(build_ui_router())
+    # Static assets for the UI. CDN-only for MVP so the directory is
+    # essentially empty, but the mount must exist so relative asset
+    # URLs resolve cleanly.
+    app.mount(
+        "/ui/static",
+        StaticFiles(directory=str(UI_STATIC_DIR)),
+        name="ui-static",
+    )
     return app
 
 

src/ix/contracts/request.py

@@ -22,6 +22,11 @@ class FileRef(BaseModel):
     Used when the file URL needs authentication (e.g. Paperless ``Token``) or a
     tighter size cap than :envvar:`IX_FILE_MAX_BYTES`. Plain URLs that need no
     headers can stay as bare ``str`` values in :attr:`Context.files`.
+
+    ``display_name`` is pure UI metadata — the pipeline never consults it for
+    execution. When the UI uploads a PDF under a random ``{uuid}.pdf`` name on
+    disk, it stashes the client-provided filename here so the browser can
+    surface "your_statement.pdf" instead of "8f3a...pdf" back to the user.
     """
 
     model_config = ConfigDict(extra="forbid")
@@ -29,6 +34,7 @@ class FileRef(BaseModel):
     url: str
     headers: dict[str, str] = Field(default_factory=dict)
     max_bytes: int | None = None
+    display_name: str | None = None
 
 
 class Context(BaseModel):
@@ -83,6 +89,44 @@ class Options(BaseModel):
     provenance: ProvenanceOptions = Field(default_factory=ProvenanceOptions)
 
 
+class UseCaseFieldDef(BaseModel):
+    """One field in an ad-hoc, caller-defined extraction schema.
+
+    The UI (and any other caller that doesn't want to wait on a backend
+    registry entry) ships one of these per desired output field. The pipeline
+    builds a fresh Pydantic response class from the list on each request.
+
+    ``choices`` only applies to ``type == "str"`` — it turns the field into a
+    ``Literal[*choices]``. For any other type the builder raises
+    ``IX_001_001``.
+    """
+
+    model_config = ConfigDict(extra="forbid")
+
+    name: str  # must be a valid Python identifier
+    type: Literal["str", "int", "float", "decimal", "date", "datetime", "bool"]
+    required: bool = False
+    description: str | None = None
+    choices: list[str] | None = None
+
+
+class InlineUseCase(BaseModel):
+    """Caller-defined use case bundled into the :class:`RequestIX`.
+
+    When present on a request, the pipeline builds the ``(Request, Response)``
+    Pydantic class pair on the fly from :attr:`fields` and skips the
+    registered use-case lookup. The registry-based ``use_case`` field is still
+    required on the request for metrics/logging but becomes a free-form label.
+    """
+
+    model_config = ConfigDict(extra="forbid")
+
+    use_case_name: str
+    system_prompt: str
+    default_model: str | None = None
+    fields: list[UseCaseFieldDef]
+
+
 class RequestIX(BaseModel):
     """Top-level job request.
 
@@ -90,6 +134,12 @@ class RequestIX(BaseModel):
     it; the REST adapter / pg-queue adapter populates it on insert. The field
     is kept here so the contract is closed-over-construction round-trips
     (e.g. when the worker re-hydrates a job out of the store).
+
+    When ``use_case_inline`` is present, the pipeline uses it verbatim to
+    build an ad-hoc ``(Request, Response)`` class pair and skips the registry
+    lookup; ``use_case`` becomes a free-form label (still required for
+    metrics/logging). When ``use_case_inline`` is absent, ``use_case`` is
+    looked up in :data:`ix.use_cases.REGISTRY` as before.
     """
 
     model_config = ConfigDict(extra="forbid")
@@ -101,3 +151,4 @@ class RequestIX(BaseModel):
     context: Context
     options: Options = Field(default_factory=Options)
     callback_url: str | None = None
+    use_case_inline: InlineUseCase | None = None

src/ix/ocr/surya_client.py

@@ -48,6 +48,11 @@ class SuryaOCRClient:
     def __init__(self) -> None:
         self._recognition_predictor: Any = None
         self._detection_predictor: Any = None
+        # ``None`` until warm_up() has run at least once. After that it's the
+        # observed value of ``torch.cuda.is_available()`` at load time. We
+        # cache it on the instance so ``/healthz`` / the UI can surface a
+        # CPU-mode warning without re-probing torch each request.
+        self.gpu_available: bool | None = None
 
     def warm_up(self) -> None:
         """Load the detection + recognition predictors. Idempotent.
@@ -72,6 +77,18 @@ class SuryaOCRClient:
         self._recognition_predictor = RecognitionPredictor(foundation)
         self._detection_predictor = DetectionPredictor()
 
+        # Best-effort CUDA probe — only after predictors loaded cleanly so we
+        # know torch is fully importable. ``torch`` is a Surya transitive
+        # dependency so if we got this far it's on sys.path. We swallow any
+        # exception to keep warm_up() sturdy: the attribute stays None and the
+        # UI falls back to "unknown" gracefully.
+        try:
+            import torch  # type: ignore[import-not-found]
+
+            self.gpu_available = bool(torch.cuda.is_available())
+        except Exception:
+            self.gpu_available = None
+
     async def ocr(
         self,
         pages: list[Page],

src/ix/pipeline/setup_step.py

@@ -34,6 +34,7 @@ from ix.ingestion import (
 )
 from ix.pipeline.step import Step
 from ix.use_cases import get_use_case
+from ix.use_cases.inline import build_use_case_classes
 
 
 class _Fetcher(Protocol):
@@ -88,9 +89,18 @@ class SetupStep(Step):
     async def process(
         self, request_ix: RequestIX, response_ix: ResponseIX
     ) -> ResponseIX:
-        # 1. Load the use-case pair — early so an unknown name fails before
+        # 1. Load the use-case pair — either from the caller's inline
-        # we waste time downloading files.
+        # definition (wins over registry) or from the registry by name.
-        use_case_request_cls, use_case_response_cls = get_use_case(request_ix.use_case)
+        # Done early so an unknown name / bad inline definition fails
+        # before we waste time downloading files.
+        if request_ix.use_case_inline is not None:
+            use_case_request_cls, use_case_response_cls = build_use_case_classes(
+                request_ix.use_case_inline
+            )
+        else:
+            use_case_request_cls, use_case_response_cls = get_use_case(
+                request_ix.use_case
+            )
         use_case_request = use_case_request_cls()
 
         # 2. Resolve the per-request scratch directory. ix_id is assigned

src/ix/store/jobs_repo.py

@@ -157,6 +157,76 @@ async def get(session: AsyncSession, job_id: UUID) -> Job | None:
     return _orm_to_job(row) if row is not None else None
 
 
+async def queue_position(
+    session: AsyncSession, job_id: UUID
+) -> tuple[int, int]:
+    """Return ``(ahead, total_active)`` for a pending/running job.
+
+    ``ahead`` counts active jobs (``pending`` or ``running``) that would be
+    claimed by the worker before this one:
+
+    * any ``running`` job is always ahead — it has the worker already.
+    * other ``pending`` jobs with a strictly older ``created_at`` are ahead
+      (the worker picks pending rows in ``ORDER BY created_at`` per
+      :func:`claim_next_pending`).
+
+    ``total_active`` is the total count of ``pending`` + ``running`` rows.
+
+    Terminal jobs (``done`` / ``error``) always return ``(0, 0)`` — there is
+    no meaningful "position" for a finished job.
+    """
+
+    row = await session.scalar(select(IxJob).where(IxJob.job_id == job_id))
+    if row is None:
+        return (0, 0)
+    if row.status not in ("pending", "running"):
+        return (0, 0)
+
+    total_active = int(
+        await session.scalar(
+            select(func.count())
+            .select_from(IxJob)
+            .where(IxJob.status.in_(("pending", "running")))
+        )
+        or 0
+    )
+
+    if row.status == "running":
+        # A running row is at the head of the queue for our purposes.
+        return (0, total_active)
+
+    # Pending: count running rows (always ahead) + older pending rows.
+    # We tiebreak on ``job_id`` for deterministic ordering when multiple
+    # rows share a ``created_at`` (e.g. the same transaction inserts two
+    # jobs, which Postgres stamps with identical ``now()`` values).
+    running_ahead = int(
+        await session.scalar(
+            select(func.count())
+            .select_from(IxJob)
+            .where(IxJob.status == "running")
+        )
+        or 0
+    )
+    pending_ahead = int(
+        await session.scalar(
+            select(func.count())
+            .select_from(IxJob)
+            .where(
+                IxJob.status == "pending",
+                (
+                    (IxJob.created_at < row.created_at)
+                    | (
+                        (IxJob.created_at == row.created_at)
+                        & (IxJob.job_id < row.job_id)
+                    )
+                ),
+            )
+        )
+        or 0
+    )
+    return (running_ahead + pending_ahead, total_active)
+
+
 async def get_by_correlation(
     session: AsyncSession, client_id: str, request_id: str
 ) -> Job | None:

src/ix/ui/__init__.py

@@ -0,0 +1,13 @@
+"""Minimal browser UI served alongside the REST API at ``/ui``.
+
+The module is intentionally thin: templates + HTMX + Pico CSS (all from
+CDNs, no build step). Uploads land in ``{cfg.tmp_dir}/ui/<uuid>.pdf`` and
+are submitted through the same :func:`ix.store.jobs_repo.insert_pending`
+entry point the REST adapter uses — the UI does not duplicate that logic.
+"""
+
+from __future__ import annotations
+
+from ix.ui.routes import build_router
+
+__all__ = ["build_router"]

src/ix/ui/routes.py

@@ -0,0 +1,428 @@
+"""``/ui`` router — thin HTML wrapper over the existing jobs pipeline.
+
+Design notes:
+
+* Uploads stream to ``{cfg.tmp_dir}/ui/{uuid4()}.pdf`` via aiofiles; the
+  file persists for the lifetime of the ``ix_id`` (no cleanup cron — spec
+  deferred).
+* The submission handler builds a :class:`RequestIX` (inline use case
+  supported) and inserts it via the same
+  :func:`ix.store.jobs_repo.insert_pending` the REST adapter uses.
+* Responses are HTML. For HTMX-triggered submissions the handler returns
+  ``HX-Redirect`` so the whole page swaps; for plain form posts it returns
+  a 303 redirect.
+* The fragment endpoint powers the polling loop: while the job is
+  pending/running, the fragment auto-refreshes every 2s via
+  ``hx-trigger="every 2s"``; when terminal, the trigger is dropped and the
+  pretty-printed response is rendered with highlight.js.
+* A process-wide 60-second cache of the OCR GPU flag (read from the
+  injected :class:`Probes`) gates a "Surya is running on CPU" notice on
+  the fragment. The fragment is polled every 2 s; re-probing the OCR
+  client on every poll is waste — one probe per minute is plenty.
+"""
+
+from __future__ import annotations
+
+import json
+import time
+import uuid
+from pathlib import Path
+from typing import Annotated
+from urllib.parse import unquote, urlsplit
+from uuid import UUID
+
+import aiofiles
+from fastapi import (
+    APIRouter,
+    Depends,
+    File,
+    Form,
+    HTTPException,
+    Request,
+    UploadFile,
+)
+from fastapi.responses import HTMLResponse, RedirectResponse, Response
+from fastapi.templating import Jinja2Templates
+from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
+
+from ix.adapters.rest.routes import Probes, get_probes, get_session_factory_dep
+from ix.config import AppConfig, get_config
+from ix.contracts.request import (
+    Context,
+    FileRef,
+    GenAIOptions,
+    InlineUseCase,
+    OCROptions,
+    Options,
+    ProvenanceOptions,
+    RequestIX,
+    UseCaseFieldDef,
+)
+from ix.store import jobs_repo
+from ix.use_cases import REGISTRY
+
+TEMPLATES_DIR = Path(__file__).parent / "templates"
+STATIC_DIR = Path(__file__).parent / "static"
+
+# Module-level cache for the OCR GPU flag. The tuple is ``(value, expires_at)``
+# where ``expires_at`` is a monotonic-clock deadline. A per-request call to
+# :func:`_cached_ocr_gpu` re-probes only once the deadline has passed.
+_OCR_GPU_CACHE: tuple[bool | None, float] = (None, 0.0)
+_OCR_GPU_TTL_SECONDS = 60.0
+
+
+def _templates() -> Jinja2Templates:
+    """One Jinja env per process; cheap enough to build per DI call."""
+
+    return Jinja2Templates(directory=str(TEMPLATES_DIR))
+
+
+def _ui_tmp_dir(cfg: AppConfig) -> Path:
+    """Where uploads land. Created on first use; never cleaned up."""
+
+    d = Path(cfg.tmp_dir) / "ui"
+    d.mkdir(parents=True, exist_ok=True)
+    return d
+
+
+def _cached_ocr_gpu(probes: Probes) -> bool | None:
+    """Read the cached OCR GPU flag, re-probing if the TTL has elapsed.
+
+    Used by the index + fragment routes so the HTMX poll loop doesn't hit
+    the OCR client's torch-probe every 2 seconds. Falls back to ``None``
+    (unknown) on any probe error.
+    """
+
+    global _OCR_GPU_CACHE
+    value, expires_at = _OCR_GPU_CACHE
+    now = time.monotonic()
+    if now >= expires_at:
+        try:
+            value = probes.ocr_gpu()
+        except Exception:
+            value = None
+        _OCR_GPU_CACHE = (value, now + _OCR_GPU_TTL_SECONDS)
+    return value
+
+
+def _file_display_entries(
+    request: RequestIX | None,
+) -> list[str]:
+    """Human-readable filename(s) for a request's context.files.
+
+    Prefers :attr:`FileRef.display_name`. Falls back to the URL's basename
+    (``unquote``ed so ``%20`` → space). Plain string entries use the same
+    basename rule. Empty list for a request with no files.
+    """
+
+    if request is None:
+        return []
+    out: list[str] = []
+    for entry in request.context.files:
+        if isinstance(entry, FileRef):
+            if entry.display_name:
+                out.append(entry.display_name)
+                continue
+            url = entry.url
+        else:
+            url = entry
+        basename = unquote(urlsplit(url).path.rsplit("/", 1)[-1]) or url
+        out.append(basename)
+    return out
+
+
+def build_router() -> APIRouter:
+    """Return a fresh router. Kept as a factory so :mod:`ix.app` can wire DI."""
+
+    router = APIRouter(prefix="/ui", tags=["ui"])
+
+    @router.get("", response_class=HTMLResponse)
+    @router.get("/", response_class=HTMLResponse)
+    async def index(
+        request: Request,
+        probes: Annotated[Probes, Depends(get_probes)],
+    ) -> Response:
+        tpl = _templates()
+        return tpl.TemplateResponse(
+            request,
+            "index.html",
+            {
+                "registered_use_cases": sorted(REGISTRY.keys()),
+                "job": None,
+                "form_error": None,
+                "form_values": {},
+                "file_names": [],
+                "cpu_mode": _cached_ocr_gpu(probes) is False,
+            },
+        )
+
+    @router.get("/jobs/{job_id}", response_class=HTMLResponse)
+    async def job_page(
+        request: Request,
+        job_id: UUID,
+        session_factory: Annotated[
+            async_sessionmaker[AsyncSession], Depends(get_session_factory_dep)
+        ],
+        probes: Annotated[Probes, Depends(get_probes)],
+    ) -> Response:
+        async with session_factory() as session:
+            job = await jobs_repo.get(session, job_id)
+        if job is None:
+            raise HTTPException(status_code=404, detail="job not found")
+        tpl = _templates()
+        return tpl.TemplateResponse(
+            request,
+            "index.html",
+            {
+                "registered_use_cases": sorted(REGISTRY.keys()),
+                "job": job,
+                "form_error": None,
+                "form_values": {},
+                "file_names": _file_display_entries(job.request),
+                "cpu_mode": _cached_ocr_gpu(probes) is False,
+            },
+        )
+
+    @router.get("/jobs/{job_id}/fragment", response_class=HTMLResponse)
+    async def job_fragment(
+        request: Request,
+        job_id: UUID,
+        session_factory: Annotated[
+            async_sessionmaker[AsyncSession], Depends(get_session_factory_dep)
+        ],
+        probes: Annotated[Probes, Depends(get_probes)],
+    ) -> Response:
+        async with session_factory() as session:
+            job = await jobs_repo.get(session, job_id)
+            if job is None:
+                raise HTTPException(status_code=404, detail="job not found")
+            ahead, total_active = await jobs_repo.queue_position(
+                session, job_id
+            )
+
+        response_json: str | None = None
+        if job.response is not None:
+            response_json = json.dumps(
+                job.response.model_dump(mode="json"),
+                indent=2,
+                sort_keys=True,
+                default=str,
+            )
+
+        elapsed_text = _format_elapsed(job)
+        file_names = _file_display_entries(job.request)
+
+        tpl = _templates()
+        return tpl.TemplateResponse(
+            request,
+            "job_fragment.html",
+            {
+                "job": job,
+                "response_json": response_json,
+                "ahead": ahead,
+                "total_active": total_active,
+                "elapsed_text": elapsed_text,
+                "file_names": file_names,
+                "cpu_mode": _cached_ocr_gpu(probes) is False,
+            },
+        )
+
+    @router.post("/jobs")
+    async def submit_job(
+        request: Request,
+        session_factory: Annotated[
+            async_sessionmaker[AsyncSession], Depends(get_session_factory_dep)
+        ],
+        pdf: Annotated[UploadFile, File()],
+        use_case_name: Annotated[str, Form()],
+        use_case_mode: Annotated[str, Form()] = "registered",
+        texts: Annotated[str, Form()] = "",
+        ix_client_id: Annotated[str, Form()] = "ui",
+        request_id: Annotated[str, Form()] = "",
+        system_prompt: Annotated[str, Form()] = "",
+        default_model: Annotated[str, Form()] = "",
+        fields_json: Annotated[str, Form()] = "",
+        use_ocr: Annotated[str, Form()] = "",
+        ocr_only: Annotated[str, Form()] = "",
+        include_ocr_text: Annotated[str, Form()] = "",
+        include_geometries: Annotated[str, Form()] = "",
+        gen_ai_model_name: Annotated[str, Form()] = "",
+        include_provenance: Annotated[str, Form()] = "",
+        max_sources_per_field: Annotated[str, Form()] = "10",
+    ) -> Response:
+        cfg = get_config()
+        form_values = {
+            "use_case_mode": use_case_mode,
+            "use_case_name": use_case_name,
+            "ix_client_id": ix_client_id,
+            "request_id": request_id,
+            "texts": texts,
+            "system_prompt": system_prompt,
+            "default_model": default_model,
+            "fields_json": fields_json,
+            "use_ocr": use_ocr,
+            "ocr_only": ocr_only,
+            "include_ocr_text": include_ocr_text,
+            "include_geometries": include_geometries,
+            "gen_ai_model_name": gen_ai_model_name,
+            "include_provenance": include_provenance,
+            "max_sources_per_field": max_sources_per_field,
+        }
+
+        def _rerender(error: str, status: int = 200) -> Response:
+            tpl = _templates()
+            return tpl.TemplateResponse(
+                request,
+                "index.html",
+                {
+                    "registered_use_cases": sorted(REGISTRY.keys()),
+                    "job": None,
+                    "form_error": error,
+                    "form_values": form_values,
+                },
+                status_code=status,
+            )
+
+        # --- Inline use case (optional) ---
+        inline: InlineUseCase | None = None
+        if use_case_mode == "custom":
+            try:
+                raw_fields = json.loads(fields_json)
+            except json.JSONDecodeError as exc:
+                return _rerender(f"Invalid fields JSON: {exc}", status=422)
+            if not isinstance(raw_fields, list):
+                return _rerender(
+                    "Invalid fields JSON: must be a list of field objects",
+                    status=422,
+                )
+            try:
+                parsed = [UseCaseFieldDef.model_validate(f) for f in raw_fields]
+                inline = InlineUseCase(
+                    use_case_name=use_case_name,
+                    system_prompt=system_prompt,
+                    default_model=default_model or None,
+                    fields=parsed,
+                )
+            except Exception as exc:  # pydantic ValidationError or similar
+                return _rerender(
+                    f"Invalid inline use-case definition: {exc}",
+                    status=422,
+                )
+
+        # --- PDF upload ---
+        upload_dir = _ui_tmp_dir(cfg)
+        target = upload_dir / f"{uuid.uuid4().hex}.pdf"
+        # Stream copy with a size cap matching IX_FILE_MAX_BYTES.
+        total = 0
+        limit = cfg.file_max_bytes
+        async with aiofiles.open(target, "wb") as out:
+            while True:
+                chunk = await pdf.read(64 * 1024)
+                if not chunk:
+                    break
+                total += len(chunk)
+                if total > limit:
+                    # Drop the partial file; no stored state.
+                    from contextlib import suppress
+
+                    with suppress(FileNotFoundError):
+                        target.unlink()
+                    return _rerender(
+                        f"PDF exceeds IX_FILE_MAX_BYTES ({limit} bytes)",
+                        status=413,
+                    )
+                await out.write(chunk)
+
+        # --- Build RequestIX ---
+        ctx_texts: list[str] = []
+        if texts.strip():
+            ctx_texts = [texts.strip()]
+
+        req_id = request_id.strip() or uuid.uuid4().hex
+        # Preserve the client-provided filename so the UI can surface the
+        # original name to the user (the on-disk name is a UUID). Strip any
+        # path prefix a browser included.
+        original_name = (pdf.filename or "").rsplit("/", 1)[-1].rsplit(
+            "\\", 1
+        )[-1] or None
+        try:
+            request_ix = RequestIX(
+                use_case=use_case_name or "adhoc",
+                use_case_inline=inline,
+                ix_client_id=(ix_client_id.strip() or "ui"),
+                request_id=req_id,
+                context=Context(
+                    files=[
+                        FileRef(
+                            url=f"file://{target.resolve()}",
+                            display_name=original_name,
+                        )
+                    ],
+                    texts=ctx_texts,
+                ),
+                options=Options(
+                    ocr=OCROptions(
+                        use_ocr=_flag(use_ocr, default=True),
+                        ocr_only=_flag(ocr_only, default=False),
+                        include_ocr_text=_flag(include_ocr_text, default=False),
+                        include_geometries=_flag(include_geometries, default=False),
+                    ),
+                    gen_ai=GenAIOptions(
+                        gen_ai_model_name=(gen_ai_model_name.strip() or None),
+                    ),
+                    provenance=ProvenanceOptions(
+                        include_provenance=_flag(include_provenance, default=True),
+                        max_sources_per_field=int(max_sources_per_field or 10),
+                    ),
+                ),
+            )
+        except Exception as exc:
+            return _rerender(f"Invalid request: {exc}", status=422)
+
+        async with session_factory() as session:
+            job = await jobs_repo.insert_pending(
+                session, request_ix, callback_url=None
+            )
+            await session.commit()
+
+        redirect_to = f"/ui/jobs/{job.job_id}"
+        if request.headers.get("HX-Request", "").lower() == "true":
+            return Response(status_code=200, headers={"HX-Redirect": redirect_to})
+        return RedirectResponse(url=redirect_to, status_code=303)
+
+    return router
+
+
+def _flag(value: str, *, default: bool) -> bool:
+    """HTML forms omit unchecked checkboxes. Treat absence as ``default``."""
+
+    if value == "":
+        return default
+    return value.lower() in ("on", "true", "1", "yes")
+
+
+def _format_elapsed(job) -> str | None:  # type: ignore[no-untyped-def]
+    """Render a ``MM:SS`` elapsed string for the fragment template.
+
+    * running → time since ``started_at``
+    * done/error → ``finished_at - created_at`` (total wall-clock including
+      queue time)
+    * pending / missing timestamps → ``None`` (template omits the line)
+    """
+
+    from datetime import UTC, datetime
+
+    def _fmt(seconds: float) -> str:
+        s = max(0, int(seconds))
+        return f"{s // 60:02d}:{s % 60:02d}"
+
+    if job.status == "running" and job.started_at is not None:
+        now = datetime.now(UTC)
+        return _fmt((now - job.started_at).total_seconds())
+    if (
+        job.status in ("done", "error")
+        and job.finished_at is not None
+        and job.created_at is not None
+    ):
+        return _fmt((job.finished_at - job.created_at).total_seconds())
+    return None

src/ix/ui/templates/index.html

@@ -0,0 +1,241 @@
+<!doctype html>
+<html lang="en" data-theme="light">
+<head>
+  <meta charset="utf-8" />
+  <meta name="viewport" content="width=device-width, initial-scale=1" />
+  <title>
+    InfoXtractor{% if job %} &mdash; job {{ job.job_id }}{% endif %}
+  </title>
+  <link
+    rel="stylesheet"
+    href="https://cdn.jsdelivr.net/npm/@picocss/pico@2/css/pico.min.css"
+  />
+  <link
+    rel="stylesheet"
+    href="https://cdnjs.cloudflare.com/ajax/libs/highlight.js/11.10.0/styles/atom-one-light.min.css"
+  />
+  <script src="https://unpkg.com/htmx.org@1.9.12"></script>
+  <script src="https://cdnjs.cloudflare.com/ajax/libs/highlight.js/11.10.0/highlight.min.js"></script>
+  <style>
+    main { padding-top: 1.5rem; padding-bottom: 4rem; }
+    pre code.hljs { padding: 1rem; border-radius: 0.4rem; }
+    .form-error { color: var(--pico-del-color, #c44); font-weight: 600; }
+    details[open] > summary { margin-bottom: 0.5rem; }
+    .field-hint { font-size: 0.85rem; color: var(--pico-muted-color); }
+    nav.ix-header {
+      display: flex; gap: 1rem; align-items: baseline;
+      padding: 0.6rem 0; border-bottom: 1px solid var(--pico-muted-border-color, #ddd);
+      margin-bottom: 1rem; flex-wrap: wrap;
+    }
+    nav.ix-header .brand { font-weight: 700; margin-right: auto; }
+    nav.ix-header code { font-size: 0.9em; }
+    .status-panel, .result-panel { margin-top: 0.75rem; }
+    .status-panel header, .result-panel header { font-size: 0.95rem; }
+    .job-files code { font-size: 0.9em; }
+    .cpu-notice { margin-top: 0.6rem; font-size: 0.9rem; color: var(--pico-muted-color); }
+    .live-dot {
+      display: inline-block; margin-left: 0.3rem;
+      animation: ix-blink 1.2s ease-in-out infinite;
+      color: var(--pico-primary, #4f8cc9);
+    }
+    @keyframes ix-blink {
+      0%, 100% { opacity: 0.2; }
+      50% { opacity: 1; }
+    }
+    .copy-btn {
+      margin-left: 0.3rem; padding: 0.1rem 0.5rem;
+      font-size: 0.8rem; line-height: 1.2;
+    }
+  </style>
+</head>
+<body>
+  <main class="container">
+    <nav class="ix-header" aria-label="InfoXtractor navigation">
+      <span class="brand">InfoXtractor</span>
+      <a href="/ui">Upload a new extraction</a>
+      {% if job %}
+      <span>
+        Job:
+        <code id="current-job-id">{{ job.job_id }}</code>
+        <button
+          type="button"
+          class="secondary outline copy-btn"
+          onclick="navigator.clipboard && navigator.clipboard.writeText('{{ job.job_id }}')"
+          aria-label="Copy job id to clipboard"
+        >Copy</button>
+      </span>
+      {% endif %}
+    </nav>
+
+    <hgroup>
+      <h1>infoxtractor</h1>
+      <p>Drop a PDF, pick or define a use case, run the pipeline.</p>
+    </hgroup>
+
+    {% if form_error %}
+      <article class="form-error">
+        <p><strong>Form error:</strong> {{ form_error }}</p>
+      </article>
+    {% endif %}
+
+    {% if not job %}
+    <article>
+      <form
+        action="/ui/jobs"
+        method="post"
+        enctype="multipart/form-data"
+        hx-post="/ui/jobs"
+        hx-encoding="multipart/form-data"
+      >
+        <label>
+          PDF file
+          <input type="file" name="pdf" accept="application/pdf" required />
+        </label>
+
+        <label>
+          Extra texts (optional, e.g. Paperless OCR output)
+          <textarea
+            name="texts"
+            rows="3"
+            placeholder="Plain text passed as context.texts[0]"
+          >{{ form_values.get("texts", "") }}</textarea>
+          <small class="field-hint">Whatever you type is submitted as a single entry in <code>context.texts</code>.</small>
+        </label>
+
+        <fieldset>
+          <legend>Use case</legend>
+          <label>
+            <input
+              type="radio"
+              name="use_case_mode"
+              value="registered"
+              {% if form_values.get("use_case_mode", "registered") == "registered" %}checked{% endif %}
+              onchange="document.getElementById('custom-fields').hidden = true"
+            />
+            Registered
+          </label>
+          <label>
+            <input
+              type="radio"
+              name="use_case_mode"
+              value="custom"
+              {% if form_values.get("use_case_mode") == "custom" %}checked{% endif %}
+              onchange="document.getElementById('custom-fields').hidden = false"
+            />
+            Custom (inline)
+          </label>
+
+          <label>
+            Use case name
+            <input
+              type="text"
+              name="use_case_name"
+              list="registered-use-cases"
+              value="{{ form_values.get('use_case_name', 'bank_statement_header') }}"
+              required
+            />
+            <datalist id="registered-use-cases">
+              {% for name in registered_use_cases %}
+              <option value="{{ name }}"></option>
+              {% endfor %}
+            </datalist>
+          </label>
+
+          <div id="custom-fields" {% if form_values.get("use_case_mode") != "custom" %}hidden{% endif %}>
+            <label>
+              System prompt
+              <textarea name="system_prompt" rows="3">{{ form_values.get("system_prompt", "") }}</textarea>
+            </label>
+            <label>
+              Default model (optional)
+              <input
+                type="text"
+                name="default_model"
+                value="{{ form_values.get('default_model', '') }}"
+                placeholder="qwen3:14b"
+              />
+            </label>
+            <label>
+              Fields (JSON list of {name, type, required?, choices?, description?})
+              <textarea name="fields_json" rows="6" placeholder='[{"name": "vendor", "type": "str", "required": true}]'>{{ form_values.get("fields_json", "") }}</textarea>
+              <small class="field-hint">Types: str, int, float, decimal, date, datetime, bool. <code>choices</code> works on <code>str</code> only.</small>
+            </label>
+          </div>
+        </fieldset>
+
+        <details>
+          <summary>Advanced options</summary>
+          <label>
+            Client id
+            <input type="text" name="ix_client_id" value="{{ form_values.get('ix_client_id', 'ui') }}" />
+          </label>
+          <label>
+            Request id (blank → random)
+            <input type="text" name="request_id" value="{{ form_values.get('request_id', '') }}" />
+          </label>
+
+          <fieldset>
+            <legend>OCR</legend>
+            <label><input type="checkbox" name="use_ocr" {% if form_values.get("use_ocr", "on") %}checked{% endif %} /> use_ocr</label>
+            <label><input type="checkbox" name="ocr_only" {% if form_values.get("ocr_only") %}checked{% endif %} /> ocr_only</label>
+            <label><input type="checkbox" name="include_ocr_text" {% if form_values.get("include_ocr_text") %}checked{% endif %} /> include_ocr_text</label>
+            <label><input type="checkbox" name="include_geometries" {% if form_values.get("include_geometries") %}checked{% endif %} /> include_geometries</label>
+          </fieldset>
+
+          <label>
+            GenAI model override (optional)
+            <input type="text" name="gen_ai_model_name" value="{{ form_values.get('gen_ai_model_name', '') }}" />
+          </label>
+
+          <fieldset>
+            <legend>Provenance</legend>
+            <label><input type="checkbox" name="include_provenance" {% if form_values.get("include_provenance", "on") %}checked{% endif %} /> include_provenance</label>
+            <label>
+              max_sources_per_field
+              <input type="number" name="max_sources_per_field" min="1" max="100" value="{{ form_values.get('max_sources_per_field', '10') }}" />
+            </label>
+          </fieldset>
+        </details>
+
+        <button type="submit">Submit</button>
+      </form>
+    </article>
+    {% endif %}
+
+    {% if job %}
+    <article id="job-panel">
+      <header>
+        <strong>Job</strong> <code>{{ job.job_id }}</code>
+        <br /><small>ix_id: <code>{{ job.ix_id }}</code></small>
+        {% if file_names %}
+        <br /><small>
+          File{% if file_names|length > 1 %}s{% endif %}:
+          {% for name in file_names %}
+          <code>{{ name }}</code>{% if not loop.last %}, {% endif %}
+          {% endfor %}
+        </small>
+        {% endif %}
+      </header>
+      <div
+        id="job-status"
+        hx-get="/ui/jobs/{{ job.job_id }}/fragment"
+        hx-trigger="load"
+        hx-swap="innerHTML"
+      >
+        Loading&hellip;
+      </div>
+    </article>
+    {% endif %}
+  </main>
+
+  <script>
+    document.body.addEventListener("htmx:afterSettle", () => {
+      if (window.hljs) {
+        document.querySelectorAll("pre code").forEach((el) => {
+          try { hljs.highlightElement(el); } catch (_) { /* noop */ }
+        });
+      }
+    });
+  </script>
+</body>
+</html>

src/ix/ui/templates/job_fragment.html

@@ -0,0 +1,77 @@
+{#- HTMX fragment rendered into #job-status on the results panel.
+    Pending/running → keep polling every 2s; terminal → render JSON.   -#}
+{% set terminal = job.status in ("done", "error") %}
+<div
+  id="job-fragment"
+  {% if not terminal %}
+  hx-get="/ui/jobs/{{ job.job_id }}/fragment"
+  hx-trigger="every 2s"
+  hx-swap="outerHTML"
+  {% endif %}
+>
+  <article class="status-panel">
+    <header>
+      <strong>Job status</strong>
+    </header>
+
+    <p>
+      Status:
+      <strong>{{ job.status }}</strong>
+      {% if not terminal %}
+      <span class="live-dot" aria-hidden="true">&#9679;</span>
+      {% endif %}
+    </p>
+
+    {% if file_names %}
+    <p class="job-files">
+      File{% if file_names|length > 1 %}s{% endif %}:
+      {% for name in file_names %}
+      <code>{{ name }}</code>{% if not loop.last %}, {% endif %}
+      {% endfor %}
+    </p>
+    {% endif %}
+
+    {% if job.status == "pending" %}
+    <p>
+      {% if ahead == 0 %}
+      About to start &mdash; the worker just freed up.
+      {% else %}
+      Queue position: {{ ahead }} ahead &mdash; {{ total_active }} job{% if total_active != 1 %}s{% endif %} total in flight (single worker).
+      {% endif %}
+    </p>
+    <progress></progress>
+    {% elif job.status == "running" %}
+    {% if elapsed_text %}
+    <p>Running for {{ elapsed_text }}.</p>
+    {% endif %}
+    <progress></progress>
+    {% elif terminal %}
+    {% if elapsed_text %}
+    <p>Finished in {{ elapsed_text }}.</p>
+    {% endif %}
+    {% endif %}
+
+    {% if cpu_mode and not terminal %}
+    <details class="cpu-notice">
+      <summary>Surya is running on CPU (~1&ndash;2 min/page)</summary>
+      <p>
+        A host NVIDIA driver upgrade would unlock GPU extraction; tracked in
+        <code>docs/deployment.md</code>.
+      </p>
+    </details>
+    {% endif %}
+  </article>
+
+  <article class="result-panel">
+    <header>
+      <strong>Result</strong>
+    </header>
+    {% if terminal and response_json %}
+    <pre><code class="language-json">{{ response_json }}</code></pre>
+    {% elif terminal %}
+    <p><em>No response body.</em></p>
+    {% else %}
+    <p><em>Waiting for the pipeline to finish&hellip;</em></p>
+    {% endif %}
+  </article>
+</div>

src/ix/use_cases/inline.py

@@ -0,0 +1,132 @@
+"""Dynamic Pydantic class builder for caller-supplied use cases.
+
+Input: an :class:`ix.contracts.request.InlineUseCase` carried on the
+:class:`~ix.contracts.request.RequestIX`.
+
+Output: a fresh ``(RequestClass, ResponseClass)`` pair with the same shape
+as a registered use case. The :class:`~ix.pipeline.setup_step.SetupStep`
+calls this when ``request_ix.use_case_inline`` is set, bypassing the
+registry lookup entirely.
+
+The builder returns brand-new classes on every call — safe to call per
+request, so two concurrent jobs can't step on each other's schemas even if
+they happen to share a ``use_case_name``. Validation errors map to
+``IX_001_001`` (same code the registry-miss path uses); the error is
+recoverable from the caller's perspective (fix the JSON and retry), not an
+infra problem.
+"""
+
+from __future__ import annotations
+
+import keyword
+import re
+from datetime import date, datetime
+from decimal import Decimal
+from typing import Any, Literal, cast
+
+from pydantic import BaseModel, ConfigDict, Field, create_model
+
+from ix.contracts.request import InlineUseCase, UseCaseFieldDef
+from ix.errors import IXErrorCode, IXException
+
+# Map the ``UseCaseFieldDef.type`` literal to concrete Python types.
+_TYPE_MAP: dict[str, type] = {
+    "str": str,
+    "int": int,
+    "float": float,
+    "decimal": Decimal,
+    "date": date,
+    "datetime": datetime,
+    "bool": bool,
+}
+
+
+def _fail(detail: str) -> IXException:
+    return IXException(IXErrorCode.IX_001_001, detail=detail)
+
+
+def _valid_field_name(name: str) -> bool:
+    """Require a valid Python identifier that isn't a reserved keyword."""
+
+    return name.isidentifier() and not keyword.iskeyword(name)
+
+
+def _resolve_field_type(field: UseCaseFieldDef) -> Any:
+    """Return the annotation for a single field, with ``choices`` honoured."""
+
+    base = _TYPE_MAP[field.type]
+    if field.choices:
+        if field.type != "str":
+            raise _fail(
+                f"field {field.name!r}: 'choices' is only allowed for "
+                f"type='str' (got {field.type!r})"
+            )
+        return Literal[tuple(field.choices)]  # type: ignore[valid-type]
+    return base
+
+
+def _sanitise_class_name(raw: str) -> str:
+    """``re.sub(r"\\W", "_", name)`` + ``Inline_`` prefix.
+
+    Keeps the generated class name debuggable (shows up in repr / tracebacks)
+    while ensuring it's always a valid Python identifier.
+    """
+
+    return "Inline_" + re.sub(r"\W", "_", raw)
+
+
+def build_use_case_classes(
+    inline: InlineUseCase,
+) -> tuple[type[BaseModel], type[BaseModel]]:
+    """Build a fresh ``(RequestClass, ResponseClass)`` from ``inline``.
+
+    * Every call returns new classes. The caller may cache if desired; the
+      pipeline intentionally does not.
+    * Raises :class:`~ix.errors.IXException` with code
+      :attr:`~ix.errors.IXErrorCode.IX_001_001` on any structural problem
+      (empty fields, bad name, dup name, bad ``choices``).
+    """
+
+    if not inline.fields:
+        raise _fail("inline use case must define at least one field")
+
+    seen: set[str] = set()
+    for fd in inline.fields:
+        if not _valid_field_name(fd.name):
+            raise _fail(f"field name {fd.name!r} is not a valid Python identifier")
+        if fd.name in seen:
+            raise _fail(f"duplicate field name {fd.name!r}")
+        seen.add(fd.name)
+
+    response_fields: dict[str, Any] = {}
+    for fd in inline.fields:
+        annotation = _resolve_field_type(fd)
+        field_info = Field(
+            ...,
+            description=fd.description,
+        ) if fd.required else Field(
+            default=None,
+            description=fd.description,
+        )
+        if not fd.required:
+            annotation = annotation | None
+        response_fields[fd.name] = (annotation, field_info)
+
+    response_cls = create_model(  # type: ignore[call-overload]
+        _sanitise_class_name(inline.use_case_name),
+        __config__=ConfigDict(extra="forbid"),
+        **response_fields,
+    )
+
+    request_cls = create_model(  # type: ignore[call-overload]
+        "Inline_Request_" + re.sub(r"\W", "_", inline.use_case_name),
+        __config__=ConfigDict(extra="forbid"),
+        use_case_name=(str, inline.use_case_name),
+        system_prompt=(str, inline.system_prompt),
+        default_model=(str | None, inline.default_model),
+    )
+
+    return cast(type[BaseModel], request_cls), cast(type[BaseModel], response_cls)
+
+
+__all__ = ["build_use_case_classes"]

tests/integration/test_jobs_repo.py

@@ -341,6 +341,117 @@ async def test_sweep_orphans_leaves_fresh_running_alone(
     assert after.status == "running"
 
 
+async def test_queue_position_pending_only(
+    session_factory: async_sessionmaker[AsyncSession],
+) -> None:
+    """Three pending rows in insertion order → positions 0, 1, 2; total 3.
+
+    Each row is committed in its own transaction so the DB stamps a
+    distinct ``created_at`` per row (``now()`` is transaction-stable).
+    """
+
+    async with session_factory() as session:
+        a = await jobs_repo.insert_pending(
+            session, _make_request("c", "qp-a"), callback_url=None
+        )
+        await session.commit()
+    async with session_factory() as session:
+        b = await jobs_repo.insert_pending(
+            session, _make_request("c", "qp-b"), callback_url=None
+        )
+        await session.commit()
+    async with session_factory() as session:
+        c = await jobs_repo.insert_pending(
+            session, _make_request("c", "qp-c"), callback_url=None
+        )
+        await session.commit()
+
+    async with session_factory() as session:
+        pa = await jobs_repo.queue_position(session, a.job_id)
+        pb = await jobs_repo.queue_position(session, b.job_id)
+        pc = await jobs_repo.queue_position(session, c.job_id)
+
+    # All three active; total == 3.
+    assert pa == (0, 3)
+    assert pb == (1, 3)
+    assert pc == (2, 3)
+
+
+async def test_queue_position_running_plus_pending(
+    session_factory: async_sessionmaker[AsyncSession],
+) -> None:
+    """One running + two pending → running:(0,3), next:(1,3), last:(2,3)."""
+
+    async with session_factory() as session:
+        first = await jobs_repo.insert_pending(
+            session, _make_request("c", "qp-r-1"), callback_url=None
+        )
+        await session.commit()
+    async with session_factory() as session:
+        second = await jobs_repo.insert_pending(
+            session, _make_request("c", "qp-r-2"), callback_url=None
+        )
+        await session.commit()
+    async with session_factory() as session:
+        third = await jobs_repo.insert_pending(
+            session, _make_request("c", "qp-r-3"), callback_url=None
+        )
+        await session.commit()
+
+    # Claim the first → it becomes running.
+    async with session_factory() as session:
+        claimed = await jobs_repo.claim_next_pending(session)
+        await session.commit()
+    assert claimed is not None
+    assert claimed.job_id == first.job_id
+
+    async with session_factory() as session:
+        p_running = await jobs_repo.queue_position(session, first.job_id)
+        p_second = await jobs_repo.queue_position(session, second.job_id)
+        p_third = await jobs_repo.queue_position(session, third.job_id)
+
+    # Running row reports 0 ahead (itself is the head).
+    assert p_running == (0, 3)
+    # Second pending: running is ahead (1) + zero older pendings.
+    assert p_second == (1, 3)
+    # Third pending: running ahead + one older pending.
+    assert p_third == (2, 3)
+
+
+async def test_queue_position_terminal_returns_zero_zero(
+    session_factory: async_sessionmaker[AsyncSession],
+) -> None:
+    """Finished jobs have no queue position — always (0, 0)."""
+
+    async with session_factory() as session:
+        inserted = await jobs_repo.insert_pending(
+            session, _make_request("c", "qp-term"), callback_url=None
+        )
+        await session.commit()
+
+    response = ResponseIX(
+        use_case="bank_statement_header",
+        ix_client_id="c",
+        request_id="qp-term",
+    )
+    async with session_factory() as session:
+        await jobs_repo.mark_done(session, inserted.job_id, response)
+        await session.commit()
+
+    async with session_factory() as session:
+        pos = await jobs_repo.queue_position(session, inserted.job_id)
+
+    assert pos == (0, 0)
+
+
+async def test_queue_position_unknown_id_returns_zero_zero(
+    session_factory: async_sessionmaker[AsyncSession],
+) -> None:
+    async with session_factory() as session:
+        pos = await jobs_repo.queue_position(session, uuid4())
+    assert pos == (0, 0)
+
+
 async def test_concurrent_claim_never_double_dispatches(
     session_factory: async_sessionmaker[AsyncSession],
 ) -> None:

tests/integration/test_ui_routes.py

@@ -0,0 +1,537 @@
+"""Integration tests for the `/ui` router (spec §PR 2).
+
+Covers the full round-trip through `POST /ui/jobs` — the handler parses
+multipart form data into a `RequestIX` and hands it to
+`ix.store.jobs_repo.insert_pending`, the same entry point the REST adapter
+uses. Tests assert the job row exists with the right client/request ids and
+that custom-use-case forms produce a `use_case_inline` block in the stored
+request JSON.
+
+The DB-touching tests depend on the shared integration conftest which
+spins up migrations against the configured Postgres; the pure-template
+tests (`GET /ui` and the fragment renderer) still need a factory but
+won't actually query — they're cheap.
+"""
+
+from __future__ import annotations
+
+import json
+from collections.abc import Iterator
+from pathlib import Path
+from uuid import UUID, uuid4
+
+import pytest
+from fastapi.testclient import TestClient
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
+
+from ix.adapters.rest.routes import Probes, get_probes, get_session_factory_dep
+from ix.app import create_app
+from ix.store.models import IxJob
+
+FIXTURE_DIR = Path(__file__).resolve().parents[1] / "fixtures"
+FIXTURE_PDF = FIXTURE_DIR / "synthetic_giro.pdf"
+
+
+def _factory_for_url(postgres_url: str):  # type: ignore[no-untyped-def]
+    def _factory():  # type: ignore[no-untyped-def]
+        eng = create_async_engine(postgres_url, pool_pre_ping=True)
+        return async_sessionmaker(eng, expire_on_commit=False)
+
+    return _factory
+
+
+@pytest.fixture
+def app(postgres_url: str) -> Iterator[TestClient]:
+    app_obj = create_app(spawn_worker=False)
+    app_obj.dependency_overrides[get_session_factory_dep] = _factory_for_url(
+        postgres_url
+    )
+    app_obj.dependency_overrides[get_probes] = lambda: Probes(
+        ollama=lambda: "ok", ocr=lambda: "ok"
+    )
+    with TestClient(app_obj) as client:
+        yield client
+
+
+class TestIndexPage:
+    def test_index_returns_html(self, app: TestClient) -> None:
+        resp = app.get("/ui")
+        assert resp.status_code == 200
+        assert "text/html" in resp.headers["content-type"]
+        body = resp.text
+        # Dropdown prefilled with the registered use case.
+        assert "bank_statement_header" in body
+        # Marker for the submission form.
+        assert '<form' in body
+
+    def test_static_mount_is_reachable(self, app: TestClient) -> None:
+        # StaticFiles returns 404 for the keepfile; the mount itself must
+        # exist so asset URLs resolve. We probe the directory root instead.
+        resp = app.get("/ui/static/.gitkeep")
+        # .gitkeep exists in the repo — expect 200 (or at minimum not a 404
+        # due to missing mount). A 405/403 would also indicate the mount is
+        # wired; we assert the response is *not* a 404 from a missing route.
+        assert resp.status_code != 404
+
+
+class TestSubmitJobRegistered:
+    def test_post_registered_use_case_creates_row(
+        self,
+        app: TestClient,
+        postgres_url: str,
+    ) -> None:
+        request_id = f"ui-reg-{uuid4().hex[:8]}"
+        with FIXTURE_PDF.open("rb") as fh:
+            resp = app.post(
+                "/ui/jobs",
+                data={
+                    "use_case_mode": "registered",
+                    "use_case_name": "bank_statement_header",
+                    "ix_client_id": "ui-test",
+                    "request_id": request_id,
+                    "texts": "",
+                    "use_ocr": "on",
+                    "include_provenance": "on",
+                    "max_sources_per_field": "10",
+                },
+                files={"pdf": ("sample.pdf", fh, "application/pdf")},
+                follow_redirects=False,
+            )
+        assert resp.status_code in (200, 303), resp.text
+
+        # Assert the row exists in the DB.
+        job_row = _find_job(postgres_url, "ui-test", request_id)
+        assert job_row is not None
+        assert job_row.status == "pending"
+        assert job_row.request["use_case"] == "bank_statement_header"
+        # Context.files must reference a local file:// path.
+        files = job_row.request["context"]["files"]
+        assert len(files) == 1
+        entry = files[0]
+        url = entry if isinstance(entry, str) else entry["url"]
+        assert url.startswith("file://")
+
+    def test_htmx_submit_uses_hx_redirect_header(
+        self,
+        app: TestClient,
+    ) -> None:
+        request_id = f"ui-htmx-{uuid4().hex[:8]}"
+        with FIXTURE_PDF.open("rb") as fh:
+            resp = app.post(
+                "/ui/jobs",
+                data={
+                    "use_case_mode": "registered",
+                    "use_case_name": "bank_statement_header",
+                    "ix_client_id": "ui-test",
+                    "request_id": request_id,
+                },
+                files={"pdf": ("sample.pdf", fh, "application/pdf")},
+                headers={"HX-Request": "true"},
+                follow_redirects=False,
+            )
+        assert resp.status_code == 200
+        assert "HX-Redirect" in resp.headers
+
+
+class TestSubmitJobCustom:
+    def test_post_custom_use_case_stores_inline(
+        self,
+        app: TestClient,
+        postgres_url: str,
+    ) -> None:
+        request_id = f"ui-cust-{uuid4().hex[:8]}"
+        fields_json = json.dumps(
+            [
+                {"name": "vendor", "type": "str", "required": True},
+                {"name": "total", "type": "decimal"},
+            ]
+        )
+        with FIXTURE_PDF.open("rb") as fh:
+            resp = app.post(
+                "/ui/jobs",
+                data={
+                    "use_case_mode": "custom",
+                    "use_case_name": "invoice_adhoc",
+                    "ix_client_id": "ui-test",
+                    "request_id": request_id,
+                    "system_prompt": "Extract vendor and total.",
+                    "default_model": "qwen3:14b",
+                    "fields_json": fields_json,
+                },
+                files={"pdf": ("sample.pdf", fh, "application/pdf")},
+                follow_redirects=False,
+            )
+        assert resp.status_code in (200, 303), resp.text
+        job_row = _find_job(postgres_url, "ui-test", request_id)
+        assert job_row is not None
+        stored = job_row.request["use_case_inline"]
+        assert stored is not None
+        assert stored["use_case_name"] == "invoice_adhoc"
+        assert stored["system_prompt"] == "Extract vendor and total."
+        names = [f["name"] for f in stored["fields"]]
+        assert names == ["vendor", "total"]
+
+    def test_post_malformed_fields_json_rejected(
+        self,
+        app: TestClient,
+        postgres_url: str,
+    ) -> None:
+        request_id = f"ui-bad-{uuid4().hex[:8]}"
+        with FIXTURE_PDF.open("rb") as fh:
+            resp = app.post(
+                "/ui/jobs",
+                data={
+                    "use_case_mode": "custom",
+                    "use_case_name": "adhoc_bad",
+                    "ix_client_id": "ui-test",
+                    "request_id": request_id,
+                    "system_prompt": "p",
+                    "fields_json": "this is not json",
+                },
+                files={"pdf": ("sample.pdf", fh, "application/pdf")},
+                follow_redirects=False,
+            )
+        # Either re-rendered form (422 / 200 with error) — what matters is
+        # that no row was inserted.
+        assert resp.status_code in (200, 400, 422)
+        job_row = _find_job(postgres_url, "ui-test", request_id)
+        assert job_row is None
+        # A helpful error should appear somewhere in the body.
+        assert (
+            "error" in resp.text.lower()
+            or "invalid" in resp.text.lower()
+            or "json" in resp.text.lower()
+        )
+
+
+class TestDisplayName:
+    def test_post_persists_display_name_in_file_ref(
+        self,
+        app: TestClient,
+        postgres_url: str,
+    ) -> None:
+        """The client-provided upload filename lands in FileRef.display_name."""
+
+        request_id = f"ui-name-{uuid4().hex[:8]}"
+        with FIXTURE_PDF.open("rb") as fh:
+            resp = app.post(
+                "/ui/jobs",
+                data={
+                    "use_case_mode": "registered",
+                    "use_case_name": "bank_statement_header",
+                    "ix_client_id": "ui-test",
+                    "request_id": request_id,
+                },
+                files={
+                    "pdf": ("my statement.pdf", fh, "application/pdf")
+                },
+                follow_redirects=False,
+            )
+        assert resp.status_code in (200, 303), resp.text
+
+        job_row = _find_job(postgres_url, "ui-test", request_id)
+        assert job_row is not None
+        entry = job_row.request["context"]["files"][0]
+        assert isinstance(entry, dict)
+        assert entry["display_name"] == "my statement.pdf"
+
+
+class TestFragment:
+    def test_fragment_pending_has_trigger(
+        self,
+        app: TestClient,
+        postgres_url: str,
+    ) -> None:
+        request_id = f"ui-frag-p-{uuid4().hex[:8]}"
+        with FIXTURE_PDF.open("rb") as fh:
+            app.post(
+                "/ui/jobs",
+                data={
+                    "use_case_mode": "registered",
+                    "use_case_name": "bank_statement_header",
+                    "ix_client_id": "ui-test",
+                    "request_id": request_id,
+                },
+                files={"pdf": ("sample.pdf", fh, "application/pdf")},
+                follow_redirects=False,
+            )
+        job_row = _find_job(postgres_url, "ui-test", request_id)
+        assert job_row is not None
+
+        resp = app.get(f"/ui/jobs/{job_row.job_id}/fragment")
+        assert resp.status_code == 200
+        body = resp.text
+        # Pending → auto-refresh every 2s.
+        assert "hx-trigger" in body
+        assert "2s" in body
+        assert "pending" in body.lower() or "running" in body.lower()
+        # New queue-awareness copy.
+        assert "Queue position" in body or "About to start" in body
+
+    def test_fragment_pending_shows_filename(
+        self,
+        app: TestClient,
+        postgres_url: str,
+    ) -> None:
+        request_id = f"ui-frag-pf-{uuid4().hex[:8]}"
+        with FIXTURE_PDF.open("rb") as fh:
+            app.post(
+                "/ui/jobs",
+                data={
+                    "use_case_mode": "registered",
+                    "use_case_name": "bank_statement_header",
+                    "ix_client_id": "ui-test",
+                    "request_id": request_id,
+                },
+                files={
+                    "pdf": (
+                        "client-side-name.pdf",
+                        fh,
+                        "application/pdf",
+                    )
+                },
+                follow_redirects=False,
+            )
+        job_row = _find_job(postgres_url, "ui-test", request_id)
+        assert job_row is not None
+        resp = app.get(f"/ui/jobs/{job_row.job_id}/fragment")
+        assert resp.status_code == 200
+        assert "client-side-name.pdf" in resp.text
+
+    def test_fragment_running_shows_elapsed(
+        self,
+        app: TestClient,
+        postgres_url: str,
+    ) -> None:
+        """After flipping a row to running with a backdated started_at, the
+        fragment renders a ``Running for MM:SS`` line."""
+
+        request_id = f"ui-frag-r-{uuid4().hex[:8]}"
+        with FIXTURE_PDF.open("rb") as fh:
+            app.post(
+                "/ui/jobs",
+                data={
+                    "use_case_mode": "registered",
+                    "use_case_name": "bank_statement_header",
+                    "ix_client_id": "ui-test",
+                    "request_id": request_id,
+                },
+                files={"pdf": ("sample.pdf", fh, "application/pdf")},
+                follow_redirects=False,
+            )
+        job_row = _find_job(postgres_url, "ui-test", request_id)
+        assert job_row is not None
+
+        _force_running(postgres_url, job_row.job_id)
+
+        resp = app.get(f"/ui/jobs/{job_row.job_id}/fragment")
+        assert resp.status_code == 200
+        body = resp.text
+        assert "Running for" in body
+        # MM:SS; our backdate is ~10s so expect 00:1? or higher.
+        import re
+
+        assert re.search(r"\d{2}:\d{2}", body), body
+
+    def test_fragment_backward_compat_no_display_name(
+        self,
+        app: TestClient,
+        postgres_url: str,
+    ) -> None:
+        """Older rows (stored before display_name existed) must still render."""
+
+        from ix.contracts.request import Context, FileRef, RequestIX
+
+        legacy_req = RequestIX(
+            use_case="bank_statement_header",
+            ix_client_id="ui-test",
+            request_id=f"ui-legacy-{uuid4().hex[:8]}",
+            context=Context(
+                files=[
+                    FileRef(url="file:///tmp/ix/ui/legacy.pdf")
+                ]
+            ),
+        )
+
+        import asyncio
+
+        from ix.store import jobs_repo as _repo
+
+        async def _insert() -> UUID:
+            eng = create_async_engine(postgres_url)
+            sf = async_sessionmaker(eng, expire_on_commit=False)
+            try:
+                async with sf() as session:
+                    job = await _repo.insert_pending(
+                        session, legacy_req, callback_url=None
+                    )
+                    await session.commit()
+                    return job.job_id
+            finally:
+                await eng.dispose()
+
+        job_id = asyncio.run(_insert())
+        resp = app.get(f"/ui/jobs/{job_id}/fragment")
+        assert resp.status_code == 200
+        body = resp.text
+        # Must not crash; must include the fallback basename from the URL.
+        assert "legacy.pdf" in body
+
+    def test_fragment_done_shows_pretty_json(
+        self,
+        app: TestClient,
+        postgres_url: str,
+    ) -> None:
+        request_id = f"ui-frag-d-{uuid4().hex[:8]}"
+        with FIXTURE_PDF.open("rb") as fh:
+            app.post(
+                "/ui/jobs",
+                data={
+                    "use_case_mode": "registered",
+                    "use_case_name": "bank_statement_header",
+                    "ix_client_id": "ui-test",
+                    "request_id": request_id,
+                },
+                files={
+                    "pdf": (
+                        "my-done-doc.pdf",
+                        fh,
+                        "application/pdf",
+                    )
+                },
+                follow_redirects=False,
+            )
+        job_row = _find_job(postgres_url, "ui-test", request_id)
+        assert job_row is not None
+
+        # Hand-tick the row to done with a fake response.
+        _force_done(
+            postgres_url,
+            job_row.job_id,
+            response_body={
+                "use_case": "bank_statement_header",
+                "ix_result": {"result": {"bank_name": "UBS AG", "currency": "CHF"}},
+            },
+        )
+
+        resp = app.get(f"/ui/jobs/{job_row.job_id}/fragment")
+        assert resp.status_code == 200
+        body = resp.text
+        # Terminal → no auto-refresh.
+        assert "every 2s" not in body and "every  2s" not in body
+        # JSON present.
+        assert "UBS AG" in body
+        assert "CHF" in body
+        # Filename surfaced on the done fragment.
+        assert "my-done-doc.pdf" in body
+
+
+def _find_job(postgres_url: str, client_id: str, request_id: str):  # type: ignore[no-untyped-def]
+    """Look up an ``ix_jobs`` row via the async engine, wrapping the coroutine
+    for test convenience."""
+
+    import asyncio
+    import json as _json
+
+    async def _go():  # type: ignore[no-untyped-def]
+        eng = create_async_engine(postgres_url)
+        sf = async_sessionmaker(eng, expire_on_commit=False)
+        try:
+            async with sf() as session:
+                r = await session.scalar(
+                    select(IxJob).where(
+                        IxJob.client_id == client_id,
+                        IxJob.request_id == request_id,
+                    )
+                )
+                if r is None:
+                    return None
+
+                class _JobRow:
+                    pass
+
+                out = _JobRow()
+                out.job_id = r.job_id
+                out.client_id = r.client_id
+                out.request_id = r.request_id
+                out.status = r.status
+                if isinstance(r.request, str):
+                    out.request = _json.loads(r.request)
+                else:
+                    out.request = r.request
+                return out
+        finally:
+            await eng.dispose()
+
+    return asyncio.run(_go())
+
+
+def _force_done(
+    postgres_url: str,
+    job_id,  # type: ignore[no-untyped-def]
+    response_body: dict,
+) -> None:
+    """Flip a pending job to ``done`` with the given response payload."""
+
+    import asyncio
+    from datetime import UTC, datetime
+
+    from sqlalchemy import text
+
+    async def _go():  # type: ignore[no-untyped-def]
+        eng = create_async_engine(postgres_url)
+        try:
+            async with eng.begin() as conn:
+                await conn.execute(
+                    text(
+                        "UPDATE ix_jobs SET status='done', "
+                        "response=CAST(:resp AS JSONB), finished_at=:now "
+                        "WHERE job_id=:jid"
+                    ),
+                    {
+                        "resp": json.dumps(response_body),
+                        "now": datetime.now(UTC),
+                        "jid": str(job_id),
+                    },
+                )
+        finally:
+            await eng.dispose()
+
+    asyncio.run(_go())
+
+
+def _force_running(
+    postgres_url: str,
+    job_id,  # type: ignore[no-untyped-def]
+    seconds_ago: int = 10,
+) -> None:
+    """Flip a pending job to ``running`` with a backdated ``started_at``.
+
+    The fragment renders "Running for MM:SS" which needs a ``started_at`` in
+    the past; 10s is enough to produce a deterministic non-zero MM:SS.
+    """
+
+    import asyncio
+    from datetime import UTC, datetime, timedelta
+
+    from sqlalchemy import text
+
+    async def _go():  # type: ignore[no-untyped-def]
+        eng = create_async_engine(postgres_url)
+        try:
+            async with eng.begin() as conn:
+                await conn.execute(
+                    text(
+                        "UPDATE ix_jobs SET status='running', started_at=:t "
+                        "WHERE job_id=:jid"
+                    ),
+                    {
+                        "t": datetime.now(UTC) - timedelta(seconds=seconds_ago),
+                        "jid": str(job_id),
+                    },
+                )
+        finally:
+            await eng.dispose()
+
+    asyncio.run(_go())

tests/unit/test_contracts.py

@@ -31,6 +31,7 @@ from ix.contracts import (
     ResponseIX,
     SegmentCitation,
 )
+from ix.contracts.request import InlineUseCase, UseCaseFieldDef
 
 
 class TestFileRef:
@@ -49,6 +50,24 @@ class TestFileRef:
         assert fr.headers == {"Authorization": "Token abc"}
         assert fr.max_bytes == 1_000_000
 
+    def test_display_name_defaults_to_none(self) -> None:
+        fr = FileRef(url="file:///tmp/ix/ui/abc.pdf")
+        assert fr.display_name is None
+
+    def test_display_name_roundtrip(self) -> None:
+        fr = FileRef(
+            url="file:///tmp/ix/ui/abc.pdf",
+            display_name="my statement.pdf",
+        )
+        assert fr.display_name == "my statement.pdf"
+        dumped = fr.model_dump_json()
+        rt = FileRef.model_validate_json(dumped)
+        assert rt.display_name == "my statement.pdf"
+        # Backward-compat: a serialised FileRef without display_name still
+        # validates cleanly (older stored jobs predate the field).
+        legacy = FileRef.model_validate({"url": "file:///x.pdf"})
+        assert legacy.display_name is None
+
 
 class TestOptionDefaults:
     def test_ocr_defaults_match_spec(self) -> None:
@@ -182,6 +201,32 @@ class TestRequestIX:
         with pytest.raises(ValidationError):
             RequestIX.model_validate({"use_case": "x"})
 
+    def test_use_case_inline_defaults_to_none(self) -> None:
+        r = RequestIX(**self._minimal_payload())
+        assert r.use_case_inline is None
+
+    def test_use_case_inline_roundtrip(self) -> None:
+        payload = self._minimal_payload()
+        payload["use_case_inline"] = {
+            "use_case_name": "adhoc",
+            "system_prompt": "extract stuff",
+            "fields": [
+                {"name": "a", "type": "str", "required": True},
+                {"name": "b", "type": "int"},
+            ],
+        }
+        r = RequestIX.model_validate(payload)
+        assert r.use_case_inline is not None
+        assert isinstance(r.use_case_inline, InlineUseCase)
+        assert r.use_case_inline.use_case_name == "adhoc"
+        assert len(r.use_case_inline.fields) == 2
+        assert isinstance(r.use_case_inline.fields[0], UseCaseFieldDef)
+        # Round-trip through JSON
+        dumped = r.model_dump_json()
+        r2 = RequestIX.model_validate_json(dumped)
+        assert r2.use_case_inline is not None
+        assert r2.use_case_inline.fields[1].type == "int"
+
 
 class TestOCRResult:
     def test_minimal_defaults(self) -> None:

tests/unit/test_setup_step.py

@@ -15,6 +15,7 @@ from ix.contracts import (
     RequestIX,
     ResponseIX,
 )
+from ix.contracts.request import InlineUseCase, UseCaseFieldDef
 from ix.contracts.response import _InternalContext
 from ix.errors import IXErrorCode, IXException
 from ix.ingestion import FetchConfig
@@ -244,6 +245,102 @@ class TestTextOnly:
         assert ctx.texts == ["hello", "there"]
 
 
+class TestInlineUseCase:
+    def _make_inline_request(
+        self,
+        inline: InlineUseCase,
+        use_case: str = "adhoc-label",
+        texts: list[str] | None = None,
+    ) -> RequestIX:
+        return RequestIX(
+            use_case=use_case,
+            use_case_inline=inline,
+            ix_client_id="test",
+            request_id="r-inline",
+            context=Context(files=[], texts=texts or ["hello"]),
+            options=Options(
+                ocr=OCROptions(use_ocr=True),
+                provenance=ProvenanceOptions(include_provenance=True),
+            ),
+        )
+
+    async def test_inline_use_case_overrides_registry(self, tmp_path: Path) -> None:
+        fetcher = FakeFetcher({})
+        ingestor = FakeIngestor([])
+        step = SetupStep(
+            fetcher=fetcher,
+            ingestor=ingestor,
+            tmp_dir=tmp_path / "work",
+            fetch_config=_make_cfg(),
+            mime_detector=_AlwaysMimePdf(),
+        )
+        inline = InlineUseCase(
+            use_case_name="adhoc",
+            system_prompt="Extract things.",
+            fields=[
+                UseCaseFieldDef(name="vendor", type="str", required=True),
+                UseCaseFieldDef(name="amount", type="decimal"),
+            ],
+        )
+        req = self._make_inline_request(inline)
+        resp = _make_response()
+        resp = await step.process(req, resp)
+
+        ctx = resp.context
+        assert ctx is not None
+        # The response class must have been built from our field list.
+        resp_cls = ctx.use_case_response  # type: ignore[union-attr]
+        assert set(resp_cls.model_fields.keys()) == {"vendor", "amount"}
+        # Public display name reflects the inline label.
+        assert resp.use_case_name == "adhoc"
+
+    async def test_inline_precedence_when_both_set(self, tmp_path: Path) -> None:
+        # ``use_case`` is a valid registered name; ``use_case_inline`` is also
+        # present. Inline MUST win (documented precedence).
+        fetcher = FakeFetcher({})
+        ingestor = FakeIngestor([])
+        step = SetupStep(
+            fetcher=fetcher,
+            ingestor=ingestor,
+            tmp_dir=tmp_path / "work",
+            fetch_config=_make_cfg(),
+            mime_detector=_AlwaysMimePdf(),
+        )
+        inline = InlineUseCase(
+            use_case_name="override",
+            system_prompt="override prompt",
+            fields=[UseCaseFieldDef(name="just_me", type="str", required=True)],
+        )
+        req = self._make_inline_request(
+            inline, use_case="bank_statement_header"
+        )
+        resp = await step.process(req, _make_response())
+        resp_cls = resp.context.use_case_response  # type: ignore[union-attr]
+        assert set(resp_cls.model_fields.keys()) == {"just_me"}
+
+    async def test_inline_with_bad_field_raises_ix_001_001(
+        self, tmp_path: Path
+    ) -> None:
+        fetcher = FakeFetcher({})
+        ingestor = FakeIngestor([])
+        step = SetupStep(
+            fetcher=fetcher,
+            ingestor=ingestor,
+            tmp_dir=tmp_path / "work",
+            fetch_config=_make_cfg(),
+            mime_detector=_AlwaysMimePdf(),
+        )
+        inline = InlineUseCase(
+            use_case_name="bad",
+            system_prompt="p",
+            fields=[UseCaseFieldDef(name="123bad", type="str")],
+        )
+        req = self._make_inline_request(inline)
+        with pytest.raises(IXException) as ei:
+            await step.process(req, _make_response())
+        assert ei.value.code is IXErrorCode.IX_001_001
+
+
 class TestInternalContextShape:
     async def test_context_is_internal_context_instance(self, tmp_path: Path) -> None:
         fetcher = FakeFetcher({})

tests/unit/test_surya_client.py

@@ -161,6 +161,78 @@ def _write_tiny_png(path: Path) -> None:
     Image.new("RGB", (2, 2), color="white").save(path, format="PNG")
 
 
+class TestGpuAvailableFlag:
+    def test_default_is_none(self) -> None:
+        client = SuryaOCRClient()
+        assert client.gpu_available is None
+
+    def test_warm_up_probes_cuda_true(self) -> None:
+        """When torch reports CUDA, warm_up records True on the instance."""
+
+        client = SuryaOCRClient()
+        fake_foundation = MagicMock()
+        fake_recognition = MagicMock()
+        fake_detection = MagicMock()
+        fake_torch = SimpleNamespace(
+            cuda=SimpleNamespace(is_available=lambda: True)
+        )
+
+        module_patches = {
+            "surya.detection": SimpleNamespace(
+                DetectionPredictor=lambda: fake_detection
+            ),
+            "surya.foundation": SimpleNamespace(
+                FoundationPredictor=lambda: fake_foundation
+            ),
+            "surya.recognition": SimpleNamespace(
+                RecognitionPredictor=lambda _f: fake_recognition
+            ),
+            "torch": fake_torch,
+        }
+        with patch.dict("sys.modules", module_patches):
+            client.warm_up()
+
+        assert client.gpu_available is True
+        assert client._recognition_predictor is fake_recognition
+        assert client._detection_predictor is fake_detection
+
+    def test_warm_up_probes_cuda_false(self) -> None:
+        """CPU-mode host → warm_up records False."""
+
+        client = SuryaOCRClient()
+        fake_torch = SimpleNamespace(
+            cuda=SimpleNamespace(is_available=lambda: False)
+        )
+        module_patches = {
+            "surya.detection": SimpleNamespace(
+                DetectionPredictor=lambda: MagicMock()
+            ),
+            "surya.foundation": SimpleNamespace(
+                FoundationPredictor=lambda: MagicMock()
+            ),
+            "surya.recognition": SimpleNamespace(
+                RecognitionPredictor=lambda _f: MagicMock()
+            ),
+            "torch": fake_torch,
+        }
+        with patch.dict("sys.modules", module_patches):
+            client.warm_up()
+
+        assert client.gpu_available is False
+
+    def test_warm_up_is_idempotent_for_probe(self) -> None:
+        """Second warm_up short-circuits; probed flag is preserved."""
+
+        client = SuryaOCRClient()
+        client._recognition_predictor = MagicMock()
+        client._detection_predictor = MagicMock()
+        client.gpu_available = True
+
+        # No module patches — warm_up must NOT touch sys.modules or torch.
+        client.warm_up()
+        assert client.gpu_available is True
+
+
 @pytest.mark.parametrize("unused", [None])  # keep pytest happy if file ever runs alone
 def test_module_imports(unused: None) -> None:
     assert SuryaOCRClient is not None

tests/unit/test_use_case_inline.py

@@ -0,0 +1,313 @@
+"""Tests for :mod:`ix.use_cases.inline` — dynamic Pydantic class builder.
+
+The builder takes an :class:`InlineUseCase` (carried on :class:`RequestIX` as
+``use_case_inline``) and produces a fresh ``(RequestClass, ResponseClass)``
+pair that the pipeline can consume in place of a registered use case.
+"""
+
+from __future__ import annotations
+
+import json
+from datetime import date, datetime
+from decimal import Decimal
+
+import pytest
+from pydantic import BaseModel, ValidationError
+
+from ix.contracts.request import InlineUseCase, UseCaseFieldDef
+from ix.errors import IXErrorCode, IXException
+from ix.use_cases.inline import build_use_case_classes
+
+
+class TestUseCaseFieldDef:
+    def test_minimal(self) -> None:
+        fd = UseCaseFieldDef(name="foo", type="str")
+        assert fd.name == "foo"
+        assert fd.type == "str"
+        assert fd.required is False
+        assert fd.description is None
+        assert fd.choices is None
+
+    def test_extra_forbidden(self) -> None:
+        with pytest.raises(ValidationError):
+            UseCaseFieldDef.model_validate(
+                {"name": "foo", "type": "str", "bogus": 1}
+            )
+
+    def test_invalid_type_rejected(self) -> None:
+        with pytest.raises(ValidationError):
+            UseCaseFieldDef.model_validate({"name": "foo", "type": "list"})
+
+
+class TestInlineUseCaseRoundtrip:
+    def test_json_roundtrip(self) -> None:
+        iuc = InlineUseCase(
+            use_case_name="Vendor Total",
+            system_prompt="Extract invoice total and vendor.",
+            default_model="qwen3:14b",
+            fields=[
+                UseCaseFieldDef(name="vendor", type="str", required=True),
+                UseCaseFieldDef(
+                    name="total",
+                    type="decimal",
+                    required=True,
+                    description="total amount due",
+                ),
+                UseCaseFieldDef(
+                    name="currency",
+                    type="str",
+                    choices=["USD", "EUR", "CHF"],
+                ),
+            ],
+        )
+        dumped = iuc.model_dump_json()
+        round = InlineUseCase.model_validate_json(dumped)
+        assert round == iuc
+        # JSON is well-formed
+        json.loads(dumped)
+
+    def test_extra_forbidden(self) -> None:
+        with pytest.raises(ValidationError):
+            InlineUseCase.model_validate(
+                {
+                    "use_case_name": "X",
+                    "system_prompt": "p",
+                    "fields": [],
+                    "bogus": 1,
+                }
+            )
+
+
+class TestBuildBasicTypes:
+    @pytest.mark.parametrize(
+        "type_name, sample_value, bad_value",
+        [
+            ("str", "hello", 123),
+            ("int", 42, "nope"),
+            ("float", 3.14, "nope"),
+            ("bool", True, "nope"),
+        ],
+    )
+    def test_simple_type(
+        self, type_name: str, sample_value: object, bad_value: object
+    ) -> None:
+        iuc = InlineUseCase(
+            use_case_name="X",
+            system_prompt="p",
+            fields=[UseCaseFieldDef(name="field", type=type_name, required=True)],
+        )
+        _req_cls, resp_cls = build_use_case_classes(iuc)
+        instance = resp_cls(field=sample_value)
+        assert instance.field == sample_value
+        with pytest.raises(ValidationError):
+            resp_cls(field=bad_value)
+
+    def test_decimal_type(self) -> None:
+        iuc = InlineUseCase(
+            use_case_name="X",
+            system_prompt="p",
+            fields=[UseCaseFieldDef(name="amount", type="decimal", required=True)],
+        )
+        _req_cls, resp_cls = build_use_case_classes(iuc)
+        instance = resp_cls(amount="12.34")
+        assert isinstance(instance.amount, Decimal)
+        assert instance.amount == Decimal("12.34")
+
+    def test_date_type(self) -> None:
+        iuc = InlineUseCase(
+            use_case_name="X",
+            system_prompt="p",
+            fields=[UseCaseFieldDef(name="d", type="date", required=True)],
+        )
+        _req_cls, resp_cls = build_use_case_classes(iuc)
+        instance = resp_cls(d="2026-04-18")
+        assert instance.d == date(2026, 4, 18)
+
+    def test_datetime_type(self) -> None:
+        iuc = InlineUseCase(
+            use_case_name="X",
+            system_prompt="p",
+            fields=[UseCaseFieldDef(name="ts", type="datetime", required=True)],
+        )
+        _req_cls, resp_cls = build_use_case_classes(iuc)
+        instance = resp_cls(ts="2026-04-18T10:00:00")
+        assert isinstance(instance.ts, datetime)
+
+
+class TestOptionalVsRequired:
+    def test_required_field_cannot_be_missing(self) -> None:
+        iuc = InlineUseCase(
+            use_case_name="X",
+            system_prompt="p",
+            fields=[UseCaseFieldDef(name="must", type="str", required=True)],
+        )
+        _req_cls, resp_cls = build_use_case_classes(iuc)
+        with pytest.raises(ValidationError):
+            resp_cls()
+
+    def test_optional_field_defaults_to_none(self) -> None:
+        iuc = InlineUseCase(
+            use_case_name="X",
+            system_prompt="p",
+            fields=[UseCaseFieldDef(name="maybe", type="str", required=False)],
+        )
+        _req_cls, resp_cls = build_use_case_classes(iuc)
+        instance = resp_cls()
+        assert instance.maybe is None
+
+    def test_optional_field_schema_allows_null(self) -> None:
+        iuc = InlineUseCase(
+            use_case_name="X",
+            system_prompt="p",
+            fields=[UseCaseFieldDef(name="maybe", type="str", required=False)],
+        )
+        _req_cls, resp_cls = build_use_case_classes(iuc)
+        schema = resp_cls.model_json_schema()
+        # "maybe" accepts string or null
+        prop = schema["properties"]["maybe"]
+        # Pydantic may express Optional as anyOf [str, null] or a type list.
+        # Either is fine — just assert null is allowed somewhere.
+        dumped = json.dumps(prop)
+        assert "null" in dumped
+
+
+class TestChoices:
+    def test_choices_for_str_produces_literal(self) -> None:
+        iuc = InlineUseCase(
+            use_case_name="X",
+            system_prompt="p",
+            fields=[
+                UseCaseFieldDef(
+                    name="kind",
+                    type="str",
+                    required=True,
+                    choices=["a", "b", "c"],
+                )
+            ],
+        )
+        _req_cls, resp_cls = build_use_case_classes(iuc)
+        inst = resp_cls(kind="a")
+        assert inst.kind == "a"
+        with pytest.raises(ValidationError):
+            resp_cls(kind="nope")
+        schema = resp_cls.model_json_schema()
+        # enum or const wind up in a referenced definition; walk the schema
+        dumped = json.dumps(schema)
+        assert '"a"' in dumped and '"b"' in dumped and '"c"' in dumped
+
+    def test_choices_for_non_str_raises_ix_001_001(self) -> None:
+        iuc = InlineUseCase(
+            use_case_name="X",
+            system_prompt="p",
+            fields=[
+                UseCaseFieldDef(
+                    name="kind",
+                    type="int",
+                    required=True,
+                    choices=["1", "2"],
+                )
+            ],
+        )
+        with pytest.raises(IXException) as exc:
+            build_use_case_classes(iuc)
+        assert exc.value.code is IXErrorCode.IX_001_001
+
+    def test_empty_choices_list_ignored(self) -> None:
+        # An explicitly empty list is as-if choices were unset; builder must
+        # not break. If the caller sent choices=[] we treat the field as
+        # plain str.
+        iuc = InlineUseCase(
+            use_case_name="X",
+            system_prompt="p",
+            fields=[
+                UseCaseFieldDef(
+                    name="kind", type="str", required=True, choices=[]
+                )
+            ],
+        )
+        _req_cls, resp_cls = build_use_case_classes(iuc)
+        inst = resp_cls(kind="anything")
+        assert inst.kind == "anything"
+
+
+class TestValidation:
+    def test_duplicate_field_names_raise(self) -> None:
+        iuc = InlineUseCase(
+            use_case_name="X",
+            system_prompt="p",
+            fields=[
+                UseCaseFieldDef(name="foo", type="str"),
+                UseCaseFieldDef(name="foo", type="int"),
+            ],
+        )
+        with pytest.raises(IXException) as exc:
+            build_use_case_classes(iuc)
+        assert exc.value.code is IXErrorCode.IX_001_001
+
+    def test_invalid_field_name_raises(self) -> None:
+        iuc = InlineUseCase(
+            use_case_name="X",
+            system_prompt="p",
+            fields=[UseCaseFieldDef(name="123abc", type="str")],
+        )
+        with pytest.raises(IXException) as exc:
+            build_use_case_classes(iuc)
+        assert exc.value.code is IXErrorCode.IX_001_001
+
+    def test_empty_fields_list_raises(self) -> None:
+        iuc = InlineUseCase(
+            use_case_name="X", system_prompt="p", fields=[]
+        )
+        with pytest.raises(IXException) as exc:
+            build_use_case_classes(iuc)
+        assert exc.value.code is IXErrorCode.IX_001_001
+
+
+class TestResponseClassNaming:
+    def test_class_name_sanitised(self) -> None:
+        iuc = InlineUseCase(
+            use_case_name="Bank / Statement — header!",
+            system_prompt="p",
+            fields=[UseCaseFieldDef(name="x", type="str")],
+        )
+        _req_cls, resp_cls = build_use_case_classes(iuc)
+        assert resp_cls.__name__.startswith("Inline_")
+        # Only alphanumerics and underscores remain.
+        assert all(c.isalnum() or c == "_" for c in resp_cls.__name__)
+
+    def test_fresh_instances_per_call(self) -> None:
+        iuc = InlineUseCase(
+            use_case_name="X",
+            system_prompt="p",
+            fields=[UseCaseFieldDef(name="x", type="str")],
+        )
+        req1, resp1 = build_use_case_classes(iuc)
+        req2, resp2 = build_use_case_classes(iuc)
+        assert resp1 is not resp2
+        assert req1 is not req2
+
+
+class TestRequestClassShape:
+    def test_request_class_exposes_prompt_and_default(self) -> None:
+        iuc = InlineUseCase(
+            use_case_name="My Case",
+            system_prompt="Follow directions.",
+            default_model="qwen3:14b",
+            fields=[UseCaseFieldDef(name="x", type="str")],
+        )
+        req_cls, _resp_cls = build_use_case_classes(iuc)
+        inst = req_cls()
+        assert inst.use_case_name == "My Case"
+        assert inst.system_prompt == "Follow directions."
+        assert inst.default_model == "qwen3:14b"
+        assert issubclass(req_cls, BaseModel)
+
+    def test_default_model_none_when_unset(self) -> None:
+        iuc = InlineUseCase(
+            use_case_name="My Case",
+            system_prompt="Follow directions.",
+            fields=[UseCaseFieldDef(name="x", type="str")],
+        )
+        req_cls, _resp_cls = build_use_case_classes(iuc)
+        inst = req_cls()
+        assert inst.default_model is None

uv.lock

@@ -7,6 +7,15 @@ resolution-markers = [
     "(platform_machine != 'aarch64' and sys_platform == 'linux') or (sys_platform != 'darwin' and sys_platform != 'linux')",
 ]
 
+[[package]]
+name = "aiofiles"
+version = "25.1.0"
+source = { registry = "https://pypi.org/simple" }
+sdist = { url = "https://files.pythonhosted.org/packages/41/c3/534eac40372d8ee36ef40df62ec129bee4fdb5ad9706e58a29be53b2c970/aiofiles-25.1.0.tar.gz", hash = "sha256:a8d728f0a29de45dc521f18f07297428d56992a742f0cd2701ba86e44d23d5b2", size = 46354, upload-time = "2025-10-09T20:51:04.358Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/bc/8a/340a1555ae33d7354dbca4faa54948d76d89a27ceef032c8c3bc661d003e/aiofiles-25.1.0-py3-none-any.whl", hash = "sha256:abe311e527c862958650f9438e859c1fa7568a141b22abcd015e120e86a85695", size = 14668, upload-time = "2025-10-09T20:51:03.174Z" },
+]
+
 [[package]]
 name = "alembic"
 version = "1.18.4"
@@ -521,16 +530,19 @@ name = "infoxtractor"
 version = "0.1.0"
 source = { editable = "." }
 dependencies = [
+    { name = "aiofiles" },
     { name = "alembic" },
     { name = "asyncpg" },
     { name = "fastapi" },
     { name = "httpx" },
+    { name = "jinja2" },
     { name = "pillow" },
     { name = "pydantic" },
     { name = "pydantic-settings" },
     { name = "pymupdf" },
     { name = "python-dateutil" },
     { name = "python-magic" },
+    { name = "python-multipart" },
     { name = "sqlalchemy", extra = ["asyncio"] },
     { name = "uvicorn", extra = ["standard"] },
 ]
@@ -550,10 +562,12 @@ ocr = [
 
 [package.metadata]
 requires-dist = [
+    { name = "aiofiles", specifier = ">=24.1" },
     { name = "alembic", specifier = ">=1.14" },
     { name = "asyncpg", specifier = ">=0.30" },
     { name = "fastapi", specifier = ">=0.115" },
     { name = "httpx", specifier = ">=0.27" },
+    { name = "jinja2", specifier = ">=3.1" },
     { name = "mypy", marker = "extra == 'dev'", specifier = ">=1.13" },
     { name = "pillow", specifier = ">=10.2,<11.0" },
     { name = "pydantic", specifier = ">=2.9" },
@@ -564,6 +578,7 @@ requires-dist = [
     { name = "pytest-httpx", marker = "extra == 'dev'", specifier = ">=0.32" },
     { name = "python-dateutil", specifier = ">=2.9" },
     { name = "python-magic", specifier = ">=0.4.27" },
+    { name = "python-multipart", specifier = ">=0.0.12" },
     { name = "ruff", marker = "extra == 'dev'", specifier = ">=0.8" },
     { name = "sqlalchemy", extras = ["asyncio"], specifier = ">=2.0.36" },
     { name = "surya-ocr", marker = "extra == 'ocr'", specifier = ">=0.17,<0.18" },
@@ -1350,6 +1365,15 @@ wheels = [
     { url = "https://files.pythonhosted.org/packages/6c/73/9f872cb81fc5c3bb48f7227872c28975f998f3e7c2b1c16e95e6432bbb90/python_magic-0.4.27-py2.py3-none-any.whl", hash = "sha256:c212960ad306f700aa0d01e5d7a325d20548ff97eb9920dcd29513174f0294d3", size = 13840, upload-time = "2022-06-07T20:16:57.763Z" },
 ]
 
+[[package]]
+name = "python-multipart"
+version = "0.0.26"
+source = { registry = "https://pypi.org/simple" }
+sdist = { url = "https://files.pythonhosted.org/packages/88/71/b145a380824a960ebd60e1014256dbb7d2253f2316ff2d73dfd8928ec2c3/python_multipart-0.0.26.tar.gz", hash = "sha256:08fadc45918cd615e26846437f50c5d6d23304da32c341f289a617127b081f17", size = 43501, upload-time = "2026-04-10T14:09:59.473Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/9a/22/f1925cdda983ab66fc8ec6ec8014b959262747e58bdca26a4e3d1da29d56/python_multipart-0.0.26-py3-none-any.whl", hash = "sha256:c0b169f8c4484c13b0dcf2ef0ec3a4adb255c4b7d18d8e420477d2b1dd03f185", size = 28847, upload-time = "2026-04-10T14:09:58.131Z" },
+]
+
 [[package]]
 name = "pyyaml"
 version = "6.0.3"