diff --git a/__tests__/core/connection.test.ts b/__tests__/core/connection.test.ts
new file mode 100644
index 0000000..f72e8a7
--- /dev/null
+++ b/__tests__/core/connection.test.ts
@@ -0,0 +1,321 @@
+import { afterAll, beforeAll, describe, expect, test } from "bun:test";
+import { Pool } from "pg";
+import { Connection } from "../../src";
+import specHelper from "../utils/specHelper";
+
+describe("connection", () => {
+  beforeAll(async () => {
+    await specHelper.connect();
+    await specHelper.dropSchema();
+    await specHelper.migrate();
+    await specHelper.cleanup();
+  });
+
+  afterAll(async () => {
+    await specHelper.cleanup();
+    await specHelper.disconnect();
+  });
+
+  test("should start with no redis keys in the namespace", async () => {
+    // Adapt: after cleanup, no job rows and no pgrq_* rows
+    const pool = await specHelper.connect();
+    const jobCount = await pool.query<{ count: string }>(
+      `SELECT count(*)::text AS count FROM ${specHelper.schema}.job`,
+    );
+    const lockCount = await pool.query<{ count: string }>(
+      `SELECT count(*)::text AS count FROM ${specHelper.schema}.pgrq_locks`,
+    );
+    const workerCount = await pool.query<{ count: string }>(
+      `SELECT count(*)::text AS count FROM ${specHelper.schema}.pgrq_workers`,
+    );
+    const leaderCount = await pool.query<{ count: string }>(
+      `SELECT count(*)::text AS count FROM ${specHelper.schema}.pgrq_leader`,
+    );
+    const statsCount = await pool.query<{ count: string }>(
+      `SELECT count(*)::text AS count FROM ${specHelper.schema}.pgrq_stats`,
+    );
+
+    expect(Number(jobCount.rows[0]?.count)).toBe(0);
+    expect(Number(lockCount.rows[0]?.count)).toBe(0);
+    expect(Number(workerCount.rows[0]?.count)).toBe(0);
+    expect(Number(leaderCount.rows[0]?.count)).toBe(0);
+    expect(Number(statsCount.rows[0]?.count)).toBe(0);
+  });
+
+  test.skip("it has loaded Lua commands", () => {
+    // Skip: No Lua (Redis-only)
+  });
+
+  describe("keys and namespaces", () => {
+    let connection: Connection;
+
+    beforeAll(async () => {
+      connection = new Connection(specHelper.cleanConnectionDetails());
+      await connection.connect();
+    });
+
+    afterAll(async () => {
+      await connection.end();
+    });
+
+    test.skip("getKeys returns appropriate keys based on matcher given", () => {
+      // Skip: Redis SCAN
+    });
+
+    test.skip("keys built with the default namespace are correct", () => {
+      // Skip: Redis key prefix
+    });
+
+    test.skip("ioredis transparent key prefix writes keys with the prefix even if they are not returned", () => {
+      // Skip: Redis keyPrefix
+    });
+
+    test("keys built with a custom namespace are correct", async () => {
+      // Adapt: `schema` option sets pg-boss schema; migrate sees that schema
+      const customSchema = "custom_namespace_test";
+      const custom = new Connection({
+        connectionString: process.env.DATABASE_URL,
+        schema: customSchema,
+      });
+      await custom.connect();
+      await custom.migrate();
+
+      expect(custom.schema).toBe(customSchema);
+
+      const result = await custom.query<{ exists: boolean }>(
+        `SELECT EXISTS (
+           SELECT 1 FROM information_schema.tables
+           WHERE table_schema = $1 AND table_name = 'pgrq_locks'
+         ) AS exists`,
+        [customSchema],
+      );
+      expect(result.rows[0]?.exists).toBe(true);
+
+      await custom.end();
+      const pool = await specHelper.connect();
+      await pool.query(`DROP SCHEMA IF EXISTS ${customSchema} CASCADE`);
+    });
+
+    test.skip("keys built with a array namespace are correct", () => {
+      // Skip: array namespace not supported
+    });
+
+    test.skip("will properly build namespace strings dynamically", () => {
+      // Skip: Redis namespace string building
+    });
+
+    test("key helper joins parts without a schema prefix", () => {
+      expect(connection.key("lock", "add", "default")).toBe("lock:add:default");
+      expect(connection.key("lock", "", "x")).toBe("lock:x");
+    });
+  });
+
+  test("will select redis db from options", async () => {
+    // Adapt: `database` string selects Postgres database
+    const databaseUrl = process.env.DATABASE_URL;
+    expect(databaseUrl).toBeDefined();
+    const url = new URL(databaseUrl ?? "");
+    const connection = new Connection({
+      host: url.hostname,
+      port: Number(url.port || 5432),
+      user: url.username,
+      password: decodeURIComponent(url.password),
+      database: url.pathname.replace(/^\//, ""),
+      schema: specHelper.schema,
+    });
+    await connection.connect();
+    const result = await connection.query<{ current_database: string }>(
+      "SELECT current_database()",
+    );
+    expect(result.rows[0]?.current_database).toBe(
+      url.pathname.replace(/^\//, ""),
+    );
+    await connection.end();
+  });
+
+  test.skip("removes empty namespace from generated key", () => {
+    // Skip: empty schema illegal; we reject
+  });
+
+  test("removes the redis event listeners when end", async () => {
+    // Adapt: pool / boss error listeners removed on end()
+    const pool = new Pool({ connectionString: process.env.DATABASE_URL });
+    const connection = new Connection({
+      pool,
+      schema: specHelper.schema,
+    });
+    await connection.connect();
+    expect(pool.listenerCount("error")).toBe(1);
+    expect(connection.boss.listenerCount("error")).toBe(1);
+    const boss = connection.boss;
+    await connection.end();
+    expect(pool.listenerCount("error")).toBe(0);
+    expect(boss.listenerCount("error")).toBe(0);
+    await pool.end();
+  });
+
+  test("connect with connectionString", async () => {
+    const connection = new Connection(specHelper.cleanConnectionDetails());
+    await connection.connect();
+    expect(connection.connected).toBe(true);
+    const result = await connection.query<{ value: number }>(
+      "SELECT 1 AS value",
+    );
+    expect(result.rows[0]?.value).toBe(1);
+    await connection.end();
+  });
+
+  test("connect with discrete host/port/user/password/database", async () => {
+    const databaseUrl = process.env.DATABASE_URL;
+    expect(databaseUrl).toBeDefined();
+    const url = new URL(databaseUrl ?? "");
+    const connection = new Connection({
+      host: url.hostname || "127.0.0.1",
+      port: Number(url.port || 5432),
+      user: url.username,
+      password: decodeURIComponent(url.password),
+      database: url.pathname.replace(/^\//, ""),
+      schema: specHelper.schema,
+    });
+    await connection.connect();
+    expect(connection.connected).toBe(true);
+    await connection.end();
+  });
+
+  test("connect with shared pool (ending Connection does not end the pool)", async () => {
+    const pool = new Pool({ connectionString: process.env.DATABASE_URL });
+    const connection = new Connection({
+      pool,
+      schema: specHelper.schema,
+    });
+    await connection.connect();
+    await connection.end();
+
+    const result = await pool.query<{ value: number }>("SELECT 1 AS value");
+    expect(result.rows[0]?.value).toBe(1);
+    await pool.end();
+  });
+
+  test("reject illegal schema", () => {
+    expect(() => new Connection({ schema: "pgboss-queue" })).toThrow(
+      /Invalid schema/,
+    );
+    expect(() => new Connection({ schema: "public; drop" })).toThrow(
+      /Invalid schema/,
+    );
+    expect(() => new Connection({ schema: "" })).toThrow(/Invalid schema/);
+  });
+
+  test("reject Redis options", () => {
+    expect(
+      () =>
+        new Connection({
+          // @ts-expect-error Redis option must be rejected at runtime
+          pkg: "ioredis",
+        }),
+    ).toThrow(/pkg/);
+    expect(
+      () =>
+        new Connection({
+          // @ts-expect-error Redis option must be rejected at runtime
+          redis: {},
+        }),
+    ).toThrow(/redis/);
+    expect(
+      () =>
+        new Connection({
+          // @ts-expect-error numeric database is Redis-only
+          database: 0,
+        }),
+    ).toThrow(/database/);
+  });
+
+  test("migrate() creates pg-boss job table and pgrq_* tables", async () => {
+    const freshSchema = "pgrq_migrate_once";
+    const pool = await specHelper.connect();
+    await pool.query(`DROP SCHEMA IF EXISTS ${freshSchema} CASCADE`);
+
+    const connection = new Connection({
+      connectionString: process.env.DATABASE_URL,
+      schema: freshSchema,
+    });
+    await connection.connect();
+    await connection.migrate();
+
+    const tables = await connection.query<{ table_name: string }>(
+      `SELECT table_name FROM information_schema.tables
+       WHERE table_schema = $1
+         AND table_name = ANY($2::text[])
+       ORDER BY table_name`,
+      [
+        freshSchema,
+        ["job", "pgrq_leader", "pgrq_locks", "pgrq_stats", "pgrq_workers"],
+      ],
+    );
+    expect(tables.rows.map((row) => row.table_name)).toEqual([
+      "job",
+      "pgrq_leader",
+      "pgrq_locks",
+      "pgrq_stats",
+      "pgrq_workers",
+    ]);
+
+    await connection.end();
+    await pool.query(`DROP SCHEMA IF EXISTS ${freshSchema} CASCADE`);
+  });
+
+  test("second migrate() is a no-op", async () => {
+    const connection = new Connection(specHelper.cleanConnectionDetails());
+    await connection.connect();
+    await connection.migrate();
+    await connection.migrate();
+    await connection.end();
+  });
+
+  test("tryLeader: only one of two connections wins; after expiry the other wins", async () => {
+    await specHelper.cleanup();
+    const a = new Connection(specHelper.cleanConnectionDetails());
+    const b = new Connection(specHelper.cleanConnectionDetails());
+    await a.connect();
+    await b.connect();
+
+    expect(await a.tryLeader("leader-a", 2)).toBe(true);
+    expect(await b.tryLeader("leader-b", 2)).toBe(false);
+    expect(await a.currentLeader()).toBe("leader-a");
+
+    // Wait for TTL expiry
+    await Bun.sleep(2100);
+    expect(await b.tryLeader("leader-b", 30)).toBe(true);
+    expect(await b.currentLeader()).toBe("leader-b");
+
+    expect(await b.releaseLeader("leader-b")).toBe(true);
+    expect(await b.currentLeader()).toBeNull();
+
+    await a.end();
+    await b.end();
+  });
+
+  test("setLockNx / expire / delLock", async () => {
+    await specHelper.cleanup();
+    const connection = new Connection(specHelper.cleanConnectionDetails());
+    await connection.connect();
+
+    expect(await connection.setLockNx("lock:test", "owner-a", 30)).toBe(true);
+    expect(await connection.setLockNx("lock:test", "owner-b", 30)).toBe(false);
+    expect(await connection.getLock("lock:test")).toBe("owner-a");
+
+    await connection.expireLock("lock:test", 1);
+    await Bun.sleep(1100);
+    expect(await connection.getLock("lock:test")).toBeNull();
+    expect(await connection.setLockNx("lock:test", "owner-b", 30)).toBe(true);
+
+    expect(await connection.delLock("lock:test")).toBe(1);
+    expect(await connection.getLock("lock:test")).toBeNull();
+
+    await connection.incrStat("processed", 2);
+    await connection.decrStat("processed", 1);
+    expect(await connection.getStats()).toEqual({ processed: 1 });
+
+    await connection.end();
+  });
+});
diff --git a/__tests__/core/connectionError.test.ts b/__tests__/core/connectionError.test.ts
new file mode 100644
index 0000000..df5b7ee
--- /dev/null
+++ b/__tests__/core/connectionError.test.ts
@@ -0,0 +1,39 @@
+import { describe, expect, test } from "bun:test";
+import { Connection } from "../../src";
+
+describe("connection error", () => {
+  test("can provide an error if connection failed", async () => {
+    await new Promise<void>((resolve, reject) => {
+      const brokenConnection = new Connection({
+        host: "127.0.0.1",
+        port: 1,
+        database: "pgboss_queue_test",
+        user: "postgres",
+        password: "postgres",
+        schema: "pgboss_queue_test",
+      });
+
+      let sawErrorEvent = false;
+
+      brokenConnection.on("error", (error: Error) => {
+        sawErrorEvent = true;
+        expect(error.message).toMatch(
+          /ECONNREFUSED|ENOTFOUND|ETIMEDOUT|EAI_AGAIN|connect/i,
+        );
+      });
+
+      brokenConnection
+        .connect()
+        .then(() => {
+          reject(new Error("expected connect() to fail"));
+        })
+        .catch((error: Error) => {
+          expect(error.message).toMatch(
+            /ECONNREFUSED|ENOTFOUND|ETIMEDOUT|EAI_AGAIN|connect/i,
+          );
+          expect(sawErrorEvent).toBe(true);
+          resolve();
+        });
+    });
+  }, 60_000);
+});
diff --git a/__tests__/utils/specHelper.ts b/__tests__/utils/specHelper.ts
index 9ac71d7..3ace860 100644
--- a/__tests__/utils/specHelper.ts
+++ b/__tests__/utils/specHelper.ts
@@ -1,4 +1,8 @@
 import { Pool, type PoolConfig } from "pg";
+import {
+  Connection,
+  type ConnectionOptions,
+} from "../../src/core/connection.js";
 
 const connectionString = process.env.DATABASE_URL;
 
@@ -8,19 +12,39 @@ if (!connectionString) {
   );
 }
 
-export const connectionDetails: PoolConfig = { connectionString };
+export const schema = "pgboss_queue_test";
 export const timeout = 500;
 export const queue = "default";
-export const schema = "pgboss_queue_test";
+
+/** Connection options shared by tests (Postgres URL + isolated schema). */
+export const connectionDetails: ConnectionOptions = {
+  connectionString,
+  schema,
+};
 
 let pool: Pool | undefined;
 
+/**
+ * Clone connection details for a fresh Connection (avoids shared mutation).
+ * @returns A shallow copy of {@link connectionDetails}.
+ */
+export function cleanConnectionDetails(): ConnectionOptions {
+  return { ...connectionDetails };
+}
+
+/**
+ * Open (or reuse) the shared helper pool and verify connectivity.
+ * @returns The shared `pg.Pool`.
+ */
 export async function connect(): Promise<Pool> {
-  pool ??= new Pool(connectionDetails);
+  pool ??= new Pool(connectionDetails as PoolConfig);
   await pool.query("SELECT 1");
   return pool;
 }
 
+/**
+ * End the shared helper pool if open.
+ */
 export async function disconnect(): Promise<void> {
   if (!pool) return;
 
@@ -28,11 +52,92 @@ export async function disconnect(): Promise<void> {
   pool = undefined;
 }
 
+/**
+ * Install pg-boss + `pgrq_*` tables into the test schema (idempotent).
+ */
+export async function migrate(): Promise<void> {
+  const connection = new Connection(cleanConnectionDetails());
+  await connection.connect();
+  await connection.migrate();
+  await connection.end();
+}
+
+/**
+ * Truncate job and metadata tables so tests start from an empty schema.
+ * No-op if the schema has not been migrated yet.
+ */
 export async function cleanup(): Promise<void> {
   const connection = await connect();
-  await connection.query("SELECT 1");
+
+  const schemaExists = await connection.query<{ exists: boolean }>(
+    `SELECT EXISTS (
+       SELECT 1 FROM information_schema.schemata WHERE schema_name = $1
+     ) AS exists`,
+    [schema],
+  );
+
+  if (!schemaExists.rows[0]?.exists) {
+    return;
+  }
+
+  const tables = await connection.query<{ table_name: string }>(
+    `SELECT table_name
+     FROM information_schema.tables
+     WHERE table_schema = $1
+       AND table_name = ANY($2::text[])`,
+    [
+      schema,
+      ["pgrq_leader", "pgrq_workers", "pgrq_locks", "pgrq_stats", "job"],
+    ],
+  );
+
+  const names = new Set(tables.rows.map((row) => row.table_name));
+
+  if (names.has("job")) {
+    await connection.query(`TRUNCATE TABLE ${schema}.job CASCADE`);
+  }
+
+  const meta = [
+    "pgrq_leader",
+    "pgrq_workers",
+    "pgrq_locks",
+    "pgrq_stats",
+  ].filter((name) => names.has(name));
+
+  if (meta.length > 0) {
+    await connection.query(
+      `TRUNCATE TABLE ${meta.map((name) => `${schema}.${name}`).join(", ")}`,
+    );
+  }
 }
 
+/**
+ * Drop the entire test schema (CASCADE). Used between files when needed.
+ */
+export async function dropSchema(): Promise<void> {
+  const connection = await connect();
+  await connection.query(`DROP SCHEMA IF EXISTS ${schema} CASCADE`);
+}
+
+/**
+ * @throws Always — dequeue lands in Phase 3.
+ */
 export async function popFromQueue(): Promise<never> {
   throw new Error("not implemented");
 }
+
+const specHelper = {
+  connectionDetails,
+  cleanConnectionDetails,
+  timeout,
+  queue,
+  schema,
+  connect,
+  disconnect,
+  migrate,
+  cleanup,
+  dropSchema,
+  popFromQueue,
+};
+
+export default specHelper;
diff --git a/bun.lock b/bun.lock
index b84ad40..311104c 100644
--- a/bun.lock
+++ b/bun.lock
@@ -6,6 +6,7 @@
       "name": "pgboss-queue",
       "dependencies": {
         "pg": "^8.23.0",
+        "pg-boss": "^12.28.0",
       },
       "devDependencies": {
         "@biomejs/biome": "^2.5.10",
@@ -83,8 +84,16 @@
 
     "bun-types": ["bun-types@1.4.0", "", { "dependencies": { "@types/node": "*" } }, "sha512-iIKw23BspnQQYd3prITOBxeUsxBHnwzX6YJfGMuNOZzeNcMmVqzIIVGRm1l69ogaPQmb4wB6BN8mA5bE9YuC5Q=="],
 
+    "cron-parser": ["cron-parser@5.10.0", "", { "dependencies": { "luxon": "^3.7.2" } }, "sha512-izNAxJyRWUP8ljBoDSub5WyrVOUlT4SLGShswE7eoRBpp6QUsSycYxLBMJlbshgPBMcPT/nrfgjNY2918ayv2A=="],
+
+    "luxon": ["luxon@3.7.2", "", {}, "sha512-vtEhXh/gNjI9Yg1u4jX/0YVPMvxzHuGgCm6tC5kZyb08yjGWGnqAjGJvcXbqQR2P3MyMEFnRbpcdFS6PBcLqew=="],
+
+    "non-error": ["non-error@0.1.0", "", {}, "sha512-TMB1uHiGsHRGv1uYclfhivcnf0/PdFp2pNqRxXjncaAsjYMoisaQJI+SSZCqRq+VliwRTC8tsMQfmrWjDMhkPQ=="],
+
     "pg": ["pg@8.23.0", "", { "dependencies": { "pg-connection-string": "^2.14.0", "pg-pool": "^3.14.0", "pg-protocol": "^1.16.0", "pg-types": "2.2.0", "pgpass": "1.0.5" }, "optionalDependencies": { "pg-cloudflare": "^1.4.0" }, "peerDependencies": { "pg-native": ">=3.0.1" }, "optionalPeers": ["pg-native"] }, "sha512-Ip2EQCngowJLGOfCwkFhPXU7/ljlhn6Rxlmy4XYfL2Y+vyRM59+8uR2xqRWKdYmbXmxCFOAmKxBuSUCdF34qLg=="],
 
+    "pg-boss": ["pg-boss@12.28.0", "", { "dependencies": { "cron-parser": "^5.10.0", "pg": "^8.23.0", "serialize-error": "^13.0.1" }, "bin": { "pg-boss": "dist/cli.js" } }, "sha512-7OaS/sYcQ8jcA9fSSlcdJc3Z9GD+/7GtQD3TlfweKpJQ/UnSbZYQHs1TECXMRY2sZo8ah4khQu9+6Z9v86z1dg=="],
+
     "pg-cloudflare": ["pg-cloudflare@1.4.0", "", {}, "sha512-Vo7z/6rrQYxpNRylp4Tlob2elzbh+N/MOQbxFVWCxS7oEx6jF53GTJFxK2WWpKuBRkmiin4Mt+xofFDjx09R0A=="],
 
     "pg-connection-string": ["pg-connection-string@2.14.0", "", {}, "sha512-XwWDGcLRGCXAR8F/AM5bG7Q+A3Wm2s6QeEjlOKZLlH3UYcguiqCWKyWXVag5TLTIjR7oOJUY8kcADaZgWPyLeg=="],
@@ -107,8 +116,14 @@
 
     "postgres-interval": ["postgres-interval@1.2.0", "", { "dependencies": { "xtend": "^4.0.0" } }, "sha512-9ZhXKM/rw350N1ovuWHbGxnGh/SNJ4cnxHiM0rxE4VN41wsg8P8zWn9hv/buK00RP4WvlOyr/RBDiptyxVbkZQ=="],
 
+    "serialize-error": ["serialize-error@13.0.1", "", { "dependencies": { "non-error": "^0.1.0", "type-fest": "^5.4.1" } }, "sha512-bBZaRwLH9PN5HbLCjPId4dP5bNGEtumcErgOX952IsvOhVPrm3/AeK1y0UHA/QaPG701eg0yEnOKsCOC6X/kaA=="],
+
     "split2": ["split2@4.2.0", "", {}, "sha512-UcjcJOWknrNkF6PLX83qcHM6KHgVKNkV62Y8a5uYDVv9ydGQVwAHMKqHdJje1VTWpljG0WYpCDhrCdAOYH4TWg=="],
 
+    "tagged-tag": ["tagged-tag@1.0.0", "", {}, "sha512-yEFYrVhod+hdNyx7g5Bnkkb0G6si8HJurOoOEgC8B/O0uXLHlaey/65KRv6cuWBNhBgHKAROVpc7QyYqE5gFng=="],
+
+    "type-fest": ["type-fest@5.8.0", "", { "dependencies": { "tagged-tag": "^1.0.0" } }, "sha512-YGYEVz3Fm5iy/AybuA0oyNFq7H4CgQNfRp/qfe8nurE1kuCeNm3/vfm9X4Mtl+qLyaKJUh5xrFZwogr41SMjYA=="],
+
     "typescript": ["typescript@7.0.2", "", { "optionalDependencies": { "@typescript/typescript-aix-ppc64": "7.0.2", "@typescript/typescript-darwin-arm64": "7.0.2", "@typescript/typescript-darwin-x64": "7.0.2", "@typescript/typescript-freebsd-arm64": "7.0.2", "@typescript/typescript-freebsd-x64": "7.0.2", "@typescript/typescript-linux-arm": "7.0.2", "@typescript/typescript-linux-arm64": "7.0.2", "@typescript/typescript-linux-loong64": "7.0.2", "@typescript/typescript-linux-mips64el": "7.0.2", "@typescript/typescript-linux-ppc64": "7.0.2", "@typescript/typescript-linux-riscv64": "7.0.2", "@typescript/typescript-linux-s390x": "7.0.2", "@typescript/typescript-linux-x64": "7.0.2", "@typescript/typescript-netbsd-arm64": "7.0.2", "@typescript/typescript-netbsd-x64": "7.0.2", "@typescript/typescript-openbsd-arm64": "7.0.2", "@typescript/typescript-openbsd-x64": "7.0.2", "@typescript/typescript-sunos-x64": "7.0.2", "@typescript/typescript-win32-arm64": "7.0.2", "@typescript/typescript-win32-x64": "7.0.2" }, "bin": { "tsc": "bin/tsc" } }, "sha512-8FYau96o3NKOhbjKi/qNvG/W5jhzxkbdm5sj9AbZ/5T5sWqn3hJgLfGx27sRKZWTvyzCP8dLRBTf5tBTSRVUNA=="],
 
     "undici-types": ["undici-types@8.3.0", "", {}, "sha512-j375ScV60dom+YkPFIfTLcOiPxkN/buHz5GobjLhixFuANaNs3C9l4GmrWqejgXWJ7BbJcFYpTEUkS1Ge8bpZQ=="],
diff --git a/docs/plans/01-repo-scaffold.md b/docs/plans/01-repo-scaffold.md
index 072a1b7..2d02d33 100644
--- a/docs/plans/01-repo-scaffold.md
+++ b/docs/plans/01-repo-scaffold.md
@@ -119,3 +119,4 @@ A compiling package, a shared `specHelper`, and CI that will run every subsequen
 - 2026-08-26: The `node-package` CI step must invoke `node` directly. `bun run test:node-package` can still shell out to `node`, but the workflow should not go through Bun for that check.
 - 2026-08-26: `noImplicitAny` is set on both `tsconfig.json` and `tsconfig.test.json` so relaxing `strict` later still bans implicit `any` in src and tests.
 - 2026-08-26: Required checks on `main` should be `complete` and `Cursor Bugbot`. The cloud agent GitHub token is not a repo admin (403 on branch protection and rulesets), so a maintainer must set that in the GitHub UI.
+- 2026-08-26: Phase 2 filled `specHelper.cleanup()` / `migrate()` / `dropSchema()` and added `pg-boss`. Smoke test remains valid; connection suite uses `*.test.ts` filenames because Bun will not discover bare `connection.ts`.
diff --git a/docs/plans/02-connection-and-schema.md b/docs/plans/02-connection-and-schema.md
index 939660e..429bf57 100644
--- a/docs/plans/02-connection-and-schema.md
+++ b/docs/plans/02-connection-and-schema.md
@@ -1,6 +1,6 @@
 # Phase 2 — Connection, schema, and automigrate primitives
 
-**Status:** not-started  
+**Status:** done  
 **Depends on:** Phase 1
 
 ## Goal
@@ -66,7 +66,7 @@ export interface MultiWorkerOptions extends WorkerOptions {
 }
 ```
 
-`Queue` / `Worker` / `Scheduler` still take `{ connection: ConnectionOptions, ... }` to match node-resque call sites.
+`Queue` / `Worker` / `Scheduler` still take `{ connection: ConnectionOptions, ... }` to match node-resque call sites. Option interfaces are exported from `src/core/connection.ts` / `src/index.ts` now so later phases do not redefine them.
 
 ### Mapping help (document in JSDoc + README)
 
@@ -77,12 +77,14 @@ export interface MultiWorkerOptions extends WorkerOptions {
 | `{ namespace: "resque" }` | `{ schema: "resque" }` (legal SQL identifier only) |
 | `{ namespace: ["a","b"] }` | not supported; use one schema name |
 
+Runtime rejection: `pkg`, `redis`, and numeric `database` throw from the `Connection` constructor.
+
 ## `Connection` class
 
 Port `src/core/connection.ts` *behavior*, not Redis:
 
-- `connect()` — create pool unless `pool` was provided; construct a `PgBoss` instance with `{ connectionString | host…, schema, migrate: false, supervise: false, schedule: false }`. Call `boss.start()` so the client is usable **without** migrating (pg-boss allows this when schema already exists; if it does not, start() may try to install — **disable migrate** and catch "schema missing" until `migrate()` runs). Verify against pg-boss version actually used: if `start()` always migrates, use the constructor `migrate: false` (documented: throws if schema absent). Tests that only connect after migrate are fine.
-- `end()` — `boss.stop({ graceful: true })`; `pool.end()` only if we created the pool.
+- `connect()` — create pool unless `pool` was provided; always wrap the pool as pg-boss `db.executeSql` so `connection.pool` is a real `pg.Pool`. Construct `PgBoss` with `{ db, schema, migrate: false, supervise: false, schedule: false, application_name }`. Call `boss.start()` when the schema is installed; if pg-boss reports "not installed", leave the pool connected so `migrate()` can run, then start boss after migrate.
+- `end()` — `boss.stop({ graceful: true, close: false })` (we own / borrow the pool); `pool.end()` only if we created the pool; remove forwarded `error` listeners.
 - `connected` boolean
 - Event `error` forwarded from pool and pg-boss
 - `key(...parts)` — **keep** as a helper for lock key strings (`["lock", func, queue, args].join(":")`) stored in `locks.key`. Do not prefix Redis-style. Tests that assert `resque-test-0:thing` are Redis-only (Phase 8 skip).
@@ -90,9 +92,9 @@ Port `src/core/connection.ts` *behavior*, not Redis:
 Expose:
 
 - `connection.boss: PgBoss`
-- `connection.pool: pg.Pool` (from boss or provided)
+- `connection.pool: pg.Pool` (owned or provided)
 - `connection.schema: string`
-- `connection.query<T>(text, values)` — parameterized, always-quoted schema interpolation only via validated identifier
+- `connection.query<T>(text, values)` — parameterized; schema identifiers are validated once and interpolated only after `assertSchema`
 
 ```ts
 async migrate(): Promise<void>
@@ -101,10 +103,11 @@ async migrate(): Promise<void>
 `migrate()` is idempotent:
 
 1. `CREATE SCHEMA IF NOT EXISTS {schema}`
-2. pg-boss migrate (instantiate a short-lived PgBoss with `migrate: true` **or** `boss.start()` on a migrator instance). Prefer the pg-boss CLI-equivalent API used in-process.
-3. Apply our metadata DDL (`CREATE TABLE IF NOT EXISTS`).
+2. Short-lived `PgBoss` with `{ db, schema, migrate: true, supervise: false, schedule: false }` → `start()` / `stop({ close: false })`
+3. Apply our metadata DDL (`CREATE TABLE IF NOT EXISTS` + indexes)
+4. `boss.start()` on the long-lived instance if it was waiting on install
 
-Workers never call this. Scheduler leader will. `specHelper` will call it in `beforeAll`.
+Workers never call this. Scheduler leader will. `specHelper.migrate()` calls it in `beforeAll`.
 
 ## Metadata DDL (ours)
 
@@ -143,8 +146,8 @@ CREATE TABLE IF NOT EXISTS {schema}.pgrq_stats (
 
 Indexes:
 
-- `pgrq_workers (ping_at)`
-- `pgrq_locks (expires_at)`
+- `pgrq_workers_ping_at_idx` on `pgrq_workers (ping_at)`
+- `pgrq_locks_expires_at_idx` on `pgrq_locks (expires_at)`
 
 Lock helpers on `Connection` (used by plugins and leader):
 
@@ -158,7 +161,7 @@ decrStat(name: string, by?: number): Promise<void>
 getStats(): Promise<Record<string, number>>
 ```
 
-Expired lock rows are treated as absent (`DELETE` where `expires_at < now()` on read, and by the scheduler sweeper).
+Expired lock rows are treated as absent (`DELETE` where `expires_at < now()` on read; `setLockNx` may take over an expired row via `ON CONFLICT … WHERE expires_at < now()`).
 
 Leader helpers:
 
@@ -171,7 +174,7 @@ releaseLeader(name: string): Promise<boolean>
 currentLeader(): Promise<string | null>
 ```
 
-Use a single transaction. This is the Redis `SET NX EX` + refresh-if-mine pattern from `scheduler.tryForLeader`.
+`tryLeader` uses a single transaction. This is the Redis `SET NX EX` + refresh-if-mine pattern from `scheduler.tryForLeader`.
 
 ## pg-boss constructor flags (every instance)
 
@@ -183,9 +186,13 @@ Use a single transaction. This is the Redis `SET NX EX` + refresh-if-mine patter
 
 We do not want two maintenance systems. pg-boss's built-in delete/archive would race our retention policy and might drop failed jobs.
 
+Dependency: `pg-boss` (installed in this phase; currently `^12`).
+
 ## Tests (this phase)
 
-Port in **this PR** (Phase 8 matrix: `connection.ts` + `connectionError.ts`). CI from Phase 1 must stay green.
+Port in **this PR** (Phase 8 matrix: `connection.test.ts` + `connectionError.test.ts`). CI from Phase 1 must stay green.
+
+Bun's test runner only discovers `*.test.ts` / `*.spec.ts` (and `_test_` / `_spec_` variants). Files live at `__tests__/core/connection.test.ts` and `__tests__/core/connectionError.test.ts` so `bun test` picks them up; **describe/test titles** still match node-resque.
 
 Port-inspired plus the Adapt rows from Phase 8:
 
@@ -193,12 +200,13 @@ Port-inspired plus the Adapt rows from Phase 8:
 - connect with discrete `host/port/user/password/database`
 - connect with shared `pool` (ending Connection does not end the pool)
 - reject illegal `schema` (`pgboss-queue`, `public; drop`, empty)
+- reject Redis options (`pkg`, `redis`, numeric `database`)
 - `migrate()` creates pg-boss `job` table and `pgrq_*` tables
 - second `migrate()` is a no-op
 - `tryLeader` : only one of two connections wins; after expiry the other wins
-- `setLockNx` / expire / `delLock`
-- connectionError (bad host)
-- fill `specHelper.cleanup()` to truncate `pgrq_*` (and `job` once pg-boss exists)
+- `setLockNx` / expire / `delLock` (+ stats smoke)
+- connectionError (bad host / port `127.0.0.1:1`)
+- `specHelper.cleanup()` truncates `pgrq_*` and `job` (CASCADE); `specHelper.migrate()` / `dropSchema()` available
 
 Do not defer these to Phase 8.
 
@@ -209,6 +217,7 @@ Do not defer these to Phase 8.
 - No job enqueue yet (that is Phase 3)
 - **`test.yaml` is green on the PR** (Postgres job runs the new files)
 - `specHelper.cleanup()` leaves no leftover rows
+- `node scripts/assert-node-package.mjs` imports `Connection` from compiled `dist`
 
 ## Next phase needs
 
@@ -216,4 +225,9 @@ Do not defer these to Phase 8.
 
 ## Lessons learned
 
-_None yet._
+- 2026-08-26: Bun only discovers test files whose names contain `.test` / `.spec` / `_test_` / `_spec_`. Porting node-resque's `__tests__/core/connection.ts` verbatim meant `bun test` silently ran only `smoke.test.ts`. Use `connection.test.ts` / `connectionError.test.ts` and keep the upstream `describe`/`test` titles; document the path rename in Phase 8.
+- 2026-08-26: Always pass a `pg.Pool` into pg-boss via `db: { executeSql }`. That keeps `connection.pool` typed as `Pool`, makes BYO-pool `end()` semantics obvious, and requires `boss.stop({ close: false })` so we do not double-close the pool.
+- 2026-08-26: With `migrate: false`, `boss.start()` throws `pg-boss is not installed` before migrate. `connect()` treats that as "pool ready, boss deferred" so `migrate()` can install, then starts the long-lived boss.
+- 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.
diff --git a/docs/plans/03-queue.md b/docs/plans/03-queue.md
index 9ed71ed..6ca8fab 100644
--- a/docs/plans/03-queue.md
+++ b/docs/plans/03-queue.md
@@ -129,4 +129,4 @@ Recommended split:
 
 ## Lessons learned
 
-_None yet._
+- 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.
diff --git a/docs/plans/08-conformance-tests.md b/docs/plans/08-conformance-tests.md
index 68105bd..d21e0cf 100644
--- a/docs/plans/08-conformance-tests.md
+++ b/docs/plans/08-conformance-tests.md
@@ -16,12 +16,12 @@ Source of truth: [actionhero/node-resque](https://github.com/actionhero/node-res
 | Landed in | Tests |
 | --- | --- |
 | Phase 1 | `specHelper` skeleton, smoke `SELECT 1`, `test.yaml` (lint / build / Postgres / complete) |
-| Phase 2 | connection + connectionError (+ illegal schema, BYO pool, migrate) |
-| Phase 3 | `__tests__/core/queue.ts` (minus live-worker slices deferred to 4) |
-| Phase 4 | `__tests__/core/worker.ts`, remaining queue worker-status, multi-process + priority extras |
-| Phase 5 | `__tests__/core/scheduler.ts`, automigrate + sweeper extras |
-| Phase 6 | `__tests__/plugins/*` |
-| Phase 7 | `__tests__/core/multiWorker.ts` |
+| 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 5 | `__tests__/core/scheduler.test.ts`, automigrate + sweeper extras |
+| Phase 6 | `__tests__/plugins/*.test.ts` |
+| Phase 7 | `__tests__/core/multiWorker.test.ts` |
 
 If a row above is missing when you start this phase, that is a **bug in an earlier phase** — go back and fix that plan/PR. Do not dump the entire suite into one late PR.
 
@@ -38,40 +38,43 @@ If a row above is missing when you start this phase, that is a **bug in an earli
 
 ```ts
 connectionDetails: ConnectionOptions
+cleanConnectionDetails(): ConnectionOptions
 timeout: number
 queue: string
 schema: string
-connect / disconnect / cleanup
-popFromQueue(): Promise<string | null>
+connect / disconnect / cleanup / migrate / dropSchema
+popFromQueue(): Promise<never> // throws until Phase 3
 ```
 
 **Isolation:** truncate + migrate once in `beforeAll` for speed, or per-file schema. Keep Bun `--max-concurrency=1` until proven otherwise. Do not pass `bun test --concurrency=1` — that is not a Bun flag. Node compatibility is the Phase 1 `node-package` job, not a second test runner.
 
+**File naming:** Bun discovers only `*.test.ts` / `*.spec.ts` (and `_test_` / `_spec_`). Keep node-resque's `describe`/`test` titles; use `__tests__/core/<name>.test.ts` instead of bare `<name>.ts`.
+
 ## Matrix
 
 Legend: **Port** = must exist and pass (ideally already, from Phases 2–7). **Skip** = Redis-only; `test.skip` with `// redis-only: …` **or** omit the file and keep the row here.
 
 Use this table as a **checklist in this phase's PR**: tick what is already green, add anything missing.
 
-### `__tests__/core/connection.ts`
+### `__tests__/core/connection.test.ts` (node-resque: `connection.ts`)
 
 | Test | Verdict | Notes |
 | --- | --- | --- |
-| should start with no redis keys in the namespace | **Adapt** | After cleanup, no `job` rows and no `pgrq_*` rows |
+| should start with no redis keys in the namespace | **Adapt** ✅ Phase 2 | After cleanup, no `job` rows and no `pgrq_*` rows |
 | it has loaded Lua commands | **Skip** | No Lua |
 | getKeys returns appropriate keys | **Skip** | Redis SCAN |
 | keys built with the default namespace | **Skip** | Redis key prefix |
 | ioredis transparent key prefix… | **Skip** | |
-| keys built with a custom namespace | **Adapt** | `schema` option sets pg-boss schema; `migrate` sees that schema |
+| keys built with a custom namespace | **Adapt** ✅ Phase 2 | `schema` option sets pg-boss schema; `migrate` sees that schema |
 | keys built with a array namespace | **Skip** | array namespace not supported |
 | will properly build namespace strings dynamically | **Skip** | |
-| will select redis db from options | **Adapt** | `database` string selects Postgres database (integration: skip if we cannot create DBs; then skip with reason) |
+| will select redis db from options | **Adapt** ✅ Phase 2 | `database` string selects Postgres database via discrete ConnectionOptions |
 | removes empty namespace from generated key | **Skip** | empty schema illegal; we reject |
-| removes the redis event listeners when end | **Adapt** | pool / boss error listeners removed on `end()` |
+| removes the redis event listeners when end | **Adapt** ✅ Phase 2 | pool / boss error listeners removed on `end()` (BYO pool) |
 
-### `__tests__/core/connectionError.ts`
+### `__tests__/core/connectionError.test.ts` (node-resque: `connectionError.ts`)
 
-**Port** — connecting to a bad host emits error / rejects. Point at `127.0.0.1:1` or invalid user.
+**Port** ✅ Phase 2 — connecting to a bad host/port emits error / rejects. Point at `127.0.0.1:1` or invalid user.
 
 ### `__tests__/core/queue.ts`
 
@@ -146,7 +149,8 @@ If an assertion cannot be identical, add a row (may already have rows from earli
 
 | Test name | node-resque assertion | Ours | Why |
 | --- | --- | --- | --- |
-| *(none yet)* | | | |
+| 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 |
 
 PRs that add rows must explain. "Postgres is different" is not enough if the Queue API can still match.
 
@@ -171,3 +175,4 @@ Docs site can describe a real API. Phase 10 can trust tests that have been runni
 - 2026-08-26 (plan): This phase is an audit, not the first test suite. Tests ship with Phases 1–7; CI has been running since Phase 1.
 - 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.
diff --git a/package.json b/package.json
index 2e9c5b4..513adab 100644
--- a/package.json
+++ b/package.json
@@ -1,6 +1,6 @@
 {
   "name": "pgboss-queue",
-  "version": "0.0.1",
+  "version": "0.1.0",
   "description": "A PostgreSQL-backed background job queue with the node-resque runtime model",
   "type": "module",
   "main": "./dist/index.js",
@@ -26,7 +26,8 @@
     "format": "biome check --write ."
   },
   "dependencies": {
-    "pg": "^8.23.0"
+    "pg": "^8.23.0",
+    "pg-boss": "^12.28.0"
   },
   "devDependencies": {
     "@biomejs/biome": "^2.5.10",
diff --git a/scripts/assert-node-package.mjs b/scripts/assert-node-package.mjs
index 6fedb03..392b01e 100644
--- a/scripts/assert-node-package.mjs
+++ b/scripts/assert-node-package.mjs
@@ -22,7 +22,8 @@ const mod = await import(url);
 
 assert.equal(typeof mod, "object");
 assert.ok(mod);
+assert.equal(typeof mod.Connection, "function");
 
 process.stdout.write(
-  `imported ${pkg.name} from ${entry} on node ${process.versions.node}\n`,
+  `imported ${pkg.name} Connection from ${entry} on node ${process.versions.node}\n`,
 );
diff --git a/src/core/connection.ts b/src/core/connection.ts
new file mode 100644
index 0000000..7ef7ff3
--- /dev/null
+++ b/src/core/connection.ts
@@ -0,0 +1,678 @@
+import { EventEmitter } from "node:events";
+import {
+  Pool,
+  type PoolConfig,
+  type QueryResult,
+  type QueryResultRow,
+} from "pg";
+import { PgBoss } from "pg-boss";
+
+const DEFAULT_SCHEMA = "pgboss_queue";
+const DEFAULT_APPLICATION_NAME = "pgboss-queue";
+const SCHEMA_PATTERN = /^[a-zA-Z_][a-zA-Z0-9_]*$/;
+const LEADER_SLOT = "default";
+
+/**
+ * Postgres connection options for pgboss-queue.
+ *
+ * Maps from node-resque Redis options as follows:
+ * - `{ host, port, password, database: 0 }` → `{ connectionString }` or
+ *   `{ host, port, user, password, database: "myapp" }`
+ * - `{ redis: ioredis }` → `{ pool: pg.Pool }`
+ * - `{ namespace: "resque" }` → `{ schema: "resque" }` (legal SQL identifier only)
+ * - `{ namespace: ["a","b"] }` → not supported; use one schema name
+ *
+ * Do not pass Redis-only options (`pkg`, `redis`, or a numeric `database`).
+ */
+export interface ConnectionOptions {
+  /** postgres:// URL. Preferred. Parsed by `pg`. */
+  connectionString?: string;
+  /** Postgres host. Default `127.0.0.1`. */
+  host?: string;
+  /** Postgres port. Default `5432`. */
+  port?: number;
+  /** Database *name* (string), not a Redis logical DB index. */
+  database?: string;
+  /** Postgres user. */
+  user?: string;
+  /** Postgres password. */
+  password?: string;
+  /** TLS flag or `pg` SSL options object. */
+  ssl?: boolean | object;
+  /**
+   * Existing node-postgres Pool. When set, we do not create or end a pool.
+   * Analogous to passing `redis: ioredisInstance`.
+   */
+  pool?: Pool;
+  /**
+   * pg-boss schema AND our metadata schema. Default `pgboss_queue`.
+   * Must match `^[a-zA-Z_][a-zA-Z0-9_]*$` (reject otherwise).
+   */
+  schema?: string;
+  /** Reported to Postgres as `application_name`. Default `pgboss-queue`. */
+  application_name?: string;
+}
+
+/**
+ * Options shared by Queue / Worker / Scheduler constructors
+ * (`{ connection: ConnectionOptions, ... }`).
+ */
+export interface QueueOptions {
+  connection?: ConnectionOptions;
+}
+
+/**
+ * Worker constructor options (Phase 4). Defined here for a stable options surface.
+ */
+export interface WorkerOptions extends QueueOptions {
+  name?: string;
+  queues?: string[] | string;
+  timeout?: number;
+  looping?: boolean;
+  id?: number;
+}
+
+/**
+ * Scheduler constructor options (Phase 5). Defined here for a stable options surface.
+ */
+export interface SchedulerOptions extends QueueOptions {
+  name?: string;
+  timeout?: number;
+  /** Leader lock TTL in seconds. Default `180`. */
+  leaderLockTimeout?: number;
+  stuckWorkerTimeout?: number | false;
+  retryStuckJobs?: boolean;
+  /** Leader runs pg-boss migrate + metadata DDL. Default `true`. */
+  automigrate?: boolean;
+  /**
+   * Leader deletes completed/cancelled jobs older than this. Default 24h.
+   * `false` disables.
+   */
+  completeJobRetentionMs?: number | false;
+}
+
+/**
+ * MultiWorker constructor options (Phase 7). Defined here for a stable options surface.
+ */
+export interface MultiWorkerOptions extends WorkerOptions {
+  minTaskProcessors?: number;
+  maxTaskProcessors?: number;
+  checkTimeout?: number;
+  maxEventLoopDelay?: number;
+}
+
+/**
+ * Postgres + pg-boss connection used by Queue, Worker, and Scheduler.
+ *
+ * Call {@link Connection.migrate} (via the elected scheduler, or explicitly in
+ * tests/deploy) before workers expect a usable schema. Instances constructed for
+ * workers/queues always use `migrate: false`, `supervise: false`, and
+ * `schedule: false` on pg-boss.
+ */
+export class Connection extends EventEmitter {
+  /** Resolved connection options (defaults applied). */
+  options: ConnectionOptions;
+  /** Whether {@link Connection.connect} succeeded (pool ready). */
+  connected: boolean;
+
+  private eventListeners: {
+    poolError?: (error: Error) => void;
+    bossError?: (error: Error) => void;
+  } = {};
+
+  private _pool: Pool | undefined;
+  private _boss: PgBoss | undefined;
+  private _schema: string;
+  private ownsPool = false;
+  private bossStarted = false;
+
+  /**
+   * @param options - Postgres connection options. Redis options are rejected.
+   * @throws If `schema` is illegal or Redis-only options are present.
+   */
+  constructor(options: ConnectionOptions = {}) {
+    super();
+    rejectRedisOptions(options);
+
+    const schema = options.schema ?? DEFAULT_SCHEMA;
+    assertSchema(schema);
+
+    this.options = {
+      host: options.host ?? "127.0.0.1",
+      port: options.port ?? 5432,
+      application_name: options.application_name ?? DEFAULT_APPLICATION_NAME,
+      schema,
+      connectionString: options.connectionString,
+      database: options.database,
+      user: options.user,
+      password: options.password,
+      ssl: options.ssl,
+      pool: options.pool,
+    };
+    this._schema = schema;
+    this.connected = false;
+  }
+
+  /** Validated pg-boss / metadata schema name. */
+  get schema(): string {
+    return this._schema;
+  }
+
+  /**
+   * Underlying `pg` pool. Available after {@link Connection.connect}.
+   * @throws If not connected.
+   */
+  get pool(): Pool {
+    if (!this._pool) {
+      throw new Error("Connection is not connected");
+    }
+    return this._pool;
+  }
+
+  /**
+   * pg-boss client started with `migrate`/`supervise`/`schedule` disabled.
+   * @throws If not connected.
+   */
+  get boss(): PgBoss {
+    if (!this._boss) {
+      throw new Error("Connection is not connected");
+    }
+    return this._boss;
+  }
+
+  /**
+   * Open the pool (unless `pool` was provided), construct pg-boss, and start it
+   * when the schema is already installed. If pg-boss is not installed yet,
+   * the pool stays usable so {@link Connection.migrate} can run.
+   *
+   * @throws On connection failure (emits `error` as well).
+   */
+  async connect(): Promise<void> {
+    if (this.connected) return;
+
+    try {
+      await this.ensurePoolAndBoss();
+      await this.tryStartBoss();
+      this.connected = true;
+    } catch (error) {
+      const err = toError(error);
+      this.emit("error", err);
+      await this.teardownPartialConnect();
+      throw err;
+    }
+  }
+
+  /**
+   * Stop pg-boss and, if we created the pool, end it. Provided pools are left open.
+   * Removes forwarded `error` listeners from the pool and boss.
+   */
+  async end(): Promise<void> {
+    this.removeForwardedListeners();
+
+    if (this._boss && this.bossStarted) {
+      await this._boss.stop({ graceful: true, close: false });
+      this.bossStarted = false;
+    }
+
+    if (this.ownsPool && this._pool) {
+      await this._pool.end();
+    }
+
+    this._boss = undefined;
+    this._pool = undefined;
+    this.ownsPool = false;
+    this.connected = false;
+  }
+
+  /**
+   * Build a lock / metadata key string (no schema prefix).
+   * Empty / whitespace-only parts are dropped.
+   *
+   * @param parts - Key segments (e.g. `lock`, function, queue, args).
+   * @returns Colon-joined key suitable for `pgrq_locks.key`.
+   */
+  key(...parts: Array<string | number | boolean | null | undefined>): string {
+    return parts
+      .map((part) => String(part ?? ""))
+      .filter((part) => part.trim().length > 0)
+      .join(":");
+  }
+
+  /**
+   * Run a parameterized query on the connection pool.
+   * Schema names must already be validated identifiers when interpolated by callers.
+   *
+   * @param text - SQL text with `$1`-style placeholders.
+   * @param values - Bound parameter values.
+   * @returns pg query result.
+   * @throws If not connected.
+   */
+  async query<T extends QueryResultRow = QueryResultRow>(
+    text: string,
+    values: unknown[] = [],
+  ): Promise<QueryResult<T>> {
+    return this.pool.query<T>(text, values);
+  }
+
+  /**
+   * Idempotently install the schema: `CREATE SCHEMA`, pg-boss migrate, and
+   * `pgrq_*` metadata tables/indexes. Intended for the scheduler leader (or tests).
+   *
+   * @throws If the pool cannot be opened or migration SQL fails.
+   */
+  async migrate(): Promise<void> {
+    if (!this._pool || !this._boss) {
+      await this.ensurePoolAndBoss();
+      this.connected = true;
+    }
+
+    const schema = this._schema;
+    await this.pool.query(`CREATE SCHEMA IF NOT EXISTS ${schema}`);
+
+    const migrator = new PgBoss({
+      db: this.dbAdapter(),
+      schema,
+      migrate: true,
+      supervise: false,
+      schedule: false,
+      application_name: this.options.application_name,
+    });
+
+    await migrator.start();
+    await migrator.stop({ graceful: true, close: false });
+
+    await this.applyMetadataDdl();
+
+    if (!this.bossStarted) {
+      await this.boss.start();
+      this.bossStarted = true;
+    }
+  }
+
+  /**
+   * Set a lock only if absent (or expired). Redis `SET NX EX` analogue.
+   *
+   * @param key - Lock key.
+   * @param value - Lock owner / payload string.
+   * @param ttlSeconds - Time-to-live in seconds.
+   * @returns `true` if this caller acquired the lock.
+   */
+  async setLockNx(
+    key: string,
+    value: string,
+    ttlSeconds: number,
+  ): Promise<boolean> {
+    const schema = this._schema;
+    const result = await this.query<{ key: string }>(
+      `INSERT INTO ${schema}.pgrq_locks (key, value, expires_at)
+       VALUES ($1, $2, now() + make_interval(secs => $3::double precision))
+       ON CONFLICT (key) DO UPDATE
+         SET value = EXCLUDED.value,
+             expires_at = EXCLUDED.expires_at
+         WHERE ${schema}.pgrq_locks.expires_at < now()
+       RETURNING key`,
+      [key, value, ttlSeconds],
+    );
+    return (result.rowCount ?? 0) > 0;
+  }
+
+  /**
+   * Read a non-expired lock value. Expired rows are deleted and treated as absent.
+   *
+   * @param key - Lock key.
+   * @returns Lock value, or `null` if missing/expired.
+   */
+  async getLock(key: string): Promise<string | null> {
+    const schema = this._schema;
+    await this.query(
+      `DELETE FROM ${schema}.pgrq_locks WHERE key = $1 AND expires_at < now()`,
+      [key],
+    );
+    const result = await this.query<{ value: string | null }>(
+      `SELECT value FROM ${schema}.pgrq_locks
+       WHERE key = $1 AND expires_at >= now()`,
+      [key],
+    );
+    return result.rows[0]?.value ?? null;
+  }
+
+  /**
+   * Delete a lock row.
+   *
+   * @param key - Lock key.
+   * @returns Number of rows deleted (`0` or `1`).
+   */
+  async delLock(key: string): Promise<number> {
+    const result = await this.query(
+      `DELETE FROM ${this._schema}.pgrq_locks WHERE key = $1`,
+      [key],
+    );
+    return result.rowCount ?? 0;
+  }
+
+  /**
+   * Refresh a lock's TTL (even if already expired, as long as the row exists).
+   *
+   * @param key - Lock key.
+   * @param ttlSeconds - New TTL from now, in seconds.
+   */
+  async expireLock(key: string, ttlSeconds: number): Promise<void> {
+    await this.query(
+      `UPDATE ${this._schema}.pgrq_locks
+       SET expires_at = now() + make_interval(secs => $2::double precision)
+       WHERE key = $1`,
+      [key, ttlSeconds],
+    );
+  }
+
+  /**
+   * Increment a named counter in `pgrq_stats`.
+   *
+   * @param name - Stat name.
+   * @param by - Amount to add. Default `1`.
+   */
+  async incrStat(name: string, by = 1): Promise<void> {
+    await this.query(
+      `INSERT INTO ${this._schema}.pgrq_stats (name, value)
+       VALUES ($1, $2)
+       ON CONFLICT (name) DO UPDATE
+         SET value = ${this._schema}.pgrq_stats.value + EXCLUDED.value`,
+      [name, by],
+    );
+  }
+
+  /**
+   * Decrement a named counter in `pgrq_stats`.
+   *
+   * @param name - Stat name.
+   * @param by - Amount to subtract. Default `1`.
+   */
+  async decrStat(name: string, by = 1): Promise<void> {
+    await this.incrStat(name, -by);
+  }
+
+  /**
+   * Read all stats as a name → value map.
+   *
+   * @returns Record of counters (missing names are absent, not zero).
+   */
+  async getStats(): Promise<Record<string, number>> {
+    const result = await this.query<{ name: string; value: string }>(
+      `SELECT name, value FROM ${this._schema}.pgrq_stats`,
+    );
+    const stats: Record<string, number> = {};
+    for (const row of result.rows) {
+      stats[row.name] = Number(row.value);
+    }
+    return stats;
+  }
+
+  /**
+   * Try to become (or refresh) the cluster leader for the default slot.
+   * Redis `SET NX EX` + refresh-if-mine pattern.
+   *
+   * @param name - Candidate leader name (e.g. `hostname:pid`).
+   * @param ttlSeconds - Leadership TTL in seconds.
+   * @returns `true` if this `name` holds leadership after the call.
+   */
+  async tryLeader(name: string, ttlSeconds: number): Promise<boolean> {
+    const schema = this._schema;
+    const client = await this.pool.connect();
+    try {
+      await client.query("BEGIN");
+      const result = await client.query<{ name: string }>(
+        `INSERT INTO ${schema}.pgrq_leader (slot, name, expires_at)
+         VALUES ($1, $2, now() + make_interval(secs => $3::double precision))
+         ON CONFLICT (slot) DO UPDATE
+           SET name = EXCLUDED.name,
+               expires_at = EXCLUDED.expires_at
+           WHERE ${schema}.pgrq_leader.expires_at < now()
+              OR ${schema}.pgrq_leader.name = EXCLUDED.name
+         RETURNING name`,
+        [LEADER_SLOT, name, ttlSeconds],
+      );
+      await client.query("COMMIT");
+      return result.rows[0]?.name === name;
+    } catch (error) {
+      await client.query("ROLLBACK");
+      throw error;
+    } finally {
+      client.release();
+    }
+  }
+
+  /**
+   * Release leadership if currently held by `name`.
+   *
+   * @param name - Leader name that should release the lock.
+   * @returns `true` if a row was deleted.
+   */
+  async releaseLeader(name: string): Promise<boolean> {
+    const result = await this.query(
+      `DELETE FROM ${this._schema}.pgrq_leader
+       WHERE slot = $1 AND name = $2`,
+      [LEADER_SLOT, name],
+    );
+    return (result.rowCount ?? 0) > 0;
+  }
+
+  /**
+   * Current non-expired leader name, if any.
+   *
+   * @returns Leader name or `null`.
+   */
+  async currentLeader(): Promise<string | null> {
+    const result = await this.query<{ name: string }>(
+      `SELECT name FROM ${this._schema}.pgrq_leader
+       WHERE slot = $1 AND expires_at >= now()`,
+      [LEADER_SLOT],
+    );
+    return result.rows[0]?.name ?? null;
+  }
+
+  private async ensurePoolAndBoss(): Promise<void> {
+    if (!this._pool) {
+      if (this.options.pool) {
+        this._pool = this.options.pool;
+        this.ownsPool = false;
+      } else {
+        this._pool = new Pool(this.buildPoolConfig());
+        this.ownsPool = true;
+      }
+    }
+
+    if (!this._boss) {
+      this._boss = new PgBoss({
+        db: this.dbAdapter(),
+        schema: this._schema,
+        migrate: false,
+        supervise: false,
+        schedule: false,
+        application_name: this.options.application_name,
+      });
+    }
+
+    this.attachForwardedListeners();
+  }
+
+  private async tryStartBoss(): Promise<void> {
+    if (!this._boss || this.bossStarted) return;
+
+    try {
+      await this._boss.start();
+      this.bossStarted = true;
+    } catch (error) {
+      if (isPgBossNotInstalled(error)) {
+        return;
+      }
+      throw error;
+    }
+  }
+
+  private async applyMetadataDdl(): Promise<void> {
+    const schema = this._schema;
+    await this.pool.query(`
+      CREATE TABLE IF NOT EXISTS ${schema}.pgrq_leader (
+        slot        text PRIMARY KEY DEFAULT 'default',
+        name        text NOT NULL,
+        expires_at  timestamptz NOT NULL
+      );
+
+      CREATE TABLE IF NOT EXISTS ${schema}.pgrq_workers (
+        name        text PRIMARY KEY,
+        queues      text NOT NULL,
+        started_at  timestamptz NOT NULL DEFAULT now(),
+        ping_at     timestamptz NOT NULL DEFAULT now(),
+        working_on  jsonb
+      );
+
+      CREATE TABLE IF NOT EXISTS ${schema}.pgrq_locks (
+        key         text PRIMARY KEY,
+        value       text,
+        expires_at  timestamptz NOT NULL
+      );
+
+      CREATE TABLE IF NOT EXISTS ${schema}.pgrq_stats (
+        name        text PRIMARY KEY,
+        value       bigint NOT NULL DEFAULT 0
+      );
+
+      CREATE INDEX IF NOT EXISTS pgrq_workers_ping_at_idx
+        ON ${schema}.pgrq_workers (ping_at);
+
+      CREATE INDEX IF NOT EXISTS pgrq_locks_expires_at_idx
+        ON ${schema}.pgrq_locks (expires_at);
+    `);
+  }
+
+  private buildPoolConfig(): PoolConfig {
+    const {
+      connectionString,
+      host,
+      port,
+      database,
+      user,
+      password,
+      ssl,
+      application_name,
+    } = this.options;
+
+    if (connectionString) {
+      return { connectionString, application_name, ssl };
+    }
+
+    return {
+      host,
+      port,
+      database,
+      user,
+      password,
+      ssl,
+      application_name,
+    };
+  }
+
+  private dbAdapter(): {
+    executeSql: (
+      text: string,
+      values?: unknown[],
+    ) => Promise<QueryResult<QueryResultRow>>;
+  } {
+    return {
+      executeSql: (text: string, values?: unknown[]) =>
+        this.pool.query(text, values),
+    };
+  }
+
+  private attachForwardedListeners(): void {
+    if (!this._pool || !this._boss) return;
+
+    this.removeForwardedListeners();
+
+    this.eventListeners.poolError = (error: Error) => {
+      this.emit("error", error);
+    };
+    this.eventListeners.bossError = (error: Error) => {
+      this.emit("error", error);
+    };
+
+    this._pool.on("error", this.eventListeners.poolError);
+    this._boss.on("error", this.eventListeners.bossError);
+  }
+
+  private removeForwardedListeners(): void {
+    if (this._pool && this.eventListeners.poolError) {
+      this._pool.off("error", this.eventListeners.poolError);
+    }
+    if (this._boss && this.eventListeners.bossError) {
+      this._boss.off("error", this.eventListeners.bossError);
+    }
+    this.eventListeners = {};
+  }
+
+  private async teardownPartialConnect(): Promise<void> {
+    this.removeForwardedListeners();
+    if (this._boss && this.bossStarted) {
+      await this._boss.stop({ graceful: false, close: false }).catch(() => {
+        // best-effort cleanup after a failed connect
+      });
+    }
+    if (this.ownsPool && this._pool) {
+      await this._pool.end().catch(() => {
+        // best-effort cleanup after a failed connect
+      });
+    }
+    this._boss = undefined;
+    this._pool = undefined;
+    this.ownsPool = false;
+    this.bossStarted = false;
+    this.connected = false;
+  }
+}
+
+/**
+ * @param schema - Candidate schema identifier.
+ * @throws If `schema` is not a bare SQL identifier.
+ */
+export function assertSchema(schema: string): void {
+  if (!SCHEMA_PATTERN.test(schema)) {
+    throw new Error(`Invalid schema "${schema}": must match ${SCHEMA_PATTERN}`);
+  }
+}
+
+function rejectRedisOptions(options: ConnectionOptions): void {
+  const raw = options as ConnectionOptions & {
+    pkg?: unknown;
+    redis?: unknown;
+    database?: unknown;
+  };
+
+  if (raw.pkg !== undefined) {
+    throw new Error(
+      'Redis option "pkg" is not supported; use Postgres ConnectionOptions',
+    );
+  }
+  if (raw.redis !== undefined) {
+    throw new Error(
+      'Redis option "redis" is not supported; pass a pg Pool as "pool"',
+    );
+  }
+  if (typeof raw.database === "number") {
+    throw new Error(
+      'Connection option "database" must be a Postgres database name (string), not a Redis DB index',
+    );
+  }
+}
+
+function isPgBossNotInstalled(error: unknown): boolean {
+  const message = toError(error).message.toLowerCase();
+  return (
+    message.includes("not installed") ||
+    (message.includes("schema") && message.includes("missing"))
+  );
+}
+
+function toError(error: unknown): Error {
+  if (error instanceof Error) return error;
+  return new Error(String(error));
+}
diff --git a/src/index.ts b/src/index.ts
index cb0ff5c..e59d706 100644
--- a/src/index.ts
+++ b/src/index.ts
@@ -1 +1,14 @@
-export {};
+/**
+ * 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,
+  Connection,
+  type ConnectionOptions,
+  type MultiWorkerOptions,
+  type QueueOptions,
+  type SchedulerOptions,
+  type WorkerOptions,
+} from "./core/connection.js";