diff --git a/scripts/wiki/post_merge_verification_receipt.py b/scripts/wiki/post_merge_verification_receipt.py new file mode 100644 index 00000000..e60e60a7 --- /dev/null +++ b/scripts/wiki/post_merge_verification_receipt.py @@ -0,0 +1,243 @@ +#!/usr/bin/env python3 +"""Emit clean-origin post-merge verification receipts for docs drift controls.""" + +from __future__ import annotations + +import argparse +import json +import re +import shlex +import shutil +import subprocess +import sys +import tempfile +from pathlib import Path +from typing import Sequence + +DEFAULT_COMMANDS = ( + "python3 scripts/wiki/check_derived_outputs.py", + "python3 -m doc_steward.cli wiki-lint .", + "python3 scripts/workflow/quality_gate.py check --scope changed --json", +) + +DRIFT_MARKERS = ( + "refresh_needed", + "pending_diff", + "needs refresh", + "stale", + "wiki-lint", + "derived output", + "provenance", + "corpus", +) +TOOLING_MARKERS = ( + "no such file or directory", + "command not found", + "modulenotfounderror", + "permission denied", + "traceback", + "failed to clone", + "not a git repository", +) +BASELINE_MARKERS = ( + "baseline", + "pre-existing", + "unrelated", +) +SENSITIVE_PATTERNS = ( + re.compile( + r"(?i)(api[_-]?key|access[_-]?token|refresh[_-]?token|secret|password|credential)" + r"([\s:=]+)([^\s,;]+)" + ), + re.compile(r"(?i)bearer\s+[A-Za-z0-9._~+/=-]{16,}"), + re.compile(r"ghp_[A-Za-z0-9_]{20,}"), + re.compile(r"github_pat_[A-Za-z0-9_]{20,}"), + re.compile(r"xox[baprs]-[A-Za-z0-9-]{10,}"), + re.compile( + r"-----BEGIN (?:OPENSSH|RSA|EC|DSA)? ?PRIVATE KEY-----.*?" + r"-----END (?:OPENSSH|RSA|EC|DSA)? ?PRIVATE KEY-----", + re.DOTALL, + ), +) + + +def _run(cmd: Sequence[str] | str, cwd: Path) -> subprocess.CompletedProcess[str]: + argv = shlex.split(cmd) if isinstance(cmd, str) else list(cmd) + return subprocess.run(argv, cwd=cwd, capture_output=True, text=True, check=False) + + +def redact_sensitive(text: str) -> str: + redacted = text + for pattern in SENSITIVE_PATTERNS: + redacted = pattern.sub( + lambda match: ( + f"{match.group(1)}{match.group(2)}[REDACTED]" + if match.lastindex == 3 + else "[REDACTED]" + ), + redacted, + ) + return redacted + + +def _summarise(stdout: str, stderr: str, limit: int = 240) -> str: + text = "\n".join(part.strip() for part in (stdout, stderr) if part.strip()).strip() + if not text: + return "command produced no output" + text = " ".join(redact_sensitive(text).split()) + if len(text) > limit: + return text[: limit - 1].rstrip() + "…" + return text + + +def classify_failure(summary: str) -> str: + lowered = summary.lower() + if any(marker in lowered for marker in TOOLING_MARKERS): + return "tooling_failure" + if any(marker in lowered for marker in BASELINE_MARKERS): + return "baseline_noise" + if any(marker in lowered for marker in DRIFT_MARKERS): + return "new_drift" + return "tooling_failure" + + +def is_dirty(repo_root: Path) -> bool: + status = _run(["git", "status", "--short"], repo_root) + return bool(status.stdout.strip()) or status.returncode != 0 + + +def rev_parse(repo_root: Path, ref: str) -> str: + result = _run(["git", "rev-parse", ref], repo_root) + if result.returncode != 0: + raise RuntimeError(_summarise(result.stdout, result.stderr)) + return result.stdout.strip() + + +def emit_yaml(receipt: dict) -> str: + def scalar(value: object) -> str: + if isinstance(value, bool): + return "true" if value else "false" + if value is None: + return "null" + text = str(value) + if not text or any(ch in text for ch in ":#[]{}\n") or text.strip() != text: + return json.dumps(text) + return text + + lines: list[str] = [] + for key in ("repo", "base_ref", "verified_commit"): + lines.append(f"{key}: {scalar(receipt[key])}") + lines.append("commands:") + for command in receipt["commands"]: + lines.append(f" - {scalar(command)}") + lines.append("results:") + for result in receipt["results"]: + lines.append(f" - command: {scalar(result['command'])}") + lines.append(f" status: {scalar(result['status'])}") + lines.append(f" summary: {scalar(result['summary'])}") + if result.get("failure_classification"): + lines.append(f" failure_classification: {scalar(result['failure_classification'])}") + lines.append( + f"dirty_worktree_after_verification: {scalar(receipt['dirty_worktree_after_verification'])}" + ) + lines.append("follow_up_issues_created:") + for url in receipt["follow_up_issues_created"]: + lines.append(f" - {scalar(url)}") + if not receipt["follow_up_issues_created"]: + lines[-1] += " []" + lines.append(f"completion_claim: {scalar(receipt['completion_claim'])}") + return "\n".join(lines) + "\n" + + +def build_receipt( + repo_root: Path, + commands: Sequence[str], + follow_up_issues: Sequence[str] = (), + keep_worktree: bool = False, +) -> tuple[dict, Path | None]: + fetch = _run(["git", "fetch", "origin", "main", "--prune"], repo_root) + if fetch.returncode != 0: + raise RuntimeError(_summarise(fetch.stdout, fetch.stderr)) + + verified_commit = rev_parse(repo_root, "origin/main") + temp_parent = Path(tempfile.mkdtemp(prefix="lifeos-post-merge-verify.")) + verify_root = temp_parent / "origin-main" + added = _run(["git", "worktree", "add", "--detach", str(verify_root), "origin/main"], repo_root) + if added.returncode != 0: + shutil.rmtree(temp_parent, ignore_errors=True) + raise RuntimeError(_summarise(added.stdout, added.stderr)) + + results: list[dict[str, str]] = [] + try: + for command in commands: + outcome = _run(command, verify_root) + status = "pass" if outcome.returncode == 0 else "fail" + summary = _summarise(outcome.stdout, outcome.stderr) + row = {"command": command, "status": status, "summary": summary} + if status == "fail": + row["failure_classification"] = classify_failure(summary) + results.append(row) + dirty_after = is_dirty(verify_root) + finally: + if not keep_worktree: + _run(["git", "worktree", "remove", "--force", str(verify_root)], repo_root) + shutil.rmtree(temp_parent, ignore_errors=True) + + failed = [row for row in results if row["status"] == "fail"] + completion_claim = "conductor_verified" + if failed: + completion_claim = "follow_up_required" if follow_up_issues else "failed" + + return ( + { + "repo": "marcusglee11/LifeOS", + "base_ref": "origin/main", + "verified_commit": verified_commit, + "commands": list(commands), + "results": results, + "dirty_worktree_after_verification": dirty_after, + "follow_up_issues_created": list(follow_up_issues), + "completion_claim": completion_claim, + }, + verify_root if keep_worktree else None, + ) + + +def parse_args(argv: Sequence[str]) -> argparse.Namespace: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument( + "--repo-root", default=".", help="LifeOS checkout to control worktrees from" + ) + parser.add_argument( + "--command", + action="append", + dest="commands", + help="Verification command to run in the clean origin/main worktree; repeatable", + ) + parser.add_argument( + "--follow-up-issue", + action="append", + default=[], + help="Follow-up issue URL created for failed verification", + ) + parser.add_argument("--json", action="store_true", help="Emit JSON instead of YAML") + parser.add_argument( + "--keep-worktree", action="store_true", help="Keep verification worktree for debugging" + ) + return parser.parse_args(argv) + + +def main(argv: Sequence[str] | None = None) -> int: + args = parse_args(argv or sys.argv[1:]) + commands = args.commands or list(DEFAULT_COMMANDS) + receipt, verify_root = build_receipt( + Path(args.repo_root).resolve(), commands, args.follow_up_issue, args.keep_worktree + ) + if verify_root is not None: + receipt["verification_worktree"] = str(verify_root) + print(json.dumps(receipt, indent=2) if args.json else emit_yaml(receipt)) + return 0 if receipt["completion_claim"] == "conductor_verified" else 1 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/tests_doc/test_post_merge_verification_receipt.py b/tests_doc/test_post_merge_verification_receipt.py new file mode 100644 index 00000000..ba106419 --- /dev/null +++ b/tests_doc/test_post_merge_verification_receipt.py @@ -0,0 +1,121 @@ +from __future__ import annotations + +import subprocess +from pathlib import Path + +from scripts.wiki import post_merge_verification_receipt as receipt + + +def test_classify_failure_distinguishes_drift_baseline_and_tooling() -> None: + assert receipt.classify_failure("wiki-lint found stale source_commit_max") == "new_drift" + assert receipt.classify_failure("pre-existing markdown baseline noise") == "baseline_noise" + assert ( + receipt.classify_failure("ModuleNotFoundError: No module named doc_steward") + == "tooling_failure" + ) + + +def test_summary_redacts_sensitive_command_output() -> None: + summary = receipt._summarise( + "api_key=abc123 ghp_abcdefghijklmnopqrstuvwxyz123456", + "password: hunter2\nAuthorization: Bearer abcdefghijklmnopqrstuvwxyz", + ) + + assert "abc123" not in summary + assert "hunter2" not in summary + assert "ghp_" not in summary + assert "Bearer abcdef" not in summary + assert "[REDACTED]" in summary + + +def test_emit_yaml_matches_required_receipt_shape() -> None: + rendered = receipt.emit_yaml( + { + "repo": "marcusglee11/LifeOS", + "base_ref": "origin/main", + "verified_commit": "abc123", + "commands": ["python3 scripts/wiki/check_derived_outputs.py"], + "results": [ + { + "command": "python3 scripts/wiki/check_derived_outputs.py", + "status": "pass", + "summary": "ok", + } + ], + "dirty_worktree_after_verification": False, + "follow_up_issues_created": [], + "completion_claim": "conductor_verified", + } + ) + + assert "repo: marcusglee11/LifeOS" in rendered + assert "base_ref: origin/main" in rendered + assert "verified_commit: abc123" in rendered + assert "dirty_worktree_after_verification: false" in rendered + assert "follow_up_issues_created: []" in rendered + assert "completion_claim: conductor_verified" in rendered + + +def test_build_receipt_fetches_and_runs_commands_in_clean_origin_main_worktree( + monkeypatch, tmp_path: Path +) -> None: + calls: list[tuple[tuple[str, ...] | str, Path]] = [] + verify_root_holder: dict[str, Path] = {} + + def fake_run(cmd, cwd: Path): + calls.append((tuple(cmd) if isinstance(cmd, list) else cmd, cwd)) + if isinstance(cmd, list) and cmd[:3] == ["git", "rev-parse", "origin/main"]: + return subprocess.CompletedProcess(cmd, 0, stdout="deadbeef\n", stderr="") + if isinstance(cmd, list) and cmd[:3] == ["git", "worktree", "add"]: + verify_root_holder["path"] = Path(cmd[-2]) + return subprocess.CompletedProcess(cmd, 0, stdout="", stderr="") + if isinstance(cmd, list) and cmd[:3] == ["git", "status", "--short"]: + return subprocess.CompletedProcess(cmd, 0, stdout="", stderr="") + return subprocess.CompletedProcess(cmd, 0, stdout="ok", stderr="") + + monkeypatch.setattr(receipt, "_run", fake_run) + monkeypatch.setattr(receipt.tempfile, "mkdtemp", lambda prefix: str(tmp_path / "verify")) + monkeypatch.setattr(receipt.shutil, "rmtree", lambda *args, **kwargs: None) + + result, kept = receipt.build_receipt(Path("/repo"), ["python3 smoke.py"]) + + assert kept is None + assert result["base_ref"] == "origin/main" + assert result["verified_commit"] == "deadbeef" + assert result["completion_claim"] == "conductor_verified" + assert result["dirty_worktree_after_verification"] is False + assert (("git", "fetch", "origin", "main", "--prune"), Path("/repo")) in calls + assert ("python3 smoke.py", verify_root_holder["path"]) in calls + assert any( + call[0][:3] == ("git", "worktree", "remove") for call in calls if isinstance(call[0], tuple) + ) + + +def test_failed_receipt_records_classification_and_follow_up_state( + monkeypatch, tmp_path: Path +) -> None: + def fake_run(cmd, cwd: Path): + if isinstance(cmd, list) and cmd[:3] == ["git", "rev-parse", "origin/main"]: + return subprocess.CompletedProcess(cmd, 0, stdout="deadbeef\n", stderr="") + if isinstance(cmd, list) and cmd[:3] == ["git", "status", "--short"]: + return subprocess.CompletedProcess(cmd, 0, stdout="", stderr="") + if cmd == "python3 scripts/wiki/check_derived_outputs.py": + return subprocess.CompletedProcess(cmd, 1, stdout="stale wiki provenance", stderr="") + return subprocess.CompletedProcess(cmd, 0, stdout="", stderr="") + + monkeypatch.setattr(receipt, "_run", fake_run) + monkeypatch.setattr(receipt.tempfile, "mkdtemp", lambda prefix: str(tmp_path / "verify")) + monkeypatch.setattr(receipt.shutil, "rmtree", lambda *args, **kwargs: None) + + result, _ = receipt.build_receipt( + Path("/repo"), + ["python3 scripts/wiki/check_derived_outputs.py"], + ["https://github.com/marcusglee11/LifeOS/issues/999"], + ) + + assert result["completion_claim"] == "follow_up_required" + assert result["results"][0]["status"] == "fail" + assert result["results"][0]["failure_classification"] == "new_drift" + assert result["follow_up_issues_created"] == [ + "https://github.com/marcusglee11/LifeOS/issues/999" + ]