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>

.gitignore

@@ -31,3 +31,6 @@ htmlcov/
 
 # uv
 .python-version
+
+# Worktrees
+.worktrees/

README.md

@@ -0,0 +1,147 @@
+# SneakyCode
+
+A privacy-first, locally-running Python coding agent that uses a local LLM (via Ollama) to perform autonomous coding tasks inside a project directory.
+
+SneakyCode accepts natural language tasks and executes them using a defined toolset for filesystem operations, shell execution, code search, and file manipulation. It runs a ReAct-style tool-call loop: send conversation history to the LLM, receive tool calls, execute them with permission checks, and feed results back until the task is complete.
+
+## Prerequisites
+
+- Python 3.11+
+- [Ollama](https://ollama.ai/) running locally with a model that supports function calling (e.g., `qwen3.5`, `llama3.1`, `mistral-nemo`)
+- [uv](https://docs.astral.sh/uv/) (recommended) or pip
+
+## Installation
+
+```bash
+# Clone the repository
+git clone <repo-url>
+cd SneakyCode
+
+# Install dependencies
+uv sync --dev
+
+# Or with pip
+pip install -e ".[dev]"
+```
+
+## Configuration
+
+Edit `config/config.yaml` to configure the agent. Key settings:
+
+```yaml
+llm:
+  model: "qwen3.5:latest"        # Ollama model name
+  endpoint: "http://localhost:11434"  # Ollama endpoint
+  max_retries: 3                   # Retry attempts on transient errors
+  retry_backoff_base: 1.0          # Exponential backoff base (seconds)
+
+agent:
+  max_iterations: 25               # Max tool-call iterations per turn
+  max_conversation_tokens: 32000   # Token budget for conversation
+  workspace_root: "."              # Project directory for file operations
+  truncation_keep_recent: 10       # Messages preserved during truncation
+  truncation_threshold: 0.85       # Budget fraction that triggers truncation
+
+session:
+  auto_save: true                  # Save session after each turn
+  max_session_age_hours: 72        # Auto-cleanup old sessions
+  offer_resume: true               # Offer to resume on startup
+
+permissions:
+  auto_approve: [read_file, list_dir, grep_files, find_files, finish]
+  prompt_user: [write_file, delete_file, run_command, str_replace, patch_apply, make_dir]
+  deny: []
+```
+
+Environment variable `SNEAKYCODE_CONFIG` can override the config file path.
+
+## Usage
+
+```bash
+# Start the interactive REPL
+sneakycode
+
+# Or run directly
+python -m app.main
+
+# With options
+sneakycode --config path/to/config.yaml --verbose --log-file sneakycode.log
+```
+
+### REPL Commands
+
+| Command    | Description                          |
+|------------|--------------------------------------|
+| `/quit`    | Save session and exit                |
+| `/history` | Show conversation history            |
+| `/clear`   | Clear conversation history           |
+| `/save`    | Manually save session                |
+| `/session` | Show session info (messages, tokens) |
+
+### Session Persistence
+
+Sessions are automatically saved after each agent turn and on exit. On startup, SneakyCode offers to resume the most recent session for the current workspace.
+
+Session files are stored in `.sneakycode/sessions/` within the workspace root.
+
+## Available Tools
+
+SneakyCode provides 11 tools across 5 categories. See [docs/tools.md](docs/tools.md) for the full reference.
+
+| Category   | Tools                                           | Permission    |
+|------------|-------------------------------------------------|---------------|
+| Read       | `read_file`, `list_dir`                         | Auto-approved |
+| Search     | `grep_files`, `find_files`                      | Auto-approved |
+| Write      | `write_file`, `make_dir`, `delete_file`         | User confirm  |
+| Edit       | `str_replace`, `patch_apply`                    | User confirm  |
+| Shell      | `run_command`                                   | User confirm  |
+| Control    | `finish`                                        | Auto-approved |
+
+## Development
+
+```bash
+# Run tests
+.venv/bin/python -m pytest tests/ -v
+
+# Run with coverage
+.venv/bin/python -m pytest tests/ --cov=app
+
+# Lint
+.venv/bin/ruff check app/ tests/
+
+# Format
+.venv/bin/ruff format app/ tests/
+```
+
+### Project Structure
+
+```
+app/
+├── agent/           # Agent loop and session context
+├── models/          # Pydantic config and message schemas
+├── services/        # LLM client, streaming, permissions, session persistence
+├── tools/           # Tool implementations (one file per group)
+└── utils/           # Logging, display, file helpers, token counter
+config/
+└── config.yaml      # Application configuration
+tests/
+├── unit/            # Unit tests for individual components
+└── integration/     # End-to-end workflow tests with mocked LLM
+```
+
+## Architecture
+
+SneakyCode follows a **ReAct-style** agent pattern:
+
+1. User provides a task in natural language
+2. Agent sends conversation history + tool schemas to the LLM
+3. LLM responds with either text (task complete) or tool calls
+4. Agent executes tool calls with permission checks
+5. Results are fed back to the LLM for the next iteration
+6. Loop continues until the LLM produces a plain-text response or calls `finish`
+
+The LLM client is abstracted behind an OpenAI-compatible interface, so any endpoint implementing the `/v1/chat/completions` SSE streaming protocol works as a backend.
+
+## License
+
+MIT

app/agent/context.py

@@ -77,3 +77,101 @@ class SessionContext:
     def start_time(self) -> datetime:
         """Session start timestamp (UTC)."""
         return self._start_time
+
+    def truncate_history(self, system_token_estimate: int = 0) -> int:
+        """Drop oldest messages to bring token usage under budget.
+
+        Preserves the first user message and the most recent N messages
+        (configured by ``truncation_keep_recent``). Cleans up orphaned tool
+        messages after truncation.
+
+        Args:
+            system_token_estimate: Estimated tokens used by the system prompt.
+
+        Returns:
+            Number of messages dropped.
+        """
+        budget = self._token_counter.budget
+        threshold = self._config.agent.truncation_threshold
+        keep_recent = self._config.agent.truncation_keep_recent
+
+        estimated = self._token_counter.estimate_messages_tokens(self._history) + system_token_estimate
+        if estimated < threshold * budget:
+            return 0
+
+        target = int(budget * 0.75)  # headroom
+        if len(self._history) <= keep_recent + 1:
+            return 0
+
+        # Split: first user message | droppable middle | recent tail
+        first_msg = self._history[0] if self._history and self._history[0].role == "user" else None
+        start_idx = 1 if first_msg else 0
+        tail_start = max(start_idx, len(self._history) - keep_recent)
+
+        dropped = 0
+        drop_indices: set[int] = set()
+
+        for i in range(start_idx, tail_start):
+            drop_indices.add(i)
+            dropped += 1
+            # Recalculate with remaining messages
+            remaining = [m for j, m in enumerate(self._history) if j not in drop_indices]
+            est = self._token_counter.estimate_messages_tokens(remaining) + system_token_estimate
+            if est < target:
+                break
+
+        if dropped == 0:
+            return 0
+
+        self._history = [m for j, m in enumerate(self._history) if j not in drop_indices]
+
+        # Clean up orphaned tool messages
+        self._cleanup_orphaned_tool_messages()
+
+        return dropped
+
+    def _cleanup_orphaned_tool_messages(self) -> None:
+        """Remove tool messages whose tool_call_id doesn't match any assistant tool_call."""
+        # Collect all tool_call IDs from assistant messages
+        valid_tc_ids: set[str] = set()
+        for msg in self._history:
+            if msg.role == "assistant" and msg.tool_calls:
+                for tc in msg.tool_calls:
+                    valid_tc_ids.add(tc.id)
+
+        # Remove tool messages referencing missing tool calls
+        self._history = [
+            msg for msg in self._history
+            if msg.role != "tool" or (msg.tool_call_id and msg.tool_call_id in valid_tc_ids)
+        ]
+
+    def to_serializable(self) -> dict:
+        """Export messages and token state for session persistence.
+
+        Returns:
+            Dict with messages and token usage data.
+        """
+        return {
+            "messages": [m.model_dump(exclude_none=True) for m in self._history],
+            "token_usage": self._token_counter.cumulative_usage.model_dump(),
+        }
+
+    def restore_from(self, data: dict) -> None:
+        """Clear and replay from serialized data.
+
+        Args:
+            data: Dict with messages and optional token_usage as produced by to_serializable().
+        """
+        self._history.clear()
+        self._message_count = 0
+
+        for msg_data in data.get("messages", []):
+            msg = Message(**msg_data)
+            self._history.append(msg)
+            self._message_count += 1
+
+        token_data = data.get("token_usage")
+        if token_data:
+            from app.utils.token_counter import TokenUsage
+            usage = TokenUsage(**token_data)
+            self._token_counter.count_usage(usage)

app/agent/loop.py

@@ -7,18 +7,11 @@ from app.agent.context import SessionContext
 from app.models.config import AppConfig
 from app.models.message import Message
 from app.models.tool_call import ToolCall, ToolResult, ToolResultStatus
-from app.services.llm import LLMClient, LLMConnectionError, LLMError
+from app.services.llm import LLMClient, LLMConnectionError, LLMError, LLMStreamError
 from app.services.permissions import PermissionsService
 from app.services.streaming import StreamHandler
 from app.tools.registry import ToolRegistry
-from app.utils.display import (
+from app.utils.display import DisplayAdapter
-    print_error,
-    print_iteration_header,
-    print_tool_call,
-    print_tool_result,
-    print_token_usage,
-    print_warning,
-)
 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,8 +43,14 @@ 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
+
+    def cancel(self) -> None:
+        """Request cancellation of the current agent turn."""
+        self._cancelled = True
 
     def _build_system_prompt(self) -> str:
         """Build the system prompt including tool schemas and agent instructions."""
@@ -81,17 +81,31 @@ class AgentLoop:
             user_input: The user's message text.
         """
         self._ctx.add_message("user", user_input)
+        self._cancelled = False
 
         max_iter = self._config.agent.max_iterations
         reasoning_only_streak = 0
         for iteration in range(1, max_iter + 1):
-            # Check token budget
+            if self._cancelled:
-            if self._ctx.token_counter.is_over_budget():
+                if self._display:
-                print_warning("Token budget exceeded. Stopping agent loop.")
+                    self._display.write_warning("Agent loop cancelled.")
                 break
 
+            # Check token budget — try truncation before giving up
+            if self._ctx.token_counter.is_over_budget():
+                system_tokens = self._ctx.token_counter.estimate_tokens(self._system_prompt)
+                dropped = self._ctx.truncate_history(system_tokens)
+                if dropped > 0:
+                    if self._display:
+                        self._display.write_warning(f"Token budget pressure: dropped {dropped} oldest messages.")
+                else:
+                    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()
@@ -112,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()
 
@@ -127,17 +141,19 @@ class AgentLoop:
 
                 if reasoning_only_streak >= _MAX_REASONING_RETRIES:
                     # Nudge the model by injecting a user hint
-                    print_warning(
+                    if self._display:
-                        f"Model produced reasoning but no response {reasoning_only_streak} times. "
+                        self._display.write_warning(
-                        "Nudging model to respond..."
+                            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
@@ -145,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:
@@ -164,30 +182,42 @@ 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.
 
+        Uses retry-enabled streaming. On mid-stream errors, attempts to recover
+        partial content if available.
+
         Returns:
             The assistant Message, or None if an error occurred.
         """
         messages = self._get_messages_with_system_prompt()
         try:
-            chunk_iter = self._client.stream_chat(messages, tools=self._tools_schema)
+            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 as e:
+        except (LLMConnectionError, LLMStreamError) as e:
-            print_error(f"Connection error: {e}")
+            partial = self._handler.get_partial_message()
+            if partial is not None:
+                if self._display:
+                    self._display.write_warning(f"Stream interrupted ({e}), returning partial response.")
+                return partial
+            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:
@@ -204,8 +234,8 @@ class AgentLoop:
             tc_id = tc.id
 
             # Display the tool call
-            if self._config.display.show_tool_calls:
+            if self._config.display.show_tool_calls and self._display:
-                print_tool_call(name, tc.function.arguments)
+                self._display.write_tool_call(name, tc.function.arguments)
 
             # Parse arguments
             try:
@@ -218,8 +248,8 @@ class AgentLoop:
                     error=f"Invalid JSON in arguments: {e}",
                 )
                 results.append(result)
-                if self._config.display.show_tool_calls:
+                if self._config.display.show_tool_calls and self._display:
-                    print_tool_result(name, result.error or "", is_error=True)
+                    self._display.write_tool_result(name, result.error or "", is_error=True)
                 continue
 
             # Look up tool
@@ -232,13 +262,13 @@ class AgentLoop:
                     error=f"Unknown tool '{name}'. Available: {available_names}",
                 )
                 results.append(result)
-                if self._config.display.show_tool_calls:
+                if self._config.display.show_tool_calls and self._display:
-                    print_tool_result(name, result.error or "", is_error=True)
+                    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,
@@ -246,17 +276,17 @@ class AgentLoop:
                     error=f"Permission denied for tool '{name}'",
                 )
                 results.append(result)
-                if self._config.display.show_tool_calls:
+                if self._config.display.show_tool_calls and self._display:
-                    print_tool_result(name, result.error or "", is_error=True)
+                    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,37 +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 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.utils.display import print_banner, print_error, print_info, print_success
-from app.tools.registry import create_default_registry
+from app.utils.logging import get_logger, setup_logging
-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
 
 
 def parse_args() -> argparse.Namespace:
-    """Parse command-line arguments.
+    """Parse command-line arguments."""
-
-    Returns:
-        Parsed arguments namespace.
-    """
     parser = argparse.ArgumentParser(
         prog="sneakycode",
         description="SneakyCode — A privacy-first local AI coding agent",
@@ -63,61 +45,11 @@ async def _preflight(config: AppConfig) -> None:
         await client.preflight_check()
 
 
-async def _run_repl(
-    ctx: SessionContext,
-    config: AppConfig,
-    logger: structlog.stdlib.BoundLogger,
-) -> None:
-    """Run the interactive REPL loop with streaming LLM responses.
-
-    Args:
-        ctx: Session context for conversation state.
-        config: Application configuration.
-        logger: Structured logger instance.
-    """
-    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 True:
-            try:
-                user_input = console.input("[bold cyan]> [/bold cyan]")
-            except (KeyboardInterrupt, EOFError):
-                console.print("\n[dim]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":
-                    console.print("[dim]Goodbye![/dim]")
-                    break
-                elif command == "/history":
-                    print_history(ctx.get_history())
-                elif command == "/clear":
-                    ctx.clear_history()
-                    print_success("Conversation history cleared.")
-                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)
-
-
 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,
@@ -133,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}")
@@ -154,12 +86,19 @@ def main() -> None:
 
     print_success("Ollama connected, model ready.")
 
-    # Create session and start REPL
+    # Create session manager
-    ctx = SessionContext(config)
+    session_mgr = SessionManager(config.session, config.agent.workspace_root, config.llm.model)
-    logger.info("startup_complete")
 
-    print_info("Commands: /quit, /history, /clear")
+    # Clean up old session files
-    asyncio.run(_run_repl(ctx, config, logger))
+    cleaned = session_mgr.cleanup_old()
+    if cleaned > 0:
+        logger.info("old_sessions_cleaned", count=cleaned)
+
+    # Launch Textual TUI
+    from app.ui.app import SneakyCodeApp
+
+    app = SneakyCodeApp(config, session_mgr=session_mgr)
+    app.run()
 
 
 if __name__ == "__main__":

app/models/config.py

@@ -16,6 +16,9 @@ class LLMConfig(BaseModel):
     temperature: float = Field(default=0.1, description="Sampling temperature")
     max_tokens: int = Field(default=4096, description="Maximum tokens in LLM response")
     timeout: int = Field(default=120, description="Request timeout in seconds")
+    max_retries: int = Field(default=3, description="Max retry attempts on transient errors")
+    retry_backoff_base: float = Field(default=1.0, description="Base seconds for exponential backoff")
+    retry_backoff_max: float = Field(default=30.0, description="Maximum backoff seconds")
 
 
 class AgentConfig(BaseModel):
@@ -28,6 +31,12 @@ class AgentConfig(BaseModel):
     workspace_root: Path = Field(
         default=Path("."), description="Root directory for file operations"
     )
+    truncation_keep_recent: int = Field(
+        default=10, description="Number of recent messages to preserve during truncation"
+    )
+    truncation_threshold: float = Field(
+        default=0.85, description="Token budget fraction that triggers truncation"
+    )
 
 
 class PermissionsConfig(BaseModel):
@@ -60,6 +69,19 @@ class ToolsConfig(BaseModel):
     filesystem: FilesystemToolConfig = Field(default_factory=FilesystemToolConfig)
 
 
+class SessionConfig(BaseModel):
+    """Session persistence configuration."""
+
+    session_dir: Path = Field(
+        default=Path(".sneakycode/sessions"), description="Directory for session files"
+    )
+    auto_save: bool = Field(default=True, description="Auto-save session after each turn")
+    max_session_age_hours: int = Field(
+        default=72, description="Max age in hours before session files are cleaned up"
+    )
+    offer_resume: bool = Field(default=True, description="Offer to resume previous sessions on startup")
+
+
 class DisplayConfig(BaseModel):
     """Terminal display preferences."""
 
@@ -76,6 +98,7 @@ class AppConfig(BaseModel):
     permissions: PermissionsConfig = Field(default_factory=PermissionsConfig)
     tools: ToolsConfig = Field(default_factory=ToolsConfig)
     display: DisplayConfig = Field(default_factory=DisplayConfig)
+    session: SessionConfig = Field(default_factory=SessionConfig)
 
     @model_validator(mode="after")
     def resolve_workspace_root(self) -> "AppConfig":

app/services/llm.py

@@ -1,6 +1,8 @@
 """LLM client wrapper for Ollama / OpenAI-compatible endpoints."""
 
+import asyncio
 import json
+import random
 from collections.abc import AsyncIterator
 from typing import Any, Self
 
@@ -162,6 +164,61 @@ class LLMClient:
         except httpx.HTTPError as e:
             raise LLMError(f"HTTP error communicating with LLM: {e}") from e
 
+    async def stream_chat_with_retry(
+        self,
+        messages: list[Message],
+        tools: list[dict[str, Any]] | None = None,
+    ) -> AsyncIterator[dict]:
+        """Stream chat with automatic retry on transient errors.
+
+        Retries on LLMConnectionError and LLMResponseError with status >= 500.
+        Does NOT retry on 4xx errors (client-side, not transient).
+        Uses exponential backoff with jitter.
+
+        Args:
+            messages: Conversation history to send to the model.
+            tools: Optional OpenAI function-calling tool schemas.
+
+        Yields:
+            Parsed JSON dicts from each SSE data line.
+
+        Raises:
+            LLMConnectionError: After exhausting retries on connection failures.
+            LLMResponseError: After exhausting retries on server errors, or immediately on 4xx.
+        """
+        max_retries = self._config.max_retries
+        last_exception: LLMError | None = None
+
+        for attempt in range(max_retries + 1):
+            try:
+                async for chunk in self.stream_chat(messages, tools=tools):
+                    yield chunk
+                return
+            except LLMConnectionError as e:
+                last_exception = e
+            except LLMResponseError as e:
+                if e.status_code is not None and e.status_code < 500:
+                    raise
+                last_exception = e
+            except LLMStreamError as e:
+                last_exception = e
+
+            if attempt < max_retries:
+                backoff = min(
+                    self._config.retry_backoff_base * (2 ** attempt) + random.uniform(0, 1),
+                    self._config.retry_backoff_max,
+                )
+                logger.warning(
+                    "llm_retry",
+                    attempt=attempt + 1,
+                    max_retries=max_retries,
+                    backoff_seconds=round(backoff, 2),
+                    error=str(last_exception),
+                )
+                await asyncio.sleep(backoff)
+
+        raise last_exception  # type: ignore[misc]
+
     async def close(self) -> None:
         """Close the underlying HTTP client."""
         await self._client.aclose()

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
+        # Prompt user via callback (TUI modal, etc.)
-        return self._prompt_user(tool_name, description)
+        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:
+        # No callback set — deny by default (safe fallback)
-        """Prompt the user for approval via the terminal."""
+        logger.warning("Tool '%s' requires approval but no prompt callback set — denied", tool_name)
-        prompt_text = f"Allow tool [bold]{tool_name}[/bold]"
+        return False
-        if description:
-            prompt_text += f" — {description}"
-        prompt_text += "?"
-
-        return Confirm.ask(prompt_text, default=False)

app/services/session.py

@@ -0,0 +1,148 @@
+"""Session persistence — auto-save and restore conversation state."""
+
+import hashlib
+import json
+from datetime import UTC, datetime
+from pathlib import Path
+from typing import TYPE_CHECKING
+
+from pydantic import BaseModel, Field
+
+from app.models.config import SessionConfig
+from app.utils.logging import get_logger
+
+if TYPE_CHECKING:
+    from app.agent.context import SessionContext
+
+logger = get_logger(__name__)
+
+
+class SessionData(BaseModel):
+    """Serialized session state for persistence."""
+
+    version: int = Field(default=1, description="Schema version for forward compatibility")
+    session_id: str = Field(description="Unique session identifier")
+    created_at: str = Field(description="ISO timestamp of session creation")
+    updated_at: str = Field(description="ISO timestamp of last update")
+    model: str = Field(description="LLM model name used in session")
+    workspace_root: str = Field(description="Workspace root path")
+    messages: list[dict] = Field(default_factory=list, description="Serialized messages")
+    token_usage: dict = Field(default_factory=dict, description="Cumulative token usage")
+
+
+class SessionManager:
+    """Manages session file I/O: save, load, restore, and cleanup.
+
+    Session files are keyed by a hash of the workspace root path so that
+    each project directory has its own session history.
+    """
+
+    def __init__(self, config: SessionConfig, workspace_root: Path, model: str) -> None:
+        """Initialize session manager.
+
+        Args:
+            config: Session configuration.
+            workspace_root: Absolute path to workspace root.
+            model: LLM model name for session metadata.
+        """
+        self._config = config
+        self._workspace_root = workspace_root
+        self._model = model
+        self._workspace_hash = hashlib.sha256(str(workspace_root).encode()).hexdigest()[:12]
+        self._session_dir = workspace_root / config.session_dir
+        self._session_id = f"{self._workspace_hash}_{datetime.now(UTC).strftime('%Y%m%d_%H%M%S')}"
+
+    def save(self, ctx: "SessionContext") -> Path:
+        """Save session state to a JSON file via atomic write.
+
+        Args:
+            ctx: Session context to persist.
+
+        Returns:
+            Path to the saved session file.
+        """
+        self._session_dir.mkdir(parents=True, exist_ok=True)
+
+        serialized = ctx.to_serializable()
+        data = SessionData(
+            session_id=self._session_id,
+            created_at=ctx.start_time.isoformat(),
+            updated_at=datetime.now(UTC).isoformat(),
+            model=self._model,
+            workspace_root=str(self._workspace_root),
+            messages=serialized["messages"],
+            token_usage=serialized["token_usage"],
+        )
+
+        file_path = self._session_dir / f"{self._session_id}.json"
+        tmp_path = file_path.with_suffix(".tmp")
+
+        tmp_path.write_text(data.model_dump_json(indent=2), encoding="utf-8")
+        tmp_path.rename(file_path)
+
+        logger.debug("session_saved", path=str(file_path))
+        return file_path
+
+    def load_latest(self) -> SessionData | None:
+        """Find and load the newest session file for this workspace.
+
+        Returns:
+            SessionData if a valid session is found, None otherwise.
+        """
+        if not self._session_dir.exists():
+            return None
+
+        session_files = sorted(
+            self._session_dir.glob(f"{self._workspace_hash}_*.json"),
+            key=lambda p: p.stat().st_mtime,
+            reverse=True,
+        )
+
+        for path in session_files:
+            try:
+                raw = json.loads(path.read_text(encoding="utf-8"))
+                return SessionData(**raw)
+            except (json.JSONDecodeError, ValueError, OSError) as e:
+                logger.warning("session_load_error", path=str(path), error=str(e))
+                continue
+
+        return None
+
+    def restore(self, data: SessionData, ctx: "SessionContext") -> None:
+        """Replay session data into a SessionContext.
+
+        Args:
+            data: Saved session data to restore.
+            ctx: Session context to populate.
+        """
+        ctx.restore_from({
+            "messages": data.messages,
+            "token_usage": data.token_usage,
+        })
+        # Preserve the original session ID for continuity
+        self._session_id = data.session_id
+        logger.info("session_restored", session_id=data.session_id, messages=len(data.messages))
+
+    def cleanup_old(self) -> int:
+        """Delete session files older than max_session_age_hours.
+
+        Returns:
+            Number of files deleted.
+        """
+        if not self._session_dir.exists():
+            return 0
+
+        cutoff = datetime.now(UTC).timestamp() - (self._config.max_session_age_hours * 3600)
+        deleted = 0
+
+        for path in self._session_dir.glob("*.json"):
+            try:
+                if path.stat().st_mtime < cutoff:
+                    path.unlink()
+                    deleted += 1
+            except OSError:
+                continue
+
+        if deleted > 0:
+            logger.info("sessions_cleaned", deleted=deleted)
+        return deleted

app/services/streaming.py

@@ -1,41 +1,56 @@
 """Streaming response handler — accumulates SSE chunks into a complete Message."""
 
-from collections.abc import AsyncIterator
+import time
-
+from collections.abc import AsyncIterator, Callable
-from rich.live import Live
-from rich.markdown import Markdown
-from rich.panel import Panel
 
 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
+    Accumulates content deltas and tool call fragments. Notifies the UI via
-    panel during streaming, and produces a complete assistant Message on finish.
+    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:
+        thinking_notified = False
-            async for chunk in chunk_iter:
+        last_update_time = 0.0
-                self._process_chunk(chunk)
 
-                # Show reasoning while waiting for content
+        async for chunk in chunk_iter:
-                display_text = self._accumulated_content
+            self._process_chunk(chunk)
-                if not display_text and self._accumulated_reasoning:
-                    display_text = "*thinking...*"
 
-                if display_text and self._display_config.stream_output:
+            if not self._display_config.stream_output:
-                    # Render inside the same Assistant panel used for final output
+                continue
-                    # so the live display and final frame are visually consistent
+
-                    live.update(
+            # Notify thinking once
-                        Panel(
+            if (
-                            Markdown(display_text),
+                not thinking_notified
-                            title="Assistant",
+                and not self._accumulated_content
-                            border_style="green",
+                and self._accumulated_reasoning
-                            expand=True,
+                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.
+        """Extract content, tool calls, and usage from a single SSE chunk."""
-
-        Args:
-            chunk: Parsed JSON dict from one SSE data line.
-        """
-        # Content delta
         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.
+        """Convert accumulated tool call fragments into sorted ToolCall list."""
-
-        Returns:
-            List of ToolCall objects sorted by stream index.
-        """
         if not self._tool_calls:
             return []
 
@@ -142,6 +161,17 @@ class StreamHandler:
             )
         return result
 
+    def get_partial_message(self) -> Message | None:
+        """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
+        return Message(
+            role="assistant",
+            content=self._accumulated_content or None,
+            tool_calls=tool_calls,
+        )
+
     @property
     def usage(self) -> TokenUsage | None:
         """Token usage reported by the API, if available."""
@@ -158,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.
+    """Print a compact tool result."""
-
-    Args:
-        name: Tool name.
-        output: Tool output or error message.
-        is_error: Whether this is an error 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.
+    """Print conversation history as a Rich table."""
-
+    console.print(render_history(messages))
-    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)

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.
 

config/config.yaml

@@ -7,11 +7,16 @@ llm:
   temperature: 0.1
   max_tokens: 4096
   timeout: 120
+  max_retries: 3
+  retry_backoff_base: 1.0
+  retry_backoff_max: 30.0
 
 agent:
   max_iterations: 25
   max_conversation_tokens: 32000
   workspace_root: "."
+  truncation_keep_recent: 10
+  truncation_threshold: 0.85
 
 permissions:
   auto_approve:
@@ -56,6 +61,12 @@ tools:
     max_file_size_bytes: 1048576  # 1 MB
     binary_detection: true
 
+session:
+  session_dir: ".sneakycode/sessions"
+  auto_save: true
+  max_session_age_hours: 72
+  offer_resume: true
+
 display:
   show_tool_calls: true
   show_token_usage: true

docs/tools.md

@@ -0,0 +1,240 @@
+# Tool Reference
+
+SneakyCode provides 11 agent-callable tools organized into 5 categories. All file path arguments must be **relative to the workspace root**.
+
+## Permission Tiers
+
+| Tier          | Behavior                              | Tools                                                          |
+|---------------|---------------------------------------|----------------------------------------------------------------|
+| Auto-approved | Executed without user confirmation    | `read_file`, `list_dir`, `grep_files`, `find_files`, `finish` |
+| User confirm  | Prompts user before execution         | `write_file`, `make_dir`, `delete_file`, `str_replace`, `patch_apply`, `run_command` |
+| Denied        | Blocked entirely (configurable)       | Any tool added to `permissions.deny` in config                 |
+
+---
+
+## Read Tools
+
+### read_file
+
+Read the full contents of a text file.
+
+| Parameter   | Type | Required | Description                                    |
+|-------------|------|----------|------------------------------------------------|
+| `file_path` | str  | Yes      | Path to the file to read (relative to workspace) |
+
+**Permission:** Auto-approved
+
+**Example:**
+```json
+{"file_path": "app/main.py"}
+```
+
+**Notes:** Binary files are detected and rejected. Files exceeding `max_file_size_bytes` (default 1 MB) are rejected.
+
+---
+
+### list_dir
+
+List the contents of a directory. Directories are suffixed with `/`. Results are sorted with directories first, then files.
+
+| Parameter        | Type | Required | Default | Description                          |
+|------------------|------|----------|---------|--------------------------------------|
+| `directory_path` | str  | No       | `"."`   | Path to directory (relative)         |
+| `recursive`      | bool | No       | `false` | If true, list entries recursively    |
+
+**Permission:** Auto-approved
+
+**Example:**
+```json
+{"directory_path": "app/tools", "recursive": true}
+```
+
+---
+
+## Search Tools
+
+### grep_files
+
+Search for a regex pattern in file contents. Returns matching lines with file paths and line numbers.
+
+| Parameter      | Type       | Required | Default | Description                              |
+|----------------|------------|----------|---------|------------------------------------------|
+| `pattern`      | str        | Yes      |         | Regular expression pattern to search for |
+| `path`         | str        | No       | `"."`   | Directory or file to search in           |
+| `file_pattern` | str\|null  | No       | `null`  | Glob pattern to filter files (e.g. `*.py`) |
+
+**Permission:** Auto-approved
+
+**Example:**
+```json
+{"pattern": "def main", "path": "app/", "file_pattern": "*.py"}
+```
+
+---
+
+### find_files
+
+Search for files matching a name pattern. Returns relative file paths.
+
+| Parameter | Type | Required | Default | Description                                       |
+|-----------|------|----------|---------|---------------------------------------------------|
+| `pattern` | str  | Yes      |         | File name pattern (e.g. `*.py`, `config.yaml`)    |
+| `path`    | str  | No       | `"."`   | Directory to search in                             |
+
+**Permission:** Auto-approved
+
+**Example:**
+```json
+{"pattern": "*.yaml", "path": "config/"}
+```
+
+---
+
+## Write Tools
+
+### write_file
+
+Write text content to a file. Creates parent directories if needed. Overwrites existing file content.
+
+| Parameter   | Type | Required | Description                     |
+|-------------|------|----------|---------------------------------|
+| `file_path` | str  | Yes      | Path to the file to write       |
+| `content`   | str  | Yes      | Content to write to the file    |
+
+**Permission:** User confirmation required
+
+**Example:**
+```json
+{"file_path": "app/utils/helpers.py", "content": "def greet():\n    return 'hello'\n"}
+```
+
+---
+
+### make_dir
+
+Create a directory and any necessary parent directories.
+
+| Parameter        | Type | Required | Description                      |
+|------------------|------|----------|----------------------------------|
+| `directory_path` | str  | Yes      | Path to the directory to create  |
+
+**Permission:** User confirmation required
+
+**Example:**
+```json
+{"directory_path": "app/services/new_module"}
+```
+
+---
+
+### delete_file
+
+Delete a single file. Does not delete directories.
+
+| Parameter   | Type | Required | Description                     |
+|-------------|------|----------|---------------------------------|
+| `file_path` | str  | Yes      | Path to the file to delete      |
+
+**Permission:** User confirmation required
+
+**Example:**
+```json
+{"file_path": "app/utils/deprecated.py"}
+```
+
+---
+
+## Edit Tools
+
+### str_replace
+
+Replace exactly one occurrence of `old_str` with `new_str` in a file. Fails if `old_str` is not found or appears more than once.
+
+| Parameter   | Type | Required | Description                                        |
+|-------------|------|----------|----------------------------------------------------|
+| `file_path` | str  | Yes      | Path to the file to edit                           |
+| `old_str`   | str  | Yes      | The exact string to find and replace (must be unique) |
+| `new_str`   | str  | Yes      | The replacement string                             |
+
+**Permission:** User confirmation required
+
+**Example:**
+```json
+{
+  "file_path": "app/main.py",
+  "old_str": "def old_function():",
+  "new_str": "def new_function():"
+}
+```
+
+---
+
+### patch_apply
+
+Apply a unified diff (patch) to a file. The patch must be in standard unified diff format.
+
+| Parameter   | Type | Required | Description                                |
+|-------------|------|----------|--------------------------------------------|
+| `file_path` | str  | Yes      | Path to the file to patch                  |
+| `patch`     | str  | Yes      | Unified diff format patch to apply         |
+
+**Permission:** User confirmation required
+
+**Example:**
+```json
+{
+  "file_path": "app/main.py",
+  "patch": "--- a/app/main.py\n+++ b/app/main.py\n@@ -1,3 +1,3 @@\n-old line\n+new line\n"
+}
+```
+
+---
+
+## Shell Tools
+
+### run_command
+
+Run a shell command in the workspace directory. Only allowed commands may be executed; dangerous commands are blocked.
+
+| Parameter | Type     | Required | Default | Description                       |
+|-----------|----------|----------|---------|-----------------------------------|
+| `command` | str      | Yes      |         | Shell command to execute          |
+| `timeout` | int\|null | No      | `30`    | Timeout in seconds                |
+
+**Permission:** User confirmation required. Subject to `tools.shell.allowed_commands` and `tools.shell.denied_commands` in config.
+
+**Example:**
+```json
+{"command": "git status", "timeout": 10}
+```
+
+**Notes:** Output is truncated to `max_output_bytes` (default 64 KB). The command's first word is checked against allow/deny lists.
+
+---
+
+## Control Tools
+
+### finish
+
+Signal that the task is complete. Terminates the agent loop.
+
+| Parameter | Type | Required | Default            | Description                  |
+|-----------|------|----------|--------------------|------------------------------|
+| `message` | str  | No       | `"Task complete."` | Final message to the user    |
+
+**Permission:** Auto-approved
+
+**Example:**
+```json
+{"message": "Created the new module with tests."}
+```
+
+---
+
+## Security Notes
+
+- All file paths are resolved against `workspace_root` with path traversal protection
+- Binary file detection prevents reading/writing binary files
+- File size limits prevent reading/writing excessively large files
+- Shell commands are validated against configurable allow/deny lists
+- Tool call arguments from the LLM are validated against JSON schema before execution

pyproject.toml

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

tests/integration/conftest.py

@@ -0,0 +1,150 @@
+"""Shared fixtures for integration tests."""
+
+import json
+from collections.abc import AsyncIterator
+from pathlib import Path
+from typing import Any
+from unittest.mock import AsyncMock
+
+import pytest
+
+from app.models.config import AgentConfig, AppConfig, DisplayConfig, LLMConfig, PermissionsConfig, SessionConfig
+from app.models.message import Message
+from app.services.llm import LLMClient
+
+
+@pytest.fixture
+def tmp_workspace(tmp_path: Path) -> Path:
+    """Create a temporary workspace directory with a sample file."""
+    ws = tmp_path / "workspace"
+    ws.mkdir()
+    (ws / "hello.txt").write_text("Hello, world!")
+    return ws
+
+
+@pytest.fixture
+def test_config(tmp_workspace: Path) -> AppConfig:
+    """AppConfig suitable for integration tests."""
+    return AppConfig(
+        llm=LLMConfig(
+            model="test-model",
+            endpoint="http://localhost:11434",
+            max_retries=2,
+            retry_backoff_base=0.01,
+            retry_backoff_max=0.02,
+        ),
+        agent=AgentConfig(
+            max_iterations=10,
+            max_conversation_tokens=32000,
+            workspace_root=tmp_workspace,
+            truncation_keep_recent=4,
+            truncation_threshold=0.85,
+        ),
+        permissions=PermissionsConfig(
+            auto_approve=["read_file", "list_dir", "grep_files", "find_files", "finish"],
+        ),
+        display=DisplayConfig(
+            show_tool_calls=False,
+            show_token_usage=False,
+            stream_output=False,
+        ),
+        session=SessionConfig(
+            session_dir=tmp_workspace / ".sneakycode" / "sessions",
+        ),
+    )
+
+
+def make_text_chunks(content: str) -> list[dict[str, Any]]:
+    """Create SSE chunk dicts for a plain text response."""
+    chunks = []
+    for char in content:
+        chunks.append({
+            "choices": [{"delta": {"content": char}, "index": 0}]
+        })
+    chunks.append({
+        "choices": [{"delta": {}, "finish_reason": "stop", "index": 0}],
+        "usage": {"prompt_tokens": 10, "completion_tokens": len(content), "total_tokens": 10 + len(content)},
+    })
+    return chunks
+
+
+def make_tool_call_chunks(name: str, args: dict[str, Any], tc_id: str = "call_001") -> list[dict[str, Any]]:
+    """Create SSE chunk dicts for a tool call response."""
+    args_str = json.dumps(args)
+    chunks = [
+        {
+            "choices": [{
+                "delta": {
+                    "tool_calls": [{
+                        "index": 0,
+                        "id": tc_id,
+                        "function": {"name": name, "arguments": ""},
+                    }]
+                },
+                "index": 0,
+            }]
+        },
+        {
+            "choices": [{
+                "delta": {
+                    "tool_calls": [{
+                        "index": 0,
+                        "function": {"arguments": args_str},
+                    }]
+                },
+                "index": 0,
+            }]
+        },
+        {
+            "choices": [{"delta": {}, "finish_reason": "tool_calls", "index": 0}],
+            "usage": {"prompt_tokens": 10, "completion_tokens": 20, "total_tokens": 30},
+        },
+    ]
+    return chunks
+
+
+class MockLLMClient:
+    """LLM client that returns scripted SSE chunk sequences."""
+
+    def __init__(self, responses: list[list[dict[str, Any]]]) -> None:
+        self._responses = list(responses)
+        self._call_count = 0
+
+    async def stream_chat(
+        self,
+        messages: list[Message],
+        tools: list[dict[str, Any]] | None = None,
+    ) -> AsyncIterator[dict]:
+        if self._call_count >= len(self._responses):
+            raise RuntimeError("MockLLMClient ran out of scripted responses")
+        chunks = self._responses[self._call_count]
+        self._call_count += 1
+        for chunk in chunks:
+            yield chunk
+
+    async def stream_chat_with_retry(
+        self,
+        messages: list[Message],
+        tools: list[dict[str, Any]] | None = None,
+    ) -> AsyncIterator[dict]:
+        async for chunk in self.stream_chat(messages, tools=tools):
+            yield chunk
+
+    @property
+    def call_count(self) -> int:
+        return self._call_count
+
+    async def close(self) -> None:
+        pass
+
+    async def __aenter__(self):
+        return self
+
+    async def __aexit__(self, *exc):
+        pass
+
+
+@pytest.fixture
+def mock_llm_client():
+    """Factory fixture for creating MockLLMClient instances."""
+    return MockLLMClient

tests/integration/test_agent_workflows.py

@@ -0,0 +1,155 @@
+"""Integration tests for end-to-end agent workflows with mocked LLM."""
+
+from pathlib import Path
+from unittest.mock import AsyncMock, patch
+
+import pytest
+
+from app.agent.context import SessionContext
+from app.agent.loop import AgentLoop
+from app.models.config import AgentConfig, AppConfig, LLMConfig
+from app.services.llm import LLMConnectionError
+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 .conftest import MockLLMClient, make_text_chunks, make_tool_call_chunks
+
+
+@pytest.fixture
+def agent_factory(test_config: AppConfig, tmp_workspace: Path):
+    """Factory that creates an AgentLoop wired to a MockLLMClient."""
+
+    def create(responses):
+        ctx = SessionContext(test_config)
+        mock_client = MockLLMClient(responses)
+        handler = StreamHandler(test_config.display)
+        registry = create_default_registry(test_config.agent.workspace_root, test_config)
+        permissions = PermissionsService(test_config.permissions)
+        agent = AgentLoop(test_config, ctx, mock_client, handler, registry, permissions)
+        return agent, ctx, mock_client
+
+    return create
+
+
+class TestAgentWorkflows:
+    @pytest.mark.asyncio
+    async def test_multi_turn_read_workflow(self, agent_factory, tmp_workspace: Path) -> None:
+        """Agent reads a file then responds with text — full 2-turn workflow."""
+        responses = [
+            make_tool_call_chunks("read_file", {"file_path": "hello.txt"}),
+            make_text_chunks("The file contains: Hello, world!"),
+        ]
+        agent, ctx, mock_client = agent_factory(responses)
+
+        await agent.run_turn("What's in hello.txt?")
+
+        assert mock_client.call_count == 2
+        history = ctx.get_history()
+        # user, assistant (tool_call), tool (result), assistant (text)
+        assert len(history) == 4
+        assert history[0].role == "user"
+        assert history[1].role == "assistant"
+        assert history[1].tool_calls is not None
+        assert history[2].role == "tool"
+        assert "Hello, world!" in (history[2].content or "")
+        assert history[3].role == "assistant"
+        assert "Hello, world!" in (history[3].content or "")
+
+    @pytest.mark.asyncio
+    async def test_token_budget_truncation(self, test_config: AppConfig, tmp_workspace: Path) -> None:
+        """When token budget is exceeded, truncation drops messages instead of stopping."""
+        # Use a tiny budget to trigger truncation
+        test_config.agent.max_conversation_tokens = 100
+        test_config.agent.truncation_keep_recent = 2
+        test_config.agent.truncation_threshold = 0.5
+
+        ctx = SessionContext(test_config)
+
+        # Fill history with enough to exceed budget
+        for i in range(10):
+            ctx.add_message("user", f"Message {i} " * 20)
+            ctx.add_message("assistant", f"Response {i} " * 20)
+
+        # Force token counter over budget
+        from app.utils.token_counter import TokenUsage
+        ctx.token_counter.count_usage(TokenUsage(total_tokens=100))
+
+        original_count = len(ctx.get_history())
+        dropped = ctx.truncate_history()
+
+        assert dropped > 0
+        assert len(ctx.get_history()) < original_count
+        # Recent messages should still be present
+        assert len(ctx.get_history()) >= 2
+
+    @pytest.mark.asyncio
+    async def test_session_save_and_restore(self, test_config: AppConfig, tmp_workspace: Path) -> None:
+        """Session can be saved after a turn and restored into a fresh context."""
+        responses = [make_text_chunks("Hello from the agent!")]
+        ctx = SessionContext(test_config)
+        mock_client = MockLLMClient(responses)
+        handler = StreamHandler(test_config.display)
+        registry = create_default_registry(test_config.agent.workspace_root, test_config)
+        permissions = PermissionsService(test_config.permissions)
+        agent = AgentLoop(test_config, ctx, mock_client, handler, registry, permissions)
+
+        await agent.run_turn("Hi")
+
+        # Save session
+        session_mgr = SessionManager(
+            test_config.session, test_config.agent.workspace_root, test_config.llm.model
+        )
+        path = session_mgr.save(ctx)
+        assert path.exists()
+
+        # Restore into fresh context
+        fresh_ctx = SessionContext(test_config)
+        loaded = session_mgr.load_latest()
+        assert loaded is not None
+        session_mgr.restore(loaded, fresh_ctx)
+
+        assert fresh_ctx.message_count == ctx.message_count
+        original_history = ctx.get_history()
+        restored_history = fresh_ctx.get_history()
+        for orig, restored in zip(original_history, restored_history):
+            assert orig.role == restored.role
+            assert orig.content == restored.content
+
+    @pytest.mark.asyncio
+    async def test_retry_on_transient_error(self, test_config: AppConfig, tmp_workspace: Path) -> None:
+        """Agent recovers from transient LLM errors via retry."""
+        from app.services.llm import LLMClient
+
+        ctx = SessionContext(test_config)
+
+        # Create a real client but mock stream_chat to fail then succeed
+        client = LLMClient(test_config.llm)
+        call_count = 0
+
+        async def flaky_stream(*args, **kwargs):
+            nonlocal call_count
+            call_count += 1
+            if call_count == 1:
+                raise LLMConnectionError("Temporary failure")
+            for chunk in make_text_chunks("Recovered!"):
+                yield chunk
+
+        client.stream_chat = flaky_stream  # type: ignore[assignment]
+
+        handler = StreamHandler(test_config.display)
+        registry = create_default_registry(test_config.agent.workspace_root, test_config)
+        permissions = PermissionsService(test_config.permissions)
+        agent = AgentLoop(test_config, ctx, client, handler, registry, permissions)
+
+        with patch("app.services.llm.asyncio.sleep", new_callable=AsyncMock):
+            await agent.run_turn("Test retry")
+
+        history = ctx.get_history()
+        # Should have succeeded: user + assistant
+        assert len(history) == 2
+        assert history[1].content == "Recovered!"
+        assert call_count == 2
+
+        await client.close()

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
@@ -62,6 +63,7 @@ def handler() -> MagicMock:
     mock.usage = None
     mock.had_reasoning_only = False
     mock.reset = MagicMock()
+    mock.get_partial_message = MagicMock(return_value=None)
     return mock
 
 
@@ -75,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,
@@ -83,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_retry.py

@@ -0,0 +1,125 @@
+"""Unit tests for LLM retry with exponential backoff."""
+
+from unittest.mock import AsyncMock, patch
+
+import pytest
+
+from app.models.config import LLMConfig
+from app.models.message import Message
+from app.services.llm import LLMClient, LLMConnectionError, LLMResponseError
+
+
+@pytest.fixture
+def llm_config() -> LLMConfig:
+    return LLMConfig(
+        model="test-model",
+        endpoint="http://localhost:11434",
+        max_retries=3,
+        retry_backoff_base=0.01,
+        retry_backoff_max=0.05,
+    )
+
+
+@pytest.fixture
+def client(llm_config: LLMConfig) -> LLMClient:
+    return LLMClient(llm_config)
+
+
+@pytest.fixture
+def messages() -> list[Message]:
+    return [Message(role="user", content="Hello")]
+
+
+class TestRetry:
+    @pytest.mark.asyncio
+    async def test_succeeds_without_retry(self, client: LLMClient, messages: list[Message]) -> None:
+        """Successful stream doesn't retry."""
+        call_count = 0
+
+        async def fake_stream(*args, **kwargs):
+            nonlocal call_count
+            call_count += 1
+            yield {"choices": [{"delta": {"content": "Hi"}}]}
+
+        client.stream_chat = fake_stream  # type: ignore[assignment]
+
+        collected = []
+        async for chunk in client.stream_chat_with_retry(messages):
+            collected.append(chunk)
+
+        assert len(collected) == 1
+        assert call_count == 1
+
+    @pytest.mark.asyncio
+    async def test_retries_on_connection_error(self, client: LLMClient, messages: list[Message]) -> None:
+        """Retries on LLMConnectionError, then succeeds."""
+        call_count = 0
+
+        async def flaky_stream(*args, **kwargs):
+            nonlocal call_count
+            call_count += 1
+            if call_count < 3:
+                raise LLMConnectionError("Connection refused")
+            yield {"choices": [{"delta": {"content": "OK"}}]}
+
+        client.stream_chat = flaky_stream  # type: ignore[assignment]
+
+        with patch("app.services.llm.asyncio.sleep", new_callable=AsyncMock):
+            collected = []
+            async for chunk in client.stream_chat_with_retry(messages):
+                collected.append(chunk)
+
+        assert len(collected) == 1
+        assert call_count == 3
+
+    @pytest.mark.asyncio
+    async def test_retries_on_5xx(self, client: LLMClient, messages: list[Message]) -> None:
+        """Retries on 5xx LLMResponseError."""
+        call_count = 0
+
+        async def server_error_stream(*args, **kwargs):
+            nonlocal call_count
+            call_count += 1
+            if call_count < 2:
+                raise LLMResponseError("Internal Server Error", status_code=500)
+            yield {"choices": [{"delta": {"content": "OK"}}]}
+
+        client.stream_chat = server_error_stream  # type: ignore[assignment]
+
+        with patch("app.services.llm.asyncio.sleep", new_callable=AsyncMock):
+            collected = []
+            async for chunk in client.stream_chat_with_retry(messages):
+                collected.append(chunk)
+
+        assert len(collected) == 1
+        assert call_count == 2
+
+    @pytest.mark.asyncio
+    async def test_no_retry_on_4xx(self, client: LLMClient, messages: list[Message]) -> None:
+        """Does NOT retry on 4xx errors — raises immediately."""
+        async def bad_request_stream(*args, **kwargs):
+            raise LLMResponseError("Bad Request", status_code=400)
+            yield  # pragma: no cover — make this an async generator
+
+        client.stream_chat = bad_request_stream  # type: ignore[assignment]
+
+        with pytest.raises(LLMResponseError, match="Bad Request"):
+            async for _ in client.stream_chat_with_retry(messages):
+                pass  # pragma: no cover
+
+    @pytest.mark.asyncio
+    async def test_respects_max_retries(self, client: LLMClient, messages: list[Message]) -> None:
+        """After exhausting retries, re-raises the last exception."""
+        async def always_fail(*args, **kwargs):
+            raise LLMConnectionError("Down forever")
+            yield  # pragma: no cover
+
+        client.stream_chat = always_fail  # type: ignore[assignment]
+
+        with patch("app.services.llm.asyncio.sleep", new_callable=AsyncMock) as mock_sleep:
+            with pytest.raises(LLMConnectionError, match="Down forever"):
+                async for _ in client.stream_chat_with_retry(messages):
+                    pass  # pragma: no cover
+
+        # Should have slept max_retries times (3 retries after initial attempt)
+        assert mock_sleep.call_count == 3

tests/unit/test_session.py

@@ -0,0 +1,122 @@
+"""Unit tests for session persistence."""
+
+import json
+import time
+from pathlib import Path
+
+import pytest
+
+from app.agent.context import SessionContext
+from app.models.config import AgentConfig, AppConfig, LLMConfig, SessionConfig
+from app.services.session import SessionManager
+
+
+@pytest.fixture
+def tmp_workspace(tmp_path: Path) -> Path:
+    return tmp_path / "workspace"
+
+
+@pytest.fixture
+def session_config(tmp_workspace: Path) -> SessionConfig:
+    return SessionConfig(
+        session_dir=tmp_workspace / ".sneakycode" / "sessions",
+        auto_save=True,
+        max_session_age_hours=72,
+        offer_resume=True,
+    )
+
+
+@pytest.fixture
+def config(tmp_workspace: Path) -> AppConfig:
+    return AppConfig(
+        llm=LLMConfig(model="test-model", endpoint="http://localhost:11434"),
+        agent=AgentConfig(workspace_root=tmp_workspace),
+    )
+
+
+@pytest.fixture
+def ctx(config: AppConfig) -> SessionContext:
+    return SessionContext(config)
+
+
+@pytest.fixture
+def session_mgr(session_config: SessionConfig, tmp_workspace: Path) -> SessionManager:
+    tmp_workspace.mkdir(parents=True, exist_ok=True)
+    return SessionManager(session_config, tmp_workspace, "test-model")
+
+
+class TestSessionPersistence:
+    def test_save_creates_file(self, session_mgr: SessionManager, ctx: SessionContext, session_config: SessionConfig) -> None:
+        """Saving a session creates a JSON file in the session directory."""
+        ctx.add_message("user", "Hello")
+        ctx.add_message("assistant", "Hi there!")
+
+        path = session_mgr.save(ctx)
+
+        assert path.exists()
+        assert path.suffix == ".json"
+
+        data = json.loads(path.read_text())
+        assert data["model"] == "test-model"
+        assert len(data["messages"]) == 2
+
+    def test_load_latest_returns_newest(self, session_mgr: SessionManager, ctx: SessionContext, session_config: SessionConfig) -> None:
+        """load_latest returns the most recently modified session."""
+        ctx.add_message("user", "First session")
+        session_mgr.save(ctx)
+
+        # Create a second session manager (simulates a new startup)
+        mgr2 = SessionManager(session_config, session_mgr._workspace_root, "test-model")
+        ctx.add_message("assistant", "Second response")
+        path2 = mgr2.save(ctx)
+
+        loaded = mgr2.load_latest()
+        assert loaded is not None
+        assert loaded.session_id == mgr2._session_id
+        assert len(loaded.messages) == 2
+
+    def test_restore_populates_context(self, session_mgr: SessionManager, ctx: SessionContext, config: AppConfig) -> None:
+        """Restoring a session populates the context with saved messages."""
+        ctx.add_message("user", "Hello")
+        ctx.add_message("assistant", "World")
+        session_mgr.save(ctx)
+
+        # Load and restore into a fresh context
+        fresh_ctx = SessionContext(config)
+        loaded = session_mgr.load_latest()
+        assert loaded is not None
+
+        session_mgr.restore(loaded, fresh_ctx)
+
+        history = fresh_ctx.get_history()
+        assert len(history) == 2
+        assert history[0].role == "user"
+        assert history[0].content == "Hello"
+        assert history[1].role == "assistant"
+        assert history[1].content == "World"
+
+    def test_cleanup_removes_old_files(self, session_config: SessionConfig, tmp_workspace: Path) -> None:
+        """cleanup_old deletes files older than max_session_age_hours."""
+        tmp_workspace.mkdir(parents=True, exist_ok=True)
+
+        # Create session config with very short max age
+        short_config = SessionConfig(
+            session_dir=session_config.session_dir,
+            max_session_age_hours=0,  # 0 hours = everything is old
+        )
+        mgr = SessionManager(short_config, tmp_workspace, "test-model")
+
+        # Create a session file manually with old timestamp
+        session_dir = tmp_workspace / short_config.session_dir
+        session_dir.mkdir(parents=True, exist_ok=True)
+        old_file = session_dir / "old_session.json"
+        old_file.write_text('{"version": 1}')
+
+        # Set mtime to the past
+        import os
+        old_time = time.time() - 3600  # 1 hour ago
+        os.utime(old_file, (old_time, old_time))
+
+        deleted = mgr.cleanup_old()
+        assert deleted == 1
+        assert not old_file.exists()

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)
+    @pytest.mark.asyncio
-    def test_prompt_user_approved(self, mock_ask: object) -> None:
+    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)
+        async def approve(tool_name: str, description: str) -> bool:
-    def test_prompt_user_denied(self, mock_ask: object) -> None:
+            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)
+        async def deny(tool_name: str, description: str) -> bool:
-    def test_unlisted_tool_prompts(self, mock_ask: object) -> None:
+            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 ---

tests/unit/test_truncation.py

@@ -0,0 +1,128 @@
+"""Unit tests for conversation truncation logic."""
+
+from pathlib import Path
+
+import pytest
+
+from app.agent.context import SessionContext
+from app.models.config import AgentConfig, AppConfig, LLMConfig
+
+
+@pytest.fixture
+def config() -> AppConfig:
+    return AppConfig(
+        llm=LLMConfig(model="test-model", endpoint="http://localhost:11434"),
+        agent=AgentConfig(
+            max_conversation_tokens=200,
+            truncation_keep_recent=3,
+            truncation_threshold=0.85,
+            workspace_root=Path("/tmp/test"),
+        ),
+    )
+
+
+@pytest.fixture
+def ctx(config: AppConfig) -> SessionContext:
+    return SessionContext(config)
+
+
+class TestTruncation:
+    def test_no_truncation_under_threshold(self, ctx: SessionContext) -> None:
+        """No messages dropped when under threshold."""
+        ctx.add_message("user", "Hello")
+        ctx.add_message("assistant", "Hi there!")
+
+        dropped = ctx.truncate_history()
+        assert dropped == 0
+        assert ctx.message_count == 2
+
+    def test_drops_oldest_messages(self, ctx: SessionContext) -> None:
+        """Drops middle messages when over budget."""
+        # Fill with enough content to exceed the small 200-token budget
+        ctx.add_message("user", "First message " * 20)
+        for i in range(8):
+            ctx.add_message("assistant", f"Response {i} " * 15)
+            ctx.add_message("user", f"Follow-up {i} " * 15)
+
+        # Force the token counter to report over budget
+        from app.utils.token_counter import TokenUsage
+        ctx.token_counter.count_usage(TokenUsage(total_tokens=200))
+
+        original_count = len(ctx.get_history())
+        dropped = ctx.truncate_history()
+
+        assert dropped > 0
+        assert len(ctx.get_history()) < original_count
+
+    def test_preserves_recent_messages(self, ctx: SessionContext) -> None:
+        """The most recent N messages are always preserved."""
+        ctx.add_message("user", "First message " * 20)
+        for i in range(10):
+            ctx.add_message("assistant", f"Response {i} " * 10)
+            ctx.add_message("user", f"Follow-up {i} " * 10)
+
+        from app.utils.token_counter import TokenUsage
+        ctx.token_counter.count_usage(TokenUsage(total_tokens=200))
+
+        history_before = ctx.get_history()
+        recent_before = history_before[-3:]  # keep_recent=3
+
+        ctx.truncate_history()
+
+        history_after = ctx.get_history()
+        recent_after = history_after[-3:]
+
+        # Recent messages should be preserved
+        for before, after in zip(recent_before, recent_after):
+            assert before.content == after.content
+
+    def test_preserves_first_user_message(self, ctx: SessionContext) -> None:
+        """First user message is always kept."""
+        first_content = "This is the very first user message"
+        ctx.add_message("user", first_content)
+        for i in range(10):
+            ctx.add_message("assistant", f"Response {i} " * 10)
+            ctx.add_message("user", f"Follow-up {i} " * 10)
+
+        from app.utils.token_counter import TokenUsage
+        ctx.token_counter.count_usage(TokenUsage(total_tokens=200))
+
+        ctx.truncate_history()
+
+        history = ctx.get_history()
+        assert history[0].role == "user"
+        assert history[0].content == first_content
+
+    def test_orphaned_tool_messages_cleaned(self, ctx: SessionContext) -> None:
+        """Tool messages without matching tool_call are cleaned up."""
+        from app.models.tool_call import ToolCall, ToolCallFunction
+
+        ctx.add_message("user", "Do something " * 20)
+        # Assistant with tool call
+        ctx.add_message(
+            "assistant",
+            None,
+            tool_calls=[ToolCall(id="tc_1", type="function", function=ToolCallFunction(name="read_file", arguments='{"path": "x"}'))],
+        )
+        # Tool result for tc_1
+        ctx.add_message("tool", "file contents " * 20, tool_call_id="tc_1", name="read_file")
+        # More padding to push over budget
+        for i in range(8):
+            ctx.add_message("assistant", f"Analysis {i} " * 15)
+            ctx.add_message("user", f"Next {i} " * 15)
+
+        from app.utils.token_counter import TokenUsage
+        ctx.token_counter.count_usage(TokenUsage(total_tokens=200))
+
+        ctx.truncate_history()
+
+        history = ctx.get_history()
+        # If the assistant message with tc_1 was dropped, the orphaned tool message should also be gone
+        has_tc1_assistant = any(
+            m.role == "assistant" and m.tool_calls and any(tc.id == "tc_1" for tc in m.tool_calls)
+            for m in history
+        )
+        has_tc1_tool = any(m.role == "tool" and m.tool_call_id == "tc_1" for m in history)
+
+        # Either both exist or neither exists
+        assert has_tc1_assistant == has_tc1_tool

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" },
+]