feat: add explicitly approved supervised aicp run

aicp/__main__.py

@@ -8,6 +8,7 @@
 from . import __version__
 from .tasks import create_task
 from .task_lifecycle import show_task, list_tasks
+from .execution import run_task
 from .verification import verify as run_verification
 from .entrypoints import ENTRYPOINTS, install_entrypoint
 from .planning import plan_init
@@ -115,6 +116,15 @@ def main(argv=None):
         command.add_argument("--path", type=Path, default=Path("."))
         if action != "list":
             command.add_argument("--name", required=True)
+    run_parser = commands.add_parser("run", help="Run approved local agent and checks (not sandboxed)")
+    run_parser.add_argument("--path", type=Path, default=Path("."))
+    run_parser.add_argument("--name", required=True)
+    run_parser.add_argument("--tool", choices=("codex", "claude", "gemini"), required=True)
+    run_parser.add_argument("--checks", type=Path, required=True)
+    run_parser.add_argument("--output", type=Path, required=True)
+    run_parser.add_argument("--timeout", type=int, default=300)
+    run_parser.add_argument("--approve-host-execution", action="store_true", required=True)
+    run_parser.add_argument("agent_argv", nargs=argparse.REMAINDER)
     verify_parser = commands.add_parser("verify", help="Run explicit project checks")
     verify_parser.add_argument("--path", type=Path, default=Path("."))
     verify_parser.add_argument("--checks", type=Path, required=True)
@@ -154,6 +164,12 @@ def main(argv=None):
                 if not check["passed"]:
                     print(f"  Check {index}: {check['status']} (exit={check['exit_code']})")
             return 0 if result["passed"] else 1
+        elif args.command == "run":
+            validate_config(args.path)
+            argv = args.agent_argv[1:] if args.agent_argv and args.agent_argv[0] == "--" else args.agent_argv
+            report = run_task(args.path, args.name, args.tool, argv, args.checks, args.output, args.timeout)
+            print(f"Execution {'passed' if report['passed'] else 'failed'}: {args.output}")
+            return 0 if report["passed"] else 1
         elif args.command == "task":
             validate_config(args.path)
             if args.task_command == "create":

aicp/execution.py

@@ -0,0 +1,56 @@
+"""Explicitly approved, non-sandboxed task execution and verification."""
+import hashlib
+import json
+from pathlib import Path
+import subprocess
+import time
+
+from .task_lifecycle import show_task
+from .verification import verify
+
+
+def run_task(project, name, tool, argv, checks_path, output, timeout=300):
+    if tool not in ("codex", "claude", "gemini"):
+        raise ValueError("Only explicitly supported local agent CLIs may run")
+    if not argv or argv[0] != tool or not all(isinstance(a, str) and a for a in argv):
+        raise ValueError("Agent argv must begin with selected tool")
+    if type(timeout) is not int or not 1 <= timeout <= 1800:
+        raise ValueError("Agent timeout must be between 1 and 1800 seconds")
+    if output.exists() or output.is_symlink():
+        raise FileExistsError("Refusing to overwrite execution report")
+    task = show_task(project, name)
+    started = time.monotonic()
+    try:
+        proc = subprocess.run(argv, cwd=project, capture_output=True, timeout=timeout, check=False)
+        status, exit_code, stdout, stderr = "completed", proc.returncode, proc.stdout, proc.stderr
+    except subprocess.TimeoutExpired as exc:
+        status, exit_code, stdout, stderr = "timeout", None, exc.stdout or b"", exc.stderr or b""
+    except OSError as exc:
+        status, exit_code, stdout, stderr = "launch_error", None, b"", str(exc).encode()
+    agent = {
+        "tool": tool,
+        "status": status,
+        "exit_code": exit_code,
+        "duration_seconds": round(time.monotonic() - started, 3),
+        "stdout_sha256": hashlib.sha256(stdout).hexdigest(),
+        "stderr_sha256": hashlib.sha256(stderr).hexdigest(),
+        "stdout_bytes": len(stdout),
+        "stderr_bytes": len(stderr),
+        "passed": status == "completed" and exit_code == 0,
+    }
+    # Never execute verification if agent execution itself failed.
+    checks = verify(project, checks_path) if agent["passed"] else None
+    report = {
+        "schema_version": 1,
+        "task": name,
+        "objective": task["objective"],
+        "agent": agent,
+        "verification": checks,
+        "passed": bool(agent["passed"] and checks and checks["passed"]),
+        "security_boundary": "trusted-host-execution-no-sandbox",
+        "scope_enforced": False,
+    }
+    with output.open("x", encoding="utf-8") as handle:
+        json.dump(report, handle, indent=2)
+        handle.write("\n")
+    return report

docs/CLI.md

@@ -41,3 +41,16 @@ Task creation rejects symlinked `tasks/` directories, non-directory destination
 ## Review existing tasks
 
 Use `python3 -m aicp task list --path /path/to/project` to list and validate local task drafts, `python3 -m aicp task show --path /path/to/project --name fix-retry` to print a task, and `python3 -m aicp task validate --path /path/to/project --name fix-retry` for a machine-readable validity result. Invalid or symlinked task files are rejected. Validation checks structure and basic allowed-path syntax; it **does not enforce agent permissions, inspect whether a task is completed, or execute acceptance criteria**.
+
+## Supervised agent execution (experimental)
+
+The `aicp run` command invokes an explicitly selected local Codex, Claude, or Gemini CLI and, **only after a successful agent exit**, runs trusted checks. It requires `--approve-host-execution`. Example, in a disposable isolated environment with the vendor CLI already installed and authenticated:
+
+```sh
+aicp run --path /path/to/project --name fix-retry \
+  --tool codex --checks /path/to/project/checks.json \
+  --output /tmp/aicp-run.json --approve-host-execution \
+  -- codex exec "Read tasks/fix-retry.json and implement the approved task"
+```
+
+The report records agent exit status, output hashes, verification results, and whether both stages passed. This is **not a sandbox**, and task allowed paths are **not enforced**. The agent executes on your host with your permissions and can change the checks file before verification; independently review the resulting code and checks before trusting the outcome. Do not pass secrets in command arguments, run on sensitive workspaces, or treat a passing report as permission to merge. CI uses mocked agent execution only. The command does not create branches, PRs, or merges.

tests/test_execution.py

@@ -0,0 +1,53 @@
+"""Supervised execution orchestration tests; no real agent is launched."""
+import json
+from pathlib import Path
+import tempfile
+import unittest
+from unittest.mock import patch
+
+from aicp.execution import run_task
+from aicp.tasks import create_task
+
+
+class RunTaskTests(unittest.TestCase):
+    def setUp(self):
+        self.temp = tempfile.TemporaryDirectory()
+        self.addCleanup(self.temp.cleanup)
+        self.project = Path(self.temp.name)
+        create_task(self.project, "fix-retry", "Fix retry", ["src/"], ["Tests pass"])
+        self.checks = self.project / "checks.json"
+        self.checks.write_text('{"schema_version":1,"checks":[["python3","--version"]]}', encoding="utf-8")
+        self.output = self.project / "report.json"
+
+    def test_successful_agent_and_checks(self):
+        with patch("aicp.execution.subprocess.run") as run:
+            run.return_value.returncode = 0
+            run.return_value.stdout = b"ok"
+            run.return_value.stderr = b""
+            result = run_task(self.project, "fix-retry", "codex", ["codex", "exec", "example"], self.checks, self.output)
+        self.assertTrue(result["passed"])
+        self.assertFalse(result["scope_enforced"])
+        self.assertEqual(json.loads(self.output.read_text(encoding="utf-8"))["task"], "fix-retry")
+
+    def test_agent_failure_skips_checks(self):
+        with patch("aicp.execution.subprocess.run") as run:
+            run.return_value.returncode = 1
+            run.return_value.stdout = b""
+            run.return_value.stderr = b"error"
+            result = run_task(self.project, "fix-retry", "codex", ["codex", "exec", "example"], self.checks, self.output)
+        self.assertFalse(result["passed"])
+        self.assertIsNone(result["verification"])
+
+    def test_invalid_agent_command_rejected(self):
+        with self.assertRaises(ValueError):
+            run_task(self.project, "fix-retry", "codex", ["sh", "-c", "true"], self.checks, self.output)
+
+    def test_report_never_overwritten(self):
+        self.output.write_text("existing", encoding="utf-8")
+        with self.assertRaises(FileExistsError):
+            run_task(self.project, "fix-retry", "codex", ["codex"], self.checks, self.output)
+        self.assertEqual(self.output.read_text(encoding="utf-8"), "existing")
+
+
+if __name__ == "__main__":
+    unittest.main()