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
This commit is contained in:
2026-10-15 16:28:17 +00:00
parent 09ca1c6ebe
commit df0cc2d0ac
+95 -1
View File
@@ -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")