diff --git a/scripts/clean_origin_post_merge_receipt.py b/scripts/clean_origin_post_merge_receipt.py new file mode 100644 index 00000000..ed296908 --- /dev/null +++ b/scripts/clean_origin_post_merge_receipt.py @@ -0,0 +1,256 @@ +#!/usr/bin/env python3 +"""Emit a clean-origin post-merge verification receipt for documentation drift controls.""" + +from __future__ import annotations + +import argparse +import json +import re +import shutil +import subprocess +import sys +import tempfile +from dataclasses import dataclass +from pathlib import Path +from typing import Sequence + +REDACTION_PATTERNS: tuple[re.Pattern[str], ...] = ( + re.compile(r"(?i)authorization\s*:\s*(bearer\s+)?([^\s,'\"}]+)"), + re.compile( + r"(?i)(api[_-]?key|access[_-]?token|refresh[_-]?token|id[_-]?token|password|passwd|secret|authorization|bearer)\s*[:=]\s*([^\s,'\"}]+)" + ), + re.compile(r"\bgh[pousr]_[A-Za-z0-9_]{20,}\b"), + re.compile(r"\bsk-[A-Za-z0-9][A-Za-z0-9_-]{16,}\b"), + re.compile(r"\b[A-Za-z0-9_-]{20,}\.[A-Za-z0-9_-]{20,}\.[A-Za-z0-9_-]{20,}\b"), +) + +DRIFT_MARKERS = ( + "drift", + "stale", + "out of date", + "source_commit", + "refresh_needed", + "wiki-lint", + "derived_outputs", +) +TOOLING_MARKERS = ( + "command not found", + "no such file or directory", + "modulenotfounderror", + "importerror", + "permission denied", + "timed out", +) + + +@dataclass(frozen=True) +class CommandResult: + command: str + status: str + summary: str + failure_class: str | None = None + + +def redact(text: str) -> str: + redacted = text + for pattern in REDACTION_PATTERNS: + + def repl(match: re.Match[str]) -> str: + if match.re.pattern.lower().startswith("(?i)authorization"): + return "Authorization: " + if match.lastindex and match.lastindex >= 2: + return f"{match.group(1)}=" + return "" + + redacted = pattern.sub(repl, redacted) + return redacted + + +def summarize(stdout: str, stderr: str, limit: int) -> str: + combined = redact("\n".join(part for part in (stdout, stderr) if part).strip()) + if not combined: + return "no output" + single_line = " | ".join(line.strip() for line in combined.splitlines() if line.strip()) + if len(single_line) <= limit: + return single_line + return single_line[: max(0, limit - 15)].rstrip() + " …" + + +def classify_failure(command: str, summary: str, returncode: int) -> str: + haystack = f"{command}\n{summary}".lower() + if any(marker in haystack for marker in TOOLING_MARKERS): + return "tooling_failure" + if any(marker in haystack for marker in DRIFT_MARKERS): + return "new_drift" + if returncode != 0: + return "baseline_noise" + return "none" + + +def run_git(args: Sequence[str], repo_root: Path) -> str: + completed = subprocess.run( + ["git", *args], + cwd=repo_root, + text=True, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + check=True, + ) + return completed.stdout.strip() + + +def _as_text(value: str | bytes | None) -> str: + if value is None: + return "" + if isinstance(value, bytes): + return value.decode("utf-8", errors="replace") + return value + + +def run_command(command: str, cwd: Path, timeout: int, summary_limit: int) -> CommandResult: + try: + completed = subprocess.run( + ["bash", "-lc", command], + cwd=cwd, + shell=False, + text=True, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + timeout=timeout, + ) + except subprocess.TimeoutExpired as exc: + summary = summarize( + _as_text(exc.stdout), + _as_text(exc.stderr) or "timed out", + summary_limit, + ) + return CommandResult( + command=command, + status="fail", + summary=summary, + failure_class="tooling_failure", + ) + + summary = summarize(completed.stdout, completed.stderr, summary_limit) + if completed.returncode == 0: + return CommandResult(command=command, status="pass", summary=summary) + return CommandResult( + command=command, + status="fail", + summary=summary, + failure_class=classify_failure(command, summary, completed.returncode), + ) + + +def create_clean_worktree(source_repo: Path, base_ref: str) -> Path: + temp_parent = Path(tempfile.mkdtemp(prefix="lifeos-clean-origin-receipt.")) + worktree = temp_parent / "worktree" + subprocess.run( + ["git", "worktree", "add", "--detach", str(worktree), base_ref], + cwd=source_repo, + text=True, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + check=True, + ) + return worktree + + +def remove_clean_worktree(source_repo: Path, worktree: Path) -> None: + subprocess.run( + ["git", "worktree", "remove", "--force", str(worktree)], + cwd=source_repo, + text=True, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + check=False, + ) + shutil.rmtree(worktree.parent, ignore_errors=True) + + +def build_receipt(args: argparse.Namespace) -> tuple[dict[str, object], int]: + repo_root = Path(args.repo_root).resolve() + if not repo_root.exists(): + raise SystemExit(f"repo root does not exist: {repo_root}") + + run_git(["fetch", "origin", "main", "--prune"], repo_root) + verified_commit = run_git(["rev-parse", args.base_ref], repo_root) + worktree = create_clean_worktree(repo_root, args.base_ref) + try: + results = [ + run_command(command, worktree, args.timeout, args.summary_limit) + for command in args.command + ] + dirty_status = run_git(["status", "--short"], worktree) + finally: + remove_clean_worktree(repo_root, worktree) + + failed = [result for result in results if result.status != "pass"] + if not failed: + completion_claim = "conductor_verified" + elif any(result.failure_class == "new_drift" for result in failed): + completion_claim = "follow_up_required" + else: + completion_claim = "failed" + + receipt: dict[str, object] = { + "repo": args.repo, + "base_ref": args.base_ref, + "verified_commit": verified_commit, + "commands": list(args.command), + "results": [ + { + key: value + for key, value in { + "command": result.command, + "status": result.status, + "summary": result.summary, + "failure_class": result.failure_class, + }.items() + if value is not None + } + for result in results + ], + "dirty_worktree_after_verification": bool(dirty_status), + "follow_up_issues_created": list(args.follow_up_issue), + "completion_claim": completion_claim, + } + return receipt, 0 if completion_claim == "conductor_verified" else 1 + + +def parse_args(argv: Sequence[str]) -> argparse.Namespace: + parser = argparse.ArgumentParser( + description=( + "Run post-merge verification from a clean origin/main worktree " + "and emit a redacted receipt." + ) + ) + parser.add_argument("--repo-root", default=Path(__file__).resolve().parent.parent) + parser.add_argument("--repo", default="marcusglee11/LifeOS") + parser.add_argument("--base-ref", default="origin/main") + parser.add_argument( + "--command", + action="append", + required=True, + help="Verification command to run inside the clean detached worktree; repeatable.", + ) + parser.add_argument("--follow-up-issue", action="append", default=[]) + parser.add_argument("--timeout", type=int, default=300) + parser.add_argument("--summary-limit", type=int, default=500) + parser.add_argument("--output", type=Path) + return parser.parse_args(argv) + + +def main(argv: Sequence[str] | None = None) -> int: + args = parse_args(argv or sys.argv[1:]) + receipt, exit_code = build_receipt(args) + output = json.dumps(receipt, indent=2, sort_keys=False) + "\n" + if args.output: + args.output.parent.mkdir(parents=True, exist_ok=True) + args.output.write_text(output, encoding="utf-8") + print(output, end="") + return exit_code + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tests_doc/test_clean_origin_post_merge_receipt.py b/tests_doc/test_clean_origin_post_merge_receipt.py new file mode 100644 index 00000000..9170b0ee --- /dev/null +++ b/tests_doc/test_clean_origin_post_merge_receipt.py @@ -0,0 +1,121 @@ +from __future__ import annotations + +import importlib.util +import sys +from argparse import Namespace +from pathlib import Path + +SCRIPT = Path(__file__).resolve().parents[1] / "scripts" / "clean_origin_post_merge_receipt.py" +spec = importlib.util.spec_from_file_location("clean_origin_post_merge_receipt", SCRIPT) +assert spec is not None +receipt = importlib.util.module_from_spec(spec) +sys.modules[spec.name] = receipt +assert spec.loader is not None +spec.loader.exec_module(receipt) + + +def test_redacts_token_shaped_output_from_summary() -> None: + summary = receipt.summarize( + "access_token=ghp_abcdefghijklmnopqrstuvwxyz123456 password=hunter2", + "Authorization: Bearer sk-testsecretsecretsecretsecret", + 500, + ) + + assert "ghp_" not in summary + assert "hunter2" not in summary + assert "sk-test" not in summary + assert "testsecret" not in summary + assert "Bearer" not in summary + assert "" in summary + + +def test_build_receipt_uses_origin_main_clean_worktree_and_records_success( + monkeypatch, tmp_path +) -> None: + calls: list[tuple[str, object]] = [] + clean = tmp_path / "clean" + clean.mkdir() + + def fake_run_git(args, repo_root): + calls.append(("git", tuple(args))) + if tuple(args) == ("rev-parse", "origin/main"): + return "a" * 40 + if tuple(args) == ("status", "--short"): + return "" + return "" + + monkeypatch.setattr(receipt, "run_git", fake_run_git) + monkeypatch.setattr(receipt, "create_clean_worktree", lambda repo_root, base_ref: clean) + monkeypatch.setattr( + receipt, + "remove_clean_worktree", + lambda repo_root, worktree: calls.append(("remove", worktree)), + ) + monkeypatch.setattr( + receipt, + "run_command", + lambda command, cwd, timeout, summary_limit: receipt.CommandResult( + command=command, status="pass", summary="ok" + ), + ) + + args = Namespace( + repo_root=tmp_path, + repo="marcusglee11/LifeOS", + base_ref="origin/main", + command=["python3 -m doc_steward.cli wiki-lint ."], + follow_up_issue=[], + timeout=10, + summary_limit=200, + ) + data, exit_code = receipt.build_receipt(args) + + assert exit_code == 0 + assert ("git", ("fetch", "origin", "main", "--prune")) in calls + assert data["base_ref"] == "origin/main" + assert data["verified_commit"] == "a" * 40 + assert data["dirty_worktree_after_verification"] is False + assert data["completion_claim"] == "conductor_verified" + + +def test_failed_wiki_drift_is_follow_up_required(monkeypatch, tmp_path) -> None: + clean = tmp_path / "clean" + clean.mkdir() + monkeypatch.setattr( + receipt, + "run_git", + lambda args, repo_root: "b" * 40 if tuple(args) == ("rev-parse", "origin/main") else "", + ) + monkeypatch.setattr(receipt, "create_clean_worktree", lambda repo_root, base_ref: clean) + monkeypatch.setattr(receipt, "remove_clean_worktree", lambda repo_root, worktree: None) + monkeypatch.setattr( + receipt, + "run_command", + lambda command, cwd, timeout, summary_limit: receipt.CommandResult( + command=command, + status="fail", + summary="wiki-lint found stale source_commit_max drift", + failure_class="new_drift", + ), + ) + + args = Namespace( + repo_root=tmp_path, + repo="marcusglee11/LifeOS", + base_ref="origin/main", + command=["python3 -m doc_steward.cli wiki-lint ."], + follow_up_issue=["https://github.com/marcusglee11/LifeOS/issues/120"], + timeout=10, + summary_limit=200, + ) + data, exit_code = receipt.build_receipt(args) + + assert exit_code == 1 + assert data["completion_claim"] == "follow_up_required" + assert data["results"][0]["failure_class"] == "new_drift" + assert data["follow_up_issues_created"] == ["https://github.com/marcusglee11/LifeOS/issues/120"] + + +def test_script_never_uses_shell_true() -> None: + source = SCRIPT.read_text() + assert "shell=True" not in source