feat(bench): add a pinned wiki snapshot for the wiki knowledge level

benchmarks/agent/agent_bench.py

@@ -21,6 +21,7 @@
 import bench_grade
 import bench_report
 import bench_trace
+import bench_wiki
 
 HERE = Path(__file__).resolve().parent
 ROOT = HERE.parent.parent
@@ -97,8 +98,8 @@ def check_cell(cell: dict, args: argparse.Namespace) -> None:
         raise SystemExit("knowledge 'web' requires network 'on'")
     if cell["wright"] == "bin+skill" and not args.skill_dir:
         raise SystemExit("wright level 'bin+skill' requires --skill-dir")
-    if cell["knowledge"] == "wiki" and not args.wiki_dir:
-        raise SystemExit("knowledge 'wiki' requires --wiki-dir")
+    if cell["knowledge"] == "wiki" and not (args.wiki_dir and (Path(args.wiki_dir) / "SNAPSHOT.json").is_file()):
+        raise SystemExit("knowledge 'wiki' requires --wiki-dir pointing at a snapshot (see `agent_bench.py wiki-snapshot`)")
 
 
 def build_env(cell: dict, args: argparse.Namespace, out: Path, workspace: Path) -> dict:
@@ -250,6 +251,7 @@ def base_result(scenario: dict, cell: dict, args: argparse.Namespace, out: Path,
             "os": platform.platform(), "python": platform.python_version(),
             "wright": subprocess.run([args.wright, "--version"], capture_output=True, text=True).stdout.strip(),
             "timestamp": datetime.now(timezone.utc).isoformat(timespec="seconds"),
+            **({"wiki": bench_wiki.identity(Path(args.wiki_dir))} if cell["knowledge"] == "wiki" else {}),
         },
     }
 
@@ -297,6 +299,12 @@ def work(job: tuple) -> None:
     return 0
 
 
+def cmd_wiki_snapshot(args: argparse.Namespace) -> int:
+    record = bench_wiki.snapshot(args.base, args.dir)
+    print(f"{len(record['documents'])} document(s) from {record['source']} into {args.dir}\nsnapshotSha256 {record['snapshotSha256']}")
+    return 0
+
+
 def cmd_setup_oracle(_: argparse.Namespace) -> int:
     return subprocess.call(["npm", "ci", "--silent"], cwd=bench_grade.ORACLE)
 
@@ -329,6 +337,9 @@ def main() -> int:
     run.add_argument("--trials", type=int, default=1)
     sub.choices["matrix"].add_argument("config", type=Path, help="JSON: agents[{id,cmd}], cells[{wright,knowledge,network}], scenarios, trials, parallel, seed, options")
     sub.add_parser("setup-oracle", help="install the pinned upstream OverPy oracle")
+    wiki = sub.add_parser("wiki-snapshot", help="fetch the Workshop wiki Markdown mirror into a pinned local snapshot")
+    wiki.add_argument("--dir", type=Path, default=Path.home() / ".cache/wright-agent-bench-wiki")
+    wiki.add_argument("--base", default=bench_wiki.BASE)
     report = sub.add_parser("report", help="summarize result.json files")
     report.add_argument("dirs", nargs="+", type=Path)
     report.add_argument("--regrade", action="store_true", help="re-grade stored workspaces twice and flag unstable graders")
@@ -342,6 +353,8 @@ def main() -> int:
         return 0 if validate(args.wright, args.out) else 1
     if args.command == "setup-oracle":
         return cmd_setup_oracle(args)
+    if args.command == "wiki-snapshot":
+        return cmd_wiki_snapshot(args)
     if args.command == "report":
         return bench_report.main(args.dirs, args.wright, args.regrade, lambda s: load_scenario(s))
     return cmd_run(args) if args.command == "run" else cmd_matrix(args)

benchmarks/agent/bench_wiki.py

@@ -0,0 +1,65 @@
+"""Snapshot of the Workshop wiki Markdown mirror for the benchmark's `wiki` knowledge level (#414, SPEC-414)."""
+
+from __future__ import annotations
+
+import hashlib
+import json
+import re
+import subprocess
+import time
+from datetime import datetime, timezone
+from pathlib import Path
+
+BASE = "https://md.wrightkit.dev"
+USER_AGENT = "wright-agent-bench/1 (+https://github.com/wrightkit/wright)"
+SLUG = re.compile(r"^[A-Za-z0-9_-]+$")
+NOTICE = (
+    "Workshop.codes wiki content, rendered to Markdown by the mirror recorded in SNAPSHOT.json.\n"
+    "Use and redistribution follow the Workshop.codes Terms of Service (https://workshop.codes/tos).\n"
+    "This snapshot is for local benchmark runs; do not commit or redistribute it.\n"
+)
+
+
+def fetch(url: str) -> bytes:
+    """GET through curl: the mirror rejects Python's HTTP client fingerprint with 403."""
+    proc = subprocess.run(["curl", "-fsSL", "-m", "30", "-A", USER_AGENT, url], capture_output=True)
+    if proc.returncode != 0:
+        raise SystemExit(f"fetch failed for {url}: {proc.stderr.decode(errors='replace').strip()}")
+    return proc.stdout
+
+
+def snapshot(base: str, out: Path, delay: float = 0.2) -> dict:
+    """Fetch the manifest and every article once, and write the files plus SNAPSHOT.json with content hashes."""
+    if (out / "SNAPSHOT.json").exists():
+        raise SystemExit(f"{out} already holds a snapshot; snapshots are pinned, so choose a new directory")
+    manifest_bytes = fetch(f"{base}/manifest.json")
+    manifest = json.loads(manifest_bytes)
+    if manifest.get("schemaVersion") != 1:
+        raise SystemExit(f"unsupported manifest schemaVersion {manifest.get('schemaVersion')!r}")
+    articles = out / "articles"
+    articles.mkdir(parents=True, exist_ok=True)
+    documents = []
+    for doc in manifest["documents"]:
+        slug = doc["slug"]
+        if not SLUG.match(slug):
+            raise SystemExit(f"refusing unsafe slug {slug!r}")
+        body = fetch(f"{base}/wiki/articles/{slug}")
+        (articles / f"{slug}.md").write_bytes(body)
+        documents.append({"slug": slug, "title": doc.get("title"), "updatedAt": doc.get("updatedAt"), "sourceUrl": doc.get("sourceUrl"), "sha256": hashlib.sha256(body).hexdigest()})
+        time.sleep(delay)
+    (out / "index.md").write_bytes(fetch(f"{base}/wiki/articles"))
+    (out / "NOTICE.txt").write_text(NOTICE)
+    identity = "\n".join(f"{d['slug']} {d['sha256']}" for d in sorted(documents, key=lambda d: d["slug"]))
+    record = {
+        "source": base, "fetchedAt": datetime.now(timezone.utc).isoformat(timespec="seconds"),
+        "manifestSha256": hashlib.sha256(manifest_bytes).hexdigest(), "documents": documents,
+        "snapshotSha256": hashlib.sha256(identity.encode()).hexdigest(),
+    }
+    (out / "SNAPSHOT.json").write_text(json.dumps(record, indent=2) + "\n")
+    return record
+
+
+def identity(wiki_dir: Path) -> dict:
+    """The pinned identity of a snapshot, recorded in every result that used it."""
+    record = json.loads((wiki_dir / "SNAPSHOT.json").read_text())
+    return {k: record[k] for k in ("source", "fetchedAt", "snapshotSha256")} | {"documents": len(record["documents"])}

benchmarks/agent/test_agent_bench.py

@@ -26,14 +26,25 @@ def setUp(self):
         self.out = Path(tempfile.mkdtemp(dir=agent_bench.ROOT / "target")).resolve()
         self.addCleanup(shutil.rmtree, self.out, True)
 
-    def trial(self, agent_cmd: str, wright: str = "bin", scenario: str = SCENARIO, **options) -> dict:
+    def trial(self, agent_cmd: str, wright: str = "bin", scenario: str = SCENARIO, knowledge: str = "none", **options) -> dict:
         args = argparse.Namespace(**{
             "wright": str(Path(WRIGHT).resolve()), "agent_id": "fake", "agent_cmd": agent_cmd, "timeout": 60, "infra_retries": 2,
             "env_pass": [], "canary_cmd": None, "skill_dir": None, "wiki_dir": None, "check_ancestors": False, **options,
         })
-        cell = {"wright": wright, "knowledge": "none", "network": "off"}
+        cell = {"wright": wright, "knowledge": knowledge, "network": "off"}
         return agent_bench.run_trial(agent_bench.load_scenario(scenario), cell, args, self.out / f"{scenario}-{wright}")
 
+    def test_wiki_snapshot_is_linked_read_only_and_identified(self):
+        snapshot = self.out / "snapshot"
+        (snapshot / "articles").mkdir(parents=True)
+        (snapshot / "articles/wait-until.md").write_text("# Wait Until\n")
+        (snapshot / "SNAPSHOT.json").write_text(json.dumps({"source": "https://mirror.example", "fetchedAt": "2026-09-30T00:00:00+00:00", "snapshotSha256": "ab" * 32, "documents": [{"slug": "wait-until"}]}))
+        result = self.trial("ls wiki/articles > listing.txt", knowledge="wiki", wiki_dir=snapshot)
+        self.assertEqual(result["environment"]["wiki"]["snapshotSha256"], "ab" * 32)
+        self.assertEqual(result["environment"]["wiki"]["documents"], 1)
+        self.assertEqual((self.out / f"{SCENARIO}-bin/workspace/listing.txt").read_text().strip(), "wait-until.md")
+        self.assertNotIn("wiki", " ".join(result["unsafeEdits"]))
+
     def test_scenarios_are_solvable_and_not_vacuous(self):
         self.assertTrue(agent_bench.validate(WRIGHT, self.out / "validate"))
 

benchmarks/agent/test_wiki.py

@@ -0,0 +1,72 @@
+import argparse
+import json
+import shutil
+import tempfile
+import threading
+import unittest
+from http.server import BaseHTTPRequestHandler, HTTPServer
+from pathlib import Path
+
+import agent_bench
+import bench_wiki
+
+ARTICLES = {"wait-until": b"# Wait Until\nbody\n", "count-of": b"# Count Of\nbody\n"}
+
+
+class Handler(BaseHTTPRequestHandler):
+    manifest = {"schemaVersion": 1, "documents": [{"slug": s, "title": s, "updatedAt": "2026-01-01T00:00:00Z", "sourceUrl": f"https://workshop.codes/wiki/articles/{s}"} for s in ARTICLES]}
+
+    def do_GET(self):
+        if self.path == "/manifest.json":
+            body = json.dumps(self.manifest).encode()
+        elif self.path == "/wiki/articles":
+            body = b"# index\n"
+        elif self.path.startswith("/wiki/articles/") and self.path.rsplit("/", 1)[1] in ARTICLES:
+            body = ARTICLES[self.path.rsplit("/", 1)[1]]
+        else:
+            self.send_error(404)
+            return
+        self.send_response(200)
+        self.end_headers()
+        self.wfile.write(body)
+
+    def log_message(self, *args):
+        pass
+
+
+class WikiSnapshotTest(unittest.TestCase):
+    def setUp(self):
+        self.server = HTTPServer(("127.0.0.1", 0), Handler)
+        threading.Thread(target=self.server.serve_forever, daemon=True).start()
+        self.addCleanup(self.server.shutdown)
+        self.base = f"http://127.0.0.1:{self.server.server_port}"
+        self.tmp = Path(tempfile.mkdtemp())
+        self.addCleanup(shutil.rmtree, self.tmp, True)
+
+    def test_snapshot_writes_documents_and_a_stable_identity(self):
+        first = bench_wiki.snapshot(self.base, self.tmp / "a", delay=0)
+        second = bench_wiki.snapshot(self.base, self.tmp / "b", delay=0)
+        self.assertEqual((self.tmp / "a/articles/wait-until.md").read_bytes(), ARTICLES["wait-until"])
+        self.assertTrue((self.tmp / "a/index.md").is_file() and (self.tmp / "a/NOTICE.txt").is_file())
+        self.assertEqual(first["snapshotSha256"], second["snapshotSha256"])
+        self.assertEqual(bench_wiki.identity(self.tmp / "a")["documents"], 2)
+
+    def test_snapshots_are_pinned_and_slugs_are_checked(self):
+        bench_wiki.snapshot(self.base, self.tmp / "a", delay=0)
+        with self.assertRaises(SystemExit):
+            bench_wiki.snapshot(self.base, self.tmp / "a", delay=0)
+        Handler.manifest = {"schemaVersion": 1, "documents": [{"slug": "../evil"}]}
+        self.addCleanup(lambda: Handler.manifest.update(documents=[{"slug": s} for s in ARTICLES]))
+        with self.assertRaises(SystemExit):
+            bench_wiki.snapshot(self.base, self.tmp / "c", delay=0)
+
+    def test_wiki_level_needs_a_snapshot(self):
+        cell = {"wright": "none", "knowledge": "wiki", "network": "off"}
+        with self.assertRaises(SystemExit):
+            agent_bench.check_cell(cell, argparse.Namespace(skill_dir=None, wiki_dir=self.tmp))
+        bench_wiki.snapshot(self.base, self.tmp / "snap", delay=0)
+        agent_bench.check_cell(cell, argparse.Namespace(skill_dir=None, wiki_dir=self.tmp / "snap"))
+
+
+if __name__ == "__main__":
+    unittest.main()

docs/agent-benchmark.md

@@ -31,9 +31,21 @@ in every cell.
 | Factor | Levels |
 | --- | --- |
 | `wright` | `none`: no directory providing `wright` is on `PATH`. `bin`: `wright` on `PATH` through a tracing shim. `bin+skill`: `bin`, plus the guide directory given by `--skill-dir`, which the adapter installs. |
-| `knowledge` | `none`; `wiki`: `--wiki-dir` is linked read-only as `./wiki` (never counted as an edit); `web`: the adapter enables its web tools. |
+| `knowledge` | `none`; `wiki`: a pinned snapshot of the Workshop wiki Markdown mirror, given by `--wiki-dir` and linked as `./wiki` (never counted as an edit); `web`: the adapter enables its web tools. |
 | `network` | `off` or `on`; `web` requires `on`. |
 
+`agent_bench.py wiki-snapshot [--dir DIR]` builds the `wiki` snapshot from the
+mirror at `md.wrightkit.dev`: it reads `manifest.json`, fetches every article once
+(through `curl`, because the mirror rejects Python's HTTP client with 403), and
+writes `articles/`, `index.md`, `NOTICE.txt`, and `SNAPSHOT.json` with per-document
+hashes and a `snapshotSha256`. A snapshot is never overwritten, a run with
+`knowledge` `wiki` refuses a directory without `SNAPSHOT.json`, and the result
+records the snapshot identity in `environment.wiki`. The mirror serves the
+Workshop.codes wiki, whose content follows the Workshop.codes Terms of Service, so
+keep the snapshot local and do not commit it. It is a small set of articles, not a
+complete Workshop reference: the `wiki` and `web` levels measure documentation that
+exists today.
+
 Each run is scrubbed: a fresh `HOME`, an allowlisted environment (`--env-pass`
 names host variables to keep), and no host instruction files. Two canaries run
 before the agent; a failed canary marks the run `invalid` and it is excluded