feat: Add conversational AI assistant panel for project codebase Q&A

Implement a slide-in chat panel that allows users to ask questions about their codebase using Claude Opus 4.5 with read-only access to project files. Backend changes: - Add SQLAlchemy models for conversation persistence (assistant_database.py) - Create AssistantChatSession with read-only Claude SDK client - Add WebSocket endpoint for real-time chat streaming - Include read-only MCP tools: feature_get_stats, feature_get_next, etc. Frontend changes: - Add floating action button (bottom-right) to toggle panel - Create slide-in panel component (400px width) - Implement WebSocket hook with reconnection logic - Add keyboard shortcut 'A' to toggle assistant Key features: - Read-only access: Only Read, Glob, Grep, WebFetch, WebSearch tools - Persistent history: Conversations saved to SQLite per project - Real-time streaming: Text chunks streamed as Claude generates response - Tool visibility: Shows when assistant is using tools to explore code 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
2026-01-30 06:12:06 +00:00 · 2026-01-04 14:57:58 +02:00
parent 88951e454a
commit 908754302a
13 changed files with 1657 additions and 6 deletions
--- a/server/services/assistant_chat_session.py
+++ b/server/services/assistant_chat_session.py
@@ -0,0 +1,374 @@
+"""
+Assistant Chat Session
+======================
+
+Manages read-only conversational assistant sessions for projects.
+The assistant can answer questions about the codebase and features
+but cannot modify any files.
+"""
+
+import json
+import logging
+import os
+import shutil
+import sys
+import threading
+from datetime import datetime
+from pathlib import Path
+from typing import AsyncGenerator, Optional
+
+from claude_agent_sdk import ClaudeAgentOptions, ClaudeSDKClient
+
+from .assistant_database import (
+    create_conversation,
+    add_message,
+    get_conversation,
+)
+
+logger = logging.getLogger(__name__)
+
+# Root directory of the project
+ROOT_DIR = Path(__file__).parent.parent.parent
+
+# Read-only feature MCP tools (no mark_passing, skip, create_bulk)
+READONLY_FEATURE_MCP_TOOLS = [
+    "mcp__features__feature_get_stats",
+    "mcp__features__feature_get_next",
+    "mcp__features__feature_get_for_regression",
+]
+
+# Read-only built-in tools (no Write, Edit, Bash)
+READONLY_BUILTIN_TOOLS = [
+    "Read",
+    "Glob",
+    "Grep",
+    "WebFetch",
+    "WebSearch",
+]
+
+
+def get_system_prompt(project_name: str, project_dir: Path) -> str:
+    """Generate the system prompt for the assistant with project context."""
+    # Try to load app_spec.txt for context
+    app_spec_content = ""
+    app_spec_path = project_dir / "prompts" / "app_spec.txt"
+    if app_spec_path.exists():
+        try:
+            app_spec_content = app_spec_path.read_text(encoding="utf-8")
+            # Truncate if too long
+            if len(app_spec_content) > 5000:
+                app_spec_content = app_spec_content[:5000] + "\n... (truncated)"
+        except Exception as e:
+            logger.warning(f"Failed to read app_spec.txt: {e}")
+
+    return f"""You are a helpful project assistant for the "{project_name}" project.
+
+Your role is to help users understand the codebase, answer questions about features, and explain how code works. You have READ-ONLY access to the project files.
+
+IMPORTANT: You CANNOT modify any files. You can only:
+- Read and analyze source code files
+- Search for patterns in the codebase
+- Look up documentation online
+- Check feature progress and status
+
+If the user asks you to make changes, politely explain that you're a read-only assistant and they should use the main coding agent for modifications.
+
+## Project Specification
+
+{app_spec_content if app_spec_content else "(No app specification found)"}
+
+## Available Tools
+
+You have access to these read-only tools:
+- **Read**: Read file contents
+- **Glob**: Find files by pattern (e.g., "**/*.tsx")
+- **Grep**: Search file contents with regex
+- **WebFetch/WebSearch**: Look up documentation online
+- **feature_get_stats**: Get feature completion progress
+- **feature_get_next**: See the next pending feature
+- **feature_get_for_regression**: See passing features
+
+## Guidelines
+
+1. Be concise and helpful
+2. When explaining code, reference specific file paths and line numbers
+3. Use the feature tools to answer questions about project progress
+4. Search the codebase to find relevant information before answering
+5. If you're unsure, say so rather than guessing"""
+
+
+class AssistantChatSession:
+    """
+    Manages a read-only assistant conversation for a project.
+
+    Uses Claude Opus 4.5 with only read-only tools enabled.
+    Persists conversation history to SQLite.
+    """
+
+    def __init__(self, project_name: str, project_dir: Path, conversation_id: Optional[int] = None):
+        """
+        Initialize the session.
+
+        Args:
+            project_name: Name of the project
+            project_dir: Absolute path to the project directory
+            conversation_id: Optional existing conversation ID to resume
+        """
+        self.project_name = project_name
+        self.project_dir = project_dir
+        self.conversation_id = conversation_id
+        self.client: Optional[ClaudeSDKClient] = None
+        self._client_entered: bool = False
+        self.created_at = datetime.now()
+
+    async def close(self) -> None:
+        """Clean up resources and close the Claude client."""
+        if self.client and self._client_entered:
+            try:
+                await self.client.__aexit__(None, None, None)
+            except Exception as e:
+                logger.warning(f"Error closing Claude client: {e}")
+            finally:
+                self._client_entered = False
+                self.client = None
+
+    async def start(self) -> AsyncGenerator[dict, None]:
+        """
+        Initialize session with the Claude client.
+
+        Creates a new conversation if none exists, then sends an initial greeting.
+        Yields message chunks as they stream in.
+        """
+        # Create a new conversation if we don't have one
+        if self.conversation_id is None:
+            conv = create_conversation(self.project_dir, self.project_name)
+            self.conversation_id = conv.id
+            yield {"type": "conversation_created", "conversation_id": self.conversation_id}
+
+        # Build permissions list for read-only access
+        permissions_list = [
+            "Read(./**)",
+            "Glob(./**)",
+            "Grep(./**)",
+            "WebFetch",
+            "WebSearch",
+            *READONLY_FEATURE_MCP_TOOLS,
+        ]
+
+        # Create security settings file
+        security_settings = {
+            "sandbox": {"enabled": False},  # No bash, so sandbox not needed
+            "permissions": {
+                "defaultMode": "bypassPermissions",  # Read-only, no dangerous ops
+                "allow": permissions_list,
+            },
+        }
+        settings_file = self.project_dir / ".claude_assistant_settings.json"
+        with open(settings_file, "w") as f:
+            json.dump(security_settings, f, indent=2)
+
+        # Build MCP servers config - only features MCP for read-only access
+        mcp_servers = {
+            "features": {
+                "command": sys.executable,
+                "args": ["-m", "mcp_server.feature_mcp"],
+                "env": {
+                    **os.environ,
+                    "PROJECT_DIR": str(self.project_dir.resolve()),
+                    "PYTHONPATH": str(ROOT_DIR.resolve()),
+                },
+            },
+        }
+
+        # Get system prompt with project context
+        system_prompt = get_system_prompt(self.project_name, self.project_dir)
+
+        # Use system Claude CLI
+        system_cli = shutil.which("claude")
+
+        try:
+            self.client = ClaudeSDKClient(
+                options=ClaudeAgentOptions(
+                    model="claude-opus-4-5-20251101",
+                    cli_path=system_cli,
+                    system_prompt=system_prompt,
+                    allowed_tools=[*READONLY_BUILTIN_TOOLS, *READONLY_FEATURE_MCP_TOOLS],
+                    mcp_servers=mcp_servers,
+                    permission_mode="bypassPermissions",
+                    max_turns=100,
+                    cwd=str(self.project_dir.resolve()),
+                    settings=str(settings_file.resolve()),
+                )
+            )
+            await self.client.__aenter__()
+            self._client_entered = True
+        except Exception as e:
+            logger.exception("Failed to create Claude client")
+            yield {"type": "error", "content": f"Failed to initialize assistant: {str(e)}"}
+            return
+
+        # Send initial greeting
+        try:
+            greeting = f"Hello! I'm your project assistant for **{self.project_name}**. I can help you understand the codebase, explain features, and answer questions about the project. What would you like to know?"
+
+            # Store the greeting in the database
+            add_message(self.project_dir, self.conversation_id, "assistant", greeting)
+
+            yield {"type": "text", "content": greeting}
+            yield {"type": "response_done"}
+        except Exception as e:
+            logger.exception("Failed to send greeting")
+            yield {"type": "error", "content": f"Failed to start conversation: {str(e)}"}
+
+    async def send_message(self, user_message: str) -> AsyncGenerator[dict, None]:
+        """
+        Send user message and stream Claude's response.
+
+        Args:
+            user_message: The user's message
+
+        Yields:
+            Message chunks:
+            - {"type": "text", "content": str}
+            - {"type": "tool_call", "tool": str, "input": dict}
+            - {"type": "response_done"}
+            - {"type": "error", "content": str}
+        """
+        if not self.client:
+            yield {"type": "error", "content": "Session not initialized. Call start() first."}
+            return
+
+        if self.conversation_id is None:
+            yield {"type": "error", "content": "No conversation ID set."}
+            return
+
+        # Store user message in database
+        add_message(self.project_dir, self.conversation_id, "user", user_message)
+
+        try:
+            async for chunk in self._query_claude(user_message):
+                yield chunk
+            yield {"type": "response_done"}
+        except Exception as e:
+            logger.exception("Error during Claude query")
+            yield {"type": "error", "content": f"Error: {str(e)}"}
+
+    async def _query_claude(self, message: str) -> AsyncGenerator[dict, None]:
+        """
+        Internal method to query Claude and stream responses.
+
+        Handles tool calls and text responses.
+        """
+        if not self.client:
+            return
+
+        # Send message to Claude
+        await self.client.query(message)
+
+        full_response = ""
+
+        # Stream the response
+        async for msg in self.client.receive_response():
+            msg_type = type(msg).__name__
+
+            if msg_type == "AssistantMessage" and hasattr(msg, "content"):
+                for block in msg.content:
+                    block_type = type(block).__name__
+
+                    if block_type == "TextBlock" and hasattr(block, "text"):
+                        text = block.text
+                        if text:
+                            full_response += text
+                            yield {"type": "text", "content": text}
+
+                    elif block_type == "ToolUseBlock" and hasattr(block, "name"):
+                        tool_name = block.name
+                        tool_input = getattr(block, "input", {})
+                        yield {
+                            "type": "tool_call",
+                            "tool": tool_name,
+                            "input": tool_input,
+                        }
+
+        # Store the complete response in the database
+        if full_response and self.conversation_id:
+            add_message(self.project_dir, self.conversation_id, "assistant", full_response)
+
+    def get_conversation_id(self) -> Optional[int]:
+        """Get the current conversation ID."""
+        return self.conversation_id
+
+
+# Session registry with thread safety
+_sessions: dict[str, AssistantChatSession] = {}
+_sessions_lock = threading.Lock()
+
+
+def get_session(project_name: str) -> Optional[AssistantChatSession]:
+    """Get an existing session for a project."""
+    with _sessions_lock:
+        return _sessions.get(project_name)
+
+
+async def create_session(
+    project_name: str,
+    project_dir: Path,
+    conversation_id: Optional[int] = None
+) -> AssistantChatSession:
+    """
+    Create a new session for a project, closing any existing one.
+
+    Args:
+        project_name: Name of the project
+        project_dir: Absolute path to the project directory
+        conversation_id: Optional conversation ID to resume
+    """
+    old_session: Optional[AssistantChatSession] = None
+
+    with _sessions_lock:
+        old_session = _sessions.pop(project_name, None)
+        session = AssistantChatSession(project_name, project_dir, conversation_id)
+        _sessions[project_name] = session
+
+    if old_session:
+        try:
+            await old_session.close()
+        except Exception as e:
+            logger.warning(f"Error closing old session for {project_name}: {e}")
+
+    return session
+
+
+async def remove_session(project_name: str) -> None:
+    """Remove and close a session."""
+    session: Optional[AssistantChatSession] = None
+
+    with _sessions_lock:
+        session = _sessions.pop(project_name, None)
+
+    if session:
+        try:
+            await session.close()
+        except Exception as e:
+            logger.warning(f"Error closing session for {project_name}: {e}")
+
+
+def list_sessions() -> list[str]:
+    """List all active session project names."""
+    with _sessions_lock:
+        return list(_sessions.keys())
+
+
+async def cleanup_all_sessions() -> None:
+    """Close all active sessions. Called on server shutdown."""
+    sessions_to_close: list[AssistantChatSession] = []
+
+    with _sessions_lock:
+        sessions_to_close = list(_sessions.values())
+        _sessions.clear()
+
+    for session in sessions_to_close:
+        try:
+            await session.close()
+        except Exception as e:
+            logger.warning(f"Error closing session {session.project_name}: {e}")
--- a/server/services/assistant_database.py
+++ b/server/services/assistant_database.py
@@ -0,0 +1,219 @@
+"""
+Assistant Database
+==================
+
+SQLAlchemy models and functions for persisting assistant conversations.
+Each project has its own assistant.db file in the project directory.
+"""
+
+import logging
+from datetime import datetime
+from pathlib import Path
+from typing import Optional
+
+from sqlalchemy import create_engine, Column, Integer, String, Text, DateTime, ForeignKey
+from sqlalchemy.orm import sessionmaker, relationship, declarative_base
+
+logger = logging.getLogger(__name__)
+
+Base = declarative_base()
+
+
+class Conversation(Base):
+    """A conversation with the assistant for a project."""
+    __tablename__ = "conversations"
+
+    id = Column(Integer, primary_key=True, index=True)
+    project_name = Column(String(100), nullable=False, index=True)
+    title = Column(String(200), nullable=True)  # Optional title, derived from first message
+    created_at = Column(DateTime, default=datetime.utcnow)
+    updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
+
+    messages = relationship("ConversationMessage", back_populates="conversation", cascade="all, delete-orphan")
+
+
+class ConversationMessage(Base):
+    """A single message within a conversation."""
+    __tablename__ = "conversation_messages"
+
+    id = Column(Integer, primary_key=True, index=True)
+    conversation_id = Column(Integer, ForeignKey("conversations.id"), nullable=False, index=True)
+    role = Column(String(20), nullable=False)  # "user" | "assistant" | "system"
+    content = Column(Text, nullable=False)
+    timestamp = Column(DateTime, default=datetime.utcnow)
+
+    conversation = relationship("Conversation", back_populates="messages")
+
+
+def get_db_path(project_dir: Path) -> Path:
+    """Get the path to the assistant database for a project."""
+    return project_dir / "assistant.db"
+
+
+def get_engine(project_dir: Path):
+    """Get or create a SQLAlchemy engine for a project's assistant database."""
+    db_path = get_db_path(project_dir)
+    # Use as_posix() for cross-platform compatibility with SQLite connection strings
+    db_url = f"sqlite:///{db_path.as_posix()}"
+    engine = create_engine(db_url, echo=False)
+    Base.metadata.create_all(engine)
+    return engine
+
+
+def get_session(project_dir: Path):
+    """Get a new database session for a project."""
+    engine = get_engine(project_dir)
+    Session = sessionmaker(bind=engine)
+    return Session()
+
+
+# ============================================================================
+# Conversation Operations
+# ============================================================================
+
+def create_conversation(project_dir: Path, project_name: str, title: Optional[str] = None) -> Conversation:
+    """Create a new conversation for a project."""
+    session = get_session(project_dir)
+    try:
+        conversation = Conversation(
+            project_name=project_name,
+            title=title,
+        )
+        session.add(conversation)
+        session.commit()
+        session.refresh(conversation)
+        logger.info(f"Created conversation {conversation.id} for project {project_name}")
+        return conversation
+    finally:
+        session.close()
+
+
+def get_conversations(project_dir: Path, project_name: str) -> list[dict]:
+    """Get all conversations for a project with message counts."""
+    session = get_session(project_dir)
+    try:
+        conversations = (
+            session.query(Conversation)
+            .filter(Conversation.project_name == project_name)
+            .order_by(Conversation.updated_at.desc())
+            .all()
+        )
+        return [
+            {
+                "id": c.id,
+                "project_name": c.project_name,
+                "title": c.title,
+                "created_at": c.created_at.isoformat() if c.created_at else None,
+                "updated_at": c.updated_at.isoformat() if c.updated_at else None,
+                "message_count": len(c.messages),
+            }
+            for c in conversations
+        ]
+    finally:
+        session.close()
+
+
+def get_conversation(project_dir: Path, conversation_id: int) -> Optional[dict]:
+    """Get a conversation with all its messages."""
+    session = get_session(project_dir)
+    try:
+        conversation = session.query(Conversation).filter(Conversation.id == conversation_id).first()
+        if not conversation:
+            return None
+        return {
+            "id": conversation.id,
+            "project_name": conversation.project_name,
+            "title": conversation.title,
+            "created_at": conversation.created_at.isoformat() if conversation.created_at else None,
+            "updated_at": conversation.updated_at.isoformat() if conversation.updated_at else None,
+            "messages": [
+                {
+                    "id": m.id,
+                    "role": m.role,
+                    "content": m.content,
+                    "timestamp": m.timestamp.isoformat() if m.timestamp else None,
+                }
+                for m in sorted(conversation.messages, key=lambda x: x.timestamp or datetime.min)
+            ],
+        }
+    finally:
+        session.close()
+
+
+def delete_conversation(project_dir: Path, conversation_id: int) -> bool:
+    """Delete a conversation and all its messages."""
+    session = get_session(project_dir)
+    try:
+        conversation = session.query(Conversation).filter(Conversation.id == conversation_id).first()
+        if not conversation:
+            return False
+        session.delete(conversation)
+        session.commit()
+        logger.info(f"Deleted conversation {conversation_id}")
+        return True
+    finally:
+        session.close()
+
+
+# ============================================================================
+# Message Operations
+# ============================================================================
+
+def add_message(project_dir: Path, conversation_id: int, role: str, content: str) -> Optional[dict]:
+    """Add a message to a conversation."""
+    session = get_session(project_dir)
+    try:
+        conversation = session.query(Conversation).filter(Conversation.id == conversation_id).first()
+        if not conversation:
+            return None
+
+        message = ConversationMessage(
+            conversation_id=conversation_id,
+            role=role,
+            content=content,
+        )
+        session.add(message)
+
+        # Update conversation's updated_at timestamp
+        conversation.updated_at = datetime.utcnow()
+
+        # Auto-generate title from first user message if not set
+        if not conversation.title and role == "user":
+            # Take first 50 chars of first user message as title
+            conversation.title = content[:50] + ("..." if len(content) > 50 else "")
+
+        session.commit()
+        session.refresh(message)
+
+        logger.debug(f"Added {role} message to conversation {conversation_id}")
+        return {
+            "id": message.id,
+            "role": message.role,
+            "content": message.content,
+            "timestamp": message.timestamp.isoformat() if message.timestamp else None,
+        }
+    finally:
+        session.close()
+
+
+def get_messages(project_dir: Path, conversation_id: int) -> list[dict]:
+    """Get all messages for a conversation."""
+    session = get_session(project_dir)
+    try:
+        messages = (
+            session.query(ConversationMessage)
+            .filter(ConversationMessage.conversation_id == conversation_id)
+            .order_by(ConversationMessage.timestamp.asc())
+            .all()
+        )
+        return [
+            {
+                "id": m.id,
+                "role": m.role,
+                "content": m.content,
+                "timestamp": m.timestamp.isoformat() if m.timestamp else None,
+            }
+            for m in messages
+        ]
+    finally:
+        session.close()