feat(session-controller): admit files across command flows

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

@@ -4,12 +4,13 @@ import { randomUUID } from 'node:crypto'
 import type { Context } from '@deepseek-ai/cordis'
 import { brandString } from '@deepseek-ai/dsh-brand'
 import type { Agent, ModelSelection as AgentModelSelection } from '@deepseek-ai/dsh-agent'
-import { AttachmentError, admitPromptContent } from '@deepseek-ai/dsh-attachment'
-import type { ImageAttachmentRef } from '@deepseek-ai/dsh-attachment'
+import { AttachmentError, admitEncodedFile, admitPromptContent } from '@deepseek-ai/dsh-attachment'
+import type { CommandFileReceiptResolver } from '@deepseek-ai/dsh-commands'
+import type { FileAttachmentRef, ImageAttachmentRef } from '@deepseek-ai/dsh-attachment'
 import {
   ReasoningEffortId, createUserMessage, freezeMessage,
 } from '@deepseek-ai/dsh-llm'
-import type { MessageSource } from '@deepseek-ai/dsh-llm'
+import type { ContentBlock, MessageSource } from '@deepseek-ai/dsh-llm'
 import type { SessionEvent, SessionHeader, SessionId, UserMessage } from '@deepseek-ai/dsh-session'
 import { SessionQueryError, type SessionObservation } from '@deepseek-ai/dsh-session-query'
 import { SessionTitleInvalidError } from '@deepseek-ai/dsh-session-title'
@@ -35,6 +36,7 @@ import type {
   SessionCreateValue,
   SessionForkRequest,
   SessionForkValue,
+  FileUploadReceiptId,
   SessionPromptRequest,
   SessionPromptValue,
   SessionRenameRequest,
@@ -43,6 +45,9 @@ import type {
   SessionSelectModelValue,
   SessionUpdateQueueRequest,
   SessionUpdateQueueValue,
+  SessionUploadFileRequest,
+  SessionUploadFileValue,
+  SessionRequestId,
 } from './types.ts'
 
 interface SessionReadState {
@@ -51,8 +56,21 @@ interface SessionReadState {
   readonly events: readonly SessionEvent[]
 }
 
+interface StagedFileUpload {
+  readonly file: FileAttachmentRef
+  /** Prompt that accepted this receipt; absent until successful admission. */
+  requestId?: SessionRequestId
+}
+
 /** Implements Session business commands delegated by the Session Controller Remote service. */
 export class SessionCommandController {
+  /**
+   * Staged file uploads awaiting a prompt, keyed by Session. Entries are the
+   * prompt-time authority for file references: a prompt may only cite a file
+   * previously uploaded for the same Session in this process.
+   */
+  private readonly stagedFiles = new Map<SessionId, Map<FileUploadReceiptId, StagedFileUpload>>()
+
   /**
    * @param ctx - Host context carrying Agent, model, attachment, title, and Workspace services.
    * @param agents - sole owner of create, resume, and Session-local model selection.
@@ -62,7 +80,115 @@ export class SessionCommandController {
     private readonly ctx: Context,
     private readonly agents: ApiSessionAgentController,
     private readonly defaultCwd: string,
-  ) {}
+  ) {
+    ctx.inject(['commands'], (commandCtx) => {
+      const resolve: CommandFileReceiptResolver = (agent, receiptId) =>
+        this.resolveStagedFile(agent.id, receiptId as FileUploadReceiptId)
+      commandCtx.effect(
+        () => commandCtx.commands.registerFileReceiptResolver(resolve),
+        'session-controller: command file receipt resolver',
+      )
+    })
+  }
+
+  /**
+   * Persist one browser file upload verbatim and stage it for later prompts.
+   * @param request - Session identity, base64 payload, and optional display name.
+   * @returns an opaque per-upload receipt and the durable file reference.
+   */
+  async uploadFile(request: SessionUploadFileRequest): Promise<SessionUploadFileValue> {
+    const agent = await this.resolveAgent(request.sessionId)
+    return this.commitFileUpload(agent, async () => admitEncodedFile(this.ctx.attachments, {
+      data: request.data,
+      ...(request.name === undefined ? {} : { name: request.name }),
+    }))
+  }
+
+  /**
+   * Persist raw upload chunks without collecting the complete file in memory.
+   * @param request - Session identity, ordered exact bytes, cancellation, and optional display name.
+   * @returns an opaque per-upload receipt and the durable file reference.
+   */
+  async uploadFileStream(request: {
+    readonly sessionId: SessionId
+    readonly data: AsyncIterable<Uint8Array>
+    readonly signal?: AbortSignal
+    readonly name?: string
+  }): Promise<SessionUploadFileValue> {
+    const agent = await this.resolveAgent(request.sessionId)
+    return this.commitFileUpload(agent, async () => this.ctx.attachments.saveFileStream({
+      data: request.data,
+      ...(request.signal === undefined ? {} : { signal: request.signal }),
+      ...(request.name === undefined ? {} : { name: request.name }),
+    }))
+  }
+
+  private async commitFileUpload(
+    agent: Agent,
+    save: () => Promise<FileAttachmentRef>,
+  ): Promise<SessionUploadFileValue> {
+    let file: FileAttachmentRef
+    try {
+      file = await save()
+    } catch (error) {
+      if (error instanceof AttachmentError) {
+        throw new RemoteError('session/attachment-invalid', error.message, { reason: error.code })
+      }
+      throw new RemoteError(
+        'gateway/internal',
+        `failed to store file upload: ${String(error)}`,
+        {},
+        { cause: error },
+      )
+    }
+    if (this.ctx.agents.get(agent.id) !== agent) {
+      throw new RemoteError(
+        'session/not-found',
+        `session "${agent.id}" was disposed before its file upload completed`,
+        { sessionId: agent.id },
+      )
+    }
+    let staged = this.stagedFiles.get(agent.id)
+    if (staged === undefined) {
+      staged = new Map()
+      this.stagedFiles.set(agent.id, staged)
+    }
+    const receiptId = randomUUID() as FileUploadReceiptId
+    staged.set(receiptId, { file })
+    return { receiptId, file }
+  }
+
+  /**
+   * Resolve one staged upload for the same Session without exposing the receipt table.
+   * @param sessionId - receiving Session identity.
+   * @param receiptId - Host-minted upload receipt.
+   * @returns the durable file reference, or `undefined` when the receipt is absent or belongs elsewhere.
+   */
+  resolveStagedFile(sessionId: SessionId, receiptId: FileUploadReceiptId): FileAttachmentRef | undefined {
+    return this.stagedFiles.get(sessionId)?.get(receiptId)?.file
+  }
+
+  /**
+   * Retire file receipts only after their accepted prompt becomes observable.
+   * @param sessionId - Session whose log emitted the prompt.
+   * @param requestId - browser prompt identity echoed by the event.
+   */
+  retireObservedPrompt(sessionId: SessionId, requestId: SessionRequestId): void {
+    const staged = this.stagedFiles.get(sessionId)
+    if (staged === undefined) return
+    for (const [receiptId, upload] of staged) {
+      if (upload.requestId === requestId) staged.delete(receiptId)
+    }
+    if (staged.size === 0) this.stagedFiles.delete(sessionId)
+  }
+
+  /**
+   * Drop one Session's staged uploads (the stored objects remain durable).
+   * @param sessionId - Session leaving the live registry.
+   */
+  releaseStagedFiles(sessionId: SessionId): void {
+    this.stagedFiles.delete(sessionId)
+  }
 
   /**
    * Create or idempotently adopt one ordinary Session.
@@ -292,6 +418,7 @@ export class SessionCommandController {
       )
     }
     const agent = await this.resolveAgent(request.sessionId)
+    if (hasPromptRequest(agent, request.requestId)) return { accepted: true }
     const selection = this.agents.selectionFor(agent).current
     if (!routeServed(this.ctx, selection.provider)) {
       throw new RemoteError(
@@ -319,10 +446,42 @@ export class SessionCommandController {
             )
           }
         }
-        const content = await admitPromptContent(this.ctx.attachments, request.content)
-        const message: UserMessage = createUserMessage({ content, source })
-        if (request.mode === 'steer') agent.steer(message)
-        else agent.followup(message)
+        const staged = this.stagedFiles.get(request.sessionId)
+        const durable = await durablePromptContent(
+          this.ctx,
+          request.content,
+          receiptId => staged?.get(receiptId)?.file,
+        )
+        const message: UserMessage = createUserMessage({ content: durable.content, source })
+        if (this.ctx.agents.get(agent.id) !== agent) {
+          throw new RemoteError(
+            'session/not-found',
+            `session "${agent.id}" was disposed during prompt admission`,
+            { sessionId: agent.id },
+          )
+        }
+        const bound = durable.receiptIds.map((receiptId) => {
+          const upload = staged?.get(receiptId)
+          if (upload === undefined) {
+            throw new RemoteError(
+              'session/attachment-invalid',
+              'File was not uploaded for this session.',
+              { reason: 'FILE_NOT_STAGED' },
+            )
+          }
+          return { upload, previous: upload.requestId }
+        })
+        for (const { upload } of bound) upload.requestId = request.requestId
+        try {
+          if (request.mode === 'steer') agent.steer(message)
+          else agent.followup(message)
+        } catch (error) {
+          for (const { upload, previous } of bound) {
+            if (previous === undefined) delete upload.requestId
+            else upload.requestId = previous
+          }
+          throw error
+        }
       } catch (error) {
         if (remoteErrorOf(error) !== undefined) throw error
         if (error instanceof AttachmentError) {
@@ -416,6 +575,12 @@ export class SessionCommandController {
       }))
     } else {
       agent.inbox.remove(request.itemId)
+      if (request.action.kind === 'remove') {
+        const source = message.source
+        if (source.kind === 'user' && 'rpcId' in source) {
+          this.retireObservedPrompt(request.sessionId, source.rpcId)
+        }
+      }
       if (request.action.kind === 'steer') agent.steer(message)
     }
     return { accepted: true }
@@ -492,6 +657,51 @@ export class SessionCommandController {
   }
 }
 
+async function durablePromptContent(
+  ctx: Context,
+  content: readonly SessionPromptRequest['content'][number][],
+  stagedFile: (receiptId: FileUploadReceiptId) => FileAttachmentRef | undefined,
+): Promise<{ readonly content: ContentBlock[]; readonly receiptIds: readonly FileUploadReceiptId[] }> {
+  const files = new Map<FileUploadReceiptId, FileAttachmentRef>()
+  for (const part of content) {
+    if (part.type !== 'file' || files.has(part.receiptId)) continue
+    const file = stagedFile(part.receiptId)
+    if (file === undefined) {
+      throw new RemoteError(
+        'session/attachment-invalid',
+        'File was not uploaded for this session.',
+        { reason: 'FILE_NOT_STAGED' },
+      )
+    }
+    files.set(part.receiptId, file)
+  }
+  type NonFilePart = Exclude<SessionPromptRequest['content'][number], { readonly type: 'file' }>
+  const admitted = await admitPromptContent(
+    ctx.attachments,
+    content.filter((part): part is NonFilePart => part.type !== 'file'),
+  )
+  let next = 0
+  const durable = content.map((part) => {
+    if (part.type === 'file') {
+      return { type: 'file' as const, attachment: files.get(part.receiptId) as FileAttachmentRef }
+    }
+    return admitted[next++] as ContentBlock
+  })
+  return { content: durable, receiptIds: [...files.keys()] }
+}
+
+function hasPromptRequest(agent: Agent, requestId: SessionRequestId): boolean {
+  const matches = (message: UserMessage): boolean => {
+    const source = message.source
+    return source.kind === 'user' && 'rpcId' in source && source.rpcId === requestId
+  }
+  if (agent.inbox.nextTurn.some(matches) || agent.inbox.nextStep.some(matches)) return true
+  return agent.session.snapshotEvents().some((event) => {
+    if (event.type !== 'user/message') return false
+    const source = event.data.source
+    return source.kind === 'user' && 'rpcId' in source && source.rpcId === requestId
+  })
+}
 function imageBlockIn(
   content: unknown,
   match: (ref: ImageAttachmentRef) => boolean,

packages/api/session-controller/src/file-upload-http.ts

@@ -0,0 +1,116 @@
+/** Raw Fetch file intake used by the browser background-upload carrier. */
+
+import type { Context } from '@deepseek-ai/cordis'
+import { SessionId } from '@deepseek-ai/dsh-session'
+import { remoteErrorOf } from '@deepseek-ai/dsh-typert-protocol'
+import { SESSION_FILE_UPLOAD_PATH } from './file-upload-path.ts'
+import type { SessionUploadFileValue } from './types.ts'
+import type { SessionCommandController } from './commands.ts'
+
+interface FileUploadConnection {
+  readonly fetch: {
+    register(route: {
+      readonly path: string
+      readonly methods: readonly ['POST']
+      readonly requestBody: 'streaming'
+      readonly fetch: (request: Request) => Promise<Response>
+    }): () => Promise<void>
+  }
+}
+
+type FileUploadResult =
+  | { readonly ok: true; readonly value: SessionUploadFileValue }
+  | {
+    readonly ok: false
+    readonly error: { readonly code: string; readonly message: string; readonly details: object }
+  }
+
+/**
+ * Install the authenticated raw-byte route on Connection's shared Fetch registry.
+ * @param ctx - Host context that provides Connection.
+ * @param commands - Session command owner that validates and stages stored bytes.
+ */
+export function registerSessionFileUploadHttp(ctx: Context, commands: SessionCommandController): void {
+  ctx.inject(['connection'], (connectionCtx) => {
+    const connection = connectionCtx.get('connection') as FileUploadConnection
+    connection.fetch.register({
+      path: SESSION_FILE_UPLOAD_PATH,
+      methods: ['POST'],
+      requestBody: 'streaming',
+      fetch: request => handleSessionFileUploadHttp(commands, request),
+    })
+  })
+}
+
+/**
+ * Handle one authenticated raw-byte upload after the physical carrier applies
+ * its trust policy.
+ * @param commands - Session command owner that validates and stages stored bytes.
+ * @param request - Fetch request carrying the raw file body.
+ * @returns JSON receipt or a precise validation response.
+ */
+export async function handleSessionFileUploadHttp(
+  commands: SessionCommandController,
+  request: Request,
+): Promise<Response> {
+  if (request.method !== 'POST') {
+    return new Response(null, { status: 405, headers: { allow: 'POST' } })
+  }
+  const mediaType = request.headers.get('content-type')?.split(';', 1)[0]?.trim().toLowerCase()
+  if (mediaType !== 'application/octet-stream') {
+    return new Response('content type must be application/octet-stream', { status: 415 })
+  }
+  const url = new URL(request.url)
+  const sessionId = url.searchParams.get('sessionId')
+  if (sessionId === null || sessionId === '') {
+    return new Response('sessionId is required', { status: 400 })
+  }
+  const name = url.searchParams.get('name') ?? undefined
+  let result: FileUploadResult
+  try {
+    result = {
+      ok: true,
+      value: await commands.uploadFileStream({
+        sessionId: SessionId(sessionId),
+        data: requestBodyChunks(request.body),
+        signal: request.signal,
+        ...(name === undefined ? {} : { name }),
+      }),
+    }
+  } catch (error) {
+    const failure = remoteErrorOf(error)
+    result = {
+      ok: false,
+      error: failure !== undefined
+        ? { code: failure.code, message: failure.message, details: failure.details }
+        : {
+          code: 'gateway/internal',
+          message: error instanceof Error ? error.message : String(error),
+          details: {},
+        },
+    }
+  }
+  return new Response(JSON.stringify(result), {
+    status: 200,
+    headers: {
+      'content-type': 'application/json; charset=utf-8',
+      'cache-control': 'no-store',
+    },
+  })
+}
+
+async function* requestBodyChunks(
+  body: ReadableStream<Uint8Array> | null,
+): AsyncIterable<Uint8Array> {
+  if (body === null) return
+  const reader = body.getReader()
+  try {
+    while (true) {
+      const chunk = await reader.read()
+      if (chunk.done) return
+      yield chunk.value
+    }
+  } finally {
+    reader.releaseLock()
+  }
+}

packages/api/session-controller/src/file-upload-path.ts

@@ -0,0 +1,2 @@
+/** Authenticated raw-byte upload route shared by Host and Client faces. */
+export const SESSION_FILE_UPLOAD_PATH = '/api/session/uploadFileBinary'

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

@@ -16,6 +16,7 @@ import { SessionCommandController } from './commands.ts'
 import { SessionControlController } from './control.ts'
 import { SessionHistoryController } from './history.ts'
 import { SessionFileReferences } from './file-references.ts'
+import { registerSessionFileUploadHttp } from './file-upload-http.ts'
 import { ApiSessionList, DEFAULT_COLD_BLANK_PROBE_MAX_BYTES } from './list.ts'
 import { buildModelCatalog } from './catalog.ts'
 import { installModelSelectionProjection } from './model-selection-projection.ts'
@@ -49,6 +50,8 @@ import type {
   SessionSelectModelValue,
   SessionUpdateQueueRequest,
   SessionUpdateQueueValue,
+  SessionUploadFileRequest,
+  SessionUploadFileValue,
 } from './types.ts'
 
 export type * from './types.ts'
@@ -116,6 +119,7 @@ export class SessionController extends TypertRemoteService {
     installModelSelectionProjection(ctx)
     this.agents = new ApiSessionAgentController(ctx)
     this.commands = new SessionCommandController(ctx, this.agents, process.cwd())
+    registerSessionFileUploadHttp(ctx, this.commands)
     this.controlState = new SessionControlController(ctx)
     // Registered before history so reverse-order teardown closes every
     // follower before waiting for already-admitted promotions.
@@ -137,6 +141,7 @@ export class SessionController extends TypertRemoteService {
       ctx.emit('api-session/added', this.listState.summaryFor(session))
     })
     ctx.on('session/disposed', (session) => {
+      this.commands.releaseStagedFiles(session.id)
       ctx.emit('api-session/removed', session.id)
     })
     ctx.on('agent/status', ({ agent, status }) => {
@@ -146,6 +151,10 @@ export class SessionController extends TypertRemoteService {
       ctx.emit('api-session/error', agent.id, errorChain(error))
     })
     ctx.on('session/event', (session, event) => {
+      if (event.type === 'user/message' && event.data.source.kind === 'user'
+        && 'rpcId' in event.data.source) {
+        this.commands.retireObservedPrompt(session.id, event.data.source.rpcId)
+      }
       if (event.type === 'request/header') {
         const agent = ctx.agents.get(session.id)
         if (agent?.session === session) this.agents.consumeSelection(
@@ -334,6 +343,19 @@ export class SessionController extends TypertRemoteService {
     return this.commands.attachment(request)
   }
 
+  /**
+   * Persist one encoded file upload verbatim and stage it for a later prompt
+   * on the same Session.
+   * @param request - Session identity, base64 payload, and optional display name.
+   * @param signal - caller cancellation before storage begins.
+   * @returns an opaque per-upload receipt and the durable file reference.
+   */
+  @Remote('uploadFile')
+  uploadFile(request: SessionUploadFileRequest, signal: AbortSignal): Promise<SessionUploadFileValue> {
+    signal.throwIfAborted()
+    return this.commands.uploadFile(request)
+  }
+
   /**
    * Mutate one still-pending queue occurrence on a live Agent.
    * @param request - Session, queue item, and requested mutation.

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

@@ -1,7 +1,7 @@
 /** Browser-safe request, result, and lifecycle vocabulary for the Session Remote service. */
 
 import type {
-  AttachmentIdType, ImageAttachmentLimits, ImageAttachmentRef, ImageMediaType,
+  AttachmentIdType, FileAttachmentRef, ImageAttachmentLimits, ImageAttachmentRef, ImageMediaType,
 } from '@deepseek-ai/dsh-attachment'
 import type { Branded } from '@deepseek-ai/dsh-brand'
 import type { MessageId } from '@deepseek-ai/dsh-llm/brand'
@@ -68,7 +68,11 @@ export interface SessionProjectionBaseline {
 export type SessionProjectionValues = Partial<SessionProjectionMap>
   & Readonly<Record<string, SessionProjectionValue>>
 
-/** Browser-submitted prompt content; the Host promotes image bytes to durable references. */
+/**
+ * Browser-submitted prompt content; the Host promotes image bytes to durable
+ * references. File parts carry the opaque receipt returned by a preceding
+ * `uploadFile` call on the same Session.
+ */
 export type PromptContentPart =
   | { readonly type: 'text'; readonly text: string }
   | {
@@ -77,6 +81,7 @@ export type PromptContentPart =
     readonly data: string
     readonly name?: string
   }
+  | { readonly type: 'file'; readonly receiptId: FileUploadReceiptId }
 
 /** Complete model selection for one Session. */
 export interface ModelSelection {
@@ -313,6 +318,22 @@ export interface SessionPromptValue {
   readonly accepted: true
 }
 
+/** One base64 file upload staged for a later prompt on the same Session. */
+export interface SessionUploadFileRequest {
+  readonly sessionId: SessionId
+  /** Canonical base64 encoding of the exact file bytes. */
+  readonly data: string
+  /** Optional display name; the Host sanitizes it into the stored leaf name. */
+  readonly name?: string
+}
+
+/** Durable receipt for one staged file upload. */
+export interface SessionUploadFileValue {
+  /** Per-upload authority consumed by a later prompt on the same Session. */
+  readonly receiptId: FileUploadReceiptId
+  readonly file: FileAttachmentRef
+}
+
 /** Durable image read request. */
 export interface SessionAttachmentRequest {
   readonly sessionId: SessionId
@@ -361,6 +382,9 @@ export interface SessionOpenWorkspacePathValue {
 /** Client-minted prompt identity used to reconcile optimistic and durable messages. */
 export type SessionRequestId = Branded<'session-request-id'>
 
+/** Host-minted authority for one staged file upload on one Session. */
+export type FileUploadReceiptId = Branded<'file-upload-receipt-id'>
+
 declare module '@deepseek-ai/dsh-llm' {
   interface MessageSourceMap {
     /** Browser prompt correlation and optional Host-validated time zone. */