Rifqi Hafizuddin
Claude Opus 5
[NOTICKET] feat: assert on invariants, report a band, classify the genre (Steps 0 + 5)
f1d64b1 | """Step 0 — assert on FIELDS AND INVARIANTS, never on exact text. | |
| Why this file exists. `temperature=0` is argmax over logits that a shared | |
| inference server does not hand back identically twice, and this corpus has a | |
| recorded case of identical input scoring 6/7 one day and 0/12 the next. So an | |
| assertion of the form "the definition equals this sentence" is not a test — it | |
| is a coin flip that fails on a day the wording moves by a comma, and passes on a | |
| day the pipeline is quietly broken in a way the wording does not show. | |
| What survives that noise is structure. "A definition, if present, is locatable | |
| verbatim in the document" is true of every correct run and false of every broken | |
| one, no matter which sentence the model picked. These are those claims. | |
| Every check here is free: deterministic, local, no API call. They run after the | |
| paid stage on whatever it produced, so they cost nothing to keep on. | |
| **A violation is a defect in the pipeline, not a bad score.** A metric moving is | |
| information; an invariant breaking means a control leaked, so the runner exits | |
| non-zero on any violation regardless of how the scores looked. | |
| """ | |
| from __future__ import annotations | |
| from dataclasses import dataclass, field | |
| from typing import Any | |
| from src.knowledge_extraction.validate import span_present | |
| # Entry kinds and the field each one calls its identity. Checked by name rather | |
| # than by isinstance: entries reach here as `model_dump()` dicts, which is the | |
| # same shape that gets persisted, so this tests what is stored. | |
| ID_FIELDS = { | |
| "glossary": "term_id", | |
| "rule": "rule_id", | |
| "formula": "formula_id", | |
| # The domain entry kept `brief_id` through the v3 rename. Recorded here | |
| # rather than corrected: renaming it is a migration that invalidates every | |
| # expert approval recorded against the old prefix (K12's own reasoning). | |
| "document": "brief_id", | |
| } | |
| class Violation: | |
| check: str | |
| kind: str | |
| detail: str | |
| def as_dict(self) -> dict[str, str]: | |
| return {"check": self.check, "kind": self.kind, "detail": self.detail} | |
| class InvariantReport: | |
| violations: list[Violation] = field(default_factory=list) | |
| checked: dict[str, int] = field(default_factory=dict) | |
| def ok(self) -> bool: | |
| return not self.violations | |
| def as_dict(self) -> dict[str, Any]: | |
| return { | |
| "ok": self.ok, | |
| "n_violations": len(self.violations), | |
| "checked": self.checked, | |
| "violations": [v.as_dict() for v in self.violations[:50]], | |
| } | |
| def check_invariants(result, doc) -> InvariantReport: | |
| """Structural claims that must hold of ANY correct run of ANY document.""" | |
| report = InvariantReport() | |
| add = report.violations.append | |
| source = "\n".join(chunk.text or "" for chunk in doc.chunks) | |
| n_pages = doc.n_pages or 0 | |
| by_kind = { | |
| "glossary": result.glossary, | |
| "rule": result.rules, | |
| "formula": result.formulas, | |
| "document": [result.brief] if result.brief else [], | |
| } | |
| report.checked = {kind: len(entries) for kind, entries in by_kind.items()} | |
| for kind, entries in by_kind.items(): | |
| id_field = ID_FIELDS[kind] | |
| seen: set[str] = set() | |
| for entry in entries: | |
| label = entry.get("term") or entry.get("name") or entry.get(id_field) or "?" | |
| # 1 — identity. Without it a re-run cannot tell an expert which rows | |
| # they already cleared, which is worse than a wrong definition. | |
| entity_id = entry.get(id_field) | |
| if not entity_id: | |
| add(Violation("entity_id_present", kind, f"{label}: {id_field} missing")) | |
| elif entity_id in seen: | |
| add(Violation("entity_id_unique", kind, f"{entity_id} appears twice")) | |
| else: | |
| seen.add(entity_id) | |
| provenance = entry.get("provenance") or {} | |
| span = (provenance.get("span") or "").strip() | |
| # 2 — a claim that survived carries a span, and that span is real. | |
| # This is the span check restated from the outside: if it ever | |
| # fires, the guard leaked rather than the model misbehaved. | |
| if _has_content(entry, kind): | |
| if not span: | |
| add(Violation("span_present", kind, f"{label}: content with no span")) | |
| elif not span_present(span, source): | |
| add( | |
| Violation( | |
| "span_locatable", | |
| kind, | |
| f"{label}: span not found verbatim in the document " | |
| f"({span[:60]!r})", | |
| ) | |
| ) | |
| # 3 — provenance points somewhere that exists. | |
| page = provenance.get("page") | |
| if page is not None and n_pages and not 0 <= page < n_pages: | |
| add( | |
| Violation( | |
| "page_in_range", | |
| kind, | |
| f"{label}: page {page} outside 0..{n_pages - 1}", | |
| ) | |
| ) | |
| # 4 — a rejected field is NULL on the entry it came from. The contract says | |
| # rejected, never repaired; this is the only check that can tell the two | |
| # apart after the fact. | |
| for rejection in result.rejected: | |
| entry = _find_entry(by_kind, rejection.entry_term) | |
| if entry is not None and entry.get(rejection.field) not in (None, "", []): | |
| add( | |
| Violation( | |
| "rejection_is_null", | |
| rejection.branch, | |
| f"{rejection.entry_term}.{rejection.field} was rejected but still " | |
| "carries a value — a repaired span is an unfalsifiable claim", | |
| ) | |
| ) | |
| _check_queue(result.review_queue, report) | |
| return report | |
| def _check_queue(queue: list[dict], report: InvariantReport) -> None: | |
| """The consumption surface. A UI binds these, so their shape is a contract.""" | |
| add = report.violations.append | |
| report.checked["review_queue"] = len(queue) | |
| ranks = [row.get("rank") for row in queue] | |
| if ranks != list(range(1, len(queue) + 1)): | |
| add(Violation("queue_ranks_contiguous", "review_queue", "ranks are not 1..N")) | |
| for row in queue: | |
| kind = row.get("row_kind") | |
| if kind not in ("term", "key_parameter"): | |
| add(Violation("row_kind_known", "review_queue", f"row_kind={kind!r}")) | |
| if kind == "key_parameter" and row.get("term_id") is not None: | |
| add( | |
| Violation( | |
| "key_parameter_has_no_term", | |
| "review_queue", | |
| f"{row.get('term')!r} carries a term_id, which is the one thing " | |
| "a key_parameter row is defined by NOT having", | |
| ) | |
| ) | |
| if "_conflict" in row or "_sort" in row: | |
| add( | |
| Violation( | |
| "no_private_keys", | |
| "review_queue", | |
| f"{row.get('term')!r} leaks a sorting key into the payload", | |
| ) | |
| ) | |
| def _has_content(entry: dict, kind: str) -> bool: | |
| """Did this entry actually assert something? An abstention is correct | |
| behaviour and carries no span obligation — 48 of 58 entries abstain on the | |
| reference document, and treating that as a violation would make the whole | |
| report noise.""" | |
| if kind == "glossary": | |
| return bool(entry.get("definition")) | |
| if kind == "rule": | |
| return bool(entry.get("statement")) | |
| if kind == "formula": | |
| return bool(entry.get("formula_latex")) | |
| return bool(entry.get("purpose_verbatim")) | |
| def _find_entry(by_kind: dict[str, list[dict]], label: str) -> dict | None: | |
| for entries in by_kind.values(): | |
| for entry in entries: | |
| if label in (entry.get("term"), entry.get("name")): | |
| return entry | |
| return None | |