feat(copilot): add user-aware LLM calls and project cleanup

- Modified `llm_call` and `generate_title` functions to accept `config` parameter, extracting `user_id` and `jwt_token` for per-user LLM configuration and API authentication
- Updated `create_base_model_with_tools` and `create_title_model` calls to pass user authentication details
- Added `jwt_token` to state for tool usage in GNS3 API calls
- Integrated chat router into controller API routes under `/chat` endpoint
- Implemented `_cleanup_copilot_agent` method in `Project` class to remove AgentService resources upon project closure, preventing resource leaks
- Enhanced error handling in agent cleanup to avoid interrupting project close operations

gns3server/agent/gns3_copilot/agent/gns3_copilot.py

@@ -124,9 +124,14 @@ class MessagesState(TypedDict):
 
 
 # Define llm call  node
-def llm_call(state: dict):
+def llm_call(state: dict, config: dict = None):
     """LLM decides whether to call a tool or not"""
 
+    # Extract user authentication info from LangGraph config
+    configurable = config.get("configurable", {}) if config else {}
+    user_id = configurable.get("user_id")
+    jwt_token = configurable.get("jwt_token")
+
     # Defensive check: skip LLM call if no user messages
     messages = state.get("messages", [])
     if not messages or len(messages) == 0:
@@ -203,23 +208,35 @@ def llm_call(state: dict):
     # print(full_messages)
 
     # Create fresh model with tools for each LLM call
-    # This ensures configuration changes in .env take effect immediately
-    model_with_tools = create_base_model_with_tools(tools)
+    # This ensures configuration changes take effect immediately
+    # Pass user_id and jwt_token for per-user LLM config and API authentication
+    model_with_tools = create_base_model_with_tools(
+        tools,
+        user_id=user_id,
+        jwt_token=jwt_token
+    )
 
+    # Store jwt_token in state for Tools to use when calling GNS3 API
     return {
         "messages": [model_with_tools.invoke(full_messages)],
         "llm_calls": state.get("llm_calls", 0) + 1,
         "topology_info": topology_info,
+        "jwt_token": jwt_token,
     }
 
 
 # Define generate title node
-def generate_title(state: MessagesState) -> dict:
+def generate_title(state: MessagesState, config: dict = None) -> dict:
     """
     Generate a conversation title using a lightweight assistant LLM (title_model).
     This node is only executed when no title has been set yet (first round only).
     """
 
+    # Extract user authentication info from LangGraph config
+    configurable = config.get("configurable", {}) if config else {}
+    user_id = configurable.get("user_id")
+    jwt_token = configurable.get("jwt_token")
+
     # Only generate a title if it hasn't been set yet
     current_title = state.get("conversation_title")
     if current_title in [None, "New Conversation"]:
@@ -236,7 +253,8 @@ def generate_title(state: MessagesState) -> dict:
         # Call the title generation model (create fresh instance for each call)
         try:
             # Create fresh title model instance from current env configuration
-            title_model = create_title_model()
+            # Pass user_id and jwt_token for per-user LLM config
+            title_model = create_title_model(user_id=user_id, jwt_token=jwt_token)
             response = title_model.invoke(
                 title_prompt_messages, config={"configurable": {"foo_temperature": 1.0}}
             )

gns3server/agent/gns3_copilot/agent_service.py

@@ -0,0 +1,282 @@
+"""
+GNS3 Copilot Agent Service
+
+Provides project-level Agent instances with SQLite checkpoint management.
+Each project has its own AgentService with a dedicated checkpoint database
+in the project directory.
+"""
+
+import asyncio
+import logging
+import os
+from typing import AsyncGenerator, Dict, Any, Optional
+from uuid import uuid4
+
+import aiosqlite
+from langchain_core.messages import HumanMessage, AIMessage, ToolMessage
+from langgraph.checkpoint.sqlite.aio import AsyncSqliteSaver
+
+from gns3_copilot.agent.gns3_copilot import agent_builder
+
+log = logging.getLogger(__name__)
+
+
+class AgentService:
+    """
+    Project-level Agent Service with async checkpoint management.
+
+    Manages a LangGraph agent instance with SQLite-based state persistence
+    for a single GNS3 project.
+    """
+
+    def __init__(self, project_path: str):
+        """
+        Initialize AgentService for a project.
+
+        Args:
+            project_path: Path to the GNS3 project directory
+        """
+        self.project_path = project_path
+        self._checkpointer: Optional[AsyncSqliteSaver] = None
+        self._checkpointer_conn: Optional[aiosqlite.Connection] = None
+        self._checkpointer_path: Optional[str] = None
+        self._graph = None
+        self._init_lock = asyncio.Lock()
+        self._initialized = False
+
+    def _get_checkpoint_dir(self) -> str:
+        """Get or create the checkpoint directory for this project."""
+        checkpoint_dir = os.path.join(self.project_path, ".gns3-copilot")
+        os.makedirs(checkpoint_dir, exist_ok=True)
+        return checkpoint_dir
+
+    async def _get_checkpointer(self) -> AsyncSqliteSaver:
+        """
+        Get or create the SQLite checkpointer for this project.
+
+        Returns:
+            AsyncSqliteSaver instance
+        """
+        async with self._init_lock:
+            if self._checkpointer is not None:
+                return self._checkpointer
+
+            checkpoint_dir = self._get_checkpoint_dir()
+            checkpointer_path = os.path.join(checkpoint_dir, "copilot_checkpoints.db")
+
+            log.debug("Creating checkpointer at: %s", checkpointer_path)
+
+            # Close existing connection if switching projects
+            if self._checkpointer_conn:
+                try:
+                    await self._checkpointer_conn.close()
+                    log.debug("Closed previous checkpointer connection")
+                except Exception as e:
+                    log.warning("Error closing old checkpointer connection: %s", e)
+
+            # Create new connection
+            conn = await aiosqlite.connect(checkpointer_path)
+            # Enable WAL mode for better concurrent performance
+            await conn.execute("PRAGMA journal_mode=WAL;")
+            self._checkpointer_conn = conn  # Save connection reference to prevent GC
+            self._checkpointer = AsyncSqliteSaver(conn)
+
+            # CRITICAL: Initialize database schema
+            await self._checkpointer.setup()
+
+            self._checkpointer_path = checkpointer_path
+            self._initialized = True
+
+            log.info("Project checkpointer created at: %s", checkpointer_path)
+            return self._checkpointer
+
+    async def _get_graph(self):
+        """Get or compile the LangGraph agent."""
+        if self._graph is None:
+            checkpointer = await self._get_checkpointer()
+            self._graph = agent_builder.compile(checkpointer=checkpointer)
+            log.info("LangGraph agent compiled for project: %s", self.project_path)
+        return self._graph
+
+    async def stream_chat(
+        self,
+        message: str,
+        session_id: str,
+        project_id: Optional[str] = None,
+        user_id: Optional[str] = None,
+        jwt_token: Optional[str] = None,
+        mode: str = "text"
+    ) -> AsyncGenerator[Dict[str, Any], None]:
+        """
+        Stream chat responses from the agent.
+
+        Args:
+            message: User message
+            session_id: Session/thread ID for conversation continuity
+            project_id: GNS3 project ID (optional, for context)
+            user_id: User ID for LLM config lookup (optional)
+            jwt_token: JWT token for API authentication (optional)
+            mode: Interaction mode (default: "text")
+
+        Yields:
+            Dict containing SSE-compatible response chunks
+        """
+        # Build config with user authentication info
+        config = {
+            "configurable": {
+                "thread_id": session_id,
+                "user_id": user_id,
+                "jwt_token": jwt_token,
+            }
+        }
+
+        # Build inputs
+        inputs = {
+            "messages": [HumanMessage(content=message)],
+            "llm_calls": 0,
+            "remaining_steps": 20,
+            "mode": mode,
+        }
+
+        # Get the compiled graph
+        graph = await self._get_graph()
+
+        # Stream events
+        try:
+            async for event in graph.astream_events(inputs, config=config, version="v2"):
+                chunk = self._convert_event_to_chunk(event, session_id)
+                if chunk:
+                    yield chunk
+
+            yield {"type": "done", "session_id": session_id}
+
+        except Exception as e:
+            log.error("Error in stream_chat: %s", e, exc_info=True)
+            yield {"type": "error", "error": str(e), "session_id": session_id}
+
+    def _convert_event_to_chunk(self, event: Dict[str, Any], session_id: str) -> Optional[Dict[str, Any]]:
+        """
+        Convert LangGraph event to API response chunk.
+
+        Args:
+            event: LangGraph event from astream_events
+            session_id: Session ID for the response
+
+        Returns:
+            Dict for SSE response or None if event should be filtered
+        """
+        event_type = event.get("event", "")
+        data = event.get("data", {})
+
+        if event_type == "on_chat_model_stream":
+            # Streaming text content from LLM
+            chunk = data.get("chunk", {})
+            content = chunk.get("content", "")
+            if content:
+                return {"type": "content", "content": content}
+
+        elif event_type == "on_tool_start":
+            # Tool execution started
+            return {
+                "type": "tool_start",
+                "tool_name": event.get("name", ""),
+                "session_id": session_id
+            }
+
+        elif event_type == "on_tool_end":
+            # Tool execution completed
+            output = data.get("output", "")
+            # Convert output to string if it's not already
+            if not isinstance(output, str):
+                output = str(output)
+            return {
+                "type": "tool_end",
+                "tool_name": event.get("name", ""),
+                "tool_output": output,
+                "session_id": session_id
+            }
+
+        return None
+
+    async def get_history(self, session_id: str, limit: int = 100) -> Dict[str, Any]:
+        """
+        Get conversation history for a session.
+
+        Args:
+            session_id: Session/thread ID
+            limit: Maximum number of messages to retrieve
+
+        Returns:
+            Dict containing thread_id, title, and messages
+        """
+        config = {"configurable": {"thread_id": session_id}}
+
+        try:
+            graph = await self._get_graph()
+            state = await graph.aget_state(config)
+
+            if state and "messages" in state.values:
+                messages = []
+                for msg in state.values["messages"][-limit:]:
+                    messages.append(self._convert_message_to_dict(msg))
+
+                title = state.values.get("conversation_title", "New Conversation")
+
+                return {
+                    "thread_id": session_id,
+                    "title": title,
+                    "messages": messages
+                }
+        except Exception as e:
+            log.error("Error getting history: %s", e, exc_info=True)
+
+        return {
+            "thread_id": session_id,
+            "title": "New Conversation",
+            "messages": []
+        }
+
+    def _convert_message_to_dict(self, msg) -> Dict[str, Any]:
+        """Convert a LangChain message to dict format."""
+        from datetime import datetime
+
+        msg_type = type(msg).__name__
+
+        result = {
+            "id": getattr(msg, "id", str(uuid4())),
+            "role": "user",
+            "content": getattr(msg, "content", str(msg)),
+            "created_at": datetime.utcnow().isoformat() + "Z",
+        }
+
+        if msg_type == "HumanMessage":
+            result["role"] = "user"
+        elif msg_type == "AIMessage":
+            result["role"] = "assistant"
+            if hasattr(msg, "tool_calls") and msg.tool_calls:
+                result["tool_calls"] = msg.tool_calls
+        elif msg_type == "ToolMessage":
+            result["role"] = "tool"
+            result["tool_call_id"] = getattr(msg, "tool_call_id", None)
+            result["name"] = getattr(msg, "name", None)
+        elif msg_type == "SystemMessage":
+            result["role"] = "system"
+
+        return result
+
+    async def close(self):
+        """
+        Close the checkpointer connection and cleanup resources.
+        """
+        async with self._init_lock:
+            if self._checkpointer_conn:
+                try:
+                    await self._checkpointer_conn.close()
+                    log.debug("Checkpointer connection closed for: %s", self.project_path)
+                except Exception as e:
+                    log.warning("Error closing checkpointer connection: %s", e)
+                finally:
+                    self._checkpointer_conn = None
+                    self._checkpointer = None
+                    self._graph = None
+                    self._initialized = False

gns3server/agent/gns3_copilot/project_agent_manager.py

@@ -0,0 +1,121 @@
+"""
+Project Agent Manager
+
+Manages AgentService instances for GNS3 projects using a singleton pattern.
+Each project has its own AgentService with a dedicated SQLite checkpoint database.
+"""
+
+import asyncio
+import logging
+from typing import Dict, Optional
+
+from gns3server.agent.gns3_copilot.agent_service import AgentService
+
+log = logging.getLogger(__name__)
+
+
+class ProjectAgentManager:
+    """
+    Singleton manager for project-level Agent services.
+
+    Manages the lifecycle of AgentService instances, ensuring that each project
+    has exactly one AgentService instance. Handles cleanup of resources when
+    projects are closed.
+    """
+
+    _instance: Optional["ProjectAgentManager"] = None
+    _lock = asyncio.Lock()
+
+    def __new__(cls):
+        if cls._instance is None:
+            cls._instance = super().__new__(cls)
+            cls._instance._agents: Dict[str, AgentService] = {}
+            cls._instance._lock = asyncio.Lock()
+        return cls._instance
+
+    async def get_agent(self, project_id: str, project_path: str) -> AgentService:
+        """
+        Get or create an AgentService for a project.
+
+        Args:
+            project_id: GNS3 project ID
+            project_path: Path to the GNS3 project directory
+
+        Returns:
+            AgentService instance for the project
+        """
+        async with self._lock:
+            if project_id not in self._agents:
+                log.info("Creating new AgentService for project: %s at %s", project_id, project_path)
+                self._agents[project_id] = AgentService(project_path)
+            return self._agents[project_id]
+
+    async def remove_agent(self, project_id: str):
+        """
+        Remove and cleanup an AgentService for a project.
+
+        Should be called when a project is closed to free resources.
+
+        Args:
+            project_id: GNS3 project ID
+        """
+        async with self._lock:
+            if project_id in self._agents:
+                log.info("Removing AgentService for project: %s", project_id)
+                agent = self._agents.pop(project_id)
+                await agent.close()
+
+    async def close_all(self):
+        """
+        Close all AgentService instances and cleanup resources.
+
+        Should be called on server shutdown.
+        """
+        async with self._lock:
+            log.info("Closing all AgentService instances (%d projects)", len(self._agents))
+            for project_id, agent in self._agents.items():
+                log.debug("Closing AgentService for project: %s", project_id)
+                await agent.close()
+            self._agents.clear()
+
+    def has_agent(self, project_id: str) -> bool:
+        """
+        Check if an AgentService exists for a project.
+
+        Args:
+            project_id: GNS3 project ID
+
+        Returns:
+            True if AgentService exists, False otherwise
+        """
+        return project_id in self._agents
+
+    @property
+    def active_projects(self) -> list[str]:
+        """
+        Get list of project IDs with active AgentService instances.
+
+        Returns:
+            List of project IDs
+        """
+        return list(self._agents.keys())
+
+
+# Global singleton instance
+_project_agent_manager: Optional[ProjectAgentManager] = None
+_manager_lock = asyncio.Lock()
+
+
+async def get_project_agent_manager() -> ProjectAgentManager:
+    """
+    Get the global ProjectAgentManager singleton instance.
+
+    Returns:
+        ProjectAgentManager instance
+    """
+    global _project_agent_manager
+    async with _manager_lock:
+        if _project_agent_manager is None:
+            _project_agent_manager = ProjectAgentManager()
+            log.info("ProjectAgentManager singleton created")
+        return _project_agent_manager