From dcbaa6c4cf6877ea254d6f6dc552073f1c4cb64f Mon Sep 17 00:00:00 2001 From: AstroHan Date: Fri, 4 Sep 2026 11:44:12 +0800 Subject: [PATCH 1/9] feat(storage): reclaim the prepared-request captures already on disk Removing the capture sink stops the growth but leaves the residue, and the residue is not the user's to clear: captures are `userVisible: false`, so no UI lists them, and the only thing that ever deleted one was purging its whole conversation. One workspace measured here holds 772 MB of them. The store made them, so the store disposes of them. `purgeRetiredCaptures` takes a bounded batch through the same mutation queue and purge-intent file as every other deletion, and reports what is left; the sweep started at host composition drains the rest behind live turns and stops when there is none. A store that never held captures does one empty pass. Interrupting it is safe by construction rather than by a checkpoint: each batch is durable on its own and the next pass reads whatever remains, so a crash, a close, or a stop all resume the same way. `purge` now shares its body with the sweep instead of restating it. Refs #4037 Generated-by: Claude Code (cherry picked from commit 89628f20e670278457e7dd041b583114379d642d) --- .../src/server/execution-composition.ts | 17 +++- .../src/__tests__/artifact-store.test.ts | 31 +++++++ .../src/__tests__/artifact-stores.test.ts | 86 +++++++++++++++++++ packages/storage/src/artifact-store.ts | 57 +++++++++--- packages/storage/src/artifact-stores.ts | 41 +++++++++ 5 files changed, 221 insertions(+), 11 deletions(-) diff --git a/packages/runtime-host/src/server/execution-composition.ts b/packages/runtime-host/src/server/execution-composition.ts index bb9145e053..0b3cea6a2f 100644 --- a/packages/runtime-host/src/server/execution-composition.ts +++ b/packages/runtime-host/src/server/execution-composition.ts @@ -80,7 +80,10 @@ import { type MakaTool } from '@maka/runtime/tool-runtime'; import { type RuntimeHostedRootAuthority } from '@maka/runtime/message-authority'; import { isHostedExecutionTerminal } from './hosted-execution-authority.js'; import { createAgentGraphControlStore } from '@maka/storage/agent-graph-control-store'; -import { createArtifactAttachmentResourceReader } from '@maka/storage/artifact-stores'; +import { + createArtifactAttachmentResourceReader, + startRetiredCaptureSweep, +} 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'; @@ -263,6 +266,16 @@ export async function createExecutionRuntimeHostComposition( `[runtime-host] optional context-offload Store could not be opened: ${generalizedErrorMessage(storage.contextOffloadUnavailable.cause)}`, ); } + let stopRetiredCaptureSweep: (() => void) | undefined; + // Reclaims what the retired prepared-request capture sink left on disk. It + // paces itself behind live turns and stops when the residue is gone, so a + // store that never held captures does one empty pass and finishes. + stopRetiredCaptureSweep = startRetiredCaptureSweep(storage.artifacts, { + onError: (error) => + console.error( + `[runtime-host] retired provider-request captures could not be reclaimed: ${generalizedErrorMessage(error)}`, + ), + }); const stores = storage.execution; let graphControlStore: ReturnType | undefined; let graphClient: HostAgentGraphCoordinator | undefined; @@ -1775,6 +1788,7 @@ export async function createExecutionRuntimeHostComposition( () => { unsubscribeTranscriptChanges?.(); unsubscribeUsageChanges?.(); + stopRetiredCaptureSweep?.(); }, ], releaseConnection: [(connectionId) => artifacts.releaseConnection(connectionId)], @@ -2001,6 +2015,7 @@ export async function createExecutionRuntimeHostComposition( try { unsubscribeTranscriptChanges?.(); unsubscribeUsageChanges?.(); + stopRetiredCaptureSweep?.(); } catch (closeError) { errors.push(closeError); } diff --git a/packages/storage/src/__tests__/artifact-store.test.ts b/packages/storage/src/__tests__/artifact-store.test.ts index 31c8c759ae..aafd71d204 100644 --- a/packages/storage/src/__tests__/artifact-store.test.ts +++ b/packages/storage/src/__tests__/artifact-store.test.ts @@ -322,6 +322,37 @@ describe('SQLite Artifact store', () => { }); }); + test('reclaims retired request captures in bounded batches and leaves everything else', async () => { + await withWorkspace(async (root) => { + const authority = createArtifactStoreWriteAuthority(root); + await authority.recover(); + const { store } = authority; + for (let index = 0; index < 3; index += 1) { + await store.create({ + ...artifactInput(`capture-${index}`, `request-${index}`, 10 + index), + source: 'provider_request_capture', + }); + } + await store.create(artifactInput('kept-artifact', 'kept', 20)); + + assert.deepEqual(await store.purgeRetiredCaptures(2), { purged: 2, remaining: 1 }); + assert.deepEqual(await store.purgeRetiredCaptures(2), { purged: 1, remaining: 0 }); + // The sweep stops on its own rather than spinning once the residue is gone. + assert.deepEqual(await store.purgeRetiredCaptures(2), { purged: 0, remaining: 0 }); + + assert.deepEqual( + (await store.list('session-1', { includeDeleted: true })).map((record) => record.id), + ['kept-artifact'], + ); + // Purge, not a tombstone: the bytes are what this reclaims. + const kept = await store.getInSession('session-1', 'kept-artifact'); + assert.ok(kept.record); + assert.deepEqual(await readdir(join(root, 'artifacts', 'session-1')), [ + basename(kept.record.relativePath), + ]); + }); + }); + test('excludes selected Artifacts from a conversation snapshot', async () => { await withWorkspace(async (root) => { const authority = createArtifactStoreWriteAuthority(root); diff --git a/packages/storage/src/__tests__/artifact-stores.test.ts b/packages/storage/src/__tests__/artifact-stores.test.ts index 8a137ec040..ecc0c3a869 100644 --- a/packages/storage/src/__tests__/artifact-stores.test.ts +++ b/packages/storage/src/__tests__/artifact-stores.test.ts @@ -25,6 +25,7 @@ import { after, describe, test } from 'node:test'; import { authenticateInteractiveArtifactStoreWriter, openInteractiveArtifactStoreForWrite, + startRetiredCaptureSweep, type InteractiveArtifactStoreWriter, } from '../artifact-stores.js'; import { ARTIFACT_WRITER_LOCK_FILE } from '../artifact-storage-layout.js'; @@ -153,6 +154,91 @@ describe('interactive artifact store authority', () => { }); }); +describe('retired request capture sweep', () => { + test('keeps taking batches until the residue is gone, then stops', async () => { + const limits: number[] = []; + // A store that purges fewer than asked still has to be revisited. + let residue = 5; + startRetiredCaptureSweep({ + purgeRetiredCaptures: async (limit) => { + limits.push(limit); + const purged = Math.min(2, residue); + residue -= purged; + return { purged, remaining: residue }; + }, + }); + + await settled(() => residue === 0); + const passes = limits.length; + assert.equal(passes, 3); + assert.ok( + limits.every((limit) => limit > 0 && Number.isSafeInteger(limit)), + 'every pass asks for a bounded batch', + ); + + await idleLongerThanOnePause(); + assert.equal(limits.length, passes, 'an empty residue does not schedule another pass'); + }); + + test('reports a failing batch once and gives up rather than retrying', async () => { + let calls = 0; + const errors: unknown[] = []; + startRetiredCaptureSweep( + { + purgeRetiredCaptures: async () => { + calls += 1; + throw new Error('artifact store is unavailable'); + }, + }, + { onError: (error) => errors.push(error) }, + ); + + await settled(() => errors.length === 1); + await idleLongerThanOnePause(); + assert.equal(calls, 1); + assert.match(String(errors[0]), /artifact store is unavailable/); + }); + + test('stop keeps the next batch from starting', async () => { + let calls = 0; + let release!: () => void; + const firstBatch = new Promise((resolve) => { + release = resolve; + }); + const stop = startRetiredCaptureSweep({ + purgeRetiredCaptures: async () => { + calls += 1; + await firstBatch; + return { purged: 1, remaining: 99 }; + }, + }); + + await settled(() => calls === 1); + stop(); + release(); + await idleLongerThanOnePause(); + assert.equal(calls, 1, 'a residue that remains is left for a later run'); + }); +}); + +/** Lets the sweep's own timers run until it reaches the state under test. */ +async function settled(done: () => boolean): Promise { + for (let attempt = 0; attempt < 500; attempt += 1) { + if (done()) return; + await new Promise((resolve) => { + setTimeout(resolve, 10); + }); + } + throw new Error('The capture sweep did not reach the expected state'); +} + +/** Long enough that a sweep which meant to continue would have called again. */ +async function idleLongerThanOnePause(): Promise { + await new Promise((resolve) => { + setTimeout(resolve, 400); + }); +} + function artifactInput(id: string, content: string | Uint8Array) { return { id, diff --git a/packages/storage/src/artifact-store.ts b/packages/storage/src/artifact-store.ts index 4f82cbd3bf..23e22cc08d 100644 --- a/packages/storage/src/artifact-store.ts +++ b/packages/storage/src/artifact-store.ts @@ -85,6 +85,11 @@ const PURGE_INTENT_SCHEMA_VERSION = 1 as const; const MAX_PURGE_INTENT_BYTES = 64 * 1024 * 1024; const ARTIFACT_PURGE_RESOLVE_CONCURRENCY = 8; +/** + * The source of the artifacts the retired capture sink wrote. The value stays + * a valid source so the records still decode; nothing produces new ones. + */ +const RETIRED_CAPTURE_ARTIFACT_SOURCE: ArtifactSource = 'provider_request_capture'; interface ArtifactSessionSnapshot { readonly records: readonly ArtifactRecord[]; @@ -219,6 +224,8 @@ export interface ArtifactAuthorityStore extends ArtifactStore { input: ConversationArtifactCopyInput, ): Promise; purgeSessionArtifacts(sessionId: string): Promise; + /** Drops up to `limit` retired prepared-request captures; reports what remains. */ + purgeRetiredCaptures(limit: number): Promise<{ purged: number; remaining: number }>; deleteUserArtifactInSession( sessionId: string, artifactId: string, @@ -882,19 +889,49 @@ class SqliteArtifactStore implements ArtifactAuthorityStore { await this.enqueueMutation(async () => { await this.prepareMutationUnlocked({ kind: 'purge' }); const ids = new Set(acceptedArtifactIds); - const records = this.records.filter((record) => ids.has(record.id)); - if (records.length === 0) return; - const paths = await this.preparePurgePathsUnlocked(records); - try { - await this.publishPurgeIntentUnlocked(records.map((record) => record.id)); - await this.completePurgeUnlocked(ids, paths); - } catch (error) { - this.invalidateWriterState(); - throw error; - } + await this.purgeRecordsUnlocked(this.records.filter((record) => ids.has(record.id))); }); } + /** + * Drops a bounded batch of the prepared-request captures left behind by the + * retired capture sink, and reports what is still there. + * + * These are not the user's to clean up: they never appear in the UI, and the + * only thing that ever reclaimed one was purging its whole conversation. The + * store made them, so the store disposes of them. + * + * Bounded so a large residue cannot monopolise the mutation queue, and safe + * to stop at any point: purge publishes its intent before touching a file, + * and the next call reads whatever is left. + */ + async purgeRetiredCaptures(limit: number): Promise<{ purged: number; remaining: number }> { + let outcome = { purged: 0, remaining: 0 }; + await this.enqueueMutation(async () => { + await this.prepareMutationUnlocked({ kind: 'purge' }); + const retired = this.records.filter( + (record) => record.source === RETIRED_CAPTURE_ARTIFACT_SOURCE, + ); + const batch = retired.slice(0, Math.max(0, limit)); + await this.purgeRecordsUnlocked(batch); + outcome = { purged: batch.length, remaining: retired.length - batch.length }; + }); + return outcome; + } + + private async purgeRecordsUnlocked(records: readonly ArtifactRecord[]): Promise { + if (records.length === 0) return; + const ids = new Set(records.map((record) => record.id)); + const paths = await this.preparePurgePathsUnlocked(records); + try { + await this.publishPurgeIntentUnlocked([...ids]); + await this.completePurgeUnlocked(ids, paths); + } catch (error) { + this.invalidateWriterState(); + throw error; + } + } + private async preparePurgePathsUnlocked( records: readonly ArtifactRecord[], ): Promise { diff --git a/packages/storage/src/artifact-stores.ts b/packages/storage/src/artifact-stores.ts index cea55fc60a..1255d19c95 100644 --- a/packages/storage/src/artifact-stores.ts +++ b/packages/storage/src/artifact-stores.ts @@ -87,6 +87,7 @@ export interface InteractiveArtifactStoreWriter extends DurableArtifactAttachmen input: ConversationArtifactCopyInput, ): Promise; purgeSessionArtifacts(sessionId: string): Promise; + purgeRetiredCaptures: ArtifactAuthorityStore['purgeRetiredCaptures']; listPage: ArtifactAuthorityStore['listPage']; listTurnArtifacts: ArtifactAuthorityStore['listTurnArtifacts']; getInSession: ArtifactAuthorityStore['getInSession']; @@ -210,6 +211,7 @@ function createWriterFacade( return run(() => store.copyConversationArtifacts(acceptedInput)); }, purgeSessionArtifacts: (sessionId) => run(() => store.purgeSessionArtifacts(sessionId)), + purgeRetiredCaptures: (limit) => run(() => store.purgeRetiredCaptures(limit)), deleteUserArtifactInSession: (sessionId, artifactId) => run(() => store.deleteUserArtifactInSession(sessionId, artifactId)), close: () => { @@ -220,6 +222,45 @@ function createWriterFacade( return Object.freeze(facade); } +/** + * Small enough that a live turn never queues behind a long purge, large enough + * that a store holding tens of thousands of captures drains within a session. + */ +const RETIRED_CAPTURE_SWEEP_BATCH = 256; +const RETIRED_CAPTURE_SWEEP_PAUSE_MS = 250; + +/** + * Drains the prepared-request captures the retired capture sink left behind. + * + * The sweep shares one mutation queue with live turns, so it takes bounded + * batches and waits between them rather than holding the queue for the whole + * residue. Stopping only means the next batch does not start: each batch is + * already durable on its own, and a later run continues from what is left. + */ +export function startRetiredCaptureSweep( + artifacts: Pick, + options: { readonly onError?: (error: unknown) => void } = {}, +): () => void { + let stopped = false; + void (async () => { + while (!stopped) { + try { + const { remaining } = await artifacts.purgeRetiredCaptures(RETIRED_CAPTURE_SWEEP_BATCH); + if (remaining === 0) return; + } catch (error) { + options.onError?.(error); + return; + } + await new Promise((resolve) => { + setTimeout(resolve, RETIRED_CAPTURE_SWEEP_PAUSE_MS).unref(); + }); + } + })(); + return () => { + stopped = true; + }; +} + function snapshotCreateInput(input: CreateArtifactInput): CreateArtifactInput { return Object.freeze({ ...input, From ad44076994710166b0cb021df1ff12669778f629 Mon Sep 17 00:00:00 2001 From: AstroHan Date: Fri, 4 Sep 2026 14:39:32 +0800 Subject: [PATCH 2/9] fix(runtime): let a copy outlive the capture it points at MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `captureArtifactId` was a required mapping, so once the sweep reclaimed a capture Artifact, branching or copying any Session holding a historical model call threw and could never succeed again. Every other Artifact reference may keep throwing: the bytes and the events naming them have always been removed together. This one cannot, because the sweep removes bytes an append-only ledger still names — so the key now leaves with them and the attempt keeps its record without the join. Also drops the two capture decoders left behind by #4631, which retired both provider-request event types from the emitted catalogue: a copy no longer reaches either one. Refs #4037 Generated-by: Claude Code --- .../src/__tests__/conversation-copy.test.ts | 90 +++++++++++++++++++ packages/runtime/src/conversation-copy.ts | 77 +++++----------- 2 files changed, 112 insertions(+), 55 deletions(-) diff --git a/packages/runtime/src/__tests__/conversation-copy.test.ts b/packages/runtime/src/__tests__/conversation-copy.test.ts index 38821c8462..bdc03f1b09 100644 --- a/packages/runtime/src/__tests__/conversation-copy.test.ts +++ b/packages/runtime/src/__tests__/conversation-copy.test.ts @@ -1602,6 +1602,96 @@ test('conversation copy rewrites the nested identity of a model call attempt', a } }); +test('conversation copy survives a capture the store has already reclaimed', async () => { + const root = await mkdtemp(join(tmpdir(), 'maka-conversation-reclaimed-capture-')); + try { + const runStore = createSqliteAgentRunStore(root); + const runtimeEventStore = createWorkspaceRuntimeStore(root); + await seedRun(runtimeEventStore, { + runId: 'run-source', + invocationId: 'invocation-source', + turnId: 'turn-1', + cwd: root, + }); + for (const event of [ + runtimeEvent({ + id: 'event-user', + role: 'user', + author: 'user', + content: { kind: 'text', text: 'copy this turn' }, + }), + runtimeEvent({ id: 'event-terminal', ts: 2, status: 'completed' }), + ]) { + await runtimeEventStore.appendRuntimeEvent('session-source', 'run-source', event); + } + await runStore.appendEvent('session-source', 'run-source', { + type: 'model_call_attempt_recorded', + id: 'attempt-source', + runId: 'run-source', + sessionId: 'session-source', + turnId: 'turn-1', + ts: 2, + data: { + schemaVersion: 1, + logicalCallId: 'logical-source', + attemptId: 'attempt-source', + traceId: 'trace-source', + sessionId: 'session-source', + runId: 'run-source', + turnId: 'turn-1', + step: 0, + attempt: 0, + callKind: 'main', + providerId: 'provider', + modelId: 'model', + captureArtifactId: 'artifact-gone', + startedAt: 1, + completedAt: 2, + latencyMs: 1, + status: 'completed', + usageBasis: 'reported', + inputTokens: 10, + outputTokens: 5, + costBasis: 'priced', + costUsd: 0.01, + }, + }); + const source = await new RuntimeReadModel({ + runtimeEventStore, + }).getSessionView('session-source'); + + // The sweep purged the capture Artifact, so the copy never sees it. Before + // the join keys were made droppable this threw and no Session holding a + // historical model call could be branched or copied again. + await cloneConversationRuntimeLedger({ + plan: await prepareTestCopyPlan(source, source.messages, runStore, runtimeEventStore), + copiedMessages: source.messages, + referenceMap: { + mode: 'exact', + linkedChildren: { mode: 'reject' }, + sourceSessionId: 'session-source', + targetSessionId: 'session-target', + artifactIds: new Map(), + relativePaths: new Map(), + }, + runStore, + runtimeEventStore, + newId: () => crypto.randomUUID(), + }); + const [targetRun] = await runtimeEventStore.listSessionInvocations('session-target'); + assert.ok(targetRun); + const events = await runStore.readEvents('session-target', targetRun.runId); + const attempt = events.find((event) => event.type === 'model_call_attempt_recorded'); + assert.ok(attempt, 'the attempt itself still copies'); + assert.equal(attempt.data?.captureArtifactId, undefined); + // Still a valid accounting authority without the join. + const decoded = decodeModelCallAttempt(attempt.data); + assert.equal(decoded.attemptId, attempt.id); + } finally { + await rm(root, { recursive: true, force: true }); + } +}); + test('conversation copy repairs a model call attempt stranded by a pre-fix copy', async () => { const root = await mkdtemp(join(tmpdir(), 'maka-conversation-model-call-legacy-')); try { diff --git a/packages/runtime/src/conversation-copy.ts b/packages/runtime/src/conversation-copy.ts index fc158e67a8..71ee9d92da 100644 --- a/packages/runtime/src/conversation-copy.ts +++ b/packages/runtime/src/conversation-copy.ts @@ -583,7 +583,6 @@ export async function cloneConversationRuntimeLedger( checkpointIds, transitionIds, transitionState, - operationalEventIds, providerTraceIds, logicalCallIds, ); @@ -874,7 +873,6 @@ function cloneAgentRunEvent( checkpointIds: Map, transitionIds: Map, transitionState: Map, - operationalEventIds: ReadonlyMap, providerTraceIds: ReadonlyMap, logicalCallIds: ReadonlyMap, ): EmittedAgentRunEvent | null { @@ -1084,7 +1082,9 @@ function rewriteModelCallAttempt( // nested identity is repaired rather than trusted and the *output* still // satisfies the `event.id === attemptId` contract. `decodeModelCallAttempt` // still rejects a schema-invalid payload. - const attempt = decodeModelCallAttempt(event.data); + // The join key is dropped by leaving it out of the spread: a conditional + // spread of `{}` cannot remove a key the spread above already placed. + const { captureArtifactId, ...attempt } = decodeModelCallAttempt(event.data); return { ...attempt, sessionId: ids.sessionId, @@ -1092,58 +1092,7 @@ function rewriteModelCallAttempt( attemptId: ids.attemptId, logicalCallId: requiredMappedId(logicalCallIds, attempt.logicalCallId, 'logical model call'), traceId: requiredMappedId(providerTraceIds, attempt.traceId, 'provider trace'), - ...(attempt.captureArtifactId !== undefined - ? { captureArtifactId: rewriteOwnedArtifactId(attempt.captureArtifactId, references) } - : {}), - }; -} - -function providerRequestCapture(event: AgentRunEvent): Record & { - readonly traceId: string; - readonly captureId: string; - readonly artifactId: string; -} { - const data = event.data; - if ( - !data || - data.captureId !== event.id || - typeof data.traceId !== 'string' || - typeof data.artifactId !== 'string' - ) { - throw new Error(`Cannot copy invalid provider request capture ${event.id}`); - } - return { - ...data, - traceId: data.traceId, - captureId: data.captureId, - artifactId: data.artifactId, - }; -} - -function providerRequestAttempt(event: AgentRunEvent): Record & { - readonly traceId: string; - readonly attemptId: string; - readonly captureId?: string; - readonly captureArtifactId?: string; -} { - const data = event.data; - const hasCaptureId = typeof data?.captureId === 'string'; - const hasArtifactId = typeof data?.captureArtifactId === 'string'; - if ( - !data || - data.attemptId !== event.id || - typeof data.traceId !== 'string' || - hasCaptureId !== hasArtifactId - ) { - throw new Error(`Cannot copy invalid provider request attempt ${event.id}`); - } - return { - ...data, - traceId: data.traceId, - attemptId: data.attemptId, - ...(hasCaptureId && hasArtifactId - ? { captureId: data.captureId as string, captureArtifactId: data.captureArtifactId as string } - : {}), + ...(captureArtifactId !== undefined ? capturedArtifactJoin(captureArtifactId, references) : {}), }; } @@ -1165,6 +1114,24 @@ function rewriteOwnedArtifactId( return rewriteOwnedId(sourceArtifactId, references.artifactIds, 'Artifact'); } +/** + * The `captureArtifactId` join, or nothing when its Artifact is gone. + * + * Every other Artifact reference throws on a missing target, because the bytes + * and the events naming them have always been removed together. This one is the + * exception: the capture Artifacts are reclaimed from disk on their own, while + * the attempts pointing at them stay in an append-only ledger. So a copy drops + * the key instead of failing on it. + */ +function capturedArtifactJoin( + sourceArtifactId: string, + references: ConversationCopyArtifactReferenceMap, +): { captureArtifactId?: string } { + if (references.mode === 'preserve_external') return { captureArtifactId: sourceArtifactId }; + const targetArtifactId = references.artifactIds.get(sourceArtifactId); + return targetArtifactId === undefined ? {} : { captureArtifactId: targetArtifactId }; +} + function rewriteOwnedId(sourceId: string, ids: ReadonlyMap, kind: string): string { return requiredMappedId(ids, sourceId, kind); } From 51117deb6e1c24700f0e411173d576068cfc7c84 Mon Sep 17 00:00:00 2001 From: AstroHan Date: Fri, 4 Sep 2026 14:39:32 +0800 Subject: [PATCH 3/9] fix(runtime-host): start the capture sweep once the store can be written A write authority refuses every mutation until it has recovered, and the sweep gives up after one failure. Starting it at composition meant its first batch always landed before recovery ran, so it reclaimed nothing and never retried. Both new tests recovered first and missed it. Refs #4037 Generated-by: Claude Code (cherry picked from commit d25a8ed6f89cdf8c439e236db2577ee9a0dab591) --- .../src/server/execution-composition.ts | 17 ++++++++--------- 1 file changed, 8 insertions(+), 9 deletions(-) diff --git a/packages/runtime-host/src/server/execution-composition.ts b/packages/runtime-host/src/server/execution-composition.ts index 0b3cea6a2f..57fc60314d 100644 --- a/packages/runtime-host/src/server/execution-composition.ts +++ b/packages/runtime-host/src/server/execution-composition.ts @@ -267,15 +267,6 @@ export async function createExecutionRuntimeHostComposition( ); } let stopRetiredCaptureSweep: (() => void) | undefined; - // Reclaims what the retired prepared-request capture sink left on disk. It - // paces itself behind live turns and stops when the residue is gone, so a - // store that never held captures does one empty pass and finishes. - stopRetiredCaptureSweep = startRetiredCaptureSweep(storage.artifacts, { - onError: (error) => - console.error( - `[runtime-host] retired provider-request captures could not be reclaimed: ${generalizedErrorMessage(error)}`, - ), - }); const stores = storage.execution; let graphControlStore: ReturnType | undefined; let graphClient: HostAgentGraphCoordinator | undefined; @@ -1772,6 +1763,14 @@ export async function createExecutionRuntimeHostComposition( state: async () => { await skills.recover(); await openedArtifactStore.recover(); + // Only now: a write authority refuses every mutation until it has + // recovered, and the sweep gives up on its first failure. + stopRetiredCaptureSweep = startRetiredCaptureSweep(storage.artifacts, { + onError: (error) => + console.error( + `[runtime-host] retired provider-request captures could not be reclaimed: ${generalizedErrorMessage(error)}`, + ), + }); }, }, drain: [ From 70fe70b71fb41240f588a17f58b0ba52482641c3 Mon Sep 17 00:00:00 2001 From: AstroHan Date: Fri, 4 Sep 2026 14:54:10 +0800 Subject: [PATCH 4/9] perf(storage): pace the capture sweep by what its last batch cost A batch holds the writer lock and its cost rises with everything the store holds, so a fixed 250 ms pause would not stay behind live turns on a store big enough for the sweep to matter. The pause is now at least three times the batch it follows. Also drops a sweep assertion that could not fail: the batch size is a module constant, so asserting it is a positive integer pinned nothing. Generated-by: Claude Code --- .../src/__tests__/artifact-stores.test.ts | 32 ++++++++++++++++--- packages/storage/src/artifact-stores.ts | 14 +++++++- 2 files changed, 40 insertions(+), 6 deletions(-) diff --git a/packages/storage/src/__tests__/artifact-stores.test.ts b/packages/storage/src/__tests__/artifact-stores.test.ts index ecc0c3a869..0107ba9c94 100644 --- a/packages/storage/src/__tests__/artifact-stores.test.ts +++ b/packages/storage/src/__tests__/artifact-stores.test.ts @@ -170,16 +170,38 @@ describe('retired request capture sweep', () => { await settled(() => residue === 0); const passes = limits.length; - assert.equal(passes, 3); - assert.ok( - limits.every((limit) => limit > 0 && Number.isSafeInteger(limit)), - 'every pass asks for a bounded batch', - ); + assert.equal(passes, 3, 'a batch that clears part of the residue is followed by another'); await idleLongerThanOnePause(); assert.equal(limits.length, passes, 'an empty residue does not schedule another pass'); }); + test('waits longer after a batch that took longer', async () => { + const gaps: number[] = []; + let previousEnd = 0; + let residue = 3; + // A batch that holds the writer lock for 300 ms must not be followed + // straight away on a store large enough for that to happen. + startRetiredCaptureSweep({ + purgeRetiredCaptures: async () => { + if (previousEnd) gaps.push(Date.now() - previousEnd); + await new Promise((resolve) => { + setTimeout(resolve, 300); + }); + residue -= 1; + previousEnd = Date.now(); + return { purged: 1, remaining: residue }; + }, + }); + + await settled(() => residue === 0); + assert.ok(gaps.length >= 1, 'the sweep took more than one batch'); + assert.ok( + gaps.every((gap) => gap >= 600), + `a 300 ms batch must be followed by a pause of at least 600 ms, saw ${gaps.join(', ')}`, + ); + }); + test('reports a failing batch once and gives up rather than retrying', async () => { let calls = 0; const errors: unknown[] = []; diff --git a/packages/storage/src/artifact-stores.ts b/packages/storage/src/artifact-stores.ts index 1255d19c95..a1ec89d695 100644 --- a/packages/storage/src/artifact-stores.ts +++ b/packages/storage/src/artifact-stores.ts @@ -228,6 +228,8 @@ function createWriterFacade( */ const RETIRED_CAPTURE_SWEEP_BATCH = 256; const RETIRED_CAPTURE_SWEEP_PAUSE_MS = 250; +/** Keeps the sweep to a quarter of the time, however long a batch takes. */ +const RETIRED_CAPTURE_SWEEP_DUTY_DIVISOR = 3; /** * Drains the prepared-request captures the retired capture sink left behind. @@ -244,15 +246,25 @@ export function startRetiredCaptureSweep( let stopped = false; void (async () => { while (!stopped) { + let batchMs: number; try { + const startedAt = Date.now(); const { remaining } = await artifacts.purgeRetiredCaptures(RETIRED_CAPTURE_SWEEP_BATCH); + batchMs = Date.now() - startedAt; if (remaining === 0) return; } catch (error) { options.onError?.(error); return; } + // A batch holds the writer lock, and its cost rises with everything the + // store holds. Waiting a multiple of what the last one took keeps the + // sweep behind live turns on a store large enough for that to matter. + const pauseMs = Math.max( + RETIRED_CAPTURE_SWEEP_PAUSE_MS, + batchMs * RETIRED_CAPTURE_SWEEP_DUTY_DIVISOR, + ); await new Promise((resolve) => { - setTimeout(resolve, RETIRED_CAPTURE_SWEEP_PAUSE_MS).unref(); + setTimeout(resolve, pauseMs).unref(); }); } })(); From a8f3087fb53809117fe569c497853ae936914e89 Mon Sep 17 00:00:00 2001 From: AstroHan Date: Fri, 4 Sep 2026 19:05:00 +0800 Subject: [PATCH 5/9] fix(storage): stop conversation copies from reintroducing retired captures A fork copied every Artifact of the source Session, retired provider-request captures included. That put condemned bytes back into a new Session -- and because the sweep stops for good once it sees an empty residue, anything copied after it finished was never reclaimed at all, which is the whole point of retiring them. It also let the copy and the sweep race over the same record. The copy now leaves that source out, which is also the only source a reader never asks for. Generated-by: Claude Code --- .../src/__tests__/artifact-store.test.ts | 32 +++++++++++++++++++ packages/storage/src/artifact-store.ts | 10 ++++-- 2 files changed, 40 insertions(+), 2 deletions(-) diff --git a/packages/storage/src/__tests__/artifact-store.test.ts b/packages/storage/src/__tests__/artifact-store.test.ts index aafd71d204..f60dbb5193 100644 --- a/packages/storage/src/__tests__/artifact-store.test.ts +++ b/packages/storage/src/__tests__/artifact-store.test.ts @@ -322,6 +322,38 @@ describe('SQLite Artifact store', () => { }); }); + test('a conversation copy leaves retired captures behind rather than reintroducing them', async () => { + await withWorkspace(async (root) => { + const authority = createArtifactStoreWriteAuthority(root); + await authority.recover(); + const { store } = authority; + const kept = await store.create(artifactInput('kept-artifact', 'kept', 10)); + await store.create({ + ...artifactInput('capture-artifact', 'request', 11), + source: 'provider_request_capture', + }); + + const copied = await store.copyConversationArtifacts({ + sourceSessionId: 'session-1', + targetSessionId: 'session-copy', + turnIds: ['turn-1'], + }); + + // The sweep runs once and stops when nothing is left. A copy that + // carried captures would put them back afterwards, so every fork would + // hand the new Session a fresh set of bytes nobody reads and nothing + // will come back for. + assert.equal(copied.artifactIds.get('capture-artifact'), undefined); + assert.ok(copied.artifactIds.get(kept.id)); + assert.deepEqual( + (await store.list('session-copy', { includeDeleted: true })).map((record) => record.source), + ['fixture'], + ); + // And the source keeps its own until the sweep takes them. + assert.deepEqual(await store.purgeRetiredCaptures(8), { purged: 1, remaining: 0 }); + }); + }); + test('reclaims retired request captures in bounded batches and leaves everything else', async () => { await withWorkspace(async (root) => { const authority = createArtifactStoreWriteAuthority(root); diff --git a/packages/storage/src/artifact-store.ts b/packages/storage/src/artifact-store.ts index 23e22cc08d..8c1513b2a6 100644 --- a/packages/storage/src/artifact-store.ts +++ b/packages/storage/src/artifact-store.ts @@ -428,7 +428,13 @@ class SqliteArtifactStore implements ArtifactAuthorityStore { (record) => record.sessionId === input.sourceSessionId && turnIds.has(record.turnId) && - !excludedArtifactIds.has(record.id), + !excludedArtifactIds.has(record.id) && + // A retired capture is on its way off disk and nothing reads one. + // Copying it would hand the new Session bytes already condemned, + // and would put records back after the sweep finished and stopped + // looking. Leaving them out also keeps the two from racing: the + // sweep can no longer delete a record this copy is holding. + record.source !== RETIRED_CAPTURE_ARTIFACT_SOURCE, ) .map((record) => ({ ...record })); for (const [sessionId, artifactIds] of requestedLinkedArtifactIds) { @@ -912,7 +918,7 @@ class SqliteArtifactStore implements ArtifactAuthorityStore { const retired = this.records.filter( (record) => record.source === RETIRED_CAPTURE_ARTIFACT_SOURCE, ); - const batch = retired.slice(0, Math.max(0, limit)); + const batch = retired.slice(0, limit); await this.purgeRecordsUnlocked(batch); outcome = { purged: batch.length, remaining: retired.length - batch.length }; }); From 82ff4a487d9aae0f7296d62c6c3efb9b36f18aa1 Mon Sep 17 00:00:00 2001 From: AstroHan Date: Fri, 4 Sep 2026 19:05:08 +0800 Subject: [PATCH 6/9] fix(storage): let the capture sweep survive a failed batch The sweep gave up on its first failed batch and logged it. Wired ahead of the store's recovery, every batch was refused, so it reclaimed nothing at all for anyone -- and the failure that says the least about the next batch is exactly the one it quit on: a purge that fails part way leaves the write authority refusing every mutation until something recovers it. It now retries, and gives up only after five consecutive failures. `onError` became the place that repairs what made the batch fail, which is why the host recovers the store there -- that recovery is also what hands the live turn's own writes back, not just this sweep's. Generated-by: Claude Code --- .../src/server/execution-composition.ts | 12 +- .../src/__tests__/artifact-stores.test.ts | 105 ++++++++++++++++-- packages/storage/src/artifact-stores.ts | 81 ++++++++++---- 3 files changed, 167 insertions(+), 31 deletions(-) diff --git a/packages/runtime-host/src/server/execution-composition.ts b/packages/runtime-host/src/server/execution-composition.ts index 57fc60314d..2e38c38f8d 100644 --- a/packages/runtime-host/src/server/execution-composition.ts +++ b/packages/runtime-host/src/server/execution-composition.ts @@ -1766,10 +1766,17 @@ export async function createExecutionRuntimeHostComposition( // Only now: a write authority refuses every mutation until it has // recovered, and the sweep gives up on its first failure. stopRetiredCaptureSweep = startRetiredCaptureSweep(storage.artifacts, { - onError: (error) => + onError: async (error) => { console.error( `[runtime-host] retired provider-request captures could not be reclaimed: ${generalizedErrorMessage(error)}`, - ), + ); + // A purge that fails part way leaves the write authority + // refusing every mutation until something recovers it -- not + // just this sweep's, but the live turn's tool results and the + // user's uploads. Recovering here is what hands those back, + // and it replays the purge intent the failed batch left. + await openedArtifactStore.recover(); + }, }); }, }, @@ -2014,7 +2021,6 @@ export async function createExecutionRuntimeHostComposition( try { unsubscribeTranscriptChanges?.(); unsubscribeUsageChanges?.(); - stopRetiredCaptureSweep?.(); } catch (closeError) { errors.push(closeError); } diff --git a/packages/storage/src/__tests__/artifact-stores.test.ts b/packages/storage/src/__tests__/artifact-stores.test.ts index 0107ba9c94..30f82b5085 100644 --- a/packages/storage/src/__tests__/artifact-stores.test.ts +++ b/packages/storage/src/__tests__/artifact-stores.test.ts @@ -155,6 +155,37 @@ describe('interactive artifact store authority', () => { }); describe('retired request capture sweep', () => { + test('drains a real writer, and cannot run before that writer has recovered', async () => { + await withInteractiveOwner(async (owner, _root, track) => { + const writer = track(await openInteractiveArtifactStoreForWrite(owner.lease)); + await writer.recover(); + for (let index = 0; index < 40; index += 1) { + await writer.create({ + ...artifactInput(`capture-${index}`, `request-${index}`), + source: 'provider_request_capture', + }); + } + await writer.create(artifactInput('kept', 'kept')); + + // Every other test here injects a fake, which is why the failure that + // actually shipped got through: wired ahead of recovery, each batch was + // refused, the sweep gave up on the first one, and it reclaimed nothing + // at all for anyone. Only the real writer and its real queue show that. + const errors: unknown[] = []; + startRetiredCaptureSweep(writer, { + onError: (error) => { + errors.push(error); + }, + }); + await settled( + async () => (await writer.listPage('session-1', { offset: 0, limit: 100 })).total === 1, + ); + assert.deepEqual(errors, []); + const remaining = await writer.listPage('session-1', { offset: 0, limit: 100 }); + assert.equal(remaining.records[0]?.id, 'kept'); + }); + }); + test('keeps taking batches until the residue is gone, then stops', async () => { const limits: number[] = []; // A store that purges fewer than asked still has to be revisited. @@ -202,7 +233,38 @@ describe('retired request capture sweep', () => { ); }); - test('reports a failing batch once and gives up rather than retrying', async () => { + test('retries a failed batch, and lets onError repair what made it fail', async () => { + let calls = 0; + const errors: unknown[] = []; + let residue = 2; + // The first failure says nothing about the second: a write authority that + // another mutation left needing recovery refuses this batch too, until + // something recovers it. That something is onError. + let recovered = false; + startRetiredCaptureSweep( + { + purgeRetiredCaptures: async () => { + calls += 1; + if (!recovered) throw new Error('Artifact write recovery is required'); + residue -= 1; + return { purged: 1, remaining: residue }; + }, + }, + { + onError: async (error) => { + errors.push(error); + recovered = true; + }, + }, + ); + + await settled(() => residue === 0); + assert.equal(errors.length, 1); + assert.match(String(errors[0]), /recovery is required/); + assert.ok(calls > 1, 'the sweep came back after the failure'); + }); + + test('gives up once failures stop looking temporary', async () => { let calls = 0; const errors: unknown[] = []; startRetiredCaptureSweep( @@ -212,13 +274,39 @@ describe('retired request capture sweep', () => { throw new Error('artifact store is unavailable'); }, }, - { onError: (error) => errors.push(error) }, + { + onError: (error) => { + errors.push(error); + }, + }, ); - await settled(() => errors.length === 1); + await settled(() => errors.length === 5); await idleLongerThanOnePause(); - assert.equal(calls, 1); - assert.match(String(errors[0]), /artifact store is unavailable/); + assert.equal(calls, 5, 'a permanent failure does not retry forever'); + }); + + test('shrinks the batch when one costs a live turn too much', async () => { + const limits: number[] = []; + let residue = 400; + // A batch costs what the whole store costs, not what its own size costs, + // so an expensive one has to be answered by asking for less. + startRetiredCaptureSweep({ + purgeRetiredCaptures: async (limit) => { + limits.push(limit); + await new Promise((resolve) => { + setTimeout(resolve, 400); + }); + residue -= limit; + return { purged: limit, remaining: Math.max(0, residue) }; + }, + }); + + await settled(() => limits.length >= 2); + assert.ok( + limits[1]! <= limits[0]!, + `a 400 ms batch must not be followed by a larger one, saw ${limits.join(' then ')}`, + ); }); test('stop keeps the next batch from starting', async () => { @@ -244,9 +332,10 @@ describe('retired request capture sweep', () => { }); /** Lets the sweep's own timers run until it reaches the state under test. */ -async function settled(done: () => boolean): Promise { - for (let attempt = 0; attempt < 500; attempt += 1) { - if (done()) return; +async function settled(done: () => boolean | Promise): Promise { + const deadline = Date.now() + 15_000; + while (Date.now() < deadline) { + if (await done()) return; await new Promise((resolve) => { setTimeout(resolve, 10); }); diff --git a/packages/storage/src/artifact-stores.ts b/packages/storage/src/artifact-stores.ts index a1ec89d695..5bd53e5780 100644 --- a/packages/storage/src/artifact-stores.ts +++ b/packages/storage/src/artifact-stores.ts @@ -223,46 +223,77 @@ function createWriterFacade( } /** - * Small enough that a live turn never queues behind a long purge, large enough - * that a store holding tens of thousands of captures drains within a session. + * How long a live turn may be made to wait for one batch. + * + * The mutation queue is FIFO with no priority, so a turn arriving while a + * batch runs waits the whole batch out. The only lever on that wait is how + * much a batch does -- and its cost is not its own size but the size of the + * store, because the purge guard resolves the path of every record it is NOT + * deleting. Measured at roughly 0.04 ms per record held, so 9,000 records + * cost about 370 ms whether the batch deletes 256 of them or 16. */ -const RETIRED_CAPTURE_SWEEP_BATCH = 256; +const RETIRED_CAPTURE_SWEEP_TARGET_BATCH_MS = 100; +const RETIRED_CAPTURE_SWEEP_MIN_BATCH = 16; +const RETIRED_CAPTURE_SWEEP_MAX_BATCH = 256; const RETIRED_CAPTURE_SWEEP_PAUSE_MS = 250; /** Keeps the sweep to a quarter of the time, however long a batch takes. */ const RETIRED_CAPTURE_SWEEP_DUTY_DIVISOR = 3; +/** + * How many batches may fail in a row before the sweep gives up. + * + * Most of what fails here is not permanent. Another mutation's failure makes + * the write authority refuse everything until something recovers it, and a + * full or briefly unavailable disk clears on its own -- so the first failure + * says nothing about the second. Giving up on it is how this sweep once + * reclaimed nothing at all, for every user, without saying so. + */ +const RETIRED_CAPTURE_SWEEP_MAX_CONSECUTIVE_FAILURES = 5; +const RETIRED_CAPTURE_SWEEP_RETRY_MS = 1_000; /** * Drains the prepared-request captures the retired capture sink left behind. * - * The sweep shares one mutation queue with live turns, so it takes bounded - * batches and waits between them rather than holding the queue for the whole - * residue. Stopping only means the next batch does not start: each batch is - * already durable on its own, and a later run continues from what is left. + * The sweep shares one mutation queue with live turns, so it takes batches + * sized to what the last one cost and waits between them rather than holding + * the queue for the whole residue. Stopping only means the next batch does not + * start: each batch is already durable on its own, and a later run continues + * from what is left. + * + * `onError` is where the decision to repair belongs -- the sweep knows a batch + * failed, not what would make the next one succeed. */ export function startRetiredCaptureSweep( artifacts: Pick, - options: { readonly onError?: (error: unknown) => void } = {}, + options: { readonly onError?: (error: unknown) => void | Promise } = {}, ): () => void { let stopped = false; void (async () => { + let batch = RETIRED_CAPTURE_SWEEP_MIN_BATCH; + let failures = 0; while (!stopped) { - let batchMs: number; + let pauseMs: number; try { const startedAt = Date.now(); - const { remaining } = await artifacts.purgeRetiredCaptures(RETIRED_CAPTURE_SWEEP_BATCH); - batchMs = Date.now() - startedAt; + const { remaining } = await artifacts.purgeRetiredCaptures(batch); + const batchMs = Date.now() - startedAt; if (remaining === 0) return; + failures = 0; + batch = nextSweepBatch(batch, batchMs); + pauseMs = Math.max( + RETIRED_CAPTURE_SWEEP_PAUSE_MS, + batchMs * RETIRED_CAPTURE_SWEEP_DUTY_DIVISOR, + ); } catch (error) { - options.onError?.(error); - return; + failures += 1; + try { + await options.onError?.(error); + } catch { + // A repair that fails leaves the same state the batch did, and the + // failure below is already being counted. + } + if (failures >= RETIRED_CAPTURE_SWEEP_MAX_CONSECUTIVE_FAILURES) return; + pauseMs = RETIRED_CAPTURE_SWEEP_RETRY_MS; } - // A batch holds the writer lock, and its cost rises with everything the - // store holds. Waiting a multiple of what the last one took keeps the - // sweep behind live turns on a store large enough for that to matter. - const pauseMs = Math.max( - RETIRED_CAPTURE_SWEEP_PAUSE_MS, - batchMs * RETIRED_CAPTURE_SWEEP_DUTY_DIVISOR, - ); await new Promise((resolve) => { setTimeout(resolve, pauseMs).unref(); }); @@ -273,6 +304,16 @@ export function startRetiredCaptureSweep( }; } +/** Steers the next batch toward one that costs about the target. */ +function nextSweepBatch(batch: number, batchMs: number): number { + const scaled = + batchMs > 0 ? Math.round((batch * RETIRED_CAPTURE_SWEEP_TARGET_BATCH_MS) / batchMs) : batch * 2; + return Math.min( + RETIRED_CAPTURE_SWEEP_MAX_BATCH, + Math.max(RETIRED_CAPTURE_SWEEP_MIN_BATCH, scaled), + ); +} + function snapshotCreateInput(input: CreateArtifactInput): CreateArtifactInput { return Object.freeze({ ...input, From 266338ea25d82f8b82bfe10cae771fe53b6791b6 Mon Sep 17 00:00:00 2001 From: AstroHan Date: Fri, 4 Sep 2026 21:39:24 +0800 Subject: [PATCH 7/9] fix(storage): exclude retired captures from every copy selection pass A conversation copy names records three ways -- by turn, by a linked child Session, and by an explicit include list -- and only the turn-scoped pass excluded the captures on their way off disk. The property the exclusion exists for belongs to the copy, not to one of its passes, so it now lives in one predicate all three ask. Naming a capture outright still fails the copy rather than silently dropping it: a caller asking for a linked Artifact that is being reclaimed is asking for something that will not be there. Generated-by: Claude Code --- .../src/__tests__/artifact-store.test.ts | 52 +++++++++++++++++++ packages/storage/src/artifact-store.ts | 30 +++++++---- 2 files changed, 71 insertions(+), 11 deletions(-) diff --git a/packages/storage/src/__tests__/artifact-store.test.ts b/packages/storage/src/__tests__/artifact-store.test.ts index f60dbb5193..f3df5577c0 100644 --- a/packages/storage/src/__tests__/artifact-store.test.ts +++ b/packages/storage/src/__tests__/artifact-store.test.ts @@ -354,6 +354,58 @@ describe('SQLite Artifact store', () => { }); }); + test('and leaves them behind on the linked and included paths too', async () => { + await withWorkspace(async (root) => { + const authority = createArtifactStoreWriteAuthority(root); + await authority.recover(); + const { store } = authority; + // A Side Conversation copy names artifacts three ways: by turn, by a + // linked child Session, and by an explicit include list for the refs + // that carry no conversation turn. All three reach the same records. + const linkedKept = await store.create({ + ...artifactInput('linked-kept', 'child result', 10), + sessionId: 'session-child', + }); + await store.create({ + ...artifactInput('linked-capture', 'child request', 11), + sessionId: 'session-child', + source: 'provider_request_capture', + }); + await store.create({ + ...artifactInput('upload-capture', 'uploaded request', 12), + turnId: 'upload-1', + source: 'provider_request_capture', + }); + + const copied = await store.copyConversationArtifacts({ + sourceSessionId: 'session-1', + targetSessionId: 'session-copy', + turnIds: ['turn-1'], + includeArtifactIds: ['upload-capture'], + linkedArtifacts: [{ sessionId: 'session-child', artifactIds: [linkedKept.id] }], + }); + + assert.ok(copied.artifactIds.get(linkedKept.id), 'ordinary linked artifacts still copy'); + assert.equal(copied.artifactIds.get('upload-capture'), undefined); + assert.deepEqual( + (await store.list('session-copy', { includeDeleted: true })).map((record) => record.source), + ['fixture'], + ); + + // A caller that names a capture outright is asking for a record that is + // on its way off disk, and the copy refuses rather than carrying it. + await assert.rejects( + store.copyConversationArtifacts({ + sourceSessionId: 'session-1', + targetSessionId: 'session-copy-2', + turnIds: ['turn-1'], + linkedArtifacts: [{ sessionId: 'session-child', artifactIds: ['linked-capture'] }], + }), + /Linked Artifact linked-capture could not be copied/, + ); + }); + }); + test('reclaims retired request captures in bounded batches and leaves everything else', async () => { await withWorkspace(async (root) => { const authority = createArtifactStoreWriteAuthority(root); diff --git a/packages/storage/src/artifact-store.ts b/packages/storage/src/artifact-store.ts index 8c1513b2a6..2a4d584b7f 100644 --- a/packages/storage/src/artifact-store.ts +++ b/packages/storage/src/artifact-store.ts @@ -91,6 +91,19 @@ const ARTIFACT_PURGE_RESOLVE_CONCURRENCY = 8; */ const RETIRED_CAPTURE_ARTIFACT_SOURCE: ArtifactSource = 'provider_request_capture'; +/** + * A record on its way off disk, which no copy may carry anywhere. + * + * Copying one would hand the target Session bytes already condemned, and would + * put records back after the sweep finished and stopped looking. Leaving them + * out also keeps the two from racing: the sweep can no longer delete a record a + * copy is holding. That property belongs to the copy, not to one of its three + * selection passes, so every pass asks the same question here. + */ +function isRetiredCapture(record: ArtifactRecord): boolean { + return record.source === RETIRED_CAPTURE_ARTIFACT_SOURCE; +} + interface ArtifactSessionSnapshot { readonly records: readonly ArtifactRecord[]; readonly revision: ArtifactListRevision; @@ -429,12 +442,7 @@ class SqliteArtifactStore implements ArtifactAuthorityStore { record.sessionId === input.sourceSessionId && turnIds.has(record.turnId) && !excludedArtifactIds.has(record.id) && - // A retired capture is on its way off disk and nothing reads one. - // Copying it would hand the new Session bytes already condemned, - // and would put records back after the sweep finished and stopped - // looking. Leaving them out also keeps the two from racing: the - // sweep can no longer delete a record this copy is holding. - record.source !== RETIRED_CAPTURE_ARTIFACT_SOURCE, + !isRetiredCapture(record), ) .map((record) => ({ ...record })); for (const [sessionId, artifactIds] of requestedLinkedArtifactIds) { @@ -443,7 +451,8 @@ class SqliteArtifactStore implements ArtifactAuthorityStore { (candidate) => candidate.sessionId === sessionId && candidate.id === artifactId && - candidate.status !== 'deleted', + candidate.status !== 'deleted' && + !isRetiredCapture(candidate), ); if (!record) throw new Error(`Linked Artifact ${artifactId} could not be copied`); selected.push({ ...record }); @@ -455,7 +464,8 @@ class SqliteArtifactStore implements ArtifactAuthorityStore { record.sessionId === input.sourceSessionId && includedArtifactIds.has(record.id) && !excludedArtifactIds.has(record.id) && - !selectedIds.has(record.id) + !selectedIds.has(record.id) && + !isRetiredCapture(record) ) { selected.push({ ...record }); selectedIds.add(record.id); @@ -915,9 +925,7 @@ class SqliteArtifactStore implements ArtifactAuthorityStore { let outcome = { purged: 0, remaining: 0 }; await this.enqueueMutation(async () => { await this.prepareMutationUnlocked({ kind: 'purge' }); - const retired = this.records.filter( - (record) => record.source === RETIRED_CAPTURE_ARTIFACT_SOURCE, - ); + const retired = this.records.filter(isRetiredCapture); const batch = retired.slice(0, limit); await this.purgeRecordsUnlocked(batch); outcome = { purged: batch.length, remaining: retired.length - batch.length }; From 17b36a21ff2edd8db59abefa582a6cf4692663a2 Mon Sep 17 00:00:00 2001 From: AstroHan Date: Fri, 4 Sep 2026 21:39:32 +0800 Subject: [PATCH 8/9] perf(storage): drop the adaptive batch the purge guard makes useless The sweep steered its batch size toward a batch costing ~100 ms, which it can never reach and never should have tried: the purge guard resolves the path of every record it is NOT deleting, so a batch costs what the store costs. Asking for fewer records pays that same toll again for less work, which is the opposite of what steering down was for -- the constant's own comment already said the cost is the store's, not the batch's. A fixed batch and the pause that scales with what the last one measured. That pause is the lever that was doing the work all along. Generated-by: Claude Code --- .../src/__tests__/artifact-stores.test.ts | 12 +++-- packages/storage/src/artifact-stores.ts | 45 +++++++------------ 2 files changed, 22 insertions(+), 35 deletions(-) diff --git a/packages/storage/src/__tests__/artifact-stores.test.ts b/packages/storage/src/__tests__/artifact-stores.test.ts index 30f82b5085..c50bdab1b5 100644 --- a/packages/storage/src/__tests__/artifact-stores.test.ts +++ b/packages/storage/src/__tests__/artifact-stores.test.ts @@ -286,11 +286,12 @@ describe('retired request capture sweep', () => { assert.equal(calls, 5, 'a permanent failure does not retry forever'); }); - test('shrinks the batch when one costs a live turn too much', async () => { + test('does not shrink a batch that cost a lot, because the cost is not the batch', async () => { const limits: number[] = []; let residue = 400; - // A batch costs what the whole store costs, not what its own size costs, - // so an expensive one has to be answered by asking for less. + // A batch costs what the whole store costs, not what its own size costs. + // Asking for less would pay that same toll again for fewer records, so an + // expensive batch is answered by waiting longer, not by taking less. startRetiredCaptureSweep({ purgeRetiredCaptures: async (limit) => { limits.push(limit); @@ -303,10 +304,7 @@ describe('retired request capture sweep', () => { }); await settled(() => limits.length >= 2); - assert.ok( - limits[1]! <= limits[0]!, - `a 400 ms batch must not be followed by a larger one, saw ${limits.join(' then ')}`, - ); + assert.deepEqual(limits.slice(0, 2), [256, 256]); }); test('stop keeps the next batch from starting', async () => { diff --git a/packages/storage/src/artifact-stores.ts b/packages/storage/src/artifact-stores.ts index 5bd53e5780..2a8ad31455 100644 --- a/packages/storage/src/artifact-stores.ts +++ b/packages/storage/src/artifact-stores.ts @@ -223,18 +223,19 @@ function createWriterFacade( } /** - * How long a live turn may be made to wait for one batch. + * How much one batch deletes -- as much as it may, not as little. * - * The mutation queue is FIFO with no priority, so a turn arriving while a - * batch runs waits the whole batch out. The only lever on that wait is how - * much a batch does -- and its cost is not its own size but the size of the - * store, because the purge guard resolves the path of every record it is NOT - * deleting. Measured at roughly 0.04 ms per record held, so 9,000 records - * cost about 370 ms whether the batch deletes 256 of them or 16. + * A batch costs what the store costs, not what its own size costs: the purge + * guard resolves the path of every record it is NOT deleting, measured at + * roughly 0.04 ms per record held, so 9,000 records cost about 370 ms whether + * the batch deletes 256 of them or 16. That fixed cost is per batch, so a + * smaller batch cannot shorten the wait a live turn takes -- it only makes the + * residue take more batches, each paying the same toll again. + * + * The lever that does work is the pause below, which keeps the sweep out of the + * queue for three times as long as it was in it. */ -const RETIRED_CAPTURE_SWEEP_TARGET_BATCH_MS = 100; -const RETIRED_CAPTURE_SWEEP_MIN_BATCH = 16; -const RETIRED_CAPTURE_SWEEP_MAX_BATCH = 256; +const RETIRED_CAPTURE_SWEEP_BATCH = 256; const RETIRED_CAPTURE_SWEEP_PAUSE_MS = 250; /** Keeps the sweep to a quarter of the time, however long a batch takes. */ const RETIRED_CAPTURE_SWEEP_DUTY_DIVISOR = 3; @@ -253,11 +254,11 @@ const RETIRED_CAPTURE_SWEEP_RETRY_MS = 1_000; /** * Drains the prepared-request captures the retired capture sink left behind. * - * The sweep shares one mutation queue with live turns, so it takes batches - * sized to what the last one cost and waits between them rather than holding - * the queue for the whole residue. Stopping only means the next batch does not - * start: each batch is already durable on its own, and a later run continues - * from what is left. + * The sweep shares one mutation queue with live turns, so it takes bounded + * batches and waits between them for as long as the last one cost, rather than + * holding the queue for the whole residue. Stopping only means the next batch + * does not start: each batch is already durable on its own, and a later run + * continues from what is left. * * `onError` is where the decision to repair belongs -- the sweep knows a batch * failed, not what would make the next one succeed. @@ -268,17 +269,15 @@ export function startRetiredCaptureSweep( ): () => void { let stopped = false; void (async () => { - let batch = RETIRED_CAPTURE_SWEEP_MIN_BATCH; let failures = 0; while (!stopped) { let pauseMs: number; try { const startedAt = Date.now(); - const { remaining } = await artifacts.purgeRetiredCaptures(batch); + const { remaining } = await artifacts.purgeRetiredCaptures(RETIRED_CAPTURE_SWEEP_BATCH); const batchMs = Date.now() - startedAt; if (remaining === 0) return; failures = 0; - batch = nextSweepBatch(batch, batchMs); pauseMs = Math.max( RETIRED_CAPTURE_SWEEP_PAUSE_MS, batchMs * RETIRED_CAPTURE_SWEEP_DUTY_DIVISOR, @@ -304,16 +303,6 @@ export function startRetiredCaptureSweep( }; } -/** Steers the next batch toward one that costs about the target. */ -function nextSweepBatch(batch: number, batchMs: number): number { - const scaled = - batchMs > 0 ? Math.round((batch * RETIRED_CAPTURE_SWEEP_TARGET_BATCH_MS) / batchMs) : batch * 2; - return Math.min( - RETIRED_CAPTURE_SWEEP_MAX_BATCH, - Math.max(RETIRED_CAPTURE_SWEEP_MIN_BATCH, scaled), - ); -} - function snapshotCreateInput(input: CreateArtifactInput): CreateArtifactInput { return Object.freeze({ ...input, From 9b73fa9d81d604ec6db1c174a929d46a8700ad94 Mon Sep 17 00:00:00 2001 From: AstroHan Date: Fri, 4 Sep 2026 22:24:51 +0800 Subject: [PATCH 9/9] fix: let every capture reference tolerate a reclaimed referent This branch already established the rule -- a reference to a capture must survive its target being reclaimed, because the ledger naming it is append-only -- and then applied it to exactly one field, `captureArtifactId`. The same ids live in a second place. A child result carries `artifactIds`, and that list is every Artifact the child's turn held: `listTurnArtifacts` filters on turn and status, never on source, so a Session that ever spawned a subagent has capture ids inside a typed tool result today. Three readers resolved them and failed on a miss: the Agent Graph reference validator, the copy's linked-child selection, and the copy's id rewrite. Reclaiming a capture therefore made a Session unable to take a Side Conversation or a revision -- the very Sessions this branch exists to reclaim bytes from. All three now carry what is still there and drop the rest. The boundary they still enforce is the one that was never about survival: a reference that crosses a Session or a lineage was never admissible, and still fails. An archived tool result's Artifact keeps throwing too -- it holds that result's own bytes, and the two are removed together. Generated-by: Claude Code --- .../session-revision-graph-references.test.ts | 25 +++--- .../session-revision-graph-references.ts | 11 ++- .../src/__tests__/conversation-copy.test.ts | 81 ++++++++++++------- packages/runtime/src/conversation-copy.ts | 38 ++++++--- .../src/__tests__/artifact-store.test.ts | 25 +++--- packages/storage/src/artifact-store.ts | 8 +- 6 files changed, 122 insertions(+), 66 deletions(-) diff --git a/packages/runtime-host/src/__tests__/session-revision-graph-references.test.ts b/packages/runtime-host/src/__tests__/session-revision-graph-references.test.ts index d42a9b9f59..a963491729 100644 --- a/packages/runtime-host/src/__tests__/session-revision-graph-references.test.ts +++ b/packages/runtime-host/src/__tests__/session-revision-graph-references.test.ts @@ -208,16 +208,6 @@ test('Agent Graph revision references reject incomplete or mismatched provenance input: { artifactTurnId: 'other-turn' }, code: 'operation_unavailable', }, - { - name: 'deleted Artifact', - input: { artifactStatus: 'deleted' }, - code: 'operation_unavailable', - }, - { - name: 'missing Artifact', - input: { artifactMissing: true }, - code: 'operation_unavailable', - }, { name: 'active child Session', input: { childActive: true }, @@ -231,6 +221,21 @@ test('Agent Graph revision references reject incomplete or mismatched provenance } }); +test('Agent Graph revision references outlive the Artifacts they name', async () => { + // A child result lists every Artifact its turn held, in a ledger that can + // never be rewritten -- so an id in it outlives what it named. The retired + // provider-request captures are reclaimed on their own, and a user may + // delete a child's Artifact; neither may cost the Session its ability to + // take a revision. What this checks is that a reference does not reach + // outside its own child and lineage, which `wrong Artifact turn` above + // still fails on. + const reclaimed = await prepare({ artifactMissing: true }); + assert.equal(reclaimed.ok, true); + + const userDeleted = await prepare({ artifactStatus: 'deleted' }); + assert.equal(userDeleted.ok, true); +}); + test('Agent Graph revision references reject invalid ownership boundaries', async () => { const genericChild = childHeader({ graph: false }); const generic = await prepare({ sessionHeaders: [sessionHeader(ROOT_SESSION_ID), genericChild] }); diff --git a/packages/runtime-host/src/server/session-revision-graph-references.ts b/packages/runtime-host/src/server/session-revision-graph-references.ts index d358f992df..9e9385e462 100644 --- a/packages/runtime-host/src/server/session-revision-graph-references.ts +++ b/packages/runtime-host/src/server/session-revision-graph-references.ts @@ -241,14 +241,21 @@ export async function prepareAgentGraphRevisionReferences( ) { return failure('operation_unavailable', 'Retained Agent Graph run reference is unavailable'); } + // A child result names every Artifact its turn held, and the ledger that + // records it can never be rewritten -- so an id in it outlives whatever it + // named. What this checks is therefore that a reference does not reach + // outside its own child and lineage, not that its target survived: the + // retired provider-request captures are reclaimed on their own, and a user + // may delete a child's Artifact. A reference whose target is gone stays + // admissible and simply resolves to nothing, while one that crosses a + // Session or a lineage was never admissible and still fails. for (const artifactId of request.artifactIds) { const artifact = await dependencies.artifacts .getInSession(childSessionId, artifactId) .catch(() => null); + if (!artifact?.record || artifact.record.status === 'deleted') continue; if ( - !artifact?.record || artifact.record.sessionId !== childSessionId || - artifact.record.status === 'deleted' || !lineage.turnIds.has(artifact.record.turnId) ) { return failure('operation_unavailable', 'Retained Agent Graph Artifact is unavailable'); diff --git a/packages/runtime/src/__tests__/conversation-copy.test.ts b/packages/runtime/src/__tests__/conversation-copy.test.ts index bdc03f1b09..4bb9115a0f 100644 --- a/packages/runtime/src/__tests__/conversation-copy.test.ts +++ b/packages/runtime/src/__tests__/conversation-copy.test.ts @@ -1001,16 +1001,31 @@ test('conversation copy rewrites owned references without changing opaque tool p preserved.type === 'user' ? preserved.attachments?.[0]?.ref : undefined, messages[0]?.type === 'user' ? messages[0].attachments?.[0]?.ref : undefined, ); - for (const message of [messages[2]!, messages[3]!]) { - assert.throws( - () => - rewriteConversationCopyMessage(message, { - ...references, - artifactIds: new Map(), - }), - /missing Artifact artifact-source/, - ); - } + // An archived tool result's Artifact holds that result's own bytes, and the + // two are removed together, so a copy that lost it has lost what a reader + // will ask for. + assert.throws( + () => + rewriteConversationCopyMessage(messages[2]!, { + ...references, + artifactIds: new Map(), + }), + /missing Artifact artifact-source/, + ); + // A child result is the opposite case: it lists every Artifact its turn + // held, in a ledger that cannot be rewritten, so an id in it outlives what + // it named. The copy carries what is still there and drops the rest, rather + // than making a whole Session uncopyable over a reclaimed byte nobody reads. + const reclaimed = rewriteConversationCopyMessage(messages[3]!, { + ...references, + artifactIds: new Map(), + }); + assert.deepEqual( + reclaimed.type === 'tool_result' && reclaimed.content.kind === 'agent_swarm' + ? reclaimed.content.items[0]?.artifactIds + : undefined, + [], + ); assert.throws( () => rewriteConversationCopyMessage(messages[3]!, { @@ -2079,29 +2094,37 @@ test('conversation copy clones one terminal Runtime ledger with new owned identi const source = await new RuntimeReadModel({ runtimeEventStore, }).getSessionView('session-source'); - await assert.rejects( - async () => - cloneConversationRuntimeLedger({ - plan: await prepareTestCopyPlan(source, source.messages, runStore, runtimeEventStore), - copiedMessages: source.messages, - referenceMap: { - mode: 'exact', - linkedChildren: { mode: 'reject' }, - sourceSessionId: 'session-source', - targetSessionId: 'session-missing-artifact', - artifactIds: new Map([['artifact-source', 'artifact-target']]), - relativePaths: new Map(), - }, - runStore, - runtimeEventStore, - newId: () => crypto.randomUUID(), - }), - /missing Artifact artifact-deleted/, + // The child result names an Artifact the copy has no mapping for, because + // it was reclaimed after the ledger recorded it. The copy carries the run + // and drops that one id, rather than making the Session uncopyable. + const withReclaimed = await cloneConversationRuntimeLedger({ + plan: await prepareTestCopyPlan(source, source.messages, runStore, runtimeEventStore), + copiedMessages: source.messages, + referenceMap: { + mode: 'exact', + linkedChildren: { mode: 'reject' }, + sourceSessionId: 'session-source', + targetSessionId: 'session-missing-artifact', + artifactIds: new Map([['artifact-source', 'artifact-target']]), + relativePaths: new Map(), + }, + runStore, + runtimeEventStore, + newId: () => crypto.randomUUID(), + }); + const reclaimedResult = withReclaimed.copiedMessages.find( + (message) => message.type === 'tool_result' && message.content.kind === 'subagent', ); assert.deepEqual( - await runtimeEventStore.listSessionInvocations('session-missing-artifact'), + reclaimedResult?.type === 'tool_result' && reclaimedResult.content.kind === 'subagent' + ? reclaimedResult.content.artifactIds + : undefined, [], ); + assert.equal( + (await runtimeEventStore.listSessionInvocations('session-missing-artifact')).length, + 1, + ); // A copied run and its copied invocation share one fresh identity, so the // copy mints one id here rather than two. const ids = [ diff --git a/packages/runtime/src/conversation-copy.ts b/packages/runtime/src/conversation-copy.ts index 71ee9d92da..9edd63ffcc 100644 --- a/packages/runtime/src/conversation-copy.ts +++ b/packages/runtime/src/conversation-copy.ts @@ -1115,20 +1115,30 @@ function rewriteOwnedArtifactId( } /** - * The `captureArtifactId` join, or nothing when its Artifact is gone. + * A reference whose target may have been reclaimed, mapped or dropped. * - * Every other Artifact reference throws on a missing target, because the bytes - * and the events naming them have always been removed together. This one is the - * exception: the capture Artifacts are reclaimed from disk on their own, while - * the attempts pointing at them stay in an append-only ledger. So a copy drops - * the key instead of failing on it. + * An Artifact reference normally throws on a missing target, because the bytes + * and the record naming them are removed together and a copy that lost one has + * lost something a reader will ask for. These references are the exception: + * they live in an append-only ledger that outlives what it names, and the + * retired provider-request captures are reclaimed from disk on their own. A + * copy carries what is still there and drops the rest, because failing would + * make a whole Session uncopyable over a byte nothing reads. */ +function reclaimableArtifactReference( + sourceArtifactId: string, + references: ConversationCopyArtifactReferenceMap, +): string | undefined { + if (references.mode === 'preserve_external') return sourceArtifactId; + return references.artifactIds.get(sourceArtifactId); +} + +/** The `captureArtifactId` join, or nothing when its Artifact is gone. */ function capturedArtifactJoin( sourceArtifactId: string, references: ConversationCopyArtifactReferenceMap, ): { captureArtifactId?: string } { - if (references.mode === 'preserve_external') return { captureArtifactId: sourceArtifactId }; - const targetArtifactId = references.artifactIds.get(sourceArtifactId); + const targetArtifactId = reclaimableArtifactReference(sourceArtifactId, references); return targetArtifactId === undefined ? {} : { captureArtifactId: targetArtifactId }; } @@ -1598,7 +1608,10 @@ function rewriteArtifactIds( artifactIds: readonly string[], references: ConversationCopyArtifactReferenceMap, ): readonly string[] { - return artifactIds.map((artifactId) => rewriteOwnedArtifactId(artifactId, references)); + return artifactIds.flatMap((artifactId) => { + const targetArtifactId = reclaimableArtifactReference(artifactId, references); + return targetArtifactId === undefined ? [] : [targetArtifactId]; + }); } function validatedExternalChildReferences( @@ -1628,9 +1641,10 @@ function rewriteSnapshotArtifactIds( if (references.mode !== 'exact' || references.linkedChildren.mode !== 'snapshot') { return artifactIds; } - return artifactIds.map((artifactId) => - requiredMappedId(references.artifactIds, artifactId, 'linked Artifact'), - ); + return artifactIds.flatMap((artifactId) => { + const targetArtifactId = references.artifactIds.get(artifactId); + return targetArtifactId === undefined ? [] : [targetArtifactId]; + }); } function rewriteArchivedSnapshot( diff --git a/packages/storage/src/__tests__/artifact-store.test.ts b/packages/storage/src/__tests__/artifact-store.test.ts index f3df5577c0..15e6eb4334 100644 --- a/packages/storage/src/__tests__/artifact-store.test.ts +++ b/packages/storage/src/__tests__/artifact-store.test.ts @@ -392,17 +392,20 @@ describe('SQLite Artifact store', () => { ['fixture'], ); - // A caller that names a capture outright is asking for a record that is - // on its way off disk, and the copy refuses rather than carrying it. - await assert.rejects( - store.copyConversationArtifacts({ - sourceSessionId: 'session-1', - targetSessionId: 'session-copy-2', - turnIds: ['turn-1'], - linkedArtifacts: [{ sessionId: 'session-child', artifactIds: ['linked-capture'] }], - }), - /Linked Artifact linked-capture could not be copied/, - ); + // A child result lists every Artifact its turn held, captures included, + // and the ledger holding that list cannot be rewritten. So a request + // naming one is answered with what is still copyable, not refused: the + // alternative is a Session that can never be forked again. + const named = await store.copyConversationArtifacts({ + sourceSessionId: 'session-1', + targetSessionId: 'session-copy-2', + turnIds: ['turn-1'], + linkedArtifacts: [ + { sessionId: 'session-child', artifactIds: ['linked-capture', linkedKept.id] }, + ], + }); + assert.equal(named.artifactIds.get('linked-capture'), undefined); + assert.ok(named.artifactIds.get(linkedKept.id)); }); }); diff --git a/packages/storage/src/artifact-store.ts b/packages/storage/src/artifact-store.ts index 2a4d584b7f..174d12a49e 100644 --- a/packages/storage/src/artifact-store.ts +++ b/packages/storage/src/artifact-store.ts @@ -454,8 +454,12 @@ class SqliteArtifactStore implements ArtifactAuthorityStore { candidate.status !== 'deleted' && !isRetiredCapture(candidate), ); - if (!record) throw new Error(`Linked Artifact ${artifactId} could not be copied`); - selected.push({ ...record }); + // A linked child result names every Artifact its turn held, and the + // ledger naming them cannot be rewritten. One that is no longer + // there is copied as nothing rather than failing the copy -- the + // caller is asking for what a past turn had, not asserting that all + // of it survived. + if (record) selected.push({ ...record }); } } const selectedIds = new Set(selected.map((record) => record.id));