diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index c526d698..51c8649b 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -100,7 +100,10 @@ jobs: ../../qa/test_root_cause_analyzer.py \ ../../qa/test_evidence_audit.py \ ../../qa/test_triage_failure.py \ - ../../qa/test_behavioral_gate_corpus.py + ../../qa/test_behavioral_gate_corpus.py \ + ../../qa/test_deterministic_rri_gate.py \ + ../../qa/test_orchestrate_split_rri.py \ + ../../qa/test_visual_regression_check.py server-contracts: # Phase-1 (DETERMINISTIC-2): deterministic cross-service MCP contract tests diff --git a/.github/workflows/release-readiness.yml b/.github/workflows/release-readiness.yml new file mode 100644 index 00000000..0fbacf49 --- /dev/null +++ b/.github/workflows/release-readiness.yml @@ -0,0 +1,124 @@ +name: Release Readiness (deterministic, advisory) + +# The full Release Readiness Index (RRI) needs a live five-persona sweep + the Mac built-app +# handoff, which CI cannot mint. But the DETERMINISTIC subset of the release gates — native +# built-app transition, ui-audit, image-render, palette-live, and the additive per-beat latency +# budget — needs NO live LLM/persona evidence. This cadence runs +# `qa/release_readiness.py --deterministic-only` as an EARLY advisory answer to "do the +# deterministic release gates hold?" — the LLM/persona gates are reported SKIPPED (never FAILED). +# +# ADVISORY, NEVER BLOCKING. `continue-on-error` keeps the job from ever going red / blocking a +# merge; a below-bar deterministic result ANNOTATES (::warning) and uploads RRI-deterministic.json +# as an artifact for triage. This is a signal, not a gate. +# +# Gateway-free / null-backend: no live model, no Eva, no gateway, no global mcp config. It builds +# a small deterministic fixture in a TEMP dir and points the reader at it, then writes the rollup +# to the WORKSPACE artifact path — NEVER the committed qa/RRI.json / qa/scores.db / ledger / +# transcripts. The reader is a pure on-disk reader (the engine stays the sole state writer). + +on: + # Manual dispatch (the agent / a maintainer can ask "does the deterministic subset hold?"). + workflow_dispatch: + inputs: + build_sha: + description: "Build SHA to stamp the deterministic rollup with (defaults to the checked-out HEAD)." + required: false + default: "" + # Standing cadence. 23:41 UTC daily, off the top of the hour to dodge the cron thundering herd. + schedule: + - cron: "41 23 * * *" + +# One advisory run at a time; a newer trigger supersedes an in-flight run. +concurrency: + group: release-readiness-deterministic + cancel-in-progress: false + +permissions: + contents: read + +jobs: + deterministic-rri: + runs-on: ubuntu-latest + # ADVISORY: never blocks. A below-bar deterministic subset (or a harness hiccup) annotates + # and uploads the rollup; it does not fail the workflow. + continue-on-error: true + steps: + - uses: actions/checkout@v4 + + - name: Install uv + uses: astral-sh/setup-uv@v5 + + - name: Run release_readiness.py --deterministic-only (advisory) + id: rri + # Builds a deterministic five-persona fixture in a TEMP dir (clean part_a PASS, 200 + # image traffic, healthy under-budget latency) so the deterministic gate machinery runs + # end-to-end WITHOUT a live model. The rollup JSON is written to the workspace artifact + # path. We never let a non-zero exit fail the step — this cadence ALERTS, never blocks. + run: | + set -uo pipefail + BUILD_SHA="${{ github.event.inputs.build_sha }}" + [ -n "$BUILD_SHA" ] || BUILD_SHA="$(git rev-parse --short=12 HEAD)" + echo "build_sha=$BUILD_SHA" + + FIX="$(mktemp -d)" + ART="${GITHUB_WORKSPACE}/RRI-deterministic.json" + echo "ART=$ART" >> "$GITHUB_ENV" + + # Deterministic evidence inputs (no live model / persona involved). + printf '{"overall": 5}\n' > "$FIX/story.json" + printf '{"overall": 5}\n' > "$FIX/mech.json" + printf 'GREEN\n' > "$FIX/behavioral.txt" + printf 'PASS\n' > "$FIX/audit.log" + printf '{"can_act": true}\n' > "$FIX/session_surface.final.json" + + RUNS="" + for persona in newbie veteran adversarial narrative optimizer; do + d="$FIX/gate-$persona" + mkdir -p "$d/player" + printf '{"run":"gate-%s","persona":"%s","completed_intro_flow":true,"persona_satisfaction":9,"gave_up":false,"bug_reports_critical":0,"console_errors":0,"image_404s":0}\n' \ + "$persona" "$persona" > "$d/score.json" + printf '{"build_sha":"%s","part_a":{"result":"PASS"},"part_b":{"persona_loop":"PASS","score_pass":true}}\n' \ + "$BUILD_SHA" > "$d/run.json" + printf '{"url":"http://127.0.0.1/image?scope=%s","status":200}\n' "$persona" > "$d/player/network.ndjson" + # Healthy under-budget latency sidecar (the additive latency gate path exercises). + printf '{"s_per_beat":78.2,"coldopen_s":157.0,"turns_per_beat":4.4}\n' > "$d/latency.json" + RUNS="${RUNS:+$RUNS,}$d" + done + + # --deterministic-only: evaluates ONLY the gates that need no live LLM/persona evidence + # and marks the LLM/persona gates SKIPPED. Pure on-disk reader; --out is the workspace + # artifact, never the committed qa/RRI.json. + uv run --directory servers/engine --no-project python "${GITHUB_WORKSPACE}/qa/release_readiness.py" \ + --runs "$RUNS" \ + --expected-personas "newbie,veteran,adversarial,narrative,optimizer" \ + --story "$FIX/story.json" --mech "$FIX/mech.json" \ + --behavioral GREEN --behavioral-path "$FIX/behavioral.txt" \ + --ui-audit PASS --ui-audit-log "$FIX/audit.log" \ + --palette-live true --palette-source "$FIX/session_surface.final.json" \ + --build-sha "$BUILD_SHA" \ + --deterministic-only \ + --out "$ART" | tee "$FIX/rri.log" || true + + rm -rf "$FIX"/gate-* "$FIX"/*.json "$FIX"/*.txt "$FIX"/*.log "$FIX" 2>/dev/null || true + + if [ ! -s "$ART" ]; then + echo "::warning title=Release Readiness::deterministic rollup produced no RRI-deterministic.json — see logs (advisory)." + exit 0 + fi + + dpass="$(python3 -c 'import json,sys;print(json.load(open(sys.argv[1])).get("deterministic_pass"))' "$ART" 2>/dev/null || echo unknown)" + dfail="$(python3 -c 'import json,sys;print(",".join(json.load(open(sys.argv[1])).get("deterministic_failed_gates") or []) or "none")' "$ART" 2>/dev/null || echo unknown)" + if [ "$dpass" = "True" ]; then + echo "::notice title=Release Readiness::Deterministic release gates HOLD (deterministic_pass=true). LLM/persona gates skipped (advisory)." + else + echo "::warning title=Release Readiness::Deterministic release gates DID NOT all hold (deterministic_pass=$dpass; failed: $dfail). Advisory — not blocking. See RRI-deterministic.json." + fi + + - name: Upload RRI-deterministic.json + if: always() + uses: actions/upload-artifact@v4 + with: + name: rri-deterministic-${{ github.run_id }} + path: RRI-deterministic.json + if-no-files-found: ignore + retention-days: 14 diff --git a/qa/latency_baseline.json b/qa/latency_baseline.json new file mode 100644 index 00000000..417330e4 --- /dev/null +++ b/qa/latency_baseline.json @@ -0,0 +1,11 @@ +{ + "schema": "worldos.latency-baseline.v1", + "_comment": "ADDITIVE per-beat latency BUDGET for the release-readiness latency gates (Phase-3). These budgets gate ONLY when a run actually carries latency evidence AND that evidence exceeds the budget; when latency data is ABSENT the gate is a documented EVIDENCE-GAP/skip, never a new false fail, so every existing RRI result is unchanged. Budgets are the healthy ledger figures (qa/scores_ledger.md: duo-baseline s/beat ~78.2, cold-open ~157-161) with headroom so normal scorer/host variance does not trip the gate. s/beat is the MEAN GENERATION seconds over CONTINUING (routine) beats (cold open excluded); coldopen_s is the one-time world-build first beat. Both are sourced from the same on-disk artifacts release_readiness already reads (a run's latency.json sidecar, or a latency block in run.json / score.json) via qa/latency_rollup.py's columns.", + "s_per_beat_budget": 120.0, + "coldopen_s_budget": 240.0, + "_baseline_evidence": { + "s_per_beat_healthy": 78.2, + "coldopen_s_healthy": 157.0, + "source": "qa/scores_ledger.md duo-baseline-rc1b (s/beat 78.2, cold-open 161) and the GUI VM sweeps (cold-open 156.6); budgets add headroom above the healthy figure so routine variance does not false-fail." + } +} diff --git a/qa/orchestrate_split_rri.py b/qa/orchestrate_split_rri.py new file mode 100644 index 00000000..47df3c43 --- /dev/null +++ b/qa/orchestrate_split_rri.py @@ -0,0 +1,486 @@ +#!/usr/bin/env python3 +"""WorldOS split-lane RRI orchestrator — run the VM(part-B personas)+Mac(part-A +handoff) RRI rollup as ONE coordinated, auditable step instead of hand-run SSH/scp. + +WHY THIS EXISTS + The support-VM lane (heavy 5-persona real-browser sweep) and the Mac lane + (built-app native #356 + handoff.json) must roll up into ONE same-SHA RRI. Doing + it by hand is a long fragile chain of `ssh`/`scp`/preflight/`release_readiness.py` + steps, each a place a stale build / mixed SHA / silent auth failure can produce a + meaningless green. This tool composes the EXISTING primitives: + qa/support_vm_preflight.py — the same-SHA / auth / tool / private-art readiness gate + qa/ui_playtest_app.sh — the part-B persona sweep (run ON the VM) + qa/release_readiness.py — the RRI rollup (--handoff-json + --support-preflight-json) + It does NOT reimplement any of them; it sequences them and refuses loudly on the traps. + +HARD SAFETY (the whole point of the default mode) + The actual remote SSH / persona-run step is OPERATOR-APPROVED ONLY: + * The DEFAULT mode is --plan (a DRY RUN). It PRINTS the exact commands it WOULD + run (preflight, persona sweep, fetch, rollup) and executes NOTHING remote. + * Remote execution happens ONLY behind an explicit --execute flag, and even then + the SSH command is shown FIRST. + * Importing this module SSHes nowhere; the default runner is a no-op placeholder. + * Connection/auth details (VM host, key path, remote checkout) are NEVER hardcoded + — the operator supplies them as flags (the runbook keeps them in operator-only + runbooks, not the tracked repo). + Before ANY execution (or even when assembling the rollup) the tool REFUSES when the + support preflight is missing/blocked or the build SHA does not match the preflight / + handoff — a mismatched-SHA rollup is exactly the corruption RRI exists to prevent. + +This module is a pure reader/reporter + an opt-in, operator-gated remote step. It +does not write any committed data artifact (qa/RRI.json etc.) on its own — the RRI +output path is operator-chosen and the rollup is the operator's to run. + +Usage: + orchestrate_split_rri.py \ + --build-sha SHA \ + --handoff-json /path/to/handoff.json \ + --support-preflight-json /path/to/support_vm_preflight.json \ + --ssh-host root@ --ssh-key /path/to/key \ + --remote-repo /root/worldos-qa/WorldOS \ + [--vm-run-root /root/worldos-qa/runs] [--local-fetch-dir DIR] \ + [--rri-out qa/RRI.json] [--personas newbie,veteran,...] \ + [--plan | --execute] [--json] +""" +from __future__ import annotations + +import argparse +import json +import shlex +import sys +from dataclasses import dataclass, field +from pathlib import Path +from typing import Callable, Sequence + +# Reuse the release-authority validators rather than duplicating the same-SHA / +# readiness contract. qa/ has no __init__.py, so make the sibling import resolve +# regardless of the caller's cwd (CI runs pytest from servers/engine). +_HERE = Path(__file__).resolve().parent +_ROOT = _HERE.parent +if str(_ROOT) not in sys.path: + sys.path.insert(0, str(_ROOT)) + +from qa.release_readiness import ( # noqa: E402 + build_sha_matches, + read_json_with_error, + validate_support_preflight_json, +) + +CANONICAL_PERSONAS = ["newbie", "veteran", "adversarial", "narrative", "optimizer"] +DEFAULT_BUDGET = "12.00" +DEFAULT_PORT = 8785 + + +RunResult = dict +# A runner takes an argv sequence and returns {ok, exit_code, stdout, stderr}. +Runner = Callable[..., RunResult] + + +def _noop_runner(argv: Sequence[str], **kwargs) -> RunResult: + """The default runner. It NEVER runs anything — calling it is a programming + error (run() in plan mode must not invoke a runner, and execute mode must be + given a real runner explicitly). This guarantees `import` and accidental + default use SSH nowhere.""" + raise RuntimeError( + "orchestrate_split_rri: no runner supplied for a remote step. " + "Remote execution is operator-gated; pass an explicit runner (or use --plan)." + ) + + +@dataclass +class OrchestratorConfig: + repo: Path + build_sha: str + handoff_json: Path + support_preflight_json: Path + ssh_host: str + ssh_key: str + remote_repo: str + vm_run_root: str = "/root/worldos-qa/runs" + local_fetch_dir: Path = field(default_factory=lambda: Path("qa/ui_playtest_runs/split-vm-fetch")) + rri_out: Path = field(default_factory=lambda: Path("qa/RRI.json")) + personas: list[str] = field(default_factory=lambda: list(CANONICAL_PERSONAS)) + budget: str = DEFAULT_BUDGET + port: int = DEFAULT_PORT + provider: str = "codex" + player_agent: str = "codex" + + def run_prefix(self) -> str: + short = (self.build_sha or "SHA").strip()[:7] or "SHA" + return f"gate-{short}" + + +def q(value: object) -> str: + return shlex.quote(str(value)) + + +# ── precondition validation ────────────────────────────────────────────────────── +def validate_preconditions(config: OrchestratorConfig) -> list[dict]: + """Return refusal gaps. EMPTY == safe to assemble/execute the split rollup. + + Reuses release_readiness' same-SHA validators so the orchestrator and the RRI + rollup agree on what 'same-SHA, ready, not-blocked' means — no second source of + truth. A non-empty result means the orchestrator must REFUSE to execute. + """ + gaps: list[dict] = [] + + if not config.build_sha.strip(): + gaps.append({"gate": "orchestrate", "missing": "--build-sha", "detail": "a build SHA is required for a same-SHA split rollup"}) + + # Support preflight is MANDATORY for the split lane (VM persona evidence rides + # the Mac --handoff-json for native proof, which release_readiness only honors + # when a same-SHA support preflight is present and green). + if not config.support_preflight_json or str(config.support_preflight_json) in ("", ".") or not Path(config.support_preflight_json).is_file(): + where = str(config.support_preflight_json) if str(config.support_preflight_json) not in ("", ".") else "(not supplied)" + gaps.append({ + "gate": "support_preflight", + "missing": "--support-preflight-json", + "detail": f"support preflight artifact not found: {where}", + }) + else: + _proof, preflight_gaps = validate_support_preflight_json(str(config.support_preflight_json), config.build_sha) + gaps.extend(preflight_gaps) + + # The Mac handoff is what supplies the native #356 / built-app proof for the VM + # persona lane. The orchestrator's job is SAME-SHA COORDINATION, not re-deriving + # the release verdict — the DEEP per-gate handoff evidence check is performed by + # release_readiness.py during the rollup itself (validate_handoff_json), so we do + # not duplicate it here. We refuse upfront only on the cheap, meaningful coordination + # signals: presence, status=passed, not-dirty, and same-SHA (a mismatched-SHA or + # failed/dirty handoff is exactly the corruption the split rollup must not paper over). + gaps.extend(_handoff_coordination_gaps(config)) + + return gaps + + +def _handoff_coordination_gaps(config: OrchestratorConfig) -> list[dict]: + gaps: list[dict] = [] + if not config.handoff_json or str(config.handoff_json) in ("", ".") or not Path(config.handoff_json).is_file(): + where = str(config.handoff_json) if str(config.handoff_json) not in ("", ".") else "(not supplied)" + gaps.append({ + "gate": "native_gate", + "missing": "--handoff-json", + "detail": f"Mac handoff artifact not found: {where}", + }) + return gaps + payload, error = read_json_with_error(Path(config.handoff_json)) + if error: + gaps.append({"gate": "native_gate", "missing": str(config.handoff_json), "detail": f"handoff JSON {error}"}) + return gaps + if payload.get("schema") != "worldos.app-handoff.v1": + gaps.append({"gate": "native_gate", "missing": str(config.handoff_json), "detail": "handoff schema is missing or wrong"}) + if payload.get("status") != "passed": + gaps.append({"gate": "native_gate", "missing": str(config.handoff_json), "detail": f"handoff status is {payload.get('status') or 'missing'}"}) + if payload.get("dirty") is not False: + gaps.append({"gate": "native_gate", "missing": str(config.handoff_json), "detail": "handoff evidence was recorded from a dirty worktree"}) + commit_sha = str(payload.get("commit_sha") or "") + if config.build_sha and not build_sha_matches(commit_sha, config.build_sha): + gaps.append({ + "gate": "native_gate", + "missing": str(config.handoff_json), + "detail": f"handoff commit_sha {commit_sha or 'missing'} does not match --build-sha {config.build_sha}", + }) + return gaps + + +# ── command assembly (the "what it WOULD run" surface) ───────────────────────────── +def remote_persona_sweep_commands(config: OrchestratorConfig) -> list[str]: + """The part-B persona sweep commands as they would run ON the VM (cwd = remote + repo). Mirrors the env contract qa/support_vm_preflight.build_vm_persona_commands + emits, so the two stay in lockstep.""" + commands: list[str] = [] + prefix = config.run_prefix() + for persona in config.personas: + commands.append( + " ".join( + [ + f"WORLDOS_ART_REPO_ROOT={q(config.remote_repo)}", + "WOS_APP_PART=B", + "WOS_APP_SKIP_BUILD=1", + "WOS_APP_NO_GLOBAL_KILL=1", + f"WOS_APP_SELECTED_PROVIDER={q(config.provider)}", + f"WOS_APP_PLAYER_AGENT={q(config.player_agent)}", + f"WOS_APP_PREFERRED_PORT={q(config.port)}", + "qa/ui_playtest_app.sh", + q(f"{prefix}-{persona}"), + "baldurs-gate", + q(persona), + "40", + q(config.budget), + ] + ) + ) + return commands + + +def ssh_base(config: OrchestratorConfig) -> list[str]: + base = ["ssh"] + if config.ssh_key: + base += ["-i", config.ssh_key] + base += [config.ssh_host] + return base + + +def preflight_command(config: OrchestratorConfig) -> str: + """The same-SHA readiness gate run on the VM before any persona sweep.""" + remote = " ".join( + [ + f"cd {q(config.remote_repo)} &&", + "python3 qa/support_vm_preflight.py", + "--repo", q(config.remote_repo), + "--expected-sha", q(config.build_sha), + "--provider", q(config.provider), + "--player-agent", q(config.player_agent), + "--art-root", q(config.remote_repo), + "--private-art-mode", "required", + ] + ) + return " ".join(ssh_base(config) + [q(remote)]) + + +def persona_sweep_command(config: OrchestratorConfig) -> str: + """One SSH invocation that runs the part-B persona sweep on the VM.""" + inner = f"cd {q(config.remote_repo)} && " + " && ".join(remote_persona_sweep_commands(config)) + return " ".join(ssh_base(config) + [q(inner)]) + + +def local_run_dirs(config: OrchestratorConfig) -> list[Path]: + """The LOCAL per-persona run dirs the VM artifacts are fetched into.""" + prefix = config.run_prefix() + return [Path(config.local_fetch_dir) / f"{prefix}-{persona}" for persona in config.personas] + + +def fetch_command(config: OrchestratorConfig) -> str: + """scp the per-persona VM run dirs back to the local fetch dir.""" + prefix = config.run_prefix() + remote_glob = f"{config.vm_run_root}/{prefix}-*" + parts = ["scp", "-r"] + if config.ssh_key: + parts += ["-i", config.ssh_key] + parts += [f"{config.ssh_host}:{remote_glob}", str(config.local_fetch_dir)] + return " ".join(q(p) if (" " in str(p)) else str(p) for p in parts) + + +def build_rollup_args(config: OrchestratorConfig) -> list[str]: + """Assemble the release_readiness.py arg list for the same-SHA RRI rollup. + + This is the heart of the tool: it threads the Mac --handoff-json, the + --support-preflight-json, the --build-sha and the fetched VM persona run dirs + into the ONE rollup invocation. The args are intentionally a list so callers can + assert on exact pairs (no string-parsing fragility).""" + run_dirs = local_run_dirs(config) + args = [ + "--runs", ",".join(str(p) for p in run_dirs), + "--expected-personas", ",".join(config.personas), + "--handoff-json", str(config.handoff_json), + "--support-preflight-json", str(config.support_preflight_json), + "--build-sha", config.build_sha, + "--out", str(config.rri_out), + "--scorecard-row", + ] + return args + + +def build_rollup_command(config: OrchestratorConfig) -> str: + """The rollup as a single LOCAL command string (runs on the Mac, not the VM).""" + parts = ["python3", "qa/release_readiness.py", *build_rollup_args(config)] + return " ".join(q(p) for p in parts) + + +def build_plan(config: OrchestratorConfig) -> dict: + """A machine-readable plan: the ordered command sequence the tool would run, + plus the precondition gaps that would refuse it. PURE — runs nothing.""" + steps = [ + { + "kind": "preflight", + "where": "vm", + "description": "Same-SHA support-VM readiness gate (refuses on stale build / unproven auth).", + "command": preflight_command(config), + }, + { + "kind": "persona_sweep", + "where": "vm", + "description": "Part-B five-persona real-browser sweep on the VM (operator-approved remote step).", + "command": persona_sweep_command(config), + "per_persona": remote_persona_sweep_commands(config), + }, + { + "kind": "fetch", + "where": "local", + "description": "Copy the per-persona VM run dirs back to the local fetch dir.", + "command": fetch_command(config), + }, + { + "kind": "rollup", + "where": "local", + "description": "Same-SHA RRI rollup (--handoff-json + --support-preflight-json + --build-sha).", + "command": build_rollup_command(config), + }, + ] + gaps = validate_preconditions(config) + return { + "schema": "worldos.orchestrate-split-rri.plan.v1", + "build_sha": config.build_sha, + "ssh_host": config.ssh_host, + "remote_repo": config.remote_repo, + "personas": config.personas, + "local_fetch_dir": str(config.local_fetch_dir), + "rri_out": str(config.rri_out), + "rollup_args": build_rollup_args(config), + "steps": steps, + "gaps": gaps, + "ready": not gaps, + "safety": ( + "Default mode is a DRY RUN: nothing remote runs without --execute. The remote " + "persona sweep is operator-approved and the SSH command is shown before it runs." + ), + } + + +def render_plan_text(plan: dict) -> str: + lines = [ + "# WorldOS split-lane RRI orchestration plan (DRY RUN)", + "", + f"- Build SHA: {plan['build_sha'] or '(missing)'}", + f"- SSH host: {plan['ssh_host'] or '(missing)'}", + f"- Remote repo: {plan['remote_repo'] or '(missing)'}", + f"- Personas: {','.join(plan['personas'])}", + f"- Local fetch dir: {plan['local_fetch_dir']}", + f"- RRI out: {plan['rri_out']}", + f"- Ready: {str(plan['ready']).lower()}", + "", + plan["safety"], + "", + "## Command sequence (this is what would run)", + "", + ] + for idx, step in enumerate(plan["steps"], start=1): + lines.append(f"### {idx}. {step['kind']} ({step['where']})") + lines.append(f"# {step['description']}") + lines.append(step["command"]) + lines.append("") + if plan["gaps"]: + lines.append("## REFUSAL — preconditions not met (would NOT execute)") + lines.append("") + for gap in plan["gaps"]: + lines.append(f"- [{gap.get('gate')}] {gap.get('missing')}: {gap.get('detail')}") + lines.append("") + return "\n".join(lines) + + +def run(config: OrchestratorConfig, *, execute: bool = False, runner: Runner | None = None) -> dict: + """Build the plan and, only when execute=True AND preconditions hold, run the + operator-approved REMOTE step via the injected runner. + + In plan mode (default) NOTHING is run and `executed` is False. Even in execute + mode, the SSH command string is captured in `shown_command` (the caller prints it + first) and a missing/blocked preflight or SHA mismatch REFUSES execution. There is + no path to a live SSH without an explicitly injected runner.""" + plan = build_plan(config) + result = { + "mode": "execute" if execute else "plan", + "executed": False, + "ready": plan["ready"], + "gaps": plan["gaps"], + "plan": plan, + "shown_command": "", + } + + if not execute: + return result + + # --execute: refuse on any precondition gap (the whole point of the gate). + if plan["gaps"]: + result["refused"] = True + return result + + runner = runner or _noop_runner + # Show the SSH command FIRST (operator can read exactly what runs), then run only + # the remote persona sweep through the injected runner. Fetch + rollup remain the + # operator's local steps (the plan prints them) so this tool never silently writes + # a committed RRI artifact. + sweep_step = next(step for step in plan["steps"] if step["kind"] == "persona_sweep") + result["shown_command"] = sweep_step["command"] + argv = ssh_base(config) + [ + "cd " + q(config.remote_repo) + " && " + " && ".join(remote_persona_sweep_commands(config)) + ] + run_result = runner(argv) + result["executed"] = True + result["remote_result"] = run_result + return result + + +def parse_args(argv: Sequence[str]) -> argparse.Namespace: + parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) + parser.add_argument("--repo", default=".", help="Local WorldOS checkout (for the rollup; default cwd)") + parser.add_argument("--build-sha", default="", help="SHA both lanes must match (REQUIRED for a real run)") + parser.add_argument("--handoff-json", default="", help="Mac app handoff JSON (qa/app_handoff_gate.py) for native proof") + parser.add_argument("--support-preflight-json", default="", help="Support-VM preflight JSON (qa/support_vm_preflight.py)") + parser.add_argument("--ssh-host", default="", help="Operator-supplied VM ssh target, e.g. root@host (NOT hardcoded)") + parser.add_argument("--ssh-key", default="", help="Operator-supplied SSH key path") + parser.add_argument("--remote-repo", default="", help="Remote WorldOS checkout path on the VM") + parser.add_argument("--vm-run-root", default="/root/worldos-qa/runs", help="Remote dir holding per-persona run dirs") + parser.add_argument("--local-fetch-dir", default="qa/ui_playtest_runs/split-vm-fetch", help="Local dir to fetch VM run dirs into") + parser.add_argument("--rri-out", default="qa/RRI.json", help="RRI rollup output path (operator-chosen)") + parser.add_argument("--personas", default=",".join(CANONICAL_PERSONAS), help="Comma-separated persona list") + parser.add_argument("--budget", default=DEFAULT_BUDGET) + parser.add_argument("--port", type=int, default=DEFAULT_PORT) + parser.add_argument("--provider", choices=("codex", "claude"), default="codex") + parser.add_argument("--player-agent", choices=("codex", "claude"), default="codex") + mode = parser.add_mutually_exclusive_group() + mode.add_argument("--plan", action="store_true", help="DRY RUN (default): print the command sequence, run nothing remote") + mode.add_argument("--execute", action="store_true", help="Operator-approved: run the remote persona sweep (SSH shown first)") + parser.add_argument("--json", action="store_true", help="Emit the machine-readable plan/result as JSON") + return parser.parse_args(list(argv)) + + +def config_from_args(args: argparse.Namespace) -> OrchestratorConfig: + personas = [p.strip() for p in args.personas.split(",") if p.strip()] or list(CANONICAL_PERSONAS) + return OrchestratorConfig( + repo=Path(args.repo).expanduser(), + build_sha=args.build_sha.strip(), + handoff_json=Path(args.handoff_json).expanduser() if args.handoff_json else Path(""), + support_preflight_json=Path(args.support_preflight_json).expanduser() if args.support_preflight_json else Path(""), + ssh_host=args.ssh_host.strip(), + ssh_key=args.ssh_key.strip(), + remote_repo=args.remote_repo.strip(), + vm_run_root=args.vm_run_root.strip(), + local_fetch_dir=Path(args.local_fetch_dir).expanduser(), + rri_out=Path(args.rri_out).expanduser(), + personas=personas, + budget=args.budget, + port=args.port, + provider=args.provider, + player_agent=args.player_agent, + ) + + +def main(argv: Sequence[str] | None = None, *, runner: Runner | None = None) -> int: + args = parse_args(argv if argv is not None else sys.argv[1:]) + config = config_from_args(args) + execute = bool(args.execute) + + result = run(config, execute=execute, runner=runner) + + if args.json: + print(json.dumps(result, indent=2, sort_keys=True)) + else: + print(render_plan_text(result["plan"])) + if execute: + if result["executed"]: + print("\n# EXECUTED the remote persona sweep (SSH shown above). " + "Now run the fetch + rollup steps locally when the sweep completes.") + else: + print("\n# REFUSED to execute: preconditions not met (see REFUSAL above).") + + # Exit non-zero when a real run was requested but refused, OR when a plan is not + # ready (so CI/operators see the red). A clean plan exits 0 (it ran nothing). + if execute and not result["executed"]: + return 1 + if not result["ready"] and execute: + return 1 + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/qa/release_readiness.py b/qa/release_readiness.py index 606d9ead..8dcf92ed 100644 --- a/qa/release_readiness.py +++ b/qa/release_readiness.py @@ -51,6 +51,18 @@ native gate PASS · arc completed · cross-persona satisfaction >=7 & no give-up · 0 critical bugs · story >=4.3 · mech >=4.5 · behavioral GREEN · ui-audit PASS · image-render >=95% · palette-live true + +ADDITIVE Phase-3 capabilities (every one is opt-in; absent inputs == today's output): + - LATENCY GATES (latency_s_per_beat / latency_coldopen): per-beat GENERATION budget + from qa/latency_baseline.json, sourced from the SAME on-disk artifacts already read + (a run's latency.json sidecar / a latency block in run.json / score.json). They gate + ONLY when latency evidence is PRESENT and over budget; when latency is ABSENT the gate + is a documented EVIDENCE-GAP SKIP (excluded from passed/total), never a new false fail — + so every pre-existing RRI result is byte-identical. + - --deterministic-only: evaluate ONLY the gates needing no live LLM/persona evidence + (native_gate, ui_audit, image_render, palette_live, + latency when present) and mark the + LLM/persona gates SKIPPED (not FAILED) — an early advisory "do the deterministic release + gates hold?" signal. NEVER claims a release verdict (release_ready stays false). """ from __future__ import annotations @@ -59,10 +71,42 @@ import re import sys from pathlib import Path +from typing import Optional REQUIRED_RELEASE_PERSONAS = ["newbie", "veteran", "adversarial", "narrative", "optimizer"] RELEASE_VERDICT_GATE = "full_five_persona_rri" + +# --- ADDITIVE Phase-3 signal taxonomy ----------------------------------------- +# The gates that depend on LIVE LLM / persona evidence (a real `claude -p` DM beat, +# a persona's self-reported satisfaction, a model-scored story/mech lens). In +# --deterministic-only mode these are SKIPPED (not failed) so CI / the agent get an +# early "do the DETERMINISTIC release gates hold?" signal with no live model run. +LLM_PERSONA_GATES = ( + "arc_completed", + "cross_persona_sat", + "no_give_up", + "zero_critical", + "story_craft", + "mechanical", + "behavioral", +) +# The deterministic complement (need no live LLM/persona evidence). The two latency +# gates are deterministic measurements and join this set ONLY when latency evidence +# is present (otherwise they are an evidence-gap skip, never a deterministic fail). +DETERMINISTIC_GATES = ( + "native_gate", + "ui_audit", + "image_render", + "palette_live", +) +LATENCY_GATES = ("latency_s_per_beat", "latency_coldopen") + +# Default per-beat latency budget — overridden by qa/latency_baseline.json when present. +# Healthy ledger figures (qa/scores_ledger.md) are ~78 s/beat and ~157 cold-open; these +# defaults add headroom so routine scorer/host variance never trips the gate. +DEFAULT_LATENCY_BASELINE = {"s_per_beat_budget": 120.0, "coldopen_s_budget": 240.0} +LATENCY_BASELINE_PATH = Path(__file__).resolve().parent / "latency_baseline.json" REQUIRED_HANDOFF_GATES = ["web_scripted_smoke", "built_app_scripted_smoke", "built_app_codex_playtest"] REQUIRED_HANDOFF_EVIDENCE_KINDS = [ "screenshots", @@ -203,6 +247,55 @@ def image_render_rate(run: Path, score: dict) -> tuple[float, int, int, str, str ) +def _latency_float(value) -> Optional[float]: + """Coerce a latency column to a positive float, or None when it is absent/NULL. + + latency_rollup.py writes NULL (None) for s_per_beat/coldopen_s when a run has no + derivable beat — that is ABSENT evidence (skip the gate), never a fabricated 0.0 + that would silently pass. Booleans and non-numerics are also treated as absent.""" + if isinstance(value, bool) or not isinstance(value, (int, float)): + return None + f = float(value) + if f != f: # NaN guard + return None + return f + + +def read_latency(run: Path, run_json: dict, score: dict) -> tuple[Optional[float], Optional[float], str]: + """Read (s_per_beat, coldopen_s, source) from the same on-disk artifacts the rollup + already reads — a run's ``latency.json`` sidecar first (what qa/run_duo.sh writes via + qa/latency_rollup.py --out), then a ``latency`` block inside run.json, then top-level + latency fields on run.json / score.json. ABSENT everywhere -> (None, None, "none"), + which makes the latency gates a documented EVIDENCE-GAP/skip, never a new false fail.""" + sidecar = read_json(run / "latency.json") + candidates: list[tuple[dict, str]] = [ + (sidecar, str(run / "latency.json")), + (run_json.get("latency") if isinstance(run_json.get("latency"), dict) else {}, str(run / "run.json")), + (run_json, str(run / "run.json")), + (score, str(run / "score.json")), + ] + for blob, src in candidates: + if not isinstance(blob, dict): + continue + s_per_beat = _latency_float(blob.get("s_per_beat")) + coldopen_s = _latency_float(blob.get("coldopen_s")) + if s_per_beat is not None or coldopen_s is not None: + return s_per_beat, coldopen_s, src + return None, None, "none" + + +def load_latency_baseline(path: Path = LATENCY_BASELINE_PATH) -> dict: + """Per-beat latency BUDGET. Falls back to DEFAULT_LATENCY_BASELINE when the baseline + file is absent or malformed (additive: a missing baseline never changes behavior).""" + payload = read_json(path) + out = dict(DEFAULT_LATENCY_BASELINE) + for key in ("s_per_beat_budget", "coldopen_s_budget"): + val = _latency_float(payload.get(key)) + if val is not None: + out[key] = val + return out + + def split_csv(value: str) -> list[str]: return [p.strip() for p in value.split(",") if p.strip()] @@ -634,6 +727,11 @@ def main() -> int: "exists the rollup is forced to an ABORTED status (infra abort, not a product RRI)") ap.add_argument("--out", default="qa/RRI.json") ap.add_argument("--scorecard-row", action="store_true") + ap.add_argument("--deterministic-only", dest="deterministic_only", action="store_true", + help="evaluate ONLY the gates that need no live LLM/persona evidence " + "(native_gate, ui_audit, image_render, palette_live, + latency when present) " + "and mark the LLM/persona gates SKIPPED (not FAILED) — an early advisory " + "'do the deterministic release gates hold?' signal, never the release verdict") args = ap.parse_args() run_dirs = [Path(p) for p in split_csv(args.runs)] @@ -683,6 +781,7 @@ def main() -> int: }) continue rate, ok, total, image_source, image_gap = image_render_rate(rd, sc) + lat_s_per_beat, lat_coldopen, lat_source = read_latency(rd, rj, sc) persona = sc.get("persona") or expected_for_run or infer_persona(rd) if persona: completed_personas.append(str(persona)) @@ -707,6 +806,11 @@ def main() -> int: "image_evidence_gap": image_gap, "image_404s": int(sc.get("image_404s", 0) or 0), "image_404s_unexpected": unexpected_404s, + # ADDITIVE latency evidence (Phase-3): None when this run carries no derivable + # latency (a documented evidence-gap/skip for the latency gates, never a fail). + "s_per_beat": lat_s_per_beat, + "coldopen_s": lat_coldopen, + "latency_source": lat_source, "run_build_sha": rj.get("build_sha") or "", "part_b_result": (rj.get("part_b") or {}).get("persona_loop") or "n/a", "part_b_score_pass": bool((rj.get("part_b") or {}).get("score_pass")), @@ -741,6 +845,17 @@ def main() -> int: img_rate = (sum(p["image_ok"] for p in img_runs) / sum(p["image_total"] for p in img_runs)) if img_runs else 0.0 total_image_denominator = sum(p["image_total"] for p in img_runs) + # ADDITIVE latency gates (Phase-3): the WORST (max) per-beat figure across personas that + # actually recorded latency, judged against qa/latency_baseline.json. The aggregate is the + # max so a single slow persona cannot be hidden by faster ones (the worldos-latency-forensics + # discipline: a slow beat that trips a persona is a release blocker). When NO persona recorded + # latency, the aggregate is None -> the gate is a documented evidence-gap/skip, never a new fail. + latency_budget = load_latency_baseline() + s_per_beat_values = [p["s_per_beat"] for p in persona_scores if p.get("s_per_beat") is not None] + coldopen_values = [p["coldopen_s"] for p in persona_scores if p.get("coldopen_s") is not None] + agg_s_per_beat = max(s_per_beat_values) if s_per_beat_values else None + agg_coldopen_s = max(coldopen_values) if coldopen_values else None + # image_render source selection. The VM cannot serve gitignored _private art and runs # a null image provider, so every VM /image request 404s BY CONSTRUCTION — those are # DESIGNED no-art outcomes, not render failures (audit F11-1: a real sweep always @@ -950,7 +1065,25 @@ def main() -> int: image_render_ok = image_evidence_complete and img_rate >= 0.95 and "image_render" not in evidence_gap_gates image_render_detail = f"source={image_render_source}; rate={img_rate:.2%}; denominator={total_image_denominator}" - # ---- the 11 gates (each contributes to RRI; all must hold for 10/10) ---- + # ---- ADDITIVE latency gates (Phase-3) ---- + # s_per_beat / coldopen are HARD gates ONLY when latency evidence is present AND over + # budget; ABSENT latency -> the gate is SKIPPED with a documented evidence gap, never a + # new false fail. A skipped gate is excluded from passed/total so an evidence-less run's + # RRI + release_ready are byte-identical to today (every pre-existing result is unchanged). + s_per_beat_budget = latency_budget["s_per_beat_budget"] + coldopen_s_budget = latency_budget["coldopen_s_budget"] + latency_s_per_beat_ok = agg_s_per_beat is None or agg_s_per_beat <= s_per_beat_budget + latency_coldopen_ok = agg_coldopen_s is None or agg_coldopen_s <= coldopen_s_budget + latency_s_per_beat_detail = ( + f"s_per_beat={agg_s_per_beat if agg_s_per_beat is not None else 'n/a (evidence gap)'}; " + f"budget={s_per_beat_budget}" + ) + latency_coldopen_detail = ( + f"coldopen_s={agg_coldopen_s if agg_coldopen_s is not None else 'n/a (evidence gap)'}; " + f"budget={coldopen_s_budget}" + ) + + # ---- the gate set (each evaluated gate contributes to RRI; all must hold for 10/10) ---- gates = { "native_gate": (native == "PASS" and not (evidence_gap_gates & native_evidence_gap_gates), native_gate_detail), @@ -973,19 +1106,52 @@ def main() -> int: "image_render": (image_render_ok, image_render_detail), "palette_live": (args.palette_live == "true" and "palette_live" not in evidence_gap_gates, f"palette_live={args.palette_live or 'n/a'}"), + "latency_s_per_beat": (latency_s_per_beat_ok, latency_s_per_beat_detail), + "latency_coldopen": (latency_coldopen_ok, latency_coldopen_detail), } - passed = sum(1 for ok, _ in gates.values() if ok) - total_gates = len(gates) - # RRI: each gate worth 10/total; HARD FLOOR — a missed gate can't be hidden by others. - # (Equal weight keeps it honest: "10/10" literally means every gate held.) - rri = round(10.0 * passed / total_gates, 1) - failed = [name for name, (ok, _) in gates.items() if not ok] + # SKIPPED gates are excluded from passed / total_gates (never counted as pass OR fail): + # * a latency gate with NO evidence -> evidence-gap skip (additive invariant). + # * EVERY LLM/persona gate in --deterministic-only -> SKIPPED, not FAILED (early signal). + skipped_gates: list[str] = [] + if agg_s_per_beat is None: + skipped_gates.append("latency_s_per_beat") + if agg_coldopen_s is None: + skipped_gates.append("latency_coldopen") + if args.deterministic_only: + skipped_gates.extend(g for g in LLM_PERSONA_GATES if g not in skipped_gates) + skipped_set = set(skipped_gates) + + evaluated = {name: ok for name, (ok, _) in gates.items() if name not in skipped_set} + passed = sum(1 for ok in evaluated.values() if ok) + total_gates = len(evaluated) + + # RRI: each EVALUATED gate worth 10/total; HARD FLOOR — a missed gate can't be hidden by + # others. (Equal weight keeps it honest: "10/10" literally means every evaluated gate held.) + rri = round(10.0 * passed / total_gates, 1) if total_gates else 0.0 + failed = [name for name, ok in evaluated.items() if not ok] + # Deterministic-only is an early ADVISORY signal, never the release verdict: report the + # deterministic subset's verdict separately so CI/the agent can read "do the deterministic + # gates hold?" without conflating it with the full five-persona release decision. + deterministic_gate_names = [ + name for name in (*DETERMINISTIC_GATES, *LATENCY_GATES) + if name not in skipped_set + ] + deterministic_failed_gates = [name for name in deterministic_gate_names if not evaluated.get(name, True)] + deterministic_pass = not deterministic_failed_gates if missing_release_personas and "missing_release_personas" not in failed: failed.insert(0, "missing_release_personas") if missing_personas and "missing_personas" not in failed: failed.insert(0, "missing_personas") - release_ready = passed == total_gates and not evidence_gaps and not missing_personas and not harness_failures + # --deterministic-only NEVER claims a release verdict (the LLM gates are unproven, only + # skipped). The full-mode release_ready logic is unchanged. + release_ready = ( + not args.deterministic_only + and passed == total_gates + and not evidence_gaps + and not missing_personas + and not harness_failures + ) # Distinct build SHAs across ALL persona runs (diagnostic output field; the native_gate CONTRACT is # release-persona-scoped via build_sha_evidence_gaps). Restored in main()'s scope after #723 factored @@ -1046,6 +1212,13 @@ def main() -> int: "gates_passed": passed, "gates_total": total_gates, "failed_gates": failed, + # ADDITIVE Phase-3 signal fields. Empty/false in the default (full) mode with latency + # evidence absent, so every pre-existing field above is unchanged. + "deterministic_only": bool(args.deterministic_only), + "deterministic_gates": deterministic_gate_names, + "deterministic_failed_gates": deterministic_failed_gates, + "deterministic_pass": deterministic_pass, + "skipped_gates": skipped_gates, "build_sha": args.build_sha, "artifact_sources": { "behavioral": args.behavioral_path or "argument", @@ -1084,6 +1257,16 @@ def main() -> int: ], "palette_live": args.palette_live, "run_build_shas": build_shas, + # ADDITIVE latency signals (Phase-3). None when no persona recorded latency + # (the latency gates are an evidence-gap skip, never a fabricated 0.0). + "latency_s_per_beat": agg_s_per_beat, + "latency_coldopen_s": agg_coldopen_s, + "latency_s_per_beat_budget": s_per_beat_budget, + "latency_coldopen_budget": coldopen_s_budget, + "latency_sources": sorted({ + str(p["latency_source"]) for p in persona_scores + if p.get("latency_source") and p.get("latency_source") != "none" + }), }, "gate_detail": {name: detail for name, (ok, detail) in gates.items()}, "personas": persona_scores, @@ -1097,7 +1280,14 @@ def main() -> int: print(f"QUOTA-ABORTED — claude account session limit (HTTP 429): {abort_detail}") print(f" This is an INFRA abort, NOT a product RRI. The {rri}/10 below is NOT a measurement; " f"re-run after the quota resets.") - print(f"RRI {rri}/10 ({passed}/{total_gates} gates) release_ready={release_ready}") + if args.deterministic_only: + print(f"DETERMINISTIC-ONLY RRI {rri}/10 ({passed}/{total_gates} deterministic gates) " + f"deterministic_pass={deterministic_pass} (advisory — NOT the release verdict; " + f"LLM/persona gates SKIPPED: {', '.join(g for g in skipped_gates if g not in LATENCY_GATES) or 'none'})") + else: + print(f"RRI {rri}/10 ({passed}/{total_gates} gates) release_ready={release_ready}") + if skipped_gates: + print(" SKIPPED (not failed): " + ", ".join(skipped_gates)) if failed: details = [] for f in failed: @@ -1125,6 +1315,10 @@ def main() -> int: f"RRI {passed}/{total_gates}; failed: {', '.join(failed) or 'none'} |") print(row) + if args.deterministic_only: + # Advisory exit: green when the deterministic subset holds (LLM gates are skipped, + # never failed), red when a deterministic gate misses. Never claims release_ready. + return 0 if (deterministic_pass and not aborted) else 1 return 0 if release_ready else 1 diff --git a/qa/scores_db.py b/qa/scores_db.py index 55b4c87d..766159a4 100644 --- a/qa/scores_db.py +++ b/qa/scores_db.py @@ -534,6 +534,168 @@ def _g(r: dict, k: str) -> str: return "\n".join(out) +# --------------------------------------------------------------------------- +# Phase-3 observability reader (1): trends_json — machine-readable per-field time-series +# --------------------------------------------------------------------------- +# Lets the agent ask "story trend over the last N runs?" in ONE call instead of hand-reading the +# ledger. PURE READER over the same rows fetch_rows returns; additive (no schema change). Points are +# emitted OLDEST-FIRST (chronological, the natural reading order for a trend) — the opposite of +# fetch_rows' newest-first, which is for the human table. Optional fences (surface / lens ruler) let +# the caller line up an apples-to-apples series; an unset fence == every row (today's behavior). +TREND_FIELDS_DEFAULT: tuple[str, ...] = ( + "story_overall", "mech_overall", "angrydm_overall", "rri", "s_per_beat", "coldopen_s", +) + + +def trends_json( + db_path: Path | str = DB_PATH, + *, + fields: Optional[list[str]] = None, + surface: Optional[str] = None, + lens_config_version: Optional[str] = None, + limit: Optional[int] = None, +) -> dict: + """Return a machine-readable per-field time-series so a trend can be read in one call. + + Shape:: + + { + "fields": ["story_overall", ...], # the metrics being tracked + "fence": {"surface": ..., "lens_config_version": ..., "limit": ...}, + "points": [ # OLDEST-FIRST (chronological) + {"run_id": ..., "ts": ..., "build_sha": ..., "rc_label": ..., + "": value, ...}, # one entry per requested field (None if unscored) + ... + ], + } + + ``fields`` defaults to :data:`TREND_FIELDS_DEFAULT` (story/mech/angrydm/rri/s_per_beat/coldopen_s); + an unknown field raises (a typo is caught, not silently dropped). ``surface`` and + ``lens_config_version`` are optional fences — when set, only rows matching that exact value are + included (the comparability axes that fence a quality trend). ``limit`` keeps the N MOST-RECENT + matching runs (then re-orders them oldest-first). With no fence and no limit the series is every + row (additive: empty/unset == today). READ-ONLY — never writes the db. + """ + flds = list(fields) if fields is not None else list(TREND_FIELDS_DEFAULT) + unknown = [f for f in flds if f not in COLUMNS] + if unknown: + raise ValueError(f"unknown trend field(s) {unknown}; valid: {sorted(COLUMNS)}") + + rows = fetch_rows(db_path) # newest-first + if surface is not None: + rows = [r for r in rows if r.get("surface") == surface] + if lens_config_version is not None: + rows = [r for r in rows if r.get("lens_config_version") == lens_config_version] + if limit is not None: + rows = rows[:limit] # fetch_rows is newest-first → the N most recent + rows = list(reversed(rows)) # emit oldest-first (chronological trend order) + + points: list[dict] = [] + for r in rows: + pt: dict[str, Any] = { + "run_id": r.get("run_id"), + "ts": r.get("ts"), + "build_sha": r.get("build_sha"), + "rc_label": r.get("rc_label"), + } + for f in flds: + pt[f] = r.get(f) + points.append(pt) + + return { + "fields": flds, + "fence": { + "surface": surface, + "lens_config_version": lens_config_version, + "limit": limit, + }, + "points": points, + } + + +# --------------------------------------------------------------------------- +# Phase-3 observability reader (2): reconcile — READ-ONLY ledger <-> INDEX.jsonl consistency +# --------------------------------------------------------------------------- +# Reports runs in the scores ledger missing from qa/INDEX.jsonl and vice-versa, so a drift between +# the two catalogs surfaces in one call. TOLERANT READER of INDEX.jsonl: it is JSONL with ARBITRARY +# keys (an open sibling PR #573 is reshaping run-naming), so this assumes ONLY "a per-line JSON object +# with some run-id field" — it tries several known id-key names, SKIPS (and reports) any line it can't +# parse or that carries no recognizable id, and NEVER rewrites INDEX.jsonl. Strictly read-only on both +# sides (the engine/harness owns INDEX.jsonl; the ledger is appended via add_run, never here). + +# Candidate run-id key names, in priority order. The real INDEX.jsonl uses "id" (a row also carries +# "kind":"run"); "run_id"/"run"/"run_name" are accepted defensively so a PR-#573 rename doesn't make +# every row look orphaned. The FIRST present non-empty string key wins. +_INDEX_ID_KEYS: tuple[str, ...] = ("run_id", "id", "run", "run_name") + + +def _index_run_id(obj: dict) -> Optional[str]: + """Best-effort extract a run id from one INDEX.jsonl object; None if no recognizable id.""" + for k in _INDEX_ID_KEYS: + v = obj.get(k) + if isinstance(v, str) and v.strip(): + return v + return None + + +def reconcile(db_path: Path | str = DB_PATH, index_path: Path | str = QA_DIR / "INDEX.jsonl") -> dict: + """READ-ONLY consistency check between the scores ledger and ``qa/INDEX.jsonl``. + + Returns:: + + { + "in_ledger_not_index": [run_id, ...], # scored but not catalogued in the index (sorted) + "in_index_not_ledger": [run_id, ...], # catalogued but never scored into the ledger (sorted) + "matched_count": int, # run ids present in BOTH + "ledger_count": int, # distinct run ids in scores.db + "index_count": int, # distinct run ids parsed from INDEX.jsonl + "skipped_lines": [ {"line": int, "reason": str, "raw": str}, ... ], # tolerant warnings + } + + The INDEX reader is deliberately tolerant: each line must be a JSON OBJECT carrying one of + :data:`_INDEX_ID_KEYS` (``run_id`` / ``id`` / ``run`` / ``run_name``). Blank lines, malformed + JSON, non-object JSON, and objects with no recognizable id are SKIPPED and reported in + ``skipped_lines`` (never raise). A missing index file is treated as an empty index (every ledger + row becomes a ledger-only orphan). This function NEVER writes either file. + """ + ledger_ids = {r["run_id"] for r in fetch_rows(db_path)} + + index_ids: set[str] = set() + skipped: list[dict] = [] + idx = Path(index_path) + if idx.exists(): + with idx.open("r", encoding="utf-8") as fh: + for lineno, raw in enumerate(fh, start=1): + line = raw.strip() + if not line: + continue # blank line — not an error, not a row + try: + obj = json.loads(line) + except (json.JSONDecodeError, ValueError): + skipped.append({"line": lineno, "reason": "unparseable JSON", + "raw": line[:200]}) + continue + if not isinstance(obj, dict): + skipped.append({"line": lineno, "reason": "not a JSON object", + "raw": line[:200]}) + continue + rid = _index_run_id(obj) + if rid is None: + skipped.append({"line": lineno, "reason": "no recognizable run-id key", + "raw": line[:200]}) + continue + index_ids.add(rid) + + return { + "in_ledger_not_index": sorted(ledger_ids - index_ids), + "in_index_not_ledger": sorted(index_ids - ledger_ids), + "matched_count": len(ledger_ids & index_ids), + "ledger_count": len(ledger_ids), + "index_count": len(index_ids), + "skipped_lines": skipped, + } + + # --------------------------------------------------------------------------- # CLI # --------------------------------------------------------------------------- @@ -551,6 +713,21 @@ def _build_parser() -> argparse.ArgumentParser: p.add_argument("--compare-rc-surface", action="store_true", help="with --compare: also show GUI-built-app RC rows (RRI sweeps) in their own " "blocks, fenced on the FULL scoring ruler") + # --- Phase-3 observability readers (pure readers; print JSON to stdout) --- + p.add_argument("--trends-json", action="store_true", + help="print a machine-readable per-field time-series (oldest-first) as JSON; " + "fence with --surface / --lens-config-version, cap with --trends-limit, " + "pick metrics with --trends-fields") + p.add_argument("--trends-fields", default=None, + help="with --trends-json: comma-separated metric columns (default: " + "story_overall,mech_overall,angrydm_overall,rri,s_per_beat,coldopen_s)") + p.add_argument("--trends-limit", type=int, default=None, + help="with --trends-json: keep only the N most-recent matching runs") + p.add_argument("--reconcile", action="store_true", + help="READ-ONLY consistency check between qa/INDEX.jsonl and scores.db; print " + "the orphan lists (ledger-only / index-only) as JSON") + p.add_argument("--index", default=str(QA_DIR / "INDEX.jsonl"), + help="with --reconcile: path to INDEX.jsonl (default qa/INDEX.jsonl)") p.add_argument("--run-id") for col in COLUMNS: p.add_argument(f"--{col.replace('_', '-')}", dest=col, default=None) @@ -589,7 +766,20 @@ def main(argv: Optional[list[str]] = None) -> int: if args.compare: print(compare_rc(db, rc=args.rc, include_rc_surface=args.compare_rc_surface)) - if not any([args.init, args.add, args.list, args.render, args.compare]): + if args.trends_json: + flds = ([f.strip() for f in args.trends_fields.split(",") if f.strip()] + if args.trends_fields else None) + print(json.dumps( + trends_json(db, fields=flds, surface=args.surface, + lens_config_version=args.lens_config_version, limit=args.trends_limit), + indent=2, ensure_ascii=False, + )) + + if args.reconcile: + print(json.dumps(reconcile(db, args.index), indent=2, ensure_ascii=False)) + + if not any([args.init, args.add, args.list, args.render, args.compare, + args.trends_json, args.reconcile]): _build_parser().print_help() return 0 diff --git a/qa/screenshot_baselines/README.md b/qa/screenshot_baselines/README.md new file mode 100644 index 00000000..29b91af3 --- /dev/null +++ b/qa/screenshot_baselines/README.md @@ -0,0 +1,98 @@ +# Screenshot baselines — GUI visual-regression reference set + +This directory holds the **committed reference screenshots** that +`qa/visual_regression_check.py` diffs a GUI-sweep's candidate screenshots against. +It is the only place visual baselines live; the checker is a pure reader and never +writes here (or anywhere except a caller-supplied report path). + +## Directory convention + +``` +qa/screenshot_baselines/ + v1.0.4/ # one dir PER RELEASE (the version the shots were blessed under) + newbie/ # (matches the play_player_browser_* personas) + table.png # .png (one PNG per screen/view) + combat.png + dialogue.png + map.png + optimizer/ + table.png + ... + v1.0.5/ + ... +``` + +* **Per release.** Baselines are blessed against a specific release (`vX.Y.Z`). A new + release that intentionally re-skins the UI gets a NEW `vX.Y.Z/` dir — you never + silently overwrite a prior release's reference set. Point `--baseline-dir` at the + release you are gating against. +* **`/.png`.** The persona axis mirrors the GUI sweep's personas + (`newbie`, `optimizer`, `narrative`, `adversarial`, `battlemage`). The view is a + named screen; the canonical screen-hash list comes from `qa/owshot.sh` + (`launcher, table, combat, dialogue, map, character, inventory, forge, relations, + journal, bestiary, acts, merchant, create, seed, settings`). +* **View key = relative path.** The checker keys each comparison on the PNG's path + *relative to the root you pass* (e.g. `newbie/table.png`). Baseline and candidate + dirs must use the SAME relative layout so keys line up. + +## Capturing a candidate / baseline set + +Use the existing headless capture primitive — it loads a localhost URL only and is safe +to run autonomously (it never enumerates the gated external volume): + +```bash +# one screen: +qa/owshot.sh table /tmp/sweep-shots/newbie/table.png +# ... repeat per (persona, view), keeping the same relative layout under the root. +``` + +To bless a candidate set into a release baseline, copy the vetted candidate tree into +`qa/screenshot_baselines/vX.Y.Z/` and commit it (PNGs are committed normally; this is a +reference artifact, NOT a generated data artifact like scores.db/RRI.json). + +## Running the check + +```bash +# STRICT (the only mode that should ever gate a build): +python qa/visual_regression_check.py \ + --baseline-dir qa/screenshot_baselines/v1.0.4 \ + --candidate-dir /tmp/sweep-shots \ + --mode strict --json + +# AUDIT (advisory, never gates) + an HTML report for human review: +python qa/visual_regression_check.py \ + --baseline-dir qa/screenshot_baselines/v1.0.4 \ + --candidate-dir /tmp/sweep-shots \ + --mode audit --report-html /tmp/visual-audit.html --json +``` + +Exit codes: `0` = PASS / SKIPPED / EMPTY, `2` = FLAG (strict only). An empty or absent +baseline dir is an additive-by-default no-op (PASS/EMPTY, never a flag). + +## STRICT vs AUDIT, and the noise-floor caveat — START STRICT + +* **STRICT** is stdlib-only (`hashlib` + a tiny IHDR parse). It compares the **sha256 of + the PNG bytes** and the **pixel dimensions**, and gates ONLY on a *definite* change: + a byte/dimension diff, or a **baseline view with no candidate** (a whole screen/element + vanished from the run — the highest-signal regression). A candidate-only view (a brand + new screen) is reported but NOT flagged. Strict is exact: it cannot tell a meaningful + re-skin from one-pixel antialiasing jitter, so it is only trustworthy on screens that + render **byte-for-byte deterministically** (no clock, no cursor blink, no animation, a + fixed device-scale-factor and window size — exactly what `qa/owshot.sh` pins). + +* **AUDIT** is a best-effort **perceptual** diff for human review. It tries `imagehash` + (perceptual hash, Hamming distance) for a robust metric, falls back to a PIL-only mean + per-pixel difference, and **skips cleanly with a clear message if Pillow is absent** + (we deliberately do NOT add a heavy dependency to force it to run — strict mode covers + the gate without it). AUDIT **never gates the build** (always exits 0) and can emit an + HTML/JSON report. + +* **The noise-floor caveat.** Real screenshots are noisy: font hinting, subpixel + antialiasing, GPU vs software rasterisation, cursor/caret blink, and live text (clocks, + timers) all produce tiny perceptual deltas that are NOT regressions. Before you gate on + any perceptual threshold, **characterise the noise floor first**: capture the same screen + twice with no code change, run AUDIT, and observe the distribution of `distance` values. + Only once you know the floor for a given (persona, view) should you consider promoting it + into a strict gate. **Recommendation: start strict on a small set of deterministic screens; + use AUDIT purely for review until the floor is understood. Do not wire an AUDIT threshold + into CI as a gate.** diff --git a/qa/test_deterministic_rri_gate.py b/qa/test_deterministic_rri_gate.py new file mode 100644 index 00000000..df6bd96c --- /dev/null +++ b/qa/test_deterministic_rri_gate.py @@ -0,0 +1,359 @@ +"""Phase-3 ADDITIVE coverage for release_readiness.py: + + (1) --deterministic-only mode — evaluates ONLY the gates that need no live + LLM/persona evidence and marks the LLM/persona gates SKIPPED (NOT FAILED), + so CI / the agent get an early "do the deterministic release gates hold?" + signal without minting any persona/model evidence. + (2) latency gates — s_per_beat + coldopen_s as ADDITIVE hard-gates sourced from + the same on-disk artifacts release_readiness already reads (a run's + latency.json sidecar / a latency block in run.json / score.json). They gate + ONLY when latency evidence is PRESENT and exceeds the qa/latency_baseline.json + budget; when latency data is ABSENT the gate is a documented EVIDENCE-GAP/skip, + never a new false fail (every pre-existing RRI result is unchanged). + +These are pure on-disk readers; every fixture is written under tmp_path. No +committed data artifact (qa/scores.db, qa/RRI.json, qa/INDEX.jsonl, transcripts) +is ever written — --out is always a tmp file. +""" + +import json +import subprocess +import sys +import tempfile +import unittest +from pathlib import Path + + +ROOT = Path(__file__).resolve().parents[1] +SCRIPT = ROOT / "qa" / "release_readiness.py" + +# The gates that need live LLM/persona evidence — in --deterministic-only they are +# SKIPPED, not FAILED. The complement (native_gate, ui_audit, image_render, +# palette_live, + the latency gates) are deterministic and still evaluated. +LLM_PERSONA_GATES = { + "arc_completed", + "cross_persona_sat", + "no_give_up", + "zero_critical", + "story_craft", + "mechanical", + "behavioral", +} +DETERMINISTIC_GATES = { + "native_gate", + "ui_audit", + "image_render", + "palette_live", +} +LATENCY_GATES = ("latency_s_per_beat", "latency_coldopen") + + +class DeterministicAndLatencyTests(unittest.TestCase): + def run_rri(self, tmp: Path, *args: str) -> tuple[int, str, dict]: + out = tmp / "RRI.json" + cmd = [sys.executable, str(SCRIPT), *args, "--out", str(out)] + proc = subprocess.run(cmd, cwd=ROOT, text=True, capture_output=True, check=False) + payload = json.loads(out.read_text(encoding="utf-8")) if out.exists() else {} + return proc.returncode, proc.stdout + proc.stderr, payload + + def write_release_inputs(self, tmp: Path) -> tuple[Path, Path, Path, Path, Path]: + story = tmp / "story.json" + mech = tmp / "mech.json" + behavioral = tmp / "behavioral.txt" + audit = tmp / "audit.log" + palette = tmp / "session_surface.final.json" + story.write_text(json.dumps({"overall": 5}), encoding="utf-8") + mech.write_text(json.dumps({"overall": 5}), encoding="utf-8") + behavioral.write_text("GREEN\n", encoding="utf-8") + audit.write_text("PASS\n", encoding="utf-8") + palette.write_text(json.dumps({"can_act": True}), encoding="utf-8") + return story, mech, behavioral, audit, palette + + def write_persona_run( + self, + tmp: Path, + persona: str, + *, + sha: str = "deadbee", + include_part_a: bool = True, + image_status: int = 200, + latency: dict | None = None, + latency_in_run_json: bool = False, + ) -> Path: + run = tmp / f"gate-{persona}" + player = run / "player" + player.mkdir(parents=True) + (run / "score.json").write_text( + json.dumps( + { + "run": f"gate-{persona}", + "persona": persona, + "completed_intro_flow": True, + "persona_satisfaction": 9, + "gave_up": False, + "bug_reports_critical": 0, + "console_errors": 0, + "image_404s": 0, + } + ), + encoding="utf-8", + ) + payload = {"build_sha": sha, "part_b": {"persona_loop": "PASS", "score_pass": True}} + if include_part_a: + payload["part_a"] = {"result": "PASS"} + if latency is not None and latency_in_run_json: + payload["latency"] = latency + (run / "run.json").write_text(json.dumps(payload), encoding="utf-8") + (player / "network.ndjson").write_text( + json.dumps({"url": f"http://127.0.0.1/image?scope={persona}", "status": image_status}), + encoding="utf-8", + ) + if latency is not None and not latency_in_run_json: + (run / "latency.json").write_text(json.dumps(latency), encoding="utf-8") + return run + + def _five_runs(self, tmp: Path, **kwargs) -> list[Path]: + return [ + self.write_persona_run(tmp, persona, **kwargs) + for persona in ("newbie", "veteran", "adversarial", "narrative", "optimizer") + ] + + def _common_args(self, runs, story, mech, behavioral, audit, palette) -> list[str]: + return [ + "--runs", + ",".join(str(r) for r in runs), + "--expected-personas", + "newbie,veteran,adversarial,narrative,optimizer", + "--story", + str(story), + "--mech", + str(mech), + "--behavioral", + "GREEN", + "--behavioral-path", + str(behavioral), + "--ui-audit", + "PASS", + "--ui-audit-log", + str(audit), + "--palette-live", + "true", + "--palette-source", + str(palette), + "--build-sha", + "deadbee", + ] + + # ---- (1) --deterministic-only mode ---- + + def test_deterministic_only_skips_llm_gates_and_computes_deterministic_subset(self): + with tempfile.TemporaryDirectory() as td: + tmp = Path(td) + runs = self._five_runs(tmp) + story, mech, behavioral, audit, palette = self.write_release_inputs(tmp) + + rc, _text, payload = self.run_rri( + tmp, + *self._common_args(runs, story, mech, behavioral, audit, palette), + "--deterministic-only", + ) + + # Deterministic-only is an EARLY ADVISORY signal: it is never the release + # verdict, so it must not claim release_ready and the rc reflects "advisory". + self.assertTrue(payload["deterministic_only"]) + self.assertFalse(payload["release_ready"]) + # The 4 deterministic gates are all evaluated and PASS on clean evidence. + self.assertEqual(set(payload["deterministic_gates"]), DETERMINISTIC_GATES) + for gate in DETERMINISTIC_GATES: + self.assertNotIn(gate, payload["skipped_gates"], f"{gate} must be evaluated, not skipped") + self.assertTrue(payload["deterministic_pass"]) + # Every LLM/persona gate is SKIPPED, not FAILED. (The two latency gates are also + # skipped here because this fixture carries no latency evidence — an evidence-gap + # skip, never a fail — so skipped_gates is a SUPERSET of the LLM gate set.) + self.assertTrue(LLM_PERSONA_GATES.issubset(set(payload["skipped_gates"]))) + for gate in LLM_PERSONA_GATES: + self.assertNotIn(gate, payload["failed_gates"], f"{gate} must be SKIPPED not FAILED") + self.assertEqual(payload["deterministic_failed_gates"], []) + + def test_deterministic_only_reports_a_failing_deterministic_gate(self): + # A real deterministic miss (palette-live false) must FAIL in deterministic-only, + # while the LLM gates stay SKIPPED. + with tempfile.TemporaryDirectory() as td: + tmp = Path(td) + runs = self._five_runs(tmp) + story, mech, behavioral, audit, palette = self.write_release_inputs(tmp) + + args = self._common_args(runs, story, mech, behavioral, audit, palette) + # flip palette-live to false + idx = args.index("--palette-live") + args[idx + 1] = "false" + + rc, _text, payload = self.run_rri(tmp, *args, "--deterministic-only") + + self.assertTrue(payload["deterministic_only"]) + self.assertFalse(payload["deterministic_pass"]) + self.assertIn("palette_live", payload["deterministic_failed_gates"]) + self.assertIn("palette_live", payload["failed_gates"]) + # LLM gates remain skipped, NOT failed, even on a deterministic miss. + for gate in LLM_PERSONA_GATES: + self.assertIn(gate, payload["skipped_gates"]) + self.assertNotIn(gate, payload["failed_gates"]) + self.assertNotEqual(rc, 0) + + def test_default_mode_is_byte_identical_without_deterministic_flag(self): + # ADDITIVE invariant: without --deterministic-only the output carries NO new + # mode and the full gate set is evaluated exactly as before. + with tempfile.TemporaryDirectory() as td: + tmp = Path(td) + runs = self._five_runs(tmp) + story, mech, behavioral, audit, palette = self.write_release_inputs(tmp) + + rc, _text, payload = self.run_rri( + tmp, *self._common_args(runs, story, mech, behavioral, audit, palette) + ) + + self.assertEqual(rc, 0) + self.assertFalse(payload["deterministic_only"]) + self.assertTrue(payload["release_ready"]) + # ADDITIVE invariant: with no latency evidence the two latency gates are an + # evidence-gap skip, so gates_total stays the pre-Phase-3 count of 11 (byte- + # identical RRI math) — they are skipped, never failed, never counted. + self.assertEqual(set(payload["skipped_gates"]), set(LATENCY_GATES)) + self.assertEqual(payload["gates_total"], 11) + self.assertEqual(payload["rri"], 10.0) + + # ---- (2) latency gates ---- + + def test_latency_present_and_over_budget_fails_gate(self): + with tempfile.TemporaryDirectory() as td: + tmp = Path(td) + # s_per_beat 300 (>> 120 budget) and coldopen 500 (>> 240 budget) + runs = self._five_runs( + tmp, latency={"s_per_beat": 300.0, "coldopen_s": 500.0, "turns_per_beat": 5.0} + ) + story, mech, behavioral, audit, palette = self.write_release_inputs(tmp) + + rc, _text, payload = self.run_rri( + tmp, *self._common_args(runs, story, mech, behavioral, audit, palette) + ) + + self.assertEqual(rc, 1) + self.assertFalse(payload["release_ready"]) + self.assertIn("latency_s_per_beat", payload["failed_gates"]) + self.assertIn("latency_coldopen", payload["failed_gates"]) + self.assertEqual(payload["signals"]["latency_s_per_beat"], 300.0) + self.assertEqual(payload["signals"]["latency_coldopen_s"], 500.0) + self.assertEqual(payload["signals"]["latency_s_per_beat_budget"], 120.0) + self.assertEqual(payload["signals"]["latency_coldopen_budget"], 240.0) + + def test_latency_present_and_under_budget_passes_gate(self): + with tempfile.TemporaryDirectory() as td: + tmp = Path(td) + # healthy ledger figures: 78.2 s/beat, 157 cold-open — both under budget + runs = self._five_runs( + tmp, latency={"s_per_beat": 78.2, "coldopen_s": 157.0, "turns_per_beat": 4.4} + ) + story, mech, behavioral, audit, palette = self.write_release_inputs(tmp) + + rc, _text, payload = self.run_rri( + tmp, *self._common_args(runs, story, mech, behavioral, audit, palette) + ) + + self.assertEqual(rc, 0) + self.assertTrue(payload["release_ready"]) + self.assertNotIn("latency_s_per_beat", payload["failed_gates"]) + self.assertNotIn("latency_coldopen", payload["failed_gates"]) + self.assertNotIn("latency_s_per_beat", {gap["gate"] for gap in payload["evidence_gaps"]}) + + def test_latency_absent_is_skip_not_a_new_false_fail(self): + # The ADDITIVE invariant: no latency evidence anywhere -> the latency gates are a + # documented EVIDENCE-GAP/skip, NEVER a new fail. The run is still release_ready + # exactly as it was before latency gates existed. + with tempfile.TemporaryDirectory() as td: + tmp = Path(td) + runs = self._five_runs(tmp) # no latency=... -> no latency.json, no run.json latency + story, mech, behavioral, audit, palette = self.write_release_inputs(tmp) + + rc, _text, payload = self.run_rri( + tmp, *self._common_args(runs, story, mech, behavioral, audit, palette) + ) + + self.assertEqual(rc, 0) + self.assertTrue(payload["release_ready"]) + self.assertNotIn("latency_s_per_beat", payload["failed_gates"]) + self.assertNotIn("latency_coldopen", payload["failed_gates"]) + # The gate is skipped (no evidence), recorded honestly. + self.assertIn("latency_s_per_beat", payload["skipped_gates"]) + self.assertIn("latency_coldopen", payload["skipped_gates"]) + self.assertIsNone(payload["signals"]["latency_s_per_beat"]) + self.assertIsNone(payload["signals"]["latency_coldopen_s"]) + + def test_latency_read_from_run_json_block(self): + # Latency can also ride run.json's `latency` block (not only the sidecar) — + # both are artifacts release_readiness already opens. + with tempfile.TemporaryDirectory() as td: + tmp = Path(td) + runs = self._five_runs( + tmp, + latency={"s_per_beat": 300.0, "coldopen_s": 100.0}, + latency_in_run_json=True, + ) + story, mech, behavioral, audit, palette = self.write_release_inputs(tmp) + + rc, _text, payload = self.run_rri( + tmp, *self._common_args(runs, story, mech, behavioral, audit, palette) + ) + + self.assertEqual(rc, 1) + self.assertIn("latency_s_per_beat", payload["failed_gates"]) + self.assertNotIn("latency_coldopen", payload["failed_gates"]) + self.assertEqual(payload["signals"]["latency_s_per_beat"], 300.0) + self.assertEqual(payload["signals"]["latency_coldopen_s"], 100.0) + + def test_latency_null_columns_are_treated_as_absent_not_zero(self): + # latency_rollup writes NULL columns when a run has no continuing beat. A NULL + # s_per_beat must be an evidence gap/skip, NOT a fabricated 0.0 that "passes". + with tempfile.TemporaryDirectory() as td: + tmp = Path(td) + runs = self._five_runs( + tmp, latency={"s_per_beat": None, "coldopen_s": None, "turns_per_beat": None} + ) + story, mech, behavioral, audit, palette = self.write_release_inputs(tmp) + + rc, _text, payload = self.run_rri( + tmp, *self._common_args(runs, story, mech, behavioral, audit, palette) + ) + + self.assertEqual(rc, 0) # null latency does not invent a fail + self.assertTrue(payload["release_ready"]) + self.assertIn("latency_s_per_beat", payload["skipped_gates"]) + self.assertIsNone(payload["signals"]["latency_s_per_beat"]) + + def test_deterministic_only_includes_latency_gate_when_evidence_present(self): + # Latency is a DETERMINISTIC measurement — in --deterministic-only it is + # evaluated (not skipped) when evidence is present, and a breach fails the + # deterministic subset. + with tempfile.TemporaryDirectory() as td: + tmp = Path(td) + runs = self._five_runs( + tmp, latency={"s_per_beat": 300.0, "coldopen_s": 500.0} + ) + story, mech, behavioral, audit, palette = self.write_release_inputs(tmp) + + rc, _text, payload = self.run_rri( + tmp, + *self._common_args(runs, story, mech, behavioral, audit, palette), + "--deterministic-only", + ) + + self.assertTrue(payload["deterministic_only"]) + self.assertIn("latency_s_per_beat", payload["deterministic_gates"]) + self.assertIn("latency_s_per_beat", payload["deterministic_failed_gates"]) + self.assertFalse(payload["deterministic_pass"]) + # still no LLM gate failed + for gate in LLM_PERSONA_GATES: + self.assertNotIn(gate, payload["failed_gates"]) + + +if __name__ == "__main__": + unittest.main() diff --git a/qa/test_orchestrate_split_rri.py b/qa/test_orchestrate_split_rri.py new file mode 100644 index 00000000..5b1beae1 --- /dev/null +++ b/qa/test_orchestrate_split_rri.py @@ -0,0 +1,396 @@ +"""Tests for qa/orchestrate_split_rri.py — the split VM(part-B)+Mac(part-A handoff) +RRI rollup orchestrator. + +These tests prove the HARD SAFETY contract of the tool: + - importing the module runs NOTHING remote (no SSH on import / by default), + - --plan (the default) PRINTS the exact command sequence and executes no remote step, + - the RRI rollup invocation assembles the correct release_readiness.py args from + on-disk fixtures (tmp_path), + - a missing/blocked support preflight or a SHA mismatch is REFUSED, + - no live SSH is ever performed in tests (a fake runner records would-be commands). +""" +import json +import sys +import tempfile +import unittest +from pathlib import Path + +# The qa/ tree has no __init__.py; make `import qa.X` resolve regardless of the +# pytest cwd (CI runs from servers/engine). Insert the repo root onto sys.path. +ROOT = Path(__file__).resolve().parents[1] +if str(ROOT) not in sys.path: + sys.path.insert(0, str(ROOT)) + +import qa.orchestrate_split_rri as orch # noqa: E402 + + +SHA = "da05101cafef00dba5eba11deadbeef012345678" +SHORT = SHA[:7] + + +def write_support_preflight( + tmp: Path, + *, + sha: str = SHORT, + ready: bool = True, + blocking_categories: list[str] | None = None, +) -> Path: + """A support_vm_preflight.json shaped exactly like the validator in + release_readiness.validate_support_preflight_json expects (mirrors the helper + in test_release_readiness.py so the two stay in lockstep).""" + preflight = tmp / "support_vm_preflight.json" + blocking_categories = blocking_categories or [] + preflight.write_text( + json.dumps( + { + "schema": "worldos.support-vm-preflight.v1", + "verdict": "passed" if ready else "blocked", + "ready_for_rri": ready, + "release_verdict": False, + "blockers": [] if ready else ["host memory below required capacity"], + "repo": { + "head_short": sha, + "expected_sha": sha, + "expected_sha_match": True, + "dirty": False, + "origin_main_query": {"ok": True, "head_short": sha}, + }, + "readiness": { + "safe_to_run_personas": ready, + "release_verdict": False, + "expected_sha": sha, + "repo_head_short": sha, + "same_sha_ready": ready, + "provider": "codex", + "player_agent": "codex", + "provider_auth_ready": ready, + "player_agent_auth_ready": ready, + "required_tools_ready": ready, + "persona_briefs_ready": ready, + "private_art_ready": ready, + "artifact_return_ready": ready, + "host_capacity_ready": ready, + "min_memory_gb": 24, + "mac_handoff_required": True, + "blocking_categories": blocking_categories, + }, + "rri_plan": { + "support_preflight_json": str(preflight), + "support_preflight_required_for_split_rollup": True, + "expected_personas": list(orch.CANONICAL_PERSONAS), + "rri_rollup_command_template": ( + "python3 qa/release_readiness.py --runs VM_PERSONA_RUN_DIRS_CSV " + "--handoff-json SAME_SHA_MAC_HANDOFF_JSON " + "--support-preflight-json support_vm_preflight.json " + f"--build-sha {sha}" + ), + }, + } + ), + encoding="utf-8", + ) + return preflight + + +def write_handoff(tmp: Path, *, sha: str = SHORT) -> Path: + handoff = tmp / "handoff.json" + handoff.write_text( + json.dumps( + { + "schema": "worldos.app-handoff.v1", + "status": "passed", + "handoff_score": 100, + "dirty": False, + "commit_sha": sha, + "release_verdict": False, + "gates": [], + } + ), + encoding="utf-8", + ) + return handoff + + +class RecordingRunner: + """A stand-in for the remote runner. NEVER touches the network — it only + records the argv it was asked to run and returns a canned success.""" + + def __init__(self): + self.calls: list[list[str]] = [] + + def __call__(self, argv, **kwargs): + self.calls.append(list(argv)) + return {"ok": True, "exit_code": 0, "stdout": "", "stderr": ""} + + +def make_config(tmp: Path, preflight: Path, handoff: Path, *, sha: str = SHA) -> "orch.OrchestratorConfig": + return orch.OrchestratorConfig( + repo=ROOT, + build_sha=sha, + handoff_json=handoff, + support_preflight_json=preflight, + ssh_host="root@example.invalid", + ssh_key="/dev/null", + remote_repo="/root/worldos-qa/WorldOS", + vm_run_root="/root/worldos-qa/runs", + local_fetch_dir=tmp / "fetched", + rri_out=tmp / "RRI.json", + personas=list(orch.CANONICAL_PERSONAS), + ) + + +class ImportSafetyTests(unittest.TestCase): + def test_module_exposes_no_default_ssh_endpoint(self): + # HARD SAFETY: the module must NOT hardcode an operator VM endpoint. The + # runbook keeps connection/auth details in operator-only runbooks, not the + # tracked repo. So the source must not embed the known VM IP / key path. + src = Path(orch.__file__).read_text(encoding="utf-8") + self.assertNotIn("178.104.123.213", src) + self.assertNotIn("cloud-deploy-key", src) + + def test_import_does_not_ssh(self): + # Importing the module (done at top of file) must not have run anything + # remote. The default runner is a no-op placeholder, never auto-invoked. + self.assertTrue(hasattr(orch, "build_plan")) + self.assertTrue(hasattr(orch, "OrchestratorConfig")) + + +class PlanModeTests(unittest.TestCase): + def test_plan_lists_command_sequence_without_running_anything(self): + with tempfile.TemporaryDirectory() as td: + tmp = Path(td) + cfg = make_config(tmp, write_support_preflight(tmp), write_handoff(tmp)) + runner = RecordingRunner() + plan = orch.build_plan(cfg) + result = orch.run(cfg, execute=False, runner=runner) + + # Nothing remote was run in plan mode. + self.assertEqual(runner.calls, []) + self.assertEqual(result["mode"], "plan") + self.assertFalse(result["executed"]) + + # The plan enumerates the three remote phases in order. + kinds = [step["kind"] for step in plan["steps"]] + self.assertEqual(kinds, ["preflight", "persona_sweep", "fetch", "rollup"]) + # Each remote step carries the exact command string it WOULD run. + for step in plan["steps"]: + self.assertTrue(step["command"].strip()) + + def test_plan_renders_text_with_all_commands(self): + with tempfile.TemporaryDirectory() as td: + tmp = Path(td) + cfg = make_config(tmp, write_support_preflight(tmp), write_handoff(tmp)) + text = orch.render_plan_text(orch.build_plan(cfg)) + # The SSH command is shown (operator can read exactly what runs). + self.assertIn("ssh", text) + self.assertIn(cfg.ssh_host, text) + # The persona sweep references the part-B sweep tool. + self.assertIn("ui_playtest_app.sh", text) + # The rollup line is the release_readiness invocation. + self.assertIn("release_readiness.py", text) + + def test_cli_plan_is_default_and_runs_no_remote(self): + with tempfile.TemporaryDirectory() as td: + tmp = Path(td) + preflight = write_support_preflight(tmp) + handoff = write_handoff(tmp) + runner = RecordingRunner() + argv = [ + "--build-sha", SHA, + "--handoff-json", str(handoff), + "--support-preflight-json", str(preflight), + "--ssh-host", "root@example.invalid", + "--ssh-key", "/dev/null", + "--remote-repo", "/root/worldos-qa/WorldOS", + "--rri-out", str(tmp / "RRI.json"), + ] + rc = orch.main(argv, runner=runner) + self.assertEqual(rc, 0) + # Default == plan: nothing remote executed. + self.assertEqual(runner.calls, []) + + def test_cli_json_emits_machine_readable_plan(self): + with tempfile.TemporaryDirectory() as td: + tmp = Path(td) + preflight = write_support_preflight(tmp) + handoff = write_handoff(tmp) + import io + from contextlib import redirect_stdout + + buf = io.StringIO() + argv = [ + "--build-sha", SHA, + "--handoff-json", str(handoff), + "--support-preflight-json", str(preflight), + "--ssh-host", "root@example.invalid", + "--ssh-key", "/dev/null", + "--remote-repo", "/root/worldos-qa/WorldOS", + "--rri-out", str(tmp / "RRI.json"), + "--json", + ] + with redirect_stdout(buf): + rc = orch.main(argv, runner=RecordingRunner()) + self.assertEqual(rc, 0) + payload = json.loads(buf.getvalue()) + self.assertEqual(payload["mode"], "plan") + self.assertFalse(payload["executed"]) + # The full result nests the plan (which carries the command sequence). + self.assertIn("plan", payload) + self.assertIn("steps", payload["plan"]) + self.assertEqual( + [s["kind"] for s in payload["plan"]["steps"]], + ["preflight", "persona_sweep", "fetch", "rollup"], + ) + + +class RollupArgAssemblyTests(unittest.TestCase): + def test_rollup_command_assembles_release_readiness_args(self): + with tempfile.TemporaryDirectory() as td: + tmp = Path(td) + preflight = write_support_preflight(tmp) + handoff = write_handoff(tmp) + cfg = make_config(tmp, preflight, handoff) + args = orch.build_rollup_args(cfg) + + # The rollup MUST pass the same-SHA handoff + support preflight + build-sha + # through to release_readiness.py, plus the fetched VM persona run dirs. + self.assertIn("--handoff-json", args) + self.assertEqual(args[args.index("--handoff-json") + 1], str(handoff)) + self.assertIn("--support-preflight-json", args) + self.assertEqual(args[args.index("--support-preflight-json") + 1], str(preflight)) + self.assertIn("--build-sha", args) + self.assertEqual(args[args.index("--build-sha") + 1], SHA) + self.assertIn("--expected-personas", args) + self.assertEqual( + args[args.index("--expected-personas") + 1], + ",".join(orch.CANONICAL_PERSONAS), + ) + self.assertIn("--out", args) + self.assertEqual(args[args.index("--out") + 1], str(cfg.rri_out)) + + # --runs points at the LOCAL fetched per-persona dirs (one per persona). + self.assertIn("--runs", args) + runs_csv = args[args.index("--runs") + 1] + run_dirs = runs_csv.split(",") + self.assertEqual(len(run_dirs), len(orch.CANONICAL_PERSONAS)) + for persona in orch.CANONICAL_PERSONAS: + self.assertTrue( + any(persona in rd for rd in run_dirs), + f"expected a fetched run dir for persona {persona}", + ) + + def test_rollup_command_string_invokes_release_readiness(self): + with tempfile.TemporaryDirectory() as td: + tmp = Path(td) + cfg = make_config(tmp, write_support_preflight(tmp), write_handoff(tmp)) + cmd = orch.build_rollup_command(cfg) + self.assertIn("release_readiness.py", cmd) + self.assertIn("--support-preflight-json", cmd) + self.assertIn("--handoff-json", cmd) + + +class RefusalTests(unittest.TestCase): + def test_missing_support_preflight_is_refused(self): + with tempfile.TemporaryDirectory() as td: + tmp = Path(td) + handoff = write_handoff(tmp) + missing = tmp / "does_not_exist.json" + cfg = orch.OrchestratorConfig( + repo=ROOT, + build_sha=SHA, + handoff_json=handoff, + support_preflight_json=missing, + ssh_host="root@example.invalid", + ssh_key="/dev/null", + remote_repo="/root/worldos-qa/WorldOS", + vm_run_root="/root/worldos-qa/runs", + local_fetch_dir=tmp / "fetched", + rri_out=tmp / "RRI.json", + personas=list(orch.CANONICAL_PERSONAS), + ) + gaps = orch.validate_preconditions(cfg) + self.assertTrue(gaps, "missing support preflight must produce a refusal gap") + + runner = RecordingRunner() + result = orch.run(cfg, execute=True, runner=runner) + self.assertFalse(result["executed"], "must refuse to execute with a missing preflight") + self.assertEqual(runner.calls, [], "no remote command may run when refused") + self.assertTrue(result["gaps"]) + + def test_blocked_support_preflight_is_refused(self): + with tempfile.TemporaryDirectory() as td: + tmp = Path(td) + preflight = write_support_preflight(tmp, ready=False, blocking_categories=["host_capacity"]) + cfg = make_config(tmp, preflight, write_handoff(tmp)) + gaps = orch.validate_preconditions(cfg) + self.assertTrue(gaps, "a blocked preflight must refuse the rollup") + runner = RecordingRunner() + result = orch.run(cfg, execute=True, runner=runner) + self.assertFalse(result["executed"]) + self.assertEqual(runner.calls, []) + + def test_sha_mismatch_is_refused(self): + with tempfile.TemporaryDirectory() as td: + tmp = Path(td) + # Preflight proves a DIFFERENT sha than --build-sha. + preflight = write_support_preflight(tmp, sha="0000000") + handoff = write_handoff(tmp, sha=SHORT) + cfg = make_config(tmp, preflight, handoff, sha=SHA) + gaps = orch.validate_preconditions(cfg) + self.assertTrue(gaps, "a SHA mismatch must produce a refusal gap") + runner = RecordingRunner() + result = orch.run(cfg, execute=True, runner=runner) + self.assertFalse(result["executed"]) + self.assertEqual(runner.calls, []) + + def test_handoff_sha_mismatch_is_refused(self): + with tempfile.TemporaryDirectory() as td: + tmp = Path(td) + preflight = write_support_preflight(tmp, sha=SHORT) + handoff = write_handoff(tmp, sha="9999999") + cfg = make_config(tmp, preflight, handoff, sha=SHA) + gaps = orch.validate_preconditions(cfg) + self.assertTrue(gaps, "a mismatched handoff SHA must produce a refusal gap") + runner = RecordingRunner() + result = orch.run(cfg, execute=True, runner=runner) + self.assertFalse(result["executed"]) + self.assertEqual(runner.calls, []) + + def test_clean_inputs_have_no_gaps(self): + with tempfile.TemporaryDirectory() as td: + tmp = Path(td) + preflight = write_support_preflight(tmp, sha=SHORT) + handoff = write_handoff(tmp, sha=SHORT) + cfg = make_config(tmp, preflight, handoff, sha=SHA) + self.assertEqual(orch.validate_preconditions(cfg), []) + + +class ExecuteSafetyTests(unittest.TestCase): + def test_execute_shows_ssh_command_before_running(self): + with tempfile.TemporaryDirectory() as td: + tmp = Path(td) + cfg = make_config(tmp, write_support_preflight(tmp), write_handoff(tmp)) + runner = RecordingRunner() + result = orch.run(cfg, execute=True, runner=runner) + # With clean inputs, execute proceeds and the runner sees the remote step, + # but the printed plan (shown_command) is captured first. + self.assertTrue(result["executed"]) + self.assertTrue(result.get("shown_command")) + self.assertIn("ssh", result["shown_command"]) + # The runner was actually invoked for the remote sweep step. + self.assertTrue(runner.calls, "execute mode must invoke the injected runner") + + def test_execute_only_runs_remote_step_through_injected_runner(self): + # No live SSH: the only way a remote command runs is via the injected runner. + with tempfile.TemporaryDirectory() as td: + tmp = Path(td) + cfg = make_config(tmp, write_support_preflight(tmp), write_handoff(tmp)) + runner = RecordingRunner() + orch.run(cfg, execute=True, runner=runner) + # Every recorded call is an ssh invocation to the configured host. + for call in runner.calls: + self.assertIn("ssh", call[0]) + + +if __name__ == "__main__": + unittest.main() diff --git a/qa/test_release_readiness.py b/qa/test_release_readiness.py index e0a8ed61..e1c242d5 100644 --- a/qa/test_release_readiness.py +++ b/qa/test_release_readiness.py @@ -2542,6 +2542,126 @@ def test_abort_marker_forces_aborted(self): self.assertFalse(payload["release_ready"]) self.assertIn("resets 3:50pm (UTC)", payload["abort_detail"]) + # --- ADDITIVE Phase-3: latency gates + --deterministic-only mode. Every gate/field/mode + # here is opt-in; with latency absent + no flag the result is byte-identical to today + # (the five-persona-release tests above all still assert evidence_gaps==[] / release_ready). --- + + def _five_clean_runs_with_latency(self, tmp: Path, latency: dict | None) -> list[Path]: + runs = [] + for persona in ("newbie", "veteran", "adversarial", "narrative", "optimizer"): + run = self.write_persona_run(tmp, persona) + if latency is not None: + (run / "latency.json").write_text(json.dumps(latency), encoding="utf-8") + runs.append(run) + return runs + + def test_latency_evidence_over_budget_fails_release(self): + with tempfile.TemporaryDirectory() as td: + tmp = Path(td) + runs = self._five_clean_runs_with_latency(tmp, {"s_per_beat": 300.0, "coldopen_s": 500.0}) + story, mech, behavioral, audit, palette = self.write_release_inputs(tmp) + + rc, _text, payload = self.run_rri( + tmp, + "--runs", ",".join(str(r) for r in runs), + "--expected-personas", "newbie,veteran,adversarial,narrative,optimizer", + "--story", str(story), "--mech", str(mech), + "--behavioral", "GREEN", "--behavioral-path", str(behavioral), + "--ui-audit", "PASS", "--ui-audit-log", str(audit), + "--palette-live", "true", "--palette-source", str(palette), + "--build-sha", "deadbee", + ) + + self.assertEqual(rc, 1) + self.assertFalse(payload["release_ready"]) + self.assertIn("latency_s_per_beat", payload["failed_gates"]) + self.assertIn("latency_coldopen", payload["failed_gates"]) + self.assertEqual(payload["signals"]["latency_s_per_beat"], 300.0) + self.assertEqual(payload["signals"]["latency_coldopen_s"], 500.0) + # latency gates are now EVALUATED so they join the gate total. + self.assertEqual(payload["gates_total"], 13) + + def test_latency_evidence_under_budget_keeps_release_ready(self): + with tempfile.TemporaryDirectory() as td: + tmp = Path(td) + # healthy ledger figures — both under budget + runs = self._five_clean_runs_with_latency(tmp, {"s_per_beat": 78.2, "coldopen_s": 157.0}) + story, mech, behavioral, audit, palette = self.write_release_inputs(tmp) + + rc, _text, payload = self.run_rri( + tmp, + "--runs", ",".join(str(r) for r in runs), + "--expected-personas", "newbie,veteran,adversarial,narrative,optimizer", + "--story", str(story), "--mech", str(mech), + "--behavioral", "GREEN", "--behavioral-path", str(behavioral), + "--ui-audit", "PASS", "--ui-audit-log", str(audit), + "--palette-live", "true", "--palette-source", str(palette), + "--build-sha", "deadbee", + ) + + self.assertEqual(rc, 0) + self.assertTrue(payload["release_ready"]) + self.assertEqual(payload["evidence_gaps"], []) + self.assertNotIn("latency_s_per_beat", payload["failed_gates"]) + self.assertNotIn("latency_coldopen", payload["failed_gates"]) + self.assertEqual(payload["gates_total"], 13) + + def test_latency_absent_is_skip_and_byte_identical_release(self): + # The ADDITIVE invariant proof: no latency evidence -> latency gates are skipped + # (never failed), gate total stays 11, and a clean five-persona run is STILL + # release_ready with evidence_gaps == [] exactly as before Phase-3. + with tempfile.TemporaryDirectory() as td: + tmp = Path(td) + runs = self._five_clean_runs_with_latency(tmp, None) + story, mech, behavioral, audit, palette = self.write_release_inputs(tmp) + + rc, _text, payload = self.run_rri( + tmp, + "--runs", ",".join(str(r) for r in runs), + "--expected-personas", "newbie,veteran,adversarial,narrative,optimizer", + "--story", str(story), "--mech", str(mech), + "--behavioral", "GREEN", "--behavioral-path", str(behavioral), + "--ui-audit", "PASS", "--ui-audit-log", str(audit), + "--palette-live", "true", "--palette-source", str(palette), + "--build-sha", "deadbee", + ) + + self.assertEqual(rc, 0) + self.assertTrue(payload["release_ready"]) + self.assertEqual(payload["evidence_gaps"], []) + self.assertEqual(payload["gates_total"], 11) + self.assertEqual(sorted(payload["skipped_gates"]), ["latency_coldopen", "latency_s_per_beat"]) + self.assertIsNone(payload["signals"]["latency_s_per_beat"]) + self.assertIsNone(payload["signals"]["latency_coldopen_s"]) + + def test_deterministic_only_marks_llm_gates_skipped_not_failed(self): + with tempfile.TemporaryDirectory() as td: + tmp = Path(td) + runs = self._five_clean_runs_with_latency(tmp, None) + story, mech, behavioral, audit, palette = self.write_release_inputs(tmp) + + rc, _text, payload = self.run_rri( + tmp, + "--runs", ",".join(str(r) for r in runs), + "--expected-personas", "newbie,veteran,adversarial,narrative,optimizer", + "--story", str(story), "--mech", str(mech), + "--behavioral", "GREEN", "--behavioral-path", str(behavioral), + "--ui-audit", "PASS", "--ui-audit-log", str(audit), + "--palette-live", "true", "--palette-source", str(palette), + "--build-sha", "deadbee", + "--deterministic-only", + ) + + self.assertEqual(rc, 0) # deterministic subset holds + self.assertTrue(payload["deterministic_only"]) + self.assertFalse(payload["release_ready"]) # never the release verdict + self.assertTrue(payload["deterministic_pass"]) + for gate in ("arc_completed", "cross_persona_sat", "no_give_up", "zero_critical", + "story_craft", "mechanical", "behavioral"): + self.assertIn(gate, payload["skipped_gates"]) + self.assertNotIn(gate, payload["failed_gates"]) + self.assertEqual(payload["deterministic_failed_gates"], []) + if __name__ == "__main__": unittest.main() diff --git a/qa/test_scores_db.py b/qa/test_scores_db.py index fe3ced9a..b6997148 100644 --- a/qa/test_scores_db.py +++ b/qa/test_scores_db.py @@ -348,3 +348,221 @@ def test_seed_flags_the_decisive_rows(tmp_path): nar = by_id["str2-narrative"] assert nar["surface"] == "GUI-headless-proxy" and nar["cross_persona_sat"] == 9 assert "latency" in nar["notes"].lower() + + +# --------------------------------------------------------------------------- +# Phase-3 observability reader (1): trends_json — per-field time-series +# --------------------------------------------------------------------------- + +def test_trends_json_shape_and_default_fields(tmp_path): + """Default call returns the documented per-field time-series shape over the spec fields.""" + db = tmp_path / "t.db" + scores_db.add_run("a", db_path=db, surface="engine-duo", ts="2026-05-01T00:00:00+00:00", + build_sha="aaa", story_overall=3.9, mech_overall=3.5) + scores_db.add_run("b", db_path=db, surface="engine-duo", ts="2026-05-02T00:00:00+00:00", + build_sha="bbb", story_overall=4.2, mech_overall=3.8, rri=6.0) + out = scores_db.trends_json(db) + # top-level shape + assert set(out) >= {"fields", "fence", "points"} + # the spec's default fields are all present + assert set(out["fields"]) == { + "story_overall", "mech_overall", "angrydm_overall", "rri", "s_per_beat", "coldopen_s", + } + # one point per run, each carries identity + ts + the field values + assert len(out["points"]) == 2 + pt = out["points"][0] + assert {"run_id", "ts"} <= set(pt) + # every requested field key is present on every point (NULL/None when unscored) + for p in out["points"]: + for f in out["fields"]: + assert f in p + + +def test_trends_json_is_chronological_oldest_first(tmp_path): + """A trend reads left-to-right in time: points are ordered oldest-first (opposite of fetch_rows).""" + db = tmp_path / "t.db" + scores_db.add_run("new", db_path=db, surface="engine-duo", ts="2026-05-30T00:00:00+00:00") + scores_db.add_run("old", db_path=db, surface="engine-duo", ts="2026-05-01T00:00:00+00:00") + scores_db.add_run("mid", db_path=db, surface="engine-duo", ts="2026-05-15T00:00:00+00:00") + out = scores_db.trends_json(db) + assert [p["run_id"] for p in out["points"]] == ["old", "mid", "new"] + + +def test_trends_json_values_carry_through(tmp_path): + db = tmp_path / "t.db" + scores_db.add_run("a", db_path=db, surface="engine-duo", ts="2026-05-01T00:00:00+00:00", + story_overall=4.0, s_per_beat=80.5, coldopen_s=170.0) + out = scores_db.trends_json(db, fields=["story_overall", "s_per_beat", "coldopen_s"]) + p = out["points"][0] + assert p["story_overall"] == 4.0 + assert p["s_per_beat"] == 80.5 + assert p["coldopen_s"] == 170.0 + + +def test_trends_json_custom_fields(tmp_path): + db = tmp_path / "t.db" + scores_db.add_run("a", db_path=db, surface="engine-duo", angrydm_overall=4.1) + out = scores_db.trends_json(db, fields=["angrydm_overall"]) + assert out["fields"] == ["angrydm_overall"] + assert out["points"][0]["angrydm_overall"] == 4.1 + assert "story_overall" not in out["points"][0] + + +def test_trends_json_rejects_unknown_field(tmp_path): + db = tmp_path / "t.db" + scores_db.connect(db).close() + with pytest.raises(ValueError): + scores_db.trends_json(db, fields=["not_a_column"]) + + +def test_trends_json_fences_by_surface(tmp_path): + """surface= keeps only matching-surface runs out of the trend (fencing).""" + db = tmp_path / "t.db" + scores_db.add_run("duo", db_path=db, surface="engine-duo", ts="2026-05-01T00:00:00+00:00", + story_overall=4.0) + scores_db.add_run("gui", db_path=db, surface="GUI-built-app", ts="2026-05-02T00:00:00+00:00", + cross_persona_sat=2.0) + out = scores_db.trends_json(db, surface="engine-duo") + assert [p["run_id"] for p in out["points"]] == ["duo"] + assert out["fence"]["surface"] == "engine-duo" + + +def test_trends_json_fences_by_lens_config_version(tmp_path): + """lens_config_version= keeps only runs scored under that lens ruler (the comparability fence).""" + db = tmp_path / "t.db" + scores_db.add_run("rulerA", db_path=db, surface="engine-duo", ts="2026-05-01T00:00:00+00:00", + lens_config_version="lc_aaaaaaaaaaaa", story_overall=4.5) + scores_db.add_run("rulerB", db_path=db, surface="engine-duo", ts="2026-05-02T00:00:00+00:00", + lens_config_version="lc_bbbbbbbbbbbb", story_overall=3.6) + out = scores_db.trends_json(db, lens_config_version="lc_aaaaaaaaaaaa") + assert [p["run_id"] for p in out["points"]] == ["rulerA"] + assert out["fence"]["lens_config_version"] == "lc_aaaaaaaaaaaa" + + +def test_trends_json_limit_keeps_last_n(tmp_path): + """limit=N keeps the N most-recent runs (still emitted oldest-first).""" + db = tmp_path / "t.db" + for d in range(1, 6): # 5 runs, ts 2026-05-01..05 + scores_db.add_run(f"r{d}", db_path=db, surface="engine-duo", + ts=f"2026-05-0{d}T00:00:00+00:00", story_overall=float(d)) + out = scores_db.trends_json(db, limit=2) + # the two newest runs (r4, r5), chronological + assert [p["run_id"] for p in out["points"]] == ["r4", "r5"] + + +def test_trends_json_empty_db(tmp_path): + db = tmp_path / "t.db" + scores_db.connect(db).close() + out = scores_db.trends_json(db) + assert out["points"] == [] + assert set(out["fields"]) >= {"story_overall", "rri"} + + +# --------------------------------------------------------------------------- +# Phase-3 observability reader (2): reconcile — READ-ONLY ledger<->INDEX.jsonl check +# --------------------------------------------------------------------------- + +def _write_index(path: Path, lines: list) -> None: + """Write an INDEX.jsonl-shaped file (one JSON object per line).""" + with path.open("w", encoding="utf-8") as fh: + for obj in lines: + if isinstance(obj, str): + fh.write(obj + "\n") # raw line (for malformed-line tests) + else: + fh.write(json.dumps(obj) + "\n") + + +def test_reconcile_detects_orphans_both_directions(tmp_path): + """A ledger row with no index line, and an index line with no ledger row, are each reported.""" + db = tmp_path / "scores.db" + idx = tmp_path / "INDEX.jsonl" + # ledger: two rows + scores_db.add_run("shared-run", db_path=db, surface="engine-duo", story_overall=4.0) + scores_db.add_run("orphan-ledger", db_path=db, surface="engine-duo", story_overall=3.5) + # index: the shared run + one index-only row (real INDEX uses the "id" key) + _write_index(idx, [ + {"kind": "run", "id": "shared-run", "path": "qa/ui_playtest_runs/shared-run"}, + {"kind": "run", "id": "orphan-index", "path": "qa/ui_playtest_runs/orphan-index"}, + ]) + rep = scores_db.reconcile(db, idx) + assert rep["in_ledger_not_index"] == ["orphan-ledger"] + assert rep["in_index_not_ledger"] == ["orphan-index"] + assert "shared-run" not in rep["in_ledger_not_index"] + assert "shared-run" not in rep["in_index_not_ledger"] + assert rep["matched_count"] == 1 + assert rep["ledger_count"] == 2 + assert rep["index_count"] == 2 + + +def test_reconcile_is_tolerant_of_run_id_key_variants(tmp_path): + """INDEX may use id / run_id / run for the run identifier — all are recognized.""" + db = tmp_path / "scores.db" + idx = tmp_path / "INDEX.jsonl" + scores_db.add_run("by-id", db_path=db, surface="engine-duo") + scores_db.add_run("by-run-id", db_path=db, surface="engine-duo") + scores_db.add_run("by-run", db_path=db, surface="engine-duo") + _write_index(idx, [ + {"id": "by-id"}, + {"run_id": "by-run-id"}, + {"run": "by-run"}, + ]) + rep = scores_db.reconcile(db, idx) + assert rep["in_ledger_not_index"] == [] + assert rep["in_index_not_ledger"] == [] + assert rep["matched_count"] == 3 + + +def test_reconcile_skips_unparseable_and_idless_lines(tmp_path): + """Malformed JSON, blank lines, and JSON objects with no run-id key are SKIPPED + warned, never crash.""" + db = tmp_path / "scores.db" + idx = tmp_path / "INDEX.jsonl" + scores_db.add_run("good", db_path=db, surface="engine-duo") + _write_index(idx, [ + {"id": "good"}, + "", # blank line + "{not valid json", # malformed + {"kind": "rubric", "note": "no id here"}, # object w/o any run-id key + "[1,2,3]", # valid JSON but not an object + ]) + rep = scores_db.reconcile(db, idx) + assert rep["in_ledger_not_index"] == [] # "good" matched + assert rep["in_index_not_ledger"] == [] + assert rep["matched_count"] == 1 + # the unparseable / id-less lines are reported as skipped (tolerant, never raises) + assert len(rep["skipped_lines"]) >= 2 + + +def test_reconcile_does_not_rewrite_index(tmp_path): + """reconcile is READ-ONLY: INDEX.jsonl bytes are unchanged after the check.""" + db = tmp_path / "scores.db" + idx = tmp_path / "INDEX.jsonl" + scores_db.add_run("r", db_path=db, surface="engine-duo") + _write_index(idx, [{"id": "r"}, {"id": "extra"}]) + before = idx.read_bytes() + scores_db.reconcile(db, idx) + assert idx.read_bytes() == before + + +def test_reconcile_missing_index_file(tmp_path): + """A missing INDEX.jsonl yields all ledger rows as orphans, with index_count 0 (no crash).""" + db = tmp_path / "scores.db" + idx = tmp_path / "does-not-exist.jsonl" + scores_db.add_run("only-ledger", db_path=db, surface="engine-duo") + rep = scores_db.reconcile(db, idx) + assert rep["index_count"] == 0 + assert rep["in_ledger_not_index"] == ["only-ledger"] + assert rep["in_index_not_ledger"] == [] + + +def test_reconcile_orphan_lists_are_sorted(tmp_path): + """Orphan lists are deterministically sorted so the report diffs cleanly.""" + db = tmp_path / "scores.db" + idx = tmp_path / "INDEX.jsonl" + for rid in ("zeta", "alpha", "mike"): + scores_db.add_run(rid, db_path=db, surface="engine-duo") + _write_index(idx, [{"id": "yankee"}, {"id": "bravo"}]) + rep = scores_db.reconcile(db, idx) + assert rep["in_ledger_not_index"] == sorted(rep["in_ledger_not_index"]) + assert rep["in_index_not_ledger"] == sorted(rep["in_index_not_ledger"]) + assert rep["in_ledger_not_index"] == ["alpha", "mike", "zeta"] + assert rep["in_index_not_ledger"] == ["bravo", "yankee"] diff --git a/qa/test_visual_regression_check.py b/qa/test_visual_regression_check.py new file mode 100644 index 00000000..a9e9ea9a --- /dev/null +++ b/qa/test_visual_regression_check.py @@ -0,0 +1,272 @@ +#!/usr/bin/env python3 +"""Tests for qa/visual_regression_check.py — the GUI screenshot visual-regression signal. + +Run (single-process; NEVER xdist): + uv run --directory servers/engine python -m pytest qa/test_visual_regression_check.py -q -p no:xdist + +These tests synthesize tiny PNG / byte files in ``tmp_path`` only — this is a pure reader; +it never writes any committed data artifact and never runs a game / scores a transcript. +""" + +from __future__ import annotations + +import io +import json +import struct +import sys +from pathlib import Path + +import pytest + +QA_DIR = Path(__file__).resolve().parent +if str(QA_DIR) not in sys.path: + sys.path.insert(0, str(QA_DIR)) + +import visual_regression_check as vrc # noqa: E402 + +try: # PIL is optional; AUDIT-mode tests adapt to its presence. + from PIL import Image # noqa: F401 + + _HAVE_PIL = True +except Exception: # pragma: no cover - depends on the host env + _HAVE_PIL = False + + +# -------------------------------------------------------------------------------------- +# Fixture helpers — synthesize tiny PNG bytes (PIL when present, otherwise a hand-rolled +# valid minimal PNG so strict-mode tests never depend on PIL). +# -------------------------------------------------------------------------------------- +def _png_bytes_pil(w: int, h: int, color) -> bytes: + from PIL import Image + + buf = io.BytesIO() + Image.new("RGB", (w, h), color).save(buf, "PNG") + return buf.getvalue() + + +def _png_bytes_stdlib(w: int, h: int, byte_fill: int = 0) -> bytes: + """A structurally-valid-enough PNG for hash/dimension purposes: real 8-byte signature + + a real IHDR chunk (so dimension parsing works) followed by deterministic filler bytes. + Strict mode only hashes bytes + reads IHDR, so this is sufficient and PIL-free.""" + sig = b"\x89PNG\r\n\x1a\n" + ihdr_body = struct.pack(">IIBBBBB", w, h, 8, 2, 0, 0, 0) # 8-bit, truecolor RGB + ihdr = struct.pack(">I", len(ihdr_body)) + b"IHDR" + ihdr_body + struct.pack(">I", 0) + filler = bytes([byte_fill]) * 16 + return sig + ihdr + filler + + +def _make_png(w: int, h: int, color=(10, 20, 30), byte_fill: int = 0) -> bytes: + if _HAVE_PIL: + return _png_bytes_pil(w, h, color) + return _png_bytes_stdlib(w, h, byte_fill) + + +def _layout(root: Path, files: dict[str, bytes]) -> Path: + """Write ``{rel_path: bytes}`` under ``root`` (creating parent dirs). Returns ``root``.""" + for rel, data in files.items(): + p = root / rel + p.parent.mkdir(parents=True, exist_ok=True) + p.write_bytes(data) + return root + + +# -------------------------------------------------------------------------------------- +# stdlib helpers +# -------------------------------------------------------------------------------------- +def test_sha256_and_dims_are_pure_stdlib(tmp_path): + png = _make_png(13, 7) + f = tmp_path / "a.png" + f.write_bytes(png) + h = vrc.sha256_file(f) + assert isinstance(h, str) and len(h) == 64 + # Same bytes -> same hash. + assert vrc.sha256_file(f) == h + dims = vrc.png_dimensions(png) + assert dims == (13, 7) + + +def test_png_dimensions_returns_none_for_non_png(): + assert vrc.png_dimensions(b"not a png at all") is None + assert vrc.png_dimensions(b"") is None + + +# -------------------------------------------------------------------------------------- +# STRICT mode +# -------------------------------------------------------------------------------------- +def test_strict_identical_is_pass(tmp_path): + png = _make_png(8, 8) + base = _layout(tmp_path / "base", {"newbie/table.png": png, "newbie/combat.png": png}) + cand = _layout(tmp_path / "cand", {"newbie/table.png": png, "newbie/combat.png": png}) + res = vrc.compare(base, cand, mode="strict") + assert res["mode"] == "strict" + assert res["verdict"] == "PASS" + assert res["counts"]["changed"] == 0 + assert res["counts"]["matched"] == 2 + assert res["counts"]["missing_candidate"] == 0 + assert res["counts"]["missing_baseline"] == 0 + # Every compared view is PASS with matching hashes. + for v in res["views"]: + assert v["status"] == "PASS" + assert v["baseline_sha256"] == v["candidate_sha256"] + + +def test_strict_changed_byte_is_flag(tmp_path): + png = _make_png(8, 8, color=(10, 20, 30), byte_fill=0) + changed = _make_png(8, 8, color=(200, 10, 10), byte_fill=255) + assert png != changed + base = _layout(tmp_path / "base", {"newbie/table.png": png}) + cand = _layout(tmp_path / "cand", {"newbie/table.png": changed}) + res = vrc.compare(base, cand, mode="strict") + assert res["verdict"] == "FLAG" + assert res["counts"]["changed"] == 1 + view = res["views"][0] + assert view["status"] == "CHANGED" + assert view["baseline_sha256"] != view["candidate_sha256"] + + +def test_strict_dimension_change_is_flag(tmp_path): + a = _make_png(8, 8) + b = _make_png(16, 8) # same-ish content, different dimensions + base = _layout(tmp_path / "base", {"p/table.png": a}) + cand = _layout(tmp_path / "cand", {"p/table.png": b}) + res = vrc.compare(base, cand, mode="strict") + assert res["verdict"] == "FLAG" + view = res["views"][0] + assert view["status"] == "CHANGED" + assert view["baseline_dimensions"] != view["candidate_dimensions"] + + +# -------------------------------------------------------------------------------------- +# Missing files — reported, never a crash +# -------------------------------------------------------------------------------------- +def test_missing_candidate_is_reported_not_crash(tmp_path): + png = _make_png(8, 8) + base = _layout(tmp_path / "base", {"newbie/table.png": png, "newbie/map.png": png}) + cand = _layout(tmp_path / "cand", {"newbie/table.png": png}) # map.png absent + res = vrc.compare(base, cand, mode="strict") + assert res["counts"]["missing_candidate"] == 1 + missing = [v for v in res["views"] if v["status"] == "MISSING_CANDIDATE"] + assert len(missing) == 1 + assert missing[0]["view"] == "newbie/map.png" + # A baseline with no candidate is a regression-worthy FLAG (the view vanished from the run). + assert res["verdict"] == "FLAG" + + +def test_missing_baseline_is_reported_not_a_regression(tmp_path): + png = _make_png(8, 8) + base = _layout(tmp_path / "base", {"newbie/table.png": png}) + cand = _layout(tmp_path / "cand", {"newbie/table.png": png, "newbie/new_view.png": png}) + res = vrc.compare(base, cand, mode="strict") + assert res["counts"]["missing_baseline"] == 1 + extra = [v for v in res["views"] if v["status"] == "MISSING_BASELINE"] + assert len(extra) == 1 + assert extra[0]["view"] == "newbie/new_view.png" + # A new candidate view with no baseline is NOT a regression — nothing to compare against. + assert res["verdict"] == "PASS" + + +def test_nonexistent_dirs_do_not_crash(tmp_path): + res = vrc.compare(tmp_path / "nope-base", tmp_path / "nope-cand", mode="strict") + assert res["verdict"] in ("PASS", "EMPTY") + assert res["counts"]["matched"] == 0 + + +# -------------------------------------------------------------------------------------- +# AUDIT mode — perceptual; SKIP cleanly when PIL is absent, run when present +# -------------------------------------------------------------------------------------- +def test_audit_mode_behaves_per_pil_availability(tmp_path): + png = _make_png(8, 8, color=(10, 20, 30)) + base = _layout(tmp_path / "base", {"newbie/table.png": png}) + cand = _layout(tmp_path / "cand", {"newbie/table.png": png}) + res = vrc.compare(base, cand, mode="audit") + assert res["mode"] == "audit" + if not vrc.perceptual_available(): + # Clean skip: explicit, never a crash, never a false FLAG. + assert res["verdict"] == "SKIPPED" + assert "skip_reason" in res + assert res["skip_reason"] # non-empty human-readable message + else: + # Identical images -> zero perceptual distance -> PASS. + assert res["verdict"] == "PASS" + view = res["views"][0] + assert view.get("distance") == 0 or view.get("distance") == pytest.approx(0.0) + + +@pytest.mark.skipif(not _HAVE_PIL, reason="PIL absent; perceptual diff cannot run") +def test_audit_mode_detects_difference_when_pil_present(tmp_path): + a = _png_bytes_pil(8, 8, (0, 0, 0)) + b = _png_bytes_pil(8, 8, (255, 255, 255)) + base = _layout(tmp_path / "base", {"newbie/table.png": a}) + cand = _layout(tmp_path / "cand", {"newbie/table.png": b}) + res = vrc.compare(base, cand, mode="audit") + assert res["verdict"] == "PASS" # AUDIT never gates; it reports for human review + view = res["views"][0] + assert view["distance"] > 0 + # An above-threshold change is annotated as DIFF (advisory), not a hard FLAG. + assert view["status"] in ("DIFF", "PASS") + + +@pytest.mark.skipif(not _HAVE_PIL, reason="PIL absent; perceptual diff cannot run") +def test_audit_html_report_written_to_tmp(tmp_path): + a = _png_bytes_pil(8, 8, (0, 0, 0)) + b = _png_bytes_pil(8, 8, (255, 255, 255)) + base = _layout(tmp_path / "base", {"newbie/table.png": a}) + cand = _layout(tmp_path / "cand", {"newbie/table.png": b}) + out_html = tmp_path / "report.html" + res = vrc.compare(base, cand, mode="audit", report_html=out_html) + assert out_html.exists() + text = out_html.read_text() + assert "newbie/table.png" in text + assert res["report_html"] == str(out_html) + + +# -------------------------------------------------------------------------------------- +# ADDITIVE-BY-DEFAULT — empty inputs == no change (no FLAG, no crash) +# -------------------------------------------------------------------------------------- +def test_empty_baseline_dir_is_additive_noop(tmp_path): + base = _layout(tmp_path / "base", {}) # exists but empty + cand = _layout(tmp_path / "cand", {"newbie/table.png": _make_png(8, 8)}) + res = vrc.compare(base, cand, mode="strict") + # No baselines to compare -> not a regression. Candidate-only views reported, not flagged. + assert res["verdict"] in ("PASS", "EMPTY") + assert res["counts"]["changed"] == 0 + + +# -------------------------------------------------------------------------------------- +# CLI smoke — exit codes + --json, writing nothing outside tmp +# -------------------------------------------------------------------------------------- +def test_cli_strict_pass_exit_zero_and_json(tmp_path, capsys): + png = _make_png(8, 8) + base = _layout(tmp_path / "base", {"newbie/table.png": png}) + cand = _layout(tmp_path / "cand", {"newbie/table.png": png}) + rc = vrc.main( + ["--baseline-dir", str(base), "--candidate-dir", str(cand), "--mode", "strict", "--json"] + ) + assert rc == 0 + out = json.loads(capsys.readouterr().out) + assert out["verdict"] == "PASS" + assert out["mode"] == "strict" + + +def test_cli_strict_flag_exit_two(tmp_path): + base = _layout(tmp_path / "base", {"newbie/table.png": _make_png(8, 8, byte_fill=0)}) + cand = _layout(tmp_path / "cand", {"newbie/table.png": _make_png(8, 8, byte_fill=255, color=(9, 9, 9))}) + rc = vrc.main( + ["--baseline-dir", str(base), "--candidate-dir", str(cand), "--mode", "strict", "--json"] + ) + assert rc == 2 # FLAG -> non-zero so CI can gate + + +def test_cli_audit_never_gates_exit_zero(tmp_path): + png = _make_png(8, 8) + base = _layout(tmp_path / "base", {"newbie/table.png": png}) + cand = _layout(tmp_path / "cand", {"newbie/table.png": png}) + rc = vrc.main( + ["--baseline-dir", str(base), "--candidate-dir", str(cand), "--mode", "audit", "--json"] + ) + # AUDIT is advisory: PASS or SKIPPED, both exit 0 (never gates the build). + assert rc == 0 + + +if __name__ == "__main__": # pragma: no cover + raise SystemExit(pytest.main([__file__, "-q", "-p", "no:xdist"])) diff --git a/qa/visual_regression_check.py b/qa/visual_regression_check.py new file mode 100644 index 00000000..5b37be9c --- /dev/null +++ b/qa/visual_regression_check.py @@ -0,0 +1,431 @@ +#!/usr/bin/env python3 +"""Did a GUI screen visually REGRESS vs the committed baseline? — the QA harness's +machine-readable visual-regression signal for the implementing agent. + +After a GUI sweep captures one PNG per (persona, view) (e.g. via ``qa/owshot.sh`` / +``qa/screen_coverage.py``), the agent asks: *did any screen change in a way that warrants +a human look — or worse, did a whole element/screen VANISH?* This tool answers it by +diffing a candidate screenshot dir against a committed baseline dir. It is a PURE READER: +it never writes state, never runs a game, never touches a committed data artifact. + +TWO MODES (start strict; characterise the audit noise floor before gating on it) +-------------------------------------------------------------------------------- +* STRICT — stdlib only (``hashlib`` + a tiny IHDR parse). Compares the sha256 of the PNG + bytes and the pixel dimensions. Gates ONLY on a DEFINITE change: a byte/dimension diff + (something rendered differently), or a baseline view with NO candidate (a whole screen + vanished from the run). A new candidate-only view is NOT a regression (nothing to compare). + Verdict FLAG -> exit 2 so CI can gate. This is the only mode that should ever gate a build. + +* AUDIT — best-effort PERCEPTUAL diff for human review. Tries ``imagehash`` (perceptual + hash, Hamming distance), else falls back to a PIL-only mean per-pixel difference, else + SKIPS cleanly with a clear message if PIL is absent (we deliberately do NOT add a heavy + dependency to force it to run). AUDIT NEVER gates the build — it always exits 0 (PASS or + SKIPPED) and optionally emits an HTML/JSON report. Its purpose is to characterise the + noise floor (font hinting, antialiasing, cursor blink, clock text) BEFORE anyone decides + which views are stable enough to gate on in strict mode. + +ADDITIVE-BY-DEFAULT: an empty/absent baseline dir == today's behaviour (PASS / no flag). +Every gate here reads engine-rendered pixels on disk, never fiction. + +BASELINE LAYOUT (see qa/screenshot_baselines/README.md): + qa/screenshot_baselines/v//.png +A "view" key is the path of a PNG RELATIVE to the baseline (or candidate) root, so +``newbie/table.png`` in baseline is compared against ``newbie/table.png`` in candidate. + +USAGE +----- + python qa/visual_regression_check.py \ + --baseline-dir qa/screenshot_baselines/v1.0.4 \ + --candidate-dir /tmp/sweep-shots \ + --mode strict --json + + # From Python: + from visual_regression_check import compare + res = compare(baseline_dir, candidate_dir, mode="strict") # -> verdict dict + +Exit codes (so CI / the agent can gate): 0 = PASS / SKIPPED / EMPTY, 2 = FLAG (strict only). +""" + +from __future__ import annotations + +import argparse +import hashlib +import html +import json +import struct +import sys +from pathlib import Path +from typing import Optional + +_PNG_SIG = b"\x89PNG\r\n\x1a\n" + +# Verdict -> process exit code. AUDIT never gates, so it can only ever yield 0-mapped verdicts. +_EXIT = {"PASS": 0, "SKIPPED": 0, "EMPTY": 0, "FLAG": 2} + +# AUDIT noise-floor default: an advisory DIFF threshold, NOT a gate. Annotates views whose +# perceptual distance exceeds it as "DIFF" for human attention; below it stays "PASS". +# (See README: characterise the real noise floor on your host before trusting this number.) +DEFAULT_AUDIT_THRESHOLD = 0.0 + + +# ====================================================================================== +# stdlib primitives (STRICT mode) — no third-party imports +# ====================================================================================== +def sha256_file(path: Path | str) -> str: + """sha256 hex digest of a file's raw bytes (read in chunks; pure stdlib).""" + h = hashlib.sha256() + with open(path, "rb") as fh: + for chunk in iter(lambda: fh.read(65536), b""): + h.update(chunk) + return h.hexdigest() + + +def png_dimensions(data: bytes) -> Optional[tuple[int, int]]: + """(width, height) parsed from a PNG's IHDR chunk, or None if ``data`` is not a PNG. + Pure stdlib — does NOT need PIL. The IHDR is the first chunk: width/height are the two + big-endian uint32s at offsets 16/20 (8-byte signature + 4-byte length + b'IHDR').""" + if len(data) < 24 or data[:8] != _PNG_SIG or data[12:16] != b"IHDR": + return None + try: + w, h = struct.unpack(">II", data[16:24]) + except struct.error: + return None + return (int(w), int(h)) + + +def _collect_pngs(root: Path) -> dict[str, Path]: + """Map ``{view_key: path}`` for every ``*.png`` under ``root`` (recursive). view_key is the + POSIX-style path relative to ``root`` so baseline and candidate keys line up regardless of + OS path separators. A missing/empty dir yields ``{}`` (additive-by-default: no crash).""" + root = Path(root) + if not root.exists() or not root.is_dir(): + return {} + out: dict[str, Path] = {} + for p in sorted(root.rglob("*.png")): + if p.is_file(): + out[p.relative_to(root).as_posix()] = p + return out + + +# ====================================================================================== +# perceptual primitives (AUDIT mode) — optional, dependency-light +# ====================================================================================== +def perceptual_available() -> bool: + """True if AUDIT mode can run a perceptual diff at all (PIL present). We try ``imagehash`` + first for a better metric but fall back to a PIL-only diff, so PIL alone is sufficient.""" + try: + import PIL # noqa: F401 + except Exception: + return False + return True + + +def _perceptual_backend() -> str: + """Which AUDIT backend is available: 'imagehash' > 'pil' > 'none'.""" + if not perceptual_available(): + return "none" + try: + import imagehash # noqa: F401 + + return "imagehash" + except Exception: + return "pil" + + +def _perceptual_distance(baseline: Path, candidate: Path, backend: str) -> Optional[float]: + """A best-effort perceptual distance in [0, 1]-ish terms. 0 == perceptually identical. + Returns None if the images cannot be opened (reported, never crashes the run).""" + try: + from PIL import Image + + with Image.open(baseline) as a_im, Image.open(candidate) as b_im: + a_im.load() + b_im.load() + if backend == "imagehash": + import imagehash + + ha = imagehash.phash(a_im) + hb = imagehash.phash(b_im) + # Hamming distance normalised by the hash bit length -> [0, 1]. + bits = len(ha.hash) * len(ha.hash[0]) + return float(ha - hb) / float(bits) if bits else 0.0 + # PIL-only fallback: normalise to a common size + mode, then mean abs per-channel + # difference / 255 -> [0, 1]. Crude but dependency-free; good enough to characterise + # a noise floor for human review (NEVER gates). + a_rgb = a_im.convert("RGB").resize((32, 32)) + b_rgb = b_im.convert("RGB").resize((32, 32)) + # .tobytes() yields flat R,G,B,R,G,B... and is stable across Pillow versions + # (avoids the deprecated .getdata()). Mean abs per-channel diff / 255 -> [0, 1]. + a_px = a_rgb.tobytes() + b_px = b_rgb.tobytes() + n = min(len(a_px), len(b_px)) + if n == 0: + return 0.0 + total = sum(abs(a_px[i] - b_px[i]) for i in range(n)) + return (total / n) / 255.0 + except Exception: + return None + + +# ====================================================================================== +# comparison core +# ====================================================================================== +def compare( + baseline_dir: Path | str, + candidate_dir: Path | str, + *, + mode: str = "strict", + audit_threshold: float = DEFAULT_AUDIT_THRESHOLD, + report_html: Optional[Path | str] = None, +) -> dict: + """Compare candidate screenshots against baselines. Returns a machine-readable verdict dict. + + mode='strict' : stdlib sha256+dims; FLAG on any byte/dim change or a vanished baseline view. + mode='audit' : perceptual diff for human review; never FLAGs (PASS/SKIPPED only).""" + mode = (mode or "strict").lower() + if mode not in ("strict", "audit"): + raise ValueError(f"unknown mode {mode!r} (expected 'strict' or 'audit')") + + base_root = Path(baseline_dir) + cand_root = Path(candidate_dir) + baselines = _collect_pngs(base_root) + candidates = _collect_pngs(cand_root) + + result: dict = { + "mode": mode, + "baseline_dir": str(base_root), + "candidate_dir": str(cand_root), + "views": [], + "counts": { + "matched": 0, + "changed": 0, + "diff": 0, + "missing_candidate": 0, + "missing_baseline": 0, + "errored": 0, + }, + } + + all_keys = sorted(set(baselines) | set(candidates)) + + if mode == "audit": + return _compare_audit(result, all_keys, baselines, candidates, audit_threshold, report_html) + return _compare_strict(result, all_keys, baselines, candidates) + + +def _compare_strict(result: dict, keys, baselines, candidates) -> dict: + counts = result["counts"] + for view in keys: + b = baselines.get(view) + c = candidates.get(view) + if b is not None and c is None: + # A baseline view with no candidate = the screen vanished from the run -> regression. + counts["missing_candidate"] += 1 + result["views"].append({"view": view, "status": "MISSING_CANDIDATE", + "baseline_sha256": _safe_hash(b), "candidate_sha256": None}) + continue + if b is None and c is not None: + # A candidate view with no baseline = a NEW screen. Not a regression (nothing to + # compare); reported so a human can promote it into the next baseline set. + counts["missing_baseline"] += 1 + result["views"].append({"view": view, "status": "MISSING_BASELINE", + "baseline_sha256": None, "candidate_sha256": _safe_hash(c)}) + continue + # Both present: compare content hash + dimensions. + b_hash, b_dims, b_err = _hash_and_dims(b) + c_hash, c_dims, c_err = _hash_and_dims(c) + entry = { + "view": view, + "baseline_sha256": b_hash, + "candidate_sha256": c_hash, + "baseline_dimensions": list(b_dims) if b_dims else None, + "candidate_dimensions": list(c_dims) if c_dims else None, + } + if b_err or c_err: + counts["errored"] += 1 + entry["status"] = "ERROR" + entry["error"] = b_err or c_err + result["views"].append(entry) + continue + if b_hash == c_hash and b_dims == c_dims: + counts["matched"] += 1 + entry["status"] = "PASS" + else: + counts["changed"] += 1 + entry["status"] = "CHANGED" + result["views"].append(entry) + + # FLAG on a definite change OR a vanished baseline view. A new candidate-only view does + # NOT flag. No comparable baselines at all -> EMPTY (additive-by-default no-op). + flagged = counts["changed"] > 0 or counts["missing_candidate"] > 0 or counts["errored"] > 0 + if flagged: + result["verdict"] = "FLAG" + elif not baselines: + result["verdict"] = "EMPTY" + else: + result["verdict"] = "PASS" + result["message"] = _summary(result) + return result + + +def _compare_audit(result, keys, baselines, candidates, threshold, report_html) -> dict: + backend = _perceptual_backend() + result["backend"] = backend + result["audit_threshold"] = threshold + + if backend == "none": + # Clean skip — explicit, never a crash, never a false FLAG. + result["verdict"] = "SKIPPED" + result["skip_reason"] = ( + "AUDIT (perceptual) mode needs Pillow (PIL); it is not importable in this " + "environment. Install it (e.g. `uv pip install pillow`, optionally `imagehash` " + "for a perceptual-hash metric) or use --mode strict, which is stdlib-only. " + "AUDIT is advisory and never gates the build." + ) + result["message"] = result["skip_reason"] + return result + + counts = result["counts"] + for view in keys: + b = baselines.get(view) + c = candidates.get(view) + if b is not None and c is None: + counts["missing_candidate"] += 1 + result["views"].append({"view": view, "status": "MISSING_CANDIDATE", "distance": None}) + continue + if b is None and c is not None: + counts["missing_baseline"] += 1 + result["views"].append({"view": view, "status": "MISSING_BASELINE", "distance": None}) + continue + dist = _perceptual_distance(b, c, backend) + if dist is None: + counts["errored"] += 1 + result["views"].append({"view": view, "status": "ERROR", "distance": None, + "error": "could not open one of the images"}) + continue + if dist > threshold: + counts["diff"] += 1 + status = "DIFF" + else: + counts["matched"] += 1 + status = "PASS" + result["views"].append({"view": view, "status": status, "distance": round(dist, 6), + "baseline": str(b), "candidate": str(c)}) + + # AUDIT NEVER gates: PASS regardless of how many DIFFs — it is for human review only. + result["verdict"] = "PASS" + if report_html is not None: + out = Path(report_html) + out.write_text(_render_html(result)) + result["report_html"] = str(out) + result["message"] = _summary(result) + return result + + +# ====================================================================================== +# helpers +# ====================================================================================== +def _safe_hash(path: Path) -> Optional[str]: + try: + return sha256_file(path) + except OSError: + return None + + +def _hash_and_dims(path: Path): + """Return (sha256|None, (w,h)|None, error|None). Never raises.""" + try: + data = path.read_bytes() + except OSError as e: + return (None, None, f"read failed: {e}") + return (hashlib.sha256(data).hexdigest(), png_dimensions(data), None) + + +def _summary(result: dict) -> str: + c = result["counts"] + lines = [f"{result['verdict']} ({result['mode']}): " + f"matched={c['matched']} changed={c['changed']} diff={c['diff']} " + f"missing_candidate={c['missing_candidate']} missing_baseline={c['missing_baseline']} " + f"errored={c['errored']}"] + if result.get("skip_reason"): + return result["skip_reason"] + for v in result["views"]: + if v["status"] in ("PASS",): + continue + extra = "" + if "distance" in v and v["distance"] is not None: + extra = f" distance={v['distance']}" + elif v["status"] == "CHANGED": + extra = f" {v.get('baseline_dimensions')} -> {v.get('candidate_dimensions')}" + lines.append(f" {v['status']:18s} {v['view']}{extra}") + return "\n".join(lines) + + +def _render_html(result: dict) -> str: + """A minimal self-contained HTML report for human review of AUDIT diffs. Embeds nothing + binary (just links the on-disk paths) so it stays tiny and writes wherever the caller asks.""" + rows = [] + for v in result["views"]: + view = html.escape(v["view"]) + status = html.escape(v["status"]) + dist = v.get("distance") + dist_s = "" if dist is None else f"{dist}" + base = html.escape(v.get("baseline", "") or "") + cand = html.escape(v.get("candidate", "") or "") + rows.append( + f"{view}{status}{dist_s}" + f"{base}{cand}" + ) + c = result["counts"] + return ( + "" + "WorldOS visual-regression AUDIT report" + "" + "

Visual-regression AUDIT report

" + f"

mode=audit backend={html.escape(str(result.get('backend')))} " + f"threshold={result.get('audit_threshold')}

" + f"

matched={c['matched']} diff={c['diff']} " + f"missing_candidate={c['missing_candidate']} missing_baseline={c['missing_baseline']} " + f"errored={c['errored']}

" + "

AUDIT is advisory and never gates the build. Characterise the noise floor " + "before promoting any view into a strict gate.

" + "" + "" + + "".join(rows) + + "
viewstatusdistancebaselinecandidate
" + ) + + +# ====================================================================================== +# CLI +# ====================================================================================== +def main(argv: Optional[list[str]] = None) -> int: + p = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) + p.add_argument("--baseline-dir", required=True, help="committed baseline screenshot dir") + p.add_argument("--candidate-dir", required=True, help="candidate (this-run) screenshot dir") + p.add_argument("--mode", choices=["strict", "audit"], default="strict", + help="strict = stdlib hash/dim gate (default); audit = perceptual review") + p.add_argument("--audit-threshold", type=float, default=DEFAULT_AUDIT_THRESHOLD, + help="AUDIT-only advisory DIFF threshold in [0,1] (never gates)") + p.add_argument("--report-html", help="AUDIT-only: write an HTML report to this path") + p.add_argument("--json", action="store_true", help="emit the machine-readable verdict as JSON") + args = p.parse_args(argv) + + result = compare( + args.baseline_dir, + args.candidate_dir, + mode=args.mode, + audit_threshold=args.audit_threshold, + report_html=args.report_html, + ) + + if args.json: + print(json.dumps(result, indent=2, default=str)) + else: + print(result.get("message", result["verdict"])) + return _EXIT.get(result["verdict"], 0) + + +if __name__ == "__main__": + raise SystemExit(main())