diff --git a/packages/zero-cache/src/config/normalize.test.ts b/packages/zero-cache/src/config/normalize.test.ts
index 8f6d34d5cd..debcfe6b39 100644
--- a/packages/zero-cache/src/config/normalize.test.ts
+++ b/packages/zero-cache/src/config/normalize.test.ts
@@ -10,6 +10,7 @@ function configWith(litestream: Partial<ZeroConfig['litestream']>): ZeroConfig {
     changeStreamer: {
       port: 4849,
       address: 'localhost',
+      pgChangeLogEnabled: true,
       sqliteChangeLogMode: 'off',
       sqliteChangeLogReadPercent: 0,
       sqliteChangeLogColdReadPercent: 0,
@@ -128,6 +129,46 @@ describe('config/normalize litestream v5 gating', () => {
 });
 
 describe('config/normalize SQLite change log', () => {
+  test('PG change log is enabled by default configuration', () => {
+    const config = configWith({});
+
+    expect(config.changeStreamer.pgChangeLogEnabled).toBe(true);
+    expect(() => assertNormalized(config)).not.toThrow();
+  });
+
+  test('disabling the PG change log requires authoritative SQLite and v5 backup settings', () => {
+    const config = configWith({});
+    config.changeStreamer.pgChangeLogEnabled = false;
+
+    expect(() => assertNormalized(config)).toThrow(
+      'requires --change-streamer-sqlite-change-log-mode=serve',
+    );
+
+    config.changeStreamer.sqliteChangeLogMode = 'serve';
+    expect(() => assertNormalized(config)).toThrow(
+      'requires --change-streamer-sqlite-change-log-read-percent=100',
+    );
+
+    config.changeStreamer.sqliteChangeLogReadPercent = 100;
+    expect(() => assertNormalized(config)).toThrow(
+      'requires --change-streamer-sqlite-change-log-cold-read-percent=100',
+    );
+
+    config.changeStreamer.sqliteChangeLogColdReadPercent = 100;
+    expect(() => assertNormalized(config)).toThrow(
+      'requires a litestream v5 backup',
+    );
+
+    Object.assign(config.litestream, {
+      backupURL: 's3://bucket/replica',
+      backupUsingV5: true,
+      restoreUsingV5: true,
+      executableV5: '/bin/litestream-v5',
+      vfsQueryExecutable: '/bin/vfs-query',
+    });
+    expect(() => assertNormalized(config)).not.toThrow();
+  });
+
   test('read percentage is only allowed in serve mode', () => {
     const config = configWith({});
     config.changeStreamer.sqliteChangeLogMode = 'compare';
diff --git a/packages/zero-cache/src/config/normalize.ts b/packages/zero-cache/src/config/normalize.ts
index 81d1396244..ed98a2bba7 100644
--- a/packages/zero-cache/src/config/normalize.ts
+++ b/packages/zero-cache/src/config/normalize.ts
@@ -53,6 +53,7 @@ export function assertNormalized(
   assert(config.changeStreamer.port, 'missing --change-streamer-port');
   assert(config.changeStreamer.address, 'missing --change-streamer-address');
   const {
+    pgChangeLogEnabled,
     sqliteChangeLogMode,
     sqliteChangeLogReadPercent,
     sqliteChangeLogColdReadPercent,
@@ -97,6 +98,24 @@ export function assertNormalized(
     sqliteChangeLogReadPercent > 0 || sqliteChangeLogColdReadPercent === 0,
     '--change-streamer-sqlite-change-log-cold-read-percent must be 0 when --change-streamer-sqlite-change-log-read-percent is 0',
   );
+  if (!pgChangeLogEnabled) {
+    assert(
+      sqliteChangeLogMode === 'serve',
+      '--change-streamer-pg-change-log-enabled=false requires --change-streamer-sqlite-change-log-mode=serve',
+    );
+    assert(
+      sqliteChangeLogReadPercent === 100,
+      '--change-streamer-pg-change-log-enabled=false requires --change-streamer-sqlite-change-log-read-percent=100',
+    );
+    assert(
+      sqliteChangeLogColdReadPercent === 100,
+      '--change-streamer-pg-change-log-enabled=false requires --change-streamer-sqlite-change-log-cold-read-percent=100',
+    );
+    assert(
+      config.litestream.backupURL && config.litestream.backupUsingV5,
+      '--change-streamer-pg-change-log-enabled=false requires a litestream v5 backup',
+    );
+  }
   for (const [flag, value] of [
     ['retention-ms', sqliteChangeLogRetentionMs],
     ['read-batch-rows', sqliteChangeLogReadBatchRows],
diff --git a/packages/zero-cache/src/config/zero-config.test.ts b/packages/zero-cache/src/config/zero-config.test.ts
index e6f8e90a25..993c0cc429 100644
--- a/packages/zero-cache/src/config/zero-config.test.ts
+++ b/packages/zero-cache/src/config/zero-config.test.ts
@@ -905,6 +905,24 @@ test('--enable-query-covering can be disabled', () => {
   expect(config.enableQueryCovering).toBe(false);
 });
 
+test('PG change log is enabled by default and can be disabled by env', () => {
+  const defaults = parseOptionsAdvanced(zeroOptions, {
+    envNamePrefix: 'ZERO_',
+    allowUnknown: false,
+    allowPartial: true,
+    env: {},
+  }).config;
+  const disabled = parseOptionsAdvanced(zeroOptions, {
+    envNamePrefix: 'ZERO_',
+    allowUnknown: false,
+    allowPartial: true,
+    env: {ZERO_CHANGE_STREAMER_PG_CHANGE_LOG_ENABLED: 'false'},
+  }).config;
+
+  expect(defaults.changeStreamer.pgChangeLogEnabled).toBe(true);
+  expect(disabled.changeStreamer.pgChangeLogEnabled).toBe(false);
+});
+
 test('legacy queries are disabled by default', () => {
   const {config} = parseOptionsAdvanced(zeroOptions, {
     envNamePrefix: 'ZERO_',
diff --git a/packages/zero-cache/src/config/zero-config.ts b/packages/zero-cache/src/config/zero-config.ts
index 3dfe4f58a0..fe523e5856 100644
--- a/packages/zero-cache/src/config/zero-config.ts
+++ b/packages/zero-cache/src/config/zero-config.ts
@@ -737,6 +737,16 @@ export const zeroOptions = {
       hidden: true,
     },
 
+    pgChangeLogEnabled: {
+      type: v.boolean().default(true),
+      desc: [
+        `Whether the legacy Postgres change log remains authoritative for`,
+        `stream initialization, persistence, catchup, and upstream ACKs.`,
+        `Disabling it requires SQLite serve mode at 100 percent and a v5 backup.`,
+      ],
+      hidden: true,
+    },
+
     sqliteChangeLogReadPercent: {
       type: v.number().default(0),
       desc: [
diff --git a/packages/zero-cache/src/server/change-streamer.ts b/packages/zero-cache/src/server/change-streamer.ts
index b3587e30b1..7eda4881d3 100644
--- a/packages/zero-cache/src/server/change-streamer.ts
+++ b/packages/zero-cache/src/server/change-streamer.ts
@@ -67,6 +67,7 @@ export default async function runWorker(
       backPressureLimitHeapProportion,
       flowControlConsensusTimeoutProportion,
       flowControlSlowSubscriberGracePeriodSeconds,
+      pgChangeLogEnabled,
       sqliteChangeLogMode,
       sqliteChangeLogReadPercent,
       sqliteChangeLogColdReadPercent,
@@ -123,7 +124,7 @@ export default async function runWorker(
   // purges. This ensures that (this) change-streamer will be able to resume
   // from the backup.
   let purgeLock =
-    litestream.backupURL && litestream.executable
+    pgChangeLogEnabled && litestream.backupURL && litestream.executable
       ? await new PurgeLocker(lc, shard, changeDB).acquire()
       : null;
   const restoreOptions = {litestream, constraints: purgeLock ?? undefined};
@@ -220,6 +221,7 @@ export default async function runWorker(
         purgeLock,
         autoReset ?? false,
         {
+          pgChangeLogEnabled,
           backPressureLimitHeapProportion,
           flowControlConsensusTimeoutProportion,
           flowControlSlowSubscriberGracePeriodMs:
@@ -258,14 +260,15 @@ export default async function runWorker(
               }
             : undefined,
           // Compare mode runs both advisory checks. Postgres remains authoritative.
-          sqliteChangeLogCompare: sqliteChangeLogComparing
-            ? {
-                replicaFile: replica.file,
-                comparePercent: sqliteChangeLogComparePercent,
-                retentionMs: sqliteChangeLogRetentionMs,
-                readBatchRows: sqliteChangeLogReadBatchRows,
-              }
-            : undefined,
+          sqliteChangeLogCompare:
+            pgChangeLogEnabled && sqliteChangeLogComparing
+              ? {
+                  replicaFile: replica.file,
+                  comparePercent: sqliteChangeLogComparePercent,
+                  retentionMs: sqliteChangeLogRetentionMs,
+                  readBatchRows: sqliteChangeLogReadBatchRows,
+                }
+              : undefined,
           // Slice 11 lands dark by default: serve mode constructs the stable
           // router, while readPercent=0 keeps every catchup on PG and emits
           // eligibility metrics before any canary traffic is enabled.
diff --git a/packages/zero-cache/src/services/change-streamer/change-streamer-service.pg.test.ts b/packages/zero-cache/src/services/change-streamer/change-streamer-service.pg.test.ts
index 2df17c592f..bb8db4a459 100644
--- a/packages/zero-cache/src/services/change-streamer/change-streamer-service.pg.test.ts
+++ b/packages/zero-cache/src/services/change-streamer/change-streamer-service.pg.test.ts
@@ -71,6 +71,7 @@ import {SQLiteChangeLogWriter} from './sqlite-change-log-writer.ts';
 import {PurgeLocker, Storer} from './storer.ts';
 
 const opts: TuningOptions = {
+  pgChangeLogEnabled: true,
   backPressureLimitHeapProportion: 0.04,
   flowControlConsensusTimeoutProportion: 2,
   statementTimeoutMs: 20_000,
@@ -350,9 +351,11 @@ describe('change-streamer/service', () => {
    * and can serve catchup from what it wrote.
    */
   type InlineChangeLogWriterOptions = {
+    pgChangeLogEnabled?: boolean | undefined;
     sqliteCatchup?: Partial<SQLiteCatchupOptions> | undefined;
     sqliteChangeLogPurge?: TuningOptions['sqliteChangeLogPurge'] | undefined;
     backupURL?: string | undefined;
+    backupVersion?: LitestreamVersion | undefined;
     sqliteChangeLogCompare?:
       | TuningOptions['sqliteChangeLogCompare']
       | undefined;
@@ -362,18 +365,27 @@ describe('change-streamer/service', () => {
   async function restartWithInlineChangeLogWriter(
     logFile: DbFile,
     {
+      pgChangeLogEnabled = true,
       sqliteCatchup,
       sqliteChangeLogPurge,
       backupURL,
+      backupVersion = 'legacy',
       sqliteChangeLogCompare,
       sqliteChangeLogServe,
     }: InlineChangeLogWriterOptions = {},
-  ): Promise<void> {
+  ): Promise<Mock<ChangeSource['startStream']>> {
     await streamer.stop();
     await streamerDone;
 
     changes = Subscription.create();
     acks = new Queue();
+    const startStream = vi.fn(() =>
+      Promise.resolve({
+        initialWatermark: REPLICA_VERSION,
+        changes,
+        acks: {push: (status: UpstreamStatusMessage) => acks.enqueue(status)},
+      }),
+    );
     streamer = await initializeStreamer(
       lc,
       shard,
@@ -382,22 +394,20 @@ describe('change-streamer/service', () => {
       'ws',
       sql,
       {
-        startStream: () =>
-          Promise.resolve({
-            initialWatermark: REPLICA_VERSION,
-            changes,
-            acks: {push: status => acks.enqueue(status)},
-          }),
+        startStream,
         startLagReporter: () => null,
         stop: () => Promise.resolve(),
       },
       ReplicationStatusPublisher.forTesting(),
       replicaConfig,
-      backupURL === undefined ? null : {backupURL, litestreamVersion: 'legacy'},
+      backupURL === undefined
+        ? null
+        : {backupURL, litestreamVersion: backupVersion},
       null,
       true,
       {
         ...opts,
+        pgChangeLogEnabled,
         // No replica beside it: the writer's anchor is the watermark its stream
         // connection resumes from, which is Postgres's `lastWatermark`.
         sqliteChangeLogWriter: {
@@ -427,6 +437,7 @@ describe('change-streamer/service', () => {
       setTimeoutFn as unknown as typeof setTimeout,
     );
     await run(streamer);
+    return startStream;
   }
 
   /** The oldest watermark retained in the SQLite change log beside `logFile`. */
@@ -1189,6 +1200,161 @@ describe('change-streamer/service', () => {
     }
   });
 
+  test('PG change-log flag makes SQLite authoritative and leaves PG frozen', async () => {
+    const readerRead = vi.spyOn(SQLiteChangeLogReader.prototype, 'read');
+    const replicaFile = new DbFile('sqlite-change-log-authoritative');
+    const replica = replicaFile.connect(lc);
+    replica.pragma('journal_mode = wal');
+    initReplicationState(replica, ['zero_data'], REPLICA_VERSION);
+
+    await sql`
+      INSERT INTO "zoro_3/cdc"."changeLog" (watermark, pos, change)
+        VALUES
+          ('04', 0, '{"tag":"begin"}'::json),
+          ('04', 1, '{"tag":"commit"}'::json)`;
+    await sql`
+      UPDATE "zoro_3/cdc"."replicationState" SET "lastWatermark" = '04'`;
+    const pgRowsBefore = await sql`
+      SELECT watermark, pos FROM "zoro_3/cdc"."changeLog"
+        ORDER BY watermark, pos`;
+
+    try {
+      const startStream = await restartWithInlineChangeLogWriter(replicaFile, {
+        pgChangeLogEnabled: false,
+        backupURL: 's3://foo/bar',
+        backupVersion: 'v5',
+        sqliteCatchup: {barrierPollIntervalMs: 10},
+        sqliteChangeLogPurge: {
+          retentionMs: 60_000,
+          batchRows: 100,
+        },
+        sqliteChangeLogServe: {
+          readPercent: 100,
+          coldReadPercent: 100,
+          retentionMs: 60_000,
+        },
+      });
+
+      // The replica, not the stale PG log, supplies the stream resume point.
+      expect(startStream).toHaveBeenCalledWith(REPLICA_VERSION, []);
+
+      changes.push(['begin', messages.begin(), {commitWatermark: '06'}]);
+      changes.push(['data', messages.insert('foo', {id: 'sqlite-only'})]);
+      changes.push(['commit', messages.commit(), {watermark: '06'}]);
+
+      await vi.waitFor(() => {
+        using log = openChangeLogDB(lc, replicaFile.path, {readonly: true});
+        expect(readChangeLogHead(log)).toBe('06');
+      });
+      expect(acks.size()).toBe(0);
+
+      // The backup replicator advances the canonical replica that Litestream
+      // backs up, so it must catch up from SQLite when PG is disabled.
+      const backupSub = await streamer.subscribe({
+        protocolVersion: PROTOCOL_VERSION,
+        taskID: 'backup-task',
+        id: 'backup-replicator',
+        mode: 'backup',
+        watermark: REPLICA_VERSION,
+        replicaVersion: REPLICA_VERSION,
+        initial: true,
+        logsChangeStream: false,
+      });
+      const backupOutput = drainToQueue(backupSub);
+      expect(await nextChange(backupOutput)).toMatchObject({tag: 'status'});
+      expect(await nextChange(backupOutput)).toMatchObject({tag: 'begin'});
+      expect(await nextChange(backupOutput)).toMatchObject({
+        tag: 'insert',
+        new: {id: 'sqlite-only'},
+      });
+      expect(await nextChange(backupOutput)).toMatchObject({tag: 'commit'});
+      expect(readerRead).toHaveBeenCalled();
+      backupSub.cancel();
+
+      const sub = await subscribeServing('sqlite-only');
+      const output = drainToQueue(sub);
+      expect(await nextChange(output)).toMatchObject({tag: 'status'});
+      expect(await nextChange(output)).toMatchObject({tag: 'begin'});
+      expect(await nextChange(output)).toMatchObject({
+        tag: 'insert',
+        new: {id: 'sqlite-only'},
+      });
+      expect(await nextChange(output)).toMatchObject({tag: 'commit'});
+      expect(readerRead).toHaveBeenCalled();
+      sub.cancel();
+
+      // Once SQLite has discarded the requested history, there is no stale
+      // PG fallback: the subscriber receives the terminal signal that causes
+      // its view-syncer to restore a fresh replica.
+      using changeLog = openChangeLogDB(lc, replicaFile.path, {
+        readonly: false,
+      });
+      changeLog
+        .prepare(`DELETE FROM "_zero.changeLogStream" WHERE watermark < '06'`)
+        .run();
+
+      // A backup below the SQLite seed cannot be demoted to PG. Its snapshot
+      // reservation stays pending until a later backup reaches the log.
+      const reservation =
+        await streamer.startSnapshotReservation('sqlite-restore');
+      const snapshots = drainSnapshotMessages(reservation);
+      streamer.trackBackupWatermark('04');
+      await expectAcks('04');
+      const timedOut: SnapshotMessage = [
+        'status',
+        {
+          tag: 'status',
+          backupURL: 'timed-out',
+          replicaVersion: 'timed-out',
+          minWatermark: 'timed-out',
+        },
+      ];
+      expect(await snapshots.dequeue(timedOut, 50)).toBe(timedOut);
+
+      // With PG persistence out of the ACK set, the verified v5 backup owns
+      // source ACK progress and releases the held reservation.
+      streamer.trackBackupWatermark('06');
+      await expectAcks('06');
+      expect(await snapshots.dequeue()).toEqual([
+        'status',
+        {
+          tag: 'status',
+          backupURL: 's3://foo/bar',
+          replicaVersion: REPLICA_VERSION,
+          minWatermark: '06',
+        },
+      ]);
+      reservation.cancel();
+
+      const tooOld = await subscribeServing('sqlite-too-old');
+      const tooOldOutput = drainToQueue(tooOld);
+      expect(await tooOldOutput.dequeue()).toEqual([
+        'error',
+        {
+          type: ErrorType.WatermarkTooOld,
+          message: 'earliest supported watermark is 06 (requested 01)',
+        },
+      ]);
+
+      expect(
+        await sql`
+          SELECT watermark, pos FROM "zoro_3/cdc"."changeLog"
+            ORDER BY watermark, pos`,
+      ).toEqual(pgRowsBefore);
+      expect(
+        await sql`
+          SELECT "lastWatermark" FROM "zoro_3/cdc"."replicationState"`,
+      ).toEqual([{lastWatermark: '04'}]);
+    } finally {
+      readerRead.mockRestore();
+      await streamer.stop();
+      await streamerDone;
+      replica.close();
+      deleteChangeLogDB(replicaFile.path);
+      replicaFile.delete();
+    }
+  });
+
   /**
    * The SQLite floor's live constraint is level-triggered, not edge-triggered:
    * backup monitors never resend an unchanged floor, so once a laggard's ACK
diff --git a/packages/zero-cache/src/services/change-streamer/change-streamer-service.ts b/packages/zero-cache/src/services/change-streamer/change-streamer-service.ts
index 41cad4d495..6c934f656a 100644
--- a/packages/zero-cache/src/services/change-streamer/change-streamer-service.ts
+++ b/packages/zero-cache/src/services/change-streamer/change-streamer-service.ts
@@ -38,6 +38,7 @@ import {
   RunningState,
   UnrecoverableError,
 } from '../running-state.ts';
+import {serializeChangeStreamData} from './change-log-codec.ts';
 import {
   ChangeLogInitializer,
   replicaInitializationSource,
@@ -147,6 +148,12 @@ export type SQLiteChangeLogServeOptions = {
 };
 
 export type TuningOptions = StorerOptions & {
+  /**
+   * Keeps the legacy Postgres change log on the initialization, persistence,
+   * catchup, purge, and ACK paths. Disable only after SQLite and the replica
+   * backup are authoritative for all of those responsibilities.
+   */
+  pgChangeLogEnabled: boolean;
   flowControlConsensusTimeoutProportion: number;
   flowControlSlowSubscriberGracePeriodMs?: number | undefined;
   sqliteCatchup?: SQLiteCatchupOptions | undefined;
@@ -210,6 +217,7 @@ export async function initializeStreamer(
     autoReset,
     purgeLock ?? undefined,
     setTimeoutFn,
+    opts.pgChangeLogEnabled,
   );
 
   // Dynamically creates connection pools that the implementation uses to
@@ -407,6 +415,7 @@ class ChangeStreamerImpl implements ChangeStreamerService {
   readonly #replicaVersion: string;
   readonly #source: ChangeSource;
   readonly #storer: Storer;
+  readonly #pgChangeLogEnabled: boolean;
   readonly #forwarder: Forwarder;
   readonly #reservations: SnapshotReservations | undefined;
   readonly #replicationStatusPublisher: ReplicationStatusPublisher;
@@ -521,6 +530,22 @@ class ChangeStreamerImpl implements ChangeStreamerService {
     this.#changeDBProvider = changeDBProvider;
     this.#replicaVersion = replicaVersion;
     this.#source = source;
+    this.#pgChangeLogEnabled = opts.pgChangeLogEnabled;
+    if (!this.#pgChangeLogEnabled) {
+      assert(
+        opts.sqliteChangeLogWriter &&
+          opts.sqliteChangeLogPurge &&
+          opts.sqliteCatchup &&
+          opts.sqliteChangeLogServe?.readPercent === 100 &&
+          opts.sqliteChangeLogServe.coldReadPercent === 100 &&
+          backupConfig?.litestreamVersion === 'v5',
+        'disabling the PG change log requires SQLite serving at 100 percent and a v5 backup',
+      );
+      assert(
+        opts.sqliteChangeLogCompare === undefined,
+        'SQLite change-log comparison requires the PG change log',
+      );
+    }
     this.#storer = new Storer(
       lc,
       shard,
@@ -610,16 +635,21 @@ class ChangeStreamerImpl implements ChangeStreamerService {
           )
         : undefined;
     this.#acker = new UpstreamAcker({
-      trackPgChangeLog: true, // TODO: set false when retiring PG
+      trackPgChangeLog: this.#pgChangeLogEnabled,
       trackBackup: backupConfig?.litestreamVersion === 'v5',
     });
-    const replicaSource = opts.sqliteChangeLogCompare
-      ? replicaInitializationSource(lc, opts.sqliteChangeLogCompare.replicaFile)
+    const replicaFileForInitialization =
+      opts.sqliteChangeLogCompare?.replicaFile ??
+      (!this.#pgChangeLogEnabled
+        ? opts.sqliteChangeLogWriter?.replicaFile
+        : undefined);
+    const replicaSource = replicaFileForInitialization
+      ? replicaInitializationSource(lc, replicaFileForInitialization)
       : undefined;
     this.#initializer = new ChangeLogInitializer(
       lc,
       {
-        initFromPgChangeLog: true, // TODO: set false when retiring PG
+        initFromPgChangeLog: this.#pgChangeLogEnabled,
         initFromReplica: replicaSource !== undefined,
       },
       {
@@ -649,6 +679,7 @@ class ChangeStreamerImpl implements ChangeStreamerService {
       : undefined;
     // Compare mode requires the writer and catchup configuration.
     this.#comparator =
+      this.#pgChangeLogEnabled &&
       opts.sqliteChangeLogCompare &&
       opts.sqliteChangeLogWriter &&
       opts.sqliteCatchup
@@ -704,14 +735,16 @@ class ChangeStreamerImpl implements ChangeStreamerService {
         const {lastWatermark, backfillRequests} =
           await this.#initializer.initialize();
         // SQLite catchup must not be eligible until this has been initialized
-        // from the durable PG head. Commits observed only since process startup
-        // are insufficient after a change-streamer restart.
+        // from the selected durable head. Commits observed only since process
+        // startup are insufficient after a change-streamer restart.
         this.#lastForwardedCommitWatermark = lastWatermark;
         const stream = await this.#source.startStream(
           lastWatermark,
           backfillRequests,
         );
-        this.#storer.run().catch(e => stream.changes.cancel(e));
+        if (this.#pgChangeLogEnabled) {
+          this.#storer.run().catch(e => stream.changes.cancel(e));
+        }
 
         this.#stream = stream;
         if (
@@ -776,16 +809,17 @@ class ChangeStreamerImpl implements ChangeStreamerService {
               break;
           }
 
-          const json = this.#storer.store(watermark, change);
+          const json = this.#pgChangeLogEnabled
+            ? this.#storer.store(watermark, change)
+            : serializeChangeStreamData(change);
           // The SQLite change log commits at transaction boundaries, and its
           // commit for this transaction lands here -- before the forward of the
           // `commit` message, and before #recordForwardedTransactionBoundary
           // advances what #captureRequiredHead reads. No `await` separates it
-          // from #storer.store() above, which is the assertable form of
-          // invariant 1: the storer only enqueues, and its Postgres commit
-          // cannot complete without yielding, so a synchronous SQLite commit in
-          // the same loop iteration always precedes anything that can advance
-          // the watermark this stream would resume from. Never throws; a write
+          // from serialization or the optional PG enqueue above. This is the
+          // assertable form of invariant 1: a synchronous SQLite commit in the
+          // same loop iteration always precedes anything that can advance the
+          // watermark this stream would resume from. Never throws; a write
           // failure disables the writer rather than stopping replication.
           this.#changeLogWriter?.write(change, json);
           const entry: WatermarkedChange = [watermark, change[1].tag, json];
@@ -820,8 +854,10 @@ class ChangeStreamerImpl implements ChangeStreamerService {
             watermark = null;
           }
 
-          // Allow the storer to exert back pressure.
-          const readyForMore = this.#storer.readyForMore();
+          // Allow the PG storer to exert back pressure when it is enabled.
+          const readyForMore = this.#pgChangeLogEnabled
+            ? this.#storer.readyForMore()
+            : undefined;
           if (readyForMore) {
             await promiseOrAbort(
               readyForMore,
@@ -840,7 +876,9 @@ class ChangeStreamerImpl implements ChangeStreamerService {
       // When the change stream is interrupted, abort any pending transaction.
       if (watermark) {
         this.#lc.warn?.(`aborting interrupted transaction ${watermark}`);
-        this.#storer.abort();
+        if (this.#pgChangeLogEnabled) {
+          this.#storer.abort();
+        }
         // Rolling back the log leaves no rows for the interrupted transaction,
         // so the next connection's reconciliation sees a head at or below its
         // resume watermark rather than a partial transaction.
@@ -934,17 +972,32 @@ class ChangeStreamerImpl implements ChangeStreamerService {
       lc.info?.(`adding subscriber ${subscriber.id}`);
 
       const catchupFromPG = () => {
+        assert(
+          this.#pgChangeLogEnabled,
+          'cannot catch up from a disabled PG change log',
+        );
         // Keep the existing PG registration/catchup lockstep unchanged when
         // SQLite was not selected before Forwarder.add().
         cleanupSubscriber = removeFromForwarder;
         this.#forwarder.add(subscriber);
         this.#storer.catchup(subscriber, mode);
       };
-      const sqliteSelection = this.#selectSQLiteCatchup(lc, ctx);
-      if (!sqliteSelection) {
+      const sqliteDecision = this.#selectSQLiteCatchup(lc, ctx);
+      if (!sqliteDecision) {
         catchupFromPG();
+      } else if (sqliteDecision.kind === 'rejected') {
+        lc.warn?.(
+          `${
+            sqliteDecision.terminal ? 'rejecting' : 'ending'
+          } subscription for ${ctx.id}: ${sqliteDecision.message}`,
+        );
+        if (sqliteDecision.terminal) {
+          subscriber.close(ErrorType.WatermarkTooOld, sqliteDecision.message);
+        } else {
+          subscriber.close();
+        }
       } else {
-        const {catchup, reason, coverage, logWarm} = sqliteSelection;
+        const {catchup, reason, coverage, logWarm} = sqliteDecision;
         cleanupSubscriber = () => catchup.remove(subscriber);
         const registration = await catchup.catchup(
           subscriber,
@@ -960,28 +1013,51 @@ class ChangeStreamerImpl implements ChangeStreamerService {
             this.#recordCatchupRoute('sqlite', reason);
             break;
           case 'uncovered':
-            lc.info?.(
-              `serving ${ctx.id} from PG catchup: subscriber watermark ` +
-                `${ctx.watermark} is below the SQLite change-log minimum ` +
-                registration.minWatermark,
-            );
-            this.#recordCatchupRoute('pg', 'watermark-uncovered');
-            catchupFromPG();
+            if (this.#pgChangeLogEnabled) {
+              lc.info?.(
+                `serving ${ctx.id} from PG catchup: subscriber watermark ` +
+                  `${ctx.watermark} is below the SQLite change-log minimum ` +
+                  registration.minWatermark,
+              );
+              this.#recordCatchupRoute('pg', 'watermark-uncovered');
+              catchupFromPG();
+            } else {
+              const message =
+                `earliest supported watermark is ` +
+                `${registration.minWatermark} (requested ${ctx.watermark})`;
+              lc.warn?.(
+                `rejecting subscriber at watermark ${ctx.watermark}: ` +
+                  `the SQLite change log starts at ` +
+                  `${registration.minWatermark} and the PG change log is disabled`,
+              );
+              this.#recordCatchupRoute('none', 'watermark-uncovered');
+              subscriber.close(ErrorType.WatermarkTooOld, message);
+            }
             break;
           case 'declined':
             // Registration failed before the subscriber was committed to
-            // SQLite, so PG is still available. The coordinator has already
-            // tripped the breaker, which keeps the retry off SQLite for the
-            // cooldown; it is deliberately not closed here, since closing it
-            // would abort the catchups of every other subscriber it is
-            // serving.
-            lc.error?.(
-              `serving ${ctx.id} from PG catchup: SQLite catchup ` +
-                `registration failed`,
-              registration.error,
-            );
-            this.#recordCatchupRoute('pg', 'registration-failed');
-            catchupFromPG();
+            // SQLite. Use PG when enabled; otherwise end this subscription so
+            // it retries. The coordinator has already tripped the breaker,
+            // which keeps the retry off SQLite for the cooldown; it is
+            // deliberately not closed here, since closing it would abort the
+            // catchups of every other subscriber it is serving.
+            if (this.#pgChangeLogEnabled) {
+              lc.error?.(
+                `serving ${ctx.id} from PG catchup: SQLite catchup ` +
+                  `registration failed`,
+                registration.error,
+              );
+              this.#recordCatchupRoute('pg', 'registration-failed');
+              catchupFromPG();
+            } else {
+              lc.error?.(
+                `ending subscription for ${ctx.id} to retry SQLite catchup: ` +
+                  `registration failed and the PG change log is disabled`,
+                registration.error,
+              );
+              this.#recordCatchupRoute('none', 'registration-failed');
+              subscriber.fail(registration.error);
+            }
             break;
           case 'handled':
             // The coordinator closed or failed the subscriber itself, so
@@ -1065,14 +1141,15 @@ class ChangeStreamerImpl implements ChangeStreamerService {
       return;
     }
 
-    // Resolve PG bounds before touching any pin. Everything below runs to
-    // completion without awaiting, which is what keeps a reservation's
+    // Resolve PG bounds, when enabled, before touching any pin. Everything
+    // below runs to completion without awaiting, which keeps a reservation's
     // advertised bounds and its pin in agreement: a concurrent /snapshot
     // retry for the same task replaces the reservation and re-pins it, and
     // confirming that replacement with the previous pin's bounds is exactly
-    // the mismatch pinning exists to prevent. The cost is one PG round trip
-    // per call even when every reservation is pinned to SQLite.
-    const pgState = await this.#getChangeLogState();
+    // the mismatch pinning exists to prevent.
+    const pgState = this.#pgChangeLogEnabled
+      ? await this.#getChangeLogState()
+      : undefined;
     for (const taskID of reservations.unconfirmedTaskIDs()) {
       let route = this.#readRouter?.peek(taskID);
       // startSnapshotReservation pins only after an in-flight purge has
@@ -1081,12 +1158,34 @@ class ChangeStreamerImpl implements ChangeStreamerService {
         continue;
       }
 
+      // A transiently unavailable or broken SQLite log normally pins the
+      // reservation to PG. With PG disabled, re-evaluate that choice on each
+      // backup notification and keep the reservation pending until SQLite can
+      // supply bounds.
+      if (!this.#pgChangeLogEnabled && route?.source === 'pg') {
+        this.#readRouter?.release(taskID);
+        route = this.#readRouter?.pin(taskID);
+        if (route?.source === 'pg') {
+          if (reservations.noteConfirmationDelayed(taskID)) {
+            this.#reservationConfirmDelays.add(1);
+          }
+          this.#lc.warn?.(
+            `delaying snapshot reservation for ${taskID}: SQLite change log ` +
+              `is unavailable and the PG change log is disabled`,
+          );
+          continue;
+        }
+      }
+
       if (route?.source === 'sqlite') {
         const coverage = must(
           route.coverage,
           'a pinned SQLite route must carry its covered range',
         );
-        if (coverage.minWatermark > backupWatermark) {
+        if (
+          this.#pgChangeLogEnabled &&
+          coverage.minWatermark > backupWatermark
+        ) {
           // A log seeded after this backup, most often. Holding the
           // reservation until a backup reaches the log's minimum would stall
           // a follower that PG can serve now, so move it -- pin included.
@@ -1110,13 +1209,16 @@ class ChangeStreamerImpl implements ChangeStreamerService {
           'a pinned SQLite route must carry its covered range',
         ).minWatermark;
       } else {
-        ({minWatermark} = pgState);
+        minWatermark = must(
+          pgState,
+          'a PG reservation route requires the PG change log',
+        ).minWatermark;
       }
 
       if (minWatermark <= backupWatermark) {
         reservations.confirmFor(
           taskID,
-          pgState.replicaVersion,
+          this.#replicaVersion,
           backupWatermark,
           source,
         );
@@ -1128,7 +1230,7 @@ class ChangeStreamerImpl implements ChangeStreamerService {
           this.#reservationConfirmDelays.add(1);
         }
         this.#lc.error?.(
-          `pg change-log minWatermark ${minWatermark} is later than ` +
+          `${source} change-log minWatermark ${minWatermark} is later than ` +
             `backupWatermark ${backupWatermark}. Delaying confirmation of ` +
             `snapshot reservation until next backup.`,
         );
@@ -1139,6 +1241,7 @@ class ChangeStreamerImpl implements ChangeStreamerService {
   #maybeSchedulePGPurge(): void {
     const backupWatermark = this.#backupWatermark;
     if (
+      !this.#pgChangeLogEnabled ||
       this.#pgPurgeScheduled ||
       this.#pgPurgeRunning ||
       backupWatermark === undefined ||
@@ -1369,20 +1472,24 @@ class ChangeStreamerImpl implements ChangeStreamerService {
   #selectSQLiteCatchup(
     lc: LogContext,
     ctx: SubscriberContext,
-  ): SQLiteCatchupSelection | undefined {
+  ): SQLiteCatchupDecision | undefined {
     const opts = this.#sqliteCatchupOptions;
     if (this.#lastForwardedCommitWatermark === undefined || !opts) {
-      this.#recordCatchupRoute('pg', 'not-ready');
-      return undefined;
+      return this.#fallbackOrReject(
+        'not-ready',
+        'SQLite catchup is not ready and the PG change log is disabled',
+      );
     }
-    // SQLite catchup is only for disposable serving replicas. Backup
-    // subscribers retain the existing PG recovery policy until RMv2 can
-    // resume the canonical replica directly from replica/backup/slot state.
-    // Enforce this before the selector so no canary policy can override it.
-    if (ctx.mode !== 'serving') {
+    // While the PG change log is available, backup subscribers retain its
+    // existing recovery policy and cannot be moved by a canary selector.
+    // Once PG is disabled, the backup replicator must use SQLite catchup: it
+    // advances the canonical replica that Litestream backs up.
+    if (ctx.mode === 'backup' && this.#pgChangeLogEnabled) {
       lc.info?.(`not serving backup subscriber ${ctx.id} from SQLite catchup`);
-      this.#recordCatchupRoute('pg', 'ineligible-mode');
-      return undefined;
+      return this.#fallbackOrReject(
+        'ineligible-mode',
+        'backup subscribers retain the PG change-log recovery policy',
+      );
     }
 
     let route: ChangeLogReadRoute | undefined;
@@ -1409,8 +1516,11 @@ class ChangeStreamerImpl implements ChangeStreamerService {
               : []),
           );
         }
-        this.#recordCatchupRoute('pg', route.reason);
-        return undefined;
+        return this.#fallbackOrReject(
+          route.reason,
+          `SQLite catchup route ${route.reason} is unavailable and the PG ` +
+            `change log is disabled`,
+        );
       }
       const coverage = route.coverage;
       assert(coverage, 'a SQLite route must carry its covered range');
@@ -1421,26 +1531,35 @@ class ChangeStreamerImpl implements ChangeStreamerService {
             coverage.minWatermark,
           {sqliteChangeLogCoverage: coverage},
         );
-        this.#recordCatchupRoute('pg', 'watermark-uncovered');
-        return undefined;
+        return this.#fallbackOrReject(
+          'watermark-uncovered',
+          `earliest supported watermark is ${coverage.minWatermark} ` +
+            `(requested ${ctx.watermark})`,
+          true,
+        );
       }
     }
 
     // `shouldUse` remains as a test hook and an optional extra policy gate.
     // Production selection comes from #readRouter.
     if (!this.#readRouter && !opts.shouldUse?.(ctx)) {
-      this.#recordCatchupRoute('pg', 'selector');
-      return undefined;
+      return this.#fallbackOrReject(
+        'selector',
+        'SQLite catchup was not selected and the PG change log is disabled',
+      );
     }
     if (this.#readRouter && opts.shouldUse && !opts.shouldUse(ctx)) {
-      this.#recordCatchupRoute('pg', 'selector');
-      return undefined;
+      return this.#fallbackOrReject(
+        'selector',
+        'SQLite catchup was not selected and the PG change log is disabled',
+      );
     }
 
     const catchup =
       this.#sqliteCatchup ?? this.#openSQLiteCatchup(lc, opts, ctx);
     if (catchup) {
       return {
+        kind: 'selected',
         catchup,
         reason: route?.reason ?? 'selector',
         coverage: route?.coverage,
@@ -1449,12 +1568,25 @@ class ChangeStreamerImpl implements ChangeStreamerService {
           route === undefined ? undefined : route.reason !== 'selected-cold',
       };
     } else {
-      this.#recordCatchupRoute('pg', 'log-unavailable');
+      return this.#fallbackOrReject(
+        'log-unavailable',
+        'SQLite change log is unavailable and the PG change log is disabled',
+      );
     }
-    return undefined;
   }
 
-  #recordCatchupRoute(source: 'pg' | 'sqlite', reason: string): void {
+  #fallbackOrReject(
+    reason: string,
+    message: string,
+    terminal = false,
+  ): SQLiteCatchupRejection | undefined {
+    this.#recordCatchupRoute(this.#pgChangeLogEnabled ? 'pg' : 'none', reason);
+    return this.#pgChangeLogEnabled
+      ? undefined
+      : {kind: 'rejected', message, terminal};
+  }
+
+  #recordCatchupRoute(source: 'pg' | 'sqlite' | 'none', reason: string): void {
     this.#catchupRoutes.add(1, {source, reason});
   }
 
@@ -1466,10 +1598,11 @@ class ChangeStreamerImpl implements ChangeStreamerService {
    * first reconcile, and deletes it when it fails soft -- or may exist without
    * content, or may be unreadable. None of those can be allowed to fail a
    * subscription:
-   * this is the last point at which PG catchup is still available -- past
-   * `Forwarder.add()` the subscriber is committed to SQLite -- so each declines
-   * here instead. Neither the failure nor the reader is retained, so a later
-   * subscription retries from scratch.
+   * this is the last point at which PG catchup can be selected when it is
+   * enabled -- past `Forwarder.add()` the subscriber is committed to SQLite --
+   * so each declines here instead. With PG disabled, the caller ends the
+   * subscription so that it retries. Neither the failure nor the reader is
+   * retained, so a later subscription retries from scratch.
    */
   #openSQLiteCatchup(
     lc: LogContext,
@@ -1551,7 +1684,10 @@ class ChangeStreamerImpl implements ChangeStreamerService {
       opts.notReadyWarnThresholdMs ??
       DEFAULT_CHANGE_LOG_UNAVAILABLE_WARN_THRESHOLD_MS;
     lc[unavailableMs >= threshold ? 'warn' : 'debug']?.(
-      `serving ${ctx.id} from PG catchup: ${reason} ` +
+      (this.#pgChangeLogEnabled
+        ? `serving ${ctx.id} from PG catchup: `
+        : `cannot serve ${ctx.id} from SQLite catchup: `) +
+        `${reason} ` +
         `(unavailable for ${unavailableMs} ms)`,
       ...(error === undefined ? [] : [error]),
     );
@@ -1563,6 +1699,7 @@ type ForwardedTransactionCompletion =
   | {kind: 'rolled-back'; watermark: string};
 
 type SQLiteCatchupSelection = {
+  readonly kind: 'selected';
   readonly catchup: SQLiteChangeLogCatchup;
   readonly reason: string;
   readonly coverage: SQLiteChangeLogCoverage | undefined;
@@ -1573,6 +1710,15 @@ type SQLiteCatchupSelection = {
   readonly logWarm: boolean | undefined;
 };
 
+type SQLiteCatchupRejection = {
+  readonly kind: 'rejected';
+  readonly message: string;
+  /** A terminal rejection forces the view-syncer to restore a fresh replica. */
+  readonly terminal: boolean;
+};
+
+type SQLiteCatchupDecision = SQLiteCatchupSelection | SQLiteCatchupRejection;
+
 // The delay between receiving an initial, backup-based watermark
 // and performing a check of whether to purge records before it.
 // This delay should be long enough to handle situations like the following:
diff --git a/packages/zero-cache/src/services/change-streamer/schema/tables.pg.test.ts b/packages/zero-cache/src/services/change-streamer/schema/tables.pg.test.ts
index 200ff5ae63..04f3f99105 100644
--- a/packages/zero-cache/src/services/change-streamer/schema/tables.pg.test.ts
+++ b/packages/zero-cache/src/services/change-streamer/schema/tables.pg.test.ts
@@ -268,6 +268,53 @@ describe('change-streamer/schema/tables', () => {
     expect(purgeLock.release).not.toHaveBeenCalled();
   });
 
+  test('disabled PG change log is neither initialized nor reset', async () => {
+    await ensureReplicationConfig(
+      lc,
+      sql,
+      {
+        replicaVersion: '183',
+        publications: ['zero_data'],
+        watermark: '183',
+      },
+      shard,
+      true,
+      undefined,
+      undefined,
+      false,
+    );
+    expect(
+      await sql`SELECT watermark, pos FROM "rezo_8/cdc"."changeLog"`,
+    ).toEqual([]);
+
+    await sql`
+      INSERT INTO "rezo_8/cdc"."changeLog" (watermark, pos, change)
+        VALUES ('stale', 0, '{"tag":"begin"}'::json)`;
+
+    await ensureReplicationConfig(
+      lc,
+      sql,
+      {
+        replicaVersion: '1g8',
+        publications: ['zero_data'],
+        watermark: '1g8',
+      },
+      shard,
+      true,
+      undefined,
+      undefined,
+      false,
+    );
+
+    expect(
+      await sql`SELECT watermark, pos FROM "rezo_8/cdc"."changeLog"`,
+    ).toEqual([{watermark: 'stale', pos: 0n}]);
+    expect(
+      await sql`
+        SELECT "lastWatermark" FROM "rezo_8/cdc"."replicationState"`,
+    ).toEqual([{lastWatermark: '1g8'}]);
+  });
+
   test('no deadlocks when table is reset', async () => {
     // Set up initial replication config.
     await ensureReplicationConfig(
diff --git a/packages/zero-cache/src/services/change-streamer/schema/tables.ts b/packages/zero-cache/src/services/change-streamer/schema/tables.ts
index f7bae2ec29..d72456c90a 100644
--- a/packages/zero-cache/src/services/change-streamer/schema/tables.ts
+++ b/packages/zero-cache/src/services/change-streamer/schema/tables.ts
@@ -174,6 +174,7 @@ export async function ensureReplicationConfig(
   autoReset: boolean,
   purgeLock?: PurgeLock,
   setTimeoutFn: typeof setTimeout = setTimeout,
+  pgChangeLogEnabled = true,
 ) {
   const {publications, replicaVersion, watermark} = subscriptionState;
   const replicaConfig = {publications, replicaVersion};
@@ -230,7 +231,9 @@ export async function ensureReplicationConfig(
         needsTruncate = true;
         stmts.push(
           sql`TRUNCATE TABLE ${sql(schema)}."replicationState"`,
-          sql`TRUNCATE TABLE ${sql(schema)}."changeLog"`,
+          ...(pgChangeLogEnabled
+            ? [sql`TRUNCATE TABLE ${sql(schema)}."changeLog"`]
+            : []),
           sql`TRUNCATE TABLE ${sql(schema)}."replicationConfig"`,
           sql`TRUNCATE TABLE ${sql(schema)}."tableMetadata"`,
           sql`TRUNCATE TABLE ${sql(schema)}."backfilling"`,
@@ -239,18 +242,20 @@ export async function ensureReplicationConfig(
     }
     // Initialize (or re-initialize TRUNCATED) tables
     if (results.length === 0 || needsTruncate) {
-      // The storer uses the earliest changeLog entry as the safe watermark
-      // from which subscribers can be resumed. These initial entries ensure
-      // that subscribers can start from a freshly synced replica, even if
-      // new changes have been replicated and not purged from the changeLog.
+      // When enabled, the PG storer uses the earliest changeLog entry as the
+      // safe watermark from which subscribers can be resumed. These initial
+      // entries ensure that subscribers can start from a freshly synced
+      // replica, even if new changes have been replicated and not purged.
       //
       // TODO: Replace this with an explicit `firstWatermark` column in the
       //       change db.
       const watermark = replicaConfig.replicaVersion;
-      const initialTx: FullChangeLogEntry[] = [
-        {watermark, pos: 0, change: {tag: 'begin'}},
-        {watermark, pos: 1, change: {tag: 'commit'}},
-      ];
+      const initialTx: FullChangeLogEntry[] = pgChangeLogEnabled
+        ? [
+            {watermark, pos: 0, change: {tag: 'begin'}},
+            {watermark, pos: 1, change: {tag: 'commit'}},
+          ]
+        : [];
 
       stmts.push(
         sql`INSERT INTO ${sql(schema)}."replicationConfig" ${sql(replicaConfig)}`,
diff --git a/packages/zero-cache/src/services/replicator/replication-resumption.pg.test.ts b/packages/zero-cache/src/services/replicator/replication-resumption.pg.test.ts
index ef9b91808c..1c705eb6b2 100644
--- a/packages/zero-cache/src/services/replicator/replication-resumption.pg.test.ts
+++ b/packages/zero-cache/src/services/replicator/replication-resumption.pg.test.ts
@@ -51,6 +51,7 @@ const shard = {
 };
 
 const streamerOptions: TuningOptions = {
+  pgChangeLogEnabled: true,
   backPressureLimitHeapProportion: 0.04,
   flowControlConsensusTimeoutProportion: 2,
   statementTimeoutMs: 20_000,
diff --git a/packages/zero-cache/src/services/replicator/replication-throughput.bench.pg.ts b/packages/zero-cache/src/services/replicator/replication-throughput.bench.pg.ts
index e43a933e40..f80ddddf3e 100644
--- a/packages/zero-cache/src/services/replicator/replication-throughput.bench.pg.ts
+++ b/packages/zero-cache/src/services/replicator/replication-throughput.bench.pg.ts
@@ -59,6 +59,7 @@ const shard = {
 };
 const benchmarkRecorder = createManualBenchmarkRecorder();
 const streamerOptions = {
+  pgChangeLogEnabled: true,
   backPressureLimitHeapProportion: 0.04,
   flowControlConsensusTimeoutProportion: 2,
   statementTimeoutMs: 60_000,