diff --git a/doc_steward/cli.py b/doc_steward/cli.py index 0d2af881..cf2d5132 100644 --- a/doc_steward/cli.py +++ b/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 .entrypoint_freshness_sweep import run as run_entrypoint_freshness_sweep from .freshness_validator import check_freshness, get_freshness_mode from .global_archive_link_ban_validator import check_global_archive_link_ban from .index_checker import check_index @@ -158,6 +159,19 @@ def cmd_freshness_check(args: argparse.Namespace) -> int: return 0 +def cmd_entrypoint_freshness_sweep(args: argparse.Namespace) -> int: + """Run entrypoint freshness issue-creator adapter.""" + run_entrypoint_freshness_sweep( + Path(args.repo_root).resolve(), + repo=args.repo, + create_issue=args.create_issue, + dry_run=args.dry_run, + json_output=args.json, + record_run=args.record_run, + ) + return 0 + + def cmd_protocols_structure_check(args: argparse.Namespace) -> int: """Run protocols structure validation.""" repo_root = str(Path(args.repo_root).resolve()) @@ -330,6 +344,23 @@ def main() -> int: p_freshness.add_argument("repo_root", help="Repository root directory") p_freshness.set_defaults(func=cmd_freshness_check) + # entrypoint-freshness-sweep + p_entrypoint_sweep = subparsers.add_parser( + "entrypoint-freshness-sweep", + help="Run read-only entrypoint freshness sweep issue-creator adapter", + ) + p_entrypoint_sweep.add_argument("repo_root", help="Repository root directory") + p_entrypoint_sweep.add_argument("--repo", default="marcusglee11/lifeos-operational-bus") + p_entrypoint_sweep.add_argument("--dry-run", action="store_true") + p_entrypoint_sweep.add_argument("--create-issue", action="store_true") + p_entrypoint_sweep.add_argument( + "--record-run", + action="store_true", + help="write a sweep_lib run receipt; rejected with --dry-run", + ) + p_entrypoint_sweep.add_argument("--json", action="store_true") + p_entrypoint_sweep.set_defaults(func=cmd_entrypoint_freshness_sweep) + # protocols-structure-check p_protocols = subparsers.add_parser( "protocols-structure-check", help="Validate docs/02_protocols/ structure" diff --git a/doc_steward/entrypoint_freshness_sweep.py b/doc_steward/entrypoint_freshness_sweep.py new file mode 100644 index 00000000..414f7e7c --- /dev/null +++ b/doc_steward/entrypoint_freshness_sweep.py @@ -0,0 +1,300 @@ +"""Issue-creator adapter for doc entrypoint freshness findings. + +This module is intentionally inert unless called by a sweep wrapper. It converts +read-only entrypoint freshness findings into sweep_lib-compatible fingerprint, +upsert, and issue payload records. +""" + +from __future__ import annotations + +import argparse +import hashlib +import json +import os +import re +import subprocess +import sys +import tempfile +from pathlib import Path +from typing import Any + +from .freshness_validator import check_entrypoint_freshness + +SWEEP_ID = "inventory-hygiene-sweep" +TARGET = "lifeos-doc-entrypoint" +CHECK_ID = "readme-entrypoint-freshness" +DEFAULT_REPO = "marcusglee11/lifeos-operational-bus" +DEFAULT_LABELS = ["sweep:inventory-hygiene", "severity:warning"] + + +class SweepLibUnavailable(RuntimeError): + """Raised when sweep_lib is required but unavailable.""" + + +def _load_sweep_lib() -> dict[str, Any]: + sweep_root = Path(os.environ.get("HERMES_SWEEP_LIB", Path.home() / ".hermes" / "sweep")) + sys.path.insert(0, str(sweep_root)) + try: + from lib import ( # type: ignore[import-not-found] + FindingsDB, + make_fingerprint, + record_sweep_run, + validate_issue_payload, + ) + except Exception as exc: # pragma: no cover - exercised through require_sweep_lib + raise SweepLibUnavailable(f"sweep_lib unavailable at {sweep_root}: {exc}") from exc + return { + "FindingsDB": FindingsDB, + "make_fingerprint": make_fingerprint, + "record_sweep_run": record_sweep_run, + "validate_issue_payload": validate_issue_payload, + } + + +def _dry_run_fingerprint(normalized_error: str) -> str: + """Deterministic dry-run fingerprint without importing live sweep_lib.""" + raw = "|".join([SWEEP_ID, TARGET, CHECK_ID, normalized_error, "warning"]) + return hashlib.sha256(raw.encode("utf-8")).hexdigest()[:16] + + +def normalize_error(findings: list[dict[str, object]]) -> str: + """Stable normalized finding summary for one deduped sweep issue.""" + parts = [] + for finding in sorted(findings, key=lambda item: str(item.get("id", ""))): + parts.append(f"{finding.get('id')}: {finding.get('evidence')}") + return " | ".join(" ".join(part.split()).strip().lower() for part in parts) + + +def issue_title(findings: list[dict[str, object]]) -> str: + ids = ", ".join(str(f.get("id")) for f in findings[:3]) + suffix = "" if len(findings) <= 3 else f" +{len(findings) - 3} more" + return f"[Inventory Hygiene] LifeOS doc entrypoint freshness drift: {ids}{suffix}"[:190] + + +def issue_body(findings: list[dict[str, object]]) -> str: + evidence_lines = [] + next_actions = [] + for finding in findings: + raw_paths = finding.get("paths", []) + paths = raw_paths if isinstance(raw_paths, list) else [] + evidence_lines.append( + "- {id}: paths={paths}; evidence={evidence}; authority={authority}".format( + id=finding.get("id"), + paths=", ".join(str(path) for path in paths), + evidence=finding.get("evidence"), + authority=finding.get("authority_class"), + ) + ) + recovery = str(finding.get("recommended_recovery", "")).strip() + if recovery and recovery not in next_actions: + next_actions.append(recovery) + return ( + "**Finding:** LifeOS README/operator entrypoint freshness drift was detected.\n\n" + f"**Target:** {TARGET}\n\n" + "**Evidence:**\n```text\n" + "\n".join(evidence_lines)[:1800] + "\n```\n\n" + "**Next action:** " + + "; ".join(next_actions)[:1200] + + ( + f"\n\nSweep: `{SWEEP_ID}`. Check: `{CHECK_ID}`. " + "Authority: read-only detector; no docs were modified." + ) + ) + + +def _as_int(value: object) -> int: + return value if isinstance(value, int) else 0 + + +def gh_create_issue(repo: str, title: str, body: str, labels: list[str]) -> int: + with tempfile.NamedTemporaryFile("w", suffix=".md", delete=False, encoding="utf-8") as handle: + handle.write(body) + body_path = handle.name + try: + cmd = ["gh", "issue", "create", "-R", repo, "--title", title, "--body-file", body_path] + for label in labels: + cmd.extend(["--label", label]) + proc = subprocess.run(cmd, text=True, capture_output=True, timeout=90) + if proc.returncode != 0: + raise RuntimeError(proc.stderr.strip() or proc.stdout.strip()) + url = proc.stdout.strip().splitlines()[-1] + match = re.search(r"/issues/(\d+)", url) + if not match: + raise RuntimeError(f"could not parse issue number from gh output: {url}") + return int(match.group(1)) + finally: + try: + os.unlink(body_path) + except OSError: + pass + + +def process_findings( + findings: list[dict[str, object]], + *, + repo: str = DEFAULT_REPO, + create_issue: bool = False, + dry_run: bool = False, + sweep_lib: dict[str, Any] | None = None, +) -> dict[str, object]: + """Process detector findings through sweep_lib dedupe semantics.""" + if not findings: + return {"findings": 0, "findings_created": 0, "findings_updated": 0, "rows": []} + + normalized = normalize_error(findings) + title = issue_title(findings) + body = issue_body(findings) + labels = list(DEFAULT_LABELS) + + if dry_run: + if sweep_lib is None: + fingerprint = _dry_run_fingerprint(normalized) + else: + sweep_lib["validate_issue_payload"](title, body, labels) + fingerprint = sweep_lib["make_fingerprint"]( + SWEEP_ID, TARGET, CHECK_ID, normalized, "warning" + ) + return { + "findings": len(findings), + "findings_created": 0, + "findings_updated": 0, + "rows": [ + { + "fingerprint": fingerprint, + "severity": "warning", + "title": title, + "issue": None, + "action": "dry-run", + } + ], + } + + libs = sweep_lib or _load_sweep_lib() + fingerprint = libs["make_fingerprint"](SWEEP_ID, TARGET, CHECK_ID, normalized, "warning") + libs["validate_issue_payload"](title, body, labels) + + row = { + "fingerprint": fingerprint, + "severity": "warning", + "title": title, + "issue": None, + } + + db = libs["FindingsDB"]() + try: + result = db.upsert_finding(fingerprint, SWEEP_ID, TARGET, CHECK_ID, normalized, "warning") + action = result["action"] + issue_num = None + created = 0 + updated = 0 + if action == "created": + if create_issue: + issue_num = gh_create_issue(repo, title, body, labels) + db.upsert_finding( + fingerprint, + SWEEP_ID, + TARGET, + CHECK_ID, + normalized, + "warning", + gh_issue_number=issue_num, + ) + created = 1 + elif action == "updated": + updated = 1 + row.update({"action": action, "issue": issue_num}) + return { + "findings": len(findings), + "findings_created": created, + "findings_updated": updated, + "rows": [row], + } + finally: + close = getattr(db, "close", None) + if callable(close): + close() + + +def run( + repo_root: str | Path, + *, + repo: str = DEFAULT_REPO, + create_issue: bool = False, + dry_run: bool = False, + json_output: bool = False, + record_run: bool = False, +) -> dict[str, object]: + if dry_run and record_run: + raise ValueError("record_run mutates sweep receipts and cannot be combined with dry_run") + + findings = check_entrypoint_freshness(repo_root) + libs = None if dry_run else _load_sweep_lib() + result = process_findings( + findings, + repo=repo, + create_issue=create_issue, + dry_run=dry_run, + sweep_lib=libs, + ) + receipt = None + if record_run: + assert libs is not None + receipt = libs["record_sweep_run"]( + SWEEP_ID, + 0, + findings_created=_as_int(result.get("findings_created", 0)), + findings_updated=_as_int(result.get("findings_updated", 0)), + telegram_sent=False, + model_used="no_agent:doc-entrypoint-freshness", + ) + payload = { + "sweep_id": SWEEP_ID, + "target": TARGET, + "check_id": CHECK_ID, + "receipt": receipt, + **result, + } + if json_output: + print(json.dumps(payload, indent=2, sort_keys=True)) + elif findings: + print(f"{CHECK_ID}: {len(findings)} finding(s)") + else: + print("[SILENT]") + return payload + + +def main(argv: list[str] | None = None) -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("repo_root", help="LifeOS repository root") + parser.add_argument("--repo", default=DEFAULT_REPO) + parser.add_argument( + "--dry-run", + action="store_true", + help="do not mutate sweep findings DB or GitHub", + ) + parser.add_argument( + "--create-issue", + action="store_true", + help="create one deduped GitHub issue for a new finding", + ) + parser.add_argument( + "--record-run", + action="store_true", + help="write a sweep_lib run receipt; rejected with --dry-run", + ) + parser.add_argument("--json", action="store_true") + args = parser.parse_args(argv) + if args.dry_run and args.record_run: + parser.error("--record-run mutates sweep receipts and cannot be combined with --dry-run") + run( + args.repo_root, + repo=args.repo, + create_issue=args.create_issue, + dry_run=args.dry_run, + json_output=args.json, + record_run=args.record_run, + ) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tests_doc/test_entrypoint_freshness_sweep.py b/tests_doc/test_entrypoint_freshness_sweep.py new file mode 100644 index 00000000..c79305a8 --- /dev/null +++ b/tests_doc/test_entrypoint_freshness_sweep.py @@ -0,0 +1,194 @@ +"""Tests for entrypoint freshness sweep adapter.""" + +from __future__ import annotations + +from typing import Any, cast + +from doc_steward.entrypoint_freshness_sweep import ( + CHECK_ID, + SWEEP_ID, + TARGET, + issue_body, + normalize_error, + process_findings, + run, +) + + +class FakeDB: + def __init__(self, action: str = "created") -> None: + self.action = action + self.calls = [] + self.closed = False + + def upsert_finding(self, *args, **kwargs): + self.calls.append((args, kwargs)) + return {"action": self.action} + + def close(self) -> None: + self.closed = True + + +def finding(identifier: str = "entrypoint-read-order-missing-links") -> dict[str, object]: + return { + "id": identifier, + "severity": "warning", + "paths": ["README.md"], + "evidence": "README operator read-order is missing: docs/INDEX.md", + "recommended_recovery": "Add the missing canonical read-order links to README.md.", + "authority_class": "canonical", + } + + +def fake_lib(db: FakeDB): + return { + "FindingsDB": lambda: db, + "make_fingerprint": lambda sweep_id, target, check_id, normalized_error, severity: ( + f"fp:{sweep_id}:{target}:{check_id}:{severity}:{len(normalized_error)}" + ), + "validate_issue_payload": lambda title, body, labels: True, + "record_sweep_run": lambda *args, **kwargs: "/tmp/receipt.json", + } + + +def test_normalize_error_is_stable_by_finding_id(): + first = [finding("b"), finding("a")] + second = [finding("a"), finding("b")] + + assert normalize_error(first) == normalize_error(second) + + +def test_issue_body_contains_sweep_lib_required_sections(): + body = issue_body([finding()]) + + assert "**Finding:**" in body + assert f"**Target:** {TARGET}" in body + assert "**Evidence:**" in body + assert "**Next action:**" in body + assert SWEEP_ID in body + assert CHECK_ID in body + + +def test_process_findings_dry_run_does_not_touch_db(): + db = FakeDB() + + result = process_findings([finding()], dry_run=True, sweep_lib=fake_lib(db)) + + assert result["findings"] == 1 + assert result["findings_created"] == 0 + assert result["findings_updated"] == 0 + rows = cast(list[dict[str, Any]], result["rows"]) + assert rows[0]["action"] == "dry-run" + assert db.calls == [] + + +def test_process_findings_created_records_one_finding_without_issue_creation(): + db = FakeDB(action="created") + + result = process_findings( + [finding()], dry_run=False, create_issue=False, sweep_lib=fake_lib(db) + ) + + assert result["findings_created"] == 1 + assert result["findings_updated"] == 0 + rows = cast(list[dict[str, Any]], result["rows"]) + assert rows[0]["action"] == "created" + assert db.calls[0][0][0].startswith("fp:inventory-hygiene-sweep:lifeos-doc-entrypoint") + assert db.closed is True + + +def test_process_findings_updated_skips_duplicate_creation(): + db = FakeDB(action="updated") + + result = process_findings([finding()], dry_run=False, create_issue=True, sweep_lib=fake_lib(db)) + + assert result["findings_created"] == 0 + assert result["findings_updated"] == 1 + rows = cast(list[dict[str, Any]], result["rows"]) + assert rows[0]["action"] == "updated" + assert db.closed is True + + +def test_process_findings_clean_state_has_no_rows(): + result = process_findings([], dry_run=False, sweep_lib=fake_lib(FakeDB())) + + assert result == {"findings": 0, "findings_created": 0, "findings_updated": 0, "rows": []} + + +def test_process_findings_dirty_dry_run_does_not_require_sweep_lib(): + result = process_findings([finding()], dry_run=True, sweep_lib=None) + + assert result["findings_created"] == 0 + assert result["findings_updated"] == 0 + rows = cast(list[dict[str, Any]], result["rows"]) + assert rows[0]["action"] == "dry-run" + assert isinstance(rows[0]["fingerprint"], str) + + +def test_run_dirty_dry_run_does_not_load_sweep_lib(monkeypatch, tmp_path): + import doc_steward.entrypoint_freshness_sweep as sweep + + monkeypatch.setattr(sweep, "check_entrypoint_freshness", lambda repo_root: [finding()]) + monkeypatch.setattr( + sweep, + "_load_sweep_lib", + lambda: (_ for _ in ()).throw(AssertionError("sweep_lib should not load")), + ) + + result = run(tmp_path, dry_run=True, json_output=False, record_run=False) + + assert result["findings"] == 1 + rows = cast(list[dict[str, Any]], result["rows"]) + assert rows[0]["action"] == "dry-run" + + +def test_cli_rejects_dry_run_record_run(tmp_path): + import pytest + + from doc_steward.entrypoint_freshness_sweep import main + + with pytest.raises(SystemExit) as excinfo: + main([str(tmp_path), "--dry-run", "--record-run"]) + + assert excinfo.value.code == 2 + + +def test_run_rejects_dry_run_record_run_before_loading_sweep_lib(monkeypatch, tmp_path): + import pytest + + import doc_steward.entrypoint_freshness_sweep as sweep + + monkeypatch.setattr( + sweep, + "_load_sweep_lib", + lambda: (_ for _ in ()).throw(AssertionError("sweep_lib should not load")), + ) + + with pytest.raises(ValueError): + run(tmp_path, dry_run=True, record_run=True) + + +def test_doc_steward_cli_rejects_dry_run_record_run(monkeypatch, tmp_path): + import pytest + + import doc_steward.entrypoint_freshness_sweep as sweep + from doc_steward import cli + + monkeypatch.setattr( + sweep, + "_load_sweep_lib", + lambda: (_ for _ in ()).throw(AssertionError("sweep_lib should not load")), + ) + monkeypatch.setattr( + "sys.argv", + [ + "doc_steward.cli", + "entrypoint-freshness-sweep", + str(tmp_path), + "--dry-run", + "--record-run", + ], + ) + + with pytest.raises(ValueError): + cli.main()