feat: add advisory doc drift preview

doc_steward/cli.py

@@ -23,6 +23,7 @@
 from .artefact_index_validator import check_artefact_index
 from .dap_validator import check_dap_compliance
 from .doc_authority_manifest import check_doc_authority_manifest
+from .drift_sweep_issue_creator import run as run_drift_sweep_issue_creator
 from .entrypoint_freshness_sweep import (
     DEFAULT_LABELS as ENTRYPOINT_FRESHNESS_LABELS,
 )
@@ -316,6 +317,17 @@ def cmd_check_manifest(args: argparse.Namespace) -> int:
     return 0
 
 
+def cmd_drift_sweep_preview(args: argparse.Namespace) -> int:
+    """Run drift sweep preview (dry-run only, no GitHub mutation)."""
+    run_drift_sweep_issue_creator(
+        Path(args.repo_root).resolve(),
+        output_dir=args.output_dir,
+        print_json=args.json,
+        parent_link=args.parent_link,
+    )
+    return 0
+
+
 def main() -> int:
     parser = argparse.ArgumentParser(
         prog="doc_steward.cli", description="LifeOS Documentation Steward CLI"
@@ -387,6 +399,17 @@ def main() -> int:
     )
     p_entrypoint_sweep.set_defaults(func=cmd_entrypoint_freshness_sweep)
 
+    # drift-sweep-preview
+    p_drift = subparsers.add_parser(
+        "drift-sweep-preview",
+        help="Preview advisory documentation drift sweep issues (dry-run only, no GitHub mutation)",
+    )
+    p_drift.add_argument("repo_root", help="Repository root directory")
+    p_drift.add_argument("--output-dir", default=None, help="Output dir for preview files")
+    p_drift.add_argument("--json", action="store_true", help="Print JSON preview to stdout")
+    p_drift.add_argument("--parent-link", default=None, help="URL to parent tracking issue")
+    p_drift.set_defaults(func=cmd_drift_sweep_preview)
+
     # protocols-structure-check
     p_protocols = subparsers.add_parser(
         "protocols-structure-check", help="Validate docs/02_protocols/ structure"

doc_steward/drift_sweep_issue_creator.py

@@ -0,0 +1,266 @@
+"""
+Self-contained drift sweep issue preview creator.
+
+Read-only dry-run module that detects advisory documentation drift via
+freshness_validator.check_entrypoint_freshness() and produces structured
+JSON previews and human-readable markdown summaries without any GitHub
+mutation, sweep_lib dependency, or external state.
+"""
+
+from __future__ import annotations
+
+import argparse
+import hashlib
+import json
+import sys
+from datetime import datetime, timezone
+from pathlib import Path
+
+from .freshness_validator import check_entrypoint_freshness
+
+SWEEP_ID = "inventory-hygiene-sweep"
+TARGET = "lifeos-doc-entrypoint"
+CHECK_ID = "readme-entrypoint-freshness"
+SCHEMA_VERSION = "drift-sweep-preview-v1"
+
+
+def _fingerprint(
+    sweep_id: str,
+    target: str,
+    check_id: str,
+    evidence: str,
+    authority_class: str,
+) -> str:
+    """Deterministic SHA-256 fingerprint for a drift class group."""
+    raw = "|".join([sweep_id, target, check_id, evidence, authority_class])
+    return hashlib.sha256(raw.encode("utf-8")).hexdigest()[:16]
+
+
+def _normalize_evidence(findings: list[dict]) -> str:
+    """Stable normalized text for one drift-class group.
+
+    Sorts findings by id, lowercases all evidence, collapses whitespace.
+    """
+    parts = []
+    for finding in sorted(findings, key=lambda f: str(f.get("id", ""))):
+        ev = str(finding.get("evidence", ""))
+        parts.append(" ".join(ev.split()).strip().lower())
+    return " | ".join(parts)
+
+
+def _group_findings(findings: list[dict]) -> dict[str, list[dict]]:
+    """Group findings by their 'id' key, preserving first-occurrence order."""
+    groups: dict[str, list[dict]] = {}
+    for finding in findings:
+        fid = str(finding.get("id", ""))
+        if fid not in groups:
+            groups[fid] = []
+        groups[fid].append(finding)
+    return groups
+
+
+def _compute_fingerprints(groups: dict[str, list[dict]]) -> dict[str, str]:
+    """Compute per-class fingerprints for each drift class."""
+    fingerprints: dict[str, str] = {}
+    for fid, findings in groups.items():
+        evidence = _normalize_evidence(findings)
+        authority_class = str(findings[0].get("authority_class", ""))
+        fingerprints[fid] = _fingerprint(
+            SWEEP_ID,
+            TARGET,
+            CHECK_ID,
+            evidence,
+            authority_class,
+        )
+    return fingerprints
+
+
+def build_preview(
+    findings: list[dict],
+    parent_link: str | None = None,
+) -> dict:
+    """Build structured JSON preview from detector findings."""
+    groups = _group_findings(findings)
+    fingerprints = _compute_fingerprints(groups)
+    now = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
+
+    classes = {}
+    for fid, fgroup in groups.items():
+        classes[fid] = {
+            "count": len(fgroup),
+            "fingerprint": fingerprints.get(fid, ""),
+        }
+
+    findings_by_class: dict[str, list[dict]] = {}
+    for fid, fgroup in groups.items():
+        findings_by_class[fid] = [dict(f) for f in fgroup]
+
+    return {
+        "sweep_metadata": {
+            "sweep_id": SWEEP_ID,
+            "target": TARGET,
+            "check_id": CHECK_ID,
+            "generation_timestamp": now,
+            "schema_version": SCHEMA_VERSION,
+        },
+        "summary": {
+            "total_findings": len(findings),
+            "total_classes": len(groups),
+            "classes": classes,
+        },
+        "findings_by_class": findings_by_class,
+        "fingerprints": fingerprints,
+        "parent_link": parent_link,
+        "receipt": {
+            "tool": "drift_sweep_issue_creator",
+            "version": "1",
+            "mutated": False,
+            "dry_run": True,
+            "generated_at": now,
+        },
+    }
+
+
+def _footer() -> str:
+    return (
+        "---\n\n"
+        "_Generated by `drift_sweep_issue_creator`. "
+        "No files were mutated. This is a dry-run preview._"
+    )
+
+
+def render_markdown(preview: dict) -> str:
+    """Render a drift preview dict as human-readable markdown."""
+    lines: list[str] = []
+    sweep_md = preview.get("sweep_metadata", {})
+    summary = preview.get("summary", {})
+
+    lines.append("# Advisory Documentation Drift Sweep Preview")
+    lines.append("")
+    lines.append(f"- **Sweep:** `{sweep_md.get('sweep_id')}`")
+    lines.append(f"- **Target:** `{sweep_md.get('target')}`")
+    lines.append(f"- **Check:** `{sweep_md.get('check_id')}`")
+    lines.append(f"- **Generated:** {sweep_md.get('generation_timestamp')}")
+    lines.append("")
+
+    parent_link = preview.get("parent_link")
+    if parent_link:
+        lines.append(f"- **Parent issue:** {parent_link}")
+        lines.append("")
+
+    total_findings = summary.get("total_findings", 0)
+
+    if total_findings == 0:
+        lines.append("## No drift findings detected.")
+        lines.append("")
+        lines.append(_footer())
+        return "\n".join(lines)
+
+    total_classes = summary.get("total_classes", 0)
+    lines.append(f"**{total_findings} finding(s) across {total_classes} drift class(es)**")
+    lines.append("")
+    lines.append("| Drift Class | Count | Fingerprint |")
+    lines.append("|---|---|---|")
+
+    classes = summary.get("classes", {})
+    for fid, info in classes.items():
+        lines.append(f"| `{fid}` | {info.get('count', 0)} | `{info.get('fingerprint', '')}` |")
+    lines.append("")
+
+    findings_by_class = preview.get("findings_by_class", {})
+    for fid in classes:
+        fgroup = findings_by_class.get(fid, [])
+        lines.append("---")
+        lines.append("")
+        fp = classes.get(fid, {}).get("fingerprint", "")
+        lines.append(f"## {fid} ({len(fgroup)} finding(s))")
+        lines.append("")
+        lines.append(f"- **Fingerprint:** `{fp}`")
+
+        for finding in fgroup:
+            lines.append(f"- **Severity:** {finding.get('severity', 'N/A')}")
+            paths = finding.get("paths", [])
+            if paths:
+                lines.append(f"- **Paths:** {', '.join(str(p) for p in paths)}")
+            lines.append(f"- **Evidence:** {finding.get('evidence', 'N/A')}")
+            recovery = finding.get("recommended_recovery", "")
+            if recovery:
+                lines.append(f"- **Recovery:** {recovery}")
+            authority = finding.get("authority_class", "")
+            if authority:
+                lines.append(f"- **Authority:** {authority}")
+        lines.append("")
+
+    lines.append(_footer())
+    return "\n".join(lines)
+
+
+def write_preview(preview: dict, output_dir: str | Path) -> list[Path]:
+    """Write JSON and MD preview files to output_dir.
+
+    Creates output_dir if needed. Writes timestamped copies plus _latest files.
+    Returns list of written file paths.
+    """
+    output_path = Path(output_dir)
+    output_path.mkdir(parents=True, exist_ok=True)
+
+    now = datetime.now(timezone.utc)
+    timestamp = now.strftime("%Y%m%dT%H%M%SZ")
+
+    json_path = output_path / f"drift_sweep_preview_{timestamp}.json"
+    md_path = output_path / f"drift_sweep_preview_{timestamp}.md"
+
+    json_text = json.dumps(preview, indent=2, sort_keys=False) + "\n"
+    md_text = render_markdown(preview) + "\n"
+
+    json_path.write_text(json_text, encoding="utf-8")
+    md_path.write_text(md_text, encoding="utf-8")
+
+    latest_json = output_path / "drift_sweep_preview_latest.json"
+    latest_md = output_path / "drift_sweep_preview_latest.md"
+
+    latest_json.write_text(json_text, encoding="utf-8")
+    latest_md.write_text(md_text, encoding="utf-8")
+
+    return [json_path, md_path, latest_json, latest_md]
+
+
+def run(
+    repo_root: str | Path,
+    output_dir: str | Path | None = None,
+    print_json: bool = False,
+    parent_link: str | None = None,
+) -> dict:
+    """Orchestrate: detect -> build_preview -> optional write/print."""
+    findings = check_entrypoint_freshness(repo_root)
+    preview = build_preview(findings, parent_link=parent_link)
+
+    if output_dir is not None:
+        write_preview(preview, output_dir)
+
+    if print_json:
+        print(json.dumps(preview, indent=2, sort_keys=False))
+
+    return preview
+
+
+def main(argv: list[str] | None = None) -> int:
+    """Standalone CLI entry point."""
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("repo_root", help="Repository root directory")
+    parser.add_argument("--output-dir", default=None, help="Output dir for preview files")
+    parser.add_argument("--json", action="store_true", help="Print JSON preview to stdout")
+    parser.add_argument("--parent-link", default=None, help="URL to parent tracking issue")
+    args = parser.parse_args(argv)
+
+    run(
+        Path(args.repo_root).resolve(),
+        output_dir=args.output_dir,
+        print_json=args.json,
+        parent_link=args.parent_link,
+    )
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())