From a1835c3228682bc459556ab2264ebd52ec23c18b Mon Sep 17 00:00:00 2001 From: Yichen Jiang Date: Mon, 27 Jul 2026 14:29:19 +0800 Subject: [PATCH] fix(gui): close attachment durability gaps --- .../attachment/attachment-local/src/store.ts | 36 ++++++++---- .../attachment-local/tests/store.spec.ts | 37 +++++++++++- packages/host/apiproxy/src/api/host.schema.ts | 4 +- .../host/apiproxy/tests/fetch-carrier.spec.ts | 56 ++++++++++++++++++- .../host/apiproxy/tests/rpc-schemas.spec.ts | 25 ++++++++- 5 files changed, 142 insertions(+), 16 deletions(-) diff --git a/packages/attachment/attachment-local/src/store.ts b/packages/attachment/attachment-local/src/store.ts index 34431f5a2b..e64647bee2 100644 --- a/packages/attachment/attachment-local/src/store.ts +++ b/packages/attachment/attachment-local/src/store.ts @@ -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 { } } +/** + * Create one private directory tree and persist every newly published ancestor. + * @param path - absolute directory to create. + */ +async function ensureDurableDirectory(path: string): Promise { + 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) diff --git a/packages/attachment/attachment-local/tests/store.spec.ts b/packages/attachment/attachment-local/tests/store.spec.ts index d7ee6de31b..73ebb3bdec 100644 --- a/packages/attachment/attachment-local/tests/store.spec.ts +++ b/packages/attachment/attachment-local/tests/store.spec.ts @@ -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() + return { + ...actual, + async open(...args: Parameters): ReturnType { + 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, { diff --git a/packages/host/apiproxy/src/api/host.schema.ts b/packages/host/apiproxy/src/api/host.schema.ts index 4b484ae594..4e42ad55ce 100644 --- a/packages/host/apiproxy/src/api/host.schema.ts +++ b/packages/host/apiproxy/src/api/host.schema.ts @@ -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 /** host.describe request payload (empty object literal). */ export const hostDescribeRequestSchema = z.object({}) satisfies z.ZodType>> diff --git a/packages/host/apiproxy/tests/fetch-carrier.spec.ts b/packages/host/apiproxy/tests/fetch-carrier.spec.ts index 9f4f11b9df..00616da0b8 100644 --- a/packages/host/apiproxy/tests/fetch-carrier.spec.ts +++ b/packages/host/apiproxy/tests/fetch-carrier.spec.ts @@ -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(frames: F[], signal: AbortSignal): AsyncGenerator> { @@ -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 }) diff --git a/packages/host/apiproxy/tests/rpc-schemas.spec.ts b/packages/host/apiproxy/tests/rpc-schemas.spec.ts index d257020298..042e5aa2a0 100644 --- a/packages/host/apiproxy/tests/rpc-schemas.spec.ts +++ b/packages/host/apiproxy/tests/rpc-schemas.spec.ts @@ -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() }) })