diff --git a/__tests__/core/queue.test.ts b/__tests__/core/queue.test.ts
new file mode 100644
index 0000000..10865a4
--- /dev/null
+++ b/__tests__/core/queue.test.ts
@@ -0,0 +1,788 @@
+import {
+  afterAll,
+  beforeAll,
+  beforeEach,
+  describe,
+  expect,
+  test,
+} from "bun:test";
+import {
+  type JobDefinition,
+  type ParsedJob,
+  Plugin,
+  Queue,
+} from "../../src/index.js";
+import specHelper from "../utils/specHelper.js";
+
+let queue: Queue;
+const delayedBase = () => Math.round((Date.now() + 60_000) / 1000) * 1000;
+
+async function seedActiveWorker(
+  target: Queue,
+  worker: string,
+  queueName: string,
+  args: unknown[],
+  runAt: Date,
+  includeId = true,
+): Promise<string> {
+  await target.enqueue(queueName, "slowJob", args);
+  const jobs = await target.connection.boss.fetch<ParsedJob>(queueName, {
+    batchSize: 1,
+  });
+  const job = jobs[0];
+  if (!job) throw new Error("expected an active job");
+  await target.connection.query(
+    `INSERT INTO ${specHelper.schema}.pgrq_workers (name, queues, working_on)
+     VALUES ($1, $2, $3::jsonb)`,
+    [
+      worker,
+      queueName,
+      JSON.stringify({
+        ...(includeId ? { id: job.id } : {}),
+        run_at: runAt.toString(),
+        queue: queueName,
+        worker,
+        payload: job.data,
+      }),
+    ],
+  );
+  return job.id;
+}
+
+describe("queue", () => {
+  beforeAll(async () => {
+    await specHelper.connect();
+    await specHelper.dropSchema();
+    await specHelper.migrate();
+  });
+
+  afterAll(async () => {
+    await queue?.end();
+    await specHelper.cleanup();
+    await specHelper.disconnect();
+  });
+
+  test("can connect", async () => {
+    const standalone = new Queue({
+      connection: specHelper.cleanConnectionDetails(),
+      queue: specHelper.queue,
+    });
+    await standalone.connect();
+    await standalone.end();
+  });
+
+  describe("[with connection]", () => {
+    beforeAll(async () => {
+      queue = new Queue(
+        { connection: specHelper.cleanConnectionDetails() },
+        {},
+      );
+      await queue.connect();
+    });
+
+    beforeEach(async () => {
+      await specHelper.cleanup();
+    });
+
+    test("can add a normal job", async () => {
+      expect(await queue.enqueue(specHelper.queue, "someJob", [1, 2, 3])).toBe(
+        true,
+      );
+      const raw = await specHelper.popFromQueue();
+      expect(raw).not.toBeNull();
+      const job = JSON.parse(raw ?? "{}") as ParsedJob;
+      expect(job.class).toBe("someJob");
+      expect(job.args).toEqual([1, 2, 3]);
+    });
+
+    test("can add delayed job (enqueueAt)", async () => {
+      const timestamp = delayedBase();
+      await queue.enqueueAt(timestamp, specHelper.queue, "someJob", [1, 2, 3]);
+      const { tasks } = await queue.delayedAt(timestamp);
+      expect(tasks).toHaveLength(1);
+      expect(tasks[0]?.class).toBe("someJob");
+      expect(tasks[0]?.args).toEqual([1, 2, 3]);
+    });
+
+    test("can add delayed job whose timestamp is a string (enqueueAt)", async () => {
+      const timestamp = delayedBase();
+      await queue.enqueueAt(
+        String(timestamp),
+        specHelper.queue,
+        "someJob",
+        [1, 2, 3],
+      );
+      expect((await queue.delayedAt(timestamp)).tasks).toHaveLength(1);
+    });
+
+    test("will not enqueue a delayed job at the same time with matching params with error", async () => {
+      const timestamp = delayedBase();
+      await queue.enqueueAt(timestamp, specHelper.queue, "someJob", [1, 2, 3]);
+      await expect(
+        queue.enqueueAt(timestamp, specHelper.queue, "someJob", [1, 2, 3]),
+      ).rejects.toThrow(
+        "Job already enqueued at this time with same arguments",
+      );
+      expect((await queue.delayedAt(timestamp)).tasks).toHaveLength(1);
+    });
+
+    test("concurrent Queue instances only enqueue one matching delayed job", async () => {
+      const other = new Queue({
+        connection: specHelper.cleanConnectionDetails(),
+      });
+      await other.connect();
+      const timestamp = delayedBase();
+      const settled = await Promise.allSettled([
+        queue.enqueueAt(timestamp, specHelper.queue, "someJob", [{ id: 1 }]),
+        other.enqueueAt(timestamp, specHelper.queue, "someJob", [{ id: 1 }]),
+      ]);
+      expect(
+        settled.filter((result) => result.status === "fulfilled"),
+      ).toHaveLength(1);
+      expect(
+        settled.filter((result) => result.status === "rejected"),
+      ).toHaveLength(1);
+      expect((await queue.delayedAt(timestamp)).tasks).toHaveLength(1);
+      await other.end();
+    });
+
+    test("can schedule a delayed job whose payload exceeds a btree key", async () => {
+      const timestamp = delayedBase();
+      const args = ["x".repeat(8_000)];
+      expect(
+        await queue.enqueueAt(timestamp, specHelper.queue, "someJob", args),
+      ).toBe(true);
+      await expect(
+        queue.enqueueAt(timestamp, specHelper.queue, "someJob", args),
+      ).rejects.toThrow(
+        "Job already enqueued at this time with same arguments",
+      );
+    });
+
+    test("will not enqueue a delayed job at the same time with matching params with error suppressed", async () => {
+      const timestamp = delayedBase();
+      await queue.enqueueAt(timestamp, specHelper.queue, "someJob", [1, 2, 3]);
+      expect(
+        await queue.enqueueAt(
+          timestamp,
+          specHelper.queue,
+          "someJob",
+          [1, 2, 3],
+          true,
+        ),
+      ).toBeUndefined();
+      expect((await queue.delayedAt(timestamp)).tasks).toHaveLength(1);
+    });
+
+    test("can add delayed job (enqueueIn)", async () => {
+      await queue.enqueueIn(60_000, specHelper.queue, "someJob", [1, 2, 3]);
+      expect(await queue.timestamps()).toHaveLength(1);
+    });
+
+    test("can add a delayed job whose time is a string (enqueueIn)", async () => {
+      await queue.enqueueIn("60000", specHelper.queue, "someJob", [1, 2, 3]);
+      expect(await queue.timestamps()).toHaveLength(1);
+    });
+
+    test("can get the number of jobs currently enqueued", async () => {
+      await queue.enqueue(specHelper.queue, "someJob", [1]);
+      await queue.enqueue(specHelper.queue, "someJob", [2]);
+      await queue.enqueueIn(60_000, specHelper.queue, "someJob", [3]);
+      expect(await queue.length(specHelper.queue)).toBe(2);
+    });
+
+    test("can get the jobs in the queue", async () => {
+      await queue.enqueue(specHelper.queue, "someJob", [1, 2, 3]);
+      await queue.enqueue(specHelper.queue, "someJob", [4, 5, 6]);
+      const jobs = await queue.queued(specHelper.queue, 0, -1);
+      expect(jobs.map((job) => job.args)).toEqual([
+        [1, 2, 3],
+        [4, 5, 6],
+      ]);
+      expect(await queue.queued(specHelper.queue, 1, 1)).toHaveLength(1);
+    });
+
+    test("can find previously scheduled jobs", async () => {
+      const timestamp = delayedBase();
+      await queue.enqueueAt(timestamp, specHelper.queue, "someJob", [1, 2, 3]);
+      expect(
+        await queue.scheduledAt(specHelper.queue, "someJob", [1, 2, 3]),
+      ).toEqual([timestamp / 1000]);
+    });
+
+    test("will not match previously scheduled jobs with differnt args", async () => {
+      await queue.enqueueAt(
+        delayedBase(),
+        specHelper.queue,
+        "someJob",
+        [1, 2, 3],
+      );
+      expect(
+        await queue.scheduledAt(specHelper.queue, "someJob", [3, 2, 1]),
+      ).toEqual([]);
+    });
+
+    test("can delete an enqueued job", async () => {
+      await queue.enqueue(specHelper.queue, "someJob", [1, 2, 3]);
+      expect(await queue.del(specHelper.queue, "someJob", [1, 2, 3])).toBe(1);
+      expect(await queue.length(specHelper.queue)).toBe(0);
+    });
+
+    test("del honors positive and negative count direction", async () => {
+      const ids = async () => {
+        const result = await queue.connection.query<{ id: string }>(
+          `SELECT id
+           FROM ${specHelper.schema}.job
+           WHERE name = $1 AND state = 'created'
+           ORDER BY created_on, id`,
+          [specHelper.queue],
+        );
+        return result.rows.map((row) => row.id);
+      };
+
+      for (let index = 0; index < 3; index += 1) {
+        await queue.enqueue(specHelper.queue, "sameJob", [1]);
+      }
+      const original = await ids();
+      expect(await queue.del(specHelper.queue, "sameJob", [1], 1)).toBe(1);
+      expect(await ids()).toEqual(original.slice(1));
+
+      await specHelper.cleanup();
+      for (let index = 0; index < 3; index += 1) {
+        await queue.enqueue(specHelper.queue, "sameJob", [1]);
+      }
+      const reloaded = await ids();
+      expect(await queue.del(specHelper.queue, "sameJob", [1], -1)).toBe(1);
+      expect(await ids()).toEqual(reloaded.slice(0, -1));
+    });
+
+    test("can delete all enqueued jobs of a particular function/class", async () => {
+      await queue.enqueue(specHelper.queue, "someJob1", [1]);
+      await queue.enqueue(specHelper.queue, "someJob1", [2]);
+      await queue.enqueue(specHelper.queue, "someJob2", [3]);
+      expect(await queue.delByFunction(specHelper.queue, "someJob1")).toBe(2);
+      expect(await queue.length(specHelper.queue)).toBe(1);
+    });
+
+    test("delByFunction only deletes matches inside its slice", async () => {
+      await queue.enqueue(specHelper.queue, "someJob1", [1]);
+      await queue.enqueue(specHelper.queue, "someJob2", [2]);
+      await queue.enqueue(specHelper.queue, "someJob1", [3]);
+      expect(
+        await queue.delByFunction(specHelper.queue, "someJob1", 1, 2),
+      ).toBe(1);
+      expect(
+        (await queue.queued(specHelper.queue, 0, -1)).map((job) => job.class),
+      ).toEqual(["someJob1", "someJob2"]);
+    });
+
+    test("can delete a delayed job", async () => {
+      const timestamp = delayedBase();
+      await queue.enqueueAt(timestamp, specHelper.queue, "someJob", [1, 2, 3]);
+      expect(
+        await queue.delDelayed(specHelper.queue, "someJob", [1, 2, 3]),
+      ).toEqual([timestamp / 1000]);
+    });
+
+    test("can delete a delayed job, and delayed queue should be empty", async () => {
+      await queue.enqueueAt(
+        delayedBase(),
+        specHelper.queue,
+        "someJob",
+        [1, 2, 3],
+      );
+      await queue.delDelayed(specHelper.queue, "someJob", [1, 2, 3]);
+      expect(await queue.allDelayed()).toEqual({});
+    });
+
+    test("can re-schedule after delQueue of object args", async () => {
+      const timestamp = delayedBase();
+      const args = [{ z: 1, a: 2 }];
+      await queue.enqueueAt(timestamp, "object-queue", "someJob", args);
+      expect(await queue.delQueue("object-queue")).toBe(1);
+      expect(
+        await queue.enqueueAt(timestamp, "object-queue", "someJob", args),
+      ).toBe(true);
+    });
+
+    test("can handle single arguments without explicit array", async () => {
+      await queue.enqueue(specHelper.queue, "someJob", 1);
+      const job = JSON.parse(
+        (await specHelper.popFromQueue()) ?? "{}",
+      ) as ParsedJob;
+      expect(job.args).toEqual([1]);
+    });
+
+    test("allows omitting arguments when enqueuing", async () => {
+      await queue.enqueue(specHelper.queue, "noParams");
+      expect((await queue.queued(specHelper.queue, 0, -1))[0]?.args).toEqual(
+        [],
+      );
+    });
+
+    test("allows omitting arguments when deleting", async () => {
+      await queue.enqueue(specHelper.queue, "noParams");
+      await queue.enqueue(specHelper.queue, "noParams");
+      expect(await queue.del(specHelper.queue, "noParams")).toBe(2);
+    });
+
+    test("allows omitting arguments when adding delayed job", async () => {
+      const timestamp = delayedBase();
+      await queue.enqueueAt(timestamp, specHelper.queue, "noParams");
+      await queue.enqueueAt(timestamp + 2000, specHelper.queue, "noParams");
+      expect(
+        await queue.scheduledAt(specHelper.queue, "noParams"),
+      ).toHaveLength(2);
+    });
+
+    test("allows omitting arguments when deleting a delayed job", async () => {
+      await queue.enqueueAt(delayedBase(), specHelper.queue, "noParams");
+      expect(await queue.delDelayed(specHelper.queue, "noParams")).toHaveLength(
+        1,
+      );
+      expect(await queue.allDelayed()).toEqual({});
+    });
+
+    test("can determine who the leader is", async () => {
+      expect(await queue.connection.tryLeader("the_scheduler", 60)).toBe(true);
+      expect(await queue.leader()).toBe("the_scheduler");
+      expect(queue.leaderKey()).not.toBe("");
+    });
+
+    test("can load stats", async () => {
+      await queue.connection.incrStat("failed", 1);
+      await queue.connection.incrStat("processed", 2);
+      expect(await queue.stats()).toEqual({ failed: "1", processed: "2" });
+    });
+
+    describe("locks", () => {
+      beforeEach(async () => {
+        await queue.connection.setLockNx(
+          "lock:lists:queueName:jobName:[{}]",
+          "123",
+          60,
+        );
+        await queue.connection.setLockNx(
+          "workerslock:lists:queueName:jobName:[{}]",
+          "456",
+          60,
+        );
+      });
+
+      test("can get locks", async () => {
+        expect(await queue.locks()).toEqual({
+          "lock:lists:queueName:jobName:[{}]": "123",
+          "workerslock:lists:queueName:jobName:[{}]": "456",
+        });
+      });
+
+      test("can remove locks", async () => {
+        expect(
+          await queue.delLock("workerslock:lists:queueName:jobName:[{}]"),
+        ).toBe(1);
+      });
+
+      test("does not return expired locks", async () => {
+        await queue.connection.query(
+          `UPDATE ${specHelper.schema}.pgrq_locks
+           SET expires_at = now() - interval '1 second'`,
+        );
+        expect(await queue.locks()).toEqual({});
+        const count = await queue.connection.query<{ count: string }>(
+          `SELECT count(*)::text AS count
+           FROM ${specHelper.schema}.pgrq_locks`,
+        );
+        expect(Number(count.rows[0]?.count)).toBe(0);
+      });
+    });
+
+    describe("failed job managment", () => {
+      beforeEach(async () => {
+        for (let id = 1; id <= 3; id += 1) {
+          await queue.enqueue("busted-queue", "busted_job", [id, 2, 3]);
+          await queue.connection.query(
+            `UPDATE ${specHelper.schema}.job
+             SET state = 'failed',
+                 completed_on = now() + make_interval(secs => $1),
+                 output = $2::jsonb
+             WHERE data->'args'->>0 = $3`,
+            [
+              id,
+              JSON.stringify({
+                worker: `busted-worker-${id}`,
+                queue: "busted-queue",
+                exception: "ERROR_NAME",
+                error: "I broke",
+                backtrace: [],
+              }),
+              String(id),
+            ],
+          );
+        }
+      });
+
+      test("can list how many failed jobs there are", async () => {
+        expect(await queue.failedCount()).toBe(3);
+      });
+
+      test("can get the body content for a collection of failed jobs", async () => {
+        const failed = await queue.failed(1, 2);
+        expect(failed).toHaveLength(2);
+        expect(failed[0]?.worker).toBe("busted-worker-2");
+        expect(failed[1]?.payload.args).toEqual([3, 2, 3]);
+      });
+
+      test("can remove a failed job by payload", async () => {
+        const [failed] = await queue.failed(1, 1);
+        if (!failed) throw new Error("expected a failed job");
+        expect(await queue.removeFailed(failed)).toBe(1);
+        expect(await queue.failedCount()).toBe(2);
+      });
+
+      test("can re-enqueue a specific job, removing it from the failed queue", async () => {
+        const failed = await queue.failed(0, -1);
+        const target = failed[2];
+        if (!target) throw new Error("expected a failed job");
+        expect(await queue.retryAndRemoveFailed(target)).toBe(true);
+        expect(await queue.failedCount()).toBe(2);
+        expect(await queue.length("busted-queue")).toBe(1);
+      });
+
+      test("will return an error when trying to retry a job not in the failed queue", async () => {
+        const [failed] = await queue.failed(2, 2);
+        if (!failed) throw new Error("expected a failed job");
+        failed.worker = "a-fake-worker";
+        await expect(queue.retryAndRemoveFailed(failed)).rejects.toThrow(
+          "This job is not in failed queue",
+        );
+        expect(await queue.failedCount()).toBe(3);
+      });
+    });
+
+    describe("delayed status", () => {
+      test("can list the timestamps that exist", async () => {
+        const timestamp = delayedBase();
+        await queue.enqueueAt(timestamp, specHelper.queue, "job1", [1]);
+        await queue.enqueueAt(timestamp, specHelper.queue, "job2", [1]);
+        await queue.enqueueAt(timestamp + 2000, specHelper.queue, "job3", [1]);
+        expect(await queue.timestamps()).toEqual([timestamp, timestamp + 2000]);
+      });
+
+      test("can list the jobs delayed at a timestamp", async () => {
+        const timestamp = delayedBase();
+        await queue.enqueueAt(timestamp, specHelper.queue, "job1", [1]);
+        await queue.enqueueAt(timestamp, specHelper.queue, "job2", [1]);
+        const delayed = await queue.delayedAt(timestamp);
+        expect(delayed.rTimestamp).toBe(timestamp / 1000);
+        expect(delayed.tasks.map((task) => task.class)).toEqual([
+          "job1",
+          "job2",
+        ]);
+      });
+
+      test("can also return a hash with all delayed tasks", async () => {
+        const timestamp = delayedBase();
+        await queue.enqueueAt(timestamp, specHelper.queue, "job1", [1]);
+        await queue.enqueueAt(timestamp + 2000, specHelper.queue, "job2", [1]);
+        expect(Object.keys(await queue.allDelayed())).toEqual([
+          String(timestamp),
+          String(timestamp + 2000),
+        ]);
+      });
+
+      test("does not list already-ready jobs as delayed", async () => {
+        const timestamp = Math.round((Date.now() - 5000) / 1000) * 1000;
+        await queue.enqueueAt(timestamp, specHelper.queue, "job1", [1]);
+        expect((await queue.delayedAt(timestamp)).tasks).toHaveLength(0);
+        expect(await queue.length(specHelper.queue)).toBe(1);
+      });
+    });
+
+    test("runs enqueue plugins in order and honors vetoes", async () => {
+      const calls: string[] = [];
+      class TrackingPlugin extends Plugin {
+        override beforeEnqueue(): boolean {
+          calls.push("before");
+          return this.options.veto !== true;
+        }
+
+        override afterEnqueue(): boolean {
+          calls.push("after");
+          return true;
+        }
+      }
+      const jobs: Record<string, JobDefinition> = {
+        allowed: {
+          plugins: [TrackingPlugin],
+          perform: async () => undefined,
+        },
+        vetoed: {
+          plugins: [TrackingPlugin],
+          pluginOptions: { TrackingPlugin: { veto: true } },
+          perform: async () => undefined,
+        },
+      };
+      const pluginQueue = new Queue(
+        { connection: specHelper.cleanConnectionDetails() },
+        jobs,
+      );
+      await pluginQueue.connect();
+      expect(await pluginQueue.enqueue(specHelper.queue, "allowed")).toBe(true);
+      expect(await pluginQueue.enqueue(specHelper.queue, "vetoed")).toBe(false);
+      expect(calls).toEqual(["before", "after", "before"]);
+      await pluginQueue.end();
+    });
+
+    test("lists configured queues and deletes a queue", async () => {
+      await queue.enqueue("temporary", "job");
+      expect(await queue.queues()).toContain("temporary");
+      expect(await queue.delQueue("temporary")).toBe(1);
+      expect(await queue.queues()).not.toContain("temporary");
+    });
+
+    test("can re-enqueue after another instance deletes the queue", async () => {
+      const other = new Queue({
+        connection: specHelper.cleanConnectionDetails(),
+      });
+      await other.connect();
+      await queue.enqueue("recycle", "job", [1]);
+      expect(await other.delQueue("recycle")).toBe(1);
+      expect(await queue.enqueue("recycle", "job", [2])).toBe(true);
+      expect(await queue.length("recycle")).toBe(1);
+      await other.end();
+    });
+
+    test("queue row locking serializes enqueue with queue deletion", async () => {
+      const other = new Queue({
+        connection: specHelper.cleanConnectionDetails(),
+      });
+      await other.connect();
+      await queue.enqueue("serialized", "job", [1]);
+      const client = await queue.connection.pool.connect();
+      await client.query("BEGIN");
+      await client.query(
+        `SELECT name
+         FROM ${specHelper.schema}.queue
+         WHERE name = 'serialized'
+         FOR UPDATE`,
+      );
+
+      let settled = false;
+      const enqueue = other.enqueue("serialized", "job", [2]).then((value) => {
+        settled = true;
+        return value;
+      });
+      await Bun.sleep(25);
+      expect(settled).toBe(false);
+      await client.query("COMMIT");
+      client.release();
+
+      expect(await enqueue).toBe(true);
+      expect(await queue.length("serialized")).toBe(2);
+      await other.end();
+    });
+
+    test("does not drop jobs that remain after delQueue", async () => {
+      await queue.enqueue("busy", "job", [1]);
+      await queue.connection.query(
+        `UPDATE ${specHelper.schema}.job
+         SET state = 'active'
+         WHERE name = 'busy'`,
+      );
+      expect(await queue.delQueue("busy")).toBe(0);
+      expect(await queue.queues()).toContain("busy");
+      expect(await queue.enqueue("busy", "job", [2])).toBe(true);
+      expect(await queue.length("busy")).toBe(1);
+    });
+
+    test("forceCleanWorker fails the original active job", async () => {
+      expect(await queue.enqueue("stuck", "slowJob", [{ a: 1 }])).toBe(true);
+      const fetched = await queue.connection.boss.fetch<{
+        class: string;
+        queue: string;
+        args: unknown[];
+      }>("stuck", { batchSize: 1 });
+      const job = fetched[0];
+      if (!job) throw new Error("expected an active job");
+      await queue.connection.query(
+        `INSERT INTO ${specHelper.schema}.pgrq_workers (name, queues, working_on)
+         VALUES ('workerA', $1, $2::jsonb)`,
+        [
+          "stuck",
+          JSON.stringify({
+            id: job.id,
+            run_at: new Date().toString(),
+            queue: "stuck",
+            worker: "workerA",
+            payload: job.data,
+          }),
+        ],
+      );
+
+      const errorPayload = await queue.forceCleanWorker("workerA");
+      expect(errorPayload?.exception).toBe("Worker Timeout (killed manually)");
+      expect(await queue.failedCount()).toBe(1);
+      const failed = await queue.failed(0, -1);
+      expect(failed[0]?.id).toBe(job.id);
+      expect(failed[0]?.payload.args).toEqual([{ a: 1 }]);
+
+      const active = await queue.connection.query<{ count: string }>(
+        `SELECT count(*)::text AS count
+         FROM ${specHelper.schema}.job
+         WHERE name = 'stuck' AND state = 'active'`,
+      );
+      expect(Number(active.rows[0]?.count)).toBe(0);
+    });
+
+    test("forceCleanWorker without a job id fails only one matching active job", async () => {
+      await queue.enqueue("twins", "slowJob", [1]);
+      await queue.enqueue("twins", "slowJob", [1]);
+      const fetched = await queue.connection.boss.fetch<{
+        class: string;
+        queue: string;
+        args: unknown[];
+      }>("twins", { batchSize: 2 });
+      expect(fetched).toHaveLength(2);
+      await queue.connection.query(
+        `INSERT INTO ${specHelper.schema}.pgrq_workers (name, queues, working_on)
+         VALUES ('workerA', 'twins', $1::jsonb)`,
+        [
+          JSON.stringify({
+            run_at: new Date().toString(),
+            queue: "twins",
+            worker: "workerA",
+            payload: fetched[0]?.data,
+          }),
+        ],
+      );
+
+      await queue.forceCleanWorker("workerA");
+      expect(await queue.failedCount()).toBe(1);
+      const active = await queue.connection.query<{ count: string }>(
+        `SELECT count(*)::text AS count
+         FROM ${specHelper.schema}.job
+         WHERE name = 'twins' AND state = 'active'`,
+      );
+      expect(Number(active.rows[0]?.count)).toBe(1);
+    });
+
+    test("can list running workers", async () => {
+      await queue.connection.query(
+        `INSERT INTO ${specHelper.schema}.pgrq_workers (name, queues)
+         VALUES ('workerA', $1), ('workerB', $1)`,
+        [specHelper.queue],
+      );
+      expect(await queue.workers()).toEqual({
+        workerA: specHelper.queue,
+        workerB: specHelper.queue,
+      });
+    });
+
+    test("we can see what workers are working on (idle)", async () => {
+      await queue.connection.query(
+        `INSERT INTO ${specHelper.schema}.pgrq_workers (name, queues)
+         VALUES ('workerA', $1), ('workerB', $1)`,
+        [specHelper.queue],
+      );
+      expect(await queue.allWorkingOn()).toEqual({
+        workerA: "started",
+        workerB: "started",
+      });
+      expect(await queue.workingOn("workerA", specHelper.queue)).toBeNull();
+    });
+
+    test("we can see what workers are working on (active)", async () => {
+      const runAt = new Date();
+      await seedActiveWorker(
+        queue,
+        "workerA",
+        "active-status",
+        [{ a: 1 }],
+        runAt,
+      );
+      const working = await queue.allWorkingOn();
+      expect(working.workerA).not.toBe("started");
+      if (working.workerA === "started" || !working.workerA) {
+        throw new Error("expected active worker payload");
+      }
+      expect(working.workerA.payload.args).toEqual([{ a: 1 }]);
+      expect(Date.parse(working.workerA.run_at)).toBe(
+        Math.floor(runAt.getTime() / 1000) * 1000,
+      );
+    });
+
+    test("can remove stuck workers and re-enqueue their jobs", async () => {
+      await seedActiveWorker(
+        queue,
+        "workerA",
+        "stuck-clean",
+        [{ a: 1 }],
+        new Date(Date.now() - 10_000),
+      );
+      const cleaned = await queue.cleanOldWorkers(1_000);
+      expect(cleaned.workerA?.payload.args).toEqual([{ a: 1 }]);
+      expect(await queue.failedCount()).toBe(1);
+      await queue.retryStuckJobs();
+      expect(await queue.failedCount()).toBe(0);
+      expect(await queue.length("stuck-clean")).toBe(1);
+    });
+
+    test("will not remove stuck jobs within the time limit", async () => {
+      await seedActiveWorker(queue, "workerA", "recent-worker", [], new Date());
+      expect(await queue.cleanOldWorkers(60_000)).toEqual({});
+      expect(await queue.workers()).toEqual({ workerA: "recent-worker" });
+    });
+
+    test("can forceClean a worker, returning the error payload", async () => {
+      await seedActiveWorker(
+        queue,
+        "workerA",
+        "force-clean",
+        [{ a: 1 }],
+        new Date(),
+      );
+      const failure = await queue.forceCleanWorker("workerA");
+      expect(failure?.worker).toBe("workerA");
+      expect(failure?.queue).toBe("force-clean");
+      expect(failure?.payload.args).toEqual([{ a: 1 }]);
+      expect(failure?.backtrace[1]).toBe("queue#forceCleanWorker");
+    });
+
+    test("can forceClean a worker, returning the error payload and removing all keys it had set in redis", async () => {
+      const id = await seedActiveWorker(
+        queue,
+        "workerA",
+        "force-clean-keys",
+        [],
+        new Date(),
+      );
+      await queue.forceCleanWorker("workerA");
+      expect(await queue.workers()).toEqual({});
+      expect(await queue.workingOn("workerA", "force-clean-keys")).toBeNull();
+      expect((await queue.failed(0, -1))[0]?.id).toBe(id);
+    });
+
+    test("forceCleanWorker emits an error for an unknown worker", async () => {
+      const error = new Promise<Error>((resolve) => {
+        queue.once("error", resolve);
+      });
+      expect(await queue.forceCleanWorker("missing-worker")).toBeUndefined();
+      expect((await error).message).toContain("cannot find queues");
+    });
+
+    test("retryStuckJobs", async () => {
+      await seedActiveWorker(queue, "workerA", "retry-one", [1], new Date());
+      await queue.forceCleanWorker("workerA");
+      await seedActiveWorker(queue, "workerB", "retry-two", [2], new Date());
+      await queue.forceCleanWorker("workerB");
+
+      await queue.retryStuckJobs(1);
+      expect(await queue.failedCount()).toBe(1);
+      expect(
+        (await queue.length("retry-one")) + (await queue.length("retry-two")),
+      ).toBe(1);
+      await queue.retryStuckJobs(0);
+      expect(await queue.failedCount()).toBe(1);
+    });
+  });
+});
diff --git a/__tests__/utils/specHelper.ts b/__tests__/utils/specHelper.ts
index 3ace860..25d9875 100644
--- a/__tests__/utils/specHelper.ts
+++ b/__tests__/utils/specHelper.ts
@@ -120,10 +120,26 @@ export async function dropSchema(): Promise<void> {
 }
 
 /**
- * @throws Always — dequeue lands in Phase 3.
+ * Fetch and remove one ready job from the default test queue.
+ *
+ * @returns The node-resque encoded payload, or `null` when the queue is empty.
  */
-export async function popFromQueue(): Promise<never> {
-  throw new Error("not implemented");
+export async function popFromQueue(): Promise<string | null> {
+  const connection = new Connection(cleanConnectionDetails());
+  await connection.connect();
+  try {
+    const jobs = await connection.boss.fetch<{
+      class: string;
+      queue: string;
+      args: unknown[];
+    }>(queue, { batchSize: 1 });
+    const job = jobs[0];
+    if (!job) return null;
+    await connection.boss.deleteJob(queue, job.id);
+    return JSON.stringify(job.data);
+  } finally {
+    await connection.end();
+  }
 }
 
 const specHelper = {
diff --git a/docs/plans/02-connection-and-schema.md b/docs/plans/02-connection-and-schema.md
index 429bf57..acf99c7 100644
--- a/docs/plans/02-connection-and-schema.md
+++ b/docs/plans/02-connection-and-schema.md
@@ -36,6 +36,7 @@ export interface ConnectionOptions {
 
 export interface QueueOptions {
   connection?: ConnectionOptions;
+  queue?: string | string[]; // node-resque constructor compatibility
 }
 
 export interface WorkerOptions extends QueueOptions {
@@ -231,3 +232,4 @@ Do not defer these to Phase 8.
 - 2026-08-26: pg-boss is a named ESM export (`import { PgBoss } from "pg-boss"`), not a default export. Migrator instances use `migrate: true` + `supervise: false` + `schedule: false`.
 - 2026-08-26: Version bumped `0.0.1` → `0.1.0` (first user-facing API: `Connection`).
 - 2026-08-26: Node ESM (`"type": "module"`) requires relative import specifiers with `.js` extensions in emitted `dist/` (e.g. `from "./core/connection.js"`). Without them, `node scripts/assert-node-package.mjs` fails with `ERR_MODULE_NOT_FOUND` even though `tsc` and Bun tests pass.
+- 2026-08-29: Phase 3 restored node-resque's optional `QueueOptions.queue` field. Queue methods still take an explicit queue name, but retaining the constructor field lets existing typed call sites migrate without an excess-property error.
diff --git a/docs/plans/03-queue.md b/docs/plans/03-queue.md
index 6ca8fab..1527af3 100644
--- a/docs/plans/03-queue.md
+++ b/docs/plans/03-queue.md
@@ -1,6 +1,6 @@
 # Phase 3 — Queue
 
-**Status:** not-started  
+**Status:** done  
 **Depends on:** Phase 2
 
 ## Goal
@@ -116,12 +116,12 @@ Skip only tests that poke Redis keys directly if any remain inside queue.ts (the
 ## Acceptance criteria
 
 - All Queue methods exist with JSDoc copied/adapted from node-resque
-- `__tests__/core/queue.ts` port is green on CI (worker-status tests that start a Worker wait for Phase 4 — split those into a `describe` marked pending **or** implement after Phase 4; prefer implementing worker methods against empty tables so idle tests pass, and mark `active workingOn` pending)
+- `__tests__/core/queue.test.ts` is green. Worker-status methods are tested with seeded `pgrq_workers` and active pg-boss jobs; Phase 4 will additionally exercise them through a live Worker.
 
 Recommended split:
 
-- Phase 3: enqueue, delayed, delete, failed (inject failed rows via SQL/`fail`), locks, stats, leader (null), idle workers
-- Phase 4: active `workingOn`, `forceCleanWorker` with a live worker
+- Phase 3: enqueue, delayed, delete, failed (inject failed rows via SQL/`fail`), locks, stats, leader, and worker-table behavior
+- Phase 4: repeat active `workingOn` / cleanup behavior end-to-end through a live Worker
 
 ## Next phase needs
 
@@ -130,3 +130,16 @@ Recommended split:
 ## Lessons learned
 
 - 2026-08-26: Bun requires `*.test.ts` filenames for discovery; Phase 3 ports should use `__tests__/core/queue.test.ts` (not bare `queue.ts`) while keeping node-resque describe/test titles.
+- 2026-08-29: pg-boss v12 queues are explicit configuration rows/partitions, so Queue lazily calls `getQueue`/`createQueue` before `send`. Queue defaults are `retryLimit: 0` and `deleteAfterSeconds: 0` (pg-boss defines `0` as never auto-delete); scheduler retention remains authoritative.
+- 2026-08-29: Delayed duplicate identity is reserved in `pgrq_locks` under a private `timestamps:{payload}:delayed:{second}` key until the scheduled second passes. This makes concurrent duplicate checks atomic without modifying pg-boss's partitioned `job` schema; `locks()` intentionally exposes only plugin `lock:*` and `workerslock:*` rows.
+- 2026-08-29: Upstream queue tests schedule at Unix millisecond `10000`, but pg-boss correctly treats 1970 timestamps as immediately runnable. The PostgreSQL port uses future rounded timestamps while retaining the same test titles and timestamp-unit assertions.
+- 2026-08-29: Phase 3 initially completed with 36 Queue tests passing against PostgreSQL. Follow-up review added direct metadata coverage for all Queue worker-status methods; Phase 4 retains responsibility for end-to-end live-Worker coverage.
+- 2026-08-29: `pgrq_stats` stores numeric counters, but `Queue.stats()` stringifies them to retain node-resque's Redis `MGET` response shape (`{ processed: "2", failed: "1" }`).
+- 2026-08-29: Bugbot: an in-process `knownQueues` cache survived `delQueue` on another `Queue` instance, so later `send` skipped `createQueue`. `ensureQueue` now always checks pg-boss and retries once on `Queue does not exist`.
+- 2026-08-29: Bugbot: `delQueue` only skipped `delete_queue` when `active` rows remained, so a concurrent `created` insert could be dropped. It now locks the pg-boss queue row, deletes non-active jobs, and drops the queue only when no rows remain.
+- 2026-08-29: Bugbot: `forceCleanWorker` inserted a second `failed` row and left the original job `active`. It now updates the in-flight job to `failed` (by id when recorded, otherwise by matching `data`) and only inserts if no active row exists.
+- 2026-08-29: Bugbot: the data-only fallback could fail every identical `active` payload. The update now selects a single matching row (`LIMIT 1 … FOR UPDATE`).
+- 2026-08-29: Bugbot: `delayedAt` omitted `start_after > now()`, so a timestamp whose second had arrived still listed jobs that `length`/`queued` already treated as ready. It now uses the same delayed filter as `timestamps` / `scheduledAt` / `delDelayed`.
+- 2026-08-29: Bugbot: delayed duplicate lock keys embedded the encoded JSON and overflowed the `pgrq_locks` btree for large payloads. Keys now use `sha256(encoded)` plus the timestamp second.
+- 2026-08-29: Bugbot: `delQueue` rebuilt those keys from jsonb-loaded args, whose object key order can differ from `JSON.stringify` at enqueue time. The hash now canonicalizes nested object keys so delete and re-enqueue agree.
+- 2026-08-29: Coverage audit found untested `del(count)`, `delByFunction(start, stop)`, expired-lock cleanup, concurrent delayed enqueue, queue-row serialization, `cleanOldWorkers`, `retryStuckJobs`, active `workingOn`, and unknown-worker errors. These now have focused PostgreSQL tests rather than being deferred wholesale to Phase 4.
diff --git a/docs/plans/08-conformance-tests.md b/docs/plans/08-conformance-tests.md
index d21e0cf..1e6885e 100644
--- a/docs/plans/08-conformance-tests.md
+++ b/docs/plans/08-conformance-tests.md
@@ -17,8 +17,8 @@ Source of truth: [actionhero/node-resque](https://github.com/actionhero/node-res
 | --- | --- |
 | Phase 1 | `specHelper` skeleton, smoke `SELECT 1`, `test.yaml` (lint / build / Postgres / complete) |
 | Phase 2 | `__tests__/core/connection.test.ts` + `connectionError.test.ts` (+ illegal schema, BYO pool, migrate, locks, leader) |
-| Phase 3 | `__tests__/core/queue.test.ts` (minus live-worker slices deferred to 4) |
-| Phase 4 | `__tests__/core/worker.test.ts`, remaining queue worker-status, multi-process + priority extras |
+| Phase 3 | `__tests__/core/queue.test.ts` (worker-table behavior seeded directly) |
+| Phase 4 | `__tests__/core/worker.test.ts`, live-Worker queue-status integration, multi-process + priority extras |
 | Phase 5 | `__tests__/core/scheduler.test.ts`, automigrate + sweeper extras |
 | Phase 6 | `__tests__/plugins/*.test.ts` |
 | Phase 7 | `__tests__/core/multiWorker.test.ts` |
@@ -151,6 +151,7 @@ If an assertion cannot be identical, add a row (may already have rows from earli
 | --- | --- | --- | --- |
 | keys built with a custom namespace | `connection.key("thing") === "customNamespace:thing"` | `connection.schema === customSchema` and `pgrq_locks` exists in that schema | Keys are not Redis-prefixed; schema replaces namespace |
 | removes the redis event listeners when end | `redis.listenerCount("error"|"end")` | `pool`/`boss` `listenerCount("error")` with BYO pool | No Redis `end` event; we forward `error` only |
+| queue delayed-job tests using timestamp `10000` | Redis keeps the 1970 timestamp in a delayed list until Scheduler transfers it | Use a future rounded timestamp and assert the same seconds/ms conversions | pg-boss `startAfter` is eligibility time, so a past timestamp is immediately ready by design |
 
 PRs that add rows must explain. "Postgres is different" is not enough if the Queue API can still match.
 
@@ -176,3 +177,5 @@ Docs site can describe a real API. Phase 10 can trust tests that have been runni
 - 2026-08-26: Phase 1 corrected the runner to `node:test` on a Bun + Node matrix. Isolation uses `--max-concurrency=1` / `--test-concurrency=1`, not `bun test --concurrency=1`.
 - 2026-08-26: Phase 1 reverted the suite to `bun:test`. Node is covered by importing `dist/` (`test:node-package`), not by running this matrix on `node --test`.
 - 2026-08-26: Phase 2 — Bun requires `.test.ts` (or `.spec` / `_test_` / `_spec_`) in the filename. Matrix paths are `__tests__/core/<name>.test.ts` while describe/test titles stay node-resque-identical. Later phases must not copy bare `connection.ts`-style names or CI will skip them.
+- 2026-08-29: Phase 3 preserves upstream Queue test titles but replaces hard-coded 1970 delayed timestamps with future rounded values. This is a required semantic adaptation because pg-boss uses `startAfter` directly rather than waiting for a scheduler to move a Redis-list item.
+- 2026-08-29: Phase 3's Queue suite now covers active/old worker metadata, force-clean, and retry-stuck behavior through seeded pg-boss and `pgrq_workers` rows. Phase 4 still repeats these paths with a live Worker but no Queue titles remain skipped.
diff --git a/package.json b/package.json
index 513adab..54864a3 100644
--- a/package.json
+++ b/package.json
@@ -1,6 +1,6 @@
 {
   "name": "pgboss-queue",
-  "version": "0.1.0",
+  "version": "0.2.0",
   "description": "A PostgreSQL-backed background job queue with the node-resque runtime model",
   "type": "module",
   "main": "./dist/index.js",
diff --git a/src/core/connection.ts b/src/core/connection.ts
index 7ef7ff3..6296291 100644
--- a/src/core/connection.ts
+++ b/src/core/connection.ts
@@ -59,6 +59,8 @@ export interface ConnectionOptions {
  */
 export interface QueueOptions {
   connection?: ConnectionOptions;
+  /** Optional default queue retained for node-resque constructor compatibility. */
+  queue?: string | string[];
 }
 
 /**
diff --git a/src/core/plugin.ts b/src/core/plugin.ts
new file mode 100644
index 0000000..0c0e3de
--- /dev/null
+++ b/src/core/plugin.ts
@@ -0,0 +1,69 @@
+import type { JobDefinition, PluginHost } from "../types/job.js";
+import type { Queue } from "./queue.js";
+
+/**
+ * Base class for node-resque-compatible job plugins.
+ *
+ * Subclasses may implement any enqueue/perform hook. Returning `false` from a
+ * before hook prevents the operation.
+ */
+export abstract class Plugin {
+  /** Plugin name used to look up `pluginOptions`. */
+  readonly name: string;
+  /** Queue or Worker instance invoking this plugin. */
+  readonly worker: PluginHost;
+  /** Queue instance associated with the host, when available. */
+  readonly queueObject: Queue | undefined;
+  /** Queue name for this operation. */
+  readonly queue: string;
+  /** Registered job name. */
+  readonly func: string;
+  /** Registered job definition. */
+  readonly job: JobDefinition;
+  /** Arguments encoded for the job. */
+  readonly args: unknown[];
+  /** Options selected by plugin name. */
+  readonly options: Record<string, unknown>;
+
+  /**
+   * @param worker - Queue or Worker invoking the hook.
+   * @param func - Registered job name.
+   * @param queue - Queue name for this operation.
+   * @param job - Registered job definition.
+   * @param args - Job arguments.
+   * @param options - Plugin-specific options.
+   */
+  constructor(
+    worker: PluginHost,
+    func: string,
+    queue: string,
+    job: JobDefinition,
+    args: unknown[],
+    options: Record<string, unknown>,
+  ) {
+    this.name = this.constructor.name || "Node Resque Plugin";
+    this.worker = worker;
+    this.func = func;
+    this.queue = queue;
+    this.job = job;
+    this.args = args;
+    this.options = options;
+
+    const host = worker as PluginHost & { queueObject?: Queue };
+    this.queueObject =
+      host.queueObject ?? (isQueue(worker) ? worker : undefined);
+  }
+
+  /** Run before enqueue. Return `false` to suppress enqueueing. */
+  beforeEnqueue?(): boolean | Promise<boolean>;
+  /** Run after enqueue. */
+  afterEnqueue?(): boolean | Promise<boolean>;
+  /** Run before performing. Return `false` to suppress execution. */
+  beforePerform?(): boolean | Promise<boolean>;
+  /** Run after performing. */
+  afterPerform?(): boolean | Promise<boolean>;
+}
+
+function isQueue(host: PluginHost): host is Queue {
+  return "enqueue" in host;
+}
diff --git a/src/core/pluginRunner.ts b/src/core/pluginRunner.ts
new file mode 100644
index 0000000..e63cdaf
--- /dev/null
+++ b/src/core/pluginRunner.ts
@@ -0,0 +1,122 @@
+import {
+  type JobDefinition,
+  type Jobs,
+  jobDefinition,
+  type PluginConstructor,
+  type PluginHost,
+} from "../types/job.js";
+import type { Plugin } from "./plugin.js";
+
+/** Hook names understood by the plugin runner. */
+export type PluginHook =
+  | "beforeEnqueue"
+  | "afterEnqueue"
+  | "beforePerform"
+  | "afterPerform";
+
+/**
+ * Run all plugins for one job in declaration order.
+ *
+ * @param self - Queue or Worker invoking the plugins.
+ * @param type - Hook to invoke.
+ * @param func - Registered job name.
+ * @param queue - Queue name.
+ * @param job - Job declaration, including function-form jobs.
+ * @param args - Job arguments.
+ * @returns `false` when a plugin vetoes the operation, otherwise `true`.
+ */
+export async function runPlugins(
+  self: PluginHost,
+  type: PluginHook,
+  func: string,
+  queue: string,
+  job: Jobs[string] | undefined,
+  args: unknown[],
+): Promise<boolean> {
+  const definition = jobDefinition(job);
+  if (!definition?.plugins?.length) return true;
+
+  for (const reference of definition.plugins) {
+    const result = await runPlugin(
+      self,
+      reference,
+      type,
+      func,
+      queue,
+      definition,
+      args,
+    );
+    if (result === false) return false;
+  }
+
+  return true;
+}
+
+/**
+ * Construct and run one plugin hook.
+ *
+ * @param self - Queue or Worker invoking the plugin.
+ * @param reference - Plugin constructor or built-in plugin name.
+ * @param type - Hook to invoke.
+ * @param func - Registered job name.
+ * @param queue - Queue name.
+ * @param job - Normalized job definition.
+ * @param args - Job arguments.
+ * @returns Hook result, defaulting to `true` when the hook is absent.
+ * @throws If a named plugin module does not export the requested plugin.
+ */
+export async function runPlugin(
+  self: PluginHost,
+  reference: string | PluginConstructor,
+  type: PluginHook,
+  func: string,
+  queue: string,
+  job: JobDefinition,
+  args: unknown[],
+): Promise<boolean> {
+  const Constructor =
+    typeof reference === "string"
+      ? await loadNamedPlugin(reference)
+      : reference;
+
+  const name = Constructor.name || "Node Resque Plugin";
+  const options = job.pluginOptions?.[name] ?? {};
+  const plugin = new Constructor(self, func, queue, job, args, options);
+  const hook = plugin[type];
+  if (typeof hook !== "function") return true;
+
+  return (await hook.call(plugin)) !== false;
+}
+
+async function loadNamedPlugin(name: string): Promise<PluginConstructor> {
+  const module: unknown = await import(`../plugins/${name}.js`);
+  if (!isModuleRecord(module)) {
+    throw new Error(`Plugin module "${name}" is invalid`);
+  }
+
+  const Constructor = module[name];
+  if (typeof Constructor !== "function") {
+    throw new Error(`Plugin "${name}" is not exported`);
+  }
+
+  return Constructor as PluginConstructor;
+}
+
+function isModuleRecord(value: unknown): value is Record<string, unknown> {
+  return typeof value === "object" && value !== null;
+}
+
+/** @deprecated Use {@link runPlugins}. Kept for node-resque source compatibility. */
+export const RunPlugins = runPlugins;
+
+/** @deprecated Use {@link runPlugin}. Kept for node-resque source compatibility. */
+export const RunPlugin = runPlugin;
+
+// Assert the indexed hook surface remains compatible with Plugin.
+const _pluginTypeCheck: PluginHook[] = [
+  "beforeEnqueue",
+  "afterEnqueue",
+  "beforePerform",
+  "afterPerform",
+];
+void (_pluginTypeCheck satisfies Array<keyof Plugin>);
diff --git a/src/core/queue.ts b/src/core/queue.ts
new file mode 100644
index 0000000..fa7ef71
--- /dev/null
+++ b/src/core/queue.ts
@@ -0,0 +1,1055 @@
+import { createHash } from "node:crypto";
+import { EventEmitter } from "node:events";
+import { hostname } from "node:os";
+import type { QueryResultRow } from "pg";
+import type { ErrorPayload } from "../types/errorPayload.js";
+import type { Jobs } from "../types/job.js";
+import { Connection, type QueueOptions } from "./connection.js";
+import { runPlugins } from "./pluginRunner.js";
+
+const QUEUED_STATES = ["created", "retry"] as const;
+const DUPLICATE_ERROR = "Job already enqueued at this time with same arguments";
+
+/** Payload stored in pg-boss's `job.data` column. */
+export interface ParsedJob {
+  /** Registered job name. */
+  class: string;
+  /** Queue name. */
+  queue: string;
+  /** Positional job arguments. */
+  args: unknown[];
+  /** Optional per-plugin configuration copied by callers when needed. */
+  pluginOptions?: Record<string, Record<string, unknown>>;
+}
+
+/** Description of work currently assigned to a worker. */
+export interface ParsedWorkerPayload {
+  /** Date string for when work started. */
+  run_at: string;
+  /** Queue being worked. */
+  queue: string;
+  /** Worker name. */
+  worker: string;
+  /** Encoded job payload. */
+  payload: ParsedJob;
+  /** pg-boss job id recorded by the worker, when present. */
+  id?: string;
+}
+
+/** Failed-job representation returned by Queue inspection methods. */
+export interface ParsedFailedJobPayload extends ErrorPayload {
+  /** pg-boss job id used for precise removal and retry. */
+  id?: string;
+}
+
+interface JobRow extends QueryResultRow {
+  id: string;
+  name: string;
+  data: unknown;
+  output?: unknown;
+  created_on: Date;
+  start_after: Date;
+  completed_on?: Date | null;
+}
+
+interface WorkerRow extends QueryResultRow {
+  name: string;
+  queues: string;
+  working_on: unknown;
+}
+
+/**
+ * PostgreSQL-backed node-resque Queue API.
+ *
+ * Queue methods use pg-boss for insertion and its `job` table for compatible
+ * inspection and administration.
+ */
+export class Queue extends EventEmitter {
+  /** Resolved Queue options. */
+  readonly options: QueueOptions;
+  /** Named jobs used by enqueue plugins and, later, Workers. */
+  readonly jobs: Jobs;
+  /** Underlying PostgreSQL / pg-boss connection. */
+  readonly connection: Connection;
+
+  /**
+   * @param options - Queue options containing PostgreSQL connection settings.
+   * @param jobs - Named job implementations. Functions are accepted as shorthand.
+   */
+  constructor(options: QueueOptions = {}, jobs: Jobs = {}) {
+    super();
+    this.options = options;
+    this.jobs = jobs;
+    this.connection = new Connection(options.connection);
+    this.connection.on("error", (error: Error) => this.emit("error", error));
+  }
+
+  /** Connect the underlying PostgreSQL and pg-boss clients. */
+  async connect(): Promise<void> {
+    await this.connection.connect();
+  }
+
+  /** Stop the underlying clients and owned pool. */
+  async end(): Promise<void> {
+    await this.connection.end();
+  }
+
+  /**
+   * Encode a node-resque job payload.
+   *
+   * @param q - Queue name.
+   * @param func - Registered job name.
+   * @param args - Positional arguments.
+   * @returns Stable JSON payload.
+   */
+  encode(q: string, func: string, args: unknown[] = []): string {
+    return JSON.stringify({ class: func, queue: q, args });
+  }
+
+  /**
+   * Enqueue a named job for immediate processing.
+   *
+   * @param q - Queue name.
+   * @param func - Registered job name.
+   * @param args - Array or single argument; omitted means no arguments.
+   * @returns `false` when a before-enqueue plugin vetoes the job, otherwise `true`.
+   */
+  async enqueue(q: string, func: string, args: unknown = []): Promise<boolean> {
+    const normalizedArgs = arrayify(args);
+    const toRun = await runPlugins(
+      this,
+      "beforeEnqueue",
+      func,
+      q,
+      this.jobs[func],
+      normalizedArgs,
+    );
+    if (!toRun) return false;
+
+    await this.sendJob(q, this.payload(q, func, normalizedArgs));
+
+    await runPlugins(
+      this,
+      "afterEnqueue",
+      func,
+      q,
+      this.jobs[func],
+      normalizedArgs,
+    );
+    return true;
+  }
+
+  /**
+   * Schedule a job at a Unix timestamp in milliseconds.
+   *
+   * Duplicate identity is `(queue, class, args, rounded timestamp second)`.
+   * Enqueue plugins intentionally do not run until scheduler transfer semantics.
+   *
+   * @param timestamp - Unix milliseconds; numeric strings are accepted.
+   * @param q - Queue name.
+   * @param func - Registered job name.
+   * @param args - Array or single argument.
+   * @param suppressDuplicateTaskError - Return `undefined` instead of throwing.
+   * @returns `true` when inserted, or `undefined` when a duplicate is suppressed.
+   * @throws If the timestamp is invalid or the same task is already scheduled.
+   */
+  async enqueueAt(
+    timestamp: number | string,
+    q: string,
+    func: string,
+    args: unknown = [],
+    suppressDuplicateTaskError = false,
+  ): Promise<true | undefined> {
+    const normalizedArgs = arrayify(args);
+    const timestampMs = parseFiniteNumber(timestamp, "timestamp");
+    const second = Math.round(timestampMs / 1000);
+    const startAfter = new Date(second * 1000);
+    const payload = this.payload(q, func, normalizedArgs);
+    const duplicateKey = delayedLockKey(
+      this.encode(q, func, normalizedArgs),
+      second,
+    );
+
+    const acquired = await this.acquireDelayedLock(duplicateKey, startAfter);
+    if (!acquired) {
+      if (suppressDuplicateTaskError) return undefined;
+      throw new Error(DUPLICATE_ERROR);
+    }
+
+    try {
+      await this.sendJob(q, payload, { startAfter });
+    } catch (error) {
+      await this.connection.delLock(duplicateKey);
+      throw error;
+    }
+
+    return true;
+  }
+
+  /**
+   * Schedule a job after a delay.
+   *
+   * @param time - Delay in milliseconds; numeric strings are accepted.
+   * @param q - Queue name.
+   * @param func - Registered job name.
+   * @param args - Array or single argument.
+   * @param suppressDuplicateTaskError - Suppress duplicate-task errors.
+   * @returns Result from {@link enqueueAt}.
+   */
+  async enqueueIn(
+    time: number | string,
+    q: string,
+    func: string,
+    args: unknown = [],
+    suppressDuplicateTaskError = false,
+  ): Promise<true | undefined> {
+    return this.enqueueAt(
+      Date.now() + parseFiniteNumber(time, "time"),
+      q,
+      func,
+      args,
+      suppressDuplicateTaskError,
+    );
+  }
+
+  /** @returns All known pg-boss queue names and queue names present in jobs. */
+  async queues(): Promise<string[]> {
+    const [configured, jobs] = await Promise.all([
+      this.connection.boss.getQueues(),
+      this.connection.query<{ name: string }>(
+        `SELECT DISTINCT name FROM ${this.connection.schema}.job`,
+      ),
+    ]);
+    return Array.from(
+      new Set([
+        ...configured.map((queue) => queue.name),
+        ...jobs.rows.map((row) => row.name),
+      ]),
+    ).sort();
+  }
+
+  /**
+   * Delete a queue and all non-active jobs in it.
+   *
+   * @param q - Queue name.
+   * @returns Number of jobs deleted.
+   */
+  async delQueue(q: string): Promise<number> {
+    const schema = this.connection.schema;
+    const client = await this.connection.pool.connect();
+    try {
+      await client.query("BEGIN");
+      const locked = await client.query(
+        `SELECT name FROM ${schema}.queue WHERE name = $1 FOR UPDATE`,
+        [q],
+      );
+      const result = await client.query<{
+        data: unknown;
+        start_after: Date;
+      }>(
+        `DELETE FROM ${schema}.job
+         WHERE name = $1 AND state <> 'active'
+         RETURNING data, start_after`,
+        [q],
+      );
+      const remaining = await client.query<{ exists: boolean }>(
+        `SELECT EXISTS (
+           SELECT 1 FROM ${schema}.job WHERE name = $1
+         ) AS exists`,
+        [q],
+      );
+      if (!remaining.rows[0]?.exists && (locked.rowCount ?? 0) > 0) {
+        await client.query(`SELECT ${schema}.delete_queue($1)`, [q]);
+      }
+      await client.query("COMMIT");
+
+      await Promise.all(
+        result.rows
+          .filter((row) => new Date(row.start_after).getTime() > Date.now())
+          .map((row) => {
+            const payload = parseJob(row.data, q);
+            const second = Math.round(
+              new Date(row.start_after).getTime() / 1000,
+            );
+            return this.connection.delLock(
+              delayedLockKey(
+                this.encode(payload.queue, payload.class, payload.args),
+                second,
+              ),
+            );
+          }),
+      );
+      return result.rowCount ?? 0;
+    } catch (error) {
+      await client.query("ROLLBACK").catch(() => {
+        // Transaction may already be closed.
+      });
+      throw error;
+    } finally {
+      client.release();
+    }
+  }
+
+  /**
+   * Count ready jobs in a queue. Delayed jobs are excluded.
+   *
+   * @param q - Queue name.
+   * @returns Number of ready jobs.
+   */
+  async length(q: string): Promise<number> {
+    const result = await this.connection.query<{ count: string }>(
+      `SELECT count(*)::text AS count
+       FROM ${this.connection.schema}.job
+       WHERE name = $1
+         AND state = ANY($2::${this.connection.schema}.job_state[])
+         AND start_after <= now()`,
+      [q, QUEUED_STATES],
+    );
+    return Number(result.rows[0]?.count ?? 0);
+  }
+
+  /**
+   * List ready jobs using Redis `LRANGE`-style inclusive indices.
+   *
+   * @param q - Queue name.
+   * @param start - Zero-based start index.
+   * @param stop - Inclusive stop index; `-1` means the end.
+   * @returns Encoded job payloads in FIFO order.
+   */
+  async queued(q: string, start = 0, stop = -1): Promise<ParsedJob[]> {
+    const range = sqlRange(start, stop);
+    const result = await this.connection.query<JobRow>(
+      `SELECT id, name, data, created_on, start_after
+       FROM ${this.connection.schema}.job
+       WHERE name = $1
+         AND state = ANY($2::${this.connection.schema}.job_state[])
+         AND start_after <= now()
+       ORDER BY created_on, id
+       OFFSET $3
+       ${range.limitSql}`,
+      [q, QUEUED_STATES, range.offset, ...range.values],
+    );
+    return result.rows.map((row) => parseJob(row.data, row.name));
+  }
+
+  /**
+   * Delete ready jobs matching class and arguments.
+   *
+   * @param q - Queue name.
+   * @param func - Registered job name.
+   * @param args - Array or single argument.
+   * @param count - `0` deletes all; positive deletes from the front; negative from the end.
+   * @returns Number of jobs deleted.
+   */
+  async del(
+    q: string,
+    func: string,
+    args: unknown = [],
+    count = 0,
+  ): Promise<number> {
+    const payload = JSON.stringify(this.payload(q, func, arrayify(args)));
+    const direction = count < 0 ? "DESC" : "ASC";
+    const limitSql = count === 0 ? "" : "LIMIT $3";
+    const values: unknown[] = [q, payload];
+    if (count !== 0) values.push(Math.abs(count));
+
+    const result = await this.connection.query(
+      `WITH selected AS (
+         SELECT id
+         FROM ${this.connection.schema}.job
+         WHERE name = $1
+           AND state = ANY(ARRAY['created','retry']::${this.connection.schema}.job_state[])
+           AND start_after <= now()
+           AND data = $2::jsonb
+         ORDER BY created_on ${direction}, id ${direction}
+         ${limitSql}
+       )
+       DELETE FROM ${this.connection.schema}.job
+       WHERE id IN (SELECT id FROM selected)`,
+      values,
+    );
+    return result.rowCount ?? 0;
+  }
+
+  /**
+   * Delete ready jobs of one class within an inclusive queue slice.
+   *
+   * @param q - Queue name.
+   * @param func - Job class/name to remove.
+   * @param start - Zero-based slice start.
+   * @param stop - Inclusive slice end; `-1` means the end.
+   * @returns Number of jobs deleted.
+   */
+  async delByFunction(
+    q: string,
+    func: string,
+    start = 0,
+    stop = -1,
+  ): Promise<number> {
+    const range = sqlRange(start, stop);
+    const result = await this.connection.query(
+      `WITH sliced AS (
+         SELECT id, data
+         FROM ${this.connection.schema}.job
+         WHERE name = $1
+           AND state = ANY(ARRAY['created','retry']::${this.connection.schema}.job_state[])
+           AND start_after <= now()
+         ORDER BY created_on, id
+         OFFSET $3
+         ${range.limitSql}
+       )
+       DELETE FROM ${this.connection.schema}.job
+       WHERE id IN (
+         SELECT id FROM sliced WHERE data->>'class' = $2
+       )`,
+      [q, func, range.offset, ...range.values],
+    );
+    return result.rowCount ?? 0;
+  }
+
+  /**
+   * Delete all delayed jobs matching a payload.
+   *
+   * @param q - Queue name.
+   * @param func - Registered job name.
+   * @param args - Array or single argument.
+   * @returns Rounded Unix timestamps in seconds for deleted jobs.
+   */
+  async delDelayed(
+    q: string,
+    func: string,
+    args: unknown = [],
+  ): Promise<number[]> {
+    const encoded = this.encode(q, func, arrayify(args));
+    const result = await this.connection.query<{ start_after: Date }>(
+      `DELETE FROM ${this.connection.schema}.job
+       WHERE name = $1
+         AND state = ANY(ARRAY['created','retry']::${this.connection.schema}.job_state[])
+         AND start_after > now()
+         AND data = $2::jsonb
+       RETURNING start_after`,
+      [q, encoded],
+    );
+    const seconds = result.rows.map((row) =>
+      Math.round(new Date(row.start_after).getTime() / 1000),
+    );
+    await Promise.all(
+      seconds.map((second) =>
+        this.connection.delLock(delayedLockKey(encoded, second)),
+      ),
+    );
+    return seconds;
+  }
+
+  /**
+   * Find times at which a matching job is delayed.
+   *
+   * @param q - Queue name.
+   * @param func - Registered job name.
+   * @param args - Array or single argument.
+   * @returns Rounded Unix timestamps in seconds.
+   */
+  async scheduledAt(
+    q: string,
+    func: string,
+    args: unknown = [],
+  ): Promise<number[]> {
+    const result = await this.connection.query<{ start_after: Date }>(
+      `SELECT start_after
+       FROM ${this.connection.schema}.job
+       WHERE name = $1
+         AND state = ANY(ARRAY['created','retry']::${this.connection.schema}.job_state[])
+         AND start_after > now()
+         AND data = $2::jsonb
+       ORDER BY start_after, created_on, id`,
+      [q, this.encode(q, func, arrayify(args))],
+    );
+    return result.rows.map((row) =>
+      Math.round(new Date(row.start_after).getTime() / 1000),
+    );
+  }
+
+  /** @returns Distinct delayed timestamps in milliseconds, sorted ascending. */
+  async timestamps(): Promise<number[]> {
+    const result = await this.connection.query<{ start_after: Date }>(
+      `SELECT DISTINCT start_after
+       FROM ${this.connection.schema}.job
+       WHERE state = ANY(ARRAY['created','retry']::${this.connection.schema}.job_state[])
+         AND start_after > now()
+       ORDER BY start_after`,
+    );
+    return result.rows.map(
+      (row) => Math.round(new Date(row.start_after).getTime() / 1000) * 1000,
+    );
+  }
+
+  /**
+   * List jobs delayed at one rounded timestamp second.
+   *
+   * @param timestamp - Unix milliseconds; numeric strings are accepted.
+   * @returns Tasks and rounded Unix timestamp in seconds.
+   */
+  async delayedAt(
+    timestamp: number | string,
+  ): Promise<{ tasks: ParsedJob[]; rTimestamp: number }> {
+    const rTimestamp = Math.round(
+      parseFiniteNumber(timestamp, "timestamp") / 1000,
+    );
+    const result = await this.connection.query<JobRow>(
+      `SELECT id, name, data, created_on, start_after
+       FROM ${this.connection.schema}.job
+       WHERE state = ANY(ARRAY['created','retry']::${this.connection.schema}.job_state[])
+         AND start_after > now()
+         AND start_after >= to_timestamp($1)
+         AND start_after < to_timestamp($1) + interval '1 second'
+       ORDER BY created_on, id`,
+      [rTimestamp],
+    );
+    return {
+      tasks: result.rows.map((row) => parseJob(row.data, row.name)),
+      rTimestamp,
+    };
+  }
+
+  /**
+   * Load all delayed jobs grouped by timestamp milliseconds.
+   *
+   * This can be expensive for a large delayed queue.
+   *
+   * @returns Timestamp-to-task-list mapping.
+   */
+  async allDelayed(): Promise<Record<string, ParsedJob[]>> {
+    const result: Record<string, ParsedJob[]> = {};
+    for (const timestamp of await this.timestamps()) {
+      const { tasks, rTimestamp } = await this.delayedAt(timestamp);
+      result[String(rTimestamp * 1000)] = tasks;
+    }
+    return result;
+  }
+
+  /** @returns Non-expired plugin lock values keyed by lock name. */
+  async locks(): Promise<Record<string, string | null>> {
+    await this.connection.query(
+      `DELETE FROM ${this.connection.schema}.pgrq_locks
+       WHERE expires_at < now()`,
+    );
+    const result = await this.connection.query<{
+      key: string;
+      value: string | null;
+    }>(
+      `SELECT key, value
+       FROM ${this.connection.schema}.pgrq_locks
+       WHERE key LIKE 'lock:%' OR key LIKE 'workerslock:%'
+       ORDER BY key`,
+    );
+    return Object.fromEntries(result.rows.map((row) => [row.key, row.value]));
+  }
+
+  /**
+   * Delete a plugin lock.
+   *
+   * @param key - Lock key without a schema prefix.
+   * @returns Number of rows deleted.
+   */
+  async delLock(key: string): Promise<number> {
+    return this.connection.delLock(key);
+  }
+
+  /** @returns Registered workers mapped to their queue string. */
+  async workers(): Promise<Record<string, string>> {
+    const result = await this.connection.query<WorkerRow>(
+      `SELECT name, queues, working_on
+       FROM ${this.connection.schema}.pgrq_workers
+       ORDER BY name`,
+    );
+    return Object.fromEntries(result.rows.map((row) => [row.name, row.queues]));
+  }
+
+  /**
+   * Read one worker's current assignment as JSON.
+   *
+   * @param workerName - Worker name.
+   * @param queues - Expected queue string; mismatches return `null`.
+   * @returns JSON payload or `null` when idle/missing.
+   */
+  async workingOn(workerName: string, queues: string): Promise<string | null> {
+    const result = await this.connection.query<{ working_on: unknown }>(
+      `SELECT working_on
+       FROM ${this.connection.schema}.pgrq_workers
+       WHERE name = $1 AND queues = $2`,
+      [workerName, queues],
+    );
+    const value = result.rows[0]?.working_on;
+    return value == null ? null : JSON.stringify(value);
+  }
+
+  /** @returns Every worker mapped to `"started"` or its active payload. */
+  async allWorkingOn(): Promise<
+    Record<string, "started" | ParsedWorkerPayload>
+  > {
+    const result = await this.connection.query<WorkerRow>(
+      `SELECT name, queues, working_on
+       FROM ${this.connection.schema}.pgrq_workers
+       ORDER BY name`,
+    );
+    const workers: Record<string, "started" | ParsedWorkerPayload> = {};
+    for (const row of result.rows) {
+      workers[row.name] =
+        row.working_on == null
+          ? "started"
+          : parseWorkerPayload(row.working_on, row.name);
+    }
+    return workers;
+  }
+
+  /**
+   * Remove a worker and convert any active assignment into a failed job.
+   *
+   * @param workerName - Worker to clean.
+   * @returns Failure payload when work was active, otherwise `undefined`.
+   */
+  async forceCleanWorker(
+    workerName: string,
+  ): Promise<ErrorPayload | undefined> {
+    const result = await this.connection.query<WorkerRow>(
+      `DELETE FROM ${this.connection.schema}.pgrq_workers
+       WHERE name = $1
+       RETURNING name, queues, working_on`,
+      [workerName],
+    );
+    const row = result.rows[0];
+    if (!row) {
+      this.emit(
+        "error",
+        new Error(
+          `force-cleaning worker ${workerName}, but cannot find queues`,
+        ),
+      );
+      return undefined;
+    }
+    if (row.working_on == null) return undefined;
+
+    const working = parseWorkerPayload(row.working_on, workerName);
+    const message = "Worker Timeout (killed manually)";
+    const errorPayload: ErrorPayload = {
+      worker: workerName,
+      queue: working.queue,
+      payload: working.payload,
+      exception: message,
+      error: message,
+      backtrace: [
+        `killed by ${hostname()} at ${new Date()}`,
+        "queue#forceCleanWorker",
+        "node-resque",
+      ],
+      failed_at: new Date().toString(),
+    };
+
+    await this.failActiveJob(working, errorPayload);
+    await this.connection.incrStat("failed");
+    return errorPayload;
+  }
+
+  /**
+   * Mark the worker's in-flight pg-boss job failed in place.
+   *
+   * @param working - Worker assignment, including optional job id.
+   * @param errorPayload - Resque failure payload stored in `output`.
+   */
+  private async failActiveJob(
+    working: ParsedWorkerPayload,
+    errorPayload: ErrorPayload,
+  ): Promise<void> {
+    const schema = this.connection.schema;
+    const updated = await this.connection.query(
+      `WITH selected AS (
+         SELECT name, id
+         FROM ${schema}.job
+         WHERE name = $1
+           AND state = 'active'
+           AND (
+             ($4::uuid IS NOT NULL AND id = $4)
+             OR ($4::uuid IS NULL AND data = $2::jsonb)
+           )
+         ORDER BY started_on NULLS FIRST, created_on, id
+         LIMIT 1
+         FOR UPDATE
+       )
+       UPDATE ${schema}.job AS job
+       SET state = 'failed',
+           completed_on = now(),
+           output = $3::jsonb
+       FROM selected
+       WHERE job.name = selected.name AND job.id = selected.id`,
+      [
+        working.queue,
+        JSON.stringify(working.payload),
+        JSON.stringify(errorPayload),
+        working.id ?? null,
+      ],
+    );
+    if ((updated.rowCount ?? 0) > 0) return;
+    if (working.id) return;
+
+    await this.ensureQueue(working.queue);
+    await this.connection.query(
+      `INSERT INTO ${this.connection.schema}.job
+         (name, data, state, retry_limit, completed_on, output)
+       VALUES ($1, $2::jsonb, 'failed', 0, now(), $3::jsonb)`,
+      [
+        working.queue,
+        JSON.stringify(working.payload),
+        JSON.stringify(errorPayload),
+      ],
+    );
+  }
+
+  /**
+   * Force-clean workers whose active payload predates an age limit.
+   *
+   * @param age - Maximum active age in milliseconds.
+   * @returns Cleaned failures keyed by worker name.
+   */
+  async cleanOldWorkers(age: number): Promise<Record<string, ErrorPayload>> {
+    const result: Record<string, ErrorPayload> = {};
+    const workers = await this.allWorkingOn();
+    for (const [workerName, payload] of Object.entries(workers)) {
+      if (
+        payload !== "started" &&
+        Date.now() - Date.parse(payload.run_at) > age
+      ) {
+        const failure = await this.forceCleanWorker(workerName);
+        if (failure) result[workerName] = failure;
+      }
+    }
+    return result;
+  }
+
+  /** @returns Number of failed pg-boss jobs. */
+  async failedCount(): Promise<number> {
+    const result = await this.connection.query<{ count: string }>(
+      `SELECT count(*)::text AS count
+       FROM ${this.connection.schema}.job
+       WHERE state = 'failed'`,
+    );
+    return Number(result.rows[0]?.count ?? 0);
+  }
+
+  /**
+   * List failed jobs using inclusive Redis list indices.
+   *
+   * @param start - Zero-based start index.
+   * @param stop - Inclusive stop index; `-1` means all remaining jobs.
+   * @returns Resque-compatible failure payloads.
+   */
+  async failed(start = 0, stop = -1): Promise<ParsedFailedJobPayload[]> {
+    const range = sqlRange(start, stop, "$2");
+    const result = await this.connection.query<JobRow>(
+      `SELECT id, name, data, output, created_on, start_after, completed_on
+       FROM ${this.connection.schema}.job
+       WHERE state = 'failed'
+       ORDER BY completed_on, created_on, id
+       OFFSET $1
+       ${range.limitSql}`,
+      [range.offset, ...range.values],
+    );
+    return result.rows.map(mapFailedRow);
+  }
+
+  /**
+   * Remove one failed job when its current payload matches.
+   *
+   * @param failedJob - Failure object previously returned by {@link failed}.
+   * @returns Number removed (`0` or `1`).
+   */
+  async removeFailed(failedJob: ErrorPayload): Promise<number> {
+    const candidates = failedJob.id
+      ? await this.connection.query<JobRow>(
+          `SELECT id, name, data, output, created_on, start_after, completed_on
+           FROM ${this.connection.schema}.job
+           WHERE id = $1 AND state = 'failed'`,
+          [failedJob.id],
+        )
+      : await this.connection.query<JobRow>(
+          `SELECT id, name, data, output, created_on, start_after, completed_on
+           FROM ${this.connection.schema}.job
+           WHERE state = 'failed'
+           ORDER BY completed_on, created_on, id`,
+        );
+
+    const match = candidates.rows.find((row) =>
+      sameFailure(mapFailedRow(row), failedJob),
+    );
+    if (!match) return 0;
+
+    const deleted = await this.connection.query(
+      `DELETE FROM ${this.connection.schema}.job
+       WHERE id = $1 AND state = 'failed'`,
+      [match.id],
+    );
+    return deleted.rowCount ?? 0;
+  }
+
+  /**
+   * Remove a failed job and enqueue its original payload.
+   *
+   * @param failedJob - Failure object previously returned by {@link failed}.
+   * @returns Enqueue result.
+   * @throws If the failure no longer exists.
+   */
+  async retryAndRemoveFailed(failedJob: ErrorPayload): Promise<boolean> {
+    if ((await this.removeFailed(failedJob)) < 1) {
+      throw new Error("This job is not in failed queue");
+    }
+    return this.enqueue(
+      failedJob.queue,
+      failedJob.payload.class,
+      failedJob.payload.args,
+    );
+  }
+
+  /**
+   * Retry failures created by {@link forceCleanWorker}.
+   *
+   * @param upperLimit - Maximum number of failed jobs to inspect.
+   */
+  async retryStuckJobs(upperLimit = Infinity): Promise<void> {
+    const limit = Number.isFinite(upperLimit)
+      ? Math.max(0, Math.floor(upperLimit))
+      : Number.MAX_SAFE_INTEGER;
+    if (limit === 0) return;
+    const jobs = await this.failed(0, limit - 1);
+    for (const job of jobs) {
+      if (job.backtrace.includes("queue#forceCleanWorker")) {
+        await this.retryAndRemoveFailed(job);
+      }
+    }
+  }
+
+  /** @returns Current non-expired scheduler leader, or `null`. */
+  async leader(): Promise<string | null> {
+    return this.connection.currentLeader();
+  }
+
+  /**
+   * Read queue counters using node-resque's string-valued response shape.
+   *
+   * @returns Named processed/failed counters as decimal strings.
+   */
+  async stats(): Promise<Record<string, string>> {
+    const stats = await this.connection.getStats();
+    return Object.fromEntries(
+      Object.entries(stats).map(([name, value]) => [name, String(value)]),
+    );
+  }
+
+  /** @returns Stable metadata slot name used for scheduler leadership. */
+  leaderKey(): string {
+    return "default";
+  }
+
+  private payload(q: string, func: string, args: unknown[]): ParsedJob {
+    return { class: func, queue: q, args };
+  }
+
+  private async sendJob(
+    q: string,
+    payload: ParsedJob,
+    options: { startAfter?: Date } = {},
+  ): Promise<string> {
+    let lastError: Error | undefined;
+    for (let attempt = 0; attempt < 2; attempt += 1) {
+      await this.ensureQueue(q);
+      try {
+        const id = await this.connection.boss.send(q, payload, {
+          retryLimit: 0,
+          deleteAfterSeconds: 0,
+          ...options,
+        });
+        if (!id) {
+          throw new Error(
+            `pg-boss did not enqueue job "${payload.class}" on queue "${q}"`,
+          );
+        }
+        return id;
+      } catch (error) {
+        lastError = toError(error);
+        if (attempt === 0 && isMissingQueue(lastError, q)) continue;
+        throw lastError;
+      }
+    }
+    throw lastError ?? new Error(`pg-boss did not enqueue job on queue "${q}"`);
+  }
+
+  private async ensureQueue(q: string): Promise<void> {
+    const existing = await this.connection.boss.getQueue(q);
+    if (existing) return;
+    try {
+      await this.connection.boss.createQueue(q, {
+        retryLimit: 0,
+        deleteAfterSeconds: 0,
+      });
+    } catch (error) {
+      if (!(await this.connection.boss.getQueue(q))) throw error;
+    }
+  }
+
+  private async acquireDelayedLock(
+    key: string,
+    startAfter: Date,
+  ): Promise<boolean> {
+    const expiresAt = new Date(
+      Math.max(startAfter.getTime() + 1000, Date.now() + 1000),
+    );
+    const result = await this.connection.query<{ key: string }>(
+      `INSERT INTO ${this.connection.schema}.pgrq_locks (key, value, expires_at)
+       VALUES ($1, NULL, $2)
+       ON CONFLICT (key) DO UPDATE
+         SET expires_at = EXCLUDED.expires_at
+         WHERE ${this.connection.schema}.pgrq_locks.expires_at < now()
+       RETURNING key`,
+      [key, expiresAt],
+    );
+    return (result.rowCount ?? 0) > 0;
+  }
+}
+
+function arrayify(value: unknown): unknown[] {
+  return Array.isArray(value) ? value : [value];
+}
+
+function parseFiniteNumber(value: number | string, name: string): number {
+  const parsed = Number(value);
+  if (!Number.isFinite(parsed)) {
+    throw new Error(`${name} must be a finite number`);
+  }
+  return parsed;
+}
+
+function delayedLockKey(encoded: string, second: number): string {
+  const digest = createHash("sha256")
+    .update(stableJson(JSON.parse(encoded)))
+    .digest("hex");
+  return `timestamps:${digest}:delayed:${second}`;
+}
+
+function stableJson(value: unknown): string {
+  return JSON.stringify(sortUnknown(value));
+}
+
+function sortUnknown(value: unknown): unknown {
+  if (Array.isArray(value)) return value.map(sortUnknown);
+  if (isRecord(value)) {
+    return Object.fromEntries(
+      Object.keys(value)
+        .sort()
+        .map((key) => [key, sortUnknown(value[key])]),
+    );
+  }
+  return value;
+}
+
+function parseJob(value: unknown, fallbackQueue: string): ParsedJob {
+  if (!isRecord(value)) {
+    throw new Error("Stored job payload is not an object");
+  }
+  const func = value.class;
+  const queue = value.queue;
+  const args = value.args;
+  if (typeof func !== "string" || !Array.isArray(args)) {
+    throw new Error("Stored job payload is invalid");
+  }
+  return {
+    class: func,
+    queue: typeof queue === "string" ? queue : fallbackQueue,
+    args,
+  };
+}
+
+function parseWorkerPayload(
+  value: unknown,
+  fallbackWorker: string,
+): ParsedWorkerPayload {
+  if (!isRecord(value)) throw new Error("Stored worker payload is invalid");
+  const payload = parseJob(value.payload, String(value.queue ?? ""));
+  return {
+    run_at: String(value.run_at ?? ""),
+    queue: typeof value.queue === "string" ? value.queue : payload.queue,
+    worker: typeof value.worker === "string" ? value.worker : fallbackWorker,
+    payload,
+    id: typeof value.id === "string" ? value.id : undefined,
+  };
+}
+
+function mapFailedRow(row: JobRow): ParsedFailedJobPayload {
+  const payload = parseJob(row.data, row.name);
+  const output = isRecord(row.output) ? row.output : {};
+  const nested = isRecord(output.error) ? output.error : output;
+  const stack = typeof nested.stack === "string" ? nested.stack : undefined;
+  const backtrace = Array.isArray(output.backtrace)
+    ? output.backtrace.map(String)
+    : (stack?.split("\n").slice(1) ?? []);
+  const completed = row.completed_on ? new Date(row.completed_on) : new Date();
+
+  return {
+    id: row.id,
+    worker: typeof output.worker === "string" ? output.worker : "",
+    queue: typeof output.queue === "string" ? output.queue : payload.queue,
+    payload,
+    exception:
+      typeof output.exception === "string"
+        ? output.exception
+        : typeof nested.name === "string"
+          ? nested.name
+          : "Error",
+    error:
+      typeof output.error === "string"
+        ? output.error
+        : typeof nested.message === "string"
+          ? nested.message
+          : String(row.output ?? "Error"),
+    backtrace,
+    failed_at:
+      typeof output.failed_at === "string"
+        ? output.failed_at
+        : completed.toString(),
+  };
+}
+
+function sameFailure(
+  left: ParsedFailedJobPayload,
+  right: ErrorPayload,
+): boolean {
+  return (
+    left.worker === right.worker &&
+    left.queue === right.queue &&
+    left.exception === right.exception &&
+    left.error === right.error &&
+    left.failed_at === right.failed_at &&
+    JSON.stringify(left.payload) === JSON.stringify(right.payload) &&
+    JSON.stringify(left.backtrace) === JSON.stringify(right.backtrace)
+  );
+}
+
+function sqlRange(
+  start: number,
+  stop: number,
+  limitPlaceholder = "$4",
+): { offset: number; limitSql: string; values: number[] } {
+  const offset = Math.max(0, Math.floor(start));
+  if (stop < 0) return { offset, limitSql: "", values: [] };
+  const limit = Math.max(0, Math.floor(stop) - offset + 1);
+  return { offset, limitSql: `LIMIT ${limitPlaceholder}`, values: [limit] };
+}
+
+function isRecord(value: unknown): value is Record<string, unknown> {
+  return typeof value === "object" && value !== null;
+}
+
+function isMissingQueue(error: Error, q: string): boolean {
+  return error.message.includes(`Queue ${q} does not exist`);
+}
+
+function toError(error: unknown): Error {
+  return error instanceof Error ? error : new Error(String(error));
+}
diff --git a/src/index.ts b/src/index.ts
index e59d706..569561c 100644
--- a/src/index.ts
+++ b/src/index.ts
@@ -1,7 +1,5 @@
 /**
  * pgboss-queue — node-resque runtime model on PostgreSQL via pg-boss.
- *
- * Phase 2 exports {@link Connection}. Queue / Worker / Scheduler arrive in later phases.
  */
 export {
   assertSchema,
@@ -12,3 +10,17 @@ export {
   type SchedulerOptions,
   type WorkerOptions,
 } from "./core/connection.js";
+export { Plugin } from "./core/plugin.js";
+export {
+  type ParsedFailedJobPayload,
+  type ParsedJob,
+  type ParsedWorkerPayload,
+  Queue,
+} from "./core/queue.js";
+export type { ErrorPayload } from "./types/errorPayload.js";
+export type {
+  Job,
+  JobDefinition,
+  Jobs,
+  PluginConstructor,
+} from "./types/job.js";
diff --git a/src/types/errorPayload.ts b/src/types/errorPayload.ts
new file mode 100644
index 0000000..72f93fd
--- /dev/null
+++ b/src/types/errorPayload.ts
@@ -0,0 +1,21 @@
+import type { ParsedJob } from "../core/queue.js";
+
+/** Resque-compatible representation of a failed job. */
+export interface ErrorPayload {
+  /** pg-boss job id, when the failure came from the job table. */
+  id?: string;
+  /** Worker that failed the job, or an empty string when unknown. */
+  worker: string;
+  /** Queue containing the job. */
+  queue: string;
+  /** Original encoded job payload. */
+  payload: ParsedJob;
+  /** Error class/name. */
+  exception: string;
+  /** Human-readable error message. */
+  error: string;
+  /** Stack frames, excluding the leading error message. */
+  backtrace: string[];
+  /** Date string matching node-resque's failed payload. */
+  failed_at: string;
+}
diff --git a/src/types/job.ts b/src/types/job.ts
new file mode 100644
index 0000000..6dff499
--- /dev/null
+++ b/src/types/job.ts
@@ -0,0 +1,41 @@
+import type { Plugin } from "../core/plugin.js";
+
+/** A job implementation registered by name. */
+export interface JobDefinition {
+  /** Plugins run in declaration order around enqueue and perform operations. */
+  plugins?: Array<string | PluginConstructor>;
+  /** Options keyed by plugin class name. */
+  pluginOptions?: Record<string, Record<string, unknown>>;
+  /** Execute the job with its encoded arguments. */
+  perform: (...args: never[]) => unknown | Promise<unknown>;
+}
+
+/** Constructor shape accepted in a job's `plugins` list. */
+export type PluginConstructor = new (
+  worker: PluginHost,
+  func: string,
+  queue: string,
+  job: JobDefinition,
+  args: unknown[],
+  options: Record<string, unknown>,
+) => Plugin;
+
+/** Minimal host surface needed by plugin instances. */
+export interface PluginHost {
+  jobs: Jobs;
+  /** Mutable execution error exposed to plugin hooks by Worker. */
+  error?: Error;
+}
+
+/** Function-form jobs are shorthand for `{ perform: fn }`. */
+export type Job = JobDefinition | JobDefinition["perform"];
+
+/** Named job registry supplied to Queue and Worker. */
+export interface Jobs {
+  [jobName: string]: Job;
+}
+
+/** Normalize either supported job declaration into an object definition. */
+export function jobDefinition(job: Job | undefined): JobDefinition | undefined {
+  return typeof job === "function" ? { perform: job } : job;
+}