UN-3211 [FEAT] HTTP session lifecycle management for workers API clients (#1782)

* UN-3211 [FEAT] HTTP session lifecycle management for workers API clients

- Add _owns_session flag to prevent singleton shared session from being
  closed by individual clients
- Wire API_CLIENT_POOL_SIZE into HTTPAdapter connection pools
- Add idempotent close() and __del__ destructor to BaseAPIClient
- Add try/finally cleanup in api-deployment and callback tasks
- Add on_worker_process_shutdown hook and early-return guard in postrun
- Add 25 unit tests for session lifecycle behavior

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

* UN-3211 [FIX] Address CodeRabbit review: log reset failures, thread-safe counter

- Log warning instead of silently swallowing exceptions in reset_singleton()
- Add threading.Lock around task counter increment for thread safety with
  threads/gevent/eventlet pools

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

* UN-3211 [FIX] Address CodeRabbit review round 2: optimize lock scope, document thread-safety

- Move WorkerConfig() instantiation outside lock in increment_task_counter()
- Remove redundant _task_counter=0 (already done inside reset_singleton)
- Document thread-safety caveat in reset_singleton() docstring
- Log close failures in task cleanup instead of silently swallowing

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

* UN-3211 [FIX] Address CodeRabbit review round 3: test real on_task_postrun handler

Tests now call the real worker.on_task_postrun() signal handler instead of
simulating the guard logic inline, catching divergence if the handler's
guard, try/except, or import path changes.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

* UN-3211 [FIX] Optimize singleton reset with cached threshold and improve close() logging

Cache the singleton_reset_task_threshold to avoid re-importing WorkerConfig on
every task increment. Promote api_client.close() failure logs from debug to
warning for better production visibility. Update tests to reset cached threshold.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

* UN-3211 [FIX] Improve API execution error handling and clean up worker imports

Refactor api-deployment tasks to handle setup failures early with proper
cleanup, move shared imports to module level in worker.py, and fix type
annotations in client_factory.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 4.6 <noreply@anthropic.com>
Co-authored-by: Chandrasekharan M <117059509+chandrasekharan-zipstack@users.noreply.github.com>
Co-authored-by: Kirtiman Mishra <110175055+kirtimanmishrazipstack@users.noreply.github.com>

Author: 4 people

workers/api-deployment/tasks.py

@@ -168,13 +168,30 @@ def _unified_api_execution(
     Returns:
         Execution result dictionary
     """
+    # Set up execution context - exit early if setup fails
+    organization_id = schema_name
     try:
-        # Set up execution context using shared utilities
-        organization_id = schema_name
-        config, api_client = WorkerExecutionContext.setup_execution_context(
+        _, api_client = WorkerExecutionContext.setup_execution_context(
             organization_id, execution_id, workflow_id
         )
+    except Exception as e:
+        logger.error(f"Failed to setup execution context: {e}")
+        # Clean up StateStore since setup_execution_context may have
+        # partially populated it before failing
+        try:
+            from shared.infrastructure.context import StateStore
 
+            StateStore.clear_all()
+        except Exception:
+            logger.debug("StateStore cleanup also failed during early exit")
+        return {
+            "execution_id": execution_id,
+            "status": "ERROR",
+            "error": str(e),
+            "files_processed": 0,
+        }
+
+    try:
         # Log task start with standardized format
         WorkerExecutionContext.log_task_start(
             f"unified_api_execution_{task_type}",
@@ -216,12 +233,12 @@ def _unified_api_execution(
             schema_name=organization_id,
             workflow_id=workflow_id,
             execution_id=execution_id,
-            hash_values_of_files=converted_files,  # Changed parameter name
+            hash_values_of_files=converted_files,
             scheduled=scheduled,
             execution_mode=execution_mode,
             pipeline_id=pipeline_id,
             use_file_history=use_file_history,
-            task_id=task_instance.request.id,  # Add required task_id
+            task_id=task_instance.request.id,
             **kwargs,
         )
 
@@ -233,25 +250,14 @@ def _unified_api_execution(
             f"files_processed={len(converted_files)}",
         )
 
-        # CRITICAL: Clean up StateStore to prevent data leaks between tasks
-        try:
-            from shared.infrastructure.context import StateStore
-
-            StateStore.clear_all()
-            logger.debug("🧹 Cleaned up StateStore context to prevent data leaks")
-        except Exception as cleanup_error:
-            logger.warning(f"Failed to cleanup StateStore context: {cleanup_error}")
-
         return result
 
     except Exception as e:
         logger.error(f"API execution failed: {e}")
 
-        # Handle execution error with standardized pattern
-        if "api_client" in locals():
-            WorkerExecutionContext.handle_execution_error(
-                api_client, execution_id, e, logger, f"api_execution_{task_type}"
-            )
+        WorkerExecutionContext.handle_execution_error(
+            api_client, execution_id, e, logger, f"api_execution_{task_type}"
+        )
 
         # Log completion with error
         WorkerExecutionContext.log_task_completion(
@@ -261,26 +267,27 @@ def _unified_api_execution(
             f"error={str(e)}",
         )
 
-        # CRITICAL: Clean up StateStore to prevent data leaks between tasks (error path)
-        try:
-            from shared.infrastructure.context import StateStore
-
-            StateStore.clear_all()
-            logger.debug(
-                "🧹 Cleaned up StateStore context to prevent data leaks (error path)"
-            )
-        except Exception as cleanup_error:
-            logger.warning(
-                f"Failed to cleanup StateStore context on error: {cleanup_error}"
-            )
-
         return {
             "execution_id": execution_id,
             "status": "ERROR",
             "error": str(e),
             "files_processed": 0,
         }
 
+    finally:
+        try:
+            api_client.close()
+        except Exception as e:
+            logger.warning("api_client.close() failed during cleanup: %s", e)
+
+        # Clean up StateStore to prevent data leaks between tasks
+        try:
+            from shared.infrastructure.context import StateStore
+
+            StateStore.clear_all()
+        except Exception as cleanup_error:
+            logger.warning(f"Failed to cleanup StateStore context: {cleanup_error}")
+
 
 @app.task(
     bind=True,