tegwick
cd9602c9d9
The previous example used the prose "Session owner", so providers recorded a space instead of the fixture token. Assistant: grok Assistant-Session: 01a10286-9185-7bb2-b411-89cf6ab4c0c6
166 lines
9.4 KiB
Python
166 lines
9.4 KiB
Python
#!/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. When the card names that recipient as a role, record the role and "
|
|
"keep the supporting repository in sources. Record the role with one of these "
|
|
"tokens: session-owner, delegator, platform, tenant-admin, root-steward, pep-owner. "
|
|
"When the card names a "
|
|
"repository as the recipient, record that repository. Route is 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())
|