Add graph visualization queries to approval workflow samples

Add get_graph_ascii() and get_graph_mermaid() queries to both
human_in_the_loop workflow samples, demonstrating how to expose
graph execution progress via Temporal queries.

These queries return:
- ASCII art diagram with progress indicators (✓/▶/○)
- Mermaid flowchart with colored nodes by status

Author: mfateev

langgraph_plugin/human_in_the_loop/approval_graph_interrupt/workflow.py

@@ -47,6 +47,7 @@ class ApprovalWorkflow:
     def __init__(self) -> None:
         self._approval_response: dict[str, Any] | None = None
         self._interrupt_value: dict[str, Any] | None = None
+        self._app: Any = None  # Store runner for visualization queries
 
     @workflow.signal
     def provide_approval(self, response: dict[str, Any]) -> None:
@@ -76,6 +77,32 @@ def get_status(self) -> str:
         else:
             return "approved" if self._approval_response.get("approved") else "rejected"
 
+    @workflow.query
+    def get_graph_ascii(self) -> str:
+        """Query to get ASCII art visualization of graph execution progress.
+
+        Returns an ASCII diagram showing which nodes have completed,
+        which is currently executing/interrupted, and which are pending.
+        """
+        if self._app is None:
+            return "Graph not yet initialized"
+        return self._app.get_graph_ascii()
+
+    @workflow.query
+    def get_graph_mermaid(self) -> str:
+        """Query to get Mermaid diagram of graph execution progress.
+
+        Returns a Mermaid flowchart with nodes colored by status:
+        - Green: completed nodes
+        - Yellow: current/interrupted node
+        - Gray: pending nodes
+
+        Can be rendered in GitHub, Notion, or any Mermaid-compatible viewer.
+        """
+        if self._app is None:
+            return "Graph not yet initialized"
+        return self._app.get_graph_mermaid()
+
     @workflow.run
     async def run(
         self,
@@ -92,7 +119,7 @@ async def run(
         Returns:
             The final state containing result and executed status.
         """
-        app = lg_compile("approval_workflow")
+        self._app = lg_compile("approval_workflow")
 
         # Handle both dataclass and dict input (Temporal deserializes to dict)
         if isinstance(request, dict):
@@ -112,7 +139,7 @@ async def run(
         }
 
         # First invocation - should hit interrupt at request_approval node
-        result = await app.ainvoke(initial_state)
+        result = await self._app.ainvoke(initial_state)
 
         # Check for interrupt
         if "__interrupt__" in result:
@@ -153,6 +180,6 @@ async def run(
             )
 
             # Resume with the approval response
-            result = await app.ainvoke(Command(resume=self._approval_response))
+            result = await self._app.ainvoke(Command(resume=self._approval_response))
 
         return result

langgraph_plugin/human_in_the_loop/approval_wait_condition/workflow.py

@@ -39,6 +39,7 @@ class ApprovalWorkflow:
     def __init__(self) -> None:
         self._approval_response: dict[str, Any] | None = None
         self._pending_approval: dict[str, Any] | None = None
+        self._app: Any = None  # Store runner for visualization queries
 
     @workflow.signal
     def provide_approval(self, response: dict[str, Any]) -> None:
@@ -68,6 +69,32 @@ def get_status(self) -> str:
         else:
             return "approved" if self._approval_response.get("approved") else "rejected"
 
+    @workflow.query
+    def get_graph_ascii(self) -> str:
+        """Query to get ASCII art visualization of graph execution progress.
+
+        Returns an ASCII diagram showing which nodes have completed,
+        which is currently executing/interrupted, and which are pending.
+        """
+        if self._app is None:
+            return "Graph not yet initialized"
+        return self._app.get_graph_ascii()
+
+    @workflow.query
+    def get_graph_mermaid(self) -> str:
+        """Query to get Mermaid diagram of graph execution progress.
+
+        Returns a Mermaid flowchart with nodes colored by status:
+        - Green: completed nodes
+        - Yellow: current/interrupted node
+        - Gray: pending nodes
+
+        Can be rendered in GitHub, Notion, or any Mermaid-compatible viewer.
+        """
+        if self._app is None:
+            return "Graph not yet initialized"
+        return self._app.get_graph_mermaid()
+
     @workflow.run
     async def run(self, request: ApprovalRequest) -> dict[str, Any]:
         """Run the approval workflow.
@@ -78,7 +105,7 @@ async def run(self, request: ApprovalRequest) -> dict[str, Any]:
         Returns:
             The final state containing result and executed status.
         """
-        app = lg_compile("approval_workflow")
+        self._app = lg_compile("approval_workflow")
 
         # Handle both dataclass and dict input (Temporal deserializes to dict)
         if isinstance(request, dict):
@@ -98,6 +125,6 @@ async def run(self, request: ApprovalRequest) -> dict[str, Any]:
         }
 
         # Run the graph - the request_approval node will wait for signal internally
-        result = await app.ainvoke(initial_state)
+        result = await self._app.ainvoke(initial_state)
 
         return result