fix(import): serialize job mutations and freeze terminal jobs

The background run reads the cancel flag and writes progress in separate
round trips, so a DELETE landing between one of those reads and its write
was overwritten by the stale snapshot the read returned: the job went back
to active, the next isCancelled said so, and the import kept going.

Every mutation now goes through the per-key lock that only confirmStart
held, and a job that has reached a terminal status no longer accepts a
status or phase change - only the partial report describing what it wrote
before it stopped.

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

@@ -98,6 +98,76 @@ describe('ImportJobStore', () => {
     expect(owned?.updatedAt).toBe(job.updatedAt);
   });
 
+  describe('terminal jobs', () => {
+    it('ignores a phase update that lands after cancellation', async () => {
+      const job = await store.create({ userId: 'u1', filepath: '/tmp/a.zip', filename: 'a.zip' });
+      await store.cancel('u1', job.jobId);
+
+      const patched = await store.patch('u1', job.jobId, { phase: 'conversations' });
+
+      expect(patched?.phase).toBe('cancelled');
+      expect(patched?.status).toBe('cancelled');
+      expect(await store.isCancelled('u1', job.jobId)).toBe(true);
+    });
+
+    it('still records the partial report a cancelled run produced', async () => {
+      const job = await store.create({ userId: 'u1', filepath: '/tmp/a.zip', filename: 'a.zip' });
+      await store.cancel('u1', job.jobId);
+
+      const report = {
+        imported: 12,
+        skipped: 0,
+        assetsImported: 3,
+        assetsUnavailable: 0,
+        errors: [],
+      };
+      const patched = await store.patch('u1', job.jobId, { report });
+
+      expect(patched?.report).toEqual(report);
+      expect(patched?.phase).toBe('cancelled');
+    });
+
+    /** The background run reads the cancel flag and writes progress in two
+     * separate round trips. Without serialization the progress write lands on a
+     * job it read before the cancellation, resurrecting `status: 'active'` —
+     * and the very next `isCancelled` then tells the run to keep importing. */
+    it('does not let a progress update in flight during a cancellation resurrect the job', async () => {
+      const backing = new Keyv();
+      const read = backing.get.bind(backing);
+      /** Only the first read stalls, which is the progress update's: the
+       * cancellation then reads and writes entirely inside that window, so the
+       * progress write is left holding a job snapshot taken before it. */
+      let stall = true;
+      jest.spyOn(backing, 'get').mockImplementation(async (key: string) => {
+        const value = await read(key);
+        if (stall) {
+          stall = false;
+          await new Promise((resolve) => setTimeout(resolve, 20));
+        }
+        return value;
+      });
+      const slowStore = new ImportJobStore(backing, 60000);
+      const job = await slowStore.create({
+        userId: 'u1',
+        filepath: '/tmp/a.zip',
+        filename: 'a.zip',
+      });
+
+      const progress = slowStore.patch('u1', job.jobId, {
+        progress: {
+          conversations: { done: 5, total: 10 },
+          messages: { done: 40, total: 80 },
+          assets: { done: 0, total: 0 },
+        },
+      });
+      await new Promise((resolve) => setTimeout(resolve, 1));
+      await slowStore.cancel('u1', job.jobId);
+      await progress;
+
+      expect(await slowStore.isCancelled('u1', job.jobId)).toBe(true);
+    });
+  });
+
   describe('confirmStart', () => {
     it('returns not_found for a job that does not exist', async () => {
       expect(await store.confirmStart('u1', 'missing')).toEqual({ status: 'not_found' });

packages/api/src/import/job.ts

@@ -18,14 +18,38 @@ export type StartTransitionResult =
   | { status: 'not_found' }
   | { status: 'conflict'; job: ImportJob };
 
+/**
+ * Statuses a job never moves out of. Reaching one freezes `status` and
+ * `phase`: the background run's `isCancelled`/`onPhase`/`onProgress`
+ * callbacks each read and write separately, so a `DELETE` landing between
+ * one of those reads and its write would otherwise walk a cancelled job back
+ * into `phase: 'conversations'` — leaving the client polling forever a job
+ * whose run has already stopped. Every other field (`report`, `progress`)
+ * still applies, so a cancelled job still gains the partial report
+ * describing what was written before it stopped.
+ */
+const TERMINAL_STATUSES = new Set<ImportJob['status']>(['cancelled', 'completed', 'failed']);
+
+type ImportJobPatch = Omit<Partial<ImportJob>, 'userId' | 'jobId'>;
+
+function applyTerminalGuard(existing: ImportJob, patch: ImportJobPatch): ImportJobPatch {
+  if (!TERMINAL_STATUSES.has(existing.status)) {
+    return patch;
+  }
+  const { status: _status, phase: _phase, ...rest } = patch;
+  return rest;
+}
+
 export class ImportJobStore {
   private readonly store: Keyv;
   private readonly ttl: number;
-  /** Serializes `confirmStart` calls sharing the same job key. `Keyv`'s
+  /** Serializes every mutation sharing the same job key. `Keyv`'s
    * `get`/`set` pair is not itself atomic — without this, two racing
    * `/start` requests can both read `awaiting_confirmation` before either
-   * write lands, launching the background run twice over the same
-   * archive. Scoped to this process; see `confirmStart`. */
+   * write lands (launching the background run twice over the same archive),
+   * and a progress update that read an active job can land after a
+   * cancellation and resurrect it. Scoped to this process; see
+   * `confirmStart`. */
   private readonly transitionLocks = new Map<string, Promise<void>>();
 
   constructor(store: Keyv, ttl: number = DEFAULT_TTL) {
@@ -63,21 +87,34 @@ export class ImportJobStore {
     return job ?? null;
   }
 
-  async patch(
+  /** The read-modify-write itself, without the lock. Only ever called from
+   * inside `withTransitionLock`; re-entering the lock here would deadlock
+   * `confirmStart`, which already holds it. */
+  private async applyPatch(
     userId: string,
     jobId: string,
-    patch: Omit<Partial<ImportJob>, 'userId' | 'jobId'>,
+    patch: ImportJobPatch,
   ): Promise<ImportJob | null> {
     const existing = await this.get(userId, jobId);
     if (!existing) {
       return null;
     }
 
-    const updated: ImportJob = { ...existing, ...patch, updatedAt: Date.now() };
+    const updated: ImportJob = {
+      ...existing,
+      ...applyTerminalGuard(existing, patch),
+      updatedAt: Date.now(),
+    };
     await this.store.set(this.key(userId, jobId), updated, this.ttl);
     return updated;
   }
 
+  async patch(userId: string, jobId: string, patch: ImportJobPatch): Promise<ImportJob | null> {
+    return this.withTransitionLock(this.key(userId, jobId), () =>
+      this.applyPatch(userId, jobId, patch),
+    );
+  }
+
   async cancel(userId: string, jobId: string): Promise<boolean> {
     const updated = await this.patch(userId, jobId, {
       status: 'cancelled',
@@ -129,7 +166,7 @@ export class ImportJobStore {
       if (existing.phase !== 'awaiting_confirmation') {
         return { status: 'conflict', job: existing };
       }
-      const updated = await this.patch(userId, jobId, { phase: 'queued' });
+      const updated = await this.applyPatch(userId, jobId, { phase: 'queued' });
       if (!updated) {
         return { status: 'not_found' };
       }