"""The `/slots` shape contract, driven from the live capture (phase 07, task 01). The old gufo slot table read field names the server never published and rendered dashes forever; this file is the fix for the *process*, not just the markup: the shape the slot grid (task 04) will read is pinned against `tests/fixtures/slots_gufo.json` — the verbatim `/slots` reply recorded by `./scripts/capture-fixtures.sh` — and against the running gufo in `tests/test_live_fleet.py::test_gufo_slots_answer_the_pinned_shape`. If the fixture (or the live server) drifts, a test fails here, not a dashboard at 2am (`AGENTS.md` rule 7). The table and the assertion live in `tests/slots_contract.py` so both suites read one list, exactly like `tests/card_contract.py` does for `/system`. """ from __future__ import annotations import copy import re from collections.abc import Iterator from pathlib import Path from typing import Any import pytest from fastapi.testclient import TestClient from auth_client import sign_in from auth_contract import ADMIN_PASSWORD, FAST_PBKDF2 from dashboards.app import create_app from dashboards.config import Settings from dashboards.store import SampleStore from slots_contract import ( CANONICAL_TASK_ID_FIELD, REQUIRED_NEXT_TOKEN_FIELDS, REQUIRED_ROW_FIELDS, SLOT_CELLS, TASK_ID_ALIASES, assert_slot_shape, load_slots_capture, ) @pytest.fixture def rows() -> list[dict[str, Any]]: """A fresh mutable copy of the capture — mutations in one test cannot leak.""" return copy.deepcopy(load_slots_capture()) @pytest.fixture def client(tmp_path: Path) -> Iterator[TestClient]: """A running app, so the deletion assertions run on the bytes a browser would get.""" db = str(tmp_path / "slots-contract.sqlite3") store = SampleStore(db) app = create_app(Settings(db_path=db, admin_password=ADMIN_PASSWORD, pbkdf2_iterations=FAST_PBKDF2), store) test_client = TestClient(app) test_client.__enter__() try: # Signed in because this fixture reads `/llm` as served bytes, and `TestClient` follows # the 303 to /login: an unsigned client would check the deleted slot table against the # *login page* and pass without ever looking at /llm. sign_in(test_client) yield test_client finally: test_client.__exit__(None, None, None) store.close() def _code(body: str) -> str: """An asset with its comments stripped — prose may *name* a dead thing the code must not be.""" body = re.sub(r"/\*.*?\*/", " ", body, flags=re.S) body = re.sub(r"", " ", body, flags=re.S) return re.sub(r"^\s*//.*$", " ", body, flags=re.M) # --------------------------------------------------------------------------- # # The capture is real, and the contract passes on it # --------------------------------------------------------------------------- # def test_the_gufo_slots_capture_is_on_disk_and_is_a_real_answer( rows: list[dict[str, Any]]) -> None: """Every row carries a dozen fields — the whole point of capturing the wire. Without this, a two-key hand-written stub would satisfy "the key exists" and the contract would prove nothing (`test_ups_contract.py` holds the same line for the UPS captures). """ assert rows, ( "gufo answered /slots with no rows — even an idle engine reports one row per " "parallel slot, and that idle shape is exactly what the grid must handle" ) for row in rows: assert len(row) >= 10, f"{len(row)} keys — this is not a capture, it is a guess" def test_the_slot_grid_fields_hold_on_every_captured_row( rows: list[dict[str, Any]]) -> None: assert_slot_shape(rows) def test_the_capture_records_the_idle_shape_the_grid_must_handle( rows: list[dict[str, Any]]) -> None: """The assumption written into the phase: a fresh capture is an idle engine. Every row is idle with a blank prompt — the grid must render that as N idle cards, not as "no data". If a re-capture lands mid-request this test names the drift loudly instead of letting the contract quietly widen; busy rows are covered below. """ for row in rows: assert row["is_processing"] is False, "the capture caught a busy slot; re-run it when idle" assert row["state"] == 0 assert row["n_prompt_tokens"] == 0 def test_a_busy_row_of_the_same_shape_still_satisfies_the_contract( rows: list[dict[str, Any]]) -> None: """The contract is the row shape, not the idle numbers — a busy slot must pass too.""" busy = rows[0] busy.update({"is_processing": True, "state": 1, "id_task": 42, "n_prompt_tokens": 128, "n_prompt_tokens_processed": 64, "n_prompt_tokens_cache": 16}) busy["next_token"][0]["n_decoded"] = 7 busy["next_token"][0]["prob"] = 0.93 # llm_monitor prints it only when present assert_slot_shape(rows) def test_an_empty_row_list_passes_the_contract() -> None: """Zero slots is a state the grid renders (an explicit empty message), not a crash.""" assert_slot_shape([]) def test_the_contract_names_every_cell_the_grid_reads() -> None: """The doc table covers the llm_monitor cell set; a rename here is a rename there.""" assert set(SLOT_CELLS) == { "card key", "state pill", "request", "prompt tok", "cache hit", "generated", "remaining", "context", "model name", } for label, paths in SLOT_CELLS.items(): assert paths, f"{label} lists no row path" # --------------------------------------------------------------------------- # # Drift fails: the contract is only real if mutations are caught # --------------------------------------------------------------------------- # @pytest.mark.parametrize("field", sorted(REQUIRED_ROW_FIELDS)) def test_the_contract_fails_when_a_row_drops_a_field_the_grid_reads( rows: list[dict[str, Any]], field: str) -> None: del rows[0][field] with pytest.raises(AssertionError, match=field): assert_slot_shape(rows) @pytest.mark.parametrize(("field", "value"), [ ("id", "0"), # a string id would key two cards "0" and 0 apart ("is_processing", 0), # the legacy numeric state is not the authoritative bool ("state", "OL"), # this is not NUT ("n_ctx", None), # None would reach the number formatter ("n_prompt_tokens", True), # bool sneaks past a bare isinstance(x, int) ("next_token", {"n_decoded": 1}), ]) def test_the_contract_fails_on_a_type_the_grid_cannot_render( rows: list[dict[str, Any]], field: str, value: Any) -> None: rows[0][field] = value with pytest.raises(AssertionError, match=field): assert_slot_shape(rows) @pytest.mark.parametrize("field", sorted(REQUIRED_NEXT_TOKEN_FIELDS)) def test_the_contract_fails_when_next_token_loses_a_field( rows: list[dict[str, Any]], field: str) -> None: del rows[0]["next_token"][0][field] with pytest.raises(AssertionError, match=field): assert_slot_shape(rows) def test_the_contract_fails_when_the_canonical_request_id_is_missing( rows: list[dict[str, Any]]) -> None: """`id_task` is the key the grid prints, so its absence is drift, not a blank cell. The alias exists in the same row, and `llm_monitor` would have fallen back to it. This contract does not: a renamed fact fails here rather than rendering a plausible-looking card out of a different field (`page-llm.js`'s `SLOT_FIELDS` names one key per cell). """ del rows[0][CANONICAL_TASK_ID_FIELD] with pytest.raises(AssertionError, match=CANONICAL_TASK_ID_FIELD): assert_slot_shape(rows) def test_the_request_id_alias_is_recorded_and_never_needed(rows: list[dict[str, Any]]) -> None: """Dropping the alias must not fail the contract — proof it is not a second source.""" assert set(TASK_ID_ALIASES) == {"task_id"} for row in rows: del row["task_id"] assert_slot_shape(rows) def test_the_contract_fails_on_duplicate_slot_ids(rows: list[dict[str, Any]]) -> None: assert len(rows) > 1, "the capture needs two slots for a dupe to be possible" rows[1]["id"] = rows[0]["id"] with pytest.raises(AssertionError, match="duplicate slot ids"): assert_slot_shape(rows) def test_the_contract_fails_when_the_two_state_fields_disagree( rows: list[dict[str, Any]]) -> None: rows[0]["is_processing"] = True # state stays 0 — which field is the pill to believe? with pytest.raises(AssertionError, match="state"): assert_slot_shape(rows) def test_the_contract_fails_when_a_row_is_not_an_object(rows: list[dict[str, Any]]) -> None: broken: list[Any] = list(rows) broken[0] = "idle" with pytest.raises(AssertionError, match="not an object"): assert_slot_shape(broken) # --------------------------------------------------------------------------- # # The deleted table stays deleted (completion criterion, on served bytes) # --------------------------------------------------------------------------- # def test_the_broken_slot_table_is_gone_from_html_js_and_css(client: TestClient) -> None: """No `#slots-card` / `#slots-table` / `#slots-empty` / `renderSlots` survives. Asserted on served bytes because there is no JS test runner (a locked anchor). The guessed field names go with it — those are the memory of an upstream the capture replaced. The `.chip.info` rule only ever painted that table's "has msg" cell, so it goes too (orphan-rule policy, `AGENTS.md`). """ html = _code(client.get("/llm").text) for gone in ("slots-card", "slots-table", "slots-empty"): assert f'id="{gone}"' not in html, f"/llm still renders the {gone}" js = _code(client.get("/static/page-llm.js").text) for gone in ("renderSlots", "with_message", "n_predict", "decode_per_second", "cache_usable"): assert gone not in js, f"page-llm.js still reads {gone}" css = client.get("/static/dash.css").text assert ".chip.info" not in css, "dash.css still styles a class nothing renders" def test_the_prompt_field_is_user_content_that_never_reaches_the_page( client: TestClient, rows: list[dict[str, Any]]) -> None: """A busy slot's `prompt` is the operator's in-flight request text. The capture script blanks it (fixture stays shape-true, content-free), and the page never reads the key — the grid renders tokens, not somebody's prompt. """ for row in rows: assert row.get("prompt", "") == "", "the capture script must blank prompt content" js = _code(client.get("/static/page-llm.js").text) # A slot-row read is `slot.prompt` / `row.prompt` / `s.prompt` / `["prompt"]`. A bare # `.prompt` ban is impossible — `state.charts.prompt` is a chart object — so the name # before the dot must be a whole word (the lookbehind is what stops `charts.prompt` # from matching through its trailing "s"). assert not re.search(r"(?