fix(embeddings): preserve recipe snapshot in status reads (#3951)

polylogue/storage/embeddings/status_payload.py

@@ -17,6 +17,7 @@
 
 from typing_extensions import TypedDict
 
+from polylogue.storage.embeddings.identity import EmbeddingRecipe
 from polylogue.storage.embeddings.materialization import (
     archive_embeddable_message_where,
     archive_embeddable_messages_relation,
@@ -475,6 +476,7 @@ def _archive_embedding_session_state_exact_with_timeout(
     *,
     status_table: str,
     timeout_ms: int,
+    recipe: EmbeddingRecipe,
 ) -> tuple[int, int, int] | None:
     """Return exact embedded/pending/blocked counts, or ``None`` when too costly."""
 
@@ -487,7 +489,12 @@ def _interrupt_when_expired() -> int:
 
     conn.set_progress_handler(_interrupt_when_expired, 10_000)
     try:
-        session_state = count_archive_embedding_session_state(conn, status_table=status_table, rebuild=False)
+        session_state = count_archive_embedding_session_state(
+            conn,
+            status_table=status_table,
+            rebuild=False,
+            recipe=recipe,
+        )
     except sqlite3.OperationalError as exc:
         message = str(exc).lower()
         if is_missing_table_error(exc):
@@ -721,6 +728,10 @@ def _archive_embedding_status_payload(
     index_db = _archive_index_path(db_path)
     if index_db is None:
         return None
+    recipe = EmbeddingRecipe.current(
+        model=str(getattr(cfg, "embedding_model", "")),
+        dimensions=_payload_int(getattr(cfg, "embedding_dimension", 0)),
+    )
     root = configured_root if configured_root is not None else db_path.parent
     conn = open_readonly_connection(index_db, timeout=STATUS_READ_BUSY_TIMEOUT_MS / 1000.0)
     conn.execute(f"PRAGMA busy_timeout = {STATUS_READ_BUSY_TIMEOUT_MS}")
@@ -768,6 +779,7 @@ def _archive_embedding_status_payload(
             conn,
             status_table=status_table,
             timeout_ms=DETAIL_QUERY_TIMEOUT_MS if include_detail else METADATA_SUMMARY_TIMEOUT_MS,
+            recipe=recipe,
         )
         pending_messages_exact = include_detail
         if exact_session_state is None:

tests/unit/storage/test_embedding_contracts.py

@@ -36,7 +36,7 @@
 
 
 class _FakeV1VectorProvider:
-    model = "voyage-4"
+    model = "voyage-4-lite"
     dimension = 1024
 
     def __init__(self) -> None:

tests/unit/storage/test_embedding_freshness_invariant.py

@@ -87,6 +87,18 @@ def __init__(self, *, model: str = "voyage-4", api_key: str | None = "test-key")
         self.embedding_max_cost_usd = 0.0
 
 
+@pytest.fixture(autouse=True)
+def _materialization_uses_freshness_baseline_recipe(monkeypatch: pytest.MonkeyPatch) -> None:
+    """Keep initial materialization on the test's explicit voyage-4 baseline.
+
+    Individual tests replace this patch with voyage-5 to prove a recipe change
+    invalidates stale vectors before it can be published as current.
+    """
+    from polylogue.storage.embeddings import materialization
+
+    monkeypatch.setattr(materialization, "load_polylogue_config", lambda: _EmbeddingConfig())
+
+
 def _write_archive_session(root: Path, *, native_id: str, text: str) -> str:
     with ArchiveStore(root) as archive:
         return archive.write_parsed(
@@ -268,6 +280,26 @@ def test_recipe_model_swap_makes_every_materialized_session_stale(tmp_path: Path
     assert {item.session_id for item in swapped} == set(session_ids)
 
 
+def test_status_payload_uses_its_resolved_recipe_for_exact_archive_counts(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    """The operator status route must not reload a different ambient recipe."""
+    from polylogue import config as config_module
+    from polylogue.storage.embeddings.status_payload import embedding_status_payload
+
+    root = tmp_path / "archive"
+    session_id = _write_archive_session(root, native_id="status-recipe", text=_INITIAL_TEXT)
+    initialize_archive_database(root / "embeddings.db", ArchiveTier.EMBEDDINGS)
+    assert embed_archive_session_sync(root / "index.db", _FakeVectorProvider(), session_id).status == "embedded"
+
+    monkeypatch.setattr(config_module, "load_polylogue_config", lambda: _EmbeddingConfig(model="voyage-5"))
+    payload = embedding_status_payload(SimpleNamespace(config=SimpleNamespace(db_path=root / "index.db")))
+
+    assert payload["configured_model"] == "voyage-5"
+    assert payload["embedded_sessions"] == 0
+    assert payload["pending_sessions"] == 1
+
+
 def test_config_change_then_old_terminal_error_cannot_clear_new_generation(
     tmp_path: Path, monkeypatch: pytest.MonkeyPatch
 ) -> None: