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/packages/runtime-host/src/__tests__/execution-composition.test.ts b/packages/runtime-host/src/__tests__/execution-composition.test.ts index b43bafeccb..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, @@ -103,8 +104,34 @@ 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 }) => { + const requestFingerprint = `sha256:${'a'.repeat(64)}` as const; + const preparingSessionId = 'preparing-context-copy'; + const sessionStore = createSessionStore(root); + await 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 sessionStore.close?.(); await mkdir(join(root, CONTEXT_OFFLOAD_DATABASE_NAME)); const originalConsoleError = console.error; const diagnostics: string[] = []; @@ -114,12 +141,30 @@ 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, + ); + await composition.recover(); + assert.equal( + diagnostics.some((message) => + message.includes('conversation copy cleanup deferred during recovery'), + ), true, ); } finally { console.error = originalConsoleError; - await composition?.close(); + if (composition) { + await composition.close(); + } + } + const reopened = createSessionStore(root); + try { + assert.equal( + (await reopened.readHeaderSnapshot(preparingSessionId)).conversationCopy?.state, + 'preparing', + ); + } finally { + await reopened.close?.(); } }); }); diff --git a/packages/runtime-host/src/__tests__/protocol.test.ts b/packages/runtime-host/src/__tests__/protocol.test.ts index 1989f4b504..2ee9d5104d 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 > 93); + }); + 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..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({ @@ -239,7 +271,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 +1059,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 +1098,7 @@ async function withHarness( retiredCapabilities: [], retiredMessages: [], purgedArtifacts: [], - checkedContext: [], + retiredContext: [], purgedTasks: [], purgedOperationalState: [], purgedAgentGraphs: [], @@ -1232,8 +1264,12 @@ 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 }; + }, + collectGarbage: async () => ({ deletedBlobs: 0, deletedBytes: 0, hasMore: false }), }, 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..503ccaa4f7 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,28 @@ export async function createExecutionRuntimeHostComposition( const openedContextOffloadReader = openedContextOffloadStore ? createInteractiveContextOffloadReader(openedContextOffloadStore) : undefined; + 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, + }); + }, + collectGarbage: async (): Promise => { + throw new Error( + 'Context-offload Store is unavailable during context garbage collection', + { cause: storage.contextOffloadUnavailable?.cause }, + ); + }, + } + : undefined; const openedUsageStores = storage.usage; const openedShellRunStore = storage.shellRuns; const worktreeChildExecutor = createGitWorktreeChildExecutor({ @@ -403,7 +421,27 @@ 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, + }), + releaseImageSnapshot: async (input: { + readonly sessionId: string; + readonly refId: string; + }) => { + await openedContextOffloadStore.releaseReference(input); + }, + } + : {}), ...(sandboxManager ? { sandboxManager } : {}), ...(filesystemWorker ? { filesystemWorker } : {}), }; @@ -1548,6 +1586,7 @@ export async function createExecutionRuntimeHostComposition( stores, artifacts: openedArtifactStore, sessionTodo: sessionTodoStore, + ...(contextOffloadAuthority ? { contextOffload: contextOffloadAuthority } : {}), manager, admission: sessionAdmission, continuity: continuityCoordinator, @@ -1572,20 +1611,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', - ); - } - } - }, + ...(contextOffloadAuthority ? { contextOffload: contextOffloadAuthority } : {}), 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..a4e6c6d776 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,10 @@ export interface HostSessionRetirementCoordinatorOptions { readonly continuity: RetirementContinuity; readonly artifacts: Pick; readonly sessionTodo: Pick; - readonly assertNoContextOffloadReferences?: (sessionIds: readonly string[]) => Promise; + readonly contextOffload?: Pick< + InteractiveContextOffloadWriter, + 'retireSession' | 'collectGarbage' + >; readonly purgeOperationalState: (sessionId: string) => Promise; readonly purgeAgentGraphState: (sessionId: string) => Promise; readonly worktrees?: Pick; @@ -194,7 +198,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 +226,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 +342,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 +728,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..44b00079e1 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' | 'collectGarbage' + >; readonly manager: SessionManager; readonly admission: SessionAdmissionGate; readonly continuity: SessionContinuityCoordinator; @@ -140,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'); @@ -179,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, @@ -503,6 +519,29 @@ export class HostSessionRevisionCoordinator { ) .map(({ descriptor, serializedResult }) => [descriptor.artifactId, serializedResult]), ); + const sourceContextRefIds = collectConversationCopySessionContextRefIds({ + sourceSessionId: input.sourceSessionId, + 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'); + } + 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 +567,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 +861,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..f3ec1661f3 100644 --- a/packages/runtime-host/src/server/session-sidecar-purge.ts +++ b/packages/runtime-host/src/server/session-sidecar-purge.ts @@ -19,13 +19,35 @@ 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< + 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, @@ -33,6 +55,9 @@ export async function purgeSessionSidecars( const outcomes = await Promise.allSettled([ authority.artifacts.purgeSessionArtifacts(sessionId), authority.sessionTodo.purgeSessionState(sessionId), + ...(authority.contextOffload + ? [retireContextSession(authority.contextOffload, 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..26050bea7c 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,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, '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 () => { @@ -423,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 36bed08f12..5bf195594d 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', + }, + }, ], }, { @@ -733,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', @@ -800,6 +817,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 +830,66 @@ 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', + 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-selected-event', '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 +1082,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..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 } : {}), @@ -435,7 +436,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/__tests__/tool-runtime-durable-boundary.test.ts b/packages/runtime/src/__tests__/tool-runtime-durable-boundary.test.ts index 71b470df11..2c68e24c32 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,38 @@ 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 dbb1d7dc65..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'; @@ -184,11 +184,11 @@ export interface BuildBuiltinToolsOptions { sandboxPlatform?: SandboxPlatform; snapshotImage?: (input: { sessionId: string; - turnId: string; - name: string; + ownerId: string; bytes: Uint8Array; mimeType: string; - }) => Promise>; + }) => Promise>; + releaseImageSnapshot?: (input: { sessionId: string; refId: string }) => Promise; } export function buildBuiltinTools(options: BuildBuiltinToolsOptions = {}): MakaTool[] { @@ -356,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) { @@ -398,10 +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, - turnId: ctx.turnId, - name: basename(path), + 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 f013c89302..15bc6f8824 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 { 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,64 @@ export interface ConversationRuntimeLedgerCopyPlan { }[]; } +interface ConversationCopyStorageReferenceInput { + readonly messages: readonly StoredMessage[]; + readonly runtimeEvents: readonly RuntimeEvent[]; + readonly archivedResults: readonly string[]; +} + +/** 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') 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 StorageRef. + } + }; + for (const message of input.messages) { + if (message.type === 'user' && message.attachments) { + 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) refs.push(attachment.ref); + } else if (event.content?.kind === 'function_response') { + addSerialized(event.content.result); + } + } + 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(); +} + export interface CloneConversationRuntimeLedgerResult { readonly copiedMessages: readonly StoredMessage[]; readonly runIdMap: readonly { @@ -646,7 +705,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); @@ -771,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; } @@ -1779,12 +1813,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 { 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 = {