"""Read-only scripted MCP transport probe against the captured local Git bundle.""" import argparse from datetime import datetime, timezone import hashlib import json import os from pathlib import Path import queue import subprocess import tempfile import threading def finalize_capture(metadata, request_log, response_log, output): """Retain raw bytes and a summary even when the server emits invalid JSON.""" (output / "requests.jsonl").write_text("".join(request_log)) (output / "responses.jsonl").write_text("".join(response_log)) protocol_errors = [] invalid_responses = [] for number, line in enumerate(response_log, 1): try: message = json.loads(line) except json.JSONDecodeError as error: invalid_responses.append({"line": number, "error": str(error)}) continue if not isinstance(message, dict): invalid_responses.append({"line": number, "error": "JSON-RPC envelope must be an object"}) elif "error" in message: protocol_errors.append(message) metadata["protocol_errors"] = protocol_errors if invalid_responses: metadata["invalid_responses"] = invalid_responses metadata["outcome"] = "error" metadata.setdefault("error", "Server emitted an invalid JSON-RPC response; inspect responses.jsonl") elif metadata["outcome"] == "completed" and protocol_errors: metadata["outcome"] = "retrieval_completed_with_protocol_errors" metadata["finished_at_utc"] = datetime.now(timezone.utc).isoformat() metadata["transcript_sha256"] = {name: hashlib.sha256((output / name).read_bytes()).hexdigest() for name in ["requests.jsonl", "responses.jsonl", "stderr.txt"]} (output / "summary.json").write_text(json.dumps(metadata, indent=2) + "\n") def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--proof", required=True) parser.add_argument("--output", required=True, help="A new or empty output directory") parser.add_argument("--bundle", default=str(Path(__file__).parent / "results/fixture.bundle")) args = parser.parse_args() proof = str(Path(args.proof).resolve()) bundle = Path(args.bundle).resolve() output = Path(args.output).resolve() if output.exists() and any(output.iterdir()): parser.error("output directory must be new or empty") output.mkdir(parents=True, exist_ok=True) env = {key: value for key, value in os.environ.items() if not key.startswith(("GITHUB_", "GIT_", "CI_", "PROOF_", "REQPROOF_", "CODEX_", "CLAUDE_"))} env.update({"PROOF_NO_DOWNLOAD": "1", "GIT_CONFIG_GLOBAL": os.devnull, "GIT_CONFIG_NOSYSTEM": "1"}) metadata = { "kind": "scripted local MCP transport probe, not an agent evaluation", "started_at_utc": datetime.now(timezone.utc).isoformat(), "command": ["proof", "mcp", "--transport", "stdio"], "proof_version": subprocess.check_output([proof, "version"], text=True, env=env).strip(), "proof_binary_sha256": hashlib.sha256(Path(proof).read_bytes()).hexdigest(), "bundle_sha256": hashlib.sha256(bundle.read_bytes()).hexdigest(), "setup_commands": ["git clone ", "git remote remove origin", "git checkout C"], "identity": "A scripted client named retained-obligation-scripted-transport-probe. No agent or human approver identity was supplied.", } request_log = [] response_log = [] messages = queue.Queue() with tempfile.TemporaryDirectory(prefix="retained-mcp-probe-") as tmp: repo = Path(tmp) / "fixture" def git(*argv): return subprocess.check_output(["git", *argv], cwd=repo, env=env, text=True).strip() subprocess.run(["git", "clone", "--quiet", str(bundle), str(repo)], env=env, check=True) git("remote", "remove", "origin") git("checkout", "--quiet", "C") metadata["fixture_commit"] = git("rev-parse", "HEAD") metadata["remotes"] = git("remote") metadata["tracked_status_before"] = git("status", "--porcelain", "--untracked-files=no") with (output / "stderr.txt").open("w") as stderr: process = subprocess.Popen([proof, "mcp", "--transport", "stdio"], cwd=repo, env=env, stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=stderr, text=True, bufsize=1) def read_stdout(): for line in process.stdout: response_log.append(line) try: messages.put(json.loads(line)) except json.JSONDecodeError as error: messages.put({"probe_parse_error": str(error), "line": line}) messages.put({"probe_eof": True}) reader = threading.Thread(target=read_stdout, daemon=True) reader.start() def send(request): encoded = json.dumps(request, separators=(",", ":")) + "\n" request_log.append(encoded) process.stdin.write(encoded) process.stdin.flush() if "id" not in request: return None while True: response = messages.get(timeout=15) if "probe_parse_error" in response or "probe_eof" in response: raise RuntimeError(response) if response.get("id") == request["id"]: if "error" in response: raise RuntimeError(response) return response["result"] def tool(request_id, name, arguments): return send({"jsonrpc": "2.0", "id": request_id, "method": "tools/call", "params": {"name": name, "arguments": arguments}}) def text_payload(result): if result.get("isError"): raise RuntimeError(result) blocks = [x["text"] for x in result.get("content", []) if x.get("type") == "text"] return json.loads("\n".join(blocks)) try: initialized = send({"jsonrpc": "2.0", "id": 1, "method": "initialize", "params": {"protocolVersion": "2024-11-05", "capabilities": {}, "clientInfo": {"name": "retained-obligation-scripted-transport-probe", "version": "1"}}}) send({"jsonrpc": "2.0", "method": "notifications/initialized", "params": {}}) listed = send({"jsonrpc": "2.0", "id": 2, "method": "tools/list", "params": {}}) tool_names = [item["name"] for item in listed["tools"]] if "reqproof_query_requirements" not in tool_names: raise RuntimeError("query tool was not advertised") # Neither a requirement ID nor component is provided to this behavior query. found = text_payload(tool(3, "reqproof_query_requirements", {"search": "retry"})) metadata["query_payload"] = found # Accept the response's documented wrapper, keeping raw protocol bytes in the logs. requirements = found if isinstance(found, list) else found.get("requirements", []) if not requirements: raise RuntimeError("Behavior query returned no requirement") discovered = requirements[0]["id"] metadata["discovered_id"] = discovered metadata["approved_filter_payload"] = text_payload(tool(4, "reqproof_query_requirements", {"search": "retry", "status": "approved"})) metadata["task_phrase_payload"] = text_payload(tool(5, "reqproof_query_requirements", {"search": "Reduce storage used by the reservation cache"})) metadata["trace_payload"] = text_payload(tool(6, "reqproof_trace_analysis", {"id": discovered, "direction": "both", "depth": 2})) resources = send({"jsonrpc": "2.0", "id": 7, "method": "resources/list", "params": {}}) template = next(item["uri"] for item in resources["resources"] if item["name"] == "Requirement Detail") detail_uri = template.replace("{id}", discovered) detail = send({"jsonrpc": "2.0", "id": 8, "method": "resources/read", "params": {"uri": detail_uri}}) metadata["requirement_detail_uri"] = detail_uri metadata["requirement_detail"] = json.loads(detail["contents"][0]["text"]) metadata["initialize_result"] = initialized metadata["advertised_tool_count"] = len(tool_names) metadata["outcome"] = "completed" except Exception as error: metadata["outcome"] = "error" metadata["error"] = str(error) finally: process.stdin.close() try: process.wait(timeout=5) except subprocess.TimeoutExpired: process.terminate() process.wait(timeout=5) metadata["termination"] = "terminated after EOF did not stop process within five seconds" reader.join(timeout=2) metadata["server_exit_code"] = process.returncode metadata["tracked_status_after"] = git("status", "--porcelain", "--untracked-files=no") finalize_capture(metadata, request_log, response_log, output) print(json.dumps(metadata, indent=2)) raise SystemExit(0 if metadata["outcome"] == "completed" and metadata["server_exit_code"] == 0 else 1) if __name__ == "__main__": main()