Module: Protege::IntrospectionBroadcaster

Extended by:
IntrospectionBroadcaster
Included in:
IntrospectionBroadcaster
Defined in:
app/subscribers/protege/introspection_broadcaster.rb

Overview

Drives the console's introspection panel from inference events — an app-level subscriber for the console experience, not part of the engine's inference core. A peer of Subscribers::HookDispatcher and Subscribers::Tracing, installed once at boot.

It subscribes to Protege events and translates streaming + tool-call notifications into Turbo Stream broadcasts, building the panel's live activity feed: one chronological stream where the model's reasoning tokens and its tool calls are interleaved in the order they happen. Installed via the install! contract (see the engine initializer) and holds process-local state for the lifetime of the process; reset! detaches its subscriptions and clears that state.

What each event does (all broadcast to the thread's Turbo channel, target dom_id(thread, :activity) — the feed container in _introspection.html.slim):

  • InferenceChunkEvent — stream the token into the run's current reasoning block, opening a fresh block after each tool call so reasoning is grouped into spaced paragraphs. Interleaving is automatic: every append lands at the feed's end, so reasoning blocks and cards stay in chronological order.
  • ToolCallStartedEvent — append a pending tool card the instant the model requests the call.
  • ToolCallCompletedEvent / ToolCallFailedEvent — replace that card in place with the outcome (a response preview, or the error message).

Routing. Only InferenceChunkEvent carries the thread; the tool-call events carry just the correlation_id that Protege::Event.emit injects into every event. So the broadcaster keeps a correlation_id → thread map, populated from InferenceStartedEvent (which carries the message, hence its thread) and cleared on +InferenceCompletedEvent+/+InferenceFailedEvent+. This is the single stream- routing convention — no emit_* signature carries the thread just for the panel. @mutex guards the map because callbacks may run concurrently across threads. Runs with no thread (scheduled +ResponsibilityRun+s) never populate the map, so their tool events broadcast nothing.

The feed is live-only by design: it is not rehydrated from persistence, so a page reload shows an empty feed until the next run streams in.

Instance Method Summary collapse

Instance Method Details

#install! ⇒ void

This method returns an undefined value.

Wire the event listeners — the subscriber's install! contract. Idempotent: retains the subscription handles and no-ops until #reset! runs, so a repeat call never double-subscribes. Called once at boot from the engine's after_initialize.



59
60
61
62
63
64
65
66
67
68
 
# File 'app/subscribers/protege/introspection_broadcaster.rb', line 59

def install!
  return unless @handles.empty?

  @handles = [
    install_start_listener,
    install_chunk_listener,
    *install_tool_listeners,
    *install_completion_listeners
  ]
end

#reset! ⇒ void

This method returns an undefined value.

Detach every subscription and drop all per-run state — the subscriber's reset! contract, for boot re-wiring and tests.



74
75
76
77
78
79
80
81
82
 
# File 'app/subscribers/protege/introspection_broadcaster.rb', line 74

def reset!
  @handles.each { |handle| ActiveSupport::Notifications.unsubscribe(handle) }
  @handles = []
  @mutex.synchronize do
    @threads.clear
    @reasoning.clear
    @reasoning_seq.clear
  end
end