AiStreamManager

What it is

AiStreamManager is the Main-process active-stream registry and the broker for every stream event. It owns the full life cycle of an AI streaming reply — from sendMessages until the assistant turn finishes persisting — including multicast fan-out, reconnect, abort, steering (queue + yield + continuation), and persistence triggering.

The renderer no longer holds a direct reference to the stream. Closing a window does not abort the stream; it continues on Main and persists normally. When the user returns, attach re-subscribes and the manager replays any chunks that landed in between.

Key: topicId. A topic has at most one active stream at a time; "streaming" is one phase of a topic, and every subscriber on a topic is equal — there is no "owner" window.

Why it exists

v1 ran the stream lifecycle, fan-out, and persistence on the renderer, which produced three structural bug classes:

• Window-bound lifecycle — unmounting the chat (topic switch, window close, route change) cancelled the transport stream, which aborted the upstream request and dropped the in-flight reply.
• No reconnect — reconnectToStream() always returned null, so returning to a topic lost live progress until the row hit the DB.
• Renderer-owned persistence — the DB write lived in the renderer, so a crash/close between stream-end and commit lost the reply.

Goal: move stream lifecycle, multicast fan-out, and persistence to Main; the renderer's only job is rendering chunks. The sections below are the reference for that Main-side design.

Architecture

┌──────────────── Renderer ────────────────────────────────────┐
│                                                              │
│  useChat({ id: topicId, transport: IpcChatTransport })       │
│    ├─ sendMessages   → Ai_Stream_Open  (topicId, trigger, userMessageParts, …)
│    ├─ reconnectToStream → Ai_Stream_Attach ({ topicId })     │
│    └─ abort signal   → Ai_Stream_Abort  ({ topicId })        │
│                                                              │
│  History:           useQuery('/topics/:id/messages')         │
│  Topic-level state: useTopicStreamStatus → shared cache       │
└──────────────────────────────────────────────────────────────┘
                 ↕ IPC (all keyed by topicId)
┌──────────────── Main ────────────────────────────────────────┐
│                                                              │
│  dispatchStreamRequest(manager, subscriber, req)             │
│    │ pick first ChatContextProvider whose canHandle matches  │
│    │ provider.prepareDispatch(subscriber, req, ctx)          │
│    └ manager.send(prepared)                                  │
│                                                              │
│  AiStreamManager                                             │
│  ┌────────────────────────────────────────────────────────┐  │
│  │ activeStreams: Map<topicId, ActiveStream>              │  │
│  │   listeners:  Map<listenerId, StreamListener>          │  │
│  │   executions: Map<modelId, StreamExecution>            │  │
│  │     ├─ abortController / status                        │  │
│  │     └─ buffer (ring) + droppedChunks                   │  │
│  │   lifecycle: StreamLifecycle  (chat or prompt)         │  │
│  └────────────────────────────────────────────────────────┘  │
│         ↓ createAndLaunchExecution → runExecutionLoop        │
│  AiService.streamText(request) → ReadableStream<UIMessageChunk> │
│         ↓ pipeStreamLoop (tees: broadcast + readUIMessageStream) │
│                                                              │
│  terminal → dispatchToListeners → every StreamListener:      │
│    WebContentsListener    → wc.send(Ai_StreamDone)           │
│    PersistenceListener    → PersistenceBackend.persistAssistant
│      • MessageServiceBackend  (SQLite tree)                  │
│      • TemporaryChatBackend   (in-memory)                    │
│      • AgentSessionMessageBackend (agent-session DB)         │
│      • TranslationBackend     (translate row)                │
│    TraceFlushListener    → TraceStorageService.saveSpans(topicId)
│    ChannelAdapterListener → adapter.onStreamComplete         │
│    SseListener            → res.write('[DONE]')              │
└──────────────────────────────────────────────────────────────┘

Pub/sub model

The manager is a broker: one set of producers feeds it, one set of consumers subscribes. The system uses the observer pattern, and splits dispatch into two semantically distinct channels based on payload volume × audience width.

Producers

Consumers

Two channels: targeted listener dispatch vs SharedCache mirror

Channel selection rule

Choose by consumer / producer fanout:

• chunk stream: one execution produces it, only the window rendering that topic needs it → targeted listener dispatch, no irrelevant pushes.
• topic status: one transition, every UI mirror wants it → SharedCache, reuse generic cache sync, no bespoke IPC.

Rules that follow from the channel split

• Ai_Stream_Attach is required. The listener channel requires explicit consumer registration; attach is the entry point and also returns a compact replay to fill the "before I subscribed" gap.
• Bootstrap needs no extra IPC. A new window pulls all shared cache entries via Cache_GetAllShared on mount; every topic.stream.statuses.${topicId} entry comes through without a bespoke snapshot IPC.
• Snapshot vs delta race. Handled by the shared cache sync layer itself — initial pull and Cache_Sync delta share the Main-side source of truth; late arrivals overwrite stale state.
• Grace-period cleanup does NOT clear the SharedCache entry. Terminal values (done / aborted / error) stay so renderer-side consumers (useTopicDbRefreshOnTerminal, useChatWithHistory, awaiting-approval indicators, sidebar badges) can observe them. The fulfilled-badge gate is a read-receipt: the entry's lastCompletedAt (bumped only on done) compared against topic.stream.last_seen_completion.${topicId} (cross-window shared cache, written when the user acknowledges). Memory tier — both reset on app restart.
• PersistenceListener placement. Terminal-only consumer — doesn't need chunk bandwidth → not added via attach; the provider includes it in the listeners array passed to send().
• TraceFlushListener placement. Terminal-only consumer that flushes TraceStorageService.saveSpans(topicId) after a chat / agent turn completes. It belongs with the turn owner (PersistentChatContextProvider or AgentSessionRuntimeService), not inside AiStreamManager and not in trace viewer UI.

File layout

src/main/ai/
├── AiService.ts                       lifecycle service: streamText + non-streaming IPC gateway
└── runtime/aiSdk/
    └── Agent.ts                       single-pass `Agent.stream` wrapper (see Agent Loop)

src/main/ai/streamManager/
├── AiStreamManager.ts                 the registry + execution loop + multicast
├── pipeStreamLoop.ts                  shared chunk-pipe primitive (used by AiStreamManager.runExecutionLoop)
├── buildCompactReplay.ts              attach-time chunk compaction (merge text-delta / reasoning-delta)
├── types.ts                           ActiveStream / StreamExecution / StreamListener / timings
├── index.ts                           barrel
│
├── context/                           per-topicId namespace dispatch
│   ├── ChatContextProvider.ts            interface + PreparedDispatch
│   ├── dispatch.ts                       single manager.send entry; MainContinueConversationRequest
│   ├── PersistentChatContextProvider.ts  uuid topics → SQLite tree
│   ├── TemporaryChatContextProvider.ts   in-memory (TemporaryChatService)
│   ├── AgentChatContextProvider.ts       `agent-session:` → agents DB
│   └── modelResolution.ts                resolveModels / siblingsGroupId
│
├── lifecycle/                         strategy: chat vs ad-hoc prompt
│   ├── StreamLifecycle.ts             interface
│   ├── ChatStreamLifecycle.ts         cross-window broadcast + 30 s grace period + attach
│   └── PromptStreamLifecycle.ts       silent, no attach, immediate eviction
│
├── listeners/
│   ├── WebContentsListener.ts         chunks → renderer windows
│   ├── PersistenceListener.ts         observer protocol + delegates to PersistenceBackend
│   ├── TraceFlushListener.ts          terminal trace-cache flush to local history
│   ├── ChannelAdapterListener.ts      text → Discord / Slack / Feishu
│   └── SseListener.ts                 UIMessageChunk → SSE response (API server)
│
└── persistence/
    ├── PersistenceBackend.ts          strategy interface + runtime-only stats input
    └── backends/
        ├── MessageServiceBackend.ts   finalize a SQLite pending placeholder
        ├── TemporaryChatBackend.ts    append to in-memory topic
        └── TranslationBackend.ts      attach `data-translation` part to a target message

Agent session persistence is implemented under agentSession/persistence because it writes the agent-session domain tables.

StreamListener interface

The manager treats every consumer through one interface; it dispatches each event by calling these methods uniformly:

interface StreamListener {
  readonly id: string
  onChunk(chunk: UIMessageChunk, sourceModelId?: UniqueModelId, anchorMessageId?: string): void
  onDone(result: StreamDoneResult): void | Promise<void>      // { finalMessage?, status: 'success', ... }
  onPaused(result: StreamPausedResult): void | Promise<void>  // { finalMessage?, status: 'paused',  ... }
  onError(result: StreamErrorResult): void | Promise<void>    // { finalMessage?, error, status: 'error', ... }
  isAlive(): boolean
}

All three terminal shapes share the same finalMessage? field — the UIMessage accumulated by readUIMessageStream in the execution loop. Whether the stream ended naturally, was aborted, or errored, it's the same variable, only the stop point differs. Earlier designs called the error-path partial a partialMessage; this turned out to be just a finalMessage that ended early. Unifying the shape means PersistenceBackend needs one persistAssistant method, not separate write paths per status.

Renderer-facing listeners also receive anchorMessageId, the assistant row the execution writes to. This is part of the stream branch identity: sourceModelId distinguishes parallel model executions, while anchorMessageId distinguishes same-model chained turns such as steer continuations.

Built-in implementations

Unified liveness policy

AiStreamManager.dispatchToListeners is the single funnel for terminal events (onDone / onPaused / onError). Per listener it:

• Calls listener.isAlive() before each broadcast — false removes the listener from stream.listeners (cleans up dead consumers).
• Wraps each call in try/catch — one bad listener can't starve the rest.
• Logs by event name + listener id for easy triage.

onChunk keeps a synchronous contract (the execution loop can't await a listener) so it inlines the loop instead of going through dispatchToListeners, but the dead-listener cleanup is the same.

PersistenceListener — strategy pattern

One listener + four backends:

interface PersistenceBackend {
  readonly kind: string   // "sqlite" | "temp" | "agents-db" | "translation"
  persistAssistant(input: {
    finalMessage?: CherryUIMessage
    status: 'success' | 'paused' | 'error'
    modelId?: UniqueModelId
    runtimeStats?: MessageRuntimeStatsInput
  }): Promise<void>
  afterPersist?(finalMessage: CherryUIMessage): Promise<void>
}

Backends expose one write method; the three statuses share its shape. On the error branch, PersistenceListener folds the SerializedError into a trailing data-error part on finalMessage.parts and then calls persistAssistant({ status: 'error' }), so backends never have to know how to encode an error into a UIMessage — they just write.

The listener owns the observer protocol: filter by modelId (multi-model topics have one listener per execution), merge the error part exactly once, swallow exceptions so they don't break downstream dispatch, fire afterPersist only when status === 'success' and finalMessage is present (best-effort). Adding a fifth storage path (e.g. an outbox) is a 60-line backend, no listener boilerplate to copy.

ActiveStream & StreamExecution

interface ActiveStream {
  topicId: string
  executions: Map<UniqueModelId, StreamExecution>   // 1 entry single-model, N multi-model
  listeners: Map<string, StreamListener>            // shared across executions
  // 'pending' on creation; flips to 'streaming' on first chunk; derived
  // from executions on terminal (done / aborted / error /
  // awaiting-approval).
  status: TopicStreamStatus
  isMultiModel: boolean                             // fixed at create; tags onChunk's sourceModelId
  lifecycle: StreamLifecycle                        // chat or prompt strategy
  expiresAt?: number
  cleanupTimer?: ReturnType<typeof setTimeout>
}

interface StreamExecution {
  modelId: UniqueModelId
  anchorMessageId?: string  // placeholder id for submit/regen, anchor id for continue
  abortController: AbortController
  status: 'streaming' | 'done' | 'error' | 'aborted'

  // Per-execution ring buffer for reconnect replay. Hitting
  // `maxBufferChunks` drops the oldest entry and bumps `droppedChunks`.
  // Independent buffers prevent a chatty model from evicting a slower
  // model's replay (a shared buffer would).
  buffer: StreamChunkPayload[]
  droppedChunks: number

  finalMessage?: CherryUIMessage

  // Set the moment a `tool-approval-request` chunk arrives, cleared on
  // response. Read by `resolveTerminalStatus` to surface
  // `awaiting-approval` on the topic.
  awaitingApproval?: boolean

  error?: SerializedError
  siblingsGroupId?: number
  loopPromise: Promise<void>     // awaited by onStop for graceful shutdown

  // Transport-side timings owned by the execution loop — chunk-shape-agnostic.
  timings: TransportTimings

  // OTel root span set as active context around runExecutionLoop so
  // AI SDK spans become children. Created by the context provider.
  rootSpan?: Span
}

interface TransportTimings {
  readonly startedAt: number   // execution loop entry
  completedAt?: number         // execution loop exit (both try and catch paths)
}

Topic-level status is derived from executions, with 'pending' as the initial pre-first-chunk window:

• Created (send() returned) → 'pending'
• Any execution emits its first chunk → 'streaming'
• All terminal, all done → 'done'
• All terminal, all aborted → 'aborted'
• Has error, none streaming → 'error'
• Any execution still has awaitingApproval true on a terminal topic → 'awaiting-approval'

pending → streaming is a one-time transition (first chunk anywhere). The terminal status is derived once when the last execution terminates.

Runtime timing persistence

Ownership (key invariant: persistence cannot write record-owned fields):

AiStreamManager creates and owns the timing collector for an execution. The context provider supplies runtimeTimingSeed when a continuation must resume persisted timing; the manager consumes that seed without loading chat or agent tables itself. The listener is chunk-shape-agnostic and carries only the collector snapshot to persistence:

const parts = finalizeInterruptedParts(finalMessage.parts, status)
const finalMessageForPersistence = { ...finalMessage, parts }
await this.opts.backend.persistAssistant({
  finalMessage: finalMessageForPersistence,
  status,
  modelId,
  runtimeStats: result.runtimeTiming
    ? { runtimeTiming: result.runtimeTiming }
    : undefined
})

New messages write only runtimeTiming; the legacy scalar timeFirstTokenMs / timeCompletionMs / timeThinkingMs fields are read only for historical rows without a runtime timeline. Backends never derive message stats themselves. Data-layer merge helpers accept only MessageRuntimeStatsInput, preserve the record-owned projection, and remove legacy scalar timing once a runtime timeline exists. Temporary chat append reads the current projection internally, and promotion only rebuilds it; no persistence path creates or repairs a usage record. See AI Usage Records.

Public API

class AiStreamManager {
  // Lifecycle container invokes with no args (DEFAULT_CONFIG); tests can
  // override `gracePeriodMs`, `backgroundMode`, `maxBufferChunks`.
  constructor(config?: Partial<AiStreamManagerConfig>)

  readonly chatLifecycle: StreamLifecycle

  // ── Single dispatch entry ─────────────────────────────────────────
  // Live topic → inject (upsert listeners onto the running stream, models
  // ignored — reached by chat steers and agent-session follow-ups whose user
  // row was already persisted/enqueued by their provider). Otherwise → start
  // (evict any grace-period stream, launch one execution per `models` entry).
  // Multi-model is detected from `models.length > 1`.
  send(input: SendInput): SendResult

  // ── Ad-hoc prompt stream (translate / topic-naming / model probes)
  // Bypasses the chat dispatcher; uses promptStreamLifecycle (silent, no
  // attach, immediate eviction).
  streamPrompt(input: {
    streamId: string                                       // doubles as topicId
    uniqueModelId: UniqueModelId
    prompt?: string
    messages?: CherryUIMessage[]
    listener: StreamListener | StreamListener[]
  }): SendResult

  // ── Subscription management ───────────────────────────────────────
  attach(sender: WebContents, req: { topicId }): AiStreamAttachResponse
  detach(sender: WebContents, req: { topicId }): void
  addListener(topicId: string, listener: StreamListener): boolean
  removeListener(topicId: string, listenerId: string): void

  // ── Control ───────────────────────────────────────────────────────
  abort(topicId: string, reason: string): void
  hasLiveStream(topicId: string): boolean
  // Queue a steer user row persisted while a turn was live; the running turn
  // yields and `onExecutionDone` chains a `steer-continuation` to answer it.
  enqueuePendingSteer(topicId: string, userMessageId: string): void
  hasPendingSteer(topicId: string): boolean

  // ── Execution-loop callbacks (driven internally; public for tests) ─
  onChunk(topicId, modelId, chunk): void
  onExecutionDone(topicId, modelId): Promise<void>
  onExecutionPaused(topicId, modelId): Promise<void>
  onExecutionError(topicId, modelId, error): Promise<void>

  // ── Inspection (read-only snapshot) ───────────────────────────────
  inspect(topicId: string): TopicSnapshot | undefined

  // ── Write quiesce (backup restore) — see the dedicated section ────
  get isWriteQuiesced(): boolean
  pause(reason?: string): Disposable
  drainInFlight(opts: { timeoutMs: number }): Promise<{ stragglerIds: string[] }>
  listActiveWork(): Array<{ id: string; summary: string }>
}

send contract

interface SendInput {
  topicId: string
  models: ReadonlyArray<{ modelId: UniqueModelId; request: AiStreamRequest; rootSpan?: Span }>
  listeners: StreamListener[]
  siblingsGroupId?: number
  lifecycle?: StreamLifecycle        // omit → chatLifecycle; streamPrompt passes promptStreamLifecycle
}

interface SendResult {
  mode: 'started' | 'injected'
  executionIds: UniqueModelId[]      // started → fresh ids; injected → already running
}

• injected: topic has a live stream (pending or streaming) → models is ignored and listeners upsert by id; no models are launched. Reached by (a) a chat steer — the provider already persisted the steer user row and dispatch enqueued it on pendingSteers; and (b) an agent-session follow-up already enqueued on the session's pendingTurns. An empty-models send with no live stream is likewise a no-op (the row is already enqueued) — send() never throws on empty models.
• started: topic is idle or grace-period (terminal) → any leftover grace-period stream is evicted, a new ActiveStream is created with isMultiModel = models.length > 1, one execution launched per model.

isMultiModel is not an input — it's derived from models.length.

Execution loop — runExecutionLoop + pipeStreamLoop

Each execution runs an independent loop that bridges "the single ReadableStream from AI SDK" to "what the manager has to do": broadcast to listeners, buffer for reconnect, and accumulate a persistable finalMessage.

Step 1 — get the raw chunk stream.

const stream: ReadableStream<UIMessageChunk> = await aiService.streamText({
  ...request,
  requestOptions: { ...request.requestOptions, signal }
})

streamText returns AI SDK's raw chunk stream. signal comes from StreamExecution.abortController; abort() triggers it.

Step 2 — wrap with withIdleTimeout. Resets per chunk; on idle timeout it aborts exec.abortController, which the upstream request is already wired to.

Step 3 — pipeStreamLoop tees the chunk stream.

pipeStreamLoop is the shared chunk-pipe primitive (the one AiStreamManager.runExecutionLoop uses). It tee()s the stream into two independent branches:

The accumulator reader is not cancelled directly on abort — Agent.stream honours the same signal upstream and propagates done through tee(), so the accumulator drains naturally. Cancelling the accumulator reader directly would race AI SDK's internal controller.close() and produce an ERR_INVALID_STATE unhandledRejection.

Step 4 — terminal dispatch.

Write quiesce (pause / drainInFlight)

Serves backup restore (#16849, same contract as JobManager's — see job overview): after the restore snapshot is staged, any main-side write to the old live DB fails the fingerprint re-check and wastes the whole restore attempt. Three AI-side writers carry the contract — AiStreamManager, AgentSessionRuntimeService, and channel intake (ChannelManager → ChannelMessageHandler) — each exposing pause(reason?): Disposable + drainInFlight({ timeoutMs }) → { stragglerIds } (empty = clean) + an advisory read-only listActiveWork().

Orchestration order (grandfather-free, per #16850) — the channel step MUST fully complete before the AI writers are paused:

1. ChannelManager.pause() — gate new adapter messages/commands and immediately flush the buffered debounce batches (never cancel: adapters ack at the transport layer on receipt, so the in-memory buffer is the only copy).
2. await ChannelManager.drainInFlight() — flush only schedules each batch's admission (its processIncoming runs on the per-chat queue microtask), so pause() returning does NOT mean the batches admitted. This await is the flush-to-admission barrier: it resolves once every flushed batch passed agent-turn admission (not turn completion — the flushed turns land in the AI in-flight set, covered from there by the AI drains).
3. only then pause AI + JobManager (any order) → joint drain → verdict → snapshot.

Why the barrier is load-bearing: if the AI writers are paused while a flushed batch is still between flush and startAgentSessionRun, the batch hits the closed AI gate and is rejected. Because the adapter already ACKed it, it cannot be recovered by aborting the restore. The channel-drain-before-AI-pause ordering is therefore a correctness precondition, not merely a performance optimization. On any drain timeout the orchestrator aborts the attempt (dispose all holds); the happy path never disposes — the holds stand until relaunch, and a lost hold fails closed.

AiStreamManager specifics:

AgentSessionRuntimeService gates its two autonomous turn starters (startNextTurn / startContinuationTurn) before they consume queue/roll state or write the assistant placeholder — suppressed starts stay queued (isSessionBusy holds) and are re-kicked on release; its drain awaits inFlightTurnStarts (a launch admitted pre-pause through its placeholder write + startRuntimeTurn handoff). See agent-session-runtime.md.

Lifecycle strategy — chat vs prompt

The manager stays policy-free. Behaviour that differs between chat streams and one-shot ad-hoc prompts (translate, topic-naming, model probes) lives in StreamLifecycle:

interface StreamLifecycle {
  readonly name: string
  onCreated(stream): void                         // freshly registered
  onPromotedToStreaming(stream): void             // first chunk
  onTerminal(stream): void                        // every isTopicDone
  canAttach(stream): boolean                      // gate for `attach`
  cleanup(stream, evict: () => void): void        // when to remove from activeStreams
}

send() defaults to chatLifecycle; streamPrompt() passes promptStreamLifecycle.

Multi-model

User mentions multiple models for one turn:

User: "Explain quantum mechanics" @gpt-4o @claude-sonnet
                                ↓
PersistentChatContextProvider.prepareDispatch
    ├─ persist user message (tree node)
    ├─ resolveModels → [gpt-4o, claude-sonnet]
    ├─ siblingsGroupId = (monotonic counter)
    ├─ create one pending assistant placeholder per model (SQLite)
    ├─ build listeners: subscriber + 2 PersistenceListener (one per backend)
    ├─ build models: 2 × { modelId, request, rootSpan }
    └─ return PreparedDispatch

dispatchStreamRequest → manager.send({ models, listeners, siblingsGroupId })
                          │
                          ├─ create ActiveStream (isMultiModel = true, 2 executions)
                          ├─ launch one execution loop per model, each with its own
                          │  ring buffer
                          └─ return { mode: 'started', executionIds: [gpt-4o, claude-sonnet] }

Steering

Steering a chat turn is enqueue + yield + chain, not abort-and-restart and not mid-turn injection. When a new Ai_Stream_Open arrives for a chat topic that is still streaming:

1. PersistentChatContextProvider (its hasLiveStream branch) persists the steer message as a normal user row and returns an enqueue-only PreparedDispatch — no models, pendingSteerUserMessageId set.
2. dispatchStreamRequest calls manager.enqueuePendingSteer(topicId, id), pushing the row onto the topic's pendingSteers FIFO, then send() — which, seeing the live stream, just upserts the subscriber (inject).
3. The running turn's steerYield stop condition (OR'd into stopWhen) sees hasPendingSteer and stops the turn cleanly at the next step boundary (persisted as success, not paused).
4. onExecutionDone sees the queued steer and, instead of finalizing the topic, chains a steer-continuation dispatch (startNextChatTurn) that answers the head of the queue, carrying the prior turn's renderer listeners forward. The FIFO drains one continuation per completed turn.

Drop-on-abort: a steer chains only after a clean done. If the turn is aborted (Stop) or errors, the queue is dropped and its persisted user rows stay in history as dangling messages the user can resend (onExecutionPaused / onExecutionError clear pendingSteers; a late steer landing after a non-clean terminal is dropped by enqueuePendingSteer). A steer queued while a turn ends awaiting-approval does not chain until the approval's continue-conversation turn completes — chaining earlier would let the approval response be swallowed by the inject branch. If the continuation itself fails to launch, the topic is driven to a terminal error rather than sticking at streaming.

Agent-session topics use a parallel, queue-based mechanism — never an interrupt. A live follow-up is steered into the running turn via connection.redirect() (no abort); if there is no live turn, or the steer is never injected, it is enqueued on the session's pendingTurns for the next turn. send() only upserts the new subscriber. See Agent Session Runtime → Live follow-up.

End-to-end flows

One row per flow. The two with dedicated docs are cross-linked rather than duplicated; the rest are stream-manager-specific.

Topic status needs no attach. Observers that only care "is this topic live?" (sidebar loading indicators, topic-list status dots) don't register a WebContentsListener. Every status transition writes the SharedCache key topic.stream.statuses.${topicId}; observers read it via useSharedCache directly. Ai_Stream_Attach is only needed when a window wants live chunks.

Channel / Agent listener composition

Channel adapters and the agent scheduler call AiStreamManager.send directly inside Main — no IPC. The scenario differences are entirely in the listener composition:

IPC contract

Request channels (Renderer → Main)

Topic status snapshots need no dedicated IPC: a new window pulls every topic.stream.statuses.${topicId} entry via Cache_GetAllShared on mount, and useSharedCache subscribes by topicId.

Push channels (Main → Renderer)

Topic-level status transitions are NOT a bespoke IPC — they live in the SharedCache key topic.stream.statuses.${topicId} (Main setShared → built-in Cache_Sync broadcast). The entry shape is TopicStatusSnapshotEntry:

{
  status: 'pending' | 'streaming' | 'done' | 'aborted' | 'awaiting-approval' | 'error'
  activeExecutions: ActiveExecution[]         // execs currently `streaming`
  awaitingApprovalAnchors: ActiveExecution[]  // execs with awaitingApproval = true
  lastCompletedAt?: number                    // bumped only on `done`; the fulfilled-badge read-receipt gate
}

pending doubles as the "new stream just created" signal — the old Ai_StreamStarted IPC is gone. Grace-period cleanup does NOT clear the entry — terminal values (done / aborted / error) stay so renderer consumers (DB-refresh trigger, awaiting-approval indicators, sidebar badges) can observe them. The badge "should I show this?" gate is a read-receipt: entry.lastCompletedAt (authoritative, bumped only on done) compared against topic.stream.last_seen_completion.${topicId} (cross-window shared cache, written by the renderer when the user acknowledges).

All traffic is keyed by topicId; multi-model uses executionId to demux chunks per model.

Topic status vs message status. Don't conflate:

• Topic stream status (SharedCache topic.stream.statuses.${topicId}): one entry per topic, source of truth is ActiveStream.status, valid only while the ActiveStream exists (+ grace period).
• Assistant message status (AssistantMessageStatus: PENDING / PROCESSING / SUCCESS / ERROR): one per assistant message, persisted in SQLite, written by PersistenceListener.onDone/onError. In multi-model, a single topic-level transition corresponds to N separate message rows.

ChatContextProvider — per-topicId namespace dispatch

Ai_Stream_Open is handled in Main by dispatchStreamRequest (context/dispatch.ts):

dispatchStreamRequest(manager, subscriber, req)
  → provider = providers.find(p => p.canHandle(req.topicId))
  → prepared = await provider.prepareDispatch(subscriber, req, { hasLiveStream })
  → result   = manager.send(prepared)        // ← the only manager.send call
  → return { mode, executionIds?, userMessageId?, placeholderIds? }

Providers only "prepare" — they never call manager.send directly. Two benefits:

• Provider unit tests assert on PreparedDispatch shape without mocking the manager.
• The restart / start / multi-model fan-out routing lives in exactly one place.

Provider interface

interface ChatContextProvider {
  readonly name: string
  canHandle(topicId: string): boolean
  prepareDispatch(
    subscriber: StreamListener,
    req: MainDispatchRequest,
    ctx: { hasLiveStream: boolean }
  ): Promise<PreparedDispatch>
}

interface PreparedDispatch {
  topicId: string
  models: ReadonlyArray<{ modelId: UniqueModelId; request: AiStreamRequest; rootSpan?: Span }>
  listeners: StreamListener[]   // subscriber + per-execution PersistenceListener(s)
  userMessageId?: string
  pendingSteerUserMessageId?: string   // persistent steer branch only; marks the dispatch enqueue-only
  reservedMessages?: CherryUIMessage[] // user/assistant skeletons created for this dispatch
  siblingsGroupId?: number
  isMultiModel: boolean
  lifecycle?: StreamLifecycle
}

// dispatch.ts also accepts two Main-internal variants synthesised internally —
// `continue-conversation` (tool-approval IPC handler) and `steer-continuation`
// (chat steer drain) — neither exposed over the renderer ↔ main contract.
type MainDispatchRequest = AiStreamOpenRequest | MainContinueConversationRequest | MainSteerContinuationRequest

Built-in providers

Order: Agent → Temporary → Persistent (first canHandle === true wins).

Persistence path comparison

One PersistenceListener across all topic kinds

Persistent / Temporary / Agent / Translation all share the same PersistenceListener class — only the injected PersistenceBackend differs. The observer protocol (modelId filter, error part folding, skip-when-no-finalMessage, swallow errors) is implemented once.

AiService integration

AiService is a lifecycle service:

• Streaming. streamText(request) returns Promise<ReadableStream<UIMessageChunk>>, consumed by AiStreamManager.runExecutionLoop.
• Non-streaming IPC gateway. generateText / checkModel / embedMany / generateImage / listModels, registered as IPC handlers in onInit.

AiStreamManager calls await application.get('AiService').streamText(...). Pre-stream errors (provider / model resolution, agent param build) reject the returned Promise; mid-stream errors come through the returned stream's error path — the two error paths never overlap.

Grace period & reconnect

After a stream terminates, ActiveStream stays in memory for 30 s (config.gracePeriodMs). During that window a returning user can attach and pull finalMessage without a DB read. After expiry the entry is evicted; subsequent attach returns not-found and the renderer reads from the DB through useQuery (PersistenceListener has already written by then).

If the user stops and immediately retries on the same topic, send takes the start branch: evictStream first clears the grace-period remnant (cancels the cleanup timer and drops the entry from activeStreams), then the new stream is created — the old never blocks the new.

Edge case cheat sheet

Design notes

Testing strategy

• Manager tests. new AiStreamManager({ maxBufferChunks: 3 }) via the optional config arg; state assertions go through mgr.inspect(topicId); listener upsert / abort / backgroundMode are tested via behaviour (drive a chunk, assert which listeners received it).
• Provider tests. Assert on the returned PreparedDispatch shape directly — no manager mock.
• PersistenceListener tests. TemporaryChatBackend as the test vehicle covers the observer protocol once for every backend.
• All internal state has a public inspection API; production and tests share the same contract.