presenter_events.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# oc_chat/core/presenter_events.py
 
"""
Bus → presenter bridge for the approval and sub-agent lanes — plan 188 T2.
 
Purpose
-------
``run_turn`` and ``run_sub_loop`` already publish ``tool_call_proposed``,
``permission_decision``, ``subagent_started`` and ``subagent_completed``
on the M2 bus, and the NDJSON :class:`~oc_chat.core.event_writer.EventWriter`
already persists them — but nothing ever put them on screen. The
plan→approve→act lane the substrate models (synthesis §1) was invisible,
and a sub-agent's *completion* reached only ``_dbg``.
 
This module is the read side for the terminal: one subscriber that folds
those kinds into :meth:`~oc_ui.presenter_base.Presenter.show_activity`
calls. It is the direct analogue of
:mod:`oc_chat.core.event_bootstrap` — same one-per-process, idempotent,
attach-at-session-bootstrap shape — and it is deliberately a *subscriber*
rather than call-site rendering:
 
- the approval and sub-agent lanes exist ONLY as events (the loop's own
  call sites hand the presenter nothing about them), so the bus is where
  the information actually is;
- it keeps the ReAct loop free of another presenter branch;
- "a dashboard is just another subscriber" is the banked architecture —
  this is that same seam used for the terminal, which is exactly the
  reuse Phase 1's projection/reducer will inherit.
 
**Nothing here publishes.** The bus stays observability, not control
(synthesis §2), and no event kind or payload changes.
 
Responsibilities
----------------
- :class:`PresenterEventBridge` — the subscriber: event → activity line,
  plus per-turn counters exposed via :meth:`PresenterEventBridge.snapshot`.
- :func:`attach_presenter_bridge` — idempotently subscribe one bridge.
- :func:`detach_presenter_bridge` — unsubscribe and clear the singleton.
 
Diagnostics
-----------
Domain: OC_CHAT.MAIN
Levels:
    L2 — attach / detach lifecycle.
    L3 — per-event rendering decisions.
    L4 — module-load coverage marker.
 
Contracts
---------
- ``attach_presenter_bridge`` is idempotent: a second call returns the
  same bridge and never double-subscribes (which would double-render).
- A presenter without ``show_activity`` (one predating plan 188) is
  tolerated: the bridge checks and skips. The bus already isolates
  subscriber faults, but a renderer must not rely on that.
- ``tool_call_proposed`` is COUNTED, never rendered: ``show_tool_call``
  already announces every invocation, so a second inline line per
  proposal would be pure duplication. The count is still tracked so the
  proposed-vs-executed delta is available to
  :meth:`PresenterEventBridge.snapshot` (and to Phase 1's reducer).
- ``permission_decision`` carries no tool linkage in the M2 envelope, so
  the rendered line names the decision and the decider ONLY. Guessing
  which proposal it belongs to would be a false attribution; adding a
  linkage field would be a Phase-1 substrate change.
 
Dependencies
------------
- oc_diagnostics (structured debug logging via _dbg)
- oc_chat.core.event_bus (EventBus / default_bus — the subscription seam)
- oc_chat.core.events (lazy; TurnEvent type hint for the event handlers)
- oc_ui.presenter_base (lazy; Presenter type hint for the rendering target)
"""
from __future__ import annotations
 
from typing import TYPE_CHECKING, Optional
 
from oc_diagnostics import _dbg
 
from oc_chat.core.event_bus import EventBus, Subscription, default_bus
 
if TYPE_CHECKING:
    from oc_chat.core.events import TurnEvent
    from oc_ui.presenter_base import Presenter
 
_dbg("OC_CHAT.MAIN", "presenter_events", "module", 4)
 
 
#: Sub-agent outcomes that are not a clean finish — rendered as a warning.
_BAD_OUTCOMES: frozenset[str] = frozenset({"error", "cancelled", "cap_hit"})
 
#: How much of a sub-agent session id is worth showing (it is a UUID4).
_ID_PREFIX = 8
 
 
class PresenterEventBridge:
    """Renders the approval + sub-agent lanes onto a presenter.
 
    Construct via :func:`attach_presenter_bridge`; the bridge owns its
    :class:`~oc_chat.core.event_bus.Subscription` handles and releases
    them in :meth:`detach`.
    """
 
    def __init__(self, presenter: "Presenter") -> None:
        self._presenter = presenter
        self._subs: list[Subscription] = []
        self._proposed = 0
        self._allowed = 0
        self._denied = 0
        self._subagents_started = 0
        self._subagents_completed = 0
 
    # ── Wiring ────────────────────────────────────────────────────────────
 
    def attach(self, bus: EventBus) -> None:
        """Subscribe to the four kinds this bridge renders (or counts)."""
        func = "attach"
        self._subs = [
            bus.subscribe("tool_call_proposed", self._on_proposed),
            bus.subscribe("permission_decision", self._on_decision),
            bus.subscribe("subagent_started", self._on_subagent_started),
            bus.subscribe("subagent_completed", self._on_subagent_completed),
        ]
        self._bus = bus
        _dbg("OC_CHAT.MAIN", func, f"{len(self._subs)} subscription(s)", 2)
 
    def detach(self) -> None:
        """Release every subscription. Safe to call twice."""
        func = "detach"
        bus = getattr(self, "_bus", None)
        if bus is not None:
            for sub in self._subs:
                bus.unsubscribe(sub)
        self._subs = []
        _dbg("OC_CHAT.MAIN", func, "detached", 2)
 
    # ── Rendering ─────────────────────────────────────────────────────────
 
    def _emit(self, label: str, detail: str, tone: str) -> None:
        """Call ``show_activity`` if the presenter has it; skip otherwise."""
        show = getattr(self._presenter, "show_activity", None)
        if not callable(show):
            return
        show(label, detail, tone=tone)
 
    def _on_proposed(self, event: "TurnEvent") -> None:
        """Count a proposal. Deliberately renders nothing — see Contracts."""
        self._proposed += 1
 
    def _on_decision(self, event: "TurnEvent") -> None:
        """The pending-approval lane: how a gated call was resolved."""
        payload = event.payload
        decision = str(getattr(payload, "decision", "") or "?")
        decided_by = str(getattr(payload, "decided_by", "") or "?")
        if decision == "allow":
            self._allowed += 1
            tone = "ok"
        else:
            self._denied += 1
            tone = "warn"
        # Plan 196 follow-up: M8.1 WU-0a emits a pre-dispatch permission_decision
        # for EVERY tool call, so rendering each one puts an approval line under
        # every call. Routine auto-allows are still COUNTED (the snapshot stays
        # complete) but not shown — this lane exists to surface decisions a human
        # or a policy actually arbitrated. Decisions with no authorization_mode
        # (pre-M8.1 shape, and the post-dispatch relay rows) render as before.
        if str(getattr(payload, "authorization_mode", "") or "") == "auto_allow":
            return
        self._emit("approval", f"{decision} · decided by {decided_by}", tone)
 
    def _on_subagent_started(self, event: "TurnEvent") -> None:
        self._subagents_started += 1
        sid = str(getattr(event.payload, "subagent_session_id", "") or "")
        self._emit("sub-agent", f"started · {sid[:_ID_PREFIX]}", "info")
 
    def _on_subagent_completed(self, event: "TurnEvent") -> None:
        self._subagents_completed += 1
        payload = event.payload
        outcome = str(getattr(payload, "outcome", "") or "?")
        sid = str(getattr(payload, "subagent_session_id", "") or "")
        tone = "warn" if outcome in _BAD_OUTCOMES else "ok"
        self._emit("sub-agent", f"{outcome} · {sid[:_ID_PREFIX]}", tone)
 
    # ── Read side ─────────────────────────────────────────────────────────
 
    def snapshot(self) -> dict:
        """Cumulative counters for the lanes this bridge observes.
 
        The proposed-vs-decided delta is the pending-approval view Phase 1's
        projection/reducer will formalise; here it is a plain counter so the
        information is at least available.
        """
        return {
            "proposed": self._proposed,
            "allowed": self._allowed,
            "denied": self._denied,
            "subagents_started": self._subagents_started,
            "subagents_completed": self._subagents_completed,
        }
 
 
# Module-level singleton guarding re-attach attempts — the same idempotency
# shape as event_bootstrap's ``_active_writer``. Two bridges on one bus would
# render every line twice.
_active_bridge: Optional[PresenterEventBridge] = None
 
 
def attach_presenter_bridge(
    presenter: "Presenter",
    *,
    bus: Optional[EventBus] = None,
) -> PresenterEventBridge:
    """Attach a singleton :class:`PresenterEventBridge` (idempotent).
 
    Returns the active bridge. A second call returns the SAME instance
    without re-subscribing, so the rendered line count stays at one per
    event. *bus* defaults to the process-wide
    :func:`~oc_chat.core.event_bus.default_bus`; tests inject their own.
    """
    func = "attach_presenter_bridge"
    global _active_bridge
    if _active_bridge is not None:
        return _active_bridge
    bridge = PresenterEventBridge(presenter)
    bridge.attach(bus if bus is not None else default_bus())
    _active_bridge = bridge
    _dbg("OC_CHAT.MAIN", func, "presenter bridge attached", 2)
    return bridge
 
 
def detach_presenter_bridge() -> None:
    """Unsubscribe the active bridge and clear the singleton. No-op when unattached."""
    func = "detach_presenter_bridge"
    global _active_bridge
    if _active_bridge is not None:
        _active_bridge.detach()
        _active_bridge = None
        _dbg("OC_CHAT.MAIN", func, "presenter bridge detached", 2)
 
 
__all__ = [
    "PresenterEventBridge",
    "attach_presenter_bridge",
    "detach_presenter_bridge",
]