refactor(client): extract background file upload service

2026-09-13 04:03:30 +00:00 · 2026-09-03 13:53:25 +08:00
parent a1956fa3e2
commit bbb2ca7c9b
61 changed files with 1225 additions and 500 deletions
@@ -2,5 +2,5 @@
 # side as of the last confirmed-consistent state. Both languages carry equal authority;
 # after editing either side, bring the other along and re-record with:
 #   pnpm run verify-translation-pairing --write packages/api/session-controller/README.md
-README.md: 4f5d41ae34690e74a880c57434f99f06a12b23b8
-README.zh.md: 2404da3973e36514fdf27a78055403baffe2a478
+README.md: ab4001052306be1b052d1ca6a0e8ff8ddeffd459
+README.zh.md: 5100959e826e1099007a85d34b345a7d7d8481e8
@@ -27,7 +27,7 @@ History pages and follow opening snapshots carry a discriminated `SessionHistory

 Each endpoint states its activation policy. List, search, attachment, history pages, log following, skill discovery, and workspace-path opening can inspect persistence without activating an Agent; `canOpenWorkspacePath()` reports native-opening availability without addressing a Session. Queue mutation and cancellation require live state; model, rename, prompt, file-reference, and file-upload operations may resolve or resume an ordinary Session. A generated Remote accepts base64 file bytes for non-browser carriers, while an authenticated exact Fetch route streams raw chunks from the browser background-upload carrier without entering Connection's buffered JSON path. The streaming path hashes and writes each chunk with backpressure, publishes the completed object atomically, and removes its staging file on cancellation or failure. Both paths store the file verbatim and stage it in this process for the same exact Agent; completion after that Agent is disposed stores no receipt. A later prompt or attachment-accepting command cites the opaque per-upload receipt. Both admission paths validate every same-Session receipt before persisting images. Prompt retries whose `requestId` is already queued or logged return the original acceptance without inserting another message; prompt receipts retire when that `rpcId` becomes a user event or its queued occurrence is removed, while command-only receipts remain until Session disposal. Create and fork are the only operations that create a new Agent directly. The skill catalog instead uses a live Agent when present or the recorded preset's standing scope when cold, so listing never starts an Agent.

-The Client adapter exposes `SessionEventStream`, a Gateway `RemoteJournalStream` bound to one ordinary or direct-subagent address. It opens follow before the initial page, publishes only contiguous `replace`, `prepend`, and `append` changes, and repairs reconnect or sequence gaps through a tail page. Backwards paging has two verbs: `loadOlder()` pulls one 50-message page, and `loadThrough(seq)` — the turn-jump loader — loops 200-message pages until the window covers the target seq, lowering a shared target on repeated calls, stopping on a page that makes no progress, and reporting busy through the same `loadingOlder` snapshot bit. Ordinary records cover `[event.seq, event.seq]`; packed rows cover `[event.seq, event.seq + memberCount - 1]`. A business, persistence, or unresolved continuity failure terminates the stream, while only physical carrier loss selects automatic resumption. `SessionControlStream` is a Gateway `RemoteSnapshotStream`; every generation opens with a complete process-local baseline, so reconnect replaces queue, jobs, and projection state instead of treating transient values as durable events.
+The Client adapter exposes `SessionEventStream`, a Gateway `RemoteJournalStream` bound to one ordinary or direct-subagent address. It opens follow before the initial page, publishes only contiguous `replace`, `prepend`, and `append` changes, and repairs reconnect or sequence gaps through a tail page. Backwards paging has two verbs: `loadOlder()` pulls one 50-message page, and `loadThrough(seq)` — the turn-jump loader — loops 200-message pages until the window covers the target seq, lowering a shared target on repeated calls, stopping on a page that makes no progress, and reporting busy through the same `loadingOlder` snapshot bit. Ordinary records cover `[event.seq, event.seq]`; packed rows cover `[event.seq, event.seq + memberCount - 1]`. A business, persistence, or unresolved continuity failure terminates the stream, while only physical carrier loss selects automatic resumption. `SessionControlStream` is a Gateway `RemoteSnapshotStream`; every generation opens with a complete process-local baseline, so reconnect replaces queue, jobs, and projection state instead of treating transient values as durable events. A Client Session resolves the independent [`fileUpload`](../../client/file-upload/README.md) service from its Agent scope for `Blob` and one-shot `ReadableStream<Uint8Array>` bodies. A fixture or bare Session keeps the generated Remote fallback for `Blob` and exact-byte inputs; a stream requires the background service because it cannot be replayed through the base64 Remote.

 The Session object also carries local submission echoes: `session.beginSubmission` inserts one into `SessionSnapshot.pendingSubmissions` synchronously, before the caller serializes and prompts, so a conversation UI can show the message on the submit click's own frame. The echo stores ordered image previews and durable file references. Session derives its `transcript`, `queued`, or `steering` placement from the current running state and requested delivery mode, then retains that placement while serialization is in flight. The prompt's `requestId` is the correlation identity: the Host echoes it as the durable user source's `rpcId`, and queue occurrences project it as `SessionQueuedItem.rpcId`. An echo retires one animation frame after its durable event or queue occurrence is observed, immediately when its identified prompt fails or is abandoned, and as failed on disposal. Each retirement fires `onRetire` exactly once; an observed retirement includes the ordered durable attachment references so the composer can release successful cards while preserving failed drafts. Echoes are Client memory only; reload and reconnect rebuild the conversation from durable events alone.

@@ -27,7 +27,7 @@ kind: "package-reference"

 每个 endpoint 都声明自己的激活策略。列表、搜索、附件、历史页、日志跟随、skill 发现和工作区路径打开可以在不激活 Agent 的情况下检查 persistence；`canOpenWorkspacePath()` 无需指定 Session 即可报告原生打开能力。queue 变更与取消要求 live 状态；模型、重命名、prompt、文件引用和文件上传操作可以解析或恢复普通 Session。生成 Remote 为非浏览器载体接收 base64 文件字节，经过认证的精确 Fetch 路由则从浏览器后台上传载体流式接收原始分块，不进入 Connection 的缓冲 JSON 路径。流式路径以背压逐块计算摘要和写入，完整接收后以原子方式发布对象，并在取消或失败时移除暂存文件。两条路径都按字节原样存储文件，并在本进程内按同一个准确 Agent 暂存；如果保存完成时该 Agent 已销毁，就不发布凭证。之后的 prompt 或接受附件的命令引用每次上传独有的不透明凭证。两条准入路径都先验证所有凭证属于同一 Session，再持久化图片。`requestId` 已进入 queue 或日志时，prompt 重试直接返回原来的接受结果，不会重复插入消息；prompt 凭证在同一 `rpcId` 进入 user event 或对应 queue occurrence 被移除后退休，只被命令使用的凭证则保留到 Session 销毁。只有 create 与 fork 会直接创建新 Agent。skill 目录则优先使用已有 live Agent，否则使用所记录 preset 的常驻 scope，因此列表查询绝不会启动 Agent。

-Client adapter 提供 `SessionEventStream`，即绑定到一个普通 Session 或 direct subagent address 的 Gateway `RemoteJournalStream`。它在读取首个 page 前打开 follow，只发布连续的 `replace`、`prepend` 和 `append` 变更，并通过 tail page 修复重连或 seq 缺口。向后分页有两个动词：`loadOlder()` 拉一页 50 条 message，而 `loadThrough(seq)`——轮次跳转加载器——按 200 条 message 一页循环拉取直到窗口覆盖目标 seq，重复调用会下调共享目标，遇到无进展的页即停止，忙碌状态复用同一个 `loadingOlder` 快照位。普通 record 覆盖 `[event.seq, event.seq]`，packed row 覆盖 `[event.seq, event.seq + memberCount - 1]`。业务、persistence 或无法恢复的连续性错误会终止 stream，只有物理载体断开才触发自动恢复。`SessionControlStream` 是 Gateway `RemoteSnapshotStream`；每代都以完整的进程本地 baseline 开始，因此重连会替换 queue、jobs 和 projection 状态，而不会把瞬态值当作 durable event。
+Client adapter 提供 `SessionEventStream`，即绑定到一个普通 Session 或 direct subagent address 的 Gateway `RemoteJournalStream`。它在读取首个 page 前打开 follow，只发布连续的 `replace`、`prepend` 和 `append` 变更，并通过 tail page 修复重连或 seq 缺口。向后分页有两个动词：`loadOlder()` 拉一页 50 条 message，而 `loadThrough(seq)`——轮次跳转加载器——按 200 条 message 一页循环拉取直到窗口覆盖目标 seq，重复调用会下调共享目标，遇到无进展的页即停止，忙碌状态复用同一个 `loadingOlder` 快照位。普通 record 覆盖 `[event.seq, event.seq]`，packed row 覆盖 `[event.seq, event.seq + memberCount - 1]`。业务、persistence 或无法恢复的连续性错误会终止 stream，只有物理载体断开才触发自动恢复。`SessionControlStream` 是 Gateway `RemoteSnapshotStream`；每代都以完整的进程本地 baseline 开始，因此重连会替换 queue、jobs 和 projection 状态，而不会把瞬态值当作 durable event。Client Session 从自己的 Agent scope 取得独立的 [`fileUpload`](../../client/file-upload/README.zh.md) 服务，用于 `Blob` 与只能使用一次的 `ReadableStream<Uint8Array>` 请求体。fixture 或未绑定的 Session 会让 `Blob` 与精确字节输入继续使用生成的 Remote；stream 无法通过 base64 Remote 重放，因此必须有后台服务。

 Session 对象还承载本地提交回显：`session.beginSubmission` 在调用方序列化与 prompt 之前，同步把一条回显写入 `SessionSnapshot.pendingSubmissions`，会话 UI 因此能在点击提交的当帧显示消息。回显按顺序存放图片预览与持久文件引用。Session 根据当前运行状态与请求的投递模式推导其 `transcript`、`queued` 或 `steering` 位置，并在序列化期间保留该位置。prompt 的 `requestId` 是关联标识：Host 把它回显为 durable user source 的 `rpcId`，queue occurrence 也把它投影为 `SessionQueuedItem.rpcId`。回显在观察到其 durable event 或 queue occurrence 后延迟一个动画帧退休，带标识的 prompt 失败或被放弃时立即退休，销毁时按 failed 退休。每次退休恰好触发一次 `onRetire`；observed 退休还会携带有序的持久附件引用，让 composer 释放成功卡片并保留失败草稿。回显只存在于 Client 内存；刷新与重连只从 durable event 重建会话。

@@ -47,7 +47,8 @@
        "@deepseek-ai/dsh-api-gateway/client"
      ],
      "inject": [
-        "@deepseek-ai/dsh-api-gateway"
+        "@deepseek-ai/dsh-api-gateway",
+        "@deepseek-ai/dsh-client-file-upload"
      ],
      "platform": "web"
    }
@@ -120,6 +121,7 @@
    "@deepseek-ai/dsh-api-gateway": "workspace:^",
    "@deepseek-ai/dsh-attachment": "workspace:^",
    "@deepseek-ai/dsh-client-connection": "workspace:^",
+    "@deepseek-ai/dsh-client-file-upload": "workspace:^",
    "@deepseek-ai/dsh-commands": "workspace:^",
    "@deepseek-ai/dsh-client-store": "workspace:^",
    "@deepseek-ai/dsh-file-reference": "workspace:^",
@@ -92,14 +92,14 @@ export interface ISession {
  /**
   * 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 data - browser Blob, exact file bytes, or a one-shot byte stream.
   * @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.
+   * @param onProgress - optional byte-progress observer for background Blob or stream uploads.
   * @returns the staged-upload receipt and durable file reference, or the business error.
   */
  uploadFile(
-    data: Blob | Uint8Array,
+    data: Blob | Uint8Array | ReadableStream<Uint8Array>,
    name?: string,
    signal?: AbortSignal,
    onProgress?: (progress: { readonly loaded: number; readonly total?: number }) => void,
@@ -31,10 +31,22 @@ export interface PendingSubmissionImage {
  readonly height?: number
 }

+/** Image branch of a local submission echo attachment. */
+export interface PendingSubmissionImageAttachment {
+  readonly type: 'image'
+  readonly value: PendingSubmissionImage
+}
+
+/** File branch of a local submission echo attachment. */
+export interface PendingSubmissionFileAttachment {
+  readonly type: 'file'
+  readonly value: FileAttachmentRef
+}
+
 /** One attachment displayed by a local submission echo, in prompt order. */
 export type PendingSubmissionAttachment =
-  | ({ readonly type: 'image' } & PendingSubmissionImage)
-  | { readonly type: 'file'; readonly attachment: FileAttachmentRef }
+  | PendingSubmissionImageAttachment
+  | PendingSubmissionFileAttachment

 /** Client surface selected when a local submission begins. */
 export type PendingSubmissionPlacement = 'transcript' | 'queued' | 'steering'
@@ -2,7 +2,8 @@

 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 type {} from '@deepseek-ai/dsh-client-connection/client'
+import type {} from '@deepseek-ai/dsh-client-file-upload/client'
 import { createSessionControlStream } from './transport.ts'
 import { ClientSessions } from './sessions/service.ts'
 import type { SessionRemotes } from './sessions/remotes.ts'
@@ -61,7 +62,9 @@ export type {
  OpenState,
  PendingSubmission,
  PendingSubmissionAttachment,
+  PendingSubmissionFileAttachment,
  PendingSubmissionImage,
+  PendingSubmissionImageAttachment,
  PendingSubmissionPlacement,
  PromptError,
  QueuedMessage,
@@ -78,6 +81,7 @@ declare module '@deepseek-ai/cordis' {
 /** Required Remote and Context projection services. */
 export const inject = [
  'connection',
+  'fileUpload',
  'typert',
  'remote',
  'remote.commands',
@@ -90,9 +94,8 @@ 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, connection.backgroundUploads)
+  const sessions = new ClientSessions(ctx, remotes)
  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,7 +25,6 @@ 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'

 function sessionSeqCursor(value: number): SessionSeqCursor {
  return value === -1 ? -1 : SessionSeq(value)
@@ -154,7 +153,6 @@ 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)
@@ -334,7 +332,6 @@ export class SessionManager {
        this.recordMutation({ kind: 'engaged', sessionId: engaged.sessionId })
      },
      projections: this.projectionStore(sessionId),
-      ...(this.backgroundUploads === undefined ? {} : { backgroundUploads: this.backgroundUploads }),
    })
  }

@@ -32,7 +32,6 @@ 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'

@@ -222,7 +221,6 @@ export class ClientSessions implements ISessions {
  constructor(
    private readonly rootCtx: Context,
    remote: SessionRemotes,
-    backgroundUploads?: BackgroundUploadTransport,
  ) {
    this.selection = createSnapshotStore<SessionSelection>(
      {},
@@ -232,7 +230,6 @@ export class ClientSessions implements ISessions {
      remote,
      restored.sessionId,
      restored.subagentAddress,
-      backgroundUploads,
    )
    this.list = createSnapshotStore<SessionListState>({
      ids: [], byId: {}, current: undefined, phase: 'pending',
@@ -2,9 +2,7 @@

 import type { Context } from '@deepseek-ai/cordis'
 import { bytesToBase64, randomUUID } from '@deepseek-ai/dsh-util-crypto'
-import type {
-  BackgroundUploadProgress, BackgroundUploadTransport,
-} from '@deepseek-ai/dsh-client-connection/client'
+import type { FileUploadProgress } from '@deepseek-ai/dsh-client-file-upload/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'
@@ -76,8 +74,6 @@ export interface SessionOptions {
   * private store (bare object-layer construction).
   */
  projections?: ProjectionValueStore
-  /** Physical large-body carrier supplied by the active browser Connection. */
-  backgroundUploads?: BackgroundUploadTransport
 }

 /**
@@ -302,15 +298,17 @@ export class Session implements SessionFace {
  /**
   * 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 data - exact bytes, a browser Blob, or a one-shot byte stream.
   * @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 bodies.
   * @returns the staged-upload receipt and durable file reference, or the business error.
   */
  async uploadFile(
-    data: Blob | Uint8Array,
+    data: Blob | Uint8Array | ReadableStream<Uint8Array>,
    name?: string,
    signal?: AbortSignal,
-    onProgress?: (progress: BackgroundUploadProgress) => void,
+    onProgress?: (progress: FileUploadProgress) => void,
  ): Promise<RemoteResult<SessionUploadFileValue>> {
    if (this.address !== undefined) {
      return {
@@ -322,10 +320,10 @@ export class Session implements SessionFace {
        ),
      }
    }
-    if (data instanceof Blob && this.options.backgroundUploads !== undefined) {
+    if (!(data instanceof Uint8Array) && this.actx?.fileUpload.available === true) {
      const query = new URLSearchParams({ sessionId: this.sessionId })
      if (name !== undefined) query.set('name', name)
-      const response = await this.options.backgroundUploads.post({
+      const response = await this.actx.fileUpload.post({
        path: `${SESSION_FILE_UPLOAD_PATH}?${query.toString()}`,
        body: data,
        headers: { 'content-type': 'application/octet-stream' },
@@ -337,6 +335,9 @@ export class Session implements SessionFace {
      }
      return parseFileUploadResult(response.body)
    }
+    if (!(data instanceof Uint8Array) && !(data instanceof Blob)) {
+      throw new Error('stream file upload requires a bound Client file-upload service')
+    }
    const bytes = data instanceof Uint8Array ? data : new Uint8Array(await data.arrayBuffer())
    return this.remote.session.uploadFile({
      sessionId: this.sessionId,
@@ -84,14 +84,18 @@ describe('beginSubmission', () => {
    const handle = session.beginSubmission({
      mode: 'queue',
      text: '你好',
-      attachments: [{ type: 'image', previewUrl: 'blob:p1', name: 'a.png', width: 4, height: 3 }],
+      attachments: [{
+        type: 'image', value: { previewUrl: 'blob:p1', name: 'a.png', width: 4, height: 3 },
+      }],
    })
    expect(session.getSnapshot().promptAttempted).toBe(true)
    expect(session.getSnapshot().pendingSubmissions).toMatchObject([{
      requestId: handle.requestId,
      placement: 'transcript',
      text: '你好',
-      attachments: [{ type: 'image', previewUrl: 'blob:p1', name: 'a.png', width: 4, height: 3 }],
+      attachments: [{
+        type: 'image', value: { previewUrl: 'blob:p1', name: 'a.png', width: 4, height: 3 },
+      }],
    }])
  })

@@ -168,7 +172,7 @@ describe('observed retirement', () => {
    const handle = session.beginSubmission({
      mode: 'queue',
      text: '发送',
-      attachments: [{ type: 'image', previewUrl: 'blob:p1' }],
+      attachments: [{ type: 'image', value: { previewUrl: 'blob:p1' } }],
      onRetire: retirement => retirements.push(retirement),
    })
    const refs = [imageRef('att-1')]
@@ -188,7 +192,7 @@ describe('observed retirement', () => {
    const handle = session.beginSubmission({
      mode: 'queue',
      text: '排队',
-      attachments: [{ type: 'image', previewUrl: 'blob:p1' }],
+      attachments: [{ type: 'image', value: { previewUrl: 'blob:p1' } }],
      onRetire: retirement => retirements.push(retirement),
    })
    const refs = [imageRef('att-q')]
@@ -210,9 +214,9 @@ describe('observed retirement', () => {
      mode: 'queue',
      text: 'mixed',
      attachments: [
-        { type: 'image', previewUrl: 'blob:first' },
-        { type: 'file', attachment: file },
-        { type: 'image', previewUrl: 'blob:last' },
+        { type: 'image', value: { previewUrl: 'blob:first' } },
+        { type: 'file', value: file },
+        { type: 'image', value: { previewUrl: 'blob:last' } },
      ],
      onRetire: retirement => retirements.push(retirement),
    })
@@ -1,6 +1,8 @@
 /** Session object lifecycle, event-window transport, commands, and resync behavior. */

 import { afterEach, describe, expect, it, vi } from 'vitest'
+import { Context } from '@deepseek-ai/cordis'
+import type { FileUploadService } from '@deepseek-ai/dsh-client-file-upload/client'
 import { SessionSeq, type SessionEvent } from '@deepseek-ai/dsh-session/types'
 import type { SessionId } from '@deepseek-ai/dsh-api-remotes/client'
 import { RemoteStreamCarrierError } from '@deepseek-ai/dsh-api-gateway/client'
@@ -23,6 +25,16 @@ function makeSession(
  return { api, session: new Session(SID, fakeRemote(api), options) }
 }

+function bindFileUpload(
+  session: Session,
+  post: FileUploadService['post'],
+  available = true,
+): void {
+  const ctx = new Context()
+  ctx.reflect.provide('fileUpload', { available, post })
+  session.bindScope(ctx)
+}
+
 function follow(
  api: FakeApiClient,
  event: SessionEvent,
@@ -50,7 +62,7 @@ describe('Session file upload', () => {
    const progress = vi.fn()
    const post = vi.fn(async (request: {
      path: string
-      body: Blob
+      body: Blob | ReadableStream<Uint8Array>
      headers?: Readonly<Record<string, string>>
      signal?: AbortSignal
      onProgress?: (progress: { loaded: number; total?: number }) => void
@@ -67,7 +79,8 @@ describe('Session file upload', () => {
        }),
      }
    })
-    const { api, session } = makeSession(undefined, { backgroundUploads: { post } })
+    const { api, session } = makeSession()
+    bindFileUpload(session, post)
    const abort = new AbortController()
    const file = new Blob([Uint8Array.of(1, 2, 3, 4)])

@@ -96,7 +109,8 @@ describe('Session file upload', () => {
        error: { code: 'session/attachment-invalid', message: 'denied', details: { reason: 'NOPE' } },
      }),
    }))
-    const { session } = makeSession(undefined, { backgroundUploads: { post } })
+    const { session } = makeSession()
+    bindFileUpload(session, post)
    await expect(session.uploadFile(new Blob([]))).resolves.toMatchObject({
      ok: false,
      error: { code: 'session/attachment-invalid', message: 'denied', details: { reason: 'NOPE' } },
@@ -119,6 +133,56 @@ describe('Session file upload', () => {
    ])
  })

+  it('keeps fixture Blob fallback on the Remote and refuses an uncarried stream', async () => {
+    const { api, session } = makeSession()
+    const post = vi.fn<FileUploadService['post']>()
+    bindFileUpload(session, post, false)
+
+    await expect(session.uploadFile(new Blob([Uint8Array.of(1)]), 'fixture.bin'))
+      .resolves.toMatchObject({ ok: true })
+    const stream = new ReadableStream<Uint8Array>({ start(controller) { controller.close() } })
+    await expect(session.uploadFile(stream, 'stream.bin'))
+      .rejects.toThrow('stream file upload requires a bound Client file-upload service')
+    expect(post).not.toHaveBeenCalled()
+    expect(api.callsOf('session.uploadFile')).toEqual([
+      { sessionId: SID, data: 'AQ==', name: 'fixture.bin' },
+    ])
+  })
+
+  it('hands a one-shot ReadableStream to the scoped file-upload service', async () => {
+    const post = vi.fn(() => Promise.resolve({
+      status: 200,
+      body: JSON.stringify({
+        ok: true,
+        value: {
+          receiptId: 'stream-receipt',
+          file: { attachmentId: 'stream-file', name: 'stream.bin', bytes: 3 },
+        },
+      }),
+    }))
+    const { session } = makeSession()
+    bindFileUpload(session, post)
+    const stream = new ReadableStream<Uint8Array>({
+      start(controller) {
+        controller.enqueue(Uint8Array.of(1, 2, 3))
+        controller.close()
+      },
+    })
+    await expect(session.uploadFile(stream, 'stream.bin')).resolves.toMatchObject({ ok: true })
+    expect(post).toHaveBeenCalledWith(expect.objectContaining({
+      path: '/api/session/uploadFileBinary?sessionId=fk-s1&name=stream.bin',
+      body: stream,
+    }))
+  })
+
+  it('refuses a stream when a bare Session has no scoped upload service', async () => {
+    const { session } = makeSession()
+    const stream = new ReadableStream<Uint8Array>({ start(controller) { controller.close() } })
+    await expect(session.uploadFile(stream)).rejects.toThrow(
+      'stream file upload requires a bound Client file-upload service',
+    )
+  })
+
  it('folds non-200 and malformed background responses into transport failures', async () => {
    const bodies: unknown[] = [
      null,
@@ -137,15 +201,13 @@ describe('Session file upload', () => {
      { ok: true, value: { receiptId: 'r', file: { attachmentId: 'a', name: 'x', bytes: -1 } } },
    ]
    for (const body of bodies) {
-      const { session } = makeSession(undefined, {
-        backgroundUploads: { post: () => Promise.resolve({ status: 200, body: JSON.stringify(body) }) },
-      })
+      const { session } = makeSession()
+      bindFileUpload(session, () => Promise.resolve({ status: 200, body: JSON.stringify(body) }))
      await expect(session.uploadFile(new Blob([])))
        .rejects.toThrow(/file upload transport returned an invalid/)
    }
-    const { session } = makeSession(undefined, {
-      backgroundUploads: { post: () => Promise.resolve({ status: 503, body: 'unavailable' }) },
-    })
+    const { session } = makeSession()
+    bindFileUpload(session, () => Promise.resolve({ status: 503, body: 'unavailable' }))
    await expect(session.uploadFile(new Blob([])))
      .rejects.toThrow('file upload transport failed with HTTP 503')
  })
@@ -155,8 +217,8 @@ describe('Session file upload', () => {
    const api = new FakeApiClient()
    const session = new Session(SID, fakeRemote(api), {
      address: { parentSessionId: PARENT, childSessionId: SID, mode: 'continuable' },
-      backgroundUploads: { post },
    })
+    bindFileUpload(session, post)
    await expect(session.uploadFile(new Blob([]))).resolves.toMatchObject({
      ok: false,
      error: { code: 'subagent/attachment-invalid', details: { reason: 'SUBAGENT_FILE_UNSUPPORTED' } },
@@ -16,6 +16,7 @@
    { "path": "../gateway/tsconfig.client.json" },
    { "path": "../../attachment/attachment" },
    { "path": "../../client/connection/tsconfig.client.json" },
+    { "path": "../../client/file-upload" },
    { "path": "../../client/store" },
    { "path": "../../context/file-reference" },
    { "path": "../../core/session" },