Add interactive TUI for memex and dagit

Textual-based chat interface with OpenAI function calling.
Combines memex knowledge graph tools (search, traverse, create)
with dagit social network tools (post, read, reply, verify).

Author: systemshift

tui/.gitignore

@@ -0,0 +1,3 @@
+memex-tui-venv/
+__pycache__/
+*.egg-info/

tui/memex_tui/__init__.py

@@ -0,0 +1,12 @@
+"""Memex TUI - Interactive control plane for memex and dagit."""
+
+from .app import MemexApp
+
+
+def main():
+    """Entry point for memex-tui command."""
+    app = MemexApp()
+    app.run()
+
+
+__all__ = ["main", "MemexApp"]

tui/memex_tui/app.py

@@ -0,0 +1,152 @@
+"""Memex TUI Application."""
+
+from textual.app import App, ComposeResult
+from textual.widgets import Input, Footer, Header
+from textual.containers import Container
+from textual.binding import Binding
+
+from .chat import ChatPanel, ChatEngine
+
+
+class MemexApp(App):
+    """Interactive TUI for memex and dagit."""
+
+    CSS = """
+    Screen {
+        layout: grid;
+        grid-size: 1;
+        grid-rows: 1fr auto;
+    }
+
+    #chat-container {
+        height: 100%;
+        border: solid $primary;
+        padding: 1;
+    }
+
+    #chat-log {
+        height: 100%;
+        scrollbar-gutter: stable;
+    }
+
+    #input-container {
+        height: auto;
+        padding: 1;
+    }
+
+    #input {
+        dock: bottom;
+    }
+
+    Footer {
+        height: auto;
+    }
+    """
+
+    BINDINGS = [
+        Binding("ctrl+c", "quit", "Quit"),
+        Binding("ctrl+l", "clear", "Clear"),
+        Binding("escape", "focus_input", "Focus Input", show=False),
+    ]
+
+    TITLE = "Memex"
+
+    def __init__(self):
+        super().__init__()
+        self.chat_engine = ChatEngine()
+        self._current_response = ""
+
+    def compose(self) -> ComposeResult:
+        """Create the UI layout."""
+        yield Header()
+        with Container(id="chat-container"):
+            yield ChatPanel(id="chat-log")
+        with Container(id="input-container"):
+            yield Input(placeholder="Ask anything... (Ctrl+C to quit)", id="input")
+        yield Footer()
+
+    def on_mount(self) -> None:
+        """Focus input on start."""
+        self.query_one("#input", Input).focus()
+        chat = self.query_one("#chat-log", ChatPanel)
+        chat.add_system_message(
+            "Welcome to Memex TUI. Ask questions about your knowledge graph or dagit network."
+        )
+        chat.add_system_message('Type "help" for commands, Ctrl+C to quit.')
+
+    async def on_input_submitted(self, event: Input.Submitted) -> None:
+        """Handle user input."""
+        user_input = event.value.strip()
+        if not user_input:
+            return
+
+        # Clear input
+        input_widget = self.query_one("#input", Input)
+        input_widget.value = ""
+
+        chat = self.query_one("#chat-log", ChatPanel)
+
+        # Handle special commands
+        if user_input.lower() in ("exit", "quit"):
+            self.exit()
+            return
+
+        if user_input.lower() == "clear":
+            self.action_clear()
+            return
+
+        if user_input.lower() == "help":
+            self._show_help(chat)
+            return
+
+        # Show user message
+        chat.add_user_message(user_input)
+
+        # Start assistant response
+        chat.start_assistant_response()
+        self._current_response = ""
+
+        # Stream response
+        async def on_text(text: str) -> None:
+            self._current_response += text
+            chat.add_assistant_text(text)
+
+        async def on_tool(tool_name: str) -> None:
+            chat.add_tool_indicator(tool_name)
+
+        try:
+            await self.chat_engine.send(user_input, on_text, on_tool)
+        except Exception as e:
+            chat.add_error(str(e))
+
+        # Ensure we end on a new line
+        if self._current_response and not self._current_response.endswith("\n"):
+            chat.write("")
+
+    def _show_help(self, chat: ChatPanel) -> None:
+        """Show help message."""
+        chat.add_system_message("Commands:")
+        chat.add_system_message("  help  - Show this help")
+        chat.add_system_message("  clear - Clear chat history")
+        chat.add_system_message("  exit  - Quit the application")
+        chat.add_system_message("")
+        chat.add_system_message("Examples:")
+        chat.add_system_message('  "search for notes about topic"')
+        chat.add_system_message('  "what\'s my dagit identity"')
+        chat.add_system_message('  "save this as a note: <your content>"')
+        chat.add_system_message('  "post to dagit: <your message>"')
+
+    def action_clear(self) -> None:
+        """Clear chat history."""
+        self.chat_engine.clear()
+        chat = self.query_one("#chat-log", ChatPanel)
+        chat.clear()
+        chat.add_system_message("Chat cleared.")
+
+    def action_focus_input(self) -> None:
+        """Focus the input field."""
+        self.query_one("#input", Input).focus()
+
+    def action_quit(self) -> None:
+        """Quit the application."""
+        self.exit()

tui/memex_tui/chat.py

@@ -0,0 +1,146 @@
+"""Chat panel widget for Memex TUI."""
+
+import json
+from typing import Callable, Awaitable
+
+from textual.widgets import RichLog
+from rich.markdown import Markdown
+from rich.text import Text
+
+from .provider import ChatProvider, Chunk, ToolCall
+from .tools import get_all_tools, execute_tool
+
+
+SYSTEM_PROMPT = """You are Memex, an intelligent assistant with access to a knowledge graph (memex) and a decentralized social network (dagit).
+
+Available capabilities:
+- Search and query the knowledge graph (memex_search, memex_get_node, memex_traverse, etc.)
+- Create notes and save information (memex_create_node)
+- Post to the dagit social network (dagit_post)
+- Read posts from dagit (dagit_read)
+- Check your identity (dagit_whoami)
+
+When users ask questions:
+1. Use the appropriate tools to find information
+2. Synthesize the results into a helpful response
+3. If saving information, confirm what was saved
+
+Be concise but helpful. Use the tools proactively when they would help answer the user's question."""
+
+
+class ChatEngine:
+    """Manages chat state and model interactions."""
+
+    def __init__(self):
+        self.provider = ChatProvider()
+        self.messages: list[dict] = []
+        self.tools = get_all_tools()
+
+    async def send(
+        self,
+        user_input: str,
+        on_text: Callable[[str], Awaitable[None]],
+        on_tool: Callable[[str], Awaitable[None]],
+    ) -> None:
+        """Send a message and stream the response.
+
+        Args:
+            user_input: User's message
+            on_text: Callback for text chunks
+            on_tool: Callback for tool call notifications
+        """
+        self.messages.append({"role": "user", "content": user_input})
+
+        max_turns = 10
+        for _ in range(max_turns):
+            text_buffer = ""
+            tool_calls: list[ToolCall] = []
+
+            for chunk in self.provider.stream(
+                SYSTEM_PROMPT, self.messages, self.tools
+            ):
+                if chunk.type == "text":
+                    await on_text(chunk.text)
+                    text_buffer += chunk.text
+
+                elif chunk.type == "tool_call":
+                    tool_calls.append(chunk.tool_call)
+                    await on_tool(f"[{chunk.tool_call.name}]")
+
+                elif chunk.type == "error":
+                    await on_text(f"\nError: {chunk.error}")
+                    return
+
+            # If there were tool calls, execute them and continue
+            if tool_calls:
+                # Add assistant message with tool calls
+                self.messages.append(
+                    {
+                        "role": "assistant",
+                        "content": text_buffer or None,
+                        "tool_calls": [
+                            {
+                                "id": tc.id,
+                                "type": "function",
+                                "function": {
+                                    "name": tc.name,
+                                    "arguments": json.dumps(tc.arguments),
+                                },
+                            }
+                            for tc in tool_calls
+                        ],
+                    }
+                )
+
+                # Execute tools and add results
+                for tc in tool_calls:
+                    result = execute_tool(tc.name, tc.arguments)
+                    self.messages.append(
+                        {
+                            "role": "tool",
+                            "tool_call_id": tc.id,
+                            "content": result,
+                        }
+                    )
+                continue
+
+            # No tool calls - conversation turn complete
+            if text_buffer:
+                self.messages.append({"role": "assistant", "content": text_buffer})
+            return
+
+    def clear(self) -> None:
+        """Clear conversation history."""
+        self.messages.clear()
+
+
+class ChatPanel(RichLog):
+    """Chat display panel with rich formatting."""
+
+    def __init__(self, **kwargs):
+        super().__init__(markup=True, wrap=True, **kwargs)
+
+    def add_user_message(self, text: str) -> None:
+        """Add a user message to the display."""
+        self.write(Text.from_markup(f"[bold cyan]You:[/bold cyan] {text}"))
+
+    def add_assistant_text(self, text: str) -> None:
+        """Add assistant text (streaming)."""
+        # For streaming, we append to the current line
+        self.write(text, scroll_end=True)
+
+    def start_assistant_response(self) -> None:
+        """Start a new assistant response line."""
+        self.write(Text.from_markup("[bold green]Memex:[/bold green] "), scroll_end=True)
+
+    def add_tool_indicator(self, tool_name: str) -> None:
+        """Show a tool being called."""
+        self.write(Text.from_markup(f"[dim]{tool_name}[/dim]"))
+
+    def add_error(self, error: str) -> None:
+        """Display an error message."""
+        self.write(Text.from_markup(f"[bold red]Error:[/bold red] {error}"))
+
+    def add_system_message(self, text: str) -> None:
+        """Display a system message."""
+        self.write(Text.from_markup(f"[dim italic]{text}[/dim italic]"))