fix(apiproxy): address session-export review — surrogate-safe chunks, backpressure drain, strict flag

Chunk boundaries never split a surrogate pair (a lone high surrogate
re-encodes as U+FFFD and silently corrupts the exported artifact), production
yields whenever the response queue fills so a slow consumer bounds the
accumulation, includeDescendants rejects values other than true/false instead
of silently under-exporting, the dead missing-services arm is deleted by
narrowing the streaming deps, and the readRaw failure answers 500 without
leaking host paths into the browser error bar.

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

@@ -45,6 +45,7 @@ import {
   sessionLogExportDeps,
   sessionLogZipFilename,
   streamSessionLogZip,
+  type SessionLogExportReady,
 } from './session-export.ts'
 import type { SessionRawArtifact } from '@deepseek-ai/dsh-session-persistence'
 import {
@@ -3366,17 +3367,23 @@ export function createApiProxy(ctx: Context, defaults: ApiProxyDefaults): ApiPro
             { status: 500 },
           )
         }
+        const ready: SessionLogExportReady = {
+          sessionQuery: deps.sessionQuery,
+          sessionPersistence: deps.sessionPersistence,
+        }
         let root: SessionRawArtifact | undefined
         try {
-          root = await deps.sessionPersistence.readRaw(request.sessionId)
-        } catch (error: unknown) {
-          return new Response(String(error), { status: 500 })
+          root = await deps.sessionPersistence.readRaw(request.sessionId, signal)
+        } 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 })
         }
         if (root === undefined) {
           return new Response('session not found', { status: 404 })
         }
         return new Response(
-          streamSessionLogZip(deps, root, request.sessionId, request.includeDescendants === true, signal),
+          streamSessionLogZip(ready, root, request.sessionId, request.includeDescendants === true, signal),
           {
             headers: {
               'content-type': 'application/zip',

packages/host/apiproxy/src/api/downloads.schema.ts

@@ -10,11 +10,15 @@ import { z } from 'zod'
 import type { DownloadsApi } from './downloads.ts'
 import { sessionIdSchema } from './sessions.schema.ts'
 
-/** session.export query params → the sessionLog request. */
+/**
+ * session.export query params → the sessionLog request. `includeDescendants`
+ * accepts exactly `true`/`false`/absent; any other value is rejected (400) so
+ * a misspelled flag cannot silently under-export.
+ */
 export const sessionLogQuerySchema = z
   .object({
     sessionId: sessionIdSchema,
-    includeDescendants: z.string().optional(),
+    includeDescendants: z.union([z.literal('true'), z.literal('false')]).optional(),
   })
   .transform(query => ({
     sessionId: query.sessionId,

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

@@ -4,9 +4,13 @@
  * original base name (`session.jsonl`); each subagent descendant under
  * `subagents/<id>/<filename>`. No manifest is written — every file is
  * byte-identical to the backend's durable artifact and self-describing
- * through its own header line. Compression happens on the host with fflate's
- * streaming Zip API, so the response is chunked as it is produced and the
- * host never materializes the whole archive.
+ * through its own header line. 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).
  * @module
  */
 
@@ -22,6 +26,12 @@ export interface SessionLogExportDeps {
   readonly sessionPersistence: SessionPersistence | undefined
 }
 
+/** The export services narrowed to the mounted ones streaming actually reads. */
+export interface SessionLogExportReady {
+  readonly sessionQuery: SessionQueryService
+  readonly sessionPersistence: SessionPersistence
+}
+
 /**
  * Resolve the persistence and session-query services a log export needs.
  * @param ctx - the composed host context.
@@ -33,6 +43,7 @@ export function sessionLogExportDeps(ctx: Context): SessionLogExportDeps {
     sessionPersistence: ctx.get('sessionPersistence'),
   }
 }
+
 /** One exported artifact: the stored text plus the zip path it lands at. */
 export interface SessionLogZipEntry {
   /** Zip entry path (root filename verbatim; descendants under `subagents/<id>/`). */
@@ -43,13 +54,15 @@ export interface SessionLogZipEntry {
 
 /**
  * One safe zip path segment from an untrusted session id. Session ids are
- * host-controlled, but the brand allows any non-empty string, so `../` and
- * separator characters are neutralized before they can shape archive entries.
+ * host-controlled, but the brand allows any non-empty string, so `../`, dot
+ * segments, and separator characters are neutralized before they can shape
+ * archive entries. Distinct ids may collapse onto one segment (id collision
+ * is impossible for the host-minted UUIDs, so no uniqueness suffix is kept).
  * @param id - the raw session id.
  * @returns a filesystem-safe single path segment.
  */
 function safeSessionIdSegment(id: string): string {
-  return id.replace(/[^A-Za-z0-9._-]/g, '_')
+  return id.replace(/[^A-Za-z0-9_-]/g, '_')
 }
 
 /**
@@ -60,13 +73,14 @@ function safeSessionIdSegment(id: string): string {
 export function sessionLogZipFilename(sessionId: string): string {
   return `dsh-session-${safeSessionIdSegment(sessionId)}.zip`
 }
+
 /**
  * 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 (the host holds at most one descendant's artifact text at
  * a time beyond the root).
- * @param deps - the export services.
+ * @param deps - the mounted export services (the caller answered 500 before this runs).
  * @param root - the already-read root artifact (read by the caller so the
  * missing-session path can answer cleanly before streaming starts).
  * @param sessionId - the root session id.
@@ -75,7 +89,7 @@ export function sessionLogZipFilename(sessionId: string): string {
  * @returns the export entries in zip order.
  */
 export async function* sessionLogZipEntries(
-  deps: SessionLogExportDeps,
+  deps: SessionLogExportReady,
   root: SessionRawArtifact,
   sessionId: SessionId,
   includeDescendants: boolean,
@@ -83,13 +97,6 @@ export async function* sessionLogZipEntries(
 ): AsyncGenerator<SessionLogZipEntry> {
   yield { path: root.filename, content: root.content }
   if (!includeDescendants) return
-  const sessionQuery = deps.sessionQuery
-  const sessionPersistence = deps.sessionPersistence
-  if (sessionQuery === undefined || sessionPersistence === undefined) {
-    // The caller validated services before the stream started; this arm is
-    // unreachable today and guards a future caller that skips the check.
-    throw new Error('session log export is unavailable: missing session-query or session-persistence service')
-  }
   const seen = new Set<SessionId>([sessionId])
   const collect = async function* (
     nodes: readonly SessionLineageNode[],
@@ -99,7 +106,7 @@ export async function* sessionLogZipEntries(
       const id = node.session.header.id
       if (seen.has(id)) continue
       seen.add(id)
-      const raw = await sessionPersistence.readRaw(id)
+      const raw = await deps.sessionPersistence.readRaw(id)
       if (raw === undefined) {
         throw new Error(`subagent "${id}" has no stored log artifact`)
       }
@@ -110,12 +117,48 @@ export async function* sessionLogZipEntries(
       yield* collect(node.descendants)
     }
   }
-  const lineage = await sessionQuery.traceSession(sessionId)
+  const lineage = await deps.sessionQuery.traceSession(sessionId)
   yield* collect(lineage.descendants)
 }
 
-/** How many code points of artifact text one zip push carries (bounded encode memory). */
-const PUSH_CHUNK_CODE_POINTS = 1 << 16
+/** How many code units of artifact text one zip push carries (bounded encode memory). */
+const PUSH_CHUNK_CODE_UNITS = 1 << 16
+
+/**
+ * Push one artifact's text into a deflate stream in bounded chunks, never
+ * splitting a surrogate pair across a chunk boundary (a lone high surrogate
+ * 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.
+ */
+async function pushArtifactChunks(
+  deflate: ZipDeflate,
+  content: string,
+  controller: ReadableStreamDefaultController<Uint8Array>,
+  signal?: AbortSignal,
+): Promise<void> {
+  const encoder = new TextEncoder()
+  let offset = 0
+  let finalChunk: boolean
+  do {
+    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
+      // pair: the pair then starts the next chunk whole.
+      const last = content.charCodeAt(end - 1)
+      if (last >= 0xd800 && last <= 0xdbff) end -= 1
+    }
+    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))
+    }
+  } while (!finalChunk)
+}
 
 /**
  * Stream one session-log ZIP as a WHATWG ReadableStream. The root artifact is
@@ -124,7 +167,7 @@ const PUSH_CHUNK_CODE_POINTS = 1 << 16
  * then encoded and deflated in bounded chunks as it is produced, so the
  * archive bytes arrive incrementally. A descendant that fails to read errors
  * the stream (fail-loud, never silent under-export).
- * @param deps - the export services.
+ * @param deps - the mounted export services (the caller answered 500 before this runs).
  * @param root - the already-read root artifact (first zip entry).
  * @param sessionId - the root session id.
  * @param includeDescendants - whether to include every subagent descendant.
@@ -132,23 +175,25 @@ const PUSH_CHUNK_CODE_POINTS = 1 << 16
  * @returns the zip byte stream.
  */
 export function streamSessionLogZip(
-  deps: SessionLogExportDeps,
+  deps: SessionLogExportReady,
   root: SessionRawArtifact,
   sessionId: SessionId,
   includeDescendants: boolean,
   signal?: AbortSignal,
 ): ReadableStream<Uint8Array> {
-  const encoder = new TextEncoder()
   return new ReadableStream<Uint8Array>({
     start(controller) {
-      // fflate invokes the callback synchronously per compressed chunk;
-      // enqueued bytes stay bounded by the compressed archive size (the body
-      // consumer drains them over the wire as the stream is pulled).
+      // 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) => {
+        /* v8 ignore next 3 -- fflate reports only internal zip failures, unreachable for valid inputs */
         if (error) {
           controller.error(error)
           return
         }
+        /* v8 ignore next -- fflate may emit empty chunks; not controllable from tests */
         if (data.byteLength > 0) controller.enqueue(data)
         if (final) controller.close()
       })
@@ -157,17 +202,13 @@ export function streamSessionLogZip(
           for await (const entry of sessionLogZipEntries(deps, root, sessionId, includeDescendants, signal)) {
             const deflate = new ZipDeflate(entry.path, { level: 6 })
             zip.add(deflate)
-            const content = entry.content
-            for (let offset = 0; offset < content.length; offset += PUSH_CHUNK_CODE_POINTS) {
-              signal?.throwIfAborted()
-              const finalChunk = offset + PUSH_CHUNK_CODE_POINTS >= content.length
-              deflate.push(encoder.encode(content.slice(offset, offset + PUSH_CHUNK_CODE_POINTS)), finalChunk)
-            }
+            await pushArtifactChunks(deflate, entry.content, controller, signal)
           }
           zip.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 */
           controller.error(error instanceof Error ? error : new Error(String(error)))
         }
       })()

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

@@ -43,7 +43,7 @@ function node(id: string, ...descendants: SessionLineageNode[]): SessionLineageN
 async function buildApi(
   artifacts: Record<string, SessionRawArtifact>,
   descendants: SessionLineageNode[] = [],
-  services: { query?: boolean; persistence?: boolean } = { query: true, persistence: true },
+  services: { query?: boolean; persistence?: boolean | 'throw' } = { query: true, persistence: true },
 ) {
   const ctx = new Context()
   await ctx.plugin(UserInteractionService)
@@ -60,7 +60,10 @@ async function buildApi(
   }
   if (services.persistence) {
     ctx.provide('sessionPersistence', {
-      readRaw: async (id: SessionId) => artifacts[id],
+      readRaw: async (id: SessionId) => {
+        if (services.persistence === 'throw') throw new Error('/host/private/session.jsonl')
+        return artifacts[id]
+      },
     } as never)
   }
   return createApiProxy(ctx, {
@@ -125,6 +128,14 @@ describe('session.export download endpoint', () => {
     expect(response.status).toBe(400)
   })
 
+  it('answers 400 for an includeDescendants value other than true or false', async () => {
+    const api = await buildApi({ 'session-root': artifact('session-root') })
+    const response = await toFetchHandler(api).fetch(
+      new Request('http://host/api/session.export?sessionId=session-root&includeDescendants=1'),
+    )
+    expect(response.status).toBe(400)
+  })
+
   it('answers 500 when the deployment mounts no persistence or session-query service', async () => {
     const api = await buildApi({}, [], { query: false, persistence: false })
     const response = await toFetchHandler(api).fetch(
@@ -146,4 +157,73 @@ describe('session.export download endpoint', () => {
     // than returning a truncated-but-valid archive.
     await expect(response.arrayBuffer()).rejects.toThrow()
   })
+
+  it('keeps an astral character whole when its surrogate pair straddles a push boundary', async () => {
+    // The push loop slices by 2^16 code units and must back off one unit when
+    // the boundary lands inside a surrogate pair; otherwise the pair re-encodes
+    // as U+FFFD and the exported artifact is silently corrupted.
+    const root = { ...artifact('session-root'), content: `${'a'.repeat((1 << 16) - 1)}😀tail` }
+    const api = await buildApi({ 'session-root': root })
+    const response = await toFetchHandler(api).fetch(
+      new Request('http://host/api/session.export?sessionId=session-root'),
+    )
+    const files = unzipSync(await responseBytes(response))
+    expect(strFromU8(files['session.jsonl'] as Uint8Array)).toBe(root.content)
+  })
+
+  it('splits a long artifact on a plain code-unit boundary without backoff', async () => {
+    // A boundary that lands on a BMP character needs no surrogate backoff; the
+    // round trip must still be byte-identical across the multi-chunk push.
+    const root = { ...artifact('session-root'), content: 'z'.repeat((1 << 16) + 4096) }
+    const api = await buildApi({ 'session-root': root })
+    const response = await toFetchHandler(api).fetch(
+      new Request('http://host/api/session.export?sessionId=session-root'),
+    )
+    const files = unzipSync(await responseBytes(response))
+    expect(strFromU8(files['session.jsonl'] as Uint8Array)).toBe(root.content)
+  })
+
+  it('exports an empty artifact as an empty zip entry', async () => {
+    const root = { ...artifact('session-root'), content: '' }
+    const api = await buildApi({ 'session-root': root })
+    const response = await toFetchHandler(api).fetch(
+      new Request('http://host/api/session.export?sessionId=session-root'),
+    )
+    const files = unzipSync(await responseBytes(response))
+    expect(Object.keys(files)).toEqual(['session.jsonl'])
+    expect(strFromU8(files['session.jsonl'] as Uint8Array)).toBe('')
+  })
+
+  it('exports a shared lineage node once (seen-set dedup)', async () => {
+    const api = await buildApi({
+      'session-root': artifact('session-root'),
+      'child-a': artifact('child-a', sid('session-root')),
+      'child-b': artifact('child-b', sid('session-root')),
+      shared: artifact('shared', sid('child-a')),
+    }, [
+      node('child-a', node('shared')),
+      node('child-b', node('shared')),
+    ])
+    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(Object.keys(files).sort()).toEqual([
+      'session.jsonl',
+      'subagents/child-a/session.jsonl',
+      'subagents/child-b/session.jsonl',
+      'subagents/shared/session.jsonl',
+    ])
+  })
+
+  it('answers 500 without leaking the backend error when the root artifact read fails', async () => {
+    const api = await buildApi({}, [], { query: true, persistence: 'throw' })
+    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 read the stored artifact')
+    expect(body).not.toContain('/host/private/')
+  })
 })