Merge branch 'master' into worktree/steer-followup-images

# Conflicts:
#	packages/api/session-controller/src/client/contract/session.ts

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

@@ -7,14 +7,44 @@
  * must stub); implementation-internal entry points (history staging, wire-frame
  * dispatch) stay on the class, invisible out here.
  */
-import type { AttachmentIdType, ImageAttachmentRef, PromptContentPart } from '@deepseek-ai/dsh-attachment'
+import type {
+  AttachmentIdType, ImageAttachmentRef, PromptContentPart,
+} from '@deepseek-ai/dsh-attachment'
 import type { MessageId } from '@deepseek-ai/dsh-llm/brand'
 import type { SessionId } from '@deepseek-ai/dsh-session/types'
 import type { RemoteResult } from '@deepseek-ai/dsh-typert-protocol'
 import type { ObservableSnapshot } from '@deepseek-ai/dsh-client-store'
-import type { QueueAction } from '../../types.ts'
+import type { QueueAction, SessionRequestId } from '../../types.ts'
 import type { ClientResult } from './result.ts'
-import type { SessionSnapshot } from './snapshot.ts'
+import type { PendingSubmissionImage, SessionSnapshot } from './snapshot.ts'
+
+/**
+ * Why a local submission echo left the snapshot: `observed` when its durable
+ * `user/message` event or host queue occurrence arrived (with the admitted
+ * image references in prompt order), `failed` when the prompt was rejected,
+ * threw, or was aborted before acceptance.
+ */
+export type PendingSubmissionRetirement =
+  | { readonly reason: 'observed'; readonly attachments: readonly ImageAttachmentRef[] }
+  | { readonly reason: 'failed' }
+
+/** Input registering one local submission echo ahead of its prompt call. */
+export interface BeginSubmissionInput {
+  /** Prompt text exactly as the upcoming prompt will send it. */
+  readonly text: string
+  /** Ordered image previews matching the upcoming prompt's image parts. */
+  readonly images: readonly PendingSubmissionImage[]
+  /** Settlement callback fired exactly once when the echo retires. */
+  readonly onRetire?: (retirement: PendingSubmissionRetirement) => void
+}
+
+/** One registered submission echo: the identity its prompt must carry, and the pre-prompt escape hatch. */
+export interface SubmissionHandle {
+  /** The prompt RPC identity; pass it to {@link ISession.prompt}. */
+  readonly requestId: SessionRequestId
+  /** Retire the echo as failed when the caller cannot reach prompt() (serialization failure); no-op after any other settlement. */
+  abandon(): void
+}
 
 /** Key-addressed projection read face (the useProjection resolution path; see ProjectionValueStore). */
 export interface ProjectionsFace {
@@ -33,16 +63,29 @@ export interface ISession {
   readonly sessionId: SessionId
   /** Host-computed projection values by key (the useProjection seat). */
   readonly projections: ProjectionsFace
+  /**
+   * Register one local submission echo in `snapshot.pendingSubmissions`,
+   * synchronously, before the caller serializes and sends the prompt. The
+   * echo retires when a durable `user/message` event or queue occurrence
+   * carrying the returned identity arrives, or when the identified prompt
+   * call fails.
+   * @param input - echo content and the optional settlement callback.
+   * @returns the minted identity for {@link prompt} plus the pre-prompt abandon path.
+   */
+  beginSubmission(input: BeginSubmissionInput): SubmissionHandle
   /**
    * Send a prompt into the session.
    * @param content - text plus browser-owned temporary image uploads.
    * @param mode - 'queue' appends a turn; 'steer' interrupts the running one.
+   * @param signal - optional caller cancellation for the complete admission round-trip.
+   * @param requestId - identity from {@link beginSubmission}; a failed identified prompt retires its echo.
    * @returns acceptance, or the business error (also mirrored into snapshot.promptError).
    */
   prompt(
     content: PromptContentPart[],
     mode: 'queue' | 'steer',
     signal?: AbortSignal,
+    requestId?: SessionRequestId,
   ): Promise<ClientResult<{ accepted: true }>>
   /**
    * Resolve one durable image referenced by this session.

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

@@ -3,6 +3,7 @@ import type { ContentBlock } from '@deepseek-ai/dsh-llm/types'
 import type { MessageId } from '@deepseek-ai/dsh-llm/brand'
 import type { SessionId } from '@deepseek-ai/dsh-session/types'
 import type { SubagentAddress } from '@deepseek-ai/dsh-subagent/client'
+import type { SessionRequestId } from '../../types.ts'
 import type { ClientFailure } from './result.ts'
 
 /** One transient inbox occurrence from the authoritative queue snapshot. */
@@ -10,11 +11,42 @@ export interface QueuedMessage {
   readonly id: MessageId
   readonly messageId: MessageId
   readonly placement: 'queued' | 'steering' | 'context'
+  /** Prompt-RPC identity of a browser-submitted occurrence; correlates the local submission echo. */
+  readonly rpcId?: SessionRequestId
   readonly content: readonly ContentBlock[]
   readonly preview: string
   readonly text: string | null
 }
 
+/** One image displayed by a local submission echo before durable admission. */
+export interface PendingSubmissionImage {
+  /** Browser-owned preview URL; its lifecycle belongs to the submitter, never this snapshot. */
+  readonly previewUrl: string
+  /** Browser file name, when the file had one. */
+  readonly name?: string
+  /** Intrinsic pixel width, when the submitter has probed it. */
+  readonly width?: number
+  /** Intrinsic pixel height, when the submitter has probed it. */
+  readonly height?: number
+}
+
+/**
+ * One local prompt-submission echo: inserted synchronously when a submission
+ * begins, so the conversation can show the message before serialization,
+ * transport, and durable admission complete. Client-memory only — reload and
+ * reconnect rebuild the conversation from durable events alone.
+ */
+export interface PendingSubmission {
+  /** The prompt RPC identity; the durable `user/message` source echoes it as `rpcId`. */
+  readonly requestId: SessionRequestId
+  /** Client wall-clock ms when the submission began. */
+  readonly time: number
+  /** Prompt text exactly as it will be sent (one text block). */
+  readonly text: string
+  /** Ordered image previews matching the prompt's image parts. */
+  readonly images: readonly PendingSubmissionImage[]
+}
+
 /** History-open lifecycle of a Session event window. */
 export type OpenState = 'cold' | 'loading' | 'open' | 'error'
 
@@ -28,6 +60,8 @@ export interface PromptError {
 export interface SessionSnapshot {
   readonly sessionId: SessionId
   readonly queue: readonly QueuedMessage[]
+  /** Local prompt-submission echoes not yet observed as durable events or queue occurrences. */
+  readonly pendingSubmissions: readonly PendingSubmission[]
   readonly running: boolean
   readonly subagent: {
     readonly address: SubagentAddress

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

@@ -40,7 +40,14 @@ export type {
   SessionProjectionMap,
   UseProjection,
 } from './sessions/projection-store.ts'
-export type { ISession, ProjectionsFace, SessionFace } from './contract/session.ts'
+export type {
+  BeginSubmissionInput,
+  ISession,
+  PendingSubmissionRetirement,
+  ProjectionsFace,
+  SessionFace,
+  SubmissionHandle,
+} from './contract/session.ts'
 export type { ISessions } from './contract/sessions.ts'
 export { MutableSessionEventSource } from './contract/events.ts'
 export type {
@@ -53,6 +60,8 @@ export type {
 } from './contract/events.ts'
 export type {
   OpenState,
+  PendingSubmission,
+  PendingSubmissionImage,
   PromptError,
   QueuedMessage,
   SessionSnapshot,

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

@@ -46,6 +46,7 @@ export class SessionQueueMirror {
         id: item.id,
         messageId: item.message.id,
         placement: item.placement,
+        ...(item.rpcId === undefined ? {} : { rpcId: item.rpcId }),
         content,
         preview: previewOf(content),
         text: textOf(content),

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

@@ -22,9 +22,11 @@ import type {
 } from '../../types.ts'
 import type { ClientFailure, ClientResult } from '../contract/result.ts'
 import { transportResult } from '../contract/result.ts'
-import type { SessionFace } from '../contract/session.ts'
 import type {
-  OpenState, PromptError, SessionSnapshot,
+  BeginSubmissionInput, PendingSubmissionRetirement, SessionFace, SubmissionHandle,
+} from '../contract/session.ts'
+import type {
+  OpenState, PendingSubmission, PromptError, SessionSnapshot,
 } from '../contract/snapshot.ts'
 import { MutableSessionEventSource } from '../contract/events.ts'
 import type {
@@ -99,6 +101,14 @@ export class Session implements SessionFace {
   private removed = false
   private promptError: PromptError | null = null
   private lastAgentError: string | null = null
+  /** Local submission echoes, insertion-ordered (see SessionSnapshot.pendingSubmissions). */
+  private pendingSubmissions: readonly PendingSubmission[] = []
+  /** Per-echo settlement state; `retiring` latches the first observation so a
+   *  queue frame and its durable event cannot both retire one echo. */
+  private readonly submissionSettlements = new Map<SessionRequestId, {
+    readonly onRetire?: ((retirement: PendingSubmissionRetirement) => void) | undefined
+    retiring: boolean
+  }>()
   /** Owns the addressed page/follow lifecycle while this Session is open. */
   private events: SessionEventStream | undefined
 
@@ -168,16 +178,42 @@ export class Session implements SessionFace {
 
   // ---- Operations ----
 
+  /**
+   * Register one local submission echo (see the ISession declaration).
+   * Synchronous through markDirty: the echo is in the very next snapshot, so
+   * the conversation can paint it before the caller starts serializing.
+   * @param input - echo content and the optional settlement callback.
+   * @returns the minted identity for {@link prompt} plus the pre-prompt abandon path.
+   */
+  beginSubmission(input: BeginSubmissionInput): SubmissionHandle {
+    const requestId = randomUUID() as SessionRequestId
+    this.pendingSubmissions = [...this.pendingSubmissions, {
+      requestId,
+      time: Date.now(),
+      text: input.text,
+      images: input.images,
+    }]
+    this.submissionSettlements.set(requestId, { onRetire: input.onRetire, retiring: false })
+    // The blank → engaging edge flips here, ahead of prompt(): the composer
+    // docks and the echo renders on the click's own frame.
+    this.promptAttempted = true
+    this.notifier.markDirty()
+    return { requestId, abandon: () => { this.retireFailedSubmission(requestId) } }
+  }
+
   /**
    * Send (queue/steer passed through 1:1); failures land in the snapshot's promptError.
    * @param content - text plus browser-owned temporary image uploads.
    * @param mode - queue appends after the current turn; steer interrupts it.
+   * @param signal - optional caller cancellation for the complete admission round-trip.
+   * @param requestId - identity from {@link beginSubmission}; a failed identified prompt retires its echo.
    * @returns the prompt result (also mirrored into promptError on failure).
    */
   async prompt(
     content: PromptContentPart[],
     mode: 'queue' | 'steer',
     signal?: AbortSignal,
+    requestId?: SessionRequestId,
   ): Promise<ClientResult<{ accepted: true }>> {
     this.promptError = null
     this.lastAgentError = null
@@ -192,7 +228,7 @@ export class Session implements SessionFace {
       if (this.address === undefined) {
         const clientTimeZone = resolvedClientTimeZone()
         result = toSessionResult(await this.remote.session.prompt({
-          requestId: randomUUID() as SessionRequestId,
+          requestId: requestId ?? randomUUID() as SessionRequestId,
           sessionId: this.sessionId,
           mode,
           content,
@@ -222,6 +258,7 @@ export class Session implements SessionFace {
       result = transportResult(error)
     }
     if (!result.ok) {
+      if (requestId !== undefined) this.retireFailedSubmission(requestId)
       this.promptError = { op: 'send', error: result.error }
       this.notifier.markDirty()
       return result
@@ -424,6 +461,7 @@ export class Session implements SessionFace {
    */
   replaceControl(queue: readonly SessionQueuedItem[]): void {
     this.queueMirror.replace(queue)
+    this.observeSubmissionQueue(queue)
     this.notifier.markDirty()
   }
 
@@ -433,6 +471,7 @@ export class Session implements SessionFace {
    */
   handleControlFrame(frame: Extract<SessionControlFrame, { type: 'queue' }>): void {
     this.queueMirror.replace(frame.items)
+    this.observeSubmissionQueue(frame.items)
     this.notifier.markDirty()
   }
 
@@ -513,6 +552,12 @@ export class Session implements SessionFace {
    * @returns when the Remote iterator has completed teardown.
    */
   async dispose(): Promise<void> {
+    // Unsettled echoes retire as failed so their owners can restore or
+    // release browser resources; echoes already scheduled as observed keep
+    // that settlement.
+    for (const requestId of [...this.submissionSettlements.keys()]) {
+      this.retireFailedSubmission(requestId)
+    }
     this.openGeneration++
     const events = this.events
     this.events = undefined
@@ -571,6 +616,7 @@ export class Session implements SessionFace {
     if (entries.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.notifier.markDirty()
   }
 
@@ -588,9 +634,70 @@ export class Session implements SessionFace {
     if (event.type === 'turn/start') this.firstPromptPendingTurn = false
     const queueChanged = this.queueMirror.acceptDurable(event)
     this.eventSource.append(entry)
+    // After the feed append: the conversation assembly's animation frame is
+    // registered by the feed subscribers above, so the echo-retirement frame
+    // scheduled here always runs after the durable node became renderable.
+    this.observeSubmissionEvent(event)
     return queueChanged || awaitingFirstTurn !== this.firstPromptPendingTurn
   }
 
+  /** Retire the matching echo when a durable browser-prompt `user/message` becomes visible. */
+  private observeSubmissionEvent(event: { readonly type: string; readonly data?: unknown }): void {
+    if (this.submissionSettlements.size === 0 || event.type !== 'user/message') return
+    // Structural read: window entries may be compact history records, so the
+    // fields are narrowed rather than trusted (same posture as Conversation
+    // assembly matchers).
+    const data = event.data as { readonly source?: unknown; readonly content?: unknown } | undefined
+    const source = data?.source as { readonly kind?: unknown; readonly rpcId?: unknown } | undefined
+    if (source?.kind !== 'user' || typeof source.rpcId !== 'string') return
+    this.scheduleObservedRetirement(source.rpcId as SessionRequestId, imageRefsIn(data?.content))
+  }
+
+  /** Retire echoes whose prompts landed in the host inbox instead of the log (running-turn submissions). */
+  private observeSubmissionQueue(items: readonly SessionQueuedItem[]): void {
+    if (this.submissionSettlements.size === 0) return
+    for (const item of items) {
+      if (item.rpcId !== undefined) {
+        this.scheduleObservedRetirement(item.rpcId, imageRefsIn(item.message.content))
+      }
+    }
+  }
+
+  /**
+   * Latch one observed settlement and remove the echo an animation frame
+   * later. The delay keeps the echo in the snapshot until the frame in which
+   * the durable node (whose assembly frame was registered first) is
+   * renderable; the render-time rpcId dedupe hides the one-frame overlap.
+   */
+  private scheduleObservedRetirement(
+    requestId: SessionRequestId,
+    attachments: readonly ImageAttachmentRef[],
+  ): void {
+    const settlement = this.submissionSettlements.get(requestId)
+    if (settlement === undefined || settlement.retiring) return
+    settlement.retiring = true
+    scheduleFrame(() => { this.finishSubmission(requestId, { reason: 'observed', attachments }) })
+  }
+
+  /** Remove one unsettled echo immediately (prompt rejection, abort, or disposal). */
+  private retireFailedSubmission(requestId: SessionRequestId): void {
+    const settlement = this.submissionSettlements.get(requestId)
+    if (settlement === undefined || settlement.retiring) return
+    settlement.retiring = true
+    this.finishSubmission(requestId, { reason: 'failed' })
+  }
+
+  /** Single removal point: drop the echo, publish, then notify the owner. */
+  private finishSubmission(requestId: SessionRequestId, retirement: PendingSubmissionRetirement): void {
+    const settlement = this.submissionSettlements.get(requestId)
+    /* v8 ignore next -- retiring latches before every schedule, so one settlement never finishes twice. */
+    if (settlement === undefined) return
+    this.submissionSettlements.delete(requestId)
+    this.pendingSubmissions = this.pendingSubmissions.filter(echo => echo.requestId !== requestId)
+    this.notifier.markDirty()
+    settlement.onRetire?.(retirement)
+  }
+
   /** Publish a terminal background failure only while this stream still owns the Session. */
   private failEventStream(events: SessionEventStream, generation: number, error: unknown): void {
     if (generation !== this.openGeneration || this.events !== events) return
@@ -607,6 +714,7 @@ export class Session implements SessionFace {
     return {
       sessionId: this.sessionId,
       queue: this.queueMirror.snapshot(),
+      pendingSubmissions: this.pendingSubmissions,
       running: this.running,
       subagent: this.address === undefined
         ? null
@@ -634,6 +742,26 @@ export class Session implements SessionFace {
   }
 }
 
+/** Run one callback on the next animation frame, or a macrotask where no frame clock exists. */
+function scheduleFrame(fn: () => void): void {
+  if (typeof requestAnimationFrame === 'function') requestAnimationFrame(() => { fn() })
+  else setTimeout(fn, 0)
+}
+
+/** Image attachment references in one structurally-read content block list, in block order. */
+function imageRefsIn(content: unknown): readonly ImageAttachmentRef[] {
+  if (!Array.isArray(content)) return []
+  const refs: ImageAttachmentRef[] = []
+  for (const block of content) {
+    if (typeof block !== 'object' || block === null) continue
+    const candidate = block as { readonly type?: unknown; readonly attachment?: unknown }
+    if (candidate.type === 'image' && typeof candidate.attachment === 'object' && candidate.attachment !== null) {
+      refs.push(candidate.attachment as ImageAttachmentRef)
+    }
+  }
+  return refs
+}
+
 /** Convert a terminal Session stream failure to the Client error vocabulary. */
 function openFailure(error: unknown): ClientFailure {
   const failure = sessionStreamFailure(error)

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

@@ -188,16 +188,24 @@ function queueItems(
     ...project('next-turn').map(message => ({
       id: message.id,
       placement: 'queued' as const,
+      ...promptRpcId(message),
       message: { id: message.id, content: message.content as unknown as JsonValue[] },
     })),
     ...project('next-step').map(message => ({
       id: message.id,
       placement: message.source.kind === 'user' ? 'steering' as const : 'context' as const,
+      ...promptRpcId(message),
       message: { id: message.id, content: message.content as unknown as JsonValue[] },
     })),
   ]
 }
 
+/** Prompt-RPC identity carried by a browser-submitted message's user source. */
+function promptRpcId(message: UserMessage): Pick<SessionQueuedItem, 'rpcId'> {
+  const source = message.source
+  return source.kind === 'user' && 'rpcId' in source ? { rpcId: source.rpcId } : {}
+}
+
 function jobView(job: JobSnapshot): SessionJob {
   return {
     id: job.id,

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

@@ -424,6 +424,8 @@ export type SessionFollowFrame =
 export interface SessionQueuedItem {
   readonly id: MessageId
   readonly placement: 'queued' | 'steering' | 'context'
+  /** Prompt-RPC identity from the queued message's user source; clients retire the matching local submission echo on it. */
+  readonly rpcId?: SessionRequestId
   /** JSON-safe message fields consumed by pending-queue presentation. */
   readonly message: {
     readonly id: MessageId