weichen
bdad8ffc91
Why: Establish one validated project-local routing document before any real Agent execution is enabled. Evidence: Cursor observed red then 12 green configuration tests; the host replays those tests, the temporary validation target, and whitespace checks during integration. Impact: Adds the U1 configuration contract and parser only; real routing, adapters, controller execution, and governance adoption remain disabled. Refs: docs/superpowers/plans/2026-10-15-project-agent-execution.md
540 lines
18 KiB
Python
540 lines
18 KiB
Python
"""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"})
|
|
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
|
|
|
|
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",
|
|
"summary_bytes",
|
|
"raw_output_bytes",
|
|
}
|
|
)
|
|
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
|
|
|
|
|
|
@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}")
|
|
|
|
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_KEYS - 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
|
|
)
|
|
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")
|
|
|
|
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,
|
|
)
|
|
|
|
|
|
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)
|