diff --git a/docs/CODEX_QUALIFICATION_REGRESSION_AUDIT_2026-09-03.md b/docs/CODEX_QUALIFICATION_REGRESSION_AUDIT_2026-09-03.md new file mode 100644 index 0000000..441441a --- /dev/null +++ b/docs/CODEX_QUALIFICATION_REGRESSION_AUDIT_2026-09-03.md @@ -0,0 +1,108 @@ +# Codex qualification regression audit — 2026-09-03 + +## Scope + +This audit was completed before another qualification or CI run. It reviews the live-qualification controller chain, the full sanitized evidence from qualification run #9 (`33732852918`, source commit `7bf68e3fa8d6caff8ed8af1bf98f6be30d4670cd`), and the Codex 0.152.0 runtime contracts relevant to C03, C06, C08, C09, C13, and C16. + +The purpose is to remove harness-induced failure modes before spending another self-hosted Codex run. + +## Run #9 control result + +Run #9 completed the controller and uploaded a valid evidence artifact. It produced 10/16 `REPRODUCED` results. The six non-reproduced capabilities were C03, C06, C08, C09, C13, and C16. + +A capability is treated as a harness defect only when the evidence and controller establish that the test itself prevented or obscured the behavior being measured. Contradictory runtime behavior is not relabeled as a harness defect. + +## Findings + +### R1 — read-only hook telemetry can break the hook under test (C08, C09, C13) + +Severity: critical. + +The v4/v5 proxies execute the real hook, then write recorder JSONL under `.pursue/` inside the fixture repository, and only after that write forward the real hook stdout/stderr and return code to Codex. C08, C09, and C13 execute these paths with `sandbox=read-only`. A failed recorder write can therefore terminate the proxy before the real hook result reaches Codex. + +Repair: telemetry moves to a disposable external sidecar, recorder I/O is fail-open, and the real hook stdout/stderr/return code is always forwarded. Missing telemetry may block qualification but may never change hook semantics. + +### R2 — C08/C09 auto-compaction must be independent of recorder success + +Severity: high. + +Run #9 showed zero PreCompact/PostCompact records. Because R1 can suppress the real hook response, that absence does not distinguish “no compaction” from “compaction attempted but proxy failed”. + +Repair: fix R1 first, retain redundant project + CLI auto-compaction configuration, use deterministic low `body_after_prefix` thresholds, and require real sidecar PreCompact/PostCompact records. C09 still requires two complete compaction cycles and a real tool call after the second PostCompact. + +### R3 — C13 transport/discovery fallback works; remaining result is contaminated by R1 + +Severity: high. + +Run #9 established that the baseline-2.3 non-ephemeral fallback creates a session and starts the home-scoped `fixture_agent`: the child returned `C13_CONTEXT_MISSING`. Parent-thread persistence and home-scoped agent discovery are therefore no longer the immediate blocker. The same trial recorded zero SubagentStart proxy events, but that proxy uses the R1 recorder pattern. + +Repair: retain ephemeral-first, the known-error-only fallback, the home-scoped fallback agent, and the project-scoped hook; move telemetry to the external fail-open sidecar; require exactly one real SubagentStart record, `continue=false`, `additionalContext`, child echo of the opaque outer proof, repository immutability, cleanup and auth-metadata invariants. + +If the repaired probe still starts the child without a SubagentStart record/context proof, that becomes runtime evidence rather than a recorder artifact. + +### R4 — C03 uses an implicit workspace Git repository whose `.git` remains protected + +Severity: high. + +C03 reached the real PlanAnvil bootstrap, but the mandatory reversible Git probe could not create a temporary ref. `workspace-write` does not make the implicit workspace `.git` metadata writable. + +Repair: use a command-driver repository and place the synthetic PlanAnvil source repository in an explicit auxiliary writable root. Run the real outer `plan_anvil.py start`, assert source branch/head/index/file preservation and cleanup, verify current agents configuration and flat direct-child execution contract, and calculate C03 deterministically in the outer harness rather than with a model evaluator. + +### R5 — C16 has successful real trials but the model evaluator discards diagnostic basis + +Severity: critical false-blocker. + +Run #9 produced PASS for `GIT_READY`, `GIT_SIGNING_BLOCKED`, and `GIT_HOOK_BLOCKED`, with empty snapshot changes and cleanup errors. The capability was nevertheless marked `BLOCKED` because the evaluator requested diagnostic basis that the sanitized trial payload omitted. + +The product probe already returns bounded per-step diagnostics in `steps[].detail` and the fixtures use deterministic markers. + +Repair: execute/parse the real probe in the outer harness, reduce controlled diagnostics to booleans such as `signing_diagnostic_observed` and `hook_diagnostic_observed` before sanitization, and calculate C16 deterministically. Raw stderr, private paths and session data are not retained. + +### R6 — C06 must not be converted to green without an isolated runtime repro + +Severity: high. + +Run #9 completed a real direct `apply_patch` file change while the PlanAnvil PreToolUse recorder observed zero `apply_patch` events. This trial used `workspace-write`, so R1 does not explain the result. Codex 0.152.0 defines a PreToolUse payload for direct `apply_patch`, and PlanAnvil's matcher includes `apply_patch`. + +Repair: retain the integrated PlanAnvil trial and add a second minimal repository containing only one project-scoped `PreToolUse` hook matching `^apply_patch$`. If the minimal hook fires but PlanAnvil does not, C06 remains `BLOCKED` as an integration defect. If even the minimal current-runtime hook does not fire after a successful direct patch, C06 is `FAILED`. Only both hook observations plus mandatory postcondition detection may produce `REPRODUCED`. + +### R7 — model planner/evaluator dependence remains in release-gating paths + +Severity: high. + +C03 still used the generic model-driven fixture planner/evaluator and C16 delegated its final decision to the evaluator. This creates avoidable non-determinism. + +Repair: C03 and C16 join C06/C08/C09/C13 as outer-deterministic release-gate decisions. Model output may supply live runtime observations but is not the sole arbiter when the outer harness can directly verify the assertion. + +### R8 — wrapper-chain growth increases regression risk + +Severity: medium. + +The controller already chains historical overrides through v1-v6. Adding v7 would make ownership harder to audit. + +Repair: keep `live_codex_qualification_harness_v6.py` as the workflow entrypoint. It explicitly owns C03, C06, C08, C09, C13, and C16 through one regression helper and delegates all other capabilities to the already-proven chain. + +### R9 — previously reproduced capabilities + +Run #9 evidence for C01, C02, C04, C05, C07, C10, C11, C12, C14, and C15 was reviewed for the same controller-induced false-green patterns. No release-gate assertion was found to depend on the failing recorder pattern. + +C10 contains a blocked PostCompact observation, but its required expected assertions are pointer/context recovery and reconstruction from canonical files/Git; the successful SessionStart trial directly demonstrates those assertions. No C10 change is included in this repair set. + +## Repair invariants + +1. qualification remains `workflow_dispatch` only from `main`; +2. self-hosted runner labels and Environment remain unchanged; +3. `approval=never` remains unchanged; +4. model-tool network access remains disabled; +5. no `danger-full-access`, approval/sandbox bypass, privileged container, or `SYS_ADMIN` is introduced; +6. project-hook trust bypass may only bypass the interactive hook-trust prompt; +7. every Git-mutating fixture is disposable and explicitly scoped; +8. evidence stores structural booleans/counts/hashes, not transcripts, credentials, private paths, or session/thread identifiers; +9. all six repaired capabilities use deterministic outer release-gate decisions; +10. no full/self-hosted qualification is run until hosted regression tests for this repair set are green. + +## Exit criteria before the next full run + +The repair PR may be merged only if hosted CI verifies the consolidated v6 ownership/delegation contract, fail-open external hook telemetry, deterministic C03 auxiliary Git-root setup, C06 isolated PreToolUse comparison, deterministic C16 diagnostic basis, valid materialized C01-C16 packages, absence of forbidden sandbox/container flags, and the existing cross-platform core/distribution checks. + +Only after merge and green post-merge CI should a new `mode=full` qualification be dispatched. \ No newline at end of file diff --git a/tests/test_live_codex_qualification_harness_v6.py b/tests/test_live_codex_qualification_harness_v6.py index 4afbf9f..bcadc84 100644 --- a/tests/test_live_codex_qualification_harness_v6.py +++ b/tests/test_live_codex_qualification_harness_v6.py @@ -6,6 +6,7 @@ ROOT = Path(__file__).resolve().parents[1] MODULE_PATH = ROOT / "tools" / "live_codex_qualification_harness_v6.py" +REGRESSION_PATH = ROOT / "tools" / "live_codex_qualification_regression.py" WORKFLOW_PATH = ROOT / ".github" / "workflows" / "plananvil-codex-qualification.yml" BASELINE_PATH = ROOT / "docs" / "CODEX_CAPABILITY_BASELINE.md" RUNBOOK_PATH = ROOT / "docs" / "CODEX_SANDBOX_RUNBOOK.md" @@ -15,14 +16,19 @@ class LiveCodexHarnessV6Tests(unittest.TestCase): @classmethod def setUpClass(cls) -> None: cls.source = MODULE_PATH.read_text(encoding="utf-8") + cls.regression = REGRESSION_PATH.read_text(encoding="utf-8") cls.workflow = WORKFLOW_PATH.read_text(encoding="utf-8") cls.baseline = BASELINE_PATH.read_text(encoding="utf-8") cls.runbook = RUNBOOK_PATH.read_text(encoding="utf-8") - def test_v6_overrides_only_c13_and_inherits_v5(self) -> None: - self.assertIn('TARGET_CAPABILITIES = {"C13"}', self.source) - self.assertIn("import live_codex_qualification_harness_v5 as prior", self.source) + def test_v6_consolidates_exact_regression_audit_targets(self) -> None: + self.assertIn("import live_codex_qualification_regression as regression", self.source) + self.assertIn("TARGET_CAPABILITIES = regression.TARGET_CAPABILITIES", self.source) + self.assertIn('TARGET_CAPABILITIES = {"C03", "C06", "C08", "C09", "C13", "C16"}', self.regression) self.assertIn("_ORIGINAL_CAPABILITY_RUNTIME = prior.capability_runtime", self.source) + for capability_id in ("C03", "C06", "C08", "C09", "C13"): + self.assertIn(f'if capability_id == "{capability_id}"', self.source) + self.assertIn("return regression.run_c16(**common)", self.source) def test_agent_identity_is_aligned(self) -> None: self.assertIn('HOME_AGENT_NAME = "fixture_agent"', self.source) @@ -41,6 +47,7 @@ def test_fallback_separates_agent_discovery_from_project_hook(self) -> None: self.assertIn('trial_n["agent_fixture_scope"] = "disposable_CODEX_HOME"', self.source) self.assertIn('trial_n["project_agent_present"] = False', self.source) self.assertIn('trial_n["project_scoped_subagent_start_hook"] = True', self.source) + self.assertIn("regression.run_c13(_c13_runtime", self.source) def test_fallback_is_still_known_error_gated(self) -> None: self.assertIn("known_e and ALLOW_NON_EPHEMERAL_FALLBACK", self.source) @@ -54,13 +61,13 @@ def test_non_ephemeral_cleanup_and_auth_invariants_remain_required(self) -> None self.assertIn("auth_metadata_unchanged", self.source) self.assertIn("home_scoped_fixture_agent_materialized", self.source) - def test_full_workflow_enables_baseline23_fallback(self) -> None: + def test_full_workflow_stays_on_v6_and_enables_baseline23_fallback(self) -> None: self.assertIn("python3 tools/live_codex_qualification_harness_v6.py", self.workflow) self.assertIn("qualification_args=(--allow-c13-non-ephemeral-fallback)", self.workflow) self.assertIn("--only C13", self.workflow) self.assertIn("inputs.mode == 'full'", self.workflow) - def test_baseline_and_runbook_are_23(self) -> None: + def test_baseline_and_runbook_remain_23(self) -> None: self.assertIn("Baseline version:** 2.3", self.baseline) self.assertIn("ephemeral-first", self.baseline) self.assertIn("home-scoped", self.baseline) @@ -69,12 +76,13 @@ def test_baseline_and_runbook_are_23(self) -> None: self.assertIn("project-scoped", self.runbook) def test_safety_boundary_is_not_weakened(self) -> None: - self.assertNotIn("--dangerously-bypass-approvals-and-sandbox", self.source) - self.assertNotIn("danger-full-access", self.source) - self.assertNotIn("--privileged", self.source) - self.assertNotIn("SYS_ADMIN", self.source) + combined = self.source + "\n" + self.regression + self.assertNotIn("--dangerously-bypass-approvals-and-sandbox", combined) + self.assertNotIn("danger-full-access", combined) + self.assertNotIn("--privileged", combined) + self.assertNotIn("SYS_ADMIN", combined) self.assertIn('sandbox_mode = "read-only"', self.source) - self.assertIn("base.git_snapshot", self.source) + self.assertIn("base.git_snapshot", combined) if __name__ == "__main__": diff --git a/tests/test_live_codex_qualification_regression.py b/tests/test_live_codex_qualification_regression.py new file mode 100644 index 0000000..ac41b14 --- /dev/null +++ b/tests/test_live_codex_qualification_regression.py @@ -0,0 +1,95 @@ +from __future__ import annotations + +from pathlib import Path +import unittest + + +ROOT = Path(__file__).resolve().parents[1] +SOURCE_PATH = ROOT / "tools" / "live_codex_qualification_regression.py" +AUDIT_PATH = ROOT / "docs" / "CODEX_QUALIFICATION_REGRESSION_AUDIT_2026-09-03.md" + + +class QualificationRegressionAuditTests(unittest.TestCase): + @classmethod + def setUpClass(cls) -> None: + cls.source = SOURCE_PATH.read_text(encoding="utf-8") + cls.audit = AUDIT_PATH.read_text(encoding="utf-8") + + def test_exact_audited_capability_set(self) -> None: + self.assertIn('TARGET_CAPABILITIES = {"C03", "C06", "C08", "C09", "C13", "C16"}', self.source) + + def test_hook_telemetry_is_external_and_fail_open(self) -> None: + self.assertIn('HOOK_LOG_ENV = "PLANANVIL_QUAL_HOOK_LOG"', self.source) + self.assertIn('os.environ["TMPDIR"] = str(path.parent.resolve())', self.source) + self.assertIn('os.environ[HOOK_LOG_ENV] = str(path.resolve())', self.source) + self.assertIn("except Exception:\n pass\n# Recorder failure is never allowed to change the hook result.", self.source) + recorder_index = self.source.index("except Exception:\n pass\n# Recorder failure is never allowed to change the hook result.") + stdout_index = self.source.index("sys.stdout.write(completed.stdout)") + self.assertLess(recorder_index, stdout_index) + self.assertNotIn('qualification-hook-events.jsonl"\nlog.parent.mkdir', self.source) + + def test_compaction_repairs_use_low_redundant_triggers(self) -> None: + self.assertIn("v4.C08_COMPACT_LIMIT = 40", self.source) + self.assertIn("v4.C09_COMPACT_LIMIT = 200", self.source) + self.assertIn("expanded = [name for name in names for _ in range(2)]", self.source) + self.assertIn("return v4._c08_runtime(**kwargs)", self.source) + self.assertIn("return v4._c09_runtime(**kwargs)", self.source) + + def test_c03_uses_explicit_auxiliary_git_source_and_outer_decision(self) -> None: + self.assertIn('source = worktrees / "source"', self.source) + self.assertIn('planning = worktrees / "planning"', self.source) + self.assertIn('".agents/skills/plan-anvil/scripts/plan_anvil.py", "start"', self.source) + self.assertIn("source_core_unchanged", self.source) + self.assertIn("execution_contract_findings", self.source) + self.assertIn("flat\\s+direct[- ]child\\s+topology", self.source) + c03 = self.source[self.source.index("def run_c03("):self.source.index("# C06")] + self.assertNotIn("evaluator", c03.lower()) + + def test_c06_has_integrated_and_minimal_apply_patch_repros(self) -> None: + self.assertIn('"matcher": "^apply_patch$"', self.source) + self.assertIn("minimal_apply_patch_pretooluse", self.source) + self.assertIn("supported_apply_patch", self.source) + self.assertIn("The isolated PreToolUse hook fired, but the installed PlanAnvil PreToolUse hook did not.", self.source) + self.assertIn("Direct apply_patch completed but the isolated current-runtime PreToolUse hook did not fire.", self.source) + self.assertIn("outer_non_intercepted_postcondition", self.source) + + def test_c13_reuses_baseline23_transport_with_repaired_telemetry(self) -> None: + self.assertIn("def run_c13(current_runtime", self.source) + self.assertIn("_patched_v5_c13", self.source) + self.assertIn("v5._c13_hook_proxy_source = _fail_open_proxy_source", self.source) + self.assertIn("v5._run_c13_codex = run_c13", self.source) + + def test_c16_uses_outer_diagnostic_basis_without_raw_diagnostics(self) -> None: + self.assertIn("def _outer_probe", self.source) + self.assertIn("signing_diagnostic_observed", self.source) + self.assertIn("hook_diagnostic_observed", self.source) + self.assertIn('"raw_diagnostics_retained=false"', self.source) + c16 = self.source[self.source.index("def run_c16("):] + self.assertNotIn("evaluator", c16.lower()) + + def test_safety_invariants(self) -> None: + for forbidden in ( + "--dangerously-bypass-approvals-and-sandbox", + "danger-full-access", + "--privileged", + "SYS_ADMIN", + ): + self.assertNotIn(forbidden, self.source) + self.assertIn('sandbox="workspace-write"', self.source) + self.assertIn("_patched_v4", self.source) + self.assertIn("_patched_v5_c13", self.source) + + def test_audit_records_root_causes_and_exit_gate(self) -> None: + for marker in ( + "R1 — read-only hook telemetry", + "R4 — C03", + "R5 — C16", + "R6 — C06", + "R8 — wrapper-chain growth", + "no full/self-hosted qualification is run until hosted regression tests", + ): + self.assertIn(marker, self.audit) + + +if __name__ == "__main__": + unittest.main() diff --git a/tools/live_codex_qualification_harness_v6.py b/tools/live_codex_qualification_harness_v6.py index 2d69cbb..490fa11 100644 --- a/tools/live_codex_qualification_harness_v6.py +++ b/tools/live_codex_qualification_harness_v6.py @@ -7,10 +7,11 @@ from typing import Any import live_codex_qualification_harness_v5 as prior +import live_codex_qualification_regression as regression base = prior.base -TARGET_CAPABILITIES = {"C13"} +TARGET_CAPABILITIES = regression.TARGET_CAPABILITIES _ORIGINAL_CAPABILITY_RUNTIME = prior.capability_runtime ALLOW_NON_EPHEMERAL_FALLBACK = False HOME_AGENT_NAME = "fixture_agent" @@ -136,8 +137,6 @@ def _c13_runtime( with prior.prior.v2._python_bytecode_disabled(): proof = prior._context_proof(source_commit) - # Attempt 1 remains project-scoped and ephemeral. Filename, declared name, - # prompt target, and hook matcher are deliberately identical. base.ensure_git_repo(project_repo) _seed_project_fixture(project_repo, proof, include_project_agent=True) project_fixture_commit = base.commit_fixture_baseline(project_repo) @@ -190,11 +189,6 @@ def _c13_runtime( else "ephemeral_unclassified_blocker" ) else: - # Run #8 proved that non-ephemeral registration survives the parent-thread - # failure but the project-scoped synthetic agent can still be rejected before - # SubagentStart. Baseline 2.3 therefore isolates the semantic probe from that - # independent discovery limitation: the synthetic child is home-scoped inside - # a disposable CODEX_HOME while the hook under test remains project-scoped. fallback_used = True base.ensure_git_repo(fallback_repo) _seed_project_fixture(fallback_repo, proof, include_project_agent=False) @@ -354,7 +348,17 @@ def capability_runtime(**kwargs: Any) -> tuple[str, bool]: if capability_id not in TARGET_CAPABILITIES: return _ORIGINAL_CAPABILITY_RUNTIME(**kwargs) common = {key: value for key, value in kwargs.items() if key != "capability_id"} - return _c13_runtime(**common) + if capability_id == "C03": + return regression.run_c03(**common) + if capability_id == "C06": + return regression.run_c06(**common) + if capability_id == "C08": + return regression.run_c08(**common) + if capability_id == "C09": + return regression.run_c09(**common) + if capability_id == "C13": + return regression.run_c13(_c13_runtime, **common) + return regression.run_c16(**common) def main(argv: list[str] | None = None) -> int: @@ -364,8 +368,6 @@ def main(argv: list[str] | None = None) -> int: ALLOW_NON_EPHEMERAL_FALLBACK = allow_flag in args args = [item for item in args if item != allow_flag] - # The v5 C13-only controller is still useful as a short diagnostic surface, - # but baseline 2.3 also permits the same narrowly gated fallback in full mode. base.capability_runtime = capability_runtime prior.capability_runtime = capability_runtime if "--only" in args: diff --git a/tools/live_codex_qualification_regression.py b/tools/live_codex_qualification_regression.py new file mode 100644 index 0000000..36e860e --- /dev/null +++ b/tools/live_codex_qualification_regression.py @@ -0,0 +1,612 @@ +from __future__ import annotations + +import contextlib +import hashlib +import json +import os +import re +import shutil +import subprocess +import sys +from pathlib import Path +from typing import Any, Callable, Iterator + +import live_codex_qualification_harness as v1 +import live_codex_qualification_harness_v2 as v2 +import live_codex_qualification_harness_v4 as v4 +import live_codex_qualification_harness_v5 as v5 + +base = v5.base +TARGET_CAPABILITIES = {"C03", "C06", "C08", "C09", "C13", "C16"} +HOOK_LOG_ENV = "PLANANVIL_QUAL_HOOK_LOG" + + +def _write(path: Path, text: str) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(text, encoding="utf-8") + + +def _sha256(path: Path) -> str: + return hashlib.sha256(path.read_bytes()).hexdigest() + + +def _runtime_paths(**kwargs: Any): + return v2._runtime_paths(**kwargs) + + +def _write_result(**kwargs: Any): + return v4._write_result(**kwargs) + + +def _sidecar(cap_runtime: Path, name: str) -> Path: + path = cap_runtime / "hook-telemetry" / f"{name}.jsonl" + path.parent.mkdir(parents=True, exist_ok=True) + return path + + +def _read_records(path: Path) -> list[dict[str, Any]]: + if not path.is_file(): + return [] + out: list[dict[str, Any]] = [] + for line in path.read_text(encoding="utf-8").splitlines(): + try: + item = json.loads(line) + except json.JSONDecodeError: + continue + if isinstance(item, dict): + out.append(item) + return out + + +@contextlib.contextmanager +def _hook_env(path: Path) -> Iterator[None]: + path.unlink(missing_ok=True) + path.parent.mkdir(parents=True, exist_ok=True) + previous_log = os.environ.get(HOOK_LOG_ENV) + previous_tmp = os.environ.get("TMPDIR") + os.environ[HOOK_LOG_ENV] = str(path.resolve()) + os.environ["TMPDIR"] = str(path.parent.resolve()) + try: + yield + finally: + if previous_log is None: + os.environ.pop(HOOK_LOG_ENV, None) + else: + os.environ[HOOK_LOG_ENV] = previous_log + if previous_tmp is None: + os.environ.pop("TMPDIR", None) + else: + os.environ["TMPDIR"] = previous_tmp + + +def _fail_open_proxy_source() -> str: + return r'''from __future__ import annotations +import json, os, subprocess, sys +from pathlib import Path + +event_name, script_name = sys.argv[1], sys.argv[2] +raw = sys.stdin.read() +try: + event = json.loads(raw) if raw.strip() else {} +except json.JSONDecodeError: + event = {} +root = Path(subprocess.check_output(["git", "rev-parse", "--show-toplevel"], text=True).strip()) +completed = subprocess.run( + [sys.executable, str(root / ".codex" / "hooks" / script_name)], + input=raw, text=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, check=False, +) +record = {"event": event_name, "returncode": completed.returncode} +if isinstance(event, dict) and isinstance(event.get("tool_name"), str): + record["tool_name"] = event["tool_name"] +try: + parsed = json.loads(completed.stdout) if completed.stdout.strip() else {} +except json.JSONDecodeError: + parsed = {} +if isinstance(parsed, dict): + if "continue" in parsed: + record["continue"] = parsed.get("continue") + reason = parsed.get("stopReason") + if isinstance(reason, str): + lowered = reason.lower() + record["stop_reason_mentions_checkpoint"] = "checkpoint" in lowered + record["stop_reason_mentions_recovery"] = "recovery" in lowered or "canonical state" in lowered + output = parsed.get("hookSpecificOutput") + if isinstance(output, dict): + record["additional_context"] = bool(output.get("additionalContext")) + if isinstance(output.get("permissionDecision"), str): + record["permission_decision"] = output["permissionDecision"] +try: + log_value = os.environ.get("PLANANVIL_QUAL_HOOK_LOG") + if log_value: + log = Path(log_value) + log.parent.mkdir(parents=True, exist_ok=True) + with log.open("a", encoding="utf-8") as handle: + handle.write(json.dumps(record, sort_keys=True) + "\n") +except Exception: + pass +# Recorder failure is never allowed to change the hook result. +sys.stdout.write(completed.stdout) +sys.stderr.write(completed.stderr) +raise SystemExit(completed.returncode) +''' + + +def _compact_prompt(capability_id: str, names: list[str]) -> str: + # Duplicate every large read so a low body-after-prefix threshold must be + # crossed before the final status commands, while preserving one model turn. + expanded = [name for name in names for _ in range(2)] + commands = "\n".join( + f"- run `cat qualification-payload/{name}` in a separate shell-tool call" + for name in expanded + ) + return f"""Exercise genuine Codex automatic compaction for {capability_id}. +Do not invoke hook scripts directly, simulate hook events, or inspect qualification telemetry. +Execute these reads in order, continuing normally after any automatic compaction: +{commands} +After the reads, run `git status --porcelain=v1 --untracked-files=all` and `git rev-parse HEAD`. +Return capability_id {capability_id} with only minimal structural observations. +""" + + +@contextlib.contextmanager +def _patched_v4(cap_runtime: Path, capability_id: str) -> Iterator[None]: + old_proxy = v4._hook_proxy_source + old_hook_log = v4._hook_log + old_run = v4._run_codex_probe + old_prompt = v4._compact_probe_prompt + old_c08 = v4.C08_COMPACT_LIMIT + old_c09 = v4.C09_COMPACT_LIMIT + log = _sidecar(cap_runtime, capability_id.lower()) + + def hook_log(_repo: Path) -> Path: + return log + + def run_probe(**kwargs: Any): + with _hook_env(log): + return old_run(**kwargs) + + v4._hook_proxy_source = _fail_open_proxy_source + v4._hook_log = hook_log + v4._run_codex_probe = run_probe + v4._compact_probe_prompt = _compact_prompt + v4.C08_COMPACT_LIMIT = 40 + v4.C09_COMPACT_LIMIT = 200 + try: + yield + finally: + v4._hook_proxy_source = old_proxy + v4._hook_log = old_hook_log + v4._run_codex_probe = old_run + v4._compact_probe_prompt = old_prompt + v4.C08_COMPACT_LIMIT = old_c08 + v4.C09_COMPACT_LIMIT = old_c09 + + +def run_c08(**kwargs: Any): + cap_runtime = Path(kwargs["runtime_root"]) / "C08" + with _patched_v4(cap_runtime, "C08"): + return v4._c08_runtime(**kwargs) + + +def run_c09(**kwargs: Any): + cap_runtime = Path(kwargs["runtime_root"]) / "C09" + with _patched_v4(cap_runtime, "C09"): + return v4._c09_runtime(**kwargs) + + +@contextlib.contextmanager +def _patched_v5_c13(cap_runtime: Path) -> Iterator[None]: + old_proxy = v5._c13_hook_proxy_source + old_hook_log = v5._hook_log + old_run = v5._run_c13_codex + + def hook_log(repo: Path) -> Path: + return _sidecar(cap_runtime, f"c13-{repo.name}") + + def run_c13(**kwargs: Any): + log = hook_log(Path(kwargs["repo"])) + with _hook_env(log): + return old_run(**kwargs) + + v5._c13_hook_proxy_source = _fail_open_proxy_source + v5._hook_log = hook_log + v5._run_c13_codex = run_c13 + try: + yield + finally: + v5._c13_hook_proxy_source = old_proxy + v5._hook_log = old_hook_log + v5._run_c13_codex = old_run + + +def run_c13(current_runtime: Callable[..., tuple[str, bool]], **kwargs: Any): + cap_runtime = Path(kwargs["runtime_root"]) / "C13" + with _patched_v5_c13(cap_runtime): + return current_runtime(**kwargs) + + +# C03 ----------------------------------------------------------------------- + +def run_c03( + *, root: Path, runtime_root: Path, schemas: dict[str, Path], version: str, + os_name: str, source_commit: str, date: str, +) -> tuple[str, bool]: + cid = "C03" + _cap, cap_runtime, _spec, driver, worktrees, results, _eval = _runtime_paths( + root=root, runtime_root=runtime_root, capability_id=cid + ) + with v2._python_bytecode_disabled(): + base.ensure_git_repo(driver) + _write(driver / "README.md", "C03 command driver.\n") + base.commit_fixture_baseline(driver) + source = worktrees / "source" + base.ensure_git_repo(source) + branch = base.git(source, "branch", "--show-current", check=False) + if branch and branch != "main": + base.git(source, "branch", "-m", "main") + v1._install_plananvil_release(root, source) + _write(source / "README.md", "C03 topology fixture.\n") + fixture_commit = base.commit_fixture_baseline(source) + before = base.git_snapshot(source) + planning = worktrees / "planning" + start = base.run([ + sys.executable, ".agents/skills/plan-anvil/scripts/plan_anvil.py", "start", + "--source", ".", "--destination", str(planning), + "--goal", "Qualify PlanAnvil flat direct-child topology", + "--codex-version", version, "--model", base.MODEL, + "--permission-mode", "approval=never; sandbox=workspace-write", + "--project-trust", "TRUSTED", "--hook-mode", "HOOKS_TRUSTED", + ], cwd=source, check=False, timeout=240) + try: + start_payload = v1._parse_json_stdout(start, "PlanAnvil start") + except Exception: + start_payload = {} + bootstrap_ok = start.returncode == 0 and bool(start_payload.get("planning_worktree")) + after_bootstrap = base.git_snapshot(source) + source_core_unchanged = all( + before.get(key) == after_bootstrap.get(key) + for key in ("head", "branch", "status", "index_tree", "cached_paths") + ) + config = (source / ".codex" / "config.toml").read_text(encoding="utf-8") + config_ok = ( + "enabled = true" in config + and "max_concurrent_threads_per_session" in config + and "max_depth" not in config + ) + contract = root / ".agents" / "skills" / "plan-anvil" / "references" / "execution-contract.md" + contract_text = contract.read_text(encoding="utf-8") + contract_ok = ( + re.search(r"flat\s+direct[- ]child\s+topology", contract_text, re.I) is not None + and "agents.max_depth" not in contract_text + and "max_depth" not in contract_text + ) + golden = root / "examples" / "small-change" / "run" / "PLAN.md" + code = r'''import json, sys +from pathlib import Path +sys.path.insert(0, str(Path('.agents/skills/plan-anvil/scripts').resolve())) +from execution_contract import execution_contract_findings +print(json.dumps(execution_contract_findings(Path(sys.argv[1]).read_text(encoding='utf-8')), sort_keys=True)) +''' + check = base.run([sys.executable, "-B", "-c", code, str(golden.resolve())], cwd=source, check=False, timeout=120) + try: + findings = json.loads(check.stdout) if check.returncode == 0 else [] + except json.JSONDecodeError: + findings = [] + validator_ok = check.returncode == 0 and not any( + isinstance(item, dict) and item.get("kind") == "execution-contract-topology-missing" + for item in findings + ) + prompt = ( + "Use only shell tools to verify the explicit auxiliary C03 source is reachable and clean. " + "Run exactly `cd ../worktrees/source && git status --porcelain=v1 && git rev-parse HEAD`. " + "Do not modify files or start subagents. Return capability_id C03, trial explicit_plananvil_topology, " + "outcome PASS when both commands succeed and status is clean." + ) + payload, events, error = v4._run_codex_probe( + cwd=driver, prompt=prompt, schemas=schemas, results_dir=results, + position=1, sandbox="workspace-write", add_dir=worktrees, timeout=600, + ) + live_ok = error is None and payload.get("outcome") == "PASS" + assertions = { + "bootstrap_git_probe_completed": bootstrap_ok, + "source_branch_head_index_files_preserved": source_core_unchanged, + "agents_config_current_and_no_max_depth": config_ok, + "plananvil_contract_flat_direct_child_and_no_max_depth": contract_ok, + "current_golden_plan_passes_topology_validator": validator_ok, + "live_codex_auxiliary_source_probe": live_ok, + } + if not bootstrap_ok or not live_ok: + result, met = "BLOCKED", False + blocker = error or "C03 auxiliary PlanAnvil bootstrap/live probe did not complete." + elif not all(assertions.values()): + result, met = "FAILED", False + blocker = "Current PlanAnvil topology/configuration contradicted the C03 contract." + else: + result, met, blocker = "REPRODUCED", True, None + trial = { + "capability_id": cid, "trial": "explicit_plananvil_topology", "trial_name": "explicit_plananvil_topology", + "outcome": "PASS" if met else ("BLOCKED" if result == "BLOCKED" else "FAIL"), + "assertions": [ + {"name": key, "status": "PASS" if value else ("BLOCKED" if result == "BLOCKED" else "FAIL"), "evidence": str(value).lower()} + for key, value in assertions.items() + ], + "observations": [f"{key}={str(value).lower()}" for key, value in assertions.items()], + "blocker": blocker, "event_summary": events, "model_payload": payload, + "contract_sha256": _sha256(contract), "golden_plan_sha256": _sha256(golden), + } + return _write_result( + root=root, cap_runtime=cap_runtime, capability_id=cid, result=result, + expected_met=met, observations=trial["observations"], blocker=blocker, + summary=("C03 reproduced with explicit writable Git bootstrap and deterministic flat direct-child contract validation." if met else "C03 did not fully reproduce after regression-audited setup."), + trials=[trial], fixture_commit=fixture_commit, version=version, os_name=os_name, + source_commit=source_commit, date=date, + ) + + +# C06 ----------------------------------------------------------------------- + +def _minimal_hook_source() -> str: + return r'''from __future__ import annotations +import json, os, sys +from pathlib import Path +try: + event = json.load(sys.stdin) +except Exception: + event = {} +try: + log = Path(os.environ["PLANANVIL_QUAL_HOOK_LOG"]) + log.parent.mkdir(parents=True, exist_ok=True) + with log.open("a", encoding="utf-8") as handle: + handle.write(json.dumps({"event": "PreToolUse", "tool_name": event.get("tool_name")}, sort_keys=True) + "\n") +except Exception: + pass +print(json.dumps({"hookSpecificOutput": {"hookEventName": "PreToolUse", "permissionDecision": "allow"}}, sort_keys=True)) +''' + + +def _patch_prompt(trial: str, target: str, marker: str) -> str: + return ( + f"Use the direct apply_patch tool exactly once to add `{target}` containing only `{marker}`. " + "Do not create it with shell commands and do not execute hook scripts manually. " + f"Return capability_id C06, trial {trial}, outcome PASS after the direct patch completes." + ) + + +def run_c06( + *, root: Path, runtime_root: Path, schemas: dict[str, Path], version: str, + os_name: str, source_commit: str, date: str, +) -> tuple[str, bool]: + cid = "C06" + _cap, cap_runtime, _spec, repo, _worktrees, results, _eval = _runtime_paths( + root=root, runtime_root=runtime_root, capability_id=cid + ) + with v2._python_bytecode_disabled(), _patched_v4(cap_runtime, cid): + base.ensure_git_repo(repo) + v1._install_plananvil_release(root, repo) + v4._instrument_hooks(repo, event_to_script={"PreToolUse": "plan-anvil-guard.py"}) + _write(repo / "README.md", "C06 integrated fixture.\n") + fixture_commit = base.commit_fixture_baseline(repo) + log = v4._hook_log(repo) + before = base.git_snapshot(repo) + payload_i, events_i, error_i = v4._run_codex_probe( + cwd=repo, prompt=_patch_prompt("supported_apply_patch", ".pursue/c06-supported.txt", "C06_SUPPORTED"), + schemas=schemas, results_dir=results, position=1, sandbox="workspace-write", timeout=600, + ) + after = base.git_snapshot(repo) + records_i = _read_records(log) + apply_i = [r for r in records_i if r.get("event") == "PreToolUse" and r.get("tool_name") == "apply_patch"] + integrated_mutation = (repo / ".pursue" / "c06-supported.txt").is_file() + count_before = len(records_i) + outer_before = base.git_snapshot(repo) + with (repo / "README.md").open("a", encoding="utf-8") as handle: + handle.write("C06_OUTSIDE_HOOK_LIFECYCLE\n") + outer_after = base.git_snapshot(repo) + postcondition = any( + isinstance(line, str) and line.endswith("README.md") + for line in (outer_after.get("status") or []) + ) + no_hook = len(_read_records(log)) == count_before + + minimal = cap_runtime / "minimal-repro" + base.ensure_git_repo(minimal) + _write(minimal / ".codex" / "hooks" / "minimal-pretooluse.py", _minimal_hook_source()) + _write(minimal / ".codex" / "hooks.json", json.dumps({ + "hooks": {"PreToolUse": [{"matcher": "^apply_patch$", "hooks": [{ + "type": "command", + "command": 'python3 "$(git rev-parse --show-toplevel)/.codex/hooks/minimal-pretooluse.py"', + "timeout": 30, + }]}]} + }, indent=2, sort_keys=True) + "\n") + _write(minimal / "README.md", "C06 minimal runtime hook repro.\n") + base.commit_fixture_baseline(minimal) + minimal_log = _sidecar(cap_runtime, "c06-minimal") + minimal_log.unlink(missing_ok=True) + with _hook_env(minimal_log): + payload_m, events_m, error_m = v4._run_codex_probe( + cwd=minimal, prompt=_patch_prompt("minimal_apply_patch_pretooluse", "c06-minimal.txt", "C06_MINIMAL"), + schemas=schemas, results_dir=results, position=2, sandbox="workspace-write", timeout=600, + ) + records_m = _read_records(minimal_log) + apply_m = [r for r in records_m if r.get("event") == "PreToolUse" and r.get("tool_name") == "apply_patch"] + minimal_mutation = (minimal / "c06-minimal.txt").is_file() + + if error_i or error_m or not integrated_mutation or not minimal_mutation: + result, met = "BLOCKED", False + blocker = error_i or error_m or "A direct apply_patch fixture mutation did not complete." + elif not apply_m: + result, met = "FAILED", False + blocker = "Direct apply_patch completed but the isolated current-runtime PreToolUse hook did not fire." + elif not apply_i: + result, met = "BLOCKED", False + blocker = "The isolated PreToolUse hook fired, but the installed PlanAnvil PreToolUse hook did not." + elif not (postcondition and no_hook): + result, met = "FAILED", False + blocker = "Mandatory postcondition detection failed for the intentionally unhooked mutation." + else: + result, met, blocker = "REPRODUCED", True, None + trials = [ + {"capability_id": cid, "trial": "supported_apply_patch", "trial_name": "supported_apply_patch", "outcome": "PASS" if apply_i and integrated_mutation else ("BLOCKED" if error_i else "FAIL"), "assertions": [{"name": "plananvil_apply_patch_pretooluse", "status": "PASS" if apply_i else ("BLOCKED" if error_i else "FAIL"), "evidence": f"events={len(apply_i)}; mutation={str(integrated_mutation).lower()}"}], "observations": [f"apply_patch_pretooluse_events={len(apply_i)}", f"supported_mutation={str(integrated_mutation).lower()}"], "blocker": error_i, "event_summary": events_i, "git_before": before, "git_after": after, "model_payload": payload_i}, + {"capability_id": cid, "trial": "minimal_apply_patch_pretooluse", "trial_name": "minimal_apply_patch_pretooluse", "outcome": "PASS" if apply_m and minimal_mutation else ("BLOCKED" if error_m else "FAIL"), "assertions": [{"name": "isolated_apply_patch_pretooluse", "status": "PASS" if apply_m else ("BLOCKED" if error_m else "FAIL"), "evidence": f"events={len(apply_m)}; mutation={str(minimal_mutation).lower()}"}], "observations": [f"minimal_apply_patch_pretooluse_events={len(apply_m)}", f"minimal_mutation={str(minimal_mutation).lower()}"], "blocker": error_m, "event_summary": events_m, "model_payload": payload_m}, + {"capability_id": cid, "trial": "outer_non_intercepted_postcondition", "trial_name": "outer_non_intercepted_postcondition", "outcome": "PASS" if postcondition and no_hook else "FAIL", "assertions": [{"name": "postcondition_detection", "status": "PASS" if postcondition and no_hook else "FAIL", "evidence": f"detected={str(postcondition).lower()}; hook_count_unchanged={str(no_hook).lower()}"}], "observations": [f"git_postcondition_detected={str(postcondition).lower()}", f"hook_record_count_unchanged={str(no_hook).lower()}"], "blocker": None, "git_before": outer_before, "git_after": outer_after}, + ] + return _write_result( + root=root, cap_runtime=cap_runtime, capability_id=cid, result=result, + expected_met=met, observations=[f"plananvil_events={len(apply_i)}", f"minimal_events={len(apply_m)}", f"postcondition={str(postcondition).lower()}"], + blocker=blocker, + summary=("C06 reproduced with integrated and isolated PreToolUse plus mandatory postconditions." if met else "C06 did not fully reproduce after the isolated current-runtime comparison."), + trials=trials, fixture_commit=fixture_commit, version=version, os_name=os_name, + source_commit=source_commit, date=date, + ) + + +# C16 ----------------------------------------------------------------------- + +def _normalize_main(source: Path) -> None: + branch = base.git(source, "branch", "--show-current", check=False) + if not branch: + base.git(source, "switch", "-c", "main") + elif branch != "main": + base.git(source, "branch", "-m", "main") + + +def _prepare_c16(root: Path, source: Path) -> str: + _write(source / "source.txt", "C16 source sentinel.\n") + base.ensure_git_repo(source) + _normalize_main(source) + v1._install_plananvil_release(root, source) + base.git(source, "add", "-A") + base.git(source, "commit", "--allow-empty", "-q", "-m", "Install PlanAnvil C16 fixture") + return base.git(source, "rev-parse", "HEAD") + + +def _clear_signing(source: Path) -> None: + base.git(source, "config", "commit.gpgsign", "false") + for key in ("gpg.program", "user.signingkey"): + base.git(source, "config", "--unset-all", key, check=False) + + +def _signing_failure(source: Path, fake: Path) -> None: + _write(fake, "#!/bin/sh\necho 'gpg: signing failed: C16 fixture signing failure' >&2\nexit 1\n") + fake.chmod(0o755) + base.git(source, "config", "commit.gpgsign", "true") + base.git(source, "config", "gpg.format", "openpgp") + base.git(source, "config", "user.signingkey", "C16-FIXTURE-KEY") + base.git(source, "config", "gpg.program", str(fake.resolve())) + + +def _hook_failure(source: Path) -> Path: + hook = source / ".git" / "hooks" / "pre-commit" + _write(hook, "#!/bin/sh\necho 'pre-commit hook failed: C16 fixture hook rejection' >&2\nexit 1\n") + hook.chmod(0o755) + return hook + + +def _outer_probe(source: Path, run_id: str, probes: Path) -> tuple[dict[str, Any], str | None]: + completed = base.run([ + sys.executable, "-B", ".agents/skills/plan-anvil/scripts/test_git_capabilities.py", + "--source", ".", "--run-id", run_id, "--temp-parent", str(probes), + ], cwd=source, check=False, timeout=240) + try: + value = json.loads(completed.stdout) + except json.JSONDecodeError: + return {}, base.sanitize_text((completed.stderr or completed.stdout)[-1500:]) + return (value if isinstance(value, dict) else {}), None + + +def _diag(payload: dict[str, Any], needles: tuple[str, ...]) -> bool: + return any( + any(needle in str(step.get("detail") or "").lower() for needle in needles) + for step in payload.get("steps", []) if isinstance(step, dict) + ) + + +def _c16_prompt(run_id: str, expected: str) -> str: + return ( + "Run the real PlanAnvil reversible Git capability probe against `../worktrees/source`: " + f"`cd ../worktrees/source && PYTHONDONTWRITEBYTECODE=1 python .agents/skills/plan-anvil/scripts/test_git_capabilities.py --source . --run-id {run_id} --temp-parent ../probes`. " + f"Return PASS only when result is {expected}, source_snapshot_changed and cleanup_errors are empty, and no probe artifacts remain." + ) + + +def run_c16( + *, root: Path, runtime_root: Path, schemas: dict[str, Path], version: str, + os_name: str, source_commit: str, date: str, +) -> tuple[str, bool]: + cid = "C16" + _cap, cap_runtime, _spec, driver, worktrees, results, _eval = _runtime_paths( + root=root, runtime_root=runtime_root, capability_id=cid + ) + with v2._python_bytecode_disabled(): + base.ensure_git_repo(driver) + _write(driver / "README.md", "C16 command driver.\n") + base.commit_fixture_baseline(driver) + source = worktrees / "source" + fixture_commit = _prepare_c16(root, source) + probes = worktrees / "probes" + probes.mkdir(parents=True, exist_ok=True) + scenarios = [ + ("git_ready_probe", "c16-ready", "GIT_READY", "ready"), + ("signing_failure_probe", "c16-signing", "GIT_SIGNING_BLOCKED", "signing"), + ("repository_hook_failure_probe", "c16-hook", "GIT_HOOK_BLOCKED", "hook"), + ] + trials: list[dict[str, Any]] = [] + checks: list[bool] = [] + model_checks: list[bool] = [] + for position, (name, run_id, expected, kind) in enumerate(scenarios, 1): + hook: Path | None = None + if kind == "signing": + _signing_failure(source, worktrees / "support" / "fake-gpg") + elif kind == "hook": + _clear_signing(source) + hook = _hook_failure(source) + else: + _clear_signing(source) + outer, outer_error = _outer_probe(source, run_id + "-outer", probes) + signing_diag = _diag(outer, ("gpg failed to sign", "signing failed", "failed to sign")) + hook_diag = _diag(outer, ("pre-commit hook failed", "hook rejection", "hook failed")) + outer_ok = ( + outer_error is None and outer.get("result") == expected + and not outer.get("source_snapshot_changed") and not outer.get("cleanup_errors") + and (kind != "signing" or signing_diag) and (kind != "hook" or hook_diag) + ) + before = base.git_snapshot(source) + payload = v2._run_trial( + capability_id=cid, + trial={"name": name, "sandbox": "workspace-write", "prompt": _c16_prompt(run_id, expected)}, + cwd=driver, snapshot_repo=source, schemas=schemas, results_dir=results, + position=position, add_dir=worktrees, timeout=600, + ) + after = base.git_snapshot(source) + model_ok = payload.get("outcome") == "PASS" and before == after + payload["outer_diagnostic_basis"] = { + "expected_classification": expected, + "outer_result_matches": outer.get("result") == expected, + "source_snapshot_changed_empty": not bool(outer.get("source_snapshot_changed")), + "cleanup_errors_empty": not bool(outer.get("cleanup_errors")), + "signing_diagnostic_observed": signing_diag, + "hook_diagnostic_observed": hook_diag, + "outer_probe_error": outer_error, + } + trials.append(base.sanitize(payload)) + checks.append(outer_ok) + model_checks.append(model_ok) + if hook is not None: + hook.unlink(missing_ok=True) + if kind == "signing": + _clear_signing(source) + if not all(checks): + result, met = "FAILED", False + blocker = "A deterministic outer C16 classification/diagnostic/cleanup assertion failed." + elif not all(model_checks): + result, met = "BLOCKED", False + blocker = "Deterministic C16 probes passed, but one matching live Codex invocation was incomplete." + else: + result, met, blocker = "REPRODUCED", True, None + return _write_result( + root=root, cap_runtime=cap_runtime, capability_id=cid, result=result, + expected_met=met, observations=[f"outer_all_pass={str(all(checks)).lower()}", f"live_codex_all_pass={str(all(model_checks)).lower()}", "raw_diagnostics_retained=false"], + blocker=blocker, + summary=("C16 reproduced with deterministic diagnostic basis and matching live Codex probes." if met else "C16 did not fully reproduce after deterministic diagnostic verification."), + trials=trials, fixture_commit=fixture_commit, version=version, os_name=os_name, + source_commit=source_commit, date=date, + )