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/execution-composition.ts b/packages/runtime-host/src/server/execution-composition.ts index bb9145e053..2e38c38f8d 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,7 @@ export async function createExecutionRuntimeHostComposition( `[runtime-host] optional context-offload Store could not be opened: ${generalizedErrorMessage(storage.contextOffloadUnavailable.cause)}`, ); } + let stopRetiredCaptureSweep: (() => void) | undefined; const stores = storage.execution; let graphControlStore: ReturnType | undefined; let graphClient: HostAgentGraphCoordinator | undefined; @@ -1759,6 +1763,21 @@ 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: 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(); + }, + }); }, }, drain: [ @@ -1775,6 +1794,7 @@ export async function createExecutionRuntimeHostComposition( () => { unsubscribeTranscriptChanges?.(); unsubscribeUsageChanges?.(); + stopRetiredCaptureSweep?.(); }, ], releaseConnection: [(connectionId) => artifacts.releaseConnection(connectionId)], 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 38821c8462..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]!, { @@ -1602,6 +1617,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 { @@ -1989,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 fc158e67a8..9edd63ffcc 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,34 @@ function rewriteOwnedArtifactId( return rewriteOwnedId(sourceArtifactId, references.artifactIds, 'Artifact'); } +/** + * A reference whose target may have been reclaimed, mapped or dropped. + * + * 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 } { + const targetArtifactId = reclaimableArtifactReference(sourceArtifactId, references); + return targetArtifactId === undefined ? {} : { captureArtifactId: targetArtifactId }; +} + function rewriteOwnedId(sourceId: string, ids: ReadonlyMap, kind: string): string { return requiredMappedId(ids, sourceId, kind); } @@ -1631,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( @@ -1661,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 31c8c759ae..15e6eb4334 100644 --- a/packages/storage/src/__tests__/artifact-store.test.ts +++ b/packages/storage/src/__tests__/artifact-store.test.ts @@ -322,6 +322,124 @@ 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('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 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)); + }); + }); + + 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..c50bdab1b5 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,200 @@ 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. + 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, '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('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( + { + purgeRetiredCaptures: async () => { + calls += 1; + throw new Error('artifact store is unavailable'); + }, + }, + { + onError: (error) => { + errors.push(error); + }, + }, + ); + + await settled(() => errors.length === 5); + await idleLongerThanOnePause(); + assert.equal(calls, 5, 'a permanent failure does not retry forever'); + }); + + 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. + // 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); + await new Promise((resolve) => { + setTimeout(resolve, 400); + }); + residue -= limit; + return { purged: limit, remaining: Math.max(0, residue) }; + }, + }); + + await settled(() => limits.length >= 2); + assert.deepEqual(limits.slice(0, 2), [256, 256]); + }); + + 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): Promise { + const deadline = Date.now() + 15_000; + while (Date.now() < deadline) { + if (await 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..174d12a49e 100644 --- a/packages/storage/src/artifact-store.ts +++ b/packages/storage/src/artifact-store.ts @@ -85,6 +85,24 @@ 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'; + +/** + * 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[]; @@ -219,6 +237,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, @@ -421,7 +441,8 @@ class SqliteArtifactStore implements ArtifactAuthorityStore { (record) => record.sessionId === input.sourceSessionId && turnIds.has(record.turnId) && - !excludedArtifactIds.has(record.id), + !excludedArtifactIds.has(record.id) && + !isRetiredCapture(record), ) .map((record) => ({ ...record })); for (const [sessionId, artifactIds] of requestedLinkedArtifactIds) { @@ -430,10 +451,15 @@ 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 }); + // 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)); @@ -442,7 +468,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); @@ -882,19 +909,47 @@ 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(isRetiredCapture); + const batch = retired.slice(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..2a8ad31455 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,87 @@ function createWriterFacade( return Object.freeze(facade); } +/** + * How much one batch deletes -- as much as it may, not as little. + * + * 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_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 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. + */ +export function startRetiredCaptureSweep( + artifacts: Pick, + options: { readonly onError?: (error: unknown) => void | Promise } = {}, +): () => void { + let stopped = false; + void (async () => { + let failures = 0; + while (!stopped) { + let pauseMs: number; + try { + const startedAt = Date.now(); + const { remaining } = await artifacts.purgeRetiredCaptures(RETIRED_CAPTURE_SWEEP_BATCH); + const batchMs = Date.now() - startedAt; + if (remaining === 0) return; + failures = 0; + pauseMs = Math.max( + RETIRED_CAPTURE_SWEEP_PAUSE_MS, + batchMs * RETIRED_CAPTURE_SWEEP_DUTY_DIVISOR, + ); + } catch (error) { + 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; + } + await new Promise((resolve) => { + setTimeout(resolve, pauseMs).unref(); + }); + } + })(); + return () => { + stopped = true; + }; +} + function snapshotCreateInput(input: CreateArtifactInput): CreateArtifactInput { return Object.freeze({ ...input,