open/llm-ingestion-okf
1
0
Fork 0
Code Issues Releases 1 Activity

feat(tools): okf_contract_check reads the contract's checkable half

Browse source 
Fourteen rules, each with its own code: one "invalid" verdict over fourteen
different defects is a diagnostic no caller can act on. The report quotes its
own denominators -- rules run, excerpts and withheld entries examined -- because
a checker that exempted itself from section 5 would be stating the rule it
breaks.

Three exit codes, not two. "The check did not run" and "the check failed" are
different outcomes, and an unread file reported as a failed check is the fourth
face of the verification law.

The shipped template plus its example payload is the known-positive arm, so a
checker that refuses everything cannot be green on the thirteen negative ones.
A last test asserts every literal the checker enforces appears in the contract
document: two copies of a closed set drift, and the copy nobody reads is the
one that goes wrong.

996 -> 1023 tests.
This commit is contained in:
Kjell Tore Guttormsen 2026-09-02 16:09:27 +02:00
commit bc0b4130f1
2 changed files with 716 additions and 0 deletions
Download patch file Download diff file Expand all files Collapse all files

473
tools/okf_contract_check.py Normal file
View file

@ -0,0 +1,473 @@
"""Check a consumption skill and one pre-pass payload against the contract.
`docs/consumption-contract.md` is normative prose. Prose has no test, so this
command reads the mechanically checkable half of it: the payload shape (SS 3, 5,
6, 7, 8) and the fixed vocabulary a conformant `SKILL.md` must carry (SS 4, 6).
**What it cannot see, said here rather than implied.** The division of labour
(SS 2) and the prohibitions (SS 9) are properties of a RUN -- whether an agent
went looking for context the pre-pass withheld, whether a retrieval tool was
pointed at the verdict layer. No static check reaches them. Conformance here is
the floor, never the proof, and a green run means the payload carries what a
claim would have to rest on -- not that the claim was made honestly.
**Every rule has its own code.** One "invalid" verdict over fourteen different
defects is a diagnostic no caller can act on. The codes are the contract's
paragraphs made addressable.
**The report carries denominators**, because the contract requires them of its
consumers and a checker exempting itself would be stating the rule it breaks:
how many rules ran, over how many excerpts and withheld entries.
Exit codes are three, not two: 0 conformant, 1 non-conformant, 2 the check did
not run. Collapsing 2 into 1 would report an unread file as a failed check.
It lives outside `src/`, so it never enters a wheel and no consumer's install
surface changes because it exists.
"""
from __future__ import annotations
import argparse
import json
from collections.abc import Callable, Mapping, Sequence
from dataclasses import dataclass
from pathlib import Path
from typing import Any
#: SS 4.1. Five literals, spelled exactly. `[unverifiable-from-bundle]` is one
#: literal string -- no variants, no translations -- so this is a membership
#: test on the literal, never on a translated concept name.
REQUIRED_MARKINGS = (
"extracted",
"derived",
"[unverifiable-from-bundle]",
"[unread]",
"[sourced-not-sufficient]",
)
#: SS 6.1. A closed set of three, and the third is a real state: `unknown` says
#: the concept carries no `adjudication` key, which is what an older bundle
#: looks like. Collapsing it into `proposed` or into absence reports "not
#: judged" where the truth is "we cannot tell", and only one of those is a fact
#: about the concept.
ADJUDICATION_STATES = ("proposed", "adjudicated", "unknown")
#: SS 6.2, from SPEC SS 5.3: derived from `verified`, lowest to highest.
TRUST_TIERS = ("unverified", "machine-confirmed", "human-reviewed")
#: The level-2 headings a conformant skill carries. Checked by name because the
#: alternative -- searching the prose for the obligation -- passes on any
#: document that happens to use the words.
REQUIRED_SECTIONS = (
"Pre-pass",
"Division of labour",
"Markings",
"States",
"Budget",
"Denominators",
"Prohibitions",
)
@dataclass(frozen=True)
class Finding:
"""One contract paragraph, unmet, named by its code."""
code: str
message: str
@dataclass(frozen=True)
class Report:
"""Findings plus the denominators they were measured over."""
findings: tuple[Finding, ...]
rules_evaluated: int
excerpts_examined: int
withheld_examined: int
def render(self) -> str:
denominator = (
f"{self.rules_evaluated} rules over {self.excerpts_examined} excerpts "
f"and {self.withheld_examined} withheld entries"
)
if not self.findings:
return f"conformant: {denominator}, 0 findings"
lines = [f"NOT conformant: {denominator}, {len(self.findings)} findings"]
lines += [f" {finding.code}: {finding.message}" for finding in self.findings]
return "\n".join(lines)
@dataclass(frozen=True)
class Context:
"""What every rule reads. `payload` is empty when the payload is not a
mapping at all, so each rule stays a total function over its input."""
skill: str
payload: Mapping[str, Any]
payload_is_mapping: bool
def _mapping(value: object) -> Mapping[str, Any]:
return value if isinstance(value, Mapping) else {}
def _sequence(value: object) -> Sequence[Any]:
if isinstance(value, Sequence) and not isinstance(value, (str, bytes)):
return value
return ()
def _text(value: object) -> str:
return value if isinstance(value, str) else ""
def _whole(value: object) -> int | None:
# `bool` is an `int` in Python and would make `True` a budget. Excluded by
# name rather than trusted not to arrive.
return value if isinstance(value, int) and not isinstance(value, bool) else None
# --- Payload rules -----------------------------------------------------------
def rule_payload_shape(ctx: Context) -> list[Finding]:
if ctx.payload_is_mapping:
return []
return [Finding("payload_invalid", "the payload is not a JSON object (SS 8)")]
def rule_contract_version(ctx: Context) -> list[Finding]:
if not ctx.payload_is_mapping or _text(ctx.payload.get("contract")):
return []
return [
Finding(
"contract_unversioned",
"the payload names no `contract` revision, so a reader cannot tell "
"which revision it is holding (SS 8.2)",
)
]
def rule_bundle_ref(ctx: Context) -> list[Finding]:
if not ctx.payload_is_mapping:
return []
bundle = _mapping(ctx.payload.get("bundle"))
findings = []
if not _text(bundle.get("ref")):
findings.append(
Finding(
"ref_missing",
"the payload names no bundle `ref`; a version is the producer's "
"assertion, a ref is a fact about bytes (SS 3.3)",
)
)
if not _text(bundle.get("bundle_id")):
findings.append(Finding("ref_missing", "the payload names no `bundle.bundle_id` (SS 3.1)"))
return findings
def rule_excerpt_source_marking(ctx: Context) -> list[Finding]:
if not ctx.payload_is_mapping:
return []
findings = []
for position, raw in enumerate(_sequence(ctx.payload.get("excerpts"))):
excerpt = _mapping(raw)
for key in ("bundle_id", "concept_id", "sha256"):
if not _text(excerpt.get(key)):
findings.append(
Finding(
"source_marking_incomplete",
f"excerpt {position} carries no {key!r}; identity across "
"bundles is the (bundle_id, concept_id) tuple with the "
"digest of the bytes it was taken from (SS 3.1, SS 3.2)",
)
)
return findings
def rule_excerpt_states(ctx: Context) -> list[Finding]:
if not ctx.payload_is_mapping:
return []
findings = []
closed: tuple[tuple[str, tuple[str, ...]], ...] = (
("adjudication", ADJUDICATION_STATES),
("trust_tier", TRUST_TIERS),
)
for position, raw in enumerate(_sequence(ctx.payload.get("excerpts"))):
excerpt = _mapping(raw)
for key, allowed in closed:
if key not in excerpt:
findings.append(
Finding(
"state_missing",
f"excerpt {position} carries no {key!r}; the third state is "
"written explicitly, never omitted -- omission collapses "
'"we cannot tell" into a fact about the concept (SS 6.1)',
)
)
continue
value = _text(excerpt.get(key))
if value not in allowed:
findings.append(
Finding(
"state_not_in_set",
f"excerpt {position} has {key}={value!r}, outside the closed "
f"set {allowed} (SS 6)",
)
)
return findings
def rule_denominator_identity(ctx: Context) -> list[Finding]:
if not ctx.payload_is_mapping:
return []
counts = _mapping(ctx.payload.get("denominators"))
values = {key: _whole(counts.get(key)) for key in ("considered", "withheld", "delivered")}
missing = sorted(key for key, value in values.items() if value is None)
if missing:
return [
Finding(
"denominator_identity",
f"the payload reports no whole-number {', '.join(missing)}; a count "
"that is not reported is unmeasured, not zero (SS 5.1)",
)
]
considered = values["considered"]
withheld = values["withheld"]
delivered = values["delivered"]
assert considered is not None and withheld is not None and delivered is not None
if considered != withheld + delivered:
return [
Finding(
"denominator_identity",
f"considered ({considered}) != withheld ({withheld}) + delivered "
f"({delivered}); a count that does not close is not a denominator "
"(SS 5.2)",
)
]
return []
def rule_denominator_lists(ctx: Context) -> list[Finding]:
if not ctx.payload_is_mapping:
return []
counts = _mapping(ctx.payload.get("denominators"))
findings = []
for key, member in (("delivered", "excerpts"), ("withheld", "withheld")):
declared = _whole(counts.get(key))
if declared is None:
continue
actual = len(_sequence(ctx.payload.get(member)))
if declared != actual:
findings.append(
Finding(
"denominator_list_mismatch",
f"denominators.{key} is {declared} but {member} holds {actual}; "
"the count and the list are two statements of one fact (SS 8.1)",
)
)
return findings
def rule_withheld_rules(ctx: Context) -> list[Finding]:
if not ctx.payload_is_mapping:
return []
findings = []
for position, raw in enumerate(_sequence(ctx.payload.get("withheld"))):
entry = _mapping(raw)
for key in ("concept_id", "rule"):
if not _text(entry.get(key)):
findings.append(
Finding(
"cut_undeclared",
f"withheld entry {position} carries no {key!r}; a visible "
"drop is worth more than a silent one (SS 5.3)",
)
)
return findings
def rule_budget_declared(ctx: Context) -> list[Finding]:
if not ctx.payload_is_mapping:
return []
budget = _mapping(ctx.payload.get("budget"))
findings = []
for key in ("unit", "instrument"):
if not _text(budget.get(key)):
findings.append(
Finding(
"budget_undeclared",
f"the budget names no {key!r}; a number without its instrument "
"and unit is not a measurement (SS 7.1)",
)
)
limit = _whole(budget.get("limit"))
if limit is None or limit <= 0:
findings.append(
Finding(
"budget_undeclared",
'"bounded" without a bound is a denominator failure in prose (SS 7.1)',
)
)
if _whole(budget.get("spent")) is None:
findings.append(Finding("budget_undeclared", "the budget reports no `spent` (SS 7.2)"))
return findings
def rule_budget_gate(ctx: Context) -> list[Finding]:
if not ctx.payload_is_mapping:
return []
budget = _mapping(ctx.payload.get("budget"))
limit = _whole(budget.get("limit"))
spent = _whole(budget.get("spent"))
if limit is None or spent is None or spent <= limit:
return []
return [
Finding(
"budget_exceeded",
f"spent ({spent}) exceeds limit ({limit}); the cut strategy is wrong "
"for this bundle, which is a finding requiring a decision and never "
"a retry with a narrower question (SS 7.3)",
)
]
def rule_instrument_validated(ctx: Context) -> list[Finding]:
if not ctx.payload_is_mapping:
return []
known = _mapping(_mapping(ctx.payload.get("budget")).get("known_positive"))
expected = _whole(known.get("expected"))
measured = _whole(known.get("measured"))
if expected is None or measured is None or expected <= 0 or not _text(known.get("case")):
return [
Finding(
"instrument_unvalidated",
"the budget carries no usable `known_positive` (case, expected, "
"measured); an instrument that has not reproduced a known figure "
"has not been shown to count (SS 7.4)",
)
]
if expected != measured:
return [
Finding(
"instrument_unvalidated",
f"the known-positive expected {expected} and the instrument "
f"measured {measured} (SS 7.4)",
)
]
return []
# --- Skill rules -------------------------------------------------------------
def rule_skill_sections(ctx: Context) -> list[Finding]:
return [
Finding(
"skill_section_missing",
f"the skill carries no `## {section}` section (SS 8 of this checker's "
"reading; the headings are fixed so they can be checked by name)",
)
for section in REQUIRED_SECTIONS
if f"## {section}" not in ctx.skill
]
def rule_skill_markings(ctx: Context) -> list[Finding]:
return [
Finding(
"skill_marking_missing",
f"the skill does not carry the required marking {marking!r} verbatim "
"(SS 4.1); one literal string, no variants, no translations",
)
for marking in REQUIRED_MARKINGS
if marking not in ctx.skill
]
def rule_skill_states(ctx: Context) -> list[Finding]:
return [
Finding(
"skill_state_missing",
f"the skill does not name the state {state!r}, so its consumer cannot "
"be held to the closed set (SS 6)",
)
for state in (*ADJUDICATION_STATES, *TRUST_TIERS)
if f"`{state}`" not in ctx.skill
]
#: Every rule, in report order. `len(RULES)` is the denominator the report
#: quotes: "how many rules ran" is the number that makes "0 findings" mean
#: something.
RULES: tuple[Callable[[Context], list[Finding]], ...] = (
rule_payload_shape,
rule_contract_version,
rule_bundle_ref,
rule_excerpt_source_marking,
rule_excerpt_states,
rule_denominator_identity,
rule_denominator_lists,
rule_withheld_rules,
rule_budget_declared,
rule_budget_gate,
rule_instrument_validated,
rule_skill_sections,
rule_skill_markings,
rule_skill_states,
)
def check(skill_text: str, payload: object) -> Report:
"""Run every rule. No rule short-circuits another: a caller fixing one
defect should not discover a second only on the next run."""
is_mapping = isinstance(payload, Mapping)
ctx = Context(
skill=skill_text,
payload=payload if isinstance(payload, Mapping) else {},
payload_is_mapping=is_mapping,
)
findings: list[Finding] = []
for rule in RULES:
findings.extend(rule(ctx))
return Report(
findings=tuple(findings),
rules_evaluated=len(RULES),
excerpts_examined=len(_sequence(ctx.payload.get("excerpts"))),
withheld_examined=len(_sequence(ctx.payload.get("withheld"))),
)
def parse_args(argv: list[str] | None) -> argparse.Namespace:
parser = argparse.ArgumentParser(
description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter
)
parser.add_argument("--skill", type=Path, required=True, help="the SKILL.md to check")
parser.add_argument("--payload", type=Path, required=True, help="one pre-pass payload (JSON)")
return parser.parse_args(argv)
def main(argv: list[str] | None = None) -> int:
args = parse_args(argv)
try:
skill_text = args.skill.read_text(encoding="utf-8")
except OSError as exc:
print(f"could not read the skill: {exc}")
return 2
try:
payload = json.loads(args.payload.read_text(encoding="utf-8"))
except OSError as exc:
print(f"could not read the payload: {exc}")
return 2
except json.JSONDecodeError as exc:
print(f"the payload is not readable JSON: {exc}")
return 2
report = check(skill_text, payload)
print(report.render())
return 1 if report.findings else 0
if __name__ == "__main__":
raise SystemExit(main())