Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
72 changes: 59 additions & 13 deletions langgraph_engine/diagrams/legacy_generator.py
Original file line number Diff line number Diff line change
Expand Up @@ -1032,13 +1032,25 @@ def generate_component_diagram(self, dep_graph=None, groups=None):
# Tier 2: AST + LLM hybrid
# ------------------------------------------------------------------

def generate_sequence_diagram(self, call_chains=None, context=""):
def generate_sequence_diagram(self, call_chains=None, entry_function=""):
"""Generate Mermaid sequenceDiagram from call chains.

When CallGraph is available, uses class-aware participants:
- Class names become participants (not raw function names)
- Method calls show Class.method() notation
- Call paths are followed for proper sequencing

Args:
entry_function: When given, filters call_chains to only the
calls transitively reachable from this function (BFS on
caller->callee), instead of rendering the whole project.
This is a real filter, not an LLM-enrichment hint -
issue #312 found this parameter was previously passed
straight into an `_llm_enrich()` call under the name
`context`, silently replacing the deterministic AST
rendering with LLM output whenever it was non-empty,
despite this being documented and labelled a Tier 2 AST
tool at the MCP tool layer.
"""
if call_chains is None:
# Try CallGraph first; fall back to per-file extraction
Expand All @@ -1058,12 +1070,20 @@ def generate_sequence_diagram(self, call_chains=None, context=""):
if not call_chains:
return "sequenceDiagram\n Note over System: No call chains found"

if entry_function:
call_chains = self._filter_chains_from_entry(call_chains, entry_function)
if not call_chains:
return (
"sequenceDiagram\n Note over System: No calls found "
"reachable from entry_function '%s'" % entry_function
)

# Build class-aware participant mapping from FQN data
# If call_chains have caller_fqn/callee_fqn, use class context
has_fqn = any(c.get("caller_fqn") for c in call_chains)

if has_fqn:
return self._sequence_from_fqn_chains(call_chains, context)
return self._sequence_from_fqn_chains(call_chains)

# Legacy path: flat caller/callee names
lines = ["sequenceDiagram"]
Expand All @@ -1079,14 +1099,45 @@ def generate_sequence_diagram(self, call_chains=None, context=""):
if count >= 30:
break

if context:
enriched = self._llm_enrich("\n".join(lines), "sequence diagram", context)
if enriched:
return enriched

return "\n".join(lines)

def _sequence_from_fqn_chains(self, call_chains, context=""):
@staticmethod
def _filter_chains_from_entry(call_chains, entry_function):
"""Keep only the call chains transitively reachable from entry_function.

BFS on caller->callee (matched by the flat, non-FQN name every
chain dict carries, per ast_analyzer.extract_call_chains's own
docstring) - a chain is included once its caller is known
reachable, which may in turn make its callee reachable too.

Args:
call_chains: List of chain dicts, each with at least
caller/callee keys.
entry_function: Name to trace the call chain from.

Returns:
Filtered list, in original relative order, or empty if
entry_function calls nothing in call_chains.
"""
reachable = {entry_function}
included_keys = set()
filtered = []
changed = True
while changed:
changed = False
for chain in call_chains:
key = (chain.get("caller"), chain.get("callee"), chain.get("line"))
if key in included_keys:
continue
if chain.get("caller") in reachable:
filtered.append(chain)
included_keys.add(key)
if chain.get("callee") not in reachable:
reachable.add(chain.get("callee"))
changed = True
return filtered

def _sequence_from_fqn_chains(self, call_chains):
"""Build class-aware sequence diagram from FQN call chains.

Uses caller_fqn/callee_fqn to extract class participants and
Expand Down Expand Up @@ -1174,11 +1225,6 @@ def _method_name(fqn):
if count >= 40:
break

if context:
enriched = self._llm_enrich("\n".join(lines), "sequence diagram", context)
if enriched:
return enriched

return "\n".join(lines)

def generate_activity_diagram(self, function_code="", context="", data=None):
Expand Down
12 changes: 6 additions & 6 deletions plugin/README.md
Original file line number Diff line number Diff line change
Expand Up @@ -29,12 +29,12 @@ writer become reachable only after the separate `register-mcp` command writes
user-scope MCP registrations, which the user runs once, by choice.
`unregister-mcp` reverses it.

`register-mcp` ships in this build. What it can register is bounded by what
exists: the progress writer is registrable today, and the version push gate is
catalogued but not yet built, so `register-mcp` reports that capability as
unavailable rather than writing an entry pointing at a file that does not exist.
Any command needing an unregistered capability says so rather than failing
quietly, and `doctor` reports the whole picture in one place.
`register-mcp` ships in this build. Both catalogued capabilities are built and
registrable today: the progress writer and the version push gate. If a
server's entry point is ever missing on disk, `register-mcp` reports that
capability as unavailable rather than writing an entry pointing at a file that
does not exist. Any command needing an unregistered capability says so rather
than failing quietly, and `doctor` reports the whole picture in one place.

## Layout

Expand Down
7 changes: 5 additions & 2 deletions plugin/scripts/snapshot_status.py
Original file line number Diff line number Diff line change
Expand Up @@ -155,7 +155,10 @@ def locate_library_root(explicit=None, search_from=None):

Args:
explicit: Caller-supplied path that wins when given.
search_from: Directory whose siblings are searched.
search_from: The plugin root, whose grandparent's siblings are
searched -- the plugin root's parent is the engine repository
root, and the library checkout sits beside that, not beside the
plugin subdirectory itself.

Returns:
Path or None.
Expand All @@ -167,7 +170,7 @@ def locate_library_root(explicit=None, search_from=None):
if from_env:
candidates.append(Path(from_env))
if search_from:
candidates.append(Path(search_from).parent / LIBRARY_DIR_NAME)
candidates.append(Path(search_from).parent.parent / LIBRARY_DIR_NAME)
for candidate in candidates:
if (candidate / "VERSION").is_file():
return candidate.resolve()
Expand Down
23 changes: 23 additions & 0 deletions tests/test_library_snapshot.py
Original file line number Diff line number Diff line change
Expand Up @@ -398,6 +398,29 @@ def test_detects_ahead(self, tmp_path):
assert outcome.status == ss.STATUS_AHEAD
assert outcome.fires is True

def test_auto_discovers_sibling_library_two_levels_up(self, tmp_path):
"""REGRESSION (#315): the no-flag auto-discovery path must find the
library beside the ENGINE repo, not beside the plugin subdirectory.

The real workspace layout is ``<workspace>/claude-workflow-engine/plugin``
and ``<workspace>/claude-global-library`` as siblings of the engine repo.
Every other test in this class passes ``library_root=`` explicitly and so
never exercises ``locate_library_root``'s ``search_from`` branch at all;
this is the one test that does, with a layout that mirrors reality
instead of a flat tmp_path.
"""
workspace = tmp_path / "workspace"
engine_root = workspace / "claude-workflow-engine"
plugin_root = engine_root / "plugin"
library = _make_library(workspace / "claude-global-library", version="1.2.3")
_make_plugin(plugin_root, snapshot_from=library, snapshot_version="1.0.0")

outcome = ss.check_snapshot(plugin_root=plugin_root, library_root=None)

assert outcome.status == ss.STATUS_BEHIND
assert outcome.fires is True
assert "1.0.0" in outcome.detail and "1.2.3" in outcome.detail

@pytest.mark.parametrize(
"pinned,live,expected",
[
Expand Down
64 changes: 64 additions & 0 deletions tests/test_uml_generators.py
Original file line number Diff line number Diff line change
Expand Up @@ -906,3 +906,67 @@ def test_sequence_diagram_from_call_graph(self, tmp_path):
# CallGraph build failed; verify fallback still works
syntax = gen.generate_sequence_diagram()
assert "sequenceDiagram" in syntax


# ==================================================================
# TestSequenceDiagramEntryFunctionFilter
# ==================================================================


class TestSequenceDiagramEntryFunctionFilter:
"""Regression coverage for #312: entry_function must be a real AST
filter, never an LLM-enrichment hint. Previously this parameter was
passed straight into `_llm_enrich()` under the name `context`,
silently replacing the deterministic rendering with LLM output
whenever it was non-empty - despite this being a labelled Tier 2
AST tool.
"""

LEGACY_CHAINS = [
{"caller": "main", "callee": "handler", "file": "app.py", "line": 1},
{"caller": "handler", "callee": "validate", "file": "app.py", "line": 2},
{"caller": "handler", "callee": "save", "file": "app.py", "line": 3},
# Unrelated branch, must NOT appear when filtering from "handler"
{"caller": "unrelated_entry", "callee": "cleanup", "file": "app.py", "line": 4},
]

def test_entry_function_never_reaches_an_llm_call(self, tmp_project, monkeypatch):
"""A non-empty entry_function must not trigger any LLM path."""

class LLMCalled(AssertionError):
pass

def _boom(*args, **kwargs):
raise LLMCalled("entry_function filtering reached an LLM call: args=%r kwargs=%r" % (args, kwargs))

monkeypatch.setattr(UMLDiagramGenerator, "_llm_enrich", _boom)

gen = UMLDiagramGenerator(str(tmp_project))
syntax = gen.generate_sequence_diagram(call_chains=self.LEGACY_CHAINS, entry_function="handler")
assert "sequenceDiagram" in syntax

def test_entry_function_filters_to_reachable_calls_only(self, tmp_project):
"""Only calls transitively reachable from entry_function are rendered."""
gen = UMLDiagramGenerator(str(tmp_project))
syntax = gen.generate_sequence_diagram(call_chains=self.LEGACY_CHAINS, entry_function="handler")
assert "validate" in syntax
assert "save" in syntax
assert "unrelated_entry" not in syntax
assert "cleanup" not in syntax

def test_entry_function_matching_nothing_says_so_explicitly(self, tmp_project):
"""A named entry_function with no matching caller gets its own
message, distinct from the generic 'no call chains found' stub -
so a caller can tell 'wrong function name' from 'nothing scanned'.
"""
gen = UMLDiagramGenerator(str(tmp_project))
syntax = gen.generate_sequence_diagram(call_chains=self.LEGACY_CHAINS, entry_function="does_not_exist")
assert "does_not_exist" in syntax
assert "No calls found" in syntax

def test_empty_entry_function_renders_everything_unfiltered(self, tmp_project):
"""No entry_function (the default) must not filter anything."""
gen = UMLDiagramGenerator(str(tmp_project))
syntax = gen.generate_sequence_diagram(call_chains=self.LEGACY_CHAINS)
assert "unrelated_entry" in syntax
assert "cleanup" in syntax
Loading