Jeff/feat/graph

dimos/core/introspection/blueprint/dot.py

@@ -51,13 +51,19 @@ class LayoutAlgo(Enum):
     # "FoxgloveBridge",
 }
 
+# Modules only ignored when show_disconnected is False (compact view)
+_COMPACT_ONLY_IGNORED_MODULES = {
+    "WebsocketVisModule",
+}
+
 
 def render(
     blueprint_set: Blueprint,
     *,
     layout: set[LayoutAlgo] | None = None,
     ignored_streams: set[tuple[str, str]] | None = None,
     ignored_modules: set[str] | None = None,
+    show_disconnected: bool = False,
 ) -> str:
     """Generate a hub-style DOT graph from a Blueprint.
 
@@ -69,6 +75,8 @@ def render(
         layout: Set of layout algorithms to apply. Default is none (let graphviz decide).
         ignored_streams: Set of (name, type_name) tuples to ignore.
         ignored_modules: Set of module names to ignore.
+        show_disconnected: If True, show streams that have a producer but no consumer
+            (or vice versa) as dashed stub nodes.
 
     Returns:
         A string in DOT format showing modules as nodes, type nodes as
@@ -79,7 +87,10 @@ def render(
     if ignored_streams is None:
         ignored_streams = DEFAULT_IGNORED_CONNECTIONS
     if ignored_modules is None:
-        ignored_modules = DEFAULT_IGNORED_MODULES
+        if show_disconnected:
+            ignored_modules = DEFAULT_IGNORED_MODULES - _COMPACT_ONLY_IGNORED_MODULES
+        else:
+            ignored_modules = DEFAULT_IGNORED_MODULES
 
     # Collect all outputs: (name, type) -> list of producer modules
     producers: dict[tuple[str, type], list[type[Module]]] = defaultdict(list)
@@ -116,6 +127,23 @@ def render(
         label = f"{name}:{type_name}"
         active_channels[key] = color_for_string(TYPE_COLORS, label)
 
+    # Find disconnected channels (producer-only or consumer-only)
+    disconnected_channels: dict[tuple[str, type], str] = {}
+    if show_disconnected:
+        all_keys = set(producers.keys()) | set(consumers.keys())
+        for key in all_keys:
+            if key in active_channels:
+                continue
+            name, type_ = key
+            type_name = type_.__name__
+            if (name, type_name) in ignored_streams:
+                continue
+            relevant_modules = producers.get(key, []) + consumers.get(key, [])
+            if all(m.__name__ in ignored_modules for m in relevant_modules):
+                continue
+            label = f"{name}:{type_name}"
+            disconnected_channels[key] = color_for_string(TYPE_COLORS, label)
+
     # Group modules by package
     def get_group(mod_class: type[Module]) -> str:
         module_path = mod_class.__module__
@@ -218,6 +246,37 @@ def get_group(mod_class: type[Module]) -> str:
                 continue
             lines.append(f'    {node_id} -> {consumer.__name__} [color="{color}"];')
 
+    # Disconnected channels (dashed stub nodes)
+    if disconnected_channels:
+        lines.append("")
+        lines.append("    // Disconnected streams")
+        for key, color in sorted(
+            disconnected_channels.items(), key=lambda x: f"{x[0][0]}:{x[0][1].__name__}"
+        ):
+            name, type_ = key
+            type_name = type_.__name__
+            node_id = sanitize_id(f"chan_{name}_{type_name}")
+            label = f"{name}:{type_name}"
+            lines.append(
+                f'    {node_id} [label="{label}", shape=note, '
+                f'style="filled,dashed", fillcolor="{color}15", color="{color}", '
+                f'fontcolor="{color}", width=0, height=0, margin="0.1,0.05", fontsize=10];'
+            )
+
+            for producer in producers.get(key, []):
+                if producer.__name__ in ignored_modules:
+                    continue
+                lines.append(
+                    f"    {producer.__name__} -> {node_id} "
+                    f'[color="{color}", style=dashed, arrowhead=none];'
+                )
+            for consumer in consumers.get(key, []):
+                if consumer.__name__ in ignored_modules:
+                    continue
+                lines.append(
+                    f'    {node_id} -> {consumer.__name__} [color="{color}", style=dashed];'
+                )
+
     lines.append("}")
     return "\n".join(lines)
 
@@ -227,20 +286,22 @@ def render_svg(
     output_path: str,
     *,
     layout: set[LayoutAlgo] | None = None,
+    show_disconnected: bool = False,
 ) -> None:
     """Generate an SVG file from a Blueprint using graphviz.
 
     Args:
         blueprint_set: The blueprint set to visualize.
         output_path: Path to write the SVG file.
         layout: Set of layout algorithms to apply.
+        show_disconnected: If True, show streams with no matching counterpart.
     """
     import subprocess
 
     if layout is None:
         layout = set()
 
-    dot_code = render(blueprint_set, layout=layout)
+    dot_code = render(blueprint_set, layout=layout, show_disconnected=show_disconnected)
     engine = "fdp" if LayoutAlgo.FDP in layout else "dot"
     result = subprocess.run(
         [engine, "-Tsvg", "-o", output_path],

dimos/core/introspection/blueprint/test_dot.py

@@ -0,0 +1,62 @@
+# Copyright 2025-2026 Dimensional Inc.
+#
+# 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.
+
+
+from dimos.core.blueprints import autoconnect
+from dimos.core.introspection.blueprint.dot import render
+from dimos.core.module import Module
+from dimos.core.stream import In, Out
+
+
+class MsgA:
+    pass
+
+
+class MsgB:
+    pass
+
+
+class ProducerModule(Module):
+    output_a: Out[MsgA]
+    output_b: Out[MsgB]
+
+
+class ConsumerModule(Module):
+    output_a: In[MsgA]
+
+
+# output_a connects (same name+type), output_b is disconnected (no consumer)
+_combined = autoconnect(ProducerModule.blueprint(), ConsumerModule.blueprint())
+
+
+def test_render_without_disconnected() -> None:
+    dot = render(_combined, ignored_streams=set(), ignored_modules=set(), show_disconnected=False)
+    # Connected channel should be present
+    assert "output_a:MsgA" in dot
+    # Disconnected output_b should NOT appear
+    assert "output_b:MsgB" not in dot
+
+
+def test_render_with_disconnected() -> None:
+    dot = render(_combined, ignored_streams=set(), ignored_modules=set(), show_disconnected=True)
+    # Connected channel should be present
+    assert "output_a:MsgA" in dot
+    # Disconnected output_b SHOULD appear with dashed style
+    assert "output_b:MsgB" in dot
+    assert "style=dashed" in dot
+
+
+def test_disconnected_default_is_false() -> None:
+    dot = render(_combined, ignored_streams=set(), ignored_modules=set())
+    assert "output_b:MsgB" not in dot

dimos/core/introspection/svg.py

@@ -29,6 +29,7 @@ def to_svg(
     output_path: str,
     *,
     layout: set[LayoutAlgo] | None = None,
+    show_disconnected: bool = False,
 ) -> None:
     """Render a module or blueprint to SVG.
 
@@ -40,6 +41,7 @@ def to_svg(
         target: Either a ModuleInfo (single module) or Blueprint (blueprint graph).
         output_path: Path to write the SVG file.
         layout: Layout algorithms (only used for blueprints).
+        show_disconnected: If True, show streams with no matching counterpart (blueprints only).
     """
     # Avoid circular imports by importing here
     from dimos.core.blueprints import Blueprint
@@ -52,6 +54,8 @@ def to_svg(
     elif isinstance(target, Blueprint):
         from dimos.core.introspection.blueprint import dot as blueprint_dot
 
-        blueprint_dot.render_svg(target, output_path, layout=layout)
+        blueprint_dot.render_svg(
+            target, output_path, layout=layout, show_disconnected=show_disconnected
+        )
     else:
         raise TypeError(f"Expected ModuleInfo or Blueprint, got {type(target).__name__}")

dimos/robot/cli/dimos.py

@@ -550,6 +550,20 @@ def send(
     topic_send(topic, message_expr)
 
 
+@main.command()
+def graph(
+    python_file: str = typer.Argument(..., help="Python file containing Blueprint globals"),
+    no_disconnected: bool = typer.Option(
+        False, "--no-disconnected", help="Hide disconnected streams"
+    ),
+    port: int = typer.Option(0, "--port", help="HTTP server port (0 = random free port)"),
+) -> None:
+    """Render blueprint graphs from a Python file and open in browser."""
+    from dimos.utils.cli.graph import main as graph_main
+
+    graph_main(python_file, show_disconnected=not no_disconnected, port=port)
+
+
 @main.command(name="rerun-bridge")
 def rerun_bridge_cmd(
     viewer_mode: str = typer.Option(

dimos/utils/cli/graph.py

@@ -0,0 +1,110 @@
+# Copyright 2025-2026 Dimensional Inc.
+#
+# 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.
+
+"""Render Blueprint graphs from a Python file and open in the browser."""
+
+from __future__ import annotations
+
+import importlib.util
+import os
+import shutil
+import tempfile
+import webbrowser
+
+
+def _build_html(python_file: str, *, show_disconnected: bool = True) -> str:
+    """Import a Python file, find all Blueprint globals, and return rendered HTML."""
+    filepath = os.path.abspath(python_file)
+    if not os.path.isfile(filepath):
+        raise FileNotFoundError(filepath)
+
+    spec = importlib.util.spec_from_file_location("_render_target", filepath)
+    if spec is None or spec.loader is None:
+        raise RuntimeError(f"Could not load {filepath}")
+    mod = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(mod)
+
+    from dimos.core.blueprints import Blueprint
+    from dimos.core.introspection.svg import to_svg
+
+    blueprints: list[tuple[str, Blueprint]] = []
+    for name, obj in vars(mod).items():
+        if name.startswith("_"):
+            continue
+        if isinstance(obj, Blueprint):
+            blueprints.append((name, obj))
+
+    if not blueprints:
+        raise RuntimeError("No Blueprint instances found in module globals.")
+
+    print(f"Found {len(blueprints)} blueprint(s): {', '.join(n for n, _ in blueprints)}")
+
+    if not shutil.which("dot"):
+        raise RuntimeError(
+            "graphviz is not installed (the 'dot' command was not found).\n"
+            "Install it with:  brew install graphviz   (macOS)\n"
+            "                  apt install graphviz    (Debian/Ubuntu)"
+        )
+
+    sections = []
+    for name, bp in blueprints:
+        fd, svg_path = tempfile.mkstemp(suffix=".svg", prefix=f"dimos_{name}_")
+        os.close(fd)
+        to_svg(bp, svg_path, show_disconnected=show_disconnected)
+        with open(svg_path) as f:
+            svg_content = f.read()
+        os.unlink(svg_path)
+        sections.append(f'<h2>{name}</h2>\n<div class="diagram">{svg_content}</div>')
+
+    return f"""\
+<!DOCTYPE html>
+<html><head>
+<meta charset="utf-8">
+<title>Blueprint Diagrams</title>
+<style>
+body {{ background: #1e1e1e; color: #ccc; font-family: sans-serif; margin: 2em; }}
+h2 {{ border-bottom: 1px solid #444; padding-bottom: 0.3em; }}
+.diagram {{ margin-bottom: 3em; }}
+.diagram svg {{ max-width: 100%; height: auto; }}
+</style>
+</head><body>
+{"".join(sections)}
+</body></html>"""
+
+
+def main(python_file: str, *, show_disconnected: bool = True, port: int = 0) -> None:
+    """Render Blueprint SVG diagrams and display them via a one-shot HTTP server."""
+    from http.server import BaseHTTPRequestHandler, HTTPServer
+
+    html = _build_html(python_file, show_disconnected=show_disconnected)
+    html_bytes = html.encode("utf-8")
+
+    class Handler(BaseHTTPRequestHandler):
+        def do_GET(self) -> None:
+            self.send_response(200)
+            self.send_header("Content-Type", "text/html; charset=utf-8")
+            self.send_header("Content-Length", str(len(html_bytes)))
+            self.end_headers()
+            self.wfile.write(html_bytes)
+
+        def log_message(self, format: str, *args: object) -> None:
+            pass
+
+    server = HTTPServer(("0.0.0.0", port), Handler)
+    actual_port = server.server_address[1]
+    url = f"http://localhost:{actual_port}"
+    print(f"Serving at {url}  (will exit after first request)")
+    webbrowser.open(url)
+    server.handle_request()
+    print("Served. Exiting.")

dimos/utils/cli/test_graph.py

@@ -0,0 +1,41 @@
+# Copyright 2025-2026 Dimensional Inc.
+#
+# 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.
+
+
+import pytest
+
+from dimos.utils.cli.graph import main
+
+
+def test_file_not_found() -> None:
+    with pytest.raises(FileNotFoundError):
+        main("/nonexistent/path.py")
+
+
+def test_no_blueprints(tmp_path: object) -> None:
+    import pathlib
+
+    p = pathlib.Path(str(tmp_path)) / "empty.py"
+    p.write_text("x = 42\n")
+    with pytest.raises(RuntimeError, match="No Blueprint instances"):
+        main(str(p))
+
+
+def test_module_load_failure(tmp_path: object) -> None:
+    import pathlib
+
+    p = pathlib.Path(str(tmp_path)) / "bad.py"
+    p.write_text("raise ImportError('boom')\n")
+    with pytest.raises(ImportError, match="boom"):
+        main(str(p))