open/portfolio-optimiser-claude
1
0
Fork 0
Code Issues Releases 1 Activity

Compare commits

base: open:main
Branches Tags
open:main
open:v0.1.0
..
compare: open:v0.1.0
Branches Tags
open:main
open:v0.1.0

No commits in common. "main" and "v0.1.0" have entirely different histories.
main ... v0.1.0

18 changed files with 41 additions and 2075 deletions
Download patch file Download diff file Expand all files Collapse all files

25
CHANGELOG.md
View file

@ -7,31 +7,6 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
## [Unreleased]
### Changed
- **The SDK pin now has a proof behind it, not only a permission.** `claude-agent-sdk`
moved 0.2.120 → 0.2.139, and the version guard gained `_VERIFIED_THROUGH`: the newest
build whose source was actually read. The pin (`>=0.2.111,<0.3`, unchanged) says what
`uv` may resolve; it never said anyone had looked. Those are different questions, and
conflating them left 0.2.111–0.2.999 admissible while the premises were source-verified
only through 0.2.110 — `uv sync --upgrade` would have kept the suite green on an
unexamined build. A build newer than the last one read now fails RED, naming the
premises to re-check. A third question — "is something newer upstream?" — is
deliberately NOT asked: it needs the network, and this suite stays offline.
- **The SDK premises are checked, not merely printed.** They had existed only as a
sentence the failure message recited; nothing verified them, so a premise that stopped
being true would have surfaced on the one live paid run. They are now a table
introspected against the installed package, with the operator-facing prose derived from
that same table so the two cannot drift. The one premise introspection structurally
cannot see — that `query()` yields an `AssistantMessage` then a closing `ResultMessage`
— is named separately, and is the honest reason a human reading is still required.
- Re-verified at 0.2.139 source level: the public `query.py` is byte-identical to 0.2.120,
every premise field keeps its type and default, and the parser changes are additive
(a new `origin` passthrough, a new `ConversationResetMessage` this client ignores).
0.2.139 also added a skills path that can default `setting_sources` to
`["user", "project"]`; it fires only on `None`, so the explicit `[]` that carries the
S10 isolation fix is out of its reach.
## [0.1.0] - 2026-08-17
### Added

17
CLAUDE.md
View file

@ -15,9 +15,7 @@ alene; det er datert i README og skal ikke påstås om arbeid etter den datoen.
`docs/plan/`. Løpende state: `STATE.md` (local-only).
## Stack
Python ≥3.10. **Claude Agent SDK** (`claude-agent-sdk` ≥0.2, kildeverifisert 0.2.139 per
2026-08-18 — pinnen `>=0.2.111,<0.3` er en TILLATELSE, `_VERIFIED_THROUGH` i
`tests/test_sdk_version_guard.py` er beviset og går rødt på en ulest nyere build. Per
Python ≥3.10. **Claude Agent SDK** (`claude-agent-sdk` ≥0.2, verifisert 0.2.110 per
2026-07-03: MIT-lisens, bundler Claude Code CLI — ingen separat installasjon; import og
konstruksjon fungerer uten API-nøkkel, nøkkel trengs kun ved faktisk `query()`).
Pakkehåndtering: `uv`. Pydantic for kontrakts-validering.
@ -45,8 +43,7 @@ Pakkehåndtering: `uv`. Pydantic for kontrakts-validering.
- Sync: `uv sync`
- Test: `uv run pytest`
- Lint: `uv run ruff check .` + `uv run ruff format .`
- Type: `uv run mypy src scripts`
- Bevis en søm: `uv run python scripts/mutation_harness.py --help` (grønn-før/rød-etter)
- Type: `uv run mypy src`
## Arbeidsflyt (invarianter)
- **Rent teknisk rammeverk:** deployer eier DPIA/ROS/behandlingsformål — kun tekniske
@ -55,16 +52,6 @@ Pakkehåndtering: `uv`. Pydantic for kontrakts-validering.
artefakt påstår mer enn implementasjonen gjør; skriptede stand-ins merkes som det.
- **Deterministisk validator obligatorisk + blokkerende** (§3 steg 4); golden-suiten
(`shared/examples/bygg-energi-mikro/golden.json`) er ENESTE fasit (§7).
- **Mutasjonsharnesset er verktøyet, ikke en huskeregel** (beslutning 2026-08-25, etter ti
håndbygg): `scripts/mutation_harness.py` kjører grønn-før/rød-etter mot det EKTE treet og
håndhever de fire kontrollene som ellers svikter stille — unikt anker, node-id-en faktisk
kjørt (pytest rc=4 er IKKE rødt), restaurering sha256-verifisert fra disk, og rødhet på en
pinnet linje. Tre roller, og rekkefølgen følger av hva hver måler: `--red` grønn FØR
(måleapparatets positivkontroll) og rød ETTER (selve målingen, ikke besvarbar før
mutasjonen finnes), `--green` grønn i BEGGE (treffer den, landet mutasjonen for bredt).
Verktøyet ligger bevisst UTENFOR `tests/` — suiten kjører hver økt, og en fixture som
skriver til disk gjør enhver avbrutt kjøring til et mutert tre. Mutasjon av en KOPI i minnet
hører derimot hjemme i suiten og er der alt (`test_guard_red_when_*`).
- **Load-bearing tester** (§11): hver søm bevises med en test som blir RØD når sømmen
detaches — grønn-men-død er feilmoden regelen finnes for.
- **Stoppkriterier + budsjett-tak påkrevd ved oppstart** (§8; `contracts.py` fail-fast).

41
README.md
View file

@ -351,47 +351,6 @@ rather than borrowing the installed one — red the moment the stamp reads the e
instead of the producing client — and the committed S10 record stays un-back-filled),
and `test_sdk_isolation.py` (local config cannot capture the checker).
**Proving it — `scripts/mutation_harness.py`.**
A test that *claims* to be load-bearing is a claim. The proof is to detach the seam, run the
test, and require it to go **red** — then restore the tree and verify the restore. That
procedure was hand-written ten times before it became a script, and the reason it became one
is not the typing. It is that four of its controls fail *silently*:
| Control | What it catches | Why prose cannot enforce it |
|---|---|---|
| Anchor occurs exactly once | A mutation landing wider than the seam | Nothing complains when it matches twice |
| The node id actually ran | A mistyped id — `pytest` exits **4**, which is non-zero and so *mimics* red | `--collect-only -q` exits **0** on an id that does not exist |
| Restore read back from disk and sha256-compared | A mutation left behind under `.venv/`, which `git status` does not track | The run still prints "restored" |
| Redness lands on a pinned line | A red caused by a broken import rather than by the seam | Any garbage edit reds a test |
Three roles, and the order follows from what each one measures — not from a convention:
- `--red` must be green **before** (the measuring apparatus: the id resolves and passes right
now) and red **after** (the measurement itself — *can this go red at all?* is not answerable
until the mutation exists).
- `--green` must hold in **both** runs. If a control reds, the mutation landed wider than the
seam and the target's redness attributes to nothing.
```bash
uv run python scripts/mutation_harness.py \
--target src/portfolio_optimiser_claude/some_module.py \
--anchor 'the exact text, unique in the file' \
--replacement 'what detaches the seam' \
--red tests/test_x_loadbearing.py::TestY::test_the_seam \
--green tests/test_x_loadbearing.py::TestY::test_the_population
```
It refuses rather than reports: an ambiguous anchor, a target already red, an id pytest never
collected, a control that reddened, and a mutation the target did not notice are each a named
error, never a quiet pass. `tests/test_mutation_harness.py` proves it in both directions — it
catches a known-detaching mutation and refuses to call a harmless one a proof.
The *other* mutation class — mutate a copy in memory and call the guard directly — belongs in
the suite and is already there (`test_guard_red_when_*`). This tool deliberately lives outside
`tests/`: the suite runs constantly, and a fixture that writes to disk would turn every
interrupted run into a mutated tree.
## The ingest layer — CSV and SQL, in front of the loop
The method spec forbids query-time retrieval against the bundle (§3 Step 1), so **data

2
SECURITY.md
View file

@ -8,7 +8,7 @@ We take security seriously. If you discover a security vulnerability, please rep
### How to Report
Email: security@fromaitochitta.com
Email: hello@fromaitochitta.com
Include:
- Description of the vulnerability

142
docs/2026-08-18-funn-koeer-og-gater.md
View file

@ -1,142 +0,0 @@
# Målte funn, køer og gater (flyttet ut av STATE.md 2026-08-18)
> Bakgrunn: `STATE.md` er current state-of-play og har et tak på ~120 linjer. Den 18.08 lå
> fila på 149. Operatørbeslutningen samme dag var å flytte tre punkter hit og la STATE
> beholde ÉN peker per punkt på formen `fil § Overskrift` — et grep-bart paragraf-anker,
> aldri et linjenummer. **Ingenting er slettet:** state er flyttet, ikke fjernet.
>
> Alt under er MÅLERESULTATER med en dato, ikke fakta om verden. Datoen står ved hver
> påstand, og en påstand herfra er et premiss som skal verifiseres på nytt før den handles
> på — særlig tall som råtner (versjoner, testantall, oppstrøms-tilstand).
## Negative funn
Steder som ER undersøkt for en manglende søm, og hvor svaret var «ingen søm å feste». De er
ferdig avklart i økt 17–22 og skal **ikke re-måles** — enumerert her nettopp for at ingen
senere økt skal bruke tid på dem igjen:
- `test_preflight` (9 tilfeller)
- `test_zero_model_calls`
- `notify` :224
- `ir` :69
- `validator` :56
- `step1_expel` :99
- cli_paritet-parene
- `method_spec` + `ingest_spec` (6 tilfeller — merk: **ikke lenger dekkende**, se under)
- alle 8 fra økt 22
**Korreksjon 2026-08-18 (økt 32):** `ingest_spec`-oppføringen gjaldt den §12-ankrede vakten.
Økt 32 målte et hull den enumereringen ikke fanget — enhver seksjon §1–§11 kunne tømmes uten
at én test i suiten ble rød (10 av 11 seksjoner; §7 var eneste overlapp, via O2-ratchet-en).
Hullet er lukket med `_STRUCTURE_MARKERS` i `tests/test_ingest_spec_loadbearing.py` (commit
`40b4631`). Lærdommen er generell: **et negativt funn har et scope, og scopet er den
spørringen som ble kjørt** — ikke fila den ble kjørt mot.
## D7-speilingskøen
Åtte kandidater for speiling mellom D7-søsknene. **Ingen er besluttet** — de står som
kandidater, ikke som planlagt arbeid:
- S2.7
- S3.2
- S4.0 (`126807a`)
- (p) `to_ore` — TO kallsteder
- ~~(a)/(i) `unquote_scalar`~~ — **MÅLT 2026-08-31, se under**
- `mandate.py`
- A5 — TO halvdeler
- ~~B4 `ToolCallRecorder`~~ — **BESVART 2026-08-25, se under**
**B4 spurte OSS:** gir Claude-SDK-siden serveridentiteten gratis? MAF får verktøynavn UTEN
server-prefiks. Åpent hos søskenet: MAF S3.4 (F10).
### B4 — svaret er NEI (målt 2026-08-25, offline, mot `claude-agent-sdk` 0.2.139)
`create_sdk_mcp_server` emitterer det **bare** verktøynavnet. Serveridentiteten finnes — på
`McpSdkServerConfig["name"]` og `Server.name` — men den er **disjunkt fra hvert navn
verktøylista bærer**. Strengen `mcp__` forekommer i **0 av 24** Python-filer i pakken
(positiv kontroll: `create_sdk_mcp_server` blir funnet av samme spørring, så spørringen KAN
finne). Konstruksjonssiden namespacer altså ingenting: en `ToolCallRecorder` hengt der ser
`record_call`, ikke `mcp__tool_call_recorder__record_call`, og må få servernavnet fortalt.
Det er **samme pris som søskenet betaler**.
**Ærlig grense — hva målingen IKKE sier.** Den navnerommede formen `mcp__<server>__<tool>`
eksisterer: den bygges inne i CLI-en som følger med SDK-en (`_bundled/claude`, 178 literale
`mcp__`-forekomster, konstruksjonen på formen `` `mcp__${…}__${…}` ``). Det er **lest av
artefaktet, ikke observert i en kjøring hos oss** — å se den emittert krever en live
`query()`, som både D6-kostnadsregelen og suitens offline-invariant forbyr. Funnet er derfor
scopet til den sømmen vi faktisk kan bygge på: den in-process konstruksjonssiden. Prefikset
finnes på et lag vi bevisst ikke kjører, og et lag vi ikke kjører er ikke en søm vi kan feste
en recorder i.
**Pinnet av** `tests/test_sdk_tool_namespace_loadbearing.py` (4 tester). Value-beviset er
kjørt, ikke påstått: SDK-en ble mutert til å namespace ved konstruksjon (`"name":
f"mcp__{name}__{tool_def.name}"`) — **grønn før, 3 av 4 røde etter**, og den ene som forble
grønn er nettopp populasjons-kontrollen, som den skal. SDK-fila ble restaurert byte-identisk
(sha256 verifisert begge veier).
**Datering (D7-rammen):** dette er arbeid ETTER 2026-08-09 og skal **ikke** leses som
uavhengig konvergens selv om svaret er identisk med søskenets.
### (a)/(i) `unquote_scalar` — svaret er JA, den var ÉN regel, men pinnet bare i kanten (målt 2026-08-31)
Kandidaten står i køen fordi søskenet vokste en navngitt `unquote_scalar` etter at en
**duplisert konvertering hadde driftet** — to steder som fjernet anførselstegn fra en
frontmatter-skalar, etter to regler. Speilings-spørsmålet er derfor ikke «har vi funksjonen»,
men det søskenets defekt faktisk stiller: **er den ÉN regel hos oss, og er selve regelen
load-bearing?**
**Populasjonen først.** `unquote_scalar` finnes ikke hos oss: **0 treff av 76 undersøkte
`.py`-filer** i `src`+`tests` (positiv kontroll: samme spørring finner `parse_frontmatter` i
`okf.py`), og `unquote` finnes ikke i repoet utenfor `.venv`. Vår motpart er
`okf._strip_matching_quotes` — **én definisjon, ett kallsted** (`_parse_frontmatter_and_body`).
Søskenets drift-form finnes altså ikke her.
**Men regelen var dekket bare i kanten.** Med `scripts/mutation_harness.py`, nevner `tests/`
(hele suiten, 950 tester), hver kjøring sha256-restaurert:
- Å detache regelen helt (`return value[1:-1]` → `return value`) er **RØD** —
`test_okf.py::TestFrontmatter::test_unknown_fields_preserved_and_quotes_stripped` fanger den.
- Hver av de tre **klausulene** var **grønn-men-død**: å svekke lengdevakten (`>= 2` → `>= 1`),
å droppe matching-kravet (`value[0] == value[-1] and`), og å utvide quote-settet med en
symmetrisk delimiter (backtick) lot hele suiten stå grønn.
Samme klasse som `_STRUCTURE_MARKERS`-hullet i økt 32: sømmen var NAVNGITT og kant-dekket, som
ikke er det samme som dekket. **Pinnet av** `tests/test_okf_unquote_loadbearing.py` (5 tester,
950 → 955). Value-beviset er kjørt, ikke påstått: hver klausul-test er grønn før og rød etter
nøyaktig sin egen mutasjon, med populasjons-kontrollen grønn i begge (mutasjonen landet ikke
for bredt), og ledd 3 pinnet på linje via `--red-at`.
**EN MÅLETRAP, MÅLT — ny lærdom om harnesset.** Første forsøk på klausul 3 utvidet settet med
`'['`. Suiten forble grønn, hvilket leses som «klausulen er ikke dekket» — men mutasjonen er en
**no-op**: `'['` kan aldri tilfredsstille matching-kravet, siden `[` ikke er `]`. Flow-form-
verdier er altså beskyttet av **matching-klausulen, ikke av quote-settet**, og en grønn kjøring
under den mutasjonen var aldri bevis om quote-settet i det hele tatt. **Harnesset kan ikke
skille en oppførsels-bevarende mutasjon fra en udekket søm — begge kommer ut som «stayed
GREEN».** En mutasjon må vises å endre oppførsel før dens grønne leses som et hull. Dette er
Verifiseringsloven ansikt 4 anvendt på selve måleapparatet: et negativt resultat fra en
spørring som ikke KAN finne, er ikke null — det er ikke målt.
**Ærlig grense — hva dette IKKE sier.** Å pinne at flow-verdier passerer **urørt** er ikke
flow-DEKODING, og fila legger ingen til: `tags: [a, b]` forblir strengen `"[a, b]"`. Den
linje-orienterte parseren har ingen nesting-modell ved design (§1 ærlighets-regelen). Den
additive flow-dekoderen er søskenets B4-arbeid i `portfolio-optimiser` og er bevisst **ikke**
bygget her; testene pinner dagens grense slik at det arbeidet ikke kan lande stille på denne
siden.
**Datering (D7-rammen):** arbeid ETTER 2026-08-09 — skal **ikke** leses som uavhengig konvergens.
Rammen rundt køen: å lese søskenets kode er tillatt (`3bdf7f0`), men kopiering skal kun skje
der det tjener løsningen, aldri som snarvei. **Uavhengighets-beviset er DATERT** t.o.m.
2026-08-09; arbeid etter den datoen kan ikke leses som uavhengig konvergens.
## D-A-gater og fasit-berøring
- **D-A#5 mangler story-etikett oppstrøms** — avklar med MAF før teksten låses. En ny
hovedbok-kontrakt MÅ inn i §12s kryssjekktabell i SAMME amendment; ellers blir tabellen
ufullstendig i det øyeblikket kontrakten finnes.
- **D-A#2 rører fasiten.** commons er meldt at vår §12-vakt keyer på ordrett `| `generated` |`
⇒ et amendment som ERSTATTER raden gjør oss RØDE. Det er **by design**: en fasit-endring
skal koste en synlig rød test, ikke gli gjennom.
- **Gates:** D-F/D-G → K13 · D-B → K14/K15 · D-E · okf-toolkit-§8 · delbarhet av ledger-/
outbox-format. R-9 er valgfri.

32
llms.txt
View file

@ -1,32 +0,0 @@
# portfolio-optimiser-claude
> Claude Agent SDK sibling implementation of the portfolio-optimiser method: an open Python
> framework that finds cost savings *inside* each project in a portfolio. Agents propose
> measures, a mandatory deterministic validator gates every number before it leaves the
> system, domain experts judge via human-in-the-loop, and the loop learns from the verdicts.
> A technical framework only — the deployer owns DPIA, risk assessment, and legal basis.
Not published to a package index. This is the way in — from a clean clone, offline, without
an API key:
```bash
git clone https://git.fromaitochitta.com/open/portfolio-optimiser-claude.git
cd portfolio-optimiser-claude
uv sync
uv run pytest # the full suite, offline, no API key
```
## Docs
- [README.md](README.md): what the framework does, the seams it is built from, the operator CLI, and the one budgeted live run
- [shared/method-spec.md](shared/method-spec.md): the normative method specification (RFC 2119) both reference implementations build
- [shared/ingest-spec.md](shared/ingest-spec.md): the normative ingest specification — connectors and bundle materialization
- [CLAUDE.md](CLAUDE.md): the invariants of this repo — offline test suite, load-bearing tests, budget ceilings
- [CHANGELOG.md](CHANGELOG.md): released versions
## Optional
- [docs/extending.md](docs/extending.md): how the ingest layer is implemented here and how to extend it
- [docs/oppskrift-kunnskapsbase.md](docs/oppskrift-kunnskapsbase.md): recipe for building a knowledge base for a project
- [CONTRIBUTING.md](CONTRIBUTING.md): how to propose changes
- [SECURITY.md](SECURITY.md): how to report a vulnerability

385
scripts/mutation_harness.py
View file

@ -1,385 +0,0 @@
#!/usr/bin/env python3
"""Green-before / red-after, with the four silent controls made loud.
This repo does not accept "the test is load-bearing" as a claim. It mutates
what the test rests on, runs the test, and requires it to go RED -- then
restores the tree and proves the restore was byte-identical. That procedure was
hand-built ten times before it became this file, and the reason it became a
file is not the typing. It is that four of its controls fail SILENTLY:
1. AN ANCHOR THAT IS NOT UNIQUE mutates more than the seam, so the redness is
not attributable to the seam.
2. A MISTYPED NODE ID never runs the test. pytest exits 4, which is non-zero,
so a harness that asks "rc != 0?" reads a typo as a proof. (MEASURED
2026-08-25: green rc=0, real failure rc=1, unknown id or broken collection
rc=4, nothing collected rc=5. And `--collect-only -q` exits 0 on an id
that does not exist, so it is not an existence check either.)
3. AN UNVERIFIED RESTORE leaves the tree mutated and says "restored". Under
`.venv/` nothing is tracked, so `git status` will not catch it.
4. RED ON THE WRONG LINE -- a NameError or a broken import -- is achievable by
any garbage edit and proves nothing about the seam.
Each one turns a broken query into a positive-looking fact, which is the defect
class the operating model names. Prose cannot enforce them; this can.
THREE ROLES, AND THE ORDER FOLLOWS FROM WHAT EACH ONE MEASURES. The rule
"positive controls before negatives" was never in tension with this procedure;
it only looked that way because "the control" was read as one thing when it is
three:
* BEFORE, mandatory -- the measuring apparatus. Every node id is collected and
GREEN right now. Without this, "red after" may be red because the id was
always wrong. This IS the positive control, and it runs first.
* AFTER, by definition -- the measurement. "Can this test go red at all?" is
not answerable until the mutation exists. This is not a control that ran
late; it is the thing being measured.
* BOTH, must stay green -- the population/collateral control. If it reds, the
mutation landed wider than the seam and the target's redness is void.
SCOPE. This is for mutating the REAL tree and running pytest as a subprocess.
The other mutation class -- mutate a copy in memory, call the guard directly --
already lives in the suite as ``test_guard_red_when_*`` and needs no tool. This
one deliberately does NOT live in the suite: the suite runs every session, and a
fixture that writes to disk turns every interrupted run into a mutated tree.
Usage:
python scripts/mutation_harness.py \
--target .venv/lib/python3.x/site-packages/some/module.py \
--anchor 'the exact text, unique in the file' \
--replacement 'what detaches the seam' \
--red tests/test_x.py::TestY::test_the_seam \
--green tests/test_x.py::TestY::test_the_population
"""
from __future__ import annotations
import argparse
import hashlib
import re
import subprocess
import sys
from dataclasses import dataclass, field
from pathlib import Path
from typing import Literal, Mapping, Sequence
Verdict = Literal["green", "red", "not-collected"]
# MEASURED 2026-08-25, pytest 8, this repo's .venv. The point of the table is
# that 4 and 5 are NOT red -- they are the apparatus reporting that it never
# measured anything.
_GREEN = 0
_FAILED = 1
_USAGE_OR_COLLECTION_ERROR = 4
_NOTHING_COLLECTED = 5
# Reads the exception NAME off pytest's `E <dotted.Name>: <msg>` lines. Kept
# deliberately dumb: it reports what pytest emitted, and nothing downstream is
# allowed to gate on the value (see CONTROL 4). `E assert 1 == 2` has no
# colon-terminated name and correctly yields nothing.
_ERROR_LINE = re.compile(r"^E\s+([A-Za-z_][\w.]*)\s*:", re.MULTILINE)
class HarnessError(Exception):
"""Base for every refusal. A refusal is a result, never a silent pass."""
class AnchorNotUnique(HarnessError):
"""The anchor does not occur exactly once, so the mutation is not targeted."""
class NodeNotCollected(HarnessError):
"""pytest never ran the id. Non-zero here mimics red and must not be read as it."""
class NotGreenBefore(HarnessError):
"""A target was already red, so redness after is not caused by the mutation."""
class ControlWentRed(HarnessError):
"""A control reddened: the mutation landed wider than the seam."""
class NotAValueProof(HarnessError):
"""The mutation ran, and the seam did not care. That is a finding, not an error."""
class RestoreFailed(HarnessError):
"""The bytes on disk are not the bytes we started with. Loudest possible failure."""
@dataclass(frozen=True)
class Outcome:
node_id: str
returncode: int
verdict: Verdict
error_types: tuple[str, ...]
output: str = ""
@property
def summary(self) -> str:
kinds = ", ".join(self.error_types) if self.error_types else "no named exception"
return f"{self.verdict} (rc={self.returncode}, {kinds})"
@dataclass(frozen=True)
class Mutation:
target: Path
anchor: str
replacement: str
@dataclass(frozen=True)
class Report:
mutation: Mutation
before: tuple[Outcome, ...]
after: tuple[Outcome, ...]
sha256_before: str
sha256_after: str
targets: tuple[str, ...] = field(default_factory=tuple)
controls: tuple[str, ...] = field(default_factory=tuple)
pins: dict[str, str] = field(default_factory=dict)
@property
def after_by_id(self) -> dict[str, Outcome]:
return {o.node_id: o for o in self.after}
@property
def before_by_id(self) -> dict[str, Outcome]:
return {o.node_id: o for o in self.before}
@property
def value_proved(self) -> bool:
after = self.after_by_id
return bool(self.targets) and all(after[n].verdict == "red" for n in self.targets)
def classify(returncode: int, output: str, node_id: str = "") -> Outcome:
"""Map a pytest exit code to a verdict, keeping "never ran" out of "red".
The whole tool turns on this function. Everything else is bookkeeping.
"""
if returncode == _GREEN:
verdict: Verdict = "green"
elif returncode == _FAILED:
verdict = "red"
elif returncode in (_USAGE_OR_COLLECTION_ERROR, _NOTHING_COLLECTED):
verdict = "not-collected"
else:
# Unknown code: refuse to guess which side of the line it falls on.
verdict = "not-collected"
return Outcome(
node_id=node_id,
returncode=returncode,
verdict=verdict,
error_types=tuple(dict.fromkeys(m.rsplit(".", 1)[-1] for m in _ERROR_LINE.findall(output))),
output=output,
)
def run_node(node_id: str, cwd: Path) -> Outcome:
completed = subprocess.run(
[sys.executable, "-m", "pytest", node_id, "-q", "--no-header", "-p", "no:cacheprovider"],
cwd=cwd,
capture_output=True,
text=True,
)
return classify(completed.returncode, completed.stdout + completed.stderr, node_id)
def _sha256(data: bytes) -> str:
return hashlib.sha256(data).hexdigest()
def _verify_restore(target: Path, original: bytes, expected_sha: str) -> str:
"""Write the original bytes back and prove from DISK that they are there.
Reads the file again rather than trusting the write, because the failure
this exists to catch is precisely a write that did not fully land.
"""
target.write_bytes(original)
actual = _sha256(target.read_bytes())
if actual != expected_sha:
raise RestoreFailed(
f"restore of {target} is NOT byte-identical: expected {expected_sha}, got {actual}. "
"The tree is dirty -- fix it by hand before trusting anything else."
)
return actual
def prove(
mutation: Mutation,
expect_red: Sequence[str],
expect_green: Sequence[str] = (),
*,
red_at: Mapping[str, str] | None = None,
cwd: Path | None = None,
) -> Report:
"""Run the full procedure, refusing loudly at every control it fails."""
targets = tuple(expect_red)
controls = tuple(expect_green)
pins = dict(red_at or {})
if not targets:
raise ValueError(
"a proof with no --red target reds nothing and proves nothing; name at least one"
)
work_dir = Path(cwd) if cwd is not None else Path.cwd()
target_file = mutation.target
# CONTROL 1 -- the anchor, checked before a single byte is written.
original = target_file.read_bytes()
text = original.decode("utf-8")
occurrences = text.count(mutation.anchor)
if occurrences != 1:
raise AnchorNotUnique(
f"anchor must occur exactly once in {target_file}; measured {occurrences} "
f"occurrence(s). A wider anchor mutates more than the seam."
)
sha_before = _sha256(original)
node_ids = list(dict.fromkeys([*targets, *controls]))
# CONTROL 2 -- the measuring apparatus, and it runs FIRST.
before = tuple(run_node(n, work_dir) for n in node_ids)
for outcome in before:
if outcome.verdict == "not-collected":
raise NodeNotCollected(
f"pytest never ran {outcome.node_id} (rc={outcome.returncode}). "
"That exit is non-zero but it is not red -- the id is wrong, or "
"collection is broken. Nothing measured."
)
if outcome.verdict == "red":
raise NotGreenBefore(
f"{outcome.node_id} was ALREADY red before the mutation, so red after "
"is not caused by it. Fix the tree, then measure."
)
mutated = text.replace(mutation.anchor, mutation.replacement, 1)
try:
target_file.write_text(mutated, encoding="utf-8")
after = tuple(run_node(n, work_dir) for n in node_ids)
finally:
# CONTROL 3 -- the restore, verified from disk, in `finally` or not at all.
sha_after = _verify_restore(target_file, original, sha_before)
report = Report(
mutation=mutation,
before=before,
after=after,
sha256_before=sha_before,
sha256_after=sha_after,
targets=targets,
controls=controls,
pins=pins,
)
results = report.after_by_id
for node_id in controls:
if results[node_id].verdict != "green":
raise ControlWentRed(
f"control {node_id} went {results[node_id].verdict} under the mutation. "
"The mutation landed wider than the seam, so the target's redness "
"attributes to nothing."
)
for node_id in targets:
outcome = results[node_id]
if outcome.verdict == "not-collected":
raise NodeNotCollected(
f"after the mutation, pytest could not collect {node_id} "
f"(rc={outcome.returncode}). Collection broke -- which any garbage edit "
"achieves. That is the loudest way to prove nothing."
)
if outcome.verdict == "green":
raise NotAValueProof(
f"{node_id} stayed GREEN under the mutation. The seam it claims to guard "
"is detached and the test did not notice -- green-but-dead, measured."
)
# CONTROL 4 -- red, but WHERE? Not derivable from the exception type: a
# legitimately red test may die as AssertionError, as `Failed: DID NOT
# RAISE`, or as a custom exception. So the caller PINS the line the
# proof is about and this checks the pin against pytest's real output.
# Unpinned, the type is reported instead of silently blessed.
pin = pins.get(node_id)
if pin is not None and pin not in outcome.output:
raise NotAValueProof(
f"{node_id} went red ({outcome.summary}), but the failure output does "
f"not contain the pinned location {pin!r}. It died somewhere else than "
"the line this proof is about."
)
return report
def _format(report: Report) -> str:
lines = [
f"VALUE-PROVED {report.mutation.target}",
f" anchor {report.mutation.anchor!r} -> {report.mutation.replacement!r}",
f" restored sha256 {report.sha256_after} (byte-identical, verified from disk)",
]
for node_id in report.targets:
outcome = report.after_by_id[node_id]
pinned = f", pinned at {report.pins[node_id]!r}" if node_id in report.pins else ""
lines.append(f" TARGET green before / red after {node_id}")
lines.append(f" died of: {outcome.summary}{pinned}")
for node_id in report.controls:
lines.append(f" CONTROL green before / green after {node_id}")
return "\n".join(lines)
def main(argv: Sequence[str] | None = None) -> int:
parser = argparse.ArgumentParser(
prog="mutation_harness.py",
description="Prove a test is load-bearing: green before, red after, restore verified.",
)
parser.add_argument("--target", required=True, type=Path, help="file to mutate in place")
parser.add_argument("--anchor", required=True, help="exact text, must occur exactly once")
parser.add_argument("--replacement", required=True, help="what the anchor becomes")
parser.add_argument(
"--red",
action="append",
default=[],
metavar="NODE_ID",
help="test that must be green before and red after (repeatable)",
)
parser.add_argument(
"--green",
action="append",
default=[],
metavar="NODE_ID",
help="control that must stay green in both runs (repeatable)",
)
parser.add_argument(
"--red-at",
action="append",
default=[],
metavar="NODE_ID=TEXT",
help="require this text in that target's failure output (repeatable)",
)
parser.add_argument("--cwd", type=Path, default=None, help="directory to run pytest from")
args = parser.parse_args(argv)
try:
report = prove(
Mutation(target=args.target, anchor=args.anchor, replacement=args.replacement),
expect_red=args.red,
expect_green=args.green,
red_at=dict(pin.split("=", 1) for pin in args.red_at),
cwd=args.cwd,
)
except HarnessError as exc:
print(f"{type(exc).__name__}: {exc}")
if isinstance(exc, NotAValueProof):
print("NOT a value proof.")
return 1
except ValueError as exc:
print(f"ValueError: {exc}")
return 2
print(_format(report))
return 0
if __name__ == "__main__": # pragma: no cover
raise SystemExit(main())

4
src/portfolio_optimiser_claude/preflight.py
View file

@ -21,9 +21,7 @@ Four offline checks (for a given backend profile, ``anthropic`` today):
refusal naming ``uv sync``, never an ImportError), the run-path symbols exist,
and the bundled Claude Code CLI is present on disk — located offline via the
SDK package's own files, mirroring the SDK's ``_find_cli`` order (bundled
first, then a ``claude`` on PATH). Verified against installed 0.2.139 by
reading its ``_find_cli`` (bundled at line 250, PATH at 256) — the order is
unchanged from 0.2.120.
first, then a ``claude`` on PATH). Verified against installed 0.2.120.
* **termination** — the §8 stop contract constructs with positive round/token
caps and the per-call USD cap is positive (the budget belt is SET).

27
src/portfolio_optimiser_claude/sdk_client.py
View file

@ -9,18 +9,13 @@ turn, a first-class USD cap (``ClaudeAgentOptions.max_budget_usd``) ON TOP of
the §8 token/round meter that the loop already charges, and NO filesystem
settings (``setting_sources=[]``).
Verified against claude-agent-sdk 0.2.139 AT SOURCE LEVEL (2026-08-18; the
public ``query.py`` is byte-identical to 0.2.120 and every field below is
present with an unchanged type and default). Pinned ``>=0.2.111,<0.3``, but the
pin is only a permission — ``tests/test_sdk_version_guard.py`` carries the
proof: it checks each attribute below against the INSTALLED package and goes
red on any build newer than the last one actually read. The premises:
``query()`` yields ``AssistantMessage`` (text blocks + real model id) and a
closing ``ResultMessage`` (provider-reported ``usage`` + ``total_cost_usd``).
Unknown message types are ignored by the loop below, so an SDK that adds one
(0.2.139 added ``ConversationResetMessage``) does not disturb this client.
A reply without usage is passed through as ``None`` so the meter fails CLOSED
(§8) — this client never invents a count.
Verified against claude-agent-sdk 0.2.110 at source level and release notes
through 0.2.120 — pinned ``>=0.2.111,<0.3`` with a version guard
(``tests/test_sdk_version_guard.py``) that forces re-verification of these
premises before any widening: ``query()`` yields ``AssistantMessage`` (text
blocks + real model id) and a closing ``ResultMessage`` (provider-reported
``usage`` + ``total_cost_usd``). A reply without usage is passed through as
``None`` so the meter fails CLOSED (§8) — this client never invents a count.
"""
from __future__ import annotations
@ -52,16 +47,14 @@ _USAGE_TOKEN_FIELDS = (
def build_call_options(model_id: str, *, max_budget_usd: float) -> ClaudeAgentOptions:
"""One bounded, ISOLATED completion call (§8 + S10 post-mortem).
``setting_sources=[]`` is the SDK's documented isolation mode (re-read at
0.2.139): the spawned CLI loads NO filesystem settings — no
``setting_sources=[]`` is the SDK's documented isolation mode (verified
against 0.2.110): the spawned CLI loads NO filesystem settings — no
session hooks, no CLAUDE.md, no operator instructions. The default
(``None``) loads ALL sources: in the S10 live run that injected the
operator's config into every call (~10-15k uncached tokens each) and
mandated a confirmation preamble that made pure-JSON replies impossible.
``system_prompt=None`` serializes to an EMPTY system prompt, not the
Claude Code preset. 0.2.139 added a skills path that can DEFAULT
``setting_sources`` to ``["user", "project"]`` — it fires only on ``None``,
so the explicit ``[]`` here is not reachable by it (read at source).
Claude Code preset.
"""
return ClaudeAgentOptions(
model=model_id,

44
tests/test_cli_paritet_loadbearing.py
View file

@ -25,11 +25,9 @@ from __future__ import annotations
import argparse
import contextlib
import importlib
import importlib.util
import io
import json
import re
import sys
from pathlib import Path
from typing import Callable
@ -560,28 +558,6 @@ def _full_help(module_name: str) -> str:
return text
def _harness_help() -> str:
"""Capture ``scripts/mutation_harness.py --help``.
Loaded from its published path rather than as a package module, because
that path IS the contract: the harness is a standalone script by design
(it must not run inside the suite it mutates for), so the README's claim
about its flags can only be checked against the file an operator runs.
"""
path = Path(__file__).resolve().parents[1] / "scripts" / "mutation_harness.py"
spec = importlib.util.spec_from_file_location("_harness_for_parity", path)
assert spec is not None and spec.loader is not None
module = importlib.util.module_from_spec(spec)
# Register BEFORE exec: @dataclass resolves annotations through
# sys.modules[cls.__module__], which is None for an unregistered module.
sys.modules[spec.name] = module
spec.loader.exec_module(module)
buffer = io.StringIO()
with contextlib.redirect_stdout(buffer), contextlib.suppress(SystemExit):
module.main(["--help"])
return buffer.getvalue()
def _readme_documented_modules() -> list[str]:
return sorted(set(_MODULE.findall(README.read_text(encoding="utf-8"))))
@ -611,32 +587,14 @@ class TestReadmeClaimsMatchTheCli:
def test_every_documented_flag_exists_in_a_documented_cli(self) -> None:
# RED the moment the README claims a flag the code does not offer —
# the drift K12 exists to close, kept closed from here on.
available = "\n".join(
[*(_full_help(name) for name in _readme_documented_modules()), _harness_help()]
)
available = "\n".join(_full_help(name) for name in _readme_documented_modules())
assert "--bundle" in available, "help capture is broken — the grep would be vacuous"
assert "--anchor" in available, (
"harness help missing — its README flags would be unmeasured"
)
undelivered = sorted(flag for flag in _readme_documented_flags() if flag not in available)
assert undelivered == [], (
f"README documents flags no CLI offers: {undelivered} — "
"either wire them or stop claiming them (§1)"
)
def test_the_harness_help_is_captured_and_the_detector_still_fires(self) -> None:
# The README documents a second command-line surface -- the mutation
# harness -- and it is NOT a `portfolio_optimiser_claude.*` module, so
# the module walk above cannot see it. It is also not third-party
# tooling, so the foreign-marker skip would be a lie. Capturing its
# --help keeps its documented flags MEASURED rather than exempted.
help_text = _harness_help()
assert "--target" in help_text, "harness help capture is broken -- the grep is vacuous"
assert "--red-at" in help_text
# POSITIVE CONTROL: the same query must be able to MISS, or the
# inclusion above proves nothing about flags the harness lacks.
assert "--tomorrows-flag" not in help_text
def test_the_operator_surfaces_are_all_documented(self) -> None:
# The other direction, bounded to the flags K12 promises the operator
# can drive from the command line: the run entrance's collecting

436
tests/test_ingest_spec_loadbearing.py
View file

@ -10,65 +10,10 @@ Form mirrored from the sibling ``test_method_spec_loadbearing.py``: every predic
the spec TEXT as an argument, so the detach-proofs are tests in the suite rather than a
one-off spot-check that dies with the session. Red-proofs run against a mutated COPY of
the spec in ``tmp_path`` — never against ``shared/`` itself.
**Second half — structure markers (added after a measurement).** Everything above anchors
§12 plus two whole-document properties. Measured AT THE TIME (806-test suite, before the
§11 row anchors below existed): gutting ANY of §1–§11 to its heading alone left this guard
green for 11 of 11 sections — and measured wider, for 10 of the 11 nothing in the whole
suite went red either, so a commons pull could empty a normative section and no test would
notice. (§7 was the single overlap: the O2 ratchet reads its `generated` row.) That claim is
now STALE for §11 specifically — gutting §11 today also reds all 11 row anchors below, since
their tables live in the same body the marker phrase does. ``_STRUCTURE_MARKERS`` closes that
hole by
binding ONE verbatim normative phrase to the section that OWNS it, the form already proven in
``test_method_spec_rule_phrases_loadbearing.py``. Two properties make the binding real rather
than decorative: the phrase must appear in ITS OWN section body (presence anywhere is what a
whole-document check already cannot see), and it must be UNIQUE in the document (a phrase
occurring twice could be satisfied from a section other than its owner).
The SELECTION is ours, not mirrored: each phrase names a seam THIS repo actually carries and
cites the owning module, because a phrase anchoring a seam we do NOT implement would be
green-but-dead of the opposite kind — protecting spec text nothing here depends on. §7's
``generated`` row is deliberately NOT the chosen phrase: the O2 ratchet
(``test_ingest_stamp_conformance_loadbearing.py``) already anchors that row, so §7 is bound
here on a second, independent sentence.
Matching is whitespace- and emphasis-normalized so a reflow or a bolding change does not
false-red; the normalization has its own controls below.
**Third half — §11 row anchors (STATE pkt. 4, ORDER 20260818T155933Z).** ``_STRUCTURE_MARKERS``
binds ONE phrase per section, so it cannot see a row DROPPED from a section that keeps its
other prose intact — exactly the §12 gap ``_CONTRACT_FIELDS`` closes for the machine-readable
contract fields. §11 carries two of its own machine-readable tables, keyed on a plain-text
(not backtick) FIRST column, never covered by a row anchor: the golden extraction format
(4 rows) and the load-bearing conformance seams (9 rows). This section repeats the §12 form
— ``*_table`` slice, ``_undocumented_*`` row check, row-removed/row-renamed red-proofs,
fail-closed heading rename — for both.
The SELECTION is, again, ours — and here it is the hard part, because §11's 9 seam rows are
not machine-readable contract fields, they are CLAIMS that a test in this repo goes RED when
the row's seam detaches. A row anchored without such a test would be green-but-dead of the
opposite kind: protecting spec text nothing here depends on. Two of the nine are NOT anchored
because no test in this repo makes that claim true (see ``_LOAD_BEARING_SEAMS`` below for the
per-row citation, and the comment beside it for the two exclusions and why).
DENOMINATOR, measured on disk (not just detach-proved in-process): each of the 11 anchored
rows was individually deleted from the real ``shared/ingest-spec.md`` (git-restored after
every mutation, `git status` clean + sha256 unchanged before and after), and the full 923-test
suite run against the mutation. Every one of the 11 reds EXACTLY its own two new tests
(``test_spec_documents_*`` + ``test_guard_red_when_*_row_removed``) — nothing pre-existing
reds for any of them, so these anchors are the only thing standing there, not a second copy
of coverage that already existed.
KNOWN LIMITATION (inherited, not fixed here): the row match is exact-whitespace
(``f"| {row} |"``), like ``_CONTRACT_FIELDS`` above and unlike ``_STRUCTURE_MARKERS``'s
normalized match. A commons reflow that pads these two tables for column alignment would
false-red all 11 row anchors at once. Left as-is for consistency with the §12 form this
section mirrors; not a live problem today (measured against the current spec).
"""
from __future__ import annotations
import re
from pathlib import Path
import pytest
@ -252,384 +197,3 @@ def test_slice_guard_red_when_anchor_degenerates_to_whole_spec() -> None:
# what catches that, and that it is not itself green by accident.
defects = _slice_defects(SPEC.read_text(encoding="utf-8"))
assert defects, "the slice guard accepts the whole spec as the §12 table — it is dead"
# --- Structure markers: one normative phrase bound to the section that OWNS it -------
# section marker -> (verbatim normative phrase, the seam it anchors / owning module)
_STRUCTURE_MARKERS: dict[str, tuple[str, str]] = {
"## 1.": (
"data reaches the model ONLY via OKF bundles",
"bundle-only data path — context by navigation, never query-time retrieval (okf.py)",
),
"## 2.": (
"A bundle without `index.md` is an error",
"a missing root index is fatal, not skipped (okf.py navigate_bundle)",
),
"## 3.": (
"The promotion gate is the ONLY path into the verdict layer",
"verdict reservation — promote() is the sole verdict writer (promotion.py, ingest.py)",
),
"## 4.": (
"Credentials never live in the manifest",
"`connection_ref` is a NAME resolved from the environment at run time (ingest.py)",
),
"## 5.": (
"there is NO wall-clock default",
"`ingested_at` is a required argument of materialize() (ingest.py)",
),
"## 6.": (
"ALL other links — curated and promoted — are preserved verbatim",
"re-ingest layer safety — a promoted verdict's index link survives (promotion.py)",
),
"## 7.": (
"OKF consumers preserve unknown frontmatter fields",
"frontmatter parsing keeps unknown keys, so §7 rides through (okf.py)",
),
"## 8.": (
"the manifest cannot grant itself network access",
"the offline invariant at the consumer seam — no network opt-in is passed (ingest.py)",
),
"## 9.": (
"ingest NEVER runs automatically",
"no scheduler, no run-path call site — the manifest path is explicit (ingest.py)",
),
"## 10.": (
"given the same source content, manifest, and `ingested_at`, the result is byte-identical",
"the byte-exact golden extractions (examples/ingest-golden-*, test_ingest_golden.py)",
),
"## 11.": (
"Test suites MUST run without credentials and without network access",
"the suite-wide offline invariant (test_sdk_isolation.py, no key and no network)",
),
}
def _normalize(text: str) -> str:
"""Whitespace- and emphasis-insensitive: a reflow or a bolding change must not red."""
return re.sub(r"\s+", " ", text.replace("*", "")).strip()
def _section_body(text: str, marker: str) -> str:
"""Heading to the next section. RED (ValueError) if the section is renamed or dropped."""
start = text.index(marker)
end = text.find("\n## ", start + len(marker))
return text[start:] if end == -1 else text[start:end]
def _is_anchored(text: str, marker: str) -> bool:
phrase, _ = _STRUCTURE_MARKERS[marker]
return _normalize(phrase) in _normalize(_section_body(text, marker))
def _occurrences(text: str, marker: str) -> int:
phrase, _ = _STRUCTURE_MARKERS[marker]
return _normalize(text).count(_normalize(phrase))
def _body_emptied(text: str, marker: str) -> str:
"""The section, gutted to its heading alone — the pull this guard exists to catch."""
body = _section_body(text, marker)
heading = body.splitlines()[0]
return text.replace(body, heading + "\n\n")
@pytest.mark.parametrize("marker", list(_STRUCTURE_MARKERS))
def test_section_keeps_its_structure_marker(marker: str) -> None:
# RED when the owning section loses the normative content this repo builds on — which
# the §12-anchored half above cannot see, because §12 survives an emptied §1-§11.
phrase, seam = _STRUCTURE_MARKERS[marker]
assert _is_anchored(SPEC.read_text(encoding="utf-8"), marker), (
f"{marker} no longer carries its structure marker ({seam}): {phrase!r}"
)
@pytest.mark.parametrize("marker", list(_STRUCTURE_MARKERS))
def test_structure_marker_is_unique_in_the_spec(marker: str) -> None:
# A phrase occurring twice could be satisfied from a section other than its owner,
# reopening the hole quietly. Binding is only meaningful while the phrase is unique.
count = _occurrences(SPEC.read_text(encoding="utf-8"), marker)
assert count == 1, f"{marker}'s structure marker occurs {count} times — binding is ambiguous"
@pytest.mark.parametrize("marker", list(_STRUCTURE_MARKERS))
def test_guard_red_when_section_body_emptied(marker: str) -> None:
# The value-proof: GREEN before the mutation, RED after it, same mutation both ways.
text = SPEC.read_text(encoding="utf-8")
assert _is_anchored(text, marker) # positive control first
assert not _is_anchored(_body_emptied(text, marker), marker)
@pytest.mark.parametrize("marker", list(_STRUCTURE_MARKERS))
def test_the_older_half_is_blind_to_the_emptied_section(marker: str) -> None:
# The DENOMINATOR, machine-checked: 11 of 11 gutted sections leave the §12-anchored
# half fully green. Without this the value-proof above would only show that the new
# marker CAN red, not that it is the only thing standing there.
gutted = _body_emptied(SPEC.read_text(encoding="utf-8"), marker)
assert not _named_toolkits(gutted), "the toolkit rule reds here — this section is not blind"
assert not _undocumented_fields(gutted), "a §12 row went missing — not a clean measurement"
assert not _slice_defects(_cross_check_table(gutted)), "the §12 slice moved — not blind"
assert not _is_anchored(gutted, marker), "and the new marker is what does red"
@pytest.mark.parametrize("marker", list(_STRUCTURE_MARKERS))
def test_guard_red_when_marker_moves_to_another_section(marker: str) -> None:
# Presence is not anchoring. The phrase survives in the document — appended after §12 —
# and the owning section must still report unanchored.
text = SPEC.read_text(encoding="utf-8")
phrase, _ = _STRUCTURE_MARKERS[marker]
moved = _body_emptied(text, marker).rstrip("\n") + f"\n\n{phrase}\n"
assert _normalize(phrase) in _normalize(moved) # control: it IS still in the document
assert not _is_anchored(moved, marker)
@pytest.mark.parametrize("marker", list(_STRUCTURE_MARKERS))
def test_guard_red_when_section_heading_renamed(marker: str) -> None:
# Fail-closed, as the §12 slice already is: no heading means no anchor, and the slice
# must raise rather than silently fall back to a wider (green-but-dead) match.
renamed = SPEC.read_text(encoding="utf-8").replace(marker, marker.replace("## ", "## §"))
with pytest.raises(ValueError):
_section_body(renamed, marker)
@pytest.mark.parametrize("marker", list(_STRUCTURE_MARKERS))
def test_normalization_tolerates_reflow_and_emphasis(marker: str) -> None:
# A line-rewrap or a bolding change is not a contract change; it must not red.
text = SPEC.read_text(encoding="utf-8")
phrase, _ = _STRUCTURE_MARKERS[marker]
reflowed = text.replace(phrase, phrase.replace(" ", "\n ", 1))
assert _is_anchored(reflowed, marker), "a reflow false-reds the guard"
bolded = text.replace(phrase, f"**{phrase}**")
assert _is_anchored(bolded, marker), "an emphasis change false-reds the guard"
def test_normalization_is_not_so_loose_that_anything_matches() -> None:
# The tolerance above is only safe while a CHANGED WORD still reds. Without this,
# `_normalize` could degenerate into something that matches everything.
assert _normalize("a b\n c") == "a b c"
assert _normalize("**bold** text") == "bold text"
assert _normalize("MUST fail fast") != _normalize("MUST fail slowly")
# --- §11 row anchors: two tables, plain-text FIRST column, the §12 form repeated ------
_GOLDEN_FORMAT_HEADING = "**Golden extraction format:**"
_LOAD_BEARING_HEADING = "**Load-bearing conformance tests**"
# Golden extraction format (§11), entry -> the test that reads/writes it by this exact name.
_GOLDEN_FORMAT_ENTRIES: dict[str, str] = {
"manifest.json": "test_ingest_golden.py (GOLDEN / 'manifest.json', every test)",
"fixture/": "test_ingest_golden.py::TestMutationControl (case / 'fixture' / 'costs.csv')",
"ingested-at.txt": "test_ingest_golden.py::_ingested_at (the fixed §5 timestamp)",
"expected-bundle/": "test_ingest_golden.py::test_materializes_golden_byte_for_byte",
}
# Load-bearing conformance seams (§11), seam -> the test that goes RED when it detaches.
# Anchored: seven of nine — each cites a test in THIS repo that makes the row's claim true.
_LOAD_BEARING_SEAMS: dict[str, str] = {
"Provenance stamping": (
"test_ingest_loadbearing.py::TestProvenanceStamping"
"::test_generated_file_carries_the_provenance_layer"
),
"Navigability": (
"test_ingest_loadbearing.py::TestNavigability::test_generated_bundle_navigates_via_unchanged_okf"
),
"Verdict reservation": (
"test_ingest_loadbearing.py::TestVerdictReservation::test_verdict_okf_type_is_rejected"
),
"Re-ingest layer safety": (
"test_ingest_loadbearing.py::TestReingestLayerSafety"
"::test_promoted_verdict_and_curated_survive_reingest"
),
"Golden regression": "test_ingest_golden.py::test_materializes_golden_byte_for_byte",
# We carry this seam as a REFUSAL, not as an http connector (§1 makes http optional):
# the adapter never passes the network opt-in, so an http source is refused fail-fast.
"Network gate": (
"test_ingest_adoption.py::TestOfflineInvariant::test_http_source_is_refused_at_the_network_gate"
),
# Self-referential and deliberately so: this row's three clauses ARE the three guards
# already in this file.
"Spec integrity": (
"test_spec_is_present / test_spec_names_no_agent_toolkit / "
"test_spec_documents_contract_field (this file)"
),
}
# NOT anchored — no test in this repo makes the row's claim true; anchoring either would be
# green-but-dead, protecting spec text nothing here depends on:
#
# - "Stamp integrity (curated writers)": the reject-on-complete-stamp check is the §3
# verbatim-authoring-path's job. It lives in the llm-ingestion-okf LIBRARY, not in this
# repo's src/ — verified: the library's public surface (Extraction, FileSource, HttpSource,
# IngestError, IngestResult, Manifest, ManifestError, MaterializationError, NetworkGateError,
# RenderError, SourceError, SqlSource, connectors, errors, load_manifest, manifest,
# materialize, materialize_bundle, render) exposes no curated-writer / authoring primitive
# at all, and no test here forges the ownership stamp and asserts rejection. `inbox.py`'s
# "authoring primitive" is a different one (method-spec §5 verdict JSON, not OKF concept
# frontmatter) and does not touch this seam.
# - "Title link-safety": no test here supplies a title containing `[` or `]` and asserts
# fail-fast rejection. `test_ingest.py::TestManifestValidation::test_malformed_manifest_is_rejected`
# mutates title to `"two\nlines"` — a different rejection, not this one.
def _golden_format_table(text: str) -> str:
"""The golden-format table body — heading to the start of the next table."""
start = text.index(_GOLDEN_FORMAT_HEADING) # RED (ValueError) if the heading is renamed
end = text.index(_LOAD_BEARING_HEADING, start)
return text[start:end]
def _load_bearing_table(text: str) -> str:
"""The load-bearing table body — heading to end-of-spec or the next section."""
start = text.index(_LOAD_BEARING_HEADING) # RED (ValueError) if the heading is renamed
end = text.find("\n## ", start + len(_LOAD_BEARING_HEADING))
return text[start:] if end == -1 else text[start:end]
def _golden_slice_defects(table: str) -> list[str]:
defects = []
if not table.startswith(_GOLDEN_FORMAT_HEADING):
defects.append("does not start at the golden-format heading")
if _LOAD_BEARING_HEADING in table:
defects.append("leaked the load-bearing table")
if "\n## " in table:
defects.append("leaked past §11 into a following section")
return defects
def _load_bearing_slice_defects(table: str) -> list[str]:
defects = []
if not table.startswith(_LOAD_BEARING_HEADING):
defects.append("does not start at the load-bearing heading")
if _CROSS_CHECK_HEADING in table:
defects.append("leaked §12")
if "\n## " in table:
defects.append("leaked a following section")
return defects
def _undocumented_golden_entries(text: str) -> list[str]:
table = _golden_format_table(text)
return [entry for entry in _GOLDEN_FORMAT_ENTRIES if f"| `{entry}` |" not in table]
def _undocumented_loadbearing_seams(text: str) -> list[str]:
table = _load_bearing_table(text)
return [seam for seam in _LOAD_BEARING_SEAMS if f"| {seam} |" not in table]
def _golden_row_removed(text: str, entry: str) -> str:
table = _golden_format_table(text)
marker = f"| `{entry}` |"
kept = [ln for ln in table.splitlines(keepends=True) if not ln.startswith(marker)]
return text.replace(table, "".join(kept))
def _golden_row_renamed(text: str, entry: str) -> str:
table = _golden_format_table(text)
marker = f"| `{entry}` |"
return text.replace(table, table.replace(marker, f"| `{entry}_renamed` |"))
def _loadbearing_row_removed(text: str, seam: str) -> str:
table = _load_bearing_table(text)
marker = f"| {seam} |"
kept = [ln for ln in table.splitlines(keepends=True) if not ln.startswith(marker)]
return text.replace(table, "".join(kept))
def _loadbearing_row_renamed(text: str, seam: str) -> str:
table = _load_bearing_table(text)
marker = f"| {seam} |"
return text.replace(table, table.replace(marker, f"| {seam} (renamed) |"))
def test_golden_format_slice_is_a_slice_and_not_the_whole_spec() -> None:
defects = _golden_slice_defects(_golden_format_table(SPEC.read_text(encoding="utf-8")))
assert not defects, f"the golden-format anchor degenerated: {defects}"
def test_load_bearing_slice_is_a_slice_and_not_the_whole_spec() -> None:
defects = _load_bearing_slice_defects(_load_bearing_table(SPEC.read_text(encoding="utf-8")))
assert not defects, f"the load-bearing anchor degenerated: {defects}"
@pytest.mark.parametrize("entry", _GOLDEN_FORMAT_ENTRIES)
def test_spec_documents_golden_format_entry(entry: str) -> None:
# RED when an amendment drops or renames a golden-format row.
undocumented = _undocumented_golden_entries(SPEC.read_text(encoding="utf-8"))
assert entry not in undocumented, f"golden-format entry {entry!r} is no longer a row in §11"
@pytest.mark.parametrize("seam", _LOAD_BEARING_SEAMS)
def test_spec_documents_loadbearing_seam(seam: str) -> None:
# RED when an amendment drops or renames an anchored load-bearing row.
undocumented = _undocumented_loadbearing_seams(SPEC.read_text(encoding="utf-8"))
assert seam not in undocumented, f"load-bearing seam {seam!r} is no longer a row in §11"
@pytest.mark.parametrize("entry", _GOLDEN_FORMAT_ENTRIES)
def test_guard_red_when_golden_row_removed(tmp_path: Path, entry: str) -> None:
text = SPEC.read_text(encoding="utf-8")
assert entry not in _undocumented_golden_entries(text) # positive control first
copy = tmp_path / "ingest-spec.md"
copy.write_text(_golden_row_removed(text, entry), encoding="utf-8")
assert entry in _undocumented_golden_entries(copy.read_text(encoding="utf-8"))
@pytest.mark.parametrize("entry", _GOLDEN_FORMAT_ENTRIES)
def test_guard_red_when_golden_row_renamed(tmp_path: Path, entry: str) -> None:
# A detach-proof is not a value-proof — the row still exists, under another name.
copy = tmp_path / "ingest-spec.md"
copy.write_text(_golden_row_renamed(SPEC.read_text(encoding="utf-8"), entry), encoding="utf-8")
assert entry in _undocumented_golden_entries(copy.read_text(encoding="utf-8"))
@pytest.mark.parametrize("seam", _LOAD_BEARING_SEAMS)
def test_guard_red_when_loadbearing_row_removed(tmp_path: Path, seam: str) -> None:
text = SPEC.read_text(encoding="utf-8")
assert seam not in _undocumented_loadbearing_seams(text) # positive control first
copy = tmp_path / "ingest-spec.md"
copy.write_text(_loadbearing_row_removed(text, seam), encoding="utf-8")
assert seam in _undocumented_loadbearing_seams(copy.read_text(encoding="utf-8"))
@pytest.mark.parametrize("seam", _LOAD_BEARING_SEAMS)
def test_guard_red_when_loadbearing_row_renamed(tmp_path: Path, seam: str) -> None:
copy = tmp_path / "ingest-spec.md"
copy.write_text(
_loadbearing_row_renamed(SPEC.read_text(encoding="utf-8"), seam), encoding="utf-8"
)
assert seam in _undocumented_loadbearing_seams(copy.read_text(encoding="utf-8"))
def test_guard_red_when_golden_format_heading_renamed(tmp_path: Path) -> None:
# Fail-closed: no heading means no anchor, and the slice must raise, not silently fall
# back to a wider (green-but-dead) match.
mutated = SPEC.read_text(encoding="utf-8").replace(
_GOLDEN_FORMAT_HEADING, "**Golden fixture format:**"
)
copy = tmp_path / "ingest-spec.md"
copy.write_text(mutated, encoding="utf-8")
with pytest.raises(ValueError):
_golden_format_table(copy.read_text(encoding="utf-8"))
def test_guard_red_when_load_bearing_heading_renamed(tmp_path: Path) -> None:
mutated = SPEC.read_text(encoding="utf-8").replace(
_LOAD_BEARING_HEADING, "**Load-bearing regression tests**"
)
copy = tmp_path / "ingest-spec.md"
copy.write_text(mutated, encoding="utf-8")
with pytest.raises(ValueError):
_load_bearing_table(copy.read_text(encoding="utf-8"))
def test_golden_slice_guard_red_when_anchor_degenerates_to_whole_spec() -> None:
defects = _golden_slice_defects(SPEC.read_text(encoding="utf-8"))
assert defects, "the golden-format slice guard accepts the whole spec — it is dead"
def test_load_bearing_slice_guard_red_when_anchor_degenerates_to_whole_spec() -> None:
defects = _load_bearing_slice_defects(SPEC.read_text(encoding="utf-8"))
assert defects, "the load-bearing slice guard accepts the whole spec — it is dead"

444
tests/test_mutation_harness.py
View file

@ -1,444 +0,0 @@
"""The green-before / red-after procedure, made a tool instead of a memory.
WHY THIS EXISTS. This repo proves a test is load-bearing by mutating what it
rests on and watching it go red. That procedure has been hand-built ten times
(2x oekt 23, 4x oekt 31, 2x oekt 32, 1x oekt 33, 1x oekt 37) and rewritten from
memory in a scratchpad every time. Four of its controls fail SILENTLY, which is
exactly the class STATE's operating model calls oensket 4 of the verification
law -- a negative result from a broken query, consumed as a positive fact:
1. the anchor is not unique, so the mutation lands somewhere else too;
2. the node id was mistyped, so pytest never ran the test and the non-zero
exit MIMICS red (MEASURED 2026-08-25: rc=4, not rc=1 -- distinguishable,
but only if something looks at the number);
3. the restore is not verified, so the tree keeps the mutation;
4. the test goes red on a NameError or a broken import rather than on the
assertion the proof is about (oekt 33: "not on a NameError, not on the
control").
Prose cannot enforce any of those. That is the whole argument for a tool.
WHAT THIS IS NOT. The other mutation class -- mutate a COPY held in memory and
call the guard directly -- already has a home in this suite and needs no tool
(``test_guard_red_when_*``, five files). This tool is for the class that mutates
the REAL tree and runs pytest as a subprocess, which is why it must NOT live
inside the suite as a fixture: the suite runs every session, and a fixture that
writes to disk makes every interrupted run a mutated tree.
So every test below runs against a SANDBOX built in ``tmp_path``. Nothing here
mutates this repo. That is a property of the design, not a precaution.
"""
from __future__ import annotations
import hashlib
import importlib.util
import sys
from pathlib import Path
from types import ModuleType
import pytest
HARNESS_PATH = Path(__file__).resolve().parents[1] / "scripts" / "mutation_harness.py"
def _load_harness() -> ModuleType:
"""Import the tool from its published path, not from a package alias.
The path is part of the contract: the order asked for a standalone script,
and a test that imported it through some other name would stay green if the
script moved out from under the operator running it by hand.
"""
assert HARNESS_PATH.is_file(), f"the harness is not at its documented path: {HARNESS_PATH}"
spec = importlib.util.spec_from_file_location("mutation_harness", HARNESS_PATH)
assert spec is not None and spec.loader is not None
module = importlib.util.module_from_spec(spec)
sys.modules["mutation_harness"] = module
spec.loader.exec_module(module)
return module
mh = _load_harness()
# --------------------------------------------------------------------------
# The sandbox: a seam, a test that rests on it, and a control that does not.
# --------------------------------------------------------------------------
SEAM_SOURCE = '''\
"""A stand-in seam with one uniquely anchored behaviour."""
PREFIX = "mcp__"
def qualify(server: str, tool: str) -> str:
return f"{PREFIX}{server}__{tool}"
def population() -> list[str]:
return ["record_call", "flush_calls"]
'''
SEAM_TESTS = """\
from seam import population, qualify
def test_the_population_is_readable_and_is_two() -> None:
# CONTROL: green before AND after. If a mutation reds this, it landed wider
# than the seam and the target's redness proves nothing.
assert len(population()) == 2
def test_the_name_carries_the_server(){ANNOT} -> None:
# TARGET: green before, red after.
assert qualify("recorder", "record_call") == "mcp__recorder__record_call"
""".replace("{ANNOT}", "")
TARGET_ID = "test_seam.py::test_the_name_carries_the_server"
CONTROL_ID = "test_seam.py::test_the_population_is_readable_and_is_two"
@pytest.fixture()
def sandbox(tmp_path: Path) -> Path:
(tmp_path / "seam.py").write_text(SEAM_SOURCE, encoding="utf-8")
(tmp_path / "test_seam.py").write_text(SEAM_TESTS, encoding="utf-8")
return tmp_path
def _sha256(path: Path) -> str:
return hashlib.sha256(path.read_bytes()).hexdigest()
# --------------------------------------------------------------------------
# 1. Classification: the layer where oensket 4 actually bites.
# --------------------------------------------------------------------------
class TestExitCodesAreClassifiedNotJustTestedForZero:
def test_the_four_measured_exit_codes_map_to_three_distinct_verdicts(self) -> None:
# MEASURED 2026-08-25 against pytest 8 in this repo's .venv. Kept as a
# table so the mapping reads as the measurement, not as a belief.
assert mh.classify(0, "1 passed").verdict == "green"
assert mh.classify(1, "E AssertionError: nope").verdict == "red"
assert mh.classify(4, "no tests ran").verdict == "not-collected"
assert mh.classify(5, "no tests collected").verdict == "not-collected"
def test_a_mistyped_node_id_is_never_read_as_red(self) -> None:
# THE control this tool exists for. rc=4 is non-zero, so any harness
# that asks "rc != 0?" calls a typo a proof. This one cannot.
outcome = mh.classify(4, "no tests ran in 0.59s")
assert outcome.verdict == "not-collected"
assert outcome.verdict != "red"
def test_the_error_type_is_read_off_the_output_not_assumed(self) -> None:
assert mh.classify(1, "E AssertionError: x").error_types == ("AssertionError",)
assert mh.classify(1, "E NameError: name 'q' is not defined").error_types == (
"NameError",
)
def test_dotted_and_non_error_suffixed_failures_are_read_too(self) -> None:
# MEASURED, and it cost a rewrite. The first version of this tool gated
# on "is it an AssertionError?" and REJECTED three of the four real
# proofs run against itself, because a legitimately red test can die as
# `Failed: DID NOT RAISE`, or as a dotted custom exception. A gate that
# refuses real evidence gets switched off, which is worse than none --
# so the type is REPORTED, never used to model what pytest may emit
# (oekt 28: the fixture is the tool's emission, not a model of it).
assert mh.classify(1, "E Failed: DID NOT RAISE").error_types == ("Failed",)
assert mh.classify(1, "E mutation_harness.NotAValueProof: stayed green").error_types == (
"NotAValueProof",
)
assert mh.classify(1, "E assert 1 == 2").error_types == ()
# --------------------------------------------------------------------------
# 2. The anchor, checked before anything is written.
# --------------------------------------------------------------------------
class TestTheAnchorIsProvenUniqueBeforeTheFileIsTouched:
def test_an_ambiguous_anchor_is_refused_and_the_file_is_untouched(self, sandbox: Path) -> None:
target = sandbox / "seam.py"
before = _sha256(target)
with pytest.raises(mh.AnchorNotUnique) as excinfo:
mh.prove(
mh.Mutation(target=target, anchor="server", replacement="srv"),
expect_red=[TARGET_ID],
cwd=sandbox,
)
assert "3" in str(excinfo.value) or "occurrence" in str(excinfo.value).lower()
assert _sha256(target) == before, "refusal must happen BEFORE the write"
def test_an_absent_anchor_is_refused(self, sandbox: Path) -> None:
target = sandbox / "seam.py"
before = _sha256(target)
with pytest.raises(mh.AnchorNotUnique):
mh.prove(
mh.Mutation(target=target, anchor="not in this file", replacement="x"),
expect_red=[TARGET_ID],
cwd=sandbox,
)
assert _sha256(target) == before
def test_a_proof_with_no_target_is_vacuous_and_refused(self, sandbox: Path) -> None:
# A harness run that reds nothing proves nothing, and would otherwise
# report success. Refuse it rather than emit an empty proof.
with pytest.raises(ValueError):
mh.prove(
mh.Mutation(target=sandbox / "seam.py", anchor='PREFIX = "mcp__"', replacement=""),
expect_red=[],
cwd=sandbox,
)
# --------------------------------------------------------------------------
# 3. The value proof of the tool itself: it catches a known-red mutation and
# refuses to call a known-harmless one a proof.
# --------------------------------------------------------------------------
DETACHING = ('PREFIX = "mcp__"', 'PREFIX = ""')
HARMLESS = ("A stand-in seam", "A stand-in seam (comment touched)")
class TestTheToolIsValueProvedInBothDirections:
def test_a_detaching_mutation_is_caught_and_reported_as_a_value_proof(
self, sandbox: Path
) -> None:
target = sandbox / "seam.py"
report = mh.prove(
mh.Mutation(target=target, anchor=DETACHING[0], replacement=DETACHING[1]),
expect_red=[TARGET_ID],
expect_green=[CONTROL_ID],
cwd=sandbox,
)
assert report.value_proved
assert all(o.verdict == "green" for o in report.before)
assert report.after_by_id[TARGET_ID].verdict == "red"
assert report.after_by_id[TARGET_ID].error_types == ("AssertionError",)
assert report.after_by_id[CONTROL_ID].verdict == "green"
def test_a_harmless_mutation_is_not_flagged_as_a_proof(self, sandbox: Path) -> None:
# The other half of the value proof, and the one that is easy to skip:
# a tool that reports success on everything reports nothing.
target = sandbox / "seam.py"
with pytest.raises(mh.NotAValueProof) as excinfo:
mh.prove(
mh.Mutation(target=target, anchor=HARMLESS[0], replacement=HARMLESS[1]),
expect_red=[TARGET_ID],
expect_green=[CONTROL_ID],
cwd=sandbox,
)
assert TARGET_ID in str(excinfo.value)
assert "green" in str(excinfo.value).lower()
def test_the_target_must_be_green_before_or_it_is_not_a_measurement(
self, sandbox: Path
) -> None:
# POSITIVE CONTROL OF THE MEASURING APPARATUS, and it runs FIRST. If the
# target is already red, "red after" is not caused by the mutation.
(sandbox / "test_seam.py").write_text(
SEAM_TESTS.replace('== "mcp__recorder__record_call"', '== "already wrong"'),
encoding="utf-8",
)
with pytest.raises(mh.NotGreenBefore):
mh.prove(
mh.Mutation(
target=sandbox / "seam.py", anchor=DETACHING[0], replacement=DETACHING[1]
),
expect_red=[TARGET_ID],
cwd=sandbox,
)
def test_a_mistyped_node_id_raises_instead_of_proving_anything(self, sandbox: Path) -> None:
with pytest.raises(mh.NodeNotCollected) as excinfo:
mh.prove(
mh.Mutation(
target=sandbox / "seam.py", anchor=DETACHING[0], replacement=DETACHING[1]
),
expect_red=["test_seam.py::test_this_id_does_not_exist"],
cwd=sandbox,
)
assert "test_this_id_does_not_exist" in str(excinfo.value)
def test_a_mutation_that_breaks_the_import_is_not_a_value_proof(self, sandbox: Path) -> None:
# rc=4 AFTER the mutation means collection broke, which any garbage
# edit achieves. It is the loudest possible way to prove nothing.
with pytest.raises(mh.NodeNotCollected):
mh.prove(
mh.Mutation(
target=sandbox / "seam.py",
anchor='PREFIX = "mcp__"',
replacement="import definitely_not_a_module",
),
expect_red=[TARGET_ID],
cwd=sandbox,
)
def test_a_pinned_red_location_is_verified_against_the_real_output(self, sandbox: Path) -> None:
# oekt 33 said the redness must land on the line the proof is about
# ("not on a NameError, not on the control"). That is not derivable
# from the exception type -- it is derivable from WHERE it died. So the
# caller pins it and the tool checks the pin against pytest's actual
# output, rather than guessing from a type it does not control.
report = mh.prove(
mh.Mutation(target=sandbox / "seam.py", anchor=DETACHING[0], replacement=DETACHING[1]),
expect_red=[TARGET_ID],
red_at={TARGET_ID: "mcp__recorder__record_call"},
cwd=sandbox,
)
assert report.value_proved
def test_a_red_that_misses_the_pinned_location_is_refused(self, sandbox: Path) -> None:
with pytest.raises(mh.NotAValueProof) as excinfo:
mh.prove(
mh.Mutation(
target=sandbox / "seam.py", anchor=DETACHING[0], replacement=DETACHING[1]
),
expect_red=[TARGET_ID],
red_at={TARGET_ID: "a line this failure never prints"},
cwd=sandbox,
)
assert "a line this failure never prints" in str(excinfo.value)
def test_a_red_on_a_broken_module_is_still_reported_with_its_type(self, sandbox: Path) -> None:
# The honest limit, stated rather than papered over: a NameError inside
# a test that RAN is red, and the tool does not silently bless or
# reject it -- it names the type so the operator can see what died.
# (A mutation that breaks COLLECTION is a different case and is refused
# outright above: rc=4 is not red.)
report = mh.prove(
mh.Mutation(
target=sandbox / "seam.py",
anchor=' return f"{PREFIX}{server}__{tool}"',
replacement=" return undefined_name",
),
expect_red=[TARGET_ID],
cwd=sandbox,
)
assert report.after_by_id[TARGET_ID].error_types == ("NameError",)
assert "NameError" in mh._format(report)
def test_a_control_that_goes_red_fails_the_proof(self, sandbox: Path) -> None:
# The mutation lands wider than the seam: the population control reds
# too, so the target's redness is not attributable to the seam.
with pytest.raises(mh.ControlWentRed) as excinfo:
mh.prove(
mh.Mutation(
target=sandbox / "seam.py",
anchor='return ["record_call", "flush_calls"]',
replacement='return ["record_call"]',
),
expect_red=[CONTROL_ID],
expect_green=[CONTROL_ID],
cwd=sandbox,
)
assert CONTROL_ID in str(excinfo.value)
# --------------------------------------------------------------------------
# 4. The restore, verified in both directions, git or no git.
# --------------------------------------------------------------------------
class TestTheRestoreIsVerifiedNotAssumed:
def test_the_target_is_restored_byte_identical_after_a_successful_proof(
self, sandbox: Path
) -> None:
target = sandbox / "seam.py"
before = _sha256(target)
report = mh.prove(
mh.Mutation(target=target, anchor=DETACHING[0], replacement=DETACHING[1]),
expect_red=[TARGET_ID],
cwd=sandbox,
)
assert _sha256(target) == before
assert report.sha256_before == before == report.sha256_after
def test_the_target_is_restored_even_when_the_proof_raises(self, sandbox: Path) -> None:
# The restore lives in `finally` or it does not exist. Every failure
# path above would otherwise leave the tree mutated.
target = sandbox / "seam.py"
before = _sha256(target)
with pytest.raises(mh.NotAValueProof):
mh.prove(
mh.Mutation(target=target, anchor=HARMLESS[0], replacement=HARMLESS[1]),
expect_red=[TARGET_ID],
cwd=sandbox,
)
assert _sha256(target) == before, "a failed proof must not leave a mutated tree"
def test_the_restore_does_not_go_through_git(self, sandbox: Path) -> None:
# The .venv case (oekt 37): the target is not tracked, so `git status`
# would never catch the mutation and `git checkout --` cannot undo it.
# The harness holds the original bytes itself, which is why the same
# mechanism covers tracked and untracked targets alike.
assert not (sandbox / ".git").exists()
target = sandbox / "seam.py"
before = _sha256(target)
mh.prove(
mh.Mutation(target=target, anchor=DETACHING[0], replacement=DETACHING[1]),
expect_red=[TARGET_ID],
cwd=sandbox,
)
assert _sha256(target) == before
def test_a_corrupted_restore_is_reported_rather_than_passed_over(self, sandbox: Path) -> None:
# Force the one failure the sha256 check exists to catch. Without it,
# a partial write leaves a silently wrong tree and the run still says
# "restored".
target = sandbox / "seam.py"
with pytest.raises(mh.RestoreFailed):
mh._verify_restore(target, b"not what was written", "0" * 64)
# --------------------------------------------------------------------------
# 5. The command line, wired to the same code path the tests prove.
# --------------------------------------------------------------------------
class TestTheCommandLineReachesTheProvenCodePath:
def test_a_successful_proof_exits_zero_and_says_what_it_measured(
self, sandbox: Path, capsys: pytest.CaptureFixture[str]
) -> None:
code = mh.main(
[
"--target",
str(sandbox / "seam.py"),
"--anchor",
DETACHING[0],
"--replacement",
DETACHING[1],
"--red",
TARGET_ID,
"--green",
CONTROL_ID,
"--cwd",
str(sandbox),
]
)
out = capsys.readouterr().out
assert code == 0
assert "VALUE-PROVED" in out
assert TARGET_ID in out
assert _sha256(sandbox / "seam.py") in out, "the verified restore is part of the report"
def test_a_failed_proof_exits_non_zero_and_names_the_reason(
self, sandbox: Path, capsys: pytest.CaptureFixture[str]
) -> None:
code = mh.main(
[
"--target",
str(sandbox / "seam.py"),
"--anchor",
HARMLESS[0],
"--replacement",
HARMLESS[1],
"--red",
TARGET_ID,
"--cwd",
str(sandbox),
]
)
out = capsys.readouterr().out
assert code != 0
assert "NOT a value proof" in out or "NotAValueProof" in out

140
tests/test_okf_unquote_loadbearing.py
View file

@ -1,140 +0,0 @@
"""(a)/(i) `unquote_scalar`: is our one unquoting rule pinned, or pinned only at the edge?
The D7 mirroring queue (``docs/2026-08-18-funn-koeer-og-gater.md § D7-speilingskøen``)
carries this candidate because the MAF sibling grew a named ``unquote_scalar`` after
a DUPLICATED conversion had drifted — two places unquoting a frontmatter scalar, by
two rules. Our counterpart is ``okf._strip_matching_quotes``: one definition, one
call site (``_parse_frontmatter_and_body``). The mirroring question is therefore not
"do we have the function" but the one the sibling's defect actually poses — **is it
the ONE rule, and is the rule itself load-bearing in each of its three clauses?**
MEASURED 2026-08-31 with ``scripts/mutation_harness.py``, denominator ``tests/``
(the whole suite, 950 tests), each run restored sha256-verified:
- Detaching the rule outright (``return value[1:-1]`` -> ``return value``) is RED —
``test_okf.py::TestFrontmatter::test_unknown_fields_preserved_and_quotes_stripped``
catches it. The seam has an edge guard.
- Each of the three CLAUSES is GREEN-BUT-DEAD. Weakening the length guard
(``>= 2`` -> ``>= 1``), dropping the matching requirement
(``value[0] == value[-1] and``), and widening the quote set with a SYMMETRIC
delimiter (adding a backtick) each left the pre-existing suite green. Nothing
noticed.
So the answer mirrors økt 32's ``_STRUCTURE_MARKERS`` hole, not B4's: the seam was
NAMED and edge-covered, which is not the same as covered. This file closes the three
clauses. Each test below was proved RED under exactly the mutation it names.
WHY THIS CANDIDATE, NOW. ``llm-ingestion-okf`` pins its emission to FLOW form, so a
frontmatter value may legally arrive as ``[a, b]`` or ``{k: v}``. Which clause
actually protects those turned out NOT to be the one reached for first — see below.
A MEASURING TRAP, MEASURED. The first attempt at the third clause widened the set
with ``'['``. The suite stayed green, which reads as "clause not covered" — but the
mutation is a NO-OP: ``'['`` can never satisfy the matching clause, since ``[`` is
not ``]``. So flow-form values are protected by the MATCHING clause, not by the
quote set, and a green run under that mutation was never evidence about the quote
set at all. The harness cannot tell a behaviour-preserving mutation from an
undetected seam — both surface as "stayed GREEN". A mutation must be shown to
change behaviour before its green is read as a hole; the backtick, being
symmetric, does change it and is what the clause is pinned with below.
HONEST LIMIT — what this does NOT say. Pinning that flow values pass through
UNTOUCHED is not flow DECODING, and this file does not add any: ``tags: [a, b]``
stays the string ``"[a, b]"``. The line-oriented parser has no nesting model by
design (``_parse_frontmatter_and_body``'s own docstring, §1 honesty rule). An
additive flow decoder is the sibling's B4 work in ``portfolio-optimiser`` and is
deliberately NOT built here; these tests pin today's boundary so that work cannot
land silently on this side.
Dated under the D7 frame: this is work AFTER 2026-08-09 and must NOT be read as
independent convergence with the sibling.
"""
from __future__ import annotations
import ast
from pathlib import Path
from portfolio_optimiser_claude.okf import parse_concept_file
SRC_PKG = Path(__file__).resolve().parents[1] / "src" / "portfolio_optimiser_claude"
OKF_SOURCE = SRC_PKG / "okf.py"
RULE_NAME = "_strip_matching_quotes"
def _concept_with(tmp_path: Path, value: str) -> dict[str, str]:
"""Parse ``title: <value>`` through the PUBLIC path and hand back the frontmatter.
Through ``parse_concept_file``, never by calling the private rule directly: a
layer you do not run is not a seam (økt 37). What is asserted below is what the
parser actually hands downstream.
"""
path = tmp_path / "a.md"
path.write_text(f"---\ntype: reference\ntitle: {value}\n---\nBody.\n", encoding="utf-8")
return parse_concept_file(path).frontmatter
class TestTheUnquotingRuleIsOneRule:
"""The population control — and it runs first, because every clause below is
worth only as much as the claim that there is exactly ONE place to pin."""
def test_exactly_one_definition_and_one_call_site(self) -> None:
# This is the sibling's defect made measurable on our side: their duplicate
# drifted. RED the day a second unquoting rule appears, or the single call
# site becomes two. Read from the AST, not from a grep of prose.
tree = ast.parse(OKF_SOURCE.read_text(encoding="utf-8"))
definitions = [
n for n in ast.walk(tree) if isinstance(n, ast.FunctionDef) and n.name == RULE_NAME
]
call_sites = [
n
for n in ast.walk(tree)
if isinstance(n, ast.Call) and isinstance(n.func, ast.Name) and n.func.id == RULE_NAME
]
assert len(definitions) == 1, f"the ONE unquoting rule is now {len(definitions)}"
assert len(call_sites) == 1, (
f"{RULE_NAME} gained a second call site — the sibling's drift shape: "
f"lines {[c.lineno for c in call_sites]}"
)
def test_the_rule_still_strips_the_ordinary_case(self, tmp_path: Path) -> None:
# POSITIVE CONTROL FOR EVERY NEGATIVE BELOW. Each clause test asserts that
# some value comes through UNSTRIPPED; that is vacuous unless stripping
# demonstrably happens at all on this path.
frontmatter = _concept_with(tmp_path, '"Quoted title"')
assert frontmatter["title"] == "Quoted title"
class TestEachClauseOfTheRuleIsLoadBearing:
"""One test per clause of ``len(value) >= 2 and value[0] == value[-1] and
value[0] in {'"', "'"}`` — each proved RED under its own mutation."""
def test_length_guard_protects_a_single_quote_character(self, tmp_path: Path) -> None:
# RED under `len(value) >= 2` -> `len(value) >= 1`.
# A lone quote is its own first AND last character, so without the length
# guard it satisfies the other two clauses and `value[1:-1]` erases it.
# A malformed scalar must survive as itself — OKF robustness: tolerate,
# never silently rewrite.
assert _concept_with(tmp_path, '"')["title"] == '"'
def test_matching_requirement_protects_an_unterminated_quote(self, tmp_path: Path) -> None:
# RED under dropping `value[0] == value[-1] and`.
# Without it, an unterminated `"Oslo` is stripped from BOTH ends and
# arrives as `Osl` — a silent corruption strictly worse than the
# malformed input it came from.
assert _concept_with(tmp_path, '"Oslo')["title"] == '"Oslo'
assert _concept_with(tmp_path, "Oslo'")["title"] == "Oslo'"
def test_the_quote_set_is_exactly_two_characters(self, tmp_path: Path) -> None:
# RED under widening the set with a SYMMETRIC delimiter:
# `{'"', "'"}` -> `{'"', "'", '`'}`. A backtick pair is not a YAML quote,
# and the set is closed at two.
assert _concept_with(tmp_path, "`code`")["title"] == "`code`"
# Flow-form values (llm-ingestion-okf pins its emission to flow form) must
# ride through navigation UNTOUCHED (ingest-spec §7). These are guarded by
# the MATCHING clause above, not by the quote set — `[` is not `]` — and
# they are asserted here because that is where a flow decoder would first
# be reached for. Stripping the delimiters would hand downstream a string
# that is neither the flow value nor its decoding.
assert _concept_with(tmp_path, "[a, b]")["title"] == "[a, b]"
assert _concept_with(tmp_path, "{k: v}")["title"] == "{k: v}"

2
tests/test_preflight.py
View file

@ -124,7 +124,7 @@ class TestPlaceholderDetector:
class TestSdkAndBundledCli:
"""Nøkkelantakelse (verified against installed claude-agent-sdk 0.2.139): the
"""Nøkkelantakelse (verified against installed claude-agent-sdk 0.2.120): the
SDK imports and its bundled CLI is present on disk — checked OFFLINE via the
package's own files, mirroring the SDK's ``_find_cli`` order. Never a net call."""

8
tests/test_sdk_isolation.py
View file

@ -2,7 +2,7 @@
The S10 live run leaked the operator's Claude Code configuration into every
spawned CLI session: ``ClaudeAgentOptions.setting_sources`` defaults to
``None``, which loads ALL filesystem settings (verified against SDK 0.2.139)
``None``, which loads ALL filesystem settings (verified against SDK 0.2.110)
— session-start hooks injected STATE.md into the model's context, every reply
opened with a mandated confirmation line (so a reply was NEVER pure JSON),
and each call paid ~10-15k uncached context tokens. ``[]`` is the SDK's
@ -49,13 +49,11 @@ class TestBuildCallOptions:
def test_the_system_prompt_is_not_the_claude_code_preset(self) -> None:
# Pins the OPTION value against the Claude Code preset. That None
# reaches the spawned CLI as --system-prompt "" was verified by
# READING subprocess_cli.py (0.2.110–0.2.139, re-read at 0.2.139:
# `system_prompt is None` still serializes to `--system-prompt ""`)
# — this test does NOT
# READING subprocess_cli.py (0.2.110–0.2.120) — this test does NOT
# bind that transport serialization; doing so would couple the suite
# to SDK-private API (the F11 fragility this repo retired).
#
# HONEST LIMIT (measured, re-measured at 0.2.139): `system_prompt=None` is NOT
# HONEST LIMIT (measured, 0.2.120): `system_prompt=None` is NOT
# distinguishable from leaving the field untouched — the SDK default
# is None too, so deleting `system_prompt=None` from build_call_options
# left the old `is None` assertion GREEN. It pinned the SDK's default,

128
tests/test_sdk_tool_namespace_loadbearing.py
View file

@ -1,128 +0,0 @@
"""B4: does the Claude Agent SDK hand us server identity for free in tool names?
The D7 mirroring queue (``docs/2026-08-18-funn-koeer-og-gater.md § D7-speilingskøen``)
asks one question of this side: when ``create_sdk_mcp_server`` emits a tool, does
that tool's name carry the server it belongs to? The MAF sibling gets tool names
WITHOUT a server prefix (their S3.4/F10), so a recorder there must carry server
identity itself. Whether we pay the same price was never measured here.
MEASURED 2026-08-25, offline, against claude-agent-sdk 0.2.139 (the version
``tests/test_sdk_version_guard.py`` pins as read): **NO — not free.** The Python
package emits the bare tool name; ``mcp__`` appears in ZERO of its 24 source files
(positive control: ``create_sdk_mcp_server`` IS findable by the same query).
Server identity lives on ``McpSdkServerConfig["name"]`` and ``Server.name``,
disjoint from every name the tool list carries.
HONEST LIMIT — what this does NOT say. The namespaced form ``mcp__<server>__<tool>``
does exist: it is built inside the CLI bundled with the SDK (178 literal
occurrences of ``mcp__`` in ``_bundled/claude``; the construction reads
``` `mcp__${...}__${...}` ```). That was READ from the artifact, not observed in a
run of ours — seeing it emitted would take a live ``query()``, which the D6 cost
rule and this suite's offline invariant both forbid. So the finding is scoped to
the seam we can actually build on: the in-process construction side. A recorder
hung there sees the bare name and must be told which server it came from —
the same shape the sibling pays for, arrived at after 2026-08-09 and therefore
NOT readable as independent convergence.
Offline-safe: constructing an SDK MCP server touches no network and needs no API
key. Nothing here starts the bundled CLI.
"""
from __future__ import annotations
import asyncio
from typing import Any
from claude_agent_sdk import McpSdkServerConfig, create_sdk_mcp_server, tool
SERVER_NAME = "tool_call_recorder"
BARE_TOOL_NAMES = ("record_call", "flush_calls")
# What a "free" server identity would have to look like for B4 to be answered
# YES. Kept as data so the assertions below read as the question, not as a
# restatement of the answer.
NAMESPACED_PREFIX = f"mcp__{SERVER_NAME}__"
@tool("record_call", "Record one tool call", {"name": str})
async def _record_call(args: dict[str, Any]) -> dict[str, Any]:
return {"content": [{"type": "text", "text": args["name"]}]}
@tool("flush_calls", "Flush recorded calls", {})
async def _flush_calls(args: dict[str, Any]) -> dict[str, Any]:
return {"content": [{"type": "text", "text": "flushed"}]}
def _server(*, tools: list[Any] | None = None) -> McpSdkServerConfig:
return create_sdk_mcp_server(
name=SERVER_NAME,
version="1.0.0",
tools=[_record_call, _flush_calls] if tools is None else tools,
)
def _emitted_tool_names(config: McpSdkServerConfig) -> list[str]:
"""Read the tool names the server ACTUALLY emits.
Not a model of the emission — the registered ``tools/list`` handler is
invoked and its result read, because a check that models a tool instead of
reading it is green-but-dead (økt 28). The coupling that buys this is the
MCP lowlevel ``Server.request_handlers`` dispatch table; should its shape
change, the lookup below raises rather than quietly returning nothing, so
the failure mode is RED, not a false negative.
"""
from mcp.types import ListToolsRequest
handler = config["instance"].request_handlers[ListToolsRequest]
async def _call() -> list[str]:
result = await handler(ListToolsRequest(method="tools/list"))
return [t.name for t in result.root.tools]
return asyncio.run(_call())
class TestEmittedToolNamesCarryNoServerIdentity:
def test_the_emission_is_readable_and_the_population_is_two(self) -> None:
# POSITIVE CONTROL, and it runs FIRST: every negative below is worth
# exactly as much as this query's ability to find anything at all. An
# empty tool list would make "no name carries the prefix" vacuously
# true — so the denominator is asserted, not assumed.
names = _emitted_tool_names(_server())
assert len(names) == 2, f"emission unreadable or empty — measured {names!r}"
assert all(isinstance(n, str) and n for n in names)
def test_no_emitted_name_carries_the_server_prefix(self) -> None:
# THE B4 ANSWER, pinned. Red the day the SDK starts namespacing at
# construction time — which is the day this repo can stop carrying
# server identity by hand and the D7 queue note becomes wrong.
names = _emitted_tool_names(_server())
assert names == list(BARE_TOOL_NAMES)
assert not any(n.startswith("mcp__") for n in names), (
f"the SDK now namespaces at construction time — B4 flipped to YES: {names!r}"
)
assert not any(SERVER_NAME in n for n in names)
def test_the_name_is_passed_through_untouched(self) -> None:
# POSITIVE CONTROL FOR THE NEGATIVE. The assertion above only means
# something if a prefix WOULD have shown up had one been there. Feed
# the server a tool already wearing the namespaced form: it comes back
# verbatim. So the emission is pass-through, the query can see a
# prefix, and its absence above is a measurement — not a blind spot.
@tool(f"{NAMESPACED_PREFIX}record_call", "Pre-namespaced", {"name": str})
async def _prefixed(args: dict[str, Any]) -> dict[str, Any]:
return {"content": [{"type": "text", "text": args["name"]}]}
names = _emitted_tool_names(_server(tools=[_prefixed]))
assert names == [f"{NAMESPACED_PREFIX}record_call"]
def test_server_identity_exists_but_lives_off_the_tool_name(self) -> None:
# The disjointness is the whole finding: identity is AVAILABLE, just
# not on the tool. A recorder must join the two itself — this pins
# both halves so "free" cannot be assumed from either one alone.
config = _server()
assert config["type"] == "sdk"
assert config["name"] == SERVER_NAME
assert config["instance"].name == SERVER_NAME
assert SERVER_NAME not in "".join(_emitted_tool_names(config))

223
tests/test_sdk_version_guard.py
View file

@ -1,38 +1,17 @@
"""SDK version guard (C2.5, closes C-N3) — LOAD-BEARING (§11).
The seam this file keeps alive: every SDK attribute premise in
``sdk_client.py`` rests on a version whose source was actually READ.
``sdk_client.py`` was verified against a CONCRETE version range
(0.2.110 read at source level, release notes through 0.2.120; sdk-review
2026-07-16). The pin ``claude-agent-sdk>=0.2.111,<0.3`` freezes that range —
this guard makes an upgrade outside it a RED test naming exactly which
premises must be re-verified, instead of a silent behaviour drift.
Three questions hide inside "is the SDK fine?", and conflating them is what
let a whole version range go unexamined (measured 2026-08-09: installed
0.2.120, pin ``>=0.2.111,<0.3``, premises source-verified only through
0.2.110 — so 0.2.111–0.2.999 satisfied the pin while nobody had read them):
(a) is the build inside the PIN? — what ``uv`` is allowed to resolve
(b) is the build inside the VERIFIED? — what a human has actually read
(c) is something newer upstream? — REQUIRES NETWORK, never in pytest
A pin is a permission, not a proof, and it can reach further than the reading
that justified it. ``_VERIFIED_THROUGH`` is the ratchet for (b): a build newer
than the last one read at source goes RED naming the premises to re-verify,
even though the pin admits it. Question (c) is deliberately absent — an
offline suite cannot answer it, and a test that reached the network to try
would trade a silent gap for a flaky one.
Two instruments, because they fail differently: ``TestTheSdkSurfaceInventory``
introspects the INSTALLED package, so a premise naming an attribute that no
longer exists goes red on ANY build — but introspection only sees the names it
was told to look for, and cannot see a field whose MEANING changed underneath a
stable name. ``_VERIFIED_THROUGH`` covers exactly that blind spot by demanding
a human read. Neither subsumes the other.
Offline-safe: reads installed package metadata and already-imported classes —
no key, no network.
Offline-safe: reads installed package metadata only — no key, no network.
"""
from __future__ import annotations
import dataclasses
import importlib.metadata
from pathlib import Path
@ -44,14 +23,6 @@ _PYPROJECT = Path(__file__).resolve().parents[1] / "pyproject.toml"
_VERIFIED_FLOOR = (0, 2, 111)
_VERIFIED_CEILING = (0, 3)
# The newest build whose SOURCE was read for the premises below — question (b).
# Raising this is a claim that someone opened the package and checked, so it
# moves ONLY together with that reading. 0.2.139 read 2026-08-18: the public
# ``query.py`` is byte-identical to 0.2.120, every premise field is present with
# an unchanged type and default, and the parser changes are purely additive
# (a new ``origin`` passthrough, a new ``ConversationResetMessage``).
_VERIFIED_THROUGH = (0, 2, 139)
def _pin_for(floor: tuple[int, ...], ceiling: tuple[int, ...]) -> str:
"""The pyproject requirement string the verified range implies.
@ -67,39 +38,16 @@ def _pin_for(floor: tuple[int, ...], ceiling: tuple[int, ...]) -> str:
_PIN = _pin_for(_VERIFIED_FLOOR, _VERIFIED_CEILING)
# The sdk_client.py attribute premises, as a table the suite can CHECK rather
# than a sentence it can only print. Every entry is read by sdk_client.py; the
# inventory below asserts each one against the installed package, so a premise
# that quietly stopped being true fails HERE instead of on the one live run.
_SDK_SURFACE: tuple[tuple[str, tuple[str, ...]], ...] = (
("AssistantMessage", ("error", "model", "content")),
("ResultMessage", ("usage", "total_cost_usd", "is_error", "subtype", "errors")),
(
"ClaudeAgentOptions",
("max_budget_usd", "setting_sources", "system_prompt", "max_turns", "model", "tools"),
),
("TextBlock", ("text",)),
# The sdk_client.py attribute premises the verified range vouches for
# (sdk-review 2026-07-16, verified against package source through 0.2.120).
_SDK_PREMISES = (
"AssistantMessage.error/.model/.content",
"ResultMessage.usage/.total_cost_usd/.is_error/.subtype/.errors",
"ClaudeAgentOptions.max_budget_usd/.setting_sources/.system_prompt/.max_turns/.model/.tools",
"TextBlock.text",
"query() yields AssistantMessage then a closing ResultMessage",
)
# The premise introspection is STRUCTURALLY unable to see: an ordering fact
# about a stream, not an attribute on a class. It is named separately rather
# than dropped, because the honest reason the version ratchet still exists is
# that this line can only be checked by a human reading the source.
_SEMANTIC_PREMISES = ("query() yields AssistantMessage then a closing ResultMessage",)
def _premise_text(surface: tuple[tuple[str, tuple[str, ...]], ...]) -> tuple[str, ...]:
"""The prose the failure message prints — DERIVED from the checked table.
Not a second hand-maintained copy: an attribute added to the table appears in
the operator-facing message for free, and one removed cannot linger there
claiming a premise nobody verifies any more (value-proof below).
"""
return tuple(f"{cls}." + "/.".join(attrs) for cls, attrs in surface) + _SEMANTIC_PREMISES
_SDK_PREMISES = _premise_text(_SDK_SURFACE)
def _parse(raw: str) -> tuple[int, ...]:
return tuple(int(part) for part in raw.split(".")[:3])
@ -135,33 +83,6 @@ def _pin_is_declared(text: str, pin: str) -> bool:
return f'"{pin}"' in body
def missing_sdk_attributes(
surface: tuple[tuple[str, tuple[str, ...]], ...] = _SDK_SURFACE,
) -> tuple[str, ...]:
"""Premises the INSTALLED build fails to satisfy — introspected, never assumed.
Fail-closed on three distinct absences, because each would otherwise read as
"no gaps found": the class is gone from the package, the class is no longer a
dataclass (so its fields are not what this check knows how to read), or the
attribute is missing. Returning () must mean "checked and clean", never
"could not look" — that conflation is the whole defect class.
"""
import claude_agent_sdk
gaps: list[str] = []
for cls_name, attrs in surface:
cls = getattr(claude_agent_sdk, cls_name, None)
if cls is None:
gaps.append(f"{cls_name}: class absent from claude_agent_sdk")
continue
if not dataclasses.is_dataclass(cls):
gaps.append(f"{cls_name}: no longer a dataclass — fields unreadable")
continue
declared = {field.name for field in dataclasses.fields(cls)}
gaps.extend(f"{cls_name}.{attr}" for attr in attrs if attr not in declared)
return tuple(gaps)
def check_sdk_version() -> str:
"""Fail if the installed SDK is outside the verified range — naming the premises."""
raw = importlib.metadata.version("claude-agent-sdk")
@ -173,19 +94,6 @@ def check_sdk_version() -> str:
"BEFORE widening the pin (pyproject.toml + this guard together): "
+ "; ".join(_SDK_PREMISES)
)
if _parse(raw) > _VERIFIED_THROUGH:
# Question (b), the one the pin cannot answer: this build is PERMITTED
# and unread. Green here would mean `uv sync --upgrade` silently retires
# the verification that justified the pin in the first place.
through = ".".join(str(part) for part in _VERIFIED_THROUGH)
raise AssertionError(
f"claude-agent-sdk {raw} satisfies the pin "
f"{_PIN.removeprefix('claude-agent-sdk')} but is NEWER than the last build "
f"read at source ({through}). The pin permits it; nobody has verified it. "
"Read the new version's source, then raise _VERIFIED_THROUGH in the SAME "
"commit that installs it — these sdk_client.py premises are what to re-check: "
+ "; ".join(_SDK_PREMISES)
)
return raw
@ -214,114 +122,11 @@ class TestSdkVersionGuard:
with pytest.raises(AssertionError):
check_sdk_version()
def test_a_version_inside_the_pin_but_beyond_the_verified_trips_the_guard(
self, monkeypatch: pytest.MonkeyPatch
) -> None:
# THE HOLE THIS RATCHET CLOSES (measured 2026-08-09): 0.2.140 satisfies
# the pin, so `uv sync --upgrade` installs it without a word, and before
# _VERIFIED_THROUGH existed the whole suite stayed GREEN on a build whose
# premises nobody had read. A permission is not a proof.
monkeypatch.setattr(importlib.metadata, "version", lambda name: "0.2.140")
with pytest.raises(AssertionError) as err:
check_sdk_version()
message = str(err.value)
assert "sdk_client.py" in message
# The message must separate the two questions, or the operator re-reads
# the pin — the thing that was never wrong — instead of the source.
assert "satisfies the pin" in message
assert "NEWER than the last build read at source" in message
for premise in _SDK_PREMISES:
assert premise in message
def test_the_build_read_at_source_itself_passes(self, monkeypatch: pytest.MonkeyPatch) -> None:
# BOUNDARY CONTROL: a ratchet that rejected everything would pass the test
# above for the wrong reason. The verified build itself must be accepted.
through = ".".join(str(part) for part in _VERIFIED_THROUGH)
monkeypatch.setattr(importlib.metadata, "version", lambda name: through)
assert check_sdk_version() == through
def test_the_verified_reading_lies_inside_the_pin_it_justifies(self) -> None:
# A reading outside the installable range would vouch for a build uv can
# never resolve — the ratchet would be green and inert.
assert _VERIFIED_FLOOR <= _VERIFIED_THROUGH < _VERIFIED_CEILING
def test_the_pyproject_pin_matches_the_verified_range(self) -> None:
# The seam: what uv installs and what this guard vouches for are ONE range.
assert _pin_is_declared(_PYPROJECT.read_text(encoding="utf-8"), _PIN)
class TestTheSdkSurfaceInventory:
"""The premise table, checked against the INSTALLED package (§11).
Until now the premises existed only as a sentence printed on failure — prose
the suite could recite but never test. Any of them could have stopped being
true and nothing would have gone red until the one live run (S10, D6) spent
real money to find out.
"""
def test_every_premise_holds_on_the_installed_build(self) -> None:
assert missing_sdk_attributes() == ()
def test_the_table_is_not_empty_of_the_things_it_claims_to_check(self) -> None:
# POSITIVE CONTROL, ahead of the negative: "no gaps" is also what an
# EMPTY table returns. Without this, the assertion above would hold just
# as well for a premise list somebody had quietly deleted.
assert len(_SDK_SURFACE) == 4
assert sum(len(attrs) for _, attrs in _SDK_SURFACE) == 15
def test_every_checked_attribute_is_named_in_the_operator_facing_message(self) -> None:
printed = "; ".join(_SDK_PREMISES)
for cls_name, attrs in _SDK_SURFACE:
for attr in attrs:
assert attr in printed, f"{cls_name}.{attr} is checked but never reported"
class TestTheInventoryGoesRedWhenTheSurfaceMoves:
"""LOAD-BEARING (§11): prove the inventory can SEE absence, on a mutated COPY.
A checker that reports no gaps is making a claim about the package; it is
indistinguishable from a checker that cannot look. Each proof below breaks the
surface deliberately and requires the specific gap to be named — mirroring the
known-positive discipline the version ratchet above already follows.
"""
def test_red_when_a_premise_names_an_attribute_the_build_lacks(self) -> None:
gaps = missing_sdk_attributes((("ResultMessage", ("total_cost_usd", "no_such_field")),))
# Not merely "non-empty": the REAL attribute must survive as satisfied and
# only the fabricated one be reported, or a checker that flagged everything
# would pass this too.
assert gaps == ("ResultMessage.no_such_field",)
def test_red_when_the_class_disappears_from_the_package(self) -> None:
gaps = missing_sdk_attributes((("NoSuchMessage", ("text",)),))
assert gaps == ("NoSuchMessage: class absent from claude_agent_sdk",)
def test_red_when_the_checked_name_is_not_a_dataclass(self) -> None:
# `query` is a REAL export of the package — a genuine known-positive, not a
# fabricated stand-in. It is a function, so its fields are not readable the
# way this check reads fields, and the honest answer is a reported gap
# rather than a silent () from a `dataclasses.fields` TypeError.
import claude_agent_sdk
assert hasattr(claude_agent_sdk, "query"), "control is broken: the export is gone"
gaps = missing_sdk_attributes((("query", ("text",)),))
assert gaps == ("query: no longer a dataclass — fields unreadable",)
def test_red_when_the_prose_stops_covering_the_table(self) -> None:
# VALUE-proof on the derivation: an attribute added to the table appears in
# the operator-facing message WITHOUT a second edit. Were the prose a hand-
# maintained copy, this new attribute would be checked and never reported —
# the exact drift that made the old sentence outlive its verification.
widened = _premise_text((("ResultMessage", ("usage", "a_newly_relied_on_field")),))
assert "a_newly_relied_on_field" in "; ".join(widened)
assert "a_newly_relied_on_field" not in "; ".join(_SDK_PREMISES)
def test_the_semantic_premise_survives_derivation(self) -> None:
# The stream-ordering premise has no attribute to introspect, so nothing
# would go red if it silently dropped out of the derived prose.
assert _SEMANTIC_PREMISES[0] in "; ".join(_SDK_PREMISES)
class TestTheSeamGoesRedWhenDetached:
"""LOAD-BEARING (§11): the assertions above must FAIL on a detached pyproject.

14
uv.lock generated
View file

@ -177,7 +177,7 @@ wheels = [
[[package]]
name = "claude-agent-sdk"
version = "0.2.139"
version = "0.2.120"
source = { registry = "https://pypi.org/simple" }
dependencies = [
{ name = "anyio" },
@ -185,13 +185,13 @@ dependencies = [
{ name = "sniffio" },
{ name = "typing-extensions", marker = "python_full_version < '3.11'" },
]
sdist = { url = "https://files.pythonhosted.org/packages/11/b6/cfcdefed1f866a8ba372ef3884c8020dd54338d15d8b45d5a1ff7432cea1/claude_agent_sdk-0.2.139.tar.gz", hash = "sha256:4395ed541cdd4c13aeb1213b3b414b7e8a94cc060a773137e961882e81c174a7", size = 319519, upload-time = "2026-08-14T22:34:48.038Z" }
sdist = { url = "https://files.pythonhosted.org/packages/eb/7f/7b69aed292a4edecae132e4dbe6b6decb4e88ec142fc91d117b19058c9e0/claude_agent_sdk-0.2.120.tar.gz", hash = "sha256:e428552f79a76e0d85789369eeb58249b33f350200124e5fc86b24168bd00805", size = 268639, upload-time = "2026-07-15T23:18:50.997Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/ea/7f/f04c33553cbc69bb96d045dc38a6266726fad72130f22f405dfe9eb54bf1/claude_agent_sdk-0.2.139-py3-none-macosx_11_0_arm64.whl", hash = "sha256:cbc50cc475ec633cabfa36347646097e9b1466d53130e4a04a87308ff830c87b", size = 88043656, upload-time = "2026-08-14T22:34:53.027Z" },
{ url = "https://files.pythonhosted.org/packages/85/d0/a17f5318ca0220479f20fdf83fa54a838a0a13ee203495ff67c72c3f43a7/claude_agent_sdk-0.2.139-py3-none-macosx_11_0_x86_64.whl", hash = "sha256:1c08206b1603444582cd365effaf95d2a8248661f1492281fb2d529b0887c047", size = 93000433, upload-time = "2026-08-14T22:34:58.225Z" },
{ url = "https://files.pythonhosted.org/packages/c5/2e/5bcec31700d76ad2d5b9fc28521a75a66f464063dac11373cf8d61446a4f/claude_agent_sdk-0.2.139-py3-none-manylinux_2_17_aarch64.whl", hash = "sha256:e69ae1a0b2af684c64839cc16e10b70800d9d2f57622b8c0d1739dd878cd7346", size = 97396659, upload-time = "2026-08-14T22:35:03.734Z" },
{ url = "https://files.pythonhosted.org/packages/c8/7f/582b3c1936c9f4ebc1bdc55a3923f1b680ef3c01928ffff1ea38eb84f637/claude_agent_sdk-0.2.139-py3-none-manylinux_2_17_x86_64.whl", hash = "sha256:34b289b3436fe24013f7b9cfe9f0a4e0806917a9ef8bbe829cda9a7b12d41a77", size = 98391889, upload-time = "2026-08-14T22:35:09.683Z" },
{ url = "https://files.pythonhosted.org/packages/56/54/d94af31d19b4e8d63d1b15002fd333ea77a040b7ab7a388044e511c8f9f6/claude_agent_sdk-0.2.139-py3-none-win_amd64.whl", hash = "sha256:9b76f0ffe216d6ca290d5f4f295ecb030dc496f101986ac99480a89d4abc6426", size = 100746507, upload-time = "2026-08-14T22:35:15.144Z" },
{ url = "https://files.pythonhosted.org/packages/80/85/5e8958704db0f8195e63f8ec4a80c5fb14756edc785bb3535e0dc5d91104/claude_agent_sdk-0.2.120-py3-none-macosx_11_0_arm64.whl", hash = "sha256:c5ead9fb4bdaf70069978703ec6d74b30bd269e9632a4aea4dc8c4e999ac3a1b", size = 71110966, upload-time = "2026-07-15T23:18:54.743Z" },
{ url = "https://files.pythonhosted.org/packages/c1/53/c6cdad82ac100c8a45887999614e9fe206b77b43790dacc6de42b156ad4c/claude_agent_sdk-0.2.120-py3-none-macosx_11_0_x86_64.whl", hash = "sha256:1248591c7bffeb6e10e8cd169e0766854951fba8816e3e7d81a003f8bfca6f08", size = 76067995, upload-time = "2026-07-15T23:18:58.398Z" },
{ url = "https://files.pythonhosted.org/packages/bf/f8/248e3f58d0f0aa7d76bd34b11b18135cc124f5b9a9b55219cc1ca03d662a/claude_agent_sdk-0.2.120-py3-none-manylinux_2_17_aarch64.whl", hash = "sha256:abc73ccdf3decca566cd18084e74bc2f2d10b8b77cc1fd5ed4299d5c15e5078b", size = 81027434, upload-time = "2026-07-15T23:19:03.318Z" },
{ url = "https://files.pythonhosted.org/packages/11/59/6adb0c53534646f1d5ddc41226ff37b2adc413a472e2f87a9011548a137f/claude_agent_sdk-0.2.120-py3-none-manylinux_2_17_x86_64.whl", hash = "sha256:888070c246c92e102c52001d26532cd3646a700656d7c368f2f91a1d3c16b534", size = 82084704, upload-time = "2026-07-15T23:19:08.729Z" },
{ url = "https://files.pythonhosted.org/packages/2a/06/036b8dce1e86ecd5e2e1ddc281736cdb33bad6d24e748b9553e235b028fe/claude_agent_sdk-0.2.120-py3-none-win_amd64.whl", hash = "sha256:bc1441c94f60c9e7b4b8c641742fedf68451f573062ef69dd43d86a61b1fb219", size = 81958518, upload-time = "2026-07-15T23:19:13.082Z" },
]
[[package]]