SessionPool Global Permission/Question Listing API Design¶

# agentpool/models/pending_interaction.py
from typing import Protocol
from datetime import datetime

class PendingQuestion(Protocol):
    """Protocol for pending questions."""
    id: str
    session_id: str
    tool_name: str
    content: str
    created_at: datetime

class PendingPermission(Protocol):
    """Protocol for pending permissions."""
    id: str
    session_id: str
    tool_name: str
    content: str
    created_at: datetime

# agentpool_server/opencode_server/models/question_permission.py
from dataclasses import dataclass, field
from datetime import datetime

@dataclass
class OpenCodePendingQuestion:
    """Concrete implementation for OpenCode protocol."""
    id: str
    session_id: str
    tool_name: str
    content: str
    created_at: datetime = field(default_factory=datetime.utcnow)

@dataclass
class OpenCodePendingPermission:
    """Concrete implementation for OpenCode protocol."""
    id: str
    session_id: str
    tool_name: str
    content: str
    created_at: datetime = field(default_factory=datetime.utcnow)

class SessionPool:
    async def list_pending_questions(
        self,
        *,
        session_id: str | None = None,
        tool_name: str | None = None,
        limit: int = 100,
        offset: int = 0,
    ) -> list[PendingQuestion]:
        """List pending questions across all sessions.

        Args:
            session_id: Filter by session ID. If None, returns questions
                from all sessions.
            tool_name: Filter by tool name. If None, returns questions
                from all tools.
            limit: Maximum number of questions to return (default 100).
            offset: Number of questions to skip (default 0).

        Returns:
            List of pending questions ordered by created_at (newest first).
        """

    async def list_pending_permissions(
        self,
        *,
        session_id: str | None = None,
        tool_name: str | None = None,
        limit: int = 100,
        offset: int = 0,
    ) -> list[PendingPermission]:
        """List pending permissions across all sessions.

        Args:
            session_id: Filter by session ID. If None, returns permissions
                from all sessions.
            tool_name: Filter by tool name. If None, returns permissions
                from all tools.
            limit: Maximum number of permissions to return (default 100).
            offset: Number of permissions to skip (default 0).

        Returns:
            List of pending permissions ordered by created_at (newest first).
        """

    async def cancel_all_pending_questions(
        self,
        session_id: str | None = None,
    ) -> list[str]:
        """Cancel all pending questions.

        Args:
            session_id: If set, only cancel questions for this session.
                If None, cancel questions across all sessions.

        Returns:
            List of cancelled question IDs.
        """

    async def cancel_all_pending_permissions(
        self,
        session_id: str | None = None,
    ) -> list[str]:
        """Cancel all pending permissions.

        Args:
            session_id: If set, only cancel permissions for this session.
                If None, cancel permissions across all sessions.

        Returns:
            List of cancelled permission IDs.
        """

# During Migration A, SessionPool.list_pending_questions() must:
# 1. Iterate ServerState.pending_questions (global dict)
# 2. Also check session.input_provider for permissions

class SessionPool:
    async def list_pending_questions(self, *, session_id=None, tool_name=None, limit=100, offset=0):
        all_questions: list[PendingQuestion] = []

        # Source 1: ServerState.pending_questions (Migration A compat)
        # This requires ServerState shim to expose pending_questions
        # or SessionPool maintains its own index

        # Source 2: session.input_provider (after A1.7 migration)
        sessions = [self.sessions.get_session(session_id)] if session_id else self.sessions._sessions.values()
        for session in sessions:
            if session is not None and session.input_provider is not None:
                # input_provider has get_pending_questions() after A1.7
                questions = session.input_provider.get_pending_questions()
                for q in questions:
                    if tool_name is None or q.tool_name == tool_name:
                        all_questions.append(q)

        # Sort by created_at descending
        all_questions.sort(key=lambda q: q.created_at, reverse=True)

        # Apply pagination
        return all_questions[offset:offset + limit]

# Cursor-based (recommended for live data)
async def list_pending_questions(
    self,
    *,
    session_id: str | None = None,
    tool_name: str | None = None,
    limit: int = 100,
    after_cursor: str | None = None,  # Question ID to start after
) -> tuple[list[PendingQuestion], str | None]:
    """Returns (questions, next_cursor). next_cursor is None when no more results."""

class SessionController:
    def __init__(self):
        self._pending_question_index: dict[str, set[str]] = {}  # session_id -> {question_ids}
        self._pending_permission_index: dict[str, set[str]] = {}  # session_id -> {permission_ids}

@router.get("/")
async def list_questions(state: StateDep):
    pending = _get_all_pending_questions(state)
    return [
        QuestionRequest(
            id=question_id,
            session_id=i.session_id,
            questions=i.questions,
            tool=i.tool,
        )
        for question_id, i in pending.items()
    ]

@router.get("/questions")
async def list_questions(session_pool: SessionPool):
    questions = await session_pool.list_pending_questions()
    return [dataclasses.asdict(q) for q in questions]

state.cancel_all_pending_questions()

session_controller.cancel_all_pending_questions()