diff --git a/reflexio/server/services/playbook/components/aggregator.py b/reflexio/server/services/playbook/components/aggregator.py index 305a72c5..e06d7db1 100644 --- a/reflexio/server/services/playbook/components/aggregator.py +++ b/reflexio/server/services/playbook/components/aggregator.py @@ -1769,26 +1769,37 @@ def run(self, playbook_aggregator_request: PlaybookAggregatorRequest) -> dict: playbook_aggregator_request.rerun and self.aggregation_claim is not None ): - if not saved_fb.embedding: + # Mock mode clusters by trigger rather than by vector + # (see the MOCK_LLM_RESPONSE branch in + # ``get_clusters``), so no centroid exists + # to persist and cluster bookkeeping is skipped. Every + # other caller still aborts on a missing embedding: a + # centroid-less cluster row would silently break the + # incremental re-aggregation this table exists to feed. + if ( + not saved_fb.embedding + and os.getenv("MOCK_LLM_RESPONSE", "").lower() != "true" + ): raise RuntimeError( "rerun agent playbook has no centroid embedding" ) - cluster_id = self._stable_aggregation_cluster_id(fp_key) - self.storage.create_playbook_aggregation_cluster( # type: ignore[attr-defined] - cluster_id=cluster_id, - agent_version=self.agent_version, - agent_playbook_id=saved_fb.agent_playbook_id, - centroid_embedding=saved_fb.embedding, - member_count=len(raw_ids), - embedding_model=self.storage.embedding_model_name, - ) - self.storage.set_playbook_aggregation_disposition( # type: ignore[attr-defined] - self.agent_version, - raw_ids, - disposition="cluster_member", - cluster_id=cluster_id, - reason="full_rerun", - ) + if saved_fb.embedding: + cluster_id = self._stable_aggregation_cluster_id(fp_key) + self.storage.create_playbook_aggregation_cluster( # type: ignore[attr-defined] + cluster_id=cluster_id, + agent_version=self.agent_version, + agent_playbook_id=saved_fb.agent_playbook_id, + centroid_embedding=saved_fb.embedding, + member_count=len(raw_ids), + embedding_model=self.storage.embedding_model_name, + ) + self.storage.set_playbook_aggregation_disposition( # type: ignore[attr-defined] + self.agent_version, + raw_ids, + disposition="cluster_member", + cluster_id=cluster_id, + reason="full_rerun", + ) for prev_fp in previous_fingerprints_for_changed_clusters.get( fp_key, {} ): diff --git a/tests/server/services/playbook/test_cluster_change_detection.py b/tests/server/services/playbook/test_cluster_change_detection.py index 1e1765c1..3427ab35 100644 --- a/tests/server/services/playbook/test_cluster_change_detection.py +++ b/tests/server/services/playbook/test_cluster_change_detection.py @@ -5,6 +5,7 @@ and clustering stability. """ +from types import SimpleNamespace from unittest.mock import MagicMock, patch import numpy as np @@ -946,5 +947,100 @@ def test_adding_playbook_only_affects_its_cluster(self, mock_playbook_aggregator assert group_b_cluster_members == group_b_cluster_members2 +class TestRerunCentroidRequirement: + """The rerun path persists a cluster centroid, so it needs an embedding. + + The invariant was added when local embeddings were computed in-process, so + ``saved_fb.embedding`` was always populated and the raise was unreachable + outside a genuine bug. Once local inference moved behind a separate + service, an unreachable embedder became a normal runtime state: storage + logs "continuing without vector" and saves the playbook with no embedding, + and the invariant fired on infrastructure rather than on a bug -- rolling + back every ``run_playbook_aggregation`` call, which always sets + ``rerun=True``. Mock mode clusters by trigger and has no centroid to + persist at all, so it must skip the bookkeeping rather than abort. + """ + + def _run_rerun_aggregation(self, *, embedding): + group_a = create_similar_embeddings(3, base_seed=42) + group_b = create_similar_embeddings(3, base_seed=100) + user_playbooks = create_user_playbooks_with_embeddings(group_a + group_b) + # Mock mode clusters by trigger, not by vector, so a shared trigger is + # what groups these there. Vector clustering ignores the field. + for playbook in user_playbooks: + playbook.trigger = "When something happens" + + config = PlaybookAggregatorConfig( + min_cluster_size=2, reaggregation_trigger_count=1 + ) + mock_ctx = MagicMock() + mock_ctx.storage = MagicMock() + mock_ctx.configurator = MagicMock() + temp = PlaybookAggregator(MagicMock(), mock_ctx, "1.0") + prev_fingerprints = {} + for cluster_id, cluster_playbooks in temp.get_clusters( + user_playbooks, config + ).items(): + fp = PlaybookAggregator._compute_cluster_fingerprint(cluster_playbooks) + prev_fingerprints[fp] = { + "agent_playbook_id": cluster_id + 100, + "user_playbook_ids": sorted( + fb.user_playbook_id for fb in cluster_playbooks + ), + } + + harness = TestAggregatorRunWithChangeDetection() + aggregator, mock_storage, _llm = harness._setup_aggregator_for_run( + user_playbooks=user_playbooks, + operation_state=prev_fingerprints, + config=config, + ) + # The centroid branch is reached only under a held aggregation claim, + # and that path sources its corpus from the rerun snapshot rather than + # from get_user_playbooks. + aggregator.aggregation_claim = MagicMock() + mock_storage.capture_playbook_aggregation_rerun_snapshot.return_value = ( + SimpleNamespace( + user_playbooks=user_playbooks, + user_high_watermark=max(fb.user_playbook_id for fb in user_playbooks), + invalidation_ids=[], + ) + ) + + saved_count = [0] + + def save_side_effect(playbooks, **_kwargs): # noqa: ANN001 + saved_count[0] += 1 + playbook = playbooks[0] + playbook.agent_playbook_id = saved_count[0] + playbook.embedding = embedding + return [playbook] + + mock_storage.save_agent_playbooks.side_effect = save_side_effect + + aggregator.run(PlaybookAggregatorRequest(agent_version="1.0", rerun=True)) + return mock_storage + + def test_missing_centroid_aborts_the_rerun(self): + """Outside mock mode a centroid-less cluster row must never be written.""" + with pytest.raises(RuntimeError, match="no centroid embedding"): + self._run_rerun_aggregation(embedding=None) + + def test_mock_mode_skips_cluster_bookkeeping(self, monkeypatch): + """Mock mode reaches the save, then skips the centroid write.""" + monkeypatch.setenv("MOCK_LLM_RESPONSE", "true") + + mock_storage = self._run_rerun_aggregation(embedding=None) + + mock_storage.save_agent_playbooks.assert_called() + mock_storage.create_playbook_aggregation_cluster.assert_not_called() + + def test_present_centroid_persists_the_cluster(self): + """The happy path still records the cluster.""" + mock_storage = self._run_rerun_aggregation(embedding=[0.1] * 8) + + mock_storage.create_playbook_aggregation_cluster.assert_called() + + if __name__ == "__main__": pytest.main([__file__, "-v"])