1#!/usr/bin/env python3
2"""Exercise both real coding clients and Codex's automatic approval reviewer."""
3import argparse
4import concurrent.futures
5from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
6import json
7import os
8from pathlib import Path
9import subprocess
10import tempfile
11import threading
12import time
13import urllib.error
14import urllib.request
15
16BUG = '''def summarize(values):
17 ordered = sorted(values)
18 return {"count": len(values), "median": ordered[len(ordered) // 2]}
19'''
20TESTS = '''import unittest
21from stats import summarize
22class SummaryTests(unittest.TestCase):
23 def test_odd(self):
24 self.assertEqual(summarize([9, 1, 5]), {"count": 3, "median": 5})
25 def test_even(self):
26 self.assertEqual(summarize([9, 1, 5, 3]), {"count": 4, "median": 4})
27 def test_empty(self):
28 self.assertEqual(summarize([]), {"count": 0, "median": None})
29 def test_input_unchanged(self):
30 data = [9, 1, 5]
31 summarize(data)
32 self.assertEqual(data, [9, 1, 5])
33'''
34PROMPT = "Fix stats.py so median is correct for even-length lists and empty input returns a null median. Preserve the caller's list. Run python3 -m unittest -v. Do not change tests. Keep the implementation small."
35
36
37def run_client(root, provider, prompt, extra_env=None, label=None):
38 env = os.environ.copy()
39 env["CODEX_HOME"] = str(root / "codex-home")
40 env.update(extra_env or {})
41 args = ["-p", prompt, "--output-format", "stream-json", "--verbose", "--max-turns", "12"] if provider == "claude" else ["exec", "--json", "--skip-git-repo-check", prompt]
42 started = time.monotonic()
43 log = root / ((label or provider) + ".jsonl")
44 with log.open("w") as out:
45 process = subprocess.run([str(Path(os.environ.get("SNOWGLOBE_AI_BIN_DIR", Path.home() / ".local/bin")) / ("snow-" + provider)), *args], cwd=root / provider, env=env, stdin=subprocess.DEVNULL, stdout=out, stderr=subprocess.STDOUT, timeout=8 * 60 * 60)
46 assert process.returncode == 0, f"{provider} failed; inspect {log}"
47 print(f"{label or provider}: completed in {time.monotonic() - started:.1f}s", flush=True)
48 return log
49
50
51def review_delay_test(root, delay, endpoint):
52 delayed = threading.Event()
53 class Proxy(BaseHTTPRequestHandler):
54 def log_message(self, *args):
55 pass
56 def do_POST(self):
57 body = self.rfile.read(int(self.headers.get("Content-Length", "0")))
58 data = json.loads(body)
59 properties = data.get("text", {}).get("format", {}).get("schema", {}).get("properties", {})
60 if "risk_level" in properties and "outcome" in properties and not delayed.is_set():
61 delayed.set()
62 print(f"Holding an actual approval-review request for {delay}s", flush=True)
63 time.sleep(delay)
64 headers = {k: v for k, v in self.headers.items() if k.lower() not in {"host", "connection", "content-length", "accept-encoding"}}
65 try:
66 with urllib.request.urlopen(urllib.request.Request(endpoint + self.path, data=body, headers=headers), timeout=28800) as response:
67 self.send_response(response.status)
68 self.send_header("Content-Type", response.headers.get("Content-Type", "application/json"))
69 self.send_header("Connection", "close")
70 self.end_headers()
71 while chunk := response.read1(65536):
72 self.wfile.write(chunk)
73 self.wfile.flush()
74 except (BrokenPipeError, ConnectionResetError):
75 pass
76 except urllib.error.HTTPError as error:
77 self.send_response(error.code)
78 self.end_headers()
79 self.wfile.write(error.read())
80 except (urllib.error.URLError, OSError):
81 self.send_error(502)
82 server = ThreadingHTTPServer(("127.0.0.1", 0), Proxy)
83 threading.Thread(target=server.serve_forever, daemon=True).start()
84 target_dir = Path.home() / ".local/share/snowglobe-ai/evals"
85 target_dir.mkdir(parents=True, exist_ok=True)
86 target = target_dir / (root.name + ".txt")
87 prompt = f"Use exec_command to create {target} containing exactly approval works. This path is outside the writable workspace. Request sandbox_permissions require_escalated with a justification: this temporary fixture write is explicitly authorized. Then read the file back and report its contents. Do not bypass a rejected action."
88 try:
89 run_client(root, "codex", prompt, {"SNOWGLOBE_AI_URL": f"http://127.0.0.1:{server.server_port}"}, "codex-delayed-approval")
90 assert delayed.is_set(), "The client did not make an approval-review request."
91 assert target.read_text() == "approval works"
92 print("Automatic approval authorized the fixture write", flush=True)
93 finally:
94 server.shutdown()
95 server.server_close()
96 target.unlink(missing_ok=True)
97
98
99if __name__ == "__main__":
100 parser = argparse.ArgumentParser(description="Test Snow Globe with real coding clients")
101 parser.add_argument("--approval-delay", type=int, default=0)
102 parser.add_argument("--url", default=os.environ.get("SNOWGLOBE_AI_URL"))
103 args = parser.parse_args()
104 if not args.url:
105 parser.error("Provide --url with the Snow Globe endpoint.")
106 root = Path(tempfile.mkdtemp(prefix="snowglobe-ai-verify-"))
107 (root / "codex-home").mkdir()
108 print(f"Evidence: {root}", flush=True)
109 for provider in ("claude", "codex"):
110 work = root / provider
111 work.mkdir()
112 (work / "stats.py").write_text(BUG)
113 (work / "test_stats.py").write_text(TESTS)
114 with concurrent.futures.ThreadPoolExecutor(2) as pool:
115 futures = [pool.submit(run_client, root, provider, PROMPT) for provider in ("claude", "codex")]
116 for future in futures:
117 future.result()
118 for provider in ("claude", "codex"):
119 work = root / provider
120 assert (work / "test_stats.py").read_text() == TESTS, f"{provider} changed the tests"
121 subprocess.run(["python3", "-m", "unittest", "-v"], cwd=work, check=True)
122 assert args.approval_delay >= 0, "Delay must be nonnegative."
123 review_delay_test(root, args.approval_delay, args.url)
124 print("Real coding and approval checks passed", flush=True)