diff --git a/cmd/ateapi/internal/store/atepg/atepg.go b/cmd/ateapi/internal/store/atepg/atepg.go
index ead0b22f8d..2e0f0a8350 100644
--- a/cmd/ateapi/internal/store/atepg/atepg.go
+++ b/cmd/ateapi/internal/store/atepg/atepg.go
@@ -23,10 +23,10 @@ package atepg
 
 import (
 	"context"
-	"encoding/json"
 	"errors"
 	"fmt"
 	"log/slog"
+	"strings"
 	"time"
 
 	"github.com/agent-substrate/substrate/cmd/ateapi/internal/store"
@@ -36,15 +36,16 @@ import (
 	"github.com/jackc/pgx/v5"
 	"github.com/jackc/pgx/v5/pgconn"
 	"github.com/jackc/pgx/v5/pgxpool"
-	"google.golang.org/protobuf/encoding/protojson"
 	"google.golang.org/protobuf/proto"
 	"google.golang.org/protobuf/types/known/timestamppb"
 )
 
 // Persistence is a service that stores ate state in PostgreSQL.
 type Persistence struct {
-	pool    *pgxpool.Pool
-	lockTTL time.Duration
+	pool            *pgxpool.Pool
+	lockTTL         time.Duration
+	stopMaintenance context.CancelFunc
+	maintenanceDone chan struct{}
 }
 
 var _ store.Interface = (*Persistence)(nil)
@@ -75,7 +76,27 @@ func NewPersistence(ctx context.Context, pool *pgxpool.Pool) (*Persistence, erro
 	if err := applySchema(ctx, pool); err != nil {
 		return nil, err
 	}
-	return &Persistence{pool: pool, lockTTL: defaultLockTTL}, nil
+	maintenanceCtx, stopMaintenance := context.WithCancel(context.Background())
+	p := &Persistence{pool: pool, lockTTL: defaultLockTTL, stopMaintenance: stopMaintenance, maintenanceDone: make(chan struct{})}
+	// Cover the partition lead before accepting writes; from then on the
+	// maintenance loop keeps partitions ahead of the clock (and the
+	// DEFAULT partition catches writes if it ever falls behind).
+	if err := p.createWorkerChangesPartitions(ctx, changeFeedPartitionLeadTimes(time.Now())...); err != nil {
+		stopMaintenance()
+		return nil, err
+	}
+	go func() {
+		defer close(p.maintenanceDone)
+		p.changeFeedMaintenance(maintenanceCtx)
+	}()
+	return p, nil
+}
+
+// Close stops the change-feed maintenance loop and waits for it to exit.
+// It does not close the pool, which the caller owns.
+func (p *Persistence) Close() {
+	p.stopMaintenance()
+	<-p.maintenanceDone
 }
 
 // querier is satisfied by both *pgxpool.Pool and pgx.Tx, letting read helpers
@@ -1234,68 +1255,51 @@ func (p *Persistence) DeleteActorSnapshotTag(ctx context.Context, atespace, name
 
 // --- Workers ---
 
-const (
-	// workerChangeChannel is the fixed LISTEN/NOTIFY channel for worker changes.
-	workerChangeChannel = "worker_changes"
-	// maxNotifyPayloadBytes reflects PostgreSQL's NOTIFY payload size limit.
-	// Writes fail rather than silently omit a notification if exceeded.
-	maxNotifyPayloadBytes = 8000
-)
-
-type workerEventEnvelope struct {
-	Type   int    `json:"t"`
-	Worker string `json:"w"` // protojson-encoded Worker
-}
-
+// Feed payload format: one event-type byte followed by the binary Worker proto.
+// The tag byte is read by other replicas during rolling deploys, so
+// store.WorkerEventType values must stay append-only stable and fit a byte.
 func marshalWorkerEvent(eventType store.WorkerEventType, worker *ateapipb.Worker) ([]byte, error) {
-	workerJSON, err := protojson.Marshal(worker)
-	if err != nil {
-		return nil, fmt.Errorf("in protojson.Marshal: %w", err)
-	}
-	msg, err := json.Marshal(workerEventEnvelope{Type: int(eventType), Worker: string(workerJSON)})
+	b, err := proto.Marshal(worker)
 	if err != nil {
-		return nil, fmt.Errorf("in json.Marshal: %w", err)
+		return nil, fmt.Errorf("in proto.Marshal: %w", err)
 	}
-	return msg, nil
+	return append([]byte{byte(eventType)}, b...), nil
 }
 
-func unmarshalWorkerEvent(payload string) (store.WorkerEvent, error) {
-	var env workerEventEnvelope
-	if err := json.Unmarshal([]byte(payload), &env); err != nil {
-		return store.WorkerEvent{}, fmt.Errorf("in json.Unmarshal: %w", err)
+func unmarshalWorkerEvent(payload []byte) (store.WorkerEvent, error) {
+	if len(payload) == 0 {
+		return store.WorkerEvent{}, fmt.Errorf("empty worker event payload")
 	}
 	worker := &ateapipb.Worker{}
-	if err := protojson.Unmarshal([]byte(env.Worker), worker); err != nil {
-		return store.WorkerEvent{}, fmt.Errorf("in protojson.Unmarshal: %w", err)
+	if err := proto.Unmarshal(payload[1:], worker); err != nil {
+		return store.WorkerEvent{}, fmt.Errorf("in proto.Unmarshal: %w", err)
 	}
-	return store.WorkerEvent{Type: store.WorkerEventType(env.Type), Worker: worker}, nil
+	return store.WorkerEvent{Type: store.WorkerEventType(payload[0]), Worker: worker}, nil
 }
 
-// writeAndNotify runs fn inside a transaction, then--only if fn reports a
-// change worth notifying--calls pg_notify in the same transaction so
-// delivery happens if and only if the transaction commits.
-func (p *Persistence) writeAndNotify(ctx context.Context, eventType store.WorkerEventType, worker *ateapipb.Worker, fn func(ctx context.Context, tx pgx.Tx) (notify bool, err error)) error {
+// writeAndAppendChange runs fn inside a transaction, then--only if fn
+// reports a change worth publishing--appends the event to the worker_changes
+// feed in the same transaction, so watchers see it if and only if the
+// transaction commits.
+func (p *Persistence) writeAndAppendChange(ctx context.Context, eventType store.WorkerEventType, worker *ateapipb.Worker, fn func(ctx context.Context, tx pgx.Tx) (changed bool, err error)) error {
 	tx, err := p.pool.Begin(ctx)
 	if err != nil {
 		return fmt.Errorf("beginning transaction: %w", err)
 	}
 	defer tx.Rollback(ctx) //nolint:errcheck // no-op once committed
 
-	notify, err := fn(ctx, tx)
+	changed, err := fn(ctx, tx)
 	if err != nil {
 		return err
 	}
 
-	if notify {
+	if changed {
 		payload, err := marshalWorkerEvent(eventType, worker)
 		if err != nil {
 			return fmt.Errorf("marshaling worker event: %w", err)
 		}
-		if len(payload) > maxNotifyPayloadBytes {
-			return fmt.Errorf("worker event payload of %d bytes exceeds PostgreSQL NOTIFY limit of %d bytes", len(payload), maxNotifyPayloadBytes)
-		}
-		if _, err := tx.Exec(ctx, `SELECT pg_notify($1, $2)`, workerChangeChannel, string(payload)); err != nil {
-			return fmt.Errorf("notifying worker change: %w", err)
+		if _, err := tx.Exec(ctx, `INSERT INTO worker_changes (payload) VALUES ($1)`, payload); err != nil {
+			return fmt.Errorf("appending worker change feed: %w", err)
 		}
 	}
 
@@ -1314,7 +1318,7 @@ func (p *Persistence) CreateWorker(ctx context.Context, worker *ateapipb.Worker)
 		return fmt.Errorf("marshaling worker: %w", err)
 	}
 
-	err = p.writeAndNotify(ctx, store.WorkerEventCreated, dbWorker, func(ctx context.Context, tx pgx.Tx) (bool, error) {
+	err = p.writeAndAppendChange(ctx, store.WorkerEventCreated, dbWorker, func(ctx context.Context, tx pgx.Tx) (bool, error) {
 		_, err := tx.Exec(ctx, `
 			INSERT INTO workers (worker_namespace, worker_pool, worker_pod, version, proto)
 			VALUES ($1, $2, $3, $4, $5)`,
@@ -1365,7 +1369,7 @@ func (p *Persistence) UpdateWorker(ctx context.Context, worker *ateapipb.Worker,
 		return fmt.Errorf("marshaling worker: %w", err)
 	}
 
-	return p.writeAndNotify(ctx, store.WorkerEventUpdated, dbWorker, func(ctx context.Context, tx pgx.Tx) (bool, error) {
+	return p.writeAndAppendChange(ctx, store.WorkerEventUpdated, dbWorker, func(ctx context.Context, tx pgx.Tx) (bool, error) {
 		var returned []byte
 		err := tx.QueryRow(ctx, `
 			UPDATE workers
@@ -1395,7 +1399,7 @@ func (p *Persistence) UpdateWorker(ctx context.Context, worker *ateapipb.Worker,
 
 func (p *Persistence) DeleteWorker(ctx context.Context, namespace, poolName, pod string) error {
 	deletedEvent := &ateapipb.Worker{WorkerNamespace: namespace, WorkerPod: pod}
-	return p.writeAndNotify(ctx, store.WorkerEventDeleted, deletedEvent, func(ctx context.Context, tx pgx.Tx) (bool, error) {
+	return p.writeAndAppendChange(ctx, store.WorkerEventDeleted, deletedEvent, func(ctx context.Context, tx pgx.Tx) (bool, error) {
 		var protoBytes []byte
 		err := tx.QueryRow(ctx, `
 			DELETE FROM workers
@@ -1403,7 +1407,7 @@ func (p *Persistence) DeleteWorker(ctx context.Context, namespace, poolName, pod
 			RETURNING proto`, namespace, poolName, pod).Scan(&protoBytes)
 		if err != nil {
 			if errors.Is(err, pgx.ErrNoRows) {
-				// Idempotent: nothing existed, so nothing to notify either.
+				// Idempotent: nothing existed, so no event to publish either.
 				return false, nil
 			}
 			return false, fmt.Errorf("deleting worker %s/%s/%s: %w", namespace, poolName, pod, err)
@@ -1462,47 +1466,352 @@ func (p *Persistence) ListWorkers(ctx context.Context, opts store.ListOptions) (
 	return store.ListResponse[*ateapipb.Worker]{Items: result, NextPageToken: nextToken}, nil
 }
 
-// WatchWorkers acquires a dedicated connection (hijacked out of the pool, so
-// it's never handed back for unrelated queries), LISTENs on the fixed
-// worker-change channel, and forwards decoded notifications until the
-// context is cancelled or the caller closes the watch.
-func (p *Persistence) WatchWorkers(ctx context.Context) (*store.WorkerWatch, error) {
-	watchCtx, cancel := context.WithCancel(ctx)
+const (
+	// Bound worker-event delivery latency in the absence of an xmin stall.
+	changeFeedPollInterval = 100 * time.Millisecond
+
+	// Cap rows fetched per poll; a burst beyond it carries over to the next poll 
+	// (events are delayed, never dropped).
+	changeFeedBatch = 1024
+
+	// Minimum time retention keeps feed rows.
+	changeFeedRetentionAge = 15 * time.Minute
+
+	// Paces partition maintenance.
+	changeFeedMaintenanceInterval = time.Minute
 
-	poolConn, err := p.pool.Acquire(watchCtx)
+	// The feed partition range width.
+	changeFeedPartitionInterval = 15 * time.Minute
+
+	// How many intervals ahead partitions are pre-created: creation must stall past 
+	// lead-1 intervals before any write detours into the DEFAULT partition backstop.
+	changeFeedPartitionLead = 2
+)
+
+// Maintains worker_changes partitions on a fixed timer.
+func (p *Persistence) changeFeedMaintenance(ctx context.Context) {
+	ticker := time.NewTicker(changeFeedMaintenanceInterval)
+	defer ticker.Stop()
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case <-ticker.C:
+		}
+		if err := p.maintainWorkerChangesPartitions(ctx); err != nil && ctx.Err() == nil {
+			slog.WarnContext(ctx, "worker change feed maintenance failed", slog.Any("err", err))
+		}
+	}
+}
+
+// Database-scoped advisory lock used to elect a single replica to run the retention transaction (drops + trim).
+const changeFeedMaintenanceLockKey = "atepg-change-feed-maintenance"
+
+// pollWorkerChangesSQL is the watch's batch query. The xid::text cast MUST
+// carry an alias so that ORDER BY xid sorts numerically by the table's xid8
+// column instead of alphabetically by the string output.
+const pollWorkerChangesSQL = `
+	SELECT xid::text AS xid_text, payload FROM worker_changes
+	WHERE xid > $1::xid8
+	  AND xid < pg_snapshot_xmin(pg_current_snapshot())
+	ORDER BY xid LIMIT $2`
+
+// pollSafetySQL returns cheap safety scalars fetched on every poll:
+// 1. A fell-behind check (trim mark is past both cursor and baseline).
+// 2. The postmaster start time (to detect database restarts).
+const pollSafetySQL = `
+	SELECT EXISTS(
+		SELECT 1 FROM worker_changes_trim
+		WHERE xid > $1::xid8 AND xid > $2::xid8),
+	pg_postmaster_start_time()::text`
+
+// maintainWorkerChangesPartitions is one maintenance pass. Partition
+// creation runs on every replica, unelected — it is idempotent.
+func (p *Persistence) maintainWorkerChangesPartitions(ctx context.Context) error {
+	now := time.Now().UTC()
+	if err := p.createWorkerChangesPartitions(ctx, changeFeedPartitionLeadTimes(now)...); err != nil {
+		return err
+	}
+
+	tx, err := p.pool.Begin(ctx)
 	if err != nil {
-		cancel()
-		return nil, fmt.Errorf("acquiring watch connection: %w", err)
+		return fmt.Errorf("beginning feed retention transaction: %w", err)
+	}
+	defer tx.Rollback(ctx) //nolint:errcheck // no-op once committed
+
+	var elected bool
+	if err := tx.QueryRow(ctx, `SELECT pg_try_advisory_xact_lock(hashtext(current_database() || ':' || $1))`, changeFeedMaintenanceLockKey).Scan(&elected); err != nil {
+		return fmt.Errorf("electing feed maintenance: %w", err)
+	}
+	if !elected {
+		return nil // another replica is maintaining; next tick retries
+	}
+	// A non-empty DEFAULT partition means partition creation stalled and writes
+	// detoured here. Watchers that lose events will detect the trim mark and resync.
+	var strays bool
+	if err := tx.QueryRow(ctx, `SELECT EXISTS(SELECT 1 FROM worker_changes_default)`).Scan(&strays); err != nil {
+		return fmt.Errorf("checking feed default partition: %w", err)
+	}
+	if strays {
+		slog.WarnContext(ctx, "change feed DEFAULT partition is non-empty; partition creation has stalled and writes are detouring")
+		if err := p.truncateWorkerChangesDefault(ctx, tx); err != nil {
+			return err
+		}
+	}
+	// Dropping a partition takes ACCESS EXCLUSIVE on the parent, blocking every worker write's feed append.
+	if err := p.dropExpiredWorkerChangesPartitions(ctx, tx, now); err != nil {
+		return err
+	}
+	if err := tx.Commit(ctx); err != nil {
+		return fmt.Errorf("committing feed retention transaction: %w", err)
+	}
+	return nil
+}
+
+// workerChangesPartitionName names the partition covering the given instant.
+func workerChangesPartitionName(at time.Time) string {
+	// it truncates to the partition boundary itself so callers can pass any moment within the range.
+	return "worker_changes_p" + at.UTC().Truncate(changeFeedPartitionInterval).Format("200601021504")
+}
+
+// changeFeedPartitionLeadTimes lists instants covering now through the creation lead, one per partition interval.
+func changeFeedPartitionLeadTimes(now time.Time) []time.Time {
+	times := make([]time.Time, changeFeedPartitionLead+1)
+	for i := range times {
+		times[i] = now.UTC().Add(time.Duration(i) * changeFeedPartitionInterval)
+	}
+	return times
+}
+
+// createWorkerChangesPartitions idempotently creates the feed partitions covering the given instants.
+func (p *Persistence) createWorkerChangesPartitions(ctx context.Context, instants ...time.Time) error {
+	tx, err := p.pool.Begin(ctx)
+	if err != nil {
+		return fmt.Errorf("beginning feed partition transaction: %w", err)
+	}
+	defer tx.Rollback(ctx) //nolint:errcheck // no-op once committed
+
+	if _, err := tx.Exec(ctx, `SELECT pg_advisory_xact_lock(hashtext('agent-substrate-atepg-feed-partitions'))`); err != nil {
+		return fmt.Errorf("locking feed partition DDL: %w", err)
+	}
+	for _, at := range instants {
+		start := at.UTC().Truncate(changeFeedPartitionInterval)
+		// UNLOGGED: see schema comment for the durability trade-off.
+		// autovacuum off: partitions are insert-only and discarded whole,
+		// so autovacuum is unnecessary and its scans would cause latency spikes.
+		stmt := fmt.Sprintf(`CREATE UNLOGGED TABLE IF NOT EXISTS %s PARTITION OF worker_changes FOR VALUES FROM ('%s') TO ('%s') WITH (autovacuum_enabled = off)`,
+			workerChangesPartitionName(start), start.Format(time.RFC3339), start.Add(changeFeedPartitionInterval).Format(time.RFC3339))
+		if _, err := tx.Exec(ctx, stmt); err != nil {
+			return fmt.Errorf("creating feed partition for %s: %w", start, err)
+		}
+	}
+	if err := tx.Commit(ctx); err != nil {
+		return fmt.Errorf("committing feed partition transaction: %w", err)
+	}
+	return nil
+}
+
+// dropExpiredWorkerChangesPartitions drops every feed partition whose
+// entire range is older than retention.
+func (p *Persistence) dropExpiredWorkerChangesPartitions(ctx context.Context, q querier, now time.Time) error {
+	rows, err := q.Query(ctx, `
+		SELECT c.relname FROM pg_inherits i
+		JOIN pg_class c ON c.oid = i.inhrelid
+		JOIN pg_class parent ON parent.oid = i.inhparent
+		WHERE parent.relname = 'worker_changes'`)
+	if err != nil {
+		return fmt.Errorf("listing feed partitions: %w", err)
+	}
+	var names []string
+	for rows.Next() {
+		var name string
+		if err := rows.Scan(&name); err != nil {
+			rows.Close()
+			return fmt.Errorf("scanning feed partition name: %w", err)
+		}
+		names = append(names, name)
+	}
+	rows.Close()
+	if err := rows.Err(); err != nil {
+		return fmt.Errorf("listing feed partitions: %w", err)
 	}
-	conn := poolConn.Hijack()
 
-	if _, err := conn.Exec(watchCtx, "LISTEN "+workerChangeChannel); err != nil {
-		conn.Close(watchCtx) //nolint:errcheck
+	for _, name := range names {
+		// The DEFAULT partition (worker_changes_default) doesn't match
+		// the range prefix and is skipped here naturally.
+		suffix, ok := strings.CutPrefix(name, "worker_changes_p")
+		if !ok {
+			continue
+		}
+		start, err := time.Parse("200601021504", suffix)
+		if err != nil {
+			continue // not a partition this maintenance loop manages
+		}
+		if now.Sub(start.Add(changeFeedPartitionInterval)) < changeFeedRetentionAge {
+			continue
+		}
+		if err := p.dropWorkerChangesPartition(ctx, q, name); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+// dropWorkerChangesPartition records the trim mark and drops the
+// partition on the caller's (elected, single) retention transaction.
+func (p *Persistence) dropWorkerChangesPartition(ctx context.Context, q querier, name string) error {
+	ident := pgx.Identifier{name}.Sanitize()
+	// The mark is the partition's greatest xid.
+	if _, err := q.Exec(ctx, fmt.Sprintf(`
+		INSERT INTO worker_changes_trim (xid)
+		SELECT xid FROM %s ORDER BY xid DESC LIMIT 1
+		ON CONFLICT (id) DO UPDATE SET xid = EXCLUDED.xid
+		WHERE EXCLUDED.xid > worker_changes_trim.xid`, ident)); err != nil {
+		return fmt.Errorf("recording trim mark for feed partition %s: %w", name, err)
+	}
+	if _, err := q.Exec(ctx, `DROP TABLE `+ident); err != nil {
+		return fmt.Errorf("dropping feed partition %s: %w", name, err)
+	}
+	return nil
+}
+
+// truncateWorkerChangesDefault discards the DEFAULT partition wholesale to
+// un-stall partition creation.
+func (p *Persistence) truncateWorkerChangesDefault(ctx context.Context, q querier) error {
+	// Highest xid is recorded as a trim mark in the same transaction so lagging watchers detect the loss and resync.
+	if _, err := q.Exec(ctx, `
+		INSERT INTO worker_changes_trim (xid)
+		SELECT xid FROM worker_changes_default ORDER BY xid DESC LIMIT 1
+		ON CONFLICT (id) DO UPDATE SET xid = EXCLUDED.xid
+		WHERE EXCLUDED.xid > worker_changes_trim.xid`); err != nil {
+		return fmt.Errorf("recording trim mark for feed default partition: %w", err)
+	}
+	if _, err := q.Exec(ctx, `TRUNCATE worker_changes_default`); err != nil {
+		return fmt.Errorf("truncating feed default partition: %w", err)
+	}
+	return nil
+}
+
+// WatchWorkers subscribes by polling the worker_changes feed using an xid cursor.
+// It fences reads behind pg_snapshot_xmin (the oldest in-flight transaction)
+// to guarantee gap-free delivery. Note that a long-running transaction anywhere
+// in the database will stall this feed.
+//
+// Events are delivered in xid order, so consumers must reconcile worker versions.
+// If the watcher detects missed events—either by lagging behind retention drops
+// or if a database restart truncates the UNLOGGED partitions—it closes the channel
+// to force the consumer to resync from the primary tables.
+func (p *Persistence) WatchWorkers(ctx context.Context) (*store.WorkerWatch, error) {
+	watchCtx, cancel := context.WithCancel(ctx)
+
+	// cursorXid starts at xmin - 1. The minus 1 is required because the poll
+	// query is exclusive (xid > cursor); starting exactly at xmin on an idle
+	// system would skip the very next event.
+	// baselineXid records the highest xid or trim mark at subscribe time so
+	// past garbage collection isn't mistaken for events lost during this watch.
+	// Xids are passed as decimal strings end-to-end; all ordering happens in SQL.
+	var cursorXid, baselineXid, baselineStart string
+	if err := p.pool.QueryRow(watchCtx, `
+		SELECT (pg_snapshot_xmin(pg_current_snapshot())::text::numeric - 1)::text,
+		       GREATEST(
+		           COALESCE((SELECT xid FROM worker_changes ORDER BY xid DESC LIMIT 1), '0'::xid8),
+		           COALESCE((SELECT xid FROM worker_changes_trim), '0'::xid8))::text,
+		       pg_postmaster_start_time()::text`).Scan(&cursorXid, &baselineXid, &baselineStart); err != nil {
 		cancel()
-		return nil, fmt.Errorf("listening for worker changes: %w", err)
+		return nil, fmt.Errorf("reading worker change feed cursor: %w", err)
 	}
 
 	ch := make(chan store.WorkerEvent, 128)
 	go func() {
 		defer close(ch)
-		defer conn.Close(context.Background()) //nolint:errcheck
+		ticker := time.NewTicker(changeFeedPollInterval)
+		defer ticker.Stop()
 		for {
-			notification, err := conn.WaitForNotification(watchCtx)
-			if err != nil {
-				// Context cancelled (caller closed the watch) or the
-				// connection was lost. Either way, the caller must
-				// re-subscribe; matches ateredis's WatchWorkers contract.
-				return
-			}
-			event, err := unmarshalWorkerEvent(notification.Payload)
-			if err != nil {
-				slog.ErrorContext(ctx, "worker event unmarshal failed", slog.Any("err", err))
-				continue
-			}
 			select {
-			case ch <- event:
 			case <-watchCtx.Done():
 				return
+			case <-ticker.C:
+			}
+			// Drain until a batch is partial. Sleeping between full batches would
+			// cap throughput and cause unrecoverable lag during bursts.
+			for {
+				// Safety checks share the batch round trip but must remain separate
+				// queries so we can detect gaps and restarts even when no rows match.
+				b := &pgx.Batch{}
+				b.Queue(pollWorkerChangesSQL, cursorXid, changeFeedBatch)
+				b.Queue(pollSafetySQL, cursorXid, baselineXid)
+				br := p.pool.SendBatch(watchCtx, b)
+
+				type feedRow struct {
+					xid     string
+					payload []byte
+				}
+				var batch []feedRow
+				rows, err := br.Query()
+				if err == nil {
+					for rows.Next() {
+						var r feedRow
+						if err = rows.Scan(&r.xid, &r.payload); err != nil {
+							batch = nil
+							break
+						}
+						batch = append(batch, r)
+					}
+					rows.Close()
+				}
+				var fellBehind bool
+				var pmStart string
+				if err == nil {
+					err = br.QueryRow().Scan(&fellBehind, &pmStart)
+				}
+				if closeErr := br.Close(); err == nil {
+					err = closeErr
+				}
+				if err != nil {
+					if watchCtx.Err() != nil {
+						return
+					}
+					// Transient poll failure: keep the cursor, try again on the next tick. 
+					// If the outage was a restart, the next successful safety check catches it.
+					slog.WarnContext(watchCtx, "worker change feed poll failed", slog.Any("err", err))
+					break
+				}
+				// A restarted postmaster truncated the UNLOGGED feed:
+				// committed-but-undelivered events may be gone, so close
+				// before the cursor can skip past them; consumers resync
+				// with a full relist.
+				if pmStart != baselineStart {
+					slog.WarnContext(watchCtx, "database restarted under the change feed; closing watch for resync",
+						slog.String("was", baselineStart), slog.String("now", pmStart))
+					return
+				}
+				// Retention safety: if retention's recorded trim high-water
+				// mark is ahead of everything this watcher has seen, a row
+				// it never consumed was discarded. Close before delivering
+				// anything past the gap.
+				if fellBehind {
+					slog.WarnContext(watchCtx, "worker watch fell behind change feed retention; closing for resync",
+						slog.String("cursor_xid", cursorXid))
+					return
+				}
+
+				for _, r := range batch {
+					event, err := unmarshalWorkerEvent(r.payload)
+					if err != nil {
+						slog.ErrorContext(watchCtx, "worker event unmarshal failed", slog.Any("err", err))
+						cursorXid = r.xid
+						continue
+					}
+					select {
+					case ch <- event:
+						cursorXid = r.xid
+					case <-watchCtx.Done():
+						return
+					}
+				}
+				if len(batch) < changeFeedBatch {
+					break // caught up; wait for the next tick
+				}
 			}
 		}
 	}()
@@ -1655,7 +1964,7 @@ func (p *Persistence) releaseLease(ctx context.Context, key, token string) error
 // --- Debug ---
 
 func (p *Persistence) DebugClearAll(ctx context.Context) error {
-	if _, err := p.pool.Exec(ctx, `TRUNCATE atespaces, actors, actor_templates, actor_template_versions, actor_snapshots, actor_snapshot_tags, workers, leases`); err != nil {
+	if _, err := p.pool.Exec(ctx, `TRUNCATE atespaces, actors, actor_templates, actor_template_versions, actor_snapshots, actor_snapshot_tags, workers, leases, worker_changes, worker_changes_trim`); err != nil {
 		return fmt.Errorf("truncating tables: %w", err)
 	}
 	return nil
diff --git a/cmd/ateapi/internal/store/atepg/atepg_test.go b/cmd/ateapi/internal/store/atepg/atepg_test.go
index e3f1c3b1d1..c7433aa472 100644
--- a/cmd/ateapi/internal/store/atepg/atepg_test.go
+++ b/cmd/ateapi/internal/store/atepg/atepg_test.go
@@ -19,6 +19,7 @@ import (
 	"errors"
 	"fmt"
 	"os"
+	"strings"
 	"sync"
 	"testing"
 	"time"
@@ -115,6 +116,7 @@ func setupPostgresPersistence(t *testing.T) *Persistence {
 	if err != nil {
 		t.Fatalf("NewPersistence failed: %v", err)
 	}
+	t.Cleanup(p.Close)
 	if err := p.DebugClearAll(ctx); err != nil {
 		t.Fatalf("DebugClearAll failed: %v", err)
 	}
@@ -220,10 +222,11 @@ func TestCreateActor_MissingAtespace_FailedPrecondition(t *testing.T) {
 	}
 }
 
-// TestWorkerNotification_OnlyAfterCommit proves the doc's atomicity claim: a
-// worker write's pg_notify shares the write's transaction, so a rolled-back
-// write never notifies, while a committed write always does.
-func TestWorkerNotification_OnlyAfterCommit(t *testing.T) {
+// TestWorkerEvent_OnlyAfterCommit proves the doc's atomicity claim: a
+// worker write's change-feed insert shares the write's transaction, so a
+// rolled-back write never produces an event, while a committed write always
+// does.
+func TestWorkerEvent_OnlyAfterCommit(t *testing.T) {
 	s := setupPostgresStore(t).(*Persistence)
 	ctx := context.Background()
 
@@ -239,9 +242,9 @@ func TestWorkerNotification_OnlyAfterCommit(t *testing.T) {
 		t.Fatalf("marshaling worker: %v", err)
 	}
 
-	// Write the row and roll back instead of committing: no notification
-	// should ever arrive, proving pg_notify's effect is undone with the rest
-	// of the transaction.
+	// Write the row and roll back instead of committing: no event should
+	// ever arrive, proving the feed insert is undone with the rest of the
+	// transaction.
 	tx, err := s.pool.Begin(ctx)
 	if err != nil {
 		t.Fatalf("Begin failed: %v", err)
@@ -252,8 +255,8 @@ func TestWorkerNotification_OnlyAfterCommit(t *testing.T) {
 		worker.GetWorkerNamespace(), worker.GetWorkerPool(), worker.GetWorkerPod(), int64(1), protoBytes); err != nil {
 		t.Fatalf("insert failed: %v", err)
 	}
-	if _, err := tx.Exec(ctx, `SELECT pg_notify($1, $2)`, workerChangeChannel, "rolled-back-payload"); err != nil {
-		t.Fatalf("pg_notify failed: %v", err)
+	if _, err := tx.Exec(ctx, `INSERT INTO worker_changes (payload) VALUES ($1)`, []byte("rolled-back-payload")); err != nil {
+		t.Fatalf("feed insert failed: %v", err)
 	}
 	if err := tx.Rollback(ctx); err != nil {
 		t.Fatalf("Rollback failed: %v", err)
@@ -261,12 +264,12 @@ func TestWorkerNotification_OnlyAfterCommit(t *testing.T) {
 
 	select {
 	case event := <-watch.Events:
-		t.Fatalf("received event %+v from a rolled-back transaction; NOTIFY should not survive rollback", event)
+		t.Fatalf("received event %+v from a rolled-back transaction; the feed insert must be undone with the rest of the transaction", event)
 	case <-time.After(500 * time.Millisecond):
 		// Expected: nothing arrives.
 	}
 
-	// The equivalent committed write must notify.
+	// The equivalent committed write must produce an event.
 	if err := s.CreateWorker(ctx, worker); err != nil {
 		t.Fatalf("CreateWorker failed: %v", err)
 	}
@@ -284,6 +287,501 @@ func TestWorkerNotification_OnlyAfterCommit(t *testing.T) {
 	}
 }
 
+// TestWatchWorkers_OutOfOrderCommitNotSkipped reproduces the commit-order
+// gap: xids are assigned at a transaction's first write but rows appear at
+// COMMIT, so a transaction holding a lower xid can commit after a
+// higher-xid sibling. A watcher that advanced past every visible row would
+// skip the in-flight one and lose its event permanently. The xmin fence
+// must instead hold the committed sibling back until the older
+// transaction resolves, then deliver both in order.
+func TestWatchWorkers_OutOfOrderCommitNotSkipped(t *testing.T) {
+	s := setupPostgresStore(t).(*Persistence)
+	ctx := context.Background()
+
+	watch, err := s.WatchWorkers(ctx)
+	if err != nil {
+		t.Fatalf("WatchWorkers failed: %v", err)
+	}
+	defer watch.Close()
+
+	mkPayload := func(pod string) []byte {
+		payload, err := marshalWorkerEvent(store.WorkerEventCreated,
+			&ateapipb.Worker{WorkerNamespace: "ns", WorkerPool: "pool", WorkerPod: pod})
+		if err != nil {
+			t.Fatalf("marshaling event for %q: %v", pod, err)
+		}
+		return payload
+	}
+
+	// tx1 appends first (lower xid) and stays open.
+	tx1, err := s.pool.Begin(ctx)
+	if err != nil {
+		t.Fatalf("Begin tx1 failed: %v", err)
+	}
+	defer tx1.Rollback(ctx) //nolint:errcheck // no-op once committed
+	if _, err := tx1.Exec(ctx, `INSERT INTO worker_changes (payload) VALUES ($1)`, mkPayload("first-xid-late-commit")); err != nil {
+		t.Fatalf("tx1 feed insert failed: %v", err)
+	}
+
+	// tx2 appends second (higher xid) and commits immediately.
+	tx2, err := s.pool.Begin(ctx)
+	if err != nil {
+		t.Fatalf("Begin tx2 failed: %v", err)
+	}
+	if _, err := tx2.Exec(ctx, `INSERT INTO worker_changes (payload) VALUES ($1)`, mkPayload("second-xid-early-commit")); err != nil {
+		t.Fatalf("tx2 feed insert failed: %v", err)
+	}
+	if err := tx2.Commit(ctx); err != nil {
+		t.Fatalf("tx2 Commit failed: %v", err)
+	}
+
+	// While tx1 is in flight, tx2's committed event must be held back by
+	// the xmin fence — otherwise the cursor has already skipped tx1's row.
+	select {
+	case event := <-watch.Events:
+		t.Fatalf("event %q delivered while an older feed transaction was still in flight; its sibling event is now unreachable", event.Worker.GetWorkerPod())
+	case <-time.After(500 * time.Millisecond):
+		// Expected: fence holds both events back.
+	}
+
+	if err := tx1.Commit(ctx); err != nil {
+		t.Fatalf("tx1 Commit failed: %v", err)
+	}
+
+	var got []string
+	for len(got) < 2 {
+		select {
+		case event := <-watch.Events:
+			got = append(got, event.Worker.GetWorkerPod())
+		case <-time.After(5 * time.Second):
+			t.Fatalf("timed out waiting for both events; delivered so far: %v", got)
+		}
+	}
+	want := []string{"first-xid-late-commit", "second-xid-early-commit"}
+	if diff := cmp.Diff(want, got); diff != "" {
+		t.Errorf("event delivery order mismatch (-want +got):\n%s", diff)
+	}
+}
+
+// TestWorkerChangesPartitionRetention verifies partition-based
+// retention: an hourly partition wholly past changeFeedRetentionAge is
+// dropped (with its greatest xid recorded in worker_changes_trim), fresh
+// rows survive, and aged strays in the DEFAULT partition are trimmed by the
+// row-wise fallback.
+func TestWorkerChangesPartitionRetention(t *testing.T) {
+	s := setupPostgresPersistence(t)
+	ctx := context.Background()
+
+	// A partition two hours back, holding one aged event.
+	stale := time.Now().UTC().Add(-2 * time.Hour)
+	if err := s.createWorkerChangesPartitions(ctx, stale); err != nil {
+		t.Fatalf("creating stale partition failed: %v", err)
+	}
+	var staleXid string
+	if err := s.pool.QueryRow(ctx, `INSERT INTO worker_changes (payload, created_at) VALUES ($1, $2) RETURNING xid::text`,
+		[]byte("old"), stale).Scan(&staleXid); err != nil {
+		t.Fatalf("inserting aged row failed: %v", err)
+	}
+	// An aged stray in the DEFAULT partition (no hourly partition covers a
+	// day ago), and a fresh row in the current partition.
+	if _, err := s.pool.Exec(ctx, `INSERT INTO worker_changes (payload, created_at) VALUES ($1, now() - interval '1 day')`, []byte("stray")); err != nil {
+		t.Fatalf("inserting default-partition stray failed: %v", err)
+	}
+	if _, err := s.pool.Exec(ctx, `INSERT INTO worker_changes (payload) VALUES ($1)`, []byte("fresh")); err != nil {
+		t.Fatalf("inserting fresh row failed: %v", err)
+	}
+
+	if err := s.maintainWorkerChangesPartitions(ctx); err != nil {
+		t.Fatalf("maintainWorkerChangesPartitions failed: %v", err)
+	}
+
+	var staleExists bool
+	if err := s.pool.QueryRow(ctx, `SELECT to_regclass($1) IS NOT NULL`,
+		workerChangesPartitionName(stale)).Scan(&staleExists); err != nil {
+		t.Fatalf("checking stale partition failed: %v", err)
+	}
+	if staleExists {
+		t.Errorf("stale partition %s still exists, want dropped", workerChangesPartitionName(stale))
+	}
+	var remaining int
+	if err := s.pool.QueryRow(ctx, `SELECT count(*) FROM worker_changes`).Scan(&remaining); err != nil {
+		t.Fatalf("counting remaining rows failed: %v", err)
+	}
+	if remaining != 1 {
+		t.Errorf("%d rows remain, want 1 (the fresh row; aged partition row and default stray gone)", remaining)
+	}
+	var trim bool
+	if err := s.pool.QueryRow(ctx, `SELECT (SELECT xid FROM worker_changes_trim) >= $1::xid8`, staleXid).Scan(&trim); err != nil {
+		t.Fatalf("reading trim mark failed: %v", err)
+	}
+	if !trim {
+		t.Errorf("trim mark does not cover dropped partition's xid %s", staleXid)
+	}
+}
+
+// TestChangeFeedMaintenance_SingleMaintainer verifies the retention
+// election: while another replica holds the advisory lock, a pass skips
+// retention cleanly (no error, nothing dropped); once released, the next
+// pass does the work. (Partition creation is deliberately unelected.)
+func TestChangeFeedMaintenance_SingleMaintainer(t *testing.T) {
+	s := setupPostgresPersistence(t)
+	ctx := context.Background()
+
+	stale := time.Now().UTC().Add(-2 * time.Hour)
+	if err := s.createWorkerChangesPartitions(ctx, stale); err != nil {
+		t.Fatalf("creating stale partition failed: %v", err)
+	}
+
+	// Another "replica" mid-pass: hold the advisory lock in an open
+	// transaction of our own.
+	holder, err := s.pool.Begin(ctx)
+	if err != nil {
+		t.Fatalf("Begin holder failed: %v", err)
+	}
+	defer holder.Rollback(ctx) //nolint:errcheck // released below
+	if _, err := holder.Exec(ctx, `SELECT pg_advisory_xact_lock(hashtext(current_database() || ':' || $1))`, changeFeedMaintenanceLockKey); err != nil {
+		t.Fatalf("taking maintenance lock failed: %v", err)
+	}
+
+	if err := s.maintainWorkerChangesPartitions(ctx); err != nil {
+		t.Fatalf("pass with lock held must skip cleanly, got: %v", err)
+	}
+	var staleExists bool
+	if err := s.pool.QueryRow(ctx, `SELECT to_regclass($1) IS NOT NULL`, workerChangesPartitionName(stale)).Scan(&staleExists); err != nil {
+		t.Fatalf("checking stale partition failed: %v", err)
+	}
+	if !staleExists {
+		t.Fatal("stale partition was dropped by a pass that lost the election")
+	}
+
+	if err := holder.Rollback(ctx); err != nil {
+		t.Fatalf("releasing maintenance lock failed: %v", err)
+	}
+	if err := s.maintainWorkerChangesPartitions(ctx); err != nil {
+		t.Fatalf("pass after lock release failed: %v", err)
+	}
+	if err := s.pool.QueryRow(ctx, `SELECT to_regclass($1) IS NOT NULL`, workerChangesPartitionName(stale)).Scan(&staleExists); err != nil {
+		t.Fatalf("re-checking stale partition failed: %v", err)
+	}
+	if staleExists {
+		t.Error("stale partition survived a pass that held the election")
+	}
+}
+
+// TestWorkerEvents_OneRowPerTransaction pins the invariant the xid-only
+// watch cursor rests on: writeAndAppendChange appends exactly one feed row
+// per transaction, so xids are distinct across the feed and a poll batch
+// can never split a same-xid group.
+func TestWorkerEvents_OneRowPerTransaction(t *testing.T) {
+	s := setupPostgresPersistence(t)
+	ctx := context.Background()
+
+	worker := &ateapipb.Worker{WorkerNamespace: "ns", WorkerPool: "pool", WorkerPod: "pod"}
+	if err := s.CreateWorker(ctx, worker); err != nil {
+		t.Fatalf("CreateWorker failed: %v", err)
+	}
+	for i := 0; i < 10; i++ {
+		stored, err := s.GetWorker(ctx, "ns", "pool", "pod")
+		if err != nil {
+			t.Fatalf("GetWorker failed: %v", err)
+		}
+		if err := s.UpdateWorker(ctx, stored, stored.GetVersion()); err != nil {
+			t.Fatalf("UpdateWorker %d failed: %v", i, err)
+		}
+	}
+	if err := s.DeleteWorker(ctx, "ns", "pool", "pod"); err != nil {
+		t.Fatalf("DeleteWorker failed: %v", err)
+	}
+
+	var total, distinct int
+	if err := s.pool.QueryRow(ctx, `SELECT count(*), count(DISTINCT xid) FROM worker_changes`).Scan(&total, &distinct); err != nil {
+		t.Fatalf("counting feed rows failed: %v", err)
+	}
+	if total == 0 || total != distinct {
+		t.Errorf("feed has %d rows but %d distinct xids; the one-row-per-transaction invariant is broken", total, distinct)
+	}
+}
+
+// TestWatchWorkers_DeliveryFencedByOldestTransaction documents the xmin
+// fence's real bound: one old transaction anywhere holds back delivery of
+// everything committed after it, for as long as it lives.
+func TestWatchWorkers_DeliveryFencedByOldestTransaction(t *testing.T) {
+	s := setupPostgresPersistence(t)
+	ctx := context.Background()
+
+	watch, err := s.WatchWorkers(ctx)
+	if err != nil {
+		t.Fatalf("WatchWorkers failed: %v", err)
+	}
+	defer watch.Close()
+
+	// An unrelated transaction that merely holds an xid.
+	blocker, err := s.pool.Begin(ctx)
+	if err != nil {
+		t.Fatalf("Begin blocker failed: %v", err)
+	}
+	defer blocker.Rollback(ctx) //nolint:errcheck // released below
+	if _, err := blocker.Exec(ctx, `SELECT pg_current_xact_id()`); err != nil {
+		t.Fatalf("assigning blocker xid failed: %v", err)
+	}
+
+	worker := &ateapipb.Worker{WorkerNamespace: "ns", WorkerPool: "pool", WorkerPod: "fenced"}
+	if err := s.CreateWorker(ctx, worker); err != nil {
+		t.Fatalf("CreateWorker failed: %v", err)
+	}
+
+	select {
+	case event := <-watch.Events:
+		t.Fatalf("event %+v delivered through the fence while an older transaction was in flight", event)
+	case <-time.After(600 * time.Millisecond):
+		// Expected: committed but fenced behind the blocker's xid.
+	}
+
+	if err := blocker.Rollback(ctx); err != nil {
+		t.Fatalf("ending blocker failed: %v", err)
+	}
+	select {
+	case event := <-watch.Events:
+		if got := event.Worker.GetWorkerPod(); got != "fenced" {
+			t.Errorf("delivered %q, want %q", got, "fenced")
+		}
+	case <-time.After(5 * time.Second):
+		t.Fatal("event not delivered after the fencing transaction ended")
+	}
+}
+
+// TestClose_StopsMaintenance pins that Close ends the background
+// maintenance goroutine (main.go defers it for exactly this): Close blocks
+// on the loop's done channel, so its return IS the assertion.
+func TestClose_StopsMaintenance(t *testing.T) {
+	ctx := context.Background()
+	p, err := NewPersistence(ctx, requirePool(t))
+	if err != nil {
+		t.Fatalf("NewPersistence failed: %v", err)
+	}
+	closed := make(chan struct{})
+	go func() {
+		p.Close()
+		close(closed)
+	}()
+	select {
+	case <-closed:
+	case <-time.After(5 * time.Second):
+		t.Fatal("Close did not stop the maintenance loop")
+	}
+}
+
+// TestPollQueryPlanStaysOnIndex pins the poll's plan shape against the
+// output-column shadowing bug: an unaliased xid::text captures the bare
+// ORDER BY name, sorting xids as text — which both diverges from the
+// cursor predicate's xid8 order (silently skipping events across digit
+// boundaries) and forces full scans with a top-N sort. Behavioural tests
+// cannot see this; the plan can.
+func TestPollQueryPlanStaysOnIndex(t *testing.T) {
+	s := setupPostgresPersistence(t)
+	ctx := context.Background()
+
+	// Seed enough rows (and stats) for the planner to have a real choice:
+	// on empty partitions it costs bitmap scans plus an explicit Sort as
+	// cheapest regardless of the index, which would make the Merge Append
+	// assertion below vacuously unreachable.
+	if _, err := s.pool.Exec(ctx, `INSERT INTO worker_changes (payload) SELECT 'x'::bytea FROM generate_series(1, 3000)`); err != nil {
+		t.Fatalf("seeding feed rows failed: %v", err)
+	}
+	if _, err := s.pool.Exec(ctx, `ANALYZE worker_changes`); err != nil {
+		t.Fatalf("ANALYZE failed: %v", err)
+	}
+
+	rows, err := s.pool.Query(ctx, "EXPLAIN "+pollWorkerChangesSQL, "100", changeFeedBatch)
+	if err != nil {
+		t.Fatalf("EXPLAIN failed: %v", err)
+	}
+	defer rows.Close()
+	var plan strings.Builder
+	for rows.Next() {
+		var line string
+		if err := rows.Scan(&line); err != nil {
+			t.Fatalf("scanning plan line: %v", err)
+		}
+		plan.WriteString(line)
+		plan.WriteString("\n")
+	}
+	got := plan.String()
+	if strings.Contains(got, "::text") {
+		t.Errorf("poll plan sorts by a text expression (output-column shadowing is back):\n%s", got)
+	}
+	if !strings.Contains(got, "Merge Append") {
+		t.Errorf("poll plan is not an index-ordered Merge Append:\n%s", got)
+	}
+}
+
+// TestChangeFeedMaintenance_ConcurrentPassesAreHarmless backs the doc's
+// claim: two replicas racing a maintenance pass produce no errors and the
+// correct end state (one wins the election, the loser skips).
+func TestChangeFeedMaintenance_ConcurrentPassesAreHarmless(t *testing.T) {
+	s := setupPostgresPersistence(t)
+	ctx := context.Background()
+
+	replica, err := NewPersistence(ctx, s.pool)
+	if err != nil {
+		t.Fatalf("second Persistence failed: %v", err)
+	}
+	t.Cleanup(replica.Close)
+
+	stale := time.Now().UTC().Add(-2 * time.Hour)
+	if err := s.createWorkerChangesPartitions(ctx, stale); err != nil {
+		t.Fatalf("creating stale partition failed: %v", err)
+	}
+
+	var wg sync.WaitGroup
+	errs := make([]error, 2)
+	for i, p := range []*Persistence{s, replica} {
+		wg.Add(1)
+		go func(i int, p *Persistence) {
+			defer wg.Done()
+			errs[i] = p.maintainWorkerChangesPartitions(ctx)
+		}(i, p)
+	}
+	wg.Wait()
+	for i, err := range errs {
+		if err != nil {
+			t.Errorf("concurrent pass %d returned error: %v", i, err)
+		}
+	}
+	var staleExists bool
+	if err := s.pool.QueryRow(ctx, `SELECT to_regclass($1) IS NOT NULL`, workerChangesPartitionName(stale)).Scan(&staleExists); err != nil {
+		t.Fatalf("checking stale partition failed: %v", err)
+	}
+	if staleExists {
+		t.Error("stale partition survived both concurrent passes")
+	}
+}
+
+// TestWorkerChangesPartitionsAreUnlogged pins the maintenance profile the
+// schema documents: every feed partition must be UNLOGGED (relpersistence
+// 'u') with autovacuum disabled (all are insert-only and discarded whole,
+// by drop or truncate — an in-window insert-autovacuum is a measured p99
+// spike); and worker_changes_trim — the loss-detection high-water mark —
+// must remain logged so it survives a crash.
+func TestWorkerChangesPartitionsAreUnlogged(t *testing.T) {
+	s := setupPostgresPersistence(t)
+	ctx := context.Background()
+
+	rows, err := s.pool.Query(ctx, `
+		SELECT c.relname, c.relpersistence, COALESCE(array_to_string(c.reloptions, ','), '') FROM pg_inherits i
+		JOIN pg_class c ON c.oid = i.inhrelid
+		JOIN pg_class parent ON parent.oid = i.inhparent
+		WHERE parent.relname = 'worker_changes'`)
+	if err != nil {
+		t.Fatalf("listing feed partitions: %v", err)
+	}
+	defer rows.Close()
+	checked := 0
+	for rows.Next() {
+		var name, persistence, options string
+		if err := rows.Scan(&name, &persistence, &options); err != nil {
+			t.Fatalf("scanning partition row: %v", err)
+		}
+		if persistence != "u" {
+			t.Errorf("partition %s has relpersistence %q, want 'u' (unlogged)", name, persistence)
+		}
+		if !strings.Contains(options, "autovacuum_enabled=off") {
+			t.Errorf("partition %s does not disable autovacuum (reloptions %q)", name, options)
+		}
+		checked++
+	}
+	if checked == 0 {
+		t.Fatal("no feed partitions found to check")
+	}
+	var trimPersistence string
+	if err := s.pool.QueryRow(ctx, `SELECT relpersistence FROM pg_class WHERE relname = 'worker_changes_trim'`).Scan(&trimPersistence); err != nil {
+		t.Fatalf("checking worker_changes_trim persistence: %v", err)
+	}
+	if trimPersistence != "p" {
+		t.Errorf("worker_changes_trim has relpersistence %q, want 'p' (logged) — the trim mark must survive a crash", trimPersistence)
+	}
+}
+
+// The restart escape hatch (a changed pg_postmaster_start_time() closes
+// the watch, because a restart truncates the UNLOGGED feed) has no e2e
+// test here: restarting the testcontainer remaps its host port, severing
+// the pool permanently — unlike production, where the database endpoint is
+// stable across restarts. The comparison itself is four lines in
+// WatchWorkers' poll loop; the trimmed-past-cursor test below covers the
+// shared close-for-resync path.
+
+// TestWatchWorkers_ClosesWhenTrimmedPastCursor verifies the retention
+// escape hatch: when rows a watcher has not consumed are deleted out from
+// under it (a retention trim on a badly lagging watcher), the watcher must
+// close its channel — the signal consumers treat as resync-and-relist —
+// rather than silently skip the gap.
+func TestWatchWorkers_ClosesWhenTrimmedPastCursor(t *testing.T) {
+	s := setupPostgresPersistence(t)
+	ctx := context.Background()
+
+	watch, err := s.WatchWorkers(ctx)
+	if err != nil {
+		t.Fatalf("WatchWorkers failed: %v", err)
+	}
+	defer watch.Close()
+
+	// Deliver one event normally so the cursor is established.
+	worker := &ateapipb.Worker{WorkerNamespace: "ns", WorkerPool: "pool", WorkerPod: "pod"}
+	if err := s.CreateWorker(ctx, worker); err != nil {
+		t.Fatalf("CreateWorker failed: %v", err)
+	}
+	select {
+	case <-watch.Events:
+	case <-time.After(2 * time.Second):
+		t.Fatal("timed out waiting for the first event")
+	}
+
+	// Atomically append three events and trim them away unconsumed —
+	// the watcher never gets a chance to see them, exactly as if
+	// retention took rows a lagging watcher had not reached.
+	payload, err := marshalWorkerEvent(store.WorkerEventUpdated, worker)
+	if err != nil {
+		t.Fatalf("marshaling event: %v", err)
+	}
+	tx, err := s.pool.Begin(ctx)
+	if err != nil {
+		t.Fatalf("Begin failed: %v", err)
+	}
+	defer tx.Rollback(ctx) //nolint:errcheck // no-op once committed
+	if _, err := tx.Exec(ctx, `INSERT INTO worker_changes (payload) VALUES ($1), ($1), ($1)`, payload); err != nil {
+		t.Fatalf("feed inserts failed: %v", err)
+	}
+	// Mirrors trimWorkerChangesDefault's shape: the mark is the deleted
+	// set's greatest xid. (The three rows above share one transaction —
+	// fine here: this test only needs the recorded mark to land past the
+	// watcher's cursor, and deletes everything the watcher has not seen.)
+	if _, err := tx.Exec(ctx, `
+		WITH doomed AS (
+			DELETE FROM worker_changes WHERE xid > (SELECT COALESCE((SELECT xid FROM worker_changes_trim), '0'::xid8))
+			RETURNING xid
+		)
+		INSERT INTO worker_changes_trim (xid)
+		SELECT xid FROM doomed ORDER BY xid DESC LIMIT 1
+		ON CONFLICT (id) DO UPDATE SET xid = EXCLUDED.xid
+		WHERE EXCLUDED.xid > worker_changes_trim.xid`); err != nil {
+		t.Fatalf("trim failed: %v", err)
+	}
+	if err := tx.Commit(ctx); err != nil {
+		t.Fatalf("Commit failed: %v", err)
+	}
+
+	// The watcher must close the channel, not deliver past the gap.
+	select {
+	case event, ok := <-watch.Events:
+		if ok {
+			t.Fatalf("received event %+v past a trimmed gap; expected the channel to close for resync", event)
+		}
+		// Expected: channel closed.
+	case <-time.After(5 * time.Second):
+		t.Fatal("timed out waiting for the watch channel to close after a trim past the cursor")
+	}
+}
+
 func TestListActors_InvalidPageToken(t *testing.T) {
 	s := setupPostgresStore(t).(*Persistence)
 	ctx := context.Background()
diff --git a/cmd/ateapi/internal/store/atepg/schema.go b/cmd/ateapi/internal/store/atepg/schema.go
index f4f0ce0ad1..abf77c3d37 100644
--- a/cmd/ateapi/internal/store/atepg/schema.go
+++ b/cmd/ateapi/internal/store/atepg/schema.go
@@ -95,6 +95,56 @@ CREATE TABLE IF NOT EXISTS workers (
     PRIMARY KEY (worker_namespace, worker_pool, worker_pod)
 );
 
+-- Transactional change feed backing WatchWorkers. Events are appended in
+-- the same transaction as the worker write and delivered by polling past
+-- an xid cursor;
+-- payload is one event-type byte followed by the binary Worker proto.
+--
+-- xid is the whole ordering: writeAndAppendChange appends exactly ONE row
+-- per transaction (the only insert site), so every feed row has a distinct
+-- xid and a poll batch can never split a same-xid group. That invariant is
+-- load-bearing for the watch cursor and pinned by a test.
+--
+-- Partitioned by created_at range (width: changeFeedPartitionInterval,
+-- kept <= retention so rows outlive it by at most one interval) so
+-- retention is a partition DROP — a metadata operation with no row
+-- deletes, dead tuples, or vacuum debt — instead of bulk DELETEs whose I/O
+-- competes with foreground traffic. The maintenance loop
+-- (changeFeedMaintenance) creates upcoming partitions and drops expired
+-- ones; the DEFAULT partition only receives writes if partition creation
+-- ever stalls, and is truncated wholesale once maintenance notices
+-- (watchers that lose events to that resync via the trim mark).
+--
+-- Partitions are UNLOGGED: the feed is ephemeral by design (cursors are
+-- not durable, subscriptions start "from now", and every consumer rebuilds
+-- from the workers table on resync), so paying WAL on every event — inside
+-- every worker-write transaction — buys nothing. Crash/failover truncates
+-- unlogged tables; see WatchWorkers for how watchers recover.
+-- worker_changes_trim stays logged — the trim mark must survive a crash.
+CREATE TABLE IF NOT EXISTS worker_changes (
+    xid         xid8 NOT NULL DEFAULT pg_current_xact_id(),
+    -- clock_timestamp(), not now(): now() is transaction-START time, so a
+    -- slow transaction would route its event by a stale timestamp — worst
+    -- case into an already-dropped partition (the DEFAULT partition would
+    -- catch it). The feed insert is the last statement before commit, so
+    -- statement time routes into the partition closest to commit time.
+    created_at  timestamptz NOT NULL DEFAULT clock_timestamp(),
+    payload     bytea NOT NULL
+) PARTITION BY RANGE (created_at);
+
+CREATE INDEX IF NOT EXISTS worker_changes_xid ON worker_changes (xid);
+
+CREATE UNLOGGED TABLE IF NOT EXISTS worker_changes_default PARTITION OF worker_changes DEFAULT WITH (autovacuum_enabled = off);
+
+-- Single-row high-water mark of retention: the greatest xid ever discarded
+-- from worker_changes (dropped with an expired partition, or truncated
+-- with the DEFAULT partition). Watchers compare it against their cursor to
+-- detect exactly that unconsumed rows were discarded out from under them.
+CREATE TABLE IF NOT EXISTS worker_changes_trim (
+    id   boolean PRIMARY KEY DEFAULT true CHECK (id),
+    xid  xid8 NOT NULL
+);
+
 CREATE TABLE IF NOT EXISTS leases (
     key         text PRIMARY KEY,
     token       text NOT NULL,
@@ -110,6 +160,17 @@ func applySchema(ctx context.Context, pool *pgxpool.Pool) error {
 	}
 	defer tx.Rollback(ctx) //nolint:errcheck // no-op once committed
 
+	// The schema needs PostgreSQL 13+ (xid8, pg_current_xact_id,
+	// pg_current_snapshot); fail with a clear message rather than an
+	// opaque DDL or function error.
+	var version int
+	if err := tx.QueryRow(ctx, `SELECT current_setting('server_version_num')::int`).Scan(&version); err != nil {
+		return fmt.Errorf("reading PostgreSQL version: %w", err)
+	}
+	if version < 130000 {
+		return fmt.Errorf("atepg requires PostgreSQL 13 or newer (xid8 and pg_current_snapshot); server_version_num is %d", version)
+	}
+
 	// Multiple ateapi replicas can start against an empty database together.
 	// PostgreSQL's IF NOT EXISTS does not eliminate every concurrent-DDL race,
 	// so serialize schema application with a transaction-scoped advisory lock.
diff --git a/cmd/ateapi/main.go b/cmd/ateapi/main.go
index 5d1d3e0d5a..63ca29c2c4 100644
--- a/cmd/ateapi/main.go
+++ b/cmd/ateapi/main.go
@@ -137,6 +137,11 @@ func main() {
 	if err != nil {
 		serverboot.Fatal(ctx, "Failed to set up persistence backend", err)
 	}
+	// Backends may run background maintenance rooted in their own context
+	// (atepg's change-feed maintenance loop); stop it on shutdown.
+	if closer, ok := persistence.(interface{ Close() }); ok {
+		defer closer.Close()
+	}
 
 	clientset, ateClient, err := newKubeClients()
 	if err != nil {