feat(fase2): wire Step-1 ExpeL retrieval into the hypothesis prompt

Closes maalbilde §5 gap #1 (the one missing "feedback-into-prompt" dataflow)
for the OKF-bundle path. Before, ExpeL was computed AFTER generation into a
discarded SessionContext, so a prior verdict could not influence any hypothesis
(context_providers=0).

- New okf.py: framework-neutral OKF bundle navigation (index + frontmatter +
  cross-links), pure stdlib, no agent_framework/mcp (D7-portable), enforced by
  test_okf_is_maf_free.
- verdicts.py: seed_store_from_bundle + bundle_candidate_features build the
  ExpeL substrate + the pre-hypothesis query key from a bundle.
- run_project(bundle_dir=...): folds the candidate's prior verdicts into the
  generation context BEFORE generate_via_llm; the road path is unchanged.

Load-bearing (maalbilde §7): test_step1_expel_loadbearing proves a prior verdict
reaches the hypothesis prompt and goes RED when the fold is detached (shown via
TDD red->green). The marker is the minted verdict id (content hash) because
docs_dir==bundle_dir lets keyword chunk-stuffing leak the realization rate;
clean layer separation is Fase 2b.

Suite 121->133 passed; mypy + ruff check clean.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01MHR8iKxJRxDiDfNw8HZmWE
This commit is contained in:
Kjell Tore Guttormsen 2026-06-29 10:56:48 +02:00
commit d6d83d42b5
9 changed files with 529 additions and 7 deletions

View file

@ -0,0 +1,135 @@
"""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 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 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