Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
11 changes: 9 additions & 2 deletions aicp/execution.py
Original file line number Diff line number Diff line change
Expand Up @@ -6,6 +6,7 @@
import time

from .task_lifecycle import show_task
from .scope_audit import git_snapshot, audit_scope
from .verification import run_checks, validate_checks


Expand Down Expand Up @@ -37,6 +38,9 @@ def run_task(project, name, tool, argv, checks_path, output, timeout=300):
if output.exists() or output.is_symlink():
raise FileExistsError("Refusing to overwrite execution report")
task, task_path, task_bytes, checks_bytes, trusted_checks = _load_trusted_inputs(project, name, checks_path)
baseline = git_snapshot(project)
if baseline:
raise ValueError('Scope audit requires a clean Git worktree before execution')
started = time.monotonic()
try:
proc = subprocess.run(argv, cwd=project, capture_output=True, timeout=timeout, check=False)
Expand Down Expand Up @@ -64,20 +68,23 @@ def run_task(project, name, tool, argv, checks_path, output, timeout=300):
and task_path.read_bytes() == task_bytes
and checks_path.read_bytes() == checks_bytes
)
checks = run_checks(project, trusted_checks) if agent["passed"] and inputs_unchanged else None
after = git_snapshot(project)
scope = audit_scope(baseline, after, task["scope"]["allowed_paths"])
checks = run_checks(project, trusted_checks) if agent["passed"] and inputs_unchanged and scope["passed"] else None
report = {
"schema_version": 1,
"task": name,
"objective": task["objective"],
"agent": agent,
"verification": checks,
"scope_audit": scope,
"trusted_inputs": {
"task_sha256": hashlib.sha256(task_bytes).hexdigest(),
"checks_sha256": hashlib.sha256(checks_bytes).hexdigest(),
"unchanged_after_agent": inputs_unchanged,
"verification_source": "pre-agent-snapshot",
},
"passed": bool(agent["passed"] and inputs_unchanged and checks and checks["passed"]),
"passed": bool(agent["passed"] and inputs_unchanged and scope["passed"] and checks and checks["passed"]),
"security_boundary": "trusted-host-execution-no-sandbox",
"scope_enforced": False,
}
Expand Down
51 changes: 51 additions & 0 deletions aicp/scope_audit.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,51 @@
"""Best-effort Git working-tree change auditing (not filesystem enforcement)."""
import subprocess
from pathlib import PurePosixPath


def git_snapshot(project):
proc = subprocess.run(
["git", "status", "--porcelain=v1", "-z", "--untracked-files=all"],
cwd=project, capture_output=True, check=False, timeout=30,
)
if proc.returncode:
raise ValueError("Git status failed; scope audit requires a Git worktree")
entries = proc.stdout.split(b"\0")
paths = set()
i = 0
while i < len(entries):
entry = entries[i]
i += 1
if not entry:
continue
if len(entry) < 4:
raise ValueError("Unexpected Git status output")
path = entry[3:].decode("utf-8", errors="surrogateescape")
paths.add(path)
if entry[:1] in (b"R", b"C") or entry[1:2] in (b"R", b"C"):
if i >= len(entries) or not entries[i]:
raise ValueError("Incomplete Git rename entry")
paths.add(entries[i].decode("utf-8", errors="surrogateescape"))
i += 1
return paths


def audit_scope(before, after, allowed_paths):
added = sorted(after - before)
allowed = []
outside = []
for path in added:
parts = PurePosixPath(path).parts
match = any(
path == prefix.rstrip("/") or path.startswith(prefix.rstrip("/") + "/")
for prefix in allowed_paths
)
(allowed if match else outside).append(path)
return {
"newly_reported_paths": added,
"within_allowed_paths": allowed,
"outside_allowed_paths": outside,
"preexisting_changed_paths": sorted(before),
"passed": not outside,
"limitations": "git-status-delta-only; does not detect modifications to already dirty files, committed changes, ignored files, or reverted writes",
}
4 changes: 4 additions & 0 deletions docs/CLI.md
Original file line number Diff line number Diff line change
Expand Up @@ -54,3 +54,7 @@ aicp run --path /path/to/project --name fix-retry \
```

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. Task and check definitions are now validated and snapshotted **before** agent execution; changes to either input after the agent runs cause the run to fail without executing verification. Checks run from the pre-agent command snapshot. This does **not** make the underlying test code, dependencies, environment, or workspace immutable, and does not protect against concurrent filesystem races. Independently review the resulting code and tests 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.

### Git workspace scope audit

Supervised `aicp run` now requires a **clean Git worktree** at startup and records newly changed tracked/untracked paths after the agent exits. If any reported path is outside the task's `allowed_paths`, verification is skipped and the run fails. The JSON report includes `scope_audit` with paths and status. This is a **best-effort audit, not permission enforcement**: Git status cannot detect ignored files, committed changes, writes that are reverted, or writes outside the repository. The agent retains full host permissions. Use a disposable isolated environment, and independently review diffs and tests.
22 changes: 22 additions & 0 deletions tests/test_execution.py
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,9 @@ def setUp(self):
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"
git_patch = patch("aicp.execution.git_snapshot", return_value=set())
git_patch.start()
self.addCleanup(git_patch.stop)

def test_successful_agent_and_checks(self):
with patch("aicp.execution.subprocess.run") as run:
Expand Down Expand Up @@ -69,6 +72,25 @@ def test_invalid_checks_fail_before_agent_execution(self):
run_task(self.project, "fix-retry", "codex", ["codex"], self.checks, self.output)
agent.assert_not_called()

def test_out_of_scope_changes_block_verification(self):
with patch("aicp.execution.git_snapshot", side_effect=[set(), {"src/fix.py", "secrets.txt"}]):
with patch("aicp.execution.subprocess.run") as agent:
agent.return_value.returncode = 0
agent.return_value.stdout = b""
agent.return_value.stderr = b""
with patch("aicp.execution.run_checks") as checks:
result = run_task(self.project, "fix-retry", "codex", ["codex"], self.checks, self.output)
checks.assert_not_called()
self.assertFalse(result["passed"])
self.assertEqual(result["scope_audit"]["outside_allowed_paths"], ["secrets.txt"])

def test_dirty_worktree_rejected_before_agent(self):
with patch("aicp.execution.git_snapshot", return_value={"existing.txt"}):
with patch("aicp.execution.subprocess.run") as agent:
with self.assertRaisesRegex(ValueError, "clean Git worktree"):
run_task(self.project, "fix-retry", "codex", ["codex"], self.checks, self.output)
agent.assert_not_called()

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)
Expand Down
22 changes: 22 additions & 0 deletions tests/test_scope_audit.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,22 @@
"""Scope auditing unit tests."""
import unittest
from aicp.scope_audit import audit_scope


class ScopeAuditTests(unittest.TestCase):
def test_prefixes_are_directory_boundaries(self):
result = audit_scope(set(), {"src/a.py", "src-extra/b.py", "tests/test_a.py"}, ["src/", "tests/"])
self.assertEqual(result["outside_allowed_paths"], ["src-extra/b.py"])
self.assertFalse(result["passed"])

def test_exact_file_allowed(self):
result = audit_scope(set(), {"README.md"}, ["README.md"])
self.assertTrue(result["passed"])

def test_clean_delta(self):
result = audit_scope(set(), set(), ["src/"])
self.assertTrue(result["passed"])


if __name__ == "__main__":
unittest.main()
Loading