diff --git a/packages/core/src/context-offload.ts b/packages/core/src/context-offload.ts index 165a1db2c3..77c9e225e7 100644 --- a/packages/core/src/context-offload.ts +++ b/packages/core/src/context-offload.ts @@ -53,16 +53,18 @@ export interface ContextOffloadLimits { readonly workspacePhysicalBytes: number; } +export type ContextOffloadPutFailureReason = + | 'too_large' + | 'session_quota_exceeded' + | 'workspace_quota_exceeded' + | 'identity_conflict' + | 'unavailable'; + export type ContextOffloadPutResult = | { readonly ok: true; readonly record: ContextOffloadRecord } | { readonly ok: false; - readonly reason: - | 'too_large' - | 'session_quota_exceeded' - | 'workspace_quota_exceeded' - | 'identity_conflict' - | 'unavailable'; + readonly reason: ContextOffloadPutFailureReason; }; export type ContextOffloadReadResult = @@ -106,6 +108,24 @@ export interface ContextOffloadGarbageCollectionResult { readonly hasMore: boolean; } +export class ReadImageSnapshotStoreError extends Error { + constructor(readonly reason: ContextOffloadPutFailureReason) { + super(`Read image snapshot storage failed: ${reason}`); + this.name = 'ReadImageSnapshotStoreError'; + } +} + +export interface ReadImageSnapshotStore { + snapshot(input: { + /** Stable identity of the Read result within its Session. */ + readonly ownerId: string; + readonly bytes: Uint8Array; + readonly mimeType: string; + }): Promise; + + read(input: SessionContextRef): Promise; +} + /** * Storage contract for capped, whole-object Agent context offload. * diff --git a/packages/storage/src/__tests__/fixtures/context-offload-managed-publication-crash-child.ts b/packages/storage/src/__tests__/fixtures/context-offload-managed-publication-crash-child.ts new file mode 100644 index 0000000000..398f0aa798 --- /dev/null +++ b/packages/storage/src/__tests__/fixtures/context-offload-managed-publication-crash-child.ts @@ -0,0 +1,55 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +import { join } from 'node:path'; +import { + CONTEXT_OFFLOAD_DATABASE_NAME, + SqliteContextOffloadStore, +} from '../../sqlite-context-offload-store.js'; + +const root = process.env.MAKA_CONTEXT_OFFLOAD_CRASH_ROOT; +if (!root) throw new Error('Missing context-offload crash fixture root'); + +const store = new SqliteContextOffloadStore(join(root, CONTEXT_OFFLOAD_DATABASE_NAME), { + limits: { + ownerMaxBytes: { + read_image_snapshot: 5 * 1024 * 1024, + tool_result_archive: 8 * 1024 * 1024, + }, + sessionLogicalBytes: 16 * 1024 * 1024, + workspacePhysicalBytes: 32 * 1024 * 1024, + }, + failpoint(point) { + const requested = process.env.MAKA_CONTEXT_OFFLOAD_CRASH_POINT ?? 'after_managed_file_publish'; + if (point === requested) process.exit(73); + }, +}); + +await store.put({ + sessionId: 'session-1', + owner: { + kind: 'read_image_snapshot', + ownerId: process.env.MAKA_CONTEXT_OFFLOAD_OWNER_ID ?? 'read-call-1', + }, + bytes: new TextEncoder().encode( + process.env.MAKA_CONTEXT_OFFLOAD_VALUE ?? 'crash-safe-managed-value', + ), + mediaType: 'image/png', +}); +throw new Error('Managed-file publication crash failpoint was not reached'); diff --git a/packages/storage/src/__tests__/read-image-snapshot-store.test.ts b/packages/storage/src/__tests__/read-image-snapshot-store.test.ts new file mode 100644 index 0000000000..c44d433773 --- /dev/null +++ b/packages/storage/src/__tests__/read-image-snapshot-store.test.ts @@ -0,0 +1,232 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +import assert from 'node:assert/strict'; +import { mkdtemp, rm } from 'node:fs/promises'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; +import { after, test } from 'node:test'; +import { MAX_READ_IMAGE_BYTES } from '@maka/core/attachments'; +import { ReadImageSnapshotStoreError, type ContextOffloadLimits } from '@maka/core/context-offload'; +import { + openInteractiveContextOffloadStoreForWrite, + type InteractiveContextOffloadWriter, +} from '../context-offload-store.js'; +import { createReadImageSnapshotStore } from '../read-image-snapshot-store.js'; +import { + resolveStorageRoot, + tryAcquireInteractiveRootOwner, + type InteractiveRootOwner, +} from '../root-authority.js'; +import { + removeTrackedControlDirectories, + trackControlDirectory, +} from './fixtures/control-directory-hygiene.js'; + +after(removeTrackedControlDirectories); + +test('derives Read image storage only from an authentic context writer', () => { + assert.throws( + () => createReadImageSnapshotStore({} as InteractiveContextOffloadWriter, 'session-1'), + /authentic interactive context-offload writer/u, + ); +}); + +test('snapshots one stable Read image identity and authorizes reads by Session', async () => { + await withReadImageStore(defaultLimits(), async (images, writer) => { + const bytes = new TextEncoder().encode('image'); + const input = { + ownerId: 'read-call-1', + bytes, + mimeType: 'image/png', + }; + const snapshotting = images.snapshot(input); + input.ownerId = 'mutated-owner'; + input.mimeType = 'image/jpeg'; + bytes.fill(0x78); + + const ref = await snapshotting; + assert.deepEqual(ref, { + kind: 'session_context', + sessionId: 'session-1', + refId: ref.refId, + }); + assert.deepEqual( + await images.snapshot({ + ownerId: 'read-call-1', + bytes: new TextEncoder().encode('image'), + mimeType: 'image/png', + }), + ref, + ); + const read = await images.read(ref); + assert.equal(read.ok, true); + if (!read.ok) return; + assert.equal(read.record.owner.kind, 'read_image_snapshot'); + assert.equal(read.record.owner.ownerId, 'read-call-1'); + assert.equal(read.record.mediaType, 'image/png'); + assert.deepEqual(read.bytes, new TextEncoder().encode('image')); + assert.deepEqual(await images.read({ ...ref, sessionId: 'session-2' }), { + ok: false, + reason: 'session_mismatch', + }); + const sessionTwoImages = createReadImageSnapshotStore(writer, 'session-2'); + assert.deepEqual(await sessionTwoImages.read(ref), { + ok: false, + reason: 'session_mismatch', + }); + + await assert.rejects( + images.snapshot({ + ownerId: 'read-call-1', + bytes: new TextEncoder().encode('changed'), + mimeType: 'image/png', + }), + (error) => + error instanceof ReadImageSnapshotStoreError && error.reason === 'identity_conflict', + ); + await assert.rejects( + images.snapshot({ + ownerId: 'read-call-1', + bytes: new TextEncoder().encode('image'), + mimeType: 'image/jpeg', + }), + (error) => + error instanceof ReadImageSnapshotStoreError && error.reason === 'identity_conflict', + ); + await assert.rejects( + images.snapshot({ + ownerId: 'not-an-image', + bytes: new Uint8Array([1]), + mimeType: 'text/plain', + }), + /media type must be an image/u, + ); + }); +}); + +test('maps configured quota failures and rejects non-image owner references', async () => { + await withReadImageStore( + { + ownerMaxBytes: { + read_image_snapshot: 4, + tool_result_archive: 64, + }, + sessionLogicalBytes: 64, + workspacePhysicalBytes: 64, + }, + async (images, writer) => { + await assert.rejects( + images.snapshot({ + ownerId: 'over-configured-limit', + bytes: new Uint8Array(5), + mimeType: 'image/png', + }), + (error) => error instanceof ReadImageSnapshotStoreError && error.reason === 'too_large', + ); + + const archive = await writer.put({ + sessionId: 'session-1', + owner: { kind: 'tool_result_archive', ownerId: 'archive-1' }, + bytes: new TextEncoder().encode('{}'), + mediaType: 'application/json', + }); + assert.equal(archive.ok, true); + if (!archive.ok) return; + assert.deepEqual( + await images.read({ + kind: 'session_context', + sessionId: 'session-1', + refId: archive.record.refId, + }), + { ok: false, reason: 'corrupt' }, + ); + }, + ); +}); + +test('enforces the Read image product cap before touching storage', async () => { + const limits: ContextOffloadLimits = { + ownerMaxBytes: { + read_image_snapshot: MAX_READ_IMAGE_BYTES + 1, + tool_result_archive: 64, + }, + sessionLogicalBytes: MAX_READ_IMAGE_BYTES + 1, + workspacePhysicalBytes: MAX_READ_IMAGE_BYTES + 1, + }; + await withReadImageStore(limits, async (images, writer) => { + await assert.rejects( + images.snapshot({ + ownerId: 'over-product-limit', + bytes: new Uint8Array(MAX_READ_IMAGE_BYTES + 1), + mimeType: 'image/png', + }), + (error) => error instanceof ReadImageSnapshotStoreError && error.reason === 'too_large', + ); + assert.deepEqual(await writer.usage(), { + references: 0, + logicalBytes: 0, + physicalBytes: 0, + }); + }); +}); + +function defaultLimits(): ContextOffloadLimits { + return { + ownerMaxBytes: { + read_image_snapshot: MAX_READ_IMAGE_BYTES, + tool_result_archive: 64, + }, + sessionLogicalBytes: MAX_READ_IMAGE_BYTES * 2, + workspacePhysicalBytes: MAX_READ_IMAGE_BYTES * 2, + }; +} + +async function withReadImageStore( + limits: ContextOffloadLimits, + run: ( + images: ReturnType, + writer: InteractiveContextOffloadWriter, + ) => Promise, +): Promise { + await withInteractiveOwner(async (owner) => { + const writer = await openInteractiveContextOffloadStoreForWrite(owner.lease, { limits }); + try { + await run(createReadImageSnapshotStore(writer, 'session-1'), writer); + } finally { + await writer.close(); + } + }); +} + +async function withInteractiveOwner(run: (owner: InteractiveRootOwner) => Promise) { + const root = await mkdtemp(join(tmpdir(), 'maka-read-image-context-store-')); + const capability = trackControlDirectory( + await resolveStorageRoot({ path: root, kind: 'interactive' }), + ); + const owner = await tryAcquireInteractiveRootOwner(capability); + assert.ok(owner); + if (!owner) return; + try { + await run(owner); + } finally { + await owner.close(); + await rm(root, { recursive: true, force: true }); + } +} diff --git a/packages/storage/src/__tests__/sqlite-context-offload-store.test.ts b/packages/storage/src/__tests__/sqlite-context-offload-store.test.ts index deb64e40f7..d590997080 100644 --- a/packages/storage/src/__tests__/sqlite-context-offload-store.test.ts +++ b/packages/storage/src/__tests__/sqlite-context-offload-store.test.ts @@ -18,25 +18,34 @@ */ import assert from 'node:assert/strict'; +import { execFile } from 'node:child_process'; import { createHash } from 'node:crypto'; -import { mkdtemp, rm } from 'node:fs/promises'; +import { mkdtemp, readFile, readdir, rm, stat, symlink, unlink, writeFile } from 'node:fs/promises'; import { tmpdir } from 'node:os'; import { join } from 'node:path'; import { DatabaseSync } from 'node:sqlite'; +import { fileURLToPath } from 'node:url'; +import { promisify } from 'node:util'; import test, { type TestContext } from 'node:test'; import type { ContextOffloadLimits } from '@maka/core/context-offload'; import { CONTEXT_OFFLOAD_DATABASE_NAME, + CONTEXT_OFFLOAD_VALUES_DIRECTORY_NAME, SqliteContextOffloadStore, } from '../sqlite-context-offload-store.js'; +const execFileAsync = promisify(execFile); +const managedPublicationCrashChild = fileURLToPath( + new URL('./fixtures/context-offload-managed-publication-crash-child.js', import.meta.url), +); + test('creates the dedicated WAL schema with incremental auto-vacuum', async (t) => { const fixture = await createFixture(t); fixture.store.close(); const database = new DatabaseSync(fixture.path); t.after(() => database.close()); - assert.equal(pragmaNumber(database, 'user_version'), 2); + assert.equal(pragmaNumber(database, 'user_version'), 3); assert.equal(pragmaNumber(database, 'auto_vacuum'), 2); assert.equal(pragmaText(database, 'journal_mode'), 'wal'); assert.deepEqual( @@ -50,6 +59,7 @@ test('creates the dedicated WAL schema with incremental auto-vacuum', async (t) .map((row) => row.name), [ 'context_blobs', + 'context_file_deletions', 'context_gc_candidates', 'context_refs', 'context_session_usage', @@ -109,6 +119,329 @@ test('atomically persists one idempotent owner identity and verifies reads', asy ok: false, reason: 'identity_conflict', }); + assert.deepEqual(await fixture.store.put({ ...input, mediaType: 'image/jpeg' }), { + ok: false, + reason: 'identity_conflict', + }); +}); + +test('stores managed binary values as durable file locators instead of SQLite payloads', async (t) => { + const fixture = await createFixture(t); + const bytes = new Uint8Array(1_024).fill(0x5a); + const blobId = sha256(bytes); + const stored = await fixture.store.put({ + sessionId: 'session-1', + owner: { kind: 'read_image_snapshot', ownerId: 'read-call-1' }, + bytes, + mediaType: 'image/png', + }); + assert.equal(stored.ok, true); + if (!stored.ok) return; + + const database = new DatabaseSync(fixture.path); + const row = database + .prepare('SELECT storage_kind, payload, size_bytes FROM context_blobs WHERE blob_id = ?') + .get(Buffer.from(blobId, 'hex')) as { + storage_kind: string; + payload: Uint8Array; + size_bytes: number; + }; + assert.equal( + ( + database.prepare('SELECT COUNT(*) AS count FROM context_file_deletions').get() as { + count: number; + } + ).count, + 0, + ); + database.close(); + const locator = Buffer.from(row.payload).toString('utf8'); + assert.equal(row.storage_kind, 'managed_file'); + assert.equal(row.size_bytes, bytes.byteLength); + assert.equal(locator, `sha256/${blobId.slice(0, 2)}/${blobId}`); + assert.ok(row.payload.byteLength < bytes.byteLength); + + const valuePath = join(fixture.root, CONTEXT_OFFLOAD_VALUES_DIRECTORY_NAME, locator); + assert.deepEqual(new Uint8Array(await readFile(valuePath)), bytes); + assert.equal((await stat(valuePath)).isFile(), true); + assert.deepEqual( + await fixture.store.read({ + sessionId: 'session-1', + refId: stored.record.refId, + maxBytes: bytes.byteLength, + }), + { ok: true, record: stored.record, bytes }, + ); + + await writeFile(valuePath, new Uint8Array(bytes.byteLength)); + assert.deepEqual( + await fixture.store.read({ + sessionId: 'session-1', + refId: stored.record.refId, + maxBytes: bytes.byteLength, + }), + { ok: false, reason: 'corrupt' }, + ); + + await fixture.store.releaseReference({ + sessionId: 'session-1', + refId: stored.record.refId, + }); + assert.deepEqual( + await fixture.store.collectGarbage({ + olderThan: 1_001, + maxBlobs: 1, + maxBytes: bytes.byteLength, + }), + { deletedBlobs: 1, deletedBytes: bytes.byteLength, hasMore: false }, + ); + await assert.rejects(stat(valuePath), (error) => isNodeError(error, 'ENOENT')); + const afterGc = new DatabaseSync(fixture.path); + assert.equal( + ( + afterGc.prepare('SELECT COUNT(*) AS count FROM context_file_deletions').get() as { + count: number; + } + ).count, + 0, + ); + afterGc.close(); +}); + +test('repairs a missing managed blob when an inline owner retries identical bytes', async (t) => { + const fixture = await createFixture(t); + const bytes = new TextEncoder().encode('shared-value'); + const tool = await fixture.store.put({ + sessionId: 'session-1', + owner: { kind: 'tool_result_archive', ownerId: 'tool-1' }, + bytes, + mediaType: 'application/octet-stream', + }); + const image = await fixture.store.put({ + sessionId: 'session-1', + owner: { kind: 'read_image_snapshot', ownerId: 'read-1' }, + bytes, + mediaType: 'image/png', + }); + assert.equal(tool.ok, true); + assert.equal(image.ok, true); + if (!tool.ok || !image.ok) return; + + const blobId = sha256(bytes); + const valuePath = join( + fixture.root, + CONTEXT_OFFLOAD_VALUES_DIRECTORY_NAME, + `sha256/${blobId.slice(0, 2)}/${blobId}`, + ); + await unlink(valuePath); + + assert.deepEqual( + await fixture.store.put({ + sessionId: 'session-1', + owner: { kind: 'tool_result_archive', ownerId: 'tool-1' }, + bytes, + mediaType: 'application/octet-stream', + }), + tool, + ); + assert.deepEqual( + await fixture.store.read({ + sessionId: 'session-1', + refId: tool.record.refId, + maxBytes: bytes.byteLength, + }), + { ok: true, record: tool.record, bytes }, + ); +}); + +test('removes managed publication state when quota admission fails', async (t) => { + const fixture = await createFixture(t, { + ownerMaxBytes: TEST_OWNER_MAX_BYTES, + sessionLogicalBytes: 64, + workspacePhysicalBytes: 0, + }); + const bytes = new TextEncoder().encode('over-quota'); + const blobId = sha256(bytes); + assert.deepEqual( + await fixture.store.put({ + sessionId: 'session-1', + owner: { kind: 'read_image_snapshot', ownerId: 'read-call-1' }, + bytes, + mediaType: 'image/png', + }), + { ok: false, reason: 'workspace_quota_exceeded' }, + ); + await assert.rejects( + stat( + join( + fixture.root, + CONTEXT_OFFLOAD_VALUES_DIRECTORY_NAME, + `sha256/${blobId.slice(0, 2)}/${blobId}`, + ), + ), + (error) => isNodeError(error, 'ENOENT'), + ); + const database = new DatabaseSync(fixture.path); + assert.equal( + ( + database.prepare('SELECT COUNT(*) AS count FROM context_file_deletions').get() as { + count: number; + } + ).count, + 0, + ); + database.close(); +}); + +test('recovers a durable managed-file publication intent after process exit', async (t) => { + const root = await mkdtemp(join(tmpdir(), 'maka-context-offload-publication-crash-')); + t.after(() => rm(root, { recursive: true, force: true })); + await assert.rejects( + execFileAsync(process.execPath, [managedPublicationCrashChild], { + env: { ...process.env, MAKA_CONTEXT_OFFLOAD_CRASH_ROOT: root, NODE_NO_WARNINGS: '1' }, + windowsHide: true, + }), + (error: unknown) => error instanceof Error && 'code' in error && Number(error.code) === 73, + ); + + const bytes = new TextEncoder().encode('crash-safe-managed-value'); + const blobId = sha256(bytes); + const locator = `sha256/${blobId.slice(0, 2)}/${blobId}`; + const valuePath = join(root, CONTEXT_OFFLOAD_VALUES_DIRECTORY_NAME, locator); + assert.deepEqual(new Uint8Array(await readFile(valuePath)), bytes); + const path = join(root, CONTEXT_OFFLOAD_DATABASE_NAME); + const crashed = new DatabaseSync(path); + assert.equal( + ( + crashed + .prepare('SELECT COUNT(*) AS count FROM context_file_deletions WHERE locator = ?') + .get(Buffer.from(locator, 'utf8')) as { count: number } + ).count, + 1, + ); + assert.equal( + (crashed.prepare('SELECT COUNT(*) AS count FROM context_blobs').get() as { count: number }) + .count, + 0, + ); + crashed.close(); + + const recovered = new SqliteContextOffloadStore(path, { limits: defaultLimits() }); + t.after(() => recovered.close()); + assert.deepEqual( + await recovered.collectGarbage({ olderThan: 1, maxBlobs: 1, maxBytes: bytes.byteLength }), + { deletedBlobs: 0, deletedBytes: 0, hasMore: false }, + ); + await assert.rejects(stat(valuePath), (error) => isNodeError(error, 'ENOENT')); + const afterRecovery = new DatabaseSync(path); + assert.equal( + ( + afterRecovery.prepare('SELECT COUNT(*) AS count FROM context_file_deletions').get() as { + count: number; + } + ).count, + 0, + ); + afterRecovery.close(); +}); + +test('recovers deterministic managed-file staging after process exit', async (t) => { + const root = await mkdtemp(join(tmpdir(), 'maka-context-offload-staging-crash-')); + t.after(() => rm(root, { recursive: true, force: true })); + await assert.rejects( + execFileAsync(process.execPath, [managedPublicationCrashChild], { + env: { + ...process.env, + MAKA_CONTEXT_OFFLOAD_CRASH_ROOT: root, + MAKA_CONTEXT_OFFLOAD_CRASH_POINT: 'after_managed_file_staging', + NODE_NO_WARNINGS: '1', + }, + windowsHide: true, + }), + (error: unknown) => error instanceof Error && 'code' in error && Number(error.code) === 73, + ); + + const bytes = new TextEncoder().encode('crash-safe-managed-value'); + const blobId = sha256(bytes); + const directory = join( + root, + CONTEXT_OFFLOAD_VALUES_DIRECTORY_NAME, + `sha256/${blobId.slice(0, 2)}`, + ); + const stagingPath = join(directory, `.${blobId}.publish.tmp`); + assert.deepEqual(new Uint8Array(await readFile(stagingPath)), bytes); + + const path = join(root, CONTEXT_OFFLOAD_DATABASE_NAME); + const recovered = new SqliteContextOffloadStore(path, { limits: defaultLimits() }); + t.after(() => recovered.close()); + assert.deepEqual(await recovered.usage(), { + references: 0, + logicalBytes: 0, + physicalBytes: bytes.byteLength, + }); + assert.deepEqual( + await recovered.collectGarbage({ olderThan: 1, maxBlobs: 1, maxBytes: bytes.byteLength }), + { deletedBlobs: 0, deletedBytes: 0, hasMore: false }, + ); + await assert.rejects(stat(stagingPath), (error) => isNodeError(error, 'ENOENT')); + assert.equal((await recovered.usage()).physicalBytes, 0); +}); + +test('reports continuation while pending managed-file deletions remain', async (t) => { + const root = await mkdtemp(join(tmpdir(), 'maka-context-offload-pending-files-')); + t.after(() => rm(root, { recursive: true, force: true })); + const values = ['pending-one', 'pending-two', 'pending-three']; + for (const [index, value] of values.entries()) { + await assert.rejects( + execFileAsync(process.execPath, [managedPublicationCrashChild], { + env: { + ...process.env, + MAKA_CONTEXT_OFFLOAD_CRASH_ROOT: root, + MAKA_CONTEXT_OFFLOAD_CRASH_POINT: 'after_managed_file_staging', + MAKA_CONTEXT_OFFLOAD_OWNER_ID: `read-${index}`, + MAKA_CONTEXT_OFFLOAD_VALUE: value, + NODE_NO_WARNINGS: '1', + }, + windowsHide: true, + }), + (error: unknown) => error instanceof Error && 'code' in error && Number(error.code) === 73, + ); + } + + const path = join(root, CONTEXT_OFFLOAD_DATABASE_NAME); + const recovered = new SqliteContextOffloadStore(path, { limits: defaultLimits() }); + t.after(() => recovered.close()); + const totalBytes = values.reduce((total, value) => total + Buffer.byteLength(value), 0); + assert.equal((await recovered.usage()).physicalBytes, totalBytes); + for (const hasMore of [true, true, false]) { + assert.deepEqual( + await recovered.collectGarbage({ olderThan: 1, maxBlobs: 1, maxBytes: totalBytes }), + { deletedBlobs: 0, deletedBytes: 0, hasMore }, + ); + } + assert.equal((await recovered.usage()).physicalBytes, 0); +}); + +test('rejects a managed-value directory that resolves outside the Storage Root', async (t) => { + const fixture = await createFixture(t); + const outside = await mkdtemp(join(tmpdir(), 'maka-context-offload-outside-')); + t.after(() => rm(outside, { recursive: true, force: true })); + await symlink( + outside, + join(fixture.root, CONTEXT_OFFLOAD_VALUES_DIRECTORY_NAME), + process.platform === 'win32' ? 'junction' : 'dir', + ); + + assert.deepEqual( + await fixture.store.put({ + sessionId: 'session-1', + owner: { kind: 'read_image_snapshot', ownerId: 'read-call-1' }, + bytes: new TextEncoder().encode('image'), + mediaType: 'image/png', + }), + { ok: false, reason: 'unavailable' }, + ); + assert.deepEqual(await readdir(outside), []); }); test('reopens durable records and preserves owner idempotency', async (t) => { @@ -141,7 +474,7 @@ test('rejects a database schema newer than this authority understands', async (t const path = join(root, CONTEXT_OFFLOAD_DATABASE_NAME); t.after(() => rm(root, { recursive: true, force: true })); const database = new DatabaseSync(path); - database.exec('PRAGMA user_version = 3'); + database.exec('PRAGMA user_version = 4'); database.close(); assert.throws( @@ -153,7 +486,7 @@ test('rejects a database schema newer than this authority understands', async (t workspacePhysicalBytes: 1, }, }), - /schema 3 is newer than supported version 2/u, + /schema 4 is newer than supported version 3/u, ); }); @@ -367,7 +700,7 @@ test('releases only the authorized Session reference without deleting shared byt test('copies references atomically without copying physical bytes', async (t) => { const fixture = await createFixture(t, { ownerMaxBytes: TEST_OWNER_MAX_BYTES, - sessionLogicalBytes: 12, + sessionLogicalBytes: 16, workspacePhysicalBytes: 16, }); const first = await fixture.store.put( @@ -376,9 +709,14 @@ test('copies references atomically without copying physical bytes', async (t) => const second = await fixture.store.put( putInput('source', 'source-2', new TextEncoder().encode('second')), ); + const third = await fixture.store.put({ + ...putInput('source', 'source-3', new TextEncoder().encode('first')), + mediaType: 'text/plain', + }); assert.equal(first.ok, true); assert.equal(second.ok, true); - if (!first.ok || !second.ok) return; + assert.equal(third.ok, true); + if (!first.ok || !second.ok || !third.ok) return; const copyInput = { sourceSessionId: 'source', @@ -420,7 +758,7 @@ test('copies references atomically without copying physical bytes', async (t) => targetSessionId: 'target', references: [ { - sourceRefId: first.record.refId, + sourceRefId: second.record.refId, targetOwner: { kind: 'tool_result_archive', ownerId: 'target-over-quota' }, }, ], @@ -445,7 +783,38 @@ test('copies references atomically without copying physical bytes', async (t) => }), { ok: false, reason: 'identity_conflict' }, ); + assert.deepEqual( + await fixture.store.copyReferences({ + sourceSessionId: 'source', + targetSessionId: 'target', + references: [ + { + sourceRefId: third.record.refId, + targetOwner: { kind: 'tool_result_archive', ownerId: 'target-1' }, + }, + ], + }), + { ok: false, reason: 'identity_conflict' }, + ); + assert.deepEqual( + await fixture.store.copyReferences({ + sourceSessionId: 'source', + targetSessionId: 'mime-conflict-target', + references: [ + { + sourceRefId: first.record.refId, + targetOwner: { kind: 'tool_result_archive', ownerId: 'target-1' }, + }, + { + sourceRefId: third.record.refId, + targetOwner: { kind: 'tool_result_archive', ownerId: 'target-1' }, + }, + ], + }), + { ok: false, reason: 'identity_conflict' }, + ); assert.equal((await fixture.store.usage('target')).references, 2); + assert.equal((await fixture.store.usage('mime-conflict-target')).references, 0); }); test('retires only one Session and collects shared blobs after the last reference', async (t) => { @@ -532,7 +901,9 @@ test('migrates v1 orphan blobs into the indexed garbage candidate set', async (t fixture.store.close(); const database = new DatabaseSync(fixture.path); - database.exec('DROP TABLE context_gc_candidates; PRAGMA user_version = 1'); + database.exec( + 'DROP TABLE context_gc_candidates; DROP TABLE context_file_deletions; PRAGMA user_version = 1', + ); database.close(); const migrated = new SqliteContextOffloadStore(fixture.path, { limits: fixture.limits }); t.after(() => migrated.close()); @@ -573,6 +944,16 @@ test('lifecycle queries use Session and garbage eligibility indexes', async (t) .all(1_001, 2); assert.match(JSON.stringify(garbagePlan), /context_gc_candidates_eligible/u); assert.doesNotMatch(JSON.stringify(garbagePlan), /SCAN b(?:\W|$)/u); + + const fileDeletionPlan = database + .prepare( + `EXPLAIN QUERY PLAN + SELECT locator FROM context_file_deletions + ORDER BY enqueued_at, locator + LIMIT ?`, + ) + .all(2); + assert.match(JSON.stringify(fileDeletionPlan), /context_file_deletions_pending/u); }); function putInput(sessionId: string, ownerId: string, bytes: Uint8Array) { @@ -606,7 +987,7 @@ async function createFixture( store.close(); await rm(root, { recursive: true, force: true }); }); - return { limits, path, store }; + return { limits, path, root, store }; } const TEST_OWNER_MAX_BYTES = Object.freeze({ @@ -614,10 +995,22 @@ const TEST_OWNER_MAX_BYTES = Object.freeze({ tool_result_archive: 8 * 1024 * 1024, }); +function defaultLimits(): ContextOffloadLimits { + return { + ownerMaxBytes: TEST_OWNER_MAX_BYTES, + sessionLogicalBytes: 16 * 1024 * 1024, + workspacePhysicalBytes: 32 * 1024 * 1024, + }; +} + function sha256(bytes: Uint8Array): string { return createHash('sha256').update(bytes).digest('hex'); } +function isNodeError(error: unknown, code: string): error is NodeJS.ErrnoException { + return error instanceof Error && 'code' in error && error.code === code; +} + function pragmaNumber(database: DatabaseSync, name: string): number { const row = database.prepare(`PRAGMA ${name}`).get() as Record; const value = row[name]; diff --git a/packages/storage/src/read-image-snapshot-store.ts b/packages/storage/src/read-image-snapshot-store.ts new file mode 100644 index 0000000000..ab1c73aa8f --- /dev/null +++ b/packages/storage/src/read-image-snapshot-store.ts @@ -0,0 +1,94 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +import { MAX_READ_IMAGE_BYTES } from '@maka/core/attachments'; +import { + ReadImageSnapshotStoreError, + type ContextOffloadReadResult, + type ReadImageSnapshotStore, +} from '@maka/core/context-offload'; +import { + authenticateInteractiveContextOffloadWriter, + type InteractiveContextOffloadWriter, +} from './context-offload-store.js'; + +/** Derives the Read image domain contract from the authenticated byte authority. */ +export function createReadImageSnapshotStore( + writer: InteractiveContextOffloadWriter, + sessionId: string, +): ReadImageSnapshotStore { + const store = authenticateInteractiveContextOffloadWriter(writer); + if (!sessionId) throw new Error('Read image snapshot Session id is required'); + const facade: ReadImageSnapshotStore = { + async snapshot(input) { + const accepted = Object.freeze({ + sessionId, + ownerId: input.ownerId, + bytes: new Uint8Array(input.bytes), + mimeType: input.mimeType, + }); + if (!accepted.mimeType.toLowerCase().startsWith('image/')) { + throw new Error('Read image snapshot media type must be an image'); + } + if (accepted.bytes.byteLength > MAX_READ_IMAGE_BYTES) { + throw new ReadImageSnapshotStoreError('too_large'); + } + const result = await store.put({ + sessionId: accepted.sessionId, + owner: { kind: 'read_image_snapshot', ownerId: accepted.ownerId }, + bytes: accepted.bytes, + mediaType: accepted.mimeType, + }); + if (!result.ok) throw new ReadImageSnapshotStoreError(result.reason); + const { record } = result; + if ( + record.sessionId !== accepted.sessionId || + record.owner.kind !== 'read_image_snapshot' || + record.owner.ownerId !== accepted.ownerId || + record.mediaType !== accepted.mimeType || + record.sizeBytes !== accepted.bytes.byteLength + ) { + throw new Error('Read image snapshot authority returned an inconsistent reference'); + } + return Object.freeze({ + kind: 'session_context', + sessionId: record.sessionId, + refId: record.refId, + }); + }, + + async read(input): Promise { + if (input.sessionId !== sessionId) return { ok: false, reason: 'session_mismatch' }; + const result = await store.read({ + sessionId, + refId: input.refId, + maxBytes: MAX_READ_IMAGE_BYTES, + }); + if (!result.ok) return result; + if ( + result.record.owner.kind !== 'read_image_snapshot' || + !result.record.mediaType.toLowerCase().startsWith('image/') + ) { + return { ok: false, reason: 'corrupt' }; + } + return result; + }, + }; + return Object.freeze(facade); +} diff --git a/packages/storage/src/sqlite-context-offload-schema.ts b/packages/storage/src/sqlite-context-offload-schema.ts index 5f3495474f..5558fd5602 100644 --- a/packages/storage/src/sqlite-context-offload-schema.ts +++ b/packages/storage/src/sqlite-context-offload-schema.ts @@ -19,7 +19,7 @@ import type { DatabaseSync } from 'node:sqlite'; -export const SQLITE_CONTEXT_OFFLOAD_SCHEMA_VERSION = 2; +export const SQLITE_CONTEXT_OFFLOAD_SCHEMA_VERSION = 3; const SQLITE_INITIALIZATION_BUSY_TIMEOUT_MS = 5_000; const SQLITE_INITIALIZATION_RETRY_DELAY_MS = 10; const initializationRetryGate = new Int32Array(new SharedArrayBuffer(4)); @@ -27,9 +27,14 @@ const initializationRetryGate = new Int32Array(new SharedArrayBuffer(4)); const INITIAL_SCHEMA = ` CREATE TABLE context_blobs ( blob_id BLOB PRIMARY KEY CHECK(length(blob_id) = 32), + storage_kind TEXT NOT NULL CHECK(storage_kind IN ('inline', 'managed_file')), payload BLOB NOT NULL, - size_bytes INTEGER NOT NULL CHECK(size_bytes >= 0 AND length(payload) = size_bytes), - created_at INTEGER NOT NULL CHECK(created_at >= 0) + size_bytes INTEGER NOT NULL CHECK(size_bytes >= 0), + created_at INTEGER NOT NULL CHECK(created_at >= 0), + CHECK( + (storage_kind = 'inline' AND length(payload) = size_bytes) OR + (storage_kind = 'managed_file' AND length(payload) BETWEEN 1 AND 512) + ) ); CREATE TABLE context_refs ( @@ -60,6 +65,15 @@ const INITIAL_SCHEMA = ` CREATE INDEX context_gc_candidates_eligible ON context_gc_candidates(unreferenced_at, blob_id); + CREATE TABLE context_file_deletions ( + locator BLOB PRIMARY KEY CHECK(length(locator) BETWEEN 1 AND 512), + size_bytes INTEGER NOT NULL CHECK(size_bytes >= 0), + enqueued_at INTEGER NOT NULL CHECK(enqueued_at >= 0) + ); + + CREATE INDEX context_file_deletions_pending + ON context_file_deletions(enqueued_at, locator); + CREATE TABLE context_session_usage ( session_id TEXT PRIMARY KEY, reference_count INTEGER NOT NULL CHECK(reference_count >= 0), @@ -94,19 +108,90 @@ const SCHEMA_V2_MIGRATION = ` ); `; +const SCHEMA_V3_MIGRATION = ` + CREATE TABLE context_blobs_v3 ( + blob_id BLOB PRIMARY KEY CHECK(length(blob_id) = 32), + storage_kind TEXT NOT NULL CHECK(storage_kind IN ('inline', 'managed_file')), + payload BLOB NOT NULL, + size_bytes INTEGER NOT NULL CHECK(size_bytes >= 0), + created_at INTEGER NOT NULL CHECK(created_at >= 0), + CHECK( + (storage_kind = 'inline' AND length(payload) = size_bytes) OR + (storage_kind = 'managed_file' AND length(payload) BETWEEN 1 AND 512) + ) + ); + + CREATE TABLE context_refs_v3 ( + ref_id TEXT PRIMARY KEY, + session_id TEXT NOT NULL, + owner_kind TEXT NOT NULL CHECK( + owner_kind IN ('read_image_snapshot', 'tool_result_archive') + ), + owner_id TEXT NOT NULL, + blob_id BLOB NOT NULL REFERENCES context_blobs_v3(blob_id) ON DELETE RESTRICT, + media_type TEXT NOT NULL, + created_at INTEGER NOT NULL CHECK(created_at >= 0), + UNIQUE(session_id, owner_kind, owner_id) + ); + + CREATE TABLE context_gc_candidates_v3 ( + blob_id BLOB PRIMARY KEY + REFERENCES context_blobs_v3(blob_id) ON DELETE CASCADE, + unreferenced_at INTEGER NOT NULL CHECK(unreferenced_at >= 0) + ); + + INSERT INTO context_blobs_v3(blob_id, storage_kind, payload, size_bytes, created_at) + SELECT blob_id, 'inline', payload, size_bytes, created_at FROM context_blobs; + + INSERT INTO context_refs_v3( + ref_id, session_id, owner_kind, owner_id, blob_id, media_type, created_at + ) + SELECT ref_id, session_id, owner_kind, owner_id, blob_id, media_type, created_at + FROM context_refs; + + INSERT INTO context_gc_candidates_v3(blob_id, unreferenced_at) + SELECT blob_id, unreferenced_at FROM context_gc_candidates; + + DROP TABLE context_refs; + DROP TABLE context_gc_candidates; + DROP TABLE context_blobs; + + ALTER TABLE context_blobs_v3 RENAME TO context_blobs; + ALTER TABLE context_refs_v3 RENAME TO context_refs; + ALTER TABLE context_gc_candidates_v3 RENAME TO context_gc_candidates; + + CREATE INDEX context_refs_session + ON context_refs(session_id, created_at, ref_id); + CREATE INDEX context_refs_blob + ON context_refs(blob_id); + CREATE INDEX context_gc_candidates_eligible + ON context_gc_candidates(unreferenced_at, blob_id); + + CREATE TABLE context_file_deletions ( + locator BLOB PRIMARY KEY CHECK(length(locator) BETWEEN 1 AND 512), + size_bytes INTEGER NOT NULL CHECK(size_bytes >= 0), + enqueued_at INTEGER NOT NULL CHECK(enqueued_at >= 0) + ); + + CREATE INDEX context_file_deletions_pending + ON context_file_deletions(enqueued_at, locator); +`; + const REQUIRED_SCHEMA_OBJECTS = Object.freeze([ ['table', 'context_blobs'], ['table', 'context_refs'], ['table', 'context_session_usage'], ['table', 'context_store_usage'], ['table', 'context_gc_candidates'], + ['table', 'context_file_deletions'], ['index', 'context_refs_session'], ['index', 'context_refs_blob'], ['index', 'context_gc_candidates_eligible'], + ['index', 'context_file_deletions_pending'], ] as const); const REQUIRED_TABLE_COLUMNS = Object.freeze({ - context_blobs: ['blob_id', 'payload', 'size_bytes', 'created_at'], + context_blobs: ['blob_id', 'storage_kind', 'payload', 'size_bytes', 'created_at'], context_refs: [ 'ref_id', 'session_id', @@ -119,12 +204,14 @@ const REQUIRED_TABLE_COLUMNS = Object.freeze({ context_session_usage: ['session_id', 'reference_count', 'logical_bytes'], context_store_usage: ['singleton', 'blob_count', 'physical_bytes'], context_gc_candidates: ['blob_id', 'unreferenced_at'], + context_file_deletions: ['locator', 'size_bytes', 'enqueued_at'], } as const); const REQUIRED_INDEX_COLUMNS = Object.freeze({ context_refs_session: ['session_id', 'created_at', 'ref_id'], context_refs_blob: ['blob_id'], context_gc_candidates_eligible: ['unreferenced_at', 'blob_id'], + context_file_deletions_pending: ['enqueued_at', 'locator'], } as const); export function configureSqliteContextOffloadDatabase(db: DatabaseSync): void { @@ -152,7 +239,7 @@ export function migrateSqliteContextOffloadDatabase(db: DatabaseSync): void { db.exec('BEGIN IMMEDIATE'); try { - const current = readSqliteContextOffloadSchemaVersion(db); + let current = readSqliteContextOffloadSchemaVersion(db); if (current > SQLITE_CONTEXT_OFFLOAD_SCHEMA_VERSION) throw newerSchemaError(current); if (current === 0) { if (hasApplicationSchemaObjects(db)) { @@ -160,10 +247,17 @@ export function migrateSqliteContextOffloadDatabase(db: DatabaseSync): void { } db.exec(INITIAL_SCHEMA); db.exec(`PRAGMA user_version = ${SQLITE_CONTEXT_OFFLOAD_SCHEMA_VERSION}`); + current = SQLITE_CONTEXT_OFFLOAD_SCHEMA_VERSION; } if (current === 1) { db.exec(SCHEMA_V2_MIGRATION); + db.exec('PRAGMA user_version = 2'); + current = 2; + } + if (current === 2) { + db.exec(SCHEMA_V3_MIGRATION); db.exec(`PRAGMA user_version = ${SQLITE_CONTEXT_OFFLOAD_SCHEMA_VERSION}`); + current = SQLITE_CONTEXT_OFFLOAD_SCHEMA_VERSION; } validateSchema(db); db.exec('COMMIT'); diff --git a/packages/storage/src/sqlite-context-offload-store.ts b/packages/storage/src/sqlite-context-offload-store.ts index e4f6be2d1e..5b96a75143 100644 --- a/packages/storage/src/sqlite-context-offload-store.ts +++ b/packages/storage/src/sqlite-context-offload-store.ts @@ -18,9 +18,10 @@ */ import { createHash, randomUUID } from 'node:crypto'; -import { mkdirSync } from 'node:fs'; +import { mkdirSync, realpathSync } from 'node:fs'; +import { link, lstat, mkdir, open, realpath, unlink } from 'node:fs/promises'; import { createRequire } from 'node:module'; -import { dirname } from 'node:path'; +import { dirname, isAbsolute, join, relative, sep } from 'node:path'; import type { DatabaseSync } from 'node:sqlite'; import { type ContextOffloadCopyResult, @@ -38,17 +39,27 @@ import { configureSqliteContextOffloadDatabase, migrateSqliteContextOffloadDatabase, } from './sqlite-context-offload-schema.js'; +import { + readStableBoundedFile, + syncDirectory, + syncDirectoryChain, + syncFile, +} from './stable-storage.js'; const MAX_ID_CODE_POINTS = 512; const MAX_MEDIA_TYPE_CODE_POINTS = 256; const SHA256_PATTERN = /^[0-9a-f]{64}$/; +const MANAGED_FILE_LOCATOR_PATTERN = /^sha256\/([0-9a-f]{2})\/([0-9a-f]{64})$/; const require = createRequire(import.meta.url); export const CONTEXT_OFFLOAD_DATABASE_NAME = 'context-offload.sqlite'; +export const CONTEXT_OFFLOAD_VALUES_DIRECTORY_NAME = 'context-offload-values'; export type SqliteContextOffloadStoreFailpoint = | 'after_blob_insert' | 'after_ref_insert' + | 'after_managed_file_staging' + | 'after_managed_file_publish' | 'after_gc_blob_delete'; export interface SqliteContextOffloadStoreOptions { @@ -68,9 +79,12 @@ interface ContextReferenceRow { size_bytes: unknown; media_type: unknown; created_at: unknown; + storage_kind?: unknown; + payload?: unknown; } interface ContextBlobRow { + storage_kind: unknown; payload: unknown; size_bytes: unknown; } @@ -88,8 +102,24 @@ interface StoreUsageRow { interface GarbageCandidateRow { blob_id: unknown; size_bytes: unknown; + storage_kind: unknown; + payload: unknown; +} + +interface ManagedFilePublication { + readonly locator: string; } +type ContextBlobStorageKind = 'inline' | 'managed_file'; + +type PreparedContextRead = + | ContextOffloadReadResult + | { + readonly kind: 'managed_file'; + readonly record: ContextOffloadRecord; + readonly locator: string; + }; + /** Low-level implementation; production callers must use the Storage Root authority facade. */ export class SqliteContextOffloadStore implements ContextOffloadStore { readonly #database: DatabaseSync; @@ -98,6 +128,9 @@ export class SqliteContextOffloadStore implements ContextOffloadStore { readonly #idFactory: () => string; readonly #failpoint?: (point: SqliteContextOffloadStoreFailpoint) => void; readonly #onUnavailable?: (error: unknown) => void; + readonly #storageRoot: string | undefined; + readonly #valueRoot: string | undefined; + #managedValueMutationTail: Promise = Promise.resolve(); #closed = false; constructor(path: string, options: SqliteContextOffloadStoreOptions) { @@ -107,7 +140,12 @@ export class SqliteContextOffloadStore implements ContextOffloadStore { this.#idFactory = options.idFactory ?? randomUUID; this.#failpoint = options.failpoint; this.#onUnavailable = options.onUnavailable; - if (path !== ':memory:') mkdirSync(dirname(path), { recursive: true }); + const storageRoot = path === ':memory:' ? undefined : dirname(path); + if (storageRoot) mkdirSync(storageRoot, { recursive: true }); + this.#storageRoot = storageRoot ? realpathSync(storageRoot) : undefined; + this.#valueRoot = this.#storageRoot + ? join(this.#storageRoot, CONTEXT_OFFLOAD_VALUES_DIRECTORY_NAME) + : undefined; const Database = loadDatabaseSync(); this.#database = new Database(path); try { @@ -147,13 +185,40 @@ export class SqliteContextOffloadStore implements ContextOffloadStore { return { ok: false, reason: 'identity_conflict' }; } - try { - this.#assertOpen(); - return this.#writeTransaction(() => this.#put({ ...input, bytes, blobId })); - } catch (error) { - this.#onUnavailable?.(error); - return { ok: false, reason: 'unavailable' }; - } + const operation = async (): Promise => { + let publication: ManagedFilePublication | undefined; + let deletionIntentLocator: string | undefined; + try { + this.#assertOpen(); + const existingStorageKind = this.#readBlobStorageKind(blobId); + const storageKind = + preferredStorageKind(input.owner) === 'managed_file' || + existingStorageKind === 'managed_file' + ? 'managed_file' + : 'inline'; + if (storageKind === 'managed_file') { + const locator = managedFileLocator(blobId); + this.#recordManagedFileDeletionIntent(locator, bytes.byteLength); + deletionIntentLocator = locator; + publication = await this.#publishManagedFile(locator, blobId, bytes); + this.#failpoint?.('after_managed_file_publish'); + } + const result = this.#writeTransaction(() => + this.#put({ ...input, bytes, blobId, storageKind, publication }), + ); + if (!result.ok && publication) { + await this.#drainFileDeletion(publication.locator).catch(() => undefined); + } + return result; + } catch (error) { + if (deletionIntentLocator) { + await this.#drainFileDeletion(deletionIntentLocator).catch(() => undefined); + } + this.#onUnavailable?.(error); + return { ok: false, reason: 'unavailable' }; + } + }; + return this.#runManagedValueMutation(operation); } async read(input: { @@ -166,7 +231,9 @@ export class SqliteContextOffloadStore implements ContextOffloadStore { assertNonNegativeSafeInteger(input.maxBytes, 'Context read byte limit'); try { this.#assertOpen(); - return this.#readTransaction(() => this.#read(input)); + const prepared = this.#readTransaction(() => this.#prepareRead(input)); + if ('ok' in prepared) return prepared; + return await this.#readManagedFile(prepared); } catch (error) { this.#onUnavailable?.(error); return { ok: false, reason: 'unavailable' }; @@ -317,63 +384,108 @@ export class SqliteContextOffloadStore implements ContextOffloadStore { if (input.maxBlobs === Number.MAX_SAFE_INTEGER) { throw new Error('Context garbage blob limit is too large'); } - this.#assertOpen(); - return this.#writeTransaction(() => { - const rows = this.#database - .prepare( - `SELECT c.blob_id, b.size_bytes + return this.#runManagedValueMutation(async () => { + this.#assertOpen(); + if (this.#hasPendingFileDeletions()) { + await this.#drainPendingFileDeletions(input.maxBlobs); + return { + deletedBlobs: 0, + deletedBytes: 0, + hasMore: this.#hasPendingFileDeletions() || this.#hasEligibleGarbage(input.olderThan), + }; + } + const collected = this.#writeTransaction(() => { + const rows = this.#database + .prepare( + `SELECT c.blob_id, b.size_bytes, b.storage_kind, b.payload FROM context_gc_candidates c INDEXED BY context_gc_candidates_eligible JOIN context_blobs b ON b.blob_id = c.blob_id WHERE c.unreferenced_at < ? ORDER BY c.unreferenced_at, c.blob_id LIMIT ?`, - ) - .all(input.olderThan, input.maxBlobs + 1) as unknown as GarbageCandidateRow[]; - const selected: Uint8Array[] = []; - let deletedBytes = 0; - for (const row of rows) { - if (selected.length === input.maxBlobs) break; - const blobId = decodeBlobId(row.blob_id); - if (!blobId || !isNonNegativeSafeInteger(row.size_bytes)) { - throw new Error('Invalid context garbage candidate'); - } - if (exceedsLimit(deletedBytes, row.size_bytes, input.maxBytes)) { - if (selected.length === 0) { - throw new Error( - `Context garbage byte limit ${input.maxBytes} cannot fit eligible blob of ${row.size_bytes} bytes`, + ) + .all(input.olderThan, input.maxBlobs + 1) as unknown as GarbageCandidateRow[]; + const selected: Array<{ + readonly blobId: Uint8Array; + readonly sizeBytes: number; + readonly managedLocator?: string; + }> = []; + let deletedBytes = 0; + let inlineDeletedBytes = 0; + for (const row of rows) { + if (selected.length === input.maxBlobs) break; + const blobId = decodeBlobId(row.blob_id); + if (!blobId || !isNonNegativeSafeInteger(row.size_bytes)) { + throw new Error('Invalid context garbage candidate'); + } + const value = decodeBlobValue(row, Buffer.from(blobId).toString('hex')); + if (!value) throw new Error('Invalid context garbage candidate value'); + if (exceedsLimit(deletedBytes, row.size_bytes, input.maxBytes)) { + if (selected.length === 0) { + throw new Error( + `Context garbage byte limit ${input.maxBytes} cannot fit eligible blob of ${row.size_bytes} bytes`, + ); + } + break; + } + deletedBytes = addSafeInteger(deletedBytes, row.size_bytes, 'Collected context bytes'); + selected.push({ + blobId, + sizeBytes: row.size_bytes, + ...(value.kind === 'managed_file' ? { managedLocator: value.locator } : {}), + }); + if (value.kind === 'inline') { + inlineDeletedBytes = addSafeInteger( + inlineDeletedBytes, + row.size_bytes, + 'Collected inline context bytes', ); } - break; } - deletedBytes = addSafeInteger(deletedBytes, row.size_bytes, 'Collected context bytes'); - selected.push(blobId); - } - const deleteBlob = this.#database.prepare( - `DELETE FROM context_blobs + const deleteBlob = this.#database.prepare( + `DELETE FROM context_blobs WHERE blob_id = ? AND NOT EXISTS (SELECT 1 FROM context_refs WHERE blob_id = ?)`, - ); - for (const blobId of selected) { - const deleted = deleteBlob.run(blobId, blobId); - if (deleted.changes !== 1) { - throw new Error('Context garbage candidate is still referenced or missing'); + ); + const enqueueFileDeletion = this.#database.prepare( + `INSERT INTO context_file_deletions(locator, size_bytes, enqueued_at) + VALUES (?, ?, ?) + ON CONFLICT(locator) DO NOTHING`, + ); + for (const selectedBlob of selected) { + if (selectedBlob.managedLocator) { + enqueueFileDeletion.run( + Buffer.from(selectedBlob.managedLocator, 'utf8'), + selectedBlob.sizeBytes, + this.#readNow(), + ); + } + const deleted = deleteBlob.run(selectedBlob.blobId, selectedBlob.blobId); + if (deleted.changes !== 1) { + throw new Error('Context garbage candidate is still referenced or missing'); + } + this.#failpoint?.('after_gc_blob_delete'); } - this.#failpoint?.('after_gc_blob_delete'); - } - if (selected.length > 0) { - const updated = this.#database - .prepare( - `UPDATE context_store_usage + if (selected.length > 0) { + const updated = this.#database + .prepare( + `UPDATE context_store_usage SET blob_count = blob_count - ?, physical_bytes = physical_bytes - ? WHERE singleton = 1`, - ) - .run(selected.length, deletedBytes); - if (updated.changes !== 1) throw new Error('Missing context store usage row'); - } + ) + .run(selected.length, inlineDeletedBytes); + if (updated.changes !== 1) throw new Error('Missing context store usage row'); + } + return { + deletedBlobs: selected.length, + deletedBytes, + hasMore: rows.length > selected.length, + }; + }); + await this.#drainPendingFileDeletions(input.maxBlobs); return { - deletedBlobs: selected.length, - deletedBytes, - hasMore: rows.length > selected.length, + ...collected, + hasMore: collected.hasMore || this.#hasPendingFileDeletions(), }; }); } @@ -404,21 +516,52 @@ export class SqliteContextOffloadStore implements ContextOffloadStore { this.#database.close(); } + #readBlobStorageKind(blobId: string): ContextBlobStorageKind | undefined { + const row = this.#database + .prepare('SELECT storage_kind FROM context_blobs WHERE blob_id = ?') + .get(Buffer.from(blobId, 'hex')) as { storage_kind?: unknown } | undefined; + if (!row) return undefined; + if (row.storage_kind !== 'inline' && row.storage_kind !== 'managed_file') { + throw new Error('Invalid context blob storage kind'); + } + return row.storage_kind; + } + #put(input: { readonly sessionId: string; readonly owner: ContextOffloadOwner; readonly bytes: Uint8Array; readonly mediaType: string; readonly blobId: string; + readonly storageKind: ContextBlobStorageKind; + readonly publication?: ManagedFilePublication; }): ContextOffloadPutResult { const existingReference = this.#readReferenceByOwner(input.sessionId, input.owner); if (existingReference) { - if (existingReference.blobId !== input.blobId) { + if ( + existingReference.blobId !== input.blobId || + existingReference.mediaType !== input.mediaType + ) { return { ok: false, reason: 'identity_conflict' }; } - if (!this.#verifyBlob(input.blobId, input.bytes)) { - throw new Error(`Context blob failed integrity verification: ${input.blobId}`); + } + + const blobIdBytes = Buffer.from(input.blobId, 'hex'); + const existingBlob = this.#database + .prepare('SELECT storage_kind, payload, size_bytes FROM context_blobs WHERE blob_id = ?') + .get(blobIdBytes) as ContextBlobRow | undefined; + if (existingBlob) { + if (!blobMatchesInput(existingBlob, input.blobId, input.bytes)) { + throw new Error(`Context blob identity is inconsistent: ${input.blobId}`); } + } + + if (input.storageKind === 'managed_file' && !input.publication) { + throw new Error('Managed context value was not durably published'); + } + if (existingReference) { + this.#cancelPendingFileDeletion(input.publication, input.bytes.byteLength); + this.#promoteToManagedFile(existingBlob, input, blobIdBytes); return { ok: true, record: existingReference }; } @@ -430,27 +573,25 @@ export class SqliteContextOffloadStore implements ContextOffloadStore { if (exceedsLimit(logicalBytes, input.bytes.byteLength, this.#limits.sessionLogicalBytes)) { return { ok: false, reason: 'session_quota_exceeded' }; } - - const blobIdBytes = Buffer.from(input.blobId, 'hex'); - const existingBlob = this.#database - .prepare('SELECT payload, size_bytes FROM context_blobs WHERE blob_id = ?') - .get(blobIdBytes) as ContextBlobRow | undefined; - const storeUsage = this.#readStoreUsage(); - if (existingBlob) { - if (!blobMatches(existingBlob, input.blobId, input.bytes)) { - throw new Error(`Context blob identity is inconsistent: ${input.blobId}`); - } - } else { + if (!existingBlob) { const physicalBytes = readNonNegativeInteger( - storeUsage.physical_bytes, + this.#readStoreUsage().physical_bytes, 'Workspace physical bytes', ); - if ( - exceedsLimit(physicalBytes, input.bytes.byteLength, this.#limits.workspacePhysicalBytes) - ) { + const exceedsWorkspaceQuota = + input.storageKind === 'managed_file' + ? physicalBytes > this.#limits.workspacePhysicalBytes + : exceedsLimit( + physicalBytes, + input.bytes.byteLength, + this.#limits.workspacePhysicalBytes, + ); + if (exceedsWorkspaceQuota) { return { ok: false, reason: 'workspace_quota_exceeded' }; } } + this.#cancelPendingFileDeletion(input.publication, input.bytes.byteLength); + this.#promoteToManagedFile(existingBlob, input, blobIdBytes); const createdAt = this.#readNow(); const refId = this.#idFactory(); @@ -458,10 +599,18 @@ export class SqliteContextOffloadStore implements ContextOffloadStore { if (!existingBlob) { this.#database .prepare( - `INSERT INTO context_blobs(blob_id, payload, size_bytes, created_at) - VALUES (?, ?, ?, ?)`, + `INSERT INTO context_blobs(blob_id, storage_kind, payload, size_bytes, created_at) + VALUES (?, ?, ?, ?, ?)`, ) - .run(blobIdBytes, input.bytes, input.bytes.byteLength, createdAt); + .run( + blobIdBytes, + input.storageKind, + input.storageKind === 'inline' + ? input.bytes + : Buffer.from(input.publication?.locator ?? '', 'utf8'), + input.bytes.byteLength, + createdAt, + ); this.#database .prepare( `UPDATE context_store_usage @@ -512,6 +661,44 @@ export class SqliteContextOffloadStore implements ContextOffloadStore { }; } + #promoteToManagedFile( + existingBlob: ContextBlobRow | undefined, + input: { + readonly storageKind: ContextBlobStorageKind; + readonly publication?: ManagedFilePublication; + }, + blobId: Uint8Array, + ): void { + if (existingBlob?.storage_kind !== 'inline' || input.storageKind !== 'managed_file') return; + this.#database + .prepare( + `UPDATE context_blobs + SET storage_kind = 'managed_file', payload = ? + WHERE blob_id = ? AND storage_kind = 'inline'`, + ) + .run(Buffer.from(input.publication?.locator ?? '', 'utf8'), blobId); + } + + #cancelPendingFileDeletion( + publication: ManagedFilePublication | undefined, + sizeBytes: number, + ): void { + if (!publication) return; + const locator = Buffer.from(publication.locator, 'utf8'); + const row = this.#database + .prepare('SELECT size_bytes FROM context_file_deletions WHERE locator = ?') + .get(locator) as { size_bytes?: unknown } | undefined; + if (!row) return; + if (row.size_bytes !== sizeBytes) { + throw new Error('Pending context file deletion has an inconsistent size'); + } + const deleted = this.#database + .prepare('DELETE FROM context_file_deletions WHERE locator = ?') + .run(locator); + if (deleted.changes !== 1) throw new Error('Pending context file deletion disappeared'); + this.#releasePendingFileBytes(sizeBytes); + } + #copyReferences(input: { readonly sourceSessionId: string; readonly targetSessionId: string; @@ -551,14 +738,16 @@ export class SqliteContextOffloadStore implements ContextOffloadStore { const ownerKey = `${reference.targetOwner.kind}\0${reference.targetOwner.ownerId}`; const pending = pendingByOwner.get(ownerKey); if (pending) { - if (pending.blobId !== source.blobId) return { ok: false, reason: 'identity_conflict' }; + if (pending.blobId !== source.blobId || pending.mediaType !== source.mediaType) { + return { ok: false, reason: 'identity_conflict' }; + } copied.push({ sourceRefId: reference.sourceRefId, targetRefId: pending.refId }); continue; } const existing = this.#readReferenceByOwner(input.targetSessionId, reference.targetOwner); if (existing) { - if (existing.blobId !== source.blobId) { + if (existing.blobId !== source.blobId || existing.mediaType !== source.mediaType) { return { ok: false, reason: 'identity_conflict' }; } pendingByOwner.set(ownerKey, { @@ -636,20 +825,20 @@ export class SqliteContextOffloadStore implements ContextOffloadStore { return { ok: true, copied }; } - #read(input: { + #prepareRead(input: { readonly sessionId: string; readonly refId: string; readonly maxBytes: number; - }): ContextOffloadReadResult { + }): PreparedContextRead { const row = this.#database .prepare( `SELECT r.ref_id, r.session_id, r.owner_kind, r.owner_id, r.blob_id, - b.size_bytes, r.media_type, r.created_at + b.size_bytes, b.storage_kind, b.payload, r.media_type, r.created_at FROM context_refs r JOIN context_blobs b ON b.blob_id = r.blob_id WHERE r.ref_id = ?`, ) - .get(input.refId) as ContextReferenceRow | undefined; + .get(input.refId) as (ContextReferenceRow & ContextBlobRow) | undefined; if (!row) return { ok: false, reason: 'not_found' }; const record = decodeReferenceRow(row); if (!record) return { ok: false, reason: 'corrupt' }; @@ -660,20 +849,17 @@ export class SqliteContextOffloadStore implements ContextOffloadStore { ) { return { ok: false, reason: 'too_large' }; } - const blob = this.#database - .prepare('SELECT payload, size_bytes FROM context_blobs WHERE blob_id = ?') - .get(Buffer.from(record.blobId, 'hex')) as ContextBlobRow | undefined; - if (!blob) return { ok: false, reason: 'corrupt' }; - const bytes = decodeBytes(blob.payload); - if ( - !bytes || - blob.size_bytes !== record.sizeBytes || - bytes.byteLength !== record.sizeBytes || - createHash('sha256').update(bytes).digest('hex') !== record.blobId - ) { + const value = decodeBlobValue(row, record.blobId); + if (!value || row.size_bytes !== record.sizeBytes) { return { ok: false, reason: 'corrupt' }; } - return { ok: true, record, bytes }; + if (value.kind === 'managed_file') { + return { kind: 'managed_file', record, locator: value.locator }; + } + if (createHash('sha256').update(value.bytes).digest('hex') !== record.blobId) { + return { ok: false, reason: 'corrupt' }; + } + return { ok: true, record, bytes: value.bytes }; } #readReferenceByOwner( @@ -695,11 +881,291 @@ export class SqliteContextOffloadStore implements ContextOffloadStore { return record; } - #verifyBlob(blobId: string, bytes: Uint8Array): boolean { - const blob = this.#database - .prepare('SELECT payload, size_bytes FROM context_blobs WHERE blob_id = ?') - .get(Buffer.from(blobId, 'hex')) as ContextBlobRow | undefined; - return blob !== undefined && blobMatches(blob, blobId, bytes); + async #readManagedFile(input: { + readonly record: ContextOffloadRecord; + readonly locator: string; + }): Promise { + let bytes: Uint8Array; + try { + const path = this.#managedFilePath(input.locator, input.record.blobId); + await this.#assertManagedDirectory(dirname(path)); + bytes = await readStableBoundedFile({ + path, + maxBytes: input.record.sizeBytes, + invalidFile: () => new InvalidManagedContextFileError(), + }); + } catch (error) { + if (error instanceof InvalidManagedContextFileError || isNodeError(error, 'ENOENT')) { + return { ok: false, reason: 'corrupt' }; + } + throw error; + } + if ( + bytes.byteLength !== input.record.sizeBytes || + createHash('sha256').update(bytes).digest('hex') !== input.record.blobId + ) { + return { ok: false, reason: 'corrupt' }; + } + return { ok: true, record: input.record, bytes: new Uint8Array(bytes) }; + } + + async #publishManagedFile( + locator: string, + blobId: string, + bytes: Uint8Array, + ): Promise { + const target = this.#managedFilePath(locator, blobId); + const targetDirectory = dirname(target); + const storageRoot = this.#storageRoot; + if (!storageRoot) throw new Error('Managed context files require a durable Storage Root'); + await this.#ensureManagedDirectory(targetDirectory); + const temporary = managedFileStagingPath(target, blobId); + let handle: Awaited> | undefined; + try { + await unlink(temporary).then( + () => syncDirectory(targetDirectory), + (error: unknown) => { + if (!isNodeError(error, 'ENOENT')) throw error; + }, + ); + handle = await open(temporary, 'wx', 0o600); + await handle.writeFile(bytes); + await handle.sync(); + await handle.close(); + handle = undefined; + this.#failpoint?.('after_managed_file_staging'); + try { + await link(temporary, target); + await this.#assertManagedDirectory(targetDirectory); + await syncDirectoryChain(targetDirectory, storageRoot); + } catch (error) { + if (!isNodeError(error, 'EEXIST')) throw error; + await this.#verifyManagedFile(target, blobId, bytes.byteLength); + await syncFile(target); + await this.#assertManagedDirectory(targetDirectory); + await syncDirectoryChain(targetDirectory, storageRoot); + } + } finally { + await handle?.close().catch(() => undefined); + await unlink(temporary).then( + () => syncDirectory(targetDirectory), + (error: unknown) => { + if (!isNodeError(error, 'ENOENT')) throw error; + }, + ); + } + return { locator }; + } + + async #verifyManagedFile(path: string, blobId: string, sizeBytes: number): Promise { + await this.#assertManagedDirectory(dirname(path)); + const bytes = await readStableBoundedFile({ + path, + maxBytes: sizeBytes, + invalidFile: () => new InvalidManagedContextFileError(), + }); + if ( + bytes.byteLength !== sizeBytes || + createHash('sha256').update(bytes).digest('hex') !== blobId + ) { + throw new InvalidManagedContextFileError(); + } + } + + #recordManagedFileDeletionIntent(locator: string, sizeBytes: number): void { + const locatorBytes = Buffer.from(locator, 'utf8'); + this.#writeTransaction(() => { + const inserted = this.#database + .prepare( + `INSERT INTO context_file_deletions(locator, size_bytes, enqueued_at) + VALUES (?, ?, ?) + ON CONFLICT(locator) DO NOTHING`, + ) + .run(locatorBytes, sizeBytes, this.#readNow()); + if (inserted.changes === 1) { + const updated = this.#database + .prepare( + `UPDATE context_store_usage + SET physical_bytes = physical_bytes + ? + WHERE singleton = 1`, + ) + .run(sizeBytes); + if (updated.changes !== 1) throw new Error('Missing context store usage row'); + } + const row = this.#database + .prepare('SELECT size_bytes FROM context_file_deletions WHERE locator = ?') + .get(locatorBytes) as { size_bytes?: unknown } | undefined; + if (row?.size_bytes !== sizeBytes) { + throw new Error('Pending context file deletion has an inconsistent size'); + } + }); + } + + async #drainPendingFileDeletions(limit: number): Promise { + const rows = this.#database + .prepare('SELECT locator FROM context_file_deletions ORDER BY enqueued_at, locator LIMIT ?') + .all(limit) as Array<{ locator?: unknown }>; + for (const row of rows) { + const locator = decodeManagedFileLocator(row.locator); + if (!locator) throw new Error('Invalid pending context file deletion locator'); + await this.#drainFileDeletion(locator); + } + } + + #hasPendingFileDeletions(): boolean { + return Boolean( + this.#database.prepare('SELECT 1 AS present FROM context_file_deletions LIMIT 1').get(), + ); + } + + #hasEligibleGarbage(olderThan: number): boolean { + return Boolean( + this.#database + .prepare( + `SELECT 1 AS present + FROM context_gc_candidates INDEXED BY context_gc_candidates_eligible + WHERE unreferenced_at < ? + LIMIT 1`, + ) + .get(olderThan), + ); + } + + async #drainFileDeletion(locator: string): Promise { + const locatorBytes = Buffer.from(locator, 'utf8'); + const live = this.#database + .prepare( + `SELECT 1 AS present FROM context_blobs + WHERE storage_kind = 'managed_file' AND payload = ? LIMIT 1`, + ) + .get(locatorBytes) as { present?: unknown } | undefined; + await this.#deleteManagedFile(locator, live?.present !== 1); + this.#writeTransaction(() => { + const row = this.#database + .prepare('SELECT size_bytes FROM context_file_deletions WHERE locator = ?') + .get(locatorBytes) as { size_bytes?: unknown } | undefined; + if (!row) return; + const sizeBytes = readNonNegativeInteger( + row.size_bytes, + 'Pending context file deletion bytes', + ); + const deleted = this.#database + .prepare('DELETE FROM context_file_deletions WHERE locator = ?') + .run(locatorBytes); + if (deleted.changes !== 1) throw new Error('Pending context file deletion disappeared'); + this.#releasePendingFileBytes(sizeBytes); + }); + } + + #releasePendingFileBytes(sizeBytes: number): void { + const updated = this.#database + .prepare( + `UPDATE context_store_usage + SET physical_bytes = physical_bytes - ? + WHERE singleton = 1 AND physical_bytes >= ?`, + ) + .run(sizeBytes, sizeBytes); + if (updated.changes !== 1) throw new Error('Context physical byte accounting underflow'); + } + + async #deleteManagedFile(locator: string, deleteTarget: boolean): Promise { + const path = this.#managedFilePath(locator); + const blobId = MANAGED_FILE_LOCATOR_PATTERN.exec(locator)?.[2]; + if (!blobId) throw new InvalidManagedContextFileError(); + const staging = managedFileStagingPath(path, blobId); + try { + await this.#assertManagedDirectory(dirname(path)); + let deleted = false; + for (const candidate of deleteTarget ? [path, staging] : [staging]) { + await unlink(candidate).then( + () => { + deleted = true; + }, + (error: unknown) => { + if (!isNodeError(error, 'ENOENT')) throw error; + }, + ); + } + if (deleted) await syncDirectory(dirname(path)); + } catch (error) { + if (!isNodeError(error, 'ENOENT')) throw error; + } + } + + async #ensureManagedDirectory(directory: string): Promise { + for (const path of this.#managedDirectoryChain(directory)) { + try { + await mkdir(path, { mode: 0o700 }); + } catch (error) { + if (!isNodeError(error, 'EEXIST')) throw error; + } + await this.#assertManagedDirectoryEntry(path); + } + } + + async #assertManagedDirectory(directory: string): Promise { + for (const path of this.#managedDirectoryChain(directory)) { + await this.#assertManagedDirectoryEntry(path); + } + } + + #managedDirectoryChain(directory: string): readonly string[] { + const valueRoot = this.#valueRoot; + if (!valueRoot) throw new Error('Managed context files require a durable Storage Root'); + const shaRoot = join(valueRoot, 'sha256'); + const shard = relative(shaRoot, directory); + if (!/^[0-9a-f]{2}$/u.test(shard) || isAbsolute(shard) || shard.includes(sep)) { + throw new InvalidManagedContextFileError(); + } + return [valueRoot, shaRoot, directory]; + } + + async #assertManagedDirectoryEntry(path: string): Promise { + const storageRoot = this.#storageRoot; + if (!storageRoot) throw new Error('Managed context files require a durable Storage Root'); + let entry: Awaited>; + let resolved: string; + try { + [entry, resolved] = await Promise.all([lstat(path), realpath(path)]); + } catch (error) { + if (isNodeError(error, 'ENOTDIR')) { + throw new InvalidManagedContextFileError(); + } + throw error; + } + const fromRoot = relative(storageRoot, resolved); + if ( + !entry.isDirectory() || + entry.isSymbolicLink() || + fromRoot === '..' || + fromRoot.startsWith(`..${sep}`) || + isAbsolute(fromRoot) + ) { + throw new InvalidManagedContextFileError(); + } + } + + #managedFilePath(locator: string, expectedBlobId?: string): string { + const valueRoot = this.#valueRoot; + if (!valueRoot) throw new Error('Managed context files require a durable Storage Root'); + const match = MANAGED_FILE_LOCATOR_PATTERN.exec(locator); + const blobId = match?.[2]; + if (!match || !blobId || match[1] !== blobId.slice(0, 2)) { + throw new InvalidManagedContextFileError(); + } + if (expectedBlobId !== undefined && blobId !== expectedBlobId) { + throw new InvalidManagedContextFileError(); + } + return join(valueRoot, 'sha256', match[1], blobId); + } + + #runManagedValueMutation(operation: () => Promise): Promise { + const pending = this.#managedValueMutationTail.then(operation, operation); + this.#managedValueMutationTail = pending.then( + () => undefined, + () => undefined, + ); + return pending; } #markBlobUnreferencedIfEligible(blobId: Uint8Array, unreferencedAt: number): void { @@ -800,14 +1266,39 @@ function decodeReferenceRow(row: ContextReferenceRow): ContextOffloadRecord | un }; } -function blobMatches(row: ContextBlobRow, blobId: string, expectedBytes: Uint8Array): boolean { - const storedBytes = decodeBytes(row.payload); +type DecodedBlobValue = + | { readonly kind: 'inline'; readonly bytes: Uint8Array } + | { readonly kind: 'managed_file'; readonly locator: string }; + +function decodeBlobValue( + row: ContextBlobRow, + expectedBlobId?: string, +): DecodedBlobValue | undefined { + if (!isNonNegativeSafeInteger(row.size_bytes)) return undefined; + if (row.storage_kind === 'inline') { + const bytes = decodeBytes(row.payload); + return bytes?.byteLength === row.size_bytes ? { kind: 'inline', bytes } : undefined; + } + if (row.storage_kind !== 'managed_file') return undefined; + const locator = decodeManagedFileLocator(row.payload); + if (!locator) return undefined; + const blobId = MANAGED_FILE_LOCATOR_PATTERN.exec(locator)?.[2]; + if (expectedBlobId !== undefined && blobId !== expectedBlobId) return undefined; + return { kind: 'managed_file', locator }; +} + +function blobMatchesInput(row: ContextBlobRow, blobId: string, expectedBytes: Uint8Array): boolean { + if ( + row.size_bytes !== expectedBytes.byteLength || + createHash('sha256').update(expectedBytes).digest('hex') !== blobId + ) { + return false; + } + const value = decodeBlobValue(row, blobId); return ( - storedBytes !== undefined && - isNonNegativeSafeInteger(row.size_bytes) && - row.size_bytes === expectedBytes.byteLength && - storedBytes.byteLength === expectedBytes.byteLength && - createHash('sha256').update(storedBytes).digest('hex') === blobId + value !== undefined && + (value.kind === 'managed_file' || + createHash('sha256').update(value.bytes).digest('hex') === blobId) ); } @@ -820,6 +1311,21 @@ function decodeBlobId(value: unknown): Uint8Array | undefined { return bytes?.byteLength === 32 ? bytes : undefined; } +function managedFileLocator(blobId: string): string { + if (!SHA256_PATTERN.test(blobId)) throw new Error('Invalid managed context blob identity'); + return `sha256/${blobId.slice(0, 2)}/${blobId}`; +} + +function decodeManagedFileLocator(value: unknown): string | undefined { + const bytes = decodeBytes(value); + if (!bytes || bytes.byteLength === 0 || bytes.byteLength > 512) return undefined; + const locator = Buffer.from(bytes).toString('utf8'); + if (!Buffer.from(locator, 'utf8').equals(Buffer.from(bytes))) return undefined; + const match = MANAGED_FILE_LOCATOR_PATTERN.exec(locator); + const blobId = match?.[2]; + return match && blobId && match[1] === blobId.slice(0, 2) ? locator : undefined; +} + function usageFromRows(session: SessionUsageRow, store: StoreUsageRow): ContextOffloadUsage { return { references: readNonNegativeInteger(session.reference_count, 'Context reference count'), @@ -849,6 +1355,14 @@ function assertOwner(owner: ContextOffloadOwner): void { assertBoundedIdentity(owner.ownerId, 'Context owner id'); } +function preferredStorageKind(owner: ContextOffloadOwner): ContextBlobStorageKind { + return owner.kind === 'read_image_snapshot' ? 'managed_file' : 'inline'; +} + +function managedFileStagingPath(target: string, blobId: string): string { + return join(dirname(target), `.${blobId}.publish.tmp`); +} + function isOwnerKind(value: unknown): value is ContextOffloadOwner['kind'] { return value === 'read_image_snapshot' || value === 'tool_result_archive'; } @@ -905,3 +1419,9 @@ function rollback(database: DatabaseSync): void { // Preserve the operation failure that triggered rollback. } } + +function isNodeError(error: unknown, code: string): error is NodeJS.ErrnoException { + return error instanceof Error && 'code' in error && error.code === code; +} + +class InvalidManagedContextFileError extends Error {}