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/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/cli/shared/embed_stats.py b/polylogue/cli/shared/embed_stats.py index f3575e76a2..a5d9832812 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: @@ -110,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") @@ -133,6 +153,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/polylogue/daemon/embedding_readiness.py b/polylogue/daemon/embedding_readiness.py index 24ed7136ac..0233d69e9c 100644 --- a/polylogue/daemon/embedding_readiness.py +++ b/polylogue/daemon/embedding_readiness.py @@ -29,6 +29,9 @@ 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, "embedding_latest_material_catchup_run": None, @@ -88,6 +91,9 @@ 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"], "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..47dd0ecd3a 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")) @@ -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 + terminal_failures = _int_value(info.get("embedding_terminal_failure_count")) or 0 if config_enabled and not has_key: rows.append( @@ -975,14 +976,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/materialization.py b/polylogue/storage/embeddings/materialization.py index 507eb50a71..08ebfd7fbc 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 or "timed out" 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.""" @@ -887,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, @@ -906,6 +923,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,9 +988,13 @@ 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] - embeddings = text_provider._get_embeddings([str(row["text"]) for row in batch], input_type="document") + attempted_message_refs = tuple(str(row["message_id"]) for row in batch) + 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: @@ -998,18 +1020,32 @@ def embed_archive_session_sync( ) 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( + 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=provider, + model=text_provider.model, + 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): @@ -1081,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/reconcile.py b/polylogue/storage/embeddings/reconcile.py index 098aa0b27d..5133609ad9 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,55 @@ 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.""" + + 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: + 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" + ) from exc + 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")) + 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/polylogue/storage/embeddings/status_payload.py b/polylogue/storage/embeddings/status_payload.py index 7d0a09df6d..340270ea2e 100644 --- a/polylogue/storage/embeddings/status_payload.py +++ b/polylogue/storage/embeddings/status_payload.py @@ -7,6 +7,8 @@ from __future__ import annotations +import json +import shlex import sqlite3 import time from datetime import UTC, datetime @@ -36,6 +38,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 +96,24 @@ 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 + supported_actions: list[str] + resolution_command: str + + class EmbeddingStatusPayload(TypedDict): config_enabled: bool has_voyage_api_key: bool @@ -103,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 @@ -122,6 +144,9 @@ 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 latest_material_catchup_run: EmbeddingCatchupRunPayload | None @@ -248,6 +273,62 @@ 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]), + "supported_actions": ["acknowledge", "requeue", "supersede"], + "resolution_command": ( + f"polylogue ops embed resolve-failure {shlex.quote(str(row[0]))}" + " --action --yes" + ), + } + ) + return details + + def _uniform_embedding_metadata_counts( conn: sqlite3.Connection, meta_table: str, @@ -439,12 +520,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" @@ -474,6 +556,7 @@ def _next_action( retrieval_ready: bool, stale_messages: int, failure_count: int, + blocked_sessions: int, ) -> EmbeddingNextActionPayload: if total_sessions <= 0: return { @@ -493,6 +576,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 { @@ -551,14 +643,19 @@ 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, + 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" @@ -578,6 +675,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, @@ -604,6 +702,9 @@ 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, "latest_material_catchup_run": latest_material_catchup_run, @@ -616,6 +717,7 @@ def _payload_from_stats( retrieval_ready=retrieval_ready, stale_messages=stats.stale_messages, failure_count=stats.failure_count, + blocked_sessions=blocked_sessions, ), } @@ -640,14 +742,30 @@ 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) 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, @@ -664,6 +782,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, @@ -688,16 +807,48 @@ 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 + ) + ) + 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 ) - 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 candidate_prose_messages_exact = False @@ -755,6 +906,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}", @@ -763,12 +921,11 @@ 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 "" exact_pending_messages = _scalar_int_with_timeout( conn, @@ -782,6 +939,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, ) @@ -822,10 +980,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, ) @@ -869,6 +1029,10 @@ 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, + terminal_failure_count=terminal_failure_count, + retryable_failure_count=retryable_failure_count, + blocked_sessions=blocked_sessions, ) diff --git a/polylogue/storage/sqlite/archive_tiers/embedding_write.py b/polylogue/storage/sqlite/archive_tiers/embedding_write.py index 9975453e79..2e4299da76 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, cast 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, *, @@ -126,9 +154,41 @@ def mark_session_embedding_error( error_message: str, retryable: bool = True, ) -> ArchiveEmbeddingStatus: - """Record a resumable embedding error for one session.""" + """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) + + +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) - needs_reindex = 1 if retryable else 0 + refs = tuple(dict.fromkeys(str(ref) for ref in message_refs)) with conn: conn.execute( """ @@ -140,9 +200,170 @@ def mark_session_embedding_error( needs_reindex = excluded.needs_reindex, error_message = excluded.error_message """, - (session_id, origin_value, needs_reindex, error_message), + (session_id, origin_value, 1 if retryable else 0, error_message), ) - return read_embedding_status(conn, session_id) + 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 ( + 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 = 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')", + (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]),), + ) + 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) + + +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=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]), + ) def read_embedding_meta(conn: sqlite3.Connection, target_id: str) -> ArchiveEmbeddingMeta: @@ -196,11 +417,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/cli/test_embed_status_fast.py b/tests/unit/cli/test_embed_status_fast.py index d1499881c7..13412b9635 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: @@ -149,13 +150,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) @@ -214,6 +220,236 @@ 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. + + 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) + 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["terminal_failure_count"] == 1 + assert payload["retryable_failure_count"] == 0 + 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 --yes" + ), + } + ] + 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_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_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" diff --git a/tests/unit/storage/test_archive_tiers_embedding_write.py b/tests/unit/storage/test_archive_tiers_embedding_write.py index a78f5f08fc..68a11564c9 100644 --- a/tests/unit/storage/test_archive_tiers_embedding_write.py +++ b/tests/unit/storage/test_archive_tiers_embedding_write.py @@ -9,11 +9,16 @@ 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_failure, read_embedding_status, + record_embedding_failure, + resolve_embedding_failure, upsert_message_embedding, upsert_message_embeddings, ) @@ -145,3 +150,167 @@ 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: + """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 + + +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" + + +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" diff --git a/tests/unit/storage/test_embedding_orphan_reconcile.py b/tests/unit/storage/test_embedding_orphan_reconcile.py index e65399413e..2be7949e77 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 @@ -54,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.""" @@ -76,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: @@ -192,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 @@ -650,3 +731,88 @@ 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: []}, + 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") + (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 + + +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 an active index generation pointer"): + 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