feat(batch): add opt-in dedicated projection drain pump

backend/config/batch_pipeline.py

@@ -1,4 +1,4 @@
 """Default-off batch admission cohort. Existing lane classification stays upstream."""
 
 from config.refactor_switches import default_off as rollout_default_off, default_on as rollout_default_on
 
@@ -573,6 +573,39 @@
     return kill_switch('BATCH_PIPELINE_OWNER_DRAIN_ENABLED')
 
 
+def serving_projection_sweep_enabled() -> bool:
+    """Only ``off`` disables serving-instance discovery; owner wakes still run."""
+    from config.policy import kill_switch
+
+    return kill_switch('BATCH_PIPELINE_SERVING_PROJECTION_SWEEP')
+
+
+def _pump_bound(raw: str, default: int, maximum: int) -> int:
+    try:
+        return max(1, min(int(raw), maximum))
+    except ValueError:
+        return default
+
+
+def projection_pump_enabled() -> bool:
+    """Process-role opt-in: only ``true`` starts the otherwise inert pump."""
+    from config.policy import kill_switch
+
+    return kill_switch('BATCH_PIPELINE_PROJECTION_PUMP_ENABLED')
+
+
+def projection_pump_tick_seconds() -> int:
+    return _pump_bound(os.getenv('BATCH_PIPELINE_PUMP_TICK_SECONDS', '20'), 20, 60)
+
+
+def projection_pump_sweep_pages() -> int:
+    return _pump_bound(os.getenv('BATCH_PIPELINE_PUMP_SWEEP_PAGES', '1'), 1, 10)
+
+
+def max_owner_drains() -> int:
+    return _pump_bound(os.getenv('BATCH_PIPELINE_MAX_OWNER_DRAINS', '8'), 8, 32)
+
+
 def projection_index_deferred_enabled() -> bool:
     """Kill switch: walks converge search/people-stats after the chain, once per conversation."""
     from config.policy import kill_switch

backend/config/flag_budget.json

@@ -24,8 +24,13 @@
   },
   "seed_review_by": "2026-11-15",
   "budget": 169,
-  "count": 248,
+  "count": 253,
   "names": {
+    "BATCH_PIPELINE_SERVING_PROJECTION_SWEEP": {"owner": "omi-pipeline / lead2", "review_by": "2026-12-05", "registry": "config/feature-flags.yaml"},
+    "BATCH_PIPELINE_PROJECTION_PUMP_ENABLED": {"owner": "omi-pipeline / lead2", "review_by": "2026-12-05", "registry": "config/feature-flags.yaml"},
+    "BATCH_PIPELINE_PUMP_SWEEP_PAGES": {"owner": "omi-pipeline / lead2", "review_by": "2026-12-05"},
+    "BATCH_PIPELINE_PUMP_TICK_SECONDS": {"owner": "omi-pipeline / lead2", "review_by": "2026-12-05"},
+    "BATCH_PIPELINE_MAX_OWNER_DRAINS": {"owner": "omi-pipeline / lead2", "review_by": "2026-12-05"},
     "PARAKEET_SHAPE_CANARY_FILE": {"owner": "two-lane lead", "review_by": "2026-12-04"},
     "PARAKEET_LID_MODEL": {"owner": "planscost", "review_by": "2026-11-30"},
     "PARAKEET_LID_ONLY": {"owner": "planscost", "review_by": "2026-11-30"},

backend/config/policy/kill_switches.py

@@ -19,6 +19,8 @@
     'BATCH_PIPELINE_PROJECTION_PENDING_PAGE_ENABLED': True,
     'BATCH_PIPELINE_PROJECTION_FOLLOWUP_PAGE_ENABLED': True,
     'BATCH_PIPELINE_OWNER_DRAIN_ENABLED': True,
+    'BATCH_PIPELINE_SERVING_PROJECTION_SWEEP': True,
+    'BATCH_PIPELINE_PROJECTION_PUMP_ENABLED': False,
     'BATCH_PIPELINE_PROJECTION_INDEX_DEFERRED_ENABLED': True,
     'PARAKEET_WINDOW_STRANDED_DRIFT_SETTLE_ENABLED': True,
     'BATCH_PIPELINE_HELD_TERMINAL_STATUS_ENABLED': True,
@@ -34,12 +36,18 @@ def kill_switch(name: str, *, policy_id: str = 'control', flags=None) -> bool:
         raise ValueError('unregistered live batch switch')
     env = os.environ if flags is None else flags
     value = env.get(name, str(KILL_SWITCHES[name])).strip().lower()
-    if name == 'PARAKEET_WINDOW_STRANDED_DRIFT_SETTLE_ENABLED':
+    if name == 'BATCH_PIPELINE_SERVING_PROJECTION_SWEEP':
+        enabled = value != 'off'
+    elif name == 'PARAKEET_WINDOW_STRANDED_DRIFT_SETTLE_ENABLED':
         enabled = value not in {'false', '0', 'off'}
     else:
         enabled = value != 'false' if name in DISABLE_ONLY_SWITCHES else value == 'true'
     from config.refactor_switches import default_on
 
-    enabled = default_on(name, enabled, environ=env)
+    # Instance selectors retain their explicit token grammar even during the
+    # shared boolean-parser migration: only off skips serving sweeps, and only
+    # true opts a process into the otherwise inert dedicated pump.
+    if name not in {'BATCH_PIPELINE_SERVING_PROJECTION_SWEEP', 'BATCH_PIPELINE_PROJECTION_PUMP_ENABLED'}:
+        enabled = default_on(name, enabled, environ=env)
     logger.info('event=batch_policy_kill_switch policy_id=%s switch=%s enabled=%s count=1', policy_id, name, enabled)
     return enabled

backend/docs/feature-flag-registry.md

@@ -528,6 +528,7 @@ and an explicit empty literal renders as `''`.
 | `BATCH_PIPELINE_LANGUAGE_INHERITANCE_ENABLED` | Default on unless false; batch pieces inherit a known bound conversation language, else the user language, before per-piece LID | backend | env | open | — | — | — | — | keep | keep | — | dazheng |
 | `BATCH_PIPELINE_LEADER_LEASE_ENABLED` | Default on unless false; one batch worker or serving replica per Firestore lease runs periodic recovery, drain, cleanup and quarantine discovery | backend | env | open | — | — | — | — | keep | keep | — | dazheng |
 | `BATCH_PIPELINE_LONG_SPEECH_HANDOFF_ENABLED` | Default on unless false; a zero-word batch unit with >=5 s worker VAD speech whose intake STT language is unknown, multi or outside Parakeet TDT v3 skips the padded Parakeet retry and goes straight to quarantine recovery | backend | env | open | — | — | — | — | keep | keep | — | dazheng |
+| `BATCH_PIPELINE_MAX_OWNER_DRAINS` | Default 8; concurrent owner drains per process, clamped to 1-32 at import; malformed keeps 8 | backend | env | closed | — | — | — | — | keep | keep | — | lead2 |
 | `BATCH_PIPELINE_MERGE_CATCH_UP_DEPTH` | Default 16; fresh batch uploads (capture ended at most 10 min before admission) commit ahead of the owner's pending catch-up rows, and catch-up commits wait while the owner already has this many committed rows awaiting projection | backend | env | closed | — | — | — | — | keep | keep | — | dazheng |
 | `BATCH_PIPELINE_MERGE_HEAD_WAIT_SECONDS` | Default 60; once a batch result has waited this long behind an unready earlier-captured merge head, the owner's ready results commit in capture order and the head appends later as late_capture | backend | env | closed | — | — | — | — | keep | keep | — | dazheng |
 | `BATCH_PIPELINE_OWNER_CHAIN_PROJECTION_ENABLED` | Default on; a settled quarantine recovery walks that owner's result log from its projection watermark in chain order and runs its contents' follow-ups first | backend | env | open | — | — | — | — | keep | keep | — | dazheng |
@@ -542,7 +543,10 @@ and an explicit empty literal renders as `''`.
 | `BATCH_PIPELINE_PROJECTION_PENDING_PAGE_ENABLED` | Default on; the projection sweep's assignment page holds only pending results and a separate persisted/ready page finishes synchronous-policy receipts | backend | env | open | — | — | — | — | keep | keep | — | dazheng |
 | `BATCH_PIPELINE_PROJECTION_PREDECESSOR_REPLAY_ENABLED` | Default on; a blocked projection successor schedules its owner's missing predecessor through the existing ordered walk | backend | env | open | — | — | — | — | keep | keep | — | dazheng |
 | `BATCH_PIPELINE_PROJECTION_PREFETCH` | Default 4; an owner-chain walk reads the next N rows' successor links, job bindings, result payloads and audio manifests ahead of the chain, and runs the candidate lookup beside context and build | backend | env | open | — | — | — | — | keep | keep | — | dazheng |
+| `BATCH_PIPELINE_PROJECTION_PUMP_ENABLED` | Default false; true selects a dedicated projection-only pump instead of the three serving loops, independently of BATCH_PIPELINE_SERVING_ENABLED | backend | env | closed | — | — | — | — | keep | keep | — | lead2 |
 | `BATCH_PIPELINE_PROJECTION_SOURCE_DELETED_ENABLED` | Default on; settle deleted original sources without a tombstone or redirect as source_deleted, without content publication | backend | env | open | — | — | — | — | keep | keep | — | cl2-cost / lead2 |
+| `BATCH_PIPELINE_PUMP_SWEEP_PAGES` | Default 1; projection pump sweep pages per tick, clamped to 1-10; malformed keeps 1 | backend | env | closed | — | — | — | — | keep | keep | — | lead2 |
+| `BATCH_PIPELINE_PUMP_TICK_SECONDS` | Default 20 seconds; projection pump tick sleep, clamped to 1-60; malformed keeps 20 | backend | env | closed | — | — | — | — | keep | keep | — | lead2 |
 | `BATCH_PIPELINE_RAMP_EXCLUDE_HEAVY_OWNERS` | Ship-on partial-ramp admission isolation; only off disables; owners over 30 open jobs or six hours pending audio use legacy | backend | env | closed | — | — | — | — | keep | keep | — | dazheng |
 | `BATCH_PIPELINE_RECOVERED_MERGE_RELEASE_ENABLED` | Default on unless false; the post-recovery owner wake commits the recovered unit, then its content's ready siblings, at once past an unready merge head (the unit itself also past the catch-up depth hold), so the same wake's chain walk projects them | backend | env | open | — | — | — | — | keep | keep | — | dazheng |
 | `BATCH_PIPELINE_RECOVERY_DISPATCH_INTERVAL_SECONDS` | Default 5; batch serving dispatches quarantine recovery on its own loop at this interval, on the maintenance leader only | backend | env | closed | — | — | — | — | keep | keep | — | dazheng |
@@ -552,6 +556,7 @@ and an explicit empty literal renders as `''`.
 | `BATCH_PIPELINE_RECOVERY_WAKE_LEASE_SECONDS` | One outstanding quarantine-recovery delivery task per paid intent; lease before a lost wake is repaired; default 600, 0 = legacy 120s wake on every sweep | backend | env | closed | — | — | — | — | keep | keep | — | dazheng |
 | `BATCH_PIPELINE_SECOND_MULTILINGUAL_ENABLED` | Ship-on Soniox second chance after valid-empty Modulate held recovery; false restores Modulate-only ordinary recovery; independent language-guard recovery retains its separate switch and cap | backend | env | open | — | — | — | — | keep | keep | — | dazheng |
 | `BATCH_PIPELINE_SERVING_METRICS_MIN_REFRESH_SECONDS` | Default 30; each batch serving instance refreshes its queue gauges at most once per interval | backend | env | closed | — | — | 60 | — | keep | keep | — | dazheng |
+| `BATCH_PIPELINE_SERVING_PROJECTION_SWEEP` | Default on; only off skips Serving.reconcile projection sweep, preserving owner wakes, held drains, metrics and leader-leased maintenance | backend | env | open | — | — | — | — | keep | keep | — | lead2 |
 | `BATCH_PIPELINE_SHARED_EMBEDDING_TRUNK_ENABLED` | Default on unless false; the serial batch GPU chain runs the WeSpeaker embedding trunk once per 10 s chunk of a unit and pools it per local speaker, instead of once per (chunk, speaker) pair | backend | env | open | — | — | — | — | keep | keep | — | dazheng |
 | `BATCH_PIPELINE_SWEEP_OWNER_CHAIN_ENABLED` | Default on; when the periodic projection sweep advances an owner's chain head it keeps walking that owner's result log in chain order instead of waiting for its cursor to wrap | backend | env | open | — | — | — | — | keep | keep | — | dazheng |
 | `BATCH_PIPELINE_VAD_SEGMENTATION_REUSE_ENABLED` | Default on unless false; on the serial batch GPU chain a unit diarized at full length takes the raw segmentation scores its own VAD pass computed instead of a second segmentation forward | backend | env | open | — | — | — | — | keep | keep | — | dazheng |
@@ -1084,6 +1089,7 @@ their code default (`fail` tells you which way a missing value resolves).
 - `BATCH_PIPELINE_LANGUAGE_INHERITANCE_ENABLED` — Default on unless false; batch pieces inherit a known bound conversation language, else the user language, before per-piece LID (fail: open)
 - `BATCH_PIPELINE_LEADER_LEASE_ENABLED` — Default on unless false; one batch worker or serving replica per Firestore lease runs periodic recovery, drain, cleanup and quarantine discovery (fail: open)
 - `BATCH_PIPELINE_LONG_SPEECH_HANDOFF_ENABLED` — Default on unless false; a zero-word batch unit with >=5 s worker VAD speech whose intake STT language is unknown, multi or outside Parakeet TDT v3 skips the padded Parakeet retry and goes straight to quarantine recovery (fail: open)
+- `BATCH_PIPELINE_MAX_OWNER_DRAINS` — Default 8; concurrent owner drains per process, clamped to 1-32 at import; malformed keeps 8 (fail: closed)
 - `BATCH_PIPELINE_MERGE_CATCH_UP_DEPTH` — Default 16; fresh batch uploads (capture ended at most 10 min before admission) commit ahead of the owner's pending catch-up rows, and catch-up commits wait while the owner already has this many committed rows awaiting projection (fail: closed)
 - `BATCH_PIPELINE_MERGE_HEAD_WAIT_SECONDS` — Default 60; once a batch result has waited this long behind an unready earlier-captured merge head, the owner's ready results commit in capture order and the head appends later as late_capture (fail: closed)
 - `BATCH_PIPELINE_MODEL_BATCH_SIZE` — GPU chain maximum model window batch size; default 8; hard ceiling 32 (fail: closed)
@@ -1101,8 +1107,11 @@ their code default (`fail` tells you which way a missing value resolves).
 - `BATCH_PIPELINE_PROJECTION_PENDING_PAGE_ENABLED` — Default on; the projection sweep's assignment page holds only pending results and a separate persisted/ready page finishes synchronous-policy receipts (fail: open)
 - `BATCH_PIPELINE_PROJECTION_PREDECESSOR_REPLAY_ENABLED` — Default on; a blocked projection successor schedules its owner's missing predecessor through the existing ordered walk (fail: open)
 - `BATCH_PIPELINE_PROJECTION_PREFETCH` — Default 4; an owner-chain walk reads the next N rows' successor links, job bindings, result payloads and audio manifests ahead of the chain, and runs the candidate lookup beside context and build (fail: open)
+- `BATCH_PIPELINE_PROJECTION_PUMP_ENABLED` — Default false; true selects a dedicated projection-only pump instead of the three serving loops, independently of BATCH_PIPELINE_SERVING_ENABLED (fail: closed)
 - `BATCH_PIPELINE_PROJECTION_SOURCE_DELETED_ENABLED` — Default on; settle deleted original sources without a tombstone or redirect as source_deleted, without content publication (fail: open)
 - `BATCH_PIPELINE_PUBLISH_CONCURRENCY` — GPU worker post-flight publication concurrency: bounded fenced transitions and one ordered drain per owner per flight; default 4, 1 = legacy serial kill switch, max 16 (fail: closed)
+- `BATCH_PIPELINE_PUMP_SWEEP_PAGES` — Default 1; projection pump sweep pages per tick, clamped to 1-10; malformed keeps 1 (fail: closed)
+- `BATCH_PIPELINE_PUMP_TICK_SECONDS` — Default 20 seconds; projection pump tick sleep, clamped to 1-60; malformed keeps 20 (fail: closed)
 - `BATCH_PIPELINE_RAMP_EXCLUDE_HEAVY_OWNERS` — Ship-on partial-ramp admission isolation; only off disables; owners over 30 open jobs or six hours pending audio use legacy (fail: closed)
 - `BATCH_PIPELINE_RECORD_PROTOCOL_READY` — One-time native record protocol cutover; default closed retains record_pending obligations without expiry (fail: closed)
 - `BATCH_PIPELINE_RECOVERED_MERGE_RELEASE_ENABLED` — Default on unless false; the post-recovery owner wake commits the recovered unit, then its content's ready siblings, at once past an unready merge head (the unit itself also past the catch-up depth hold), so the same wake's chain walk projects them (fail: open)
@@ -1113,6 +1122,7 @@ their code default (`fail` tells you which way a missing value resolves).
 - `BATCH_PIPELINE_RECOVERY_WAKE_LEASE_SECONDS` — One outstanding quarantine-recovery delivery task per paid intent; lease before a lost wake is repaired; default 600, 0 = legacy 120s wake on every sweep (fail: closed)
 - `BATCH_PIPELINE_REUSE_LIVE_ENABLED` — Default-off GPU-chain pre-STT reuse; requires existing lineage gates and full-span words-of-record quality proof (fail: closed)
 - `BATCH_PIPELINE_SECOND_MULTILINGUAL_ENABLED` — Ship-on Soniox second chance after valid-empty Modulate held recovery; false restores Modulate-only ordinary recovery; independent language-guard recovery retains its separate switch and cap (fail: open)
+- `BATCH_PIPELINE_SERVING_PROJECTION_SWEEP` — Default on; only off skips Serving.reconcile projection sweep, preserving owner wakes, held drains, metrics and leader-leased maintenance (fail: open)
 - `BATCH_PIPELINE_SHADOW_LANGUAGE_ONLY` — Default false; classify one canonical batch unit per selected legacy upload instead of entering the idle GPU shadow; no route or publication (fail: closed)
 - `BATCH_PIPELINE_SHADOW_LANGUAGE_PROBES_PER_HOUR` — Default and maximum 500 fleet-wide pair attempts per UTC hour; zero or malformed denies; minute pacing and two 120-second slots also apply (fail: closed)
 - `BATCH_PIPELINE_SHADOW_RECOVERY_API_URL` — Explicit HTTPS existing record-recovery CPU role for shadow recovery; URL is also its OIDC audience; blank fails closed (fail: closed)

backend/docs/listen_pusher_pipeline.mdx

@@ -9299,3 +9299,21 @@ frames, one automatic notes-only refresh is claimed. A user title prevents both
 the claim and a concurrent refresh commit. An all-rejected pass may still refresh
 late OCR/calendar context, but rejected frame bytes never reach notes. This
 refresh does not run derived memory, app, embedding, or speech-learning jobs.
+
+### Dedicated projection pump
+
+`BATCH_PIPELINE_PROJECTION_PUMP_ENABLED` defaults to `false`; only `true`
+selects one projection-only loop in place of the serving runtime's three loops,
+even when `BATCH_PIPELINE_SERVING_ENABLED` is unset/false. It uses the same
+stage/store/artifact construction and never calls reconcile, maintenance or
+recovery dispatch. `BATCH_PIPELINE_PUMP_SWEEP_PAGES` defaults to 1 (clamped
+1-10) and `BATCH_PIPELINE_PUMP_TICK_SECONDS` to 20 seconds (clamped 1-60).
+Malformed numeric values use those defaults. Each tick sweeps pages through
+`sync_executor`, starts owed owner drains on the event loop and sleeps.
+
+`BATCH_PIPELINE_SERVING_PROJECTION_SWEEP` defaults on; only `off` skips the
+serving reconcile sweep. Owner wakes, held drains, metrics refresh, maintenance
+and recovery dispatch keep their existing behavior. The pump ignores this
+serving-only switch. `BATCH_PIPELINE_MAX_OWNER_DRAINS` sets the process's
+owner drain capacity at import (default 8, clamped 1-32; malformed uses 8).
+[Change and ramp](listen_pusher_pipeline/changes/2026-11-05-projection-pump.md).

backend/docs/listen_pusher_pipeline/changes/2026-11-05-projection-pump.md

@@ -0,0 +1,19 @@
+# Dedicated projection discovery pump
+
+The [serving projection contract](../../listen_pusher_pipeline.mdx#dedicated-projection-pump)
+can select one projection-only loop without starting serving maintenance,
+record admission or quarantine dispatch. Unset env preserves the existing three
+serving loops, sweep cadence and eight owner drains. No deploy configuration is
+changed here; service shape remains a separate decision.
+
+The dedicated loop offloads each sweep to `sync_executor`, starts owed owner
+drains on the event loop, and emits a heartbeat with successful sweep count,
+requested owners and elapsed time on every tick, including failed ticks. Failures
+log their exception type and retry on the next tick. Durable projection receipts
+and advisory owner claims are unchanged.
+
+Ramp owner: lead2, tonight (2026-11-05). Start with a pump service of N=3-4,
+verify owner_drain rows/s stays at least 70% of pre-cutover and claim_busy ratio
+does not exceed pre-cutover, then disable serving sweeps only after at least
+30 minutes stable. Rollback: restore serving sweeps, disable the pump, and return
+owner drain capacity to eight. This PR only adds code and unit coverage.