feat(conversation): stream files through submission lifecycle

2026-09-11 04:00:38 +00:00 · 2026-09-01 16:11:43 +08:00
parent 8a0ff3aff8
commit bafa6ae11d
27 changed files with 559 additions and 145 deletions
@@ -7,22 +7,25 @@
 * must stub); implementation-internal entry points (history staging, wire-frame
 * dispatch) stay on the class, invisible out here.
 */
-import type { AttachmentIdType, ImageAttachmentRef } from '@deepseek-ai/dsh-attachment'
+import type { AttachmentIdType, FileAttachmentRef, ImageAttachmentRef } 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 { PromptContentPart, QueueAction, SessionRequestId } from '../../types.ts'
-import type { PendingSubmissionImage, SessionSnapshot } from './snapshot.ts'
+import type { FileUploadReceiptId, PromptContentPart, QueueAction, SessionRequestId } from '../../types.ts'
+import type { PendingSubmissionAttachment, 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,
+ * attachment 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: 'observed'
+    readonly attachments: readonly (ImageAttachmentRef | FileAttachmentRef)[]
+  }
  | { readonly reason: 'failed' }

 /** Input registering one local submission echo ahead of its prompt call. */
@@ -31,8 +34,8 @@ export interface BeginSubmissionInput {
  readonly mode: 'queue' | 'steer'
  /** 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[]
+  /** Ordered image previews and durable file metadata matching the upcoming prompt attachments. */
+  readonly attachments: readonly PendingSubmissionAttachment[]
  /** Settlement callback fired exactly once when the echo retires. */
  readonly onRetire?: (retirement: PendingSubmissionRetirement) => void
 }
@@ -86,6 +89,21 @@ export interface ISession {
    signal?: AbortSignal,
    requestId?: SessionRequestId,
  ): Promise<RemoteResult<{ accepted: true }>>
+  /**
+   * Persist one browser file verbatim and stage it for a later prompt on this
+   * session. The returned opaque receipt is what a prompt file part cites.
+   * @param data - browser Blob or exact file bytes.
+   * @param name - optional display name; the host sanitizes the stored leaf name.
+   * @param signal - optional cancellation for the active upload.
+   * @param onProgress - optional byte-progress observer for background Blob uploads.
+   * @returns the staged-upload receipt and durable file reference, or the business error.
+   */
+  uploadFile(
+    data: Blob | Uint8Array,
+    name?: string,
+    signal?: AbortSignal,
+    onProgress?: (progress: { readonly loaded: number; readonly total?: number }) => void,
+  ): Promise<RemoteResult<{ receiptId: FileUploadReceiptId; file: FileAttachmentRef }>>
  /**
   * Resolve one durable image referenced by this session.
   * @param attachmentId - opaque id found in the folded session log.
@@ -1,5 +1,6 @@
 /** Session-owned observable state excluding Conversation target data. */
 import type { ContentBlock } from '@deepseek-ai/dsh-llm/types'
+import type { FileAttachmentRef } 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 { SubagentAddress } from '@deepseek-ai/dsh-subagent/client'
@@ -30,6 +31,11 @@ export interface PendingSubmissionImage {
  readonly height?: number
 }

+/** One attachment displayed by a local submission echo, in prompt order. */
+export type PendingSubmissionAttachment =
+  | ({ readonly type: 'image' } & PendingSubmissionImage)
+  | { readonly type: 'file'; readonly attachment: FileAttachmentRef }
+
 /** Client surface selected when a local submission begins. */
 export type PendingSubmissionPlacement = 'transcript' | 'queued' | 'steering'

@@ -48,8 +54,8 @@ export interface PendingSubmission {
  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[]
+  /** Ordered image previews and durable file metadata matching the prompt attachments. */
+  readonly attachments: readonly PendingSubmissionAttachment[]
 }

 /** History-open lifecycle of a Session event window. */
@@ -2,6 +2,7 @@

 import type { Context } from '@deepseek-ai/cordis'
 import type {} from '@deepseek-ai/dsh-agent/types'
+import type { ConnectionHandle } from '@deepseek-ai/dsh-client-connection/client'
 import { createSessionControlStream } from './transport.ts'
 import { ClientSessions } from './sessions/service.ts'
 import type { SessionRemotes } from './sessions/remotes.ts'
@@ -59,6 +60,7 @@ export type {
 export type {
  OpenState,
  PendingSubmission,
+  PendingSubmissionAttachment,
  PendingSubmissionImage,
  PendingSubmissionPlacement,
  PromptError,
@@ -75,6 +77,7 @@ declare module '@deepseek-ai/cordis' {

 /** Required Remote and Context projection services. */
 export const inject = [
+  'connection',
  'typert',
  'remote',
  'remote.commands',
@@ -87,8 +90,9 @@ export const inject = [
 * @param ctx - Client Cordis context.
 */
 export function apply(ctx: Context): void {
+  const connection = ctx.get('connection') as ConnectionHandle
  const remotes = ctx.remote as unknown as SessionRemotes
-  const sessions = new ClientSessions(ctx, remotes)
+  const sessions = new ClientSessions(ctx, remotes, connection.backgroundUploads)
  ctx.remote.$on('api-session/added', (summary) => { sessions.handleSessionAdded(summary) })
  ctx.remote.$on('api-session/removed', (sessionId) => { sessions.handleSessionRemoved(sessionId) })
  ctx.remote.$on('api-session/status', (sessionId, running) => {
@@ -25,6 +25,7 @@ import { Notifier } from './notifier.ts'
 import { ProjectionValueStore } from './projection-store.ts'
 import { Session } from './session.ts'
 import type { SessionRemotes } from './remotes.ts'
+import type { BackgroundUploadTransport } from '@deepseek-ai/dsh-client-connection/client'

 /**
 * List arrival lifecycle, orthogonal to the pull-activity `state` axis:
@@ -149,6 +150,7 @@ export class SessionManager {
    private readonly remote: SessionRemotes,
    restoredSelection?: SessionId,
    restoredAddress?: SubagentAddress,
+    private readonly backgroundUploads?: BackgroundUploadTransport,
  ) {
    this.selected = restoredSelection
    if (restoredAddress !== undefined) this.addresses.set(restoredAddress.childSessionId, restoredAddress)
@@ -328,6 +330,7 @@ export class SessionManager {
        this.recordMutation({ kind: 'engaged', sessionId: engaged.sessionId })
      },
      projections: this.projectionStore(sessionId),
+      ...(this.backgroundUploads === undefined ? {} : { backgroundUploads: this.backgroundUploads }),
    })
  }

@@ -5,11 +5,11 @@ import type { QueuedMessage } from '../contract/snapshot.ts'

 const QUEUE_PREVIEW_CHARS = 200

-// Image blocks are excluded: queue presentation renders them as thumbnails
-// from `content`, so the text preview covers only what has no visual form.
+// Attachment blocks are excluded: queue presentation renders them from
+// `content`, so the text preview covers only what has no visual form.
 function previewOf(content: readonly ContentBlock[]): string {
  const flat = content
-    .filter(block => block.type !== 'image')
+    .filter(block => block.type !== 'image' && block.type !== 'file')
    .map(block => (block.type === 'text' ? block.text : `[${block.type}]`))
    .join(' ').replace(/\s+/g, ' ').trim()
  const chars = Array.from(flat)
@@ -5,8 +5,8 @@
 * @module @deepseek-ai/dsh-api-session-controller/client/sessions/remotes
 */

-import type { EncodedImageAttachment } from '@deepseek-ai/dsh-attachment/types'
 import type { ClientRemote } from '@deepseek-ai/dsh-api-gateway/client'
+import type { CommandSubmitAttachment } from '@deepseek-ai/dsh-commands/types'
 import type { SessionId } from '@deepseek-ai/dsh-session/types'
 import type {
  SubagentCatalog, SubagentInterruptReceipt, SubagentPromptReceipt, SubagentPromptRequest,
@@ -19,7 +19,7 @@ export interface SessionCommandsRemote {
  execute(
    agentId: SessionId,
    line: string,
-    images: readonly EncodedImageAttachment[],
+    attachments: readonly CommandSubmitAttachment[],
    signal?: AbortSignal,
  ): Promise<RemoteResult<object | undefined>>
 }
@@ -32,6 +32,7 @@ import type { AgentContext, ISessions } from '../contract/sessions.ts'
 import { createScope, scopeOf as scopeTagOf } from '../scope.ts'
 import { SessionManager } from './manager.ts'
 import type { SessionRemotes } from './remotes.ts'
+import type { BackgroundUploadTransport } from '@deepseek-ai/dsh-client-connection/client'
 import type { SessionListPhase, SessionSearchResultItem, SubagentCatalogSnapshot } from './manager.ts'
 import type { Session } from './session.ts'

@@ -221,6 +222,7 @@ export class ClientSessions implements ISessions {
  constructor(
    private readonly rootCtx: Context,
    remote: SessionRemotes,
+    backgroundUploads?: BackgroundUploadTransport,
  ) {
    this.selection = createSnapshotStore<SessionSelection>(
      {},
@@ -230,6 +232,7 @@ export class ClientSessions implements ISessions {
      remote,
      restored.sessionId,
      restored.subagentAddress,
+      backgroundUploads,
    )
    this.list = createSnapshotStore<SessionListState>({
      ids: [], byId: {}, current: undefined, phase: 'pending',
@@ -1,8 +1,11 @@
 // Sessions remain resident after creation so their open Remote sources keep running off-screen.

 import type { Context } from '@deepseek-ai/cordis'
-import { randomUUID } from '@deepseek-ai/dsh-util-crypto'
-import type { AttachmentIdType, ImageAttachmentRef } from '@deepseek-ai/dsh-attachment'
+import { bytesToBase64, randomUUID } from '@deepseek-ai/dsh-util-crypto'
+import type {
+  BackgroundUploadProgress, BackgroundUploadTransport,
+} from '@deepseek-ai/dsh-client-connection/client'
+import type { AttachmentIdType, FileAttachmentRef, ImageAttachmentRef } from '@deepseek-ai/dsh-attachment'
 import type { SubagentAddress } from '@deepseek-ai/dsh-subagent/client'
 import type { MessageId } from '@deepseek-ai/dsh-llm/brand'
 import type { SessionId } from '@deepseek-ai/dsh-session/types'
@@ -15,7 +18,9 @@ import type {
  SessionControlFrame,
  SessionQueuedItem,
  SessionRequestId,
+  SessionUploadFileValue,
 } from '../../types.ts'
+import { SESSION_FILE_UPLOAD_PATH } from '../../file-upload-path.ts'
 import type {
  BeginSubmissionInput, PendingSubmissionRetirement, SessionFace, SubmissionHandle,
 } from '../contract/session.ts'
@@ -28,6 +33,7 @@ import type {
 } from '../contract/events.ts'
 import { Notifier } from './notifier.ts'
 import { isRemoteFailure } from '@deepseek-ai/dsh-api-gateway/client'
+import { RemoteError } from '@deepseek-ai/dsh-typert-protocol'
 import type { RemoteFailure, RemoteResult } from '@deepseek-ai/dsh-typert-protocol'
 import type { SessionRemotes } from './remotes.ts'
 import { ProjectionValueStore } from './projection-store.ts'
@@ -62,6 +68,8 @@ export interface SessionOptions {
   * private store (bare object-layer construction).
   */
  projections?: ProjectionValueStore
+  /** Physical large-body carrier supplied by the active browser Connection. */
+  backgroundUploads?: BackgroundUploadTransport
 }

 /**
@@ -196,7 +204,7 @@ export class Session implements SessionFace {
        : 'transcript',
      time: Date.now(),
      text: input.text,
-      images: input.images,
+      attachments: input.attachments,
    }]
    this.submissionSettlements.set(requestId, { onRetire: input.onRetire, retiring: false })
    // The blank → engaging edge flips here, ahead of prompt(): the composer
@@ -208,7 +216,7 @@ export class Session implements SessionFace {

  /**
   * Send (queue/steer passed through 1:1); failures land in the snapshot's promptError.
-   * @param content - text plus browser-owned temporary image uploads.
+   * @param content - text, browser-owned temporary image uploads, and staged-file receipts.
   * @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.
@@ -238,13 +246,25 @@ export class Session implements SessionFace {
        content,
        clientTimeZone,
      }, signal)
+    } else if (content.some(part => part.type === 'file')) {
+      result = {
+        ok: false,
+        error: new RemoteError(
+          'subagent/attachment-invalid',
+          'subagent continuation does not accept files',
+          { reason: 'SUBAGENT_FILE_UNSUPPORTED' },
+        ),
+      }
    } else {
+      // The preceding branch rejects file parts before the narrower subagent
+      // wire type is used; this array is not filtered or reordered.
+      const routedContent = content as Exclude<PromptContentPart, { readonly type: 'file' }>[]
      const routed = await this.remote.subagents.prompt({
        requestId: randomUUID() as SessionRequestId,
        parentSessionId: this.address.parentSessionId,
        childSessionId: this.address.childSessionId,
        mode: 'continuable',
-        content,
+        content: routedContent,
        clientTimeZone: resolvedClientTimeZone(),
      }, signal)
      result = routed.ok ? { ok: true, value: { accepted: true } } : routed
@@ -271,6 +291,52 @@ export class Session implements SessionFace {
    return result
  }

+  /**
+   * Persist one browser file verbatim and stage it for a later prompt on this
+   * session (ordinary sessions only; subagent conversations refuse).
+   * @param data - exact file bytes.
+   * @param name - optional display name; the host sanitizes the stored leaf name.
+   * @returns the staged-upload receipt and durable file reference, or the business error.
+   */
+  async uploadFile(
+    data: Blob | Uint8Array,
+    name?: string,
+    signal?: AbortSignal,
+    onProgress?: (progress: BackgroundUploadProgress) => void,
+  ): Promise<RemoteResult<SessionUploadFileValue>> {
+    if (this.address !== undefined) {
+      return {
+        ok: false,
+        error: new RemoteError(
+          'subagent/attachment-invalid',
+          'subagent conversations do not accept file uploads',
+          { reason: 'SUBAGENT_FILE_UNSUPPORTED' },
+        ),
+      }
+    }
+    if (data instanceof Blob && this.options.backgroundUploads !== undefined) {
+      const query = new URLSearchParams({ sessionId: this.sessionId })
+      if (name !== undefined) query.set('name', name)
+      const response = await this.options.backgroundUploads.post({
+        path: `${SESSION_FILE_UPLOAD_PATH}?${query.toString()}`,
+        body: data,
+        headers: { 'content-type': 'application/octet-stream' },
+        ...(signal === undefined ? {} : { signal }),
+        ...(onProgress === undefined ? {} : { onProgress }),
+      })
+      if (response.status !== 200) {
+        throw new Error(`file upload transport failed with HTTP ${String(response.status)}`)
+      }
+      return parseFileUploadResult(response.body)
+    }
+    const bytes = data instanceof Uint8Array ? data : new Uint8Array(await data.arrayBuffer())
+    return this.remote.session.uploadFile({
+      sessionId: this.sessionId,
+      data: bytesToBase64(bytes),
+      ...(name === undefined ? {} : { name }),
+    }, signal)
+  }
+
  /**
   * Resolve one image referenced by this session into browser-consumable bytes.
   * @param attachmentId - opaque id found in the folded session log.
@@ -657,7 +723,7 @@ export class Session implements SessionFace {
    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))
+    this.scheduleObservedRetirement(source.rpcId as SessionRequestId, attachmentRefsIn(data?.content))
  }

  /** Retire echoes whose prompts landed in the host inbox instead of the log (running-turn submissions). */
@@ -665,7 +731,7 @@ export class Session implements SessionFace {
    if (this.submissionSettlements.size === 0) return
    for (const item of items) {
      if (item.rpcId !== undefined) {
-        this.scheduleObservedRetirement(item.rpcId, imageRefsIn(item.message.content))
+        this.scheduleObservedRetirement(item.rpcId, attachmentRefsIn(item.message.content))
      }
    }
  }
@@ -678,7 +744,7 @@ export class Session implements SessionFace {
   */
  private scheduleObservedRetirement(
    requestId: SessionRequestId,
-    attachments: readonly ImageAttachmentRef[],
+    attachments: readonly (ImageAttachmentRef | FileAttachmentRef)[],
  ): void {
    const settlement = this.submissionSettlements.get(requestId)
    if (settlement === undefined || settlement.retiring) return
@@ -750,21 +816,62 @@ export class Session implements SessionFace {
  }
 }

+function parseFileUploadResult(body: string): RemoteResult<SessionUploadFileValue> {
+  const value = JSON.parse(body) as unknown
+  if (!isRecord(value) || typeof value.ok !== 'boolean') {
+    throw new TypeError('file upload transport returned an invalid result')
+  }
+  if (!value.ok) {
+    const error = value.error
+    if (!isRecord(error) || typeof error.code !== 'string'
+      || typeof error.message !== 'string' || !isRecord(error.details)) {
+      throw new TypeError('file upload transport returned an invalid failure')
+    }
+    return {
+      ok: false,
+      error: new RemoteError(error.code as never, error.message, error.details as never),
+    }
+  }
+  const result = value.value
+  const file = isRecord(result) ? result.file : undefined
+  if (!isRecord(result) || typeof result.receiptId !== 'string' || !isRecord(file)
+    || typeof file.attachmentId !== 'string' || typeof file.name !== 'string'
+    || typeof file.bytes !== 'number' || !Number.isSafeInteger(file.bytes) || file.bytes < 0) {
+    throw new TypeError('file upload transport returned an invalid receipt')
+  }
+  return {
+    ok: true,
+    value: {
+      receiptId: result.receiptId as SessionUploadFileValue['receiptId'],
+      file: {
+        attachmentId: file.attachmentId as SessionUploadFileValue['file']['attachmentId'],
+        name: file.name,
+        bytes: file.bytes,
+      },
+    },
+  }
+}
+
+function isRecord(value: unknown): value is Record<PropertyKey, unknown> {
+  return typeof value === 'object' && value !== null && !Array.isArray(value)
+}
+
 /** 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[] {
+/** Attachment references in one structurally-read content block list, in block order. */
+function attachmentRefsIn(content: unknown): readonly (ImageAttachmentRef | FileAttachmentRef)[] {
  if (!Array.isArray(content)) return []
-  const refs: ImageAttachmentRef[] = []
+  const refs: Array<ImageAttachmentRef | FileAttachmentRef> = []
  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)
+    if ((candidate.type === 'image' || candidate.type === 'file')
+      && typeof candidate.attachment === 'object' && candidate.attachment !== null) {
+      refs.push(candidate.attachment as ImageAttachmentRef | FileAttachmentRef)
    }
  }
  return refs