From 4a3b620c11557771b2b8c95bc2a9a2177bf9c198 Mon Sep 17 00:00:00 2001
From: Christian Cabauatan <christian@tryferris.ai>
Date: Wed, 15 Apr 2026 09:57:01 -0700
Subject: [PATCH] fix: files/retain upload problems and orphaned retains

Include task_payload in the async_operations INSERT atomically instead
of the previous two-step INSERT-then-UPDATE approach. When a crash or
timeout occurred between the two statements, rows were left with
task_payload IS NULL. The worker claim query filters on
task_payload IS NOT NULL, so those orphaned rows became permanently
stuck as unclaimed pending tasks.
---
 .../hindsight_api/engine/memory_engine.py     | 30 +++++++++++--------
 1 file changed, 18 insertions(+), 12 deletions(-)

diff --git a/hindsight-api-slim/hindsight_api/engine/memory_engine.py b/hindsight-api-slim/hindsight_api/engine/memory_engine.py
index dc30994302..ba8bef51a8 100644
--- a/hindsight-api-slim/hindsight_api/engine/memory_engine.py
+++ b/hindsight-api-slim/hindsight_api/engine/memory_engine.py
@@ -854,26 +854,37 @@ async def _handle_file_convert_retain(self, task_dict: dict[str, Any]):
             "file_content_type": task_dict["content_type"],
         }
 
-        # In one transaction: create the retain async operation AND mark this conversion as completed
+        # Include task_payload in the INSERT atomically. Previously this was a
+        # two-step process (INSERT without payload, then UPDATE to set it) which
+        # left null-payload rows when a crash or timeout occurred between the two
+        # statements. The worker claim query filters on `task_payload IS NOT NULL`,
+        # so those orphaned rows became permanently stuck as unclaimed pending tasks.
         retain_operation_id = uuid.uuid4()
+        full_retain_payload = {
+            "type": "batch_retain",
+            "operation_id": str(retain_operation_id),
+            "bank_id": bank_id,
+            **retain_task_payload,
+        }
+        payload_json = json.dumps(full_retain_payload, default=_json_default)
+
         pool = await self._get_pool()
         async with acquire_with_retry(pool) as conn:
             async with conn.transaction():
-                # Create the retain operation record
                 await conn.execute(
                     f"""
                     INSERT INTO {fq_table("async_operations")}
-                    (operation_id, bank_id, operation_type, result_metadata, status)
-                    VALUES ($1, $2, $3, $4, $5)
+                    (operation_id, bank_id, operation_type, result_metadata, status, task_payload)
+                    VALUES ($1, $2, $3, $4, $5, $6::jsonb)
                     """,
                     retain_operation_id,
                     bank_id,
                     "retain",
                     json.dumps({}),
                     "pending",
+                    payload_json,
                 )
 
-                # Mark this file_convert_retain operation as completed
                 if operation_id:
                     await conn.execute(
                         f"""
@@ -884,13 +895,8 @@ async def _handle_file_convert_retain(self, task_dict: dict[str, Any]):
                         uuid.UUID(operation_id),
                     )
 
-        # Submit the retain task to the task backend (outside the transaction)
-        full_retain_payload = {
-            "type": "batch_retain",
-            "operation_id": str(retain_operation_id),
-            "bank_id": bank_id,
-            **retain_task_payload,
-        }
+        # For SyncTaskBackend: executes the retain task inline.
+        # For BrokerTaskBackend: idempotent UPDATE (payload already set above).
         await self._task_backend.submit_task(full_retain_payload)
 
         logger.info(