fix(import): claim assets only once their conversation is written, and stop the pre-scan on cancel

A conversation is buffered when it is converted; the flush is what writes
it. Claiming its assets at buffer time meant a flush that rejected left
those files behind, referenced by a conversation that never landed and
skipped by the cleanup. Claims are held pending and promoted only when a
flush actually commits, so maybeFlush now reports whether it ran.

The pre-scan also never consulted the cancel flag, so a cancelled job kept
inflating and parsing every remaining shard after the user was told it had
stopped.

api/server/utils/import/importBatchBuilder.js

@@ -183,13 +183,16 @@ class ImportBatchBuilder {
    * Flushes the buffered batch once the number of buffered conversations
    * reaches flushThreshold. Intended to be called periodically while importing
    * to bound peak memory and Mongo op size.
-   * @returns {Promise<void>} A promise that resolves once any triggered flush completes.
+   * @returns {Promise<boolean>} Whether a flush actually ran. Callers that
+   *   promote bookkeeping on commit (the importer's asset claims) need to know
+   *   the difference between "buffered" and "written".
    */
   async maybeFlush() {
     if (this.conversations.length < this.flushThreshold) {
-      return;
+      return false;
     }
     await this.flush();
+    return true;
   }
 
   /**

packages/api/src/import/service.spec.ts

@@ -26,7 +26,7 @@ function recorder(): { sink: Parameters<typeof runImport>[0]['batch']; recorded:
       finishConversation: (title, _createdAt, convo, model) => {
         recorded.conversations.push({ title, convo, model });
       },
-      maybeFlush: async () => undefined,
+      maybeFlush: async () => false,
       saveBatch: async () => undefined,
     },
   };
@@ -207,6 +207,46 @@ describe('runImport', () => {
     expect(recorded.conversations).toEqual([]);
   });
 
+  /** A conversation is only buffered when it is converted; the flush is what
+   * writes it. Claiming its assets at buffer time meant a flush that rejected
+   * left those files behind, referenced by a conversation that was never
+   * written and skipped by the cleanup. */
+  it('releases assets whose conversations were buffered but never flushed', async () => {
+    const filepath = await buildFixtureExport();
+    const deleted: string[] = [];
+    const recorded: string[] = [];
+
+    await expect(
+      runImport({
+        filepath,
+        userId: 'u1',
+        defaultModel: 'gpt-4o',
+        deps: {
+          ...DEPS,
+          deleteFile: async (asset: { file_id: string }) => {
+            deleted.push(asset.file_id);
+          },
+        },
+        batch: {
+          startConversation: () => undefined,
+          saveMessage: () => undefined,
+          finishConversation: (title: string) => {
+            recorded.push(title);
+          },
+          maybeFlush: async () => false,
+          saveBatch: async () => {
+            throw new Error('mongo unavailable');
+          },
+        },
+        existingExternalIds: new Set(),
+      }),
+    ).rejects.toThrow('mongo unavailable');
+
+    expect(recorded.length).toBeGreaterThan(0);
+    /** Every asset the fixture ships, since no conversation was committed. */
+    expect(deleted).toHaveLength(3);
+  });
+
   it('reports progress as it advances', async () => {
     const filepath = await buildFixtureExport();
     const { sink } = recorder();
@@ -372,17 +412,64 @@ describe('runImport', () => {
       deps: DEPS,
       batch: sink,
       existingExternalIds: new Set(),
+      /** Keyed off what has actually been written rather than a probe count,
+       * so it keeps meaning "cancelled after the first conversation" however
+       * many times the run checks along the way. */
       isCancelled: async () => {
         checks += 1;
-        return checks > 1;
+        return recorded.conversations.length >= 1;
       },
     });
 
+    expect(checks).toBeGreaterThan(0);
     expect(recorded.conversations).toHaveLength(1);
     expect(report.imported).toBe(1);
     expect(report.skipped).toBe(0);
   });
 
+  /** The scan runs before any phase is announced and spends real time
+   * inflating and parsing a large sharded export. Without a check here the job
+   * reports cancelled while the process works on through every shard. */
+  it('abandons the pre-scan when the job is already cancelled', async () => {
+    const filepath = await writeZip({
+      'conversations-000.json': JSON.stringify([
+        textConversation('ext-first', 'First convo', 1700005000),
+      ]),
+      'export_manifest.json': shardedManifest(['conversations-000.json']),
+    });
+    const { sink, recorded } = recorder();
+    const reads: string[] = [];
+    const realOpen = archiveModule.openArchive;
+    jest.spyOn(archiveModule, 'openArchive').mockImplementation(async (path, options) => {
+      const archive = await realOpen(path, options);
+      return {
+        ...archive,
+        read: async (name: string) => {
+          reads.push(name);
+          return archive.read(name);
+        },
+      };
+    });
+
+    try {
+      const report = await runImport({
+        filepath,
+        userId: 'u1',
+        defaultModel: 'gpt-4o',
+        deps: DEPS,
+        batch: sink,
+        existingExternalIds: new Set(),
+        isCancelled: async () => true,
+      });
+
+      expect(recorded.conversations).toEqual([]);
+      expect(report.imported).toBe(0);
+      expect(reads).not.toContain('conversations-000.json');
+    } finally {
+      jest.restoreAllMocks();
+    }
+  });
+
   it('records a shard parse failure and still imports the other shard', async () => {
     const filepath = await writeZip({
       'conversations-000.json': 'not valid json{',

packages/api/src/import/service.ts

@@ -36,6 +36,9 @@ interface ExportScan {
   pointers: string[];
   attachments: Map<string, ChatGptAttachment>;
   references: Map<string, AssetReference>;
+  /** Set when the scan stopped early because the job was cancelled, so the run
+   * does not go on to ingest assets for a partial pointer list. */
+  cancelled?: boolean;
   /** The format the first shard that parsed turned out to be. A Claude or Grok
    * export aborts the scan immediately — neither has assets a conversation can
    * resolve, so nothing this pass collects applies to them. */
@@ -116,6 +119,7 @@ async function scanExport(
   shards: string[],
   errors: string[],
   existingExternalIds: ReadonlySet<string>,
+  isCancelled?: () => Promise<boolean>,
 ): Promise<ExportScan> {
   const scan: ExportScan = {
     shards: [],
@@ -129,6 +133,14 @@ async function scanExport(
   let detected = false;
 
   for (const shard of shards) {
+    /** The scan runs before any phase is announced, and a large sharded export
+     * spends real time here inflating and parsing. Without this the job reports
+     * cancelled while the process works on through every remaining shard. */
+    if (isCancelled && (await isCancelled())) {
+      scan.cancelled = true;
+      return scan;
+    }
+
     try {
       const parsed = await readShardJson(archive, shard);
 
@@ -181,7 +193,7 @@ function importConversation(
   conv: ChatGptConversation,
   input: RunImportInput,
   assets: Map<string, ImportedAsset>,
-  usedPointers: Set<string>,
+  pendingPointers: Set<string>,
 ): number {
   const converted = convertConversation(conv, {
     userId: input.userId,
@@ -205,11 +217,13 @@ function importConversation(
     converted.model,
   );
 
-  /** Recorded only once the conversation is buffered, so an asset whose
-   * conversation never made it is left unclaimed and gets cleaned up. */
+  /** Held pending, not claimed. The conversation is only buffered at this
+   * point; a flush that rejects loses it, and an asset claimed here would then
+   * survive the cleanup with nothing referencing it. `commitPending` promotes
+   * these once a flush actually lands. */
   for (const message of converted.messages) {
     for (const pointer of message.assetPointers) {
-      usedPointers.add(pointer);
+      pendingPointers.add(pointer);
     }
   }
 
@@ -254,7 +268,18 @@ export async function runImport(input: RunImportInput): Promise<ImportReport> {
    * claimed. Declared out here so the `finally` can release the difference on
    * every exit path — completed, cancelled, or thrown. */
   const ingested = new Map<string, ImportedAsset>();
+  /** Claimed: referenced by a conversation the sink has actually committed. */
   const usedPointers = new Set<string>();
+  /** Buffered but not yet flushed. Promoted on a successful flush, dropped if
+   * one rejects — the conversations it held were never written. */
+  const pendingPointers = new Set<string>();
+
+  const commitPending = (): void => {
+    for (const pointer of pendingPointers) {
+      usedPointers.add(pointer);
+    }
+    pendingPointers.clear();
+  };
 
   const report: ImportReport = {
     imported: 0,
@@ -300,8 +325,13 @@ export async function runImport(input: RunImportInput): Promise<ImportReport> {
       layout.conversationShards,
       report.errors,
       input.existingExternalIds,
+      input.isCancelled,
     );
 
+    if (scan.cancelled) {
+      return report;
+    }
+
     if (scan.format === 'claude') {
       await runClaudeImport(providerRun);
       return report;
@@ -383,7 +413,12 @@ export async function runImport(input: RunImportInput): Promise<ImportReport> {
         }
 
         try {
-          progress.messages.done += importConversation(conv, input, assetResult.map, usedPointers);
+          progress.messages.done += importConversation(
+            conv,
+            input,
+            assetResult.map,
+            pendingPointers,
+          );
           report.imported += 1;
           /** The skip set is a snapshot taken once at job start, so without
            * this a `conversation_id` appearing twice in one export — across
@@ -398,12 +433,15 @@ export async function runImport(input: RunImportInput): Promise<ImportReport> {
         }
 
         progress.conversations.done += 1;
-        await input.batch.maybeFlush();
+        if (await input.batch.maybeFlush()) {
+          commitPending();
+        }
         await input.onProgress?.(progress);
       }
     }
 
     await input.batch.saveBatch();
+    commitPending();
     return report;
   } catch (error) {
     logger.error('[import] Import run failed', error);

packages/api/src/import/sink.ts

@@ -38,7 +38,10 @@ export interface BatchSink {
     convo: ConversationOverrides,
     model: string,
   ): void;
-  maybeFlush(): Promise<void>;
+  /** Resolves `true` when a flush actually ran. "Buffered" and "written" are
+   * different states, and the importer's asset cleanup depends on the
+   * difference. */
+  maybeFlush(): Promise<boolean>;
   saveBatch(): Promise<void>;
 }