fix(batch): stripe projection discovery startup cursors (#24716)

## Problem
Projection discovery instances begin at the same keyset head after a
roll, repeatedly offering the same owners to the advisory claim.
Revision, hostname and PID alone may be identical across Cloud Run
replicas.

## Fix
Select a random 128-bit process seed on first use, mix it with the
revision, and cache the resulting SHA-256 stripe modulo 4096 for the
process lifetime. A local lock makes first-use initialization and
logging occur once even with concurrent scanners. No metadata server or
network I/O is used. The larger stripe space reduces collisions at 30–50
instances; restarts may choose a different stripe. Log the chosen stripe
once in the process selector, rather than for every scanner.

Seed each scanner at its committed_at lower bound between the oldest
matching result and startup time. Existing committed_at + document ID
ordering remains oldest-first after that bound, then wraps to the head.
Every row remains discoverable; no durable state changes. The existing
follow-up seed takes precedence. Page sizes, sweep cadence, owner-drain
concurrency and owner claims are unchanged.

## Tests
314 passed across 8 files using `BACKEND_UNIT_TEST_FILE_LIST=<file> bash
backend/test.sh` through `bin/omi-heavy run --`: projection discovery,
batch-stage environment ratchet, flag budget, unit policy, follow-up
desync, follow-up reads, sweep owner-chain and owner-drain tests.
Injected seeds model separate processes with identical metadata and
verify process-stable selection, different stripes and one log per
process. Full-cycle coverage covers stripe boundaries through 4095,
timestamp ties, wrap and empty/short collections.

B1 from the supplied correctness review is fixed. The
insert-behind-cursor follow-up remains outside this change.

## Risk
This is discovery desynchronization, not exclusive partitioning. Hash
collisions and sparse/skewed timestamp ranges can still share starting
rows. Startup adds one scoped head read per scanner; a seed beyond the
newest result retries the head immediately. Throughput and claim-busy
impact require a separately authorized production canary. No production
access, merge or deployment in this lane.

## Rollback
The switch defaults on. Set
`BATCH_PIPELINE_PROJECTION_STRIPING_ENABLED=off` and restart scanners to
recover the exact pre-change cursor sequence and query budget.
Pagination off also bypasses seeding.

Failure-Class: none
This is cross-instance discovery contention; the blocked-page class
covers dependency starvation and the scheduler-shard class covers shared
control mutations. Transaction fences remain unchanged.

Author: Git-on-my-level

backend/config/batch_pipeline.py

@@ -3,8 +3,12 @@
 from config.refactor_switches import default_off as rollout_default_off, default_on as rollout_default_on
 
 from enum import IntEnum
+import hashlib
+import logging
 import math
 import os
+import secrets
+import threading
 
 from config.rollout_cohorts import rollout_cohort_members
 from typing import Callable, TypeVar
@@ -515,6 +519,38 @@ def projection_pagination_enabled() -> bool:
     return kill_switch('BATCH_PIPELINE_PROJECTION_PAGINATION_ENABLED')
 
 
+def projection_striping_enabled() -> bool:
+    from config.policy import kill_switch
+
+    return kill_switch('BATCH_PIPELINE_PROJECTION_STRIPING_ENABLED')
+
+
+PROJECTION_DISCOVERY_STRIPES = 4096
+_projection_discovery_stripe_value: int | None = None
+_PROJECTION_DISCOVERY_LOCK = threading.Lock()
+
+
+def projection_discovery_stripe() -> int:
+    """Choose and log one process-local random phase, without network I/O.
+
+    Replicas may share hostname/PID. Fresh entropy separates them; 4096 buckets
+    reduce collisions at 30-50 instances. Restarts may choose a different phase.
+    """
+    global _projection_discovery_stripe_value
+    with _PROJECTION_DISCOVERY_LOCK:
+        if _projection_discovery_stripe_value is None:
+            identity = f'{os.getenv("K_REVISION", "")}:{secrets.randbits(128)}'
+            _projection_discovery_stripe_value = (
+                int.from_bytes(hashlib.sha256(identity.encode()).digest()[:8], 'big') % PROJECTION_DISCOVERY_STRIPES
+            )
+            logging.getLogger(__name__).info(
+                'event=batch_projection_discovery action=start stripe=%d stripes=%d',
+                _projection_discovery_stripe_value,
+                PROJECTION_DISCOVERY_STRIPES,
+            )
+        return _projection_discovery_stripe_value
+
+
 def owner_chain_projection_enabled() -> bool:
     from config.policy import kill_switch
 

backend/config/flag_budget.json

@@ -24,9 +24,11 @@
   },
   "seed_review_by": "2026-11-15",
   "budget": 169,
-  "count": 249,
+  "count": 251,
   "names": {
     "PROJECTION_DRAIN_CLAIM_AWARE": {"owner": "two-lane lead", "review_by": "2026-12-05", "registry": "config/feature-flags.yaml"},
+    "BATCH_PIPELINE_PROJECTION_STRIPING_ENABLED": {"owner": "omi-pipeline", "review_by": "2026-12-06", "registry": "config/feature-flags.yaml"},
+    "K_REVISION": {"owner": "omi-pipeline", "review_by": "2026-12-06"},
     "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

@@ -11,6 +11,7 @@
     'BATCH_PIPELINE_LEADER_LEASE_ENABLED': True,
     'BATCH_PIPELINE_RECOVERY_TASKS_ENABLED': True,
     'BATCH_PIPELINE_PROJECTION_PAGINATION_ENABLED': True,
+    'BATCH_PIPELINE_PROJECTION_STRIPING_ENABLED': True,
     'BATCH_PIPELINE_OWNER_CHAIN_PROJECTION_ENABLED': True,
     'BATCH_PIPELINE_RECOVERED_MERGE_RELEASE_ENABLED': True,
     'BATCH_PIPELINE_RECOVERY_OWNER_ONLY_ENABLED': True,
@@ -35,16 +36,15 @@ 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 == 'PROJECTION_DRAIN_CLAIM_AWARE':
+    if name in {'BATCH_PIPELINE_PROJECTION_STRIPING_ENABLED', 'PROJECTION_DRAIN_CLAIM_AWARE'}:
         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
 
-    # This switch keeps its narrower off-only grammar in every parser mode.
-    if name != 'PROJECTION_DRAIN_CLAIM_AWARE':
+    if name not in {'BATCH_PIPELINE_PROJECTION_STRIPING_ENABLED', 'PROJECTION_DRAIN_CLAIM_AWARE'}:
         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

@@ -543,6 +543,7 @@ and an explicit empty literal renders as `''`.
 | `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_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_PROJECTION_STRIPING_ENABLED` | Default on unless off; seed projection discovery at an instance-specific committed_at stripe, then wrap through every row | backend | env | open | — | — | — | — | keep | keep | — | omi-pipeline |
 | `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 |
@@ -1104,6 +1105,7 @@ their code default (`fail` tells you which way a missing value resolves).
 - `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_SOURCE_DELETED_ENABLED` — Default on; settle deleted original sources without a tombstone or redirect as source_deleted, without content publication (fail: open)
+- `BATCH_PIPELINE_PROJECTION_STRIPING_ENABLED` — Default on unless off; seed projection discovery at an instance-specific committed_at stripe, then wrap through every row (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_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)
@@ -1461,6 +1463,7 @@ reason:
 
 | Key | Reason | owner | decision | review_by | notes |
 | --- | --- | --- | --- | --- | --- |
+| `K_REVISION` | Cloud Run revision identity for process-local discovery phase, not a behavior flag | — | — | — | — |
 | `DD_TRACE_ENABLED` | Datadog telemetry instrumentation switch, not a product behavior flag | — | — | — | — |
 | `DD_LOGS_ENABLED` | Datadog telemetry log routing switch, not a product behavior flag | — | — | — | — |
 | `OMI_FORCE_CONTEXT_BUCKETS` | Local-only dev override, not shipped remote authority | — | — | — | — |

backend/tests/unit/test_batch_pipeline_followup_desync.py

@@ -63,6 +63,8 @@ def keep(entry):
 
 
 def runner(rows, monkeypatch, *, fraction=0.0, now=None):
+    # Isolate the pre-existing follow-up offset/backoff switch from global striping.
+    monkeypatch.setenv('BATCH_PIPELINE_PROJECTION_STRIPING_ENABLED', 'off')
     query = ResultQuery(rows)
     store = Mock()
     store.projection_page.side_effect = lambda **kw: projection_page(query, 'batch_pipeline_dev', **kw)

backend/tests/unit/test_batch_pipeline_followup_reads.py

@@ -55,6 +55,7 @@ def stream(self):
 
 @pytest.fixture(autouse=True)
 def head_first_discovery(monkeypatch):
+    monkeypatch.setenv('BATCH_PIPELINE_PROJECTION_STRIPING_ENABLED', 'off')
     # These read budgets pin the head-first pass (desync kill switch). The
     # desync seed adds one 1-row read per process: test_batch_pipeline_followup_desync.py.
     monkeypatch.setenv('BATCH_PIPELINE_PROJECTION_FOLLOWUP_BACKOFF_SECONDS', '0')

backend/tests/unit/test_batch_pipeline_projection_discovery.py

@@ -2,11 +2,19 @@
 
 from unittest.mock import Mock
 
+import pytest
+
 from utils.batch_pipeline.projection_discovery import ProjectionDiscovery
+from config import batch_pipeline
+from config.policy import kill_switch
+from database.batch_pipeline_store import projection_page
+from tests.unit.test_batch_pipeline_followup_desync import ResultQuery
+from utils.batch_pipeline import projection_discovery
 
 
 def test_default_on_independent_scans_wrap_and_kill_switch_restores_first_page(monkeypatch):
     monkeypatch.delenv('BATCH_PIPELINE_PROJECTION_PAGINATION_ENABLED', raising=False)
+    monkeypatch.setenv('BATCH_PIPELINE_PROJECTION_STRIPING_ENABLED', 'off')
     store = Mock()
     store.projection_page.side_effect = lambda *, limit, after: (
         (['ready'] * 100, (2.0, 'ready-100')) if after is None else (['sibling', 'other-owner'], None)
@@ -23,7 +31,8 @@ def test_default_on_independent_scans_wrap_and_kill_switch_restores_first_page(m
     assert assignment.after is followup.after is None
 
 
-def test_failed_query_retries_cursor_without_losing_page():
+def test_failed_query_retries_cursor_without_losing_page(monkeypatch):
+    monkeypatch.setenv('BATCH_PIPELINE_PROJECTION_STRIPING_ENABLED', 'off')
     store = Mock()
     store.projection_page.side_effect = [(['head'], (2.0, 'head')), RuntimeError('query unavailable'), (['tail'], None)]
     scan = ProjectionDiscovery(store)
@@ -40,3 +49,97 @@ def test_failed_query_retries_cursor_without_losing_page():
         (2.0, 'head'),
         (2.0, 'head'),
     ]
+
+
+def scanner(monkeypatch, rows, stripe):
+    monkeypatch.delenv('BATCH_PIPELINE_PROJECTION_STRIPING_ENABLED', raising=False)
+    monkeypatch.delenv('BATCH_PIPELINE_PROJECTION_PAGINATION_ENABLED', raising=False)
+    monkeypatch.setattr(projection_discovery, 'projection_discovery_stripe', lambda: stripe)
+    monkeypatch.setattr(projection_discovery.time, 'time', lambda: 1256.0)
+    query = ResultQuery(rows)
+    store = Mock()
+    store.projection_page.side_effect = lambda **kw: projection_page(query, 'synthetic', **kw)
+    return ProjectionDiscovery(store, states=('pending',)), query
+
+
+@pytest.mark.parametrize('stripe', [0, 1, 1024, 2048, 4095])
+@pytest.mark.parametrize('limit', [1, 17, 100])
+def test_striped_cycle_visits_every_row_exactly_once_including_timestamp_ties(monkeypatch, stripe, limit):
+    rows = {f'r{i:03d}': {'committed_at': 1000.0 + i // 2, 'projection_state': 'pending'} for i in range(512)}
+    scan, query = scanner(monkeypatch, rows, stripe)
+    seen = []
+    while len(seen) < 2 * len(rows):
+        seen.extend(scan.next(limit))
+    ordered = sorted(rows)
+    first = 2 * ((stripe + 15) // 16)
+    rotated = ordered[first:] + ordered[:first]
+    assert seen[: len(rows)] == seen[len(rows) : 2 * len(rows)] == rotated
+    assert query.calls[0] == (None, 1)  # one scoped startup head read
+    assert query.calls[1] == ((1000.0 + stripe / 16,), limit)
+
+
+def test_process_entropy_is_sampled_and_logged_once_per_process(monkeypatch, caplog):
+    monkeypatch.setenv('K_REVISION', 'same-revision-across-replicas')
+    entropy = Mock(side_effect=[11, 22])
+    monkeypatch.setattr(batch_pipeline.secrets, 'randbits', entropy)
+    monkeypatch.setattr(batch_pipeline, '_projection_discovery_stripe_value', None)
+    try:
+        with caplog.at_level('INFO', logger=batch_pipeline.__name__):
+            first = ProjectionDiscovery(Mock()).stripe
+            assert ProjectionDiscovery(Mock()).stripe == first
+            assert entropy.call_count == 1
+            # Simulate another process with identical revision/hostname/PID,
+            # injecting a fresh seed rather than relying on probabilistic tests.
+            monkeypatch.setattr(batch_pipeline, '_projection_discovery_stripe_value', None)
+            second = ProjectionDiscovery(Mock()).stripe
+            assert ProjectionDiscovery(Mock()).stripe == second
+        assert first != second
+        assert 0 <= first < 4096 and 0 <= second < 4096
+        assert entropy.call_count == 2
+        assert caplog.text.count('event=batch_projection_discovery action=start') == 2
+        assert caplog.text.count('stripes=4096') == 2
+    finally:
+        monkeypatch.setattr(batch_pipeline, '_projection_discovery_stripe_value', None)
+
+
+@pytest.mark.parametrize('raw', ['', 'on', 'false', 'garbage', 'off', ' OFF '])
+@pytest.mark.parametrize('mode', ['off', 'serve'])
+def test_only_off_disables_striping(monkeypatch, raw, mode):
+    monkeypatch.setenv('CAPTURE_COVERAGE_MODE', mode)
+    monkeypatch.setenv('BATCH_PIPELINE_PROJECTION_STRIPING_ENABLED', raw)
+    assert batch_pipeline.projection_striping_enabled() == (raw.strip().lower() != 'off')
+    assert kill_switch('BATCH_PIPELINE_PROJECTION_STRIPING_ENABLED', flags={})
+
+
+@pytest.mark.parametrize('count', [0, 1, 3])
+def test_empty_short_or_stale_collection_returns_head_without_tick_delay(monkeypatch, count):
+    rows = {f'r{i}': {'committed_at': 1000.0, 'projection_state': 'pending'} for i in range(count)}
+    scan, _ = scanner(monkeypatch, rows, 4095)
+    assert scan.next() == sorted(rows)
+    assert scan.next() == sorted(rows)
+
+
+def test_stripe_seed_query_failure_retries_without_advancing(monkeypatch):
+    rows = {'r0': {'committed_at': 1000.0, 'projection_state': 'pending'}}
+    scan, _ = scanner(monkeypatch, rows, 128)
+    page = scan.store.projection_page.side_effect
+    scan.store.projection_page.side_effect = RuntimeError('unavailable')
+    with pytest.raises(RuntimeError):
+        scan.next()
+    assert scan.after is None and not scan.seeded
+    scan.store.projection_page.side_effect = page
+    assert scan.next() == ['r0']
+
+
+def test_pagination_off_bypasses_startup_seed(monkeypatch):
+    scan, query = scanner(monkeypatch, {}, 128)
+    monkeypatch.setenv('BATCH_PIPELINE_PROJECTION_PAGINATION_ENABLED', 'false')
+    assert scan.next() == []
+    assert query.calls == [(None, 100)]
+
+
+def test_existing_followup_seed_is_preserved(monkeypatch):
+    scan, query = scanner(monkeypatch, {}, 128)
+    scan.after = (1200.0, '')
+    assert scan.next() == []
+    assert query.calls == [((1200.0,), 100)]

backend/tests/unit/test_batch_pipeline_sweep_owner_chain.py

@@ -30,6 +30,8 @@
 @pytest.fixture(autouse=True)
 def claims(monkeypatch):
     """Advisory owner claim that records nesting; a busy set refuses entry."""
+    # Trace assertions exercise the exact head-first discovery rollback.
+    monkeypatch.setenv('BATCH_PIPELINE_PROJECTION_STRIPING_ENABLED', 'off')
     state = SimpleClaims()
 
     @contextmanager

backend/tests/unit/test_batch_unit_policy.py

@@ -87,8 +87,16 @@ def test_cohort_registry_and_live_switch_contract():
     for key in KILL_SWITCHES:
         assert by_key[key]['lifecycle'] == 'ops_kill'
         assert kill_switch(key, flags={})
-        disabled = 'off' if key == 'PROJECTION_DRAIN_CLAIM_AWARE' else 'false'
-        assert not kill_switch(key, flags={key: disabled})
+        assert not kill_switch(
+            key,
+            flags={
+                key: (
+                    'off'
+                    if key in {'BATCH_PIPELINE_PROJECTION_STRIPING_ENABLED', 'PROJECTION_DRAIN_CLAIM_AWARE'}
+                    else 'false'
+                )
+            },
+        )
     assert 'nested' in by_key['BATCH_LIVE_STT_RETIRE_PERCENT']['notes'].lower()
     with pytest.raises(ValueError):
         kill_switch('BATCH_PIPELINE_ENABLED')

backend/tests/unit/test_flag_budget.py

@@ -50,6 +50,7 @@
 SEED_BUDGET = 170  # 2026-10-25 seed; lower it when the surface shrinks, never raise it.
 HELPER_KILL_SWITCHES = {
     'PROJECTION_DRAIN_CLAIM_AWARE',
+    'BATCH_PIPELINE_PROJECTION_STRIPING_ENABLED',
     'CONVERSATION_OCCASION_RELEVANCE_ENABLED',
     'LATE_AUDIO_SYNC_INTAKE_ENABLED',
     'EPISODE_RECORDED_TIME_ENABLED',
@@ -217,14 +218,15 @@ def test_helper_kill_switch_budget_entries_match_the_default_on_registry():
             assert entry['owner'] == 'two-lane lead'
             assert entry['review_by'] == '2026-12-05'
         else:
-            assert entry['owner'] == 'rollout-meetings / occasion architecture'
-            assert entry['review_by'] == '2026-11-28'
-        assert flag['owner'] == 'dazheng'
+            striped = name == 'BATCH_PIPELINE_PROJECTION_STRIPING_ENABLED'
+            assert entry['owner'] == ('omi-pipeline' if striped else 'rollout-meetings / occasion architecture')
+            assert entry['review_by'] == ('2026-12-06' if striped else '2026-11-28')
+        assert flag['owner'] == ('omi-pipeline' if name == 'BATCH_PIPELINE_PROJECTION_STRIPING_ENABLED' else 'dazheng')
         assert flag['kind'] == 'env'
         assert flag['lifecycle'] == 'ops_kill'
         assert 'backend' in flag['surfaces']
         assert flag['fail'] == 'open'
-        assert flag['summary'].startswith('Default on;')
+        assert flag['summary'].startswith('Default on')
 
 
 def _data(retired=None, **names):