diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md
index b3e00a27..59b01cf9 100644
--- a/CONTRIBUTING.md
+++ b/CONTRIBUTING.md
@@ -219,6 +219,40 @@ uv run python -m src.deriver     # background worker
 Everything Python goes through `uv run`. Redis is optional for local development; without it
 caching is simply disabled.
 
+### Running without a model provider
+
+`src/mock_provider/` is a deterministic, OpenAI-compatible endpoint, so you can run the full
+stack with no provider account, no API key, and no spend. It answers `/v1/chat/completions`
+and `/v1/embeddings` with obviously-synthetic content derived from the request, and the same
+request always produces the same response. Run it from the standard image or the repo:
+
+```bash
+uv run fastapi run --host 0.0.0.0 --port 8106 src/mock_provider/main.py
+```
+
+Then point Honcho at it. All three variables are required:
+
+```bash
+export LLM_OPENAI_API_KEY=sandbox          # gates client construction, not just auth
+export LLM_OPENAI_BASE_URL=http://localhost:8106/v1
+export EMBEDDING_MODEL_CONFIG__OVERRIDES__BASE_URL=http://localhost:8106/v1
+```
+
+Embeddings resolve through a separate client that reads the base URL only from the per-module
+override, so without the third variable your embedding calls go to `api.openai.com` for real.
+Do not set any per-module credential override (`..._OVERRIDES__API_KEY` / `API_KEY_ENV`) —
+that makes the module ignore the global base URL.
+
+Two things to know:
+
+- **A repo `.env` beats your exported environment.** `src/config.py` calls
+  `load_dotenv(override=True)` at import, so a stale `.env` silently wins over the variables
+  above. Set `PYTHON_DOTENV_DISABLED=1` (and `HONCHO_CONFIG_TOML_DISABLED=1` for a local
+  `config.toml`) when you need the environment to be the only input.
+- **Mock embeddings are hash-derived and carry no semantic similarity.** Two paraphrases are as
+  far apart as two unrelated strings. Recall against this provider must use lexical/full-text
+  search; anything asserting on vector ranking needs a real embedding provider.
+
 ## Making the change
 
 ### Branches and commits
diff --git a/src/mock_provider/__init__.py b/src/mock_provider/__init__.py
new file mode 100644
index 00000000..3a085944
--- /dev/null
+++ b/src/mock_provider/__init__.py
@@ -0,0 +1 @@
+"""Deterministic OpenAI-compatible provider for local and CI use."""
diff --git a/src/mock_provider/chat.py b/src/mock_provider/chat.py
new file mode 100644
index 00000000..df79a507
--- /dev/null
+++ b/src/mock_provider/chat.py
@@ -0,0 +1,212 @@
+"""OpenAI-compatible ``/chat/completions``, answered without inference."""
+
+from __future__ import annotations
+
+import hashlib
+import json
+import re
+from collections.abc import AsyncIterator
+from typing import Any
+
+from fastapi import APIRouter, Request
+from fastapi.responses import StreamingResponse
+
+from src.mock_provider.coerce import as_dict, as_list, as_str
+from src.mock_provider.schema_gen import generate
+
+router = APIRouter(tags=["mock-provider"])
+
+# Honcho's json_object mode injects the schema into the prompt text rather than
+# into response_format (see _apply_json_object_mode in the OpenAI backend), so
+# the only machine-readable copy of the schema is inside a message.
+_SCHEMA_HINT = re.compile(r"schema:\s*(\{)", re.IGNORECASE)
+
+
+def _completion_id(payload: dict[str, Any]) -> str:
+    """Stable id, so a replayed request is byte-identical."""
+    digest = hashlib.sha256(
+        json.dumps(payload, sort_keys=True, default=str).encode()
+    ).hexdigest()
+    return f"chatcmpl-mock-{digest[:24]}"
+
+
+def _extract_balanced_json(text: str, start: int) -> dict[str, Any] | None:
+    """Read one balanced ``{...}`` beginning at ``start`` and parse it.
+
+    A plain regex cannot do this — a JSON Schema contains nested objects, and
+    braces inside string literals must not count toward the depth.
+    """
+    depth = 0
+    in_string = False
+    escaped = False
+    for index in range(start, len(text)):
+        char = text[index]
+        if in_string:
+            if escaped:
+                escaped = False
+            elif char == "\\":
+                escaped = True
+            elif char == '"':
+                in_string = False
+            continue
+        if char == '"':
+            in_string = True
+        elif char == "{":
+            depth += 1
+        elif char == "}":
+            depth -= 1
+            if depth == 0:
+                try:
+                    parsed = json.loads(text[start : index + 1])
+                except json.JSONDecodeError:
+                    return None
+                return as_dict(parsed)
+    return None
+
+
+def _schema_from_messages(messages: list[Any]) -> dict[str, Any] | None:
+    """Recover an injected schema from the prompt, for json_object mode."""
+    for message in reversed(messages):
+        message_dict = as_dict(message)
+        if message_dict is None:
+            continue
+        content = as_str(message_dict.get("content"))
+        if content is None:
+            continue
+        for match in _SCHEMA_HINT.finditer(content):
+            candidate = _extract_balanced_json(content, match.start(1))
+            if candidate and ("properties" in candidate or "$defs" in candidate):
+                return candidate
+    return None
+
+
+def _response_content(payload: dict[str, Any]) -> str:
+    """The assistant message body: schema-conforming JSON, or prose."""
+    response_format = as_dict(payload.get("response_format"))
+    messages = as_list(payload.get("messages")) or []
+
+    if response_format is not None:
+        kind = as_str(response_format.get("type"))
+        if kind == "json_schema":
+            wrapper = as_dict(response_format.get("json_schema"))
+            if wrapper is not None:
+                schema = as_dict(wrapper.get("schema"))
+                if schema is not None:
+                    return json.dumps(generate(schema))
+            # A json_schema request whose schema we cannot read must not fall
+            # through to prose — that is the silent-empty failure this mock
+            # exists to avoid. An empty object at least parses.
+            return "{}"
+        if kind == "json_object":
+            schema = _schema_from_messages(messages)
+            return json.dumps(generate(schema)) if schema else "{}"
+
+    return (
+        "[mock] This is a synthetic response from Honcho's mock provider. "
+        "No model was called."
+    )
+
+
+def _usage(payload: dict[str, Any], content: str) -> dict[str, int]:
+    """Rough token accounting, so cost telemetry has plausible numbers."""
+    messages = as_list(payload.get("messages")) or []
+    prompt_chars = 0
+    for message in messages:
+        message_dict = as_dict(message)
+        text = as_str(message_dict.get("content")) if message_dict else None
+        if text is not None:
+            prompt_chars += len(text)
+    prompt_tokens = max(1, prompt_chars // 4)
+    completion_tokens = max(1, len(content) // 4)
+    return {
+        "prompt_tokens": prompt_tokens,
+        "completion_tokens": completion_tokens,
+        "total_tokens": prompt_tokens + completion_tokens,
+    }
+
+
+def _created() -> int:
+    # Fixed rather than time-based: a mock that changes its output between
+    # identical calls defeats the point.
+    return 1577836800  # 2020-01-01T00:00:00Z
+
+
+async def _stream(
+    completion_id: str, model: str, content: str, usage: dict[str, int]
+) -> AsyncIterator[bytes]:
+    def chunk(payload: dict[str, Any]) -> bytes:
+        return f"data: {json.dumps(payload)}\n\n".encode()
+
+    base = {
+        "id": completion_id,
+        "object": "chat.completion.chunk",
+        "created": _created(),
+        "model": model,
+    }
+    yield chunk(
+        {
+            **base,
+            "choices": [
+                {
+                    "index": 0,
+                    "delta": {"role": "assistant", "content": ""},
+                    "finish_reason": None,
+                }
+            ],
+        }
+    )
+    yield chunk(
+        {
+            **base,
+            "choices": [
+                {"index": 0, "delta": {"content": content}, "finish_reason": None}
+            ],
+        }
+    )
+    yield chunk(
+        {
+            **base,
+            "choices": [{"index": 0, "delta": {}, "finish_reason": "stop"}],
+        }
+    )
+    # The backend requests stream_options.include_usage and ends the stream on
+    # the usage chunk, so it must come last and must carry choices: [].
+    yield chunk({**base, "choices": [], "usage": usage})
+    yield b"data: [DONE]\n\n"
+
+
+@router.post("/chat/completions")
+async def chat_completions(request: Request) -> Any:
+    payload = as_dict(await request.json()) or {}
+
+    model = as_str(payload.get("model")) or "mock-model"
+    content = _response_content(payload)
+    usage = _usage(payload, content)
+    completion_id = _completion_id(payload)
+
+    if payload.get("stream"):
+        return StreamingResponse(
+            _stream(completion_id, model, content, usage),
+            media_type="text/event-stream",
+        )
+
+    return {
+        "id": completion_id,
+        "object": "chat.completion",
+        "created": _created(),
+        "model": model,
+        "choices": [
+            {
+                "index": 0,
+                "message": {
+                    "role": "assistant",
+                    "content": content,
+                    "refusal": None,
+                    "tool_calls": None,
+                },
+                "logprobs": None,
+                "finish_reason": "stop",
+            }
+        ],
+        "usage": usage,
+    }
diff --git a/src/mock_provider/coerce.py b/src/mock_provider/coerce.py
new file mode 100644
index 00000000..6b5b4cac
--- /dev/null
+++ b/src/mock_provider/coerce.py
@@ -0,0 +1,34 @@
+"""Typed narrowing for values decoded from JSON.
+
+``isinstance(value, dict)`` on an ``Any`` narrows to ``dict[Unknown, Unknown]``,
+which spreads unknown types through everything downstream. These helpers narrow
+and pin the element types in one step.
+"""
+
+from __future__ import annotations
+
+from typing import Any, cast
+
+
+def as_dict(value: object) -> dict[str, Any] | None:
+    """The value as a JSON object, or None if it is not one."""
+    return cast("dict[str, Any]", value) if isinstance(value, dict) else None
+
+
+def as_list(value: object) -> list[Any] | None:
+    """The value as a JSON array, or None if it is not one."""
+    return cast("list[Any]", value) if isinstance(value, list) else None
+
+
+def as_str(value: object) -> str | None:
+    """The value as a JSON string, or None if it is not one."""
+    return value if isinstance(value, str) else None
+
+
+def as_int(value: object) -> int | None:
+    """The value as a JSON integer, or None if it is not one.
+
+    ``bool`` is excluded: it is an ``int`` subclass, and a JSON ``true`` reaching
+    a size or dimension field is a malformed request, not the number one.
+    """
+    return value if isinstance(value, int) and not isinstance(value, bool) else None
diff --git a/src/mock_provider/embeddings.py b/src/mock_provider/embeddings.py
new file mode 100644
index 00000000..522f6a5e
--- /dev/null
+++ b/src/mock_provider/embeddings.py
@@ -0,0 +1,103 @@
+"""OpenAI-compatible ``/embeddings``, answered from a content hash."""
+
+from __future__ import annotations
+
+import base64
+import hashlib
+import struct
+from typing import Any
+
+from fastapi import APIRouter, Request
+
+from src.mock_provider.coerce import as_dict, as_int, as_list, as_str
+
+router = APIRouter(tags=["mock-provider"])
+
+# Honcho's default. EmbeddingClient._validate_embedding_dimensions raises when a
+# vector comes back at the wrong width, and validate_embedding_schema refuses to
+# boot when the width disagrees with the pgvector column, so the request's own
+# `dimensions` is honoured whenever it is present.
+DEFAULT_DIMENSIONS = 1536
+
+
+def content_to_embedding(content: str, dimensions: int) -> list[float]:
+    """A deterministic vector for ``content``.
+
+    Identical input yields an identical vector, and different inputs differ —
+    which is what deduplication logic needs. It carries no semantic similarity:
+    two paraphrases are as far apart as two unrelated strings. Anything
+    asserting on ranking quality must not use this provider.
+
+    Mirrors ``_content_to_embedding`` in tests/conftest.py.
+    """
+    digest = hashlib.sha256(content.encode()).digest()
+    return [(digest[i % len(digest)] / 255.0) * 2 - 1 for i in range(dimensions)]
+
+
+def _encode_base64(vector: list[float]) -> str:
+    """Little-endian float32, which is what the OpenAI SDK decodes."""
+    return base64.b64encode(struct.pack(f"<{len(vector)}f", *vector)).decode()
+
+
+def _normalize_input(raw: object) -> list[str]:
+    """Accept every input shape the OpenAI embeddings API allows.
+
+    Token-array inputs (list[int] / list[list[int]]) are rendered back to a
+    stable string rather than rejected — the vector only has to be deterministic.
+    """
+    if raw is None:
+        return []
+    if (text := as_str(raw)) is not None:
+        return [text]
+
+    items = as_list(raw)
+    if items is None:
+        return [str(raw)]
+
+    # A flat list of ints is one tokenized input, not many single-token ones.
+    if items and all(as_int(item) is not None for item in items):
+        return [",".join(str(item) for item in items)]
+
+    texts: list[str] = []
+    for item in items:
+        if (item_text := as_str(item)) is not None:
+            texts.append(item_text)
+        elif (parts := as_list(item)) is not None:
+            texts.append(",".join(str(part) for part in parts))
+        else:
+            texts.append(str(item))
+    return texts
+
+
+@router.post("/embeddings")
+async def embeddings(request: Request) -> Any:
+    payload = as_dict(await request.json()) or {}
+
+    texts = _normalize_input(payload.get("input"))
+    requested = as_int(payload.get("dimensions"))
+    dimensions = requested if requested and requested > 0 else DEFAULT_DIMENSIONS
+
+    # The SDK omits encoding_format only when it wants base64 (it sets the
+    # parameter explicitly for float), so absent means base64.
+    encoding_format = as_str(payload.get("encoding_format")) or "base64"
+
+    data: list[dict[str, Any]] = []
+    for index, text in enumerate(texts):
+        vector = content_to_embedding(text, dimensions)
+        data.append(
+            {
+                "object": "embedding",
+                "index": index,
+                "embedding": (
+                    vector if encoding_format == "float" else _encode_base64(vector)
+                ),
+            }
+        )
+
+    prompt_tokens = max(1, sum(len(text) for text in texts) // 4)
+    return {
+        "object": "list",
+        "data": data,
+        "model": as_str(payload.get("model")) or "mock-embedding",
+        "usage": {"prompt_tokens": prompt_tokens, "total_tokens": prompt_tokens},
+    }
diff --git a/src/mock_provider/main.py b/src/mock_provider/main.py
new file mode 100644
index 00000000..154a62a3
--- /dev/null
+++ b/src/mock_provider/main.py
@@ -0,0 +1,59 @@
+"""A deterministic, OpenAI-compatible provider for local and CI use.
+
+Lets Honcho run with no model provider, no API key, and no spend. It answers
+``/v1/chat/completions`` and ``/v1/embeddings`` with obviously-synthetic content
+derived from the request, so the same request always produces the same response.
+
+Runs as its own service from the standard Honcho image:
+
+    fastapi run --host 0.0.0.0 src/mock_provider/main.py
+
+Point Honcho at it with three variables — all three are required:
+
+    LLM_OPENAI_API_KEY=sandbox            # gates client construction, not just auth
+    LLM_OPENAI_BASE_URL=http://mock-provider:8000/v1
+    EMBEDDING_MODEL_CONFIG__OVERRIDES__BASE_URL=http://mock-provider:8000/v1
+
+Embeddings resolve through a separate client that reads the base URL only from
+the per-module override, so without the third variable embedding calls go to
+api.openai.com for real. Do not set any per-module credential override
+(``..._OVERRIDES__API_KEY`` / ``API_KEY_ENV``) — that makes the module ignore the
+global base URL.
+
+Embeddings are hash-derived and carry no semantic similarity. Recall assertions
+against this provider must use lexical/full-text search, not vector ranking.
+"""
+
+from __future__ import annotations
+
+from typing import Any
+
+from fastapi import FastAPI
+
+from src.mock_provider import chat, embeddings
+
+app = FastAPI(
+    title="Honcho Mock Provider",
+    description="Deterministic OpenAI-compatible endpoint for local and CI use.",
+    version="1.0.0",
+)
+
+# Mounted at both prefixes so the base URL works with or without /v1.
+for _router in (chat.router, embeddings.router):
+    app.include_router(_router, prefix="/v1")
+    app.include_router(_router)
+
+
+@app.get("/health")
+async def health() -> dict[str, str]:
+    return {"status": "ok", "provider": "mock"}
+
+
+@app.get("/{path:path}")
+async def catch_all(path: str) -> dict[str, Any]:
+    """Answer any other GET, so a bare ``/`` works as a container healthcheck.
+
+    Deliberately GET-only: an unimplemented POST returns 405 rather than a
+    plausible-looking 200, so a missing endpoint fails loudly.
+    """
+    return {"object": "mock", "path": path, "detail": "mock provider placeholder"}
diff --git a/src/mock_provider/schema_gen.py b/src/mock_provider/schema_gen.py
new file mode 100644
index 00000000..eb0f2533
--- /dev/null
+++ b/src/mock_provider/schema_gen.py
@@ -0,0 +1,285 @@
+"""Generate a conforming instance from a JSON Schema.
+
+The deriver is a structured-output caller: it sends a schema and parses the
+reply back into a Pydantic model. A mock that answers with prose does not fail
+loudly — ``repair_response_model_json`` swallows the error and hands back an
+empty ``PromptRepresentation``, which reads as "the deriver found nothing"
+rather than "the mock is wrong". So generation is driven by the schema that was
+actually sent, ``$ref`` indirection and all.
+
+Values are derived from a hash of the property path, so the same schema always
+produces the same instance and two different fields never collide.
+"""
+
+from __future__ import annotations
+
+import hashlib
+from typing import Any
+
+from src.mock_provider.coerce import as_dict, as_int, as_list, as_str
+
+# Depth cap for self-referential schemas. Reasoning-tree models nest premises
+# inside conclusions, so a $ref cycle is normal input, not a malformed schema.
+MAX_DEPTH = 6
+
+_WORDS = (
+    "synthetic",
+    "placeholder",
+    "mock",
+    "sample",
+    "fixture",
+    "stub",
+    "generated",
+    "example",
+    "inert",
+    "dummy",
+)
+
+
+def _seed(path: str) -> int:
+    return int.from_bytes(hashlib.sha256(path.encode()).digest()[:8], "big")
+
+
+def _phrase(path: str, words: int = 6) -> str:
+    """An obviously-synthetic sentence, stable for a given path."""
+    seed = _seed(path)
+    picked = [_WORDS[(seed >> (i * 5)) % len(_WORDS)] for i in range(words)]
+    return f"[mock] {' '.join(picked)}"
+
+
+def _resolve(schema: dict[str, Any], root: dict[str, Any]) -> dict[str, Any]:
+    """Follow a local ``$ref`` chain to the schema it points at.
+
+    Only local refs are supported: the mock never fetches over the network, and
+    Pydantic's ``model_json_schema()`` only ever emits ``#/$defs/...``.
+    """
+    seen: set[str] = set()
+    current = schema
+    while "$ref" in current:
+        ref = as_str(current["$ref"])
+        if ref is None or not ref.startswith("#/") or ref in seen:
+            return {}
+        seen.add(ref)
+
+        target: dict[str, Any] | None = root
+        for part in ref[2:].split("/"):
+            if target is None or part not in target:
+                return {}
+            target = as_dict(target[part])
+        if target is None:
+            return {}
+        current = target
+    return current
+
+
+def _merge_all_of(schema: dict[str, Any], root: dict[str, Any]) -> dict[str, Any]:
+    """Flatten ``allOf`` into the parent so one pass can read properties off it."""
+    branches = as_list(schema.get("allOf"))
+    if branches is None:
+        return schema
+
+    merged: dict[str, Any] = {k: v for k, v in schema.items() if k != "allOf"}
+    for branch in branches:
+        resolved_branch = as_dict(branch)
+        if resolved_branch is None:
+            continue
+        resolved = _resolve(resolved_branch, root)
+        for key, value in resolved.items():
+            if key == "properties":
+                properties = as_dict(value)
+                if properties is not None:
+                    existing = as_dict(merged.get("properties")) or {}
+                    merged["properties"] = {**properties, **existing}
+                    continue
+            if key == "required":
+                required = as_list(value)
+                if required is not None:
+                    previous = as_list(merged.get("required")) or []
+                    merged["required"] = list({*previous, *required})
+                    continue
+            merged.setdefault(key, value)
+    return merged
+
+
+def _infer_type(schema: dict[str, Any]) -> str:
+    """Best-effort type when the schema omits an explicit ``type``."""
+    declared = schema.get("type")
+    if (name := as_str(declared)) is not None:
+        return name
+    if (names := as_list(declared)) is not None:
+        # Nullable unions arrive as ["string", "null"]; prefer the real type.
+        for candidate in names:
+            if (candidate_name := as_str(candidate)) and candidate_name != "null":
+                return candidate_name
+        return "null"
+    if "properties" in schema:
+        return "object"
+    if "items" in schema:
+        return "array"
+    return "string"
+
+
+def generate(schema: dict[str, Any], root: dict[str, Any] | None = None) -> Any:
+    """Build a value satisfying ``schema``.
+
+    ``root`` carries the document that ``$ref`` resolves against; it defaults to
+    ``schema`` itself, which is the shape Pydantic emits.
+    """
+    return _generate(schema, root if root is not None else schema, "$", 0)
+
+
+def _generate(
+    schema: dict[str, Any], root: dict[str, Any], path: str, depth: int
+) -> Any:
+    resolved = _merge_all_of(_resolve(schema, root), root)
+
+    if "const" in resolved:
+        return resolved["const"]
+
+    enum = as_list(resolved.get("enum"))
+    if enum:
+        return enum[_seed(path) % len(enum)]
+
+    if depth >= MAX_DEPTH and "default" in resolved:
+        return resolved["default"]
+
+    for key in ("anyOf", "oneOf"):
+        branches = as_list(resolved.get(key))
+        if branches:
+            return _generate(_pick_branch(branches, root, depth), root, path, depth)
+
+    kind = _infer_type(resolved)
+    if kind == "object":
+        return _generate_object(resolved, root, path, depth)
+    if kind == "array":
+        return _generate_array(resolved, root, path, depth)
+    if kind == "integer":
+        return _bounded_int(resolved, path)
+    if kind == "number":
+        return float(_bounded_int(resolved, path))
+    if kind == "boolean":
+        return _seed(path) % 2 == 0
+    if kind == "null":
+        return None
+    return _generate_string(resolved, path)
+
+
+def _pick_branch(
+    branches: list[Any], root: dict[str, Any], depth: int
+) -> dict[str, Any]:
+    """Choose a union member, preferring a non-null one.
+
+    Past the depth cap the order flips: a nullable recursive field terminates on
+    ``null`` instead of nesting another level.
+    """
+    resolved: list[dict[str, Any]] = []
+    for branch in branches:
+        branch_dict = as_dict(branch)
+        if branch_dict is not None:
+            resolved.append(_resolve(branch_dict, root))
+    if not resolved:
+        return {}
+
+    if depth >= MAX_DEPTH:
+        nulls = [b for b in resolved if _infer_type(b) == "null"]
+        if nulls:
+            return nulls[0]
+    non_null = [b for b in resolved if _infer_type(b) != "null"]
+    return non_null[0] if non_null else resolved[0]
+
+
+def _generate_object(
+    schema: dict[str, Any], root: dict[str, Any], path: str, depth: int
+) -> dict[str, Any]:
+    properties = as_dict(schema.get("properties"))
+    if properties is None:
+        return {}
+
+    # OpenAI structured outputs run in strict mode, where every property is
+    # required. Emitting the full property set satisfies both strict and loose
+    # schemas, so `required` is only consulted to decide what to drop once the
+    # depth cap has been hit.
+    declared_required = as_list(schema.get("required"))
+    required: set[str] = (
+        {name for name in (as_str(item) for item in declared_required) if name}
+        if declared_required is not None
+        else set(properties)
+    )
+
+    result: dict[str, Any] = {}
+    for name, subschema in properties.items():
+        if depth >= MAX_DEPTH and name not in required:
+            continue
+        child = as_dict(subschema)
+        if child is None:
+            continue
+        result[name] = _generate(child, root, f"{path}.{name}", depth + 1)
+    return result
+
+
+def _generate_array(
+    schema: dict[str, Any], root: dict[str, Any], path: str, depth: int
+) -> list[Any]:
+    items = as_dict(schema.get("items"))
+    min_items = as_int(schema.get("minItems"))
+    max_items = as_int(schema.get("maxItems"))
+
+    count = 2
+    if min_items is not None:
+        count = max(count, min_items)
+    if max_items is not None:
+        count = min(count, max_items)
+    if depth >= MAX_DEPTH:
+        count = min_items or 0
+    if items is None or count <= 0:
+        return []
+
+    return [_generate(items, root, f"{path}[{i}]", depth + 1) for i in range(count)]
+
+
+def _generate_string(schema: dict[str, Any], path: str) -> str:
+    fmt = as_str(schema.get("format"))
+    if fmt == "date-time":
+        return "2020-01-01T00:00:00Z"
+    if fmt == "date":
+        return "2020-01-01"
+    if fmt == "uuid":
+        stem = hashlib.sha256(path.encode()).hexdigest()[:8]
+        return f"{stem}-0000-4000-8000-000000000000"
+    if fmt in ("uri", "url"):
+        return "https://mock.invalid/placeholder"
+    if fmt == "email":
+        return "placeholder@mock.invalid"
+
+    value = _phrase(path)
+    min_length = as_int(schema.get("minLength"))
+    max_length = as_int(schema.get("maxLength"))
+    if min_length is not None and len(value) < min_length:
+        value = value.ljust(min_length, "x")
+    if max_length is not None and len(value) > max_length:
+        value = value[:max_length]
+    return value
+
+
+def _bounded_int(schema: dict[str, Any], path: str) -> int:
+    low = as_int(schema.get("minimum"))
+    if (
+        low is None
+        and (exclusive := as_int(schema.get("exclusiveMinimum"))) is not None
+    ):
+        low = exclusive + 1
+    high = as_int(schema.get("maximum"))
+    if (
+        high is None
+        and (exclusive := as_int(schema.get("exclusiveMaximum"))) is not None
+    ):
+        high = exclusive - 1
+
+    if low is not None and high is not None:
+        span = high - low
+        return low + (_seed(path) % (span + 1) if span > 0 else 0)
+    if low is not None:
+        return low + (_seed(path) % 8)
+    if high is not None:
+        return high - (_seed(path) % 8)
+    return _seed(path) % 100
diff --git a/tests/conftest.py b/tests/conftest.py
index 090d5395..411f5210 100644
--- a/tests/conftest.py
+++ b/tests/conftest.py
@@ -91,6 +91,9 @@ _RUNTIME_MOCK_TEST_BLOCKLIST_PREFIXES = (
     # Pure JWT scope tests — operate on src.security directly, no DB needed.
     "tests/test_security.py",
     "tests/test_generate_jwt_script.py",
+    # The mock provider is a standalone ASGI app with no database or LLM of its
+    # own; the runtime mocks would patch the very seams it exists to replace.
+    "tests/mock_provider/",
 )
 
 _LIVE_LLM_MARKER = "live_llm"
diff --git a/tests/mock_provider/test_honcho_contract.py b/tests/mock_provider/test_honcho_contract.py
new file mode 100644
index 00000000..6c362603
--- /dev/null
+++ b/tests/mock_provider/test_honcho_contract.py
@@ -0,0 +1,221 @@
+"""Drive Honcho's real provider clients against the mock over ASGI.
+
+The unit tests assert the mock's own output. These assert the hop that actually
+matters: ``OpenAIBackend`` and ``EmbeddingClient`` — the production classes,
+unpatched — talking to the mock through the genuine OpenAI SDK, including the
+``strict: true`` json_schema transform that ``chat.completions.parse()`` applies
+on the way out and the Pydantic validation it applies on the way back.
+"""
+
+from __future__ import annotations
+
+from typing import Any
+
+import httpx
+import pytest
+from openai import AsyncOpenAI
+
+from src.config import EmbeddingModelConfig
+from src.embedding_client import _EmbeddingClient  # pyright: ignore[reportPrivateUsage]
+from src.llm.backends.openai import OpenAIBackend
+from src.mock_provider.embeddings import content_to_embedding
+from src.mock_provider.main import app
+from src.utils.representation import PromptRepresentation
+
+MESSAGES: list[dict[str, Any]] = [
+    {"role": "user", "content": "I switched the service from pip to uv last week."}
+]
+
+
+@pytest.fixture
+def openai_client() -> AsyncOpenAI:
+    return AsyncOpenAI(
+        api_key="sandbox",
+        base_url="http://mock-provider.invalid/v1",
+        http_client=httpx.AsyncClient(transport=httpx.ASGITransport(app=app)),
+    )
+
+
+@pytest.mark.asyncio
+async def test_backend_parses_the_deriver_response_model(
+    openai_client: AsyncOpenAI,
+) -> None:
+    """The production path: parse() with a Pydantic response_format."""
+    backend = OpenAIBackend(openai_client)
+
+    result = await backend.complete(
+        model="mock-model",
+        messages=MESSAGES,
+        max_tokens=512,
+        response_format=PromptRepresentation,
+    )
+
+    assert isinstance(result.content, PromptRepresentation)
+    # An empty explicit list is what a prose-answering mock silently produces,
+    # so it is the specific thing worth asserting against.
+    assert result.content.explicit
+    assert result.output_tokens > 0
+
+
+@pytest.mark.asyncio
+async def test_backend_json_object_mode_recovers_the_schema(
+    openai_client: AsyncOpenAI,
+) -> None:
+    """json_object mode carries the schema in the prompt, not response_format."""
+    backend = OpenAIBackend(openai_client)
+
+    result = await backend.complete(
+        model="mock-model",
+        messages=MESSAGES,
+        max_tokens=512,
+        response_format=PromptRepresentation,
+        extra_params={"structured_output_mode": "json_object"},
+    )
+
+    assert isinstance(result.content, PromptRepresentation)
+    assert result.content.explicit
+
+
+@pytest.mark.asyncio
+async def test_backend_with_tools_uses_json_schema_and_still_parses(
+    openai_client: AsyncOpenAI,
+) -> None:
+    """Non-strict tools force create() + explicit json_schema instead of parse()."""
+    backend = OpenAIBackend(openai_client)
+
+    result = await backend.complete(
+        model="mock-model",
+        messages=MESSAGES,
+        max_tokens=512,
+        response_format=PromptRepresentation,
+        tools=[
+            {
+                "type": "function",
+                "function": {
+                    "name": "search_memory",
+                    "description": "Search memory",
+                    "parameters": {
+                        "type": "object",
+                        "properties": {"query": {"type": "string"}},
+                    },
+                },
+            }
+        ],
+    )
+
+    assert isinstance(result.content, PromptRepresentation)
+    assert result.content.explicit
+
+
+@pytest.mark.asyncio
+async def test_backend_plain_completion(openai_client: AsyncOpenAI) -> None:
+    backend = OpenAIBackend(openai_client)
+
+    result = await backend.complete(
+        model="mock-model", messages=MESSAGES, max_tokens=128
+    )
+
+    assert isinstance(result.content, str)
+    assert "[mock]" in result.content
+    assert result.finish_reason == "stop"
+
+
+@pytest.mark.asyncio
+async def test_backend_stream_yields_content_then_a_usage_terminator(
+    openai_client: AsyncOpenAI,
+) -> None:
+    backend = OpenAIBackend(openai_client)
+
+    chunks = [
+        chunk
+        async for chunk in backend.stream(
+            model="mock-model", messages=MESSAGES, max_tokens=128
+        )
+    ]
+
+    assert "[mock]" in "".join(chunk.content or "" for chunk in chunks)
+
+    terminator = chunks[-1]
+    assert terminator.is_done
+    assert terminator.finish_reason == "stop"
+    # None here means the stream ended without a usage chunk, which is the
+    # failure mode when stream_options.include_usage goes unanswered.
+    assert terminator.output_tokens is not None
+    assert terminator.output_tokens > 0
+
+
+def _embedding_client(dimensions: int, encoding_format: str) -> _EmbeddingClient:
+    # The public EmbeddingClient is a settings-driven singleton wrapper; the
+    # transport behaviour under test lives on the implementation it wraps.
+    return _EmbeddingClient(
+        EmbeddingModelConfig(
+            model="text-embedding-3-small",
+            transport="openai",
+            api_key="sandbox",
+            base_url="http://mock-provider.invalid/v1",
+        ),
+        vector_dimensions=dimensions,
+        max_input_tokens=8192,
+        max_tokens_per_request=300000,
+        send_dimensions=True,
+        encoding_format=encoding_format,  # pyright: ignore[reportArgumentType]
+    )
+
+
+@pytest.fixture(autouse=True)
+def _route_embedding_client_over_asgi(  # pyright: ignore[reportUnusedFunction]
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """Give the embedding client's AsyncOpenAI an ASGI transport.
+
+    EmbeddingClient builds its own client internally, so the transport has to be
+    injected at construction rather than passed in.
+    """
+    original = AsyncOpenAI.__init__
+
+    def patched(self: AsyncOpenAI, *args: Any, **kwargs: Any) -> None:
+        kwargs.setdefault(
+            "http_client",
+            httpx.AsyncClient(transport=httpx.ASGITransport(app=app)),
+        )
+        original(self, *args, **kwargs)
+
+    monkeypatch.setattr(AsyncOpenAI, "__init__", patched)
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("encoding_format", ["float", "base64"])
+async def test_embedding_client_round_trip(encoding_format: str) -> None:
+    """Covers both wire encodings; base64 is what the SDK uses by default."""
+    client = _embedding_client(1536, encoding_format)
+
+    vector = await client.embed("I switched the service from pip to uv.")
+
+    # _validate_embedding_dimensions raises on a width mismatch, so reaching
+    # here already proves the width is right; assert the values too.
+    assert len(vector) == 1536
+    assert vector == pytest.approx(  # pyright: ignore[reportUnknownMemberType]
+        content_to_embedding("I switched the service from pip to uv.", 1536),
+        abs=1e-6,
+    )
+
+
+@pytest.mark.asyncio
+async def test_embedding_client_honours_a_non_default_dimension() -> None:
+    """send_dimensions=True forwards `dimensions`; the mock must obey it."""
+    client = _embedding_client(256, "float")
+
+    assert len(await client.embed("hello")) == 256
+
+
+@pytest.mark.asyncio
+async def test_embedding_client_batches() -> None:
+    """_validate_embedding_count rejects a mismatched count."""
+    client = _embedding_client(1536, "float")
+    texts = [f"observation number {index}" for index in range(12)]
+
+    vectors = await client.simple_batch_embed(texts)
+
+    assert len(vectors) == len(texts)
+    assert all(len(vector) == 1536 for vector in vectors)
+    assert len({tuple(vector) for vector in vectors}) == len(texts)
diff --git a/tests/mock_provider/test_mock_provider.py b/tests/mock_provider/test_mock_provider.py
new file mode 100644
index 00000000..f7863c2d
--- /dev/null
+++ b/tests/mock_provider/test_mock_provider.py
@@ -0,0 +1,439 @@
+"""Contract tests for the mock provider.
+
+The failure this guards against is silent: when the mock answers a structured
+request with something the deriver cannot parse, ``repair_response_model_json``
+falls back to an empty ``PromptRepresentation`` and the run looks like "the
+deriver found nothing" rather than "the mock is broken". So the assertions here
+are about parseability against real Honcho models, not about response shape.
+"""
+
+from __future__ import annotations
+
+import base64
+import hashlib
+import json
+import struct
+from collections.abc import Callable
+from typing import Any
+
+import pytest
+from fastapi.testclient import TestClient
+from pydantic import BaseModel, Field
+
+from src.mock_provider.embeddings import content_to_embedding
+from src.mock_provider.main import app
+from src.mock_provider.schema_gen import generate
+from src.utils.representation import PromptRepresentation
+
+# A $ref/$defs schema, which is what Pydantic emits for any nested model and the
+# indirection a naive generator silently drops.
+PROBE_SCHEMA: dict[str, Any] = {
+    "$defs": {
+        "Item": {
+            "type": "object",
+            "properties": {
+                "name": {"type": "string"},
+                "count": {"type": "integer", "minimum": 1, "maximum": 5},
+            },
+            "required": ["name", "count"],
+        }
+    },
+    "type": "object",
+    "properties": {
+        "label": {"type": "string"},
+        "items": {"type": "array", "items": {"$ref": "#/$defs/Item"}},
+    },
+    "required": ["label", "items"],
+}
+
+
+class ProbeItem(BaseModel):
+    name: str
+    count: int = Field(ge=1, le=5)
+
+
+class Probe(BaseModel):
+    label: str
+    items: list[ProbeItem]
+
+
+@pytest.fixture
+def client() -> TestClient:
+    return TestClient(app)
+
+
+def _post_chat(client: TestClient, **payload: Any) -> dict[str, Any]:
+    payload.setdefault("model", "mock-model")
+    payload.setdefault("messages", [{"role": "user", "content": "hello"}])
+    response = client.post("/v1/chat/completions", json=payload)
+    assert response.status_code == 200, response.text
+    return response.json()
+
+
+def _json_schema_format(schema: dict[str, Any], name: str) -> dict[str, Any]:
+    return {
+        "type": "json_schema",
+        "json_schema": {"name": name, "schema": schema, "strict": True},
+    }
+
+
+# --- structured output ------------------------------------------------------
+
+
+def test_json_schema_request_round_trips_into_its_pydantic_model(
+    client: TestClient,
+) -> None:
+    body = _post_chat(
+        client, response_format=_json_schema_format(PROBE_SCHEMA, "Probe")
+    )
+    content = body["choices"][0]["message"]["content"]
+
+    probe = Probe.model_validate_json(content)
+    assert probe.label
+    assert probe.items, "$ref array must not come back empty"
+    assert all(1 <= item.count <= 5 for item in probe.items)
+
+
+def test_deriver_response_model_round_trips() -> None:
+    """The real model the deriver parses, not a stand-in."""
+    schema = PromptRepresentation.model_json_schema()
+    content = json.dumps(generate(schema))
+
+    representation = PromptRepresentation.model_validate_json(content)
+    assert (
+        representation.explicit
+    ), "an empty explicit list is exactly the silent failure this mock avoids"
+
+
+def test_json_schema_response_is_never_prose(client: TestClient) -> None:
+    body = _post_chat(
+        client, response_format=_json_schema_format(PROBE_SCHEMA, "Probe")
+    )
+    json.loads(body["choices"][0]["message"]["content"])
+
+
+def test_unreadable_json_schema_still_returns_parseable_json(
+    client: TestClient,
+) -> None:
+    body = _post_chat(
+        client,
+        response_format={"type": "json_schema", "json_schema": {"name": "Broken"}},
+    )
+    assert json.loads(body["choices"][0]["message"]["content"]) == {}
+
+
+def test_json_object_mode_recovers_the_schema_from_the_prompt(
+    client: TestClient,
+) -> None:
+    """json_object mode puts the schema in the prompt, not in response_format."""
+    body = _post_chat(
+        client,
+        messages=[
+            {"role": "user", "content": "Extract facts."},
+            {
+                "role": "user",
+                "content": "Respond with valid JSON matching this schema:\n"
+                + json.dumps(PROBE_SCHEMA),
+            },
+        ],
+        response_format={"type": "json_object"},
+    )
+    Probe.model_validate_json(body["choices"][0]["message"]["content"])
+
+
+def test_json_object_mode_without_a_schema_returns_an_empty_object(
+    client: TestClient,
+) -> None:
+    body = _post_chat(client, response_format={"type": "json_object"})
+    assert json.loads(body["choices"][0]["message"]["content"]) == {}
+
+
+def test_plain_request_returns_prose(client: TestClient) -> None:
+    body = _post_chat(client)
+    content = body["choices"][0]["message"]["content"]
+    assert "[mock]" in content
+    with pytest.raises(json.JSONDecodeError):
+        json.loads(content)
+
+
+def test_tools_request_does_not_emit_tool_calls(client: TestClient) -> None:
+    """The tool loop must terminate; a mock that calls tools would spin."""
+    body = _post_chat(
+        client,
+        tools=[
+            {
+                "type": "function",
+                "function": {"name": "search_memory", "parameters": {}},
+            }
+        ],
+    )
+    assert body["choices"][0]["message"]["tool_calls"] is None
+    assert body["choices"][0]["finish_reason"] == "stop"
+
+
+def test_identical_requests_are_byte_identical(client: TestClient) -> None:
+    payload: dict[str, Any] = {
+        "model": "mock-model",
+        "messages": [{"role": "user", "content": "determinism"}],
+        "response_format": _json_schema_format(PROBE_SCHEMA, "Probe"),
+    }
+    first = client.post("/v1/chat/completions", json=payload).json()
+    second = client.post("/v1/chat/completions", json=payload).json()
+    assert first == second
+
+
+def test_usage_is_reported(client: TestClient) -> None:
+    body = _post_chat(client)
+    usage = body["usage"]
+    assert usage["total_tokens"] == usage["prompt_tokens"] + usage["completion_tokens"]
+    assert usage["completion_tokens"] > 0
+
+
+# --- schema generation edge cases -------------------------------------------
+
+
+def test_recursive_schema_terminates() -> None:
+    """Reasoning trees nest premises inside conclusions, so cycles are normal."""
+    schema: dict[str, Any] = {
+        "$defs": {
+            "Node": {
+                "type": "object",
+                "properties": {
+                    "value": {"type": "string"},
+                    "child": {"anyOf": [{"$ref": "#/$defs/Node"}, {"type": "null"}]},
+                },
+                "required": ["value", "child"],
+            }
+        },
+        "$ref": "#/$defs/Node",
+    }
+    node: dict[str, Any] | None = generate(schema)
+    depth = 0
+    while node is not None and node.get("child") is not None:
+        node = node["child"]
+        depth += 1
+        assert depth < 50, "recursive schema did not terminate"
+
+
+_SCALAR_CASES: list[tuple[str, dict[str, Any], Callable[[Any], bool]]] = [
+    ("enum", {"type": "string", "enum": ["a", "b"]}, lambda v: v in ("a", "b")),
+    ("const", {"const": 7}, lambda v: v == 7),
+    ("boolean", {"type": "boolean"}, lambda v: isinstance(v, bool)),
+    ("null", {"type": "null"}, lambda v: v is None),
+    ("number", {"type": "number"}, lambda v: isinstance(v, float)),
+    ("nullable-union", {"type": ["string", "null"]}, lambda v: isinstance(v, str)),
+    ("pinned-int", {"type": "integer", "minimum": 3, "maximum": 3}, lambda v: v == 3),
+    (
+        "exclusive-bounds",
+        {"type": "integer", "exclusiveMinimum": 1, "exclusiveMaximum": 3},
+        lambda v: v == 2,
+    ),
+    (
+        "date-time",
+        {"type": "string", "format": "date-time"},
+        lambda v: str(v).endswith("Z"),
+    ),
+    ("min-length", {"type": "string", "minLength": 400}, lambda v: len(v) >= 400),
+    ("max-length", {"type": "string", "maxLength": 4}, lambda v: len(v) == 4),
+    (
+        "min-items",
+        {"type": "array", "items": {"type": "string"}, "minItems": 3},
+        lambda v: len(v) >= 3,
+    ),
+    (
+        "max-items",
+        {"type": "array", "items": {"type": "string"}, "maxItems": 1},
+        lambda v: len(v) == 1,
+    ),
+]
+
+
+@pytest.mark.parametrize(
+    ("schema", "check"),
+    [(schema, check) for _, schema, check in _SCALAR_CASES],
+    ids=[name for name, _, _ in _SCALAR_CASES],
+)
+def test_scalar_schema_forms(
+    schema: dict[str, Any], check: Callable[[Any], bool]
+) -> None:
+    assert check(generate(schema))
+
+
+def test_all_of_is_flattened() -> None:
+    schema: dict[str, Any] = {
+        "allOf": [
+            {
+                "type": "object",
+                "properties": {"a": {"type": "string"}},
+                "required": ["a"],
+            },
+            {
+                "type": "object",
+                "properties": {"b": {"type": "integer"}},
+                "required": ["b"],
+            },
+        ]
+    }
+    result = generate(schema)
+    assert isinstance(result["a"], str)
+    assert isinstance(result["b"], int)
+
+
+def test_generation_is_stable_across_calls() -> None:
+    assert generate(PROBE_SCHEMA) == generate(PROBE_SCHEMA)
+
+
+def test_sibling_fields_of_the_same_type_differ() -> None:
+    """Path-seeded, so a schema of identical fields is not all one value."""
+    schema: dict[str, Any] = {
+        "type": "object",
+        "properties": {
+            "first": {"type": "string"},
+            "second": {"type": "string"},
+        },
+        "required": ["first", "second"],
+    }
+    result = generate(schema)
+    assert result["first"] != result["second"]
+
+
+# --- embeddings -------------------------------------------------------------
+
+
+def test_embeddings_default_to_1536_and_are_stable(client: TestClient) -> None:
+    payload = {
+        "model": "text-embedding-3-small",
+        "input": "hello",
+        "encoding_format": "float",
+    }
+    first = client.post("/v1/embeddings", json=payload)
+    assert first.status_code == 200, first.text
+    vector = first.json()["data"][0]["embedding"]
+
+    assert len(vector) == 1536
+    assert all(-1.0 <= value <= 1.0 for value in vector)
+    assert client.post("/v1/embeddings", json=payload).json() == first.json()
+
+
+def test_embeddings_honour_the_requested_dimension(client: TestClient) -> None:
+    """A width mismatch raises in EmbeddingClient and blocks startup."""
+    response = client.post(
+        "/v1/embeddings",
+        json={"input": "hello", "dimensions": 256, "encoding_format": "float"},
+    )
+    assert len(response.json()["data"][0]["embedding"]) == 256
+
+
+def test_different_inputs_give_different_vectors(client: TestClient) -> None:
+    response = client.post(
+        "/v1/embeddings",
+        json={"input": ["alpha", "beta"], "encoding_format": "float"},
+    )
+    data = response.json()["data"]
+    assert len(data) == 2
+    assert [item["index"] for item in data] == [0, 1]
+    assert data[0]["embedding"] != data[1]["embedding"]
+
+
+def test_batch_returns_one_embedding_per_input(client: TestClient) -> None:
+    """EmbeddingClient._validate_embedding_count rejects any other count."""
+    texts = [f"text-{index}" for index in range(17)]
+    response = client.post(
+        "/v1/embeddings", json={"input": texts, "encoding_format": "float"}
+    )
+    assert len(response.json()["data"]) == len(texts)
+
+
+def test_base64_is_the_default_encoding_and_decodes_to_the_float_vector(
+    client: TestClient,
+) -> None:
+    """The SDK omits encoding_format precisely when it wants base64."""
+    response = client.post("/v1/embeddings", json={"input": "hello"})
+    encoded = response.json()["data"][0]["embedding"]
+    assert isinstance(encoded, str)
+
+    raw = base64.b64decode(encoded)
+    decoded = list(struct.unpack(f"<{len(raw) // 4}f", raw))
+    assert len(decoded) == 1536
+    expected = content_to_embedding("hello", 1536)
+    assert decoded == pytest.approx(expected, abs=1e-6)  # pyright: ignore[reportUnknownMemberType]
+
+
+def test_embedding_matches_the_test_suite_helper() -> None:
+    """Kept in step with _content_to_embedding in tests/conftest.py.
+
+    Both must derive the same vector from the same text, so a suite that mocks
+    the embedding client in-process and one that talks to this provider over
+    HTTP agree on what a given string embeds to.
+    """
+    digest = hashlib.sha256(b"hello").digest()
+    expected = [(digest[i % len(digest)] / 255.0) * 2 - 1 for i in range(8)]
+
+    assert content_to_embedding("hello", 8) == pytest.approx(expected)  # pyright: ignore[reportUnknownMemberType]
+
+
+# --- routing ----------------------------------------------------------------
+
+
+def test_routes_are_mounted_with_and_without_the_v1_prefix(
+    client: TestClient,
+) -> None:
+    for path in ("/v1/chat/completions", "/chat/completions"):
+        response = client.post(
+            path, json={"model": "m", "messages": [{"role": "user", "content": "x"}]}
+        )
+        assert response.status_code == 200, path
+
+
+def test_unimplemented_post_returns_405_not_a_plausible_200(
+    client: TestClient,
+) -> None:
+    """A catch-all POST would make a missing endpoint look like it worked."""
+    assert client.post("/v1/completions", json={}).status_code == 405
+
+
+def test_health_and_catch_all_get(client: TestClient) -> None:
+    assert client.get("/health").json()["status"] == "ok"
+    assert client.get("/").status_code == 200
+
+
+# --- streaming --------------------------------------------------------------
+
+
+def test_stream_emits_content_then_a_final_usage_chunk(client: TestClient) -> None:
+    """The backend ends the stream on the usage chunk, so it must come last."""
+    with client.stream(
+        "POST",
+        "/v1/chat/completions",
+        json={
+            "model": "mock-model",
+            "messages": [{"role": "user", "content": "stream please"}],
+            "stream": True,
+            "stream_options": {"include_usage": True},
+        },
+    ) as response:
+        assert response.status_code == 200
+        lines = [
+            line[len("data: ") :]
+            for line in response.iter_lines()
+            if line.startswith("data: ")
+        ]
+
+    assert lines[-1] == "[DONE]"
+    chunks = [json.loads(line) for line in lines[:-1]]
+
+    content = "".join(
+        chunk["choices"][0]["delta"].get("content", "")
+        for chunk in chunks
+        if chunk["choices"]
+    )
+    assert "[mock]" in content
+
+    assert any(
+        chunk["choices"] and chunk["choices"][0]["finish_reason"] == "stop"
+        for chunk in chunks
+    )
+
+    usage_chunk = chunks[-1]
+    assert usage_chunk["usage"]["completion_tokens"] > 0
+    assert usage_chunk["choices"] == []