"""Markdown extraction and constrained YAML parsing for AGENT-EXECUTION.md.""" from __future__ import annotations import ast import re from collections.abc import Mapping from dataclasses import dataclass from pathlib import Path from typing import Any SCHEMA_V1 = "agent-execution/v1" CONFIG_NAME = "AGENT-EXECUTION.md" BLOCK_OPEN = "```agent-execution" BLOCK_CLOSE = "```" BUILTIN_ADAPTERS = frozenset( {"cursor", "antigravity", "claude", "codex", "pi", "kimi", "opencode", "custom"} ) REQUIRED_ROLES = ("implementation", "repair", "test", "review") ALLOWED_PLACEHOLDERS = frozenset( {"${workspace}", "${unit_packet}", "${result_path}", "${model}"} ) PLACEHOLDER_RE = re.compile(r"\$\{[A-Za-z0-9_]+\}") PERMISSION_VALUES = frozenset({"run-everything"}) FALLBACK_VALUES = frozenset({"all-unavailable", "never"}) # Implementation/repair/test roles each need exactly one enabled profile per # adapter here, with or without emergency_profile; review is unrestricted. NORMAL_ROLE_ADAPTERS = frozenset({"cursor", "antigravity", "pi"}) MAX_ATTEMPTS_PER_PROFILE = 3 PROMPT_TRANSPORTS = frozenset({"stdin", "positional", "flag"}) OUTPUT_FORMATS = frozenset({"json", "stream-json", "jsonl", "text"}) MAX_PARALLEL = 32 MAX_TIMEOUT_SECONDS = 86400 MAX_SUMMARY_BYTES = 1_048_576 MAX_RAW_OUTPUT_BYTES = 104_857_600 # Omitted idle_timeout_seconds uses this cap, then clamps to timeout_seconds. DEFAULT_IDLE_TIMEOUT_SECONDS = 600 SHELL_META_RE = re.compile(r"[;&|`$<>(){}]|&&|\|\||\$\(") SECRET_KEY_RE = re.compile( r"(password|secret|token|api[_-]?key|authorization|credential|passwd)", re.IGNORECASE, ) SECRET_VALUE_RE = re.compile( r"(?i)^(bearer\s+\S+|sk-[A-Za-z0-9_\-]{8,}|[A-Fa-f0-9]{32,})$" ) ROOT_KEYS = frozenset({"schema", "execution", "roles", "profiles", "custom_adapters"}) EXECUTION_KEYS = frozenset( { "fresh_process_per_unit", "permission", "fallback_to_host", "max_parallel", "timeout_seconds", "idle_timeout_seconds", "summary_bytes", "raw_output_bytes", "max_attempts_per_profile", "emergency_profile", } ) EXECUTION_REQUIRED = EXECUTION_KEYS - frozenset( {"idle_timeout_seconds", "max_attempts_per_profile", "emergency_profile"} ) PROFILE_KEYS = frozenset({"adapter", "executable", "model", "enabled"}) CUSTOM_ADAPTER_KEYS = frozenset( { "base_argv", "run_everything_argv", "workspace_argv", "model_argv", "prompt_transport", "output_format", } ) class ConfigError(ValueError): """Raised when AGENT-EXECUTION.md cannot be parsed or fails schema checks.""" @dataclass(frozen=True) class ExecutionPolicy: fresh_process_per_unit: bool permission: str fallback_to_host: str max_parallel: int timeout_seconds: int summary_bytes: int raw_output_bytes: int idle_timeout_seconds: int = DEFAULT_IDLE_TIMEOUT_SECONDS max_attempts_per_profile: int = 1 emergency_profile: str | None = None @dataclass(frozen=True) class Profile: name: str adapter: str executable: str model: str | None enabled: bool @dataclass(frozen=True) class CustomAdapter: name: str base_argv: tuple[str, ...] run_everything_argv: tuple[str, ...] workspace_argv: tuple[str, ...] model_argv: tuple[str, ...] prompt_transport: str output_format: str @dataclass(frozen=True) class ExecutionConfig: schema: str execution: ExecutionPolicy roles: Mapping[str, tuple[str, ...]] profiles: Mapping[str, Profile] custom_adapters: Mapping[str, CustomAdapter] def load_config(repo: Path) -> ExecutionConfig: """Load and validate the single agent-execution block from repo/AGENT-EXECUTION.md.""" path = Path(repo) / CONFIG_NAME if not path.is_file(): raise ConfigError(f"missing {CONFIG_NAME}") text = path.read_text(encoding="utf-8") raw = _parse_subset_yaml(_extract_block(text)) return _build_config(raw) def validate_config(repo: Path) -> list[str]: """Return an empty list when configuration is valid; raise ConfigError otherwise.""" load_config(repo) return [] def _extract_block(text: str) -> str: lines = text.splitlines() starts: list[int] = [] for i, line in enumerate(lines): if line.strip() == BLOCK_OPEN: starts.append(i) if not starts: raise ConfigError("missing agent-execution fenced block") if len(starts) > 1: raise ConfigError("duplicate agent-execution fenced blocks") start = starts[0] end = None for j in range(start + 1, len(lines)): if lines[j].strip() == BLOCK_CLOSE: end = j break if end is None: raise ConfigError("unclosed agent-execution fenced block") return "\n".join(lines[start + 1 : end]) def _parse_subset_yaml(text: str) -> dict[str, Any]: if "\t" in text: raise ConfigError("tabs are not allowed; use two-space indentation") lines = text.splitlines() root: dict[str, Any] = {} stack: list[tuple[int, dict[str, Any]]] = [(-2, root)] i = 0 while i < len(lines): raw = lines[i] if not raw.strip() or raw.lstrip().startswith("#"): i += 1 continue indent = len(raw) - len(raw.lstrip(" ")) stripped = raw.lstrip(" ") if stripped.startswith(("- ", "?", "|", ">", "&", "*", "!")): raise ConfigError(f"unsupported YAML construct: {raw.strip()!r}") if re.search(r"(^|\s)<<", stripped): raise ConfigError(f"unsupported YAML merge/alias: {raw.strip()!r}") content = raw[indent:] if ":" not in content: raise ConfigError(f"expected key: value, got {raw.strip()!r}") key, _, rest = content.partition(":") key = key.strip() rest = rest.strip() if not key or not re.fullmatch(r"[A-Za-z_][A-Za-z0-9_]*", key): raise ConfigError(f"invalid mapping key: {key!r}") _reject_secret_key(key) # Pop to parent whose indent is strictly less than current. while len(stack) > 1 and stack[-1][0] >= indent: stack.pop() parent_indent, parent = stack[-1] expected = parent_indent + 2 if indent != expected: raise ConfigError( f"indentation error: expected {expected} spaces, got {indent}" ) if key in parent: raise ConfigError(f"duplicate mapping key: {key!r}") if rest == "": child: dict[str, Any] = {} parent[key] = child stack.append((indent, child)) i += 1 continue value = _parse_scalar(rest, key=key) parent[key] = value i += 1 return root def _parse_scalar(text: str, *, key: str) -> Any: if text.startswith(("|", ">")): raise ConfigError("multiline scalars are not supported") if text.startswith("["): return _parse_flow_sequence(text) if text in ("true", "false"): return text == "true" if text in ("null", "~"): return None if re.fullmatch(r"-?\d+", text): return int(text) if re.fullmatch(r"\d{4}-\d{2}-\d{2}.*", text): raise ConfigError(f"implicit dates are not supported: {text!r}") if text[:1] in ("'", '"'): value = _parse_quoted_string(text) else: if any(ch in text for ch in ":#[]{},"): raise ConfigError(f"unsafe unquoted scalar: {text!r}") value = text _reject_shell_or_secret_value(key, value) return value def _parse_quoted_string(text: str) -> str: quote = text[0] if quote not in ("'", '"'): raise ConfigError(f"invalid quoted string: {text!r}") if quote == "'": # Single-quoted YAML-ish: only '' escape; keep non-ASCII intact. if not text.endswith("'") or len(text) < 2: raise ConfigError(f"unclosed single-quoted string: {text!r}") inner = text[1:-1] return inner.replace("''", "'") # Double-quoted: exact Python/JSON-compatible escapes via ast.literal_eval. try: value = ast.literal_eval(text) except (SyntaxError, ValueError) as exc: raise ConfigError(f"invalid double-quoted string: {text!r}") from exc if not isinstance(value, str): raise ConfigError(f"quoted scalar must be a string: {text!r}") return value def _parse_flow_sequence(text: str) -> list[Any]: if not text.startswith("[") or not text.endswith("]"): raise ConfigError(f"invalid flow sequence: {text!r}") inner = text[1:-1].strip() if not inner: return [] items: list[Any] = [] buf: list[str] = [] in_quote: str | None = None escape = False i = 0 while i < len(inner): ch = inner[i] if in_quote: buf.append(ch) if escape: escape = False elif ch == "\\" and in_quote == '"': escape = True elif ch == in_quote: in_quote = None i += 1 continue if ch in ("'", '"'): in_quote = ch buf.append(ch) i += 1 continue if ch == ",": items.append(_parse_scalar("".join(buf).strip(), key="list_item")) buf = [] i += 1 continue buf.append(ch) i += 1 if in_quote: raise ConfigError(f"unclosed quote in flow sequence: {text!r}") if buf and "".join(buf).strip(): items.append(_parse_scalar("".join(buf).strip(), key="list_item")) return items def _reject_secret_key(key: str) -> None: if SECRET_KEY_RE.search(key): raise ConfigError(f"credential-looking key rejected: {key!r}") def _reject_shell_or_secret_value(key: str, value: Any) -> None: if not isinstance(value, str): return # Whole-item placeholders are validated later for argv contexts. if value in ALLOWED_PLACEHOLDERS: return if SHELL_META_RE.search(value): raise ConfigError(f"shell metacharacter template rejected for {key!r}") if SECRET_VALUE_RE.match(value.strip()): raise ConfigError(f"secret-shaped value rejected for {key!r}") def _build_config(raw: dict[str, Any]) -> ExecutionConfig: if not isinstance(raw, dict): raise ConfigError("root must be a mapping") unknown = set(raw) - ROOT_KEYS if unknown: raise ConfigError(f"unknown top-level keys: {sorted(unknown)}") for required in ("schema", "execution", "roles", "profiles"): if required not in raw: raise ConfigError(f"missing required key: {required}") schema = raw["schema"] if schema != SCHEMA_V1: raise ConfigError(f"unknown schema: {schema!r}") execution = _build_execution(raw["execution"]) profiles = _build_profiles(raw["profiles"]) custom_adapters = _build_custom_adapters(raw.get("custom_adapters", {})) roles = _build_roles(raw["roles"], profiles) for profile in profiles.values(): if profile.adapter == "custom": if profile.name not in custom_adapters: raise ConfigError( f"profile {profile.name!r} uses adapter custom without " f"custom_adapters.{profile.name}" ) elif profile.adapter not in BUILTIN_ADAPTERS - {"custom"}: if profile.adapter not in custom_adapters: raise ConfigError(f"unknown adapter: {profile.adapter!r}") _validate_emergency_profile(execution, profiles, roles) _validate_normal_role_adapters(profiles, roles) _validate_normal_role_exactness(profiles, roles) return ExecutionConfig( schema=schema, execution=execution, roles=roles, profiles=profiles, custom_adapters=custom_adapters, ) def _build_execution(raw: Any) -> ExecutionPolicy: if not isinstance(raw, dict): raise ConfigError("execution must be a mapping") unknown = set(raw) - EXECUTION_KEYS if unknown: raise ConfigError(f"unknown execution keys: {sorted(unknown)}") missing = EXECUTION_REQUIRED - set(raw) if missing: raise ConfigError(f"missing execution keys: {sorted(missing)}") fresh = raw["fresh_process_per_unit"] if fresh is not True: raise ConfigError("fresh_process_per_unit must be true") permission = raw["permission"] if permission not in PERMISSION_VALUES: raise ConfigError(f"unsupported permission: {permission!r}") fallback = raw["fallback_to_host"] if fallback not in FALLBACK_VALUES: raise ConfigError(f"unsupported fallback_to_host: {fallback!r}") max_parallel = _positive_bounded( raw["max_parallel"], "max_parallel", 1, MAX_PARALLEL ) timeout_seconds = _positive_bounded( raw["timeout_seconds"], "timeout_seconds", 1, MAX_TIMEOUT_SECONDS ) if "idle_timeout_seconds" in raw: idle_timeout_seconds = _positive_bounded( raw["idle_timeout_seconds"], "idle_timeout_seconds", 1, timeout_seconds, ) else: idle_timeout_seconds = min(DEFAULT_IDLE_TIMEOUT_SECONDS, timeout_seconds) summary_bytes = _positive_bounded( raw["summary_bytes"], "summary_bytes", 1, MAX_SUMMARY_BYTES ) raw_output_bytes = _positive_bounded( raw["raw_output_bytes"], "raw_output_bytes", 1, MAX_RAW_OUTPUT_BYTES ) if raw_output_bytes < summary_bytes: raise ConfigError("raw_output_bytes must be >= summary_bytes") if "max_attempts_per_profile" in raw: max_attempts_per_profile = _positive_bounded( raw["max_attempts_per_profile"], "max_attempts_per_profile", 1, MAX_ATTEMPTS_PER_PROFILE, ) else: max_attempts_per_profile = 1 if "emergency_profile" in raw: emergency_profile = raw["emergency_profile"] if not isinstance(emergency_profile, str) or not emergency_profile: raise ConfigError("emergency_profile must be a non-empty string") else: emergency_profile = None return ExecutionPolicy( fresh_process_per_unit=True, permission=permission, fallback_to_host=fallback, max_parallel=max_parallel, timeout_seconds=timeout_seconds, summary_bytes=summary_bytes, raw_output_bytes=raw_output_bytes, idle_timeout_seconds=idle_timeout_seconds, max_attempts_per_profile=max_attempts_per_profile, emergency_profile=emergency_profile, ) def _validate_emergency_profile( execution: ExecutionPolicy, profiles: Mapping[str, Profile], roles: Mapping[str, tuple[str, ...]], ) -> None: name = execution.emergency_profile if name is None: return if execution.max_attempts_per_profile != 3: raise ConfigError( "max_attempts_per_profile must equal 3 when emergency_profile is configured" ) profile = profiles.get(name) if profile is None: raise ConfigError(f"emergency_profile {name!r} does not exist") if not profile.enabled: raise ConfigError(f"emergency_profile {name!r} must be enabled") if profile.adapter != "codex": raise ConfigError(f"emergency_profile {name!r} must use adapter codex") for role in ("implementation", "repair", "test"): if name in roles[role]: raise ConfigError( f"emergency_profile {name!r} must not appear in role {role!r}" ) def _validate_normal_role_adapters( profiles: Mapping[str, Profile], roles: Mapping[str, tuple[str, ...]], ) -> None: """Reject adapters outside ``NORMAL_ROLE_ADAPTERS`` on implementation, repair, and test. Applies to every profile in those roles, enabled or not, in every mode. """ for role in ("implementation", "repair", "test"): for name in roles[role]: adapter = profiles[name].adapter if adapter not in NORMAL_ROLE_ADAPTERS: raise ConfigError( f"profile {name!r} in role {role!r} must use adapter " "cursor, antigravity, or pi" ) def _validate_normal_role_exactness( profiles: Mapping[str, Profile], roles: Mapping[str, tuple[str, ...]], ) -> None: """Each normal role needs exactly one enabled profile per normal adapter. Disabled profiles do not count and may appear alongside the enabled one. Review is exempt. """ 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") if value < lo or value > hi: raise ConfigError(f"{name} out of bounds: {value}") return value def _build_profiles(raw: Any) -> dict[str, Profile]: if not isinstance(raw, dict) or not raw: raise ConfigError("profiles must be a non-empty mapping") out: dict[str, Profile] = {} for name, body in raw.items(): if not isinstance(body, dict): raise ConfigError(f"profile {name!r} must be a mapping") unknown = set(body) - PROFILE_KEYS if unknown: raise ConfigError(f"unknown keys in profile {name!r}: {sorted(unknown)}") missing = PROFILE_KEYS - set(body) if missing: raise ConfigError(f"missing keys in profile {name!r}: {sorted(missing)}") adapter = body["adapter"] executable = body["executable"] enabled = body["enabled"] model_raw = body["model"] if not isinstance(adapter, str) or not adapter: raise ConfigError(f"profile {name!r} adapter must be a string") if not isinstance(executable, str) or not executable: raise ConfigError(f"profile {name!r} executable must be a string") if type(enabled) is not bool: raise ConfigError(f"profile {name!r} enabled must be a boolean") if model_raw is None or model_raw == "auto": model: str | None = None elif isinstance(model_raw, str) and model_raw: model = model_raw else: raise ConfigError(f"profile {name!r} model must be a string or auto") out[name] = Profile( name=name, adapter=adapter, executable=executable, model=model, enabled=enabled, ) return out def _build_roles( raw: Any, profiles: Mapping[str, Profile] ) -> dict[str, tuple[str, ...]]: if not isinstance(raw, dict): raise ConfigError("roles must be a mapping") unknown = set(raw) - set(REQUIRED_ROLES) if unknown: raise ConfigError(f"unknown roles: {sorted(unknown)}") missing = set(REQUIRED_ROLES) - set(raw) if missing: raise ConfigError(f"missing roles: {sorted(missing)}") out: dict[str, tuple[str, ...]] = {} for role in REQUIRED_ROLES: value = raw[role] if not isinstance(value, list) or not value: raise ConfigError(f"role {role!r} must be a non-empty list") names: list[str] = [] seen: set[str] = set() for item in value: if not isinstance(item, str) or not item: raise ConfigError(f"role {role!r} entries must be strings") if item in seen: raise ConfigError(f"duplicate profile reference in role {role!r}: {item}") if item not in profiles: raise ConfigError(f"unknown profile reference: {item!r}") seen.add(item) names.append(item) out[role] = tuple(names) return out def _build_custom_adapters(raw: Any) -> dict[str, CustomAdapter]: if raw is None: return {} if not isinstance(raw, dict): raise ConfigError("custom_adapters must be a mapping") out: dict[str, CustomAdapter] = {} for name, body in raw.items(): if not isinstance(body, dict): raise ConfigError(f"custom adapter {name!r} must be a mapping") unknown = set(body) - CUSTOM_ADAPTER_KEYS if unknown: raise ConfigError( f"unknown keys in custom adapter {name!r}: {sorted(unknown)}" ) missing = CUSTOM_ADAPTER_KEYS - set(body) if missing: raise ConfigError( f"missing keys in custom adapter {name!r}: {sorted(missing)}" ) base_argv = _argv_tuple(body["base_argv"], f"custom_adapters.{name}.base_argv") run_everything_argv = _argv_tuple( body["run_everything_argv"], f"custom_adapters.{name}.run_everything_argv", ) workspace_argv = _argv_tuple( body["workspace_argv"], f"custom_adapters.{name}.workspace_argv" ) model_argv = _argv_tuple( body["model_argv"], f"custom_adapters.{name}.model_argv" ) prompt_transport = body["prompt_transport"] output_format = body["output_format"] if prompt_transport not in PROMPT_TRANSPORTS: raise ConfigError( f"unsupported prompt_transport for {name!r}: {prompt_transport!r}" ) if output_format not in OUTPUT_FORMATS: raise ConfigError( f"unsupported output_format for {name!r}: {output_format!r}" ) out[name] = CustomAdapter( name=name, base_argv=base_argv, run_everything_argv=run_everything_argv, workspace_argv=workspace_argv, model_argv=model_argv, prompt_transport=prompt_transport, output_format=output_format, ) return out def _argv_tuple(value: Any, label: str) -> tuple[str, ...]: if not isinstance(value, list): raise ConfigError(f"{label} must be a list") items: list[str] = [] for item in value: if not isinstance(item, str): raise ConfigError(f"{label} entries must be strings") matches = PLACEHOLDER_RE.findall(item) if matches: if len(matches) != 1 or item != matches[0]: raise ConfigError( f"placeholder must fill whole argv item in {label}: {item!r}" ) if matches[0] not in ALLOWED_PLACEHOLDERS: raise ConfigError(f"unknown placeholder {matches[0]!r} in {label}") else: _reject_shell_or_secret_value(label, item) items.append(item) return tuple(items)