From a80e74306979fb340c6bf7bbb36db1dafc8ebb44 Mon Sep 17 00:00:00 2001 From: saagpatel Date: Sat, 6 Jun 2026 02:46:59 -0700 Subject: [PATCH] feat(automation): gated executor with safety rails (Arc D phase 3a) Add src/automation_executor.py: the only layer that mutates real repos, built safe by construction. Every git/gh call goes through an injected CommandRunner (real subprocess in prod, fake in tests), so the full command sequence and all rails are unit-tested without touching a repo or network. Rails (all tested): - require_approved is the hard pre-gate; an unapproved proposal runs nothing. - dry_run defaults to True; in dry-run no mutating command and no apply_change run. - context-PR refuses protected/default branches (case-insensitive), rejects unsafe/flag-like branch names, requires a non-empty default branch, never uses --force, skips dirty worktrees and non-git repos. - fail-fast: a failed checkout/commit never reaches push or gh pr create; an apply_change exception returns 'failed' and best-effort restores the branch. - catalog-seed apply is local + reversible (only fills missing fields). - mark_executed (added to automation_proposals): APPROVED -> EXECUTED only. Content generation (managed block, catalog seeds) is supplied by the caller and computed fresh at execution time; phase 3b wires it to the recovery primitives and adds the propose/approve/execute CLI. 27 executor tests + mark_executed tests. Full suite 2394 passed, ruff clean. /code-review (adversarial) applied: fixed case-insensitive branch bypass, empty default_branch, flag-like branch injection, and apply_change strand. --- src/automation_executor.py | 234 +++++++++++++++++++ src/automation_proposals.py | 22 ++ tests/test_automation_executor.py | 373 ++++++++++++++++++++++++++++++ 3 files changed, 629 insertions(+) create mode 100644 src/automation_executor.py create mode 100644 tests/test_automation_executor.py diff --git a/src/automation_executor.py b/src/automation_executor.py new file mode 100644 index 0000000..a6dd382 --- /dev/null +++ b/src/automation_executor.py @@ -0,0 +1,234 @@ +"""Gated executor for bounded automation (Arc D, phase 3). + +This is the only layer that mutates real repos, so it is built to be safe by +construction: + +* every git/gh interaction goes through an injected ``CommandRunner`` (real + subprocess in production, a fake in tests); +* ``require_approved`` is the hard pre-gate — an unapproved proposal never runs; +* ``dry_run`` defaults to ``True`` — callers must opt in to apply; +* context-PR work refuses the default/protected branch, never uses ``--force``, + skips dirty worktrees and non-git repos, and aborts fast on the first failed + command (so a failed commit never reaches push/PR); +* catalog-seed work is local + reversible and only fills missing fields. + +Content generation (the managed block, the catalog seed dict) is supplied by the +caller via ``ExecutionPlan.apply_change`` / the ``seeds`` argument, computed fresh +at execution time — phase-3b wires those to the existing recovery primitives. +""" + +from __future__ import annotations + +import re +import subprocess +from dataclasses import dataclass +from pathlib import Path +from typing import Callable + +from src.automation_proposals import AutomationProposal, require_approved + +CONTRACT_VERSION = "automation_execution_v1" + +# Branches the executor will never commit automation onto. +PROTECTED_BRANCHES = frozenset({"main", "master"}) + +# A safe feature-branch name: alphanumeric/dot/underscore first char (never a +# leading '-', which git/gh would parse as a flag), then word/./_/-/ characters. +_SAFE_BRANCH = re.compile(r"^[A-Za-z0-9._][A-Za-z0-9._/-]*$") + +_GIT_TIMEOUT_SECONDS = 120 + + +class AutomationExecutionError(RuntimeError): + """Raised when an execution safety rail (e.g. branch policy) is violated.""" + + +@dataclass(frozen=True) +class CommandResult: + returncode: int + stdout: str = "" + stderr: str = "" + + +# (args, cwd) -> CommandResult. Injected so tests never touch git/gh/network. +CommandRunner = Callable[[list[str], Path], CommandResult] + + +def default_command_runner(args: list[str], cwd: Path) -> CommandResult: + """Run a command via subprocess (the production ``CommandRunner``).""" + proc = subprocess.run( + args, + cwd=cwd, + capture_output=True, + text=True, + timeout=_GIT_TIMEOUT_SECONDS, + check=False, + ) + return CommandResult(proc.returncode, proc.stdout, proc.stderr) + + +@dataclass(frozen=True) +class ExecutionPlan: + """The concrete change to apply for one context-PR proposal. + + ``apply_change`` writes the file changes into ``repo_path`` (computed fresh + by the caller); the executor handles all git/gh mechanics around it. + """ + + repo_path: Path + default_branch: str + branch_name: str + commit_message: str + pr_title: str + pr_body: str + apply_change: Callable[[Path], None] + has_git: bool = True + + +@dataclass(frozen=True) +class ExecutionResult: + proposal_id: str + outcome: str # applied | dry-run | skipped | failed + detail: str + reference: str = "" + commands: tuple[str, ...] = () + + +def _dirty_worktree(repo_path: Path, runner: CommandRunner) -> bool: + result = runner(["git", "status", "--porcelain"], repo_path) + if result.returncode != 0: + # A failed status check is treated as "do not touch" (conservative). + return True + return bool(result.stdout.strip()) + + +def execute_context_pr( + proposal: AutomationProposal, + plan: ExecutionPlan, + *, + dry_run: bool = True, + runner: CommandRunner = default_command_runner, +) -> ExecutionResult: + """Open a context-improvement PR for an APPROVED proposal, behind every rail.""" + require_approved(proposal) # hard gate — raises if not approved + + branch = plan.branch_name.strip() + default_branch = plan.default_branch.strip() + if not branch: + raise AutomationExecutionError("branch_name must be non-empty") + if not default_branch: + raise AutomationExecutionError("default_branch must be non-empty") + if not _SAFE_BRANCH.match(branch): + # Rejects leading '-' (flag injection into git/gh) and whitespace. + raise AutomationExecutionError(f"unsafe branch name {branch!r}") + if branch.lower() in PROTECTED_BRANCHES or branch.lower() == default_branch.lower(): + raise AutomationExecutionError( + f"refusing to commit automation onto protected/default branch {branch!r}" + ) + + if not plan.has_git: + return ExecutionResult(proposal.proposal_id, "skipped", "no-git-repo") + + planned = [ + f"git checkout -b {branch}", + "", + "git add -A", + f"git commit -m {plan.commit_message!r}", + f"git push -u origin {branch}", + f"gh pr create --base {default_branch} --head {branch} ...", + ] + + if _dirty_worktree(plan.repo_path, runner): + return ExecutionResult(proposal.proposal_id, "skipped", "dirty-worktree") + + if dry_run: + return ExecutionResult( + proposal.proposal_id, + "dry-run", + f"Would open a PR on branch {branch}.", + commands=tuple(planned), + ) + + # Run checkout first, then apply the change, then stage/commit/push/PR. + checkout = runner(["git", "checkout", "-b", branch], plan.repo_path) + if checkout.returncode != 0: + return ExecutionResult( + proposal.proposal_id, "failed", f"checkout: {checkout.stderr}".strip() + ) + + try: + plan.apply_change(plan.repo_path) + except Exception as error: # noqa: BLE001 - surface as a failed result, not a strand + # The change failed after the branch was created; best-effort return to + # the default branch so the repo is not left stranded on a partial branch. + runner(["git", "checkout", default_branch], plan.repo_path) + return ExecutionResult(proposal.proposal_id, "failed", f"apply-change: {error}".strip()) + + for args in ( + ["git", "add", "-A"], + ["git", "commit", "-m", plan.commit_message], + ["git", "push", "-u", "origin", branch], + ): + result = runner(args, plan.repo_path) + if result.returncode != 0: + return ExecutionResult( + proposal.proposal_id, "failed", f"{args[1]}: {result.stderr}".strip() + ) + + pr = runner( + [ + "gh", + "pr", + "create", + "--base", + default_branch, + "--head", + branch, + "--title", + plan.pr_title, + "--body", + plan.pr_body, + ], + plan.repo_path, + ) + if pr.returncode != 0: + return ExecutionResult(proposal.proposal_id, "failed", f"gh pr create: {pr.stderr}".strip()) + + return ExecutionResult( + proposal.proposal_id, + "applied", + f"Opened PR on branch {branch}.", + reference=pr.stdout.strip(), + ) + + +def execute_catalog_seed( + proposal: AutomationProposal, + *, + catalog_path: Path, + seeds: dict[str, dict[str, str]], + dry_run: bool = True, + apply_seeds: Callable[[Path, dict[str, dict[str, str]]], None] | None = None, +) -> ExecutionResult: + """Apply catalog seed updates for an APPROVED proposal (local + reversible).""" + require_approved(proposal) # hard gate + + repo_keys = ", ".join(sorted(seeds)) or "(none)" + if dry_run: + return ExecutionResult( + proposal.proposal_id, + "dry-run", + f"Would fill missing catalog fields for: {repo_keys}.", + ) + + if apply_seeds is None: + from src.portfolio_context_recovery import _merge_catalog_seeds + + apply_seeds = _merge_catalog_seeds + apply_seeds(catalog_path, seeds) + return ExecutionResult( + proposal.proposal_id, + "applied", + f"Filled missing catalog fields for: {repo_keys}.", + reference=str(catalog_path), + ) diff --git a/src/automation_proposals.py b/src/automation_proposals.py index e1f2919..d263b3f 100644 --- a/src/automation_proposals.py +++ b/src/automation_proposals.py @@ -239,6 +239,28 @@ def reject_proposal( ) +def mark_executed( + proposals: Iterable[AutomationProposal], + proposal_id: str, + *, + executed_at: str, + execution_ref: str, +) -> list[AutomationProposal]: + """Transition an APPROVED proposal to EXECUTED after a successful apply. + + Only an APPROVED proposal can be executed — this mirrors the ``require_approved`` + gate so the execution lifecycle cannot skip approval. + """ + return _transition( + proposals, + proposal_id, + expected_status=STATUS_APPROVED, + status=STATUS_EXECUTED, + executed_at=executed_at, + execution_ref=execution_ref, + ) + + # --- enforcement gate ------------------------------------------------------ diff --git a/tests/test_automation_executor.py b/tests/test_automation_executor.py new file mode 100644 index 0000000..b29bb64 --- /dev/null +++ b/tests/test_automation_executor.py @@ -0,0 +1,373 @@ +"""Tests for the Arc D phase-3 gated executor. + +Every git/gh interaction goes through an injected ``CommandRunner``, so these +tests verify the exact command sequence and every safety rail (approval gate, +never-default-branch, never-force, skip-dirty, dry-run default, fail-fast) with +no real repository or network involved. +""" + +from __future__ import annotations + +from pathlib import Path + +import pytest + +from src.automation_executor import ( + AutomationExecutionError, + CommandResult, + ExecutionPlan, + execute_catalog_seed, + execute_context_pr, +) +from src.automation_proposals import ( + ACTION_CATALOG_SEED, + ACTION_CONTEXT_PR, + STATUS_APPROVED, + STATUS_EXECUTED, + STATUS_PENDING, + AutomationProposal, + ProposalApprovalError, + ProposalNotFoundError, + mark_executed, +) + +NOW = "2026-04-14T12:00:00+00:00" + + +class FakeRunner: + """Records every command and returns canned results keyed by the first 2 args.""" + + def __init__(self, results: dict[tuple[str, ...], CommandResult] | None = None) -> None: + self.calls: list[tuple[list[str], Path]] = [] + self._results = results or {} + + def __call__(self, args: list[str], cwd: Path) -> CommandResult: + self.calls.append((list(args), cwd)) + key = tuple(args[:2]) + return self._results.get(key, CommandResult(returncode=0, stdout="", stderr="")) + + @property + def command_args(self) -> list[list[str]]: + return [args for args, _ in self.calls] + + +def _approved(action_type: str = ACTION_CONTEXT_PR) -> AutomationProposal: + return AutomationProposal( + proposal_id=f"{action_type}:owner/Repo", + action_type=action_type, + display_name="Repo", + repo_full_name="owner/Repo", + description="d", + status=STATUS_APPROVED, + created_at=NOW, + ) + + +def _plan(tmp_path: Path, applied: list[Path], **overrides) -> ExecutionPlan: + defaults = dict( + repo_path=tmp_path, + default_branch="main", + branch_name="auto/context-repo", + commit_message="docs: refresh managed context block", + pr_title="docs: refresh managed context block", + pr_body="Automated context improvement.", + apply_change=lambda path: applied.append(path), + has_git=True, + ) + defaults.update(overrides) + return ExecutionPlan(**defaults) + + +# --- approval gate --------------------------------------------------------- + + +def test_context_pr_requires_approved_proposal(tmp_path: Path) -> None: + pending = AutomationProposal( + proposal_id="context-pr:owner/Repo", + action_type=ACTION_CONTEXT_PR, + display_name="Repo", + repo_full_name="owner/Repo", + description="d", + status=STATUS_PENDING, + ) + runner = FakeRunner() + with pytest.raises(ProposalApprovalError): + execute_context_pr(pending, _plan(tmp_path, []), dry_run=False, runner=runner) + assert runner.calls == [] # nothing ran + + +# --- never the default branch ---------------------------------------------- + + +@pytest.mark.parametrize("branch", ["main", "master"]) +def test_context_pr_refuses_protected_branch(tmp_path: Path, branch: str) -> None: + runner = FakeRunner() + with pytest.raises(AutomationExecutionError): + execute_context_pr( + _approved(), _plan(tmp_path, [], branch_name=branch), dry_run=False, runner=runner + ) + assert runner.calls == [] + + +def test_context_pr_refuses_branch_equal_to_default(tmp_path: Path) -> None: + runner = FakeRunner() + with pytest.raises(AutomationExecutionError): + execute_context_pr( + _approved(), + _plan(tmp_path, [], default_branch="develop", branch_name="develop"), + dry_run=False, + runner=runner, + ) + + +def test_context_pr_refuses_empty_branch(tmp_path: Path) -> None: + runner = FakeRunner() + with pytest.raises(AutomationExecutionError): + execute_context_pr( + _approved(), _plan(tmp_path, [], branch_name=""), dry_run=False, runner=runner + ) + + +@pytest.mark.parametrize("branch", ["Main", "MASTER"]) +def test_context_pr_protected_branch_guard_is_case_insensitive(tmp_path: Path, branch: str) -> None: + runner = FakeRunner() + with pytest.raises(AutomationExecutionError): + execute_context_pr( + _approved(), _plan(tmp_path, [], branch_name=branch), dry_run=False, runner=runner + ) + assert runner.calls == [] + + +def test_context_pr_refuses_branch_equal_to_default_case_insensitive(tmp_path: Path) -> None: + runner = FakeRunner() + with pytest.raises(AutomationExecutionError): + execute_context_pr( + _approved(), + _plan(tmp_path, [], default_branch="Develop", branch_name="develop"), + dry_run=False, + runner=runner, + ) + + +@pytest.mark.parametrize("branch", ["--force", "-q", "has space", "--head"]) +def test_context_pr_refuses_flag_like_or_unsafe_branch(tmp_path: Path, branch: str) -> None: + runner = FakeRunner() + with pytest.raises(AutomationExecutionError): + execute_context_pr( + _approved(), _plan(tmp_path, [], branch_name=branch), dry_run=False, runner=runner + ) + assert runner.calls == [] + + +def test_context_pr_refuses_empty_default_branch(tmp_path: Path) -> None: + runner = FakeRunner() + with pytest.raises(AutomationExecutionError): + execute_context_pr( + _approved(), _plan(tmp_path, [], default_branch=""), dry_run=False, runner=runner + ) + + +def test_context_pr_apply_change_failure_returns_failed_and_restores_branch( + tmp_path: Path, +) -> None: + runner = FakeRunner() + + def boom(_path: Path) -> None: + raise OSError("disk full") + + plan = _plan(tmp_path, [], apply_change=boom) + result = execute_context_pr(_approved(), plan, dry_run=False, runner=runner) + assert result.outcome == "failed" + assert "disk full" in result.detail + # Best-effort restore to the default branch; never reaches commit/push/PR. + assert ["git", "checkout", "main"] in runner.command_args + assert not any(args[:2] == ["git", "commit"] for args in runner.command_args) + assert not any(args[:2] == ["gh", "pr"] for args in runner.command_args) + + +# --- skip rails ------------------------------------------------------------ + + +def test_context_pr_skips_dirty_worktree(tmp_path: Path) -> None: + runner = FakeRunner({("git", "status"): CommandResult(0, " M file.py\n")}) + applied: list[Path] = [] + result = execute_context_pr(_approved(), _plan(tmp_path, applied), dry_run=False, runner=runner) + assert result.outcome == "skipped" + assert "dirty" in result.detail + assert applied == [] # change never applied + # Only the read-only status check ran; no mutating commands. + assert runner.command_args == [["git", "status", "--porcelain"]] + + +def test_context_pr_skips_when_no_git(tmp_path: Path) -> None: + runner = FakeRunner() + applied: list[Path] = [] + result = execute_context_pr( + _approved(), _plan(tmp_path, applied, has_git=False), dry_run=False, runner=runner + ) + assert result.outcome == "skipped" + assert applied == [] + assert runner.calls == [] + + +# --- dry run (default) ----------------------------------------------------- + + +def test_context_pr_dry_run_is_default_and_mutates_nothing(tmp_path: Path) -> None: + runner = FakeRunner() + applied: list[Path] = [] + result = execute_context_pr( + _approved(), _plan(tmp_path, applied), runner=runner + ) # no dry_run arg + assert result.outcome == "dry-run" + assert applied == [] + # Only the read-only dirty check ran; planned commands are reported, not run. + assert runner.command_args == [["git", "status", "--porcelain"]] + assert any("gh pr create" in cmd for cmd in result.commands) + assert any("checkout -b auto/context-repo" in cmd for cmd in result.commands) + + +# --- apply ----------------------------------------------------------------- + + +def test_context_pr_apply_runs_full_sequence_and_opens_pr(tmp_path: Path) -> None: + runner = FakeRunner({("gh", "pr"): CommandResult(0, "https://github.com/owner/Repo/pull/7\n")}) + applied: list[Path] = [] + result = execute_context_pr(_approved(), _plan(tmp_path, applied), dry_run=False, runner=runner) + assert result.outcome == "applied" + assert result.reference == "https://github.com/owner/Repo/pull/7" + assert applied == [tmp_path] # change applied exactly once + + cmds = runner.command_args + assert cmds == [ + ["git", "status", "--porcelain"], + ["git", "checkout", "-b", "auto/context-repo"], + ["git", "add", "-A"], + ["git", "commit", "-m", "docs: refresh managed context block"], + ["git", "push", "-u", "origin", "auto/context-repo"], + [ + "gh", + "pr", + "create", + "--base", + "main", + "--head", + "auto/context-repo", + "--title", + "docs: refresh managed context block", + "--body", + "Automated context improvement.", + ], + ] + + +def test_context_pr_never_uses_force(tmp_path: Path) -> None: + runner = FakeRunner() + execute_context_pr(_approved(), _plan(tmp_path, []), dry_run=False, runner=runner) + flat = [token for args in runner.command_args for token in args] + assert not any("force" in token for token in flat) + + +def test_context_pr_apply_change_runs_after_branch_before_add(tmp_path: Path) -> None: + order: list[str] = [] + + def record_runner(args: list[str], cwd: Path) -> CommandResult: + order.append(" ".join(args[:2])) + return CommandResult(0, "") + + plan = _plan(tmp_path, [], apply_change=lambda path: order.append("APPLY")) + execute_context_pr(_approved(), plan, dry_run=False, runner=record_runner) + assert order.index("git checkout") < order.index("APPLY") < order.index("git add") + + +def test_context_pr_aborts_on_commit_failure(tmp_path: Path) -> None: + runner = FakeRunner({("git", "commit"): CommandResult(1, "", "nothing to commit")}) + result = execute_context_pr(_approved(), _plan(tmp_path, []), dry_run=False, runner=runner) + assert result.outcome == "failed" + assert "nothing to commit" in result.detail + # push / pr create must NOT run after a failed commit. + assert ["git", "push", "-u", "origin", "auto/context-repo"] not in runner.command_args + assert not any(args[:2] == ["gh", "pr"] for args in runner.command_args) + + +# --- catalog seed (local, reversible) -------------------------------------- + + +def test_catalog_seed_requires_approved(tmp_path: Path) -> None: + pending = AutomationProposal( + proposal_id="catalog-seed:owner/Repo", + action_type=ACTION_CATALOG_SEED, + display_name="Repo", + repo_full_name="owner/Repo", + description="d", + status=STATUS_PENDING, + ) + with pytest.raises(ProposalApprovalError): + execute_catalog_seed( + pending, + catalog_path=tmp_path / "catalog.yaml", + seeds={"repo": {"path_confidence": "high"}}, + dry_run=False, + ) + + +def test_catalog_seed_dry_run_does_not_write(tmp_path: Path) -> None: + catalog = tmp_path / "catalog.yaml" + result = execute_catalog_seed( + _approved(ACTION_CATALOG_SEED), + catalog_path=catalog, + seeds={"repo": {"path_confidence": "high"}}, + ) # dry_run defaults True + assert result.outcome == "dry-run" + assert not catalog.exists() + + +def test_catalog_seed_apply_writes_only_missing_fields(tmp_path: Path) -> None: + catalog = tmp_path / "catalog.yaml" + catalog.write_text("repos:\n repo:\n purpose: existing\n") + result = execute_catalog_seed( + _approved(ACTION_CATALOG_SEED), + catalog_path=catalog, + seeds={"repo": {"path_confidence": "high", "purpose": "should-not-overwrite"}}, + dry_run=False, + ) + assert result.outcome == "applied" + text = catalog.read_text() + assert "path_confidence: high" in text + assert "purpose: existing" in text # never overwritten + assert "should-not-overwrite" not in text + + +# --- mark_executed lifecycle ---------------------------------------------- + + +def test_mark_executed_transitions_approved_to_executed() -> None: + proposals = [_approved()] + updated = mark_executed( + proposals, + "context-pr:owner/Repo", + executed_at=NOW, + execution_ref="https://github.com/owner/Repo/pull/7", + ) + [proposal] = updated + assert proposal.status == STATUS_EXECUTED + assert proposal.executed_at == NOW + assert proposal.execution_ref == "https://github.com/owner/Repo/pull/7" + + +def test_mark_executed_rejects_non_approved() -> None: + pending = AutomationProposal( + proposal_id="context-pr:owner/Repo", + action_type=ACTION_CONTEXT_PR, + display_name="Repo", + repo_full_name="owner/Repo", + description="d", + status=STATUS_PENDING, + ) + with pytest.raises(ProposalApprovalError): + mark_executed([pending], "context-pr:owner/Repo", executed_at=NOW, execution_ref="x") + + +def test_mark_executed_unknown_id_raises() -> None: + with pytest.raises(ProposalNotFoundError): + mark_executed([], "context-pr:ghost", executed_at=NOW, execution_ref="x")