ccproxy.api.ui

ccproxy.api.ui

UI components for CCProxy API.

TerminalPermissionHandler

TerminalPermissionHandler()

Handles confirmation requests in the terminal using Textual with request stacking.

Implements ConfirmationHandlerProtocol for type safety and interoperability.

Source code in ccproxy/api/ui/terminal_permission_handler.py

def __init__(self) -> None:
    """Initialize the terminal confirmation handler."""
    self._request_queue: (
        asyncio.Queue[tuple[PermissionRequest, asyncio.Future[bool]]] | None
    ) = None
    self._cancelled_requests: set[str] = set()
    self._processing_task: asyncio.Task[None] | None = None
    self._active_apps: dict[str, ConfirmationApp] = {}

handle_permission async

handle_permission(request)

Handle a permission request.

Parameters:

Name	Type	Description	Default
request	PermissionRequest	The permission request to handle	required

Returns:

Name	Type	Description
bool	bool	True if the user confirmed, False otherwise

Source code in ccproxy/api/ui/terminal_permission_handler.py

async def handle_permission(self, request: PermissionRequest) -> bool:
    """Handle a permission request.

    Args:
        request: The permission request to handle

    Returns:
        bool: True if the user confirmed, False otherwise
    """
    try:
        logger.info(
            "handling_confirmation_request",
            request_id=request.id,
            tool_name=request.tool_name,
            time_remaining=request.time_remaining(),
        )

        # Check if request has already expired
        if request.time_remaining() <= 0:
            logger.info("confirmation_request_expired", request_id=request.id)
            return False

        # Ensure processing task is running
        self._ensure_processing_task_running()

        # Queue request and wait for result
        result = await self._queue_and_wait_for_result(request)

        logger.info(
            "confirmation_request_completed", request_id=request.id, result=result
        )

        return result

    except Exception as e:
        logger.error(
            "confirmation_handling_error",
            request_id=request.id,
            error=str(e),
            exc_info=True,
        )
        return False

cancel_confirmation

cancel_confirmation(request_id, reason='cancelled')

Cancel an ongoing confirmation request.

Parameters:

Name	Type	Description	Default
request_id	str	The ID of the request to cancel	required
reason	str	The reason for cancellation	'cancelled'

Source code in ccproxy/api/ui/terminal_permission_handler.py

def cancel_confirmation(self, request_id: str, reason: str = "cancelled") -> None:
    """Cancel an ongoing confirmation request.

    Args:
        request_id: The ID of the request to cancel
        reason: The reason for cancellation
    """
    logger.info("cancelling_confirmation", request_id=request_id, reason=reason)
    self._cancelled_requests.add(request_id)

    # If there's an active dialog for this request, close it immediately
    if request_id in self._active_apps:
        app = self._active_apps[request_id]
        # Schedule the cancellation feedback asynchronously
        asyncio.create_task(self._cancel_active_dialog(app, reason))

shutdown async

shutdown()

Shutdown the handler and cleanup resources.

Source code in ccproxy/api/ui/terminal_permission_handler.py

async def shutdown(self) -> None:
    """Shutdown the handler and cleanup resources."""
    if self._processing_task and not self._processing_task.done():
        self._processing_task.cancel()
        with contextlib.suppress(asyncio.CancelledError):
            await self._processing_task

    self._processing_task = None