
Dev Context Code Graph
- 35 installs
- 73 repo stars
- Updated July 13, 2026
- vasilyu1983/ai-agents-public
Helps with ai & agent building tasks.
About
dev-context-code-graph is a Claude Code skill for ai & agent building. It helps solo builders move faster with AI-assisted development.
- dev-context-code-graph
- AI & Agent Building
- AI-coding skill
Dev Context Code Graph by the numbers
- 35 all-time installs (skills.sh)
- +2 installs in the week ending Jul 27, 2026 (Skillselion tracking)
- Ranked #8,710 of 16,546 AI & Agent Building skills by installs in the Skillselion catalog
- Data as of Aug 5, 2026 (Skillselion catalog sync)
npx skills add https://github.com/vasilyu1983/ai-agents-public --skill dev-context-code-graphAdd your badge
Show developers this skill is listed on Skillselion. Paste this into your README.
| Installs | 35 |
|---|---|
| repo stars | ★ 73 |
| Last updated | July 13, 2026 |
| Repository | vasilyu1983/ai-agents-public ↗ |
What it does
Helps with ai & agent building tasks.
Files
Code Graph
Build a deterministic, machine-readable code graph for a single repository. Treat the graph as a machine-readable substrate for an LLM-maintained repo wiki or context hub, not just a terminal-only report. This skill mirrors the dev-context-multi-repo artifact workflow, but works at file and symbol level instead of portfolio level.
Use this skill when you need:
- a committable
graphs/code-graph.jsonartifact - file and symbol maps for one repo
- import, call, inheritance, and test-link analysis
- graph-theory review signals such as articulation points, bridges, cycles, topological order, and alternate paths
- blast radius and minimal review context for changed files or symbols
- budget-bounded context retrieval around 1–3 hot symbols via Personalized PageRank
- a grounded repo description or module description generated from graph data instead of ad hoc codebase prose
Do not use this skill for:
- portfolio-wide repo discovery or cross-repo system maps
- architecture or migration planning across many repos
- prose documentation cleanup without graph generation
Use related skills instead:
- dev-context-multi-repo for repo portfolios and hub-level knowledge graphs
- dev-context-engineering for deciding when code graph vs context graph vs repo graph is the right artifact
- docs-ai-prd for code-graph specs and acceptance criteria
- docs-codebase for publishing graph-backed docs and reports
Quick Reference
| Need | Start here |
|---|---|
| Generate the base artifact set | ## Workflow |
| Validate schema and graph integrity | ### Phase 3: Validate |
| Query blast radius or symbol neighborhoods | ### Phase 4: Query |
| Surface structural graph risk | query_code_graph.py --articulation-points, --bridges, --cycles, --topo-sort, --from ... --to ... --k N |
| Run hot-symbol PPR retrieval | query_code_graph.py --ppr --seed <id> [--seed <id> ...] --top N |
| Detect modules via communities | query_code_graph.py --communities [--resolution γ] [--community-seed N] |
| Pick the right query for a review | references/query-recipes.md |
| Load scripts, schemas, and reports | ## Navigation |
Standard Outputs
Every run should target the same output model:
code-profiles/<repo>.json
Machine-readable code profile conforming to schemas/code-profile.schema.json.
graphs/code-graph.json
Primary machine-readable artifact. Queryable symbol graph conforming to schemas/code-graph.schema.json.
reports/code-graph-validation.json(optional but recommended)
Output of scripts/validate_code_graph.py --output ....
reports/code-graph-report.md(optional but recommended)
Human-readable summary report generated from the graph.
reports/query-*.md(optional but recommended)
Persisted blast-radius, neighborhood, and relationship answers worth filing back into a repo knowledge base.
reports/code-graph-report.html(optional)
Static HTML report for local review.
reports/code-graph.mmd(optional)
Mermaid diagram export for graph neighborhoods or impact views.
Workflow
Phase 1: Scan
1. Identify supported source files by extension while skipping generated and vendored build trees. 2. Classify files as source, test, config, or unknown. 3. Parse each file with the best deterministic strategy available. 4. Record parse status explicitly: parsed, heuristic, unsupported, error, or skipped.
Primary helper: scripts/scan_code_repo.py
ASCII Flow
single-repo code graph request
-> scan files and classify source, test, config, or unknown
-> parse symbols with deterministic or heuristic parser
-> build nodes and edges into graphs/code-graph.json
-> validate schema, references, duplicates, orphans, and parse confidence
-> query neighborhoods, impact, paths, PPR, communities, or structural risk
-> persist useful reports back into repo context
-> use findings for review, docs, onboarding, or blast-radius planningPhase 2: Build
1. Read one or more code-profiles/*.json files. 2. Materialize canonical nodes and edges into graphs/code-graph.json. 3. Add structural edges from repo to file and from parent symbol or file to child symbol. 4. Synthesize external_symbol nodes for unresolved or third-party import/call targets.
Primary helper: scripts/build_code_graph.py
Phase 3: Validate
Run validation after every build:
python3 scripts/validate_code_graph.py graphs/code-graph.json \
--output reports/code-graph-validation.jsonThe validator checks:
- schema and enum compliance
- dangling references
- orphan nodes
- duplicate IDs
- containment / parent edge consistency
- circular imports
- stale verification metadata
- parse-status and confidence bounds
Phase 4: Query
Common query patterns:
# Neighborhood around a file or symbol
python3 scripts/query_code_graph.py graphs/code-graph.json --node <id> --hops 1
# Blast radius from a file or symbol
python3 scripts/query_code_graph.py graphs/code-graph.json --impact <id> --hops 2
# Path between two nodes
python3 scripts/query_code_graph.py graphs/code-graph.json --from <id> --to <id>
# Up to three node-disjoint shortest paths
python3 scripts/query_code_graph.py graphs/code-graph.json --from <id> --to <id> --k 3
# Personalized PageRank from one or more hot symbols
python3 scripts/query_code_graph.py graphs/code-graph.json --ppr --seed <id> --top 30
# Module detection via Louvain communities (γ=1 default; >1 = smaller modules, <1 = larger)
python3 scripts/query_code_graph.py graphs/code-graph.json --communities --format table
python3 scripts/query_code_graph.py graphs/code-graph.json --communities --resolution 1.4 --top 25
# Search by label, path, summary, or tags
python3 scripts/query_code_graph.py graphs/code-graph.json --search "invoice service"
# Structural risk
python3 scripts/query_code_graph.py graphs/code-graph.json --articulation-points --relations imports --top 20
python3 scripts/query_code_graph.py graphs/code-graph.json --bridges --relations imports --top 20
python3 scripts/query_code_graph.py graphs/code-graph.json --cycles --relations imports,inherits,calls
python3 scripts/query_code_graph.py graphs/code-graph.json --topo-sort imports
# Mermaid diagram export
python3 scripts/query_code_graph.py graphs/code-graph.json --diagram --output reports/code-graph.mmdPrimary helper: scripts/query_code_graph.py
When a query produces durable knowledge, write it to markdown or Mermaid instead of leaving it only in chat output.
Use query outputs to generate:
- repo-area descriptions
- module summaries
- change-impact notes
- review packets for risky symbols or subsystems
Phase 5: Publish
Generate static reports for humans:
python3 scripts/export_code_graph_report.py graphs/code-graph.json \
--output-dir reports/Prefer markdown, Mermaid, or HTML outputs that can be reviewed in Obsidian or filed back into a broader context hub.
Phase 6: Enhance
Run lightweight health checks over the graph-backed knowledge:
1. Review validation results for missing edges, parse gaps, and unsupported areas that need explanation. 2. Ask the LLM to suggest reusable reports, concept pages, or follow-up questions based on dense or weakly connected graph regions. 3. File durable findings back into reports/ or a higher-level repo knowledge base instead of repeating the same exploration later.
If the graph is feeding repo descriptions, keep the generation bounded:
- summarize only what the graph can support
- call out parse gaps explicitly
- do not invent ownership, runtime behavior, or architectural roles from symbol names alone
Supported V1 Scope
The v1 parser pipeline is deterministic and conservative.
- Full parser: Python via
ast - Heuristic parser: JavaScript, TypeScript, TSX, C#, Swift
- Unsupported languages: emit file nodes only and mark them explicitly
Swift support is intentionally heuristic in v1. The scanner extracts top-level type and function symbols from .swift files, records imports and inheritance/protocol edges conservatively, and avoids generated Apple build output such as .build, DerivedData, Pods, Carthage, and SourcePackages.
The graph covers:
repo,file,class,function,method,test,external_symbolnodescontains,defines,imports,calls,inherits,references,testsedges
It does not attempt:
- semantic embeddings
- watch mode or IDE sync
- SQLite persistence
- whole-program type inference
- guaranteed perfect call resolution across dynamic code
For small and medium repos, direct graph queries plus compact markdown reports are often enough. Do not add heavier retrieval or indexing layers until graph JSON and summaries stop being operationally sufficient.
Command Pattern
Start with one of these instructions:
- "Use dev-context-code-graph to build a code graph for this repo."
- "Use dev-context-code-graph to find the blast radius of
<file-or-symbol>." - "Use dev-context-code-graph to map tests that cover
<module>." - "Use dev-context-code-graph to export a Mermaid diagram for this area of the codebase."
- "Use dev-context-code-graph to generate a repo-area description from graph evidence."
Recommended Layout
code-hub/
├── AGENTS.md
├── code-profiles/
├── graphs/
├── reports/
├── schemas/
└── scripts/Keep generated graph data in code-profiles/, graphs/, and reports/. Do not paste inventories or raw symbol lists into root instruction files.
Stable Types
The skill assumes these canonical contracts:
- schemas/code-profile.schema.json
- schemas/code-graph.schema.json
Do not invent alternate shapes unless a downstream system requires a transform layer.
Validation Checklist
- [ ] Every scanned repo emits one
code-profiles/<repo>.json. - [ ] Every node has a stable normalized ID.
- [ ] Every node with
parent_idhas a matchingcontainsordefinesedge. - [ ] All unsupported or failed parses are marked explicitly.
- [ ]
graphs/code-graph.jsonvalidates cleanly. - [ ] Impact and path queries return bounded, readable results.
- [ ] External targets are represented as
external_symbolnodes instead of dangling edges. - [ ] Findings worth reusing are written back as markdown or diagram files instead of being left only in chat history.
- [ ] Any generated repo or module description is grounded in graph evidence and explicitly notes parse gaps.
Known Traps
- Treating heuristic parser output as semantic truth when the file format or language support is only partial.
- Skipping explicit parse-status and confidence fields, which makes downstream summaries look more certain than the graph can justify.
- Building graph edges directly from ad hoc scripts or LLM summaries instead of normalizing through the profile and schema contracts.
- Assuming unsupported languages can be silently ignored instead of recording the coverage gap that affects blast-radius answers.
- Using one giant graph query for every question when a bounded neighborhood or impact query is enough and far easier to validate.
- Treating the code graph as a replacement for reading source files; it is a routing and blast-radius substrate, not the final authority for behavior.
- Pulling whole-repo graph output into an LLM prompt instead of using graph node IDs, paths, and bounded queries for just-in-time context.
Common Anti-Patterns
- Turning the graph into a prose-heavy report generator and losing the machine-readable substrate that made it trustworthy.
- Inventing new node or edge types opportunistically instead of preserving the stable schema and adding a transform layer if needed.
- Treating external libraries and unresolved symbols as missing data rather than representing them as
external_symbolnodes. - Publishing repo or module descriptions from graph output without calling out parse gaps, heuristics, and unsupported areas.
- Adding heavier retrieval or embeddings before the deterministic graph and query workflow is already operationally sufficient.
- Feeding generated summaries back into graph extraction as if they were source evidence.
- Letting per-repo symbol nodes leak into the portfolio knowledge graph instead of publishing linked code-graph reports or selected summaries.
Navigation
References
- references/code-graph-patterns.md
- references/parser-support-matrix.md
- references/query-recipes.md
Schemas
- schemas/code-profile.schema.json
- schemas/code-graph.schema.json
Scripts
- scripts/scan_code_repo.py
- scripts/build_code_graph.py
- scripts/validate_code_graph.py
- scripts/query_code_graph.py
- scripts/export_code_graph_report.py
- scripts/test_code_graph_regressions.py
Examples
- examples/python-mini-repo.md
- examples/typescript-mini-repo.md
Fact-Checking
- Known bugs, regressions, framework/compiler/runtime footguns, and version-specific crash or workaround guidance must be verified against current primary web sources before being treated as current fact.
- Re-verify parser support claims before user-facing answers if the implementation changes.
- Prefer primary references for parser runtimes, AST behavior, and JSON Schema semantics.
- Use data/sources.json as the curated source map.
{
"metadata": {
"title": "dev-context-code-graph sources",
"skill": "dev-context-code-graph",
"description": "Primary references for code graph extraction, JSON graph contracts, and parser behavior.",
"last_updated": "2026-04-25",
"updated": "2026-04-25",
"total_sources": 6
},
"external_sources": [
{
"name": "garrytan/gbrain",
"url": "https://github.com/garrytan/gbrain",
"commit_sha": "adb02b7826a010700efc968b18df8aaf17d8ffa1",
"license": "MIT",
"extracted_date": "2026-04-13",
"patterns_used": [
"deterministic-collectors-discipline"
],
"research_pack": "docs/research/2026-04-13-context-engineering-skill-scan.md",
"notes": "Cited for convergent-evidence validation of the deterministic-first parser pipeline discipline. No structural changes made to this skill; one-line citation in references/code-graph-patterns.md only."
}
],
"primary_sources": [
{
"name": "code-review-graph",
"url": "https://github.com/tirth8205/code-review-graph",
"description": "Reference implementation for graph-as-artifact code review workflows and blast-radius analysis.",
"add_as_web_search": true
},
{
"name": "JSON Schema",
"url": "https://json-schema.org/",
"description": "Canonical schema reference for the code-profile and code-graph contracts.",
"add_as_web_search": true
},
{
"name": "Python ast",
"url": "https://docs.python.org/3/library/ast.html",
"description": "Primary source for Python AST parsing and node semantics.",
"add_as_web_search": true
},
{
"name": "tree-sitter",
"url": "https://tree-sitter.github.io/tree-sitter/",
"description": "Reference for parser-driven syntax trees and language adapter design.",
"add_as_web_search": true
},
{
"name": "Sourcegraph Code Graph",
"url": "https://sourcegraph.com/docs/cody/core-concepts/code-graph",
"description": "Current product documentation describing code graph data as structural context for codebase-aware AI assistance.",
"add_as_web_search": true
}
]
}
Python Mini Repo Example
Input repo:
app.pydefineshelper()service.pyimportshelper()and definesService.run()test_service.pycallsService.run()
Expected outputs:
- one
reponode - three
filenodes function,class,method, andtestnodesimports,calls, andtestsedges- impact query from
service.pyreachesService.runandtest_service.py
TypeScript Mini Repo Example
Input repo:
src/math.tsexportssum()src/app.tsimportssum()and calls itsrc/app.test.tsimportsrunApp()and asserts behavior
Expected outputs:
- file and function nodes for all three files
- local
importsedges between files callsedges fromrunApptosumtestsedges from the test symbol torunApp
Code Graph Patterns
Table of Contents
- Node Types
- Edge Relations
- ID Rules
- Confidence Rules
- Query Semantics
- Structural Risk Analysis
- V1 Boundaries
Node Types
repo: root node for one repository scanfile: individual source or test filemodule: reserved for future file-to-module enrichmentclass: class or equivalent type declarationfunction: top-level functionmethod: class or instance methodtest: executable test function or test methodexternal_symbol: unresolved or third-party import/call target
Edge Relations
contains: repo contains filedefines: file or class defines a symbolimports: file imports a local file or external symbolcalls: function or method calls another symbolinherits: class extends or derives from another symbolreferences: unresolved or weakly-resolved symbol usagetests: test symbol exercises a function, method, or class
ID Rules
- Normalize IDs to lowercase and replace non-safe characters with
-. - Scope file IDs by repo:
{repo_id}#file#{normalized_path}. - Scope symbol IDs by parent:
{parent_id}#{type}#{normalized_label}. - Scope unresolved targets as
external#{kind}#{normalized_label}. - Never reuse one raw ID for two ontology types.
Confidence Rules
- Python AST extraction:
0.9to0.98 - Heuristic extraction:
0.45to0.75 - Explicit external targets:
0.4to0.7 - Unsupported-language file node:
0.35
Query Semantics
--node: bounded neighborhood in both directions--impact: downstream blast radius using outgoing edges only--from/--to: shortest path search--search: lexical search across labels, paths, tags, summaries, and properties
Use the smallest query that proves the answer. A one-hop neighborhood is the default for orientation; two or three hops are for explicit blast-radius or path questions. Whole-graph export belongs in reports and diagrams, not in the model context window.
When the user asks for code behavior, use the graph to choose files and symbols to inspect, then read the source. The graph is evidence about structure; source files, tests, and runtime traces remain the authority for behavior.
Structural Risk Analysis
Beyond bounded blast radius, three classical algorithms surface structural risk that fan-in and BFS miss. All three are O(V+E) on the graph and run in milliseconds for repos under ~50k symbols.
Articulation Points and Bridges (Tarjan)
An articulation point (cut vertex) is a node whose removal disconnects the graph. A bridge (cut edge) is an edge whose removal does the same. These are the literal single points of failure: the file or import that, if changed wrong, severs whole subsystems.
python3 query_code_graph.py graphs/code-graph.json --articulation-points --top 20
python3 query_code_graph.py graphs/code-graph.json --bridges --top 20When to use:
- Pre-merge review for changes that touch high-degree files. If the changed file is an articulation point, the change deserves extra scrutiny — not because of edit volume, but because a regression breaks more of the graph than a typical edit would.
- Onboarding: articulation points are the files a new engineer should read first. They sit on the most paths.
- Refactoring planning: bridges are candidates for explicit interface boundaries.
Output: ranked list with node_id, path, components_disconnected_if_removed. A node listed alongside weighted_fan_in from the portfolio graph (§5 in dev-context-multi-repo/references/knowledge-graph-patterns.md) gives a richer picture than either alone — fan-in says "many things point at this," articulation says "many things are reachable only through this."
Extended Cycle Detection
The portfolio validator (§11 of knowledge-graph-patterns.md) runs cycle detection only on depends_on. The single-repo code graph should run it on three more relations:
| Relation | Why cycles matter |
|---|---|
imports | Circular imports are a real footgun in Python, JS/TS, Swift; they cause runtime errors, partial-init bugs, or silent type-resolution failures depending on the language. |
inherits | Cyclic inheritance is a compile error in most languages but possible to construct via heuristic extraction over partial files. A reported cycle here usually means the parser misidentified a class — useful as a parser sanity check. |
calls | Recursion is legal but worth surfacing: cycles in calls highlight intentional recursion, mutual recursion across modules, and accidental call loops that bypass tail-call optimization. |
python3 query_code_graph.py graphs/code-graph.json --cycles --relations imports,inherits,callsOutput groups cycles by relation and lists the smallest cycle covering each strongly-connected component. Treat imports cycles as findings to investigate; inherits cycles as parser-correctness signals; calls cycles as informational.
Topological Sort
For DAGs derived from the graph (imports after cycle removal, depends_on, inherits), a topological sort gives canonical build / load / read order:
python3 query_code_graph.py graphs/code-graph.json --topo-sort --relation importsUseful for:
- Generating module-by-module documentation in the right reading order (read leaves first, root last).
- Picking a valid edit sequence when a refactor must touch multiple modules without breaking intermediate states.
- Planning incremental migrations: edit leaves first, then walk up.
If the chosen relation has cycles, return them via --cycles first. Topological sort is only defined on DAGs.
k-Shortest Paths
--from <a> --to <b> returns one shortest path. For propagation analysis ("how could a change in A reach B"), one path understates the risk surface. Add --k 3 to return the top three node-disjoint shortest paths instead.
python3 query_code_graph.py graphs/code-graph.json --from <id> --to <id> --k 3When to use:
- Estimating coupling: more paths between two nodes = harder to cleanly separate them.
- Test planning: each disjoint path is a separate way the change can propagate, so each path deserves its own test.
- Architecture review: zero paths between two modules is a healthy sign of separation; many paths is a coupling smell.
V1 Boundaries
- Prefer deterministic extraction over speculative graph enrichment.
- Emit partial but labeled output rather than hallucinating precise relationships.
- Treat test-link resolution as best effort and confidence-scored, not exact coverage proof.
- Do not merge symbol-level nodes into portfolio-level knowledge graphs. Publish code graph reports and link them from repo catalog pages.
- Do not re-ingest LLM-authored module summaries as source evidence unless the original file paths and graph node IDs remain attached.
This discipline — code for data, LLMs for judgment — is the same one adopted independently in document-ingest systems for knowledge hubs. See garrytan/gbrain docs/guides/deterministic-collectors.md (MIT, commit adb02b7) for a parallel articulation in the note-ingest domain. Convergent evidence that deterministic collectors + latent judgment on structured output is a robust pattern across both symbol-level code graphs and entity-level knowledge graphs.
Parser Support Matrix
| Language | Strategy | Symbol support | Import support | Call support | Notes |
|---|---|---|---|---|---|
| Python | ast | Strong | Strong | Strong | Preferred v1 parser |
| JavaScript | Heuristic | Medium | Strong | Medium | Regex-backed extraction |
| TypeScript | Heuristic | Medium | Strong | Medium | Handles common import and function patterns |
| TSX | Heuristic | Medium | Strong | Medium | Treats component functions as functions |
| C# | Heuristic | Medium | Medium | Medium | Extracts common class and method shapes |
| Swift | Heuristic | Medium | Medium | Medium | Extracts top-level types, functions, imports, and inheritance/protocol edges while skipping generated Apple build trees |
| Other | File-only | None | None | None | Emit unsupported parse status |
Rules:
- Unsupported languages still emit
filenodes. - Unsupported files must be labeled with
parse_status: unsupported. - Heuristic parsers must use lower confidence than AST-backed extraction.
For the staged plan to upgrade the heuristic languages to tree-sitter while keeping Python on ast, see tree-sitter-migration-plan.md.
Code Graph Query Recipes
Concrete invocation patterns for the most common review packets. Each recipe assumes you have already built graphs/code-graph.json via scripts/build_code_graph.py.
All commands use python3 scripts/query_code_graph.py graphs/code-graph.json as the prefix; the prefix is omitted below for brevity.
Table of Contents
- PR-impact packet — what does this change touch
- Refactor-risk packet — where is the structural risk
- Hot-symbol context (PPR) — budget-bounded retrieval around a symbol
- Dead-code candidates — what is unused or weakly connected
- Test-coverage cone — which tests cover this code
- Cycle inventory — circular imports and inheritance loops
- Module-boundary check — articulation points and bridges
- Diff vs canonical layering — topo-sort drift checks
- Module discovery via communities — what are the natural modules
When to use which recipe
| Trigger | Recipe |
|---|---|
| About to merge a PR | PR-impact + Test-coverage cone |
| Considering renaming or moving a symbol | PR-impact + Refactor-risk |
| Onboarding to an unfamiliar module | Hot-symbol context + Module-boundary check |
| Cleaning up before a release | Dead-code + Cycle inventory |
| Planning architectural changes | Refactor-risk + Diff vs canonical layering |
---
PR-impact packet
Goal: list every symbol and file that reaches the changed symbol, plus their tests, within 2 hops.
# 1. Two-hop blast radius from the changed symbol
--impact "fn:src/api.py:create_user" --hops 2 --format json --output reports/pr-impact.json
# 2. Mermaid view for the PR description
--impact "fn:src/api.py:create_user" --hops 2 --format mermaid --output reports/pr-impact.mmd
# 3. Tests that exercise this symbol
--node "fn:src/api.py:create_user" --hops 1 --format table | grep -E "^test|tests_"Sanity checks:
- If the impact node count is > 50, your symbol is too central — break the change into smaller commits.
- If no test edges appear, flag missing coverage before merging.
---
Refactor-risk packet
Goal: surface the structural risk that a refactor would carry — articulation points, bridges, and cycles in the affected subgraph.
# 1. Articulation points across the import graph
--articulation-points --relations imports --top 30 --format table --output reports/refactor-articulation.md
# 2. Bridges (single edges whose removal disconnects the graph)
--bridges --relations imports,inherits --top 30 --format table --output reports/refactor-bridges.md
# 3. Cycles that the refactor must preserve or break carefully
--cycles --relations imports,inherits,calls --format json --output reports/refactor-cycles.json
# 4. Topological depth of the affected files
--topo-sort imports --format json --output reports/refactor-topo.jsonReading the output:
- An articulation point with high
betweenness_proxyis a load-bearing module — refactor with feature-flag rollout. - A bridge between two large components is a likely seam for splitting the codebase.
- Cycles in
importsare usually fixable; cycles ininheritsindicate deeper coupling.
---
Hot-symbol context (PPR)
Goal: budget-bounded retrieval around 1–3 hot symbols. Use this instead of --impact --hops N when the symbol's neighbourhood blows past 100 nodes.
# Single hot symbol — top 30 most-related nodes
--ppr --seed "fn:src/api.py:create_user" --top 30 --format table
# Multiple hot symbols (tax-rate calculation cluster)
--ppr \
--seed "fn:src/tax/calc.py:apply_vat" \
--seed "fn:src/tax/calc.py:apply_corp_tax" \
--seed "class:src/tax/rates.py:RateTable" \
--top 50 \
--filter-type function,method,class \
--output reports/tax-context.json
# Tighter teleport (smaller neighbourhood, less drift)
--ppr --seed "fn:src/api.py:create_user" --alpha 0.3 --top 20When PPR beats hop-bounded BFS:
- Your symbol has high fan-out (>20 callers) so 2-hop BFS already exceeds 200 nodes.
- You want a ranked retrieval set, not the full neighbourhood.
- The reviewer needs context for an LLM that has a token budget.
α (alpha) tuning:
- α = 0.10–0.15 — wide context, follows long call chains
- α = 0.20–0.30 — tight context, stays close to the seed
- α = 0.50+ — almost only the seed and direct neighbours
---
Dead-code candidates
Goal: find symbols with zero or near-zero fan-in.
# Rank by fan-in, ascending — bottom of the list is suspicious
--rank --top 0 --format json --output reports/rank-all.json
# Then surface bottom-fan-in functions in code (jq):
jq '.results | map(select(.type == "function" and .importance == 0)) | .[].id' reports/rank-all.json
# Cross-check against tests — a function with fan-in 0 but a test edge is a public API, not dead
--node "fn:src/utils.py:helper" --hops 1 --format tableDisambiguation:
- Fan-in 0 + no test edge + no
external_symbolreference → strong dead-code candidate. - Fan-in 0 + a test edge → public API; do not delete.
- Fan-in 0 + a CLI/handler decorator → entrypoint; do not delete.
---
Test-coverage cone
Goal: confirm that a module is covered by tests via the tests edge type.
# Outgoing tests edges from the test files in the module
--node "file:tests/test_api.py" --hops 2 --format table
# Reverse: which tests cover a production symbol
--impact "fn:src/api.py:create_user" --hops 2 --format json \
| jq '.edges | map(select(.relation == "tests")) | .[].source'Failure modes:
- Heuristic parsers (JS/TS/Swift/C#) under-resolve test edges — treat absence of edges as "unknown coverage", not "no coverage".
- Always cross-check parse-status fields:
confidence < 0.7means the call resolution may be wrong.
---
Cycle inventory
Goal: locate circular imports, mutually-recursive class hierarchies, and call cycles.
# All cycles, all relations
--cycles --format json --output reports/cycles.json
# Imports-only (most actionable)
--cycles --relations imports --format table
# Class-level cycles (rare but dangerous)
--cycles --relations inherits --format jsonTriage:
- Import cycles → break with lazy import or interface extraction.
- Inheritance cycles → almost always a bug or invalid heuristic parse.
- Call cycles → expected for recursive algorithms; ignore if intentional.
---
Module-boundary check
Goal: validate that a logical module corresponds to a structurally-cohesive subgraph.
# Articulation points limited to the imports relation
--articulation-points --relations imports --top 50 --format table
# Bridges between subsystems
--bridges --relations imports --top 50 --format table
# Manually inspect a candidate boundary
--from "file:src/payments/checkout.py" --to "file:src/notifications/email.py" --max-hops 5 --k 3 --format tableHealthy signs:
- Articulation points are a small set of well-named "facade" or "interface" modules.
- Bridges run through these facades, not through ad-hoc helper files.
- 3+ disjoint paths exist between two subsystems → loose coupling.
- 1 disjoint path → fragile boundary; refactor before further decoupling.
---
Diff vs canonical layering
Goal: detect when a new commit adds an edge that violates the canonical layer order (e.g. domain → infra instead of infra → domain).
# 1. Topological sort of imports
--topo-sort imports --format json --output reports/topo-current.json
# 2. Compare against the previous run committed at reports/topo-baseline.json
diff <(jq '.results' reports/topo-baseline.json) <(jq '.results' reports/topo-current.json)
# 3. If the topo-sort errors with cycles_detected, drill into the cycle
--cycles --relations imports --format json | jq '.results[].nodes'Workflow:
- Commit a
reports/topo-baseline.jsonafter each architectural review. - Run a CI check that fails if the diff introduces backward edges.
- Treat new cycles as PR-blocking unless explicitly approved.
---
Module discovery via Louvain communities
Goal: discover the codebase's natural module boundaries from call/import structure, then compare them against the directory layout.
# 1. Detect communities at default resolution
--communities --format table
# 2. Tighten or loosen modularity (γ=1 default)
# Higher γ → smaller, more cohesive modules; lower γ → larger, looser modules
--communities --resolution 1.4 --top 30 --format json --output reports/communities-tight.json
--communities --resolution 0.7 --top 30 --format json --output reports/communities-loose.json
# 3. Inspect cross-directory cohesion in each community
jq -r '.communities[] | {community_id, size, top_files: (.top_files | keys)}' reports/communities-tight.jsonReading the output:
modularityQ ≈ 0.4–0.7 = strong community structure; <0.3 = weak partition (tightly coupled or single-module repo).- A community whose
top_filesspan more than two directories is a cross-cutting concern — refactor candidate or evidence the directory structure has drifted from actual coupling. --community-seedis deterministic; freeze it for reproducible review reports.
When to use: onboarding into an unfamiliar codebase, validating a planned package split, finding hidden coupling, or grounding a "natural module map" before proposing architectural moves.
Limitation: pure-python Louvain is fine to ~50k edges. For very large graphs, swap in a native Leiden implementation (igraph) — same input/output contract, better partition stability.
---
Output discipline
When a recipe produces durable knowledge, file the report into a stable path:
reports/pr-impact-<pr-number>.md— one-shot, link from the PRreports/refactor-<area>.md— durable, link from the area's onboarding docreports/cycles.json+reports/cycles.md— pair JSON + summary, regenerate on each releasereports/topo-baseline.json— long-lived, regenerate on architectural reviewsreports/communities.json+reports/modules.md— pair JSON + narrative summary, regenerate during architectural reviews
Anything reused once should not stay only in chat output.
{
"$schema": "https://json-schema.org/draft/2020-12/schema",
"$id": "https://example.com/dev-context-code-graph/schemas/code-graph.schema.json",
"title": "CodeGraph",
"type": "object",
"required": ["meta", "nodes", "edges"],
"properties": {
"meta": {
"type": "object",
"required": [
"generated_at",
"version",
"graph_contract_version",
"build_source",
"node_count",
"edge_count",
"repo_count"
],
"properties": {
"generated_at": { "type": "string", "format": "date-time" },
"version": { "type": "string" },
"graph_contract_version": { "type": "string" },
"build_source": { "type": "string", "enum": ["code_profiles"] },
"node_count": { "type": "integer" },
"edge_count": { "type": "integer" },
"repo_count": { "type": "integer" },
"base_commit_shas": {
"type": "object",
"additionalProperties": { "type": "string" }
},
"validation": {
"type": "object",
"properties": {
"checks_passed": { "type": "integer" },
"checks_total": { "type": "integer" },
"last_validated_at": { "type": "string", "format": "date-time" }
}
}
}
},
"nodes": {
"type": "array",
"items": {
"type": "object",
"required": ["id", "type", "label"],
"properties": {
"id": { "type": "string" },
"type": {
"type": "string",
"enum": ["repo", "file", "module", "class", "function", "method", "test", "external_symbol"]
},
"label": { "type": "string" },
"summary": { "type": "string" },
"tags": {
"type": "array",
"items": { "type": "string" }
},
"importance": {
"type": "number",
"minimum": 0
},
"parent_id": { "type": "string" },
"path": { "type": "string" },
"language": { "type": "string" },
"kind": { "type": "string" },
"parse_status": { "type": "string" },
"line_start": { "type": "integer", "minimum": 1 },
"line_end": { "type": "integer", "minimum": 1 },
"properties": { "type": "object" },
"confidence": {
"type": "number",
"minimum": 0,
"maximum": 1
},
"evidence": {
"type": "array",
"items": { "type": "object" }
},
"first_seen_at": { "type": "string", "format": "date-time" },
"last_verified_at": { "type": "string", "format": "date-time" },
"stale": { "type": "boolean" },
"community_id": { "type": "string", "description": "Louvain community label (community_0, community_1, ...) — populated by query_code_graph.py --communities or persisted by an enrichment pass" },
"ppr_score": { "type": "number", "minimum": 0, "description": "Cached Personalized PageRank score from a prior --ppr run; runtime queries should regenerate rather than trust this value" }
}
}
},
"edges": {
"type": "array",
"items": {
"type": "object",
"required": ["source", "target", "relation", "group"],
"properties": {
"source": { "type": "string" },
"target": { "type": "string" },
"relation": {
"type": "string",
"enum": ["contains", "defines", "imports", "calls", "inherits", "references", "tests"]
},
"group": {
"type": "string",
"enum": ["structural", "dependency", "behavioral", "semantic"]
},
"weight": {
"type": "number",
"minimum": 0,
"maximum": 1
},
"confidence": {
"type": "number",
"minimum": 0,
"maximum": 1
},
"notes": { "type": "string" },
"evidence": {
"type": "array",
"items": { "type": "object" }
},
"first_seen_at": { "type": "string", "format": "date-time" },
"last_verified_at": { "type": "string", "format": "date-time" },
"stale": { "type": "boolean" }
}
}
}
}
}
{
"$schema": "https://json-schema.org/draft/2020-12/schema",
"$id": "https://example.com/dev-context-code-graph/schemas/code-profile.schema.json",
"title": "CodeProfile",
"type": "object",
"additionalProperties": false,
"required": [
"repo_id",
"repo_name",
"repo_path",
"languages",
"files",
"symbols",
"relations",
"summary",
"evidence",
"confidence_score",
"last_scanned_at"
],
"properties": {
"repo_id": { "type": "string" },
"repo_name": { "type": "string" },
"repo_path": { "type": "string" },
"languages": {
"type": "array",
"items": { "type": "string" }
},
"files": {
"type": "array",
"items": {
"type": "object",
"additionalProperties": false,
"required": ["id", "path", "language", "kind", "parse_status"],
"properties": {
"id": { "type": "string" },
"path": { "type": "string" },
"language": { "type": "string" },
"kind": {
"type": "string",
"enum": ["source", "test", "config", "unknown"]
},
"parse_status": {
"type": "string",
"enum": ["parsed", "heuristic", "unsupported", "error", "skipped"]
},
"confidence": {
"type": "number",
"minimum": 0,
"maximum": 1
}
}
}
},
"symbols": {
"type": "array",
"items": {
"type": "object",
"additionalProperties": false,
"required": ["id", "type", "label", "path", "language", "parent_id"],
"properties": {
"id": { "type": "string" },
"type": {
"type": "string",
"enum": ["module", "class", "function", "method", "test"]
},
"label": { "type": "string" },
"path": { "type": "string" },
"language": { "type": "string" },
"parent_id": { "type": "string" },
"line_start": { "type": "integer", "minimum": 1 },
"line_end": { "type": "integer", "minimum": 1 },
"signature": { "type": "string" },
"confidence": {
"type": "number",
"minimum": 0,
"maximum": 1
}
}
}
},
"relations": {
"type": "array",
"items": {
"type": "object",
"additionalProperties": false,
"required": ["source", "target", "relation", "group", "confidence"],
"properties": {
"source": { "type": "string" },
"target": { "type": "string" },
"relation": {
"type": "string",
"enum": ["contains", "defines", "imports", "calls", "inherits", "references", "tests"]
},
"group": {
"type": "string",
"enum": ["structural", "dependency", "behavioral", "semantic"]
},
"confidence": {
"type": "number",
"minimum": 0,
"maximum": 1
},
"notes": { "type": "string" }
}
}
},
"summary": { "type": "string" },
"evidence": {
"type": "array",
"items": {
"type": "object",
"additionalProperties": false,
"required": ["path", "reason"],
"properties": {
"path": { "type": "string" },
"reason": { "type": "string" }
}
}
},
"confidence_score": {
"type": "number",
"minimum": 0,
"maximum": 1
},
"last_scanned_at": {
"type": "string",
"format": "date-time"
}
}
}
#!/usr/bin/env python3
"""Build a code graph JSON from code-profile artifacts."""
from __future__ import annotations
import argparse
import json
import re
import sys
from datetime import datetime, timezone
from pathlib import Path
GRAPH_CONTRACT_VERSION = "1.0"
EDGE_WEIGHTS = {
"contains": 1.0,
"defines": 0.95,
"imports": 0.72,
"calls": 0.8,
"inherits": 0.78,
"references": 0.55,
"tests": 0.7,
}
def now_iso() -> str:
return datetime.now(timezone.utc).isoformat()
def normalize_id(text: str) -> str:
return re.sub(r"[^a-z0-9._-]", "-", text.lower().strip()).strip("-")
class GraphBuilder:
def __init__(self) -> None:
self.nodes: dict[str, dict] = {}
self.edge_keys: set[tuple[str, str, str]] = set()
self.edges: list[dict] = []
self.repo_ids: set[str] = set()
def add_node(self, node_id: str, node_type: str, label: str, **kwargs) -> None:
if node_id not in self.nodes:
node = {"id": node_id, "type": node_type, "label": label}
for key, value in kwargs.items():
if value is not None:
node[key] = value
self.nodes[node_id] = node
return
existing = self.nodes[node_id]
for key, value in kwargs.items():
if value is None:
continue
if key == "tags":
existing.setdefault("tags", [])
existing["tags"] = sorted(set(existing["tags"]) | set(value))
elif key == "properties":
existing.setdefault("properties", {})
for prop_key, prop_value in value.items():
existing["properties"].setdefault(prop_key, prop_value)
elif key == "evidence":
existing.setdefault("evidence", [])
existing["evidence"].extend(value)
elif key not in existing:
existing[key] = value
def add_edge(self, source: str, target: str, relation: str, group: str, confidence: float | None = None, notes: str | None = None) -> None:
key = (source, target, relation)
if key in self.edge_keys:
return
self.edge_keys.add(key)
payload = {
"source": source,
"target": target,
"relation": relation,
"group": group,
"weight": EDGE_WEIGHTS.get(relation, 0.5),
}
if confidence is not None:
payload["confidence"] = round(confidence, 3)
if notes:
payload["notes"] = notes
self.edges.append(payload)
def ensure_external_node(self, node_id: str) -> None:
if node_id in self.nodes:
return
label = node_id.split("#", 2)[-1].replace("-", " ")
self.add_node(node_id, "external_symbol", label, confidence=0.45)
def to_dict(self) -> dict:
return {
"meta": {
"generated_at": now_iso(),
"version": "1.0",
"graph_contract_version": GRAPH_CONTRACT_VERSION,
"build_source": "code_profiles",
"node_count": len(self.nodes),
"edge_count": len(self.edges),
"repo_count": len(self.repo_ids),
},
"nodes": list(self.nodes.values()),
"edges": self.edges,
}
def build_from_profiles(profiles_dir: Path, graph: GraphBuilder) -> None:
json_files = sorted(profiles_dir.glob("*.json"))
if not json_files:
raise SystemExit(f"No profile JSON files found in {profiles_dir}")
for path in json_files:
profile = json.loads(path.read_text(encoding="utf-8"))
repo_id = profile["repo_id"]
graph.repo_ids.add(repo_id)
graph.add_node(
repo_id,
"repo",
profile["repo_name"],
path=profile.get("repo_path"),
summary=profile.get("summary"),
confidence=profile.get("confidence_score"),
first_seen_at=profile.get("last_scanned_at"),
last_verified_at=profile.get("last_scanned_at"),
)
file_ids = {entry["id"] for entry in profile.get("files", [])}
symbol_ids = {entry["id"] for entry in profile.get("symbols", [])}
for entry in profile.get("files", []):
graph.add_node(
entry["id"],
"file",
entry["path"],
parent_id=repo_id,
path=entry["path"],
language=entry["language"],
kind=entry.get("kind"),
parse_status=entry.get("parse_status"),
confidence=entry.get("confidence"),
first_seen_at=profile.get("last_scanned_at"),
last_verified_at=profile.get("last_scanned_at"),
)
graph.add_edge(repo_id, entry["id"], "contains", "structural", 1.0)
for entry in profile.get("symbols", []):
graph.add_node(
entry["id"],
entry["type"],
entry["label"],
parent_id=entry["parent_id"],
path=entry.get("path"),
language=entry.get("language"),
line_start=entry.get("line_start"),
line_end=entry.get("line_end"),
properties={"signature": entry.get("signature")} if entry.get("signature") else {},
confidence=entry.get("confidence"),
first_seen_at=profile.get("last_scanned_at"),
last_verified_at=profile.get("last_scanned_at"),
)
graph.add_edge(entry["parent_id"], entry["id"], "defines", "structural", entry.get("confidence", 0.8))
known_nodes = file_ids | symbol_ids | {repo_id}
for relation in profile.get("relations", []):
source = relation["source"]
target = relation["target"]
if source not in known_nodes and source not in graph.nodes:
graph.ensure_external_node(source)
if target not in known_nodes and target not in graph.nodes:
graph.ensure_external_node(target)
graph.add_edge(
source,
target,
relation["relation"],
relation["group"],
relation.get("confidence"),
relation.get("notes"),
)
def main() -> int:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--profiles", required=True, help="Directory containing code-profile JSON files")
parser.add_argument("--output", help="Output path for code-graph.json")
args = parser.parse_args()
profiles_dir = Path(args.profiles).expanduser().resolve()
output = Path(args.output).expanduser().resolve() if args.output else (profiles_dir.parent / "graphs" / "code-graph.json")
output.parent.mkdir(parents=True, exist_ok=True)
graph = GraphBuilder()
build_from_profiles(profiles_dir, graph)
payload = graph.to_dict()
output.write_text(json.dumps(payload, indent=2), encoding="utf-8")
print(f"[ok] Wrote code graph to {output}")
return 0
if __name__ == "__main__":
raise SystemExit(main())
#!/usr/bin/env python3
"""Generate static Markdown and HTML reports from code-graph.json."""
from __future__ import annotations
import argparse
import json
from collections import Counter, defaultdict
from pathlib import Path
def load_json(path: Path | None) -> dict | None:
if path is None or not path.exists():
return None
return json.loads(path.read_text(encoding="utf-8"))
def top_nodes(nodes: list[dict], edges: list[dict], limit: int = 20) -> list[dict]:
incoming = defaultdict(float)
for edge in edges:
incoming[edge["target"]] += float(edge.get("weight", 1.0))
ranked = []
for node in nodes:
ranked.append(
{
"label": node.get("label", node["id"]),
"type": node.get("type"),
"path": node.get("path", ""),
"importance": round(incoming.get(node["id"], 0.0), 3),
}
)
ranked.sort(key=lambda row: (-row["importance"], row["label"]))
return ranked[:limit]
def render_markdown(graph: dict, validation: dict | None) -> str:
nodes = graph.get("nodes", [])
edges = graph.get("edges", [])
node_counts = Counter(node.get("type") for node in nodes)
edge_counts = Counter(edge.get("relation") for edge in edges)
lines = ["# Code Graph Report", ""]
lines.append(f"- Generated: {graph.get('meta', {}).get('generated_at', 'unknown')}")
lines.append(f"- Repos: {graph.get('meta', {}).get('repo_count', 0)}")
lines.append(f"- Nodes: {graph.get('meta', {}).get('node_count', len(nodes))}")
lines.append(f"- Edges: {graph.get('meta', {}).get('edge_count', len(edges))}")
if validation:
lines.append(f"- Validation: {validation.get('checks_passed', 0)}/{validation.get('checks_total', 0)} checks passed")
lines.append("")
lines.append("## Node Types")
lines.append("")
lines.append("| Type | Count |")
lines.append("|------|-------|")
for node_type, count in sorted(node_counts.items()):
lines.append(f"| {node_type} | {count} |")
lines.append("")
lines.append("## Edge Relations")
lines.append("")
lines.append("| Relation | Count |")
lines.append("|----------|-------|")
for relation, count in sorted(edge_counts.items()):
lines.append(f"| {relation} | {count} |")
lines.append("")
lines.append("## Most Referenced Nodes")
lines.append("")
lines.append("| Label | Type | Path | Importance |")
lines.append("|-------|------|------|------------|")
for row in top_nodes(nodes, edges):
lines.append(f"| {row['label']} | {row['type']} | {row['path']} | {row['importance']} |")
return "\n".join(lines) + "\n"
def render_html(markdown_report: str) -> str:
body = markdown_report.replace("&", "&").replace("<", "<").replace(">", ">")
return (
"<!doctype html><html><head><meta charset='utf-8'>"
"<title>Code Graph Report</title>"
"<style>body{font-family:system-ui,sans-serif;max-width:1080px;margin:40px auto;padding:0 24px;line-height:1.5}pre{white-space:pre-wrap}</style>"
"</head><body><pre>"
+ body
+ "</pre></body></html>"
)
def main() -> int:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("graph", help="Path to code-graph.json")
parser.add_argument("--output-dir", required=True, help="Directory for markdown and html reports")
parser.add_argument("--validation", help="Optional path to code-graph-validation.json")
args = parser.parse_args()
graph = load_json(Path(args.graph).expanduser().resolve())
if graph is None:
raise SystemExit("Graph file not found")
validation = load_json(Path(args.validation).expanduser().resolve()) if args.validation else None
output_dir = Path(args.output_dir).expanduser().resolve()
output_dir.mkdir(parents=True, exist_ok=True)
markdown_report = render_markdown(graph, validation)
html_report = render_html(markdown_report)
md_path = output_dir / "code-graph-report.md"
html_path = output_dir / "code-graph-report.html"
md_path.write_text(markdown_report, encoding="utf-8")
html_path.write_text(html_report, encoding="utf-8")
print(f"[ok] Wrote {md_path}")
print(f"[ok] Wrote {html_path}")
return 0
if __name__ == "__main__":
raise SystemExit(main())
#!/usr/bin/env python3
"""Query a code graph by neighborhood, impact, path, type, rank, structural risk, search, or diagram export."""
from __future__ import annotations
import argparse
import json
import re
import sys
from collections import defaultdict, deque
from pathlib import Path
MERMAID_DIRECTIONS = ("LR", "RL", "TD", "TB", "BT")
MERMAID_TYPE_STYLES = {
"repo": "fill:#dbeafe,stroke:#1d4ed8,color:#0f172a",
"file": "fill:#f3f4f6,stroke:#4b5563,color:#111827",
"class": "fill:#ede9fe,stroke:#7c3aed,color:#1f2937",
"function": "fill:#dcfce7,stroke:#15803d,color:#14532d",
"method": "fill:#ccfbf1,stroke:#0f766e,color:#134e4a",
"test": "fill:#fee2e2,stroke:#dc2626,color:#7f1d1d",
"external_symbol": "fill:#fde68a,stroke:#d97706,color:#78350f",
}
def load_graph(graph_path: str) -> tuple[list[dict], list[dict], dict[str, dict]]:
data = json.loads(Path(graph_path).read_text(encoding="utf-8"))
nodes = data.get("nodes", [])
edges = data.get("edges", [])
return nodes, edges, {node["id"]: node for node in nodes if "id" in node}
def build_adjacency(edges: list[dict]) -> tuple[dict[str, list[tuple[str, dict]]], dict[str, list[tuple[str, dict]]]]:
forward: dict[str, list[tuple[str, dict]]] = {}
backward: dict[str, list[tuple[str, dict]]] = {}
for edge in edges:
source = edge.get("source")
target = edge.get("target")
if not source or not target:
continue
forward.setdefault(source, []).append((target, edge))
backward.setdefault(target, []).append((source, edge))
return forward, backward
def bfs_neighborhood(start: str, forward: dict, backward: dict, hops: int) -> tuple[set[str], list[dict]]:
visited_nodes = {start}
visited_edges: list[dict] = []
edge_keys: set[tuple[str, str, str]] = set()
queue = deque([(start, 0)])
seen = {start}
while queue:
node_id, depth = queue.popleft()
if depth >= hops:
continue
for neighbor, edge in forward.get(node_id, []):
key = (edge["source"], edge["target"], edge["relation"])
if key not in edge_keys:
edge_keys.add(key)
visited_edges.append(edge)
visited_nodes.add(neighbor)
if neighbor not in seen:
seen.add(neighbor)
queue.append((neighbor, depth + 1))
for neighbor, edge in backward.get(node_id, []):
key = (edge["source"], edge["target"], edge["relation"])
if key not in edge_keys:
edge_keys.add(key)
visited_edges.append(edge)
visited_nodes.add(neighbor)
if neighbor not in seen:
seen.add(neighbor)
queue.append((neighbor, depth + 1))
return visited_nodes, visited_edges
def bfs_impact(start: str, forward: dict, hops: int) -> tuple[set[str], list[dict]]:
visited_nodes = {start}
visited_edges: list[dict] = []
edge_keys: set[tuple[str, str, str]] = set()
queue = deque([(start, 0)])
seen = {start}
while queue:
node_id, depth = queue.popleft()
if depth >= hops:
continue
for neighbor, edge in forward.get(node_id, []):
key = (edge["source"], edge["target"], edge["relation"])
if key not in edge_keys:
edge_keys.add(key)
visited_edges.append(edge)
visited_nodes.add(neighbor)
if neighbor not in seen:
seen.add(neighbor)
queue.append((neighbor, depth + 1))
return visited_nodes, visited_edges
def bfs_paths(start: str, end: str, forward: dict, max_hops: int) -> list[list[str]]:
if start == end:
return [[start]]
queue: deque[tuple[str, list[str]]] = deque([(start, [start])])
found: list[list[str]] = []
shortest = None
while queue:
node_id, path = queue.popleft()
if shortest is not None and len(path) - 1 >= shortest:
continue
if len(path) - 1 >= max_hops:
continue
for neighbor, _edge in forward.get(node_id, []):
if neighbor in path:
continue
new_path = path + [neighbor]
if neighbor == end:
shortest = len(new_path) - 1
found.append(new_path)
continue
queue.append((neighbor, new_path))
return found
def k_node_disjoint_paths(start: str, end: str, forward: dict, max_hops: int, k: int) -> list[list[str]]:
"""Return up to k shortest paths, avoiding intermediate nodes from prior paths."""
if k <= 1:
return bfs_paths(start, end, forward, max_hops)
blocked: set[str] = set()
paths: list[list[str]] = []
for _ in range(k):
queue: deque[tuple[str, list[str]]] = deque([(start, [start])])
found_path: list[str] | None = None
while queue:
node_id, path = queue.popleft()
if len(path) - 1 >= max_hops:
continue
for neighbor, _edge in forward.get(node_id, []):
if neighbor in path:
continue
if neighbor in blocked and neighbor not in {start, end}:
continue
next_path = path + [neighbor]
if neighbor == end:
found_path = next_path
queue.clear()
break
queue.append((neighbor, next_path))
if not found_path:
break
paths.append(found_path)
blocked.update(found_path[1:-1])
return paths
def relation_set(raw: str | None) -> set[str]:
if not raw:
return set()
return {item.strip() for item in raw.split(",") if item.strip()}
def filtered_edges(edges: list[dict], relations: set[str]) -> list[dict]:
if not relations:
return edges
return [edge for edge in edges if edge.get("relation") in relations]
def undirected_neighbors(edges: list[dict]) -> dict[str, set[str]]:
graph: dict[str, set[str]] = defaultdict(set)
for edge in edges:
source = edge.get("source")
target = edge.get("target")
if not source or not target:
continue
graph[source].add(target)
graph[target].add(source)
return graph
def articulation_points(edges: list[dict], top: int = 20) -> list[dict]:
graph = undirected_neighbors(edges)
timer = 0
discovery: dict[str, int] = {}
low: dict[str, int] = {}
parent: dict[str, str | None] = {}
points: dict[str, int] = {}
def dfs(node: str) -> None:
nonlocal timer
discovery[node] = low[node] = timer
timer += 1
children = 0
separated = 0
for neighbor in sorted(graph[node]):
if neighbor not in discovery:
parent[neighbor] = node
children += 1
dfs(neighbor)
low[node] = min(low[node], low[neighbor])
if parent.get(node) is None:
continue
if low[neighbor] >= discovery[node]:
separated += 1
elif neighbor != parent.get(node):
low[node] = min(low[node], discovery[neighbor])
if parent.get(node) is None and children > 1:
points[node] = children
elif parent.get(node) is not None and separated:
points[node] = separated + 1
for node in sorted(graph):
if node not in discovery:
parent[node] = None
dfs(node)
rows = [
{"node_id": node, "components_disconnected_if_removed": count}
for node, count in points.items()
]
rows.sort(key=lambda item: (-item["components_disconnected_if_removed"], item["node_id"]))
return rows if top == 0 else rows[:top]
def bridge_edges(edges: list[dict], top: int = 20) -> list[dict]:
graph = undirected_neighbors(edges)
timer = 0
discovery: dict[str, int] = {}
low: dict[str, int] = {}
parent: dict[str, str | None] = {}
bridges: list[dict] = []
def dfs(node: str) -> None:
nonlocal timer
discovery[node] = low[node] = timer
timer += 1
for neighbor in sorted(graph[node]):
if neighbor not in discovery:
parent[neighbor] = node
dfs(neighbor)
low[node] = min(low[node], low[neighbor])
if low[neighbor] > discovery[node]:
bridges.append({"source": node, "target": neighbor})
elif neighbor != parent.get(node):
low[node] = min(low[node], discovery[neighbor])
for node in sorted(graph):
if node not in discovery:
parent[node] = None
dfs(node)
bridges.sort(key=lambda item: (item["source"], item["target"]))
return bridges if top == 0 else bridges[:top]
def cycles_by_relation(edges: list[dict], relations: set[str]) -> list[dict]:
selected_relations = relations or {str(edge.get("relation")) for edge in edges if edge.get("relation")}
rows: list[dict] = []
for relation in sorted(selected_relations):
forward = defaultdict(list)
for edge in edges:
if edge.get("relation") == relation:
forward[edge.get("source")].append(edge.get("target"))
visited: set[str] = set()
stack: list[str] = []
in_stack: set[str] = set()
seen_cycles: set[tuple[str, ...]] = set()
def dfs(node: str) -> None:
visited.add(node)
stack.append(node)
in_stack.add(node)
for neighbor in sorted(item for item in forward.get(node, []) if item):
if neighbor not in visited:
dfs(neighbor)
elif neighbor in in_stack:
idx = stack.index(neighbor)
cycle = stack[idx:] + [neighbor]
canonical = tuple(sorted(cycle[:-1]))
if canonical not in seen_cycles:
seen_cycles.add(canonical)
rows.append({"relation": relation, "cycle": cycle, "length": len(cycle) - 1})
stack.pop()
in_stack.remove(node)
for node in sorted(forward):
if node not in visited:
dfs(node)
rows.sort(key=lambda item: (item["relation"], item["length"], item["cycle"]))
return rows
def topological_sort(edges: list[dict], relation: str) -> tuple[list[str], list[dict]]:
relation_edges = [edge for edge in edges if edge.get("relation") == relation]
cycles = cycles_by_relation(relation_edges, {relation})
if cycles:
return [], cycles
nodes: set[str] = set()
indegree: dict[str, int] = defaultdict(int)
forward: dict[str, list[str]] = defaultdict(list)
for edge in relation_edges:
source = edge.get("source")
target = edge.get("target")
if not source or not target:
continue
nodes.update([source, target])
forward[source].append(target)
indegree[target] += 1
indegree.setdefault(source, 0)
queue = deque(sorted(node for node in nodes if indegree[node] == 0))
ordered: list[str] = []
while queue:
node = queue.popleft()
ordered.append(node)
for neighbor in sorted(forward.get(node, [])):
indegree[neighbor] -= 1
if indegree[neighbor] == 0:
queue.append(neighbor)
return ordered, []
def query_search(nodes: list[dict], query: str, types: set[str] | None = None, limit: int = 20) -> list[dict]:
terms = [term for term in re.split(r"\s+", query.lower().strip()) if term]
matches: list[tuple[int, dict]] = []
for node in nodes:
if types and node.get("type") not in types:
continue
haystack = " ".join(
[
str(node.get("id", "")),
str(node.get("label", "")),
str(node.get("path", "")),
str(node.get("summary", "")),
" ".join(str(tag) for tag in node.get("tags", [])),
" ".join(f"{key} {value}" for key, value in (node.get("properties") or {}).items()),
]
).lower()
score = sum(term in haystack for term in terms)
if score:
matches.append((score, node))
matches.sort(key=lambda item: (-item[0], item[1].get("label", item[1]["id"])))
limited = matches if limit == 0 else matches[:limit]
return [node for _score, node in limited]
def personalized_pagerank(
nodes: list[dict],
edges: list[dict],
seeds: list[str],
*,
alpha: float = 0.15,
max_iter: int = 100,
tol: float = 1e-9,
) -> dict[str, float]:
"""
Weighted Personalized PageRank for hot-symbol blast-radius retrieval.
Mirrors the multi-repo implementation: dependency-free, dangling mass
redistributed through the seed vector, weights honoured per-edge.
"""
node_ids = [node["id"] for node in nodes if "id" in node]
if not node_ids:
return {}
node_set = set(node_ids)
seed_set = [seed for seed in seeds if seed in node_set]
if not seed_set:
raise ValueError("at least one seed must exist in the graph")
personalization = {node_id: 0.0 for node_id in node_ids}
seed_weight = 1.0 / len(seed_set)
for seed in seed_set:
personalization[seed] = seed_weight
outgoing: dict[str, list[tuple[str, float]]] = {node_id: [] for node_id in node_ids}
for edge in edges:
source = edge.get("source")
target = edge.get("target")
if source not in node_set or target not in node_set:
continue
try:
weight = float(edge.get("weight", 1.0))
except (TypeError, ValueError):
weight = 1.0
if weight <= 0:
continue
outgoing[source].append((target, weight))
scores = dict(personalization)
for _ in range(max_iter):
next_scores = {node_id: alpha * personalization[node_id] for node_id in node_ids}
dangling_mass = 0.0
for node_id, score in scores.items():
weighted_targets = outgoing.get(node_id, [])
if not weighted_targets:
dangling_mass += score
continue
total_weight = sum(weight for _target, weight in weighted_targets)
if total_weight <= 0:
dangling_mass += score
continue
walk_mass = (1.0 - alpha) * score
for target, weight in weighted_targets:
next_scores[target] += walk_mass * (weight / total_weight)
if dangling_mass:
redistributed = (1.0 - alpha) * dangling_mass
for node_id, seed_score in personalization.items():
if seed_score:
next_scores[node_id] += redistributed * seed_score
delta = sum(abs(next_scores[node_id] - scores.get(node_id, 0.0)) for node_id in node_ids)
scores = next_scores
if delta < tol:
break
total = sum(scores.values())
if total > 0:
scores = {node_id: value / total for node_id, value in scores.items()}
return scores
def _undirected_adjacency(
nodes: list[dict],
edges: list[dict],
) -> tuple[dict[str, dict[str, float]], dict[str, float], float]:
"""Symmetric weighted adjacency for community detection."""
node_ids = [node["id"] for node in nodes if "id" in node]
node_set = set(node_ids)
adjacency: dict[str, dict[str, float]] = {nid: {} for nid in node_ids}
for edge in edges:
source = edge.get("source")
target = edge.get("target")
if source not in node_set or target not in node_set or source == target:
continue
try:
weight = float(edge.get("weight", 1.0))
except (TypeError, ValueError):
weight = 1.0
if weight <= 0:
continue
adjacency[source][target] = adjacency[source].get(target, 0.0) + weight
adjacency[target][source] = adjacency[target].get(source, 0.0) + weight
node_strength = {nid: sum(neighbours.values()) for nid, neighbours in adjacency.items()}
total_weight = sum(node_strength.values()) / 2.0
return adjacency, node_strength, total_weight
class _DeterministicShuffler:
"""LCG-based shuffler so community detection stays reproducible."""
def __init__(self, seed: int):
self.state = seed & 0xFFFFFFFF or 1
def _next(self) -> int:
self.state = (1103515245 * self.state + 12345) & 0x7FFFFFFF
return self.state
def shuffle(self, items: list) -> None:
for i in range(len(items) - 1, 0, -1):
j = self._next() % (i + 1)
items[i], items[j] = items[j], items[i]
def _louvain_pass(
adjacency: dict[str, dict[str, float]],
node_strength: dict[str, float],
total_weight: float,
resolution: float,
seed: int,
) -> dict[str, str]:
if total_weight <= 0:
return {nid: nid for nid in adjacency}
community = {nid: nid for nid in adjacency}
community_strength: dict[str, float] = dict(node_strength)
two_m = 2.0 * total_weight
rng = _DeterministicShuffler(seed)
nodes_in_order = list(adjacency.keys())
improved = True
iterations = 0
max_iterations = 20
while improved and iterations < max_iterations:
improved = False
iterations += 1
rng.shuffle(nodes_in_order)
for nid in nodes_in_order:
current = community[nid]
ki = node_strength[nid]
if ki <= 0:
continue
community_links: dict[str, float] = {}
for neighbour, weight in adjacency[nid].items():
if neighbour == nid:
continue
community_links[community[neighbour]] = (
community_links.get(community[neighbour], 0.0) + weight
)
community_strength[current] -= ki
best = current
best_gain = 0.0
for candidate, ki_in_candidate in community_links.items():
sigma_tot = community_strength.get(candidate, 0.0)
gain = (ki_in_candidate / total_weight) - resolution * (sigma_tot * ki) / (two_m * total_weight)
if gain > best_gain + 1e-12:
best_gain = gain
best = candidate
if best_gain <= 0:
best = current
community_strength[best] = community_strength.get(best, 0.0) + ki
if best != current:
community[nid] = best
improved = True
return community
def detect_communities(
nodes: list[dict],
edges: list[dict],
*,
resolution: float = 1.0,
seed: int = 42,
max_levels: int = 5,
) -> dict[str, str]:
"""
Louvain-style community detection over the code graph.
Returns node_id -> community_id mapping. Pure-python; for very large graphs
(>50k edges), prefer a native Leiden implementation.
"""
adjacency, node_strength, total_weight = _undirected_adjacency(nodes, edges)
if total_weight <= 0:
return {nid: nid for nid in adjacency}
membership = _louvain_pass(adjacency, node_strength, total_weight, resolution, seed)
for level in range(max_levels):
super_adjacency: dict[str, dict[str, float]] = {}
super_strength: dict[str, float] = {}
for nid, neighbours in adjacency.items():
community = membership[nid]
super_adjacency.setdefault(community, {})
super_strength[community] = super_strength.get(community, 0.0)
for neighbour, weight in neighbours.items():
neighbour_community = membership[neighbour]
super_adjacency[community][neighbour_community] = (
super_adjacency[community].get(neighbour_community, 0.0) + weight
)
super_strength[community] += weight
super_strength = {c: total / 2.0 + total / 2.0 for c, total in super_strength.items()}
super_membership = _louvain_pass(
super_adjacency,
super_strength,
total_weight,
resolution,
seed + level + 1,
)
if all(super_membership[c] == c for c in super_adjacency):
break
membership = {nid: super_membership[membership[nid]] for nid in adjacency}
adjacency = super_adjacency
node_strength = super_strength
counts: dict[str, int] = {}
for community in membership.values():
counts[community] = counts.get(community, 0) + 1
ordered = sorted(counts.items(), key=lambda item: (-item[1], item[0]))
label_map = {c: f"community_{i}" for i, (c, _n) in enumerate(ordered)}
return {nid: label_map[c] for nid, c in membership.items()}
def compute_modularity(edges: list[dict], membership: dict[str, str]) -> float:
if not membership:
return 0.0
strength: dict[str, float] = {}
intra: dict[str, float] = {}
total_weight = 0.0
for edge in edges:
source = edge.get("source")
target = edge.get("target")
if source is None or target is None or source == target:
continue
if source not in membership or target not in membership:
continue
try:
weight = float(edge.get("weight", 1.0))
except (TypeError, ValueError):
weight = 1.0
if weight <= 0:
continue
strength[source] = strength.get(source, 0.0) + weight
strength[target] = strength.get(target, 0.0) + weight
total_weight += weight
if membership[source] == membership[target]:
intra[membership[source]] = intra.get(membership[source], 0.0) + 2 * weight
if total_weight <= 0:
return 0.0
two_m = 2.0 * total_weight
community_strength: dict[str, float] = {}
for nid, community in membership.items():
community_strength[community] = community_strength.get(community, 0.0) + strength.get(nid, 0.0)
modularity = 0.0
for community, intra_weight in intra.items():
sigma_tot = community_strength.get(community, 0.0)
modularity += (intra_weight / two_m) - (sigma_tot / two_m) ** 2
return modularity
def summarize_communities(nodes: list[dict], membership: dict[str, str], top: int) -> list[dict]:
grouped: dict[str, list[dict]] = {}
for node in nodes:
nid = node.get("id")
if not nid or nid not in membership:
continue
grouped.setdefault(membership[nid], []).append(node)
summaries = []
for community, members in sorted(grouped.items(), key=lambda item: (-len(item[1]), item[0])):
types: dict[str, int] = {}
files: dict[str, int] = {}
sample_labels = []
for node in members:
type_key = node.get("type", "?")
types[type_key] = types.get(type_key, 0) + 1
path_key = node.get("path") or "?"
files[path_key] = files.get(path_key, 0) + 1
if len(sample_labels) < 5:
sample_labels.append(node.get("label") or node.get("id"))
summaries.append({
"community_id": community,
"size": len(members),
"type_breakdown": dict(sorted(types.items(), key=lambda item: -item[1])),
"top_files": dict(sorted(files.items(), key=lambda item: -item[1])[:5]),
"sample_labels": sample_labels,
})
return summaries if top == 0 else summaries[:top]
def rank_nodes(nodes: list[dict], edges: list[dict], filter_type: str | None, top: int) -> list[dict]:
incoming = defaultdict(float)
for edge in edges:
incoming[edge["target"]] += float(edge.get("weight", 1.0))
ranked = []
for node in nodes:
if filter_type and node.get("type") != filter_type:
continue
ranked.append(
{
"id": node["id"],
"type": node.get("type"),
"label": node.get("label"),
"importance": round(incoming.get(node["id"], 0.0), 3),
"path": node.get("path"),
}
)
ranked.sort(key=lambda item: (-item["importance"], item["label"]))
return ranked if top == 0 else ranked[:top]
def format_json(payload: dict) -> str:
return json.dumps(payload, indent=2)
def format_table(payload: dict) -> str:
if "communities" in payload and payload.get("query") == "communities":
summaries = payload["communities"]
modularity = payload.get("modularity", 0.0)
count = payload.get("community_count", len(summaries))
if not summaries:
return f"No communities (modularity={modularity}, count={count})."
lines = [
f"# Communities (modularity={modularity}, count={count})",
"",
"community_id | size | top_types | sample_labels",
"--- | --- | --- | ---",
]
for entry in summaries:
top_types = ", ".join(f"{k}:{v}" for k, v in entry.get("type_breakdown", {}).items())
samples = ", ".join(entry.get("sample_labels", [])[:3])
lines.append(f"{entry['community_id']} | {entry['size']} | {top_types} | {samples}")
return "\n".join(lines)
rows = payload.get("results") or payload.get("nodes") or payload.get("paths") or []
if not rows:
return "No results."
if isinstance(rows[0], list):
return "\n".join(" -> ".join(item) for item in rows)
headers = sorted({key for row in rows for key in row.keys()})
lines = [" | ".join(headers), " | ".join("---" for _ in headers)]
for row in rows:
lines.append(" | ".join(str(row.get(header, "")) for header in headers))
return "\n".join(lines)
def mermaid_safe_id(value: str) -> str:
cleaned = re.sub(r"[^A-Za-z0-9_]", "_", value)
if not cleaned:
return "node"
if cleaned[0].isdigit():
return f"n_{cleaned}"
return cleaned
def mermaid_escape(value: str) -> str:
return value.replace('"', '\\"')
def format_mermaid(payload: dict, direction: str) -> str:
nodes = payload.get("nodes", [])
edges = payload.get("edges", [])
lines = [f"flowchart {direction}"]
for node in nodes:
node_id = mermaid_safe_id(node["id"])
label = mermaid_escape(node.get("label", node["id"]))
lines.append(f' {node_id}["{label}"]')
for edge in edges:
source = mermaid_safe_id(edge["source"])
target = mermaid_safe_id(edge["target"])
label = mermaid_escape(edge.get("relation", ""))
lines.append(f" {source} -->|{label}| {target}")
for node_type, style in MERMAID_TYPE_STYLES.items():
class_name = mermaid_safe_id(f"class_{node_type}")
lines.append(f" classDef {class_name} {style}")
for node in nodes:
if node.get("type") == node_type:
lines.append(f" class {mermaid_safe_id(node['id'])} {class_name}")
return "\n".join(lines)
def main() -> int:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("graph", help="Path to code-graph.json")
parser.add_argument("--node", help="Neighborhood around a node")
parser.add_argument("--from", dest="from_id", help="Path search source node")
parser.add_argument("--to", help="Path search target node")
parser.add_argument("--type", dest="node_type", help="List all nodes of a given type")
parser.add_argument("--impact", help="Downstream impact from a node")
parser.add_argument("--rank", action="store_true", help="Rank nodes by weighted incoming edges")
parser.add_argument("--ppr", action="store_true", help="Personalized PageRank from one or more --seed nodes")
parser.add_argument("--seed", action="append", default=[], help="Seed node id for --ppr (repeatable)")
parser.add_argument("--alpha", type=float, default=0.15, help="Teleport probability for --ppr (default: 0.15)")
parser.add_argument("--include-seeds", action="store_true", help="Include seed nodes in --ppr results")
parser.add_argument("--search", help="Lexical search query")
parser.add_argument("--diagram", action="store_true", help="Export full graph or filtered subgraph as Mermaid")
parser.add_argument("--articulation-points", action="store_true", help="List articulation points in the selected graph")
parser.add_argument("--bridges", action="store_true", help="List bridge edges in the selected graph")
parser.add_argument("--cycles", action="store_true", help="List directed cycles grouped by relation")
parser.add_argument("--topo-sort", metavar="RELATION", help="Topologically sort the selected relation if it is acyclic")
parser.add_argument("--communities", action="store_true", help="Detect modules via Louvain communities")
parser.add_argument("--resolution", type=float, default=1.0, help="Resolution γ for --communities (default: 1.0; >1 favours smaller, <1 favours larger)")
parser.add_argument("--community-seed", type=int, default=42, help="Deterministic seed for --communities (default: 42)")
parser.add_argument("--hops", type=int, default=1, help="Hop depth for neighborhood and impact")
parser.add_argument("--max-hops", type=int, default=3, help="Maximum path-finding depth")
parser.add_argument("--k", type=int, default=1, help="Number of node-disjoint shortest paths for --from/--to")
parser.add_argument("--top", type=int, default=20, help="Top rows for --rank")
parser.add_argument("--filter-type", help="Optional node type filter for --rank")
parser.add_argument("--types", help="Comma-separated node types for --search")
parser.add_argument("--relations", help="Comma-separated relations for structural graph queries")
parser.add_argument("--limit", type=int, default=20, help="Result limit for --search")
parser.add_argument("--format", choices=("json", "table", "mermaid"), default="json")
parser.add_argument("--output", help="Optional output file")
parser.add_argument("--mermaid-direction", choices=MERMAID_DIRECTIONS, default="LR")
args = parser.parse_args()
nodes, edges, node_index = load_graph(args.graph)
forward, backward = build_adjacency(edges)
if sum(bool(value) for value in (args.node, args.from_id, args.node_type, args.impact, args.rank, args.ppr, args.search, args.diagram, args.articulation_points, args.bridges, args.cycles, args.topo_sort, args.communities)) != 1:
print("Error: choose exactly one query mode", file=sys.stderr)
return 1
result: dict
if args.node:
node_ids, edge_rows = bfs_neighborhood(args.node, forward, backward, args.hops)
result = {"query": "node", "nodes": [node_index[node_id] for node_id in sorted(node_ids) if node_id in node_index], "edges": edge_rows}
elif args.impact:
node_ids, edge_rows = bfs_impact(args.impact, forward, args.hops)
result = {"query": "impact", "nodes": [node_index[node_id] for node_id in sorted(node_ids) if node_id in node_index], "edges": edge_rows}
elif args.from_id:
if not args.to:
print("Error: --from requires --to", file=sys.stderr)
return 1
result = {"query": "path", "paths": k_node_disjoint_paths(args.from_id, args.to, forward, args.max_hops, args.k)}
elif args.node_type:
result = {"query": "type", "results": [node for node in nodes if node.get("type") == args.node_type]}
elif args.rank:
result = {"query": "rank", "results": rank_nodes(nodes, edges, args.filter_type, args.top)}
elif args.ppr:
if not args.seed:
print("Error: --ppr requires at least one --seed", file=sys.stderr)
return 1
if not 0 < args.alpha < 1:
print("Error: --alpha must be between 0 and 1", file=sys.stderr)
return 1
unknown_seeds = [seed for seed in args.seed if seed not in node_index]
if unknown_seeds:
print(f"Error: unknown seed(s): {', '.join(unknown_seeds)}", file=sys.stderr)
return 1
scores = personalized_pagerank(nodes, edges, args.seed, alpha=args.alpha)
ranked = []
seed_set = set(args.seed)
allowed_types = {value.strip() for value in args.filter_type.split(",")} if args.filter_type else None
for node in nodes:
nid = node.get("id")
if not nid:
continue
if not args.include_seeds and nid in seed_set:
continue
if allowed_types and node.get("type") not in allowed_types:
continue
ranked.append({
"id": nid,
"type": node.get("type"),
"label": node.get("label"),
"path": node.get("path"),
"ppr_score": round(scores.get(nid, 0.0), 10),
})
ranked.sort(key=lambda item: (-item["ppr_score"], item.get("type") or "", item.get("label") or "", item["id"]))
if args.top > 0:
ranked = ranked[:args.top]
result = {
"query": "ppr",
"seeds": args.seed,
"alpha": args.alpha,
"include_seeds": args.include_seeds,
"results": ranked,
}
elif args.search:
requested_types = {value.strip() for value in args.types.split(",")} if args.types else None
result = {"query": "search", "results": query_search(nodes, args.search, requested_types, args.limit)}
elif args.articulation_points:
selected_edges = filtered_edges(edges, relation_set(args.relations))
rows = articulation_points(selected_edges, args.top)
for row in rows:
node = node_index.get(row["node_id"], {})
row["type"] = node.get("type")
row["label"] = node.get("label")
row["path"] = node.get("path")
result = {"query": "articulation_points", "results": rows}
elif args.bridges:
selected_edges = filtered_edges(edges, relation_set(args.relations))
result = {"query": "bridges", "results": bridge_edges(selected_edges, args.top)}
elif args.cycles:
result = {"query": "cycles", "results": cycles_by_relation(edges, relation_set(args.relations))}
elif args.communities:
if args.resolution <= 0:
print("Error: --resolution must be > 0", file=sys.stderr)
return 1
membership = detect_communities(
nodes,
edges,
resolution=args.resolution,
seed=args.community_seed,
)
modularity = compute_modularity(edges, membership)
summaries = summarize_communities(nodes, membership, args.top)
result = {
"query": "communities",
"resolution": args.resolution,
"seed": args.community_seed,
"modularity": round(modularity, 6),
"community_count": len({c for c in membership.values()}),
"communities": summaries,
}
elif args.topo_sort:
ordered, cycles = topological_sort(edges, args.topo_sort)
if cycles:
result = {"query": "topo_sort", "relation": args.topo_sort, "error": "cycles_detected", "results": cycles}
else:
result = {"query": "topo_sort", "relation": args.topo_sort, "results": [{"order": index + 1, "node_id": node_id, "label": node_index.get(node_id, {}).get("label")} for index, node_id in enumerate(ordered)]}
else:
result = {"query": "diagram", "nodes": nodes, "edges": edges}
if args.format == "json":
rendered = format_json(result)
elif args.format == "table":
rendered = format_table(result)
else:
if "nodes" not in result:
print("Error: Mermaid output requires nodes and edges", file=sys.stderr)
return 1
rendered = format_mermaid(result, args.mermaid_direction)
if args.output:
output_path = Path(args.output).expanduser().resolve()
output_path.parent.mkdir(parents=True, exist_ok=True)
output_path.write_text(rendered, encoding="utf-8")
else:
print(rendered)
return 0
if __name__ == "__main__":
raise SystemExit(main())
#!/usr/bin/env python3
"""Extract a deterministic code profile from a single repository."""
from __future__ import annotations
import argparse
import ast
import json
import re
from datetime import datetime, timezone
from pathlib import Path
SKIP_DIRS = {
".git",
".hg",
".svn",
".build",
".swiftpm",
"node_modules",
"dist",
"build",
".next",
"vendor",
".archive",
".venv",
"venv",
"__pycache__",
"coverage",
".turbo",
"DerivedData",
"Pods",
"Carthage",
"SourcePackages",
"xcuserdata",
}
LANGUAGE_BY_SUFFIX = {
".py": "Python",
".js": "JavaScript",
".jsx": "JavaScript",
".ts": "TypeScript",
".tsx": "TSX",
".cs": "C#",
".swift": "Swift",
}
CALL_KEYWORDS = {
"if",
"for",
"while",
"switch",
"catch",
"return",
"new",
"typeof",
"await",
"nameof",
"assert",
"guard",
"defer",
"init",
}
IMPORT_RE = re.compile(r'^\s*import\s+.*?from\s+[\'"]([^\'"]+)[\'"]', re.MULTILINE)
REQUIRE_RE = re.compile(r'require\([\'"]([^\'"]+)[\'"]\)')
USING_RE = re.compile(r'^\s*using\s+([A-Za-z0-9_.]+)\s*;', re.MULTILINE)
JS_CLASS_RE = re.compile(r'\bclass\s+([A-Za-z_][A-Za-z0-9_]*)\b(?:\s+extends\s+([A-Za-z_][A-Za-z0-9_]*))?')
CS_CLASS_RE = re.compile(r'\bclass\s+([A-Za-z_][A-Za-z0-9_]*)\b(?:\s*:\s*([A-Za-z_][A-Za-z0-9_]*))?')
SWIFT_IMPORT_RE = re.compile(r'^\s*import\s+([A-Za-z0-9_.]+)', re.MULTILINE)
SWIFT_TYPE_RE = re.compile(
r'^\s*(?:@\w+(?:\([^)]*\))?\s*)*'
r'(?:(?:public|private|fileprivate|internal|open|final|indirect|nonisolated)\s+)*'
r'(class|struct|actor|enum|protocol)\s+([A-Za-z_][A-Za-z0-9_]*)\b'
r'(?:\s*:\s*([^{]+))?',
re.MULTILINE,
)
JS_FUNCTION_RE = re.compile(r'^\s*(?:export\s+)?(?:async\s+)?function\s+([A-Za-z_][A-Za-z0-9_]*)\s*\(', re.MULTILINE)
JS_ARROW_RE = re.compile(r'^\s*(?:export\s+)?(?:const|let|var)\s+([A-Za-z_][A-Za-z0-9_]*)\s*=\s*(?:async\s*)?\([^)]*\)\s*=>', re.MULTILINE)
JS_METHOD_RE = re.compile(r'^\s*(?:public\s+|private\s+|protected\s+|static\s+|async\s+)*(?:get\s+|set\s+)?([A-Za-z_][A-Za-z0-9_]*)\s*\([^;]*\)\s*\{', re.MULTILINE)
SWIFT_FUNCTION_RE = re.compile(
r'^\s*(?:@\w+(?:\([^)]*\))?\s*)*'
r'(?:(?:public|private|fileprivate|internal|open|final|override|mutating|nonmutating|static|class|convenience|required)\s+)*'
r'func\s+([A-Za-z_][A-Za-z0-9_]*)\s*\(',
re.MULTILINE,
)
CS_METHOD_RE = re.compile(
r'^\s*(?:\[.*\]\s*)*(?:(?:public|private|protected|internal|static|async|virtual|override|sealed|partial|new|extern)\s+)+'
r'[\w<>\[\],?.]+\s+([A-Za-z_][A-Za-z0-9_]*)\s*\([^;]*\)\s*\{',
re.MULTILINE,
)
TEST_CASE_RE = re.compile(r'^\s*(?:test|it|describe)\s*\(\s*[\'"]([^\'"]+)[\'"]')
CALL_RE = re.compile(r'([A-Za-z_][A-Za-z0-9_]*)\s*\(')
ATTR_CALL_RE = re.compile(r'\.([A-Za-z_][A-Za-z0-9_]*)\s*\(')
def now_iso() -> str:
return datetime.now(timezone.utc).isoformat()
def normalize_id(text: str) -> str:
return re.sub(r"[^a-z0-9._-]", "-", text.lower().strip()).strip("-")
def file_node_id(repo_id: str, rel_path: Path) -> str:
return f"{repo_id}#file#{normalize_id(rel_path.as_posix())}"
def symbol_node_id(parent_id: str, symbol_type: str, label: str) -> str:
return f"{parent_id}#{symbol_type}#{normalize_id(label)}"
def external_node_id(kind: str, label: str) -> str:
return f"external#{kind}#{normalize_id(label)}"
def detect_file_kind(rel_path: Path) -> str:
lowered = rel_path.as_posix().lower()
parts = {part.lower() for part in rel_path.parts}
if "__tests__" in parts or "tests" in parts or "specs" in parts:
return "test"
if lowered.endswith((".test.js", ".test.ts", ".test.tsx", ".spec.js", ".spec.ts", ".spec.tsx", "_test.py", "_test.cs")):
return "test"
if rel_path.name.lower().startswith("test_"):
return "test"
return "source"
def should_skip(path: Path) -> bool:
return any(part in SKIP_DIRS for part in path.parts)
def iter_code_files(repo_root: Path) -> list[Path]:
files: list[Path] = []
for path in repo_root.rglob("*"):
if not path.is_file():
continue
if should_skip(path.relative_to(repo_root)):
continue
if path.suffix.lower() in LANGUAGE_BY_SUFFIX:
files.append(path)
return sorted(files)
def add_relation(relations: list[dict], seen: set[tuple[str, str, str]], source: str, target: str, relation: str, group: str, confidence: float, notes: str | None = None) -> None:
key = (source, target, relation)
if key in seen:
return
seen.add(key)
payload = {
"source": source,
"target": target,
"relation": relation,
"group": group,
"confidence": round(confidence, 3),
}
if notes:
payload["notes"] = notes
relations.append(payload)
def is_test_name(name: str) -> bool:
lowered = name.lower()
return lowered.startswith("test") or lowered.endswith("test")
def resolve_relative_import(repo_root: Path, current_file: Path, raw_target: str) -> Path | None:
if not raw_target.startswith("."):
return None
base = current_file.parent
remainder = raw_target
while remainder.startswith("../"):
base = base.parent
remainder = remainder[3:]
if remainder.startswith("./"):
remainder = remainder[2:]
candidate_base = (base / remainder).resolve()
candidates = [
candidate_base,
candidate_base.with_suffix(".ts"),
candidate_base.with_suffix(".tsx"),
candidate_base.with_suffix(".js"),
candidate_base.with_suffix(".jsx"),
candidate_base.with_suffix(".py"),
candidate_base.with_suffix(".cs"),
candidate_base / "index.ts",
candidate_base / "index.tsx",
candidate_base / "index.js",
candidate_base / "index.jsx",
candidate_base / "__init__.py",
]
for candidate in candidates:
if candidate.exists() and candidate.is_file():
return candidate
return None
class PythonAnalyzer(ast.NodeVisitor):
def __init__(self, repo_root: Path, repo_id: str, file_path: Path, rel_path: Path, file_id: str, file_kind: str) -> None:
self.repo_root = repo_root
self.repo_id = repo_id
self.file_path = file_path
self.rel_path = rel_path
self.file_id = file_id
self.file_kind = file_kind
self.symbols: list[dict] = []
self.relations: list[dict] = []
self.relation_seen: set[tuple[str, str, str]] = set()
self.symbol_stack: list[tuple[str, str]] = []
self.class_stack: list[str] = []
self.pending_calls: list[tuple[str, str, bool]] = []
self.pending_inherits: list[tuple[str, str]] = []
self.label_index: dict[str, list[str]] = {}
def add_symbol(self, symbol_type: str, label: str, parent_id: str, node: ast.AST) -> str:
symbol_id = symbol_node_id(parent_id, symbol_type, label)
line_end = getattr(node, "end_lineno", getattr(node, "lineno", None))
entry = {
"id": symbol_id,
"type": symbol_type,
"label": label,
"path": self.rel_path.as_posix(),
"language": "Python",
"parent_id": parent_id,
"line_start": getattr(node, "lineno", None),
"line_end": line_end,
"confidence": 0.97,
}
self.symbols.append(entry)
self.label_index.setdefault(label, []).append(symbol_id)
add_relation(self.relations, self.relation_seen, parent_id, symbol_id, "defines", "structural", 0.98)
return symbol_id
def current_parent(self) -> str:
if self.class_stack:
return self.class_stack[-1]
return self.file_id
def current_symbol(self) -> tuple[str, str] | None:
if not self.symbol_stack:
return None
return self.symbol_stack[-1]
def visit_Import(self, node: ast.Import) -> None:
for alias in node.names:
target = self.resolve_import(alias.name, 0)
add_relation(self.relations, self.relation_seen, self.file_id, target, "imports", "dependency", 0.92)
def visit_ImportFrom(self, node: ast.ImportFrom) -> None:
module = node.module or ""
target = self.resolve_import(module, node.level)
add_relation(self.relations, self.relation_seen, self.file_id, target, "imports", "dependency", 0.9)
def visit_ClassDef(self, node: ast.ClassDef) -> None:
class_id = self.add_symbol("class", node.name, self.file_id, node)
self.class_stack.append(class_id)
self.symbol_stack.append((class_id, "class"))
for base in node.bases:
base_name = self.extract_name(base)
if base_name:
self.pending_inherits.append((class_id, base_name))
self.generic_visit(node)
self.symbol_stack.pop()
self.class_stack.pop()
def visit_FunctionDef(self, node: ast.FunctionDef) -> None:
self._visit_callable(node)
def visit_AsyncFunctionDef(self, node: ast.AsyncFunctionDef) -> None:
self._visit_callable(node)
def _visit_callable(self, node: ast.FunctionDef | ast.AsyncFunctionDef) -> None:
symbol_type = "method" if self.class_stack else "function"
if self.file_kind == "test" and is_test_name(node.name):
symbol_type = "test"
parent_id = self.current_parent()
symbol_id = self.add_symbol(symbol_type, node.name, parent_id, node)
self.symbol_stack.append((symbol_id, symbol_type))
self.generic_visit(node)
self.symbol_stack.pop()
def visit_Call(self, node: ast.Call) -> None:
current = self.current_symbol()
if current is not None:
symbol_id, symbol_type = current
call_name = self.extract_name(node.func)
if call_name:
self.pending_calls.append((symbol_id, call_name, symbol_type == "test"))
self.generic_visit(node)
def resolve_import(self, module: str, level: int) -> str:
if level > 0:
base = self.file_path.parent
for _ in range(level - 1):
base = base.parent
candidate_base = base
if module:
candidate_base = candidate_base.joinpath(*module.split("."))
for candidate in (
candidate_base.with_suffix(".py"),
candidate_base / "__init__.py",
):
if candidate.exists():
return file_node_id(self.repo_id, candidate.relative_to(self.repo_root))
else:
candidate_base = self.repo_root.joinpath(*module.split(".")) if module else self.repo_root
for candidate in (
candidate_base.with_suffix(".py"),
candidate_base / "__init__.py",
):
if candidate.exists():
return file_node_id(self.repo_id, candidate.relative_to(self.repo_root))
label = module or "relative-import"
return external_node_id("import", label)
@staticmethod
def extract_name(node: ast.AST) -> str | None:
if isinstance(node, ast.Name):
return node.id
if isinstance(node, ast.Attribute):
return node.attr
return None
def finalize(self) -> tuple[list[dict], list[dict]]:
for source_id, target_label, is_test_caller in self.pending_calls:
resolved_target = self.resolve_symbol_label(target_label, "call")
relation = "tests" if is_test_caller else "calls"
group = "behavioral" if is_test_caller else "behavioral"
confidence = 0.82 if relation == "calls" else 0.72
add_relation(self.relations, self.relation_seen, source_id, resolved_target, relation, group, confidence)
for class_id, base_label in self.pending_inherits:
resolved_target = self.resolve_symbol_label(base_label, "inherit")
add_relation(self.relations, self.relation_seen, class_id, resolved_target, "inherits", "semantic", 0.8)
return self.symbols, self.relations
def resolve_symbol_label(self, label: str, kind: str) -> str:
matches = self.label_index.get(label, [])
if len(matches) == 1:
return matches[0]
return external_node_id(kind, label)
def extract_brace_calls(line: str) -> list[str]:
names = set(ATTR_CALL_RE.findall(line))
names.update(CALL_RE.findall(line))
cleaned = []
for name in names:
if name in CALL_KEYWORDS:
continue
cleaned.append(name)
return sorted(set(cleaned))
def parse_swift_inheritance_targets(raw_clause: str | None) -> list[str]:
if not raw_clause:
return []
targets: list[str] = []
for candidate in raw_clause.split(","):
cleaned = candidate.strip()
if not cleaned:
continue
cleaned = cleaned.split("where", 1)[0].strip()
cleaned = cleaned.split("<", 1)[0].strip()
cleaned = cleaned.split("(", 1)[0].strip()
cleaned = cleaned.split(":", 1)[0].strip()
if not cleaned:
continue
tokens = re.findall(r"[A-Za-z_][A-Za-z0-9_]*", cleaned)
if tokens:
targets.append(tokens[-1])
return targets
def analyze_heuristic_file(repo_root: Path, repo_id: str, file_path: Path, rel_path: Path, file_id: str, language: str, file_kind: str) -> tuple[str, list[dict], list[dict]]:
content = file_path.read_text(encoding="utf-8", errors="ignore")
symbols: list[dict] = []
relations: list[dict] = []
relation_seen: set[tuple[str, str, str]] = set()
label_index: dict[str, list[str]] = {}
pending_calls: list[tuple[str, str, str]] = []
pending_inherits: list[tuple[str, str]] = []
file_kind = detect_file_kind(rel_path)
lines = content.splitlines()
brace_depth = 0
class_stack: list[tuple[str, int]] = []
symbol_context: list[tuple[str, str, int]] = []
def add_symbol(symbol_type: str, label: str, parent_id: str, line_no: int, confidence: float) -> str:
symbol_id = symbol_node_id(parent_id, symbol_type, label)
entry = {
"id": symbol_id,
"type": symbol_type,
"label": label,
"path": rel_path.as_posix(),
"language": language,
"parent_id": parent_id,
"line_start": line_no,
"line_end": line_no,
"confidence": confidence,
}
if not any(existing["id"] == symbol_id for existing in symbols):
symbols.append(entry)
label_index.setdefault(label, []).append(symbol_id)
add_relation(relations, relation_seen, parent_id, symbol_id, "defines", "structural", min(confidence + 0.1, 0.95))
return symbol_id
for line_no, line in enumerate(lines, start=1):
stripped = line.strip()
open_count = line.count("{")
close_count = line.count("}")
if not stripped:
brace_depth += open_count - close_count
while symbol_context and brace_depth < symbol_context[-1][2]:
symbol_context.pop()
while class_stack and brace_depth < class_stack[-1][1]:
class_stack.pop()
continue
if language in {"JavaScript", "TypeScript", "TSX"}:
for target in IMPORT_RE.findall(line):
resolved = resolve_relative_import(repo_root, file_path, target)
target_id = file_node_id(repo_id, resolved.relative_to(repo_root)) if resolved else external_node_id("import", target)
add_relation(relations, relation_seen, file_id, target_id, "imports", "dependency", 0.78)
for target in REQUIRE_RE.findall(line):
resolved = resolve_relative_import(repo_root, file_path, target)
target_id = file_node_id(repo_id, resolved.relative_to(repo_root)) if resolved else external_node_id("import", target)
add_relation(relations, relation_seen, file_id, target_id, "imports", "dependency", 0.72)
if language == "C#":
for target in USING_RE.findall(line):
add_relation(relations, relation_seen, file_id, external_node_id("import", target), "imports", "dependency", 0.7)
elif language == "Swift":
for target in SWIFT_IMPORT_RE.findall(line):
add_relation(relations, relation_seen, file_id, external_node_id("import", target), "imports", "dependency", 0.74)
if language in {"JavaScript", "TypeScript", "TSX"}:
class_match = JS_CLASS_RE.search(line)
elif language == "C#":
class_match = CS_CLASS_RE.search(line)
else:
class_match = SWIFT_TYPE_RE.search(line) if language == "Swift" else None
if class_match:
class_name = class_match.group(2) if language == "Swift" else class_match.group(1)
parent_id = file_id
class_id = add_symbol("class", class_name, parent_id, line_no, 0.66)
if language == "Swift":
for base_name in parse_swift_inheritance_targets(class_match.group(3)):
pending_inherits.append((class_id, base_name))
else:
base_name = class_match.group(2)
if base_name:
pending_inherits.append((class_id, base_name))
function_match = None
if language in {"JavaScript", "TypeScript", "TSX"}:
function_match = JS_FUNCTION_RE.match(line) or JS_ARROW_RE.match(line) or JS_METHOD_RE.match(line)
elif language == "C#":
function_match = CS_METHOD_RE.match(line)
elif language == "Swift":
function_match = SWIFT_FUNCTION_RE.match(line)
test_case_match = TEST_CASE_RE.match(line) if file_kind == "test" and language in {"JavaScript", "TypeScript", "TSX"} else None
if function_match:
label = function_match.group(1)
class_parent = class_stack[-1][0] if class_stack else None
symbol_type = "method" if class_parent else "function"
if file_kind == "test" and is_test_name(label):
symbol_type = "test"
parent_id = class_parent or file_id
symbol_id = add_symbol(symbol_type, label, parent_id, line_no, 0.6 if symbol_type == "function" else 0.58)
symbol_context.append((symbol_id, symbol_type, brace_depth + max(open_count - close_count, 1)))
elif test_case_match:
label = test_case_match.group(1) or f"test-{line_no}"
symbol_id = add_symbol("test", label, file_id, line_no, 0.56)
symbol_context.append((symbol_id, "test", brace_depth + max(open_count - close_count, 1)))
active_symbol = symbol_context[-1] if symbol_context else None
if active_symbol:
source_id, symbol_type, _ = active_symbol
relation_name = "tests" if symbol_type == "test" else "calls"
for target_name in extract_brace_calls(line):
pending_calls.append((source_id, target_name, relation_name))
post_depth = brace_depth + open_count - close_count
if class_match and open_count > close_count:
class_label = class_match.group(2) if language == "Swift" else class_match.group(1)
class_id = symbol_node_id(file_id, "class", class_label)
class_stack.append((class_id, max(post_depth, brace_depth + 1)))
brace_depth = post_depth
while symbol_context and brace_depth < symbol_context[-1][2]:
symbol_context.pop()
while class_stack and brace_depth < class_stack[-1][1]:
class_stack.pop()
for source_id, target_name, relation_name in pending_calls:
matches = label_index.get(target_name, [])
target_id = matches[0] if len(matches) == 1 else external_node_id("call", target_name)
confidence = 0.58 if relation_name == "calls" else 0.52
add_relation(relations, relation_seen, source_id, target_id, relation_name, "behavioral", confidence)
for class_id, base_name in pending_inherits:
matches = label_index.get(base_name, [])
target_id = matches[0] if len(matches) == 1 else external_node_id("inherit", base_name)
add_relation(relations, relation_seen, class_id, target_id, "inherits", "semantic", 0.62)
return "heuristic", symbols, relations
def analyze_python_file(repo_root: Path, repo_id: str, file_path: Path, rel_path: Path, file_id: str, file_kind: str) -> tuple[str, list[dict], list[dict]]:
try:
tree = ast.parse(file_path.read_text(encoding="utf-8", errors="ignore"))
except SyntaxError:
return "error", [], []
analyzer = PythonAnalyzer(repo_root, repo_id, file_path, rel_path, file_id, file_kind)
analyzer.visit(tree)
symbols, relations = analyzer.finalize()
return "parsed", symbols, relations
def scan_repo(repo_root: Path) -> dict:
repo_root = repo_root.expanduser().resolve()
repo_id = normalize_id(repo_root.name)
files: list[dict] = []
symbols: list[dict] = []
relations: list[dict] = []
evidence: list[dict] = []
languages: list[str] = []
relation_seen: set[tuple[str, str, str]] = set()
for file_path in iter_code_files(repo_root):
rel_path = file_path.relative_to(repo_root)
language = LANGUAGE_BY_SUFFIX[file_path.suffix.lower()]
if language not in languages:
languages.append(language)
file_kind = detect_file_kind(rel_path)
file_id = file_node_id(repo_id, rel_path)
parse_status = "unsupported"
extracted_symbols: list[dict] = []
extracted_relations: list[dict] = []
if language == "Python":
parse_status, extracted_symbols, extracted_relations = analyze_python_file(repo_root, repo_id, file_path, rel_path, file_id, file_kind)
elif language in {"JavaScript", "TypeScript", "TSX", "C#", "Swift"}:
parse_status, extracted_symbols, extracted_relations = analyze_heuristic_file(repo_root, repo_id, file_path, rel_path, file_id, language, file_kind)
file_confidence = {
"parsed": 0.96,
"heuristic": 0.62,
"unsupported": 0.35,
"error": 0.25,
"skipped": 0.1,
}[parse_status]
files.append(
{
"id": file_id,
"path": rel_path.as_posix(),
"language": language,
"kind": file_kind,
"parse_status": parse_status,
"confidence": file_confidence,
}
)
if len(evidence) < 25:
evidence.append({"path": rel_path.as_posix(), "reason": f"{parse_status} code file"})
add_relation(relations, relation_seen, repo_id, file_id, "contains", "structural", 1.0)
for symbol in extracted_symbols:
if not any(existing["id"] == symbol["id"] for existing in symbols):
symbols.append(symbol)
for relation in extracted_relations:
add_relation(
relations,
relation_seen,
relation["source"],
relation["target"],
relation["relation"],
relation["group"],
relation["confidence"],
relation.get("notes"),
)
parsed_files = sum(1 for entry in files if entry["parse_status"] == "parsed")
heuristic_files = sum(1 for entry in files if entry["parse_status"] == "heuristic")
summary = (
f"{repo_root.name} has {len(files)} code file(s), {len(symbols)} symbol node(s), and "
f"{len(relations)} relation(s); {parsed_files} file(s) parsed strongly and {heuristic_files} file(s) parsed heuristically."
)
supported_files = [entry for entry in files if entry["parse_status"] in {"parsed", "heuristic"}]
confidence_score = round(sum(entry["confidence"] for entry in supported_files) / max(len(files), 1), 3) if files else 0.0
return {
"repo_id": repo_id,
"repo_name": repo_root.name,
"repo_path": str(repo_root),
"languages": languages,
"files": files,
"symbols": symbols,
"relations": relations,
"summary": summary,
"evidence": evidence,
"confidence_score": confidence_score,
"last_scanned_at": now_iso(),
}
def main() -> int:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("repo", help="Path to the repository root")
args = parser.parse_args()
print(json.dumps(scan_repo(Path(args.repo)), indent=2))
return 0
if __name__ == "__main__":
raise SystemExit(main())
#!/usr/bin/env python3
"""Regression tests for the dev-context-code-graph scripts."""
from __future__ import annotations
import importlib.util
import json
import tempfile
import unittest
from pathlib import Path
SCRIPT_DIR = Path(__file__).resolve().parent
def load_module(filename: str, module_name: str):
spec = importlib.util.spec_from_file_location(module_name, SCRIPT_DIR / filename)
module = importlib.util.module_from_spec(spec)
assert spec.loader is not None
spec.loader.exec_module(module)
return module
scan_code_repo = load_module("scan_code_repo.py", "dev_context_code_graph_scan")
build_code_graph = load_module("build_code_graph.py", "dev_context_code_graph_build")
validate_code_graph = load_module("validate_code_graph.py", "dev_context_code_graph_validate")
query_code_graph = load_module("query_code_graph.py", "dev_context_code_graph_query")
export_code_graph_report = load_module("export_code_graph_report.py", "dev_context_code_graph_report")
class CodeGraphRegressionTests(unittest.TestCase):
def test_python_repo_scan_build_validate_and_report(self) -> None:
with tempfile.TemporaryDirectory() as tmpdir:
repo = Path(tmpdir) / "py-repo"
repo.mkdir()
(repo / "app.py").write_text(
"def helper():\n return 1\n\n"
"class Service:\n def run(self):\n return helper()\n",
encoding="utf-8",
)
(repo / "test_app.py").write_text(
"from app import helper\n\n"
"def test_helper():\n assert helper() == 1\n",
encoding="utf-8",
)
profile = scan_code_repo.scan_repo(repo)
self.assertEqual(profile["repo_id"], "py-repo")
self.assertGreaterEqual(len(profile["files"]), 2)
self.assertTrue(any(symbol["type"] == "function" and symbol["label"] == "helper" for symbol in profile["symbols"]))
self.assertTrue(any(relation["relation"] == "tests" for relation in profile["relations"]))
profiles_dir = Path(tmpdir) / "code-profiles"
profiles_dir.mkdir()
(profiles_dir / "py-repo.json").write_text(json.dumps(profile, indent=2), encoding="utf-8")
graph = build_code_graph.GraphBuilder()
build_code_graph.build_from_profiles(profiles_dir, graph)
graph_payload = graph.to_dict()
self.assertEqual(graph_payload["meta"]["repo_count"], 1)
self.assertTrue(any(node["type"] == "file" for node in graph_payload["nodes"]))
self.assertTrue(any(edge["relation"] == "calls" for edge in graph_payload["edges"]))
report = validate_code_graph.validate_graph(graph_payload, 90)
self.assertEqual(report["issues_total"], 0)
graph_path = Path(tmpdir) / "graphs" / "code-graph.json"
graph_path.parent.mkdir()
graph_path.write_text(json.dumps(graph_payload, indent=2), encoding="utf-8")
markdown = export_code_graph_report.render_markdown(graph_payload, None)
self.assertIn("Code Graph Report", markdown)
def test_typescript_repo_links_relative_imports(self) -> None:
with tempfile.TemporaryDirectory() as tmpdir:
repo = Path(tmpdir) / "ts-repo"
(repo / "src").mkdir(parents=True)
(repo / "src" / "math.ts").write_text("export function sum(a: number, b: number) { return a + b; }\n", encoding="utf-8")
(repo / "src" / "app.ts").write_text(
"import { sum } from './math'\n"
"export function runApp() { return sum(1, 2) }\n",
encoding="utf-8",
)
(repo / "src" / "app.test.ts").write_text(
"import { runApp } from './app'\n"
"test('runs', () => { runApp() })\n",
encoding="utf-8",
)
profile = scan_code_repo.scan_repo(repo)
imports = [relation for relation in profile["relations"] if relation["relation"] == "imports"]
self.assertTrue(imports)
self.assertTrue(any(relation["relation"] == "tests" for relation in profile["relations"]))
self.assertTrue(any("math.ts" in file_entry["path"] for file_entry in profile["files"]))
self.assertTrue(any(symbol["label"] == "runApp" for symbol in profile["symbols"]))
def test_query_search_rank_and_path(self) -> None:
nodes = [
{"id": "repo#file#a", "type": "file", "label": "a.py"},
{"id": "repo#file#a#function#helper", "type": "function", "label": "helper", "parent_id": "repo#file#a"},
{"id": "repo#file#test#test-helper", "type": "test", "label": "test_helper", "parent_id": "repo#file#test"},
]
edges = [
{"source": "repo#file#a", "target": "repo#file#a#function#helper", "relation": "defines", "group": "structural", "weight": 0.95},
{"source": "repo#file#test#test-helper", "target": "repo#file#a#function#helper", "relation": "tests", "group": "behavioral", "weight": 0.7},
]
results = query_code_graph.query_search(nodes, "helper", None, 10)
self.assertEqual(results[0]["label"], "helper")
ranked = query_code_graph.rank_nodes(nodes, edges, None, 10)
self.assertEqual(ranked[0]["label"], "helper")
forward, _backward = query_code_graph.build_adjacency(edges)
paths = query_code_graph.bfs_paths("repo#file#test#test-helper", "repo#file#a#function#helper", forward, 2)
self.assertEqual(paths[0][-1], "repo#file#a#function#helper")
def test_apply_fixes_adds_missing_parent_edge(self) -> None:
graph = {
"meta": {"generated_at": "2026-03-22T00:00:00+00:00", "version": "1.0", "graph_contract_version": "1.0", "build_source": "code_profiles", "node_count": 2, "edge_count": 0, "repo_count": 1},
"nodes": [
{"id": "repo", "type": "repo", "label": "repo"},
{"id": "repo#file#a", "type": "file", "label": "a.py", "parent_id": "repo", "parse_status": "parsed"},
],
"edges": [],
}
fixed = validate_code_graph.apply_fixes(graph)
self.assertTrue(any(edge["relation"] == "contains" for edge in fixed["edges"]))
if __name__ == "__main__":
unittest.main()