feat(consolidation): warn when a run dead-letters a large share of its work

hindsight-api-slim/hindsight_api/api/http.py

@@ -2732,6 +2732,13 @@ class BankTemplateConfig(BaseModel):
     consolidation_max_memories_per_round: int | None = Field(
         default=None, description="Max memory units fed into a single consolidation round"
     )
+    consolidation_dead_letter_warn_fraction: float | None = Field(
+        default=None,
+        description=(
+            "Share of a consolidation run's attempted memories that must dead-letter "
+            "before an end-of-run warning is logged (0 disables)"
+        ),
+    )
     consolidation_llm_parallelism: int | None = Field(
         default=None, description="Number of consolidation LLM batches processed concurrently"
     )

hindsight-api-slim/hindsight_api/config.py

@@ -664,6 +664,7 @@ def _parse_boolean_env(env_name: str, default: bool) -> bool:
 ENV_ENABLE_AUTO_CONSOLIDATION = "HINDSIGHT_API_ENABLE_AUTO_CONSOLIDATION"
 ENV_CONSOLIDATION_BATCH_SIZE = "HINDSIGHT_API_CONSOLIDATION_BATCH_SIZE"
 ENV_CONSOLIDATION_MAX_MEMORIES_PER_ROUND = "HINDSIGHT_API_CONSOLIDATION_MAX_MEMORIES_PER_ROUND"
+ENV_CONSOLIDATION_DEAD_LETTER_WARN_FRACTION = "HINDSIGHT_API_CONSOLIDATION_DEAD_LETTER_WARN_FRACTION"
 ENV_CONSOLIDATION_LLM_BATCH_SIZE = "HINDSIGHT_API_CONSOLIDATION_LLM_BATCH_SIZE"
 ENV_CONSOLIDATION_DEDUP_THRESHOLD = "HINDSIGHT_API_CONSOLIDATION_DEDUP_THRESHOLD"
 ENV_CONSOLIDATION_LLM_PARALLELISM = "HINDSIGHT_API_CONSOLIDATION_LLM_PARALLELISM"
@@ -1263,6 +1264,9 @@ def _parse_strategy_boosts(raw: str | None) -> dict[str, str]:
 DEFAULT_CONSOLIDATION_MAX_MEMORIES_PER_ROUND = (
     100  # Max memories per consolidation round (0 = unlimited). Limits how long one bank holds a worker slot.
 )
+# Warn at end of run when this share of a run's attempted memories dead-lettered.
+# 0 disables the warning; 1.0 warns only when every attempted memory failed.
+DEFAULT_CONSOLIDATION_DEAD_LETTER_WARN_FRACTION = 0.5
 DEFAULT_CONSOLIDATION_LLM_BATCH_SIZE = 8  # Facts per LLM call (1 = no batching; >1 = batch mode)
 # Cosine >= this between a newly-created or freshly-updated observation and an existing one
 # triggers a focused 1-by-1 LLM "merge or keep" pass (the LLM reads both, so numbers/negation/
@@ -2514,6 +2518,7 @@ class HindsightConfig:
     consolidation_batch_size: int
     consolidation_dedup_threshold: float
     consolidation_max_memories_per_round: int
+    consolidation_dead_letter_warn_fraction: float
     consolidation_llm_batch_size: int
     consolidation_llm_parallelism: int
     consolidation_max_tokens: int
@@ -2811,6 +2816,7 @@ class HindsightConfig:
         "consolidation_llm_batch_size",
         "consolidation_llm_parallelism",
         "consolidation_max_memories_per_round",
+        "consolidation_dead_letter_warn_fraction",
         "consolidation_source_facts_max_tokens",
         "consolidation_source_facts_max_tokens_per_observation",
         "observations_mission",
@@ -3795,6 +3801,12 @@ def from_env(cls) -> "HindsightConfig":
                     str(DEFAULT_CONSOLIDATION_MAX_MEMORIES_PER_ROUND),
                 )
             ),
+            consolidation_dead_letter_warn_fraction=float(
+                os.getenv(
+                    ENV_CONSOLIDATION_DEAD_LETTER_WARN_FRACTION,
+                    str(DEFAULT_CONSOLIDATION_DEAD_LETTER_WARN_FRACTION),
+                )
+            ),
             consolidation_dedup_threshold=float(
                 os.getenv(ENV_CONSOLIDATION_DEDUP_THRESHOLD, str(DEFAULT_CONSOLIDATION_DEDUP_THRESHOLD))
             ),

hindsight-api-slim/hindsight_api/engine/consolidation/consolidator.py

@@ -31,7 +31,7 @@
 import asyncpg
 from pydantic import BaseModel, field_validator
 
-from ...config import get_config
+from ...config import ENV_CONSOLIDATION_DEAD_LETTER_WARN_FRACTION, get_config
 from ...worker.stage import set_stage
 from ..db import DatabaseBackend
 from ..db_utils import acquire_with_retry
@@ -1097,6 +1097,61 @@ async def _count_unconsolidated_rows(
     )
 
 
+#: A run must dead-letter at least this many memories before the end-of-run warning
+#: fires, regardless of the configured fraction. Without a floor, a bank consolidating
+#: one or two memories at a time would warn on every isolated failure, and a warning
+#: that fires constantly is one operators learn to scroll past -- the same end state as
+#: not having it, but with false confidence that the case is covered.
+DEAD_LETTER_WARN_MIN_MEMORIES = 3
+
+
+def dead_letter_warning(bank_id: str, processed: int, failed: int, warn_fraction: float) -> str | None:
+    """Return an operator-facing warning if this run dead-lettered a large share of its work.
+
+    Memories that exhaust consolidation retries are stamped ``consolidation_failed_at``
+    and are then skipped by every later run. That is the correct conservative behaviour
+    -- the engine cannot know whether the provider refused this content or was simply
+    unavailable -- but it is silent: the rows leave the pending set and nothing says so.
+    A provider outage or a quota window can therefore park a whole batch indefinitely,
+    and in practice it is noticed weeks later, if at all.
+
+    This deliberately infers nothing about WHY the failures happened. It reads two
+    counters the job already keeps and compares them to a threshold. There is no
+    inspection of provider exception types, status codes or message text, so it cannot
+    go stale as providers change their error vocabulary, and it behaves identically for
+    a failure mode nobody has seen yet.
+
+    Returns None when there is nothing to say, so the caller logs only on a real signal
+    and the decision stays testable without capturing log output.
+    """
+    # Counters only, and both must be sane: a negative one means the caller's
+    # accounting is broken, and a broken counter must not be reported as a
+    # dead-letter event (processed=-5, failed=10 would otherwise read as "200%").
+    if processed < 0 or failed <= 0:
+        return None
+    attempted = processed + failed
+    if attempted <= 0:
+        return None
+    # A non-positive fraction disables the warning outright; operators who have accepted
+    # a lossy provider should be able to silence it without patching.
+    if warn_fraction <= 0:
+        return None
+    if failed < DEAD_LETTER_WARN_MIN_MEMORIES:
+        return None
+    fraction = failed / attempted
+    if fraction < warn_fraction:
+        return None
+    return (
+        f"[CONSOLIDATION] bank={bank_id} dead-lettered {failed}/{attempted} memories "
+        f"({fraction:.0%}) in this run. They are stamped consolidation_failed_at and "
+        f"will NOT be retried by later runs. If this was a provider outage or quota "
+        f"window rather than unusable content, clear the stamps with: "
+        f"POST /v1/default/banks/{bank_id}/consolidation/recover "
+        f"(CLI: hindsight bank consolidation-recover {bank_id}). "
+        f"Set {ENV_CONSOLIDATION_DEAD_LETTER_WARN_FRACTION}=0 to silence this."
+    )
+
+
 def _as_op_uuid(operation_id: str | uuid.UUID) -> uuid.UUID:
     return uuid.UUID(operation_id) if isinstance(operation_id, str) else operation_id
 
@@ -1835,6 +1890,29 @@ def _fmt(key: str) -> str:
 
     perf.flush()
 
+    # Surface a run that parked a large share of its work. Emitted once per run, after
+    # the totals are final, so it reports the run's actual outcome rather than a
+    # mid-flight batch that later batches may recover from.
+    #
+    # The two counters mean what the warning says they mean: ``memories_failed`` is
+    # incremented only for an ``action == "failed"`` result, which is emitted at the
+    # one site that marks a memory with ``consolidation_failed_at`` -- after adaptive
+    # splitting has already retried it down to a single-memory batch. So it counts
+    # terminal dead-letter stamps from THIS run, not transient batch errors that a
+    # later split recovered, and ``memories_processed`` is its success counterpart.
+    # Read from the bank-resolved config, not get_config(): this knob is per-bank
+    # configurable and exportable in a bank template, so reading the process-global
+    # would let a bank-level value store, export and import cleanly while never taking
+    # effect. Same object the sibling consolidation knobs above are read from.
+    warning = dead_letter_warning(
+        bank_id,
+        stats["memories_processed"],
+        stats["memories_failed"],
+        config.consolidation_dead_letter_warn_fraction,
+    )
+    if warning:
+        logger.warning(warning)
+
     return {"status": "completed", "bank_id": bank_id, **stats}
 
 

hindsight-api-slim/tests/test_bank_template_full_roundtrip.py

@@ -111,6 +111,7 @@
     "store_document_text": False,
     "enable_auto_consolidation": False,
     "consolidation_max_memories_per_round": 42,
+    "consolidation_dead_letter_warn_fraction": 0.25,
     "consolidation_llm_parallelism": 3,
     "recall_include_chunks": True,
     "recall_max_tokens": 9000,

hindsight-api-slim/tests/test_consolidation_dead_letter_warning.py

@@ -0,0 +1,226 @@
+"""Tests for the end-of-run dead-letter warning.
+
+The warning exists because ``consolidation_failed_at`` is silent: a stamped memory
+leaves the pending set and no later run touches it, so a provider outage can park a
+whole batch with nothing in the logs to prompt the operator to call
+``/consolidation/recover``.
+
+The decision logic is a pure function, so these tests need no database and no LLM. They
+pin three things: that it fires when a run really did park a large share of its work,
+that it stays quiet on the shapes that would otherwise make it noise (which is what
+decides whether operators keep reading it), and that the message actually tells the
+operator what to run.
+"""
+
+import pytest
+
+from hindsight_api.config import ENV_CONSOLIDATION_DEAD_LETTER_WARN_FRACTION
+from hindsight_api.engine.consolidation.consolidator import (
+    DEAD_LETTER_WARN_MIN_MEMORIES,
+    dead_letter_warning,
+)
+
+BANK = "bank-under-test"
+HALF = 0.5
+
+
+class TestFires:
+    """Shapes that must produce a warning."""
+
+    def test_whole_batch_parked(self):
+        # The case the warning exists for: a quota window takes out an entire run.
+        msg = dead_letter_warning(BANK, processed=0, failed=40, warn_fraction=HALF)
+        assert msg is not None
+        assert "40/40" in msg
+        assert "100%" in msg
+
+    def test_majority_parked(self):
+        msg = dead_letter_warning(BANK, processed=10, failed=30, warn_fraction=HALF)
+        assert msg is not None
+        assert "30/40" in msg
+        assert "75%" in msg
+
+    def test_exactly_at_threshold_fires(self):
+        # Boundary: fraction == threshold must fire, not fall through the >= edge.
+        msg = dead_letter_warning(BANK, processed=20, failed=20, warn_fraction=HALF)
+        assert msg is not None
+        assert "50%" in msg
+
+    def test_exactly_at_the_floor_fires(self):
+        # The floor is inclusive: failed == DEAD_LETTER_WARN_MIN_MEMORIES must warn.
+        # Without this, tightening `failed < FLOOR` to `failed <= FLOOR` (requiring 4)
+        # would pass the whole suite -- the quiet-side tests only cover 1 and 2.
+        msg = dead_letter_warning(BANK, processed=0, failed=DEAD_LETTER_WARN_MIN_MEMORIES, warn_fraction=HALF)
+        assert msg is not None
+        assert "3/3" in msg
+
+    def test_low_threshold_catches_a_small_share(self):
+        # An operator who wants to hear about any material loss can lower the bar; the
+        # absolute floor still applies, so failed=4 is above DEAD_LETTER_WARN_MIN_MEMORIES.
+        msg = dead_letter_warning(BANK, processed=96, failed=4, warn_fraction=0.01)
+        assert msg is not None
+        assert "4/100" in msg
+
+
+class TestStaysQuiet:
+    """Shapes that must NOT warn.
+
+    These matter more than the firing cases. A guard that cries wolf gets ignored, and
+    an ignored guard is indistinguishable from a missing one except that it also
+    supplies false confidence.
+    """
+
+    def test_clean_run(self):
+        assert dead_letter_warning(BANK, processed=100, failed=0, warn_fraction=HALF) is None
+
+    def test_empty_run(self):
+        # No work attempted at all: 0/0 must not divide, and must not warn.
+        assert dead_letter_warning(BANK, processed=0, failed=0, warn_fraction=HALF) is None
+
+    def test_below_the_fraction(self):
+        assert dead_letter_warning(BANK, processed=90, failed=10, warn_fraction=HALF) is None
+
+    def test_the_floor_is_three(self):
+        # Pinned as a literal so the cases below cannot drift with the constant. If this
+        # value is changed deliberately, this test is the place that records it.
+        assert DEAD_LETTER_WARN_MIN_MEMORIES == 3
+
+    @pytest.mark.parametrize("failed", [1, 2])
+    def test_below_the_absolute_floor_even_at_100_percent(self, failed):
+        # A bank consolidating one or two memories at a time hits 100% constantly, so
+        # the floor is the whole reason this warning stays readable.
+        #
+        # These inputs are LITERAL on purpose. They were originally
+        # `range(1, DEAD_LETTER_WARN_MIN_MEMORIES)`, which derives the test's own inputs
+        # from the constant under test: setting the floor to 0 emptied the range, and
+        # the test SKIPPED instead of failing. Caught by planting exactly that
+        # regression -- a guard whose trigger disappears along with the thing it guards
+        # is not a guard.
+        assert dead_letter_warning(BANK, processed=0, failed=failed, warn_fraction=HALF) is None
+
+    def test_zero_fraction_disables(self):
+        assert dead_letter_warning(BANK, processed=0, failed=1000, warn_fraction=0.0) is None
+
+    def test_negative_fraction_disables(self):
+        assert dead_letter_warning(BANK, processed=0, failed=1000, warn_fraction=-1.0) is None
+
+    def test_negative_failed_count_is_not_a_signal(self):
+        # Defensive: a counter regression must not be reported as a dead-letter event.
+        assert dead_letter_warning(BANK, processed=10, failed=-5, warn_fraction=HALF) is None
+
+    def test_negative_processed_count_is_not_a_signal(self):
+        # The asymmetric case: guarding only `failed` leaves processed=-5, failed=10
+        # summing to attempted=5, so the fraction reads 2.0 and the message would claim
+        # "10/5 (200%)". A broken counter must produce silence, not a nonsense number.
+        assert dead_letter_warning(BANK, processed=-5, failed=10, warn_fraction=HALF) is None
+        assert dead_letter_warning(BANK, processed=-1, failed=3, warn_fraction=HALF) is None
+
+
+class TestMessageIsActionable:
+    """The message has to be usable on its own, in a log, by someone paged at 3am."""
+
+    def test_names_the_recovery_endpoint_and_bank(self):
+        msg = dead_letter_warning(BANK, processed=0, failed=25, warn_fraction=HALF)
+        assert msg is not None
+        assert f"/v1/default/banks/{BANK}/consolidation/recover" in msg
+        assert f"hindsight bank consolidation-recover {BANK}" in msg
+
+    def test_says_the_rows_will_not_retry(self):
+        # Without this the reader cannot tell whether the situation is self-healing.
+        msg = dead_letter_warning(BANK, processed=0, failed=25, warn_fraction=HALF)
+        assert msg is not None
+        assert "consolidation_failed_at" in msg
+        assert "NOT be retried" in msg
+
+    def test_says_how_to_silence_it(self):
+        msg = dead_letter_warning(BANK, processed=0, failed=25, warn_fraction=HALF)
+        assert msg is not None
+        assert ENV_CONSOLIDATION_DEAD_LETTER_WARN_FRACTION in msg
+
+
+class TestClassifiesNothing:
+    """The reason #3309 was rejected: no provider-shaped surface may creep back in.
+
+    The rejected approach read provider exception types, then status codes, then message
+    substrings, so it needed updating as every provider's wire vocabulary changed. This
+    replacement must depend on counters only -- a property worth pinning, because the
+    natural "improvement" to this function is to start explaining WHY the failures
+    happened.
+    """
+
+    def test_signature_takes_only_counters(self):
+        import inspect
+
+        params = set(inspect.signature(dead_letter_warning).parameters)
+        assert params == {"bank_id", "processed", "failed", "warn_fraction"}
+
+    def test_executable_code_mentions_no_provider_error_vocabulary(self):
+        import ast
+        import inspect
+        import textwrap
+
+        from hindsight_api.engine.consolidation import consolidator
+
+        fn = ast.parse(textwrap.dedent(inspect.getsource(consolidator.dead_letter_warning))).body[0]
+        # Scan the BODY only. The docstring explains that no classification happens, so
+        # including prose would fail on the very sentence documenting the property under
+        # test -- and would then be "fixed" by deleting the explanation, which is worse
+        # than the test. ast.unparse also drops comments, for the same reason.
+        first = fn.body[0] if fn.body else None
+        has_docstring = (
+            isinstance(first, ast.Expr) and isinstance(first.value, ast.Constant) and isinstance(first.value.value, str)
+        )
+        body = fn.body[1:] if has_docstring else fn.body
+        assert body, "function body is empty -- this check would pass vacuously"
+        code = "\n".join(ast.unparse(node) for node in body).lower()
+        for token in ("rate_limit", "ratelimited", "429", "quota_exceeded", "status_code", "exception"):
+            assert token not in code, f"dead_letter_warning must not reason about {token!r}"
+
+
+class TestCallSiteWiring:
+    """The pure function is well covered; the glue that calls it is the weak seam.
+
+    A correct decision that is never logged, or is logged from the process-global
+    config instead of the bank's, is invisible in exactly the way this warning exists
+    to prevent -- so the wiring is pinned rather than assumed.
+    """
+
+    def test_job_reads_the_bank_resolved_config_not_the_process_global(self):
+        # `_run_consolidation_job` receives `config` already resolved for the bank.
+        # Reading `get_config()` here instead would let a per-bank value store, export
+        # and import cleanly while never taking effect -- a silent no-op, and the same
+        # invisible-failure shape that got the classification approach rejected.
+        import ast
+        import inspect
+        import textwrap
+
+        from hindsight_api.engine.consolidation import consolidator
+
+        src = textwrap.dedent(inspect.getsource(consolidator._run_consolidation_job))
+        call = next(
+            node
+            for node in ast.walk(ast.parse(src))
+            if isinstance(node, ast.Call) and isinstance(node.func, ast.Name) and node.func.id == "dead_letter_warning"
+        )
+        rendered = [ast.unparse(arg) for arg in call.args]
+        assert "config.consolidation_dead_letter_warn_fraction" in rendered, rendered
+        assert not any("get_config()" in arg for arg in rendered), rendered
+
+    def test_the_decision_is_logged_as_a_warning(self):
+        # Pins that a non-None decision actually reaches the log. The function could be
+        # perfect and the caller could drop it on the floor.
+        import ast
+        import inspect
+        import textwrap
+
+        from hindsight_api.engine.consolidation import consolidator
+
+        src = textwrap.dedent(inspect.getsource(consolidator._run_consolidation_job))
+        guarded_logs = [
+            node
+            for node in ast.walk(ast.parse(src))
+            if isinstance(node, ast.If)
+            and ast.unparse(node.test) == "warning"
+            and "logger.warning(warning)" in ast.unparse(node)
+        ]
+        assert guarded_logs, "the dead-letter decision is computed but never logged"