#!/usr/bin/env python3 """Generic upstream "fork" version drift check for the drift monitor (REQ-081). Some repos that use PRD Plugin also track an external *fork* -- either a running node they connect to (poll its /v1/status) or a fork repo they are pinned to (poll a small published manifest). This check lets such a repo see, on Stop, when a newer fork version is available than the one it is on. It is deliberately generic: PRD Plugin knows nothing fork-specific. It polls a configured JSON URL, reads a configured version field (dotted paths supported), and compares it to a local marker file (default FORK-VERSION). It is: * inert by default (fork.version_check.enabled is false), * a no-op in any repo that is not a fork consumer -- disabled, no source_url, or no local marker file -- so it ships safely to every prd-plugin repo, * TTL-cached and fail-open: a network or parse failure resolves to "no update", never an error that blocks a prompt. Config block (.prd_plugin/config.json): fork.version_check.enabled bool default false fork.version_check.source_url str "" -- a static manifest URL or a node /v1/status fork.version_check.version_field str "fork_version" -- dotted path into the JSON fork.version_check.local_marker str "FORK-VERSION" -- repo-relative marker file fork.version_check.ttl_hours number 6 """ from __future__ import annotations import argparse import json import time import urllib.request from pathlib import Path CACHE_REL = Path(".prd_plugin") / "local" / "fork-version-cache.json" DEFAULT_FIELD = "fork_version" DEFAULT_MARKER = "FORK-VERSION" DEFAULT_TTL_HOURS = 6 def _read_json(path): try: return json.loads(Path(path).read_text(encoding="utf-8-sig")) except Exception: return None def _config(root): cfg = _read_json(Path(root) / ".prd_plugin" / "config.json") return cfg if isinstance(cfg, dict) else {} def _settings(root): fv = _config(root).get("fork", {}) vc = fv.get("version_check", {}) if isinstance(fv, dict) else {} if not isinstance(vc, dict): vc = {} return { "enabled": bool(vc.get("enabled", False)), "source_url": str(vc.get("source_url", "") or ""), "version_field": str(vc.get("version_field", DEFAULT_FIELD) or DEFAULT_FIELD), "local_marker": str(vc.get("local_marker", DEFAULT_MARKER) or DEFAULT_MARKER), "ttl_hours": vc.get("ttl_hours", DEFAULT_TTL_HOURS), } def local_fork_version(root, marker): """The repo's current fork version from its marker file, or None if absent.""" path = Path(root) / marker if not path.is_file(): return None try: text = path.read_text(encoding="utf-8-sig").strip() except Exception: return None return text or None def _extract(payload, field): """Pull a version out of a JSON payload by a dotted field path.""" node = payload for part in str(field).split("."): if isinstance(node, dict) and part in node: node = node[part] else: return None if isinstance(node, (str, int, float)): return str(node) return None def fetch_latest(url, field, timeout=8): """Latest fork version from a JSON URL. Returns a string or raises. Only http(s) URLs are honored; the URL comes from repo config (trusted).""" if not url or not url.startswith(("http://", "https://")): raise RuntimeError("no http(s) source_url configured") req = urllib.request.Request(url, headers={"Accept": "application/json"}) with urllib.request.urlopen(req, timeout=timeout) as resp: # noqa: S310 (config URL) raw = resp.read().decode("utf-8", "replace") data = json.loads(raw) value = _extract(data, field) if value is None: raise RuntimeError(f"field {field!r} not found in fork status response") return value def _cache_path(root): return Path(root) / CACHE_REL def _write_cache(root, data): try: p = _cache_path(root) p.parent.mkdir(parents=True, exist_ok=True) p.write_text(json.dumps(data, indent=2) + "\n", encoding="utf-8") except Exception: pass # best-effort def _newer(current, latest): """True when latest is strictly ahead of current. Numeric when both parse as ints (the FORK-VERSION counter); otherwise a conservative string difference.""" if current is None or latest is None: return False try: return int(str(latest)) > int(str(current)) except (TypeError, ValueError): return str(latest) != str(current) def check(root, fetch=fetch_latest, now=None, force=False): """Return the fork version status, polling only when the cache is stale. Fail-open: any failure resolves to a no-update status with an error note.""" root = Path(root) now = time.time() if now is None else now s = _settings(root) current = local_fork_version(root, s["local_marker"]) base = {"enabled": s["enabled"], "current": current, "latest": None, "update_available": False, "source": s["source_url"]} if not s["enabled"] or not s["source_url"] or current is None: base["reason"] = "not configured" if not (s["enabled"] and s["source_url"]) \ else "no local marker" return base cache = _read_json(_cache_path(root)) ttl = float(s["ttl_hours"]) * 3600.0 if (not force and isinstance(cache, dict) and cache.get("latest") and cache.get("source_url") == s["source_url"] and (now - float(cache.get("checked_at", 0))) < ttl): latest = cache.get("latest") return {**base, "latest": latest, "update_available": _newer(current, latest), "checked_at": cache.get("checked_at"), "source_kind": "cache"} try: latest = fetch(s["source_url"], s["version_field"]) _write_cache(root, {"source_url": s["source_url"], "latest": latest, "checked_at": now}) return {**base, "latest": latest, "update_available": _newer(current, latest), "checked_at": now, "source_kind": "network"} except Exception as exc: last = cache.get("latest") if isinstance(cache, dict) else None return {**base, "latest": last, "update_available": _newer(current, last), "source_kind": "cache" if last else "error", "error": str(exc)[:200]} def fork_drift(root, fetch=fetch_latest): """Drift-monitor validator contract {status, findings, summary}. No-op unless the repo is a configured fork consumer. Fail-open.""" try: s = check(root, fetch=fetch) if not s.get("enabled") or not s.get("source") or s.get("current") is None: return {"status": "ok", "findings": [], "summary": "fork version check not configured (nothing to check)"} if s.get("update_available"): return {"status": "attention", "findings": [{"category": "fork_drift", "severity": "info", "summary": f"a new fork version is available: " f"{s['latest']} (you are on {s['current']})"}], "summary": f"fork {s['current']} -> {s['latest']} available"} return {"status": "ok", "findings": [], "summary": f"fork up to date (on {s['current']})"} except Exception as exc: # pragma: no cover - fail-open return {"status": "ok", "findings": [], "summary": f"fork version check skipped: {exc}"} def main(argv=None): parser = argparse.ArgumentParser(description=__doc__.splitlines()[0]) parser.add_argument("--repo-root", default=".") parser.add_argument("--drift", action="store_true", help="Report fork version drift for the drift monitor.") parser.add_argument("--format", choices=("json", "markdown"), default=None) parser.add_argument("--output", help="Write the result here instead of stdout.") args = parser.parse_args(argv) result = fork_drift(Path(args.repo_root)) text = (result["summary"] if args.format == "markdown" else json.dumps(result, indent=2)) if args.output: out = Path(args.output) out.parent.mkdir(parents=True, exist_ok=True) out.write_text(text, encoding="utf-8") else: print(text) return 0 if __name__ == "__main__": import sys sys.exit(main())