feat(graph): LangGraph parity shims — sync invoke + mermaid/ascii renderers

src/locus/multiagent/graph.py

@@ -1448,6 +1448,37 @@ async def ainvoke(
         result = await self.execute(inputs or {}, config=config)
         return result.final_state
 
+    def invoke(
+        self,
+        inputs: dict[str, Any] | None = None,
+        config: GraphConfig | None = None,
+        **kwargs: Any,
+    ) -> dict[str, Any]:
+        """Synchronous entry point — LangGraph parity for ``CompiledStateGraph.invoke``.
+
+        Thin sync wrapper around :meth:`ainvoke`. Refuses to run when called
+        from inside a live event loop — use :meth:`ainvoke` there instead.
+        """
+        try:
+            asyncio.get_running_loop()
+        except RuntimeError:
+            return asyncio.run(self.ainvoke(inputs, config=config, **kwargs))
+        msg = (
+            "StateGraph.invoke() called from inside a running event loop. "
+            "Use `await graph.ainvoke(...)` from async code."
+        )
+        raise RuntimeError(msg)
+
+    def run_sync(
+        self,
+        inputs: dict[str, Any] | None = None,
+        config: GraphConfig | None = None,
+        **kwargs: Any,
+    ) -> dict[str, Any]:
+        """Alias for :meth:`invoke` — matches the spelling used in the
+        ``docs/concepts/multi-agent/graph.md`` examples."""
+        return self.invoke(inputs, config=config, **kwargs)
+
     async def astream(
         self,
         inputs: dict[str, Any] | None = None,
@@ -1459,9 +1490,39 @@ async def astream(
             yield event
 
     def get_graph(self) -> StateGraph:
-        """LangGraph-compatible alias — returns self (graph exposes .nodes dict)."""
+        """LangGraph-compatible alias — returns self.
+
+        Self carries ``.nodes``, ``.edges``, ``.draw_mermaid()``, and
+        ``.draw_ascii()`` so the LangGraph chain ``compiled.get_graph().draw_mermaid()``
+        works out of the box.
+        """
         return self
 
+    def draw_mermaid(self, *, direction: str = "TD") -> str:
+        """Render this graph as a Mermaid flowchart.
+
+        LangGraph parity for ``compiled.get_graph().draw_mermaid()``.
+        Delegates to :func:`locus.multiagent.visualize.draw_mermaid`.
+        """
+        from locus.multiagent.visualize import draw_mermaid as _draw_mermaid
+
+        return _draw_mermaid(self, direction=direction)
+
+    def draw_ascii(self) -> str:
+        """Render this graph as ASCII.
+
+        LangGraph parity for ``compiled.get_graph().draw_ascii()``.
+        Delegates to :func:`locus.multiagent.visualize.draw_ascii`.
+        """
+        from locus.multiagent.visualize import draw_ascii as _draw_ascii
+
+        return _draw_ascii(self)
+
+    def get_mermaid(self, *, direction: str = "TD") -> str:
+        """Alias for :meth:`draw_mermaid` — matches the spelling used in
+        ``docs/concepts/multi-agent/graph.md``."""
+        return self.draw_mermaid(direction=direction)
+
     async def aget_state(self, config: Any = None) -> None:
         """LangGraph-compatible stub — locus uses checkpointer.load directly."""
         return

tests/unit/test_graph.py

@@ -1337,3 +1337,62 @@ async def node(inputs):
 
         repr_str = repr(graph)
         assert "StateGraph" in repr_str or "test_graph" in repr_str
+
+
+class TestLangGraphParityShims:
+    """Regression tests for the LangGraph-shape sync + rendering shims.
+
+    These match what ``docs/concepts/multi-agent/graph.md`` documents
+    and what LangGraph's ``CompiledStateGraph`` exposes.
+    """
+
+    def _trivial(self) -> StateGraph:
+        from locus.multiagent.graph import END, START
+
+        g = StateGraph()
+
+        async def bump(inputs):
+            return {"x": (inputs.get("x") or 0) + 1}
+
+        g.add_node("bump", bump)
+        g.add_edge(START, "bump")
+        g.add_edge("bump", END)
+        return g.compile()
+
+    def test_invoke_is_sync_and_runs(self):
+        compiled = self._trivial()
+        out = compiled.invoke({"x": 1})
+        assert out["x"] == 2
+
+    def test_run_sync_alias(self):
+        compiled = self._trivial()
+        assert compiled.run_sync({"x": 1})["x"] == compiled.invoke({"x": 1})["x"]
+
+    def test_invoke_refuses_inside_running_loop(self):
+        import asyncio
+
+        compiled = self._trivial()
+
+        async def caller():
+            compiled.invoke({"x": 1})
+
+        with pytest.raises(RuntimeError, match="running event loop"):
+            asyncio.run(caller())
+
+    def test_get_mermaid_and_draw_mermaid_equal(self):
+        compiled = self._trivial()
+        mermaid = compiled.get_mermaid()
+        assert mermaid == compiled.draw_mermaid()
+        assert mermaid.startswith("graph TD")
+        assert "bump" in mermaid
+
+    def test_get_graph_chain_to_draw_mermaid(self):
+        # LangGraph parity: `compiled.get_graph().draw_mermaid()` works.
+        compiled = self._trivial()
+        assert compiled.get_graph().draw_mermaid() == compiled.draw_mermaid()
+
+    def test_draw_ascii_runs(self):
+        compiled = self._trivial()
+        ascii_art = compiled.draw_ascii()
+        assert isinstance(ascii_art, str)
+        assert ascii_art