"""Configuration parser and schema contract for AGENT-EXECUTION.md.""" from __future__ import annotations import tempfile import textwrap import unittest from pathlib import Path from scripts.agent_exec_config import ( DEFAULT_IDLE_TIMEOUT_SECONDS, ConfigError, ExecutionConfig, load_config, validate_config, ) SCHEMA = "agent-execution/v1" MINIMAL_BLOCK = textwrap.dedent( """\ schema: agent-execution/v1 execution: fresh_process_per_unit: true permission: run-everything fallback_to_host: all-unavailable max_parallel: 3 timeout_seconds: 7200 summary_bytes: 16384 raw_output_bytes: 10485760 roles: implementation: [cursor, antigravity, pi] repair: [cursor, antigravity, pi] test: [pi, antigravity, cursor] review: [pi] profiles: cursor: adapter: cursor executable: cursor-agent model: auto enabled: true antigravity: adapter: antigravity executable: agy model: auto enabled: true pi: adapter: pi executable: pi model: auto enabled: true """ ) def write_repo(root: Path, body: str, *, extra_blocks: list[str] | None = None) -> Path: parts = ["# Fixture\n", "```agent-execution\n", body.rstrip(), "\n```\n"] if extra_blocks: for block in extra_blocks: parts.extend(["\n```agent-execution\n", block.rstrip(), "\n```\n"]) (root / "AGENT-EXECUTION.md").write_text("".join(parts), encoding="utf-8") return root class AgentExecConfigTests(unittest.TestCase): def setUp(self) -> None: self.tmp = tempfile.TemporaryDirectory(prefix="agent-exec-config-") self.addCleanup(self.tmp.cleanup) self.repo = Path(self.tmp.name) def test_loads_single_agent_execution_block(self) -> None: write_repo(self.repo, MINIMAL_BLOCK) self.assertTrue(issubclass(ConfigError, ValueError)) cfg = load_config(self.repo) self.assertIsInstance(cfg, ExecutionConfig) self.assertEqual(cfg.schema, SCHEMA) self.assertTrue(cfg.execution.fresh_process_per_unit) self.assertEqual(cfg.execution.permission, "run-everything") self.assertEqual(cfg.execution.fallback_to_host, "all-unavailable") self.assertEqual(cfg.roles["implementation"], ("cursor", "antigravity", "pi")) self.assertEqual(cfg.profiles["cursor"].executable, "cursor-agent") self.assertEqual(cfg.custom_adapters, {}) self.assertEqual(validate_config(self.repo), []) def test_rejects_missing_or_duplicate_blocks(self) -> None: (self.repo / "AGENT-EXECUTION.md").write_text("# no block\n", encoding="utf-8") with self.assertRaises(ConfigError): load_config(self.repo) write_repo(self.repo, MINIMAL_BLOCK, extra_blocks=[MINIMAL_BLOCK]) with self.assertRaises(ConfigError): load_config(self.repo) def test_rejects_unknown_schema_and_unknown_keys(self) -> None: bad_schema = MINIMAL_BLOCK.replace( "schema: agent-execution/v1", "schema: agent-execution/v0" ) write_repo(self.repo, bad_schema) with self.assertRaises(ConfigError): load_config(self.repo) bad_key = MINIMAL_BLOCK + "\nextra_top_level: true\n" write_repo(self.repo, bad_key) with self.assertRaises(ConfigError): load_config(self.repo) bad_exec = MINIMAL_BLOCK.replace( "raw_output_bytes: 10485760\n", "raw_output_bytes: 10485760\n mystery: 1\n", ) write_repo(self.repo, bad_exec) with self.assertRaises(ConfigError): load_config(self.repo) def test_role_order_does_not_follow_profile_order(self) -> None: # profiles listed pi before cursor; role order must stay cursor -> pi block = textwrap.dedent( """\ schema: agent-execution/v1 execution: fresh_process_per_unit: true permission: run-everything fallback_to_host: all-unavailable max_parallel: 1 timeout_seconds: 60 summary_bytes: 1024 raw_output_bytes: 4096 roles: implementation: [cursor, antigravity, pi] repair: [cursor, antigravity, pi] test: [cursor, antigravity, pi] review: [pi] profiles: pi: adapter: pi executable: pi model: auto enabled: true antigravity: adapter: antigravity executable: agy model: auto enabled: true cursor: adapter: cursor executable: cursor-agent model: auto enabled: true """ ) write_repo(self.repo, block) cfg = load_config(self.repo) self.assertEqual(cfg.roles["implementation"], ("cursor", "antigravity", "pi")) self.assertEqual(list(cfg.profiles.keys())[0], "pi") def test_model_auto_is_preserved_as_none(self) -> None: write_repo(self.repo, MINIMAL_BLOCK) cfg = load_config(self.repo) self.assertIsNone(cfg.profiles["cursor"].model) self.assertIsNone(cfg.profiles["pi"].model) def test_rejects_unknown_profile_reference(self) -> None: block = MINIMAL_BLOCK.replace( "implementation: [cursor, antigravity, pi]", "implementation: [cursor, antigravity, pi, missing]", ) write_repo(self.repo, block) with self.assertRaises(ConfigError): load_config(self.repo) def test_rejects_shell_command_and_secret_shaped_values(self) -> None: shell = MINIMAL_BLOCK.replace( "executable: cursor-agent", "executable: \"cursor-agent && rm -rf /\"", ) write_repo(self.repo, shell) with self.assertRaises(ConfigError): load_config(self.repo) secret_key = MINIMAL_BLOCK.replace( "enabled: true\n\n pi:", "enabled: true\n api_token: \"sk-secret-value\"\n\n pi:", ) write_repo(self.repo, secret_key) with self.assertRaises(ConfigError): load_config(self.repo) secret_val = MINIMAL_BLOCK.replace( "executable: pi", "executable: \"Bearer abcdef0123456789abcdef0123456789\"", ) write_repo(self.repo, secret_val) with self.assertRaises(ConfigError): load_config(self.repo) def test_custom_placeholders_must_fill_whole_argv_items(self) -> None: good = MINIMAL_BLOCK.replace("review: [pi]", "review: [mycustom]") + textwrap.dedent( """\ mycustom: adapter: custom executable: mycli model: auto enabled: true custom_adapters: mycustom: base_argv: ["run", "${unit_packet}"] run_everything_argv: ["--yes"] workspace_argv: ["--cwd", "${workspace}"] model_argv: ["--model", "${model}"] prompt_transport: stdin output_format: json """ ) write_repo(self.repo, good) cfg = load_config(self.repo) self.assertEqual( cfg.custom_adapters["mycustom"].workspace_argv, ("--cwd", "${workspace}"), ) bad = good.replace( 'workspace_argv: ["--cwd", "${workspace}"]', 'workspace_argv: ["--cwd=${workspace}"]', ) write_repo(self.repo, bad) with self.assertRaises(ConfigError): load_config(self.repo) unknown_ph = good.replace( 'base_argv: ["run", "${unit_packet}"]', 'base_argv: ["run", "${other}"]', ) write_repo(self.repo, unknown_ph) with self.assertRaises(ConfigError): load_config(self.repo) def test_numeric_limits_are_positive_and_bounded(self) -> None: zero = MINIMAL_BLOCK.replace("max_parallel: 3", "max_parallel: 0") write_repo(self.repo, zero) with self.assertRaises(ConfigError): load_config(self.repo) negative = MINIMAL_BLOCK.replace("timeout_seconds: 7200", "timeout_seconds: -1") write_repo(self.repo, negative) with self.assertRaises(ConfigError): load_config(self.repo) huge = MINIMAL_BLOCK.replace("summary_bytes: 16384", "summary_bytes: 999999999") write_repo(self.repo, huge) with self.assertRaises(ConfigError): load_config(self.repo) inverted = MINIMAL_BLOCK.replace( "summary_bytes: 16384\n raw_output_bytes: 10485760", "summary_bytes: 10485760\n raw_output_bytes: 16384", ) write_repo(self.repo, inverted) with self.assertRaises(ConfigError): load_config(self.repo) def test_fresh_process_per_unit_must_be_true(self) -> None: block = MINIMAL_BLOCK.replace( "fresh_process_per_unit: true", "fresh_process_per_unit: false", ) write_repo(self.repo, block) with self.assertRaises(ConfigError) as ctx: load_config(self.repo) self.assertIn("fresh_process_per_unit", str(ctx.exception)) def test_rejects_indentation_jump(self) -> None: # root mapping child jumps from 0 spaces to 4 spaces (must be exactly +2) jumped = textwrap.dedent( """\ schema: agent-execution/v1 execution: fresh_process_per_unit: true permission: run-everything fallback_to_host: all-unavailable max_parallel: 1 timeout_seconds: 60 summary_bytes: 1024 raw_output_bytes: 4096 roles: implementation: [cursor] repair: [cursor] test: [cursor] review: [cursor] profiles: cursor: adapter: cursor executable: cursor-agent model: auto enabled: true """ ) write_repo(self.repo, jumped) with self.assertRaises(ConfigError) as ctx: load_config(self.repo) self.assertRegex(str(ctx.exception).lower(), r"indent|indentation") def test_quoted_unicode_string_round_trips(self) -> None: block = MINIMAL_BLOCK.replace( "executable: pi", 'executable: "模型-β-测试"', ) write_repo(self.repo, block) cfg = load_config(self.repo) self.assertEqual(cfg.profiles["pi"].executable, "模型-β-测试") escaped = MINIMAL_BLOCK.replace( "executable: pi", r'executable: "say \"hi\" 模型"', ) write_repo(self.repo, escaped) cfg = load_config(self.repo) self.assertEqual(cfg.profiles["pi"].executable, 'say "hi" 模型') def test_idle_timeout_defaults_within_total_and_rejects_bounds(self) -> None: self.assertEqual(DEFAULT_IDLE_TIMEOUT_SECONDS, 600) write_repo(self.repo, MINIMAL_BLOCK) cfg = load_config(self.repo) self.assertEqual(cfg.execution.idle_timeout_seconds, 600) self.assertLessEqual(cfg.execution.idle_timeout_seconds, cfg.execution.timeout_seconds) short = MINIMAL_BLOCK.replace("timeout_seconds: 7200", "timeout_seconds: 60") write_repo(self.repo, short) cfg = load_config(self.repo) self.assertEqual(cfg.execution.timeout_seconds, 60) self.assertEqual(cfg.execution.idle_timeout_seconds, 60) explicit = MINIMAL_BLOCK.replace( "timeout_seconds: 7200\n", "timeout_seconds: 7200\n idle_timeout_seconds: 600\n", ) write_repo(self.repo, explicit) self.assertEqual(load_config(self.repo).execution.idle_timeout_seconds, 600) equal = MINIMAL_BLOCK.replace( "timeout_seconds: 7200\n", "timeout_seconds: 100\n idle_timeout_seconds: 100\n", ) write_repo(self.repo, equal) cfg = load_config(self.repo) self.assertEqual(cfg.execution.idle_timeout_seconds, 100) self.assertEqual(cfg.execution.timeout_seconds, 100) above = MINIMAL_BLOCK.replace( "timeout_seconds: 7200\n", "timeout_seconds: 100\n idle_timeout_seconds: 101\n", ) write_repo(self.repo, above) with self.assertRaises(ConfigError) as ctx: load_config(self.repo) self.assertIn("idle_timeout_seconds", str(ctx.exception)) zero = MINIMAL_BLOCK.replace( "timeout_seconds: 7200\n", "timeout_seconds: 7200\n idle_timeout_seconds: 0\n", ) write_repo(self.repo, zero) with self.assertRaises(ConfigError): load_config(self.repo) negative = MINIMAL_BLOCK.replace( "timeout_seconds: 7200\n", "timeout_seconds: 7200\n idle_timeout_seconds: -5\n", ) write_repo(self.repo, negative) with self.assertRaises(ConfigError): load_config(self.repo) flagged = MINIMAL_BLOCK.replace( "timeout_seconds: 7200\n", "timeout_seconds: 7200\n idle_timeout_seconds: true\n", ) write_repo(self.repo, flagged) with self.assertRaises(ConfigError): load_config(self.repo) fractional = MINIMAL_BLOCK.replace( "timeout_seconds: 7200\n", "timeout_seconds: 7200\n idle_timeout_seconds: 1.5\n", ) write_repo(self.repo, fractional) with self.assertRaises(ConfigError): load_config(self.repo) def test_exposes_max_attempts_per_profile_and_emergency_profile(self) -> None: 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") def test_configured_emergency_profile_must_exist_be_enabled_and_use_adapter_codex( self, ) -> None: write_repo(self.repo, _emergency_block(include_profile=False)) with self.assertRaises(ConfigError) as ctx: load_config(self.repo) self.assertIn("does not exist", str(ctx.exception)) write_repo(self.repo, _emergency_block(enabled="false")) with self.assertRaises(ConfigError) as ctx: load_config(self.repo) self.assertIn("must be enabled", str(ctx.exception)) write_repo(self.repo, _emergency_block(adapter="pi")) with self.assertRaises(ConfigError) as ctx: load_config(self.repo) self.assertIn("adapter codex", str(ctx.exception)) def test_emergency_profile_is_excluded_from_normal_roles(self) -> None: for role, replacement in ( ("implementation", "[cursor, pi, codex]"), ("repair", "[cursor, codex]"), ("test", "[pi, codex]"), ): write_repo(self.repo, _emergency_block(**{role: replacement})) with self.assertRaises(ConfigError) as ctx: load_config(self.repo) self.assertIn("must not appear", str(ctx.exception)) self.assertIn(role, str(ctx.exception)) def test_review_may_include_emergency_profile(self) -> None: write_repo(self.repo, _emergency_block(review="[pi, codex]")) cfg = load_config(self.repo) self.assertEqual(cfg.execution.emergency_profile, "codex") self.assertTrue(cfg.profiles["codex"].enabled) self.assertEqual(cfg.profiles["codex"].adapter, "codex") self.assertEqual(cfg.roles["review"], ("pi", "codex")) self.assertNotIn("codex", cfg.roles["implementation"]) self.assertNotIn("codex", cfg.roles["repair"]) self.assertNotIn("codex", cfg.roles["test"]) def test_accepts_fallback_to_host_never(self) -> None: write_repo(self.repo, _strict_routing_block(fallback="never")) cfg = load_config(self.repo) self.assertEqual(cfg.execution.fallback_to_host, "never") self.assertEqual(cfg.execution.max_attempts_per_profile, 3) self.assertEqual(cfg.execution.emergency_profile, "codex") def test_legacy_all_unavailable_remains_parseable(self) -> None: write_repo(self.repo, _strict_routing_block(fallback="all-unavailable")) cfg = load_config(self.repo) self.assertEqual(cfg.execution.fallback_to_host, "all-unavailable") def test_emergency_profile_requires_max_attempts_exactly_three(self) -> None: for raw_max in ("1", "2", "4", None): with self.subTest(max_attempts=raw_max): write_repo( self.repo, _strict_routing_block(fallback="all-unavailable", max_attempts=raw_max), ) with self.assertRaises(ConfigError) as ctx: load_config(self.repo) self.assertIn("max_attempts_per_profile", str(ctx.exception)) def test_normal_roles_reject_forbidden_adapters_and_alternate_codex_profile( self, ) -> None: claude = textwrap.dedent( """\ claude: adapter: claude executable: claude model: auto enabled: true """ ) claude = "\n".join( f" {line}" if line else line for line in claude.strip("\n").splitlines() ) write_repo( self.repo, _strict_routing_block( fallback="all-unavailable", implementation="[cursor, claude]", extra_profiles=claude, ), ) with self.assertRaises(ConfigError) as ctx: load_config(self.repo) self.assertIn("implementation", str(ctx.exception)) self.assertIn("cursor, antigravity, or pi", str(ctx.exception)) planner = textwrap.dedent( """\ planner: adapter: codex executable: planner model: auto enabled: true """ ) planner = "\n".join( f" {line}" if line else line for line in planner.strip("\n").splitlines() ) write_repo( self.repo, _strict_routing_block( fallback="all-unavailable", implementation="[cursor, planner]", extra_profiles=planner, ), ) with self.assertRaises(ConfigError) as ctx: load_config(self.repo) self.assertIn("planner", str(ctx.exception)) self.assertIn("implementation", str(ctx.exception)) self.assertNotIn("host-fallback", str(ctx.exception).lower()) def test_strict_emergency_mode_rejects_custom_normal_role_adapters(self) -> None: worker = textwrap.dedent( """\ worker: adapter: custom executable: worker model: auto enabled: true """ ) worker = "\n".join( f" {line}" if line else line for line in worker.strip("\n").splitlines() ) adapters = textwrap.dedent( """\ custom_adapters: worker: base_argv: ["${workspace}", "${unit_packet}", "${result_path}"] run_everything_argv: [] workspace_argv: [] model_argv: [] prompt_transport: stdin output_format: text """ ) for role in ("implementation", "repair", "test"): with self.subTest(role=role): roles = { "implementation": "[cursor]", "repair": "[cursor]", "test": "[pi]", } roles[role] = "[worker]" block = _strict_routing_block( implementation=roles["implementation"], repair=roles["repair"], test=roles["test"], extra_profiles=worker, ) write_repo(self.repo, block.rstrip() + "\n\n" + adapters) with self.assertRaises(ConfigError) as ctx: load_config(self.repo) message = str(ctx.exception) self.assertIn("worker", message) 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_without_emergency_custom_normal_roles_are_rejected(self) -> None: write_repo(self.repo, _legacy_custom_block()) with self.assertRaises(ConfigError) as ctx: load_config(self.repo) message = str(ctx.exception) self.assertIn("mycustom", message) self.assertIn("cursor, antigravity, or pi", message) def test_without_emergency_review_may_use_custom_adapter(self) -> None: block = _legacy_custom_block().replace( "implementation: [mycustom]\n" " repair: [mycustom]\n" " test: [mycustom]\n", "implementation: [cursor, antigravity, pi]\n" " repair: [cursor, antigravity, pi]\n" " test: [pi, antigravity, cursor]\n", ).replace( "\ncustom_adapters:", "\n" + _profile_lines("cursor", "cursor", "true") + "\n" + _profile_lines("antigravity", "antigravity", "true") + "\n" + _profile_lines("pi", "pi", "true") + "\n\ncustom_adapters:", ) write_repo(self.repo, block) cfg = load_config(self.repo) self.assertIsNone(cfg.execution.emergency_profile) self.assertEqual(cfg.roles["review"], ("mycustom",)) self.assertEqual(cfg.profiles["mycustom"].adapter, "custom") def test_without_emergency_normal_roles_need_exact_enabled_adapter_set(self) -> None: no_emergency = {"emergency": None, "max_attempts": None, "review": "[pi]"} worker = _profile_lines("worker", "custom", "true") adapters = ( "custom_adapters:\n" " worker:\n" ' base_argv: ["${workspace}", "${unit_packet}", "${result_path}"]\n' " run_everything_argv: []\n" " workspace_argv: []\n" " model_argv: []\n" " prompt_transport: stdin\n" " output_format: text\n" ) claude = _profile_lines("claude", "claude", "true") for role in ("implementation", "repair", "test"): cases = ( ("missing", "[cursor, pi]", "", "exactly one enabled"), ( "disabled-only", "[cursor, agy_off, pi]", _profile_lines("agy_off", "antigravity", "false"), "exactly one enabled", ), ( "duplicate-enabled", "[cursor, antigravity, pi, cursor2]", _profile_lines("cursor2", "cursor", "true"), "exactly one enabled", ), ( "custom-extra", "[cursor, antigravity, pi, worker]", worker, "cursor, antigravity, or pi", ), ( "custom-only", "[worker]", worker, "cursor, antigravity, or pi", ), ( "other-builtin", "[cursor, antigravity, pi, claude]", claude, "cursor, antigravity, or pi", ), ) for label, replacement, extra, expected in cases: with self.subTest(role=role, case=label): block = _strict_routing_block( **no_emergency, **{role: replacement}, extra_profiles=extra ) if extra is worker: block = block.rstrip() + "\n\n" + adapters write_repo(self.repo, block) with self.assertRaises(ConfigError) as ctx: load_config(self.repo) message = str(ctx.exception) self.assertIn(role, message) self.assertIn(expected, message) def test_without_emergency_disabled_extras_are_ignored_and_order_kept(self) -> None: write_repo( self.repo, _strict_routing_block( emergency=None, max_attempts=None, review="[pi]", implementation="[pi, cursor_off, antigravity, cursor]", test="[antigravity, cursor, pi, cursor_off]", extra_profiles=_profile_lines("cursor_off", "cursor", "false"), ), ) cfg = load_config(self.repo) self.assertIsNone(cfg.execution.emergency_profile) self.assertEqual( cfg.roles["implementation"], ("pi", "cursor_off", "antigravity", "cursor") ) self.assertEqual(cfg.roles["repair"], ("cursor", "antigravity", "pi")) self.assertEqual(cfg.roles["test"], ("antigravity", "cursor", "pi", "cursor_off")) def test_max_attempts_per_profile_never_exceeds_three(self) -> None: for raw_max in ("1", "2", "3"): with self.subTest(max_attempts=raw_max): write_repo( self.repo, _strict_routing_block( emergency=None, max_attempts=raw_max, review="[pi]" ), ) cfg = load_config(self.repo) self.assertEqual(cfg.execution.max_attempts_per_profile, int(raw_max)) for raw_max in ("4", "10"): for emergency in (None, "codex"): with self.subTest(max_attempts=raw_max, emergency=emergency): write_repo( self.repo, _strict_routing_block(emergency=emergency, max_attempts=raw_max), ) with self.assertRaises(ConfigError) as ctx: load_config(self.repo) self.assertIn("max_attempts_per_profile", str(ctx.exception)) def test_checked_in_project_config_never_authorizes_host_fallback(self) -> None: repo = Path(__file__).resolve().parents[1] cfg = load_config(repo) self.assertEqual(cfg.execution.fallback_to_host, "never") self.assertEqual(cfg.execution.max_attempts_per_profile, 3) self.assertEqual(cfg.execution.emergency_profile, "codex") allowed = {"cursor", "antigravity", "pi"} for role in ("implementation", "repair", "test"): self.assertNotIn("codex", cfg.roles[role]) for name in cfg.roles[role]: self.assertIn(cfg.profiles[name].adapter, allowed) self.assertIn("codex", cfg.roles["review"]) self.assertTrue(cfg.profiles["codex"].enabled) self.assertEqual(cfg.profiles["codex"].adapter, "codex") def _legacy_custom_block() -> str: """No emergency_profile and adapter custom on every role.""" return textwrap.dedent( """\ schema: agent-execution/v1 execution: fresh_process_per_unit: true permission: run-everything fallback_to_host: all-unavailable max_parallel: 1 timeout_seconds: 60 summary_bytes: 1024 raw_output_bytes: 4096 roles: implementation: [mycustom] repair: [mycustom] test: [mycustom] review: [mycustom] profiles: mycustom: adapter: custom executable: mycli model: auto enabled: true custom_adapters: mycustom: base_argv: ["run", "${unit_packet}"] run_everything_argv: ["--yes"] workspace_argv: ["--cwd", "${workspace}"] model_argv: ["--model", "${model}"] prompt_transport: stdin output_format: json """ ) def _strict_routing_block( *, fallback: str = "never", max_attempts: str | None = "3", emergency: str | None = "codex", implementation: str = "[cursor, antigravity, pi]", repair: str = "[cursor, antigravity, pi]", test: str = "[pi, antigravity, cursor]", review: str = "[pi, codex]", extra_profiles: str = "", ) -> str: lines = [ "schema: agent-execution/v1", "", "execution:", " fresh_process_per_unit: true", " permission: run-everything", f" fallback_to_host: {fallback}", " max_parallel: 1", " timeout_seconds: 60", " summary_bytes: 1024", " raw_output_bytes: 4096", ] if max_attempts is not None: lines.append(f" max_attempts_per_profile: {max_attempts}") if emergency is not None: lines.append(f" emergency_profile: {emergency}") lines.extend( [ "", "roles:", f" implementation: {implementation}", f" repair: {repair}", f" test: {test}", f" review: {review}", "", "profiles:", " cursor:", " adapter: cursor", " executable: cursor-agent", " model: auto", " enabled: true", "", " antigravity:", " adapter: antigravity", " executable: agy", " model: auto", " enabled: true", "", " pi:", " adapter: pi", " executable: pi", " model: auto", " enabled: true", "", " codex:", " adapter: codex", " executable: codex", " model: auto", " enabled: true", ] ) if extra_profiles: lines.append(extra_profiles.rstrip("\n")) 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, antigravity, pi]", repair: str = "[cursor, antigravity, pi]", test: str = "[pi, antigravity, cursor]", review: str = "[pi]", ) -> str: block = MINIMAL_BLOCK.replace( " raw_output_bytes: 10485760\n", " raw_output_bytes: 10485760\n" " max_attempts_per_profile: 3\n" " emergency_profile: codex\n", ) block = block.replace( "implementation: [cursor, antigravity, pi]", f"implementation: {implementation}" ) block = block.replace("repair: [cursor, antigravity, pi]", f"repair: {repair}") block = block.replace("test: [pi, antigravity, cursor]", f"test: {test}") block = block.replace("review: [pi]", f"review: {review}") if not include_profile: return block return block.replace( " enabled: true\n", " enabled: true\n" "\n" " codex:\n" f" adapter: {adapter}\n" " executable: codex\n" " model: auto\n" f" enabled: {enabled}\n", 1, ) if __name__ == "__main__": unittest.main()