feat(attachment): store and project generic files

This commit is contained in:
creatixchu
2026-09-01 16:11:43 +08:00
parent dead2b2324
commit 65d2015b09
25 changed files with 1469 additions and 128 deletions
@@ -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))
},
)
}
@@ -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<FileAttachmentRef> {
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<FileAttachmentRef> {
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<Uint8Array> {
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')
}
}
@@ -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<T> {
}, (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<FileAttachmentRef> {
return saveFileVerbatim(this.root, input)
}
override async saveFileStream(input: SaveFileStreamAttachment): Promise<FileAttachmentRef> {
return saveFileStreamVerbatim(this.root, input)
}
override readFileStream(ref: FileAttachmentRef, signal?: AbortSignal): AsyncIterable<Uint8Array> {
return readFileStreamVerbatim(this.root, ref, signal)
}
override fileHostPath(ref: FileAttachmentRef): string {
return storedFilePath(this.root, ref)
}
override async readImageRequest(
ref: ImageAttachmentRef,
policy: ImageRequestPolicy,
+186 -36
View File
@@ -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<void> {
const staged = await stageImmutableObject(root, (function* (): Iterable<Uint8Array> {
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<Uint8Array>,
targetFor: (sha256: string, bytes: number) => string,
signal?: AbortSignal,
): Promise<StreamedImmutableObject> {
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<void> {
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<Uint8Array> | Iterable<Uint8Array>,
signal?: AbortSignal,
): Promise<StagedImmutableObject> {
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<void> {
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<string> {
const hash = createHash('sha256')
for await (const chunk of createReadStream(path) as AsyncIterable<Buffer>) hash.update(chunk)
return hash.digest('hex')
}
async function removeTemporary(path: string): Promise<void> {
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
},
)
}
/**
@@ -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<string> {
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<Uint8Array>): Promise<Uint8Array> {
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('a<b>c: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<Uint8Array> {
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<Uint8Array> {
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<Uint8Array> {
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)
})
})
@@ -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<Uint8Array> { 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 })
}
@@ -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()
@@ -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<FileAttachmentRef> {
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.
@@ -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<string> = new Set(IMAGE_ADMISSION_ERROR_CODES)
+70 -1
View File
@@ -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<FileAttachmentRef> {
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<FileAttachmentRef> {
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<Uint8Array> {
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.
@@ -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<Uint8Array>
/** 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
@@ -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') } }
@@ -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<Uint8Array> { 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<void> => {
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)
})
})
+75 -1
View File
@@ -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
+48 -7
View File
@@ -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) {
+15 -1
View File
@@ -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
}
+80
View File
@@ -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<ContentBlock, { type: 'file' }> {
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<ContentBlock, { type: 'tool-result' }>
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')
})
})
+67
View File
@@ -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 {
+12 -3
View File
@@ -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<LlmRuntime['fileRequestText']>[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`).
+20 -12
View File
@@ -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<ContentBlock, { type: 'file' }>['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]!
+41 -13
View File
@@ -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<ContentBlock, { type: 'file' }>['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,
}
}
@@ -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<Harness> {
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')
@@ -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/<id>/<filename>`; each image referenced
* by any included log under `media/<attachmentId>.<ext>` (content-addressed,
* so one archive never duplicates a shared image). No manifest is written —
* every file is byte-identical to the backend's durable artifact or attachment
* store and self-describing through its own header line or media type. Before
* so one archive never duplicates a shared image); each file under
* `files/<prefix>/<digest>/<name>`. 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<Uint8Array> }
/** Zip extension for each accepted raster media type. */
const MEDIA_TYPE_EXTENSIONS: Record<ImageAttachmentRef['mediaType'], string> = {
@@ -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<string, ImageAttachmentRef>): void {
function collectAttachmentRefs(
content: unknown,
images: Map<string, ImageAttachmentRef>,
files: Map<string, FileAttachmentRef>,
): 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<string, ImageAttachmentRef
const block = value as { type?: unknown; attachment?: unknown; content?: unknown }
if (block.type === 'image' && typeof block.attachment === 'object' && block.attachment !== null) {
const ref = block.attachment as ImageAttachmentRef
refs.set(String(ref.attachmentId), ref)
images.set(String(ref.attachmentId), ref)
}
if (block.type === 'file' && typeof block.attachment === 'object' && block.attachment !== null) {
const ref = block.attachment as FileAttachmentRef
files.set(`${String(ref.attachmentId)}\u0000${ref.name}`, ref)
}
if (Array.isArray(block.content)) {
for (const item of block.content) pending.push(item)
@@ -133,13 +153,18 @@ function collectImageRefs(content: unknown, refs: Map<string, ImageAttachmentRef
}
/**
* Collect every image reference one session event carries, across the same
* Collect every attachment reference one session event carries, across the same
* carriers the live attachment route scans (direct content, message content,
* inserted messages, and completed assistant chunk blocks).
* @param event - one parsed JSONL event object.
* @param refs - the dedupe map being filled (keyed by attachment id).
* @param images - image dedupe map keyed by attachment id.
* @param files - file dedupe map keyed by attachment id and stored name.
*/
function collectEventImageRefs(event: unknown, refs: Map<string, ImageAttachmentRef>): void {
function collectEventAttachmentRefs(
event: unknown,
images: Map<string, ImageAttachmentRef>,
files: Map<string, FileAttachmentRef>,
): 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<string, ImageAttachment
inserted?: Array<{ content?: unknown }>
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<string, ImageAttachmentRef> {
const refs = new Map<string, ImageAttachmentRef>()
function attachmentRefsInArtifact(content: string): {
readonly images: Map<string, ImageAttachmentRef>
readonly files: Map<string, FileAttachmentRef>
} {
const images = new Map<string, ImageAttachmentRef>()
const files = new Map<string, FileAttachmentRef>()
for (const line of content.split('\n')) {
if (line === '') continue
let event: unknown
@@ -173,9 +202,9 @@ function imageRefsInArtifact(content: string): Map<string, ImageAttachmentRef> {
} 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<SessionLogZipEntry> {
const media = new Map<string, ImageAttachmentRef>()
const rememberMedia = (content: string): void => {
for (const [id, ref] of imageRefsInArtifact(content)) media.set(id, ref)
const files = new Map<string, FileAttachmentRef>()
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>([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<Uint8Array>,
controller: ReadableStreamDefaultController<Uint8Array>,
capacity: ResponseCapacityGate,
signal: AbortSignal,
): Promise<void> {
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()
@@ -55,6 +55,7 @@ interface SessionLogConnection {
register(route: {
readonly path: string
readonly methods: readonly ('GET' | 'HEAD')[]
readonly requestBody: 'buffered'
readonly fetch: (request: Request) => Promise<Response>
}): () => Promise<void>
}
@@ -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,
@@ -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<string, SessionRawArtifact>,
descendants: SessionLineageNode[] = [],
@@ -62,6 +67,7 @@ async function buildApi(
query?: boolean
persistence?: boolean | 'throw' | 'unsupported'
attachments?: boolean | ((ref: ImageAttachmentRef, signal?: AbortSignal) => Promise<ReturnType<typeof storedImage>>)
readFileStream?: (ref: FileAttachmentRef, signal?: AbortSignal) => AsyncIterable<Uint8Array>
sessions?: {
get(id: SessionId): { readonly id: SessionId } | undefined
flush(session: { readonly id: SessionId }): Promise<boolean>
@@ -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<Uint8Array> {
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<Uint8Array> {
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, [