diff --git a/packages/outpost/db/prisma/migrations/20260820120000_add_job_claim_fencing/migration.sql b/packages/outpost/db/prisma/migrations/20260820120000_add_job_claim_fencing/migration.sql
new file mode 100644
index 0000000..59d9db6
--- /dev/null
+++ b/packages/outpost/db/prisma/migrations/20260820120000_add_job_claim_fencing/migration.sql
@@ -0,0 +1,27 @@
+-- Fences writes to the worker execution that owns a claim, and records the
+-- deadline that execution was granted so reclaim has a single authority.
+--
+-- Additive and nullable with no backfill, which is deliberate: a NULL claimToken
+-- on an in-flight row means "claimed before this deployed", and the reclaim's
+-- legacy branch handles those on an absolute ceiling instead of a granted
+-- deadline.
+--
+-- On ordering: `apps/worker/start.sh` runs `migrate deploy` at container start,
+-- so this ships WITH the new code rather than ahead of it. There is no two-phase
+-- deploy here. What makes that safe is the additive-nullable shape, not any
+-- sequencing guarantee -- during the rollout an old replica's unfenced
+-- `prisma.job.update` writes race the new replica's fenced ones, and the old
+-- replica simply does not see these columns.
+--
+-- Takes ACCESS EXCLUSIVE on "Job" for the length of this transaction. That is
+-- brief for a nullable ADD COLUMN with no default (a catalog-only change in
+-- PG11+), but it does queue behind in-flight claim UPDATEs and block everything
+-- behind it while it waits. Deliberately no `lock_timeout`: a statement that
+-- times out here aborts the migration, and `migrate deploy` records the failure
+-- in `_prisma_migrations` with `finished_at = NULL`, which is the P3009 state
+-- `start.sh` exists to diagnose and which needs a manual `migrate resolve`.
+-- Waiting is recoverable; a half-recorded migration is not. Same trade `start.sh`
+-- makes when it leaves `migrate deploy` unbounded.
+ALTER TABLE "Job"
+ADD COLUMN "claimToken" TEXT,
+ADD COLUMN "lockUntil" TIMESTAMP(3);
diff --git a/packages/outpost/db/prisma/migrations/20260820120100_index_job_status_lockuntil/migration.sql b/packages/outpost/db/prisma/migrations/20260820120100_index_job_status_lockuntil/migration.sql
new file mode 100644
index 0000000..13241f5
--- /dev/null
+++ b/packages/outpost/db/prisma/migrations/20260820120100_index_job_status_lockuntil/migration.sql
@@ -0,0 +1,20 @@
+-- Index the columns the reclaim sweep scans.
+--
+-- The sweep runs once per poll interval on every replica:
+--   status = 'PROCESSING' AND "lockUntil" < NOW() - grace
+-- plus a NULL-"lockUntil" arm for rows claimed before the column existed. The
+-- first arm keeps "lockUntil" bare with the interval arithmetic on the other
+-- side, which is what lets the second column do any work; the legacy arms
+-- discriminate on "lockedAt"/"updatedAt" and ride the (status, "lockUntil" IS
+-- NULL) prefix, then filter. That is fine -- those arms empty out after one
+-- rollout. See reclaimStaleJobs() in packages/outpost/queue/src/worker.ts.
+--
+-- Its own migration on purpose. Prisma wraps each migration file in one
+-- transaction, so keeping this with the ADD COLUMN would hold that statement's
+-- ACCESS EXCLUSIVE lock across the index build and block reads for the duration.
+-- Split, the ALTER commits first and this takes only SHARE: writes wait, reads
+-- do not. Same reason CONCURRENTLY is not used -- it cannot run inside Prisma's
+-- transaction at all.
+--
+-- Purely additive: creates an index only, no column or data changes.
+CREATE INDEX "Job_status_lockUntil_idx" ON "Job"("status", "lockUntil");
diff --git a/packages/outpost/db/prisma/schema.prisma b/packages/outpost/db/prisma/schema.prisma
index 6a19ec1..9d6f212 100644
--- a/packages/outpost/db/prisma/schema.prisma
+++ b/packages/outpost/db/prisma/schema.prisma
@@ -241,6 +241,12 @@ model Job {
     progress    Int?      // Optional percentage 0-100
     runAt       DateTime  @default(now())
     lockedAt    DateTime?
+    // The deadline the CLAIMING worker was granted, written at claim time from
+    // that worker's own timeout config. Reclaim compares against this rather than
+    // re-deriving a window from whichever worker happens to notice, so a replica
+    // on an older config cannot decide a live claim has expired.
+    lockUntil   DateTime?
+    claimToken  String?   // Fences writes to the worker execution that owns this claim
     completedAt DateTime?
     error       String?
     createdAt   DateTime  @default(now())
@@ -248,6 +254,29 @@ model Job {
 
     @@index([status, runAt])
     @@index([type])
+    // Serves the reclaim sweep, which runs once per poll interval per replica.
+    // [status, runAt] has the wrong second column for it and [type] is far too
+    // low-selectivity to help.
+    //
+    // Only usable while the sweep's predicate keeps lockUntil bare on one side —
+    // see reclaimStaleJobs() in packages/outpost/queue/src/worker.ts. Wrapping
+    // the column in a CASE, or moving the interval arithmetic onto it, leaves the
+    // planner with the status prefix and a filter over every PROCESSING row.
+    //
+    // Worth stating what it does NOT buy, because the gain is smaller than it
+    // looks: PROCESSING is a tiny set (bounded by replicas × maxConcurrency, and
+    // JOB_CLEANUP keeps the table small besides), and [status, runAt]'s prefix
+    // already narrows to it. Meanwhile status and lockUntil both change on every
+    // claim and every release, so this index adds maintenance to the two hottest
+    // write paths on the table. It is here because the sweep's cost should not be
+    // a function of table size as the queue grows, not because it pays for itself
+    // at today's volumes.
+    //
+    // Ideally partial (WHERE status = 'PROCESSING'), which schema.prisma cannot
+    // express. Taking the wider index rather than hand-writing one, because a
+    // hand-written index puts the database permanently in drift against the CI
+    // gate that caught the missing claimToken field on this same branch.
+    @@index([status, lockUntil])
 }
 
 enum JobStatus {
diff --git a/packages/outpost/queue/src/__tests__/queue.test.ts b/packages/outpost/queue/src/__tests__/queue.test.ts
index befdd4d..4db6123 100644
--- a/packages/outpost/queue/src/__tests__/queue.test.ts
+++ b/packages/outpost/queue/src/__tests__/queue.test.ts
@@ -19,11 +19,13 @@ import type { JobResult, JobHandlerContext, WorkerHealthStatus } from '../types.
 const mockPrismaJob = {
     create: vi.fn(),
     update: vi.fn(),
+    updateMany: vi.fn(),
     findFirst: vi.fn(),
 };
 
 const mockPrisma = {
     job: mockPrismaJob,
+    $executeRaw: vi.fn(),
     $queryRaw: vi.fn(),
 };
 
@@ -38,6 +40,21 @@ vi.mock('@copilotkit/outpost/shared', () => ({
     calculateBackoff: (attempt: number) => 1000 * Math.pow(2, attempt),
 }));
 
+// The `shared` mock above re-declares MAX_JOB_ATTEMPTS, BACKOFF_BASE_MS and
+// BACKOFF_MAX_MS as literals. Every assertion written against those literals was
+// checking the mock against itself: editing the real constants left the whole
+// file green. Importing the real ones straight from source — past both the mock
+// and the package alias — gives the suite something honest to compare against.
+// `importActual` deliberately, not a relative path into shared/src: the queue
+// tsconfig sets rootDir to queue/src, so reaching across the package boundary by
+// path is a typecheck error. This goes through the same specifier the mock
+// intercepts, and gets the real module behind it.
+const realConstants = (await vi.importActual('@copilotkit/outpost/shared')) as {
+    MAX_JOB_ATTEMPTS: number;
+    BACKOFF_BASE_MS: number;
+    BACKOFF_MAX_MS: number;
+};
+
 // Import after mocks are set up
 const { createJob, updateJobProgress } = await import('../create-job.js');
 const { Worker } = await import('../worker.js');
@@ -52,6 +69,7 @@ function makeJobRow(
         payload: unknown;
         attempts: number;
         maxAttempts: number;
+        claimToken: string;
     }> = {},
 ) {
     return {
@@ -60,6 +78,7 @@ function makeJobRow(
         payload: overrides.payload ?? { ticketId: 'tkt-1', source: 'discord' },
         attempts: overrides.attempts ?? 0,
         maxAttempts: overrides.maxAttempts ?? 5,
+        claimToken: overrides.claimToken ?? 'claim-1',
     };
 }
 
@@ -83,7 +102,7 @@ describe('createJob', () => {
 
         const callArg = mockPrismaJob.create.mock.calls[0][0];
         expect(callArg.data.type).toBe('AI_RESPONSE');
-        expect(callArg.data.maxAttempts).toBe(5); // MAX_JOB_ATTEMPTS
+        expect(callArg.data.maxAttempts).toBe(realConstants.MAX_JOB_ATTEMPTS);
         expect(callArg.data.payload).toEqual({ ticketId: 'tkt-1', source: 'discord' });
         expect(callArg.data.runAt).toBeInstanceOf(Date);
     });
@@ -120,42 +139,73 @@ describe('updateJobProgress', () => {
     });
 
     it('updates progress clamped between 0 and 100', async () => {
-        mockPrismaJob.update.mockResolvedValue({});
+        mockPrismaJob.updateMany.mockResolvedValue({ count: 1 });
 
-        await updateJobProgress('job-1', 50);
-        expect(mockPrismaJob.update).toHaveBeenCalledWith({
-            where: { id: 'job-1' },
+        await updateJobProgress('job-1', 50, 'claim-1');
+        expect(mockPrismaJob.updateMany).toHaveBeenCalledWith({
+            where: { id: 'job-1', status: 'PROCESSING', claimToken: 'claim-1' },
             data: { progress: 50 },
         });
     });
 
     it('clamps progress above 100 to 100', async () => {
-        mockPrismaJob.update.mockResolvedValue({});
+        mockPrismaJob.updateMany.mockResolvedValue({ count: 1 });
 
-        await updateJobProgress('job-1', 150);
-        expect(mockPrismaJob.update).toHaveBeenCalledWith({
-            where: { id: 'job-1' },
+        await updateJobProgress('job-1', 150, 'claim-1');
+        expect(mockPrismaJob.updateMany).toHaveBeenCalledWith({
+            where: { id: 'job-1', status: 'PROCESSING', claimToken: 'claim-1' },
             data: { progress: 100 },
         });
     });
 
     it('clamps negative progress to 0', async () => {
-        mockPrismaJob.update.mockResolvedValue({});
+        mockPrismaJob.updateMany.mockResolvedValue({ count: 1 });
 
-        await updateJobProgress('job-1', -10);
-        expect(mockPrismaJob.update).toHaveBeenCalledWith({
-            where: { id: 'job-1' },
+        await updateJobProgress('job-1', -10, 'claim-1');
+        expect(mockPrismaJob.updateMany).toHaveBeenCalledWith({
+            where: { id: 'job-1', status: 'PROCESSING', claimToken: 'claim-1' },
             data: { progress: 0 },
         });
     });
 });
 
+describe('updateJobProgress resilience', () => {
+    beforeEach(() => {
+        vi.clearAllMocks();
+    });
+
+    it('cannot fail the job it is only describing', async () => {
+        const warn = vi.spyOn(console, 'warn').mockImplementation(() => {});
+        mockPrismaJob.updateMany.mockRejectedValue(new Error('pool exhausted'));
+
+        // Handlers `await` this. A rejection propagated into the handler, the
+        // worker recorded it as a job failure, and work that was running perfectly
+        // well got retried — repeating every side effect it had already produced.
+        // Progress is telemetry; it must never be able to fail the job.
+        await expect(updateJobProgress('job-1', 50, 'claim-1')).resolves.toBeUndefined();
+
+        expect(warn.mock.calls.map((c: unknown[]) => String(c[0])).join('\n')).toContain(
+            'failed and was ignored',
+        );
+        warn.mockRestore();
+    });
+});
+
 describe('Worker', () => {
     let worker: InstanceType<typeof Worker>;
 
     beforeEach(() => {
         vi.clearAllMocks();
         vi.useFakeTimers();
+        mockPrisma.$executeRaw.mockResolvedValue(0);
+        mockPrismaJob.updateMany.mockResolvedValue({ count: 1 });
+        // `clearAllMocks` clears call records but keeps implementations and any
+        // unconsumed `Once` queue, and nothing here re-armed `$queryRaw` — so a
+        // test that set no claim result silently inherited the previous test's,
+        // and `poll()` swallowed the resulting TypeError in its own catch. Reset
+        // and give it an explicit default; per-test `Once` values still win.
+        mockPrisma.$queryRaw.mockReset();
+        mockPrisma.$queryRaw.mockResolvedValue([]);
         worker = new Worker({
             pollIntervalMs: 100,
             maxConcurrency: 2,
@@ -186,14 +236,241 @@ describe('Worker', () => {
         await vi.advanceTimersByTimeAsync(0);
 
         expect(results).toEqual(['tkt-1']);
-        expect(mockPrismaJob.update).toHaveBeenCalledWith(
+        const claimSql = mockPrisma.$queryRaw.mock.calls[0][0].join(' ');
+        expect(claimSql).toContain('"claimToken" = gen_random_uuid()::text');
+        expect(claimSql).toContain('"maxAttempts", "claimToken"');
+        expect(mockPrismaJob.updateMany).toHaveBeenCalledWith(
             expect.objectContaining({
-                where: { id: 'job-1' },
+                where: expect.objectContaining({ id: 'job-1', claimToken: 'claim-1' }),
                 data: expect.objectContaining({ status: 'COMPLETED', attempts: 1 }),
             }),
         );
     });
 
+    it('reclaims on the deadline the claiming worker recorded, not its own config', async () => {
+        const now = new Date('2026-08-11T12:00:00.000Z');
+        vi.setSystemTime(now);
+
+        await worker.stop();
+        worker = new Worker({
+            pollIntervalMs: 100,
+            maxConcurrency: 2,
+            defaultTimeoutMs: 5000,
+            // Deliberately not 1000: that is BACKOFF_BASE_MS, and the assertion
+            // below is that this worker's config does NOT reach the predicate, so
+            // it has to be a value nothing else could have put there.
+            jobTimeouts: {
+                [JobType.AI_RESPONSE]: 7000,
+            },
+        });
+
+        const recoveredJob = makeJobRow();
+        mockPrisma.$executeRaw.mockResolvedValue(1);
+        mockPrisma.$queryRaw.mockResolvedValueOnce([recoveredJob]);
+        mockPrisma.$queryRaw.mockResolvedValue([]);
+        mockPrismaJob.update.mockResolvedValue({});
+
+        const handled: string[] = [];
+        worker.on(JobType.AI_RESPONSE, async (payload) => {
+            handled.push(payload.ticketId);
+            return { success: true };
+        });
+        worker.on(JobType.ESCALATION, async () => ({ success: true }));
+
+        worker.start();
+        await vi.advanceTimersByTimeAsync(0);
+
+        const firstReclaim = mockPrisma.$executeRaw.mock.calls[0];
+        const reclaimSql = firstReclaim[0].join(' ');
+        expect(reclaimSql).toContain("WHERE job.status = 'PROCESSING'");
+
+        // The whole point of B2: this worker's `jobTimeouts` must not appear in the
+        // predicate. A replica configured differently from the one that claimed the
+        // row would otherwise decide a live claim had expired, and the original
+        // handler's success would be fenced out and silently discarded.
+        expect(reclaimSql).toContain('job."lockUntil"');
+        expect(reclaimSql).not.toContain('jsonb_to_recordset');
+        expect(firstReclaim.slice(1)).not.toContain(7000);
+        expect(firstReclaim.slice(1)).not.toContain(5000);
+
+        // Rows claimed before `lockUntil` existed still need a way out, on an
+        // absolute ceiling rather than a guessed deadline.
+        expect(reclaimSql).toContain('job."lockedAt"');
+        expect(firstReclaim.slice(1)).toContain(900_000);
+
+        expect(handled).toEqual(['tkt-1']);
+    });
+
+    it("writes the claiming worker's own timeout onto the row it claims", async () => {
+        await worker.stop();
+        worker = new Worker({
+            pollIntervalMs: 100,
+            maxConcurrency: 2,
+            defaultTimeoutMs: 5000,
+            jobTimeouts: {
+                [JobType.AI_RESPONSE]: 120_000,
+            },
+        });
+
+        mockPrisma.$queryRaw.mockResolvedValue([]);
+        worker.on(JobType.AI_RESPONSE, async () => ({ success: true }));
+
+        worker.start();
+        await vi.advanceTimersByTimeAsync(0);
+
+        const claim = mockPrisma.$queryRaw.mock.calls[0];
+        const claimSql = claim[0].join(' ');
+        expect(claimSql).toContain('"lockUntil" = NOW()');
+        // Per-type, looked up by the row's own type, with the worker default as the
+        // fallback — so a type this worker has no entry for still gets a deadline.
+        expect(claimSql).toContain('->> "Job".type');
+        const timeoutMap = claim
+            .slice(1)
+            .find((v: unknown): v is string => typeof v === 'string' && v.includes('AI_RESPONSE'));
+        expect(JSON.parse(timeoutMap as string)).toEqual({
+            [JobType.AI_RESPONSE]: 120_000,
+        });
+        expect(claim.slice(1)).toContain(5000);
+    });
+
+    it('does not let an old execution clobber the reclaimed claim', async () => {
+        const persisted = {
+            id: 'job-1',
+            status: 'PROCESSING',
+            claimToken: 'claim-old',
+            attempts: 0,
+            progress: null as number | null,
+        };
+        mockPrismaJob.updateMany.mockImplementation(async ({ where, data }) => {
+            if (
+                where.id !== persisted.id ||
+                where.status !== persisted.status ||
+                where.claimToken !== persisted.claimToken
+            ) {
+                return { count: 0 };
+            }
+            Object.assign(persisted, data);
+            return { count: 1 };
+        });
+
+        let oldStarted!: () => void;
+        const oldIsRunning = new Promise<void>((resolve) => {
+            oldStarted = resolve;
+        });
+        let releaseOld!: () => void;
+        const oldMayFinish = new Promise<void>((resolve) => {
+            releaseOld = resolve;
+        });
+        worker.on(JobType.AI_RESPONSE, async (payload, context) => {
+            if (payload.ticketId === 'old-execution') {
+                oldStarted();
+                await oldMayFinish;
+                await context.reportProgress(25);
+            }
+            return { success: true };
+        });
+
+        type ClaimedJob = ReturnType<typeof makeJobRow>;
+        const processJob = (
+            worker as unknown as { processJob(job: ClaimedJob): Promise<void> }
+        ).processJob.bind(worker);
+        const oldExecution = processJob(
+            makeJobRow({
+                payload: { ticketId: 'old-execution', source: 'discord' },
+                claimToken: 'claim-old',
+            }),
+        );
+        await oldIsRunning;
+
+        // Model stale reclamation followed by a new exclusive claim.
+        persisted.claimToken = 'claim-new';
+        persisted.attempts = 1;
+        const newExecution = processJob(
+            makeJobRow({
+                payload: { ticketId: 'new-execution', source: 'discord' },
+                attempts: 1,
+                claimToken: 'claim-new',
+            }),
+        );
+        await newExecution;
+
+        releaseOld();
+        await oldExecution;
+
+        expect(persisted).toMatchObject({
+            status: 'COMPLETED',
+            claimToken: null,
+            attempts: 2,
+            progress: 100,
+        });
+        expect(mockPrismaJob.updateMany).toHaveBeenCalledWith(
+            expect.objectContaining({
+                where: {
+                    id: 'job-1',
+                    status: 'PROCESSING',
+                    claimToken: 'claim-old',
+                },
+            }),
+        );
+        expect(mockPrismaJob.update).not.toHaveBeenCalled();
+    });
+
+    it('counts crash-abandoned claims toward dead letter only after a recovery grace', async () => {
+        worker.on(JobType.AI_RESPONSE, async () => ({ success: true }));
+        mockPrisma.$queryRaw.mockResolvedValue([]);
+
+        worker.start();
+        await vi.advanceTimersByTimeAsync(0);
+
+        const reclaimCall = mockPrisma.$executeRaw.mock.calls[0];
+        const reclaimSql = reclaimCall[0].join(' ');
+        expect(reclaimSql).toContain('job."attempts" + 1');
+        expect(reclaimSql).toContain("THEN 'DEAD_LETTER'");
+        expect(reclaimSql).toContain('"claimToken" = NULL');
+        expect(reclaimSql).toContain('"lockUntil" = NULL');
+        // The grace sits on top of the recorded deadline: the normal timeout path
+        // must have time to release its own claim before another worker calls it
+        // crash-abandoned.
+        expect(reclaimCall.slice(1)).toContain(30_000);
+    });
+
+    it('spaces a reclaimed retry by the same backoff a handler failure would', async () => {
+        worker.on(JobType.AI_RESPONSE, async () => ({ success: true }));
+        mockPrisma.$queryRaw.mockResolvedValue([]);
+
+        worker.start();
+        await vi.advanceTimersByTimeAsync(0);
+
+        const reclaimCall = mockPrisma.$executeRaw.mock.calls[0];
+        const reclaimSql = reclaimCall[0].join(' ');
+
+        // `runAt = NOW()` let a crash-looping worker burn every attempt on a job
+        // back to back and drive it to DEAD_LETTER at full speed. A reclaim and a
+        // handler failure both mean "this attempt did not finish", so they have to
+        // space retries the same way.
+        // Asserting the shape of the `runAt` arm rather than the absence of one
+        // spelling of the regression: `not.toContain('ELSE NOW()\n')` only fired
+        // when a newline happened to follow, so `ELSE NOW() END` on one line —
+        // the same bug — walked straight past it.
+        expect(reclaimSql).toMatch(/"runAt" = CASE[\s\S]*ELSE NOW\(\)\s*\+/);
+        expect(reclaimSql).toContain('random()');
+        expect(reclaimSql).toContain('LEAST');
+
+        // The exponent is clamped. `maxAttempts` is per-row and settable through
+        // `createJob`, and float8 overflows around 2^1024 — which would abort the
+        // whole sweep for every row, not just the offending one. JS saturates
+        // gracefully here (`Math.min(Infinity, MAX)` is MAX), so without the clamp
+        // the SQL and `calculateBackoff` diverge at the extreme.
+        expect(reclaimSql).toContain('POWER(2, LEAST(job."attempts" + 1, 30))');
+
+        // Against the REAL constants, not the mock's copies of them.
+        expect(reclaimCall.slice(1)).toContain(realConstants.BACKOFF_BASE_MS);
+        expect(reclaimCall.slice(1)).toContain(realConstants.BACKOFF_MAX_MS);
+        // Mirrors calculateBackoff: BACKOFF_BASE_MS with jitter, BACKOFF_MAX_MS cap.
+        expect(reclaimCall.slice(1)).toContain(1000);
+        expect(reclaimCall.slice(1)).toContain(300_000);
+    });
+
     it('marks job DEAD_LETTER after maxAttempts exhausted', async () => {
         const jobRow = makeJobRow({ attempts: 4, maxAttempts: 5 }); // attempt will be 5
         mockPrisma.$queryRaw.mockResolvedValueOnce([jobRow]);
@@ -207,9 +484,9 @@ describe('Worker', () => {
         worker.start();
         await vi.advanceTimersByTimeAsync(0);
 
-        expect(mockPrismaJob.update).toHaveBeenCalledWith(
+        expect(mockPrismaJob.updateMany).toHaveBeenCalledWith(
             expect.objectContaining({
-                where: { id: 'job-1' },
+                where: expect.objectContaining({ id: 'job-1', claimToken: 'claim-1' }),
                 data: expect.objectContaining({
                     status: 'DEAD_LETTER',
                     attempts: 5,
@@ -232,9 +509,9 @@ describe('Worker', () => {
         worker.start();
         await vi.advanceTimersByTimeAsync(0);
 
-        expect(mockPrismaJob.update).toHaveBeenCalledWith(
+        expect(mockPrismaJob.updateMany).toHaveBeenCalledWith(
             expect.objectContaining({
-                where: { id: 'job-1' },
+                where: expect.objectContaining({ id: 'job-1', claimToken: 'claim-1' }),
                 data: expect.objectContaining({
                     status: 'PENDING',
                     attempts: 2,
@@ -268,7 +545,7 @@ describe('Worker', () => {
         await vi.advanceTimersByTimeAsync(200);
 
         // Should have been marked as retryable (attempt 1 of 5)
-        expect(mockPrismaJob.update).toHaveBeenCalledWith(
+        expect(mockPrismaJob.updateMany).toHaveBeenCalledWith(
             expect.objectContaining({
                 data: expect.objectContaining({
                     status: 'PENDING',
@@ -288,7 +565,7 @@ describe('Worker', () => {
         worker.start();
         await vi.advanceTimersByTimeAsync(0);
 
-        expect(mockPrismaJob.update).toHaveBeenCalledWith(
+        expect(mockPrismaJob.updateMany).toHaveBeenCalledWith(
             expect.objectContaining({
                 data: expect.objectContaining({
                     status: 'FAILED',
@@ -329,7 +606,7 @@ describe('Worker', () => {
         await vi.advanceTimersByTimeAsync(100);
 
         // All 3 should complete (they run concurrently via Promise.allSettled)
-        expect(mockPrismaJob.update).toHaveBeenCalledTimes(3);
+        expect(mockPrismaJob.updateMany).toHaveBeenCalledTimes(3);
     });
 
     it('provides accurate health check information', () => {
@@ -374,6 +651,66 @@ describe('Worker', () => {
         expect(jobFinished).toBe(true);
     });
 
+    it('shares the active-job drain across repeated stop calls', async () => {
+        let jobFinished = false;
+        const jobRow = makeJobRow();
+        mockPrisma.$queryRaw.mockResolvedValueOnce([jobRow]);
+        mockPrisma.$queryRaw.mockResolvedValue([]);
+        mockPrismaJob.update.mockResolvedValue({});
+
+        worker.on(JobType.AI_RESPONSE, async () => {
+            await new Promise((resolve) => setTimeout(resolve, 200));
+            jobFinished = true;
+            return { success: true };
+        });
+
+        worker.start();
+        await vi.advanceTimersByTimeAsync(0);
+
+        const signalStop = worker.stop();
+        let appStopResolved = false;
+        const appStop = worker.stop().then(() => {
+            appStopResolved = true;
+        });
+
+        await Promise.resolve();
+        expect(appStopResolved).toBe(false);
+        expect(jobFinished).toBe(false);
+
+        await vi.advanceTimersByTimeAsync(300);
+        await Promise.all([signalStop, appStop]);
+
+        expect(jobFinished).toBe(true);
+        expect(appStopResolved).toBe(true);
+    });
+
+    it('waits for an in-flight poll and does not claim after shutdown begins', async () => {
+        let releaseReclaim!: () => void;
+        mockPrisma.$executeRaw.mockImplementationOnce(
+            () =>
+                new Promise<number>((resolve) => {
+                    releaseReclaim = () => resolve(0);
+                }),
+        );
+
+        worker.on(JobType.AI_RESPONSE, async () => ({ success: true }));
+        worker.start();
+
+        const stopPromise = worker.stop();
+        let stopped = false;
+        void stopPromise.then(() => {
+            stopped = true;
+        });
+        await Promise.resolve();
+        expect(stopped).toBe(false);
+
+        releaseReclaim();
+        await stopPromise;
+
+        expect(mockPrisma.$queryRaw).not.toHaveBeenCalled();
+        expect(worker.healthCheck().running).toBe(false);
+    });
+
     it('handler receives context with progress reporting', async () => {
         const jobRow = makeJobRow();
         mockPrisma.$queryRaw.mockResolvedValueOnce([jobRow]);
@@ -393,12 +730,582 @@ describe('Worker', () => {
 
         expect(receivedContext).not.toBeNull();
         expect(receivedContext!.jobId).toBe('job-1');
-        // reportProgress should have called prisma.job.update with progress: 50
-        const progressCall = mockPrismaJob.update.mock.calls.find(
+        // reportProgress should fence the update to this execution's claim.
+        const progressCall = mockPrismaJob.updateMany.mock.calls.find(
             (call: Array<Record<string, Record<string, unknown>>>) => call[0].data.progress === 50,
         );
         expect(progressCall).toBeDefined();
     });
+
+    // Production sets `concurrencyByType` (apps/worker/src/index.ts), so every job
+    // it claims goes through `claimJobsForType` — and that query had no coverage
+    // of the deadline it writes. Deleting the `lockUntil` SET from it alone left
+    // all 1113 tests green while every production claim got `lockUntil = NULL` and
+    // fell into the 15-minute legacy branch forever, which is precisely the
+    // failure this whole change exists to remove.
+    describe('the claim path production actually runs', () => {
+        const setClauseOf = (call: unknown[]) => {
+            const sql = (call[0] as TemplateStringsArray).join(' ');
+            return sql.slice(
+                sql.indexOf("SET status = 'PROCESSING'"),
+                sql.indexOf('WHERE id IN ('),
+            );
+        };
+
+        it('writes a deadline onto every row it claims', async () => {
+            await worker.stop();
+            worker = new Worker({
+                pollIntervalMs: 100,
+                maxConcurrency: 2,
+                defaultTimeoutMs: 5000,
+                // The presence of this is what routes claiming through
+                // `claimJobsForType` instead of `claimAndProcessJobs`.
+                concurrencyByType: { [JobType.AI_RESPONSE]: 1 },
+                jobTimeouts: { [JobType.AI_RESPONSE]: 120_000 },
+            });
+            worker.on(JobType.AI_RESPONSE, async () => ({ success: true }));
+
+            worker.start();
+            await vi.advanceTimersByTimeAsync(0);
+
+            const claim = mockPrisma.$queryRaw.mock.calls[0];
+            const claimSql = claim[0].join(' ');
+            expect(claimSql).toContain('"lockUntil" = NOW()');
+            // Per-type, looked up by the row's own type, with the worker default
+            // as the fallback — the same contract the other claim path has.
+            expect(claimSql).toContain('->> "Job".type');
+            const timeoutMap = claim
+                .slice(1)
+                .find(
+                    (v: unknown): v is string => typeof v === 'string' && v.includes('AI_RESPONSE'),
+                );
+            expect(JSON.parse(timeoutMap as string)).toEqual({
+                [JobType.AI_RESPONSE]: 120_000,
+            });
+            expect(claim.slice(1)).toContain(5000);
+        });
+
+        it('claims identically whichever path is taken', async () => {
+            await worker.stop();
+            const perType = new Worker({
+                pollIntervalMs: 100,
+                defaultTimeoutMs: 5000,
+                concurrencyByType: { [JobType.AI_RESPONSE]: 1 },
+            });
+            perType.on(JobType.AI_RESPONSE, async () => ({ success: true }));
+            perType.start();
+            await vi.advanceTimersByTimeAsync(0);
+            const perTypeSet = setClauseOf(mockPrisma.$queryRaw.mock.calls[0]);
+            await perType.stop();
+
+            mockPrisma.$queryRaw.mockClear();
+
+            worker = new Worker({ pollIntervalMs: 100, defaultTimeoutMs: 5000 });
+            worker.on(JobType.AI_RESPONSE, async () => ({ success: true }));
+            worker.start();
+            await vi.advanceTimersByTimeAsync(0);
+            const batchSet = setClauseOf(mockPrisma.$queryRaw.mock.calls[0]);
+
+            // The two claim queries carry byte-identical SET clauses, duplicated by
+            // hand. Nothing else notices when one is edited and the other is not,
+            // and a divergence there is silent in production and invisible in CI.
+            expect(perTypeSet).toBe(batchSet);
+            expect(perTypeSet).toContain('"lockUntil" = NOW()');
+            expect(perTypeSet).toContain('"claimToken" = gen_random_uuid()::text');
+        });
+    });
+
+    // A claim whose writes cannot be fenced is worse than no claim: Prisma drops a
+    // `where` key whose value is undefined, so every fenced write in `processJob`
+    // would quietly become an unfenced update-by-id.
+    describe('a claim with no token', () => {
+        it('is refused rather than run unfenced', async () => {
+            const error = vi.spyOn(console, 'error').mockImplementation(() => {});
+            mockPrisma.$queryRaw.mockResolvedValueOnce([
+                { ...makeJobRow(), claimToken: undefined as unknown as string },
+            ]);
+
+            let ran = false;
+            worker.on(JobType.AI_RESPONSE, async () => {
+                ran = true;
+                return { success: true };
+            });
+            worker.start();
+            await vi.advanceTimersByTimeAsync(0);
+
+            expect(ran).toBe(false);
+            // Left PROCESSING with its deadline intact, so the sweep recovers it.
+            expect(mockPrismaJob.updateMany).not.toHaveBeenCalled();
+            expect(error.mock.calls.map((c: unknown[]) => String(c[0])).join('\n')).toContain(
+                'could not be fenced',
+            );
+            error.mockRestore();
+        });
+    });
+
+    // The bookkeeping that records a result is not the work itself, and the two
+    // must not fail the same way.
+    describe('when the database fails after the handler succeeded', () => {
+        it('does not re-queue work that already ran', async () => {
+            const error = vi.spyOn(console, 'error').mockImplementation(() => {});
+            mockPrisma.$queryRaw.mockResolvedValueOnce([makeJobRow()]);
+            mockPrismaJob.updateMany.mockRejectedValueOnce(new Error('connection reset by peer'));
+
+            worker.on(JobType.AI_RESPONSE, async () => ({ success: true }));
+            worker.start();
+            await vi.advanceTimersByTimeAsync(0);
+
+            // The old shape wrapped the completion write in the handler's own try,
+            // so a Prisma blip was laundered into "the job failed": the row went
+            // back to PENDING carrying the DB error as the job's error, and the
+            // retry re-ran every external side effect the handler had already
+            // produced. That is the duplicate execution the claim token exists to
+            // detect, manufactured by the worker from a bookkeeping error.
+            const retried = mockPrismaJob.updateMany.mock.calls
+                .map((call: Array<{ data: Record<string, unknown> }>) => call[0].data)
+                .find((data: Record<string, unknown>) => data.status === 'PENDING');
+            expect(retried).toBeUndefined();
+
+            const messages = error.mock.calls.map((c: unknown[]) => String(c[0])).join('\n');
+            expect(messages).toContain('succeeded but its');
+            error.mockRestore();
+        });
+
+        it('reports a processJob that threw outright', async () => {
+            const error = vi.spyOn(console, 'error').mockImplementation(() => {});
+            mockPrisma.$queryRaw.mockResolvedValueOnce([makeJobRow({ attempts: 1 })]);
+            // Both the failure write and its fallback fail — the DB is down, which
+            // is exactly when this happens.
+            mockPrismaJob.updateMany.mockRejectedValue(new Error('pool exhausted'));
+
+            worker.on(JobType.AI_RESPONSE, async () => {
+                throw new Error('handler said no');
+            });
+            worker.start();
+            await vi.advanceTimersByTimeAsync(0);
+
+            // `Promise.allSettled` absorbed the rejection and nobody read the
+            // result, so this produced no output at all: not the Prisma error, and
+            // not the handler failure it was trying to record.
+            const messages = error.mock.calls.map((c: unknown[]) => String(c[0])).join('\n');
+            expect(messages).toContain('processJob threw');
+            error.mockRestore();
+        });
+    });
+
+    // The sweep is the only thing standing between a crashed claim and a row that
+    // is PROCESSING forever, and it is also the first `await` in every poll. Both
+    // properties are load-bearing and neither was pinned.
+    describe('the reclaim sweep', () => {
+        let warn: ReturnType<typeof vi.spyOn>;
+        let error: ReturnType<typeof vi.spyOn>;
+
+        beforeEach(() => {
+            warn = vi.spyOn(console, 'warn').mockImplementation(() => {});
+            error = vi.spyOn(console, 'error').mockImplementation(() => {});
+        });
+
+        afterEach(() => {
+            warn.mockRestore();
+            error.mockRestore();
+        });
+
+        it('cannot stop the worker claiming when it fails', async () => {
+            // The sweep runs ahead of every claim in the same `try`, and
+            // `lastPollTime` is already set by then. Without its own catch, one
+            // failing sweep ends all claiming while `healthCheck()` still reports
+            // healthy: a total outage with nothing to restart it. Recovering
+            // abandoned work is a nice-to-have; claiming new work is the job.
+            mockPrisma.$executeRaw.mockRejectedValueOnce(new Error('reclaim exploded'));
+            mockPrisma.$queryRaw.mockResolvedValue([]);
+
+            worker.on(JobType.AI_RESPONSE, async () => ({ success: true }));
+            worker.start();
+            await vi.advanceTimersByTimeAsync(0);
+
+            // The claim ran anyway. This is the assertion that dies if the
+            // try/catch around the sweep is deleted.
+            expect(mockPrisma.$queryRaw).toHaveBeenCalled();
+
+            // And it is not a silent recovery.
+            const logged = error.mock.calls.map((c: unknown[]) => String(c[0])).join('\n');
+            expect(logged).toContain('Reclaim sweep failed');
+        });
+
+        it('keeps lockUntil bare on one side so the index can serve the predicate', async () => {
+            mockPrisma.$queryRaw.mockResolvedValue([]);
+            worker.on(JobType.AI_RESPONSE, async () => ({ success: true }));
+            worker.start();
+            await vi.advanceTimersByTimeAsync(0);
+
+            const reclaimSql = mockPrisma.$executeRaw.mock.calls[0][0].join(' ');
+            const where = reclaimSql.slice(reclaimSql.indexOf("WHERE job.status = 'PROCESSING'"));
+
+            // `Job_status_lockUntil_idx` exists for this predicate. A `CASE` over
+            // the column is not sargable, so the planner would take the `status`
+            // prefix and then filter every PROCESSING row — the index would be
+            // there and unusable, which is worse than not adding it.
+            expect(where).not.toContain('CASE');
+            expect(where).toContain('job."lockUntil"');
+
+            // Same reason, one level down: the interval arithmetic has to sit on
+            // the right-hand side. `lockUntil + interval < NOW()` is a disjunction
+            // and still non-sargable.
+            expect(where).toContain('< NOW() -');
+            expect(where).not.toMatch(/job\."lockUntil"\s*\+/);
+
+            // Deliberately not asserting the same of `lockedAt`. That arm cannot
+            // use the index's second column either way — `lockedAt` is not in the
+            // index — so it rides the status + `lockUntil IS NULL` prefix and then
+            // filters. An assertion there would look like it defended the index
+            // and defend nothing.
+        });
+
+        it('leaves no PROCESSING row without a way out', async () => {
+            worker.start();
+            await vi.advanceTimersByTimeAsync(0);
+
+            const reclaimSql = mockPrisma.$executeRaw.mock.calls[0][0].join(' ');
+            const where = reclaimSql.slice(reclaimSql.indexOf("WHERE job.status = 'PROCESSING'"));
+
+            // Both timestamp arms NULL-guard one column and compare the other, and
+            // `NULL < x` is NULL — so a PROCESSING row carrying neither timestamp
+            // matched no arm and sat there forever, invisible to the one mechanism
+            // that exists to rescue it. Unreachable from the two claim paths, which
+            // is exactly the assumption a last-resort sweep should not be making.
+            // `updatedAt` is written by every path, so it closes the blind spot.
+            expect(where).toContain('job."updatedAt"');
+        });
+
+        it('keeps the failure that actually killed a dead-lettered job', async () => {
+            worker.start();
+            await vi.advanceTimersByTimeAsync(0);
+
+            const reclaimSql = mockPrisma.$executeRaw.mock.calls[0][0].join(' ');
+
+            // The sweep used to overwrite `error` unconditionally, including on the
+            // arm that lands on DEAD_LETTER — so the one surface where the *why*
+            // matters most was left holding a generic string. Bounded with `left`
+            // so repeated reclaims of a crash-looping job cannot grow it without
+            // limit.
+            expect(reclaimSql).toContain('previous error');
+            expect(reclaimSql).toContain('left(job.error, 200)');
+        });
+
+        it('says how many rows it returned to the queue', async () => {
+            const warn = vi.spyOn(console, 'warn').mockImplementation(() => {});
+            mockPrisma.$executeRaw.mockResolvedValue(7);
+
+            worker.start();
+            await vi.advanceTimersByTimeAsync(0);
+
+            // Every row this moves is a claim a worker took and never released.
+            // The count was computed once per poll on every replica and discarded,
+            // which threw away the earliest signal that replicas are dying.
+            const messages = warn.mock.calls.map((c: unknown[]) => String(c[0])).join('\n');
+            expect(messages).toContain('7 abandoned');
+            warn.mockRestore();
+        });
+
+        it('stays silent when it reclaimed nothing', async () => {
+            const warn = vi.spyOn(console, 'warn').mockImplementation(() => {});
+            mockPrisma.$executeRaw.mockResolvedValue(0);
+
+            worker.start();
+            await vi.advanceTimersByTimeAsync(0);
+
+            const messages = warn.mock.calls.map((c: unknown[]) => String(c[0])).join('\n');
+            expect(messages).not.toContain('abandoned');
+            warn.mockRestore();
+        });
+
+        it('still runs when this replica has no free slots', async () => {
+            const saturated = new Worker({ pollIntervalMs: 100, maxConcurrency: 0 });
+            saturated.on(JobType.AI_RESPONSE, async () => ({ success: true }));
+
+            saturated.start();
+            await vi.advanceTimersByTimeAsync(0);
+
+            // Reclaiming is a global sweep over rows other replicas abandoned. It
+            // has nothing to do with this replica's spare capacity, and gating it
+            // behind the capacity check stopped recovery exactly when the backlog
+            // that produced the abandoned rows was largest.
+            expect(mockPrisma.$executeRaw).toHaveBeenCalled();
+            await saturated.stop();
+        });
+
+        it('consumes an attempt, so a crash-looping job still reaches DEAD_LETTER', async () => {
+            mockPrisma.$queryRaw.mockResolvedValue([]);
+            worker.on(JobType.AI_RESPONSE, async () => ({ success: true }));
+            worker.start();
+            await vi.advanceTimersByTimeAsync(0);
+
+            const reclaimSql = mockPrisma.$executeRaw.mock.calls[0][0].join(' ');
+
+            // Asserting the `SET` fragment specifically. A bare
+            // `toContain('job."attempts" + 1')` is satisfied by the occurrences in
+            // the `status`, `completedAt` and `runAt` arms, so mutating the
+            // assignment to `"attempts" = job."attempts"` walks straight past it —
+            // and a job that never accrues an attempt is reclaimed forever.
+            expect(reclaimSql).toContain('"attempts" = job."attempts" + 1');
+        });
+
+        it("is blind to job type, so no replica can strand another replica's work", async () => {
+            mockPrisma.$queryRaw.mockResolvedValue([]);
+            worker.on(JobType.AI_RESPONSE, async () => ({ success: true }));
+            worker.start();
+            await vi.advanceTimersByTimeAsync(0);
+
+            const reclaimSql = mockPrisma.$executeRaw.mock.calls[0][0].join(' ');
+
+            // `lockUntil` makes the row self-describing, so nothing about the type
+            // is needed to tell that a claim has expired. Asserting the absence of
+            // any type reference rather than one spelling of the old predicate:
+            // `AND job.type = ANY(...)` would have satisfied the previous guard,
+            // and it reintroduces the bug where a crashed claim of a type this
+            // replica does not register sits PROCESSING forever.
+            expect(reclaimSql).not.toContain('job.type');
+            expect(reclaimSql).not.toContain('"Job".type');
+        });
+    });
+
+    // Every release path has to clear the deadline it set. Inert while the
+    // predicate gates on `status = 'PROCESSING'`, but the first future path that
+    // sets PROCESSING without writing a fresh `lockUntil` inherits a deadline
+    // already in the past and gets reclaimed mid-flight.
+    describe('releasing a claim clears its deadline', () => {
+        const releaseData = (status: string) =>
+            mockPrismaJob.updateMany.mock.calls
+                .map((call: Array<{ data: Record<string, unknown> }>) => call[0].data)
+                .find((data: Record<string, unknown>) => data.status === status);
+
+        it('clears lockUntil when a job completes', async () => {
+            mockPrisma.$queryRaw.mockResolvedValueOnce([makeJobRow()]);
+            mockPrisma.$queryRaw.mockResolvedValue([]);
+
+            worker.on(JobType.AI_RESPONSE, async () => ({ success: true }));
+            worker.start();
+            await vi.advanceTimersByTimeAsync(0);
+
+            expect(releaseData('COMPLETED')).toMatchObject({ lockUntil: null });
+        });
+
+        it('clears lockUntil when a job is scheduled for retry', async () => {
+            mockPrisma.$queryRaw.mockResolvedValueOnce([makeJobRow({ attempts: 1 })]);
+            mockPrisma.$queryRaw.mockResolvedValue([]);
+
+            worker.on(JobType.AI_RESPONSE, async () => ({
+                success: false,
+                error: 'transient',
+            }));
+            worker.start();
+            await vi.advanceTimersByTimeAsync(0);
+
+            expect(releaseData('PENDING')).toMatchObject({ lockUntil: null });
+        });
+
+        it('clears lockUntil when a job is dead-lettered', async () => {
+            mockPrisma.$queryRaw.mockResolvedValueOnce([
+                makeJobRow({ attempts: 4, maxAttempts: 5 }),
+            ]);
+            mockPrisma.$queryRaw.mockResolvedValue([]);
+
+            worker.on(JobType.AI_RESPONSE, async () => ({
+                success: false,
+                error: 'permanently broken',
+            }));
+            worker.start();
+            await vi.advanceTimersByTimeAsync(0);
+
+            expect(releaseData('DEAD_LETTER')).toMatchObject({ lockUntil: null });
+        });
+    });
+
+    // A fenced write means two executions of the same row overlapped — the exact
+    // event the claim token exists to produce. Before these, changing both
+    // `count > 0` checks to `count >= 0` passed the whole suite: the fence fired
+    // and said nothing, on every path.
+    describe('fence rejections are reported', () => {
+        let warn: ReturnType<typeof vi.spyOn>;
+        let error: ReturnType<typeof vi.spyOn>;
+
+        beforeEach(() => {
+            warn = vi.spyOn(console, 'warn').mockImplementation(() => {});
+            error = vi.spyOn(console, 'error').mockImplementation(() => {});
+        });
+
+        afterEach(() => {
+            warn.mockRestore();
+            error.mockRestore();
+        });
+
+        const fencedMessages = () => warn.mock.calls.map((c: unknown[]) => String(c[0])).join('\n');
+
+        it('warns when a completed job can no longer write its own result', async () => {
+            mockPrisma.$queryRaw.mockResolvedValueOnce([makeJobRow()]);
+            mockPrisma.$queryRaw.mockResolvedValue([]);
+            // Someone else owns the row: it was reclaimed while this execution was
+            // still live, so this handler's side effects have now happened twice.
+            mockPrismaJob.updateMany.mockResolvedValue({ count: 0 });
+
+            worker.on(JobType.AI_RESPONSE, async () => ({ success: true }));
+            worker.start();
+            await vi.advanceTimersByTimeAsync(0);
+
+            expect(fencedMessages()).toContain('job-1');
+            expect(fencedMessages()).toContain('reclaimed while still live');
+        });
+
+        it('keeps the underlying failure when a retry write is fenced', async () => {
+            mockPrisma.$queryRaw.mockResolvedValueOnce([
+                makeJobRow({ attempts: 1, maxAttempts: 5 }),
+            ]);
+            mockPrisma.$queryRaw.mockResolvedValue([]);
+            mockPrismaJob.updateMany.mockResolvedValue({ count: 0 });
+
+            worker.on(JobType.AI_RESPONSE, async () => {
+                throw new Error('upstream timed out');
+            });
+            worker.start();
+            await vi.advanceTimersByTimeAsync(0);
+
+            // Pre-fence this path always logged. Losing the retry log would also
+            // lose the failure that caused the timeout in the first place.
+            expect(fencedMessages()).toContain('upstream timed out');
+            expect(fencedMessages()).toContain('fenced');
+        });
+
+        it('keeps the underlying failure when a dead-letter write is fenced', async () => {
+            mockPrisma.$queryRaw.mockResolvedValueOnce([
+                makeJobRow({ attempts: 4, maxAttempts: 5 }),
+            ]);
+            mockPrisma.$queryRaw.mockResolvedValue([]);
+            mockPrismaJob.updateMany.mockResolvedValue({ count: 0 });
+
+            worker.on(JobType.AI_RESPONSE, async () => ({
+                success: false,
+                error: 'permanently broken',
+            }));
+            worker.start();
+            await vi.advanceTimersByTimeAsync(0);
+
+            expect(fencedMessages()).toContain('permanently broken');
+            expect(fencedMessages()).toContain('dead-letter write was fenced');
+        });
+
+        it('warns when the no-handler tombstone is fenced', async () => {
+            // The fifth fenced write in the file, and the last one that was still
+            // silent. It also clears the claim, so a fenced tombstone means some
+            // other execution owns a row this one just tried to mark FAILED.
+            mockPrisma.$queryRaw.mockResolvedValueOnce([makeJobRow({ type: JobType.SLA_CHECK })]);
+            mockPrisma.$queryRaw.mockResolvedValue([]);
+            mockPrismaJob.updateMany.mockResolvedValue({ count: 0 });
+
+            // No handler registered for SLA_CHECK on this worker.
+            worker.start();
+            await vi.advanceTimersByTimeAsync(0);
+
+            expect(fencedMessages()).toContain('no-handler failure');
+            expect(fencedMessages()).toContain('reclaimed while still live');
+        });
+
+        it('records the attempt the no-handler tombstone consumed', async () => {
+            mockPrisma.$queryRaw.mockResolvedValueOnce([
+                makeJobRow({ type: JobType.SLA_CHECK, attempts: 2 }),
+            ]);
+
+            worker.start();
+            await vi.advanceTimersByTimeAsync(0);
+
+            // Every other terminal path writes the attempt it used. This one did
+            // not, so a job that was claimed and dispatched read `attempts: 2`
+            // afterwards — indistinguishable from one that was never picked up.
+            const tombstone = mockPrismaJob.updateMany.mock.calls
+                .map((call: Array<{ data: Record<string, unknown> }>) => call[0].data)
+                .find((data: Record<string, unknown>) => data.status === 'FAILED');
+            expect(tombstone).toMatchObject({ attempts: 3 });
+        });
+
+        it('clears lockUntil on the no-handler tombstone', async () => {
+            mockPrisma.$queryRaw.mockResolvedValueOnce([makeJobRow({ type: JobType.SLA_CHECK })]);
+            mockPrisma.$queryRaw.mockResolvedValue([]);
+
+            worker.start();
+            await vi.advanceTimersByTimeAsync(0);
+
+            const tombstone = mockPrismaJob.updateMany.mock.calls
+                .map((call: Array<{ data: Record<string, unknown> }>) => call[0].data)
+                .find((data: Record<string, unknown>) => data.status === 'FAILED');
+            expect(tombstone).toMatchObject({ lockUntil: null });
+        });
+
+        it('warns when a progress report is fenced', async () => {
+            mockPrisma.$queryRaw.mockResolvedValueOnce([makeJobRow()]);
+            mockPrisma.$queryRaw.mockResolvedValue([]);
+            mockPrismaJob.updateMany.mockResolvedValue({ count: 0 });
+
+            worker.on(JobType.AI_RESPONSE, async (_payload, ctx) => {
+                await ctx.reportProgress(50);
+                return { success: true };
+            });
+            worker.start();
+            await vi.advanceTimersByTimeAsync(0);
+
+            // Earliest observable sign that this execution has lost its claim while
+            // the handler is still running.
+            expect(fencedMessages()).toContain('Progress update for job job-1 was fenced');
+        });
+    });
+});
+
+// `LEGACY_RECLAIM_CEILING_MS` is fixed on purpose — deriving it from the observing
+// worker's config is the bug `lockUntil` was added to remove. But fixed means it
+// does not follow `jobTimeouts` upward, and the two live in different packages, so
+// nothing else would notice them drifting apart.
+describe('Worker legacy-reclaim ceiling guard', () => {
+    let warn: ReturnType<typeof vi.spyOn>;
+
+    beforeEach(() => {
+        vi.clearAllMocks();
+        warn = vi.spyOn(console, 'warn').mockImplementation(() => {});
+    });
+
+    afterEach(() => {
+        warn.mockRestore();
+    });
+
+    const warnings = () => warn.mock.calls.map((c: unknown[]) => String(c[0])).join('\n');
+
+    it('stays quiet for the timeouts the worker actually ships with', () => {
+        // apps/worker/src/index.ts: the two longest are HUBSPOT_SYNC and
+        // ACCOUNT_SCORING, tied at 300s. 300_000 + 30_000 grace < 900_000.
+        new Worker({
+            defaultTimeoutMs: 30_000,
+            jobTimeouts: {
+                [JobType.AI_RESPONSE]: 120_000,
+                [JobType.HUBSPOT_SYNC]: 300_000,
+                [JobType.ACCOUNT_SCORING]: 300_000,
+            },
+        });
+
+        expect(warnings()).not.toContain('LEGACY_RECLAIM_CEILING_MS');
+    });
+
+    it('says so when a configured timeout outgrows the ceiling', () => {
+        // During the one rollout where rows still carry a NULL `lockUntil`, a
+        // timeout this long means the sweep calls a live claim abandoned and the
+        // original execution's completion is fenced out and discarded.
+        new Worker({ jobTimeouts: { [JobType.HUBSPOT_SYNC]: 1_200_000 } });
+
+        expect(warnings()).toContain('LEGACY_RECLAIM_CEILING_MS');
+        expect(warnings()).toContain('1230000');
+    });
+
+    it('counts defaultTimeoutMs too, not just the per-type map', () => {
+        new Worker({ defaultTimeoutMs: 1_200_000 });
+
+        expect(warnings()).toContain('LEGACY_RECLAIM_CEILING_MS');
+    });
 });
 
 describe('Scheduler', () => {
diff --git a/packages/outpost/queue/src/__tests__/worker-concurrency.test.ts b/packages/outpost/queue/src/__tests__/worker-concurrency.test.ts
index 441c5f5..20eec81 100644
--- a/packages/outpost/queue/src/__tests__/worker-concurrency.test.ts
+++ b/packages/outpost/queue/src/__tests__/worker-concurrency.test.ts
@@ -16,11 +16,13 @@ import type { JobHandlerContext, WorkerHealthStatus } from '../types.js';
 const mockPrismaJob = {
     create: vi.fn(),
     update: vi.fn(),
+    updateMany: vi.fn(),
     findFirst: vi.fn(),
 };
 
 const mockPrisma = {
     job: mockPrismaJob,
+    $executeRaw: vi.fn(),
     $queryRaw: vi.fn(),
 };
 
@@ -39,19 +41,23 @@ const { Worker } = await import('../worker.js');
 
 // ─── Helpers ────────────────────────────────────────────────────────────────
 
-function makeJobRow(overrides: Partial<{
-    id: string;
-    type: string;
-    payload: unknown;
-    attempts: number;
-    maxAttempts: number;
-}> = {}) {
+function makeJobRow(
+    overrides: Partial<{
+        id: string;
+        type: string;
+        payload: unknown;
+        attempts: number;
+        maxAttempts: number;
+        claimToken: string;
+    }> = {},
+) {
     return {
         id: overrides.id ?? 'job-1',
         type: overrides.type ?? JobType.AI_RESPONSE,
         payload: overrides.payload ?? { ticketId: 'tkt-1', source: 'discord' },
         attempts: overrides.attempts ?? 0,
         maxAttempts: overrides.maxAttempts ?? 5,
+        claimToken: overrides.claimToken ?? `claim-${overrides.id ?? 'job-1'}`,
     };
 }
 
@@ -63,6 +69,8 @@ describe('Worker per-type concurrency', () => {
     beforeEach(() => {
         vi.clearAllMocks();
         vi.useFakeTimers();
+        mockPrisma.$executeRaw.mockResolvedValue(0);
+        mockPrismaJob.updateMany.mockResolvedValue({ count: 1 });
     });
 
     afterEach(async () => {
@@ -103,14 +111,48 @@ describe('Worker per-type concurrency', () => {
 
         // Process should have picked up the job
         expect(mockPrisma.$queryRaw).toHaveBeenCalled();
+        const claimSql = mockPrisma.$queryRaw.mock.calls[0][0].join(' ');
+        expect(claimSql).toContain('"claimToken" = gen_random_uuid()::text');
+        expect(claimSql).toContain('"maxAttempts", "claimToken"');
         // Job should have been completed
-        expect(mockPrismaJob.update).toHaveBeenCalledWith(
+        expect(mockPrismaJob.updateMany).toHaveBeenCalledWith(
             expect.objectContaining({
                 data: expect.objectContaining({ status: 'COMPLETED' }),
             }),
         );
     });
 
+    it('reclaims stale processing jobs before per-type claims', async () => {
+        const now = new Date('2026-08-11T12:00:00.000Z');
+        vi.setSystemTime(now);
+        worker = new Worker({
+            pollIntervalMs: 100,
+            maxConcurrency: 2,
+            concurrencyByType: {
+                [JobType.AI_RESPONSE]: 1,
+            },
+            jobTimeouts: {
+                [JobType.AI_RESPONSE]: 2000,
+            },
+        });
+
+        mockPrisma.$queryRaw.mockResolvedValue([]);
+        worker.on(JobType.AI_RESPONSE, async () => ({ success: true }));
+
+        worker.start();
+        await vi.advanceTimersByTimeAsync(0);
+
+        expect(mockPrisma.$executeRaw).toHaveBeenCalledTimes(1);
+        // The sweep no longer carries a per-type policy set. `lockUntil` is on the
+        // row, written by whoever claimed it, so this worker's own timeout config
+        // is not an input to the decision.
+        const reclaimSql = mockPrisma.$executeRaw.mock.calls[0][0].join(' ');
+        expect(reclaimSql).toContain('job."lockUntil"');
+        expect(reclaimSql).not.toContain('jsonb_to_recordset');
+        expect(reclaimSql).not.toContain('job.type = policy.type');
+        expect(mockPrisma.$queryRaw).toHaveBeenCalled();
+    });
+
     it('falls back to global limit when concurrencyByType is not specified', async () => {
         worker = new Worker({
             pollIntervalMs: 100,
@@ -187,7 +229,7 @@ describe('Worker per-type concurrency', () => {
         await vi.advanceTimersByTimeAsync(0);
 
         // Both AI jobs should have completed
-        const completedCalls = mockPrismaJob.update.mock.calls.filter(
+        const completedCalls = mockPrismaJob.updateMany.mock.calls.filter(
             (call: Array<Record<string, Record<string, unknown>>>) =>
                 call[0].data.status === 'COMPLETED',
         );
@@ -220,11 +262,13 @@ describe('Worker per-type concurrency', () => {
 
         // Per-type claims: first call for AI_RESPONSE, second for ESCALATION
         mockPrisma.$queryRaw
+            .mockResolvedValueOnce([makeJobRow({ id: 'ai-1', type: JobType.AI_RESPONSE })])
             .mockResolvedValueOnce([
-                makeJobRow({ id: 'ai-1', type: JobType.AI_RESPONSE }),
-            ])
-            .mockResolvedValueOnce([
-                makeJobRow({ id: 'esc-1', type: JobType.ESCALATION, payload: { ticketId: 'tkt-esc', reason: 'test' } }),
+                makeJobRow({
+                    id: 'esc-1',
+                    type: JobType.ESCALATION,
+                    payload: { ticketId: 'tkt-esc', reason: 'test' },
+                }),
             ])
             .mockResolvedValue([]);
 
diff --git a/packages/outpost/queue/src/create-job.ts b/packages/outpost/queue/src/create-job.ts
index 59ac2f6..5b8b95d 100644
--- a/packages/outpost/queue/src/create-job.ts
+++ b/packages/outpost/queue/src/create-job.ts
@@ -25,13 +25,43 @@ export async function createJob<T extends JobType>(
 }
 
 /**
- * Update the progress of a running job.
+ * Update the progress of the running job claim that owns this execution.
  * Progress is a percentage from 0 to 100.
  */
-export async function updateJobProgress(jobId: string, percent: number): Promise<void> {
+export async function updateJobProgress(
+    jobId: string,
+    percent: number,
+    claimToken: string,
+): Promise<void> {
     const clamped = Math.max(0, Math.min(100, Math.round(percent)));
-    await prisma.job.update({
-        where: { id: jobId },
-        data: { progress: clamped },
-    });
+
+    // Handlers `await` this, so a rejection here propagates into the handler and
+    // the worker records it as a job failure — retrying work that was running
+    // perfectly well and repeating every side effect it had already produced.
+    // Progress is telemetry; it must never be able to fail the job it describes.
+    let result: { count: number };
+    try {
+        result = await prisma.job.updateMany({
+            where: { id: jobId, status: 'PROCESSING', claimToken },
+            data: { progress: clamped },
+        });
+    } catch (error) {
+        console.warn(
+            `[Queue] Progress update for job ${jobId} failed and was ignored, ` +
+                `so it cannot fail the running job:`,
+            error,
+        );
+        return;
+    }
+
+    // A dropped progress update is harmless in itself, but it is the earliest
+    // observable sign that this execution has lost its claim — the handler is
+    // still running while something else owns the row. Worth a line, since the
+    // fence is otherwise indistinguishable from a successful write.
+    if (result.count === 0) {
+        console.warn(
+            `[Queue] Progress update for job ${jobId} was fenced: ` +
+                `claim ${claimToken} no longer owns the row.`,
+        );
+    }
 }
diff --git a/packages/outpost/queue/src/worker.ts b/packages/outpost/queue/src/worker.ts
index 6c7738a..8ca273e 100644
--- a/packages/outpost/queue/src/worker.ts
+++ b/packages/outpost/queue/src/worker.ts
@@ -1,5 +1,5 @@
 import { prisma } from '@copilotkit/outpost/db';
-import { calculateBackoff } from '@copilotkit/outpost/shared';
+import { BACKOFF_BASE_MS, BACKOFF_MAX_MS, calculateBackoff } from '@copilotkit/outpost/shared';
 import { updateJobProgress } from './create-job.js';
 import type {
     JobType,
@@ -10,6 +10,75 @@ import type {
     JobHandlerContext,
 } from './types.js';
 
+/**
+ * Margin added on top of a claim's own deadline before the sweep calls it dead.
+ *
+ * The normal timeout path has to be given a chance to release its own claim
+ * first; without this, a handler that times out at exactly `lockUntil` races the
+ * sweep that is about to reclaim it. Also the amount `warnIfLegacyCeilingTooLow`
+ * adds to the longest configured timeout when checking the legacy ceiling.
+ */
+const STALE_RECOVERY_GRACE_MS = 30_000;
+
+/**
+ * How long a `PROCESSING` row with no `lockUntil` is left alone.
+ *
+ * Only reachable for rows a pre-`lockUntil` worker claimed, i.e. during the one
+ * deploy that rolls this out. We genuinely do not know what deadline those were
+ * granted, so the ceiling is a fixed value set well above the largest configured
+ * timeout rather than derived from the observing worker's config — that
+ * derivation is exactly the bug `lockUntil` exists to remove, so it must not come
+ * back through this door. (At the time of writing the longest are `HUBSPOT_SYNC`
+ * and `ACCOUNT_SCORING`, tied at 300s; `warnIfLegacyCeilingTooLow` below is what
+ * keeps that honest, so treat the guard rather than this sentence as the source
+ * of truth.)
+ *
+ * Being fixed means it does not follow `jobTimeouts` upward. A timeout raised
+ * past this ceiling would make the legacy branch reclaim live claims for the
+ * length of one rollout, so the constructor checks the two against each other
+ * and says so rather than letting it pass silently.
+ */
+const LEGACY_RECLAIM_CEILING_MS = 900_000;
+
+/**
+ * Report a fenced write.
+ *
+ * `count === 0` on any of these updates is the event the claim token exists to
+ * produce, and it must never be inferred from the absence of a log. The pre-fence
+ * code always logged on these paths; suppressing the log when the fence fires
+ * would make the interesting case the quiet one.
+ *
+ * What it proves is narrower than it first looks: the row no longer matches
+ * `(id, PROCESSING, claimToken)`. That means the claim was lost — the reclaim
+ * sweep took the row — but not necessarily that a second execution has happened.
+ * The sweep sets `claimToken = NULL` and may have landed on DEAD_LETTER, in which
+ * case nothing will run again. The message says what is known and what is at
+ * risk, rather than asserting a concurrent run that may not exist.
+ */
+function warnIfFenced(
+    count: number,
+    job: Pick<ClaimedJob, 'id' | 'type' | 'claimToken'>,
+    what: string,
+): void {
+    if (count > 0) return;
+    console.warn(
+        `[Queue Worker] ${what} write for job ${job.id} (${job.type}) was fenced: ` +
+            `claim ${job.claimToken} no longer owns the row, so this claim was ` +
+            `reclaimed while still live. The row has since been retried or ` +
+            `dead-lettered, and any external side effect of this execution may ` +
+            `already have been repeated.`,
+    );
+}
+
+interface ClaimedJob {
+    id: string;
+    type: string;
+    payload: unknown;
+    attempts: number;
+    maxAttempts: number;
+    claimToken: string;
+}
+
 /**
  * A worker that polls the Postgres job queue and processes jobs using
  * SELECT ... FOR UPDATE SKIP LOCKED for safe concurrent processing.
@@ -33,21 +102,53 @@ export class Worker {
     private jobTimeouts: Partial<Record<JobType, number>>;
     private defaultTimeoutMs: number;
     private pollTimer: ReturnType<typeof setTimeout> | null = null;
+    private pollPromise: Promise<void> | null = null;
     private activeJobs = new Set<string>();
     /** Track active job counts per type for per-type concurrency enforcement */
     private activeJobsByType = new Map<string, number>();
     private lastPollTime: Date | null = null;
     private upSince: Date | null = null;
     private shutdownResolve: (() => void) | null = null;
+    private stopPromise: Promise<void> | null = null;
     private signalHandlers: { signal: string; handler: () => void }[] = [];
 
     constructor(options?: WorkerOptions) {
         this.pollIntervalMs = options?.pollIntervalMs ?? 1000;
         this.batchSize = options?.batchSize ?? 10;
         this.maxConcurrency = options?.maxConcurrency ?? 5;
-        this.concurrencyByType = (options?.concurrencyByType ?? {}) as Partial<Record<string, number>>;
+        this.concurrencyByType = (options?.concurrencyByType ?? {}) as Partial<
+            Record<string, number>
+        >;
         this.jobTimeouts = options?.jobTimeouts ?? {};
         this.defaultTimeoutMs = options?.defaultTimeoutMs ?? 30_000;
+        this.warnIfLegacyCeilingTooLow();
+    }
+
+    /**
+     * Say so when a configured timeout outgrows `LEGACY_RECLAIM_CEILING_MS`.
+     *
+     * The legacy branch only runs against rows claimed before `lockUntil`
+     * existed, so this is a one-rollout concern — but during that rollout a
+     * timeout above the ceiling means the sweep calls a still-running claim
+     * abandoned, and the original execution's completion is then fenced out and
+     * discarded. Checked here rather than left to a comment because the ceiling
+     * and the timeouts live in different packages, so nothing else would notice
+     * them drifting apart. Once every `PROCESSING` row carries a `lockUntil`,
+     * the branch is unreachable and this is only advisory.
+     */
+    private warnIfLegacyCeilingTooLow(): void {
+        const configured = Object.values(this.jobTimeouts).filter(
+            (ms): ms is number => typeof ms === 'number',
+        );
+        const longest = Math.max(this.defaultTimeoutMs, ...configured);
+        const needed = longest + STALE_RECOVERY_GRACE_MS;
+        if (needed <= LEGACY_RECLAIM_CEILING_MS) return;
+        console.warn(
+            `[Queue Worker] Longest job timeout (${longest}ms) plus the recovery grace ` +
+                `(${STALE_RECOVERY_GRACE_MS}ms) exceeds LEGACY_RECLAIM_CEILING_MS ` +
+                `(${LEGACY_RECLAIM_CEILING_MS}ms). Rows claimed before "lockUntil" existed ` +
+                `can be reclaimed while still running. Raise the ceiling above ${needed}ms.`,
+        );
     }
 
     /**
@@ -65,10 +166,11 @@ export class Worker {
         if (this.running) return;
         this.running = true;
         this.shuttingDown = false;
+        this.stopPromise = null;
         this.upSince = new Date();
         console.log('[Queue Worker] Started');
         this.registerSignalHandlers();
-        this.poll();
+        this.runPoll();
     }
 
     /**
@@ -76,6 +178,11 @@ export class Worker {
      * Waits for all active jobs to complete before resolving.
      */
     async stop(): Promise<void> {
+        // Signal handlers and the worker app can both request shutdown. Share
+        // the same drain promise so a second caller cannot observe
+        // `running=false`, return early, and disconnect Prisma/exit while the
+        // first caller is still waiting for active jobs.
+        if (this.stopPromise) return this.stopPromise;
         if (!this.running) return;
         this.shuttingDown = true;
         this.running = false;
@@ -87,21 +194,32 @@ export class Worker {
 
         this.removeSignalHandlers();
 
-        // Wait for active jobs to finish
-        if (this.activeJobs.size > 0) {
-            console.log(`[Queue Worker] Waiting for ${this.activeJobs.size} active jobs to complete...`);
-            await new Promise<void>((resolve) => {
-                this.shutdownResolve = resolve;
-                // Check immediately in case jobs finished between the check and setting the resolver
-                if (this.activeJobs.size === 0) {
-                    this.shutdownResolve = null;
-                    resolve();
-                }
-            });
-        }
+        this.stopPromise = (async () => {
+            // A poll may be between its running check and its atomic claim. Let
+            // that cycle finish before deciding whether the active set is
+            // drained, otherwise stop() can resolve just before it claims work.
+            await this.pollPromise;
 
-        this.upSince = null;
-        console.log('[Queue Worker] Stopped');
+            // Wait for active jobs to finish
+            if (this.activeJobs.size > 0) {
+                console.log(
+                    `[Queue Worker] Waiting for ${this.activeJobs.size} active jobs to complete...`,
+                );
+                await new Promise<void>((resolve) => {
+                    this.shutdownResolve = resolve;
+                    // Check immediately in case jobs finished between the check and setting the resolver
+                    if (this.activeJobs.size === 0) {
+                        this.shutdownResolve = null;
+                        resolve();
+                    }
+                });
+            }
+
+            this.upSince = null;
+            console.log('[Queue Worker] Stopped');
+        })();
+
+        return this.stopPromise;
     }
 
     /**
@@ -136,16 +254,53 @@ export class Worker {
         this.signalHandlers = [];
     }
 
+    private runPoll(): void {
+        const currentPoll = this.poll();
+        this.pollPromise = currentPoll;
+        void currentPoll.finally(() => {
+            if (this.pollPromise === currentPoll) this.pollPromise = null;
+        });
+    }
+
+    private schedulePoll(delayMs: number): void {
+        if (!this.running) return;
+        this.pollTimer = setTimeout(() => this.runPoll(), delayMs);
+    }
+
     private async poll(): Promise<void> {
         if (!this.running) return;
 
         try {
             this.lastPollTime = new Date();
+
+            // Ahead of the capacity check, not behind it. Reclaiming is a global
+            // sweep over abandoned rows and has nothing to do with this replica's
+            // spare capacity, so gating it on free slots would stop recovery
+            // exactly when the backlog that produced the abandoned rows is
+            // largest. Latent while `availableSlots` can never reach 0 (the poll
+            // awaits its whole batch, so `activeJobs` is empty here), and live the
+            // moment that changes.
+            //
+            // Isolated on purpose. It sits ahead of every claim in the same try,
+            // and `lastPollTime` is already set, so a failing reclaim would stop
+            // all claiming while the process stayed up. Note the health server in
+            // `apps/worker/src/index.ts` answers 200 unconditionally and never
+            // consults `healthCheck()`, so nothing would have restarted it either —
+            // that half is the /health rework's problem, not this catch's.
+            // Recovering abandoned work is a nice-to-have; claiming new work is the
+            // job.
+            try {
+                await this.reclaimStaleJobs();
+            } catch (error) {
+                console.error('[Queue Worker] Reclaim sweep failed, continuing:', error);
+            }
+            if (!this.running) return;
+
             const availableSlots = this.maxConcurrency - this.activeJobs.size;
 
             if (availableSlots <= 0) {
                 // At capacity, wait and retry
-                this.pollTimer = setTimeout(() => this.poll(), this.pollIntervalMs);
+                this.schedulePoll(this.pollIntervalMs);
                 return;
             }
 
@@ -162,13 +317,127 @@ export class Worker {
 
             // If we processed jobs, poll immediately for more
             const nextPollDelay = processedCount > 0 ? 0 : this.pollIntervalMs;
-            this.pollTimer = setTimeout(() => this.poll(), nextPollDelay);
+            this.schedulePoll(nextPollDelay);
         } catch (error) {
             console.error('[Queue Worker] Poll error:', error);
-            this.pollTimer = setTimeout(() => this.poll(), this.pollIntervalMs);
+            this.schedulePoll(this.pollIntervalMs);
+        }
+    }
+
+    /**
+     * Return abandoned PROCESSING jobs to the pending queue before claiming work.
+     *
+     * `lockUntil` is written with the database clock at claim time, so the stale
+     * comparison also uses the database clock, and it is the *claiming* worker's
+     * deadline rather than a window re-derived here. A recovery grace is still
+     * added on top: the normal timeout path must have time to release its own
+     * claim before another worker calls it crash-abandoned. A true abandonment
+     * consumes an attempt, clears its claim token, and moves toward DEAD_LETTER
+     * like every other failed execution — spaced by the same backoff.
+     *
+     * Deliberately not filtered to types this worker registers. `lockUntil` makes
+     * the row self-describing, so there is no longer anything a worker needs to
+     * know about a type in order to tell that its claim has expired — and the old
+     * type predicate meant a crashed claim of a type this replica does not handle
+     * stayed PROCESSING forever, never reclaimed and never dead-lettered.
+     *
+     * The deadline test is a disjunction rather than a `CASE` over `lockUntil`,
+     * because a `CASE` over the column is not sargable: the planner would take the
+     * `status` prefix of `Job_status_lockUntil_idx` and then filter every
+     * `PROCESSING` row. In the first arm `lockUntil` is kept bare with the interval
+     * arithmetic on the right, which is what lets the index's second column do any
+     * work at all.
+     *
+     * The legacy arm cannot use the index the same way — it discriminates on
+     * `lockedAt`, which is not in it — so it rides the `status` + `lockUntil IS
+     * NULL` prefix and filters. That is fine: the arm is dead after one rollout.
+     */
+    private async reclaimStaleJobs(): Promise<void> {
+        // `runAt` mirrors `calculateBackoff` in shared/utils.ts — base * 2^attempt
+        // plus jitter under a ceiling — because a reclaim and a handler failure
+        // both mean "this attempt did not finish, try again later" and must space
+        // retries the same way. Setting NOW() here let a crash-looping worker burn
+        // every attempt on a job back to back and drive it to DEAD_LETTER at full
+        // speed. The attempt number is the one being scheduled, `attempts + 1`,
+        // matching `handleFailure`.
+        const reclaimed = await prisma.$executeRaw`
+            UPDATE "Job" AS job
+            SET status = CASE
+                    WHEN job."attempts" + 1 >= job."maxAttempts"
+                    THEN 'DEAD_LETTER'::"JobStatus"
+                    ELSE 'PENDING'::"JobStatus"
+                END,
+                "attempts" = job."attempts" + 1,
+                "lockedAt" = NULL,
+                "lockUntil" = NULL,
+                "claimToken" = NULL,
+                progress = NULL,
+                error = 'Worker claim was abandoned before completion'
+                    || COALESCE(' (previous error: ' || left(job.error, 200) || ')', ''),
+                "completedAt" = CASE
+                    WHEN job."attempts" + 1 >= job."maxAttempts" THEN NOW()
+                    ELSE NULL
+                END,
+                "runAt" = CASE
+                    WHEN job."attempts" + 1 >= job."maxAttempts" THEN job."runAt"
+                    ELSE NOW() + (
+                        LEAST(
+                            ${BACKOFF_BASE_MS} * POWER(2, LEAST(job."attempts" + 1, 30))
+                                + random() * ${BACKOFF_BASE_MS},
+                            ${BACKOFF_MAX_MS}
+                        ) * INTERVAL '1 millisecond'
+                    )
+                END,
+                "updatedAt" = NOW()
+            WHERE job.status = 'PROCESSING'
+            AND (
+                (
+                    job."lockUntil" IS NOT NULL
+                    AND job."lockUntil"
+                        < NOW() - (${STALE_RECOVERY_GRACE_MS} * INTERVAL '1 millisecond')
+                )
+                OR (
+                    job."lockUntil" IS NULL
+                    AND job."lockedAt"
+                        < NOW() - (${LEGACY_RECLAIM_CEILING_MS} * INTERVAL '1 millisecond')
+                )
+                OR (
+                    job."lockUntil" IS NULL
+                    AND job."lockedAt" IS NULL
+                    AND job."updatedAt"
+                        < NOW() - (${LEGACY_RECLAIM_CEILING_MS} * INTERVAL '1 millisecond')
+                )
+            )
+        `;
+
+        // Every row this moved is a claim some worker took and never released —
+        // a crash, an OOM kill, an evicted container. It is the earliest signal
+        // that replicas are dying, and it was being computed once per poll and
+        // thrown away. Only spoken when non-zero: the healthy case is silence.
+        if (reclaimed > 0) {
+            console.warn(
+                `[Queue Worker] Reclaim sweep returned ${reclaimed} abandoned ` +
+                    `PROCESSING job(s) to the queue. Each is a claim a worker took ` +
+                    `and never released — check for crashed or OOM-killed replicas.`,
+            );
         }
     }
 
+    /**
+     * This worker's per-type claim durations, as a jsonb object for the claim SQL.
+     *
+     * Written onto the row at claim time so the deadline belongs to the execution
+     * that owns the claim. Deriving it at reclaim time from the *observing*
+     * worker's config meant a replica on an older revision — one without an entry
+     * for a long-running type, so falling back to `defaultTimeoutMs` — would
+     * reclaim a claim that was still live, and the original handler's eventual
+     * success would then be fenced out and silently discarded while the job ran
+     * a second time.
+     */
+    private timeoutMapJson(): string {
+        return JSON.stringify(this.jobTimeouts);
+    }
+
     /**
      * Claim jobs respecting per-type concurrency limits.
      * For each registered job type that has available capacity, claim up to
@@ -212,8 +481,7 @@ export class Worker {
             const jobs = await this.claimJobsForType(type, limit);
 
             if (jobs.length > 0) {
-                const promises = jobs.map((job) => this.processJob(job));
-                await Promise.allSettled(promises);
+                await this.processClaimedJobs(jobs);
                 totalProcessed += jobs.length;
                 remainingGlobalSlots -= jobs.length;
             }
@@ -225,29 +493,17 @@ export class Worker {
     /**
      * Claim pending jobs of a specific type using SKIP LOCKED.
      */
-    private async claimJobsForType(
-        type: string,
-        limit: number,
-    ): Promise<
-        Array<{
-            id: string;
-            type: string;
-            payload: unknown;
-            attempts: number;
-            maxAttempts: number;
-        }>
-    > {
-        return prisma.$queryRaw<
-            Array<{
-                id: string;
-                type: string;
-                payload: unknown;
-                attempts: number;
-                maxAttempts: number;
-            }>
-        >`
+    private async claimJobsForType(type: string, limit: number): Promise<Array<ClaimedJob>> {
+        return prisma.$queryRaw<Array<ClaimedJob>>`
             UPDATE "Job"
-            SET status = 'PROCESSING', "lockedAt" = NOW(), "updatedAt" = NOW()
+            SET status = 'PROCESSING', "lockedAt" = NOW(),
+                "lockUntil" = NOW() + (
+                    COALESCE(
+                        (${this.timeoutMapJson()}::jsonb ->> "Job".type)::double precision,
+                        ${this.defaultTimeoutMs}
+                    ) * INTERVAL '1 millisecond'
+                ),
+                "claimToken" = gen_random_uuid()::text, "updatedAt" = NOW()
             WHERE id IN (
                 SELECT id FROM "Job"
                 WHERE status = 'PENDING'
@@ -257,24 +513,23 @@ export class Worker {
                 LIMIT ${limit}
                 FOR UPDATE SKIP LOCKED
             )
-            RETURNING id, type, payload, attempts, "maxAttempts"
+            RETURNING id, type, payload, attempts, "maxAttempts", "claimToken"
         `;
     }
 
     private async claimAndProcessJobs(limit: number): Promise<number> {
         // Use raw query with SKIP LOCKED for safe concurrent job processing.
         // This atomically selects and locks pending jobs that are ready to run.
-        const jobs = await prisma.$queryRaw<
-            Array<{
-                id: string;
-                type: string;
-                payload: unknown;
-                attempts: number;
-                maxAttempts: number;
-            }>
-        >`
+        const jobs = await prisma.$queryRaw<Array<ClaimedJob>>`
             UPDATE "Job"
-            SET status = 'PROCESSING', "lockedAt" = NOW(), "updatedAt" = NOW()
+            SET status = 'PROCESSING', "lockedAt" = NOW(),
+                "lockUntil" = NOW() + (
+                    COALESCE(
+                        (${this.timeoutMapJson()}::jsonb ->> "Job".type)::double precision,
+                        ${this.defaultTimeoutMs}
+                    ) * INTERVAL '1 millisecond'
+                ),
+                "claimToken" = gen_random_uuid()::text, "updatedAt" = NOW()
             WHERE id IN (
                 SELECT id FROM "Job"
                 WHERE status = 'PENDING'
@@ -283,42 +538,85 @@ export class Worker {
                 LIMIT ${limit}
                 FOR UPDATE SKIP LOCKED
             )
-            RETURNING id, type, payload, attempts, "maxAttempts"
+            RETURNING id, type, payload, attempts, "maxAttempts", "claimToken"
         `;
 
-        // Process jobs concurrently (each tracked in activeJobs)
-        const promises = jobs.map((job: { id: string; type: string; payload: unknown; attempts: number; maxAttempts: number }) => this.processJob(job));
-        await Promise.allSettled(promises);
+        await this.processClaimedJobs(jobs);
 
         return jobs.length;
     }
 
-    private async processJob(job: {
-        id: string;
-        type: string;
-        payload: unknown;
-        attempts: number;
-        maxAttempts: number;
-    }): Promise<void> {
+    /**
+     * Run a claimed batch concurrently, and say something when one of them dies.
+     *
+     * `processJob` has a `finally` but no `catch`, so anything thrown outside its
+     * inner try — a Prisma error on the no-handler tombstone, on the dead-letter
+     * write, on the retry write — propagates out. `Promise.allSettled` then
+     * absorbs it and returns a result object nobody was reading, so a database
+     * blip could take out the whole failure-recording path and produce no output
+     * at all: not the Prisma error, and not the handler failure it was recording.
+     * The row survives (it stays PROCESSING for the sweep); the explanation did not.
+     */
+    private async processClaimedJobs(jobs: ClaimedJob[]): Promise<void> {
+        const settled = await Promise.allSettled(jobs.map((job) => this.processJob(job)));
+        settled.forEach((outcome, i) => {
+            if (outcome.status !== 'rejected') return;
+            const job = jobs[i];
+            console.error(
+                `[Queue Worker] processJob threw for job ${job.id} (${job.type}). ` +
+                    `The row is left PROCESSING for the reclaim sweep:`,
+                outcome.reason,
+            );
+        });
+    }
+
+    private async processJob(job: ClaimedJob): Promise<void> {
+        // `ClaimedJob.claimToken` is typed `string`, but it arrives from
+        // `$queryRaw`, which validates nothing — the type is an assertion about a
+        // column the schema declares nullable. That matters more than it reads:
+        // Prisma's `updateMany` silently DROPS a `where` key whose value is
+        // `undefined`, so a token that ever went missing would turn every fenced
+        // write below into an unfenced update-by-id, and the exactly-once
+        // mechanism would disappear with no error and no log. Refuse the row
+        // instead. It stays PROCESSING with its `lockUntil` intact, so the sweep
+        // recovers it on the normal deadline rather than it being lost.
+        if (!job.claimToken) {
+            console.error(
+                `[Queue Worker] Refusing job ${job.id} (${job.type}): the claim ` +
+                    `returned no token, so its writes could not be fenced. Leaving ` +
+                    `the row for the reclaim sweep. This is a bug in the claim query.`,
+            );
+            return;
+        }
+
         this.activeJobs.add(job.id);
-        this.activeJobsByType.set(
-            job.type,
-            (this.activeJobsByType.get(job.type) ?? 0) + 1,
-        );
+        this.activeJobsByType.set(job.type, (this.activeJobsByType.get(job.type) ?? 0) + 1);
 
         try {
             const handler = this.handlers.get(job.type);
 
             if (!handler) {
                 console.warn(`[Queue Worker] No handler for job type: ${job.type}`);
-                await prisma.job.update({
-                    where: { id: job.id },
+                const tombstone = await prisma.job.updateMany({
+                    where: {
+                        id: job.id,
+                        status: 'PROCESSING',
+                        claimToken: job.claimToken,
+                    },
                     data: {
                         status: 'FAILED',
+                        // Every other terminal path records the attempt it used.
+                        // Without this the row reads `attempts: 0` for a job that
+                        // was claimed and dispatched.
+                        attempts: job.attempts + 1,
                         error: `No handler registered for job type: ${job.type}`,
                         completedAt: new Date(),
+                        lockedAt: null,
+                        lockUntil: null,
+                        claimToken: null,
                     },
                 });
+                warnIfFenced(tombstone.count, job, 'no-handler failure');
                 return;
             }
 
@@ -328,32 +626,84 @@ export class Worker {
             // Build handler context
             const context: JobHandlerContext = {
                 jobId: job.id,
-                reportProgress: (percent: number) => updateJobProgress(job.id, percent),
+                reportProgress: (percent: number) =>
+                    updateJobProgress(job.id, percent, job.claimToken),
             };
 
+            // The handler's own try. Only what the handler does belongs in here:
+            // a throw from the bookkeeping below is a different kind of event and
+            // must not be laundered into "the job failed".
+            let result: JobResult;
             try {
-                const result = await this.runWithTimeout(
+                result = await this.runWithTimeout(
                     handler(job.payload as Record<string, never>, context),
                     timeoutMs,
                 );
-
-                if (result.success) {
-                    await prisma.job.update({
-                        where: { id: job.id },
-                        data: {
-                            status: 'COMPLETED',
-                            attempts: attempt,
-                            progress: 100,
-                            completedAt: new Date(),
-                            lockedAt: null,
-                        },
-                    });
-                } else {
-                    await this.handleFailure(job.id, attempt, job.maxAttempts, result.error ?? 'Unknown error');
-                }
             } catch (error) {
                 const errorMessage = error instanceof Error ? error.message : String(error);
-                await this.handleFailure(job.id, attempt, job.maxAttempts, errorMessage);
+                await this.handleFailure(
+                    job.id,
+                    job.type,
+                    job.claimToken,
+                    attempt,
+                    job.maxAttempts,
+                    errorMessage,
+                );
+                return;
+            }
+
+            if (!result.success) {
+                await this.handleFailure(
+                    job.id,
+                    job.type,
+                    job.claimToken,
+                    attempt,
+                    job.maxAttempts,
+                    result.error ?? 'Unknown error',
+                );
+                return;
+            }
+
+            // The work succeeded and its side effects have already happened. A
+            // throw from here is the database failing to record that, which is
+            // emphatically not a job failure: routing it through `handleFailure`
+            // set the row back to PENDING carrying a Prisma error as the job's
+            // error, and the retry re-ran every one of those side effects. So the
+            // row is left PROCESSING for the sweep instead — still a re-run, but
+            // one that is logged as what it is rather than recorded as a fault in
+            // the handler.
+            try {
+                const completion = await prisma.job.updateMany({
+                    where: {
+                        id: job.id,
+                        status: 'PROCESSING',
+                        claimToken: job.claimToken,
+                    },
+                    data: {
+                        status: 'COMPLETED',
+                        attempts: attempt,
+                        progress: 100,
+                        completedAt: new Date(),
+                        lockedAt: null,
+                        lockUntil: null,
+                        claimToken: null,
+                    },
+                });
+                // The single most important thing this fence can tell us: the work
+                // finished, and the row no longer accepts this claim. It was
+                // reclaimed while still live, so this success is being discarded
+                // and the row has been re-queued or dead-lettered. Whether the side
+                // effects actually ran twice depends on which of those happened,
+                // which is why the message says "may".
+                warnIfFenced(completion.count, job, 'completion');
+            } catch (error) {
+                console.error(
+                    `[Queue Worker] Job ${job.id} (${job.type}) succeeded but its ` +
+                        `COMPLETED write failed. The row stays PROCESSING and the ` +
+                        `reclaim sweep will re-queue it, which WILL run the handler ` +
+                        `again and repeat its side effects:`,
+                    error,
+                );
             }
         } finally {
             this.activeJobs.delete(job.id);
@@ -374,7 +724,10 @@ export class Worker {
     private async runWithTimeout<T>(promise: Promise<T>, timeoutMs: number): Promise<T> {
         let timer: ReturnType<typeof setTimeout>;
         const timeout = new Promise<never>((_resolve, reject) => {
-            timer = setTimeout(() => reject(new Error(`Job timed out after ${timeoutMs}ms`)), timeoutMs);
+            timer = setTimeout(
+                () => reject(new Error(`Job timed out after ${timeoutMs}ms`)),
+                timeoutMs,
+            );
         });
 
         try {
@@ -386,45 +739,69 @@ export class Worker {
 
     private async handleFailure(
         jobId: string,
+        jobType: string,
+        claimToken: string,
         attempt: number,
         maxAttempts: number,
         error: string,
     ): Promise<void> {
         if (attempt >= maxAttempts) {
             // Dead letter: job has exhausted all retries
-            await prisma.job.update({
-                where: { id: jobId },
+            const result = await prisma.job.updateMany({
+                where: { id: jobId, status: 'PROCESSING', claimToken },
                 data: {
                     status: 'DEAD_LETTER',
                     attempts: attempt,
                     error,
                     completedAt: new Date(),
                     lockedAt: null,
+                    lockUntil: null,
+                    claimToken: null,
                 },
             });
-            console.error(
-                `[Queue Worker] Job ${jobId} moved to dead letter queue after ${attempt} attempts: ${error}`,
-            );
+            if (result.count > 0) {
+                console.error(
+                    `[Queue Worker] Job ${jobId} (${jobType}) moved to dead letter queue after ${attempt} attempts: ${error}`,
+                );
+            } else {
+                // Pre-fence this always logged. Staying silent here would lose both
+                // the fence rejection and the failure that caused it.
+                console.warn(
+                    `[Queue Worker] Job ${jobId} (${jobType}) dead-letter write was fenced ` +
+                        `(claim ${claimToken} no longer owns the row); ` +
+                        `the failure it was recording was: ${error}`,
+                );
+            }
         } else {
             // Schedule retry with exponential backoff
             const backoffMs = calculateBackoff(attempt);
             const runAt = new Date(Date.now() + backoffMs);
 
-            await prisma.job.update({
-                where: { id: jobId },
+            const result = await prisma.job.updateMany({
+                where: { id: jobId, status: 'PROCESSING', claimToken },
                 data: {
                     status: 'PENDING',
                     attempts: attempt,
                     error,
                     runAt,
                     lockedAt: null,
+                    lockUntil: null,
+                    claimToken: null,
                     progress: null,
                 },
             });
-            console.warn(
-                `[Queue Worker] Job ${jobId} failed (attempt ${attempt}/${maxAttempts}), ` +
-                    `retrying at ${runAt.toISOString()}: ${error}`,
-            );
+            if (result.count > 0) {
+                console.warn(
+                    `[Queue Worker] Job ${jobId} (${jobType}) failed (attempt ${attempt}/${maxAttempts}), ` +
+                        `retrying at ${runAt.toISOString()}: ${error}`,
+                );
+            } else {
+                console.warn(
+                    `[Queue Worker] Job ${jobId} (${jobType}) retry write was fenced ` +
+                        `(claim ${claimToken} no longer owns the row); ` +
+                        `the failure it was recording was: ${error}`,
+                );
+            }
         }
     }
 }