weichen/Agent-System-Dev
1
0
Fork 0
Code Issues 1 Pull Requests 3 Actions 3 Packages Projects Releases Wiki Activity

Enforce complete normal CLI routing

Browse Source 
Why: Emergency Codex fallback must not run before Cursor, Antigravity, and PI have each been exhausted.

Evidence: Cursor transport 5e6c93f765c232783a7dccb05544e26c386fb031; 29 config tests, focused emergency controller test, and 449 governance tests passed with 19 skips.

Impact: Emergency-mode implementation, repair, and test roles now require exactly one enabled profile for each normal adapter while preserving configured order.

Refs: .task/feature-agent-cli-enforcement.md
This commit is contained in:
weichen
2026-10-15 19:52:47 +00:00
parent 59232669d0
commit 93aba6ba95
3 changed files with 180 additions and 57 deletions
Download Patch File Download Diff File Expand all files Collapse all files
scripts/agent_exec_config.py
+28
View File
@@ -350,6 +350,7 @@ def _build_config(raw: dict[str, Any]) -> ExecutionConfig:
_validate_emergency_profile(execution, profiles, roles)
_validate_normal_role_adapters(execution, profiles, roles)
_validate_normal_role_exactness(execution, profiles, roles)
return ExecutionConfig(
schema=schema,
@@ -487,6 +488,33 @@ def _validate_normal_role_adapters(
)
def _validate_normal_role_exactness(
execution: ExecutionPolicy,
profiles: Mapping[str, Profile],
roles: Mapping[str, tuple[str, ...]],
) -> None:
"""In active project mode each normal role needs one enabled profile per adapter.
Disabled profiles do not count and may appear alongside the enabled one.
Review is exempt.
"""
if execution.emergency_profile is None:
return
for role in ("implementation", "repair", "test"):
enabled: dict[str, list[str]] = {}
for name in roles[role]:
profile = profiles[name]
if profile.enabled:
enabled.setdefault(profile.adapter, []).append(name)
for adapter in sorted(NORMAL_ROLE_ADAPTERS):
names = enabled.get(adapter, [])
if len(names) != 1:
raise ConfigError(
f"role {role!r} must reference exactly one enabled "
f"{adapter} profile; found {names or 'none'}"
)
def _positive_bounded(value: Any, name: str, lo: int, hi: int) -> int:
if type(value) is not int or isinstance(value, bool):
raise ConfigError(f"{name} must be an integer")
tests/test_agent_exec.py
+53 -27
View File
@@ -69,6 +69,7 @@ def _write_config(
profiles: dict[str, dict[str, object]],
role_order: list[str],
review_order: list[str] | None = None,
repair_test_order: list[str] | None = None,
timeout_seconds: int = 30,
idle_timeout_seconds: int | None = None,
max_attempts_per_profile: int | None = None,
@@ -95,6 +96,9 @@ def _write_config(
order = ", ".join(role_order)
first = role_order[0]
review = ", ".join(review_order) if review_order is not None else first
repair_test = (
", ".join(repair_test_order) if repair_test_order is not None else first
)
idle_line = (
f" idle_timeout_seconds: {idle_timeout_seconds}\n"
if idle_timeout_seconds is not None
@@ -125,8 +129,8 @@ def _write_config(
roles:
implementation: [{order}]
repair: [{first}]
test: [{first}]
repair: [{repair_test}]
test: [{repair_test}]
review: [{review}]
profiles:
@@ -202,6 +206,12 @@ def _pi_help() -> str:
return "\n".join(_REQUIRED_HELP_MARKERS["pi"]) + "\n"
def _antigravity_help() -> str:
from scripts.agent_exec_adapters import _REQUIRED_HELP_MARKERS
return "\n".join(_REQUIRED_HELP_MARKERS["antigravity"]) + "\n"
class AgentExecControllerTests(unittest.TestCase):
def setUp(self) -> None:
self.tmp = tempfile.TemporaryDirectory(prefix="agent-exec-ctrl-")
@@ -627,20 +637,26 @@ class AgentExecControllerTests(unittest.TestCase):
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",
help_text=_pi_help(),
body=textwrap.dedent(
f"""\
printf 'regular\\n' >> "{launches}"
mkdir -p src
printf 'tmp\\n' > src/marker.txt
printf '%s\\n' '{{"terminal_status":"failed","summary":"regular failed clean","changed_files":["src/marker.txt"]}}'
exit 9
"""
),
normal = (
("cursor", "cursor-regular", _cursor_help()),
("antigravity", "agy-regular", _antigravity_help()),
("pi", "pi-regular", _pi_help()),
)
for adapter, executable, help_text in normal:
_write_fake_worker(
self.bindir,
executable,
help_text=help_text,
body=textwrap.dedent(
f"""\
printf '{adapter}\\n' >> "{launches}"
mkdir -p src
printf 'tmp\\n' > src/marker.txt
printf '%s\\n' '{{"terminal_status":"failed","summary":"{adapter} failed clean","changed_files":["src/marker.txt"]}}'
exit 9
"""
),
)
_write_fake_worker(
self.bindir,
"codex",
@@ -670,12 +686,16 @@ class AgentExecControllerTests(unittest.TestCase):
"""
),
)
order = [adapter for adapter, _, _ in normal]
profiles: dict[str, dict[str, object]] = {
adapter: {"adapter": adapter, "executable": executable, "model": "auto"}
for adapter, executable, _ in normal
}
profiles["codex"] = {"adapter": "codex", "executable": "codex", "model": "auto"}
self._configure(
profiles={
"regular": {"adapter": "pi", "executable": "regular", "model": "auto"},
"codex": {"adapter": "codex", "executable": "codex", "model": "auto"},
},
role_order=["regular"],
profiles=profiles,
role_order=order,
repair_test_order=order,
max_attempts_per_profile=3,
emergency_profile="codex",
)
@@ -683,7 +703,10 @@ class AgentExecControllerTests(unittest.TestCase):
self.assertEqual(loaded.execution.max_attempts_per_profile, 3)
self.assertEqual(loaded.execution.emergency_profile, "codex")
self.assertIn("codex", loaded.profiles)
self.assertEqual(loaded.profiles["regular"].adapter, "pi")
for role in ("implementation", "repair", "test"):
self.assertEqual(loaded.roles[role], tuple(order))
for adapter in order:
self.assertEqual(loaded.profiles[adapter].adapter, adapter)
self.assertEqual(loaded.profiles["codex"].adapter, "codex")
for role_names in loaded.roles.values():
self.assertNotIn("codex", role_names)
@@ -713,11 +736,11 @@ class AgentExecControllerTests(unittest.TestCase):
self.assertEqual(
recorded,
[
("regular", 1, "failed-clean", False),
("regular", 2, "failed-clean", False),
("regular", 3, "failed-clean", False),
("codex", 1, "success", True),
],
(adapter, ordinal, "failed-clean", False)
for adapter in order
for ordinal in (1, 2, 3)
]
+ [("codex", 1, "success", True)],
)
self.assertTrue(recorded[-1][3])
self.assertFalse(any(flag for _, _, _, flag in recorded[:-1]))
@@ -725,7 +748,10 @@ class AgentExecControllerTests(unittest.TestCase):
observed = [
line for line in launches.read_text(encoding="utf-8").splitlines() if line
]
self.assertEqual(observed, ["regular", "regular", "regular", "codex"])
self.assertEqual(
observed,
["cursor"] * 3 + ["antigravity"] * 3 + ["pi"] * 3 + ["codex"],
)
self.assertEqual(data["attempts"][-1]["summary"], "done")
self.assertEqual(proc.returncode, agent_exec.EXIT_SUCCESS, proc.stderr + proc.stdout)
self._assert_canonical_untouched()
tests/test_agent_exec_config.py
+99 -30
View File
@@ -406,30 +406,7 @@ class AgentExecConfigTests(unittest.TestCase):
load_config(self.repo)
def test_exposes_max_attempts_per_profile_and_emergency_profile(self) -> None:
block = MINIMAL_BLOCK.replace(
" raw_output_bytes: 10485760\n",
" raw_output_bytes: 10485760\n"
" max_attempts_per_profile: 3\n"
" emergency_profile: codex\n",
).replace(
" pi:\n"
" adapter: pi\n"
" executable: pi\n"
" model: auto\n"
" enabled: true\n",
" pi:\n"
" adapter: pi\n"
" executable: pi\n"
" model: auto\n"
" enabled: true\n"
"\n"
" codex:\n"
" adapter: codex\n"
" executable: codex\n"
" model: auto\n"
" enabled: true\n",
)
write_repo(self.repo, block)
write_repo(self.repo, _emergency_block())
cfg = load_config(self.repo)
self.assertEqual(cfg.execution.max_attempts_per_profile, 3)
self.assertEqual(cfg.execution.emergency_profile, "codex")
@@ -599,6 +576,77 @@ class AgentExecConfigTests(unittest.TestCase):
self.assertIn(role, message)
self.assertIn("cursor, antigravity, or pi", message)
def test_emergency_normal_roles_reject_missing_required_adapter(self) -> None:
for role, replacement, missing in (
("implementation", "[cursor, pi]", "antigravity"),
("repair", "[cursor, antigravity]", "pi"),
("test", "[pi, antigravity]", "cursor"),
):
with self.subTest(role=role):
write_repo(self.repo, _strict_routing_block(**{role: replacement}))
with self.assertRaises(ConfigError) as ctx:
load_config(self.repo)
message = str(ctx.exception)
self.assertIn(role, message)
self.assertIn(missing, message)
self.assertIn("exactly one enabled", message)
def test_emergency_normal_roles_reject_disabled_only_adapter(self) -> None:
for role in ("implementation", "repair", "test"):
with self.subTest(role=role):
write_repo(
self.repo,
_strict_routing_block(
**{role: "[cursor, agy_off, pi]"},
extra_profiles=_profile_lines("agy_off", "antigravity", "false"),
),
)
with self.assertRaises(ConfigError) as ctx:
load_config(self.repo)
message = str(ctx.exception)
self.assertIn(role, message)
self.assertIn("antigravity", message)
self.assertIn("exactly one enabled", message)
def test_emergency_normal_roles_reject_multiple_enabled_same_adapter(self) -> None:
for role in ("implementation", "repair", "test"):
with self.subTest(role=role):
write_repo(
self.repo,
_strict_routing_block(
**{role: "[cursor, antigravity, pi, cursor2]"},
extra_profiles=_profile_lines("cursor2", "cursor", "true"),
),
)
with self.assertRaises(ConfigError) as ctx:
load_config(self.repo)
message = str(ctx.exception)
self.assertIn(role, message)
self.assertIn("cursor2", message)
self.assertIn("exactly one enabled", message)
def test_emergency_normal_roles_allow_extra_disabled_and_keep_order(self) -> None:
write_repo(
self.repo,
_strict_routing_block(
implementation="[cursor_off, pi, cursor, antigravity]",
repair="[antigravity, pi, cursor_off, cursor]",
extra_profiles=_profile_lines("cursor_off", "cursor", "false"),
),
)
cfg = load_config(self.repo)
self.assertEqual(
cfg.roles["implementation"], ("cursor_off", "pi", "cursor", "antigravity")
)
self.assertEqual(cfg.roles["repair"], ("antigravity", "pi", "cursor_off", "cursor"))
self.assertEqual(cfg.roles["test"], ("pi", "antigravity", "cursor"))
self.assertFalse(cfg.profiles["cursor_off"].enabled)
def test_emergency_exactness_does_not_apply_to_review(self) -> None:
write_repo(self.repo, _strict_routing_block(review="[codex]"))
cfg = load_config(self.repo)
self.assertEqual(cfg.roles["review"], ("codex",))
def test_legacy_without_emergency_custom_remains_parseable(self) -> None:
write_repo(self.repo, _legacy_custom_block())
cfg = load_config(self.repo)
@@ -671,9 +719,9 @@ def _strict_routing_block(
fallback: str = "never",
max_attempts: str | None = "3",
emergency: str | None = "codex",
implementation: str = "[cursor, pi]",
repair: str = "[cursor]",
test: str = "[pi]",
implementation: str = "[cursor, antigravity, pi]",
repair: str = "[cursor, antigravity, pi]",
test: str = "[pi, antigravity, cursor]",
review: str = "[pi, codex]",
extra_profiles: str = "",
) -> str:
@@ -709,6 +757,12 @@ def _strict_routing_block(
" model: auto",
" enabled: true",
"",
" antigravity:",
" adapter: antigravity",
" executable: agy",
" model: auto",
" enabled: true",
"",
" pi:",
" adapter: pi",
" executable: pi",
@@ -727,14 +781,24 @@ def _strict_routing_block(
return "\n".join(lines) + "\n"
def _profile_lines(name: str, adapter: str, enabled: str) -> str:
return (
f" {name}:\n"
f" adapter: {adapter}\n"
f" executable: {name}\n"
" model: auto\n"
f" enabled: {enabled}"
)
def _emergency_block(
*,
include_profile: bool = True,
adapter: str = "codex",
enabled: str = "true",
implementation: str = "[cursor, pi]",
repair: str = "[cursor]",
test: str = "[pi]",
implementation: str = "[cursor, antigravity, pi]",
repair: str = "[cursor, antigravity, pi]",
test: str = "[pi, antigravity, cursor]",
review: str = "[pi]",
) -> str:
block = MINIMAL_BLOCK.replace(
@@ -747,6 +811,11 @@ def _emergency_block(
block = block.replace("repair: [cursor]", f"repair: {repair}")
block = block.replace("test: [pi]", f"test: {test}")
block = block.replace("review: [pi]", f"review: {review}")
block = block.replace(
" pi:\n",
_profile_lines("antigravity", "antigravity", "true") + "\n\n pi:\n",
1,
)
if not include_profile:
return block
return block.replace(