dashboards/tests/slots_contract.py

"""The `/slots` row-shape contract behind the `/llm` slot grid (phase 07, task 01).

The old gufo slot table was deleted unread: it looked for `slot.state` as a *string*,
`slot.n_prompt`, `slot.n_predict`, `slot.cache` and `slot.decode_per_second`, and the live
server publishes none of those names — every row rendered em-dashes forever. That is
`AGENTS.md` rule 7 failing in slow motion ("trust a memory of an upstream"), so the
replacement grid (task 04) is written against a *recorded* ground truth instead:
`tests/fixtures/slots_gufo.json`, the verbatim `/slots` reply from the running gufo,
re-captured by `./scripts/capture-fixtures.sh`.

The same device as `tests/card_contract.py`: the tables live here so the hermetic test
(`tests/test_llm_slots_contract.py`, driven from the fixture) and the live test
(`tests/test_live_fleet.py::test_gufo_slots_answer_the_pinned_shape`) assert one list and
cannot drift apart.

======================  ==========================================================
Slot card cell          Path in a `/slots` row
======================  ==========================================================
card key                ``id`` (unique per row — the grid keys one card per slot)
state pill              ``is_processing`` (authoritative bool) / ``state`` (legacy 0/1)
request                 ``id_task`` — canonical; the ``task_id`` alias is captured but
                        **never rendered**
prompt tok              ``n_prompt_tokens`` (+ ``n_prompt_tokens_processed`` sub-line)
cache hit               ``n_prompt_tokens_cache`` (+ ratio against ``n_prompt_tokens``)
generated               ``next_token[0].n_decoded`` (+ optional ``prob`` sub-line)
remaining               ``next_token[0].n_remain`` — where the capture puts it, never a
                        row-level ``n_remain`` fallback
context                 ``n_ctx``
model name              ``model``
======================  ==========================================================

The field names and types are the live capture's. `llm_monitor` reads the same set with
``a ?? b`` fallbacks (the task-id alias pair, ``n_remain`` at either level); the grid drops
them. Task 04's rule is that **this contract names the canonical key of each fact and the
page renders that one**, so a rename fails a test here instead of quietly reading a different
field. `llm_monitor` is the design reference, not a dependency.

``prompt`` is deliberately **not** a card cell: on a busy slot it is the user's in-flight
request text. `./scripts/capture-fixtures.sh` blanks it in the fixture, and the page must
never read it.

The capture is an **idle** engine — gufo answers one row per parallel slot even when
nothing is generating, so "every slot idle, prompt blank" is the shape the grid must
handle; that assumption is pinned in the hermetic test, and the mutation tests below keep
the contract honest for busy rows too.
"""
from __future__ import annotations

import json
from pathlib import Path
from typing import Any

FIXTURES = Path(__file__).resolve().parent / "fixtures"

#: The one name the grid prints for "which request is in this slot". The live capture answers
#: **both** `id_task` and `task_id` for the same fact (and `llm_monitor` reads `id_task`
#: first), so a name has to be made canonical rather than chained: `slot.id_task ??
#: slot.task_id` is exactly the guessing that made the deleted table impossible to trust. If a
#: re-capture ever answers only the alias, this constant changes and the grid follows — the
#: fallback never moves into the page (`page-llm.js`'s `SLOT_FIELDS` names this key once).
CANONICAL_TASK_ID_FIELD: str = "id_task"

#: Other names a row may answer for that one fact: type-checked when present, rendered by
#: nothing.
TASK_ID_ALIASES: tuple[str, ...] = ("task_id",)

#: Every name known for the request id, canonical first (for messages and cross-checks).
TASK_ID_FIELDS: tuple[str, ...] = (CANONICAL_TASK_ID_FIELD, *TASK_ID_ALIASES)

#: Doc table: slot-grid cell -> the row path(s) that cell reads. The authoritative
#: presence/type rules are the three tables below it.
SLOT_CELLS: dict[str, tuple[str, ...]] = {
    "card key": ("id",),
    "state pill": ("is_processing", "state"),
    "request": (CANONICAL_TASK_ID_FIELD,),
    "prompt tok": ("n_prompt_tokens", "n_prompt_tokens_processed"),
    "cache hit": ("n_prompt_tokens_cache",),
    "generated": ("next_token.n_decoded", "next_token.prob"),
    "remaining": ("next_token.n_remain",),
    "context": ("n_ctx",),
    "model name": ("model",),
}

#: Field -> (accepted types, why the grid needs it). Every row must carry every one of
#: these; a missing key is the drift that made the old table render dashes.
REQUIRED_ROW_FIELDS: dict[str, tuple[tuple[type, ...], str]] = {
    "id": ((int,), "the grid keys one card per slot — a duplicate or absent id is two "
                   "cards for one slot or a slot nobody updates"),
    "is_processing": (((bool,), "the authoritative busy/idle verdict behind the state pill")),
    "state": ((int,), "the legacy numeric state (0 idle / 1 processing); the pill prefers "
                      "is_processing but labels anything else it cannot name"),
    CANONICAL_TASK_ID_FIELD: ((int,), "the request cell — the only name the grid reads for "
                                      "it, so its absence is drift rather than a blank cell"),
    "n_ctx": ((int,), "the context cell — the slot's window size"),
    "n_prompt_tokens": ((int, float), "the prompt cell"),
    "n_prompt_tokens_processed": ((int, float), "the `processed …` sub-line while prefill runs"),
    "n_prompt_tokens_cache": ((int, float), "the cache-hit cell (ratio against n_prompt_tokens)"),
    "next_token": ((list,), "the generated cell reads the first entry's n_decoded"),
}

#: Fields the grid reads *when present* — absence degrades to `n/a`, presence must keep the
#: type. `model` names the slot header; `prob` is a sub-line llm_monitor prints only if
#: the server publishes it (this gufo does not, on the captured rows).
OPTIONAL_ROW_FIELDS: dict[str, tuple[tuple[type, ...], str]] = {
    "model": ((str,), "the model name beside the slot id"),
    "speculative": ((bool,), "whether speculative decoding is on for this slot"),
    "n_remain": ((int, float), "some builds answer it at the row level; the grid reads "
                               "next_token[0].n_remain, which every captured row carries"),
    "task_id": ((int,), f"the alias of {CANONICAL_TASK_ID_FIELD} — captured, never rendered"),
}

#: Every entry of `next_token` that is present must carry these; an empty list is legal
#  (the grid prints `n/a` for generated/remaining, never a fabricated 0).
REQUIRED_NEXT_TOKEN_FIELDS: dict[str, tuple[tuple[type, ...], str]] = {
    "n_decoded": ((int, float), "the generated cell"),
    "n_remain": ((int, float), "the remaining cell — the captured rows carry it here, "
                               "not at the row's top level"),
}

OPTIONAL_NEXT_TOKEN_FIELDS: dict[str, tuple[tuple[type, ...], str]] = {
    "prob": ((int, float), "the `p=…` sub-line; this capture omits it and the grid copes"),
}

#: Row fields that are user content: captured blanked, never rendered by the page.
USER_CONTENT_FIELDS: tuple[str, ...] = ("prompt",)


def load_slots_capture(name: str = "gufo") -> list[dict[str, Any]]:
    """The verbatim `/slots` capture, as the live server answered it."""
    path = FIXTURES / f"slots_{name}.json"
    if not path.exists():
        raise AssertionError(
            f"{path} is missing — run ./scripts/capture-fixtures.sh (it captures gufo's "
            "/slots whenever GUFO_BASE_URL is set)"
        )
    rows = json.loads(path.read_text())
    assert isinstance(rows, list), f"{path.name}: /slots answered {type(rows).__name__}, not a list"
    return rows


def _matches(value: Any, types: tuple[type, ...]) -> bool:
    """`bool` is a subclass of `int`; never let `true` answer for a number or vice versa."""
    if types == (bool,):
        return isinstance(value, bool)
    if any(t in (int, float) for t in types):
        return isinstance(value, (int, float)) and not isinstance(value, bool)
    return isinstance(value, types)


def _check_fields(row: dict[str, Any], table: dict[str, tuple[tuple[type, ...], str]],
                  *, require: bool, where: str) -> None:
    for field, (types, why) in table.items():
        if require:
            assert field in row, (
                f"{where}: no key {field!r} — {why}. The deleted slot table rendered dashes "
                "for exactly this kind of drift; fail here instead"
            )
        elif field not in row:
            continue
        assert _matches(row[field], types), (
            f"{where}.{field} is {type(row[field]).__name__}, expected "
            f"{'/'.join(t.__name__ for t in types)} — {why}"
        )


def assert_slot_shape(rows: list[Any]) -> None:
    """Every `/slots` row has the shape the slot grid reads. Empty list is legal.

    Drift in either direction fails: a renamed/missing key (including the canonical request
    id — the alias does not excuse its absence), a type the grid cannot format, or a
    duplicated slot id (keyed cards would collide).
    """
    assert isinstance(rows, list), f"/slots answered {type(rows).__name__}, not a list of rows"
    seen_ids: list[int] = []
    for index, row in enumerate(rows):
        where = f"row[{index}]"
        assert isinstance(row, dict), f"{where} is {type(row).__name__}, not an object"
        _check_fields(row, REQUIRED_ROW_FIELDS, require=True, where=where)
        _check_fields(row, OPTIONAL_ROW_FIELDS, require=False, where=where)

        # The canonical request id is required above. An alias may ride along (every
        # captured row answers both) and the grid ignores it.
        tokens = row["next_token"]
        for entry in tokens:
            assert isinstance(entry, dict), f"{where}.next_token entry is not an object"
            _check_fields(entry, REQUIRED_NEXT_TOKEN_FIELDS, require=True,
                          where=f"{where}.next_token")
            _check_fields(entry, OPTIONAL_NEXT_TOKEN_FIELDS, require=False,
                          where=f"{where}.next_token")

        # The pill: `is_processing` is authoritative, and `state == 0` must not contradict
        # it — llm_monitor's exact reading, so a busy row and an idle row stay tellable.
        if row["is_processing"]:
            assert row["state"] != 0, (
                f"{where}: is_processing=True with state=0 — the pill cannot tell the operator "
                "which field to believe"
            )

        seen_ids.append(row["id"])
    duplicates = sorted({i for i in seen_ids if seen_ids.count(i) > 1})
    assert not duplicates, (
        f"duplicate slot ids {duplicates} — the grid keys one card per slot and a dupe "
        "renders one card twice (or drops the other)"
    )