fix(sync): join a landed sync clip to the live episode at intake, not after notes

.github/failure-classes/FC-episode-membership-waits-for-enrichment.json

@@ -0,0 +1,10 @@
+{
+  "schema_version": 1,
+  "id": "FC-episode-membership-waits-for-enrichment",
+  "violated_contract": "A conversation that belongs to the live occasion must be a member of the live episode from the moment intake persists it, so Home folds it under the live episode row; membership may not wait for the conversation's own enrichment to finalize.",
+  "canonical_prevention": "Join at the boundary that creates the visible row (sync intake persistence), keep the finalize-time join only as an idempotent safety net, and cover both boundaries plus the started-before-episode grace in unit tests against the episode service.",
+  "canonical_prevention_artifact": ["backend/tests/unit/test_episodes_service.py"],
+  "evidence_prs": [],
+  "scope_hints": ["backend/utils/episodes/", "backend/utils/sync/"],
+  "status": "open"
+}

backend/tests/unit/test_episodes_finalize_paths.py

@@ -260,3 +260,37 @@ def saturated(*_args, **_kwargs):
 
     monkeypatch.setattr(episode_finalize, 'submit_with_context', saturated)
     episode_finalize.schedule_after_finalization(_UID, 'c1', ProcessingTrigger.CLIENT_FINALIZE)
+
+
+def test_attach_synced_at_intake_joins_through_the_service_and_normalizes_the_row(monkeypatch):
+    from utils.episodes import service
+
+    seen = []
+    monkeypatch.setattr(episode_finalize.client_gate, 'client_seen_cached', lambda uid: True)
+    monkeypatch.setattr(
+        service,
+        'attach_synced_intake',
+        lambda uid, cid, row, collaborators=None: seen.append((uid, cid, row['source'])) or 'joined',
+    )
+    row = {'id': 'c-wal', 'source': ConversationSource.omi}
+    assert episode_finalize.attach_synced_at_intake(_UID, 'c-wal', row) == 'joined'
+    # The boundary rules compare wire values; the pydantic enum member never reaches them.
+    assert seen == [(_UID, 'c-wal', 'omi')]
+
+
+def test_attach_synced_at_intake_is_gated_and_fail_open(monkeypatch):
+    monkeypatch.setattr(episode_finalize.client_gate, 'client_seen_cached', lambda uid: False)
+    assert episode_finalize.attach_synced_at_intake(_UID, 'c-wal', {'id': 'c-wal'}) == 'skipped_no_client'
+    monkeypatch.setattr(episode_finalize.client_gate, 'client_seen_cached', lambda uid: True)
+    monkeypatch.setenv(config.EPISODES_ENABLED_ENV, 'off')
+    assert episode_finalize.attach_synced_at_intake(_UID, 'c-wal', {'id': 'c-wal'}) == 'disabled'
+    monkeypatch.delenv(config.EPISODES_ENABLED_ENV)
+    assert episode_finalize.attach_synced_at_intake(_UID, '', {'id': 'c-wal'}) == 'disabled'
+
+    def boom(*_args, **_kwargs):
+        raise RuntimeError('episode store unavailable')
+
+    from utils.episodes import service
+
+    monkeypatch.setattr(service, 'attach_synced_intake', boom)
+    assert episode_finalize.attach_synced_at_intake(_UID, 'c-wal', {'id': 'c-wal'}) == 'error'

backend/tests/unit/test_episodes_service.py

@@ -1,4 +1,4 @@
 """Episode orchestration over the strict Firestore fake: moments, boundaries, reconciliation, wrap-up."""
 
 import json
 from dataclasses import replace
@@ -1164,11 +1164,12 @@
         == 'no_episode'
     )
     assert episodes_db.get_episode(UID, episode['id'], firestore_client=store)['status'] == 'live'
-    # Audio older than the episode belongs to an occasion already over.
+    # Audio from an occasion already over: it began far before this episode and
+    # its span ended long before the episode's first recorded words.
     _attach(store, 'c-next', minutes=60, now_minutes=60)
     current = episodes_db.get_current_episode(UID, firestore_client=store)
     assert (
-        service.on_conversation_finalized(UID, 'c-wal', _wal_row(minutes=59), trigger='sync_update', collaborators=c)
+        service.on_conversation_finalized(UID, 'c-wal', _wal_row(minutes=45), trigger='sync_update', collaborators=c)
         == 'no_episode'
     )
     after = episodes_db.get_current_episode(UID, firestore_client=store)
@@ -1195,3 +1196,104 @@
     }
     opened = episodes_db.get_episode_for_conversation(UID, 'c-cap', firestore_client=store)
     assert opened is not None and 'c-cap' in opened['conversation_ids']
+
+
+# --------------------------------------------------------------------------- sync intake attach
+
+
+def test_a_sync_clip_joins_the_live_episode_at_intake_and_the_finalize_hook_stays_idempotent():
+    store = StrictFirestore()
+    _attach(store, 'c-live', minutes=0)
+    episode = episodes_db.get_current_episode(UID, firestore_client=store)
+    model = FakeModel()
+    c = _collaborators(model, now=T0 + timedelta(minutes=2), store=store)
+    # The WAL row exists the moment intake persisted it; Home must fold it now, not
+    # after enrichment finishes.
+    assert service.attach_synced_intake(UID, 'c-wal', _wal_row(), collaborators=c) == 'joined'
+    after = episodes_db.get_current_episode(UID, firestore_client=store)
+    assert after['id'] == episode['id'] and after['status'] == 'live'
+    assert after['conversation_ids'] == ['c-live', 'c-wal']
+    assert episodes_db.get_episode_for_conversation(UID, 'c-wal', firestore_client=store)['id'] == episode['id']
+    # The cheap rules alone decided: no model call, and no text fed.
+    assert model.calls == [] and after['applied'].keys() == set() and after['moments'] == []
+    # Enrichment finishing later neither re-joins nor replays the duplicate transcript.
+    assert service.on_conversation_finalized(UID, 'c-wal', _wal_row(), trigger='sync_update', collaborators=c) == (
+        'finalized'
+    )
+    replay = episodes_db.get_current_episode(UID, firestore_client=store)
+    assert replay['conversation_ids'] == ['c-live', 'c-wal'] and model.calls == []
+    # A replayed intake is a no-op on the same membership.
+    assert service.attach_synced_intake(UID, 'c-wal', _wal_row(), collaborators=c) == 'attached'
+
+
+def test_intake_attach_respects_the_gates_and_an_existing_episode_ownership(monkeypatch):
+    store = StrictFirestore()
+    _attach(store, 'c-live', minutes=0)
+    c = _collaborators(FakeModel(), now=T0 + timedelta(minutes=2), store=store)
+    monkeypatch.setenv(config.EPISODES_ENABLED_ENV, 'off')
+    assert service.attach_synced_intake(UID, 'c-wal', _wal_row(), collaborators=c) == 'disabled'
+    monkeypatch.delenv(config.EPISODES_ENABLED_ENV)
+    monkeypatch.setattr(service.client_gate, 'client_seen_cached', lambda uid: False)
+    assert service.attach_synced_intake(UID, 'c-wal', _wal_row(), collaborators=c) == 'skipped_no_client'
+    assert episodes_db.get_current_episode(UID, firestore_client=store)['conversation_ids'] == ['c-live']
+    monkeypatch.setattr(service.client_gate, 'client_seen_cached', lambda uid: True)
+    # A member of the live episode never re-joins, and an episode owner never moves.
+    assert service.attach_synced_intake(UID, 'c-live', _wal_row('c-live'), collaborators=c) == 'attached'
+    assert service.close_on_finalize(UID, 'c-live', trigger='client_finalize', collaborators=c) == 'closed'
+    assert service.attach_synced_intake(UID, 'c-live', _wal_row('c-live'), collaborators=c) == 'no_episode'
+    assert episodes_db.get_current_episode(UID, firestore_client=store) is None
+    assert episodes_db.get_episode_for_conversation(UID, 'c-wal', firestore_client=store) is None
+
+
+def test_intake_attach_with_no_live_episode_creates_nothing():
+    store = StrictFirestore()
+    c = _collaborators(FakeModel(), now=T0 + timedelta(minutes=2), store=store)
+    assert service.attach_synced_intake(UID, 'c-wal', _wal_row(), collaborators=c) == 'no_episode'
+    assert episodes_db.get_current_episode(UID, firestore_client=store) is None
+    assert episodes_db.get_episode_for_conversation(UID, 'c-wal', firestore_client=store) is None
+
+
+def test_a_clip_starting_slightly_before_the_episode_start_joins_within_the_continue_gap():
+    store = StrictFirestore()
+    # The pendant caught the opening words before the phone's stream recorded the start.
+    _attach(store, 'c-live', minutes=1)
+    episode = episodes_db.get_current_episode(UID, firestore_client=store)
+    c = _collaborators(FakeModel(), now=T0 + timedelta(minutes=2), store=store)
+    # Span overlap: started before the episode, finished inside it.
+    assert service.attach_synced_intake(UID, 'c-ov', _wal_row('c-ov', minutes=0.75), collaborators=c) == 'joined'
+    # Clock skew: began a minute before the recorded start, ended half a minute before it.
+    skew = _wal_row('c-skew', minutes=0)
+    skew['finished_at'] = skew['started_at'] + timedelta(seconds=30)
+    assert service.attach_synced_intake(UID, 'c-skew', skew, collaborators=c) == 'joined'
+    after = episodes_db.get_current_episode(UID, firestore_client=store)
+    assert after['id'] == episode['id'] and after['conversation_ids'] == ['c-live', 'c-ov', 'c-skew']
+
+
+def test_a_clip_that_ended_before_the_episode_stays_out_beyond_the_continue_gap():
+    store = StrictFirestore()
+    _attach(store, 'c-live', minutes=10)
+    episode = episodes_db.get_current_episode(UID, firestore_client=store)
+    c = _collaborators(FakeModel(), now=T0 + timedelta(minutes=11), store=store)
+    assert service.attach_synced_intake(UID, 'c-old', _wal_row('c-old'), collaborators=c) == 'no_episode'
+    assert episodes_db.get_episode_for_conversation(UID, 'c-old', firestore_client=store) is None
+    after = episodes_db.get_current_episode(UID, firestore_client=store)
+    assert after['id'] == episode['id'] and after['conversation_ids'] == ['c-live']
+
+
+def test_an_intake_joined_clip_merged_away_follows_membership_to_the_survivor():
+    store = StrictFirestore()
+    _attach(store, 'survivor', minutes=0)
+    c = _collaborators(FakeModel(), now=T0 + timedelta(minutes=2), store=store)
+    assert service.attach_synced_intake(UID, 'donor', _wal_row('donor'), collaborators=c) == 'joined'
+    # Smart merge folds the donor away: the episode keeps standing for the survivor
+    # only, and the deleted donor row can never render on Home.
+    donor_row = {'id': 'donor', 'smart_merge': {'role': 'donor', 'survivor_id': 'survivor'}, 'finished_at': T0}
+    assert (
+        service.on_conversation_finalized(
+            UID, 'survivor', donor_row, trigger='sync_update', absorbed_from='donor', collaborators=c
+        )
+        == 'finalized'
+    )
+    episode = episodes_db.get_current_episode(UID, firestore_client=store)
+    assert episode is not None and episode['conversation_ids'] == ['survivor']
+    assert episodes_db.get_episode_for_conversation(UID, 'survivor', firestore_client=store)['id'] == episode['id']

backend/tests/unit/test_sync_capture_pipeline.py

@@ -1,4 +1,4 @@
 """Real VAD export and process_segment preserve speech extents; silence creates nothing."""
 
 import logging
 from copy import deepcopy
@@ -66,6 +66,9 @@
         ]
         store = StrictFirestore()
         lifecycle.ingest_sync_conversation = lambda uid, row, **kw: intake(store, row, **kw)
+        # The episodes intake attach is fail-open; the harness stand-in skips like a
+        # user whose app never called the episodes API. The wiring test spies on it.
+        module.attach_synced_at_intake = lambda uid, cid, row: 'skipped_no_client'
         yield module, store
 
 
@@ -840,3 +843,24 @@
     assert caught.value is error
     assert not errors and not outcome and not store.rows
     turnstile.complete.assert_called_once_with('1700000000.wav')
+
+
+def test_process_segment_attaches_the_landed_row_to_the_live_episode(pipeline, monkeypatch):
+    module, _store = pipeline
+    attached = []
+    monkeypatch.setattr(module, 'sync_enabled_for', lambda _: True)
+    monkeypatch.setattr(module, '_download_audio_bytes', lambda _: b'synthetic')
+    monkeypatch.setattr(
+        module.final_pass_test_seam, 'run_sync_final_pass', lambda *a, **kw: (None, {'outcome': 'skipped_budget'})
+    )
+    monkeypatch.setattr(
+        module,
+        'attach_synced_at_intake',
+        lambda uid, cid, row: attached.append((uid, cid, dict(row))) or 'joined',
+    )
+    response = {'new_memories': set(), 'updated_memories': set()}
+    assert module.process_segment('1700000000.wav', 'u', response, threading.Lock(), [])
+    # The row the intake persisted is the row handed to the episode join.
+    assert len(attached) == 1
+    uid, conversation_id, row = attached[0]
+    assert uid == 'u' and row['id'] == conversation_id and conversation_id in response['new_memories']

backend/tests/unit/test_sync_cross_job_assignment.py

@@ -291,6 +291,10 @@ def independent_job_pipeline():
         pipeline = load_module_fresh(
             'utils.sync.pipeline', Path(__file__).resolve().parents[2] / 'utils/sync/pipeline.py'
         )
+        # The episodes intake attach is fail-open; the harness stand-in skips like
+        # a user whose app never called the episodes API, keeping the overlay's
+        # deferred service import out of this file's behavior timing budget.
+        pipeline.attach_synced_at_intake = lambda uid, cid, row: 'skipped_no_client'
         yield pipeline, transcript_segment, lifecycle
 
 

backend/tests/unit/test_sync_lineage_diagnostics.py

@@ -1,4 +1,4 @@
 """Bounded diagnostics for lineage resolution: stage counts, id probe, span buckets.
 
 The ``event=sync_lineage_resolve`` line reports how many returned candidate rows
 each filter stage dropped (never ids, text or field values), and a no-rows
@@ -794,3 +794,36 @@
     assert assigned['id'] == (landed['id'] if retry_anchor else LIVE_ID) and not created
     assert assigned['_sync_lineage_repeat_only'] is True
     assert assigned['_sync_lineage_stamp_gap_refused'] is True
+
+
+@pytest.mark.parametrize('repeat_only', [False, True])
+def test_complete_sync_intake_folds_a_landed_row_into_the_live_episode_before_downstream_work(repeat_only):
+    import threading
+
+    from utils.sync.intake_completion import complete_sync_intake
+
+    order = []
+    assigned = {'id': SENTINEL_ID, 'sync_relevance': 'keep'}
+    if repeat_only:
+        assigned['_sync_lineage_repeat_only'] = True
+    complete_sync_intake(
+        uid='synthetic-owner',
+        assigned=assigned,
+        created=not repeat_only,
+        survivors=[],
+        response={'new_memories': set(), 'updated_memories': set()},
+        lock=threading.Lock(),
+        language='en',
+        audio_enabled=True,
+        store_audio=lambda cid: order.append(('store_audio', cid)),
+        finish=lambda *a, **k: order.append(('finish',)),
+        mark_finalize=lambda: None,
+        acknowledge=lambda: None,
+        attach_episode=lambda uid, cid, row: order.append(('attach', uid, cid)) or 'joined',
+    )
+    # Membership exists before audio storage or enrichment can stall behind it;
+    # a transient repeat-only intake wrote nothing and owes no membership work.
+    if repeat_only:
+        assert order == [('store_audio', SENTINEL_ID)]
+    else:
+        assert order == [('attach', 'synthetic-owner', SENTINEL_ID), ('store_audio', SENTINEL_ID), ('finish',)]

backend/tests/unit/test_sync_lineage_flags_off_parity.py

@@ -320,10 +320,19 @@ def old_process(*args, lineage_binding=None, **kwargs):
                         'finish': calls,
                         # Shadow comparison telemetry is new; serving events,
                         # persisted rows and results must retain their old bytes.
+                        # The episodes intake attach (with its capability-marker
+                        # read) is likewise new additive telemetry, not a sync
+                        # serving change.
                         'logs': [
                             record.getMessage()
                             for record in caplog.records
-                            if not record.getMessage().startswith('event=refactor_shadow ')
+                            if not record.getMessage().startswith(
+                                (
+                                    'event=refactor_shadow ',
+                                    'event=episodes_client_seen_',
+                                    'event=episode_sync_intake_attach',
+                                )
+                            )
                         ],
                     }
                 )

backend/utils/episodes/ARCHITECTURE.md

@@ -53,7 +53,14 @@ one capture window's evidence.
      sealed with the user's key. A lost race retries once.
 3. When a conversation's finalize starts, its live episode closes
    (`service.close_on_finalize`, through `finalize.close_now`; see "Close and wrap-up").
-4. After a conversation's finalization job completes,
+4. A sync (WAL) conversation joins the live episode covering its span the
+   moment intake persisted it (`service.attach_synced_intake`, through
+   `finalize.attach_synced_at_intake`, called from the sync pipeline's
+   `complete_sync_intake`). The join is the cheap parallel-capture one below,
+   so Home folds the clip under the live row from its first appearance instead
+   of only after its enrichment finalizes. Fail-open and gated on the same
+   admission and client-capability markers as the finalize-time hook.
+5. After a conversation's finalization job completes,
    `utils/conversations/finalizer.py` calls `service.on_conversation_finalized`
    (bounded, fail-open, after `complete_finalization_fanout`). The synchronous
    `POST /v1/conversations` route runs the same hook off the request path
@@ -64,11 +71,15 @@ one capture window's evidence.
      per-conversation applied offset), to the conversation's own episode even
      when that one has closed.
    - A sync-finalized conversation that never attached (no realtime stream
-     carried it) joins the live episode covering its span first, as a parallel
-     capture: a member whose text the live stream already fed, so no moment and
-     no reconcile. It never closes the live episode, never opens one, and a
-     closed covering episode stays closed; a discarded conversation never
-     joins.
+     carried it, and intake did not or could not join it) joins the live
+     episode covering its span first, as a parallel capture: a member whose
+     text the live stream already fed, so no moment and no reconcile. It never
+     closes the live episode, never opens one, and a closed covering episode
+     stays closed; a discarded conversation never joins. A clip that started
+     before the episode's recorded start still joins when its span reaches the
+     episode or it began within `CONTINUE_GAP_SECONDS` of that start (the
+     pendant and the phone clock separately); one that ended earlier belongs
+     to an occasion already over.
    - It repoints a smart-merge donor to its survivor.
    - It reconciles the provisional items heard in that conversation against the
      tasks its normal processing saved: each becomes `kept` (with the task id)