From f0ee1f0b19ffac1ede6ba03ce7b0c39eabccc096 Mon Sep 17 00:00:00 2001 From: Zerlight Wu Date: Thu, 3 Sep 2026 14:14:09 +0800 Subject: [PATCH 1/5] feat(schema,engine): attachment records and a filesystem blob store --- .../foundation/schema/src/model/attachment.ts | 83 +++++++++ packages/foundation/schema/src/model/index.ts | 1 + .../engine/src/__tests__/blob-store.test.ts | 120 +++++++++++++ .../host/engine/src/attachment/blob-store.ts | 170 ++++++++++++++++++ .../host/engine/src/attachment/mime-sniff.ts | 31 ++++ packages/host/engine/src/index.ts | 1 + 6 files changed, 406 insertions(+) create mode 100644 packages/foundation/schema/src/model/attachment.ts create mode 100644 packages/host/engine/src/__tests__/blob-store.test.ts create mode 100644 packages/host/engine/src/attachment/blob-store.ts create mode 100644 packages/host/engine/src/attachment/mime-sniff.ts diff --git a/packages/foundation/schema/src/model/attachment.ts b/packages/foundation/schema/src/model/attachment.ts new file mode 100644 index 000000000..5a7f642d0 --- /dev/null +++ b/packages/foundation/schema/src/model/attachment.ts @@ -0,0 +1,83 @@ +import { z } from 'zod'; +import { AttachmentIdSchema, TimestampSchema } from './primitives'; + +/** + * The immutable attachment store: prompts and session resources reference `AttachmentRecord`s, + * whose bytes live in a content-addressed blob store keyed by their own SHA-256. Records and + * reference edges are rows; bytes never are. + */ + +const rSha256Hex = /^[0-9a-f]{64}$/; + +/** Lowercase hex SHA-256 digest of a blob's bytes. */ +export const Sha256HexSchema = z.string().regex(rSha256Hex); + +/** Content address: `sha256:`. The storage path is derived from it, never stored or exposed. */ +export const BlobIdSchema = z + .string() + .regex(/^sha256:[0-9a-f]{64}$/) + .brand<'BlobId'>(); +export type BlobId = z.infer; + +export function blobIdFromSha256(hex: string): BlobId { + return BlobIdSchema.parse(`sha256:${hex.toLowerCase()}`); +} + +/** Upload ID: daemon-minted identity of one in-flight upload lease. */ +export const UploadIdSchema = z.string().min(1).brand<'UploadId'>(); +export type UploadId = z.infer; + +export const BlobRecordSchema = z.object({ + blobId: BlobIdSchema, + sizeBytes: z.number().int().nonnegative(), + createdAt: TimestampSchema, +}); +export type BlobRecord = z.infer; + +/** Open string on purpose: a client without a renderer for a kind shows a generic card. */ +export const AttachmentKindSchema = z.string().min(1).max(32); + +export const MAX_ATTACHMENT_METADATA_BYTES = 4096; + +/** Business identity of one attachment; many records may share one blob. */ +export const AttachmentRecordSchema = z.object({ + attachmentId: AttachmentIdSchema, + kind: AttachmentKindSchema, + name: z.string().min(1).max(255), + mimeType: z.string().min(1).max(255), + sizeBytes: z.number().int().nonnegative(), + metadata: z + .record(z.string(), z.unknown()) + .refine((value) => JSON.stringify(value).length <= MAX_ATTACHMENT_METADATA_BYTES, { + message: `Attachment metadata exceeds ${MAX_ATTACHMENT_METADATA_BYTES} bytes`, + }), + createdAt: TimestampSchema, +}); +export type AttachmentRecord = z.infer; + +/** Which blob holds which representation of an attachment; `original` is the only variant today. */ +export const AttachmentBlobSchema = z.object({ + attachmentId: AttachmentIdSchema, + variant: z.string().min(1).max(32), + blobId: BlobIdSchema, +}); +export type AttachmentBlob = z.infer; + +/** + * An in-flight or committed-but-unclaimed upload. The lease is a GC root: it pins `blobId` once + * the declared hash is known to exist, and `attachmentId` once committed, until a prompt or + * session resource references the attachment (the claim) or the lease expires (the reaper). + */ +export const UploadLeaseSchema = z.object({ + uploadId: UploadIdSchema, + declaredSha256: Sha256HexSchema, + declaredSize: z.number().int().nonnegative(), + name: z.string().min(1).max(255), + mimeType: z.string().min(1).max(255).optional(), + kind: AttachmentKindSchema, + blobId: BlobIdSchema.optional(), + attachmentId: AttachmentIdSchema.optional(), + expiresAt: TimestampSchema, + createdAt: TimestampSchema, +}); +export type UploadLease = z.infer; diff --git a/packages/foundation/schema/src/model/index.ts b/packages/foundation/schema/src/model/index.ts index 8eab83ae9..53f15a07e 100644 --- a/packages/foundation/schema/src/model/index.ts +++ b/packages/foundation/schema/src/model/index.ts @@ -2,6 +2,7 @@ export * from './account'; export * from './agent'; export * from './agent-runtime'; export * from './artifact'; +export * from './attachment'; export * from './browser'; export * from './content'; export * from './conversation'; diff --git a/packages/host/engine/src/__tests__/blob-store.test.ts b/packages/host/engine/src/__tests__/blob-store.test.ts new file mode 100644 index 000000000..42aad0740 --- /dev/null +++ b/packages/host/engine/src/__tests__/blob-store.test.ts @@ -0,0 +1,120 @@ +import { createHash } from 'node:crypto'; +import { mkdtemp, readdir, readFile, rm, stat } from 'node:fs/promises'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; +import { blobIdFromSha256 } from '@linkcode/schema'; +import { afterEach, describe, expect, it } from 'vitest'; +import { BlobIntegrityError, FsBlobStore } from '../attachment/blob-store'; +import { declaredMimeTypeMatches, sniffImageMimeType } from '../attachment/mime-sniff'; + +const temporaryDirectories: string[] = []; + +afterEach(async () => { + await Promise.all( + temporaryDirectories.splice(0).map((path) => rm(path, { recursive: true, force: true })), + ); +}); + +async function storeInTempDir(): Promise<{ store: FsBlobStore; root: string }> { + const root = await mkdtemp(join(tmpdir(), 'linkcode-blob-store-')); + temporaryDirectories.push(root); + return { store: new FsBlobStore(root), root }; +} + +function sha256(bytes: Uint8Array): string { + return createHash('sha256').update(bytes).digest('hex'); +} + +describe('FsBlobStore', () => { + it('publishes verified bytes under their hash and dedupes identical commits', async () => { + const { store, root } = await storeInTempDir(); + const bytes = Buffer.from('hello attachment store'); + const expected = { sha256: sha256(bytes), sizeBytes: bytes.byteLength }; + + const first = await store.stage('upload-1'); + await first.write(0, bytes.subarray(0, 5)); + await first.write(5, bytes.subarray(5)); + const blobId = await first.commit(expected); + + expect(blobId).toBe(blobIdFromSha256(expected.sha256)); + expect(store.pathOf(blobId)).toBe( + join(root, 'sha256', expected.sha256.slice(0, 2), expected.sha256.slice(2)), + ); + expect(await readFile(store.pathOf(blobId))).toEqual(bytes); + expect(await store.stat(blobId)).toEqual({ sizeBytes: bytes.byteLength }); + expect((await stat(store.pathOf(blobId))).mode & 0o222).toBe(0); + + const second = await store.stage('upload-2'); + await second.write(0, bytes); + expect(await second.commit(expected)).toBe(blobId); + expect(await store.list()).toEqual([blobId]); + expect(await readdir(join(root, 'tmp'))).toEqual([]); + }); + + it('refuses a size or hash mismatch and leaves nothing behind', async () => { + const { store, root } = await storeInTempDir(); + const bytes = Buffer.from('payload'); + + const shortStage = await store.stage('short'); + await shortStage.write(0, bytes); + await expect( + shortStage.commit({ sha256: sha256(bytes), sizeBytes: bytes.byteLength + 1 }), + ).rejects.toBeInstanceOf(BlobIntegrityError); + + const wrongHash = await store.stage('wrong-hash'); + await wrongHash.write(0, bytes); + await expect( + wrongHash.commit({ sha256: sha256(Buffer.from('other')), sizeBytes: bytes.byteLength }), + ).rejects.toBeInstanceOf(BlobIntegrityError); + + const aborted = await store.stage('aborted'); + await aborted.write(0, bytes); + await aborted.abort(); + + expect(await readdir(join(root, 'tmp'))).toEqual([]); + expect(await store.list()).toEqual([]); + await expect(store.stage('../escape')).rejects.toThrow('Invalid upload id'); + }); + + it('purges staging orphans and deletes blobs on request', async () => { + const { store, root } = await storeInTempDir(); + const bytes = Buffer.from('to be deleted'); + const stage = await store.stage('committed'); + await stage.write(0, bytes); + const blobId = await stage.commit({ sha256: sha256(bytes), sizeBytes: bytes.byteLength }); + const orphan = await store.stage('orphan'); + await orphan.write(0, bytes); + + await store.purgeStaging(); + await expect(stat(join(root, 'tmp'))).rejects.toMatchObject({ code: 'ENOENT' }); + expect(await store.list()).toEqual([blobId]); + + await store.delete(blobId); + expect(await store.stat(blobId)).toBeUndefined(); + expect(await store.list()).toEqual([]); + await expect(store.delete(blobId)).resolves.toBeUndefined(); + }); +}); + +describe('mime sniff', () => { + const png = new Uint8Array([0x89, 0x50, 0x4e, 0x47, 0x0d, 0x0a, 0x1a, 0x0a, 0, 0]); + const webp = Buffer.concat([Buffer.from('RIFF'), Buffer.alloc(4), Buffer.from('WEBPVP8 ')]); + + it('recognizes the supported image types and nothing else', () => { + expect(sniffImageMimeType(png)).toBe('image/png'); + expect(sniffImageMimeType(new Uint8Array([0xff, 0xd8, 0xff, 0xe0]))).toBe('image/jpeg'); + expect(sniffImageMimeType(Buffer.from('GIF89a......'))).toBe('image/gif'); + expect(sniffImageMimeType(webp)).toBe('image/webp'); + expect(sniffImageMimeType(Buffer.from('RIFF....WAVE'))).toBeUndefined(); + expect(sniffImageMimeType(Buffer.from('%PDF-1.7'))).toBeUndefined(); + expect(sniffImageMimeType(new Uint8Array(0))).toBeUndefined(); + }); + + it('holds image declarations to their bytes and trusts the rest', () => { + expect(declaredMimeTypeMatches('image/png', png)).toBe(true); + expect(declaredMimeTypeMatches('image/jpeg', png)).toBe(false); + expect(declaredMimeTypeMatches('image/svg+xml', Buffer.from(''))).toBe(false); + expect(declaredMimeTypeMatches('application/pdf', Buffer.from('%PDF-1.7'))).toBe(true); + expect(declaredMimeTypeMatches('text/plain', png)).toBe(true); + }); +}); diff --git a/packages/host/engine/src/attachment/blob-store.ts b/packages/host/engine/src/attachment/blob-store.ts new file mode 100644 index 000000000..54e216546 --- /dev/null +++ b/packages/host/engine/src/attachment/blob-store.ts @@ -0,0 +1,170 @@ +import { createHash } from 'node:crypto'; +import { createReadStream } from 'node:fs'; +import type { FileHandle } from 'node:fs/promises'; +import { chmod, mkdir, open, readdir, rename, rm, stat } from 'node:fs/promises'; +import { dirname, join } from 'node:path'; +import type { BlobId } from '@linkcode/schema'; +import { blobIdFromSha256 } from '@linkcode/schema'; + +const BLOB_ID_PREFIX = 'sha256:'; +const rShard = /^[0-9a-f]{2}$/; +const rShardRest = /^[0-9a-f]{62}$/; +const rUploadId = /^[\w-]{1,128}$/; + +/** The declared size or SHA-256 did not match the staged bytes; the staging file is gone. */ +export class BlobIntegrityError extends Error { + constructor(message: string, options?: ErrorOptions) { + super(message, options); + this.name = 'BlobIntegrityError'; + } +} + +/** + * Content-addressed byte storage. Dedupe-by-hash is scoped to one trust domain: every peer of + * this store is the same account, so a known hash counts as proof of possession. A multi-tenant + * implementation must scope dedupe per tenant or demand real proof-of-possession. + */ +export interface BlobStore { + /** Absolute path for same-host consumers (materialization, hosted files); existence not implied. */ + pathOf(blobId: BlobId): string; + stat(blobId: BlobId): Promise<{ sizeBytes: number } | undefined>; + /** Open staging for one upload; readers cannot observe the bytes until `commit`. */ + stage(uploadId: string): Promise; + delete(blobId: BlobId): Promise; + /** Every committed blob on disk — the "on disk" side of the boot mark-and-sweep. */ + list(): Promise; + /** Drop every staging file: no upload survives a restart. */ + purgeStaging(): Promise; +} + +export interface BlobStage { + write(offset: number, chunk: Uint8Array): Promise; + /** Verify size and SHA-256, then publish with one same-volume atomic rename. */ + commit(expected: { sha256: string; sizeBytes: number }): Promise; + abort(): Promise; +} + +class FsBlobStage implements BlobStage { + private handle: FileHandle | undefined; + + constructor( + private readonly store: FsBlobStore, + private readonly path: string, + handle: FileHandle, + ) { + this.handle = handle; + } + + async write(offset: number, chunk: Uint8Array): Promise { + if (!this.handle) throw new Error('Blob stage is closed'); + await this.handle.write(chunk, 0, chunk.byteLength, offset); + } + + async commit(expected: { sha256: string; sizeBytes: number }): Promise { + await this.close(); + const sizeBytes = (await stat(this.path)).size; + const sha256 = await sha256OfFile(this.path); + if (sizeBytes !== expected.sizeBytes || sha256 !== expected.sha256.toLowerCase()) { + await rm(this.path, { force: true }); + throw new BlobIntegrityError( + sizeBytes === expected.sizeBytes + ? 'Uploaded bytes do not match the declared SHA-256' + : `Uploaded ${sizeBytes} bytes, declared ${expected.sizeBytes}`, + ); + } + const blobId = blobIdFromSha256(sha256); + await this.store.publish(this.path, blobId); + return blobId; + } + + async abort(): Promise { + await this.close(); + await rm(this.path, { force: true }); + } + + private async close(): Promise { + const handle = this.handle; + this.handle = undefined; + await handle?.close(); + } +} + +/** `/sha256/ab/cdef…` for committed blobs, `/tmp/` while staging. */ +export class FsBlobStore implements BlobStore { + constructor(private readonly root: string) {} + + pathOf(blobId: BlobId): string { + const hex = blobId.slice(BLOB_ID_PREFIX.length); + return join(this.root, 'sha256', hex.slice(0, 2), hex.slice(2)); + } + + async stat(blobId: BlobId): Promise<{ sizeBytes: number } | undefined> { + try { + const info = await stat(this.pathOf(blobId)); + return info.isFile() ? { sizeBytes: info.size } : undefined; + } catch (error) { + if ((error as NodeJS.ErrnoException).code === 'ENOENT') return undefined; + throw error; + } + } + + async stage(uploadId: string): Promise { + if (!rUploadId.test(uploadId)) throw new Error(`Invalid upload id: ${uploadId}`); + const dir = join(this.root, 'tmp'); + await mkdir(dir, { recursive: true }); + const path = join(dir, uploadId); + return new FsBlobStage(this, path, await open(path, 'w')); + } + + async delete(blobId: BlobId): Promise { + await rm(this.pathOf(blobId), { force: true }); + } + + async list(): Promise { + const ids: BlobId[] = []; + const shards = await readdirOrEmpty(join(this.root, 'sha256')); + for (let i = 0, len = shards.length; i < len; i++) { + const shard = shards[i]; + if (!rShard.test(shard)) continue; + // eslint-disable-next-line no-await-in-loop -- one directory at a time keeps memory flat + const entries = await readdirOrEmpty(join(this.root, 'sha256', shard)); + for (let j = 0, jlen = entries.length; j < jlen; j++) { + const rest = entries[j]; + if (rShardRest.test(rest)) ids.push(blobIdFromSha256(`${shard}${rest}`)); + } + } + return ids; + } + + async purgeStaging(): Promise { + await rm(join(this.root, 'tmp'), { recursive: true, force: true }); + } + + /** Publish a verified staging file. Losing the rename race to identical bytes is success. */ + async publish(stagingPath: string, blobId: BlobId): Promise { + const dest = this.pathOf(blobId); + await mkdir(dirname(dest), { recursive: true }); + await chmod(stagingPath, 0o444); + try { + await rename(stagingPath, dest); + } catch (error) { + if ((await this.stat(blobId)) === undefined) throw error; + await rm(stagingPath, { force: true }); + } + } +} + +async function readdirOrEmpty(path: string): Promise { + try { + return await readdir(path); + } catch (error) { + if ((error as NodeJS.ErrnoException).code === 'ENOENT') return []; + throw error; + } +} + +async function sha256OfFile(path: string): Promise { + const hash = createHash('sha256'); + for await (const chunk of createReadStream(path)) hash.update(chunk as Uint8Array); + return hash.digest('hex'); +} diff --git a/packages/host/engine/src/attachment/mime-sniff.ts b/packages/host/engine/src/attachment/mime-sniff.ts new file mode 100644 index 000000000..0d318dc82 --- /dev/null +++ b/packages/host/engine/src/attachment/mime-sniff.ts @@ -0,0 +1,31 @@ +import type { SupportedAttachmentImageMimeType } from '@linkcode/schema'; + +const PNG_MAGIC = [0x89, 0x50, 0x4e, 0x47, 0x0d, 0x0a, 0x1a, 0x0a]; + +function startsWith(bytes: Uint8Array, magic: readonly number[], offset = 0): boolean { + if (bytes.byteLength < offset + magic.length) return false; + for (let i = 0, len = magic.length; i < len; i++) { + if (bytes[offset + i] !== magic[i]) return false; + } + return true; +} + +function ascii(text: string): number[] { + return Array.from(text, (char) => char.charCodeAt(0)); +} + +/** The image type the leading bytes actually are, for the four types adapters accept. */ +export function sniffImageMimeType(head: Uint8Array): SupportedAttachmentImageMimeType | undefined { + if (startsWith(head, [0xff, 0xd8, 0xff])) return 'image/jpeg'; + if (startsWith(head, PNG_MAGIC)) return 'image/png'; + if (startsWith(head, ascii('GIF87a')) || startsWith(head, ascii('GIF89a'))) return 'image/gif'; + if (startsWith(head, ascii('RIFF')) && startsWith(head, ascii('WEBP'), 8)) return 'image/webp'; + return undefined; +} + +/** A declared `image/*` type must match its bytes — model APIs refuse the mismatch later and + * less legibly. Other declarations have no reliable sniff and are trusted. */ +export function declaredMimeTypeMatches(declared: string, head: Uint8Array): boolean { + if (!declared.startsWith('image/')) return true; + return sniffImageMimeType(head) === declared; +} diff --git a/packages/host/engine/src/index.ts b/packages/host/engine/src/index.ts index 807735e7c..31647962b 100644 --- a/packages/host/engine/src/index.ts +++ b/packages/host/engine/src/index.ts @@ -7,6 +7,7 @@ export type { ProviderConfigStore } from './agent/provider-config'; export type { TranslatorService, TranslatorUpstream } from './agent/translator'; export type { AssetService } from './asset/service'; +export { type BlobStage, type BlobStore, FsBlobStore } from './attachment/blob-store'; export type { LoopStore, ScheduleStore } from './automation'; export { ConversationSessionBusyError, From abdb20bee32b117af5dac881e2ed2759faa89cd2 Mon Sep 17 00:00:00 2001 From: Zerlight Wu Date: Thu, 3 Sep 2026 14:22:27 +0800 Subject: [PATCH 2/5] feat(engine): attachment store, upload leases, and reference-rooted GC --- .../schema/src/model/session-resource.ts | 4 +- .../src/__tests__/attachment-gc.test.ts | 188 ++++++++++++++++++ .../engine/src/attachment/attachment-store.ts | 143 +++++++++++++ packages/host/engine/src/attachment/gc.ts | 70 +++++++ .../src/conversation/conversation-store.ts | 16 ++ packages/host/engine/src/deps.ts | 8 + packages/host/engine/src/engine.ts | 45 ++++- packages/host/engine/src/index.ts | 7 + .../engine/src/resource/resource-store.ts | 11 +- 9 files changed, 483 insertions(+), 9 deletions(-) create mode 100644 packages/host/engine/src/__tests__/attachment-gc.test.ts create mode 100644 packages/host/engine/src/attachment/attachment-store.ts create mode 100644 packages/host/engine/src/attachment/gc.ts diff --git a/packages/foundation/schema/src/model/session-resource.ts b/packages/foundation/schema/src/model/session-resource.ts index c8f8b92c9..5ed746ea4 100644 --- a/packages/foundation/schema/src/model/session-resource.ts +++ b/packages/foundation/schema/src/model/session-resource.ts @@ -1,5 +1,5 @@ import { z } from 'zod'; -import { SessionIdSchema, TimestampSchema } from './primitives'; +import { AttachmentIdSchema, SessionIdSchema, TimestampSchema } from './primitives'; export const SessionResourceIdSchema = z.string().min(1).brand<'SessionResourceId'>(); export type SessionResourceId = z.infer; @@ -19,6 +19,8 @@ export const SessionResourceSchema = z.object({ kind: z.enum(['file', 'image', 'document', 'site', 'link']), status: z.enum(['processing', 'generating', 'ready', 'failed', 'unavailable']), locator: SessionResourceLocatorSchema, + /** Set when the bytes live in the attachment store; the resource row is then a GC root. */ + attachmentId: AttachmentIdSchema.optional(), mimeType: z.string().min(1).optional(), sizeBytes: z.number().int().nonnegative().optional(), error: z.string().min(1).optional(), diff --git a/packages/host/engine/src/__tests__/attachment-gc.test.ts b/packages/host/engine/src/__tests__/attachment-gc.test.ts new file mode 100644 index 000000000..a82a3ccf3 --- /dev/null +++ b/packages/host/engine/src/__tests__/attachment-gc.test.ts @@ -0,0 +1,188 @@ +import { createHash } from 'node:crypto'; +import { mkdtemp, rm, stat } from 'node:fs/promises'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; +import type { AttachmentId, BlobId, UploadLease } from '@linkcode/schema'; +import { + AttachmentIdSchema, + ConversationOperationSchema, + ConversationTurnSchema, + PromptRecordSchema, + SessionResourceSchema, + UploadIdSchema, +} from '@linkcode/schema'; +import { afterEach, describe, expect, it } from 'vitest'; +import { InMemoryAttachmentStore } from '../attachment/attachment-store'; +import { FsBlobStore } from '../attachment/blob-store'; +import { ATTACHMENT_GC_GRACE_MS, AttachmentGc, UPLOAD_LEASE_TTL_MS } from '../attachment/gc'; +import { InMemoryConversationStore } from '../conversation/conversation-store'; +import { InMemoryResourceStore } from '../resource/resource-store'; + +const temporaryDirectories: string[] = []; + +afterEach(async () => { + await Promise.all( + temporaryDirectories.splice(0).map((path) => rm(path, { recursive: true, force: true })), + ); +}); + +function sha256(bytes: Uint8Array): string { + return createHash('sha256').update(bytes).digest('hex'); +} + +async function fixture() { + const root = await mkdtemp(join(tmpdir(), 'linkcode-attachment-gc-')); + temporaryDirectories.push(root); + const blobs = new FsBlobStore(root); + const conversations = new InMemoryConversationStore(); + const resources = new InMemoryResourceStore(); + const store = new InMemoryAttachmentStore(() => [ + ...conversations.referencedAttachmentIds(), + ...resources.referencedAttachmentIds(), + ]); + const clock = { now: 1_000_000 }; + const gc = new AttachmentGc(store, blobs, () => clock.now); + + async function publish(text: string): Promise { + const bytes = Buffer.from(text); + const stage = await blobs.stage(`stage-${sha256(bytes).slice(0, 8)}`); + await stage.write(0, bytes); + return stage.commit({ sha256: sha256(bytes), sizeBytes: bytes.byteLength }); + } + + async function commit(id: string, text: string, uploadId?: string): Promise { + const blobId = await publish(text); + await store.commitAttachment({ + blob: { blobId, sizeBytes: Buffer.byteLength(text), createdAt: clock.now }, + attachment: { + attachmentId: AttachmentIdSchema.parse(id), + kind: 'file', + name: `${id}.txt`, + mimeType: 'text/plain', + sizeBytes: Buffer.byteLength(text), + metadata: {}, + createdAt: clock.now, + }, + uploadId: uploadId === undefined ? undefined : UploadIdSchema.parse(uploadId), + }); + return blobId; + } + + function lease(uploadId: string, text: string): UploadLease { + return { + uploadId: UploadIdSchema.parse(uploadId), + declaredSha256: sha256(Buffer.from(text)), + declaredSize: Buffer.byteLength(text), + name: 'draft.txt', + kind: 'file', + expiresAt: clock.now + UPLOAD_LEASE_TTL_MS, + createdAt: clock.now, + }; + } + + return { blobs, clock, commit, conversations, gc, lease, publish, resources, store }; +} + +describe('AttachmentGc', () => { + it('collects only what no prompt, resource, or lease roots, once the grace window passes', async () => { + const f = await fixture(); + const promptBlob = await f.commit('att-prompt', 'referenced by a prompt'); + const resourceBlob = await f.commit('att-resource', 'referenced by a resource'); + await f.store.beginUpload(f.lease('up-draft', 'held by a lease')); + const leasedBlob = await f.commit('att-leased', 'held by a lease', 'up-draft'); + const strayBlob = await f.commit('att-stray', 'nothing roots this'); + + await f.conversations.persistTurnIntent({ + turn: ConversationTurnSchema.parse({ + turnId: 't-1', + sessionId: 's-1', + parentTurnId: null, + siblingOrdinal: 1, + input: { type: 'prompt', promptId: 'p-1' }, + runId: 'run-1', + state: 'preparing', + createdAt: 1, + }), + prompt: PromptRecordSchema.parse({ + promptId: 'p-1', + blocks: [{ type: 'attachment_ref', attachmentId: 'att-prompt' }], + contextAttachmentIds: [], + createdAt: 1, + }), + operation: ConversationOperationSchema.parse({ + operationId: 'op-1', + sessionId: 's-1', + kind: 'turn.submit', + state: 'open', + createdAt: 1, + }), + }); + await f.resources.save( + SessionResourceSchema.parse({ + resourceId: 'resource-1', + sessionId: 's-1', + direction: 'source', + name: 'brief.txt', + kind: 'file', + status: 'ready', + locator: { type: 'managed-file', path: f.blobs.pathOf(resourceBlob) }, + attachmentId: 'att-resource', + createdAt: 1, + updatedAt: 1, + }), + ); + + expect(await f.gc.sweep()).toEqual({ removedBlobs: [] }); + + f.clock.now += ATTACHMENT_GC_GRACE_MS + 1; + expect(await f.gc.sweep()).toEqual({ removedBlobs: [strayBlob] }); + expect(await f.blobs.stat(strayBlob)).toBeUndefined(); + expect(await f.store.getAttachment(AttachmentIdSchema.parse('att-stray'))).toBeUndefined(); + const survivors = await Promise.all( + [promptBlob, resourceBlob, leasedBlob].map((blobId) => f.blobs.stat(blobId)), + ); + expect(survivors.every(Boolean)).toBe(true); + + f.clock.now += UPLOAD_LEASE_TTL_MS; + expect(await f.gc.sweep()).toEqual({ removedBlobs: [leasedBlob] }); + expect(await f.store.getLease(UploadIdSchema.parse('up-draft'))).toBeUndefined(); + expect(await f.store.getAttachment(AttachmentIdSchema.parse('att-prompt'))).toMatchObject({ + blobId: promptBlob, + }); + expect(await f.store.getAttachment(AttachmentIdSchema.parse('att-resource'))).toMatchObject({ + blobId: resourceBlob, + }); + }); + + it('pins an existing blob at begin so a dedupe hit survives the reaper', async () => { + const f = await fixture(); + const blobId = await f.commit('att-old', 'shared bytes'); + f.clock.now += ATTACHMENT_GC_GRACE_MS + 1; + const pinned = await f.store.beginUpload(f.lease('up-dedupe', 'shared bytes')); + expect(pinned.blobId).toBe(blobId); + + expect(await f.gc.sweep()).toEqual({ removedBlobs: [] }); + expect(await f.store.getAttachment(AttachmentIdSchema.parse('att-old'))).toBeUndefined(); + expect(await f.store.getBlob(blobId)).toBeDefined(); + expect(await f.blobs.stat(blobId)).toBeDefined(); + await expect(f.store.beginUpload(f.lease('up-dedupe', 'shared bytes'))).rejects.toThrow( + 'already exists', + ); + }); + + it('boot sweep drops staging files and bytes that have no blob row', async () => { + const f = await fixture(); + const kept = await f.commit('att-kept', 'row and bytes'); + const orphan = await f.publish('bytes without a row'); + const staged = await f.blobs.stage('never-committed'); + await staged.write(0, Buffer.from('half an upload')); + + const report = await f.gc.bootSweep(); + expect(report.removedBlobs).toEqual([orphan]); + expect(await f.blobs.list()).toEqual([kept]); + await expect(stat(join(f.blobs.pathOf(kept), '..', '..', '..', 'tmp'))).rejects.toMatchObject({ + code: 'ENOENT', + }); + expect(f.store.getAttachment('att-kept' as AttachmentId)).resolves.toBeDefined(); + }); +}); diff --git a/packages/host/engine/src/attachment/attachment-store.ts b/packages/host/engine/src/attachment/attachment-store.ts new file mode 100644 index 000000000..d0ea54873 --- /dev/null +++ b/packages/host/engine/src/attachment/attachment-store.ts @@ -0,0 +1,143 @@ +import type { + AttachmentId, + AttachmentRecord, + BlobId, + BlobRecord, + Timestamp, + UploadId, + UploadLease, +} from '@linkcode/schema'; +import { blobIdFromSha256 } from '@linkcode/schema'; + +/** An attachment record joined with the blob holding its `original` bytes. */ +export interface StoredAttachment extends AttachmentRecord { + readonly blobId: BlobId; +} + +export interface AttachmentCommit { + readonly blob: BlobRecord; + readonly attachment: AttachmentRecord; + /** The lease that staged the bytes; it keeps pinning the new attachment until claimed. */ + readonly uploadId?: UploadId; +} + +export interface AttachmentSweepWindow { + readonly now: Timestamp; + /** Rows created at or after this instant are never collected: their root may be one + * transaction away. */ + readonly graceBefore: Timestamp; +} + +/** + * Attachment metadata, reference roots, and upload leases. The daemon implements it on the graph + * connection so a lease claim rides the submit transaction. `sweep` MUST read its roots — prompt + * refs, session-resource refs, unexpired leases — inside its own transaction: a claim landing + * between a root read and the delete would otherwise lose a referenced attachment. + */ +export interface AttachmentStore { + getAttachment(attachmentId: AttachmentId): Promise; + listAttachments(attachmentIds: readonly AttachmentId[]): Promise; + getBlob(blobId: BlobId): Promise; + getLease(uploadId: UploadId): Promise; + /** Atomic: insert the lease and, when a blob row already carries its declared hash, pin that + * blob — the returned lease then names it (the dedupe short-circuit). */ + beginUpload(lease: UploadLease): Promise; + deleteLease(uploadId: UploadId): Promise; + /** Atomic: the blob row (if new), the attachment with its `original` variant, and the lease + * pointed at the attachment. */ + commitAttachment(commit: AttachmentCommit): Promise; + /** Atomic reaper: expired leases first; then attachments, then blobs, that nothing roots and + * that predate the grace window. Returns the blob ids whose bytes the caller must delete. */ + sweep(window: AttachmentSweepWindow): Promise; +} + +export class InMemoryAttachmentStore implements AttachmentStore { + private readonly blobs = new Map(); + private readonly attachments = new Map(); + private readonly leases = new Map(); + + /** `roots` lists every attachment id a prompt or session resource currently references. */ + constructor(private readonly roots: () => Iterable = () => []) {} + + getAttachment(attachmentId: AttachmentId): Promise { + const attachment = this.attachments.get(attachmentId); + return Promise.resolve(attachment && structuredClone(attachment)); + } + + listAttachments(attachmentIds: readonly AttachmentId[]): Promise { + const found: StoredAttachment[] = []; + for (let i = 0, len = attachmentIds.length; i < len; i++) { + const attachment = this.attachments.get(attachmentIds[i]); + if (attachment) found.push(structuredClone(attachment)); + } + return Promise.resolve(found); + } + + getBlob(blobId: BlobId): Promise { + const blob = this.blobs.get(blobId); + return Promise.resolve(blob && structuredClone(blob)); + } + + getLease(uploadId: UploadId): Promise { + const lease = this.leases.get(uploadId); + return Promise.resolve(lease && structuredClone(lease)); + } + + beginUpload(lease: UploadLease): Promise { + if (this.leases.has(lease.uploadId)) { + return Promise.reject(new Error(`Upload lease already exists: ${lease.uploadId}`)); + } + const blobId = blobIdFromSha256(lease.declaredSha256); + const pinned = this.blobs.has(blobId) ? { ...lease, blobId } : lease; + this.leases.set(lease.uploadId, structuredClone(pinned)); + return Promise.resolve(structuredClone(pinned)); + } + + deleteLease(uploadId: UploadId): Promise { + this.leases.delete(uploadId); + return Promise.resolve(); + } + + commitAttachment({ attachment, blob, uploadId }: AttachmentCommit): Promise { + if (!this.blobs.has(blob.blobId)) this.blobs.set(blob.blobId, structuredClone(blob)); + this.attachments.set(attachment.attachmentId, { + ...structuredClone(attachment), + blobId: blob.blobId, + }); + const lease = uploadId === undefined ? undefined : this.leases.get(uploadId); + if (lease) { + this.leases.set(lease.uploadId, { + ...lease, + blobId: blob.blobId, + attachmentId: attachment.attachmentId, + }); + } + return Promise.resolve(); + } + + sweep({ graceBefore, now }: AttachmentSweepWindow): Promise { + for (const [uploadId, lease] of this.leases) { + if (lease.expiresAt <= now) this.leases.delete(uploadId); + } + const rooted = new Set(this.roots()); + const pinnedBlobs = new Set(); + for (const lease of this.leases.values()) { + if (lease.attachmentId !== undefined) rooted.add(lease.attachmentId); + if (lease.blobId !== undefined) pinnedBlobs.add(lease.blobId); + } + for (const [attachmentId, attachment] of this.attachments) { + if (attachment.createdAt < graceBefore && !rooted.has(attachmentId)) { + this.attachments.delete(attachmentId); + } + } + for (const attachment of this.attachments.values()) pinnedBlobs.add(attachment.blobId); + const doomed: BlobId[] = []; + for (const [blobId, blob] of this.blobs) { + if (blob.createdAt < graceBefore && !pinnedBlobs.has(blobId)) { + this.blobs.delete(blobId); + doomed.push(blobId); + } + } + return Promise.resolve(doomed); + } +} diff --git a/packages/host/engine/src/attachment/gc.ts b/packages/host/engine/src/attachment/gc.ts new file mode 100644 index 000000000..893903ec7 --- /dev/null +++ b/packages/host/engine/src/attachment/gc.ts @@ -0,0 +1,70 @@ +import type { BlobId } from '@linkcode/schema'; +import { Effect, Schedule } from 'effect'; +import type { AttachmentStore } from './attachment-store'; +import type { BlobStore } from './blob-store'; + +/** A draft's upload outlives any composer session, not a forgotten one. */ +export const UPLOAD_LEASE_TTL_MS = 24 * 60 * 60 * 1000; +/** Rows younger than this are never collected: their root may be one transaction away. */ +export const ATTACHMENT_GC_GRACE_MS = 60 * 60 * 1000; +export const ATTACHMENT_GC_INTERVAL_MS = 60 * 60 * 1000; + +export interface AttachmentGcReport { + readonly removedBlobs: BlobId[]; +} + +/** Reference-rooted collection: the store's transaction decides, then bytes follow. */ +export class AttachmentGc { + constructor( + private readonly store: AttachmentStore, + private readonly blobs: BlobStore, + private readonly clock: () => number = Date.now, + ) {} + + /** A failed unlink leaves an orphan file the next boot sweep removes. */ + async sweep(): Promise { + const now = this.clock(); + const removedBlobs = await this.store.sweep({ + now, + graceBefore: now - ATTACHMENT_GC_GRACE_MS, + }); + for (let i = 0, len = removedBlobs.length; i < len; i++) { + // eslint-disable-next-line no-await-in-loop -- sequential unlinks of a short list + await this.blobs.delete(removedBlobs[i]); + } + return { removedBlobs }; + } + + /** + * Run before requests are accepted: no upload survives a restart, so staging goes wholesale; + * then a sweep; then bytes with no blob row (a crash between publish and the row insert). + * Safe only while no commit can be publishing — that is what the boot ordering guarantees. + */ + async bootSweep(): Promise { + await this.blobs.purgeStaging(); + const report = await this.sweep(); + const onDisk = await this.blobs.list(); + const orphans: BlobId[] = []; + for (let i = 0, len = onDisk.length; i < len; i++) { + const blobId = onDisk[i]; + // eslint-disable-next-line no-await-in-loop -- one row lookup per file on disk + if (await this.store.getBlob(blobId)) continue; + // eslint-disable-next-line no-await-in-loop -- same + await this.blobs.delete(blobId); + orphans.push(blobId); + } + return { removedBlobs: [...report.removedBlobs, ...orphans] }; + } + + /** One sweep per interval until interrupted; a failed sweep is logged and the cadence goes on. */ + cadence(): Effect.Effect { + const sweep = Effect.tryPromise({ try: () => this.sweep(), catch: (error) => error }).pipe( + Effect.catch((error) => Effect.logError('Attachment GC sweep failed', error)), + Effect.asVoid, + ); + return Effect.sleep(ATTACHMENT_GC_INTERVAL_MS).pipe( + Effect.andThen(sweep.pipe(Effect.repeat(Schedule.spaced(ATTACHMENT_GC_INTERVAL_MS)))), + Effect.asVoid, + ); + } +} diff --git a/packages/host/engine/src/conversation/conversation-store.ts b/packages/host/engine/src/conversation/conversation-store.ts index 1bd500951..d70cce86d 100644 --- a/packages/host/engine/src/conversation/conversation-store.ts +++ b/packages/host/engine/src/conversation/conversation-store.ts @@ -1,4 +1,5 @@ import type { + AttachmentId, ConversationOperation, ConversationTurn, OperationId, @@ -65,6 +66,21 @@ export class InMemoryConversationStore implements ConversationStore { private readonly bindings = new Map(); private readonly operations = new Map(); + /** GC roots for the in-memory attachment store: every attachment a persisted prompt references. */ + referencedAttachmentIds(): AttachmentId[] { + const ids: AttachmentId[] = []; + for (const prompt of this.prompts.values()) { + for (let i = 0, len = prompt.contextAttachmentIds.length; i < len; i++) { + ids.push(prompt.contextAttachmentIds[i]); + } + for (let i = 0, len = prompt.blocks.length; i < len; i++) { + const block = prompt.blocks[i]; + if (block.type === 'attachment_ref') ids.push(block.attachmentId); + } + } + return ids; + } + listTurns(sessionId: SessionId): Promise { const turns = []; for (const turn of this.turns.values()) { diff --git a/packages/host/engine/src/deps.ts b/packages/host/engine/src/deps.ts index bb4b1803d..ce262d64b 100644 --- a/packages/host/engine/src/deps.ts +++ b/packages/host/engine/src/deps.ts @@ -5,6 +5,8 @@ import type { ModelProbe } from './agent/model-probe'; import type { ProviderConfigStore } from './agent/provider-config'; import type { TranslatorService } from './agent/translator'; import type { AssetService } from './asset/service'; +import type { AttachmentStore } from './attachment/attachment-store'; +import type { BlobStore } from './attachment/blob-store'; import type { LoopStore, ScheduleStore } from './automation'; import type { ConversationStore } from './conversation/conversation-store'; import type { GitService } from './git/git-service'; @@ -28,6 +30,12 @@ export interface EngineDeps { /** Durable turn-tree/prompt storage. The daemon injects the single-connection SQLite store its * multi-table transactions require; the in-memory default keeps bare engines and tests free. */ conversationStore?: ConversationStore; + /** Attachment metadata and upload leases. Inject it together with `conversationStore` and + * `resourceStore`: the reaper's roots live in those tables, and the in-memory default can only + * read roots from the in-memory stores. */ + attachmentStore?: AttachmentStore; + /** Attachment bytes; defaults to a filesystem store under `stateDir`. */ + blobStore?: BlobStore; resourceStore?: ResourceStore; /** Daemon profile state directory containing managed resource bytes. */ stateDir?: string; diff --git a/packages/host/engine/src/engine.ts b/packages/host/engine/src/engine.ts index 20cfcf498..fd699e755 100644 --- a/packages/host/engine/src/engine.ts +++ b/packages/host/engine/src/engine.ts @@ -1,3 +1,7 @@ +import { mkdtempSync } from 'node:fs'; +import { rm } from 'node:fs/promises'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; import type { PluginDiscoveryOptions } from '@linkcode/agent-adapter'; import { createAdapter, createPluginProviderAdapter } from '@linkcode/agent-adapter'; import type { WorkspaceRecord } from '@linkcode/schema'; @@ -12,6 +16,9 @@ import { InMemoryProviderConfigStore } from './agent/provider-config'; import { AgentRequestHandler } from './agent/request-handler'; import { AgentRuntimeService } from './agent/runtime-service'; import { ManagedAssetService } from './asset/service'; +import { InMemoryAttachmentStore } from './attachment/attachment-store'; +import { FsBlobStore } from './attachment/blob-store'; +import { AttachmentGc } from './attachment/gc'; import { InMemoryLoopStore, InMemoryScheduleStore, @@ -100,13 +107,16 @@ export const createEngineRuntime = Effect.fn('Engine.create')(function* ( ); const routes = deps.previewRoutes ?? new PreviewRouteRegistry(); const fileHost = new FileHostService(routes); - const resources = new ResourceService( - transport, - deps.resourceStore ?? new InMemoryResourceStore(), - records, - deps.stateDir, - fileHost, - ); + // A bare engine gets its own state dir: blob GC in a shared tmp path would reap another + // engine's bytes, since each in-memory store only knows its own roots. + const stateDir = + deps.stateDir ?? + (yield* Effect.acquireRelease( + Effect.sync(() => mkdtempSync(join(tmpdir(), 'linkcode-engine-'))), + (dir) => Effect.promise(() => rm(dir, { recursive: true, force: true })), + )); + const resourceStore = deps.resourceStore ?? new InMemoryResourceStore(); + const resources = new ResourceService(transport, resourceStore, records, stateDir, fileHost); const plugins = new PluginService(deps.pluginFactory ?? createPluginProviderAdapter); const translator = deps.translator; const startOptions = new SessionStartOptionsResolver( @@ -150,6 +160,18 @@ export const createEngineRuntime = Effect.fn('Engine.create')(function* ( const simulators = deps.simulators; const browserBroker = new BrowserBrokerService(transport); const conversationStore = deps.conversationStore ?? new InMemoryConversationStore(); + const blobStore = deps.blobStore ?? new FsBlobStore(join(stateDir, 'blobs')); + const attachmentStore = + deps.attachmentStore ?? + new InMemoryAttachmentStore(() => [ + ...(conversationStore instanceof InMemoryConversationStore + ? conversationStore.referencedAttachmentIds() + : []), + ...(resourceStore instanceof InMemoryResourceStore + ? resourceStore.referencedAttachmentIds() + : []), + ]); + const attachmentGc = new AttachmentGc(attachmentStore, blobStore); const conversationTurns = new ConversationTurnService( conversationStore, records, @@ -321,6 +343,15 @@ export const createEngineRuntime = Effect.fn('Engine.create')(function* ( // Before requests are accepted: open operations and non-terminal turns cannot outlive the // adapters that ran them, and a retried operation must replay a terminal result. yield* conversationTurns.recover(Array.from(records.values(), ({ sessionId }) => sessionId)); + // Before the transport connects: the boot sweep deletes bytes without a row, which is only + // safe while no upload can be publishing. GC never takes the boot down. + yield* tryOperation( + 'filesystem', + 'attachments.boot-sweep', + 'Failed to sweep the attachment store', + () => attachmentGc.bootSweep(), + ).pipe(Effect.catch((error) => Effect.logWarning('Attachment boot sweep failed', error))); + runTask(attachmentGc.cadence()); yield* worktrees.start(new Set(Array.from(records.values(), ({ sessionId }) => sessionId))); yield* tryOperation('store', 'workspaces.load', 'Failed to load workspaces', () => workspaces.start(), diff --git a/packages/host/engine/src/index.ts b/packages/host/engine/src/index.ts index 31647962b..ed92743e7 100644 --- a/packages/host/engine/src/index.ts +++ b/packages/host/engine/src/index.ts @@ -7,6 +7,13 @@ export type { ProviderConfigStore } from './agent/provider-config'; export type { TranslatorService, TranslatorUpstream } from './agent/translator'; export type { AssetService } from './asset/service'; +export { + type AttachmentCommit, + type AttachmentStore, + type AttachmentSweepWindow, + InMemoryAttachmentStore, + type StoredAttachment, +} from './attachment/attachment-store'; export { type BlobStage, type BlobStore, FsBlobStore } from './attachment/blob-store'; export type { LoopStore, ScheduleStore } from './automation'; export { diff --git a/packages/host/engine/src/resource/resource-store.ts b/packages/host/engine/src/resource/resource-store.ts index 86fb19a77..c05a41da1 100644 --- a/packages/host/engine/src/resource/resource-store.ts +++ b/packages/host/engine/src/resource/resource-store.ts @@ -1,4 +1,4 @@ -import type { SessionId, SessionResource, SessionResourceId } from '@linkcode/schema'; +import type { AttachmentId, SessionId, SessionResource, SessionResourceId } from '@linkcode/schema'; export interface ResourceStore { list(sessionId: SessionId): Promise; @@ -16,6 +16,15 @@ export class InMemoryResourceStore implements ResourceStore { private readonly resources = new Map(); private readonly locatorKeys = new Map(); + /** GC roots for the in-memory attachment store: every attachment a resource is backed by. */ + referencedAttachmentIds(): AttachmentId[] { + const ids: AttachmentId[] = []; + for (const resource of this.resources.values()) { + if (resource.attachmentId !== undefined) ids.push(resource.attachmentId); + } + return ids; + } + list(sessionId: SessionId): Promise { const resources: SessionResource[] = []; for (const resource of this.resources.values()) { From 3ca914a8b5e71e5271ef9e54ddba91ca5b237393 Mon Sep 17 00:00:00 2001 From: Zerlight Wu Date: Thu, 3 Sep 2026 14:31:01 +0800 Subject: [PATCH 3/5] feat(daemon): attachment tables, upload leases, and lease claims on the graph connection --- .../daemon/drizzle/0014_elite_mister_fear.sql | 41 + apps/daemon/drizzle/meta/0014_snapshot.json | 1571 +++++++++++++++++ apps/daemon/drizzle/meta/_journal.json | 7 + .../src/__tests__/attachment-store.test.ts | 212 +++ .../src/__tests__/resource-store.test.ts | 3 + apps/daemon/src/attachment-store.ts | 230 +++ apps/daemon/src/conversation-store.ts | 3 + apps/daemon/src/db/schema.ts | 62 + apps/daemon/src/index.ts | 2 + apps/daemon/src/resource-store.ts | 2 + 10 files changed, 2133 insertions(+) create mode 100644 apps/daemon/drizzle/0014_elite_mister_fear.sql create mode 100644 apps/daemon/drizzle/meta/0014_snapshot.json create mode 100644 apps/daemon/src/__tests__/attachment-store.test.ts create mode 100644 apps/daemon/src/attachment-store.ts diff --git a/apps/daemon/drizzle/0014_elite_mister_fear.sql b/apps/daemon/drizzle/0014_elite_mister_fear.sql new file mode 100644 index 000000000..9a4f7dff6 --- /dev/null +++ b/apps/daemon/drizzle/0014_elite_mister_fear.sql @@ -0,0 +1,41 @@ +CREATE TABLE `attachment_blobs` ( + `attachment_id` text NOT NULL, + `variant` text NOT NULL, + `blob_id` text NOT NULL, + PRIMARY KEY(`attachment_id`, `variant`), + FOREIGN KEY (`attachment_id`) REFERENCES `attachments`(`attachment_id`) ON UPDATE no action ON DELETE cascade, + FOREIGN KEY (`blob_id`) REFERENCES `blobs`(`blob_id`) ON UPDATE no action ON DELETE no action +); +--> statement-breakpoint +CREATE INDEX `attachment_blobs_blob_idx` ON `attachment_blobs` (`blob_id`);--> statement-breakpoint +CREATE TABLE `attachments` ( + `attachment_id` text PRIMARY KEY NOT NULL, + `kind` text NOT NULL, + `name` text NOT NULL, + `mime_type` text NOT NULL, + `size_bytes` integer NOT NULL, + `metadata_json` text NOT NULL, + `created_at` integer NOT NULL +); +--> statement-breakpoint +CREATE TABLE `blobs` ( + `blob_id` text PRIMARY KEY NOT NULL, + `size_bytes` integer NOT NULL, + `created_at` integer NOT NULL +); +--> statement-breakpoint +CREATE TABLE `upload_leases` ( + `upload_id` text PRIMARY KEY NOT NULL, + `declared_sha256` text NOT NULL, + `declared_size` integer NOT NULL, + `name` text NOT NULL, + `mime_type` text, + `kind` text NOT NULL, + `blob_id` text, + `attachment_id` text, + `expires_at` integer NOT NULL, + `created_at` integer NOT NULL +); +--> statement-breakpoint +CREATE INDEX `upload_leases_expires_at_idx` ON `upload_leases` (`expires_at`);--> statement-breakpoint +ALTER TABLE `session_resources` ADD `attachment_id` text; \ No newline at end of file diff --git a/apps/daemon/drizzle/meta/0014_snapshot.json b/apps/daemon/drizzle/meta/0014_snapshot.json new file mode 100644 index 000000000..1cb26c39a --- /dev/null +++ b/apps/daemon/drizzle/meta/0014_snapshot.json @@ -0,0 +1,1571 @@ +{ + "version": "6", + "dialect": "sqlite", + "id": "a9ae9955-d7e4-45a9-a33c-59f8b342efd7", + "prevId": "882ec2b6-a29e-44e2-893e-b19adff08199", + "tables": { + "attachment_blobs": { + "name": "attachment_blobs", + "columns": { + "attachment_id": { + "name": "attachment_id", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "variant": { + "name": "variant", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "blob_id": { + "name": "blob_id", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + } + }, + "indexes": { + "attachment_blobs_blob_idx": { + "name": "attachment_blobs_blob_idx", + "columns": ["blob_id"], + "isUnique": false + } + }, + "foreignKeys": { + "attachment_blobs_attachment_id_attachments_attachment_id_fk": { + "name": "attachment_blobs_attachment_id_attachments_attachment_id_fk", + "tableFrom": "attachment_blobs", + "tableTo": "attachments", + "columnsFrom": ["attachment_id"], + "columnsTo": ["attachment_id"], + "onDelete": "cascade", + "onUpdate": "no action" + }, + "attachment_blobs_blob_id_blobs_blob_id_fk": { + "name": "attachment_blobs_blob_id_blobs_blob_id_fk", + "tableFrom": "attachment_blobs", + "tableTo": "blobs", + "columnsFrom": ["blob_id"], + "columnsTo": ["blob_id"], + "onDelete": "no action", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": { + "attachment_blobs_attachment_id_variant_pk": { + "columns": ["attachment_id", "variant"], + "name": "attachment_blobs_attachment_id_variant_pk" + } + }, + "uniqueConstraints": {}, + "checkConstraints": {} + }, + "attachments": { + "name": "attachments", + "columns": { + "attachment_id": { + "name": "attachment_id", + "type": "text", + "primaryKey": true, + "notNull": true, + "autoincrement": false + }, + "kind": { + "name": "kind", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "name": { + "name": "name", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "mime_type": { + "name": "mime_type", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "size_bytes": { + "name": "size_bytes", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "metadata_json": { + "name": "metadata_json", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "created_at": { + "name": "created_at", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false + } + }, + "indexes": {}, + "foreignKeys": {}, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "checkConstraints": {} + }, + "blobs": { + "name": "blobs", + "columns": { + "blob_id": { + "name": "blob_id", + "type": "text", + "primaryKey": true, + "notNull": true, + "autoincrement": false + }, + "size_bytes": { + "name": "size_bytes", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "created_at": { + "name": "created_at", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false + } + }, + "indexes": {}, + "foreignKeys": {}, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "checkConstraints": {} + }, + "conversation_operations": { + "name": "conversation_operations", + "columns": { + "operation_id": { + "name": "operation_id", + "type": "text", + "primaryKey": true, + "notNull": true, + "autoincrement": false + }, + "session_id": { + "name": "session_id", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "kind": { + "name": "kind", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "state": { + "name": "state", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "turn_id": { + "name": "turn_id", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "error_code": { + "name": "error_code", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "error_message": { + "name": "error_message", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "created_at": { + "name": "created_at", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "resolved_at": { + "name": "resolved_at", + "type": "integer", + "primaryKey": false, + "notNull": false, + "autoincrement": false + } + }, + "indexes": { + "conversation_operations_session_idx": { + "name": "conversation_operations_session_idx", + "columns": ["session_id"], + "isUnique": false + }, + "conversation_operations_open_session_unique": { + "name": "conversation_operations_open_session_unique", + "columns": ["session_id"], + "isUnique": true, + "where": "state = 'open'" + } + }, + "foreignKeys": { + "conversation_operations_session_id_sessions_session_id_fk": { + "name": "conversation_operations_session_id_sessions_session_id_fk", + "tableFrom": "conversation_operations", + "tableTo": "sessions", + "columnsFrom": ["session_id"], + "columnsTo": ["session_id"], + "onDelete": "cascade", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "checkConstraints": {} + }, + "conversation_turns": { + "name": "conversation_turns", + "columns": { + "turn_id": { + "name": "turn_id", + "type": "text", + "primaryKey": true, + "notNull": true, + "autoincrement": false + }, + "session_id": { + "name": "session_id", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "parent_turn_id": { + "name": "parent_turn_id", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "sibling_ordinal": { + "name": "sibling_ordinal", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "input_type": { + "name": "input_type", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "prompt_id": { + "name": "prompt_id", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "command_name": { + "name": "command_name", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "command_arguments": { + "name": "command_arguments", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "shell_command": { + "name": "shell_command", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "run_id": { + "name": "run_id", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "state": { + "name": "state", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "created_at": { + "name": "created_at", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false + } + }, + "indexes": { + "conversation_turns_session_idx": { + "name": "conversation_turns_session_idx", + "columns": ["session_id"], + "isUnique": false + }, + "conversation_turns_sibling_unique": { + "name": "conversation_turns_sibling_unique", + "columns": ["session_id", "parent_turn_id", "sibling_ordinal"], + "isUnique": true, + "where": "parent_turn_id IS NOT NULL" + }, + "conversation_turns_root_sibling_unique": { + "name": "conversation_turns_root_sibling_unique", + "columns": ["session_id", "sibling_ordinal"], + "isUnique": true, + "where": "parent_turn_id IS NULL" + } + }, + "foreignKeys": { + "conversation_turns_session_id_sessions_session_id_fk": { + "name": "conversation_turns_session_id_sessions_session_id_fk", + "tableFrom": "conversation_turns", + "tableTo": "sessions", + "columnsFrom": ["session_id"], + "columnsTo": ["session_id"], + "onDelete": "cascade", + "onUpdate": "no action" + }, + "conversation_turns_parent_turn_id_conversation_turns_turn_id_fk": { + "name": "conversation_turns_parent_turn_id_conversation_turns_turn_id_fk", + "tableFrom": "conversation_turns", + "tableTo": "conversation_turns", + "columnsFrom": ["parent_turn_id"], + "columnsTo": ["turn_id"], + "onDelete": "cascade", + "onUpdate": "no action" + }, + "conversation_turns_prompt_id_prompts_prompt_id_fk": { + "name": "conversation_turns_prompt_id_prompts_prompt_id_fk", + "tableFrom": "conversation_turns", + "tableTo": "prompts", + "columnsFrom": ["prompt_id"], + "columnsTo": ["prompt_id"], + "onDelete": "no action", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "checkConstraints": {} + }, + "loop_iterations": { + "name": "loop_iterations", + "columns": { + "loop_id": { + "name": "loop_id", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "index": { + "name": "index", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "status": { + "name": "status", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "worker_session_id": { + "name": "worker_session_id", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "verifier_session_id": { + "name": "verifier_session_id", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "checks_json": { + "name": "checks_json", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "verdict_json": { + "name": "verdict_json", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "error": { + "name": "error", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "started_at": { + "name": "started_at", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "ended_at": { + "name": "ended_at", + "type": "integer", + "primaryKey": false, + "notNull": false, + "autoincrement": false + } + }, + "indexes": {}, + "foreignKeys": { + "loop_iterations_loop_id_loops_loop_id_fk": { + "name": "loop_iterations_loop_id_loops_loop_id_fk", + "tableFrom": "loop_iterations", + "tableTo": "loops", + "columnsFrom": ["loop_id"], + "columnsTo": ["loop_id"], + "onDelete": "cascade", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": { + "loop_iterations_loop_id_index_pk": { + "columns": ["loop_id", "index"], + "name": "loop_iterations_loop_id_index_pk" + } + }, + "uniqueConstraints": {}, + "checkConstraints": {} + }, + "loops": { + "name": "loops", + "columns": { + "loop_id": { + "name": "loop_id", + "type": "text", + "primaryKey": true, + "notNull": true, + "autoincrement": false + }, + "spec_json": { + "name": "spec_json", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "status": { + "name": "status", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "iteration_count": { + "name": "iteration_count", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false, + "default": 0 + }, + "error": { + "name": "error", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "summary": { + "name": "summary", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "started_at": { + "name": "started_at", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "updated_at": { + "name": "updated_at", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "ended_at": { + "name": "ended_at", + "type": "integer", + "primaryKey": false, + "notNull": false, + "autoincrement": false + } + }, + "indexes": {}, + "foreignKeys": {}, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "checkConstraints": {} + }, + "prompt_attachment_refs": { + "name": "prompt_attachment_refs", + "columns": { + "prompt_id": { + "name": "prompt_id", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "attachment_id": { + "name": "attachment_id", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + } + }, + "indexes": {}, + "foreignKeys": { + "prompt_attachment_refs_prompt_id_prompts_prompt_id_fk": { + "name": "prompt_attachment_refs_prompt_id_prompts_prompt_id_fk", + "tableFrom": "prompt_attachment_refs", + "tableTo": "prompts", + "columnsFrom": ["prompt_id"], + "columnsTo": ["prompt_id"], + "onDelete": "cascade", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": { + "prompt_attachment_refs_prompt_id_attachment_id_pk": { + "columns": ["prompt_id", "attachment_id"], + "name": "prompt_attachment_refs_prompt_id_attachment_id_pk" + } + }, + "uniqueConstraints": {}, + "checkConstraints": {} + }, + "prompts": { + "name": "prompts", + "columns": { + "prompt_id": { + "name": "prompt_id", + "type": "text", + "primaryKey": true, + "notNull": true, + "autoincrement": false + }, + "blocks_json": { + "name": "blocks_json", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "context_attachment_ids_json": { + "name": "context_attachment_ids_json", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "created_at": { + "name": "created_at", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false + } + }, + "indexes": {}, + "foreignKeys": {}, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "checkConstraints": {} + }, + "provider_turn_bindings": { + "name": "provider_turn_bindings", + "columns": { + "turn_id": { + "name": "turn_id", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "run_id": { + "name": "run_id", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "history_id": { + "name": "history_id", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "checkpoint": { + "name": "checkpoint", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "captured_from": { + "name": "captured_from", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + } + }, + "indexes": {}, + "foreignKeys": { + "provider_turn_bindings_turn_id_conversation_turns_turn_id_fk": { + "name": "provider_turn_bindings_turn_id_conversation_turns_turn_id_fk", + "tableFrom": "provider_turn_bindings", + "tableTo": "conversation_turns", + "columnsFrom": ["turn_id"], + "columnsTo": ["turn_id"], + "onDelete": "cascade", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": { + "provider_turn_bindings_turn_id_history_id_pk": { + "columns": ["turn_id", "history_id"], + "name": "provider_turn_bindings_turn_id_history_id_pk" + } + }, + "uniqueConstraints": {}, + "checkConstraints": {} + }, + "schedule_runs": { + "name": "schedule_runs", + "columns": { + "run_id": { + "name": "run_id", + "type": "text", + "primaryKey": true, + "notNull": true, + "autoincrement": false + }, + "schedule_id": { + "name": "schedule_id", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "status": { + "name": "status", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "trigger": { + "name": "trigger", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "session_id": { + "name": "session_id", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "error": { + "name": "error", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "summary": { + "name": "summary", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "started_at": { + "name": "started_at", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "ended_at": { + "name": "ended_at", + "type": "integer", + "primaryKey": false, + "notNull": false, + "autoincrement": false + } + }, + "indexes": { + "schedule_runs_schedule_started_idx": { + "name": "schedule_runs_schedule_started_idx", + "columns": ["schedule_id", "started_at"], + "isUnique": false + } + }, + "foreignKeys": { + "schedule_runs_schedule_id_schedules_schedule_id_fk": { + "name": "schedule_runs_schedule_id_schedules_schedule_id_fk", + "tableFrom": "schedule_runs", + "tableTo": "schedules", + "columnsFrom": ["schedule_id"], + "columnsTo": ["schedule_id"], + "onDelete": "cascade", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "checkConstraints": {} + }, + "schedules": { + "name": "schedules", + "columns": { + "schedule_id": { + "name": "schedule_id", + "type": "text", + "primaryKey": true, + "notNull": true, + "autoincrement": false + }, + "name": { + "name": "name", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "prompt": { + "name": "prompt", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "cadence_type": { + "name": "cadence_type", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "cron_expression": { + "name": "cron_expression", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "cron_timezone": { + "name": "cron_timezone", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "interval_ms": { + "name": "interval_ms", + "type": "integer", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "target_type": { + "name": "target_type", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "target_session_id": { + "name": "target_session_id", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "target_config_json": { + "name": "target_config_json", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "status": { + "name": "status", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "completed_reason": { + "name": "completed_reason", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "misfire_policy": { + "name": "misfire_policy", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "next_run_at": { + "name": "next_run_at", + "type": "integer", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "last_run_at": { + "name": "last_run_at", + "type": "integer", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "run_count": { + "name": "run_count", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false, + "default": 0 + }, + "max_runs": { + "name": "max_runs", + "type": "integer", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "expires_at": { + "name": "expires_at", + "type": "integer", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "created_at": { + "name": "created_at", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "updated_at": { + "name": "updated_at", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false + } + }, + "indexes": { + "schedules_next_run_at_idx": { + "name": "schedules_next_run_at_idx", + "columns": ["next_run_at"], + "isUnique": false + } + }, + "foreignKeys": {}, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "checkConstraints": {} + }, + "session_resources": { + "name": "session_resources", + "columns": { + "resource_id": { + "name": "resource_id", + "type": "text", + "primaryKey": true, + "notNull": true, + "autoincrement": false + }, + "session_id": { + "name": "session_id", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "direction": { + "name": "direction", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "name": { + "name": "name", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "kind": { + "name": "kind", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "status": { + "name": "status", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "locator_type": { + "name": "locator_type", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "locator": { + "name": "locator", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "normalized_locator_key": { + "name": "normalized_locator_key", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "attachment_id": { + "name": "attachment_id", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "mime_type": { + "name": "mime_type", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "size_bytes": { + "name": "size_bytes", + "type": "integer", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "error": { + "name": "error", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "created_at": { + "name": "created_at", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "updated_at": { + "name": "updated_at", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false + } + }, + "indexes": { + "session_resources_session_idx": { + "name": "session_resources_session_idx", + "columns": ["session_id"], + "isUnique": false + }, + "session_resources_locator_idx": { + "name": "session_resources_locator_idx", + "columns": ["session_id", "normalized_locator_key"], + "isUnique": true + } + }, + "foreignKeys": { + "session_resources_session_id_sessions_session_id_fk": { + "name": "session_resources_session_id_sessions_session_id_fk", + "tableFrom": "session_resources", + "tableTo": "sessions", + "columnsFrom": ["session_id"], + "columnsTo": ["session_id"], + "onDelete": "cascade", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "checkConstraints": {} + }, + "session_runs": { + "name": "session_runs", + "columns": { + "id": { + "name": "id", + "type": "integer", + "primaryKey": true, + "notNull": true, + "autoincrement": true + }, + "session_id": { + "name": "session_id", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "seq": { + "name": "seq", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "run_id": { + "name": "run_id", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "base_turn_id": { + "name": "base_turn_id", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "history_id": { + "name": "history_id", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "account_id": { + "name": "account_id", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "model": { + "name": "model", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "effort": { + "name": "effort", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "approval_policy_id": { + "name": "approval_policy_id", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "started_at": { + "name": "started_at", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "ended_at": { + "name": "ended_at", + "type": "integer", + "primaryKey": false, + "notNull": false, + "autoincrement": false + } + }, + "indexes": { + "session_runs_session_id_idx": { + "name": "session_runs_session_id_idx", + "columns": ["session_id"], + "isUnique": false + }, + "session_runs_run_id_unique": { + "name": "session_runs_run_id_unique", + "columns": ["run_id"], + "isUnique": true + } + }, + "foreignKeys": { + "session_runs_session_id_sessions_session_id_fk": { + "name": "session_runs_session_id_sessions_session_id_fk", + "tableFrom": "session_runs", + "tableTo": "sessions", + "columnsFrom": ["session_id"], + "columnsTo": ["session_id"], + "onDelete": "cascade", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "checkConstraints": {} + }, + "sessions": { + "name": "sessions", + "columns": { + "session_id": { + "name": "session_id", + "type": "text", + "primaryKey": true, + "notNull": true, + "autoincrement": false + }, + "kind": { + "name": "kind", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "cwd": { + "name": "cwd", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "title": { + "name": "title", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "origin_type": { + "name": "origin_type", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "origin_history_id": { + "name": "origin_history_id", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "origin_imported_at": { + "name": "origin_imported_at", + "type": "integer", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "origin_source_session_id": { + "name": "origin_source_session_id", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "origin_source_turn_id": { + "name": "origin_source_turn_id", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "origin_forked_at": { + "name": "origin_forked_at", + "type": "integer", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "created_via": { + "name": "created_via", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "automation_kind": { + "name": "automation_kind", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "automation_id": { + "name": "automation_id", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "active_leaf_turn_id": { + "name": "active_leaf_turn_id", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "graph_revision": { + "name": "graph_revision", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false, + "default": 0 + }, + "event_epoch": { + "name": "event_epoch", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false, + "default": 0 + }, + "created_at": { + "name": "created_at", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "updated_at": { + "name": "updated_at", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false + } + }, + "indexes": { + "sessions_updated_at_idx": { + "name": "sessions_updated_at_idx", + "columns": ["updated_at"], + "isUnique": false + } + }, + "foreignKeys": {}, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "checkConstraints": {} + }, + "upload_leases": { + "name": "upload_leases", + "columns": { + "upload_id": { + "name": "upload_id", + "type": "text", + "primaryKey": true, + "notNull": true, + "autoincrement": false + }, + "declared_sha256": { + "name": "declared_sha256", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "declared_size": { + "name": "declared_size", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "name": { + "name": "name", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "mime_type": { + "name": "mime_type", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "kind": { + "name": "kind", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "blob_id": { + "name": "blob_id", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "attachment_id": { + "name": "attachment_id", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "expires_at": { + "name": "expires_at", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "created_at": { + "name": "created_at", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false + } + }, + "indexes": { + "upload_leases_expires_at_idx": { + "name": "upload_leases_expires_at_idx", + "columns": ["expires_at"], + "isUnique": false + } + }, + "foreignKeys": {}, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "checkConstraints": {} + }, + "workspaces": { + "name": "workspaces", + "columns": { + "workspace_id": { + "name": "workspace_id", + "type": "text", + "primaryKey": true, + "notNull": true, + "autoincrement": false + }, + "cwd": { + "name": "cwd", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "name": { + "name": "name", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "kind": { + "name": "kind", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false, + "default": "'project'" + }, + "parent_workspace_id": { + "name": "parent_workspace_id", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "created_at": { + "name": "created_at", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "last_used_at": { + "name": "last_used_at", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false + } + }, + "indexes": { + "workspaces_cwd_unique": { + "name": "workspaces_cwd_unique", + "columns": ["cwd"], + "isUnique": true + }, + "workspaces_last_used_at_idx": { + "name": "workspaces_last_used_at_idx", + "columns": ["last_used_at"], + "isUnique": false + } + }, + "foreignKeys": {}, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "checkConstraints": {} + }, + "worktrees": { + "name": "worktrees", + "columns": { + "worktree_path": { + "name": "worktree_path", + "type": "text", + "primaryKey": true, + "notNull": true, + "autoincrement": false + }, + "repo_root": { + "name": "repo_root", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "branch": { + "name": "branch", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "session_id": { + "name": "session_id", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "created_at": { + "name": "created_at", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "state": { + "name": "state", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + } + }, + "indexes": { + "worktrees_repo_root_branch_unique": { + "name": "worktrees_repo_root_branch_unique", + "columns": ["repo_root", "branch"], + "isUnique": true + }, + "worktrees_session_id_unique": { + "name": "worktrees_session_id_unique", + "columns": ["session_id"], + "isUnique": true + } + }, + "foreignKeys": {}, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "checkConstraints": {} + } + }, + "views": {}, + "enums": {}, + "_meta": { + "schemas": {}, + "tables": {}, + "columns": {} + }, + "internal": { + "indexes": {} + } +} diff --git a/apps/daemon/drizzle/meta/_journal.json b/apps/daemon/drizzle/meta/_journal.json index d70f3890e..14611f1c9 100644 --- a/apps/daemon/drizzle/meta/_journal.json +++ b/apps/daemon/drizzle/meta/_journal.json @@ -99,6 +99,13 @@ "when": 1788261880957, "tag": "0013_famous_randall", "breakpoints": true + }, + { + "idx": 14, + "version": "6", + "when": 1788416689309, + "tag": "0014_elite_mister_fear", + "breakpoints": true } ] } diff --git a/apps/daemon/src/__tests__/attachment-store.test.ts b/apps/daemon/src/__tests__/attachment-store.test.ts new file mode 100644 index 000000000..b5b876760 --- /dev/null +++ b/apps/daemon/src/__tests__/attachment-store.test.ts @@ -0,0 +1,212 @@ +import { createHash } from 'node:crypto'; +import { mkdtemp, rm } from 'node:fs/promises'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; +import type { AttachmentStore } from '@linkcode/engine'; +import type { AttachmentRecord, BlobRecord, UploadLease } from '@linkcode/schema'; +import { + AttachmentIdSchema, + AttachmentRecordSchema, + blobIdFromSha256, + ConversationOperationSchema, + ConversationTurnSchema, + PromptRecordSchema, + SessionRecordSchema, + SessionResourceSchema, + UploadIdSchema, + UploadLeaseSchema, +} from '@linkcode/schema'; +import { afterEach, describe, expect, it } from 'vitest'; +import { createAttachmentStore } from '../attachment-store'; +import { createConversationStore } from '../conversation-store'; +import type { DaemonDatabase } from '../db/database'; +import { openDaemonDatabase } from '../db/database'; +import { createResourceStore } from '../resource-store'; +import { createSessionStore } from '../session-store'; + +const temporaryDirectories: string[] = []; +const openDatabases = new Set(); + +afterEach(async () => { + for (const database of openDatabases) database.close(); + openDatabases.clear(); + await Promise.all( + temporaryDirectories.splice(0).map((path) => rm(path, { recursive: true, force: true })), + ); +}); + +const GRACE = 60 * 60 * 1000; +const NOW = 1_000_000; + +function sha256(text: string): string { + return createHash('sha256').update(text).digest('hex'); +} + +function blob(text: string, createdAt = NOW): BlobRecord { + return { blobId: blobIdFromSha256(sha256(text)), sizeBytes: text.length, createdAt }; +} + +function attachment(id: string, createdAt = NOW): AttachmentRecord { + return AttachmentRecordSchema.parse({ + attachmentId: id, + kind: 'image', + name: `${id}.png`, + mimeType: 'image/png', + sizeBytes: 3, + metadata: { width: 2, height: 1 }, + createdAt, + }); +} + +function lease(uploadId: string, text: string): UploadLease { + return UploadLeaseSchema.parse({ + uploadId, + declaredSha256: sha256(text), + declaredSize: text.length, + name: 'draft.png', + mimeType: 'image/png', + kind: 'image', + expiresAt: NOW + 24 * 60 * 60 * 1000, + createdAt: NOW, + }); +} + +async function fixture(): Promise<{ + readonly database: DaemonDatabase; + readonly path: string; + readonly store: AttachmentStore; +}> { + const directory = await mkdtemp(join(tmpdir(), 'linkcode-attachment-store-')); + temporaryDirectories.push(directory); + const path = join(directory, 'daemon.db'); + const database = openDaemonDatabase(path); + openDatabases.add(database); + await createSessionStore(database.client).save( + SessionRecordSchema.parse({ + sessionId: 's-1', + kind: 'claude-code', + cwd: '/repo', + origin: { type: 'created' }, + createdAt: 1, + updatedAt: 1, + runs: [], + }), + ); + return { database, path, store: createAttachmentStore(database.client) }; +} + +describe('SQLite attachment store', () => { + it('round-trips attachments, blobs, and leases through a fresh store instance', async () => { + const { database, store } = await fixture(); + const begun = await store.beginUpload(lease('up-1', 'abc')); + expect(begun.blobId).toBeUndefined(); + await store.commitAttachment({ + blob: blob('abc'), + attachment: attachment('att-1'), + uploadId: UploadIdSchema.parse('up-1'), + }); + + const reopened = createAttachmentStore(database.client); + expect(await reopened.getAttachment(AttachmentIdSchema.parse('att-1'))).toEqual({ + ...attachment('att-1'), + blobId: blob('abc').blobId, + }); + expect(await reopened.getBlob(blob('abc').blobId)).toEqual(blob('abc')); + expect(await reopened.getLease(UploadIdSchema.parse('up-1'))).toEqual({ + ...lease('up-1', 'abc'), + blobId: blob('abc').blobId, + attachmentId: 'att-1', + }); + expect(await reopened.listAttachments([AttachmentIdSchema.parse('missing')])).toEqual([]); + + const pinned = await reopened.beginUpload(lease('up-2', 'abc')); + expect(pinned.blobId).toBe(blob('abc').blobId); + await expect(async () => reopened.beginUpload(lease('up-2', 'abc'))).rejects.toThrow('UNIQUE'); + await reopened.deleteLease(UploadIdSchema.parse('up-2')); + expect(await reopened.getLease(UploadIdSchema.parse('up-2'))).toBeUndefined(); + }); + + it('reaps only unrooted rows past the grace window and claims leases on prompt persist', async () => { + const { database, path, store } = await fixture(); + const conversations = createConversationStore(database.client); + const resources = createResourceStore(path); + const shared = blob('shared'); + const leasedOnly = blob('leased'); + const viaResource = blob('resource'); + + await store.beginUpload(lease('up-prompt', 'shared')); + await store.commitAttachment({ + blob: shared, + attachment: attachment('att-prompt'), + uploadId: UploadIdSchema.parse('up-prompt'), + }); + await store.commitAttachment({ blob: shared, attachment: attachment('att-stray') }); + await store.beginUpload(lease('up-leased', 'leased')); + await store.commitAttachment({ + blob: leasedOnly, + attachment: attachment('att-leased'), + uploadId: UploadIdSchema.parse('up-leased'), + }); + await store.commitAttachment({ blob: viaResource, attachment: attachment('att-resource') }); + + await conversations.persistTurnIntent({ + turn: ConversationTurnSchema.parse({ + turnId: 't-1', + sessionId: 's-1', + parentTurnId: null, + siblingOrdinal: 1, + input: { type: 'prompt', promptId: 'p-1' }, + runId: 'run-1', + state: 'preparing', + createdAt: 1, + }), + prompt: PromptRecordSchema.parse({ + promptId: 'p-1', + blocks: [{ type: 'attachment_ref', attachmentId: 'att-prompt' }], + contextAttachmentIds: [], + createdAt: 1, + }), + operation: ConversationOperationSchema.parse({ + operationId: 'op-1', + sessionId: 's-1', + kind: 'turn.submit', + state: 'open', + createdAt: 1, + }), + }); + // The prompt is the root now; the draft lease that pinned the attachment is released. + expect(await store.getLease(UploadIdSchema.parse('up-prompt'))).toBeUndefined(); + await resources.save( + SessionResourceSchema.parse({ + resourceId: 'resource-1', + sessionId: 's-1', + direction: 'source', + name: 'brief.png', + kind: 'image', + status: 'ready', + locator: { type: 'managed-file', path: '/state/blobs/x' }, + attachmentId: 'att-resource', + createdAt: 1, + updatedAt: 1, + }), + ); + + expect(await store.sweep({ now: NOW, graceBefore: NOW })).toEqual([]); + expect(await store.getAttachment(AttachmentIdSchema.parse('att-stray'))).toBeDefined(); + + expect(await store.sweep({ now: NOW + 1, graceBefore: NOW + GRACE })).toEqual([]); + expect(await store.getAttachment(AttachmentIdSchema.parse('att-stray'))).toBeUndefined(); + expect(await store.getBlob(shared.blobId)).toEqual(shared); + expect(await store.getAttachment(AttachmentIdSchema.parse('att-prompt'))).toBeDefined(); + expect(await store.getAttachment(AttachmentIdSchema.parse('att-leased'))).toBeDefined(); + expect(await store.getAttachment(AttachmentIdSchema.parse('att-resource'))).toBeDefined(); + + const afterExpiry = lease('up-leased', 'leased').expiresAt; + expect(await store.sweep({ now: afterExpiry, graceBefore: NOW + GRACE })).toEqual([ + leasedOnly.blobId, + ]); + expect(await store.getAttachment(AttachmentIdSchema.parse('att-leased'))).toBeUndefined(); + expect(await store.getBlob(leasedOnly.blobId)).toBeUndefined(); + expect(await store.getBlob(viaResource.blobId)).toEqual(viaResource); + }); +}); diff --git a/apps/daemon/src/__tests__/resource-store.test.ts b/apps/daemon/src/__tests__/resource-store.test.ts index 2da6ec343..c76ff758d 100644 --- a/apps/daemon/src/__tests__/resource-store.test.ts +++ b/apps/daemon/src/__tests__/resource-store.test.ts @@ -46,6 +46,9 @@ describe('SQLite resource store', () => { kind: 'document', status: 'ready', locator: { type: 'workspace-file', path: locatorKey }, + attachmentId: 'att-report', + mimeType: 'application/pdf', + sizeBytes: 6, createdAt: 2, updatedAt: 2, }); diff --git a/apps/daemon/src/attachment-store.ts b/apps/daemon/src/attachment-store.ts new file mode 100644 index 000000000..0282b8cf2 --- /dev/null +++ b/apps/daemon/src/attachment-store.ts @@ -0,0 +1,230 @@ +import type { + AttachmentCommit, + AttachmentStore, + AttachmentSweepWindow, + StoredAttachment, +} from '@linkcode/engine'; +import type { AttachmentId, BlobId, BlobRecord, UploadId, UploadLease } from '@linkcode/schema'; +import { + AttachmentRecordSchema, + BlobIdSchema, + BlobRecordSchema, + blobIdFromSha256, + UploadLeaseSchema, +} from '@linkcode/schema'; +import { and, eq, inArray, isNotNull, lt, lte, notInArray } from 'drizzle-orm'; +import type { DaemonDatabaseClient } from './db/database'; +import { + attachmentBlobs, + attachments, + blobs, + promptAttachmentRefs, + sessionResources, + uploadLeases, +} from './db/schema'; + +const ORIGINAL_VARIANT = 'original'; + +type AttachmentRow = typeof attachments.$inferSelect; +type LeaseRow = typeof uploadLeases.$inferSelect; + +/** + * SQLite-backed `AttachmentStore` on the daemon's shared graph connection: the reaper reads its + * roots inside its own transaction, and the conversation store's submit transaction claims leases. + */ +export function createAttachmentStore(db: DaemonDatabaseClient): AttachmentStore { + function list(attachmentIds: readonly AttachmentId[]): StoredAttachment[] { + if (attachmentIds.length === 0) return []; + const rows = db + .select({ attachment: attachments, blobId: attachmentBlobs.blobId }) + .from(attachments) + .innerJoin( + attachmentBlobs, + and( + eq(attachmentBlobs.attachmentId, attachments.attachmentId), + eq(attachmentBlobs.variant, ORIGINAL_VARIANT), + ), + ) + .where(inArray(attachments.attachmentId, Array.from(attachmentIds))) + .all(); + return rows.map((row) => toStoredAttachment(row.attachment, row.blobId)); + } + + return { + getAttachment(attachmentId: AttachmentId): Promise { + return Promise.resolve(list([attachmentId])[0]); + }, + + listAttachments(attachmentIds: readonly AttachmentId[]): Promise { + return Promise.resolve(list(attachmentIds)); + }, + + getBlob(blobId: BlobId): Promise { + const row = db.select().from(blobs).where(eq(blobs.blobId, blobId)).get(); + return Promise.resolve(row ? BlobRecordSchema.parse(row) : undefined); + }, + + getLease(uploadId: UploadId): Promise { + const row = db.select().from(uploadLeases).where(eq(uploadLeases.uploadId, uploadId)).get(); + return Promise.resolve(row ? toLease(row) : undefined); + }, + + beginUpload(lease: UploadLease): Promise { + const pinned = db.transaction((tx) => { + const declared = blobIdFromSha256(lease.declaredSha256); + const existing = tx + .select({ blobId: blobs.blobId }) + .from(blobs) + .where(eq(blobs.blobId, declared)) + .get(); + const row: UploadLease = existing ? { ...lease, blobId: declared } : lease; + // Plain insert: a reused upload id conflicts instead of silently re-leasing. + tx.insert(uploadLeases).values(toLeaseRow(row)).run(); + return row; + }); + return Promise.resolve(pinned); + }, + + deleteLease(uploadId: UploadId): Promise { + db.delete(uploadLeases).where(eq(uploadLeases.uploadId, uploadId)).run(); + return Promise.resolve(); + }, + + commitAttachment({ attachment, blob, uploadId }: AttachmentCommit): Promise { + db.transaction((tx) => { + tx.insert(blobs).values(blob).onConflictDoNothing().run(); + tx.insert(attachments).values(toAttachmentRow(attachment)).run(); + tx.insert(attachmentBlobs) + .values({ + attachmentId: attachment.attachmentId, + variant: ORIGINAL_VARIANT, + blobId: blob.blobId, + }) + .run(); + if (uploadId !== undefined) { + tx.update(uploadLeases) + .set({ blobId: blob.blobId, attachmentId: attachment.attachmentId }) + .where(eq(uploadLeases.uploadId, uploadId)) + .run(); + } + }); + return Promise.resolve(); + }, + + sweep({ graceBefore, now }: AttachmentSweepWindow): Promise { + const doomed = db.transaction((tx) => { + tx.delete(uploadLeases).where(lte(uploadLeases.expiresAt, now)).run(); + // NOT IN over a nullable column needs the IS NOT NULL filter: one NULL would make the + // predicate unknown for every row and the reaper would silently collect nothing. + tx.delete(attachments) + .where( + and( + lt(attachments.createdAt, graceBefore), + notInArray( + attachments.attachmentId, + tx.select({ id: promptAttachmentRefs.attachmentId }).from(promptAttachmentRefs), + ), + notInArray( + attachments.attachmentId, + tx + .select({ id: sessionResources.attachmentId }) + .from(sessionResources) + .where(isNotNull(sessionResources.attachmentId)), + ), + notInArray( + attachments.attachmentId, + tx + .select({ id: uploadLeases.attachmentId }) + .from(uploadLeases) + .where(isNotNull(uploadLeases.attachmentId)), + ), + ), + ) + .run(); + const rows = tx + .select({ blobId: blobs.blobId }) + .from(blobs) + .where( + and( + lt(blobs.createdAt, graceBefore), + notInArray( + blobs.blobId, + tx.select({ id: attachmentBlobs.blobId }).from(attachmentBlobs), + ), + notInArray( + blobs.blobId, + tx + .select({ id: uploadLeases.blobId }) + .from(uploadLeases) + .where(isNotNull(uploadLeases.blobId)), + ), + ), + ) + .all(); + const blobIds = rows.map((row) => BlobIdSchema.parse(row.blobId)); + if (blobIds.length > 0) tx.delete(blobs).where(inArray(blobs.blobId, blobIds)).run(); + return blobIds; + }); + return Promise.resolve(doomed); + }, + }; +} + +function toAttachmentRow( + attachment: AttachmentCommit['attachment'], +): typeof attachments.$inferInsert { + return { + attachmentId: attachment.attachmentId, + kind: attachment.kind, + name: attachment.name, + mimeType: attachment.mimeType, + sizeBytes: attachment.sizeBytes, + metadataJson: JSON.stringify(attachment.metadata), + createdAt: attachment.createdAt, + }; +} + +function toStoredAttachment(row: AttachmentRow, blobId: string): StoredAttachment { + return { + ...AttachmentRecordSchema.parse({ + attachmentId: row.attachmentId, + kind: row.kind, + name: row.name, + mimeType: row.mimeType, + sizeBytes: row.sizeBytes, + metadata: JSON.parse(row.metadataJson), + createdAt: row.createdAt, + }), + blobId: BlobIdSchema.parse(blobId), + }; +} + +function toLeaseRow(lease: UploadLease): typeof uploadLeases.$inferInsert { + return { + uploadId: lease.uploadId, + declaredSha256: lease.declaredSha256, + declaredSize: lease.declaredSize, + name: lease.name, + mimeType: lease.mimeType ?? null, + kind: lease.kind, + blobId: lease.blobId ?? null, + attachmentId: lease.attachmentId ?? null, + expiresAt: lease.expiresAt, + createdAt: lease.createdAt, + }; +} + +function toLease(row: LeaseRow): UploadLease { + return UploadLeaseSchema.parse({ + uploadId: row.uploadId, + declaredSha256: row.declaredSha256, + declaredSize: row.declaredSize, + name: row.name, + mimeType: row.mimeType ?? undefined, + kind: row.kind, + blobId: row.blobId ?? undefined, + attachmentId: row.attachmentId ?? undefined, + expiresAt: row.expiresAt, + createdAt: row.createdAt, + }); +} diff --git a/apps/daemon/src/conversation-store.ts b/apps/daemon/src/conversation-store.ts index 1fa67902f..e69a196e2 100644 --- a/apps/daemon/src/conversation-store.ts +++ b/apps/daemon/src/conversation-store.ts @@ -24,6 +24,7 @@ import { promptAttachmentRefs, prompts, providerTurnBindings, + uploadLeases, } from './db/schema'; type TurnRow = typeof conversationTurns.$inferSelect; @@ -154,6 +155,8 @@ export function createConversationStore(db: DaemonDatabaseClient): ConversationS ) .onConflictDoNothing() .run(); + // The prompt is the root now; release the draft leases that pinned these attachments. + tx.delete(uploadLeases).where(inArray(uploadLeases.attachmentId, referenced)).run(); } } // Plain inserts: a replayed operationId must conflict here, never re-open a terminal row. diff --git a/apps/daemon/src/db/schema.ts b/apps/daemon/src/db/schema.ts index ed90820f1..78c2c8f52 100644 --- a/apps/daemon/src/db/schema.ts +++ b/apps/daemon/src/db/schema.ts @@ -90,6 +90,9 @@ export const sessionResources = sqliteTable( }).notNull(), locator: text('locator').notNull(), normalizedLocatorKey: text('normalized_locator_key'), + /** Set when the bytes live in the attachment store; a GC root read by ../attachment-store.ts. + * No FK: the reaper deletes an attachment only once no resource names it. */ + attachmentId: text('attachment_id'), mimeType: text('mime_type'), sizeBytes: integer('size_bytes'), error: text('error'), @@ -207,6 +210,65 @@ export const conversationOperations = sqliteTable( ], ); +/** + * Attachment store metadata (`Blob*`/`Attachment*`/`UploadLease` schemas). Bytes live in the blob + * store on disk under `blob_id`; these rows are the GC roots and edges, written by + * ../attachment-store.ts on the shared connection so lease claims ride the submit transaction. + */ +export const blobs = sqliteTable('blobs', { + blobId: text('blob_id').primaryKey(), + sizeBytes: integer('size_bytes').notNull(), + createdAt: integer('created_at').notNull(), +}); + +export const attachments = sqliteTable('attachments', { + attachmentId: text('attachment_id').primaryKey(), + kind: text('kind').notNull(), + name: text('name').notNull(), + mimeType: text('mime_type').notNull(), + sizeBytes: integer('size_bytes').notNull(), + /** JSON object, size-capped by the zod schema. */ + metadataJson: text('metadata_json').notNull(), + createdAt: integer('created_at').notNull(), +}); + +/** `original` today; a derived variant (thumbnail, extracted text) is an insert, not a migration. */ +export const attachmentBlobs = sqliteTable( + 'attachment_blobs', + { + attachmentId: text('attachment_id') + .notNull() + .references(() => attachments.attachmentId, { onDelete: 'cascade' }), + variant: text('variant').notNull(), + blobId: text('blob_id') + .notNull() + .references(() => blobs.blobId), + }, + (table) => [ + primaryKey({ columns: [table.attachmentId, table.variant] }), + index('attachment_blobs_blob_idx').on(table.blobId), + ], +); + +/** In-flight and unclaimed uploads; a GC root until claimed or expired. No FKs: the reaper itself + * deletes a pinned blob or attachment only once no lease names it. */ +export const uploadLeases = sqliteTable( + 'upload_leases', + { + uploadId: text('upload_id').primaryKey(), + declaredSha256: text('declared_sha256').notNull(), + declaredSize: integer('declared_size').notNull(), + name: text('name').notNull(), + mimeType: text('mime_type'), + kind: text('kind').notNull(), + blobId: text('blob_id'), + attachmentId: text('attachment_id'), + expiresAt: integer('expires_at').notNull(), + createdAt: integer('created_at').notNull(), + }, + (table) => [index('upload_leases_expires_at_idx').on(table.expiresAt)], +); + /** * Recurring automations; mirrors `Schedule` from `@linkcode/schema` (spec fields flattened into * columns). `target_session_id` deliberately has no foreign key — a deleted target is the signal the diff --git a/apps/daemon/src/index.ts b/apps/daemon/src/index.ts index df9e5098e..5f69a83a3 100644 --- a/apps/daemon/src/index.ts +++ b/apps/daemon/src/index.ts @@ -28,6 +28,7 @@ import { Cause, Context, Effect, Exit, Layer, Option } from 'effect'; import { extractErrorMessage } from 'foxts/extract-error-message'; import { createAiGatewaySidecar } from './ai-gateway'; import { installAsarSpawnFix } from './asar-spawn'; +import { createAttachmentStore } from './attachment-store'; import { adoptLegacyDeviceKeyFile } from './cloud/device-key'; import { runLoginCommand, runLogoutCommand } from './cloud/login'; import { startCloudUplink } from './cloud/uplink'; @@ -280,6 +281,7 @@ async function main(): Promise { simulatorConsent, sessionStore: createSessionStore(database.client), conversationStore: createConversationStore(database.client), + attachmentStore: createAttachmentStore(database.client), resourceStore: createResourceStore(databasePath()), stateDir: daemonStateDir(), scheduleStore: createScheduleStore(databasePath()), diff --git a/apps/daemon/src/resource-store.ts b/apps/daemon/src/resource-store.ts index 54ab23e93..f1ccdd57b 100644 --- a/apps/daemon/src/resource-store.ts +++ b/apps/daemon/src/resource-store.ts @@ -94,6 +94,7 @@ function toResource(row: Row): SessionResource { row.locatorType === 'url' ? { type: 'url', url: row.locator } : { type: row.locatorType, path: row.locator }, + attachmentId: row.attachmentId ?? undefined, mimeType: row.mimeType ?? undefined, sizeBytes: row.sizeBytes ?? undefined, error: row.error ?? undefined, @@ -112,6 +113,7 @@ function toRow(resource: SessionResource, key?: string): typeof sessionResources locatorType: resource.locator.type, locator: resource.locator.type === 'url' ? resource.locator.url : resource.locator.path, normalizedLocatorKey: key ?? null, + attachmentId: resource.attachmentId ?? null, mimeType: resource.mimeType ?? null, sizeBytes: resource.sizeBytes ?? null, error: resource.error ?? null, From 7d0a59ae43bfeb85e8c377ad5a83b812097bc419 Mon Sep 17 00:00:00 2001 From: Zerlight Wu Date: Thu, 3 Sep 2026 14:37:04 +0800 Subject: [PATCH 4/5] feat(engine): store session resource bytes in the blob store --- .../foundation/schema/src/model/attachment.ts | 8 +-- .../src/__tests__/engine-resources.test.ts | 31 ++++++++- .../session-event-sequencing.test.ts | 6 ++ packages/host/engine/src/engine.ts | 36 +++++++---- packages/host/engine/src/resource/service.ts | 64 +++++++++++++++---- 5 files changed, 111 insertions(+), 34 deletions(-) diff --git a/packages/foundation/schema/src/model/attachment.ts b/packages/foundation/schema/src/model/attachment.ts index 5a7f642d0..7f40e3bdb 100644 --- a/packages/foundation/schema/src/model/attachment.ts +++ b/packages/foundation/schema/src/model/attachment.ts @@ -43,8 +43,8 @@ export const MAX_ATTACHMENT_METADATA_BYTES = 4096; export const AttachmentRecordSchema = z.object({ attachmentId: AttachmentIdSchema, kind: AttachmentKindSchema, - name: z.string().min(1).max(255), - mimeType: z.string().min(1).max(255), + name: z.string().min(1), + mimeType: z.string().min(1), sizeBytes: z.number().int().nonnegative(), metadata: z .record(z.string(), z.unknown()) @@ -72,8 +72,8 @@ export const UploadLeaseSchema = z.object({ uploadId: UploadIdSchema, declaredSha256: Sha256HexSchema, declaredSize: z.number().int().nonnegative(), - name: z.string().min(1).max(255), - mimeType: z.string().min(1).max(255).optional(), + name: z.string().min(1), + mimeType: z.string().min(1).optional(), kind: AttachmentKindSchema, blobId: BlobIdSchema.optional(), attachmentId: AttachmentIdSchema.optional(), diff --git a/packages/host/engine/src/__tests__/engine-resources.test.ts b/packages/host/engine/src/__tests__/engine-resources.test.ts index d1e449199..89ddbc57b 100644 --- a/packages/host/engine/src/__tests__/engine-resources.test.ts +++ b/packages/host/engine/src/__tests__/engine-resources.test.ts @@ -8,6 +8,7 @@ import { RESOURCE_CONTEXT_SENTINEL } from '../resource/service'; import { createSessionHarness, startedSessionId } from './fixtures/session-harness'; const temporaryDirectories: string[] = []; +const rAttachmentId = /^att-/; afterEach(async () => { await Promise.all( @@ -72,9 +73,12 @@ describe('engine session resources', () => { name: 'brief.txt', status: 'ready', locator: { type: 'managed-file' }, + attachmentId: expect.stringMatching(rAttachmentId), }); if (source.locator.type !== 'managed-file') throw new Error('expected managed source'); + expect(source.locator.path.startsWith(join(stateDir, 'blobs', 'sha256'))).toBe(true); expect(await readFile(source.locator.path, 'utf8')).toBe('source material'); + expect((await stat(source.locator.path)).mode & 0o222).toBe(0); const mark = h.sent.length; await h.inject({ @@ -111,7 +115,30 @@ describe('engine session resources', () => { await vi.waitFor(() => { expect(h.sent).toContainEqual({ kind: 'request.succeeded', replyTo: 'remove' }); }); - await expect(stat(source.locator.path)).rejects.toMatchObject({ code: 'ENOENT' }); + // Store bytes are shared and immutable: removal drops the reference, the reaper takes the file. + expect(await readFile(source.locator.path, 'utf8')).toBe('source material'); + await h.inject({ kind: 'resource.list', clientReqId: 'list-after-remove', sessionId }); + expect(listedResources(h.sent, 'list-after-remove')).toEqual([]); + + await h.inject({ + kind: 'resource.source.upload', + clientReqId: 'upload-mismatch', + sessionId, + name: 'photo.png', + mimeType: 'image/png', + data: Buffer.from('not a png').toString('base64'), + }); + await vi.waitFor(() => { + expect(h.sent).toContainEqual( + expect.objectContaining({ + kind: 'request.failed', + replyTo: 'upload-mismatch', + code: 'invalid_request', + }), + ); + }); + await h.inject({ kind: 'resource.list', clientReqId: 'list-after-mismatch', sessionId }); + expect(listedResources(h.sent, 'list-after-mismatch')).toEqual([]); await h.inject({ kind: 'resource.source.upload', @@ -134,7 +161,7 @@ describe('engine session resources', () => { await vi.waitFor(() => { expect(h.sent).toContainEqual({ kind: 'request.succeeded', replyTo: 'delete-session' }); }); - await expect(stat(uploaded.resource.locator.path)).rejects.toMatchObject({ code: 'ENOENT' }); + expect(await readFile(uploaded.resource.locator.path, 'utf8')).toBe('temporary'); const retained = join(stateDir, 'retained.txt'); await writeFile(retained, 'safe'); diff --git a/packages/host/engine/src/__tests__/session-event-sequencing.test.ts b/packages/host/engine/src/__tests__/session-event-sequencing.test.ts index 0dc9b4b83..a7341f334 100644 --- a/packages/host/engine/src/__tests__/session-event-sequencing.test.ts +++ b/packages/host/engine/src/__tests__/session-event-sequencing.test.ts @@ -1,3 +1,5 @@ +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; import type { AgentEvent, ConversationWatermark, @@ -15,6 +17,8 @@ import { Deferred, Effect, Scope } from 'effect'; import { noop } from 'foxts/noop'; import { describe, expect, it } from 'vitest'; import { AgentRuntimeService } from '../agent/runtime-service'; +import { InMemoryAttachmentStore } from '../attachment/attachment-store'; +import { FsBlobStore } from '../attachment/blob-store'; import { InMemoryConversationStore } from '../conversation/conversation-store'; import { ConversationLiveJournals } from '../conversation/live-journal'; import { ConversationTurnService } from '../conversation/turn-service'; @@ -269,6 +273,8 @@ describe('stale-run events at saga cutover', () => { registry, undefined, new FileHostService(new PreviewRouteRegistry()), + new FsBlobStore(join(tmpdir(), 'linkcode-sequencing-blobs')), + new InMemoryAttachmentStore(), ), turns, journals, diff --git a/packages/host/engine/src/engine.ts b/packages/host/engine/src/engine.ts index fd699e755..9ec49d518 100644 --- a/packages/host/engine/src/engine.ts +++ b/packages/host/engine/src/engine.ts @@ -116,7 +116,28 @@ export const createEngineRuntime = Effect.fn('Engine.create')(function* ( (dir) => Effect.promise(() => rm(dir, { recursive: true, force: true })), )); const resourceStore = deps.resourceStore ?? new InMemoryResourceStore(); - const resources = new ResourceService(transport, resourceStore, records, stateDir, fileHost); + const conversationStore = deps.conversationStore ?? new InMemoryConversationStore(); + const blobStore = deps.blobStore ?? new FsBlobStore(join(stateDir, 'blobs')); + const attachmentStore = + deps.attachmentStore ?? + new InMemoryAttachmentStore(() => [ + ...(conversationStore instanceof InMemoryConversationStore + ? conversationStore.referencedAttachmentIds() + : []), + ...(resourceStore instanceof InMemoryResourceStore + ? resourceStore.referencedAttachmentIds() + : []), + ]); + const attachmentGc = new AttachmentGc(attachmentStore, blobStore); + const resources = new ResourceService( + transport, + resourceStore, + records, + stateDir, + fileHost, + blobStore, + attachmentStore, + ); const plugins = new PluginService(deps.pluginFactory ?? createPluginProviderAdapter); const translator = deps.translator; const startOptions = new SessionStartOptionsResolver( @@ -159,19 +180,6 @@ export const createEngineRuntime = Effect.fn('Engine.create')(function* ( // predicate that gates claims on a live session. const simulators = deps.simulators; const browserBroker = new BrowserBrokerService(transport); - const conversationStore = deps.conversationStore ?? new InMemoryConversationStore(); - const blobStore = deps.blobStore ?? new FsBlobStore(join(stateDir, 'blobs')); - const attachmentStore = - deps.attachmentStore ?? - new InMemoryAttachmentStore(() => [ - ...(conversationStore instanceof InMemoryConversationStore - ? conversationStore.referencedAttachmentIds() - : []), - ...(resourceStore instanceof InMemoryResourceStore - ? resourceStore.referencedAttachmentIds() - : []), - ]); - const attachmentGc = new AttachmentGc(attachmentStore, blobStore); const conversationTurns = new ConversationTurnService( conversationStore, records, diff --git a/packages/host/engine/src/resource/service.ts b/packages/host/engine/src/resource/service.ts index d7e53df3d..55806f428 100644 --- a/packages/host/engine/src/resource/service.ts +++ b/packages/host/engine/src/resource/service.ts @@ -1,14 +1,22 @@ -import { randomUUID } from 'node:crypto'; -import { mkdir, rm, stat, writeFile } from 'node:fs/promises'; +import { createHash, randomUUID } from 'node:crypto'; +import { rm, stat } from 'node:fs/promises'; import { tmpdir } from 'node:os'; import { basename, extname, join, resolve, sep, win32 } from 'node:path'; import { fileURLToPath } from 'node:url'; import type { SessionId, SessionResource, SessionResourceId } from '@linkcode/schema'; -import { MAX_ATTACHMENT_BYTES, SessionResourceIdSchema } from '@linkcode/schema'; +import { + AttachmentIdSchema, + blobIdFromSha256, + MAX_ATTACHMENT_BYTES, + SessionResourceIdSchema, +} from '@linkcode/schema'; import type { Transport } from '@linkcode/transport'; import { createWireMessage } from '@linkcode/transport'; import { Effect } from 'effect'; import { noop } from 'foxts/noop'; +import type { AttachmentStore } from '../attachment/attachment-store'; +import type { BlobStore } from '../attachment/blob-store'; +import { declaredMimeTypeMatches } from '../attachment/mime-sniff'; import { OperationError, RequestError } from '../failure'; import type { FileHostService } from '../preview/file-host-service'; import type { SessionRecordRegistry } from '../session/session-record-registry'; @@ -57,6 +65,8 @@ export class ResourceService { private readonly records: SessionRecordRegistry, private readonly stateDir: string | undefined, private readonly fileHost: FileHostService, + private readonly blobs: BlobStore, + private readonly attachments: AttachmentStore, ) {} list(sessionId: SessionId): Effect.Effect { @@ -69,7 +79,7 @@ export class ResourceService { mimeType: string | undefined, data: string, ): Effect.Effect { - const { records, stateDir, transport } = this; + const { attachments, blobs, records, transport } = this; return Effect.gen({ self: this }, function* () { if (!records.has(sessionId)) { return yield* new RequestError({ code: 'not_found', message: 'Session not found' }); @@ -81,16 +91,18 @@ export class ResourceService { message: 'Resource exceeds the 8 MiB limit', }); } - const resourceId = SessionResourceIdSchema.parse(`resource-${randomUUID()}`); - const directory = sessionResourceDirectory(stateDir, sessionId); - if (!directory) { + if (mimeType !== undefined && !declaredMimeTypeMatches(mimeType, bytes.subarray(0, 16))) { return yield* new RequestError({ code: 'invalid_request', - message: 'Session resource path is invalid', + message: `File contents are not ${mimeType}`, }); } - const path = resolve(directory, resourceId); + const resourceId = SessionResourceIdSchema.parse(`resource-${randomUUID()}`); + const attachmentId = AttachmentIdSchema.parse(`att-${randomUUID()}`); + const sha256 = createHash('sha256').update(bytes).digest('hex'); + const blobId = blobIdFromSha256(sha256); const now = Date.now(); + // The harness reads the blob's own path: immutable, shared, mode 0444. let resource: SessionResource = { resourceId, sessionId, @@ -98,7 +110,8 @@ export class ResourceService { name, kind: classify(name, mimeType), status: 'processing', - locator: { type: 'managed-file', path }, + locator: { type: 'managed-file', path: blobs.pathOf(blobId) }, + attachmentId, mimeType, sizeBytes: bytes.byteLength, createdAt: now, @@ -108,8 +121,26 @@ export class ResourceService { transport.send(createWireMessage({ kind: 'resource.changed', resource })); const written = yield* Effect.tryPromise({ async try() { - await mkdir(resolve(path, '..'), { recursive: true }); - await writeFile(path, bytes); + const stage = await blobs.stage(resourceId); + try { + await stage.write(0, bytes); + await stage.commit({ sha256, sizeBytes: bytes.byteLength }); + } catch (error) { + await stage.abort().catch(noop); + throw error; + } + await attachments.commitAttachment({ + blob: { blobId, sizeBytes: bytes.byteLength, createdAt: now }, + attachment: { + attachmentId, + kind: resource.kind, + name, + mimeType: mimeType ?? 'application/octet-stream', + sizeBytes: bytes.byteLength, + metadata: {}, + createdAt: now, + }, + }); }, catch: (cause) => cause, }).pipe( @@ -124,7 +155,6 @@ export class ResourceService { error: 'Failed to persist uploaded resource', updatedAt: Date.now(), }; - if (!written) yield* Effect.promise(() => rm(path, { force: true }).catch(noop)); yield* this.run('save', () => this.store.save(resource)); transport.send(createWireMessage({ kind: 'resource.changed', resource })); return resource; @@ -136,7 +166,13 @@ export class ResourceService { Effect.flatMap((resource) => Effect.promise(async () => { if (!resource) return; - if (resource.direction === 'source' && resource.locator.type === 'managed-file') { + // Store-backed bytes are shared and reaped by the attachment GC; only a pre-store + // resource owns its file. + if ( + resource.direction === 'source' && + resource.locator.type === 'managed-file' && + resource.attachmentId === undefined + ) { await rm(resource.locator.path, { force: true }); } this.transport.send( From a9339dc70d974af3e029ded536337e56b7911e4a Mon Sep 17 00:00:00 2001 From: Zerlight Wu Date: Thu, 3 Sep 2026 15:32:46 +0800 Subject: [PATCH 5/5] fix(engine): serialize blob GC unlinks with publish and persist resources after the blob row --- .../src/__tests__/attachment-gc.test.ts | 25 ++++- .../engine/src/__tests__/blob-store.test.ts | 7 +- .../src/__tests__/engine-resources.test.ts | 30 ++++++ .../host/engine/src/attachment/blob-store.ts | 9 ++ packages/host/engine/src/attachment/gc.ts | 45 +++++--- .../host/engine/src/attachment/io-mutex.ts | 18 ++++ .../host/engine/src/attachment/mime-sniff.ts | 16 ++- packages/host/engine/src/engine.ts | 5 +- packages/host/engine/src/resource/service.ts | 102 ++++++++++-------- 9 files changed, 188 insertions(+), 69 deletions(-) create mode 100644 packages/host/engine/src/attachment/io-mutex.ts diff --git a/packages/host/engine/src/__tests__/attachment-gc.test.ts b/packages/host/engine/src/__tests__/attachment-gc.test.ts index a82a3ccf3..f3bbb6b31 100644 --- a/packages/host/engine/src/__tests__/attachment-gc.test.ts +++ b/packages/host/engine/src/__tests__/attachment-gc.test.ts @@ -2,7 +2,7 @@ import { createHash } from 'node:crypto'; import { mkdtemp, rm, stat } from 'node:fs/promises'; import { tmpdir } from 'node:os'; import { join } from 'node:path'; -import type { AttachmentId, BlobId, UploadLease } from '@linkcode/schema'; +import type { BlobId, UploadLease } from '@linkcode/schema'; import { AttachmentIdSchema, ConversationOperationSchema, @@ -183,6 +183,27 @@ describe('AttachmentGc', () => { await expect(stat(join(f.blobs.pathOf(kept), '..', '..', '..', 'tmp'))).rejects.toMatchObject({ code: 'ENOENT', }); - expect(f.store.getAttachment('att-kept' as AttachmentId)).resolves.toBeDefined(); + await expect( + f.store.getAttachment(AttachmentIdSchema.parse('att-kept')), + ).resolves.toBeDefined(); + }); + + it('does not unlink a blob that is re-committed after the reaper transaction', async () => { + const f = await fixture(); + const blobId = await f.commit('att-old', 'shared bytes'); + f.clock.now += ATTACHMENT_GC_GRACE_MS + 1; + + const originalSweep = f.store.sweep.bind(f.store); + f.store.sweep = async (window) => { + const doomed = await originalSweep(window); + await f.commit('att-new', 'shared bytes'); + return doomed; + }; + + expect(await f.gc.sweep()).toEqual({ removedBlobs: [] }); + expect(await f.blobs.stat(blobId)).toBeDefined(); + expect(await f.store.getAttachment(AttachmentIdSchema.parse('att-new'))).toMatchObject({ + blobId, + }); }); }); diff --git a/packages/host/engine/src/__tests__/blob-store.test.ts b/packages/host/engine/src/__tests__/blob-store.test.ts index 42aad0740..d611649b5 100644 --- a/packages/host/engine/src/__tests__/blob-store.test.ts +++ b/packages/host/engine/src/__tests__/blob-store.test.ts @@ -110,10 +110,13 @@ describe('mime sniff', () => { expect(sniffImageMimeType(new Uint8Array(0))).toBeUndefined(); }); - it('holds image declarations to their bytes and trusts the rest', () => { + it('holds sniffable image declarations to their bytes and trusts the rest', () => { expect(declaredMimeTypeMatches('image/png', png)).toBe(true); expect(declaredMimeTypeMatches('image/jpeg', png)).toBe(false); - expect(declaredMimeTypeMatches('image/svg+xml', Buffer.from(''))).toBe(false); + expect(declaredMimeTypeMatches('image/jpeg', Buffer.from(''))).toBe(false); + expect(declaredMimeTypeMatches('image/svg+xml', Buffer.from(''))).toBe(true); + expect(declaredMimeTypeMatches('image/svg+xml', png)).toBe(false); + expect(declaredMimeTypeMatches('image/heic', Buffer.from('ftypheic'))).toBe(true); expect(declaredMimeTypeMatches('application/pdf', Buffer.from('%PDF-1.7'))).toBe(true); expect(declaredMimeTypeMatches('text/plain', png)).toBe(true); }); diff --git a/packages/host/engine/src/__tests__/engine-resources.test.ts b/packages/host/engine/src/__tests__/engine-resources.test.ts index 89ddbc57b..774e544da 100644 --- a/packages/host/engine/src/__tests__/engine-resources.test.ts +++ b/packages/host/engine/src/__tests__/engine-resources.test.ts @@ -140,6 +140,36 @@ describe('engine session resources', () => { await h.inject({ kind: 'resource.list', clientReqId: 'list-after-mismatch', sessionId }); expect(listedResources(h.sent, 'list-after-mismatch')).toEqual([]); + await h.inject({ + kind: 'resource.source.upload', + clientReqId: 'upload-svg', + sessionId, + name: 'icon.svg', + mimeType: 'image/svg+xml', + data: Buffer.from('').toString('base64'), + }); + await vi.waitFor(() => { + expect(h.sent).toContainEqual( + expect.objectContaining({ kind: 'resource.uploaded', replyTo: 'upload-svg' }), + ); + }); + await h.inject({ kind: 'resource.list', clientReqId: 'list-svg', sessionId }); + expect(listedResources(h.sent, 'list-svg')).toEqual([ + expect.objectContaining({ + name: 'icon.svg', + status: 'ready', + mimeType: 'image/svg+xml', + }), + ]); + await h.inject({ + kind: 'resource.remove', + clientReqId: 'remove-svg', + resourceId: listedResources(h.sent, 'list-svg')[0].resourceId, + }); + await vi.waitFor(() => { + expect(h.sent).toContainEqual({ kind: 'request.succeeded', replyTo: 'remove-svg' }); + }); + await h.inject({ kind: 'resource.source.upload', clientReqId: 'upload-for-delete', diff --git a/packages/host/engine/src/attachment/blob-store.ts b/packages/host/engine/src/attachment/blob-store.ts index 54e216546..f751b4f93 100644 --- a/packages/host/engine/src/attachment/blob-store.ts +++ b/packages/host/engine/src/attachment/blob-store.ts @@ -149,7 +149,16 @@ export class FsBlobStore implements BlobStore { await rename(stagingPath, dest); } catch (error) { if ((await this.stat(blobId)) === undefined) throw error; + // Dest existing is not proof of identical bytes: Windows rename does not replace a + // read-only dest, so a truncated or bitrot file would otherwise count as success. + const existing = await sha256OfFile(dest); + const expected = blobId.slice(BLOB_ID_PREFIX.length); await rm(stagingPath, { force: true }); + if (existing !== expected) { + throw new BlobIntegrityError('Existing blob bytes do not match the declared SHA-256', { + cause: error, + }); + } } } } diff --git a/packages/host/engine/src/attachment/gc.ts b/packages/host/engine/src/attachment/gc.ts index 893903ec7..921e01e2b 100644 --- a/packages/host/engine/src/attachment/gc.ts +++ b/packages/host/engine/src/attachment/gc.ts @@ -2,6 +2,7 @@ import type { BlobId } from '@linkcode/schema'; import { Effect, Schedule } from 'effect'; import type { AttachmentStore } from './attachment-store'; import type { BlobStore } from './blob-store'; +import { AttachmentIoMutex } from './io-mutex'; /** A draft's upload outlives any composer session, not a forgotten one. */ export const UPLOAD_LEASE_TTL_MS = 24 * 60 * 60 * 1000; @@ -19,41 +20,53 @@ export class AttachmentGc { private readonly store: AttachmentStore, private readonly blobs: BlobStore, private readonly clock: () => number = Date.now, + private readonly io: AttachmentIoMutex = new AttachmentIoMutex(), ) {} /** A failed unlink leaves an orphan file the next boot sweep removes. */ async sweep(): Promise { + return this.io.run(() => this.sweepBody()); + } + + private async sweepBody(): Promise { const now = this.clock(); const removedBlobs = await this.store.sweep({ now, graceBefore: now - ATTACHMENT_GC_GRACE_MS, }); + const unlinked: BlobId[] = []; for (let i = 0, len = removedBlobs.length; i < len; i++) { + const blobId = removedBlobs[i]; // eslint-disable-next-line no-await-in-loop -- sequential unlinks of a short list - await this.blobs.delete(removedBlobs[i]); + if (await this.store.getBlob(blobId)) continue; + // eslint-disable-next-line no-await-in-loop -- same + await this.blobs.delete(blobId); + unlinked.push(blobId); } - return { removedBlobs }; + return { removedBlobs: unlinked }; } /** * Run before requests are accepted: no upload survives a restart, so staging goes wholesale; * then a sweep; then bytes with no blob row (a crash between publish and the row insert). - * Safe only while no commit can be publishing — that is what the boot ordering guarantees. + * File unlinks share the publish mutex: they are only safe while no commit can be publishing. */ async bootSweep(): Promise { - await this.blobs.purgeStaging(); - const report = await this.sweep(); - const onDisk = await this.blobs.list(); - const orphans: BlobId[] = []; - for (let i = 0, len = onDisk.length; i < len; i++) { - const blobId = onDisk[i]; - // eslint-disable-next-line no-await-in-loop -- one row lookup per file on disk - if (await this.store.getBlob(blobId)) continue; - // eslint-disable-next-line no-await-in-loop -- same - await this.blobs.delete(blobId); - orphans.push(blobId); - } - return { removedBlobs: [...report.removedBlobs, ...orphans] }; + return this.io.run(async () => { + await this.blobs.purgeStaging(); + const report = await this.sweepBody(); + const onDisk = await this.blobs.list(); + const orphans: BlobId[] = []; + for (let i = 0, len = onDisk.length; i < len; i++) { + const blobId = onDisk[i]; + // eslint-disable-next-line no-await-in-loop -- one row lookup per file on disk + if (await this.store.getBlob(blobId)) continue; + // eslint-disable-next-line no-await-in-loop -- same + await this.blobs.delete(blobId); + orphans.push(blobId); + } + return { removedBlobs: [...report.removedBlobs, ...orphans] }; + }); } /** One sweep per interval until interrupted; a failed sweep is logged and the cadence goes on. */ diff --git a/packages/host/engine/src/attachment/io-mutex.ts b/packages/host/engine/src/attachment/io-mutex.ts new file mode 100644 index 000000000..38eec60f1 --- /dev/null +++ b/packages/host/engine/src/attachment/io-mutex.ts @@ -0,0 +1,18 @@ +import { noop } from 'foxts/noop'; + +/** + * Serializes blob publish + row insert with GC unlinks. A doomed id can grow a new row between + * the reaper transaction and the unlink; those two must not overlap. + */ +export class AttachmentIoMutex { + private tail: Promise = Promise.resolve(); + + run(work: () => Promise): Promise { + const previous = this.tail; + let release: () => void = noop; + this.tail = new Promise((resolve) => { + release = resolve; + }); + return previous.catch(noop).then(work).finally(release); + } +} diff --git a/packages/host/engine/src/attachment/mime-sniff.ts b/packages/host/engine/src/attachment/mime-sniff.ts index 0d318dc82..baca02d0c 100644 --- a/packages/host/engine/src/attachment/mime-sniff.ts +++ b/packages/host/engine/src/attachment/mime-sniff.ts @@ -23,9 +23,19 @@ export function sniffImageMimeType(head: Uint8Array): SupportedAttachmentImageMi return undefined; } -/** A declared `image/*` type must match its bytes — model APIs refuse the mismatch later and - * less legibly. Other declarations have no reliable sniff and are trusted. */ +const SNIFFABLE_IMAGE_TYPES = new Set([ + 'image/jpeg', + 'image/png', + 'image/gif', + 'image/webp', +]); + +/** A declared sniffable `image/*` type must match its bytes — model APIs refuse the mismatch + * later and less legibly. Other declarations (svg, heic, pdf, …) have no reliable sniff here + * and are trusted. */ export function declaredMimeTypeMatches(declared: string, head: Uint8Array): boolean { if (!declared.startsWith('image/')) return true; - return sniffImageMimeType(head) === declared; + const sniffed = sniffImageMimeType(head); + if (sniffed !== undefined) return sniffed === declared; + return !SNIFFABLE_IMAGE_TYPES.has(declared); } diff --git a/packages/host/engine/src/engine.ts b/packages/host/engine/src/engine.ts index 9ec49d518..6dc387d9c 100644 --- a/packages/host/engine/src/engine.ts +++ b/packages/host/engine/src/engine.ts @@ -19,6 +19,7 @@ import { ManagedAssetService } from './asset/service'; import { InMemoryAttachmentStore } from './attachment/attachment-store'; import { FsBlobStore } from './attachment/blob-store'; import { AttachmentGc } from './attachment/gc'; +import { AttachmentIoMutex } from './attachment/io-mutex'; import { InMemoryLoopStore, InMemoryScheduleStore, @@ -128,7 +129,8 @@ export const createEngineRuntime = Effect.fn('Engine.create')(function* ( ? resourceStore.referencedAttachmentIds() : []), ]); - const attachmentGc = new AttachmentGc(attachmentStore, blobStore); + const attachmentIo = new AttachmentIoMutex(); + const attachmentGc = new AttachmentGc(attachmentStore, blobStore, Date.now, attachmentIo); const resources = new ResourceService( transport, resourceStore, @@ -137,6 +139,7 @@ export const createEngineRuntime = Effect.fn('Engine.create')(function* ( fileHost, blobStore, attachmentStore, + attachmentIo, ); const plugins = new PluginService(deps.pluginFactory ?? createPluginProviderAdapter); const translator = deps.translator; diff --git a/packages/host/engine/src/resource/service.ts b/packages/host/engine/src/resource/service.ts index 55806f428..0f6726c6b 100644 --- a/packages/host/engine/src/resource/service.ts +++ b/packages/host/engine/src/resource/service.ts @@ -16,6 +16,7 @@ import { Effect } from 'effect'; import { noop } from 'foxts/noop'; import type { AttachmentStore } from '../attachment/attachment-store'; import type { BlobStore } from '../attachment/blob-store'; +import { AttachmentIoMutex } from '../attachment/io-mutex'; import { declaredMimeTypeMatches } from '../attachment/mime-sniff'; import { OperationError, RequestError } from '../failure'; import type { FileHostService } from '../preview/file-host-service'; @@ -67,6 +68,7 @@ export class ResourceService { private readonly fileHost: FileHostService, private readonly blobs: BlobStore, private readonly attachments: AttachmentStore, + private readonly io: AttachmentIoMutex = new AttachmentIoMutex(), ) {} list(sessionId: SessionId): Effect.Effect { @@ -79,7 +81,7 @@ export class ResourceService { mimeType: string | undefined, data: string, ): Effect.Effect { - const { attachments, blobs, records, transport } = this; + const { attachments, blobs, io, records, transport } = this; return Effect.gen({ self: this }, function* () { if (!records.has(sessionId)) { return yield* new RequestError({ code: 'not_found', message: 'Session not found' }); @@ -102,44 +104,32 @@ export class ResourceService { const sha256 = createHash('sha256').update(bytes).digest('hex'); const blobId = blobIdFromSha256(sha256); const now = Date.now(); - // The harness reads the blob's own path: immutable, shared, mode 0444. - let resource: SessionResource = { - resourceId, - sessionId, - direction: 'source', - name, - kind: classify(name, mimeType), - status: 'processing', - locator: { type: 'managed-file', path: blobs.pathOf(blobId) }, - attachmentId, - mimeType, - sizeBytes: bytes.byteLength, - createdAt: now, - updatedAt: now, - }; - yield* this.run('save', () => this.store.save(resource)); - transport.send(createWireMessage({ kind: 'resource.changed', resource })); + const kind = classify(name, mimeType); + const locator = { type: 'managed-file' as const, path: blobs.pathOf(blobId) }; const written = yield* Effect.tryPromise({ async try() { - const stage = await blobs.stage(resourceId); - try { - await stage.write(0, bytes); - await stage.commit({ sha256, sizeBytes: bytes.byteLength }); - } catch (error) { - await stage.abort().catch(noop); - throw error; - } - await attachments.commitAttachment({ - blob: { blobId, sizeBytes: bytes.byteLength, createdAt: now }, - attachment: { - attachmentId, - kind: resource.kind, - name, - mimeType: mimeType ?? 'application/octet-stream', - sizeBytes: bytes.byteLength, - metadata: {}, - createdAt: now, - }, + await io.run(async () => { + const stage = await blobs.stage(resourceId); + try { + await stage.write(0, bytes); + await stage.commit({ sha256, sizeBytes: bytes.byteLength }); + await attachments.commitAttachment({ + blob: { blobId, sizeBytes: bytes.byteLength, createdAt: now }, + attachment: { + attachmentId, + kind, + name, + mimeType: mimeType ?? 'application/octet-stream', + sizeBytes: bytes.byteLength, + metadata: {}, + createdAt: now, + }, + }); + } catch (error) { + await stage.abort().catch(noop); + await blobs.delete(blobId); + throw error; + } }); }, catch: (cause) => cause, @@ -147,14 +137,36 @@ export class ResourceService { Effect.as(true), Effect.catch(() => Effect.succeed(false)), ); - resource = written - ? { ...resource, status: 'ready', updatedAt: Date.now() } - : { - ...resource, - status: 'failed', - error: 'Failed to persist uploaded resource', - updatedAt: Date.now(), - }; + if (!written) { + return { + resourceId, + sessionId, + direction: 'source', + name, + kind, + status: 'failed', + locator, + error: 'Failed to persist uploaded resource', + mimeType, + sizeBytes: bytes.byteLength, + createdAt: now, + updatedAt: Date.now(), + }; + } + const resource: SessionResource = { + resourceId, + sessionId, + direction: 'source', + name, + kind, + status: 'ready', + locator, + attachmentId, + mimeType, + sizeBytes: bytes.byteLength, + createdAt: now, + updatedAt: Date.now(), + }; yield* this.run('save', () => this.store.save(resource)); transport.send(createWireMessage({ kind: 'resource.changed', resource })); return resource;