fix(backend): drop only the ungrounded memory candidate, not the whole run (#11685) (#11688)

Canonical L1 extraction raised ValueError as soon as any single candidate's
evidence quotes failed to bind to exactly one transcript segment. That verdict
is per candidate, but raising it aborted the entire run — and, because
_extract_memories runs synchronously inside process_conversation, the rest of
conversation finalization with it (action items, goal progress, the deferred
flag reset, the desktop meeting Chat receipt).

In prod over 2026-08-16 03:00Z-15:30Z that was 27 of 39 failed lazy
enrichments (42 completed in the same window), plus a continuous stream on
backend-sync where the merged conversation just keeps its stale enrichment.
Grounding is not systematically broken there: 332 conversations saved >=1
canonical memory in that window and 209 of them saved >=2, so candidates fail
grounding individually.

Skip the ungrounded candidate instead. The capture fence is unchanged —
nothing unbound is ever written — and the run's grounded siblings survive.
Every candidate failing to ground still raises: replace_conversation_memories
is called unconditionally, so continuing there would submit an empty
replacement and retract the source's existing memories on the strength of a
run we just decided not to trust.

Failure-Class: none

Author: kodjima33

backend/tests/unit/test_memory_replace_policy.py

@@ -225,6 +225,70 @@ def test_canonical_capture_preserves_prior_state_when_candidate_has_any_unground
     mock_service.replace_conversation_memories.assert_not_called()
 
 
+def test_canonical_capture_drops_only_the_ungrounded_candidate(monkeypatch):
+    """One ungrounded candidate must not discard its grounded siblings."""
+    pc = _load_process_conversation()
+    from models.conversation import Conversation
+    from models.conversation_enums import CategoryEnum, ConversationSource
+    from models.structured import Structured
+    from models.transcript_segment import TranscriptSegment
+
+    mock_service = MagicMock()
+    monkeypatch.setattr(pc, "MemoryService", lambda db_client: mock_service)
+    monkeypatch.setattr(
+        pc,
+        "extract_canonical_l1_memory_candidates",
+        MagicMock(
+            return_value=[
+                SimpleNamespace(
+                    content="The user works at Acme.",
+                    evidence_quotes=["I work at Acme"],
+                    speaker_label="SPEAKER_00",
+                    speaker_scope="session-local",
+                    about="the user",
+                    risk_flags=[],
+                    archive_class="general",
+                ),
+                SimpleNamespace(
+                    content="The user was diagnosed with condition X.",
+                    evidence_quotes=["I was diagnosed with condition X"],
+                    speaker_label="SPEAKER_00",
+                    speaker_scope="session-local",
+                    about="the user",
+                    risk_flags=[],
+                    archive_class="general",
+                ),
+            ]
+        ),
+    )
+    monkeypatch.setattr(pc.users_db, "get_user_language_preference", lambda uid: "en")
+
+    conversation = Conversation(
+        id="conv-partially-grounded",
+        created_at=datetime(2026, 6, 1, tzinfo=timezone.utc),
+        started_at=datetime(2026, 6, 1, tzinfo=timezone.utc),
+        finished_at=datetime(2026, 6, 1, 1, tzinfo=timezone.utc),
+        source=ConversationSource.omi,
+        structured=Structured(title="Test", overview="Overview", category=CategoryEnum.personal),
+        transcript_segments=[
+            TranscriptSegment(
+                text="I work at Acme and we talked about the grocery list.",
+                speaker="SPEAKER_00",
+                is_user=True,
+                start=0.0,
+                end=4.0,
+            )
+        ],
+    )
+
+    result = pc._extract_memories_canonical("uid-partial-grounding", conversation, db_client=MagicMock())
+
+    assert result.count == 1
+    replacement_payloads = mock_service.replace_conversation_memories.call_args.args[2]
+    assert len(replacement_payloads) == 1
+    assert replacement_payloads[0]["content"] == "The user works at Acme."
+
+
 @pytest.mark.parametrize(
     "matched_segments",
     [

backend/utils/conversations/process_conversation.py

@@ -909,13 +909,19 @@ def _extract_memories_canonical(
             language=language,
             strict=True,
         )
+        ungrounded_candidates = 0
         for candidate in extracted_candidates:
             evidence_quotes = _grounded_l1_evidence_quotes(
                 candidate.evidence_quotes,
                 conversation.transcript_segments,
             )
             if not evidence_quotes:
-                raise ValueError("canonical memory extraction returned evidence without a unique source binding")
+                # A quote that binds to no single segment is a verdict on this
+                # candidate, not on the run. Dropping it keeps the capture fence
+                # intact — nothing unbound is written — without discarding the
+                # grounded siblings and the rest of conversation finalization.
+                ungrounded_candidates += 1
+                continue
             subject_entity_id, subject_attribution, subject_kind = _l1_candidate_subject(
                 source_id=conversation.id,
                 about=candidate.about,
@@ -943,6 +949,25 @@ def _extract_memories_canonical(
                     True,
                 )
             )
+        if extracted_candidates and not capture_candidates:
+            # Every candidate failed grounding: the run itself is untrustworthy,
+            # so it must not submit the empty replacement that would retract the
+            # source's existing memories.
+            raise ValueError("canonical memory extraction returned evidence without a unique source binding")
+        if ungrounded_candidates:
+            logger.warning(
+                "canonical memory extraction dropped %s of %s ungrounded candidates conversation=%s",
+                ungrounded_candidates,
+                len(extracted_candidates),
+                conversation.id,
+            )
+            record_fallback(
+                component='other',
+                from_mode='canonical_memory_extraction',
+                to_mode='grounded_candidates_only',
+                reason='other',
+                outcome='degraded',
+            )
 
     is_locked = conversation.is_locked
     parsed_memories: List[Tuple[MemoryDB, List[str], str, List[str]]] = []