fix(session): harden embedded assistant streams

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

@@ -2,6 +2,7 @@
 
 import type { AssistantStreamFrame } from '@deepseek-ai/dsh-agent'
 import { AssistantStreamAccumulator } from '@deepseek-ai/dsh-llm'
+import type { SessionSeqCursor } from '@deepseek-ai/dsh-session'
 import type { JsonValue } from '@deepseek-ai/dsh-util-values'
 import type {
   SessionAssistantStreamAttempt,
@@ -11,6 +12,7 @@ import type {
 interface MutableAttempt {
   readonly attemptId: SessionAssistantStreamAttempt['attemptId']
   readonly startedTime: number
+  readonly startedAfterSeq: SessionSeqCursor
   readonly turn: number
   readonly step: number
   readonly stream: AssistantStreamAccumulator
@@ -32,8 +34,9 @@ export class SessionAssistantStreamAccumulator {
   /**
    * Fold one trusted frame from the current attached Agent lifecycle.
    * @param frame - next dense process-local Assistant frame.
+   * @param durableCursor - last committed Session seq when this frame was observed.
    */
-  accept(frame: AssistantStreamFrame): void {
+  accept(frame: AssistantStreamFrame, durableCursor: SessionSeqCursor): void {
     if (frame.type === 'start' && frame.revision === 1 && this.revision !== 0) {
       this.activeAttempt = undefined
       this.revision = 0
@@ -50,6 +53,7 @@ export class SessionAssistantStreamAccumulator {
         this.activeAttempt = {
           attemptId: frame.attemptId,
           startedTime: frame.startedTime,
+          startedAfterSeq: durableCursor,
           turn: frame.turn,
           step: frame.step,
           stream: new AssistantStreamAccumulator(),
@@ -87,6 +91,7 @@ export class SessionAssistantStreamAccumulator {
         activeAttempt: {
           attemptId: this.activeAttempt.attemptId,
           startedTime: this.activeAttempt.startedTime,
+          startedAfterSeq: this.activeAttempt.startedAfterSeq,
           turn: this.activeAttempt.turn,
           step: this.activeAttempt.step,
           nextIndex: this.activeAttempt.nextIndex,

packages/api/session-controller/src/client/contract/events.ts

@@ -163,6 +163,20 @@ export class MutableSessionEventSource implements SessionEventSource {
     })
   }
 
+  /**
+   * Replace one attempt's transient rows with its committed durable settlement.
+   * @param attemptId - process-local attempt whose live rows are now redundant.
+   * @param entry - durable settlement committed for that attempt.
+   */
+  settleAssistant(attemptId: string, entry: SessionLiveEventEntry): void {
+    const entries = materialize(this.window).filter(candidate => (
+      candidate.type !== 'transient' || String(candidate.event.data.attemptId) !== attemptId
+    ))
+    entries.push(entry)
+    this.window = leaf(entries)
+    this.publish(this.snapshot.hasMore, { kind: 'replace', entries })
+  }
+
   private publish(
     hasMore: boolean,
     change: SessionEventChange,

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

@@ -14,7 +14,7 @@ import type {
 
 interface ActiveAttempt {
   readonly attemptId: string
-  readonly startedTime: number
+  readonly startedAfterSeq: number
   readonly turn: number
   readonly step: number
   nextIndex: number
@@ -23,6 +23,11 @@ interface ActiveAttempt {
 /** One Web publication decision from the assistant stream fold. */
 export type ClientAssistantStreamResult =
   | { readonly type: 'publish'; readonly entry: SessionLiveEventEntry }
+  | {
+    readonly type: 'settlement'
+    readonly attemptId: string
+    readonly entry: SessionLiveEventEntry
+  }
   | { readonly type: 'transient'; readonly entry: SessionTransientEventEntry }
   | { readonly type: 'rebaseline' }
   | undefined
@@ -52,7 +57,7 @@ export class ClientAssistantStream {
     if (opening !== undefined) {
       this.activeAttempt = {
         attemptId: String(opening.attemptId),
-        startedTime: opening.startedTime,
+        startedAfterSeq: opening.startedAfterSeq,
         turn: opening.turn,
         step: opening.step,
         nextIndex: opening.nextIndex,
@@ -120,7 +125,7 @@ export class ClientAssistantStream {
         this.pending.clear()
         this.activeAttempt = {
           attemptId: String(frame.attemptId),
-          startedTime: frame.startedTime,
+          startedAfterSeq: frame.startedAfterSeq,
           turn: frame.turn,
           step: frame.step,
           nextIndex: 0,
@@ -170,7 +175,8 @@ export class ClientAssistantStream {
           return { type: 'rebaseline' }
         }
         this.pending.delete(frame.outcome.seq)
-        return this.publish(entry)
+        this.publishedSeqs.add(entry.event.seq)
+        return { type: 'settlement', attemptId: attempt.attemptId, entry }
       }
     }
   }
@@ -182,7 +188,7 @@ export class ClientAssistantStream {
     if (attempt === undefined
       || (event.type !== 'assistant/message' && event.type !== 'assistant/attempt')
       || (event.type === 'assistant/message' && event.surfaceOp !== 'append')
-      || event.time < attempt.startedTime
+      || event.seq <= attempt.startedAfterSeq
       || attempt.turn !== event.data.turn
       || attempt.step !== event.data.step) return undefined
     return attempt

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

@@ -668,6 +668,12 @@ export class Session implements SessionFace {
       })
       return
     }
+    if (result?.type === 'settlement') {
+      this.eventSource.settleAssistant(result.attemptId, result.entry)
+      this.observeSubmissionEvent(result.entry.event)
+      this.notifier.markDirty()
+      return
+    }
     if (result?.type === 'publish' && this.appendLive(result.entry)) {
       this.notifier.markDirty()
     } else if (result?.type === 'transient') {

packages/api/session-controller/src/history.ts

@@ -57,7 +57,7 @@ export class SessionHistoryController {
         stream = new SessionAssistantStreamAccumulator()
         this.assistantStreams.set(agent.session.id, stream)
       }
-      stream.accept(frame)
+      stream.accept(frame, cursorBeforeNext(agent.session.seq))
     }, { global: true })
     ctx.on('agent/disposed', ({ agent }) => {
       this.assistantStreams.delete(agent.session.id)
@@ -166,7 +166,7 @@ export class SessionHistoryController {
         if (agent.session.id !== target) return
         buffered.pushBack({
           type: 'assistant-stream',
-          frame: wireAssistantStreamFrame(frame),
+          frame: wireAssistantStreamFrame(frame, cursorBeforeNext(agent.session.seq)),
           ordinal: ++assistantStreamOrdinal,
         })
         notify()
@@ -274,8 +274,16 @@ export class SessionHistoryController {
 
 }
 
-function wireAssistantStreamFrame(frame: AssistantStreamFrame): SessionAssistantStreamFrame {
-  if (frame.type !== 'chunk') return frame
+function cursorBeforeNext(nextSeq: SessionLogOffsetType): SessionSeqCursor {
+  return nextSeq === 0 ? -1 : SessionSeq(nextSeq - 1)
+}
+
+function wireAssistantStreamFrame(
+  frame: AssistantStreamFrame,
+  durableCursor: SessionSeqCursor,
+): SessionAssistantStreamFrame {
+  if (frame.type === 'start') return { ...frame, startedAfterSeq: durableCursor }
+  if (frame.type === 'end') return frame
   return {
     ...frame,
     chunk: frame.chunk as JsonValue,

packages/api/session-controller/src/types.ts

@@ -6,7 +6,7 @@ import type {
 import type { Branded } from '@deepseek-ai/dsh-brand'
 import type { LlmAttemptId, MessageId } from '@deepseek-ai/dsh-llm/brand'
 import type { ContentBlock } from '@deepseek-ai/dsh-llm'
-import type { SessionId } from '@deepseek-ai/dsh-session/types'
+import type { SessionId, SessionSeqCursor } from '@deepseek-ai/dsh-session/types'
 import type { SessionProjectionMap } from '@deepseek-ai/dsh-session-projection/types'
 import type { JobId } from '@deepseek-ai/dsh-jobs/brand'
 import type { JsonValue } from '@deepseek-ai/dsh-util-values'
@@ -438,6 +438,8 @@ export interface SessionAssistantStreamAttempt {
   readonly attemptId: LlmAttemptId
   /** Safe-integer wall-clock time copied from the attempt's start frame. */
   readonly startedTime: number
+  /** Last durable Session seq observed when this attempt started. */
+  readonly startedAfterSeq: SessionSeqCursor
   readonly turn: number
   readonly step: number
   /** Dense position expected for the next live chunk frame. */
@@ -459,6 +461,7 @@ export type SessionAssistantStreamFrame =
     readonly attemptId: LlmAttemptId
     readonly revision: number
     readonly startedTime: number
+    readonly startedAfterSeq: SessionSeqCursor
     readonly turn: number
     readonly step: number
   }