Merge remote-tracking branch 'origin/master' into worktree/web-plugin-config

packages/host/apiproxy/README.i18n.yaml

@@ -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/host/apiproxy/README.md
-README.md: eb42aa3f16a4991aa5b5de22b3015491b5b15754
-README.zh.md: fca7644a6beb94f7dd2e13bb95c2bfcd45d5df20
+README.md: 056a4e640adb667ecffef358b8a2f31a174a45c7
+README.zh.md: 63a41baf84d5e7bf23fc66234781bb868be86ddc

packages/host/apiproxy/README.md

@@ -2,7 +2,7 @@
 
 English | [中文](README.zh.md)
 
-The API gateway shared by every client consists of the TypeScript API contract (`src/api/`, zero Node dependencies, importable from the browser), the fetch carrier pair (`src/fetch/`: `toFetchHandler` on the host side, `AbstractApiClient` plus platform subclasses on the client side), and the host-side implementation (`src/api-proxy.ts`: `createApiProxy` plus the default-exported `ApiProxyService` gateway plugin — config `{nativeOpen?}`, provides `ctx.apiProxy`). This package registers no routes; carriers such as HTTP wrap `ctx.apiProxy` themselves. The shipped Web composition lives in [`packages/bundle/web-app/cordis.patch.yml`](../../bundle/web-app/cordis.patch.yml), while its default Agent model selection belongs to [`@deepseek-ai/dsh-agent-default-model`](../../core/agent-default-model/README.md) in the base bundle.
+The API gateway shared by every client consists of the TypeScript API contract (`src/api/`, zero Node dependencies, importable from the browser), the fetch carrier pair (`src/fetch/`: `toFetchHandler` on the host side, `AbstractApiClient` plus platform subclasses on the client side), and the host-side implementation (`src/api-proxy.ts`: `createApiProxy` plus the default-exported `ApiProxyService` gateway plugin — config `{nativeOpen?, sessionExportCompressionLevel?}`, provides `ctx.apiProxy`). This package registers no routes; carriers such as HTTP wrap `ctx.apiProxy` themselves. The shipped Web composition lives in [`packages/bundle/web-app/cordis.patch.yml`](../../bundle/web-app/cordis.patch.yml), while its default Agent model selection belongs to [`@deepseek-ai/dsh-agent-default-model`](../../core/agent-default-model/README.md) in the base bundle.
 
 ## The shared Agent default (`agent-default-model` Settings section)
 
@@ -28,7 +28,7 @@ Question responses are validated against their pending request before the first
 
 `session.history`'s tail page (`beforeSeq` absent) additionally carries an optional `projections` block — the watermark snapshot of every unit registered on `ctx.sessionProjections` (`@deepseek-ai/dsh-session-projection`), with `asOfSeq` = the last event seq the values reflect (`-1` on an empty log). The gateway also subscribes to the registry's change feed and mints a `session/projection` mux frame per changed unit (`{sessionId, key, value, seq}` — live push state, never logged; clients hold one generic per-session value store under higher-seq-wins). The carrier holds zero domain knowledge (each value passed its unit's own schema inside the registry; the wire schemas keep `values`/`value` wide); loadOlder pages never carry the block, and a composition without the registry serves histories without either surface.
 
-Session-log export is a host-only download surface, not an RPC: `GET /api/session.export?sessionId=…&includeDescendants=true` streams a ZIP whose files are each session's stored artifact text verbatim (the persistence backend's `readRaw` — exact durable bytes decoded from the physical encoding, never a reconstruction from parsed events), root under its original base name plus each subagent descendant under `subagents/<id>/`, and every image any included log references under `media/<attachmentId>.<ext>` (read and verified from the attachment store; a shared image appears once). Compression runs on the host with fflate's streaming Zip API, so the response is chunked as it is produced and the host never holds the whole archive in one buffer, and production yields whenever the response queue fills, so a slow consumer bounds the accumulation (fflate's callback is synchronous — the drain point is the only backpressure). It requires the persistence, session-query, and attachment services: a deployment without any answers 500, a missing root session 404, and a descendant without a stored artifact or a referenced image that cannot be read fails the stream (fail-loud, never silent under-export). The carrier mounts the endpoint; `ApiProxy.downloads.sessionLog` implements it.
+Session-log export is a host-only download surface, not an RPC: `GET /api/session.export?sessionId=…&includeDescendants=true` streams a ZIP whose files are each session's stored artifact text verbatim (the persistence backend's `readRaw` — exact durable bytes decoded from the physical encoding, never a reconstruction from parsed events), root under its original base name plus each subagent descendant under `subagents/<id>/`, and every image any included log references under `media/<attachmentId>.<ext>` (read and verified from the attachment store; a shared image appears once). Each live root or descendant crosses the authoritative `SessionStore.flush` durability barrier immediately before its raw artifact read; cold sessions have no in-memory work to flush. Compression runs on the host with fflate's streaming Zip API at validated `sessionExportCompressionLevel` 0–9 (default 6), so deployments can trade CPU and latency against archive size; the response is chunked as it is produced and the host never holds the whole archive in one buffer. Once the response queue reaches its 64 KiB byte high-water mark, production waits until consumer pull restores positive capacity; fflate's synchronous callback can overshoot that bound only by the output of one bounded input push. Request abort and response-body cancellation stop lineage and artifact work, terminate the active compressor, and propagate as cancellation rather than an HTTP 500. It requires the persistence, session-query, and attachment services: a deployment without any answers 500, a persistence backend without per-session raw artifacts answers 501, a missing root session answers 404, and a descendant without a stored artifact or a referenced image that cannot be read fails the stream (fail-loud, never silent under-export). The carrier mounts the endpoint; `ApiProxy.downloads.sessionLog` implements it.
 
 Session titles ride the generic projection pair like every other domain — the history-tail `projections` block plus `session/projection` frames under the `title` key. Titles do not join `session.list`; cold sessions remain metadata-only there until opening or resuming attaches their logs. `session.rename` accepts an explicit user title (resuming a cold session first), delegating to `ctx.sessionTitle.rename` — the accepted `session/title` event pins the title against automatic regeneration — and returns the normalized title plus its event seq so a client settles its `title` projection cell ahead of the push frame; a title that normalizes to empty returns `title-invalid`.
 

packages/host/apiproxy/README.zh.md

@@ -2,7 +2,7 @@
 
 [English](README.md) | 中文
 
-所有客户端共用的 API 网关由三部分组成：TypeScript API 约定（`src/api/`，不依赖 Node，可从浏览器导入）、fetch 载体对（`src/fetch/`：宿主侧的 `toFetchHandler`，以及客户端侧的 `AbstractApiClient` 与平台子类）和宿主侧实现（`src/api-proxy.ts`：`createApiProxy` 加上默认导出的 `ApiProxyService` 网关插件，其配置为 `{nativeOpen?}`，提供 `ctx.apiProxy`）。该包不注册任何路由；HTTP 等载体自行包装 `ctx.apiProxy`。随发行版交付的 Web 组合位于 [`packages/bundle/web-app/cordis.patch.yml`](../../bundle/web-app/cordis.patch.yml)，其默认 Agent（智能体）模型选择属于 base 组合包中的 [`@deepseek-ai/dsh-agent-default-model`](../../core/agent-default-model/README.md)。
+所有客户端共用的 API 网关由三部分组成：TypeScript API 约定（`src/api/`，不依赖 Node，可从浏览器导入）、fetch 载体对（`src/fetch/`：宿主侧的 `toFetchHandler`，以及客户端侧的 `AbstractApiClient` 与平台子类）和宿主侧实现（`src/api-proxy.ts`：`createApiProxy` 加上默认导出的 `ApiProxyService` 网关插件，其配置为 `{nativeOpen?, sessionExportCompressionLevel?}`，提供 `ctx.apiProxy`）。该包不注册任何路由；HTTP 等载体自行包装 `ctx.apiProxy`。随发行版交付的 Web 组合位于 [`packages/bundle/web-app/cordis.patch.yml`](../../bundle/web-app/cordis.patch.yml)，其默认 Agent（智能体）模型选择属于 base 组合包中的 [`@deepseek-ai/dsh-agent-default-model`](../../core/agent-default-model/README.md)。
 
 ## 共享 Agent 默认值（`agent-default-model` Settings 分节）
 
@@ -28,7 +28,7 @@ Settings 分节中的 `reasoningEffort` 在 agent-default-model 插件配置中
 
 `session.history` 的尾页（不带 `beforeSeq`）额外携带一个可选的 `projections` 块——`ctx.sessionProjections`（`@deepseek-ai/dsh-session-projection`）上每个已注册单元的水位线快照，`asOfSeq` = 这些值共同反映到的最后一个事件 seq（空日志为 `-1`）。网关还订阅注册表的变更流，为每个状态发生变化的单元生成一个 `session/projection` mux 帧（`{sessionId, key, value, seq}`——实时推送状态，绝不入日志；客户端按 seq 高者胜维护一个按会话的通用值仓）。载体不持有任何领域知识（每个值在注册表内部已过其单元自己的 schema；协议 schema 对 `values`/`value` 保持宽松）；loadOlder 页永不携带该块，未装注册表的组合则两个面都不提供。
 
-会话日志导出是宿主侧的下载面，不是 RPC：`GET /api/session.export?sessionId=…&includeDescendants=true` 流式返回一个 ZIP，其中每个文件都是会话存储工件的逐字原文（持久化后端的 `readRaw`——按物理编码解码的确切持久化字节，绝非从解析后事件重建），根会话放在其原始基础文件名下，每个子代理后代放在 `subagents/<id>/` 下，每个被任何包含的日志引用的图片放在 `media/<attachmentId>.<ext>` 下（从附件存储读取并校验；共享图片只出现一次）。压缩在宿主侧用 fflate 的流式 Zip API 完成，响应边生成边分块写出，宿主从不把整个归档放进单个缓冲区，且每当响应队列填满时生产会让出，慢消费者因此只产生有界的积压（fflate 的回调是同步的——让出点是唯一的背压手段）。它要求同时挂载持久化、session-query 与附件服务：任一缺失应答 500，根会话缺失应答 404，后代缺少存储工件或引用的图片无法读取则整个流失败（fail-loud，绝不静默少导出）。端点由传输层挂载，`ApiProxy.downloads.sessionLog` 实现它。
+会话日志导出是宿主侧的下载面，不是 RPC：`GET /api/session.export?sessionId=…&includeDescendants=true` 流式返回一个 ZIP，其中每个文件都是会话存储工件的逐字原文（持久化后端的 `readRaw`——按物理编码解码的确切持久化字节，绝非从解析后事件重建），根会话放在其原始基础文件名下，每个子代理后代放在 `subagents/<id>/` 下，每个被任何包含的日志引用的图片放在 `media/<attachmentId>.<ext>` 下（从附件存储读取并校验；共享图片只出现一次）。每个实时根会话或后代都会在读取原始工件前立即通过权威的 `SessionStore.flush` 持久性屏障；冷会话没有需要 flush 的内存工作。压缩在宿主侧使用 fflate 流式 Zip API 和已验证的 `sessionExportCompressionLevel` 0–9（默认 6），使部署可以在 CPU／延迟与归档大小之间取舍；响应边生成边分块写出，宿主从不把整个归档放进单个缓冲区。响应队列达到 64 KiB 字节高水位后，生产会等待 Consumer pull 恢复正容量；fflate 的同步回调最多只会让该界限多出一次有界输入 push 的输出。请求中止或响应 body 取消会停止血缘与工件工作、终止活跃压缩器，并继续按取消传播，而不会变成 HTTP 500。它要求同时挂载持久化、session-query 与附件服务：任一缺失应答 500，持久化后端不提供每会话原始工件时应答 501，根会话缺失时应答 404，后代缺少存储工件或引用的图片无法读取则整个流失败（fail-loud，绝不静默少导出）。端点由传输层挂载，`ApiProxy.downloads.sessionLog` 实现它。
 
 会话标题与其他所有领域一样搭乘这对通用投影机制——历史尾页的 `projections` 块外加 `title` 键下的 `session/projection` 帧。标题不会加入 `session.list`；冷会话在其中仍只有元数据，直到打开或恢复操作附加其日志。`session.rename` 接受用户显式标题（冷会话先恢复），委托给 `ctx.sessionTitle.rename`——被接受的 `session/title` 事件将标题钉住、不再被自动生成覆盖——并返回规范化后的标题及其事件 seq，让 client 在推送帧到达前就结算自己的 `title` 投影格；规范化后为空的标题返回 `title-invalid`。
 

packages/host/apiproxy/src/api-proxy.ts

@@ -43,10 +43,13 @@ import type {
   WorkspaceId, WorkspaceView,
 } from './api/index.ts'
 import {
+  DEFAULT_SESSION_LOG_COMPRESSION_LEVEL,
+  flushLiveSessionLog,
   sessionLogExportDeps,
   sessionLogZipFilename,
   streamSessionLogZip,
   type SessionLogExportReady,
+  type SessionLogCompressionLevel,
 } from './session-export.ts'
 import type { SessionRawArtifact } from '@deepseek-ai/dsh-session-persistence'
 import {
@@ -554,6 +557,8 @@ export interface ApiProxyDefaults {
   openPath?: (path: string, signal: AbortSignal) => Promise<void>
   /** Native text-editor handoff; injectable for settings-document tests. */
   openTextFile?: (path: string, signal: AbortSignal) => Promise<void>
+  /** Validated DEFLATE level for session-log ZIP entries; defaults to 6. */
+  sessionExportCompressionLevel?: SessionLogCompressionLevel
   /**
    * Whether handing a path to the native opener can work at all — the
    * `hasDocument` capability the preset roster reports, and the switch
@@ -999,6 +1004,8 @@ function changedWorkspaceView(workspaceId: string, value: unknown): WorkspaceVie
  * @returns the ApiProxy implementation.
  */
 export function createApiProxy(ctx: Context, defaults: ApiProxyDefaults): ApiProxy {
+  const sessionExportCompressionLevel = defaults.sessionExportCompressionLevel
+    ?? DEFAULT_SESSION_LOG_COMPRESSION_LEVEL
   /** The seed model each create/resume declares; re-read so it never goes stale. */
   const agentOptions = (): AgentOptions => {
     const { provider, model } = defaults.defaultModelSelection()
@@ -3500,24 +3507,41 @@ export function createApiProxy(ctx: Context, defaults: ApiProxyDefaults): ApiPro
             { status: 500 },
           )
         }
+        if (!deps.sessionPersistence.supportsRawArtifacts) {
+          return new Response(
+            'session log export is unavailable: the persistence backend does not expose per-session raw artifacts',
+            { status: 501 },
+          )
+        }
         const ready: SessionLogExportReady = {
           sessionQuery: deps.sessionQuery,
           sessionPersistence: deps.sessionPersistence,
           attachments: deps.attachments,
+          sessions: deps.sessions,
         }
         let root: SessionRawArtifact | undefined
         try {
+          await flushLiveSessionLog(deps, request.sessionId, signal)
           root = await deps.sessionPersistence.readRaw(request.sessionId, signal)
+          signal.throwIfAborted()
         } catch {
-          // Backend read failure: answer 500 without echoing the error, which
-          // may carry absolute host paths into the browser error bar.
-          return new Response('session log export failed to read the stored artifact', { status: 500 })
+          signal.throwIfAborted()
+          // Root preparation failure: answer 500 without echoing the error,
+          // which may carry absolute host paths into the browser error bar.
+          return new Response('session log export failed to prepare the stored artifact', { status: 500 })
         }
         if (root === undefined) {
           return new Response('session not found', { status: 404 })
         }
         return new Response(
-          streamSessionLogZip(ready, root, request.sessionId, request.includeDescendants === true, signal),
+          streamSessionLogZip(
+            ready,
+            root,
+            request.sessionId,
+            request.includeDescendants === true,
+            sessionExportCompressionLevel,
+            signal,
+          ),
           {
             headers: {
               'content-type': 'application/zip',

packages/host/apiproxy/src/index.ts

@@ -17,6 +17,10 @@ import z from '@deepseek-ai/schemastery'
 import type {} from '@deepseek-ai/dsh-agent-default-model'
 import type { ApiProxy } from './api/index.ts'
 import { createApiProxy } from './api-proxy.ts'
+import {
+  DEFAULT_SESSION_LOG_COMPRESSION_LEVEL,
+  type SessionLogCompressionLevel,
+} from './session-export.ts'
 
 export type * from './api/index.ts'
 export { RpcId } from './api/rpc.ts'
@@ -33,7 +37,7 @@ declare module '@deepseek-ai/cordis' {
   }
 }
 
-/** Gateway plugin config for native Host integration. */
+/** Gateway plugin configuration. */
 export interface Config {
   /**
    * Whether this deployment can hand paths to a native desktop opener —
@@ -43,6 +47,12 @@ export interface Config {
    * container whose DISPLAY points nowhere a user can see.
    */
   nativeOpen?: boolean
+  /**
+   * DEFLATE level for every session-log ZIP entry: `0` stores without
+   * compression, `1` favors CPU/latency, and `9` favors archive size.
+   * @default 6
+   */
+  sessionExportCompressionLevel?: 0 | 1 | 2 | 3 | 4 | 5 | 6 | 7 | 8 | 9
 }
 
 /**
@@ -58,6 +68,8 @@ export class ApiProxyService extends Service implements ApiProxy {
 
   static Config: z<Config> = z.object({
     nativeOpen: z.boolean(),
+    sessionExportCompressionLevel: z.number().step(1).min(0).max(9)
+      .default(DEFAULT_SESSION_LOG_COMPRESSION_LEVEL) as z<SessionLogCompressionLevel>,
   })
 
   readonly sessions: ApiProxy['sessions']
@@ -82,6 +94,9 @@ export class ApiProxyService extends Service implements ApiProxy {
       saveDefaultModelSelection: selection => ctx.agentDefaultModel.saveSelection(selection),
       cwd: process.cwd(),
       ...config.nativeOpen === undefined ? {} : { canOpenPath: () => config.nativeOpen as boolean },
+      ...(config.sessionExportCompressionLevel === undefined
+        ? {}
+        : { sessionExportCompressionLevel: config.sessionExportCompressionLevel }),
     })
     this.sessions = api.sessions
     this.subagents = api.subagents

packages/host/apiproxy/src/session-export.ts

@@ -6,13 +6,16 @@
  * by any included log under `media/<attachmentId>.<ext>` (content-addressed,
  * so one archive never duplicates a shared image). No manifest is written —
  * every file is byte-identical to the backend's durable artifact or attachment
- * store and self-describing through its own header line or media type.
+ * store and self-describing through its own header line or media type. Before
+ * each live session's artifact read, the SessionStore flush barrier makes the
+ * current in-memory log durable; cold sessions need no barrier. Request abort
+ * and response-consumer cancellation share one producer signal and terminate
+ * the active compressor.
  * Compression runs on the host with fflate's streaming Zip API, so the archive
  * bytes are produced incrementally and the host never holds the whole archive
- * in one buffer; production yields to the consumer whenever the response queue
- * fills past its high-water mark, so a slow consumer bounds the accumulation
- * instead of piling up the whole archive (fflate's callback is synchronous —
- * this drain point is the only backpressure available).
+ * in one buffer; production waits for consumer pull whenever the response queue
+ * reaches its byte high-water mark, so a slow consumer bounds accumulation to
+ * the fixed 64 KiB response queue plus one synchronous fflate push.
  * @module
  */
 
@@ -20,14 +23,21 @@ import { Zip, ZipDeflate } from 'fflate'
 import type { Context } from '@deepseek-ai/cordis'
 import type { AttachmentStore, ImageAttachmentRef } from '@deepseek-ai/dsh-attachment'
 import type { SessionLineageNode, SessionQueryService } from '@deepseek-ai/dsh-session-query'
-import type { SessionId } from '@deepseek-ai/dsh-session'
+import type { SessionId, SessionStore } from '@deepseek-ai/dsh-session'
 import type { SessionPersistence, SessionRawArtifact } from '@deepseek-ai/dsh-session-persistence'
 
-/** The services a session-log export needs (absent → the export is unavailable). */
+/** Valid fflate DEFLATE levels accepted by session-log export. */
+export type SessionLogCompressionLevel = 0 | 1 | 2 | 3 | 4 | 5 | 6 | 7 | 8 | 9
+
+/** Balanced default used when a direct createApiProxy caller omits deployment config. */
+export const DEFAULT_SESSION_LOG_COMPRESSION_LEVEL: SessionLogCompressionLevel = 6
+
+/** The services a session-log export needs (the live-session store is optional). */
 export interface SessionLogExportDeps {
   readonly sessionQuery: SessionQueryService | undefined
   readonly sessionPersistence: SessionPersistence | undefined
   readonly attachments: AttachmentStore | undefined
+  readonly sessions: SessionStore | undefined
 }
 
 /** The export services narrowed to the mounted ones streaming actually reads. */
@@ -35,6 +45,7 @@ export interface SessionLogExportReady {
   readonly sessionQuery: SessionQueryService
   readonly sessionPersistence: SessionPersistence
   readonly attachments: AttachmentStore
+  readonly sessions: SessionStore | undefined
 }
 
 /**
@@ -47,9 +58,32 @@ export function sessionLogExportDeps(ctx: Context): SessionLogExportDeps {
     sessionQuery: ctx.get('sessionQuery'),
     sessionPersistence: ctx.get('sessionPersistence'),
     attachments: ctx.get('attachments'),
+    sessions: ctx.get('sessions'),
   }
 }
 
+/**
+ * Flush one currently live session through the store's authoritative durability
+ * barrier immediately before its raw artifact is read. A cold or absent id has
+ * no in-memory work to flush.
+ * @param deps - export services, including the optional live-session store.
+ * @param id - the session whose artifact is about to be read.
+ * @param signal - optional cancellation observed around the flush barrier.
+ */
+export async function flushLiveSessionLog(
+  deps: Pick<SessionLogExportDeps, 'sessions'>,
+  id: SessionId,
+  signal?: AbortSignal,
+): Promise<void> {
+  signal?.throwIfAborted()
+  const sessions = deps.sessions
+  if (sessions === undefined) return
+  const session = sessions.get(id)
+  if (session === undefined) return
+  await sessions.flush(session)
+  signal?.throwIfAborted()
+}
+
 /** One exported file: a stored artifact text or one referenced media object. */
 export type SessionLogZipEntry =
   | { readonly path: string; readonly content: string }
@@ -168,9 +202,9 @@ export function sessionLogZipFilename(sessionId: string): string {
 
 /**
  * Yield the export entries in zip order: the preloaded root artifact first,
- * then every subagent descendant in lineage order (each read from the
- * persistence backend right before it is yielded and dropped after the
- * consumer moves on), then every distinct media object referenced by any of
+ * then every subagent descendant in lineage order (each flushed when live,
+ * read from the persistence backend right before it is yielded, and dropped
+ * after the consumer moves on), then every distinct media object referenced by any of
  * the included logs (read and verified from the attachment store, one archive
  * entry per attachment id). The host holds at most one descendant's artifact
  * text and one media object at a time beyond the root.
@@ -179,7 +213,7 @@ export function sessionLogZipFilename(sessionId: string): string {
  * missing-session path can answer cleanly before streaming starts).
  * @param sessionId - the root session id.
  * @param includeDescendants - whether to include every subagent descendant.
- * @param signal - optional cancellation for read work.
+ * @param signal - optional cancellation forwarded to lineage, persistence, and attachment reads.
  * @returns the export entries in zip order.
  */
 export async function* sessionLogZipEntries(
@@ -205,7 +239,9 @@ export async function* sessionLogZipEntries(
         const id = node.session.header.id
         if (seen.has(id)) continue
         seen.add(id)
-        const raw = await deps.sessionPersistence.readRaw(id)
+        await flushLiveSessionLog(deps, id, signal)
+        const raw = await deps.sessionPersistence.readRaw(id, signal)
+        signal?.throwIfAborted()
         if (raw === undefined) {
           throw new Error(`subagent "${id}" has no stored log artifact`)
         }
@@ -217,12 +253,14 @@ export async function* sessionLogZipEntries(
         yield* collect(node.descendants)
       }
     }
-    const lineage = await deps.sessionQuery.traceSession(sessionId)
+    const lineage = await deps.sessionQuery.traceSession(sessionId, signal)
+    signal?.throwIfAborted()
     yield* collect(lineage.descendants)
   }
   for (const ref of media.values()) {
     signal?.throwIfAborted()
-    const stored = await deps.attachments.readImage(ref)
+    const stored = await deps.attachments.readImage(ref, signal)
+    signal?.throwIfAborted()
     yield { path: mediaEntryPath(ref), data: stored.data }
   }
 }
@@ -233,30 +271,66 @@ const PUSH_CHUNK_CODE_UNITS = 1 << 16
 /** How many bytes of media one zip push carries (bounded memory; images are already size-capped). */
 const PUSH_CHUNK_BYTES = 1 << 16
 
+/** Byte capacity retained by the response stream before ZIP production waits for pull. */
+const RESPONSE_HIGH_WATER_MARK_BYTES = 1 << 16
+
+/** One producer waiter released only when ReadableStream pull restores capacity. */
+class ResponseCapacityGate {
+  private releasePending: (() => void) | undefined
+
+  /**
+   * Wait until the response queue has positive byte capacity or cancellation wins.
+   * @param controller - response controller whose desired size owns capacity.
+   * @param signal - combined request/consumer cancellation.
+   */
+  async wait(
+    controller: ReadableStreamDefaultController<Uint8Array>,
+    signal: AbortSignal,
+  ): Promise<void> {
+    signal.throwIfAborted()
+    if (controller.desiredSize === null || controller.desiredSize > 0) return
+    await new Promise<void>((resolve) => {
+      const release = (): void => {
+        this.releasePending = undefined
+        signal.removeEventListener('abort', release)
+        resolve()
+      }
+      this.releasePending = release
+      signal.addEventListener('abort', release, { once: true })
+    })
+    signal.throwIfAborted()
+  }
+
+  /** Release the current producer waiter after a consumer pull. */
+  pulled(): void {
+    this.releasePending?.()
+  }
+}
+
 /**
  * Push one media object's bytes into a deflate stream in bounded chunks,
- * yielding to a slow consumer between chunks like the artifact path does.
+ * waiting for consumer capacity between chunks like the artifact path does.
  * @param deflate - the zip entry's deflate stream.
  * @param data - the stored image bytes.
- * @param signal - optional cancellation; throws when aborted.
+ * @param controller - response queue controller.
+ * @param capacity - pull-driven response-capacity gate.
+ * @param signal - cancellation; throws when aborted.
  */
 async function pushBinaryChunks(
   deflate: ZipDeflate,
   data: Uint8Array,
   controller: ReadableStreamDefaultController<Uint8Array>,
-  signal?: AbortSignal,
+  capacity: ResponseCapacityGate,
+  signal: AbortSignal,
 ): Promise<void> {
   let offset = 0
   do {
-    signal?.throwIfAborted()
+    signal.throwIfAborted()
     const end = Math.min(offset + PUSH_CHUNK_BYTES, data.byteLength)
     const finalChunk = end >= data.byteLength
     deflate.push(data.subarray(offset, end), finalChunk)
     offset = end
-    /* v8 ignore next 2 -- only fires when a slow consumer leaves the queue over-full */
-    if (controller.desiredSize !== null && controller.desiredSize < 0) {
-      await new Promise(resolve => setTimeout(resolve, 0))
-    }
+    await capacity.wait(controller, signal)
   } while (offset < data.byteLength)
 }
 
@@ -266,19 +340,22 @@ async function pushBinaryChunks(
  * re-encodes as U+FFFD and would silently corrupt the exported artifact).
  * @param deflate - the zip entry's deflate stream.
  * @param content - the artifact text verbatim.
- * @param signal - optional cancellation; throws when aborted.
+ * @param controller - response queue controller.
+ * @param capacity - pull-driven response-capacity gate.
+ * @param signal - cancellation; throws when aborted.
  */
 async function pushArtifactChunks(
   deflate: ZipDeflate,
   content: string,
   controller: ReadableStreamDefaultController<Uint8Array>,
-  signal?: AbortSignal,
+  capacity: ResponseCapacityGate,
+  signal: AbortSignal,
 ): Promise<void> {
   const encoder = new TextEncoder()
   let offset = 0
   let finalChunk: boolean
   do {
-    signal?.throwIfAborted()
+    signal.throwIfAborted()
     let end = Math.min(offset + PUSH_CHUNK_CODE_UNITS, content.length)
     if (end < content.length && end - offset > 1) {
       // Back off one code unit when the boundary lands inside a surrogate
@@ -289,10 +366,7 @@ async function pushArtifactChunks(
     finalChunk = end >= content.length
     deflate.push(encoder.encode(content.slice(offset, end)), finalChunk)
     offset = end
-    /* v8 ignore next 2 -- only fires when a slow consumer leaves the queue over-full */
-    if (controller.desiredSize !== null && controller.desiredSize < 0) {
-      await new Promise(resolve => setTimeout(resolve, 0))
-    }
+    await capacity.wait(controller, signal)
   } while (!finalChunk)
 }
 
@@ -307,7 +381,8 @@ async function pushArtifactChunks(
  * @param root - the already-read root artifact (first zip entry).
  * @param sessionId - the root session id.
  * @param includeDescendants - whether to include every subagent descendant.
- * @param signal - optional cancellation for read work.
+ * @param compressionLevel - validated fflate DEFLATE level for every ZIP entry.
+ * @param signal - request cancellation combined with response-consumer cancellation.
  * @returns the zip byte stream.
  */
 export function streamSessionLogZip(
@@ -315,15 +390,26 @@ export function streamSessionLogZip(
   root: SessionRawArtifact,
   sessionId: SessionId,
   includeDescendants: boolean,
-  signal?: AbortSignal,
+  compressionLevel: SessionLogCompressionLevel,
+  signal: AbortSignal,
 ): ReadableStream<Uint8Array> {
+  const consumerAbort = new AbortController()
+  const producerSignal = AbortSignal.any([signal, consumerAbort.signal])
+  let zip: Zip | undefined
+  let zipTerminated = false
+  const capacity = new ResponseCapacityGate()
+  const terminateZip = (): void => {
+    if (zip === undefined || zipTerminated) return
+    zipTerminated = true
+    zip.terminate()
+  }
   return new ReadableStream<Uint8Array>({
     start(controller) {
       // fflate invokes the callback synchronously per compressed chunk, so a
-      // single push can enqueue ahead of a slow consumer; pushArtifactChunks
-      // yields between chunks once the queue is over-full, bounding the
-      // accumulation to the queue high-water mark plus one push.
-      const zip = new Zip((error, data, final) => {
+      // single push can enqueue ahead of a slow consumer; the capacity gate
+      // waits for pull between pushes once the byte queue is full, bounding
+      // accumulation to the queue high-water mark plus one synchronous push.
+      const archive = new Zip((error, data, final) => {
         /* v8 ignore next 3 -- fflate reports only internal zip failures, unreachable for valid inputs */
         if (error) {
           controller.error(error)
@@ -333,25 +419,39 @@ export function streamSessionLogZip(
         if (data.byteLength > 0) controller.enqueue(data)
         if (final) controller.close()
       })
+      zip = archive
       void (async () => {
         try {
-          for await (const entry of sessionLogZipEntries(deps, root, sessionId, includeDescendants, signal)) {
-            const deflate = new ZipDeflate(entry.path, { level: 6 })
-            zip.add(deflate)
+          for await (const entry of sessionLogZipEntries(deps, root, sessionId, includeDescendants, producerSignal)) {
+            const deflate = new ZipDeflate(entry.path, { level: compressionLevel })
+            archive.add(deflate)
             if ('content' in entry) {
-              await pushArtifactChunks(deflate, entry.content, controller, signal)
+              await pushArtifactChunks(deflate, entry.content, controller, capacity, producerSignal)
             } else {
-              await pushBinaryChunks(deflate, entry.data, controller, signal)
+              await pushBinaryChunks(deflate, entry.data, controller, capacity, producerSignal)
             }
           }
-          zip.end()
+          archive.end()
         } catch (error) {
           // A mid-stream failure (missing descendant, cancellation, read
           // error) must fail the download rather than ship a truncated archive.
           /* v8 ignore next -- typed backends reject with Error, and DOMException is one in Node */
+          terminateZip()
           controller.error(error instanceof Error ? error : new Error(String(error)))
         }
       })()
     },
+    pull() {
+      capacity.pulled()
+    },
+    cancel(reason) {
+      consumerAbort.abort(
+        reason instanceof Error ? reason : new Error('session log export stream cancelled'),
+      )
+      terminateZip()
+    },
+  }, {
+    highWaterMark: RESPONSE_HIGH_WATER_MARK_BYTES,
+    size: chunk => chunk.byteLength,
   })
 }

packages/host/apiproxy/tests/session-export.spec.ts

@@ -5,7 +5,8 @@
  * root → 404, missing descendant → errored stream).
  */
 
-import { describe, expect, it } from 'vitest'
+import { randomBytes } from 'node:crypto'
+import { describe, expect, it, vi } from 'vitest'
 import { Context } from '@deepseek-ai/cordis'
 import { unzipSync, strFromU8 } from 'fflate'
 import type { ImageAttachmentRef } from '@deepseek-ai/dsh-attachment'
@@ -13,8 +14,7 @@ import UserInteractionService from '@deepseek-ai/dsh-user-interaction'
 import type { SessionHeader, SessionId } from '@deepseek-ai/dsh-session'
 import type { SessionLineageNode } from '@deepseek-ai/dsh-session-query'
 import type { SessionRawArtifact } from '@deepseek-ai/dsh-session-persistence'
-import { toFetchHandler } from '@deepseek-ai/dsh-host-apiproxy'
-import { createApiProxy } from '@deepseek-ai/dsh-host-apiproxy'
+import ApiProxyService, { createApiProxy, toFetchHandler } from '@deepseek-ai/dsh-host-apiproxy'
 
 const sid = (id: string): SessionId => id as SessionId
 
@@ -59,8 +59,21 @@ async function buildApi(
   descendants: SessionLineageNode[] = [],
   services: {
     query?: boolean
-    persistence?: boolean | 'throw'
-    attachments?: boolean | ((ref: ImageAttachmentRef) => Promise<ReturnType<typeof storedImage>>)
+    persistence?: boolean | 'throw' | 'unsupported'
+    attachments?: boolean | ((ref: ImageAttachmentRef, signal?: AbortSignal) => Promise<ReturnType<typeof storedImage>>)
+    sessions?: {
+      get(id: SessionId): { readonly id: SessionId } | undefined
+      flush(session: { readonly id: SessionId }): Promise<boolean>
+    }
+    readRaw?: (id: SessionId, signal?: AbortSignal) => Promise<SessionRawArtifact | undefined>
+    traceSession?: (id: SessionId, signal?: AbortSignal) => Promise<{
+      target: { header: SessionHeader; live: boolean; persisted: boolean }
+      ancestors: readonly SessionLineageNode[]
+      complete: boolean
+      root: { header: SessionHeader; live: boolean; persisted: boolean }
+      descendants: readonly SessionLineageNode[]
+    }>
+    compressionLevel?: 0 | 1 | 2 | 3 | 4 | 5 | 6 | 7 | 8 | 9
   } = {},
 ) {
   const ctx = new Context()
@@ -69,21 +82,22 @@ async function buildApi(
   const persistence = services.persistence ?? true
   if (query) {
     ctx.provide('sessionQuery', {
-      traceSession: async () => ({
+      traceSession: services.traceSession ?? (async () => ({
         target: { header: header('session-root'), live: false, persisted: true },
         ancestors: [],
         complete: true,
         root: { header: header('session-root'), live: false, persisted: true },
         descendants,
-      }),
+      })),
     } as never)
   }
   if (persistence) {
     ctx.provide('sessionPersistence', {
-      readRaw: async (id: SessionId) => {
+      supportsRawArtifacts: persistence !== 'unsupported',
+      readRaw: services.readRaw ?? (async (id: SessionId) => {
         if (persistence === 'throw') throw new Error('/host/private/session.jsonl')
         return artifacts[id]
-      },
+      }),
     } as never)
   }
   if (services.attachments !== false) {
@@ -97,9 +111,13 @@ async function buildApi(
       readImage,
     } as never)
   }
+  if (services.sessions !== undefined) ctx.provide('sessions', services.sessions as never)
   return createApiProxy(ctx, {
     defaultModelSelection: () => ({ provider: 'p', model: 'm' }),
     cwd: '/tmp',
+    ...services.compressionLevel === undefined
+      ? {}
+      : { sessionExportCompressionLevel: services.compressionLevel },
   })
 }
 
@@ -107,6 +125,19 @@ async function responseBytes(response: Response): Promise<Uint8Array> {
   return new Uint8Array(await response.arrayBuffer())
 }
 
+describe('session export compression config', () => {
+  it('defaults to level 6 and rejects values outside the integer 0-9 range', () => {
+    expect(ApiProxyService.Config({})).toEqual({ sessionExportCompressionLevel: 6 })
+    expect(ApiProxyService.Config({ sessionExportCompressionLevel: 0 }))
+      .toEqual({ sessionExportCompressionLevel: 0 })
+    expect(ApiProxyService.Config({ sessionExportCompressionLevel: 9 }))
+      .toEqual({ sessionExportCompressionLevel: 9 })
+    for (const value of [-1, 10, 1.5]) {
+      expect(() => ApiProxyService.Config({ sessionExportCompressionLevel: value } as never)).toThrow()
+    }
+  })
+})
+
 describe('session.export download endpoint', () => {
   it('streams a ZIP with the root artifact verbatim under its original filename', async () => {
     const api = await buildApi({ 'session-root': artifact('session-root') })
@@ -121,6 +152,24 @@ describe('session.export download endpoint', () => {
     expect(strFromU8(files['session.jsonl'] as Uint8Array)).toBe(artifact('session-root').content)
   })
 
+  it('uses the resolved compression level for ZIP entries', async () => {
+    const root = artifact('session-root', undefined, 'compressible\n'.repeat(32 * 1024))
+    const storedApi = await buildApi({ 'session-root': root }, [], { compressionLevel: 0 })
+    const compressedApi = await buildApi({ 'session-root': root }, [], { compressionLevel: 9 })
+    const stored = await storedApi.downloads.sessionLog(
+      { sessionId: sid('session-root'), includeDescendants: false },
+      new AbortController().signal,
+    )
+    const compressed = await compressedApi.downloads.sessionLog(
+      { sessionId: sid('session-root'), includeDescendants: false },
+      new AbortController().signal,
+    )
+    const storedBytes = await responseBytes(stored)
+    const compressedBytes = await responseBytes(compressed)
+    expect(compressedBytes.byteLength).toBeLessThan(storedBytes.byteLength)
+    expect(strFromU8(unzipSync(compressedBytes)['session.jsonl'] as Uint8Array)).toBe(root.content)
+  })
+
   it('includes descendant artifacts under subagents/<id>/ when requested', async () => {
     const api = await buildApi({
       'session-root': artifact('session-root'),
@@ -143,6 +192,55 @@ describe('session.export download endpoint', () => {
       .toBe(artifact('child-a').content)
   })
 
+  it('flushes each live root and descendant immediately before reading its artifact', async () => {
+    const stored: Record<string, SessionRawArtifact> = {
+      'session-root': artifact('session-root', undefined, 'stale root'),
+      'child-a': artifact('child-a', sid('session-root'), 'stale child'),
+    }
+    const durable: Record<string, SessionRawArtifact> = {
+      'session-root': artifact('session-root', undefined, 'durable root'),
+      'child-a': artifact('child-a', sid('session-root'), 'durable child'),
+    }
+    const flushed: SessionId[] = []
+    const api = await buildApi(stored, [node('child-a')], {
+      sessions: {
+        get: id => durable[id] === undefined ? undefined : { id },
+        flush: async (session) => {
+          const artifactAfterFlush = durable[session.id]
+          if (artifactAfterFlush === undefined) throw new Error('unexpected session')
+          flushed.push(session.id)
+          stored[session.id] = artifactAfterFlush
+          return true
+        },
+      },
+    })
+    const response = await toFetchHandler(api).fetch(
+      new Request('http://host/api/session.export?sessionId=session-root&includeDescendants=true'),
+    )
+    const files = unzipSync(await responseBytes(response))
+    expect(flushed).toEqual([sid('session-root'), sid('child-a')])
+    expect(strFromU8(files['session.jsonl'] as Uint8Array)).toBe('durable root')
+    expect(strFromU8(files['subagents/child-a/session.jsonl'] as Uint8Array)).toBe('durable child')
+  })
+
+  it('reads a cold artifact without asking the live-session store to flush', async () => {
+    const flush = vi.fn(async () => true)
+    const root = artifact('session-root')
+    const api = await buildApi({ 'session-root': root }, [], {
+      sessions: {
+        get: () => undefined,
+        flush,
+      },
+    })
+    const response = await api.downloads.sessionLog(
+      { sessionId: sid('session-root'), includeDescendants: false },
+      new AbortController().signal,
+    )
+    const files = unzipSync(await responseBytes(response))
+    expect(flush).not.toHaveBeenCalled()
+    expect(strFromU8(files['session.jsonl'] as Uint8Array)).toBe(root.content)
+  })
+
   it('answers 404 for a missing root session', async () => {
     const api = await buildApi({})
     const response = await toFetchHandler(api).fetch(
@@ -151,6 +249,15 @@ describe('session.export download endpoint', () => {
     expect(response.status).toBe(404)
   })
 
+  it('answers 501 when the persistence backend has no per-session raw artifacts', async () => {
+    const api = await buildApi({}, [], { persistence: 'unsupported' })
+    const response = await toFetchHandler(api).fetch(
+      new Request('http://host/api/session.export?sessionId=session-root'),
+    )
+    expect(response.status).toBe(501)
+    expect(await response.text()).toContain('does not expose per-session raw artifacts')
+  })
+
   it('answers 400 when the sessionId query parameter is absent', async () => {
     const api = await buildApi({ 'session-root': artifact('session-root') })
     const response = await toFetchHandler(api).fetch(
@@ -214,6 +321,37 @@ describe('session.export download endpoint', () => {
     expect(strFromU8(files['session.jsonl'] as Uint8Array)).toBe(root.content)
   })
 
+  it('waits for response pull capacity before reading the next archive entry', async () => {
+    const root = artifact('session-root', undefined, [
+      imageEventLine('after-root'),
+      randomBytes(512 * 1024).toString('base64'),
+    ].join('\n'))
+    let imageReads = 0
+    const api = await buildApi({ 'session-root': root }, [], {
+      attachments: async (ref) => {
+        imageReads += 1
+        return storedImage(String(ref.attachmentId), ref.mediaType)
+      },
+    })
+    vi.useFakeTimers()
+    let response: Response | undefined
+    try {
+      response = await toFetchHandler(api).fetch(
+        new Request('http://host/api/session.export?sessionId=session-root'),
+      )
+      // Exhausting timer turns must not advance a producer whose byte queue is
+      // full; only a consumer pull can release it.
+      await vi.runAllTimersAsync()
+      expect(imageReads).toBe(0)
+    } finally {
+      vi.useRealTimers()
+    }
+    if (response === undefined) throw new Error('missing export response')
+    const files = unzipSync(await responseBytes(response))
+    expect(imageReads).toBe(1)
+    expect(files['media/after-root.png']).toEqual(storedImage('after-root').data)
+  })
+
   it('exports an empty artifact as an empty zip entry', async () => {
     const root = { ...artifact('session-root'), content: '' }
     const api = await buildApi({ 'session-root': root })
@@ -254,10 +392,179 @@ describe('session.export download endpoint', () => {
     )
     expect(response.status).toBe(500)
     const body = await response.text()
-    expect(body).toBe('session log export failed to read the stored artifact')
+    expect(body).toBe('session log export failed to prepare the stored artifact')
     expect(body).not.toContain('/host/private/')
   })
 
+  it('answers the private-error-safe 500 when the live root flush fails', async () => {
+    const api = await buildApi({ 'session-root': artifact('session-root') }, [], {
+      sessions: {
+        get: id => ({ id }),
+        flush: async () => { throw new Error('/host/private/flush-state') },
+      },
+    })
+    const response = await toFetchHandler(api).fetch(
+      new Request('http://host/api/session.export?sessionId=session-root'),
+    )
+    expect(response.status).toBe(500)
+    const body = await response.text()
+    expect(body).toBe('session log export failed to prepare the stored artifact')
+    expect(body).not.toContain('/host/private/')
+  })
+
+  it('forwards one request signal through root, lineage, and descendant reads', async () => {
+    const reads: Array<{ id: SessionId; signal: AbortSignal | undefined }> = []
+    const traces: AbortSignal[] = []
+    const api = await buildApi({}, [node('child-a')], {
+      readRaw: async (id, signal) => {
+        reads.push({ id, signal })
+        return id === sid('session-root')
+          ? artifact('session-root')
+          : artifact('child-a', sid('session-root'))
+      },
+      traceSession: async (_id, signal) => {
+        if (signal !== undefined) traces.push(signal)
+        return {
+          target: { header: header('session-root'), live: false, persisted: true },
+          ancestors: [],
+          complete: true,
+          root: { header: header('session-root'), live: false, persisted: true },
+          descendants: [node('child-a')],
+        }
+      },
+    })
+    const controller = new AbortController()
+    const response = await api.downloads.sessionLog(
+      { sessionId: sid('session-root'), includeDescendants: true },
+      controller.signal,
+    )
+    await response.arrayBuffer()
+    const producerSignal = traces[0]
+    if (producerSignal === undefined) throw new Error('missing lineage signal')
+    expect(reads[0]).toEqual({ id: sid('session-root'), signal: controller.signal })
+    expect(reads[1]).toEqual({ id: sid('child-a'), signal: producerSignal })
+    const cancellation = new Error('request cancelled after response')
+    controller.abort(cancellation)
+    expect(producerSignal.aborted).toBe(true)
+    expect(producerSignal.reason).toBe(cancellation)
+  })
+
+  it('preserves request cancellation instead of translating it to HTTP 500', async () => {
+    const api = await buildApi({ 'session-root': artifact('session-root') })
+    const controller = new AbortController()
+    const cancellation = new Error('request cancelled')
+    controller.abort(cancellation)
+    await expect(api.downloads.sessionLog(
+      { sessionId: sid('session-root'), includeDescendants: false },
+      controller.signal,
+    )).rejects.toBe(cancellation)
+  })
+
+  it('aborts descendant work and terminates ZIP production when its reader cancels', async () => {
+    let reportDescendantStarted!: (signal: AbortSignal) => void
+    const descendantStarted = new Promise<AbortSignal>((resolve) => {
+      reportDescendantStarted = resolve
+    })
+    const api = await buildApi({}, [node('child-a')], {
+      readRaw: async (id, signal) => {
+        if (id === sid('session-root')) return artifact('session-root')
+        if (signal === undefined) throw new Error('missing descendant signal')
+        reportDescendantStarted(signal)
+        return new Promise((_, reject) => {
+          signal.addEventListener('abort', () => {
+            reject(signal.reason as Error)
+          }, { once: true })
+        })
+      },
+    })
+    const response = await api.downloads.sessionLog(
+      { sessionId: sid('session-root'), includeDescendants: true },
+      new AbortController().signal,
+    )
+    const reader = response.body?.getReader()
+    if (reader === undefined) throw new Error('missing response body')
+    const descendantSignal = await descendantStarted
+    const cancellation = new Error('download consumer left')
+    await reader.cancel(cancellation)
+    expect(descendantSignal.aborted).toBe(true)
+    expect(descendantSignal.reason).toBe(cancellation)
+  })
+
+  it('aborts attachment reads when its reader cancels', async () => {
+    let reportAttachmentStarted!: (signal: AbortSignal) => void
+    const attachmentStarted = new Promise<AbortSignal>((resolve) => {
+      reportAttachmentStarted = resolve
+    })
+    const root = artifact('session-root', undefined, [
+      '{"type":"session","version":0,"id":"session-root","createdAt":1000}',
+      imageEventLine('slow-img'),
+    ].join('\n') + '\n')
+    const api = await buildApi({ 'session-root': root }, [], {
+      attachments: async (_ref, signal) => {
+        if (signal === undefined) throw new Error('missing attachment signal')
+        reportAttachmentStarted(signal)
+        return new Promise((_, reject) => {
+          signal.addEventListener('abort', () => {
+            reject(signal.reason as Error)
+          }, { once: true })
+        })
+      },
+    })
+    const response = await api.downloads.sessionLog(
+      { sessionId: sid('session-root'), includeDescendants: false },
+      new AbortController().signal,
+    )
+    const reader = response.body?.getReader()
+    if (reader === undefined) throw new Error('missing response body')
+    const attachmentSignal = await attachmentStarted
+    const cancellation = new Error('download consumer left during attachment read')
+    await reader.cancel(cancellation)
+    expect(attachmentSignal.aborted).toBe(true)
+    expect(attachmentSignal.reason).toBe(cancellation)
+  })
+
+  it('uses a stable Error reason when its reader cancels without one', async () => {
+    let reportDescendantStarted!: (signal: AbortSignal) => void
+    const descendantStarted = new Promise<AbortSignal>((resolve) => {
+      reportDescendantStarted = resolve
+    })
+    const api = await buildApi({}, [node('child-a')], {
+      readRaw: async (id, signal) => {
+        if (id === sid('session-root')) return artifact('session-root')
+        if (signal === undefined) throw new Error('missing descendant signal')
+        reportDescendantStarted(signal)
+        return new Promise((_, reject) => {
+          signal.addEventListener('abort', () => {
+            reject(signal.reason as Error)
+          }, { once: true })
+        })
+      },
+    })
+    const response = await api.downloads.sessionLog(
+      { sessionId: sid('session-root'), includeDescendants: true },
+      new AbortController().signal,
+    )
+    const reader = response.body?.getReader()
+    if (reader === undefined) throw new Error('missing response body')
+    const descendantSignal = await descendantStarted
+    await reader.cancel()
+    expect(descendantSignal.reason).toEqual(new Error('session log export stream cancelled'))
+  })
+
+  it('normalizes a non-Error descendant failure before erroring the stream', async () => {
+    const api = await buildApi({}, [node('child-a')], {
+      readRaw: async (id) => {
+        if (id === sid('session-root')) return artifact('session-root')
+        throw 'descendant read failed'
+      },
+    })
+    const response = await api.downloads.sessionLog(
+      { sessionId: sid('session-root'), includeDescendants: true },
+      new AbortController().signal,
+    )
+    await expect(response.arrayBuffer()).rejects.toEqual(new Error('descendant read failed'))
+  })
+
   it('includes media objects referenced by the root log under media/<id>.<ext>', async () => {
     const root = artifact('session-root', undefined, [
       '{"type":"session","version":0,"id":"session-root","createdAt":1000}',