From 8b30353c4d0c978d38b6cbd581abe079e65aad60 Mon Sep 17 00:00:00 2001 From: weichen Date: Thu, 15 Oct 2026 17:12:33 +0000 Subject: [PATCH] Enforce execution-agent provenance at integration 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 --- Makefile | 11 +- scripts/agent_exec.py | 332 ++++++++++++++++++++- scripts/hooks/pre-commit | 3 +- tests/test_agent_exec_admission.py | 446 +++++++++++++++++++++++++++++ 4 files changed, 789 insertions(+), 3 deletions(-) create mode 100644 tests/test_agent_exec_admission.py diff --git a/Makefile b/Makefile index f1246f76..158dd4df 100644 --- a/Makefile +++ b/Makefile @@ -13,7 +13,7 @@ PLUGINS := plugins COMBO := core=$(shell git -C $(CORE) rev-parse --short HEAD 2>/dev/null)+backend=$(shell git -C $(BACKEND) rev-parse --short HEAD 2>/dev/null)+frontend=$(shell git -C $(FRONTEND) rev-parse --short HEAD 2>/dev/null)+ui=$(shell git -C $(UI) rev-parse --short HEAD 2>/dev/null)+agentcmd=$(shell git -C deps/agent-command rev-parse --short HEAD 2>/dev/null)+documents=$(shell git -C modules/documents rev-parse --short HEAD 2>/dev/null)+native=$(shell git -C deps/agent-native rev-parse --short HEAD 2>/dev/null)+squad=$(shell git -C deps/agent-squad rev-parse --short HEAD 2>/dev/null)+intent=$(shell git -C deps/agent-intent rev-parse --short HEAD 2>/dev/null) VERSION := $(shell git describe --tags --always --dirty 2>/dev/null)+$(COMBO) -.PHONY: help sync status build frontend svc svc-list svc-info svc-commits svc-history svc-update svc-recover svc-watchdog test-backend test-backend-pg test-multi-instance test-lifecycle test-frontend test-ui test-core test-agentcmd test-pins test-libs dev pin clean setup precheck task-start task-resume task-sync task-close task-finish facts-verify evidence version agent-exec-validate +.PHONY: help sync status build frontend svc svc-list svc-info svc-commits svc-history svc-update svc-recover svc-watchdog test-backend test-backend-pg test-multi-instance test-lifecycle test-frontend test-ui test-core test-agentcmd test-pins test-libs dev pin clean setup precheck task-start task-resume task-sync task-close task-finish facts-verify evidence version agent-exec-validate agent-exec-apply agent-exec-admit version: @echo "agent-system $(VERSION)" @@ -78,10 +78,19 @@ help: @echo "make pin 推进 submodule 指针到各仓远端 main（提交前须 make test 通过）" @echo "make status submodule 状态" @echo "make agent-exec-validate 校验 AGENT-EXECUTION.md（agent_exec validate）" + @echo "make agent-exec-apply RECEIPT= 快进当前分支到成功收据的 transport" + @echo "make agent-exec-admit 校验暂存路径具备 transport 来源或属于 host 豁免" agent-exec-validate: PYTHONDONTWRITEBYTECODE=1 python3 scripts/agent_exec.py validate --repo . +agent-exec-apply: + @if [ -z "$(RECEIPT)" ]; then echo "usage: make agent-exec-apply RECEIPT="; exit 2; fi + PYTHONDONTWRITEBYTECODE=1 python3 scripts/agent_exec.py apply --repo . --receipt "$(RECEIPT)" + +agent-exec-admit: + PYTHONDONTWRITEBYTECODE=1 python3 scripts/agent_exec.py admit --repo . + sync: git submodule update --init --recursive diff --git a/scripts/agent_exec.py b/scripts/agent_exec.py index 7f317f7f..4f98e4c2 100644 --- a/scripts/agent_exec.py +++ b/scripts/agent_exec.py @@ -1,4 +1,4 @@ -"""Recoverable agent-exec controller: validate, isolate, supervise, receipt, transport.""" +"""Recoverable agent-exec controller: validate, run, apply transports, admit provenance.""" from __future__ import annotations import argparse @@ -169,6 +169,19 @@ def main(argv: list[str] | None = None) -> int: p_run.add_argument("--unit", required=True) p_run.add_argument("--receipt", required=True) + p_apply = sub.add_parser( + "apply", + help="Fast-forward the current branch to a success transport", + ) + p_apply.add_argument("--repo", required=True) + p_apply.add_argument("--receipt", required=True) + + p_admit = sub.add_parser( + "admit", + help="Admit staged paths backed by a success transport or host exemption", + ) + p_admit.add_argument("--repo", required=True) + args = parser.parse_args(argv) if args.command == "validate": return cmd_validate(Path(args.repo)) @@ -179,6 +192,10 @@ def main(argv: list[str] | None = None) -> int: unit_path=Path(args.unit), receipt_path=Path(args.receipt), ) + if args.command == "apply": + return cmd_apply(Path(args.repo), Path(args.receipt)) + if args.command == "admit": + return cmd_admit(Path(args.repo)) return EXIT_INVALID_CONFIG @@ -2117,5 +2134,318 @@ def _cap_bytes(data: bytes, limit: int) -> bytes: return data[:limit] +# Host may commit task cards and superpowers docs without a worker transport. +_HOST_MANAGED_PREFIXES = ( + ".task/", + "docs/superpowers/specs/", + "docs/superpowers/plans/", +) +_EMBEDDED_SECRET_RE = re.compile(r"(?i)(?:bearer\s+\S+|sk-[A-Za-z0-9_\-]{8,})") + + +def _reject(message: str, secrets: set[str], code: int = EXIT_BLOCKED) -> int: + print(_redact_text(message, secrets), file=sys.stderr) + return code + + +def _secret_set(*texts: str) -> set[str]: + found: set[str] = set() + _collect_secret_values(found) + for text in texts: + found.update(_EMBEDDED_SECRET_RE.findall(text)) + return found + + +def _normalize_repo_path(path: str) -> str: + """Normalize a repo-relative path without eating dotted names such as ``.task``.""" + norm = path.replace("\\", "/").strip() + while norm.startswith("./"): + norm = norm[2:] + return norm + + +def _host_managed(path: str) -> bool: + norm = _normalize_repo_path(path) + if not norm or ".." in Path(norm).parts: + return False + return any(norm.startswith(prefix) for prefix in _HOST_MANAGED_PREFIXES) + + +def _non_exempt(paths: list[str]) -> set[str]: + return {_normalize_repo_path(path) for path in paths if not _host_managed(path)} + + +def _same_repo(repo: Path, raw: object) -> bool: + if not isinstance(raw, str) or not raw.strip(): + return False + candidate = Path(raw) + if not candidate.is_absolute(): + candidate = repo / candidate + try: + return candidate.resolve() == repo.resolve() + except OSError: + return False + + +def _resolve_commit(repo: Path, rev: str) -> str | None: + if not rev or rev.startswith("-") or "\n" in rev or "\x00" in rev: + return None + proc = _git(repo, "rev-parse", "--verify", "--quiet", f"{rev}^{{commit}}", check=False) + if proc.returncode != 0: + return None + resolved = proc.stdout.strip() + return resolved or None + + +def _commit_parents(repo: Path, commit: str) -> list[str] | None: + proc = _git(repo, "rev-list", "--parents", "-n", "1", commit, check=False) + if proc.returncode != 0: + return None + parts = proc.stdout.split() + if not parts or parts[0] != commit: + return None + return parts[1:] + + +def _name_status_paths(text: str) -> list[str]: + paths: list[str] = [] + for line in text.splitlines(): + if not line.strip(): + continue + parts = line.split("\t") + status = parts[0] + if status[:1] in {"R", "C"} and len(parts) >= 3: + paths.append(parts[2]) + elif len(parts) >= 2: + paths.append(parts[1]) + return paths + + +def _diff_paths(repo: Path, parent: str, commit: str) -> list[str]: + proc = _git( + repo, + "diff", + "--name-status", + "--find-renames", + parent, + commit, + check=False, + ) + if proc.returncode != 0: + raise RuntimeError(proc.stderr.strip() or "transport diff failed") + return _name_status_paths(proc.stdout) + + +def _staged_paths(repo: Path) -> list[str]: + proc = _git( + repo, + "diff", + "--cached", + "--name-status", + "--find-renames", + check=False, + ) + if proc.returncode != 0: + raise RuntimeError(proc.stderr.strip() or "staged diff failed") + return _name_status_paths(proc.stdout) + + +def _index_matches_commit(repo: Path, commit: str, paths: set[str]) -> bool: + """True when each named index path has the same content as commit.""" + if not paths: + return True + proc = _git( + repo, + "diff", + "--cached", + "--quiet", + commit, + "--", + *sorted(paths), + check=False, + ) + return proc.returncode == 0 + + +def _valid_transport_ref(ref: object) -> str | None: + if not isinstance(ref, str) or not ref.startswith("refs/agent-exec/"): + return None + if ".." in ref.split("/") or ref.endswith("/"): + return None + return ref + + +def _history_file(repo: Path) -> Path | None: + common = _git( + repo, + "rev-parse", + "--path-format=absolute", + "--git-common-dir", + check=False, + ) + if common.returncode != 0 or not common.stdout.strip(): + raise RuntimeError(common.stderr.strip() or "unable to resolve git common dir") + common_dir = Path(common.stdout.strip()) + if not common_dir.is_absolute(): + common_dir = (repo / common_dir).resolve() + history_dir = common_dir / "agent-exec" + history_path = history_dir / "execution-history.jsonl" + if history_dir.is_symlink() or history_path.is_symlink(): + return None + return history_path + + +def _load_history_records(repo: Path) -> list[dict[str, Any]]: + history_path = _history_file(repo) + if history_path is None or not history_path.is_file(): + return [] + records: list[dict[str, Any]] = [] + for line in history_path.read_text(encoding="utf-8").splitlines(): + if not line.strip(): + continue + try: + data = json.loads(line) + except json.JSONDecodeError: + continue + if isinstance(data, dict): + records.append(data) + return records + + +def _transport_covers( + repo: Path, + record: dict[str, Any], + *, + head: str, + required: set[str], +) -> bool: + if record.get("schema") != HISTORY_SCHEMA: + return False + if record.get("terminal_status") != "success": + return False + if not _same_repo(repo, record.get("repo")): + return False + transport = record.get("transport") + if not isinstance(transport, dict): + return False + ref = _valid_transport_ref(transport.get("ref")) + parent = transport.get("parent") + if ref is None or not isinstance(parent, str): + return False + if _resolve_commit(repo, parent) != head: + return False + commit = _resolve_commit(repo, ref) + if commit is None or _commit_parents(repo, commit) != [head]: + return False + changed = record.get("changed_paths") + if not isinstance(changed, list) or any(not isinstance(path, str) for path in changed): + return False + diff_paths = _diff_paths(repo, head, commit) + if _non_exempt(changed) != required or _non_exempt(diff_paths) != required: + return False + return _index_matches_commit(repo, commit, required) + + +def cmd_apply(repo: Path, receipt_path: Path) -> int: + """Fast-forward the clean current branch onto one success transport.""" + repo = repo.resolve() + text = "" + try: + text = receipt_path.read_text(encoding="utf-8") + receipt = json.loads(text) + except (OSError, UnicodeError, json.JSONDecodeError) as exc: + return _reject( + f"apply: invalid receipt: {exc}", + _secret_set(text), + EXIT_INVALID_CONFIG, + ) + secrets = _secret_set(text) + if not isinstance(receipt, dict) or receipt.get("schema") != RECEIPT_SCHEMA: + return _reject("apply: invalid receipt", secrets, EXIT_INVALID_CONFIG) + if not _same_repo(repo, receipt.get("repo")): + return _reject("apply: receipt repo mismatch", secrets) + if receipt.get("terminal_status") != "success": + message = "apply: non-success receipt" + reason = receipt.get("reason") + if isinstance(reason, str) and reason: + message = f"{message}: {reason}" + return _reject(message, secrets) + + base = receipt.get("base") + base_head = base.get("head") if isinstance(base, dict) else None + if not isinstance(base_head, str): + return _reject("apply: stale HEAD", secrets) + resolved_base = _resolve_commit(repo, base_head) + head = _resolve_commit(repo, "HEAD") + status = _git( + repo, "status", "--porcelain", "--untracked-files=all", check=False + ) + if status.returncode != 0: + return _reject( + f"apply: dirty worktree: {status.stderr.strip()}", + secrets, + ) + if status.stdout.strip(): + return _reject("apply: dirty worktree", secrets) + branch = _git(repo, "rev-parse", "--abbrev-ref", "HEAD", check=False) + if branch.returncode != 0 or branch.stdout.strip() in {"", "HEAD"}: + return _reject("apply: detached HEAD", secrets) + if head is None or resolved_base is None or head != resolved_base: + return _reject("apply: stale HEAD", secrets) + + transport = receipt.get("transport") + if not isinstance(transport, dict): + return _reject("apply: missing transport", secrets) + ref = _valid_transport_ref(transport.get("ref")) + parent = transport.get("parent") + if ref is None or not isinstance(parent, str) or not parent: + return _reject("apply: missing transport", secrets) + commit = _resolve_commit(repo, ref) + if commit is None: + return _reject("apply: missing transport", secrets) + if ( + _resolve_commit(repo, parent) != resolved_base + or _commit_parents(repo, commit) != [resolved_base] + ): + return _reject("apply: mismatched transport", secrets) + + merge = _git(repo, "merge", "--ff-only", "--quiet", commit, check=False) + if merge.returncode != 0: + return _reject( + "apply: fast-forward failed: " + (merge.stderr.strip() or merge.stdout.strip()), + secrets, + ) + print(f"apply: fast-forwarded {branch.stdout.strip()} {head}..{commit}") + return EXIT_SUCCESS + + +def cmd_admit(repo: Path) -> int: + """Accept staged host-managed paths, or paths exactly covered by one transport.""" + repo = repo.resolve() + secrets = _secret_set() + try: + staged = _staged_paths(repo) + required = _non_exempt(staged) + if not required: + print("admit: ok") + return EXIT_SUCCESS + head = _resolve_commit(repo, "HEAD") + if head is None: + return _reject("admit: no provenance", secrets) + history_text = "" + history_path = _history_file(repo) + if history_path is not None and history_path.is_file(): + history_text = history_path.read_text(encoding="utf-8") + secrets = _secret_set(history_text) + for record in _load_history_records(repo): + if _transport_covers(repo, record, head=head, required=required): + print("admit: ok") + return EXIT_SUCCESS + except Exception as exc: + return _reject(f"admit: no provenance: {exc}", secrets) + listed = ", ".join(sorted(required)) + return _reject(f"admit: no provenance for {listed}", secrets) + + if __name__ == "__main__": sys.exit(main()) diff --git a/scripts/hooks/pre-commit b/scripts/hooks/pre-commit index c2eebd7b..08778e53 100755 --- a/scripts/hooks/pre-commit +++ b/scripts/hooks/pre-commit @@ -10,4 +10,5 @@ echo "$branch" | grep -qE '^(feature|fix|hotfix|release)/[A-Za-z0-9._-]+$' || { echo "pre-commit: 拒绝——分支名须为 feature|fix|hotfix|release/，当前：${branch}" >&2 exit 1 } -exit 0 +root=$(git rev-parse --show-toplevel) +PYTHONDONTWRITEBYTECODE=1 python3 "$root/scripts/agent_exec.py" admit --repo "$root" diff --git a/tests/test_agent_exec_admission.py b/tests/test_agent_exec_admission.py new file mode 100644 index 00000000..f0c50eab --- /dev/null +++ b/tests/test_agent_exec_admission.py @@ -0,0 +1,446 @@ +"""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, + )