portfolio-optimiser/src/portfolio_optimiser/okf.py

"""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 typing import Any, Final, Literal

from portfolio_optimiser.ir import CostBaseline, CostBaselineLine
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"))


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_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("- "))
        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"]

    @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))


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 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."""
    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 "
            "(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


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