fix(evaluation): keep verdicts through learning merges

reflexio/models/api_schema/domain/entities.py

@@ -454,8 +454,8 @@ class RetrievedLearningEvaluationResult(BaseModel):
         agent_version (str): Version supplied to group evaluation;
             informational, not part of the uniqueness key.
         kind (RetrievedLearningKind): The learning kind.
-        learning_id (str): Stable storage id, matching
-            ``RetrievedLearning.learning_id``.
+        learning_id (str): Current stable storage id. When an attached learning
+            has been merged or superseded, this is the live survivor's id.
         is_relevant (bool | None): Whether the learning applies to the
             session. ``None`` only when the relevance judge/chunk failed.
         relevance_reason (str): Judge reasoning; empty when ``is_relevant``

reflexio/server/services/agent_success_evaluation/components/retrieved_learning_evaluator.py

@@ -18,7 +18,7 @@
 
 import logging
 from dataclasses import dataclass, field
-from typing import TYPE_CHECKING, Any, Literal
+from typing import TYPE_CHECKING, Any, Literal, cast
 
 from pydantic import ConfigDict, Field
 
@@ -28,6 +28,7 @@
 )
 from reflexio.models.structured_output import StrictStructuredOutput
 from reflexio.server.llm.model_defaults import ModelRole, resolve_model_name
+from reflexio.server.services.lineage.resolve import EntityType, resolve_current
 from reflexio.server.services.service_utils import (
     log_llm_messages,
     log_model_response,
@@ -219,6 +220,10 @@ def evaluate(
         """
         diagnostics: dict[str, Any] = {
             "invalid_ref_count": 0,
+            "resolved_via_lineage": 0,
+            "unresolvable_ref_count": 0,
+            "purged_ref_count": 0,
+            "ineligible_ref_count": 0,
             "failed_relevance_chunks": 0,
             "failed_impact_chunks": 0,
         }
@@ -364,24 +369,37 @@ def _resolve_candidates(
         storage = self.request_context.storage
         if storage is None:
             return []
-        profile_ids = [lid for (kind, lid) in refs if kind == "profile"]
-        user_playbook_ids: list[int] = []
-        agent_playbook_ids: list[int] = []
+        current_refs: dict[tuple[str, str], None] = {}
         for kind, lid in refs:
-            if kind not in ("user_playbook", "agent_playbook"):
-                continue
-            try:
-                parsed = int(lid)
-            except ValueError:
-                diagnostics["invalid_ref_count"] += 1
+            lookup_id: str | int = lid
+            if kind in ("user_playbook", "agent_playbook"):
+                try:
+                    lookup_id = int(lid)
+                except ValueError:
+                    diagnostics["invalid_ref_count"] += 1
+                    continue
+                if lookup_id <= 0:
+                    diagnostics["invalid_ref_count"] += 1
+                    continue
+            current = resolve_current(storage, cast(EntityType, kind), lookup_id)
+            if current is None:
+                diagnostics["unresolvable_ref_count"] += 1
                 continue
-            if parsed <= 0:
-                diagnostics["invalid_ref_count"] += 1
+            if current.is_purged:
+                diagnostics["purged_ref_count"] += 1
                 continue
-            if kind == "user_playbook":
-                user_playbook_ids.append(parsed)
-            else:
-                agent_playbook_ids.append(parsed)
+            current_key = (kind, str(current.id))
+            if current_key != (kind, lid):
+                diagnostics["resolved_via_lineage"] += 1
+            current_refs.setdefault(current_key, None)
+
+        profile_ids = [lid for (kind, lid) in current_refs if kind == "profile"]
+        user_playbook_ids = [
+            int(lid) for (kind, lid) in current_refs if kind == "user_playbook"
+        ]
+        agent_playbook_ids = [
+            int(lid) for (kind, lid) in current_refs if kind == "agent_playbook"
+        ]
 
         resolved: dict[tuple[str, str], LearningCandidate] = {}
         if profile_ids:
@@ -420,8 +438,11 @@ def _resolve_candidates(
                 content=playbook.content,
                 trigger=playbook.trigger or "",
             )
-        # Preserve first-seen order of the attached refs.
-        return [resolved[key] for key in refs if key in resolved]
+        diagnostics["ineligible_ref_count"] += sum(
+            key not in resolved for key in current_refs
+        )
+        # Preserve first-seen order after refs that share a survivor collapse.
+        return [resolved[key] for key in current_refs if key in resolved]
 
     # ===============================
     # judging

reflexio/server/services/storage/sqlite_storage/playbook/_eval_results.py

@@ -2,12 +2,13 @@
 
 import sqlite3
 from datetime import UTC, datetime
-from typing import Any
+from typing import Any, cast
 
 from reflexio.models.api_schema.service_schemas import (
     AgentSuccessEvaluationResult,
     RetrievedLearningEvaluationResult,
 )
+from reflexio.server.services.lineage.resolve import EntityType, resolve_current
 
 from ...storage_base.retrieved_learning_state import (
     CANONICAL_RETRIEVED_KINDS,
@@ -35,6 +36,7 @@ class AgentEvaluationResultStoreMixin:
     # Type hints for instance attributes/methods provided by SQLiteStorageBase via MRO
     _lock: Any
     conn: sqlite3.Connection
+    org_id: str
     _execute: Any
     _fetchall: Any
     _fetchone: Any
@@ -256,18 +258,74 @@ def _rle_fingerprint_now(self, user_id: str, session_id: str) -> str:
         return builder.hexdigest()
 
     def _rle_attached_refs(self, user_id: str, session_id: str) -> set[tuple[str, str]]:
-        """Canonical ``(kind, learning_id)`` refs attached to the live session."""
+        """Return the live identities reached from the session's attachments."""
         cur = self.conn.execute(
             """SELECT i.retrieved_learnings
                FROM interactions i JOIN requests r ON i.request_id = r.request_id
                WHERE r.session_id = ? AND i.user_id = ?""",
             (session_id, user_id),
         )
-        attached: set[tuple[str, str]] = set()
+        original: set[tuple[str, str]] = set()
         for row in cur:
-            attached.update(_parse_attachment_refs(row["retrieved_learnings"]))
+            original.update(_parse_attachment_refs(row["retrieved_learnings"]))
+        attached: set[tuple[str, str]] = set()
+        for kind, learning_id in original:
+            try:
+                current = resolve_current(self, cast(EntityType, kind), learning_id)
+            except (TypeError, ValueError):
+                continue
+            if current is None:
+                continue
+            if not current.is_purged:
+                attached.add((kind, str(current.id)))
         return attached
 
+    def _rle_lineage_changed(self, refs: set[tuple[str, str]]) -> bool:
+        """Whether any evaluated identity now points elsewhere or is purged."""
+        for kind, learning_id in refs:
+            try:
+                current = resolve_current(self, cast(EntityType, kind), learning_id)
+            except (TypeError, ValueError):
+                continue
+            if current is None:
+                if self._rle_ref_has_history(kind, learning_id):
+                    return True
+                continue
+            if current.is_purged:
+                return True
+            if str(current.id) != learning_id:
+                return True
+        return False
+
+    def _rle_ref_has_history(self, kind: str, learning_id: str) -> bool:
+        """Whether an unresolvable identity previously existed or had lineage."""
+        table_and_key = {
+            "profile": ("profiles", "profile_id"),
+            "user_playbook": ("user_playbooks", "user_playbook_id"),
+            "agent_playbook": ("agent_playbooks", "agent_playbook_id"),
+        }.get(kind)
+        if table_and_key is None:
+            return False
+        table, key = table_and_key
+        if self.conn.execute(
+            f"SELECT 1 FROM {table} WHERE {key} = ? LIMIT 1",  # noqa: S608
+            (learning_id,),
+        ).fetchone():
+            return True
+        return (
+            self.conn.execute(
+                """SELECT 1 FROM lineage_event e
+                   WHERE e.org_id = ? AND e.entity_type = ?
+                     AND (e.entity_id = ? OR EXISTS (
+                         SELECT 1 FROM json_each(e.source_ids)
+                         WHERE CAST(value AS TEXT) = ?
+                     ))
+                   LIMIT 1""",
+                (self.org_id, kind, learning_id, learning_id),
+            ).fetchone()
+            is not None
+        )
+
     def _rle_eligible_refs(
         self, user_id: str, results: list[RetrievedLearningEvaluationResult]
     ) -> set[tuple[str, str]]:
@@ -461,6 +519,13 @@ def replace_retrieved_learning_evaluation_results(
                 # to trip the UNIQUE index and roll back the commit — caller
                 # bugs fail loud rather than silently dropping data.
                 attached = self._rle_attached_refs(user_id, session_id)
+                proposed = {(r.kind, r.learning_id) for r in results}
+                # A candidate that now resolves elsewhere changed while the
+                # judge was running. Retry instead of caching a false terminal
+                # ``not_applicable`` result.
+                if self._rle_lineage_changed(proposed):
+                    self.conn.rollback()
+                    return RetrievedLearningCommitResult(disposition="stale")
                 eligible = self._rle_eligible_refs(user_id, results)
                 kept = [
                     r

reflexio/server/services/storage/storage_base/playbook/_eval_results.py

@@ -224,10 +224,12 @@ def replace_retrieved_learning_evaluation_results(
         In one transaction: locks the session's state row, requires the
         current generation to equal ``generation`` (else ``superseded``),
         recomputes the session fingerprint from live rows and requires it to
-        equal ``session_fingerprint`` (else ``stale``), rechecks every
-        result's source row for retrieval eligibility (ineligible rows are
-        dropped), then deletes the prior session set, inserts the filtered
-        set, and persists completion state — all or nothing.
+        equal ``session_fingerprint`` (else ``stale``), requires every result
+        to match a session attachment after lineage resolution, and rechecks
+        its source row for retrieval eligibility. Rows that fail either check
+        are dropped before the prior session set is atomically replaced.
+        If a proposed identity no longer resolves to itself or is purged, the
+        method returns ``stale`` without replacing the prior result set.
 
         When commit-time eligibility removes every candidate, prior rows are
         cleared and the final status is ``not_applicable`` regardless of

tests/server/services/agent_success_evaluation/test_retrieved_learning_evaluator.py

@@ -11,6 +11,7 @@
 
 from reflexio.models.api_schema.domain import (
     AgentPlaybook,
+    LineageContext,
     PlaybookStatus,
     UserPlaybook,
     UserProfile,
@@ -192,13 +193,147 @@ def test_resolution_skips_missing_and_ineligible(storage: SQLiteStorage) -> None
         ("agent_playbook", str(apb_id)),
     }
     assert run.diagnostics["invalid_ref_count"] == 1
+    assert run.diagnostics["unresolvable_ref_count"] == 1
+    assert run.diagnostics["ineligible_ref_count"] == 1
     row = next(r for r in run.rows if r.kind == "profile")
     assert row.is_relevant is True and row.impact == "positive"
     assert row.agent_version == "evaluated-v2"
     assert row.created_at == 1_700_000_000
     bulk_agent_lookup.assert_called_once()
 
 
+def test_merged_user_playbook_is_judged_as_its_live_survivor(
+    storage: SQLiteStorage,
+) -> None:
+    source = UserPlaybook(
+        user_id=USER,
+        playbook_name="old checklist",
+        request_id="r-source",
+        agent_version="v1",
+        content="use the outdated deployment steps",
+    )
+    survivor = UserPlaybook(
+        user_id=USER,
+        playbook_name="current checklist",
+        request_id="r-survivor",
+        agent_version="v1",
+        content="use the current deployment steps",
+    )
+    storage.save_user_playbooks([source, survivor])
+    storage.merge_records(
+        entity_type="user_playbook",
+        survivor_id=str(survivor.user_playbook_id),
+        source_ids=[str(source.user_playbook_id)],
+        context=LineageContext(op_kind="merge", actor="test", request_id="r-merge"),
+    )
+    llm = _echoing_llm()
+
+    run = _make_evaluator(storage, llm).evaluate(
+        USER,
+        SESSION,
+        "v1",
+        _snapshot({1: [("user_playbook", str(source.user_playbook_id))]}),
+    )
+
+    assert [(row.kind, row.learning_id) for row in run.rows] == [
+        ("user_playbook", str(survivor.user_playbook_id))
+    ]
+    assert run.diagnostics["resolved_via_lineage"] == 1
+    prompts = [call.kwargs["messages"][0]["content"] for call in llm.mock_calls]
+    assert all("use the current deployment steps" in prompt for prompt in prompts)
+    assert all("use the outdated deployment steps" not in prompt for prompt in prompts)
+
+
+def test_merged_profile_refs_collapsing_to_one_survivor_are_judged_once(
+    storage: SQLiteStorage,
+) -> None:
+    profiles = [
+        UserProfile(
+            profile_id=profile_id,
+            user_id=USER,
+            content=content,
+            last_modified_timestamp=1,
+            generated_from_request_id="r1",
+        )
+        for profile_id, content in (
+            ("profile-source-1", "old preference one"),
+            ("profile-source-2", "old preference two"),
+            ("profile-survivor", "current preference"),
+        )
+    ]
+    storage.add_user_profile(USER, profiles)
+    storage.merge_records(
+        entity_type="profile",
+        survivor_id="profile-survivor",
+        source_ids=["profile-source-1", "profile-source-2"],
+        context=LineageContext(op_kind="merge", actor="test", request_id="r-merge"),
+    )
+    llm = _echoing_llm()
+
+    run = _make_evaluator(storage, llm).evaluate(
+        USER,
+        SESSION,
+        "v1",
+        _snapshot(
+            {
+                1: [
+                    ("profile", "profile-source-1"),
+                    ("profile", "profile-source-2"),
+                ]
+            }
+        ),
+    )
+
+    assert [(row.kind, row.learning_id) for row in run.rows] == [
+        ("profile", "profile-survivor")
+    ]
+    assert run.diagnostics["resolved_via_lineage"] == 2
+    assert llm.generate_chat_response.call_count == 2
+    prompts = [
+        call.kwargs["messages"][0]["content"]
+        for call in llm.generate_chat_response.call_args_list
+    ]
+    assert all("current preference" in prompt for prompt in prompts)
+    assert all("old preference" not in prompt for prompt in prompts)
+
+
+def test_purged_lineage_survivor_is_not_judged(storage: SQLiteStorage) -> None:
+    profiles = [
+        UserProfile(
+            profile_id=profile_id,
+            user_id=USER,
+            content=content,
+            last_modified_timestamp=1,
+            generated_from_request_id="r1",
+        )
+        for profile_id, content in (
+            ("profile-source", "old preference"),
+            ("profile-survivor", "current preference"),
+        )
+    ]
+    storage.add_user_profile(USER, profiles)
+    storage.merge_records(
+        entity_type="profile",
+        survivor_id="profile-survivor",
+        source_ids=["profile-source"],
+        context=LineageContext(op_kind="merge", actor="test", request_id="r-merge"),
+    )
+    assert storage.purge_content(entity_type="profile", entity_id="profile-survivor")
+    llm = _echoing_llm()
+
+    run = _make_evaluator(storage, llm).evaluate(
+        USER,
+        SESSION,
+        "v1",
+        _snapshot({1: [("profile", "profile-source")]}),
+    )
+
+    assert run.outcome == "evaluated"
+    assert run.rows == []
+    assert run.diagnostics["purged_ref_count"] == 1
+    llm.generate_chat_response.assert_not_called()
+
+
 def test_unapproved_agent_playbook_is_ineligible(storage: SQLiteStorage) -> None:
     storage.save_agent_playbooks(
         [