From cb82fb77cd70f167758619e5304fa2ccb8eb1d0f Mon Sep 17 00:00:00 2001 From: Kavish Date: Wed, 14 Oct 2026 19:11:30 -0700 Subject: [PATCH 1/3] fix(cli): keep graph edges attached to node names with a colon graphviz.Digraph.edge() splits an endpoint on ":" before quoting it, so an edge to a node or tool named "srv:tool" was written as `a -> srv:tool`. DOT reads that as node "srv" with port "tool", so the edge attached to a phantom node while the real node was drawn disconnected. Quote edge endpoints through a small Digraph subclass so the real name stays the DOT id. adk-web reads the SVG title as the node name, so renaming ids would break click handling. Output is unchanged for graphs without ":" in any name, including __START__ and __END__. Add the first tests for plot_workflow_graph, using format="dot" so no graphviz binary is needed. Fixes #7543 --- .../adk/cli/utils/graph_visualization.py | 19 ++- .../cli/utils/test_graph_visualization.py | 142 ++++++++++++++++++ 2 files changed, 160 insertions(+), 1 deletion(-) create mode 100644 tests/unittests/cli/utils/test_graph_visualization.py diff --git a/src/google/adk/cli/utils/graph_visualization.py b/src/google/adk/cli/utils/graph_visualization.py index ee53037cf96..1b2883d7996 100644 --- a/src/google/adk/cli/utils/graph_visualization.py +++ b/src/google/adk/cli/utils/graph_visualization.py @@ -21,10 +21,27 @@ from typing import cast import graphviz +from graphviz import quoting from ...workflow._node_status import NodeStatus +class _Digraph(graphviz.Digraph): # type: ignore[misc] # graphviz is untyped + """A Digraph whose edge endpoints are quoted as plain node ids. + + Digraph.edge() splits an endpoint on ":" before quoting it, treating the + rest as a port, so an edge to a node named "srv:tool" is written as + `a -> srv:tool` and attaches to a phantom node "srv". Quoting the whole + endpoint keeps the real name as the DOT id, which adk-web reads back from + the SVG title as the node name. No port syntax is used in these graphs. + + This overrides the private `_quote_edge` hook. Verified on graphviz 0.20.2 + and 0.21; a unit test fails if the hook is renamed or stops working. + """ + + _quote_edge = staticmethod(quoting.quote) + + def plot_workflow_graph( app_info: dict[str, Any], agent_state: dict[str, Any] | None = None, @@ -64,7 +81,7 @@ def _traverse_sub_agents( graph = {"nodes": nodes, "edges": edges} nodes_state = agent_state.get("nodes", {}) - dot = graphviz.Digraph(comment="Workflow Visualization") + dot = _Digraph(comment="Workflow Visualization") if dark_mode: graph_bgcolor = "#0F172A" diff --git a/tests/unittests/cli/utils/test_graph_visualization.py b/tests/unittests/cli/utils/test_graph_visualization.py new file mode 100644 index 00000000000..d4df10d6d20 --- /dev/null +++ b/tests/unittests/cli/utils/test_graph_visualization.py @@ -0,0 +1,142 @@ +# Copyright 2026 Google LLC +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Tests for plot_workflow_graph DOT output.""" + +import re +from typing import Any + +from google.adk.cli.utils.graph_visualization import _Digraph +from google.adk.cli.utils.graph_visualization import plot_workflow_graph +from google.adk.workflow._node_status import NodeStatus +import graphviz + +_TOKEN = r'"(?:[^"\\]|\\.)*"|[^\s"\[]+' +_NODE_LINE = re.compile(rf'^\t({_TOKEN}) \[', re.MULTILINE) +_EDGE_LINE = re.compile(rf'^\t({_TOKEN}) -> ({_TOKEN})', re.MULTILINE) + + +def _workflow_info( + nodes: list[dict[str, Any]], edges: list[tuple[str, str]] +) -> dict[str, Any]: + return { + 'root_agent': { + 'name': 'wf', + 'graph': { + 'nodes': nodes, + 'edges': [ + {'from_node': {'name': a}, 'to_node': {'name': b}} + for a, b in edges + ], + }, + } + } + + +def _plot(info: dict[str, Any], state: dict[str, Any] | None = None) -> str: + dot = plot_workflow_graph(info, state, format='dot') + assert isinstance(dot, str) + return dot + + +def _node_line(dot: str, name: str) -> str: + return next( + line for line in dot.splitlines() if f' {name}>' in line and '[' in line + ) + + +def test_edges_attach_to_nodes_whose_names_contain_colons(): + """Edges to colon-named nodes hit the real nodes, not phantom ports. + + The real name must stay the DOT id because adk-web reads the SVG title as + the node name for click handling. + """ + info = _workflow_info( + [ + {'name': '__START__'}, + { + 'name': 'a', + 'type': 'agent', + 'tools': [{'name': 'srv:tool', 'type': 'tool'}], + }, + {'name': 'step:two', 'type': 'function'}, + ], + [('__START__', 'a'), ('a', 'step:two')], + ) + + dot = _plot(info) + + declared = set(_NODE_LINE.findall(dot)) + endpoints = {end for edge in _EDGE_LINE.findall(dot) for end in edge} + assert {'"srv:tool"', '"step:two"'} <= declared + assert endpoints <= declared + assert not {'srv', 'step'} & endpoints + + +def test_digraph_edge_hook_quotes_colon_endpoints(): + """The graphviz hook that keeps colon names whole still exists and works.""" + assert hasattr(graphviz.Digraph, '_quote_edge'), ( + 'graphviz renamed Digraph._quote_edge; update _Digraph in' + ' graph_visualization.py' + ) + + dot = _Digraph() + dot.edge('a', 'srv:tool') + + assert 'a -> "srv:tool"' in dot.source + + +def test_start_and_end_nodes_keep_their_identifiers(): + """START and END keep the __START__ and __END__ identifiers.""" + info = _workflow_info( + [{'name': '__START__'}, {'name': 'a', 'type': 'agent'}], + [('__START__', 'a')], + ) + + dot = _plot(info) + + assert '\t__START__ [label=START' in dot + assert '\t__START__ -> a' in dot + assert '\t__END__ [label=END' in dot + assert '\ta -> __END__' in dot + + +def test_tools_are_injected_as_dashed_nodes_with_edges_from_their_agent(): + """A node's tools show up as dashed tool nodes linked from the node.""" + info = _workflow_info( + [{'name': 'a', 'type': 'agent', 'tools': [{'name': 'search'}]}], [] + ) + + dot = _plot(info) + + assert 'rounded,filled,dashed' in _node_line(dot, 'search') + assert '\ta -> search [' in dot + assert 'style=dashed' in dot + + +def test_completed_node_is_filled_with_completed_status_color(): + """A completed node is filled with the completed color, others are not.""" + info = _workflow_info( + [ + {'name': 'a', 'type': 'agent'}, + {'name': 'b', 'type': 'agent'}, + ], + [('a', 'b')], + ) + state = {'nodes': {'a': {'status': NodeStatus.COMPLETED.value}}} + + dot = _plot(info, state) + + assert 'fillcolor="#16A34A"' in _node_line(dot, 'a') + assert 'fillcolor="#16A34A"' not in _node_line(dot, 'b') From f2e36d088f4673fcb649cdee68f469f821c4b2ed Mon Sep 17 00:00:00 2001 From: Kavish Date: Thu, 15 Oct 2026 00:26:33 -0700 Subject: [PATCH 2/3] test(cli): cover colon node names as edge target, source and terminal Add tests for a workflow node named "b:x" in each edge position handled by plot_workflow_graph: as an edge target (a -> b:x), as an edge source (b:x -> c) and as a terminal node linked to END (b:x -> __END__). Each test asserts on the DOT source that every edge endpoint is a declared node, that no phantom "b" endpoint appears, and that the node is declared under its real name. --- .../cli/utils/test_graph_visualization.py | 52 +++++++++++++++++++ 1 file changed, 52 insertions(+) diff --git a/tests/unittests/cli/utils/test_graph_visualization.py b/tests/unittests/cli/utils/test_graph_visualization.py index d4df10d6d20..f5337861ae6 100644 --- a/tests/unittests/cli/utils/test_graph_visualization.py +++ b/tests/unittests/cli/utils/test_graph_visualization.py @@ -84,6 +84,58 @@ def test_edges_attach_to_nodes_whose_names_contain_colons(): assert not {'srv', 'step'} & endpoints +def _assert_colon_node_is_wired(dot: str, edge: str) -> None: + """Checks `edge` is present and the colon node "b:x" has no phantom port.""" + declared = set(_NODE_LINE.findall(dot)) + endpoints = {end for pair in _EDGE_LINE.findall(dot) for end in pair} + assert '"b:x"' in declared + assert endpoints <= declared + assert 'b' not in endpoints + assert f'\t{edge}' in dot + + +def test_workflow_node_with_colon_can_be_an_edge_target(): + """An edge into a node named "b:x" lands on the declared node.""" + info = _workflow_info( + [ + {'name': 'a', 'type': 'agent'}, + {'name': 'b:x', 'type': 'function'}, + ], + [('a', 'b:x')], + ) + + _assert_colon_node_is_wired(_plot(info), 'a -> "b:x"') + + +def test_workflow_node_with_colon_can_be_an_edge_source(): + """An edge out of a node named "b:x" starts at the declared node.""" + info = _workflow_info( + [ + {'name': 'b:x', 'type': 'function'}, + {'name': 'c', 'type': 'agent'}, + ], + [('b:x', 'c')], + ) + + _assert_colon_node_is_wired(_plot(info), '"b:x" -> c') + + +def test_workflow_terminal_node_with_colon_is_linked_to_end(): + """A terminal node named "b:x" gets its edge to END from the real node.""" + info = _workflow_info( + [ + {'name': 'a', 'type': 'agent'}, + {'name': 'b:x', 'type': 'function'}, + ], + [('a', 'b:x')], + ) + + dot = _plot(info) + + _assert_colon_node_is_wired(dot, '"b:x" -> __END__') + assert '\t__END__ [label=END' in dot + + def test_digraph_edge_hook_quotes_colon_endpoints(): """The graphviz hook that keeps colon names whole still exists and works.""" assert hasattr(graphviz.Digraph, '_quote_edge'), ( From b10c0245178a0c3c60397f427e36e3cfffc23713 Mon Sep 17 00:00:00 2001 From: Kavish Date: Thu, 15 Oct 2026 09:32:42 -0700 Subject: [PATCH 3/3] fix(cli): quote tool names with a colon in the agent graph get_agent_graph draws every edge through Digraph.edge(), which splits an endpoint on ":" before quoting it. An edge to a tool named "srv:tool" was written as `root -> srv:tool`, which DOT reads as node "srv" with port "tool", so the edge attached to a phantom node and the real tool node was drawn disconnected. This is the same bug fixed for plot_workflow_graph in this PR. Build the top-level graph and the cluster subgraphs for Sequential, Loop and Parallel agents from the _Digraph subclass added earlier, so edge endpoints are quoted whole and the real name stays the DOT id. Tool edges inside a cluster are drawn on the cluster's own Digraph, so it needs the subclass too. Output is unchanged for graphs without a colon in any name. Add tests for a colon-named tool on a plain agent and inside each cluster type. --- src/google/adk/cli/agent_graph.py | 7 +-- tests/unittests/cli/test_agent_graph.py | 64 +++++++++++++++++++++++++ 2 files changed, 68 insertions(+), 3 deletions(-) diff --git a/src/google/adk/cli/agent_graph.py b/src/google/adk/cli/agent_graph.py index cf6bcaf9fca..69112504ab3 100644 --- a/src/google/adk/cli/agent_graph.py +++ b/src/google/adk/cli/agent_graph.py @@ -28,6 +28,7 @@ from ..tools.agent_tool import AgentTool from ..tools.base_tool import BaseTool from ..tools.function_tool import FunctionTool +from .utils.graph_visualization import _Digraph if TYPE_CHECKING: from collections.abc import Sequence @@ -235,7 +236,7 @@ async def draw_node(tool_or_agent: Union[BaseAgent, BaseTool]) -> None: if name in highlight_tuple: # if in highlight, draw highlight node if as_cluster: - cluster = graphviz.Digraph( + cluster = _Digraph( name='cluster_' + name ) # adding "cluster_" to the name makes the graph render as a cluster subgraph await build_cluster(cluster, agent, name) @@ -253,7 +254,7 @@ async def draw_node(tool_or_agent: Union[BaseAgent, BaseTool]) -> None: return # if not in highlight, draw non-highlight node if as_cluster: - cluster = graphviz.Digraph( + cluster = _Digraph( name='cluster_' + name ) # adding "cluster_" to the name makes the graph render as a cluster subgraph await build_cluster(cluster, agent, name) @@ -314,7 +315,7 @@ async def get_agent_graph( dark_mode: bool = True, ) -> graphviz.Digraph | bytes: bg_color = '#333537' if dark_mode else '#ffffff' - graph = graphviz.Digraph( + graph = _Digraph( graph_attr={'rankdir': 'LR', 'bgcolor': bg_color}, strict=True ) await build_graph(graph, root_agent, highlights_pairs) diff --git a/tests/unittests/cli/test_agent_graph.py b/tests/unittests/cli/test_agent_graph.py index 249cb77519e..f5c3464c74f 100644 --- a/tests/unittests/cli/test_agent_graph.py +++ b/tests/unittests/cli/test_agent_graph.py @@ -24,6 +24,7 @@ from google.adk.agents.sequential_agent import SequentialAgent from google.adk.cli.agent_graph import get_agent_graph from google.adk.tools.agent_tool import AgentTool +from google.adk.tools.function_tool import FunctionTool import pytest _DARK_GREEN = '#0F5223' @@ -250,3 +251,66 @@ async def test_get_agent_graph_is_strict_so_repeated_edges_collapse(): assert graph.strict assert graph.source.count('child -> roll_dice') == 1 + + +def _colon_tool() -> FunctionTool: + """A tool whose name contains ':', which Gemini function names allow.""" + tool = FunctionTool(roll_dice) + tool.name = 'srv:tool' + return tool + + +def _assert_tool_edge_lands_on_declared_node(source: str, agent: str) -> None: + """Checks `agent -> "srv:tool"` targets the node, not a phantom port. + + _parse() unquotes names, so it cannot tell `a -> srv:tool` (port syntax) + from `a -> "srv:tool"`; compare the raw tokens instead. + """ + declared = { + match['name'] + for line in source.splitlines() + if (match := _NODE_RE.match(line.strip())) + } + endpoints = { + match[end] + for line in source.splitlines() + if (match := _EDGE_RE.match(line.strip())) + for end in ('src', 'dst') + } + assert '"srv:tool"' in declared + assert endpoints <= declared + assert 'srv' not in endpoints + assert f'{agent} -> "srv:tool"' in source + + +@pytest.mark.asyncio +async def test_build_graph_tool_name_with_colon_is_linked_to_its_agent(): + root = LlmAgent(name='root', model='gemini-2.0-flash', tools=[_colon_tool()]) + + graph = await get_agent_graph(root, []) + + _assert_tool_edge_lands_on_declared_node(graph.source, 'root') + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + 'workflow_agent_class', [SequentialAgent, LoopAgent, ParallelAgent] +) +async def test_build_graph_tool_name_with_colon_is_linked_inside_a_cluster( + workflow_agent_class, +): + workflow = workflow_agent_class( + name='steps', + sub_agents=[ + LlmAgent( + name='first', model='gemini-2.0-flash', tools=[_colon_tool()] + ), + LlmAgent(name='second', model='gemini-2.0-flash'), + ], + ) + + graph = await get_agent_graph(workflow, []) + + # The tool edge is drawn on the cluster's own Digraph. + assert 'subgraph "cluster_steps' in graph.source + _assert_tool_edge_lands_on_declared_node(graph.source, 'first')