"""The MCP gate for exposing OKF bundles (capability loop, step 3). One command, one exit code. It asks whether a bundle can be reached over the Model Context Protocol in two shapes -- one server in front of ONE bundle, and one server that knows no bundle by name -- and it measures the thing the operator actually asked about: what has to be rebuilt when a bundle is rebuilt, and whether a stale artefact refuses out loud or answers quietly. WRITTEN RED, before any capability. Nothing in this module serves a bundle, ranks a concept or writes a skill; it only measures. The architecture choice -- one-to-one, one-to-many, skill, MCP, or both -- is the operator's, and this file exists so that choice is made on numbers. THE SERVER IS A SUBPROCESS AND NEVER AN IMPORT. Every row below speaks newline-delimited JSON-RPC over the server's stdin/stdout, starting with the `initialize` handshake. Importing the server's functions and calling them would measure the functions; the thing a client meets is the process, its framing and its error envelope, and those are what break. DENOMINATORS DO NOT COME FROM THE RUN. `REQUIRED_TOOLS`, `DRILL_ARTEFACTS`, `DISCOVERY_CHECKS`, `CROSS_CHECKS` and `HOSTILE_CASES` are pinned here. A row that counted whatever the server happened to offer would go green by offering less -- the defect this repository has already met once, in the retrieval gate's rows 2 and 3. THE QUESTION SET IS AN INPUT, NEVER A CONSTANT HERE. This repository is public and the graded set names a consumer's documents, so row 2 takes a path plus the freeze file that pins it, verifies the sha256 itself, and stays RED with a stated reason when it is not supplied or does not match. What IS committed is the synthetic corpus this module generates, whose subject matter is invented here and names no real document. SILENT IS RED REGARDLESS (row 3). An artefact that has gone stale may refuse, and may be inherently incapable of going stale. It may not answer as though nothing happened: a confident answer from yesterday's bundle is the one failure mode a consumer cannot detect from the outside. """ from __future__ import annotations import argparse import hashlib import json import os import re import subprocess import sys import tempfile import threading import time from collections.abc import Iterator, Mapping, Sequence from contextlib import contextmanager, suppress from dataclasses import dataclass, field from pathlib import Path from typing import Any TOOLS = Path(__file__).resolve().parent REPO = TOOLS.parent if str(REPO / "src") not in sys.path: sys.path.insert(0, str(REPO / "src")) GREEN = "GREEN" RED = "RED" SERVER_MODULE = "llm_ingestion_okf.mcp_server" # The protocol revision this gate speaks. A server answering `initialize` with # a different one is not a finding by itself -- the spec allows a server to # name the revision it supports -- but the handshake must complete. PROTOCOL_VERSION = "2025-06-18" # The minimum tool set per variant, with the reason each one exists. Pinned # here so a server cannot pass row 1 by shipping fewer tools. REQUIRED_TOOLS: Mapping[str, tuple[str, ...]] = { # A server in front of one bundle still has to say WHICH bundle, or an # answer cannot be attributed; and it has to hand over a concept's bytes, # because a ranked cut is a selection and row 2 asks about presence. "one-to-one": ("okf_describe", "okf_ask", "okf_fetch"), # A server that knows no bundle by name needs one more: the discovery # call. Everything else takes the bundle as an argument. "one-to-many": ("okf_list", "okf_describe", "okf_ask", "okf_fetch"), } # Every tool answer must carry both, or an arm cannot fill the answer contract: # which bundle the claim comes from, and which concept inside it. BUNDLE_KEY = "bundle_id" SOURCE_KEY = "sources" # Row 3 measures four artefact classes, not two. The operator's question was # about maintenance, and the skill is the incumbent -- comparing MCP against # nothing would answer half of it. DRILL_ARTEFACTS: tuple[str, ...] = ( "mcp-one-to-one", "mcp-one-to-many", "skill-one-to-one", "skill-one-to-many", ) DISCOVERY_CHECKS: tuple[str, ...] = ("list", "describe", "fetch") DISCOVERY_BUNDLES = 3 CROSS_CHECKS: tuple[str, ...] = ( "both-bundles-named", "both-sources-carried", "documented-sequence", ) #: Each hostile case with the refusal CODES that count as the right refusal. #: A code set rather than a bare "was refused": the first run of this gate had #: the 10 MB concept refused as `concept_unknown`, because the fixture wrote #: the file without naming it in the index -- the size ceiling never ran, and #: the row was green for a reason that had nothing to do with the attack. Two #: checks giving the same verdict are not the same guarantee. HOSTILE_CASES: Mapping[str, frozenset[str]] = { "traversal-in-bundle-id": frozenset({"bundle_unknown", "path_escape", "bundle_id_invalid"}), "traversal-in-concept-id": frozenset({"concept_unknown", "path_escape"}), "symlink-out-of-root": frozenset({"bundle_unknown", "path_escape"}), "broken-manifest": frozenset({"bundle_unreadable"}), "oversized-concept": frozenset({"concept_too_large"}), "unknown-bundle-id": frozenset({"bundle_unknown"}), } # A concept large enough that reading it whole is a decision rather than an # accident. The order names 10 MB; the gate writes exactly that. OVERSIZED_BYTES = 10 * 1024 * 1024 CALL_TIMEOUT = 30.0 class GateUsage(Exception): """Wrong input to the gate itself: exit 2, never a red row.""" class RpcError(Exception): """The server answered with a JSON-RPC error. That is a LOUD refusal and several rows want exactly this.""" def __init__(self, code: int, message: str) -> None: super().__init__(f"{code}: {message}") self.code = code self.message = message class ServerGone(Exception): """The process died, or said nothing within the timeout.""" # -------------------------------------------------------------------------- # A JSON-RPC stdio client, written narrowly on purpose. # -------------------------------------------------------------------------- class Stdio: """Newline-delimited JSON-RPC 2.0 over a child process's stdio. Hand-written because the gate must not share a code path with the thing it judges: a client built from the server's own framing helpers would agree with the server by construction. """ def __init__(self, argv: Sequence[str], *, cwd: Path | None = None) -> None: env = dict(os.environ) env["PYTHONPATH"] = os.pathsep.join( [str(REPO / "src"), *([env["PYTHONPATH"]] if env.get("PYTHONPATH") else [])] ) self.proc = subprocess.Popen( list(argv), stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE, cwd=str(cwd) if cwd else None, env=env, text=True, encoding="utf-8", bufsize=1, ) self._next_id = 0 self._stderr: list[str] = [] self._pump = threading.Thread(target=self._drain_stderr, daemon=True) self._pump.start() def _drain_stderr(self) -> None: assert self.proc.stderr is not None for line in self.proc.stderr: self._stderr.append(line.rstrip("\n")) @property def stderr_tail(self) -> str: return " | ".join(self._stderr[-4:]) def _send(self, message: Mapping[str, Any]) -> None: assert self.proc.stdin is not None if self.proc.poll() is not None: raise ServerGone(f"exited {self.proc.returncode}: {self.stderr_tail}") try: self.proc.stdin.write(json.dumps(message, ensure_ascii=False) + "\n") self.proc.stdin.flush() except (BrokenPipeError, ValueError) as error: # pragma: no cover - race raise ServerGone(f"{error}: {self.stderr_tail}") from error def _read(self, timeout: float) -> dict[str, Any]: assert self.proc.stdout is not None result: list[str] = [] def reader() -> None: line = self.proc.stdout.readline() # type: ignore[union-attr] result.append(line) thread = threading.Thread(target=reader, daemon=True) thread.start() thread.join(timeout) if thread.is_alive(): raise ServerGone(f"no answer within {timeout}s: {self.stderr_tail}") if not result or not result[0]: raise ServerGone(f"stdout closed: {self.stderr_tail}") try: parsed = json.loads(result[0]) except json.JSONDecodeError as error: raise ServerGone(f"not JSON: {result[0][:120]!r}") from error if not isinstance(parsed, dict): raise ServerGone(f"not a JSON-RPC object: {result[0][:120]!r}") return parsed def request( self, method: str, params: Mapping[str, Any] | None = None, *, timeout: float = CALL_TIMEOUT ) -> dict[str, Any]: self._next_id += 1 message: dict[str, Any] = {"jsonrpc": "2.0", "id": self._next_id, "method": method} if params is not None: message["params"] = params self._send(message) deadline = time.monotonic() + timeout while True: answer = self._read(max(0.1, deadline - time.monotonic())) if "id" not in answer: # a notification from the server; skip it continue if answer.get("id") != self._next_id: continue if "error" in answer: error = answer["error"] raise RpcError(int(error.get("code", 0)), str(error.get("message", ""))) payload = answer.get("result") if not isinstance(payload, dict): raise ServerGone(f"result is not an object: {answer!r}") return payload def notify(self, method: str, params: Mapping[str, Any] | None = None) -> None: message: dict[str, Any] = {"jsonrpc": "2.0", "method": method} if params is not None: message["params"] = params self._send(message) def handshake(self) -> dict[str, Any]: result = self.request( "initialize", { "protocolVersion": PROTOCOL_VERSION, "capabilities": {}, "clientInfo": {"name": "okf-mcp-gate", "version": "1"}, }, ) self.notify("notifications/initialized") return result def tool_names(self) -> tuple[str, ...]: listed = self.request("tools/list").get("tools", []) return tuple(str(tool.get("name", "")) for tool in listed if isinstance(tool, dict)) def call(self, name: str, arguments: Mapping[str, Any]) -> dict[str, Any]: """One `tools/call`, unwrapped to the object the tool returned. A tool that fails INSIDE the envelope (`isError: true`) is raised as an RpcError too: to a client the two are the same refusal, and a gate that treated one as loud and the other as silent would be scoring the server's transport taste. """ result = self.request("tools/call", {"name": name, "arguments": dict(arguments)}) text = "" for block in result.get("content", []) or []: if isinstance(block, dict) and block.get("type") == "text": text = str(block.get("text", "")) break if result.get("isError"): raise RpcError(-32000, text or "isError") structured = result.get("structuredContent") if isinstance(structured, dict): return structured try: parsed = json.loads(text) except json.JSONDecodeError as error: raise ServerGone(f"tool {name} returned no readable object: {text[:120]!r}") from error if not isinstance(parsed, dict): raise ServerGone(f"tool {name} returned {type(parsed).__name__}, not an object") return parsed def close(self) -> None: with suppress(Exception): assert self.proc.stdin is not None self.proc.stdin.close() try: self.proc.wait(timeout=5) except subprocess.TimeoutExpired: # pragma: no cover - only on a hang self.proc.kill() self.proc.wait(timeout=5) @contextmanager def server(argv: Sequence[str]) -> Iterator[Stdio]: client = Stdio(argv) try: yield client finally: client.close() def variant_argv(variant: str, target: Path) -> list[str]: flag = "--bundle" if variant == "one-to-one" else "--root" return [sys.executable, "-m", SERVER_MODULE, flag, str(target)] def server_exists() -> tuple[bool, str]: """Measured, not assumed: ask the interpreter to import the module. `0 of N` is a measurement only when the query behind it is shown, so the reason string carries the interpreter's own words. """ probe = subprocess.run( [sys.executable, "-c", f"import {SERVER_MODULE}"], capture_output=True, text=True, env={**os.environ, "PYTHONPATH": str(REPO / "src")}, cwd=str(REPO), check=False, ) if probe.returncode == 0: return True, "module imports" tail = (probe.stderr or "").strip().splitlines() return False, tail[-1] if tail else f"exit {probe.returncode}" # -------------------------------------------------------------------------- # Synthetic bundles. Invented subject matter, no real document named. # -------------------------------------------------------------------------- CONCEPT = """--- type: reference title: {title} source_file: {source} ingested_at: 2026-01-01T00:00:00Z generated: true sources: [{{ resource: {source}, title: {source} }}] adjudication: proposed bundle_id: {bundle_id} segment_id: {segment} source_lines: [1, 4] --- ## {title} {body} """ def write_bundle(root: Path, bundle_id: str, concepts: Sequence[tuple[str, str, str]]) -> Path: """`concepts` is (slug, title, body). One flat directory, one index.""" root.mkdir(parents=True, exist_ok=True) lines = [ "---", "okf_version: 0.2", f"bundle_id: {bundle_id}", "---", "", ] for position, (slug, title, body) in enumerate(concepts, start=1): (root / f"{slug}.md").write_text( CONCEPT.format( title=title, source=f"{slug}.txt", bundle_id=bundle_id, segment=f"s{position}", body=body, ), encoding="utf-8", ) lines.append(f"- [{title}]({slug}.md) — adjudication: proposed") (root / "index.md").write_text("\n".join(lines) + "\n", encoding="utf-8") return root def corpus(scratch: Path) -> dict[str, Path]: """Two bundles that share no vocabulary, so a cross-bundle answer cannot be produced by one of them alone.""" bridges = write_bundle( scratch / "bundles" / "bridge-notes", "bridge-notes", [ ("spennvidde", "Spennvidde for gangbru", "En gangbru med spennvidde 24 meter."), ("rekkverk", "Rekkverk paa gangbru", "Rekkverket skal vaere 1,2 meter hoeyt."), ], ) kitchens = write_bundle( scratch / "bundles" / "kitchen-notes", "kitchen-notes", [ ("surdeig", "Surdeig og heving", "Surdeigen hever i 14 timer ved 22 grader."), ("ovn", "Ovnstemperatur", "Ovnen forvarmes til 250 grader."), ], ) return {"bridge-notes": bridges, "kitchen-notes": kitchens} # -------------------------------------------------------------------------- # Rows # -------------------------------------------------------------------------- @dataclass class Row: number: int name: str k: int m: int status: str reason: str details: list[str] = field(default_factory=list) @property def fails(self) -> bool: return self.status != GREEN def to_json(self) -> dict[str, Any]: return { "row": self.number, "name": self.name, "k": self.k, "m": self.m, "status": self.status, "reason": self.reason, "details": self.details, } def _row(number: int, name: str, k: int, m: int, reason: str, details: list[str]) -> Row: """`m == 0` is NEVER green: a row with nothing to count did not measure.""" return Row(number, name, k, m, GREEN if m > 0 and k == m else RED, reason, details) # Which keys an answer must carry, per tool. `okf_list` and `okf_describe` # answer ABOUT a bundle; the other two answer FROM one, and an answer from a # bundle without a concept id cannot be cited. ANSWER_KEYS: Mapping[str, tuple[str, ...]] = { "okf_list": (BUNDLE_KEY,), "okf_describe": (BUNDLE_KEY,), "okf_ask": (BUNDLE_KEY, SOURCE_KEY), "okf_fetch": (BUNDLE_KEY, SOURCE_KEY), } def _bundle_ids(answer: Mapping[str, Any]) -> set[str]: """Every bundle id anywhere in the answer, at any depth. A `list` answer carries one id per bundle and an `ask` answer carries one per excerpt, so a rule reading only the top level would score the shapes rather than the contract. """ found: set[str] = set() def walk(node: Any) -> None: if isinstance(node, Mapping): value = node.get(BUNDLE_KEY) if isinstance(value, str) and value: found.add(value) for child in node.values(): walk(child) elif isinstance(node, (list, tuple)): for child in node: walk(child) walk(answer) return found def _source_ids(answer: Mapping[str, Any]) -> set[str]: found: set[str] = set() def walk(node: Any) -> None: if isinstance(node, Mapping): for key in ("concept_id", "source_id"): value = node.get(key) if isinstance(value, str) and value: found.add(value) for child in node.values(): walk(child) elif isinstance(node, (list, tuple)): for child in node: walk(child) walk(answer) return found _REFUSAL_CODE = re.compile(r"refused \(([a-z_]+)\)") def refusal_code(error: RpcError) -> str: """The code a refusal carries, or `""`. Read from the message because that is where a client sees it: a tool-level refusal travels in the result envelope, and its JSON-RPC number is the same for every one of them. """ match = _REFUSAL_CODE.search(error.message) return match.group(1) if match else "" def _probe_arguments(tool: str, bundle_id: str, concept_id: str, *, named: bool) -> dict[str, Any]: """One representative call per tool. `named` is False for the one-to-one variant, whose bundle is fixed at startup and takes no bundle argument.""" bundle: dict[str, Any] = {BUNDLE_KEY: bundle_id} if named else {} if tool == "okf_list": return {} if tool == "okf_describe": return bundle if tool == "okf_ask": return {**bundle, "question": "spennvidde for gangbru"} return {**bundle, "concept_id": concept_id} def _first_concept_id(client: Stdio, bundle_id: str, *, named: bool) -> str: answer = client.call("okf_describe", {BUNDLE_KEY: bundle_id} if named else {}) for key in ("concepts", "concept_ids"): value = answer.get(key) if isinstance(value, list) and value: head = value[0] if isinstance(head, str): return head if isinstance(head, Mapping) and isinstance(head.get("concept_id"), str): return str(head["concept_id"]) raise ServerGone("okf_describe named no concept, so okf_fetch has no argument") def row_one(bundles: Mapping[str, Path], reachable: tuple[bool, str]) -> Row: """The handshake, the tool list, and one call per required tool. `m` is the pinned roster across both variants, so a server shipping three tools where four are required scores 3 of 7 rather than 3 of 3. """ m = sum(len(names) for names in REQUIRED_TOOLS.values()) details: list[str] = [] if not reachable[0]: return _row( 1, "protocol: initialize -> tools/list -> tools/call over stdio", 0, m, f"`python -c 'import {SERVER_MODULE}'` fails: {reachable[1]}", [f"roster: {json.dumps(REQUIRED_TOOLS)}"], ) k = 0 for variant, required in REQUIRED_TOOLS.items(): named = variant == "one-to-many" target = bundles["bridge-notes"] if not named else bundles["bridge-notes"].parent try: with server(variant_argv(variant, target)) as client: client.handshake() listed = set(client.tool_names()) concept = "" with suppress(Exception): concept = _first_concept_id(client, "bridge-notes", named=named) for tool in required: if tool not in listed: details.append(f"{variant}: {tool} not in tools/list") continue try: answer = client.call( tool, _probe_arguments(tool, "bridge-notes", concept, named=named) ) except (RpcError, ServerGone) as error: details.append(f"{variant}: {tool} did not answer: {error}") continue missing = [] if BUNDLE_KEY in ANSWER_KEYS[tool] and not _bundle_ids(answer): missing.append(BUNDLE_KEY) if SOURCE_KEY in ANSWER_KEYS[tool] and not _source_ids(answer): missing.append("concept_id") if missing: details.append(f"{variant}: {tool} answered without {', '.join(missing)}") continue k += 1 except (ServerGone, OSError) as error: details.append(f"{variant}: server did not start: {error}") return _row( 1, "protocol: initialize -> tools/list -> tools/call over stdio", k, m, "every required tool answers over real stdio, carrying bundle id and concept id", details, ) @dataclass(frozen=True) class AnchorSet: """A (bundle, anchor, quote) triple set, read from a frozen question file. The set is never committed here. `path` and `sha256` are the whole registration, and a set whose bytes moved is a usage error, not a red row: a gate that measured a moving set would be reporting on nothing. """ label: str pairs: tuple[tuple[str, str, str], ...] version: int note: str def read_anchor_set(questions: Path, freeze: Path, *, want_version: int) -> AnchorSet: if not questions.is_file(): raise GateUsage(f"question set not found: {questions}") if not freeze.is_file(): raise GateUsage(f"freeze file not found: {freeze}") frozen = json.loads(freeze.read_text(encoding="utf-8")) declared = frozen.get("sha256", {}).get(questions.name) measured = hashlib.sha256(questions.read_bytes()).hexdigest() version = int(frozen.get("versjon", 0)) if declared != measured: raise GateUsage( f"{questions.name} is not the file {freeze.name} pins: " f"declared {declared}, measured {measured}" ) if version < want_version: raise GateUsage( f"{freeze.name} declares version {version}; this gate was ordered " f"against version {want_version}" ) document = json.loads(questions.read_text(encoding="utf-8")) pairs: dict[tuple[str, str], str] = {} for question in document.get("sporsmal", []): # The set names its bundles as a LIST per question, and a question may # name several. A pair is (bundle, anchor), so a question naming two # bundles and one anchor is two pairs: an anchor reachable in one # bundle and not the other is two different facts. named = [str(entry) for entry in (question.get("bundles") or []) if entry] for atom in question.get("atomer", []) or []: anchor = atom.get("kilde_anker") quote = str(atom.get("kilde_sitat") or "") if isinstance(anchor, str) and anchor: for bundle in named: if pairs.get((bundle, anchor), "") == "": pairs[(bundle, anchor)] = quote for cite in question.get("must_cite", []) or []: entries = cite if isinstance(cite, list) else [cite] for entry in entries: if isinstance(entry, str) and entry: for bundle in named: pairs.setdefault((bundle, entry), "") return AnchorSet( label=f"{questions.parent.name}/{questions.name}", pairs=tuple((bundle, anchor, quote) for (bundle, anchor), quote in sorted(pairs.items())), version=version, note=str(frozen.get("merknad", ""))[:120], ) # The row-2 machinery's known-positive, over the synthetic corpus: four # (bundle, anchor, quote) triples this file invented, whose text the tool # surface must be able to hand back verbatim. Without it, a row reporting # `0 of M` could be reporting that the lookup was never able to find anything. SYNTHETIC_ANCHORS: tuple[tuple[str, str, str], ...] = ( ("bridge-notes", "spennvidde", "En gangbru med spennvidde 24 meter."), ("bridge-notes", "rekkverk", "Rekkverket skal vaere 1,2 meter hoeyt."), ("kitchen-notes", "surdeig", "Surdeigen hever i 14 timer ved 22 grader."), ("kitchen-notes", "ovn", "Ovnen forvarmes til 250 grader."), ) _WHITESPACE = re.compile(r"\s+") def fold(text: str) -> str: """The one normalisation both sides of a quote comparison get. Two removals and nothing else. U+00AD, because `okf build` strips soft hyphens from extracted text (`extract.normalise_extracted`) while the publisher's own JSON keeps them, so a quote carrying one could never match text that is otherwise identical. And whitespace runs, because a quote cut out of a paragraph carries the line breaks of wherever it was cut. Case is NOT folded and no character is transliterated: a quote is a quote. """ return _WHITESPACE.sub(" ", text.replace("\u00ad", "")).strip() def _reach_anchor(client: Stdio, bundle_id: str, anchor: str, quote: str) -> tuple[bool, str]: """A documented two-step sequence for one anchor, with the route recorded. Step one asks whether the anchor IS a concept id -- the cheap case, and a true ceiling. Step two asks the surface the anchor as a question and reads the delivered excerpts. Step two goes through the ranker, so a pair met only there is a FLOOR on the ceiling and never the ceiling itself: the row prints both counts rather than one. """ needle = fold(quote) try: answer = client.call("okf_fetch", {BUNDLE_KEY: bundle_id, "concept_id": anchor}) except (RpcError, ServerGone): pass else: if not needle or needle in fold(json.dumps(answer, ensure_ascii=False)): return True, "fetch" if not needle: return False, "no quote to look for" try: answer = client.call("okf_ask", {BUNDLE_KEY: bundle_id, "question": anchor}) except (RpcError, ServerGone) as error: return False, f"ask refused: {error}" return (needle in fold(json.dumps(answer, ensure_ascii=False))), "ask" def _fetch_carries(client: Stdio, bundle_id: str, anchor: str, quote: str, *, named: bool) -> bool: """The synthetic known-positive's route: fetch by id, nothing else.""" arguments: dict[str, Any] = {"concept_id": anchor} if named: arguments[BUNDLE_KEY] = bundle_id try: answer = client.call("okf_fetch", arguments) except (RpcError, ServerGone): return False text = json.dumps(answer, ensure_ascii=False) return quote in text if quote else bool(_source_ids(answer)) def present_in_bundle(bundle_root: Path, quotes: Sequence[str]) -> list[bool]: """Is the quote anywhere in the bundle's concept bodies? Read with the LIBRARY rather than the surface, on purpose: this separates "the bundle does not carry it" from "the surface could not reach it", and the order asks for the first to be reported as a fact about the bundle rather than a defect in the server. """ from llm_ingestion_okf import consume as okf_consume haystack = fold( "\n".join( (bundle_root / f"{concept_id}.md").read_text(encoding="utf-8", errors="replace") for concept_id in okf_consume.enumerate_concepts(bundle_root) ) ) return [bool(quote) and fold(quote) in haystack for quote in quotes] def row_two( bundles: Mapping[str, Path], reachable: tuple[bool, str], anchors: AnchorSet | None, real: Mapping[str, Path] | None = None, ) -> Row: details: list[str] = [] known_positive = 0 if reachable[0]: root = next(iter(bundles.values())).parent try: with server(variant_argv("one-to-many", root)) as client: client.handshake() for bundle_id, anchor, quote in SYNTHETIC_ANCHORS: if _fetch_carries(client, bundle_id, anchor, quote, named=True): known_positive += 1 except (ServerGone, OSError) as error: details.append(f"known-positive could not run: {error}") details.append( f"known-positive (synthetic, this file's own text): " f"{known_positive} of {len(SYNTHETIC_ANCHORS)} anchors fetched verbatim" ) name = "coverage ceiling: every anchor the frozen set points at, fetched verbatim" if anchors is None: return _row( 2, name, 0, 0, "no frozen question set supplied (--sett/--frys); the denominator is " "counted from that file at run time and is not known here", details, ) m = len(anchors.pairs) details.append( f"set {anchors.label}, freeze version {anchors.version}, " f"M = {m} (bundle, anchor) pairs counted from the file" ) if not real: return _row( 2, name, 0, m, "no bundle root supplied (--bundle-root); the set's bundles are not " "committed here and the row cannot be measured without them", details, ) if not reachable[0]: return _row(2, name, 0, m, f"server missing: {reachable[1]}", details) by_bundle: dict[str, list[tuple[str, str]]] = {} for bundle_id, anchor, quote in anchors.pairs: by_bundle.setdefault(bundle_id, []).append((anchor, quote)) root = next(iter(real.values())).parent k = 0 carried = 0 routes = {"fetch": 0, "ask": 0} with server(variant_argv("one-to-many", root)) as client: client.handshake() served = {str(entry) for entry in _bundle_ids(client.call("okf_list", {}))} for bundle_id in sorted(by_bundle): entries = by_bundle[bundle_id] target = real.get(bundle_id) if target is None: details.append( f"{bundle_id}: RED for the BUNDLE -- no root supplied holds it " f"({len(entries)} of {m} pairs)" ) continue served_id = _served_name(target, served) if served_id is None: details.append( f"{bundle_id}: RED for the BUNDLE -- {target.name} is not served " f"({len(entries)} of {m} pairs)" ) continue here = present_in_bundle(target, [quote for _anchor, quote in entries]) carried += sum(here) hit = 0 for (anchor, quote), _in_bundle in zip(entries, here, strict=True): met, route = _reach_anchor(client, served_id, anchor, quote) if met: hit += 1 routes[route] = routes.get(route, 0) + 1 k += hit details.append( f"{bundle_id} (served as `{served_id}`): {hit} of {len(entries)} reached; " f"{sum(here)} of {len(entries)} present in the bundle at all" ) details.append( f"routes: {routes.get('fetch', 0)} met by `okf_fetch` on the anchor as a concept id " f"(a true ceiling), {routes.get('ask', 0)} met only through `okf_ask` (a FLOOR on the " "ceiling: that route runs the ranker)" ) details.append( f"present in some bundle at all: {carried} of {m} -- a pair the bundle does not " "carry is red for the bundle, not for the server" ) return _row(2, name, k, m, "the ceiling an arm can reach, per (bundle, anchor) pair", details) def _served_name(target: Path, served: set[str]) -> str | None: """The id the server calls this directory, read from the bundle itself. Never guessed from the directory name: the frozen set names `n100-2023` and the bundle declares `vegnormal-n100-2023`, so a rule matching names would have to invent an alias rule of its own. """ from llm_ingestion_okf import consume as okf_consume declared = okf_consume.root_bundle_id_of(target) return declared if declared in served else None # -------------------------------------------------------------------------- # Row 3: the update drill. The operator's question, measured. # -------------------------------------------------------------------------- @dataclass(frozen=True) class DrillResult: artefact: str loud: bool recreate: tuple[str, ...] note: str def _mutate(bundle: Path) -> None: """Change one concept, so the bundle's fingerprint moves.""" target = sorted(p for p in bundle.rglob("*.md") if p.name != "index.md")[0] text = target.read_text(encoding="utf-8") target.write_text( text.rstrip("\n") + "\n\nEn setning som ikke sto her foer.\n", encoding="utf-8" ) def _drill_mcp(variant: str, bundle: Path, bundle_id: str) -> DrillResult: """A running server, a bundle rebuilt under it, and one question: does the answer move with the bytes, or does yesterday's answer keep coming?""" named = variant == "one-to-many" target = bundle if not named else bundle.parent try: with server(variant_argv(variant, target)) as client: client.handshake() before = client.call("okf_describe", {BUNDLE_KEY: bundle_id} if named else {}) _mutate(bundle) after = client.call("okf_describe", {BUNDLE_KEY: bundle_id} if named else {}) moved = before.get("ref") != after.get("ref") return DrillResult( f"mcp-{variant}", loud=bool(moved), recreate=(), note=( "the identity moved with the bytes; nothing to recreate" if moved else f"SILENT: still reports ref {before.get('ref')!r} after the rebuild" ), ) except (ServerGone, OSError, RpcError) as error: return DrillResult(f"mcp-{variant}", loud=False, recreate=(), note=f"did not run: {error}") def _drill_skill_one_to_one(bundle: Path, scratch: Path) -> DrillResult: """Today's generated skill: it carries the fingerprint in its own prose, so a rebuilt bundle leaves it declaring an identity the payload no longer has. `okf check`'s `bundle_mismatch` is the loud refusal, and regenerating it is the cost.""" try: from llm_ingestion_okf import consume as okf_consume from llm_ingestion_okf import contract_check, skill except ImportError as error: # pragma: no cover - the package is always present return DrillResult("skill-one-to-one", loud=False, recreate=(), note=str(error)) out = scratch / "skill-1-1" try: skill.generate(bundle, out=out, question="spennvidde for gangbru", force=True) except Exception as error: return DrillResult( "skill-one-to-one", loud=False, recreate=(), note=f"not generated: {error}" ) declared = contract_check.skill_identity((out / "SKILL.md").read_text(encoding="utf-8")) _mutate(bundle) current = okf_consume.bundle_ref(bundle) loud = declared is not None and declared[1] != current return DrillResult( "skill-one-to-one", loud=loud, recreate=("okf skill --out --force",), note=( f"declares ref {declared[1][:16] if declared else None}..., bundle now " f"{current[:16]}...; `okf check --skill` refuses (bundle_mismatch)" if loud else "the stale skill declares an identity no rule can compare" ), ) def _drill_skill_one_to_many(bundles: Mapping[str, Path], scratch: Path) -> DrillResult: """The candidate: one installable skill with no bundle-specific number in it, and the per-bundle numbers read at run time. Its staleness test is not a refusal -- it is that the artefact cannot go stale, which is stronger. Proved by a PROPERTY and not by reading it: the text generated for two bundles that share nothing must be byte-identical, and must carry neither bundle's id or ref. """ try: from llm_ingestion_okf import consume as okf_consume from llm_ingestion_okf import skill except ImportError as error: # pragma: no cover return DrillResult("skill-one-to-many", loud=False, recreate=(), note=str(error)) render_generic = getattr(skill, "render_generic", None) if render_generic is None: return DrillResult( "skill-one-to-many", loud=False, recreate=(), note="`skill.render_generic` does not exist: the candidate has not been built", ) first, second = (bundles["bridge-notes"], bundles["kitchen-notes"]) text_first = render_generic() text_second = render_generic() identical = text_first == text_second leaked = [ token for token in ( "bridge-notes", "kitchen-notes", okf_consume.bundle_ref(first), okf_consume.bundle_ref(second), ) if token in text_first ] (scratch / "skill-1-many").mkdir(parents=True, exist_ok=True) (scratch / "skill-1-many" / "SKILL.md").write_text(text_first, encoding="utf-8") return DrillResult( "skill-one-to-many", loud=identical and not leaked, recreate=(), note=( "carries no bundle id, no ref and no count, so a rebuild cannot make it wrong" if identical and not leaked else f"bundle-specific after all: {', '.join(leaked) or 'not reproducible'}" ), ) def row_three(scratch: Path) -> Row: """Scenario 1: one concept changes. Per artefact class -- does the stale copy refuse out loud, and what has to be made again?""" m = len(DRILL_ARTEFACTS) details: list[str] = [] results: list[DrillResult] = [] for artefact in DRILL_ARTEFACTS: stage = scratch / "drill" / artefact stage.mkdir(parents=True, exist_ok=True) bundles = corpus(stage) if artefact == "mcp-one-to-one": results.append(_drill_mcp("one-to-one", bundles["bridge-notes"], "bridge-notes")) elif artefact == "mcp-one-to-many": results.append(_drill_mcp("one-to-many", bundles["bridge-notes"], "bridge-notes")) elif artefact == "skill-one-to-one": results.append(_drill_skill_one_to_one(bundles["bridge-notes"], stage)) else: results.append(_drill_skill_one_to_many(bundles, stage)) for result in results: details.append( f"{result.artefact}: {'LOUD' if result.loud else 'SILENT/absent'}, " f"{len(result.recreate)} artefact(s) to recreate, " f"{len(result.recreate)} manual step(s) -- {result.note}" ) return _row( 3, "update drill 1: one concept changes; stale must refuse, never answer quietly", sum(1 for result in results if result.loud), m, "a silent stale answer is RED regardless of every other number", details, ) def row_four(scratch: Path) -> Row: """Scenario 2: three unknown bundles appear while the one-to-many server is RUNNING. No restart, no config, no code.""" m = DISCOVERY_BUNDLES * len(DISCOVERY_CHECKS) stage = scratch / "discovery" stage.mkdir(parents=True, exist_ok=True) bundles = corpus(stage) root = bundles["bridge-notes"].parent newcomers = { "ferry-notes": ("avgang", "Avgangstider", "Fergen gaar hver time fra klokken 06."), "lichen-notes": ("vekst", "Vekstrate", "Lav vokser under en millimeter i aaret."), "sextant-notes": ("horisont", "Horisontmaaling", "Kimingen justeres for oeyehoeyde."), } details: list[str] = [] k = 0 try: with server(variant_argv("one-to-many", root)) as client: client.handshake() first = _bundle_ids(client.call("okf_list", {})) details.append( f"before: {len(first)} bundle(s) -- {', '.join(sorted(first)) or 'none'}" ) for bundle_id, (slug, title, body) in newcomers.items(): write_bundle(root / bundle_id, bundle_id, [(slug, title, body)]) served = _bundle_ids(client.call("okf_list", {})) for bundle_id, (slug, _title, body) in newcomers.items(): if bundle_id in served: k += 1 else: details.append(f"{bundle_id}: not seen by a later `okf_list`") continue try: described = client.call("okf_describe", {BUNDLE_KEY: bundle_id}) if bundle_id in _bundle_ids(described): k += 1 else: details.append(f"{bundle_id}: okf_describe did not name it") except (RpcError, ServerGone) as error: details.append(f"{bundle_id}: okf_describe refused: {error}") if _fetch_carries(client, bundle_id, slug, body, named=True): k += 1 else: details.append(f"{bundle_id}: okf_fetch did not return its text") except (ServerGone, OSError, RpcError) as error: details.append(f"the running server could not be reached: {error}") return _row( 4, "update drill 2: three unknown bundles appear while the server runs", k, m, "discovery happens per call; a restart, a config edit or a code change is RED", details, ) def row_five(scratch: Path) -> Row: """One documented call sequence that answers from two bundles at once, with both sources. An arm that cannot do this can only ever answer from one.""" m = len(CROSS_CHECKS) stage = scratch / "cross" stage.mkdir(parents=True, exist_ok=True) bundles = corpus(stage) root = bundles["bridge-notes"].parent details: list[str] = [] k = 0 try: with server(variant_argv("one-to-many", root)) as client: client.handshake() calls = 0 answer = client.call("okf_ask", {"question": "spennvidde og surdeig"}) calls += 1 named = _bundle_ids(answer) sources = _source_ids(answer) if {"bridge-notes", "kitchen-notes"} <= named: k += 1 else: details.append(f"one answer named {sorted(named) or 'no bundle'}") if len(sources) >= 2: k += 1 else: details.append(f"one answer carried {len(sources)} source id(s)") if calls == 1: k += 1 details.append( f"documented sequence: {calls} call(s) -- okf_ask with no bundle argument" ) except (ServerGone, OSError, RpcError) as error: details.append(f"cross-bundle answer could not be produced: {error}") return _row( 5, "across bundles: one documented sequence, two bundles, both sources", k, m, "an answer spanning two bundles must carry both bundles' citations", details, ) def row_six(scratch: Path) -> Row: """Hostile input. Every case must be refused OUT LOUD, and nothing outside the served root may appear in any answer.""" m = len(HOSTILE_CASES) stage = scratch / "hostile" stage.mkdir(parents=True, exist_ok=True) bundles = corpus(stage) root = bundles["bridge-notes"].parent outside = stage / "outside" outside.mkdir(parents=True, exist_ok=True) canary = "HEMMELIGHET-SOM-ALDRI-SKAL-LESES" write_bundle(outside / "secret-notes", "secret-notes", [("hemmelig", "Hemmelig", canary)]) with suppress(OSError, NotImplementedError): (root / "linked-notes").symlink_to(outside / "secret-notes", target_is_directory=True) write_bundle(root / "broken-notes", "broken-notes", [("x", "X", "y")]) (root / "broken-notes" / "index.md").write_bytes(b"\xff\xfe not a manifest at all\x00") huge = root / "bridge-notes" / "svulmende.md" huge.write_text( CONCEPT.format( title="Svulmende", source="svulmende.txt", bundle_id="bridge-notes", segment="s9", body="A" * OVERSIZED_BYTES, ), encoding="utf-8", ) # Named in the index, or it is not a concept and the ceiling never runs: # the bundle's own index is what makes a file reachable at all. index = root / "bridge-notes" / "index.md" index.write_text( index.read_text(encoding="utf-8").rstrip("\n") + "\n- [Svulmende](svulmende.md) \u2014 adjudication: proposed\n", encoding="utf-8", ) details: list[str] = [] k = 0 attempts: list[tuple[str, str, dict[str, Any]]] = [ ("traversal-in-bundle-id", "okf_describe", {BUNDLE_KEY: "../outside/secret-notes"}), ( "traversal-in-concept-id", "okf_fetch", {BUNDLE_KEY: "bridge-notes", "concept_id": "../../outside/secret-notes/hemmelig"}, ), ("symlink-out-of-root", "okf_describe", {BUNDLE_KEY: "linked-notes"}), ("broken-manifest", "okf_describe", {BUNDLE_KEY: "broken-notes"}), ("oversized-concept", "okf_fetch", {BUNDLE_KEY: "bridge-notes", "concept_id": "svulmende"}), ("unknown-bundle-id", "okf_describe", {BUNDLE_KEY: "does-not-exist"}), ] try: with server(variant_argv("one-to-many", root)) as client: client.handshake() for case, tool, arguments in attempts: try: answer = client.call(tool, arguments) except RpcError as error: code = refusal_code(error) if code in HOSTILE_CASES[case]: k += 1 details.append(f"{case}: refused loudly ({code})") else: details.append( f"{case}: refused as `{code or error.code}`, which is not the " f"check this case attacks ({'/'.join(sorted(HOSTILE_CASES[case]))}) " f"-- {error}" ) continue except ServerGone as error: details.append(f"{case}: the server died instead of refusing ({error})") continue body = json.dumps(answer, ensure_ascii=False) if canary in body: details.append(f"{case}: READ OUTSIDE THE ROOT -- the canary is in the answer") else: details.append(f"{case}: answered instead of refusing ({len(body)} bytes)") except (ServerGone, OSError) as error: details.append(f"the server could not be reached: {error}") return _row( 6, "hostile input: traversal, symlink, broken manifest, 10 MB concept, unknown id", k, m, "every case refused out loud, and no byte from outside the served root in any answer", details, ) # -------------------------------------------------------------------------- # Verdict # -------------------------------------------------------------------------- LIMITS: tuple[str, ...] = ( "The gate measures a CEILING and a maintenance cost. It runs no arm, calls " "no model and ranks nothing: whether an arm answers well is a different " "question, asked by tools/okf_retrieval_gate.py.", "Row 2's denominator comes from the frozen question set at run time. " "Without --sett/--frys the row is red with m = 0, which is the honest " "shape: nothing was counted.", "Row 3 counts artefacts and manual steps INSIDE this machine. A consuming " "project that has installed a generated skill pays one more step per " "project, and that number is not measurable from here.", "The architecture choice is the operator's. These rows are the input to " "it, never the decision.", ) def evaluate( scratch: Path, *, anchors: AnchorSet | None = None, real: Mapping[str, Path] | None = None, ) -> list[Row]: reachable = server_exists() bundles = corpus(scratch / "base") return [ row_one(bundles, reachable), row_two(bundles, reachable, anchors, real), row_three(scratch), row_four(scratch), row_five(scratch), row_six(scratch), ] def render(rows: Sequence[Row]) -> str: lines = ["okf MCP gate", ""] for row in sorted(rows, key=lambda r: r.number): lines.append(f"{row.status:5} row {row.number}: {row.name}") lines.append(f" {row.k} of {row.m} -- {row.reason}") lines.extend(f" - {detail}" for detail in row.details) lines.append("") failing = [row.number for row in sorted(rows, key=lambda r: r.number) if row.fails] lines.append( f"GATE {RED}: rows {', '.join(str(number) for number in failing)}" if failing else f"GATE {GREEN}" ) lines.append("") lines.append("Limits:") lines.extend(f" - {limit}" for limit in LIMITS) return "\n".join(lines) + "\n" def _real_bundles(roots: Sequence[Path], anchors: AnchorSet | None) -> dict[str, Path]: """Map every bundle the set names to a directory under the given roots. The match is on the SET's name against the directory name's leading segment, and the served id is then read from the bundle itself. A set name matching two directories is a usage error: choosing one would make the measurement depend on directory order. """ if not roots or anchors is None: return {} wanted = sorted({bundle for bundle, _anchor, _quote in anchors.pairs}) found: dict[str, Path] = {} for name in wanted: hits = [ child for root in roots for child in sorted(root.iterdir()) if child.is_dir() and child.name.startswith(name) ] if len(hits) > 1: raise GateUsage( f"`{name}` matches {len(hits)} directories under the given roots: " f"{', '.join(hit.name for hit in hits)}" ) if hits: found[name] = hits[0] return found def main(argv: list[str] | None = None) -> int: parser = argparse.ArgumentParser(description=__doc__.split("\n", 1)[0]) parser.add_argument("--json", action="store_true", help="emit the rows as JSON") parser.add_argument( "--sett", type=Path, help="the frozen graded question set row 2 counts its denominator from", ) parser.add_argument( "--frys", type=Path, help="the freeze file that pins --sett by sha256 and declares its version", ) parser.add_argument( "--bundle-root", type=Path, action="append", default=[], help=( "a directory holding the set's bundles, for row 2 (repeatable). The " "bundles are never committed here; the row maps a set name to a served " "bundle by reading the bundle's own declared id" ), ) parser.add_argument( "--sett-versjon", type=int, default=4, help="the freeze version this gate was ordered against (default 4)", ) args = parser.parse_args(argv) try: anchors = ( read_anchor_set(args.sett, args.frys, want_version=args.sett_versjon) if args.sett or args.frys else None ) if (args.sett is None) != (args.frys is None): raise GateUsage("--sett and --frys are given together or not at all") real = _real_bundles(args.bundle_root, anchors) with tempfile.TemporaryDirectory(prefix="okf-mcp-gate-") as scratch: rows = evaluate(Path(scratch), anchors=anchors, real=real) except GateUsage as error: print(f"okf-mcp-gate: {error}", file=sys.stderr) return 2 # Broad on purpose, and the house pattern: a gate that dies with a # traceback must not be readable as a finding. except Exception as error: print(f"okf-mcp-gate: did not run: {type(error).__name__}: {error}", file=sys.stderr) return 2 if args.json: print( json.dumps( { "rows": [row.to_json() for row in sorted(rows, key=lambda r: r.number)], "required_tools": {k: list(v) for k, v in REQUIRED_TOOLS.items()}, "limits": list(LIMITS), "gate": RED if any(row.fails for row in rows) else GREEN, }, indent=2, ensure_ascii=False, ) ) else: print(render(rows), end="") return 1 if any(row.fails for row in rows) else 0 if __name__ == "__main__": raise SystemExit(main())