diff --git a/aicp/__main__.py b/aicp/__main__.py index 1019b9a..1ae7181 100644 --- a/aicp/__main__.py +++ b/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": diff --git a/aicp/execution.py b/aicp/execution.py new file mode 100644 index 0000000..b46413e --- /dev/null +++ b/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 diff --git a/docs/CLI.md b/docs/CLI.md index b531ea1..39c1465 100644 --- a/docs/CLI.md +++ b/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. diff --git a/tests/test_execution.py b/tests/test_execution.py new file mode 100644 index 0000000..afdb2a9 --- /dev/null +++ b/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()