diff --git a/apps/worker/__tests__/coexist-instagram-sync.test.ts b/apps/worker/__tests__/coexist-instagram-sync.test.ts
index c1b663088a..e5a9b5ed8b 100644
--- a/apps/worker/__tests__/coexist-instagram-sync.test.ts
+++ b/apps/worker/__tests__/coexist-instagram-sync.test.ts
@@ -63,6 +63,14 @@ vi.mock("@chatbotx.io/worker-config", () => ({
     add: mockQueueAdd,
     addBulk: mockQueueAddBulk,
   },
+  LowJobAction: {
+    coexistAttachmentDownload: "coexistAttachmentDownload",
+    updateContactAvatar: "updateContactAvatar",
+  },
+  lowQueue: {
+    add: mockQueueAdd,
+    addBulk: mockQueueAddBulk,
+  },
 }))
 
 vi.mock("../src/lib/logger", () => ({
diff --git a/apps/worker/__tests__/coexist-messenger-sync.test.ts b/apps/worker/__tests__/coexist-messenger-sync.test.ts
index 7daf4790d5..0a93143024 100644
--- a/apps/worker/__tests__/coexist-messenger-sync.test.ts
+++ b/apps/worker/__tests__/coexist-messenger-sync.test.ts
@@ -104,6 +104,14 @@ vi.mock("@chatbotx.io/worker-config", () => ({
     add: mockQueueAdd,
     addBulk: vi.fn().mockResolvedValue(undefined),
   },
+  LowJobAction: {
+    coexistAttachmentDownload: "coexistAttachmentDownload",
+    updateContactAvatar: "updateContactAvatar",
+  },
+  lowQueue: {
+    add: vi.fn().mockResolvedValue(undefined),
+    addBulk: vi.fn().mockResolvedValue(undefined),
+  },
 }))
 
 vi.mock("@chatbotx.io/database/schema", () => ({
diff --git a/apps/worker/__tests__/coexist-whatsapp-flush-lifecycle.test.ts b/apps/worker/__tests__/coexist-whatsapp-flush-lifecycle.test.ts
index fcef438288..294fe6307d 100644
--- a/apps/worker/__tests__/coexist-whatsapp-flush-lifecycle.test.ts
+++ b/apps/worker/__tests__/coexist-whatsapp-flush-lifecycle.test.ts
@@ -151,6 +151,11 @@ vi.mock("@chatbotx.io/worker-config", () => ({
     coexistAttachmentDownload: "coexistAttachmentDownload",
   },
   integrationQueue: { add: mockQueueAdd, addBulk: vi.fn() },
+  LowJobAction: {
+    coexistAttachmentDownload: "coexistAttachmentDownload",
+    updateContactAvatar: "updateContactAvatar",
+  },
+  lowQueue: { add: vi.fn(), addBulk: vi.fn() },
 }))
 
 // Carries the real schema forward and overrides only the models these tests
diff --git a/apps/worker/__tests__/coexist-whatsapp-flush.test.ts b/apps/worker/__tests__/coexist-whatsapp-flush.test.ts
index 0f647e1c88..aaf2fc7ca7 100644
--- a/apps/worker/__tests__/coexist-whatsapp-flush.test.ts
+++ b/apps/worker/__tests__/coexist-whatsapp-flush.test.ts
@@ -190,6 +190,11 @@ vi.mock("@chatbotx.io/worker-config", () => ({
     coexistMessengerSync: "coexistMessengerSync",
   },
   integrationQueue: { add: mockQueueAdd },
+  LowJobAction: {
+    coexistAttachmentDownload: "coexistAttachmentDownload",
+    updateContactAvatar: "updateContactAvatar",
+  },
+  lowQueue: { add: vi.fn(), addBulk: vi.fn() },
 }))
 
 // Carries the real schema forward and overrides only the models these tests
diff --git a/apps/worker/__tests__/enqueue-attachment-downloads.test.ts b/apps/worker/__tests__/enqueue-attachment-downloads.test.ts
new file mode 100644
index 0000000000..6fb779a56e
--- /dev/null
+++ b/apps/worker/__tests__/enqueue-attachment-downloads.test.ts
@@ -0,0 +1,100 @@
+import { beforeEach, describe, expect, it, vi } from "vitest"
+
+const { mockAddBulk } = vi.hoisted(() => ({
+  mockAddBulk: vi.fn(),
+}))
+
+// Coexist attachment downloads are light but high-volume and low-priority, so
+// they must be enqueued on the dedicated `low` queue — never the
+// latency-sensitive `integration` queue that drives customer replies.
+vi.mock("@chatbotx.io/worker-config", () => ({
+  LowJobAction: {
+    coexistAttachmentDownload: "coexistAttachmentDownload",
+    updateContactAvatar: "updateContactAvatar",
+  },
+  lowQueue: { addBulk: mockAddBulk },
+}))
+
+import { enqueueAttachmentDownloadJobs } from "../src/integration/handlers/coexist/enqueue-attachment-downloads"
+
+describe("enqueueAttachmentDownloadJobs", () => {
+  beforeEach(() => {
+    vi.clearAllMocks()
+    mockAddBulk.mockResolvedValue(undefined)
+  })
+
+  it.each([
+    "messenger",
+    "whatsapp",
+    "instagram",
+  ] as const)("routes %s attachment jobs to the low queue with preserved options", async (channel) => {
+    await enqueueAttachmentDownloadJobs({
+      workspaceId: "ws-1",
+      integrationId: "int-1",
+      channel,
+      attachmentIds: ["a1", "a2"],
+    })
+
+    expect(mockAddBulk).toHaveBeenCalledTimes(1)
+    const jobs = mockAddBulk.mock.calls[0][0]
+    expect(jobs).toHaveLength(2)
+    expect(jobs[0]).toEqual({
+      name: "coexistAttachmentDownload",
+      data: {
+        type: "coexistAttachmentDownload",
+        data: {
+          attachmentId: "a1",
+          workspaceId: "ws-1",
+          channel,
+          integrationId: "int-1",
+        },
+      },
+      opts: {
+        jobId: "att-a1",
+        attempts: 5,
+        backoff: { type: "exponential", delay: 30_000 },
+        removeOnComplete: true,
+        removeOnFail: { count: 100 },
+      },
+    })
+  })
+
+  it("produces jobIds free of the ':' delimiter BullMQ forbids", async () => {
+    await enqueueAttachmentDownloadJobs({
+      workspaceId: "ws-1",
+      integrationId: "int-1",
+      channel: "messenger",
+      attachmentIds: ["a1", "a2", "a3"],
+    })
+
+    const jobs = mockAddBulk.mock.calls[0][0]
+    for (const job of jobs) {
+      expect(job.opts.jobId).not.toContain(":")
+    }
+  })
+
+  it("no-ops without touching the queue when there are no attachments", async () => {
+    await enqueueAttachmentDownloadJobs({
+      workspaceId: "ws-1",
+      integrationId: "int-1",
+      channel: "whatsapp",
+      attachmentIds: [],
+    })
+
+    expect(mockAddBulk).not.toHaveBeenCalled()
+  })
+
+  it("propagates an addBulk failure so callers control their own error policy", async () => {
+    const failure = new Error("redis down")
+    mockAddBulk.mockRejectedValueOnce(failure)
+
+    await expect(
+      enqueueAttachmentDownloadJobs({
+        workspaceId: "ws-1",
+        integrationId: "int-1",
+        channel: "instagram",
+        attachmentIds: ["a1"],
+      }),
+    ).rejects.toThrow(failure)
+  })
+})
diff --git a/apps/worker/__tests__/enqueue-avatar-jobs.test.ts b/apps/worker/__tests__/enqueue-avatar-jobs.test.ts
new file mode 100644
index 0000000000..942446a43a
--- /dev/null
+++ b/apps/worker/__tests__/enqueue-avatar-jobs.test.ts
@@ -0,0 +1,87 @@
+import { beforeEach, describe, expect, it, vi } from "vitest"
+
+const { mockAddBulk } = vi.hoisted(() => ({
+  mockAddBulk: vi.fn(),
+}))
+
+// The avatar backfill jobs are light but high-volume and low-priority, so they
+// must be enqueued on the dedicated `low` queue — never the latency-sensitive
+// `integration` queue that drives customer replies.
+vi.mock("@chatbotx.io/worker-config", () => ({
+  LowJobAction: {
+    updateContactAvatar: "updateContactAvatar",
+    coexistAttachmentDownload: "coexistAttachmentDownload",
+  },
+  lowQueue: { addBulk: mockAddBulk },
+}))
+
+vi.mock("../src/lib/logger", () => ({
+  logger: { error: vi.fn(), warn: vi.fn(), info: vi.fn() },
+}))
+
+import { enqueueContactAvatarJobs } from "../src/integration/handlers/contact/enqueue-avatar-jobs"
+
+describe("enqueueContactAvatarJobs", () => {
+  beforeEach(() => {
+    vi.clearAllMocks()
+    mockAddBulk.mockResolvedValue(undefined)
+  })
+
+  it("enqueues one updateContactAvatar job per contact on the low queue", async () => {
+    await enqueueContactAvatarJobs({
+      workspaceId: "ws-1",
+      contactInboxIds: new Map([
+        ["source-a", { contactInboxId: "ci-a" }],
+        ["source-b", { contactInboxId: "ci-b" }],
+      ]),
+    })
+
+    expect(mockAddBulk).toHaveBeenCalledTimes(1)
+    const jobs = mockAddBulk.mock.calls[0][0]
+    expect(jobs).toHaveLength(2)
+    expect(jobs[0]).toMatchObject({
+      name: "updateContactAvatar",
+      data: {
+        type: "updateContactAvatar",
+        data: {
+          workspaceId: "ws-1",
+          contactInboxId: "ci-a",
+          sourceId: "source-a",
+        },
+      },
+      opts: { jobId: "update-avatar-ci-a" },
+    })
+  })
+
+  it("produces jobIds free of the ':' delimiter BullMQ forbids", async () => {
+    await enqueueContactAvatarJobs({
+      workspaceId: "ws-1",
+      contactInboxIds: new Map([["source-a", { contactInboxId: "ci-a" }]]),
+    })
+
+    const jobs = mockAddBulk.mock.calls[0][0]
+    for (const job of jobs) {
+      expect(job.opts.jobId).not.toContain(":")
+    }
+  })
+
+  it("no-ops without touching the queue when there are no contacts", async () => {
+    await enqueueContactAvatarJobs({
+      workspaceId: "ws-1",
+      contactInboxIds: new Map(),
+    })
+
+    expect(mockAddBulk).not.toHaveBeenCalled()
+  })
+
+  it("swallows an addBulk failure so the caller's run is never failed", async () => {
+    mockAddBulk.mockRejectedValueOnce(new Error("redis down"))
+
+    await expect(
+      enqueueContactAvatarJobs({
+        workspaceId: "ws-1",
+        contactInboxIds: new Map([["source-a", { contactInboxId: "ci-a" }]]),
+      }),
+    ).resolves.toBeUndefined()
+  })
+})
diff --git a/apps/worker/__tests__/low-worker-boot.test.ts b/apps/worker/__tests__/low-worker-boot.test.ts
new file mode 100644
index 0000000000..4e77ea731f
--- /dev/null
+++ b/apps/worker/__tests__/low-worker-boot.test.ts
@@ -0,0 +1,178 @@
+import { beforeEach, describe, expect, test, vi } from "vitest"
+
+// Boots the real `src/low/worker.ts` (which starts itself on import) and asserts
+// it creates exactly one BullMQ `Worker` on the `low` queue, routes each
+// LowJobAction to the correct shared handler, and gates every job behind
+// `withBlockedOwnerGuard`. All of the worker's imports are mocked to keep this a
+// fast, isolated unit test.
+
+type CapturedWorker = {
+  queueName: unknown
+  processor: (job: { data: unknown; id?: string }) => Promise<unknown>
+  options: Record<string, unknown>
+}
+
+const workerState = vi.hoisted(() => ({
+  capturedWorkers: [] as CapturedWorker[],
+  ensureBootstrapped: vi.fn(async () => undefined),
+  coexistAttachmentDownload: vi.fn(async () => undefined),
+  updateContactAvatar: vi.fn(async () => undefined),
+  withBlockedOwnerGuard: vi.fn(
+    async (_workspaceId: unknown, fn: () => Promise<unknown>) => await fn(),
+  ),
+  workerClose: vi.fn(async () => undefined),
+  workerOn: vi.fn(),
+}))
+
+vi.mock("bullmq", () => {
+  class WorkerMock {
+    close = workerState.workerClose
+    on = workerState.workerOn
+
+    constructor(
+      queueName: unknown,
+      processor: CapturedWorker["processor"],
+      options: Record<string, unknown>,
+    ) {
+      workerState.capturedWorkers.push({ queueName, processor, options })
+    }
+  }
+
+  return { Worker: WorkerMock }
+})
+
+vi.mock("@chatbotx.io/worker-config", () => ({
+  LowJobAction: {
+    coexistAttachmentDownload: "coexistAttachmentDownload",
+    updateContactAvatar: "updateContactAvatar",
+  },
+  queueNames: { enum: { low: "low" } },
+  defaultWorkerOptions: { concurrency: 5, removeOnComplete: { count: 1000 } },
+  getRedisConnection: vi.fn(() => ({})),
+}))
+
+vi.mock("@chatbotx.io/business", () => ({
+  withBlockedOwnerGuard: workerState.withBlockedOwnerGuard,
+}))
+
+vi.mock("../src/env", () => ({
+  env: { LOW_WORKER_CONCURRENCY: 30 },
+}))
+
+vi.mock("../src/lib/bootstrap", () => ({
+  ensureBootstrapped: workerState.ensureBootstrapped,
+}))
+
+vi.mock("../src/lib/logger", () => ({
+  logger: { error: vi.fn(), info: vi.fn(), warn: vi.fn() },
+}))
+
+vi.mock("../src/lib/run-job-with-audit-context", () => ({
+  runJobWithAuditContext: vi.fn(
+    async (_params: unknown, fn: () => Promise<unknown>) => await fn(),
+  ),
+}))
+
+vi.mock("../src/integration/handlers/coexist/attachment-download", () => ({
+  coexistAttachmentDownload: workerState.coexistAttachmentDownload,
+}))
+
+vi.mock("../src/integration/handlers/contact/update-avatar", () => ({
+  updateContactAvatar: workerState.updateContactAvatar,
+}))
+
+// Importing the worker module boots it exactly once (ESM module cache).
+await import("../src/low/worker")
+await vi.waitFor(() => {
+  expect(workerState.capturedWorkers).toHaveLength(1)
+})
+
+describe("low worker process boot", () => {
+  beforeEach(() => {
+    workerState.coexistAttachmentDownload.mockClear()
+    workerState.updateContactAvatar.mockClear()
+    workerState.withBlockedOwnerGuard.mockClear()
+    workerState.withBlockedOwnerGuard.mockImplementation(
+      async (_workspaceId: unknown, fn: () => Promise<unknown>) => await fn(),
+    )
+  })
+
+  test("boots exactly one Worker on the low queue at the env-tunable concurrency", () => {
+    expect(workerState.capturedWorkers).toHaveLength(1)
+    expect(workerState.capturedWorkers[0]?.queueName).toBe("low")
+    expect(workerState.capturedWorkers[0]?.options.concurrency).toBe(30)
+  })
+
+  test("routes coexistAttachmentDownload to its handler with the job payload", async () => {
+    const [worker] = workerState.capturedWorkers
+    const data = {
+      attachmentId: "att-1",
+      workspaceId: "ws-1",
+      channel: "messenger",
+      integrationId: "int-1",
+    }
+
+    await worker?.processor({
+      data: { type: "coexistAttachmentDownload", data },
+    })
+
+    expect(workerState.coexistAttachmentDownload).toHaveBeenCalledWith(data)
+    expect(workerState.updateContactAvatar).not.toHaveBeenCalled()
+    expect(workerState.withBlockedOwnerGuard).toHaveBeenCalledWith(
+      "ws-1",
+      expect.any(Function),
+    )
+  })
+
+  test("routes updateContactAvatar to its handler with the job payload", async () => {
+    const [worker] = workerState.capturedWorkers
+    const data = {
+      workspaceId: "ws-2",
+      contactInboxId: "ci-2",
+      sourceId: "src-2",
+    }
+
+    await worker?.processor({
+      data: { type: "updateContactAvatar", data },
+    })
+
+    expect(workerState.updateContactAvatar).toHaveBeenCalledWith(data)
+    expect(workerState.coexistAttachmentDownload).not.toHaveBeenCalled()
+    expect(workerState.withBlockedOwnerGuard).toHaveBeenCalledWith(
+      "ws-2",
+      expect.any(Function),
+    )
+  })
+
+  test("a frozen workspace short-circuits before any handler runs", async () => {
+    workerState.withBlockedOwnerGuard.mockImplementationOnce(
+      async () => undefined,
+    )
+    const [worker] = workerState.capturedWorkers
+
+    await worker?.processor({
+      data: {
+        type: "coexistAttachmentDownload",
+        data: {
+          attachmentId: "att-3",
+          workspaceId: "ws-3",
+          channel: "whatsapp",
+          integrationId: "int-3",
+        },
+      },
+    })
+
+    expect(workerState.coexistAttachmentDownload).not.toHaveBeenCalled()
+  })
+
+  test("an unknown action is a no-op — no handler is invoked", async () => {
+    const [worker] = workerState.capturedWorkers
+
+    await worker?.processor({
+      data: { type: "somethingElse", data: { workspaceId: "ws-4" } },
+    })
+
+    expect(workerState.coexistAttachmentDownload).not.toHaveBeenCalled()
+    expect(workerState.updateContactAvatar).not.toHaveBeenCalled()
+  })
+})
diff --git a/apps/worker/package.json b/apps/worker/package.json
index 880cb44ec6..b0a5ad179a 100644
--- a/apps/worker/package.json
+++ b/apps/worker/package.json
@@ -17,6 +17,7 @@
     "worker:events": "dotenv -e ../../.env -- tsx --watch src/events/worker.ts",
     "worker:heavy": "dotenv -e ../../.env -- tsx --watch src/heavy/worker.ts",
     "worker:integration": "dotenv -e ../../.env -- tsx --watch src/integration/worker.ts",
+    "worker:low": "dotenv -e ../../.env -- tsx --watch src/low/worker.ts",
     "worker:notification": "dotenv -e ../../.env -- tsx --watch src/notification/worker.ts",
     "worker:sequence-consumer": "dotenv -e ../../.env -- tsx --watch src/sequence-scheduler/worker-consumer.ts",
     "worker:sequence-producer": "dotenv -e ../../.env -- tsx --watch src/sequence-scheduler/worker-producer.ts",
diff --git a/apps/worker/src/env.ts b/apps/worker/src/env.ts
index 95d4cc776f..44fe594e32 100644
--- a/apps/worker/src/env.ts
+++ b/apps/worker/src/env.ts
@@ -21,6 +21,11 @@ export const env = createEnv({
       .min(1)
       .max(200)
       .default(10),
+    // Light-but-bulky, low-priority jobs (Coexist/Customer-Scan media backfill)
+    // run on their own `low` queue/worker so a historical-import burst never
+    // starves the latency-sensitive integration queue. I/O-bound → higher
+    // default than integration; tune per node bandwidth / Graph rate limits.
+    LOW_WORKER_CONCURRENCY: z.coerce.number().int().min(1).max(200).default(30),
     AI_AGENT_WORKER_CONCURRENCY: z.coerce
       .number()
       .int()
diff --git a/apps/worker/src/integration/handlers/coexist/attachment-download.ts b/apps/worker/src/integration/handlers/coexist/attachment-download.ts
index 8379d3720a..688216c9a5 100644
--- a/apps/worker/src/integration/handlers/coexist/attachment-download.ts
+++ b/apps/worker/src/integration/handlers/coexist/attachment-download.ts
@@ -9,7 +9,7 @@ import {
 } from "@chatbotx.io/integration-whatsapp"
 import { SdkException } from "@chatbotx.io/sdk"
 import { createId } from "@chatbotx.io/utils"
-import type { IntegrationJobCoexistAttachmentDownload } from "@chatbotx.io/worker-config"
+import type { LowJobCoexistAttachmentDownload } from "@chatbotx.io/worker-config"
 import imageSize from "image-size"
 import { logger } from "../../../lib/logger"
 
@@ -217,8 +217,7 @@ export const downloadWhatsappMedia = async (
   }
 }
 
-type AttachmentChannel =
-  IntegrationJobCoexistAttachmentDownload["data"]["channel"]
+type AttachmentChannel = LowJobCoexistAttachmentDownload["data"]["channel"]
 type BearerTokenAuth = { tokens: { accessToken: string } }
 type AttachmentDownloadContext = { auth: BearerTokenAuth }
 
@@ -273,7 +272,7 @@ const mediaDownloaders = {
  * — BullMQ `jobId: att-<id>` dedup plus this prefix guard cover idempotency.
  */
 export const coexistAttachmentDownload = async (
-  data: IntegrationJobCoexistAttachmentDownload["data"],
+  data: LowJobCoexistAttachmentDownload["data"],
 ): Promise<void> => {
   const { attachmentId, workspaceId, channel, integrationId } = data
 
diff --git a/apps/worker/src/integration/handlers/coexist/enqueue-attachment-downloads.ts b/apps/worker/src/integration/handlers/coexist/enqueue-attachment-downloads.ts
new file mode 100644
index 0000000000..1348876f93
--- /dev/null
+++ b/apps/worker/src/integration/handlers/coexist/enqueue-attachment-downloads.ts
@@ -0,0 +1,54 @@
+import {
+  LowJobAction,
+  type LowJobCoexistAttachmentDownload,
+  lowQueue,
+} from "@chatbotx.io/worker-config"
+
+type AttachmentDownloadChannel =
+  LowJobCoexistAttachmentDownload["data"]["channel"]
+
+/**
+ * Bulk-enqueue one `coexistAttachmentDownload` job per Attachment id onto the
+ * low-priority `low` queue — kept off the latency-sensitive `integration` queue
+ * so a historical-import burst never starves customer replies.
+ *
+ * Channel-agnostic: the caller passes its `channel`, so the Messenger, Instagram
+ * and WhatsApp coexist syncs share one enqueue path instead of each duplicating
+ * the same job shape/options. Idempotent — the jobId `att-<id>` dedups concurrent
+ * enqueues and the handler prefix-checks `originPath`, so re-enqueuing the same
+ * id is harmless.
+ *
+ * This builds and enqueues only; it does NOT catch. Each caller keeps its own
+ * failure policy: WhatsApp/Messenger swallow (best-effort — bytes stay pending,
+ * a later sync re-enqueues) while Instagram lets it propagate so the run fails
+ * and re-drives before the resume watermark advances past un-enqueued
+ * attachments.
+ */
+export const enqueueAttachmentDownloadJobs = async (input: {
+  workspaceId: string
+  integrationId: string
+  channel: AttachmentDownloadChannel
+  attachmentIds: string[]
+}): Promise<void> => {
+  const { workspaceId, integrationId, channel, attachmentIds } = input
+  if (attachmentIds.length === 0) {
+    return
+  }
+
+  await lowQueue.addBulk(
+    attachmentIds.map((attachmentId) => ({
+      name: LowJobAction.coexistAttachmentDownload,
+      data: {
+        type: LowJobAction.coexistAttachmentDownload,
+        data: { attachmentId, workspaceId, channel, integrationId },
+      },
+      opts: {
+        jobId: `att-${attachmentId}`,
+        attempts: 5,
+        backoff: { type: "exponential" as const, delay: 30_000 },
+        removeOnComplete: true,
+        removeOnFail: { count: 100 },
+      },
+    })),
+  )
+}
diff --git a/apps/worker/src/integration/handlers/coexist/instagram-sync.ts b/apps/worker/src/integration/handlers/coexist/instagram-sync.ts
index 77136304c2..45cbb6a6e3 100644
--- a/apps/worker/src/integration/handlers/coexist/instagram-sync.ts
+++ b/apps/worker/src/integration/handlers/coexist/instagram-sync.ts
@@ -21,6 +21,7 @@ import {
   createHistoricalIdFactory,
   type HistoricalMessage,
 } from "./bulk-historical-import"
+import { enqueueAttachmentDownloadJobs } from "./enqueue-attachment-downloads"
 import { instagramCoexistAdapter } from "./instagram-adapter"
 import { instagramFacebookCoexistAdapter } from "./instagram-facebook-adapter"
 import { splitDisplayName } from "./instagram-normalize"
@@ -357,29 +358,16 @@ const runInstagramCoexistPull = async <
       skippedTotal += pageSkipped
       failedTotal += pageFailed
 
-      if (attachmentIds.length > 0) {
-        await integrationQueue.addBulk(
-          attachmentIds.map((attachmentId) => ({
-            name: IntegrationJobAction.coexistAttachmentDownload,
-            data: {
-              type: IntegrationJobAction.coexistAttachmentDownload,
-              data: {
-                attachmentId,
-                workspaceId,
-                channel: "instagram" as const,
-                integrationId,
-              },
-            },
-            opts: {
-              jobId: `att-${attachmentId}`,
-              attempts: 5,
-              backoff: { type: "exponential", delay: 30_000 },
-              removeOnComplete: true,
-              removeOnFail: { count: 100 },
-            },
-          })),
-        )
-      }
+      // Not best-effort by design: a failed enqueue propagates so the run is
+      // marked failed and re-driven BEFORE the resume watermark advances past
+      // these attachments (updateProgress below writes lastSyncedAt). This
+      // preserves Instagram's stricter original behavior.
+      await enqueueAttachmentDownloadJobs({
+        workspaceId,
+        integrationId,
+        channel: "instagram",
+        attachmentIds,
+      })
 
       await coexistService.updateProgress({
         runId,
diff --git a/apps/worker/src/integration/handlers/coexist/messenger-sync.ts b/apps/worker/src/integration/handlers/coexist/messenger-sync.ts
index 7e9de6f548..d085c4384a 100644
--- a/apps/worker/src/integration/handlers/coexist/messenger-sync.ts
+++ b/apps/worker/src/integration/handlers/coexist/messenger-sync.ts
@@ -34,6 +34,7 @@ import {
   maxNumericId,
 } from "./bulk-historical-import"
 import { filterConversationWindow } from "./conversation-window"
+import { enqueueAttachmentDownloadJobs } from "./enqueue-attachment-downloads"
 import {
   fetchConvMessages,
   messengerAuthSchema,
@@ -466,38 +467,22 @@ async function runMessagesPhase(ctx: SyncContext): Promise<PhaseResult> {
       // One UPDATE per table for the whole chunk (not per conv/page in the loop).
       await applyCoexistActivityUpdates(activityUpdates, { workspaceId })
 
-      // Bulk-enqueue per-attachment download jobs. The handler is idempotent
-      // (prefix-checked + jobId-dedup'd), so a retry of this whole chunk
-      // re-enqueues the same jobIds harmlessly.
-      if (pageAttachmentIds.length > 0) {
-        try {
-          await integrationQueue.addBulk(
-            pageAttachmentIds.map((attachmentId) => ({
-              name: IntegrationJobAction.coexistAttachmentDownload,
-              data: {
-                type: IntegrationJobAction.coexistAttachmentDownload,
-                data: {
-                  attachmentId,
-                  workspaceId,
-                  channel: "messenger" as const,
-                  integrationId: ctx.integrationId,
-                },
-              },
-              opts: {
-                jobId: `att-${attachmentId}`,
-                attempts: 5,
-                backoff: { type: "exponential", delay: 30_000 },
-                removeOnComplete: true,
-                removeOnFail: { count: 100 },
-              },
-            })),
-          )
-        } catch (error) {
-          logger.error(
-            { error, runId, pageNumber, count: pageAttachmentIds.length },
-            "[coexist] Messenger attachment download enqueue failed — bytes left as pending",
-          )
-        }
+      // Bulk-enqueue per-attachment download jobs onto the low-priority queue.
+      // The handler is idempotent (prefix-checked + jobId-dedup'd), so a retry
+      // of this whole chunk re-enqueues the same jobIds harmlessly. Best-effort:
+      // a failed enqueue leaves the bytes pending and must not fail the page.
+      try {
+        await enqueueAttachmentDownloadJobs({
+          workspaceId,
+          integrationId: ctx.integrationId,
+          channel: "messenger",
+          attachmentIds: pageAttachmentIds,
+        })
+      } catch (error) {
+        logger.error(
+          { err: error, runId, pageNumber, count: pageAttachmentIds.length },
+          "[coexist] Messenger attachment download enqueue failed — bytes left as pending",
+        )
       }
 
       await coexistService.incrementProgress({
diff --git a/apps/worker/src/integration/handlers/coexist/whatsapp-flush.ts b/apps/worker/src/integration/handlers/coexist/whatsapp-flush.ts
index 3d5464dfef..467851f19e 100644
--- a/apps/worker/src/integration/handlers/coexist/whatsapp-flush.ts
+++ b/apps/worker/src/integration/handlers/coexist/whatsapp-flush.ts
@@ -17,6 +17,7 @@ import {
 } from "@chatbotx.io/worker-config"
 import { logger } from "../../../lib/logger"
 import { bulkImportHistorical } from "./bulk-historical-import"
+import { enqueueAttachmentDownloadJobs } from "./enqueue-attachment-downloads"
 import {
   abandon,
   type FlushContext,
@@ -80,47 +81,6 @@ const resolveTerminalStatus = (counters: {
   return "succeeded"
 }
 
-/**
- * Enqueues one download job per Attachment inserted by this batch (inline or
- * post-batch). Never throws: the bytes stay pending and the row is recoverable.
- */
-const enqueueAttachmentDownloads = async (
-  context: FlushContext,
-  attachmentIds: string[],
-): Promise<void> => {
-  if (attachmentIds.length === 0) {
-    return
-  }
-  try {
-    await integrationQueue.addBulk(
-      attachmentIds.map((attachmentId) => ({
-        name: IntegrationJobAction.coexistAttachmentDownload,
-        data: {
-          type: IntegrationJobAction.coexistAttachmentDownload,
-          data: {
-            attachmentId,
-            workspaceId: context.integration.workspaceId,
-            channel: "whatsapp" as const,
-            integrationId: context.integration.id,
-          },
-        },
-        opts: {
-          jobId: `att-${attachmentId}`,
-          attempts: 5,
-          backoff: { type: "exponential", delay: 30_000 },
-          removeOnComplete: true,
-          removeOnFail: { count: 100 },
-        },
-      })),
-    )
-  } catch (error) {
-    logger.error(
-      { error, runId: context.runId, count: attachmentIds.length },
-      "[coexist] WhatsApp attachment download enqueue failed — bytes left as pending",
-    )
-  }
-}
-
 /** Replays carried patches plus this batch's, and re-caps what stays pending. */
 const applyBatchPatches = async (
   context: FlushContext,
@@ -252,10 +212,25 @@ const importAndPatch = async (
 ): Promise<void> => {
   const batchResult = await importBatch(context, state, reduced, stagedRows)
   const patchAttachmentIds = await applyBatchPatches(context, state, reduced)
-  await enqueueAttachmentDownloads(context, [
+  // Best-effort: a failed enqueue leaves the bytes pending and recoverable, so
+  // it must never fail the flush chunk.
+  const attachmentIds = [
     ...batchResult.insertedAttachmentIds,
     ...patchAttachmentIds,
-  ])
+  ]
+  try {
+    await enqueueAttachmentDownloadJobs({
+      workspaceId: context.integration.workspaceId,
+      integrationId: context.integration.id,
+      channel: "whatsapp",
+      attachmentIds,
+    })
+  } catch (error) {
+    logger.error(
+      { err: error, runId: context.runId, count: attachmentIds.length },
+      "[coexist] WhatsApp attachment download enqueue failed — bytes left as pending",
+    )
+  }
 }
 
 /**
diff --git a/apps/worker/src/integration/handlers/contact/enqueue-avatar-jobs.ts b/apps/worker/src/integration/handlers/contact/enqueue-avatar-jobs.ts
index 269b34e54c..f85bd949dc 100644
--- a/apps/worker/src/integration/handlers/contact/enqueue-avatar-jobs.ts
+++ b/apps/worker/src/integration/handlers/contact/enqueue-avatar-jobs.ts
@@ -1,8 +1,5 @@
 import type { ChannelContactImportLink } from "@chatbotx.io/business"
-import {
-  IntegrationJobAction,
-  integrationQueue,
-} from "@chatbotx.io/worker-config"
+import { LowJobAction, lowQueue } from "@chatbotx.io/worker-config"
 import { logger } from "../../../lib/logger"
 
 /**
@@ -40,9 +37,9 @@ export const enqueueContactAvatarJobs = async (input: {
   }
 
   const avatarJobs = Array.from(contactInboxIds, ([sourceId, link]) => ({
-    name: IntegrationJobAction.updateContactAvatar,
+    name: LowJobAction.updateContactAvatar,
     data: {
-      type: IntegrationJobAction.updateContactAvatar,
+      type: LowJobAction.updateContactAvatar,
       data: {
         workspaceId,
         contactInboxId: link.contactInboxId,
@@ -58,7 +55,7 @@ export const enqueueContactAvatarJobs = async (input: {
   }))
 
   try {
-    await integrationQueue.addBulk(avatarJobs)
+    await lowQueue.addBulk(avatarJobs)
   } catch (error) {
     logger.error(
       { err: error, jobCount: avatarJobs.length, ...logContext },
diff --git a/apps/worker/src/integration/handlers/contact/update-avatar.ts b/apps/worker/src/integration/handlers/contact/update-avatar.ts
index 5310bacd77..8693a0755e 100644
--- a/apps/worker/src/integration/handlers/contact/update-avatar.ts
+++ b/apps/worker/src/integration/handlers/contact/update-avatar.ts
@@ -1,5 +1,5 @@
 import { contactInboxService, contactService } from "@chatbotx.io/business"
-import type { IntegrationJobUpdateContactAvatar } from "@chatbotx.io/worker-config"
+import type { LowJobUpdateContactAvatar } from "@chatbotx.io/worker-config"
 import { logger } from "../../../lib/logger"
 import {
   allIntegrations,
@@ -16,7 +16,7 @@ import {
  * overwrites a live-set or previously-mirrored avatar.
  */
 export const updateContactAvatar = async (
-  data: IntegrationJobUpdateContactAvatar["data"],
+  data: LowJobUpdateContactAvatar["data"],
 ): Promise<void> => {
   const { workspaceId, contactInboxId, sourceId } = data
 
diff --git a/apps/worker/src/low/worker.ts b/apps/worker/src/low/worker.ts
new file mode 100644
index 0000000000..218b42cc9a
--- /dev/null
+++ b/apps/worker/src/low/worker.ts
@@ -0,0 +1,104 @@
+import { withBlockedOwnerGuard } from "@chatbotx.io/business"
+import {
+  defaultWorkerOptions,
+  getRedisConnection,
+  LowJobAction,
+  type LowJobData,
+  queueNames,
+} from "@chatbotx.io/worker-config"
+import { type Job, Worker } from "bullmq"
+import { env } from "../env"
+import { coexistAttachmentDownload } from "../integration/handlers/coexist/attachment-download"
+import { updateContactAvatar } from "../integration/handlers/contact/update-avatar"
+import { ensureBootstrapped } from "../lib/bootstrap"
+import { logger } from "../lib/logger"
+import { runJobWithAuditContext } from "../lib/run-job-with-audit-context"
+
+/**
+ * Consumer for the `low` workload-class queue: light, high-volume, low-priority
+ * jobs deliberately kept off the latency-sensitive `integration` queue so a
+ * historical-import burst never starves customer replies.
+ *
+ * The queue split only isolates host resources (CPU, network, provider rate
+ * limits) when this runs as its OWN process: production launches it as a
+ * dedicated `worker low` service on the `worker-batch` node pool, separate from
+ * `worker integration` on the webhook nodes, at `LOW_WORKER_CONCURRENCY`. The
+ * image's `worker all` default (dev/fallback) runs every worker in one
+ * container and does NOT provide that host-level isolation.
+ *
+ * Handlers are shared with the integration worker during the two-phase cutover
+ * (integration still handles any jobs already queued under the old actions);
+ * once that queue is drained, the integration-side cases are removed.
+ */
+async function startLowWorker() {
+  try {
+    await ensureBootstrapped()
+  } catch (err) {
+    logger.error({ err }, "Failed to bootstrap low worker")
+    process.exit(1)
+  }
+
+  const worker = new Worker(
+    queueNames.enum.low,
+    async (job: Job<LowJobData>) => {
+      const workspaceId = job.data.data.workspaceId
+      await withBlockedOwnerGuard(workspaceId, async () => {
+        await runJobWithAuditContext(
+          { workspaceId, source: `low:${job.data.type}` },
+          async () => {
+            switch (job.data.type) {
+              case LowJobAction.coexistAttachmentDownload: {
+                await coexistAttachmentDownload(job.data.data)
+                return
+              }
+              case LowJobAction.updateContactAvatar: {
+                await updateContactAvatar(job.data.data)
+                return
+              }
+              default: {
+                // Exhaustiveness guard — a new LowJobData variant without a
+                // case here becomes a compile error.
+                const _exhaustive: never = job.data
+                logger.warn({ data: _exhaustive }, "Unhandled low job type")
+                return
+              }
+            }
+          },
+        )
+      })
+    },
+    {
+      connection: getRedisConnection(),
+      ...defaultWorkerOptions,
+      concurrency: env.LOW_WORKER_CONCURRENCY,
+    },
+  )
+
+  worker.on("failed", (job, err) => {
+    if (job) {
+      logger.error({ err }, `Low job ${job.id} has failed`)
+    }
+  })
+
+  let isShuttingDown = false
+  async function shutdown() {
+    if (isShuttingDown) {
+      return
+    }
+    isShuttingDown = true
+    try {
+      await worker.close()
+      process.exit(0)
+    } catch (err) {
+      logger.error(err, "[LowWorker] Error during shutdown")
+      process.exit(1)
+    }
+  }
+  process.once("SIGINT", shutdown)
+  process.once("SIGTERM", shutdown)
+}
+
+startLowWorker().catch((err) => {
+  logger.error({ err }, "Failed to start low worker")
+  process.exit(1)
+})
diff --git a/apps/worker/tsdown.config.ts b/apps/worker/tsdown.config.ts
index aa9f2eb8c4..6093d06c94 100644
--- a/apps/worker/tsdown.config.ts
+++ b/apps/worker/tsdown.config.ts
@@ -16,6 +16,7 @@ export default defineConfig({
     "src/schedule/worker.ts",
     "src/events/worker.ts",
     "src/notification/worker.ts",
+    "src/low/worker.ts",
   ],
   dts: false,
   shims: true,
diff --git a/packages/worker-config/src/index.ts b/packages/worker-config/src/index.ts
index 857749a794..5a301322bd 100644
--- a/packages/worker-config/src/index.ts
+++ b/packages/worker-config/src/index.ts
@@ -10,6 +10,7 @@ export * from "./queues/chat"
 export * from "./queues/default"
 export * from "./queues/heavy"
 export * from "./queues/integration"
+export * from "./queues/low"
 export * from "./queues/notification"
 export * from "./queues/quota"
 export * from "./queues/schedule"
diff --git a/packages/worker-config/src/lib/types.ts b/packages/worker-config/src/lib/types.ts
index 65b248db1e..fef05dc369 100644
--- a/packages/worker-config/src/lib/types.ts
+++ b/packages/worker-config/src/lib/types.ts
@@ -15,4 +15,5 @@ export const queueNames = z.enum([
   "notification",
   "callTranscription",
   "whatsappVoipSignaling",
+  "low",
 ])
diff --git a/packages/worker-config/src/queues/low/index.ts b/packages/worker-config/src/queues/low/index.ts
new file mode 100644
index 0000000000..4fba1d97b5
--- /dev/null
+++ b/packages/worker-config/src/queues/low/index.ts
@@ -0,0 +1,71 @@
+import { Queue } from "bullmq"
+import {
+  defaultJobOptions,
+  fakeQueue,
+  getRedisConnection,
+  isNoRedisEnv,
+} from "../../lib/connection"
+import { queueNames } from "../../lib/types"
+
+/**
+ * Workload-class queue for jobs that are individually *light* (short, I/O-bound)
+ * but arrive in *high volume* and are *low priority* — they must always yield to
+ * the latency-sensitive `integration` queue that drives customer replies.
+ *
+ * Like `heavy`, this queue is intentionally NOT tied to one product domain:
+ * enqueue work here whenever running it on a latency-sensitive domain worker
+ * would starve customer-facing throughput during a burst. Current tenants are
+ * the Coexist/Customer-Scan media backfill jobs (attachment mirroring, contact
+ * avatar mirroring); future light-but-bulky jobs can join with their own action.
+ */
+export const LowJobAction = {
+  coexistAttachmentDownload: "coexistAttachmentDownload",
+  updateContactAvatar: "updateContactAvatar",
+} as const
+
+export type LowJobAction = (typeof LowJobAction)[keyof typeof LowJobAction]
+
+/**
+ * Mirror a Coexist historical attachment's bytes to object storage and persist
+ * the resulting S3 path on the `Attachment` row. `originPath` carries a pending
+ * sentinel (Graph URL or `wa-media:<id>`) until this job resolves it.
+ *
+ * Idempotency: jobId `att-${attachmentId}` dedups concurrent enqueues; the
+ * handler additionally checks the originPath prefix to no-op on retries where a
+ * prior worker already finished the upload.
+ */
+export type LowJobCoexistAttachmentDownload = {
+  type: typeof LowJobAction.coexistAttachmentDownload
+  data: {
+    attachmentId: string
+    workspaceId: string
+    channel: "messenger" | "whatsapp" | "instagram"
+    integrationId: string
+  }
+}
+
+/**
+ * Fetch a contact's profile picture from the channel's Graph/API, mirror the
+ * bytes to our object storage, and persist the storage path on the Contact row.
+ * Dispatched per-contact after Coexist historical sync / Automatic Customer Scan
+ * upsert contacts (which only carry name/sourceId, not avatar).
+ */
+export type LowJobUpdateContactAvatar = {
+  type: typeof LowJobAction.updateContactAvatar
+  data: {
+    workspaceId: string
+    contactInboxId: string
+    sourceId: string
+  }
+}
+
+export type LowJobData =
+  | LowJobCoexistAttachmentDownload
+  | LowJobUpdateContactAvatar
+
+export const lowQueue = isNoRedisEnv()
+  ? fakeQueue
+  : new Queue<LowJobData>(queueNames.enum.low, {
+      connection: getRedisConnection(),
+      defaultJobOptions,
+    })