From 1d9f0e87ecbaca3387259e8da8888fef1b6695b9 Mon Sep 17 00:00:00 2001 From: Sinity Date: Sun, 12 Jul 2026 23:25:06 +0200 Subject: [PATCH 01/16] fix(embeddings): require active index generation for cleanup Problem: a matching index schema alone does not prove an inactive\nblue-green rebuild candidate is safe deletion truth for embeddings.\n\nWhat changed: require generation-managed cleanup to use the active\nsource-snapshotted index generation, before deletion and again before\ncommit. Legacy direct-index archives retain the existing schema and inode\nguards.\n\nVerification: pending focused embedding storage test; host PSI is currently\ntoo high for a safe pytest run.\n\nRef polylogue-1dk1\n\nCo-Authored-By: Claude --- polylogue/storage/embeddings/reconcile.py | 49 +++++++++++++++++++ .../test_embedding_orphan_reconcile.py | 47 ++++++++++++++++++ 2 files changed, 96 insertions(+) diff --git a/polylogue/storage/embeddings/reconcile.py b/polylogue/storage/embeddings/reconcile.py index 098aa0b27d..987bbdd72c 100644 --- a/polylogue/storage/embeddings/reconcile.py +++ b/polylogue/storage/embeddings/reconcile.py @@ -40,6 +40,7 @@ from __future__ import annotations +import json import sqlite3 import time from dataclasses import dataclass @@ -247,6 +248,7 @@ def reconcile_embedding_orphans( "embedding orphan reconciliation apply requires an authoritative index schema: " f"active index is v{actual_index_schema_version}, packaged index is v{INDEX_SCHEMA_VERSION}" ) + _assert_active_index_generation(index_path) conn.execute("BEGIN" if dry_run else "BEGIN IMMEDIATE") scanned_message_meta_rows = _scalar(conn, "SELECT COUNT(*) FROM message_embeddings_meta") @@ -299,6 +301,7 @@ def reconcile_embedding_orphans( if not dry_run: _assert_index_identity(index_path, expected_index_identity) + _assert_active_index_generation(index_path) affected_sessions: set[str] = set() for row in limited_message: message_id = str(row["message_id"]) @@ -337,6 +340,7 @@ def reconcile_embedding_orphans( removed_status_rows += max(0, cursor.rowcount) _assert_index_identity(index_path, expected_index_identity) + _assert_active_index_generation(index_path) conn.commit() samples: list[EmbeddingOrphanSample] = [ @@ -482,6 +486,51 @@ def _assert_index_identity(index_path: Path, expected: _IndexIdentity) -> None: ) +def _assert_active_index_generation(index_path: Path) -> None: + """Require the active source-snapshotted generation when generations exist. + + Legacy archives have a direct ``index.db`` and no generation metadata, so + the schema and inode guards remain their authority proof. Once a rebuild + creates generation metadata, however, a same-schema inactive candidate is + not safe deletion truth: only the generation named by the active pointer + and marked ``active`` after a source snapshot may drive reconciliation. + """ + + root = index_path.parent + generations = root / ".index-generations" + metadata_paths = tuple(generations.glob("*/generation.json")) if generations.is_dir() else () + if not metadata_paths: + return + + pointer = root / ".index-active-pointer" + if not pointer.is_file(): + raise RuntimeError("embedding orphan reconciliation requires an active index generation pointer") + try: + pointed_path = Path(pointer.read_text(encoding="utf-8").strip()).resolve(strict=True) + except (OSError, ValueError) as exc: + raise RuntimeError( + "embedding orphan reconciliation found an unreadable active index generation pointer" + ) from exc + if pointed_path != index_path.resolve(strict=True): + raise RuntimeError("embedding orphan reconciliation refuses a non-active index generation") + + for metadata_path in metadata_paths: + try: + payload = json.loads(metadata_path.read_text(encoding="utf-8")) + generation_path = Path(str(payload["index_path"])).resolve(strict=True) + except (KeyError, OSError, TypeError, ValueError, json.JSONDecodeError): + continue + if generation_path == pointed_path: + if ( + payload.get("state") == "active" + and isinstance(payload.get("source_snapshot"), str) + and payload["source_snapshot"] + ): + return + raise RuntimeError("embedding orphan reconciliation requires an active source-snapshotted index generation") + raise RuntimeError("embedding orphan reconciliation active index is missing generation readiness evidence") + + def _is_recent(timestamp_ms: int | None, now_ms: int, quiet_window_ms: int) -> bool: if timestamp_ms is None: return False diff --git a/tests/unit/storage/test_embedding_orphan_reconcile.py b/tests/unit/storage/test_embedding_orphan_reconcile.py index e65399413e..1fad67d506 100644 --- a/tests/unit/storage/test_embedding_orphan_reconcile.py +++ b/tests/unit/storage/test_embedding_orphan_reconcile.py @@ -11,6 +11,7 @@ from __future__ import annotations +import json import sqlite3 from pathlib import Path from unittest.mock import patch @@ -650,3 +651,49 @@ def test_index_identity_is_revalidated_before_atomic_commit(tmp_path: Path) -> N (session_id,), ).fetchone() assert status is not None and tuple(status) == (1, 0) + + +def test_apply_refuses_inactive_generation_even_when_schema_matches(tmp_path: Path) -> None: + """Only the source-snapshotted active generation may authorize deletion. + + Anti-vacuity: removing the generation-authority guard lets an inactive + v35 rebuild candidate delete this real orphan vector and metadata row. + """ + session_id = "codex-session:inactive-generation" + message_id = f"{session_id}:orphan" + index_db = tmp_path / "index.db" + _connect_index(index_db, sessions=[session_id], messages={session_id: []}) + generations = tmp_path / ".index-generations" / "gen-inactive" + generations.mkdir(parents=True) + (tmp_path / ".index-active-pointer").write_text(str(index_db.resolve()), encoding="utf-8") + (generations / "generation.json").write_text( + json.dumps( + { + "generation_id": "gen-inactive", + "owner_id": "test", + "archive_root": str(tmp_path), + "index_path": str(index_db), + "state": "inactive", + "created_at_ms": _NOW_MS, + "source_snapshot": "source-at-rebuild-start", + } + ), + encoding="utf-8", + ) + embeddings_db = tmp_path / "embeddings.db" + conn = _connect_embeddings(embeddings_db) + _write_embedding(conn, message_id=message_id, session_id=session_id, embedded_at_ms=_OLD_MS) + conn.close() + + with pytest.raises(RuntimeError, match="active source-snapshotted index generation"): + reconcile_embedding_orphans( + index_db, + embeddings_db, + dry_run=False, + now_ms=_NOW_MS, + mutation_authority="offline-exclusive", + ) + + with _connect_embeddings(embeddings_db) as verify: + assert verify.execute("SELECT COUNT(*) FROM message_embeddings_meta").fetchone()[0] == 1 + assert verify.execute("SELECT COUNT(*) FROM message_embeddings").fetchone()[0] == 1 From 68a196cfc19c6cd317e7eada8c2b31ec24908d19 Mon Sep 17 00:00:00 2001 From: Sinity Date: Sun, 12 Jul 2026 23:29:25 +0200 Subject: [PATCH 02/16] feat(embeddings): preserve terminal failure lifecycle Problem: terminal provider failures were represented only by a status-row\nerror string, so operators could not inspect identities or resolve debt\nwithout losing evidence.\n\nWhat changed: add a rebuildable embedding-failure ledger with source refs,\nprovider/model/error classification, explicit lifecycle transitions, and\nrequeue semantics. Archive materialization records the attempted batch and\nmarks open failures resolved after a later successful embed.\n\nVerification: Ruff format and lint pass. Focused embedding tests pending the\nnext safe host-resource window.\n\nRef polylogue-egm8\n\nCo-Authored-By: Claude --- .../storage/embeddings/materialization.py | 26 ++- .../sqlite/archive_tiers/embedding_write.py | 212 ++++++++++++++++++ .../sqlite/archive_tiers/embeddings.py | 26 ++- .../test_archive_tiers_embedding_write.py | 82 +++++++ 4 files changed, 343 insertions(+), 3 deletions(-) diff --git a/polylogue/storage/embeddings/materialization.py b/polylogue/storage/embeddings/materialization.py index 507eb50a71..55d87a0b74 100644 --- a/polylogue/storage/embeddings/materialization.py +++ b/polylogue/storage/embeddings/materialization.py @@ -75,6 +75,19 @@ def is_terminal_embedding_provider_error(error_message: object) -> bool: return any(marker in normalized for marker in TERMINAL_PROVIDER_ERROR_MARKERS) +def embedding_error_class(error_message: object) -> str: + """Classify provider failures without discarding their original evidence.""" + + normalized = " ".join(str(error_message).lower().split()) + if "http 400" in normalized or "status 400" in normalized or "400 bad request" in normalized: + return "provider_http_400" + if "http 429" in normalized or "status 429" in normalized: + return "provider_http_429" + if "timeout" in normalized: + return "provider_timeout" + return "provider_error" + + def archive_embeddable_message_where(alias: str = "m") -> str: """SQL predicate for authored prose messages eligible for embedding.""" @@ -906,6 +919,7 @@ def embed_archive_session_sync( index_conn = sqlite3.connect(f"file:{index_db_path}?mode=ro", uri=True, timeout=30.0) index_conn.row_factory = sqlite3.Row embeddings_conn = sqlite3.connect(embeddings_db_path, timeout=30.0) + attempted_message_refs: tuple[str, ...] = () try: from polylogue.storage.sqlite.sqlite_vec_extension import try_load_sqlite_vec @@ -970,6 +984,7 @@ def embed_archive_session_sync( batch_size = max(1, ARCHIVE_EMBED_MESSAGE_BATCH_SIZE) for start in range(0, len(embeddable), batch_size): batch = embeddable[start : start + batch_size] + attempted_message_refs = tuple(str(row["message_id"]) for row in batch) embeddings = text_provider._get_embeddings([str(row["text"]) for row in batch], input_type="document") if len(embeddings) != len(batch): raise RuntimeError("embedding provider returned a mismatched vector count") @@ -996,18 +1011,25 @@ def embed_archive_session_sync( message_count=len(embeddable), model=text_provider.model, ) + from polylogue.storage.sqlite.archive_tiers.embedding_write import resolve_open_embedding_failures_for_session + + resolve_open_embedding_failures_for_session(embeddings_conn, session_id=session_id) except Exception as exc: try: - from polylogue.storage.sqlite.archive_tiers.embedding_write import mark_session_embedding_error + from polylogue.storage.sqlite.archive_tiers.embedding_write import record_embedding_failure origin_row = index_conn.execute( "SELECT origin FROM sessions WHERE session_id = ?", (session_id,) ).fetchone() if origin_row is not None: - mark_session_embedding_error( + record_embedding_failure( embeddings_conn, session_id=session_id, origin=str(origin_row["origin"]), + message_refs=attempted_message_refs, + provider="voyage", + model=text_provider.model, + error_class=embedding_error_class(exc), error_message=str(exc), retryable=not is_terminal_embedding_provider_error(str(exc)), ) diff --git a/polylogue/storage/sqlite/archive_tiers/embedding_write.py b/polylogue/storage/sqlite/archive_tiers/embedding_write.py index 9975453e79..373b4b4bf5 100644 --- a/polylogue/storage/sqlite/archive_tiers/embedding_write.py +++ b/polylogue/storage/sqlite/archive_tiers/embedding_write.py @@ -5,9 +5,13 @@ from __future__ import annotations +import json import sqlite3 +import time +import uuid from collections.abc import Sequence from dataclasses import dataclass +from typing import Literal from polylogue.core.enums import Origin from polylogue.storage.search_providers.sqlite_vec_support import _serialize_f32 @@ -45,6 +49,30 @@ class ArchiveEmbeddingWrite: content_hash: bytes +EmbeddingFailureState = Literal["retryable", "terminal", "acknowledged", "superseded", "resolved"] +EmbeddingFailureResolution = Literal["acknowledge", "requeue", "supersede"] + + +@dataclass(frozen=True, slots=True) +class ArchiveEmbeddingFailure: + failure_id: str + session_id: str + origin: str + message_refs: tuple[str, ...] + provider: str + model: str + error_class: str + error_message: str + retryable: bool + lifecycle_state: EmbeddingFailureState + created_at_ms: int + updated_at_ms: int + resolved_at_ms: int | None + resolution_action: str | None + resolution_note: str | None + superseded_by: str | None + + def upsert_message_embedding( conn: sqlite3.Connection, *, @@ -145,6 +173,182 @@ def mark_session_embedding_error( return read_embedding_status(conn, session_id) +def record_embedding_failure( + conn: sqlite3.Connection, + *, + session_id: str, + origin: Origin | str, + message_refs: Sequence[str], + provider: str, + model: str, + error_class: str, + error_message: str, + retryable: bool, + occurred_at_ms: int | None = None, +) -> ArchiveEmbeddingFailure: + """Persist one inspectable failure event and its current retry lifecycle.""" + + now_ms = int(time.time() * 1000) if occurred_at_ms is None else occurred_at_ms + failure_id = f"embedding-failure:{uuid.uuid4()}" + state: EmbeddingFailureState = "retryable" if retryable else "terminal" + origin_value = _enum_value(origin) + refs = tuple(dict.fromkeys(str(ref) for ref in message_refs)) + with conn: + conn.execute( + """ + INSERT INTO embedding_status ( + session_id, origin, message_count_embedded, last_embedded_at_ms, needs_reindex, error_message + ) VALUES (?, ?, 0, NULL, ?, ?) + ON CONFLICT(session_id) DO UPDATE SET + origin = excluded.origin, + needs_reindex = excluded.needs_reindex, + error_message = excluded.error_message + """, + (session_id, origin_value, 1 if retryable else 0, error_message), + ) + conn.execute( + """ + INSERT INTO embedding_failures ( + failure_id, session_id, origin, message_refs_json, provider, model, + error_class, error_message, retryable, lifecycle_state, created_at_ms, updated_at_ms + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + """, + ( + failure_id, + session_id, + origin_value, + json.dumps(refs), + provider, + model, + error_class, + error_message, + int(retryable), + state, + now_ms, + now_ms, + ), + ) + return read_embedding_failure(conn, failure_id) + + +def resolve_embedding_failure( + conn: sqlite3.Connection, + *, + failure_id: str, + action: EmbeddingFailureResolution, + note: str | None = None, + superseded_by: str | None = None, + resolved_at_ms: int | None = None, +) -> ArchiveEmbeddingFailure: + """Explicitly acknowledge, supersede, or requeue an active failure.""" + + now_ms = int(time.time() * 1000) if resolved_at_ms is None else resolved_at_ms + state: EmbeddingFailureState = { + "acknowledge": "acknowledged", + "supersede": "superseded", + "requeue": "resolved", + }[action] + with conn: + row = conn.execute( + "SELECT session_id FROM embedding_failures WHERE failure_id = ? AND lifecycle_state IN ('retryable', 'terminal')", + (failure_id,), + ).fetchone() + if row is None: + raise KeyError(failure_id) + conn.execute( + """ + UPDATE embedding_failures + SET lifecycle_state = ?, updated_at_ms = ?, resolved_at_ms = ?, resolution_action = ?, + resolution_note = ?, superseded_by = ? + WHERE failure_id = ? + """, + (state, now_ms, now_ms, action, note, superseded_by, failure_id), + ) + if action == "requeue": + conn.execute( + "UPDATE embedding_status SET needs_reindex = 1, error_message = NULL WHERE session_id = ?", + (str(row[0]),), + ) + return read_embedding_failure(conn, failure_id) + + +def resolve_open_embedding_failures_for_session( + conn: sqlite3.Connection, *, session_id: str, resolved_at_ms: int | None = None +) -> int: + """Preserve prior failures while marking a later successful embedding as resolution.""" + + now_ms = int(time.time() * 1000) if resolved_at_ms is None else resolved_at_ms + with conn: + cursor = conn.execute( + """ + UPDATE embedding_failures + SET lifecycle_state = 'resolved', updated_at_ms = ?, resolved_at_ms = ?, resolution_action = 'embedded' + WHERE session_id = ? AND lifecycle_state IN ('retryable', 'terminal') + """, + (now_ms, now_ms, session_id), + ) + return max(0, cursor.rowcount) + + +def list_active_embedding_failures(conn: sqlite3.Connection, *, limit: int = 25) -> tuple[ArchiveEmbeddingFailure, ...]: + """Return bounded current failure identities for status and agent surfaces.""" + + rows = conn.execute( + """ + SELECT failure_id, session_id, origin, message_refs_json, provider, model, error_class, error_message, + retryable, lifecycle_state, created_at_ms, updated_at_ms, resolved_at_ms, resolution_action, + resolution_note, superseded_by + FROM embedding_failures + WHERE lifecycle_state IN ('retryable', 'terminal') + ORDER BY updated_at_ms DESC, failure_id ASC + LIMIT ? + """, + (max(0, limit),), + ).fetchall() + return tuple(_failure_from_row(row) for row in rows) + + +def read_embedding_failure(conn: sqlite3.Connection, failure_id: str) -> ArchiveEmbeddingFailure: + row = conn.execute( + """ + SELECT failure_id, session_id, origin, message_refs_json, provider, model, error_class, error_message, + retryable, lifecycle_state, created_at_ms, updated_at_ms, resolved_at_ms, resolution_action, + resolution_note, superseded_by + FROM embedding_failures WHERE failure_id = ? + """, + (failure_id,), + ).fetchone() + if row is None: + raise KeyError(failure_id) + return _failure_from_row(row) + + +def _failure_from_row(row: sqlite3.Row | tuple[object, ...]) -> ArchiveEmbeddingFailure: + message_refs_raw = row[3] + try: + message_refs = tuple(str(item) for item in json.loads(str(message_refs_raw))) + except (TypeError, ValueError, json.JSONDecodeError): + message_refs = () + return ArchiveEmbeddingFailure( + failure_id=str(row[0]), + session_id=str(row[1]), + origin=str(row[2]), + message_refs=message_refs, + provider=str(row[4]), + model=str(row[5]), + error_class=str(row[6]), + error_message=str(row[7]), + retryable=bool(row[8]), + lifecycle_state=str(row[9]), + created_at_ms=int(row[10]), + updated_at_ms=int(row[11]), + resolved_at_ms=None if row[12] is None else int(row[12]), + resolution_action=None if row[13] is None else str(row[13]), + resolution_note=None if row[14] is None else str(row[14]), + superseded_by=None if row[15] is None else str(row[15]), + ) + + def read_embedding_meta(conn: sqlite3.Connection, target_id: str) -> ArchiveEmbeddingMeta: conn.row_factory = sqlite3.Row row = conn.execute( @@ -196,11 +400,19 @@ def _enum_value(value: object) -> str: __all__ = [ "ArchiveEmbeddingMeta", + "ArchiveEmbeddingFailure", + "EmbeddingFailureResolution", + "EmbeddingFailureState", "ArchiveEmbeddingStatus", "ArchiveEmbeddingWrite", + "list_active_embedding_failures", "mark_session_embedding_error", + "read_embedding_failure", "read_embedding_meta", "read_embedding_status", "upsert_message_embedding", "upsert_message_embeddings", + "record_embedding_failure", + "resolve_embedding_failure", + "resolve_open_embedding_failures_for_session", ] diff --git a/polylogue/storage/sqlite/archive_tiers/embeddings.py b/polylogue/storage/sqlite/archive_tiers/embeddings.py index 3ede7dd196..842807e01f 100644 --- a/polylogue/storage/sqlite/archive_tiers/embeddings.py +++ b/polylogue/storage/sqlite/archive_tiers/embeddings.py @@ -2,7 +2,7 @@ from __future__ import annotations -EMBEDDINGS_SCHEMA_VERSION = 1 +EMBEDDINGS_SCHEMA_VERSION = 2 EMBEDDING_DIMENSION = 1024 EMBEDDINGS_DDL = f""" @@ -31,6 +31,30 @@ error_message TEXT ) STRICT; +CREATE TABLE IF NOT EXISTS embedding_failures ( + failure_id TEXT PRIMARY KEY, + session_id TEXT NOT NULL, + origin TEXT NOT NULL, + message_refs_json TEXT NOT NULL DEFAULT '[]', + provider TEXT NOT NULL, + model TEXT NOT NULL, + error_class TEXT NOT NULL, + error_message TEXT NOT NULL, + retryable INTEGER NOT NULL CHECK(retryable IN (0, 1)), + lifecycle_state TEXT NOT NULL CHECK(lifecycle_state IN ( + 'retryable', 'terminal', 'acknowledged', 'superseded', 'resolved' + )), + created_at_ms INTEGER NOT NULL, + updated_at_ms INTEGER NOT NULL, + resolved_at_ms INTEGER, + resolution_action TEXT, + resolution_note TEXT, + superseded_by TEXT +) STRICT; + +CREATE INDEX IF NOT EXISTS idx_embedding_failures_active +ON embedding_failures(lifecycle_state, updated_at_ms DESC, failure_id); + """ __all__ = ["EMBEDDING_DIMENSION", "EMBEDDINGS_DDL", "EMBEDDINGS_SCHEMA_VERSION"] diff --git a/tests/unit/storage/test_archive_tiers_embedding_write.py b/tests/unit/storage/test_archive_tiers_embedding_write.py index a78f5f08fc..2805307922 100644 --- a/tests/unit/storage/test_archive_tiers_embedding_write.py +++ b/tests/unit/storage/test_archive_tiers_embedding_write.py @@ -9,11 +9,15 @@ from polylogue.storage.embeddings.materialization import _record_archive_embedding_success from polylogue.storage.sqlite.archive_tiers.bootstrap import initialize_archive_tier from polylogue.storage.sqlite.archive_tiers.embedding_write import ( + ArchiveEmbeddingFailure, ArchiveEmbeddingMeta, ArchiveEmbeddingStatus, ArchiveEmbeddingWrite, + list_active_embedding_failures, mark_session_embedding_error, read_embedding_status, + record_embedding_failure, + resolve_embedding_failure, upsert_message_embedding, upsert_message_embeddings, ) @@ -145,3 +149,81 @@ def test_archive_tiers_embedding_writer_records_terminal_errors(tmp_path: Path) needs_reindex=False, error_message="Embedding generation failed: HTTP 400", ) + + +def test_embedding_failure_lifecycle_preserves_audit_and_requeues(tmp_path: Path) -> None: + """Terminal rows are inspectable, acknowledgeable, and requeueable. + + Anti-vacuity: replacing the ledger with aggregate-only status rows loses + message/provider/error identities; deleting on resolution loses the audit; + omitting the requeue mutation leaves the session permanently excluded. + """ + conn = _connect(tmp_path / "embeddings.db") + terminal = record_embedding_failure( + conn, + session_id="aistudio-drive:poisoned", + origin="aistudio-drive", + message_refs=("aistudio-drive:poisoned:m1",), + provider="voyage", + model="voyage-4", + error_class="provider_http_400", + error_message="Embedding generation failed: HTTP 400", + retryable=False, + occurred_at_ms=1_800_000_000_000, + ) + assert terminal == ArchiveEmbeddingFailure( + failure_id=terminal.failure_id, + session_id="aistudio-drive:poisoned", + origin="aistudio-drive", + message_refs=("aistudio-drive:poisoned:m1",), + provider="voyage", + model="voyage-4", + error_class="provider_http_400", + error_message="Embedding generation failed: HTTP 400", + retryable=False, + lifecycle_state="terminal", + created_at_ms=1_800_000_000_000, + updated_at_ms=1_800_000_000_000, + resolved_at_ms=None, + resolution_action=None, + resolution_note=None, + superseded_by=None, + ) + assert list_active_embedding_failures(conn) == (terminal,) + + acknowledged = resolve_embedding_failure( + conn, + failure_id=terminal.failure_id, + action="acknowledge", + note="provider rejects this historical payload", + resolved_at_ms=1_800_000_000_100, + ) + assert acknowledged.lifecycle_state == "acknowledged" + assert acknowledged.resolution_action == "acknowledge" + assert list_active_embedding_failures(conn) == () + assert conn.execute("SELECT COUNT(*) FROM embedding_failures").fetchone()[0] == 1 + assert read_embedding_status(conn, "aistudio-drive:poisoned").needs_reindex is False + + requeued = record_embedding_failure( + conn, + session_id="codex-session:retry-me", + origin=Origin.CODEX_SESSION, + message_refs=("codex-session:retry-me:m1",), + provider="voyage", + model="voyage-4", + error_class="provider_http_400", + error_message="Embedding generation failed: HTTP 400", + retryable=False, + occurred_at_ms=1_800_000_000_200, + ) + resolved = resolve_embedding_failure( + conn, + failure_id=requeued.failure_id, + action="requeue", + resolved_at_ms=1_800_000_000_300, + ) + assert resolved.lifecycle_state == "resolved" + assert resolved.resolution_action == "requeue" + status = read_embedding_status(conn, "codex-session:retry-me") + assert status.needs_reindex is True + assert status.error_message is None From c0afe01d28886036df49997fb481f80dafc58152 Mon Sep 17 00:00:00 2001 From: Sinity Date: Sun, 12 Jul 2026 23:31:04 +0200 Subject: [PATCH 03/16] fix(embeddings): register failure ledger writer paths Problem: the writer-ownership gate and strict types did not recognize\nthe new lifecycle mutation paths.\n\nWhat changed: inventory the ledger entrypoints and narrow the lifecycle\nrow conversions to their schema-backed types.\n\nVerification: Ruff format and lint pass; full quick gate will run on push.\n\nRef polylogue-egm8\n\nCo-Authored-By: Claude --- docs/plans/layering.yaml | 4 +++- .../sqlite/archive_tiers/embedding_write.py | 23 +++++++++++-------- 2 files changed, 16 insertions(+), 11 deletions(-) diff --git a/docs/plans/layering.yaml b/docs/plans/layering.yaml index 9fd0ea1721..8a7d4946bb 100644 --- a/docs/plans/layering.yaml +++ b/docs/plans/layering.yaml @@ -71,7 +71,9 @@ writer_modules: - tier: embeddings durability: rebuildable interruption: replayable - entrypoints: [mark_session_embedding_error, upsert_message_embedding, upsert_message_embeddings] + entrypoints: + [mark_session_embedding_error, record_embedding_failure, resolve_embedding_failure, + resolve_open_embedding_failures_for_session, upsert_message_embedding, upsert_message_embeddings] - path: polylogue/storage/sqlite/archive_tiers/user_write.py surfaces: - tier: user diff --git a/polylogue/storage/sqlite/archive_tiers/embedding_write.py b/polylogue/storage/sqlite/archive_tiers/embedding_write.py index 373b4b4bf5..521fc1e8f2 100644 --- a/polylogue/storage/sqlite/archive_tiers/embedding_write.py +++ b/polylogue/storage/sqlite/archive_tiers/embedding_write.py @@ -11,7 +11,7 @@ import uuid from collections.abc import Sequence from dataclasses import dataclass -from typing import Literal +from typing import Literal, cast from polylogue.core.enums import Origin from polylogue.storage.search_providers.sqlite_vec_support import _serialize_f32 @@ -243,11 +243,14 @@ def resolve_embedding_failure( """Explicitly acknowledge, supersede, or requeue an active failure.""" now_ms = int(time.time() * 1000) if resolved_at_ms is None else resolved_at_ms - state: EmbeddingFailureState = { - "acknowledge": "acknowledged", - "supersede": "superseded", - "requeue": "resolved", - }[action] + state = cast( + EmbeddingFailureState, + { + "acknowledge": "acknowledged", + "supersede": "superseded", + "requeue": "resolved", + }[action], + ) with conn: row = conn.execute( "SELECT session_id FROM embedding_failures WHERE failure_id = ? AND lifecycle_state IN ('retryable', 'terminal')", @@ -339,10 +342,10 @@ def _failure_from_row(row: sqlite3.Row | tuple[object, ...]) -> ArchiveEmbedding error_class=str(row[6]), error_message=str(row[7]), retryable=bool(row[8]), - lifecycle_state=str(row[9]), - created_at_ms=int(row[10]), - updated_at_ms=int(row[11]), - resolved_at_ms=None if row[12] is None else int(row[12]), + lifecycle_state=cast(EmbeddingFailureState, str(row[9])), + created_at_ms=int(str(row[10])), + updated_at_ms=int(str(row[11])), + resolved_at_ms=None if row[12] is None else int(str(row[12])), resolution_action=None if row[13] is None else str(row[13]), resolution_note=None if row[14] is None else str(row[14]), superseded_by=None if row[15] is None else str(row[15]), From d66ca438d1a0ee8077d4c2c4157c58957c9140f1 Mon Sep 17 00:00:00 2001 From: Sinity Date: Sun, 12 Jul 2026 23:34:35 +0200 Subject: [PATCH 04/16] feat(embeddings): expose actionable failure debt Problem: active terminal failures remained aggregate-only critical debt,\nwithout bounded identities or an explicit lifecycle action.\n\nWhat changed: expose active failure identities through the shared status\npayload (and therefore MCP), derive archive debt from lifecycle state, and\nadd a confirmed CLI resolution command.\n\nVerification: Ruff format/lint and strict mypy pass for the changed\nmodules. Focused embedding tests pending the next safe resource window.\n\nRef polylogue-egm8\n\nCo-Authored-By: Claude --- polylogue/cli/commands/embed.py | 58 +++++++++++- polylogue/daemon/embedding_readiness.py | 2 + polylogue/operations/archive_debt.py | 17 +++- .../storage/embeddings/status_payload.py | 89 ++++++++++++++++++- 4 files changed, 160 insertions(+), 6 deletions(-) diff --git a/polylogue/cli/commands/embed.py b/polylogue/cli/commands/embed.py index c61d3bb335..b480f27175 100644 --- a/polylogue/cli/commands/embed.py +++ b/polylogue/cli/commands/embed.py @@ -22,7 +22,7 @@ import sqlite3 import time from pathlib import Path -from typing import TypedDict, cast +from typing import Literal, TypedDict, cast import click @@ -267,6 +267,62 @@ def embed_command() -> None: """Manage the embedding pipeline (activation, preflight, backfill).""" +@embed_command.command("resolve-failure") +@click.argument("failure_id") +@click.option("--action", "resolution", type=click.Choice(["acknowledge", "requeue", "supersede"]), required=True) +@click.option("--note", default=None, help="Durable operator rationale for the resolution.") +@click.option("--superseded-by", default=None, help="Replacement failure or remediation reference for supersession.") +@click.option("--yes", is_flag=True, help="Confirm this lifecycle mutation.") +@click.option("--format", "output_format", type=click.Choice(["text", "json"]), default="text") +@click.pass_obj +def resolve_failure_subcommand( + env: AppEnv, + failure_id: str, + resolution: str, + note: str | None, + superseded_by: str | None, + yes: bool, + output_format: str, +) -> None: + """Acknowledge, supersede, or requeue one active embedding failure.""" + + if not yes and not click.confirm(f"Apply {resolution} to embedding failure {failure_id}?", default=False): + raise click.Abort() + index_db = _active_archive_index_path(env.config.db_path) + if index_db is None: + raise click.ClickException("index.db not found") + embeddings_db = index_db.with_name("embeddings.db") + if not embeddings_db.exists(): + raise click.ClickException("embeddings.db not found") + from polylogue.storage.sqlite.archive_tiers.embedding_write import resolve_embedding_failure + + try: + with sqlite3.connect(embeddings_db, timeout=30.0) as conn: + failure = resolve_embedding_failure( + conn, + failure_id=failure_id, + action=cast(Literal["acknowledge", "requeue", "supersede"], resolution), + note=note, + superseded_by=superseded_by, + ) + except KeyError as exc: + raise click.ClickException(f"active embedding failure not found: {failure_id}") from exc + payload = { + "failure_id": failure.failure_id, + "session_id": failure.session_id, + "lifecycle_state": failure.lifecycle_state, + "resolution_action": failure.resolution_action, + "resolution_note": failure.resolution_note, + "superseded_by": failure.superseded_by, + } + if output_format == "json": + click.echo(json.dumps(payload, sort_keys=True)) + else: + click.echo( + f"Resolved embedding failure {failure.failure_id}: {failure.lifecycle_state} ({failure.resolution_action})" + ) + + def _check_sqlite_vec_available() -> tuple[bool, str | None]: import importlib.util diff --git a/polylogue/daemon/embedding_readiness.py b/polylogue/daemon/embedding_readiness.py index 24ed7136ac..21e47c2066 100644 --- a/polylogue/daemon/embedding_readiness.py +++ b/polylogue/daemon/embedding_readiness.py @@ -29,6 +29,7 @@ def _defaults(*, enabled: bool, config_enabled: bool, has_key: bool, model: str, "embedding_stale_count": 0, "embedding_coverage_percent": 0.0, "embedding_failure_count": 0, + "embedding_failure_details": [], "embedding_estimated_cost_usd": 0.0, "embedding_latest_catchup_run": None, "embedding_latest_material_catchup_run": None, @@ -88,6 +89,7 @@ def embedding_readiness_info(db_file: Path, *, detail: bool = False) -> dict[str "embedding_stale_count": payload["stale_messages"], "embedding_coverage_percent": payload["embedding_coverage_percent"], "embedding_failure_count": payload["failure_count"], + "embedding_failure_details": payload["failure_details"], "embedding_estimated_cost_usd": payload["total_estimated_cost_usd"], "embedding_latest_catchup_run": payload["latest_catchup_run"], "embedding_latest_material_catchup_run": payload["latest_material_catchup_run"], diff --git a/polylogue/operations/archive_debt.py b/polylogue/operations/archive_debt.py index 247b1ae264..b48d955b0c 100644 --- a/polylogue/operations/archive_debt.py +++ b/polylogue/operations/archive_debt.py @@ -945,6 +945,7 @@ def _embedding_rows(index_db: Path) -> list[ArchiveDebtRowPayload]: pending_messages_exact = _bool_value(info.get("embedding_pending_message_count_exact")) stale = _int_value(info.get("embedding_stale_count")) or 0 failures = _int_value(info.get("embedding_failure_count")) or 0 + failure_details = info.get("embedding_failure_details") if config_enabled and not has_key: rows.append( @@ -966,6 +967,13 @@ def _embedding_rows(index_db: Path) -> list[ArchiveDebtRowPayload]: ) ) if failures: + terminal_failures = 0 + if isinstance(failure_details, list): + terminal_failures = sum( + 1 + for detail in failure_details + if isinstance(detail, dict) and detail.get("lifecycle_state") == "terminal" + ) rows.append( ArchiveDebtRowPayload( debt_ref="debt:embedding:catchup:failures", @@ -975,14 +983,19 @@ def _embedding_rows(index_db: Path) -> list[ArchiveDebtRowPayload]: severity="critical", status="actionable" if enabled else "blocked", owner="daemon", - summary=f"{failures} embedding catch-up failure(s) recorded", + summary=f"{failures} active embedding failure(s) recorded", evidence_refs=(f"archive-tier:{index_db.with_name('embeddings.db')}",), actions=( ArchiveDebtActionPayload( - label="Inspect embedding status", + label="Inspect active embedding failures", command=("polylogue", "ops", "embed", "status", "--detail"), ), ), + caveats=( + (f"{terminal_failures} terminal failure(s) require explicit acknowledge, supersede, or requeue.",) + if terminal_failures + else ("Retryable failures remain eligible for automatic catch-up.",) + ), ) ) if pending or stale: diff --git a/polylogue/storage/embeddings/status_payload.py b/polylogue/storage/embeddings/status_payload.py index 7d0a09df6d..95fa29a34b 100644 --- a/polylogue/storage/embeddings/status_payload.py +++ b/polylogue/storage/embeddings/status_payload.py @@ -7,6 +7,7 @@ from __future__ import annotations +import json import sqlite3 import time from datetime import UTC, datetime @@ -36,6 +37,7 @@ DETAIL_CANDIDATE_PROSE_TIMEOUT_MS = 10_000 METADATA_SUMMARY_TIMEOUT_MS = 5_000 STATUS_READ_BUSY_TIMEOUT_MS = 1_000 +EMBEDDING_FAILURE_DETAIL_LIMIT = 25 class _HasConfig(Protocol): @@ -93,6 +95,22 @@ class EmbeddingNextActionPayload(TypedDict): reason: str +class EmbeddingFailureDetailPayload(TypedDict): + failure_id: str + session_id: str + origin: str + message_refs: list[str] + provider: str + model: str + error_class: str + error_message: str + retryable: bool + lifecycle_state: str + created_at: str | None + updated_at: str | None + resolution_action: str | None + + class EmbeddingStatusPayload(TypedDict): config_enabled: bool has_voyage_api_key: bool @@ -122,6 +140,7 @@ class EmbeddingStatusPayload(TypedDict): embedding_dimensions: dict[int, int] retrieval_bands: dict[str, dict[str, object]] failure_count: int + failure_details: list[EmbeddingFailureDetailPayload] total_estimated_cost_usd: float | None latest_catchup_run: EmbeddingCatchupRunPayload | None latest_material_catchup_run: EmbeddingCatchupRunPayload | None @@ -248,6 +267,57 @@ def _interrupt_when_expired() -> int: return list(rows) +def _active_failure_details( + conn: sqlite3.Connection, + failure_table: str, + *, + include_detail: bool, +) -> list[EmbeddingFailureDetailPayload]: + """Return bounded active lifecycle rows, never historical acknowledgements.""" + + if not include_detail or not failure_table: + return [] + rows = _rows_with_timeout( + conn, + f""" + SELECT failure_id, session_id, origin, message_refs_json, provider, model, error_class, error_message, + retryable, lifecycle_state, created_at_ms, updated_at_ms, resolution_action + FROM {failure_table} + WHERE lifecycle_state IN ('retryable', 'terminal') + ORDER BY updated_at_ms DESC, failure_id ASC + LIMIT ? + """, + timeout_ms=DETAIL_QUERY_TIMEOUT_MS, + params=(EMBEDDING_FAILURE_DETAIL_LIMIT,), + ) + if rows is None: + return [] + details: list[EmbeddingFailureDetailPayload] = [] + for row in rows: + try: + message_refs = [str(item) for item in json.loads(str(row[3]))] + except (TypeError, ValueError, json.JSONDecodeError): + message_refs = [] + details.append( + { + "failure_id": str(row[0]), + "session_id": str(row[1]), + "origin": str(row[2]), + "message_refs": message_refs, + "provider": str(row[4]), + "model": str(row[5]), + "error_class": str(row[6]), + "error_message": str(row[7]), + "retryable": bool(row[8]), + "lifecycle_state": str(row[9]), + "created_at": _iso_from_epoch_ms(row[10]), + "updated_at": _iso_from_epoch_ms(row[11]), + "resolution_action": None if row[12] is None else str(row[12]), + } + ) + return details + + def _uniform_embedding_metadata_counts( conn: sqlite3.Connection, meta_table: str, @@ -551,6 +621,7 @@ def _payload_from_stats( latest_catchup_run: EmbeddingCatchupRunPayload | None, latest_material_catchup_run: EmbeddingCatchupRunPayload | None, pending_messages_exact: bool, + failure_details: list[EmbeddingFailureDetailPayload] | None = None, ) -> EmbeddingStatusPayload: embedded_sessions = stats.embedded_sessions pending_sessions = stats.pending_sessions @@ -604,6 +675,7 @@ def _payload_from_stats( "embedding_dimensions": stats.dimension_counts, "retrieval_bands": stats.retrieval_bands, "failure_count": stats.failure_count, + "failure_details": failure_details or [], "total_estimated_cost_usd": stats.total_estimated_cost_usd, "latest_catchup_run": latest_catchup_run, "latest_material_catchup_run": latest_material_catchup_run, @@ -640,10 +712,12 @@ def _archive_embedding_status_payload( status_table = _attached_table_name(conn, "embeddings", "embedding_status") vector_table = _attached_table_name(conn, "embeddings", "message_embeddings") meta_table = _attached_table_name(conn, "embeddings", "message_embeddings_meta") + failure_table = _attached_table_name(conn, "embeddings", "embedding_failures") else: status_table = "" vector_table = "" meta_table = "" + failure_table = "" has_messages = _table_exists(conn, "messages") has_status = bool(status_table) has_meta = bool(meta_table) @@ -688,16 +762,24 @@ def _archive_embedding_status_payload( failure_count = ( _scalar_int( conn, - f""" + f"SELECT COUNT(*) FROM {failure_table} WHERE lifecycle_state IN ('retryable', 'terminal')", + ) + if failure_table + else ( + _scalar_int( + conn, + f""" SELECT COUNT(*) FROM {status_table} AS e JOIN sessions AS s ON s.session_id = e.session_id WHERE e.error_message IS NOT NULL """, + ) + if has_status + else 0 ) - if has_status - else 0 ) + failure_details = _active_failure_details(conn, failure_table, include_detail=include_detail) pending_messages = 0 candidate_prose_messages: int | None = None candidate_prose_messages_exact = False @@ -869,6 +951,7 @@ def _archive_embedding_status_payload( latest_catchup_run=latest_catchup_run, latest_material_catchup_run=latest_material_catchup_run, pending_messages_exact=pending_messages_exact, + failure_details=failure_details, ) From 41dac5ba23b8c1699101f1341cd5bd81148238c0 Mon Sep 17 00:00:00 2001 From: Sinity Date: Sun, 12 Jul 2026 23:51:11 +0200 Subject: [PATCH 05/16] fix(embeddings): surface terminal failure actions Problem: archive debt discarded bounded failure detail and status rows\ndid not name the supported lifecycle command.\n\nWhat changed: make debt request detailed readiness, add explicit supported\nactions and a resolution command to each active failure payload, and cover\nthe real CLI status route.\n\nVerification: devtools test tests/unit/cli/test_embed_status_fast.py -k\nterminal_failure_resolution (1 passed).\n\nRef polylogue-egm8\n\nCo-Authored-By: Claude --- polylogue/operations/archive_debt.py | 2 +- .../storage/embeddings/status_payload.py | 6 ++ tests/unit/cli/test_embed_status_fast.py | 66 +++++++++++++++++++ 3 files changed, 73 insertions(+), 1 deletion(-) diff --git a/polylogue/operations/archive_debt.py b/polylogue/operations/archive_debt.py index b48d955b0c..27d8df4473 100644 --- a/polylogue/operations/archive_debt.py +++ b/polylogue/operations/archive_debt.py @@ -935,7 +935,7 @@ def _table_exists(conn: sqlite3.Connection, table: str) -> bool: def _embedding_rows(index_db: Path) -> list[ArchiveDebtRowPayload]: - info = embedding_readiness_info(index_db, detail=False) + info = embedding_readiness_info(index_db, detail=True) rows: list[ArchiveDebtRowPayload] = [] config_enabled = _bool_value(info.get("embedding_config_enabled")) has_key = _bool_value(info.get("embedding_has_voyage_key")) diff --git a/polylogue/storage/embeddings/status_payload.py b/polylogue/storage/embeddings/status_payload.py index 95fa29a34b..856128acc9 100644 --- a/polylogue/storage/embeddings/status_payload.py +++ b/polylogue/storage/embeddings/status_payload.py @@ -109,6 +109,8 @@ class EmbeddingFailureDetailPayload(TypedDict): created_at: str | None updated_at: str | None resolution_action: str | None + supported_actions: list[str] + resolution_command: str class EmbeddingStatusPayload(TypedDict): @@ -313,6 +315,10 @@ def _active_failure_details( "created_at": _iso_from_epoch_ms(row[10]), "updated_at": _iso_from_epoch_ms(row[11]), "resolution_action": None if row[12] is None else str(row[12]), + "supported_actions": ["acknowledge", "requeue", "supersede"], + "resolution_command": ( + f"polylogue ops embed resolve-failure {str(row[0])} --action acknowledge|requeue|supersede --yes" + ), } ) return details diff --git a/tests/unit/cli/test_embed_status_fast.py b/tests/unit/cli/test_embed_status_fast.py index d1499881c7..ca94a521ef 100644 --- a/tests/unit/cli/test_embed_status_fast.py +++ b/tests/unit/cli/test_embed_status_fast.py @@ -214,6 +214,72 @@ def test_status_json_reports_archive_embedding_metadata_without_detail(tmp_path: assert payload["newest_embedded_at"] is None +def test_status_detail_exposes_bounded_terminal_failure_resolution(tmp_path: Path) -> None: + """Status must name an actionable terminal row rather than only its aggregate.""" + db_anchor = tmp_path / "custom.sqlite" + index_db = tmp_path / "index.db" + _seed_archive_file_set_from_archive_tiers(index_db) + with sqlite3.connect(index_db.with_name("embeddings.db")) as conn: + conn.execute( + """ + CREATE TABLE embedding_failures ( + failure_id TEXT PRIMARY KEY, + session_id TEXT NOT NULL, + origin TEXT NOT NULL, + message_refs_json TEXT NOT NULL, + provider TEXT NOT NULL, + model TEXT NOT NULL, + error_class TEXT NOT NULL, + error_message TEXT NOT NULL, + retryable INTEGER NOT NULL, + lifecycle_state TEXT NOT NULL, + created_at_ms INTEGER NOT NULL, + updated_at_ms INTEGER NOT NULL, + resolved_at_ms INTEGER, + resolution_action TEXT, + resolution_note TEXT, + superseded_by TEXT + ) + """ + ) + conn.execute( + """ + INSERT INTO embedding_failures VALUES ( + 'embedding-failure:terminal', 'codex-session:pending', 'codex-session', + '[\"codex-session:pending:m1\"]', 'voyage', 'voyage-4', 'provider_http_400', + 'Embedding generation failed: HTTP 400', 0, 'terminal', 1800000000000, 1800000000000, + NULL, NULL, NULL, NULL + ) + """ + ) + + payload = _run_status(db_anchor, "--detail", cfg=_Cfg(embedding_enabled=True, voyage_api_key="vk-live")) + + assert payload["failure_count"] == 1 + assert payload["failure_details"] == [ + { + "failure_id": "embedding-failure:terminal", + "session_id": "codex-session:pending", + "origin": "codex-session", + "message_refs": ["codex-session:pending:m1"], + "provider": "voyage", + "model": "voyage-4", + "error_class": "provider_http_400", + "error_message": "Embedding generation failed: HTTP 400", + "retryable": False, + "lifecycle_state": "terminal", + "created_at": "2027-01-15T08:00:00+00:00", + "updated_at": "2027-01-15T08:00:00+00:00", + "resolution_action": None, + "supported_actions": ["acknowledge", "requeue", "supersede"], + "resolution_command": ( + "polylogue ops embed resolve-failure embedding-failure:terminal " + "--action acknowledge|requeue|supersede --yes" + ), + } + ] + + def test_status_json_detail_does_not_derive_coverage_from_analyzed_prose_estimate(tmp_path: Path) -> None: db_anchor = tmp_path / "custom.sqlite" index_db = tmp_path / "index.db" From 28b3d6e2873fdc19824ae4e199b900fdd96f9ad3 Mon Sep 17 00:00:00 2001 From: Sinity Date: Sun, 12 Jul 2026 23:54:52 +0200 Subject: [PATCH 06/16] fix(embeddings): count terminal debt beyond detail window Problem: archive debt inferred terminal state from the bounded detail sample,\nwhich can misclassify a larger active failure backlog.\n\nWhat changed: publish exact active terminal and retryable lifecycle counts\nthrough readiness and consume the terminal aggregate in archive debt.\n\nVerification: Ruff, strict mypy, and terminal_failure_resolution focused\nCLI test pass.\n\nRef polylogue-egm8\n\nCo-Authored-By: Claude --- polylogue/daemon/embedding_readiness.py | 4 +++ polylogue/operations/archive_debt.py | 9 +----- .../storage/embeddings/status_payload.py | 32 +++++++++++++++++++ tests/unit/cli/test_embed_status_fast.py | 2 ++ 4 files changed, 39 insertions(+), 8 deletions(-) diff --git a/polylogue/daemon/embedding_readiness.py b/polylogue/daemon/embedding_readiness.py index 21e47c2066..0233d69e9c 100644 --- a/polylogue/daemon/embedding_readiness.py +++ b/polylogue/daemon/embedding_readiness.py @@ -29,6 +29,8 @@ def _defaults(*, enabled: bool, config_enabled: bool, has_key: bool, model: str, "embedding_stale_count": 0, "embedding_coverage_percent": 0.0, "embedding_failure_count": 0, + "embedding_terminal_failure_count": 0, + "embedding_retryable_failure_count": 0, "embedding_failure_details": [], "embedding_estimated_cost_usd": 0.0, "embedding_latest_catchup_run": None, @@ -89,6 +91,8 @@ def embedding_readiness_info(db_file: Path, *, detail: bool = False) -> dict[str "embedding_stale_count": payload["stale_messages"], "embedding_coverage_percent": payload["embedding_coverage_percent"], "embedding_failure_count": payload["failure_count"], + "embedding_terminal_failure_count": payload["terminal_failure_count"], + "embedding_retryable_failure_count": payload["retryable_failure_count"], "embedding_failure_details": payload["failure_details"], "embedding_estimated_cost_usd": payload["total_estimated_cost_usd"], "embedding_latest_catchup_run": payload["latest_catchup_run"], diff --git a/polylogue/operations/archive_debt.py b/polylogue/operations/archive_debt.py index 27d8df4473..47dd0ecd3a 100644 --- a/polylogue/operations/archive_debt.py +++ b/polylogue/operations/archive_debt.py @@ -945,7 +945,7 @@ def _embedding_rows(index_db: Path) -> list[ArchiveDebtRowPayload]: pending_messages_exact = _bool_value(info.get("embedding_pending_message_count_exact")) stale = _int_value(info.get("embedding_stale_count")) or 0 failures = _int_value(info.get("embedding_failure_count")) or 0 - failure_details = info.get("embedding_failure_details") + terminal_failures = _int_value(info.get("embedding_terminal_failure_count")) or 0 if config_enabled and not has_key: rows.append( @@ -967,13 +967,6 @@ def _embedding_rows(index_db: Path) -> list[ArchiveDebtRowPayload]: ) ) if failures: - terminal_failures = 0 - if isinstance(failure_details, list): - terminal_failures = sum( - 1 - for detail in failure_details - if isinstance(detail, dict) and detail.get("lifecycle_state") == "terminal" - ) rows.append( ArchiveDebtRowPayload( debt_ref="debt:embedding:catchup:failures", diff --git a/polylogue/storage/embeddings/status_payload.py b/polylogue/storage/embeddings/status_payload.py index 856128acc9..99eb4c32dd 100644 --- a/polylogue/storage/embeddings/status_payload.py +++ b/polylogue/storage/embeddings/status_payload.py @@ -142,6 +142,8 @@ class EmbeddingStatusPayload(TypedDict): embedding_dimensions: dict[int, int] retrieval_bands: dict[str, dict[str, object]] failure_count: int + terminal_failure_count: int + retryable_failure_count: int failure_details: list[EmbeddingFailureDetailPayload] total_estimated_cost_usd: float | None latest_catchup_run: EmbeddingCatchupRunPayload | None @@ -628,6 +630,8 @@ def _payload_from_stats( latest_material_catchup_run: EmbeddingCatchupRunPayload | None, pending_messages_exact: bool, failure_details: list[EmbeddingFailureDetailPayload] | None = None, + terminal_failure_count: int = 0, + retryable_failure_count: int = 0, ) -> EmbeddingStatusPayload: embedded_sessions = stats.embedded_sessions pending_sessions = stats.pending_sessions @@ -681,6 +685,8 @@ def _payload_from_stats( "embedding_dimensions": stats.dimension_counts, "retrieval_bands": stats.retrieval_bands, "failure_count": stats.failure_count, + "terminal_failure_count": terminal_failure_count, + "retryable_failure_count": retryable_failure_count, "failure_details": failure_details or [], "total_estimated_cost_usd": stats.total_estimated_cost_usd, "latest_catchup_run": latest_catchup_run, @@ -785,6 +791,30 @@ def _archive_embedding_status_payload( else 0 ) ) + terminal_failure_count = ( + _scalar_int(conn, f"SELECT COUNT(*) FROM {failure_table} WHERE lifecycle_state = 'terminal'") + if failure_table + else ( + _scalar_int( + conn, + f"SELECT COUNT(*) FROM {status_table} WHERE error_message IS NOT NULL AND needs_reindex = 0", + ) + if has_status + else 0 + ) + ) + retryable_failure_count = ( + _scalar_int(conn, f"SELECT COUNT(*) FROM {failure_table} WHERE lifecycle_state = 'retryable'") + if failure_table + else ( + _scalar_int( + conn, + f"SELECT COUNT(*) FROM {status_table} WHERE error_message IS NOT NULL AND needs_reindex = 1", + ) + if has_status + else 0 + ) + ) failure_details = _active_failure_details(conn, failure_table, include_detail=include_detail) pending_messages = 0 candidate_prose_messages: int | None = None @@ -958,6 +988,8 @@ def _archive_embedding_status_payload( latest_material_catchup_run=latest_material_catchup_run, pending_messages_exact=pending_messages_exact, failure_details=failure_details, + terminal_failure_count=terminal_failure_count, + retryable_failure_count=retryable_failure_count, ) diff --git a/tests/unit/cli/test_embed_status_fast.py b/tests/unit/cli/test_embed_status_fast.py index ca94a521ef..85316d07f9 100644 --- a/tests/unit/cli/test_embed_status_fast.py +++ b/tests/unit/cli/test_embed_status_fast.py @@ -256,6 +256,8 @@ def test_status_detail_exposes_bounded_terminal_failure_resolution(tmp_path: Pat payload = _run_status(db_anchor, "--detail", cfg=_Cfg(embedding_enabled=True, voyage_api_key="vk-live")) assert payload["failure_count"] == 1 + assert payload["terminal_failure_count"] == 1 + assert payload["retryable_failure_count"] == 0 assert payload["failure_details"] == [ { "failure_id": "embedding-failure:terminal", From abdc734e3e99e8bcf14c4781285e92e7b2756cb7 Mon Sep 17 00:00:00 2001 From: Sinity Date: Sun, 12 Jul 2026 23:57:37 +0200 Subject: [PATCH 07/16] fix(embeddings): render actionable failure identities Problem: detail JSON carried failure ids and commands, but the default\nhuman-facing status renderer omitted them.\n\nWhat changed: render bounded active failure identities, refs, classifications,\nerrors, and resolution commands in text status output.\n\nVerification: Ruff, mypy, and terminal_failure_resolution focused CLI test\npass.\n\nRef polylogue-egm8\n\nCo-Authored-By: Claude --- polylogue/cli/shared/embed_stats.py | 20 ++++++++++++++++++++ tests/unit/cli/test_embed_status_fast.py | 19 ++++++++++++++++--- 2 files changed, 36 insertions(+), 3 deletions(-) diff --git a/polylogue/cli/shared/embed_stats.py b/polylogue/cli/shared/embed_stats.py index f3575e76a2..afa96eb8b2 100644 --- a/polylogue/cli/shared/embed_stats.py +++ b/polylogue/cli/shared/embed_stats.py @@ -92,6 +92,25 @@ def _render_next_actions(payload: EmbeddingStatusPayload) -> None: _render_field("Command", command) +def _render_failure_details(payload: EmbeddingStatusPayload) -> None: + """Render bounded actionable identities for the human status surface.""" + + details = payload["failure_details"] + if not details: + return + click.echo(" Active embedding failures:") + for detail in details: + refs = ", ".join(detail["message_refs"]) or "session-only" + click.echo( + f" {detail['failure_id']}: {detail['lifecycle_state']}; " + f"{detail['origin']} / {detail['session_id']}; {detail['provider']} {detail['model']}; " + f"{detail['error_class']}" + ) + click.echo(f" refs: {refs}") + click.echo(f" error: {detail['error_message']}") + click.echo(f" resolve: {detail['resolution_command']}") + + def render_embedding_stats(payload: EmbeddingStatusPayload, *, json_output: bool = False) -> None: """Render an embedding statistics payload.""" if json_output: @@ -133,6 +152,7 @@ def render_embedding_stats(payload: EmbeddingStatusPayload, *, json_output: bool else: _render_field("Estimated total cost", f"~${payload['total_estimated_cost_usd']:.2f}") _render_next_actions(payload) + _render_failure_details(payload) _render_embedding_window(payload) _render_named_counts("Models", payload["embedding_models"]) _render_named_counts("Dimensions", payload["embedding_dimensions"]) diff --git a/tests/unit/cli/test_embed_status_fast.py b/tests/unit/cli/test_embed_status_fast.py index 85316d07f9..5c13206d7a 100644 --- a/tests/unit/cli/test_embed_status_fast.py +++ b/tests/unit/cli/test_embed_status_fast.py @@ -149,13 +149,18 @@ def _run_status(db_path: Path, *args: str, cfg: _Cfg | None = None) -> dict[str, return _payload(result.output) -def _run_status_text(db_path: Path, *, cfg: _Cfg | None = None) -> str: +def _run_status_text(db_path: Path, *, detail: bool = False, cfg: _Cfg | None = None) -> str: runner = CliRunner(env={"POLYLOGUE_FORCE_PLAIN": "1"}) with patch( "polylogue.config.load_polylogue_config", return_value=cfg or _Cfg(embedding_enabled=False, voyage_api_key=None), ): - result = runner.invoke(embed_command, ["status"], obj=_env(db_path), catch_exceptions=False) + result = runner.invoke( + embed_command, + ["status", *(["--detail"] if detail else [])], + obj=_env(db_path), + catch_exceptions=False, + ) assert result.exit_code == 0 return str(result.output) @@ -215,7 +220,11 @@ def test_status_json_reports_archive_embedding_metadata_without_detail(tmp_path: def test_status_detail_exposes_bounded_terminal_failure_resolution(tmp_path: Path) -> None: - """Status must name an actionable terminal row rather than only its aggregate.""" + """Status must name an actionable terminal row rather than only its aggregate. + + Anti-vacuity: omitting the text renderer leaves the exact failure id and + resolution command unreachable to the operator despite JSON detail. + """ db_anchor = tmp_path / "custom.sqlite" index_db = tmp_path / "index.db" _seed_archive_file_set_from_archive_tiers(index_db) @@ -280,6 +289,10 @@ def test_status_detail_exposes_bounded_terminal_failure_resolution(tmp_path: Pat ), } ] + text = _run_status_text(db_anchor, detail=True, cfg=_Cfg(embedding_enabled=True, voyage_api_key="vk-live")) + assert "embedding-failure:terminal: terminal" in text + assert "refs: codex-session:pending:m1" in text + assert "resolve: polylogue ops embed resolve-failure embedding-failure:terminal" in text def test_status_json_detail_does_not_derive_coverage_from_analyzed_prose_estimate(tmp_path: Path) -> None: From 41a9930389276d69162527515a83fd68706665fe Mon Sep 17 00:00:00 2001 From: Sinity Date: Mon, 13 Jul 2026 00:02:57 +0200 Subject: [PATCH 08/16] fix(embeddings): make failure commands executable Problem: status rendered shell-invalid action alternatives and the operator resolution command had no end-to-end coverage. What changed: render a quoted failure identity with an ACTION placeholder, and exercise the Click requeue path against the archive index/embeddings pair. Ref polylogue-egm8 --- .../storage/embeddings/status_payload.py | 3 +- tests/unit/cli/test_embed_status_fast.py | 81 ++++++++++++++++++- 2 files changed, 81 insertions(+), 3 deletions(-) diff --git a/polylogue/storage/embeddings/status_payload.py b/polylogue/storage/embeddings/status_payload.py index 99eb4c32dd..9d37823498 100644 --- a/polylogue/storage/embeddings/status_payload.py +++ b/polylogue/storage/embeddings/status_payload.py @@ -8,6 +8,7 @@ from __future__ import annotations import json +import shlex import sqlite3 import time from datetime import UTC, datetime @@ -319,7 +320,7 @@ def _active_failure_details( "resolution_action": None if row[12] is None else str(row[12]), "supported_actions": ["acknowledge", "requeue", "supersede"], "resolution_command": ( - f"polylogue ops embed resolve-failure {str(row[0])} --action acknowledge|requeue|supersede --yes" + f"polylogue ops embed resolve-failure {shlex.quote(str(row[0]))} --action ACTION --yes" ), } ) diff --git a/tests/unit/cli/test_embed_status_fast.py b/tests/unit/cli/test_embed_status_fast.py index 5c13206d7a..c30f599b8c 100644 --- a/tests/unit/cli/test_embed_status_fast.py +++ b/tests/unit/cli/test_embed_status_fast.py @@ -284,8 +284,7 @@ def test_status_detail_exposes_bounded_terminal_failure_resolution(tmp_path: Pat "resolution_action": None, "supported_actions": ["acknowledge", "requeue", "supersede"], "resolution_command": ( - "polylogue ops embed resolve-failure embedding-failure:terminal " - "--action acknowledge|requeue|supersede --yes" + "polylogue ops embed resolve-failure embedding-failure:terminal --action ACTION --yes" ), } ] @@ -295,6 +294,84 @@ def test_status_detail_exposes_bounded_terminal_failure_resolution(tmp_path: Pat assert "resolve: polylogue ops embed resolve-failure embedding-failure:terminal" in text +def test_resolve_failure_cli_requeues_terminal_failure(tmp_path: Path) -> None: + """The operator command must mutate the failure ledger and retry status. + + Anti-vacuity: replacing the Click command with output-only formatting, or + removing its call to ``resolve_embedding_failure``, leaves the terminal row + active and the session excluded from a future embedding pass. + """ + db_anchor = tmp_path / "custom.sqlite" + index_db = tmp_path / "index.db" + _seed_archive_file_set_from_archive_tiers(index_db) + embeddings_db = index_db.with_name("embeddings.db") + with sqlite3.connect(embeddings_db) as conn: + conn.executescript( + """ + CREATE TABLE embedding_failures ( + failure_id TEXT PRIMARY KEY, + session_id TEXT NOT NULL, + origin TEXT NOT NULL, + message_refs_json TEXT NOT NULL, + provider TEXT NOT NULL, + model TEXT NOT NULL, + error_class TEXT NOT NULL, + error_message TEXT NOT NULL, + retryable INTEGER NOT NULL, + lifecycle_state TEXT NOT NULL, + created_at_ms INTEGER NOT NULL, + updated_at_ms INTEGER NOT NULL, + resolved_at_ms INTEGER, + resolution_action TEXT, + resolution_note TEXT, + superseded_by TEXT + ); + INSERT INTO embedding_status VALUES ( + 'codex-session:pending', 'codex-session', 0, 0, 'Embedding generation failed: HTTP 400' + ); + INSERT INTO embedding_failures VALUES ( + 'embedding-failure:terminal', 'codex-session:pending', 'codex-session', + '["codex-session:pending:m1"]', 'voyage', 'voyage-4', 'provider_http_400', + 'Embedding generation failed: HTTP 400', 0, 'terminal', 1800000000000, 1800000000000, + NULL, NULL, NULL, NULL + ); + """ + ) + + runner = CliRunner(env={"POLYLOGUE_FORCE_PLAIN": "1"}) + result = runner.invoke( + embed_command, + [ + "resolve-failure", + "embedding-failure:terminal", + "--action", + "requeue", + "--yes", + "--format", + "json", + ], + obj=_env(db_anchor), + catch_exceptions=False, + ) + + assert result.exit_code == 0 + assert _payload(result.output) == { + "failure_id": "embedding-failure:terminal", + "lifecycle_state": "resolved", + "resolution_action": "requeue", + "resolution_note": None, + "session_id": "codex-session:pending", + "superseded_by": None, + } + with sqlite3.connect(embeddings_db) as conn: + assert conn.execute( + "SELECT lifecycle_state FROM embedding_failures WHERE failure_id = 'embedding-failure:terminal'" + ).fetchone() == ("resolved",) + assert conn.execute( + "SELECT needs_reindex, error_message FROM embedding_status WHERE session_id = 'codex-session:pending'" + ).fetchone() == (1, None) + + def test_status_json_detail_does_not_derive_coverage_from_analyzed_prose_estimate(tmp_path: Path) -> None: db_anchor = tmp_path / "custom.sqlite" index_db = tmp_path / "index.db" From e757c207a3fa5a211f9613b7d94b6a24cbd274ac Mon Sep 17 00:00:00 2001 From: Sinity Date: Mon, 13 Jul 2026 00:05:05 +0200 Subject: [PATCH 09/16] fix(embeddings): supersede stale failure attempts Problem: every repeated embedding failure remained current debt, inflating active-failure counts for one session despite a newer failure event. What changed: supersede prior active rows for the session when recording the new failure, retaining their identities and linking them to the current event. Ref polylogue-egm8 --- .../sqlite/archive_tiers/embedding_write.py | 9 ++++ .../test_archive_tiers_embedding_write.py | 44 +++++++++++++++++++ 2 files changed, 53 insertions(+) diff --git a/polylogue/storage/sqlite/archive_tiers/embedding_write.py b/polylogue/storage/sqlite/archive_tiers/embedding_write.py index 521fc1e8f2..bac0140818 100644 --- a/polylogue/storage/sqlite/archive_tiers/embedding_write.py +++ b/polylogue/storage/sqlite/archive_tiers/embedding_write.py @@ -206,6 +206,15 @@ def record_embedding_failure( """, (session_id, origin_value, 1 if retryable else 0, error_message), ) + conn.execute( + """ + UPDATE embedding_failures + SET lifecycle_state = 'superseded', updated_at_ms = ?, resolved_at_ms = ?, + resolution_action = 'superseded', superseded_by = ? + WHERE session_id = ? AND lifecycle_state IN ('retryable', 'terminal') + """, + (now_ms, now_ms, failure_id, session_id), + ) conn.execute( """ INSERT INTO embedding_failures ( diff --git a/tests/unit/storage/test_archive_tiers_embedding_write.py b/tests/unit/storage/test_archive_tiers_embedding_write.py index 2805307922..38014b44ff 100644 --- a/tests/unit/storage/test_archive_tiers_embedding_write.py +++ b/tests/unit/storage/test_archive_tiers_embedding_write.py @@ -15,6 +15,7 @@ ArchiveEmbeddingWrite, list_active_embedding_failures, mark_session_embedding_error, + read_embedding_failure, read_embedding_status, record_embedding_failure, resolve_embedding_failure, @@ -227,3 +228,46 @@ def test_embedding_failure_lifecycle_preserves_audit_and_requeues(tmp_path: Path status = read_embedding_status(conn, "codex-session:retry-me") assert status.needs_reindex is True assert status.error_message is None + + +def test_new_failure_supersedes_prior_active_failure_for_same_session(tmp_path: Path) -> None: + """A later attempt is current debt; earlier attempts remain durable evidence. + + Anti-vacuity: deleting the lifecycle transition leaves both rows active, so + status and archive debt overcount a single repeatedly failing session. + """ + conn = _connect(tmp_path / "embeddings.db") + prior = record_embedding_failure( + conn, + session_id="codex-session:retry-loop", + origin=Origin.CODEX_SESSION, + message_refs=("codex-session:retry-loop:m1",), + provider="voyage", + model="voyage-4", + error_class="provider_timeout", + error_message="Embedding generation timed out", + retryable=True, + occurred_at_ms=1_800_000_000_000, + ) + current = record_embedding_failure( + conn, + session_id="codex-session:retry-loop", + origin=Origin.CODEX_SESSION, + message_refs=("codex-session:retry-loop:m2",), + provider="voyage", + model="voyage-4", + error_class="provider_http_400", + error_message="Embedding generation failed: HTTP 400", + retryable=False, + occurred_at_ms=1_800_000_100_000, + ) + + superseded = read_embedding_failure(conn, prior.failure_id) + assert superseded.lifecycle_state == "superseded" + assert superseded.resolution_action == "superseded" + assert superseded.superseded_by == current.failure_id + assert superseded.resolved_at_ms == current.created_at_ms + assert list_active_embedding_failures(conn) == (current,) + status = read_embedding_status(conn, "codex-session:retry-loop") + assert status.needs_reindex is False + assert status.error_message == "Embedding generation failed: HTTP 400" From 0193f03248a326673cf1800942e2b903273616f2 Mon Sep 17 00:00:00 2001 From: Sinity Date: Mon, 13 Jul 2026 00:10:49 +0200 Subject: [PATCH 10/16] fix(embeddings): distinguish terminal exclusions from backlog Problem: acknowledged terminal failures could be reported as retry backlog or as complete coverage, depending on the status-detail path. What changed: report blocked sessions separately, exclude them from retry and stale-message counts, and retain partial coverage with an explicit lifecycle explanation. Ref polylogue-egm8 --- polylogue/cli/shared/embed_stats.py | 1 + .../storage/embeddings/status_payload.py | 49 +++++++++++- tests/unit/cli/test_embed_status_fast.py | 77 +++++++++++++++++++ 3 files changed, 123 insertions(+), 4 deletions(-) diff --git a/polylogue/cli/shared/embed_stats.py b/polylogue/cli/shared/embed_stats.py index afa96eb8b2..a5d9832812 100644 --- a/polylogue/cli/shared/embed_stats.py +++ b/polylogue/cli/shared/embed_stats.py @@ -129,6 +129,7 @@ def render_embedding_stats(payload: EmbeddingStatusPayload, *, json_output: bool _render_field("Status", payload["status"]) _render_field("Total sessions", payload["total_sessions"]) _render_field("Embedded sessions", payload["embedded_sessions"]) + _render_field("Blocked sessions", payload["blocked_sessions"]) _render_field("Embedded messages", payload["embedded_messages"]) _render_field("Session coverage", f"{payload['embedding_coverage_percent']:.1f}%") candidate_prose_messages = payload.get("candidate_prose_messages") diff --git a/polylogue/storage/embeddings/status_payload.py b/polylogue/storage/embeddings/status_payload.py index 9d37823498..135ed57815 100644 --- a/polylogue/storage/embeddings/status_payload.py +++ b/polylogue/storage/embeddings/status_payload.py @@ -124,6 +124,7 @@ class EmbeddingStatusPayload(TypedDict): status: str total_sessions: int embedded_sessions: int + blocked_sessions: int embedded_messages: int pending_sessions: int pending_messages: int | None @@ -518,12 +519,13 @@ def _embedding_status( total_sessions: int, embedded_sessions: int, pending_sessions: int, + blocked_sessions: int, ) -> str: if total_sessions <= 0: return "empty" - if pending_sessions <= 0: + if pending_sessions <= 0 and blocked_sessions <= 0: return "complete" - if embedded_sessions <= 0: + if embedded_sessions <= 0 and blocked_sessions <= 0: return "none" return "partial" @@ -553,6 +555,7 @@ def _next_action( retrieval_ready: bool, stale_messages: int, failure_count: int, + blocked_sessions: int, ) -> EmbeddingNextActionPayload: if total_sessions <= 0: return { @@ -572,6 +575,15 @@ def _next_action( "command": "polylogue ops embed status --detail", "reason": "Embedding failures exist and need inspection before treating coverage as clean.", } + if blocked_sessions > 0: + return { + "code": "acknowledged_terminal_exclusions", + "command": None, + "reason": ( + "Some sessions have acknowledged or superseded terminal embedding failures; " + "they are retained as audit evidence but excluded from automatic retry." + ), + } if not config_enabled: if pending_sessions <= 0 and retrieval_ready: return { @@ -633,14 +645,16 @@ def _payload_from_stats( failure_details: list[EmbeddingFailureDetailPayload] | None = None, terminal_failure_count: int = 0, retryable_failure_count: int = 0, + blocked_sessions: int = 0, ) -> EmbeddingStatusPayload: embedded_sessions = stats.embedded_sessions pending_sessions = stats.pending_sessions - eligible_sessions = embedded_sessions + pending_sessions + eligible_sessions = embedded_sessions + pending_sessions + blocked_sessions status = _embedding_status( total_sessions=total_sessions, embedded_sessions=embedded_sessions, pending_sessions=pending_sessions, + blocked_sessions=blocked_sessions, ) if stats.failure_count > 0 and status == "complete": status = "partial" @@ -660,6 +674,7 @@ def _payload_from_stats( "status": status, "total_sessions": total_sessions, "embedded_sessions": embedded_sessions, + "blocked_sessions": blocked_sessions, "embedded_messages": stats.embedded_messages, "pending_sessions": pending_sessions, "pending_messages": stats.pending_messages if pending_messages_exact else None, @@ -701,6 +716,7 @@ def _payload_from_stats( retrieval_ready=retrieval_ready, stale_messages=stats.stale_messages, failure_count=stats.failure_count, + blocked_sessions=blocked_sessions, ), } @@ -735,6 +751,20 @@ def _archive_embedding_status_payload( has_status = bool(status_table) has_meta = bool(meta_table) total_sessions = _scalar_int(conn, "SELECT COUNT(*) FROM sessions") + blocked_sessions = ( + _scalar_int( + conn, + f""" + SELECT COUNT(*) + FROM {status_table} AS e + JOIN sessions AS s ON s.session_id = e.session_id + WHERE COALESCE(e.needs_reindex, 0) = 0 + AND e.error_message IS NOT NULL + """, + ) + if has_status + else 0 + ) embedded_sessions, pending_sessions = _archive_embedding_session_state_summary( conn, status_table=status_table, @@ -751,6 +781,7 @@ def _archive_embedding_status_payload( pending_messages_exact = False else: embedded_sessions, pending_sessions = exact_session_state + pending_sessions = max(pending_sessions - blocked_sessions, 0) if has_status: embedded_messages = _scalar_int( conn, @@ -882,13 +913,19 @@ def _archive_embedding_status_payload( if total_messages is None: total_messages = 0 pending_messages_exact = False - if has_meta and embedded_messages == 0: + if has_meta and embedded_messages == 0 and blocked_sessions == 0: pending_messages = total_messages elif has_meta: meta_join = "ON em.message_id = m.message_id" meta_missing_column = "em.message_id" status_join = f"LEFT JOIN {status_table} e ON e.session_id = m.session_id" if has_status else "" status_reindex_clause = "OR COALESCE(e.needs_reindex, 0) = 1" if has_status else "" + blocked_session_clause = ( + "AND NOT (e.session_id IS NOT NULL AND COALESCE(e.needs_reindex, 0) = 0 " + "AND e.error_message IS NOT NULL)" + if has_status + else "" + ) exact_pending_messages = _scalar_int_with_timeout( conn, f""" @@ -901,6 +938,7 @@ def _archive_embedding_status_payload( OR COALESCE(em.needs_reindex, 0) = 1 {status_reindex_clause} ) + {blocked_session_clause} """, timeout_ms=DETAIL_QUERY_TIMEOUT_MS, ) @@ -941,10 +979,12 @@ def _archive_embedding_status_payload( SELECT COUNT(*) FROM {messages_ref} JOIN {meta_table} em {meta_join} + {status_join} WHERE ( COALESCE(em.needs_reindex, 0) = 1 OR (em.content_hash IS NOT NULL AND em.content_hash != m.content_hash) ) + {blocked_session_clause} """, timeout_ms=DETAIL_QUERY_TIMEOUT_MS, ) @@ -991,6 +1031,7 @@ def _archive_embedding_status_payload( failure_details=failure_details, terminal_failure_count=terminal_failure_count, retryable_failure_count=retryable_failure_count, + blocked_sessions=blocked_sessions, ) diff --git a/tests/unit/cli/test_embed_status_fast.py b/tests/unit/cli/test_embed_status_fast.py index c30f599b8c..54f5f852e1 100644 --- a/tests/unit/cli/test_embed_status_fast.py +++ b/tests/unit/cli/test_embed_status_fast.py @@ -18,6 +18,7 @@ finish_embedding_catchup_run, start_embedding_catchup_run, ) +from polylogue.storage.sqlite.archive_tiers.embedding_write import resolve_embedding_failure class _Cfg: @@ -372,6 +373,82 @@ def test_resolve_failure_cli_requeues_terminal_failure(tmp_path: Path) -> None: ).fetchone() == (1, None) +def test_status_excludes_acknowledged_terminal_failure_from_retry_backlog(tmp_path: Path) -> None: + """An acknowledgement clears critical debt without falsifying coverage. + + Anti-vacuity: removing the blocked-session lifecycle query makes detail + status report this unembedded session as complete, while summary status + presents it as a retryable backlog even though the writer excludes it. + """ + db_anchor = tmp_path / "custom.sqlite" + index_db = tmp_path / "index.db" + _seed_archive_file_set_from_archive_tiers(index_db) + embeddings_db = index_db.with_name("embeddings.db") + with sqlite3.connect(index_db) as conn: + conn.execute("DELETE FROM messages WHERE session_id = 'codex-session:complete'") + conn.execute("DELETE FROM sessions WHERE session_id = 'codex-session:complete'") + with sqlite3.connect(embeddings_db) as conn: + conn.execute("DELETE FROM message_embeddings WHERE message_id = 'codex-session:complete:m1'") + conn.execute("DELETE FROM message_embeddings_meta WHERE message_id = 'codex-session:complete:m1'") + conn.execute("DELETE FROM embedding_status WHERE session_id = 'codex-session:complete'") + with sqlite3.connect(embeddings_db) as conn: + conn.executescript( + """ + CREATE TABLE embedding_failures ( + failure_id TEXT PRIMARY KEY, + session_id TEXT NOT NULL, + origin TEXT NOT NULL, + message_refs_json TEXT NOT NULL, + provider TEXT NOT NULL, + model TEXT NOT NULL, + error_class TEXT NOT NULL, + error_message TEXT NOT NULL, + retryable INTEGER NOT NULL, + lifecycle_state TEXT NOT NULL, + created_at_ms INTEGER NOT NULL, + updated_at_ms INTEGER NOT NULL, + resolved_at_ms INTEGER, + resolution_action TEXT, + resolution_note TEXT, + superseded_by TEXT + ); + INSERT INTO embedding_status VALUES ( + 'codex-session:pending', 'codex-session', 0, 0, 'Embedding generation failed: HTTP 400' + ); + INSERT INTO embedding_failures VALUES ( + 'embedding-failure:terminal', 'codex-session:pending', 'codex-session', + '["codex-session:pending:m1"]', 'voyage', 'voyage-4', 'provider_http_400', + 'Embedding generation failed: HTTP 400', 0, 'terminal', 1800000000000, 1800000000000, + NULL, NULL, NULL, NULL + ); + """ + ) + resolve_embedding_failure( + conn, + failure_id="embedding-failure:terminal", + action="acknowledge", + resolved_at_ms=1_800_000_001_000, + ) + + payload = _run_status(db_anchor, "--detail", cfg=_Cfg(embedding_enabled=True, voyage_api_key="vk-live")) + + assert payload["failure_count"] == 0 + assert payload["terminal_failure_count"] == 0 + assert payload["blocked_sessions"] == 1 + assert payload["pending_sessions"] == 0 + assert payload["pending_messages"] == 0 + assert payload["embedding_coverage_percent"] == 0.0 + assert payload["status"] == "partial" + assert payload["next_action"] == { + "code": "acknowledged_terminal_exclusions", + "command": None, + "reason": ( + "Some sessions have acknowledged or superseded terminal embedding failures; " + "they are retained as audit evidence but excluded from automatic retry." + ), + } + + def test_status_json_detail_does_not_derive_coverage_from_analyzed_prose_estimate(tmp_path: Path) -> None: db_anchor = tmp_path / "custom.sqlite" index_db = tmp_path / "index.db" From ed695789d854b9f462065451b7ea034a25e2ef01 Mon Sep 17 00:00:00 2001 From: Sinity Date: Mon, 13 Jul 2026 00:12:06 +0200 Subject: [PATCH 11/16] fix(embeddings): scope blocked status joins Problem: the blocked-session status query reused SQL fragments outside their conditional definition, failing strict type verification. What changed: define the shared status join and blocked predicate for every detail query path. Ref polylogue-egm8 EOF && nix develop --command git push --- polylogue/storage/embeddings/status_payload.py | 14 +++++++------- 1 file changed, 7 insertions(+), 7 deletions(-) diff --git a/polylogue/storage/embeddings/status_payload.py b/polylogue/storage/embeddings/status_payload.py index 135ed57815..7418f00e32 100644 --- a/polylogue/storage/embeddings/status_payload.py +++ b/polylogue/storage/embeddings/status_payload.py @@ -905,6 +905,13 @@ def _archive_embedding_status_payload( if include_detail and has_messages: candidate_prose_messages, candidate_prose_messages_exact = _candidate_prose_message_count(conn) messages_ref = archive_embeddable_messages_relation(conn, alias="m") + status_join = f"LEFT JOIN {status_table} e ON e.session_id = m.session_id" if has_status else "" + blocked_session_clause = ( + "AND NOT (e.session_id IS NOT NULL AND COALESCE(e.needs_reindex, 0) = 0 " + "AND e.error_message IS NOT NULL)" + if has_status + else "" + ) total_messages = _scalar_int_with_timeout( conn, f"SELECT COUNT(*) FROM {messages_ref}", @@ -918,14 +925,7 @@ def _archive_embedding_status_payload( elif has_meta: meta_join = "ON em.message_id = m.message_id" meta_missing_column = "em.message_id" - status_join = f"LEFT JOIN {status_table} e ON e.session_id = m.session_id" if has_status else "" status_reindex_clause = "OR COALESCE(e.needs_reindex, 0) = 1" if has_status else "" - blocked_session_clause = ( - "AND NOT (e.session_id IS NOT NULL AND COALESCE(e.needs_reindex, 0) = 0 " - "AND e.error_message IS NOT NULL)" - if has_status - else "" - ) exact_pending_messages = _scalar_int_with_timeout( conn, f""" From 52cbbc46a74dea7913828fabe08950c7a92f5e38 Mon Sep 17 00:00:00 2001 From: Sinity Date: Mon, 13 Jul 2026 00:16:27 +0200 Subject: [PATCH 12/16] fix(embeddings): require generation proof for cleanup Problem: orphan cleanup could accept a current-schema index without active rebuild readiness evidence as deletion authority. What changed: require an active, source-snapshotted index generation for every apply and make reconciliation fixtures model that current authority. Ref polylogue-1dk1 --- polylogue/storage/embeddings/reconcile.py | 11 +-- .../test_embedding_orphan_reconcile.py | 67 ++++++++++++++++++- 2 files changed, 67 insertions(+), 11 deletions(-) diff --git a/polylogue/storage/embeddings/reconcile.py b/polylogue/storage/embeddings/reconcile.py index 987bbdd72c..aa7fb8d061 100644 --- a/polylogue/storage/embeddings/reconcile.py +++ b/polylogue/storage/embeddings/reconcile.py @@ -487,20 +487,13 @@ def _assert_index_identity(index_path: Path, expected: _IndexIdentity) -> None: def _assert_active_index_generation(index_path: Path) -> None: - """Require the active source-snapshotted generation when generations exist. - - Legacy archives have a direct ``index.db`` and no generation metadata, so - the schema and inode guards remain their authority proof. Once a rebuild - creates generation metadata, however, a same-schema inactive candidate is - not safe deletion truth: only the generation named by the active pointer - and marked ``active`` after a source snapshot may drive reconciliation. - """ + """Require the active source-snapshotted generation for deletion truth.""" root = index_path.parent generations = root / ".index-generations" metadata_paths = tuple(generations.glob("*/generation.json")) if generations.is_dir() else () if not metadata_paths: - return + raise RuntimeError("embedding orphan reconciliation requires active index generation readiness evidence") pointer = root / ".index-active-pointer" if not pointer.is_file(): diff --git a/tests/unit/storage/test_embedding_orphan_reconcile.py b/tests/unit/storage/test_embedding_orphan_reconcile.py index 1fad67d506..007e865aaf 100644 --- a/tests/unit/storage/test_embedding_orphan_reconcile.py +++ b/tests/unit/storage/test_embedding_orphan_reconcile.py @@ -55,7 +55,13 @@ """ -def _connect_index(path: Path, *, sessions: list[str], messages: dict[str, list[str]]) -> None: +def _connect_index( + path: Path, + *, + sessions: list[str], + messages: dict[str, list[str]], + authoritative_generation: bool = True, +) -> None: """Build a minimal synthetic ``index.db`` with only the sessions/messages listed — standing in for a rebuilt index that dropped some identities.""" @@ -77,6 +83,24 @@ def _connect_index(path: Path, *, sessions: list[str], messages: dict[str, list[ conn.commit() finally: conn.close() + if authoritative_generation: + generations = path.parent / ".index-generations" / "gen-current" + generations.mkdir(parents=True) + (path.parent / ".index-active-pointer").write_text(str(path.resolve()), encoding="utf-8") + (generations / "generation.json").write_text( + json.dumps( + { + "generation_id": "gen-current", + "owner_id": "test", + "archive_root": str(path.parent), + "index_path": str(path), + "state": "active", + "created_at_ms": _NOW_MS, + "source_snapshot": "source-at-rebuild-start", + } + ), + encoding="utf-8", + ) def _connect_embeddings(path: Path) -> sqlite3.Connection: @@ -662,7 +686,12 @@ def test_apply_refuses_inactive_generation_even_when_schema_matches(tmp_path: Pa session_id = "codex-session:inactive-generation" message_id = f"{session_id}:orphan" index_db = tmp_path / "index.db" - _connect_index(index_db, sessions=[session_id], messages={session_id: []}) + _connect_index( + index_db, + sessions=[session_id], + messages={session_id: []}, + authoritative_generation=False, + ) generations = tmp_path / ".index-generations" / "gen-inactive" generations.mkdir(parents=True) (tmp_path / ".index-active-pointer").write_text(str(index_db.resolve()), encoding="utf-8") @@ -697,3 +726,37 @@ def test_apply_refuses_inactive_generation_even_when_schema_matches(tmp_path: Pa with _connect_embeddings(embeddings_db) as verify: assert verify.execute("SELECT COUNT(*) FROM message_embeddings_meta").fetchone()[0] == 1 assert verify.execute("SELECT COUNT(*) FROM message_embeddings").fetchone()[0] == 1 + + +def test_apply_refuses_generationless_index_even_when_schema_matches(tmp_path: Path) -> None: + """Deletion requires rebuild-readiness evidence, not schema version alone. + + Anti-vacuity: restoring the generation-less compatibility path lets a + current-schema index with no active source snapshot delete this orphan. + """ + session_id = "codex-session:generationless" + message_id = f"{session_id}:orphan" + index_db = tmp_path / "index.db" + _connect_index( + index_db, + sessions=[session_id], + messages={session_id: []}, + authoritative_generation=False, + ) + embeddings_db = tmp_path / "embeddings.db" + conn = _connect_embeddings(embeddings_db) + _write_embedding(conn, message_id=message_id, session_id=session_id, embedded_at_ms=_OLD_MS) + conn.close() + + with pytest.raises(RuntimeError, match="requires active index generation readiness evidence"): + reconcile_embedding_orphans( + index_db, + embeddings_db, + dry_run=False, + now_ms=_NOW_MS, + mutation_authority="offline-exclusive", + ) + + with _connect_embeddings(embeddings_db) as verify: + assert verify.execute("SELECT COUNT(*) FROM message_embeddings_meta").fetchone()[0] == 1 + assert verify.execute("SELECT COUNT(*) FROM message_embeddings").fetchone()[0] == 1 From f69c3374e108b09cecb5dad5ee669abd173965cd Mon Sep 17 00:00:00 2001 From: Sinity Date: Mon, 13 Jul 2026 00:18:13 +0200 Subject: [PATCH 13/16] fix(embeddings): ledger every status error Problem: a public status-error writer could produce terminal debt without the failure identity needed for inspection or resolution. What changed: route that helper through the lifecycle recorder while preserving its status return contract. Ref polylogue-egm8 --- .../sqlite/archive_tiers/embedding_write.py | 28 ++++++++----------- .../test_archive_tiers_embedding_write.py | 6 ++++ 2 files changed, 18 insertions(+), 16 deletions(-) diff --git a/polylogue/storage/sqlite/archive_tiers/embedding_write.py b/polylogue/storage/sqlite/archive_tiers/embedding_write.py index bac0140818..6889d9d95c 100644 --- a/polylogue/storage/sqlite/archive_tiers/embedding_write.py +++ b/polylogue/storage/sqlite/archive_tiers/embedding_write.py @@ -154,22 +154,18 @@ def mark_session_embedding_error( error_message: str, retryable: bool = True, ) -> ArchiveEmbeddingStatus: - """Record a resumable embedding error for one session.""" - origin_value = _enum_value(origin) - needs_reindex = 1 if retryable else 0 - with conn: - conn.execute( - """ - INSERT INTO embedding_status ( - session_id, origin, message_count_embedded, last_embedded_at_ms, needs_reindex, error_message - ) VALUES (?, ?, 0, NULL, ?, ?) - ON CONFLICT(session_id) DO UPDATE SET - origin = excluded.origin, - needs_reindex = excluded.needs_reindex, - error_message = excluded.error_message - """, - (session_id, origin_value, needs_reindex, error_message), - ) + """Record a failure with a status projection for compatibility callers.""" + record_embedding_failure( + conn, + session_id=session_id, + origin=origin, + message_refs=(), + provider="unknown", + model="unknown", + error_class="embedding_error", + error_message=error_message, + retryable=retryable, + ) return read_embedding_status(conn, session_id) diff --git a/tests/unit/storage/test_archive_tiers_embedding_write.py b/tests/unit/storage/test_archive_tiers_embedding_write.py index 38014b44ff..aaf5945db3 100644 --- a/tests/unit/storage/test_archive_tiers_embedding_write.py +++ b/tests/unit/storage/test_archive_tiers_embedding_write.py @@ -150,6 +150,12 @@ def test_archive_tiers_embedding_writer_records_terminal_errors(tmp_path: Path) needs_reindex=False, error_message="Embedding generation failed: HTTP 400", ) + [failure] = list_active_embedding_failures(conn) + assert failure.session_id == "codex-session:bad-input" + assert failure.lifecycle_state == "terminal" + assert failure.message_refs == () + assert failure.provider == "unknown" + assert failure.error_class == "embedding_error" def test_embedding_failure_lifecycle_preserves_audit_and_requeues(tmp_path: Path) -> None: From b057079111b0351ad6d0760053eedaf549c8e63d Mon Sep 17 00:00:00 2001 From: Sinity Date: Mon, 13 Jul 2026 06:23:06 +0200 Subject: [PATCH 14/16] fix(embeddings): classify 'timed out' provider errors as provider_timeout Review follow-up on #2796: the timeout classifier matched only the single-token spelling. Co-Authored-By: Claude --- polylogue/storage/embeddings/materialization.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/polylogue/storage/embeddings/materialization.py b/polylogue/storage/embeddings/materialization.py index 55d87a0b74..5dcdc3d215 100644 --- a/polylogue/storage/embeddings/materialization.py +++ b/polylogue/storage/embeddings/materialization.py @@ -83,7 +83,7 @@ def embedding_error_class(error_message: object) -> str: return "provider_http_400" if "http 429" in normalized or "status 429" in normalized: return "provider_http_429" - if "timeout" in normalized: + if "timeout" in normalized or "timed out" in normalized: return "provider_timeout" return "provider_error" From 4326d07dc70c68d4d2956368f4fc1e0e644c1161 Mon Sep 17 00:00:00 2001 From: Sinity Date: Mon, 13 Jul 2026 07:13:27 +0200 Subject: [PATCH 15/16] fix(embeddings): resolve generation metadata from pointer anchor Problem: archive file-set deployments expose index.db through a public\nsymlink while storing generation metadata beside the database-tier\npointer anchor. The orphan reconciler checked the public symlink\ndirectory, so it could never authorize a correctly active external tier.\n\nWhat changed: validate the configured active-pointer anchor, resolve the\nactive index through it, and read generation metadata beside that anchor.\nThe generation state, source-snapshot, schema, identity, and pre-commit\nguards remain mandatory.\n\nVerification: devtools test tests/unit/storage/test_embedding_orphan_reconcile.py\n(16 passed); devtools test tests/unit/storage -k embedding (89 passed).\n\nRef polylogue-1dk1\n\nCo-Authored-By: Claude --- polylogue/storage/embeddings/reconcile.py | 27 ++++++--- .../test_embedding_orphan_reconcile.py | 58 ++++++++++++++++++- 2 files changed, 76 insertions(+), 9 deletions(-) diff --git a/polylogue/storage/embeddings/reconcile.py b/polylogue/storage/embeddings/reconcile.py index aa7fb8d061..5133609ad9 100644 --- a/polylogue/storage/embeddings/reconcile.py +++ b/polylogue/storage/embeddings/reconcile.py @@ -489,17 +489,19 @@ def _assert_index_identity(index_path: Path, expected: _IndexIdentity) -> None: def _assert_active_index_generation(index_path: Path) -> None: """Require the active source-snapshotted generation for deletion truth.""" - root = index_path.parent - generations = root / ".index-generations" - metadata_paths = tuple(generations.glob("*/generation.json")) if generations.is_dir() else () - if not metadata_paths: - raise RuntimeError("embedding orphan reconciliation requires active index generation readiness evidence") - - pointer = root / ".index-active-pointer" + configured_root = index_path.parent + pointer = configured_root / ".index-active-pointer" if not pointer.is_file(): raise RuntimeError("embedding orphan reconciliation requires an active index generation pointer") try: - pointed_path = Path(pointer.read_text(encoding="utf-8").strip()).resolve(strict=True) + pointer_anchor = Path(pointer.read_text(encoding="utf-8").strip()) + if ( + not pointer_anchor.is_absolute() + or pointer_anchor.name != "index.db" + or ".index-generations" in pointer_anchor.parts + ): + raise ValueError("invalid active index generation pointer anchor") + pointed_path = pointer_anchor.resolve(strict=True) except (OSError, ValueError) as exc: raise RuntimeError( "embedding orphan reconciliation found an unreadable active index generation pointer" @@ -507,6 +509,15 @@ def _assert_active_index_generation(index_path: Path) -> None: if pointed_path != index_path.resolve(strict=True): raise RuntimeError("embedding orphan reconciliation refuses a non-active index generation") + # The configured archive root may expose ``index.db`` as a symlink into a + # separately mounted database tier. IndexGenerationStore deliberately + # keeps generation metadata beside the pointer anchor, not beside that + # public symlink, so resolve readiness from the anchor's parent. + generations = pointer_anchor.parent / ".index-generations" + metadata_paths = tuple(generations.glob("*/generation.json")) if generations.is_dir() else () + if not metadata_paths: + raise RuntimeError("embedding orphan reconciliation requires active index generation readiness evidence") + for metadata_path in metadata_paths: try: payload = json.loads(metadata_path.read_text(encoding="utf-8")) diff --git a/tests/unit/storage/test_embedding_orphan_reconcile.py b/tests/unit/storage/test_embedding_orphan_reconcile.py index 007e865aaf..2be7949e77 100644 --- a/tests/unit/storage/test_embedding_orphan_reconcile.py +++ b/tests/unit/storage/test_embedding_orphan_reconcile.py @@ -217,6 +217,62 @@ def test_reconcile_removes_message_orphaned_by_index_rebuild(tmp_path: Path) -> assert [item.session_id for item in pending] == [session_id] +def test_apply_accepts_generation_metadata_beside_external_pointer_anchor(tmp_path: Path) -> None: + """A public archive symlink must use the anchor tier's generation metadata. + + Anti-vacuity: looking beside ``public_root/index.db`` rather than beside + the active-pointer anchor makes this production-shaped layout refuse the + otherwise authorized orphan deletion. + """ + public_root = tmp_path / "archive" + tier_root = tmp_path / "db-tier" + public_root.mkdir() + tier_root.mkdir() + session_id = "codex-session:external-tier" + message_id = f"{session_id}:orphan" + tier_index = tier_root / "index.db" + _connect_index(tier_index, sessions=[session_id], messages={session_id: []}, authoritative_generation=False) + + public_index = public_root / "index.db" + public_index.symlink_to(tier_index) + (public_root / ".index-active-pointer").write_text(str(tier_index), encoding="utf-8") + generation_dir = tier_root / ".index-generations" / "gen-current" + generation_dir.mkdir(parents=True) + (generation_dir / "generation.json").write_text( + json.dumps( + { + "generation_id": "gen-current", + "owner_id": "test", + "archive_root": str(public_root), + "index_path": str(tier_index), + "state": "active", + "created_at_ms": _NOW_MS, + "source_snapshot": "source-at-rebuild-start", + } + ), + encoding="utf-8", + ) + + embeddings_db = public_root / "embeddings.db" + conn = _connect_embeddings(embeddings_db) + _write_embedding(conn, message_id=message_id, session_id=session_id, embedded_at_ms=_OLD_MS) + conn.close() + + report = reconcile_embedding_orphans( + public_index, + embeddings_db, + dry_run=False, + now_ms=_NOW_MS, + mutation_authority="offline-exclusive", + ) + + assert report.removed_message_rows == 1 + assert report.removed_vector_rows == 1 + with _connect_embeddings(embeddings_db) as verify: + assert verify.execute("SELECT COUNT(*) FROM message_embeddings_meta").fetchone()[0] == 0 + assert verify.execute("SELECT COUNT(*) FROM message_embeddings").fetchone()[0] == 0 + + def test_reconcile_preserves_content_hash_mismatch_when_identity_present(tmp_path: Path) -> None: """Content-hash guard: a message that still exists (identity present) is never removed merely because its stored content_hash is stale — that is @@ -748,7 +804,7 @@ def test_apply_refuses_generationless_index_even_when_schema_matches(tmp_path: P _write_embedding(conn, message_id=message_id, session_id=session_id, embedded_at_ms=_OLD_MS) conn.close() - with pytest.raises(RuntimeError, match="requires active index generation readiness evidence"): + with pytest.raises(RuntimeError, match="requires an active index generation pointer"): reconcile_embedding_orphans( index_db, embeddings_db, From 16ce55fd5d55ad8cedba8c06ad3726fe62fdf605 Mon Sep 17 00:00:00 2001 From: Sinity Date: Mon, 13 Jul 2026 08:16:55 +0200 Subject: [PATCH 16/16] fix(embeddings): resolve review findings on failure lifecycle honesty Problem: CodeRabbit review of #2796 surfaced three valid defects. The no_messages/no_embeddable_messages success exits skipped failure resolution, leaving phantom active debt. The blanket exception handler ledgered local faults (sqlite-vec load, SQL, content-hash, write) as provider=voyage failures. Acknowledge/supersede of a retryable failure left needs_reindex=1, so the session stayed in the automatic backlog despite the operator resolution. The rendered resolution_command also used a bare ACTION placeholder that was not visibly a template. What changed: failure resolution moved into _record_archive_embedding_success so every terminal success outcome clears open failures; provider exceptions are wrapped in _ProviderRequestError so only genuine provider faults carry provider=voyage while local faults record provider=local / internal_error; non-requeue resolutions clear needs_reindex but keep error_message, moving the session from pending backlog to the visible blocked count; resolution_command now renders --action . Regression tests cover all three behaviors plus provider-attribution preservation. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_013v95mgEuw3uo9AsnpvhDbm --- .../storage/embeddings/materialization.py | 35 +++- .../storage/embeddings/status_payload.py | 3 +- .../sqlite/archive_tiers/embedding_write.py | 9 + tests/unit/cli/test_embed_status_fast.py | 3 +- .../test_archive_tiers_embedding_write.py | 37 ++++ .../unit/storage/test_embedding_contracts.py | 158 ++++++++++++++++++ 6 files changed, 235 insertions(+), 10 deletions(-) diff --git a/polylogue/storage/embeddings/materialization.py b/polylogue/storage/embeddings/materialization.py index 5dcdc3d215..08ebfd7fbc 100644 --- a/polylogue/storage/embeddings/materialization.py +++ b/polylogue/storage/embeddings/materialization.py @@ -900,6 +900,10 @@ async def _view_title() -> Session | None: ) +class _ProviderRequestError(RuntimeError): + """Marks an exception raised by the embedding provider call itself.""" + + def embed_archive_session_sync( index_db_path: Path, vec_provider: VectorProvider, @@ -985,9 +989,12 @@ def embed_archive_session_sync( for start in range(0, len(embeddable), batch_size): batch = embeddable[start : start + batch_size] attempted_message_refs = tuple(str(row["message_id"]) for row in batch) - embeddings = text_provider._get_embeddings([str(row["text"]) for row in batch], input_type="document") + try: + embeddings = text_provider._get_embeddings([str(row["text"]) for row in batch], input_type="document") + except Exception as exc: + raise _ProviderRequestError(str(exc)) from exc if len(embeddings) != len(batch): - raise RuntimeError("embedding provider returned a mismatched vector count") + raise _ProviderRequestError("embedding provider returned a mismatched vector count") writes: list[ArchiveEmbeddingWrite] = [] for row, embedding in zip(batch, embeddings, strict=True): if row["content_hash"] is None: @@ -1011,9 +1018,6 @@ def embed_archive_session_sync( message_count=len(embeddable), model=text_provider.model, ) - from polylogue.storage.sqlite.archive_tiers.embedding_write import resolve_open_embedding_failures_for_session - - resolve_open_embedding_failures_for_session(embeddings_conn, session_id=session_id) except Exception as exc: try: from polylogue.storage.sqlite.archive_tiers.embedding_write import record_embedding_failure @@ -1022,16 +1026,26 @@ def embed_archive_session_sync( "SELECT origin FROM sessions WHERE session_id = ?", (session_id,) ).fetchone() if origin_row is not None: + if isinstance(exc, _ProviderRequestError): + provider = "voyage" + error_class = embedding_error_class(exc) + retryable = not is_terminal_embedding_provider_error(str(exc)) + else: + # Local faults (sqlite-vec load, SQL, content-hash validation, + # write) must not masquerade as provider failures in the ledger. + provider = "local" + error_class = "internal_error" + retryable = True record_embedding_failure( embeddings_conn, session_id=session_id, origin=str(origin_row["origin"]), message_refs=attempted_message_refs, - provider="voyage", + provider=provider, model=text_provider.model, - error_class=embedding_error_class(exc), + error_class=error_class, error_message=str(exc), - retryable=not is_terminal_embedding_provider_error(str(exc)), + retryable=retryable, ) finally: with contextlib.suppress(sqlite3.Error): @@ -1103,6 +1117,11 @@ def _record_archive_embedding_success( """, (session_id, origin, message_count, now_ms, needs_reindex), ) + # Every terminal success outcome — including "nothing to embed" — resolves + # the session's open failures, or they linger as phantom debt. + from polylogue.storage.sqlite.archive_tiers.embedding_write import resolve_open_embedding_failures_for_session + + resolve_open_embedding_failures_for_session(conn, session_id=session_id) _PROSE_MATERIAL_ORIGINS = frozenset({"human_authored", "assistant_authored"}) diff --git a/polylogue/storage/embeddings/status_payload.py b/polylogue/storage/embeddings/status_payload.py index 7418f00e32..340270ea2e 100644 --- a/polylogue/storage/embeddings/status_payload.py +++ b/polylogue/storage/embeddings/status_payload.py @@ -321,7 +321,8 @@ def _active_failure_details( "resolution_action": None if row[12] is None else str(row[12]), "supported_actions": ["acknowledge", "requeue", "supersede"], "resolution_command": ( - f"polylogue ops embed resolve-failure {shlex.quote(str(row[0]))} --action ACTION --yes" + f"polylogue ops embed resolve-failure {shlex.quote(str(row[0]))}" + " --action --yes" ), } ) diff --git a/polylogue/storage/sqlite/archive_tiers/embedding_write.py b/polylogue/storage/sqlite/archive_tiers/embedding_write.py index 6889d9d95c..2e4299da76 100644 --- a/polylogue/storage/sqlite/archive_tiers/embedding_write.py +++ b/polylogue/storage/sqlite/archive_tiers/embedding_write.py @@ -277,6 +277,15 @@ def resolve_embedding_failure( "UPDATE embedding_status SET needs_reindex = 1, error_message = NULL WHERE session_id = ?", (str(row[0]),), ) + else: + # acknowledge/supersede end the retry loop: clear the failure-driven + # requeue flag but keep error_message, which moves the session from + # the pending backlog into the visible blocked count. The guard keeps + # config-change reindex marks (which never set error_message) intact. + conn.execute( + "UPDATE embedding_status SET needs_reindex = 0 WHERE session_id = ? AND error_message IS NOT NULL", + (str(row[0]),), + ) return read_embedding_failure(conn, failure_id) diff --git a/tests/unit/cli/test_embed_status_fast.py b/tests/unit/cli/test_embed_status_fast.py index 54f5f852e1..13412b9635 100644 --- a/tests/unit/cli/test_embed_status_fast.py +++ b/tests/unit/cli/test_embed_status_fast.py @@ -285,7 +285,8 @@ def test_status_detail_exposes_bounded_terminal_failure_resolution(tmp_path: Pat "resolution_action": None, "supported_actions": ["acknowledge", "requeue", "supersede"], "resolution_command": ( - "polylogue ops embed resolve-failure embedding-failure:terminal --action ACTION --yes" + "polylogue ops embed resolve-failure embedding-failure:terminal" + " --action --yes" ), } ] diff --git a/tests/unit/storage/test_archive_tiers_embedding_write.py b/tests/unit/storage/test_archive_tiers_embedding_write.py index aaf5945db3..68a11564c9 100644 --- a/tests/unit/storage/test_archive_tiers_embedding_write.py +++ b/tests/unit/storage/test_archive_tiers_embedding_write.py @@ -277,3 +277,40 @@ def test_new_failure_supersedes_prior_active_failure_for_same_session(tmp_path: status = read_embedding_status(conn, "codex-session:retry-loop") assert status.needs_reindex is False assert status.error_message == "Embedding generation failed: HTTP 400" + + +def test_acknowledging_retryable_failure_stops_auto_requeue_but_stays_blocked(tmp_path: Path) -> None: + """Operator acknowledgement of a retryable failure ends the retry loop. + + Anti-vacuity: without the non-requeue resolution clearing needs_reindex, + the session stays in the automatic backlog despite the operator resolution; + clearing error_message too would erase it from the visible blocked count. + """ + conn = _connect(tmp_path / "embeddings.db") + failure = record_embedding_failure( + conn, + session_id="codex-session:flaky", + origin=Origin.CODEX_SESSION, + message_refs=("codex-session:flaky:m1",), + provider="voyage", + model="voyage-4", + error_class="provider_timeout", + error_message="Embedding generation timed out", + retryable=True, + occurred_at_ms=1_800_000_000_000, + ) + before = read_embedding_status(conn, "codex-session:flaky") + assert before.needs_reindex is True + assert before.error_message == "Embedding generation timed out" + + acknowledged = resolve_embedding_failure( + conn, + failure_id=failure.failure_id, + action="acknowledge", + note="known flaky historical payload", + resolved_at_ms=1_800_000_000_100, + ) + assert acknowledged.lifecycle_state == "acknowledged" + after = read_embedding_status(conn, "codex-session:flaky") + assert after.needs_reindex is False + assert after.error_message == "Embedding generation timed out" diff --git a/tests/unit/storage/test_embedding_contracts.py b/tests/unit/storage/test_embedding_contracts.py index 4120268765..63380232ea 100644 --- a/tests/unit/storage/test_embedding_contracts.py +++ b/tests/unit/storage/test_embedding_contracts.py @@ -1468,3 +1468,161 @@ def test_missing_vec_module_treated_as_optional(self) -> None: assert stats.embedded_sessions == 0 assert stats.embedded_messages == 0 assert stats.pending_sessions == 0 + + +def _write_archive_session(archive_root: Path, *, native_id: str, embeddable: bool) -> str: + from polylogue.archive.message.roles import Role + from polylogue.core.enums import BlockType, MaterialOrigin, Provider + from polylogue.sources.parsers.base import ParsedContentBlock, ParsedMessage, ParsedSession + from polylogue.storage.sqlite.archive_tiers.archive import ArchiveStore + + text = "This archive message is long enough to embed for semantic search." + with ArchiveStore(archive_root) as archive: + return archive.write_parsed( + ParsedSession( + source_name=Provider.CODEX, + provider_session_id=native_id, + title="failure resolution session", + messages=[ + ParsedMessage( + provider_message_id="m1", + role=Role.USER, + text=text, + blocks=[ParsedContentBlock(type=BlockType.TEXT, text=text)], + material_origin=(MaterialOrigin.HUMAN_AUTHORED if embeddable else MaterialOrigin.TOOL_RESULT), + ) + ], + ) + ) + + +def _seed_open_archive_failure(embeddings_db: Path, session_id: str) -> str: + from polylogue.storage.sqlite.archive_tiers.bootstrap import initialize_archive_database + from polylogue.storage.sqlite.archive_tiers.embedding_write import record_embedding_failure + from polylogue.storage.sqlite.archive_tiers.types import ArchiveTier + + initialize_archive_database(embeddings_db, ArchiveTier.EMBEDDINGS) + conn = sqlite3.connect(embeddings_db) + try: + failure = record_embedding_failure( + conn, + session_id=session_id, + origin="codex-session", + message_refs=(f"{session_id}:m1",), + provider="voyage", + model="voyage-4", + error_class="provider_error", + error_message="Embedding generation failed: HTTP 500", + retryable=True, + ) + return failure.failure_id + finally: + conn.close() + + +def _failure_lifecycle_state(embeddings_db: Path, failure_id: str) -> str: + conn = sqlite3.connect(embeddings_db) + try: + row = conn.execute( + "SELECT lifecycle_state FROM embedding_failures WHERE failure_id = ?", (failure_id,) + ).fetchone() + assert row is not None + return str(row[0]) + finally: + conn.close() + + +def test_archive_success_outcomes_resolve_open_failures(tmp_path: Path) -> None: + """Every terminal success outcome clears prior failure debt. + + Anti-vacuity: dropping resolve_open_embedding_failures_for_session from + _record_archive_embedding_success leaves both failures active, so a session + that later embeds (or turns out to have nothing embeddable) reports phantom + current debt forever. + """ + archive_root = tmp_path / "archive" + embedded_session = _write_archive_session(archive_root, native_id="embed-ok", embeddable=True) + noop_session = _write_archive_session(archive_root, native_id="embed-noop", embeddable=False) + index_db = archive_root / "index.db" + embeddings_db = archive_root / "embeddings.db" + embedded_failure = _seed_open_archive_failure(embeddings_db, embedded_session) + noop_failure = _seed_open_archive_failure(embeddings_db, noop_session) + + embedded_outcome = embed_archive_session_sync(index_db, _FakeV1VectorProvider(), embedded_session) + noop_outcome = embed_archive_session_sync(index_db, _FakeV1VectorProvider(), noop_session) + + assert embedded_outcome.status == "embedded" + assert noop_outcome.status == "no_embeddable_messages" + assert _failure_lifecycle_state(embeddings_db, embedded_failure) == "resolved" + assert _failure_lifecycle_state(embeddings_db, noop_failure) == "resolved" + + +def test_archive_local_fault_is_not_ledgered_as_provider_failure( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """Local storage faults must not masquerade as Voyage provider failures. + + Anti-vacuity: recording every materialization exception with + provider="voyage" falsifies the audit ledger; the production handler must + branch on whether the provider call itself raised. + """ + from polylogue.storage.sqlite.archive_tiers import embedding_write as embedding_write_module + from polylogue.storage.sqlite.archive_tiers.bootstrap import initialize_archive_database + from polylogue.storage.sqlite.archive_tiers.types import ArchiveTier + + archive_root = tmp_path / "archive" + session_id = _write_archive_session(archive_root, native_id="embed-local-fault", embeddable=True) + index_db = archive_root / "index.db" + embeddings_db = archive_root / "embeddings.db" + initialize_archive_database(embeddings_db, ArchiveTier.EMBEDDINGS) + + def _raise_write_fault(*args: object, **kwargs: object) -> None: + raise sqlite3.OperationalError("database is locked") + + monkeypatch.setattr(embedding_write_module, "upsert_message_embeddings", _raise_write_fault) + outcome = embed_archive_session_sync(index_db, _FakeV1VectorProvider(), session_id) + assert outcome.status == "error" + + conn = sqlite3.connect(embeddings_db) + try: + provider, error_class, retryable = conn.execute( + "SELECT provider, error_class, retryable FROM embedding_failures WHERE session_id = ?", + (session_id,), + ).fetchone() + finally: + conn.close() + assert provider == "local" + assert error_class == "internal_error" + assert bool(retryable) is True + + +def test_archive_provider_fault_keeps_provider_attribution(tmp_path: Path) -> None: + """A genuine provider exception still ledgers as a Voyage failure.""" + archive_root = tmp_path / "archive" + session_id = _write_archive_session(archive_root, native_id="embed-provider-fault", embeddable=True) + index_db = archive_root / "index.db" + embeddings_db = archive_root / "embeddings.db" + from polylogue.storage.sqlite.archive_tiers.bootstrap import initialize_archive_database + from polylogue.storage.sqlite.archive_tiers.types import ArchiveTier + + initialize_archive_database(embeddings_db, ArchiveTier.EMBEDDINGS) + + provider = _FakeV1VectorProvider() + + def _raise_provider_fault(texts: list[str], input_type: str = "document") -> list[list[float]]: + raise RuntimeError("Embedding generation failed: HTTP 429") + + provider._get_embeddings = _raise_provider_fault # type: ignore[method-assign] + outcome = embed_archive_session_sync(index_db, provider, session_id) + assert outcome.status == "error" + + conn = sqlite3.connect(embeddings_db) + try: + recorded_provider, error_class = conn.execute( + "SELECT provider, error_class FROM embedding_failures WHERE session_id = ?", + (session_id,), + ).fetchone() + finally: + conn.close() + assert recorded_provider == "voyage" + assert error_class == "provider_http_429"