[FIX] Executor Queue for Agentic extraction (#1893)

harini-venkataraman · Deepak-Kesavan · pre-commit-ci[bot] · web-flow · commit 7983c98fc1b8 · 2026-04-03T12:41:40.000+05:30
* Execution backend - revamp * async flow * Streaming progress to FE * Removing multi hop in Prompt studio ide and structure tool * UN-3234 [FIX] Add beta tag to agentic prompt studio navigation item * Added executors for agentic prompt studio * Added executors for agentic prompt studio * Removed redundant envs * Removed redundant envs * Removed redundant envs * [pre-commit.ci] auto fixes from pre-commit.com hooks for more information, see https://pre-commit.ci * Removed redundant envs * Removed redundant envs * Removed redundant envs * Removed redundant envs * Removed redundant envs * Removed redundant envs * Removed redundant envs * Removed redundant envs * [pre-commit.ci] auto fixes from pre-commit.com hooks for more information, see https://pre-commit.ci * Removed redundant envs * adding worker for callbacks * adding worker for callbacks * [pre-commit.ci] auto fixes from pre-commit.com hooks for more information, see https://pre-commit.ci * adding worker for callbacks * adding worker for callbacks * adding worker for callbacks * [pre-commit.ci] auto fixes from pre-commit.com hooks for more information, see https://pre-commit.ci * Pluggable apps and plugins to fit the new async prompt execution architecture * [pre-commit.ci] auto fixes from pre-commit.com hooks for more information, see https://pre-commit.ci * Pluggable apps and plugins to fit the new async prompt execution architecture * [pre-commit.ci] auto fixes from pre-commit.com hooks for more information, see https://pre-commit.ci * Pluggable apps and plugins to fit the new async prompt execution architecture * adding worker for callbacks * adding worker for callbacks * adding worker for callbacks * adding worker for callbacks * adding worker for callbacks * adding worker for callbacks * adding worker for callbacks * adding worker for callbacks * fix: write output files in agentic extraction pipeline Agentic extraction returned early without writing INFILE (JSON) or METADATA.json, causing destination connectors to read the original PDF and fail with "Expected tool output type: TXT, got: application/pdf". Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com> * UN-3266 fix: replace hardcoded /tmp paths with secure temp dirs in tests (#1850) * UN-3266 fix: replace hardcoded /tmp paths with secure temp dirs in tests Replace hardcoded /tmp/ paths (SonarCloud S5443 security hotspots) with pytest's tmp_path fixture or module-level tempfile.mkdtemp() constants in all affected test files to avoid world-writable directory vulnerabilities. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com> * [pre-commit.ci] auto fixes from pre-commit.com hooks for more information, see https://pre-commit.ci --------- Co-authored-by: Claude Sonnet 4.6 <noreply@anthropic.com> Co-authored-by: pre-commit-ci[bot] <66853113+pre-commit-ci[bot]@users.noreply.github.com> * [pre-commit.ci] auto fixes from pre-commit.com hooks for more information, see https://pre-commit.ci * Update docs * UN-3266 fix: remove dead code with undefined names in fetch_response Remove unreachable code block after the async callback return in fetch_response that still referenced output_count_before and response from the old synchronous implementation, causing ruff F821 errors. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com> * Un 3266 fix security hotspot tmp paths (#1851) * UN-3266 fix: replace hardcoded /tmp paths with secure temp dirs in tests Replace hardcoded /tmp/ paths (SonarCloud S5443 security hotspots) with pytest's tmp_path fixture or module-level tempfile.mkdtemp() constants in all affected test files to avoid world-writable directory vulnerabilities. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com> * UN-3266 fix: resolve ruff linting failures across multiple files - B026: pass url positionally in worker_celery.py to avoid star-arg after keyword - N803: rename MockAsyncResult to mock_async_result in test_tasks.py - E501/I001: fix long line and import sort in llm_whisperer helper - ANN401: replace Any with object|None in dispatcher.py; add noqa in test helpers - F841: remove unused workflow_id and result assignments Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com> * [pre-commit.ci] auto fixes from pre-commit.com hooks for more information, see https://pre-commit.ci --------- Co-authored-by: Claude Sonnet 4.6 <noreply@anthropic.com> Co-authored-by: pre-commit-ci[bot] <66853113+pre-commit-ci[bot]@users.noreply.github.com> * UN-3266 fix: resolve SonarCloud bugs S2259 and S1244 in PR #1849 - S2259: guard against None after _discover_plugins() in loader.py to satisfy static analysis on the dict[str,type]|None field type - S1244: replace float equality checks with pytest.approx() in test_answer_prompt.py and test_phase2h.py Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com> * UN-3266 fix: resolve SonarCloud code smells in PR #1849 - S5799: Merge all implicit string concatenations in log messages (legacy_executor.py, tasks.py, dispatcher.py, orchestrator.py, registry.py, variable_replacement.py, structure_tool_task.py) - S1192: Extract duplicate literal to _NO_CELERY_APP_MSG constant in dispatcher.py - S1871: Merge identical elif/else branches in tasks.py and test_sanity_phase6j.py - S1186: Add comment to empty stub method in test_sanity_phase6a.py - S1481: Remove unused local variables in test_sanity_phase6d/e/f/g/h/j and test_phase5d.py - S117: Rename PascalCase local variables to snake_case in test_sanity_phase3/5/6i.py - S5655: Broaden tool type annotation to StreamMixin in IndexingUtils.generate_index_key and PlatformHelper.get_adapter_config - docker:S7031: Merge consecutive RUN instructions in worker-unified.Dockerfile - javascript:S1128: Remove unused pollForCompletion import in usePromptRun.js Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com> * [pre-commit.ci] auto fixes from pre-commit.com hooks for more information, see https://pre-commit.ci * UN-3266 fix: wrap long log message in dispatcher.py to fix E501 Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com> * UN-3266 fix: resolve remaining SonarCloud S117 naming violations Rename PascalCase local variables to snake_case to comply with S117: - legacy_executor.py: rename tuple-unpacked _get_prompt_deps() results (AnswerPromptService→answer_prompt_svc, RetrievalService→retrieval_svc, VariableReplacementService→variable_replacement_svc, LLM→llm_cls, EmbeddingCompat→embedding_compat_cls, VectorDB→vector_db_cls) and update all downstream usages including _apply_type_conversion and _handle_summarize - test_phase1_log_streaming.py: rename Mock* local variables to mock_* snake_case equivalents - test_sanity_phase3.py: rename MockDispatcher→mock_dispatcher_cls and MockShim→mock_shim_cls across all 10 test methods - test_sanity_phase5.py: rename MockShim→mock_shim, MockX2Text→mock_x2text in 6 test methods; MockDispatcher→mock_dispatcher_cls in dispatch test; fix LLM_cls→llm_cls, EmbeddingCompat→embedding_compat_cls, VectorDB→vector_db_cls in _mock_prompt_deps helper Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com> * [pre-commit.ci] auto fixes from pre-commit.com hooks for more information, see https://pre-commit.ci * UN-3266 fix: resolve remaining SonarCloud code smells in PR #1849 - test_sanity_phase2/4.py, test_answer_prompt.py: rename PascalCase local variables in _mock_prompt_deps/_mock_deps to snake_case (RetrievalService→retrieval_svc, VariableReplacementService→ variable_replacement_svc, Index→index_cls, LLM_cls→llm_cls, EmbeddingCompat→embedding_compat_cls, VectorDB→vector_db_cls, AnswerPromptService→answer_prompt_svc_cls) — fixes S117 - test_sanity_phase3.py: remove unused local variable "result" — fixes S1481 - structure_tool_task.py: remove redundant json.JSONDecodeError from except clause (subclass of ValueError) — fixes S5713 - shared/workflow/execution/service.py: replace generic Exception with RuntimeError for structure tool failure — fixes S112 - run-worker-docker.sh: define EXECUTOR_WORKER_TYPE constant and replace 10 literal "executor" occurrences — fixes S1192 Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com> * UN-3266 fix: resolve SonarCloud cognitive complexity and code smell violations - Reduce cognitive complexity in answer_prompt.py: - Extract _build_grammar_notes, _run_webhook_postprocess helpers - _is_safe_public_url: extracted _resolve_host_addresses helper - handle_json: early-return pattern eliminates nesting - construct_prompt: delegates grammar loop to _build_grammar_notes - Reduce cognitive complexity in legacy_executor.py: - Extract _execute_single_prompt, _run_table_extraction helpers - Extract _run_challenge_if_enabled, _run_evaluation_if_enabled - Extract _inject_table_settings, _finalize_pipeline_result - Extract _convert_number_answer, _convert_scalar_answer - Extract _sanitize_dict_values helper - _handle_answer_prompt CC reduced from 50 to ~7 - Reduce CC in structure_tool_task.py: guard-clause refactor - Reduce CC in backend: dto.py, deployment_helper.py, api_deployment_views.py, prompt_studio_helper.py - Fix S117: rename PascalCase local vars in test_answer_prompt.py - Fix S1192: extract EXECUTOR_WORKER_TYPE constant in run-worker.sh - Fix S1172: remove unused params from structure_tool_task.py - Fix S5713: remove redundant JSONDecodeError in json_repair_helper.py - Fix S112/S5727 in test_execution.py Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com> * UN-3266 fix: remove unused RetrievalStrategy import from _handle_answer_prompt Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com> * [pre-commit.ci] auto fixes from pre-commit.com hooks for more information, see https://pre-commit.ci * UN-3266 fix: rename UsageHelper params to lowercase (N803) Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com> * UN-3266 fix: resolve remaining SonarCloud issues from check run 66691002192 - Add @staticmethod to _sanitize_null_values (fixes S2325 missing self) - Reduce _execute_single_prompt params from 25 to 11 (S107) by grouping services as deps tuple and extracting exec params from context.executor_params - Add NOSONAR suppression for raise exc in test helper (S112) Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com> * [pre-commit.ci] auto fixes from pre-commit.com hooks for more information, see https://pre-commit.ci * UN-3266 fix: remove unused locals in _handle_answer_prompt (F841) execution_id, file_hash, log_events_id, custom_data are now extracted inside _execute_single_prompt from context.executor_params. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com> * fix: resolve Biome linting errors in frontend source files Auto-fixed 48 lint errors across 56 files: import ordering, block statements, unused variable prefixing, and formatting issues. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com> * fix: replace dynamic import of SharePermission with static import in Workflows Resolves vite build warning about SharePermission.jsx being both dynamically and statically imported across the codebase. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com> * fix: resolve SonarCloud warnings in frontend components - Remove unnecessary try-catch around PostHog event calls - Flip negated condition in PromptOutput.handleTable for clarity Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com> * [pre-commit.ci] auto fixes from pre-commit.com hooks for more information, see https://pre-commit.ci * Address PR #1849 review comments: fix null guards, dead code, and test drift - Remove redundant inline `import uuid as _uuid` in views.py (use module-level uuid) - URL-encode DB_USER in worker_celery.py result backend connection string - Remove misleading task_queues=[Queue("executor")] from dispatch-only Celery app - Remove dead `if not tool:` guards after objects.get() (already raises DoesNotExist) - Move profile_manager/default_profile null checks before first dereference - Reorder ProfileManager.objects.get before mark_document_indexed in tasks.py - Handle ProfileManager.DoesNotExist as warning, not hard failure - Wrap PostHog analytics in try/catch so failures don't block prompt execution - Handle pending-indexing 200 response in usePromptRun.js (clear RUNNING status) - Reset formData when metadata is missing in ConfigureDs.jsx - Fix test_should_skip_extraction tests: function now takes 1 arg (outputs only) - Fix agentic routing tests: mock X2Text.process, remove stale platform_helper kwarg Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com> * [pre-commit.ci] auto fixes from pre-commit.com hooks for more information, see https://pre-commit.ci * Fix missing llm_usage_reason for summarize LLM usage tracking Add PSKeys.LLM_USAGE_REASON to usage_kwargs in _handle_summarize() so summarization costs appear under summarize_llm in API response metadata. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com> * UN-3266 [FIX] Fix single-pass extraction routing in LegacyExecutor - Route _handle_structure_pipeline to _handle_single_pass_extraction when is_single_pass=True (was always calling _handle_answer_prompt) - Delegate _handle_single_pass_extraction to cloud plugin via ExecutorRegistry, falling back to _handle_answer_prompt if plugin not installed Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com> * Fixing API depployment response mismatches * Fix single-pass extraction showing only one prompt result in real-time - Fix accumulation bug in usePromptOutput updateCoverage() where each loop iteration spread the original promptOutputs instead of the accumulating updatedPromptOutputs, causing only the last prompt to render - Improve index success toast to show document name - Strip adapter names from index key configs for consistent hashing - Update sdk1 uv.lock Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com> * Move summarize from sync Django plugin to executor worker for IDE index When "Summarize" is enabled and a user indexes a new document in Prompt Studio, the backend returned a 500 because the sync Django plugin tried to read the extracted .txt file before extraction happened in the worker. Fix: defer summarization to the executor worker's _handle_ide_index (extract → summarize → index), build summarize_params in the payload, and track the summarize index in the ide_index_complete callback. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com> * Address PR #1849 review comments: null guards, thread safety - Fix null guard ordering in build_index_payload: add DefaultProfileError check before calling validators on default_profile - Fix null guard ordering in _fetch_single_pass_response: move check before .llm.id access and validators (was dead code after dereference) - Add threading.Lock to worker_celery singleton to prevent race under concurrent gunicorn threads Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com> * Add documentation to ExecutionResponse DTO describing result structure Documents the ExecutionResponse dataclass fields, especially the result attribute's per-file dict structure (output, metadata, metrics, error keys) as requested in PR review. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com> * Fix PR review issues: IDOR, null guards, rollback, spinner, summarize, prompt lookup - Strip result payload from task_status to prevent IDOR data leak - Move null guard before validators in build_fetch_response_payload - Roll back indexing flag on broker failure in index_document - Use explicit IDs for spinner clearing instead of ORM serialization format - Catch broad exceptions in summarize tracking to prevent false failures - Guard prompt_id lookup in fetch_response with 400/404 responses Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com> * [pre-commit.ci] auto fixes from pre-commit.com hooks for more information, see https://pre-commit.ci * Fix CI, tests, and add async prompt studio improvements - Fix biome import ordering in frontend test files - Fix 22 stale backend test assertions in test_tasks.py (patch targets, return values, view dispatch pattern, remove TestCeleryConfig) - Add ide_prompt_complete callback tests - Add frontend vitest config and regression tests for null guards, stale closures, and single-pass loading guard - Add LLMCompat llama-index compatibility wrapper in SDK1 - Add litellm cohere embed timeout monkey-patch (v1.82.3) - Improve S3 filesystem helper (region_name, empty credential handling) - Add RetrieverLLM and lazy LLM creation for retrievers - Improve worker cleanup: api_client.close() in finally blocks, early return on setup failure in API deployment tasks - Add workers conftest.py for test environment setup Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com> * [pre-commit.ci] auto fixes from pre-commit.com hooks for more information, see https://pre-commit.ci * Fix pre-existing biome CI errors: import ordering and formatting Auto-fix 18 pre-existing organizeImports and formatting errors across 17 frontend files that were blocking biome ci on the entire codebase. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com> * Fix ruff F821: add missing transaction import in prompt_studio_helper Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com> * Add input validation guards to bulk_fetch_response endpoint Validate prompt_ids (non-empty), document_id (required), and handle DoesNotExist for both prompts and document to return proper 400/404 instead of dispatching no-op tasks or raising unhandled 500 errors. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com> * IDE Call backs * Sonar issues fix * [pre-commit.ci] auto fixes from pre-commit.com hooks for more information, see https://pre-commit.ci * Fix ruff errors: restore summary_profile variable, suppress TC001 in dispatcher Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com> * [pre-commit.ci] auto fixes from pre-commit.com hooks for more information, see https://pre-commit.ci * Update bun.lock to match package.json dependency ranges Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com> * Fix all biome lint warnings: empty blocks, missing braces, forEach returns, unused vars Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com> * Move ExecutionContext import into TYPE_CHECKING block Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com> * [pre-commit.ci] auto fixes from pre-commit.com hooks for more information, see https://pre-commit.ci * Fix SonarQube issues: duplication, naming, nesting, unused var - Extract _parse_json_body() helper and _ERR_INVALID_JSON constant to deduplicate 5 identical JSON parsing blocks in internal_views.py - Rename `User` local variable to `user_model` (naming convention) - Merge _emit_result/_emit_error into unified _emit_event() in ide_callback tasks to reduce code duplication - Extract _get_task_error() helper to deduplicate AsyncResult error retrieval in ide_index_error and ide_prompt_error - Remove unused `mock_ar_cls` variable in test_ide_callback.py - Add security note documenting why @csrf_exempt is safe on internal endpoints Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com> * Replace worker-ide-callback Dockerfile with worker-unified The IDE callback worker should use the unified worker image (worker-unified.Dockerfile) consistent with all other v2 workers. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com> * Add celery_executor_agentic queue to executor worker The executor worker only consumed from celery_executor_legacy, but the agentic prompt studio dispatches tasks to celery_executor_agentic. This caused agentic operations to sit in RabbitMQ with no consumer, resulting in timeouts and stuck-in-progress states. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com> * FIxing email enforce type * Removing line-item from select choices * Update workers/shared/enums/worker_enums_base.py Co-authored-by: greptile-apps[bot] <165735046+greptile-apps[bot]@users.noreply.github.com> Signed-off-by: harini-venkataraman <115449948+harini-venkataraman@users.noreply.github.com> * Update backend/workflow_manager/workflow_v2/workflow_helper.py Co-authored-by: greptile-apps[bot] <165735046+greptile-apps[bot]@users.noreply.github.com> Signed-off-by: harini-venkataraman <115449948+harini-venkataraman@users.noreply.github.com> * [pre-commit.ci] auto fixes from pre-commit.com hooks for more information, see https://pre-commit.ci * Fix false success logs and silent failures in ETL destination pipelines - Widen except clause in structure_tool_task to catch FileOperationError and log write paths for diagnostics - Add diagnostic logging at all silent return-None points in destination connector so missing INFILE/METADATA paths are visible in logs - Raise RuntimeError instead of silently skipping when no tool execution result is available for DB/FS destinations, preventing false success - Remove dead retry config from execute_bin task (max_retries=0) - Fix duplicate EXECUTOR/IDE_CALLBACK enum entries in WorkerType Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com> * [pre-commit.ci] auto fixes from pre-commit.com hooks for more information, see https://pre-commit.ci * Revert ETL destination pipeline changes — deferring to next cut Reverts diagnostic logging and error-raising changes in structure_tool_task.py and destination_connector.py. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com> --------- Signed-off-by: harini-venkataraman <115449948+harini-venkataraman@users.noreply.github.com> Co-authored-by: Ghost Jake <89829542+Deepak-Kesavan@users.noreply.github.com> Co-authored-by: pre-commit-ci[bot] <66853113+pre-commit-ci[bot]@users.noreply.github.com> Co-authored-by: Claude Sonnet 4.6 <noreply@anthropic.com> Co-authored-by: Ritwik G <100672805+ritwik-g@users.noreply.github.com> Co-authored-by: Kirtiman Mishra <110175055+kirtimanmishrazipstack@users.noreply.github.com> Co-authored-by: greptile-apps[bot] <165735046+greptile-apps[bot]@users.noreply.github.com>
diff --git a/backend/prompt_studio/prompt_studio_core_v2/static/select_choices.json b/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":{
diff --git a/backend/workflow_manager/workflow_v2/workflow_helper.py b/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,
diff --git a/docker/docker-compose.yaml b/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}
diff --git a/workers/executor/executors/legacy_executor.py b/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 = (
diff --git a/workers/executor/worker.py b/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(","),
             },
         )
 
diff --git a/workers/shared/enums/worker_enums_base.py b/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"
diff --git a/workers/shared/infrastructure/config/registry.py b/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,
diff --git a/workers/tests/test_answer_prompt.py b/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."""
