code_of_conquest_dnd/docs/superpowers/plans/2026-07-09-narrator-ollama-pipeline.md

Narrator / Ollama Pipeline (server) Implementation Plan

Status: ✅ COMPLETE (2026-07-10). All 8 tasks landed via subagent-driven development (per-task spec+quality review + opus whole-branch review). Merged --no-ff to dev as ddc00d4 (branch feature/narrator-ollama-pipeline, commits 42b3720..f7c51b7, deleted post-merge). Suite: 49 passed / 1 skipped (live smoke, --run-live), pristine. Live wire proven against real qwen3.5. Two conflicts caught in execution: the Plan-A test_endpoints loop was made hermetic after wiring (Task 7), and pytest.ini was merged not replaced to keep filterwarnings. Opus review found + fixed two seam-level robustness gaps: best-effort call logging (§13) and non-JSON 200 → ModelError (§12). Deferred follow-ups for later plans: prompts.system_prompt should raise on a missing --- delimiter; render_digest's trailing narrator instruction should move into the system prompt when a second role reuses the digest.

For agentic workers: REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (- [ ]) syntax for tracking.

Goal: Make /dm/narrate real — assemble the Narrator prompt + a curated canon-log digest, call Ollama, return the prose, log every call (§10), and signal failure honestly — building the reusable server-side model-call pipeline the other four roles later drop into.

Architecture: Layered modules under api/app/ — config (env), routing (role→model), prompts (system-prompt loader + digest renderer), ollama_client (httpx → Ollama /api/chat, one retry, typed ModelError), call_log (§10 JSON-lines), narrate (the Narrator service). main.py's /dm/narrate handler stays thin: 200 {"prose": …} or 502 {"detail": {"model_error": …}}. Mocks enter at the ollama_client boundary.

Tech Stack: Python 3.12, FastAPI, httpx (non-streaming; MockTransport for unit tests — no new dependency), pytest. Ollama in dev.

Global Constraints

• Python 3.12, FastAPI. No new dependencies — httpx==0.28.1 (already pinned) provides MockTransport; the live-test gate is a conftest.py option.
• Test setup: venv from Plan A at repo root (.venv). All test commands run from api/ with the venv active (source .venv/bin/activate from repo root, then cd api); api/pytest.ini puts app on the path. Fixture api/tests/fixtures/canon_log_valid.json already exists (Plan A).
• Server returns raw prose — /dm/narrate → {"prose": "<text with tags intact>"}. The server never parses [FACT:]/tags and never mutates state; the client's TagExtractor (Plan B) does that. §2 + DRY.
• §7/§2 boundary: render_digest is a curated projection — it omits raw integers (companion disposition values render as names only), renders luck_descriptor as flavor never a number, and drops schema_version/ids. The Narrator gets nouns and situation, never anything to do arithmetic on.
• Narrator model: default qwen3.5:latest via OLLAMA_NARRATOR_MODEL (charter §5 "the good model"; swappable per §4). think: false is sent as a top-level /api/chat field (qwen3.x thinking OFF; verified graceful on non-thinking models).
• Per-call seed (§10): the server sets an explicit options.seed per call and logs it (replay/eval).
• Failure contract (§12/§13): exactly one retry on transport error / timeout / non-2xx / empty content; then ModelError → 502 {"detail": {"model_error": "<reason>"}} (same family as the existing 422). No server-side fallback prose; the client falls back on any non-200.
• Non-streaming — stream: false. Streaming is a later plan.
• Ollama reachable at http://localhost:11434 for the gated live smoke (pytest --run-live).
• Execution branch: run this plan on feature/narrator-ollama-pipeline branched off dev (CLAUDE.md §18). Per-task commits land there; the human merges to dev.

---

Phase 1 — Pipeline foundation

The reusable machinery, each module mock/unit-tested. The endpoint is untouched until Phase 2.

Task 1: config + routing (role → model, env-driven)

Files:

• Create: api/app/config.py
• Create: api/app/routing.py
• Test: api/tests/test_routing.py

Interfaces:

• Produces: config.ollama_base_url() -> str, config.ollama_timeout_seconds() -> float, config.narrator_model() -> str, config.call_log_path() -> str | None (all read env at call time). routing.RoleConfig(model: str, options: dict, think: bool | None); routing.for_role(role: str) -> RoleConfig (raises KeyError for an unknown role).

• Step 1: Write the failing test

Create api/tests/test_routing.py:

import pytest

from app.routing import for_role


def test_narrator_defaults():
    cfg = for_role("narrator")
    assert cfg.model == "qwen3.5:latest"
    assert cfg.think is False
    assert cfg.options["num_predict"] == 300
    assert cfg.options["temperature"] == 0.8


def test_env_overrides_model(monkeypatch):
    monkeypatch.setenv("OLLAMA_NARRATOR_MODEL", "llama3.1:latest")
    assert for_role("narrator").model == "llama3.1:latest"


def test_unknown_role_raises():
    with pytest.raises(KeyError):
        for_role("wizard")

• Step 2: Run test to verify it fails

Run: cd api && python -m pytest tests/test_routing.py -v Expected: FAIL — ModuleNotFoundError: No module named 'app.routing'.

• Step 3: Write config.py

Create api/app/config.py:

"""Server config, read from the environment at call time so tests can override
without reimport. Charter §4 — all of this lives server-side; the client never
sees it.
"""

import os


def ollama_base_url() -> str:
    return os.environ.get("OLLAMA_BASE_URL", "http://localhost:11434")


def ollama_timeout_seconds() -> float:
    return float(os.environ.get("OLLAMA_TIMEOUT_SECONDS", "30"))


def narrator_model() -> str:
    return os.environ.get("OLLAMA_NARRATOR_MODEL", "qwen3.5:latest")


def call_log_path() -> str | None:
    return os.environ.get("CALL_LOG_PATH") or None

• Step 4: Write routing.py

Create api/app/routing.py:

"""Role → model routing (charter §4/§5). Model choice is config, not law — a
deploy, not a client patch. Read at call time so an env override needs no reimport.
The provider abstraction (Ollama → Replicate) slots in here later (YAGNI now).
"""

from dataclasses import dataclass

from . import config


@dataclass(frozen=True)
class RoleConfig:
    model: str
    options: dict
    think: bool | None = None  # Ollama top-level `think`; None omits it


def for_role(role: str) -> RoleConfig:
    if role == "narrator":
        return RoleConfig(
            model=config.narrator_model(),
            options={"temperature": 0.8, "top_p": 0.9, "num_predict": 300},
            think=False,  # qwen3.x thinking OFF; harmless on non-thinking models
        )
    raise KeyError(f"no routing for role: {role}")

• Step 5: Run test to verify it passes

Run: cd api && python -m pytest tests/test_routing.py -v Expected: PASS — 3 passed.

• Step 6: Commit

git add api/app/config.py api/app/routing.py api/tests/test_routing.py
git commit -m "feat(api): config + role→model routing (qwen3.5 narrator default, think off)"

---

Task 2: prompts — system-prompt loader + canon-log digest

Files:

• Create: api/app/prompts.py
• Test: api/tests/test_prompts.py

Interfaces:

• Consumes: the fixture api/tests/fixtures/canon_log_valid.json; the existing api/prompts/narrator.md.

• Produces: prompts.system_prompt(role: str) -> str (returns the file content below the first \n---\n — the header above it is human-facing front-matter; cached). prompts.render_digest(canon_log: dict) -> str (curated labeled digest; omits integers).

• Step 1: Write the failing test

Create api/tests/test_prompts.py:

import json
from pathlib import Path

from app.prompts import render_digest, system_prompt

VALID = json.loads((Path(__file__).parent / "fixtures" / "canon_log_valid.json").read_text())


def test_system_prompt_strips_the_header():
    body = system_prompt("narrator")
    assert isinstance(body, str) and body.strip() != ""
    # The human-facing markdown title above the '---' is not part of the prompt.
    assert "# Narrator prompt" not in body


def test_digest_has_the_narrative_context():
    d = render_digest(VALID)
    assert "Location: the Greywater docks" in d
    assert "Arrived at Greywater by barge before dawn" in d
    assert "Established facts:" in d
    assert "the harbourmaster is named Oda Fenn" in d
    assert "Companions present: Brannoc Thane, Cadwyn Vell" in d
    assert "Active quest: The Missing Ledger — Find who took Fenn's ledger" in d
    assert "Fortune: Fortune spits on you" in d
    assert "Player: Aldric, a sellsword" in d
    assert d.rstrip().endswith("Describe the scene as it is now.")


def test_digest_omits_raw_numbers_and_ids():
    d = render_digest(VALID)
    # §7/§2: no disposition integers, no ids, no schema_version leak into the prompt.
    assert "disposition" not in d
    assert "40" not in d and "15" not in d      # the deserter dispositions
    assert "brannoc_thane" not in d             # ids stay out; names only
    assert "schema_version" not in d

• Step 2: Run test to verify it fails

Run: cd api && python -m pytest tests/test_prompts.py -v Expected: FAIL — ModuleNotFoundError: No module named 'app.prompts'.

• Step 3: Write prompts.py

Create api/app/prompts.py:

"""Load a role's system prompt (§16 — prompts are source code) and render the
canon log into a curated digest for the user message (charter §11).

The digest is a projection, not a dump: it surfaces narrative context and
deliberately omits raw integers — companion dispositions render as names only,
luck as its descriptor, and ids/schema_version are dropped (charter §7/§2).
"""

from functools import lru_cache
from pathlib import Path

_PROMPT_DIR = Path(__file__).resolve().parent.parent / "prompts"


@lru_cache(maxsize=None)
def system_prompt(role: str) -> str:
    text = (_PROMPT_DIR / f"{role}.md").read_text(encoding="utf-8")
    # Everything below the first '---' line is the model-facing prompt; the
    # header above it is human/dev front-matter.
    parts = text.split("\n---\n", 1)
    body = parts[1] if len(parts) == 2 else text
    return body.strip()


def render_digest(canon_log: dict) -> str:
    lines: list[str] = []

    location = canon_log.get("location", {})
    lines.append(f"Location: {location.get('name', 'an unknown place')}")

    events = canon_log.get("recent_events", [])
    if events:
        lines.append("Recent events:")
        lines += [f"- {e}" for e in events]

    facts = canon_log.get("established_facts", [])
    if facts:
        lines.append("Established facts:")
        lines += [f"- {f}" for f in facts]

    party = canon_log.get("party", [])
    if party:
        lines.append("Companions present: " + ", ".join(m.get("name", "") for m in party))

    for quest in canon_log.get("active_quests", []):
        if quest.get("status") == "active":
            lines.append(f"Active quest: {quest.get('name', '')} — {quest.get('objective', '')}")

    player = canon_log.get("player", {})
    descriptor = player.get("luck_descriptor")
    if descriptor:
        lines.append(f"Fortune: {descriptor}")
    lines.append(f"Player: {player.get('name', '')}, a {player.get('class_id', '')}")

    lines.append("")
    lines.append("Describe the scene as it is now.")
    return "\n".join(lines)

• Step 4: Run test to verify it passes

Run: cd api && python -m pytest tests/test_prompts.py -v Expected: PASS — 3 passed. (system_prompt returns whatever is below --- in the current narrator.md; its body is authored in Task 5.)

• Step 5: Commit

git add api/app/prompts.py api/tests/test_prompts.py
git commit -m "feat(api): prompt loader (header split) + curated canon-log digest"

---

Task 3: ollama_client — call /api/chat with one retry

Files:

• Create: api/app/ollama_client.py
• Test: api/tests/test_ollama_client.py

Interfaces:

• Consumes: config.ollama_base_url, config.ollama_timeout_seconds.

• Produces: ollama_client.ModelError (Exception); ollama_client.chat(model: str, messages: list[dict], options: dict, *, think: bool | None = None, client: httpx.Client | None = None) -> str — returns the assistant content, raises ModelError after one retry. Sends {"model", "messages", "stream": False, "options"} and, only when think is not None, a top-level "think".

• Step 1: Write the failing test

Create api/tests/test_ollama_client.py:

import json

import httpx
import pytest

from app.ollama_client import ModelError, chat


def _client(handler):
    return httpx.Client(transport=httpx.MockTransport(handler), base_url="http://ollama")


def test_success_returns_content_and_sends_payload():
    seen = {}

    def handler(request):
        seen["body"] = json.loads(request.content)
        return httpx.Response(200, json={"message": {"content": "You stand on the wharf."}})

    out = chat("qwen3.5:latest", [{"role": "user", "content": "hi"}], {"seed": 7},
               think=False, client=_client(handler))
    assert out == "You stand on the wharf."
    assert seen["body"]["model"] == "qwen3.5:latest"
    assert seen["body"]["stream"] is False
    assert seen["body"]["options"]["seed"] == 7
    assert seen["body"]["think"] is False


def test_think_omitted_when_none():
    seen = {}

    def handler(request):
        seen["body"] = json.loads(request.content)
        return httpx.Response(200, json={"message": {"content": "ok"}})

    chat("m", [{"role": "user", "content": "x"}], {}, client=_client(handler))
    assert "think" not in seen["body"]


def test_retries_once_then_succeeds():
    calls = {"n": 0}

    def handler(request):
        calls["n"] += 1
        if calls["n"] == 1:
            raise httpx.ConnectError("boom")
        return httpx.Response(200, json={"message": {"content": "recovered"}})

    out = chat("m", [{"role": "user", "content": "x"}], {}, client=_client(handler))
    assert out == "recovered"
    assert calls["n"] == 2


def test_two_transport_failures_raise_modelerror():
    def handler(request):
        raise httpx.ConnectError("down")

    with pytest.raises(ModelError):
        chat("m", [{"role": "user", "content": "x"}], {}, client=_client(handler))


def test_timeout_exhausted_raises_modelerror():
    def handler(request):
        raise httpx.TimeoutException("slow")

    with pytest.raises(ModelError):
        chat("m", [{"role": "user", "content": "x"}], {}, client=_client(handler))


def test_non_2xx_retried_then_modelerror():
    def handler(request):
        return httpx.Response(500, json={"error": "nope"})

    with pytest.raises(ModelError):
        chat("m", [{"role": "user", "content": "x"}], {}, client=_client(handler))


def test_empty_content_retried_then_modelerror():
    def handler(request):
        return httpx.Response(200, json={"message": {"content": "   "}})

    with pytest.raises(ModelError):
        chat("m", [{"role": "user", "content": "x"}], {}, client=_client(handler))

• Step 2: Run test to verify it fails

Run: cd api && python -m pytest tests/test_ollama_client.py -v Expected: FAIL — ModuleNotFoundError: No module named 'app.ollama_client'.

• Step 3: Write ollama_client.py

Create api/app/ollama_client.py:

"""Call Ollama's /api/chat (non-streaming) with a single retry. Callers see
success or ModelError (charter §12/§13 — one retry, then fail; the client
degrades on any non-200). Built so a streaming mode is an additive path later.
"""

import httpx

from . import config


class ModelError(Exception):
    """The model call failed after one retry (transport / timeout / non-2xx / empty)."""


def chat(
    model: str,
    messages: list[dict],
    options: dict,
    *,
    think: bool | None = None,
    client: httpx.Client | None = None,
) -> str:
    payload: dict = {"model": model, "messages": messages, "stream": False, "options": options}
    if think is not None:
        payload["think"] = think

    owns = client is None
    http = client or httpx.Client(
        base_url=config.ollama_base_url(), timeout=config.ollama_timeout_seconds()
    )
    try:
        last = "no attempt made"
        for _ in range(2):  # one try + one retry (§12: never more than once)
            try:
                resp = http.post("/api/chat", json=payload)
                resp.raise_for_status()
                content = resp.json().get("message", {}).get("content", "")
                if content.strip():
                    return content
                last = "empty response from model"
            except httpx.HTTPError as exc:  # TimeoutException, TransportError, HTTPStatusError
                last = f"{type(exc).__name__}: {exc}"
        raise ModelError(last)
    finally:
        if owns:
            http.close()

• Step 4: Run test to verify it passes

Run: cd api && python -m pytest tests/test_ollama_client.py -v Expected: PASS — 7 passed.

• Step 5: Commit

git add api/app/ollama_client.py api/tests/test_ollama_client.py
git commit -m "feat(api): Ollama /api/chat client with one retry + typed ModelError"

---

Task 4: call_log — §10 structured JSON-lines logging

Files:

• Create: api/app/call_log.py
• Test: api/tests/test_call_log.py

Interfaces:

• Consumes: config.call_log_path.

• Produces: call_log.record(*, role, model, options, messages, canon_log, ok, latency_ms, response=None, error=None, write=<default>) -> dict — builds the record, writes one JSON line via write (default: CALL_LOG_PATH file if set, else stdout), returns the record. response present only when ok; error present only when not.

• Step 1: Write the failing test

Create api/tests/test_call_log.py:

import json

from app.call_log import record


def test_success_record_shape():
    captured = []
    rec = record(
        role="narrator", model="qwen3.5:latest", options={"seed": 7},
        messages=[{"role": "user", "content": "x"}], canon_log={"a": 1},
        ok=True, latency_ms=120, response="prose", write=captured.append,
    )
    assert len(captured) == 1
    parsed = json.loads(captured[0])
    assert parsed["ok"] is True
    assert parsed["response"] == "prose"
    assert parsed["options"]["seed"] == 7
    assert parsed["role"] == "narrator"
    assert parsed["canon_log"] == {"a": 1}
    assert "error" not in parsed
    assert "ts" in parsed
    assert rec["response"] == "prose"


def test_failure_record_shape():
    captured = []
    record(
        role="narrator", model="m", options={}, messages=[], canon_log={},
        ok=False, latency_ms=5, error="ModelError: down", write=captured.append,
    )
    parsed = json.loads(captured[0])
    assert parsed["ok"] is False
    assert parsed["error"] == "ModelError: down"
    assert "response" not in parsed

• Step 2: Run test to verify it fails

Run: cd api && python -m pytest tests/test_call_log.py -v Expected: FAIL — ModuleNotFoundError: No module named 'app.call_log'.

• Step 3: Write call_log.py

Create api/app/call_log.py:

"""Structured JSON-lines logging of every model call (charter §10 — the seed and
the full prompt logged with every call; §4 — free eval/replay infrastructure).

One JSON object per line: everything needed to replay the call (canon_log +
messages + model + seed) against a candidate model. No secrets to redact — the
client never sends keys, the canon log carries no PII, and the prompt is exactly
what we want on record.
"""

import json
import sys
from datetime import datetime, timezone
from typing import Callable

from . import config


def _default_write(line: str) -> None:
    path = config.call_log_path()
    if path:
        with open(path, "a", encoding="utf-8") as handle:
            handle.write(line + "\n")
    else:
        sys.stdout.write(line + "\n")


def record(
    *,
    role: str,
    model: str,
    options: dict,
    messages: list[dict],
    canon_log: dict,
    ok: bool,
    latency_ms: int,
    response: str | None = None,
    error: str | None = None,
    write: Callable[[str], None] = _default_write,
) -> dict:
    rec: dict = {
        "ts": datetime.now(timezone.utc).isoformat(),
        "role": role,
        "model": model,
        "options": options,
        "messages": messages,
        "canon_log": canon_log,
        "ok": ok,
        "latency_ms": latency_ms,
    }
    if ok:
        rec["response"] = response
    else:
        rec["error"] = error
    write(json.dumps(rec, ensure_ascii=False))
    return rec

• Step 4: Run test to verify it passes

Run: cd api && python -m pytest tests/test_call_log.py -v Expected: PASS — 2 passed.

• Step 5: Run the whole Phase 1 suite

Run: cd api && python -m pytest -v Expected: PASS — all Task 1–4 tests plus the pre-existing Plan A suite green.

• Step 6: Commit

git add api/app/call_log.py api/tests/test_call_log.py
git commit -m "feat(api): §10 JSON-lines call logging (full prompt + seed, injectable sink)"

---

Phase 2 — Narrator online

Author the prompt, wire the service and endpoint, prove the wire against a real model.

Task 5: Author the narrator.md system-prompt body

Files:

• Modify: api/prompts/narrator.md
• Test: api/tests/test_prompts.py (append one assertion)

Interfaces:

• Consumes: prompts.system_prompt (Task 2). No new interface.

• Step 1: Author the prompt body

Replace the entire contents of api/prompts/narrator.md with (header above --- stays human-facing; everything below is the model-facing system prompt):

# Narrator prompt

**Role:** Describe scenes, outcomes, transitions. **Quality matters most — use the good model** (charter §5).

**Input:** scene state + canon log (§11), rendered as a labeled digest in the user message.
**Output:** prose with tags. Never invent a proper noun without emitting `[FACT: ...]` so code can harvest it into `established_facts` (§11).

**Voice:** dry, gritty not grim, plays the world straight. Never winks. Nothing is "hilariously" anything (§3).

---

You are the Narrator of a gritty high-fantasy RPG. You describe the world; you never control the player.

Given the current scene state below, describe the scene as it is right now, in the second person ("you"), present tense. Cover what the player sees, hears, and smells, and what the people present are doing. Three to five sentences, then stop.

Voice:
- Dry and economical. Gritty, not grim. Play the world straight — never wink, never point out that a thing is funny. Nothing is "hilariously" or "comically" anything.
- Profanity and ugliness are allowed when the moment earns them, never as decoration.

Hard rules:
- Do not introduce a proper noun — a person's name, a place name, a named object — unless it already appears below. If you must introduce one, append a tag on its own: `[FACT: <the new fact as a short clause>]`, one per new proper noun, so the game can record it.
- Never contradict anything under "Established facts".
- Never state a number, a statistic, a hit-point total, or a Luck value. "Fortune" is a feeling, never a figure.
- Never decide, narrate, or assume the player's choices, words, or actions. Describe the world and the other characters; leave the player's next move to the player.
- Output prose only, plus any `[FACT: ...]` tags. No JSON, no lists, no headings, no notes about what you are doing.

• Step 2: Append the body-authored assertion

Add to api/tests/test_prompts.py:

def test_narrator_body_is_authored():
    body = system_prompt.__wrapped__("narrator")  # bypass lru_cache in case of prior load
    assert "TBD" not in body
    assert "[FACT:" in body           # the tag instruction is present
    assert "second person" in body    # the core task is present

• Step 3: Run the prompt tests

Run: cd api && python -m pytest tests/test_prompts.py -v Expected: PASS — 4 passed (the 3 from Task 2 + the new body assertion). The .__wrapped__ call bypasses lru_cache so the freshly authored file is read.

• Step 4: Commit

git add api/prompts/narrator.md api/tests/test_prompts.py
git commit -m "feat(api): author narrator system prompt (voice, [FACT] rule, no-numbers, no player agency)"

---

Task 6: narrate — the Narrator service

Files:

• Create: api/app/narrate.py
• Test: api/tests/test_narrate.py

Interfaces:

• Consumes: routing.for_role, prompts.system_prompt/render_digest, ollama_client.chat/ModelError, call_log.record (all imported as modules so tests can monkeypatch the attribute).

• Produces: narrate.run(canon_log: dict) -> str — renders digest, routes model, calls Ollama with a per-call seed and think, logs success or failure, returns raw prose or re-raises ModelError.

• Step 1: Write the failing test

Create api/tests/test_narrate.py:

import json
from pathlib import Path

import pytest

import app.narrate as narrate
from app.ollama_client import ModelError

VALID = json.loads((Path(__file__).parent / "fixtures" / "canon_log_valid.json").read_text())


def test_run_returns_prose_and_logs_success(monkeypatch):
    seen = {}

    def fake_chat(model, messages, options, *, think=None):
        seen.update(model=model, messages=messages, options=options, think=think)
        return "You stand on the wharf. [FACT: the gulls have gone quiet]"

    logs = []
    monkeypatch.setattr(narrate.ollama_client, "chat", fake_chat)
    monkeypatch.setattr(narrate.call_log, "record", lambda **kw: logs.append(kw))

    out = narrate.run(VALID)
    assert out.startswith("You stand on the wharf.")
    assert seen["model"] == "qwen3.5:latest"
    assert seen["think"] is False
    assert seen["messages"][0]["role"] == "system"
    assert seen["messages"][1]["role"] == "user"
    assert "seed" in seen["options"]
    assert logs and logs[0]["ok"] is True and logs[0]["response"] == out


def test_run_logs_failure_and_reraises(monkeypatch):
    def boom(*args, **kwargs):
        raise ModelError("upstream down")

    logs = []
    monkeypatch.setattr(narrate.ollama_client, "chat", boom)
    monkeypatch.setattr(narrate.call_log, "record", lambda **kw: logs.append(kw))

    with pytest.raises(ModelError):
        narrate.run(VALID)
    assert logs and logs[0]["ok"] is False and "down" in logs[0]["error"]

• Step 2: Run test to verify it fails

Run: cd api && python -m pytest tests/test_narrate.py -v Expected: FAIL — ModuleNotFoundError: No module named 'app.narrate'.

• Step 3: Write narrate.py

Create api/app/narrate.py:

"""The Narrator role service (charter §5). Renders the digest, routes the model,
calls Ollama with one retry, logs the call (§10), and returns raw prose — tags
left intact for the client's TagExtractor to harvest (§2/§6). Raises ModelError
on failure for the handler to map to a 502.
"""

import random
from time import perf_counter

from . import call_log, ollama_client, prompts, routing


def run(canon_log: dict) -> str:
    cfg = routing.for_role("narrator")
    options = {**cfg.options, "seed": random.randint(0, 2**31 - 1)}
    messages = [
        {"role": "system", "content": prompts.system_prompt("narrator")},
        {"role": "user", "content": prompts.render_digest(canon_log)},
    ]

    start = perf_counter()
    try:
        text = ollama_client.chat(cfg.model, messages, options, think=cfg.think)
    except ollama_client.ModelError as exc:
        call_log.record(
            role="narrator", model=cfg.model, options=options, messages=messages,
            canon_log=canon_log, ok=False,
            latency_ms=int((perf_counter() - start) * 1000), error=str(exc),
        )
        raise

    call_log.record(
        role="narrator", model=cfg.model, options=options, messages=messages,
        canon_log=canon_log, ok=True,
        latency_ms=int((perf_counter() - start) * 1000), response=text,
    )
    return text

• Step 4: Run test to verify it passes

Run: cd api && python -m pytest tests/test_narrate.py -v Expected: PASS — 2 passed.

• Step 5: Commit

git add api/app/narrate.py api/tests/test_narrate.py
git commit -m "feat(api): narrate service — render→route→call→log, returns raw prose"

---

Task 7: Wire /dm/narrate

Files:

• Modify: api/app/main.py
• Test: api/tests/test_narrate_endpoint.py

Interfaces:

• Consumes: narrate.run, ollama_client.ModelError.

• Produces: POST /dm/narrate → 200 {"prose": text} on success; 502 {"detail": {"model_error": "<reason>"}} on ModelError; 422 (existing) on an invalid canon log. The other four role endpoints stay stubbed.

• Step 1: Write the failing test

Create api/tests/test_narrate_endpoint.py:

import json
from pathlib import Path

from fastapi.testclient import TestClient

import app.main as main
import app.narrate as narrate
from app.ollama_client import ModelError

client = TestClient(main.app)
VALID = json.loads((Path(__file__).parent / "fixtures" / "canon_log_valid.json").read_text())


def test_valid_log_returns_prose(monkeypatch):
    monkeypatch.setattr(narrate.ollama_client, "chat", lambda *a, **k: "You stand on the wharf.")
    monkeypatch.setattr(narrate.call_log, "record", lambda **kw: None)  # keep test output pristine
    r = client.post("/dm/narrate", json={"canon_log": VALID})
    assert r.status_code == 200
    assert r.json() == {"prose": "You stand on the wharf."}


def test_model_error_maps_to_502(monkeypatch):
    def boom(*a, **k):
        raise ModelError("upstream down")

    monkeypatch.setattr(narrate.ollama_client, "chat", boom)
    monkeypatch.setattr(narrate.call_log, "record", lambda **kw: None)
    r = client.post("/dm/narrate", json={"canon_log": VALID})
    assert r.status_code == 502
    assert "model_error" in r.json()["detail"]


def test_invalid_log_still_422(monkeypatch):
    monkeypatch.setattr(narrate.ollama_client, "chat", lambda *a, **k: "x")
    bad = json.loads(json.dumps(VALID))
    bad["player"]["luck"] = 5  # §7 leak — schema rejects
    r = client.post("/dm/narrate", json={"canon_log": bad})
    assert r.status_code == 422
    assert "canon_log_errors" in r.json()["detail"]

• Step 2: Run test to verify it fails

Run: cd api && python -m pytest tests/test_narrate_endpoint.py -v Expected: FAIL — the current /dm/narrate returns 200 {"detail": "not implemented"}, so test_valid_log_returns_prose and test_model_error_maps_to_502 fail.

• Step 3: Wire the endpoint

In api/app/main.py, add these imports near the existing imports:

from .narrate import run as narrate_run
from .ollama_client import ModelError

Then replace the existing narrate handler:

@app.post("/dm/narrate")
def narrate(req: TurnRequest = Depends(valid_turn)) -> dict:
    return {"detail": "not implemented"}

with:

@app.post("/dm/narrate")
def narrate(req: TurnRequest = Depends(valid_turn)) -> dict:
    try:
        return {"prose": narrate_run(req.canon_log)}
    except ModelError as exc:
        raise HTTPException(status_code=502, detail={"model_error": str(exc)})

Leave /dm/adjudicate, /dm/improvise, /npc/speak, /party/banter returning {"detail": "not implemented"}.

• Step 4: Run test to verify it passes

Run: cd api && python -m pytest tests/test_narrate_endpoint.py -v Expected: PASS — 3 passed.

• Step 5: Commit

git add api/app/main.py api/tests/test_narrate_endpoint.py
git commit -m "feat(api): wire /dm/narrate — 200 prose | 502 model_error | 422 invalid log"

---

Task 8: Gated live smoke test + env docs

Files:

• Create: api/conftest.py
• Modify: api/pytest.ini
• Create: api/tests/test_live_smoke.py
• Modify: api/.env.example

Interfaces:

• Consumes: narrate.run. Produces: a live marker + --run-live gate; the live test is skipped by default and hits the real Ollama when opted in.

• Step 1: Add the --run-live gate

Create api/conftest.py:

"""Gate `@pytest.mark.live` tests (they hit a real Ollama) behind --run-live so
the default suite stays hermetic. Run the live layer with: pytest --run-live
"""

import pytest


def pytest_addoption(parser):
    parser.addoption(
        "--run-live", action="store_true", default=False,
        help="run tests marked @pytest.mark.live (require a real Ollama)",
    )


def pytest_collection_modifyitems(config, items):
    if config.getoption("--run-live"):
        return
    skip_live = pytest.mark.skip(reason="needs --run-live (real Ollama)")
    for item in items:
        if "live" in item.keywords:
            item.add_marker(skip_live)

• Step 2: Register the marker

Replace api/pytest.ini with:

[pytest]
pythonpath = .
testpaths = tests
markers =
    live: exercises a real Ollama; skipped unless --run-live

• Step 3: Write the live smoke test

Create api/tests/test_live_smoke.py:

import json
from pathlib import Path

import pytest

import app.narrate as narrate

VALID = json.loads((Path(__file__).parent / "fixtures" / "canon_log_valid.json").read_text())


@pytest.mark.live
def test_narrator_returns_real_prose():
    """Hits the real Ollama (OLLAMA_BASE_URL). Proves the wire end-to-end:
    prompt assembly → model → non-empty prose. Not asserted: tag content
    (nondeterministic). Run with: pytest --run-live
    """
    text = narrate.run(VALID)
    assert isinstance(text, str)
    assert len(text.strip()) > 40  # real prose, not a blip

• Step 4: Verify the default suite SKIPS it, then run it live

Run (default — must skip): cd api && python -m pytest tests/test_live_smoke.py -v Expected: 1 skipped (needs --run-live).

Run (live — real model): cd api && python -m pytest tests/test_live_smoke.py -v --run-live Expected: PASS — 1 passed; a JSON call-log line is emitted to stdout, and the returned prose is a few sentences of scene description. This proves the whole loop against the real qwen3.5.

• Step 5: Document the new env vars

Append to api/.env.example:

# Narrator model (charter §5 "the good model"). qwen3.x runs with thinking OFF.
OLLAMA_NARRATOR_MODEL=qwen3.5:latest
# Per-attempt model-call timeout (seconds).
OLLAMA_TIMEOUT_SECONDS=30
# Optional: append call logs (JSON lines, §10) to this file instead of stdout.
CALL_LOG_PATH=

• Step 6: Run the whole suite (default, hermetic)

Run: cd api && python -m pytest -v Expected: PASS — every unit test from Tasks 1–7 green, the live smoke skipped, and the pre-existing Plan A suite green.

• Step 7: Commit

git add api/conftest.py api/pytest.ini api/tests/test_live_smoke.py api/.env.example
git commit -m "test(api): gated live Ollama smoke (--run-live) + env docs"

---

Self-Review

Spec coverage:

• Shared pipeline + Narrator first (decision 1): config/routing/prompts/ollama_client/call_log (Phase 1) + narrate/endpoint (Phase 2). ✔
• Mock boundary + gated live smoke (decision 2): httpx MockTransport unit tests (Tasks 3, 6, 7); @pytest.mark.live + --run-live gate (Task 8). ✔
• Non-streaming now (decision 3): stream: False in ollama_client; streaming named out-of-scope. ✔
• Typed error → client fallback (decision 4): ModelError → 502 {"detail":{"model_error":…}} (Task 7); 422 preserved. ✔
• Rendered digest, curated projection (decision 5): render_digest omits dispositions/ids/schema_version, luck as descriptor (Task 2, asserted). ✔
• Tag handling client-side (decision 6): server returns raw {"prose": …}; no tag parsing anywhere in the server. ✔
• qwen3.5 default + think:false top-level, graceful on non-thinking models (routing Task 1; payload asserted Task 3). ✔
• Per-call seed logged (§10): narrate sets options.seed; call_log records full messages + canon_log + options (Tasks 4, 6). ✔
• Prompt authored as source code (§16), header/body split convention (Tasks 2, 5). ✔

Placeholder scan: No TBD/TODO in code. The narrator.md body is fully authored in Task 5 (and a test asserts TBD is gone). ✔

Type consistency: chat(model, messages, options, *, think=None, client=None) -> str used identically in Tasks 3/6. for_role(role) -> RoleConfig(model, options, think) consistent (Tasks 1/6). call_log.record(*, role, model, options, messages, canon_log, ok, latency_ms, response=, error=, write=) consistent (Tasks 4/6). narrate.run(canon_log) -> str consistent (Tasks 6/7/8). Modules imported as attributes (narrate.ollama_client.chat, narrate.call_log.record) so the monkeypatches in Tasks 6/7 target the right objects. ✔

Deviations from spec (flag for reviewer):

• routing builds RoleConfig at call time reading env (not a module-level ROLES dict frozen at import as the spec sketched) — so an env override is testable without reimport. Strictly a testability improvement; same behavior.
• prompts.system_prompt uses lru_cache; Task 5's test calls .__wrapped__ to read the freshly authored file. Noted so the reviewer doesn't read it as a hack.

Out of scope (later plans): streaming; client HTTP layer + degraded-DM fallback display + [FACT] harvest; the other four roles; Replicate provider; auth/metering.