"""Machine-first traceability graph over PRD Plugin state (read-only). Walks every record in `.prd_plugin/state` (JSON state files + nested artifact records) and the links between them, and builds a directed, labeled graph that agents can traverse and reason over: - **nodes** — one per canonical ID (`id`, `type`, `status`, `summary`, source file) - **edges** — `{cause, effect, rel}`: an upstream `cause` leads to a downstream `effect`, labeled with the relation (`derived_from`, `implements`, `implemented_by`, `evidenced_by`, ...). This is the shape AI-Collab-v3's causal graph ingests (`{cause, effect, label}`), so the `--format cause-effect` export drops straight into its substrate. - **adjacency** — `in` (causes) / `out` (effects) per node for O(1) traversal - **findings** — orphans, dangling links, and incomplete chains (a requirement with no implementing task, a task with no validation/evidence) Agent-facing queries: prd_graph.py --node IMP-TASK-001 --provenance # why does this exist (ancestors) prd_graph.py --node PRD-REQ-001 --impact # blast radius (descendants) prd_graph.py --node PRD-REQ-001 --neighbors # direct causes/effects prd_graph.py --gaps # incomplete chains + orphans/dangling prd_graph.py --format cause-effect # AI-Collab edge list prd_graph.py --out .prd_plugin/local/traceability-graph.json # write the graph It never mutates project state. `--set-auto on|off` only flips the `automation.graph_auto_refresh` config flag. """ import argparse import json import re from pathlib import Path SCHEMA_VERSION = "0.1" ID_RE = re.compile(r"\b([A-Z]{2,}(?:-[A-Z]{1,})*-\d{1,})\b") STATE_FILES = ( ".prd_plugin/state/requests.json", ".prd_plugin/state/tracking.json", ".prd_plugin/state/decisions.json", ".prd_plugin/state/health.json", ".prd_plugin/state/changelog.json", ".prd_plugin/state/evidence.json", ".prd_plugin/state/memory.json", ) TRACKING_BRANCH_DIR = ".prd_plugin/state/tracking-branches" DOCUMENT_BRANCH_DIR = "docs/doc-branches" # Node type by ID prefix (longest prefix wins). PREFIX_TYPE = { "BR-REQ": "brainstorm_requirement", "BR-DEC": "brainstorm_decision", "BR-Q": "brainstorm_question", "BR-RISK": "brainstorm_risk", "PRD-REQ": "requirement", "PRD-NFR": "requirement_nfr", "PRD-ACC": "acceptance", "ARCH-COMP": "architecture_component", "ARCH-DEC": "architecture_decision", "ARCH-IF": "architecture_interface", "ARCH-RISK": "architecture_risk", "IMP-PHASE": "phase", "IMP-TASK": "task", "IMP-VAL": "validation", "DEC": "decision", "EV": "evidence", "AGENT": "agent", "SES": "session", "MEM": "memory", "OBS": "observation", "TRK": "tracking", "CHG": "changelog", "REQ": "request", "MSG": "message", "HLT": "health", "BLK": "blocker", "DBR-DELTA": "document_branch_delta", "DBR-MERGE": "document_branch_merge", "DBR": "document_branch", } # field -> (relation label, owner role). owner role "effect" means the linked # target is upstream (the cause); "cause" means the target is downstream. Covers # both the canonical planning-artifact schema (source, source_prd_id, satisfies, # sources, validation, tasks, acceptance_criteria, ...) and the ai-collab-style # *_ids schema, so the graph works whichever a repo uses. FIELD_REL = { # --- upstream pointers (owner is the downstream effect) --- "source": ("derived_from", "effect"), "sources": ("derived_from", "effect"), "source_ids": ("derived_from", "effect"), "source_refs": ("derived_from", "effect"), "source_request_id": ("derived_from", "effect"), "source_brainstorm_id": ("derived_from", "effect"), "source_prd_id": ("derived_from", "effect"), "source_architecture_id": ("derived_from", "effect"), "requirement_ids": ("implements", "effect"), "architecture_ids": ("implements", "effect"), "satisfies": ("satisfies", "effect"), # canonical implementation-plan template: a task names its phase; a # validation names the tasks it applies to (both point at an upstream cause) "phase_id": ("in_phase", "effect"), "applies_to": ("validates", "effect"), "supports": ("supports", "effect"), "supported_ids": ("supports", "effect"), "superseded_by": ("superseded_by", "effect"), # --- downstream pointers (owner is the upstream cause) --- "validation": ("validated_by", "cause"), "validations": ("validated_by", "cause"), "validation_ids": ("validated_by", "cause"), # the shipped template's task field "tasks": ("contains", "cause"), "implementation_task_ids": ("implemented_by", "cause"), "acceptance_criteria": ("has_acceptance", "cause"), "acceptance_ids": ("has_acceptance", "cause"), "graduated_to": ("graduated_to", "cause"), "linked_ids": ("linked", "cause"), "affected_ids": ("affects", "cause"), "supersedes": ("supersedes", "cause"), "blocks": ("blocks", "cause"), "routed_to": ("routed_to", "cause"), "linked_health_findings": ("health", "cause"), "tracking_branch_ids": ("promoted_from", "effect"), "tracking_id": ("promoted_to", "cause"), } # Node-type groups for gap detection (by ID prefix). ARCH_OR_IMPL_PREFIXES = ("ARCH-", "IMP-") def _read_json(path): return json.loads(Path(path).read_text(encoding="utf-8-sig")) def _ids_in(value): out = set() if isinstance(value, str): out.update(ID_RE.findall(value)) elif isinstance(value, dict): for v in value.values(): out |= _ids_in(v) elif isinstance(value, list): for v in value: out |= _ids_in(v) return out def _node_type(node_id): parts = node_id.split("-") for n in (3, 2, 1): if len(parts) > n: cand = "-".join(parts[:n]) if cand in PREFIX_TYPE: return PREFIX_TYPE[cand] # bare prefix (no numeric tail consumed) e.g. "EV-001" -> "EV" pre = node_id.rsplit("-", 1)[0] return PREFIX_TYPE.get(pre, pre.lower()) def _walk_records(obj, source, out, parent=None): """Yield (id, dict, source_file, parent_id) for every dict with a canonical id. `parent` is the nearest enclosing record id, so a container document (e.g. PRD-001) is linked to its nested records (PRD-REQ-001, ...). """ if isinstance(obj, dict): rid = obj.get("id") if isinstance(rid, str) and ID_RE.fullmatch(rid): out.append((rid, obj, source, parent)) parent = rid for v in obj.values(): _walk_records(v, source, out, parent) elif isinstance(obj, list): for v in obj: _walk_records(v, source, out, parent) def _collect_records(root): root = Path(root).resolve() found, unreadable = [], [] for rel in STATE_FILES: p = root / rel if p.is_file(): try: _walk_records(_read_json(p), rel, found) except Exception as e: unreadable.append({"path": rel, "error": str(e)}) art = root / ".prd_plugin" / "state" / "artifacts" if art.is_dir(): for p in sorted(art.rglob("*.json")): rel = str(p.relative_to(root)).replace("\\", "/") try: _walk_records(_read_json(p), rel, found) except Exception as e: unreadable.append({"path": rel, "error": str(e)}) branches = root / TRACKING_BRANCH_DIR if branches.is_dir(): for p in sorted(branches.glob("*.json")): rel = str(p.relative_to(root)).replace("\\", "/") try: _walk_records(_read_json(p), rel, found) except Exception as e: unreadable.append({"path": rel, "error": str(e)}) document_branches = root / DOCUMENT_BRANCH_DIR if document_branches.is_dir(): for p in sorted(document_branches.glob("DBR-*/BRANCH.json")): rel = str(p.relative_to(root)).replace("\\", "/") try: _walk_records(_read_json(p), rel, found) except Exception as e: unreadable.append({"path": rel, "error": str(e)}) return found, unreadable # Code anchors (TRK-125): agents stamp canonical IDs at implementation sites # (comments/docstrings); the graph indexes those stamps so features trace from # record to code and back. Only IDs that exist as records become anchors. CODE_EXTENSIONS = {".py", ".cjs", ".mjs", ".js", ".ts", ".tsx", ".jsx", ".rs", ".go", ".java", ".cs", ".rb", ".sh", ".ps1", ".c", ".cpp", ".h", ".sql", ".toml", ".yaml", ".yml"} _ANCHOR_SKIP_DIRS = {".git", "node_modules", "wiki", "raw", "templates", ".prd_plugin", ".agents", ".opencode", "__pycache__", "dist", "build", "target", "vendor"} _ANCHOR_MAX_BYTES = 512 * 1024 def _scan_code_anchors(root, known_ids): """{relative_path: {record_id: [line, ...]}} for stamped known IDs.""" root = Path(root) anchors = {} stack = [root] while stack: directory = stack.pop() try: entries = sorted(directory.iterdir()) except OSError: continue for entry in entries: name = entry.name if entry.is_dir(): if name in _ANCHOR_SKIP_DIRS: continue # hub mirrors: .claude/skills duplicates skills/; hooks are code if directory.name == ".claude" and name == "skills": continue stack.append(entry) continue if entry.suffix.lower() not in CODE_EXTENSIONS: continue try: if entry.stat().st_size > _ANCHOR_MAX_BYTES: continue text = entry.read_text(encoding="utf-8-sig") except (OSError, UnicodeError): continue per_id = {} for lineno, line in enumerate(text.splitlines(), start=1): for match in ID_RE.findall(line): if match in known_ids: per_id.setdefault(match, []).append(lineno) if per_id: anchors[entry.relative_to(root).as_posix()] = per_id return anchors def build_graph(root="."): records, unreadable = _collect_records(root) nodes = {} edge_set = set() for rid, obj, source, parent in records: node = nodes.setdefault(rid, {"id": rid, "type": _node_type(rid), "status": None, "summary": None, "source": source}) if node["summary"] is None and isinstance(obj.get("summary"), str): node["summary"] = obj["summary"] if node["status"] is None and isinstance(obj.get("status"), str): node["status"] = obj["status"] # structural containment: document -> nested record if parent and parent != rid: edge_set.add((parent, rid, "contains")) for field, (rel, owner_role) in FIELD_REL.items(): for target in _ids_in(obj.get(field)): if target == rid: continue if owner_role == "effect": edge_set.add((target, rid, rel)) # target(cause) -> owner(effect) else: edge_set.add((rid, target, rel)) # owner(cause) -> target(effect) for path, per_id in sorted(_scan_code_anchors(root, set(nodes)).items()): code_id = f"code:{path}" nodes[code_id] = {"id": code_id, "type": "code", "status": None, "summary": f"code anchors: {', '.join(sorted(per_id))}", "source": path, "anchors": per_id} for record_id in per_id: edge_set.add((record_id, code_id, "anchored_in")) edges = [{"cause": c, "effect": e, "rel": r} for (c, e, r) in sorted(edge_set)] adjacency = {nid: {"in": [], "out": []} for nid in nodes} for e in edges: adjacency.setdefault(e["cause"], {"in": [], "out": []})["out"].append(e["effect"]) adjacency.setdefault(e["effect"], {"in": [], "out": []})["in"].append(e["cause"]) for a in adjacency.values(): a["in"] = sorted(set(a["in"])) a["out"] = sorted(set(a["out"])) findings = _findings(nodes, edges, adjacency) findings["unreadable"] = unreadable return { "schema_version": SCHEMA_VERSION, "repo_root": str(Path(root).resolve()), "counts": {"nodes": len(nodes), "edges": len(edges), "orphans": len(findings["orphans"]), "dangling": len(findings["dangling"]), "incomplete_chains": len(findings["incomplete_chains"]), "unreadable": len(unreadable)}, "nodes": nodes, "edges": edges, "adjacency": adjacency, "findings": findings, } def _reach(adjacency, start, direction): """Transitive closure following 'out' (impact) or 'in' (provenance).""" seen = set() frontier = list(adjacency.get(start, {}).get(direction, [])) while frontier: nid = frontier.pop() if nid in seen: continue seen.add(nid) frontier.extend(adjacency.get(nid, {}).get(direction, [])) seen.discard(start) return sorted(seen) def impact(graph, node): """Everything downstream of node (blast radius).""" return _reach(graph["adjacency"], node, "out") def provenance(graph, node): """Everything upstream of node (why it exists / ancestors).""" return _reach(graph["adjacency"], node, "in") def neighbors(graph, node): a = graph["adjacency"].get(node, {"in": [], "out": []}) return {"causes": a["in"], "effects": a["out"]} def _findings(nodes, edges, adjacency): node_ids = set(nodes) referenced = {} for e in edges: referenced.setdefault(e["cause"], set()).add(e["effect"]) referenced.setdefault(e["effect"], set()).add(e["cause"]) dangling = sorted( ({"id": tid, "linked_with": sorted(srcs)} for tid, srcs in referenced.items() if tid not in node_ids), key=lambda d: d["id"], ) orphans = sorted(nid for nid in node_ids if not adjacency.get(nid, {}).get("in") and not adjacency.get(nid, {}).get("out")) incomplete = [] for nid, node in nodes.items(): # orphans are reported on their own; don't double-flag them as gaps if nid in orphans: continue down = _reach(adjacency, nid, "out") if node["type"] == "requirement": # covered if any architecture or implementation record sits downstream if not any(d.startswith(ARCH_OR_IMPL_PREFIXES) for d in down): incomplete.append({"id": nid, "gap": "requirement_without_coverage", "detail": "no ARCH-* or IMP-* record reachable downstream"}) elif node["type"] == "task": if not any(d.startswith("IMP-VAL") or _node_type(d) == "evidence" for d in down): incomplete.append({"id": nid, "gap": "task_without_validation", "detail": "no IMP-VAL or EV reachable downstream"}) incomplete.sort(key=lambda f: f["id"]) return {"orphans": orphans, "dangling": dangling, "incomplete_chains": incomplete} def to_cause_effect(graph): """AI-Collab-v3 causal-edge shape: [{cause, effect, label}].""" return [{"cause": e["cause"], "effect": e["effect"], "label": e["rel"]} for e in graph["edges"]] def to_mermaid(graph): lines = ["flowchart LR"] for e in graph["edges"]: lines.append(f' {e["cause"].replace("-", "_")}["{e["cause"]}"] -->|{e["rel"]}| ' f'{e["effect"].replace("-", "_")}["{e["effect"]}"]') return "\n".join(lines) + "\n" def _config_path(root): return Path(root) / ".prd_plugin" / "config.json" def set_auto_refresh(root, value): p = _config_path(root) if p.is_file(): try: cfg = _read_json(p) except Exception as e: raise SystemExit(f"refusing to write: {p} is not valid JSON ({e})") if not isinstance(cfg, dict): raise SystemExit(f"refusing to write: {p} is not a JSON object") else: cfg = {} cfg.setdefault("automation", {})["graph_auto_refresh"] = bool(value) p.parent.mkdir(parents=True, exist_ok=True) tmp = p.with_suffix(p.suffix + ".tmp") tmp.write_text(json.dumps(cfg, indent=2, ensure_ascii=False) + "\n", encoding="utf-8") import os os.replace(tmp, p) return bool(value) def auto_refresh_enabled(root): p = _config_path(root) if not p.is_file(): return False try: return bool(_read_json(p).get("automation", {}).get("graph_auto_refresh", False)) except Exception: return False def _summary(graph): c = graph["counts"] return (f"Graph: {c['nodes']} nodes, {c['edges']} edges | " f"orphans: {c['orphans']} | dangling: {c['dangling']} | " f"incomplete chains: {c['incomplete_chains']}") def main(argv=None): ap = argparse.ArgumentParser(description="Machine-first traceability graph (read-only).") ap.add_argument("--repo-root", default=".") ap.add_argument("--node") ap.add_argument("--impact", action="store_true", help="downstream blast radius of --node") ap.add_argument("--provenance", action="store_true", help="upstream ancestors of --node") ap.add_argument("--neighbors", action="store_true", help="direct causes/effects of --node") ap.add_argument("--gaps", action="store_true", help="incomplete chains, orphans, dangling") ap.add_argument("--anchors", metavar="ID", help="code files and lines stamped with this record id") ap.add_argument("--format", choices=("json", "cause-effect", "mermaid"), default="json") ap.add_argument("--out", help="write the full graph JSON to this path") ap.add_argument("--set-auto", choices=("on", "off", "status"), help="flip/show automation.graph_auto_refresh") args = ap.parse_args(argv) if args.set_auto: if args.set_auto == "status": print(json.dumps({"graph_auto_refresh": auto_refresh_enabled(args.repo_root)})) else: val = set_auto_refresh(args.repo_root, args.set_auto == "on") print(json.dumps({"graph_auto_refresh": val})) print(f"Graph auto-refresh {'on' if val else 'off'}. " "Takes effect in the next Claude Code session (it runs in the Stop hook).") return 0 graph = build_graph(args.repo_root) if args.out: Path(args.out).parent.mkdir(parents=True, exist_ok=True) Path(args.out).write_text(json.dumps(graph, indent=2, ensure_ascii=False) + "\n", encoding="utf-8") if args.anchors: sites = {node["source"]: node["anchors"][args.anchors] for node in graph["nodes"].values() if node.get("type") == "code" and args.anchors in node.get("anchors", {})} print(json.dumps({"id": args.anchors, "anchored_in": sites}, indent=2)) elif args.node and args.impact: print(json.dumps({"node": args.node, "impact": impact(graph, args.node)}, indent=2)) elif args.node and args.provenance: print(json.dumps({"node": args.node, "provenance": provenance(graph, args.node)}, indent=2)) elif args.node and args.neighbors: print(json.dumps({"node": args.node, **neighbors(graph, args.node)}, indent=2)) elif args.gaps: print(json.dumps(graph["findings"], indent=2)) elif args.format == "cause-effect": print(json.dumps(to_cause_effect(graph), indent=2)) elif args.format == "mermaid": print(to_mermaid(graph)) elif args.out: print(_summary(graph)) else: print(json.dumps(graph, indent=2, ensure_ascii=False)) return 0 if __name__ == "__main__": raise SystemExit(main())