diff --git a/src/advisory-lock.test.ts b/src/advisory-lock.test.ts
index 690fd99..06beccb 100644
--- a/src/advisory-lock.test.ts
+++ b/src/advisory-lock.test.ts
@@ -1,5 +1,5 @@
 import { describe, expect } from "vitest";
-import { createPool } from "slonik";
+import { createPool, sql } from "slonik";
 
 import { pgsliceTest as test } from "./testing/index.js";
 import { AdvisoryLock, AdvisoryLockError } from "./advisory-lock.js";
@@ -20,7 +20,9 @@ describe("AdvisoryLock.withLock", () => {
     expect(result).toBe("success");
   });
 
-  test("releases lock even if handler throws", async ({ transaction }) => {
+  test("propagates the handler's error and stays re-acquirable in the transaction", async ({
+    transaction,
+  }) => {
     const table = Table.parse("test_table");
 
     await expect(
@@ -29,7 +31,7 @@ describe("AdvisoryLock.withLock", () => {
       }),
     ).rejects.toThrow("handler error");
 
-    // Should be able to acquire the lock again since it was released
+    // Re-acquirable within the same transaction (xact advisory locks are re-entrant)
     const result = await AdvisoryLock.withLock(
       transaction,
       table,
@@ -75,6 +77,49 @@ describe("AdvisoryLock.withLock", () => {
       await pool2.end();
     }
   });
+
+  test("does not leak the lock when the handler aborts the transaction", async ({
+    databaseUrl,
+  }) => {
+    const table = Table.parse("test_table");
+    const operation = "test_op";
+
+    const pool1 = await createPool(databaseUrl.toString(), {
+      maximumPoolSize: 1,
+      queryRetryLimit: 0,
+    });
+    const pool2 = await createPool(databaseUrl.toString(), {
+      maximumPoolSize: 1,
+      queryRetryLimit: 0,
+    });
+
+    try {
+      // The handler aborts its transaction with a failing statement. A
+      // session-scoped lock survived the rollback (leaked); a transaction-scoped
+      // lock is released when the transaction rolls back.
+      await expect(
+        pool1.transaction((tx1) =>
+          AdvisoryLock.withLock(tx1, table, operation, async () => {
+            await tx1.query(sql.unsafe`SELECT 1 FROM pgslice_missing_relation`);
+          }),
+        ),
+      ).rejects.toThrow();
+
+      // A different session can now take the lock — proof it was released.
+      await pool2.transaction(async (tx2) => {
+        const result = await AdvisoryLock.withLock(
+          tx2,
+          table,
+          operation,
+          async () => "acquired after rollback",
+        );
+        expect(result).toBe("acquired after rollback");
+      });
+    } finally {
+      await pool1.end();
+      await pool2.end();
+    }
+  });
 });
 
 describe("AdvisoryLock.acquire", () => {
diff --git a/src/advisory-lock.ts b/src/advisory-lock.ts
index e6d5aa2..31f4e2c 100644
--- a/src/advisory-lock.ts
+++ b/src/advisory-lock.ts
@@ -16,8 +16,15 @@ export class AdvisoryLockError extends Error {
 
 export abstract class AdvisoryLock {
   /**
-   * Executes a handler while holding an advisory lock.
-   * The lock is automatically released when the handler completes or throws.
+   * Runs a handler while holding a transaction-scoped advisory lock. The lock is
+   * acquired on the given transaction and released automatically when that
+   * transaction commits or rolls back — so a handler that aborts the transaction
+   * cannot leak the lock, and there is no unlock query that could fail on an
+   * aborted transaction and mask the handler's error.
+   *
+   * Must be called inside a transaction (every pgslice command wraps it in
+   * `connection.transaction(...)`). For a lock that must span multiple
+   * transactions — e.g. a batched generator — use {@link acquire} instead.
    */
   static async withLock<T>(
     connection: CommonQueryMethods,
@@ -25,12 +32,11 @@ export abstract class AdvisoryLock {
     operation: string,
     handler: () => Promise<T>,
   ): Promise<T> {
-    const release = await this.acquire(connection, table, operation);
-    try {
-      return await handler();
-    } finally {
-      await release();
+    const key = await this.#getKey(connection, table, operation);
+    if (!(await this.#tryAcquireXact(connection, key))) {
+      throw new AdvisoryLockError(table, operation);
     }
+    return handler();
   }
 
   /**
@@ -80,6 +86,18 @@ export abstract class AdvisoryLock {
     return result.acquired;
   }
 
+  static async #tryAcquireXact(
+    connection: CommonQueryMethods,
+    key: bigint,
+  ): Promise<boolean> {
+    const result = await connection.one(
+      sql.type(z.object({ acquired: z.boolean() }))`
+        SELECT pg_try_advisory_xact_lock(${key}) AS acquired
+      `,
+    );
+    return result.acquired;
+  }
+
   static async #release(
     connection: CommonQueryMethods,
     key: bigint,
diff --git a/src/cli.ts b/src/cli.ts
index d9156d6..9f0e2a5 100644
--- a/src/cli.ts
+++ b/src/cli.ts
@@ -7,6 +7,7 @@ import { DisableRetiredMirroringCommand } from "./commands/disable-retired-mirro
 import { EnableMirroringCommand } from "./commands/enable-mirroring.js";
 import { EnableRetiredMirroringCommand } from "./commands/enable-retired-mirroring.js";
 import { FillCommand } from "./commands/fill.js";
+import { MaintainCommand } from "./commands/maintain.js";
 import { PrepCommand } from "./commands/prep.js";
 import { StatusCommand } from "./commands/status.js";
 import { SwapCommand } from "./commands/swap.js";
@@ -30,6 +31,7 @@ export function createCli(): Cli {
   cli.register(EnableMirroringCommand);
   cli.register(EnableRetiredMirroringCommand);
   cli.register(FillCommand);
+  cli.register(MaintainCommand);
   cli.register(PrepCommand);
   cli.register(StatusCommand);
   cli.register(SwapCommand);
diff --git a/src/commands/base.ts b/src/commands/base.ts
index c90c04e..768aa6e 100644
--- a/src/commands/base.ts
+++ b/src/commands/base.ts
@@ -25,7 +25,7 @@ export abstract class BaseCommand extends Command<Context> {
         new URL(this.getDatabaseUrl()),
       );
 
-      await this.perform(this.context.pgslice);
+      return await this.perform(this.context.pgslice);
     } catch (error) {
       if (error instanceof Error) {
         this.context.stderr.write(`${error.message}\n`);
diff --git a/src/commands/maintain.test.ts b/src/commands/maintain.test.ts
new file mode 100644
index 0000000..5ebd94e
--- /dev/null
+++ b/src/commands/maintain.test.ts
@@ -0,0 +1,233 @@
+import { describe, expect } from "vitest";
+import { sql, type DatabaseTransactionConnection } from "slonik";
+
+import { commandTest as test } from "../testing/index.js";
+import { MaintainCommand } from "./maintain.js";
+
+interface LogEntry {
+  jobId: string;
+  msg: string;
+  level: string;
+  target: { db: string; host: string; schema?: string; table?: string };
+  future?: { daily: number; weekly: number; monthly: number; yearly: number };
+  partitions?: { new: number; total: number };
+  success?: number;
+  succeeded?: { count: number; tables: string[] };
+  failed?: { count: number; tables: string[] };
+}
+
+/** Parse the command's stdout as JSONL (one log record per line). */
+function jsonLines(output: string | undefined): LogEntry[] {
+  return (output ?? "")
+    .split("\n")
+    .filter((line) => line.trim().length > 0)
+    .map((line): LogEntry => JSON.parse(line));
+}
+
+describe("MaintainCommand", () => {
+  test.scoped({ commandClass: ({}, use) => use(MaintainCommand) });
+
+  async function createPosts(transaction: DatabaseTransactionConnection) {
+    await transaction.query(sql.unsafe`
+      CREATE TABLE posts (
+        id bigint NOT NULL,
+        created_at timestamp without time zone NOT NULL,
+        PRIMARY KEY (id, created_at)
+      ) PARTITION BY RANGE (created_at)
+    `);
+    await transaction.query(sql.unsafe`
+      COMMENT ON TABLE posts IS 'column:created_at,period:month,cast:date,version:3'
+    `);
+    await transaction.query(sql.unsafe`
+      CREATE TABLE posts_y2026m01 PARTITION OF posts
+      FOR VALUES FROM ('2026-01-01') TO ('2026-02-01')
+    `);
+  }
+
+  test("emits JSONL and exits 0 on a healthy fleet", async ({
+    cli,
+    commandContext,
+    transaction,
+  }) => {
+    await createPosts(transaction);
+
+    const exitCode = await cli.run(["maintain"], commandContext);
+    expect(exitCode).toBe(0);
+
+    const logs = jsonLines(commandContext.stdout.read()?.toString());
+
+    // Every record shares the shape, uses only info/error, and never leaks the
+    // partitioning model.
+    for (const entry of logs) {
+      expect(typeof entry.msg).toBe("string");
+      expect(["info", "error"]).toContain(entry.level);
+      expect(entry.target.db).toBeTypeOf("string");
+      expect(entry.target.host).toBe("localhost");
+      expect(JSON.stringify(entry)).not.toContain("model");
+    }
+
+    // Every record from one run shares a single generated jobId (a UUID).
+    const jobIds = new Set(logs.map((entry) => entry.jobId));
+    expect(jobIds.size).toBe(1);
+    expect([...jobIds][0]).toMatch(
+      /^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/i,
+    );
+
+    const start = logs[0];
+    expect(start.msg).toBe("Running pgslice maintain");
+    expect(start.level).toBe("info");
+    expect(start.future).toEqual({
+      daily: 90,
+      weekly: 26,
+      monthly: 6,
+      yearly: 1,
+    });
+
+    const table = logs.find((entry) => entry.target.table === "posts");
+    expect(table?.msg).toBe("Extended table successfully");
+    expect(table?.level).toBe("info");
+    expect(table?.success).toBe(1);
+    expect(table?.target.schema).toBe("public");
+    expect(table?.partitions).toEqual({
+      new: expect.any(Number),
+      total: expect.any(Number),
+    });
+
+    const final = logs.at(-1);
+    expect(final?.msg).toBe("Finished pgslice maintain successfully");
+    expect(final?.level).toBe("info");
+    expect(final?.succeeded).toEqual({ count: 1, tables: ["public.posts"] });
+    expect(final?.failed).toEqual({ count: 0, tables: [] });
+  });
+
+  test("exits 1 and logs an error record when a leaf has no usable replica identity", async ({
+    cli,
+    commandContext,
+    transaction,
+  }) => {
+    await createPosts(transaction);
+    await transaction.query(sql.unsafe`
+      ALTER TABLE posts_y2026m01 REPLICA IDENTITY NOTHING
+    `);
+
+    const exitCode = await cli.run(["maintain"], commandContext);
+    expect(exitCode).toBe(1);
+
+    const logs = jsonLines(commandContext.stdout.read()?.toString());
+
+    const table = logs.find((entry) => entry.target.table === "posts");
+    expect(table?.level).toBe("error");
+    expect(table?.success).toBe(0);
+    expect(table?.msg).toContain("replica identity");
+    expect(table?.target.host).toBe("localhost");
+
+    const final = logs.at(-1);
+    expect(final?.msg).toBe("Finished pgslice maintain with errors");
+    expect(final?.level).toBe("error");
+    expect(final?.failed).toEqual({ count: 1, tables: ["public.posts"] });
+  });
+
+  test("exits 0 with an empty summary when no managed tables exist", async ({
+    cli,
+    commandContext,
+    transaction,
+  }) => {
+    await transaction.query(sql.unsafe`
+      CREATE TABLE plain (id bigint NOT NULL, created_at timestamptz NOT NULL)
+      PARTITION BY RANGE (created_at)
+    `);
+
+    const exitCode = await cli.run(["maintain"], commandContext);
+    expect(exitCode).toBe(0);
+
+    const logs = jsonLines(commandContext.stdout.read()?.toString());
+    expect(logs[0].msg).toBe("Running pgslice maintain");
+    expect(logs.some((entry) => entry.target.table !== undefined)).toBe(false);
+
+    const final = logs.at(-1);
+    expect(final?.msg).toBe("Finished pgslice maintain successfully");
+    expect(final?.succeeded).toEqual({ count: 0, tables: [] });
+    expect(final?.failed).toEqual({ count: 0, tables: [] });
+  });
+
+  test("logs a no-op record when a table already has runway beyond the horizon", async ({
+    cli,
+    commandContext,
+    transaction,
+  }) => {
+    // A managed parent whose only partition is far in the future: the forward
+    // horizon is already covered, so maintain creates nothing.
+    await transaction.query(sql.unsafe`
+      CREATE TABLE posts (
+        id bigint NOT NULL,
+        created_at timestamp without time zone NOT NULL,
+        PRIMARY KEY (id, created_at)
+      ) PARTITION BY RANGE (created_at)
+    `);
+    await transaction.query(sql.unsafe`
+      COMMENT ON TABLE posts IS 'column:created_at,period:month,cast:date,version:3'
+    `);
+    await transaction.query(sql.unsafe`
+      CREATE TABLE posts_y2099m01 PARTITION OF posts
+      FOR VALUES FROM ('2099-01-01') TO ('2099-02-01')
+    `);
+
+    const exitCode = await cli.run(["maintain"], commandContext);
+    expect(exitCode).toBe(0);
+
+    const logs = jsonLines(commandContext.stdout.read()?.toString());
+    const table = logs.find((entry) => entry.target.table === "posts");
+    expect(table?.msg).toBe("Table already up to date; no extension needed");
+    expect(table?.level).toBe("info");
+    expect(table?.success).toBe(1);
+    expect(table?.partitions).toEqual({ new: 0, total: 1 });
+    expect(table?.target.host).toBe("localhost");
+  });
+
+  test("reads per-period horizons from PGSLICE_FUTURE_* env vars", async ({
+    cli,
+    commandContext,
+    transaction,
+  }) => {
+    await createPosts(transaction);
+    process.env.PGSLICE_FUTURE_MONTHLY = "3";
+    process.env.PGSLICE_FUTURE_WEEKLY = "9";
+    try {
+      const exitCode = await cli.run(["maintain"], commandContext);
+      expect(exitCode).toBe(0);
+
+      const logs = jsonLines(commandContext.stdout.read()?.toString());
+      // env overrides the baked default for its period; unset periods keep theirs.
+      expect(logs[0].future).toEqual({
+        daily: 90,
+        weekly: 9,
+        monthly: 3,
+        yearly: 1,
+      });
+    } finally {
+      delete process.env.PGSLICE_FUTURE_MONTHLY;
+      delete process.env.PGSLICE_FUTURE_WEEKLY;
+    }
+  });
+
+  test("an explicit --future-monthly flag overrides the env var", async ({
+    cli,
+    commandContext,
+    transaction,
+  }) => {
+    await createPosts(transaction);
+    process.env.PGSLICE_FUTURE_MONTHLY = "3";
+    try {
+      const exitCode = await cli.run(
+        ["maintain", "--future-monthly", "7"],
+        commandContext,
+      );
+      expect(exitCode).toBe(0);
+
+      const logs = jsonLines(commandContext.stdout.read()?.toString());
+      expect(logs[0].future?.monthly).toBe(7);
+    } finally {
+      delete process.env.PGSLICE_FUTURE_MONTHLY;
+    }
+  });
+});
diff --git a/src/commands/maintain.ts b/src/commands/maintain.ts
new file mode 100644
index 0000000..d36f7e0
--- /dev/null
+++ b/src/commands/maintain.ts
@@ -0,0 +1,114 @@
+import { Command, Option } from "clipanion";
+import * as t from "typanion";
+
+import { BaseCommand } from "./base.js";
+import { Pgslice } from "../pgslice.js";
+import type { MaintainLog } from "../types.js";
+
+export class MaintainCommand extends BaseCommand {
+  static override paths = [["maintain"]];
+
+  static override usage = Command.Usage({
+    description:
+      "Extend every managed partitioned table discovered from the catalog",
+    details: `
+      Discovers every partitioned table that carries a valid pgslice settings
+      comment and extends each one with add_partitions, so newly partitioned
+      tables are picked up automatically with no per-table configuration.
+
+      Native tables (the partitioned parent owns the primary key) and classic
+      pgslice tables (each partition owns its own) are both handled. After
+      extending each table this checks that every leaf partition has a replica
+      identity usable for logical replication, and exits non-zero if any does
+      not.
+    `,
+    examples: [
+      [
+        "Extend all managed tables with the default per-period horizons",
+        "$0 maintain",
+      ],
+      [
+        "Override one period's horizon (12 months of monthly runway)",
+        "$0 maintain --future-monthly 12",
+      ],
+      [
+        "Configure a horizon via environment (e.g. from a scheduled job)",
+        "PGSLICE_FUTURE_MONTHLY=12 $0 maintain",
+      ],
+      ["Restrict to a single schema", "$0 maintain --schema analytics"],
+    ],
+  });
+
+  past = Option.String("--past", "0", {
+    description: "Number of past partitions to add to each table",
+    validator: t.cascade(t.isNumber(), [t.isInteger(), t.isAtLeast(0)]),
+  });
+  futureDaily = Option.String("--future-daily", "90", {
+    description: "Future partitions to keep ahead for daily tables",
+    env: "PGSLICE_FUTURE_DAILY",
+    validator: t.cascade(t.isNumber(), [t.isInteger(), t.isAtLeast(0)]),
+  });
+  futureWeekly = Option.String("--future-weekly", "26", {
+    description: "Future partitions to keep ahead for weekly tables",
+    env: "PGSLICE_FUTURE_WEEKLY",
+    validator: t.cascade(t.isNumber(), [t.isInteger(), t.isAtLeast(0)]),
+  });
+  futureMonthly = Option.String("--future-monthly", "6", {
+    description: "Future partitions to keep ahead for monthly tables",
+    env: "PGSLICE_FUTURE_MONTHLY",
+    validator: t.cascade(t.isNumber(), [t.isInteger(), t.isAtLeast(0)]),
+  });
+  futureYearly = Option.String("--future-yearly", "1", {
+    description: "Future partitions to keep ahead for yearly tables",
+    env: "PGSLICE_FUTURE_YEARLY",
+    validator: t.cascade(t.isNumber(), [t.isInteger(), t.isAtLeast(0)]),
+  });
+  schema = Option.String("--schema", {
+    description: "Restrict maintenance to partitioned tables in this schema",
+  });
+  tablespace = Option.String("--tablespace", {
+    description: "Tablespace to use for new partitions",
+  });
+  inheritGrants = Option.Boolean("--inherit-grants", true, {
+    description:
+      "Copy each parent table's privileges onto its new partitions (default: true; disable with --no-inherit-grants)",
+  });
+
+  override async perform(pgslice: Pgslice): Promise<number | void> {
+    // maintain emits structured records (start, per-table, final); write each as
+    // one JSON object per line (JSONL).
+    const log: MaintainLog = (entry) => {
+      this.context.stdout.write(`${JSON.stringify(entry)}\n`);
+    };
+
+    // Host only — never the credentials in the URL — so the logs identify the
+    // endpoint being maintained without leaking secrets.
+    const host = new URL(this.getDatabaseUrl()).hostname;
+
+    const results = await pgslice.start((connection) =>
+      pgslice.maintain(
+        connection,
+        {
+          host,
+          past: this.past,
+          futureDaily: this.futureDaily,
+          futureWeekly: this.futureWeekly,
+          futureMonthly: this.futureMonthly,
+          futureYearly: this.futureYearly,
+          schema: this.schema,
+          tablespace: this.tablespace,
+          inheritGrants: this.inheritGrants,
+        },
+        log,
+      ),
+    );
+
+    // Exit non-zero if any table failed to extend or shipped a CDC-unsafe leaf;
+    // as a CronJob that surfaces as a failed Job. The final summary log already
+    // carries the detail, so no separate error line is written.
+    const failed = results.some(
+      (result) => result.error !== null || !result.replicaIdentityReady,
+    );
+    return failed ? 1 : undefined;
+  }
+}
diff --git a/src/maintain.test.ts b/src/maintain.test.ts
new file mode 100644
index 0000000..5c25f92
--- /dev/null
+++ b/src/maintain.test.ts
@@ -0,0 +1,737 @@
+import { afterEach, beforeEach, describe, expect, vi } from "vitest";
+import { sql } from "slonik";
+import { z } from "zod";
+
+import { pgsliceTest as test } from "./testing/index.js";
+import {
+  addChild,
+  addDefault,
+  addMinvalueChild,
+  childNames,
+  createCdcRole,
+  DATE,
+  nativeParent,
+  pgsliceParent,
+  TS,
+  TSTZ,
+} from "./testing/shapes.js";
+
+/** A captured maintain JSONL log record, for asserting record shape from tests. */
+interface MaintainLogRecord {
+  jobId: string;
+  msg: string;
+  level: string;
+  success?: number;
+  target: { host?: string; db: string; schema?: string; table?: string };
+}
+
+/**
+ * End-to-end validation of `maintain` against the real production table shapes
+ * on PostgreSQL 13.20: catalog discovery, the native (parent-owned PK) vs
+ * classic-pgslice (per-partition PK) distinction, grant inheritance, the
+ * replica-identity behavior verified on PG 13 (the partitioned replica-identity
+ * index propagates to new partitions automatically; the per-leaf identity falls
+ * back to the composite key), idempotency, and the CDC-readiness guard.
+ */
+describe("Pgslice.maintain (fleet)", () => {
+  beforeEach(() => {
+    vi.useFakeTimers();
+    // 2026-01-15 is a Thursday; its ISO week is 2026-W03 (Monday 2026-01-12).
+    vi.setSystemTime(new Date(Date.UTC(2026, 0, 15)));
+  });
+
+  afterEach(() => {
+    vi.useRealTimers();
+  });
+
+  describe("the full production fleet on one database", () => {
+    test.beforeEach(async ({ transaction }) => {
+      // 7 native tables.
+      for (const posts of ["posts", "comments", "reactions"]) {
+        await nativeParent(
+          transaction,
+          posts,
+          "created_at",
+          TS,
+          "month",
+          "date",
+        );
+        await addChild(
+          transaction,
+          posts,
+          `${posts}_y2026m01`,
+          "2026-01-01",
+          "2026-02-01",
+        );
+      }
+
+      await nativeParent(
+        transaction,
+        "daily_reports",
+        "created_date",
+        DATE,
+        "month",
+        "date",
+      );
+      await addChild(
+        transaction,
+        "daily_reports",
+        "daily_reports_202601",
+        "2026-01-01",
+        "2026-02-01",
+      );
+
+      await nativeParent(
+        transaction,
+        "analytics.visits",
+        "created_at",
+        TSTZ,
+        "week",
+        "timestamptz",
+      );
+      await addChild(
+        transaction,
+        "analytics.visits",
+        "visits_2026w03",
+        "2026-01-12 00:00:00+00",
+        "2026-01-19 00:00:00+00",
+      );
+      await addDefault(transaction, "analytics.visits", "visits_default");
+
+      await nativeParent(
+        transaction,
+        "audit_log",
+        "occurred_at",
+        TS,
+        "week",
+        "date",
+      );
+      await addChild(
+        transaction,
+        "audit_log",
+        "audit_log_y2026w02",
+        "2026-01-05",
+        "2026-01-12",
+      );
+
+      await nativeParent(
+        transaction,
+        "outbox",
+        "created_at",
+        TSTZ,
+        "week",
+        "timestamptz",
+      );
+      // MINVALUE historic partition, ignored when anchoring.
+      await addMinvalueChild(
+        transaction,
+        "outbox",
+        "outbox_historic",
+        "2025-01-01 00:00:00+00",
+      );
+      await addChild(
+        transaction,
+        "outbox",
+        "outbox_p2026w03",
+        "2026-01-12 00:00:00+00",
+        "2026-01-19 00:00:00+00",
+      );
+
+      // 3 classic-pgslice tables (parent has no PK; each partition owns its own).
+      for (const sessions of ["sessions", "tokens", "messages"]) {
+        await pgsliceParent(
+          transaction,
+          sessions,
+          "created_at",
+          TSTZ,
+          "week",
+          "timestamptz",
+        );
+        await addChild(
+          transaction,
+          sessions,
+          `${sessions}_2026w03`,
+          "2026-01-12 00:00:00+00",
+          "2026-01-19 00:00:00+00",
+          ["id", "created_at"],
+        );
+      }
+
+      // Noise that discovery must ignore: a partitioned parent with no settings
+      // comment, and an ordinary (non-partitioned) table.
+      await transaction.query(sql.unsafe`
+        CREATE TABLE unmanaged_part (
+          id bigint NOT NULL,
+          created_at timestamp with time zone NOT NULL,
+          PRIMARY KEY (id, created_at)
+        ) PARTITION BY RANGE (created_at)
+      `);
+      await addChild(
+        transaction,
+        "unmanaged_part",
+        "unmanaged_part_2026m01",
+        "2026-01-01 00:00:00+00",
+        "2026-02-01 00:00:00+00",
+      );
+      await transaction.query(
+        sql.unsafe`CREATE TABLE regular_table (id bigint PRIMARY KEY)`,
+      );
+    });
+
+    test("discovers exactly the settings-commented partitioned tables", async ({
+      pgslice,
+      transaction,
+    }) => {
+      const results = await pgslice.maintain(transaction, {
+        futureDaily: 3,
+        futureWeekly: 3,
+        futureMonthly: 3,
+        futureYearly: 3,
+      });
+
+      expect(results.map((r) => r.table).sort()).toEqual(
+        [
+          "public.outbox",
+          "public.daily_reports",
+          "public.tokens",
+          "public.audit_log",
+          "public.reactions",
+          "public.comments",
+          "public.posts",
+          "public.sessions",
+          "public.messages",
+          "analytics.visits",
+        ].sort(),
+      );
+    });
+
+    test("classifies native vs pgslice tables correctly", async ({
+      pgslice,
+      transaction,
+    }) => {
+      const results = await pgslice.maintain(transaction, {
+        futureDaily: 3,
+        futureWeekly: 3,
+        futureMonthly: 3,
+        futureYearly: 3,
+      });
+      const model = new Map(results.map((r) => [r.table, r.model]));
+
+      for (const native of [
+        "public.posts",
+        "public.comments",
+        "public.reactions",
+        "public.daily_reports",
+        "analytics.visits",
+        "public.audit_log",
+        "public.outbox",
+      ]) {
+        expect(model.get(native)).toBe("native");
+      }
+      for (const classic of [
+        "public.sessions",
+        "public.tokens",
+        "public.messages",
+      ]) {
+        expect(model.get(classic)).toBe("pgslice");
+      }
+    });
+
+    test("extends every table by bounds with the expected pgslice-named partitions", async ({
+      pgslice,
+      transaction,
+    }) => {
+      const results = await pgslice.maintain(transaction, {
+        futureDaily: 3,
+        futureWeekly: 3,
+        futureMonthly: 3,
+        futureYearly: 3,
+      });
+      const created = new Map(
+        results.map((r) => [r.table, [...r.partitionsCreated].sort()]),
+      );
+
+      expect(created.get("public.posts")).toEqual([
+        "posts_202602",
+        "posts_202603",
+        "posts_202604",
+      ]);
+      expect(created.get("public.daily_reports")).toEqual([
+        "daily_reports_202602",
+        "daily_reports_202603",
+        "daily_reports_202604",
+      ]);
+      expect(created.get("analytics.visits")).toEqual([
+        "visits_2026w04",
+        "visits_2026w05",
+        "visits_2026w06",
+      ]);
+      expect(created.get("public.audit_log")).toEqual([
+        "audit_log_2026w03",
+        "audit_log_2026w04",
+        "audit_log_2026w05",
+        "audit_log_2026w06",
+      ]);
+      expect(created.get("public.outbox")).toEqual([
+        "outbox_2026w04",
+        "outbox_2026w05",
+        "outbox_2026w06",
+      ]);
+      expect(created.get("public.sessions")).toEqual([
+        "sessions_2026w04",
+        "sessions_2026w05",
+        "sessions_2026w06",
+      ]);
+
+      // Legacy-named and special partitions are left in place.
+      expect(await childNames(transaction, "outbox")).toContain(
+        "outbox_historic",
+      );
+      expect(await childNames(transaction, "analytics.visits")).toContain(
+        "visits_default",
+      );
+      expect(await childNames(transaction, "public.posts")).toContain(
+        "posts_y2026m01",
+      );
+    });
+
+    test("every managed table ends up CDC-ready", async ({
+      pgslice,
+      transaction,
+    }) => {
+      const results = await pgslice.maintain(transaction, {
+        futureDaily: 3,
+        futureWeekly: 3,
+        futureMonthly: 3,
+        futureYearly: 3,
+      });
+      for (const result of results) {
+        expect(result.replicaIdentityReady).toBe(true);
+        expect(result.unsafePartitions).toEqual([]);
+      }
+    });
+
+    test("is idempotent: a second run creates nothing", async ({
+      pgslice,
+      transaction,
+    }) => {
+      await pgslice.maintain(transaction, {
+        futureDaily: 3,
+        futureWeekly: 3,
+        futureMonthly: 3,
+        futureYearly: 3,
+      });
+      const second = await pgslice.maintain(transaction, {
+        futureDaily: 3,
+        futureWeekly: 3,
+        futureMonthly: 3,
+        futureYearly: 3,
+      });
+      for (const result of second) {
+        expect(result.partitionsCreated).toEqual([]);
+      }
+    });
+
+    test("can be restricted to a single schema", async ({
+      pgslice,
+      transaction,
+    }) => {
+      const results = await pgslice.maintain(transaction, {
+        futureDaily: 3,
+        futureWeekly: 3,
+        futureMonthly: 3,
+        futureYearly: 3,
+        schema: "analytics",
+      });
+      expect(results.map((r) => r.table)).toEqual(["analytics.visits"]);
+    });
+  });
+
+  describe("native tables", () => {
+    test("inherit the parent composite PK and the parent's grants on new partitions", async ({
+      pgslice,
+      transaction,
+    }) => {
+      await createCdcRole(transaction, "replica_reader_maint");
+      await nativeParent(
+        transaction,
+        "posts",
+        "created_at",
+        TS,
+        "month",
+        "date",
+      );
+      await addChild(
+        transaction,
+        "posts",
+        "posts_y2026m01",
+        "2026-01-01",
+        "2026-02-01",
+      );
+      await transaction.query(
+        sql.unsafe`GRANT SELECT ON TABLE posts TO replica_reader_maint`,
+      );
+
+      await pgslice.maintain(transaction, {
+        futureDaily: 1,
+        futureWeekly: 1,
+        futureMonthly: 1,
+        futureYearly: 1,
+      });
+
+      const pkDefs = await transaction.any(
+        sql.type(z.object({ def: z.string() }))`
+          SELECT pg_get_constraintdef(oid) AS def FROM pg_constraint
+          WHERE conrelid = 'public.posts_202602'::regclass AND contype = 'p'
+        `,
+      );
+      expect(pkDefs.map((c) => c.def)).toEqual([
+        "PRIMARY KEY (id, created_at)",
+      ]);
+
+      const grants = await transaction.any(
+        sql.type(z.object({ grantee: z.string(), privilege: z.string() }))`
+          SELECT grantee_role.rolname AS grantee, acl.privilege_type AS privilege
+          FROM pg_class c
+            CROSS JOIN LATERAL aclexplode(c.relacl) AS acl
+            JOIN pg_roles grantee_role ON grantee_role.oid = acl.grantee
+          WHERE c.oid = 'public.posts_202602'::regclass
+            AND grantee_role.rolname = 'replica_reader_maint'
+        `,
+      );
+      expect(grants.map((g) => g.privilege)).toContain("SELECT");
+    });
+  });
+
+  describe("classic pgslice tables", () => {
+    test("get a per-partition composite primary key", async ({
+      pgslice,
+      transaction,
+    }) => {
+      await pgsliceParent(
+        transaction,
+        "sessions",
+        "created_at",
+        TSTZ,
+        "week",
+        "timestamptz",
+      );
+      await addChild(
+        transaction,
+        "sessions",
+        "sessions_2026w03",
+        "2026-01-12 00:00:00+00",
+        "2026-01-19 00:00:00+00",
+        ["id", "created_at"],
+      );
+
+      const results = await pgslice.maintain(transaction, {
+        futureDaily: 1,
+        futureWeekly: 1,
+        futureMonthly: 1,
+        futureYearly: 1,
+      });
+      expect(results[0].model).toBe("pgslice");
+
+      const pkDefs = await transaction.any(
+        sql.type(z.object({ def: z.string() }))`
+          SELECT pg_get_constraintdef(oid) AS def FROM pg_constraint
+          WHERE conrelid = 'public.sessions_2026w04'::regclass AND contype = 'p'
+        `,
+      );
+      expect(pkDefs.map((c) => c.def)).toEqual([
+        "PRIMARY KEY (id, created_at)",
+      ]);
+    });
+  });
+
+  describe("replica identity", () => {
+    test("a parent REPLICA IDENTITY USING INDEX propagates its partitioned index to new partitions automatically", async ({
+      pgslice,
+      transaction,
+    }) => {
+      await nativeParent(
+        transaction,
+        "analytics.visits",
+        "created_at",
+        TSTZ,
+        "week",
+        "timestamptz",
+      );
+      await addChild(
+        transaction,
+        "analytics.visits",
+        "visits_2026w03",
+        "2026-01-12 00:00:00+00",
+        "2026-01-19 00:00:00+00",
+      );
+      await addDefault(transaction, "analytics.visits", "visits_default");
+      // Build the partitioned replica-identity index exactly as the runner does:
+      // ON ONLY the parent, then attach a child index for every existing leaf,
+      // then point REPLICA IDENTITY at it.
+      await transaction.query(sql.unsafe`
+        CREATE UNIQUE INDEX visits_ri_uidx
+        ON ONLY analytics.visits (id, created_at)
+      `);
+      await transaction.query(sql.unsafe`
+        CREATE UNIQUE INDEX visits_2026w03_ri
+        ON analytics.visits_2026w03 (id, created_at)
+      `);
+      await transaction.query(sql.unsafe`
+        ALTER INDEX analytics.visits_ri_uidx
+        ATTACH PARTITION analytics.visits_2026w03_ri
+      `);
+      await transaction.query(sql.unsafe`
+        CREATE UNIQUE INDEX visits_default_ri
+        ON analytics.visits_default (id, created_at)
+      `);
+      await transaction.query(sql.unsafe`
+        ALTER INDEX analytics.visits_ri_uidx
+        ATTACH PARTITION analytics.visits_default_ri
+      `);
+      await transaction.query(sql.unsafe`
+        ALTER TABLE analytics.visits
+        REPLICA IDENTITY USING INDEX visits_ri_uidx
+      `);
+
+      const results = await pgslice.maintain(transaction, {
+        futureDaily: 1,
+        futureWeekly: 1,
+        futureMonthly: 1,
+        futureYearly: 1,
+      });
+      expect(results[0].replicaIdentityReady).toBe(true);
+
+      // The new partition auto-got a unique (id, created_at) index attached to
+      // the parent's partitioned replica-identity index, and it is valid. (The
+      // new partition also has its own inherited composite PK index — that is a
+      // second unique index, which is why we filter to the one attached to the
+      // replica-identity index here.)
+      const idx = await transaction.any(
+        sql.type(z.object({ valid: z.boolean() }))`
+          SELECT ix.indisvalid AS valid
+          FROM pg_inherits pi
+            JOIN pg_class parent_idx ON parent_idx.oid = pi.inhparent
+            JOIN pg_class ic ON ic.oid = pi.inhrelid
+            JOIN pg_index ix ON ix.indexrelid = ic.oid
+            JOIN pg_class t ON t.oid = ix.indrelid
+            JOIN pg_namespace n ON n.oid = t.relnamespace
+          WHERE parent_idx.relname = 'visits_ri_uidx'
+            AND n.nspname = 'analytics'
+            AND t.relname = 'visits_2026w04'
+        `,
+      );
+      expect(idx).toHaveLength(1);
+      expect(idx[0].valid).toBe(true);
+
+      // The parent's replica-identity index stays valid after extension.
+      const parentIdx = await transaction.one(
+        sql.type(z.object({ valid: z.boolean() }))`
+          SELECT ix.indisvalid AS valid FROM pg_index ix
+            JOIN pg_class ic ON ic.oid = ix.indexrelid
+          WHERE ic.relname = 'visits_ri_uidx'
+        `,
+      );
+      expect(parentIdx.valid).toBe(true);
+    });
+
+    test("flags a pre-existing partition that has no usable replica identity", async ({
+      pgslice,
+      transaction,
+    }) => {
+      await nativeParent(
+        transaction,
+        "weird",
+        "created_at",
+        TS,
+        "month",
+        "date",
+      );
+      await addChild(
+        transaction,
+        "weird",
+        "weird_2026m01",
+        "2026-01-01",
+        "2026-02-01",
+      );
+      // Simulate a leaf created outside pgslice that is not CDC-safe.
+      await transaction.query(sql.unsafe`
+        ALTER TABLE weird_2026m01 REPLICA IDENTITY NOTHING
+      `);
+
+      const results = await pgslice.maintain(transaction, {
+        futureDaily: 1,
+        futureWeekly: 1,
+        futureMonthly: 1,
+        futureYearly: 1,
+      });
+      const weird = results.find((r) => r.table === "public.weird");
+
+      expect(weird?.replicaIdentityReady).toBe(false);
+      expect(weird?.unsafePartitions).toEqual(["weird_2026m01"]);
+    });
+  });
+
+  describe("failure isolation", () => {
+    test("a non-empty DEFAULT blocking the next partition fails that table alone; the rest of the fleet still extends", async ({
+      pgslice,
+      transaction,
+    }) => {
+      // `blocked` has a DEFAULT holding a row that belongs in the next range,
+      // so creating that partition will fail (the production drain footgun).
+      await nativeParent(
+        transaction,
+        "blocked",
+        "created_at",
+        TS,
+        "month",
+        "date",
+      );
+      await addChild(
+        transaction,
+        "blocked",
+        "blocked_y2026m01",
+        "2026-01-01",
+        "2026-02-01",
+      );
+      await addDefault(transaction, "blocked", "blocked_default");
+      await transaction.query(sql.unsafe`
+        INSERT INTO blocked (id, created_at, payload) VALUES (1, '2026-02-15', 'absorbed')
+      `);
+      // `healthy` is an ordinary native monthly table.
+      await nativeParent(
+        transaction,
+        "healthy",
+        "created_at",
+        TS,
+        "month",
+        "date",
+      );
+      await addChild(
+        transaction,
+        "healthy",
+        "healthy_y2026m01",
+        "2026-01-01",
+        "2026-02-01",
+      );
+
+      const results = await pgslice.maintain(transaction, {
+        futureDaily: 1,
+        futureWeekly: 1,
+        futureMonthly: 1,
+        futureYearly: 1,
+      });
+      const blocked = results.find((r) => r.table === "public.blocked");
+      const healthy = results.find((r) => r.table === "public.healthy");
+
+      // The blocked table's CREATE is rejected (slonik surfaces it as a check
+      // integrity constraint violation); it is recorded as failed, not thrown.
+      expect(blocked?.error).toBeTruthy();
+      expect(blocked?.partitionsCreated).toEqual([]);
+      // The healthy table is extended despite the other table's failure.
+      expect(healthy?.error).toBeNull();
+      expect(healthy?.partitionsCreated).toEqual(["healthy_202602"]);
+    });
+
+    test("emits an error record carrying the failure message and host", async ({
+      pgslice,
+      transaction,
+    }) => {
+      await nativeParent(
+        transaction,
+        "blocked",
+        "created_at",
+        TS,
+        "month",
+        "date",
+      );
+      await addChild(
+        transaction,
+        "blocked",
+        "blocked_y2026m01",
+        "2026-01-01",
+        "2026-02-01",
+      );
+      await addDefault(transaction, "blocked", "blocked_default");
+      await transaction.query(sql.unsafe`
+        INSERT INTO blocked (id, created_at, payload) VALUES (1, '2026-02-15', 'absorbed')
+      `);
+
+      const lines: string[] = [];
+      const results = await pgslice.maintain(
+        transaction,
+        { futureMonthly: 1, host: "clone.example", jobId: "job-under-test" },
+        (entry) => {
+          lines.push(JSON.stringify(entry));
+        },
+      );
+      const records = lines.map((line): MaintainLogRecord => JSON.parse(line));
+
+      const blocked = results.find((r) => r.table === "public.blocked");
+      const record = records.find((r) => r.target.table === "blocked");
+      // The caught error is surfaced verbatim as the record's msg.
+      expect(record?.msg).toBe(blocked?.error);
+      expect(record?.level).toBe("error");
+      expect(record?.success).toBe(0);
+      expect(record?.target.host).toBe("clone.example");
+      // The run's jobId is stamped on every record.
+      expect(records.every((r) => r.jobId === "job-under-test")).toBe(true);
+    });
+  });
+
+  describe("per-period horizons", () => {
+    test("extends each table by the horizon for its own period", async ({
+      pgslice,
+      transaction,
+    }) => {
+      // One run over a monthly and a weekly table with different per-period
+      // futures. The weekly table (shorter period + larger future) must get
+      // strictly more new partitions than the monthly one — proving the horizon
+      // is applied per period, not uniformly.
+      await nativeParent(
+        transaction,
+        "m_tbl",
+        "created_at",
+        TS,
+        "month",
+        "date",
+      );
+      await addChild(
+        transaction,
+        "m_tbl",
+        "m_tbl_y2026m01",
+        "2026-01-01",
+        "2026-02-01",
+      );
+      await nativeParent(
+        transaction,
+        "w_tbl",
+        "created_at",
+        TS,
+        "week",
+        "date",
+      );
+      await addChild(
+        transaction,
+        "w_tbl",
+        "w_tbl_2026w02",
+        "2026-01-05",
+        "2026-01-12",
+      );
+
+      const results = await pgslice.maintain(transaction, {
+        futureMonthly: 2,
+        futureWeekly: 6,
+      });
+      const m = results.find((r) => r.table === "public.m_tbl");
+      const w = results.find((r) => r.table === "public.w_tbl");
+
+      expect(m?.error).toBeNull();
+      expect(w?.error).toBeNull();
+      const mCount = m?.partitionsCreated.length ?? 0;
+      const wCount = w?.partitionsCreated.length ?? 0;
+      expect(mCount).toBeGreaterThan(0);
+      expect(wCount).toBeGreaterThan(mCount);
+    });
+  });
+});
diff --git a/src/pgslice.ts b/src/pgslice.ts
index 0006731..c363aab 100644
--- a/src/pgslice.ts
+++ b/src/pgslice.ts
@@ -7,6 +7,9 @@ import {
 } from "slonik";
 import { createQueryLoggingInterceptor } from "slonik-interceptor-query-logging";
 
+import { z } from "zod";
+import { randomUUID } from "node:crypto";
+
 import type {
   AddPartitionsOptions,
   AnalyzeOptions,
@@ -15,6 +18,10 @@ import type {
   EnableMirroringOptions,
   FillBatchResult,
   FillOptions,
+  MaintainLog,
+  MaintainOptions,
+  MaintainResult,
+  PartitionModel,
   Period,
   PrepOptions,
   StatusOptions,
@@ -496,6 +503,225 @@ export class Pgslice {
     );
   }
 
+  /**
+   * Maintains every managed partitioned table the connection can see: each
+   * partitioned parent (`relkind = 'p'`) carrying a valid pgslice settings
+   * comment is discovered from the catalog and extended via {@link addPartitions}.
+   * New partitioned tables are therefore picked up automatically with no
+   * per-table configuration.
+   *
+   * The native-vs-pgslice distinction is handled by {@link addPartitions}: it
+   * skips the per-partition primary key when the parent owns one (the inherited
+   * composite key and any partitioned indexes propagate automatically) and adds
+   * a per-partition key otherwise. New partitions are created with the default
+   * replica identity, so each leaf's row identity is its own (or inherited)
+   * primary key and no replica-identity DDL is required. After extending each
+   * table this verifies every leaf has a usable replica identity, surfacing a
+   * misconfigured table rather than silently shipping a CDC-unsafe partition.
+   *
+   * Each table is maintained independently: a failure on one table (for example
+   * a non-empty DEFAULT partition that blocks creating the next one) is recorded
+   * on that table's result and does not stop the rest of the fleet.
+   *
+   * The optional `log` sink receives structured records — one on start, one per
+   * table as it is extended, and one final summary — for the command to emit as
+   * JSONL. It defaults to a no-op, so callers that only need the results (such
+   * as tests) can ignore it.
+   */
+  async maintain(
+    connection: DatabasePoolConnection,
+    options: MaintainOptions,
+    log: MaintainLog = () => {},
+  ): Promise<MaintainResult[]> {
+    const past = options.past ?? 0;
+    // Future runway is set per period: "N periods" is very different runway for a
+    // weekly vs a monthly vs a yearly table, so each table uses the horizon for
+    // its own period, giving the fleet comparable forward coverage.
+    const futureByPeriod: Record<Period, number> = {
+      day: options.futureDaily ?? 90,
+      week: options.futureWeekly ?? 26,
+      month: options.futureMonthly ?? 6,
+      year: options.futureYearly ?? 1,
+    };
+    // Capture one reference instant for the whole fleet, so a run that straddles
+    // a period boundary uses a consistent horizon for every table.
+    const now = options.now ?? new Date();
+    // One correlation id per run, stamped on every record so all logs from this
+    // invocation can be grouped.
+    const jobId = options.jobId ?? randomUUID();
+
+    const { db } = await connection.one(
+      sql.type(z.object({ db: z.string() }))`SELECT current_database() AS db`,
+    );
+    // Every record carries the endpoint host (from the connection URL) and the
+    // database name, so a run identifies which host and DB it extended.
+    const target = { host: options.host, db };
+
+    log({
+      jobId,
+      msg: "Running pgslice maintain",
+      level: "info",
+      target,
+      future: {
+        daily: futureByPeriod.day,
+        weekly: futureByPeriod.week,
+        monthly: futureByPeriod.month,
+        yearly: futureByPeriod.year,
+      },
+    });
+
+    const managed = await this.#discoverManagedTables(
+      connection,
+      options.schema,
+    );
+
+    const results: MaintainResult[] = [];
+    const succeeded: string[] = [];
+    const failed: string[] = [];
+
+    for (const { table, period } of managed) {
+      let result: MaintainResult;
+      try {
+        const partitionsCreated = await this.addPartitions(connection, {
+          table: table.toString(),
+          past,
+          future: futureByPeriod[period],
+          tablespace: options.tablespace,
+          inheritGrants: options.inheritGrants,
+          now,
+        });
+
+        const { model, partitionCount, unsafePartitions } =
+          await connection.transaction(async (tx) => {
+            const parentPrimaryKey = await table.primaryKeyColumns(tx);
+            const partitions = await table.partitions(tx);
+            const unsafe = await table.unsafeReplicaIdentityPartitions(tx);
+            const model: PartitionModel =
+              parentPrimaryKey.length > 0 ? "native" : "pgslice";
+            return {
+              model,
+              partitionCount: partitions.length,
+              unsafePartitions: unsafe,
+            };
+          });
+
+        result = {
+          table: table.toString(),
+          model,
+          partitionsCreated,
+          partitionCount,
+          replicaIdentityReady: unsafePartitions.length === 0,
+          unsafePartitions,
+          error: null,
+        };
+      } catch (error) {
+        result = {
+          table: table.toString(),
+          model: null,
+          partitionsCreated: [],
+          partitionCount: 0,
+          replicaIdentityReady: false,
+          unsafePartitions: [],
+          error: error instanceof Error ? error.message : String(error),
+        };
+      }
+
+      results.push(result);
+
+      // A table is a success only when it extended without error and every leaf
+      // is CDC-safe; a created-but-CDC-unsafe table is a failure the run reports.
+      const ok = result.error === null && result.replicaIdentityReady;
+      (ok ? succeeded : failed).push(result.table);
+
+      let msg: string;
+      if (result.error !== null) {
+        msg = result.error;
+      } else if (!result.replicaIdentityReady) {
+        msg = `Partitions without a usable replica identity (CDC-unsafe): ${result.unsafePartitions.join(", ")}`;
+      } else if (result.partitionsCreated.length === 0) {
+        msg = "Table already up to date; no extension needed";
+      } else {
+        msg = "Extended table successfully";
+      }
+
+      log({
+        jobId,
+        msg,
+        level: ok ? "info" : "error",
+        target: { ...target, schema: table.schema, table: table.name },
+        partitions: {
+          new: result.partitionsCreated.length,
+          total: result.partitionCount,
+        },
+        success: ok ? 1 : 0,
+      });
+    }
+
+    log({
+      jobId,
+      msg:
+        failed.length === 0
+          ? "Finished pgslice maintain successfully"
+          : "Finished pgslice maintain with errors",
+      level: failed.length === 0 ? "info" : "error",
+      target,
+      succeeded: { count: succeeded.length, tables: succeeded },
+      failed: { count: failed.length, tables: failed },
+    });
+
+    return results;
+  }
+
+  /**
+   * Discovers the managed partitioned parents visible to this connection: a
+   * top-level partitioned table (`relkind = 'p'`, not itself a partition) whose
+   * table comment parses as pgslice settings. Optionally restricted to a single
+   * schema. The same `pg_partitioned_table`-style catalog scan the runway
+   * monitor uses, so scheduler and monitor share one source of truth.
+   */
+  async #discoverManagedTables(
+    connection: DatabasePoolConnection,
+    schema?: string,
+  ): Promise<Array<{ table: Table; period: Period }>> {
+    const schemaFilter = schema
+      ? sql.fragment`AND n.nspname = ${schema}`
+      : sql.fragment``;
+
+    const rows = await connection.any(
+      sql.type(
+        z.object({
+          schema: z.string(),
+          name: z.string(),
+          comment: z.string().nullable(),
+        }),
+      )`
+        SELECT
+          n.nspname AS schema,
+          c.relname AS name,
+          obj_description(c.oid, 'pg_class') AS comment
+        FROM pg_class c
+          JOIN pg_namespace n ON n.oid = c.relnamespace
+        WHERE c.relkind = 'p'
+          AND n.nspname NOT IN ('pg_catalog', 'information_schema')
+          AND NOT EXISTS (
+            SELECT 1 FROM pg_inherits inh WHERE inh.inhrelid = c.oid
+          )
+          ${schemaFilter}
+        ORDER BY n.nspname, c.relname
+      `,
+    );
+
+    return rows.flatMap((row) => {
+      if (!row.comment) {
+        return [];
+      }
+      const settings = TableSettings.parseFromComment(row.comment);
+      return settings
+        ? [{ table: new Table(row.schema, row.name), period: settings.period }]
+        : [];
+    });
+  }
+
   /**
    * Resolves the primary key columns to place on each new partition in the
    * classic pgslice model (where the parent has no primary key of its own).
diff --git a/src/table.test.ts b/src/table.test.ts
index 36cd7b6..40414a2 100644
--- a/src/table.test.ts
+++ b/src/table.test.ts
@@ -738,3 +738,44 @@ describe("Table.rangePartitionBounds", () => {
     expect(finiteRange?.upper).toEqual(new Date("2026-01-19T00:00:00Z"));
   });
 });
+
+describe("Table.unsafeReplicaIdentityPartitions", () => {
+  test("flags an unsafe leaf beneath a sub-partitioned child", async ({
+    transaction,
+  }) => {
+    await transaction.query(sql.unsafe`
+      CREATE TABLE nested_parent (
+        id bigint NOT NULL,
+        created_at timestamp NOT NULL,
+        PRIMARY KEY (id, created_at)
+      ) PARTITION BY RANGE (created_at)
+    `);
+    // A direct child that is itself partitioned (multi-level partitioning).
+    await transaction.query(sql.unsafe`
+      CREATE TABLE nested_parent_2026 PARTITION OF nested_parent
+        FOR VALUES FROM ('2026-01-01') TO ('2027-01-01')
+        PARTITION BY RANGE (created_at)
+    `);
+    await transaction.query(sql.unsafe`
+      CREATE TABLE nested_parent_2026m01 PARTITION OF nested_parent_2026
+        FOR VALUES FROM ('2026-01-01') TO ('2026-02-01')
+    `);
+    await transaction.query(sql.unsafe`
+      CREATE TABLE nested_parent_2026m02 PARTITION OF nested_parent_2026
+        FOR VALUES FROM ('2026-02-01') TO ('2026-03-01')
+    `);
+    await transaction.query(sql.unsafe`
+      ALTER TABLE nested_parent_2026m02 REPLICA IDENTITY NOTHING
+    `);
+
+    const unsafe =
+      await Table.parse("nested_parent").unsafeReplicaIdentityPartitions(
+        transaction,
+      );
+
+    // Only true leaves are inspected. The unsafe nested leaf is caught; a
+    // direct-children-only check would see the (safe) sub-partitioned child and
+    // miss it.
+    expect(unsafe).toEqual(["nested_parent_2026m02"]);
+  });
+});
diff --git a/src/table.ts b/src/table.ts
index 6999647..5f16d3b 100644
--- a/src/table.ts
+++ b/src/table.ts
@@ -497,6 +497,43 @@ export class Table {
     });
   }
 
+  /**
+   * Returns the leaf partitions of this table that lack a replica identity
+   * usable for logical replication. A partition is unsafe when it has
+   * `REPLICA IDENTITY NOTHING`, or the default identity with no primary key —
+   * in either case Postgres rejects `UPDATE`/`DELETE` on a published table.
+   * Partitions covered by a primary key (the native parent-owned key or a
+   * per-partition key), or an explicit FULL / USING INDEX identity, are safe.
+   * Read-only: it only inspects the catalog and never emits replica-identity
+   * DDL. Walks the whole partition tree, so leaf partitions beneath a
+   * sub-partitioned child are inspected too — not just direct children.
+   */
+  async unsafeReplicaIdentityPartitions(
+    tx: CommonQueryMethods,
+  ): Promise<string[]> {
+    const rows = await tx.any(
+      sql.type(z.object({ name: z.string() }))`
+        SELECT leaf.relname AS name
+        FROM pg_partition_tree(${this.regclassLiteral}::regclass) tree
+          JOIN pg_class leaf ON leaf.oid = tree.relid
+        WHERE tree.isleaf
+          AND (
+            leaf.relreplident = 'n'
+            OR (
+              leaf.relreplident = 'd'
+              AND NOT EXISTS (
+                SELECT 1 FROM pg_index pk
+                WHERE pk.indrelid = leaf.oid AND pk.indisprimary
+              )
+            )
+          )
+        ORDER BY leaf.relname ASC
+      `,
+    );
+
+    return rows.map((r) => r.name);
+  }
+
   /**
    * Fetches the partition settings from this table's comment.
    */
diff --git a/src/types.ts b/src/types.ts
index e50f451..8872461 100644
--- a/src/types.ts
+++ b/src/types.ts
@@ -225,3 +225,92 @@ export interface TableStatus {
   retiredMirrorTriggerExists: boolean;
   originalIsPartitioned: boolean;
 }
+
+/**
+ * Options for the `maintain` command.
+ */
+export interface MaintainOptions {
+  past?: number;
+  /**
+   * Future partitions to keep ahead, set per period. "N periods" is very
+   * different runway for a weekly vs a monthly vs a yearly table, so the horizon
+   * is per period and every table gets comparable forward coverage; each
+   * discovered table uses the value for its own period. Defaults when omitted:
+   * daily 90, weekly 26, monthly 6, yearly 1.
+   */
+  futureDaily?: number;
+  futureWeekly?: number;
+  futureMonthly?: number;
+  futureYearly?: number;
+  /**
+   * Host of the endpoint being maintained, recorded on every log record's
+   * `target` alongside the database name so operators can see which host was
+   * extended. The command derives it from the connection URL (host only, no
+   * credentials); when omitted, `target.host` is simply absent.
+   */
+  host?: string;
+  /**
+   * Restrict maintenance to partitioned tables in this schema. When omitted,
+   * every managed partitioned table the connection can see is maintained.
+   */
+  schema?: string;
+  tablespace?: string;
+  inheritGrants?: boolean;
+  /**
+   * The reference "now" for choosing each table's partition horizon. Defaults
+   * to a single instant captured when the run starts, so a fleet run that
+   * straddles a period boundary uses one consistent horizon for every table.
+   */
+  now?: Date;
+  /**
+   * Correlation id stamped on every log record for this run, so all records
+   * from one invocation can be grouped. A fresh id is generated per run when
+   * omitted.
+   */
+  jobId?: string;
+}
+
+/**
+ * Sink for maintain's structured JSONL logs. Each call receives one log record;
+ * the `maintain` command serializes it to stdout as one JSON object per line so
+ * downstream log tooling can lift the keys into attributes. Only `info` and
+ * `error` records are emitted — there is no debug/warning level.
+ */
+export type MaintainLog = (entry: Record<string, unknown>) => void;
+
+/**
+ * How a managed table's primary key is owned, which decides how
+ * `add_partitions` treats each new partition.
+ * - "native": the partitioned parent owns the (often composite) primary key, so
+ *   Postgres propagates it — and any partitioned indexes — to each new partition.
+ * - "pgslice": the classic pgslice model where the parent has no primary key and
+ *   each partition owns its own.
+ */
+export type PartitionModel = "native" | "pgslice";
+
+/**
+ * Outcome of maintaining a single managed table.
+ */
+export interface MaintainResult {
+  table: string;
+  /** The partitioning model, or null if maintenance failed before it was determined. */
+  model: PartitionModel | null;
+  partitionsCreated: string[];
+  partitionCount: number;
+  /**
+   * Whether every leaf partition has a replica identity usable for logical
+   * replication. New partitions are created with the default replica identity,
+   * so each leaf's row identity is its own (or inherited) primary key and no
+   * replica-identity DDL is required; this read-only flag surfaces a leaf
+   * lacking a usable identity rather than shipping a CDC-unsafe partition.
+   */
+  replicaIdentityReady: boolean;
+  /** Leaf partitions that would not be CDC-safe (no usable replica identity). */
+  unsafePartitions: string[];
+  /**
+   * Error message if maintaining this table failed; null on success. One
+   * table's failure (e.g. a non-empty DEFAULT blocking the next partition) is
+   * recorded here and does not stop the rest of the fleet from being extended.
+   */
+  error: string | null;
+}