ccproxy.api.ui.terminal_permission_handler¶

ccproxy.api.ui.terminal_permission_handler ¶

Terminal UI handler for confirmation requests using Textual with request stacking support.

PendingRequest `dataclass` ¶

PendingRequest(request, future, cancelled=False)

Represents a pending confirmation request with its response future.

ConfirmationScreen ¶

ConfirmationScreen(request)

Bases: ModalScreen[bool]

Modal screen for displaying a single confirmation request.

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

def __init__(self, request: PermissionRequest) -> None:
    super().__init__()
    self.request = request
    self.start_time = time.time()
    self.countdown_timer: Timer | None = None

compose ¶

compose()

Compose the confirmation dialog.

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

def compose(self) -> ComposeResult:
    """Compose the confirmation dialog."""
    with Container(id="confirmation-dialog"):
        yield Vertical(
            Label("[bold red]Permission Request[/bold red]", id="title"),
            self._create_info_display(),
            Label("Calculating timeout...", id="countdown", classes="countdown"),
            Label(
                "[bold white]Allow this operation? (y/N):[/bold white]",
                id="question",
            ),
            id="content",
        )

on_mount ¶

on_mount()

Start the countdown timer when mounted.

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

def on_mount(self) -> None:
    """Start the countdown timer when mounted."""
    self.update_countdown()
    self.countdown_timer = self.set_interval(0.1, self.update_countdown)

update_countdown ¶

update_countdown()

Update the countdown display.

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

def update_countdown(self) -> None:
    """Update the countdown display."""
    elapsed = time.time() - self.start_time
    remaining = max(0, self.request.time_remaining() - elapsed)
    self.time_remaining = remaining

    if remaining <= 0:
        self._timeout()
    else:
        countdown_widget = self.query_one("#countdown", Label)
        if remaining > 10:
            style = "yellow"
        elif remaining > 5:
            style = "orange1"
        else:
            style = "red"
        countdown_widget.update(f"[{style}]Timeout in {remaining:.1f}s[/{style}]")

action_confirm ¶

action_confirm()

Confirm the request.

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

def action_confirm(self) -> None:
    """Confirm the request."""
    if self.countdown_timer:
        self.countdown_timer.stop()
        self.countdown_timer = None
    self.call_later(self._show_result, True, "ALLOWED")

action_deny ¶

action_deny()

Deny the request.

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

def action_deny(self) -> None:
    """Deny the request."""
    if self.countdown_timer:
        self.countdown_timer.stop()
        self.countdown_timer = None
    self.call_later(self._show_result, False, "DENIED")

action_cancel ¶

action_cancel()

Cancel the request (Ctrl+C).

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

def action_cancel(self) -> None:
    """Cancel the request (Ctrl+C)."""
    if self.countdown_timer:
        self.countdown_timer.stop()
        self.countdown_timer = None
    self.call_later(self._show_result, False, "CANCELLED")
    # Raise KeyboardInterrupt to forward it up
    raise KeyboardInterrupt("User cancelled confirmation")

ConfirmationApp ¶

ConfirmationApp(request)

Bases: App[bool]

Simple Textual app for a single confirmation request.

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

def __init__(self, request: PermissionRequest) -> None:
    super().__init__()
    self.theme = "textual-ansi"
    self.request = request
    self.result = False
    self.start_time = time.time()
    self.countdown_timer: Timer | None = None

compose ¶

compose()

Compose the confirmation dialog directly.

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

def compose(self) -> ComposeResult:
    """Compose the confirmation dialog directly."""
    with Container(id="confirmation-dialog"):
        yield Vertical(
            Label("[bold red]Permission Request[/bold red]", id="title"),
            self._create_info_display(),
            Label("Calculating timeout...", id="countdown", classes="countdown"),
            Label(
                "[bold white]Allow this operation? (y/N):[/bold white]",
                id="question",
            ),
            id="content",
        )

on_mount ¶

on_mount()

Start the countdown timer when mounted.

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

def on_mount(self) -> None:
    """Start the countdown timer when mounted."""
    self.update_countdown()
    self.countdown_timer = self.set_interval(0.1, self.update_countdown)

update_countdown ¶

update_countdown()

Update the countdown display.

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

def update_countdown(self) -> None:
    """Update the countdown display."""
    elapsed = time.time() - self.start_time
    remaining = max(0, self.request.time_remaining() - elapsed)
    self.time_remaining = remaining

    if remaining <= 0:
        self._timeout()
    else:
        countdown_widget = self.query_one("#countdown", Label)
        if remaining > 10:
            style = "yellow"
        elif remaining > 5:
            style = "orange1"
        else:
            style = "red"
        countdown_widget.update(f"[{style}]Timeout in {remaining:.1f}s[/{style}]")

action_confirm ¶

action_confirm()

Confirm the request.

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

def action_confirm(self) -> None:
    """Confirm the request."""
    if self.countdown_timer:
        self.countdown_timer.stop()
        self.countdown_timer = None
    self.call_later(self._show_result, True, "ALLOWED")

action_deny ¶

action_deny()

Deny the request.

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

def action_deny(self) -> None:
    """Deny the request."""
    if self.countdown_timer:
        self.countdown_timer.stop()
        self.countdown_timer = None
    self.call_later(self._show_result, False, "DENIED")

action_cancel ¶

action_cancel()

Cancel the request (Ctrl+C).

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

def action_cancel(self) -> None:
    """Cancel the request (Ctrl+C)."""
    if self.countdown_timer:
        self.countdown_timer.stop()
        self.countdown_timer = None
    self.call_later(self._show_result, False, "CANCELLED")
    # Raise KeyboardInterrupt to forward it up
    raise KeyboardInterrupt("User cancelled confirmation")

on_key `async` ¶

on_key(event)

Handle global key events, especially Ctrl+C.

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

async def on_key(self, event: Key) -> None:
    """Handle global key events, especially Ctrl+C."""
    if event.key == "ctrl+c":
        # Forward the KeyboardInterrupt
        self.exit(False)
        raise KeyboardInterrupt("User cancelled confirmation")

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

ccproxy.api.ui.terminal_permission_handler¶