clearance_screen

clearance_screen

In-TUI clearance screen for live D-Bus shield verdict handling.

Provides a ClearanceScreen backed by terok_clearance.CallbackNotifier plugged into terok_clearance.EventSubscriber. The subscriber handles the full signal-to-verdict cycle; the callback notifier bridges D-Bus events into Textual messages so the screen can render blocked connections and route operator Allow/Deny actions back through D-Bus.

The screen listens on the whole session bus — all containers' events are shown, with the container name displayed prominently on every row.

Dual use:

• Embedded — pushed as a screen inside terok-tui.
• Standalone — terok clearance launches ClearanceApp, a minimal Textual app containing only this screen.

ClearanceScreen()

Bases: Screen[None]

Full-page screen for live D-Bus shield clearance verdicts.

Initialise clearance screen state.

Source code in src/terok/tui/clearance_screen.py

def __init__(self) -> None:
    """Initialise clearance screen state."""
    super().__init__()
    self._notifier: Any = None  # CallbackNotifier
    self._subscriber: Any = None  # EventSubscriber
    self._pending: dict[int, _PendingRequest] = {}

BINDINGS = [_modal_binding('escape', 'dismiss_screen', 'Back'), _modal_binding('q', 'dismiss_screen', 'Back'), _modal_binding('a', 'allow_selected', 'Allow'), _modal_binding('x', 'deny_selected', 'Deny')] class-attribute instance-attribute

CSS = '\n ClearanceScreen {\n layout: vertical;\n background: $background;\n }\n #clearance-header {\n height: 1;\n background: $primary;\n color: $text;\n padding: 0 1;\n }\n #pending-list {\n height: auto;\n max-height: 40%;\n border: round $primary;\n border-title-align: right;\n background: $surface;\n }\n #event-log {\n height: 1fr;\n }\n ' class-attribute instance-attribute

compose()

Build header, pending list, event log.

The footer is not composed here: when this screen is pushed inside the host terok-tui app its parent Footer already renders the active screen's bindings, and doubling up would produce two footer bars. The standalone ClearanceApp composes its own Footer so the bindings still show.

Source code in src/terok/tui/clearance_screen.py

def compose(self) -> ComposeResult:
    """Build header, pending list, event log.

    The footer is *not* composed here: when this screen is pushed
    inside the host ``terok-tui`` app its parent ``Footer`` already
    renders the active screen's bindings, and doubling up would
    produce two footer bars.  The standalone [`ClearanceApp`][terok.tui.clearance_screen.ClearanceApp]
    composes its own ``Footer`` so the bindings still show.
    """
    yield Static(" Shield Clearance", id="clearance-header")
    pending = ListView(id="pending-list")
    pending.border_title = "Pending (0)"
    yield pending
    yield RichLog(auto_scroll=True, max_lines=1000, id="event-log")

on_mount() async

Connect to the clearance hub and start the event subscriber.

Source code in src/terok/tui/clearance_screen.py

async def on_mount(self) -> None:
    """Connect to the clearance hub and start the event subscriber."""
    log = self.query_one(_ID_EVENT_LOG, RichLog)
    try:
        from terok.lib.api.clearance import CallbackNotifier, EventSubscriber

        self._notifier = CallbackNotifier(
            on_notify=self._on_notify,
            on_container_started=self._on_container_started,
            on_container_exited=self._on_container_exited,
        )
        # Identity resolution is no longer a TUI concern: the shield
        # reader resolves the orchestrator dossier at emit time and
        # ships it on every event, so the subscriber just reads it.
        self._subscriber = EventSubscriber(self._notifier)
        await self._subscriber.start()
        log.write(Text("Connected to clearance hub...", style=_STYLE_INFO))
    except Exception as exc:
        _log.debug("clearance hub connection failed: %s", exc)
        log.write(Text(f"clearance hub unavailable: {exc}", style=_STYLE_ERROR))
        if self._subscriber:
            try:
                await self._subscriber.stop()
            except Exception:
                _log.debug("Failed to stop subscriber during error cleanup", exc_info=True)
        if self._notifier:
            try:
                await self._notifier.disconnect()
            except Exception:
                _log.debug("Failed to disconnect notifier during error cleanup", exc_info=True)
        self._notifier = None
        self._subscriber = None

on_app_focus(_event)

Cut short any reconnect back-off when the operator refocuses.

Source code in src/terok/tui/clearance_screen.py

def on_app_focus(self, _event: events.AppFocus) -> None:
    """Cut short any reconnect back-off when the operator refocuses."""
    if self._subscriber is not None:
        with contextlib.suppress(Exception):
            self._subscriber.poke_reconnect()

on_unmount() async

Stop the subscriber and release resources.

Source code in src/terok/tui/clearance_screen.py

async def on_unmount(self) -> None:
    """Stop the subscriber and release resources."""
    try:
        if self._subscriber:
            await self._subscriber.stop()
    except Exception as exc:
        _log.debug("Failed to stop EventSubscriber: %s", exc)
    finally:
        if self._notifier:
            try:
                await self._notifier.disconnect()
            except Exception as exc:
                _log.debug("Failed to disconnect CallbackNotifier: %s", exc)

on__notification_posted(message)

Handle notifications from the CallbackNotifier.

Source code in src/terok/tui/clearance_screen.py

def on__notification_posted(self, message: _NotificationPosted) -> None:
    """Handle notifications from the CallbackNotifier."""
    try:
        log = self.query_one(_ID_EVENT_LOG, RichLog)
        pending_list = self.query_one(_ID_PENDING, ListView)
    except NoMatches:
        return

    rendered = _render_notification(message)
    if message.replaces_id and message.replaces_id in self._pending:
        # Verdict applied — remove from pending, log result
        del self._pending[message.replaces_id]
        self._remove_pending_item(message.replaces_id)
        style = _STYLE_ALLOWED if "Allowed" in message.summary else _STYLE_DENIED
        log.write(Text(rendered, style=style))
    elif message.actions:
        # New blocked connection — add to pending
        req = _PendingRequest(nid=message.nid, summary=message.summary, body=message.body)
        self._pending[message.nid] = req
        label = Static(rendered, markup=False)
        item = ListItem(label)
        item.clearance_nid = message.nid  # type: ignore[attr-defined]
        pending_list.append(item)
        # The style alone communicates "this is a block" — drop the redundant
        # "BLOCKED  " prefix that used to double up with the "Blocked:" title.
        log.write(Text(rendered, style=_STYLE_BLOCKED))
    else:
        # Informational (e.g. verdict details)
        log.write(Text(rendered, style=_STYLE_INFO))

    pending_list.border_title = f"Pending ({len(self._pending)})"

on__lifecycle_posted(message)

Render container-lifecycle events in the scrolling log.

Source code in src/terok/tui/clearance_screen.py

def on__lifecycle_posted(self, message: _LifecyclePosted) -> None:
    """Render container-lifecycle events in the scrolling log."""
    try:
        log = self.query_one(_ID_EVENT_LOG, RichLog)
    except NoMatches:
        return
    if message.event == "started":
        log.write(Text(f"Container connected: {message.container}", style=_STYLE_INFO))
    else:
        tail = f" ({message.reason})" if message.reason else ""
        log.write(Text(f"Container gone: {message.container}{tail}", style=_STYLE_INFO))

action_allow_selected()

Send an allow verdict for the highlighted pending request.

Source code in src/terok/tui/clearance_screen.py

def action_allow_selected(self) -> None:
    """Send an ``allow`` verdict for the highlighted pending request."""
    self._send_verdict("allow")

action_deny_selected()

Send a deny verdict for the highlighted pending request.

Source code in src/terok/tui/clearance_screen.py

def action_deny_selected(self) -> None:
    """Send a ``deny`` verdict for the highlighted pending request."""
    self._send_verdict("deny")

action_dismiss_screen()

Close the clearance screen.

Source code in src/terok/tui/clearance_screen.py

def action_dismiss_screen(self) -> None:
    """Close the clearance screen."""
    self.dismiss(None)

ClearanceApp

Bases: App

Minimal Textual app containing only the ClearanceScreen.

The app-level Footer auto-renders the pushed screen's bindings, so operators see a Allow x Deny Esc Back without us maintaining a hand-written hint string. The command palette (^p) is disabled — this tool's surface is four verdict keys, and a palette prompt would just confuse.

TITLE = 'terok clearance' class-attribute instance-attribute

ENABLE_COMMAND_PALETTE = False class-attribute instance-attribute

compose()

Pair an app-level Footer with the pushed clearance screen.

Source code in src/terok/tui/clearance_screen.py

def compose(self) -> ComposeResult:
    """Pair an app-level ``Footer`` with the pushed clearance screen."""
    yield Footer()

on_mount()

Push the clearance screen on startup.

Source code in src/terok/tui/clearance_screen.py

def on_mount(self) -> None:
    """Push the clearance screen on startup."""
    self.push_screen(ClearanceScreen(), callback=lambda _: self.exit())

main()

Entry point for terok clearance standalone command.

Source code in src/terok/tui/clearance_screen.py

def main() -> None:
    """Entry point for ``terok clearance`` standalone command."""
    ClearanceApp().run()