Add clean-origin docs verification receipts

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())

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"
+    ]