feat(hints): scaffold static phrase lookup server
Context: Phase P1 of the hints engine spec (docs/superpowers/specs/2026-05-03-hints-engine-design.md). Establishes a stateless FastAPI service surface so later phases (LLM rewrite, anti-cheat, adaptive level, ESP32 integration, auto-trigger) have a stable contract to build on. Currently the master ESP32 has no hint endpoint at all — players rely on the static MP3 pool fired ad hoc from the NPC engine. Approach: Standalone package under tools/hints/ that loads the existing npc_phrases.yaml at startup, normalizes the schema, and exposes three endpoints. No project-level dependency churn — the service runs via uv overlay (uv run --with fastapi --with uvicorn --with pyyaml --with pydantic) so it stays decoupled from the unittest-only test suite documented in tests/CLAUDE.md. Changes: - tools/hints/__init__.py + server.py: FastAPI app with /healthz, /hints/ask (puzzle_id+level → base phrase), and /hints/coverage (audit summary). Loader merges canonical hints.* puzzles with V3 P1-P7 puzzles currently mis-nested under endings.* in the YAML — works around the authoring bug without editing source. - Pydantic request validation (level constrained to 1-3), random.choice from the per-level phrase list, structured JSON access logs to stdout per request. - tests/hints/test_server.py: 6 pytest tests via fastapi TestClient covering healthz, coverage shape, happy path against a real puzzle from the YAML, unknown-puzzle 404, invalid-level 422, and a French diacritic round-trip. - Makefile: hints-serve (uvicorn --reload --port 8300) and hints-test (pytest tests/hints/) targets, both via uv overlay so they need zero project pip install. Impact: Unblocks P2 (LiteLLM rewrite via Qwen 72B on MacStudio :4000), P3 (anti-cheat), P4 (adaptive level), P5 (ESP32 REST client), and P6 (auto-trigger). Surfaces a documented YAML schema bug (P1_SON..P7_COFFRE indented under endings:) for later cleanup. No firmware impact, no YAML edits, no LLM calls.
This commit is contained in:
@@ -2,7 +2,7 @@ PYTHON ?= python3
|
||||
SCENARIO ?= game/scenarios/zacus_v2.yaml
|
||||
FRONTEND_DIR ?= frontend-v3
|
||||
|
||||
.PHONY: bootstrap-validators bootstrap-docs scenario-validate audio-validate printables-validate export validate-runtime-bundle content-checks runtime3-compile runtime3-simulate runtime3-verify runtime3-test runtime3-firmware-bundle frontend-typecheck frontend-test frontend-build docs-build docs-serve all-validate images playtest
|
||||
.PHONY: bootstrap-validators bootstrap-docs scenario-validate audio-validate printables-validate export validate-runtime-bundle content-checks runtime3-compile runtime3-simulate runtime3-verify runtime3-test runtime3-firmware-bundle frontend-typecheck frontend-test frontend-build docs-build docs-serve all-validate images playtest hints-serve hints-test
|
||||
|
||||
bootstrap-validators:
|
||||
bash tools/setup/install_validators.sh
|
||||
@@ -69,3 +69,11 @@ playtest:
|
||||
--scenario game/scenarios/zacus_v2.yaml \
|
||||
--playtest game/scenarios/playtests/zacus_v3_60min_tech.playtest.yaml \
|
||||
--snapshot game/scenarios/playtests/snapshots/zacus_v3_60min_tech.snapshot.json
|
||||
|
||||
hints-serve:
|
||||
uv run --with fastapi --with uvicorn --with pyyaml --with pydantic \
|
||||
uvicorn tools.hints.server:app --reload --port 8300
|
||||
|
||||
hints-test:
|
||||
uv run --with fastapi --with uvicorn --with pyyaml --with pydantic --with pytest --with httpx \
|
||||
pytest tests/hints/ -v
|
||||
|
||||
@@ -0,0 +1,158 @@
|
||||
"""Tests for tools.hints.server (P1 — static phrase lookup).
|
||||
|
||||
Run via:
|
||||
make hints-test
|
||||
or:
|
||||
uv run --with fastapi --with uvicorn --with pyyaml --with pydantic \
|
||||
--with pytest --with httpx pytest tests/hints/ -v
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
from fastapi.testclient import TestClient
|
||||
|
||||
REPO_ROOT = Path(__file__).resolve().parents[2]
|
||||
sys.path.insert(0, str(REPO_ROOT))
|
||||
|
||||
from tools.hints.server import ( # noqa: E402 (import after sys.path tweak)
|
||||
DEFAULT_PHRASES_PATH,
|
||||
create_app,
|
||||
load_phrase_bank,
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Shared fixtures
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@pytest.fixture(scope="module")
|
||||
def app():
|
||||
return create_app(DEFAULT_PHRASES_PATH)
|
||||
|
||||
|
||||
@pytest.fixture(scope="module")
|
||||
def client(app):
|
||||
return TestClient(app)
|
||||
|
||||
|
||||
@pytest.fixture(scope="module")
|
||||
def real_puzzle_id():
|
||||
"""First puzzle in the bank that has a non-empty level_1.
|
||||
|
||||
Using a real puzzle from the actual YAML keeps the test honest — if
|
||||
`npc_phrases.yaml` drifts, the suite catches it.
|
||||
"""
|
||||
bank = load_phrase_bank(DEFAULT_PHRASES_PATH)
|
||||
for pid, levels in bank.items():
|
||||
if levels.get("level_1"):
|
||||
return pid
|
||||
pytest.fail("no puzzle with level_1 found in npc_phrases.yaml")
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Health + coverage
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_healthz_returns_ok_with_counts(client):
|
||||
resp = client.get("/healthz")
|
||||
assert resp.status_code == 200
|
||||
body = resp.json()
|
||||
assert body["status"] == "ok"
|
||||
assert body["phrases_loaded"] > 0
|
||||
assert body["puzzles_loaded"] > 0
|
||||
assert body["phrases_path"].endswith("npc_phrases.yaml")
|
||||
|
||||
|
||||
def test_coverage_summary_shape_is_complete(client):
|
||||
resp = client.get("/hints/coverage")
|
||||
assert resp.status_code == 200
|
||||
body = resp.json()
|
||||
|
||||
# Required top-level fields
|
||||
for key in ("total_puzzles", "puzzles_by_level_count",
|
||||
"percent_by_level_count", "per_puzzle"):
|
||||
assert key in body, f"coverage missing field {key!r}"
|
||||
|
||||
# Bucket counts must sum to total
|
||||
bucket_sum = sum(body["puzzles_by_level_count"].values())
|
||||
assert bucket_sum == body["total_puzzles"]
|
||||
assert body["total_puzzles"] >= 1
|
||||
|
||||
# Each per_puzzle row has the expected sub-fields
|
||||
for row in body["per_puzzle"]:
|
||||
assert "puzzle_id" in row
|
||||
assert "levels_present" in row
|
||||
assert "levels_missing" in row
|
||||
assert "phrase_counts" in row
|
||||
assert set(row["phrase_counts"].keys()) == {"level_1", "level_2", "level_3"}
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# /hints/ask happy path + errors
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_hints_ask_happy_path_returns_real_phrase(client, real_puzzle_id):
|
||||
resp = client.post(
|
||||
"/hints/ask",
|
||||
json={"puzzle_id": real_puzzle_id, "level": 1, "session_id": "test-session"},
|
||||
)
|
||||
assert resp.status_code == 200, resp.text
|
||||
body = resp.json()
|
||||
assert body["puzzle_id"] == real_puzzle_id
|
||||
assert body["level"] == 1
|
||||
assert body["source"] == "static"
|
||||
assert isinstance(body["hint"], str)
|
||||
assert len(body["hint"]) > 0
|
||||
|
||||
# The phrase must be one of the real phrases in the bank — guards
|
||||
# against accidental string mangling in the loader.
|
||||
bank = load_phrase_bank(DEFAULT_PHRASES_PATH)
|
||||
assert body["hint"] in bank[real_puzzle_id]["level_1"]
|
||||
|
||||
|
||||
def test_hints_ask_unknown_puzzle_returns_404(client):
|
||||
resp = client.post(
|
||||
"/hints/ask",
|
||||
json={"puzzle_id": "P999_DOES_NOT_EXIST", "level": 1},
|
||||
)
|
||||
assert resp.status_code == 404
|
||||
assert "unknown puzzle_id" in resp.json()["detail"]
|
||||
|
||||
|
||||
def test_hints_ask_invalid_level_returns_422(client, real_puzzle_id):
|
||||
# Pydantic validation rejects out-of-range level → 422 (Unprocessable)
|
||||
resp = client.post(
|
||||
"/hints/ask",
|
||||
json={"puzzle_id": real_puzzle_id, "level": 7},
|
||||
)
|
||||
assert resp.status_code == 422
|
||||
|
||||
|
||||
def test_hints_ask_preserves_french_diacritics(client, real_puzzle_id):
|
||||
"""UTF-8 round-trip must keep accented characters intact."""
|
||||
bank = load_phrase_bank(DEFAULT_PHRASES_PATH)
|
||||
has_accent = any(
|
||||
any(ch in phrase for ch in "éèêàùôîçÉÈÊÀÙÔÎÇ")
|
||||
for level in bank[real_puzzle_id].values()
|
||||
for phrase in level
|
||||
)
|
||||
if not has_accent:
|
||||
pytest.skip(f"{real_puzzle_id} has no accented phrases — skipping diacritic check")
|
||||
|
||||
# Try several requests — at least one should land on an accented phrase
|
||||
found_accent = False
|
||||
for _ in range(20):
|
||||
resp = client.post(
|
||||
"/hints/ask", json={"puzzle_id": real_puzzle_id, "level": 1}
|
||||
)
|
||||
assert resp.status_code == 200
|
||||
if any(ch in resp.json()["hint"] for ch in "éèêàùôîçÉÈÊÀÙÔÎÇ"):
|
||||
found_accent = True
|
||||
break
|
||||
assert found_accent, "diacritics never round-tripped — encoding bug?"
|
||||
@@ -0,0 +1,6 @@
|
||||
"""Hints engine package — FastAPI service for adaptive NPC hints.
|
||||
|
||||
Phase P1: stateless static phrase lookup from `game/scenarios/npc_phrases.yaml`.
|
||||
Future phases (P2+): LLM rewrite, anti-cheat, adaptive level selection,
|
||||
ESP32 integration, auto-trigger. See docs/superpowers/specs/2026-05-03-hints-engine-design.md.
|
||||
"""
|
||||
@@ -0,0 +1,299 @@
|
||||
"""Hints engine — Phase P1 FastAPI server.
|
||||
|
||||
Stateless static phrase lookup from `game/scenarios/npc_phrases.yaml`.
|
||||
Endpoints:
|
||||
- GET /healthz — liveness probe + phrases-loaded count
|
||||
- POST /hints/ask — return base phrase for {puzzle_id, level}
|
||||
- GET /hints/coverage — per-puzzle level coverage audit
|
||||
|
||||
Run:
|
||||
uv run --with fastapi --with uvicorn --with pyyaml --with pydantic \
|
||||
uvicorn tools.hints.server:app --reload --port 8300
|
||||
|
||||
Spec: docs/superpowers/specs/2026-05-03-hints-engine-design.md
|
||||
P1 scope is intentionally narrow: no LLM rewrite, no anti-cheat,
|
||||
no adaptive level selection, no auto-trigger. Those land in P2-P6.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import logging
|
||||
import os
|
||||
import random
|
||||
import re
|
||||
import sys
|
||||
import time
|
||||
from pathlib import Path
|
||||
from typing import Any, Dict, List, Optional
|
||||
|
||||
import yaml
|
||||
from fastapi import FastAPI, HTTPException, Request
|
||||
from pydantic import BaseModel, Field, conint
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Paths and constants
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
REPO_ROOT = Path(__file__).resolve().parents[2]
|
||||
DEFAULT_PHRASES_PATH = REPO_ROOT / "game" / "scenarios" / "npc_phrases.yaml"
|
||||
PHRASES_PATH_ENV = "ZACUS_NPC_PHRASES_PATH"
|
||||
|
||||
# Puzzle IDs that look like V3 numbered puzzles (P1_FOO, P12_BAR, ...).
|
||||
# Authoring drift: `P1_SON`..`P7_COFFRE` are nested under `endings:` in the
|
||||
# current YAML (indentation bug — see audit). We surface them anyway because
|
||||
# their internal `key:` values advertise `hints.<id>.level_N`. Fix later in
|
||||
# `npc_phrases.yaml`; do NOT fix here (P1 instructions forbid YAML edits).
|
||||
PUZZLE_ID_PATTERN = re.compile(r"^P\d+_[A-Z0-9_]+$")
|
||||
|
||||
ALL_LEVELS = (1, 2, 3)
|
||||
|
||||
LOG = logging.getLogger("hints.server")
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Phrase bank loading + validation
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def _extract_text(entry: Any) -> str:
|
||||
"""Return the human-readable phrase from an entry (dict with 'text' or bare str)."""
|
||||
if isinstance(entry, dict):
|
||||
val = entry.get("text", "")
|
||||
return val if isinstance(val, str) else ""
|
||||
if isinstance(entry, str):
|
||||
return entry
|
||||
return ""
|
||||
|
||||
|
||||
def _normalize_level_block(block: Any) -> List[str]:
|
||||
"""Coerce a level block into a list of non-empty strings."""
|
||||
if block is None:
|
||||
return []
|
||||
if isinstance(block, list):
|
||||
out: List[str] = []
|
||||
for entry in block:
|
||||
text = _extract_text(entry).strip()
|
||||
if text:
|
||||
out.append(text)
|
||||
return out
|
||||
if isinstance(block, str):
|
||||
return [block.strip()] if block.strip() else []
|
||||
if isinstance(block, dict):
|
||||
# Tolerate `{text: "..."}` at the level position
|
||||
text = _extract_text(block).strip()
|
||||
return [text] if text else []
|
||||
return []
|
||||
|
||||
|
||||
def load_phrase_bank(path: Path) -> Dict[str, Dict[str, List[str]]]:
|
||||
"""Load and normalize hints from `npc_phrases.yaml`.
|
||||
|
||||
Returns a flat mapping: `{puzzle_id: {"level_1": [str, ...], ...}}`.
|
||||
Pulls from two places to honor the YAML's authorial intent:
|
||||
1. Top-level `hints.<puzzle_id>.level_N` (canonical)
|
||||
2. `endings.<puzzle_id>.level_N` where puzzle_id matches PUZZLE_ID_PATTERN
|
||||
(the V3 P1-P7 hints currently mis-nested under `endings:`)
|
||||
|
||||
Raises:
|
||||
FileNotFoundError if `path` doesn't exist.
|
||||
ValueError if the YAML lacks a top-level `hints` mapping or no usable
|
||||
puzzle entries are found.
|
||||
"""
|
||||
if not path.is_file():
|
||||
raise FileNotFoundError(f"npc_phrases.yaml not found at {path}")
|
||||
|
||||
with path.open("r", encoding="utf-8") as f:
|
||||
data = yaml.safe_load(f)
|
||||
|
||||
if not isinstance(data, dict):
|
||||
raise ValueError("npc_phrases.yaml: root is not a YAML mapping")
|
||||
|
||||
hints_section = data.get("hints")
|
||||
if not isinstance(hints_section, dict):
|
||||
raise ValueError("npc_phrases.yaml: missing or non-dict top-level 'hints'")
|
||||
|
||||
bank: Dict[str, Dict[str, List[str]]] = {}
|
||||
|
||||
# 1. Canonical hints.* puzzles
|
||||
for puzzle_id, levels in hints_section.items():
|
||||
if not isinstance(levels, dict):
|
||||
continue
|
||||
bank[puzzle_id] = {
|
||||
f"level_{n}": _normalize_level_block(levels.get(f"level_{n}"))
|
||||
for n in ALL_LEVELS
|
||||
}
|
||||
|
||||
# 2. Mis-nested V3 puzzles under endings.*
|
||||
endings_section = data.get("endings")
|
||||
if isinstance(endings_section, dict):
|
||||
for puzzle_id, levels in endings_section.items():
|
||||
if not PUZZLE_ID_PATTERN.match(puzzle_id):
|
||||
continue
|
||||
if not isinstance(levels, dict):
|
||||
continue
|
||||
if puzzle_id in bank:
|
||||
continue # canonical hints.* wins
|
||||
bank[puzzle_id] = {
|
||||
f"level_{n}": _normalize_level_block(levels.get(f"level_{n}"))
|
||||
for n in ALL_LEVELS
|
||||
}
|
||||
|
||||
if not bank:
|
||||
raise ValueError(
|
||||
"npc_phrases.yaml: no usable puzzle hints found "
|
||||
"(expected mapping under 'hints.<puzzle_id>.level_N')"
|
||||
)
|
||||
|
||||
return bank
|
||||
|
||||
|
||||
def compute_coverage(bank: Dict[str, Dict[str, List[str]]]) -> Dict[str, Any]:
|
||||
"""Build the audit summary returned by GET /hints/coverage."""
|
||||
per_puzzle: List[Dict[str, Any]] = []
|
||||
buckets = {0: 0, 1: 0, 2: 0, 3: 0}
|
||||
for puzzle_id, levels in sorted(bank.items()):
|
||||
present = [n for n in ALL_LEVELS if levels.get(f"level_{n}")]
|
||||
missing = [n for n in ALL_LEVELS if not levels.get(f"level_{n}")]
|
||||
buckets[len(present)] += 1
|
||||
per_puzzle.append({
|
||||
"puzzle_id": puzzle_id,
|
||||
"levels_present": present,
|
||||
"levels_missing": missing,
|
||||
"phrase_counts": {
|
||||
f"level_{n}": len(levels.get(f"level_{n}", [])) for n in ALL_LEVELS
|
||||
},
|
||||
})
|
||||
total = sum(buckets.values())
|
||||
pct = {
|
||||
str(k): round((v / total) * 100, 1) if total else 0.0
|
||||
for k, v in buckets.items()
|
||||
}
|
||||
return {
|
||||
"total_puzzles": total,
|
||||
"puzzles_by_level_count": {str(k): v for k, v in buckets.items()},
|
||||
"percent_by_level_count": pct,
|
||||
"per_puzzle": per_puzzle,
|
||||
}
|
||||
|
||||
|
||||
def count_phrases(bank: Dict[str, Dict[str, List[str]]]) -> int:
|
||||
"""Total number of distinct phrase strings across the whole bank."""
|
||||
return sum(len(level) for puzzle in bank.values() for level in puzzle.values())
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Pydantic request/response models
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class HintAskRequest(BaseModel):
|
||||
puzzle_id: str = Field(..., min_length=1)
|
||||
level: conint(ge=1, le=3) # type: ignore[valid-type]
|
||||
session_id: Optional[str] = None
|
||||
|
||||
|
||||
class HintAskResponse(BaseModel):
|
||||
hint: str
|
||||
level: int
|
||||
puzzle_id: str
|
||||
source: str = "static"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# FastAPI app factory
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def _resolve_phrases_path() -> Path:
|
||||
override = os.environ.get(PHRASES_PATH_ENV)
|
||||
return Path(override).resolve() if override else DEFAULT_PHRASES_PATH
|
||||
|
||||
|
||||
def _emit_log(record: Dict[str, Any]) -> None:
|
||||
"""Structured JSON log line on stdout (one dict per request)."""
|
||||
sys.stdout.write(json.dumps(record, ensure_ascii=False) + "\n")
|
||||
sys.stdout.flush()
|
||||
|
||||
|
||||
def create_app(phrases_path: Optional[Path] = None) -> FastAPI:
|
||||
"""Build the FastAPI app. Pass `phrases_path` for tests, default for prod."""
|
||||
path = phrases_path or _resolve_phrases_path()
|
||||
bank = load_phrase_bank(path) # raises if cracked → uvicorn refuses to start
|
||||
coverage_cache = compute_coverage(bank)
|
||||
phrase_total = count_phrases(bank)
|
||||
|
||||
app = FastAPI(
|
||||
title="Zacus Hints Engine",
|
||||
version="0.1.0-P1",
|
||||
description="Static NPC hint phrase lookup. Phase P1 of the hints engine spec.",
|
||||
)
|
||||
|
||||
# Stash for handlers + tests
|
||||
app.state.phrase_bank = bank
|
||||
app.state.coverage = coverage_cache
|
||||
app.state.phrase_total = phrase_total
|
||||
app.state.phrases_path = str(path)
|
||||
|
||||
@app.get("/healthz")
|
||||
def healthz() -> Dict[str, Any]:
|
||||
return {
|
||||
"status": "ok",
|
||||
"phrases_loaded": app.state.phrase_total,
|
||||
"puzzles_loaded": len(app.state.phrase_bank),
|
||||
"phrases_path": app.state.phrases_path,
|
||||
}
|
||||
|
||||
@app.get("/hints/coverage")
|
||||
def coverage() -> Dict[str, Any]:
|
||||
return app.state.coverage
|
||||
|
||||
@app.post("/hints/ask", response_model=HintAskResponse)
|
||||
def hints_ask(payload: HintAskRequest, request: Request) -> HintAskResponse:
|
||||
started = time.perf_counter()
|
||||
status_code = 200
|
||||
try:
|
||||
puzzle = app.state.phrase_bank.get(payload.puzzle_id)
|
||||
if puzzle is None:
|
||||
status_code = 404
|
||||
raise HTTPException(
|
||||
status_code=404,
|
||||
detail=f"unknown puzzle_id: {payload.puzzle_id!r}",
|
||||
)
|
||||
level_key = f"level_{payload.level}"
|
||||
phrases = puzzle.get(level_key) or []
|
||||
if not phrases:
|
||||
status_code = 404
|
||||
raise HTTPException(
|
||||
status_code=404,
|
||||
detail=(
|
||||
f"no phrases for {payload.puzzle_id} at {level_key} "
|
||||
f"(available: {[k for k, v in puzzle.items() if v]})"
|
||||
),
|
||||
)
|
||||
hint = random.choice(phrases)
|
||||
return HintAskResponse(
|
||||
hint=hint,
|
||||
level=payload.level,
|
||||
puzzle_id=payload.puzzle_id,
|
||||
)
|
||||
except HTTPException as exc:
|
||||
status_code = exc.status_code
|
||||
raise
|
||||
finally:
|
||||
_emit_log({
|
||||
"ts": time.time(),
|
||||
"event": "hints_ask",
|
||||
"puzzle_id": payload.puzzle_id,
|
||||
"level": payload.level,
|
||||
"session_id": payload.session_id,
|
||||
"client": request.client.host if request.client else None,
|
||||
"latency_ms": round((time.perf_counter() - started) * 1000, 2),
|
||||
"status": status_code,
|
||||
})
|
||||
|
||||
return app
|
||||
|
||||
|
||||
# Module-level app for `uvicorn tools.hints.server:app`
|
||||
app = create_app()
|
||||
Reference in New Issue
Block a user