"""Repo self-audit for PRD Plugin. Walks a single repo and produces a structured rule-vs-reality gap analysis: - extracts rules from AGENTS.md / CLAUDE.md by section - mines TODO / FIXME / MOCK / PLACEHOLDER / HACK / WORKAROUND / STUB markers in source code with file/line/context and permitted-zone detection - reads PRD Plugin state files for orphan IDs, stale records, missing traceability, and hub-vs-downstream contradictions - optionally analyzes raw session JSONL and promoted observations - persists actionable REQ-* and HLT-* records """ from __future__ import annotations import argparse import hashlib import json import re from dataclasses import dataclass, field from datetime import datetime, timezone from pathlib import Path from typing import Any, Iterable RULE_FILES = ("AGENTS.md", "CLAUDE.md") ALL_ID_PATTERN = re.compile( r"\b[A-Z]{2,}(?:-[A-Z]{2,})*-\d{3,}\b" ) # Debt markers and their categories, ported from Improve mining-tier2.ps1. DEBT_MARKER_CATEGORIES: dict[str, str] = { "FIXME": "known-defect", "BUG": "known-defect", "BROKEN": "known-defect", "TODO": "incomplete", "WIP": "incomplete", "PENDING": "incomplete", "HACK": "shortcut", "WORKAROUND": "shortcut", "XXX": "risk-flag", "STUB": "placeholder", "STUBS": "placeholder", "STUBBS": "placeholder", "PLACEHOLDER": "placeholder", "MOCK": "placeholder", "NOT_IMPLEMENTED": "placeholder", "NOT-IMPLEMENTED": "placeholder", "REVERT": "rollback", "WONT-FIX": "abandoned", "WONTDO": "abandoned", "DEPRECATED": "abandoned", "TEMP": "temporary", "TEMPFILE": "temporary", } # Directories and files to skip when mining source. EXCLUDED_DIR_NAMES = { "node_modules", ".git", ".venv", "venv", "target", "dist", "build", "obj", "bin", "out", "worktrees", "reruns", ".vs", ".idea", "__pycache__", ".zeusgrid", ".opencode-state", ".opencode", ".agents", ".agent-colab", ".cache", ".cargo", ".gradle", ".turbo", ".next", ".vercel", ".mypy_cache", ".pytest_cache", ".ruff_cache", } EXCLUDED_PATH_SUBSTRINGS = { "\\node_modules\\", "/node_modules/", "\\.git\\", "/.git/", "\\.prd_plugin\\state\\", "/.prd_plugin/state/", "\\.prd_plugin\\cache\\", "/.prd_plugin/cache/", "\\.prd_plugin\\local\\", "/.prd_plugin/local/", "\\.prd_plugin\\inbox\\", "/.prd_plugin/inbox/", "\\.prd_plugin\\outbox\\", "/.prd_plugin/outbox/", "\\.prd_plugin\\mailboxes\\", "/.prd_plugin/mailboxes/", "\\.opencode-state\\", "/.opencode-state/", "pnpm-lock.yaml", "package-lock.json", "yarn.lock", "Cargo.lock", "composer.lock", "poetry.lock", "Pipfile.lock", "Gemfile.lock", } SOURCE_SUFFIXES = { ".py", ".js", ".jsx", ".ts", ".tsx", ".mjs", ".cjs", ".rs", ".go", ".java", ".kt", ".swift", ".m", ".mm", ".cs", ".fs", ".vb", ".cpp", ".c", ".h", ".hpp", ".cc", ".rb", ".php", ".sh", ".ps1", ".psm1", ".bat", ".cmd", ".sql", ".toml", ".yaml", ".yml", ".vue", ".svelte", ".scss", ".css", ".md", ".rst", ".txt", } PERMITTED_ZONE_PATTERNS = ( r"(^|/|\\)tests?(/|\\)", r"(^|/|\\)fixtures?(/|\\)", r"(^|/|\\)skeleton(/|\\)", r"(^|/|\\)templates[/\\]repo-skeleton[/\\]", r"(^|/\\)_?test_[^/\\]*\.py$", r"test_.*\.py$", r".*_test\.py$", ) HUB_ONLY_INSTALL_SCOPES = {"plugin_development", "hub_runtime"} # Minimal fallback lists used only when the authoritative install-scope # manifests cannot be loaded. These must match the scopes above, not # downstream_optional or downstream_runtime helpers. _FALLBACK_HUB_ONLY_SCRIPTS = { "gap_audit.py", "release_check.py", "local_workflow_check.py", "version_advice.py", "request_import.py", "request_mailbox.py", "prd_install.py", } _FALLBACK_HUB_ONLY_SKILLS = {"project-fold-it-in"} SIGNIFICANT_VIOLATION_THRESHOLD = 3 def _load_hub_only_set(repo_root: Path, manifest_name: str, entry_key: str) -> set[str]: """Load names whose install_scope is hub-only from a PRD install-scope manifest.""" manifest = _read_json(repo_root / "templates" / manifest_name) if manifest and isinstance(manifest, dict): entries = manifest.get(entry_key, {}) if isinstance(entries, dict): return { name for name, meta in entries.items() if isinstance(meta, dict) and meta.get("install_scope") in HUB_ONLY_INSTALL_SCOPES } return set() def hub_only_scripts(repo_root: Path) -> set[str]: loaded = _load_hub_only_set(repo_root, "script-install-scope.json", "scripts") return loaded if loaded else _FALLBACK_HUB_ONLY_SCRIPTS.copy() def hub_only_skills(repo_root: Path) -> set[str]: loaded = _load_hub_only_set(repo_root, "skill-install-scope.json", "skills") return loaded if loaded else _FALLBACK_HUB_ONLY_SKILLS.copy() @dataclass(frozen=True) class Rule: phrase: str source_file: str section_path: tuple[str, ...] line: int confidence: str # explicit or inferred forbidden_markers: tuple[str, ...] category: str @dataclass class MarkerHit: file: str line: int marker: str category: str snippet: str permitted_zone: bool @dataclass class SourceRef: path: str line: int | None = None note: str = "" @dataclass class Finding: category: str severity: str # critical, high, medium, low, info summary: str recommendation: str rule_phrase: str | None = None source_refs: list[SourceRef] = field(default_factory=list) affected_ids: list[str] = field(default_factory=list) related_findings: list[str] = field(default_factory=list) def fingerprint(self) -> str: parts = [self.category, self.rule_phrase or "", self.summary[:80]] if self.source_refs: parts.append(self.source_refs[0].path) return hashlib.sha1("|".join(parts).encode("utf-8")).hexdigest()[:16] def _read_text(path: Path) -> str: try: return path.read_text(encoding="utf-8", errors="ignore") except OSError: return "" def _read_json(path: Path) -> dict[str, Any] | None: try: return json.loads(path.read_text(encoding="utf-8-sig")) except (OSError, json.JSONDecodeError): return None def _read_jsonl(path: Path) -> Iterable[dict[str, Any]]: try: for line in path.read_text(encoding="utf-8-sig").splitlines(): line = line.strip() if not line: continue try: yield json.loads(line) except json.JSONDecodeError: continue except OSError: return def _is_excluded_path(path: Path, repo_root: Path) -> bool: rel = path.relative_to(repo_root).as_posix() if any(part in EXCLUDED_DIR_NAMES for part in path.parts): return True if any(sub in rel for sub in EXCLUDED_PATH_SUBSTRINGS): return True if path.suffix.lower() not in SOURCE_SUFFIXES: return True return False def _is_permitted_zone(path: Path, repo_root: Path) -> bool: rel = path.relative_to(repo_root).as_posix() return any(re.search(pattern, rel) for pattern in PERMITTED_ZONE_PATTERNS) def _iter_source_files(repo_root: Path) -> Iterable[Path]: for path in sorted(repo_root.rglob("*")): if not path.is_file(): continue if _is_excluded_path(path, repo_root): continue yield path def _parse_markdown_sections(text: str, filename: str) -> list[dict[str, Any]]: sections = [] current = {"level": 0, "heading": "", "path": [], "body": [], "line": 0} for i, line in enumerate(text.splitlines(), start=1): match = re.match(r"^(#{1,6})\s+(.*)$", line) if match: if current["body"] or current["heading"]: sections.append({ "level": current["level"], "heading": current["heading"], "path": tuple(current["path"]), "body": "\n".join(current["body"]), "line": current["line"], "file": filename, }) level = len(match.group(1)) heading = match.group(2).strip() path = current["path"][: level - 1] + [heading] current = {"level": level, "heading": heading, "path": path, "body": [], "line": i} else: current["body"].append(line) if current["body"] or current["heading"]: sections.append({ "level": current["level"], "heading": current["heading"], "path": tuple(current["path"]), "body": "\n".join(current["body"]), "line": current["line"], "file": filename, }) return sections def extract_rules(repo_root: Path) -> list[Rule]: rules: list[Rule] = [] seen: set[tuple[str, str]] = set() for filename in RULE_FILES: path = repo_root / filename if not path.is_file(): continue text = _read_text(path) sections = _parse_markdown_sections(text, filename) for section in sections: body = (section["heading"] + "\n" + section["body"]).lower() detectors = [ ("no todos", ("TODO", "WIP", "PENDING"), "debt"), ("no mocks", ("MOCK",), "debt"), ("no stubs", ("STUB", "STUBS", "STUBBS"), "debt"), ("no placeholders", ("PLACEHOLDER", "NOT_IMPLEMENTED", "NOT-IMPLEMENTED"), "debt"), ("no hacks", ("HACK", "XXX"), "debt"), ("no workarounds", ("WORKAROUND",), "debt"), ("verify with real systems", ("MOCK",), "testing"), ("tdd", ("TODO",), "testing"), ("claim + evidence", (), "evidence"), ("no fake completion", (), "evidence"), ("check existing credentials", (), "security"), ] for phrase, markers, category in detectors: pattern = re.compile(r"\b" + re.escape(phrase.lower()) + r"\b") if pattern.search(body) and (filename, phrase) not in seen: rules.append(Rule( phrase=phrase, source_file=filename, section_path=section["path"], line=section["line"], confidence="explicit", forbidden_markers=markers, category=category, )) seen.add((filename, phrase)) heading_lower = section["heading"].lower() inferred_map = { "no todo": ("TODO",), "no mock": ("MOCK",), "no stub": ("STUB", "STUBS"), "no placeholder": ("PLACEHOLDER",), "no hack": ("HACK",), "no workaround": ("WORKAROUND",), } for key, markers in inferred_map.items(): if key in heading_lower and (filename, key) not in seen: rules.append(Rule( phrase=key, source_file=filename, section_path=section["path"], line=section["line"], confidence="inferred", forbidden_markers=markers, category="debt", )) seen.add((filename, key)) return rules def mine_debt_markers(repo_root: Path) -> list[MarkerHit]: hits: list[MarkerHit] = [] for path in _iter_source_files(repo_root): text = _read_text(path) if not text: continue lines = text.splitlines() rel = path.relative_to(repo_root).as_posix() permitted = _is_permitted_zone(path, repo_root) for line_no, line in enumerate(lines, start=1): for marker, category in DEBT_MARKER_CATEGORIES.items(): if marker in ("BUG", "BROKEN", "PENDING", "DEPRECATED", "TEMP", "TEMPFILE"): if not re.search(rf"\b{re.escape(marker)}\b", line): continue if marker not in [w for w in re.findall(r"\b\w+\b", line) if w.isupper()]: continue else: if not re.search(rf"\b{re.escape(marker)}\b", line, re.IGNORECASE): continue snippet = line.strip() if len(snippet) > 160: snippet = snippet[:157] + "..." hits.append(MarkerHit( file=rel, line=line_no, marker=marker, category=category, snippet=snippet, permitted_zone=permitted, )) return hits def _walk_values(value: Any) -> Iterable[Any]: if isinstance(value, dict): for child in value.values(): yield from _walk_values(child) elif isinstance(value, list): for child in value: yield from _walk_values(child) else: yield value def _extract_id_references(value: Any) -> set[str]: refs: set[str] = set() for v in _walk_values(value): if isinstance(v, str): refs.update(ALL_ID_PATTERN.findall(v)) return refs def _parse_time(value: Any) -> datetime | None: if not isinstance(value, str) or "T" not in value: return None try: parsed = datetime.fromisoformat(value.replace("Z", "+00:00")) except ValueError: return None if parsed.tzinfo is None: parsed = parsed.replace(tzinfo=timezone.utc) return parsed.astimezone(timezone.utc) def _state_files(repo_root: Path) -> Iterable[Path]: state_dir = repo_root / ".prd_plugin" / "state" if state_dir.exists(): for path in sorted(state_dir.rglob("*")): if path.is_file() and path.suffix in (".json", ".jsonl"): yield path promoted = repo_root / "sessions" / "shared" / "promoted-session-summaries.jsonl" if promoted.is_file(): yield promoted def collect_canonical_ids(repo_root: Path) -> dict[str, set[str]]: ids_by_prefix: dict[str, set[str]] = {} for path in _state_files(repo_root): if path.suffix == ".json": data = _read_json(path) if data is None: continue for identifier in _extract_id_references(data): prefix = identifier.rsplit("-", 1)[0] ids_by_prefix.setdefault(prefix, set()).add(identifier) elif path.suffix == ".jsonl": for obj in _read_jsonl(path): for identifier in _extract_id_references(obj): prefix = identifier.rsplit("-", 1)[0] ids_by_prefix.setdefault(prefix, set()).add(identifier) return ids_by_prefix def _load_state_object(path: Path) -> dict[str, Any] | list[Any] | None: if path.suffix == ".json": return _read_json(path) items: list[dict[str, Any]] = [] for obj in _read_jsonl(path): if isinstance(obj, dict): items.append(obj) return items def _collect_linked_ids(obj: Any) -> set[str]: linked: set[str] = set() if not isinstance(obj, dict): return linked for key in ( "linked_ids", "affected_ids", "linked_health_findings", "source_request_id", "upstream_request_id", "graduated_to", "supersedes", "superseded_by", "affected_areas", ): value = obj.get(key) if isinstance(value, list): for item in value: if isinstance(item, str): linked.update(ALL_ID_PATTERN.findall(item)) elif isinstance(item, dict): linked.update(_extract_id_references(item)) elif isinstance(value, str): linked.update(ALL_ID_PATTERN.findall(value)) return linked def mine_state_inconsistencies( repo_root: Path, canonical_ids: dict[str, set[str]], stale_days: int, now: datetime, ) -> list[Finding]: findings: list[Finding] = [] all_ids: set[str] = set() for group in canonical_ids.values(): all_ids.update(group) # Orphan / missing IDs for path in _state_files(repo_root): data = _load_state_object(path) if data is None: continue rel = path.relative_to(repo_root).as_posix() items = data if isinstance(data, list) else [data] for item in items: if not isinstance(item, dict): continue item_id = item.get("id") linked = _collect_linked_ids(item) for ref in linked: if ref not in all_ids: findings.append(Finding( category="orphan_id", severity="high", summary=f"{item_id or 'record'} in {rel} references missing {ref}.", recommendation="Create the missing canonical record or remove the stale reference.", source_refs=[SourceRef(path=rel, note=f"references {ref}")], affected_ids=[item_id] if isinstance(item_id, str) else [], )) # Stale records cutoff = now.timestamp() - stale_days * 86400 for path in _state_files(repo_root): data = _load_state_object(path) if data is None: continue rel = path.relative_to(repo_root).as_posix() items = data if isinstance(data, list) else [data] for item in items: if not isinstance(item, dict): continue item_id = item.get("id") status = item.get("status") created = _parse_time(item.get("created_at")) updated = _parse_time(item.get("updated_at")) ts = updated or created if ts and ts.timestamp() < cutoff and status in ("open", "proposed", "in_review", "needs_info"): findings.append(Finding( category="stale_record", severity="medium", summary=f"{item_id or 'record'} in {rel} has been {status} since {ts.date().isoformat()}.", recommendation="Resolve, refresh with evidence, or explicitly close the stale record.", source_refs=[SourceRef(path=rel, note=f"status={status}")], affected_ids=[item_id] if isinstance(item_id, str) else [], )) # Missing traceability: requests without health links requests_path = repo_root / ".prd_plugin" / "state" / "requests.json" requests_data = _read_json(requests_path) if requests_data and isinstance(requests_data, dict): for req in requests_data.get("requests", []): if not isinstance(req, dict): continue status = req.get("status") severity = req.get("severity", "").lower() health_links = req.get("linked_health_findings", []) or [] if status in ("accepted", "proposed") and severity in ("high", "critical", "medium") and not health_links: findings.append(Finding( category="missing_traceability", severity="medium", summary=f"{req.get('id')} ({severity}) has no linked health findings.", recommendation="Create an HLT-* finding or link an existing one to track risk.", source_refs=[SourceRef(path=".prd_plugin/state/requests.json", note=f"{req.get('id')}")], affected_ids=[req.get("id")] if isinstance(req.get("id"), str) else [], )) # Missing traceability: health without request links health_path = repo_root / ".prd_plugin" / "state" / "health.json" health_data = _read_json(health_path) if health_data and isinstance(health_data, dict): for finding in health_data.get("findings", []): if not isinstance(finding, dict): continue if finding.get("status") != "open": continue affected = finding.get("affected_ids", []) or [] if not affected: findings.append(Finding( category="missing_traceability", severity="medium", summary=f"{finding.get('id')} is open but has no affected_ids.", recommendation="Link the health finding to the REQ-* or TRK-* it tracks.", source_refs=[SourceRef(path=".prd_plugin/state/health.json", note=f"{finding.get('id')}")], affected_ids=[finding.get("id")] if isinstance(finding.get("id"), str) else [], )) return findings def mine_hub_downstream_issues(repo_root: Path) -> list[Finding]: findings: list[Finding] = [] skeleton_agents = repo_root / "templates" / "repo-skeleton" / ".agents" / "skills" skeleton_opencode = repo_root / "templates" / "repo-skeleton" / ".opencode" / "skill" hub_scripts = hub_only_scripts(repo_root) hub_skills = hub_only_skills(repo_root) # Hub-only skills in downstream skeleton for base in (skeleton_agents, skeleton_opencode): if not base.is_dir(): continue for skill_dir in base.iterdir(): if skill_dir.name in hub_skills: rel = skill_dir.relative_to(repo_root).as_posix() findings.append(Finding( category="hub_downstream_mismatch", severity="high", summary=f"Hub-only skill {skill_dir.name} found in downstream skeleton.", recommendation=f"Remove {skill_dir.name} from the downstream template; it is plugin-development/hub-only.", source_refs=[SourceRef(path=rel)], )) # Hub-only scripts referenced in downstream skills for base in (skeleton_agents, skeleton_opencode): if not base.is_dir(): continue for skill_file in base.rglob("SKILL.md"): text = _read_text(skill_file) document_context = text.lower() document_guarded = bool( re.search(r"\brun\s+helper\s+scripts\s+from\b[^\n.]*\bhub\b", document_context) or re.search(r"\bdo\s+not\s+(?:copy|run)\b[^\n.]*\bhub-only\b", document_context) or re.search(r"\bhub-only\s+script\b[^\n.]*\bdo\s+not\s+run\b", document_context) ) rel = skill_file.relative_to(repo_root).as_posix() for script in hub_scripts: for m in re.finditer(rf"\b{re.escape(script)}\b", text): if document_guarded: continue paragraph_start = text.rfind("\n\n", 0, m.start()) + 2 paragraph_end = text.find("\n\n", m.end()) if paragraph_end < 0: paragraph_end = len(text) context = text[paragraph_start:paragraph_end].lower() guarded = any(guard in context for guard in ( "do not run", "must not run", "hub-only", "hub only", "do not call", )) hub_owned_action = bool( # Allow a qualifier between "hub" and the verb, e.g. # "Hub development uses ..." - the sentence still scopes # the script to the hub, which is what matters. re.search(r"\b(?:the\s+)?hub\b[\w\s-]{0,24}?\b(?:imports|publishes|runs|uses)\b", context) or re.search(r"\bfrom\b[^\n.]*\b(?:plugin\s+)?hub\b", context) ) # A provenance note records where a file CAME FROM; it is not # an instruction, so it cannot cause a downstream agent to run # a hub-only script. Flagging it would push authors to edit # skills purely to appease the detector. provenance = bool( re.search(r"\bgenerated\s+(?:from|by)\b", context) or "do not edit by hand" in context ) if guarded or hub_owned_action or provenance: continue findings.append(Finding( category="hub_downstream_mismatch", severity="high", summary=f"Downstream skill {rel} references hub-only script {script} without a guard.", recommendation="Remove the reference or add an explicit 'do not run / hub-only' guard.", source_refs=[SourceRef(path=rel, line=text[:m.start()].count("\n") + 1, note=f"mentions {script}")], )) break return findings def _normalize_error_stem(text: str) -> str: stem = text[:120] stem = re.sub(r"\d+", "#", stem) stem = re.sub(r"[A-Za-z]:[\\/][^\s]+", "", stem) stem = re.sub(r"\b[a-f0-9]{8,}\b", "", stem, flags=re.IGNORECASE) return stem def mine_sessions(repo_root: Path) -> list[Finding]: findings: list[Finding] = [] sessions_dir = repo_root / ".prd_plugin" / "local" / "sessions" if not sessions_dir.is_dir(): return findings error_stems: dict[str, int] = {} high_error_sessions: list[tuple[str, float, int]] = [] for session_file in sorted(sessions_dir.rglob("*.jsonl")): rel = session_file.relative_to(repo_root).as_posix() calls = 0 errors = 0 for entry in _read_jsonl(session_file): if not isinstance(entry, dict): continue t = entry.get("type") or entry.get("role") if t == "assistant" and isinstance(entry.get("message"), dict): content = entry["message"].get("content", []) if isinstance(content, list): for c in content: if isinstance(c, dict) and c.get("type") == "tool_use": calls += 1 if t == "user" and isinstance(entry.get("message"), dict): content = entry["message"].get("content", []) if isinstance(content, list): for c in content: if isinstance(c, dict) and c.get("type") == "tool_result": calls += 1 if c.get("is_error"): errors += 1 content_data = c.get("content") if isinstance(content_data, str): stem = _normalize_error_stem(content_data) error_stems[stem] = error_stems.get(stem, 0) + 1 elif isinstance(content_data, list) and content_data and isinstance(content_data[0], dict): stem = _normalize_error_stem(content_data[0].get("text", "")) error_stems[stem] = error_stems.get(stem, 0) + 1 if calls > 0: rate = errors / calls if rate >= 0.05 or errors >= 5: high_error_sessions.append((rel, rate, errors)) if high_error_sessions: for rel, rate, errors in high_error_sessions[:5]: findings.append(Finding( category="session_error_pattern", severity="medium", summary=f"Session {rel} has {errors} tool errors ({rate:.1%} error rate).", recommendation="Review the session for repeated tool-usage mistakes and consider a skill or guardrail update.", source_refs=[SourceRef(path=rel)], )) for stem, count in sorted(error_stems.items(), key=lambda x: x[1], reverse=True)[:5]: if count < 2: continue findings.append(Finding( category="session_error_pattern", severity="medium", summary=f"Recurring tool error stem ({count} times): {stem}", recommendation="Add a skill guardrail or workflow step that prevents this error pattern.", source_refs=[SourceRef(path=".prd_plugin/local/sessions", note="aggregated across sessions")], )) return findings def mine_promoted_observations(repo_root: Path) -> list[Finding]: findings: list[Finding] = [] promoted_path = repo_root / "sessions" / "shared" / "promoted-session-summaries.jsonl" if not promoted_path.is_file(): return findings for obs in _read_jsonl(promoted_path): if not isinstance(obs, dict): continue obs_id = obs.get("id") status = obs.get("status") refs = obs.get("source_refs", []) or [] if status == "promoted" and not refs: findings.append(Finding( category="stale_record", severity="medium", summary=f"Promoted observation {obs_id} has no source_refs.", recommendation="Add source_refs linking the observation back to session evidence or demote it.", source_refs=[SourceRef(path="sessions/shared/promoted-session-summaries.jsonl", note=obs_id)], affected_ids=[obs_id] if isinstance(obs_id, str) else [], )) return findings def build_rule_violation_findings(rules: list[Rule], hits: list[MarkerHit]) -> list[Finding]: findings: list[Finding] = [] by_marker: dict[str, list[MarkerHit]] = {} for hit in hits: by_marker.setdefault(hit.marker, []).append(hit) for rule in rules: relevant = [] for marker in rule.forbidden_markers: relevant.extend(by_marker.get(marker, [])) relevant = [h for h in relevant if not h.permitted_zone] if not relevant: continue count = len(relevant) if count >= 20: severity = "critical" elif count >= 10: severity = "high" elif count >= 3: severity = "medium" else: severity = "low" top = relevant[:5] refs = [SourceRef(path=h.file, line=h.line, note=h.snippet) for h in top] findings.append(Finding( category="rule_violation", severity=severity, summary=f"Rule '{rule.phrase}' is violated {count} time(s) outside test/fixture/skeleton zones.", recommendation=f"Enforce '{rule.phrase}' with a CI/pre-commit check or revise the rule.", rule_phrase=rule.phrase, source_refs=refs, )) return findings def _next_id(records: list[dict[str, Any]], prefix: str) -> int: max_id = 0 for record in records: rid = str(record.get("id", "")) match = re.match(rf"{prefix}-(\d+)$", rid) if match: max_id = max(max_id, int(match.group(1))) return max_id + 1 def persist_findings( repo_root: Path, findings: list[Finding], ) -> dict[str, Any]: repo_root = Path(repo_root) state_dir = repo_root / ".prd_plugin" / "state" state_dir.mkdir(parents=True, exist_ok=True) requests_path = state_dir / "requests.json" requests_data = _read_json(requests_path) or {"schema_version": "0.1", "requests": []} requests = requests_data.setdefault("requests", []) health_path = state_dir / "health.json" health_data = _read_json(health_path) or { "schema_version": "0.1", "checked_at": None, "checked_by_agent": None, "checked_from_session": None, "status": "ok", "findings": [], } health_findings = health_data.setdefault("findings", []) created_req = 0 created_hlt = 0 updated_req = 0 existing_by_fp: dict[str, dict[str, Any]] = {} for req in requests: if not isinstance(req, dict): continue if req.get("status") in ("proposed", "in_review", "accepted"): fp = req.get("self_audit_fingerprint") if fp: existing_by_fp[fp] = req now = datetime.now().strftime("%Y-%m-%d") checked_at = datetime.now(timezone.utc).isoformat() for finding in findings: if finding.severity in ("info", "low"): continue fp = finding.fingerprint() existing = existing_by_fp.get(fp) if existing: existing["last_seen_at"] = now existing["updated_at"] = now updated_req += 1 continue req_num = _next_id(requests, "REQ") req_id = f"REQ-{req_num:03d}" hlt_num = _next_id(health_findings, "HLT") hlt_id = f"HLT-{hlt_num:03d}" request_type = "method" if finding.category == "hub_downstream_mismatch": request_type = "compatibility" elif finding.category in ("orphan_id", "stale_record", "missing_traceability"): request_type = "process" source_refs = [ {"path": ref.path, "line": ref.line, "note": ref.note} for ref in finding.source_refs ] requests.append({ "id": req_id, "status": "proposed", "request_type": request_type, "scope": "local", "origin_repo": repo_root.name, "visibility": "repo", "upstream_submission": False, "upstream_request_id": None, "source_request_id": None, "summary": finding.summary, "rationale": finding.recommendation, "severity": finding.severity, "requested_by_agent": "AGENT-001", "requested_from_session": "SES-AUDIT", "affected_versions": [], "affected_areas": list({ref.path for ref in finding.source_refs}) or ["scripts/prd_self_audit.py"], "linked_ids": finding.affected_ids, "linked_health_findings": [hlt_id], "source_refs": source_refs, "proposed_actions": [finding.recommendation], "reproduction_steps": ["Run python scripts/prd_self_audit.py --repo-root . --persist"], "expected_behavior": "No gap of this kind exists.", "actual_behavior": finding.summary, "workaround": None, "regression": False, "risk": finding.severity, "created_at": now, "updated_at": now, "last_seen_at": now, "reviewed_at": None, "reviewed_by_agent": None, "decision": None, "decision_rationale": None, "thread": {"status": "open", "participants": [], "messages": []}, "graduated_to": [], "self_audit_fingerprint": fp, }) health_findings.append({ "id": hlt_id, "severity": finding.severity, "status": "open", "summary": finding.summary, "affected_ids": [req_id] + finding.affected_ids, "source_refs": source_refs, "recommended_action": finding.recommendation, "created_at": now, "resolved_at": None, }) existing_by_fp[fp] = requests[-1] created_req += 1 created_hlt += 1 if created_req or updated_req: requests_path.write_text(json.dumps(requests_data, indent=2) + "\n", encoding="utf-8") if created_hlt: health_data["checked_at"] = checked_at health_data["checked_by_agent"] = "AGENT-001" health_data["checked_from_session"] = "SES-AUDIT" if any(f.get("severity") == "critical" for f in health_findings): health_data["status"] = "attention_needed" elif any(f.get("severity") in ("high", "medium") for f in health_findings): health_data["status"] = "warning" else: health_data["status"] = "ok" health_path.write_text(json.dumps(health_data, indent=2) + "\n", encoding="utf-8") return { "created_requests": created_req, "created_health": created_hlt, "updated_requests": updated_req, } def run_audit( repo_root: Path, stale_days: int = 30, include_sessions: bool = False, ) -> dict[str, Any]: now = datetime.now(timezone.utc) rules = extract_rules(repo_root) hits = mine_debt_markers(repo_root) canonical_ids = collect_canonical_ids(repo_root) findings: list[Finding] = [] findings.extend(build_rule_violation_findings(rules, hits)) findings.extend(mine_state_inconsistencies(repo_root, canonical_ids, stale_days, now)) findings.extend(mine_hub_downstream_issues(repo_root)) findings.extend(mine_promoted_observations(repo_root)) if include_sessions: findings.extend(mine_sessions(repo_root)) return { "repo": str(repo_root), "checked_at": now.strftime("%Y-%m-%d"), "rules": [ { "phrase": r.phrase, "source_file": r.source_file, "section_path": r.section_path, "line": r.line, "confidence": r.confidence, "category": r.category, } for r in rules ], "marker_count": len(hits), "marker_count_excluding_permitted": len([h for h in hits if not h.permitted_zone]), "findings": findings, "summary": { "total_findings": len(findings), "by_severity": { "critical": sum(1 for f in findings if f.severity == "critical"), "high": sum(1 for f in findings if f.severity == "high"), "medium": sum(1 for f in findings if f.severity == "medium"), "low": sum(1 for f in findings if f.severity == "low"), "info": sum(1 for f in findings if f.severity == "info"), }, "by_category": {}, }, } def format_markdown(report: dict[str, Any]) -> str: lines = [ "# PRD Plugin Self-Audit", "", f"Repo: `{report['repo']}`", f"Checked at: `{report['checked_at']}`", "", "## Summary", "", f"- Total findings: **{report['summary']['total_findings']}**", f"- Critical: **{report['summary']['by_severity']['critical']}**", f"- High: **{report['summary']['by_severity']['high']}**", f"- Medium: **{report['summary']['by_severity']['medium']}**", f"- Low: **{report['summary']['by_severity']['low']}**", f"- Debt markers: **{report['marker_count']}** ({report['marker_count_excluding_permitted']} outside permitted zones)", "", "## Rules Extracted", "", ] if not report["rules"]: lines.append("- No rules detected.") else: for rule in report["rules"]: section = " / ".join(rule["section_path"]) or "(top)" lines.append( f"- `{rule['phrase']}` ({rule['confidence']}, {rule['category']}) — " f"{rule['source_file']}:{rule['line']} section `{section}`" ) lines.extend(["", "## Findings", ""]) if not report["findings"]: lines.append("- None") else: for finding in report["findings"]: lines.append(f"### {finding.severity.upper()}: {finding.category}") lines.append(finding.summary) lines.append(f"Recommendation: {finding.recommendation}") if finding.source_refs: lines.append("Source refs:") for ref in finding.source_refs: loc = f":{ref.line}" if ref.line else "" lines.append(f" - `{ref.path}{loc}` {ref.note}".strip()) lines.append("") return "\n".join(lines) + "\n" def main(argv: list[str] | None = None) -> int: parser = argparse.ArgumentParser(description="PRD Plugin self-audit runner.") parser.add_argument("--repo-root", default=".") parser.add_argument("--state-dir", default=".prd_plugin/state") parser.add_argument("--sessions-dir", default=".prd_plugin/local/sessions") parser.add_argument("--promoted", default="sessions/shared/promoted-session-summaries.jsonl") parser.add_argument("--output", help="Output file path.") parser.add_argument("--format", choices=("json", "markdown"), default="markdown") parser.add_argument("--persist", action="store_true", help="Write REQ-* and HLT-* records.") parser.add_argument("--include-sessions", action="store_true", help="Analyze raw session JSONL.") parser.add_argument("--stale-days", type=int, default=30) args = parser.parse_args(argv) repo_root = Path(args.repo_root).resolve() report = run_audit(repo_root, stale_days=args.stale_days, include_sessions=args.include_sessions) persisted: dict[str, Any] = {"created_requests": 0, "created_health": 0, "updated_requests": 0} if args.persist: persisted = persist_findings(repo_root, report["findings"]) report["persisted"] = persisted content = json.dumps(report, indent=2, default=lambda o: o.__dict__) + "\n" if args.format == "json" else format_markdown(report) if args.output: output = Path(args.output) output.parent.mkdir(parents=True, exist_ok=True) output.write_text(content, encoding="utf-8") else: print(content, end="") if args.persist and args.format == "markdown": print( f"\nPersisted: {persisted['created_requests']} new REQ-*, " f"{persisted['created_health']} new HLT-*, " f"{persisted['updated_requests']} updated REQ-*." ) return 0 if __name__ == "__main__": raise SystemExit(main())