"""Selective memory promotion guardrail for PRD Plugin. The default ``promotion_policy: all`` keeps the legacy behaviour (every observation becomes a ``MEM-*`` record). The ``promotion_policy: selective`` mode applies a confidence and evidence-count threshold so the durable memory cannot grow unbounded. A weekly ``gc`` pass flags stale candidates older than ``weekly_gc_stale_after_days``. """ from __future__ import annotations import argparse import json import re from datetime import datetime, timezone from pathlib import Path from typing import Any, Iterable POLICIES = {"all", "selective"} CONFIDENCE_RANK = {"low": 1, "medium": 2, "high": 3, "unknown": 0} DEFAULT_MEMORY_CONFIG = { "promotion_policy": "all", "selective_min_confidence": "high", "selective_min_evidence_count": 2, "weekly_gc_stale_after_days": 90, } DATE_PATTERN = re.compile(r"^\d{4}-\d{2}-\d{2}$") def _read_json(path: Path) -> dict[str, Any]: return json.loads(path.read_text(encoding="utf-8-sig")) def _resolve_config_path(repo_root: Path) -> Path: return repo_root / ".prd_plugin" / "config.json" def load_memory_config(repo_root: Path) -> dict[str, Any]: """Return the merged memory config (file overrides + defaults).""" config_path = _resolve_config_path(repo_root) if not config_path.exists(): return dict(DEFAULT_MEMORY_CONFIG) config = _read_json(config_path) if not isinstance(config, dict): return dict(DEFAULT_MEMORY_CONFIG) user_memory = config.get("memory", {}) or {} if not isinstance(user_memory, dict): return dict(DEFAULT_MEMORY_CONFIG) merged = dict(DEFAULT_MEMORY_CONFIG) for key, value in user_memory.items(): if value is not None: merged[key] = value return merged def _confidence_rank(level: str) -> int: return CONFIDENCE_RANK.get(str(level).lower(), 0) def _parse_date(value: str) -> datetime | None: if not value or not isinstance(value, str) or not DATE_PATTERN.match(value): return None try: return datetime.strptime(value, "%Y-%m-%d").replace(tzinfo=timezone.utc) except ValueError: return None def evaluate_candidate(candidate: dict[str, Any], memory_config: dict[str, Any]) -> dict[str, Any]: """Return a per-candidate decision: accepted, reasons, gc info.""" reasons: list[str] = [] if memory_config.get("promotion_policy") != "selective": accepted = True else: candidate_confidence = str(candidate.get("confidence", "unknown")).lower() min_confidence = str(memory_config.get("selective_min_confidence", "high")).lower() if _confidence_rank(candidate_confidence) < _confidence_rank(min_confidence): reasons.append( f"confidence '{candidate_confidence}' is below selective_min_confidence '{min_confidence}'" ) source_refs = candidate.get("source_refs") or [] min_evidence = int(memory_config.get("selective_min_evidence_count", 2) or 0) if not isinstance(source_refs, list) or len(source_refs) < min_evidence: reasons.append( f"source_refs count is {len(source_refs) if isinstance(source_refs, list) else 0}, " f"below selective_min_evidence_count {min_evidence}" ) accepted = not reasons created_at = str(candidate.get("created_at", "")) return { "id": candidate.get("id"), "accepted": accepted, "reasons": reasons, "created_at": created_at, } def evaluate_gc(candidates: Iterable[dict[str, Any]], memory_config: dict[str, Any], now: datetime | None = None) -> dict[str, Any]: """Flag candidates older than ``weekly_gc_stale_after_days`` as stale.""" stale_after_days = int(memory_config.get("weekly_gc_stale_after_days", 90) or 90) now = now or datetime.now(timezone.utc) stale: list[dict[str, Any]] = [] for candidate in candidates: candidate_date = _parse_date(str(candidate.get("created_at", ""))) if candidate_date is None: continue age_days = (now - candidate_date).days if age_days > stale_after_days: stale.append( { "id": candidate.get("id"), "created_at": candidate.get("created_at"), "age_days": age_days, } ) return { "stale_after_days": stale_after_days, "checked_at": now.strftime("%Y-%m-%d"), "stale_count": len(stale), "stale_candidates": stale, } def evaluate(repo_root: Path, candidates: list[dict[str, Any]], now: datetime | None = None) -> dict[str, Any]: """Return a per-candidate accept/reject decision plus a gc rollup.""" memory_config = load_memory_config(repo_root) policy = str(memory_config.get("promotion_policy", "all")).lower() fallback_reason = "" if policy not in POLICIES: fallback_reason = ( f"unknown promotion_policy '{memory_config.get('promotion_policy')}'; " f"falling back to 'all'" ) policy = "all" per_candidate = [evaluate_candidate(c, memory_config) for c in candidates] accepted = [c for c in per_candidate if c["accepted"]] rejected = [c for c in per_candidate if not c["accepted"]] return { "policy": policy, "fallback_reason": fallback_reason, "memory_config": memory_config, "candidates": per_candidate, "accepted_count": len(accepted), "rejected_count": len(rejected), "accepted_ids": [c["id"] for c in accepted], "rejected_ids": [c["id"] for c in rejected], "gc": evaluate_gc(candidates, memory_config, now=now), "note": ( "selective promotion: candidates must meet confidence and evidence thresholds; " "stale candidates flagged for weekly GC review." if policy == "selective" else "all promotion: every candidate accepted; weekly GC still applies." ), } def format_markdown(decision: dict[str, Any]) -> str: lines = [ "# PRD Plugin Selective Promote", "", f"Policy: `{decision['policy']}`", f"Accepted: `{decision['accepted_count']}`", f"Rejected: `{decision['rejected_count']}`", f"Stale: `{decision['gc']['stale_count']}` (older than {decision['gc']['stale_after_days']} days)", "", decision.get("note", ""), "", ] if decision.get("fallback_reason"): lines.append(f"Fallback: {decision['fallback_reason']}") lines.append("") if decision.get("candidates"): lines.append("## Candidates") for c in decision["candidates"]: status = "ACCEPT" if c["accepted"] else "REJECT" reasons = "; ".join(c["reasons"]) if c["reasons"] else "" lines.append(f"- {status} {c['id']} {reasons}") lines.append("") if decision["gc"]["stale_candidates"]: lines.append("## Stale Candidates (Weekly GC)") for s in decision["gc"]["stale_candidates"]: lines.append(f"- {s['id']} created {s['created_at']} ({s['age_days']} days old)") return "\n".join(lines) def main(argv: list[str] | None = None) -> int: parser = argparse.ArgumentParser(description="PRD Plugin selective memory promotion guardrail.") parser.add_argument("--repo-root", default=".") parser.add_argument( "--candidates", default=".prd_plugin/state/memory.json", help="Path to candidates JSON file (list of MEM-* records).", ) parser.add_argument("--format", choices=("json", "markdown"), default="json") args = parser.parse_args(argv) repo_root = Path(args.repo_root).resolve() candidates_path = Path(args.candidates) if candidates_path.exists(): data = json.loads(candidates_path.read_text(encoding="utf-8-sig")) if isinstance(data, dict) and isinstance(data.get("records"), list): candidates = data["records"] elif isinstance(data, list): candidates = data else: candidates = [] else: candidates = [] decision = evaluate(repo_root, candidates) if args.format == "markdown": print(format_markdown(decision), end="") else: print(json.dumps(decision, indent=2)) return 0 if __name__ == "__main__": raise SystemExit(main())