test: add durable finalization, Sync, and Rewind gauntlets (#9981)

## Summary

- Add a non-production Rewind artifact/recovery bridge gauntlet covering
privacy admission, HEVC, SQLite, readback, recovery, and cleanup.
- Make memory-read fixtures inject their reference time so CI never
expires a fixed visibility window.
- Keep the shared vector-fixture seam signature-aligned with
production's explicit `now` snapshot, including callers that inject a
custom fixture clock.
- Fix Rewind lifecycle ownership: cleanup, writes, and finalization are
generation-fenced; concurrent shutdown flushes join the active trailer
write, cancellation surfaces failure, and chunk paths are
collision-proof and recover at their absolute capture time.
- Extend the upstream two-process Listen Cloud Tasks gauntlet with a
deterministic public REST finalization race: four stale-read requests
create one outbox job and one named task, prove the three
`AlreadyExists` handoffs, then rehydrate that task after listener
restart and prove exactly-once worker completion.
- Repair the Firestore contention the new REST race exposed in CI:
recreate the transaction and SDK wrapper for bounded `Aborted` retries,
preserve the atomic outbox boundary, and return the existing retryable
503 contract only after that budget is exhausted. Failed gauntlet jobs
now retain and tail their synthetic backend logs for actionable
diagnosis.
- Adopt the stronger upstream Sync (#10008) and Listen (#10000)
gauntlets during rebase instead of restoring the older overlapping
harnesses.

## Final verification (rebased onto current `origin/main`)

- `BACKEND_PYTEST_WORKERS=4 bash backend/test.sh` — all 675 backend unit
test files passed
- `backend/testing/listen_pusher_stack/run.sh --keep --state-dir
/tmp/omi-listen-pusher-final-head.X326dN` — passed all inline and Cloud
Tasks scenarios, including the deterministic REST/restart case
- `bash backend/test.sh` — all 675 backend unit test files passed again
after the contention repair
- `backend/testing/listen_pusher_stack/run.sh --state-dir
/tmp/omi-listen-pusher-contention-fix` — passed all inline and Cloud
Tasks scenarios after the contention repair
- `bash backend/scripts/typecheck.sh` — 0 errors
- `cd desktop/macos && bash test.sh` — launcher tests, Rust tests, and
isolated Swift suites passed
- `make preflight` — deterministic CI manifest passed
- `scripts/pr-preflight --pr-body-file /tmp/omi-gauntlet-pr-body.md` and
`scripts/failure-class validate --base origin/main --head HEAD
--pr-body-file /tmp/omi-gauntlet-pr-body.md` — passed

The former PR head was GitHub-conflicted, which suppressed normal
`pull_request` runs. This branch is rebased cleanly onto current
`origin/main`; these are the final-head checks before its fresh CI run.

## Guard scope

The four-party read barrier intentionally lives only in the Listen
gauntlet entrypoint, rather than in a shared or production primitive. It
coordinates a controllable harness race after the real authenticated
lookup; the production route, Firestore transaction, Cloud Tasks task
construction, listener restart, and worker all remain unmodified. It
extends the recovery surface introduced by merged
[#10000](#10000).

## Product invariants affected

- INV-MEM-1
- INV-MEM-2

## Failure class

Failure-Class: none

The Rewind lifecycle defects and bounded Firestore
transaction-contention recovery have behavioral regression guards, but
no existing semantic failure class describes either failure boundary.

Author: Git-on-my-level

.github/workflows/backend-hermetic-e2e.yml

@@ -98,7 +98,17 @@ jobs:
           sudo apt-get install --yes redis-server
 
       - name: Run listen to pusher stack gauntlet
-        run: npm run test:listen-pusher-stack:emulator
+        run: npm run test:listen-pusher-stack:emulator -- --state-dir "$RUNNER_TEMP/listen-pusher-stack"
+
+      - name: Show listen gauntlet backend logs on failure
+        if: failure()
+        run: |
+          state_dir="$RUNNER_TEMP/listen-pusher-stack"
+          if [[ -d "$state_dir" ]]; then
+            find "$state_dir" -type f -name backend.log -print -exec tail -n 160 {} \;
+          else
+            echo "Listen gauntlet state directory was not retained."
+          fi
 
   sync-cloud-tasks-stack-gauntlet:
     name: Sync Cloud Tasks Stack Gauntlet

.github/workflows/desktop-backend-contracts.yml

@@ -22,6 +22,7 @@ on:
       - "desktop/macos/e2e/**"
       - "desktop/macos/Desktop/Sources/DesktopAutomationBridge.swift"
       - "desktop/macos/Desktop/Sources/FloatingControlBar/RealtimeHubController.swift"
+      - "desktop/macos/Desktop/Sources/Rewind/Core/RewindArtifactGauntlet.swift"
 
 jobs:
   desktop-core-e2e-t0:

backend/database/conversation_finalization_jobs.py

@@ -15,6 +15,7 @@
 
 from database import conversations as conversations_db
 from database._client import document_id_from_seed, get_firestore_client
+from database.firestore_transaction_retry import run_with_transaction_contention_retry
 
 CONVERSATIONS_COLLECTION = 'conversations'
 FINALIZATION_JOBS_COLLECTION = 'conversation_finalization_jobs'
@@ -252,19 +253,30 @@ def create_or_get_finalization_intent(
 ) -> FinalizationIntent:
     client = _client(firestore_client)
     conversation_ref = _conversation_ref(client, uid, conversation_id)
-    transaction = client.transaction()
-    transactional = firestore.transactional(_create_or_get_finalization_intent_txn)
-    return transactional(
-        transaction,
-        conversation_ref,
-        client.collection(FINALIZATION_JOBS_COLLECTION),
-        uid,
-        conversation_id,
-        requires_byok,
-        finalization_admission,
-        _now(),
-        force_process=force_process,
-        extra_updates=extra_updates,
+    jobs_collection = client.collection(FINALIZATION_JOBS_COLLECTION)
+
+    def create_intent_in_transaction(transaction: Any) -> FinalizationIntent:
+        # The Firestore SDK's transactional wrapper retains retry state. Build
+        # it for this outer attempt so concurrent REST finalizers always get a
+        # fresh transaction and wrapper after read-time contention.
+        transactional = firestore.transactional(_create_or_get_finalization_intent_txn)
+        return transactional(
+            transaction,
+            conversation_ref,
+            jobs_collection,
+            uid,
+            conversation_id,
+            requires_byok,
+            finalization_admission,
+            _now(),
+            force_process=force_process,
+            extra_updates=extra_updates,
+        )
+
+    return run_with_transaction_contention_retry(
+        client.transaction,
+        create_intent_in_transaction,
+        operation_name='conversation_finalization_intent',
     )
 
 

backend/models/memory_search_gateway.py

@@ -72,6 +72,7 @@ def hydrate_and_filter_vector_hits(
     mode: SearchMode,
     required_projection_commit_id: str,
     required_account_generation: int,
+    now: Optional[datetime] = None,
 ) -> SearchGatewayResult:
     """Fail-closed vector gateway.
 
@@ -231,9 +232,9 @@ def hydrate_and_filter_vector_hits(
             )
             continue
         access = (
-            is_archive_access_eligible(item, policy)
+            is_archive_access_eligible(item, policy, now=now)
             if mode == SearchMode.archive_explicit
-            else is_default_access_eligible(item, policy)
+            else is_default_access_eligible(item, policy, now=now)
         )
         if not access.allowed:
             decisions[hit.memory_id] = SearchDecision.access_denied

backend/routers/memory_product.py

@@ -174,6 +174,7 @@ def search_vector_memory(
             limit=limit,
             required_projection_commit_id=rollout.vector_projection_commit_id,
             required_account_generation=rollout.rollout_capabilities.account_generation,
+            now=_current_time(),
         )
     except ValueError as exc:
         raise HTTPException(status_code=400, detail=str(exc)) from exc

backend/testing/listen_pusher_stack/README.md

@@ -69,13 +69,19 @@ Scenarios:
 3. a stale empty desktop recording is removed by the next-session lifecycle path and creates no job;
 4. a pusher process loses the first 104 before claim, is restarted, and the
    live backend session replays the same job ID and dispatch generation exactly once.
-5. a session closes during the deferred pending-finalization window; the real
+5. concurrent public `POST /v1/conversations/{id}/finalize` retries produce one
+   opaque named task and one outbox job, prove the `AlreadyExists` boundary,
+   then survive listener restart before the detached worker completes and safely
+   ACKs a duplicate delivery. A bounded test-entrypoint read barrier makes the
+   intended stale-read race deterministic without replacing the route,
+   lifecycle transaction, or task construction;
+6. a session closes during the deferred pending-finalization window; the real
    recovery path from #9960 enqueues one opaque Cloud Tasks task, then a real
    worker retry preserves `processing` until it completes the same job;
-6. a worker exhausting its two-attempt test budget atomically dead-letters the
+7. a worker exhausting its two-attempt test budget atomically dead-letters the
    job and marks the still-current conversation `failed`/`discarded`, while a
    later duplicate delivery is fenced;
-7. an integration failure after processing retries only durable fanout, never
+8. an integration failure after processing retries only durable fanout, never
    re-runs completed conversation processing.
 
 The inline compatibility coverage deliberately triggers stale live-session

backend/testing/listen_pusher_stack/cloud_tasks.py

@@ -138,6 +138,13 @@ def create_task(self, *, parent: str, task: Any) -> Any:
             raise RuntimeError('finalization task name does not match its opaque durable identity')
         with _task_event_lock():
             if task_name in _read_task_names():
+                _append_event(
+                    {
+                        'event': 'task_already_exists',
+                        'task_name': task_name,
+                        'payload': payload,
+                    }
+                )
                 raise AlreadyExists(f'loopback task already exists: {task_name}')
             _append_event(
                 {

backend/testing/listen_pusher_stack/finalizer_leaves.py

@@ -54,7 +54,7 @@ def _consume_failure(stage: str, conversation_id: str, **metadata: Any) -> bool:
     return True
 
 
-def _offline_process_conversation(uid: str, _language: str, conversation: Any, **_kwargs: Any) -> Any:
+def _offline_process_conversation(uid: str, _language: str, conversation: Any, **kwargs: Any) -> Any:
     conversation_id = str(conversation.id)
     if _consume_failure('process', conversation_id):
         raise RuntimeError('controlled finalization processing failure')
@@ -67,6 +67,8 @@ def _offline_process_conversation(uid: str, _language: str, conversation: Any, *
             'outcome': 'completed',
             'conversation_id': conversation_id,
             'persisted': bool(persisted),
+            'force_process': bool(kwargs.get('force_process')),
+            'defer_memory_extraction': bool(kwargs.get('defer_memory_extraction')),
         }
     )
     return conversation

backend/testing/listen_pusher_stack/listener_app.py

@@ -1,11 +1,69 @@
 """Instrumented listener entrypoint for durable finalization scenarios.
 
-The production FastAPI application and task construction remain real. Only the
-Cloud Tasks transport client is replaced by a strict loopback boundary.
+The production FastAPI application and task construction remain real. Cloud
+Tasks is the only external transport replaced; an opt-in gate merely
+coordinates real route lookups for a deterministic concurrency scenario.
 """
 
+import os
+from threading import Barrier, BrokenBarrierError, Lock
+
 from testing.listen_pusher_stack.cloud_tasks import install_loopback_tasks_client
 
 install_loopback_tasks_client()
 
 from main import app  # noqa: E402
+
+
+def _install_rest_finalization_race_barrier() -> None:
+    """Force a bounded stale-read race before the real finalization transaction.
+
+    The live gauntlet sends concurrent public REST calls, but scheduler timing
+    alone cannot guarantee every handler reads ``in_progress`` before the
+    winning Firestore transaction changes it to ``processing``. This opt-in
+    harness seam gates only the first N target reads after they have used the
+    production lookup. The route, auth, lifecycle transaction, and Cloud Tasks
+    call therefore stay real while the intended named-task race is repeatable.
+    """
+    uid = os.getenv('OMI_STACK_FINALIZATION_RACE_UID', '')
+    conversation_id = os.getenv('OMI_STACK_FINALIZATION_RACE_CONVERSATION_ID', '')
+    raw_parties = os.getenv('OMI_STACK_FINALIZATION_RACE_PARTIES', '')
+    if not any((uid, conversation_id, raw_parties)):
+        return
+    if not all((uid, conversation_id, raw_parties)):
+        raise RuntimeError('REST finalization race barrier requires uid, conversation ID, and party count')
+    try:
+        parties = int(raw_parties)
+    except ValueError as error:
+        raise RuntimeError('REST finalization race barrier party count must be an integer') from error
+    if parties < 2:
+        raise RuntimeError('REST finalization race barrier needs at least two parties')
+
+    from routers import conversations as conversations_router
+
+    original_lookup = conversations_router._get_valid_conversation_by_id
+    barrier = Barrier(parties)
+    lock = Lock()
+    remaining = parties
+
+    def lookup_with_race_gate(request_uid: str, request_conversation_id: str):
+        nonlocal remaining
+        snapshot = original_lookup(request_uid, request_conversation_id)
+        if request_uid != uid or request_conversation_id != conversation_id:
+            return snapshot
+        with lock:
+            should_wait = remaining > 0
+            if should_wait:
+                remaining -= 1
+        if not should_wait:
+            return snapshot
+        try:
+            barrier.wait(timeout=10.0)
+        except BrokenBarrierError as error:
+            raise RuntimeError('REST finalization race barrier did not receive every request') from error
+        return snapshot
+
+    conversations_router._get_valid_conversation_by_id = lookup_with_race_gate
+
+
+_install_rest_finalization_race_barrier()