#!/usr/bin/env python3 """Build a deterministic dependency graph from normalized coverage records.""" from __future__ import annotations import argparse import hashlib import json from collections import Counter, defaultdict from pathlib import Path NORMALIZED = Path("coverage/normalized-sources.jsonl") def sha256(path: Path) -> str: digest = hashlib.sha256() with path.open("rb") as handle: for block in iter(lambda: handle.read(1024 * 1024), b""): digest.update(block) return digest.hexdigest() def read_jsonl(path: Path) -> list[dict[str, object]]: rows: list[dict[str, object]] = [] with path.open("r", encoding="utf-8") as handle: for line_number, line in enumerate(handle, 1): if not line.strip(): continue row = json.loads(line) if not isinstance(row, dict): raise ValueError(f"{path}:{line_number}: expected object") rows.append(row) return rows def canonical_path(value: object) -> str | None: if not isinstance(value, str) or not value.strip(): return None raw = value.replace("\\", "/").strip() if raw.startswith("/"): return None parts: list[str] = [] for part in raw.split("/"): if part in ("", "."): continue if part == "..": if not parts: return None parts.pop() else: parts.append(part) return "/".join(parts) or None def generate(root: Path, output: Path, report_path: Path) -> tuple[int, str]: normalized_path = root / NORMALIZED if not normalized_path.is_file(): raise FileNotFoundError(f"required normalized index is missing: {NORMALIZED}") records = read_jsonl(normalized_path) file_ids = { str(row["path"]): str(row["source_id"]) for row in records if row.get("origin_index") == "source" and row.get("path") } nodes = {str(row["source_id"]) for row in records} edges: list[dict[str, object]] = [] edge_counts: Counter[str] = Counter() duplicate_edges: list[dict[str, object]] = [] unresolved: list[dict[str, object]] = [] def target_for(path_value: object, status: str = "unresolved") -> tuple[str, str]: path = canonical_path(path_value) if path and path in file_ids: return file_ids[path], "resolved" if path and (root / path).is_file(): return f"unindexed:{path}", "unindexed" return f"unresolved:{path or str(path_value) or 'empty'}", status def add_edge(from_id: str, to_id: str, edge_kind: str, record: dict[str, object], status: str, raw_value: object = None) -> None: key = f"{from_id}|{edge_kind}|{to_id}|{record.get('line', '')}" edge_counts[key] += 1 if edge_counts[key] > 1: duplicate_edges.append({"key": key, "occurrence": edge_counts[key], "source_id": record.get("source_id")}) edge_id = f"edge:{key}" + (f"~dup{edge_counts[key]}" if edge_counts[key] > 1 else "") row = { "edge_id": edge_id, "from_source_id": from_id, "to_source_id": to_id, "edge_kind": edge_kind, "raw_value": raw_value, "evidence_path": record.get("hash_path") or record.get("path"), "evidence_line": record.get("line"), "evidence_source_id": record.get("source_id"), "parse_status": status, } edges.append(row) nodes.add(to_id) if status not in {"resolved", "parsed"}: unresolved.append({"edge_id": edge_id, "to_source_id": to_id, "status": status, "raw_value": raw_value}) for record in records: source_id = str(record["source_id"]) index_name = str(record.get("origin_index")) if index_name == "resource": origin_path = canonical_path(record.get("hash_path")) from_id = file_ids.get(origin_path, source_id) target_id, status = target_for(record.get("path"), str(record.get("target_status") or "unresolved")) add_edge(from_id, target_id, f"resource:{record.get('edge_kind') or record.get('resource_kind') or 'reference'}", record, status, record.get("raw_value")) elif index_name == "test": entry_path = canonical_path(record.get("path")) entry_file = file_ids.get(entry_path) if entry_file: add_edge(entry_file, source_id, "test_entrypoint", record, "resolved", record.get("command")) for field, edge_kind in (("inputs", "test_input"), ("expected", "test_expected"), ("config_refs", "test_config_ref")): values = record.get(field, []) if not isinstance(values, list): continue for value in sorted(str(item) for item in values): target_id, status = target_for(value) add_edge(source_id, target_id, edge_kind, record, status, value) elif index_name in {"symbol", "document"}: related = record.get("related_source_ids", []) if isinstance(related, list): for related_id in sorted(str(item) for item in related): add_edge(related_id, source_id, "declares" if index_name == "symbol" else "documents", record, "resolved", record.get("path")) edges.sort(key=lambda row: str(row["edge_id"])) output.parent.mkdir(parents=True, exist_ok=True) with output.open("w", encoding="utf-8", newline="\n") as handle: for row in edges: handle.write(json.dumps(row, ensure_ascii=False, sort_keys=True, separators=(",", ":")) + "\n") digest = sha256(output) incoming = Counter(str(edge["to_source_id"]) for edge in edges) outgoing = Counter(str(edge["from_source_id"]) for edge in edges) orphan_nodes = sorted(node for node in nodes if not incoming[node] and not outgoing[node] and not node.startswith(("unresolved:", "unindexed:"))) report = { "schema": "dependency-closure-report.v1", "normalized_input_sha256": sha256(normalized_path), "nodes": len(nodes), "edges": len(edges), "resolved_edges": sum(1 for edge in edges if edge["parse_status"] == "resolved"), "unresolved_edges": len(unresolved), "edge_kinds": dict(sorted(Counter(str(edge["edge_kind"]) for edge in edges).items())), "duplicate_edges": duplicate_edges, "unresolved": unresolved, "orphan_nodes": orphan_nodes, "output_sha256": digest, } report_path.parent.mkdir(parents=True, exist_ok=True) report_path.write_text(json.dumps(report, ensure_ascii=False, sort_keys=True, indent=2) + "\n", encoding="utf-8") return len(edges), digest def main() -> int: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--repo-root", type=Path, default=Path(__file__).resolve().parents[1]) parser.add_argument("--output", type=Path, default=Path("coverage/dependency-closure.jsonl")) parser.add_argument("--report", type=Path, default=Path("coverage/dependency-closure.report.json")) args = parser.parse_args() root = args.repo_root.resolve() count, digest = generate(root, root / args.output, root / args.report) print(f"dependency_edges={count} sha256={digest}") return 0 if __name__ == "__main__": raise SystemExit(main())