fix(worker): prevent child tasks from blocking parent execution

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

@@ -2586,7 +2586,7 @@ async def lifespan(app: FastAPI):
                 schema=schema,
                 tenant_extension=memory._tenant_extension,
                 max_slots=config.worker_max_slots,
-                consolidation_max_slots=config.worker_consolidation_max_slots,
+                slot_reservations=config.worker_slot_reservations,
             )
             poller_task = asyncio.create_task(poller.run())
             logging.info(f"Worker poller started (worker_id={worker_id})")

hindsight-api-slim/hindsight_api/config.py

@@ -383,7 +383,18 @@ def normalize_config_dict(config: dict[str, Any]) -> dict[str, Any]:
 ENV_WORKER_MAX_RETRIES = "HINDSIGHT_API_WORKER_MAX_RETRIES"
 ENV_WORKER_HTTP_PORT = "HINDSIGHT_API_WORKER_HTTP_PORT"
 ENV_WORKER_MAX_SLOTS = "HINDSIGHT_API_WORKER_MAX_SLOTS"
-ENV_WORKER_CONSOLIDATION_MAX_SLOTS = "HINDSIGHT_API_WORKER_CONSOLIDATION_MAX_SLOTS"
+
+# Per-operation-type slot reservations. Each entry maps an operation_type
+# (as stored in async_operations.operation_type) to its env var and default.
+# Adding a new operation type here is the ONLY change needed to make it
+# reservable via env var — config fields, from_env(), and the
+# worker_slot_reservations property all derive from this dict.
+WORKER_SLOT_RESERVATION_TYPES: dict[str, tuple[str, int]] = {
+    "consolidation": ("HINDSIGHT_API_WORKER_CONSOLIDATION_MAX_SLOTS", 2),
+    "retain": ("HINDSIGHT_API_WORKER_RETAIN_MAX_SLOTS", 0),
+    "file_convert_retain": ("HINDSIGHT_API_WORKER_FILE_CONVERT_RETAIN_MAX_SLOTS", 0),
+    "refresh_mental_model": ("HINDSIGHT_API_WORKER_REFRESH_MENTAL_MODEL_MAX_SLOTS", 0),
+}
 ENV_RETAIN_MAX_CONCURRENT = "HINDSIGHT_API_RETAIN_MAX_CONCURRENT"
 
 # Reflect agent settings
@@ -604,7 +615,6 @@ def normalize_config_dict(config: dict[str, Any]) -> dict[str, Any]:
 DEFAULT_WORKER_MAX_RETRIES = 3  # Max retries before marking task failed
 DEFAULT_WORKER_HTTP_PORT = 8889  # HTTP port for worker metrics/health
 DEFAULT_WORKER_MAX_SLOTS = 10  # Total concurrent tasks per worker
-DEFAULT_WORKER_CONSOLIDATION_MAX_SLOTS = 2  # Max concurrent consolidation tasks per worker
 DEFAULT_RETAIN_MAX_CONCURRENT = 4  # Max concurrent retain DB phases (HNSW reads + writes). Limits I/O contention.
 
 # Reflect agent settings
@@ -1062,7 +1072,7 @@ class HindsightConfig:
     worker_max_retries: int
     worker_http_port: int
     worker_max_slots: int
-    worker_consolidation_max_slots: int
+    worker_slot_reservations: dict[str, int]
     retain_max_concurrent: int
 
     # Reflect agent settings
@@ -1275,6 +1285,16 @@ def validate(self) -> None:
                 f"provider: {self.retain_llm_provider or self.llm_provider})"
             )
 
+        # Validate that sum of per-operation slot reservations does not exceed max_slots
+        total_reserved = sum(self.worker_slot_reservations.values())
+        if total_reserved > self.worker_max_slots:
+            reservation_details = ", ".join(f"{k}={v}" for k, v in self.worker_slot_reservations.items() if v > 0)
+            raise ValueError(
+                f"Sum of per-operation slot reservations ({total_reserved}: {reservation_details}) "
+                f"exceeds worker_max_slots ({self.worker_max_slots}). "
+                f"Reduce reservations or increase HINDSIGHT_API_WORKER_MAX_SLOTS."
+            )
+
     @classmethod
     def from_env(cls) -> "HindsightConfig":
         """Create configuration from environment variables."""
@@ -1658,9 +1678,11 @@ def from_env(cls) -> "HindsightConfig":
             worker_max_retries=int(os.getenv(ENV_WORKER_MAX_RETRIES, str(DEFAULT_WORKER_MAX_RETRIES))),
             worker_http_port=int(os.getenv(ENV_WORKER_HTTP_PORT, str(DEFAULT_WORKER_HTTP_PORT))),
             worker_max_slots=int(os.getenv(ENV_WORKER_MAX_SLOTS, str(DEFAULT_WORKER_MAX_SLOTS))),
-            worker_consolidation_max_slots=int(
-                os.getenv(ENV_WORKER_CONSOLIDATION_MAX_SLOTS, str(DEFAULT_WORKER_CONSOLIDATION_MAX_SLOTS))
-            ),
+            worker_slot_reservations={
+                op_type: int(os.getenv(env_var, str(default)))
+                for op_type, (env_var, default) in WORKER_SLOT_RESERVATION_TYPES.items()
+                if int(os.getenv(env_var, str(default))) > 0
+            },
             retain_max_concurrent=int(os.getenv(ENV_RETAIN_MAX_CONCURRENT, str(DEFAULT_RETAIN_MAX_CONCURRENT))),
             # Reflect agent settings
             reflect_max_iterations=int(os.getenv(ENV_REFLECT_MAX_ITERATIONS, str(DEFAULT_REFLECT_MAX_ITERATIONS))),

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

@@ -2,7 +2,8 @@
 Task backend for distributed task processing.
 
 This provides an abstraction for task storage and execution:
-- BrokerTaskBackend: Uses PostgreSQL as broker (production)
+- BrokerTaskBackend: Uses PostgreSQL as broker (production API servers)
+- WorkerTaskBackend: No-op submit_task (production workers — child tasks are polled)
 - SyncTaskBackend: Executes tasks immediately (testing/embedded)
 """
 
@@ -125,6 +126,33 @@ async def shutdown(self):
         logger.debug("SyncTaskBackend shutdown")
 
 
+class WorkerTaskBackend(TaskBackend):
+    """
+    Task backend for worker processes.
+
+    Workers execute tasks directly via the poller (claim → execute), so they
+    don't need submit_task to run anything.  When engine code running *inside*
+    a worker-executed task calls submit_task (e.g. retain triggers consolidation),
+    the row has already been INSERTed into async_operations with task_payload by
+    _submit_async_operation — so submit_task is a no-op.  The new task will be
+    picked up by a worker on the next poll cycle instead of being executed inline,
+    which avoids blocking the parent task.
+    """
+
+    async def initialize(self):
+        self._initialized = True
+        logger.debug("WorkerTaskBackend initialized")
+
+    async def submit_task(self, task_dict: dict[str, Any]):
+        """No-op: the row already exists in async_operations; a worker will claim it."""
+        task_type = task_dict.get("type", "unknown")
+        logger.debug(f"WorkerTaskBackend: submit_task no-op for {task_type} (will be picked up by poller)")
+
+    async def shutdown(self):
+        self._initialized = False
+        logger.debug("WorkerTaskBackend shutdown")
+
+
 class BrokerTaskBackend(TaskBackend):
     """
     Task backend using PostgreSQL as broker.

hindsight-api-slim/hindsight_api/worker/main.py

@@ -18,7 +18,7 @@
 import warnings
 
 from ..config import get_config
-from ..engine.task_backend import SyncTaskBackend
+from ..engine.task_backend import WorkerTaskBackend
 from .poller import WorkerPoller
 
 # Filter deprecation warnings from third-party libraries
@@ -164,7 +164,11 @@ def main():
     print(f"  Poll interval: {args.poll_interval}ms")
     print(f"  Max retries: {args.max_retries}")
     print(f"  Max slots: {config.worker_max_slots}")
-    print(f"  Consolidation max slots: {config.worker_consolidation_max_slots}")
+    reservations = config.worker_slot_reservations
+    reservations_str = ", ".join(f"{k}={v}" for k, v in reservations.items()) if reservations else "none"
+    shared_pool = max(0, config.worker_max_slots - sum(reservations.values()))
+    print(f"  Slot reservations: {reservations_str}")
+    print(f"  Shared pool: {shared_pool}")
     print(f"  HTTP server: {args.http_host}:{args.http_port}")
     print()
 
@@ -191,11 +195,13 @@ async def run():
             logger.info(f"Loaded operation validator: {operation_validator.__class__.__name__}")
 
         # Initialize MemoryEngine
-        # Workers use SyncTaskBackend because they execute tasks directly,
-        # they don't need to store tasks (they poll from DB)
+        # Workers use WorkerTaskBackend: submit_task is a no-op because the
+        # row already exists in async_operations.  Child tasks (e.g. consolidation
+        # triggered by retain) will be picked up by the poller on the next cycle
+        # instead of being executed inline, which avoids blocking the parent task.
         memory = MemoryEngine(
             run_migrations=False,  # Workers don't run migrations
-            task_backend=SyncTaskBackend(),
+            task_backend=WorkerTaskBackend(),
             tenant_extension=tenant_extension,
             operation_validator=operation_validator,
         )
@@ -222,7 +228,7 @@ async def run():
             schema=schema,
             tenant_extension=tenant_extension,
             max_slots=config.worker_max_slots,
-            consolidation_max_slots=config.worker_consolidation_max_slots,
+            slot_reservations=config.worker_slot_reservations,
         )
 
         # Create the HTTP app for metrics/health