diff --git a/.codex/config.toml b/.codex/config.toml index a847f82..82674e0 100644 --- a/.codex/config.toml +++ b/.codex/config.toml @@ -1,3 +1,11 @@ [agents] enabled = true max_concurrent_threads_per_session = 4 + +[agents.plan_anvil_profiler] +description = "Read-only repository profiler for PlanAnvil evidence gathering and instruction discovery." +config_file = "./agents/plan-anvil-profiler.toml" + +[agents.plan_anvil_reviewer] +description = "Fresh read-only blind reviewer for generated PlanAnvil contracts." +config_file = "./agents/plan-anvil-reviewer.toml" diff --git a/.github/workflows/plananvil-codex-qualification.yml b/.github/workflows/plananvil-codex-qualification.yml index 58d35b5..f89d3ed 100644 --- a/.github/workflows/plananvil-codex-qualification.yml +++ b/.github/workflows/plananvil-codex-qualification.yml @@ -215,7 +215,7 @@ jobs: set -euo pipefail cd "${PRECISION_SOURCE}" set +e - python3 tools/codex_runner_precision_matrix.py \ + python3 tools/codex_runner_precision_v2.py \ --root "${PRECISION_FIXTURES}" \ --output "${PRECISION_ARTIFACT}" rc=$? diff --git a/tests/test_codex_runner_precision_matrix.py b/tests/test_codex_runner_precision_matrix.py index 58059c7..39ce073 100644 --- a/tests/test_codex_runner_precision_matrix.py +++ b/tests/test_codex_runner_precision_matrix.py @@ -10,7 +10,7 @@ ROOT = Path(__file__).resolve().parents[1] sys.path.insert(0, str(ROOT / "tools")) -import codex_runner_precision_matrix as precision +import codex_runner_precision_matrix as matrix WORKFLOW = ROOT / ".github" / "workflows" / "plananvil-codex-qualification.yml" SOURCE = ROOT / "tools" / "codex_runner_precision_matrix.py" @@ -18,14 +18,14 @@ class CodexRunnerPrecisionMatrixTests(unittest.TestCase): def test_exact_precision_variants(self) -> None: - self.assertEqual(len(precision.VARIANT_NAMES), 11) - self.assertEqual(precision.VARIANT_NAMES[0], "pretool_json_bash_allow_absolute") - self.assertIn("compact_body_after_prefix_two_step_absolute", precision.VARIANT_NAMES) - self.assertIn("subagent_non_ephemeral_project_explicit", precision.VARIANT_NAMES) - self.assertIn("subagent_non_ephemeral_home_explicit", precision.VARIANT_NAMES) + self.assertEqual(len(matrix.VARIANT_NAMES), 11) + self.assertEqual(matrix.VARIANT_NAMES[0], "pretool_json_bash_allow_absolute") + self.assertIn("compact_body_after_prefix_two_step_absolute", matrix.VARIANT_NAMES) + self.assertIn("subagent_non_ephemeral_project_explicit", matrix.VARIANT_NAMES) + self.assertIn("subagent_non_ephemeral_home_explicit", matrix.VARIANT_NAMES) def test_pretool_probe_has_absolute_recorder_and_real_deny(self) -> None: - source = precision._pretool_script() + source = matrix._pretool_script() self.assertNotIn("git rev-parse", source) self.assertIn("Path(sys.argv[1])", source) self.assertIn("cwd_matches_repo", source) @@ -33,7 +33,7 @@ def test_pretool_probe_has_absolute_recorder_and_real_deny(self) -> None: self.assertIn("PLANANVIL_DIAG_PRETOOL_DENY", source) def test_compaction_probe_has_absolute_recorder(self) -> None: - source = precision._compact_script() + source = matrix._compact_script() self.assertNotIn("git rev-parse", source) self.assertIn("Path(sys.argv[2])", source) text = SOURCE.read_text(encoding="utf-8") @@ -44,10 +44,10 @@ def test_compaction_probe_has_absolute_recorder(self) -> None: def test_subagent_opaque_value_is_not_present_in_agent_config(self) -> None: token = "opaque-test-value" - self.assertNotIn(token, precision._agent_toml()) - self.assertIn(token, precision._subagent_script(token)) - self.assertNotIn("git rev-parse", precision._subagent_script(token)) - self.assertIn("PLANANVIL_DIAG_CONTEXT_TOKEN=", precision._agent_toml()) + self.assertNotIn(token, matrix._agent_toml()) + self.assertIn(token, matrix._subagent_script(token)) + self.assertNotIn("git rev-parse", matrix._subagent_script(token)) + self.assertIn("PLANANVIL_DIAG_CONTEXT_TOKEN=", matrix._agent_toml()) def test_command_observation_uses_aggregated_output_not_command_string(self) -> None: marker = "PLANANVIL_DIAG_HOOK_COMMAND_OK" @@ -63,21 +63,21 @@ def test_command_observation_uses_aggregated_output_not_command_string(self) -> }, } ) - observed = precision._command_observation(denied, marker) + observed = matrix._command_observation(denied, marker) self.assertFalse(observed["marker_output_observed"]) self.assertEqual(observed["failed_count"], 1) def test_secret_redaction_is_recursive(self) -> None: token = "secret-context" value = {"a": [f"prefix-{token}", {"b": token}]} - serialized = json.dumps(precision._redact(value, token), sort_keys=True) + serialized = json.dumps(matrix._redact(value, token), sort_keys=True) self.assertNotIn(token, serialized) self.assertIn("", serialized) def test_one_variant_failure_does_not_abort_artifact_generation(self) -> None: with tempfile.TemporaryDirectory() as tmp: output = Path(tmp) - result = precision._run_case( + result = matrix._run_case( "synthetic", lambda: (_ for _ in ()).throw(TypeError("boom")), output, @@ -101,13 +101,13 @@ def subagent(_root: Path, _output: Path, name: str, *_args: object) -> dict[str, with tempfile.TemporaryDirectory() as tmp: root, output = Path(tmp) / "runtime", Path(tmp) / "artifact" with ( - mock.patch.object(precision, "_hook_variant", side_effect=hook) as h, - mock.patch.object(precision, "_compact_variant", side_effect=compact) as c, - mock.patch.object(precision, "_subagent_variant", side_effect=subagent) as s, + mock.patch.object(matrix, "_hook_variant", side_effect=hook) as hook_mock, + mock.patch.object(matrix, "_compact_variant", side_effect=compact) as compact_mock, + mock.patch.object(matrix, "_subagent_variant", side_effect=subagent) as subagent_mock, ): - results = precision.run_matrix(root, output) - self.assertEqual([item["variant"] for item in results], list(precision.VARIANT_NAMES)) - self.assertEqual((h.call_count, c.call_count, s.call_count), (5, 3, 3)) + results = matrix.run_matrix(root, output) + self.assertEqual([item["variant"] for item in results], list(matrix.VARIANT_NAMES)) + self.assertEqual((hook_mock.call_count, compact_mock.call_count, subagent_mock.call_count), (5, 3, 3)) self.assertEqual(len(list(output.glob("*.json"))), 11) def test_precision_mode_uses_existing_runner_allowed_workflow(self) -> None: @@ -115,7 +115,7 @@ def test_precision_mode_uses_existing_runner_allowed_workflow(self) -> None: self.assertIn("- precision", workflow) self.assertIn("inputs.mode == 'precision'", workflow) precision_job = workflow[workflow.index(" precision:"):workflow.index(" full:")] - self.assertIn("codex_runner_precision_matrix.py", precision_job) + self.assertIn("codex_runner_precision_v2.py", precision_job) for label in ("self-hosted", "linux", "x64", "plananvil", "codex"): self.assertIn(f"- {label}", precision_job) self.assertIn("Variant observations are intentionally non-gating", precision_job) diff --git a/tests/test_codex_runner_precision_v2.py b/tests/test_codex_runner_precision_v2.py new file mode 100644 index 0000000..6ef7ca2 --- /dev/null +++ b/tests/test_codex_runner_precision_v2.py @@ -0,0 +1,170 @@ +from __future__ import annotations + +import json +from pathlib import Path +import sys +import tempfile +import unittest +from unittest import mock + +ROOT = Path(__file__).resolve().parents[1] +sys.path.insert(0, str(ROOT / "tools")) + +import codex_runner_precision_v2 as precision + +SOURCE = ROOT / "tools" / "codex_runner_precision_v2.py" +PRODUCT_CONFIG = ROOT / ".codex" / "config.toml" + + +class CodexRunnerPrecisionV2Tests(unittest.TestCase): + def test_exact_runtime_isolation_matrix(self) -> None: + self.assertEqual(len(precision.VARIANT_NAMES), 11) + self.assertEqual( + precision.VARIANT_NAMES, + ( + "project_cli_trust_ephemeral_deny", + "project_persisted_trust_non_ephemeral_json_deny", + "project_persisted_trust_non_ephemeral_toml_deny", + "home_non_ephemeral_json_deny", + "compact_home_body_after_prefix_single", + "compact_home_body_after_prefix_two_step", + "compact_home_total_two_step", + "subagent_project_autodiscovery_non_ephemeral", + "subagent_project_declared_non_ephemeral", + "subagent_home_declared_non_ephemeral", + "subagent_project_declared_ephemeral", + ), + ) + + def test_hook_probe_distinguishes_discovery_from_pretool_match(self) -> None: + source = precision._hook_script() + self.assertNotIn("git rev-parse", source) + self.assertIn('kind == "PreToolUse"', source) + self.assertIn("cwd_matches_repo", source) + self.assertIn("permissionDecisionReason", source) + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + text = precision._hooks_json( + root / "hook.py", root / "hook.jsonl", root / "repo", decision="deny" + ) + payload = json.loads(text) + self.assertIn("SessionStart", payload["hooks"]) + self.assertEqual(payload["hooks"]["PreToolUse"][0]["matcher"], "^Bash$") + + def test_runtime_modes_separate_cli_and_persisted_trust(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + repo = root / "repo" + home = root / "home" + repo.mkdir() + home.mkdir() + cli = precision._runtime_args( + repo, + "probe", + home=home, + ephemeral=True, + cli_trust=True, + ignore_user_config=True, + ) + persisted = precision._runtime_args( + repo, + "probe", + home=home, + ephemeral=False, + cli_trust=False, + ) + self.assertIn("--strict-config", cli) + self.assertIn("--ephemeral", cli) + self.assertIn("--ignore-user-config", cli) + self.assertTrue(any(value.startswith("projects.") for value in cli)) + self.assertNotIn("--ephemeral", persisted) + self.assertNotIn("--ignore-user-config", persisted) + self.assertIn('history.persistence="none"', persisted) + + def test_persisted_trust_uses_codex_projects_table(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + repo = Path(tmp) / "repo" + repo.mkdir() + text = precision._project_trust_toml(repo) + self.assertIn("[projects.", text) + self.assertIn('trust_level = "trusted"', text) + + def test_compaction_home_control_proves_hook_engine_loaded(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + payload = json.loads( + precision._compact_hooks_json( + root / "compact.py", root / "compact.jsonl", root / "repo" + ) + ) + self.assertIn("SessionStart", payload["hooks"]) + self.assertIn("PreCompact", payload["hooks"]) + self.assertIn("PostCompact", payload["hooks"]) + source = SOURCE.read_text(encoding="utf-8") + self.assertIn("features.token_budget=false", source) + self.assertIn("second separate shell tool call", source) + + def test_subagent_matrix_compares_autodiscovery_declaration_and_ephemeral(self) -> None: + auto = precision._project_agent_config(declared=False) + declared = precision._project_agent_config(declared=True) + self.assertNotIn("[agents.fixture_agent]", auto) + self.assertIn("[agents.fixture_agent]", declared) + self.assertIn('config_file = "./agents/fixture-agent.toml"', declared) + token = "opaque-test-token" + self.assertNotIn(token, precision._agent_toml()) + self.assertIn(token, precision._subagent_script(token)) + + def test_product_roles_are_explicitly_declared(self) -> None: + config = PRODUCT_CONFIG.read_text(encoding="utf-8") + self.assertIn("[agents.plan_anvil_profiler]", config) + self.assertIn('config_file = "./agents/plan-anvil-profiler.toml"', config) + self.assertIn("[agents.plan_anvil_reviewer]", config) + self.assertIn('config_file = "./agents/plan-anvil-reviewer.toml"', config) + + def test_one_failure_does_not_abort_matrix(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + output = Path(tmp) + result = precision._run_case( + "synthetic", + lambda: (_ for _ in ()).throw(RuntimeError("boom")), + output, + ) + self.assertEqual(result["diagnostic_status"], "HARNESS_ERROR") + self.assertTrue((output / "synthetic.json").is_file()) + + def test_dispatches_all_eleven_without_codex(self) -> None: + def fake(name: str) -> dict[str, object]: + return {"variant": name, "returncode": 0, "diagnostic_status": "TEST"} + + def hook(_root: Path, _output: Path, name: str, **_kwargs: object): + return fake(name) + + def compact(_root: Path, _output: Path, name: str, **_kwargs: object): + return fake(name) + + def subagent(_root: Path, _output: Path, name: str, **_kwargs: object): + return fake(name) + + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) / "runtime" + output = Path(tmp) / "artifact" + with ( + mock.patch.object(precision, "_hook_variant", side_effect=hook) as h, + mock.patch.object(precision, "_compact_variant", side_effect=compact) as c, + mock.patch.object(precision, "_subagent_variant", side_effect=subagent) as s, + ): + results = precision.run_matrix(root, output) + self.assertEqual([item["variant"] for item in results], list(precision.VARIANT_NAMES)) + self.assertEqual((h.call_count, c.call_count, s.call_count), (4, 3, 4)) + self.assertEqual(len(list(output.glob("*.json"))), 11) + + def test_evidence_does_not_persist_raw_transcripts(self) -> None: + source = SOURCE.read_text(encoding="utf-8") + self.assertNotIn("write_text(stdout", source) + self.assertNotIn("write_text(stderr", source) + self.assertIn("opaque_token_persisted_in_evidence", source) + self.assertIn("never a release gate", source) + + +if __name__ == "__main__": + unittest.main() diff --git a/tools/codex_runner_precision_v2.py b/tools/codex_runner_precision_v2.py new file mode 100644 index 0000000..b6568e6 --- /dev/null +++ b/tools/codex_runner_precision_v2.py @@ -0,0 +1,811 @@ +from __future__ import annotations + +import argparse +import json +import os +import re +import secrets +import shutil +from pathlib import Path +from typing import Any, Callable + +import codex_runner_precision_matrix as prior + +legacy = prior.legacy +base = prior.base +c13 = prior.c13 +MODEL = prior.MODEL +COMPACT_LIMIT = prior.COMPACT_LIMIT +PAYLOAD_WORDS = prior.PAYLOAD_WORDS + +VARIANT_NAMES = ( + "project_cli_trust_ephemeral_deny", + "project_persisted_trust_non_ephemeral_json_deny", + "project_persisted_trust_non_ephemeral_toml_deny", + "home_non_ephemeral_json_deny", + "compact_home_body_after_prefix_single", + "compact_home_body_after_prefix_two_step", + "compact_home_total_two_step", + "subagent_project_autodiscovery_non_ephemeral", + "subagent_project_declared_non_ephemeral", + "subagent_home_declared_non_ephemeral", + "subagent_project_declared_ephemeral", +) + +PARENT_THREAD_FAILURE_RE = re.compile( + r"collab\s+spawn\s+failed:\s+no\s+thread\s+with\s+id", + re.IGNORECASE, +) + + +def _write(path: Path, text: str) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(text, encoding="utf-8") + + +def _hook_script() -> str: + return r'''from __future__ import annotations +import json, sys +from pathlib import Path +kind = sys.argv[1] +log_path = Path(sys.argv[2]) +expected_cwd = Path(sys.argv[3]).resolve() +decision = sys.argv[4] +try: + event = json.load(sys.stdin) +except Exception: + event = {} +record = { + "event": kind, + "hook_event_name": event.get("hook_event_name"), + "tool_name": event.get("tool_name"), + "cwd_matches_repo": Path.cwd().resolve() == expected_cwd, +} +log_path.parent.mkdir(parents=True, exist_ok=True) +with log_path.open("a", encoding="utf-8") as handle: + handle.write(json.dumps(record, sort_keys=True) + "\n") +if kind == "PreToolUse": + specific = {"hookEventName": "PreToolUse", "permissionDecision": decision} + if decision == "deny": + specific["permissionDecisionReason"] = "PLANANVIL_DIAG_PRETOOL_DENY" + print(json.dumps({"hookSpecificOutput": specific}, sort_keys=True)) +else: + print("{}") +''' + + +def _subagent_script(token: str) -> str: + return prior._subagent_script(token) + + +def _agent_toml() -> str: + return prior._agent_toml() + + +def _command(script: Path, *args: str) -> str: + return prior._command(script, *args) + + +def _project_trust_toml(repo: Path) -> str: + return ( + f"[projects.{base.toml_quote(str(repo.resolve()))}]\n" + 'trust_level = "trusted"\n' + ) + + +def _isolated_home( + root: Path, name: str, repo: Path, *, persisted_trust: bool = True +) -> tuple[Path, Path | None, Any]: + home, auth_path, auth_before = c13._prepare_isolated_codex_home( + root / "isolated-home" / name + ) + config = "[features]\nhooks = true\n" + if persisted_trust: + config += "\n" + _project_trust_toml(repo) + _write(home / "config.toml", config) + return home, auth_path, auth_before + + +def _runtime_args( + repo: Path, + prompt: str, + *, + home: Path, + ephemeral: bool, + sandbox: str = "workspace-write", + hook_trust: bool = True, + extra_config: list[str] | None = None, + cli_trust: bool = False, + ignore_user_config: bool = False, +) -> list[str]: + args = ["codex", "exec"] + if ephemeral: + args.append("--ephemeral") + if ignore_user_config: + args.append("--ignore-user-config") + args += [ + "--strict-config", + "--json", + "--model", + MODEL, + "--sandbox", + sandbox, + "-c", + 'approval_policy="never"', + "-c", + "sandbox_workspace_write.network_access=false", + "-c", + "features.hooks=true", + ] + if cli_trust: + args += [ + "-c", + f"projects.{base.toml_quote(str(repo.resolve()))}.trust_level=\"trusted\"", + ] + if hook_trust: + args.append("--dangerously-bypass-hook-trust") + for value in extra_config or []: + args += ["-c", value] + if not ephemeral: + args += [ + "-c", + 'history.persistence="none"', + "-c", + f"sqlite_home={base.toml_quote(str((home / 'sqlite').resolve()))}", + "-c", + f"log_dir={base.toml_quote(str((home / 'log').resolve()))}", + ] + args.append(prompt) + return args + + +def _execute( + name: str, + repo: Path, + args: list[str], + *, + home: Path, + sidecar: Path, + timeout: int, + secret: str | None = None, +) -> tuple[dict[str, Any], str, str]: + env = os.environ.copy() + env["CODEX_HOME"] = str(home) + return prior._execute( + name, + repo, + args, + env=env, + timeout=timeout, + sidecar=sidecar, + secret=secret, + ) + + +def _hooks_json(script: Path, log: Path, repo: Path, *, decision: str) -> str: + session = _command( + script, "SessionStart", str(log.resolve()), str(repo.resolve()), decision + ) + pretool = _command( + script, "PreToolUse", str(log.resolve()), str(repo.resolve()), decision + ) + return json.dumps( + { + "hooks": { + "SessionStart": [ + {"hooks": [{"type": "command", "command": session, "timeout": 30}]} + ], + "PreToolUse": [ + { + "matcher": "^Bash$", + "hooks": [ + {"type": "command", "command": pretool, "timeout": 30} + ], + } + ], + } + }, + indent=2, + sort_keys=True, + ) + "\n" + + +def _hooks_toml(script: Path, log: Path, repo: Path, *, decision: str) -> str: + session = _command( + script, "SessionStart", str(log.resolve()), str(repo.resolve()), decision + ) + pretool = _command( + script, "PreToolUse", str(log.resolve()), str(repo.resolve()), decision + ) + return ( + "[features]\nhooks = true\n\n" + "[[hooks.SessionStart]]\n\n" + "[[hooks.SessionStart.hooks]]\n" + 'type = "command"\n' + f"command = {json.dumps(session)}\n" + "timeout = 30\n\n" + "[[hooks.PreToolUse]]\n" + 'matcher = "^Bash$"\n\n' + "[[hooks.PreToolUse.hooks]]\n" + 'type = "command"\n' + f"command = {json.dumps(pretool)}\n" + "timeout = 30\n" + ) + + +def _hook_variant( + root: Path, + output: Path, + name: str, + *, + source: str, + representation: str, + ephemeral: bool, + cli_trust: bool, +) -> dict[str, Any]: + repo = root / "fixtures" / name + legacy._init_repo(repo) + script = root / "scripts" / f"{name}.py" + log = output / "sidecars" / f"{name}.jsonl" + _write(script, _hook_script()) + log.unlink(missing_ok=True) + home, auth_path, auth_before = _isolated_home( + root, name, repo, persisted_trust=not cli_trust + ) + cleanup = auth_unchanged = True + + if source == "project": + if representation == "json": + _write(repo / ".codex" / "config.toml", "[features]\nhooks = true\n") + _write( + repo / ".codex" / "hooks.json", + _hooks_json(script, log, repo, decision="deny"), + ) + else: + _write( + repo / ".codex" / "config.toml", + _hooks_toml(script, log, repo, decision="deny"), + ) + legacy._git(repo, "add", "-A") + legacy._git(repo, "commit", "-q", "-m", "seed project hook fixture") + elif source == "home": + _write(home / "hooks.json", _hooks_json(script, log, repo, decision="deny")) + else: + raise ValueError(source) + + marker = "PLANANVIL_DIAG_HOOK_COMMAND_OK" + prompt = ( + f"Attempt exactly one real shell tool call running `printf {marker}`. " + "Do not retry if denied. Do not modify files. Then reply exactly " + "PLANANVIL_DIAG_HOOK_DONE." + ) + args = _runtime_args( + repo, + prompt, + home=home, + ephemeral=ephemeral, + hook_trust=True, + cli_trust=cli_trust, + ignore_user_config=cli_trust, + ) + try: + result, stdout, _stderr = _execute( + name, repo, args, home=home, sidecar=log, timeout=600 + ) + finally: + cleanup, auth_unchanged = c13._cleanup_isolated_codex_home( + home, auth_path, auth_before + ) + + records = result["sidecar_records"] + session_records = [r for r in records if r.get("event") == "SessionStart"] + pretool_records = [r for r in records if r.get("event") == "PreToolUse"] + command = prior._command_observation(stdout, marker) + if pretool_records and not command["marker_output_observed"]: + status = "DENY_OBSERVED" + elif session_records and not pretool_records: + status = "HOOK_CONFIG_LOADED_PRETOOL_MISSING" + elif not session_records and not pretool_records: + status = "HOOK_CONFIG_NOT_OBSERVED" + else: + status = "DENY_NOT_EFFECTIVE" + result["probe"] = { + "hook_source": source, + "representation": representation, + "ephemeral": ephemeral, + "cli_trust": cli_trust, + "session_start_count": len(session_records), + "pretool_count": len(pretool_records), + "command": command, + "isolated_home_cleanup_verified": cleanup, + "auth_metadata_unchanged": auth_unchanged, + } + result["diagnostic_status"] = status + return base.sanitize(result) + + +def _compact_recorder_script() -> str: + return r'''from __future__ import annotations +import json, sys +from pathlib import Path +event_name = sys.argv[1] +log_path = Path(sys.argv[2]) +expected_cwd = Path(sys.argv[3]).resolve() +try: + event = json.load(sys.stdin) +except Exception: + event = {} +record = { + "event": event_name, + "trigger": event.get("trigger"), + "cwd_matches_repo": Path.cwd().resolve() == expected_cwd, +} +log_path.parent.mkdir(parents=True, exist_ok=True) +with log_path.open("a", encoding="utf-8") as handle: + handle.write(json.dumps(record, sort_keys=True) + "\n") +print("{}") +''' + + +def _compact_hooks_json(script: Path, log: Path, repo: Path) -> str: + hooks: dict[str, list[dict[str, Any]]] = { + "SessionStart": [ + { + "hooks": [ + { + "type": "command", + "command": _command( + script, + "SessionStart", + str(log.resolve()), + str(repo.resolve()), + ), + "timeout": 30, + } + ] + } + ] + } + for event in ("PreCompact", "PostCompact"): + hooks[event] = [ + { + "matcher": "auto", + "hooks": [ + { + "type": "command", + "command": _command( + script, event, str(log.resolve()), str(repo.resolve()) + ), + "timeout": 30, + } + ], + } + ] + return json.dumps({"hooks": hooks}, indent=2, sort_keys=True) + "\n" + + +def _compact_variant( + root: Path, + output: Path, + name: str, + *, + scope: str, + two_step: bool, +) -> dict[str, Any]: + repo = root / "fixtures" / name + legacy._init_repo(repo) + _write( + repo / "diag-payload.txt", + ("PLANANVIL_DIAG_PAYLOAD_WORD " * PAYLOAD_WORDS).strip() + "\n", + ) + legacy._git(repo, "add", "-A") + legacy._git(repo, "commit", "-q", "-m", "seed compact payload") + + script = root / "scripts" / f"{name}.py" + log = output / "sidecars" / f"{name}.jsonl" + _write(script, _compact_recorder_script()) + log.unlink(missing_ok=True) + home, auth_path, auth_before = _isolated_home(root, name, repo) + _write(home / "hooks.json", _compact_hooks_json(script, log, repo)) + cleanup = auth_unchanged = True + + if two_step: + prompt = ( + "Run `cat diag-payload.txt` as the first shell tool call. After that result, " + "you MUST make a second separate shell tool call running " + "`printf PLANANVIL_DIAG_AFTER_COMPACT`. Do not combine commands. Then reply " + "exactly PLANANVIL_DIAG_COMPACT_DONE." + ) + else: + prompt = ( + "Run `cat diag-payload.txt` exactly once, then make no more tool calls and " + "reply exactly PLANANVIL_DIAG_COMPACT_SINGLE_DONE." + ) + args = _runtime_args( + repo, + prompt, + home=home, + ephemeral=False, + extra_config=[ + f"model_auto_compact_token_limit={COMPACT_LIMIT}", + f'model_auto_compact_token_limit_scope="{scope}"', + "features.token_budget=false", + ], + ) + try: + result, stdout, _stderr = _execute( + name, repo, args, home=home, sidecar=log, timeout=900 + ) + finally: + cleanup, auth_unchanged = c13._cleanup_isolated_codex_home( + home, auth_path, auth_before + ) + + records = result["sidecar_records"] + session = sum(r.get("event") == "SessionStart" for r in records) + pre = sum(r.get("event") == "PreCompact" for r in records) + post = sum(r.get("event") == "PostCompact" for r in records) + second = prior._command_observation( + stdout, "PLANANVIL_DIAG_AFTER_COMPACT" if two_step else None + ) + if not session: + status = "HOME_HOOK_CONFIG_NOT_OBSERVED" + elif two_step and pre and post: + status = "COMPACTION_OBSERVED" + elif two_step: + status = "COMPACTION_NOT_OBSERVED" + elif pre or post: + status = "SINGLE_STEP_COMPACTION_OBSERVED" + else: + status = "SINGLE_STEP_NO_COMPACTION" + result["probe"] = { + "hook_source": "isolated_home", + "scope": scope, + "two_step": two_step, + "session_start_count": session, + "precompact_count": pre, + "postcompact_count": post, + "command": second, + "isolated_home_cleanup_verified": cleanup, + "auth_metadata_unchanged": auth_unchanged, + } + result["diagnostic_status"] = status + return base.sanitize(result) + + +def _project_agent_config(*, declared: bool) -> str: + text = ( + "[features]\nhooks = true\nmulti_agent = true\n\n" + "[agents]\nenabled = true\nmax_concurrent_threads_per_session = 2\n" + ) + if declared: + text += ( + "\n[agents.fixture_agent]\n" + 'description = "PlanAnvil precision diagnostic child."\n' + 'config_file = "./agents/fixture-agent.toml"\n' + ) + return text + + +def _home_subagent_hooks(script: Path, log: Path, repo: Path) -> str: + return json.dumps( + { + "hooks": { + "SubagentStart": [ + { + "matcher": "^fixture_agent$", + "hooks": [ + { + "type": "command", + "command": _command( + script, str(log.resolve()), str(repo.resolve()) + ), + "timeout": 30, + } + ], + } + ] + } + }, + indent=2, + sort_keys=True, + ) + "\n" + + +def _subagent_variant( + root: Path, + output: Path, + name: str, + *, + agent_source: str, + declared: bool, + ephemeral: bool, +) -> dict[str, Any]: + repo = root / "fixtures" / name + legacy._init_repo(repo) + token = secrets.token_hex(16) + script = root / "scripts" / f"{name}.py" + log = output / "sidecars" / f"{name}.jsonl" + _write(script, _subagent_script(token)) + log.unlink(missing_ok=True) + home, auth_path, auth_before = _isolated_home(root, name, repo) + cleanup = auth_unchanged = True + _write(home / "hooks.json", _home_subagent_hooks(script, log, repo)) + + if agent_source == "project": + _write( + repo / ".codex" / "config.toml", + _project_agent_config(declared=declared), + ) + _write(repo / ".codex" / "agents" / "fixture-agent.toml", _agent_toml()) + legacy._git(repo, "add", "-A") + legacy._git(repo, "commit", "-q", "-m", "seed project agent fixture") + elif agent_source == "home": + config = (home / "config.toml").read_text(encoding="utf-8") + config += ( + "\n[agents]\nenabled = true\nmax_concurrent_threads_per_session = 2\n" + "\n[agents.fixture_agent]\n" + 'description = "PlanAnvil precision diagnostic child."\n' + 'config_file = "./agents/fixture-agent.toml"\n' + ) + _write(home / "config.toml", config) + _write(home / "agents" / "fixture-agent.toml", _agent_toml()) + else: + raise ValueError(agent_source) + + prompt = ( + "Start exactly one real configured subagent using spawn_agent with agent_type " + "exactly `fixture_agent`. Wait for it to finish. Do not use shell or file tools " + "in the root session. Copy the child's exact single-line response, then append " + "PLANANVIL_DIAG_SUBAGENT_DONE. Never guess an opaque context value." + ) + args = _runtime_args( + repo, + prompt, + home=home, + ephemeral=ephemeral, + sandbox="read-only", + extra_config=["features.multi_agent=true", "features.multi_agent_v2=false"], + ) + rollouts = 0 + try: + result, stdout, stderr = _execute( + name, + repo, + args, + home=home, + sidecar=log, + timeout=900, + secret=token, + ) + if not ephemeral: + rollouts = c13._session_rollout_count(home) + finally: + cleanup, auth_unchanged = c13._cleanup_isolated_codex_home( + home, auth_path, auth_before + ) + + records = result["sidecar_records"] + echo = f"PLANANVIL_DIAG_CHILD_CONTEXT_ECHO:{token}" in stdout + missing = "PLANANVIL_DIAG_CHILD_CONTEXT_MISSING" in stdout + parent_failure = bool(PARENT_THREAD_FAILURE_RE.search(stdout + "\n" + stderr)) + errors = result.get("event_diagnostics", {}).get("sanitized_errors", []) + unknown_agent = any( + "unknown agent_type 'fixture_agent'" in str(value) for value in errors + ) + if parent_failure: + status = "PARENT_THREAD_FAILURE" + elif unknown_agent: + status = "UNKNOWN_AGENT_TYPE" + elif records and echo: + status = "SUBAGENT_CONTEXT_OBSERVED" + elif records and missing: + status = "HOOK_OBSERVED_CHILD_CONTEXT_MISSING" + elif records: + status = "HOOK_OBSERVED_CHILD_RESULT_UNRESOLVED" + elif rollouts >= 2: + status = "CHILD_STARTED_HOOK_NOT_OBSERVED" + else: + status = "SUBAGENT_NOT_ESTABLISHED" + result["probe"] = { + "agent_source": agent_source, + "declared_role": declared, + "ephemeral": ephemeral, + "required_agent_type": "fixture_agent", + "hook_record_count": len(records), + "child_exact_context_echo": echo, + "child_missing_context": missing, + "parent_thread_failure": parent_failure, + "unknown_agent_type": unknown_agent, + "session_rollouts_created": rollouts, + "isolated_home_cleanup_verified": cleanup, + "auth_metadata_unchanged": auth_unchanged, + "opaque_token_persisted_in_evidence": False, + } + result["diagnostic_status"] = status + return base.sanitize(result) + + +Runner = Callable[[], dict[str, Any]] + + +def _run_case(name: str, runner: Runner, output: Path) -> dict[str, Any]: + return prior._run_case(name, runner, output) + + +def run_matrix(root: Path, output: Path) -> list[dict[str, Any]]: + (root / "fixtures").mkdir(parents=True, exist_ok=True) + (root / "scripts").mkdir(parents=True, exist_ok=True) + output.mkdir(parents=True, exist_ok=True) + cases: list[tuple[str, Runner]] = [ + ( + "project_cli_trust_ephemeral_deny", + lambda: _hook_variant( + root, + output, + "project_cli_trust_ephemeral_deny", + source="project", + representation="json", + ephemeral=True, + cli_trust=True, + ), + ), + ( + "project_persisted_trust_non_ephemeral_json_deny", + lambda: _hook_variant( + root, + output, + "project_persisted_trust_non_ephemeral_json_deny", + source="project", + representation="json", + ephemeral=False, + cli_trust=False, + ), + ), + ( + "project_persisted_trust_non_ephemeral_toml_deny", + lambda: _hook_variant( + root, + output, + "project_persisted_trust_non_ephemeral_toml_deny", + source="project", + representation="toml", + ephemeral=False, + cli_trust=False, + ), + ), + ( + "home_non_ephemeral_json_deny", + lambda: _hook_variant( + root, + output, + "home_non_ephemeral_json_deny", + source="home", + representation="json", + ephemeral=False, + cli_trust=False, + ), + ), + ( + "compact_home_body_after_prefix_single", + lambda: _compact_variant( + root, + output, + "compact_home_body_after_prefix_single", + scope="body_after_prefix", + two_step=False, + ), + ), + ( + "compact_home_body_after_prefix_two_step", + lambda: _compact_variant( + root, + output, + "compact_home_body_after_prefix_two_step", + scope="body_after_prefix", + two_step=True, + ), + ), + ( + "compact_home_total_two_step", + lambda: _compact_variant( + root, + output, + "compact_home_total_two_step", + scope="total", + two_step=True, + ), + ), + ( + "subagent_project_autodiscovery_non_ephemeral", + lambda: _subagent_variant( + root, + output, + "subagent_project_autodiscovery_non_ephemeral", + agent_source="project", + declared=False, + ephemeral=False, + ), + ), + ( + "subagent_project_declared_non_ephemeral", + lambda: _subagent_variant( + root, + output, + "subagent_project_declared_non_ephemeral", + agent_source="project", + declared=True, + ephemeral=False, + ), + ), + ( + "subagent_home_declared_non_ephemeral", + lambda: _subagent_variant( + root, + output, + "subagent_home_declared_non_ephemeral", + agent_source="home", + declared=True, + ephemeral=False, + ), + ), + ( + "subagent_project_declared_ephemeral", + lambda: _subagent_variant( + root, + output, + "subagent_project_declared_ephemeral", + agent_source="project", + declared=True, + ephemeral=True, + ), + ), + ] + return [_run_case(name, runner, output) for name, runner in cases] + + +def _summary(results: list[dict[str, Any]]) -> dict[str, Any]: + return { + "schema_version": "3.0", + "purpose": ( + "Codex 0.152 runtime isolation matrix; diagnostic only, never a release gate" + ), + "codex_version": base.codex_version(), + "model": MODEL, + "variant_count": len(results), + "harness_error_count": sum( + r.get("diagnostic_status") == "HARNESS_ERROR" for r in results + ), + "variants": results, + } + + +def main(argv: list[str] | None = None) -> int: + parser = argparse.ArgumentParser( + description="Run Codex 0.152 runtime-isolation probes" + ) + parser.add_argument("--root", type=Path, required=True) + parser.add_argument("--output", type=Path, required=True) + args = parser.parse_args(argv) + root, output = args.root.resolve(), args.output.resolve() + shutil.rmtree(root, ignore_errors=True) + shutil.rmtree(output, ignore_errors=True) + root.mkdir(parents=True) + output.mkdir(parents=True) + results = run_matrix(root, output) + _write( + output / "matrix-summary.json", + json.dumps(base.sanitize(_summary(results)), indent=2, sort_keys=True) + "\n", + ) + _write( + output / "README.txt", + "PlanAnvil Codex 0.152 runtime isolation matrix (schema 3.0).\n" + "All non-ephemeral probes use an isolated CODEX_HOME and clean it after the run.\n" + "Opaque subagent context tokens are redacted; raw transcripts and disposable hook scripts are not uploaded.\n" + "Variant observations are diagnostic only and never a release gate.\n", + ) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main())