open/portfolio-optimiser
1
0
Fork 0
Code Issues Releases 3 Activity
portfolio-optimiser/src/portfolio_optimiser/okf.py
Kjell Tore Guttormsen ac0bfdba27 feat(p21): a declaration that must have LOOKED, and a refusal that names the neighbours 
C1. Round 4 produced 13 declarations over six runs and NOT ONE named a fasit concept.
The distinct documents opened before each were 1,1,1,1,1,1,1,2,5,5,6,13,13: seven
declared the base's FIRST requirement after opening exactly ONE document.

The order offered two rules and asked which discriminates. Replayed against the real
listings: "the declared document must have come back from a read_dir filtered on a word
from the approach's label" refuses 13 of 13 -- including Soraasen's 12.11, the closest
any run came -- because ZERO of the 13 were reached through a filtered listing at all.
A gate that refuses every measured case, right and wrong alike, cannot discriminate.
"fewer than k distinct documents opened" at k=3 refuses 8 of 13 and keeps the five that
navigated. k=3, 4 and 5 refuse the SAME eight -- the distribution has a gap between 2
and 5 -- so the threshold is not on a cliff, and 3 is the lowest of that plateau.
DISTINCT paths, not calls, and capped by the base's own size so a small base stays
declarable.

C2. Over the same traces 18 of 143 path-bearing calls named a path the base does not
hold, ELEVEN of them one run walking R761/4-3, 4.3, 4-2, 4-1, 4-0, 4-5, 4-6 while the
real names are R761/4, R761/41, R761/42. The refusal already named the nearest listable
ancestor; now it also names up to five of that rung's own subdirectories, ranked by
longest common prefix with the segment that failed. ONE copy shared by both refusal
sites, built from context_files through in_dimension, so every name handed back resolves
and the verdict layer can never be advertised in an apology. MEASURED after: 16 of 18.

Load-bearing MEASURED, four mutations all red against the WHOLE suite, green control
1863/5, golden byte-unchanged: C3(i) the declaration gate detached (2 red) . C3(ii) the
neighbour list empty (6) . C3(iii) built from files (1, the verdict arm alone) . C3(iv)
count CALLS instead of distinct documents (1, the repetition arm alone).

Three existing arms REWRITTEN, not weakened: all three read one document and declared,
which is the measured failure class exactly.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
2026-09-15 11:38:57 +02:00

2038 lines
108 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

"""OKF (Open Knowledge Format) bundle navigation — framework-neutral, D7-portable context seam.
Reads a bundle the way OKF intends (progressive disclosure): start at ``index.md``, follow
intra-bundle cross-links **recursively, depth-first in first-seen link order**, parse each file's
YAML frontmatter, classify by the one required ``type`` field. **NO** ``agent_framework``, **NO**
``mcp`` — stdlib + ``pydantic`` only (as ``dimension.py``; the typed contracts this module loads
live in ``ir.py``), so the SAME navigation serves both the MAF and the Claude-SDK
implementations unchanged (målbilde §4 vendor-neutrality).
Link resolution follows ``shared/method-spec.md`` §3 Step 1: a leading ``/`` denotes the **bundle
root** (NEVER a filesystem-absolute path), any other form is relative to the LINKING FILE's own
directory — so a bundle may be a hierarchy. It is **escape, not depth**, that is forbidden; this
resolve-and-boundary-check replaced the old "a path separator means out-of-bundle" heuristic, which
conflated the two and forbade valid nesting. Repeated links de-duplicate on the RESOLVED path, so
``./a.md`` and ``a.md`` are one entry and cycles terminate.
Robustness is part of the spec (OKF SPEC §4): consumers MUST tolerate broken links and unknown
fields. A target that fails to resolve for ANY reason (missing file, invalid path component,
escape) is silently skipped, never raised. Path-safety reuses ``retrieval.safe_resolve`` (also pure
stdlib): each cross-link is canonicalised and boundary-checked against the bundle dir, fail-closed
— the SOLE in-/out-of-bundle test.
Skipped is not SILENT, though: every link the walk could not follow is recorded on
``Bundle.skipped`` as a ``SkippedLink`` (which file it was written in, the link text verbatim, and
which of the two reasons applied). The tolerance is unchanged — nothing raises — but a bundle whose
other half was never reached is no longer indistinguishable from one where those documents were
never written.
"""
from __future__ import annotations
import itertools
import json
import posixpath
import re
from dataclasses import dataclass
from pathlib import Path
from collections.abc import Sequence
from typing import Any, Final, Literal
from portfolio_optimiser.ir import CostBaseline, CostBaselineLine
from portfolio_optimiser.retrieval import PathSecurityError, safe_resolve
_INDEX_NAME = "index.md"
#: The declared ``type`` of the layer the gated ExpeL fold owns. ONE copy of "what a verdict IS"
#: (kø-(p)): three surfaces now answer it — ``Bundle.verdicts`` (the seeds Step 1 retrieves),
#: ``context_files`` (the bodies agents may be shown) and ``declares_verdict_type`` (the gate on the
#: bytes leaving ``read_file``). Two copies would be free to disagree about one document, and the
#: disagreement would show up as a judgement reaching a hypothesis around the fold.
_VERDICT_TYPE = "verdict"
_IR_PROJECTION = "validator-input.json"
_COST_BASELINE = "cost-baseline.json"
# Intra-bundle markdown cross-links: ``](target.md)``. A path separator is NOT a rejection reason —
# ``_resolve_target`` decides in-/out-of-bundle, and only escape is refused (method-spec §3 Step 1).
_LINK_RE = re.compile(r"\]\(([^)]+\.md)\)")
def unquote_scalar(raw: str) -> str:
"""The ONE unquoting rule for a frontmatter scalar — ``parse_frontmatter`` preserves quotes
(OKF SPEC §4), so every consumer of a scalar has to take them off, and they must all take them
off the SAME way.
It lives here because this module owns ``parse_frontmatter``. It was a private copy in
``verdicts`` (structural key) while ``bundle_context`` stripped only ``"`` (rendered title) —
two rules, and the weaker one in the path that becomes the agent's read-context. Both quote
styles, and whitespace on either side of them, are ordinary YAML a hand-authoring curator
writes. Mirrors the (p) precedent: a duplicated conversion drifts, and the drifted copy decides
something. Gated by ``tests/test_frontmatter_unquote_loadbearing.py``."""
return raw.strip().strip('"').strip("'").strip()
def _split_frontmatter(text: str) -> tuple[list[str], str, bool]:
"""Scan the leading ``---``-delimited block ONCE: ``(frontmatter lines, body, terminated)``.
This is the module's ONLY place the delimiter is compared against. ``parse_frontmatter`` and
``_read_body`` each had their own loop over the same delimiter, which is the kø-(p) shape — and
here the two copies had already drifted, measured: given an opening ``---`` with no closing one,
``parse_frontmatter`` consumed every remaining line as frontmatter while ``_read_body`` fell
through and returned the WHOLE file, delimiter line included.
**That divergence is PINNED, not fixed.** Reconciling it would move the body-rendering path both
nav-golden fasits read, which nothing asks for. So this function reports FACTS and decides
nothing: ``terminated`` says whether a closing delimiter was found, and each caller keeps
applying its own existing rule to it. ``body`` is the text after a CLOSED block and is ``""``
whenever ``terminated`` is false — a caller that wants the whole-file fallback must say so,
rather than receive it silently from a scanner that cannot know which rule applies.
Frontmatter lines are returned VERBATIM apart from their line ending: leading indentation is
load-bearing for block-form values (``verified:`` as a sequence of mappings, SPEC §5.2), so the
decoder that consumes these lines gets a second READER of one parse, never a second parser.
Gated by ``tests/test_provenance_decoder_loadbearing.py``."""
lines = text.splitlines(keepends=True)
if not lines or lines[0].strip() != "---":
return [], "", False
for i in range(1, len(lines)):
if lines[i].strip() == "---":
return (
[line.rstrip("\r\n") for line in lines[1:i]],
"".join(lines[i + 1 :]).lstrip("\n"),
True,
)
return [line.rstrip("\r\n") for line in lines[1:]], "", False
def _frontmatter_from_text(text: str) -> dict[str, str]:
"""``parse_frontmatter``'s rule, applied to already-read text: every frontmatter line that
carries a colon becomes one ``key: value`` pair, last write winning AMONG LINES AT THE SAME
LEVEL. Unterminated blocks are parsed as if closed — the pre-split behaviour, preserved
deliberately.
A TOP-LEVEL key (no leading whitespace — ``_split_frontmatter`` preserves indentation
verbatim, so this is a real structural signal, not a guess) always wins over an indented one
of the same name: a nested line is a block-sequence/-mapping entry under some OTHER top-level
key (``sources:\\n - title: …`` names the SOURCE, not the concept) and must never overwrite
the concept's own field, however late it appears in the scan. Measured 2026-09-13 on
vegnormal-okf concepts: without this, every ``sources:``-bearing file's nested ``title``
replaced its own, collapsing 4605 concepts to 4 distinct titles. A nested line with no
top-level counterpart is still preserved (OKF SPEC §4, "unknown fields are preserved")."""
fm: dict[str, str] = {}
top_level: set[str] = set()
for line in _split_frontmatter(text)[0]:
if not line:
continue
key, sep, val = line.partition(":")
if not sep:
continue
key = key.strip()
if line[0].isspace():
if key in top_level:
continue
fm[key] = val.strip()
else:
fm[key] = val.strip()
top_level.add(key)
return fm
def _body_from_text(text: str) -> str:
"""``_read_body``'s rule, applied to already-read text: the body after a CLOSED frontmatter
block, and otherwise the whole file — which covers both "no block at all" and "opened but never
closed". This is the caller-side rule ``_split_frontmatter`` deliberately refuses to apply."""
_, body, terminated = _split_frontmatter(text)
return body if terminated else text
def parse_frontmatter(path: str | Path) -> dict[str, str]:
"""Read the leading ``---``-delimited YAML frontmatter block as key:value strings.
Minimal by design (no ``yaml`` dependency): enough for the one required ``type`` field and the
verdict's scalar fields. List values (``tags: [...]``) are kept verbatim; unknown fields are
preserved (OKF SPEC §4). Returns ``{}`` when there is no frontmatter block."""
return _frontmatter_from_text(Path(path).read_text(encoding="utf-8"))
def _read_body(path: Path) -> str:
"""The markdown body after the frontmatter block (or the whole file if there is none)."""
return _body_from_text(path.read_text(encoding="utf-8"))
class FlowDecodeError(ValueError):
"""A frontmatter value is outside the accepted single-line flow subset.
A ``ValueError`` subclass on purpose (the ``IngestStampError`` precedent): the CLI's refusal
tuple and hosting's 400 arm both catch ``ValueError``, so a malformed knowledge base is a
refusal the caller can read, never a traceback on the crash channel."""
#: The pair separator INSIDE a flow mapping. Colon-SPACE, never a bare colon — ``by: human:jsmith``
#: and ``at: 2024-01-15T10:00:00Z`` both carry colons that are part of the value, and a decoder
#: that split on ``:`` would quietly truncate every actor and every timestamp it read.
_FLOW_PAIR_SEPARATOR = ": "
def _scan_flow(text: str, separator: str) -> tuple[list[str], str, int]:
"""Split ``text`` on ``separator`` at nesting depth 0 and OUTSIDE quotes.
Returns ``(parts, unclosed_quote, depth)`` — the two trailing values are how the caller tells a
complete value from a truncated one, rather than discovering it later as a wrong answer.
"Split on a separator" is where this class of decoder fails silently, so the scan is
character-by-character and quote-aware instead of ``str.split``."""
parts: list[str] = []
buf: list[str] = []
quote = ""
depth = 0
i = 0
while i < len(text):
ch = text[i]
if quote:
buf.append(ch)
if ch == quote:
quote = ""
i += 1
continue
if ch in "\"'":
quote = ch
buf.append(ch)
i += 1
continue
if ch in "[{":
depth += 1
buf.append(ch)
i += 1
continue
if ch in "]}":
depth -= 1
buf.append(ch)
i += 1
continue
if depth == 0 and text.startswith(separator, i):
parts.append("".join(buf))
buf = []
i += len(separator)
continue
buf.append(ch)
i += 1
parts.append("".join(buf))
return parts, quote, depth
def _has_nested_collection(text: str) -> bool:
"""True when ``text`` opens a ``[`` or ``{`` outside quotes. Nesting is outside the accepted
subset, and depth alone cannot detect it — a balanced ``[x, y]`` returns to depth 0."""
quote = ""
for ch in text:
if quote:
if ch == quote:
quote = ""
continue
if ch in "\"'":
quote = ch
continue
if ch in "[{":
return True
return False
def _find_pair_separator(pair: str) -> int:
"""Index of the FIRST ``": "`` outside quotes, or ``-1``."""
quote = ""
i = 0
while i < len(pair):
ch = pair[i]
if quote:
if ch == quote:
quote = ""
i += 1
continue
if ch in "\"'":
quote = ch
i += 1
continue
if pair.startswith(_FLOW_PAIR_SEPARATOR, i):
return i
i += 1
return -1
def _decode_flow_mapping(item: str, raw: str, key: str | None) -> dict[str, str]:
"""One ``{ k: v, ... }`` flow mapping into a dict, KEY-AGNOSTICALLY."""
inner = item[1:-1]
if _has_nested_collection(inner):
raise FlowDecodeError(
f"a nested flow collection inside {raw!r} is outside the accepted subset — the "
"decoder reads one level of `{ key: value }` pairs and refuses to guess at more"
)
pairs, quote, depth = _scan_flow(inner, ",")
if quote:
raise FlowDecodeError(f"an unterminated quoted scalar in {raw!r}")
if depth != 0:
raise FlowDecodeError(f"an unterminated flow mapping in {raw!r}")
entry: dict[str, str] = {}
for pair in pairs:
at = _find_pair_separator(pair)
if at < 0:
raise FlowDecodeError(
f"{pair.strip()!r} in {raw!r} is not a `key: value` pair (the separator is "
"colon-SPACE) — refusing rather than guessing what was meant"
)
name = unquote_scalar(pair[:at])
value = unquote_scalar(pair[at + len(_FLOW_PAIR_SEPARATOR) :])
if name in entry:
raise FlowDecodeError(
f"duplicate key {name!r} in {raw!r} — last-write-wins is precisely the silent "
"overwrite this decoder exists to remove, so it is refused here too"
)
entry[name] = value
if key == "verified" and not entry.get("by"):
raise FlowDecodeError(
f"a `verified` entry in {raw!r} names no actor — SPEC §5.2 makes `by` required within "
"a verification event, and tiering an entry that names nobody would mint provenance"
)
return entry
def decode_block_mappings(
continuation: Sequence[str], *, key: str | None = None
) -> tuple[dict[str, str], ...] | None:
"""Decode a BLOCK sequence of mappings into entries, or ``None`` when it cannot be read.
``None`` means "this reader cannot decode it", NEVER "there is nothing here": an indented line
before any ``- `` opens no entry, and folding it into one would invent an entry the document
does not carry. The caller turns ``None`` into the ``UnreadableProvenance`` it already returned.
**The second carrier of the same shape, never a second grammar.** The pair separator is
colon-SPACE via ``_find_pair_separator``, names and values go through ``unquote_scalar``,
duplicate keys are refused, and the one key-specific rule SPEC §5.2 imposes (a ``verified``
entry must name an actor) applies here too — all of it the SAME rule ``_decode_flow_mapping``
applies to the flow carrier. Two grammars would be two answers to one question, and a delivered
base would read differently depending on which spelling its producer chose. The colon-SPACE part
is load-bearing rather than stylistic: every delivered ``resource`` is a URL, so a reader
splitting on the FIRST colon would truncate all 4605 of them at ``https``.
Measured 2026-09-12: all four delivered knowledge bases write ``sources`` in this form and none
in flow form (n100 446/446, n200 1133/1133, n500 270/270, r761 2756/2756). Reading it is not a
licence to WRITE it — ``write_concept_file``/``verified_field`` still refuse exactly what
``decode_flow_value`` refuses, so the emission rule and the round-trip gate are untouched.
Gated by ``tests/test_block_sources_reader_loadbearing.py``."""
entries: list[dict[str, str]] = []
from_flow = False
for line in continuation:
if not line.strip():
continue
item = line.strip()
opened = item.startswith("- ")
if opened:
entries.append({})
item = item[2:].strip()
from_flow = False
elif not entries:
return None
if item.startswith("{"):
# A FLOW mapping as the item. Decoded by ``decode_flow_value`` — the ONE decoder —
# rather than by the pair loop below, which would read ``{ id`` as a key and hand back
# an entry the document does not carry. Measured 2026-09-12: this is the shape
# ``tests/golden/block-form-provenance`` writes for ``verified``, and the pair loop
# produced ``{'{ id': 'a, resource: ... }'}`` for it.
if not opened:
return None
try:
decoded = decode_flow_value(item, key=key)
except FlowDecodeError:
return None
if len(decoded) != 1:
return None
entries[-1] = dict(decoded[0])
from_flow = True
continue
if from_flow:
# A bare pair continuing an entry that was opened as a flow mapping. Mixing the two
# spellings within one entry is refused rather than merged: merging would decide
# silently which carrier wins for a document that used both.
return None
at = _find_pair_separator(item)
if at < 0:
return None
name = unquote_scalar(item[:at])
value = unquote_scalar(item[at + len(_FLOW_PAIR_SEPARATOR) :])
if name in entries[-1]:
return None
entries[-1][name] = value
if not entries:
return None
if key == "verified" and any(not entry.get("by") for entry in entries):
return None
return tuple(entries)
def decode_flow_value(raw: str, *, key: str | None = None) -> tuple[dict[str, str], ...]:
"""Decode the accepted single-line flow subset into a tuple of entries.
Two shapes are accepted and nothing else: a flow sequence of flow mappings
``[{ k: v }, { k: v }]``, and a bare flow mapping ``{ k: v }`` which normalises to a
ONE-ELEMENT tuple (SPEC §5.2: "Consumers MUST treat a bare mapping as a one-element list").
Entries decode **key-agnostically** — whatever keys the entry carries, never a hard-coded
``{id, resource}``. The agreed segmented shape adds ``segment_id`` and ``source_offset``, so a
two-key decoder would refuse the very bundles this seam is built for. This is the ONE named
seam: a future structured reader becomes a parameter here, not a refactor everywhere.
**No YAML-1.1 coercion.** Values come back as the strings they were written as: ``yes`` / ``no``
/ ``on`` stay strings and ``1`` stays ``"1"``. This is a DELIBERATE divergence from PyYAML's
resolver, which would return ``True`` and ``1`` — written down here rather than inherited
silently, because a value that changes type between the file and the consumer is exactly the
class of surprise a provenance reader must not import.
``key`` is optional and additive: it carries the ONE key-specific rule SPEC §5.2 imposes, that
a ``verified`` entry must name an actor. Everything else stays key-agnostic.
Raises ``FlowDecodeError`` — by name, with its own message — for a bare-scalar sequence, an
empty sequence, an unterminated flow, a value continued onto the next line, a nested collection,
a duplicate key within one entry, and a ``verified`` entry with no ``by``.
Gated by ``tests/test_provenance_decoder_loadbearing.py``."""
if "\n" in raw or "\r" in raw:
raise FlowDecodeError(
f"the flow value {raw!r} does not fit on one line — a continued value is outside the "
"accepted subset, and joining the lines would decode something nobody wrote"
)
text = raw.strip()
if text.startswith("["):
if not text.endswith("]"):
raise FlowDecodeError(f"an unterminated flow sequence: {raw!r}")
inner = text[1:-1].strip()
if not inner:
raise FlowDecodeError(
f"the flow sequence {raw!r} names no source — an empty list reads as a measured "
"absence when it is the absence of a measurement"
)
chunks, quote, depth = _scan_flow(inner, ",")
if quote:
raise FlowDecodeError(f"an unterminated quoted scalar in {raw!r}")
if depth != 0:
raise FlowDecodeError(f"an unterminated flow mapping in {raw!r}")
entries: list[dict[str, str]] = []
for chunk in chunks:
item = chunk.strip()
if not (item.startswith("{") and item.endswith("}")):
raise FlowDecodeError(
f"a bare scalar entry {item!r} in {raw!r} is not a conformant entry — SPEC "
"§5.1 makes `resource` REQUIRED within an entry, so a naked filename names a "
"resource without saying so"
)
entries.append(_decode_flow_mapping(item, raw, key))
return tuple(entries)
if text.startswith("{"):
if not text.endswith("}"):
raise FlowDecodeError(f"an unterminated flow mapping: {raw!r}")
return (_decode_flow_mapping(text, raw, key),)
raise FlowDecodeError(
f"{raw!r} is neither a flow sequence nor a flow mapping — the accepted subset is "
"`[{ k: v }, ...]` or `{ k: v }` on one line"
)
@dataclass(frozen=True)
class BundleFile:
"""One OKF file: its name, declared ``type`` (``""`` if absent), frontmatter, and body."""
name: str
type: str
frontmatter: dict[str, str]
body: str
#: Why navigation did not follow a cross-link. TWO values, because the two mean different things
#: to whoever has to fix the bundle: ``outside-bundle`` is a target that resolves OUTSIDE the bundle
#: root (frequently a deliberate link to a neighbouring base), ``missing`` is a target that resolves
#: INSIDE it with no readable file there (almost always a typo in the link). Collapsing them into
#: one "skipped" would answer neither question. De-duplication is NOT among them: a repeated link
#: and a cycle are correct navigation, never a skip.
SkipReason = Literal["outside-bundle", "missing"]
@dataclass(frozen=True)
class SkippedLink:
"""One cross-link the walk did NOT follow, and why.
STRUCTURED rather than a rendered string, for the reason ``BudgetExceeded`` carries
``kind``/``limit``/``observed`` as fields (kø-(y)): "which document is missing" and "why is it
missing" are two separate operative questions, and a caller that has to re-parse prose to tell
them apart has been handed a diagnostic it cannot act on.
``target`` is the link text VERBATIM as written in the source file, never the resolved path: the
operator fixing the bundle edits that text, and a normalised form would send them looking for a
string their file does not contain."""
#: Bundle-relative name of the file the link was written in.
from_file: str
#: The link target exactly as it appears in that file.
target: str
reason: SkipReason
#: Characters an actor may not contain, because each would RESTRUCTURE the flow mapping it is
#: written into. ``": "`` is in the set for a reason that is not decoration: ``decode_flow_value``
#: splits pairs on colon-SPACE, so an actor carrying it would be written out cleanly and then
#: mis-decoded on the way back. The writer and the reader must agree on this set, or the round trip
#: lies while every byte looks fine.
_VERIFIED_UNSAFE_TOKENS = (",", "{", "}", "[", "]", "\n", "\r", ": ")
def verified_field(actor: str, at: str) -> str:
"""Compose the SPEC §5.2 single-verifier flow shorthand ``{ by: <actor>, at: <at> }``.
The ENCODER half of this seam, and a separately named function on purpose: it gives the
round-trip property one site to be measured at, and ``render_frontmatter`` cannot serve —
it collapses every newline to a space and therefore cannot emit block form at all.
**The actor is written VERBATIM and is never prefixed with ``human:``.** A caller reaches the
human tier by saying so; minting the prefix on their behalf would fabricate a sign-off nobody
gave.
**Honesty limit, stated:** only ``actor`` is checked against the unsafe set. ``at`` is an ISO
timestamp in every caller today and is not validated here — a deliberate scope line, not an
oversight, and the place to revisit it is the invariant record rather than a silent widening.
Raises ``FlowDecodeError`` — the same named refusal the reader raises, which is what makes
"the writer refuses exactly what the reader cannot read" structural rather than a convention.
Gated by ``tests/test_provenance_decoder_loadbearing.py``."""
for token in _VERIFIED_UNSAFE_TOKENS:
if token in actor:
raise FlowDecodeError(
f"the actor {actor!r} contains {token!r}, which would restructure the `verified` "
"flow mapping — refusing to write a provenance record that parses cleanly into "
"something no one wrote"
)
return f"{{ by: {actor}, at: {at} }}"
#: A concept's trust level, derived from its ``verified`` actors (SPEC §5.3), lowest to highest.
#: Derived, never stored: OKF records objective signals and refuses to persist a subjective score,
#: so this is a reading of the actors and not a field any document carries.
TrustTier = Literal["unverified", "machine-confirmed", "human-reviewed"]
#: The ONE actor prefix that raises a concept to the human tier (SPEC §7 actor convention).
_HUMAN_ACTOR_PREFIX = "human:"
def trust_tier(entries: tuple[dict[str, str], ...] | None) -> TrustTier:
"""Derive the trust tier from ``verified`` entries (SPEC §5.3).
``None`` or empty ⇒ ``unverified``; any actor whose ``by`` STARTS WITH ``human:`` ⇒
``human-reviewed``; otherwise ⇒ ``machine-confirmed``.
**Prefix, never substring.** ``by: bot/human:2`` is a machine actor whose identifier merely
contains the token, and a substring test would promote it — minting a human sign-off nobody
gave, which is the fabricated-provenance defect one level down.
**An entry that names NO actor is REFUSED, never tiered.** "Otherwise ⇒ machine-confirmed"
would derive a trust level from an entry that identifies nobody. ``decode_flow_value`` already
refuses that shape when it reads a ``verified`` value; this function is public, so it ASSERTS
the invariant at its own door instead of assuming its caller came through that one. No new
exception type: the named refusal for this condition belongs to the decoder, and a second class
here would imply a second rule.
**Entry count is not a tier.** SPEC §5.3 derives the tier from the actor prefix alone, so a
two-entry machine list stays ``machine-confirmed``.
Gated by ``tests/test_provenance_decoder_loadbearing.py``."""
if not entries:
return "unverified"
actors: list[str] = []
for entry in entries:
actor = entry.get("by", "").strip()
if not actor:
raise ValueError(
f"a verification entry {entry!r} names no `by` actor — SPEC §5.2 makes it "
"required, and deriving a trust tier from an entry that identifies nobody would "
"mint the provenance it claims to read"
)
actors.append(actor)
if any(actor.startswith(_HUMAN_ACTOR_PREFIX) for actor in actors):
return "human-reviewed"
return "machine-confirmed"
class AdjudicationValueError(ValueError):
"""A concept declares an ``adjudication`` value outside the closed vocabulary.
A ``ValueError`` subclass (the ``IngestStampError`` precedent) so a malformed knowledge base
reaches the CLI's refusal tuple and hosting's 400 arm rather than the crash channel."""
#: Whether a concept has been ADJUDICATED, per the cross-repo contract. The value set on the wire
#: is CLOSED (``proposed`` | ``adjudicated``); the third token is this consumer's, and it is what
#: absence means. See ``docs/okf-konsum-kontrakter.md`` § 2, which is the source for this rule.
AdjudicationState = Literal["proposed", "adjudicated", "unknown"]
#: The closed on-the-wire vocabulary. ``unknown`` is deliberately NOT in it: it is what this reader
#: concludes from absence, never something a document may declare.
_ADJUDICATION_WIRE_VALUES = ("proposed", "adjudicated")
def adjudication_for(path: str | Path) -> AdjudicationState:
"""Read a concept's adjudication state, with absence as a FIRST-CLASS state.
A concept that does not carry the key is ``unknown`` — *we did not learn whether this was
adjudicated*, which is what an older bundle looks like. **Collapsing that into ``absent``**
(*it was not adjudicated*) **is the same defect as collapsing ``unreadable`` into ``absent``**
one layer up, and the same defect removed from ``RunResult.verdict``.
A value outside the two named ones is **refused by name**, never mapped to ``unknown``:
validation, never repair. An EMPTY value is refused for the same reason — present-but-empty is
a different fact from absent, and folding it in would recreate the collapse this function
exists to prevent.
**Read through ``parse_frontmatter``, NOT ``evidence_for`` — and that is a MEASUREMENT, not a
preference.** The plan specified ``evidence_for(path, key="adjudication")``; the contract as
delivered makes ``adjudication`` a plain SCALAR, and ``evidence_for`` routes its value through
``decode_flow_value``. Measured 2026-09-02: BOTH valid values come back as an identical
``state='unreadable', reason='unsupported-flow'`` record, so that route cannot tell ``proposed``
from ``adjudicated`` at all. ``parse_frontmatter`` is the repo's scalar reader over the SAME
``_split_frontmatter`` scan, so this is a second READER of one parse — never a second parser,
which is the rule that route was reaching for.
Gated by ``tests/test_falsification_verdict_loadbearing.py``."""
raw = parse_frontmatter(path).get("adjudication")
if raw is None:
return "unknown"
value = unquote_scalar(raw)
if value not in _ADJUDICATION_WIRE_VALUES:
raise AdjudicationValueError(
f"{value!r} is not an `adjudication` value — the vocabulary is closed to "
f"{' | '.join(_ADJUDICATION_WIRE_VALUES)}, and mapping an unrecognised one to "
"`unknown` would invent the very state that token exists to keep honest"
)
# ``value`` is one of the wire literals, which the Literal type also admits.
return value # type: ignore[return-value]
#: What a document says about ONE provenance key. Three states, and the third is the whole point:
#: collapsing ``unreadable`` into ``absent`` turns a verdict on missing evidence into evidence of
#: absence.
EvidenceState = Literal["present", "absent", "unreadable"]
@dataclass(frozen=True)
class FalsificationEvidence:
"""What one document offers a falsification verdict, and how much of it could be read.
Each field asserts only what its state can honestly carry: ``tier`` is ``None`` unless the state
is ``present`` (an unread value tiers nothing), and ``reason`` is ``None`` when it IS present
(there is nothing to explain). ``items_seen`` is its own field rather than folded into a token,
the ``BudgetExceeded`` kø-(y) rule: "which shape" and "how many" are two operative questions."""
#: Path of the document the evidence was read from.
file: str
state: EvidenceState
#: The derived trust tier — ``None`` unless ``state`` is ``present``.
tier: TrustTier | None
#: WHY the value could not be read — ``None`` unless ``state`` is ``unreadable``.
reason: ProvenanceReason | None
#: The decoded entries, empty unless ``state`` is ``present``.
entries: tuple[dict[str, str], ...]
#: How many entries were seen, INCLUDING ones that could not be decoded. Part of the
#: ``(state, reason, items_seen)`` triple a discounted concept is reported with.
items_seen: int
#: The ONE provenance key SPEC §5.3 derives a trust tier from. Named rather than spelled inline at
#: the two places that care, so "which key is tiered" cannot answer differently in each of them.
_TIERED_KEY: Final = "verified"
def evidence_for(path: str | Path, key: str = _TIERED_KEY) -> FalsificationEvidence:
"""Read one document's provenance into the three-state answer a falsification verdict needs.
**A library primitive, deliberately NOT wired into ``run_project`` or ``explore``**, mirroring
``promote_verdict`` and ``write_verdict``: the system reads, the caller decides. Wiring it into
the run surface would move the byte-pinned demo transcript, which nothing asks for.
**A tier is derived ONLY for the key SPEC §5.3 tiers.** ``tier`` used to be derived
unconditionally, so ``key="sources"`` — or the agreed segmented form's
``segment_id``/``source_offset`` — crashed in ``trust_tier``, which refuses an entry naming no
``by`` actor. That refusal is right, and it belongs to ``verified`` alone: ``by`` is required of
a verification entry (SPEC §5.2), not of every provenance key. So the guard stays exactly where
it is and this function stops applying it to keys it was never about. Measured at the close of
B4: all 11 call sites in the repo use the default, so the parameter had never been exercised.
Gated by ``tests/test_falsification_verdict_loadbearing.py`` and
``tests/test_evidence_key_parameter_loadbearing.py``."""
result = read_provenance(path, key)
if result is None:
return FalsificationEvidence(
file=str(path), state="absent", tier=None, reason=None, entries=(), items_seen=0
)
if isinstance(result, UnreadableProvenance):
return FalsificationEvidence(
file=str(path),
state="unreadable",
tier=None,
reason=result.reason,
entries=(),
items_seen=result.items_seen,
)
return FalsificationEvidence(
file=str(path),
state="present",
tier=trust_tier(result) if key == _TIERED_KEY else None,
reason=None,
entries=result,
items_seen=len(result),
)
def evidence_notice(evidence: FalsificationEvidence) -> str | None:
"""One line about evidence that could not be used, or ``None`` when there is nothing to say.
Omission, never an empty row (the ``cost_baseline_notice`` precedent). The reason TOKEN is
printed raw rather than translated into prose, so no second display vocabulary exists to drift
from ``ProvenanceReason``."""
if evidence.state == "present":
return None
detail = "" if evidence.reason is None else f"; reason {evidence.reason}"
return (
f"provenance {evidence.state} in {evidence.file}{detail}; items_seen={evidence.items_seen}"
)
def admits_falsification(evidence: FalsificationEvidence) -> bool:
"""The K5 threshold, expressed in ONE place: ``present`` AND a tier above ``unverified``.
An operator decision, not a default. A second copy of a threshold drifts (kø-(p)), and a
threshold spelled inline at each caller is a threshold nobody can find.
Everything it refuses is meant to be REPORTED with ``(state, reason, items_seen)`` and
explicitly discounted — never silently excluded, because a dropped concept and a discounted one
are different facts and only one of them is honest about what was read.
**``author`` / ``usage_count`` / ``last_modified`` are deliberately NOT required, and the
denominator is written down rather than implied:** SPEC §5.1 names SIX entry keys and the
producer writes TWO of them. Requiring keys the producer does not emit would make the threshold
unreachable in practice while looking strict on paper. The threshold names only what is
actually written. See ``docs/okf-konsum-kontrakter.md`` § 1, which is the source for this rule.
Gated by ``tests/test_falsification_verdict_loadbearing.py``."""
# NAMES the tiers that clear it, rather than excluding the one that does not. ``tier`` is now
# ``None`` for any key other than ``verified`` (SPEC §5.3 tiers that key alone), and
# ``None != "unverified"`` is TRUE — so the exclusion spelling would let a present ``sources``
# list clear a threshold about verification. For every value reachable before that change the
# two spellings are identical, which is why this is a tightening and not a new rule.
return evidence.state == "present" and evidence.tier in ("human-reviewed", "machine-confirmed")
#: WHY a provenance value could not be read. The tokens name the SHAPE the value is written in
#: and NOTHING else — the same discipline ``SkipReason`` carries. A block sequence and a block
#: mapping are both CONFORMANT OKF (SPEC §5.2 writes ``verified`` in exactly those forms); they are
#: simply outside the accepted single-line subset, and this decoder is not entitled to an opinion
#: about whether the author erred. ``unsupported-flow`` is the one token that does denote a
#: malformation, and it says so by naming the flow FORM rather than the author.
ProvenanceReason = Literal["block-sequence", "block-mapping", "unsupported-flow"]
@dataclass(frozen=True)
class UnreadableProvenance:
"""A provenance key that IS present and could NOT be read, and why — never silence.
Mirrors ``SkippedLink``: structured rather than a rendered string, for the reason
``BudgetExceeded`` carries ``kind``/``limit``/``observed`` as fields (kø-(y)). "Which shape is
this written in" and "how many entries were there" are two separate operative questions, and a
caller forced to re-parse a token to tell them apart has been handed a diagnostic it cannot act
on. That is why the COUNT is its own field and is never folded into the token.
``value`` is the offending text VERBATIM as written in the file, indentation included — the
operator fixing the document edits that text, and a normalised form would send them looking for
a string their file does not contain."""
#: Path of the document the key was read from.
file: str
#: The frontmatter key that could not be read.
key: str
#: The offending text exactly as it appears in that file.
value: str
reason: ProvenanceReason
#: Entries seen. ``0`` when nothing countable was there — an empty key, or a flow value the
#: decoder refused and therefore never enumerated.
items_seen: int
def read_provenance(
path: str | Path, key: str
) -> tuple[dict[str, str], ...] | UnreadableProvenance | None:
"""Read one provenance key into entries, or say why it could not be read.
Three outcomes, and the three-way split is the whole point:
* ``None`` — the document does not carry the key. **Absence is ``None``, never a default**:
the F2 principle one layer down, where an unread signal must not become an asserted absent
one.
* a tuple of entries — the value was in the accepted flow subset and decoded.
* ``UnreadableProvenance`` — the key is THERE and could not be read, with the shape and the
count that say what was seen.
Driven by ``_split_frontmatter``: this is a second READER of the one parse, never a second
parser. Only top-level (unindented) frontmatter lines are considered as key sites, so an
indented ``by:`` inside a block entry can never be mistaken for a document-level key.
Gated by ``tests/test_provenance_decoder_loadbearing.py``."""
lines = _split_frontmatter(Path(path).read_text(encoding="utf-8"))[0]
for i, line in enumerate(lines):
if line[:1] in (" ", "\t"):
continue
name, sep, raw = line.partition(":")
if not sep or name.strip() != key:
continue
value = raw.strip()
if value:
try:
return decode_flow_value(value, key=key)
except FlowDecodeError:
return UnreadableProvenance(
file=str(path), key=key, value=value, reason="unsupported-flow", items_seen=0
)
continuation = list(itertools.takewhile(lambda ln: ln[:1] in (" ", "\t"), lines[i + 1 :]))
if not continuation:
return UnreadableProvenance(
file=str(path), key=key, value="", reason="block-mapping", items_seen=0
)
# An ITEM is a continuation line whose STRIPPED form opens with ``- ``. A ``- `` occurring
# inside a value is text, not an item, and counting it would inflate the number the caller
# acts on.
items = sum(1 for ln in continuation if ln.strip().startswith("- "))
# P13b: the block sequence of mappings is SPEC §5.1's own example and the form all four
# delivered bases actually write (4605/4605, measured). It is decoded by the second carrier
# of the one grammar; a block MAPPING (no item opened) and anything the grammar refuses
# still come back unreadable, which is what keeps "we could not read it" from quietly
# becoming "there was nothing to read".
block = decode_block_mappings(continuation, key=key)
if block is not None:
return block
return UnreadableProvenance(
file=str(path),
key=key,
value="\n".join(continuation),
# SPEC §5.2's one-element MUST: a bare mapping IS one entry, so a mapping that is
# present counts 1 rather than 0 — 0 is reserved for "nothing was there at all".
reason="block-sequence" if items else "block-mapping",
items_seen=items or 1,
)
return None
@dataclass(frozen=True)
class Bundle:
"""A navigated OKF bundle: ``index.md`` plus every cross-linked file that resolves — and, in
``skipped``, every cross-link that did not.
``skipped`` DEFAULTS to the empty tuple, and the default is the honest reading rather than a
convenience: an empty trace is a positive statement ("every cross-link was followed"), in the
same class as ``ProvenanceStamp.external_calls`` ("nothing outside this process was contacted").
That is the opposite of ``ProvenanceStamp.cost_baseline_anchored``, which is REQUIRED precisely
because both of its defaults would lie. The difference is what each absent value would assert:
a missing bool has to claim something about an event, while a missing trace asserts only that
the event list is empty — which is exactly what a construction with no skips means."""
dir: str
files: tuple[BundleFile, ...]
#: Every link navigation could not follow, in walk order. Read by ``run`` to render the one line
#: a run prints about its own reachability; NEVER read by ``bundle_context``, whose rendering is
#: built from ``index_summary`` + ``context_files`` alone — which is what keeps the commons-owned
#: nav-golden fasit byte-identical.
skipped: tuple[SkippedLink, ...] = ()
@property
def index_summary(self) -> str:
"""The ROOT index body — the progressive-disclosure entry point (not whole-bundle stuffing).
Bound to the root alone: a nested ``a/index.md`` is navigation, never summary prose."""
return next((f.body for f in self.files if f.name == _INDEX_NAME), "")
@property
def verdicts(self) -> list[BundleFile]:
"""Every ``type: verdict`` file (the ExpeL seeds the Step-1 wiring retrieves)."""
return [f for f in self.files if f.type == _VERDICT_TYPE]
@property
def context_files(self) -> list[BundleFile]:
"""The non-index, non-``verdict`` concept files — the bodies that form the agent context.
The verdict layer is deliberately EXCLUDED: prior verdicts reach the hypothesis prompt only
through the gated ExpeL fold, never by stuffing them into the read-context (målbilde §4).
The exclusion is a TYPE CHECK on each reached file, applied at EVERY level — never a
property of the link graph, so a mislabelled or injected navigation edge cannot smuggle a
nested verdict in. Index files are dropped by BASENAME at every level too: a nested
``a/index.md`` is navigation, not content."""
return [
f
for f in self.files
if posixpath.basename(f.name) != _INDEX_NAME and f.type != _VERDICT_TYPE
]
@property
def hypothesis(self) -> BundleFile | None:
"""The candidate ``type: hypothesis`` file, if present."""
return next((f for f in self.files if f.type == "hypothesis"), None)
def _load_file(bundle_dir: str, name: str) -> BundleFile | None:
"""Resolve the bundle-relative ``name`` within ``bundle_dir`` and read it, or ``None`` if
missing / escaping the bundle (OKF §4 broken-link tolerance + fail-closed path-safety)."""
try:
resolved = Path(safe_resolve(bundle_dir, name))
except PathSecurityError:
return None
if not resolved.is_file():
return None
text = resolved.read_text(encoding="utf-8")
fm = _frontmatter_from_text(text)
return BundleFile(
name=name, type=fm.get("type", ""), frontmatter=fm, body=_body_from_text(text)
)
def _resolve_target(bundle_dir: str, from_name: str, target: str) -> tuple[str, str] | None:
"""Resolve one cross-link into ``(bundle-relative posix name, canonical path)``, or ``None``
when it fails to resolve for ANY reason (escape, invalid path component) — the caller skips,
never raises (method-spec §3 Step 1).
A leading ``/`` denotes the BUNDLE ROOT, not the filesystem root: an implementation that let
``os.path.join`` see an absolute target would either escape to the real filesystem path or
refuse a legitimate root-relative link. Any other form resolves against the LINKING FILE's own
directory, so an index links its immediate children one segment at a time."""
if target.startswith("/"):
rel = posixpath.normpath(target.lstrip("/"))
else:
rel = posixpath.normpath(posixpath.join(posixpath.dirname(from_name), target))
try:
return rel, safe_resolve(bundle_dir, rel)
except PathSecurityError:
return None
def _walk(
bundle_dir: str,
current: BundleFile,
files: list[BundleFile],
seen: set[str],
skipped: list[SkippedLink],
) -> None:
"""Follow ``current``'s cross-links depth-first in first-seen order, appending each newly
reached file and recursing into it. De-duplication is on the CANONICAL RESOLVED path (so
``./a.md``, ``a.md`` and ``/a.md`` are one entry), which is also what terminates cycles.
A link that cannot be followed is still SKIPPED, never raised (OKF §4) — the tolerance is the
spec — but it is now RECORDED in ``skipped``, with the reason distinguishing the two cases.
The dedup branch records NOTHING: a repeated link and a cycle are correct navigation, and an
implementation that logged every ``continue`` would report a healthy bundle as half-unread.
A caller-owned accumulator rather than a return value, for the reason ``generate``'s
parse-failure sink is one: the recursion is depth-first over an unbounded tree, so every frame
appends into the SAME list and the walk's shape stays unchanged."""
for target in _LINK_RE.findall(current.body):
resolved = _resolve_target(bundle_dir, current.name, target)
if resolved is None:
# The target left the bundle. Often deliberate (a link to a neighbouring base), so it is
# reported rather than refused — the tolerance is unchanged.
skipped.append(
SkippedLink(from_file=current.name, target=target, reason="outside-bundle")
)
continue
rel, canonical = resolved
if canonical in seen:
continue # de-duplication / cycle termination: correct navigation, NOT a skip
seen.add(canonical)
linked = _load_file(bundle_dir, rel)
if linked is None:
# In-bundle, but nothing readable is there: broken link, tolerated, never raised (§4).
# Recorded once per resolved target — the ``seen`` entry above absorbs repeats.
skipped.append(SkippedLink(from_file=current.name, target=target, reason="missing"))
continue
files.append(linked)
_walk(bundle_dir, linked, files, seen, skipped)
def navigate_bundle(bundle_dir: str) -> Bundle:
"""Navigate the OKF bundle from ``index.md``: parse the root index, then follow intra-bundle
``.md`` cross-links RECURSIVELY, depth-first in first-seen link order, reading each reached
file's frontmatter + body. Fully deterministic. Broken / escaping links are skipped (§4) — and
RECORDED on the returned ``Bundle.skipped``, so "this document was never written" and "the link
to it was wrong" stop looking identical from the outside.
Navigation follows LINKS ONLY — a directory is never enumerated. Hence the missing-``index.md``
error binds the bundle ROOT alone (a bundle has no entry point without it); an intermediate
directory reached by a link needs no ``index.md`` of its own, and content nothing links to is
simply unreachable, not an error. Raises ``ValueError`` only for the unreadable root index."""
index = _load_file(bundle_dir, _INDEX_NAME)
if index is None:
raise ValueError(f"OKF bundle has no readable {_INDEX_NAME}: {bundle_dir!r}")
files: list[BundleFile] = [index]
skipped: list[SkippedLink] = []
root = _resolve_target(bundle_dir, _INDEX_NAME, _INDEX_NAME)
seen = {root[1]} if root is not None else set()
_walk(bundle_dir, index, files, seen, skipped)
return Bundle(dir=bundle_dir, files=tuple(files), skipped=tuple(skipped))
class BundleIdMismatch(ValueError):
"""A base cannot say what it is: two of its CONCEPTS declare different ``bundle_id`` values.
**This used to mean something else, and the change is an operator decision (S7a-3 pkt. 1).**
Until 2026-09-03 it also covered a declared id that disagreed with the directory the base was
mounted under. That refusal was measured against the first delivered corpus that declares its
own id (K2: 618 of 630 concept files, mounted under a different name) and it made the base
unopenable at every door, with re-mounting by hand as the only remedy. The mount is a filesystem
accident; the declaration is the artefact speaking. A disagreement between them is now RECORDED
(``ResolvedBundleId.mount`` + ``ProvenanceStamp.bundle_id_source`` + one warning line), never
refused.
What is left is the disagreement no mount name could ever settle: within ONE base, two concepts
naming two different corpora. There is no fallback that makes that base coherent, and an
artefact stamped from it would name one of the two at random.
A ``ValueError`` DELIBERATELY: it must land on ``run.main``'s refusal tuple and on hosting's
400 arm rather than on the crash channel. ``ExplorationError`` is a ``RuntimeError`` and would
give an operator a traceback and an automated caller a 500 — a configuration mistake dressed
as a server fault.
"""
#: Which source answered. THREE values, not a boolean (operator decision B1): a caller that cannot
#: tell "the concept said so" from "we fell back twice" has been handed a stamp it cannot audit.
BundleIdOrigin = Literal["declared-concept", "declared-index", "mount-derived"]
_BUNDLE_ID_KEY = "bundle_id"
@dataclass(frozen=True)
class ResolvedBundleId:
"""A base's id, WHERE it came from, and the mount it was read from.
``origin`` is REQUIRED WITHOUT DEFAULT, for the reason ``ProvenanceStamp.cost_baseline_anchored``
is: both defaults would lie about an event. Defaulting to ``"mount-derived"`` would let a
resolver that never read the base claim it had; defaulting to a declared value would claim a
declaration that never happened.
``mount`` is CARRIED rather than discarded once the declaration wins (S7a-3 pkt. 1): the
warning a run prints names both, and a caller reading an artefact months later needs to know
which directory the corpus was read from. A resolver that kept only the winner could report
that something was overridden without being able to say what.
"""
id: str
origin: BundleIdOrigin
mount: str
def _declared_bundle_id(frontmatter: dict[str, str]) -> str:
"""The declared id, de-quoted through ``unquote_scalar`` — this repo's ONE de-quoting rule."""
return unquote_scalar(frontmatter.get(_BUNDLE_ID_KEY, "")).strip()
def reconcile_bundle_id(
bundle_dir: str | Path, *, concept_name: str | None = None
) -> ResolvedBundleId:
"""Resolve a base's ``bundle_id`` against the mount it was opened from — the ONE rule.
``bundle_id`` denotes the corpus **as a mounted artefact**, never the production event that made
it (operator decision D6). That denotation cannot be changed once artefacts bear the stamp,
which is why it is written down here rather than left to each call site.
Resolution order is operator decision B1 — identity is the pair ``(bundle_id, concept_id)``, and
the CONCEPT's own frontmatter is consulted first:
1. ``concept_name``'s frontmatter, when a concept is named and declares the key →
``"declared-concept"``;
2. the root ``index.md``'s frontmatter → ``"declared-index"``;
3. the mount's basename → ``"mount-derived"``.
A declared id that DISAGREES with the mount is NOT an error (operator decision, S7a-3 pkt. 1):
the declaration wins, the mount is carried alongside on ``ResolvedBundleId.mount``, and the
caller that opened the base is the one that reports it. What still refuses is a base whose
CONCEPTS disagree with each other — see ``assert_declared_ids_agree``, which is a separate,
unconditionally called check rather than a branch here, so a door that forgot it fails a test
of its own instead of quietly skipping a refusal.
A base whose root ``index.md`` cannot be read is **unknown, not undeclared**:
``navigate_bundle``'s fail-fast propagates unchanged, because reading an unreadable base as
"it declares nothing" would widen the answer on missing evidence.
MEASURED 2026-09-02: no file under ``shared/`` declares the key (zero ``^bundle_id`` hits
against a known-positive control of 31 files carrying ``^type:``), so every base in this repo
resolves ``mount-derived`` today and both declared branches are DEFENSIVE — the first corpus in
the wild to take a declared branch was K2 (measured 2026-09-03, 619 declaring files).
**Honesty limit, stated:** without ``concept_name`` the root ``index.md`` answers, so a base
whose index declares X while its concepts declare Y resolves to X. That is B1's resolution
ORDER, unchanged here; the agreement check below is about concepts colliding with each other.
:raises ValueError: the root ``index.md`` is missing or unreadable.
"""
root = str(bundle_dir)
mount = Path(root).name
if concept_name is not None:
concept = _load_file(root, concept_name)
if concept is None:
raise ValueError(f"OKF bundle has no readable concept {concept_name!r}: {root!r}")
declared = _declared_bundle_id(concept.frontmatter)
if declared:
return ResolvedBundleId(id=declared, origin="declared-concept", mount=mount)
index = _load_file(root, _INDEX_NAME)
if index is None:
raise ValueError(f"OKF bundle has no readable {_INDEX_NAME}: {root!r}")
declared = _declared_bundle_id(index.frontmatter)
if declared:
return ResolvedBundleId(id=declared, origin="declared-index", mount=mount)
return ResolvedBundleId(id=mount, origin="mount-derived", mount=mount)
def assert_declared_ids_agree(bundle: Bundle) -> None:
"""Refuse a navigated base whose CONCEPTS declare two different ``bundle_id`` values.
This is the collision the slacken (S7a-3 pkt. 1) deliberately keeps: a declared id that
disagrees with the MOUNT is a filesystem accident and is now recorded rather than refused, but
two concepts inside one base naming two different corpora is a base that cannot say what it is.
No fallback settles it, and every artefact stamped from it would name one of the two at random.
**Concepts only — the root ``index.md`` is NOT in the set, and that is decision B1 applied a
second time.** Concept-beats-index is a PRECEDENCE rule, so an index out of step with its
concepts is the fallback losing, not a collision. Folding the index in would newly refuse
exactly the K2-shaped bases the slacken exists to admit.
Reads ``context_files``, which is also what drops the ``type: verdict`` layer and nested
``index.md`` at every level — the same property ``read_bundle`` is built from, so the two
cannot disagree about which files count as concepts.
A SEPARATE function rather than a branch inside ``reconcile_bundle_id``: it needs a navigated
bundle, and ``reconcile_bundle_id`` must stay pure (``explore._bundle_index`` resolves ids for
directories that may not exist). Being separate also gives it its own mutation — detach the call
at a door and that door's own arm goes red, instead of a refusal quietly not happening.
:raises BundleIdMismatch: two concepts declare different ids.
"""
declared: dict[str, str] = {}
for f in bundle.context_files:
value = _declared_bundle_id(f.frontmatter)
if value:
declared.setdefault(value, f.name)
if len(declared) > 1:
named = ", ".join(f"{value!r} (in {name!r})" for value, name in sorted(declared.items()))
raise BundleIdMismatch(
f"knowledge base {bundle.dir!r} cannot say what it is: its concepts declare "
f"{len(declared)} different bundle_id values — {named}; an approach names a base by "
"that id, so a run would evaluate against one corpus and report the other"
)
def in_dimension(file: BundleFile, dimension: str | None) -> bool:
"""Whether a concept file belongs to a run scoped to ``dimension`` (§4.1a).
ONE copy of the rule (kø-(p)), because two renderings now answer it: ``bundle_context`` renders
the matched bodies, and ``directory_listing`` lists the matched documents for a navigator that
reads them one at a time. Two copies of "which knowledge is in scope" would be free to disagree
about the same base, and the disagreement would show up as an agent being shown a document it
is then refused.
``dimension=None`` admits everything (byte-identical to the unscoped rendering), and a file
carrying NO ``dimension`` is never dropped: un-scoped knowledge belongs to every scope, which
is what makes a method or a cost reference usable across dimensions.
"""
if dimension is None:
return True
declared = file.frontmatter.get("dimension")
return declared is None or declared == dimension
def declares_verdict_type(path: str | Path) -> bool:
"""Whether the document AT ``path`` declares the ``type: verdict`` layer.
Deliberately keyed on the DOCUMENT rather than on the walk, and that is the whole difference
between this gate and a listing filter. ``Bundle.verdicts`` answers "which judgements did
navigation reach", which is the right question for the ExpeL seeds; the gate's question is the
other one — "may these bytes leave" — and it is asked about a path a model chose, which no
listing gave it. A verdict file no index links to is absent from the walk and would sail
straight through a navigation-keyed check.
Reads the same frontmatter ``_load_file`` reads, through the same one-scan splitter, so a file
the walk WOULD have classified as a verdict is classified identically here. A missing file
raises where it always did, in the read.
"""
return parse_frontmatter(path).get("type", "") == _VERDICT_TYPE
class BundlePathNotFound(ValueError):
"""A listing was asked for a directory the navigated bundle does not have.
A ``ValueError``, the ``BundleIdMismatch`` precedent: it must land on the CLI's refusal tuple
and hosting's 400 arm rather than the crash channel. Refusing is the point — an unknown path
rendered as an empty listing is indistinguishable from a directory that exists and holds
nothing, and the caller is a model choosing a path out of a previous listing.
"""
class DocumentPathRefused(ValueError):
"""``read_dir`` was asked for a DOCUMENT — the wrong rung of the navigation ladder.
The symmetry of ``explore.DirectoryPathRefused``, which has answered the other direction since
the live K2 run of session 95: ``read_file`` on a directory names ``read_dir``, while
``read_dir`` on a document named neither the rung that reads it nor the path it would take. A
refusal that only says "no" leaves the caller — a model choosing a path — with the same next
move it just made.
A ``ValueError``, the ``BundlePathNotFound``/``DimensionScopeRefused`` precedent, and a SIBLING
of ``BundlePathNotFound`` rather than a subclass: "this path is a document" and "this path is
nothing" are different facts, and a caller switching on the first must not be answered by the
second.
**Not the class the live two rounds fell into** (``docs/2026-09-07-syretest-s7-prepass-k2.md``
§ 4): there the path named no document at all — it was a document's name with its ``inbox-``
prefix dropped — and it is still refused as unknown. Naming the nearest look-alike would be
guessing what a caller meant, which is invention rather than validation.
"""
#: P18/A1 — the DEFAULT number of concept documents one ``read_dir`` answers with.
#:
#: CHOSEN BY MEASUREMENT, against the ceiling S7a-3 set (1 500 characters per listing). Measured
#: 14.09 over the four delivered vegnormal bases: one document entry is 121-209 characters (median
#: 145), and the worst level is ``krav/N200`` with 1 132 documents at 169 974 characters. Ten
#: entries of the WORST measured size is 2 090 characters of entries; ten of the median size is
#: 1 450 — so ten is the largest round window that keeps a default listing of the worst measured
#: level in the neighbourhood of the ceiling instead of two orders of magnitude above it.
#:
#: A default page of ten out of 1 132 is not meant to be browsed to the end: ``total`` says how many
#: there are and ``filter`` is the rung's answer to "which ones". That is the point — before this,
#: the price of finding out WHAT is at a level was set by how much is at it.
_DIRECTORY_PAGE_DEFAULT: Final = 10
#: The largest window a CALLER may ask for. A model-chosen limit is clamped to it rather than
#: refused: the caller asked for a listing, and answering with more than this is the cost the
#: pagination exists to bound.
#:
#: 50 is measured, not round: 50 entries of the worst measured size is ~10 400 characters and of the
#: median ~7 250 — the same order as the 6 073-character level S7a-3 already measured and accepted
#: as the honest price of a bundle-relative path, and 16x below the 169 974 a single unbounded call
#: cost. No single call can therefore cost O(corpus): the worst it can cost is O(window).
_DIRECTORY_PAGE_MAX: Final = 50
#: The frontmatter keys ``filter`` matches against, besides the title. TOP-LEVEL keys, read straight
#: off ``BundleFile.frontmatter``: since P15 (f13dc64) a top-level key wins over an indented one of
#: the same name, so this IS the concept's own value and a second "own frontmatter" reader here
#: would be the second copy of one rule that kø-(p) forbids. MEASURED on the delivered bases:
#: ``req_number`` on the N corpora ("Krav 4.1.2—1"), ``prosessnr`` on R761 ("'11.11'", quoted —
#: hence ``unquote_scalar``, this repo's ONE de-quoting rule).
_FILTER_FIELDS: Final = ("req_number", "prosessnr")
#: Which frontmatter keys DECLARE a reference number — the base's own vocabulary of requirement and
#: process numbers. Derived from ``_FILTER_FIELDS`` rather than restating it (two copies of the two
#: measured keys is the kø-(p) drift), plus ``seksjon``, which is the field that carries the BARE
#: form-3 number on the N corpora and on R761 (``seksjon: '10.4.3'``, ``seksjon: '11.11'``) while
#: ``req_number`` carries the composed ``Krav 10.4.3—1``. MEASURED 15.09 over the four delivered
#: bases: n100 553 distinct declared numbers, n200 1 440, n500 365, r761 2 765.
#:
#: NOT added to ``_FILTER_FIELDS`` itself, and that is deliberate: what a ``filter`` word searches
#: is measured and gated (P18), and widening it would change ``total_matches`` for every navigator
#: call. These two constants answer different questions — "what does a filter look in" and "what
#: does this document declare as its number" — over ONE list of the measured reference keys.
REFERENCE_NUMBER_FIELDS: Final = (*_FILTER_FIELDS, "seksjon")
def reference_number(file: BundleFile) -> str:
"""The document's OWN reference number as its top-level frontmatter states it, or ``""``.
First non-empty of ``_FILTER_FIELDS``, in that order: the N corpora declare ``req_number``
("Krav 4.1.2—1") and R761 declares ``prosessnr`` ("'11.11'", quoted — hence ``unquote_scalar``),
and MEASURED no delivered document declares both. ``seksjon`` is deliberately NOT read here:
this answers "which requirement IS this", and a section number names the chapter a requirement
sits in, not the requirement.
Read off ``BundleFile.frontmatter``, so P15's top-level-wins rule applies and a nested
``sources:`` entry can never answer for the concept.
"""
for key in _FILTER_FIELDS:
value = unquote_scalar(file.frontmatter.get(key, ""))
if value:
return value
return ""
def declared_reference_numbers(file: BundleFile) -> tuple[str, ...]:
"""Every reference number this ONE document declares (``REFERENCE_NUMBER_FIELDS``), de-quoted.
The unit of the reference VOCABULARY the validator's stage 0b checks a requirement-shaped code
against (P20/B). Per DOCUMENT rather than per base, because the caller composing the grounding
already walks the base once and the boundaries it composes are the ones the gate must read
(P18/B1's rule: one document is one unit, never a blob).
"""
return tuple(
value
for key in REFERENCE_NUMBER_FIELDS
if (value := unquote_scalar(file.frontmatter.get(key, "")))
)
def _matches_filter(file: BundleFile, needle: str) -> bool:
"""Case-insensitive SUBSTRING over the document's title and its reference number.
A substring and not a pattern, for ``_ground_against_input``'s reason one rung down: the caller
is a model choosing words, and a form the rule does not know would silently return nothing —
an empty listing that reads like "the base does not have this". Substring fails toward showing
MORE, which a navigator can narrow; a pattern fails toward showing nothing, which it cannot.
``description`` is deliberately NOT searched: measured, every concept in all four bases carries
one, and matching prose would return most of a level for most words — the filter would look like
it worked while bounding nothing.
"""
hay = [unquote_scalar(file.frontmatter.get("title", file.name))]
hay.extend(unquote_scalar(file.frontmatter.get(k, "")) for k in _FILTER_FIELDS)
return any(needle in value.casefold() for value in hay)
#: P21/C2 — how many neighbouring directories a "no such path" refusal may name. A refusal's job is
#: to hand back the one thing the caller can act on, not to re-list the level: five is enough to
#: show the SHAPE of the names this base uses (``4``/``41``/``42`` rather than ``4-3``), and the
#: full level is one ``read_dir`` away.
_NEIGHBOUR_LIMIT: Final = 5
def nearest_listable_directory(bundle: Bundle, path: str, *, dimension: str | None = None) -> str:
"""The deepest ANCESTOR of ``path`` that ``directory_listing`` will actually answer for.
Chosen off the NAVIGATED ``context_files`` rather than off the filesystem, for two reasons that
are the same reason: a directory can exist on disk and hold no navigated concept (nothing links
it), in which case ``read_dir`` refuses it and the refusal would have handed the caller a path
that does not resolve — ``_index_excerpt``'s rule one rung up, a path that never was is worse
than no path. And ``context_files`` is the property that drops the ``type: verdict`` layer, so a
refusal can never advertise by name the one layer no listing mentions.
Falls back to ``""``, the base's own top level, which ``directory_listing`` always answers.
"""
reachable = [f for f in bundle.context_files if in_dimension(f, dimension)]
segments = path.strip("/").split("/")
for depth in range(len(segments) - 1, 0, -1):
candidate = "/".join(segments[:depth])
if any(f.name.startswith(candidate + "/") for f in reachable):
return candidate
return ""
def nearest_subdirectories(
bundle: Bundle,
path: str,
*,
dimension: str | None = None,
limit: int = _NEIGHBOUR_LIMIT,
) -> tuple[str, ...]:
"""The directories a caller who named a path this base does not hold could have meant (P21/C2).
**The measured defect.** Over round 4's six traces, 16 of 105 ``read_dir`` calls and 2 of 38
``read_file`` calls named a path the base does not hold, and eleven of those were one run
walking ``R761/4-3``, ``R761/4.3``, ``R761/4-2``, ``R761/4-1``, ``R761/4-0``, ``R761/4-5``,
``R761/4-6`` — guessing at a chapter-number spelling the corpus does not use, while the real
neighbours are ``R761/4``, ``R761/41``, ``R761/42``. The refusal already named the nearest
LISTABLE ancestor, which is the right rung; what it could not say is which of that rung's names
the caller was reaching for.
**Every name it returns RESOLVES.** Built from ``context_files`` and through the SAME
``in_dimension`` predicate the listing uses: a suggestion read off the filesystem could name a
directory ``read_dir`` then refuses, and one built from ``files`` could name the ``type:
verdict`` layer by path — advertising in a refusal the one layer no listing mentions.
**Ranked by longest common prefix with the segment that failed, then shortest, then name.** A
ranking cannot refuse anything — this is help text on a refusal — so its failure direction is
benign: at worst it names five real directories that are not the one meant. Prefix ranking is
what puts ``4`` ahead of ``41`` for ``4-3``; with no common prefix at all every candidate ties
and the order degrades to "the shortest names at this level", which is an honest "here is what
IS here".
"""
ancestor = nearest_listable_directory(bundle, path, dimension=dimension)
prefix = f"{ancestor}/" if ancestor else ""
depth = len(prefix.split("/")) - 1 if prefix else 0
segments = path.strip("/").split("/")
missing = segments[depth] if len(segments) > depth else ""
children: set[str] = set()
for file in bundle.context_files:
if not in_dimension(file, dimension) or not file.name.startswith(prefix):
continue
rest = file.name[len(prefix) :].split("/")
if len(rest) > 1:
children.add(rest[0])
def _shared(name: str) -> int:
n = 0
for a, b in zip(name, missing):
if a != b:
break
n += 1
return n
ranked = sorted(children, key=lambda n: (-_shared(n), len(n), n))
return tuple(f"{prefix}{name}" for name in ranked[:limit])
def directory_listing(
bundle: Bundle,
path: str = "",
*,
dimension: str | None = None,
filter: str | None = None,
offset: int = 0,
limit: int | None = None,
) -> dict[str, Any]:
"""One LEVEL of a navigated bundle: the subdirectories under ``path`` with what each holds, and
the concept documents that sit directly in it.
The rung between ``list_bundles`` (which bases exist) and ``read_file`` (what one document
says). Before it existed, ``read_bundle`` answered with EVERY concept document in the base —
measured on K2 at 42 761 o200k tokens riding in 7 of 12 prompts, 90 % of the run — because the
price of finding out what a base contains was set by how much it contains.
**The tree is derived from ``context_files`` NAMES, never from ``index.md`` files.** Every name
is already the full bundle-relative posix path, so the shape was never lost — only the
rendering flattened it, which is why ``bundle_context`` and both nav-goldens are byte-identical
after this was added. Deriving from names also means a directory whose index was never linked is
still visible, and that the two renderings of one bundle cannot disagree about what a concept is.
``context_files``, NEVER ``files``: it is the property that drops the ``type: verdict`` layer
AND nested ``index.md`` at every level. A listing built from ``files`` would put prior verdicts
in front of the navigator around the gated ExpeL fold (målbilde §4) — and a directory COUNT
built from ``files`` would advertise documents the navigator is not allowed to be shown.
**Every path is BUNDLE-RELATIVE, i.e. usable verbatim as the next call's argument.** A level-
relative name would have to be composed by the caller, and the caller is a model: a path that
never existed is worse than no path (``_index_excerpt``'s rule, one rung up).
``dimension`` scopes the listing exactly as it scopes ``bundle_context`` — ONE predicate
(``in_dimension``) serves both, so a navigator is never shown a document the run would then
refuse to open. Under a scope a directory holding only foreign-dimension documents raises
``BundlePathNotFound`` like any other unknown path: within this run it holds nothing, and
answering with an empty listing is the very confusion the refusal exists to prevent.
``documents`` on a directory entry is the count of concept documents in its whole SUBTREE — what
the subtree holds, not what one ``read_dir`` on it returns. It is the price signal a navigator
chooses against, and the tool description says which of the two it is rather than leaving the
reader to guess.
**P18/A1 — the answer is a WINDOW, so one listing costs O(window) and never O(level).** S7a-3
bound the cost to entries at ONE level rather than documents in the base; P16 then measured what
one level costs on a delivered corpus: ``krav/N200`` 169 974 characters over 1 132 documents,
``krav/N100`` 69 250 over 445, and R761's root 110 912 over 2 728 SUBDIRECTORIES — which is why
the window covers both kinds and not only documents. ``offset``/``limit`` page through
directories first, then documents; ``total`` is the denominator the window is taken from.
After: 479-1 537 characters for a default listing of each of those four levels.
**P18/A2 — ``filter`` is how a navigator asks "which ones", instead of paging to find out.**
Case-insensitive substring over a document's title and reference number (``req_number`` /
``prosessnr``) and over a directory's path. A filter that matches nothing answers with an empty
window and ``total_matches: 0`` — never a refusal: "no document here is about X" is an answer,
and refusing it would make an honest negative indistinguishable from a path that does not exist.
:raises BundlePathNotFound: no navigated concept document lives under ``path``.
"""
prefix = "" if path in ("", ".") else path.strip("/") + "/"
needle = None if filter is None else filter.casefold()
directories: dict[str, int] = {}
at_level: list[BundleFile] = []
matched_documents: list[dict[str, Any]] = []
for f in bundle.context_files:
if not f.name.startswith(prefix) or not in_dimension(f, dimension):
continue
rest = f.name[len(prefix) :]
head, sep, _ = rest.partition("/")
if sep:
directories[prefix + head] = directories.get(prefix + head, 0) + 1
continue
at_level.append(f)
if needle is not None and not _matches_filter(f, needle):
continue
matched_documents.append(
{
"name": f.name,
# ``or "document"`` mirrors ``bundle_context``'s own fallback for a file with no
# declared type, so the two renderings cannot disagree about it.
"type": f.type or "document",
"title": unquote_scalar(f.frontmatter.get("title", f.name)),
# What the next rung COSTS, in the unit the ceiling is measured in. A navigator
# that cannot see the price cannot choose against a budget.
"chars": len(f.body),
}
)
# A directory has no frontmatter, so its PATH is all there is to match on — and it is what a
# navigator filtering an R761 level for "65" means. MEASURED: without this, ``read_dir`` on
# R761's root answered with 2 728 directory entries at 110 912 characters, a bigger unbounded
# call than the 69 250-character document level this order was written for.
matched_directories = [
{"path": name, "documents": count}
for name, count in sorted(directories.items())
if needle is None or needle in name.casefold()
]
if prefix and not directories and not at_level:
# The wrong RUNG, answered as such — the direction ``explore.DirectoryPathRefused`` already
# covers, measured absent here (F3). Built from ``context_files`` and through the SAME
# ``in_dimension`` predicate the listing above uses: a lookup over ``files`` would name a
# ``type: verdict`` document by path, advertising in a refusal the one layer no listing
# mentions, and one that ignored the scope would name a document this run would then refuse
# to open. The document's REAL name is quoted, never the caller's path, so what the refusal
# hands back resolves.
stem = path.strip("/")
candidate = stem if stem.endswith(".md") else stem + ".md"
named = next(
(f for f in bundle.context_files if f.name == candidate and in_dimension(f, dimension)),
None,
)
if named is not None:
raise DocumentPathRefused(
f"{path!r} in knowledge base {bundle.dir!r} is a document, not a directory; "
f"use read_file to read {named.name!r} whole"
)
# P21/C2: name the rung the caller could have meant. The SAME two helpers the read_file
# refusal uses (kø-(p)) — one question about one base must not have two answers — and both
# are built from ``context_files`` through ``in_dimension``, so every name handed back is
# one this same function would then answer for.
neighbours = nearest_subdirectories(bundle, path, dimension=dimension)
ancestor = nearest_listable_directory(bundle, path, dimension=dimension)
nearby = f"; its subdirectories include {', '.join(neighbours)}" if neighbours else ""
raise BundlePathNotFound(
f"knowledge base {bundle.dir!r} has no directory {path!r}; it holds no concept "
f"document under that path. Nearest directory that does: {ancestor!r}{nearby}"
)
# A1: the WINDOW. Clamped, never refused — a caller asking for more than the maximum asked for
# a listing, and the bound is this rung's job to keep, not the caller's to remember. A negative
# or absurd offset lands past the end and answers with an empty window over an honest ``total``,
# which is what "there is nothing here" looks like when the denominator is stated.
#
# ONE window over BOTH kinds, directories first, because a level is one thing to page through:
# two independent windows would make "show me the next ten" a question with two answers.
window = (
_DIRECTORY_PAGE_DEFAULT if limit is None else max(0, min(int(limit), _DIRECTORY_PAGE_MAX))
)
start = max(0, int(offset))
page = (matched_directories + matched_documents)[start : start + window]
listing: dict[str, Any] = {
"path": path,
"directories": [e for e in page if "path" in e],
"documents": [e for e in page if "name" in e],
# The DENOMINATOR, always: how many entries — subdirectories plus concept documents — this
# level holds within this run's scope. Carried whether or not a filter narrowed the answer,
# because a window without a total is a measurement without a denominator, which is the one
# thing every other count in this repo refuses to be (ansikt 4).
"total": len(directories) + len(at_level),
"offset": start,
"limit": window,
}
if filter is not None:
# A SECOND fact, not a second copy of the first: ``total`` says what is here, this says how
# many of it the filter admitted. A navigator reading only one of the two cannot tell a
# filter that was too narrow from a level that is nearly empty.
listing["filter"] = filter
listing["total_matches"] = len(matched_directories) + len(matched_documents)
return listing
def bundle_context(bundle: Bundle, *, dimension: str | None = None) -> str:
"""Render a navigated bundle as agent read-context via progressive disclosure: the ``index.md``
summary, then each concept file as ``## {type}: {title}\\n{body}``. ``type: verdict`` files are
EXCLUDED (målbilde §2 step 1 / §4: navigation, not chunk-stuffing — the verdict layer folds in
only via the gated ExpeL retrieval). Deterministic: index first, then context files in
navigation order; empty sections are dropped.
When ``dimension`` is given, only concept files whose frontmatter ``dimension`` matches — or that
carry no ``dimension`` at all (un-scoped knowledge is never dropped) — are rendered; the default
``dimension=None`` renders every concept file, byte-identical to the prior behavior. ``dimension``
is a plain ``str`` (not the ``Dimension`` type) so ``okf`` stays MAF-free and import-cycle-free.
Rendering is FLAT regardless of nesting depth (method-spec §3 Step 1): directory structure is
navigation, not presentation, so a nested concept file renders as the same
``## {type}: {title}`` section a root file would — there is no level heading, and nested index
bodies do not appear at all. The serialisation (heading, blank line, body; sections separated by
one blank line) is what the commons nav-golden ``expected-read-context.md`` fasit compares
against — see ``tests/test_okf.py`` nav-golden gates."""
sections = [bundle.index_summary.strip("\n")]
for f in bundle.context_files:
if not in_dimension(f, dimension):
continue
title = unquote_scalar(f.frontmatter.get("title", f.name))
body = f.body.strip("\n")
sections.append(f"## {f.type or 'document'}: {title}\n\n{body}")
return "\n\n".join(s for s in sections if s.strip())
def render_frontmatter(frontmatter: dict[str, str]) -> str:
"""Render a frontmatter dict as ``key: value`` lines (the inverse direction of
``parse_frontmatter``, used by the Step-8 promotion writer). Scalar values are **single-lined**
(every newline/CR collapses to a space) because ``parse_frontmatter`` is line-oriented and stops
at the first ``---`` line — a multi-line value would otherwise corrupt the block or terminate it
early. NOT a bijection: this only guarantees that the single-line fields it writes re-parse to
the same strings; ``parse_frontmatter`` keeps quotes and treats ``tags: [...]`` as a literal
string, so callers pass already-formatted values. Keys are emitted in insertion order."""
return "\n".join(f"{key}: {' '.join(str(value).split())}" for key, value in frontmatter.items())
class IngestStampError(ValueError):
"""A curated writer was handed frontmatter carrying the COMPLETE ingest ownership stamp
(``shared/ingest-spec.md`` §3). The stamp is the sole mark separating ingest-owned files from
curated ones, and re-materialization replaces exactly what carries it — so a curated file that
forged it could be silently deleted by a later ingest run."""
_YAML_TRUE_LITERALS = frozenset({"true", "yes", "on"})
"""Every scalar a real YAML reader parses to boolean ``True`` (measured with PyYAML's ``safe_load``
core-schema resolver: ``true``/``yes``/``on``, any case, are bool; the same resolver reads bare
``y``/``n`` and ``1``/``0`` as string/int, never bool — so those are deliberately EXCLUDED here.
Widening past what a YAML reader actually resolves would over-block curated content no ingest
pipeline ever produces, on a form nothing downstream would honour as the stamp either."""
def _claims_ingest_ownership(value: str) -> bool:
"""Whether a ``generated`` value claims ingest ownership, in EITHER spelling the pinned writer
has used.
Two forms, one claim. The pre-V1 spelling is a YAML boolean (``_YAML_TRUE_LITERALS``). The V1
spelling is the provenance mapping ``{ by: <actor>, at: <timestamp> }`` — measured 2026-09-12 as
what ``llm-ingestion-okf`` >=0.8.5 writes into every generated concept, where 0.3.2 wrote
``true``. A detector that knew only the boolean read the V1 form as NOT a stamp, so the forgery
refusal in ``write_concept_file`` went inert on exactly the output of the writer it guards
against, with the whole fail-closed suite green (P13, ``docs/2026-09-12-p13-okf-pin-r761.md``
§ 2c′). Widening the predicate from "reads as boolean True" to "claims ownership" is what makes
the guard survive the emitter, rather than the emitter's current spelling.
The recogniser for the V1 half is ``decode_flow_value`` — the module's ONE flow decoder, never a
second copy of the rule (the kø-(p) precedent, and the same argument ``write_concept_file``
already makes for ``verified``: the writer refuses exactly what the reader can read). A value the
decoder REFUSES is therefore not an ownership claim and writes through, which is what keeps this
from collapsing into "any non-empty ``generated``"."""
literal = value.strip().strip('"').lower()
if literal in _YAML_TRUE_LITERALS:
return True
try:
return bool(decode_flow_value(value.strip(), key="generated"))
except FlowDecodeError:
return False
def _carries_complete_ingest_stamp(frontmatter: dict[str, str]) -> bool:
"""Whether ``frontmatter`` carries BOTH halves of the ingest ownership stamp: a ``generated``
value claiming ingest ownership (``_claims_ingest_ownership``) together with a non-empty
``ingest_manifest`` reference (ingest-spec §7).
FAIL-CLOSED on the value literal: the field previously matched only the exact string ``"true"``,
so a pinned ingest writer emitting any other YAML-1.1 truthy form (``yes``, ``on``) would have
slipped the stamp past this gate undetected — inert only by the accident of the pinned writer's
current output, per the CLAUDE.md ingest-stamp invariant. The test is on the COMPLETE stamp,
never on the individual field names — curated content may legitimately carry a single provenance
field, and a verbatim round-trip of one half must keep working. Values are compared the way
``parse_frontmatter`` yields them (line-oriented strings, quotes retained), so surrounding quotes
and case are normalised away here."""
manifest = str(frontmatter.get("ingest_manifest", "")).strip().strip('"')
return _claims_ingest_ownership(str(frontmatter.get("generated", ""))) and bool(manifest)
def write_concept_file(bundle_dir: str, name: str, frontmatter: dict[str, str], body: str) -> Path:
"""Write a typed OKF concept file (``---`` frontmatter + markdown body) into ``bundle_dir``,
path-safe via ``safe_resolve`` (fail-closed: a ``name`` escaping the bundle raises
``PathSecurityError``). Pure stdlib — the D7-portable counterpart of ``navigate_bundle``'s read.
This is the repo's one authoring primitive that materialises a concept file from CALLER-SUPPLIED
frontmatter, so it is the surface ingest-spec §3's "no other writer may forge the stamp" binds:
a frontmatter carrying the complete ingest stamp raises ``IngestStampError`` and NOTHING is
written. A validation, never a repair — the caller is told, not silently corrected.
Returns the written path."""
verified = frontmatter.get("verified")
if verified is not None:
# Validation, never repair, and NOT a second copy of the rule: the check IS
# ``decode_flow_value``, so the writer refuses exactly what the reader cannot read. A
# duplicated rule here would be free to drift, and the drifted copy would decide what gets
# written. Scoped to ``verified`` BY NAME — ``render_frontmatter`` keeps collapsing
# newlines for every other key, which this step deliberately does not change.
decode_flow_value(verified, key="verified")
if _carries_complete_ingest_stamp(frontmatter):
raise IngestStampError(
"refusing to write a curated concept file carrying the COMPLETE ingest ownership stamp "
"(a `generated` ownership claim — `true` or the V1 `{ by: ..., at: ... }` mapping — "
"together with ingest_manifest); only the ingest materializer may claim it "
"(ingest-spec §3) — either field alone is permitted"
)
resolved = Path(safe_resolve(bundle_dir, name))
resolved.parent.mkdir(parents=True, exist_ok=True)
resolved.write_text(f"---\n{render_frontmatter(frontmatter)}\n---\n\n{body}", encoding="utf-8")
return resolved
def link_in_index(bundle_dir: str, target_name: str, label: str) -> bool:
"""Append an intra-bundle cross-link ``- [label](target_name)`` to ``index.md`` so
``navigate_bundle`` (which follows ONLY index cross-links) reaches a newly written file.
Idempotent: if a link to ``target_name`` already exists the index is left untouched. Returns
whether a link was added. ``label`` is supplied by the caller and ends up in ``index_summary``
(hence ``bundle_context``) verbatim, so the promotion policy passes a NEUTRAL label carrying no
verdict signal (målbilde §3/§6). Known MVP limitation: the read-modify-write is not atomic."""
resolved = Path(safe_resolve(bundle_dir, _INDEX_NAME))
body = resolved.read_text(encoding="utf-8")
if f"]({target_name})" in body:
return False
prefix = body if body.endswith("\n") else body + "\n"
resolved.write_text(f"{prefix}- [{label}]({target_name})\n", encoding="utf-8")
return True
def load_cost_baseline(bundle_dir: str, name: str = _COST_BASELINE) -> CostBaseline:
"""Load the bundle's cost baseline (``cost-baseline.json`` by default): the project's ACTUAL
cost lines (``{code: {quantity, unit_cost}}``), which the deterministic validator reconciles a
proposal's ``affected_items`` against (S4.0, F3).
Fail-fast, mirroring ``load_ir_projection`` and ``dimension.load_dimension``: a missing file
raises ``FileNotFoundError`` and malformed content raises ``pydantic.ValidationError``. A cost
baseline is authoritative gate input — a tolerantly-degraded one would silently un-anchor the
gate, which is precisely the failure this stage exists to prevent. (The tolerant skip rule
belongs to the RAW verdict-inbox layer, never here.)
Use ``load_optional_cost_baseline`` where the ABSENCE of the file is legitimate."""
resolved = Path(safe_resolve(bundle_dir, name))
if not resolved.is_file():
raise FileNotFoundError(f"cost baseline not found in bundle: {name!r}")
return load_cost_baseline_file(str(resolved))
def load_cost_baseline_file(path: str) -> CostBaseline:
"""Load a cost baseline from a file that is NOT inside a knowledge base: the PROJECT's own price
schedule (P21, ``--cost-baseline``).
The measured reason it exists. Four paid rounds (P16/P18/P19/P17b/P20) ran entirely UN-ANCHORED,
because the only file loader reads ``cost-baseline.json`` out of the bundle directory and no road
normal carries a price schedule: a vegnormal is KNOWLEDGE, and the price belongs to the PROJECT.
Stage 0 was therefore skipped in every one of them, and "validated" could not mean anything —
P20 G1/G2 measured real process numbers validating with invented amounts. This is the third door
into ``CostBaseline`` alongside the bundle file and ``derive_cost_baseline``, and the only one
whose input is the project rather than the corpus.
**The SAME parse, never a second one** (kø-(p)): ``load_cost_baseline`` resolves inside the
bundle and then delegates here, so the two doors cannot disagree about what a baseline file is.
What differs is the resolution — ``safe_resolve`` is the ONE in-/out-of-bundle test and stays on
the bundle door alone, because a project's own schedule is legitimately outside every base.
Fail-fast, the error CLASSES of ``load_cost_baseline``: a missing file raises
``FileNotFoundError`` and malformed content raises ``pydantic.ValidationError``. There is no
optional twin, and that is deliberate: the bundle file is absent by default (a base authored
before the amendment is legitimately un-anchored), whereas this path exists only because an
operator NAMED a file — tolerating its absence would answer an explicit order with a silently
un-anchored run (``load_mandate``'s rule)."""
resolved = Path(path)
if not resolved.is_file():
raise FileNotFoundError(f"cost baseline not found: {path!r}")
return CostBaseline.model_validate_json(resolved.read_text(encoding="utf-8"))
def load_optional_cost_baseline(bundle_dir: str, name: str = _COST_BASELINE) -> CostBaseline | None:
"""``load_cost_baseline`` where a MISSING file is legitimate: returns ``None`` instead of
raising. This is the run path's loader — a bundle authored before the baseline amendment is
simply un-anchored (``None`` = pre-S4.0 behaviour), not an error, which is what keeps every
existing bundle (including the commons-owned goldens) running byte-identically.
The tolerance stops at absence: a baseline that EXISTS but is malformed still raises. Reading a
corrupt baseline as "no baseline" would hand back an un-anchored gate under the appearance of an
anchored one (the same reasoning as ``budget.read_spend``)."""
try:
return load_cost_baseline(bundle_dir, name)
except FileNotFoundError:
return None
class CostBaselineDerivationError(ValueError):
"""A priced schedule could not be read into a ``CostBaseline`` WITHOUT judgement (MAJOR-4).
A ``ValueError`` subclass on purpose, the ``BundleIdMismatch`` precedent: it lands on
``run.main``'s refusal tuple and the hosted surface's 400 arm rather than on the crash channel,
because a schedule this reader cannot map is a caller's input being wrong, not the framework
failing.
Every raise here is VALIDATION, never repair. That is the whole point of the third projection:
K2 is pre-award, so its ``Prisskjema`` carries codes and descriptions with the price column
empty, and the one thing a deriver must never do is answer that with a zero or an invented unit
cost. A baseline is the deterministic gate's ground truth — an invented line would let the gate
reconcile a hallucinated proposal against a hallucinated baseline and report itself anchored."""
#: Header text -> the role it fills, as a CLOSED vocabulary. Matching is on the lower-cased,
#: stripped header cell and is EXACT: a substring rule would let ``Enhetspris eks. mva`` and
#: ``Enhetspris inkl. mva`` both claim the same role and the reader would be choosing between two
#: prices, which is exactly the judgement this function may not exercise. An unrecognised header is
#: simply not a role, and a table missing any role is not a candidate.
_BASELINE_ROLES: Final[dict[str, tuple[str, ...]]] = {
"code": ("postnr", "post", "kode", "kostkode", "code", "cost_code"),
"quantity": ("mengde", "antall", "quantity"),
"unit_cost": ("enhetspris", "unit_cost", "unit price", "unit_price"),
}
#: The number grammar, and it is CLOSED: optional sign, digits, optional dot-decimal. Measured
#: (pandoc 3.10.2, the producer's own writer and arguments) the xlsx path emits ``1250.0`` / ``42.5``
#: — no thousands separators and no comma decimals — so admitting a comma would buy nothing and
#: import ``1,250``'s thousands-versus-decimal ambiguity for free. A hand-authored Norwegian
#: ``1 250,50`` is therefore a REFUSED cell, by name, rather than a guess.
_BASELINE_NUMBER_RE: Final = re.compile(r"-?\d+(?:\.\d+)?")
#: A pandoc SIMPLE table's dash rule: two or more dash groups, whose spans define the columns.
_SIMPLE_RULE_RE: Final = re.compile(r" *-+(?: +-+)+ *")
#: A pipe table's separator row cell, e.g. ``---`` or ``:---:``.
_PIPE_RULE_CELL_RE: Final = re.compile(r":?-+:?")
@dataclass(frozen=True)
class _ScheduleTable:
"""One markdown table found in a concept file: its header cells, its data rows, and the
bundle-relative file it was read from (so a refusal can name the document)."""
file: str
header: tuple[str, ...]
rows: tuple[tuple[str, ...], ...]
def _split_pipe_row(line: str) -> tuple[str, ...]:
"""Split one pipe-table row on UNESCAPED ``|``. ``render.render_table`` escapes ``\\`` first and
then ``|``, so the reader has to undo them in the opposite order or a cell containing a literal
backslash comes back wrong."""
cells, current, escaped = [], [], False
for char in line.strip().strip("|"):
if escaped:
current.append(char)
escaped = False
elif char == "\\":
escaped = True
elif char == "|":
cells.append("".join(current).strip())
current = []
else:
current.append(char)
cells.append("".join(current).strip())
return tuple(cells)
def _pipe_tables(file: str, body: str) -> list[_ScheduleTable]:
"""The csv/sql-sourced form (``render.render_table``): ``| a | b |`` over ``| --- | --- |``."""
tables, lines = [], body.split("\n")
for i in range(len(lines) - 1):
header_line, rule_line = lines[i].strip(), lines[i + 1].strip()
if not header_line.startswith("|") or not rule_line.startswith("|"):
continue
rule = _split_pipe_row(rule_line)
if not rule or not all(_PIPE_RULE_CELL_RE.fullmatch(cell) for cell in rule):
continue
rows = []
for line in lines[i + 2 :]:
if not line.strip().startswith("|"):
break
rows.append(_split_pipe_row(line))
tables.append(
_ScheduleTable(file=file, header=_split_pipe_row(header_line), rows=tuple(rows))
)
return tables
def _simple_tables(file: str, body: str) -> list[_ScheduleTable]:
"""The xlsx form. MEASURED, not assumed: ``extract._extract_office`` converts through pandoc
with ``_PANDOC_WRITER = "markdown"`` and ``_PANDOC_ARGS = ("--eol=lf", "--wrap=none")``, and
``inbox.py`` hands that text to ``render_inbox_concept`` untouched — it never reaches
``render_table``. The result is a pandoc SIMPLE table, and its dash rule is the column
definition: each dash group's span slices the same columns out of the header and every row.
Span-slicing rather than splitting on whitespace runs is what makes an EMPTY cell readable. K2's
whole shape is empty price cells, and a whitespace split would collapse them and silently shift
every later value one column left — a mis-mapping that reads as data rather than as a defect."""
tables, lines = [], body.split("\n")
for i in range(1, len(lines)):
if not _SIMPLE_RULE_RE.fullmatch(lines[i]) or not lines[i - 1].strip():
continue
spans = [(m.start(), m.end()) for m in re.finditer(r"-+", lines[i])]
rows = []
for line in lines[i + 1 :]:
if not line.strip():
break
rows.append(tuple(line[a:b].strip() for a, b in spans))
tables.append(
_ScheduleTable(
file=file,
header=tuple(lines[i - 1][a:b].strip() for a, b in spans),
rows=tuple(rows),
)
)
return tables
def _role_columns(table: _ScheduleTable) -> dict[str, list[int]] | None:
"""Map each role to the column indices claiming it, or ``None`` when the table is not a
candidate at all (some role is named by no column)."""
found: dict[str, list[int]] = {role: [] for role in _BASELINE_ROLES}
for index, cell in enumerate(table.header):
for role, names in _BASELINE_ROLES.items():
if cell.strip().lower() in names:
found[role].append(index)
return None if any(not columns for columns in found.values()) else found
def _cell(row: tuple[str, ...], index: int) -> str:
"""A row shorter than its header has an EMPTY cell there, never a missing one: a truncated row
is exactly how an unfilled trailing column arrives, and it must reach the same refusal."""
return row[index].strip() if index < len(row) else ""
def _number(raw: str, *, header: str, code: str) -> float:
if not _BASELINE_NUMBER_RE.fullmatch(raw):
raise CostBaselineDerivationError(
f"row {code!r}: column {header!r} holds {raw!r}, which is not a plain decimal number "
"(the grammar is optional sign, digits, optional dot-decimal — a thousands separator "
"or a comma decimal is refused rather than guessed at)"
)
return float(raw)
def derive_cost_baseline(bundle: Bundle, *, project_id: str) -> CostBaseline:
"""Derive a ``CostBaseline`` from a priced schedule the producer already rendered into a concept
file — the THIRD projection into the type whose docstring names the other two (MAJOR-4).
Neither existing projection can serve an ingested tender corpus: ``cost-baseline.json`` is
hand-written per project and ``validator.baseline_from_project`` belongs to the road reference
domain. So a K2-shaped bundle could be navigated and never anchored. This reads the numbers that
are already IN the bundle.
**``project_id`` is a REQUIRED keyword rather than something read from the bundle.** The bundle's
``validator-input.json`` carries one, but ``run._project_from_bundle`` already fail-fasts the
run's requested id against it — reading it a second time here would make this a second reader of
a fact that has an owner (kø-(p)), and would drag an IR projection into a function whose whole
input is a table.
**NO judgement anywhere.** The header vocabulary is closed and matched exactly, the number
grammar is closed, and every ambiguity is a refusal rather than a choice:
* no table in the bundle names all three roles, or more than one does;
* two columns of one table claim the same role;
* two rows carry the same cost code (a dict would last-write-win, and the baseline would then
describe one of two lines the operator can see in the document);
* a row prices nothing — empty, unparseable, or a non-positive unit cost.
**A partly-priced schedule refuses in FULL.** Not conservatism: a half-derived baseline anchors
some codes while ``ProvenanceStamp.cost_baseline_anchored`` reports ``True``, and that bit is
required-without-default precisely because both of its defaults would lie.
Scanned over ``context_files``, never ``files`` (MAJOR-3's rule): a ``type: verdict`` file is a
prior judgement, not project cost data, and a reader over ``files`` would let one decide a
project's ground truth outside the gated ExpeL fold.
**Honesty limits, stated.** A row that is empty across all three mapped columns asserts nothing
and is skipped — that is a blank spacer, not a price. But NS 3451 section rows (a code and a
heading with no quantity, inside an otherwise priced sheet) are NOT classified, because K2 itself
was not available to measure here; such a sheet refuses, and the rule for it should be written
when someone can measure the real form. And the provenance stamp records THAT a run was
anchored, never WHICH of the three projections anchored it.
Gated by ``tests/test_cost_baseline_derivation_loadbearing.py``."""
tables: list[tuple[_ScheduleTable, dict[str, list[int]]]] = []
for concept in bundle.context_files:
for table in _pipe_tables(concept.name, concept.body) + _simple_tables(
concept.name, concept.body
):
roles = _role_columns(table)
if roles is not None:
tables.append((table, roles))
if not tables:
raise CostBaselineDerivationError(
f"no cost table found in bundle {bundle.dir!r}: no concept file carries a markdown "
f"table whose header names all three of {sorted(_BASELINE_ROLES)} "
f"(recognised headers: {_BASELINE_ROLES})"
)
if len(tables) > 1:
named = ", ".join(sorted({table.file for table, _ in tables}))
raise CostBaselineDerivationError(
f"{len(tables)} cost tables found in bundle {bundle.dir!r} ({named}); which one prices "
"the project is a question about the documents, not one this reader answers by order "
"of appearance"
)
table, roles = tables[0]
for role, columns in roles.items():
if len(columns) > 1:
duplicated = ", ".join(repr(table.header[index]) for index in columns)
raise CostBaselineDerivationError(
f"{table.file}: {len(columns)} columns claim the {role!r} role ({duplicated}); "
"a reader that took the first would be choosing between them"
)
code_at, quantity_at, unit_cost_at = (
roles[role][0] for role in ("code", "quantity", "unit_cost")
)
items: dict[str, CostBaselineLine] = {}
for row in table.rows:
code = _cell(row, code_at)
quantity_raw = _cell(row, quantity_at)
unit_cost_raw = _cell(row, unit_cost_at)
if not code and not quantity_raw and not unit_cost_raw:
continue # a blank spacer row asserts nothing
if not code:
raise CostBaselineDerivationError(
f"{table.file}: a row carries numbers under an empty {table.header[code_at]!r} "
"cell, so the line it prices cannot be named"
)
for raw, index in ((quantity_raw, quantity_at), (unit_cost_raw, unit_cost_at)):
if not raw:
raise CostBaselineDerivationError(
f"{table.file}: row {code!r} has an empty {table.header[index]!r} cell. This "
"schedule is not priced (K2's pre-award shape); a baseline is refused rather "
"than completed with a value nobody wrote"
)
quantity = _number(quantity_raw, header=table.header[quantity_at], code=code)
unit_cost = _number(unit_cost_raw, header=table.header[unit_cost_at], code=code)
if unit_cost <= 0:
raise CostBaselineDerivationError(
f"{table.file}: row {code!r} has {table.header[unit_cost_at]!r} = {unit_cost}, "
"which prices nothing; a baseline line must carry a positive unit cost"
)
if quantity < 0:
raise CostBaselineDerivationError(
f"{table.file}: row {code!r} has {table.header[quantity_at]!r} = {quantity}"
)
if code in items:
raise CostBaselineDerivationError(
f"{table.file}: cost code {code!r} appears twice; keeping either row would make "
"the baseline describe one of two lines the document shows"
)
items[code] = CostBaselineLine(quantity=quantity, unit_cost=unit_cost)
if not items:
raise CostBaselineDerivationError(f"{table.file}: the cost table has no priced rows at all")
return CostBaseline(project_id=project_id, items=items)
def load_ir_projection(bundle_dir: str, name: str = _IR_PROJECTION) -> dict[str, Any]:
"""Load the bundle's IR projection (``validator-input.json`` by default): the candidate
measure's cost-IR (``measure``, ``affected_items``, ``claimed_saving_nok``) — the
pre-hypothesis ExpeL query-key source. Raises if missing / escaping the bundle (fail-fast: it
is required input, not an optional cross-link)."""
resolved = Path(safe_resolve(bundle_dir, name))
if not resolved.is_file():
raise FileNotFoundError(f"IR projection not found in bundle: {name!r}")
data: dict[str, Any] = json.loads(resolved.read_text(encoding="utf-8"))
return data
def load_optional_ir_projection(
bundle_dir: str, name: str = _IR_PROJECTION
) -> dict[str, Any] | None:
"""``load_ir_projection`` where a MISSING file is legitimate: returns ``None`` instead of
raising (S7b søm 1). This is the run path's loader — an INGESTED corpus carries no hand-written
IR projection, and requiring one meant such a base could be navigated, catalogued and judged
through the deterministic mandate door and still not run the eight-step loop at all.
A PAIR beside the fail-fast loader rather than a ``required=`` flag on it, mirroring
``load_cost_baseline`` / ``load_optional_cost_baseline`` exactly. A flag would leave one branch
unexercised wherever callers took the default, which is the rot PM-tillegg 5 measured on
``evidence_for``; and it would falsify the several docstrings across the package that cite
``load_ir_projection`` as *the* fail-fast precedent.
The tolerance stops at absence: a projection that EXISTS but is malformed still raises, and a
name escaping the bundle still raises. Reading a corrupt projection as "no projection" would
hand back an unkeyed run under the appearance of a keyed one — the same reasoning as
``load_optional_cost_baseline`` and ``budget.read_spend``."""
try:
return load_ir_projection(bundle_dir, name)
except FileNotFoundError:
return None
Reference in a new issue View git blame Copy permalink