feat(retain): optional fail-on-extraction-errors flag (#2721)

Add opt-in HINDSIGHT_API_FAIL_ON_EXTRACTION_ERRORS (default False, preserves behavior). When enabled and a retain accumulated extraction errors (extraction_errors_count > 0), the operation is marked failed instead of silently completed. Self-contained: config flag + _mark_operation_completed decision + docs + .env.example. Deferred follow-ups (status API field, completed_with_errors status, webhook field, metric) noted in the PR.

Own change verified (config + status tests pass, verify-generated-files green, ruff/tsc clean). Remaining CI reds are unrelated: Core LLM + pg0 test-api flakes, a transient ts-client-oracle, and test-embed-windows (the #2676 regression already fixed on main by #2723; #2721 does not touch the embed daemon).

Fixes #2700

Author: benfrank241

.env.example

@@ -89,6 +89,10 @@ HINDSIGHT_API_LOG_LEVEL=info
 # Unset uses HINDSIGHT_API_RETAIN_CHUNK_SIZE as the structured-chunk limit.
 # HINDSIGHT_API_RETAIN_STRUCTURED_CHUNK_SIZE=
 
+# When true, a retain operation that hit any fact-extraction errors is marked
+# 'failed' (not 'completed'), surfacing silently-dropped facts. Default false.
+# HINDSIGHT_API_FAIL_ON_EXTRACTION_ERRORS=false
+
 # Dry-run extraction preview endpoint (POST /memories/dry-run-extract). Enabled by default; it makes
 # a real LLM call but stores nothing. Set to false to remove the endpoint (returns 404).
 # HINDSIGHT_API_ENABLE_DRY_RUN_EXTRACT=true

hindsight-api-slim/hindsight_api/config.py

@@ -665,6 +665,9 @@ def _resolve_operation_temperature(operation_env: str, default: float) -> float
 ENV_AUDIT_LOG_ACTIONS = "HINDSIGHT_API_AUDIT_LOG_ACTIONS"
 ENV_AUDIT_LOG_RETENTION_DAYS = "HINDSIGHT_API_AUDIT_LOG_RETENTION_DAYS"
 
+# Retain reliability settings
+ENV_FAIL_ON_EXTRACTION_ERRORS = "HINDSIGHT_API_FAIL_ON_EXTRACTION_ERRORS"
+
 # LLM request tracing settings
 ENV_LLM_TRACE_ENABLED = "HINDSIGHT_API_LLM_TRACE_ENABLED"
 ENV_LLM_TRACE_SCOPES = "HINDSIGHT_API_LLM_TRACE_SCOPES"
@@ -1105,6 +1108,11 @@ def _parse_strategy_boosts(raw: str | None) -> dict[str, str]:
 DEFAULT_AUDIT_LOG_ACTIONS = ""  # Empty = audit all eligible actions
 DEFAULT_AUDIT_LOG_RETENTION_DAYS = -1  # -1 = keep forever
 
+# Retain reliability defaults
+DEFAULT_FAIL_ON_EXTRACTION_ERRORS = (
+    False  # Preserve existing behavior: retain completes even if some chunks fail extraction
+)
+
 # LLM request tracing defaults
 DEFAULT_LLM_TRACE_ENABLED = True  # Enabled by default
 DEFAULT_LLM_TRACE_SCOPES = ""  # Empty = trace all call scopes
@@ -1958,6 +1966,11 @@ class HindsightConfig:
     audit_log_actions: list[str]  # Allowlist of action types (empty = all)
     audit_log_retention_days: int  # -1 = keep forever, >0 = delete after N days
 
+    # Retain reliability configuration (static - server-level only)
+    # When True, a retain operation that accumulated any fact-extraction errors is
+    # marked 'failed' instead of 'completed', surfacing silent fact loss to clients.
+    fail_on_extraction_errors: bool
+
     # LLM request tracing configuration (static - server-level only)
     llm_trace_enabled: bool  # Master switch for per-bank LLM request tracing
     llm_trace_scopes: list[str]  # Allowlist of call scopes to trace (empty = all)
@@ -3048,6 +3061,11 @@ def from_env(cls) -> "HindsightConfig":
             audit_log_retention_days=int(
                 os.getenv(ENV_AUDIT_LOG_RETENTION_DAYS, str(DEFAULT_AUDIT_LOG_RETENTION_DAYS))
             ),
+            # Retain reliability configuration (static, server-level only)
+            fail_on_extraction_errors=os.getenv(
+                ENV_FAIL_ON_EXTRACTION_ERRORS, str(DEFAULT_FAIL_ON_EXTRACTION_ERRORS)
+            ).lower()
+            == "true",
             # LLM request tracing configuration (static, server-level only)
             llm_trace_enabled=os.getenv(ENV_LLM_TRACE_ENABLED, str(DEFAULT_LLM_TRACE_ENABLED)).lower() == "true",
             llm_trace_scopes=[

hindsight-api-slim/hindsight_api/engine/memory_engine.py

@@ -2374,11 +2374,57 @@ async def _mark_operation_completed(self, operation_id: str):
 
         Also checks if this is a child operation and updates the parent if all siblings are done.
         Uses a single transaction to avoid race conditions when multiple children complete simultaneously.
+
+        Opt-in escape hatch: when ``HINDSIGHT_API_FAIL_ON_EXTRACTION_ERRORS`` is set and the
+        operation's ``result_metadata`` recorded a non-zero ``extraction_errors_count`` (written by
+        ``_write_retain_outcome_metadata`` before this call), the operation is marked ``failed``
+        instead of ``completed``. This surfaces silently-dropped facts as a hard failure rather
+        than a clean success. Default is off, so existing behavior is unchanged (see issue #2700).
         """
         try:
             backend = await self._get_backend()
             async with acquire_with_retry(backend) as conn:
                 async with conn.transaction():
+                    # Read the accumulated extraction-error count (persisted by
+                    # _write_retain_outcome_metadata) to decide the terminal status.
+                    meta_row = await conn.fetchrow(
+                        f"SELECT result_metadata FROM {fq_table('async_operations')} WHERE operation_id = $1",
+                        uuid.UUID(operation_id),
+                    )
+                    extraction_errors_count = 0
+                    if meta_row is not None:
+                        metadata = conn.parse_json(meta_row["result_metadata"]) or {}
+                        extraction_errors_count = int(metadata.get("extraction_errors_count") or 0)
+
+                    fail_on_errors = get_config().fail_on_extraction_errors
+                    if fail_on_errors and extraction_errors_count > 0:
+                        error_message = (
+                            f"Retain completed with {extraction_errors_count} fact extraction error(s); "
+                            "marked failed because HINDSIGHT_API_FAIL_ON_EXTRACTION_ERRORS is enabled. "
+                            "See result_metadata.extraction_errors_sample for details."
+                        )
+                        row = await conn.fetchrow(
+                            f"""
+                            UPDATE {fq_table("async_operations")}
+                            SET status = 'failed', error_message = $2, updated_at = NOW(), completed_at = NOW()
+                            WHERE operation_id = $1
+                            RETURNING operation_id
+                            """,
+                            uuid.UUID(operation_id),
+                            error_message,
+                        )
+                        if row is None:
+                            logger.info(
+                                f"Operation {operation_id} no longer exists (bank deleted), skipping mark-failed"
+                            )
+                            return
+                        logger.warning(
+                            f"Marked async operation as failed due to {extraction_errors_count} "
+                            f"extraction error(s): {operation_id}"
+                        )
+                        await self._maybe_update_parent_operation(operation_id, conn)
+                        return
+
                     # Mark this operation as completed
                     row = await conn.fetchrow(
                         f"""

hindsight-api-slim/tests/test_async_batch_retain.py

@@ -2,6 +2,7 @@
 
 import asyncio
 import json
+import os
 import uuid
 
 import pytest
@@ -515,6 +516,105 @@ async def empty_extract_facts_from_contents(
     assert "extraction_errors_sample" not in parent["result_metadata"]
 
 
+async def _seed_retain_op_with_errors(pool, bank_id: str, error_count: int) -> uuid.UUID:
+    """Insert a pending retain operation whose outcome metadata records extraction errors."""
+    operation_id = uuid.uuid4()
+    await pool.execute(
+        """
+        INSERT INTO async_operations (operation_id, bank_id, operation_type, result_metadata, status)
+        VALUES ($1, $2, $3, $4, $5)
+        """,
+        operation_id,
+        bank_id,
+        "retain",
+        json.dumps(
+            {
+                "unit_ids_count": 3,
+                "extraction_errors_count": error_count,
+                "extraction_errors_sample": ["chunk 2 failed to parse"],
+            }
+        ),
+        "pending",
+    )
+    return operation_id
+
+
+async def _op_row(pool, operation_id: uuid.UUID):
+    return await pool.fetchrow(
+        "SELECT status, error_message FROM async_operations WHERE operation_id = $1",
+        operation_id,
+    )
+
+
+@pytest.mark.asyncio
+async def test_completion_marks_failed_when_flag_on_and_errors_present(memory):
+    """With the escape hatch on, a retain that dropped facts ends 'failed' (issue #2700)."""
+    from hindsight_api.config import ENV_FAIL_ON_EXTRACTION_ERRORS, clear_config_cache
+
+    bank_id = "test_fail_on_extraction_errors_on"
+    pool = await memory._get_pool()
+    await _ensure_bank(pool, bank_id)
+    operation_id = await _seed_retain_op_with_errors(pool, bank_id, error_count=2)
+
+    os.environ[ENV_FAIL_ON_EXTRACTION_ERRORS] = "true"
+    clear_config_cache()
+    try:
+        await memory._mark_operation_completed(str(operation_id))
+    finally:
+        del os.environ[ENV_FAIL_ON_EXTRACTION_ERRORS]
+        clear_config_cache()
+
+    row = await _op_row(pool, operation_id)
+    assert row["status"] == "failed"
+    assert row["error_message"] is not None
+    assert "2" in row["error_message"]
+    assert "extraction error" in row["error_message"].lower()
+
+
+@pytest.mark.asyncio
+async def test_completion_stays_completed_when_flag_off(memory):
+    """Default behavior is preserved: extraction errors still complete the operation."""
+    from hindsight_api.config import ENV_FAIL_ON_EXTRACTION_ERRORS, clear_config_cache
+
+    bank_id = "test_fail_on_extraction_errors_off"
+    pool = await memory._get_pool()
+    await _ensure_bank(pool, bank_id)
+    operation_id = await _seed_retain_op_with_errors(pool, bank_id, error_count=2)
+
+    os.environ.pop(ENV_FAIL_ON_EXTRACTION_ERRORS, None)
+    clear_config_cache()
+    try:
+        await memory._mark_operation_completed(str(operation_id))
+    finally:
+        clear_config_cache()
+
+    row = await _op_row(pool, operation_id)
+    assert row["status"] == "completed"
+    assert row["error_message"] is None
+
+
+@pytest.mark.asyncio
+async def test_completion_completed_when_flag_on_but_no_errors(memory):
+    """The flag only fails operations that actually accumulated extraction errors."""
+    from hindsight_api.config import ENV_FAIL_ON_EXTRACTION_ERRORS, clear_config_cache
+
+    bank_id = "test_fail_on_extraction_errors_none"
+    pool = await memory._get_pool()
+    await _ensure_bank(pool, bank_id)
+    operation_id = await _seed_retain_op_with_errors(pool, bank_id, error_count=0)
+
+    os.environ[ENV_FAIL_ON_EXTRACTION_ERRORS] = "true"
+    clear_config_cache()
+    try:
+        await memory._mark_operation_completed(str(operation_id))
+    finally:
+        del os.environ[ENV_FAIL_ON_EXTRACTION_ERRORS]
+        clear_config_cache()
+
+    row = await _op_row(pool, operation_id)
+    assert row["status"] == "completed"
+
+
 @pytest.mark.asyncio
 async def test_retain_records_user_provided_document_ids(memory, request_context):
     """User-supplied document_ids land in child op result_metadata.document_ids."""

hindsight-api-slim/tests/test_config_validation.py

@@ -122,6 +122,30 @@ def test_retain_structured_chunk_size_reads_from_env():
     assert config.retain_structured_chunk_size == 9000
 
 
+def test_fail_on_extraction_errors_defaults_to_false(monkeypatch):
+    """Silent-success behavior is preserved by default (issue #2700)."""
+    from hindsight_api.config import ENV_FAIL_ON_EXTRACTION_ERRORS, HindsightConfig
+
+    monkeypatch.delenv(ENV_FAIL_ON_EXTRACTION_ERRORS, raising=False)
+    monkeypatch.setenv("HINDSIGHT_API_LLM_PROVIDER", "mock")
+
+    config = HindsightConfig.from_env()
+
+    assert config.fail_on_extraction_errors is False
+
+
+def test_fail_on_extraction_errors_reads_true_from_env(monkeypatch):
+    """The opt-in escape hatch parses truthy values from the environment."""
+    from hindsight_api.config import ENV_FAIL_ON_EXTRACTION_ERRORS, HindsightConfig
+
+    monkeypatch.setenv(ENV_FAIL_ON_EXTRACTION_ERRORS, "true")
+    monkeypatch.setenv("HINDSIGHT_API_LLM_PROVIDER", "mock")
+
+    config = HindsightConfig.from_env()
+
+    assert config.fail_on_extraction_errors is True
+
+
 def test_llm_ollama_num_ctx_defaults_to_none(monkeypatch):
     """Unset Ollama num_ctx override lets Ollama use its model/server default."""
     from hindsight_api.config import ENV_LLM_OLLAMA_NUM_CTX, HindsightConfig

hindsight-docs/docs/developer/configuration.md

@@ -1106,6 +1106,7 @@ Controls the retain (memory ingestion) pipeline.
 | `HINDSIGHT_API_RETAIN_DEFAULT_STRATEGY` | Default retain strategy name. When set, all retain calls without an explicit `strategy` parameter use this strategy. | - |
 | `HINDSIGHT_API_RETAIN_BATCH_POLL_INTERVAL_SECONDS` | Batch API polling interval in seconds | `60` |
 | `HINDSIGHT_API_STORE_DOCUMENT_TEXT` | Persist the raw source text alongside extracted memories. Set to `false` to skip storing it. Static, server-level. | `true` |
+| `HINDSIGHT_API_FAIL_ON_EXTRACTION_ERRORS` | When `true`, a retain operation that accumulated any fact-extraction errors is marked `failed` (with an error message including the count) instead of `completed`, so silently-dropped facts surface as a hard failure. Default preserves existing behavior. Static, server-level. | `false` |
 
 > **Batch-capable providers.** `HINDSIGHT_API_RETAIN_BATCH_ENABLED=true` only works with a retain LLM provider that implements a batch API: `openai`, `groq`, `gemini`, and `fireworks`. Batch always requires async retain (`async=true`); a sync retain with batch enabled errors. Other providers fail fast at startup.
 >

hindsight-embed/hindsight_embed/env.example

@@ -89,6 +89,10 @@ HINDSIGHT_API_LOG_LEVEL=info
 # Unset uses HINDSIGHT_API_RETAIN_CHUNK_SIZE as the structured-chunk limit.
 # HINDSIGHT_API_RETAIN_STRUCTURED_CHUNK_SIZE=
 
+# When true, a retain operation that hit any fact-extraction errors is marked
+# 'failed' (not 'completed'), surfacing silently-dropped facts. Default false.
+# HINDSIGHT_API_FAIL_ON_EXTRACTION_ERRORS=false
+
 # Dry-run extraction preview endpoint (POST /memories/dry-run-extract). Enabled by default; it makes
 # a real LLM call but stores nothing. Set to false to remove the endpoint (returns 404).
 # HINDSIGHT_API_ENABLE_DRY_RUN_EXTRACT=true

skills/hindsight-docs/references/developer/configuration.md

@@ -1106,6 +1106,7 @@ Controls the retain (memory ingestion) pipeline.
 | `HINDSIGHT_API_RETAIN_DEFAULT_STRATEGY` | Default retain strategy name. When set, all retain calls without an explicit `strategy` parameter use this strategy. | - |
 | `HINDSIGHT_API_RETAIN_BATCH_POLL_INTERVAL_SECONDS` | Batch API polling interval in seconds | `60` |
 | `HINDSIGHT_API_STORE_DOCUMENT_TEXT` | Persist the raw source text alongside extracted memories. Set to `false` to skip storing it. Static, server-level. | `true` |
+| `HINDSIGHT_API_FAIL_ON_EXTRACTION_ERRORS` | When `true`, a retain operation that accumulated any fact-extraction errors is marked `failed` (with an error message including the count) instead of `completed`, so silently-dropped facts surface as a hard failure. Default preserves existing behavior. Static, server-level. | `false` |
 
 > **Batch-capable providers.** `HINDSIGHT_API_RETAIN_BATCH_ENABLED=true` only works with a retain LLM provider that implements a batch API: `openai`, `groq`, `gemini`, and `fireworks`. Batch always requires async retain (`async=true`); a sync retain with batch enabled errors. Other providers fail fast at startup.
 >