dashboard v2

CHEATBOOK.md

@@ -0,0 +1,111 @@
+# Flowk 🌊 Cheatbook
+
+A comprehensive quick-reference guide for building, orchestrating, and observing AI agent workflows with Flowk.
+
+---
+
+## 1. Core Classes & Decorators
+
+### `Graph`
+The central coordinator for building and executing workflows.
+
+| Method | Description | Example |
+| :--- | :--- | :--- |
+| `Graph(state_schema=None, checkpoint_db=None)` | Initialize a graph. Optionally provide a Pydantic schema and a connection string for persistence. | `g = Graph(state_schema=MyState, checkpoint_db="flow.db")` |
+| `@g.node(retries=0, fallback=None)` | Decorator to register a function as a node. | `@g.node(retries=3)\ndef task(data): return data` |
+| `g.connect(from_node, to_node)` | Connect two nodes sequentially. | `g.connect(start_node, end_node)` |
+| `g.route(condition_fn, mapping_dict)` | Create a deterministic branch point. | `g.route(decider, {"yes": node_a, "no": node_b})` |
+| `@g.llm_router(targets, model="gpt-4o-mini")` | Decorator for intelligent LLM-based auto-routing. | `@g.llm_router(targets={"a": "desc", "b": "desc"})\ndef router(state): return state["text"]` |
+| `g.compile(interrupt_before=None)` | Freeze graph and set breakpoints. | `g.compile(interrupt_before=["final_step"])` |
+| `g.run(input, session_id=None)` | Execute graph synchronously. | `result = g.run("Hello")` |
+| `await g.arun(input, session_id=None)` | Execute graph asynchronously (supports parallel branches). | `res = await g.arun("Query")` |
+| `async for event in g.astream(input)` | Stream execution events (JSON objects). | `async for e in g.astream(inp): print(e)` |
+| `g.as_node(state_key=None)` | Wrap an entire graph into a single node for composition. | `sub_node = sub_graph.as_node(state_key="sub")` |
+| `g.serve(host="0.0.0.0", port=8000)` | 1-Click FastAPI deployment. | `g.serve(port=8080)` |
+| `g.show()` | Visualize graph in terminal as ASCII. | `g.show()` |
+| `g.debug(input)` | Run with verbose terminal logging. | `g.debug("test")` |
+| `g.replay(run_id)` | Replay a historical execution trace. | `g.replay("run-123")` |
+
+---
+
+## 2. Advanced Features
+
+### Persistence: `StorageRegistry` & `MemoryStore`
+Handled automatically when `checkpoint_db` is passed to `Graph()`.
+
+- **SQLite**: `checkpoint_db="flow.db"`
+- **Redis**: `checkpoint_db="redis://localhost:6379/0"`
+
+| Method | Description |
+| :--- | :--- |
+| `StorageRegistry.get_trace(run_id)` | Retrieve raw execution trace. |
+| `StorageRegistry.list_runs(session_id=None)` | List historical run IDs. |
+
+### Metrics: `MetricsRegistry` & `PluginManager`
+Track token usage and execution performance.
+
+| Method | Description | Example |
+| :--- | :--- | :--- |
+| `MetricsRegistry.get_summary()` | Returns JSON of token usage, costs, and node times. | `print(g.metrics())` |
+| `PluginManager.register(plugin)` | Register an LLM metrics plugin (OpenAI/Anthropic). | `PluginManager.register(OpenAIPlugin())` |
+
+---
+
+## 3. CLI Commands
+
+| Command | Description |
+| :--- | :--- |
+| `flowk ui` | Launch the **Production-Grade Dashboard** (v2) on port 8502. |
+| `flowk serve <file>:<graph>` | Serve a graph file as a FastAPI app. |
+
+---
+
+## 4. Examples
+
+### A. Minimal Async Graph
+```python
+from flowk import Graph
+
+g = Graph()
+
+@g.node()
+async def greet(name: str):
+    return f"Hello {name}!"
+
+@g.node()
+async def process(msg: str):
+    return msg.upper()
+
+g.connect(greet, process)
+
+# Run it
+if __name__ == "__main__":
+    import asyncio
+    res = asyncio.run(g.arun("World"))
+    print(res) # => "HELLO WORLD!"
+```
+
+### B. Intelligent Router
+```python
+@g.llm_router(targets={
+    "billing": "Queries about invoices or payments",
+    "support": "General help or technical issues"
+})
+def supervisor(state):
+    return state["user_message"]
+
+g.connect(input_node, supervisor)
+```
+
+### C. Human-in-the-Loop
+```python
+g.compile(interrupt_before=["payment"])
+
+# Running this will stop before 'payment'
+async for event in g.astream(data, session_id="abc"):
+    if event["type"] == "interrupt":
+        print("PAUSED. Waiting for review.")
+
+# Resume later with same session_id
+await g.arun(None, session_id="abc")
+```

DOCUMENTATION.md

@@ -220,13 +220,19 @@ This isolates complex pipelines without polluting supervisor-level logic domains
 
 ## 8. Observability (CLI, Time-Travel, Dashboard)
 
-### Flowk UI (Local Dashboard)
-Inspect run histories, trace logic changes, and debug global state permutations explicitly via our Streamlit UI.
+## **📊 Observability Dashboard:** Track sessions, visualize topology, and perform step-by-step state diffing through the local production-grade dashboard (`flowk ui`).
+Spin up the native **Production-Grade Dashboard** to review these checkpoints visually with interactive graph topology and state diffing:
 ```bash
 flowk ui
-# Operates at http://localhost:8501 pulling from SQLite local checkpoints
+# Launches at http://localhost:8502
 ```
 
+**Features:**
+- **Interactive Graph Visualization**: Real-time SVG rendering of your graph topology and execution paths.
+- **State Diff Engine**: Side-by-side comparison of global state snapshots between chaque node execution.
+- **Persistent Trace Storage**: Backed by the `StorageRegistry` (SQLite/Redis), allowing you to browse historical sessions and runs.
+- **Multi-Agent Tracing**: Detailed visibility into nested sub-graph executions.
+
 ### CLI Operations
 Render beautiful flow charts to your developer terminal before production to visually confirm structural intent:
 ```python

README.md

@@ -158,6 +158,8 @@ g.arun(None, session_id="user_john")
 
 ---
 
+---
+
 ## 📊 Observability Dashboard & Persistence
 
 Flowk effortlessly saves run-histories exactly as they mutate across node transactions.
@@ -169,11 +171,17 @@ g = Graph(checkpoint_db="local_traces.db")          # SQLite Storage
 g = Graph(checkpoint_db="redis://localhost:6379/0") # Redis
 ```
 
-Spin up the native **Streamlit Time-Machine Dashboard** to review these checkpoints visually without relying on generic SaaS providers:
+Spin up the native **Production-Grade Dashboard** (v2) to review these checkpoints visually with interactive graph topology and state diffing:
 ```bash
 flowk ui
+# Launches at http://localhost:8502
 ```
 
+The new dashboard provides:
+- **Interactive SVG Graphs**: Visualize your workflow logic and execution paths.
+- **State Diff Engine**: Compare state snapshots step-by-step.
+- **Session History**: Browse and resume historical agent runs from SQLite/Redis.
+
 ---
 
 ## 📦 Multi-Agent Composition

flowk/cli.py

@@ -10,16 +10,24 @@ def main():
     command = sys.argv[1]
 
     if command == "ui":
-        # Launch streamlit dashboard
+        # Launch production-grade v2 dashboard
         try:
-            import streamlit  # pyre-ignore
+            from fastapi import FastAPI
+            import uvicorn
         except ImportError:
-            print("Streamlit is not installed. Run 'pip install flowk[ui]' to enable the dashboard.")
+            print("FastAPI and Uvicorn are required for the v2 dashboard. Run 'pip install flowk[api]'")
             sys.exit(1)
 
-        ui_path = os.path.join(os.path.dirname(__file__), "ui", "dashboard.py")
-        print("🌊 Starting Flowk Observability Dashboard...")
-        subprocess.run(["streamlit", "run", ui_path])
+        from flowk import Graph
+        from flowk.server import create_app
+
+        # Create a dummy graph to host the dashboard if none is provided
+        # It will still serve the /ui/sessions etc. from the global memory store
+        dummy_graph = Graph(checkpoint_db="flowk_memory.db")
+        app = create_app(dummy_graph)
+
+        print("🔥 Starting Flowk Production Dashboard (v2)...")
+        uvicorn.run(app, host="127.0.0.1", port=8502)
     else:
         print(f"Unknown command: {command}")
 

flowk/executor.py

@@ -93,7 +93,7 @@ def execute(
             PluginManager.on_node_end(run_id, node, output, state)  # pyre-ignore
 
             if status == "error":
-                StorageRegistry.save_trace(run_id, execution_trace)  # pyre-ignore
+                StorageRegistry.save_trace(run_id, execution_trace, session_id=session_id)  # pyre-ignore
                 if session_id:
                     MemoryStore.save_state(session_id, state.to_dict())  # pyre-ignore
                 raise RuntimeError(f"Execution failed at node '{current_node_name}': {error}")
@@ -120,7 +120,7 @@ def execute(
                     )
 
         total_duration = time.time() - start_time
-        StorageRegistry.save_trace(run_id, execution_trace)  # pyre-ignore
+        StorageRegistry.save_trace(run_id, execution_trace, session_id=session_id)  # pyre-ignore
         MetricsRegistry.record_run(total_duration)  # pyre-ignore
         PluginManager.on_run_end(run_id, self.graph, current_input)  # pyre-ignore
 
@@ -240,7 +240,7 @@ async def astream(
             active_nodes = next_layer
 
         total_duration = time.time() - start_time
-        StorageRegistry.save_trace(run_id, execution_trace)  # pyre-ignore
+        StorageRegistry.save_trace(run_id, execution_trace, session_id=session_id)  # pyre-ignore
         MetricsRegistry.record_run(total_duration)  # pyre-ignore
         PluginManager.on_run_end(run_id, self.graph, None)  # pyre-ignore
 

flowk/graph.py

@@ -29,7 +29,9 @@ def __init__(
 
         if checkpoint_db:
             from flowk.memory import MemoryStore  # pyre-ignore
+            from flowk.storage import StorageRegistry
             MemoryStore.configure(checkpoint_db)
+            StorageRegistry.configure(checkpoint_db)
 
     # ------------------------------------------------------------------
     # Node registration
@@ -167,6 +169,19 @@ def compile(self, interrupt_before: Optional[List[str]] = None) -> "Graph":
                     raise RuntimeError(
                         f"Edge compilation error: Target node '{target}' from '{source}' does not exist."
                     )
+
+        # Persist topology for UI observability
+        from flowk.storage import StorageRegistry
+        nodes = [{"id": n.name, "name": n.name, "type": "agent" if "agent" in n.name.lower() else "node"} for n in self.nodes.values()]
+        edges = []
+        for src, targets in self.edges.items():
+            for tgt in targets:
+                edges.append({"source": src, "target": tgt, "type": "flow"})
+        for src, mapping in self.routes.items():
+            for val, tgt in mapping.items():
+                edges.append({"source": src, "target": tgt, "type": "route", "label": str(val)})
+
+        StorageRegistry.save_graph("default", {"nodes": nodes, "edges": edges})
 
         self.interrupt_before = interrupt_before if interrupt_before is not None else []  # pyre-ignore
         self.compiled = True

flowk/memory.py

@@ -145,6 +145,35 @@ def clear(cls, session_id: Optional[str] = None) -> None:
         else:
             cls._sessions.clear()
 
+    # ------------------------------------------------------------------
+    # UI / Observability
+    # ------------------------------------------------------------------
+
+    @classmethod
+    def list_sessions(cls) -> Dict[str, dict]:
+        """Return all persisted sessions and their latest states."""
+        redis_client: Any = cls._redis_client
+        if redis_client is not None:
+            keys = redis_client.keys("flowk:session:*")
+            sessions = {}
+            for key in keys:
+                s_id = key.decode("utf-8").replace("flowk:session:", "")
+                raw = redis_client.get(key)
+                sessions[s_id] = json.loads(raw) if raw else {}
+            return sessions
+
+        db_path: Optional[str] = cls._db_path
+        if db_path is not None:
+            sessions = {}
+            with sqlite3.connect(db_path) as conn:
+                rows = conn.execute("SELECT id, state FROM sessions").fetchall()
+                for row in rows:
+                    sessions[row[0]] = json.loads(row[1])
+            return sessions
+
+        # In-memory fallback
+        return cls._sessions.copy()
+
     # ------------------------------------------------------------------
     # Convenience
     # ------------------------------------------------------------------

flowk/server.py

@@ -3,8 +3,10 @@
 
 try:
     from fastapi import FastAPI, Request, HTTPException  # pyre-ignore
-    from fastapi.responses import StreamingResponse  # pyre-ignore
+    from fastapi.responses import StreamingResponse, FileResponse  # pyre-ignore
+    from fastapi.staticfiles import StaticFiles  # pyre-ignore
     import uvicorn  # pyre-ignore
+    import os
     _fastapi_available = True
 except ImportError:
     _fastapi_available = False
@@ -27,6 +29,18 @@ def create_app(graph: Any) -> Any:
         version="0.3.0",
     )
 
+    # ------------------------------------------------------------------
+    # Static UI Assets
+    # ------------------------------------------------------------------
+    ui_dist = os.path.join(os.path.dirname(__file__), "ui", "v2", "dist")
+
+    if os.path.exists(ui_dist):
+        app.mount("/assets", StaticFiles(directory=os.path.join(ui_dist, "assets")), name="assets")
+
+        @app.get("/")
+        async def serve_ui():
+            return FileResponse(os.path.join(ui_dist, "index.html"))
+
     @app.post("/invoke")  # pyre-ignore
     async def invoke(request: Request) -> dict:  # pyre-ignore
         """
@@ -72,4 +86,56 @@ async def event_generator():
 
         return StreamingResponse(event_generator(), media_type="text/event-stream")  # pyre-ignore
 
+    # ------------------------------------------------------------------
+    # UI / Observability Endpoints
+    # ------------------------------------------------------------------
+
+    @app.get("/ui/sessions")
+    async def get_sessions():
+        """List all active sessions and their latest state snapshots."""
+        from flowk.memory import MemoryStore
+        return MemoryStore.list_sessions()
+
+    @app.get("/ui/runs")
+    async def get_runs(session_id: Optional[str] = None):
+        """List all recorded execution run IDs, optional filter by session."""
+        from flowk.storage import StorageRegistry
+        return StorageRegistry.list_runs(session_id=session_id)
+
+    @app.get("/ui/session/{session_id}/runs")
+    async def get_session_runs(session_id: str):
+        """List all run IDs for a specific session."""
+        from flowk.storage import StorageRegistry
+        return StorageRegistry.list_runs(session_id=session_id)
+
+    @app.get("/ui/run/{run_id}")
+    async def get_run_trace(run_id: str):
+        """Retrieve the full step-by-step execution trace for a specific run."""
+        from flowk.storage import StorageRegistry
+        try:
+            return StorageRegistry.get_trace(run_id)
+        except Exception as e:
+            raise HTTPException(status_code=404, detail=str(e))
+
+    @app.get("/ui/graph")
+    async def get_graph():
+        """Export graph topology for UI visualization."""
+        from flowk.storage import StorageRegistry
+
+        # Try finding persisted graph first if local graph is dummy
+        if not graph.nodes:
+            persisted = StorageRegistry.get_graph("default")
+            if persisted:
+                return persisted
+
+        nodes = [{"id": n.name, "name": n.name, "type": "agent" if "agent" in n.name.lower() else "node"} for n in graph.nodes.values()]
+        edges = []
+        for src, targets in graph.edges.items():
+            for tgt in targets:
+                edges.append({"source": src, "target": tgt, "type": "flow"})
+        for src, mapping in graph.routes.items():
+            for val, tgt in mapping.items():
+                edges.append({"source": src, "target": tgt, "type": "route", "label": str(val)})
+        return {"nodes": nodes, "edges": edges}
+
     return app