fix(storage): reclaim the bytes the v1 upgrade orphaned

packages/runtime-host/src/server/execution-composition.ts

@@ -1755,6 +1755,15 @@ export async function createExecutionRuntimeHostComposition(
         recovery: {
           state: async () => {
             await skills.recover();
+            try {
+              await openedArtifactStore.reclaimUpgradeResidue();
+            } catch (error) {
+              // Leftover bytes are not worth refusing to start over; the next
+              // start tries again.
+              console.error(
+                `[runtime-host] upgrade residue could not be reclaimed: ${generalizedErrorMessage(error)}`,
+              );
+            }
           },
         },
         drain: [

packages/storage/src/__tests__/artifact-stores.test.ts

@@ -135,6 +135,114 @@ describe('interactive artifact store authority', () => {
     });
   });
 
+  test('reclaims the bytes the v1 upgrade orphaned, including a user-deleted upload', async () => {
+    await withInteractiveOwner(async (owner, root, track) => {
+      const initial = await openInteractiveArtifactStoreForWrite(owner.lease);
+      initial.close();
+      const db = new DatabaseSync(join(root, 'runtime.sqlite'));
+      db.exec(`
+        DROP TABLE artifact_records;
+        CREATE TABLE artifact_records (
+          storage_key TEXT PRIMARY KEY, artifact_id TEXT NOT NULL,
+          session_id TEXT NOT NULL, created_at INTEGER NOT NULL CHECK(created_at >= 0),
+          status TEXT NOT NULL CHECK(status IN ('live', 'deleted')),
+          relative_path TEXT NOT NULL, record_json TEXT NOT NULL
+        );
+        CREATE UNIQUE INDEX artifact_records_relative_path ON artifact_records(relative_path);
+        UPDATE operational_schema_migrations SET version = 1 WHERE scope = 'artifact';
+      `);
+      // One row per reason the upgrade drops one, each with bytes on disk.
+      const rows = [
+        { id: 'live', name: 'quarterly numbers.csv', status: 'live', source: 'user_upload' },
+        { id: 'erased', name: 'passport scan.pdf', status: 'deleted', source: 'user_upload' },
+        {
+          id: 'retired',
+          name: 'provider-request-step-4-cap.json',
+          status: 'live',
+          source: 'provider_request_capture',
+        },
+        { id: 'sourceless', name: 'recap-request.json', status: 'live', source: undefined },
+        { id: 'broken', name: 'unreadable.txt', status: 'live', source: 'tool_result' },
+        { id: 'mismatched', name: 'inconsistent.txt', status: 'live', source: 'tool_result' },
+        // Sorts first, and a directory cannot be unlinked, so it stands in for
+        // any leftover the store cannot remove.
+        { id: 'aborted', name: 'stuck', status: 'live', source: 'provider_request_capture' },
+      ];
+      await mkdir(join(root, 'artifacts', 'session-1'), { recursive: true });
+      for (const row of rows) {
+        const relativePath = `session-1/${row.id}-${row.name}`;
+        if (row.id === 'aborted') await mkdir(join(root, 'artifacts', relativePath));
+        else await writeFile(join(root, 'artifacts', relativePath), `bytes of ${row.id}`);
+        db.prepare('INSERT INTO artifact_records VALUES (?, ?, ?, ?, ?, ?, ?)').run(
+          row.id,
+          row.id,
+          'session-1',
+          1,
+          row.status,
+          relativePath,
+          row.id === 'broken'
+            ? '{'
+            : JSON.stringify({
+                id: row.id === 'mismatched' ? 'other-id' : row.id,
+                sessionId: 'session-1',
+                turnId: 'turn-1',
+                createdAt: 1,
+                name: row.name,
+                kind: 'file',
+                sizeBytes: `bytes of ${row.id}`.length,
+                relativePath,
+                ...(row.source ? { source: row.source } : {}),
+                status: row.status,
+              }),
+        );
+      }
+      db.close();
+
+      const store = track(await openInteractiveArtifactStoreForWrite(owner.lease));
+      // Re-creating a dropped record's exact path before the reclamation runs
+      // must keep the new bytes, not honour the note.
+      await store.create({
+        id: 'erased',
+        sessionId: 'session-1',
+        turnId: 'turn-2',
+        name: 'passport scan.pdf',
+        kind: 'file',
+        content: 'uploaded again',
+        source: 'user_upload',
+      });
+      await store.reclaimUpgradeResidue();
+      await store.reclaimUpgradeResidue();
+
+      const path = (id: string) =>
+        join(root, 'artifacts', `session-1/${id}-${rows.find((row) => row.id === id)!.name}`);
+      for (const id of ['retired', 'sourceless', 'broken', 'mismatched']) {
+        await assert.rejects(() => stat(path(id)), { code: 'ENOENT' });
+      }
+      assert.deepEqual(await store.readTextInSession('session-1', 'live'), {
+        ok: true,
+        text: 'bytes of live',
+      });
+      assert.deepEqual(await store.readTextInSession('session-1', 'erased'), {
+        ok: true,
+        text: 'uploaded again',
+      });
+      assert.equal((await stat(path('aborted'))).isDirectory(), true);
+      store.close();
+
+      // Everything behind the one that would not go was still reclaimed, and
+      // only its own note survives for a later attempt.
+      const remaining = new DatabaseSync(join(root, 'runtime.sqlite'), { readOnly: true });
+      assert.deepEqual(
+        remaining
+          .prepare('SELECT relative_path FROM artifact_upgrade_orphan_paths')
+          .all()
+          .map((row) => (row as { relative_path: string }).relative_path),
+        ['session-1/aborted-stuck'],
+      );
+      remaining.close();
+    });
+  });
+
   test('requires authentic leases and writer facades', async () => {
     await assert.rejects(
       () =>

packages/storage/src/artifact-store.ts

@@ -182,6 +182,7 @@ export interface ArtifactAuthorityStore extends DurableArtifactAttachmentReader
     input: ConversationArtifactCopyInput,
   ): Promise<ConversationArtifactCopyResult>;
   purgeSessionArtifacts(sessionId: string): Promise<void>;
+  reclaimUpgradeResidue(): Promise<void>;
   deleteOwnedArtifactInSession(
     sessionId: string,
     artifactId: string,
@@ -473,6 +474,43 @@ class SqliteArtifactStore implements ArtifactAuthorityStore {
     });
   }
 
+  /**
+   * Deletes the files the v1 upgrade recorded as no longer named by any record.
+   *
+   * A path some record has since claimed keeps its bytes. A note is discharged
+   * once its file is gone, and a file that will not go keeps only its own note
+   * rather than holding up the ones behind it.
+   */
+  async reclaimUpgradeResidue(): Promise<void> {
+    await this.enqueueMutation(async () => {
+      await this.prepareMutationUnlocked();
+      const recorded = this.metadataRepository.readUpgradeOrphanPaths();
+      if (recorded.length === 0) return;
+      const claimed = new Set(this.records.map((record) => record.relativePath));
+      const directories = new Set<string>();
+      const discharged: string[] = [];
+      try {
+        for (const relativePath of recorded) {
+          if (claimed.has(relativePath) || !isSafeRelativeArtifactPath(relativePath)) {
+            discharged.push(relativePath);
+            continue;
+          }
+          const target = join(this.artifactRoot, relativePath);
+          try {
+            await unlink(target);
+            directories.add(dirname(target));
+          } catch (error) {
+            if (!isNotFound(error)) continue;
+          }
+          discharged.push(relativePath);
+        }
+      } finally {
+        for (const directory of directories) await syncDirectory(directory);
+      }
+      if (discharged.length > 0) this.metadataRepository.forgetUpgradeOrphanPaths(discharged);
+    });
+  }
+
   private async replayExistingArtifactUnlocked(
     existing: ArtifactRecord,
     input: CreateArtifactInput,
@@ -1088,10 +1126,6 @@ function assertArtifactTurnKey(value: unknown): asserts value is string {
 const ARTIFACT_KIND_SET = new Set<ArtifactKind>(ARTIFACT_KINDS);
 const ARTIFACT_SOURCE_SET = new Set<ArtifactSource>(ARTIFACT_SOURCES);
 
-function isRecord(value: unknown): value is Record<string, unknown> {
-  return typeof value === 'object' && value !== null && !Array.isArray(value);
-}
-
 async function assertArtifactDirectory(artifactRoot: string, directory: string): Promise<void> {
   const root = await ensureRealDirectory(artifactRoot);
   const resolvedDirectory = await realpath(directory);

packages/storage/src/artifact-stores.ts

@@ -72,6 +72,7 @@ export interface InteractiveArtifactStoreWriter extends DurableArtifactAttachmen
     input: ConversationArtifactCopyInput,
   ): Promise<ConversationArtifactCopyResult>;
   purgeSessionArtifacts(sessionId: string): Promise<void>;
+  reclaimUpgradeResidue(): Promise<void>;
   listPage: ArtifactAuthorityStore['listPage'];
   listTurnArtifacts: ArtifactAuthorityStore['listTurnArtifacts'];
   getInSession: ArtifactAuthorityStore['getInSession'];
@@ -177,6 +178,7 @@ function createWriterFacade(
       return run(() => store.copyConversationArtifacts(acceptedInput));
     },
     purgeSessionArtifacts: (sessionId) => run(() => store.purgeSessionArtifacts(sessionId)),
+    reclaimUpgradeResidue: () => run(() => store.reclaimUpgradeResidue()),
     deleteUserArtifactInSession: (sessionId, artifactId) =>
       run(() => store.deleteUserArtifactInSession(sessionId, artifactId)),
     close: () => {

packages/storage/src/sqlite-artifact-metadata.ts

@@ -92,6 +92,24 @@ class SqliteArtifactMetadataRepository {
     });
   }
 
+  readUpgradeOrphanPaths(): string[] {
+    this.assertOpen();
+    const rows = this.#lease.database
+      .prepare('SELECT relative_path FROM artifact_upgrade_orphan_paths ORDER BY relative_path')
+      .all() as Array<{ relative_path: string }>;
+    return rows.map((row) => row.relative_path);
+  }
+
+  forgetUpgradeOrphanPaths(relativePaths: readonly string[]): void {
+    this.assertOpen();
+    this.#lease.transaction('write', () => {
+      const forget = this.#lease.database.prepare(
+        'DELETE FROM artifact_upgrade_orphan_paths WHERE relative_path = ?',
+      );
+      for (const relativePath of relativePaths) forget.run(relativePath);
+    });
+  }
+
   close(): void {
     if (this.#closed) return;
     this.#closed = true;

packages/storage/src/sqlite-artifact-schema.ts

@@ -18,7 +18,10 @@
  */
 
 import type { DatabaseSync } from 'node:sqlite';
-import { decodeArtifactRecordJsons } from './artifact-metadata-codec.js';
+import {
+  decodeArtifactRecordJsons,
+  isSafeRelativeArtifactPath,
+} from './artifact-metadata-codec.js';
 
 export const SQLITE_ARTIFACT_SCHEMA_VERSION = 3;
 
@@ -27,22 +30,35 @@ export function migrateSqliteArtifactDatabase(db: DatabaseSync): void {
     name?: unknown;
   }>;
   const retained: string[] = [];
-  if (columns.some(({ name }) => name === 'status' || name === 'storage_key')) {
+  // Every path the old table named. Whatever is not carried over is a file no
+  // catalog will name again, and this is the last moment anything knows it is
+  // there. Unlinking here is not an option: a rollback after one would be
+  // unrecoverable, so the paths are recorded for the store to reclaim later.
+  const scanned: string[] = [];
+  const hasStatusColumn = columns.some(({ name }) => name === 'status');
+  if (hasStatusColumn || columns.some(({ name }) => name === 'storage_key')) {
     const rows = db.prepare('SELECT * FROM artifact_records').all();
     for (const row of rows) {
-      if (columns.some(({ name }) => name === 'status') && row.status !== 'live') continue;
+      if (typeof row.relative_path === 'string' && isSafeRelativeArtifactPath(row.relative_path)) {
+        scanned.push(row.relative_path);
+      }
       try {
         const parsed = JSON.parse(String(row.record_json));
         if (!parsed || typeof parsed !== 'object' || Array.isArray(parsed)) continue;
-        if (parsed.status !== undefined && parsed.status !== 'live') continue;
-        delete parsed.status;
         if (
           parsed.id !== row.artifact_id ||
           parsed.sessionId !== row.session_id ||
           parsed.createdAt !== row.created_at ||
           parsed.relativePath !== row.relative_path
         )
           continue;
+        if (
+          [hasStatusColumn ? row.status : undefined, parsed.status].some(
+            (value) => value !== undefined && value !== null && value !== 'live',
+          )
+        )
+          continue;
+        delete parsed.status;
         retained.push(JSON.stringify(parsed));
       } catch {}
     }
@@ -62,11 +78,22 @@ export function migrateSqliteArtifactDatabase(db: DatabaseSync): void {
 
     CREATE UNIQUE INDEX IF NOT EXISTS artifact_records_relative_path
       ON artifact_records(relative_path);
+
+    CREATE TABLE IF NOT EXISTS artifact_upgrade_orphan_paths (
+      relative_path TEXT PRIMARY KEY
+    );
+  `);
+  const carried = decodeArtifactRecordJsons(retained);
+  const kept = new Set(carried.map((record) => record.relativePath));
+  const orphan = db.prepare(`
+    INSERT INTO artifact_upgrade_orphan_paths VALUES (?)
+    ON CONFLICT(relative_path) DO NOTHING
   `);
+  for (const relativePath of scanned) if (!kept.has(relativePath)) orphan.run(relativePath);
   const insert = db.prepare(`
     INSERT INTO artifact_records VALUES (?, ?, ?, ?, ?)
   `);
-  for (const record of decodeArtifactRecordJsons(retained)) {
+  for (const record of carried) {
     insert.run(
       record.id,
       record.sessionId,