From cbb11b1d28434e21c8391ec3811a70e7650fe69b Mon Sep 17 00:00:00 2001 From: likun Date: Sat, 29 Aug 2026 22:29:54 +0800 Subject: [PATCH 1/5] feat(runtime-host): write Read images to Session context --- .../__tests__/execution-composition.test.ts | 4 +- .../src/__tests__/protocol.test.ts | 4 + .../session-retirement-coordinator.test.ts | 13 ++-- packages/runtime-host/src/protocol/index.ts | 3 +- .../src/server/execution-composition.ts | 69 +++++++++-------- .../server/session-retirement-coordinator.ts | 9 ++- .../server/session-revision-coordinator.ts | 32 ++++++++ .../src/server/session-sidecar-purge.ts | 3 + .../builtin-tools-file-worker.test.ts | 15 ++-- .../src/__tests__/conversation-copy.test.ts | 77 ++++++++++--------- .../__tests__/filesystem-authority.test.ts | 2 +- packages/runtime/src/builtin-tools.ts | 8 +- packages/runtime/src/conversation-copy.ts | 50 ++++++++++-- 13 files changed, 193 insertions(+), 96 deletions(-) diff --git a/packages/runtime-host/src/__tests__/execution-composition.test.ts b/packages/runtime-host/src/__tests__/execution-composition.test.ts index b43bafeccb..a427b76a36 100644 --- a/packages/runtime-host/src/__tests__/execution-composition.test.ts +++ b/packages/runtime-host/src/__tests__/execution-composition.test.ts @@ -103,7 +103,7 @@ test('production composition owns the long-term memory database lifecycle', asyn }); }); -test('production composition reaches Ready when the optional context reader cannot open', async () => { +test('production composition reaches Ready when the optional context Store cannot open', async () => { await withCompositionRoot(async ({ root, owner }) => { await mkdir(join(root, CONTEXT_OFFLOAD_DATABASE_NAME)); const originalConsoleError = console.error; @@ -114,7 +114,7 @@ test('production composition reaches Ready when the optional context reader cann composition = await createExecutionRuntimeHostComposition(compositionContext(owner)); assert.equal(composition.workspaceExecution.state, 'ready'); assert.equal( - diagnostics.some((message) => message.includes('optional context-offload reader')), + diagnostics.some((message) => message.includes('optional context-offload Store')), true, ); } finally { diff --git a/packages/runtime-host/src/__tests__/protocol.test.ts b/packages/runtime-host/src/__tests__/protocol.test.ts index 1989f4b504..95f1e4672d 100644 --- a/packages/runtime-host/src/__tests__/protocol.test.ts +++ b/packages/runtime-host/src/__tests__/protocol.test.ts @@ -140,6 +140,10 @@ describe('Runtime Host bootstrap protocol', () => { assert.ok(RUNTIME_HOST_COMPATIBILITY_EPOCH > 78); }); + test('publishes a new compatibility epoch for Read image Session context refs', () => { + assert.ok(RUNTIME_HOST_COMPATIBILITY_EPOCH > 87); + }); + test('rejects the legacy connection update result in the current compatibility epoch', () => { assert.throws( () => 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 affd668029..ed1b35b5db 100644 --- a/packages/runtime-host/src/__tests__/session-retirement-coordinator.test.ts +++ b/packages/runtime-host/src/__tests__/session-retirement-coordinator.test.ts @@ -239,7 +239,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)); + assert.deepEqual(new Set(harness.actions.retiredContext), new Set(harness.familyIds)); }); }); @@ -1027,7 +1027,7 @@ interface RetirementActions { readonly retiredCapabilities: string[]; readonly retiredMessages: string[]; readonly purgedArtifacts: string[]; - readonly checkedContext: string[]; + readonly retiredContext: string[]; readonly purgedTasks: string[]; readonly purgedOperationalState: string[]; readonly purgedAgentGraphs: string[]; @@ -1066,7 +1066,7 @@ async function withHarness( retiredCapabilities: [], retiredMessages: [], purgedArtifacts: [], - checkedContext: [], + retiredContext: [], purgedTasks: [], purgedOperationalState: [], purgedAgentGraphs: [], @@ -1232,8 +1232,11 @@ async function withHarness( actions.purgedTasks.push(sessionId); }, }, - assertNoContextOffloadReferences: async (sessionIds) => { - actions.checkedContext.push(...sessionIds); + contextOffload: { + retireSession: async (sessionId) => { + actions.retiredContext.push(sessionId); + return { releasedReferences: 0, releasedLogicalBytes: 0 }; + }, }, purgeOperationalState: async (sessionId) => { actions.purgedOperationalState.push(sessionId); diff --git a/packages/runtime-host/src/protocol/index.ts b/packages/runtime-host/src/protocol/index.ts index bc5f27e602..6111296348 100644 --- a/packages/runtime-host/src/protocol/index.ts +++ b/packages/runtime-host/src/protocol/index.ts @@ -100,7 +100,8 @@ export const RUNTIME_HOST_REGISTRATION_SCHEMA_VERSION = 1 as const; export const RUNTIME_HOST_PROTOCOL_VERSION = 0 as const; // Increment when the same protocol version no longer guarantees safe Client-Host // interoperability. Mismatches are rejected before domain commands are admitted. -export const RUNTIME_HOST_COMPATIBILITY_EPOCH = 93 as const; +export const RUNTIME_HOST_COMPATIBILITY_EPOCH = 94 as const; +// 94: Read image tool results may carry durable `session_context` refs. // 93: Configuration credential transfer binds proxy destinations and // Connection credentials to exact Host-owned targets before secret access. // Proxy policy and credentials commit through one recoverable Host command; diff --git a/packages/runtime-host/src/server/execution-composition.ts b/packages/runtime-host/src/server/execution-composition.ts index 7200050d7a..2ae0681ad1 100644 --- a/packages/runtime-host/src/server/execution-composition.ts +++ b/packages/runtime-host/src/server/execution-composition.ts @@ -77,14 +77,9 @@ import { import { type MakaTool } from '@maka/runtime/tool-runtime'; import { type RuntimeHostedRootAuthority } from '@maka/runtime/message-authority'; import { createAgentGraphControlStore } from '@maka/storage/agent-graph-control-store'; -import { - createArtifactAttachmentResourceReader, - createReadImageSnapshotter, -} from '@maka/storage/artifact-stores'; -import { - isSessionNotFoundError, - SessionMetadataConflictError, -} from '@maka/storage/execution-stores'; +import { createArtifactAttachmentResourceReader } from '@maka/storage/artifact-stores'; +import { createReadImageSnapshotStore } from '@maka/storage/read-image-snapshot-store'; +import { isSessionNotFoundError } 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'; @@ -208,15 +203,16 @@ export interface ExecutionRuntimeHostComposition extends RuntimeHostComposition readonly plugins: HostPluginPlatform; } -const CONTEXT_OFFLOAD_READER_LIMITS: ContextOffloadLimits = Object.freeze({ +const GIBIBYTE = 1024 * 1024 * 1024; +const CONTEXT_OFFLOAD_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, + // Read images are bounded individually and logically per Session. Physical + // bytes are content-addressed across Sessions and bounded per workspace. + sessionLogicalBytes: GIBIBYTE, + workspacePhysicalBytes: 20 * GIBIBYTE, }); export interface CreateExecutionRuntimeHostCompositionOptions { @@ -245,7 +241,7 @@ export async function createExecutionRuntimeHostComposition( dependencies: ExecutionRuntimeHostCompositionDependencies = {}, ): Promise { const storage = await openStorageWriterComposition(context.owner.lease, { - contextOffloadLimits: CONTEXT_OFFLOAD_READER_LIMITS, + contextOffloadLimits: CONTEXT_OFFLOAD_LIMITS, afterRuntimePolicyOpened: async (stores) => { if (options.bootstrapRuntimePolicy !== false) { await ensureBootstrapRuntimePolicy({ @@ -261,7 +257,7 @@ export async function createExecutionRuntimeHostComposition( }); if (storage.contextOffloadUnavailable) { console.error( - `[runtime-host] optional context-offload reader could not be opened: ${generalizedErrorMessage(storage.contextOffloadUnavailable.cause)}`, + `[runtime-host] optional context-offload Store could not be opened: ${generalizedErrorMessage(storage.contextOffloadUnavailable.cause)}`, ); } const stores = storage.execution; @@ -294,6 +290,17 @@ export async function createExecutionRuntimeHostComposition( const openedContextOffloadReader = openedContextOffloadStore ? createInteractiveContextOffloadReader(openedContextOffloadStore) : undefined; + const contextOffloadRetirement = openedContextOffloadStore + ? openedContextOffloadStore + : storage.contextOffloadUnavailable + ? { + retireSession: async (_sessionId: string): Promise => { + throw new Error('Context-offload Store is unavailable during Session retirement', { + cause: storage.contextOffloadUnavailable?.cause, + }); + }, + } + : undefined; const openedUsageStores = storage.usage; const openedShellRunStore = storage.shellRuns; const worktreeChildExecutor = createGitWorktreeChildExecutor({ @@ -403,7 +410,21 @@ export async function createExecutionRuntimeHostComposition( }), backgroundTasks: runtimeResources, ptyControls: runtimeResources, - snapshotImage: createReadImageSnapshotter(openedArtifactStore), + ...(openedContextOffloadStore + ? { + snapshotImage: async (input: { + readonly sessionId: string; + readonly ownerId: string; + readonly bytes: Uint8Array; + readonly mimeType: string; + }) => + createReadImageSnapshotStore(openedContextOffloadStore, input.sessionId).snapshot({ + ownerId: input.ownerId, + bytes: input.bytes, + mimeType: input.mimeType, + }), + } + : {}), ...(sandboxManager ? { sandboxManager } : {}), ...(filesystemWorker ? { filesystemWorker } : {}), }; @@ -1548,6 +1569,7 @@ export async function createExecutionRuntimeHostComposition( stores, artifacts: openedArtifactStore, sessionTodo: sessionTodoStore, + ...(contextOffloadRetirement ? { contextOffload: contextOffloadRetirement } : {}), manager, admission: sessionAdmission, continuity: continuityCoordinator, @@ -1572,20 +1594,7 @@ export async function createExecutionRuntimeHostComposition( continuity: continuityCoordinator, artifacts: openedArtifactStore, sessionTodo: sessionTodoStore, - 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', - ); - } - } - }, + ...(contextOffloadRetirement ? { contextOffload: contextOffloadRetirement } : {}), purgeOperationalState: async (sessionId) => { await stores.purgeConversationOperationalState(sessionId); await openedPlanStore.purgeSessionState(sessionId); diff --git a/packages/runtime-host/src/server/session-retirement-coordinator.ts b/packages/runtime-host/src/server/session-retirement-coordinator.ts index 72c6e9632f..52b672de96 100644 --- a/packages/runtime-host/src/server/session-retirement-coordinator.ts +++ b/packages/runtime-host/src/server/session-retirement-coordinator.ts @@ -32,6 +32,7 @@ import { } from '@maka/storage/execution-stores'; import { type SessionManager } from '@maka/runtime/session-manager'; import type { InteractiveSessionTodoWriter } from '@maka/storage/session-todo-authority'; +import type { InteractiveContextOffloadWriter } from '@maka/storage/context-offload-store'; import { type OperationOutcome, type SessionCatalogItem, @@ -122,7 +123,7 @@ export interface HostSessionRetirementCoordinatorOptions { readonly continuity: RetirementContinuity; readonly artifacts: Pick; readonly sessionTodo: Pick; - readonly assertNoContextOffloadReferences?: (sessionIds: readonly string[]) => Promise; + readonly contextOffload?: Pick; readonly purgeOperationalState: (sessionId: string) => Promise; readonly purgeAgentGraphState: (sessionId: string) => Promise; readonly worktrees?: Pick; @@ -194,7 +195,7 @@ export class HostSessionRetirementCoordinator { readonly #continuity: RetirementContinuity; readonly #artifacts: HostSessionRetirementCoordinatorOptions['artifacts']; readonly #sessionTodo: HostSessionRetirementCoordinatorOptions['sessionTodo']; - readonly #assertNoContextOffloadReferences: HostSessionRetirementCoordinatorOptions['assertNoContextOffloadReferences']; + readonly #contextOffload: HostSessionRetirementCoordinatorOptions['contextOffload']; readonly #purgeOperationalState: HostSessionRetirementCoordinatorOptions['purgeOperationalState']; readonly #purgeAgentGraphState: HostSessionRetirementCoordinatorOptions['purgeAgentGraphState']; readonly #worktrees: HostSessionRetirementCoordinatorOptions['worktrees']; @@ -222,7 +223,7 @@ export class HostSessionRetirementCoordinator { this.#continuity = options.continuity; this.#artifacts = options.artifacts; this.#sessionTodo = options.sessionTodo; - this.#assertNoContextOffloadReferences = options.assertNoContextOffloadReferences; + this.#contextOffload = options.contextOffload; this.#purgeOperationalState = options.purgeOperationalState; this.#purgeAgentGraphState = options.purgeAgentGraphState; this.#worktrees = options.worktrees; @@ -338,7 +339,6 @@ 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); @@ -725,6 +725,7 @@ export class HostSessionRetirementCoordinator { { artifacts: this.#artifacts, sessionTodo: this.#sessionTodo, + ...(this.#contextOffload ? { contextOffload: this.#contextOffload } : {}), purgeOperationalState: this.#purgeOperationalState, }, sessionId, diff --git a/packages/runtime-host/src/server/session-revision-coordinator.ts b/packages/runtime-host/src/server/session-revision-coordinator.ts index 16c918fff4..dd051dcfc8 100644 --- a/packages/runtime-host/src/server/session-revision-coordinator.ts +++ b/packages/runtime-host/src/server/session-revision-coordinator.ts @@ -35,6 +35,7 @@ import { archivedToolResultContainsConversationOwnedReferences, cloneConversationRuntimeLedger, collectConversationCopyLinkedChildReferences, + collectConversationCopySessionContextRefIds, collectConversationCopySessionFileRefs, createConversationCopySlice, prepareConversationRuntimeLedgerCopy, @@ -61,6 +62,7 @@ import { authenticateInteractiveSessionTodoWriter, type InteractiveSessionTodoWriter, } from '@maka/storage/session-todo-authority'; +import type { InteractiveContextOffloadWriter } from '@maka/storage/context-offload-store'; import type { OperationOutcome, SessionConversationCopyInput, @@ -106,6 +108,10 @@ export interface HostSessionRevisionCoordinatorOptions { readonly stores: ExecutionStoresWriter<'interactive'>; readonly artifacts: InteractiveArtifactStoreWriter; readonly sessionTodo: InteractiveSessionTodoWriter; + readonly contextOffload?: Pick< + InteractiveContextOffloadWriter, + 'copyReferences' | 'retireSession' + >; readonly manager: SessionManager; readonly admission: SessionAdmissionGate; readonly continuity: SessionContinuityCoordinator; @@ -503,6 +509,28 @@ export class HostSessionRevisionCoordinator { ) .map(({ descriptor, serializedResult }) => [descriptor.artifactId, serializedResult]), ); + const sourceContextRefIds = collectConversationCopySessionContextRefIds({ + sourceSessionId: input.sourceSessionId, + copiedMessages: slice.messages, + plan, + }); + if (sourceContextRefIds.length > 0 && !this.options.contextOffload) { + throw new Error('Session context copy authority is unavailable'); + } + const contextCopy = + sourceContextRefIds.length === 0 + ? { ok: true as const, copied: [] } + : await this.options.contextOffload!.copyReferences({ + sourceSessionId: input.sourceSessionId, + targetSessionId: input.targetSessionId, + references: sourceContextRefIds.map((sourceRefId) => ({ + sourceRefId, + targetOwner: { kind: 'read_image_snapshot', ownerId: sourceRefId }, + })), + }); + if (!contextCopy.ok) { + throw new Error(`Session context references could not be copied: ${contextCopy.reason}`); + } const artifactCopy = await this.#artifacts.copyConversationArtifacts({ sourceSessionId: input.sourceSessionId, targetSessionId: input.targetSessionId, @@ -528,6 +556,9 @@ export class HostSessionRevisionCoordinator { targetSessionId: input.targetSessionId, artifactIds: artifactCopy.artifactIds, relativePaths: artifactCopy.relativePaths, + contextRefs: new Map( + contextCopy.copied.map(({ sourceRefId, targetRefId }) => [sourceRefId, targetRefId]), + ), linkedChildren: kind === 'side_conversation' ? { @@ -819,6 +850,7 @@ export class HostSessionRevisionCoordinator { { artifacts: this.#artifacts, sessionTodo: this.#sessionTodo, + ...(this.options.contextOffload ? { contextOffload: this.options.contextOffload } : {}), purgeOperationalState: (sessionId) => this.#stores.purgeConversationOperationalState(sessionId), }, diff --git a/packages/runtime-host/src/server/session-sidecar-purge.ts b/packages/runtime-host/src/server/session-sidecar-purge.ts index 54a09e4676..59484fcf12 100644 --- a/packages/runtime-host/src/server/session-sidecar-purge.ts +++ b/packages/runtime-host/src/server/session-sidecar-purge.ts @@ -19,10 +19,12 @@ import type { InteractiveArtifactStoreWriter } from '@maka/storage/artifact-stores'; import type { InteractiveSessionTodoWriter } from '@maka/storage/session-todo-authority'; +import type { InteractiveContextOffloadWriter } from '@maka/storage/context-offload-store'; export interface SessionSidecarPurgeAuthority { readonly artifacts: Pick; readonly sessionTodo: Pick; + readonly contextOffload?: Pick; readonly purgeOperationalState: (sessionId: string) => Promise; } @@ -33,6 +35,7 @@ export async function purgeSessionSidecars( const outcomes = await Promise.allSettled([ authority.artifacts.purgeSessionArtifacts(sessionId), authority.sessionTodo.purgeSessionState(sessionId), + ...(authority.contextOffload ? [authority.contextOffload.retireSession(sessionId)] : []), authority.purgeOperationalState(sessionId), ]); const failures = outcomes.flatMap((outcome) => diff --git a/packages/runtime/src/__tests__/builtin-tools-file-worker.test.ts b/packages/runtime/src/__tests__/builtin-tools-file-worker.test.ts index bc0eaa5886..fe9c9a1c24 100644 --- a/packages/runtime/src/__tests__/builtin-tools-file-worker.test.ts +++ b/packages/runtime/src/__tests__/builtin-tools-file-worker.test.ts @@ -169,6 +169,7 @@ describe('builtin file tools use the sandboxed worker', () => { test('uses one worker read operation for image paths', async () => { const cwd = await temporaryDirectory('maka-file-worker-cwd-'); const calls: FilesystemWorkerExecuteInput[] = []; + let snapshotOwnerId: string | undefined; const tools = buildBuiltinTools({ filesystemWorker: { execute: async (input) => { @@ -176,11 +177,14 @@ describe('builtin file tools use the sandboxed worker', () => { return { kind: 'read_image', base64: 'iVBORw0KGgo=', mimeType: 'image/png' }; }, }, - snapshotImage: async () => ({ - kind: 'session_file', - sessionId: 'session-1', - relativePath: 'artifact-1', - }), + snapshotImage: async (input) => { + snapshotOwnerId = input.ownerId; + return { + kind: 'session_context', + sessionId: 'session-1', + refId: 'context-1', + }; + }, sandboxPlatform: 'darwin', }); @@ -188,6 +192,7 @@ describe('builtin file tools use the sandboxed worker', () => { assert.equal(calls.length, 1); assert.deepEqual(calls[0]?.operation, { kind: 'read', path: 'image.png', offset: 1, limit: 1 }); + assert.equal(snapshotOwnerId, 'tool-Read'); }); test('serializes writes through real and symlinked cwd paths', async () => { diff --git a/packages/runtime/src/__tests__/conversation-copy.test.ts b/packages/runtime/src/__tests__/conversation-copy.test.ts index 36bed08f12..7c1504ad39 100644 --- a/packages/runtime/src/__tests__/conversation-copy.test.ts +++ b/packages/runtime/src/__tests__/conversation-copy.test.ts @@ -44,6 +44,7 @@ import { archivedToolResultContainsConversationOwnedReferences, cloneConversationRuntimeLedger, collectConversationCopyLinkedChildReferences, + collectConversationCopySessionContextRefIds, collectConversationCopySessionFileRefs, createConversationCopySlice, prepareConversationRuntimeLedgerCopy, @@ -721,6 +722,17 @@ test('conversation copy rewrites owned references without changing opaque tool p relativePath: 'session-source/artifact-source-file.txt', }, }, + { + kind: 'image', + name: 'snapshot.png', + mimeType: 'image/png', + bytes: 4, + ref: { + kind: 'session_context', + sessionId: 'session-source', + refId: 'context-source', + }, + }, ], }, { @@ -800,6 +812,7 @@ test('conversation copy rewrites owned references without changing opaque tool p relativePaths: new Map([ ['session-source/artifact-source-file.txt', 'session-target/artifact-target-file.txt'], ]), + contextRefs: new Map([['context-source', 'context-target']]), runIds: new Map([['run-source', 'run-target']]), invocationIds: new Map([['invocation-source', 'invocation-target']]), runtimeEventIds: new Map([['event-source', 'event-target']]), @@ -812,6 +825,32 @@ test('conversation copy rewrites owned references without changing opaque tool p sessionId: 'session-target', relativePath: 'session-target/artifact-target-file.txt', }); + assert.deepEqual(rewritten[0]?.type === 'user' ? rewritten[0].attachments?.[1]?.ref : undefined, { + kind: 'session_context', + sessionId: 'session-target', + refId: 'context-target', + }); + assert.deepEqual( + collectConversationCopySessionContextRefIds({ + sourceSessionId: 'session-source', + copiedMessages: messages, + plan: { + sourceSessionId: 'session-source', + copyTurnIds: ['turn-1'], + inlineRuntimeEvents: [], + runs: [], + }, + }), + ['context-source'], + ); + assert.throws( + () => + rewriteConversationCopyMessage(messages[0]!, { + ...references, + contextRefs: new Map(), + }), + /missing Session context context-source/, + ); const userMessage = messages[0]; assert.equal(userMessage?.type, 'user'); if (userMessage?.type !== 'user') return; @@ -1004,44 +1043,6 @@ 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/__tests__/filesystem-authority.test.ts b/packages/runtime/src/__tests__/filesystem-authority.test.ts index b70ff74418..c755002fa4 100644 --- a/packages/runtime/src/__tests__/filesystem-authority.test.ts +++ b/packages/runtime/src/__tests__/filesystem-authority.test.ts @@ -435,7 +435,7 @@ describe('file tools follow the execution boundary', () => { }, snapshotImage: async (input) => { snapshots.push(input.bytes); - return { kind: 'session_file', sessionId: input.sessionId, relativePath: 'artifact-1' }; + return { kind: 'session_context', sessionId: input.sessionId, refId: 'context-1' }; }, }); diff --git a/packages/runtime/src/builtin-tools.ts b/packages/runtime/src/builtin-tools.ts index dbb1d7dc65..37a2f43011 100644 --- a/packages/runtime/src/builtin-tools.ts +++ b/packages/runtime/src/builtin-tools.ts @@ -184,11 +184,10 @@ export interface BuildBuiltinToolsOptions { sandboxPlatform?: SandboxPlatform; snapshotImage?: (input: { sessionId: string; - turnId: string; - name: string; + ownerId: string; bytes: Uint8Array; mimeType: string; - }) => Promise>; + }) => Promise>; } export function buildBuiltinTools(options: BuildBuiltinToolsOptions = {}): MakaTool[] { @@ -400,8 +399,7 @@ export function buildBuiltinTools(options: BuildBuiltinToolsOptions = {}): MakaT throw new Error('Read image snapshots are not available in this toolset.'); const ref = await options.snapshotImage({ sessionId, - turnId: ctx.turnId, - name: basename(path), + ownerId: ctx.toolCallId, bytes: result.bytes, mimeType: result.mimeType, }); diff --git a/packages/runtime/src/conversation-copy.ts b/packages/runtime/src/conversation-copy.ts index f013c89302..fe42b81b12 100644 --- a/packages/runtime/src/conversation-copy.ts +++ b/packages/runtime/src/conversation-copy.ts @@ -25,7 +25,7 @@ import type { } from '@maka/core/agent-run'; import type { RuntimeEvent } from '@maka/core/runtime-event'; import type { RuntimeEventStore } from '@maka/core/runtime-event-store'; -import type { StorageRef, ToolResultContent } from '@maka/core/events'; +import { isStorageRef, type StorageRef, type ToolResultContent } from '@maka/core/events'; import { parseAttachmentResourceRef } from '@maka/core/attachments'; import { markPersisted } from '@maka/core/persisted-value'; import type { StoredMessage } from '@maka/core/session'; @@ -104,6 +104,7 @@ export type ConversationCopyArtifactReferenceMap = readonly mode: 'exact'; readonly artifactIds: ReadonlyMap; readonly relativePaths: ReadonlyMap; + readonly contextRefs?: ReadonlyMap; readonly linkedChildren: | { readonly mode: 'reject' } | { @@ -159,6 +160,35 @@ export interface ConversationRuntimeLedgerCopyPlan { }[]; } +/** Finds durable Session context references that the exact copy will rewrite. */ +export function collectConversationCopySessionContextRefIds(input: { + readonly sourceSessionId: string; + readonly copiedMessages: readonly StoredMessage[]; + readonly plan: ConversationRuntimeLedgerCopyPlan; +}): readonly string[] { + const refIds = new Set(); + const seen = new WeakSet(); + const visit = (value: unknown): void => { + if (isStorageRef(value)) { + if (value.kind === 'session_context' && value.sessionId === input.sourceSessionId) { + refIds.add(value.refId); + } + return; + } + if (typeof value !== 'object' || value === null || seen.has(value)) return; + seen.add(value); + if (Array.isArray(value)) { + for (const item of value) visit(item); + return; + } + for (const item of Object.values(value)) visit(item); + }; + visit(input.copiedMessages); + visit(input.plan.inlineRuntimeEvents); + for (const run of input.plan.runs) visit(run.runtimeEvents); + return [...refIds].sort(); +} + export interface CloneConversationRuntimeLedgerResult { readonly copiedMessages: readonly StoredMessage[]; readonly runIdMap: readonly { @@ -1779,12 +1809,22 @@ 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.kind !== 'session_context') || + ref.sessionId !== references.sourceSessionId + ) { + return ref; } - if (ref.kind !== 'session_file' || ref.sessionId !== references.sourceSessionId) return ref; if (references.mode === 'preserve_external') return ref; + if (ref.kind === 'session_context') { + const refId = references.contextRefs?.get(ref.refId); + if (!refId) throw new Error(`Conversation copy is missing Session context ${ref.refId}`); + return { + ...ref, + sessionId: references.targetSessionId, + refId, + }; + } const artifactId = references.artifactIds.get(ref.relativePath); if (artifactId) { return { From 13aec0ae04fa47204ed57c03331f64a37e90f4f0 Mon Sep 17 00:00:00 2001 From: likun Date: Tue, 1 Sep 2026 22:34:58 +0800 Subject: [PATCH 2/5] fix(runtime-host): close Read context lifecycle gaps --- ...-read-image-context-writer-review-fixes.md | 109 ++++++++++++++++++ .../__tests__/execution-composition.test.ts | 53 ++++++++- .../src/server/execution-composition.ts | 17 ++- .../server/session-revision-coordinator.ts | 5 +- .../builtin-tools-file-worker.test.ts | 68 ++++++++++- .../src/__tests__/conversation-copy.test.ts | 55 +++++++-- .../__tests__/filesystem-authority.test.ts | 1 + .../tool-runtime-durable-boundary.test.ts | 48 ++++++-- packages/runtime/src/builtin-tools.ts | 37 +++++- packages/runtime/src/conversation-copy.ts | 58 ++++++---- packages/runtime/src/tool-runtime.ts | 18 +++ 11 files changed, 423 insertions(+), 46 deletions(-) create mode 100644 docs/changelogs/2026-09-01-read-image-context-writer-review-fixes.md diff --git a/docs/changelogs/2026-09-01-read-image-context-writer-review-fixes.md b/docs/changelogs/2026-09-01-read-image-context-writer-review-fixes.md new file mode 100644 index 0000000000..51c217ba79 --- /dev/null +++ b/docs/changelogs/2026-09-01-read-image-context-writer-review-fixes.md @@ -0,0 +1,109 @@ +# Read image context writer review fixes + +## 1. Current state and problem + +PR #4184 moves Read image snapshots from Session artifacts to the durable Session context Store. +Review found four reachable ownership and lifecycle defects in the initial cut: + +- B1: snapshot identity used the provider-scoped tool call id rather than Runtime's durable tool + operation id. +- B2: conversation copy collected context refs from the complete source event set and opaque JSON, + not only the selected typed projection it rewrites. +- F1: a known T2 result-commit failure left the already-created snapshot reference live. +- F2: recovery could discard the durable preparing-copy metadata while context retirement was + unavailable. + +## 2. Required behavior + +- B3: Read image snapshots are owned by the Runtime-issued durable operation id and fail closed + when that identity is absent. +- B4: branch, revision, and Side Conversation copies collect only typed attachment/image refs in + copied Messages, selected AgentRun RuntimeEvents, and inspected archived results. +- B5: a known T2 failure attempts to release the Read snapshot without replacing the authoritative + persistence error if compensation also fails. +- B6: an unavailable context Store keeps preparing-copy cleanup retryable by failing retirement + before stable Session metadata deletion. + +## 3. Architecture and alternatives + +- D1: use `MakaToolContext.operationId`, already derived from invocation id plus provider tool-call + id, as the context Store owner id. Re-deriving or accepting the provider id would duplicate or + weaken Runtime authority. +- D2: mirror the existing typed Session-file collector instead of structurally walking arbitrary + payloads. This keeps collection and rewrite sites identical. +- D3: add one optional ToolRuntime T2 compensation hook and bind it only for Read image snapshots. + Compensation receives the normalized durable result so it releases exactly the ref T2 attempted + to publish. +- D4: inject a failing context copy/retirement authority when the optional Store cannot open. The + preparing Session header remains the durable retry anchor. +- R1: hard-crash or outcome-unknown windows cannot be perfectly compensated; whole-Session + retirement remains the durable backstop. +- R2: ordinary branch/revision copies continue rejecting archived source-owned image refs; exact + archived-result migration remains outside this PR. + +## 4. Implementation design + +- Runtime `Read` refuses image snapshotting without `operationId` and passes it to the snapshot + Store as `ownerId`. +- `MakaTool.compensateDurableOutcomeCommitFailure` runs only after `commitToolOutcome` rejects. + The Read binding recognizes only a same-Session `session_context` image result and calls + `releaseReference`. +- `collectConversationCopySessionContextRefIds` accepts copied Messages, selected RuntimeEvents, + and archived serialized results. It decodes only canonical `ToolResultContent` image sites. +- Runtime Host passes the same available-or-failing context authority to copy recovery and Session + retirement, so sidecar purge must succeed before metadata discard. +- Compatibility epoch 88 marks the newly emitted `session_context` tool result shape after rebasing + onto epoch 87. + +## 5. Verification plan + +- E2E Required: no. The stable cross-package boundary is the repository's two-client UDS Session + revision integration test; no external service stack is involved. +- Build `@maka/core`, `@maka/storage`, `@maka/runtime`, and `@maka/runtime-host` in dependency order. +- Run focused Runtime tests for Read wiring, filesystem authority, conversation copy, and the + durable T1/T2 boundary. +- Run focused Runtime Host tests for production composition recovery, protocol epoch, and Session + retirement. +- Run the Session revision two-client UDS integration test because copy projection and restart + recovery cross package boundaries. +- Run `git diff --check`; the repository has no `make lint-fix` target. + +## 6. Serial checklist + +- [x] Rebase the PR branch onto current `origin/main` and reconcile SessionTodo/protocol changes. +- [x] Replace provider snapshot identity with Runtime durable operation identity. +- [x] Restrict context-ref collection to selected typed rewrite sites. +- [x] Add best-effort T2 snapshot compensation. +- [x] Retain preparing-copy cleanup metadata while context retirement is unavailable. +- [x] Complete final focused and integration verification. +- [x] Complete the final six-pass diff review with no remaining actionable finding. +- [ ] Scan secrets, publish, and monitor the updated PR. + +## 7. Outcome and evidence + +| ID | Implementation | Evidence | Status | +| --- | --- | --- | --- | +| B3 / D1 | Read snapshot owner is `MakaToolContext.operationId`; missing identity fails closed. | Builtin Read tests cover the exact owner and missing-identity refusal. | DONE | +| B4 / D2 | Context refs are collected only from typed copied Messages, selected run events, and inspected archives. | Conversation-copy tests prove typed images are selected while opaque lookalikes are ignored. | DONE | +| B5 / D3 / F1 | T2 rejection invokes Read snapshot release best-effort without masking T2. | Durable-boundary and Read compensation tests cover success and compensation failure. | DONE | +| B6 / D4 / F2 | Unavailable context authority refuses preparing-copy retirement before metadata discard. | Production composition recovery test reopens storage and observes the preparing header intact. | DONE | +| R1 | Hard-crash compensation remains outside the known-failure hook. | Session retirement continues to retire all context refs durably. | DONE | +| R2 | Ordinary archived owned refs remain rejected rather than copied unsafely. | Archive preflight recognizes source-owned `session_context` images. | DONE | + +Final commands, all run from the repository root: + +- `npm --workspace @maka/core run build && npm --workspace @maka/storage run build && npm --workspace @maka/runtime run build && npm --workspace @maka/runtime-host run build` — pass. +- `node --test packages/runtime/dist/__tests__/builtin-tools-file-worker.test.js packages/runtime/dist/__tests__/filesystem-authority.test.js packages/runtime/dist/__tests__/conversation-copy.test.js packages/runtime/dist/__tests__/tool-runtime-durable-boundary.test.js` — 99 tests pass. +- `node --test packages/runtime-host/dist/__tests__/execution-composition.test.js packages/runtime-host/dist/__tests__/protocol.test.js packages/runtime-host/dist/__tests__/session-retirement-coordinator.test.js` — 105 tests pass. +- `node --test packages/runtime-host/dist/__tests__/session-revision-two-client-uds.test.js` — 1 integration test passes. +- `git diff --check origin/main` — pass. +- Lint fix was not run because the repository has no `Makefile` or `make lint-fix` target. + +The net diff was reviewed in behavior/scope, architecture, tests, security/compatibility, +operations, and documentation passes. No migration, deployment-order change, credential, or +external-service dependency is introduced. + +## 8. Remaining work + +- Scan secrets, push the rebased branch, resolve the six addressed GitHub threads, and monitor the + current head until CI and review state are clean. diff --git a/packages/runtime-host/src/__tests__/execution-composition.test.ts b/packages/runtime-host/src/__tests__/execution-composition.test.ts index a427b76a36..655752f73a 100644 --- a/packages/runtime-host/src/__tests__/execution-composition.test.ts +++ b/packages/runtime-host/src/__tests__/execution-composition.test.ts @@ -105,6 +105,32 @@ test('production composition owns the long-term memory database lifecycle', asyn test('production composition reaches Ready when the optional context Store cannot open', async () => { await withCompositionRoot(async ({ root, owner }) => { + const requestFingerprint = `sha256:${'a'.repeat(64)}` as const; + const preparingSessionId = 'preparing-context-copy'; + const stores = await openInteractiveExecutionStoresForWrite(owner.lease); + await stores.sessionStore.createStableSession({ + sessionId: preparingSessionId, + requestFingerprint, + input: { + cwd: root, + llmConnectionId: FAKE_CONNECTION_ID, + llmConnectionSlug: 'fake', + model: 'fake-model', + permissionMode: 'ask', + name: 'Preparing context copy', + labels: [], + parentSessionId: 'source-session', + branchOfTurnId: 'source-turn', + conversationCopy: { + kind: 'branch', + sourceSessionId: 'source-session', + sourceTurnId: 'source-turn', + requestFingerprint, + state: 'preparing', + }, + }, + }); + await stores.sessionStore.close?.(); await mkdir(join(root, CONTEXT_OFFLOAD_DATABASE_NAME)); const originalConsoleError = console.error; const diagnostics: string[] = []; @@ -117,9 +143,34 @@ test('production composition reaches Ready when the optional context Store canno diagnostics.some((message) => message.includes('optional context-offload Store')), true, ); + await assert.rejects( + composition.recover(), + (error) => + error instanceof AggregateError && + error.errors.some((failure) => + String(failure).includes( + 'Context-offload Store is unavailable during Session retirement', + ), + ), + ); } finally { console.error = originalConsoleError; - await composition?.close(); + if (composition) { + await assert.rejects( + composition.close(), + /Unable to close Runtime Host execution composition/, + ); + } + } + const reopened = await openInteractiveExecutionStoresForWrite(owner.lease); + try { + assert.equal( + (await reopened.sessionStore.readHeaderSnapshot(preparingSessionId)).conversationCopy + ?.state, + 'preparing', + ); + } finally { + await reopened.sessionStore.close?.(); } }); }); diff --git a/packages/runtime-host/src/server/execution-composition.ts b/packages/runtime-host/src/server/execution-composition.ts index 2ae0681ad1..72b03d558e 100644 --- a/packages/runtime-host/src/server/execution-composition.ts +++ b/packages/runtime-host/src/server/execution-composition.ts @@ -290,10 +290,15 @@ export async function createExecutionRuntimeHostComposition( const openedContextOffloadReader = openedContextOffloadStore ? createInteractiveContextOffloadReader(openedContextOffloadStore) : undefined; - const contextOffloadRetirement = openedContextOffloadStore + const contextOffloadAuthority = openedContextOffloadStore ? openedContextOffloadStore : storage.contextOffloadUnavailable ? { + copyReferences: async (): Promise => { + throw new Error('Context-offload Store is unavailable during Session copy', { + cause: storage.contextOffloadUnavailable?.cause, + }); + }, retireSession: async (_sessionId: string): Promise => { throw new Error('Context-offload Store is unavailable during Session retirement', { cause: storage.contextOffloadUnavailable?.cause, @@ -423,6 +428,12 @@ export async function createExecutionRuntimeHostComposition( bytes: input.bytes, mimeType: input.mimeType, }), + releaseImageSnapshot: async (input: { + readonly sessionId: string; + readonly refId: string; + }) => { + await openedContextOffloadStore.releaseReference(input); + }, } : {}), ...(sandboxManager ? { sandboxManager } : {}), @@ -1569,7 +1580,7 @@ export async function createExecutionRuntimeHostComposition( stores, artifacts: openedArtifactStore, sessionTodo: sessionTodoStore, - ...(contextOffloadRetirement ? { contextOffload: contextOffloadRetirement } : {}), + ...(contextOffloadAuthority ? { contextOffload: contextOffloadAuthority } : {}), manager, admission: sessionAdmission, continuity: continuityCoordinator, @@ -1594,7 +1605,7 @@ export async function createExecutionRuntimeHostComposition( continuity: continuityCoordinator, artifacts: openedArtifactStore, sessionTodo: sessionTodoStore, - ...(contextOffloadRetirement ? { contextOffload: contextOffloadRetirement } : {}), + ...(contextOffloadAuthority ? { contextOffload: contextOffloadAuthority } : {}), purgeOperationalState: async (sessionId) => { await stores.purgeConversationOperationalState(sessionId); await openedPlanStore.purgeSessionState(sessionId); diff --git a/packages/runtime-host/src/server/session-revision-coordinator.ts b/packages/runtime-host/src/server/session-revision-coordinator.ts index dd051dcfc8..68de60efac 100644 --- a/packages/runtime-host/src/server/session-revision-coordinator.ts +++ b/packages/runtime-host/src/server/session-revision-coordinator.ts @@ -511,8 +511,9 @@ export class HostSessionRevisionCoordinator { ); const sourceContextRefIds = collectConversationCopySessionContextRefIds({ sourceSessionId: input.sourceSessionId, - copiedMessages: slice.messages, - plan, + messages: slice.messages, + runtimeEvents: plan.runs.flatMap(({ runtimeEvents }) => runtimeEvents), + archivedResults: archivePreflight.serializedResults, }); if (sourceContextRefIds.length > 0 && !this.options.contextOffload) { throw new Error('Session context copy authority is unavailable'); diff --git a/packages/runtime/src/__tests__/builtin-tools-file-worker.test.ts b/packages/runtime/src/__tests__/builtin-tools-file-worker.test.ts index fe9c9a1c24..26050bea7c 100644 --- a/packages/runtime/src/__tests__/builtin-tools-file-worker.test.ts +++ b/packages/runtime/src/__tests__/builtin-tools-file-worker.test.ts @@ -192,7 +192,72 @@ describe('builtin file tools use the sandboxed worker', () => { assert.equal(calls.length, 1); assert.deepEqual(calls[0]?.operation, { kind: 'read', path: 'image.png', offset: 1, limit: 1 }); - assert.equal(snapshotOwnerId, 'tool-Read'); + assert.equal(snapshotOwnerId, 'toolop-Read'); + }); + + test('releases a Read image snapshot when durable result commit fails', async () => { + const released: Array<{ sessionId: string; refId: string }> = []; + const read = buildBuiltinTools({ + releaseImageSnapshot: async (input) => { + released.push(input); + }, + }).find((candidate) => candidate.name === 'Read'); + if (!read?.compensateDurableOutcomeCommitFailure) { + throw new Error('Read image compensation missing'); + } + + await read.compensateDurableOutcomeCommitFailure({ + result: { + kind: 'image', + mimeType: 'image/png', + ref: { kind: 'session_context', sessionId: 'session-1', refId: 'context-1' }, + }, + isError: false, + sessionId: 'session-1', + operationId: 'toolop-Read', + }); + + assert.deepEqual(released, [{ sessionId: 'session-1', refId: 'context-1' }]); + }); + + test('refuses to snapshot a Read image without a durable operation identity', async () => { + const tools = buildBuiltinTools({ + filesystemWorker: { + execute: async () => ({ + kind: 'read_image', + base64: 'iVBORw0KGgo=', + mimeType: 'image/png', + }), + }, + snapshotImage: async () => { + assert.fail('snapshot must not run without a durable operation identity'); + }, + sandboxPlatform: 'darwin', + }); + const read = tools.find((candidate) => candidate.name === 'Read'); + if (!read) throw new Error('Read tool missing'); + + await assert.rejects( + Promise.resolve( + read.impl( + { path: 'image.png' }, + { + sessionId: 'session-1', + turnId: 'turn-1', + toolCallId: 'provider-call-reused', + cwd: await temporaryDirectory('maka-file-worker-cwd-'), + permissionMode: 'ask', + executionBoundary: createManagedExecutionBoundary( + createWorkspaceWritePermissionProfile(), + 0, + ), + abortSignal: new AbortController().signal, + emitOutput: () => {}, + }, + ), + ), + /require a durable tool operation identity/, + ); }); test('serializes writes through real and symlinked cwd paths', async () => { @@ -428,6 +493,7 @@ async function runTool( sessionId: 'session-1', turnId: 'turn-1', toolCallId: `tool-${name}`, + operationId: `toolop-${name}`, cwd, permissionMode: 'ask', executionBoundary: createManagedExecutionBoundary(createWorkspaceWritePermissionProfile(), 0), diff --git a/packages/runtime/src/__tests__/conversation-copy.test.ts b/packages/runtime/src/__tests__/conversation-copy.test.ts index 7c1504ad39..5bf195594d 100644 --- a/packages/runtime/src/__tests__/conversation-copy.test.ts +++ b/packages/runtime/src/__tests__/conversation-copy.test.ts @@ -745,6 +745,11 @@ test('conversation copy rewrites owned references without changing opaque tool p sessionId: 'session-source', runId: 'run-source', artifactId: 'artifact-source', + opaqueStorageRefShape: { + kind: 'session_context', + sessionId: 'session-source', + refId: 'context-opaque', + }, }, providerOptions: { sourceInvocationId: 'invocation-source', @@ -833,15 +838,49 @@ test('conversation copy rewrites owned references without changing opaque tool p assert.deepEqual( collectConversationCopySessionContextRefIds({ sourceSessionId: 'session-source', - copiedMessages: messages, - plan: { - sourceSessionId: 'session-source', - copyTurnIds: ['turn-1'], - inlineRuntimeEvents: [], - runs: [], - }, + messages, + runtimeEvents: [ + runtimeEvent({ + id: 'selected-image-result', + role: 'tool', + author: 'tool', + content: { + kind: 'function_response', + id: 'tool-image', + name: 'Read', + result: { + kind: 'image', + mimeType: 'image/png', + ref: { + kind: 'session_context', + sessionId: 'session-source', + refId: 'context-selected-event', + }, + }, + }, + }), + runtimeEvent({ + id: 'opaque-json-result', + role: 'tool', + author: 'tool', + content: { + kind: 'function_response', + id: 'tool-opaque', + name: 'opaque', + result: { + kind: 'json', + value: { + kind: 'session_context', + sessionId: 'session-source', + refId: 'context-opaque-event', + }, + }, + }, + }), + ], + archivedResults: [], }), - ['context-source'], + ['context-selected-event', 'context-source'], ); assert.throws( () => diff --git a/packages/runtime/src/__tests__/filesystem-authority.test.ts b/packages/runtime/src/__tests__/filesystem-authority.test.ts index c755002fa4..ed20bbbe63 100644 --- a/packages/runtime/src/__tests__/filesystem-authority.test.ts +++ b/packages/runtime/src/__tests__/filesystem-authority.test.ts @@ -78,6 +78,7 @@ function runTool( turnId: 'turn-1', cwd, toolCallId: 'tool-1', + operationId: 'toolop-1', abortSignal: new AbortController().signal, emitOutput: () => {}, ...(executionBoundary ? { executionBoundary } : {}), diff --git a/packages/runtime/src/__tests__/tool-runtime-durable-boundary.test.ts b/packages/runtime/src/__tests__/tool-runtime-durable-boundary.test.ts index 71b470df11..0c87cb82a4 100644 --- a/packages/runtime/src/__tests__/tool-runtime-durable-boundary.test.ts +++ b/packages/runtime/src/__tests__/tool-runtime-durable-boundary.test.ts @@ -1593,6 +1593,7 @@ describe('ToolRuntime durable boundary', () => { it('does not publish an implementation result when T2 fails', async () => { let implementationCalls = 0; + const compensations: unknown[] = []; const harness = makeHarness({ commitToolPrepared: async () => ({ created: true, runtimeEventSeq: 1 }), commitToolOutcome: async () => { @@ -1600,15 +1601,15 @@ describe('ToolRuntime durable boundary', () => { }, }); - await assert.rejects( - harness.execute( - tool(() => { - implementationCalls += 1; - return { ok: true }; - }), - ), - /T2 unavailable/, - ); + const target = tool(() => { + implementationCalls += 1; + return { ok: true }; + }); + target.compensateDurableOutcomeCommitFailure = async (input) => { + compensations.push(input); + }; + + await assert.rejects(harness.execute(target), /T2 unavailable/); assert.equal(implementationCalls, 1); assert.equal( @@ -1619,6 +1620,35 @@ describe('ToolRuntime durable boundary', () => { harness.messages.some((message) => message.type === 'tool_result'), false, ); + assert.equal(compensations.length, 1); + const compensation = compensations[0] as { + result: unknown; + isError: boolean; + sessionId: string; + operationId: string; + }; + assert.deepEqual({ ...compensation, operationId: '' }, { + result: { kind: 'json', value: { ok: true } }, + isError: false, + sessionId: 'session-1', + operationId: '', + }); + assert.match(compensation.operationId, /^toolop_/); + }); + + it('keeps the T2 persistence error authoritative when compensation also fails', async () => { + const harness = makeHarness({ + commitToolPrepared: async () => ({ created: true, runtimeEventSeq: 1 }), + commitToolOutcome: async () => { + throw new Error('T2 unavailable'); + }, + }); + const target = tool(() => ({ ok: true })); + target.compensateDurableOutcomeCommitFailure = async () => { + throw new Error('compensation unavailable'); + }; + + await assert.rejects(harness.execute(target), /T2 unavailable/); }); it('commits a normalized error outcome before returning a thrown tool failure to the model', async () => { diff --git a/packages/runtime/src/builtin-tools.ts b/packages/runtime/src/builtin-tools.ts index 37a2f43011..18ddde6aff 100644 --- a/packages/runtime/src/builtin-tools.ts +++ b/packages/runtime/src/builtin-tools.ts @@ -39,7 +39,7 @@ import { basename, dirname, isAbsolute } from 'node:path'; import { compilePermissionProfile } from '@maka/core/permission-profile-compiler'; import { parseAttachmentResourceRef } from '@maka/core/attachments'; import { type SandboxBoundaryExpansion } from '@maka/core/sandbox-boundary'; -import { type StorageRef, type ToolResultContent } from '@maka/core/events'; +import { isStorageRef, type StorageRef, type ToolResultContent } from '@maka/core/events'; import { type PermissionProfile } from '@maka/core/permission-profile'; import { bashToolResultToModelOutput } from './bash-model-output.js'; import { fileWriteToolResultToModelOutput } from './file-tool-model-output.js'; @@ -188,6 +188,7 @@ export interface BuildBuiltinToolsOptions { bytes: Uint8Array; mimeType: string; }) => Promise>; + releaseImageSnapshot?: (input: { sessionId: string; refId: string }) => Promise; } export function buildBuiltinTools(options: BuildBuiltinToolsOptions = {}): MakaTool[] { @@ -355,6 +356,35 @@ export function buildBuiltinTools(options: BuildBuiltinToolsOptions = {}): MakaT description: readDescription, parameters: readParameters, executionFacts, + ...(options.releaseImageSnapshot + ? { + compensateDurableOutcomeCommitFailure: async (input: { + readonly result: unknown; + readonly sessionId: string; + }) => { + const result = input.result; + if ( + !result || + typeof result !== 'object' || + (result as { kind?: unknown }).kind !== 'image' + ) { + return; + } + const ref = (result as { ref?: unknown }).ref; + if ( + !isStorageRef(ref) || + ref.kind !== 'session_context' || + ref.sessionId !== input.sessionId + ) { + return; + } + await options.releaseImageSnapshot!({ + sessionId: ref.sessionId, + refId: ref.refId, + }); + }, + } + : {}), impl: async (input, ctx) => { const { cwd, sessionId, abortSignal } = ctx; if ('ref' in input) { @@ -397,9 +427,12 @@ export function buildBuiltinTools(options: BuildBuiltinToolsOptions = {}): MakaT if (result.kind === 'read_image') { if (!options.snapshotImage) throw new Error('Read image snapshots are not available in this toolset.'); + if (!ctx.operationId) { + throw new Error('Read image snapshots require a durable tool operation identity.'); + } const ref = await options.snapshotImage({ sessionId, - ownerId: ctx.toolCallId, + ownerId: ctx.operationId, bytes: result.bytes, mimeType: result.mimeType, }); diff --git a/packages/runtime/src/conversation-copy.ts b/packages/runtime/src/conversation-copy.ts index fe42b81b12..1448fcfd96 100644 --- a/packages/runtime/src/conversation-copy.ts +++ b/packages/runtime/src/conversation-copy.ts @@ -25,7 +25,7 @@ import type { } from '@maka/core/agent-run'; import type { RuntimeEvent } from '@maka/core/runtime-event'; import type { RuntimeEventStore } from '@maka/core/runtime-event-store'; -import { isStorageRef, type StorageRef, type ToolResultContent } from '@maka/core/events'; +import { type StorageRef, type ToolResultContent } from '@maka/core/events'; import { parseAttachmentResourceRef } from '@maka/core/attachments'; import { markPersisted } from '@maka/core/persisted-value'; import type { StoredMessage } from '@maka/core/session'; @@ -163,29 +163,44 @@ export interface ConversationRuntimeLedgerCopyPlan { /** Finds durable Session context references that the exact copy will rewrite. */ export function collectConversationCopySessionContextRefIds(input: { readonly sourceSessionId: string; - readonly copiedMessages: readonly StoredMessage[]; - readonly plan: ConversationRuntimeLedgerCopyPlan; + readonly messages: readonly StoredMessage[]; + readonly runtimeEvents: readonly RuntimeEvent[]; + readonly archivedResults: readonly string[]; }): readonly string[] { const refIds = new Set(); - const seen = new WeakSet(); - const visit = (value: unknown): void => { - if (isStorageRef(value)) { - if (value.kind === 'session_context' && value.sessionId === input.sourceSessionId) { - refIds.add(value.refId); - } - return; + const addRef = (ref: StorageRef): void => { + if (ref.kind === 'session_context' && ref.sessionId === input.sourceSessionId) { + refIds.add(ref.refId); } - if (typeof value !== 'object' || value === null || seen.has(value)) return; - seen.add(value); - if (Array.isArray(value)) { - for (const item of value) visit(item); - return; + }; + const addContent = (content: ToolResultContent): void => { + if (content.kind === 'image') addRef(content.ref); + }; + const addSerialized = (value: unknown): void => { + if (isArchivedToolResultPlaceholder(value)) return; + try { + addContent(decodePersistedToolResultContent(markPersisted(value))); + } catch { + // Opaque tool results carry no typed Session context reference. } - for (const item of Object.values(value)) visit(item); }; - visit(input.copiedMessages); - visit(input.plan.inlineRuntimeEvents); - for (const run of input.plan.runs) visit(run.runtimeEvents); + for (const message of input.messages) { + if (message.type === 'user' && message.attachments) { + for (const attachment of message.attachments) addRef(attachment.ref); + } else if (message.type === 'tool_result') { + addContent(message.content); + } + } + for (const event of input.runtimeEvents) { + if (event.content?.kind === 'text' && event.content.attachments) { + for (const attachment of event.content.attachments) addRef(attachment.ref); + } else if (event.content?.kind === 'function_response') { + addSerialized(event.content.result); + } + } + for (const serializedResult of input.archivedResults) { + addSerialized(deserializeToolResultArchive(serializedResult)); + } return [...refIds].sort(); } @@ -676,7 +691,10 @@ export function archivedToolResultContainsConversationOwnedReferences( if (content.kind === 'archived_tool_result') return true; if (content.kind === 'image') { - return content.ref.kind === 'session_file' && content.ref.sessionId === sourceSessionId; + return ( + (content.ref.kind === 'session_file' || content.ref.kind === 'session_context') && + content.ref.sessionId === sourceSessionId + ); } if (content.kind === 'subagent') { const [linked] = conversationCopyLinkedChildReferences(content); diff --git a/packages/runtime/src/tool-runtime.ts b/packages/runtime/src/tool-runtime.ts index 5214ffda49..cb81443d34 100644 --- a/packages/runtime/src/tool-runtime.ts +++ b/packages/runtime/src/tool-runtime.ts @@ -223,6 +223,13 @@ export interface MakaTool

{ * settlement instead of detaching, so late side effects cannot outlive `exec`. */ impl: (args: P, ctx: MakaToolContext) => Promise | R; + /** Best-effort compensation after T2 rejects a result that already produced side effects. */ + compensateDurableOutcomeCommitFailure?: (input: { + readonly result: unknown; + readonly isError: boolean; + readonly sessionId: string; + readonly operationId: string; + }) => Promise | void; /** Optional synchronous provider-visible content mapping, used for screenshot image parts. */ toModelOutput?: (options: { toolCallId: string; @@ -2312,6 +2319,17 @@ export class ToolRuntime { committedAt: responseEvent.ts, }); } catch (error) { + try { + await input.tool.compensateDurableOutcomeCommitFailure?.({ + result, + isError, + sessionId: this.input.sessionId, + operationId, + }); + } catch { + // T2 remains authoritative. Compensation is deliberately best-effort + // and must never replace the persistence failure that triggered it. + } throw new RuntimeCommitBoundaryError('T2', error); } committedOutcome = { From d884af9c2581b98a77140ac2dd9d5b02f3153495 Mon Sep 17 00:00:00 2001 From: likun Date: Tue, 1 Sep 2026 23:15:23 +0800 Subject: [PATCH 3/5] docs: add ASF header to review changelog --- ...-read-image-context-writer-review-fixes.md | 19 +++++++++++++++++++ 1 file changed, 19 insertions(+) diff --git a/docs/changelogs/2026-09-01-read-image-context-writer-review-fixes.md b/docs/changelogs/2026-09-01-read-image-context-writer-review-fixes.md index 51c217ba79..e7894b529e 100644 --- a/docs/changelogs/2026-09-01-read-image-context-writer-review-fixes.md +++ b/docs/changelogs/2026-09-01-read-image-context-writer-review-fixes.md @@ -1,3 +1,22 @@ + + # Read image context writer review fixes ## 1. Current state and problem From 4ae3b0f5741ebb8f5fbb1f8cab8c61c7b24e2e9b Mon Sep 17 00:00:00 2001 From: likun Date: Tue, 1 Sep 2026 23:28:20 +0800 Subject: [PATCH 4/5] test(runtime): format compensation assertion --- .../tool-runtime-durable-boundary.test.ts | 15 +++++++++------ 1 file changed, 9 insertions(+), 6 deletions(-) diff --git a/packages/runtime/src/__tests__/tool-runtime-durable-boundary.test.ts b/packages/runtime/src/__tests__/tool-runtime-durable-boundary.test.ts index 0c87cb82a4..2c68e24c32 100644 --- a/packages/runtime/src/__tests__/tool-runtime-durable-boundary.test.ts +++ b/packages/runtime/src/__tests__/tool-runtime-durable-boundary.test.ts @@ -1627,12 +1627,15 @@ describe('ToolRuntime durable boundary', () => { sessionId: string; operationId: string; }; - assert.deepEqual({ ...compensation, operationId: '' }, { - result: { kind: 'json', value: { ok: true } }, - isError: false, - sessionId: 'session-1', - operationId: '', - }); + assert.deepEqual( + { ...compensation, operationId: '' }, + { + result: { kind: 'json', value: { ok: true } }, + isError: false, + sessionId: 'session-1', + operationId: '', + }, + ); assert.match(compensation.operationId, /^toolop_/); }); From 34011d3bbc688cc40bea7766d31626a8e48cb21a Mon Sep 17 00:00:00 2001 From: likun Date: Wed, 2 Sep 2026 00:08:46 +0800 Subject: [PATCH 5/5] fix(runtime-host): finish context lifecycle cleanup --- CHANGELOG.md | 3 + ...-read-image-context-writer-review-fixes.md | 128 ------------------ .../__tests__/execution-composition.test.ts | 34 ++--- .../src/__tests__/protocol.test.ts | 2 +- .../session-retirement-coordinator.test.ts | 33 +++++ .../src/server/execution-composition.ts | 6 + .../server/session-retirement-coordinator.ts | 5 +- .../server/session-revision-coordinator.ts | 16 ++- .../src/server/session-sidecar-purge.ts | 26 +++- packages/runtime/src/conversation-copy.ts | 72 ++++------ 10 files changed, 127 insertions(+), 198 deletions(-) delete mode 100644 docs/changelogs/2026-09-01-read-image-context-writer-review-fixes.md diff --git a/CHANGELOG.md b/CHANGELOG.md index acffc8fe45..58a00b8c7c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -49,6 +49,9 @@ and SessionEvent-to-RuntimeEvent conversion remains a pure mapper. - Retired the Task Ledger domain: SessionTodo is now the sole authority for in-session work items, and the operational-state schema drops the `workflow_task_ledger_events` table on first open. **Unfinished Tasks are not migrated and are permanently deleted.** This affects workspaces last opened by `v0.1.0` through `v0.1.11`, `cli-v0.1.0-beta.1`, `v0.2.0-incubating-rc1`, or a `v0.2.0-dev` build; those releases wrote Tasks to a table that no shipped build ever bridged into SessionTodo. Before opening such a workspace with this build, finish or export the Tasks you still need, or copy the workspace's `runtime.sqlite` aside — the migration removes the only live copy, so afterwards recovery requires a backup made in advance. - Unified context management under one Runtime-owned policy. `MAKA_CONTEXT_*` environment overrides no longer tune or disable compaction and Tool Result pruning; model-visible archive placeholders are read on demand through bounded `ArchiveRead` calls instead of eager hydration. Previously supported overrides are ignored on upgrade: if Tool Result pruning was set to `off`, pruning is re-enabled, and there is currently no supported replacement opt-out. +- Moved Read image snapshots into the durable context-offload store with Runtime-owned + lifecycle identity, exact branch and revision copying, recovery-safe cleanup, and bounded + physical garbage collection after Session retirement. ## 0.1.11 - 2026-08-18 diff --git a/docs/changelogs/2026-09-01-read-image-context-writer-review-fixes.md b/docs/changelogs/2026-09-01-read-image-context-writer-review-fixes.md deleted file mode 100644 index e7894b529e..0000000000 --- a/docs/changelogs/2026-09-01-read-image-context-writer-review-fixes.md +++ /dev/null @@ -1,128 +0,0 @@ - - -# Read image context writer review fixes - -## 1. Current state and problem - -PR #4184 moves Read image snapshots from Session artifacts to the durable Session context Store. -Review found four reachable ownership and lifecycle defects in the initial cut: - -- B1: snapshot identity used the provider-scoped tool call id rather than Runtime's durable tool - operation id. -- B2: conversation copy collected context refs from the complete source event set and opaque JSON, - not only the selected typed projection it rewrites. -- F1: a known T2 result-commit failure left the already-created snapshot reference live. -- F2: recovery could discard the durable preparing-copy metadata while context retirement was - unavailable. - -## 2. Required behavior - -- B3: Read image snapshots are owned by the Runtime-issued durable operation id and fail closed - when that identity is absent. -- B4: branch, revision, and Side Conversation copies collect only typed attachment/image refs in - copied Messages, selected AgentRun RuntimeEvents, and inspected archived results. -- B5: a known T2 failure attempts to release the Read snapshot without replacing the authoritative - persistence error if compensation also fails. -- B6: an unavailable context Store keeps preparing-copy cleanup retryable by failing retirement - before stable Session metadata deletion. - -## 3. Architecture and alternatives - -- D1: use `MakaToolContext.operationId`, already derived from invocation id plus provider tool-call - id, as the context Store owner id. Re-deriving or accepting the provider id would duplicate or - weaken Runtime authority. -- D2: mirror the existing typed Session-file collector instead of structurally walking arbitrary - payloads. This keeps collection and rewrite sites identical. -- D3: add one optional ToolRuntime T2 compensation hook and bind it only for Read image snapshots. - Compensation receives the normalized durable result so it releases exactly the ref T2 attempted - to publish. -- D4: inject a failing context copy/retirement authority when the optional Store cannot open. The - preparing Session header remains the durable retry anchor. -- R1: hard-crash or outcome-unknown windows cannot be perfectly compensated; whole-Session - retirement remains the durable backstop. -- R2: ordinary branch/revision copies continue rejecting archived source-owned image refs; exact - archived-result migration remains outside this PR. - -## 4. Implementation design - -- Runtime `Read` refuses image snapshotting without `operationId` and passes it to the snapshot - Store as `ownerId`. -- `MakaTool.compensateDurableOutcomeCommitFailure` runs only after `commitToolOutcome` rejects. - The Read binding recognizes only a same-Session `session_context` image result and calls - `releaseReference`. -- `collectConversationCopySessionContextRefIds` accepts copied Messages, selected RuntimeEvents, - and archived serialized results. It decodes only canonical `ToolResultContent` image sites. -- Runtime Host passes the same available-or-failing context authority to copy recovery and Session - retirement, so sidecar purge must succeed before metadata discard. -- Compatibility epoch 88 marks the newly emitted `session_context` tool result shape after rebasing - onto epoch 87. - -## 5. Verification plan - -- E2E Required: no. The stable cross-package boundary is the repository's two-client UDS Session - revision integration test; no external service stack is involved. -- Build `@maka/core`, `@maka/storage`, `@maka/runtime`, and `@maka/runtime-host` in dependency order. -- Run focused Runtime tests for Read wiring, filesystem authority, conversation copy, and the - durable T1/T2 boundary. -- Run focused Runtime Host tests for production composition recovery, protocol epoch, and Session - retirement. -- Run the Session revision two-client UDS integration test because copy projection and restart - recovery cross package boundaries. -- Run `git diff --check`; the repository has no `make lint-fix` target. - -## 6. Serial checklist - -- [x] Rebase the PR branch onto current `origin/main` and reconcile SessionTodo/protocol changes. -- [x] Replace provider snapshot identity with Runtime durable operation identity. -- [x] Restrict context-ref collection to selected typed rewrite sites. -- [x] Add best-effort T2 snapshot compensation. -- [x] Retain preparing-copy cleanup metadata while context retirement is unavailable. -- [x] Complete final focused and integration verification. -- [x] Complete the final six-pass diff review with no remaining actionable finding. -- [ ] Scan secrets, publish, and monitor the updated PR. - -## 7. Outcome and evidence - -| ID | Implementation | Evidence | Status | -| --- | --- | --- | --- | -| B3 / D1 | Read snapshot owner is `MakaToolContext.operationId`; missing identity fails closed. | Builtin Read tests cover the exact owner and missing-identity refusal. | DONE | -| B4 / D2 | Context refs are collected only from typed copied Messages, selected run events, and inspected archives. | Conversation-copy tests prove typed images are selected while opaque lookalikes are ignored. | DONE | -| B5 / D3 / F1 | T2 rejection invokes Read snapshot release best-effort without masking T2. | Durable-boundary and Read compensation tests cover success and compensation failure. | DONE | -| B6 / D4 / F2 | Unavailable context authority refuses preparing-copy retirement before metadata discard. | Production composition recovery test reopens storage and observes the preparing header intact. | DONE | -| R1 | Hard-crash compensation remains outside the known-failure hook. | Session retirement continues to retire all context refs durably. | DONE | -| R2 | Ordinary archived owned refs remain rejected rather than copied unsafely. | Archive preflight recognizes source-owned `session_context` images. | DONE | - -Final commands, all run from the repository root: - -- `npm --workspace @maka/core run build && npm --workspace @maka/storage run build && npm --workspace @maka/runtime run build && npm --workspace @maka/runtime-host run build` — pass. -- `node --test packages/runtime/dist/__tests__/builtin-tools-file-worker.test.js packages/runtime/dist/__tests__/filesystem-authority.test.js packages/runtime/dist/__tests__/conversation-copy.test.js packages/runtime/dist/__tests__/tool-runtime-durable-boundary.test.js` — 99 tests pass. -- `node --test packages/runtime-host/dist/__tests__/execution-composition.test.js packages/runtime-host/dist/__tests__/protocol.test.js packages/runtime-host/dist/__tests__/session-retirement-coordinator.test.js` — 105 tests pass. -- `node --test packages/runtime-host/dist/__tests__/session-revision-two-client-uds.test.js` — 1 integration test passes. -- `git diff --check origin/main` — pass. -- Lint fix was not run because the repository has no `Makefile` or `make lint-fix` target. - -The net diff was reviewed in behavior/scope, architecture, tests, security/compatibility, -operations, and documentation passes. No migration, deployment-order change, credential, or -external-service dependency is introduced. - -## 8. Remaining work - -- Scan secrets, push the rebased branch, resolve the six addressed GitHub threads, and monitor the - current head until CI and review state are clean. diff --git a/packages/runtime-host/src/__tests__/execution-composition.test.ts b/packages/runtime-host/src/__tests__/execution-composition.test.ts index 655752f73a..0f25236dc9 100644 --- a/packages/runtime-host/src/__tests__/execution-composition.test.ts +++ b/packages/runtime-host/src/__tests__/execution-composition.test.ts @@ -41,6 +41,7 @@ import { fingerprintAgentGraphRunnableIntent } from '@maka/runtime/stream-graph- import type { AgentGraphRunnableIntent } from '@maka/runtime/stream-graph-readiness'; import { createAgentGraphControlStore } from '@maka/storage/agent-graph-control-store'; import { openInteractiveExecutionStoresForWrite } from '@maka/storage/execution-stores'; +import { createSessionStore } from '@maka/storage/session-store'; import { LONG_TERM_MEMORY_DATABASE_NAME, openInteractiveLongTermMemoryStoreForWrite, @@ -107,8 +108,8 @@ test('production composition reaches Ready when the optional context Store canno await withCompositionRoot(async ({ root, owner }) => { const requestFingerprint = `sha256:${'a'.repeat(64)}` as const; const preparingSessionId = 'preparing-context-copy'; - const stores = await openInteractiveExecutionStoresForWrite(owner.lease); - await stores.sessionStore.createStableSession({ + const sessionStore = createSessionStore(root); + await sessionStore.createStableSession({ sessionId: preparingSessionId, requestFingerprint, input: { @@ -130,7 +131,7 @@ test('production composition reaches Ready when the optional context Store canno }, }, }); - await stores.sessionStore.close?.(); + await sessionStore.close?.(); await mkdir(join(root, CONTEXT_OFFLOAD_DATABASE_NAME)); const originalConsoleError = console.error; const diagnostics: string[] = []; @@ -143,34 +144,27 @@ test('production composition reaches Ready when the optional context Store canno diagnostics.some((message) => message.includes('optional context-offload Store')), true, ); - await assert.rejects( - composition.recover(), - (error) => - error instanceof AggregateError && - error.errors.some((failure) => - String(failure).includes( - 'Context-offload Store is unavailable during Session retirement', - ), - ), + await composition.recover(); + assert.equal( + diagnostics.some((message) => + message.includes('conversation copy cleanup deferred during recovery'), + ), + true, ); } finally { console.error = originalConsoleError; if (composition) { - await assert.rejects( - composition.close(), - /Unable to close Runtime Host execution composition/, - ); + await composition.close(); } } - const reopened = await openInteractiveExecutionStoresForWrite(owner.lease); + const reopened = createSessionStore(root); try { assert.equal( - (await reopened.sessionStore.readHeaderSnapshot(preparingSessionId)).conversationCopy - ?.state, + (await reopened.readHeaderSnapshot(preparingSessionId)).conversationCopy?.state, 'preparing', ); } finally { - await reopened.sessionStore.close?.(); + await reopened.close?.(); } }); }); diff --git a/packages/runtime-host/src/__tests__/protocol.test.ts b/packages/runtime-host/src/__tests__/protocol.test.ts index 95f1e4672d..2ee9d5104d 100644 --- a/packages/runtime-host/src/__tests__/protocol.test.ts +++ b/packages/runtime-host/src/__tests__/protocol.test.ts @@ -141,7 +141,7 @@ describe('Runtime Host bootstrap protocol', () => { }); test('publishes a new compatibility epoch for Read image Session context refs', () => { - assert.ok(RUNTIME_HOST_COMPATIBILITY_EPOCH > 87); + assert.ok(RUNTIME_HOST_COMPATIBILITY_EPOCH > 93); }); test('rejects the legacy connection update result in the current compatibility epoch', () => { 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 ed1b35b5db..c191cf0a0b 100644 --- a/packages/runtime-host/src/__tests__/session-retirement-coordinator.test.ts +++ b/packages/runtime-host/src/__tests__/session-retirement-coordinator.test.ts @@ -41,6 +41,7 @@ import type { ConnectionContext } from '../server/operation-dispatcher.js'; import { SessionAdmissionGate } from '../server/session-admission-gate.js'; import { MemoryExtractionSessionLane } from '../server/memory-extraction-session-lane.js'; import { HostSessionRetirementCoordinator } from '../server/session-retirement-coordinator.js'; +import { purgeSessionSidecars } from '../server/session-sidecar-purge.js'; import { waitFor as pollFor } from '@maka/core/test-only/async-primitives'; const CONNECTION_CONTEXT: ConnectionContext = { @@ -51,6 +52,37 @@ const CONNECTION_CONTEXT: ConnectionContext = { }; describe('Host Session retirement coordinator', () => { + test('retires context refs before draining every physical garbage batch', async () => { + const contextActions: string[] = []; + let garbageBatches = 0; + await purgeSessionSidecars( + { + artifacts: { purgeSessionArtifacts: async () => {} }, + sessionTodo: { purgeSessionState: async () => {} }, + contextOffload: { + retireSession: async (sessionId) => { + contextActions.push(`retire:${sessionId}`); + return { releasedReferences: 1, releasedLogicalBytes: 10 }; + }, + collectGarbage: async (input) => { + contextActions.push(`collect:${input.maxBlobs}`); + garbageBatches += 1; + return { deletedBlobs: 1, deletedBytes: 10, hasMore: garbageBatches < 3 }; + }, + }, + purgeOperationalState: async () => {}, + }, + 'session-context', + ); + + assert.deepEqual(contextActions, [ + 'retire:session-context', + 'collect:64', + 'collect:64', + 'collect:64', + ]); + }); + test('rejects ordinary archive and remove operations for the Coordination Session', async () => { await withHarness(async (harness) => { const created = await harness.store.createStableSession({ @@ -1237,6 +1269,7 @@ async function withHarness( actions.retiredContext.push(sessionId); return { releasedReferences: 0, releasedLogicalBytes: 0 }; }, + collectGarbage: async () => ({ deletedBlobs: 0, deletedBytes: 0, hasMore: false }), }, purgeOperationalState: async (sessionId) => { actions.purgedOperationalState.push(sessionId); diff --git a/packages/runtime-host/src/server/execution-composition.ts b/packages/runtime-host/src/server/execution-composition.ts index 72b03d558e..503ccaa4f7 100644 --- a/packages/runtime-host/src/server/execution-composition.ts +++ b/packages/runtime-host/src/server/execution-composition.ts @@ -304,6 +304,12 @@ export async function createExecutionRuntimeHostComposition( cause: storage.contextOffloadUnavailable?.cause, }); }, + collectGarbage: async (): Promise => { + throw new Error( + 'Context-offload Store is unavailable during context garbage collection', + { cause: storage.contextOffloadUnavailable?.cause }, + ); + }, } : undefined; const openedUsageStores = storage.usage; diff --git a/packages/runtime-host/src/server/session-retirement-coordinator.ts b/packages/runtime-host/src/server/session-retirement-coordinator.ts index 52b672de96..a4e6c6d776 100644 --- a/packages/runtime-host/src/server/session-retirement-coordinator.ts +++ b/packages/runtime-host/src/server/session-retirement-coordinator.ts @@ -123,7 +123,10 @@ export interface HostSessionRetirementCoordinatorOptions { readonly continuity: RetirementContinuity; readonly artifacts: Pick; readonly sessionTodo: Pick; - readonly contextOffload?: Pick; + readonly contextOffload?: Pick< + InteractiveContextOffloadWriter, + 'retireSession' | 'collectGarbage' + >; readonly purgeOperationalState: (sessionId: string) => Promise; readonly purgeAgentGraphState: (sessionId: string) => Promise; readonly worktrees?: Pick; diff --git a/packages/runtime-host/src/server/session-revision-coordinator.ts b/packages/runtime-host/src/server/session-revision-coordinator.ts index 68de60efac..44b00079e1 100644 --- a/packages/runtime-host/src/server/session-revision-coordinator.ts +++ b/packages/runtime-host/src/server/session-revision-coordinator.ts @@ -110,7 +110,7 @@ export interface HostSessionRevisionCoordinatorOptions { readonly sessionTodo: InteractiveSessionTodoWriter; readonly contextOffload?: Pick< InteractiveContextOffloadWriter, - 'copyReferences' | 'retireSession' + 'copyReferences' | 'retireSession' | 'collectGarbage' >; readonly manager: SessionManager; readonly admission: SessionAdmissionGate; @@ -146,7 +146,7 @@ export class HostSessionRevisionCoordinator { (header) => header.conversationCopy !== undefined, ); for (const header of copies) { - if (header.conversationCopy!.state === 'preparing') await this.#discard(header); + if (header.conversationCopy!.state === 'preparing') await this.#discardDuringRecovery(header); } const committed = copies.filter((header) => header.conversationCopy!.state === 'committed'); @@ -185,11 +185,21 @@ export class HostSessionRevisionCoordinator { if (retained.has(header.id)) { await this.options.manager.commitRevisionVersion(header.id); } else { - await this.#discard(header); + await this.#discardDuringRecovery(header); } } } + async #discardDuringRecovery(header: SessionHeader): Promise { + try { + await this.#discard(header); + } catch (error) { + console.error( + `[runtime-host] conversation copy cleanup deferred during recovery (${header.id}): ${conversationCopyCommitFailureDiagnostic(error)}`, + ); + } + } + async #copy( kind: ConversationCopyKind, input: SessionConversationCopyInput, diff --git a/packages/runtime-host/src/server/session-sidecar-purge.ts b/packages/runtime-host/src/server/session-sidecar-purge.ts index 59484fcf12..f3ec1661f3 100644 --- a/packages/runtime-host/src/server/session-sidecar-purge.ts +++ b/packages/runtime-host/src/server/session-sidecar-purge.ts @@ -24,10 +24,30 @@ import type { InteractiveContextOffloadWriter } from '@maka/storage/context-offl export interface SessionSidecarPurgeAuthority { readonly artifacts: Pick; readonly sessionTodo: Pick; - readonly contextOffload?: Pick; + readonly contextOffload?: Pick< + InteractiveContextOffloadWriter, + 'retireSession' | 'collectGarbage' + >; readonly purgeOperationalState: (sessionId: string) => Promise; } +const CONTEXT_GARBAGE_BATCH_BLOBS = 64; + +async function retireContextSession( + contextOffload: Pick, + sessionId: string, +): Promise { + await contextOffload.retireSession(sessionId); + for (;;) { + const collected = await contextOffload.collectGarbage({ + olderThan: Number.MAX_SAFE_INTEGER, + maxBlobs: CONTEXT_GARBAGE_BATCH_BLOBS, + maxBytes: Number.MAX_SAFE_INTEGER, + }); + if (!collected.hasMore) return; + } +} + export async function purgeSessionSidecars( authority: SessionSidecarPurgeAuthority, sessionId: string, @@ -35,7 +55,9 @@ export async function purgeSessionSidecars( const outcomes = await Promise.allSettled([ authority.artifacts.purgeSessionArtifacts(sessionId), authority.sessionTodo.purgeSessionState(sessionId), - ...(authority.contextOffload ? [authority.contextOffload.retireSession(sessionId)] : []), + ...(authority.contextOffload + ? [retireContextSession(authority.contextOffload, sessionId)] + : []), authority.purgeOperationalState(sessionId), ]); const failures = outcomes.flatMap((outcome) => diff --git a/packages/runtime/src/conversation-copy.ts b/packages/runtime/src/conversation-copy.ts index 1448fcfd96..15bc6f8824 100644 --- a/packages/runtime/src/conversation-copy.ts +++ b/packages/runtime/src/conversation-copy.ts @@ -160,40 +160,38 @@ export interface ConversationRuntimeLedgerCopyPlan { }[]; } -/** Finds durable Session context references that the exact copy will rewrite. */ -export function collectConversationCopySessionContextRefIds(input: { - readonly sourceSessionId: string; +interface ConversationCopyStorageReferenceInput { readonly messages: readonly StoredMessage[]; readonly runtimeEvents: readonly RuntimeEvent[]; readonly archivedResults: readonly string[]; -}): readonly string[] { - const refIds = new Set(); - const addRef = (ref: StorageRef): void => { - if (ref.kind === 'session_context' && ref.sessionId === input.sourceSessionId) { - refIds.add(ref.refId); - } - }; +} + +/** Walks every typed StorageRef site reached by conversation-copy rewriting. */ +function collectConversationCopyStorageRefs( + input: ConversationCopyStorageReferenceInput, +): readonly StorageRef[] { + const refs: StorageRef[] = []; const addContent = (content: ToolResultContent): void => { - if (content.kind === 'image') addRef(content.ref); + if (content.kind === 'image') refs.push(content.ref); }; const addSerialized = (value: unknown): void => { if (isArchivedToolResultPlaceholder(value)) return; try { addContent(decodePersistedToolResultContent(markPersisted(value))); } catch { - // Opaque tool results carry no typed Session context reference. + // Opaque tool results carry no typed StorageRef. } }; for (const message of input.messages) { if (message.type === 'user' && message.attachments) { - for (const attachment of message.attachments) addRef(attachment.ref); + for (const attachment of message.attachments) refs.push(attachment.ref); } else if (message.type === 'tool_result') { addContent(message.content); } } for (const event of input.runtimeEvents) { if (event.content?.kind === 'text' && event.content.attachments) { - for (const attachment of event.content.attachments) addRef(attachment.ref); + for (const attachment of event.content.attachments) refs.push(attachment.ref); } else if (event.content?.kind === 'function_response') { addSerialized(event.content.result); } @@ -201,6 +199,22 @@ export function collectConversationCopySessionContextRefIds(input: { for (const serializedResult of input.archivedResults) { addSerialized(deserializeToolResultArchive(serializedResult)); } + return refs; +} + +/** Finds durable Session context references that the exact copy will rewrite. */ +export function collectConversationCopySessionContextRefIds(input: { + readonly sourceSessionId: string; + readonly messages: readonly StoredMessage[]; + readonly runtimeEvents: readonly RuntimeEvent[]; + readonly archivedResults: readonly string[]; +}): readonly string[] { + const refIds = new Set(); + for (const ref of collectConversationCopyStorageRefs(input)) { + if (ref.kind === 'session_context' && ref.sessionId === input.sourceSessionId) { + refIds.add(ref.refId); + } + } return [...refIds].sort(); } @@ -819,38 +833,10 @@ export function collectConversationCopySessionFileRefs(input: { readonly archivedResults: readonly string[]; }): ReadonlySet { const refs = new Set(); - const addRef = (ref: StorageRef): void => { + for (const ref of collectConversationCopyStorageRefs(input)) { if (ref.kind === 'session_file' && ref.sessionId === input.sourceSessionId) { refs.add(ref.relativePath); } - }; - const addContent = (content: ToolResultContent): void => { - if (content.kind === 'image') addRef(content.ref); - }; - const addSerialized = (value: unknown): void => { - if (isArchivedToolResultPlaceholder(value)) return; - try { - addContent(decodePersistedToolResultContent(markPersisted(value))); - } catch { - // Opaque tool results carry no typed Session file reference. - } - }; - for (const message of input.messages) { - if (message.type === 'user' && message.attachments) { - for (const attachment of message.attachments) addRef(attachment.ref); - } else if (message.type === 'tool_result') { - addContent(message.content); - } - } - for (const event of input.runtimeEvents) { - if (event.content?.kind === 'text' && event.content.attachments) { - for (const attachment of event.content.attachments) addRef(attachment.ref); - } else if (event.content?.kind === 'function_response') { - addSerialized(event.content.result); - } - } - for (const serializedResult of input.archivedResults) { - addSerialized(deserializeToolResultArchive(serializedResult)); } return refs; }