#!/usr/bin/env python3 """Produce a deterministic structural summary of a Brick schema-v4 graph.""" from __future__ import annotations import argparse import json import math import sys from collections import Counter, defaultdict from pathlib import Path from typing import Any, NoReturn DEPENDENCY_KINDS = {"calls", "imports", "inherits"} NODE_KINDS = {"repository", "directory", "file", "class", "function", "external", "finding"} EDGE_KINDS = {"contains", "defines", "calls", "imports", "inherits", "flags"} EVIDENCE_LANES = {"syntactic", "symbolic", "imported", "heuristic"} P4_METRIC_FIELDS = { "parser_state_count", "terminal_state_count", "parser_transition_count", "parser_cycle_count", "cyclic_parser_state_count", "unreachable_state_count", "control_apply_node_count", "control_apply_relationship_count", "control_apply_cycle_count", "cyclic_control_apply_node_count", "unresolved_dependency_count", } def fail(message: str) -> NoReturn: print(f"architecture_summary: {message}", file=sys.stderr) raise SystemExit(2) def reject_nonfinite_constant(constant: str) -> NoReturn: raise ValueError(f"non-finite JSON number {constant}") def is_normalized_absolute_root(value: str) -> bool: if not value or "\\" in value: return False if value.startswith("//"): remainder = value[2:] components = remainder.split("/") if ( len(components) < 2 or not components[0] or not components[1] or any(character in components[0] + components[1] for character in "?:") ): return False elif value.startswith("/"): remainder = value[1:] elif ( len(value) >= 3 and value[0].isascii() and value[0].isalpha() and value[1:3] == ":/" ): remainder = value[3:] else: return False return all( component not in {".", ".."} and (component != "" or remainder == "") for component in remainder.split("/") ) def is_normalized_relative_path(value: str, allow_dot: bool = False) -> bool: if allow_dot and value == ".": return True if not value or value.startswith("/") or "\\" in value: return False if len(value) >= 2 and value[0].isascii() and value[0].isalpha() and value[1] == ":": return False return all(component not in {"", ".", ".."} for component in value.split("/")) def is_valid_confidence(value: Any) -> bool: if value is None or isinstance(value, str): return True if isinstance(value, bool) or not isinstance(value, (int, float)): return False try: return math.isfinite(float(value)) except OverflowError: return False def validate_graph_records(value: dict[str, Any]) -> None: nodes = value["nodes"] edges = value["edges"] node_ids: set[str] = set() repository_count = 0 for node in nodes: if not isinstance(node, dict): fail("every node must be an object") node_id = node.get("id") kind = node.get("kind") if not isinstance(node_id, str) or not node_id: fail("every node must have a non-empty string id") if node_id in node_ids: fail(f"duplicate node ID {node_id}") node_ids.add(node_id) if kind not in NODE_KINDS: fail(f"unknown node kind {kind!r}") if kind == "repository": repository_count += 1 if repository_count != 1: fail(f"graph must contain exactly one repository node; found {repository_count}") edge_ids: set[str] = set() for edge in edges: if not isinstance(edge, dict): fail("every edge must be an object") edge_id = edge.get("id") source = edge.get("source") target = edge.get("target") kind = edge.get("kind") if not isinstance(edge_id, str) or not edge_id: fail("every edge must have a non-empty string id") if edge_id in edge_ids: fail(f"duplicate edge ID {edge_id}") if edge_id in node_ids: fail(f"entity ID {edge_id} is shared by a node and edge") edge_ids.add(edge_id) if kind not in EDGE_KINDS: fail(f"unknown edge kind {kind!r}") if source not in node_ids or target not in node_ids: fail(f"edge {edge_id} has an unknown endpoint") evidence_records = edge.get("evidence") if not isinstance(evidence_records, list) or not evidence_records: fail(f"edge {edge_id} must contain non-empty evidence") for evidence in evidence_records: if not isinstance(evidence, dict): fail(f"edge {edge_id} evidence must be an object") if evidence.get("lane") not in EVIDENCE_LANES: fail(f"edge {edge_id} has an unknown evidence lane") if not isinstance(evidence.get("backend"), str): fail(f"edge {edge_id} evidence backend must be a string") if not isinstance(evidence.get("provenance"), str): fail(f"edge {edge_id} evidence provenance must be a string") confidence = evidence.get("confidence") if not is_valid_confidence(confidence): fail(f"edge {edge_id} evidence confidence has an invalid type") def load_graph(path: Path) -> dict[str, Any]: try: value = json.loads( path.read_text(encoding="utf-8"), parse_constant=reject_nonfinite_constant, ) except OSError as error: fail(f"cannot read {path}: {error}") except (json.JSONDecodeError, ValueError) as error: fail(f"invalid JSON in {path}: {error}") if not isinstance(value, dict): fail("top-level JSON value must be an object") if value.get("schema_version") != 4: fail(f"unsupported schema_version {value.get('schema_version')!r}; expected 4") if not isinstance(value.get("generated_by"), str) or not isinstance(value.get("root"), str): fail("schema-v4 graph must contain string generated_by and root fields") if not is_normalized_absolute_root(value["root"]): fail("root must be a normalized portable absolute path") scope = value.get("scope") if not isinstance(scope, dict) or not isinstance(scope.get("path"), str): fail("schema-v4 graph must contain scan scope.path") if not is_normalized_relative_path(scope["path"], allow_dot=True): fail("scope.path must be normalized and repository-relative") if scope.get("p4_frontend") not in {"lexical", "tree-sitter-p4-pilot"}: fail(f"unsupported P4 frontend in scan scope: {scope.get('p4_frontend')!r}") excluded = scope.get("excluded_directories", []) if not isinstance(excluded, list) or not all( isinstance(directory, str) for directory in excluded ): fail("scope.excluded_directories must be an array of strings") if excluded != sorted(set(excluded)) or any( not is_normalized_relative_path(directory) or "/" in directory for directory in excluded ): fail("scope.excluded_directories must contain sorted unique path components") if not isinstance(value.get("nodes"), list) or not isinstance( value.get("edges"), list ): fail("graph must contain nodes and edges arrays") metrics = value.get("metrics") p4_metrics = metrics.get("p4") if isinstance(metrics, dict) else None if not isinstance(p4_metrics, dict) or set(p4_metrics) != P4_METRIC_FIELDS: fail("schema-v4 graph must contain the complete metrics.p4 contract") if any(type(metric) is not int for metric in p4_metrics.values()): fail("metrics.p4 values must be integers") validate_graph_records(value) return value def text_field(value: Any, field: str) -> str | None: result = value.get(field) if isinstance(value, dict) else None return result if isinstance(result, str) else None def evidence_value(value: Any, field: str) -> str: result = value.get(field) if isinstance(value, dict) else None if result is None: return "unknown" if isinstance(result, float): return f"{result:.3f}".rstrip("0").rstrip(".") return str(result) def strongly_connected_components( vertices: list[str], adjacency: dict[str, list[str]] ) -> list[list[str]]: index = 0 indices: dict[str, int] = {} lowlinks: dict[str, int] = {} stack: list[str] = [] on_stack: set[str] = set() components: list[list[str]] = [] def visit(vertex: str) -> None: nonlocal index indices[vertex] = index lowlinks[vertex] = index index += 1 stack.append(vertex) on_stack.add(vertex) for neighbor in adjacency.get(vertex, []): if neighbor not in indices: visit(neighbor) lowlinks[vertex] = min(lowlinks[vertex], lowlinks[neighbor]) elif neighbor in on_stack: lowlinks[vertex] = min(lowlinks[vertex], indices[neighbor]) if lowlinks[vertex] == indices[vertex]: component: list[str] = [] while True: member = stack.pop() on_stack.remove(member) component.append(member) if member == vertex: break components.append(sorted(component)) for vertex in sorted(vertices): if vertex not in indices: visit(vertex) return sorted(components, key=lambda item: (-len(item), item)) def ranked_counts( counts: Counter[str], labels: dict[str, str], limit: int ) -> list[dict[str, Any]]: rows = [ {"id": node_id, "label": labels.get(node_id, node_id), "count": count} for node_id, count in counts.items() ] return sorted(rows, key=lambda row: (-row["count"], row["id"]))[:limit] def node_reference(node_id: str, nodes: dict[str, dict[str, Any]]) -> dict[str, Any]: node = nodes.get(node_id, {}) result: dict[str, Any] = { "id": node_id, "name": text_field(node, "name") or node_id, "kind": text_field(node, "kind") or "unknown", } path = text_field(node, "path") if path is not None: result["path"] = path line = node.get("line") if isinstance(line, int): result["line"] = line return result def summarize(graph: dict[str, Any], limit: int) -> dict[str, Any]: nodes = [node for node in graph["nodes"] if isinstance(node, dict)] edges = [edge for edge in graph["edges"] if isinstance(edge, dict)] node_by_id = { node_id: node for node in nodes if (node_id := text_field(node, "id")) is not None } labels = { node_id: text_field(node, "name") or node_id for node_id, node in node_by_id.items() } node_kinds = Counter(text_field(node, "kind") or "unknown" for node in nodes) edge_kinds = Counter(text_field(edge, "kind") or "unknown" for edge in edges) evidence_lanes: Counter[str] = Counter() evidence_backends: Counter[str] = Counter() evidence_confidence: Counter[str] = Counter() adjacency: dict[str, list[str]] = defaultdict(list) fan_in: Counter[str] = Counter() fan_out: Counter[str] = Counter() self_loops: set[str] = set() dependency_edges = 0 unresolved_targets: Counter[str] = Counter() for edge in edges: evidence_records = edge.get("evidence") if isinstance(evidence_records, list): for evidence in evidence_records: if isinstance(evidence, dict): evidence_lanes[evidence_value(evidence, "lane")] += 1 evidence_backends[evidence_value(evidence, "backend")] += 1 evidence_confidence[evidence_value(evidence, "confidence")] += 1 kind = text_field(edge, "kind") source = text_field(edge, "source") target = text_field(edge, "target") if kind not in DEPENDENCY_KINDS or source is None or target is None: continue dependency_edges += 1 adjacency[source].append(target) fan_out[source] += 1 fan_in[target] += 1 if source == target: self_loops.add(source) target_node = node_by_id.get(target) if target_node is not None and text_field(target_node, "kind") == "external": unresolved_targets[target] += 1 vertices = sorted( set(adjacency) | {neighbor for neighbors in adjacency.values() for neighbor in neighbors} ) components = strongly_connected_components(vertices, adjacency) cyclic_components = [ component for component in components if len(component) > 1 or component[0] in self_loops ] findings = [node for node in nodes if text_field(node, "kind") == "finding"] finding_backends: Counter[str] = Counter() finding_confidence: Counter[str] = Counter() finding_rules: Counter[str] = Counter() for finding in findings: metadata = finding.get("metadata") or finding.get("attributes") provenance = finding.get("provenance") if isinstance(metadata, dict): finding_rules[ text_field(metadata, "rule_id") or text_field(metadata, "rule") or "unknown" ] += 1 finding_confidence[evidence_value(metadata, "confidence")] += 1 finding_backends[evidence_value(metadata, "backend")] += 1 elif isinstance(provenance, dict): finding_backends[text_field(provenance, "backend") or "unknown"] += 1 unresolved = [ { "id": node_id, "label": labels.get(node_id, node_id), "incoming_dependencies": count, } for node_id, count in unresolved_targets.items() ] unresolved.sort(key=lambda row: (-row["incoming_dependencies"], row["id"])) return { "schema_version": graph.get("schema_version"), "generated_by": graph.get("generated_by"), "root": graph.get("root"), "scope": graph.get("scope"), "diagnostics": graph.get("diagnostics", []), "inventory": { "nodes": len(nodes), "edges": len(edges), "node_kinds": dict(sorted(node_kinds.items())), "edge_kinds": dict(sorted(edge_kinds.items())), }, "evidence": { "lanes": dict(sorted(evidence_lanes.items())), "backends": dict(sorted(evidence_backends.items())), "confidence": dict(sorted(evidence_confidence.items())), }, "dependencies": { "edge_count": dependency_edges, "cyclic_component_count": len(cyclic_components), "cyclic_components": [ [node_reference(node_id, node_by_id) for node_id in component] for component in cyclic_components[:limit] ], "top_fan_in": ranked_counts(fan_in, labels, limit), "top_fan_out": ranked_counts(fan_out, labels, limit), "unresolved_target_count": len(unresolved_targets), "top_unresolved_targets": unresolved[:limit], }, "findings": { "count": len(findings), "backends": dict(sorted(finding_backends.items())), "confidence": dict(sorted(finding_confidence.items())), "rules": dict(sorted(finding_rules.items())), }, "metrics": graph.get("metrics"), } def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("graph", type=Path, help="Brick schema-v4 JSON artifact") parser.add_argument( "--limit", type=int, default=10, help="maximum ranked entries (default: 10)" ) args = parser.parse_args() if args.limit <= 0: parser.error("--limit must be positive") return args def main() -> None: args = parse_args() graph = load_graph(args.graph) json.dump(summarize(graph, args.limit), sys.stdout, indent=2, sort_keys=True) sys.stdout.write("\n") if __name__ == "__main__": main()