202 lines
9.9 KiB
Python
202 lines
9.9 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, parse each file's YAML frontmatter, classify by the one required
|
|
``type`` field. **NO** ``agent_framework``, **NO** ``mcp`` — pure stdlib, so the SAME navigation
|
|
serves both the MAF and the Claude-SDK implementations unchanged (målbilde §4 vendor-neutrality).
|
|
|
|
Robustness is part of the spec (OKF SPEC §4): consumers MUST tolerate broken links and unknown
|
|
fields. A link to a missing file — or one escaping the bundle — 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.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import re
|
|
from dataclasses import dataclass
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
from portfolio_optimiser.retrieval import PathSecurityError, safe_resolve
|
|
|
|
_INDEX_NAME = "index.md"
|
|
_IR_PROJECTION = "validator-input.json"
|
|
# Intra-bundle markdown cross-links: ``](target.md)``. Targets with a path separator (``/``) are
|
|
# treated as out-of-bundle and skipped (only same-dir bundle files are navigated).
|
|
_LINK_RE = re.compile(r"\]\(([^)]+\.md)\)")
|
|
|
|
|
|
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."""
|
|
lines = Path(path).read_text(encoding="utf-8").splitlines()
|
|
if not lines or lines[0].strip() != "---":
|
|
return {}
|
|
fm: dict[str, str] = {}
|
|
for line in lines[1:]:
|
|
if line.strip() == "---":
|
|
break
|
|
key, sep, val = line.partition(":")
|
|
if sep:
|
|
fm[key.strip()] = val.strip()
|
|
return fm
|
|
|
|
|
|
def _read_body(path: Path) -> str:
|
|
"""The markdown body after the frontmatter block (or the whole file if there is none)."""
|
|
lines = path.read_text(encoding="utf-8").splitlines(keepends=True)
|
|
if lines and lines[0].strip() == "---":
|
|
for i in range(1, len(lines)):
|
|
if lines[i].strip() == "---":
|
|
return "".join(lines[i + 1 :]).lstrip("\n")
|
|
return "".join(lines)
|
|
|
|
|
|
@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
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class Bundle:
|
|
"""A navigated OKF bundle: ``index.md`` plus every cross-linked file that resolves."""
|
|
|
|
dir: str
|
|
files: tuple[BundleFile, ...]
|
|
|
|
@property
|
|
def index_summary(self) -> str:
|
|
"""The index body — the progressive-disclosure entry point (not whole-bundle stuffing)."""
|
|
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)."""
|
|
return [f for f in self.files if 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 ``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
|
|
fm = parse_frontmatter(resolved)
|
|
return BundleFile(name=name, type=fm.get("type", ""), frontmatter=fm, body=_read_body(resolved))
|
|
|
|
|
|
def navigate_bundle(bundle_dir: str) -> Bundle:
|
|
"""Navigate the OKF bundle from ``index.md``: parse the index, follow its intra-bundle ``.md``
|
|
cross-links, and read each linked file's frontmatter + body. Deterministic: index first, then
|
|
links in first-seen order, de-duplicated. Broken / escaping links are skipped (§4). Raises
|
|
``ValueError`` only when ``index.md`` itself is unreadable (a bundle has no entry point)."""
|
|
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]
|
|
seen = {_INDEX_NAME}
|
|
for target in _LINK_RE.findall(index.body):
|
|
if "/" in target or target in seen:
|
|
continue # only same-dir bundle files; de-dup repeated links
|
|
seen.add(target)
|
|
linked = _load_file(bundle_dir, target)
|
|
if linked is not None:
|
|
files.append(linked)
|
|
return Bundle(dir=bundle_dir, files=tuple(files))
|
|
|
|
|
|
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."""
|
|
sections = [bundle.index_summary]
|
|
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 = f.frontmatter.get("title", f.name).strip('"')
|
|
sections.append(f"## {f.type or 'document'}: {title}\n{f.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())
|
|
|
|
|
|
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.
|
|
Returns the written path."""
|
|
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_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
|