fix(worker): stop wedged retains from holding worker slots forever

.env.example

@@ -112,6 +112,12 @@ HINDSIGHT_API_LOG_LEVEL=info
 # 'failed' (not 'completed'), surfacing silently-dropped facts. Default false.
 # HINDSIGHT_API_FAIL_ON_EXTRACTION_ERRORS=false
 
+# Wall-clock ceiling (seconds) for one retain task in the worker. A retain that
+# blocks indefinitely is cancelled and marked 'failed' — and so becomes
+# retryable — instead of holding its worker slot until the process restarts.
+# Set well above your slowest healthy retain; 0 disables. Default 3600.
+# HINDSIGHT_API_RETAIN_WALL_TIMEOUT=3600
+
 # 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

@@ -690,6 +690,7 @@ def _parse_boolean_env(env_name: str, default: bool) -> bool:
 }
 ENV_WORKER_CONSOLIDATION_BANK_PRIORITY = "HINDSIGHT_API_WORKER_CONSOLIDATION_BANK_PRIORITY"
 ENV_RETAIN_MAX_CONCURRENT = "HINDSIGHT_API_RETAIN_MAX_CONCURRENT"
+ENV_RETAIN_WALL_TIMEOUT = "HINDSIGHT_API_RETAIN_WALL_TIMEOUT"
 
 # Reflect agent settings
 ENV_REFLECT_MAX_ITERATIONS = "HINDSIGHT_API_REFLECT_MAX_ITERATIONS"
@@ -1168,6 +1169,14 @@ def _parse_strategy_boosts(raw: str | None) -> dict[str, str]:
 DEFAULT_OPERATION_RETENTION_DAYS = 0
 DEFAULT_OPERATION_CLEANUP_BATCH_SIZE = 1000
 DEFAULT_RETAIN_MAX_CONCURRENT = 4  # Max concurrent retain DB phases (HNSW reads + writes). Limits I/O contention.
+# Wall-clock ceiling for one retain task in the worker (0 disables). This is a
+# deadlock/wedge backstop, not a latency target: a retain that blocks forever on
+# a lock, an LLM permit or a queue put would otherwise hold its worker slot until
+# the process restarts, and 'processing' is neither retryable nor cancellable
+# through the API. Set well above any healthy retain so it only ever fires on a
+# genuine wedge — the per-attempt LLM timeout and the retry budget already bound
+# the normal slow path.
+DEFAULT_RETAIN_WALL_TIMEOUT = 3600  # seconds (1 hour)
 
 # Reflect agent settings
 DEFAULT_REFLECT_MAX_ITERATIONS = 10  # Max tool call iterations before forcing response
@@ -2104,6 +2113,7 @@ class HindsightConfig:
     operation_retention_days: int
     operation_cleanup_batch_size: int
     retain_max_concurrent: int
+    retain_wall_timeout: int
 
     # Reflect agent settings
     reflect_max_iterations: int
@@ -3215,6 +3225,7 @@ def from_env(cls) -> "HindsightConfig":
                 DEFAULT_OPERATION_CLEANUP_BATCH_SIZE,
             ),
             retain_max_concurrent=int(os.getenv(ENV_RETAIN_MAX_CONCURRENT, str(DEFAULT_RETAIN_MAX_CONCURRENT))),
+            retain_wall_timeout=int(os.getenv(ENV_RETAIN_WALL_TIMEOUT, str(DEFAULT_RETAIN_WALL_TIMEOUT))),
             # Reflect agent settings
             reflect_max_iterations=int(os.getenv(ENV_REFLECT_MAX_ITERATIONS, str(DEFAULT_REFLECT_MAX_ITERATIONS))),
             reflect_prompt_cache_enabled=os.getenv(

hindsight-api-slim/hindsight_api/engine/db/ops_postgresql.py

@@ -253,11 +253,20 @@ async def bulk_insert_entities(
         entity_names: list[str],
         entity_dates: list,
     ) -> dict[str, str]:
+        # ORDER BY LOWER(name) so every concurrent batch inserts in the same order
+        # as the conflict target (bank_id, LOWER(canonical_name)). ON CONFLICT DO
+        # NOTHING takes a ShareLock on the inserting transaction of any speculative
+        # row it collides with, so two batches with overlapping names inserting in
+        # different orders deadlock. The caller already sorts by Python's
+        # ``str.lower()``, which agrees with the index for ASCII but not for every
+        # locale (see the Turkish-İ note in entity_resolver) — ordering in SQL makes
+        # the database's own collation the single arbiter for all writers.
         inserted_rows = await conn.fetch(
             f"""
             INSERT INTO {table} (bank_id, canonical_name, first_seen, last_seen, mention_count)
             SELECT $1, name, COALESCE(event_date, now()), COALESCE(event_date, now()), 0
             FROM unnest($2::text[], $3::timestamptz[]) AS t(name, event_date)
+            ORDER BY LOWER(name)
             ON CONFLICT (bank_id, LOWER(canonical_name))
             DO NOTHING
             RETURNING id, LOWER(canonical_name) AS name_lower

hindsight-api-slim/hindsight_api/engine/db/postgresql.py

@@ -78,6 +78,7 @@ def run_migrations(self, dsn: str, *, schema: str | None = None) -> None:
     def __init__(self) -> None:
         self._pool: asyncpg.Pool | None = None
         self._acquire_warn_threshold_s: float = 1.0
+        self._acquire_timeout_s: float | None = None
 
     async def initialize(
         self,
@@ -93,6 +94,13 @@ async def initialize(
         from ...config import get_config
 
         self._acquire_warn_threshold_s = get_config().db_acquire_warn_threshold_ms / 1000.0
+        # Kept for acquire() below: asyncpg's ``timeout`` create_pool kwarg is a
+        # *connect* kwarg (how long establishing a new connection may take), and
+        # ``Pool.acquire()`` defaults to waiting for a free connection forever.
+        # Passing it here alone made HINDSIGHT_API_DB_ACQUIRE_TIMEOUT a no-op for
+        # the wait it names: a pool-exhaustion stall never surfaced as an error,
+        # it just hung (#3002). 0 restores the unbounded behaviour.
+        self._acquire_timeout_s = acquire_timeout if acquire_timeout > 0 else None
         self._pool = await asyncpg.create_pool(
             dsn,
             min_size=min_size,
@@ -138,15 +146,19 @@ def _pool_stats(self) -> PoolStats | None:
     async def acquire(self) -> AsyncIterator[PostgresConnection]:
         pool = self._ensure_pool()
         async with instrument_acquire(
-            pool.acquire(), pool_stats=self._pool_stats, warn_threshold_s=self._acquire_warn_threshold_s
+            pool.acquire(timeout=self._acquire_timeout_s),
+            pool_stats=self._pool_stats,
+            warn_threshold_s=self._acquire_warn_threshold_s,
         ) as conn:
             yield PostgresConnection(conn)
 
     @asynccontextmanager
     async def transaction(self) -> AsyncIterator[PostgresConnection]:
         pool = self._ensure_pool()
         async with instrument_acquire(
-            pool.acquire(), pool_stats=self._pool_stats, warn_threshold_s=self._acquire_warn_threshold_s
+            pool.acquire(timeout=self._acquire_timeout_s),
+            pool_stats=self._pool_stats,
+            warn_threshold_s=self._acquire_warn_threshold_s,
         ) as conn:
             async with conn.transaction():
                 yield PostgresConnection(conn)

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

@@ -888,7 +888,13 @@ async def call(
         from ..worker.stage import set_stage
 
         structured = "+structured" if response_format is not None else ""
-        set_stage(f"llm.{self.provider}.{scope}{structured}")
+        # `.queued` until the concurrency permits are in hand — see the acquire
+        # below. Without it, a call waiting on a saturated semaphore is
+        # indistinguishable from one the provider is actively running, and the
+        # label points at the provider (#3002: an operator lost an hour to
+        # "llm.bedrock.*" for tasks that had never reached Bedrock).
+        base_stage = f"llm.{self.provider}.{scope}{structured}"
+        set_stage(f"{base_stage}.queued")
 
         # Resolve the retry policy: explicit per-call arg wins, else the provider's
         # configured per-operation/global default, else this method's own fallback.
@@ -948,6 +954,7 @@ async def call(
             async with AsyncExitStack() as stack:
                 for sem in _semaphores_for_scope(scope):
                     await stack.enter_async_context(sem)
+                set_stage(base_stage)
 
                 # cached_prefix is only set for providers that returned a handle
                 # from get_or_create_cached_prefix() (e.g. Gemini); it's None for
@@ -1039,7 +1046,9 @@ async def call_with_tools(
         """
         from ..worker.stage import set_stage
 
-        set_stage(f"llm.{self.provider}.{scope}+tools")
+        # `.queued` until the permits are held — see the structured path above.
+        base_stage = f"llm.{self.provider}.{scope}+tools"
+        set_stage(f"{base_stage}.queued")
 
         # Resolve the retry policy: explicit per-call arg wins, else the provider's
         # configured per-operation/global default, else this method's own fallback.
@@ -1081,6 +1090,7 @@ async def call_with_tools(
             async with AsyncExitStack() as stack:
                 for sem in _semaphores_for_scope(scope):
                     await stack.enter_async_context(sem)
+                set_stage(base_stage)
 
                 # cached_prefix is only set for providers that returned a handle
                 # from get_or_create_cached_prefix() / create_incremental_cache();

hindsight-api-slim/hindsight_api/engine/providers/gemini_llm.py

@@ -424,8 +424,7 @@ def _build_generation_config(use_cache: bool) -> "genai_types.GenerateContentCon
         last_exception = None
 
         for attempt in range(max_retries + 1):
-            if attempt > 0:
-                set_stage(f"llm.gemini.{scope}.attempt={attempt + 1}/{max_retries + 1}")
+            set_stage(f"llm.gemini.{scope}.attempt={attempt + 1}/{max_retries + 1}")
             try:
                 response = await asyncio.wait_for(
                     self._client.aio.models.generate_content(
@@ -769,8 +768,7 @@ def _build_tools_config(use_cache: bool) -> "genai_types.GenerateContentConfig":
 
         last_exception = None
         for attempt in range(max_retries + 1):
-            if attempt > 0:
-                set_stage(f"llm.gemini.tools.attempt={attempt + 1}/{max_retries + 1}")
+            set_stage(f"llm.gemini.tools.attempt={attempt + 1}/{max_retries + 1}")
             try:
                 # With the cache active, send only the un-cached tail (delta);
                 # on the uncached fallback path send the full conversation so the

hindsight-api-slim/hindsight_api/engine/providers/litellm_llm.py

@@ -255,8 +255,7 @@ async def call(
         last_exception = None
 
         for attempt in range(max_retries + 1):
-            if attempt > 0:
-                set_stage(f"llm.{self._stage_label}.{scope}.attempt={attempt + 1}/{max_retries + 1}")
+            set_stage(f"llm.{self._stage_label}.{scope}.attempt={attempt + 1}/{max_retries + 1}")
             try:
                 response = await asyncio.wait_for(
                     self._acompletion(**call_kwargs),
@@ -447,8 +446,7 @@ async def call_with_tools(
 
         last_exception = None
         for attempt in range(max_retries + 1):
-            if attempt > 0:
-                set_stage(f"llm.{self._stage_label}.tools.attempt={attempt + 1}/{max_retries + 1}")
+            set_stage(f"llm.{self._stage_label}.tools.attempt={attempt + 1}/{max_retries + 1}")
             try:
                 response = await asyncio.wait_for(
                     self._acompletion(**call_kwargs),

hindsight-api-slim/hindsight_api/engine/providers/openai_compatible_llm.py

@@ -899,8 +899,7 @@ async def call(
             # Surface attempt count in worker stage so JSON-schema retry loops
             # are visible from logs (small models on strict structured output
             # often loop here). Cheap no-op outside worker context.
-            if attempt > 0:
-                set_stage(f"llm.{self.provider}.{scope}.attempt={attempt + 1}/{max_retries + 1}")
+            set_stage(f"llm.{self.provider}.{scope}.attempt={attempt + 1}/{max_retries + 1}")
             try:
                 if response_format is not None:
                     response = await self._client.chat.completions.create(**call_params)
@@ -1269,8 +1268,7 @@ async def call_with_tools(
         last_exception = None
 
         for attempt in range(max_retries + 1):
-            if attempt > 0:
-                set_stage(f"llm.{self.provider}.tools.attempt={attempt + 1}/{max_retries + 1}")
+            set_stage(f"llm.{self.provider}.tools.attempt={attempt + 1}/{max_retries + 1}")
             try:
                 response = await self._client.chat.completions.create(**call_params)
 
@@ -1474,8 +1472,7 @@ async def _call_ollama_native(
 
         async with httpx.AsyncClient(timeout=300.0) as client:
             for attempt in range(max_retries + 1):
-                if attempt > 0:
-                    set_stage(f"llm.ollama_native.{scope}.attempt={attempt + 1}/{max_retries + 1}")
+                set_stage(f"llm.ollama_native.{scope}.attempt={attempt + 1}/{max_retries + 1}")
                 try:
                     response = await client.post(native_url, json=payload, headers=headers)
                     response.raise_for_status()

hindsight-api-slim/hindsight_api/engine/retain/orchestrator.py

@@ -1412,26 +1412,38 @@ async def _extract_one(global_idx: int, chunk_text: str) -> None:
 
         tasks: list[asyncio.Task] = []
         skipped_total = 0
-        for i, chunk_text in enumerate(all_pre_chunks):
-            chunk_hash = chunk_storage.compute_chunk_hash(chunk_text)
-            if chunk_hash in existing_chunk_hashes:
-                # Memory: skipped chunks aren't needed either.
-                all_pre_chunks[i] = ""
-                skipped_total += 1
-                continue
-            tasks.append(asyncio.create_task(_extract_one(i, chunk_text)))
-
-        if skipped_total > 0:
-            log_buffer.append(f"[streaming] Producer: skipped {skipped_total}/{total_chunks} already-committed chunks")
+        try:
+            for i, chunk_text in enumerate(all_pre_chunks):
+                chunk_hash = chunk_storage.compute_chunk_hash(chunk_text)
+                if chunk_hash in existing_chunk_hashes:
+                    # Memory: skipped chunks aren't needed either.
+                    all_pre_chunks[i] = ""
+                    skipped_total += 1
+                    continue
+                tasks.append(asyncio.create_task(_extract_one(i, chunk_text)))
 
-        # Wait for all extractions; collect exceptions
-        results = await asyncio.gather(*tasks, return_exceptions=True)
-        for r in results:
-            if isinstance(r, BaseException):
-                producer_error.append(r)
+            if skipped_total > 0:
+                log_buffer.append(
+                    f"[streaming] Producer: skipped {skipped_total}/{total_chunks} already-committed chunks"
+                )
 
-        # Signal the consumer that production is done
-        await chunk_queue.put(None)
+            # Wait for all extractions; collect exceptions
+            results = await asyncio.gather(*tasks, return_exceptions=True)
+            for r in results:
+                if isinstance(r, BaseException):
+                    producer_error.append(r)
+
+            # Signal the consumer that production is done
+            await chunk_queue.put(None)
+        finally:
+            # Cancellation arriving mid-fan-out (the consumer failed, or the worker's
+            # wall-clock ceiling fired) must not strand extraction tasks. Cancelling
+            # the gather above already propagates to them, but tasks created before
+            # we reach it would otherwise survive and park on `chunk_queue.put()`
+            # for the life of the process.
+            for extraction in tasks:
+                if not extraction.done():
+                    extraction.cancel()
 
     # ---- DB Consumer ----
     # Drains enriched chunks from the queue in batches and runs
@@ -1869,8 +1881,27 @@ async def _run_mini_batch_db_work() -> None:
             logger.warning("Failed to check operation recovery state", exc_info=True)
 
     if not facts_already_committed:
-        # Run producer and consumer concurrently
-        await asyncio.gather(_llm_producer(), _db_consumer())
+        # Run producer and consumer concurrently.
+        #
+        # Cancellation is explicit because plain gather() leaks: when the consumer
+        # raises (a deadlock victim, a lock timeout) gather propagates that error
+        # immediately but leaves the producer — and every extraction task under it
+        # — running. Those tasks then block forever on `chunk_queue.put()` into a
+        # queue nobody drains, pinning their chunk payloads and still spending LLM
+        # permits and tokens on an operation that already failed (#3002). The same
+        # applies when the worker's wall-clock ceiling cancels us from above.
+        producer_task = asyncio.create_task(_llm_producer())
+        consumer_task = asyncio.create_task(_db_consumer())
+        try:
+            await asyncio.gather(producer_task, consumer_task)
+        finally:
+            for pipeline_task in (producer_task, consumer_task):
+                if not pipeline_task.done():
+                    pipeline_task.cancel()
+            # Await the cancellations so neither half outlives this call; the
+            # results are already accounted for by the gather above (or by the
+            # exception that is propagating).
+            await asyncio.gather(producer_task, consumer_task, return_exceptions=True)
 
         # Propagate producer errors (e.g. LLM failures)
         if producer_error: