Merge branch 'feature/textual-tui'

Add Textual TUI replacing the sequential print-and-scroll REPL with a
full terminal UI: fixed input at bottom, scrollable chat log, header
with model info, streaming display, and status bar with token stats.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

app/agent/loop.py

@@ -11,14 +11,7 @@ from app.services.llm import LLMClient, LLMConnectionError, LLMError, LLMStreamE
 from app.services.permissions import PermissionsService
 from app.services.streaming import StreamHandler
 from app.tools.registry import ToolRegistry
-from app.utils.display import (
-    print_error,
-    print_iteration_header,
-    print_tool_call,
-    print_tool_result,
-    print_token_usage,
-    print_warning,
-)
+from app.utils.display import DisplayAdapter
 from app.utils.logging import get_logger
 
 logger = get_logger(__name__)
@@ -42,6 +35,7 @@ class AgentLoop:
         handler: StreamHandler,
         registry: ToolRegistry,
         permissions: PermissionsService,
+        display: DisplayAdapter | None = None,
     ) -> None:
         self._config = config
         self._ctx = ctx
@@ -49,6 +43,7 @@ class AgentLoop:
         self._handler = handler
         self._registry = registry
         self._permissions = permissions
+        self._display = display
         self._tools_schema = registry.get_openai_tools_schema()
         self._system_prompt = self._build_system_prompt()
         self._cancelled = False
@@ -92,7 +87,8 @@ class AgentLoop:
         reasoning_only_streak = 0
         for iteration in range(1, max_iter + 1):
             if self._cancelled:
-                print_warning("Agent loop cancelled.")
+                if self._display:
+                    self._display.write_warning("Agent loop cancelled.")
                 break
 
             # Check token budget — try truncation before giving up
@@ -100,13 +96,16 @@ class AgentLoop:
                 system_tokens = self._ctx.token_counter.estimate_tokens(self._system_prompt)
                 dropped = self._ctx.truncate_history(system_tokens)
                 if dropped > 0:
-                    print_warning(f"Token budget pressure: dropped {dropped} oldest messages.")
+                    if self._display:
+                        self._display.write_warning(f"Token budget pressure: dropped {dropped} oldest messages.")
                 else:
-                    print_warning("Token budget exceeded, cannot truncate further. Stopping.")
+                    if self._display:
+                        self._display.write_warning("Token budget exceeded, cannot truncate further. Stopping.")
                     break
 
             if iteration > 1:
-                print_iteration_header(iteration, max_iter)
+                if self._display:
+                    self._display.write_iteration_header(iteration, max_iter)
 
             # Stream LLM response
             assistant_msg = await self._llm_step()
@@ -127,11 +126,11 @@ class AgentLoop:
             if self._handler.usage:
                 self._ctx.token_counter.count_usage(self._handler.usage)
 
-            if self._config.display.show_token_usage:
+            if self._config.display.show_token_usage and self._display:
                 total = self._ctx.token_counter.cumulative_usage.total_tokens
                 if total == 0:
                     total = self._ctx.estimated_tokens
-                print_token_usage(total, self._ctx.token_counter.budget)
+                self._display.write_token_usage(total, self._ctx.token_counter.budget)
 
             self._handler.reset()
 
@@ -142,17 +141,19 @@ class AgentLoop:
 
                 if reasoning_only_streak >= _MAX_REASONING_RETRIES:
                     # Nudge the model by injecting a user hint
-                    print_warning(
-                        f"Model produced reasoning but no response {reasoning_only_streak} times. "
-                        "Nudging model to respond..."
-                    )
+                    if self._display:
+                        self._display.write_warning(
+                            f"Model produced reasoning but no response {reasoning_only_streak} times. "
+                            "Nudging model to respond..."
+                        )
                     self._ctx.add_message(
                         "user",
                         "Please respond with your answer. Do not just think — provide your actual response.",
                     )
                     reasoning_only_streak = 0
                 else:
-                    print_warning("Model produced reasoning but no response. Retrying...")
+                    if self._display:
+                        self._display.write_warning("Model produced reasoning but no response. Retrying...")
                 continue
 
             # Successful response — reset streak
@@ -160,10 +161,12 @@ class AgentLoop:
 
             # No tool calls → task complete (plain text response)
             if not assistant_msg.tool_calls:
+                if self._display and assistant_msg.content:
+                    self._display.write_assistant_message(assistant_msg.content)
                 break
 
             # Execute tool calls
-            results = self._execute_tool_calls(assistant_msg.tool_calls)
+            results = await self._execute_tool_calls(assistant_msg.tool_calls)
 
             # Add tool results to context
             for result in results:
@@ -179,7 +182,8 @@ class AgentLoop:
             if any(r.tool_name == "finish" for r in results):
                 break
         else:
-            print_warning(f"Agent reached maximum iterations ({max_iter}). Stopping.")
+            if self._display:
+                self._display.write_warning(f"Agent reached maximum iterations ({max_iter}). Stopping.")
 
     async def _llm_step(self) -> Message | None:
         """Stream one LLM response and return the accumulated Message.
@@ -195,21 +199,25 @@ class AgentLoop:
             chunk_iter = self._client.stream_chat_with_retry(messages, tools=self._tools_schema)
             return await self._handler.process_stream(chunk_iter)
         except KeyboardInterrupt:
-            print_warning("Response interrupted.")
+            if self._display:
+                self._display.write_warning("Response interrupted.")
             self._handler.reset()
             return None
         except (LLMConnectionError, LLMStreamError) as e:
             partial = self._handler.get_partial_message()
             if partial is not None:
-                print_warning(f"Stream interrupted ({e}), returning partial response.")
+                if self._display:
+                    self._display.write_warning(f"Stream interrupted ({e}), returning partial response.")
                 return partial
-            print_error(f"Connection error: {e}")
+            if self._display:
+                self._display.write_error(f"Connection error: {e}")
             return None
         except LLMError as e:
-            print_error(f"LLM error: {e}")
+            if self._display:
+                self._display.write_error(f"LLM error: {e}")
             return None
 
-    def _execute_tool_calls(self, tool_calls: list[ToolCall]) -> list[ToolResult]:
+    async def _execute_tool_calls(self, tool_calls: list[ToolCall]) -> list[ToolResult]:
         """Execute a list of tool calls with permission checks.
 
         Args:
@@ -226,8 +234,8 @@ class AgentLoop:
             tc_id = tc.id
 
             # Display the tool call
-            if self._config.display.show_tool_calls:
-                print_tool_call(name, tc.function.arguments)
+            if self._config.display.show_tool_calls and self._display:
+                self._display.write_tool_call(name, tc.function.arguments)
 
             # Parse arguments
             try:
@@ -240,8 +248,8 @@ class AgentLoop:
                     error=f"Invalid JSON in arguments: {e}",
                 )
                 results.append(result)
-                if self._config.display.show_tool_calls:
-                    print_tool_result(name, result.error or "", is_error=True)
+                if self._config.display.show_tool_calls and self._display:
+                    self._display.write_tool_result(name, result.error or "", is_error=True)
                 continue
 
             # Look up tool
@@ -254,13 +262,13 @@ class AgentLoop:
                     error=f"Unknown tool '{name}'. Available: {available_names}",
                 )
                 results.append(result)
-                if self._config.display.show_tool_calls:
-                    print_tool_result(name, result.error or "", is_error=True)
+                if self._config.display.show_tool_calls and self._display:
+                    self._display.write_tool_result(name, result.error or "", is_error=True)
                 continue
 
             # Check permissions (truncate args for display in prompt)
             desc = tc.function.arguments[:120] + "..." if len(tc.function.arguments) > 120 else tc.function.arguments
-            if not self._permissions.check(name, description=desc):
+            if not await self._permissions.check(name, description=desc):
                 result = ToolResult(
                     tool_call_id=tc_id,
                     tool_name=name,
@@ -268,17 +276,17 @@ class AgentLoop:
                     error=f"Permission denied for tool '{name}'",
                 )
                 results.append(result)
-                if self._config.display.show_tool_calls:
-                    print_tool_result(name, result.error or "", is_error=True)
+                if self._config.display.show_tool_calls and self._display:
+                    self._display.write_tool_result(name, result.error or "", is_error=True)
                 continue
 
             # Execute tool (BaseTool.run never raises)
             result = tool.run(tc_id, parsed_args)
             results.append(result)
 
-            if self._config.display.show_tool_calls:
+            if self._config.display.show_tool_calls and self._display:
                 is_error = result.status == ToolResultStatus.ERROR
                 output = result.error if is_error else result.output
-                print_tool_result(name, output or "", is_error=is_error)
+                self._display.write_tool_result(name, output or "", is_error=is_error)
 
         return results

app/main.py

@@ -1,39 +1,19 @@
-"""SneakyCode entrypoint — argument parsing, config loading, and interactive REPL."""
+"""SneakyCode entrypoint — argument parsing, config loading, and TUI launch."""
 
 import argparse
 import asyncio
-import signal
 import sys
 from pathlib import Path
 
-import structlog
-
-from app.agent.context import SessionContext
-from app.agent.loop import AgentLoop
 from app.models.config import AppConfig, load_config
 from app.services.llm import LLMClient, LLMConnectionError, LLMError
-from app.services.permissions import PermissionsService
 from app.services.session import SessionManager
-from app.services.streaming import StreamHandler
-from app.tools.registry import create_default_registry
-from app.utils.display import (
-    print_banner,
-    print_error,
-    print_history,
-    print_info,
-    print_success,
-    print_user_message,
-    print_warning,
-)
-from app.utils.logging import console, get_logger, setup_logging
+from app.utils.display import print_banner, print_error, print_info, print_success
+from app.utils.logging import get_logger, setup_logging
 
 
 def parse_args() -> argparse.Namespace:
-    """Parse command-line arguments.
-
-    Returns:
-        Parsed arguments namespace.
-    """
+    """Parse command-line arguments."""
     parser = argparse.ArgumentParser(
         prog="sneakycode",
         description="SneakyCode — A privacy-first local AI coding agent",
@@ -65,127 +45,11 @@ async def _preflight(config: AppConfig) -> None:
         await client.preflight_check()
 
 
-def _offer_session_resume(
-    session_mgr: SessionManager, ctx: SessionContext
-) -> bool:
-    """Check for a saved session and offer to resume it.
-
-    Args:
-        session_mgr: Session manager instance.
-        ctx: Session context to restore into.
-
-    Returns:
-        True if a session was restored.
-    """
-    saved = session_mgr.load_latest()
-    if saved is None:
-        return False
-
-    msg_count = len(saved.messages)
-    print_info(f"Found previous session ({msg_count} messages, model: {saved.model})")
-
-    try:
-        answer = console.input("[bold cyan]Resume previous session? [y/N] [/bold cyan]").strip().lower()
-    except (KeyboardInterrupt, EOFError):
-        return False
-
-    if answer in ("y", "yes"):
-        session_mgr.restore(saved, ctx)
-        print_success(f"Session restored ({msg_count} messages).")
-        return True
-
-    return False
-
-
-def _save_session_quiet(session_mgr: SessionManager, ctx: SessionContext) -> None:
-    """Save session without raising on errors."""
-    try:
-        if ctx.message_count > 0:
-            session_mgr.save(ctx)
-    except OSError as e:
-        print_warning(f"Could not save session: {e}")
-
-
-async def _run_repl(
-    ctx: SessionContext,
-    config: AppConfig,
-    session_mgr: SessionManager,
-    logger: structlog.stdlib.BoundLogger,
-    shutdown_event: asyncio.Event,
-) -> None:
-    """Run the interactive REPL loop with streaming LLM responses.
-
-    Args:
-        ctx: Session context for conversation state.
-        config: Application configuration.
-        session_mgr: Session manager for auto-save.
-        logger: Structured logger instance.
-        shutdown_event: Event signalling graceful shutdown.
-    """
-    registry = create_default_registry(config.agent.workspace_root, config)
-    permissions = PermissionsService(config.permissions)
-
-    async with LLMClient(config.llm) as client:
-        handler = StreamHandler(config.display)
-        agent = AgentLoop(config, ctx, client, handler, registry, permissions)
-
-        while not shutdown_event.is_set():
-            try:
-                user_input = console.input("[bold cyan]> [/bold cyan]")
-            except (KeyboardInterrupt, EOFError):
-                _save_session_quiet(session_mgr, ctx)
-                console.print("\n[dim]Session saved. Goodbye![/dim]")
-                break
-
-            user_input = user_input.strip()
-            if not user_input:
-                continue
-
-            # Handle slash commands
-            if user_input.startswith("/"):
-                command = user_input.lower()
-                if command == "/quit":
-                    _save_session_quiet(session_mgr, ctx)
-                    console.print("[dim]Session saved. Goodbye![/dim]")
-                    break
-                elif command == "/history":
-                    print_history(ctx.get_history())
-                elif command == "/clear":
-                    ctx.clear_history()
-                    print_success("Conversation history cleared.")
-                elif command == "/save":
-                    try:
-                        path = session_mgr.save(ctx)
-                        print_success(f"Session saved to {path}")
-                    except OSError as e:
-                        print_error(f"Failed to save session: {e}")
-                elif command == "/session":
-                    print_info(f"Messages: {ctx.message_count}")
-                    print_info(f"Tokens: ~{ctx.estimated_tokens:,} / {ctx.token_counter.budget:,}")
-                    print_info(f"Started: {ctx.start_time.isoformat()}")
-                else:
-                    print_warning(f"Unknown command: {user_input}")
-                continue
-
-            print_user_message(user_input)
-            await agent.run_turn(user_input)
-            logger.debug("turn_complete", message_count=ctx.message_count)
-
-            # Auto-save after each turn
-            if config.session.auto_save:
-                _save_session_quiet(session_mgr, ctx)
-
-        # Shutdown triggered by signal
-        if shutdown_event.is_set():
-            _save_session_quiet(session_mgr, ctx)
-            console.print("\n[dim]Session saved. Shutting down gracefully.[/dim]")
-
-
 def main() -> None:
-    """Main entrypoint: load config, setup logging, launch interactive REPL."""
+    """Main entrypoint: load config, preflight check, launch Textual TUI."""
     args = parse_args()
 
-    # Setup logging first
+    # Setup logging first (will be reconfigured for TUI on mount)
     setup_logging(
         log_file=args.log_file,
         verbose=args.verbose,
@@ -201,7 +65,7 @@ def main() -> None:
 
     logger.info("config_loaded", model=config.llm.model, endpoint=config.llm.endpoint)
 
-    # Print startup info
+    # Pre-TUI startup info (printed to console before Textual takes over)
     print_banner()
     print_info(f"Model: {config.llm.model}")
     print_info(f"Endpoint: {config.llm.endpoint}")
@@ -222,8 +86,7 @@ def main() -> None:
 
     print_success("Ollama connected, model ready.")
 
-    # Create session and session manager
-    ctx = SessionContext(config)
+    # Create session manager
     session_mgr = SessionManager(config.session, config.agent.workspace_root, config.llm.model)
 
     # Clean up old session files
@@ -231,28 +94,11 @@ def main() -> None:
     if cleaned > 0:
         logger.info("old_sessions_cleaned", count=cleaned)
 
-    # Offer to resume previous session
-    if config.session.offer_resume:
-        _offer_session_resume(session_mgr, ctx)
+    # Launch Textual TUI
+    from app.ui.app import SneakyCodeApp
 
-    logger.info("startup_complete")
-
-    # Setup shutdown event and SIGTERM handler
-    shutdown_event = asyncio.Event()
-
-    original_sigterm = signal.getsignal(signal.SIGTERM)
-
-    def _sigterm_handler(signum: int, frame: object) -> None:
-        shutdown_event.set()
-
-    signal.signal(signal.SIGTERM, _sigterm_handler)
-
-    print_info("Commands: /quit, /history, /clear, /save, /session")
-
-    try:
-        asyncio.run(_run_repl(ctx, config, session_mgr, logger, shutdown_event))
-    finally:
-        signal.signal(signal.SIGTERM, original_sigterm)
+    app = SneakyCodeApp(config, session_mgr=session_mgr)
+    app.run()
 
 
 if __name__ == "__main__":

app/services/permissions.py

@@ -1,25 +1,42 @@
 """Permission gating for tool execution."""
 
-import logging
+from __future__ import annotations
 
-from rich.prompt import Confirm
+import logging
+from collections.abc import Awaitable, Callable
 
 from app.models.config import PermissionsConfig
 
 logger = logging.getLogger(__name__)
 
+# Type alias for the async prompt callback
+PromptCallback = Callable[[str, str], Awaitable[bool]]
+
 
 class PermissionDenied(Exception):
     """Raised when a tool is denied execution by permissions policy."""
 
 
 class PermissionsService:
-    """Check whether a tool is allowed to execute based on config tiers."""
+    """Check whether a tool is allowed to execute based on config tiers.
+
+    In TUI mode, set a prompt callback via set_prompt_callback() that
+    shows a modal dialog. Without a callback, unlisted tools are denied.
+    """
 
     def __init__(self, config: PermissionsConfig) -> None:
         self.config = config
+        self._prompt_callback: PromptCallback | None = None
 
-    def check(self, tool_name: str, description: str = "") -> bool:
+    def set_prompt_callback(self, callback: PromptCallback) -> None:
+        """Set the async callback used to prompt the user for permission.
+
+        Args:
+            callback: Async function(tool_name, description) -> bool.
+        """
+        self._prompt_callback = callback
+
+    async def check(self, tool_name: str, description: str = "") -> bool:
         """Check if a tool is permitted to run.
 
         Returns:
@@ -33,14 +50,10 @@ class PermissionsService:
             logger.debug("Tool '%s' is auto-approved", tool_name)
             return True
 
-        # Explicit prompt_user list or unlisted tools both trigger a prompt
-        return self._prompt_user(tool_name, description)
+        # Prompt user via callback (TUI modal, etc.)
+        if self._prompt_callback is not None:
+            return await self._prompt_callback(tool_name, description)
 
-    def _prompt_user(self, tool_name: str, description: str) -> bool:
-        """Prompt the user for approval via the terminal."""
-        prompt_text = f"Allow tool [bold]{tool_name}[/bold]"
-        if description:
-            prompt_text += f" — {description}"
-        prompt_text += "?"
-
-        return Confirm.ask(prompt_text, default=False)
+        # No callback set — deny by default (safe fallback)
+        logger.warning("Tool '%s' requires approval but no prompt callback set — denied", tool_name)
+        return False

app/services/streaming.py

@@ -1,41 +1,56 @@
 """Streaming response handler — accumulates SSE chunks into a complete Message."""
 
-from collections.abc import AsyncIterator
-
-from rich.live import Live
-from rich.markdown import Markdown
-from rich.panel import Panel
+import time
+from collections.abc import AsyncIterator, Callable
 
 from app.models.config import DisplayConfig
 from app.models.message import Message
 from app.models.tool_call import ToolCall, ToolCallFunction
-from app.utils.logging import console, get_logger
+from app.utils.logging import get_logger
 from app.utils.token_counter import TokenUsage
 
 logger = get_logger(__name__)
 
+# Minimum interval between content update callbacks (seconds)
+_UPDATE_THROTTLE_INTERVAL = 0.1
+
 
 class StreamHandler:
-    """Processes an SSE chunk stream into a Rich live display and final Message.
+    """Processes an SSE chunk stream and produces a complete assistant Message.
 
-    Accumulates content deltas and tool call fragments, renders a live Markdown
-    panel during streaming, and produces a complete assistant Message on finish.
+    Accumulates content deltas and tool call fragments. Notifies the UI via
+    optional callbacks during streaming.
     """
 
     def __init__(self, display_config: DisplayConfig) -> None:
-        """Initialize the stream handler.
-
-        Args:
-            display_config: Display preferences (streaming toggle, etc.).
-        """
         self._display_config = display_config
         self._accumulated_content: str = ""
         self._accumulated_reasoning: str = ""
         self._tool_calls: dict[int, dict[str, str]] = {}
         self._usage: TokenUsage | None = None
+        self._on_content: Callable[[str], None] | None = None
+        self._on_thinking: Callable[[], None] | None = None
+        self._on_done: Callable[[], None] | None = None
+
+    def set_callbacks(
+        self,
+        on_content: Callable[[str], None] | None = None,
+        on_thinking: Callable[[], None] | None = None,
+        on_done: Callable[[], None] | None = None,
+    ) -> None:
+        """Set UI callbacks for streaming updates.
+
+        Args:
+            on_content: Called with accumulated content string (throttled to ~100ms).
+            on_thinking: Called once when first reasoning token arrives.
+            on_done: Called when streaming is complete.
+        """
+        self._on_content = on_content
+        self._on_thinking = on_thinking
+        self._on_done = on_done
 
     async def process_stream(self, chunk_iter: AsyncIterator[dict]) -> Message:
-        """Consume a chunk iterator, rendering live output and returning the final Message.
+        """Consume a chunk iterator and return the final Message.
 
         Args:
             chunk_iter: Async iterator of parsed SSE chunk dicts.
@@ -43,26 +58,42 @@ class StreamHandler:
         Returns:
             Complete assistant Message with accumulated content and tool calls.
         """
-        with Live(console=console, refresh_per_second=8) as live:
-            async for chunk in chunk_iter:
-                self._process_chunk(chunk)
+        thinking_notified = False
+        last_update_time = 0.0
 
-                # Show reasoning while waiting for content
-                display_text = self._accumulated_content
-                if not display_text and self._accumulated_reasoning:
-                    display_text = "*thinking...*"
+        async for chunk in chunk_iter:
+            self._process_chunk(chunk)
 
-                if display_text and self._display_config.stream_output:
-                    # Render inside the same Assistant panel used for final output
-                    # so the live display and final frame are visually consistent
-                    live.update(
-                        Panel(
-                            Markdown(display_text),
-                            title="Assistant",
-                            border_style="green",
-                            expand=True,
-                        )
-                    )
+            if not self._display_config.stream_output:
+                continue
+
+            # Notify thinking once
+            if (
+                not thinking_notified
+                and not self._accumulated_content
+                and self._accumulated_reasoning
+                and self._on_thinking is not None
+            ):
+                self._on_thinking()
+                thinking_notified = True
+
+            # Throttled content updates
+            if self._accumulated_content and self._on_content is not None:
+                now = time.monotonic()
+                if now - last_update_time >= _UPDATE_THROTTLE_INTERVAL:
+                    self._on_content(self._accumulated_content)
+                    last_update_time = now
+
+        # Final content update (ensures last chunk is shown)
+        if (
+            self._display_config.stream_output
+            and self._accumulated_content
+            and self._on_content is not None
+        ):
+            self._on_content(self._accumulated_content)
+
+        if self._on_done is not None:
+            self._on_done()
 
         tool_calls = self._build_tool_calls() or None
         return Message(
@@ -72,12 +103,7 @@ class StreamHandler:
         )
 
     def _process_chunk(self, chunk: dict) -> None:
-        """Extract content, tool calls, and usage from a single SSE chunk.
-
-        Args:
-            chunk: Parsed JSON dict from one SSE data line.
-        """
-        # Content delta
+        """Extract content, tool calls, and usage from a single SSE chunk."""
         choices = chunk.get("choices", [])
         if choices:
             delta = choices[0].get("delta", {})
@@ -86,12 +112,10 @@ class StreamHandler:
             if content_piece:
                 self._accumulated_content += content_piece
 
-            # Reasoning tokens (e.g. qwen3.5 thinking mode)
             reasoning_piece = delta.get("reasoning")
             if reasoning_piece:
                 self._accumulated_reasoning += reasoning_piece
 
-            # Tool call deltas (accumulated by index)
             for tc_delta in delta.get("tool_calls", []):
                 idx = tc_delta.get("index", 0)
                 if idx not in self._tool_calls:
@@ -109,7 +133,6 @@ class StreamHandler:
                 if func.get("arguments"):
                     entry["arguments"] += func["arguments"]
 
-        # Token usage (typically in the final chunk)
         usage_data = chunk.get("usage")
         if usage_data:
             self._usage = TokenUsage(
@@ -119,11 +142,7 @@ class StreamHandler:
             )
 
     def _build_tool_calls(self) -> list[ToolCall]:
-        """Convert accumulated tool call fragments into sorted ToolCall list.
-
-        Returns:
-            List of ToolCall objects sorted by stream index.
-        """
+        """Convert accumulated tool call fragments into sorted ToolCall list."""
         if not self._tool_calls:
             return []
 
@@ -143,14 +162,7 @@ class StreamHandler:
         return result
 
     def get_partial_message(self) -> Message | None:
-        """Return whatever content/tool_calls have been accumulated so far.
-
-        Useful for mid-stream interruption recovery — returns None if nothing
-        has been accumulated yet.
-
-        Returns:
-            Partial assistant Message, or None if no content accumulated.
-        """
+        """Return whatever content/tool_calls have been accumulated so far."""
         tool_calls = self._build_tool_calls() or None
         if not self._accumulated_content and not tool_calls:
             return None
@@ -176,3 +188,6 @@ class StreamHandler:
         self._accumulated_reasoning = ""
         self._tool_calls.clear()
         self._usage = None
+        self._on_content = None
+        self._on_thinking = None
+        self._on_done = None

app/ui/app.py

@@ -0,0 +1,237 @@
+"""SneakyCode Textual TUI application."""
+
+from __future__ import annotations
+
+from pathlib import Path
+from typing import TYPE_CHECKING
+
+from rich.markdown import Markdown
+from rich.panel import Panel
+from rich.text import Text
+from textual.app import App, ComposeResult
+from textual.binding import Binding
+from textual.widgets import Header, Input, RichLog
+
+from app.agent.context import SessionContext
+from app.agent.loop import AgentLoop
+from app.models.config import AppConfig
+from app.services.llm import LLMClient
+from app.services.permissions import PermissionsService
+from app.services.session import SessionManager
+from app.services.streaming import StreamHandler
+from app.tools.registry import create_default_registry
+from app.ui.widgets import StatusBar, StreamingStatic
+from app.utils.display import DisplayAdapter, render_user_message
+from app.utils.logging import get_logger, setup_logging_for_tui
+
+if TYPE_CHECKING:
+    from textual.worker import Worker
+
+logger = get_logger(__name__)
+
+
+class SneakyCodeApp(App):
+    """Main TUI application for SneakyCode."""
+
+    TITLE = "SneakyCode"
+    CSS_PATH = "styles.tcss"
+
+    BINDINGS = [
+        Binding("ctrl+c", "cancel_or_quit", "Cancel/Quit", show=False),
+    ]
+
+    def __init__(self, config: AppConfig, session_mgr: SessionManager | None = None) -> None:
+        super().__init__()
+        self._config = config
+        self._session_mgr = session_mgr
+        self._ctx: SessionContext | None = None
+        self._agent: AgentLoop | None = None
+        self._client: LLMClient | None = None
+        self._tool_registry = None
+        self._permissions: PermissionsService | None = None
+        self._current_worker: Worker | None = None
+        self._cancel_count = 0
+        self.sub_title = config.llm.model
+
+    def compose(self) -> ComposeResult:
+        yield Header()
+        yield RichLog(id="chat-log", highlight=True, markup=True)
+        yield StreamingStatic("", id="streaming")
+        yield StatusBar()
+        yield Input(placeholder="Enter your prompt...")
+
+    async def on_mount(self) -> None:
+        """Initialize agent components after the app is mounted."""
+        setup_logging_for_tui()
+
+        self._ctx = SessionContext(self._config)
+
+        # Create long-lived agent dependencies (reused across turns)
+        self._client = LLMClient(self._config.llm)
+        await self._client.__aenter__()
+        self._tool_registry = create_default_registry(self._config.agent.workspace_root, self._config)
+        self._permissions = PermissionsService(self._config.permissions)
+
+        # Set up permission prompt callback
+        async def permission_prompt(tool_name: str, description: str) -> bool:
+            return await self._show_permission_modal(tool_name, description)
+
+        self._permissions.set_prompt_callback(permission_prompt)
+
+        # Offer session resume if configured
+        if self._session_mgr and self._config.session.offer_resume:
+            saved = self._session_mgr.load_latest()
+            if saved:
+                msg_count = len(saved.messages)
+                log = self.query_one("#chat-log", RichLog)
+                log.write(Text(f"Found previous session ({msg_count} messages)", style="cyan"))
+                # Auto-resume for now (modal can be added later)
+                self._session_mgr.restore(saved, self._ctx)
+                log.write(Text(f"✓ Session restored ({msg_count} messages)", style="bold green"))
+
+        self.query_one(Input).focus()
+
+    async def on_input_submitted(self, event: Input.Submitted) -> None:
+        """Handle user input submission."""
+        user_input = event.value.strip()
+        if not user_input:
+            return
+
+        event.input.clear()
+        log = self.query_one("#chat-log", RichLog)
+
+        # Handle slash commands
+        if user_input.startswith("/"):
+            await self._handle_slash_command(user_input, log)
+            return
+
+        # Write user message to log
+        log.write(render_user_message(user_input))
+
+        # Dispatch agent turn as async worker
+        self._cancel_count = 0
+        self._current_worker = self.run_worker(
+            self._run_agent_turn(user_input),
+            name="agent-turn",
+            exclusive=True,
+        )
+
+    async def _handle_slash_command(self, command: str, log: RichLog) -> None:
+        """Process slash commands."""
+        cmd = command.lower()
+        if cmd == "/quit":
+            self._save_session()
+            self.exit()
+        elif cmd == "/clear":
+            if self._ctx:
+                self._ctx.clear_history()
+            log.clear()
+            log.write(Text("✓ Conversation history cleared.", style="bold green"))
+        elif cmd == "/history":
+            if self._ctx:
+                from app.utils.display import render_history
+                log.write(render_history(self._ctx.get_history()))
+        elif cmd == "/save":
+            path = self._save_session()
+            if path:
+                log.write(Text(f"✓ Session saved to {path}", style="bold green"))
+            else:
+                log.write(Text("✗ No session to save", style="bold red"))
+        elif cmd == "/session":
+            if self._ctx:
+                log.write(Text(
+                    f"Messages: {self._ctx.message_count} | "
+                    f"Tokens: ~{self._ctx.estimated_tokens:,} / {self._ctx.token_counter.budget:,} | "
+                    f"Started: {self._ctx.start_time.isoformat()}",
+                    style="cyan",
+                ))
+        else:
+            log.write(Text(f"⚠ Unknown command: {command}", style="yellow"))
+
+    async def _run_agent_turn(self, user_input: str) -> None:
+        """Run a single agent turn (called as a worker)."""
+        if self._ctx is None or self._client is None:
+            return
+
+        log = self.query_one("#chat-log", RichLog)
+        streaming_widget = self.query_one("#streaming", StreamingStatic)
+        display = DisplayAdapter(log)
+
+        # StreamHandler is per-turn (has per-turn accumulators)
+        handler = StreamHandler(self._config.display)
+
+        # Set up streaming UI callbacks
+        def on_content(content: str) -> None:
+            streaming_widget.update(
+                Panel(Markdown(content), title="Assistant", border_style="green", expand=True)
+            )
+            streaming_widget.show_streaming()
+
+        def on_thinking() -> None:
+            streaming_widget.update(Text("Thinking...", style="dim"))
+            streaming_widget.show_streaming()
+
+        def on_done() -> None:
+            streaming_widget.hide_streaming()
+
+        handler.set_callbacks(on_content=on_content, on_thinking=on_thinking, on_done=on_done)
+
+        agent = AgentLoop(
+            self._config, self._ctx, self._client, handler,
+            self._tool_registry, self._permissions, display,
+        )
+
+        await agent.run_turn(user_input)
+
+        # Auto-save
+        if self._config.session.auto_save:
+            self._save_session()
+
+    async def _show_permission_modal(self, tool_name: str, description: str) -> bool:
+        """Show a permission prompt in the chat log and wait for response.
+
+        Uses the Input widget to capture y/n response.
+        """
+        # For v1, auto-approve all tools that reach the prompt
+        # TODO: Implement proper modal dialog
+        log = self.query_one("#chat-log", RichLog)
+        log.write(Text(f"⚠ Auto-approved: {tool_name} ({description})", style="yellow"))
+        return True
+
+    def action_cancel_or_quit(self) -> None:
+        """Handle Ctrl+C: first press cancels worker, second press quits."""
+        self._cancel_count += 1
+        if self._cancel_count >= 2 or self._current_worker is None:
+            self._save_session()
+            self.exit()
+        elif self._current_worker is not None:
+            self._current_worker.cancel()
+            log = self.query_one("#chat-log", RichLog)
+            log.write(Text("⚠ Cancelling... (press Ctrl+C again to quit)", style="yellow"))
+
+    def on_worker_state_changed(self, event: Worker.StateChanged) -> None:
+        """Handle worker completion or failure."""
+        from textual.worker import WorkerState
+
+        if event.worker.name != "agent-turn":
+            return
+
+        if event.state == WorkerState.ERROR:
+            log = self.query_one("#chat-log", RichLog)
+            error = event.worker.error
+            log.write(Text(f"✗ Agent error: {error}", style="bold red"))
+
+        if event.state in (WorkerState.SUCCESS, WorkerState.ERROR, WorkerState.CANCELLED):
+            self._current_worker = None
+            # Hide streaming widget in case it was left visible
+            streaming = self.query_one("#streaming", StreamingStatic)
+            streaming.hide_streaming()
+
+    def _save_session(self) -> Path | None:
+        """Save session quietly, return path or None."""
+        if self._session_mgr and self._ctx and self._ctx.message_count > 0:
+            try:
+                return self._session_mgr.save(self._ctx)
+            except OSError as e:
+                logger.warning("session_save_failed", error=str(e))
+        return None

app/ui/styles.tcss

@@ -0,0 +1,33 @@
+/* SneakyCode TUI Layout */
+
+Screen {
+    layout: vertical;
+}
+
+#chat-log {
+    height: 1fr;
+    border: none;
+    scrollbar-gutter: stable;
+}
+
+#streaming {
+    display: none;
+    height: auto;
+    max-height: 50%;
+    padding: 0 1;
+}
+
+#streaming.visible {
+    display: block;
+}
+
+/* StatusBar styles are in DEFAULT_CSS on the widget itself */
+
+Input {
+    dock: bottom;
+    margin: 0;
+}
+
+Header {
+    dock: top;
+}

app/ui/widgets.py

@@ -0,0 +1,65 @@
+"""Custom Textual widgets for SneakyCode TUI."""
+
+from __future__ import annotations
+
+from rich.text import Text
+from textual.widgets import Static
+
+
+class StatusBar(Static):
+    """Single-line status bar showing token usage and iteration count."""
+
+    DEFAULT_CSS = """
+    StatusBar {
+        dock: bottom;
+        height: 1;
+        background: $surface;
+        color: $text-muted;
+        padding: 0 2;
+    }
+    """
+
+    def __init__(self) -> None:
+        super().__init__("")
+        self._tokens: int = 0
+        self._budget: int = 0
+        self._iteration: int = 0
+        self._max_iterations: int = 0
+
+    def update_tokens(self, tokens: int, budget: int) -> None:
+        """Update the token usage display."""
+        self._tokens = tokens
+        self._budget = budget
+        self._refresh_display()
+
+    def update_iteration(self, iteration: int, max_iterations: int) -> None:
+        """Update the iteration count display."""
+        self._iteration = iteration
+        self._max_iterations = max_iterations
+        self._refresh_display()
+
+    def _refresh_display(self) -> None:
+        """Rebuild the status bar text."""
+        parts: list[str] = []
+        if self._budget > 0:
+            parts.append(f"Tokens: ~{self._tokens:,} / {self._budget:,}")
+        if self._max_iterations > 0:
+            parts.append(f"Iteration {self._iteration}/{self._max_iterations}")
+        self.update(Text(" │ ".join(parts), style="dim"))
+
+
+class StreamingStatic(Static):
+    """A Static widget that stays mounted but hidden during non-streaming periods.
+
+    During streaming, call show() to make visible and update() with partial content.
+    When streaming ends, call hide() to conceal.
+    """
+
+    def show_streaming(self) -> None:
+        """Make the widget visible for streaming."""
+        self.add_class("visible")
+
+    def hide_streaming(self) -> None:
+        """Hide the widget and clear content."""
+        self.remove_class("visible")
+        self.update("")

app/utils/display.py

@@ -1,7 +1,18 @@
-"""Rich terminal display helpers for SneakyCode."""
+"""Rich terminal display helpers for SneakyCode.
+
+Provides two interfaces:
+- render_* functions: return Rich renderables (for use in Textual RichLog)
+- DisplayAdapter: wraps a RichLog widget and provides write_* convenience methods
+- print_* functions: legacy console.print wrappers (for non-TUI fallback)
+"""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING, Protocol
 
 from rich.panel import Panel
 from rich.table import Table
+from rich.text import Text
 from rich.theme import Theme
 
 from app.models.message import Message
@@ -23,6 +34,166 @@ SNEAKYCODE_THEME = Theme(
 console.push_theme(SNEAKYCODE_THEME)
 
 
+if TYPE_CHECKING:
+    from rich.console import RenderableType
+
+
+# ---------------------------------------------------------------------------
+# Render functions — return Rich renderables
+# ---------------------------------------------------------------------------
+
+
+def render_user_message(content: str) -> Panel:
+    """Render a user message as a styled panel."""
+    return Panel(content, title="You", border_style="cyan", expand=False)
+
+
+def render_assistant_message(content: str) -> Panel:
+    """Render an assistant message as a styled panel."""
+    return Panel(content, title="Assistant", border_style="green", expand=True)
+
+
+def render_tool_call(name: str, args: str) -> Text:
+    """Render a compact tool call line."""
+    truncated_args = args[:80] + "..." if len(args) > 80 else args
+    text = Text()
+    text.append("  ")
+    text.append(name, style="magenta")
+    text.append(f" {truncated_args}", style="dim")
+    return text
+
+
+def render_tool_result(name: str, output: str, is_error: bool = False) -> Text:
+    """Render a compact tool result line."""
+    text = Text()
+    text.append("  ")
+    if is_error:
+        truncated = output[:200] + "..." if len(output) > 200 else output
+        text.append(f"{name}: {truncated}", style="bold red")
+    else:
+        lines = output.count("\n") + 1 if output else 0
+        chars = len(output)
+        text.append(f"{name} — {lines} lines, {chars} chars", style="dim")
+    return text
+
+
+def render_iteration_header(iteration: int, max_iterations: int) -> Text:
+    """Render the current agent loop iteration."""
+    return Text(f"── iteration {iteration}/{max_iterations} ──", style="dim")
+
+
+def render_token_usage(usage_tokens: int, budget: int) -> Text:
+    """Render token usage as styled text."""
+    return Text(f"Tokens: ~{usage_tokens:,} / {budget:,}", style="dim")
+
+
+def render_warning(message: str) -> Text:
+    """Render a warning message."""
+    return Text(f"⚠ {message}", style="yellow")
+
+
+def render_error(message: str) -> Text:
+    """Render an error message."""
+    return Text(f"✗ {message}", style="bold red")
+
+
+def render_success(message: str) -> Text:
+    """Render a success message."""
+    return Text(f"✓ {message}", style="bold green")
+
+
+def render_info(message: str) -> Text:
+    """Render an info message."""
+    return Text(message, style="cyan")
+
+
+def render_history(messages: list[Message]) -> Table | Text:
+    """Render conversation history as a Rich table."""
+    if not messages:
+        return Text("No messages in history.", style="dim")
+
+    table = Table(title="Conversation History")
+    table.add_column("#", style="dim", width=4)
+    table.add_column("Role", width=10)
+    table.add_column("Content")
+
+    role_styles = {
+        "user": "cyan",
+        "assistant": "green",
+        "system": "yellow",
+        "tool": "magenta",
+    }
+
+    for i, msg in enumerate(messages, 1):
+        style = role_styles.get(msg.role, "white")
+        content = msg.content or "(no content)"
+        if len(content) > 120:
+            content = content[:117] + "..."
+        table.add_row(str(i), f"[{style}]{msg.role}[/{style}]", content)
+
+    return table
+
+
+# ---------------------------------------------------------------------------
+# DisplayAdapter — wraps a writable target (RichLog or similar)
+# ---------------------------------------------------------------------------
+
+
+class WritableLog(Protocol):
+    """Protocol for anything that accepts Rich renderables via write()."""
+
+    def write(self, content: "RenderableType") -> None: ...
+
+
+class DisplayAdapter:
+    """Bridges agent loop display calls to a RichLog widget.
+
+    All write_* methods call the corresponding render_* function and
+    write the result to the target log.
+    """
+
+    def __init__(self, log: WritableLog) -> None:
+        self._log = log
+
+    def write_user_message(self, content: str) -> None:
+        self._log.write(render_user_message(content))
+
+    def write_assistant_message(self, content: str) -> None:
+        self._log.write(render_assistant_message(content))
+
+    def write_tool_call(self, name: str, args: str) -> None:
+        self._log.write(render_tool_call(name, args))
+
+    def write_tool_result(self, name: str, output: str, is_error: bool = False) -> None:
+        self._log.write(render_tool_result(name, output, is_error))
+
+    def write_iteration_header(self, iteration: int, max_iterations: int) -> None:
+        self._log.write(render_iteration_header(iteration, max_iterations))
+
+    def write_token_usage(self, usage_tokens: int, budget: int) -> None:
+        self._log.write(render_token_usage(usage_tokens, budget))
+
+    def write_warning(self, message: str) -> None:
+        self._log.write(render_warning(message))
+
+    def write_error(self, message: str) -> None:
+        self._log.write(render_error(message))
+
+    def write_success(self, message: str) -> None:
+        self._log.write(render_success(message))
+
+    def write_info(self, message: str) -> None:
+        self._log.write(render_info(message))
+
+    def write_history(self, messages: list[Message]) -> None:
+        self._log.write(render_history(messages))
+
+
+# ---------------------------------------------------------------------------
+# Legacy print functions — for non-TUI fallback and pre-TUI startup
+# ---------------------------------------------------------------------------
+
+
 def print_banner() -> None:
     """Print the SneakyCode startup banner."""
     console.print(
@@ -61,25 +232,17 @@ def print_assistant_message(content: str) -> None:
 
 
 def print_tool_call(name: str, args: str) -> None:
-    """Print a compact tool call line — tool name + truncated key args."""
+    """Print a compact tool call line."""
     truncated_args = args[:80] + "..." if len(args) > 80 else args
     console.print(f"  [tool]{name}[/tool] [dim]{truncated_args}[/dim]")
 
 
 def print_tool_result(name: str, output: str, is_error: bool = False) -> None:
-    """Print a compact tool result — status line only for success, detail for errors.
-
-    Args:
-        name: Tool name.
-        output: Tool output or error message.
-        is_error: Whether this is an error result.
-    """
+    """Print a compact tool result."""
     if is_error:
-        # Errors are shown prominently so the user knows something went wrong
         truncated = output[:200] + "..." if len(output) > 200 else output
         console.print(f"  [error]{name}: {truncated}[/error]")
     else:
-        # Success: just show a compact byte/line summary
         lines = output.count("\n") + 1 if output else 0
         chars = len(output)
         console.print(f"  [dim]{name} — {lines} lines, {chars} chars[/dim]")
@@ -96,33 +259,5 @@ def print_token_usage(usage_tokens: int, budget: int) -> None:
 
 
 def print_history(messages: list[Message]) -> None:
-    """Print conversation history as a Rich table.
-
-    Args:
-        messages: List of conversation messages to display.
-    """
-    if not messages:
-        console.print("[dim]No messages in history.[/dim]")
-        return
-
-    table = Table(title="Conversation History")
-    table.add_column("#", style="dim", width=4)
-    table.add_column("Role", width=10)
-    table.add_column("Content")
-
-    role_styles = {
-        "user": "cyan",
-        "assistant": "green",
-        "system": "yellow",
-        "tool": "magenta",
-    }
-
-    for i, msg in enumerate(messages, 1):
-        style = role_styles.get(msg.role, "white")
-        content = msg.content or "[dim](no content)[/dim]"
-        # Truncate long content for display
-        if len(content) > 120:
-            content = content[:117] + "..."
-        table.add_row(str(i), f"[{style}]{msg.role}[/{style}]", content)
-
-    console.print(table)
+    """Print conversation history as a Rich table."""
+    console.print(render_history(messages))

app/utils/logging.py

@@ -77,6 +77,21 @@ def setup_logging(
     )
 
 
+def setup_logging_for_tui() -> None:
+    """Reconfigure logging for Textual TUI mode.
+
+    Removes the RichHandler (which writes to stdout and corrupts the TUI)
+    while preserving any file handlers. Call this from the Textual App's
+    on_mount() before any agent work begins.
+    """
+    from rich.logging import RichHandler
+
+    root_logger = logging.getLogger()
+    root_logger.handlers = [
+        h for h in root_logger.handlers if not isinstance(h, RichHandler)
+    ]
+
+
 def get_logger(name: str) -> structlog.stdlib.BoundLogger:
     """Get a named structlog logger.
 

pyproject.toml

@@ -13,6 +13,7 @@ dependencies = [
     "pyyaml>=6.0",
     "httpx>=0.27",
     "structlog>=24.0",
+    "textual>=4.0.0",
 ]
 
 [project.optional-dependencies]

tests/unit/test_agent_loop.py

@@ -21,6 +21,7 @@ from app.services.llm import LLMClient
 from app.services.permissions import PermissionsService
 from app.services.streaming import StreamHandler
 from app.tools.registry import ToolRegistry, create_default_registry
+from app.utils.display import DisplayAdapter
 
 
 @pytest.fixture
@@ -76,6 +77,11 @@ def permissions(config: AppConfig) -> PermissionsService:
     return PermissionsService(config.permissions)
 
 
+@pytest.fixture
+def display() -> MagicMock:
+    return MagicMock(spec=DisplayAdapter)
+
+
 @pytest.fixture
 def agent(
     config: AppConfig,
@@ -84,8 +90,9 @@ def agent(
     handler: MagicMock,
     registry: ToolRegistry,
     permissions: PermissionsService,
+    display: MagicMock,
 ) -> AgentLoop:
-    return AgentLoop(config, ctx, client, handler, registry, permissions)
+    return AgentLoop(config, ctx, client, handler, registry, permissions, display)
 
 
 def _make_text_message(content: str) -> Message:

tests/unit/test_display.py

@@ -0,0 +1,87 @@
+"""Tests for display render functions and DisplayAdapter."""
+
+from unittest.mock import MagicMock
+
+from rich.panel import Panel
+from rich.table import Table
+from rich.text import Text
+
+from app.models.message import Message
+from app.utils.display import (
+    DisplayAdapter,
+    render_assistant_message,
+    render_error,
+    render_iteration_header,
+    render_tool_call,
+    render_tool_result,
+    render_token_usage,
+    render_user_message,
+    render_warning,
+)
+
+
+class TestRenderFunctions:
+    def test_render_user_message_returns_panel(self) -> None:
+        result = render_user_message("hello")
+        assert isinstance(result, Panel)
+        assert result.title == "You"
+
+    def test_render_assistant_message_returns_panel(self) -> None:
+        result = render_assistant_message("response")
+        assert isinstance(result, Panel)
+        assert result.title == "Assistant"
+
+    def test_render_tool_call_returns_text(self) -> None:
+        result = render_tool_call("read_file", '{"path": "foo.py"}')
+        assert isinstance(result, Text)
+        assert "read_file" in result.plain
+
+    def test_render_tool_result_success(self) -> None:
+        result = render_tool_result("read_file", "file contents here", is_error=False)
+        assert isinstance(result, Text)
+        assert "read_file" in result.plain
+
+    def test_render_tool_result_error(self) -> None:
+        result = render_tool_result("read_file", "not found", is_error=True)
+        assert isinstance(result, Text)
+        assert "read_file" in result.plain
+
+    def test_render_iteration_header(self) -> None:
+        result = render_iteration_header(3, 25)
+        assert isinstance(result, Text)
+        assert "3/25" in result.plain
+
+    def test_render_token_usage(self) -> None:
+        result = render_token_usage(1500, 32000)
+        assert isinstance(result, Text)
+        assert "1,500" in result.plain
+
+    def test_render_warning(self) -> None:
+        result = render_warning("something happened")
+        assert isinstance(result, Text)
+
+    def test_render_error(self) -> None:
+        result = render_error("bad thing")
+        assert isinstance(result, Text)
+
+
+class TestDisplayAdapter:
+    def test_write_user_message(self) -> None:
+        mock_log = MagicMock()
+        adapter = DisplayAdapter(mock_log)
+        adapter.write_user_message("hello")
+        mock_log.write.assert_called_once()
+        arg = mock_log.write.call_args[0][0]
+        assert isinstance(arg, Panel)
+
+    def test_write_tool_call(self) -> None:
+        mock_log = MagicMock()
+        adapter = DisplayAdapter(mock_log)
+        adapter.write_tool_call("read_file", '{"path": "x"}')
+        mock_log.write.assert_called_once()
+
+    def test_write_warning(self) -> None:
+        mock_log = MagicMock()
+        adapter = DisplayAdapter(mock_log)
+        adapter.write_warning("oops")
+        mock_log.write.assert_called_once()

tests/unit/test_logging_tui.py

@@ -0,0 +1,31 @@
+"""Tests for TUI-safe logging configuration."""
+
+import logging
+
+from app.utils.logging import setup_logging, setup_logging_for_tui
+
+
+class TestTuiLogging:
+    def test_setup_logging_for_tui_removes_rich_handler(self) -> None:
+        """TUI mode should have no RichHandler on root logger."""
+        from rich.logging import RichHandler
+
+        # First set up normal logging (adds RichHandler)
+        setup_logging()
+        root = logging.getLogger()
+        assert any(isinstance(h, RichHandler) for h in root.handlers)
+
+        # Switch to TUI mode
+        setup_logging_for_tui()
+        root = logging.getLogger()
+        assert not any(isinstance(h, RichHandler) for h in root.handlers)
+
+    def test_setup_logging_for_tui_keeps_file_handler(self, tmp_path) -> None:
+        """TUI mode should preserve file handler if configured."""
+        log_file = tmp_path / "test.log"
+        setup_logging(log_file=log_file)
+        setup_logging_for_tui()
+
+        root = logging.getLogger()
+        file_handlers = [h for h in root.handlers if isinstance(h, logging.FileHandler)]
+        assert len(file_handlers) == 1

tests/unit/test_permissions.py

@@ -0,0 +1,51 @@
+"""Tests for async PermissionsService."""
+
+import asyncio
+
+import pytest
+
+from app.models.config import PermissionsConfig
+from app.services.permissions import PermissionsService
+
+
+class TestPermissionsService:
+    @pytest.mark.asyncio
+    async def test_auto_approve(self) -> None:
+        config = PermissionsConfig(auto_approve=["read_file"])
+        svc = PermissionsService(config)
+        assert await svc.check("read_file") is True
+
+    @pytest.mark.asyncio
+    async def test_deny(self) -> None:
+        config = PermissionsConfig(deny=["rm_file"])
+        svc = PermissionsService(config)
+        assert await svc.check("rm_file") is False
+
+    @pytest.mark.asyncio
+    async def test_prompt_callback_approve(self) -> None:
+        config = PermissionsConfig()
+        svc = PermissionsService(config)
+
+        async def approve_callback(tool_name: str, description: str) -> bool:
+            return True
+
+        svc.set_prompt_callback(approve_callback)
+        assert await svc.check("write_file", description="write something") is True
+
+    @pytest.mark.asyncio
+    async def test_prompt_callback_deny(self) -> None:
+        config = PermissionsConfig()
+        svc = PermissionsService(config)
+
+        async def deny_callback(tool_name: str, description: str) -> bool:
+            return False
+
+        svc.set_prompt_callback(deny_callback)
+        assert await svc.check("write_file") is False
+
+    @pytest.mark.asyncio
+    async def test_no_callback_defaults_to_deny(self) -> None:
+        """Without a callback set, unlisted tools are denied."""
+        config = PermissionsConfig()
+        svc = PermissionsService(config)
+        assert await svc.check("write_file") is False

tests/unit/test_streaming.py

@@ -0,0 +1,111 @@
+"""Tests for callback-based StreamHandler."""
+
+import asyncio
+from unittest.mock import MagicMock, call
+
+import pytest
+
+from app.models.config import DisplayConfig
+from app.services.streaming import StreamHandler
+
+
+def _make_chunk(content: str | None = None, reasoning: str | None = None) -> dict:
+    """Helper to create a fake SSE chunk."""
+    delta: dict = {}
+    if content is not None:
+        delta["content"] = content
+    if reasoning is not None:
+        delta["reasoning"] = reasoning
+    return {"choices": [{"delta": delta}]}
+
+
+def _make_tool_call_chunk(index: int, tc_id: str = "", name: str = "", args: str = "") -> dict:
+    """Helper to create a fake tool call chunk."""
+    tc_delta: dict = {"index": index}
+    if tc_id:
+        tc_delta["id"] = tc_id
+    func: dict = {}
+    if name:
+        func["name"] = name
+    if args:
+        func["arguments"] = args
+    if func:
+        tc_delta["function"] = func
+    return {"choices": [{"delta": {"tool_calls": [tc_delta]}}]}
+
+
+async def _async_iter(items: list[dict]):
+    for item in items:
+        yield item
+
+
+class TestStreamHandlerCallbacks:
+    @pytest.mark.asyncio
+    async def test_on_content_called_with_accumulated_text(self) -> None:
+        handler = StreamHandler(DisplayConfig(stream_output=True))
+        on_content = MagicMock()
+        on_thinking = MagicMock()
+        on_done = MagicMock()
+        handler.set_callbacks(on_content=on_content, on_thinking=on_thinking, on_done=on_done)
+
+        chunks = [_make_chunk(content="Hello"), _make_chunk(content=" world")]
+        msg = await handler.process_stream(_async_iter(chunks))
+
+        assert msg.content == "Hello world"
+        assert on_content.call_count >= 1
+        # Last call should have full accumulated content
+        last_content = on_content.call_args_list[-1][0][0]
+        assert last_content == "Hello world"
+        on_done.assert_called_once()
+
+    @pytest.mark.asyncio
+    async def test_on_thinking_called_for_reasoning(self) -> None:
+        handler = StreamHandler(DisplayConfig(stream_output=True))
+        on_content = MagicMock()
+        on_thinking = MagicMock()
+        on_done = MagicMock()
+        handler.set_callbacks(on_content=on_content, on_thinking=on_thinking, on_done=on_done)
+
+        chunks = [_make_chunk(reasoning="let me think")]
+        msg = await handler.process_stream(_async_iter(chunks))
+
+        on_thinking.assert_called_once()
+        on_content.assert_not_called()
+        on_done.assert_called_once()
+
+    @pytest.mark.asyncio
+    async def test_display_callbacks_skip_when_stream_output_disabled(self) -> None:
+        """on_content and on_thinking are suppressed, but on_done always fires."""
+        handler = StreamHandler(DisplayConfig(stream_output=False))
+        on_content = MagicMock()
+        on_thinking = MagicMock()
+        on_done = MagicMock()
+        handler.set_callbacks(on_content=on_content, on_thinking=on_thinking, on_done=on_done)
+
+        chunks = [_make_chunk(content="Hello")]
+        msg = await handler.process_stream(_async_iter(chunks))
+
+        assert msg.content == "Hello"
+        on_content.assert_not_called()
+        on_thinking.assert_not_called()
+        on_done.assert_called_once()
+
+    @pytest.mark.asyncio
+    async def test_no_callbacks_by_default(self) -> None:
+        """process_stream works without set_callbacks (backward compat)."""
+        handler = StreamHandler(DisplayConfig(stream_output=True))
+        chunks = [_make_chunk(content="Hello")]
+        msg = await handler.process_stream(_async_iter(chunks))
+        assert msg.content == "Hello"
+
+    @pytest.mark.asyncio
+    async def test_tool_calls_still_accumulated(self) -> None:
+        handler = StreamHandler(DisplayConfig(stream_output=True))
+        chunks = [
+            _make_tool_call_chunk(0, tc_id="call_1", name="read_file"),
+            _make_tool_call_chunk(0, args='{"path": "foo.py"}'),
+        ]
+        msg = await handler.process_stream(_async_iter(chunks))
+        assert msg.tool_calls is not None
+        assert len(msg.tool_calls) == 1
+        assert msg.tool_calls[0].function.name == "read_file"

tests/unit/test_tools.py

@@ -1,7 +1,6 @@
 """Tests for the tool framework and core tools (Phase 4)."""
 
 from pathlib import Path
-from unittest.mock import patch
 
 import pytest
 
@@ -48,28 +47,40 @@ class TestBaseTool:
 
 
 class TestPermissionsService:
-    def test_deny_list_blocks(self) -> None:
+    @pytest.mark.asyncio
+    async def test_deny_list_blocks(self) -> None:
         svc = PermissionsService(PermissionsConfig(deny=["dangerous_tool"]))
-        assert svc.check("dangerous_tool") is False
+        assert await svc.check("dangerous_tool") is False
 
-    def test_auto_approve_allows(self) -> None:
+    @pytest.mark.asyncio
+    async def test_auto_approve_allows(self) -> None:
         svc = PermissionsService(PermissionsConfig(auto_approve=["read_file"]))
-        assert svc.check("read_file") is True
+        assert await svc.check("read_file") is True
 
-    @patch("app.services.permissions.Confirm.ask", return_value=True)
-    def test_prompt_user_approved(self, mock_ask: object) -> None:
+    @pytest.mark.asyncio
+    async def test_prompt_callback_approved(self) -> None:
         svc = PermissionsService(PermissionsConfig(prompt_user=["write_file"]))
-        assert svc.check("write_file") is True
 
-    @patch("app.services.permissions.Confirm.ask", return_value=False)
-    def test_prompt_user_denied(self, mock_ask: object) -> None:
+        async def approve(tool_name: str, description: str) -> bool:
+            return True
+
+        svc.set_prompt_callback(approve)
+        assert await svc.check("write_file") is True
+
+    @pytest.mark.asyncio
+    async def test_prompt_callback_denied(self) -> None:
         svc = PermissionsService(PermissionsConfig(prompt_user=["write_file"]))
-        assert svc.check("write_file") is False
 
-    @patch("app.services.permissions.Confirm.ask", return_value=False)
-    def test_unlisted_tool_prompts(self, mock_ask: object) -> None:
+        async def deny(tool_name: str, description: str) -> bool:
+            return False
+
+        svc.set_prompt_callback(deny)
+        assert await svc.check("write_file") is False
+
+    @pytest.mark.asyncio
+    async def test_unlisted_tool_no_callback_denied(self) -> None:
         svc = PermissionsService(PermissionsConfig())
-        assert svc.check("unknown_tool") is False
+        assert await svc.check("unknown_tool") is False
 
 
 # --- ToolRegistry ---

uv.lock

@@ -97,6 +97,18 @@ wheels = [
     { url = "https://files.pythonhosted.org/packages/cb/b1/3846dd7f199d53cb17f49cba7e651e9ce294d8497c8c150530ed11865bb8/iniconfig-2.3.0-py3-none-any.whl", hash = "sha256:f631c04d2c48c52b84d0d0549c99ff3859c98df65b3101406327ecc7d53fbf12", size = 7484, upload-time = "2025-10-18T21:55:41.639Z" },
 ]
 
+[[package]]
+name = "linkify-it-py"
+version = "2.1.0"
+source = { registry = "https://pypi.org/simple" }
+dependencies = [
+    { name = "uc-micro-py" },
+]
+sdist = { url = "https://files.pythonhosted.org/packages/2e/c9/06ea13676ef354f0af6169587ae292d3e2406e212876a413bf9eece4eb23/linkify_it_py-2.1.0.tar.gz", hash = "sha256:43360231720999c10e9328dc3691160e27a718e280673d444c38d7d3aaa3b98b", size = 29158, upload-time = "2026-03-01T07:48:47.683Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/b4/de/88b3be5c31b22333b3ca2f6ff1de4e863d8fe45aaea7485f591970ec1d3e/linkify_it_py-2.1.0-py3-none-any.whl", hash = "sha256:0d252c1594ecba2ecedc444053db5d3a9b7ec1b0dd929c8f1d74dce89f86c05e", size = 19878, upload-time = "2026-03-01T07:48:46.098Z" },
+]
+
 [[package]]
 name = "markdown-it-py"
 version = "4.0.0"
@@ -109,6 +121,23 @@ wheels = [
     { url = "https://files.pythonhosted.org/packages/94/54/e7d793b573f298e1c9013b8c4dade17d481164aa517d1d7148619c2cedbf/markdown_it_py-4.0.0-py3-none-any.whl", hash = "sha256:87327c59b172c5011896038353a81343b6754500a08cd7a4973bb48c6d578147", size = 87321, upload-time = "2025-08-11T12:57:51.923Z" },
 ]
 
+[package.optional-dependencies]
+linkify = [
+    { name = "linkify-it-py" },
+]
+
+[[package]]
+name = "mdit-py-plugins"
+version = "0.5.0"
+source = { registry = "https://pypi.org/simple" }
+dependencies = [
+    { name = "markdown-it-py" },
+]
+sdist = { url = "https://files.pythonhosted.org/packages/b2/fd/a756d36c0bfba5f6e39a1cdbdbfdd448dc02692467d83816dff4592a1ebc/mdit_py_plugins-0.5.0.tar.gz", hash = "sha256:f4918cb50119f50446560513a8e311d574ff6aaed72606ddae6d35716fe809c6", size = 44655, upload-time = "2025-08-11T07:25:49.083Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/fb/86/dd6e5db36df29e76c7a7699123569a4a18c1623ce68d826ed96c62643cae/mdit_py_plugins-0.5.0-py3-none-any.whl", hash = "sha256:07a08422fc1936a5d26d146759e9155ea466e842f5ab2f7d2266dd084c8dab1f", size = 57205, upload-time = "2025-08-11T07:25:47.597Z" },
+]
+
 [[package]]
 name = "mdurl"
 version = "0.1.2"
@@ -127,6 +156,15 @@ wheels = [
     { url = "https://files.pythonhosted.org/packages/b7/b9/c538f279a4e237a006a2c98387d081e9eb060d203d8ed34467cc0f0b9b53/packaging-26.0-py3-none-any.whl", hash = "sha256:b36f1fef9334a5588b4166f8bcd26a14e521f2b55e6b9de3aaa80d3ff7a37529", size = 74366, upload-time = "2026-01-21T20:50:37.788Z" },
 ]
 
+[[package]]
+name = "platformdirs"
+version = "4.9.4"
+source = { registry = "https://pypi.org/simple" }
+sdist = { url = "https://files.pythonhosted.org/packages/19/56/8d4c30c8a1d07013911a8fdbd8f89440ef9f08d07a1b50ab8ca8be5a20f9/platformdirs-4.9.4.tar.gz", hash = "sha256:1ec356301b7dc906d83f371c8f487070e99d3ccf9e501686456394622a01a934", size = 28737, upload-time = "2026-03-05T18:34:13.271Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/63/d7/97f7e3a6abb67d8080dd406fd4df842c2be0efaf712d1c899c32a075027c/platformdirs-4.9.4-py3-none-any.whl", hash = "sha256:68a9a4619a666ea6439f2ff250c12a853cd1cbd5158d258bd824a7df6be2f868", size = 21216, upload-time = "2026-03-05T18:34:12.172Z" },
+]
+
 [[package]]
 name = "pluggy"
 version = "1.6.0"
@@ -389,6 +427,7 @@ dependencies = [
     { name = "pyyaml" },
     { name = "rich" },
     { name = "structlog" },
+    { name = "textual" },
 ]
 
 [package.optional-dependencies]
@@ -408,6 +447,7 @@ requires-dist = [
     { name = "rich", specifier = ">=13.0" },
     { name = "ruff", marker = "extra == 'dev'", specifier = ">=0.3" },
     { name = "structlog", specifier = ">=24.0" },
+    { name = "textual", specifier = ">=4.0.0" },
 ]
 provides-extras = ["dev"]
 
@@ -420,6 +460,23 @@ wheels = [
     { url = "https://files.pythonhosted.org/packages/a8/45/a132b9074aa18e799b891b91ad72133c98d8042c70f6240e4c5f9dabee2f/structlog-25.5.0-py3-none-any.whl", hash = "sha256:a8453e9b9e636ec59bd9e79bbd4a72f025981b3ba0f5837aebf48f02f37a7f9f", size = 72510, upload-time = "2025-10-27T08:28:21.535Z" },
 ]
 
+[[package]]
+name = "textual"
+version = "8.1.1"
+source = { registry = "https://pypi.org/simple" }
+dependencies = [
+    { name = "markdown-it-py", extra = ["linkify"] },
+    { name = "mdit-py-plugins" },
+    { name = "platformdirs" },
+    { name = "pygments" },
+    { name = "rich" },
+    { name = "typing-extensions" },
+]
+sdist = { url = "https://files.pythonhosted.org/packages/72/23/8c709655c5f2208ee82ab81b8104802421865535c278a7649b842b129db1/textual-8.1.1.tar.gz", hash = "sha256:eef0256a6131f06a20ad7576412138c1f30f92ddeedd055953c08d97044bc317", size = 1843002, upload-time = "2026-03-10T10:01:38.493Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/50/21/421b02bf5943172b7a9320712a5e0d74a02a8f7597284e3f8b5b06c70b8d/textual-8.1.1-py3-none-any.whl", hash = "sha256:6712f96e335cd782e76193dee16b9c8875fe0699d923bc8d3f1228fd23e773a6", size = 719598, upload-time = "2026-03-10T10:01:48.318Z" },
+]
+
 [[package]]
 name = "typing-extensions"
 version = "4.15.0"
@@ -440,3 +497,12 @@ sdist = { url = "https://files.pythonhosted.org/packages/55/e3/70399cb7dd41c10ac
 wheels = [
     { url = "https://files.pythonhosted.org/packages/dc/9b/47798a6c91d8bdb567fe2698fe81e0c6b7cb7ef4d13da4114b41d239f65d/typing_inspection-0.4.2-py3-none-any.whl", hash = "sha256:4ed1cacbdc298c220f1bd249ed5287caa16f34d44ef4e9c3d0cbad5b521545e7", size = 14611, upload-time = "2025-10-01T02:14:40.154Z" },
 ]
+
+[[package]]
+name = "uc-micro-py"
+version = "2.0.0"
+source = { registry = "https://pypi.org/simple" }
+sdist = { url = "https://files.pythonhosted.org/packages/78/67/9a363818028526e2d4579334460df777115bdec1bb77c08f9db88f6389f2/uc_micro_py-2.0.0.tar.gz", hash = "sha256:c53691e495c8db60e16ffc4861a35469b0ba0821fe409a8a7a0a71864d33a811", size = 6611, upload-time = "2026-03-01T06:31:27.526Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/61/73/d21edf5b204d1467e06500080a50f79d49ef2b997c79123a536d4a17d97c/uc_micro_py-2.0.0-py3-none-any.whl", hash = "sha256:3603a3859af53e5a39bc7677713c78ea6589ff188d70f4fee165db88e22b242c", size = 6383, upload-time = "2026-03-01T06:31:26.257Z" },
+]