portfolio-optimiser/src/portfolio_optimiser/okf.py

477 lines
26 KiB
Python

"""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 json
import posixpath
import re
from dataclasses import dataclass
from pathlib import Path
from typing import Any, Literal
from portfolio_optimiser.ir import CostBaseline
from portfolio_optimiser.retrieval import PathSecurityError, safe_resolve
_INDEX_NAME = "index.md"
_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. Unterminated blocks are
parsed as if closed — the pre-split behaviour, preserved deliberately."""
fm: dict[str, str] = {}
for line in _split_frontmatter(text)[0]:
key, sep, val = line.partition(":")
if sep:
fm[key.strip()] = val.strip()
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"))
@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
@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"]
@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"
]
@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))
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 dimension is not None:
file_dim = f.frontmatter.get("dimension")
if file_dim is not None and file_dim != 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 _carries_complete_ingest_stamp(frontmatter: dict[str, str]) -> bool:
"""Whether ``frontmatter`` carries BOTH halves of the ingest ownership stamp: a ``generated``
value a YAML reader would read as boolean ``True`` (``_YAML_TRUE_LITERALS``) 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."""
generated = str(frontmatter.get("generated", "")).strip().strip('"').lower()
manifest = str(frontmatter.get("ingest_manifest", "")).strip().strip('"')
return generated in _YAML_TRUE_LITERALS 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."""
if _carries_complete_ingest_stamp(frontmatter):
raise IngestStampError(
"refusing to write a curated concept file carrying the COMPLETE ingest ownership stamp "
"(generated: true + 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 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
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