diff --git a/AGENT-EXECUTION.md b/AGENT-EXECUTION.md index 662f22b1..c817b41a 100644 --- a/AGENT-EXECUTION.md +++ b/AGENT-EXECUTION.md @@ -19,7 +19,9 @@ candidate ends `unavailable` or `blocked`, and that result does not authorize the host to write the unit. The legacy value `all-unavailable` remains parseable and likewise grants no host write authority. Cursor uses `cursor-agent-stable`. `model: auto` lets each CLI use its configured default -model; PI therefore uses the operator-selected MiniMax default. +model; PI therefore uses the operator-selected MiniMax default. The +`integration` block declares the host-exempt paths and the transport, admission, +and cleanup entry points. ```agent-execution schema: agent-execution/v1 @@ -66,4 +68,13 @@ profiles: executable: codex model: auto enabled: true + +integration: + exempt_paths: [".task/**", "docs/superpowers/specs/**", "docs/superpowers/plans/**"] + transport: + command: python3 scripts/agent_exec.py apply + admission: + command: python3 scripts/agent_exec.py admit + cleanup: + command: make task-finish ``` diff --git a/scripts/agent_exec_config.py b/scripts/agent_exec_config.py index 405e8d84..a65b0ded 100644 --- a/scripts/agent_exec_config.py +++ b/scripts/agent_exec_config.py @@ -5,7 +5,7 @@ import ast import re from collections.abc import Mapping from dataclasses import dataclass -from pathlib import Path +from pathlib import Path, PurePosixPath from typing import Any SCHEMA_V1 = "agent-execution/v1" @@ -47,7 +47,17 @@ 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"}) +ROOT_KEYS = frozenset( + {"schema", "execution", "roles", "profiles", "custom_adapters", "integration"} +) +INTEGRATION_KEYS = frozenset({"exempt_paths", "transport", "admission", "cleanup"}) +ENTRYPOINT_KEYS = frozenset({"command"}) +# Mirrors the host-managed prefixes enforced by scripts/agent_exec.py +# (_HOST_MANAGED_PREFIXES, the enforcement source). A declared +# integration.exempt_paths must equal this set so it cannot widen host authority. +HOST_EXEMPT_PATHS = frozenset( + {".task/**", "docs/superpowers/specs/**", "docs/superpowers/plans/**"} +) EXECUTION_KEYS = frozenset( { "fresh_process_per_unit", @@ -116,6 +126,14 @@ class CustomAdapter: output_format: str +@dataclass(frozen=True) +class IntegrationConfig: + exempt_paths: tuple[str, ...] + transport_command: str + admission_command: str + cleanup_command: str + + @dataclass(frozen=True) class ExecutionConfig: schema: str @@ -123,6 +141,7 @@ class ExecutionConfig: roles: Mapping[str, tuple[str, ...]] profiles: Mapping[str, Profile] custom_adapters: Mapping[str, CustomAdapter] + integration: IntegrationConfig | None = None def load_config(repo: Path) -> ExecutionConfig: @@ -351,12 +370,17 @@ def _build_config(raw: dict[str, Any]) -> ExecutionConfig: _validate_normal_role_adapters(profiles, roles) _validate_normal_role_exactness(profiles, roles) + integration = None + if "integration" in raw: + integration = _build_integration(raw["integration"]) + return ExecutionConfig( schema=schema, execution=execution, roles=roles, profiles=profiles, custom_adapters=custom_adapters, + integration=integration, ) @@ -634,6 +658,62 @@ def _build_custom_adapters(raw: Any) -> dict[str, CustomAdapter]: return out +def _build_integration(raw: Any) -> IntegrationConfig: + if not isinstance(raw, dict): + raise ConfigError("integration must be a mapping") + unknown = set(raw) - INTEGRATION_KEYS + if unknown: + raise ConfigError(f"unknown keys in integration: {sorted(unknown)}") + missing = INTEGRATION_KEYS - set(raw) + if missing: + raise ConfigError(f"missing keys in integration: {sorted(missing)}") + exempt_paths = _exempt_paths_tuple(raw["exempt_paths"]) + if set(exempt_paths) != HOST_EXEMPT_PATHS: + raise ConfigError( + "integration.exempt_paths must exactly match the host-managed set " + f"{sorted(HOST_EXEMPT_PATHS)}; got {list(exempt_paths)}" + ) + return IntegrationConfig( + exempt_paths=exempt_paths, + transport_command=_entrypoint_command(raw["transport"], "integration.transport"), + admission_command=_entrypoint_command(raw["admission"], "integration.admission"), + cleanup_command=_entrypoint_command(raw["cleanup"], "integration.cleanup"), + ) + + +def _exempt_paths_tuple(value: Any) -> tuple[str, ...]: + if not isinstance(value, list) or not value: + raise ConfigError("integration.exempt_paths must be a non-empty list") + items: list[str] = [] + for item in value: + if not isinstance(item, str) or not item: + raise ConfigError("integration.exempt_paths entries must be non-empty strings") + if item.startswith("/") or ".." in PurePosixPath(item).parts: + raise ConfigError( + f"integration.exempt_paths must be relative without '..': {item!r}" + ) + if item in items: + raise ConfigError(f"duplicate integration.exempt_paths entry: {item!r}") + items.append(item) + return tuple(items) + + +def _entrypoint_command(raw: Any, label: str) -> str: + if not isinstance(raw, dict): + raise ConfigError(f"{label} must be a mapping") + unknown = set(raw) - ENTRYPOINT_KEYS + if unknown: + raise ConfigError(f"unknown keys in {label}: {sorted(unknown)}") + missing = ENTRYPOINT_KEYS - set(raw) + if missing: + raise ConfigError(f"missing keys in {label}: {sorted(missing)}") + command = raw["command"] + if not isinstance(command, str) or not command.strip(): + raise ConfigError(f"{label}.command must be a non-empty string") + _reject_shell_or_secret_value(f"{label}.command", command) + return command + + def _argv_tuple(value: Any, label: str) -> tuple[str, ...]: if not isinstance(value, list): raise ConfigError(f"{label} must be a list") diff --git a/tests/test_agent_exec_config.py b/tests/test_agent_exec_config.py index e6f2c88a..25b779a2 100644 --- a/tests/test_agent_exec_config.py +++ b/tests/test_agent_exec_config.py @@ -792,6 +792,117 @@ class AgentExecConfigTests(unittest.TestCase): self.assertTrue(cfg.profiles["codex"].enabled) self.assertEqual(cfg.profiles["codex"].adapter, "codex") + def test_valid_integration_block_parses_and_is_exposed(self) -> None: + write_repo(self.repo, MINIMAL_BLOCK + "\n" + VALID_INTEGRATION) + cfg = load_config(self.repo) + self.assertIsNotNone(cfg.integration) + assert cfg.integration is not None + self.assertEqual( + cfg.integration.exempt_paths, + (".task/**", "docs/superpowers/specs/**", "docs/superpowers/plans/**"), + ) + self.assertEqual( + cfg.integration.transport_command, "python3 scripts/agent_exec.py apply" + ) + self.assertEqual( + cfg.integration.admission_command, "python3 scripts/agent_exec.py admit" + ) + self.assertEqual(cfg.integration.cleanup_command, "make task-finish") + + def test_missing_integration_still_parses_as_none(self) -> None: + write_repo(self.repo, MINIMAL_BLOCK) + self.assertIsNone(load_config(self.repo).integration) + + def test_integration_rejects_unknown_and_missing_keys(self) -> None: + cases = { + "unknown key in integration": VALID_INTEGRATION.replace( + " cleanup:\n", " extra: x\n cleanup:\n" + ), + "unknown key in transport": VALID_INTEGRATION.replace( + " command: python3 scripts/agent_exec.py apply\n", + " command: python3 scripts/agent_exec.py apply\n extra: x\n", + ), + "missing cleanup": VALID_INTEGRATION.split(" cleanup:")[0], + } + for label, block in cases.items(): + with self.subTest(label): + write_repo(self.repo, MINIMAL_BLOCK + "\n" + block) + with self.assertRaises(ConfigError): + load_config(self.repo) + + def test_integration_exempt_paths_must_be_unique_nonempty_and_exact(self) -> None: + cases = { + "empty": 'exempt_paths: []', + "duplicate": ( + 'exempt_paths: [".task/**", ".task/**", ' + '"docs/superpowers/specs/**", "docs/superpowers/plans/**"]' + ), + "widened": ( + 'exempt_paths: [".task/**", "docs/superpowers/specs/**", ' + '"docs/superpowers/plans/**", "src/**"]' + ), + "narrowed": 'exempt_paths: [".task/**", "docs/superpowers/specs/**"]', + "parent traversal": ( + 'exempt_paths: [".task/**", "docs/superpowers/specs/**", ' + '"docs/superpowers/plans/../../**"]' + ), + "absolute": ( + 'exempt_paths: [".task/**", "docs/superpowers/specs/**", ' + '"/docs/superpowers/plans/**"]' + ), + } + for label, value in cases.items(): + with self.subTest(label): + block = VALID_INTEGRATION.replace( + 'exempt_paths: [".task/**", "docs/superpowers/specs/**", ' + '"docs/superpowers/plans/**"]', + value, + ) + write_repo(self.repo, MINIMAL_BLOCK + "\n" + block) + with self.assertRaises(ConfigError): + load_config(self.repo) + + def test_integration_command_rejects_shell_metacharacters(self) -> None: + block = VALID_INTEGRATION.replace( + "command: python3 scripts/agent_exec.py apply", + "command: python3 scripts/agent_exec.py apply; rm -rf /", + ) + write_repo(self.repo, MINIMAL_BLOCK + "\n" + block) + with self.assertRaises(ConfigError): + load_config(self.repo) + + def test_host_exempt_paths_match_executor_enforcement(self) -> None: + from scripts import agent_exec + from scripts.agent_exec_config import HOST_EXEMPT_PATHS + + enforced = {prefix.rstrip("/") for prefix in agent_exec._HOST_MANAGED_PREFIXES} + declared = {glob.removesuffix("/**") for glob in HOST_EXEMPT_PATHS} + self.assertEqual(declared, enforced) + + def test_checked_in_project_config_declares_integration(self) -> None: + cfg = load_config(Path(__file__).resolve().parents[1]) + self.assertIsNotNone(cfg.integration) + assert cfg.integration is not None + self.assertEqual( + set(cfg.integration.exempt_paths), + {".task/**", "docs/superpowers/specs/**", "docs/superpowers/plans/**"}, + ) + self.assertEqual(cfg.integration.cleanup_command, "make task-finish") + + +VALID_INTEGRATION = textwrap.dedent( + """\ + integration: + exempt_paths: [".task/**", "docs/superpowers/specs/**", "docs/superpowers/plans/**"] + transport: + command: python3 scripts/agent_exec.py apply + admission: + command: python3 scripts/agent_exec.py admit + cleanup: + command: make task-finish + """ +) + def _legacy_custom_block() -> str: """No emergency_profile and adapter custom on every role."""