feat(agent): emit live assistant stream frames

packages/api/session-controller/src/client/sessions/assistant-stream.ts

@@ -0,0 +1,176 @@
+/** Web presentation fold joining durable v1 events with transient assistant frames. */
+
+import type {
+  SessionAssistantStreamBaseline,
+  SessionAssistantStreamFrame,
+} from '../../types.ts'
+import type {
+  SessionEventLikeEntry,
+  SessionLiveEventEntry,
+} from '../contract/events.ts'
+
+interface ActiveAttempt {
+  readonly startedTime: number
+  readonly turn: number
+  readonly step: number
+  readonly legacyChunkSeqs: Set<number>
+  nextIndex: number
+}
+
+/** One Web publication decision from the assistant stream fold. */
+export type ClientAssistantStreamResult =
+  | { readonly type: 'publish'; readonly entry: SessionLiveEventEntry }
+  | { readonly type: 'rebaseline' }
+  | undefined
+
+function positionKey(turn: number, step: number): string {
+  return `${String(turn)}:${String(step)}`
+}
+
+function sameSeqs(left: readonly number[], right: readonly number[]): boolean {
+  return left.length === right.length && left.every((seq, index) => seq === right[index])
+}
+
+/**
+ * Keeps transient Assistant presentation behind one small interface. Durable
+ * chunks and final messages publish only at their matching live frame.
+ */
+export class ClientAssistantStream {
+  private readonly attempts = new Map<string, ActiveAttempt>()
+  private readonly pendingChunks = new Map<number, SessionLiveEventEntry>()
+  private readonly pendingMessages = new Map<string, SessionLiveEventEntry>()
+  private publishedSeqs = new Set<number>()
+
+  /**
+   * Replace the durable Web window and adopt an optional reconnect baseline.
+   * @param entries - complete event window from the journal replacement.
+   * @param baseline - active process-local attempts for a follow opening.
+   * @returns the same durable window; baseline seqs suppress later duplicate live appends.
+   */
+  replace(
+    entries: readonly SessionEventLikeEntry[],
+    baseline?: SessionAssistantStreamBaseline,
+  ): readonly SessionEventLikeEntry[] {
+    this.pendingChunks.clear()
+    this.pendingMessages.clear()
+    this.attempts.clear()
+    if (baseline !== undefined) {
+      for (const attempt of baseline.attempts) {
+        this.attempts.set(String(attempt.attemptId), {
+          startedTime: attempt.startedTime,
+          turn: attempt.turn,
+          step: attempt.step,
+          legacyChunkSeqs: new Set(attempt.legacyChunkSeqs),
+          nextIndex: attempt.chunks.length,
+        })
+      }
+    }
+    const visible = entries.filter((entry) => {
+      if (entry.type !== 'event' || entry.event.type !== 'assistant/message'
+        || entry.event.surfaceOp !== 'append') return true
+      const attempt = this.attemptForSettlement(entry.event)
+      if (attempt === undefined) return true
+      this.pendingMessages.set(positionKey(attempt.turn, attempt.step), entry)
+      return false
+    })
+    this.publishedSeqs = new Set(visible.map(entry => entry.event.seq))
+    return visible
+  }
+
+  /**
+   * Stage one durable tail event when an active attempt owns its publication.
+   * @param entry - next cursor-validated durable event.
+   * @returns the entry for immediate publication, or undefined while staged.
+   */
+  acceptDurable(entry: SessionLiveEventEntry): ClientAssistantStreamResult {
+    const event = entry.event
+    if (event.type === 'assistant/chunk') {
+      const attempt = this.attemptFor(event.data.turn, event.data.step)
+      if (attempt === undefined) return this.publish(entry)
+      this.pendingChunks.set(event.seq, entry)
+      return undefined
+    }
+    if (event.type === 'assistant/message') {
+      if (event.surfaceOp !== 'append') return this.publish(entry)
+      const attempt = this.attemptFor(event.data.turn, event.data.step)
+      if (attempt === undefined) return this.publish(entry)
+      this.pendingMessages.set(positionKey(event.data.turn, event.data.step), entry)
+      return undefined
+    }
+    return this.publish(entry)
+  }
+
+  /**
+   * Fold one validated transient frame and release its matching durable event.
+   * @param frame - next dense process-local frame.
+   * @returns one durable event whose Web publication commits at this frame.
+   */
+  acceptFrame(frame: SessionAssistantStreamFrame): ClientAssistantStreamResult {
+    switch (frame.type) {
+      case 'start':
+        this.attempts.set(String(frame.attemptId), {
+          startedTime: frame.startedTime,
+          turn: frame.turn,
+          step: frame.step,
+          legacyChunkSeqs: new Set(),
+          nextIndex: 0,
+        })
+        return undefined
+      case 'chunk': {
+        const attempt = this.attempts.get(String(frame.attemptId))
+        if (attempt === undefined || frame.index !== attempt.nextIndex) return { type: 'rebaseline' }
+        attempt.nextIndex += 1
+        attempt.legacyChunkSeqs.add(frame.legacyChunkSeq)
+        if (this.publishedSeqs.has(frame.legacyChunkSeq)) return undefined
+        const entry = this.pendingChunks.get(frame.legacyChunkSeq)
+        if (entry === undefined) return { type: 'rebaseline' }
+        this.pendingChunks.delete(frame.legacyChunkSeq)
+        return this.publish(entry)
+      }
+      case 'end': {
+        const attempt = this.attempts.get(String(frame.attemptId))
+        this.attempts.delete(String(frame.attemptId))
+        if (attempt === undefined) return { type: 'rebaseline' }
+        if (frame.index !== attempt.nextIndex) return { type: 'rebaseline' }
+        if (!sameSeqs([...attempt.legacyChunkSeqs], frame.legacyChunkSeqs)) {
+          return { type: 'rebaseline' }
+        }
+        const key = positionKey(attempt.turn, attempt.step)
+        const entry = this.pendingMessages.get(key)
+        if (entry === undefined) {
+          return frame.outcome === 'aborted' ? undefined : { type: 'rebaseline' }
+        }
+        if (entry.event.type !== 'assistant/message') return { type: 'rebaseline' }
+        const sourceEventSeqs = entry.event.sourceEventSeqs
+        if (sourceEventSeqs === undefined || !sameSeqs(sourceEventSeqs, frame.legacyChunkSeqs)) {
+          return { type: 'rebaseline' }
+        }
+        this.pendingMessages.delete(key)
+        return this.publish(entry)
+      }
+    }
+  }
+
+  private attemptFor(turn: number, step: number): ActiveAttempt | undefined {
+    return [...this.attempts.values()].find(attempt => (
+      attempt.turn === turn && attempt.step === step
+    ))
+  }
+
+  private attemptForSettlement(
+    event: Extract<SessionLiveEventEntry['event'], { type: 'assistant/message' }>,
+  ): ActiveAttempt | undefined {
+    const sourceEventSeqs = event.sourceEventSeqs
+    if (sourceEventSeqs === undefined) return undefined
+    return [...this.attempts.values()].find(attempt => (
+      attempt.turn === event.data.turn
+      && attempt.step === event.data.step
+      && sameSeqs([...attempt.legacyChunkSeqs], sourceEventSeqs)
+    ))
+  }
+
+  private publish(entry: SessionLiveEventEntry): ClientAssistantStreamResult {
+    this.publishedSeqs.add(entry.event.seq)
+    return { type: 'publish', entry }
+  }
+}

packages/api/session-controller/src/client/sessions/session.ts

@@ -12,6 +12,7 @@ import type {
   PromptContentPart,
   QueueAction,
   SessionAddress,
+  SessionAssistantStreamBaseline,
   SessionControlFrame,
   SessionProjectionBaseline,
   SessionQueuedItem,
@@ -35,6 +36,10 @@ import { ProjectionValueStore } from './projection-store.ts'
 import type { ProjectionsBaseline } from './projection-store.ts'
 import { resolvedClientTimeZone } from '../time-zone.ts'
 import { SessionQueueMirror } from './queue-mirror.ts'
+import {
+  ClientAssistantStream,
+  type ClientAssistantStreamResult,
+} from './assistant-stream.ts'
 
 function projectionsBaseline(value: SessionProjectionBaseline): ProjectionsBaseline {
   return {
@@ -95,6 +100,7 @@ export class Session implements SessionFace {
   private jumpPromise: Promise<void> | null = null
   /** Authoritative stream-only inbox snapshot; pending work never hits history. */
   private readonly queueMirror = new SessionQueueMirror()
+  private readonly assistantStream = new ClientAssistantStream()
   private running = false
   private address: SubagentAddress | undefined
   private parentAvailable: boolean | undefined
@@ -620,27 +626,50 @@ export class Session implements SessionFace {
           change.entries,
           change.hasMore,
           change.page.projections === undefined ? undefined : projectionsBaseline(change.page.projections),
+          change.page.assistantStream,
         )
         return
       case 'prepend':
         this.prependWindow(change.entries, change.hasMore)
         return
       case 'append':
-        if (this.appendLive(change.entry)) this.notifier.markDirty()
+        this.publishAssistantEntry(this.assistantStream.acceptDurable(change.entry))
+        return
+      case 'assistant-stream':
+        this.publishAssistantEntry(this.assistantStream.acceptFrame(change.frame))
     }
   }
 
   /** Replace the complete contiguous window and apply page-owned projection metadata. */
-  private installWindow(entries: readonly SessionEventLikeEntry[], hasMore: boolean, projections?: ProjectionsBaseline): void {
-    this.baseSeq = SessionLogOffset(entries[0]?.event.seq ?? 0)
+  private installWindow(
+    entries: readonly SessionEventLikeEntry[],
+    hasMore: boolean,
+    projections?: ProjectionsBaseline,
+    assistantStream?: SessionAssistantStreamBaseline,
+  ): void {
+    const visible = this.assistantStream.replace(entries, assistantStream)
+    this.baseSeq = SessionLogOffset(visible[0]?.event.seq ?? 0)
     this.hasMore = hasMore
-    if (entries.some(entry => entry.event.type === 'turn/start')) this.firstPromptPendingTurn = false
+    if (visible.some(entry => entry.event.type === 'turn/start')) this.firstPromptPendingTurn = false
     if (projections !== undefined) this.projections.seed(projections)
-    this.eventSource.replace(entries, hasMore)
-    for (const entry of entries) this.observeSubmissionEvent(entry.event)
+    this.eventSource.replace(visible, hasMore)
+    for (const entry of visible) this.observeSubmissionEvent(entry.event)
     this.notifier.markDirty()
   }
 
+  private publishAssistantEntry(result: ClientAssistantStreamResult): void {
+    if (result?.type === 'rebaseline') {
+      const events = this.events
+      queueMicrotask(() => {
+        if (events !== undefined && this.events === events) events.restart()
+      })
+      return
+    }
+    if (result?.type === 'publish' && this.appendLive(result.entry)) {
+      this.notifier.markDirty()
+    }
+  }
+
   /** Prepend one stream-validated history page. */
   private prependWindow(entries: readonly SessionEventLikeEntry[], hasMore: boolean): void {
     this.baseSeq = entries[0] === undefined ? this.baseSeq : SessionLogOffset(entries[0].event.seq)

packages/api/session-controller/src/client/transport.ts

@@ -12,6 +12,8 @@ import {
 } from '@deepseek-ai/dsh-api-gateway/client'
 import type {
   SessionAddress,
+  SessionAssistantStreamBaseline,
+  SessionAssistantStreamFrame,
   SessionControlFrame,
   SessionHistoryRecord,
   SessionPage,
@@ -40,6 +42,7 @@ export type SessionRemote = ClientRemote['session']
 /** Opening metadata carried only by a follow snapshot, never by loadOlder pages. */
 interface SessionJournalPage extends SessionPage {
   readonly projections?: SessionProjectionBaseline
+  readonly assistantStream?: SessionAssistantStreamBaseline
 }
 
 /** One complete publication from the Session journal stream. */
@@ -51,9 +54,12 @@ export type SessionJournalChange =
     readonly hasMore: boolean
   }
   | { readonly type: 'append'; readonly entry: SessionLiveEventEntry }
+  | { readonly type: 'assistant-stream'; readonly frame: SessionAssistantStreamFrame }
 
 function toSessionJournalChange(
-  change: RemoteJournalChange<SessionJournalPage, SessionHistoryRecord>,
+  change: RemoteJournalChange<
+    SessionJournalPage, SessionHistoryRecord, SessionAssistantStreamFrame
+  >,
 ): SessionJournalChange {
   switch (change.type) {
     case 'replace':
@@ -72,6 +78,8 @@ function toSessionJournalChange(
         entry: change.entry as unknown as SessionLiveEventEntry,
       }
     }
+    case 'notification':
+      return { type: 'assistant-stream', frame: change.notification }
   }
 }
 
@@ -136,7 +144,8 @@ export class SessionEventStream extends RemoteJournalStream<
   SessionJournalPage,
   SessionHistoryRecord,
   number,
-  ClientSessionPageRequest
+  ClientSessionPageRequest,
+  SessionAssistantStreamFrame
 > {
   /**
    * @param remote - generated Session namespace and Gateway stream factory.
@@ -169,12 +178,24 @@ export class SessionEventStream extends RemoteJournalStream<
   protected override async * follow(
     request: ClientSessionPageRequest,
     signal: AbortSignal,
-  ): AsyncIterable<RemoteJournalFrame<SessionHistoryRecord, number, SessionJournalPage>> {
+  ): AsyncIterable<RemoteJournalFrame<
+    SessionHistoryRecord, number, SessionJournalPage, SessionAssistantStreamFrame
+  >> {
+    let assistantRevision: number | undefined
     for await (const frame of this.remote.session.follow({
       address: this.address,
+      assistantStream: true,
       ...(request.maxMessages === undefined ? {} : { maxMessages: request.maxMessages }),
     }, signal)) {
       if (frame.type === 'snapshot') {
+        if (frame.assistantStream === undefined) {
+          throw new RemoteError(
+            'gateway/internal',
+            'session assistant stream omitted its opted-in opening baseline',
+            {},
+          )
+        }
+        assistantRevision = frame.assistantStream.revision
         yield {
           type: 'opened',
           cursor: frame.cursor,
@@ -182,10 +203,22 @@ export class SessionEventStream extends RemoteJournalStream<
             records: frame.records,
             hasMore: frame.hasMore,
             projections: frame.projections,
+            assistantStream: frame.assistantStream,
           },
         }
         continue
       }
+      if (frame.type === 'assistant-stream') {
+        const expected = (assistantRevision ?? 0) + 1
+        if (frame.frame.revision !== expected) {
+          throw new RemoteStreamCarrierError(
+            `session assistant stream skipped revision ${String(expected)}`,
+          )
+        }
+        assistantRevision = frame.frame.revision
+        yield { type: 'notification', notification: frame.frame }
+        continue
+      }
       yield { type: 'entry', entry: frame }
     }
   }