From 165372f4ce53b4092e2e318b523ff3f7cec02cab Mon Sep 17 00:00:00 2001 From: Divyam Talwar Date: Mon, 24 Aug 2026 01:47:49 +0530 Subject: [PATCH 1/2] test(review): make retrospective prompt assertion deterministic --- tests/review/test_review_commands.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/tests/review/test_review_commands.py b/tests/review/test_review_commands.py index 823fcfe..1d980c1 100644 --- a/tests/review/test_review_commands.py +++ b/tests/review/test_review_commands.py @@ -775,9 +775,9 @@ def test_do_run_batches_dry_run_generates_packet_and_prompts( assert len(packet_files) == 1 blind_packet = tmp_path / ".structorium" / "review_packet_blind.json" assert blind_packet.exists() - prompt_files = list(runs_dir.glob("*/prompts/batch-*.md")) + prompt_files = sorted(runs_dir.glob("*/prompts/batch-*.md")) assert len(prompt_files) == 2 - prompt_text = prompt_files[0].read_text() + prompt_text = "\n".join(path.read_text() for path in prompt_files) assert "Blind packet:" in prompt_text assert str(blind_packet) in prompt_text assert "Previously flagged issues" in prompt_text From d01a3e575bc6ad88dafc4910e071935dcee2741e Mon Sep 17 00:00:00 2001 From: Divyam Talwar Date: Mon, 24 Aug 2026 01:47:59 +0530 Subject: [PATCH 2/2] feat(graph): explain bounded dependency impact paths --- README.md | 6 +- app/cli_support/parser.py | 5 +- app/cli_support/parser_groups.py | 26 +++ app/commands/impact_cmd.py | 72 +++++++++ app/commands/registry.py | 2 + docs/IMPACT_EXPLORER.md | 38 +++++ engine/impact.py | 253 ++++++++++++++++++++++++++++++ tests/commands/test_impact_cmd.py | 68 ++++++++ tests/engine/test_impact.py | 76 +++++++++ 9 files changed, 544 insertions(+), 2 deletions(-) create mode 100644 app/commands/impact_cmd.py create mode 100644 docs/IMPACT_EXPLORER.md create mode 100644 engine/impact.py create mode 100644 tests/commands/test_impact_cmd.py create mode 100644 tests/engine/test_impact.py diff --git a/README.md b/README.md index 1ddd742..aa2b972 100644 --- a/README.md +++ b/README.md @@ -1904,6 +1904,11 @@ def evaluate_gate(findings, changed_ranges, policy): ## 🔗 Temporal Coupling +For a source-level blast-radius explanation, run `structorium impact `. +The bounded explorer reports shortest-path witnesses for dependents and dependencies, +and can emit JSON for agents or Mermaid for review discussions. See the +[dependency impact explorer guide](docs/IMPACT_EXPLORER.md). + Temporal coupling is computed from git history: ``` @@ -2172,4 +2177,3 @@ copies or substantial portions of the Software.

↑ Back to top

- diff --git a/app/cli_support/parser.py b/app/cli_support/parser.py index 2f6b2ec..3848fc3 100644 --- a/app/cli_support/parser.py +++ b/app/cli_support/parser.py @@ -3,7 +3,8 @@ from __future__ import annotations import argparse -from importlib.metadata import PackageNotFoundError, version as get_version +from importlib.metadata import PackageNotFoundError +from importlib.metadata import version as get_version from app.cli_support.parser_groups import ( _add_config_parser, @@ -12,6 +13,7 @@ _add_exclude_parser, _add_fix_parser, _add_ignore_parser, + _add_impact_parser, _add_langs_parser, _add_move_parser, _add_next_parser, @@ -118,6 +120,7 @@ def create_parser(*, langs: list[str], detector_names: list[str]) -> argparse.Ar ) _add_scan_parser(sub) _add_status_parser(sub) + _add_impact_parser(sub) _add_tree_parser(sub) _add_show_parser(sub) _add_next_parser(sub) diff --git a/app/cli_support/parser_groups.py b/app/cli_support/parser_groups.py index d1705b5..ecc2e65 100644 --- a/app/cli_support/parser_groups.py +++ b/app/cli_support/parser_groups.py @@ -25,6 +25,7 @@ "_add_exclude_parser", "_add_fix_parser", "_add_ignore_parser", + "_add_impact_parser", "_add_langs_parser", "_add_move_parser", "_add_next_parser", @@ -40,6 +41,31 @@ ] +def _add_impact_parser(sub) -> None: + p_impact = sub.add_parser( + "impact", help="Explain bounded dependency blast radius for files or directories" + ) + p_impact.add_argument("targets", nargs="+", help="File or directory paths to explore") + p_impact.add_argument("--path", type=str, default=".", help="Project root directory") + p_impact.add_argument("--state", type=str, default=None, help="Path to state file") + p_impact.add_argument( + "--direction", + choices=["dependents", "dependencies", "both"], + default="both", + help="Graph direction to explore (default: both)", + ) + p_impact.add_argument("--depth", type=int, default=3, help="Maximum traversal depth") + p_impact.add_argument( + "--max-nodes", type=int, default=200, help="Hard impacted-node budget" + ) + p_impact.add_argument( + "--format", choices=["text", "json", "mermaid"], default="text" + ) + p_impact.add_argument( + "--output", type=str, default=None, help="Write evidence to this path" + ) + + def _add_scan_parser(sub) -> None: p_scan = sub.add_parser( "scan", diff --git a/app/commands/impact_cmd.py b/app/commands/impact_cmd.py new file mode 100644 index 0000000..b5f637d --- /dev/null +++ b/app/commands/impact_cmd.py @@ -0,0 +1,72 @@ +"""CLI command for bounded dependency blast-radius exploration.""" + +from __future__ import annotations + +import argparse +import sys +from pathlib import Path + +from app.commands.helpers.lang import resolve_lang, resolve_lang_settings +from app.commands.helpers.runtime import command_runtime +from core.discovery_api import safe_write_text +from core.output_api import colorize +from engine.impact import ( + analyze_impact, + render_impact_json, + render_impact_mermaid, + render_impact_text, +) +from languages import runtime as lang_runtime + + +def cmd_impact(args: argparse.Namespace) -> None: + """Build the language dependency graph and explain a bounded change radius.""" + lang = resolve_lang(args) + if lang is None or not lang.build_dep_graph: + print( + colorize("No dependency-graph language integration is available.", "red"), + file=sys.stderr, + ) + raise SystemExit(2) + + project_root = Path(getattr(args, "path", None) or ".").resolve() + runtime = command_runtime(args) + lang_run = lang_runtime.make_lang_run( + lang, + overrides=lang_runtime.LangRunOverrides( + runtime_settings=resolve_lang_settings(runtime.config, lang) + ), + ) + try: + graph = lang_run.build_dep_graph(project_root) + except (OSError, UnicodeDecodeError, ValueError, TypeError, RuntimeError) as exc: + print( + colorize(f"Could not build dependency graph: {exc}", "red"), + file=sys.stderr, + ) + raise SystemExit(2) from exc + + report = analyze_impact( + graph, + list(args.targets), + project_root=project_root, + direction=args.direction, + max_depth=args.depth, + max_nodes=args.max_nodes, + ) + renderers = { + "text": render_impact_text, + "json": render_impact_json, + "mermaid": render_impact_mermaid, + } + rendered = renderers[args.format](report) + if args.output: + output = Path(args.output) + output.parent.mkdir(parents=True, exist_ok=True) + safe_write_text(output, rendered) + print(colorize(f"Wrote dependency impact to {output.resolve()}", "green")) + else: + print(rendered, end="") + + +__all__ = ["cmd_impact"] diff --git a/app/commands/registry.py b/app/commands/registry.py index 2e4d8e0..011dd49 100644 --- a/app/commands/registry.py +++ b/app/commands/registry.py @@ -17,6 +17,7 @@ def _build_handlers() -> dict[str, CommandHandler]: from app.commands.dev_cmd import cmd_dev from app.commands.exclude_cmd import cmd_exclude from app.commands.fix import cmd_fix + from app.commands.impact_cmd import cmd_impact from app.commands.langs import cmd_langs from app.commands.move import cmd_move from app.commands.next import cmd_next @@ -38,6 +39,7 @@ def _build_handlers() -> dict[str, CommandHandler]: "ignore": cmd_ignore_pattern, "exclude": cmd_exclude, "fix": cmd_fix, + "impact": cmd_impact, "plan": cmd_plan, "detect": cmd_detect, "tree": cmd_tree, diff --git a/docs/IMPACT_EXPLORER.md b/docs/IMPACT_EXPLORER.md new file mode 100644 index 0000000..f5eb163 --- /dev/null +++ b/docs/IMPACT_EXPLORER.md @@ -0,0 +1,38 @@ +# Dependency impact explorer + +Before changing a shared module, ask Structorium which modules depend on it, which +dependencies it reaches, and the shortest path that proves each relationship: + +```bash +structorium impact src/domain.py --direction dependents --depth 4 +structorium impact src/domain.py --format json --output artifacts/impact.json +structorium impact src/domain.py --format mermaid --output artifacts/impact.mmd +``` + +Directory targets expand to every dependency-graph node below that prefix. Searches +are deterministic breadth-first traversals, so every result includes a shortest-path +witness. `--depth` defaults to 3 and `--max-nodes` defaults to 200; the report marks +itself as truncated instead of silently exploring an unbounded monorepo graph. + +Directions use explicit names: + +- `dependents`: files that could be affected when the target changes. +- `dependencies`: files the target itself relies on. +- `both`: both views within the same shared node budget. + +JSON is intended for agents and CI artifacts. Mermaid is intended for PR descriptions +and architecture discussions. Both describe original source-to-dependency edges; the +visual renderer never invents relationships. + +## Research provenance + +The feature is original Structorium code informed by public code-graph workflows: + +- [dependency-cruiser CLI reporters](https://github.com/sverweij/dependency-cruiser/blob/main/doc/cli.md) + provide graph outputs including Mermaid for repository-native review. +- [dependency-cruiser folder graphs](https://github.com/sverweij/dependency-cruiser/blob/main/doc/faq.md#folder-level-dependency-graph-ddot-reporter) + demonstrate why large dependency graphs need focused, summarized views. +- [CodeQL path explanations](https://codeql.github.com/docs/writing-codeql-queries/creating-path-queries/) + establish evidence paths as a useful way to explain why a result is reachable. + +No competitor code or runtime dependency is used. diff --git a/engine/impact.py b/engine/impact.py new file mode 100644 index 0000000..d2bd4d8 --- /dev/null +++ b/engine/impact.py @@ -0,0 +1,253 @@ +"""Bounded, deterministic dependency-impact exploration algorithms.""" + +from __future__ import annotations + +import hashlib +import json +from collections import deque +from collections.abc import Mapping +from pathlib import Path, PurePosixPath +from typing import Any, Literal, TypedDict + +ImpactDirection = Literal["dependents", "dependencies", "both"] + + +class ImpactEntry(TypedDict): + path: str + direction: Literal["dependents", "dependencies"] + distance: int + witness: list[str] + + +class ImpactEdge(TypedDict): + source: str + target: str + + +class ImpactReport(TypedDict): + seeds: list[str] + direction: ImpactDirection + max_depth: int + max_nodes: int + truncated: bool + entries: list[ImpactEntry] + edges: list[ImpactEdge] + node_count: int + + +def _normalize_path(value: object, project_root: Path) -> str: + raw = str(value or ".").replace("\\", "/") + candidate = Path(raw) + if candidate.is_absolute(): + try: + raw = str(candidate.resolve().relative_to(project_root.resolve())) + except (OSError, ValueError): + raw = candidate.as_posix() + normalized = PurePosixPath(raw).as_posix() + while normalized.startswith("./"): + normalized = normalized[2:] + return normalized.rstrip("/") or "." + + +def normalize_graph( + graph: Mapping[str, Mapping[str, Any]], project_root: Path +) -> dict[str, set[str]]: + """Normalize heterogeneous language graphs into source -> dependency edges.""" + normalized: dict[str, set[str]] = {} + for raw_source, raw_entry in graph.items(): + source = _normalize_path(raw_source, project_root) + normalized.setdefault(source, set()) + imports = raw_entry.get("imports", set()) + if not isinstance(imports, (set, list, tuple, frozenset)): + continue + for raw_target in imports: + target = _normalize_path(raw_target, project_root) + normalized[source].add(target) + normalized.setdefault(target, set()) + return normalized + + +def resolve_seeds(nodes: set[str], targets: list[str], project_root: Path) -> list[str]: + """Resolve exact file or directory-prefix targets against normalized graph nodes.""" + seeds: set[str] = set() + for target in targets: + normalized = _normalize_path(target, project_root) + if normalized in nodes: + seeds.add(normalized) + continue + prefix = normalized.rstrip("/") + "/" + seeds.update(node for node in nodes if node.startswith(prefix)) + return sorted(seeds) + + +def _reverse_graph(graph: Mapping[str, set[str]]) -> dict[str, set[str]]: + reverse = {node: set() for node in graph} + for source, dependencies in graph.items(): + for dependency in dependencies: + reverse.setdefault(dependency, set()).add(source) + return reverse + + +def _walk( + adjacency: Mapping[str, set[str]], + seeds: list[str], + *, + direction: Literal["dependents", "dependencies"], + max_depth: int, + budget: int, +) -> tuple[list[ImpactEntry], bool]: + witnesses = {seed: [seed] for seed in seeds} + distances = {seed: 0 for seed in seeds} + queue = deque(seeds) + entries: list[ImpactEntry] = [] + + while queue: + current = queue.popleft() + distance = distances[current] + if distance >= max_depth: + continue + for neighbor in sorted(adjacency.get(current, set())): + if neighbor in distances: + continue + if len(entries) >= budget: + return entries, True + distances[neighbor] = distance + 1 + witnesses[neighbor] = [*witnesses[current], neighbor] + entries.append( + { + "path": neighbor, + "direction": direction, + "distance": distance + 1, + "witness": witnesses[neighbor], + } + ) + queue.append(neighbor) + return entries, False + + +def analyze_impact( + graph: Mapping[str, Mapping[str, Any]], + targets: list[str], + *, + project_root: Path, + direction: ImpactDirection = "both", + max_depth: int = 3, + max_nodes: int = 200, +) -> ImpactReport: + """Explore dependency blast radius with shortest-path witnesses and hard bounds.""" + if max_depth < 1: + raise ValueError("max_depth must be at least 1") + if max_nodes < 1: + raise ValueError("max_nodes must be at least 1") + if direction not in {"dependents", "dependencies", "both"}: + raise ValueError(f"unsupported impact direction: {direction}") + + normalized = normalize_graph(graph, project_root) + seeds = resolve_seeds(set(normalized), targets, project_root) + if not seeds: + raise ValueError(f"no dependency graph nodes matched: {', '.join(targets)}") + seed_truncated = len(seeds) > max_nodes + seeds = seeds[:max_nodes] + + entries: list[ImpactEntry] = [] + truncated = seed_truncated + directions: list[Literal["dependents", "dependencies"]] = ( + ["dependents", "dependencies"] if direction == "both" else [direction] + ) + reverse = _reverse_graph(normalized) + for selected in directions: + remaining = max_nodes - len(seeds) - len(entries) + if remaining <= 0: + truncated = True + break + adjacency = reverse if selected == "dependents" else normalized + found, did_truncate = _walk( + adjacency, + seeds, + direction=selected, + max_depth=max_depth, + budget=remaining, + ) + entries.extend(found) + truncated = truncated or did_truncate + + entries.sort(key=lambda item: (item["distance"], item["direction"], item["path"])) + visible = {*seeds, *(item["path"] for item in entries)} + edges = [ + {"source": source, "target": target} + for source in sorted(visible) + for target in sorted(normalized.get(source, set())) + if target in visible + ] + return { + "seeds": seeds, + "direction": direction, + "max_depth": max_depth, + "max_nodes": max_nodes, + "truncated": truncated, + "entries": entries, + "edges": edges, + "node_count": len(visible), + } + + +def render_impact_text(report: ImpactReport) -> str: + """Render a concise operator-facing impact report.""" + lines = [ + "Dependency impact", + f"Seeds: {', '.join(report['seeds'])}", + f"Visible nodes: {report['node_count']} (depth <= {report['max_depth']})", + ] + for entry in report["entries"]: + witness = " -> ".join(entry["witness"]) + lines.append( + f"[{entry['direction']} d={entry['distance']}] {entry['path']} via {witness}" + ) + if report["truncated"]: + lines.append(f"TRUNCATED at {report['max_nodes']} impacted nodes") + return "\n".join(lines) + "\n" + + +def render_impact_json(report: ImpactReport) -> str: + """Render stable machine-readable impact evidence.""" + return json.dumps(report, indent=2, sort_keys=True) + "\n" + + +def _mermaid_id(path: str) -> str: + return "n" + hashlib.sha1(path.encode("utf-8"), usedforsecurity=False).hexdigest()[:12] + + +def _mermaid_label(path: str) -> str: + return path.replace('"', "'") + + +def render_impact_mermaid(report: ImpactReport) -> str: + """Render a review-ready Mermaid dependency graph.""" + nodes = sorted( + {*report["seeds"], *(entry["path"] for entry in report["entries"])} + ) + lines = ["flowchart LR"] + for node in nodes: + lines.append(f' {_mermaid_id(node)}["{_mermaid_label(node)}"]') + for edge in report["edges"]: + lines.append(f" {_mermaid_id(edge['source'])} --> {_mermaid_id(edge['target'])}") + for seed in report["seeds"]: + lines.append(f" class {_mermaid_id(seed)} seed") + lines.append(" classDef seed fill:#ffd166,stroke:#111,stroke-width:3px") + if report["truncated"]: + lines.append(f" %% truncated at {report['max_nodes']} impacted nodes") + return "\n".join(lines) + "\n" + + +__all__ = [ + "ImpactDirection", + "ImpactEdge", + "ImpactEntry", + "ImpactReport", + "analyze_impact", + "normalize_graph", + "render_impact_json", + "render_impact_mermaid", + "render_impact_text", + "resolve_seeds", +] diff --git a/tests/commands/test_impact_cmd.py b/tests/commands/test_impact_cmd.py new file mode 100644 index 0000000..0a79e88 --- /dev/null +++ b/tests/commands/test_impact_cmd.py @@ -0,0 +1,68 @@ +"""CLI surface tests for dependency-impact exploration.""" + +from __future__ import annotations + +from argparse import Namespace +from pathlib import Path + +from app.commands.helpers.runtime import CommandRuntime +from app.commands.impact_cmd import cmd_impact +from cli import create_parser + + +def test_parser_accepts_bounded_impact_options() -> None: + args = create_parser().parse_args( + [ + "impact", + "src/domain.py", + "--direction", + "dependents", + "--depth", + "4", + "--max-nodes", + "25", + "--format", + "json", + ] + ) + assert args.command == "impact" + assert args.targets == ["src/domain.py"] + assert args.depth == 4 + assert args.max_nodes == 25 + + +def test_command_writes_json_with_mock_language(tmp_path: Path, monkeypatch) -> None: + class FakeRun: + @staticmethod + def build_dep_graph(_path): + return { + "src/api.py": {"imports": {"src/domain.py"}}, + "src/domain.py": {"imports": set()}, + } + + class FakeLang: + name = "fake" + build_dep_graph = True + + @staticmethod + def normalize_settings(settings): + return settings + + monkeypatch.setattr("app.commands.impact_cmd.resolve_lang", lambda _args: FakeLang()) + monkeypatch.setattr( + "app.commands.impact_cmd.lang_runtime.make_lang_run", + lambda *_args, **_kwargs: FakeRun(), + ) + output = tmp_path / "impact.json" + args = Namespace( + targets=["src/domain.py"], + direction="dependents", + depth=3, + max_nodes=20, + format="json", + output=str(output), + path=str(tmp_path), + runtime=CommandRuntime(config={}, state={}, state_path=None), + ) + cmd_impact(args) + assert '"path": "src/api.py"' in output.read_text(encoding="utf-8") diff --git a/tests/engine/test_impact.py b/tests/engine/test_impact.py new file mode 100644 index 0000000..6b7c58a --- /dev/null +++ b/tests/engine/test_impact.py @@ -0,0 +1,76 @@ +"""Dependency-impact algorithm and evidence renderer tests.""" + +from __future__ import annotations + +import json +from pathlib import Path + +import pytest + +from engine.impact import ( + analyze_impact, + render_impact_json, + render_impact_mermaid, +) + + +def _graph() -> dict: + return { + "src/api.py": {"imports": {"src/service.py"}}, + "src/worker.py": {"imports": {"src/service.py"}}, + "src/service.py": {"imports": {"src/domain.py"}}, + "src/domain.py": {"imports": set()}, + "tests/test_domain.py": {"imports": {"src/domain.py"}}, + } + + +def test_dependents_include_shortest_path_witnesses(tmp_path: Path) -> None: + report = analyze_impact( + _graph(), ["src/domain.py"], project_root=tmp_path, direction="dependents" + ) + by_path = {entry["path"]: entry for entry in report["entries"]} + assert by_path["src/service.py"]["witness"] == ["src/domain.py", "src/service.py"] + assert by_path["src/api.py"]["witness"] == [ + "src/domain.py", + "src/service.py", + "src/api.py", + ] + + +def test_directory_target_expands_to_matching_graph_nodes(tmp_path: Path) -> None: + report = analyze_impact( + _graph(), ["src"], project_root=tmp_path, direction="dependencies", max_depth=1 + ) + assert report["seeds"] == [ + "src/api.py", + "src/domain.py", + "src/service.py", + "src/worker.py", + ] + + +def test_budget_is_hard_bounded_and_marked_truncated(tmp_path: Path) -> None: + report = analyze_impact( + _graph(), + ["src/domain.py"], + project_root=tmp_path, + direction="dependents", + max_nodes=2, + ) + assert len(report["entries"]) == 1 + assert report["node_count"] == 2 + assert report["truncated"] is True + + +def test_unknown_target_fails_closed(tmp_path: Path) -> None: + with pytest.raises(ValueError, match="no dependency graph nodes matched"): + analyze_impact(_graph(), ["missing"], project_root=tmp_path) + + +def test_json_and_mermaid_are_deterministic(tmp_path: Path) -> None: + report = analyze_impact(_graph(), ["src/domain.py"], project_root=tmp_path) + assert json.loads(render_impact_json(report))["seeds"] == ["src/domain.py"] + mermaid = render_impact_mermaid(report) + assert mermaid.startswith("flowchart LR\n") + assert "src/domain.py" in mermaid + assert "classDef seed" in mermaid