agent-exec transport 879b9a517efd4e3f85339649d5b00586/395b33d1abda

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
 ```

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")

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."""