diff --git a/packages/attachment/attachment-local/src/compression-limiter.ts b/packages/attachment/attachment-local/src/compression-limiter.ts index 3935f262a1..26cc1c6ecb 100644 --- a/packages/attachment/attachment-local/src/compression-limiter.ts +++ b/packages/attachment/attachment-local/src/compression-limiter.ts @@ -1,5 +1,16 @@ /** Instance-owned concurrency bound for native image transformations. */ +/** + * Preserve Error rejections and normalize non-Error native binding values. + * @param reason - rejection reason returned by a compression task. + * @returns an Error suitable for promise rejection. + */ +export function compressionFailure(reason: unknown): Error { + return reason instanceof Error + ? reason + : new Error('Image compression task rejected with a non-Error value.', { cause: reason }) +} + /** FIFO limiter for asynchronous compression work. */ export class CompressionLimiter { private active = 0 @@ -30,9 +41,7 @@ export class CompressionLimiter { }, (error: unknown) => { release() - reject(error instanceof Error - ? error - : new Error('Image compression task rejected with a non-Error value.', { cause: error })) + reject(compressionFailure(error)) }, ) } diff --git a/packages/attachment/attachment-local/src/file-store.ts b/packages/attachment/attachment-local/src/file-store.ts new file mode 100644 index 0000000000..cf4c56edb1 --- /dev/null +++ b/packages/attachment/attachment-local/src/file-store.ts @@ -0,0 +1,181 @@ +/** Verbatim content-addressed local file storage. @module @deepseek-ai/dsh-attachment-local/file-store */ + +import { createHash } from 'node:crypto' +import { createReadStream } from 'node:fs' +import { join } from 'node:path' +import { AttachmentError, AttachmentId } from '@deepseek-ai/dsh-attachment' +import type { + FileAttachmentRef, SaveFileAttachment, SaveFileStreamAttachment, +} from '@deepseek-ai/dsh-attachment' +import { + publishImmutableAlias, publishImmutableObject, publishImmutableObjectStream, +} from './store.ts' + +const FILE_ID_PATTERN = /^sha256:([a-f0-9]{64})$/ +const WINDOWS_DEVICE_NAME = /^(?:con|prn|aux|nul|com[1-9]|lpt[1-9])$/iu + +function isWindowsDeviceName(name: string): boolean { + const dot = name.indexOf('.') + const stem = (dot < 0 ? name : name.slice(0, dot)).replace(/[. ]+$/u, '') + return WINDOWS_DEVICE_NAME.test(stem) +} + +function utf8Prefix(value: string, maxBytes: number): string { + let bytes = 0 + let prefix = '' + for (const character of Buffer.from(value).toString('utf8')) { + const characterBytes = Buffer.byteLength(character) + if (bytes + characterBytes > maxBytes) break + prefix += character + bytes += characterBytes + } + return prefix +} + +/** + * Sanitize one caller display name into a safe stored leaf name. Both + * separator styles are stripped by hand: a POSIX host treats `\` as an + * ordinary character, so path.basename would keep a Windows client's full + * local path and leak it into the reference and the session log. Characters + * Windows refuses in file names become `_` so one reference stays valid on + * every supported host. + * @param value - caller-declared display name, possibly a full client path. + * @returns a non-empty leaf name safe to store on every supported filesystem. + */ +export function fileLeafName(value: string | undefined): string { + if (value === undefined) return 'file' + const leaf = value.slice(Math.max(value.lastIndexOf('/'), value.lastIndexOf('\\')) + 1) + let clean = leaf + .replace(/[\u0000-\u001f\u007f]/g, '') + .replace(/[<>:"|?*]/g, '_') + .trim() + .replace(/[. ]+$/u, '') + if (isWindowsDeviceName(clean)) clean = `_${clean}` + clean = utf8Prefix(clean, 255).replace(/[. ]+$/u, '') + return clean === '' || clean === '.' || clean === '..' ? 'file' : clean +} + +function ensureFileReference(ref: FileAttachmentRef): string { + const match = FILE_ID_PATTERN.exec(String(ref.attachmentId)) + if (match?.[1] === undefined || ref.name !== fileLeafName(ref.name)) { + throw new AttachmentError('File attachment reference is invalid.', 'INVALID_ATTACHMENT_REF') + } + return match[1] +} + +/** + * Derive the absolute immutable-object path for one stored file. The digest + * names a directory so the sanitized display name stays the stored leaf name, + * giving models and users a path that ends in the real filename. + * @param root - absolute `DSH_HOME/attachments/v1` root. + * @param ref - durable file reference from the session log or an upload receipt. + * @returns provider-local path without reading the object. + * @throws an AttachmentError when the reference digest or name is invalid. + */ +export function storedFilePath(root: string, ref: FileAttachmentRef): string { + const sha256 = ensureFileReference(ref) + return join(root, 'files', sha256.slice(0, 2), sha256, ref.name) +} + +/** Canonical object path shared by every display name for one digest. */ +function storedFileObjectPath(root: string, sha256: string): string { + return join(root, 'file-objects', sha256.slice(0, 2), sha256) +} + +/** + * Commit one file byte-for-byte below a versioned attachment root. + * @param root - absolute `DSH_HOME/attachments/v1` root. + * @param input - exact bytes and optional display name. + * @returns the durable content-addressed file reference. + */ +export async function saveFileVerbatim( + root: string, + input: SaveFileAttachment, +): Promise { + const sha256 = createHash('sha256').update(input.data).digest('hex') + const ref: FileAttachmentRef = { + attachmentId: AttachmentId(`sha256:${sha256}`), + name: fileLeafName(input.name), + bytes: input.data.byteLength, + } + const objectPath = storedFileObjectPath(root, sha256) + await publishImmutableObject(root, objectPath, input.data, sha256) + await publishImmutableAlias(root, objectPath, storedFilePath(root, ref), sha256) + return ref +} + +/** + * Commit one file byte-for-byte from bounded chunks below a versioned attachment root. + * @param root - absolute `DSH_HOME/attachments/v1` root. + * @param input - ordered exact bytes, optional cancellation, and display name. + * @returns the durable content-addressed file reference. + */ +export async function saveFileStreamVerbatim( + root: string, + input: SaveFileStreamAttachment, +): Promise { + const name = fileLeafName(input.name) + const stored = await publishImmutableObjectStream( + root, + input.data, + sha256 => storedFileObjectPath(root, sha256), + input.signal, + ) + const ref: FileAttachmentRef = { + attachmentId: AttachmentId(`sha256:${stored.sha256}`), + name, + bytes: stored.bytes, + } + input.signal?.throwIfAborted() + await publishImmutableAlias( + root, + storedFileObjectPath(root, stored.sha256), + storedFilePath(root, ref), + stored.sha256, + ) + input.signal?.throwIfAborted() + return ref +} + +/** + * Read one stored file in bounded chunks and verify its byte count and digest. + * @param root - absolute `DSH_HOME/attachments/v1` root. + * @param ref - durable file reference from the session log. + * @param signal - optional cancellation for filesystem reads. + * @returns exact stored bytes in order; integrity failures reject after the final chunk. + */ +export async function* readFileStreamVerbatim( + root: string, + ref: FileAttachmentRef, + signal?: AbortSignal, +): AsyncIterable { + signal?.throwIfAborted() + const sha256 = ensureFileReference(ref) + const stream = createReadStream(storedFilePath(root, ref), { + highWaterMark: 1 << 16, + ...(signal === undefined ? {} : { signal }), + }) + const hash = createHash('sha256') + let bytes = 0 + try { + for await (const chunk of stream) { + signal?.throwIfAborted() + const data = chunk as Buffer + hash.update(data) + bytes += data.byteLength + yield data + } + } catch (error) { + signal?.throwIfAborted() + if (error instanceof Error && 'code' in error && error.code === 'ENOENT') { + throw new AttachmentError('File attachment object is missing.', 'ATTACHMENT_NOT_FOUND') + } + throw new AttachmentError('Unable to read file attachment.', 'ATTACHMENT_READ_FAILED', { cause: error }) + } finally { + stream.destroy() + } + signal?.throwIfAborted() + if (bytes !== ref.bytes || hash.digest('hex') !== sha256) { + throw new AttachmentError('Stored file attachment failed integrity verification.', 'ATTACHMENT_CORRUPT') + } +} diff --git a/packages/attachment/attachment-local/src/index.ts b/packages/attachment/attachment-local/src/index.ts index 16b963f225..7058fb3244 100644 --- a/packages/attachment/attachment-local/src/index.ts +++ b/packages/attachment/attachment-local/src/index.ts @@ -5,23 +5,32 @@ import { Context } from '@deepseek-ai/cordis' import z from '@deepseek-ai/schemastery' import { AttachmentStore } from '@deepseek-ai/dsh-attachment' import type { + FileAttachmentRef, ImageAttachmentLimits, ImageAttachmentRef, ImageRequestPolicy, RequestImageAttachment, + SaveFileAttachment, + SaveFileStreamAttachment, SaveImageAttachment, StoredImageAttachment, } from '@deepseek-ai/dsh-attachment' import { resolveDshHome } from '@deepseek-ai/dsh-home-paths' import type { NormalizationPolicy } from './normalization.ts' -import { CompressionLimiter } from './compression-limiter.ts' +import { CompressionLimiter, compressionFailure } from './compression-limiter.ts' import { commitPreparedImageFile, normalizedImagePath, prepareImageFile, readImageFile, validateImageFile } from './store.ts' +import { + readFileStreamVerbatim, saveFileStreamVerbatim, saveFileVerbatim, storedFilePath, +} from './file-store.ts' import { readRequestImageFile, requestImageVariantId } from './request-image.ts' export { canPassThroughNormalization, normalizeImage } from './normalization.ts' export type { NormalizedImage, NormalizationPolicy } from './normalization.ts' export { commitPreparedImageFile, prepareImageFile, readImageFile, saveImageFile, validateImageFile } from './store.ts' export type { PreparedImageFile } from './store.ts' +export { + fileLeafName, readFileStreamVerbatim, saveFileStreamVerbatim, saveFileVerbatim, storedFilePath, +} from './file-store.ts' export { readRequestImageFile, requestImageVariantId } from './request-image.ts' /** Default maximum encoded bytes for one submitted image; oversized sources are refused, not shrunk. */ @@ -124,9 +133,7 @@ class SharedRequest { }, (error: unknown) => { signal.removeEventListener('abort', abort) release(false) - // CompressionLimiter normalizes task rejections before this handler. - // oxlint-disable-next-line typescript/prefer-promise-reject-errors - reject(error) + reject(compressionFailure(error)) }) }) } @@ -222,6 +229,22 @@ export class LocalAttachmentStore extends AttachmentStore { return normalizedImagePath(this.root, ref) } + override async saveFile(input: SaveFileAttachment): Promise { + return saveFileVerbatim(this.root, input) + } + + override async saveFileStream(input: SaveFileStreamAttachment): Promise { + return saveFileStreamVerbatim(this.root, input) + } + + override readFileStream(ref: FileAttachmentRef, signal?: AbortSignal): AsyncIterable { + return readFileStreamVerbatim(this.root, ref, signal) + } + + override fileHostPath(ref: FileAttachmentRef): string { + return storedFilePath(this.root, ref) + } + override async readImageRequest( ref: ImageAttachmentRef, policy: ImageRequestPolicy, diff --git a/packages/attachment/attachment-local/src/store.ts b/packages/attachment/attachment-local/src/store.ts index e5a979aec1..7fab679e34 100644 --- a/packages/attachment/attachment-local/src/store.ts +++ b/packages/attachment/attachment-local/src/store.ts @@ -1,7 +1,7 @@ /** Content-addressed, owner-private local attachment storage. */ import { createHash, randomUUID } from 'node:crypto' -import { constants } from 'node:fs' +import { constants, createReadStream } from 'node:fs' import { chmod, link, mkdir, open, readFile, unlink } from 'node:fs/promises' import { dirname, join, parse, resolve } from 'node:path' import { @@ -197,60 +197,210 @@ export async function commitPreparedImageFile( if (digest(normalized) !== sha256 || normalized.byteLength !== prepared.ref.bytes) { throw new AttachmentError('Prepared attachment bytes do not match their reference.', 'ATTACHMENT_CORRUPT') } - const bucket = join(root, 'objects', sha256.slice(0, 2)) + await publishImmutableObject(root, normalizedImagePath(root, prepared.ref), normalized, sha256) + return prepared.ref +} + +/** + * Publish one immutable content-addressed object below a versioned attachment + * root: staged write, fsync, hard-link into place, digest-verified EEXIST + * deduplication, read-only mode, and durable directory entries from the + * target's parent up to (excluding) `root`. + * @param root - absolute `DSH_HOME/attachments/v1` root. + * @param target - absolute final object path below `root`. + * @param data - exact object bytes whose digest is `sha256`. + * @param sha256 - hex digest the stored bytes must match on deduplication. + */ +export async function publishImmutableObject( + root: string, + target: string, + data: Uint8Array, + sha256: string, +): Promise { + const staged = await stageImmutableObject(root, (function* (): Iterable { + yield data + })()) + if (staged.sha256 !== sha256) { + await removeTemporary(staged.path) + throw new AttachmentError('Attachment bytes do not match their publication digest.', 'ATTACHMENT_CORRUPT') + } + await publishStagedObject(root, target, staged) +} + +/** Digest and byte count produced while streaming one immutable object to disk. */ +export interface StreamedImmutableObject { + readonly sha256: string + readonly bytes: number +} + +/** + * Stream one immutable object from bounded chunks into a staging file, then + * publish it at a digest-derived target without collecting the complete object in memory. + * @param root - absolute `DSH_HOME/attachments/v1` root. + * @param data - exact object bytes in order. + * @param targetFor - derive the final absolute target from the completed digest and byte count. + * @param signal - optional cancellation for source reads and storage writes. + * @returns digest and exact byte count of the published object. + */ +export async function publishImmutableObjectStream( + root: string, + data: AsyncIterable, + targetFor: (sha256: string, bytes: number) => string, + signal?: AbortSignal, +): Promise { + const staged = await stageImmutableObject(root, data, signal) + let target: string + try { + target = targetFor(staged.sha256, staged.bytes) + } catch (error) { + /* v8 ignore start -- The local target callback constructs a validated reference from this function's digest. */ + await removeTemporary(staged.path) + throw error + /* v8 ignore stop */ + } + await publishStagedObject(root, target, staged) + return { sha256: staged.sha256, bytes: staged.bytes } +} + +/** + * Publish another durable hard-link name for an existing immutable object. + * @param root - absolute versioned attachment root. + * @param source - existing content-addressed object below `root`. + * @param target - new alias below `root`. + * @param sha256 - expected object digest for an existing-target race. + */ +export async function publishImmutableAlias( + root: string, + source: string, + target: string, + sha256: string, +): Promise { + const parent = dirname(target) + try { + const boundary = await ensureDurableHome(dirname(dirname(resolve(root)))) + await ensureDurableDirectory(parent, boundary) + try { + await link(source, target) + } catch (error) { + /* v8 ignore next -- Private same-filesystem directories make EEXIST the only recoverable link race. */ + if (!(error instanceof Error && 'code' in error && error.code === 'EEXIST')) throw error + if (await digestFile(target) !== sha256) { + throw new AttachmentError('Stored attachment failed integrity verification.', 'ATTACHMENT_CORRUPT') + } + } + await chmod(target, 0o400) + const stop = resolve(root) + for (let level = parent; level !== stop; level = dirname(level)) { + await syncDirectory(level) + /* v8 ignore next -- filesystem-root guard: targets sit below root, so the walk reaches `stop` first. */ + if (dirname(level) === level) break + } + } catch (error) { + if (error instanceof AttachmentError) throw error + throw new AttachmentError('Unable to persist attachment.', 'ATTACHMENT_WRITE_FAILED', { cause: error }) + } +} + +interface StagedImmutableObject extends StreamedImmutableObject { + readonly path: string + readonly boundary: string +} + +async function stageImmutableObject( + root: string, + data: AsyncIterable | Iterable, + signal?: AbortSignal, +): Promise { const staging = join(root, 'tmp') // Establish DSH_HOME itself against the filesystem root once per process. // Every process performs that proof independently, so observing a directory // another process created can never be mistaken for durable publication. const boundary = await ensureDurableHome(dirname(dirname(resolve(root)))) - await ensureDurableDirectory(bucket, boundary) await ensureDurableDirectory(staging, boundary) const temporary = join(staging, randomUUID()) - const target = normalizedImagePath(root, prepared.ref) let handle try { handle = await open(temporary, constants.O_CREAT | constants.O_EXCL | constants.O_WRONLY, 0o600) - await handle.writeFile(normalized) + const hash = createHash('sha256') + let bytes = 0 + for await (const chunk of data) { + signal?.throwIfAborted() + await handle.writeFile(chunk) + hash.update(chunk) + bytes += chunk.byteLength + } + signal?.throwIfAborted() await handle.sync() + signal?.throwIfAborted() await handle.close() handle = undefined - try { - await link(temporary, target) - } catch (error) { - /* v8 ignore next -- Private same-filesystem directories make EEXIST the only recoverable link race. */ - if (!(error instanceof Error && 'code' in error && error.code === 'EEXIST')) throw error - const existing = new Uint8Array(await readFile(target)) - if (digest(existing) !== sha256) throw new AttachmentError('Stored attachment failed integrity verification.', 'ATTACHMENT_CORRUPT') - } - // Windows shares the read-only attribute across hard links and refuses to - // unlink either name once it is set, so discard the staging name first. - await unlink(temporary) - // The target remains the sole link for a new object; this also restores - // read-only mode when the deduplication path observes an existing object. - await chmod(target, 0o400) - // Persist the target entry and close a concurrent bucket-creation window - // before the reference can reach a session checkpoint. The dedup path - // repeats both syncs because it may observe another writer's link before - // that writer reaches its own durability boundary. - await syncDirectory(bucket) - await syncDirectory(join(root, 'objects')) + return { path: temporary, boundary, sha256: hash.digest('hex'), bytes } } catch (error) { - /* v8 ignore next -- A descriptor can remain open only when the underlying write/sync/close operation fails. */ + /* v8 ignore next -- A descriptor remains open only when write, sync, or close fails. */ if (handle !== undefined) await handle.close().catch( /* v8 ignore next -- Close failure is superseded by the storage operation that entered cleanup. */ () => {}, ) - await unlink(temporary).catch( - /* v8 ignore next -- The callback requires a second independent staging-unlink failure. */ - (cleanupError: unknown) => { - /* v8 ignore next -- Cleanup is best-effort only for a staging file already removed by a failed operation. */ - if (!(cleanupError instanceof Error && 'code' in cleanupError && cleanupError.code === 'ENOENT')) throw cleanupError - }, - ) - if (error instanceof AttachmentError) throw error - throw new AttachmentError('Unable to persist image attachment.', 'ATTACHMENT_WRITE_FAILED', { cause: error }) + await removeTemporary(temporary) + if (error instanceof AttachmentError || signal?.aborted === true) throw error + throw new AttachmentError('Unable to persist attachment.', 'ATTACHMENT_WRITE_FAILED', { cause: error }) } - return prepared.ref +} + +async function publishStagedObject( + root: string, + target: string, + staged: StagedImmutableObject, +): Promise { + const parent = dirname(target) + try { + await ensureDurableDirectory(parent, staged.boundary) + try { + await link(staged.path, target) + } catch (error) { + /* v8 ignore next -- Private same-filesystem directories make EEXIST the only recoverable link race. */ + if (!(error instanceof Error && 'code' in error && error.code === 'EEXIST')) throw error + if (await digestFile(target) !== staged.sha256) { + throw new AttachmentError('Stored attachment failed integrity verification.', 'ATTACHMENT_CORRUPT') + } + } + // Windows shares the read-only attribute across hard links and refuses to + // unlink either name once it is set, so discard the staging name first. + await unlink(staged.path) + // The target remains the sole link for a new object; this also restores + // read-only mode when the deduplication path observes an existing object. + await chmod(target, 0o400) + // Persist the target entry and close every concurrent parent-creation + // window before the reference can reach a session checkpoint. The dedup + // path repeats these syncs because it may observe another writer's link + // before that writer reaches its own durability boundary. + const stop = resolve(root) + for (let level = parent; level !== stop; level = dirname(level)) { + await syncDirectory(level) + /* v8 ignore next -- filesystem-root guard: targets sit below root, so the walk reaches `stop` first. */ + if (dirname(level) === level) break + } + } catch (error) { + await removeTemporary(staged.path) + if (error instanceof AttachmentError) throw error + throw new AttachmentError('Unable to persist attachment.', 'ATTACHMENT_WRITE_FAILED', { cause: error }) + } +} + +async function digestFile(path: string): Promise { + const hash = createHash('sha256') + for await (const chunk of createReadStream(path) as AsyncIterable) hash.update(chunk) + return hash.digest('hex') +} + +async function removeTemporary(path: string): Promise { + await unlink(path).catch( + /* v8 ignore next -- Cleanup can observe a staging name already removed after successful linking. */ + (cleanupError: unknown) => { + /* v8 ignore next -- Any cleanup failure except an absent staging name must remain visible. */ + if (!(cleanupError instanceof Error && 'code' in cleanupError && cleanupError.code === 'ENOENT')) throw cleanupError + }, + ) } /** diff --git a/packages/attachment/attachment-local/tests/file-store.spec.ts b/packages/attachment/attachment-local/tests/file-store.spec.ts new file mode 100644 index 0000000000..dde6ec2d06 --- /dev/null +++ b/packages/attachment/attachment-local/tests/file-store.spec.ts @@ -0,0 +1,278 @@ +import { createHash } from 'node:crypto' +import { chmod, mkdir, readdir, readFile, stat, unlink, writeFile } from 'node:fs/promises' +import { tmpdir } from 'node:os' +import { join } from 'node:path' +import { mkdtemp, rm } from 'node:fs/promises' +import { afterEach, describe, expect, it } from 'vitest' +import { AttachmentId } from '@deepseek-ai/dsh-attachment' +import type { FileAttachmentRef } from '@deepseek-ai/dsh-attachment' +import { + fileLeafName, readFileStreamVerbatim, saveFileStreamVerbatim, saveFileVerbatim, storedFilePath, +} from '../src/file-store.ts' +import { publishImmutableAlias } from '../src/store.ts' + +const roots: string[] = [] + +async function makeRoot(): Promise { + const root = join(await mkdtemp(join(tmpdir(), 'dsh-file-store-')), 'attachments', 'v1') + roots.push(root) + return root +} + +afterEach(async () => { + for (const root of roots.splice(0)) { + await rm(join(root, '..', '..'), { recursive: true, force: true, maxRetries: 3 }) + } +}) + +function sha256(data: Uint8Array): string { + return createHash('sha256').update(data).digest('hex') +} + +async function readStream(stream: AsyncIterable): Promise { + const chunks: Uint8Array[] = [] + for await (const chunk of stream) chunks.push(chunk) + return new Uint8Array(Buffer.concat(chunks)) +} + +describe('fileLeafName', () => { + it('keeps ordinary names and strips client paths of both separator styles', () => { + expect(fileLeafName('notes.pdf')).toBe('notes.pdf') + expect(fileLeafName('/home/user/data.csv')).toBe('data.csv') + expect(fileLeafName('C:\\Users\\me\\report.docx')).toBe('report.docx') + }) + + it('removes control characters, rewrites Windows-invalid characters, and bounds UTF-8 length', () => { + expect(fileLeafName('a\u0000b\u001f.txt')).toBe('ab.txt') + expect(fileLeafName('ac:d"e|f?g*h.txt')).toBe('a_b_c_d_e_f_g_h.txt') + expect(fileLeafName(`${'x'.repeat(300)}.bin`).length).toBe(255) + const multibyte = fileLeafName(`${'文'.repeat(100)}.txt`) + expect(Buffer.byteLength(multibyte)).toBeLessThanOrEqual(255) + expect(multibyte.endsWith('\ufffd')).toBe(false) + expect(fileLeafName(`safe-${'x'.repeat(248)}\ud83d\ude00`)).not.toMatch(/\ud83d$/u) + }) + + it('removes Windows trailing characters and protects reserved device names', () => { + expect(fileLeafName('report. ')).toBe('report') + expect(fileLeafName('CON')).toBe('_CON') + expect(fileLeafName('com1.txt')).toBe('_com1.txt') + expect(fileLeafName('con .txt')).toBe('_con .txt') + expect(fileLeafName('com10.txt')).toBe('com10.txt') + }) + + it('falls back to a stable name for absent, empty, and dot-only inputs', () => { + expect(fileLeafName(undefined)).toBe('file') + expect(fileLeafName('')).toBe('file') + expect(fileLeafName(' ')).toBe('file') + expect(fileLeafName('.')).toBe('file') + expect(fileLeafName('..')).toBe('file') + }) +}) + +describe('saveFileVerbatim', () => { + it('stores the exact bytes read-only at a digest-and-name path', async () => { + const root = await makeRoot() + const data = Uint8Array.from([0, 1, 2, 250, 251, 252]) + const ref = await saveFileVerbatim(root, { data, name: 'blob.bin' }) + expect(ref).toEqual({ + attachmentId: AttachmentId(`sha256:${sha256(data)}`), + name: 'blob.bin', + bytes: data.byteLength, + }) + const path = storedFilePath(root, ref) + expect(path.endsWith(join(sha256(data), 'blob.bin'))).toBe(true) + expect(new Uint8Array(await readFile(path))).toEqual(data) + if (process.platform !== 'win32') { + expect((await stat(path)).mode & 0o777).toBe(0o400) + } + }) + + it('accepts a zero-byte file', async () => { + const root = await makeRoot() + const ref = await saveFileVerbatim(root, { data: new Uint8Array(0), name: 'empty.txt' }) + expect(ref.bytes).toBe(0) + expect((await readFile(storedFilePath(root, ref))).byteLength).toBe(0) + }) + + it('stores sanitized Windows-reserved and multibyte names', async () => { + const root = await makeRoot() + const reserved = await saveFileVerbatim(root, { data: new Uint8Array(0), name: 'NUL.txt' }) + const multibyte = await saveFileVerbatim(root, { data: Uint8Array.of(1), name: '文'.repeat(100) }) + expect(reserved.name).toBe('_NUL.txt') + expect(Buffer.byteLength(multibyte.name)).toBeLessThanOrEqual(255) + await expect(readFile(storedFilePath(root, reserved))).resolves.toHaveLength(0) + await expect(readFile(storedFilePath(root, multibyte))).resolves.toEqual(Buffer.from([1])) + }) + + it('deduplicates identical bytes and stores distinct names beside one digest', async () => { + const root = await makeRoot() + const data = Uint8Array.from([7, 7, 7]) + const first = await saveFileVerbatim(root, { data, name: 'a.txt' }) + const again = await saveFileVerbatim(root, { data, name: 'a.txt' }) + expect(again).toEqual(first) + const renamed = await saveFileVerbatim(root, { data, name: 'b.txt' }) + expect(renamed.attachmentId).toBe(first.attachmentId) + expect((await stat(storedFilePath(root, first))).ino) + .toBe((await stat(storedFilePath(root, renamed))).ino) + const digestDir = join(root, 'files', sha256(data).slice(0, 2), sha256(data)) + expect((await readdir(digestDir)).sort()).toEqual(['a.txt', 'b.txt']) + }) + + it('refuses a stored object whose bytes no longer match the digest', async () => { + const root = await makeRoot() + const data = Uint8Array.from([1, 2, 3]) + const ref = await saveFileVerbatim(root, { data, name: 'c.txt' }) + const path = storedFilePath(root, ref) + await chmod(path, 0o600) + await writeFile(path, Uint8Array.from([9, 9, 9])) + await expect(saveFileVerbatim(root, { data, name: 'c.txt' })) + .rejects.toMatchObject({ code: 'ATTACHMENT_CORRUPT' }) + }) + + it('rejects a conflicting display-name alias and wraps alias publication failures', async () => { + const root = await makeRoot() + const data = Uint8Array.of(1, 2, 3) + const ref = await saveFileVerbatim(root, { data, name: 'alias.bin' }) + const alias = storedFilePath(root, ref) + await unlink(alias) + await writeFile(alias, Uint8Array.of(9, 9, 9)) + await expect(saveFileVerbatim(root, { data, name: 'alias.bin' })) + .rejects.toMatchObject({ code: 'ATTACHMENT_CORRUPT' }) + + await expect(publishImmutableAlias( + root, + join(root, 'missing-object'), + join(root, 'files', 'ff', 'missing', 'alias.bin'), + 'f'.repeat(64), + )).rejects.toMatchObject({ code: 'ATTACHMENT_WRITE_FAILED' }) + }) +}) + +describe('saveFileStreamVerbatim', () => { + it('stores ordered chunks without materializing one aggregate byte array', async () => { + const root = await makeRoot() + const ref = await saveFileStreamVerbatim(root, { + data: (async function* (): AsyncIterable { + yield Uint8Array.of(0, 1) + yield Uint8Array.of(2, 250) + yield Uint8Array.of(251, 252) + })(), + name: 'large.bin', + }) + const expected = Uint8Array.of(0, 1, 2, 250, 251, 252) + expect(ref).toEqual({ + attachmentId: AttachmentId(`sha256:${sha256(expected)}`), + name: 'large.bin', + bytes: expected.byteLength, + }) + expect(new Uint8Array(await readFile(storedFilePath(root, ref)))).toEqual(expected) + }) + + it('removes its staging file when cancellation interrupts the source', async () => { + const root = await makeRoot() + const abort = new AbortController() + const reason = new Error('upload cancelled') + await expect(saveFileStreamVerbatim(root, { + data: (async function* (): AsyncIterable { + yield Uint8Array.of(1, 2) + abort.abort(reason) + yield Uint8Array.of(3, 4) + })(), + signal: abort.signal, + name: 'cancelled.bin', + })).rejects.toBe(reason) + expect(await readdir(join(root, 'tmp'))).toEqual([]) + }) + + it('wraps source failures and removes the staging file', async () => { + const root = await makeRoot() + await expect(saveFileStreamVerbatim(root, { + data: (async function* (): AsyncIterable { + yield Uint8Array.of(1, 2) + throw new Error('source failed') + })(), + name: 'failed.bin', + })).rejects.toMatchObject({ code: 'ATTACHMENT_WRITE_FAILED' }) + expect(await readdir(join(root, 'tmp'))).toEqual([]) + }) +}) + +describe('readFileStreamVerbatim', () => { + it('returns exact bounded chunks and accepts an empty file', async () => { + const root = await makeRoot() + const data = Uint8Array.from({ length: (1 << 16) + 3 }, (_, index) => index % 251) + const ref = await saveFileVerbatim(root, { data, name: 'large.bin' }) + await expect(readStream(readFileStreamVerbatim(root, ref))).resolves.toEqual(data) + const empty = await saveFileVerbatim(root, { data: new Uint8Array(), name: 'empty.bin' }) + await expect(readStream(readFileStreamVerbatim(root, empty))).resolves.toEqual(new Uint8Array()) + }) + + it('rejects invalid, missing, and unreadable references with storage codes', async () => { + const root = await makeRoot() + const ref: FileAttachmentRef = { + attachmentId: AttachmentId(`sha256:${'c'.repeat(64)}`), + name: 'missing.bin', + bytes: 1, + } + await expect(readStream(readFileStreamVerbatim(root, { ...ref, name: '../escape' }))) + .rejects.toMatchObject({ code: 'INVALID_ATTACHMENT_REF' }) + await expect(readStream(readFileStreamVerbatim(root, ref))) + .rejects.toMatchObject({ code: 'ATTACHMENT_NOT_FOUND' }) + + const saved = await saveFileVerbatim(root, { data: Uint8Array.of(1), name: 'unreadable.bin' }) + const path = storedFilePath(root, saved) + await unlink(path) + await mkdir(path) + await expect(readStream(readFileStreamVerbatim(root, saved))) + .rejects.toMatchObject({ code: 'ATTACHMENT_READ_FAILED' }) + }) + + it('detects changed bytes and recorded lengths', async () => { + const root = await makeRoot() + const ref = await saveFileVerbatim(root, { data: Uint8Array.of(1, 2, 3), name: 'data.bin' }) + const path = storedFilePath(root, ref) + await chmod(path, 0o600) + await writeFile(path, Uint8Array.of(3, 2, 1)) + await expect(readStream(readFileStreamVerbatim(root, ref))) + .rejects.toMatchObject({ code: 'ATTACHMENT_CORRUPT' }) + await writeFile(path, Uint8Array.of(1, 2, 3)) + await expect(readStream(readFileStreamVerbatim(root, { ...ref, bytes: 4 }))) + .rejects.toMatchObject({ code: 'ATTACHMENT_CORRUPT' }) + }) + + it('preserves caller cancellation before and during a read', async () => { + const root = await makeRoot() + const ref = await saveFileVerbatim(root, { + data: Uint8Array.from({ length: 1 << 17 }, () => 7), + name: 'cancel.bin', + }) + const before = new AbortController() + const beforeReason = new Error('cancelled before read') + before.abort(beforeReason) + await expect(readStream(readFileStreamVerbatim(root, ref, before.signal))).rejects.toBe(beforeReason) + + const during = new AbortController() + const stream = readFileStreamVerbatim(root, ref, during.signal)[Symbol.asyncIterator]() + await expect(stream.next()).resolves.toMatchObject({ done: false }) + const duringReason = new Error('cancelled during read') + during.abort(duringReason) + await expect(stream.next()).rejects.toBe(duringReason) + }) +}) + +describe('storedFilePath', () => { + it('rejects malformed digests and unsanitized names before deriving a path', () => { + const good: FileAttachmentRef = { + attachmentId: AttachmentId(`sha256:${'a'.repeat(64)}`), + name: 'ok.txt', + bytes: 1, + } + expect(() => storedFilePath('/root', { ...good, attachmentId: AttachmentId('sha256:short') })) + .toThrow(expect.objectContaining({ code: 'INVALID_ATTACHMENT_REF' }) as Error) + expect(() => storedFilePath('/root', { ...good, name: '../escape.txt' })) + .toThrow(expect.objectContaining({ code: 'INVALID_ATTACHMENT_REF' }) as Error) + expect(() => storedFilePath('/root', { ...good, name: 'nested/name.txt' })) + .toThrow(expect.objectContaining({ code: 'INVALID_ATTACHMENT_REF' }) as Error) + expect(storedFilePath('/root', good).endsWith(join('a'.repeat(64), 'ok.txt'))).toBe(true) + }) +}) diff --git a/packages/attachment/attachment-local/tests/index.spec.ts b/packages/attachment/attachment-local/tests/index.spec.ts index 4d9d9f350b..5711c6d35d 100644 --- a/packages/attachment/attachment-local/tests/index.spec.ts +++ b/packages/attachment/attachment-local/tests/index.spec.ts @@ -87,6 +87,21 @@ describe('local attachment service', () => { await expect(readFile(hostPath)).resolves.toEqual(Buffer.from(data)) const request = await service.readImageRequest(ref, { maxPixels: 1, maxBytes: 1024 }) expect(request).not.toHaveProperty('access') + + const fileData = Uint8Array.of(0, 1, 2, 255) + const fileRef = await service.saveFile({ data: fileData, name: 'notes.bin' }) + const filePath = service.fileHostPath(fileRef) + expect(filePath).toContain(join('files', String(fileRef.attachmentId).slice(7, 9))) + await expect(readFile(filePath)).resolves.toEqual(Buffer.from(fileData)) + + const streamRef = await service.saveFileStream({ + data: (async function* (): AsyncIterable { yield fileData })(), + name: 'stream.bin', + }) + await expect(readFile(service.fileHostPath(streamRef))).resolves.toEqual(Buffer.from(fileData)) + const streamed: Uint8Array[] = [] + for await (const chunk of service.readFileStream(streamRef)) streamed.push(chunk) + expect(Buffer.concat(streamed)).toEqual(Buffer.from(fileData)) } finally { await rm(dshHome, { recursive: true, force: true }) } diff --git a/packages/attachment/attachment-local/tests/store.spec.ts b/packages/attachment/attachment-local/tests/store.spec.ts index 3ff58eb2da..82c2b9f3ba 100644 --- a/packages/attachment/attachment-local/tests/store.spec.ts +++ b/packages/attachment/attachment-local/tests/store.spec.ts @@ -1,6 +1,6 @@ import { createHash } from 'node:crypto' import { constants } from 'node:fs' -import { chmod, mkdir, readFile, stat, writeFile } from 'node:fs/promises' +import { chmod, mkdir, readFile, readdir, stat, writeFile } from 'node:fs/promises' import { tmpdir } from 'node:os' import { dirname, join, parse, resolve } from 'node:path' import { mkdtemp, rm } from 'node:fs/promises' @@ -8,7 +8,13 @@ import { afterEach, describe, expect, it, vi } from 'vitest' import sharp from 'sharp' import type { ImageAttachmentLimits } from '@deepseek-ai/dsh-attachment' import type { NormalizationPolicy } from '../src/normalization.ts' -import { commitPreparedImageFile, prepareImageFile, readImageFile, saveImageFile } from '../src/store.ts' +import { + commitPreparedImageFile, + prepareImageFile, + publishImmutableObject, + readImageFile, + saveImageFile, +} from '../src/store.ts' const fsControl = vi.hoisted(() => ({ readSignals: [] as AbortSignal[], @@ -89,12 +95,13 @@ describe('local attachment store', () => { // Later directory creation can then stop at that process-proven boundary. expect(fsControl.syncedDirectories).toEqual([ ...parentChainToRoot(base), - // bucket chain: every parent entry between the bucket and the boundary. - objects, + // Staging precedes publication because the streamed digest selects the + // target bucket only after every byte has been written. storageRoot, join(storageRoot, '..'), base, - // staging chain re-walks the shared ancestors after creating tmp. + // bucket chain: every parent entry between the bucket and the boundary. + objects, storageRoot, join(storageRoot, '..'), base, @@ -141,6 +148,14 @@ describe('local attachment store', () => { await expect(readImageFile(storageRoot, first)).resolves.toEqual({ ref: first, data: PNG }) }) + it('rejects publication when the supplied digest does not match the staged bytes', async () => { + const storageRoot = await root() + const target = join(storageRoot, 'objects', '00', 'mismatch') + await expect(publishImmutableObject(storageRoot, target, Uint8Array.of(1), '0'.repeat(64))) + .rejects.toMatchObject({ code: 'ATTACHMENT_CORRUPT' }) + expect(await readdir(join(storageRoot, 'tmp'))).toEqual([]) + }) + it.skipIf(process.platform !== 'win32')('publishes a new object on Windows', async () => { const storageRoot = await root() diff --git a/packages/attachment/attachment/src/admission.ts b/packages/attachment/attachment/src/admission.ts index 155b3644d8..a3f5f16c8c 100644 --- a/packages/attachment/attachment/src/admission.ts +++ b/packages/attachment/attachment/src/admission.ts @@ -5,21 +5,30 @@ import { AttachmentError } from './error.ts' import type { AttachmentStore } from './index.ts' import type { AdmittedPromptContentPart, + EncodedFileAttachment, EncodedImageAttachment, + FileAttachmentRef, ImageAttachmentRef, PromptContentPart, SaveImageAttachment, } from './types.ts' /** Decode one upload payload while rejecting non-canonical base64 forms. */ -function decodeBase64(data: string): Uint8Array { +function decodeCanonicalBase64(data: string, empty: 'reject' | 'accept', code: 'INVALID_IMAGE_BASE64' | 'INVALID_FILE_BASE64'): Uint8Array { const decoded = Buffer.from(data, 'base64') - if (data.length === 0 || decoded.toString('base64') !== data) { - throw new AttachmentError('Image upload is not canonical base64.', 'INVALID_IMAGE_BASE64') + if ((data.length === 0 && empty === 'reject') || decoded.toString('base64') !== data) { + throw new AttachmentError( + code === 'INVALID_IMAGE_BASE64' ? 'Image upload is not canonical base64.' : 'File upload is not canonical base64.', + code, + ) } return new Uint8Array(decoded) } +function decodeBase64(data: string): Uint8Array { + return decodeCanonicalBase64(data, 'reject', 'INVALID_IMAGE_BASE64') +} + /** Store input for one decoded upload. */ function saveInput(image: EncodedImageAttachment): SaveImageAttachment { return { @@ -46,6 +55,26 @@ export async function admitEncodedImages( return attachments.saveImages(images.map(saveInput)) } +/** + * Admit one wire file upload: enforce canonical base64 (an empty file is a + * valid zero-byte payload), then delegate verbatim commit to + * {@link AttachmentStore.saveFile}. The shared entry for every RPC endpoint + * accepting browser file uploads. + * @param attachments - the deployment attachment store. + * @param file - base64-encoded upload and optional display name. + * @returns the durable content-addressed file reference. + * @throws AttachmentError on a non-canonical payload or a storage failure. + */ +export async function admitEncodedFile( + attachments: AttachmentStore, + file: EncodedFileAttachment, +): Promise { + return attachments.saveFile({ + data: decodeCanonicalBase64(file.data, 'accept', 'INVALID_FILE_BASE64'), + ...file.name === undefined ? {} : { name: file.name }, + }) +} + /** * Admit one browser prompt and replace each uploaded image with its durable reference. * Text-only prompts do not access the attachment store. diff --git a/packages/attachment/attachment/src/error.ts b/packages/attachment/attachment/src/error.ts index c19229872b..7a1588b55e 100644 --- a/packages/attachment/attachment/src/error.ts +++ b/packages/attachment/attachment/src/error.ts @@ -18,12 +18,14 @@ export type ImageAdmissionErrorCode = typeof IMAGE_ADMISSION_ERROR_CODES[number] /** Stable attachment failure codes used for protocol error routing. */ export type AttachmentErrorCode = | ImageAdmissionErrorCode + | 'INVALID_FILE_BASE64' | 'INVALID_ATTACHMENT_REF' | 'ATTACHMENT_CORRUPT' | 'ATTACHMENT_WRITE_FAILED' | 'ATTACHMENT_NOT_FOUND' | 'ATTACHMENT_READ_FAILED' | 'ATTACHMENT_PROJECTION_UNSUPPORTED' + | 'ATTACHMENT_FILES_UNSUPPORTED' /** Runtime membership for structurally compatible errors crossing package boundaries. */ const IMAGE_ADMISSION_ERROR_CODE_SET: ReadonlySet = new Set(IMAGE_ADMISSION_ERROR_CODES) diff --git a/packages/attachment/attachment/src/index.ts b/packages/attachment/attachment/src/index.ts index 9dcddeae65..8ae640c947 100644 --- a/packages/attachment/attachment/src/index.ts +++ b/packages/attachment/attachment/src/index.ts @@ -3,10 +3,13 @@ import { Context, Service } from '@deepseek-ai/cordis' import { AttachmentError } from './error.ts' import type { + FileAttachmentRef, ImageAttachmentLimits, ImageAttachmentRef, ImageRequestPolicy, RequestImageAttachment, + SaveFileAttachment, + SaveFileStreamAttachment, SaveImageAttachment, StoredImageAttachment, } from './types.ts' @@ -14,18 +17,22 @@ import type { export { AttachmentId, ImageVariantId } from './brand.ts' export { AttachmentError, isImageAdmissionError } from './error.ts' export type { AttachmentErrorCode, ImageAdmissionErrorCode } from './error.ts' -export { admitEncodedImages, admitPromptContent } from './admission.ts' +export { admitEncodedFile, admitEncodedImages, admitPromptContent } from './admission.ts' export { requestImageDimensions } from './request-projection.ts' export type { AttachmentId as AttachmentIdType, AdmittedPromptContentPart, + EncodedFileAttachment, EncodedImageAttachment, + FileAttachmentRef, ImageAttachmentLimits, ImageAttachmentRef, ImageRequestPolicy, ImageMediaType, PromptContentPart, RequestImageAttachment, + SaveFileAttachment, + SaveFileStreamAttachment, SaveImageAttachment, StoredImageAttachment, } from './types.ts' @@ -121,6 +128,68 @@ export abstract class AttachmentStore extends Service { return undefined } + /** + * Durably commit one file byte-for-byte before its owning session event is + * appended. Files carry no admission limits: any byte content and length is + * accepted, and the stored object is the exact submitted bytes. Backends + * without verbatim file storage keep this default rejection. + * @param input - exact bytes and optional display name. + * @returns the durable content-addressed file reference. + */ + saveFile(input: SaveFileAttachment): Promise { + void input + return Promise.reject(new AttachmentError( + 'The mounted attachment provider cannot store verbatim files.', + 'ATTACHMENT_FILES_UNSUPPORTED', + )) + } + + /** + * Durably commit one file byte-for-byte from bounded chunks. Providers must + * apply backpressure and must not collect the complete file in memory. + * Backends without streamed verbatim storage keep this default rejection. + * @param input - ordered exact bytes, optional cancellation, and display name. + * @returns the durable content-addressed file reference. + */ + saveFileStream(input: SaveFileStreamAttachment): Promise { + void input + return Promise.reject(new AttachmentError( + 'The mounted attachment provider cannot stream verbatim files.', + 'ATTACHMENT_FILES_UNSUPPORTED', + )) + } + + /** + * Read and verify one verbatim stored file as bounded chunks. Providers must + * not collect the complete file in memory. Backends without verbatim file + * reads keep this default rejection. + * @param ref - durable reference from the session log. + * @param signal - optional cancellation for backend reads and verification work. + * @returns exact file bytes in order; integrity failures reject the iteration. + */ + async *readFileStream( + ref: FileAttachmentRef, + signal?: AbortSignal, + ): AsyncIterable { + signal?.throwIfAborted() + void ref + await Promise.reject(new AttachmentError( + 'The mounted attachment provider cannot read verbatim files.', + 'ATTACHMENT_FILES_UNSUPPORTED', + )) + } + + /** + * Locate the verbatim stored file object in the harness host filesystem. + * @param ref - durable file reference. + * @returns an absolute host path, or undefined when this backend is not host-file-backed. + * @throws an AttachmentError when the durable reference is invalid. + */ + fileHostPath(ref: FileAttachmentRef): string | undefined { + void ref + return undefined + } + /** * Generate or read one deterministic model-request version from the stored normalized image. * @param ref - durable provider-independent normalized attachment reference. diff --git a/packages/attachment/attachment/src/types.ts b/packages/attachment/attachment/src/types.ts index 7a55c6a68f..03fb45c761 100644 --- a/packages/attachment/attachment/src/types.ts +++ b/packages/attachment/attachment/src/types.ts @@ -31,6 +31,45 @@ export interface ImageAttachmentRef { } } +/** + * Durable, serializable reference to one verbatim stored file. Files are + * stored byte-for-byte with no normalization; `attachmentId` is the sha256 + * digest of exactly those bytes. + */ +export interface FileAttachmentRef { + /** Opaque content-addressed storage identifier; never a filesystem path or bearer URL. */ + attachmentId: AttachmentId + /** Sanitized display filename, also the stored object's leaf name. */ + name: string + /** Exact byte length. */ + bytes: number +} + +/** Base64-encoded file upload accompanying one wire request. */ +export interface EncodedFileAttachment { + /** Canonical base64 encoding of the file bytes. */ + data: string + /** Optional display name; it is never interpreted as a path. */ + name?: string +} + +/** Request to durably commit one file verbatim. */ +export interface SaveFileAttachment { + data: Uint8Array + /** Optional browser/provider display name; it is never interpreted as a path. */ + name?: string +} + +/** Request to durably commit one file from bounded byte chunks. */ +export interface SaveFileStreamAttachment { + /** Exact file bytes in order; providers must not retain the complete sequence in memory. */ + data: AsyncIterable + /** Optional cancellation for source reads and storage writes. */ + signal?: AbortSignal + /** Optional browser/provider display name; it is never interpreted as a path. */ + name?: string +} + /** Deployment-resolved limits used by upload admission and request buffering. */ export interface ImageAttachmentLimits { maxImageBytes: number diff --git a/packages/attachment/attachment/tests/admission.spec.ts b/packages/attachment/attachment/tests/admission.spec.ts index ba24d8fa20..60eadd421f 100644 --- a/packages/attachment/attachment/tests/admission.spec.ts +++ b/packages/attachment/attachment/tests/admission.spec.ts @@ -1,6 +1,6 @@ import { describe, expect, it, vi } from 'vitest' import type { AttachmentStore } from '@deepseek-ai/dsh-attachment' -import { admitEncodedImages, admitPromptContent } from '@deepseek-ai/dsh-attachment' +import { admitEncodedFile, admitEncodedImages, admitPromptContent } from '@deepseek-ai/dsh-attachment' import type { ImageAttachmentRef, SaveImageAttachment } from '@deepseek-ai/dsh-attachment/types' const PNG = 'AAAA' // canonical base64, 3 bytes @@ -65,6 +65,44 @@ describe('admitEncodedImages', () => { }) }) +describe('admitEncodedFile', () => { + /** Delegation double: records the exact saveFile input and answers a fixed ref. */ + function fileStoreOf() { + const store = { + saveFile: vi.fn((input: { data: Uint8Array; name?: string }) => Promise.resolve({ + attachmentId: 'file-1' as never, + name: input.name ?? 'file', + bytes: input.data.byteLength, + })), + } + return { store: store as unknown as AttachmentStore, mocks: store } + } + + it('decodes canonical base64 and delegates verbatim commit to saveFile', async () => { + const { store, mocks } = fileStoreOf() + const ref = await admitEncodedFile(store, { data: 'AAAA', name: 'blob.bin' }) + expect(mocks.saveFile).toHaveBeenCalledTimes(1) + const input = mocks.saveFile.mock.calls[0]?.[0] as { data: Uint8Array; name?: string } + expect([input.name, input.data.byteLength]).toEqual(['blob.bin', 3]) + expect(ref.bytes).toBe(3) + }) + + it('accepts an empty payload as a zero-byte file and omits an absent name', async () => { + const { store, mocks } = fileStoreOf() + const ref = await admitEncodedFile(store, { data: '' }) + const input = mocks.saveFile.mock.calls[0]?.[0] as object + expect('name' in input).toBe(false) + expect(ref.bytes).toBe(0) + }) + + it('rejects non-canonical base64 without touching the store', async () => { + const { store, mocks } = fileStoreOf() + await expect(admitEncodedFile(store, { data: 'not base64!!' })) + .rejects.toMatchObject({ code: 'INVALID_FILE_BASE64' }) + expect(mocks.saveFile).not.toHaveBeenCalled() + }) +}) + describe('admitPromptContent', () => { it('converts text-only prompts without touching the attachment store', async () => { const store = { saveImages: () => { throw new Error('text-only prompts must not reach the store') } } diff --git a/packages/attachment/attachment/tests/index.spec.ts b/packages/attachment/attachment/tests/index.spec.ts index 8675d45404..fcccd60b6e 100644 --- a/packages/attachment/attachment/tests/index.spec.ts +++ b/packages/attachment/attachment/tests/index.spec.ts @@ -147,10 +147,33 @@ describe('AttachmentStore.readImageRequest', () => { expect(() => store.readImageRequest(ref, { maxPixels: 1, maxBytes: 1 }, controller.signal)).toThrow(reason) }) - it('exposes no provider-owned host path by default', async () => { + it('rejects generic-file storage and exposes no provider-owned host path by default', async () => { const store = new RecordingStore(new Context()) const ref = await store.saveImage(image(1)) expect(store.imageHostPath(ref)).toBeUndefined() + await expect(store.saveFile({ data: Uint8Array.of(1), name: 'notes.txt' })) + .rejects.toMatchObject({ code: 'ATTACHMENT_FILES_UNSUPPORTED' }) + await expect(store.saveFileStream({ + data: (async function* (): AsyncIterable { yield Uint8Array.of(1) })(), + name: 'notes.txt', + })).rejects.toMatchObject({ code: 'ATTACHMENT_FILES_UNSUPPORTED' }) + const fileRef = { + attachmentId: AttachmentId(`sha256:${'ab'.repeat(32)}`), + name: 'notes.txt', + bytes: 1, + } + expect(store.fileHostPath(fileRef)).toBeUndefined() + const read = async (signal?: AbortSignal): Promise => { + for await (const chunk of store.readFileStream(fileRef, signal)) { + void chunk + throw new Error('unsupported store yielded a chunk') + } + } + await expect(read()).rejects.toMatchObject({ code: 'ATTACHMENT_FILES_UNSUPPORTED' }) + const controller = new AbortController() + const reason = new Error('cancel unsupported file read') + controller.abort(reason) + await expect(read(controller.signal)).rejects.toBe(reason) }) }) diff --git a/packages/llm/llm/src/content.ts b/packages/llm/llm/src/content.ts index 5880f82c98..8ec0c14d35 100644 --- a/packages/llm/llm/src/content.ts +++ b/packages/llm/llm/src/content.ts @@ -2,7 +2,9 @@ import type { ContentBlock } from './types.ts' import type { Message } from './message.ts' -import type { AttachmentStore, ImageAttachmentRef, ImageMediaType, RequestImageAttachment } from '@deepseek-ai/dsh-attachment' +import type { + AttachmentStore, FileAttachmentRef, ImageAttachmentRef, ImageMediaType, RequestImageAttachment, +} from '@deepseek-ai/dsh-attachment' import { assertNever } from '@deepseek-ai/dsh-util-values' /** Execution-world path that model tools can use to read one normalized attachment. */ @@ -126,6 +128,78 @@ export function contentHasImage(content: readonly ContentBlock[]): boolean { || (block.type === 'tool-result' && contentHasImage(block.content))) } +/** + * True when typed model content contains a file block, walking nested + * tool-result content on the same recursion every file policy shares. + * @param content - typed model content blocks. + * @returns whether any nested block is a file. + */ +export function contentHasFile(content: readonly ContentBlock[]): boolean { + return content.some(block => block.type === 'file' + || (block.type === 'tool-result' && contentHasFile(block.content))) +} + +/** + * Stable model-facing handle for one durable file reference: the address of + * the verbatim stored copy and the instruction to read it on demand. This is + * the only representation a provider ever receives for a file. + * @param ref - durable verbatim file reference. + * @param readonlyPath - execution-world path of the stored copy, when resolvable. + * @returns deterministic handle text naming the file, its size, and its address. + */ +export function fileHandleText(ref: FileAttachmentRef, readonlyPath: string | undefined): string { + const digest = String(ref.attachmentId).slice('sha256:'.length, 'sha256:'.length + 8) + const identity = `File ${quoted(ref.name)} (${ref.bytes} bytes, sha256:${digest})` + if (readonlyPath === undefined) { + return `[${identity} was uploaded, but the current execution environment cannot access a readable path. Report that limitation if its contents are needed; do not claim to have read it.]` + } + return `[${identity}: verbatim read-only copy saved at ${quoted(readonlyPath)}. Read that path with your file tools when its contents are needed; copy it to a writable location before modifying it. When delegating file work, include this saved path in the delegation prompt; only subagents sharing this execution environment can read it.]` +} + +/** Replace every file occurrence, including nested tool results, with handle text. */ +function replaceFilesWithHandles( + blocks: readonly ContentBlock[], + resolvePath: (ref: FileAttachmentRef) => string | undefined, +): ContentBlock[] { + let next: ContentBlock[] | undefined + for (const [index, block] of blocks.entries()) { + if (block.type === 'file') { + next ??= blocks.slice(0, index) + next.push({ type: 'text', text: fileHandleText(block.attachment, resolvePath(block.attachment)) }) + continue + } + if (block.type === 'tool-result') { + const content = replaceFilesWithHandles(block.content, resolvePath) + if (content !== block.content) { + next ??= blocks.slice(0, index) + next.push({ ...block, content }) + continue + } + } + next?.push(block) + } + return next ?? blocks as ContentBlock[] +} + +/** + * Project durable file history into deterministic handle text for every model + * route. Unlike images, no provider receives file blocks natively, so this + * projection is unconditional in request assembly. + * @param messages - complete request history. + * @param resolvePath - resolve one reference's current execution-world read path. + * @returns the original list without files, otherwise shallow message copies with handle text. + */ +export function projectFilesToText( + messages: readonly Message[], + resolvePath: (ref: FileAttachmentRef) => string | undefined, +): readonly Message[] { + if (!messages.some(message => contentHasFile(message.content))) return messages + return messages.map((message) => { + const content = replaceFilesWithHandles(message.content, resolvePath) + return content === message.content ? message : { ...message, content } + }) +} + /** Base64 length of raw image bytes, including padding. */ function base64Length(bytes: number): number { return Math.ceil(bytes / 3) * 4 diff --git a/packages/llm/llm/src/index.ts b/packages/llm/llm/src/index.ts index cdecc4c96e..23f0228ca9 100644 --- a/packages/llm/llm/src/index.ts +++ b/packages/llm/llm/src/index.ts @@ -32,7 +32,10 @@ import type { LlmCallConfig, LlmCallConfigAdapterDefaults } from './call-config. import { HarnessError, INVALID_CREDENTIAL_CODE } from './error.ts' import { normalizeLlmFailure } from './adapter-failure.ts' import { normalizeApiKey } from './api-key.ts' -import { contentHasImage, projectImagesForTextModel } from './content.ts' +import { + contentHasFile, contentHasImage, fileHandleText, projectFilesToText, projectImagesForTextModel, +} from './content.ts' +import type { FileAttachmentRef } from '@deepseek-ai/dsh-attachment' export * from './attribution.ts' export * from './brand.ts' @@ -660,6 +663,16 @@ export class LlmRuntime extends TypertRemoteService { return this.adapters.get(provider)?.adapter.imageRequestPricing(provider, model) } + /** + * Resolve the exact text one durable file occurrence contributes to every + * provider request in the current execution environment. + * @param ref - durable verbatim file reference from model history. + * @returns the same deterministic handle text used at adapter dispatch. + */ + fileRequestText(ref: FileAttachmentRef): string { + return fileHandleText(ref, this.fileReadPath(ref)) + } + /** Detach typed adapter-owned modality metadata. */ private detachedModalities(modalities: readonly ModelModality[] | undefined): ModelModality[] | undefined { return modalities === undefined ? undefined : [...modalities] @@ -956,6 +969,26 @@ export class LlmRuntime extends TypertRemoteService { return Object.isFrozen(options) ? deepFreeze(filtered) : filtered } + /** + * Resolve the current execution-world read path of one durable file + * reference through the mounted attachment and filesystem providers. + */ + private fileReadPath(ref: FileAttachmentRef): string | undefined { + let hostPath: string | undefined + try { + hostPath = this.ctx.get('attachments')?.fileHostPath(ref) + } catch { + // A malformed durable reference degrades this occurrence to the no-path + // handle instead of failing every later request over the same log. + return undefined + } + if (hostPath === undefined) return undefined + // Structural face: dsh-llm cannot depend on the filesystem package, and + // only this one mapping method is consumed. + const fs = this.ctx.get('fs') as { processPathFromHostPath(hostPath: string): string | undefined } | undefined + return fs?.processPathFromHostPath(hostPath) + } + /** * Final adapter boundary. Adapter selection, dispatch, iterator construction, * and iteration failures become one terminal failure chunk. Middleware and @@ -993,13 +1026,21 @@ export class LlmRuntime extends TypertRemoteService { : Object.isFrozen(options) ? deepFreeze({ ...options, ...resolvedConfig }) : { ...options, ...resolvedConfig } - const projectedOptions = modelInfo.inputModalities !== undefined + // Files are never dispatched natively: every route receives handle text. + let projectedMessages: readonly Message[] = resolvedOptions.messages + if (projectedMessages.some(message => contentHasFile(message.content))) { + projectedMessages = projectFilesToText(projectedMessages, ref => this.fileReadPath(ref)) + } + if (modelInfo.inputModalities !== undefined && !modelInfo.inputModalities.includes('image') - && resolvedOptions.messages.some(message => contentHasImage(message.content)) - ? Object.isFrozen(resolvedOptions) - ? deepFreeze({ ...resolvedOptions, messages: projectImagesForTextModel(resolvedOptions.messages) as Message[] }) - : { ...resolvedOptions, messages: projectImagesForTextModel(resolvedOptions.messages) as Message[] } - : resolvedOptions + && projectedMessages.some(message => contentHasImage(message.content))) { + projectedMessages = projectImagesForTextModel(projectedMessages) + } + const projectedOptions = projectedMessages === resolvedOptions.messages + ? resolvedOptions + : Object.isFrozen(resolvedOptions) + ? deepFreeze({ ...resolvedOptions, messages: projectedMessages as Message[] }) + : { ...resolvedOptions, messages: projectedMessages as Message[] } const stream = dispatch(this.forAdapter(projectedOptions, adapter)) iterator = stream[Symbol.asyncIterator]() } catch (error: unknown) { diff --git a/packages/llm/llm/src/types.ts b/packages/llm/llm/src/types.ts index 4303538892..c480f57b89 100644 --- a/packages/llm/llm/src/types.ts +++ b/packages/llm/llm/src/types.ts @@ -5,7 +5,7 @@ */ import type { Branded } from '@deepseek-ai/dsh-brand' -import type { ImageAttachmentRef } from '@deepseek-ai/dsh-attachment' +import type { FileAttachmentRef, ImageAttachmentRef } from '@deepseek-ai/dsh-attachment' import type { ToolCallId, ProviderRequestId, ReasoningEffortId } from './brand.ts' import type { Message } from './message.ts' @@ -74,6 +74,19 @@ export interface ImageBlock { attachment: ImageAttachmentRef } +/** + * A durable verbatim file reference, valid in user content. Files never reach + * a provider natively: request assembly projects every occurrence to + * deterministic handle text (name, byte size, and the read-only saved path), + * so adapters and providers see text in its place while the durable log keeps + * the structured reference for presentation and authorization. + */ +export interface FileBlock { + type: 'file' + /** Immutable verbatim bytes and display metadata owned by the attachment service. */ + attachment: FileAttachmentRef +} + /** A tool invocation requested by the model. */ export interface ToolCallBlock { type: 'tool-call' @@ -100,6 +113,7 @@ export interface ContentBlockMap { 'text': TextBlock 'reasoning': ReasoningBlock 'image': ImageBlock + 'file': FileBlock 'tool-call': ToolCallBlock 'tool-result': ToolResultBlock } diff --git a/packages/llm/llm/tests/content.spec.ts b/packages/llm/llm/tests/content.spec.ts index 748e20f412..af114958b1 100644 --- a/packages/llm/llm/tests/content.spec.ts +++ b/packages/llm/llm/tests/content.spec.ts @@ -3,7 +3,10 @@ import { AttachmentId, ImageVariantId } from '@deepseek-ai/dsh-attachment' import type { AttachmentStore, ImageMediaType } from '@deepseek-ai/dsh-attachment' import { ToolCallId, + contentHasFile, createUserMessage, + fileHandleText, + projectFilesToText, offloadedImageText, offloadedImagePrefixCount, offloadRequestImagesWithPolicy, @@ -360,3 +363,80 @@ describe('projectImagesForTextModel', () => { ]) }) }) + +describe('file projection', () => { + function fileBlock(name: string): Extract { + return { + type: 'file', + attachment: { + attachmentId: AttachmentId(`sha256:${'ab'.repeat(32)}`), + name, + bytes: 42, + }, + } + } + + it('detects file blocks at the top level and inside nested tool results', () => { + expect(contentHasFile([{ type: 'text', text: 'x' }])).toBe(false) + expect(contentHasFile([fileBlock('a.txt')])).toBe(true) + expect(contentHasFile([{ + type: 'tool-result', + toolCallId: ToolCallId('call-1'), + content: [{ + type: 'tool-result', + toolCallId: ToolCallId('call-2'), + content: [fileBlock('deep.txt')], + }], + }])).toBe(true) + }) + + it('renders the handle with the read path or the explicit no-path fallback', () => { + const withPath = fileHandleText(fileBlock('notes.pdf').attachment, '/home/.dsh/attachments/v1/files/ab/x/notes.pdf') + expect(withPath).toContain('"notes.pdf"') + expect(withPath).toContain('42 bytes') + expect(withPath).toContain('sha256:abababab') + expect(withPath).toContain('"/home/.dsh/attachments/v1/files/ab/x/notes.pdf"') + expect(withPath).toContain('include this saved path in the delegation prompt') + expect(withPath).toContain('only subagents sharing this execution environment can read it') + const withoutPath = fileHandleText(fileBlock('notes.pdf').attachment, undefined) + expect(withoutPath).toContain('current execution environment cannot access a readable path') + expect(withoutPath).toContain('do not claim to have read it') + }) + + it('replaces every file occurrence with handle text and keeps file-free history identical', () => { + const plain = [createUserMessage({ content: [{ type: 'text', text: 'hi' }], source })] + expect(projectFilesToText(plain, () => '/p')).toBe(plain) + const unchangedTool = { + type: 'tool-result' as const, + toolCallId: ToolCallId('call-plain'), + content: [{ type: 'text' as const, text: 'unchanged result' }], + } + const messages = [plain[0]!, createUserMessage({ + content: [ + fileBlock('top.csv'), + { type: 'text', text: 'keep' }, + unchangedTool, + { + type: 'tool-result', + toolCallId: ToolCallId('call-3'), + content: [fileBlock('nested.csv')], + }, + ], + source, + })] + const projected = projectFilesToText(messages, ref => `/copies/${ref.name}`) + expect(projected).not.toBe(messages) + expect(projected[0]).toBe(messages[0]) + const content = projected[1]!.content + expect(content[0]).toEqual({ type: 'text', text: fileHandleText(fileBlock('top.csv').attachment, '/copies/top.csv') }) + expect(content[1]).toEqual({ type: 'text', text: 'keep' }) + expect(content[2]).toBe(messages[1]!.content[2]) + const nested = content[3] as Extract + expect(nested.content[0]).toEqual({ + type: 'text', + text: fileHandleText(fileBlock('nested.csv').attachment, '/copies/nested.csv'), + }) + // The durable message is untouched: projection returns shallow copies. + expect(messages[1]!.content[0]!.type).toBe('file') + }) +}) diff --git a/packages/llm/llm/tests/service.spec.ts b/packages/llm/llm/tests/service.spec.ts index 47309daea1..2bfcd0f486 100644 --- a/packages/llm/llm/tests/service.spec.ts +++ b/packages/llm/llm/tests/service.spec.ts @@ -205,6 +205,73 @@ describe('LlmRuntime', () => { expect(adapter.lastOptions?.messages[0]).toBe(message) }) + it('projects file blocks through every host-path availability outcome', async () => { + const attachment = { + attachmentId: AttachmentId(`sha256:${'ab'.repeat(32)}`), + name: 'notes.txt', + bytes: 3, + } + const cases = [ + { + name: 'native tools under read-only permission receive the mapped read path', + attachments: { fileHostPath: () => '/host/notes.txt' }, + fs: { processPathFromHostPath: () => '/sandbox/notes.txt' }, + expected: '"/sandbox/notes.txt"', + }, + { + name: 'Code Mode under workspace-write permission receives the same mapped read path', + attachments: { fileHostPath: () => '/host/notes.txt' }, + fs: { processPathFromHostPath: () => '/code-sandbox/notes.txt' }, + expected: '"/code-sandbox/notes.txt"', + }, + { + name: 'missing attachment service', + expected: 'current execution environment cannot access a readable path', + }, + { + name: 'provider without a host path', + attachments: { fileHostPath: () => undefined }, + expected: 'current execution environment cannot access a readable path', + }, + { + name: 'invalid durable reference', + attachments: { fileHostPath: () => { throw new Error('invalid ref') } }, + expected: 'current execution environment cannot access a readable path', + }, + { + name: 'missing filesystem mapping', + attachments: { fileHostPath: () => '/host/notes.txt' }, + expected: 'current execution environment cannot access a readable path', + }, + ] + + for (const fixture of cases) { + const ctx = new Context() + if (fixture.attachments !== undefined) ctx.provide('attachments', fixture.attachments as never) + if (fixture.fs !== undefined) ctx.provide('fs', fixture.fs as never) + await ctx.plugin(LlmRuntime) + const adapter = new RecordingAdapter(SCRIPT) + ctx.llm.registerAdapter(['test-provider'], adapter) + + await collect(ctx.llm.stream({ + provider: 'test-provider', + model: 'test-model', + messages: [createUserMessage({ + content: [{ type: 'file', attachment }], + source: { kind: 'user' }, + })], + })) + + const projected = adapter.lastOptions?.messages[0]?.content[0] + expect(projected, fixture.name).toMatchObject({ type: 'text' }) + if (projected?.type !== 'text') throw new Error(`expected projected text for ${fixture.name}`) + expect(projected.text, fixture.name).toContain(fixture.expected) + if (fixture.fs !== undefined) { + expect(projected.text, fixture.name).toContain('include this saved path in the delegation prompt') + } + } + }) + it('captures provider-owned retry policy at registration and defaults omission', async () => { const configured = resolveRetryPolicy({ mode: 'always' }, 'test retryPolicy') const adapter = new class extends ScriptedAdapter { diff --git a/packages/llm/token-meter/src/index.ts b/packages/llm/token-meter/src/index.ts index d97d9d3fc2..c3c3029219 100644 --- a/packages/llm/token-meter/src/index.ts +++ b/packages/llm/token-meter/src/index.ts @@ -7,7 +7,7 @@ import { Context, Service } from '@deepseek-ai/cordis' import z from '@deepseek-ai/schemastery' import { BlockAssembler } from '@deepseek-ai/dsh-llm' -import type { LlmImageRequestPricing, Message, TokenUsage } from '@deepseek-ai/dsh-llm' +import type { LlmImageRequestPricing, LlmRuntime, Message, TokenUsage } from '@deepseek-ai/dsh-llm' import { deepFreeze } from '@deepseek-ai/dsh-util-values' import type { EpochHeader, Session, SessionEvent } from '@deepseek-ai/dsh-session' import { canonicalHeader, headerEquals, isSurfaceEvent } from '@deepseek-ai/dsh-session' @@ -136,7 +136,8 @@ export class TokenMeter extends Service { ? state.header : canonicalHeader(requestHeader) const pricing = this._routeImagePricing(header) - const surface = priceSurface(state.surface, pricing) + const fileText = this._fileRequestText() + const surface = priceSurface(state.surface, pricing, fileText) const anchor = state.anchor let baseline: TokenMeasurementBaseline @@ -145,7 +146,7 @@ export class TokenMeter extends Service { // Matching headers share one route, so the anchored snapshot reprices // under the same pricing as the current surface and the signed delta // compares like with like. - const anchorSurfaceTokens = priceSurface(anchor.nodes, pricing).surfaceTokens + const anchorSurfaceTokens = priceSurface(anchor.nodes, pricing, fileText).surfaceTokens + anchor.assistantTokens const estimatedAnchorTokens = estimateHeader(header) + anchorSurfaceTokens const usage = anchor.usage @@ -183,6 +184,14 @@ export class TokenMeter extends Service { return this.ctx.get('llm')?.imageRequestPricing(config.provider, config.model) } + /** Resolve request-time file projection when an LLM service is mounted. */ + private _fileRequestText(): ( + (ref: Parameters[0]) => string + ) | undefined { + const llm = this.ctx.get('llm') + return llm === undefined ? undefined : ref => llm.fileRequestText(ref) + } + /** * Heuristically price one model-visible message (instance face of the pure * `estimateMessage` export from `estimate.ts`). diff --git a/packages/llm/token-meter/src/route-pricing.ts b/packages/llm/token-meter/src/route-pricing.ts index 402db18ac0..03a1c519c3 100644 --- a/packages/llm/token-meter/src/route-pricing.ts +++ b/packages/llm/token-meter/src/route-pricing.ts @@ -1,18 +1,17 @@ /** - * Route-aware surface pricing: projects the fold's fixed-heuristic nodes onto - * the routed model's request, replacing every image occurrence's structural - * price with the route's declared visual tokens plus the model-visible text it - * actually sends. Without declared pricing every node keeps its fixed - * heuristic price, so provider-neutral behavior is unchanged. + * Request-projected surface pricing: replaces attachment-block heuristics with + * the image and file representations sent to the routed model. * * @module @deepseek-ai/dsh-token-meter/route-pricing */ -import type { LlmImageRequestPricing } from '@deepseek-ai/dsh-llm' +import type { ContentBlock, LlmImageRequestPricing } from '@deepseek-ai/dsh-llm' import { estimateContent } from './estimate.ts' import type { MeterSurfaceNode } from './surface-fold.ts' import type { TokenSurfaceNode } from './types.ts' +type FileAttachmentRef = Extract['attachment'] + /** One surface priced for a request route: public nodes plus their total. */ export interface PricedSurface { /** Positional nodes carrying both the route price and the fixed-heuristic price. */ @@ -22,9 +21,10 @@ export interface PricedSurface { } /** - * Price one ordered surface under a route's request-image pricing. + * Price one ordered surface under its model-request attachment projection. * @param nodes - the fold's current or snapshotted surface, in model-visible order. * @param pricing - the routed model's image pricing, or undefined to keep the fixed heuristic. + * @param fileText - exact file handle projection used by the mounted LLM service. * @returns detached public nodes and their route-priced total. * @throws when the pricing answers a different occurrence count than it was * asked — misalignment would silently misprice nodes, so it must fail loud. @@ -32,9 +32,11 @@ export interface PricedSurface { export function priceSurface( nodes: readonly MeterSurfaceNode[], pricing: LlmImageRequestPricing | undefined, + fileText?: (ref: FileAttachmentRef) => string, ): PricedSurface { const images = pricing === undefined ? [] : nodes.flatMap(node => node.images) - if (pricing === undefined || images.length === 0) { + const hasFiles = fileText !== undefined && nodes.some(node => node.files.length > 0) + if ((pricing === undefined || images.length === 0) && !hasFiles) { let surfaceTokens = 0 const publicNodes = nodes.map((node) => { surfaceTokens += node.heuristicTokens @@ -42,8 +44,8 @@ export function priceSurface( }) return { nodes: publicNodes, surfaceTokens } } - const prices = pricing.priceImages(images) - if (prices.length !== images.length) { + const prices = pricing === undefined ? [] : pricing.priceImages(images) + if (pricing !== undefined && prices.length !== images.length) { throw new Error( `token meter: route image pricing answered ${prices.length} prices for ${images.length} occurrences`, ) @@ -52,8 +54,14 @@ export function priceSurface( let surfaceTokens = 0 const publicNodes = nodes.map((node) => { let tokens = node.heuristicTokens - if (node.images.length > 0) { - tokens = node.imageFreeTokens + if (fileText !== undefined && node.files.length > 0) { + tokens -= node.fileStructuralTokens + for (const file of node.files) { + tokens += estimateContent([{ type: 'text', text: fileText(file) }]) + } + } + if (pricing !== undefined && node.images.length > 0) { + tokens -= node.imageStructuralTokens for (let occurrence = 0; occurrence < node.images.length; occurrence += 1) { // oxlint-disable-next-line typescript/no-non-null-assertion -- length equality is asserted above const price = prices[cursor]! diff --git a/packages/llm/token-meter/src/surface-fold.ts b/packages/llm/token-meter/src/surface-fold.ts index 66824a2974..f947adfc35 100644 --- a/packages/llm/token-meter/src/surface-fold.ts +++ b/packages/llm/token-meter/src/surface-fold.ts @@ -10,8 +10,8 @@ * fallible step read-only and {@link commitSurfaceTokens} mutates in place, * so a throw leaves the caller's state untouched and the same malformed * event fails identically on every retry. - * Nodes also carry their durable image occurrences and image-free heuristic - * price, so `measure()` can reprice image content for the routed model. + * Nodes also carry durable attachment occurrences and their structural prices, + * so `measure()` can price the request representation sent to the model. * * @module @deepseek-ai/dsh-token-meter/surface-fold */ @@ -22,16 +22,22 @@ import type { ContentBlock, Message } from '@deepseek-ai/dsh-llm' import type { ImageAttachmentRef } from '@deepseek-ai/dsh-attachment' import { estimateMessage, estimateStructuralBlock } from './estimate.ts' +type FileAttachmentRef = Extract['attachment'] + /** One priced surface node with the image occurrences route pricing replaces. */ export interface MeterSurfaceNode { /** Durable sequence number of the surface event. */ readonly seq: number /** Fixed-heuristic price of the node's exact message. */ readonly heuristicTokens: number - /** Fixed-heuristic price with every image occurrence's structural price removed. */ - readonly imageFreeTokens: number + /** Structural JSON price replaced when the routed request projects images. */ + readonly imageStructuralTokens: number + /** Structural JSON price replaced when request assembly projects files to text. */ + readonly fileStructuralTokens: number /** Durable image occurrences in message order; empty for image-free nodes. */ readonly images: readonly ImageAttachmentRef[] + /** Durable file occurrences in message order; empty for file-free nodes. */ + readonly files: readonly FileAttachmentRef[] } /** One validated surface transition that has not mutated the priced surface yet. */ @@ -46,31 +52,53 @@ export interface SurfaceTokenPlan { readonly target: 'append' | { readonly startIdx: number; readonly endIdx: number } } -/** Collect image occurrences recursively and total their structural prices. */ -function collectImages(blocks: readonly ContentBlock[], images: ImageAttachmentRef[]): number { - let structuralTokens = 0 +/** Collect projected attachment occurrences and their structural prices. */ +function collectProjectedAttachments( + blocks: readonly ContentBlock[], + images: ImageAttachmentRef[], + files: FileAttachmentRef[], +): { readonly imageTokens: number; readonly fileTokens: number } { + let imageTokens = 0 + let fileTokens = 0 for (const block of blocks) { if (block.type === 'image') { images.push(block.attachment) - structuralTokens += estimateStructuralBlock(block) + imageTokens += estimateStructuralBlock(block) + } else if (block.type === 'file') { + files.push(block.attachment) + fileTokens += estimateStructuralBlock(block) } else if (block.type === 'tool-result') { - structuralTokens += collectImages(block.content, images) + const nested = collectProjectedAttachments(block.content, images, files) + imageTokens += nested.imageTokens + fileTokens += nested.fileTokens } } - return structuralTokens + return { imageTokens, fileTokens } } /** Build one priced node from a surface event's derived message. */ function analyzeNode(seq: number, message: Message | null): MeterSurfaceNode { - if (message === null) return { seq, heuristicTokens: 0, imageFreeTokens: 0, images: [] } + if (message === null) { + return { + seq, + heuristicTokens: 0, + imageStructuralTokens: 0, + fileStructuralTokens: 0, + images: [], + files: [], + } + } const heuristicTokens = estimateMessage(message) const images: ImageAttachmentRef[] = [] - const imageStructuralTokens = collectImages(message.content, images) + const files: FileAttachmentRef[] = [] + const structural = collectProjectedAttachments(message.content, images, files) return { seq, heuristicTokens, - imageFreeTokens: heuristicTokens - imageStructuralTokens, + imageStructuralTokens: structural.imageTokens, + fileStructuralTokens: structural.fileTokens, images, + files, } } diff --git a/packages/llm/token-meter/tests/route-pricing.spec.ts b/packages/llm/token-meter/tests/route-pricing.spec.ts index 37ef87ce04..65fb15a2f7 100644 --- a/packages/llm/token-meter/tests/route-pricing.spec.ts +++ b/packages/llm/token-meter/tests/route-pricing.spec.ts @@ -1,9 +1,11 @@ import { describe, expect, it } from 'vitest' import { Context } from '@deepseek-ai/cordis' -import { LlmRuntime, LlmAdapter, createMessage, createUserMessage } from '@deepseek-ai/dsh-llm' +import { + LlmRuntime, LlmAdapter, createMessage, createUserMessage, projectFilesToText, +} from '@deepseek-ai/dsh-llm' import type { GenerateOptions, LlmImageRequestPricing, Message, StreamChunk, TokenUsage, UserMessage } from '@deepseek-ai/dsh-llm' import { AttachmentId } from '@deepseek-ai/dsh-attachment' -import type { ImageAttachmentRef } from '@deepseek-ai/dsh-attachment' +import type { FileAttachmentRef, ImageAttachmentRef } from '@deepseek-ai/dsh-attachment' import { Session, SessionId, canonicalHeader } from '@deepseek-ai/dsh-session' import type { EpochHeader } from '@deepseek-ai/dsh-session' import SessionProjectionRegistry from '@deepseek-ai/dsh-session-projection' @@ -53,6 +55,14 @@ function imageMessage(name: string, text = 'look at this'): UserMessage { }) } +function fileRef(name: string): FileAttachmentRef { + return { + attachmentId: AttachmentId(`sha256:${'ab'.repeat(32)}`), + name, + bytes: 2_447_000_000, + } +} + function header(model: string): EpochHeader { return canonicalHeader({ config: { provider: 'mock', model } }) } @@ -65,6 +75,12 @@ interface Harness { async function harness(pricing: (model: string) => LlmImageRequestPricing | undefined): Promise { const ctx = new Context() new SessionProjectionRegistry(ctx) + ctx.provide('attachments', { + fileHostPath: (ref: FileAttachmentRef) => `/host/${ref.name}`, + } as never) + ctx.provide('fs', { + processPathFromHostPath: (path: string) => path.replace('/host/', '/sandbox/'), + } as never) const llm = new LlmRuntime(ctx) llm.registerAdapter(['mock'], new PricingAdapter(pricing)) const meter = new TokenMeter(ctx) @@ -96,7 +112,23 @@ function appendSuccessfulCall(session: Session, value: EpochHeader, usage?: Toke session.append('step/end', { turn: 1, step: 1 }) } -describe('route-aware image pricing', () => { +describe('request projection pricing', () => { + it('prices file blocks as the exact handle text dispatched to the provider', async () => { + const { meter, session } = await harness(() => undefined) + const ref = fileRef('archive.zip') + const message = createUserMessage({ + content: [{ type: 'file', attachment: ref }], + source: { kind: 'user' }, + }) + session.append('user/message', message, { surfaceOp: 'append' }) + + const measurement = meter.measure(session) + const projected = projectFilesToText([message], file => `/sandbox/${file.name}`)[0] + if (projected === undefined) throw new Error('missing projected file message') + expect(measurement.nodes[0]?.tokens).toBe(estimateMessage(projected)) + expect(measurement.nodes[0]?.tokens).toBeGreaterThan(estimateMessage(message)) + }) + it('prices a first multimodal request estimate with the routed visual tokens', async () => { const { meter, session } = await harness(() => fixedPricing) const message = imageMessage('photo') diff --git a/packages/session-query/session-log-export/src/archive.ts b/packages/session-query/session-log-export/src/archive.ts index 2f7d1fbd11..d89e036193 100644 --- a/packages/session-query/session-log-export/src/archive.ts +++ b/packages/session-query/session-log-export/src/archive.ts @@ -1,12 +1,12 @@ /** * Host-side session-log download: streams one ZIP archive whose files are the - * sessions' stored artifact text verbatim plus every referenced media object. + * sessions' stored artifact text verbatim plus every referenced attachment. * The root artifact sits under its original base name (`session.jsonl`); each * subagent descendant under `subagents//`; each image referenced * by any included log under `media/.` (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. Before + * so one archive never duplicates a shared image); each file under + * `files///`. No manifest is written. Every entry is + * byte-identical to the backend's durable artifact or attachment store. 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 @@ -21,7 +21,9 @@ import { Zip, ZipDeflate } from 'fflate' import type { Context } from '@deepseek-ai/cordis' -import type { AttachmentStore, ImageAttachmentRef } from '@deepseek-ai/dsh-attachment' +import type { + AttachmentStore, FileAttachmentRef, ImageAttachmentRef, +} from '@deepseek-ai/dsh-attachment' import type { SessionLineageNode, SessionQueryEngine } from '@deepseek-ai/dsh-session-query' import type { SessionId, SessionStore } from '@deepseek-ai/dsh-session' import type { SessionPersistence, SessionRawArtifact } from '@deepseek-ai/dsh-session-persistence' @@ -84,10 +86,11 @@ export async function flushLiveSessionLog( signal?.throwIfAborted() } -/** One exported file: a stored artifact text or one referenced media object. */ +/** One exported artifact or referenced attachment object. */ export type SessionLogZipEntry = | { readonly path: string; readonly content: string } | { readonly path: string; readonly data: Uint8Array } + | { readonly path: string; readonly chunks: AsyncIterable } /** Zip extension for each accepted raster media type. */ const MEDIA_TYPE_EXTENSIONS: Record = { @@ -108,13 +111,26 @@ function mediaEntryPath(ref: ImageAttachmentRef): string { return `media/${String(ref.attachmentId)}.${MEDIA_TYPE_EXTENSIONS[ref.mediaType]}` } +/** Archive path that preserves one stored file reference's digest and name. */ +function fileEntryPath(ref: FileAttachmentRef): string { + const digest = String(ref.attachmentId).replace(/^sha256:/u, '') + const name = ref.name.replace(/[\\/\u0000-\u001f\u007f]/gu, '_') + const safeName = name === '.' || name === '..' || name === '' ? 'file' : name + return `files/${digest.slice(0, 2)}/${digest}/${safeName}` +} + /** - * Collect every image reference inside one content array, descending into + * Collect every attachment 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). + * @param images - image dedupe map keyed by attachment id. + * @param files - file dedupe map keyed by attachment id and stored name. */ -function collectImageRefs(content: unknown, refs: Map): void { +function collectAttachmentRefs( + content: unknown, + images: Map, + files: Map, +): void { if (!Array.isArray(content)) return const pending: unknown[] = [] for (const item of content) pending.push(item) @@ -124,7 +140,11 @@ function collectImageRefs(content: unknown, refs: Map): void { +function collectEventAttachmentRefs( + event: unknown, + images: Map, + files: Map, +): void { const data = (event as { data?: unknown }).data if (typeof data !== 'object' || data === null) return const carrier = data as { @@ -148,23 +173,27 @@ function collectEventImageRefs(event: unknown, refs: Map chunk?: { type?: unknown; block?: unknown } } - collectImageRefs(carrier.content, refs) - if (carrier.message !== undefined) collectImageRefs(carrier.message.content, refs) + collectAttachmentRefs(carrier.content, images, files) + if (carrier.message !== undefined) collectAttachmentRefs(carrier.message.content, images, files) if (carrier.inserted !== undefined) { - for (const message of carrier.inserted) collectImageRefs(message.content, refs) + for (const message of carrier.inserted) collectAttachmentRefs(message.content, images, files) } - if (carrier.chunk?.type === 'block-end') collectImageRefs([carrier.chunk.block], refs) + if (carrier.chunk?.type === 'block-end') collectAttachmentRefs([carrier.chunk.block], images, files) } /** - * Collect the distinct media references one stored artifact text names. - * Lines that fail to parse cannot reference media and are skipped (the + * Collect the distinct attachment references one stored artifact text names. + * Lines that fail to parse cannot reference attachments 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. + * @returns image and file dedupe maps. */ -function imageRefsInArtifact(content: string): Map { - const refs = new Map() +function attachmentRefsInArtifact(content: string): { + readonly images: Map + readonly files: Map +} { + const images = new Map() + const files = new Map() for (const line of content.split('\n')) { if (line === '') continue let event: unknown @@ -173,9 +202,9 @@ function imageRefsInArtifact(content: string): Map { } catch { continue } - collectEventImageRefs(event, refs) + collectEventAttachmentRefs(event, images, files) } - return refs + return { images, files } } /** @@ -204,10 +233,10 @@ 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 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. + * after the consumer moves on), then every distinct attachment referenced by + * the included logs. Images are read and verified as bounded stored objects; + * generic files remain streamed through the ZIP writer. The host holds at most + * one descendant artifact, one image, and one file chunk 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). @@ -224,10 +253,13 @@ export async function* sessionLogZipEntries( signal?: AbortSignal, ): AsyncGenerator { const media = new Map() - const rememberMedia = (content: string): void => { - for (const [id, ref] of imageRefsInArtifact(content)) media.set(id, ref) + const files = new Map() + const rememberAttachments = (content: string): void => { + const refs = attachmentRefsInArtifact(content) + for (const [id, ref] of refs.images) media.set(id, ref) + for (const [id, ref] of refs.files) files.set(id, ref) } - rememberMedia(root.content) + rememberAttachments(root.content) yield { path: root.filename, content: root.content } if (includeDescendants) { const seen = new Set([sessionId]) @@ -245,7 +277,7 @@ export async function* sessionLogZipEntries( if (raw === undefined) { throw new Error(`subagent "${id}" has no stored log artifact`) } - rememberMedia(raw.content) + rememberAttachments(raw.content) yield { path: `subagents/${safeSessionIdSegment(id)}/${raw.filename}`, content: raw.content, @@ -263,6 +295,13 @@ export async function* sessionLogZipEntries( signal?.throwIfAborted() yield { path: mediaEntryPath(ref), data: stored.data } } + for (const ref of files.values()) { + signal?.throwIfAborted() + yield { + path: fileEntryPath(ref), + chunks: deps.attachments.readFileStream(ref, signal), + } + } } /** How many code units of artifact text one zip push carries (bounded encode memory). */ @@ -334,6 +373,25 @@ async function pushBinaryChunks( } while (offset < data.byteLength) } +/** Push one streamed file entry without retaining its complete byte sequence. */ +async function pushStreamChunks( + deflate: ZipDeflate, + chunks: AsyncIterable, + controller: ReadableStreamDefaultController, + capacity: ResponseCapacityGate, + signal: AbortSignal, +): Promise { + for await (const chunk of chunks) { + signal.throwIfAborted() + if (chunk.byteLength === 0) continue + deflate.push(chunk, false) + await capacity.wait(controller, signal) + } + signal.throwIfAborted() + deflate.push(new Uint8Array(), true) + await capacity.wait(controller, signal) +} + /** * 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 @@ -427,8 +485,10 @@ export function streamSessionLogZip( archive.add(deflate) if ('content' in entry) { await pushArtifactChunks(deflate, entry.content, controller, capacity, producerSignal) - } else { + } else if ('data' in entry) { await pushBinaryChunks(deflate, entry.data, controller, capacity, producerSignal) + } else { + await pushStreamChunks(deflate, entry.chunks, controller, capacity, producerSignal) } } archive.end() diff --git a/packages/session-query/session-log-export/src/index.ts b/packages/session-query/session-log-export/src/index.ts index 88c4cc6ccb..91a97faff9 100644 --- a/packages/session-query/session-log-export/src/index.ts +++ b/packages/session-query/session-log-export/src/index.ts @@ -55,6 +55,7 @@ interface SessionLogConnection { register(route: { readonly path: string readonly methods: readonly ('GET' | 'HEAD')[] + readonly requestBody: 'buffered' readonly fetch: (request: Request) => Promise }): () => Promise } @@ -81,6 +82,7 @@ export function apply(ctx: Context, config: Config = {}): void { connectionOf(ctx).fetch.register({ path: SESSION_LOG_EXPORT_PATH, methods: ['GET', 'HEAD'], + requestBody: 'buffered', fetch: async (request) => { const response = await sessionLogExportResponse( ctx, diff --git a/packages/session-query/session-log-export/tests/archive.host.spec.ts b/packages/session-query/session-log-export/tests/archive.host.spec.ts index c41846039a..af1ea9e428 100644 --- a/packages/session-query/session-log-export/tests/archive.host.spec.ts +++ b/packages/session-query/session-log-export/tests/archive.host.spec.ts @@ -9,7 +9,7 @@ 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' +import type { FileAttachmentRef, ImageAttachmentRef } from '@deepseek-ai/dsh-attachment' 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' @@ -55,6 +55,11 @@ function imageEventLine(id: string, mediaType: ImageAttachmentRef['mediaType'] = return `{"type":"user/message","seq":1,"time":1000,"data":{"content":[{"type":"image","attachment":{"attachmentId":"${id}","mediaType":"${mediaType}","bytes":4,"width":2,"height":2}}]}}` } +/** A user/message event line carrying one generic-file reference. */ +function fileEventLine(id: string, name = 'notes.txt', bytes = 5): string { + return `{"type":"user/message","seq":1,"time":1000,"data":{"content":[{"type":"file","attachment":{"attachmentId":"${id}","name":"${name}","bytes":${bytes}}}]}}` +} + async function buildApi( artifacts: Record, descendants: SessionLineageNode[] = [], @@ -62,6 +67,7 @@ async function buildApi( query?: boolean persistence?: boolean | 'throw' | 'unsupported' attachments?: boolean | ((ref: ImageAttachmentRef, signal?: AbortSignal) => Promise>) + readFileStream?: (ref: FileAttachmentRef, signal?: AbortSignal) => AsyncIterable sessions?: { get(id: SessionId): { readonly id: SessionId } | undefined flush(session: { readonly id: SessionId }): Promise @@ -110,6 +116,9 @@ async function buildApi( validateImage: async () => {}, saveImage: async () => { throw new Error('export never saves images') }, readImage, + readFileStream: services.readFileStream ?? (async function* () { + throw new Error('fixture has no files') + }), } as never) } if (services.sessions !== undefined) ctx.provide('sessions', services.sessions as never) @@ -630,6 +639,52 @@ describe('session.export download endpoint', () => { expect(files['media/img-1.png']).toEqual(storedImage('img-1').data) }) + it('streams generic files under their content-addressed archive paths', async () => { + const digest = 'a'.repeat(64) + const id = `sha256:${digest}` + const fallbackDigest = 'c'.repeat(64) + const fallbackId = `sha256:${fallbackDigest}` + const root = artifact('session-root', undefined, [ + '{"type":"session","version":0,"id":"session-root","createdAt":1000}', + fileEventLine(id, 'notes.txt', 5), + fileEventLine(fallbackId, '.', 5), + ].join('\n') + '\n') + const reads: Array<{ ref: FileAttachmentRef; signal: AbortSignal | undefined }> = [] + const api = await buildApi({ 'session-root': root }, [], { + readFileStream: (ref, signal) => (async function* (): AsyncIterable { + reads.push({ ref, signal }) + yield new Uint8Array() + yield Uint8Array.of(1, 2) + yield Uint8Array.of(3, 4, 5) + })(), + }) + const response = await toFetchHandler(api).fetch( + new Request('http://host/api/session.export?sessionId=session-root'), + ) + const files = unzipSync(await responseBytes(response)) + expect(files[`files/aa/${digest}/notes.txt`]).toEqual(Uint8Array.of(1, 2, 3, 4, 5)) + expect(files[`files/cc/${fallbackDigest}/file`]).toEqual(Uint8Array.of(1, 2, 3, 4, 5)) + expect(reads).toHaveLength(2) + expect(reads[0]?.ref).toMatchObject({ attachmentId: id, name: 'notes.txt', bytes: 5 }) + expect(reads[0]?.signal).toBeInstanceOf(AbortSignal) + }) + + it('fails the whole export when a referenced file stream fails', async () => { + const digest = 'b'.repeat(64) + const id = `sha256:${digest}` + const root = artifact('session-root', undefined, fileEventLine(id)) + const api = await buildApi({ 'session-root': root }, [], { + readFileStream: () => (async function* (): AsyncIterable { + yield Uint8Array.of(1) + throw new Error('file bytes missing') + })(), + }) + const response = await toFetchHandler(api).fetch( + new Request('http://host/api/session.export?sessionId=session-root'), + ) + await expect(response.arrayBuffer()).rejects.toThrow('file bytes missing') + }) + it('collects media referenced from nested tool results', async () => { const nested = '{"type":"assistant/message","seq":2,"time":2000,"data":{"content":[{"type":"tool-result","content":[{"type":"image","attachment":{"attachmentId":"nested-1","mediaType":"image/webp","bytes":4,"width":2,"height":2}}]}]}}' const root = artifact('session-root', undefined, [