From df0cc2d0ac30a8f47b78cf5fd5085c05a4ee3eea Mon Sep 17 00:00:00 2001
From: weichen <weichen@home.local>
Date: Thu, 15 Oct 2026 16:28:17 +0000
Subject: [PATCH] Specify the Codex emergency routing contract

Why: Codex CLI must run only after every regular profile exhausts its three safe attempts.

Evidence: The focused test records three regular failed-clean attempts and fails because no emergency Codex attempt is launched.

Impact: Adds a RED controller contract test and test helper support only.

Refs: feature/agent-cli-enforcement
---
 tests/test_agent_exec.py | 96 +++++++++++++++++++++++++++++++++++++++-
 1 file changed, 95 insertions(+), 1 deletion(-)

diff --git a/tests/test_agent_exec.py b/tests/test_agent_exec.py
index b49d9c64..b6747a28 100644
--- a/tests/test_agent_exec.py
+++ b/tests/test_agent_exec.py
@@ -68,6 +68,7 @@ def _write_config(
     timeout_seconds: int = 30,
     idle_timeout_seconds: int | None = None,
     max_attempts_per_profile: int | None = None,
+    emergency_profile: str | None = None,
     summary_bytes: int = 256,
     raw_output_bytes: int = 1024,
 ) -> None:
@@ -99,6 +100,11 @@ def _write_config(
         if max_attempts_per_profile is not None
         else ""
     )
+    emergency_line = (
+        f"          emergency_profile: {emergency_profile}\n"
+        if emergency_profile is not None
+        else ""
+    )
     block = textwrap.dedent(
         f"""\
         schema: agent-execution/v1
@@ -109,7 +115,7 @@ def _write_config(
           fallback_to_host: all-unavailable
           max_parallel: 1
           timeout_seconds: {timeout_seconds}
-{idle_line}{attempts_line}          summary_bytes: {summary_bytes}
+{idle_line}{attempts_line}{emergency_line}          summary_bytes: {summary_bytes}
           raw_output_bytes: {raw_output_bytes}
 
         roles:
@@ -564,6 +570,94 @@ class AgentExecControllerTests(unittest.TestCase):
         )
         self._assert_canonical_untouched()
 
+    def test_emergency_profile_runs_once_after_regular_failed_clean(self) -> None:
+        launches = self.root / "emergency-launches.log"
+        launches.write_text("", encoding="utf-8")
+        _write_fake_worker(
+            self.bindir,
+            "regular",
+            body=textwrap.dedent(
+                f"""\
+                printf 'regular\\n' >> "{launches}"
+                workspace="$1"
+                mkdir -p "$workspace/src"
+                printf 'tmp\\n' > "$workspace/src/marker.txt"
+                exit 9
+                """
+            ),
+        )
+        _write_fake_worker(
+            self.bindir,
+            "codex",
+            body=textwrap.dedent(
+                f"""\
+                printf 'codex\\n' >> "{launches}"
+                workspace="$1"
+                result="$3"
+                mkdir -p "$workspace/src"
+                printf 'final\\n' > "$workspace/src/marker.txt"
+                printf '%s\\n' '{{"terminal_status":"completed","summary":"done","changed_files":["src/marker.txt"],"evidence":[],"scope_expansion":null}}' > "$result"
+                exit 0
+                """
+            ),
+        )
+        self._configure(
+            profiles={
+                "regular": {"adapter": "custom", "executable": "regular"},
+                "codex": {"adapter": "custom", "executable": "codex"},
+            },
+            role_order=["regular"],
+            max_attempts_per_profile=3,
+            emergency_profile="codex",
+        )
+        loaded = agent_exec.load_config(self.repo)
+        self.assertEqual(loaded.execution.max_attempts_per_profile, 3)
+        self.assertEqual(loaded.execution.emergency_profile, "codex")
+        self.assertIn("codex", loaded.profiles)
+        for role_names in loaded.roles.values():
+            self.assertNotIn("codex", role_names)
+        unit = _write_unit(self.root / "unit.json")
+        receipt = self.receipts / "emergency-fallback.json"
+        proc = self._run_cli(
+            "run",
+            "--repo",
+            str(self.repo),
+            "--role",
+            "implementation",
+            "--unit",
+            str(unit),
+            "--receipt",
+            str(receipt),
+        )
+        data = json.loads(receipt.read_text(encoding="utf-8"))
+        recorded = [
+            (
+                attempt["profile"],
+                attempt.get("attempt_ordinal"),
+                attempt["terminal_status"],
+                attempt["emergency_fallback"],
+            )
+            for attempt in data["attempts"]
+        ]
+        self.assertEqual(
+            recorded,
+            [
+                ("regular", 1, "failed-clean", False),
+                ("regular", 2, "failed-clean", False),
+                ("regular", 3, "failed-clean", False),
+                ("codex", 1, "success", True),
+            ],
+        )
+        self.assertTrue(recorded[-1][3])
+        self.assertFalse(any(flag for _, _, _, flag in recorded[:-1]))
+        self.assertEqual(data.get("host_fallback_authorizations"), 0)
+        observed = [
+            line for line in launches.read_text(encoding="utf-8").splitlines() if line
+        ]
+        self.assertEqual(observed, ["regular", "regular", "regular", "codex"])
+        self.assertEqual(proc.returncode, agent_exec.EXIT_SUCCESS, proc.stderr + proc.stdout)
+        self._assert_canonical_untouched()
+
     def test_failed_unknown_blocks_fallback(self) -> None:
         launched = self.root / "launched.log"
         launched.write_text("", encoding="utf-8")