portfolio-optimiser-claude/tests/test_ingest_spec_loadbearing.py

"""Spec-integrity seam for the ingest spec (ingest-spec §11).

The D7 analog of MAF's I1 framework-guard: this repo consumes ``shared/ingest-spec.md``
UNCHANGED from commons, and this test keeps the contract honest — it goes RED when the
spec goes missing, names a concrete agent toolkit (the framework-neutrality rule), or
stops documenting a contract field. It is the load-bearing guard the ingest layer relies
on to keep being implementable "from this spec alone".

Form mirrored from the sibling ``test_method_spec_loadbearing.py``: every predicate takes
the spec TEXT as an argument, so the detach-proofs are tests in the suite rather than a
one-off spot-check that dies with the session. Red-proofs run against a mutated COPY of
the spec in ``tmp_path`` — never against ``shared/`` itself.

**Second half — structure markers (added after a measurement).** Everything above anchors
§12 plus two whole-document properties. Measured: gutting ANY of §1–§11 to its heading alone
left this guard green for 11 of 11 sections — and measured wider, for 10 of the 11 nothing in
the whole 806-test suite went red either, so a commons pull could empty a normative section
and no test would notice. (§7 is the single overlap: the O2 ratchet reads its `generated`
row.) ``_STRUCTURE_MARKERS`` closes that hole by
binding ONE verbatim normative phrase to the section that OWNS it, the form already proven in
``test_method_spec_rule_phrases_loadbearing.py``. Two properties make the binding real rather
than decorative: the phrase must appear in ITS OWN section body (presence anywhere is what a
whole-document check already cannot see), and it must be UNIQUE in the document (a phrase
occurring twice could be satisfied from a section other than its owner).

The SELECTION is ours, not mirrored: each phrase names a seam THIS repo actually carries and
cites the owning module, because a phrase anchoring a seam we do NOT implement would be
green-but-dead of the opposite kind — protecting spec text nothing here depends on. §7's
``generated`` row is deliberately NOT the chosen phrase: the O2 ratchet
(``test_ingest_stamp_conformance_loadbearing.py``) already anchors that row, so §7 is bound
here on a second, independent sentence.
Matching is whitespace- and emphasis-normalized so a reflow or a bolding change does not
false-red; the normalization has its own controls below.
"""

from __future__ import annotations

import re
from pathlib import Path

import pytest

SPEC = Path(__file__).resolve().parents[1] / "shared" / "ingest-spec.md"

# Concrete agent toolkits / vendor stacks the framework-neutral spec MUST NOT name.
_FORBIDDEN_TOOLKITS = (
    "claude",
    "anthropic",
    "openai",
    "gpt",
    "gemini",
    "llama",
    "langchain",
    "autogen",
    "crewai",
    "semantic kernel",
    "microsoft agent framework",
    "agent sdk",
    "bedrock",
    "vertex",
    "foundry",
    "maf",
)

# The §12 cross-check table is the ANCHOR, and the spec appoints it itself: "Every field
# of the machine-readable contracts, mapped to its normative section (completeness is
# enforced by the spec-integrity test)" — this test is that enforcer. Asserting over the
# whole spec text instead would be green-but-dead: prose saturates the field names —
# `generated` alone appears 22 times OUTSIDE §12 and exactly once inside it (measured
# after the V1 pull) — so no amendment dropping a row could ever turn it red.
_CROSS_CHECK_HEADING = "## 12. Cross-check table"

# Every field of the machine-readable contracts the D7 implementation depends on — the
# spec's §12 cross-check table must keep documenting each (spec-integrity).
_CONTRACT_FIELDS = (
    "manifest_version",
    "source",
    "bundle_summary",
    "extractions",
    "source_system",
    "source_query",
    "ingested_at",
    "ingest_manifest",
    "generated",
    "okf_type",
    "max_rows",
    "root",
    "connection_ref",  # the sql source reference the D7 sql connector (I5) depends on
)


def _cross_check_table(text: str) -> str:
    """The §12 section body — heading to end-of-spec or the next section, whichever first."""
    start = text.index(_CROSS_CHECK_HEADING)  # RED (ValueError) if §12 is renamed or dropped
    end = text.find("\n## ", start + len(_CROSS_CHECK_HEADING))
    return text[start:] if end == -1 else text[start:end]


def _named_toolkits(text: str) -> list[str]:
    low = text.lower()
    return [tok for tok in _FORBIDDEN_TOOLKITS if tok in low]


def _undocumented_fields(text: str) -> list[str]:
    # The row's FIRST column is the documented-field claim; a field named only in another
    # row's "Contract" prose does not count.
    table = _cross_check_table(text)
    return [field for field in _CONTRACT_FIELDS if f"| `{field}` |" not in table]


def _slice_defects(table: str) -> list[str]:
    """Defects in the anchor itself — a slice that widened into the surrounding spec."""
    defects = []
    if not table.startswith(_CROSS_CHECK_HEADING):
        defects.append("does not start at the §12 heading")
    if "Honesty rule" in table:
        defects.append("leaked §1 prose — the anchor is not a slice")
    if "\n## " in table:
        defects.append("leaked a following section")
    return defects


def _row_removed(text: str, field: str) -> str:
    table = _cross_check_table(text)
    kept = [ln for ln in table.splitlines(keepends=True) if not ln.startswith(f"| `{field}` |")]
    return text.replace(table, "".join(kept))


def _row_renamed(text: str, field: str) -> str:
    table = _cross_check_table(text)
    return text.replace(table, table.replace(f"| `{field}` |", f"| `{field}_renamed` |"))


# --- The guard itself (against the real spec) ---------------------------------------


def _spec_is_present(path: Path) -> bool:
    """The presence predicate itself, so the red-proof can exercise THE SAME one."""
    return path.is_file()


def test_spec_is_present() -> None:
    # RED if the spec goes missing (the layer stops being implementable from spec alone).
    assert _spec_is_present(SPEC), "ingest-spec.md missing — subtree pull the commons contract"


def test_spec_names_no_agent_toolkit() -> None:
    present = _named_toolkits(SPEC.read_text(encoding="utf-8"))
    assert not present, f"framework-neutral spec names a concrete toolkit: {present}"


def test_cross_check_slice_is_a_slice_and_not_the_whole_spec() -> None:
    # Guards the anchor itself: a slice that degenerated into the full text would make
    # every row assertion below green-but-dead again, silently. RED if it widens.
    defects = _slice_defects(_cross_check_table(SPEC.read_text(encoding="utf-8")))
    assert not defects, f"the §12 anchor degenerated: {defects}"


@pytest.mark.parametrize("field", _CONTRACT_FIELDS)
def test_spec_documents_contract_field(field: str) -> None:
    # RED when an amendment drops or renames a §12 row.
    undocumented = _undocumented_fields(SPEC.read_text(encoding="utf-8"))
    assert field not in undocumented, (
        f"contract field {field!r} is no longer a row in the §12 cross-check table"
    )


# --- Red-proofs: the guard MUST fail on a detached spec (mutated copy, never shared/) --


def test_guard_red_when_spec_missing(tmp_path: Path) -> None:
    # Was VACUOUS — see the twin in ``test_method_spec_loadbearing.py``: it asserted
    # a file is absent from a fresh ``tmp_path``, true by construction, and never
    # touched the guard it is named for. Now it exercises THE SAME predicate the
    # guard calls, both directions, positive control first.
    assert _spec_is_present(SPEC)
    assert not _spec_is_present(tmp_path / "ingest-spec.md")


@pytest.mark.parametrize("toolkit", _FORBIDDEN_TOOLKITS)
def test_guard_red_when_toolkit_injected(tmp_path: Path, toolkit: str) -> None:
    mutated = SPEC.read_text(encoding="utf-8") + f"\n\nBuilt on {toolkit}.\n"
    copy = tmp_path / "ingest-spec.md"
    copy.write_text(mutated, encoding="utf-8")
    assert toolkit in _named_toolkits(copy.read_text(encoding="utf-8"))


@pytest.mark.parametrize("field", _CONTRACT_FIELDS)
def test_guard_red_when_row_removed_from_cross_check_table(tmp_path: Path, field: str) -> None:
    # M1: the row is gone from §12 — even though prose elsewhere may still name the field.
    copy = tmp_path / "ingest-spec.md"
    copy.write_text(_row_removed(SPEC.read_text(encoding="utf-8"), field), encoding="utf-8")
    assert field in _undocumented_fields(copy.read_text(encoding="utf-8"))


@pytest.mark.parametrize("field", _CONTRACT_FIELDS)
def test_guard_red_when_row_renamed_in_cross_check_table(tmp_path: Path, field: str) -> None:
    # M2: a detach-proof is not a value-proof — the row still exists, under another name.
    copy = tmp_path / "ingest-spec.md"
    copy.write_text(_row_renamed(SPEC.read_text(encoding="utf-8"), field), encoding="utf-8")
    assert field in _undocumented_fields(copy.read_text(encoding="utf-8"))


def test_guard_red_when_section12_heading_renamed(tmp_path: Path) -> None:
    # M3: fail-closed — no §12 heading means no anchor, and the guard must raise, not
    # silently fall back to a wider (green-but-dead) slice.
    mutated = SPEC.read_text(encoding="utf-8").replace(
        _CROSS_CHECK_HEADING, "## 12. Field reference"
    )
    copy = tmp_path / "ingest-spec.md"
    copy.write_text(mutated, encoding="utf-8")
    with pytest.raises(ValueError):
        _cross_check_table(copy.read_text(encoding="utf-8"))


def test_slice_guard_red_when_anchor_degenerates_to_whole_spec() -> None:
    # M4: the anchor can degenerate. If _cross_check_table ever returned the full text,
    # the row assertions would go green-but-dead again — this proves the slice guard is
    # what catches that, and that it is not itself green by accident.
    defects = _slice_defects(SPEC.read_text(encoding="utf-8"))
    assert defects, "the slice guard accepts the whole spec as the §12 table — it is dead"


# --- Structure markers: one normative phrase bound to the section that OWNS it -------

# section marker -> (verbatim normative phrase, the seam it anchors / owning module)
_STRUCTURE_MARKERS: dict[str, tuple[str, str]] = {
    "## 1.": (
        "data reaches the model ONLY via OKF bundles",
        "bundle-only data path — context by navigation, never query-time retrieval (okf.py)",
    ),
    "## 2.": (
        "A bundle without `index.md` is an error",
        "a missing root index is fatal, not skipped (okf.py navigate_bundle)",
    ),
    "## 3.": (
        "The promotion gate is the ONLY path into the verdict layer",
        "verdict reservation — promote() is the sole verdict writer (promotion.py, ingest.py)",
    ),
    "## 4.": (
        "Credentials never live in the manifest",
        "`connection_ref` is a NAME resolved from the environment at run time (ingest.py)",
    ),
    "## 5.": (
        "there is NO wall-clock default",
        "`ingested_at` is a required argument of materialize() (ingest.py)",
    ),
    "## 6.": (
        "ALL other links — curated and promoted — are preserved verbatim",
        "re-ingest layer safety — a promoted verdict's index link survives (promotion.py)",
    ),
    "## 7.": (
        "OKF consumers preserve unknown frontmatter fields",
        "frontmatter parsing keeps unknown keys, so §7 rides through (okf.py)",
    ),
    "## 8.": (
        "the manifest cannot grant itself network access",
        "the offline invariant at the consumer seam — no network opt-in is passed (ingest.py)",
    ),
    "## 9.": (
        "ingest NEVER runs automatically",
        "no scheduler, no run-path call site — the manifest path is explicit (ingest.py)",
    ),
    "## 10.": (
        "given the same source content, manifest, and `ingested_at`, the result is byte-identical",
        "the byte-exact golden extractions (examples/ingest-golden-*, test_ingest_golden.py)",
    ),
    "## 11.": (
        "Test suites MUST run without credentials and without network access",
        "the suite-wide offline invariant (test_sdk_isolation.py, no key and no network)",
    ),
}


def _normalize(text: str) -> str:
    """Whitespace- and emphasis-insensitive: a reflow or a bolding change must not red."""
    return re.sub(r"\s+", " ", text.replace("*", "")).strip()


def _section_body(text: str, marker: str) -> str:
    """Heading to the next section. RED (ValueError) if the section is renamed or dropped."""
    start = text.index(marker)
    end = text.find("\n## ", start + len(marker))
    return text[start:] if end == -1 else text[start:end]


def _is_anchored(text: str, marker: str) -> bool:
    phrase, _ = _STRUCTURE_MARKERS[marker]
    return _normalize(phrase) in _normalize(_section_body(text, marker))


def _occurrences(text: str, marker: str) -> int:
    phrase, _ = _STRUCTURE_MARKERS[marker]
    return _normalize(text).count(_normalize(phrase))


def _body_emptied(text: str, marker: str) -> str:
    """The section, gutted to its heading alone — the pull this guard exists to catch."""
    body = _section_body(text, marker)
    heading = body.splitlines()[0]
    return text.replace(body, heading + "\n\n")


@pytest.mark.parametrize("marker", list(_STRUCTURE_MARKERS))
def test_section_keeps_its_structure_marker(marker: str) -> None:
    # RED when the owning section loses the normative content this repo builds on — which
    # the §12-anchored half above cannot see, because §12 survives an emptied §1-§11.
    phrase, seam = _STRUCTURE_MARKERS[marker]
    assert _is_anchored(SPEC.read_text(encoding="utf-8"), marker), (
        f"{marker} no longer carries its structure marker ({seam}): {phrase!r}"
    )


@pytest.mark.parametrize("marker", list(_STRUCTURE_MARKERS))
def test_structure_marker_is_unique_in_the_spec(marker: str) -> None:
    # A phrase occurring twice could be satisfied from a section other than its owner,
    # reopening the hole quietly. Binding is only meaningful while the phrase is unique.
    count = _occurrences(SPEC.read_text(encoding="utf-8"), marker)
    assert count == 1, f"{marker}'s structure marker occurs {count} times — binding is ambiguous"


@pytest.mark.parametrize("marker", list(_STRUCTURE_MARKERS))
def test_guard_red_when_section_body_emptied(marker: str) -> None:
    # The value-proof: GREEN before the mutation, RED after it, same mutation both ways.
    text = SPEC.read_text(encoding="utf-8")
    assert _is_anchored(text, marker)  # positive control first
    assert not _is_anchored(_body_emptied(text, marker), marker)


@pytest.mark.parametrize("marker", list(_STRUCTURE_MARKERS))
def test_the_older_half_is_blind_to_the_emptied_section(marker: str) -> None:
    # The DENOMINATOR, machine-checked: 11 of 11 gutted sections leave the §12-anchored
    # half fully green. Without this the value-proof above would only show that the new
    # marker CAN red, not that it is the only thing standing there.
    gutted = _body_emptied(SPEC.read_text(encoding="utf-8"), marker)
    assert not _named_toolkits(gutted), "the toolkit rule reds here — this section is not blind"
    assert not _undocumented_fields(gutted), "a §12 row went missing — not a clean measurement"
    assert not _slice_defects(_cross_check_table(gutted)), "the §12 slice moved — not blind"
    assert not _is_anchored(gutted, marker), "and the new marker is what does red"


@pytest.mark.parametrize("marker", list(_STRUCTURE_MARKERS))
def test_guard_red_when_marker_moves_to_another_section(marker: str) -> None:
    # Presence is not anchoring. The phrase survives in the document — appended after §12 —
    # and the owning section must still report unanchored.
    text = SPEC.read_text(encoding="utf-8")
    phrase, _ = _STRUCTURE_MARKERS[marker]
    moved = _body_emptied(text, marker).rstrip("\n") + f"\n\n{phrase}\n"
    assert _normalize(phrase) in _normalize(moved)  # control: it IS still in the document
    assert not _is_anchored(moved, marker)


@pytest.mark.parametrize("marker", list(_STRUCTURE_MARKERS))
def test_guard_red_when_section_heading_renamed(marker: str) -> None:
    # Fail-closed, as the §12 slice already is: no heading means no anchor, and the slice
    # must raise rather than silently fall back to a wider (green-but-dead) match.
    renamed = SPEC.read_text(encoding="utf-8").replace(marker, marker.replace("## ", "## §"))
    with pytest.raises(ValueError):
        _section_body(renamed, marker)


@pytest.mark.parametrize("marker", list(_STRUCTURE_MARKERS))
def test_normalization_tolerates_reflow_and_emphasis(marker: str) -> None:
    # A line-rewrap or a bolding change is not a contract change; it must not red.
    text = SPEC.read_text(encoding="utf-8")
    phrase, _ = _STRUCTURE_MARKERS[marker]
    reflowed = text.replace(phrase, phrase.replace(" ", "\n  ", 1))
    assert _is_anchored(reflowed, marker), "a reflow false-reds the guard"
    bolded = text.replace(phrase, f"**{phrase}**")
    assert _is_anchored(bolded, marker), "an emphasis change false-reds the guard"


def test_normalization_is_not_so_loose_that_anything_matches() -> None:
    # The tolerance above is only safe while a CHANGED WORD still reds. Without this,
    # `_normalize` could degenerate into something that matches everything.
    assert _normalize("a  b\n c") == "a b c"
    assert _normalize("**bold** text") == "bold text"
    assert _normalize("MUST fail fast") != _normalize("MUST fail slowly")