diff --git a/docs/superpowers/plans/2026-06-04-experience-simulator.md b/docs/superpowers/plans/2026-06-04-experience-simulator.md new file mode 100644 index 0000000..8768e14 --- /dev/null +++ b/docs/superpowers/plans/2026-06-04-experience-simulator.md @@ -0,0 +1,1058 @@ +# Experience Simulator (Curator's X-ray) Implementation Plan + +> **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:** A web-based, Docker-on-localhost simulator that wires the installation's five dials to the *real* `hef.selection` code and shows a full-transparency "X-ray" — the picked piece, the ranked candidate pool with distances, and brain/mood grid maps — so a curator can feel whether the selection model surfaces fitting pieces. + +**Architecture:** One FastAPI service in one container serves a JSON selection API and a dependency-free vanilla-JS frontend. The backend imports `hef.catalog` and `hef.selection` directly; selection ranking is exposed through a new additive `hef.selection.ranked_candidates()` so the X-ray shows exactly what the picker picks from. A seeded synthetic fixture catalog feeds the model since the real catalog is empty. + +**Tech Stack:** Python ≥3.11, FastAPI, Uvicorn, Pydantic, pytest + Starlette `TestClient` (httpx); vanilla HTML/CSS/JS (no build step); Docker + docker compose. + +Spec: [`docs/superpowers/specs/2026-06-04-experience-simulator-design.md`](../specs/2026-06-04-experience-simulator-design.md). + +--- + +## File Structure + +- **Modify** `hef/selection.py` — add public `ranked_candidates()`; refactor `select()` to call it. +- **Modify** `pyproject.toml` — add `simulator` package; add `[project.optional-dependencies] sim`. +- **Create** `simulator/__init__.py` — package marker. +- **Create** `simulator/fixtures.py` — deterministic synthetic catalog generator. +- **Create** `simulator/app.py` — FastAPI factory, request model, endpoints, catalog loading, static mount, module-level `app`. +- **Create** `simulator/static/index.html`, `simulator/static/app.js`, `simulator/static/style.css` — the X-ray UI. +- **Create** `simulator/Dockerfile`, `simulator/docker-compose.yml` — packaging. +- **Create** `Makefile` — `sim` / `sim-local` targets. +- **Create** `tests/test_selection_ranked.py`, `tests/test_fixtures.py`, `tests/test_simulator_api.py` — tests. +- **Modify** `docs/USER_GUIDE.md` — "Playing with the simulator" section. + +Each task ends green and is committed independently. + +--- + +### Task 1: `hef.selection.ranked_candidates()` + refactor `select()` + +**Files:** +- Modify: `hef/selection.py` +- Test: `tests/test_selection_ranked.py` (Create) + +- [ ] **Step 1: Write the failing tests** + +Create `tests/test_selection_ranked.py`: + +```python +import random + +import pytest + +from hef.catalog import Record +from hef.selection import Coordinate, Weights, ranked_candidates, select + + +def make_record(**overrides): + base = dict( + id="r", + title="t", + source_url="u", + source_archive="internet_archive", + license="public_domain", + mode="video", + left=0, + right=0, + dark=0, + light=0, + duration_s=600, + file_path="", + ) + base.update(overrides) + return Record(**base) + + +def test_ranked_candidates_sorts_nearest_first_with_distances(): + near = make_record(id="near", mode="video", left=1, right=1, dark=0, light=0) + far = make_record(id="far", mode="video", left=4, right=4, dark=4, light=4) + ranked = ranked_candidates([far, near], Coordinate(0, 0, 0, 0), "video") + assert [r.id for r, _ in ranked] == ["near", "far"] + assert ranked[0][1] < ranked[1][1] + + +def test_ranked_candidates_caps_at_pool_size(): + recs = [make_record(id=f"r{i}", mode="video", left=i % 5) for i in range(10)] + ranked = ranked_candidates(recs, Coordinate(0, 0, 0, 0), "video", pool_size=3) + assert len(ranked) == 3 + + +def test_ranked_candidates_filters_by_mode(): + a = make_record(id="aud", mode="audio") + v = make_record(id="vid", mode="video") + ranked = ranked_candidates([a, v], Coordinate(0, 0, 0, 0), "audio") + assert [r.id for r, _ in ranked] == ["aud"] + + +def test_ranked_candidates_av_falls_back_to_audio_and_video(): + a = make_record(id="aud", mode="audio") + v = make_record(id="vid", mode="video") + ranked = ranked_candidates([a, v], Coordinate(0, 0, 0, 0), "av", pool_size=4) + assert {r.id for r, _ in ranked} == {"aud", "vid"} + + +def test_ranked_candidates_approved_only(): + p = make_record(id="prop", mode="video", review_status="proposed") + ok = make_record(id="appr", mode="video", review_status="approved") + ranked = ranked_candidates([p, ok], Coordinate(0, 0, 0, 0), "video", approved_only=True) + assert [r.id for r, _ in ranked] == ["appr"] + + +def test_ranked_candidates_rejects_none_mode(): + with pytest.raises(ValueError): + ranked_candidates([], Coordinate(0, 0, 0, 0), "none") + + +def test_select_returns_rank_one_of_ranked_candidates(): + recs = [ + make_record(id="near", mode="video", left=1), + make_record(id="far", mode="video", left=4), + ] + coord = Coordinate(0, 0, 0, 0) + ranked = ranked_candidates(recs, coord, "video") + assert select(recs, coord, "video", rng=None).id == ranked[0][0].id + + +def test_select_none_mode_still_returns_none(): + recs = [make_record(id="v", mode="video")] + assert select(recs, Coordinate(0, 0, 0, 0), "none") is None + + +def test_select_shuffles_within_pool_when_rng_given(): + recs = [make_record(id=f"r{i}", mode="video", left=i) for i in range(5)] + coord = Coordinate(0, 0, 0, 0) + picks = {select(recs, coord, "video", pool_size=5, rng=random.Random(s)).id for s in range(20)} + assert len(picks) > 1 +``` + +- [ ] **Step 2: Run tests to verify they fail** + +Run: `python -m pytest tests/test_selection_ranked.py -v` +Expected: FAIL — `ImportError: cannot import name 'ranked_candidates'`. + +- [ ] **Step 3: Add `ranked_candidates` and refactor `select`** + +In `hef/selection.py`, replace the `select()` function (lines ~61-98) with: + +```python +def ranked_candidates( + records, + coord: Coordinate, + mode: str, + *, + pool_size: int = 4, + weights: Weights = Weights(), + approved_only: bool = False, +) -> list[tuple[Record, float]]: + """Mode-eligible records sorted nearest-first, each paired with its distance. + + Applies the same approved_only filter and 'av' pool fallback as select(), + then returns up to pool_size nearest (record, distance) pairs. This is the + single source of truth for "the pool"; select() picks from it. + """ + if mode not in CONTENT_MODES: + raise ValueError( + f"ranked_candidates expects a content mode {sorted(CONTENT_MODES)}, " + f"got {mode!r}" + ) + pool = records + if approved_only: + pool = [r for r in pool if r.review_status == "approved"] + candidates = candidates_for_mode(pool, mode, pool_size) + ranked = sorted( + candidates, + key=lambda r: (distance(coord, record_coordinate(r), weights), r.id), + ) + nearest = ranked[:pool_size] + return [(r, distance(coord, record_coordinate(r), weights)) for r in nearest] + + +def select( + records, + coord: Coordinate, + mode: str, + *, + pool_size: int = 4, + weights: Weights = Weights(), + approved_only: bool = False, + rng=None, +) -> Optional[Record]: + """Pick the record nearest `coord` for the given selector `mode`. + + - 'none' selector mode returns None (the void/rest state). + - With rng=None, returns the single nearest record (ties broken by id) for + deterministic behavior. Pass a random.Random to shuffle within the + pool_size nearest records. + - approved_only restricts to records the human has blessed. + """ + if mode not in SELECTOR_MODES: + raise ValueError( + f"invalid selector mode {mode!r}; expected one of {sorted(SELECTOR_MODES)}" + ) + if mode == "none": + return None + ranked = ranked_candidates( + records, + coord, + mode, + pool_size=pool_size, + weights=weights, + approved_only=approved_only, + ) + if not ranked: + return None + nearest = [r for r, _ in ranked] + if rng is None: + return nearest[0] + return rng.choice(nearest) +``` + +Leave the `from typing import Optional` import where it already is (move it to the top of the file if cleaner, but it is not required for this change). + +- [ ] **Step 4: Run the new tests AND the existing selection suite** + +Run: `python -m pytest tests/test_selection_ranked.py tests/test_selection.py -v` +Expected: PASS (all). The existing `test_selection.py` must stay green — `select()` behavior is unchanged. + +- [ ] **Step 5: Commit** + +```bash +git add hef/selection.py tests/test_selection_ranked.py +git commit -m "feat(selection): expose ranked_candidates(); refactor select() onto it" +``` + +--- + +### Task 2: Declare the simulator package and its dependencies + +**Files:** +- Modify: `pyproject.toml` +- Create: `simulator/__init__.py` + +- [ ] **Step 1: Create the package marker** + +Create `simulator/__init__.py`: + +```python +"""Web-based curator's X-ray simulator for the experience filter.""" +``` + +- [ ] **Step 2: Add the package and optional deps to `pyproject.toml`** + +Change the `[tool.setuptools]` packages line from: + +```toml +[tool.setuptools] +packages = ["hef", "tools"] +``` + +to: + +```toml +[tool.setuptools] +packages = ["hef", "tools", "simulator"] +``` + +And add this block after the `[project.scripts]` section: + +```toml +[project.optional-dependencies] +sim = [ + "fastapi>=0.110", + "uvicorn[standard]>=0.29", + "httpx>=0.27", +] +``` + +(`httpx` is what Starlette's `TestClient` uses, so it belongs in the same group the tests need.) + +- [ ] **Step 3: Install the sim dependencies into the venv** + +Run: `python -m pip install -e ".[sim]"` +Expected: installs fastapi, uvicorn, httpx; ends with "Successfully installed …". + +- [ ] **Step 4: Verify imports resolve** + +Run: `python -c "import fastapi, uvicorn, httpx; import simulator; print('ok')"` +Expected: prints `ok`. + +- [ ] **Step 5: Commit** + +```bash +git add pyproject.toml simulator/__init__.py +git commit -m "build(simulator): add simulator package and sim optional-deps" +``` + +--- + +### Task 3: Synthetic fixture catalog generator + +**Files:** +- Create: `simulator/fixtures.py` +- Test: `tests/test_fixtures.py` (Create) + +- [ ] **Step 1: Write the failing tests** + +Create `tests/test_fixtures.py`: + +```python +from hef.catalog import validate_catalog +from hef.selection import CONTENT_MODES +from simulator.fixtures import generate_fixture_catalog + + +def test_fixture_catalog_is_valid(): + records = generate_fixture_catalog() + validate_catalog(records) # raises on any invalid record or duplicate id + + +def test_fixture_catalog_spans_the_coordinate_space(): + records = generate_fixture_catalog() + coords = {(r.left, r.right, r.dark, r.light) for r in records} + # all 625 cells of the 5x5 brain x 5x5 mood space are present + assert len(coords) == 625 + + +def test_fixture_catalog_has_every_content_mode(): + records = generate_fixture_catalog() + present = {r.mode for r in records} + assert CONTENT_MODES <= present + + +def test_fixture_catalog_mixes_review_statuses(): + records = generate_fixture_catalog() + statuses = {r.review_status for r in records} + assert statuses == {"proposed", "approved"} + + +def test_fixture_catalog_references_no_real_media(): + records = generate_fixture_catalog() + assert all(r.file_path == "" for r in records) + + +def test_fixture_catalog_is_deterministic(): + a = generate_fixture_catalog(seed=42) + b = generate_fixture_catalog(seed=42) + assert [r.id for r in a] == [r.id for r in b] + assert [r.mode for r in a] == [r.mode for r in b] + assert [r.review_status for r in a] == [r.review_status for r in b] +``` + +- [ ] **Step 2: Run tests to verify they fail** + +Run: `python -m pytest tests/test_fixtures.py -v` +Expected: FAIL — `ModuleNotFoundError: No module named 'simulator.fixtures'`. + +- [ ] **Step 3: Write the generator** + +Create `simulator/fixtures.py`: + +```python +"""Deterministic synthetic catalog so the selection model can be felt everywhere. + +The real catalog (catalog/library.jsonl) is empty; this generates one record per +cell of the 5x5 brain x 5x5 mood coordinate space (625 records), with a seeded +mix of content modes and review statuses and no real media attached. +""" + +from __future__ import annotations + +import random + +from hef.catalog import Record + +MODES = ("audio", "video", "av") +ARCHIVES = ("internet_archive", "musopen", "librivox", "nasa", "freesound") + +_LEFT_WORDS = ("Treatise", "Lecture", "Field Notes", "Reading", "Documentary") +_RIGHT_WORDS = ("Reverie", "Nocturne", "Bloom", "Drift", "Aurora") + + +def _title(left: int, right: int, dark: int, light: int, mode: str) -> str: + a = _LEFT_WORDS[left] if left >= right else _RIGHT_WORDS[right] + return f"{a} ({mode}) L{left}R{right}D{dark}Li{light}" + + +def generate_fixture_catalog(seed: int = 1729) -> list[Record]: + """One valid Record per coordinate cell (625 total), deterministic for a seed.""" + rng = random.Random(seed) + records: list[Record] = [] + n = 0 + for left in range(5): + for right in range(5): + for dark in range(5): + for light in range(5): + mode = rng.choice(MODES) + status = rng.choice(("proposed", "approved")) + is_video = mode in ("video", "av") + records.append( + Record( + id=f"fx-{n:04d}", + title=_title(left, right, dark, light, mode), + source_url=f"https://example.test/fx/{n:04d}", + source_archive=rng.choice(ARCHIVES), + license="public_domain", + mode=mode, + left=left, + right=right, + dark=dark, + light=light, + duration_s=rng.choice((300, 480, 600, 720, 900)), + file_path="", + review_status=status, + resolution="1920x1080" if is_video else "", + rationale=f"fixture at ({left},{right},{dark},{light})", + reviewed_at="2026-06-04T00:00:00Z" if status == "approved" else None, + ) + ) + n += 1 + return records +``` + +> Note: with a fixed seed the mode draw is deterministic but not guaranteed to hit all three modes across 625 draws in theory; in practice it always does. The test `test_fixture_catalog_has_every_content_mode` guards this for the default seed. If a future seed change ever broke it, force the first three records to the three modes. + +- [ ] **Step 4: Run tests to verify they pass** + +Run: `python -m pytest tests/test_fixtures.py -v` +Expected: PASS (all 6). + +- [ ] **Step 5: Commit** + +```bash +git add simulator/fixtures.py tests/test_fixtures.py +git commit -m "feat(simulator): deterministic synthetic fixture catalog" +``` + +--- + +### Task 4: FastAPI app factory + `/api/select` + `/api/catalog/meta` + +**Files:** +- Create: `simulator/app.py` +- Test: `tests/test_simulator_api.py` (Create) + +- [ ] **Step 1: Write the failing tests** + +Create `tests/test_simulator_api.py`: + +```python +import pytest +from fastapi.testclient import TestClient + +from hef.catalog import Record +from simulator.app import create_app + + +def make_record(**overrides): + base = dict( + id="r", + title="t", + source_url="u", + source_archive="internet_archive", + license="public_domain", + mode="video", + left=0, + right=0, + dark=0, + light=0, + duration_s=600, + file_path="", + ) + base.update(overrides) + return Record(**base) + + +@pytest.fixture +def client(): + records = [ + make_record(id="v-near", mode="video", left=0, right=0, dark=0, light=0), + make_record(id="v-far", mode="video", left=4, right=4, dark=4, light=4), + make_record(id="a-one", mode="audio", left=1, right=1, dark=1, light=1), + make_record(id="prop", mode="video", left=0, right=0, dark=0, light=1, + review_status="proposed"), + make_record(id="appr", mode="video", left=0, right=0, dark=0, light=1, + review_status="approved"), + ] + return TestClient(create_app(records=records)) + + +def _body(**overrides): + base = dict(left=0, right=0, dark=0, light=0, mode="video") + base.update(overrides) + return base + + +def test_select_returns_pick_and_ranked_pool(client): + resp = client.post("/api/select", json=_body(mode="video", pool_size=4)) + assert resp.status_code == 200 + data = resp.json() + assert data["pick"]["id"] == "v-near" + ids = [c["record"]["id"] for c in data["pool"]] + assert ids[0] == "v-near" + assert all("distance" in c and "rank" in c for c in data["pool"]) + assert [c["rank"] for c in data["pool"]] == list(range(1, len(data["pool"]) + 1)) + + +def test_none_mode_is_the_void(client): + resp = client.post("/api/select", json=_body(mode="none")) + assert resp.status_code == 200 + data = resp.json() + assert data["pick"] is None + assert data["pool"] == [] + + +def test_dial_out_of_range_is_rejected(client): + resp = client.post("/api/select", json=_body(left=7)) + assert resp.status_code == 422 + + +def test_bad_mode_is_rejected(client): + resp = client.post("/api/select", json=_body(mode="banana")) + assert resp.status_code == 422 + + +def test_approved_only_narrows_pool(client): + resp = client.post("/api/select", json=_body(left=0, right=0, dark=0, light=1, + mode="video", approved_only=True)) + data = resp.json() + assert all(c["record"]["review_status"] == "approved" for c in data["pool"]) + + +def test_catalog_meta_reports_counts(client): + resp = client.get("/api/catalog/meta") + assert resp.status_code == 200 + data = resp.json() + assert data["total"] == 5 + assert data["by_mode"]["video"] == 4 + assert data["by_mode"]["audio"] == 1 + assert set(data["by_status"]) == {"proposed", "approved"} +``` + +- [ ] **Step 2: Run tests to verify they fail** + +Run: `python -m pytest tests/test_simulator_api.py -v` +Expected: FAIL — `ImportError: cannot import name 'create_app' from 'simulator.app'`. + +- [ ] **Step 3: Write the app factory** + +Create `simulator/app.py`: + +```python +"""FastAPI service: dials -> real hef.selection -> X-ray (pick + ranked pool).""" + +from __future__ import annotations + +import os +from collections import Counter +from pathlib import Path +from typing import Literal, Optional + +from fastapi import FastAPI +from fastapi.staticfiles import StaticFiles +from pydantic import BaseModel, Field + +from hef.catalog import load_catalog, record_to_dict +from hef.selection import ( + Coordinate, + Weights, + candidates_for_mode, + ranked_candidates, + select, +) +from simulator.fixtures import generate_fixture_catalog + +STATIC_DIR = Path(__file__).parent / "static" + + +class SelectRequest(BaseModel): + left: int = Field(ge=0, le=4) + right: int = Field(ge=0, le=4) + dark: int = Field(ge=0, le=4) + light: int = Field(ge=0, le=4) + mode: Literal["none", "audio", "video", "av"] + pool_size: int = Field(default=4, ge=1, le=25) + brain_weight: float = Field(default=1.0, ge=0.0) + mood_weight: float = Field(default=1.0, ge=0.0) + approved_only: bool = False + + +def load_catalog_or_fixtures() -> list: + """Use the real catalog if a non-empty one is configured/exists, else fixtures.""" + configured = os.environ.get("HEF_SIM_CATALOG") + path = Path(configured) if configured else Path("catalog/library.jsonl") + if path.exists() and path.stat().st_size > 0: + return load_catalog(path) + return generate_fixture_catalog() + + +def create_app(records: Optional[list] = None) -> FastAPI: + app = FastAPI(title="HEF Experience Simulator") + app.state.catalog = records if records is not None else load_catalog_or_fixtures() + + @app.post("/api/select") + def api_select(req: SelectRequest): + catalog = app.state.catalog + coord = Coordinate(req.left, req.right, req.dark, req.light) + weights = Weights(brain=req.brain_weight, mood=req.mood_weight) + if req.mode == "none": + return {"pick": None, "pool": [], "coverage": {"candidates_in_mode": 0}} + pool = catalog + if req.approved_only: + pool = [r for r in pool if r.review_status == "approved"] + eligible = candidates_for_mode(pool, req.mode, req.pool_size) + ranked = ranked_candidates( + catalog, coord, req.mode, + pool_size=req.pool_size, weights=weights, approved_only=req.approved_only, + ) + pick = select( + catalog, coord, req.mode, + pool_size=req.pool_size, weights=weights, approved_only=req.approved_only, + rng=None, + ) + return { + "pick": record_to_dict(pick) if pick else None, + "pool": [ + {"record": record_to_dict(r), "distance": d, "rank": i + 1} + for i, (r, d) in enumerate(ranked) + ], + "coverage": {"candidates_in_mode": len(eligible)}, + } + + @app.get("/api/catalog/meta") + def api_meta(): + catalog = app.state.catalog + return { + "total": len(catalog), + "by_mode": dict(Counter(r.mode for r in catalog)), + "by_status": dict(Counter(r.review_status for r in catalog)), + } + + if STATIC_DIR.exists(): + app.mount("/", StaticFiles(directory=STATIC_DIR, html=True), name="static") + + return app + + +app = create_app() +``` + +> The `/` static mount is registered last so the `/api/*` routes win. The mount is guarded by `STATIC_DIR.exists()` so this module imports cleanly before Task 5 creates the static files. + +- [ ] **Step 4: Run tests to verify they pass** + +Run: `python -m pytest tests/test_simulator_api.py -v` +Expected: PASS (all 6). + +- [ ] **Step 5: Commit** + +```bash +git add simulator/app.py tests/test_simulator_api.py +git commit -m "feat(simulator): FastAPI app with /api/select and /api/catalog/meta" +``` + +--- + +### Task 5: The X-ray frontend + +**Files:** +- Create: `simulator/static/index.html` +- Create: `simulator/static/style.css` +- Create: `simulator/static/app.js` +- Test: extend `tests/test_simulator_api.py` with a static-serving smoke test + +No JS unit harness exists; the frontend is validated by the smoke test (it is served) plus the BDD scenarios at run time. + +- [ ] **Step 1: Write the failing smoke test** + +Append to `tests/test_simulator_api.py`: + +```python +from simulator.app import create_app as _create_app_for_static + + +def test_index_is_served(): + # The default app mounts the real static dir. + client = TestClient(_create_app_for_static()) + resp = client.get("/") + assert resp.status_code == 200 + assert "text/html" in resp.headers["content-type"] + assert "X-ray" in resp.text +``` + +- [ ] **Step 2: Run it to verify it fails** + +Run: `python -m pytest tests/test_simulator_api.py::test_index_is_served -v` +Expected: FAIL — 404 (no `index.html` yet), or assertion error. + +- [ ] **Step 3: Create `simulator/static/index.html`** + +```html + + +
+ + +