From da7c4810d88079abc4a5d997c67c542ba1836a9f Mon Sep 17 00:00:00 2001 From: likun Date: Sat, 29 Aug 2026 21:51:05 +0800 Subject: [PATCH 1/5] feat(storage): read durable Session context refs Generated-by: OpenAI Codex --- .../core/src/__tests__/runtime-event.test.ts | 39 ++++++++++++++ packages/core/src/context-offload.ts | 3 ++ packages/core/src/events.ts | 26 +++++++++- .../src/__tests__/protocol.test.ts | 15 ++++++ packages/runtime-host/src/protocol/turn.ts | 10 ++-- .../src/server/execution-composition.ts | 19 +++++++ .../src/server/execution-model-composition.ts | 11 ++++ .../provider-image-overflow-recovery.test.ts | 35 +++++++++++-- .../src/provider-image-overflow-recovery.ts | 7 +++ packages/storage/package.json | 2 + .../__tests__/artifact-attachments.test.ts | 52 +++++++++++++++++++ .../storage-writer-composition.test.ts | 47 +++++++++++++++++ packages/storage/src/artifact-attachments.ts | 10 ++++ .../src/sqlite-context-offload-store.ts | 4 +- .../storage/src/storage-writer-composition.ts | 16 ++++++ 15 files changed, 285 insertions(+), 11 deletions(-) diff --git a/packages/core/src/__tests__/runtime-event.test.ts b/packages/core/src/__tests__/runtime-event.test.ts index 120d2eac3a..f3c69c7d7e 100644 --- a/packages/core/src/__tests__/runtime-event.test.ts +++ b/packages/core/src/__tests__/runtime-event.test.ts @@ -22,6 +22,7 @@ import assert from 'node:assert/strict'; import { expect } from './test-helpers.js'; import { decodeMessageContent, + isCanonicalStorageRef, messageContentsEqual, normalizeMessageContent, type SessionEvent, @@ -173,6 +174,33 @@ describe('continuation-start protocol', () => { }); describe('RuntimeEvent content variants', () => { + test('recognizes canonical durable Session context references', () => { + assert.equal( + isCanonicalStorageRef({ + kind: 'session_context', + sessionId: 'session-1', + refId: 'read-image:owner-1', + }), + true, + ); + assert.equal( + isCanonicalStorageRef({ + kind: 'session_context', + sessionId: 'session-1', + refId: '😀'.repeat(512), + }), + true, + ); + for (const ref of [ + { kind: 'session_context', sessionId: 'bad/session', refId: 'ref-1' }, + { kind: 'session_context', sessionId: 'session-1', refId: '' }, + { kind: 'session_context', sessionId: 'session-1', refId: '😀'.repeat(513) }, + { kind: 'session_context', sessionId: 'session-1', refId: 'ref-1', extra: true }, + ]) { + assert.equal(isCanonicalStorageRef(ref), false); + } + }); + test('preserves sent inline references as message identity', () => { const inlineReferences = [ { kind: 'skill', value: '/skill:writer', label: 'Writer', start: 8 }, @@ -324,6 +352,17 @@ describe('RuntimeEvent content variants', () => { bytes: 1, ref: { kind: 'workspace_file' as const, relativePath: 'a.ts' }, }, + { + kind: 'image' as const, + name: 'snapshot.png', + mimeType: 'image/png', + bytes: 8, + ref: { + kind: 'session_context' as const, + sessionId: 'session-1', + refId: 'read-image:owner-1', + }, + }, ]; const quotes = [ { text: 'first', label: 'Assistant', sourceTurnId: 'turn-1' }, diff --git a/packages/core/src/context-offload.ts b/packages/core/src/context-offload.ts index 77c9e225e7..fa0098a3f1 100644 --- a/packages/core/src/context-offload.ts +++ b/packages/core/src/context-offload.ts @@ -23,6 +23,9 @@ export interface SessionContextRef { readonly refId: string; } +/** Maximum Unicode code points accepted for durable context-offload identities. */ +export const CONTEXT_OFFLOAD_ID_MAX_CODE_POINTS = 512; + export type ContextOffloadOwner = | { readonly kind: 'read_image_snapshot'; diff --git a/packages/core/src/events.ts b/packages/core/src/events.ts index dc9c11c41d..2185174067 100644 --- a/packages/core/src/events.ts +++ b/packages/core/src/events.ts @@ -27,6 +27,7 @@ */ import * as nodeCrypto from 'node:crypto'; +import { CONTEXT_OFFLOAD_ID_MAX_CODE_POINTS, type SessionContextRef } from './context-offload.js'; import type { AdditionalPermissionRequest, PermissionMode, @@ -74,6 +75,7 @@ type TerminalToolResultStatus = Exclude; // ============================================================================ export type StorageRef = + | SessionContextRef | { kind: 'session_file'; sessionId: string; relativePath: string } | { kind: 'workspace_file'; relativePath: string } | { kind: 'external_file'; absolutePath: string }; @@ -154,6 +156,9 @@ const SESSION_FILE_REF_SHAPE = defineObjectShape +>()(['kind', 'sessionId', 'refId'], []); const WORKSPACE_FILE_REF_SHAPE = defineObjectShape< Extract >()(['kind', 'relativePath'], []); @@ -328,6 +333,13 @@ export function isStorageRef(value: unknown): value is StorageRef { typeof value.relativePath === 'string' ); } + if (value.kind === 'session_context') { + return ( + hasExactShape(value, SESSION_CONTEXT_REF_SHAPE) && + typeof value.sessionId === 'string' && + typeof value.refId === 'string' + ); + } if (value.kind === 'workspace_file') { return hasExactShape(value, WORKSPACE_FILE_REF_SHAPE) && typeof value.relativePath === 'string'; } @@ -341,9 +353,15 @@ export function isStorageRef(value: unknown): value is StorageRef { export function isCanonicalStorageRef(value: unknown): value is StorageRef { if (!isStorageRef(value)) return false; if (value.kind === 'external_file') return isCanonicalAbsolutePath(value.absolutePath); - if (value.kind === 'session_file' && !/^[A-Za-z0-9_-]{1,128}$/.test(value.sessionId)) { + if ( + (value.kind === 'session_file' || value.kind === 'session_context') && + !/^[A-Za-z0-9_-]{1,128}$/.test(value.sessionId) + ) { return false; } + if (value.kind === 'session_context') { + return value.refId.length > 0 && [...value.refId].length <= CONTEXT_OFFLOAD_ID_MAX_CODE_POINTS; + } return isCanonicalRelativePath(value.relativePath); } @@ -445,6 +463,12 @@ function attachmentRefsEqual(left: AttachmentRef, right: AttachmentRef): boolean return false; } switch (left.ref.kind) { + case 'session_context': + return ( + right.ref.kind === 'session_context' && + left.ref.sessionId === right.ref.sessionId && + left.ref.refId === right.ref.refId + ); case 'session_file': return ( right.ref.kind === 'session_file' && diff --git a/packages/runtime-host/src/__tests__/protocol.test.ts b/packages/runtime-host/src/__tests__/protocol.test.ts index 1c5241a5e9..b9a4e7b2a3 100644 --- a/packages/runtime-host/src/__tests__/protocol.test.ts +++ b/packages/runtime-host/src/__tests__/protocol.test.ts @@ -1546,6 +1546,18 @@ describe('Runtime Host bootstrap protocol', () => { ), }), ); + assert.doesNotThrow(() => + submit({ + text: 'valid context ref', + attachments: [ + attachmentRef({ + kind: 'session_context', + sessionId: 'session-1', + refId: 'read-image:owner-1', + }), + ], + }), + ); assert.throws( () => submit({ @@ -1566,6 +1578,8 @@ describe('Runtime Host bootstrap protocol', () => { { ...attachmentRef({ kind: 'workspace_file', relativePath: 'a.ts' }), mimeType: '' }, attachmentRef({ kind: 'workspace_file', relativePath: 'a'.repeat(4097) }), attachmentRef({ kind: 'session_file', sessionId: 'bad/id', relativePath: 'a.ts' }), + attachmentRef({ kind: 'session_context', sessionId: 'session-1', refId: '' }), + attachmentRef({ kind: 'session_context', sessionId: 'session-1', refId: 'a'.repeat(513) }), attachmentRef({ kind: 'workspace_file', relativePath: '../secret' }), attachmentRef({ kind: 'workspace_file', relativePath: 'src//a.ts' }), attachmentRef({ kind: 'external_file', absolutePath: 'relative/a.ts' }), @@ -2046,6 +2060,7 @@ function retractedMessage(text = 'do this next') { function attachmentRef( ref: | { kind: 'session_file'; sessionId: string; relativePath: string } + | { kind: 'session_context'; sessionId: string; refId: string } | { kind: 'workspace_file'; relativePath: string } | { kind: 'external_file'; absolutePath: string }, ) { diff --git a/packages/runtime-host/src/protocol/turn.ts b/packages/runtime-host/src/protocol/turn.ts index 6ce0f60455..8aab204b6d 100644 --- a/packages/runtime-host/src/protocol/turn.ts +++ b/packages/runtime-host/src/protocol/turn.ts @@ -431,14 +431,16 @@ export function decodeMessageContent(value: unknown, allowEmptyText = false): Me if (attachment.bytes > MAX_ATTACHMENT_BYTES) { throw invalidProtocolFrame('Invalid AttachmentRef bytes'); } - if (attachment.ref.kind === 'session_file') { + if (attachment.ref.kind === 'session_file' || attachment.ref.kind === 'session_context') { requireEntityId(attachment.ref.sessionId, 'AttachmentRef sessionId'); } - const path = + const identity = attachment.ref.kind === 'external_file' ? attachment.ref.absolutePath - : attachment.ref.relativePath; - requireUtf8String(path, 'AttachmentRef path', ATTACHMENT_PATH_MAX_BYTES, false); + : attachment.ref.kind === 'session_context' + ? attachment.ref.refId + : attachment.ref.relativePath; + requireUtf8String(identity, 'AttachmentRef identity', ATTACHMENT_PATH_MAX_BYTES, false); } if ((content.quotes?.length ?? 0) > TURN_MESSAGE_QUOTE_MAX_COUNT) { throw invalidProtocolFrame('Invalid Message quotes'); diff --git a/packages/runtime-host/src/server/execution-composition.ts b/packages/runtime-host/src/server/execution-composition.ts index 2468419635..810d898deb 100644 --- a/packages/runtime-host/src/server/execution-composition.ts +++ b/packages/runtime-host/src/server/execution-composition.ts @@ -18,6 +18,8 @@ */ import { createHash, randomUUID } from 'node:crypto'; +import { MAX_READ_IMAGE_BYTES } from '@maka/core/attachments'; +import type { ContextOffloadLimits } from '@maka/core/context-offload'; import { messageContentDigest, normalizeMessageContent } from '@maka/core/events'; import { describeChatConfigurationReason, @@ -196,6 +198,17 @@ export interface ExecutionRuntimeHostComposition extends RuntimeHostComposition readonly workspaceExecution: RuntimeHostWorkspaceExecutionComposition; } +const CONTEXT_OFFLOAD_READER_LIMITS: ContextOffloadLimits = Object.freeze({ + ownerMaxBytes: Object.freeze({ + read_image_snapshot: MAX_READ_IMAGE_BYTES, + tool_result_archive: 0, + }), + // This expand slice opens only the reader path. Zero quotas make accidental + // non-empty puts fail closed until the writer/lifecycle cutover lands. + sessionLogicalBytes: 0, + workspacePhysicalBytes: 0, +}); + export interface CreateExecutionRuntimeHostCompositionOptions { readonly bootstrapRuntimePolicy?: boolean; readonly skillHomeDirectory?: string; @@ -222,6 +235,7 @@ export async function createExecutionRuntimeHostComposition( dependencies: ExecutionRuntimeHostCompositionDependencies = {}, ): Promise { const storage = await openStorageWriterComposition(context.owner.lease, { + contextOffloadLimits: CONTEXT_OFFLOAD_READER_LIMITS, afterRuntimePolicyOpened: async (stores) => { if (options.bootstrapRuntimePolicy !== false) { await ensureBootstrapRuntimePolicy({ @@ -258,6 +272,10 @@ export async function createExecutionRuntimeHostComposition( const longTermMemoryStore = storage.longTermMemory; const taskLedgerStore = storage.taskLedger; const openedArtifactStore = storage.artifacts; + const openedContextOffloadStore = storage.contextOffload; + if (!openedContextOffloadStore) { + throw new Error('Runtime Host context-offload reader authority is unavailable'); + } const openedUsageStores = storage.usage; const openedShellRunStore = storage.shellRuns; const worktreeChildExecutor = createGitWorktreeChildExecutor({ @@ -672,6 +690,7 @@ export async function createExecutionRuntimeHostComposition( ? {} : { memoryExtraction }), artifacts: openedArtifactStore, + contextOffload: openedContextOffloadStore, executionArtifacts, usage: openedUsageStores, childAgents: bindHostChildAgentBackend( diff --git a/packages/runtime-host/src/server/execution-model-composition.ts b/packages/runtime-host/src/server/execution-model-composition.ts index 0c450b39cd..4dd698761a 100644 --- a/packages/runtime-host/src/server/execution-model-composition.ts +++ b/packages/runtime-host/src/server/execution-model-composition.ts @@ -49,6 +49,8 @@ import { persistProviderRequestCaptureArtifact, type InteractiveArtifactStoreWriter, } from '@maka/storage/artifact-stores'; +import type { InteractiveContextOffloadWriter } from '@maka/storage/context-offload-store'; +import { createReadImageSnapshotStore } from '@maka/storage/read-image-snapshot-store'; import type { RuntimePolicyStoresWriter } from '@maka/storage/runtime-policy-stores'; import type { InteractiveUsageStoresWriter } from '@maka/storage/usage-stores'; import { @@ -70,6 +72,7 @@ export interface HostAiSdkBackendInput { readonly sandboxDiagnostics: SandboxDiagnosticsProvider; readonly memoryExtraction?: HostMemoryExtractionCoordinator; readonly artifacts: HostExecutionArtifactAuthority; + readonly contextOffload?: InteractiveContextOffloadWriter; readonly executionArtifacts: HostExecutionArtifactServices; readonly usage: HostExecutionUsageAuthority; readonly requestDrain: () => void; @@ -381,6 +384,14 @@ export async function createHostAiSdkBackend(input: HostAiSdkBackendInput): Prom readAttachmentBytes: createAttachmentByteReader({ artifactStore: input.artifacts, sessionId: input.context.sessionId, + ...(input.contextOffload + ? { + readImageSnapshots: createReadImageSnapshotStore( + input.contextOffload, + input.context.sessionId, + ), + } + : {}), }), recordToolArtifacts: input.executionArtifacts.recordToolArtifacts, toolResultArchive: input.executionArtifacts.toolResultArchive, diff --git a/packages/runtime/src/__tests__/provider-image-overflow-recovery.test.ts b/packages/runtime/src/__tests__/provider-image-overflow-recovery.test.ts index cf03db6291..a58861f151 100644 --- a/packages/runtime/src/__tests__/provider-image-overflow-recovery.test.ts +++ b/packages/runtime/src/__tests__/provider-image-overflow-recovery.test.ts @@ -21,13 +21,14 @@ import assert from 'node:assert/strict'; import { describe, test } from 'node:test'; import type { RuntimeEvent } from '@maka/core/runtime-event'; +import type { StorageRef } from '@maka/core/events'; import type { ModelMessage } from '../model-protocol.js'; import { collectHistoricalImageToolResults, omitHistoricalImageToolResults, } from '../provider-image-overflow-recovery.js'; -function imageResultEvent(toolCallId: string, relativePath: string): RuntimeEvent { +function imageResultEvent(toolCallId: string, ref: StorageRef): RuntimeEvent { return { id: `event-${toolCallId}`, sessionId: 'session-1', @@ -46,7 +47,7 @@ function imageResultEvent(toolCallId: string, relativePath: string): RuntimeEven result: { kind: 'image', mimeType: 'image/png', - ref: { kind: 'session_file', sessionId: 'session-1', relativePath }, + ref, }, isError: false, }, @@ -84,7 +85,13 @@ function prompt(messages: readonly ModelMessage[]): string { describe('provider image overflow recovery projection', () => { test('omits only eligible historical tool-result images and names their artifact', () => { - const priorEvents = [imageResultEvent('prior-image-call', 'screenshots/screenshot.png')]; + const priorEvents = [ + imageResultEvent('prior-image-call', { + kind: 'session_file', + sessionId: 'session-1', + relativePath: 'screenshots/screenshot.png', + }), + ]; const messages = [ { role: 'user', @@ -111,7 +118,11 @@ describe('provider image overflow recovery projection', () => { const messages = [toolImageMessage('prior-image-call', 'PRIOR_IMAGE')]; const original = structuredClone(messages); const eligible = collectHistoricalImageToolResults([ - imageResultEvent('prior-image-call', 'screenshot.png'), + imageResultEvent('prior-image-call', { + kind: 'session_file', + sessionId: 'session-1', + relativePath: 'screenshot.png', + }), ]); const first = omitHistoricalImageToolResults(messages, eligible); @@ -124,4 +135,20 @@ describe('provider image overflow recovery projection', () => { assert.equal(second.omittedParts, 0); assert.deepEqual(second.messages, first.messages); }); + + test('uses the durable context identity when an omitted image has no file path', () => { + const eligible = collectHistoricalImageToolResults([ + imageResultEvent('prior-image-call', { + kind: 'session_context', + sessionId: 'session-1', + refId: 'read-image:owner-1', + }), + ]); + const result = omitHistoricalImageToolResults( + [toolImageMessage('prior-image-call', 'PRIOR_IMAGE')], + eligible, + ); + + assert.match(prompt(result.messages), /read-image:owner-1/); + }); }); diff --git a/packages/runtime/src/provider-image-overflow-recovery.ts b/packages/runtime/src/provider-image-overflow-recovery.ts index ad31df5dd1..71a90e5e66 100644 --- a/packages/runtime/src/provider-image-overflow-recovery.ts +++ b/packages/runtime/src/provider-image-overflow-recovery.ts @@ -46,6 +46,13 @@ function storageRefLabel(value: unknown): string | undefined { ) { return value.relativePath; } + if ( + value.kind === 'session_context' && + typeof value.refId === 'string' && + value.refId.length > 0 + ) { + return value.refId; + } if ( value.kind === 'external_file' && typeof value.absolutePath === 'string' && diff --git a/packages/storage/package.json b/packages/storage/package.json index 2c4bc7853d..cbcbaddc18 100644 --- a/packages/storage/package.json +++ b/packages/storage/package.json @@ -10,6 +10,7 @@ "./agent-run-store": "./dist/agent-run-store.js", "./artifact-stores": "./dist/artifact-stores.js", "./config-transfer": "./dist/config-transfer.js", + "./context-offload-store": "./dist/context-offload-store.js", "./credential-store": "./dist/credential-store.js", "./daily-review-authority": "./dist/daily-review-authority.js", "./deep-research-authority": "./dist/deep-research-authority.js", @@ -36,6 +37,7 @@ "./project-catalog": "./dist/project-catalog.js", "./project-catalog-authority": "./dist/project-catalog-authority.js", "./root-authority": "./dist/root-authority.js", + "./read-image-snapshot-store": "./dist/read-image-snapshot-store.js", "./runtime-event-persistence": "./dist/runtime-event-persistence.js", "./runtime-policy-stores": "./dist/runtime-policy-stores.js", "./scheduled-task-store": "./dist/scheduled-task-store.js", diff --git a/packages/storage/src/__tests__/artifact-attachments.test.ts b/packages/storage/src/__tests__/artifact-attachments.test.ts index 7a23eadf46..fd0ba6f126 100644 --- a/packages/storage/src/__tests__/artifact-attachments.test.ts +++ b/packages/storage/src/__tests__/artifact-attachments.test.ts @@ -23,6 +23,7 @@ import { tmpdir } from 'node:os'; import { join } from 'node:path'; import { describe, test } from 'node:test'; import { MAX_ATTACHMENT_BYTES } from '@maka/core/attachments'; +import type { ReadImageSnapshotStore } from '@maka/core/context-offload'; import { type StorageRef } from '@maka/core/events'; import { createArtifactAttachmentResourceReader, @@ -124,6 +125,53 @@ describe('artifact attachment authority', () => { }); }); + test('routes durable context refs through the Session-bound snapshot reader', async () => { + await withStore(async (store) => { + const reads: string[] = []; + const readImageSnapshots: ReadImageSnapshotStore = { + async snapshot() { + throw new Error('not used by the reader path'); + }, + async read(ref) { + reads.push(ref.refId); + if (ref.refId === 'missing') return { ok: false, reason: 'not_found' }; + return { + ok: true, + record: { + refId: ref.refId, + sessionId: ref.sessionId, + owner: { kind: 'read_image_snapshot', ownerId: 'owner-1' }, + blobId: 'a'.repeat(64), + sizeBytes: png.byteLength, + mediaType: 'image/png', + createdAt: 1, + }, + bytes: png, + }; + }, + }; + const reader = createAttachmentByteReader({ + artifactStore: store, + sessionId: 'session-1', + readImageSnapshots, + }); + + assert.deepEqual(await reader(sessionContextRef('ref-1')), { + ok: true, + bytes: png, + }); + assert.deepEqual(await reader(sessionContextRef('missing')), { + ok: false, + reason: 'not_found', + }); + assert.deepEqual(await reader(sessionContextRef('ref-2', 'other-session')), { + ok: false, + reason: 'session_mismatch', + }); + assert.deepEqual(reads, ['ref-1', 'missing']); + }); + }); + test('passes through real store not-found and unsupported-mime failures', async () => { await withStore(async (store) => { const reader = createAttachmentByteReader({ @@ -172,6 +220,10 @@ function sessionFileRef(relativePath: string, sessionId = 'session-1'): StorageR return { kind: 'session_file', sessionId, relativePath }; } +function sessionContextRef(refId: string, sessionId = 'session-1'): StorageRef { + return { kind: 'session_context', sessionId, refId }; +} + async function withStore( run: (store: ReturnType) => Promise, ): Promise { diff --git a/packages/storage/src/__tests__/storage-writer-composition.test.ts b/packages/storage/src/__tests__/storage-writer-composition.test.ts index 4c65607f5f..e242eede0f 100644 --- a/packages/storage/src/__tests__/storage-writer-composition.test.ts +++ b/packages/storage/src/__tests__/storage-writer-composition.test.ts @@ -22,6 +22,7 @@ import { mkdtemp, rm } from 'node:fs/promises'; import { tmpdir } from 'node:os'; import { join } from 'node:path'; import { after, test } from 'node:test'; +import type { ContextOffloadLimits } from '@maka/core/context-offload'; import { acquireOperationalStateDatabase } from '../operational-state-store.js'; import { openStorageWriterComposition } from '../storage-writer-composition.js'; import { @@ -38,6 +39,12 @@ import { // temporary root's removal leaves it behind; reclaim the recorded rootIds here. after(removeTrackedControlDirectories); +const contextOffloadLimits: ContextOffloadLimits = Object.freeze({ + ownerMaxBytes: Object.freeze({ read_image_snapshot: 1024, tool_result_archive: 1024 }), + sessionLogicalBytes: 4096, + workspacePhysicalBytes: 4096, +}); + test('storage writer composition rejects reuse until close completes', async () => { const root = await mkdtemp(join(tmpdir(), 'maka-storage-composition-')); try { @@ -88,6 +95,46 @@ test('opening a second storage writer composition creates a usable lifecycle', a } }); +test('context-offload authority is optional and participates in composition close', async () => { + const root = await mkdtemp(join(tmpdir(), 'maka-storage-context-composition-')); + try { + const capability = trackControlDirectory( + await resolveStorageRoot({ path: root, kind: 'interactive' }), + ); + const owner = await tryAcquireInteractiveRootOwner(capability); + assert.ok(owner); + try { + const withoutContext = await openStorageWriterComposition(owner.lease); + assert.equal(withoutContext.contextOffload, undefined); + await withoutContext.close(); + + const withContext = await openStorageWriterComposition(owner.lease, { + contextOffloadLimits, + }); + assert.ok(withContext.contextOffload); + assert.deepEqual( + await withContext.contextOffload.read({ + sessionId: 'session-1', + refId: 'missing', + maxBytes: 1024, + }), + { ok: false, reason: 'not_found' }, + ); + await withContext.close(); + + const reopened = await openStorageWriterComposition(owner.lease, { + contextOffloadLimits, + }); + assert.ok(reopened.contextOffload); + await reopened.close(); + } finally { + await owner.close(); + } + } finally { + await rm(root, { recursive: true, force: true }); + } +}); + test('a failed close keeps the lease unavailable', async () => { const root = await mkdtemp(join(tmpdir(), 'maka-storage-composition-close-failure-')); try { diff --git a/packages/storage/src/artifact-attachments.ts b/packages/storage/src/artifact-attachments.ts index 075ba28dac..4b4cf35147 100644 --- a/packages/storage/src/artifact-attachments.ts +++ b/packages/storage/src/artifact-attachments.ts @@ -24,6 +24,7 @@ import { type AttachmentByteReader, } from '@maka/core/attachments'; import { type StorageRef, type ToolResultContent } from '@maka/core/events'; +import type { ReadImageSnapshotStore } from '@maka/core/context-offload'; import type { ArtifactAuthorityStore, ArtifactStore, @@ -86,10 +87,19 @@ export function createArtifactAttachmentResourceReader(input: { export function createAttachmentByteReader(input: { artifactStore: DurableArtifactAttachmentReader; sessionId: string; + readImageSnapshots?: ReadImageSnapshotStore; maxBytes?: number; }): AttachmentByteReader { const maxBytes = input.maxBytes ?? MAX_ATTACHMENT_BYTES; return async (ref) => { + if (ref.kind === 'session_context') { + if (ref.sessionId !== input.sessionId) return { ok: false, reason: 'session_mismatch' }; + if (!input.readImageSnapshots) return { ok: false, reason: 'unsupported_ref_kind' }; + const result = await input.readImageSnapshots.read(ref); + return result.ok + ? { ok: true, bytes: new Uint8Array(result.bytes) } + : { ok: false, reason: result.reason }; + } if (ref.kind !== 'session_file') return { ok: false, reason: 'unsupported_ref_kind' }; if (ref.sessionId !== input.sessionId) return { ok: false, reason: 'session_mismatch' }; const result = await input.artifactStore.readDurableAttachmentBinary({ diff --git a/packages/storage/src/sqlite-context-offload-store.ts b/packages/storage/src/sqlite-context-offload-store.ts index 5b96a75143..e8c2e4e47c 100644 --- a/packages/storage/src/sqlite-context-offload-store.ts +++ b/packages/storage/src/sqlite-context-offload-store.ts @@ -24,6 +24,7 @@ import { createRequire } from 'node:module'; import { dirname, isAbsolute, join, relative, sep } from 'node:path'; import type { DatabaseSync } from 'node:sqlite'; import { + CONTEXT_OFFLOAD_ID_MAX_CODE_POINTS, type ContextOffloadCopyResult, type ContextOffloadGarbageCollectionResult, type ContextOffloadLimits, @@ -46,7 +47,6 @@ import { 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})$/; @@ -1368,7 +1368,7 @@ function isOwnerKind(value: unknown): value is ContextOffloadOwner['kind'] { } function assertBoundedIdentity(value: string, label: string): void { - assertBoundedText(value, MAX_ID_CODE_POINTS, label); + assertBoundedText(value, CONTEXT_OFFLOAD_ID_MAX_CODE_POINTS, label); } function assertBoundedText(value: string, maxCodePoints: number, label: string): void { diff --git a/packages/storage/src/storage-writer-composition.ts b/packages/storage/src/storage-writer-composition.ts index 5261127c7b..95cbae1695 100644 --- a/packages/storage/src/storage-writer-composition.ts +++ b/packages/storage/src/storage-writer-composition.ts @@ -18,6 +18,8 @@ */ import { openInteractiveArtifactStoreForWrite } from './artifact-stores.js'; +import type { ContextOffloadLimits } from '@maka/core/context-offload'; +import { openInteractiveContextOffloadStoreForWrite } from './context-offload-store.js'; import { openInteractiveDailyReviewAuthorityForWrite } from './daily-review-authority.js'; import { openInteractiveDeepResearchStoreForWrite } from './deep-research-authority.js'; import { openInteractiveExecutionStoresForWrite } from './execution-stores.js'; @@ -38,6 +40,8 @@ export interface OpenStorageWriterCompositionOptions { afterRuntimePolicyOpened?: ( stores: Awaited>, ) => void | Promise; + /** Opens the context-offload authority only when a reader or writer is composed. */ + contextOffloadLimits?: ContextOffloadLimits; } export interface StorageWriterComposition { @@ -53,6 +57,7 @@ export interface StorageWriterComposition { readonly longTermMemory: Awaited>; readonly taskLedger: Awaited>; readonly artifacts: Awaited>; + readonly contextOffload?: Awaited>; readonly usage: Awaited>; readonly shellRuns: Awaited>; close(): Promise; @@ -153,6 +158,16 @@ async function createComposition( () => openInteractiveArtifactStoreForWrite(lease), closeWriter, ); + const contextOffloadLimits = options.contextOffloadLimits; + const contextOffload = contextOffloadLimits + ? await openWriter( + () => + openInteractiveContextOffloadStoreForWrite(lease, { + limits: contextOffloadLimits, + }), + closeWriter, + ) + : undefined; const usage = await openWriter(() => openInteractiveUsageStoresForWrite(lease), closeWriter); const shellRuns = await openWriter( () => openInteractiveShellRunStoreForWrite(lease), @@ -171,6 +186,7 @@ async function createComposition( longTermMemory, taskLedger, artifacts, + ...(contextOffload ? { contextOffload } : {}), usage, shellRuns, close, From a13b3c1b9047a9c159dd65cc61f05469c0cbc1e2 Mon Sep 17 00:00:00 2001 From: likun Date: Sat, 29 Aug 2026 21:53:17 +0800 Subject: [PATCH 2/5] ci(runtime-host): declare compatible context ref reader Generated-by: OpenAI Codex --- .../session-context-ref-reader.json | 5 +++++ 1 file changed, 5 insertions(+) create mode 100644 packages/runtime-host/protocol-compatible-changes/session-context-ref-reader.json diff --git a/packages/runtime-host/protocol-compatible-changes/session-context-ref-reader.json b/packages/runtime-host/protocol-compatible-changes/session-context-ref-reader.json new file mode 100644 index 0000000000..0523f5a773 --- /dev/null +++ b/packages/runtime-host/protocol-compatible-changes/session-context-ref-reader.json @@ -0,0 +1,5 @@ +{ + "epoch": 65, + "files": ["packages/runtime-host/src/protocol/turn.ts"], + "reason": "Widens the Host decoder for a durable context reference variant this reader-only slice does not emit; existing peers exchange the same frames until the writer cutover" +} From 2715ee2f60d68df82da40c9ac209bf37233701fa Mon Sep 17 00:00:00 2001 From: likun Date: Sat, 29 Aug 2026 22:11:32 +0800 Subject: [PATCH 3/5] fix(runtime-host): degrade unavailable context reader --- .../__tests__/execution-composition.test.ts | 22 ++++++++++++++ .../src/server/execution-composition.ts | 10 ++++--- .../storage-writer-composition.test.ts | 30 ++++++++++++++++++- .../storage/src/storage-writer-composition.ts | 27 +++++++++++------ 4 files changed, 75 insertions(+), 14 deletions(-) diff --git a/packages/runtime-host/src/__tests__/execution-composition.test.ts b/packages/runtime-host/src/__tests__/execution-composition.test.ts index 12c83148e5..4eb2376355 100644 --- a/packages/runtime-host/src/__tests__/execution-composition.test.ts +++ b/packages/runtime-host/src/__tests__/execution-composition.test.ts @@ -57,6 +57,7 @@ import { const require = createRequire(import.meta.url); const FAKE_CONNECTION_ID = 'bbbbbbbb-bbbb-4bbb-8bbb-bbbbbbbbbbbb'; +const CONTEXT_OFFLOAD_DATABASE_NAME = 'context-offload.sqlite'; test('filesystem worker follows the candidate executable runtime', () => { assert.equal(runtimeHostFilesystemWorkerRuntime({ electron: '43.1.1' }), 'electron'); @@ -97,6 +98,27 @@ test('production composition owns the long-term memory database lifecycle', asyn }); }); +test('production composition reaches Ready when the optional context reader cannot open', async () => { + await withCompositionRoot(async ({ root, owner }) => { + await mkdir(join(root, CONTEXT_OFFLOAD_DATABASE_NAME)); + const originalConsoleError = console.error; + const diagnostics: string[] = []; + console.error = (...values: unknown[]) => diagnostics.push(values.map(String).join(' ')); + let composition: Awaited> | undefined; + try { + composition = await createExecutionRuntimeHostComposition(compositionContext(owner)); + assert.equal(composition.workspaceExecution.state, 'ready'); + assert.equal( + diagnostics.some((message) => message.includes('optional context-offload reader')), + true, + ); + } finally { + console.error = originalConsoleError; + await composition?.close(); + } + }); +}); + test('production composition closes long-term memory after a later startup failure', async () => { await withCompositionRoot(async ({ root, owner }) => { const stores = await openInteractiveExecutionStoresForWrite(owner.lease); diff --git a/packages/runtime-host/src/server/execution-composition.ts b/packages/runtime-host/src/server/execution-composition.ts index 810d898deb..94dc813e38 100644 --- a/packages/runtime-host/src/server/execution-composition.ts +++ b/packages/runtime-host/src/server/execution-composition.ts @@ -249,6 +249,11 @@ export async function createExecutionRuntimeHostComposition( } }, }); + if (storage.contextOffloadUnavailable) { + console.error( + `[runtime-host] optional context-offload reader could not be opened: ${generalizedErrorMessage(storage.contextOffloadUnavailable.cause)}`, + ); + } const stores = storage.execution; let graphControlStore: ReturnType | undefined; let graphClient: HostAgentGraphCoordinator | undefined; @@ -273,9 +278,6 @@ export async function createExecutionRuntimeHostComposition( const taskLedgerStore = storage.taskLedger; const openedArtifactStore = storage.artifacts; const openedContextOffloadStore = storage.contextOffload; - if (!openedContextOffloadStore) { - throw new Error('Runtime Host context-offload reader authority is unavailable'); - } const openedUsageStores = storage.usage; const openedShellRunStore = storage.shellRuns; const worktreeChildExecutor = createGitWorktreeChildExecutor({ @@ -690,7 +692,7 @@ export async function createExecutionRuntimeHostComposition( ? {} : { memoryExtraction }), artifacts: openedArtifactStore, - contextOffload: openedContextOffloadStore, + ...(openedContextOffloadStore ? { contextOffload: openedContextOffloadStore } : {}), executionArtifacts, usage: openedUsageStores, childAgents: bindHostChildAgentBackend( diff --git a/packages/storage/src/__tests__/storage-writer-composition.test.ts b/packages/storage/src/__tests__/storage-writer-composition.test.ts index e242eede0f..d0c0c02ae1 100644 --- a/packages/storage/src/__tests__/storage-writer-composition.test.ts +++ b/packages/storage/src/__tests__/storage-writer-composition.test.ts @@ -18,12 +18,13 @@ */ import assert from 'node:assert/strict'; -import { mkdtemp, rm } from 'node:fs/promises'; +import { mkdir, mkdtemp, rm } from 'node:fs/promises'; import { tmpdir } from 'node:os'; import { join } from 'node:path'; import { after, test } from 'node:test'; import type { ContextOffloadLimits } from '@maka/core/context-offload'; import { acquireOperationalStateDatabase } from '../operational-state-store.js'; +import { CONTEXT_OFFLOAD_DATABASE_NAME } from '../sqlite-context-offload-store.js'; import { openStorageWriterComposition } from '../storage-writer-composition.js'; import { resolveStorageRoot, @@ -135,6 +136,33 @@ test('context-offload authority is optional and participates in composition clos } }); +test('an unavailable context-offload authority does not fail the storage composition', async () => { + const root = await mkdtemp(join(tmpdir(), 'maka-storage-context-unavailable-')); + try { + await mkdir(join(root, CONTEXT_OFFLOAD_DATABASE_NAME)); + const capability = trackControlDirectory( + await resolveStorageRoot({ path: root, kind: 'interactive' }), + ); + const owner = await tryAcquireInteractiveRootOwner(capability); + assert.ok(owner); + try { + const composition = await openStorageWriterComposition(owner.lease, { + contextOffloadLimits, + }); + assert.equal(composition.contextOffload, undefined); + assert.ok(composition.contextOffloadUnavailable); + assert.ok(composition.contextOffloadUnavailable.cause instanceof Error); + await composition.execution.sessionStore.list(); + await composition.artifacts.listPage('session-1', { offset: 0, limit: 1 }); + await composition.close(); + } finally { + await owner.close(); + } + } finally { + await rm(root, { recursive: true, force: true }); + } +}); + test('a failed close keeps the lease unavailable', async () => { const root = await mkdtemp(join(tmpdir(), 'maka-storage-composition-close-failure-')); try { diff --git a/packages/storage/src/storage-writer-composition.ts b/packages/storage/src/storage-writer-composition.ts index 95cbae1695..9ed4ba4680 100644 --- a/packages/storage/src/storage-writer-composition.ts +++ b/packages/storage/src/storage-writer-composition.ts @@ -58,6 +58,8 @@ export interface StorageWriterComposition { readonly taskLedger: Awaited>; readonly artifacts: Awaited>; readonly contextOffload?: Awaited>; + /** Present when the optional context-offload capability could not be opened. */ + readonly contextOffloadUnavailable?: { readonly cause: unknown }; readonly usage: Awaited>; readonly shellRuns: Awaited>; close(): Promise; @@ -159,15 +161,21 @@ async function createComposition( closeWriter, ); const contextOffloadLimits = options.contextOffloadLimits; - const contextOffload = contextOffloadLimits - ? await openWriter( - () => - openInteractiveContextOffloadStoreForWrite(lease, { - limits: contextOffloadLimits, - }), - closeWriter, - ) - : undefined; + let contextOffload: + | Awaited> + | undefined; + let contextOffloadUnavailable: { readonly cause: unknown } | undefined; + if (contextOffloadLimits) { + try { + const openedContextOffload = await openInteractiveContextOffloadStoreForWrite(lease, { + limits: contextOffloadLimits, + }); + contextOffload = openedContextOffload; + closes.push(() => closeWriter(openedContextOffload)); + } catch (cause) { + contextOffloadUnavailable = Object.freeze({ cause }); + } + } const usage = await openWriter(() => openInteractiveUsageStoresForWrite(lease), closeWriter); const shellRuns = await openWriter( () => openInteractiveShellRunStoreForWrite(lease), @@ -187,6 +195,7 @@ async function createComposition( taskLedger, artifacts, ...(contextOffload ? { contextOffload } : {}), + ...(contextOffloadUnavailable ? { contextOffloadUnavailable } : {}), usage, shellRuns, close, From d87bba84b8b192710ed012439b1f7fd0d416e092 Mon Sep 17 00:00:00 2001 From: likun Date: Sat, 29 Aug 2026 22:34:29 +0800 Subject: [PATCH 4/5] fix(runtime-host): close reader-only context authority --- packages/core/src/context-offload.ts | 8 +-- .../src/__tests__/protocol.test.ts | 25 ++++----- .../session-retirement-coordinator.test.ts | 6 +++ .../src/protocol/hosted-execution.ts | 4 +- packages/runtime-host/src/protocol/message.ts | 3 +- packages/runtime-host/src/protocol/turn.ts | 14 ++++- .../src/server/execution-composition.ts | 26 ++++++++- .../src/server/execution-model-composition.ts | 12 +++-- .../server/session-retirement-coordinator.ts | 4 ++ .../src/__tests__/conversation-copy.test.ts | 38 +++++++++++++ packages/runtime/src/conversation-copy.ts | 4 ++ .../__tests__/artifact-attachments.test.ts | 16 ++++-- .../__tests__/context-offload-store.test.ts | 19 ++++++- .../read-image-snapshot-store.test.ts | 12 ++++- packages/storage/src/artifact-attachments.ts | 12 +++-- packages/storage/src/context-offload-store.ts | 42 +++++++++++++++ .../storage/src/read-image-snapshot-store.ts | 53 +++++++++++++------ 17 files changed, 247 insertions(+), 51 deletions(-) diff --git a/packages/core/src/context-offload.ts b/packages/core/src/context-offload.ts index fa0098a3f1..9c2cb9c1d1 100644 --- a/packages/core/src/context-offload.ts +++ b/packages/core/src/context-offload.ts @@ -118,15 +118,17 @@ export class ReadImageSnapshotStoreError extends Error { } } -export interface ReadImageSnapshotStore { +export interface ReadImageSnapshotReader { + read(input: SessionContextRef): Promise; +} + +export interface ReadImageSnapshotStore extends ReadImageSnapshotReader { 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; } /** diff --git a/packages/runtime-host/src/__tests__/protocol.test.ts b/packages/runtime-host/src/__tests__/protocol.test.ts index b9a4e7b2a3..24c14a6f89 100644 --- a/packages/runtime-host/src/__tests__/protocol.test.ts +++ b/packages/runtime-host/src/__tests__/protocol.test.ts @@ -59,6 +59,7 @@ import { TURN_MESSAGE_QUOTE_MAX_COUNT, TURN_MESSAGE_QUOTE_TEXT_MAX_LENGTH, TURN_FAILURE_MESSAGE_MAX_BYTES, + decodeMessageContent, TURN_SKILL_ID_MAX_COUNT, TURN_SKILL_ID_MAX_LENGTH, } from '../protocol/turn.js'; @@ -1546,18 +1547,18 @@ describe('Runtime Host bootstrap protocol', () => { ), }), ); - assert.doesNotThrow(() => - submit({ - text: 'valid context ref', - attachments: [ - attachmentRef({ - kind: 'session_context', - sessionId: 'session-1', - refId: 'read-image:owner-1', - }), - ], - }), - ); + const contextContent = { + text: 'valid context ref', + attachments: [ + attachmentRef({ + kind: 'session_context' as const, + sessionId: 'session-1', + refId: 'read-image:owner-1', + }), + ], + }; + assert.throws(() => submit(contextContent), isInvalidFrame); + assert.deepEqual(decodeMessageContent(contextContent), contextContent); assert.throws( () => submit({ diff --git a/packages/runtime-host/src/__tests__/session-retirement-coordinator.test.ts b/packages/runtime-host/src/__tests__/session-retirement-coordinator.test.ts index 21e19ef4f9..e0ec3dbca3 100644 --- a/packages/runtime-host/src/__tests__/session-retirement-coordinator.test.ts +++ b/packages/runtime-host/src/__tests__/session-retirement-coordinator.test.ts @@ -228,6 +228,7 @@ describe('Host Session retirement coordinator', () => { 'parent retirement cleanup did not converge', ); assert.deepEqual(new Set(harness.actions.purgedArtifacts), new Set(harness.familyIds)); + assert.deepEqual(new Set(harness.actions.checkedContext), new Set(harness.familyIds)); }); }); @@ -1008,6 +1009,7 @@ interface RetirementActions { readonly retiredCapabilities: string[]; readonly retiredMessages: string[]; readonly purgedArtifacts: string[]; + readonly checkedContext: string[]; readonly purgedTasks: string[]; readonly purgedOperationalState: string[]; readonly purgedAgentGraphs: string[]; @@ -1046,6 +1048,7 @@ async function withHarness( retiredCapabilities: [], retiredMessages: [], purgedArtifacts: [], + checkedContext: [], purgedTasks: [], purgedOperationalState: [], purgedAgentGraphs: [], @@ -1211,6 +1214,9 @@ async function withHarness( actions.purgedTasks.push(sessionId); }, }, + assertNoContextOffloadReferences: async (sessionIds) => { + actions.checkedContext.push(...sessionIds); + }, purgeOperationalState: async (sessionId) => { actions.purgedOperationalState.push(sessionId); }, diff --git a/packages/runtime-host/src/protocol/hosted-execution.ts b/packages/runtime-host/src/protocol/hosted-execution.ts index f1fee49dea..8c468bb97c 100644 --- a/packages/runtime-host/src/protocol/hosted-execution.ts +++ b/packages/runtime-host/src/protocol/hosted-execution.ts @@ -28,7 +28,7 @@ import { import { invalidProtocolFrame } from './errors.js'; import { defineOperation } from './operation-spec.js'; import { decodeSessionCreateInput, type SessionCreateInput } from './session-catalog.js'; -import { decodeMessageContent } from './turn.js'; +import { decodeMessageAdmissionContent, decodeMessageContent } from './turn.js'; const ERRORS = [ 'host_not_ready', @@ -124,7 +124,7 @@ export function decodeHostedExecutionStartInput(value: unknown): HostedExecution return { executionId, session, - content: decodeMessageContent(input.content), + content: decodeMessageAdmissionContent(input.content), ...(input.maxSteps === undefined ? {} : { maxSteps: requirePositiveCount(input.maxSteps, 'maxSteps') }), diff --git a/packages/runtime-host/src/protocol/message.ts b/packages/runtime-host/src/protocol/message.ts index c30546427c..3ca77b6f7a 100644 --- a/packages/runtime-host/src/protocol/message.ts +++ b/packages/runtime-host/src/protocol/message.ts @@ -31,6 +31,7 @@ import { import { defineOperation } from './operation-spec.js'; import { decodeMessageContent, + decodeMessageAdmissionContent, decodeSkillIds, decodeTurnOrchestration, decodeTurnSnapshot, @@ -328,7 +329,7 @@ function decodeTurnMessageSubmitInput(value: unknown): TurnMessageSubmitInput { originHostEpoch: requireId(record.originHostEpoch, 'originHostEpoch'), sessionId: requireEntityId(record.sessionId, 'sessionId'), messageId: requireEntityId(record.messageId, 'messageId'), - content: decodeMessageContent(record.content, skillIds.length > 0), + content: decodeMessageAdmissionContent(record.content, skillIds.length > 0), placement, ...(skillIds.length > 0 ? { skillIds } : {}), ...(turnOrchestration !== undefined ? { turnOrchestration } : {}), diff --git a/packages/runtime-host/src/protocol/turn.ts b/packages/runtime-host/src/protocol/turn.ts index 8aab204b6d..e1a4a5b41e 100644 --- a/packages/runtime-host/src/protocol/turn.ts +++ b/packages/runtime-host/src/protocol/turn.ts @@ -355,7 +355,7 @@ function decodeTurnStartInput(value: unknown): TurnStartInput { return { sessionId: requireEntityId(record.sessionId, 'sessionId'), turnId: requireEntityId(record.turnId, 'turnId'), - content: decodeMessageContent(record.content, skillIds.length > 0), + content: decodeMessageAdmissionContent(record.content, skillIds.length > 0), ...(skillIds.length > 0 ? { skillIds } : {}), ...(record.turnOrchestration !== undefined ? { turnOrchestration: decodeTurnOrchestration(record.turnOrchestration) } @@ -458,6 +458,18 @@ export function decodeMessageContent(value: unknown, allowEmptyText = false): Me return content; } +/** Client-authored Messages cannot claim Host-owned Session context references. */ +export function decodeMessageAdmissionContent( + value: unknown, + allowEmptyText = false, +): MessageContent { + const content = decodeMessageContent(value, allowEmptyText); + if (content.attachments?.some((attachment) => attachment.ref.kind === 'session_context')) { + throw invalidProtocolFrame('Session context references are Host-owned'); + } + return content; +} + function requireUtf8String( value: unknown, label: string, diff --git a/packages/runtime-host/src/server/execution-composition.ts b/packages/runtime-host/src/server/execution-composition.ts index 94dc813e38..1f59857dff 100644 --- a/packages/runtime-host/src/server/execution-composition.ts +++ b/packages/runtime-host/src/server/execution-composition.ts @@ -81,10 +81,14 @@ import { createArtifactAttachmentResourceReader, createReadImageSnapshotter, } from '@maka/storage/artifact-stores'; -import { isSessionNotFoundError } from '@maka/storage/execution-stores'; +import { + isSessionNotFoundError, + SessionMetadataConflictError, +} from '@maka/storage/execution-stores'; import { createExternalSessionAdapterRegistry } from '@maka/storage/external-sessions'; import { createGitWorktreeChildExecutor } from '@maka/storage/git-worktree-child-executor'; import { runWithStorageRootLease } from '@maka/storage/root-authority'; +import { createInteractiveContextOffloadReader } from '@maka/storage/context-offload-store'; import { openStorageWriterComposition } from '@maka/storage/storage-writer-composition'; import type { RuntimePolicyStoresWriter } from '@maka/storage/runtime-policy-stores'; import { resolveWorkspaceIdentity } from '@maka/storage/workspace-identity'; @@ -278,6 +282,9 @@ export async function createExecutionRuntimeHostComposition( const taskLedgerStore = storage.taskLedger; const openedArtifactStore = storage.artifacts; const openedContextOffloadStore = storage.contextOffload; + const openedContextOffloadReader = openedContextOffloadStore + ? createInteractiveContextOffloadReader(openedContextOffloadStore) + : undefined; const openedUsageStores = storage.usage; const openedShellRunStore = storage.shellRuns; const worktreeChildExecutor = createGitWorktreeChildExecutor({ @@ -692,7 +699,8 @@ export async function createExecutionRuntimeHostComposition( ? {} : { memoryExtraction }), artifacts: openedArtifactStore, - ...(openedContextOffloadStore ? { contextOffload: openedContextOffloadStore } : {}), + ...(openedContextOffloadReader ? { contextOffload: openedContextOffloadReader } : {}), + ...(storage.contextOffloadUnavailable ? { contextOffloadUnavailable: true } : {}), executionArtifacts, usage: openedUsageStores, childAgents: bindHostChildAgentBackend( @@ -1453,6 +1461,20 @@ export async function createExecutionRuntimeHostComposition( continuity: continuityCoordinator, artifacts: openedArtifactStore, taskLedger: taskLedgerStore, + assertNoContextOffloadReferences: async (sessionIds) => { + if (!openedContextOffloadStore) { + throw new Error('Context-offload reader is unavailable during Session removal', { + cause: storage.contextOffloadUnavailable?.cause, + }); + } + for (const sessionId of sessionIds) { + if ((await openedContextOffloadStore.usage(sessionId)).references > 0) { + throw new SessionMetadataConflictError( + 'Session removal does not support Session context references yet', + ); + } + } + }, purgeOperationalState: async (sessionId) => { await stores.purgeConversationOperationalState(sessionId); await openedPlanStore.purgeSessionState(sessionId); diff --git a/packages/runtime-host/src/server/execution-model-composition.ts b/packages/runtime-host/src/server/execution-model-composition.ts index 4dd698761a..34337a5215 100644 --- a/packages/runtime-host/src/server/execution-model-composition.ts +++ b/packages/runtime-host/src/server/execution-model-composition.ts @@ -49,8 +49,8 @@ import { persistProviderRequestCaptureArtifact, type InteractiveArtifactStoreWriter, } from '@maka/storage/artifact-stores'; -import type { InteractiveContextOffloadWriter } from '@maka/storage/context-offload-store'; -import { createReadImageSnapshotStore } from '@maka/storage/read-image-snapshot-store'; +import type { InteractiveContextOffloadReader } from '@maka/storage/context-offload-store'; +import { createReadImageSnapshotReader } from '@maka/storage/read-image-snapshot-store'; import type { RuntimePolicyStoresWriter } from '@maka/storage/runtime-policy-stores'; import type { InteractiveUsageStoresWriter } from '@maka/storage/usage-stores'; import { @@ -72,7 +72,8 @@ export interface HostAiSdkBackendInput { readonly sandboxDiagnostics: SandboxDiagnosticsProvider; readonly memoryExtraction?: HostMemoryExtractionCoordinator; readonly artifacts: HostExecutionArtifactAuthority; - readonly contextOffload?: InteractiveContextOffloadWriter; + readonly contextOffload?: InteractiveContextOffloadReader; + readonly contextOffloadUnavailable?: boolean; readonly executionArtifacts: HostExecutionArtifactServices; readonly usage: HostExecutionUsageAuthority; readonly requestDrain: () => void; @@ -386,12 +387,15 @@ export async function createHostAiSdkBackend(input: HostAiSdkBackendInput): Prom sessionId: input.context.sessionId, ...(input.contextOffload ? { - readImageSnapshots: createReadImageSnapshotStore( + readImageSnapshots: createReadImageSnapshotReader( input.contextOffload, input.context.sessionId, ), } : {}), + ...(!input.contextOffload && input.contextOffloadUnavailable + ? { readImageSnapshotsUnavailable: true } + : {}), }), recordToolArtifacts: input.executionArtifacts.recordToolArtifacts, toolResultArchive: input.executionArtifacts.toolResultArchive, diff --git a/packages/runtime-host/src/server/session-retirement-coordinator.ts b/packages/runtime-host/src/server/session-retirement-coordinator.ts index 8d9e713a78..c3cb224fe7 100644 --- a/packages/runtime-host/src/server/session-retirement-coordinator.ts +++ b/packages/runtime-host/src/server/session-retirement-coordinator.ts @@ -121,6 +121,7 @@ export interface HostSessionRetirementCoordinatorOptions { readonly continuity: RetirementContinuity; readonly artifacts: Pick; readonly taskLedger: Pick; + readonly assertNoContextOffloadReferences?: (sessionIds: readonly string[]) => Promise; readonly purgeOperationalState: (sessionId: string) => Promise; readonly purgeAgentGraphState: (sessionId: string) => Promise; readonly worktrees?: Pick; @@ -191,6 +192,7 @@ export class HostSessionRetirementCoordinator { readonly #continuity: RetirementContinuity; readonly #artifacts: HostSessionRetirementCoordinatorOptions['artifacts']; readonly #taskLedger: HostSessionRetirementCoordinatorOptions['taskLedger']; + readonly #assertNoContextOffloadReferences: HostSessionRetirementCoordinatorOptions['assertNoContextOffloadReferences']; readonly #purgeOperationalState: HostSessionRetirementCoordinatorOptions['purgeOperationalState']; readonly #purgeAgentGraphState: HostSessionRetirementCoordinatorOptions['purgeAgentGraphState']; readonly #worktrees: HostSessionRetirementCoordinatorOptions['worktrees']; @@ -218,6 +220,7 @@ export class HostSessionRetirementCoordinator { this.#continuity = options.continuity; this.#artifacts = options.artifacts; this.#taskLedger = options.taskLedger; + this.#assertNoContextOffloadReferences = options.assertNoContextOffloadReferences; this.#purgeOperationalState = options.purgeOperationalState; this.#purgeAgentGraphState = options.purgeAgentGraphState; this.#worktrees = options.worktrees; @@ -333,6 +336,7 @@ export class HostSessionRetirementCoordinator { if (plan.archive.sessionIds.length > 0) { archiveHandles = await this.#prepareRetirement(plan.archive, 'archive'); } + await this.#assertNoContextOffloadReferences?.(plan.remove.sessionIds); const allSessionIds = [...plan.remove.sessionIds, ...plan.archive.sessionIds]; await this.#finalizeWorkspacePatches(allSessionIds); await this.#disposeBackends(allSessionIds); diff --git a/packages/runtime/src/__tests__/conversation-copy.test.ts b/packages/runtime/src/__tests__/conversation-copy.test.ts index 87223750e7..7d7db16e9e 100644 --- a/packages/runtime/src/__tests__/conversation-copy.test.ts +++ b/packages/runtime/src/__tests__/conversation-copy.test.ts @@ -887,6 +887,44 @@ test('conversation copy rewrites owned references without changing opaque tool p ); }); +test('reader-only conversation copy rejects source-owned Session context refs', () => { + const message: StoredMessage = { + type: 'user', + id: 'user-context', + turnId: 'turn-1', + ts: 1, + text: 'context', + attachments: [ + { + kind: 'image', + name: 'snapshot.png', + mimeType: 'image/png', + bytes: 4, + ref: { + kind: 'session_context', + sessionId: 'session-source', + refId: 'context-source', + }, + }, + ], + }; + assert.throws( + () => + rewriteConversationCopyMessage(message, { + mode: 'exact', + sourceSessionId: 'session-source', + targetSessionId: 'session-target', + artifactIds: new Map(), + relativePaths: new Map(), + linkedChildren: { mode: 'reject' }, + runIds: new Map(), + runtimeEventIds: new Map(), + providerTraceIds: new Map(), + }), + /does not support Session context references yet/, + ); +}); + test('conversation copy rejects continuation authority selected through the child-run closure', async () => { const parent = agentRunHeader({ runId: 'run-parent', turnId: 'turn-parent' }); const child = agentRunHeader({ diff --git a/packages/runtime/src/conversation-copy.ts b/packages/runtime/src/conversation-copy.ts index 5e1722c44d..85867ee812 100644 --- a/packages/runtime/src/conversation-copy.ts +++ b/packages/runtime/src/conversation-copy.ts @@ -1510,6 +1510,10 @@ function rewriteStorageRef( ref: StorageRef, references: ConversationCopyArtifactReferenceMap, ): StorageRef { + if (ref.kind === 'session_context' && ref.sessionId === references.sourceSessionId) { + if (references.mode === 'preserve_external') return ref; + throw new Error('Conversation copy does not support Session context references yet'); + } if (ref.kind !== 'session_file' || ref.sessionId !== references.sourceSessionId) return ref; if (references.mode === 'preserve_external') return ref; const artifactId = references.artifactIds.get(ref.relativePath); diff --git a/packages/storage/src/__tests__/artifact-attachments.test.ts b/packages/storage/src/__tests__/artifact-attachments.test.ts index fd0ba6f126..a84c3366e9 100644 --- a/packages/storage/src/__tests__/artifact-attachments.test.ts +++ b/packages/storage/src/__tests__/artifact-attachments.test.ts @@ -23,7 +23,7 @@ import { tmpdir } from 'node:os'; import { join } from 'node:path'; import { describe, test } from 'node:test'; import { MAX_ATTACHMENT_BYTES } from '@maka/core/attachments'; -import type { ReadImageSnapshotStore } from '@maka/core/context-offload'; +import type { ReadImageSnapshotReader } from '@maka/core/context-offload'; import { type StorageRef } from '@maka/core/events'; import { createArtifactAttachmentResourceReader, @@ -122,16 +122,22 @@ describe('artifact attachment authority', () => { ok: false, reason: 'too_large', }); + const unavailableReader = createAttachmentByteReader({ + artifactStore: store, + sessionId: 'session-1', + readImageSnapshotsUnavailable: true, + }); + assert.deepEqual(await unavailableReader(sessionContextRef('ref-1')), { + ok: false, + reason: 'unavailable', + }); }); }); test('routes durable context refs through the Session-bound snapshot reader', async () => { await withStore(async (store) => { const reads: string[] = []; - const readImageSnapshots: ReadImageSnapshotStore = { - async snapshot() { - throw new Error('not used by the reader path'); - }, + const readImageSnapshots: ReadImageSnapshotReader = { async read(ref) { reads.push(ref.refId); if (ref.refId === 'missing') return { ok: false, reason: 'not_found' }; diff --git a/packages/storage/src/__tests__/context-offload-store.test.ts b/packages/storage/src/__tests__/context-offload-store.test.ts index 3dae2c40f6..d82a0a7f24 100644 --- a/packages/storage/src/__tests__/context-offload-store.test.ts +++ b/packages/storage/src/__tests__/context-offload-store.test.ts @@ -24,8 +24,11 @@ import { join } from 'node:path'; import { after, test } from 'node:test'; import type { ContextOffloadLimits } from '@maka/core/context-offload'; import { + authenticateInteractiveContextOffloadReader, authenticateInteractiveContextOffloadWriter, + createInteractiveContextOffloadReader, openInteractiveContextOffloadStoreForWrite, + type InteractiveContextOffloadReader, type InteractiveContextOffloadWriter, } from '../context-offload-store.js'; import { @@ -55,6 +58,10 @@ test('requires authentic Storage Root leases and writer facades', async () => { () => authenticateInteractiveContextOffloadWriter({} as InteractiveContextOffloadWriter), invalidLease, ); + assert.throws( + () => authenticateInteractiveContextOffloadReader({} as InteractiveContextOffloadReader), + invalidLease, + ); }); test('single-flights one limit-bound writer and snapshots admitted inputs', async () => { @@ -79,6 +86,10 @@ test('single-flights one limit-bound writer and snapshots admitted inputs', asyn await conflictingOpening; assert.strictEqual(second, first); assert.strictEqual(authenticateInteractiveContextOffloadWriter(first), first); + const reader = createInteractiveContextOffloadReader(first); + assert.strictEqual(createInteractiveContextOffloadReader(first), reader); + assert.strictEqual(authenticateInteractiveContextOffloadReader(reader), reader); + assert.deepEqual(Object.keys(reader).sort(), ['access', 'kind', 'read']); assert.equal((await stat(join(root, CONTEXT_OFFLOAD_DATABASE_NAME))).isFile(), true); const bytes = new TextEncoder().encode('safe'); @@ -100,7 +111,7 @@ test('single-flights one limit-bound writer and snapshots admitted inputs', asyn assert.equal(stored.record.owner.ownerId, 'source-owner'); assert.equal(stored.record.mediaType, 'application/json'); assert.deepEqual( - await first.read({ sessionId: 'source', refId: stored.record.refId, maxBytes: 64 }), + await reader.read({ sessionId: 'source', refId: stored.record.refId, maxBytes: 64 }), { ok: true, record: stored.record, bytes: new TextEncoder().encode('safe') }, ); @@ -150,6 +161,7 @@ test('close drains admitted work, revokes the facade, and permits a clean reopen bytes: new TextEncoder().encode('image'), mediaType: 'image/png', }); + const reader = createInteractiveContextOffloadReader(writer); const closing = writer.close(); const reopening = openInteractiveContextOffloadStoreForWrite(owner.lease, { limits: testLimits(), @@ -158,6 +170,11 @@ test('close drains admitted work, revokes the facade, and permits a clean reopen await closing; await assert.rejects(writer.usage(), invalidLease); assert.throws(() => authenticateInteractiveContextOffloadWriter(writer), invalidLease); + assert.throws(() => authenticateInteractiveContextOffloadReader(reader), invalidLease); + await assert.rejects( + reader.read({ sessionId: 'session-1', refId: 'ref-1', maxBytes: 64 }), + invalidLease, + ); const reopened = await reopening; try { diff --git a/packages/storage/src/__tests__/read-image-snapshot-store.test.ts b/packages/storage/src/__tests__/read-image-snapshot-store.test.ts index c44d433773..3c9ee900b0 100644 --- a/packages/storage/src/__tests__/read-image-snapshot-store.test.ts +++ b/packages/storage/src/__tests__/read-image-snapshot-store.test.ts @@ -25,10 +25,14 @@ 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 { + createInteractiveContextOffloadReader, openInteractiveContextOffloadStoreForWrite, type InteractiveContextOffloadWriter, } from '../context-offload-store.js'; -import { createReadImageSnapshotStore } from '../read-image-snapshot-store.js'; +import { + createReadImageSnapshotReader, + createReadImageSnapshotStore, +} from '../read-image-snapshot-store.js'; import { resolveStorageRoot, tryAcquireInteractiveRootOwner, @@ -82,6 +86,12 @@ test('snapshots one stable Read image identity and authorizes reads by Session', assert.equal(read.record.owner.ownerId, 'read-call-1'); assert.equal(read.record.mediaType, 'image/png'); assert.deepEqual(read.bytes, new TextEncoder().encode('image')); + const reader = createReadImageSnapshotReader( + createInteractiveContextOffloadReader(writer), + 'session-1', + ); + assert.deepEqual(await reader.read(ref), read); + assert.deepEqual(Object.keys(reader), ['read']); assert.deepEqual(await images.read({ ...ref, sessionId: 'session-2' }), { ok: false, reason: 'session_mismatch', diff --git a/packages/storage/src/artifact-attachments.ts b/packages/storage/src/artifact-attachments.ts index 4b4cf35147..c274c0c628 100644 --- a/packages/storage/src/artifact-attachments.ts +++ b/packages/storage/src/artifact-attachments.ts @@ -24,7 +24,7 @@ import { type AttachmentByteReader, } from '@maka/core/attachments'; import { type StorageRef, type ToolResultContent } from '@maka/core/events'; -import type { ReadImageSnapshotStore } from '@maka/core/context-offload'; +import type { ReadImageSnapshotReader } from '@maka/core/context-offload'; import type { ArtifactAuthorityStore, ArtifactStore, @@ -87,14 +87,20 @@ export function createArtifactAttachmentResourceReader(input: { export function createAttachmentByteReader(input: { artifactStore: DurableArtifactAttachmentReader; sessionId: string; - readImageSnapshots?: ReadImageSnapshotStore; + readImageSnapshots?: ReadImageSnapshotReader; + readImageSnapshotsUnavailable?: boolean; maxBytes?: number; }): AttachmentByteReader { const maxBytes = input.maxBytes ?? MAX_ATTACHMENT_BYTES; return async (ref) => { if (ref.kind === 'session_context') { if (ref.sessionId !== input.sessionId) return { ok: false, reason: 'session_mismatch' }; - if (!input.readImageSnapshots) return { ok: false, reason: 'unsupported_ref_kind' }; + if (!input.readImageSnapshots) { + return { + ok: false, + reason: input.readImageSnapshotsUnavailable ? 'unavailable' : 'unsupported_ref_kind', + }; + } const result = await input.readImageSnapshots.read(ref); return result.ok ? { ok: true, bytes: new Uint8Array(result.bytes) } diff --git a/packages/storage/src/context-offload-store.ts b/packages/storage/src/context-offload-store.ts index c3e4f07b20..1aff861e80 100644 --- a/packages/storage/src/context-offload-store.ts +++ b/packages/storage/src/context-offload-store.ts @@ -35,7 +35,10 @@ import { } from './sqlite-context-offload-store.js'; const writerBrand: unique symbol = Symbol('InteractiveContextOffloadWriter'); +const readerBrand: unique symbol = Symbol('InteractiveContextOffloadReader'); const writers = new WeakSet(); +const readers = new WeakSet(); +const readerByWriter = new WeakMap(); const writerByLease = new WeakMap< object, { readonly writer: InteractiveContextOffloadWriter; readonly limitsKey: string } @@ -60,6 +63,12 @@ export interface InteractiveContextOffloadWriter extends Omit; } +export interface InteractiveContextOffloadReader extends Pick { + readonly kind: 'interactive'; + readonly access: 'read'; + readonly [readerBrand]: true; +} + export function authenticateInteractiveContextOffloadWriter( writer: InteractiveContextOffloadWriter, ): InteractiveContextOffloadWriter { @@ -72,6 +81,37 @@ export function authenticateInteractiveContextOffloadWriter( return writer; } +export function authenticateInteractiveContextOffloadReader( + reader: InteractiveContextOffloadReader, +): InteractiveContextOffloadReader { + if (!readers.has(reader)) { + throw new StorageRootAuthorityError( + 'invalid_lease', + 'Expected an authentic interactive context-offload reader', + ); + } + return reader; +} + +/** Narrows an authenticated writer to the read-only authority used by model hydration. */ +export function createInteractiveContextOffloadReader( + writer: InteractiveContextOffloadWriter, +): InteractiveContextOffloadReader { + const authenticated = authenticateInteractiveContextOffloadWriter(writer); + const existing = readerByWriter.get(authenticated); + if (existing) return existing; + const reader: InteractiveContextOffloadReader = Object.freeze({ + kind: 'interactive', + access: 'read', + [readerBrand]: true as const, + read: (input: Parameters[0]) => + authenticated.read(Object.freeze({ ...input })), + }); + readers.add(reader); + readerByWriter.set(authenticated, reader); + return reader; +} + /** * Opens context-offload storage through an authenticated interactive write * lease. Production callers must use this facade instead of constructing the @@ -203,6 +243,8 @@ function createWriterFacade( if (closeTask) return closeTask; closed = true; if (writerByLease.get(lease)?.writer === writer) writerByLease.delete(lease); + const reader = readerByWriter.get(writer); + if (reader) readers.delete(reader); writers.delete(writer); let pending!: Promise; pending = (async () => { diff --git a/packages/storage/src/read-image-snapshot-store.ts b/packages/storage/src/read-image-snapshot-store.ts index ab1c73aa8f..2bdc3d6cb0 100644 --- a/packages/storage/src/read-image-snapshot-store.ts +++ b/packages/storage/src/read-image-snapshot-store.ts @@ -21,13 +21,45 @@ import { MAX_READ_IMAGE_BYTES } from '@maka/core/attachments'; import { ReadImageSnapshotStoreError, type ContextOffloadReadResult, + type ReadImageSnapshotReader, type ReadImageSnapshotStore, + type SessionContextRef, } from '@maka/core/context-offload'; import { + authenticateInteractiveContextOffloadReader, authenticateInteractiveContextOffloadWriter, + createInteractiveContextOffloadReader, + type InteractiveContextOffloadReader, type InteractiveContextOffloadWriter, } from './context-offload-store.js'; +/** Derives the Read image hydration contract from an authenticated reader. */ +export function createReadImageSnapshotReader( + reader: InteractiveContextOffloadReader, + sessionId: string, +): ReadImageSnapshotReader { + const store = authenticateInteractiveContextOffloadReader(reader); + if (!sessionId) throw new Error('Read image snapshot Session id is required'); + return Object.freeze({ + async read(input: SessionContextRef): 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; + }, + }); +} + /** Derives the Read image domain contract from the authenticated byte authority. */ export function createReadImageSnapshotStore( writer: InteractiveContextOffloadWriter, @@ -35,6 +67,10 @@ export function createReadImageSnapshotStore( ): ReadImageSnapshotStore { const store = authenticateInteractiveContextOffloadWriter(writer); if (!sessionId) throw new Error('Read image snapshot Session id is required'); + const reader = createReadImageSnapshotReader( + createInteractiveContextOffloadReader(store), + sessionId, + ); const facade: ReadImageSnapshotStore = { async snapshot(input) { const accepted = Object.freeze({ @@ -73,22 +109,7 @@ export function createReadImageSnapshotStore( }); }, - 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; - }, + read: (input) => reader.read(input), }; return Object.freeze(facade); } From 407a687286fb284205a1de4f668b15f5613f5789 Mon Sep 17 00:00:00 2001 From: likun Date: Sat, 29 Aug 2026 22:57:48 +0800 Subject: [PATCH 5/5] ci(runtime-host): align context reader protocol declaration --- .../session-context-ref-reader.json | 10 +++++++--- 1 file changed, 7 insertions(+), 3 deletions(-) diff --git a/packages/runtime-host/protocol-compatible-changes/session-context-ref-reader.json b/packages/runtime-host/protocol-compatible-changes/session-context-ref-reader.json index 0523f5a773..06f1186853 100644 --- a/packages/runtime-host/protocol-compatible-changes/session-context-ref-reader.json +++ b/packages/runtime-host/protocol-compatible-changes/session-context-ref-reader.json @@ -1,5 +1,9 @@ { - "epoch": 65, - "files": ["packages/runtime-host/src/protocol/turn.ts"], - "reason": "Widens the Host decoder for a durable context reference variant this reader-only slice does not emit; existing peers exchange the same frames until the writer cutover" + "epoch": 67, + "files": [ + "packages/runtime-host/src/protocol/hosted-execution.ts", + "packages/runtime-host/src/protocol/message.ts", + "packages/runtime-host/src/protocol/turn.ts" + ], + "reason": "Widens Host result decoding for a durable context reference variant while client admission rejects Host-owned refs; this reader-only slice emits no new wire frames" }