weichen
8b30353c4d
Why: Implementation changes must be mechanically tied to a successful CLI transport instead of relying on agents to follow prose. Evidence: Admission tests went RED without apply/admit; 10 admission, 41 agent-exec, and 62 task tests now pass with exact staged-content matching. Impact: Adds validated transport fast-forwarding and rejects unproven implementation commits while exempting task cards and approved plan/spec paths. Refs: feature/agent-cli-enforcement
447 lines
17 KiB
Python
447 lines
17 KiB
Python
"""Provenance admission: apply a success transport, or admit exact staged paths."""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import shutil
|
|
import subprocess
|
|
import tempfile
|
|
import unittest
|
|
import uuid
|
|
from pathlib import Path
|
|
|
|
from scripts import agent_exec
|
|
|
|
|
|
SOURCE = Path(__file__).resolve().parents[1]
|
|
|
|
|
|
def _git(repo: Path, *args: str, check: bool = True) -> subprocess.CompletedProcess[str]:
|
|
return subprocess.run(
|
|
["git", "-C", str(repo), *args],
|
|
check=check,
|
|
capture_output=True,
|
|
text=True,
|
|
)
|
|
|
|
|
|
class AgentExecAdmissionTests(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
self.tmp = tempfile.TemporaryDirectory(prefix="agent-exec-admit-")
|
|
self.addCleanup(self.tmp.cleanup)
|
|
self.root = Path(self.tmp.name)
|
|
self.repo = self.root / "repo"
|
|
self.repo.mkdir()
|
|
_git(self.repo, "init", "-b", "main")
|
|
_git(self.repo, "config", "user.email", "agent-exec@example.com")
|
|
_git(self.repo, "config", "user.name", "Agent Exec")
|
|
(self.repo / "README.md").write_text("fixture\n", encoding="utf-8")
|
|
_git(self.repo, "add", "README.md")
|
|
_git(self.repo, "commit", "-m", "init")
|
|
_git(self.repo, "checkout", "-b", "feature/admission-test")
|
|
self.head = _git(self.repo, "rev-parse", "HEAD").stdout.strip()
|
|
|
|
def _cli(self, *args: str) -> subprocess.CompletedProcess[str]:
|
|
env = os.environ.copy()
|
|
env["PYTHONDONTWRITEBYTECODE"] = "1"
|
|
env["PYTHONPATH"] = str(SOURCE)
|
|
return subprocess.run(
|
|
["python3", "-m", "scripts.agent_exec", *args],
|
|
cwd=str(self.root),
|
|
capture_output=True,
|
|
text=True,
|
|
env=env,
|
|
)
|
|
|
|
def _state(self) -> dict[str, str]:
|
|
return {
|
|
"head": _git(self.repo, "rev-parse", "HEAD").stdout.strip(),
|
|
"branch": _git(self.repo, "rev-parse", "--abbrev-ref", "HEAD").stdout.strip(),
|
|
"status": _git(
|
|
self.repo, "status", "--porcelain", "--untracked-files=all"
|
|
).stdout,
|
|
"index": _git(self.repo, "ls-files", "--stage").stdout,
|
|
"readme": (self.repo / "README.md").read_text(encoding="utf-8"),
|
|
}
|
|
|
|
def _transport_commit(
|
|
self,
|
|
files: dict[str, str],
|
|
*,
|
|
parent: str | None = None,
|
|
ref: str | None = None,
|
|
) -> tuple[str, str]:
|
|
parent = parent or _git(self.repo, "rev-parse", "HEAD").stdout.strip()
|
|
ref = ref or f"refs/agent-exec/{uuid.uuid4().hex}/attempt-1"
|
|
index = self.root / f"index-{uuid.uuid4().hex}"
|
|
env = os.environ.copy()
|
|
env["GIT_INDEX_FILE"] = str(index)
|
|
subprocess.run(
|
|
["git", "-C", str(self.repo), "read-tree", parent],
|
|
check=True,
|
|
env=env,
|
|
capture_output=True,
|
|
text=True,
|
|
)
|
|
for rel, content in files.items():
|
|
blob = subprocess.run(
|
|
["git", "-C", str(self.repo), "hash-object", "-w", "--stdin"],
|
|
input=content,
|
|
text=True,
|
|
check=True,
|
|
capture_output=True,
|
|
).stdout.strip()
|
|
subprocess.run(
|
|
[
|
|
"git",
|
|
"-C",
|
|
str(self.repo),
|
|
"update-index",
|
|
"--add",
|
|
"--cacheinfo",
|
|
f"100644,{blob},{rel}",
|
|
],
|
|
check=True,
|
|
env=env,
|
|
capture_output=True,
|
|
text=True,
|
|
)
|
|
tree = subprocess.run(
|
|
["git", "-C", str(self.repo), "write-tree"],
|
|
check=True,
|
|
env=env,
|
|
capture_output=True,
|
|
text=True,
|
|
).stdout.strip()
|
|
commit = subprocess.run(
|
|
[
|
|
"git",
|
|
"-C",
|
|
str(self.repo),
|
|
"-c",
|
|
"user.name=agent-exec-transport",
|
|
"-c",
|
|
"user.email=agent-exec-transport@localhost.invalid",
|
|
"commit-tree",
|
|
tree,
|
|
"-p",
|
|
parent,
|
|
"-m",
|
|
"agent-exec transport",
|
|
],
|
|
check=True,
|
|
capture_output=True,
|
|
text=True,
|
|
).stdout.strip()
|
|
_git(self.repo, "update-ref", ref, commit)
|
|
index.unlink(missing_ok=True)
|
|
return commit, ref
|
|
|
|
def _receipt(
|
|
self,
|
|
*,
|
|
parent: str,
|
|
ref: str | None,
|
|
paths: list[str],
|
|
status: str = "success",
|
|
repo: str | None = None,
|
|
reason: str | None = None,
|
|
transport_parent: str | None = None,
|
|
omit_transport: bool = False,
|
|
) -> Path:
|
|
data: dict[str, object] = {
|
|
"schema": agent_exec.RECEIPT_SCHEMA,
|
|
"repo": repo if repo is not None else str(self.repo.resolve()),
|
|
"terminal_status": status,
|
|
"reason": reason,
|
|
"base": {
|
|
"head": parent,
|
|
"tree": _git(self.repo, "rev-parse", f"{parent}^{{tree}}").stdout.strip(),
|
|
"branch": "feature/admission-test",
|
|
},
|
|
"changed_paths": paths,
|
|
}
|
|
if not omit_transport:
|
|
data["transport"] = {
|
|
"ref": ref,
|
|
"parent": parent if transport_parent is None else transport_parent,
|
|
"attempt_id": "attempt-1",
|
|
}
|
|
path = self.root / f"receipt-{uuid.uuid4().hex}.json"
|
|
path.write_text(json.dumps(data), encoding="utf-8")
|
|
return path
|
|
|
|
def _history(
|
|
self,
|
|
*,
|
|
parent: str,
|
|
ref: str,
|
|
paths: list[str],
|
|
status: str = "success",
|
|
repo: str | None = None,
|
|
) -> None:
|
|
record = {
|
|
"schema": agent_exec.HISTORY_SCHEMA,
|
|
"terminal_status": status,
|
|
"repo": repo if repo is not None else str(self.repo.resolve()),
|
|
"changed_paths": paths,
|
|
"transport": {"ref": ref, "parent": parent},
|
|
}
|
|
common = Path(
|
|
_git(
|
|
self.repo,
|
|
"rev-parse",
|
|
"--path-format=absolute",
|
|
"--git-common-dir",
|
|
).stdout.strip()
|
|
)
|
|
dest = common / "agent-exec"
|
|
dest.mkdir(mode=0o700, exist_ok=True)
|
|
with (dest / "execution-history.jsonl").open("a", encoding="utf-8") as handle:
|
|
handle.write(json.dumps(record, sort_keys=True) + "\n")
|
|
|
|
def _stage(self, rel: str, content: str) -> None:
|
|
path = self.repo / rel
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
path.write_text(content, encoding="utf-8")
|
|
_git(self.repo, "add", "--", rel)
|
|
|
|
def test_apply_fast_forwards_clean_matching_transport(self) -> None:
|
|
commit, ref = self._transport_commit({"src/marker.txt": "shipped\n"})
|
|
receipt = self._receipt(
|
|
parent=self.head, ref=ref, paths=["src/marker.txt"]
|
|
)
|
|
proc = self._cli(
|
|
"apply", "--repo", str(self.repo), "--receipt", str(receipt)
|
|
)
|
|
self.assertEqual(proc.returncode, agent_exec.EXIT_SUCCESS, proc.stderr + proc.stdout)
|
|
self.assertIn("apply: fast-forwarded", proc.stdout)
|
|
self.assertEqual(_git(self.repo, "rev-parse", "HEAD").stdout.strip(), commit)
|
|
self.assertEqual(
|
|
_git(self.repo, "rev-parse", "feature/admission-test").stdout.strip(),
|
|
commit,
|
|
)
|
|
self.assertEqual((self.repo / "src/marker.txt").read_text(encoding="utf-8"), "shipped\n")
|
|
self.assertEqual(_git(self.repo, "status", "--porcelain").stdout, "")
|
|
parents = _git(self.repo, "rev-list", "--parents", "-n", "1", "HEAD").stdout.split()
|
|
self.assertEqual(parents, [commit, self.head])
|
|
|
|
def test_apply_rejects_without_changing_state(self) -> None:
|
|
_commit, ref = self._transport_commit({"src/marker.txt": "shipped\n"})
|
|
valid = dict(parent=self.head, ref=ref, paths=["src/marker.txt"])
|
|
|
|
def reject(label: str, receipt: Path) -> None:
|
|
before = self._state()
|
|
proc = self._cli(
|
|
"apply", "--repo", str(self.repo), "--receipt", str(receipt)
|
|
)
|
|
self.assertEqual(
|
|
proc.returncode, agent_exec.EXIT_BLOCKED, proc.stderr + proc.stdout
|
|
)
|
|
self.assertIn(label, proc.stderr)
|
|
self.assertNotIn("apply: fast-forwarded", proc.stdout)
|
|
self.assertEqual(self._state(), before)
|
|
|
|
with self.subTest(label="stale HEAD"):
|
|
(self.repo / "advance.txt").write_text("advance\n", encoding="utf-8")
|
|
_git(self.repo, "add", "advance.txt")
|
|
_git(self.repo, "commit", "-m", "advance")
|
|
reject("stale HEAD", self._receipt(**valid))
|
|
_git(self.repo, "reset", "--hard", self.head)
|
|
|
|
with self.subTest(label="dirty worktree"):
|
|
(self.repo / "README.md").write_text("dirty\n", encoding="utf-8")
|
|
reject("dirty worktree", self._receipt(**valid))
|
|
_git(self.repo, "checkout", "--", "README.md")
|
|
|
|
with self.subTest(label="non-success receipt"):
|
|
reject(
|
|
"non-success receipt",
|
|
self._receipt(**valid, status="failed-unknown"),
|
|
)
|
|
|
|
with self.subTest(label="missing transport ref"):
|
|
reject(
|
|
"missing transport",
|
|
self._receipt(
|
|
parent=self.head,
|
|
ref="refs/agent-exec/missing/attempt",
|
|
paths=["src/marker.txt"],
|
|
),
|
|
)
|
|
|
|
with self.subTest(label="omitted transport"):
|
|
reject(
|
|
"missing transport",
|
|
self._receipt(
|
|
parent=self.head,
|
|
ref=None,
|
|
paths=["src/marker.txt"],
|
|
omit_transport=True,
|
|
),
|
|
)
|
|
|
|
with self.subTest(label="mismatched transport"):
|
|
side = _git(
|
|
self.repo,
|
|
"commit-tree",
|
|
_git(self.repo, "rev-parse", "HEAD^{tree}").stdout.strip(),
|
|
"-p",
|
|
self.head,
|
|
"-m",
|
|
"side",
|
|
).stdout.strip()
|
|
_mismatch_commit, mismatch_ref = self._transport_commit(
|
|
{"src/marker.txt": "other\n"}, parent=side
|
|
)
|
|
reject(
|
|
"mismatched transport",
|
|
self._receipt(
|
|
parent=self.head, ref=mismatch_ref, paths=["src/marker.txt"]
|
|
),
|
|
)
|
|
|
|
with self.subTest(label="receipt repo mismatch"):
|
|
reject(
|
|
"receipt repo mismatch",
|
|
self._receipt(
|
|
**valid, repo=str((self.root / "other-repo").resolve())
|
|
),
|
|
)
|
|
|
|
def test_apply_redacts_secret_shaped_receipt_text(self) -> None:
|
|
_commit, ref = self._transport_commit({"src/marker.txt": "shipped\n"})
|
|
secret = "sk-admissionsecret"
|
|
receipt = self._receipt(
|
|
parent=self.head,
|
|
ref=ref,
|
|
paths=["src/marker.txt"],
|
|
status="failed-unknown",
|
|
reason=secret,
|
|
)
|
|
before = self._state()
|
|
proc = self._cli("apply", "--repo", str(self.repo), "--receipt", str(receipt))
|
|
combined = proc.stdout + proc.stderr
|
|
self.assertEqual(proc.returncode, agent_exec.EXIT_BLOCKED, combined)
|
|
self.assertIn("non-success receipt", proc.stderr)
|
|
self.assertNotIn(secret, combined)
|
|
self.assertIn("***", combined)
|
|
self.assertEqual(self._state(), before)
|
|
|
|
def test_admit_rejects_staged_paths_without_provenance(self) -> None:
|
|
self._stage("src/extra.txt", "unproven\n")
|
|
before = self._state()
|
|
proc = self._cli("admit", "--repo", str(self.repo))
|
|
self.assertEqual(proc.returncode, agent_exec.EXIT_BLOCKED, proc.stderr + proc.stdout)
|
|
self.assertIn("admit: no provenance", proc.stderr)
|
|
self.assertEqual(self._state(), before)
|
|
|
|
def test_admit_accepts_exact_transport_paths(self) -> None:
|
|
_commit, ref = self._transport_commit(
|
|
{"src/marker.txt": "shipped\n", "src/other.txt": "also\n"}
|
|
)
|
|
self._history(
|
|
parent=self.head,
|
|
ref=ref,
|
|
paths=["src/marker.txt", "src/other.txt"],
|
|
)
|
|
self._stage("src/marker.txt", "shipped\n")
|
|
self._stage("src/other.txt", "also\n")
|
|
self._stage(".task/feature-admission.md", "host card\n")
|
|
before = self._state()
|
|
proc = self._cli("admit", "--repo", str(self.repo))
|
|
self.assertEqual(proc.returncode, agent_exec.EXIT_SUCCESS, proc.stderr + proc.stdout)
|
|
self.assertIn("admit: ok", proc.stdout)
|
|
self.assertEqual(self._state(), before)
|
|
|
|
self._git_reset_index()
|
|
self._stage("src/marker.txt", "shipped\n")
|
|
partial = self._cli("admit", "--repo", str(self.repo))
|
|
self.assertEqual(partial.returncode, agent_exec.EXIT_BLOCKED, partial.stderr)
|
|
self.assertIn("admit: no provenance", partial.stderr)
|
|
|
|
def test_admit_rejects_different_content_at_transport_path(self) -> None:
|
|
_commit, ref = self._transport_commit({"src/marker.txt": "shipped\n"})
|
|
self._history(parent=self.head, ref=ref, paths=["src/marker.txt"])
|
|
self._stage("src/marker.txt", "rewritten\n")
|
|
before = self._state()
|
|
proc = self._cli("admit", "--repo", str(self.repo))
|
|
self.assertEqual(proc.returncode, agent_exec.EXIT_BLOCKED, proc.stderr + proc.stdout)
|
|
self.assertIn("admit: no provenance", proc.stderr)
|
|
self.assertEqual(self._state(), before)
|
|
|
|
def test_admit_accepts_host_managed_exemptions(self) -> None:
|
|
self._stage(".task/feature-admission.md", "card\n")
|
|
self._stage("docs/superpowers/specs/admission.md", "spec\n")
|
|
self._stage("docs/superpowers/plans/admission.md", "plan\n")
|
|
before = self._state()
|
|
proc = self._cli("admit", "--repo", str(self.repo))
|
|
self.assertEqual(proc.returncode, agent_exec.EXIT_SUCCESS, proc.stderr + proc.stdout)
|
|
self.assertIn("admit: ok", proc.stdout)
|
|
self.assertEqual(self._state(), before)
|
|
|
|
self._stage("docs/superpowers/specs-extra/not-exempt.md", "nope\n")
|
|
rejected = self._cli("admit", "--repo", str(self.repo))
|
|
self.assertEqual(rejected.returncode, agent_exec.EXIT_BLOCKED, rejected.stderr)
|
|
self.assertIn("admit: no provenance", rejected.stderr)
|
|
|
|
def test_precommit_rejects_unproven_paths_after_branch_checks(self) -> None:
|
|
self._install_project_hook()
|
|
self._stage("src/extra.txt", "unproven\n")
|
|
proc = self._hook()
|
|
combined = proc.stdout + proc.stderr
|
|
self.assertNotEqual(proc.returncode, 0, combined)
|
|
self.assertIn("admit: no provenance", combined)
|
|
|
|
_git(self.repo, "checkout", "-B", "main", self.head)
|
|
self._stage("src/extra.txt", "unproven\n")
|
|
blocked = self._hook()
|
|
text = blocked.stdout + blocked.stderr
|
|
self.assertNotEqual(blocked.returncode, 0, text)
|
|
self.assertIn("集成分支", text)
|
|
self.assertNotIn("admit:", text)
|
|
|
|
def test_precommit_accepts_host_managed_exemptions(self) -> None:
|
|
self._install_project_hook()
|
|
self._stage("docs/superpowers/plans/admission.md", "plan\n")
|
|
proc = self._hook()
|
|
self.assertEqual(proc.returncode, 0, proc.stderr + proc.stdout)
|
|
self.assertIn("admit: ok", proc.stdout)
|
|
|
|
def test_make_agent_exec_apply_requires_receipt(self) -> None:
|
|
env = os.environ.copy()
|
|
env.pop("RECEIPT", None)
|
|
proc = subprocess.run(
|
|
["make", "agent-exec-apply"],
|
|
cwd=str(SOURCE),
|
|
capture_output=True,
|
|
text=True,
|
|
env=env,
|
|
)
|
|
combined = proc.stdout + proc.stderr
|
|
self.assertNotEqual(proc.returncode, 0, combined)
|
|
self.assertIn("RECEIPT", combined)
|
|
self.assertNotIn("apply: fast-forwarded", combined)
|
|
|
|
def _git_reset_index(self) -> None:
|
|
_git(self.repo, "reset", "--hard", self.head)
|
|
_git(self.repo, "clean", "-fd")
|
|
|
|
def _install_project_hook(self) -> None:
|
|
dest = self.repo / "scripts"
|
|
hooks = dest / "hooks"
|
|
hooks.mkdir(parents=True)
|
|
for name in ("agent_exec.py", "agent_exec_config.py", "agent_exec_adapters.py"):
|
|
shutil.copy(SOURCE / "scripts" / name, dest / name)
|
|
hook = hooks / "pre-commit"
|
|
shutil.copy(SOURCE / "scripts" / "hooks" / "pre-commit", hook)
|
|
hook.chmod(0o755)
|
|
|
|
def _hook(self) -> subprocess.CompletedProcess[str]:
|
|
return subprocess.run(
|
|
["bash", str(self.repo / "scripts" / "hooks" / "pre-commit")],
|
|
cwd=str(self.repo),
|
|
capture_output=True,
|
|
text=True,
|
|
)
|