diff --git a/reflexio/models/api_schema/domain/entities.py b/reflexio/models/api_schema/domain/entities.py index 5def6d07..bc0e2fac 100644 --- a/reflexio/models/api_schema/domain/entities.py +++ b/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`` diff --git a/reflexio/server/services/agent_success_evaluation/components/retrieved_learning_evaluator.py b/reflexio/server/services/agent_success_evaluation/components/retrieved_learning_evaluator.py index fc545428..71ef5a59 100644 --- a/reflexio/server/services/agent_success_evaluation/components/retrieved_learning_evaluator.py +++ b/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 diff --git a/reflexio/server/services/storage/sqlite_storage/playbook/_eval_results.py b/reflexio/server/services/storage/sqlite_storage/playbook/_eval_results.py index a012c261..90337d76 100644 --- a/reflexio/server/services/storage/sqlite_storage/playbook/_eval_results.py +++ b/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 diff --git a/reflexio/server/services/storage/storage_base/playbook/_eval_results.py b/reflexio/server/services/storage/storage_base/playbook/_eval_results.py index 71a70b52..700959ba 100644 --- a/reflexio/server/services/storage/storage_base/playbook/_eval_results.py +++ b/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 diff --git a/tests/server/services/agent_success_evaluation/test_retrieved_learning_evaluator.py b/tests/server/services/agent_success_evaluation/test_retrieved_learning_evaluator.py index eec19ec2..89762587 100644 --- a/tests/server/services/agent_success_evaluation/test_retrieved_learning_evaluator.py +++ b/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,6 +193,8 @@ 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" @@ -199,6 +202,138 @@ def test_resolution_skips_missing_and_ineligible(storage: SQLiteStorage) -> None 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( [ diff --git a/tests/server/services/agent_success_evaluation/test_retrieved_learning_runner_integration.py b/tests/server/services/agent_success_evaluation/test_retrieved_learning_runner_integration.py index a68bdaf8..d0d418db 100644 --- a/tests/server/services/agent_success_evaluation/test_retrieved_learning_runner_integration.py +++ b/tests/server/services/agent_success_evaluation/test_retrieved_learning_runner_integration.py @@ -19,6 +19,7 @@ from reflexio.models.api_schema.domain import ( AgentPlaybook, Interaction, + LineageContext, PlaybookStatus, Request, RetrievedLearning, @@ -190,6 +191,54 @@ def test_publish_to_verdicts_end_to_end(storage: SQLiteStorage) -> None: assert row.created_at == OLD_TS +def test_force_regenerate_replaces_merged_learning_with_its_survivor( + storage: SQLiteStorage, +) -> None: + source = UserPlaybook( + user_id=USER, + playbook_name="old checklist", + request_id="source-request", + agent_version="v1", + content="use the old deployment steps", + ) + survivor = UserPlaybook( + user_id=USER, + playbook_name="current checklist", + request_id="survivor-request", + agent_version="v1", + content="use the current deployment steps", + ) + storage.save_user_playbooks([source, survivor]) + _seed_session( + storage, + refs=[ + RetrievedLearning( + kind="user_playbook", learning_id=str(source.user_playbook_id) + ) + ], + ) + first = _run(storage) + assert first.retrieved_learning_status == "complete" + assert [ + row.learning_id + for row in storage.get_retrieved_learning_evaluation_results(session_id=SESSION) + ] == [str(source.user_playbook_id)] + + 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"), + ) + regenerated = _run(storage, force=True) + + assert regenerated.retrieved_learning_status == "complete" + assert [ + row.learning_id + for row in storage.get_retrieved_learning_evaluation_results(session_id=SESSION) + ] == [str(survivor.user_playbook_id)] + + def test_rerun_short_circuits_and_force_is_idempotent( storage: SQLiteStorage, ) -> None: diff --git a/tests/server/services/storage/test_storage_contract_retrieved_learning_evals.py b/tests/server/services/storage/test_storage_contract_retrieved_learning_evals.py index 43d2b4f4..8dd28893 100644 --- a/tests/server/services/storage/test_storage_contract_retrieved_learning_evals.py +++ b/tests/server/services/storage/test_storage_contract_retrieved_learning_evals.py @@ -17,6 +17,7 @@ from reflexio.models.api_schema.domain import ( Interaction, + LineageContext, Request, RetrievedLearning, RetrievedLearningEvaluationResult, @@ -177,6 +178,134 @@ def test_content_only_edit_invalidates_fingerprint(storage) -> None: assert commit.disposition == "stale" +def test_merge_during_evaluation_makes_the_commit_stale(storage) -> None: + source_id = _seed_eligible_learnings(storage) + survivor = UserPlaybook( + user_id=USER, + playbook_name="current checklist", + request_id="r2", + agent_version="v1", + content="use the current checklist", + ) + storage.save_user_playbooks([survivor]) + _seed_session( + storage, + refs=[RetrievedLearning(kind="user_playbook", learning_id=str(source_id))], + ) + snapshot = storage.load_bounded_retrieved_learning_snapshot(USER, SESSION) + fingerprint = session_fingerprint(snapshot) + generation = storage.begin_retrieved_learning_evaluation_run(USER, SESSION) + + storage.merge_records( + entity_type="user_playbook", + survivor_id=str(survivor.user_playbook_id), + source_ids=[str(source_id)], + context=LineageContext(op_kind="merge", actor="test", request_id="r-merge"), + ) + commit = storage.replace_retrieved_learning_evaluation_results( + USER, + SESSION, + generation, + fingerprint, + "complete", + {}, + [_result("user_playbook", str(source_id))], + ) + + assert commit.disposition == "stale" + assert storage.get_retrieved_learning_evaluation_results(session_id=SESSION) == [] + + +def test_survivor_merge_during_evaluation_makes_the_commit_stale(storage) -> None: + source_id = _seed_eligible_learnings(storage) + first_survivor = UserPlaybook( + user_id=USER, + playbook_name="current checklist", + request_id="r2", + agent_version="v1", + content="use the current checklist", + ) + final_survivor = UserPlaybook( + user_id=USER, + playbook_name="final checklist", + request_id="r3", + agent_version="v1", + content="use the final checklist", + ) + storage.save_user_playbooks([first_survivor, final_survivor]) + _seed_session( + storage, + refs=[RetrievedLearning(kind="user_playbook", learning_id=str(source_id))], + ) + storage.merge_records( + entity_type="user_playbook", + survivor_id=str(first_survivor.user_playbook_id), + source_ids=[str(source_id)], + context=LineageContext(op_kind="merge", actor="test", request_id="r-merge-1"), + ) + fingerprint = session_fingerprint( + storage.load_bounded_retrieved_learning_snapshot(USER, SESSION) + ) + generation = storage.begin_retrieved_learning_evaluation_run(USER, SESSION) + + storage.merge_records( + entity_type="user_playbook", + survivor_id=str(final_survivor.user_playbook_id), + source_ids=[str(first_survivor.user_playbook_id)], + context=LineageContext(op_kind="merge", actor="test", request_id="r-merge-2"), + ) + commit = storage.replace_retrieved_learning_evaluation_results( + USER, + SESSION, + generation, + fingerprint, + "complete", + {}, + [_result("user_playbook", str(first_survivor.user_playbook_id))], + ) + + assert commit.disposition == "stale" + assert storage.get_retrieved_learning_evaluation_results(session_id=SESSION) == [] + + +def test_delete_during_evaluation_makes_commit_stale_and_keeps_prior(storage) -> None: + _seed_eligible_learnings(storage) + _seed_session( + storage, + refs=[RetrievedLearning(kind="profile", learning_id="prof-1")], + ) + fingerprint = session_fingerprint( + storage.load_bounded_retrieved_learning_snapshot(USER, SESSION) + ) + first_generation = storage.begin_retrieved_learning_evaluation_run(USER, SESSION) + first = storage.replace_retrieved_learning_evaluation_results( + USER, + SESSION, + first_generation, + fingerprint, + "complete", + {}, + [_result("profile", "prof-1")], + ) + assert first.disposition == "applied" + + generation = storage.begin_retrieved_learning_evaluation_run(USER, SESSION) + assert storage.delete_profiles_by_ids(["prof-1"]) == 1 + commit = storage.replace_retrieved_learning_evaluation_results( + USER, + SESSION, + generation, + fingerprint, + "complete", + {}, + [_result("profile", "prof-1")], + ) + + assert commit.disposition == "stale" + kept = storage.get_retrieved_learning_evaluation_results(session_id=SESSION) + assert [(row.kind, row.learning_id) for row in kept] == [("profile", "prof-1")] + + def test_replace_persists_exact_eligible_set(storage) -> None: playbook_id = _seed_eligible_learnings(storage) _seed_session(