fix(gui): close attachment durability gaps

This commit is contained in:
Yichen Jiang
2026-07-27 14:29:19 +08:00
parent b599d1d0af
commit a1835c3228
5 changed files with 142 additions and 16 deletions

View File

@@ -3,7 +3,7 @@
import { createHash, randomUUID } from 'node:crypto'
import { constants } from 'node:fs'
import { chmod, link, mkdir, open, readFile, unlink } from 'node:fs/promises'
import { basename, join } from 'node:path'
import { basename, dirname, join, resolve } from 'node:path'
import {
AttachmentError,
AttachmentId,
@@ -78,6 +78,25 @@ async function syncDirectory(path: string): Promise<void> {
}
}
/**
* Create one private directory tree and persist every newly published ancestor.
* @param path - absolute directory to create.
*/
async function ensureDurableDirectory(path: string): Promise<void> {
const target = resolve(path)
const firstCreated = await mkdir(target, { recursive: true, mode: 0o700 })
await chmod(target, 0o700)
if (firstCreated === undefined) return
const highestCreated = resolve(firstCreated)
let created = target
while (true) {
await syncDirectory(dirname(created))
if (created === highestCreated) return
created = dirname(created)
}
}
/**
* Save and verify immutable image bytes below a versioned attachment root.
* @param root - absolute `DSH_HOME/attachments/v1` root.
@@ -91,10 +110,8 @@ export async function saveImageFile(root: string, input: SaveImageAttachment, li
const sha256 = digest(input.data)
const bucket = join(root, 'objects', sha256.slice(0, 2))
const staging = join(root, 'tmp')
await mkdir(bucket, { recursive: true, mode: 0o700 })
await mkdir(staging, { recursive: true, mode: 0o700 })
await chmod(bucket, 0o700)
await chmod(staging, 0o700)
await ensureDurableDirectory(bucket)
await ensureDurableDirectory(staging)
const temporary = join(staging, randomUUID())
const target = objectPath(root, sha256)
let handle
@@ -112,11 +129,10 @@ export async function saveImageFile(root: string, input: SaveImageAttachment, li
const existing = new Uint8Array(await readFile(target))
if (digest(existing) !== sha256) throw new AttachmentError('Stored attachment failed integrity verification.', 'ATTACHMENT_CORRUPT')
}
// The synced file becomes durable only once its directory entries are: sync
// the bucket (the new object entry) and its parent (the possibly new bucket
// entry) before this reference can reach a session checkpoint. The dedup
// path syncs too — the earlier save that created the entry may have crashed
// before its own directory sync.
// 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'))
await unlink(temporary)

View File

@@ -1,12 +1,26 @@
import { createHash } from 'node:crypto'
import { constants } from 'node:fs'
import { chmod, mkdir, readFile, stat, 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 { afterEach, describe, expect, it, vi } from 'vitest'
import type { ImageAttachmentLimits } from '@deepseek-ai/dsh-attachment'
import { readImageFile, saveImageFile } from '../src/store.ts'
const fsControl = vi.hoisted(() => ({ syncedDirectories: [] as string[] }))
vi.mock('node:fs/promises', async (importOriginal) => {
const actual = await importOriginal<typeof import('node:fs/promises')>()
return {
...actual,
async open(...args: Parameters<typeof actual.open>): ReturnType<typeof actual.open> {
if (args[1] === constants.O_RDONLY) fsControl.syncedDirectories.push(String(args[0]))
return actual.open(...args)
},
}
})
const PNG = Uint8Array.from(Buffer.from(
'iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAQAAAC1HAwCAAAAC0lEQVR42mNk+A8AAQUBAScY42YAAAAASUVORK5CYII=',
'base64',
@@ -33,6 +47,27 @@ afterEach(async () => {
})
describe('local attachment store', () => {
it.skipIf(process.platform === 'win32')('syncs every newly created object ancestor before returning', async () => {
const storageRoot = await root()
const base = join(storageRoot, '..', '..')
const sha256 = createHash('sha256').update(PNG).digest('hex')
const objects = join(storageRoot, 'objects')
const bucket = join(objects, sha256.slice(0, 2))
fsControl.syncedDirectories.length = 0
await saveImageFile(storageRoot, { data: PNG, mediaType: 'image/png' }, LIMITS)
expect(fsControl.syncedDirectories).toEqual([
objects,
storageRoot,
join(storageRoot, '..'),
base,
storageRoot,
bucket,
objects,
])
})
it('publishes one private content-addressed object and deduplicates equal bytes', async () => {
const storageRoot = await root()
const first = await saveImageFile(storageRoot, {

View File

@@ -3,11 +3,13 @@
*/
import { z } from 'zod'
import type { ModelModality } from '@deepseek-ai/dsh-llm'
import type { RequestPayload, ResponseValue } from './rpc-map.ts'
import type { Wire } from './rpc.schema.ts'
import { imageMediaTypeSchema } from './sessions.schema.ts'
const modalitySchema = z.union([z.literal('text'), z.literal('image')])
/** Merge-extensible modality passthrough: declaration merging cannot extend a runtime Zod union. */
const modalitySchema = z.string() as unknown as z.ZodType<ModelModality>
/** host.describe request payload (empty object literal). */
export const hostDescribeRequestSchema = z.object({}) satisfies z.ZodType<Wire<RequestPayload<'host.describe'>>>

View File

@@ -1,12 +1,24 @@
import { describe, expect, it, vi } from 'vitest'
import type { ApiProxy, HostFrame, MuxFrame } from '../src/api/index.ts'
import type { ResponseValue } from '../src/api/rpc-map.ts'
import type { ClientResponse, RpcMessage, RpcReceipt, RpcRequest } from '../src/api/rpc.ts'
import { RpcId } from '../src/api/rpc.ts'
import { toFetchHandler } from '../src/fetch/handler.ts'
import { AbstractApiClient, InProcessApiClient } from '../src/fetch/client.ts'
declare module '@deepseek-ai/dsh-llm' {
interface ModelModalityMap {
audio: 'audio'
}
}
/** Minimal in-memory ApiProxy: echoes rpcIds, scripts one frame per stream. */
function fakeApi(overrides: Partial<{ muxFrames: MuxFrame[]; hostFrames: HostFrame[]; crashOn: string }> = {}): ApiProxy {
function fakeApi(overrides: Partial<{
muxFrames: MuxFrame[]
hostFrames: HostFrame[]
crashOn: string
hostDescription: ResponseValue<'host.describe'>
}> = {}): ApiProxy {
const muxFrames = overrides.muxFrames ?? [{ type: 'session/subscribed', sessionId: 's1' as never, lastSeq: -1 }]
const hostFrames = overrides.hostFrames ?? [{ type: 'host/session-removed', sessionId: 's1' as never }]
async function * stream<F>(frames: F[], signal: AbortSignal): AsyncGenerator<RpcRequest<F>> {
@@ -45,7 +57,13 @@ function fakeApi(overrides: Partial<{ muxFrames: MuxFrame[]; hostFrames: HostFra
},
host: {
async describe(request) {
return { rpcId: request.rpcId, result: { ok: true, value: { version: 'v', cwd: '/w', attachedSessions: 0 } } }
return {
rpcId: request.rpcId,
result: {
ok: true,
value: overrides.hostDescription ?? { version: 'v', cwd: '/w', attachedSessions: 0 },
},
}
},
},
workspace: {
@@ -137,6 +155,40 @@ describe('unary round trip (handler ⇄ client, no network)', () => {
expect((await c.host.describe({})).result.ok).toBe(true)
})
it('round-trips a declaration-merged model modality through host.describe', async () => {
const c = client(fakeApi({
hostDescription: {
version: 'v',
cwd: '/w',
activeModel: {
provider: 'future',
id: 'audio-model',
name: 'Audio Model',
inputModalities: ['text', 'audio'],
outputModalities: ['audio'],
},
attachedSessions: 0,
},
}))
const response = await c.host.describe({})
expect(response.result).toEqual({
ok: true,
value: {
version: 'v',
cwd: '/w',
activeModel: {
provider: 'future',
id: 'audio-model',
name: 'Audio Model',
inputModalities: ['text', 'audio'],
outputModalities: ['audio'],
},
attachedSessions: 0,
},
})
})
it('round-trips command.list / command.execute / skill.list through the wire form', async () => {
const c = client()
const list = await c.commands.list({ sessionId: 's' as never })

View File

@@ -155,11 +155,32 @@ describe('sessions domain schemas', () => {
})
describe('host domain schemas', () => {
it('validates describe request/value', () => {
it('validates describe request/value and preserves merge-extensible modalities', () => {
expect(hostDescribeRequestSchema.parse({})).toEqual({})
const value = hostDescribeValueSchema.parse({ version: '1', cwd: '/x', provider: 'p', model: 'm', attachedSessions: 2 })
const value = hostDescribeValueSchema.parse({
version: '1',
cwd: '/x',
provider: 'p',
model: 'm',
activeModel: {
provider: 'p',
id: 'm',
name: 'Model',
inputModalities: ['text', 'audio'],
outputModalities: ['text', 'audio'],
},
attachedSessions: 2,
})
expect(value.attachedSessions).toBe(2)
expect(value.activeModel?.inputModalities).toEqual(['text', 'audio'])
expect(value.activeModel?.outputModalities).toEqual(['text', 'audio'])
expect(hostDescribeValueSchema.parse({ version: '1', cwd: '/x', attachedSessions: 0 }).provider).toBeUndefined()
expect(() => hostDescribeValueSchema.parse({
version: '1',
cwd: '/x',
activeModel: { provider: 'p', id: 'm', name: 'Model', inputModalities: [{ type: 'audio' }] },
attachedSessions: 0,
})).toThrow()
})
})