From b06ac05e2bfb035368b9ec142ad9b0455e85a671 Mon Sep 17 00:00:00 2001 From: pk-zipstack Date: Thu, 16 Jul 2026 20:07:25 +0530 Subject: [PATCH 1/5] UN-3575 [FEAT] Thread lookup_config into agentic-table dispatch Pass the per-prompt lookup_config into the agentic_table executor's params so post-extraction lookup enrichment runs for agentic-table prompts on the deployment path. Sourced from the output dict (attached upstream at export); None when no lookup is assigned, so it's a no-op otherwise. Co-Authored-By: Claude Opus 4.8 (1M context) --- workers/file_processing/structure_tool_task.py | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/workers/file_processing/structure_tool_task.py b/workers/file_processing/structure_tool_task.py index cfa0e316a6..341ca5294a 100644 --- a/workers/file_processing/structure_tool_task.py +++ b/workers/file_processing/structure_tool_task.py @@ -473,6 +473,10 @@ def _execute_structure_tool_impl(params: dict) -> dict: "parallel_pages": at_settings.get("parallel_pages", 4), "execution_id": execution_id, "PLATFORM_SERVICE_API_KEY": platform_service_api_key, + # Lookup enrichment config, attached to the output upstream when a + # lookup is assigned to this agentic-table prompt. None otherwise — + # the agentic_table executor runs enrichment only when set. + "lookup_config": at_output.get("lookup_config"), } at_ctx = ExecutionContext( executor_name="agentic_table", From a820880f33fcea104baddfa13b8f3d7fbd2bc22c Mon Sep 17 00:00:00 2001 From: pk-zipstack Date: Thu, 16 Jul 2026 20:47:45 +0530 Subject: [PATCH 2/5] UN-3575 [FEAT] Add shared parallel_map bounded-concurrency util Length-preserving bounded-concurrency map used by per-row lookup enrichment (and reusable by other LLM call sites). max_workers=1 == sequential. Co-Authored-By: Claude Opus 4.8 (1M context) --- workers/shared/parallel_map.py | 94 ++++++++++++++++++++++++++++++++++ 1 file changed, 94 insertions(+) create mode 100644 workers/shared/parallel_map.py diff --git a/workers/shared/parallel_map.py b/workers/shared/parallel_map.py new file mode 100644 index 0000000000..30d240baea --- /dev/null +++ b/workers/shared/parallel_map.py @@ -0,0 +1,94 @@ +"""Generic bounded-concurrency map for LLM (and other I/O-bound) call sites. + +Runs ``worker(item)`` across a thread pool capped at ``max_workers`` and returns +results in input order. Unlike the agentic extractor's older ``parallel_page_map`` +(which drops ``None`` results), this is **length-preserving**: the output list is +always the same length as the input, so callers can realign results to inputs by +index — essential for per-row enrichment where row *i* in must map to row *i* out. + +``max_workers=1`` runs effectively sequentially, so a single knob covers both the +sequential and bounded-parallel strategies with no code change. + +Rate limiting / retries are intentionally NOT handled here — the SDK LLM client +already retries transient errors (429/500/503/... and provider "overloaded") +with backoff and honors ``Retry-After``, so ``max_workers`` concurrent callers +self-throttle without any extra coordination in this utility. + +NOTE: there is no early-abort. All submitted work runs to completion even if the +caller is later cancelled or times out — in-flight threads cannot be killed. +Callers that need to stop a doomed run early must gate submission themselves. +""" + +import logging +import threading +from collections.abc import Callable +from concurrent.futures import ThreadPoolExecutor, as_completed +from typing import TypeVar + +logger = logging.getLogger(__name__) + +T = TypeVar("T") +R = TypeVar("R") + + +def parallel_map( + items: list[T], + worker: Callable[[T], R], + *, + max_workers: int, + on_error: Callable[[int, T, Exception], R] | None = None, + label: str = "", +) -> list[R]: + """Apply ``worker`` to each item with bounded concurrency, order preserved. + + Args: + items: Inputs to process. An empty list returns ``[]``. + worker: Called as ``worker(item)`` for each item. + max_workers: Thread-pool cap. ``<= 1`` runs effectively sequentially. + on_error: Called as ``on_error(index, item, exception)`` to produce a + fallback result when a worker raises. If omitted, a failed item's + slot is left as ``None`` (still counted — length is preserved). + label: Optional label for the progress log line. + + Returns: + A list the SAME LENGTH as ``items``, results in input order. Failed + items hold either the ``on_error`` fallback or ``None``. + """ + if not items: + return [] + + n = len(items) + effective_workers = max(max_workers, 1) + results: list[R | None] = [None] * n + log_lock = threading.Lock() + + if effective_workers > 1: + suffix = f" ({label})" if label else "" + logger.info( + "parallel_map: %d items across up to %d workers%s", + n, + effective_workers, + suffix, + ) + + with ThreadPoolExecutor(max_workers=effective_workers) as executor: + future_to_idx = { + executor.submit(worker, item): idx for idx, item in enumerate(items) + } + for future in as_completed(future_to_idx): + idx = future_to_idx[future] + try: + results[idx] = future.result() + except Exception as e: + with log_lock: + logger.error( + "parallel_map: item %d/%d failed: %s", + idx + 1, + n, + e, + exc_info=True, + ) + if on_error is not None: + results[idx] = on_error(idx, items[idx], e) + + return results From e2c922f6d2fd478bb47e313e1877d1600ee4f4e2 Mon Sep 17 00:00:00 2001 From: pk-zipstack Date: Tue, 21 Jul 2026 20:02:21 +0530 Subject: [PATCH 3/5] UN-3575 [FIX] Address review on parallel_map + dispatch comment parallel_map: return type is now list[R | None] (matches the None-on-omitted- on_error contract); drop the needless log lock (loop body is single-threaded); cap per-item tracebacks and add a failure summary so a systemic failure can't flood logs; guard on_error so a raising fallback can't hang the pool. Docstring corrected (drop cloud-only symbol reference; don't overstate SDK retry coverage). structure_tool_task: collapse the lookup_config comment and name the source. Co-Authored-By: Claude Opus 4.8 (1M context) --- .../file_processing/structure_tool_task.py | 5 +- workers/shared/parallel_map.py | 72 +++++++++++++------ 2 files changed, 52 insertions(+), 25 deletions(-) diff --git a/workers/file_processing/structure_tool_task.py b/workers/file_processing/structure_tool_task.py index de543b2fbd..35cd3f8130 100644 --- a/workers/file_processing/structure_tool_task.py +++ b/workers/file_processing/structure_tool_task.py @@ -474,9 +474,8 @@ def _execute_structure_tool_impl(params: dict) -> dict: "execution_id": execution_id, "PLATFORM_SERVICE_API_KEY": platform_service_api_key, "group_key": at_settings.get("group_key", ""), - # Lookup enrichment config, attached to the output upstream when a - # lookup is assigned to this agentic-table prompt. None otherwise — - # the agentic_table executor runs enrichment only when set. + # Set on the output at export by prompt_studio_registry_helper.py + # when a lookup is assigned; None otherwise. "lookup_config": at_output.get("lookup_config"), } at_ctx = ExecutionContext( diff --git a/workers/shared/parallel_map.py b/workers/shared/parallel_map.py index 30d240baea..6467034939 100644 --- a/workers/shared/parallel_map.py +++ b/workers/shared/parallel_map.py @@ -1,18 +1,17 @@ """Generic bounded-concurrency map for LLM (and other I/O-bound) call sites. Runs ``worker(item)`` across a thread pool capped at ``max_workers`` and returns -results in input order. Unlike the agentic extractor's older ``parallel_page_map`` -(which drops ``None`` results), this is **length-preserving**: the output list is -always the same length as the input, so callers can realign results to inputs by -index — essential for per-row enrichment where row *i* in must map to row *i* out. +results in input order. It is **length-preserving**: the output list is always +the same length as the input and failed items keep their slot, so callers can +realign results to inputs by index (row *i* in maps to row *i* out). ``max_workers=1`` runs effectively sequentially, so a single knob covers both the sequential and bounded-parallel strategies with no code change. -Rate limiting / retries are intentionally NOT handled here — the SDK LLM client -already retries transient errors (429/500/503/... and provider "overloaded") -with backoff and honors ``Retry-After``, so ``max_workers`` concurrent callers -self-throttle without any extra coordination in this utility. +Rate limiting / retries are intentionally NOT handled here. Bounding +``max_workers`` caps concurrent calls; anything finer (provider rate limits, +``Retry-After``, backoff on 429/5xx/overloaded) is the LLM client's job, not +this generic utility's. Size ``max_workers`` conservatively for the provider. NOTE: there is no early-abort. All submitted work runs to completion even if the caller is later cancelled or times out — in-flight threads cannot be killed. @@ -20,7 +19,6 @@ """ import logging -import threading from collections.abc import Callable from concurrent.futures import ThreadPoolExecutor, as_completed from typing import TypeVar @@ -30,6 +28,11 @@ T = TypeVar("T") R = TypeVar("R") +# Cap on how many per-item failures are logged with a full traceback before +# switching to one-line records — stops a dead adapter from emitting hundreds +# of identical stack traces (and Sentry events) for one logical failure. +_MAX_TRACEBACKS = 5 + def parallel_map( items: list[T], @@ -38,7 +41,7 @@ def parallel_map( max_workers: int, on_error: Callable[[int, T, Exception], R] | None = None, label: str = "", -) -> list[R]: +) -> list[R | None]: """Apply ``worker`` to each item with bounded concurrency, order preserved. Args: @@ -48,11 +51,14 @@ def parallel_map( on_error: Called as ``on_error(index, item, exception)`` to produce a fallback result when a worker raises. If omitted, a failed item's slot is left as ``None`` (still counted — length is preserved). + Prefer passing ``on_error`` so a failed slot is distinguishable + from a legitimate ``None`` result. label: Optional label for the progress log line. Returns: - A list the SAME LENGTH as ``items``, results in input order. Failed - items hold either the ``on_error`` fallback or ``None``. + A list the SAME LENGTH as ``items``, results in input order. A failed + item holds the ``on_error`` fallback, or ``None`` when ``on_error`` is + omitted — hence the ``R | None`` element type. """ if not items: return [] @@ -60,7 +66,7 @@ def parallel_map( n = len(items) effective_workers = max(max_workers, 1) results: list[R | None] = [None] * n - log_lock = threading.Lock() + failures = 0 if effective_workers > 1: suffix = f" ({label})" if label else "" @@ -71,6 +77,8 @@ def parallel_map( suffix, ) + # The as_completed loop body runs on the calling thread (only worker() runs + # in the pool), so results/logging here need no locking. with ThreadPoolExecutor(max_workers=effective_workers) as executor: future_to_idx = { executor.submit(worker, item): idx for idx, item in enumerate(items) @@ -80,15 +88,35 @@ def parallel_map( try: results[idx] = future.result() except Exception as e: - with log_lock: - logger.error( - "parallel_map: item %d/%d failed: %s", - idx + 1, - n, - e, - exc_info=True, - ) + failures += 1 + # Full traceback for the first few, then one-liners so a + # systemic failure doesn't flood logs. + logger.error( + "parallel_map: item %d/%d failed: %s", + idx + 1, + n, + e, + exc_info=failures <= _MAX_TRACEBACKS, + ) if on_error is not None: - results[idx] = on_error(idx, items[idx], e) + # Guard on_error itself: if it raises, it would propagate + # through the pool's __exit__ (shutdown(wait=True)) and hang + # the caller until every in-flight task drains, with no log. + try: + results[idx] = on_error(idx, items[idx], e) + except Exception: + logger.error( + "parallel_map: on_error raised for item %d/%d", + idx + 1, + n, + exc_info=True, + ) + if failures: + logger.warning( + "parallel_map: %d/%d items failed%s", + failures, + n, + f" ({label})" if label else "", + ) return results From c4634047fc6507d44a081434fd0103869213e4a4 Mon Sep 17 00:00:00 2001 From: pk-zipstack Date: Thu, 23 Jul 2026 14:00:05 +0530 Subject: [PATCH 4/5] UN-3575 [FIX] Let callers preload lookup reference texts (per-row read hoist) Add preload_reference_texts() and a reference_texts passthrough on run_lookup_enrichment so per-row enrichment can read the reference file(s) once instead of once per row. None => each call loads its own (unchanged path). Co-Authored-By: Claude Opus 4.8 (1M context) --- .../executor/executors/lookup_enrichment.py | 23 +++++++++++++++++++ 1 file changed, 23 insertions(+) diff --git a/workers/executor/executors/lookup_enrichment.py b/workers/executor/executors/lookup_enrichment.py index 0b224badb5..39c3d1eadf 100644 --- a/workers/executor/executors/lookup_enrichment.py +++ b/workers/executor/executors/lookup_enrichment.py @@ -32,6 +32,24 @@ def is_blank(value: Any) -> bool: return False +def preload_reference_texts(lookup_config: dict[str, Any] | None) -> Any: + """Read a lookup's reference files once via the enrichment plugin. + + For callers that enrich many rows against one ``lookup_config`` (per-row + enrichment), pass the result as ``reference_texts`` to + ``run_lookup_enrichment`` so the remote read isn't repeated per row. + Returns ``None`` (fall back to per-call loading) when the plugin/config is + unavailable or the read fails. + """ + lookup_cls = ExecutorPluginLoader.get("lookup-enrichment") + if not (lookup_config and lookup_cls): + return None + preload = getattr(lookup_cls, "preload_reference_texts", None) + if preload is None: + return None + return preload(lookup_config) + + def run_lookup_enrichment( output: dict[str, Any], structured_output: dict[str, Any], @@ -40,11 +58,15 @@ def run_lookup_enrichment( shim: Any, llm_cls: Any, usage_kwargs: dict[str, Any] | None = None, + reference_texts: dict[str, str] | None = None, ) -> list[dict[str, Any]]: """Run lookup enrichment plugin if enabled and available. Returns any usage records the plugin emitted (recovered even on plugin failure) so the caller can extend its billing batch. + + ``reference_texts`` (from :func:`preload_reference_texts`) skips the + plugin's per-call remote read of the reference files. """ prompt_name = output[PSKeys.NAME] current_value = structured_output.get(prompt_name) @@ -73,6 +95,7 @@ def run_lookup_enrichment( prompt_name=prompt_name, shim=shim, usage_kwargs=usage_kwargs, + reference_texts=reference_texts, ) metrics.setdefault(prompt_name, {})[lookup_cls.METRICS_KEY] = outcome.llm_metrics except Exception: From f2bf6401461d4f2c346d9c69f1e33eabac14aec1 Mon Sep 17 00:00:00 2001 From: pk-zipstack Date: Thu, 23 Jul 2026 15:24:27 +0530 Subject: [PATCH 5/5] UN-3575 [FIX] Guard preload_reference_texts against plugin exceptions Honour the documented None-on-failure contract: wrap the plugin's preload call in try/except (and use callable()) so a raising plugin hook falls back to per-call reference loading instead of aborting enrichment. Co-Authored-By: Claude Opus 4.8 (1M context) --- workers/executor/executors/lookup_enrichment.py | 10 ++++++++-- 1 file changed, 8 insertions(+), 2 deletions(-) diff --git a/workers/executor/executors/lookup_enrichment.py b/workers/executor/executors/lookup_enrichment.py index 39c3d1eadf..3cfbe6b767 100644 --- a/workers/executor/executors/lookup_enrichment.py +++ b/workers/executor/executors/lookup_enrichment.py @@ -45,9 +45,15 @@ def preload_reference_texts(lookup_config: dict[str, Any] | None) -> Any: if not (lookup_config and lookup_cls): return None preload = getattr(lookup_cls, "preload_reference_texts", None) - if preload is None: + if not callable(preload): + return None + try: + return preload(lookup_config) + except Exception: + # Honour the None-on-failure contract even if the plugin's preload + # raises — the caller falls back to per-call reference loading. + logger.exception("Failed to preload lookup reference texts") return None - return preload(lookup_config) def run_lookup_enrichment(