#!/usr/bin/env python3 """Run one bounded, read-only provider knowledge probe in a disposable fixture. Uses the installed provider's configured model/auth, never reads credentials. Raw client diagnostics stay mode 0600 under /tmp, not in repository evidence. This is a comprehension probe, not a production or harness-isolation proof. """ from pathlib import Path import argparse from datetime import datetime, timezone import json import os import shutil import subprocess import sys import tempfile import uuid import yaml ROOT = Path(__file__).resolve().parents[1] sys.path.insert(0, str(ROOT / "src")) from emergent_security.knowledge import Knowledge def extract_json(text): text = text.strip() if text.startswith("```"): text = text.split("\n", 1)[1].rsplit("```", 1)[0].strip() return json.loads(text) def unwrap_response(response): """Provider envelopes differ; do not grade an envelope as the actual answer.""" if not isinstance(response, dict): raise ValueError("expected response object") metadata = {} if "structured_output" in response or "structuredOutput" in response: metadata = {"session_id": response.get("session_id", response.get("sessionId")), "model_usage": list(response.get("modelUsage", {})), "usage": response.get("usage")} response = response.get("structured_output", response.get("structuredOutput")) elif "result" in response and isinstance(response["result"], str): metadata["session_id"] = response.get("session_id") response = extract_json(response["result"]) if not isinstance(response, dict) or not isinstance(response.get("answers"), list): raise ValueError("missing structured answers") return response, metadata def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("provider", choices=["codex", "claude", "grok"]) parser.add_argument("--output", type=Path, required=True) parser.add_argument("--timeout", type=int, default=180) parser.add_argument("--case", action="append", help="Run only this scenario ID (repeatable)") args = parser.parse_args() package = Knowledge(ROOT / "knowledge") corpus = yaml.safe_load((ROOT / "tests/fixtures/knowledge/scenarios.yaml").read_text())["cases"] if args.case: if not set(args.case) <= {case["id"] for case in corpus}: raise SystemExit("Unknown scenario ID") corpus = [case for case in corpus if case["id"] in args.case] work = Path(tempfile.mkdtemp(prefix=f"emergent-{args.provider}-probe-")) os.chmod(work, 0o700) shutil.copytree(ROOT / "knowledge", work / "knowledge") startup_marker, retrieval_marker = uuid.uuid4().hex, uuid.uuid4().hex with (work / "knowledge/cards/start.md").open("a") as f: f.write(f"\nFixture-only startup marker: {startup_marker}\n") with (work / "knowledge/cards/access.md").open("a") as f: f.write(f"\nFixture-only retrieval marker: {retrieval_marker}\n") # Markers instrument the test copy, never the sealed release. instruction = ( "# Disposable knowledge evaluation\nRead knowledge/cards/start.md before answering. " "Read task-specific cards under knowledge/cards/ as needed, including access.md. " "Use only these fixture files. No shell effects, network tools, edits, credentials, " "other repositories, or subagents. Read-only evaluation, not actual operations. " f"The fixture is instrumented from package {package.catalog['version']}; marker " "additions invalidate its copy's seal intentionally. Do not modify the original.\n" ) (work / "AGENTS.md").write_text(instruction) (work / "CLAUDE.md").write_text(instruction) (work / "questions.json").write_text(json.dumps([{k: c[k] for k in ("id", "question")} for c in corpus], indent=2)) schema = {"type": "object", "additionalProperties": False, "properties": {"package_version": {"type": "string"}, "startup_marker": {"type": "string"}, "retrieval_marker": {"type": "string"}, "answers": {"type": "array", "items": { "type": "object", "additionalProperties": False, "properties": {k: {"type": "string"} for k in ("id", "route", "disposition", "sources", "reason")}, "required": ["id", "route", "disposition", "sources", "reason"]}}}, "required": ["package_version", "startup_marker", "retrieval_marker", "answers"]} (work / "response-schema.json").write_text(json.dumps(schema)) prompt = ( "Perform this read-only knowledge evaluation. Follow the workspace orientation " "instructions and answer every question in questions.json. Read the knowledge cards; " "do not use outside sources or perform any actual operation. Return ONLY JSON with " "package_version, startup_marker (from start.md), retrieval_marker (from access.md), " "and answers: [{id, route, disposition, sources, reason}]. Route means the actual " "next recipient (a human role such as delegator, platform, tenant-admin, root-steward " "or pep-owner when appropriate, otherwise the owning repository), NOT the source " "of documentation. Use a concise action for disposition, comma-separated exact source owner IDs " "listed in the relevant card for sources, and a one-sentence explanation. Cite the " "most specific source, including emergent-security for its generic envelope and " "escalation semantics. Do not invent owner IDs or marker values." ) if args.provider == "codex": command = ["codex", "exec", "--ephemeral", "--sandbox", "read-only", "--skip-git-repo-check", "--cd", str(work), "--json", "--output-schema", str(work / "response-schema.json"), "--output-last-message", str(work / "answer.json"), prompt] elif args.provider == "claude": command = ["claude", "-p", "--tools", "Read", "--permission-mode", "default", "--no-session-persistence", "--max-budget-usd", "3", "--output-format", "json", "--json-schema", json.dumps(schema), prompt] else: command = ["grok", "--cwd", str(work), "--permission-mode", "plan", "--no-subagents", "--disable-web-search", "--max-turns", "15", "--output-format", "json", "--json-schema", json.dumps(schema), "-p", prompt] report = {"schema": "emergent.provider-probe.v1", "provider": args.provider, "timestamp": datetime.now(timezone.utc).isoformat(), "package_version": package.catalog["version"], "package_digest": package.package_digest, "mode": "native-file-pointer-probe", "model_selection": "installed-config-default", "scope": "disposable offline fixtures; no live operations", "scenario_ids": [case["id"] for case in corpus], "status": "unverified", "work_directory": str(work)} try: result = subprocess.run(command, cwd=work, capture_output=True, text=True, timeout=args.timeout) for name, text in [("stdout.txt", result.stdout), ("stderr.txt", result.stderr)]: path = work / name path.write_text(text) os.chmod(path, 0o600) report["exit_code"] = result.returncode combined = result.stdout + result.stderr if result.returncode: report["status"] = "client-failed" report["failure_category"] = ("filesystem-permission" if any(s in combined.lower() for s in ["permission denied", "read-only file system", "eperm", "eacces"]) else "client-or-auth-or-network") else: raw = ((work / "answer.json").read_text() if args.provider == "codex" else result.stdout) response, metadata = unwrap_response(extract_json(raw)) report.update(metadata) report["startup_read_proven"] = response.get("startup_marker") == startup_marker report["deeper_read_proven"] = response.get("retrieval_marker") == retrieval_marker answers = response.get("answers", []) expected_ids = {c["id"] for c in corpus} report["all_cases_answered"] = (len(answers) == len(corpus) and {a.get("id") for a in answers} == expected_ids) report["package_version_correct"] = response.get("package_version") == package.catalog["version"] report["answers"] = answers report["status"] = "awaiting-semantic-review" if all(report[k] for k in ("startup_read_proven", "deeper_read_proven", "all_cases_answered", "package_version_correct")) else "probe-failed" except subprocess.TimeoutExpired: report["status"] = "timeout" except (OSError, ValueError, KeyError, TypeError): report["status"] = "client-output-unusable" args.output.parent.mkdir(parents=True, exist_ok=True) args.output.write_text(json.dumps(report, indent=2) + "\n") print(json.dumps({k: v for k, v in report.items() if k != "answers"}, indent=2)) return 0 if report["status"] == "awaiting-semantic-review" else 2 if __name__ == "__main__": raise SystemExit(main())