mirror of
https://github.com/deepseek-ai/deepseek-harness
synced 2026-08-15 21:04:50 +00:00
c2569c3f9d
The export ZIP now carries every image any included log references under media/<attachmentId>.<ext>, read and verified from the attachment store with one entry per shared image. The endpoint requires the attachments service alongside persistence and session-query; a referenced image that cannot be read fails the stream like a missing descendant.
358 lines
15 KiB
TypeScript
358 lines
15 KiB
TypeScript
/**
|
|
* Host-side session-log download: streams one ZIP archive whose files are the
|
|
* sessions' stored artifact text verbatim plus every referenced media object.
|
|
* The root artifact sits under its original base name (`session.jsonl`); each
|
|
* subagent descendant under `subagents/<id>/<filename>`; each image referenced
|
|
* 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.
|
|
* 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
|
|
*/
|
|
|
|
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 { SessionPersistence, SessionRawArtifact } from '@deepseek-ai/dsh-session-persistence'
|
|
|
|
/** The services a session-log export needs (absent → the export is unavailable). */
|
|
export interface SessionLogExportDeps {
|
|
readonly sessionQuery: SessionQueryService | undefined
|
|
readonly sessionPersistence: SessionPersistence | undefined
|
|
readonly attachments: AttachmentStore | undefined
|
|
}
|
|
|
|
/** The export services narrowed to the mounted ones streaming actually reads. */
|
|
export interface SessionLogExportReady {
|
|
readonly sessionQuery: SessionQueryService
|
|
readonly sessionPersistence: SessionPersistence
|
|
readonly attachments: AttachmentStore
|
|
}
|
|
|
|
/**
|
|
* Resolve the persistence, session-query, and attachment services a log export needs.
|
|
* @param ctx - the composed host context.
|
|
* @returns the export services (absent when the deployment does not mount them).
|
|
*/
|
|
export function sessionLogExportDeps(ctx: Context): SessionLogExportDeps {
|
|
return {
|
|
sessionQuery: ctx.get('sessionQuery'),
|
|
sessionPersistence: ctx.get('sessionPersistence'),
|
|
attachments: ctx.get('attachments'),
|
|
}
|
|
}
|
|
|
|
/** One exported file: a stored artifact text or one referenced media object. */
|
|
export type SessionLogZipEntry =
|
|
| { readonly path: string; readonly content: string }
|
|
| { readonly path: string; readonly data: Uint8Array }
|
|
|
|
/** Zip extension for each accepted raster media type. */
|
|
const MEDIA_TYPE_EXTENSIONS: Record<ImageAttachmentRef['mediaType'], string> = {
|
|
'image/png': 'png',
|
|
'image/jpeg': 'jpg',
|
|
'image/webp': 'webp',
|
|
'image/gif': 'gif',
|
|
}
|
|
|
|
/**
|
|
* The zip path for one media object: content-addressed by the opaque
|
|
* attachment id so shared images land once and the id in the log maps back to
|
|
* the archive entry without a manifest.
|
|
* @param ref - the durable reference from a session log.
|
|
* @returns the archive path.
|
|
*/
|
|
function mediaEntryPath(ref: ImageAttachmentRef): string {
|
|
return `media/${String(ref.attachmentId)}.${MEDIA_TYPE_EXTENSIONS[ref.mediaType]}`
|
|
}
|
|
|
|
/**
|
|
* Collect every image reference inside one content array, descending into
|
|
* nested tool results the way the live attachment route does.
|
|
* @param content - an event content array (or nested tool-result content).
|
|
* @param refs - the dedupe map being filled (keyed by attachment id).
|
|
*/
|
|
function collectImageRefs(content: unknown, refs: Map<string, ImageAttachmentRef>): void {
|
|
if (!Array.isArray(content)) return
|
|
const pending: unknown[] = []
|
|
for (const item of content) pending.push(item)
|
|
while (pending.length > 0) {
|
|
const value = pending.pop()
|
|
if (typeof value !== 'object' || value === null || Array.isArray(value)) continue
|
|
const block = value as { type?: unknown; attachment?: unknown; content?: unknown }
|
|
if (block.type === 'image' && typeof block.attachment === 'object' && block.attachment !== null) {
|
|
const ref = block.attachment as ImageAttachmentRef
|
|
refs.set(String(ref.attachmentId), ref)
|
|
}
|
|
if (Array.isArray(block.content)) {
|
|
for (const item of block.content) pending.push(item)
|
|
}
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Collect every image reference one session event carries, across the same
|
|
* carriers the live attachment route scans (direct content, message content,
|
|
* inserted messages, and completed assistant chunk blocks).
|
|
* @param event - one parsed JSONL event object.
|
|
* @param refs - the dedupe map being filled (keyed by attachment id).
|
|
*/
|
|
function collectEventImageRefs(event: unknown, refs: Map<string, ImageAttachmentRef>): void {
|
|
const data = (event as { data?: unknown }).data
|
|
if (typeof data !== 'object' || data === null) return
|
|
const carrier = data as {
|
|
content?: unknown
|
|
message?: { content?: unknown }
|
|
inserted?: Array<{ content?: unknown }>
|
|
chunk?: { type?: unknown; block?: unknown }
|
|
}
|
|
collectImageRefs(carrier.content, refs)
|
|
if (carrier.message !== undefined) collectImageRefs(carrier.message.content, refs)
|
|
if (carrier.inserted !== undefined) {
|
|
for (const message of carrier.inserted) collectImageRefs(message.content, refs)
|
|
}
|
|
if (carrier.chunk?.type === 'block-end') collectImageRefs([carrier.chunk.block], refs)
|
|
}
|
|
|
|
/**
|
|
* Collect the distinct media references one stored artifact text names.
|
|
* Lines that fail to parse cannot reference media and are skipped (the
|
|
* artifact text itself is exported verbatim regardless).
|
|
* @param content - the stored artifact text.
|
|
* @returns the dedupe map keyed by attachment id.
|
|
*/
|
|
function imageRefsInArtifact(content: string): Map<string, ImageAttachmentRef> {
|
|
const refs = new Map<string, ImageAttachmentRef>()
|
|
for (const line of content.split('\n')) {
|
|
if (line === '') continue
|
|
let event: unknown
|
|
try {
|
|
event = JSON.parse(line)
|
|
} catch {
|
|
continue
|
|
}
|
|
collectEventImageRefs(event, refs)
|
|
}
|
|
return refs
|
|
}
|
|
|
|
/**
|
|
* One safe zip path segment from an untrusted session id. Session ids are
|
|
* 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, '_')
|
|
}
|
|
|
|
/**
|
|
* The export archive filename for one root session.
|
|
* @param sessionId - the root session id (sanitized to one safe path segment).
|
|
* @returns the attachment filename for the session's export archive.
|
|
*/
|
|
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), 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.
|
|
* @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.
|
|
* @param includeDescendants - whether to include every subagent descendant.
|
|
* @param signal - optional cancellation for read work.
|
|
* @returns the export entries in zip order.
|
|
*/
|
|
export async function* sessionLogZipEntries(
|
|
deps: SessionLogExportReady,
|
|
root: SessionRawArtifact,
|
|
sessionId: SessionId,
|
|
includeDescendants: boolean,
|
|
signal?: AbortSignal,
|
|
): AsyncGenerator<SessionLogZipEntry> {
|
|
const media = new Map<string, ImageAttachmentRef>()
|
|
const rememberMedia = (content: string): void => {
|
|
for (const [id, ref] of imageRefsInArtifact(content)) media.set(id, ref)
|
|
}
|
|
rememberMedia(root.content)
|
|
yield { path: root.filename, content: root.content }
|
|
if (includeDescendants) {
|
|
const seen = new Set<SessionId>([sessionId])
|
|
const collect = async function* (
|
|
nodes: readonly SessionLineageNode[],
|
|
): AsyncGenerator<SessionLogZipEntry> {
|
|
for (const node of nodes) {
|
|
signal?.throwIfAborted()
|
|
const id = node.session.header.id
|
|
if (seen.has(id)) continue
|
|
seen.add(id)
|
|
const raw = await deps.sessionPersistence.readRaw(id)
|
|
if (raw === undefined) {
|
|
throw new Error(`subagent "${id}" has no stored log artifact`)
|
|
}
|
|
rememberMedia(raw.content)
|
|
yield {
|
|
path: `subagents/${safeSessionIdSegment(id)}/${raw.filename}`,
|
|
content: raw.content,
|
|
}
|
|
yield* collect(node.descendants)
|
|
}
|
|
}
|
|
const lineage = await deps.sessionQuery.traceSession(sessionId)
|
|
yield* collect(lineage.descendants)
|
|
}
|
|
for (const ref of media.values()) {
|
|
signal?.throwIfAborted()
|
|
const stored = await deps.attachments.readImage(ref)
|
|
yield { path: mediaEntryPath(ref), data: stored.data }
|
|
}
|
|
}
|
|
|
|
/** How many code units of artifact text one zip push carries (bounded encode memory). */
|
|
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
|
|
|
|
/**
|
|
* 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.
|
|
* @param deflate - the zip entry's deflate stream.
|
|
* @param data - the stored image bytes.
|
|
* @param signal - optional cancellation; throws when aborted.
|
|
*/
|
|
async function pushBinaryChunks(
|
|
deflate: ZipDeflate,
|
|
data: Uint8Array,
|
|
controller: ReadableStreamDefaultController<Uint8Array>,
|
|
signal?: AbortSignal,
|
|
): Promise<void> {
|
|
let offset = 0
|
|
do {
|
|
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))
|
|
}
|
|
} while (offset < data.byteLength)
|
|
}
|
|
|
|
/**
|
|
* 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
|
|
* read and validated by the caller before this is called (missing root or
|
|
* missing services answer cleanly before any byte is produced); each entry is
|
|
* 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 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.
|
|
* @param signal - optional cancellation for read work.
|
|
* @returns the zip byte stream.
|
|
*/
|
|
export function streamSessionLogZip(
|
|
deps: SessionLogExportReady,
|
|
root: SessionRawArtifact,
|
|
sessionId: SessionId,
|
|
includeDescendants: boolean,
|
|
signal?: AbortSignal,
|
|
): ReadableStream<Uint8Array> {
|
|
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) => {
|
|
/* 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()
|
|
})
|
|
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)
|
|
if ('content' in entry) {
|
|
await pushArtifactChunks(deflate, entry.content, controller, signal)
|
|
} else {
|
|
await pushBinaryChunks(deflate, entry.data, 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)))
|
|
}
|
|
})()
|
|
},
|
|
})
|
|
}
|