Capture harness, fixture verification, CI, and the public README
The site does no live inference. Rollouts are captured once against spark-1 and replayed at their recorded wall-clock — a public demo with no auth cannot hold an API key, and a recorded run can be scrubbed, permalinked, blind-compared and verified in ways a live one cannot. What stops it being a video is that the browser re-derives every number from the recorded moves. verify_fixtures.py is the Python half of that: it replays every committed fixture through the engine and reproduces its own rewards. All 16 land at delta 0.0. A fixture that cannot be regenerated is a claim with no receipt. First real measurement, thinking off, 8 seeds: solved 0/8. The model repeats guesses it has already played, invents words (trape, slith, postt, boomy), and contradicts its own feedback — consistency 0.09 to 0.17. That is the published failure taxonomy showing up in our own data on the first run, and it is why `consistency` is a reward component rather than a footnote. A capture failure is recorded as a turn with a null reply, never dropped. A capture that silently discarded failed turns would be reporting a better model than the one that ran. CI gates both halves and four things that fail silently in production: the word lists must rebuild byte-identically, the prerendered routes must carry their own baked og tags (crawlers do not run JS, so without them every shared link previews as the homepage), no blob: URL may reach the bundle (the site's CSP has no worker-src, so it falls back to default-src 'self' and a blob worker is blocked with no error), and the conformance digest must match across languages. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_019mt6sHQHEnEYrJZvoMCJSB
This commit is contained in:
karti-ai
+228
@@ -0,0 +1,228 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Capture real rollouts into the fixtures the site replays.
|
||||
|
||||
The site does no live inference. That is a deliberate architecture choice, not
|
||||
a limitation: a public demo with no auth cannot hold an API key, a live call is
|
||||
slow and flaky on conference wifi, and a recorded run can be scrubbed,
|
||||
verified, permalinked and blind-compared in ways a live one cannot.
|
||||
|
||||
What makes it honest rather than a video is that the browser re-derives every
|
||||
number it shows from the recorded moves, and says so. See verify.ts.
|
||||
|
||||
Usage:
|
||||
uv run python envs/capture.py --arm base-off --seeds 0-7
|
||||
uv run python envs/capture.py --arm solver --seeds 0-7 # no model needed
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
import time
|
||||
import urllib.error
|
||||
import urllib.request
|
||||
from pathlib import Path
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).parent / "wordle_five"))
|
||||
|
||||
import numpy as np # noqa: E402
|
||||
|
||||
from wordle_five import solver as S # noqa: E402
|
||||
from wordle_five.engine import MAX_GUESSES, Game, answers # noqa: E402
|
||||
from wordle_five.protocol import ( # noqa: E402
|
||||
parse_guess,
|
||||
render_feedback,
|
||||
render_rejection,
|
||||
system_prompt,
|
||||
)
|
||||
from wordle_five.reward import Episode, metrics, score # noqa: E402
|
||||
|
||||
OUT = Path(__file__).parent.parent / "public" / "traces" / "wordle"
|
||||
ENDPOINT = os.environ.get("PIG_DEMO_INFERENCE", "http://100.127.247.67:8001/v1/chat/completions")
|
||||
MODEL = os.environ.get("PIG_DEMO_MODEL", "brain-qwen38-dspark")
|
||||
|
||||
ARMS = {
|
||||
"base-off": {"label": "Out of the box", "thinking": False},
|
||||
"base-on": {"label": "Allowed to think", "thinking": True},
|
||||
"solver": {"label": "Best-known play", "thinking": None},
|
||||
}
|
||||
|
||||
|
||||
def call_model(messages: list[dict], thinking: bool) -> dict:
|
||||
"""One completion. Returns reply, reasoning and the real call metrics.
|
||||
|
||||
Never raises on an upstream failure — a dropped call becomes a turn with a
|
||||
null reply, which the game scores as a rejected guess. A capture that
|
||||
silently discarded failed turns would be reporting a better model than the
|
||||
one that ran.
|
||||
"""
|
||||
body = {
|
||||
"model": MODEL,
|
||||
"messages": messages,
|
||||
"temperature": 0.7,
|
||||
"max_tokens": 2048 if thinking else 512,
|
||||
"chat_template_kwargs": {"enable_thinking": bool(thinking)},
|
||||
}
|
||||
request = urllib.request.Request(
|
||||
ENDPOINT,
|
||||
data=json.dumps(body).encode(),
|
||||
headers={"Content-Type": "application/json"},
|
||||
)
|
||||
started = time.time()
|
||||
try:
|
||||
with urllib.request.urlopen(request, timeout=300) as response:
|
||||
payload = json.loads(response.read())
|
||||
except (urllib.error.URLError, TimeoutError, OSError) as exc:
|
||||
return {
|
||||
"reply": None,
|
||||
"reasoning": None,
|
||||
"call": {
|
||||
"promptTokens": None,
|
||||
"completionTokens": None,
|
||||
"reasoningTokens": None,
|
||||
"durationMs": round((time.time() - started) * 1000),
|
||||
"finishReason": f"error: {type(exc).__name__}",
|
||||
},
|
||||
}
|
||||
|
||||
elapsed = round((time.time() - started) * 1000)
|
||||
choice = payload["choices"][0]
|
||||
message = choice.get("message", {})
|
||||
usage = payload.get("usage", {}) or {}
|
||||
details = usage.get("completion_tokens_details") or {}
|
||||
return {
|
||||
"reply": message.get("content"),
|
||||
"reasoning": message.get("reasoning_content"),
|
||||
"call": {
|
||||
"promptTokens": usage.get("prompt_tokens"),
|
||||
"completionTokens": usage.get("completion_tokens"),
|
||||
"reasoningTokens": details.get("reasoning_tokens"),
|
||||
"durationMs": elapsed,
|
||||
"finishReason": choice.get("finish_reason"),
|
||||
},
|
||||
}
|
||||
|
||||
|
||||
def solver_turn(game: Game) -> dict:
|
||||
"""The reference player, recorded in the same shape as a model turn.
|
||||
|
||||
durationMs is null rather than invented: nothing waited for this, and the
|
||||
player must not pretend otherwise. The UI renders a null duration as an
|
||||
instant step and labels the run as generated.
|
||||
"""
|
||||
pool = answers()
|
||||
if not game.history:
|
||||
guess = pool[S._best_opener()]
|
||||
else:
|
||||
alive = S.consistent_candidates(game.history)
|
||||
guess = (
|
||||
pool[S.best_guess(np.array([pool.index(w) for w in alive]))] if alive else "tares"
|
||||
)
|
||||
return {
|
||||
"reply": f"[{guess}]",
|
||||
"reasoning": None,
|
||||
"call": {
|
||||
"promptTokens": None,
|
||||
"completionTokens": None,
|
||||
"reasoningTokens": None,
|
||||
"durationMs": None,
|
||||
"finishReason": "generated",
|
||||
},
|
||||
}
|
||||
|
||||
|
||||
def capture(arm: str, seed: int) -> dict:
|
||||
config = ARMS[arm]
|
||||
game = Game(seed=seed)
|
||||
messages = [
|
||||
{"role": "system", "content": system_prompt()},
|
||||
{"role": "user", "content": "Enter your guess to begin."},
|
||||
]
|
||||
turns: list[dict] = []
|
||||
|
||||
while not game.over and len(turns) < MAX_GUESSES * 2:
|
||||
if arm == "solver":
|
||||
result = solver_turn(game)
|
||||
else:
|
||||
result = call_model(messages, bool(config["thinking"]))
|
||||
|
||||
guess = parse_guess(result["reply"])
|
||||
left = MAX_GUESSES - len(game.history)
|
||||
|
||||
if guess is None:
|
||||
game.rejected += 1
|
||||
observation = render_rejection("no bracketed guess found", left)
|
||||
else:
|
||||
pattern, rejection = game.play(guess)
|
||||
observation = (
|
||||
render_rejection(rejection, left)
|
||||
if rejection is not None
|
||||
else render_feedback(guess, pattern or "", MAX_GUESSES - len(game.history))
|
||||
)
|
||||
|
||||
turns.append({**result, "info": {"guess": guess, "observation": observation}})
|
||||
messages.append({"role": "assistant", "content": result["reply"] or ""})
|
||||
messages.append({"role": "user", "content": observation})
|
||||
|
||||
if game.rejected >= MAX_GUESSES:
|
||||
break
|
||||
|
||||
episode = Episode(
|
||||
answer=game.answer,
|
||||
guesses=[g for g, _ in game.history],
|
||||
patterns=[p for _, p in game.history],
|
||||
rejected=game.rejected,
|
||||
reference_depth=S.reference_depth(game.answer),
|
||||
)
|
||||
# A run that never reached a terminal state is marked truncated, and the
|
||||
# browser renders its verification as "unverifiable" rather than as a zero.
|
||||
truncated = not game.over and game.rejected < MAX_GUESSES
|
||||
|
||||
return {
|
||||
"runId": f"{arm}-s{seed}",
|
||||
"seed": seed,
|
||||
"model": MODEL if arm != "solver" else "entropy-solver",
|
||||
"capturedAt": time.strftime("%Y-%m-%d"),
|
||||
"rewards": score(episode),
|
||||
"metrics": metrics(episode),
|
||||
"truncated": truncated,
|
||||
"outcome": "solved" if game.solved else "failed",
|
||||
"answer": game.answer,
|
||||
"turns": turns,
|
||||
}
|
||||
|
||||
|
||||
def parse_seeds(spec: str) -> list[int]:
|
||||
if "-" in spec:
|
||||
lo, hi = spec.split("-")
|
||||
return list(range(int(lo), int(hi) + 1))
|
||||
return [int(s) for s in spec.split(",")]
|
||||
|
||||
|
||||
def main() -> int:
|
||||
parser = argparse.ArgumentParser()
|
||||
parser.add_argument("--arm", required=True, choices=sorted(ARMS))
|
||||
parser.add_argument("--seeds", default="0-7")
|
||||
args = parser.parse_args()
|
||||
|
||||
OUT.mkdir(parents=True, exist_ok=True)
|
||||
for seed in parse_seeds(args.seeds):
|
||||
started = time.time()
|
||||
episode = capture(args.arm, seed)
|
||||
path = OUT / f"{episode['runId']}.json"
|
||||
path.write_text(json.dumps(episode, indent=2) + "\n")
|
||||
guesses = [t["info"]["guess"] for t in episode["turns"]]
|
||||
print(
|
||||
f"{episode['runId']:>16} {episode['answer']} {episode['outcome']:<7}"
|
||||
f" solved={episode['rewards']['solved']:.0f}"
|
||||
f" econ={episode['rewards']['economy']:.2f}"
|
||||
f" cons={episode['rewards']['consistency']:.2f}"
|
||||
f" {time.time()-started:5.1f}s {guesses}"
|
||||
)
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
@@ -0,0 +1,79 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Replay every committed fixture and reproduce its recorded rewards.
|
||||
|
||||
A fixture is a claim: "this model, on this seed, scored this." The claim is
|
||||
only worth anything if it can be re-derived from the moves it records. This
|
||||
does that in Python; `verify.ts` does the same thing in the browser, live, in
|
||||
front of the visitor.
|
||||
|
||||
Exits non-zero if any fixture's rewards cannot be reproduced from its own turns.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).parent / "wordle_five"))
|
||||
|
||||
from wordle_five import solver as S # noqa: E402
|
||||
from wordle_five.engine import Game # noqa: E402
|
||||
from wordle_five.protocol import parse_guess # noqa: E402
|
||||
from wordle_five.reward import Episode, score # noqa: E402
|
||||
|
||||
TRACES = Path(__file__).parent.parent / "public" / "traces" / "wordle"
|
||||
TOLERANCE = 1e-9
|
||||
|
||||
|
||||
def replay(fixture: dict) -> dict[str, float]:
|
||||
game = Game(seed=fixture["seed"])
|
||||
if game.answer != fixture["answer"]:
|
||||
raise AssertionError(
|
||||
f"seed {fixture['seed']} gives {game.answer!r}, fixture claims {fixture['answer']!r}"
|
||||
)
|
||||
for turn in fixture["turns"]:
|
||||
guess = parse_guess(turn["reply"])
|
||||
if guess is None:
|
||||
game.rejected += 1
|
||||
continue
|
||||
game.play(guess)
|
||||
|
||||
episode = Episode(
|
||||
answer=game.answer,
|
||||
guesses=[g for g, _ in game.history],
|
||||
patterns=[p for _, p in game.history],
|
||||
rejected=game.rejected,
|
||||
reference_depth=S.reference_depth(game.answer),
|
||||
)
|
||||
return score(episode)
|
||||
|
||||
|
||||
def main() -> int:
|
||||
fixtures = sorted(TRACES.glob("*.json"))
|
||||
fixtures = [f for f in fixtures if f.name != "manifest.json"]
|
||||
if not fixtures:
|
||||
print("no fixtures found — nothing to verify")
|
||||
return 0
|
||||
|
||||
failures = 0
|
||||
for path in fixtures:
|
||||
data = json.loads(path.read_text())
|
||||
recomputed = replay(data)
|
||||
recorded = data["rewards"]
|
||||
deltas = {k: abs(recomputed[k] - recorded[k]) for k in recorded}
|
||||
worst = max(deltas.values())
|
||||
status = "ok" if worst <= TOLERANCE else "MISMATCH"
|
||||
if worst > TOLERANCE:
|
||||
failures += 1
|
||||
culprit = max(deltas, key=deltas.get)
|
||||
print(f"{path.name:>20} {status} worst delta {worst:.3g} on '{culprit}'")
|
||||
else:
|
||||
print(f"{path.name:>20} {status} delta {worst:.1e}")
|
||||
|
||||
print(f"\n{len(fixtures)} fixtures, {failures} mismatched")
|
||||
return 1 if failures else 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
Reference in New Issue
Block a user