fix(embeddings): make embedding lifecycle actionable

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

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
 

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"])

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"],

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:

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"})

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