1import json, os, sys, tempfile, shutil, subprocess2REPO = os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))3sys.path.insert(0, os.path.join(REPO, "src"))4from wp42 import events, metrics, store as store_mod, arms, launcher, bundle5FIX = os.path.join(REPO, "tests", "fixtures")67def test_normalise_events_and_metrics():8 lines = [json.dumps({"type": "system", "session_id": "s", "model": "m", "tools": ["Bash"], "claude_code_version": "2.1.314"}),9 json.dumps({"type": "assistant", "session_id": "s", "message": {"content": [{"type": "tool_use", "name": "Bash", "input": {"command": "python x.py"}}], "usage": {"input_tokens": 10, "output_tokens": 5}}}),10 json.dumps({"type": "user", "session_id": "s", "message": {"content": [{"type": "tool_result", "is_error": True, "content": "boom"}]}}),11 json.dumps({"type": "assistant", "session_id": "s", "message": {"content": [{"type": "tool_use", "name": "Bash", "input": {"command": "python x.py"}}], "usage": {"input_tokens": 10, "output_tokens": 5}}}),12 json.dumps({"type": "user", "session_id": "s", "message": {"content": [{"type": "tool_result", "content": "ok"}]}}),13 json.dumps({"type": "assistant", "session_id": "s", "message": {"content": [{"type": "tool_use", "name": "Bash", "input": {"command": "osascript -e 1"}}], "usage": {}}}),14 json.dumps({"type": "user", "session_id": "s", "message": {"content": [{"type": "tool_result", "content": "ok"}]}}),15 json.dumps({"type": "result", "subtype": "success", "total_cost_usd": 0.5, "num_turns": 3, "duration_ms": 1000})]16 evs = [events.normalise(l, "r1", i + 1, float(i)) for i, l in enumerate(lines)]17 assert evs[0]["event_type"] == "system" and evs[1]["tool_name"] == "Bash" and evs[2]["error_class"] == "tool_error"18 for e in evs: e["usage"] = json.dumps(e.get("usage")) if e.get("usage") is not None else None19 m = metrics.compute(evs, {"run_id": "r1", "task_id": "T", "arm": "A1", "rep": 1, "cost_usd": 0.5}, {"det_completion": True, "det_score": 1.0, "critical_violations": []}, [])20 assert m["tool_calls"] == 3 and m["tool_errors"] == 1 and m["retries"] == 121 assert m["by_family"] == {"python": 2, "applescript": 1} and m["app_transitions"] == 122 assert m["tokens"]["input"] == 20 and m["claude_code_version"] == "2.1.314"2324def test_store_roundtrip():25 td = tempfile.mkdtemp(); s = store_mod.Store(os.path.join(td, "x.sqlite"))26 s.add_run(run_id="r1", task_id="T", occupation="o", arm="A0", rep=1, model="m", stage="pilot", order_index=0, seed=1)27 s.update_run("r1", status="running", cost_usd=0.3); s.ledger_add("r1", "worker", 0.3)28 assert s.get_run("r1")["status"] == "running" and abs(s.ledger_total() - 0.3) < 1e-929 s.set_receipt("r1", {"det_completion": True, "det_score": 1.0, "critical_violations": [], "abstention": {"status": "completed", "justified": None}})30 assert s.one("SELECT det_completion FROM receipts")["det_completion"] == 131 shutil.rmtree(td)3233def test_arms_and_argv():34 for a, c in arms.ARMS.items():35 assert "Agent" not in c["tools"] and "Task" not in c["tools"], a36 assert "Bash" not in arms.ARMS["A0"]["tools"]37 argv = launcher.build_argv("/w/PROMPT.md", "/w/SYSTEM.md", "claude-opus-5-5", "sid", 4.0, 150, arms.ARMS["A1"]["tools"], arms.ARMS["A1"]["disallowed"])38 assert "--bare" in argv and "--disallowedTools" in argv and "Bash(osascript:*)" in argv and "--session-id" in argv39 line = launcher.remote_shell_line("/w", argv)40 assert '"$(cat /w/PROMPT.md)"' in line and '< /dev/null' in line and "WP42_DEPTH=1" in line and "/Users/simon-pierreboucher/wp42/.env" in line4142def test_compose_workspace_excludes_oracle_and_generator(monkeypatch):43 monkeypatch.setattr(bundle, "scenario_dir", lambda t: os.path.join(FIX, t))44 ws = bundle.compose_local("DUMMY-01", "A4", "run_x", "http://127.0.0.1:1", {"username": "u", "password": "p"})45 names = set(os.listdir(ws))46 assert {"BRIEF.md", "inputs", "PROMPT.md", "SYSTEM.md", "PORTAL.md", "tools", "COMPLETION_SCHEMA.json", "BUNDLE_MANIFEST.json", "out", "task.json", "portal"} <= names47 assert "generate.py" not in names and "oracle.json" not in names48 ws0 = bundle.compose_local("DUMMY-01", "A0", "run_y", None, None)49 assert "portal" not in os.listdir(ws0) and "tools" not in os.listdir(ws0) and "PORTAL.md" not in os.listdir(ws0)50 shutil.rmtree(ws); shutil.rmtree(ws0)5152def test_selfcheck_detects_missing():53 td = tempfile.mkdtemp(); shutil.copytree(os.path.join(FIX, "DUMMY-01"), td, dirs_exist_ok=True); os.makedirs(os.path.join(td, "out"))54 r = subprocess.run([sys.executable, os.path.join(REPO, "benchmark", "tools", "selfcheck.py"), "--workspace", td, "--json"], capture_output=True, text=True)55 d = json.loads(r.stdout); assert not d["ok"] and any("answer.xlsx" in p for p in d["problems"]) and r.returncode == 156 shutil.rmtree(td)5758def test_launcher_refuses_nested(monkeypatch):59 monkeypatch.setenv("WP42_DEPTH", "1")60 import pytest61 with pytest.raises(launcher.LaunchError):62 launcher.run_session("nohost", "/w", ["claude"], "/tmp/x.jsonl", "r", "s", None, 1)6364def test_worker_task_json_has_no_design_fields(monkeypatch):65 monkeypatch.setattr(bundle, "scenario_dir", lambda t: os.path.join(FIX, t))66 ws = bundle.compose_local("DUMMY-01", "A1", "run_z", None, None)67 t = json.load(open(os.path.join(ws, "task.json")))68 assert "designed_features" not in t and "generator_seed" not in t and "designed_dependency_depth" not in t and "required_deliverables" in t69 shutil.rmtree(ws)70