mostlyright.weather.earnings.segment_bus

mostlyright.weather.earnings.segment_bus

In-process asyncio segment/fact pub-sub bus (Phase 27, 27-10 Task 3).

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

Design (D-27.17, RESEARCH-LIVE-STREAMING §2, codex P2):

• In-process asyncio for v1. A per-call_id set of subscriber asyncio.Queue``s (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 Redis backplane is a RESERVED seam behind the same publish interface — RedisSegmentBus is a stub only (NOT built in v1).
• Backpressure (T-27-39). Each subscriber queue is BOUNDED. On a full queue, the publisher evicts the OLDEST partial item (drop-oldest-partial) and never blocks; a final / fact-delta is NEVER dropped (settlement-adjacent — a dropped final loses a counted mention). If the queue is full of finals with no partial to evict, the queue is grown just enough to admit the final (finals are bounded by the ring-buffer size, so this cannot grow without bound).
• Resume (codex P2). subscribe(call_id, from_seq=N) first replays the ring-buffer finals with stream_seq > N (gap recovery for a reconnect), 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 — NEVER a silent gap. A brand-new subscriber (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 (D-27.9). The item union is transcript-segment | fact-delta ONLY. Publishing raw audio bytes (or any non-Segment/FactDelta object) is rejected — audio NEVER enters the bus.

Module Attributes

BusItem	The bus item union — TEXT/facts only (D-27.9).

Classes

EndOfCall(call_id)	End-of-call control sentinel: the streaming engine finished call_id.
RedisSegmentBus(*args, **kwargs)	RESERVED multi-node seam (D-27.17) — NOT implemented in v1.
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 (v1; Redis is a reserved seam).
SegmentBusProtocol(*args, **kwargs)	The publish/subscribe contract both the in-process + Redis buses satisfy.
SubscriberLagged(earliest_retained_seq)	Terminal marker: this subscriber fell too far behind to buffer safely.

mostlyright.weather.earnings.segment_bus.BusItem = mostlyright.weather.earnings.streaming_transcriber.Segment | mostlyright.weather.earnings.streaming_transcriber.FactDelta

The bus item union — TEXT/facts only (D-27.9). Segment may be partial or final; FactDelta is always final (counted off a final segment).

class mostlyright.weather.earnings.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 27-11 SSE route can emit a terminating end_of_call frame and close the connection cleanly (no dangling generator / leaked subscriber). It is a TEXT/control marker only — carries no audio (D-27.9).

• Parameters: call_id (str)

call_id : str

class mostlyright.weather.earnings.segment_bus.RedisSegmentBus(*args, **kwargs)

Bases: object

RESERVED multi-node seam (D-27.17) — NOT implemented in v1.

A Redis-pub/sub backplane behind the SAME SegmentBusProtocol publish/subscribe interface, so a multi-node deployment can fan segments across processes without changing the streaming engine or the SSE endpoint. Reserved only — constructing it raises NotImplementedError in v1 (the in-process SegmentBus is the shipped implementation).

• Parameters:
  • args (object)
  • kwargs (object)

class mostlyright.weather.earnings.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 (27-04). This is an EXPLICIT signal, never a silent gap (codex P2).

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

earliest_retained_seq : int

from_seq : int

class mostlyright.weather.earnings.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 (v1; Redis is a reserved seam).

• Parameters:
  • subscriber_queue_maxsize (int) – Bounded per-subscriber queue depth (drop-oldest-partial on overflow).
  • 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 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 / non-Segment-or-FactDelta objects (D-27.9). Appends FINAL items (final Segment or any FactDelta) to the per-call ring buffer. Puts to every subscriber queue 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, first replays ring-buffer FINALs with stream_seq > from_seq (gap recovery). If from_seq predates the ring buffer’s earliest retained seq, yields a ResumeIncomplete marker FIRST (reconcile from the ledger — never a silent gap). When from_seq is None, backfills the bounded ring buffer of recent finals. Then drains the live queue.

• 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.weather.earnings.segment_bus.SegmentBusProtocol(*args, **kwargs)

Bases: Protocol

The publish/subscribe contract both the in-process + Redis buses satisfy.

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.weather.earnings.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 — a slow /stream consumer that stopped 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, then reconnects). This bounds memory at the hard cap while NEVER silently dropping a final (a dropped final loses a counted mention — settlement-adjacent). TEXT/control only — carries no audio (D-27.9).

• Parameters: earliest_retained_seq (int)

earliest_retained_seq : int