From 904ea4de24812eeeb625e3561454b0c4930d3308 Mon Sep 17 00:00:00 2001 From: Chris Bartholomew Date: Mon, 22 Dec 2025 14:42:57 -0500 Subject: [PATCH] fix: propagate exceptions from task handlers to enable retry logic (#65) Task handlers were swallowing exceptions, causing operations to be marked as completed even when they failed. This prevented the retry logic in execute_task() from working and led to accumulation of pending operations that never completed. Fixed handlers: - _handle_batch_retain: remove try/except wrapper - _handle_access_count_update: remove try/except wrapper - _handle_regenerate_observations: remove outer try/except, keep inner one for individual entity failures --- .../hindsight_api/engine/memory_engine.py | 191 +++++++++--------- 1 file changed, 93 insertions(+), 98 deletions(-) diff --git a/hindsight-api/hindsight_api/engine/memory_engine.py b/hindsight-api/hindsight_api/engine/memory_engine.py index cd55eee7..d5bda4d8 100644 --- a/hindsight-api/hindsight_api/engine/memory_engine.py +++ b/hindsight-api/hindsight_api/engine/memory_engine.py @@ -396,22 +396,22 @@ class MemoryEngine(MemoryEngineInterface): Args: task_dict: Dict with 'node_ids' key containing list of node IDs to update + + Raises: + Exception: Any exception from database operations (propagates to execute_task for retry) """ node_ids = task_dict.get("node_ids", []) if not node_ids: return pool = await self._get_pool() - try: - # Convert string UUIDs to UUID type for faster matching - uuid_list = [uuid.UUID(nid) for nid in node_ids] - async with acquire_with_retry(pool) as conn: - await conn.execute( - f"UPDATE {fq_table('memory_units')} SET access_count = access_count + 1 WHERE id = ANY($1::uuid[])", - uuid_list, - ) - except Exception as e: - logger.error(f"Access count handler: Error updating access counts: {e}") + # Convert string UUIDs to UUID type for faster matching + uuid_list = [uuid.UUID(nid) for nid in node_ids] + async with acquire_with_retry(pool) as conn: + await conn.execute( + f"UPDATE {fq_table('memory_units')} SET access_count = access_count + 1 WHERE id = ANY($1::uuid[])", + uuid_list, + ) async def _handle_batch_retain(self, task_dict: dict[str, Any]): """ @@ -419,29 +419,27 @@ class MemoryEngine(MemoryEngineInterface): Args: task_dict: Dict with 'bank_id', 'contents' + + Raises: + ValueError: If bank_id is missing + Exception: Any exception from retain_batch_async (propagates to execute_task for retry) """ - try: - bank_id = task_dict.get("bank_id") - if not bank_id: - raise ValueError("bank_id is required for batch retain task") - contents = task_dict.get("contents", []) + bank_id = task_dict.get("bank_id") + if not bank_id: + raise ValueError("bank_id is required for batch retain task") + contents = task_dict.get("contents", []) - logger.info( - f"[BATCH_RETAIN_TASK] Starting background batch retain for bank_id={bank_id}, {len(contents)} items" - ) + logger.info( + f"[BATCH_RETAIN_TASK] Starting background batch retain for bank_id={bank_id}, {len(contents)} items" + ) - # Use internal request context for background tasks - from hindsight_api.models import RequestContext + # Use internal request context for background tasks + from hindsight_api.models import RequestContext - internal_context = RequestContext() - await self.retain_batch_async(bank_id=bank_id, contents=contents, request_context=internal_context) + internal_context = RequestContext() + await self.retain_batch_async(bank_id=bank_id, contents=contents, request_context=internal_context) - logger.info(f"[BATCH_RETAIN_TASK] Completed background batch retain for bank_id={bank_id}") - except Exception as e: - logger.error(f"Batch retain handler: Error processing batch retain: {e}") - import traceback - - traceback.print_exc() + logger.info(f"[BATCH_RETAIN_TASK] Completed background batch retain for bank_id={bank_id}") async def execute_task(self, task_dict: dict[str, Any]): """ @@ -3649,90 +3647,87 @@ Guidelines: task_dict: Dict with 'bank_id' and either: - 'entity_ids' (list): Process multiple entities - 'entity_id', 'entity_name': Process single entity (legacy) + + Raises: + ValueError: If required fields are missing + Exception: Any exception from regenerate_entity_observations (propagates to execute_task for retry) """ - try: - bank_id = task_dict.get("bank_id") - # Use internal request context for background tasks - from hindsight_api.models import RequestContext + bank_id = task_dict.get("bank_id") + # Use internal request context for background tasks + from hindsight_api.models import RequestContext - internal_context = RequestContext() + internal_context = RequestContext() - # New format: multiple entity_ids - if "entity_ids" in task_dict: - entity_ids = task_dict.get("entity_ids", []) - min_facts = task_dict.get("min_facts", 5) + # New format: multiple entity_ids + if "entity_ids" in task_dict: + entity_ids = task_dict.get("entity_ids", []) + min_facts = task_dict.get("min_facts", 5) - if not bank_id or not entity_ids: - logger.error(f"[OBSERVATIONS] Missing required fields in task: {task_dict}") - return + if not bank_id or not entity_ids: + raise ValueError(f"[OBSERVATIONS] Missing required fields in task: {task_dict}") - # Process each entity - pool = await self._get_pool() - async with pool.acquire() as conn: - for entity_id in entity_ids: - try: - # Fetch entity name and check fact count - import uuid as uuid_module + # Process each entity + pool = await self._get_pool() + async with pool.acquire() as conn: + for entity_id in entity_ids: + try: + # Fetch entity name and check fact count + import uuid as uuid_module - entity_uuid = uuid_module.UUID(entity_id) if isinstance(entity_id, str) else entity_id + entity_uuid = uuid_module.UUID(entity_id) if isinstance(entity_id, str) else entity_id - # First check if entity exists - entity_exists = await conn.fetchrow( - f"SELECT canonical_name FROM {fq_table('entities')} WHERE id = $1 AND bank_id = $2", - entity_uuid, - bank_id, - ) + # First check if entity exists + entity_exists = await conn.fetchrow( + f"SELECT canonical_name FROM {fq_table('entities')} WHERE id = $1 AND bank_id = $2", + entity_uuid, + bank_id, + ) - if not entity_exists: - logger.debug(f"[OBSERVATIONS] Entity {entity_id} not yet in bank {bank_id}, skipping") - continue - - entity_name = entity_exists["canonical_name"] - - # Count facts linked to this entity - fact_count = ( - await conn.fetchval( - f"SELECT COUNT(*) FROM {fq_table('unit_entities')} WHERE entity_id = $1", - entity_uuid, - ) - or 0 - ) - - # Only regenerate if entity has enough facts - if fact_count >= min_facts: - await self.regenerate_entity_observations( - bank_id, entity_id, entity_name, version=None, request_context=internal_context - ) - else: - logger.debug( - f"[OBSERVATIONS] Skipping {entity_name} ({fact_count} facts < {min_facts} threshold)" - ) - - except Exception as e: - logger.error(f"[OBSERVATIONS] Error processing entity {entity_id}: {e}") + if not entity_exists: + logger.debug(f"[OBSERVATIONS] Entity {entity_id} not yet in bank {bank_id}, skipping") continue - # Legacy format: single entity - else: - entity_id = task_dict.get("entity_id") - entity_name = task_dict.get("entity_name") - version = task_dict.get("version") + entity_name = entity_exists["canonical_name"] - if not all([bank_id, entity_id, entity_name]): - logger.error(f"[OBSERVATIONS] Missing required fields in task: {task_dict}") - return + # Count facts linked to this entity + fact_count = ( + await conn.fetchval( + f"SELECT COUNT(*) FROM {fq_table('unit_entities')} WHERE entity_id = $1", + entity_uuid, + ) + or 0 + ) - # Type assertions after validation - assert isinstance(bank_id, str) and isinstance(entity_id, str) and isinstance(entity_name, str) - await self.regenerate_entity_observations( - bank_id, entity_id, entity_name, version=version, request_context=internal_context - ) + # Only regenerate if entity has enough facts + if fact_count >= min_facts: + await self.regenerate_entity_observations( + bank_id, entity_id, entity_name, version=None, request_context=internal_context + ) + else: + logger.debug( + f"[OBSERVATIONS] Skipping {entity_name} ({fact_count} facts < {min_facts} threshold)" + ) - except Exception as e: - logger.error(f"[OBSERVATIONS] Error regenerating observations: {e}") - import traceback + except Exception as e: + # Log but continue processing other entities - individual entity failures + # shouldn't fail the whole batch + logger.error(f"[OBSERVATIONS] Error processing entity {entity_id}: {e}") + continue - traceback.print_exc() + # Legacy format: single entity + else: + entity_id = task_dict.get("entity_id") + entity_name = task_dict.get("entity_name") + version = task_dict.get("version") + + if not all([bank_id, entity_id, entity_name]): + raise ValueError(f"[OBSERVATIONS] Missing required fields in task: {task_dict}") + + # Type assertions after validation + assert isinstance(bank_id, str) and isinstance(entity_id, str) and isinstance(entity_name, str) + await self.regenerate_entity_observations( + bank_id, entity_id, entity_name, version=version, request_context=internal_context + ) # ========================================================================= # Statistics & Operations (for HTTP API layer)