session

@classmethod
def reset_instance(cls) -> None:
    """Reset the singleton instance (for testing)."""
    if cls._instance is not None:
        cls._instance._sessions = {}
        cls._instance._server_commands = {}
    cls._instance = None

def get_server_command_for_session(self, session_id: str) -> str | None:
    """Get the server command string that was used for a session.

    Args:
        session_id: Session identifier (format: "kind:workspace").

    Returns:
        The command string that successfully started the server, or None.
    """
    return self._server_commands.get(session_id)

async def get_or_create(
    self,
    server_kind: str,
    workspace_path: str,
    startup_timeout: float | None = None,
) -> LspClient:
    """Get existing session or create new one with fallback chain.

    Tries each configured LSP server command in order. If one fails,
    falls back to the next. Stores which command succeeded for later
    inspection.

    Args:
        server_kind: Server type ("ts", "py", etc.).
        workspace_path: Absolute path to workspace root.
        startup_timeout: Maximum seconds to wait for server initialization.
            If None, uses the value from settings.

    Returns:
        Connected LspClient instance.

    Raises:
        RuntimeError: If all server commands fail to start.
        ValueError: If server kind is not supported.
    """
    if startup_timeout is None:
        startup_timeout = get_settings().lsp_startup_timeout

    key = self._make_session_key(server_kind, workspace_path)

    if key in self._sessions:
        client = self._sessions[key]
        if client.is_connected:
            logger.debug(f"Reusing LSP session: {key}")
            return client
        # Client disconnected, remove and recreate
        logger.debug(f"Removing disconnected LSP session: {key}")
        del self._sessions[key]
        self._server_commands.pop(key, None)

    logger.info(f"Creating new LSP session: {key}")
    settings = get_settings()

    # Normalize kind for workspace config lookup
    kind = server_kind.lower()
    aliases = {"typescript": "ts", "python": "py", "javascript": "ts"}
    kind = aliases.get(kind, kind)

    commands = self._get_server_commands(server_kind)
    last_error: OSError | TimeoutError | RuntimeError | None = None

    for i, cmd in enumerate(commands):
        try:
            ws_settings = self._get_workspace_settings(kind)
            client = LspClient(
                request_timeout=float(settings.lsp_timeout),
                workspace_settings=ws_settings,
            )
            init_options = self._get_workspace_config(kind)
            await client.start(
                cmd,
                workspace_path,
                startup_timeout=startup_timeout,
                initialization_options=init_options,
            )
            self._sessions[key] = client
            self._server_commands[key] = " ".join(cmd)
            if i > 0:
                logger.info(f"LSP fallback succeeded with: {' '.join(cmd)}")
            return client
        except (OSError, TimeoutError, RuntimeError) as e:
            last_error = e
            if i < len(commands) - 1:
                logger.warning(
                    f"LSP server '{' '.join(cmd)}' failed: {e}. Trying next fallback...",
                )
            continue

    raise RuntimeError(
        f"All LSP servers failed for {server_kind}. Last error: {last_error}",
    )

def get_session(self, session_id: str) -> LspClient | None:
    """Get an existing session by ID.

    Args:
        session_id: Session identifier (format: "kind:workspace").

    Returns:
        LspClient if session exists and is connected, None otherwise.
    """
    client = self._sessions.get(session_id)
    if client and client.is_connected:
        return client
    return None

def list_sessions(self) -> list[str]:
    """List all active session IDs.

    Returns:
        List of session key strings.
    """
    return [key for key, client in self._sessions.items() if client.is_connected]

async def shutdown_session(self, session_id: str) -> bool:
    """Shutdown a specific session.

    Args:
        session_id: Session identifier to shutdown.

    Returns:
        True if session was found and shutdown, False otherwise.
    """
    client = self._sessions.pop(session_id, None)
    self._server_commands.pop(session_id, None)
    if client:
        await client.shutdown()
        logger.info(f"Shutdown LSP session: {session_id}")
        return True
    return False

async def shutdown_all(self) -> None:
    """Shutdown all active LSP sessions."""
    logger.info(f"Shutting down {len(self._sessions)} LSP sessions")
    for key, client in list(self._sessions.items()):
        try:
            await client.shutdown()
            logger.debug(f"Shutdown LSP session: {key}")
        except (OSError, TimeoutError, RuntimeError) as e:
            logger.warning(f"Error shutting down LSP session {key}: {e}")
    self._sessions.clear()
    self._server_commands.clear()