llm-ingestion-okf/tests/test_mcp_server.py

"""The MCP surface itself (`src/llm_ingestion_okf/mcp_server.py`).

`tools/okf_mcp_gate.py` measures this module over a real protocol and is the
eval it was built against. These tests hold the pieces the gate reaches only
through a verdict: what discovery does with a symlink, what the card does NOT
do to the bundle, and the property that makes the generic skill's whole claim
checkable.
"""

from __future__ import annotations

import hashlib
import json
import subprocess
import sys
from pathlib import Path

import pytest

from llm_ingestion_okf import mcp_server, skill

TOOLS = Path(__file__).resolve().parents[1] / "tools"
sys.path.insert(0, str(TOOLS))

import okf_mcp_gate as gate  # noqa: E402

GOLDEN = (
    Path(__file__).resolve().parents[1]
    / "examples"
    / "ingest-golden-segmented-okf-v0-2"
    / "expected-bundle"
)


def _tree(root: Path) -> dict[str, str]:
    return {
        str(path.relative_to(root)): hashlib.sha256(path.read_bytes()).hexdigest()
        for path in sorted(root.rglob("*"))
        if path.is_file()
    }


def test_the_card_is_derived_and_writes_nothing_into_the_bundle(tmp_path: Path) -> None:
    """The whole reason the card is not a file in the bundle.

    A stored card would move the bytes of all six `examples/*/expected-bundle`
    trees and of the pinned reference bundle, and would be one more artefact
    that can disagree with what is beside it.
    """
    bundle = tmp_path / "b"
    bundle.mkdir()
    for source in GOLDEN.rglob("*"):
        target = bundle / source.relative_to(GOLDEN)
        if source.is_dir():
            target.mkdir(parents=True, exist_ok=True)
        else:
            target.write_bytes(source.read_bytes())
    before = _tree(bundle)
    first = mcp_server.card(bundle, profile=mcp_server.okf_consume.DEFAULT_PROFILE)
    second = mcp_server.card(bundle, profile=mcp_server.okf_consume.DEFAULT_PROFILE)
    assert _tree(bundle) == before
    assert first == second
    assert first["concept_count"] == 3
    assert first["ref"].startswith("sha256-tree:")


def test_the_card_command_prints_json_a_reader_can_parse() -> None:
    """The generic skill tells its reader to run this. A command whose output
    could not be read back would make that instruction decorative."""
    run = subprocess.run(
        [sys.executable, "-m", "llm_ingestion_okf.cli", "card", str(GOLDEN)],
        capture_output=True,
        text=True,
        check=False,
    )
    assert run.returncode == 0, run.stderr
    payload = json.loads(run.stdout)
    assert payload["bundle_id"] == "b-golden-segmented-okf-v0-2"
    assert set(payload) >= {"ref", "concept_count", "conditional_fields", "whole_bundle_bytes"}


def test_discovery_never_descends_a_symlink(tmp_path: Path) -> None:
    """Driven from both sides: the real directory IS found, the link to it is
    not. Without the control, a rule that found nothing at all would pass."""
    outside = tmp_path / "outside"
    gate.write_bundle(outside / "secret", "secret-notes", [("x", "X", "y")])
    root = tmp_path / "root"
    gate.write_bundle(root / "real", "real-notes", [("x", "X", "y")])
    (root / "linked").symlink_to(outside / "secret", target_is_directory=True)

    found = mcp_server.discover([root])
    assert [served.bundle_id for served in found.bundles] == ["real-notes"]
    assert mcp_server.discover([outside]).bundles[0].bundle_id == "secret-notes"


def test_a_directory_that_cannot_be_read_as_a_bundle_is_reported_not_skipped(
    tmp_path: Path,
) -> None:
    """An absence with no denominator is not a boundary. A broken manifest that
    simply vanished from the list would be indistinguishable from a bundle that
    was never there."""
    root = tmp_path / "root"
    gate.write_bundle(root / "good", "good-notes", [("x", "X", "y")])
    gate.write_bundle(root / "bad", "bad-notes", [("x", "X", "y")])
    (root / "bad" / "index.md").write_bytes(b"\xff\xfe\x00")
    found = mcp_server.discover([root])
    assert [served.bundle_id for served in found.bundles] == ["good-notes"]
    assert [entry.path for entry in found.unreadable] == ["bad"]


def test_the_one_to_one_server_finds_the_bundle_it_was_started_on(tmp_path: Path) -> None:
    """The defect the gate found on this module's first build: `--bundle` points
    AT a bundle, and discovery that only looked at children found none."""
    gate.write_bundle(tmp_path / "b", "solo-notes", [("x", "X", "y")])
    surface = mcp_server.build_surface(bundle=tmp_path / "b", roots=[])
    assert surface.resolve(None).bundle_id == "solo-notes"


def test_a_one_to_many_call_naming_no_bundle_is_refused_rather_than_guessed(
    tmp_path: Path,
) -> None:
    """Picking one would make an answer's provenance depend on directory order."""
    root = tmp_path / "root"
    gate.write_bundle(root / "a", "a-notes", [("x", "X", "y")])
    gate.write_bundle(root / "b", "b-notes", [("x", "X", "y")])
    surface = mcp_server.build_surface(bundle=None, roots=[root])
    with pytest.raises(mcp_server.ToolError) as raised:
        surface.resolve(None)
    assert raised.value.code == "bundle_id_required"


def test_the_one_to_one_server_offers_no_listing_tool(tmp_path: Path) -> None:
    """A tool that always returns the same single row invites a client to treat
    discovery as available where the deployment does not have it."""
    gate.write_bundle(tmp_path / "b", "solo-notes", [("x", "X", "y")])
    one = mcp_server.build_surface(bundle=tmp_path / "b", roots=[])
    many = mcp_server.build_surface(bundle=None, roots=[tmp_path])
    assert [tool.name for tool in mcp_server.tools(one)] == list(gate.REQUIRED_TOOLS["one-to-one"])
    assert [tool.name for tool in mcp_server.tools(many)] == list(
        gate.REQUIRED_TOOLS["one-to-many"]
    )


def test_every_tool_carries_a_written_reason() -> None:
    """The order's rule: a tool with no reason written down is a tool nobody
    has to justify keeping."""
    surface = mcp_server.Surface(
        (Path("/nonexistent"),), None, mcp_server.okf_consume.DEFAULT_PROFILE
    )
    for tool in mcp_server.tools(surface):
        assert "Exists because" in tool.description, tool.name
        assert len(tool.description) > 200, tool.name


# --- the one-to-many skill candidate ------------------------------------------


def test_the_generic_skill_carries_no_bundles_identity(tmp_path: Path) -> None:
    """The property that makes the candidate's whole claim checkable: it takes
    no argument, so there is no bundle it could have read.

    Controlled against a per-bundle skill, which must carry exactly what this
    one does not -- without that control a test asserting an absence would pass
    on an empty string.
    """
    generic = skill.render_generic()
    skill.generate(GOLDEN, out=tmp_path / "per-bundle", question="krav", force=True)
    per_bundle = (tmp_path / "per-bundle" / "SKILL.md").read_text(encoding="utf-8")

    bundle_id = "b-golden-segmented-okf-v0-2"
    ref = mcp_server.okf_consume.bundle_ref(GOLDEN)
    assert bundle_id in per_bundle and ref in per_bundle
    assert bundle_id not in generic and ref not in generic
    assert skill.CARD_COMMAND in generic


def test_the_generic_skill_is_the_same_bytes_for_anyone(tmp_path: Path) -> None:
    """Two calls, and a written file, all identical. A rebuild of any bundle
    cannot make this artefact wrong, which is stronger than refusing loudly."""
    first = skill.render_generic()
    written = skill.generate_generic(out=tmp_path / "g")
    assert written.read_text(encoding="utf-8") == first == skill.render_generic()


def test_the_generic_skill_leaves_no_per_corpus_hole() -> None:
    """A hole left in a generic document is a number the reader is invited to
    invent -- which is the unfilled template's own defect."""
    assert skill._PLACEHOLDER.findall(skill.render_generic()) == []


def test_the_generic_skill_keeps_every_section_the_checker_reads() -> None:
    from llm_ingestion_okf import contract_check

    text = skill.render_generic()
    for section in contract_check.REQUIRED_SECTIONS:
        assert f"## {section}" in text, section