Add project execution agent configuration

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
This commit is contained in:
2026-10-15 02:17:29 +00:00
parent 05e45a9044
commit bdad8ffc91
4 changed files with 956 additions and 1 deletions
+71
View File
@@ -0,0 +1,71 @@
# Project Agent Execution
Project-local execution routing configuration. This document is the single
authority for which Agent CLI profiles may implement, repair, test, or review
bounded units. Human-readable notes may appear outside the machine block; the
parser accepts exactly one `agent-execution` fenced block.
U1 delivers configuration parsing and validation only. Routing is not enabled
until later units wire adapters and the execution controller.
```agent-execution
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, claude, codex, pi, kimi, opencode]
repair: [cursor, antigravity, codex, claude]
test: [pi, antigravity, codex]
review: [claude, codex, pi]
profiles:
cursor:
adapter: cursor
executable: cursor-agent
model: auto
enabled: true
antigravity:
adapter: antigravity
executable: agy
model: auto
enabled: true
claude:
adapter: claude
executable: claude
model: auto
enabled: true
codex:
adapter: codex
executable: codex
model: auto
enabled: true
pi:
adapter: pi
executable: pi
model: auto
enabled: true
kimi:
adapter: kimi
executable: kimi
model: auto
enabled: true
opencode:
adapter: opencode
executable: opencode
model: auto
enabled: true
```
+7 -1
View File
@@ -13,7 +13,7 @@ PLUGINS := plugins
COMBO := core=$(shell git -C $(CORE) rev-parse --short HEAD 2>/dev/null)+backend=$(shell git -C $(BACKEND) rev-parse --short HEAD 2>/dev/null)+frontend=$(shell git -C $(FRONTEND) rev-parse --short HEAD 2>/dev/null)+ui=$(shell git -C $(UI) rev-parse --short HEAD 2>/dev/null)+agentcmd=$(shell git -C deps/agent-command rev-parse --short HEAD 2>/dev/null)+documents=$(shell git -C modules/documents rev-parse --short HEAD 2>/dev/null)+native=$(shell git -C deps/agent-native rev-parse --short HEAD 2>/dev/null)+squad=$(shell git -C deps/agent-squad rev-parse --short HEAD 2>/dev/null)+intent=$(shell git -C deps/agent-intent rev-parse --short HEAD 2>/dev/null)
VERSION := $(shell git describe --tags --always --dirty 2>/dev/null)+$(COMBO)
.PHONY: help sync status build frontend svc svc-list svc-info svc-commits svc-history svc-update svc-recover svc-watchdog test-backend test-backend-pg test-multi-instance test-lifecycle test-frontend test-ui test-core test-agentcmd test-pins test-libs dev pin clean setup precheck task-start task-resume task-sync task-close task-finish facts-verify evidence version
.PHONY: help sync status build frontend svc svc-list svc-info svc-commits svc-history svc-update svc-recover svc-watchdog test-backend test-backend-pg test-multi-instance test-lifecycle test-frontend test-ui test-core test-agentcmd test-pins test-libs dev pin clean setup precheck task-start task-resume task-sync task-close task-finish facts-verify evidence version agent-exec-validate
version:
@echo "agent-system $(VERSION)"
@@ -77,6 +77,12 @@ help:
@echo "make dev 开发模式说明(前后端分离热更新)"
@echo "make pin 推进 submodule 指针到各仓远端 main(提交前须 make test 通过)"
@echo "make status submodule 状态"
@echo "make agent-exec-validate 校验 AGENT-EXECUTION.md 配置合同(U1:配置解析单测)"
# U1 temporary target: configuration contract tests. Task 3 replaces the recipe
# with `python3 scripts/agent_exec.py validate --repo .`.
agent-exec-validate:
PYTHONDONTWRITEBYTECODE=1 python3 -m unittest tests.test_agent_exec_config -v
sync:
git submodule update --init --recursive
+539
View File
@@ -0,0 +1,539 @@
"""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)
+339
View File
@@ -0,0 +1,339 @@
"""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 (
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, claude]
repair: [cursor]
test: [claude]
review: [claude]
profiles:
cursor:
adapter: cursor
executable: cursor-agent
model: auto
enabled: true
claude:
adapter: claude
executable: claude
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", "claude"))
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 claude before cursor; role order must stay cursor -> claude
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, claude]
repair: [cursor]
test: [claude]
review: [claude]
profiles:
claude:
adapter: claude
executable: claude
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", "claude"))
self.assertEqual(list(cfg.profiles.keys())[0], "claude")
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["claude"].model)
def test_rejects_unknown_profile_reference(self) -> None:
block = MINIMAL_BLOCK.replace(
"implementation: [cursor, claude]",
"implementation: [cursor, claude, 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 claude:",
"enabled: true\n api_token: \"sk-secret-value\"\n\n claude:",
)
write_repo(self.repo, secret_key)
with self.assertRaises(ConfigError):
load_config(self.repo)
secret_val = MINIMAL_BLOCK.replace(
"executable: claude",
"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 = 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
"""
)
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: claude",
'executable: "模型-β-测试"',
)
write_repo(self.repo, block)
cfg = load_config(self.repo)
self.assertEqual(cfg.profiles["claude"].executable, "模型-β-测试")
escaped = MINIMAL_BLOCK.replace(
"executable: claude",
r'executable: "say \"hi\" 模型"',
)
write_repo(self.repo, escaped)
cfg = load_config(self.repo)
self.assertEqual(cfg.profiles["claude"].executable, 'say "hi" 模型')
if __name__ == "__main__":
unittest.main()