mostlyright.finance.transcripts.segment_bus

In-process asyncio segment/fact pub-sub bus.

SegmentBus is the per-call topic the streaming STT engine publishes Segment and FactDelta items onto, and the SSE endpoint fans out to browser clients over text/event-stream.

Design:

• In-process asyncio for v1. A per-call_id set of subscriber buffers (fan-out) plus a per-call_id bounded ring buffer (collections.deque(maxlen=K)) of the last K final events, for short-gap Last-Event-ID resume. A cross-process (Redis/Memorystore) backplane behind the same publish interface would serve a multi-node deployment; it is not built in v1.
• Backpressure. Each subscriber buffer is bounded. On a full buffer the publisher evicts the oldest partial item and never blocks; a final segment or fact delta is never dropped, because dropping one loses a counted mention. If the buffer is full of finals with no partial to evict, it grows just enough to admit the final — finals are bounded by the ring-buffer size, so this cannot grow without bound.
• Resume. subscribe(call_id, from_seq=N) first replays the ring-buffer finals with stream_seq > N, plus any fact delta at exactly stream_seq == N, which shares its parent final segment’s seq and is published after it, so a strict cutoff would silently drop a counted mention (see _replay_after()). That covers the gap from a reconnect; the subscriber then goes live. If from_seq predates the ring buffer’s earliest retained seq, the subscriber first yields a ResumeIncomplete marker so the consumer reconciles from the authoritative ledger rather than gapping silently. A brand-new subscriber with no from_seq gets the bounded ring-buffer backfill of recent finals, then live items; full history is the ledger’s job, not the bus’s.
• Audio gate. The item union is transcript segment or fact delta only. Publishing raw audio bytes, or any other object, is rejected: audio never enters the bus.

Module Attributes

Attribute	Description
BusItem	text and facts only.

Classes

Class	Description
EndOfCall(call_id)	End-of-call control sentinel: the streaming engine finished call_id.
ResumeIncomplete(from_seq, earliest_retained_seq)	Resume marker: from_seq predates the ring buffer’s earliest retained seq.
SegmentBus(*[, subscriber_queue_maxsize, …])	In-process asyncio per-call pub/sub bus; single process only.
SegmentBusProtocol(*args, **kwargs)	The publish/subscribe contract the in-process bus satisfies.
SubscriberLagged(earliest_retained_seq)	Terminal marker: this subscriber fell too far behind to buffer safely.

mostlyright.finance.transcripts.segment_bus.BusItem

text and facts only. Segment may be partial or final; FactDelta is always final, since it is counted off a final segment.

• Type: The bus item union

class mostlyright.finance.transcripts.segment_bus.EndOfCall(call_id)

Bases: object

End-of-call control sentinel: the streaming engine finished call_id.

Published by SegmentBus.close() and delivered last to every live subscriber, so the SSE route can emit a terminating end_of_call frame and close the connection cleanly, leaving no dangling generator or leaked subscriber. It is a text/control marker and carries no audio.

• Parameters: call_id (str)

call_id: str

class mostlyright.finance.transcripts.segment_bus.ResumeIncomplete(from_seq, earliest_retained_seq)

Bases: object

Resume marker: from_seq predates the ring buffer’s earliest retained seq.

Yielded first to a resubscribing consumer whose requested from_seq is older than anything the bounded ring buffer still holds. The gap between from_seq and earliest_retained_seq cannot be replayed from the bus, so the consumer must reconcile from the authoritative post-call ledger. This is an explicit signal, never a silent gap.

• Parameters:
  • from_seq (int)
  • earliest_retained_seq (int)

earliest_retained_seq: int

from_seq: int

class mostlyright.finance.transcripts.segment_bus.SegmentBus(, subscriber_queue_maxsize=256, ring_buffer_size=128, subscriber_hard_maxsize=None)

Bases: object

In-process asyncio per-call pub/sub bus; single process only.

• Parameters:
  • subscriber_queue_maxsize (int) – Bounded per-subscriber buffer depth; on overflow the oldest partial is dropped.
  • ring_buffer_size (int) – Per-call ring buffer depth of the last K final events (resume backfill).
  • subscriber_hard_maxsize (int | None)

async close(call_id)

Signal end-of-call: deliver an EndOfCall marker to subscribers.

Each live subscriber receives the marker last, after any queued items; the SSE route yields it as a terminating end_of_call frame and then closes cleanly. Idempotent: closing a call with no subscribers is a no-op.

• Return type: None
• Parameters: call_id (str)

is_closed(call_id)

Whether call_id has emitted end-of-call (a finished call).

• Return type: bool
• Parameters: call_id (str)

async publish(call_id, item)

Publish item to call_id — fan-out + ring-buffer the finals.

Rejects audio bytes and any object that is not a Segment or FactDelta. Appends final items (a final Segment or any FactDelta) to the per-call ring buffer, then offers to every subscriber buffer with drop-oldest-partial backpressure. The publisher never blocks.

• Return type: None
• Parameters:
  • call_id (str)
  • item (object)

subscribe(call_id, from_seq=None)

Subscribe to call_id; return an async generator of items.

When from_seq is given, this first replays ring-buffer finals with stream_seq > from_seq, plus any fact delta at from_seq, which shares its parent segment’s seq and trails it on the wire (see _replay_after()). If from_seq predates the ring buffer’s earliest retained seq, a ResumeIncomplete marker is yielded first, telling the consumer to reconcile from the ledger rather than gapping silently. When from_seq is None, it backfills the bounded ring buffer of recent finals. It then drains the live buffer.

• Return type: AsyncIterator[Segment | FactDelta | ResumeIncomplete | EndOfCall]
• Parameters:
  • call_id (str)
  • from_seq (int | None)

subscriber_count(call_id)

Number of live subscribers on call_id (0 after clean teardown).

• Return type: int
• Parameters: call_id (str)

class mostlyright.finance.transcripts.segment_bus.SegmentBusProtocol(*args, **kwargs)

Bases: Protocol

The publish/subscribe contract the in-process bus satisfies.

A future cross-process (Redis/Memorystore) backplane would implement this same interface, so a multi-node deployment can fan segments across processes without changing the streaming engine or the SSE endpoint.

async publish(call_id, item)

• Return type: None
• Parameters:
  • call_id (str)
  • item (object)

subscribe(call_id, from_seq=None)

• Return type: AsyncIterator[Segment | FactDelta | ResumeIncomplete | EndOfCall]
• Parameters:
  • call_id (str)
  • from_seq (int | None)

class mostlyright.finance.transcripts.segment_bus.SubscriberLagged(earliest_retained_seq)

Bases: object

Terminal marker: this subscriber fell too far behind to buffer safely.

Emitted once when a subscriber’s buffer would exceed the hard ceiling with undroppable final items, which happens when a /stream consumer stops reading. It terminates that subscriber’s stream; the SSE route frames it as resume_incomplete so the client reconciles the gap from the authoritative post-call ledger and then reconnects. This bounds memory at the hard cap without silently dropping a final, which would lose a counted mention. It is a text/control marker and carries no audio.

• Parameters: earliest_retained_seq (int)

earliest_retained_seq: int