Coverage for .github / scripts / autopilot_ci_contract.py: 100.00%
219 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-09-14 22:07 +0000
1"""Single source for the Autopilot facts and assertions CI jobs share.
3Unit, dev-container, and boot jobs use this module instead of duplicating
4expected models, pins, generated-config shapes, or companion-server lists.
5Every fact is derived from production modules, so no release literal lives
6here to update separately.
8Print subcommands accept ``--engine`` (default: ``claude-code``):
10 pin selected engine's pinned CLI version
11 install-command exact lazy npm install command
12 default-model shipped Bedrock default model id
13 expected-servers expected MCP server names, one per line
15Verification subcommands:
17 verify-config PATH validate Claude's --print-config JSON
18 verify-codex-config PATH validate Codex's --print-config TOML
19 [--no-companions] expect only the gco server
20 [--expect-gco-env K=V] require a gco-only env pair (repeatable)
21 [--gco-args ARG] require exact gco args (repeatable, ordered)
22 verify-plan PATH validate a -o json --dry-run plan
23 [--engine ENGINE]
24 [--claude-binary present|absent]
25 [--codex-binary present|absent]
27Importable for pytest: ``expected_servers()``, ``verify_config()``,
28``verify_codex_config()``, ``verify_plan()``, ``main()``.
29"""
31from __future__ import annotations
33import argparse
34import json
35import sys
36import tomllib
37from pathlib import Path
38from typing import Any
40_REPO_ROOT = Path(__file__).resolve().parents[2]
41if str(_REPO_ROOT) not in sys.path:
42 sys.path.insert(0, str(_REPO_ROOT))
44from cli.autopilot import ( # noqa: E402
45 CLAUDE_CODE_PACKAGE,
46 CLAUDE_CODE_VERSION,
47 CODEX_BEDROCK_PROVIDER,
48 CODEX_MCP_STARTUP_TIMEOUT_SECONDS,
49 CODEX_PACKAGE,
50 CODEX_VERSION,
51 COMPANION_MCP_SERVERS,
52 AutopilotEngine,
53 claude_install_command,
54 codex_install_command,
55)
56from gco.bedrock import ( # noqa: E402
57 get_default_claude_code_model_id,
58 get_default_codex_model_id,
59 get_default_codex_reasoning_effort,
60)
62#: Companions deliberately pruned from the curated registry; their names
63#: must never reappear anywhere in a generated session config.
64PRUNED_PACKAGES: tuple[str, ...] = ("mcp-server-fetch", "mcp-server-calculator")
65_ENGINE_CHOICES = tuple(engine.value for engine in AutopilotEngine)
68def expected_servers(include_companions: bool = True) -> list[str]:
69 """Return the exact MCP server set a generated session config must carry."""
70 names = {"gco"}
71 if include_companions:
72 names |= {companion.name for companion in COMPANION_MCP_SERVERS}
73 return sorted(names)
76def _verify_server_mapping(
77 servers: Any,
78 *,
79 include_companions: bool,
80 expect_gco_env: dict[str, str] | None,
81 gco_args: list[str] | None,
82) -> list[str]:
83 if not isinstance(servers, dict):
84 return ["config carries no MCP server mapping"]
86 problems: list[str] = []
87 expected = expected_servers(include_companions)
88 actual = sorted(servers)
89 if actual != expected:
90 missing = sorted(set(expected) - set(actual))
91 extra = sorted(set(actual) - set(expected))
92 problems.append(f"server set mismatch: missing={missing} unexpected={extra}")
94 for name, entry in sorted(servers.items()):
95 if not isinstance(entry, dict):
96 problems.append(f"{name}: entry must be a mapping, got {entry!r}")
97 continue
98 command = entry.get("command")
99 if not isinstance(command, str) or not command:
100 problems.append(f"{name}: command must be a non-empty string, got {command!r}")
101 args = entry.get("args")
102 if not isinstance(args, list) or not all(isinstance(arg, str) for arg in args):
103 problems.append(f"{name}: args must be a list of strings, got {args!r}")
105 text = json.dumps(servers)
106 for pruned in PRUNED_PACKAGES:
107 if pruned in text:
108 problems.append(f"pruned package {pruned!r} reappeared in the generated config")
110 gco_entry = servers.get("gco", {})
111 if not isinstance(gco_entry, dict):
112 gco_entry = {}
113 if expect_gco_env is not None:
114 gco_env = gco_entry.get("env", {})
115 if not isinstance(gco_env, dict):
116 gco_env = {}
117 for key, value in sorted(expect_gco_env.items()):
118 if gco_env.get(key) != value:
119 problems.append(f"gco env {key!r}: expected {value!r}, got {gco_env.get(key)!r}")
120 for name, entry in sorted(servers.items()):
121 if name == "gco" or not isinstance(entry, dict):
122 continue
123 environment = entry.get("env", {})
124 leaked = set(expect_gco_env) & set(environment if isinstance(environment, dict) else {})
125 if leaked:
126 problems.append(f"gco-only env leaked onto {name}: {sorted(leaked)}")
127 if gco_args is not None and gco_entry.get("args") != gco_args:
128 problems.append(f"gco args: expected {gco_args!r}, got {gco_entry.get('args')!r}")
130 return problems
133def verify_config(
134 config: dict[str, Any],
135 include_companions: bool = True,
136 expect_gco_env: dict[str, str] | None = None,
137 gco_args: list[str] | None = None,
138) -> list[str]:
139 """Return every problem with Claude's generated MCP JSON (empty = valid)."""
140 servers = config.get("mcpServers")
141 if not isinstance(servers, dict):
142 return ["config carries no mcpServers mapping"]
143 return _verify_server_mapping(
144 servers,
145 include_companions=include_companions,
146 expect_gco_env=expect_gco_env,
147 gco_args=gco_args,
148 )
151def verify_codex_config(
152 config: dict[str, Any],
153 include_companions: bool = True,
154 expect_gco_env: dict[str, str] | None = None,
155 gco_args: list[str] | None = None,
156 expected_region: str | None = None,
157) -> list[str]:
158 """Return every problem with Codex's generated TOML config (empty = valid)."""
159 problems: list[str] = []
160 expected_model = get_default_codex_model_id()
161 if config.get("model") != expected_model:
162 problems.append(
163 f"Codex model {config.get('model')!r} != shipped default {expected_model!r}"
164 )
165 if config.get("model_provider") != CODEX_BEDROCK_PROVIDER:
166 problems.append(
167 f"Codex provider {config.get('model_provider')!r} != {CODEX_BEDROCK_PROVIDER!r}"
168 )
169 expected_reasoning = get_default_codex_reasoning_effort()
170 if config.get("model_reasoning_effort") != expected_reasoning:
171 problems.append(
172 "Codex reasoning effort "
173 f"{config.get('model_reasoning_effort')!r} != {expected_reasoning!r}"
174 )
175 if config.get("check_for_update_on_startup") is not False:
176 problems.append("Codex update checks must be disabled in the generated config")
178 providers = config.get("model_providers")
179 provider = providers.get(CODEX_BEDROCK_PROVIDER) if isinstance(providers, dict) else None
180 if not isinstance(provider, dict) or provider.get("wire_api") != "responses":
181 actual_wire_api = provider.get("wire_api") if isinstance(provider, dict) else None
182 problems.append(f"Codex wire API {actual_wire_api!r} != 'responses'")
183 aws = provider.get("aws") if isinstance(provider, dict) else None
184 if not isinstance(aws, dict):
185 problems.append(f"Codex config carries no {CODEX_BEDROCK_PROVIDER}.aws provider table")
186 else:
187 region = aws.get("region")
188 if not isinstance(region, str) or not region:
189 problems.append(f"Codex provider region must be non-empty, got {region!r}")
190 elif expected_region is not None and region != expected_region:
191 problems.append(f"Codex provider region {region!r} != expected {expected_region!r}")
193 servers = config.get("mcp_servers")
194 problems.extend(
195 _verify_server_mapping(
196 servers,
197 include_companions=include_companions,
198 expect_gco_env=expect_gco_env,
199 gco_args=gco_args,
200 )
201 )
202 if isinstance(servers, dict):
203 for name, entry in sorted(servers.items()):
204 if isinstance(entry, dict) and entry.get("enabled") is not True:
205 problems.append(f"{name}: Codex MCP server must be enabled")
206 if (
207 isinstance(entry, dict)
208 and entry.get("startup_timeout_sec") != CODEX_MCP_STARTUP_TIMEOUT_SECONDS
209 ):
210 problems.append(
211 f"{name}: Codex MCP startup timeout must be "
212 f"{CODEX_MCP_STARTUP_TIMEOUT_SECONDS} seconds"
213 )
214 return problems
217def _engine_facts(engine: AutopilotEngine) -> tuple[str, str, list[str], str | None]:
218 if engine is AutopilotEngine.CODEX:
219 return (
220 CODEX_VERSION,
221 f"{CODEX_PACKAGE}@{CODEX_VERSION}",
222 codex_install_command(),
223 get_default_codex_reasoning_effort(),
224 )
225 return (
226 CLAUDE_CODE_VERSION,
227 f"{CLAUDE_CODE_PACKAGE}@{CLAUDE_CODE_VERSION}",
228 claude_install_command(),
229 None,
230 )
233def _default_model(engine: AutopilotEngine) -> str:
234 if engine is AutopilotEngine.CODEX:
235 return get_default_codex_model_id()
236 return get_default_claude_code_model_id()
239def verify_plan(
240 plan: dict[str, Any],
241 claude_binary: str | None = None,
242 *,
243 engine: str | AutopilotEngine = AutopilotEngine.CLAUDE_CODE,
244 codex_binary: str | None = None,
245) -> list[str]:
246 """Return every problem with a JSON dry-run plan (empty = valid)."""
247 resolved_engine = AutopilotEngine(engine)
248 problems: list[str] = []
249 _version, expected_pin, install_command, expected_reasoning = _engine_facts(resolved_engine)
250 expected_model = _default_model(resolved_engine)
252 if plan.get("engine") != resolved_engine.value:
253 problems.append(f"plan engine {plan.get('engine')!r} != {resolved_engine.value!r}")
254 if plan.get("model") != expected_model:
255 problems.append(f"plan model {plan.get('model')!r} != shipped default {expected_model!r}")
256 if plan.get("reasoning_effort") != expected_reasoning:
257 problems.append(
258 f"plan reasoning {plan.get('reasoning_effort')!r} != {expected_reasoning!r}"
259 )
260 if sorted(plan.get("mcp_servers", [])) != expected_servers():
261 problems.append(
262 f"plan servers {sorted(plan.get('mcp_servers', []))} != {expected_servers()}"
263 )
264 if plan.get("engine_pin") != expected_pin:
265 problems.append(f"plan engine pin {plan.get('engine_pin')!r} != {expected_pin!r}")
266 if plan.get("install_command") != " ".join(install_command):
267 problems.append(
268 "plan install command does not match the selected engine's production command"
269 )
271 binary_state = codex_binary if resolved_engine is AutopilotEngine.CODEX else claude_binary
272 binary = plan.get("engine_binary")
273 if binary_state == "absent" and binary is not None:
274 problems.append(f"expected no {resolved_engine.value} binary, plan found {binary!r}")
275 if binary_state == "present" and not binary:
276 problems.append(f"expected an installed {resolved_engine.value} binary, plan detected none")
278 selected_pin_field = (
279 "codex_pin" if resolved_engine is AutopilotEngine.CODEX else "claude_code_pin"
280 )
281 selected_binary_field = (
282 "codex_binary" if resolved_engine is AutopilotEngine.CODEX else "claude_binary"
283 )
284 other_pin_field = "claude_code_pin" if resolved_engine is AutopilotEngine.CODEX else "codex_pin"
285 other_binary_field = (
286 "claude_binary" if resolved_engine is AutopilotEngine.CODEX else "codex_binary"
287 )
288 if plan.get(selected_pin_field) != expected_pin:
289 problems.append(
290 f"plan {selected_pin_field} {plan.get(selected_pin_field)!r} != {expected_pin!r}"
291 )
292 if plan.get(selected_binary_field) != binary:
293 problems.append(f"plan {selected_binary_field} disagrees with engine_binary")
294 if plan.get(other_pin_field) is not None or plan.get(other_binary_field) is not None:
295 problems.append(
296 f"plan leaks selected-engine state into {other_pin_field}/{other_binary_field}"
297 )
299 if resolved_engine is AutopilotEngine.CODEX:
300 # The public JSON formatter intentionally omits the large generated
301 # config; when an in-process caller supplies it, validate it too.
302 rendered = plan.get("codex_config")
303 if rendered is not None:
304 if not isinstance(rendered, str):
305 problems.append("Codex plan config must be TOML text when present")
306 else:
307 try:
308 codex_config = tomllib.loads(rendered)
309 except tomllib.TOMLDecodeError as exc:
310 problems.append(f"Codex plan config is invalid TOML: {exc}")
311 else:
312 problems.extend(
313 verify_codex_config(codex_config, expected_region=plan.get("region"))
314 )
315 elif plan.get("codex_config") is not None:
316 problems.append("Claude plan unexpectedly carries a Codex config")
318 return problems
321def _load_json(path: str) -> dict[str, Any]:
322 with open(path, encoding="utf-8") as handle:
323 payload: dict[str, Any] = json.load(handle)
324 return payload
327def _load_toml(path: str) -> dict[str, Any]:
328 with open(path, "rb") as handle:
329 return tomllib.load(handle)
332def _parse_env_pair(pair: str) -> tuple[str, str]:
333 key, separator, value = pair.partition("=")
334 if not separator or not key:
335 raise argparse.ArgumentTypeError(f"--expect-gco-env expects KEY=VALUE, got {pair!r}")
336 return key, value
339def _add_engine_argument(parser: argparse.ArgumentParser) -> None:
340 parser.add_argument(
341 "--engine", choices=_ENGINE_CHOICES, default=AutopilotEngine.CLAUDE_CODE.value
342 )
345def _add_config_arguments(parser: argparse.ArgumentParser) -> None:
346 parser.add_argument("path")
347 parser.add_argument("--no-companions", action="store_true")
348 parser.add_argument("--expect-gco-env", action="append", type=_parse_env_pair, default=None)
349 parser.add_argument("--gco-args", action="append", default=None)
352def main(argv: list[str] | None = None) -> int:
353 parser = argparse.ArgumentParser(description=__doc__)
354 sub = parser.add_subparsers(dest="command", required=True)
356 for name in ("pin", "install-command", "default-model"):
357 _add_engine_argument(sub.add_parser(name))
358 sub.add_parser("expected-servers")
360 config_parser = sub.add_parser("verify-config")
361 _add_config_arguments(config_parser)
363 codex_config_parser = sub.add_parser("verify-codex-config")
364 _add_config_arguments(codex_config_parser)
365 codex_config_parser.add_argument("--region", default=None)
367 plan_parser = sub.add_parser("verify-plan")
368 plan_parser.add_argument("path")
369 _add_engine_argument(plan_parser)
370 plan_parser.add_argument("--claude-binary", choices=("present", "absent"), default=None)
371 plan_parser.add_argument("--codex-binary", choices=("present", "absent"), default=None)
373 args = parser.parse_args(argv)
375 if args.command in {"pin", "install-command", "default-model"}:
376 engine = AutopilotEngine(args.engine)
377 version, _pin, install_command, _reasoning = _engine_facts(engine)
378 if args.command == "pin":
379 print(version)
380 elif args.command == "install-command":
381 print(" ".join(install_command))
382 else:
383 print(_default_model(engine))
384 return 0
385 if args.command == "expected-servers":
386 print("\n".join(expected_servers()))
387 return 0
389 expect_gco_env = dict(args.expect_gco_env) if getattr(args, "expect_gco_env", None) else None
390 if args.command == "verify-config":
391 problems = verify_config(
392 _load_json(args.path),
393 include_companions=not args.no_companions,
394 expect_gco_env=expect_gco_env,
395 gco_args=args.gco_args,
396 )
397 label = "Claude config"
398 elif args.command == "verify-codex-config":
399 problems = verify_codex_config(
400 _load_toml(args.path),
401 include_companions=not args.no_companions,
402 expect_gco_env=expect_gco_env,
403 gco_args=args.gco_args,
404 expected_region=args.region,
405 )
406 label = "Codex config"
407 else:
408 problems = verify_plan(
409 _load_json(args.path),
410 engine=args.engine,
411 claude_binary=args.claude_binary,
412 codex_binary=args.codex_binary,
413 )
414 label = f"{args.engine} plan"
416 if problems:
417 for problem in problems:
418 print(f"ERROR: {problem}", file=sys.stderr)
419 return 1
420 print(f"autopilot {label} OK ({args.path})")
421 return 0
424if __name__ == "__main__":
425 raise SystemExit(main())