feat(storage): reclaim the prepared-request captures already on disk

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] });

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<typeof createAgentGraphControlStore> | 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)],

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');

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 = [