diff --git a/apps/dev/docs/multi-db-verification.md b/apps/dev/docs/multi-db-verification.md
index 44116c91..8c0182de 100644
--- a/apps/dev/docs/multi-db-verification.md
+++ b/apps/dev/docs/multi-db-verification.md
@@ -54,6 +54,13 @@ is knowingly red (see below) and `&&` would stop the chain before Mongo ever ran
 `bootTestPayload` picks its adapter from `DB_ADAPTER` through `resolveTestDbAdapter()`, the test-side
 sibling of `resolveDbAdapter()` in `src/lib/database/resolveAdapter.ts`.
 
+**Two queue modes, not one.** `EXCLUSIVE_QUEUE=1` turns on Payload's `enableConcurrencyControl` for
+every boot, which is the setting a host enables to have the queue itself hold a second job for a
+document. `test:integration:all` runs the three adapters in both modes — six runs — because the
+plugin behaves differently under it: without the flag a second request extends the live job, with it
+the request gets a job of its own and waits. Two specs pin their own mode rather than following the
+env var (`locale-append` needs it off, `exclusive-queue` needs it on); everything else runs in both.
+
 **Isolation.** Each boot needs a database of its own: the twelve spec files run serially but share
 one server, so without it they would read each other's rows. SQLite gets this for free — a throwaway
 file per boot. Postgres and MongoDB do not, so each boot is given a namespace named by a random run
diff --git a/apps/dev/package.json b/apps/dev/package.json
index a49e6565..fb459c71 100644
--- a/apps/dev/package.json
+++ b/apps/dev/package.json
@@ -28,8 +28,12 @@
     "test:integration:sqlite": "cross-env DB_ADAPTER=sqlite bun run test:integration",
     "test:integration:postgres": "cross-env DB_ADAPTER=postgres bun run test:integration",
     "test:integration:mongo": "cross-env DB_ADAPTER=mongo bun run test:integration",
-    "test:integration:all": "bun run test:integration:sqlite; bun run test:integration:postgres; bun run test:integration:mongo",
-    "check-types": "tsgo --noEmit -p tsconfig.check.json"
+    "test:integration:all": "bun run test:integration:sqlite; bun run test:integration:postgres; bun run test:integration:mongo; bun run test:integration:exclusive:sqlite; bun run test:integration:exclusive:postgres; bun run test:integration:exclusive:mongo",
+    "check-types": "tsgo --noEmit -p tsconfig.check.json",
+    "test:integration:exclusive": "cross-env EXCLUSIVE_QUEUE=1 bun run test:integration",
+    "test:integration:exclusive:sqlite": "cross-env DB_ADAPTER=sqlite bun run test:integration:exclusive",
+    "test:integration:exclusive:postgres": "cross-env DB_ADAPTER=postgres bun run test:integration:exclusive",
+    "test:integration:exclusive:mongo": "cross-env DB_ADAPTER=mongo bun run test:integration:exclusive"
   },
   "dependencies": {
     "@focus-reactive/payload-plugin-ab": "workspace:*",
diff --git a/apps/dev/src/integration/translator/bootTestPayload.ts b/apps/dev/src/integration/translator/bootTestPayload.ts
index 041dcae6..48ad5be3 100644
--- a/apps/dev/src/integration/translator/bootTestPayload.ts
+++ b/apps/dev/src/integration/translator/bootTestPayload.ts
@@ -22,6 +22,9 @@ import { createTestDatabase } from "../../lib/database/resolveAdapter";
 import { reverseComplete } from "../../lib/translator/fakeComplete";
 import { buildTestCollections } from "./testCollections";
 
+/** Payload's `autoRun.limit` default — these specs reproduce the cron's batching, not a run of one. */
+export const CRON_BATCH_LIMIT = 50;
+
 /**
  * A booted test Payload plus the throwaway resources to tear down after the suite.
  */
@@ -49,6 +52,8 @@ export type TestPayload = {
  *   `runId` (see `resolveTestDbAdapter`), so schema `push` is a clean CREATE with no data-loss branch
  *   — Payload never drops to the interactive "accept data loss?" prompt that would hang an
  *   unattended/headless run. `cleanup()` drops the namespace and removes the temp dir even on failure.
+ * - **One boot per process:** `getPayload` caches, so a second `bootTestPayload` in the same spec
+ *   file returns the first — a case that needs its own boot needs its own file.
  * - **Sync runner:** a translation runs INLINE inside the triggering `afterChange`, so it is complete
  *   when the awaited `payload.update`/`create` resolves — no job autorun, no polling, no async race
  *   in the specs.
@@ -60,8 +65,13 @@ export type TestPayload = {
  *   publish; the enqueue route still works.
  * @param opts.collections - replaces the shared fixture set entirely (not merged). The set must
  *   still contain a `docs` collection when `autoTranslate` is passed.
+ * @param opts.failFor - target locales the fake provider should throw for, so a spec can exercise a
+ *   partial failure. Every other locale translates normally.
  * @param opts.runner - defaults to the sync runner. `createPayloadJobsRunner({ autoRun: false })`
  *   leaves queued jobs unprocessed in `payload-jobs`, so a spec can read the rows.
+ * @param opts.onTranslate - awaited before each provider call, so a spec can hold a locale mid-run.
+ * @param opts.exclusiveQueue - Payload's `enableConcurrencyControl` for this boot; defaults to
+ *   `EXCLUSIVE_QUEUE=1`.
  * @param opts.fallback - localization fallback, off by default: an unwritten locale reads as
  *   empty, not as the default locale's text. Localization-level, so it applies to the whole boot.
  */
@@ -69,6 +79,9 @@ export async function bootTestPayload(opts?: {
   autoTranslate?: { targets: string[]; strategy?: "overwrite" | "skip_existing" };
   collections?: CollectionConfig[];
   fallback?: boolean;
+  exclusiveQueue?: boolean;
+  failFor?: string[];
+  onTranslate?: (targetLng: string) => Promise<void> | void;
   runner?: TaskRunnerProvider;
 }): Promise<TestPayload> {
   const dir = mkdtempSync(join(tmpdir(), "translator-int-"));
@@ -84,11 +97,14 @@ export async function bootTestPayload(opts?: {
     : collections;
 
   const baseProvider = createTranslationProvider({ complete: reverseComplete });
+  const failFor = new Set(opts?.failFor);
   let translateCalls = 0;
   const countingProvider: TranslationProvider = {
-    translate: (input, sourceLng, targetLng) => {
+    translate: async (input, sourceLng, targetLng) => {
       translateCalls += 1;
-      return baseProvider.translate(input, sourceLng, targetLng);
+      await opts?.onTranslate?.(targetLng);
+      if (failFor.has(targetLng)) throw new Error(`provider unavailable for ${targetLng}`);
+      return await baseProvider.translate(input, sourceLng, targetLng);
     },
   };
 
@@ -107,9 +123,17 @@ export async function bootTestPayload(opts?: {
         { code: "en", label: "English" },
         { code: "de", label: "Deutsch" },
         { code: "fr", label: "Français" },
+        // Three targets, not two: with two, the failing locale is always the last and "stopped at
+        // the failure" is unobservable.
+        { code: "es", label: "Español" },
       ],
     },
     collections,
+    jobs: {
+      // Payload deletes completed jobs by default, leaving the status panels nothing to read.
+      deleteJobOnComplete: false,
+      enableConcurrencyControl: opts?.exclusiveQueue ?? process.env.EXCLUSIVE_QUEUE === "1",
+    },
     plugins: [
       translatorPlugin({
         collections: managed,
diff --git a/apps/dev/src/integration/translator/exclusive-queue.int.test.ts b/apps/dev/src/integration/translator/exclusive-queue.int.test.ts
new file mode 100644
index 00000000..ac6cfe91
--- /dev/null
+++ b/apps/dev/src/integration/translator/exclusive-queue.int.test.ts
@@ -0,0 +1,138 @@
+import { createPayloadJobsRunner } from "@focus-reactive/payload-plugin-translator";
+import { afterAll, beforeAll, describe, expect, it } from "vitest";
+
+import { bootTestPayload, CRON_BATCH_LIMIT } from "./bootTestPayload";
+import type { TestPayload } from "./bootTestPayload";
+import { callEndpoint } from "./callEndpoint";
+
+// Its own file: the setting is fixed at boot, and a boot is per process (see `bootTestPayload`).
+
+type RunResult = { jobStatus?: Record<string, unknown> };
+
+type Job = {
+  processing?: boolean;
+  input?: { collection_id?: string; target_lngs?: string[] };
+};
+
+let ctx: TestPayload;
+let release: (() => void) | undefined;
+let reached: (() => void) | undefined;
+let held: Promise<void>;
+
+const armBarrier = () => {
+  held = new Promise<void>((resolve) => {
+    reached = resolve;
+  });
+};
+
+beforeAll(async () => {
+  armBarrier();
+  ctx = await bootTestPayload({
+    exclusiveQueue: true,
+    runner: createPayloadJobsRunner({ autoRun: false }),
+    onTranslate: async (targetLng) => {
+      if (targetLng !== "de") return;
+      reached?.();
+      await new Promise<void>((resolve) => {
+        release = resolve;
+      });
+    },
+  });
+});
+afterAll(async () => {
+  release?.();
+  await ctx?.cleanup();
+});
+
+const createDoc = async (title: string) => {
+  const doc = await ctx.payload.create({
+    collection: "docs" as "pages",
+    locale: "en",
+    data: { title, _status: "published" } as never,
+  });
+  return String(doc.id);
+};
+
+const enqueue = (id: string, targets: string[]) =>
+  callEndpoint(ctx.payload, "post", "/translate/enqueue", {
+    body: {
+      source_lng: "en",
+      target_lng: targets,
+      collection_slug: "docs",
+      collection_id: [id],
+      strategy: "overwrite",
+      publish_on_translation: false,
+    },
+  });
+
+const jobsFor = async (documentId: string) => {
+  const { docs } = await ctx.payload.find({
+    collection: "payload-jobs" as "pages",
+    pagination: false,
+    where: { workflowSlug: { equals: "translate_document_locales" } } as never,
+  });
+  return (docs as Job[]).filter((job) => job.input?.collection_id === documentId);
+};
+
+const runQueue = () =>
+  ctx.payload.jobs.run({ queue: "translations", limit: CRON_BATCH_LIMIT }) as Promise<RunResult>;
+
+const titleIn = async (id: string, locale: string) =>
+  (
+    (await ctx.payload.findByID({
+      collection: "docs" as "pages",
+      id,
+      locale: locale as "en",
+      fallbackLocale: false,
+      draft: true,
+    })) as Record<string, unknown>
+  ).title;
+
+describe("with the host's concurrency control on", () => {
+  it("holds a second job for the same document, then runs it and loses nothing", async () => {
+    armBarrier();
+    const id = await createDoc("Exclusive source");
+
+    await enqueue(id, ["de"]);
+    const first = runQueue();
+    await held;
+
+    await enqueue(id, ["fr"]);
+
+    // The picker check below also passes when there was nothing to take, so first prove the second
+    // job exists.
+    const queued = await jobsFor(id);
+    expect(queued.length, "the request did not get a job of its own").toBe(2);
+    expect(
+      queued.find((job) => !job.processing)?.input?.target_lngs,
+      "the second job should carry only the locale that was asked for"
+    ).toEqual(["fr"]);
+
+    const whileRunning = await runQueue();
+    expect(
+      Object.keys(whileRunning.jobStatus ?? {}),
+      "the picker took a second job for a document already being written"
+    ).toEqual([]);
+
+    release?.();
+    await first;
+    await runQueue();
+
+    expect(await titleIn(id, "de"), "de was lost").toBe("ecruos evisulcxE");
+    expect(await titleIn(id, "fr"), "fr was lost").toBe("ecruos evisulcxE");
+  });
+
+  it("still runs jobs for different documents together", async () => {
+    const first = await createDoc("Doc one");
+    const second = await createDoc("Doc two");
+    await enqueue(first, ["fr"]);
+    await enqueue(second, ["fr"]);
+
+    const batch = await runQueue();
+
+    expect(
+      Object.keys(batch.jobStatus ?? {}).length,
+      "two documents were serialized against each other"
+    ).toBe(2);
+  });
+});
diff --git a/apps/dev/src/integration/translator/job-supersede.int.test.ts b/apps/dev/src/integration/translator/job-extend.int.test.ts
similarity index 50%
rename from apps/dev/src/integration/translator/job-supersede.int.test.ts
rename to apps/dev/src/integration/translator/job-extend.int.test.ts
index dd733310..16481d49 100644
--- a/apps/dev/src/integration/translator/job-supersede.int.test.ts
+++ b/apps/dev/src/integration/translator/job-extend.int.test.ts
@@ -1,27 +1,24 @@
 import { createPayloadJobsRunner } from "@focus-reactive/payload-plugin-translator";
 import { afterAll, beforeAll, describe, expect, it } from "vitest";
 
-import { bootTestPayload } from "./bootTestPayload";
+import { bootTestPayload, CRON_BATCH_LIMIT } from "./bootTestPayload";
 import type { TestPayload } from "./bootTestPayload";
 import { callEndpoint } from "./callEndpoint";
 
-// Rows are counted, never identified by id: SQLite reuses the rowid of a deleted row (integer primary
-// key, no AUTOINCREMENT), so a replacement job can arrive carrying the deleted job's id and an
-// id-based assertion would pass for the wrong reason.
-
 type Job = {
   id: string | number;
   completedAt?: string | null;
-  input?: { target_lng?: string; collection_id?: string };
+  processing?: boolean;
+  input?: { collection_id?: string; target_lngs?: string[] };
 };
 
 let ctx: TestPayload;
 
-const enqueue = async (id: string, target = "de") => {
+const enqueue = async (id: string, targets: string[] = ["de"]) => {
   const res = await callEndpoint(ctx.payload, "post", "/translate/enqueue", {
     body: {
       source_lng: "en",
-      target_lng: target,
+      target_lng: targets,
       collection_slug: "docs",
       collection_id: [id],
       strategy: "overwrite",
@@ -29,15 +26,15 @@ const enqueue = async (id: string, target = "de") => {
     },
   });
   expect(res.status, "the enqueue endpoint rejected the request").toBe(200);
-  return (res.data as { data: { queued: number } }).data.queued;
+  const body = res.data as { data: { queued: number } };
+  return body.data.queued;
 };
 
-// The cases share one boot, so the table also holds every earlier case's jobs.
 const jobs = async (documentId: string): Promise<Job[]> => {
   const { docs } = await ctx.payload.find({
     collection: "payload-jobs" as "pages",
     pagination: false,
-    where: { taskSlug: { equals: "translate_document" } } as never,
+    where: { workflowSlug: { equals: "translate_document_locales" } } as never,
   });
   return (docs as Job[]).filter((j) => j.input?.collection_id === documentId);
 };
@@ -65,44 +62,60 @@ afterAll(async () => {
   await ctx?.cleanup();
 });
 
-describe("re-enqueue supersedes unfinished work, not finished work", () => {
-  it("keeps a finished job when the same locale is translated again", async () => {
+describe("a second request extends the live job rather than replacing it", () => {
+  it("adds its locales to a job that has not started", async () => {
     const id = await createDoc();
-    expect(await enqueue(id), "fixture: the first enqueue queued a job").toBe(1);
-
-    const [first] = await jobs(id);
-    await markFinished(first.id);
-
-    expect(await enqueue(id), "the re-enqueue queued nothing").toBe(1);
+    await enqueue(id, ["de", "fr"]);
+    await enqueue(id, ["es"]);
 
-    const after = await jobs(id);
-    expect(after.filter((j) => j.completedAt).length, "the finished job was deleted").toBe(1);
-    expect(after.length, "finished job plus the new one").toBe(2);
+    const live = await jobs(id);
+    expect(live.length, "the second request queued a job of its own").toBe(1);
+    expect(live[0].input?.target_lngs, "the locales already owed were dropped").toEqual([
+      "de",
+      "fr",
+      "es",
+    ]);
   });
 
-  it("still supersedes an unfinished job for the same locale", async () => {
+  it("translates every locale the extended job accumulated", async () => {
     const id = await createDoc();
-    await enqueue(id);
-    expect((await jobs(id)).length, "fixture").toBe(1);
+    await enqueue(id, ["de", "fr"]);
+    await enqueue(id, ["es"]);
+
+    await ctx.payload.jobs.run({ queue: "translations", limit: CRON_BATCH_LIMIT });
+
+    for (const locale of ["de", "fr", "es"]) {
+      const doc = (await ctx.payload.findByID({
+        collection: "docs" as "pages",
+        id,
+        locale: locale as "en",
+        fallbackLocale: false,
+        draft: true,
+      })) as Record<string, unknown>;
+      expect(doc.title, `${locale} was not translated`).toBe("crS");
+    }
+  });
 
-    expect(await enqueue(id), "the re-enqueue queued nothing").toBe(1);
+  it("keeps a locale the live job already owes from being listed twice", async () => {
+    const id = await createDoc();
+    await enqueue(id, ["de"]);
+    await enqueue(id, ["de"]);
 
-    expect((await jobs(id)).length, "the unfinished job was not superseded").toBe(1);
+    const live = await jobs(id);
+    expect(live.length).toBe(1);
+    expect(live[0].input?.target_lngs, "the locale was queued a second time").toEqual(["de"]);
   });
 
-  it("leaves another locale's unfinished job alone", async () => {
+  it("leaves a finished job alone and gives the new request its own", async () => {
     const id = await createDoc();
-    await enqueue(id, "de");
-    await enqueue(id, "fr");
-    expect((await jobs(id)).length, "fixture: one job per locale").toBe(2);
+    await enqueue(id, ["de"]);
+    const [first] = await jobs(id);
+    await markFinished(first.id);
 
-    expect(await enqueue(id, "de"), "the re-enqueue queued nothing").toBe(1);
+    await enqueue(id, ["fr"]);
 
     const after = await jobs(id);
-    expect(after.length, "the fr job was collateral damage").toBe(2);
-    expect(after.map((j) => j.input?.target_lng).sort(), "one job per locale, still").toEqual([
-      "de",
-      "fr",
-    ]);
+    expect(after.filter((j) => j.completedAt).length, "the finished job was touched").toBe(1);
+    expect(after.length, "finished job plus the new one").toBe(2);
   });
 });
diff --git a/apps/dev/src/integration/translator/locale-append.int.test.ts b/apps/dev/src/integration/translator/locale-append.int.test.ts
new file mode 100644
index 00000000..42f56cf1
--- /dev/null
+++ b/apps/dev/src/integration/translator/locale-append.int.test.ts
@@ -0,0 +1,126 @@
+import { createPayloadJobsRunner } from "@focus-reactive/payload-plugin-translator";
+import { afterAll, beforeAll, describe, expect, it } from "vitest";
+
+import { bootTestPayload, CRON_BATCH_LIMIT } from "./bootTestPayload";
+import type { TestPayload } from "./bootTestPayload";
+import { callEndpoint } from "./callEndpoint";
+
+// Guards an undocumented Payload behaviour: after each task settles it re-reads the job row onto the
+// live `job` object, which is how a locale appended mid-run reaches the handler. Measured on 3.84.1;
+// peer floor is ^3.76.0.
+
+type Job = {
+  id: string | number;
+  completedAt?: string | null;
+  input?: { target_lngs?: string[] };
+  log?: Array<{ state: string; input?: { target_lng?: string } }>;
+};
+
+const rev = (value: string) => [...value].reverse().join("");
+
+let ctx: TestPayload;
+let held: (() => void) | undefined;
+let heldStarted: (() => void) | undefined;
+const heldReached = new Promise<void>((resolve) => {
+  heldStarted = resolve;
+});
+
+beforeAll(async () => {
+  ctx = await bootTestPayload({
+    // Pinned off: a running job is only extended without the host's concurrency control — the other
+    // mode is `exclusive-queue.int.test.ts`.
+    exclusiveQueue: false,
+    runner: createPayloadJobsRunner({ autoRun: false }),
+    onTranslate: async (targetLng) => {
+      if (targetLng !== "fr") return;
+      heldStarted?.();
+      await new Promise<void>((resolve) => {
+        held = resolve;
+      });
+    },
+  });
+});
+afterAll(async () => {
+  held?.();
+  await ctx?.cleanup();
+});
+
+const readJob = async (): Promise<Job> => {
+  const { docs } = await ctx.payload.find({
+    collection: "payload-jobs" as "pages",
+    pagination: false,
+    where: { workflowSlug: { equals: "translate_document_locales" } } as never,
+  });
+  return docs[0] as Job;
+};
+
+describe("adding a locale to a running job", () => {
+  it("reaches the handler, and the log written before it survives", async () => {
+    const doc = await ctx.payload.create({
+      collection: "docs" as "pages",
+      locale: "en",
+      data: { title: "Append source", _status: "published" } as never,
+    });
+    const id = String(doc.id);
+
+    await callEndpoint(ctx.payload, "post", "/translate/enqueue", {
+      body: {
+        source_lng: "en",
+        target_lng: ["de", "fr"],
+        collection_slug: "docs",
+        collection_id: [id],
+        strategy: "overwrite",
+        publish_on_translation: false,
+      },
+    });
+
+    const run = ctx.payload.jobs.run({ queue: "translations", limit: CRON_BATCH_LIMIT });
+    await heldReached;
+
+    // The surviving-log claim is vacuous unless the log is non-empty when the second request lands.
+    const before = await readJob();
+    expect(
+      (before.log ?? []).map((e) => e.input?.target_lng),
+      "fixture: de should already be logged when the second request lands"
+    ).toEqual(["de"]);
+
+    await callEndpoint(ctx.payload, "post", "/translate/enqueue", {
+      body: {
+        source_lng: "en",
+        target_lng: ["es"],
+        collection_slug: "docs",
+        collection_id: [id],
+        strategy: "overwrite",
+        publish_on_translation: false,
+      },
+    });
+
+    held?.();
+    await run;
+
+    const after = await readJob();
+    const logged = (after.log ?? []).map((e) => [e.input?.target_lng, e.state]);
+
+    const esTitle = (
+      (await ctx.payload.findByID({
+        collection: "docs" as "pages",
+        id,
+        locale: "es" as "en",
+        fallbackLocale: false,
+        draft: true,
+      })) as Record<string, unknown>
+    ).title;
+
+    expect(
+      logged.map((l) => l[0]),
+      "the appended locale never ran"
+    ).toEqual(["de", "fr", "es"]);
+    expect(esTitle, "the appended locale was not translated").toBe(rev("Append source"));
+
+    const { totalDocs } = await ctx.payload.count({
+      collection: "payload-jobs" as "pages",
+      where: { workflowSlug: { equals: "translate_document_locales" } } as never,
+    });
+    expect(totalDocs, "a second job was queued alongside the running one").toBe(1);
+  });
+});
diff --git a/apps/dev/src/integration/translator/locale-workflow-failure.int.test.ts b/apps/dev/src/integration/translator/locale-workflow-failure.int.test.ts
new file mode 100644
index 00000000..56e5c34f
--- /dev/null
+++ b/apps/dev/src/integration/translator/locale-workflow-failure.int.test.ts
@@ -0,0 +1,113 @@
+import { createPayloadJobsRunner } from "@focus-reactive/payload-plugin-translator";
+import { afterAll, beforeAll, describe, expect, it } from "vitest";
+
+import { bootTestPayload, CRON_BATCH_LIMIT } from "./bootTestPayload";
+import type { TestPayload } from "./bootTestPayload";
+import { callEndpoint } from "./callEndpoint";
+
+// Its own file: the failing provider is fixed at boot, and a boot is per process (see
+// `bootTestPayload`).
+
+const rev = (s: string) => [...s].reverse().join("");
+
+let failing: TestPayload;
+
+beforeAll(async () => {
+  failing = await bootTestPayload({
+    runner: createPayloadJobsRunner({ autoRun: false }),
+    failFor: ["fr"],
+  });
+});
+afterAll(async () => {
+  await failing?.cleanup();
+});
+
+describe("when one locale's provider fails", () => {
+  it("stops there, leaves the locales after it untouched, and resumes on retry", async () => {
+    const source = "Partial source";
+    const doc = await failing.payload.create({
+      collection: "docs" as "pages",
+      locale: "en",
+      data: { title: source, _status: "published" } as never,
+    });
+    const id = String(doc.id);
+
+    const res = await callEndpoint(failing.payload, "post", "/translate/enqueue", {
+      body: {
+        source_lng: "en",
+        target_lng: ["de", "fr", "es"],
+        collection_slug: "docs",
+        collection_id: [id],
+        strategy: "overwrite",
+        publish_on_translation: false,
+      },
+    });
+    expect(res.status).toBe(200);
+
+    await failing.payload.jobs.run({ queue: "translations", limit: CRON_BATCH_LIMIT });
+
+    const read = async (locale: string) =>
+      (
+        (await failing.payload.findByID({
+          collection: "docs" as "pages",
+          id,
+          locale: locale as "en",
+          fallbackLocale: false,
+          draft: true,
+        })) as Record<string, unknown>
+      ).title;
+
+    expect(await read("de"), "the locale before the failure should have landed").toBe(rev(source));
+    expect(await read("fr"), "the failing locale should not have landed").toBeUndefined();
+    expect(await read("es"), "the locale after the failure should be untouched").toBeUndefined();
+
+    const { docs } = await failing.payload.find({
+      collection: "payload-jobs" as "pages",
+      pagination: false,
+      where: { workflowSlug: { equals: "translate_document_locales" } } as never,
+    });
+    const log = (docs[0] as { log?: Array<{ state: string; input?: { target_lng?: string } }> })
+      .log;
+    expect(
+      log?.map((entry) => [entry.input?.target_lng, entry.state]),
+      "the job should record which locale failed, and attempt no locale after it"
+    ).toEqual([
+      ["de", "succeeded"],
+      ["fr", "failed"],
+    ]);
+
+    const status = await callEndpoint(
+      failing.payload,
+      "get",
+      "/translate/document/:collection_slug/:collection_id",
+      { routeParams: { collection_slug: "docs", collection_id: id } }
+    );
+    const rows =
+      (status.data as { data?: Array<{ status: string; input: { target_lng: string } }> }).data ??
+      [];
+    expect(
+      rows.map((row) => [row.input.target_lng, row.status]).sort(),
+      "each locale should report its own outcome"
+    ).toEqual([
+      ["de", "completed"],
+      ["es", "pending"],
+      ["fr", "failed"],
+    ]);
+
+    // Payload backs a failed job off exponentially, so the picker would skip it for the next few
+    // seconds. Clearing the delay lets the retry happen now rather than making the spec wait.
+    await failing.payload.update({
+      collection: "payload-jobs" as "pages",
+      id: (docs[0] as { id: string | number }).id,
+      data: { waitUntil: null, processing: false } as never,
+    });
+
+    const before = failing.translateCount();
+    await failing.payload.jobs.run({ queue: "translations", limit: CRON_BATCH_LIMIT });
+
+    expect(
+      failing.translateCount() - before,
+      "the retry should attempt only the locale that failed"
+    ).toBe(1);
+  });
+});
diff --git a/apps/dev/src/integration/translator/locale-workflow.int.test.ts b/apps/dev/src/integration/translator/locale-workflow.int.test.ts
new file mode 100644
index 00000000..9f0c1acc
--- /dev/null
+++ b/apps/dev/src/integration/translator/locale-workflow.int.test.ts
@@ -0,0 +1,145 @@
+import { createPayloadJobsRunner } from "@focus-reactive/payload-plugin-translator";
+import { afterAll, beforeAll, describe, expect, it } from "vitest";
+
+import { bootTestPayload, CRON_BATCH_LIMIT } from "./bootTestPayload";
+import type { TestPayload } from "./bootTestPayload";
+import { callEndpoint } from "./callEndpoint";
+
+// Must boot the real jobs runner: `createSyncRunner` translates inline and in order, so the fan-out
+// this file guards against cannot occur under it.
+
+const rev = (s: string) => [...s].reverse().join("");
+
+let ctx: TestPayload;
+
+beforeAll(async () => {
+  ctx = await bootTestPayload({ runner: createPayloadJobsRunner({ autoRun: false }) });
+});
+afterAll(async () => {
+  await ctx?.cleanup();
+});
+
+const createDoc = async (title: string) => {
+  const doc = await ctx.payload.create({
+    collection: "docs" as "pages",
+    locale: "en",
+    data: { title, _status: "published" } as never,
+  });
+  return String(doc.id);
+};
+
+const enqueue = async (id: string, targets: string[]) => {
+  const res = await callEndpoint(ctx.payload, "post", "/translate/enqueue", {
+    body: {
+      source_lng: "en",
+      target_lng: targets,
+      collection_slug: "docs",
+      collection_id: [id],
+      strategy: "overwrite",
+      publish_on_translation: false,
+    },
+  });
+  expect(res.status, "the enqueue endpoint rejected the request").toBe(200);
+};
+
+const runQueue = () => ctx.payload.jobs.run({ queue: "translations", limit: CRON_BATCH_LIMIT });
+
+const titleIn = async (id: string, locale: string) =>
+  (
+    (await ctx.payload.findByID({
+      collection: "docs" as "pages",
+      id,
+      locale: locale as "en",
+      fallbackLocale: false,
+      draft: true,
+    })) as Record<string, unknown>
+  ).title;
+
+const workflowJob = async (id: string) => {
+  const { docs } = await ctx.payload.find({
+    collection: "payload-jobs" as "pages",
+    pagination: false,
+    where: { workflowSlug: { equals: "translate_document_locales" } } as never,
+  });
+  return (
+    docs as Array<{
+      completedAt?: string | null;
+      input?: { collection_id?: string };
+      log?: Array<Record<string, unknown>>;
+    }>
+  ).find((j) => j.input?.collection_id === id);
+};
+
+describe("translating one document into several locales", () => {
+  it("translates every requested locale, not just one", async () => {
+    const source = "Multi source";
+    const id = await createDoc(source);
+
+    await enqueue(id, ["de", "fr"]);
+    await runQueue();
+
+    expect(await titleIn(id, "de"), "de was not translated").toBe(rev(source));
+    expect(await titleIn(id, "fr"), "fr was not translated").toBe(rev(source));
+  });
+
+  it("runs the locales one after another, never overlapping", async () => {
+    const id = await createDoc("Ordered source");
+
+    await enqueue(id, ["de", "fr", "es"]);
+    await runQueue();
+
+    const job = await workflowJob(id);
+    expect(job, "no workflow job was written").toBeDefined();
+    const log = job?.log ?? [];
+    expect(
+      log.map((e) => (e.input as { target_lng?: string })?.target_lng),
+      "the log should carry a row per locale, in request order"
+    ).toEqual(["de", "fr", "es"]);
+    expect(log.map((e) => e.state)).toEqual(["succeeded", "succeeded", "succeeded"]);
+
+    // The order assertion above would also pass under `Promise.all`; only non-overlap rules it out.
+    for (let i = 1; i < log.length; i++) {
+      expect(
+        Date.parse(log[i].executedAt as string),
+        `locale ${i} started before locale ${i - 1} finished`
+      ).toBeGreaterThanOrEqual(Date.parse(log[i - 1].completedAt as string));
+    }
+  });
+
+  it("the status endpoint still reports a row per target locale", async () => {
+    const id = await createDoc("Status source");
+
+    await enqueue(id, ["de", "fr"]);
+    await runQueue();
+
+    const res = await callEndpoint(
+      ctx.payload,
+      "get",
+      "/translate/document/:collection_slug/:collection_id",
+      {
+        routeParams: { collection_slug: "docs", collection_id: id },
+      }
+    );
+    expect(res.status).toBe(200);
+    const rows = (res.data as { data?: Array<{ input?: { target_lng?: string } }> }).data ?? [];
+    expect(
+      rows.map((r) => r.input?.target_lng).sort(),
+      "the panel lost its per-locale detail"
+    ).toEqual(["de", "fr"]);
+  });
+
+  it("does not re-run a workflow that already completed", async () => {
+    const id = await createDoc("Idempotent source");
+
+    await enqueue(id, ["de", "fr"]);
+    await runQueue();
+
+    // A *failed* job is skipped on the second run too (backoff), so prove it completed first.
+    expect((await workflowJob(id))?.completedAt, "the workflow did not complete").toBeTruthy();
+
+    const before = ctx.translateCount();
+    await runQueue();
+
+    expect(ctx.translateCount() - before, "a completed workflow was run again").toBe(0);
+  });
+});
diff --git a/packages/payload-plugin-translator/README.md b/packages/payload-plugin-translator/README.md
index dce52dd1..b43d8754 100644
--- a/packages/payload-plugin-translator/README.md
+++ b/packages/payload-plugin-translator/README.md
@@ -562,6 +562,43 @@ createPayloadJobsRunner({ taskName: "translate_document", queueName: "translatio
 
 > By default Payload deletes a job as soon as it completes, so the "Completed" status never shows in the UI. Set `jobs: { deleteJobOnComplete: false }` in your Payload config to keep it.
 
+##### One job per document
+
+A document's target locales are queued as a **single job** that translates them one after another.
+Every write Payload makes is a whole-document version snapshot, so two locales translated in parallel
+build from the same base and the second silently drops the first's work.
+
+A later request for the same document adds its locales to that job rather than replacing it — the
+locales the job still owes are never lost. Two cases get a job of their own instead: re-translating a
+locale the live job has already finished (its log records it as done, so it would be skipped), and a
+request that picked a different source locale, strategy or publish flag — a job carries one of each
+for all its locales, so it cannot take work that chose differently.
+
+##### Optional: strict one-at-a-time per document
+
+Two requests landing at the same instant, or a re-translation of an already-finished locale, can still
+put two jobs on one document. If your content is edited often enough for that to matter, enable
+Payload's own concurrency control:
+
+```typescript
+// payload.config.ts
+export default buildConfig({
+  jobs: { enableConcurrencyControl: true },
+  // ...
+});
+```
+
+The plugin picks this up on its own — there is no option to set here. With it on, the queue holds a
+second job for a document until the running one finishes, so two jobs can never write the same
+document at once. Jobs for *different* documents still run in parallel.
+
+The cost is yours to weigh: the setting adds an indexed `concurrencyKey` column to the jobs
+collection, so a SQL database needs a migration (`payload migrate:create` then `payload migrate`);
+MongoDB needs none. A second job also waits for the next queue run rather than starting immediately.
+
+> With the setting on, a job stuck at `processing: true` blocks every other job for that document
+> until its lock is reclaimed. The plugin clears stale locks on boot — see `staleJobTimeoutMs`.
+
 #### `createSyncRunner()`
 
 Runs translations inline (no queue) — handy for development or small datasets.
diff --git a/packages/payload-plugin-translator/docs/DEPRECATIONS.md b/packages/payload-plugin-translator/docs/DEPRECATIONS.md
index c677c652..38d82290 100644
--- a/packages/payload-plugin-translator/docs/DEPRECATIONS.md
+++ b/packages/payload-plugin-translator/docs/DEPRECATIONS.md
@@ -46,10 +46,31 @@ the single source of truth — code annotations link here by anchor instead of d
   remain readable. Removed (along with the fallback read path) in the next major.
 - **Code refs:**
   - `src/server/modules/task-runner/payload-jobs-runner/PayloadJobsRunnerProvider.ts` (inputSchema, handler input type/unpacking)
-  - `src/server/modules/task-runner/payload-jobs-runner/PayloadJobsTaskRunner.ts` (enqueue write, `findByCollection` / `findJobsInternal` query + in-memory filter)
+  - `src/server/modules/task-runner/payload-jobs-runner/PayloadJobsTaskRunner.ts` (enqueue write, `findByCollection` / `findRawJobs` query + in-memory filter)
   - `src/server/modules/task-runner/payload-jobs-runner/normalizeJob.ts` (read fallback)
   - `src/server/modules/task-runner/payload-jobs-runner/types.ts` (`PayloadJob.input` shape)
 
+### jobs-per-locale-task-shape
+
+- **What:** jobs queued as a bare task (`taskSlug: 'translate_document'`, one `target_lng` in the
+  input) rather than as the locale-walking workflow.
+- **Status:** live
+- **Deprecated:** 2026-09-04 / PR #133
+- **Replacement:** one workflow job per document (`workflowSlug: 'translate_document_locales'`,
+  `target_lngs` list).
+- **Remove in:** next major
+- **Why:** a job per locale meant Payload ran a document's locales through `Promise.all`, and every
+  write it makes is a whole-document version snapshot — so the second locale's snapshot dropped the
+  first's work. See [locale workflow](./plans/2026-09-04-locale-workflow.task.md).
+- **Migration:** expand/contract. Nothing writes the task shape any more, but rows queued before the
+  upgrade are still in `payload-jobs`, so every read keeps a fallback for them. Removing the fallback
+  strands those rows: cancel, stale-lock reclaim and the status panels stop finding them, silently.
+- **Code refs:**
+  - `src/server/modules/task-runner/payload-jobs-runner/PayloadJobsTaskRunner.ts` (`ownJobs()`)
+  - `src/server/modules/task-runner/payload-jobs-runner/planEnqueue.ts` (`pickHost` skips it)
+  - `src/server/modules/task-runner/payload-jobs-runner/normalizeJob.ts` (`normalizeJobLocales`
+    expands it to itself)
+
 ### find-by-collection-document-ids-array
 
 - **What:** passing a bare `Array<string | number>` of document ids as the second argument to
diff --git a/packages/payload-plugin-translator/docs/plans/2026-09-04-locale-workflow.task.md b/packages/payload-plugin-translator/docs/plans/2026-09-04-locale-workflow.task.md
new file mode 100644
index 00000000..a2d6340a
--- /dev/null
+++ b/packages/payload-plugin-translator/docs/plans/2026-09-04-locale-workflow.task.md
@@ -0,0 +1,188 @@
+# Task contract — one workflow per document instead of one job per locale (#114)
+
+**Risk: HIGH.** Changes what the jobs runner queues, what supersession cancels, and where the admin
+panel reads per-locale state. Four callers, and it affects data an editor sees.
+
+## The defect, measured
+
+Parallel per-locale jobs on one document lose **the translations themselves**, not only their
+publication as #114's title says. Measured on the plugin's real path — the `/translate/enqueue`
+endpoint, the Payload jobs runner, and `payload.jobs.run` (the call the autorun cron makes):
+
+| adapter | trial 1 | trial 2 | trial 3 |
+|---|---|---|---|
+| SQLite | 1 of 2 | 1 of 2 | 1 of 2 |
+| Postgres | 1 of 2 | 1 of 2 | 1 of 2 |
+| MongoDB | 1 of 2 | 1 of 2 | 1 of 2 |
+
+Nine of nine: one translation of two lands, which one varies, no error raised, both jobs report
+success.
+
+**Cause.** Every write is a whole-document version snapshot, drafts included
+(`payload/dist/collections/operations/utilities/update.js:188` reads the last published version as
+its base and merges one locale onto it). Two jobs run that read-modify-write from the same base;
+whoever writes second produces a snapshot that never contained the other's work.
+
+**The parallelism is ours.** `AutoTranslate.policy.ts:138` emits one task per target locale, and
+`runJobs` batches them through `Promise.all` unless `sequential` is passed — which the autorun cron
+does not pass and `AutorunCronConfig` cannot express.
+
+**Why the suite never caught it.** Every integration spec boots `createSyncRunner`, which runs
+translations inline and in order. The jobs runner — the production default — had no coverage until
+#126 added `bootTestPayload({ runner })`.
+
+## Design decisions
+
+**D1 — a Payload workflow with one task per locale.**
+Verified by running it before choosing it: tasks execute strictly in sequence; the job's `log` array
+records one entry per task with `taskSlug`, `input`, `output` and `state`; a failure stops the run and
+leaves later locales unattempted; a retry resumes at the failed locale rather than redoing the
+successful ones; and no migration is needed, because a job's `input` is a single JSON column and
+`payload_jobs_log` already exists.
+
+Rejected, with measurements, in #128: Payload's `enableConcurrencyControl` defers a blocked job to the
+**next cron tick** — roughly a minute per locale at the plugin's `* * * * *` autorun — and adds an
+indexed column to the jobs collection.
+
+**D2 — the grouping happens inside `PayloadJobsTaskRunner`, not at its callers.**
+`TaskRunner.enqueue(tasks: TaskInput[])` already receives one entry per locale, and the runner already
+groups them by collection. Grouping by document and queueing one workflow is a change contained to
+that class. Rejected: changing `TaskInput` to carry a locale array — it would touch both callers, the
+sync runner, the lifecycle wrapper and the endpoint, for no gain, since the array already arrives.
+
+**D3 — a re-enqueue supersedes every live job for the document, the one in flight included.**
+The user's decision, taken twice. The first version spared a running workflow so it could keep its
+already-translated locales; review showed that premise does not hold — Payload has no "queue behind",
+its picker takes any job that is not currently processing, so the spared job and the new one write the
+document at once on the next cron tick. That is the lost update at workflow granularity. Rejected on
+the second pass: leaving the running job and dropping the new request, which loses no work but also
+never translates the edit the user just asked for.
+
+What a cancelled run actually costs: nothing that was translated, because those locales are already
+written to the documents. Only the job's log and its unfinished locale go, and the new workflow
+redoes that locale against the newer source — which is what the user asked for by re-enqueueing.
+
+**D4 — the panel reads per-locale state from the job log.**
+The user's decision. Payload writes an entry per task, so the detail survives the collapse from N jobs
+to one. Rejected: one row per document — less work now, but the editor loses sight of which locale is
+done and which failed, which is the information the panel exists to show.
+
+**Placement:** the workflow registration goes beside the task registration in
+`PayloadJobsRunnerProvider.configure()`; the grouping in `PayloadJobsTaskRunner.enqueue`; the log
+reduction where `latestTaskPerTargetLocale` lives today.
+
+**New surface:** none outside the plugin. The `/translate/enqueue` contract is unchanged — it already
+accepts `target_lng` as a string or an array.
+
+**Written contract owed:** yes — what the workflow guarantees about ordering and about partial
+completion after a failure. Neither is expressible in a signature.
+
+**Escalate:** no. One module, no new dependency, no schema change.
+
+## Acceptance criteria
+
+1. **Every requested locale is translated.** Enqueue two locales for one document through the endpoint
+   with the jobs runner, run the queue, and both translations are present. *Check: integration test on
+   SQLite.* Currently fails — one of two lands.
+2. **The same holds on Postgres and MongoDB.** *Check: `test:integration:postgres` and `:mongo`.*
+3. **Locales run in order, not concurrently.** *Check: integration test asserting the job log's entries
+   are ordered and non-overlapping.*
+4. **A failure stops the run and leaves later locales unattempted**, and the job records which locale
+   failed. *Check: integration test with a provider that throws for one locale.*
+5. **A retry resumes rather than restarting** — a locale already translated is not sent to the provider
+   again. *Check: integration test using the boot's `translateCount()`.*
+6. **A re-enqueue leaves exactly one live job on the document**, whatever state the old one was in —
+   not started, queued for retry with every locale logged, or in flight. *Check: unit tests on the
+   runner plus `job-extend.int.test.ts`.*
+7. **The status endpoints still report per-locale state**, now from the job log. *Check: integration
+   test through the real endpoints.*
+8. **The sync runner is unaffected.** *Check: the existing 73 integration tests stay green on SQLite.*
+9. **Checks clean:** check-types both packages, unit tests, lint at the repo baseline.
+
+## Human choices
+
+- **Supersede every live job for the document** (D3), after the first version's premise was
+  disproved. Rejected alternative recorded above.
+- **Panel reads the job log** (D4). Rejected alternative recorded above.
+- **Fix this under #114** rather than opening a separate issue, with #114's description extended by a
+  comment to cover the wider defect.
+
+## Risks
+
+- The panel's data source changes; a mistake there is visible to editors even though no translation is
+  lost.
+- Old jobs queued before the upgrade carry the per-locale shape. `readCollectionRef` is the precedent
+  for reading two stored shapes and should be followed rather than reinvented.
+- Workflows are a Payload concept the plugin has not used before. The behaviour above was verified on
+  3.84.1; the plugin's peer floor is `^3.76.0` and that gap is unverified.
+
+## Corrected while building
+
+**`bootTestPayload` cannot boot twice in one file.** `getPayload` caches per process, so a second
+`bootTestPayload` in the same spec returns the first — the failing-provider case silently ran against
+the healthy boot and reported the wrong thing. Split into its own file, which is what the harness's
+own docblock already warns about.
+
+**The per-locale rows must keep the real job id.** The first version of `normalizeJobLocales`
+synthesised `${jobId}:${locale}` for uniqueness. That id is handed straight to `cancel()`, which
+addresses jobs — so supersession stopped cancelling anything. The rows share the job id, because they
+are rows of one job.
+
+**Completed jobs are deleted by default.** Payload's `deleteJobOnComplete` defaults to `true`, and
+`bootTestPayload` inherited that — so the panel had nothing to read after a run. The harness now sets
+it to `false`, matching the dev app and the README's own recommendation. Worth stating plainly: the
+panel reading per-locale state from the job log only works for hosts that keep completed jobs.
+
+**A retry is delayed by backoff.** Payload backs a failed job off exponentially, so a retry cannot be
+observed by simply running the queue again; the spec clears `waitUntil` rather than waiting.
+
+
+## Review log
+
+**2026-09-04 — three review passes over the first implementation.** All findings were checked against
+Payload 3.84.1's source before being acted on; four were real defects and are fixed here.
+
+- **`reclaimStaleJobs` had gone dead.** It narrowed by `taskSlug`, and `jobs.queue({ workflow })`
+  writes `workflowSlug` and leaves `taskSlug` null (`queues/localAPI.js:51-54`), so after this change
+  no job matched it. Boot-time stale-lock recovery would have reported zero reclaimed, forever,
+  without erroring. Both slug predicates now come from one `ownJobs()`.
+- **`run()` read a locale row instead of the job.** `handleTaskError.js:43` stamps `completedAt` on a
+  *failed* log entry too, so a partially-failed workflow presented rows that all carried one, and the
+  Retry button answered `already_completed` → 404 for exactly the jobs a user presses it on. The
+  runner's internal reads now go through `findRawJobs` + `normalizeJob`; only `findByCollection`,
+  which feeds the panels, expands to locale rows.
+- **The supersession predicate reopened #114.** It filtered on an expanded row's `status`, and a job
+  whose locales are all logged has no `pending` row — yet a non-final task failure leaves it queued
+  (`hasError: false`, `processing: false`). Two live jobs on one document, picked into one
+  `Promise.all` batch. See D3 above for how this was settled.
+- **N reads of the job table.** The lookup sat inside the per-document loop while `findRawJobs` is
+  unpaginated and filters in memory, so `select_all` meant one full scan per document. One read now
+  serves the whole batch, and the orphaned `groupByCollection` is gone.
+
+**Test findings, and what closed them.** The audit's central result was that D4 was not tested at all:
+`normalizeJobLocales` could have its entire `job.log` branch deleted and the suite stayed green — in
+production that is the panel reporting "completed" for a locale that failed. Seven unit cases now
+cover it, and the audit's own empty implementation was pasted in to confirm four of them go red.
+Three more gaps closed: the sequencing check asserted order, which survives `Promise.all`, and now
+asserts non-overlap (`executedAt` of each locale against the previous locale's `completedAt`) —
+verified by mutating the handler to `Promise.all`, which reddens it; the failure spec failed the
+*last* locale, so "stops there" was unobservable, and now runs three locales failing the middle one;
+and "does not re-run a completed workflow" passed equally for a job left failed, so it now asserts the
+workflow reached `completedAt` first. A third locale (`es`) was added to the shared harness for this.
+
+**Comment audit.** 110 comment lines over 569 lines of code, judged too dense. The #114 lost-update
+story had been written out in five places; `PayloadJobsTaskRunner.enqueue` now owns it and the others
+point at it or are gone. Twenty-two comments deleted or shortened, mostly narration of the line below.
+What was kept is measured Payload behaviour a reader cannot recover without opening `node_modules`:
+`completedAt` on failed log entries, the locale-as-task-id restoration, the retry backoff, the
+`getPayload` per-process cache.
+
+**Verification.** Unit 1332 in the plugin; integration 15/15 on SQLite and MongoDB, 13/15 on Postgres
+— the two failing files are the known #124 auto-translate cases, red on `main` as well. check-types
+clean in both packages; lint at the repo baseline.
+
+**Left open, deliberately.** `enqueue` cancels the old job and queues the new one without a
+transaction, so a batch spanning several documents can leave one document's old job deleted and its
+new workflow unqueued if a later document throws. _Closed by
+[2026-09-08-one-live-job-per-document](./2026-09-08-one-live-job-per-document.task.md): `enqueue` no
+longer cancels or deletes anything._
diff --git a/packages/payload-plugin-translator/docs/plans/2026-09-08-one-live-job-per-document.task.md b/packages/payload-plugin-translator/docs/plans/2026-09-08-one-live-job-per-document.task.md
new file mode 100644
index 00000000..1681073b
--- /dev/null
+++ b/packages/payload-plugin-translator/docs/plans/2026-09-08-one-live-job-per-document.task.md
@@ -0,0 +1,187 @@
+# Task contract — one live job per document, extended rather than replaced
+
+Continues [2026-09-04-locale-workflow.task.md](./2026-09-04-locale-workflow.task.md), which made a
+document's locales one workflow job. That change left the *second request* case wrong: a new request
+replaced the live job outright and dropped whatever locales it still owed.
+
+**Risk: HIGH.** Changes what `enqueue` writes, removes supersession entirely, touches a security
+defect in the cancel path, and adds a second execution mode. Data an editor sees is affected.
+
+## Stage 1 — extend the live job instead of replacing it
+
+**The defect.** `enqueue` cancelled every live job for the document and queued a replacement carrying
+only the locales of the current request. The panel's per-row "re-translate" button sends exactly one
+locale, so a live job holding `['de','fr','es']` became `['fr']` and two locales vanished with no
+trace. Introduced by the supersession rule added on 2026-09-04.
+
+**D1 — a locale is added to the live job's stored list; no cancel, no delete.**
+Measured before choosing (see *Evidence*): a locale written into a running job's row IS picked up by
+its handler, on all three adapters. So "queue behind" needs no new store — `target_lngs` already is
+the queue. Rejected: keep replacing and carry over the unfinished locales. It works, but it cancels a
+running job, and cancellation was measured not to stop one (see D3 of the earlier contract, now
+superseded) — the row is deleted before the handler can read the flag.
+
+**D2 — the write is narrow: the `input` column only, through the database adapter.**
+Measured: `payload.update` is the full document operation — it re-reads, merges and writes the row
+whole, reverting log entries pushed in between. Observed at 3/120 on Postgres and 1/120 on MongoDB.
+A narrow write showed 0 clobbering across ~1000 rounds on all three adapters. The jobs collection's
+only `beforeChange` hook guards a *cancelled* job from being revived; we never append to a cancelled
+job, so skipping hooks costs nothing.
+
+**D3 — the choice lives in one pure function, `planEnqueue`.**
+Three situations (no live job · add to it · give it a job of its own) and, from stage 3, one more
+input (is the queue exclusive). Keeping it inline would put the mode check in the middle of an
+I/O-heavy method. Rejected: a strategy object per mode — the two modes differ in exactly one branch,
+so an interface plus two implementations costs more than the branch.
+
+**D4 — after writing, verify.**
+The residual loss is the read-modify-write gap: the job can complete between our read and our write,
+leaving the locale in the list with nobody to run it. Measured at 0–5% of appends landing mid-run.
+Detectable: re-read after writing; if the job has completed, or the locale is not in the stored list,
+give those locales a job of their own. That fallback fires only when the old job is already finished,
+so it never creates two live jobs.
+
+## Stage 2 — defects found by the review passes
+
+1. **The panel crashes** when a locale failed and the job carries no final error yet: the client reads
+   `run.error.message` unconditionally for `status: "failed"`. Introduced on 2026-09-04, when a row's
+   status started coming from the job log while its error still came from the job.
+2. **`/translate/cancel` deletes any row in `payload-jobs`.** Ids come from the request body and the
+   delete is not narrowed to this plugin's jobs; the default access guard allows everyone. `ownJobs()`
+   already exists and is simply not applied there.
+3. **`cancel-by-collection` skips jobs waiting to retry** — the same expanded-row `status === "pending"`
+   predicate that was already fixed in `enqueue`.
+4. **`run()` claims success without running anything.** Payload returns `noJobsRemaining` /
+   `remainingJobsFromQueried: 0` when the picker takes nothing; we ignore it.
+
+## Stage 3 — honour the host's `enableConcurrencyControl`
+
+**D5 — we never set the flag; we adapt to it.**
+The user's decision. Enabling it adds an indexed column to `payload-jobs` and needs a migration on SQL
+— that is the host's call, not a plugin's. Rejected: a plugin option that sets it, and setting it
+unconditionally; both make the plugin responsible for someone else's schema.
+
+**D6 — no plugin option, no config field.** The fact is readable at the point of use
+(`payload.config.jobs.enableConcurrencyControl`). Adding an option would be a second source of truth
+for something Payload already stores, and Payload refuses to boot if `concurrency` is declared while
+the flag is off — so the two can never legitimately disagree.
+
+With the flag on, the workflow declares `concurrency: { key: <collection>:<id>, exclusive: true }` and
+`planEnqueue` stops appending to a *running* job: the new job is queued alongside and Payload holds it
+until the running one finishes.
+
+## Evidence gathered before designing
+
+Probes run on SQLite, Postgres and MongoDB, ~1000 rounds total.
+
+| Question | Answer |
+|---|---|
+| Does a running handler see a locale appended to its row? | Yes, on all three adapters (28/28 with a barrier) |
+| Is the probe able to fail? | Yes — reverting the handler's loop to a one-time read reddens it |
+| Does `payload.update` clobber the job log? | Yes: 3/120 Postgres, 1/120 MongoDB |
+| Does a narrow write clobber it? | No: 0 across all adapters |
+| Residual loss with a narrow write | 0–5% of appends landing mid-run; always "write landed, handler never ran it", never "write vanished" |
+| Does the handler reading the row itself help? | No measurable difference — dropped |
+
+## Acceptance criteria
+
+1. **A second request adds its locales instead of replacing the job.** Enqueue `['de','fr']`, then
+   enqueue `['es']` for the same document before the queue runs; one job exists carrying all three.
+   *Check: integration test on SQLite.* Fails now — today the second request replaces the first.
+2. **Nothing a live job still owes is dropped.** After the sequence in (1), running the queue
+   translates all three locales. *Check: same test.* Fails now.
+3. **A locale appended to a job already running is picked up.** *Check: integration test with a
+   barrier holding one locale.* Fails now — no append path exists.
+4. **The job log survives an append.** *Check: the same test asserts the entry written before the
+   append is still present.*
+5. **A locale that cannot be delivered gets a job of its own.** *Check: unit test on the runner —
+   the stored row reports completed after the write.*
+6. **`planEnqueue` returns the right plan for each situation**, in both modes. *Check: unit tests.*
+7. **The panel no longer crashes on a failed locale with no job error.** *Check: unit test on
+   `buildTranslationStatusRows`.* Fails now.
+8. **Cancel only ever deletes this plugin's jobs.** *Check: unit test asserting the delete's `where`
+   carries the own-jobs predicate.* Fails now.
+9. **`cancel-by-collection` cancels a job waiting to retry.** *Check: unit test.* Fails now.
+10. **`run()` reports failure when the picker took nothing.** *Check: unit test.* Fails now.
+11. **With the host's flag on, two jobs for one document never run at once, and none is lost.**
+    *Check: integration test booted with `enableConcurrencyControl`.*
+12. **With the flag on, jobs for different documents still run in parallel.** *Check: same file.*
+13. **The workflow declares `concurrency` only when the flag is on.** *Check: unit test on the
+    provider — Payload refuses to boot otherwise.*
+14. **The existing integration suite passes in both modes**, on all three adapters.
+15. **Checks clean:** unit tests, check-types in both packages, lint at the repo baseline.
+
+## Human choices
+
+- **Extend the live job rather than replace it** (D1), after being shown that a replacement drops the
+  locales the old job still owed.
+- **Never set `enableConcurrencyControl` ourselves** (D5). Stated as: requiring the host to flip a flag
+  to fix our bug would be offloading the bug onto them, and flipping it silently is worse.
+- **One gate, not two implementations** (D3) — the user asked explicitly not to smear two modes across
+  the code.
+- **Test the second mode too**, including the whole suite under the flag, not just a targeted test.
+- **The row/job type split is out of scope** and stays a follow-up.
+
+## Risks
+
+- The append path rests on Payload refreshing `job.input` from the row after each task — an
+  implementation detail, not a documented contract, measured on 3.84.1 against a peer floor of
+  `^3.76.0`. Criterion 3's test is what turns a future change into a red run instead of silent loss.
+- Removing supersession means a document can briefly hold a finished job and a new one. Nothing reads
+  "the" job for a document, so no caller breaks, but it changes what the panel lists.
+- Enabling the host's flag adds one query per queue run for every job in the host's app, not just ours.
+
+## Review log
+
+**2026-09-08 — three review passes, all findings checked against the code before acting.**
+
+**Design review.** Three real defects, all fixed here.
+- *The auto-translate debounce had stopped coalescing.* The common case is the same locales on every
+  save, which leaves nothing to append — so `enqueue` did nothing at all and the pending job kept the
+  first edit's `waitUntil`. The plan now names the host job even when it has nothing to add, and the
+  debounce moves with the request.
+- *A request's settings were silently discarded.* A job carries one source locale, one strategy and
+  one publish flag for all of its locales; a request that chose differently would have run under the
+  job's. `pickHost` now requires all three to match, and `enqueue` groups by them, which makes
+  `queueWorkflow` taking them from the first task true by construction rather than by luck.
+- *Two concurrent appends lost one, undetectably.* The verify step only asked about its own locales,
+  so a competing write that replaced the whole list looked like success. The write now retries once
+  from the stored row — a set union, so retrying is harmless — and anything still missing gets a job
+  of its own.
+
+Also: documents are served in parallel again (the loop had made a `select_all` enqueue N sequential
+round trips), `findRawJobs` reads at `depth: 0` so the legacy relationship field is not populated,
+and `workflowName` was removed from the public options and derived from `taskName` — nobody asked to
+set it, and keeping it out means no new public surface and no `@since` to date.
+
+Corrected while doing this: `ultracite fix` had rewritten two constructors away from the parameter
+properties every sibling handler uses, and dropped a `private` in the process. Reverted.
+
+**Test audit (mutation-based).** 13 mutations, **7 survived** the first suite. The worst: deleting
+the per-document filter on the live jobs handed to `planEnqueue` kept all 127 unit tests green — under
+it, enqueuing for one document appends to another document's job. Also surviving: ignoring
+`exclusiveQueue`; taking the last live job instead of the newest; dropping the debounce carry;
+blanking the other stored fields on append; and the plugin switching the host's
+`enableConcurrencyControl` on for them. Each now has a check, and each mutation was re-run to confirm
+it dies.
+
+One integration check was self-satisfying: `exclusive-queue` asserted the picker took nothing, which
+is equally true when there was nothing to take. It now first requires the second job to exist and to
+carry exactly the requested locale.
+
+**Comment audit.** 215 comment lines over 1313 of code, judged too dense; now 164 over ~1570. One
+comment was outright false — "groups are disjoint by document" stopped being true once the grouping
+key gained the settings — and several pointed at code the diff had moved or renamed. The recurring
+fault was one rationale written out in up to eight places; each now has a single owner and the rest
+point at it or are gone. What was kept is measured Payload behaviour a reader cannot recover without
+opening `node_modules`.
+
+**Verification.** Unit 1359 in the plugin; check-types clean in both packages; lint 8 warnings on the
+changed files against 10 on `main` for the same files, 0 errors. Integration on three adapters in both
+queue modes — six runs: SQLite 82/82 and MongoDB 82/82 in each, Postgres 79/82 in each, the three
+failures being the known #124 auto-translate cases that are red on `main` as well.
+
+**Left open.** The row/job type split (several panel rows share one job id, so per-row Cancel acts on
+the whole document) stays a follow-up — it changes the `TaskRunner` contract, both runners, five
+handlers and the client.
+
diff --git a/packages/payload-plugin-translator/src/client/entities/translation/model/statusRows.test.ts b/packages/payload-plugin-translator/src/client/entities/translation/model/statusRows.test.ts
index fe767e69..fb02387b 100644
--- a/packages/payload-plugin-translator/src/client/entities/translation/model/statusRows.test.ts
+++ b/packages/payload-plugin-translator/src/client/entities/translation/model/statusRows.test.ts
@@ -123,6 +123,21 @@ describe("buildTranslationStatusRows", () => {
     expect(buildTranslationStatusRows({})).toEqual([]);
     expect(buildTranslationStatusRows({ staleness: { locales: [] }, runs: [] })).toEqual([]);
   });
+  it("renders a failed locale whose job carries no error yet", () => {
+    const failedWithoutError = {
+      id: "job-de",
+      status: DocumentTranslationStatus.FAILED,
+      created_at: "2026-07-06T00:00:00.000Z",
+      updated_at: "2026-07-07T00:00:00.000Z",
+      input: { source_lng: "en", target_lng: "de" },
+    } as DocumentTranslation;
+
+    const rows = buildTranslationStatusRows({ runs: [failedWithoutError] });
+
+    expect(rows).toHaveLength(1);
+    expect(rows[0].state).toBe("failed");
+    expect(rows[0].error).toBeUndefined();
+  });
 });
 
 describe("STATE_DOT", () => {
diff --git a/packages/payload-plugin-translator/src/client/entities/translation/model/statusRows.ts b/packages/payload-plugin-translator/src/client/entities/translation/model/statusRows.ts
index 9d2634e6..e421fa13 100644
--- a/packages/payload-plugin-translator/src/client/entities/translation/model/statusRows.ts
+++ b/packages/payload-plugin-translator/src/client/entities/translation/model/statusRows.ts
@@ -91,7 +91,7 @@ export function buildTranslationStatusRows(input: {
         state: existing && !TRANSIENT.has(state) ? existing.state : state,
         at: run.updated_at,
         jobId: TRANSIENT.has(state) ? run.id : existing?.jobId,
-        error: run.status === "failed" ? run.error.message : undefined,
+        error: run.status === "failed" ? run.error?.message : undefined,
       });
     }
   }
diff --git a/packages/payload-plugin-translator/src/client/entities/translation/model/types.ts b/packages/payload-plugin-translator/src/client/entities/translation/model/types.ts
index fd2f02d8..ea4426d3 100644
--- a/packages/payload-plugin-translator/src/client/entities/translation/model/types.ts
+++ b/packages/payload-plugin-translator/src/client/entities/translation/model/types.ts
@@ -19,7 +19,8 @@ export type DocumentTranslationFailed = {
   created_at: string;
   updated_at: string;
   input: InputData;
-  error: {
+  /** Unset until the job stops retrying, so a row can read `failed` with nothing to show. */
+  error?: {
     message: string;
   };
 };
@@ -42,9 +43,8 @@ export type DocumentTranslationCompleted = {
 };
 
 /**
- * One translation job for a single target locale. The document status feed is an array of these —
- * the latest job per target locale (see `useDocumentTranslation`), because re-translate queues an
- * independent job per locale.
+ * One row per target locale in the document status feed. Several rows can share one job id — a job
+ * carries all of a document's locales.
  */
 export type DocumentTranslation =
   | DocumentTranslationCompleted
diff --git a/packages/payload-plugin-translator/src/server/features/cancel-by-collection/handler.test.ts b/packages/payload-plugin-translator/src/server/features/cancel-by-collection/handler.test.ts
index f4104fe8..694ded27 100644
--- a/packages/payload-plugin-translator/src/server/features/cancel-by-collection/handler.test.ts
+++ b/packages/payload-plugin-translator/src/server/features/cancel-by-collection/handler.test.ts
@@ -95,10 +95,10 @@ describe("CancelByCollectionHandler", () => {
       expect(response.body).toBeNull();
     });
 
-    it("returns 204 when all tasks are already completed", async () => {
+    it("returns 204 when all tasks are running (not pending)", async () => {
       const tasks = [
-        createMockTask({ id: "task-1", status: "completed" }),
-        createMockTask({ id: "task-2", status: "failed" }),
+        createMockTask({ id: "task-1", status: "running" }),
+        createMockTask({ id: "task-2", status: "running" }),
       ];
       (mockTaskRunner.findByCollection as ReturnType<typeof vi.fn>).mockResolvedValue(tasks);
 
@@ -108,11 +108,14 @@ describe("CancelByCollectionHandler", () => {
       expect(response.status).toBe(204);
       expect(mockTaskRunner.cancel).not.toHaveBeenCalled();
     });
+  });
 
-    it("returns 204 when all tasks are running (not pending)", async () => {
+  describe("cancelling queued jobs", () => {
+    it("cancels every queued job and leaves the one in flight alone", async () => {
       const tasks = [
-        createMockTask({ id: "task-1", status: "running" }),
+        createMockTask({ id: "task-1", status: "pending" }),
         createMockTask({ id: "task-2", status: "running" }),
+        createMockTask({ id: "task-3", status: "pending" }),
       ];
       (mockTaskRunner.findByCollection as ReturnType<typeof vi.fn>).mockResolvedValue(tasks);
 
@@ -120,25 +123,32 @@ describe("CancelByCollectionHandler", () => {
       const response = await handler.handle(req);
 
       expect(response.status).toBe(204);
-      expect(mockTaskRunner.cancel).not.toHaveBeenCalled();
+      expect(mockTaskRunner.cancel).toHaveBeenCalledWith(["task-1", "task-3"]);
     });
-  });
 
-  describe("cancelling pending tasks", () => {
-    it("cancels only pending tasks", async () => {
+    it("cancels a job waiting to retry after a failure", async () => {
       const tasks = [
-        createMockTask({ id: "task-1", status: "pending" }),
-        createMockTask({ id: "task-2", status: "running" }),
-        createMockTask({ id: "task-3", status: "pending" }),
-        createMockTask({ id: "task-4", status: "completed" }),
+        createMockTask({ id: "task-1", status: "completed" }),
+        createMockTask({ id: "task-1", status: "failed" }),
       ];
       (mockTaskRunner.findByCollection as ReturnType<typeof vi.fn>).mockResolvedValue(tasks);
 
       const req = createMockRequest({ collection_slug: "posts" });
-      const response = await handler.handle(req);
+      await handler.handle(req);
 
-      expect(response.status).toBe(204);
-      expect(mockTaskRunner.cancel).toHaveBeenCalledWith(["task-1", "task-3"]);
+      expect(mockTaskRunner.cancel).toHaveBeenCalledWith(["task-1"]);
+    });
+
+    it("names each job once, however many locale rows it has", async () => {
+      const tasks = [
+        createMockTask({ id: "task-1", status: "pending" }),
+        createMockTask({ id: "task-1", status: "pending" }),
+      ];
+      (mockTaskRunner.findByCollection as ReturnType<typeof vi.fn>).mockResolvedValue(tasks);
+
+      await handler.handle(createMockRequest({ collection_slug: "posts" }));
+
+      expect(mockTaskRunner.cancel).toHaveBeenCalledWith(["task-1"]);
     });
 
     it("calls findByCollection with correct collection slug", async () => {
@@ -146,7 +156,9 @@ describe("CancelByCollectionHandler", () => {
 
       await handler.handle(req);
 
-      expect(mockTaskRunner.findByCollection).toHaveBeenCalledWith("pages");
+      expect(mockTaskRunner.findByCollection).toHaveBeenCalledWith("pages", {
+        excludeCompleted: true,
+      });
     });
 
     it("creates task runner with request payload", async () => {
diff --git a/packages/payload-plugin-translator/src/server/features/cancel-by-collection/handler.ts b/packages/payload-plugin-translator/src/server/features/cancel-by-collection/handler.ts
index fa90f230..85e68ab1 100644
--- a/packages/payload-plugin-translator/src/server/features/cancel-by-collection/handler.ts
+++ b/packages/payload-plugin-translator/src/server/features/cancel-by-collection/handler.ts
@@ -7,9 +7,7 @@ import { isCollectionAvailable } from "../_lib/collection-utils";
 import { CancelByCollectionInputSchema } from "./model";
 import type { CancelConfig } from "./model";
 
-/**
- * Cancels all pending translation tasks for a collection
- */
+/** Cancels every queued job for a collection; jobs in flight are left alone. */
 export class CancelByCollectionHandler {
   constructor(
     private readonly config: CancelConfig,
@@ -29,13 +27,16 @@ export class CancelByCollectionHandler {
       return ServerResponse.badRequest("Collection not available for translation");
 
     const runner = this.taskRunnerFactory.create(req.payload);
-    const tasks = await runner.findByCollection(collectionSlug);
-    if (tasks.length === 0) return ServerResponse.noContent();
+    const rows = await runner.findByCollection(collectionSlug, { excludeCompleted: true });
+    if (rows.length === 0) return ServerResponse.noContent();
 
-    const pendingTaskIds = tasks.filter((task) => task.status === "pending").map((task) => task.id);
-    if (pendingTaskIds.length === 0) return ServerResponse.noContent();
+    // A job waiting to retry has every locale logged, so it has no `pending` row — filter by *not
+    // running* instead.
+    const running = new Set(rows.filter((row) => row.status === "running").map((row) => row.id));
+    const queuedJobIds = [...new Set(rows.map((row) => row.id))].filter((id) => !running.has(id));
+    if (queuedJobIds.length === 0) return ServerResponse.noContent();
 
-    await runner.cancel(pendingTaskIds);
+    await runner.cancel(queuedJobIds);
 
     return ServerResponse.noContent();
   }
diff --git a/packages/payload-plugin-translator/src/server/features/enqueue-translation/handler.ts b/packages/payload-plugin-translator/src/server/features/enqueue-translation/handler.ts
index 619cd6e1..622135bc 100644
--- a/packages/payload-plugin-translator/src/server/features/enqueue-translation/handler.ts
+++ b/packages/payload-plugin-translator/src/server/features/enqueue-translation/handler.ts
@@ -77,8 +77,6 @@ export class EnqueueTranslationHandler {
       : collection_id;
 
     const runner = this.taskRunnerFactory.create(req.payload);
-    // One task per (document x target locale). The runner keys/supersedes per (document, targetLng),
-    // so N concurrent targets of one document coexist (PR #75) — no runner change needed.
     const tasks = collectionIds.flatMap((id) =>
       targets.map((targetLng) => ({
         collectionSlug,
diff --git a/packages/payload-plugin-translator/src/server/modules/task-runner/payload-jobs-runner/PayloadJobsRunnerProvider.test.ts b/packages/payload-plugin-translator/src/server/modules/task-runner/payload-jobs-runner/PayloadJobsRunnerProvider.test.ts
index 96cba607..432b114a 100644
--- a/packages/payload-plugin-translator/src/server/modules/task-runner/payload-jobs-runner/PayloadJobsRunnerProvider.test.ts
+++ b/packages/payload-plugin-translator/src/server/modules/task-runner/payload-jobs-runner/PayloadJobsRunnerProvider.test.ts
@@ -18,6 +18,13 @@ const makePayload = (overrides?: { update?: ReturnType<typeof vi.fn> }) => ({
   logger: { error: vi.fn() },
 });
 
+const workflowOf = (config: Config) =>
+  (config.jobs?.workflows ?? []).find(
+    (w) => w.slug === "translate_document_locales"
+  ) as unknown as {
+    concurrency?: { key: (a: never) => string; exclusive?: boolean };
+  };
+
 describe("PayloadJobsRunnerProvider", () => {
   describe("configure().onInit", () => {
     it("returns a config whose onInit is a function that triggers reclaim", async () => {
@@ -85,4 +92,30 @@ describe("PayloadJobsRunnerProvider", () => {
       expect(() => createPayloadJobsRunner({ staleJobTimeoutMs: 60_000 })).not.toThrow();
     });
   });
+
+  describe("the workflow's concurrency declaration", () => {
+    it("is absent when the host has not enabled concurrency control", () => {
+      const config = createPayloadJobsRunner().configure(minimalContext)(makeConfig());
+
+      expect(workflowOf(config).concurrency).toBeUndefined();
+    });
+
+    it("does not enable the host's concurrency control on its behalf", () => {
+      const config = createPayloadJobsRunner().configure(minimalContext)(makeConfig());
+
+      expect(config.jobs?.enableConcurrencyControl).toBeUndefined();
+    });
+
+    it("keys on the document when the host has enabled it", () => {
+      const config = createPayloadJobsRunner().configure(minimalContext)(
+        makeConfig({ jobs: { enableConcurrencyControl: true } })
+      );
+
+      const concurrency = workflowOf(config).concurrency;
+      expect(concurrency?.exclusive).toBe(true);
+      expect(
+        concurrency?.key({ input: { collection_slug: "posts", collection_id: "7" } } as never)
+      ).toBe("posts:7");
+    });
+  });
 });
diff --git a/packages/payload-plugin-translator/src/server/modules/task-runner/payload-jobs-runner/PayloadJobsRunnerProvider.ts b/packages/payload-plugin-translator/src/server/modules/task-runner/payload-jobs-runner/PayloadJobsRunnerProvider.ts
index c5ca8a34..485f6fb7 100644
--- a/packages/payload-plugin-translator/src/server/modules/task-runner/payload-jobs-runner/PayloadJobsRunnerProvider.ts
+++ b/packages/payload-plugin-translator/src/server/modules/task-runner/payload-jobs-runner/PayloadJobsRunnerProvider.ts
@@ -1,7 +1,12 @@
-import type { Config, Field, Payload } from "payload";
+import type { Config, Field, Payload, WorkflowConfig } from "payload";
 
 import type { TaskRunner } from "../TaskRunner.interface";
-import type { PayloadJobsRunnerOptions, PayloadJobsRunnerConfig, AutoRunConfig } from "./types";
+import type {
+  PayloadJobsRunnerOptions,
+  PayloadJobsRunnerConfig,
+  AutoRunConfig,
+  StoredWorkflowInput,
+} from "./types";
 import { PayloadJobsTaskRunner } from "./PayloadJobsTaskRunner";
 import { readCollectionRef } from "./readCollectionRef";
 import type { TaskRunnerContext, TaskRunnerProvider } from "../TaskRunnerProvider.interface";
@@ -12,7 +17,10 @@ const defaultAutoRun: Required<AutoRunConfig> = {
   limit: 50,
 };
 
-const DEFAULT_STALE_JOB_TIMEOUT_MS = 5 * 60 * 1000; // 5 minutes
+type StoredJobInput = Partial<StoredWorkflowInput> & Record<string, unknown>;
+type RunLocaleTask = (taskID: string, args: { input: Record<string, unknown> }) => Promise<unknown>;
+
+const DEFAULT_STALE_JOB_TIMEOUT_MS = 5 * 60 * 1000;
 
 const defaultValues = {
   taskName: "translate_document",
@@ -24,16 +32,11 @@ const defaultValues = {
     attempts: 3,
     backoff: {
       type: "exponential" as const,
-      delay: 5000, // 5s, 10s, 20s
+      delay: 5000,
     },
   },
 };
 
-/**
- * TaskRunnerProvider implementation using Payload Jobs.
- *
- * Configures Payload jobs, tasks, and autorun for translation processing.
- */
 export class PayloadJobsRunnerProvider implements TaskRunnerProvider {
   private readonly config: PayloadJobsRunnerConfig;
 
@@ -54,6 +57,7 @@ export class PayloadJobsRunnerProvider implements TaskRunnerProvider {
 
     this.config = {
       taskName: options?.taskName ?? defaultValues.taskName,
+      workflowName: `${options?.taskName ?? defaultValues.taskName}_locales`,
       queueName: options?.queueName ?? defaultValues.queueName,
       jobsCollection: options?.jobsCollection ?? defaultValues.jobsCollection,
       autoRun,
@@ -67,14 +71,11 @@ export class PayloadJobsRunnerProvider implements TaskRunnerProvider {
   }
 
   configure(context: TaskRunnerContext): (config: Config) => Config {
-    const { taskName, queueName, retries, autoRun } = this.config;
+    const { taskName, workflowName, queueName, retries, autoRun } = this.config;
     const { handler, collections } = context;
 
     return (config) => {
       const inputSchema: Field[] = [
-        // Flat text reference (ID-agnostic). Current shape that jobs are
-        // written with — no relationship type validation against the target
-        // collection's ID type, so string IDs work for number-id collections.
         {
           type: "text",
           name: "collection_slug",
@@ -85,14 +86,6 @@ export class PayloadJobsRunnerProvider implements TaskRunnerProvider {
           name: "collection_id",
           required: true,
         },
-        /**
-         * Legacy relationship reference, kept as a read-only fallback so jobs
-         * queued before the ID-agnostic migration stay readable. No longer
-         * written; demoted to `required: false` so new jobs (which omit it)
-         * pass validation. Removed in the next major.
-         * See docs/DEPRECATIONS.md#jobs-input-collection-field
-         * @deprecated
-         */
         {
           type: "relationship",
           name: "collection",
@@ -128,6 +121,11 @@ export class PayloadJobsRunnerProvider implements TaskRunnerProvider {
         },
       ];
 
+      const workflowInputSchema: Field[] = [
+        ...inputSchema.filter((f) => "name" in f && f.name !== "target_lng"),
+        { type: "json", name: "target_lngs", required: true },
+      ];
+
       const task = {
         slug: taskName,
         inputSchema,
@@ -137,7 +135,6 @@ export class PayloadJobsRunnerProvider implements TaskRunnerProvider {
           input: {
             collection_slug?: string;
             collection_id?: string;
-            // Legacy fallback shape, see docs/DEPRECATIONS.md#jobs-input-collection-field
             collection?: { relationTo: string; value: string | number };
             source_lng: string;
             target_lng: string;
@@ -158,11 +155,35 @@ export class PayloadJobsRunnerProvider implements TaskRunnerProvider {
         },
       };
 
+      const workflow: WorkflowConfig<StoredJobInput> = {
+        slug: workflowName,
+        inputSchema: workflowInputSchema,
+        retries,
+        ...(config.jobs?.enableConcurrencyControl
+          ? {
+              concurrency: {
+                key: ({ input }) => `${input.collection_slug}:${input.collection_id}`,
+                exclusive: true,
+              },
+            }
+          : {}),
+        handler: async ({ job, tasks }) => {
+          const runLocale = (tasks as Record<string, RunLocaleTask>)[taskName];
+          for (let i = 0; ; i++) {
+            const { target_lngs: targets, ...shared } = job.input;
+            const target = targets?.[i];
+            if (target === undefined) return;
+            await runLocale(target, { input: { ...shared, target_lng: target } });
+          }
+        },
+      };
+
       if (!config.jobs) config.jobs = {};
       if (!config.jobs.tasks) config.jobs.tasks = [];
       config.jobs.tasks.push(task);
+      if (!config.jobs.workflows) config.jobs.workflows = [];
+      config.jobs.workflows.push(workflow);
 
-      // Skip autoRun configuration when disabled (e.g., for Vercel/serverless deployments)
       if (autoRun) {
         const autoRunConfig = {
           queue: queueName,
@@ -183,13 +204,6 @@ export class PayloadJobsRunnerProvider implements TaskRunnerProvider {
         }
       }
 
-      // Reset stale locks on boot so jobs abandoned by a killed process
-      // (deploy/crash/timeout) become eligible for the autorun picker again.
-      // The picker requires processing:false, no error, and no pending waitUntil;
-      // a mid-run casualty (no error, no waitUntil) satisfies the rest, so
-      // clearing processing is sufficient for that case. Threshold-based, so a
-      // job genuinely in flight on another live instance (fresh updatedAt) is
-      // left alone. Wrapped so a reclaim failure never blocks startup.
       const existingOnInit = config.onInit;
       config.onInit = async (payload) => {
         if (existingOnInit) await existingOnInit(payload);
@@ -208,24 +222,6 @@ export class PayloadJobsRunnerProvider implements TaskRunnerProvider {
   }
 }
 
-/**
- * Creates the **recommended** task runner: translations run as Payload Jobs
- * (queued, executed by autoRun cron or a manual run, with stale-lock recovery).
- * Durable across restarts and suited to production/serverless. Pass the result
- * as `translatorPlugin({ runner })`.
- *
- * @param options - Queue/task names, `autoRun` cron (or `false` to disable),
- *   `staleJobTimeoutMs`, and retry policy. See {@link PayloadJobsRunnerOptions}.
- * @returns A {@link TaskRunnerProvider} for the plugin's `runner` option.
- * @example
- * ```ts
- * translatorPlugin({
- *   collections: [Posts],
- *   translationProvider: createOpenAIProvider({ apiKey: process.env.OPENAI_API_KEY! }),
- *   runner: createPayloadJobsRunner({ autoRun: { cron: '* * * * *' } }),
- * })
- * ```
- */
 export function createPayloadJobsRunner(options?: PayloadJobsRunnerOptions): TaskRunnerProvider {
   return new PayloadJobsRunnerProvider(options);
 }
diff --git a/packages/payload-plugin-translator/src/server/modules/task-runner/payload-jobs-runner/PayloadJobsTaskRunner.test.ts b/packages/payload-plugin-translator/src/server/modules/task-runner/payload-jobs-runner/PayloadJobsTaskRunner.test.ts
index 1a2ee342..82282e40 100644
--- a/packages/payload-plugin-translator/src/server/modules/task-runner/payload-jobs-runner/PayloadJobsTaskRunner.test.ts
+++ b/packages/payload-plugin-translator/src/server/modules/task-runner/payload-jobs-runner/PayloadJobsTaskRunner.test.ts
@@ -9,6 +9,8 @@ describe("PayloadJobsTaskRunner", () => {
     find: ReturnType<typeof vi.fn>;
     delete: ReturnType<typeof vi.fn>;
     update: ReturnType<typeof vi.fn>;
+    db: { updateOne: ReturnType<typeof vi.fn> };
+    config: { jobs?: { enableConcurrencyControl?: boolean } };
     jobs: {
       queue: ReturnType<typeof vi.fn>;
       cancel: ReturnType<typeof vi.fn>;
@@ -24,10 +26,14 @@ describe("PayloadJobsTaskRunner", () => {
       find: vi.fn().mockResolvedValue({ docs: [] }),
       delete: vi.fn().mockResolvedValue(undefined),
       update: vi.fn().mockResolvedValue({ docs: [] }),
+      db: { updateOne: vi.fn().mockResolvedValue(undefined) },
+      config: { jobs: {} },
       jobs: {
         queue: vi.fn().mockResolvedValue(undefined),
         cancel: vi.fn().mockResolvedValue(undefined),
-        run: vi.fn().mockResolvedValue({ jobStatus: {}, remainingJobsFromQueried: 0 }),
+        run: vi
+          .fn()
+          .mockResolvedValue({ jobStatus: { "job-123": {} }, remainingJobsFromQueried: 0 }),
         // kept only so tests can assert run() never falls back to the broken
         // runByID id-path — production code does not call it.
         runByID: vi.fn().mockResolvedValue(undefined),
@@ -35,6 +41,7 @@ describe("PayloadJobsTaskRunner", () => {
     };
     config = {
       taskName: "translate_document",
+      workflowName: "translate_document_locales",
       queueName: "translations",
       jobsCollection: "payload-jobs",
       autoRun: {
@@ -69,28 +76,51 @@ describe("PayloadJobsTaskRunner", () => {
     ...overrides,
   });
 
+  /** A live workflow job whose settings match `createInput()`'s, so `pickHost` accepts it. */
+  const createLiveJob = (overrides: Partial<PayloadJob> = {}): PayloadJob =>
+    createJob({
+      input: {
+        collection_slug: "posts",
+        collection_id: "doc-123",
+        source_lng: "en",
+        strategy: "overwrite",
+        publish_on_translation: false,
+        target_lngs: ["de"],
+      },
+      ...overrides,
+    });
+
   describe("enqueue", () => {
-    it("looks for superseded jobs among unfinished ones only", async () => {
+    it("reads only unfinished jobs when planning an enqueue", async () => {
       await runner.enqueue([createInput()]);
 
       const whereArg = mockPayload.find.mock.calls[0][0].where;
       expect(whereArg).toEqual({
-        and: [{ taskSlug: { equals: "translate_document" } }, { completedAt: { exists: false } }],
+        and: [
+          {
+            or: [
+              { workflowSlug: { equals: "translate_document_locales" } },
+              { taskSlug: { equals: "translate_document" } },
+            ],
+          },
+          { completedAt: { exists: false } },
+        ],
       });
     });
 
-    it("queues tasks with correct input", async () => {
-      const input = createInput();
-      await runner.enqueue([input]);
+    it("queues one workflow per document, carrying its locales", async () => {
+      await runner.enqueue([createInput({ targetLng: "de" }), createInput({ targetLng: "fr" })]);
 
+      expect(mockPayload.jobs.queue).toHaveBeenCalledTimes(1);
       expect(mockPayload.jobs.queue).toHaveBeenCalledWith({
-        task: "translate_document",
+        workflow: "translate_document_locales",
         queue: "translations",
+        waitUntil: undefined,
         input: {
           collection_slug: "posts",
           collection_id: "doc-123",
           source_lng: "en",
-          target_lng: "de",
+          target_lngs: ["de", "fr"],
           strategy: "overwrite",
           publish_on_translation: false,
         },
@@ -113,118 +143,179 @@ describe("PayloadJobsTaskRunner", () => {
       expect(arg.waitUntil).toBeUndefined();
     });
 
-    it("queues multiple tasks", async () => {
-      const inputs = [
-        createInput({ collectionId: "doc-1" }),
-        createInput({ collectionId: "doc-2" }),
-      ];
-      await runner.enqueue(inputs);
+    it("queues one workflow per document, not one per task", async () => {
+      await runner.enqueue([
+        createInput({ collectionId: "doc-1", targetLng: "de" }),
+        createInput({ collectionId: "doc-1", targetLng: "fr" }),
+        createInput({ collectionId: "doc-2", targetLng: "de" }),
+      ]);
 
       expect(mockPayload.jobs.queue).toHaveBeenCalledTimes(2);
     });
 
-    it("cancels existing jobs before queuing new ones", async () => {
-      const existingJob = createJob({ id: "existing-job" });
-      mockPayload.find.mockResolvedValueOnce({ docs: [existingJob] });
+    it("adds the locale to a live job instead of queuing a second one", async () => {
+      const live = createLiveJob({ id: "live-job" });
+      // First `find` builds the plan, the second verifies the write.
+      mockPayload.find.mockResolvedValueOnce({ docs: [live] }).mockResolvedValueOnce({
+        docs: [{ ...live, input: { ...live.input, target_lngs: ["de", "fr"] } }],
+      });
 
-      const input = createInput();
-      await runner.enqueue([input]);
+      await runner.enqueue([createInput({ targetLng: "fr" })]);
 
-      expect(mockPayload.jobs.cancel).toHaveBeenCalledWith({
-        where: { id: { in: ["existing-job"] } },
-        queue: "translations",
-      });
-      expect(mockPayload.delete).toHaveBeenCalledWith({
-        collection: "payload-jobs",
-        where: { id: { in: ["existing-job"] } },
+      const write = mockPayload.db.updateOne.mock.calls[0][0];
+      expect(write.collection).toBe("payload-jobs");
+      expect(write.id).toBe("live-job");
+      expect(write.data.input).toEqual({
+        collection_slug: "posts",
+        collection_id: "doc-123",
+        source_lng: "en",
+        strategy: "overwrite",
+        publish_on_translation: false,
+        target_lngs: ["de", "fr"],
       });
+      expect(mockPayload.jobs.queue).not.toHaveBeenCalled();
     });
 
-    it("does not cancel a running job for a different target locale of the same document", async () => {
-      // A `de` job is in flight; the user re-translates `fr` on the same document. The `de` job must
-      // survive — cancelling it is the concurrent re-translate bug.
-      const runningDe = createJob({
-        id: "de-job",
-        processing: true,
-        input: {
-          collection: { relationTo: "posts" as CollectionSlug, value: "doc-123" },
-          source_lng: "en",
-          target_lng: "de",
-          strategy: "overwrite",
-        },
+    it("does not extend a live job that belongs to a different document", async () => {
+      mockPayload.find.mockResolvedValue({
+        docs: [
+          createJob({
+            id: "other-doc-job",
+            input: {
+              collection_slug: "posts",
+              collection_id: "doc-999",
+              source_lng: "en",
+              strategy: "overwrite",
+              target_lngs: ["de"],
+            },
+          }),
+        ],
       });
-      mockPayload.find.mockResolvedValueOnce({ docs: [runningDe] });
 
       await runner.enqueue([createInput({ targetLng: "fr" })]);
 
-      expect(mockPayload.jobs.cancel).not.toHaveBeenCalled();
-      expect(mockPayload.delete).not.toHaveBeenCalled();
+      expect(mockPayload.db.updateOne).not.toHaveBeenCalled();
       expect(mockPayload.jobs.queue).toHaveBeenCalledTimes(1);
     });
 
-    it("supersedes only the same-locale job when several locales have jobs", async () => {
-      const deJob = createJob({
-        id: "de-job",
-        input: {
-          collection: { relationTo: "posts" as CollectionSlug, value: "doc-123" },
-          source_lng: "en",
-          target_lng: "de",
-          strategy: "overwrite",
-        },
+    it("never cancels or deletes a job when enqueuing", async () => {
+      const live = createLiveJob({
+        id: "live-job",
       });
-      const frJob = createJob({
-        id: "fr-job",
-        input: {
-          collection: { relationTo: "posts" as CollectionSlug, value: "doc-123" },
-          source_lng: "en",
-          target_lng: "fr",
-          strategy: "overwrite",
-        },
+      mockPayload.find.mockResolvedValueOnce({ docs: [live] }).mockResolvedValueOnce({
+        docs: [{ ...live, input: { ...live.input, target_lngs: ["de", "fr"] } }],
       });
-      mockPayload.find.mockResolvedValueOnce({ docs: [deJob, frJob] });
 
       await runner.enqueue([createInput({ targetLng: "fr" })]);
 
-      expect(mockPayload.jobs.cancel).toHaveBeenCalledWith({
-        where: { id: { in: ["fr-job"] } },
-        queue: "translations",
+      expect(mockPayload.jobs.cancel).not.toHaveBeenCalled();
+      expect(mockPayload.delete).not.toHaveBeenCalled();
+    });
+
+    it("gives the locale its own job when the live one finished after the write landed", async () => {
+      const live = createLiveJob({ id: "live-job" });
+      mockPayload.find.mockResolvedValueOnce({ docs: [live] }).mockResolvedValueOnce({
+        docs: [
+          {
+            ...live,
+            completedAt: "2026-01-01T00:00:01Z",
+            input: { ...live.input, target_lngs: ["de", "fr"] },
+          },
+        ],
       });
-      expect(mockPayload.delete).toHaveBeenCalledWith({
-        collection: "payload-jobs",
-        where: { id: { in: ["fr-job"] } },
+
+      await runner.enqueue([createInput({ targetLng: "fr" })]);
+
+      expect(mockPayload.jobs.queue).toHaveBeenCalledTimes(1);
+      expect(mockPayload.jobs.queue).toHaveBeenCalledWith(
+        expect.objectContaining({
+          input: expect.objectContaining({ target_lngs: ["fr"] }),
+        })
+      );
+    });
+
+    it("retries the write once when a concurrent append replaced the list", async () => {
+      const live = createLiveJob({ id: "live-job" });
+      const clobbered = { ...live, input: { ...live.input, target_lngs: ["de", "es"] } };
+      mockPayload.find
+        .mockResolvedValueOnce({ docs: [live] })
+        .mockResolvedValueOnce({ docs: [clobbered] })
+        .mockResolvedValueOnce({
+          docs: [{ ...live, input: { ...live.input, target_lngs: ["de", "es", "fr"] } }],
+        });
+
+      await runner.enqueue([createInput({ targetLng: "fr" })]);
+
+      expect(mockPayload.db.updateOne).toHaveBeenCalledTimes(2);
+      expect(mockPayload.jobs.queue).not.toHaveBeenCalled();
+    });
+
+    it("gives the locale its own job when even the retry does not land it", async () => {
+      const live = createLiveJob({ id: "live-job" });
+      const withoutIt = { ...live, input: { ...live.input, target_lngs: ["de", "es"] } };
+      mockPayload.find
+        .mockResolvedValueOnce({ docs: [live] })
+        .mockResolvedValueOnce({ docs: [withoutIt] })
+        .mockResolvedValueOnce({ docs: [withoutIt] });
+
+      await runner.enqueue([createInput({ targetLng: "fr" })]);
+
+      expect(mockPayload.jobs.queue).toHaveBeenCalledWith(
+        expect.objectContaining({
+          input: expect.objectContaining({ target_lngs: ["fr"] }),
+        })
+      );
+    });
+
+    it("pushes a not-yet-started job's debounce out to this request's", async () => {
+      const live = createLiveJob({ id: "live-job" });
+      mockPayload.find.mockResolvedValue({ docs: [live] });
+      const waitUntil = new Date("2026-02-02T00:00:00.000Z");
+
+      await runner.enqueue([createInput({ targetLng: "de", waitUntil })]);
+
+      expect(mockPayload.db.updateOne.mock.calls[0][0].data).toMatchObject({
+        waitUntil: "2026-02-02T00:00:00.000Z",
       });
     });
 
-    it("does not cancel when no existing jobs", async () => {
-      mockPayload.find.mockResolvedValue({ docs: [] });
+    it("leaves a running job's schedule alone", async () => {
+      const running = createLiveJob({ id: "live-job", processing: true });
+      mockPayload.find.mockResolvedValue({ docs: [running] });
 
-      const input = createInput();
-      await runner.enqueue([input]);
+      await runner.enqueue([
+        createInput({ targetLng: "de", waitUntil: new Date("2026-02-02T00:00:00.000Z") }),
+      ]);
 
-      expect(mockPayload.jobs.cancel).not.toHaveBeenCalled();
-      expect(mockPayload.delete).not.toHaveBeenCalled();
+      expect(mockPayload.db.updateOne).not.toHaveBeenCalled();
     });
 
-    it("groups tasks by collection", async () => {
-      const inputs = [
-        createInput({
-          collectionSlug: "posts" as CollectionSlug,
-          collectionId: "post-1",
-        }),
-        createInput({
-          collectionSlug: "posts" as CollectionSlug,
-          collectionId: "post-2",
-        }),
-        createInput({
-          collectionSlug: "pages" as CollectionSlug,
-          collectionId: "page-1",
-        }),
-      ];
+    it("queues alongside a running job when the host enabled concurrency control", async () => {
+      mockPayload.config.jobs = { enableConcurrencyControl: true };
+      mockPayload.find.mockResolvedValue({
+        docs: [
+          createLiveJob({
+            id: "running-job",
+            processing: true,
+          }),
+        ],
+      });
+
+      await runner.enqueue([createInput({ targetLng: "fr" })]);
+
+      expect(mockPayload.db.updateOne).not.toHaveBeenCalled();
+      expect(mockPayload.jobs.queue).toHaveBeenCalledTimes(1);
+    });
 
-      await runner.enqueue(inputs);
+    it("reads the job table once for the whole batch, however many documents it spans", async () => {
+      await runner.enqueue([
+        createInput({ collectionSlug: "posts" as CollectionSlug, collectionId: "post-1" }),
+        createInput({ collectionSlug: "posts" as CollectionSlug, collectionId: "post-2" }),
+        createInput({ collectionSlug: "pages" as CollectionSlug, collectionId: "page-1" }),
+      ]);
 
-      // Should check for existing jobs per collection
-      expect(mockPayload.find).toHaveBeenCalledTimes(2);
+      expect(mockPayload.find).toHaveBeenCalledTimes(1);
+      expect(mockPayload.jobs.queue).toHaveBeenCalledTimes(3);
     });
 
     it("stores the reference as flat text, coercing the id to a string", async () => {
@@ -274,7 +365,29 @@ describe("PayloadJobsTaskRunner", () => {
       });
       expect(mockPayload.delete).toHaveBeenCalledWith({
         collection: "payload-jobs",
-        where: { id: { in: ["job-1", "job-2"] } },
+        where: {
+          and: [
+            {
+              or: [
+                { workflowSlug: { equals: "translate_document_locales" } },
+                { taskSlug: { equals: "translate_document" } },
+              ],
+            },
+            { id: { in: ["job-1", "job-2"] } },
+          ],
+        },
+      });
+    });
+
+    it("deletes only this plugin's jobs, whatever ids it is handed", async () => {
+      await runner.cancel(["someone-elses-job"]);
+
+      const where = mockPayload.delete.mock.calls[0][0].where as { and?: unknown[] };
+      expect(where.and?.[0]).toEqual({
+        or: [
+          { workflowSlug: { equals: "translate_document_locales" } },
+          { taskSlug: { equals: "translate_document" } },
+        ],
       });
     });
 
@@ -308,6 +421,58 @@ describe("PayloadJobsTaskRunner", () => {
       expect(result).toEqual({ success: false, error: "already_completed" });
     });
 
+    it("retries a workflow whose locales are logged but whose run failed", async () => {
+      const partiallyFailed = createJob({
+        input: {
+          collection_slug: "posts",
+          collection_id: "doc-123",
+          source_lng: "en",
+          target_lngs: ["de", "fr"],
+          strategy: "overwrite",
+        },
+        log: [
+          { state: "succeeded", completedAt: "2024-01-01T00:01:00Z", input: { target_lng: "de" } },
+          { state: "failed", completedAt: "2024-01-01T00:02:00Z", input: { target_lng: "fr" } },
+        ],
+      });
+      mockPayload.find.mockResolvedValue({ docs: [partiallyFailed] });
+
+      const result = await runner.run("job-123");
+
+      expect(result).toEqual({ success: true });
+      expect(mockPayload.jobs.run).toHaveBeenCalledWith({
+        queue: "translations",
+        where: { id: { equals: "job-123" } },
+        limit: 1,
+      });
+    });
+
+    it("reports failure when the picker took nothing", async () => {
+      mockPayload.find.mockResolvedValue({ docs: [createJob()] });
+      mockPayload.jobs.run.mockResolvedValue({ jobStatus: {}, remainingJobsFromQueried: 0 });
+
+      expect(await runner.run("job-123")).toEqual({
+        success: false,
+        error: "already_running",
+      });
+    });
+
+    it("clears what blocks the picker before retrying a failed job", async () => {
+      mockPayload.find.mockResolvedValue({
+        docs: [createJob({ error: { message: "provider down" } })],
+      });
+
+      await runner.run("job-123");
+
+      expect(mockPayload.update).toHaveBeenCalledWith(
+        expect.objectContaining({
+          collection: "payload-jobs",
+          where: { id: { equals: "job-123" } },
+          data: { processing: false, hasError: false, error: null, waitUntil: null },
+        })
+      );
+    });
+
     it("returns already_running when a job is genuinely in flight (fresh lock)", async () => {
       const runningJob = createJob({
         processing: true,
@@ -339,9 +504,8 @@ describe("PayloadJobsTaskRunner", () => {
         collection: "payload-jobs",
         depth: 0,
         where: { id: { equals: "job-123" } },
-        data: { processing: false },
+        data: { processing: false, hasError: false, error: null, waitUntil: null },
       });
-      // run via the where-based picker, NOT runByID
       expect(mockPayload.jobs.run).toHaveBeenCalledWith({
         queue: "translations",
         where: { id: { equals: "job-123" } },
@@ -369,7 +533,7 @@ describe("PayloadJobsTaskRunner", () => {
         collection: "payload-jobs",
         depth: 0,
         where: { id: { equals: "job-123" } },
-        data: { processing: false },
+        data: { processing: false, hasError: false, error: null, waitUntil: null },
       });
     });
 
@@ -385,13 +549,19 @@ describe("PayloadJobsTaskRunner", () => {
         where: { id: { equals: "job-123" } },
         limit: 1,
       });
-      // a pending job (processing:false) needs no lock reset
       expect(mockPayload.update).not.toHaveBeenCalled();
-      // findJobsInternal must narrow by taskSlug AND the given id
       expect(mockPayload.find).toHaveBeenCalledWith(
         expect.objectContaining({
           where: {
-            and: [{ taskSlug: { equals: "translate_document" } }, { id: { equals: "job-123" } }],
+            and: [
+              {
+                or: [
+                  { workflowSlug: { equals: "translate_document_locales" } },
+                  { taskSlug: { equals: "translate_document" } },
+                ],
+              },
+              { id: { equals: "job-123" } },
+            ],
           },
         })
       );
@@ -426,7 +596,7 @@ describe("PayloadJobsTaskRunner", () => {
       vi.useRealTimers();
     });
 
-    it("resets stale processing locks via a real-column where clause", async () => {
+    it("narrows the stale-lock reset to both stored slugs", async () => {
       vi.useFakeTimers();
       vi.setSystemTime(new Date("2026-01-01T00:00:00.000Z"));
 
@@ -448,7 +618,12 @@ describe("PayloadJobsTaskRunner", () => {
       ).toISOString();
       expect(arg.where).toEqual({
         and: [
-          { taskSlug: { equals: "translate_document" } },
+          {
+            or: [
+              { workflowSlug: { equals: "translate_document_locales" } },
+              { taskSlug: { equals: "translate_document" } },
+            ],
+          },
           { processing: { equals: true } },
           { completedAt: { exists: false } },
           { updatedAt: { less_than: expectedCutoff } },
@@ -524,17 +699,20 @@ describe("PayloadJobsTaskRunner", () => {
       });
     });
 
-    it("narrows the SQL where clause by taskSlug only", async () => {
-      // Slug and id are matched in memory (see PayloadJobsTaskRunner.findByCollection
-      // for the full reasoning), spanning both the new flat-text shape and the
-      // legacy relationship shape. The WHERE sent to Payload must narrow only by
-      // taskSlug — never by the collection slug or id — otherwise we both
-      // re-introduce the SQLite type-coercion bug and drop one of the two shapes.
+    it("narrows the SQL where clause by the job's own slugs only", async () => {
+      // See `findByCollection`'s docblock: a `where` on slug or id drops every pre-migration job.
       await runner.findByCollection("posts" as CollectionSlug, [5, 6]);
 
       const whereArg = mockPayload.find.mock.calls[0][0].where;
       expect(whereArg).toEqual({
-        and: [{ taskSlug: { equals: "translate_document" } }],
+        and: [
+          {
+            or: [
+              { workflowSlug: { equals: "translate_document_locales" } },
+              { taskSlug: { equals: "translate_document" } },
+            ],
+          },
+        ],
       });
       expect(JSON.stringify(whereArg)).not.toContain("collection_id");
       expect(JSON.stringify(whereArg)).not.toContain("collection.value");
@@ -545,7 +723,15 @@ describe("PayloadJobsTaskRunner", () => {
 
       const whereArg = mockPayload.find.mock.calls[0][0].where;
       expect(whereArg).toEqual({
-        and: [{ taskSlug: { equals: "translate_document" } }, { completedAt: { exists: false } }],
+        and: [
+          {
+            or: [
+              { workflowSlug: { equals: "translate_document_locales" } },
+              { taskSlug: { equals: "translate_document" } },
+            ],
+          },
+          { completedAt: { exists: false } },
+        ],
       });
       expect(JSON.stringify(whereArg)).not.toContain("collection_id");
     });
diff --git a/packages/payload-plugin-translator/src/server/modules/task-runner/payload-jobs-runner/PayloadJobsTaskRunner.ts b/packages/payload-plugin-translator/src/server/modules/task-runner/payload-jobs-runner/PayloadJobsTaskRunner.ts
index 1d4d5e5d..4d8a6b25 100644
--- a/packages/payload-plugin-translator/src/server/modules/task-runner/payload-jobs-runner/PayloadJobsTaskRunner.ts
+++ b/packages/payload-plugin-translator/src/server/modules/task-runner/payload-jobs-runner/PayloadJobsTaskRunner.ts
@@ -2,21 +2,49 @@ import type { Payload, Where, CollectionSlug } from "payload";
 
 import type { TaskFilter, TaskRunner } from "../TaskRunner.interface";
 import { toTaskFilter } from "../toTaskFilter";
-import type { Task, TaskInput, RunResult, ID } from "../types";
-import type { PayloadJobsRunnerConfig, PayloadJob } from "./types";
-import { normalizeJob } from "./normalizeJob";
+import type { Task, TaskInput, RunResult } from "../types";
+import type { PayloadJobsRunnerConfig, PayloadJob, StoredWorkflowInput } from "./types";
+import { normalizeJobLocales } from "./normalizeJob";
+import { planEnqueue } from "./planEnqueue";
+import type { RequestShape } from "./planEnqueue";
+import { readCollectionRef } from "./readCollectionRef";
 
-// A translation job's supersession identity: same document AND same target locale. IDs are
-// String()-normalized to match the stored (string) form, so a number id compares equal to its
-// persisted job.
-const documentLocaleKey = (collectionId: ID, targetLng: string): string =>
-  `${String(collectionId)}:${targetLng}`;
+const APPEND_ATTEMPTS = 2;
 
 /**
- * TaskRunner implementation using Payload Jobs.
- *
- * Handles queuing, cancellation, status tracking, and execution of translation tasks.
+ * Groups served at once: a `select_all` enqueue can span thousands, each costing two writes and a
+ * read, and the ceiling is the database pool rather than the CPU.
  */
+const ENQUEUE_CONCURRENCY = 10;
+
+type QueueWorkflow = (args: {
+  workflow: string;
+  queue: string;
+  waitUntil?: Date;
+  input: StoredWorkflowInput;
+}) => Promise<unknown>;
+
+function requestShape(task: TaskInput): RequestShape {
+  return {
+    collectionSlug: task.collectionSlug,
+    collectionId: String(task.collectionId),
+    sourceLng: task.sourceLng,
+    strategy: task.strategy,
+    publishOnTranslation: task.publishOnTranslation,
+  };
+}
+
+/** `pickHost` matches on the same shape, so no two groups can pick the same host job — which is what
+ * makes the parallel `serve` calls safe. */
+function requestKey(task: TaskInput): string {
+  return JSON.stringify(requestShape(task));
+}
+
+function documentKey(collectionSlug: string, collectionId: string): string {
+  // NUL: no slug or id can contain it, so two different documents cannot produce one key.
+  return `${collectionSlug}\u0000${collectionId}`;
+}
+
 export class PayloadJobsTaskRunner implements TaskRunner {
   constructor(
     private readonly payload: Payload,
@@ -24,138 +52,206 @@ export class PayloadJobsTaskRunner implements TaskRunner {
   ) {}
 
   async enqueue(tasks: TaskInput[]): Promise<void> {
-    const byCollection = this.groupByCollection(tasks);
-
-    for (const [collectionSlug, items] of byCollection) {
-      const documentIds = items.map((t) => t.collectionId);
-      // Finished jobs must stay out of this set: superseding deletes (`cancelAndDeleteJobs` reaches
-      // `payload.delete`), so a completed job for the same (document, locale) would be erased along
-      // with the pending one.
-      const existing = await this.findByCollection(collectionSlug, {
-        documentIds,
-        excludeCompleted: true,
-      });
-      // Supersede only jobs for the SAME (document, target locale) being re-enqueued — never a
-      // concurrent job for a *different* locale of the same document. Cancelling per-document would
-      // kill an in-flight translation of another locale (the concurrent re-translate bug).
-      const supersededKeys = new Set(
-        items.map((t) => documentLocaleKey(t.collectionId, t.targetLng))
-      );
-      const toCancel = existing.filter((t) =>
-        supersededKeys.has(documentLocaleKey(t.input.collectionId, t.input.targetLng))
+    const byRequest = new Map<string, TaskInput[]>();
+    for (const task of tasks) {
+      const key = requestKey(task);
+      const group = byRequest.get(key) ?? [];
+      group.push(task);
+      byRequest.set(key, group);
+    }
+
+    const live = await this.findRawJobs({ completedAt: { exists: false } });
+    const liveByDocument = new Map<string, PayloadJob[]>();
+    for (const job of live) {
+      const { collectionSlug, collectionId } = readCollectionRef(job.input);
+      const key = documentKey(collectionSlug, collectionId);
+      liveByDocument.set(key, [...(liveByDocument.get(key) ?? []), job]);
+    }
+    const exclusiveQueue = Boolean(this.payload.config.jobs?.enableConcurrencyControl);
+
+    const groups = [...byRequest.values()];
+    for (let i = 0; i < groups.length; i += ENQUEUE_CONCURRENCY) {
+      await Promise.all(
+        groups
+          .slice(i, i + ENQUEUE_CONCURRENCY)
+          .map((group) => this.serve(group, liveByDocument, exclusiveQueue))
       );
-      if (toCancel.length > 0) {
-        await this.cancelAndDeleteJobs(toCancel.map((t) => t.id));
-      }
     }
+  }
+
+  private async serve(
+    group: TaskInput[],
+    liveByDocument: Map<string, PayloadJob[]>,
+    exclusiveQueue: boolean
+  ): Promise<void> {
+    const [first] = group;
+    const request = requestShape(first);
+    const plan = planEnqueue({
+      live: liveByDocument.get(documentKey(request.collectionSlug, request.collectionId)) ?? [],
+      request,
+      requested: group.map((t) => t.targetLng),
+      exclusiveQueue,
+    });
 
-    await Promise.all(
-      tasks.map((task) =>
-        this.payload.jobs.queue({
-          task: this.config.taskName,
-          queue: this.config.queueName,
-          // Debounce: when set, Payload holds the job until this instant. A superseding enqueue for
-          // the same (document, targetLng) cancels the pending delayed job first (see enqueue above),
-          // so rapid source edits coalesce to the final one. Undefined for the manual path.
-          waitUntil: task.waitUntil,
-          input: {
-            // Flat text reference (ID-agnostic). Stored as a string — no
-            // relationship type validation against the collection's ID type,
-            // which is what previously left number-id jobs stuck in processing.
-            // This is the single write boundary, so `String(...)` here is the
-            // one place IDs are normalized for storage.
-            collection_slug: task.collectionSlug,
-            collection_id: String(task.collectionId),
-            source_lng: task.sourceLng,
-            target_lng: task.targetLng,
-            strategy: task.strategy,
-            publish_on_translation: task.publishOnTranslation,
-          },
-        })
-      )
-    );
+    const undelivered = plan.host
+      ? await this.extendJob(plan.host, plan.append, first.waitUntil)
+      : [];
+    const queue = [...plan.queue, ...undelivered];
+    if (queue.length > 0) await this.queueWorkflow(request, queue, first.waitUntil);
+  }
+
+  private async extendJob(job: PayloadJob, locales: string[], waitUntil?: Date): Promise<string[]> {
+    let current = job;
+    let undelivered = locales;
+    // `input` is one JSON column, so a concurrent append replaces the whole list; the union makes a
+    // retry from the stored row harmless.
+    for (let attempt = 0; attempt < APPEND_ATTEMPTS; attempt++) {
+      const listed = current.input?.target_lngs ?? [];
+      const missing = locales.filter((locale) => !listed.includes(locale));
+      const debounce = waitUntil && !current.processing ? waitUntil.toISOString() : undefined;
+      if (missing.length === 0 && !debounce) return [];
+
+      // Not `payload.update`: it rewrites the whole row and reverts log entries written in between.
+      // See D2 of docs/plans/2026-09-08-one-live-job-per-document.task.md.
+      await this.payload.db.updateOne({
+        collection: this.config.jobsCollection,
+        id: job.id,
+        data: {
+          input: { ...current.input, target_lngs: [...listed, ...missing] },
+          ...(debounce ? { waitUntil: debounce } : {}),
+        },
+        returning: false,
+      });
+
+      const reread = await this.findJobById(job.id);
+      if (!reread || reread.completedAt) return locales;
+      current = reread;
+
+      const stored = new Set(current.input?.target_lngs);
+      undelivered = locales.filter((locale) => !stored.has(locale));
+      if (undelivered.length === 0) return [];
+    }
+    return undelivered;
+  }
+
+  private async queueWorkflow(
+    request: RequestShape,
+    targetLngs: string[],
+    waitUntil?: Date
+  ): Promise<void> {
+    const input: StoredWorkflowInput = {
+      collection_slug: request.collectionSlug,
+      collection_id: request.collectionId,
+      source_lng: request.sourceLng,
+      target_lngs: targetLngs,
+      strategy: request.strategy,
+      publish_on_translation: request.publishOnTranslation,
+    };
+
+    // Cast: `jobs.queue` is typed over the host's generated slugs, which cannot include a workflow
+    // registered at config time.
+    const queueJob = this.payload.jobs.queue as unknown as QueueWorkflow;
+    await queueJob({
+      workflow: this.config.workflowName,
+      queue: this.config.queueName,
+      waitUntil,
+      input,
+    });
   }
 
   async cancel(taskIds: string[]): Promise<void> {
     if (taskIds.length === 0) return;
-    await this.cancelAndDeleteJobs(taskIds);
+
+    // Mark then delete: the delete alone would take the row out of the status feed under
+    // `deleteJobOnComplete: false` without recording why it went. The mark does not reach a running
+    // handler — see D1 of docs/plans/2026-09-08-one-live-job-per-document.task.md.
+    await this.payload.jobs.cancel({
+      where: { id: { in: taskIds } },
+      queue: this.config.queueName,
+    });
+
+    await this.payload.delete({
+      collection: this.config.jobsCollection,
+      where: { and: [this.ownJobs(), { id: { in: taskIds } }] },
+    });
   }
 
   async run(taskId: string): Promise<RunResult> {
-    const tasks = await this.findJobsInternal({ id: { equals: taskId } }, { limit: 1 });
-    const task = tasks[0];
-
-    if (!task) {
+    const job = await this.findJobById(taskId);
+    if (!job) {
       return { success: false, error: "not_found" };
     }
-    if (task.completedAt) {
+    if (job.completedAt) {
       return { success: false, error: "already_completed" };
     }
-    if (task.status === "running") {
-      // The picker below selects only `processing: false`, so a stale lock must be cleared first.
-      if (!this.isStale(task.updatedAt)) {
-        return { success: false, error: "already_running" };
-      }
-      await this.resetProcessing({ id: { equals: taskId } });
+    if (job.processing && !this.isStale(job.updatedAt)) {
+      return { success: false, error: "already_running" };
+    }
+    if (job.processing || job.error) {
+      await this.clearPickerBlockers(taskId);
     }
 
-    // `where` picker, not `payload.jobs.runByID({ id })`: in `runJobs` the guard block
-    // (processing:false, hasError not true, waitUntil due) is built only for the non-id branch, so
-    // the id path would re-run a job that already exhausted its retries. Checked against payload
-    // 3.84.1.
-    await this.payload.jobs.run({
+    // Not `jobs.runByID`: payload 3.84.1 builds the picker guard (processing / hasError / waitUntil)
+    // only on the `where` path, so the id path re-runs a job that exhausted its retries.
+    const result = (await this.payload.jobs.run({
       queue: this.config.queueName,
       where: { id: { equals: taskId } },
       limit: 1,
-    });
+    })) as { jobStatus?: Record<string, unknown> };
+
+    const pickerTookNothing = Object.keys(result?.jobStatus ?? {}).length === 0;
+    if (pickerTookNothing) {
+      return { success: false, error: "already_running" };
+    }
     return { success: true };
   }
 
   /**
-   * Clear stale `processing` locks — still processing, not completed, `updatedAt` older than
-   * `staleJobTimeoutMs` — so abandoned jobs are eligible for the autorun picker again. A job that
-   * exhausted its retries carries `hasError: true` and stays excluded from autorun even after its
-   * lock is cleared; only a manual `run()` recovers it.
+   * A job that exhausted its retries carries `hasError: true` and stays excluded from the picker even
+   * after its lock is cleared; only a manual `run()` recovers it.
    * @returns how many locks were cleared.
    */
   async reclaimStaleJobs(): Promise<number> {
     const cutoff = new Date(Date.now() - this.config.staleJobTimeoutMs).toISOString();
-    return this.resetProcessing({
-      and: [
-        { taskSlug: { equals: this.config.taskName } },
-        { processing: { equals: true } },
-        { completedAt: { exists: false } },
-        { updatedAt: { less_than: cutoff } },
-      ],
-    });
-  }
-
-  /** Clears the `processing` lock on every job matching `where`. `depth: 0` — only the count is read. */
-  private async resetProcessing(where: Where): Promise<number> {
     const result = await this.payload.update({
       collection: this.config.jobsCollection,
       depth: 0,
-      where,
+      where: {
+        and: [
+          this.ownJobs(),
+          { processing: { equals: true } },
+          { completedAt: { exists: false } },
+          { updatedAt: { less_than: cutoff } },
+        ],
+      },
       data: { processing: false },
     });
     return result.docs.length;
   }
 
+  /**
+   * `payload.update`, not the adapter write `extendJob` uses, so the jobs collection's `beforeChange`
+   * hook still runs — it is what keeps a cancelled job cancelled.
+   */
+  private async clearPickerBlockers(taskId: string): Promise<void> {
+    await this.payload.update({
+      collection: this.config.jobsCollection,
+      depth: 0,
+      where: { id: { equals: taskId } },
+      data: { processing: false, hasError: false, error: null, waitUntil: null },
+    });
+  }
+
   private isStale(updatedAt: string): boolean {
     const parsed = Date.parse(updatedAt);
-    // Unknown/corrupt timestamp → treat as stale so the job can be recovered
-    // rather than permanently refused as already-running.
     if (Number.isNaN(parsed)) return true;
     return Date.now() - parsed > this.config.staleJobTimeoutMs;
   }
 
   /**
-   * Find translation jobs for a collection.
-   *
-   * Only `taskSlug` and `completedAt` reach the database; slug and document ids are matched in memory
-   * because a job's collection reference may sit in either the flat-text fields or the legacy
-   * relationship shape (`readCollectionRef`), so a `where` on `input.collection_slug` would silently
-   * drop every pre-migration job. `excludeCompleted` is what bounds the read — see issue #108.
+   * Matched in memory, not in a `where`: a job's collection reference sits either in the flat text
+   * fields or in the legacy relationship shape (see `readCollectionRef`), so `input.collection_slug`
+   * as a filter would silently drop every pre-migration job.
    */
   async findByCollection(
     collectionSlug: CollectionSlug,
@@ -163,55 +259,44 @@ export class PayloadJobsTaskRunner implements TaskRunner {
   ): Promise<Task[]> {
     const { documentIds, excludeCompleted } = toTaskFilter(filter);
     const where = excludeCompleted ? { completedAt: { exists: false } } : undefined;
-    const all = await this.findJobsInternal(where, { pagination: false });
-    const bySlug = all.filter((t) => t.input.collectionSlug === collectionSlug);
-    if (!documentIds?.length) return bySlug;
-    const wanted = new Set(documentIds.map(String));
-    return bySlug.filter((t) => wanted.has(t.input.collectionId));
+    const wanted = documentIds?.length ? new Set(documentIds.map(String)) : undefined;
+    const jobs = await this.findRawJobs(where);
+    return jobs
+      .filter((job) => {
+        const ref = readCollectionRef(job.input);
+        return ref.collectionSlug === collectionSlug && (!wanted || wanted.has(ref.collectionId));
+      })
+      .flatMap(normalizeJobLocales);
   }
 
-  private groupByCollection(tasks: TaskInput[]): Map<CollectionSlug, TaskInput[]> {
-    const map = new Map<CollectionSlug, TaskInput[]>();
-    for (const task of tasks) {
-      const existing = map.get(task.collectionSlug) ?? [];
-      existing.push(task);
-      map.set(task.collectionSlug, existing);
-    }
-    return map;
+  private ownJobs(): Where {
+    // Pre-workflow jobs are still in the table: docs/DEPRECATIONS.md#jobs-per-locale-task-shape
+    return {
+      or: [
+        { workflowSlug: { equals: this.config.workflowName } },
+        { taskSlug: { equals: this.config.taskName } },
+      ],
+    };
   }
 
-  private async cancelAndDeleteJobs(taskIds: string[]): Promise<void> {
-    if (taskIds.length === 0) return;
-
-    // Both, in this order: `jobs.cancel` only writes `{ error: { cancelled: true }, hasError: true,
-    // processing: false }`, which is what signals a running handler to abort. The delete then removes
-    // the row — under `deleteJobOnComplete: false` a cancelled job would otherwise sit in the status
-    // feed forever.
-    await this.payload.jobs.cancel({
-      where: { id: { in: taskIds } },
-      queue: this.config.queueName,
-    });
-
-    await this.payload.delete({
-      collection: this.config.jobsCollection,
-      where: { id: { in: taskIds } },
-    });
+  private async findJobById(id: string): Promise<PayloadJob | undefined> {
+    const [job] = await this.findRawJobs({ id: { equals: id } });
+    return job;
   }
 
-  private async findJobsInternal(
-    where?: Where,
-    params?: { limit?: number; pagination?: boolean }
-  ): Promise<Task[]> {
-    const and: Where[] = [{ taskSlug: { equals: this.config.taskName } }];
+  private async findRawJobs(where?: Where): Promise<PayloadJob[]> {
+    const and: Where[] = [this.ownJobs()];
     if (where) and.push(where);
 
     const response = await this.payload.find({
       collection: this.config.jobsCollection,
-      limit: params?.limit,
-      pagination: params?.pagination,
+      // The legacy `input.collection` is a declared relationship; at the default depth Payload
+      // populates it, and `readCollectionRef` would then read a document where it wants an id.
+      depth: 0,
+      pagination: false,
       where: { and },
     });
 
-    return (response.docs as PayloadJob[]).map(normalizeJob);
+    return response.docs as PayloadJob[];
   }
 }
diff --git a/packages/payload-plugin-translator/src/server/modules/task-runner/payload-jobs-runner/normalizeJob.test.ts b/packages/payload-plugin-translator/src/server/modules/task-runner/payload-jobs-runner/normalizeJob.test.ts
index 76773258..64a71120 100644
--- a/packages/payload-plugin-translator/src/server/modules/task-runner/payload-jobs-runner/normalizeJob.test.ts
+++ b/packages/payload-plugin-translator/src/server/modules/task-runner/payload-jobs-runner/normalizeJob.test.ts
@@ -1,6 +1,6 @@
 import { describe, it, expect } from "vitest";
 import type { CollectionSlug } from "payload";
-import { normalizeJob } from "./normalizeJob";
+import { normalizeJob, normalizeJobLocales } from "./normalizeJob";
 import type { PayloadJob } from "./types";
 
 describe("normalizeJob", () => {
@@ -265,3 +265,93 @@ describe("normalizeJob", () => {
     });
   });
 });
+
+describe("normalizeJobLocales", () => {
+  const workflowJob: PayloadJob = {
+    id: "job-999",
+    createdAt: "2024-01-01T00:00:00Z",
+    updatedAt: "2024-01-01T00:05:00Z",
+    input: {
+      collection_slug: "posts",
+      collection_id: "doc-456",
+      source_lng: "en",
+      target_lngs: ["de", "fr", "es"],
+      strategy: "overwrite",
+    },
+  };
+
+  it("expands a pre-workflow job to itself", () => {
+    const legacy: PayloadJob = {
+      id: "job-1",
+      createdAt: "2024-01-01T00:00:00Z",
+      updatedAt: "2024-01-01T00:00:00Z",
+      input: { collection_slug: "posts", collection_id: "doc-1", target_lng: "de" },
+    };
+    const rows = normalizeJobLocales(legacy);
+    expect(rows).toEqual([normalizeJob(legacy)]);
+  });
+
+  it("gives every requested locale a row, in the requested order", () => {
+    const rows = normalizeJobLocales(workflowJob);
+    expect(rows.map((r) => r.input.targetLng)).toEqual(["de", "fr", "es"]);
+  });
+
+  it("reports each locale's own outcome from the job log, not the job's status", () => {
+    const rows = normalizeJobLocales({
+      ...workflowJob,
+      processing: true,
+      log: [
+        { state: "succeeded", completedAt: "2024-01-01T00:01:00Z", input: { target_lng: "de" } },
+        { state: "failed", completedAt: "2024-01-01T00:02:00Z", input: { target_lng: "fr" } },
+      ],
+    });
+    expect(rows.map((r) => [r.input.targetLng, r.status])).toEqual([
+      ["de", "completed"],
+      ["fr", "failed"],
+      ["es", "running"],
+    ]);
+  });
+
+  it("stamps completedAt only on a locale that succeeded", () => {
+    const rows = normalizeJobLocales({
+      ...workflowJob,
+      log: [
+        { state: "succeeded", completedAt: "2024-01-01T00:01:00Z", input: { target_lng: "de" } },
+        { state: "failed", completedAt: "2024-01-01T00:02:00Z", input: { target_lng: "fr" } },
+      ],
+    });
+    expect(rows[0].completedAt).toBe("2024-01-01T00:01:00Z");
+    expect(rows[1].completedAt).toBeUndefined();
+  });
+
+  it("keeps the failure on the locale that failed, off the ones that landed", () => {
+    const rows = normalizeJobLocales({
+      ...workflowJob,
+      error: { message: "provider refused fr" },
+      log: [
+        { state: "succeeded", completedAt: "2024-01-01T00:01:00Z", input: { target_lng: "de" } },
+        { state: "failed", completedAt: "2024-01-01T00:02:00Z", input: { target_lng: "fr" } },
+      ],
+    });
+    expect(rows[0].error).toBeUndefined();
+    expect(rows[1].error).toEqual({ message: "provider refused fr" });
+  });
+
+  it("takes a retried locale's most recent log entry", () => {
+    const rows = normalizeJobLocales({
+      ...workflowJob,
+      input: { ...workflowJob.input, target_lngs: ["de"] },
+      log: [
+        { state: "failed", completedAt: "2024-01-01T00:01:00Z", input: { target_lng: "de" } },
+        { state: "succeeded", completedAt: "2024-01-01T00:03:00Z", input: { target_lng: "de" } },
+      ],
+    });
+    expect(rows[0].status).toBe("completed");
+    expect(rows[0].completedAt).toBe("2024-01-01T00:03:00Z");
+  });
+
+  it("keeps the real job id on every row, because cancelling one cancels the job", () => {
+    const rows = normalizeJobLocales(workflowJob);
+    expect(rows.map((r) => r.id)).toEqual(["job-999", "job-999", "job-999"]);
+  });
+});
diff --git a/packages/payload-plugin-translator/src/server/modules/task-runner/payload-jobs-runner/normalizeJob.ts b/packages/payload-plugin-translator/src/server/modules/task-runner/payload-jobs-runner/normalizeJob.ts
index 40e2dcc2..7cdbc267 100644
--- a/packages/payload-plugin-translator/src/server/modules/task-runner/payload-jobs-runner/normalizeJob.ts
+++ b/packages/payload-plugin-translator/src/server/modules/task-runner/payload-jobs-runner/normalizeJob.ts
@@ -1,32 +1,7 @@
 import type { Task, TaskStatus } from "../types";
-import type { PayloadJob } from "./types";
+import type { JobLogEntry, PayloadJob } from "./types";
 import { readCollectionRef } from "./readCollectionRef";
 
-/**
- * Transform Payload job to normalized Task
- */
-export function normalizeJob(job: PayloadJob): Task {
-  const { collectionSlug, collectionId } = readCollectionRef(job.input);
-
-  return {
-    id: job.id,
-    status: getJobStatus(job),
-    input: {
-      collectionSlug,
-      collectionId,
-      sourceLng: job.input?.source_lng ?? "",
-      targetLng: job.input?.target_lng ?? "",
-      strategy: (job.input?.strategy as "overwrite" | "skip_existing") ?? "overwrite",
-      publishOnTranslation: job.input?.publish_on_translation ?? false,
-    },
-    createdAt: job.createdAt,
-    updatedAt: job.updatedAt,
-    completedAt: job.completedAt ?? undefined,
-    error: job.error ? { message: extractErrorMessage(job.error) } : undefined,
-    cancelled: isCancelled(job.error),
-  };
-}
-
 function getJobStatus(job: PayloadJob): TaskStatus {
   if (job.completedAt) return "completed";
   if (job.processing) return "running";
@@ -46,7 +21,7 @@ function extractErrorMessage(error: unknown): string {
   return "Unknown error";
 }
 
-function isCancelled(error: unknown): boolean {
+export function isCancelled(error: unknown): boolean {
   return (
     error !== null &&
     typeof error === "object" &&
@@ -55,3 +30,64 @@ function isCancelled(error: unknown): boolean {
     error.cancelled
   );
 }
+
+export function normalizeJob(job: PayloadJob): Task {
+  const { collectionSlug, collectionId } = readCollectionRef(job.input);
+
+  return {
+    id: job.id,
+    status: getJobStatus(job),
+    input: {
+      collectionSlug,
+      collectionId,
+      sourceLng: job.input?.source_lng ?? "",
+      targetLng: job.input?.target_lng ?? "",
+      strategy: (job.input?.strategy as "overwrite" | "skip_existing") ?? "overwrite",
+      publishOnTranslation: job.input?.publish_on_translation ?? false,
+    },
+    createdAt: job.createdAt,
+    updatedAt: job.updatedAt,
+    completedAt: job.completedAt ?? undefined,
+    error: job.error ? { message: extractErrorMessage(job.error) } : undefined,
+    cancelled: isCancelled(job.error),
+  };
+}
+
+/**
+ * One {@link Task} per target locale, its state read from {@link latestLogByLocale}. A pre-workflow
+ * job carries a single `target_lng` and expands to itself.
+ */
+export function normalizeJobLocales(job: PayloadJob): Task[] {
+  const targets = job.input?.target_lngs;
+  if (!Array.isArray(targets) || targets.length === 0) return [normalizeJob(job)];
+
+  const base = normalizeJob(job);
+  const latestByLocale = latestLogByLocale(job);
+
+  return targets.map((targetLng) => {
+    const entry = latestByLocale.get(targetLng);
+    if (!entry) return { ...base, input: { ...base.input, targetLng } };
+    const succeeded = entry.state === "succeeded";
+    return {
+      ...base,
+      status: succeeded ? "completed" : "failed",
+      completedAt: succeeded ? (entry.completedAt ?? undefined) : undefined,
+      error: succeeded ? undefined : base.error,
+      cancelled: succeeded ? false : base.cancelled,
+      input: { ...base.input, targetLng },
+    };
+  });
+}
+
+/**
+ * Each locale's most recent log entry: Payload appends to `log` chronologically, so last-write-wins
+ * leaves the latest attempt.
+ */
+export function latestLogByLocale(job: PayloadJob): Map<string, JobLogEntry> {
+  const byLocale = new Map<string, JobLogEntry>();
+  for (const entry of job.log ?? []) {
+    const lng = entry?.input?.target_lng;
+    if (typeof lng === "string") byLocale.set(lng, entry);
+  }
+  return byLocale;
+}
diff --git a/packages/payload-plugin-translator/src/server/modules/task-runner/payload-jobs-runner/planEnqueue.test.ts b/packages/payload-plugin-translator/src/server/modules/task-runner/payload-jobs-runner/planEnqueue.test.ts
new file mode 100644
index 00000000..d20708e3
--- /dev/null
+++ b/packages/payload-plugin-translator/src/server/modules/task-runner/payload-jobs-runner/planEnqueue.test.ts
@@ -0,0 +1,144 @@
+import { describe, expect, it } from "vitest";
+
+import { planEnqueue } from "./planEnqueue";
+import type { RequestShape } from "./planEnqueue";
+import type { PayloadJob } from "./types";
+
+const request: RequestShape = {
+  collectionSlug: "posts",
+  collectionId: "doc-1",
+  sourceLng: "en",
+  strategy: "overwrite",
+  publishOnTranslation: false,
+};
+
+const storedInput = {
+  collection_slug: "posts",
+  collection_id: "doc-1",
+  source_lng: "en",
+  strategy: "overwrite",
+  publish_on_translation: false,
+  target_lngs: ["de"],
+};
+
+const job = (overrides: Partial<PayloadJob> = {}): PayloadJob => ({
+  id: "job-1",
+  createdAt: "2026-01-01T00:00:00Z",
+  updatedAt: "2026-01-01T00:00:00Z",
+  input: storedInput,
+  ...overrides,
+});
+
+const plan = (live: PayloadJob[], requested: string[], exclusiveQueue = false) =>
+  planEnqueue({ live, request, requested, exclusiveQueue });
+
+describe("planEnqueue", () => {
+  it("queues everything when the document has no live job", () => {
+    expect(plan([], ["de", "fr"])).toEqual({ host: null, append: [], queue: ["de", "fr"] });
+  });
+
+  it("adds locales the live job does not carry yet", () => {
+    const live = job();
+    expect(plan([live], ["fr", "es"])).toEqual({ host: live, append: ["fr", "es"], queue: [] });
+  });
+
+  it("does nothing for a locale the live job already owes", () => {
+    const live = job();
+    expect(plan([live], ["de"])).toEqual({ host: live, append: [], queue: [] });
+  });
+
+  it("gives a locale the live job has already translated a job of its own", () => {
+    const live = job({
+      log: [{ state: "succeeded", input: { target_lng: "de" } }],
+    });
+    expect(plan([live], ["de"])).toEqual({ host: live, append: [], queue: ["de"] });
+  });
+
+  it("splits a request across both when it mixes the two", () => {
+    const live = job({ log: [{ state: "succeeded", input: { target_lng: "de" } }] });
+    expect(plan([live], ["de", "fr"])).toEqual({ host: live, append: ["fr"], queue: ["de"] });
+  });
+
+  it("ignores duplicates in the request", () => {
+    const live = job();
+    expect(plan([live], ["fr", "fr"])).toEqual({ host: live, append: ["fr"], queue: [] });
+  });
+
+  it.each([
+    ["oldest first", ["old", "new"]],
+    ["newest first", ["new", "old"]],
+  ])("extends the newest live job, %s", (_label, order) => {
+    const byId = {
+      old: job({ id: "old", createdAt: "2026-01-01T00:00:00Z" }),
+      new: job({ id: "new", createdAt: "2026-01-02T00:00:00Z" }),
+    };
+    expect(
+      plan(
+        order.map((id) => byId[id as "old" | "new"]),
+        ["fr"]
+      ).host?.id
+    ).toBe("new");
+  });
+
+  it("picks the newest job that is usable, not the newest job", () => {
+    const newerButCancelled = job({
+      id: "cancelled",
+      createdAt: "2026-01-03T00:00:00Z",
+      error: { cancelled: true },
+    });
+    const usable = job({ id: "usable", createdAt: "2026-01-02T00:00:00Z" });
+    expect(plan([newerButCancelled, usable], ["fr"]).host?.id).toBe("usable");
+  });
+
+  it("does not extend a job in the pre-workflow shape", () => {
+    const legacy = job({
+      input: { ...storedInput, target_lngs: undefined, target_lng: "de" },
+    });
+    expect(plan([legacy], ["fr"])).toEqual({ host: null, append: [], queue: ["fr"] });
+  });
+
+  it("does not extend a cancelled job", () => {
+    const cancelled = job({ error: { cancelled: true } });
+    expect(plan([cancelled], ["fr"])).toEqual({ host: null, append: [], queue: ["fr"] });
+  });
+
+  describe("when the host enabled Payload's concurrency control", () => {
+    it("queues alongside a running job instead of extending it", () => {
+      const running = job({ processing: true });
+      expect(plan([running], ["fr"], true)).toEqual({ host: null, append: [], queue: ["fr"] });
+    });
+
+    it("merges an already-translated locale into the alongside job too", () => {
+      const running = job({
+        processing: true,
+        log: [{ state: "succeeded", input: { target_lng: "de" } }],
+      });
+      expect(plan([running], ["de", "fr"], true)).toEqual({
+        host: null,
+        append: [],
+        queue: ["fr", "de"],
+      });
+    });
+
+    it("still extends a job that has not started", () => {
+      const pending = job({ processing: false });
+      expect(plan([pending], ["fr"], true)).toEqual({ host: pending, append: ["fr"], queue: [] });
+    });
+  });
+
+  it("extends a running job when the host has NOT enabled concurrency control", () => {
+    const running = job({ processing: true });
+    expect(plan([running], ["fr"])).toEqual({ host: running, append: ["fr"], queue: [] });
+  });
+
+  describe("a job can only take locales from a request it matches", () => {
+    it.each([
+      ["a different strategy", { strategy: "skip_existing" }],
+      ["a different source locale", { source_lng: "fr" }],
+      ["a different publish flag", { publish_on_translation: true }],
+    ])("starts its own job for %s", (_label, storedOverride) => {
+      const live = job({ input: { ...storedInput, ...storedOverride } });
+      expect(plan([live], ["fr"])).toEqual({ host: null, append: [], queue: ["fr"] });
+    });
+  });
+});
diff --git a/packages/payload-plugin-translator/src/server/modules/task-runner/payload-jobs-runner/planEnqueue.ts b/packages/payload-plugin-translator/src/server/modules/task-runner/payload-jobs-runner/planEnqueue.ts
new file mode 100644
index 00000000..cfdb8725
--- /dev/null
+++ b/packages/payload-plugin-translator/src/server/modules/task-runner/payload-jobs-runner/planEnqueue.ts
@@ -0,0 +1,72 @@
+import type { CollectionSlug } from "payload";
+
+import { isCancelled, latestLogByLocale } from "./normalizeJob";
+import type { PayloadJob } from "./types";
+
+/** The part of a request that every locale in it shares. */
+export type RequestShape = {
+  collectionSlug: CollectionSlug;
+  collectionId: string;
+  sourceLng: string;
+  strategy: string;
+  publishOnTranslation: boolean;
+};
+
+/**
+ * `append` and `queue` are independent: a request can both extend a live job and need a job of its
+ * own, when a locale it asks for has already been translated by that job.
+ */
+export type EnqueuePlan = {
+  host: PayloadJob | null;
+  append: string[];
+  queue: string[];
+};
+
+/**
+ * One live job per document: a later request extends that job's locale list rather than replacing it.
+ *
+ * @param live - non-completed jobs for **this document only**; the caller filters by document.
+ * @param requested - target locales in request order; duplicates ignored.
+ * @param exclusiveQueue - the host's `enableConcurrencyControl`; with it on a running job is queued
+ *   behind rather than extended.
+ */
+export function planEnqueue(args: {
+  live: PayloadJob[];
+  request: RequestShape;
+  requested: string[];
+  exclusiveQueue: boolean;
+}): EnqueuePlan {
+  const requested = [...new Set(args.requested)];
+  const host = pickHost(args.live, args.request);
+  if (!host) return { host: null, append: [], queue: requested };
+
+  // Already-succeeded locales need a fresh job: the workflow passes the locale as the task id, so
+  // Payload's restoration would skip them (see the workflow handler in PayloadJobsRunnerProvider).
+  const settled = latestLogByLocale(host);
+  const done = requested.filter((locale) => settled.get(locale)?.state === "succeeded");
+  const listed = new Set(host.input?.target_lngs);
+  const missing = requested.filter((locale) => !listed.has(locale));
+
+  if (args.exclusiveQueue && host.processing) {
+    return { host: null, append: [], queue: [...missing, ...done] };
+  }
+  return { host, append: missing, queue: done };
+}
+
+/**
+ * A job carries one source locale, one strategy and one publish flag for all of its locales, so it can
+ * host only a request that chose the same three — otherwise the request runs under settings the user
+ * did not pick.
+ */
+function pickHost(live: PayloadJob[], request: RequestShape): PayloadJob | null {
+  const usable = live.filter(
+    (job) =>
+      Array.isArray(job.input?.target_lngs) &&
+      !isCancelled(job.error) &&
+      job.input?.source_lng === request.sourceLng &&
+      job.input?.strategy === request.strategy &&
+      (job.input?.publish_on_translation ?? false) === request.publishOnTranslation
+  );
+  const newestFirst = usable.sort((a, b) => Date.parse(b.createdAt) - Date.parse(a.createdAt));
+  return newestFirst[0] ?? null;
+}
diff --git a/packages/payload-plugin-translator/src/server/modules/task-runner/payload-jobs-runner/types.ts b/packages/payload-plugin-translator/src/server/modules/task-runner/payload-jobs-runner/types.ts
index 08cfeb8d..fbf79f85 100644
--- a/packages/payload-plugin-translator/src/server/modules/task-runner/payload-jobs-runner/types.ts
+++ b/packages/payload-plugin-translator/src/server/modules/task-runner/payload-jobs-runner/types.ts
@@ -72,6 +72,8 @@ export type PayloadJobsRunnerOptions = {
  */
 export type PayloadJobsRunnerConfig = {
   taskName: string;
+  /** Derived from `taskName`; deliberately not a plugin option. */
+  workflowName: string;
   queueName: string;
   jobsCollection: CollectionSlug;
   autoRun: false | Required<AutoRunConfig>;
@@ -79,10 +81,8 @@ export type PayloadJobsRunnerConfig = {
   retries?: PayloadJobsRunnerOptions["retries"];
 };
 
-/**
- * Raw Payload job structure
- */
 export type PayloadJob = {
+  log?: JobLogEntry[];
   id: string;
   completedAt?: string | null;
   createdAt: string;
@@ -104,7 +104,25 @@ export type PayloadJob = {
     };
     source_lng?: string;
     target_lng?: string;
+    target_lngs?: string[];
     strategy?: string;
     publish_on_translation?: boolean;
   };
 };
+
+/** Snake_case because Payload persists these keys verbatim in the job row. */
+export type StoredWorkflowInput = {
+  collection_slug: CollectionSlug;
+  collection_id: string;
+  source_lng: string;
+  target_lngs: string[];
+  strategy: string;
+  publish_on_translation: boolean;
+};
+
+/** Written by Payload only once a task settles — an absent entry means that locale has not run. */
+export type JobLogEntry = {
+  state: "succeeded" | "failed";
+  completedAt?: string | null;
+  input?: { target_lng?: string };
+};