[FIX] Executor Queue for Agentic extraction

backend/prompt_studio/prompt_studio_core_v2/static/select_choices.json

@@ -14,7 +14,6 @@
         "date":"date",
         "boolean":"boolean",
         "json":"json",
-        "line_item":"line-item",
         "table":"table"
     },
     "output_processing":{

backend/workflow_manager/workflow_v2/workflow_helper.py

@@ -598,11 +598,7 @@ def execute_workflow_async(
     @staticmethod
     @celery_app.task(
         name="async_execute_bin",
-        autoretry_for=(Exception,),
         max_retries=0,
-        retry_backoff=True,
-        retry_backoff_max=500,
-        retry_jitter=True,
     )
     def execute_bin(
         schema_name: str,

docker/docker-compose.yaml

@@ -529,7 +529,7 @@ services:
       - EXECUTOR_METRICS_PORT=8088
       - HEALTH_PORT=8088
       # Configurable Celery options
-      - CELERY_QUEUES_EXECUTOR=${CELERY_QUEUES_EXECUTOR:-celery_executor_legacy}
+      - CELERY_QUEUES_EXECUTOR=${CELERY_QUEUES_EXECUTOR:-celery_executor_legacy,celery_executor_agentic}
       - CELERY_POOL=${WORKER_EXECUTOR_POOL:-prefork}
       - CELERY_PREFETCH_MULTIPLIER=${WORKER_EXECUTOR_PREFETCH_MULTIPLIER:-1}
       - CELERY_CONCURRENCY=${WORKER_EXECUTOR_CONCURRENCY:-2}

workers/executor/executors/legacy_executor.py

@@ -1059,9 +1059,7 @@ def _sanitize_null_values(
     ) -> dict[str, Any]:
         """Replace 'NA' strings with None in structured output."""
         for k, v in structured_output.items():
-            if isinstance(v, str) and v.lower() == "na":
-                structured_output[k] = None
-            elif isinstance(v, list):
+            if isinstance(v, list):
                 for i, item in enumerate(v):
                     if isinstance(item, str) and item.lower() == "na":
                         v[i] = None
@@ -1698,16 +1696,19 @@ def _apply_type_conversion(
             )
 
         elif output_type == PSKeys.EMAIL:
-            email_prompt = (
-                f"Extract the email from the following text:\n{answer}"
-                f"\n\nOutput just the email. "
-                f"The email should be directly assignable to a string "
-                f"variable. No explanation is required. If you cannot "
-                f'extract the email, output "NA".'
-            )
-            structured_output[prompt_name] = LegacyExecutor._convert_scalar_answer(
-                answer, llm, answer_prompt_svc, email_prompt
-            )
+            if answer.lower() == "na":
+                structured_output[prompt_name] = answer
+            else:
+                email_prompt = (
+                    f"Extract the email from the following text:\n{answer}"
+                    f"\n\nOutput just the email. "
+                    f"The email should be directly assignable to a string "
+                    f"variable. No explanation is required. If you cannot "
+                    f'extract the email, output "NA".'
+                )
+                structured_output[prompt_name] = answer_prompt_svc.run_completion(
+                    llm=llm, prompt=email_prompt
+                )
 
         elif output_type == PSKeys.DATE:
             date_prompt = (

workers/executor/worker.py

@@ -5,6 +5,7 @@
 """
 
 import logging
+import os
 
 from shared.enums.worker_enums import WorkerType
 from shared.infrastructure.config.builder import WorkerBuilder
@@ -43,7 +44,9 @@ def check_executor_health():
                 "worker_type": "executor",
                 "registered_executors": executors,
                 "executor_count": len(executors),
-                "queues": ["celery_executor_legacy"],
+                "queues": os.environ.get(
+                    "CELERY_QUEUES_EXECUTOR", "celery_executor_legacy"
+                ).split(","),
             },
         )
 

workers/file_processing/structure_tool_task.py

@@ -418,8 +418,13 @@ def _execute_structure_tool_impl(params: dict) -> dict:
         logger.info("Overwriting INFILE with structured output: %s", input_file_path)
         fs.json_dump(path=input_file_path, data=structured_output)
 
-        logger.info("Output written successfully to workflow storage")
-    except (OSError, json.JSONDecodeError) as e:
+        logger.info(
+            "Output files written — INFILE=%s, METADATA_DIR=%s",
+            input_file_path,
+            execution_data_dir,
+        )
+    except Exception as e:
+        logger.error("Failed to write output files: %s", e, exc_info=True)
         return ExecutionResult.failure(error=f"Error writing output file: {e}").to_dict()
 
     # Write tool result + tool_metadata to METADATA.json

workers/shared/enums/worker_enums_base.py

@@ -155,6 +155,7 @@ class QueueName(str, Enum):
     # The dispatcher derives queue names as ``celery_executor_{executor_name}``.
     # The "legacy" executor is the default OSS executor.
     EXECUTOR = "celery_executor_legacy"
+    EXECUTOR_AGENTIC = "celery_executor_agentic"
 
     # IDE callback queue (prompt studio post-execution callbacks)
     IDE_CALLBACK = "ide_callback"

workers/shared/infrastructure/config/registry.py

@@ -66,6 +66,7 @@ class WorkerRegistry:
         ),
         WorkerType.EXECUTOR: WorkerQueueConfig(
             primary_queue=QueueName.EXECUTOR,
+            additional_queues=[QueueName.EXECUTOR_AGENTIC],
         ),
         WorkerType.IDE_CALLBACK: WorkerQueueConfig(
             primary_queue=QueueName.IDE_CALLBACK,

workers/shared/workflow/destination_connector.py

@@ -658,6 +658,18 @@ def _handle_filesystem_destination(
     ):
         """Handle filesystem destination processing."""
         if not result.has_hitl:
+            if not result.tool_execution_result and not file_ctx.execution_error:
+                error_msg = (
+                    f"No tool execution result for file '{file_ctx.file_name}' "
+                    f"- filesystem copy skipped"
+                )
+                logger.error(error_msg)
+                log_file_error(
+                    exec_ctx.workflow_log,
+                    exec_ctx.file_execution_id,
+                    f"❌ {error_msg}",
+                )
+                raise RuntimeError(error_msg)
             log_file_info(
                 exec_ctx.workflow_log,
                 exec_ctx.file_execution_id,
@@ -696,9 +708,17 @@ def _handle_database_destination(
                     api_client=exec_ctx.api_client,
                 )
             else:
-                logger.warning(
-                    f"No tool execution result or execution error found for file {file_ctx.file_name}, skipping database insertion"
+                error_msg = (
+                    f"No tool execution result for file '{file_ctx.file_name}' "
+                    f"- database insertion skipped"
+                )
+                logger.error(error_msg)
+                log_file_error(
+                    exec_ctx.workflow_log,
+                    exec_ctx.file_execution_id,
+                    f"❌ {error_msg}",
                 )
+                raise RuntimeError(error_msg)
         else:
             logger.info(
                 f"File '{file_ctx.file_name}' sent to HITL queue - DATABASE processing will be handled after review"
@@ -1403,9 +1423,18 @@ def get_tool_execution_result_from_execution_context(
             file_storage = file_system.get_file_storage()
 
             if not metadata_file_path:
+                logger.warning(
+                    "No metadata_file_path (file_execution_id=%s, execution_dir=%s)",
+                    file_execution_id,
+                    file_handler.execution_dir,
+                )
                 return None
 
             if not file_storage.exists(metadata_file_path):
+                logger.warning(
+                    "METADATA.json not found at '%s'",
+                    metadata_file_path,
+                )
                 return None
 
             metadata_content = file_storage.read(path=metadata_file_path, mode="r")
@@ -1418,9 +1447,17 @@ def get_tool_execution_result_from_execution_context(
             output_file_path = file_handler.infile
 
             if not output_file_path:
+                logger.warning(
+                    "No infile path (file_execution_id=%s)",
+                    file_execution_id,
+                )
                 return None
 
             if not file_storage.exists(output_file_path):
+                logger.warning(
+                    "INFILE not found at '%s'",
+                    output_file_path,
+                )
                 return None
 
             file_type = file_storage.mime_type(path=output_file_path)

workers/tests/test_answer_prompt.py

@@ -562,8 +562,8 @@ def test_invalid_strategy_skips_retrieval(
         )
         result = executor._handle_answer_prompt(ctx)
 
-        # Answer stays "NA" which gets sanitized to None
-        assert result.data[PSKeys.OUTPUT]["field_a"] is None
+        # Answer stays "NA" (top-level NA is preserved, not sanitized)
+        assert result.data[PSKeys.OUTPUT]["field_a"] == "NA"
 
 
 class TestHandleAnswerPromptMultiPrompt:
@@ -687,21 +687,21 @@ def test_vectordb_closed(self, mock_shim_cls, mock_deps):
 class TestNullSanitization:
     """Tests for _sanitize_null_values."""
 
-    def test_na_string_becomes_none(self):
-        """Top-level 'NA' string → None."""
+    def test_na_string_preserved(self):
+        """Top-level 'NA' string is preserved (not sanitized to None)."""
         from executor.executors.legacy_executor import LegacyExecutor
 
         output = {"field": "NA"}
         result = LegacyExecutor._sanitize_null_values(output)
-        assert result["field"] is None
+        assert result["field"] == "NA"
 
-    def test_na_case_insensitive(self):
-        """'na' (lowercase) → None."""
+    def test_na_case_insensitive_preserved(self):
+        """Top-level 'na' (lowercase) is preserved (not sanitized to None)."""
         from executor.executors.legacy_executor import LegacyExecutor
 
         output = {"field": "na"}
         result = LegacyExecutor._sanitize_null_values(output)
-        assert result["field"] is None
+        assert result["field"] == "na"
 
     def test_nested_list_na(self):
         """NA in nested list items → None."""