nodegraph analsys and mutation tools for agentic use

src/talemate/game/engine/nodes/tools/__init__.py

@@ -0,0 +1,138 @@
+"""
+Static analysis, mutation, and layout tools for Talemate node graphs.
+
+This toolset is primarily aimed at helping **LLM-driven agents** author
+and modify Talemate node-graph JSON files safely. Hand-editing these
+graphs is error-prone for language models — socket names, property
+keys, registry paths, and edge wiring are all easy to typo and produce
+silently-broken files. The tools here exist so an agent can:
+
+* **Analyze** existing graphs (``analysis.py``, exposed via the
+  ``python -m talemate.game.engine.nodes.tools`` CLI) to understand
+  structure before modifying anything — ``summary``, ``stages``,
+  ``node``, ``rtrace``, ``check-registries``, etc.
+* **Mutate** graphs programmatically (``writer.py``) via
+  ``GraphWriter``, which validates socket names and property keys
+  against the live node-class registry at write time and raises
+  ``UnknownSocketError`` / ``UnknownPropertyError`` / ``CycleError`` /
+  etc. rather than letting typos slip through into the JSON.
+* **Lay out** graphs deterministically (``layout.py``) so the agent
+  never has to pick ``(x, y)`` coordinates — a stage-stratified
+  algorithm handles positioning with predecessor-aware row placement
+  and height estimation.
+* **Load** graph JSON with scene-module auto-registration
+  (``loader.py``) so sibling scene modules are visible to the writer
+  and analysis layer without manual registry setup.
+
+A human working directly in the Talemate visual editor does not need
+any of this — the editor handles socket wiring, layout, and validation
+natively. These tools exist specifically for the out-of-editor,
+machine-authored path.
+"""
+
+from .analysis import (
+    ConsumersResult,
+    EdgeInfo,
+    FeedsResult,
+    GraphSummary,
+    NodeDetails,
+    NodeListEntry,
+    RegistryCheckResult,
+    StageMapResult,
+    TraceNode,
+    check_registries,
+    consumers,
+    ensure_registry_loaded,
+    feeds,
+    get_node,
+    list_edges,
+    list_nodes,
+    resolve_node_id,
+    stage_map,
+    summarize,
+    trace_backward,
+    trace_forward,
+)
+from .layout import LayoutError, LayoutOptions, layout_graph
+from .loader import GraphLoadError, load_graph, register_scene_modules
+from .writer import (
+    AlreadyConnectedError,
+    CycleError,
+    DynamicInputError,
+    GraphWriter,
+    GroupColor,
+    GroupError,
+    NodeMetadata,
+    NodeNotFoundError,
+    NotConnectedError,
+    UnknownPropertyError,
+    UnknownRegistryError,
+    UnknownSocketError,
+    WriterError,
+    add_dynamic_input,
+    add_group,
+    add_node,
+    clear_metadata_cache,
+    connect,
+    disconnect,
+    get_node_metadata,
+    remove_dynamic_input,
+    remove_node,
+)
+
+__all__ = [
+    # analysis models
+    "GraphSummary",
+    "NodeListEntry",
+    "NodeDetails",
+    "EdgeInfo",
+    "FeedsResult",
+    "ConsumersResult",
+    "TraceNode",
+    "StageMapResult",
+    "RegistryCheckResult",
+    # analysis functions
+    "summarize",
+    "list_nodes",
+    "get_node",
+    "list_edges",
+    "feeds",
+    "consumers",
+    "trace_forward",
+    "trace_backward",
+    "stage_map",
+    "check_registries",
+    "resolve_node_id",
+    "ensure_registry_loaded",
+    # loader
+    "GraphLoadError",
+    "load_graph",
+    "register_scene_modules",
+    # writer
+    "GraphWriter",
+    "NodeMetadata",
+    "GroupColor",
+    "get_node_metadata",
+    "clear_metadata_cache",
+    "add_node",
+    "remove_node",
+    "connect",
+    "disconnect",
+    "add_dynamic_input",
+    "remove_dynamic_input",
+    "add_group",
+    "WriterError",
+    "UnknownRegistryError",
+    "UnknownSocketError",
+    "UnknownPropertyError",
+    "AlreadyConnectedError",
+    "NotConnectedError",
+    "CycleError",
+    "DynamicInputError",
+    "GroupError",
+    "NodeNotFoundError",
+    # layout
+    "layout_graph",
+    "LayoutError",
+    "LayoutOptions",
+]

src/talemate/game/engine/nodes/tools/__main__.py

@@ -0,0 +1,8 @@
+"""Allow ``python -m talemate.game.engine.nodes.tools`` to invoke the CLI."""
+
+import sys
+
+from .cli import main
+
+if __name__ == "__main__":
+    sys.exit(main())

src/talemate/game/engine/nodes/tools/analysis.py

@@ -0,0 +1,974 @@
+"""
+Read-only static analysis of Talemate node graph JSON dicts.
+
+Every public function takes an already-loaded ``graph: dict`` (the parsed
+contents of a graph .json file) and returns a pydantic model. There is no
+file I/O, no formatting, and no runtime instantiation of node classes from
+the graph itself - we only consult the live ``NODES`` registry to look up
+static socket layouts where available.
+
+This module is intended as the analysis layer for both a developer CLI
+(``tools/cli.py``) and, eventually, a UI surface. Keep it pure.
+"""
+
+from __future__ import annotations
+
+from collections import Counter, defaultdict
+from typing import Any
+
+import pydantic
+import structlog
+
+log = structlog.get_logger("talemate.game.engine.nodes.tools.analysis")
+
+__all__ = [
+    # registry loader
+    "ensure_registry_loaded",
+    # models
+    "GraphSummary",
+    "NodeListEntry",
+    "SocketRef",
+    "NodeInputInfo",
+    "NodeOutputInfo",
+    "NodeDetails",
+    "EdgeInfo",
+    "FeedsResult",
+    "ConsumersResult",
+    "TraceNode",
+    "StageInfo",
+    "StageChain",
+    "StageMapResult",
+    "RegistryProblem",
+    "RegistryCheckResult",
+    # functions
+    "summarize",
+    "list_nodes",
+    "get_node",
+    "list_edges",
+    "feeds",
+    "consumers",
+    "trace_forward",
+    "trace_backward",
+    "stage_map",
+    "check_registries",
+    # helpers
+    "resolve_node_id",
+    "split_socket_path",
+]
+
+
+# ---------------------------------------------------------------------------
+# Registry loader (lazy, side-effecty - only run when actually needed)
+# ---------------------------------------------------------------------------
+
+_REGISTRY_LOADED = False
+
+
+def ensure_registry_loaded() -> None:
+    """Import all shipped Talemate node definitions exactly once.
+
+    This is lazy because importing the registry has filesystem side effects
+    (walks ``SEARCH_PATHS``) and we want analysis functions that don't need
+    the registry to remain cheap.
+    """
+
+    global _REGISTRY_LOADED
+    if _REGISTRY_LOADED:
+        return
+    # First import the Python modules that register native node classes
+    # via @register decorators (core/MakeBool, validation/*, etc).
+    import talemate.game.engine.nodes.load_definitions  # noqa: F401
+    from talemate.game.engine.nodes.registry import import_initial_node_definitions
+
+    # Then walk SEARCH_PATHS to register JSON-defined module nodes.
+    import_initial_node_definitions()
+    _REGISTRY_LOADED = True
+
+
+def _registry_dict() -> dict[str, Any]:
+    ensure_registry_loaded()
+    from talemate.game.engine.nodes.registry import NODES
+
+    return NODES
+
+
+def _base_types_dict() -> dict[str, Any]:
+    ensure_registry_loaded()
+    from talemate.game.engine.nodes.base_types import BASE_TYPES
+
+    return BASE_TYPES
+
+
+# ---------------------------------------------------------------------------
+# Pydantic models
+# ---------------------------------------------------------------------------
+
+
+class GraphSummary(pydantic.BaseModel):
+    title: str | None
+    id: str | None
+    registry: str | None
+    base_type: str | None
+    extends: str | None
+    node_count: int
+    nodes_by_category: dict[str, int]
+    stage_node_count: int
+    input_count: int
+    output_count: int
+    module_property_count: int
+    group_titles: list[str]
+
+
+class NodeListEntry(pydantic.BaseModel):
+    short_id: str
+    id: str
+    registry: str | None
+    title: str
+
+
+class SocketRef(pydantic.BaseModel):
+    node_id: str
+    short_id: str
+    socket: str
+
+    @property
+    def full(self) -> str:
+        return f"{self.node_id}.{self.socket}"
+
+
+class NodeInputInfo(pydantic.BaseModel):
+    name: str
+    connected: bool
+    source: SocketRef | None = None
+
+
+class NodeOutputInfo(pydantic.BaseModel):
+    name: str
+    consumers: list[SocketRef] = pydantic.Field(default_factory=list)
+
+
+class NodeDetails(pydantic.BaseModel):
+    id: str
+    short_id: str
+    title: str
+    registry: str | None
+    base_type: str | None
+    properties: dict[str, Any]
+    x: int = 0
+    y: int = 0
+    width: int = 0
+    height: int = 0
+    inputs: list[NodeInputInfo]
+    outputs: list[NodeOutputInfo]
+    registered: bool
+
+
+class EdgeInfo(pydantic.BaseModel):
+    source_node: str
+    source_short: str
+    source_socket: str
+    target_node: str
+    target_short: str
+    target_socket: str
+
+
+class FeedsResult(pydantic.BaseModel):
+    target_node: str
+    target_short: str
+    target_socket: str
+    source: SocketRef | None = None
+    source_node_title: str | None = None
+    source_node_registry: str | None = None
+
+
+class ConsumersResult(pydantic.BaseModel):
+    source_node: str
+    source_short: str
+    source_socket: str
+    consumers: list[EdgeInfo] = pydantic.Field(default_factory=list)
+
+
+class TraceNode(pydantic.BaseModel):
+    node_id: str
+    short_id: str
+    title: str
+    registry: str | None
+    via_socket: str | None = None  # the socket on *this* node we walked through
+    children: list["TraceNode"] = pydantic.Field(default_factory=list)
+    cycle: bool = False
+    truncated: bool = False  # depth limit hit
+
+
+class StageInfo(pydantic.BaseModel):
+    node_id: str
+    short_id: str
+    title: str
+    stage: int
+    chain_index: int
+
+
+class StageChain(pydantic.BaseModel):
+    index: int
+    node_ids: list[str]
+    stage_node_ids: list[str]
+    stages: list[int]
+
+
+class StageMapResult(pydantic.BaseModel):
+    stage_nodes: list[StageInfo]
+    chains: list[StageChain]
+    unstaged_node_ids: list[str]
+
+
+class RegistryProblem(pydantic.BaseModel):
+    node_id: str
+    short_id: str
+    title: str
+    registry: str | None
+    base_type: str | None
+    kind: str  # "registry" | "base_type"
+
+
+class RegistryCheckResult(pydantic.BaseModel):
+    unknown_registries: list[RegistryProblem]
+    unknown_base_types: list[RegistryProblem]
+    checked_registry_count: int
+    checked_base_type_count: int
+
+    @property
+    def has_problems(self) -> bool:
+        return bool(self.unknown_registries or self.unknown_base_types)
+
+
+# ---------------------------------------------------------------------------
+# Helpers
+# ---------------------------------------------------------------------------
+
+
+def _short(node_id: str) -> str:
+    return node_id[:8]
+
+
+def _nodes(graph: dict) -> dict[str, dict]:
+    return graph.get("nodes") or {}
+
+
+def _edges(graph: dict) -> dict[str, list[str]]:
+    return graph.get("edges") or {}
+
+
+def split_socket_path(path: str) -> tuple[str, str]:
+    """Split a ``<node_id>.<socket>`` path on the FIRST dot.
+
+    Node ids (full UUIDs or short prefixes) and socket names in this
+    codebase never contain dots, so the first dot cleanly separates
+    the node-id half from the socket-name half.
+    """
+
+    if "." not in path:
+        raise ValueError(
+            f"Socket path '{path}' must be of the form '<node_id>.<socket>'"
+        )
+    node_part, socket = path.split(".", 1)
+    return node_part, socket
+
+
+def resolve_node_id(graph: dict, prefix: str) -> str:
+    """Resolve a possibly-truncated node id to a full node id.
+
+    Accepts a full UUID or any unique prefix. Raises ``ValueError`` on
+    miss or ambiguity.
+    """
+
+    nodes = _nodes(graph)
+    if not prefix:
+        raise ValueError("Empty node id prefix")
+
+    if prefix in nodes:
+        return prefix
+
+    matches = [nid for nid in nodes if nid.startswith(prefix)]
+    if not matches:
+        raise ValueError(
+            f"No node matches prefix '{prefix}'. "
+            f"(Graph contains {len(nodes)} nodes; try `list-nodes` to find one.)"
+        )
+    if len(matches) > 1:
+        joined = ", ".join(_short(m) for m in matches[:6])
+        more = "" if len(matches) <= 6 else f" (+{len(matches) - 6} more)"
+        raise ValueError(
+            f"Ambiguous node prefix '{prefix}' matches {len(matches)} nodes: {joined}{more}"
+        )
+    return matches[0]
+
+
+def _make_socket_ref(node_id: str, socket: str) -> SocketRef:
+    return SocketRef(node_id=node_id, short_id=_short(node_id), socket=socket)
+
+
+def _registered_node_class(registry: str | None) -> Any | None:
+    """Return the live class registered under ``registry`` or ``None``."""
+
+    if not registry:
+        return None
+    return _registry_dict().get(registry)
+
+
+def _instance_for_class(node_cls: Any) -> Any | None:
+    """Try to instantiate a node class to read its socket layout.
+
+    This is a best-effort, side-effect-free probe: registered classes set
+    up their sockets in ``setup()``, and that ``setup`` runs in
+    ``__init__``. If anything goes wrong (constructor expects positional
+    args, etc.), return ``None`` and let the caller fall back.
+    """
+
+    if node_cls is None:
+        return None
+    try:
+        return node_cls()
+    except Exception:
+        return None
+
+
+def _static_socket_names(registry: str | None) -> tuple[list[str] | None, list[str] | None]:
+    """Return ``(input_names, output_names)`` from the registered class.
+
+    Each element is ``None`` when we couldn't determine it.
+    """
+
+    cls = _registered_node_class(registry)
+    instance = _instance_for_class(cls)
+    if instance is None:
+        return (None, None)
+
+    input_names: list[str] | None
+    output_names: list[str] | None
+    try:
+        input_names = [s.name for s in instance.inputs]
+    except Exception:
+        input_names = None
+    try:
+        output_names = [s.name for s in instance.outputs]
+    except Exception:
+        output_names = None
+    return input_names, output_names
+
+
+def _node_input_names(graph: dict, node_id: str, node: dict) -> list[str]:
+    """Best-effort list of input socket names for a node.
+
+    Order of preference:
+        1. registered class static inputs (+ dynamic_inputs from JSON)
+        2. dynamic_inputs from JSON alone
+        3. inferred from edges where this node appears as a target
+    """
+
+    static_inputs, _ = _static_socket_names(node.get("registry"))
+    dyn = [d.get("name") for d in (node.get("dynamic_inputs") or []) if d.get("name")]
+    inferred = sorted(_inferred_input_names_from_edges(graph, node_id))
+
+    if static_inputs is not None:
+        # Combine static + dynamic, then drop duplicates while preserving order.
+        seen: set[str] = set()
+        combined: list[str] = []
+        for name in [*static_inputs, *dyn, *inferred]:
+            if name not in seen:
+                seen.add(name)
+                combined.append(name)
+        return combined
+
+    if dyn or inferred:
+        seen = set()
+        combined = []
+        for name in [*dyn, *inferred]:
+            if name not in seen:
+                seen.add(name)
+                combined.append(name)
+        return combined
+
+    return []
+
+
+def _node_output_names(graph: dict, node_id: str, node: dict) -> list[str]:
+    """Best-effort list of output socket names for a node."""
+
+    _, static_outputs = _static_socket_names(node.get("registry"))
+    inferred = sorted(_inferred_output_names_from_edges(graph, node_id))
+
+    if static_outputs is not None:
+        seen: set[str] = set()
+        combined: list[str] = []
+        for name in [*static_outputs, *inferred]:
+            if name not in seen:
+                seen.add(name)
+                combined.append(name)
+        return combined
+
+    return inferred
+
+
+def _inferred_input_names_from_edges(graph: dict, node_id: str) -> set[str]:
+    names: set[str] = set()
+    for _src, targets in _edges(graph).items():
+        for tgt in targets:
+            try:
+                tgt_node, tgt_socket = split_socket_path(tgt)
+            except ValueError:
+                continue
+            if tgt_node == node_id:
+                names.add(tgt_socket)
+    return names
+
+
+def _inferred_output_names_from_edges(graph: dict, node_id: str) -> set[str]:
+    names: set[str] = set()
+    for src, _targets in _edges(graph).items():
+        try:
+            src_node, src_socket = split_socket_path(src)
+        except ValueError:
+            continue
+        if src_node == node_id:
+            names.add(src_socket)
+    return names
+
+
+def _input_source_map(graph: dict) -> dict[tuple[str, str], tuple[str, str]]:
+    """Map ``(target_node, target_socket) -> (source_node, source_socket)``.
+
+    The graph format treats edges as ``source -> [targets]``, so this
+    inverts the relation. If a target socket somehow appears with multiple
+    sources we keep the last one we see (the runtime treats inputs as
+    single-source).
+    """
+
+    out: dict[tuple[str, str], tuple[str, str]] = {}
+    for src, targets in _edges(graph).items():
+        try:
+            src_node, src_socket = split_socket_path(src)
+        except ValueError:
+            continue
+        for tgt in targets:
+            try:
+                tgt_node, tgt_socket = split_socket_path(tgt)
+            except ValueError:
+                continue
+            out[(tgt_node, tgt_socket)] = (src_node, src_socket)
+    return out
+
+
+# ---------------------------------------------------------------------------
+# summarize
+# ---------------------------------------------------------------------------
+
+
+def summarize(graph: dict) -> GraphSummary:
+    nodes = _nodes(graph)
+
+    categories: Counter[str] = Counter()
+    stage_count = 0
+    for node in nodes.values():
+        registry = node.get("registry") or ""
+        if "/" in registry:
+            category = registry.split("/", 1)[0]
+        else:
+            category = registry or "<none>"
+        categories[category] += 1
+        if registry == "core/Stage":
+            stage_count += 1
+
+    groups = graph.get("groups") or []
+    group_titles = [g.get("title", "") for g in groups]
+
+    return GraphSummary(
+        title=graph.get("title"),
+        id=graph.get("id"),
+        registry=graph.get("registry"),
+        base_type=graph.get("base_type"),
+        extends=graph.get("extends"),
+        node_count=len(nodes),
+        nodes_by_category=dict(sorted(categories.items())),
+        stage_node_count=stage_count,
+        input_count=len(graph.get("inputs") or []),
+        output_count=len(graph.get("outputs") or []),
+        module_property_count=len(graph.get("module_properties") or {}),
+        group_titles=group_titles,
+    )
+
+
+# ---------------------------------------------------------------------------
+# list_nodes
+# ---------------------------------------------------------------------------
+
+
+def list_nodes(
+    graph: dict,
+    *,
+    registry_pattern: str | None = None,
+    title_pattern: str | None = None,
+) -> list[NodeListEntry]:
+    nodes = _nodes(graph)
+    rp = registry_pattern.lower() if registry_pattern else None
+    tp = title_pattern.lower() if title_pattern else None
+
+    out: list[NodeListEntry] = []
+    for node_id, node in nodes.items():
+        registry = node.get("registry") or ""
+        title = node.get("title") or ""
+        if rp and rp not in registry.lower():
+            continue
+        if tp and tp not in title.lower():
+            continue
+        out.append(
+            NodeListEntry(
+                short_id=_short(node_id),
+                id=node_id,
+                registry=registry or None,
+                title=title,
+            )
+        )
+    out.sort(key=lambda e: (e.registry or "", e.title))
+    return out
+
+
+# ---------------------------------------------------------------------------
+# get_node
+# ---------------------------------------------------------------------------
+
+
+def get_node(graph: dict, node_id: str) -> NodeDetails:
+    full_id = resolve_node_id(graph, node_id)
+    node = _nodes(graph)[full_id]
+
+    input_source_map = _input_source_map(graph)
+    input_names = _node_input_names(graph, full_id, node)
+    output_names = _node_output_names(graph, full_id, node)
+
+    inputs: list[NodeInputInfo] = []
+    for name in input_names:
+        src = input_source_map.get((full_id, name))
+        if src is not None:
+            src_node, src_socket = src
+            inputs.append(
+                NodeInputInfo(
+                    name=name,
+                    connected=True,
+                    source=_make_socket_ref(src_node, src_socket),
+                )
+            )
+        else:
+            inputs.append(NodeInputInfo(name=name, connected=False))
+
+    outputs: list[NodeOutputInfo] = []
+    edges = _edges(graph)
+    for name in output_names:
+        full_socket = f"{full_id}.{name}"
+        targets = edges.get(full_socket, [])
+        consumer_refs: list[SocketRef] = []
+        for tgt in targets:
+            try:
+                tgt_node, tgt_socket = split_socket_path(tgt)
+            except ValueError:
+                continue
+            consumer_refs.append(_make_socket_ref(tgt_node, tgt_socket))
+        outputs.append(NodeOutputInfo(name=name, consumers=consumer_refs))
+
+    registered = _registered_node_class(node.get("registry")) is not None
+
+    return NodeDetails(
+        id=full_id,
+        short_id=_short(full_id),
+        title=node.get("title", ""),
+        registry=node.get("registry"),
+        base_type=node.get("base_type"),
+        properties=dict(node.get("properties") or {}),
+        x=int(node.get("x", 0) or 0),
+        y=int(node.get("y", 0) or 0),
+        width=int(node.get("width", 0) or 0),
+        height=int(node.get("height", 0) or 0),
+        inputs=inputs,
+        outputs=outputs,
+        registered=registered,
+    )
+
+
+# ---------------------------------------------------------------------------
+# list_edges
+# ---------------------------------------------------------------------------
+
+
+def list_edges(
+    graph: dict,
+    *,
+    from_node: str | None = None,
+    to_node: str | None = None,
+) -> list[EdgeInfo]:
+    from_full = resolve_node_id(graph, from_node) if from_node else None
+    to_full = resolve_node_id(graph, to_node) if to_node else None
+
+    out: list[EdgeInfo] = []
+    for src, targets in _edges(graph).items():
+        try:
+            src_node, src_socket = split_socket_path(src)
+        except ValueError:
+            continue
+        if from_full and src_node != from_full:
+            continue
+        for tgt in targets:
+            try:
+                tgt_node, tgt_socket = split_socket_path(tgt)
+            except ValueError:
+                continue
+            if to_full and tgt_node != to_full:
+                continue
+            out.append(
+                EdgeInfo(
+                    source_node=src_node,
+                    source_short=_short(src_node),
+                    source_socket=src_socket,
+                    target_node=tgt_node,
+                    target_short=_short(tgt_node),
+                    target_socket=tgt_socket,
+                )
+            )
+    out.sort(key=lambda e: (e.source_short, e.source_socket, e.target_short, e.target_socket))
+    return out
+
+
+# ---------------------------------------------------------------------------
+# feeds / consumers
+# ---------------------------------------------------------------------------
+
+
+def feeds(graph: dict, target: str) -> FeedsResult:
+    """Return the source feeding ``target`` (a ``<node_or_prefix>.<socket>``).
+
+    For an input socket: returns the upstream output that connects to it.
+    """
+
+    node_part, socket = split_socket_path(target)
+    full_id = resolve_node_id(graph, node_part)
+    src = _input_source_map(graph).get((full_id, socket))
+
+    source_ref: SocketRef | None = None
+    src_title: str | None = None
+    src_registry: str | None = None
+    if src is not None:
+        src_node_id, src_socket = src
+        source_ref = _make_socket_ref(src_node_id, src_socket)
+        src_node = _nodes(graph).get(src_node_id, {})
+        src_title = src_node.get("title")
+        src_registry = src_node.get("registry")
+
+    return FeedsResult(
+        target_node=full_id,
+        target_short=_short(full_id),
+        target_socket=socket,
+        source=source_ref,
+        source_node_title=src_title,
+        source_node_registry=src_registry,
+    )
+
+
+def consumers(graph: dict, source: str) -> ConsumersResult:
+    """Return all targets reading ``source`` (a ``<node_or_prefix>.<socket>``)."""
+
+    node_part, socket = split_socket_path(source)
+    full_id = resolve_node_id(graph, node_part)
+    full_key = f"{full_id}.{socket}"
+    edges = _edges(graph).get(full_key, [])
+
+    consumer_edges: list[EdgeInfo] = []
+    for tgt in edges:
+        try:
+            tgt_node, tgt_socket = split_socket_path(tgt)
+        except ValueError:
+            continue
+        consumer_edges.append(
+            EdgeInfo(
+                source_node=full_id,
+                source_short=_short(full_id),
+                source_socket=socket,
+                target_node=tgt_node,
+                target_short=_short(tgt_node),
+                target_socket=tgt_socket,
+            )
+        )
+    return ConsumersResult(
+        source_node=full_id,
+        source_short=_short(full_id),
+        source_socket=socket,
+        consumers=consumer_edges,
+    )
+
+
+# ---------------------------------------------------------------------------
+# trace_forward / trace_backward
+# ---------------------------------------------------------------------------
+
+
+def _make_trace_leaf(graph: dict, node_id: str, via: str | None, *, cycle: bool, truncated: bool) -> TraceNode:
+    node = _nodes(graph).get(node_id, {})
+    return TraceNode(
+        node_id=node_id,
+        short_id=_short(node_id),
+        title=node.get("title", ""),
+        registry=node.get("registry"),
+        via_socket=via,
+        cycle=cycle,
+        truncated=truncated,
+    )
+
+
+def _trace(
+    graph: dict,
+    start_id: str,
+    *,
+    depth: int,
+    forward: bool,
+    visited: set[str] | None = None,
+    via: str | None = None,
+) -> TraceNode:
+    if visited is None:
+        visited = set()
+    node = _nodes(graph).get(start_id, {})
+    if start_id in visited:
+        return _make_trace_leaf(graph, start_id, via, cycle=True, truncated=False)
+
+    current = TraceNode(
+        node_id=start_id,
+        short_id=_short(start_id),
+        title=node.get("title", ""),
+        registry=node.get("registry"),
+        via_socket=via,
+    )
+
+    if depth <= 0:
+        current.truncated = True
+        return current
+
+    visited = visited | {start_id}
+    edges = _edges(graph)
+
+    if forward:
+        # walk every edge whose source is this node
+        for src_path, targets in edges.items():
+            try:
+                src_node, src_socket = split_socket_path(src_path)
+            except ValueError:
+                continue
+            if src_node != start_id:
+                continue
+            for tgt in targets:
+                try:
+                    tgt_node, tgt_socket = split_socket_path(tgt)
+                except ValueError:
+                    continue
+                child = _trace(
+                    graph,
+                    tgt_node,
+                    depth=depth - 1,
+                    forward=True,
+                    visited=visited,
+                    via=tgt_socket,
+                )
+                current.children.append(child)
+    else:
+        # walk every edge whose target is this node
+        for src_path, targets in edges.items():
+            try:
+                src_node, src_socket = split_socket_path(src_path)
+            except ValueError:
+                continue
+            for tgt in targets:
+                try:
+                    tgt_node, tgt_socket = split_socket_path(tgt)
+                except ValueError:
+                    continue
+                if tgt_node != start_id:
+                    continue
+                child = _trace(
+                    graph,
+                    src_node,
+                    depth=depth - 1,
+                    forward=False,
+                    visited=visited,
+                    via=src_socket,
+                )
+                current.children.append(child)
+
+    return current
+
+
+def trace_forward(graph: dict, start: str, *, depth: int = 3) -> TraceNode:
+    full_id = resolve_node_id(graph, start)
+    return _trace(graph, full_id, depth=depth, forward=True)
+
+
+def trace_backward(graph: dict, end: str, *, depth: int = 3) -> TraceNode:
+    full_id = resolve_node_id(graph, end)
+    return _trace(graph, full_id, depth=depth, forward=False)
+
+
+# ---------------------------------------------------------------------------
+# stage_map
+# ---------------------------------------------------------------------------
+
+
+def _weakly_connected_components(graph: dict) -> list[set[str]]:
+    """Compute weakly-connected components over the node-level edge graph."""
+
+    nodes = _nodes(graph)
+    parent: dict[str, str] = {nid: nid for nid in nodes}
+
+    def find(x: str) -> str:
+        while parent[x] != x:
+            parent[x] = parent[parent[x]]
+            x = parent[x]
+        return x
+
+    def union(a: str, b: str) -> None:
+        ra, rb = find(a), find(b)
+        if ra != rb:
+            parent[ra] = rb
+
+    for src, targets in _edges(graph).items():
+        try:
+            src_node, _ = split_socket_path(src)
+        except ValueError:
+            continue
+        if src_node not in parent:
+            continue
+        for tgt in targets:
+            try:
+                tgt_node, _ = split_socket_path(tgt)
+            except ValueError:
+                continue
+            if tgt_node not in parent:
+                continue
+            union(src_node, tgt_node)
+
+    components: dict[str, set[str]] = defaultdict(set)
+    for nid in nodes:
+        components[find(nid)].add(nid)
+    return list(components.values())
+
+
+def stage_map(graph: dict) -> StageMapResult:
+    nodes = _nodes(graph)
+    components = _weakly_connected_components(graph)
+
+    # index components for stable ordering: smallest stage first, then size
+    def comp_sort_key(comp: set[str]) -> tuple[int, int]:
+        stage_vals = [
+            int((nodes[nid].get("properties") or {}).get("stage", 0))
+            for nid in comp
+            if nodes[nid].get("registry") == "core/Stage"
+        ]
+        min_stage = min(stage_vals) if stage_vals else 10**9
+        return (min_stage, -len(comp))
+
+    components.sort(key=comp_sort_key)
+
+    stage_nodes: list[StageInfo] = []
+    chains: list[StageChain] = []
+    unstaged: list[str] = []
+
+    for idx, comp in enumerate(components):
+        comp_stage_ids = [
+            nid for nid in comp if nodes[nid].get("registry") == "core/Stage"
+        ]
+        if not comp_stage_ids:
+            unstaged.extend(sorted(comp))
+            continue
+
+        comp_stages: list[int] = []
+        for sid in comp_stage_ids:
+            stage_val = int((nodes[sid].get("properties") or {}).get("stage", 0))
+            comp_stages.append(stage_val)
+            stage_nodes.append(
+                StageInfo(
+                    node_id=sid,
+                    short_id=_short(sid),
+                    title=nodes[sid].get("title", ""),
+                    stage=stage_val,
+                    chain_index=idx,
+                )
+            )
+        chains.append(
+            StageChain(
+                index=idx,
+                node_ids=sorted(comp),
+                stage_node_ids=sorted(comp_stage_ids),
+                stages=sorted(comp_stages),
+            )
+        )
+
+    stage_nodes.sort(key=lambda s: (s.chain_index, s.stage, s.short_id))
+    return StageMapResult(
+        stage_nodes=stage_nodes,
+        chains=chains,
+        unstaged_node_ids=sorted(unstaged),
+    )
+
+
+# ---------------------------------------------------------------------------
+# check_registries
+# ---------------------------------------------------------------------------
+
+
+def check_registries(graph: dict) -> RegistryCheckResult:
+    NODES = _registry_dict()
+    BASE_TYPES = _base_types_dict()
+
+    unknown_registries: list[RegistryProblem] = []
+    unknown_base_types: list[RegistryProblem] = []
+    checked_registries = 0
+    checked_base_types = 0
+
+    def _check(node_id: str, node: dict) -> None:
+        nonlocal checked_registries, checked_base_types
+        registry = node.get("registry")
+        base_type = node.get("base_type")
+        title = node.get("title") or ""
+        if registry:
+            checked_registries += 1
+            if registry not in NODES:
+                unknown_registries.append(
+                    RegistryProblem(
+                        node_id=node_id,
+                        short_id=_short(node_id),
+                        title=title,
+                        registry=registry,
+                        base_type=base_type,
+                        kind="registry",
+                    )
+                )
+        if base_type:
+            checked_base_types += 1
+            if base_type not in BASE_TYPES:
+                unknown_base_types.append(
+                    RegistryProblem(
+                        node_id=node_id,
+                        short_id=_short(node_id),
+                        title=title,
+                        registry=registry,
+                        base_type=base_type,
+                        kind="base_type",
+                    )
+                )
+
+    # check the top-level graph itself
+    top_id = graph.get("id") or "<graph>"
+    _check(top_id, graph)
+
+    # check every interior node
+    for nid, node in _nodes(graph).items():
+        _check(nid, node)
+
+    return RegistryCheckResult(
+        unknown_registries=unknown_registries,
+        unknown_base_types=unknown_base_types,
+        checked_registry_count=checked_registries,
+        checked_base_type_count=checked_base_types,
+    )

src/talemate/game/engine/nodes/tools/cli.py

@@ -0,0 +1,398 @@
+"""
+Argparse-based CLI wrapping ``analysis.py``.
+
+Subcommands map 1:1 to analysis functions. Every subcommand supports
+``--json`` for machine-readable output. Default output is dense, grep-
+friendly text using 8-char short UUID prefixes.
+"""
+
+from __future__ import annotations
+
+import argparse
+import io
+import json
+import sys
+from typing import Any, Callable, TextIO
+
+import pydantic
+
+from . import analysis
+from .loader import GraphLoadError, load_graph
+
+__all__ = ["main", "build_parser"]
+
+
+# ---------------------------------------------------------------------------
+# Exit codes
+# ---------------------------------------------------------------------------
+
+EXIT_OK = 0
+EXIT_STRUCT_ERROR = 1
+EXIT_REGISTRY_PROBLEMS = 2
+
+
+# ---------------------------------------------------------------------------
+# Formatters
+# ---------------------------------------------------------------------------
+
+
+def _fmt_summary(s: analysis.GraphSummary) -> str:
+    out = io.StringIO()
+    out.write(f"title:        {s.title}\n")
+    out.write(f"id:           {s.id}\n")
+    out.write(f"registry:     {s.registry}\n")
+    out.write(f"base_type:    {s.base_type}\n")
+    out.write(f"extends:      {s.extends}\n")
+    out.write(f"nodes:        {s.node_count}\n")
+    out.write(f"  by category:\n")
+    for cat, count in s.nodes_by_category.items():
+        out.write(f"    {cat:<20} {count}\n")
+    out.write(f"stage nodes:  {s.stage_node_count}\n")
+    out.write(
+        f"interface:    inputs={s.input_count} outputs={s.output_count} "
+        f"module_properties={s.module_property_count}\n"
+    )
+    if s.group_titles:
+        out.write("groups:\n")
+        for t in s.group_titles:
+            out.write(f"  - {t}\n")
+    return out.getvalue().rstrip()
+
+
+def _fmt_list_nodes(entries: list[analysis.NodeListEntry]) -> str:
+    if not entries:
+        return "(no nodes match)"
+    lines = []
+    reg_w = max((len(e.registry or "") for e in entries), default=0)
+    reg_w = min(reg_w, 50)
+    for e in entries:
+        lines.append(f"{e.short_id}  {(e.registry or ''):<{reg_w}}  {e.title}")
+    return "\n".join(lines)
+
+
+def _fmt_node_details(d: analysis.NodeDetails) -> str:
+    out = io.StringIO()
+    out.write(f"{d.short_id}  {d.title}\n")
+    out.write(f"  id:        {d.id}\n")
+    out.write(f"  registry:  {d.registry}  (registered={d.registered})\n")
+    out.write(f"  base_type: {d.base_type}\n")
+    out.write(f"  pos:       x={d.x} y={d.y} w={d.width} h={d.height}\n")
+    if d.properties:
+        out.write("  properties:\n")
+        for k, v in d.properties.items():
+            v_str = repr(v)
+            if len(v_str) > 100:
+                v_str = v_str[:97] + "..."
+            out.write(f"    {k} = {v_str}\n")
+    out.write("  inputs:\n")
+    if not d.inputs:
+        out.write("    (none)\n")
+    for inp in d.inputs:
+        if inp.connected and inp.source is not None:
+            out.write(
+                f"    [x] {inp.name:<20} <- {inp.source.short_id}.{inp.source.socket}\n"
+            )
+        else:
+            out.write(f"    [ ] {inp.name:<20} (unconnected)\n")
+    out.write("  outputs:\n")
+    if not d.outputs:
+        out.write("    (none)\n")
+    for op in d.outputs:
+        if not op.consumers:
+            out.write(f"    [ ] {op.name:<20} (no consumers)\n")
+        else:
+            out.write(f"    [x] {op.name:<20} ->\n")
+            for c in op.consumers:
+                out.write(f"          {c.short_id}.{c.socket}\n")
+    return out.getvalue().rstrip()
+
+
+def _fmt_edges(edges: list[analysis.EdgeInfo]) -> str:
+    if not edges:
+        return "(no edges)"
+    lines = []
+    for e in edges:
+        lines.append(
+            f"{e.source_short}.{e.source_socket}  ->  "
+            f"{e.target_short}.{e.target_socket}"
+        )
+    return "\n".join(lines)
+
+
+def _fmt_feeds(r: analysis.FeedsResult) -> str:
+    out = io.StringIO()
+    out.write(f"target: {r.target_short}.{r.target_socket}\n")
+    if r.source is None:
+        out.write("source: (unconnected)\n")
+    else:
+        out.write(f"source: {r.source.short_id}.{r.source.socket}\n")
+        out.write(f"  title:    {r.source_node_title}\n")
+        out.write(f"  registry: {r.source_node_registry}\n")
+    return out.getvalue().rstrip()
+
+
+def _fmt_consumers(r: analysis.ConsumersResult) -> str:
+    out = io.StringIO()
+    out.write(f"source: {r.source_short}.{r.source_socket}\n")
+    if not r.consumers:
+        out.write("consumers: (none)\n")
+    else:
+        out.write(f"consumers ({len(r.consumers)}):\n")
+        for c in r.consumers:
+            out.write(f"  -> {c.target_short}.{c.target_socket}\n")
+    return out.getvalue().rstrip()
+
+
+def _fmt_trace(t: analysis.TraceNode, *, indent: int = 0) -> str:
+    out = io.StringIO()
+    prefix = "  " * indent
+    via = f" ({t.via_socket})" if t.via_socket else ""
+    markers = []
+    if t.cycle:
+        markers.append("CYCLE")
+    if t.truncated:
+        markers.append("DEPTH")
+    marker_str = f" [{', '.join(markers)}]" if markers else ""
+    out.write(f"{prefix}{t.short_id}  {t.title}  <{t.registry}>{via}{marker_str}\n")
+    for child in t.children:
+        out.write(_fmt_trace(child, indent=indent + 1))
+    return out.getvalue()
+
+
+def _fmt_stage_map(r: analysis.StageMapResult) -> str:
+    out = io.StringIO()
+    out.write(f"chains: {len(r.chains)}\n")
+    for chain in r.chains:
+        out.write(
+            f"  chain {chain.index}: nodes={len(chain.node_ids)} "
+            f"stage_nodes={len(chain.stage_node_ids)} stages={chain.stages}\n"
+        )
+    out.write(f"stage nodes: {len(r.stage_nodes)}\n")
+    for s in r.stage_nodes:
+        out.write(
+            f"  chain {s.chain_index}  stage {s.stage:<4} {s.short_id}  {s.title}\n"
+        )
+    if r.unstaged_node_ids:
+        out.write(
+            f"unstaged components: {len(r.unstaged_node_ids)} node(s) "
+            f"in chains without a stage marker (default priority)\n"
+        )
+    return out.getvalue().rstrip()
+
+
+def _fmt_check_registries(r: analysis.RegistryCheckResult) -> str:
+    out = io.StringIO()
+    out.write(
+        f"checked: {r.checked_registry_count} registry strings, "
+        f"{r.checked_base_type_count} base_type strings\n"
+    )
+    if not r.unknown_registries and not r.unknown_base_types:
+        out.write("ok: no unknown registries or base types\n")
+        return out.getvalue().rstrip()
+    if r.unknown_registries:
+        out.write(f"unknown registries ({len(r.unknown_registries)}):\n")
+        for p in r.unknown_registries:
+            out.write(f"  {p.short_id}  {p.registry}  ({p.title})\n")
+    if r.unknown_base_types:
+        out.write(f"unknown base types ({len(r.unknown_base_types)}):\n")
+        for p in r.unknown_base_types:
+            out.write(f"  {p.short_id}  {p.base_type}  ({p.title})\n")
+    return out.getvalue().rstrip()
+
+
+# ---------------------------------------------------------------------------
+# Output dispatch
+# ---------------------------------------------------------------------------
+
+
+def _emit(
+    result: pydantic.BaseModel | list[pydantic.BaseModel],
+    *,
+    json_out: bool,
+    text_formatter: Callable[[Any], str],
+    out_stream: TextIO,
+) -> None:
+    if json_out:
+        if isinstance(result, list):
+            data = [r.model_dump(mode="json") for r in result]
+            out_stream.write(json.dumps(data, indent=2))
+        else:
+            out_stream.write(result.model_dump_json(indent=2))
+        out_stream.write("\n")
+        return
+    out_stream.write(text_formatter(result))
+    out_stream.write("\n")
+
+
+# ---------------------------------------------------------------------------
+# Argparse plumbing
+# ---------------------------------------------------------------------------
+
+
+def build_parser() -> argparse.ArgumentParser:
+    parser = argparse.ArgumentParser(
+        prog="python -m talemate.game.engine.nodes.tools",
+        description="Read-only static analysis for Talemate node graph JSON files.",
+    )
+    sub = parser.add_subparsers(dest="cmd", required=True)
+
+    def add_graph_arg(p: argparse.ArgumentParser) -> None:
+        p.add_argument("graph", help="Path to a node graph JSON file")
+        p.add_argument(
+            "--json", dest="json_out", action="store_true", help="Emit JSON output"
+        )
+
+    p_summary = sub.add_parser("summary", help="Top-level graph summary")
+    add_graph_arg(p_summary)
+
+    p_list = sub.add_parser("list-nodes", help="List nodes; filter by registry/title")
+    add_graph_arg(p_list)
+    p_list.add_argument("--registry", help="Substring filter on node registry")
+    p_list.add_argument("--title", help="Substring filter on node title")
+
+    p_node = sub.add_parser("node", help="Show details for a single node")
+    add_graph_arg(p_node)
+    p_node.add_argument("node_id", help="Node id (full or unique prefix)")
+
+    p_edges = sub.add_parser("edges", help="List edges, optionally filtered")
+    add_graph_arg(p_edges)
+    p_edges.add_argument("--from", dest="from_node", help="Filter by source node")
+    p_edges.add_argument("--to", dest="to_node", help="Filter by target node")
+
+    p_feeds = sub.add_parser("feeds", help="What feeds a given input socket?")
+    add_graph_arg(p_feeds)
+    p_feeds.add_argument("target", help="<node_id_or_prefix>.<socket>")
+
+    p_cons = sub.add_parser("consumers", help="What consumes a given output socket?")
+    add_graph_arg(p_cons)
+    p_cons.add_argument("source", help="<node_id_or_prefix>.<socket>")
+
+    p_trace = sub.add_parser("trace", help="Forward trace from a node")
+    add_graph_arg(p_trace)
+    p_trace.add_argument("start", help="Node id (full or unique prefix)")
+    p_trace.add_argument("--depth", type=int, default=3)
+
+    p_rtrace = sub.add_parser("rtrace", help="Backward trace from a node")
+    add_graph_arg(p_rtrace)
+    p_rtrace.add_argument("end", help="Node id (full or unique prefix)")
+    p_rtrace.add_argument("--depth", type=int, default=3)
+
+    p_stages = sub.add_parser("stages", help="Show core/Stage chains")
+    add_graph_arg(p_stages)
+
+    p_check = sub.add_parser(
+        "check-registries", help="Verify every registry string is known to the runtime"
+    )
+    add_graph_arg(p_check)
+
+    return parser
+
+
+def _run(args: argparse.Namespace, out_stream: TextIO, err_stream: TextIO) -> int:
+    # Ensure every Talemate node type is registered before any command
+    # runs. Loading once up-front gives consistent behavior across
+    # commands and surfaces loader errors before we start analyzing.
+    analysis.ensure_registry_loaded()
+
+    try:
+        graph = load_graph(args.graph)
+    except GraphLoadError as exc:
+        err_stream.write(f"error: {exc}\n")
+        return EXIT_STRUCT_ERROR
+
+    cmd = args.cmd
+    json_out = args.json_out
+
+    try:
+        if cmd == "summary":
+            result = analysis.summarize(graph)
+            _emit(result, json_out=json_out, text_formatter=_fmt_summary, out_stream=out_stream)
+
+        elif cmd == "list-nodes":
+            result = analysis.list_nodes(
+                graph,
+                registry_pattern=args.registry,
+                title_pattern=args.title,
+            )
+            _emit(
+                result,
+                json_out=json_out,
+                text_formatter=_fmt_list_nodes,
+                out_stream=out_stream,
+            )
+
+        elif cmd == "node":
+            result = analysis.get_node(graph, args.node_id)
+            _emit(
+                result,
+                json_out=json_out,
+                text_formatter=_fmt_node_details,
+                out_stream=out_stream,
+            )
+
+        elif cmd == "edges":
+            result = analysis.list_edges(
+                graph, from_node=args.from_node, to_node=args.to_node
+            )
+            _emit(result, json_out=json_out, text_formatter=_fmt_edges, out_stream=out_stream)
+
+        elif cmd == "feeds":
+            result = analysis.feeds(graph, args.target)
+            _emit(result, json_out=json_out, text_formatter=_fmt_feeds, out_stream=out_stream)
+
+        elif cmd == "consumers":
+            result = analysis.consumers(graph, args.source)
+            _emit(
+                result,
+                json_out=json_out,
+                text_formatter=_fmt_consumers,
+                out_stream=out_stream,
+            )
+
+        elif cmd == "trace":
+            result = analysis.trace_forward(graph, args.start, depth=args.depth)
+            _emit(result, json_out=json_out, text_formatter=_fmt_trace, out_stream=out_stream)
+
+        elif cmd == "rtrace":
+            result = analysis.trace_backward(graph, args.end, depth=args.depth)
+            _emit(result, json_out=json_out, text_formatter=_fmt_trace, out_stream=out_stream)
+
+        elif cmd == "stages":
+            result = analysis.stage_map(graph)
+            _emit(
+                result,
+                json_out=json_out,
+                text_formatter=_fmt_stage_map,
+                out_stream=out_stream,
+            )
+
+        elif cmd == "check-registries":
+            result = analysis.check_registries(graph)
+            _emit(
+                result,
+                json_out=json_out,
+                text_formatter=_fmt_check_registries,
+                out_stream=out_stream,
+            )
+            if result.has_problems:
+                return EXIT_REGISTRY_PROBLEMS
+
+        else:  # pragma: no cover - argparse enforces this
+            err_stream.write(f"error: unknown command {cmd}\n")
+            return EXIT_STRUCT_ERROR
+
+    except ValueError as exc:
+        err_stream.write(f"error: {exc}\n")
+        return EXIT_STRUCT_ERROR
+
+    return EXIT_OK
+
+
+def main(argv: list[str] | None = None) -> int:
+    parser = build_parser()
+    args = parser.parse_args(argv)
+    return _run(args, sys.stdout, sys.stderr)
+
+
+if __name__ == "__main__":  # pragma: no cover
+    sys.exit(main())

src/talemate/game/engine/nodes/tools/layout.py

@@ -0,0 +1,803 @@
+"""
+Deterministic auto-layout for Talemate node graph JSON dicts.
+
+The goal is a *readable* placement of freshly added nodes — not aesthetic
+perfection. The algorithm is a stage-stratified vertical-band sweep: each
+weakly-connected component of the target set is classified by the node
+registries it contains, sorted vertically (Input chains on top, Stage
+chains in numeric order, passthrough middle chains, Output chains at the
+bottom), and then laid out horizontally using a topological depth sweep
+with greedy row assignment to avoid overlaps.
+
+Node heights are NOT read from the writer — the writer intentionally
+leaves them unset and layout estimates them from socket / property counts
+via ``_estimate_height``. This keeps collision avoidance honest for nodes
+that actually render tall.
+
+Usage::
+
+    from talemate.game.engine.nodes.tools import layout_graph
+
+    layout_graph(graph, new_node_ids=[id_a, id_b], anchor="right")
+
+``anchor`` values:
+
+* ``"right"`` — place the target set to the right of the existing bounding
+  box, sharing the existing ``min_y``.
+* ``"below"`` — place the target set below the existing bounding box,
+  sharing the existing ``min_x``.
+* ``"full"`` — relayout every node in the graph, starting at ``(0, 0)``.
+
+Notes:
+
+* Nodes outside the target set are never moved and never resized.
+* Heights of nodes inside the target set are always overwritten with an
+  estimate — the contract is "if you ask layout to position a node, it
+  also sizes it."
+* Groups, comments, and any non-node canvas elements are not touched.
+* ``extends`` is not resolved — only the raw JSON of this graph is
+  considered.
+"""
+
+from __future__ import annotations
+
+import enum
+from collections import defaultdict, deque
+from dataclasses import dataclass, field
+from itertools import groupby
+from typing import Iterable
+
+import pydantic
+import structlog
+
+from . import writer as writer_mod
+from .writer import DEFAULT_NODE_HEIGHT, DEFAULT_NODE_WIDTH
+
+log = structlog.get_logger("talemate.game.engine.nodes.tools.layout")
+
+__all__ = [
+    "layout_graph",
+    "LayoutError",
+    "LayoutOptions",
+    "WCCKind",
+]
+
+# Default canvas metrics — chosen to match the rough shapes of shipped
+# graphs. ``col_width = 360`` gives wires ~150 px of horizontal breathing
+# room between a node's right edge (default node width 210) and the
+# next column's left edge.
+#
+# ``DEFAULT_NODE_WIDTH`` / ``DEFAULT_NODE_HEIGHT`` are imported from
+# ``writer`` so both modules agree on a single source of truth. The
+# height fallback is used only when a node outside the target set is
+# missing a ``height`` for bounding-box math; nodes inside the target
+# set are always sized by ``_estimate_height``.
+DEFAULT_COL_WIDTH = 360
+DEFAULT_RIGHT_GAP = 300
+DEFAULT_BELOW_GAP = 200
+
+
+class LayoutError(ValueError):
+    """Raised for malformed layout requests."""
+
+
+class LayoutOptions(pydantic.BaseModel):
+    """Tweakable metrics for the layout pass (defaults are fine)."""
+
+    col_width: int = DEFAULT_COL_WIDTH
+    right_gap: int = DEFAULT_RIGHT_GAP
+    below_gap: int = DEFAULT_BELOW_GAP
+    # Vertical gap between stage bands after the super-band merge.
+    band_gap: int = 120
+    # Height-aware packing: minimum vertical gap between two nodes
+    # stacked in the same column.
+    min_vertical_gap: int = 50
+    # Height estimation formula inputs.
+    title_bar_height: int = 20
+    socket_row_height: int = 22
+    padding: int = 0
+    min_height: int = 60
+    # Extra vertical space for the +/- buttons that DynamicSocketNodeBase
+    # subclasses render at the bottom of the node.
+    dynamic_socket_bonus: int = 30
+
+
+# ---------------------------------------------------------------------------
+# WCC classification
+# ---------------------------------------------------------------------------
+
+
+class WCCKind(enum.Enum):
+    """The vertical band a weakly-connected component belongs to."""
+
+    TOP = "top"        # has core/Input, no core/Stage
+    STAGE = "stage"    # has core/Stage
+    MIDDLE = "middle"  # no Input, no Output, no Stage
+    BOTTOM = "bottom"  # has core/Output, no core/Input, no core/Stage
+
+
+@dataclass
+class WCCInfo:
+    """One weakly-connected component and its classification."""
+
+    nodes: list[str]
+    kind: WCCKind
+    stage_value: int | None = None  # populated when kind == STAGE
+
+
+# ---------------------------------------------------------------------------
+# Graph accessors (private)
+# ---------------------------------------------------------------------------
+
+
+def _nodes(graph: dict) -> dict[str, dict]:
+    return graph.get("nodes") or {}
+
+
+def _edges(graph: dict) -> dict[str, list[str]]:
+    return graph.get("edges") or {}
+
+
+def _split_edge_node(endpoint: str) -> str | None:
+    if "." not in endpoint:
+        return None
+    return endpoint.split(".", 1)[0]
+
+
+def _node_rect(node: dict) -> tuple[int, int, int, int]:
+    x = int(node.get("x", 0) or 0)
+    y = int(node.get("y", 0) or 0)
+    w = int(node.get("width", 0) or 0) or DEFAULT_NODE_WIDTH
+    h = int(node.get("height", 0) or 0) or DEFAULT_NODE_HEIGHT
+    return x, y, w, h
+
+
+def _existing_bounds(
+    graph: dict, target_set: set[str]
+) -> tuple[int, int, int, int]:
+    """Return (min_x, min_y, max_x, max_y) for nodes NOT in the target set.
+
+    If the "outside" set is empty (e.g. full relayout, or every node is
+    targeted) the bounds collapse to ``(0, 0, 0, 0)``.
+    """
+
+    xs: list[int] = []
+    ys: list[int] = []
+    maxes_x: list[int] = []
+    maxes_y: list[int] = []
+    for nid, node in _nodes(graph).items():
+        if nid in target_set:
+            continue
+        x, y, w, h = _node_rect(node)
+        xs.append(x)
+        ys.append(y)
+        maxes_x.append(x + w)
+        maxes_y.append(y + h)
+    if not xs:
+        return (0, 0, 0, 0)
+    return (min(xs), min(ys), max(maxes_x), max(maxes_y))
+
+
+# ---------------------------------------------------------------------------
+# Height estimation
+# ---------------------------------------------------------------------------
+
+
+def _estimate_height(graph_node: dict, options: LayoutOptions) -> int:
+    """Compute an estimated render height for a node from its class metadata.
+
+    Formula::
+
+        title_bar_height
+          + (max(num_inputs, num_outputs) + num_properties) * socket_row_height
+          + padding
+          + dynamic_socket_bonus (if the class is a DynamicSocketNodeBase)
+
+    Properties render as full widget rows **below** the socket area in
+    LiteGraph, stacking vertically with the socket rows rather than
+    being a small per-property bonus. Dynamic-socket nodes render
+    ``+``/``-`` buttons at the bottom for adding/removing inputs; those
+    take ~30 px and apply whether or not the instance currently has any
+    dynamic inputs (the buttons are a class capability).
+
+    Floored at ``options.min_height``. If the node's registry is not in
+    the live ``NODES`` dict, falls back to ``options.min_height``.
+
+    ``num_inputs`` includes any graph-level ``dynamic_inputs`` on the
+    node, because dynamic inputs also render a socket row.
+    """
+
+    registry = graph_node.get("registry")
+    if not registry:
+        return options.min_height
+    try:
+        meta = writer_mod.get_node_metadata(registry)
+    except writer_mod.UnknownRegistryError:
+        return options.min_height
+
+    num_static_inputs = len(meta.inputs)
+    num_outputs = len(meta.outputs)
+    num_properties = len(meta.properties)
+
+    dyn = graph_node.get("dynamic_inputs") or []
+    num_dynamic_inputs = len(dyn)
+    num_inputs = num_static_inputs + num_dynamic_inputs
+
+    rows = max(num_inputs, num_outputs) + num_properties
+    estimated = (
+        options.title_bar_height
+        + rows * options.socket_row_height
+        + options.padding
+    )
+    if meta.is_dynamic:
+        estimated += options.dynamic_socket_bonus
+    return max(estimated, options.min_height)
+
+
+def _apply_estimated_heights(
+    graph: dict, target_ids: set[str], options: LayoutOptions
+) -> None:
+    """Set ``height`` on each target node to an estimated value.
+
+    Always overwrites the height for nodes in the target set — the
+    contract is "if you ask layout to position a node, it also sizes it."
+    Nodes outside the target set are not touched.
+    """
+
+    nodes = _nodes(graph)
+    for nid in target_ids:
+        node = nodes.get(nid)
+        if node is None:
+            continue
+        node["height"] = _estimate_height(node, options)
+
+
+# ---------------------------------------------------------------------------
+# Weakly-connected components over a subset
+# ---------------------------------------------------------------------------
+
+
+def _wcc_over_subset(
+    graph: dict, subset: set[str]
+) -> list[list[str]]:
+    """Compute weakly-connected components restricted to ``subset``.
+
+    Only edges with both endpoints in ``subset`` contribute to adjacency;
+    edges leaving the subset (bridges to the outside world) don't merge
+    components for our purposes. A node in ``subset`` with no internal
+    edges becomes a singleton component.
+
+    Returns a list of components, each a sorted ``list[str]`` of node ids.
+    The outer list is sorted by the lexicographically-smallest node id of
+    each component for deterministic iteration order.
+    """
+
+    parent: dict[str, str] = {nid: nid for nid in subset}
+
+    def find(x: str) -> str:
+        while parent[x] != x:
+            parent[x] = parent[parent[x]]
+            x = parent[x]
+        return x
+
+    def union(a: str, b: str) -> None:
+        ra, rb = find(a), find(b)
+        if ra != rb:
+            parent[ra] = rb
+
+    for src_key, targets in _edges(graph).items():
+        s_node = _split_edge_node(src_key)
+        if s_node is None or s_node not in subset:
+            continue
+        for tgt in targets:
+            t_node = _split_edge_node(tgt)
+            if t_node is None or t_node not in subset:
+                continue
+            union(s_node, t_node)
+
+    buckets: dict[str, list[str]] = defaultdict(list)
+    for nid in subset:
+        buckets[find(nid)].append(nid)
+
+    components = [sorted(comp) for comp in buckets.values()]
+    components.sort(key=lambda comp: comp[0] if comp else "")
+    return components
+
+
+# ---------------------------------------------------------------------------
+# Component classification
+# ---------------------------------------------------------------------------
+
+
+def _classify_wcc(graph: dict, node_ids: list[str]) -> WCCInfo:
+    """Inspect the registries of the nodes in a component and pick a kind.
+
+    Precedence when multiple flags apply::
+
+        STAGE > TOP > BOTTOM > MIDDLE
+
+    Rationale: a chain containing both an ``Input`` and a ``Stage`` is
+    part of that Stage's band, not the top band. A chain with both
+    ``Output`` and ``Input`` is a TOP (passthrough — the Input anchors
+    it). A chain with ``Output`` only and no ``Stage`` is BOTTOM.
+    """
+
+    nodes = _nodes(graph)
+    has_input = False
+    has_output = False
+    stage_values: list[int] = []
+
+    for nid in node_ids:
+        node = nodes.get(nid) or {}
+        registry = node.get("registry") or ""
+        if registry == "core/Input":
+            has_input = True
+        elif registry == "core/Output":
+            has_output = True
+        elif registry == "core/Stage":
+            stage_prop = (node.get("properties") or {}).get("stage", 0)
+            try:
+                stage_values.append(int(stage_prop))
+            except (TypeError, ValueError):
+                stage_values.append(0)
+
+    if stage_values:
+        return WCCInfo(
+            nodes=node_ids,
+            kind=WCCKind.STAGE,
+            stage_value=min(stage_values),
+        )
+    if has_input:
+        return WCCInfo(nodes=node_ids, kind=WCCKind.TOP)
+    if has_output:
+        return WCCInfo(nodes=node_ids, kind=WCCKind.BOTTOM)
+    return WCCInfo(nodes=node_ids, kind=WCCKind.MIDDLE)
+
+
+def _band_group_key(info: WCCInfo) -> tuple:
+    """Grouping key for the band-merging step in ``layout_graph``.
+
+    Two WCCs share a key (and therefore get merged into one super-band)
+    when they belong to the same vertical category — e.g. all BOTTOM
+    components form one bottom band, all STAGE components with the same
+    stage number form one stage band. This eliminates the fragmentation
+    that arises when a graph has multiple WCCs of the same kind, which
+    is common when each ``GetState\u2192Output`` emission pair is its own
+    tiny 2-node WCC.
+
+    The key shape mirrors the first two elements of ``_wcc_sort_key`` so
+    that consecutive same-keyed entries are guaranteed to be adjacent in
+    the sorted list (a precondition for ``itertools.groupby``).
+    """
+
+    if info.kind is WCCKind.TOP:
+        return (0, 0)
+    if info.kind is WCCKind.STAGE:
+        return (1, info.stage_value)  # _classify_wcc guarantees non-None here
+    if info.kind is WCCKind.MIDDLE:
+        return (2, 0)
+    return (3, 0)  # BOTTOM
+
+
+def _wcc_sort_key(info: WCCInfo) -> tuple:
+    """Vertical ordering key: TOP, STAGE(0), STAGE(1), ..., MIDDLE, BOTTOM.
+
+    Within the same band, ties break on the lexicographically-smallest
+    node id so the order is stable.
+    """
+
+    smallest = info.nodes[0] if info.nodes else ""
+    if info.kind is WCCKind.TOP:
+        return (0, 0, smallest)
+    if info.kind is WCCKind.STAGE:
+        return (1, info.stage_value, smallest)  # STAGE kind guarantees non-None
+    if info.kind is WCCKind.MIDDLE:
+        return (2, 0, smallest)
+    return (3, 0, smallest)  # BOTTOM
+
+
+# ---------------------------------------------------------------------------
+# Topological column sweep for a single band
+# ---------------------------------------------------------------------------
+
+
+def _topological_depths(
+    target_ids: set[str], internal_edges: list[tuple[str, str]]
+) -> dict[str, int]:
+    """Kahn's algorithm; isolated nodes get depth 0.
+
+    A cycle (if any — the writer's ``connect()`` rejects these, but raw
+    JSON could still have them) short-circuits: remaining nodes fall
+    back to depth 0 to keep the layout robust.
+    """
+
+    adj: dict[str, set[str]] = defaultdict(set)
+    indeg: dict[str, int] = {nid: 0 for nid in target_ids}
+    for s, t in internal_edges:
+        if t not in adj[s]:
+            adj[s].add(t)
+            indeg[t] = indeg.get(t, 0) + 1
+
+    depths: dict[str, int] = {nid: 0 for nid in target_ids}
+    queue: deque[str] = deque(nid for nid, d in indeg.items() if d == 0)
+    while queue:
+        nid = queue.popleft()
+        for child in adj.get(nid, ()):
+            candidate = depths[nid] + 1
+            if candidate > depths[child]:
+                depths[child] = candidate
+            indeg[child] -= 1
+            if indeg[child] == 0:
+                queue.append(child)
+
+    # Any nodes left with indeg > 0 are part of a cycle; leave them at 0
+    # rather than dropping them.
+    return depths
+
+
+def _rects_overlap(
+    a: tuple[int, int, int, int], b: tuple[int, int, int, int]
+) -> bool:
+    ax, ay, aw, ah = a
+    bx, by, bw, bh = b
+    return not (ax + aw <= bx or bx + bw <= ax or ay + ah <= by or by + bh <= ay)
+
+
+@dataclass
+class _BandPlacement:
+    """Internal result of placing a single band."""
+
+    rects: list[tuple[int, int, int, int]] = field(default_factory=list)
+    max_y: int = 0  # highest y + h across all placed nodes; 0 for empty
+
+
+def _place_band(
+    graph: dict,
+    band_nodes: list[str],
+    *,
+    origin_x: int,
+    origin_y: int,
+    options: LayoutOptions,
+) -> _BandPlacement:
+    """Place one weakly-connected component as a topological column sweep.
+
+    All nodes in ``band_nodes`` are positioned starting at
+    ``(origin_x, origin_y)``.
+
+    Packing is height-aware (per-column ``y_cursor``) and
+    predecessor-aware (Sugiyama-layer barycenter heuristic): for each
+    column past the first, unplaced nodes are sorted by the average y
+    of their already-placed predecessors in earlier columns, so a child
+    tends to land near the row of its parent(s). This eliminates the
+    diamond wire crossings that a naive stable-id ordering would cause.
+    """
+
+    nodes = _nodes(graph)
+    band_set = set(band_nodes)
+
+    # Internal edges for topological sort (both endpoints in the band).
+    internal_edges: list[tuple[str, str]] = []
+    for src_key, targets in _edges(graph).items():
+        s_node = _split_edge_node(src_key)
+        if s_node is None or s_node not in band_set:
+            continue
+        for tgt in targets:
+            t_node = _split_edge_node(tgt)
+            if t_node is None or t_node not in band_set:
+                continue
+            if s_node == t_node:
+                continue
+            internal_edges.append((s_node, t_node))
+
+    depths = _topological_depths(band_set, internal_edges)
+
+    # Predecessor adjacency (parent set per node, restricted to the band).
+    predecessors: dict[str, set[str]] = {nid: set() for nid in band_set}
+    for s, t in internal_edges:
+        predecessors[t].add(s)
+
+    # Group by column (depth), nodes sorted by id for stable initial order.
+    by_depth: dict[int, list[str]] = defaultdict(list)
+    for nid in sorted(band_set):
+        by_depth[depths.get(nid, 0)].append(nid)
+
+    placement = _BandPlacement()
+
+    # Per-column y cursor: next free y position for that column.
+    column_cursors: dict[int, int] = {}
+    # Track placed y positions so predecessor-aware sorting can average
+    # over already-assigned rows.
+    placed_y: dict[str, int] = {}
+
+    # Build outgoing-edge index for the entry-point sort key (column 0),
+    # restricted to edges whose BOTH endpoints live in this band. Any
+    # edge that leaves the band is irrelevant to the BFS-to-deepest walk
+    # and filtering it here keeps the inner loop tight.
+    outgoing: dict[str, list[tuple[str, str]]] = defaultdict(list)
+    for src_key, targets in _edges(graph).items():
+        s_node = _split_edge_node(src_key)
+        if s_node is None or s_node not in band_set:
+            continue
+        for tgt in targets:
+            t_node, _, t_socket = tgt.partition(".")
+            if t_node and t_socket and t_node in band_set:
+                outgoing[s_node].append((t_node, t_socket))
+
+    def _input_socket_index(target_node_id: str, target_socket: str) -> int:
+        """Return the index of ``target_socket`` in the target node's input
+        list (static inputs from the class metadata followed by per-instance
+        ``dynamic_inputs``). Falls back to a large sentinel when the socket
+        can't be resolved, so it sorts to the end without crashing.
+        """
+        node = nodes.get(target_node_id) or {}
+        registry = node.get("registry") or ""
+        if not registry:
+            return 999
+        try:
+            meta = writer_mod.get_node_metadata(registry)
+        except writer_mod.UnknownRegistryError:
+            return 999
+        names = list(meta.inputs)
+        for d in node.get("dynamic_inputs") or []:
+            name = d.get("name") if isinstance(d, dict) else None
+            if name:
+                names.append(name)
+        try:
+            return names.index(target_socket)
+        except ValueError:
+            return 999
+
+    def _entry_point_sort_key(nid: str) -> tuple:
+        """Sort key for column-0 (no-predecessor) nodes.
+
+        Priority order (highest-priority case wins):
+
+        1. **``core/Input`` entry point** — sorted by its own ``num``
+           property, so module inputs appear in interface-declaration
+           order (``IN x`` with ``num=1`` sits above ``IN y`` with
+           ``num=2``). Predecessor-aware placement then aligns each
+           input's ``SET local.x`` counterpart next to it, producing
+           horizontal IN-to-SET wires.
+        2. **Deepest-descendant is ``core/Output``** — sorted by the
+           descendant's ``num`` property. Makes the bottom emission band's
+           ``GET\u2192Output`` pairs land in interface-declaration order.
+        3. **General deepest-descendant case** — sorted by
+           ``(descendant_id, socket_index)``. When entry-point chains
+           converge on a sink (e.g. four ``IN``\u2192``SET`` chains feeding a
+           ``Stage 0`` marker), siblings share their first sort component
+           and differ only on socket order, so predecessor-aware placement
+           downstream eliminates cross-column wire crossings into the
+           convergence node.
+        4. **No outgoing edges** — fall back to stable id order at the
+           end of the sort.
+
+        Cases 1 and 2 use string-prefixed tuple components (``"__input__"``
+        / ``"__output__"``) to keep them in distinct sort spaces so
+        cross-kind ordering stays predictable.
+        """
+
+        # Case 1: the entry point is itself a core/Input — use its num.
+        node = nodes.get(nid) or {}
+        if node.get("registry") == "core/Input":
+            num_val = (node.get("properties") or {}).get("num", 0)
+            try:
+                num_int = int(num_val)
+            except (TypeError, ValueError):
+                num_int = 0
+            return (0, ("__input__", num_int), nid)
+
+        # BFS forward from nid, tracking depth. ``outgoing`` is already
+        # pre-filtered to in-band edges, so no per-hop band check.
+        visited: set[str] = {nid}
+        bfs: deque[tuple[str, int]] = deque([(nid, 0)])
+        candidates: list[tuple[int, str, int]] = []  # (depth, target_id, socket_idx)
+
+        while bfs:
+            current, depth = bfs.popleft()
+            for t_node, t_socket in outgoing.get(current, ()):
+                candidates.append(
+                    (depth + 1, t_node, _input_socket_index(t_node, t_socket))
+                )
+                if t_node not in visited:
+                    visited.add(t_node)
+                    bfs.append((t_node, depth + 1))
+
+        if not candidates:
+            return (1, ("", 0), nid)
+
+        # Pick the deepest reachable (descendant, socket). Tie-break on
+        # lex-smallest (descendant_id, socket_index) so siblings that
+        # converge on the same descendant share their first sort component
+        # and differ only on socket order.
+        candidates.sort(key=lambda c: (-c[0], c[1], c[2]))
+        _, desc_id, socket_idx = candidates[0]
+
+        # Case 2: deepest descendant is a core/Output — use its num
+        # property (interface-declaration order).
+        desc_node = nodes.get(desc_id) or {}
+        if desc_node.get("registry") == "core/Output":
+            num_val = (desc_node.get("properties") or {}).get("num", 0)
+            try:
+                num_int = int(num_val)
+            except (TypeError, ValueError):
+                num_int = 0
+            return (0, ("__output__", num_int), nid)
+
+        # Case 3: general deepest-descendant case.
+        return (0, (desc_id, socket_idx), nid)
+
+    for depth in sorted(by_depth):
+        column_x = origin_x + depth * options.col_width
+        candidates = by_depth[depth]
+
+        if depth == 0:
+            # First column: no predecessors, so we can't barycenter on
+            # parents. Sort entry points by where their wires attach on
+            # their successors instead — see ``_entry_point_sort_key``.
+            ordered = sorted(candidates, key=_entry_point_sort_key)
+        else:
+            # Sort by preferred y = average y of placed predecessors in
+            # earlier columns. Nodes with no placed predecessor go to
+            # the end. Secondary key is the stable id so ties are
+            # deterministic.
+            def _preferred(nid: str) -> tuple[int, int, str]:
+                placed_parents = [
+                    placed_y[p]
+                    for p in predecessors.get(nid, ())
+                    if p in placed_y
+                ]
+                if placed_parents:
+                    avg = sum(placed_parents) // len(placed_parents)
+                    return (0, avg, nid)
+                return (1, 0, nid)
+
+            ordered = sorted(candidates, key=_preferred)
+
+        for nid in ordered:
+            node = nodes[nid]
+            _, _, w, h = _node_rect(node)
+
+            # Preferred y from predecessors (ignored in column 0).
+            preferred_y: int | None = None
+            if depth > 0:
+                placed_parents = [
+                    placed_y[p]
+                    for p in predecessors.get(nid, ())
+                    if p in placed_y
+                ]
+                if placed_parents:
+                    preferred_y = sum(placed_parents) // len(placed_parents)
+
+            cursor = column_cursors.get(depth, origin_y)
+            if preferred_y is None:
+                target_y = cursor
+            else:
+                # Never go above the cursor (that would shuffle the
+                # order of already-placed nodes in this column).
+                target_y = max(preferred_y, cursor)
+
+            candidate = (column_x, target_y, w, h)
+            # Height-aware collision check against anything already
+            # placed in the band (protects against cross-column overlap
+            # from the wider col_width or adjustment quirks).
+            while any(_rects_overlap(candidate, r) for r in placement.rects):
+                target_y += options.min_vertical_gap
+                candidate = (column_x, target_y, w, h)
+
+            cx, cy, cw, ch = candidate
+            node["x"] = cx
+            node["y"] = cy
+            placement.rects.append(candidate)
+            placed_y[nid] = cy
+            column_cursors[depth] = cy + ch + options.min_vertical_gap
+            if cy + ch > placement.max_y:
+                placement.max_y = cy + ch
+
+    return placement
+
+
+# ---------------------------------------------------------------------------
+# Public entry point
+# ---------------------------------------------------------------------------
+
+
+def layout_graph(
+    graph: dict,
+    *,
+    new_node_ids: Iterable[str] | None = None,
+    anchor: str = "right",
+    options: LayoutOptions | None = None,
+) -> None:
+    """Mutate ``graph`` so the target node set is placed without overlap.
+
+    Parameters
+    ----------
+    graph:
+        Raw graph dict (as loaded from ``loader.load_graph``). Mutated
+        in place — nothing is returned.
+    new_node_ids:
+        Exact set of nodes to reposition. If ``None``, every node
+        currently sitting at ``(0, 0)`` is treated as needing placement.
+    anchor:
+        ``"right"``, ``"below"`` or ``"full"``. See module docstring.
+    options:
+        Optional :class:`LayoutOptions` overriding canvas metrics.
+    """
+
+    opts = options or LayoutOptions()
+    if anchor not in ("right", "below", "full"):
+        raise LayoutError(
+            f"Unknown anchor '{anchor}'; expected 'right', 'below', or 'full'."
+        )
+
+    nodes = _nodes(graph)
+    if not nodes:
+        return
+
+    # Resolve the target set.
+    if anchor == "full":
+        target_ids = set(nodes.keys())
+    elif new_node_ids is not None:
+        target_ids = set(new_node_ids)
+        missing = target_ids - nodes.keys()
+        if missing:
+            raise LayoutError(
+                f"Target set references unknown node ids: {sorted(missing)}"
+            )
+    else:
+        target_ids = {
+            nid
+            for nid, n in nodes.items()
+            if int(n.get("x", 0) or 0) == 0 and int(n.get("y", 0) or 0) == 0
+        }
+
+    if not target_ids:
+        return
+
+    # Step 1: estimate heights for every target node BEFORE we read any
+    # rectangle for collision math.
+    _apply_estimated_heights(graph, target_ids, opts)
+
+    # Step 2: pick the anchor origin based on where everything else is.
+    min_x, min_y, max_x, max_y = _existing_bounds(graph, target_ids)
+    if anchor == "right":
+        anchor_x = max_x + opts.right_gap if max_x else 0
+        anchor_y = min_y
+    elif anchor == "below":
+        anchor_x = min_x
+        anchor_y = max_y + opts.below_gap if max_y else 0
+    else:  # full
+        anchor_x = 0
+        anchor_y = 0
+
+    # Step 3: compute weakly-connected components of the target set and
+    # classify each one so we can stack them vertically.
+    components = _wcc_over_subset(graph, target_ids)
+    wcc_infos = [_classify_wcc(graph, comp) for comp in components]
+    wcc_infos.sort(key=_wcc_sort_key)
+
+    # Step 4: place each band, stacking downward. WCCs of the same kind
+    # (e.g. multiple BOTTOM components, or both halves of a single STAGE)
+    # are merged into one super-band so they read as one cohesive group
+    # rather than as multiple fragmented bands separated by ``band_gap``.
+    # The y-cursor starts at anchor_y and advances by (band_height +
+    # band_gap) after each merged band.
+    y_cursor = anchor_y
+    for _key, group_iter in groupby(wcc_infos, key=_band_group_key):
+        group = list(group_iter)
+        # Flatten all nodes from all WCCs in this group into a single
+        # band. ``_place_band`` builds its own internal topology, so
+        # passing nodes from multiple disconnected sub-WCCs just means
+        # each sub-WCC contributes its own column structure to the band.
+        band_nodes: list[str] = [nid for info in group for nid in info.nodes]
+        placement = _place_band(
+            graph,
+            band_nodes,
+            origin_x=anchor_x,
+            origin_y=y_cursor,
+            options=opts,
+        )
+        if not placement.rects:
+            continue
+        band_height = placement.max_y - y_cursor
+        y_cursor = y_cursor + band_height + opts.band_gap

src/talemate/game/engine/nodes/tools/loader.py

@@ -0,0 +1,227 @@
+"""
+File-I/O choke point for the node-graph analysis tools.
+
+This module also handles **scene-module auto-registration**: when you
+load a graph file that lives under ``scenes/<name>/nodes/``, every
+sibling ``*.json`` in the same directory is registered into the live
+``NODES`` dict so that any ``add_node("...")`` or static-analysis call
+that references one of those scene-level modules can resolve it. This
+removes the footgun where ``ensure_registry_loaded()`` only walks the
+shipped ``SEARCH_PATHS`` and silently doesn't see scene modules.
+"""
+
+from __future__ import annotations
+
+import json
+from pathlib import Path
+from typing import Iterable
+
+import structlog
+
+from talemate.game.engine.nodes.registry import NODES, import_node_definition
+
+from .analysis import ensure_registry_loaded
+
+__all__ = ["load_graph", "GraphLoadError", "register_scene_modules"]
+
+log = structlog.get_logger("talemate.game.engine.nodes.tools.loader")
+
+
+class GraphLoadError(ValueError):
+    """Raised when a graph file is missing or unparseable."""
+
+
+def load_graph(
+    path: str | Path,
+    *,
+    extra_module_paths: Iterable[str | Path] | None = None,
+) -> dict:
+    """Load and parse a node graph JSON file into a plain dict.
+
+    Raises ``GraphLoadError`` for missing files or malformed JSON. The
+    returned dict is the raw JSON contents - we deliberately do NOT run
+    it through any pydantic model so analysis remains side-effect-free.
+
+    **Scene-module auto-registration** runs as a side effect:
+
+    * If ``path`` matches the pattern ``**/scenes/<name>/nodes/<file>.json``,
+      every sibling ``*.json`` in that directory is registered into the
+      live ``NODES`` dict via ``register_scene_modules``. This means the
+      writer and analysis layer can introspect scene-level modules that
+      the standard ``ensure_registry_loaded()`` walk doesn't reach.
+    * If ``extra_module_paths`` is given, every ``*.json`` under each of
+      those paths is also registered. Use this for non-standard layouts.
+
+    Auto-registration failures are logged at warn-level and never raise —
+    the loader's primary job is still to load the requested file.
+    """
+
+    p = Path(path)
+    if not p.exists():
+        raise GraphLoadError(f"Graph file not found: {p}")
+    if not p.is_file():
+        raise GraphLoadError(f"Not a regular file: {p}")
+    try:
+        with p.open("r", encoding="utf-8") as fh:
+            data = json.load(fh)
+    except json.JSONDecodeError as exc:
+        raise GraphLoadError(f"Failed to parse graph JSON {p}: {exc}") from exc
+
+    if not isinstance(data, dict):
+        raise GraphLoadError(
+            f"Graph JSON {p} did not parse to an object (got {type(data).__name__})"
+        )
+
+    # Auto-register scene modules when loading from a scene path. This
+    # fires AFTER parsing the requested file (so a malformed file still
+    # raises predictably) but BEFORE returning, so downstream callers
+    # see the populated registry.
+    _maybe_autoregister_scene_modules(p)
+    if extra_module_paths:
+        for extra in extra_module_paths:
+            register_scene_modules(extra)
+
+    return data
+
+
+def register_scene_modules(directory: str | Path) -> int:
+    """Walk ``directory`` for ``*.json`` files and register each as a node
+    type in the live ``NODES`` dict. Returns the number of modules newly
+    registered. Already-registered registries are skipped.
+
+    Files without a top-level ``registry`` key are silently ignored
+    (they aren't node modules — they may be scene shells, asset metadata,
+    or other JSON in the same dir).
+
+    The standard node registry (``ensure_registry_loaded()``) is always
+    loaded first — scene modules typically reference standard node types
+    like ``validation/ValidateValueIsSet`` and ``core/MakeBool`` in their
+    interior, and ``import_node_definition`` validates the full graph at
+    registration time. Without the standard registry loaded first, every
+    scene module that uses any standard node would fail to validate.
+
+    A retry loop handles inter-module dependencies inside the directory:
+    if module A references module B and the alphabetic walk hits A first,
+    A fails its first attempt because B isn't registered yet, then B
+    succeeds, then A retries and succeeds.
+
+    Failures after retries are logged but never raised. Useful when you
+    have node modules in a non-standard location and want them visible
+    to the writer/analysis layer.
+
+    Why not just call ``registry.import_scene_node_definitions``?
+        The native function requires a live ``Scene`` object and stores
+        its classes in a per-scene container (``scene._NODE_DEFINITIONS``)
+        rather than the global ``NODES`` dict. This helper exists because
+        the tools package runs in contexts (CLI subprocesses, tests, one-
+        shot scripts) where there is no active scene, and it needs the
+        registered classes in the **global** ``NODES`` dict so the writer
+        and analysis layer can introspect them. Additionally, this
+        version rolls back half-registered classes between retry attempts
+        (``@register`` fires synchronously at class definition, and a
+        failed ``node.model_validate(node_data)`` leaves a broken class
+        behind) — the native scene loader has that same latent issue but
+        masks it by wiping its per-scene container on the next load.
+        Consolidating these via a shared lower-level helper in
+        ``registry.py`` is a worthwhile refactor but out of scope here.
+    """
+
+    d = Path(directory)
+    if not d.is_dir():
+        return 0
+
+    # Make sure the standard registry is loaded before we try to validate
+    # scene modules — otherwise references to standard types will fail.
+    ensure_registry_loaded()
+
+    # Collect all candidate module entries first so we can iterate with
+    # a retry loop for inter-module dependencies.
+    candidates: list[tuple[Path, dict, str]] = []
+    for json_path in sorted(d.glob("*.json")):
+        try:
+            with json_path.open("r", encoding="utf-8") as fh:
+                module_data = json.load(fh)
+        except (json.JSONDecodeError, OSError) as exc:
+            log.debug("scene_module_skipped", path=str(json_path), reason=str(exc))
+            continue
+        if not isinstance(module_data, dict):
+            continue
+        registry = module_data.get("registry")
+        if not registry:
+            continue
+        if registry in NODES:
+            continue
+        candidates.append((json_path, module_data, registry))
+
+    registered = 0
+    last_errors: dict[str, str] = {}
+
+    # Retry loop: keep trying as long as at least one module made progress
+    # in the previous pass. ``import_node_definition`` registers the class
+    # via ``@register`` synchronously and THEN validates by instantiating,
+    # so a failed validation leaves a half-broken class in NODES. We
+    # explicitly remove failed registries between attempts so retries get
+    # a clean slate.
+    pending = list(candidates)
+    progress = True
+    while pending and progress:
+        progress = False
+        next_pending: list[tuple[Path, dict, str]] = []
+        for entry in pending:
+            json_path, module_data, registry = entry
+            try:
+                import_node_definition(module_data, NODES)
+                registered += 1
+                progress = True
+                last_errors.pop(registry, None)
+            except Exception as exc:  # noqa: BLE001 — see docstring
+                # Roll back the half-registered class so the next retry
+                # attempt re-creates it cleanly.
+                NODES.pop(registry, None)
+                last_errors[registry] = str(exc)
+                next_pending.append(entry)
+        pending = next_pending
+
+    # Anything left in `pending` failed every attempt — log a single
+    # summary warning per registry rather than spamming once per attempt.
+    for json_path, _module_data, registry in pending:
+        log.warning(
+            "scene_module_import_failed",
+            path=str(json_path),
+            registry=registry,
+            err=last_errors.get(registry, "unknown"),
+        )
+
+    if registered:
+        # Clear the writer's metadata cache so any newly-registered
+        # registry gets re-probed on its next ``get_node_metadata`` call.
+        # Lazy to break the writer -> loader import cycle.
+        from .writer import clear_metadata_cache
+
+        clear_metadata_cache()
+
+    return registered
+
+
+def _maybe_autoregister_scene_modules(graph_path: Path) -> None:
+    """If ``graph_path`` looks like ``**/scenes/<name>/nodes/*.json``,
+    register every JSON sibling in the same directory.
+
+    The pattern match is intentionally strict: it requires the immediate
+    parent directory to be named ``nodes`` AND its grandparent to be a
+    direct child of a directory named ``scenes``. This avoids accidentally
+    walking unrelated ``scenes/`` directories elsewhere in the filesystem.
+    """
+
+    try:
+        resolved = graph_path.resolve()
+    except OSError:
+        return
+    parts = resolved.parts
+    if len(parts) < 4:
+        return
+    if parts[-2] != "nodes":
+        return
+    if parts[-4] != "scenes":
+        return
+    register_scene_modules(resolved.parent)

src/talemate/game/engine/nodes/tools/writer.py

@@ -0,0 +1,986 @@
+"""
+Programmatic mutation API for Talemate node graph JSON dicts.
+
+The functions here take an already-loaded ``graph: dict`` and mutate it in
+place — the mirror image of ``analysis.py``, which is strictly read-only.
+Pure primitive functions are the core surface; ``GraphWriter`` is an
+ergonomic class that wraps them and adds file I/O.
+
+Note on node sizes: the writer does NOT assign a default ``height`` to new
+nodes. Height belongs to the layout pass (``tools.layout.layout_graph``),
+which estimates it from the node's socket and property counts so collision
+avoidance actually has a realistic rectangle to work with. Callers that
+need a pinned height can still pass one explicitly. Width is still set to
+a fixed default because every node in the repo uses the same canvas width.
+
+Node metadata (static inputs, outputs, properties, ``base_type``) is read
+from the live ``NODES`` class registry via a lightweight in-process cache
+that instantiates each class once. We deliberately avoid
+``registry.export_node_definitions()`` for this cache because that helper
+calls ``PropertyField.model_dump()``, which in turn calls user-supplied
+``generate_choices`` callables — several of those touch runtime-only state
+(the websocket handler, the active scene, etc.) and crash outside a live
+Talemate session. The instance-based probe here is the same pattern
+``analysis.py`` uses in ``_static_socket_names`` and matches the
+frontend-facing exporter's own instantiation step.
+
+Known limitations:
+
+* The writer does NOT keep the graph's top-level ``inputs`` / ``outputs``
+  / ``module_properties`` mirror arrays in sync when interior Input /
+  Output / ModuleProperty nodes are added or removed. Those arrays are
+  computed at runtime from the interior nodes, but some shipped graphs
+  (notably director actions) serialise them anyway. Callers that care
+  about those arrays should regenerate them manually.
+* The writer does NOT resolve ``extends``. It operates on raw JSON.
+* The writer does NOT check socket *type* compatibility; many sockets in
+  the codebase are typed ``any`` and strict type checks would fight real
+  graphs. It only checks that sockets exist and that fan-in is at most 1.
+"""
+
+from __future__ import annotations
+
+import json
+import uuid
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Any, Iterable
+
+import structlog
+
+from . import analysis
+from .analysis import ensure_registry_loaded, resolve_node_id, split_socket_path
+from .loader import load_graph
+
+log = structlog.get_logger("talemate.game.engine.nodes.tools.writer")
+
+__all__ = [
+    # class
+    "GraphWriter",
+    # pure primitives
+    "add_node",
+    "remove_node",
+    "connect",
+    "disconnect",
+    "add_dynamic_input",
+    "remove_dynamic_input",
+    "add_group",
+    # metadata helpers
+    "get_node_metadata",
+    "clear_metadata_cache",
+    "NodeMetadata",
+    "GroupColor",
+    # exceptions
+    "WriterError",
+    "UnknownRegistryError",
+    "UnknownSocketError",
+    "UnknownPropertyError",
+    "AlreadyConnectedError",
+    "NotConnectedError",
+    "CycleError",
+    "DynamicInputError",
+    "NodeNotFoundError",
+    "GroupError",
+]
+
+
+# ---------------------------------------------------------------------------
+# Group color palette — matches the LiteGraph presets exposed in the
+# frontend's "Create Group from Selection" submenu (see
+# ``talemate_frontend/src/utils/recentNodes.js``). Each constant is the
+# preset's ``groupcolor`` hex value, which is what gets stored on the
+# group's ``color`` field in the graph JSON. Use these by name for
+# readability — the agent picks which group gets which color based on
+# the cluster's purpose.
+# ---------------------------------------------------------------------------
+
+
+class GroupColor:
+    """Named group-color constants matching the LiteGraph preset palette.
+
+    These are the same colors the frontend's right-click "Create Group
+    from Selection" menu offers, so auto-generated groups visually match
+    the convention used by hand-built graphs.
+    """
+
+    INPUT = "#88A"           # blue       — input collection / passthrough
+    OUTPUT = "#8A8"          # green      — output emission
+    PROCESS = "#3f789e"      # pale_blue  — main computation stages
+    PREPARE = "#8AA"         # cyan       — setup / preconditions
+    VALIDATION = "#b58b2a"   # yellow     — validation / guards
+    FUNCTION = "#b06634"     # brown      — function definitions
+    SPECIAL = "#a1309b"      # purple     — helpers / orphans / one-offs
+    ERROR_HANDLING = "#A88"  # red        — error-handler chains
+    UX = "#207e7e"           # teal       — user-facing UX nodes
+
+
+# Default canvas dimensions used when a node in ``add_group``'s target
+# set hasn't been sized by the layout pass yet (rare — call layout first).
+DEFAULT_NODE_WIDTH = 210
+DEFAULT_NODE_HEIGHT = 100
+
+
+# Matches the frontend's group-creation padding constants (see
+# ``talemate_frontend/src/utils/groupInteractions.js``). Auto-groups
+# should look identical to user-created ones, so we mirror the same
+# numbers rather than picking our own.
+GROUP_PADDING = 25
+GROUP_TOP_PADDING = 45     # padding + 20
+GROUP_TITLE_HEIGHT = 20    # LiteGraph.NODE_TITLE_HEIGHT default
+GROUP_MIN_WIDTH = 140
+GROUP_MIN_HEIGHT = 80
+GROUP_DEFAULT_FONT_SIZE = 24
+
+
+# ---------------------------------------------------------------------------
+# Exceptions
+# ---------------------------------------------------------------------------
+
+
+class WriterError(Exception):
+    """Base class for all writer-layer errors."""
+
+
+class UnknownRegistryError(WriterError):
+    """A registry path is not present in the live ``NODES`` dict."""
+
+
+class UnknownSocketError(WriterError):
+    """A named socket does not exist on the target node."""
+
+
+class AlreadyConnectedError(WriterError):
+    """A target input socket already has a source connection."""
+
+
+class NotConnectedError(WriterError):
+    """``disconnect`` was asked to remove an edge that does not exist."""
+
+
+class CycleError(WriterError):
+    """The requested ``connect`` would introduce a cycle in the graph."""
+
+
+class GroupError(WriterError):
+    """``add_group`` was given an invalid set of node ids or empty input."""
+
+
+class UnknownPropertyError(WriterError):
+    """A property key passed to ``add_node`` does not exist on the
+    node class. Sister to ``UnknownSocketError`` — same kind of typo
+    catch, but for the ``properties`` dict instead of socket names.
+    """
+
+
+class DynamicInputError(WriterError):
+    """A dynamic-input operation was attempted on an incompatible node."""
+
+
+class NodeNotFoundError(WriterError):
+    """A node id was not present in the graph."""
+
+
+# ---------------------------------------------------------------------------
+# Node metadata cache
+# ---------------------------------------------------------------------------
+
+
+@dataclass(frozen=True)
+class NodeMetadata:
+    """Static metadata for a registered node class, cached per process.
+
+    ``inputs`` / ``outputs`` are socket names in declaration order (the
+    ones the class sets up in ``setup()``, before any graph-specific
+    dynamic inputs). ``properties`` is the tuple of class-declared
+    property names (from ``instance.properties`` after ``setup()``).
+    ``base_type`` is the class-level ``_base_type`` / ``base_type``
+    computed field. ``is_dynamic`` is True when the class is a subclass
+    of ``DynamicSocketNodeBase``.
+    """
+
+    registry: str
+    base_type: str
+    inputs: tuple[str, ...]
+    outputs: tuple[str, ...]
+    properties: tuple[str, ...]
+    is_dynamic: bool
+
+
+_METADATA_CACHE: dict[str, NodeMetadata] = {}
+
+
+def _dynamic_socket_base_cls() -> type:
+    from talemate.game.engine.nodes.core.dynamic import DynamicSocketNodeBase
+
+    return DynamicSocketNodeBase
+
+
+def get_node_metadata(registry: str) -> NodeMetadata:
+    """Return cached ``NodeMetadata`` for a registry path.
+
+    Raises ``UnknownRegistryError`` if the registry is not present in
+    the live ``NODES`` dict.
+    """
+
+    cached = _METADATA_CACHE.get(registry)
+    if cached is not None:
+        return cached
+
+    ensure_registry_loaded()
+    from talemate.game.engine.nodes.registry import NODES
+
+    node_cls = NODES.get(registry)
+    if node_cls is None:
+        raise UnknownRegistryError(
+            f"Registry '{registry}' is not registered. "
+            f"(Did you forget to call ensure_registry_loaded() or import "
+            f"the module that declares it?)"
+        )
+
+    try:
+        instance = node_cls()
+    except Exception as exc:  # pragma: no cover - defensive
+        raise UnknownRegistryError(
+            f"Registry '{registry}' is registered but the class "
+            f"({node_cls.__name__}) could not be instantiated: {exc}"
+        ) from exc
+
+    try:
+        inputs = tuple(s.name for s in instance.inputs)
+    except Exception:
+        inputs = tuple()
+    try:
+        outputs = tuple(s.name for s in instance.outputs)
+    except Exception:
+        outputs = tuple()
+    try:
+        properties = tuple((instance.properties or {}).keys())
+    except Exception:
+        properties = tuple()
+
+    base_type = ""
+    try:
+        base_type = instance.base_type  # computed_field on NodeBase
+    except Exception:
+        base_type = getattr(node_cls, "_base_type", "") or ""
+
+    try:
+        is_dynamic = isinstance(instance, _dynamic_socket_base_cls())
+    except Exception:  # pragma: no cover - defensive
+        is_dynamic = False
+
+    meta = NodeMetadata(
+        registry=registry,
+        base_type=base_type or "",
+        inputs=inputs,
+        outputs=outputs,
+        properties=properties,
+        is_dynamic=is_dynamic,
+    )
+    _METADATA_CACHE[registry] = meta
+    return meta
+
+
+def clear_metadata_cache() -> None:
+    """Flush the node-metadata cache.
+
+    Called by the loader after registering scene-level modules so that
+    their socket layouts are re-probed on next access. Also useful from
+    tests that want to observe fresh probes.
+    """
+
+    _METADATA_CACHE.clear()
+
+
+# ---------------------------------------------------------------------------
+# Graph accessors
+# ---------------------------------------------------------------------------
+
+
+def _nodes(graph: dict) -> dict[str, dict]:
+    nodes = graph.get("nodes")
+    if nodes is None:
+        nodes = {}
+        graph["nodes"] = nodes
+    return nodes
+
+
+def _edges(graph: dict) -> dict[str, list[str]]:
+    edges = graph.get("edges")
+    if edges is None:
+        edges = {}
+        graph["edges"] = edges
+    return edges
+
+
+def _require_node(graph: dict, node_id: str) -> dict:
+    nodes = _nodes(graph)
+    if node_id not in nodes:
+        raise NodeNotFoundError(f"Node id '{node_id}' is not in the graph")
+    return nodes[node_id]
+
+
+def _dynamic_input_names(node: dict) -> list[str]:
+    return [
+        d.get("name")
+        for d in (node.get("dynamic_inputs") or [])
+        if d.get("name")
+    ]
+
+
+def _input_names_for_node(node: dict) -> list[str]:
+    """Inputs = class-declared inputs plus any dynamic_inputs on this node.
+
+    Duplicates are removed while preserving declaration order (class
+    inputs first, then dynamic inputs).
+    """
+
+    registry = node.get("registry")
+    static_inputs: tuple[str, ...] = tuple()
+    if registry:
+        try:
+            static_inputs = get_node_metadata(registry).inputs
+        except UnknownRegistryError:
+            static_inputs = tuple()
+
+    dyn = _dynamic_input_names(node)
+    seen: set[str] = set()
+    out: list[str] = []
+    for name in list(static_inputs) + dyn:
+        if name and name not in seen:
+            seen.add(name)
+            out.append(name)
+    return out
+
+
+def _output_names_for_node(node: dict) -> list[str]:
+    registry = node.get("registry")
+    if not registry:
+        return []
+    try:
+        return list(get_node_metadata(registry).outputs)
+    except UnknownRegistryError:
+        return []
+
+
+# ---------------------------------------------------------------------------
+# Pure primitives
+# ---------------------------------------------------------------------------
+
+
+def add_node(
+    graph: dict,
+    registry: str,
+    *,
+    title: str | None = None,
+    properties: dict[str, Any] | None = None,
+    x: int = 0,
+    y: int = 0,
+    width: int = DEFAULT_NODE_WIDTH,
+    height: int | None = None,
+) -> str:
+    """Add a new node to ``graph`` and return its new UUID.
+
+    The new id is always a freshly generated ``uuid4``. The registry
+    must be present in the live ``NODES`` dict, otherwise
+    ``UnknownRegistryError`` is raised.
+
+    Property keys in ``properties`` are validated against the node
+    class's declared ``Fields`` — an unknown / typo'd key raises
+    ``UnknownPropertyError`` with a helpful message listing the valid
+    property names. This is the property-side equivalent of how
+    ``connect()`` validates socket names.
+
+    ``height`` defaults to ``None`` — the writer intentionally does not
+    pin a height on new nodes because height should be estimated by
+    the layout pass from the node's socket / property counts. Pass an
+    explicit integer to override and pin the height.
+    """
+
+    meta = get_node_metadata(registry)
+
+    # Validate property keys against the class metadata. We deliberately
+    # validate even an empty/None properties arg early in the function
+    # so the error fires before any partial mutation happens.
+    if properties:
+        valid = set(meta.properties)
+        unknown = [k for k in properties if k not in valid]
+        if unknown:
+            unknown_repr = ", ".join(repr(k) for k in unknown)
+            valid_repr = ", ".join(repr(k) for k in sorted(valid)) or "(none)"
+            raise UnknownPropertyError(
+                f"Node {registry!r} has no property/properties "
+                f"{unknown_repr}. Valid properties: {valid_repr}."
+            )
+
+    new_id = str(uuid.uuid4())
+    node: dict[str, Any] = {
+        "title": title if title is not None else registry.split("/")[-1],
+        "id": new_id,
+        "properties": dict(properties or {}),
+        "x": int(x),
+        "y": int(y),
+        "width": int(width),
+        "collapsed": False,
+        "inherited": False,
+        "registry": registry,
+        "base_type": meta.base_type,
+    }
+    if height is not None:
+        node["height"] = int(height)
+    if meta.is_dynamic:
+        # Mirror the shape shipped graphs use: an empty list is valid and
+        # makes future add_dynamic_input calls ergonomic.
+        node["dynamic_inputs"] = []
+
+    _nodes(graph)[new_id] = node
+    return new_id
+
+
+def remove_node(graph: dict, node_id: str) -> None:
+    """Remove a node and sweep every edge that references it.
+
+    The edge sweep removes:
+
+    * any edge key whose source node is ``node_id``
+    * any edge target entry whose target node is ``node_id``, pruning
+      now-empty target lists (and therefore their edge keys)
+    """
+
+    nodes = _nodes(graph)
+    if node_id not in nodes:
+        raise NodeNotFoundError(f"Node id '{node_id}' is not in the graph")
+
+    del nodes[node_id]
+
+    edges = _edges(graph)
+    # Drop entire edge keys whose source is this node
+    to_drop = [
+        src for src in edges if src.split(".", 1)[0] == node_id
+    ]
+    for src in to_drop:
+        del edges[src]
+
+    # For remaining edges, filter out targets pointing at this node.
+    empty_keys: list[str] = []
+    for src, targets in edges.items():
+        kept = [
+            tgt
+            for tgt in targets
+            if tgt.split(".", 1)[0] != node_id
+        ]
+        if len(kept) != len(targets):
+            edges[src] = kept
+        if not edges[src]:
+            empty_keys.append(src)
+    for key in empty_keys:
+        del edges[key]
+
+
+def _target_has_source(graph: dict, target_node: str, target_socket: str) -> bool:
+    full_tgt = f"{target_node}.{target_socket}"
+    for _src, targets in _edges(graph).items():
+        if full_tgt in targets:
+            return True
+    return False
+
+
+def connect(
+    graph: dict,
+    source_id: str,
+    source_socket: str,
+    target_id: str,
+    target_socket: str,
+) -> None:
+    """Connect ``source_id.source_socket`` to ``target_id.target_socket``.
+
+    Validates that:
+
+    * both nodes exist in the graph
+    * the source output socket exists on the source node's class
+    * the target input socket exists (class-declared or dynamic) on the
+      target node
+    * the target input is not already connected (fan-in is at most 1)
+    * adding the edge would not introduce a cycle
+
+    On cycle detection the edge is rolled back before ``CycleError`` is
+    raised.
+    """
+
+    source_node = _require_node(graph, source_id)
+    target_node = _require_node(graph, target_id)
+
+    source_outputs = _output_names_for_node(source_node)
+    if source_socket not in source_outputs:
+        raise UnknownSocketError(
+            f"Source node {source_id} ({source_node.get('registry')}) has no "
+            f"output socket '{source_socket}'. Known outputs: "
+            f"{source_outputs or '(none)'}"
+        )
+
+    target_inputs = _input_names_for_node(target_node)
+    if target_socket not in target_inputs:
+        raise UnknownSocketError(
+            f"Target node {target_id} ({target_node.get('registry')}) has no "
+            f"input socket '{target_socket}'. Known inputs: "
+            f"{target_inputs or '(none)'}"
+        )
+
+    if _target_has_source(graph, target_id, target_socket):
+        raise AlreadyConnectedError(
+            f"Target input {target_id}.{target_socket} is already connected. "
+            f"Disconnect the existing source before reassigning."
+        )
+
+    edges = _edges(graph)
+    src_key = f"{source_id}.{source_socket}"
+    tgt_value = f"{target_id}.{target_socket}"
+
+    targets_list = edges.setdefault(src_key, [])
+    targets_list.append(tgt_value)
+
+    # Cycle check AFTER adding, so we include the new edge; roll back if
+    # it would form a cycle.
+    if _would_cycle_including_new(graph, source_id, target_id):
+        targets_list.remove(tgt_value)
+        if not targets_list:
+            del edges[src_key]
+        raise CycleError(
+            f"Connecting {source_id}.{source_socket} -> "
+            f"{target_id}.{target_socket} would introduce a cycle."
+        )
+
+
+def _would_cycle_including_new(
+    graph: dict, source_node: str, target_node: str
+) -> bool:
+    """Cycle check based on current edge state (edge already inserted).
+
+    With the new edge in place, a cycle exists iff there is a path from
+    ``target_node`` back to ``source_node`` (counting the new edge as
+    part of adjacency). Self-loops are cycles.
+    """
+
+    if source_node == target_node:
+        return True
+
+    # Build node-level adjacency from every current edge. Edges are
+    # ``<node_id>.<socket>`` strings; we only care about the node halves.
+    adj: dict[str, set[str]] = {}
+    for src, targets in _edges(graph).items():
+        s_node = src.split(".", 1)[0]
+        for tgt in targets:
+            t_node = tgt.split(".", 1)[0]
+            adj.setdefault(s_node, set()).add(t_node)
+
+    # Walk forward from target_node; if we reach source_node we have a
+    # cycle because source -> ... -> target -> source.
+    stack = list(adj.get(target_node, ()))
+    seen: set[str] = set()
+    while stack:
+        current = stack.pop()
+        if current == source_node:
+            return True
+        if current in seen:
+            continue
+        seen.add(current)
+        stack.extend(adj.get(current, ()))
+    return False
+
+
+def disconnect(
+    graph: dict,
+    source_id: str,
+    source_socket: str,
+    target_id: str,
+    target_socket: str,
+) -> None:
+    """Remove the edge ``source_id.source_socket -> target_id.target_socket``.
+
+    Raises ``NotConnectedError`` if the edge is not present.
+    """
+
+    edges = _edges(graph)
+    src_key = f"{source_id}.{source_socket}"
+    tgt_value = f"{target_id}.{target_socket}"
+
+    targets = edges.get(src_key)
+    if not targets or tgt_value not in targets:
+        raise NotConnectedError(
+            f"No edge {source_id}.{source_socket} -> "
+            f"{target_id}.{target_socket} to remove."
+        )
+    targets.remove(tgt_value)
+    if not targets:
+        del edges[src_key]
+
+
+def add_dynamic_input(
+    graph: dict,
+    node_id: str,
+    name: str,
+    socket_type: str = "*",
+) -> None:
+    """Append a ``dynamic_inputs`` entry on a ``DynamicSocketNodeBase`` node.
+
+    Raises ``DynamicInputError`` if the node's registered class is not a
+    subclass of ``DynamicSocketNodeBase``, or if a dynamic input with
+    ``name`` already exists.
+
+    The default ``socket_type`` is ``"*"`` — the LiteGraph **wildcard**
+    socket type that accepts a connection from any other socket regardless
+    of type. This is the correct default for collector-style nodes like
+    ``data/DictCollector``, ``data/ListCollector``, ``data/string/AdvancedFormat``,
+    and ``data/string/Jinja2Format``, which all want to accept inputs of
+    any type. **Do NOT use ``"any"`` as the type** — despite the name,
+    ``"any"`` is a specific named type in LiteGraph, not a wildcard, and
+    the frontend's socket validation refuses to connect a typed output
+    (e.g. ``int``, ``str``) to an ``"any"`` input. Only use ``"*"``.
+    """
+
+    node = _require_node(graph, node_id)
+    registry = node.get("registry")
+    if not registry:
+        raise DynamicInputError(
+            f"Node {node_id} has no registry; cannot determine whether it "
+            f"supports dynamic inputs."
+        )
+
+    meta = get_node_metadata(registry)
+    if not meta.is_dynamic:
+        raise DynamicInputError(
+            f"Node {node_id} ({registry}) is not a DynamicSocketNodeBase "
+            f"subclass; it does not accept dynamic inputs."
+        )
+
+    dyn = node.setdefault("dynamic_inputs", [])
+    if any(d.get("name") == name for d in dyn):
+        raise DynamicInputError(
+            f"Dynamic input '{name}' already exists on node {node_id}."
+        )
+    dyn.append({"name": name, "type": socket_type})
+
+
+def remove_dynamic_input(graph: dict, node_id: str, name: str) -> None:
+    """Remove a dynamic input entry and sweep any edges using it."""
+
+    node = _require_node(graph, node_id)
+    dyn = node.get("dynamic_inputs") or []
+    new_dyn = [d for d in dyn if d.get("name") != name]
+    if len(new_dyn) == len(dyn):
+        raise DynamicInputError(
+            f"No dynamic input named '{name}' on node {node_id}."
+        )
+    node["dynamic_inputs"] = new_dyn
+
+    # Sweep any edges targeting this now-removed socket.
+    edges = _edges(graph)
+    tgt_value = f"{node_id}.{name}"
+    empty_keys: list[str] = []
+    for src, targets in edges.items():
+        if tgt_value in targets:
+            targets.remove(tgt_value)
+        if not targets:
+            empty_keys.append(src)
+    for key in empty_keys:
+        del edges[key]
+
+
+# ---------------------------------------------------------------------------
+# Group creation
+# ---------------------------------------------------------------------------
+
+
+def add_group(
+    graph: dict,
+    title: str,
+    color: str,
+    node_ids: Iterable[str],
+    *,
+    font_size: int = GROUP_DEFAULT_FONT_SIZE,
+) -> dict:
+    """Append a colored, titled group around the given nodes' bounding box.
+
+    Computes the union bounding box of every node in ``node_ids`` (using
+    each node's ``x``/``y``/``width``/``height``) and then expands it by
+    the same padding constants the frontend's "Create Group from
+    Selection" menu uses, so the resulting group is visually identical
+    to a user-created one.
+
+    Parameters
+    ----------
+    graph:
+        The graph dict being mutated.
+    title:
+        Display title for the group. Use a short human-readable label
+        (e.g. ``"Input"``, ``"Stage 0"``, ``"Output"``).
+    color:
+        The group's border / title color, as a hex string. Use one of
+        the constants on :class:`GroupColor` to match the frontend's
+        preset palette.
+    node_ids:
+        The nodes the group should encompass. Must contain at least one
+        valid node id. Short-prefix ids are NOT resolved here — pass
+        full ids (use ``GraphWriter.add_group`` for prefix support).
+    font_size:
+        Title font size. Defaults to 24, which matches every existing
+        group in shipped graphs.
+
+    Returns
+    -------
+    dict
+        The newly-created group dict (which has also been appended to
+        ``graph["groups"]`` in place). Caller can mutate further if
+        needed.
+
+    Raises
+    ------
+    GroupError
+        If ``node_ids`` is empty or any id does not exist in the graph.
+
+    Notes
+    -----
+    Call this **after** running ``layout_graph``. Group bounding boxes
+    are computed from the nodes' actual positions, so positioning the
+    nodes first is required.
+    """
+
+    nodes = _nodes(graph)
+    ids = list(node_ids)
+    if not ids:
+        raise GroupError("add_group requires at least one node id.")
+
+    rects: list[tuple[int, int, int, int]] = []
+    for nid in ids:
+        node = nodes.get(nid)
+        if node is None:
+            raise GroupError(f"add_group: unknown node id {nid!r}.")
+        x = int(node.get("x", 0) or 0)
+        y = int(node.get("y", 0) or 0)
+        w = int(node.get("width", DEFAULT_NODE_WIDTH) or DEFAULT_NODE_WIDTH)
+        h = int(node.get("height", DEFAULT_NODE_HEIGHT) or DEFAULT_NODE_HEIGHT)
+        rects.append((x, y, w, h))
+
+    min_x = min(r[0] for r in rects)
+    min_y = min(r[1] for r in rects)
+    max_x = max(r[0] + r[2] for r in rects)
+    max_y = max(r[1] + r[3] for r in rects)
+
+    # Mirror the frontend's group-creation padding (groupInteractions.js).
+    pos_x = min_x - GROUP_PADDING
+    pos_y = min_y - GROUP_TOP_PADDING - GROUP_TITLE_HEIGHT
+    raw_width = (max_x - min_x) + GROUP_PADDING * 2
+    raw_height = (max_y - min_y) + GROUP_PADDING + GROUP_TOP_PADDING + GROUP_TITLE_HEIGHT
+    width = max(GROUP_MIN_WIDTH, raw_width)
+    height = max(GROUP_MIN_HEIGHT, raw_height)
+
+    group_dict = {
+        "title": title,
+        "x": pos_x,
+        "y": pos_y,
+        "width": width,
+        "height": height,
+        "color": color,
+        "font_size": font_size,
+        "inherited": False,
+    }
+
+    groups = graph.setdefault("groups", [])
+    groups.append(group_dict)
+    return group_dict
+
+
+# ---------------------------------------------------------------------------
+# GraphWriter ergonomics
+# ---------------------------------------------------------------------------
+
+
+class GraphWriter:
+    """Ergonomic wrapper around the pure mutation primitives.
+
+    Holds a graph dict plus an optional source path for ``save()``. All
+    mutation methods delegate to the module-level functions above — the
+    class is sugar, not logic.
+    """
+
+    def __init__(self, graph: dict, *, path: str | Path | None = None) -> None:
+        self.graph = graph
+        self.path: Path | None = Path(path) if path is not None else None
+
+    # -- construction ------------------------------------------------------
+
+    @classmethod
+    def load(
+        cls,
+        path: str | Path,
+        *,
+        extra_module_paths: Iterable[str | Path] | None = None,
+    ) -> "GraphWriter":
+        """Load a graph JSON file from disk and wrap it.
+
+        If ``path`` lives under ``scenes/<name>/nodes/``, sibling JSON
+        modules in that directory are auto-registered into the live
+        ``NODES`` dict so the writer can introspect them when you
+        reference one via ``add_node("...")``. Pass ``extra_module_paths``
+        to also walk additional directories — useful for non-standard
+        layouts where related modules live elsewhere.
+        """
+
+        graph = load_graph(path, extra_module_paths=extra_module_paths)
+        return cls(graph, path=path)
+
+    def save(self, path: str | Path | None = None, *, indent: int = 2) -> Path:
+        """Persist the graph to disk, returning the path written to.
+
+        If ``path`` is omitted the writer uses the path it was loaded
+        from; if the writer was constructed without a path and no
+        argument is passed this raises ``WriterError``.
+        """
+
+        target = Path(path) if path is not None else self.path
+        if target is None:
+            raise WriterError(
+                "GraphWriter.save() needs an explicit path because this "
+                "writer was not constructed via GraphWriter.load()."
+            )
+        target.parent.mkdir(parents=True, exist_ok=True)
+        with target.open("w", encoding="utf-8") as fh:
+            json.dump(self.graph, fh, indent=indent)
+            fh.write("\n")
+        self.path = target
+        return target
+
+    # -- mutation ----------------------------------------------------------
+
+    def add_node(
+        self,
+        registry: str,
+        *,
+        title: str | None = None,
+        properties: dict[str, Any] | None = None,
+        x: int = 0,
+        y: int = 0,
+        width: int = DEFAULT_NODE_WIDTH,
+        height: int | None = None,
+    ) -> str:
+        return add_node(
+            self.graph,
+            registry,
+            title=title,
+            properties=properties,
+            x=x,
+            y=y,
+            width=width,
+            height=height,
+        )
+
+    def remove_node(self, node_id: str) -> None:
+        remove_node(self.graph, self._resolve(node_id))
+
+    def connect(self, *args: str) -> None:
+        """Connect two sockets.
+
+        Two call shapes are supported:
+
+        * ``writer.connect(src_id, src_socket, tgt_id, tgt_socket)``
+        * ``writer.connect("<src_id>.<src_socket>", "<tgt_id>.<tgt_socket>")``
+        """
+
+        src_id, src_socket, tgt_id, tgt_socket = self._parse_edge_args(args)
+        connect(
+            self.graph,
+            self._resolve(src_id),
+            src_socket,
+            self._resolve(tgt_id),
+            tgt_socket,
+        )
+
+    def disconnect(self, *args: str) -> None:
+        """Disconnect two sockets. Same call shapes as :meth:`connect`."""
+
+        src_id, src_socket, tgt_id, tgt_socket = self._parse_edge_args(args)
+        disconnect(
+            self.graph,
+            self._resolve(src_id),
+            src_socket,
+            self._resolve(tgt_id),
+            tgt_socket,
+        )
+
+    def add_dynamic_input(
+        self, node_id: str, name: str, socket_type: str = "*"
+    ) -> None:
+        add_dynamic_input(
+            self.graph, self._resolve(node_id), name, socket_type=socket_type
+        )
+
+    def remove_dynamic_input(self, node_id: str, name: str) -> None:
+        remove_dynamic_input(self.graph, self._resolve(node_id), name)
+
+    def add_group(
+        self,
+        title: str,
+        color: str,
+        node_ids: Iterable[str],
+        *,
+        font_size: int = GROUP_DEFAULT_FONT_SIZE,
+    ) -> dict:
+        """Append a colored, titled group around the given nodes.
+
+        Same as the pure :func:`add_group` but resolves short-prefix
+        node ids transparently. **Call this after layout** — group
+        bounding boxes are computed from the nodes' actual positions.
+
+        Pick a ``color`` from :class:`GroupColor` (or pass a literal hex
+        string). The agent decides which color matches the cluster's
+        purpose; common conventions:
+
+        * ``GroupColor.INPUT`` for input-collection / passthrough bands
+        * ``GroupColor.OUTPUT`` for output-emission bands
+        * ``GroupColor.PROCESS`` for main computation stages
+        * ``GroupColor.VALIDATION`` for guard / validation chains
+        * ``GroupColor.SPECIAL`` for helpers and one-offs
+        """
+
+        resolved = [self._resolve(nid) for nid in node_ids]
+        return add_group(
+            self.graph,
+            title=title,
+            color=color,
+            node_ids=resolved,
+            font_size=font_size,
+        )
+
+    # -- lookup helpers ----------------------------------------------------
+
+    def resolve(self, node_id_or_prefix: str) -> str:
+        """Resolve a short-prefix node id to a full id."""
+
+        return self._resolve(node_id_or_prefix)
+
+    def _resolve(self, node_id_or_prefix: str) -> str:
+        try:
+            return resolve_node_id(self.graph, node_id_or_prefix)
+        except ValueError as exc:
+            raise NodeNotFoundError(str(exc)) from exc
+
+    @staticmethod
+    def _parse_edge_args(
+        args: tuple[str, ...],
+    ) -> tuple[str, str, str, str]:
+        if len(args) == 4:
+            return args  # type: ignore[return-value]
+        if len(args) == 2:
+            src_node, src_socket = split_socket_path(args[0])
+            tgt_node, tgt_socket = split_socket_path(args[1])
+            return src_node, src_socket, tgt_node, tgt_socket
+        raise TypeError(
+            "connect/disconnect expects either 4 string args "
+            "(src_id, src_socket, tgt_id, tgt_socket) or 2 dotted strings "
+            "('src_id.socket', 'tgt_id.socket')."
+        )
+
+    # -- introspection -----------------------------------------------------
+
+    def summarize(self) -> analysis.GraphSummary:
+        return analysis.summarize(self.graph)

tests/test_nodegraph_tools.py

@@ -0,0 +1,382 @@
+"""
+Tests for ``talemate.game.engine.nodes.tools`` static analysis CLI/library.
+"""
+
+from __future__ import annotations
+
+import io
+import json
+from pathlib import Path
+
+import pytest
+
+from talemate.game.engine.nodes.tools import analysis, cli
+from talemate.game.engine.nodes.tools.loader import GraphLoadError, load_graph
+
+REPO_ROOT = Path(__file__).resolve().parent.parent
+DIRECTOR_FIXTURE = (
+    REPO_ROOT
+    / "src/talemate/agents/director/modules/director-action-direct-story-arc.json"
+)
+
+
+# ---------------------------------------------------------------------------
+# Inline fixture graph
+# ---------------------------------------------------------------------------
+
+
+def _make_inline_graph() -> dict:
+    """Hand-built tiny graph used for deterministic unit tests.
+
+    Layout:
+
+        a (core/MakeBool) ---value---> b (core/Stage stage=2)
+                                       |
+                                       state---> c (core/Watch)
+        d (mystery/Unknown)            (no edges)
+    """
+
+    return {
+        "title": "Inline Test Graph",
+        "id": "00000000-0000-0000-0000-000000000000",
+        "registry": "test/InlineGraph",
+        "base_type": "core/Graph",
+        "extends": None,
+        "properties": {},
+        "nodes": {
+            "aaaa1111-aaaa-aaaa-aaaa-aaaaaaaaaaaa": {
+                "title": "true",
+                "id": "aaaa1111-aaaa-aaaa-aaaa-aaaaaaaaaaaa",
+                "properties": {"value": True},
+                "registry": "core/MakeBool",
+                "base_type": "core/Node",
+            },
+            "bbbb2222-bbbb-bbbb-bbbb-bbbbbbbbbbbb": {
+                "title": "Stage 2",
+                "id": "bbbb2222-bbbb-bbbb-bbbb-bbbbbbbbbbbb",
+                "properties": {"stage": 2},
+                "registry": "core/Stage",
+                "base_type": "core/Node",
+            },
+            "cccc3333-cccc-cccc-cccc-cccccccccccc": {
+                "title": "Watcher",
+                "id": "cccc3333-cccc-cccc-cccc-cccccccccccc",
+                "properties": {},
+                "registry": "core/Watch",
+                "base_type": "core/Node",
+            },
+            "dddd4444-dddd-dddd-dddd-dddddddddddd": {
+                "title": "Mystery Node",
+                "id": "dddd4444-dddd-dddd-dddd-dddddddddddd",
+                "properties": {},
+                "registry": "mystery/Unknown",
+                "base_type": "core/Node",
+            },
+        },
+        "edges": {
+            "aaaa1111-aaaa-aaaa-aaaa-aaaaaaaaaaaa.value": [
+                "bbbb2222-bbbb-bbbb-bbbb-bbbbbbbbbbbb.state"
+            ],
+            "bbbb2222-bbbb-bbbb-bbbb-bbbbbbbbbbbb.state": [
+                "cccc3333-cccc-cccc-cccc-cccccccccccc.value"
+            ],
+        },
+        "groups": [{"title": "Inline Group"}],
+        "comments": [],
+        "inputs": [],
+        "outputs": [],
+        "module_properties": {},
+    }
+
+
+def _make_cycle_graph() -> dict:
+    """Two-node cycle: A.out -> B.in, B.out -> A.in."""
+
+    return {
+        "title": "Cycle",
+        "id": "ffffffff-ffff-ffff-ffff-ffffffffffff",
+        "registry": "test/Cycle",
+        "base_type": "core/Graph",
+        "nodes": {
+            "11111111-1111-1111-1111-111111111111": {
+                "title": "A",
+                "id": "11111111-1111-1111-1111-111111111111",
+                "properties": {},
+                "registry": "core/MakeBool",
+                "base_type": "core/Node",
+            },
+            "22222222-2222-2222-2222-222222222222": {
+                "title": "B",
+                "id": "22222222-2222-2222-2222-222222222222",
+                "properties": {},
+                "registry": "core/MakeBool",
+                "base_type": "core/Node",
+            },
+        },
+        "edges": {
+            "11111111-1111-1111-1111-111111111111.out": [
+                "22222222-2222-2222-2222-222222222222.in"
+            ],
+            "22222222-2222-2222-2222-222222222222.out": [
+                "11111111-1111-1111-1111-111111111111.in"
+            ],
+        },
+        "groups": [],
+        "comments": [],
+        "inputs": [],
+        "outputs": [],
+        "module_properties": {},
+    }
+
+
+# ---------------------------------------------------------------------------
+# Inline graph: per-function tests
+# ---------------------------------------------------------------------------
+
+
+def test_summarize_inline():
+    g = _make_inline_graph()
+    s = analysis.summarize(g)
+
+    assert s.title == "Inline Test Graph"
+    assert s.registry == "test/InlineGraph"
+    assert s.node_count == 4
+    assert s.stage_node_count == 1
+    assert s.nodes_by_category.get("core") == 3
+    assert s.nodes_by_category.get("mystery") == 1
+    assert s.group_titles == ["Inline Group"]
+
+
+def test_list_nodes_filter_inline():
+    g = _make_inline_graph()
+    all_nodes = analysis.list_nodes(g)
+    assert len(all_nodes) == 4
+
+    only_core = analysis.list_nodes(g, registry_pattern="core/")
+    assert len(only_core) == 3
+    assert all(e.registry and e.registry.startswith("core/") for e in only_core)
+
+    by_title = analysis.list_nodes(g, title_pattern="watch")
+    assert len(by_title) == 1
+    assert by_title[0].title == "Watcher"
+
+
+def test_get_node_inline_with_short_prefix():
+    g = _make_inline_graph()
+    d = analysis.get_node(g, "bbbb2222")
+    assert d.title == "Stage 2"
+    assert d.short_id == "bbbb2222"
+    assert d.registered is True
+
+    # state input is connected
+    state_in = next(i for i in d.inputs if i.name == "state")
+    assert state_in.connected
+    assert state_in.source is not None
+    assert state_in.source.short_id == "aaaa1111"
+
+    # state output is connected
+    state_out = next(o for o in d.outputs if o.name == "state")
+    assert any(c.short_id == "cccc3333" for c in state_out.consumers)
+
+    # mystery node has no static class but should still resolve via edges
+    mystery = analysis.get_node(g, "dddd4444")
+    assert mystery.registered is False
+
+
+def test_resolve_short_prefix_unique_ambiguous_missing():
+    g = _make_inline_graph()
+    # add another aaaa-prefixed node to make a collision
+    g["nodes"]["aaaa9999-zzzz-zzzz-zzzz-zzzzzzzzzzzz"] = {
+        "title": "Other",
+        "id": "aaaa9999-zzzz-zzzz-zzzz-zzzzzzzzzzzz",
+        "properties": {},
+        "registry": "core/MakeBool",
+        "base_type": "core/Node",
+    }
+
+    # unique-enough prefix still works
+    assert analysis.resolve_node_id(g, "aaaa1111") == "aaaa1111-aaaa-aaaa-aaaa-aaaaaaaaaaaa"
+
+    # ambiguous
+    with pytest.raises(ValueError, match="Ambiguous"):
+        analysis.resolve_node_id(g, "aaaa")
+
+    # missing
+    with pytest.raises(ValueError, match="No node"):
+        analysis.resolve_node_id(g, "ffff0000")
+
+
+def test_list_edges_filter():
+    g = _make_inline_graph()
+    all_edges = analysis.list_edges(g)
+    assert len(all_edges) == 2
+
+    only_from_a = analysis.list_edges(g, from_node="aaaa1111")
+    assert len(only_from_a) == 1
+    assert only_from_a[0].target_short == "bbbb2222"
+
+    only_to_c = analysis.list_edges(g, to_node="cccc3333")
+    assert len(only_to_c) == 1
+    assert only_to_c[0].source_short == "bbbb2222"
+
+
+def test_feeds_and_consumers():
+    g = _make_inline_graph()
+
+    f = analysis.feeds(g, "bbbb2222.state")
+    assert f.source is not None
+    assert f.source.short_id == "aaaa1111"
+    assert f.source_node_registry == "core/MakeBool"
+
+    f_unconnected = analysis.feeds(g, "aaaa1111.in")
+    assert f_unconnected.source is None
+
+    c = analysis.consumers(g, "bbbb2222.state")
+    assert len(c.consumers) == 1
+    assert c.consumers[0].target_short == "cccc3333"
+
+
+def test_trace_forward_inline():
+    g = _make_inline_graph()
+    t = analysis.trace_forward(g, "aaaa1111", depth=3)
+    assert t.short_id == "aaaa1111"
+    assert len(t.children) == 1
+    assert t.children[0].short_id == "bbbb2222"
+    assert t.children[0].children[0].short_id == "cccc3333"
+
+
+def test_trace_backward_inline():
+    g = _make_inline_graph()
+    t = analysis.trace_backward(g, "cccc3333", depth=3)
+    assert t.short_id == "cccc3333"
+    assert t.children[0].short_id == "bbbb2222"
+    assert t.children[0].children[0].short_id == "aaaa1111"
+
+
+def test_trace_cycle_safety():
+    g = _make_cycle_graph()
+    t = analysis.trace_forward(g, "11111111", depth=10)
+    # walk: A -> B -> A(cycle marker)
+    assert t.short_id == "11111111"
+    assert len(t.children) == 1
+    b = t.children[0]
+    assert b.short_id == "22222222"
+    assert len(b.children) == 1
+    cycled = b.children[0]
+    assert cycled.cycle is True
+    assert cycled.short_id == "11111111"
+    # no infinite recursion below the cycle marker
+    assert cycled.children == []
+
+
+def test_stage_map_inline():
+    g = _make_inline_graph()
+    sm = analysis.stage_map(g)
+    # one chain has the stage node, the other (mystery) is unstaged
+    assert len(sm.stage_nodes) == 1
+    assert sm.stage_nodes[0].stage == 2
+    assert sm.stage_nodes[0].short_id == "bbbb2222"
+    assert "dddd4444-dddd-dddd-dddd-dddddddddddd" in sm.unstaged_node_ids
+
+
+def test_check_registries_inline_finds_unknown():
+    g = _make_inline_graph()
+    r = analysis.check_registries(g)
+    unknowns = {p.registry for p in r.unknown_registries}
+    # mystery/Unknown is fake, plus the top-level test/InlineGraph
+    assert "mystery/Unknown" in unknowns
+    assert "test/InlineGraph" in unknowns
+    # core/Stage and core/MakeBool are real and should NOT appear
+    assert "core/Stage" not in unknowns
+    assert "core/MakeBool" not in unknowns
+
+
+# ---------------------------------------------------------------------------
+# Real-graph smoke tests
+# ---------------------------------------------------------------------------
+
+
+def test_real_graph_summarize_and_lists():
+    assert DIRECTOR_FIXTURE.exists(), DIRECTOR_FIXTURE
+    g = load_graph(DIRECTOR_FIXTURE)
+
+    s = analysis.summarize(g)
+    assert s.node_count > 0
+    assert s.stage_node_count >= 1
+
+    nodes = analysis.list_nodes(g)
+    assert nodes  # non-empty
+
+    sm = analysis.stage_map(g)
+    assert sm.stage_nodes  # at least one Stage node
+
+    chk = analysis.check_registries(g)
+    # Shipped graph should not contain unknown registries.
+    assert chk.unknown_registries == [], [p.model_dump() for p in chk.unknown_registries]
+    assert chk.unknown_base_types == [], [p.model_dump() for p in chk.unknown_base_types]
+
+
+def test_loader_missing_file(tmp_path):
+    with pytest.raises(GraphLoadError):
+        load_graph(tmp_path / "nope.json")
+
+
+def test_loader_bad_json(tmp_path):
+    p = tmp_path / "bad.json"
+    p.write_text("not json", encoding="utf-8")
+    with pytest.raises(GraphLoadError):
+        load_graph(p)
+
+
+# ---------------------------------------------------------------------------
+# CLI smoke tests
+# ---------------------------------------------------------------------------
+
+
+def _run_cli(argv: list[str]) -> tuple[int, str, str]:
+    parser = cli.build_parser()
+    args = parser.parse_args(argv)
+    out = io.StringIO()
+    err = io.StringIO()
+    code = cli._run(args, out, err)
+    return code, out.getvalue(), err.getvalue()
+
+
+def test_cli_summary_text():
+    code, out, err = _run_cli(["summary", str(DIRECTOR_FIXTURE)])
+    assert code == 0, err
+    assert "title:" in out
+    assert "nodes:" in out
+    assert "stage nodes:" in out
+
+
+def test_cli_summary_json_is_valid():
+    code, out, err = _run_cli(["summary", str(DIRECTOR_FIXTURE), "--json"])
+    assert code == 0, err
+    parsed = json.loads(out)
+    assert parsed["registry"] == "agents/director/chat/directorActionDirectStoryArc"
+    assert parsed["node_count"] > 0
+
+
+def test_cli_check_registries_real_graph_returns_zero():
+    code, out, err = _run_cli(["check-registries", str(DIRECTOR_FIXTURE)])
+    assert code == 0, err + out
+    assert "ok:" in out
+
+
+def test_cli_check_registries_unknown_exits_2(tmp_path):
+    g = _make_inline_graph()
+    p = tmp_path / "g.json"
+    p.write_text(json.dumps(g), encoding="utf-8")
+    code, out, err = _run_cli(["check-registries", str(p)])
+    assert code == cli.EXIT_REGISTRY_PROBLEMS
+    assert "mystery/Unknown" in out
+
+
+def test_cli_node_with_short_prefix():
+    code, out, err = _run_cli(
+        ["node", str(DIRECTOR_FIXTURE), "ccb39d43"]
+    )
+    # ccb39d43 is the top-level graph id, not a node id - expect error
+    assert code == cli.EXIT_STRUCT_ERROR
+    assert "No node" in err or "matches" in err