From ab98fc53b70cc11165d3061b0a0027b66c706202 Mon Sep 17 00:00:00 2001 From: Athul Date: Thu, 11 Jun 2026 13:38:44 +0530 Subject: [PATCH 1/7] UN-2771 Include text extraction time in API deployment metrics The structure tool timed indexing but not the text extraction (LLMWhisperer/X2Text) call, so API responses with include_metrics=True reported indexing time only. Time dynamic_extraction the same way and merge it into the result metrics as extraction.time_taken(s). Bump structure tool to 0.0.102. Co-Authored-By: Claude Fable 5 --- tools/structure/src/config/properties.json | 2 +- tools/structure/src/constants.py | 1 + tools/structure/src/main.py | 14 +++++++++++++- 3 files changed, 15 insertions(+), 2 deletions(-) diff --git a/tools/structure/src/config/properties.json b/tools/structure/src/config/properties.json index c8697e7307..a9d1a6029d 100644 --- a/tools/structure/src/config/properties.json +++ b/tools/structure/src/config/properties.json @@ -2,7 +2,7 @@ "schemaVersion": "0.0.1", "displayName": "Structure Tool", "functionName": "structure_tool", - "toolVersion": "0.0.101", + "toolVersion": "0.0.102", "description": "This is a template tool which can answer set of input prompts designed in the Prompt Studio", "input": { "description": "File that needs to be indexed and parsed for answers" diff --git a/tools/structure/src/constants.py b/tools/structure/src/constants.py index 8da6a5701a..294fb3015b 100644 --- a/tools/structure/src/constants.py +++ b/tools/structure/src/constants.py @@ -77,6 +77,7 @@ class SettingsKeys: TOOL = "tool" METRICS = "metrics" INDEXING = "indexing" + EXTRACTION = "extraction" EXECUTION_ID = "execution_id" IS_DIRECTORY_MODE = "is_directory_mode" LLM_PROFILE_ID = "llm_profile_id" diff --git a/tools/structure/src/main.py b/tools/structure/src/main.py index f68143a6c8..fa8ec39530 100644 --- a/tools/structure/src/main.py +++ b/tools/structure/src/main.py @@ -318,6 +318,7 @@ def run( ) extracted_text = "" + extraction_metrics = {} usage_kwargs: dict[Any, Any] = dict() if skip_extraction_and_indexing: self.stream_log( @@ -328,6 +329,7 @@ def run( usage_kwargs[UsageKwargs.RUN_ID] = self.file_execution_id usage_kwargs[UsageKwargs.FILE_NAME] = self.source_file_name usage_kwargs[UsageKwargs.EXECUTION_ID] = self.execution_id + extraction_start_time = datetime.datetime.now() extracted_text = STHelper.dynamic_extraction( file_path=input_file, enable_highlight=is_highlight_enabled, @@ -338,6 +340,13 @@ def run( tool=self, execution_run_data_folder=str(execution_run_data_folder), ) + extraction_metrics = { + SettingsKeys.EXTRACTION: { + "time_taken(s)": STHelper.elapsed_time( + start_time=extraction_start_time + ) + } + } index_metrics = {} if is_summarization_enabled: @@ -458,7 +467,10 @@ def run( "No text is extracted from the document to add to the metadata" ) if merged_metrics := self._merge_metrics( - structured_output.get(SettingsKeys.METRICS, {}), index_metrics + self._merge_metrics( + structured_output.get(SettingsKeys.METRICS, {}), index_metrics + ), + extraction_metrics, ): structured_output[SettingsKeys.METRICS] = merged_metrics # Update GUI From ec778bc95c32e6dbff2ec9581a76600414e7ec42 Mon Sep 17 00:00:00 2001 From: Athul Date: Thu, 11 Jun 2026 16:22:22 +0530 Subject: [PATCH 2/7] UN-2771 Rework: capture extraction time in the worker pipeline instead MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Per review, the structure tool's Docker path is deprecated — the live flow is the celery-based LegacyExecutor structure pipeline. Time the extract step there and merge {'extraction': {'time_taken(s)': ...}} into the result metrics alongside the existing per-output indexing timing. Structure tool changes reverted (no tool version bump needed). Co-Authored-By: Claude Fable 5 --- tools/structure/src/config/properties.json | 2 +- tools/structure/src/constants.py | 1 - tools/structure/src/main.py | 14 +------------- workers/executor/executors/legacy_executor.py | 14 ++++++++++++-- 4 files changed, 14 insertions(+), 17 deletions(-) diff --git a/tools/structure/src/config/properties.json b/tools/structure/src/config/properties.json index a9d1a6029d..c8697e7307 100644 --- a/tools/structure/src/config/properties.json +++ b/tools/structure/src/config/properties.json @@ -2,7 +2,7 @@ "schemaVersion": "0.0.1", "displayName": "Structure Tool", "functionName": "structure_tool", - "toolVersion": "0.0.102", + "toolVersion": "0.0.101", "description": "This is a template tool which can answer set of input prompts designed in the Prompt Studio", "input": { "description": "File that needs to be indexed and parsed for answers" diff --git a/tools/structure/src/constants.py b/tools/structure/src/constants.py index 294fb3015b..8da6a5701a 100644 --- a/tools/structure/src/constants.py +++ b/tools/structure/src/constants.py @@ -77,7 +77,6 @@ class SettingsKeys: TOOL = "tool" METRICS = "metrics" INDEXING = "indexing" - EXTRACTION = "extraction" EXECUTION_ID = "execution_id" IS_DIRECTORY_MODE = "is_directory_mode" LLM_PROFILE_ID = "llm_profile_id" diff --git a/tools/structure/src/main.py b/tools/structure/src/main.py index fa8ec39530..f68143a6c8 100644 --- a/tools/structure/src/main.py +++ b/tools/structure/src/main.py @@ -318,7 +318,6 @@ def run( ) extracted_text = "" - extraction_metrics = {} usage_kwargs: dict[Any, Any] = dict() if skip_extraction_and_indexing: self.stream_log( @@ -329,7 +328,6 @@ def run( usage_kwargs[UsageKwargs.RUN_ID] = self.file_execution_id usage_kwargs[UsageKwargs.FILE_NAME] = self.source_file_name usage_kwargs[UsageKwargs.EXECUTION_ID] = self.execution_id - extraction_start_time = datetime.datetime.now() extracted_text = STHelper.dynamic_extraction( file_path=input_file, enable_highlight=is_highlight_enabled, @@ -340,13 +338,6 @@ def run( tool=self, execution_run_data_folder=str(execution_run_data_folder), ) - extraction_metrics = { - SettingsKeys.EXTRACTION: { - "time_taken(s)": STHelper.elapsed_time( - start_time=extraction_start_time - ) - } - } index_metrics = {} if is_summarization_enabled: @@ -467,10 +458,7 @@ def run( "No text is extracted from the document to add to the metadata" ) if merged_metrics := self._merge_metrics( - self._merge_metrics( - structured_output.get(SettingsKeys.METRICS, {}), index_metrics - ), - extraction_metrics, + structured_output.get(SettingsKeys.METRICS, {}), index_metrics ): structured_output[SettingsKeys.METRICS] = merged_metrics # Update GUI diff --git a/workers/executor/executors/legacy_executor.py b/workers/executor/executors/legacy_executor.py index eae4d05b2f..db89d4b98d 100644 --- a/workers/executor/executors/legacy_executor.py +++ b/workers/executor/executors/legacy_executor.py @@ -619,10 +619,12 @@ def _failure(child_result: ExecutionResult) -> ExecutionResult: ) step = 1 + extraction_metrics: dict = {} try: # ---- Step 1: Extract ---- if not skip_extraction: step += 1 + extraction_start = time.monotonic() extract_ctx = ExecutionContext( executor_name=context.executor_name, operation=Operation.EXTRACT.value, @@ -640,6 +642,9 @@ def _failure(child_result: ExecutionResult) -> ExecutionResult: return _failure(extract_result) _absorb(extract_result) extracted_text = extract_result.data.get(IKeys.EXTRACTED_TEXT, "") + extraction_metrics = { + "extraction": {"time_taken(s)": time.monotonic() - extraction_start} + } # ---- Step 2: Summarize (if enabled) ---- if is_summarization: @@ -700,6 +705,7 @@ def _failure(child_result: ExecutionResult) -> ExecutionResult: source_file_name=source_file_name, extracted_text=extracted_text, index_metrics=index_metrics, + extraction_metrics=extraction_metrics, ) output_map = structured_output.get(PSKeys.OUTPUT, {}) or {} @@ -787,6 +793,7 @@ def _finalize_pipeline_result( source_file_name: str, extracted_text: str, index_metrics: dict, + extraction_metrics: dict | None = None, ) -> None: """Populate metadata/metrics in structured_output after pipeline completion.""" if "metadata" not in structured_output: @@ -794,10 +801,13 @@ def _finalize_pipeline_result( structured_output["metadata"]["file_name"] = source_file_name if extracted_text: structured_output["metadata"]["extracted_text"] = extracted_text - if index_metrics: + new_metrics = self._merge_pipeline_metrics( + index_metrics or {}, extraction_metrics or {} + ) + if new_metrics: existing_metrics = structured_output.get("metrics", {}) structured_output["metrics"] = self._merge_pipeline_metrics( - existing_metrics, index_metrics + existing_metrics, new_metrics ) def _run_pipeline_summarize( From 81d01ed182eae88499db5517ee0ee8ab376eb4af Mon Sep 17 00:00:00 2001 From: Athul Date: Tue, 14 Jul 2026 10:19:04 +0530 Subject: [PATCH 3/7] UN-2771 Address review: namespace the metric, share the duration key, align clocks MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Nest extraction timing under a reserved "_pipeline" namespace (PSKeys.PIPELINE) so it cannot collide with a user-defined output/prompt named "extraction" at the top level of the metrics dict. - Use PSKeys.EXTRACTION instead of the hardcoded literal, and capture the duration in a named local for parity with the indexing path. - Promote the duplicated "time_taken(s)" literal to PSKeys.TIME_TAKEN. - Align the indexing path onto time.monotonic(): it measured a duration with wall-clock datetime.now(), which is wrong across NTP/system-clock adjustments and left the two producers of time_taken(s) on different clocks. Drops the now-unused local datetime import. - Document the extraction_metrics shape on _finalize_pipeline_result and type it dict[str, dict] | None. Tests: 5 new cases in test_phase5d.py — metric recorded, name-collision guard, index+extraction merge, skip-extraction, and extract-failure (no timing recorded). Worker suite: 723 passed, same 6 pre-existing failures as main. Co-Authored-By: Claude Opus 4.8 (1M context) --- workers/executor/executors/constants.py | 5 + workers/executor/executors/legacy_executor.py | 28 ++-- workers/tests/test_phase5d.py | 120 ++++++++++++++++++ 3 files changed, 145 insertions(+), 8 deletions(-) diff --git a/workers/executor/executors/constants.py b/workers/executor/executors/constants.py index 9eddab8423..868a47c773 100644 --- a/workers/executor/executors/constants.py +++ b/workers/executor/executors/constants.py @@ -51,6 +51,11 @@ class PromptServiceConstants: CHALLENGE = "challenge" ENABLE_CHALLENGE = "enable_challenge" EXTRACTION = "extraction" + # Reserved namespace for pipeline-level metrics so they cannot collide with + # user-defined output/prompt names at the top level of the metrics dict. + PIPELINE = "_pipeline" + # Wire-format key every metrics producer reports its duration under. + TIME_TAKEN = "time_taken(s)" SUMMARIZE = "summarize" SINGLE_PASS_EXTRACTION = "single-pass-extraction" SIMPLE_PROMPT_STUDIO = "simple-prompt-studio" diff --git a/workers/executor/executors/legacy_executor.py b/workers/executor/executors/legacy_executor.py index ea5404e55e..07b7d8c4f1 100644 --- a/workers/executor/executors/legacy_executor.py +++ b/workers/executor/executors/legacy_executor.py @@ -642,8 +642,14 @@ def _failure(child_result: ExecutionResult) -> ExecutionResult: return _failure(extract_result) _absorb(extract_result) extracted_text = extract_result.data.get(IKeys.EXTRACTED_TEXT, "") + extraction_time = time.monotonic() - extraction_start + # Nest under a reserved "_pipeline" namespace so the metric never + # collides with a user-defined output named "extraction" during + # the top-level metrics merge (see _merge_pipeline_metrics). extraction_metrics = { - "extraction": {"time_taken(s)": time.monotonic() - extraction_start} + PSKeys.PIPELINE: { + PSKeys.EXTRACTION: {PSKeys.TIME_TAKEN: extraction_time} + } } # ---- Step 2: Summarize (if enabled) ---- @@ -806,9 +812,17 @@ def _finalize_pipeline_result( source_file_name: str, extracted_text: str, index_metrics: dict, - extraction_metrics: dict | None = None, + extraction_metrics: dict[str, dict] | None = None, ) -> None: - """Populate metadata/metrics in structured_output after pipeline completion.""" + """Populate metadata/metrics in structured_output after pipeline completion. + + Args: + extraction_metrics: Pipeline-level extraction timing, shaped as + ``{"_pipeline": {"extraction": {"time_taken(s)": float}}}``. + Nested under the reserved ``_pipeline`` namespace to avoid + colliding with user-defined output names during the merge. + ``None``/empty when extraction is skipped. + """ if "metadata" not in structured_output: structured_output["metadata"] = {} structured_output["metadata"]["file_name"] = source_file_name @@ -965,8 +979,6 @@ def _index_pipeline_output( index_records: list[dict], ) -> None: """Index a single structure-pipeline output entry in-place.""" - import datetime - chunk_size = output.get("chunk-size", 0) if chunk_size == 0: return @@ -987,7 +999,7 @@ def _index_pipeline_output( return seen_params.add(param_key) - indexing_start = datetime.datetime.now() + indexing_start = time.monotonic() logger.info( "Pipeline indexing: chunk_size=%s chunk_overlap=%s vector_db=%s", chunk_size, @@ -1039,9 +1051,9 @@ def _index_pipeline_output( if child_records: index_records.extend(child_records) - elapsed = (datetime.datetime.now() - indexing_start).total_seconds() + elapsed = time.monotonic() - indexing_start output_name = output.get("name", "") - index_metrics[output_name] = {"indexing": {"time_taken(s)": elapsed}} + index_metrics[output_name] = {"indexing": {PSKeys.TIME_TAKEN: elapsed}} @staticmethod def _merge_pipeline_metrics(metrics1: dict, metrics2: dict) -> dict: diff --git a/workers/tests/test_phase5d.py b/workers/tests/test_phase5d.py index 8b3252c9e6..72f1715242 100644 --- a/workers/tests/test_phase5d.py +++ b/workers/tests/test_phase5d.py @@ -304,6 +304,126 @@ def test_index_metrics_merged(self, executor): assert "indexing" in metrics["field_a"] +# --------------------------------------------------------------------------- +# Tests — Extraction timing metric (UN-2771) +# --------------------------------------------------------------------------- + + +class TestExtractionMetrics: + """Extraction duration is reported under the reserved _pipeline namespace. + + The metric is nested as ``metrics["_pipeline"]["extraction"]`` rather than a + bare top-level ``"extraction"`` key, because sibling keys in this dict are + user-defined output/prompt names — a prompt literally named "extraction" + would otherwise collide with it. + """ + + def _executor_with_extraction(self, executor, answer_metrics=None): + executor._handle_extract = MagicMock( + return_value=ExecutionResult( + success=True, data={"extracted_text": "text"} + ) + ) + executor._handle_index = MagicMock( + return_value=ExecutionResult(success=True, data={"doc_id": "d1"}) + ) + executor._handle_answer_prompt = MagicMock( + return_value=ExecutionResult( + success=True, + data={"output": {}, "metrics": answer_metrics or {}}, + ) + ) + return executor + + def _run(self, executor): + ctx = _make_pipeline_context({ + "extract_params": _base_extract_params(), + "index_template": _base_index_template(), + "answer_params": _base_answer_params(), + "pipeline_options": _base_pipeline_options(), + }) + return executor._handle_structure_pipeline(ctx) + + def test_extraction_metrics_recorded(self, executor): + """A normal run reports a non-negative float extraction duration.""" + self._executor_with_extraction(executor) + result = self._run(executor) + + assert result.success + time_taken = result.data["metrics"]["_pipeline"]["extraction"][ + "time_taken(s)" + ] + assert isinstance(time_taken, float) + assert time_taken >= 0 + + def test_extraction_metric_does_not_collide_with_output_name(self, executor): + """A prompt named "extraction" coexists with the pipeline metric.""" + self._executor_with_extraction( + executor, + answer_metrics={"extraction": {"llm": {"time_taken(s)": 2.0}}}, + ) + result = self._run(executor) + + metrics = result.data["metrics"] + # The user's prompt metrics are untouched... + assert metrics["extraction"] == {"llm": {"time_taken(s)": 2.0}} + # ...and the pipeline timing lives in its own namespace. + assert "time_taken(s)" in metrics["_pipeline"]["extraction"] + + def test_index_and_extraction_metrics_merged(self, executor): + """Per-output indexing and pipeline extraction metrics coexist.""" + self._executor_with_extraction( + executor, + answer_metrics={"field_a": {"llm": {"time_taken(s)": 2.0}}}, + ) + executor._run_pipeline_index = MagicMock( + return_value=({"field_a": {"indexing": {"time_taken(s)": 0.5}}}, []) + ) + result = self._run(executor) + + metrics = result.data["metrics"] + assert "llm" in metrics["field_a"] + assert "indexing" in metrics["field_a"] + assert "time_taken(s)" in metrics["_pipeline"]["extraction"] + + def test_skip_extraction_records_no_extraction_metric(self, executor): + """Smart-table path skips extract, so no extraction metric is added.""" + executor._handle_extract = MagicMock() + executor._handle_index = MagicMock() + executor._handle_answer_prompt = MagicMock( + return_value=ExecutionResult(success=True, data={"output": {}}) + ) + opts = _base_pipeline_options() + opts["skip_extraction_and_indexing"] = True + + ctx = _make_pipeline_context({ + "extract_params": _base_extract_params(), + "index_template": _base_index_template(), + "answer_params": _base_answer_params(), + "pipeline_options": opts, + }) + result = executor._handle_structure_pipeline(ctx) + + assert result.success + executor._handle_extract.assert_not_called() + assert "_pipeline" not in result.data.get("metrics", {}) + + def test_extract_failure_records_no_extraction_metric(self, executor): + """Timing is taken after the failure early-return, so a failed extract + must not report a duration.""" + executor._handle_extract = MagicMock( + return_value=ExecutionResult.failure(error="x2text error") + ) + executor._handle_index = MagicMock() + executor._handle_answer_prompt = MagicMock() + executor._finalize_pipeline_result = MagicMock() + + result = self._run(executor) + + assert not result.success + executor._finalize_pipeline_result.assert_not_called() + + # --------------------------------------------------------------------------- # Tests — Embedding usage record propagation # --------------------------------------------------------------------------- From 214d1a829c976e78564a3e070dbefc97b5abc76b Mon Sep 17 00:00:00 2001 From: Athul Date: Wed, 22 Jul 2026 10:13:15 +0530 Subject: [PATCH 4/7] UN-2771 Rename metric namespace to _file.text_extraction Review follow-up: the reserved bucket holds file-level (whole-document) metrics, so _file matches the existing file key of the API response better than _pipeline. The metric itself becomes text_extraction to stay distinct from extraction_llm, the extraction-purpose LLM call that sits beside it in the same metrics dict. Adds a single-pass test pinning cross-mode consistency: single pass skips indexing but shares _finalize_pipeline_result, so its flat metrics simply gain the same _file sibling the multi-prompt path produces. Co-Authored-By: Claude Opus 4.8 (1M context) --- workers/executor/executors/constants.py | 10 +- workers/executor/executors/legacy_executor.py | 16 +-- workers/tests/test_structure_pipeline.py | 107 ++++++++++++------ 3 files changed, 90 insertions(+), 43 deletions(-) diff --git a/workers/executor/executors/constants.py b/workers/executor/executors/constants.py index 868a47c773..cd5bd2ec07 100644 --- a/workers/executor/executors/constants.py +++ b/workers/executor/executors/constants.py @@ -51,9 +51,13 @@ class PromptServiceConstants: CHALLENGE = "challenge" ENABLE_CHALLENGE = "enable_challenge" EXTRACTION = "extraction" - # Reserved namespace for pipeline-level metrics so they cannot collide with - # user-defined output/prompt names at the top level of the metrics dict. - PIPELINE = "_pipeline" + # Reserved namespace for file-level (whole-document) metrics so they cannot + # collide with user-defined output/prompt names at the top level of the + # metrics dict. Mirrors the "file" key of the API response. + FILE = "_file" + # Text extraction (X2Text) of the source document. Named to stay distinct + # from "extraction_llm", the extraction-purpose LLM call beside it. + TEXT_EXTRACTION = "text_extraction" # Wire-format key every metrics producer reports its duration under. TIME_TAKEN = "time_taken(s)" SUMMARIZE = "summarize" diff --git a/workers/executor/executors/legacy_executor.py b/workers/executor/executors/legacy_executor.py index 07b7d8c4f1..4cbe93bb23 100644 --- a/workers/executor/executors/legacy_executor.py +++ b/workers/executor/executors/legacy_executor.py @@ -643,12 +643,12 @@ def _failure(child_result: ExecutionResult) -> ExecutionResult: _absorb(extract_result) extracted_text = extract_result.data.get(IKeys.EXTRACTED_TEXT, "") extraction_time = time.monotonic() - extraction_start - # Nest under a reserved "_pipeline" namespace so the metric never - # collides with a user-defined output named "extraction" during - # the top-level metrics merge (see _merge_pipeline_metrics). + # Nest under a reserved "_file" namespace so the metric never + # collides with a user-defined output named "text_extraction" + # during the top-level metrics merge (see _merge_pipeline_metrics). extraction_metrics = { - PSKeys.PIPELINE: { - PSKeys.EXTRACTION: {PSKeys.TIME_TAKEN: extraction_time} + PSKeys.FILE: { + PSKeys.TEXT_EXTRACTION: {PSKeys.TIME_TAKEN: extraction_time} } } @@ -817,9 +817,9 @@ def _finalize_pipeline_result( """Populate metadata/metrics in structured_output after pipeline completion. Args: - extraction_metrics: Pipeline-level extraction timing, shaped as - ``{"_pipeline": {"extraction": {"time_taken(s)": float}}}``. - Nested under the reserved ``_pipeline`` namespace to avoid + extraction_metrics: File-level text extraction timing, shaped as + ``{"_file": {"text_extraction": {"time_taken(s)": float}}}``. + Nested under the reserved ``_file`` namespace to avoid colliding with user-defined output names during the merge. ``None``/empty when extraction is skipped. """ diff --git a/workers/tests/test_structure_pipeline.py b/workers/tests/test_structure_pipeline.py index 1482dc26b5..80119faacf 100644 --- a/workers/tests/test_structure_pipeline.py +++ b/workers/tests/test_structure_pipeline.py @@ -8,6 +8,7 @@ from unittest.mock import MagicMock, patch import pytest + from unstract.sdk1.execution.context import ExecutionContext, Operation from unstract.sdk1.execution.result import ExecutionResult @@ -294,24 +295,22 @@ def test_index_metrics_merged(self, executor): # --------------------------------------------------------------------------- -# Tests — Extraction timing metric (UN-2771) +# Tests — Text extraction timing metric # --------------------------------------------------------------------------- class TestExtractionMetrics: - """Extraction duration is reported under the reserved _pipeline namespace. + """Extraction duration is reported under the reserved _file namespace. - The metric is nested as ``metrics["_pipeline"]["extraction"]`` rather than a - bare top-level ``"extraction"`` key, because sibling keys in this dict are - user-defined output/prompt names — a prompt literally named "extraction" - would otherwise collide with it. + The metric is nested as ``metrics["_file"]["text_extraction"]`` rather than a + bare top-level key, because sibling keys in this dict are user-defined + output/prompt names — a prompt named "text_extraction" would otherwise + collide with it. """ def _executor_with_extraction(self, executor, answer_metrics=None): executor._handle_extract = MagicMock( - return_value=ExecutionResult( - success=True, data={"extracted_text": "text"} - ) + return_value=ExecutionResult(success=True, data={"extracted_text": "text"}) ) executor._handle_index = MagicMock( return_value=ExecutionResult(success=True, data={"doc_id": "d1"}) @@ -325,12 +324,14 @@ def _executor_with_extraction(self, executor, answer_metrics=None): return executor def _run(self, executor): - ctx = _make_pipeline_context({ - "extract_params": _base_extract_params(), - "index_template": _base_index_template(), - "answer_params": _base_answer_params(), - "pipeline_options": _base_pipeline_options(), - }) + ctx = _make_pipeline_context( + { + "extract_params": _base_extract_params(), + "index_template": _base_index_template(), + "answer_params": _base_answer_params(), + "pipeline_options": _base_pipeline_options(), + } + ) return executor._handle_structure_pipeline(ctx) def test_extraction_metrics_recorded(self, executor): @@ -339,25 +340,23 @@ def test_extraction_metrics_recorded(self, executor): result = self._run(executor) assert result.success - time_taken = result.data["metrics"]["_pipeline"]["extraction"][ - "time_taken(s)" - ] + time_taken = result.data["metrics"]["_file"]["text_extraction"]["time_taken(s)"] assert isinstance(time_taken, float) assert time_taken >= 0 def test_extraction_metric_does_not_collide_with_output_name(self, executor): - """A prompt named "extraction" coexists with the pipeline metric.""" + """A prompt named "text_extraction" coexists with the file metric.""" self._executor_with_extraction( executor, - answer_metrics={"extraction": {"llm": {"time_taken(s)": 2.0}}}, + answer_metrics={"text_extraction": {"llm": {"time_taken(s)": 2.0}}}, ) result = self._run(executor) metrics = result.data["metrics"] # The user's prompt metrics are untouched... - assert metrics["extraction"] == {"llm": {"time_taken(s)": 2.0}} - # ...and the pipeline timing lives in its own namespace. - assert "time_taken(s)" in metrics["_pipeline"]["extraction"] + assert metrics["text_extraction"] == {"llm": {"time_taken(s)": 2.0}} + # ...and the file-level timing lives in its own namespace. + assert "time_taken(s)" in metrics["_file"]["text_extraction"] def test_index_and_extraction_metrics_merged(self, executor): """Per-output indexing and pipeline extraction metrics coexist.""" @@ -373,7 +372,7 @@ def test_index_and_extraction_metrics_merged(self, executor): metrics = result.data["metrics"] assert "llm" in metrics["field_a"] assert "indexing" in metrics["field_a"] - assert "time_taken(s)" in metrics["_pipeline"]["extraction"] + assert "time_taken(s)" in metrics["_file"]["text_extraction"] def test_skip_extraction_records_no_extraction_metric(self, executor): """Smart-table path skips extract, so no extraction metric is added.""" @@ -385,21 +384,65 @@ def test_skip_extraction_records_no_extraction_metric(self, executor): opts = _base_pipeline_options() opts["skip_extraction_and_indexing"] = True - ctx = _make_pipeline_context({ - "extract_params": _base_extract_params(), - "index_template": _base_index_template(), - "answer_params": _base_answer_params(), - "pipeline_options": opts, - }) + ctx = _make_pipeline_context( + { + "extract_params": _base_extract_params(), + "index_template": _base_index_template(), + "answer_params": _base_answer_params(), + "pipeline_options": opts, + } + ) result = executor._handle_structure_pipeline(ctx) assert result.success executor._handle_extract.assert_not_called() - assert "_pipeline" not in result.data.get("metrics", {}) + assert "_file" not in result.data.get("metrics", {}) + + def test_single_pass_reports_same_file_metric(self, executor): + """Single pass gains the _file bucket beside its flat metrics. + + Single pass skips indexing but shares ``_finalize_pipeline_result``, so + it needs no special-casing: its metrics stay flat and simply gain the + same ``_file`` sibling the multi-prompt path gets. + """ + self._executor_with_extraction(executor) + executor._handle_single_pass_extraction = MagicMock( + return_value=ExecutionResult( + success=True, + data={ + "output": {}, + "metrics": { + "context_retrieval": {"time_taken(s)": 0.4}, + "extraction_llm": {"time_taken(s)": 2.0}, + }, + }, + ) + ) + opts = _base_pipeline_options() + opts["is_single_pass_enabled"] = True + + ctx = _make_pipeline_context( + { + "extract_params": _base_extract_params(), + "index_template": _base_index_template(), + "answer_params": _base_answer_params(), + "pipeline_options": opts, + } + ) + result = executor._handle_structure_pipeline(ctx) + + assert result.success + metrics = result.data["metrics"] + # Flat single-pass metrics survive untouched... + assert metrics["context_retrieval"] == {"time_taken(s)": 0.4} + assert metrics["extraction_llm"] == {"time_taken(s)": 2.0} + # ...beside the same _file bucket the multi-prompt path produces. + assert "time_taken(s)" in metrics["_file"]["text_extraction"] def test_extract_failure_records_no_extraction_metric(self, executor): """Timing is taken after the failure early-return, so a failed extract - must not report a duration.""" + must not report a duration. + """ executor._handle_extract = MagicMock( return_value=ExecutionResult.failure(error="x2text error") ) From a34a106eba72975a1afcb9875016a89f0f3dbd1c Mon Sep 17 00:00:00 2001 From: Athul Date: Fri, 4 Sep 2026 00:24:34 +0530 Subject: [PATCH 5/7] UN-2771 Report extraction time for agentic_table prompts too MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The extraction timer added by this PR runs only in LegacyExecutor._handle_structure_pipeline. Agentic-table prompts are dispatched to their own executor, which performs its own X2Text, so agentic-only deployments reported no extraction time at all and mixed deployments reported only the legacy pipeline's share. The agentic dispatch was already discarding the executor's result beyond `output.tables`. Read its metrics instead — same `data["metadata"]["metrics"]` shape LegacyExecutor._run_table_extraction uses for the non-agentic table plugin — and fold them in: - per-prompt metrics land beside the legacy pipeline's, under `table_extraction`, so both table routes read alike downstream - any reported X2Text duration is added to `_file.text_extraction`, so the file-level figure covers every extractor that ran The all-agentic branch previously built its result with no `metrics` key at all; it now populates one on the same path. Durations sum rather than replace: each agentic prompt extracts the document itself, so the metric is time spent extracting, not the wall-clock span. An executor reporting no timing leaves the bucket absent rather than writing a misleading zero. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01VdHng2GqQpAxBbQs8Ks5Ej --- .../file_processing/structure_tool_task.py | 72 +++++++++++- workers/tests/test_agentic_operations.py | 106 ++++++++++++++++++ 2 files changed, 177 insertions(+), 1 deletion(-) diff --git a/workers/file_processing/structure_tool_task.py b/workers/file_processing/structure_tool_task.py index 61334e63fc..1577cdad5a 100644 --- a/workers/file_processing/structure_tool_task.py +++ b/workers/file_processing/structure_tool_task.py @@ -196,6 +196,65 @@ def _should_skip_extraction_for_smart_table( return False +def _agentic_extraction_seconds(agentic_metrics: dict[str, Any]) -> float: + """Total X2Text seconds reported by the agentic_table executor. + + Each agentic prompt extracts the document itself, so the durations sum: + the figure is time *spent* extracting, not the wall-clock span of the + agentic step. Returns 0.0 when the executor reports no extraction timing, + which leaves the file-level bucket untouched rather than writing a zero. + """ + from executor.executors.constants import PromptServiceConstants as PSKeys + + total = 0.0 + for prompt_metrics in agentic_metrics.values(): + if not isinstance(prompt_metrics, dict): + continue + entry = prompt_metrics.get(PSKeys.TEXT_EXTRACTION) + if not isinstance(entry, dict): + continue + value = entry.get(PSKeys.TIME_TAKEN) + if isinstance(value, (int, float)) and not isinstance(value, bool): + total += float(value) + return total + + +def _merge_agentic_metrics( + structured_output: dict[str, Any], agentic_metrics: dict[str, Any] +) -> None: + """Fold the agentic executor's metrics into the tool's metrics dict. + + Two things happen here, and they are separate: + + * Per-prompt metrics land beside the legacy pipeline's, keyed by prompt + name under ``table_extraction`` — the same shape + ``LegacyExecutor._run_table_extraction`` uses for the non-agentic table + plugin, so both table routes read alike downstream. + * Any X2Text duration the executor reported is *added* to the file-level + ``_file.text_extraction`` bucket. The legacy pipeline times only its own + extraction, so without this an agentic-only run reports no extraction + time at all and a mixed run under-reports it (UN-2771 / UN-4084). + """ + from executor.executors.constants import PromptServiceConstants as PSKeys + + if not agentic_metrics: + return + + metrics = structured_output.setdefault("metrics", {}) + for prompt_name, prompt_metrics in agentic_metrics.items(): + metrics.setdefault(prompt_name, {}).update({"table_extraction": prompt_metrics}) + + extraction_seconds = _agentic_extraction_seconds(agentic_metrics) + if not extraction_seconds: + return + file_bucket = metrics.setdefault(PSKeys.FILE, {}).setdefault( + PSKeys.TEXT_EXTRACTION, {} + ) + file_bucket[PSKeys.TIME_TAKEN] = ( + file_bucket.get(PSKeys.TIME_TAKEN, 0.0) + extraction_seconds + ) + + # ----------------------------------------------------------------------- # Main Celery task # ----------------------------------------------------------------------- @@ -443,6 +502,7 @@ def _execute_structure_tool_impl(params: dict) -> dict: # PDF written alongside INFILE by the source connector. agentic_source_path = str(execution_run_data_folder / "SOURCE") agentic_results: dict[str, Any] = {} + agentic_metrics: dict[str, Any] = {} for at_output in agentic_table_outputs: at_settings = at_output.get("agentic_table_settings") or {} json_structure = at_settings.get("json_structure") @@ -490,6 +550,12 @@ def _execute_structure_tool_impl(params: dict) -> dict: return at_result.to_dict() at_output_data = at_result.data.get("output", {}) or {} agentic_results[at_output[_SK.NAME]] = at_output_data.get("tables", []) + # The executor runs its own X2Text, so its metrics carry an extraction + # duration the legacy pipeline's timer never sees. Same result shape the + # non-agentic table plugin uses (LegacyExecutor._run_table_extraction). + at_metrics = (at_result.data.get("metadata") or {}).get("metrics") or {} + if at_metrics: + agentic_metrics[at_output[_SK.NAME]] = at_metrics # ---- Step 6b: Dispatch legacy structure_pipeline ---- # Skipped entirely when every prompt is agentic_table — the legacy @@ -536,12 +602,16 @@ def _execute_structure_tool_impl(params: dict) -> dict: structured_output = pipeline_result.data if agentic_results: structured_output.setdefault("output", {}).update(agentic_results) + _merge_agentic_metrics(structured_output, agentic_metrics) else: - # All-agentic case: skip the legacy pipeline entirely. + # All-agentic case: skip the legacy pipeline entirely. The metrics + # dict is still populated, so an agentic-only deployment reports the + # same _file.text_extraction the legacy path does. structured_output = { "output": agentic_results, "metadata": {"agentic_only": True}, } + _merge_agentic_metrics(structured_output, agentic_metrics) pipeline_elapsed = 0.0 # ---- Step 7: Write output files ---- diff --git a/workers/tests/test_agentic_operations.py b/workers/tests/test_agentic_operations.py index f16e566d01..3167866e25 100644 --- a/workers/tests/test_agentic_operations.py +++ b/workers/tests/test_agentic_operations.py @@ -262,3 +262,109 @@ def test_agentic_ops_use_default_log_component(self, op): "table_extract", "smart_table_extract", ) + + +# --------------------------------------------------------------------------- +# Agentic extraction timing (UN-2771) +# --------------------------------------------------------------------------- + + +class TestAgenticExtractionMetrics: + """The agentic executor's X2Text duration reaches the file-level metric. + + The legacy pipeline times only its own extraction. Agentic prompts extract + the document inside their own executor, so without folding their reported + duration in, an agentic-only deployment reports no extraction time at all + and a mixed deployment under-reports it. + """ + + @staticmethod + def _agentic(seconds): + """One prompt's metrics as the executor reports them.""" + return {"invoices": {"text_extraction": {"time_taken(s)": seconds}}} + + def test_sums_across_prompts(self): + """Each agentic prompt extracts the file itself, so durations add.""" + from file_processing.structure_tool_task import _agentic_extraction_seconds + + metrics = { + "invoices": {"text_extraction": {"time_taken(s)": 1.5}}, + "receipts": {"text_extraction": {"time_taken(s)": 2.25}}, + } + assert _agentic_extraction_seconds(metrics) == pytest.approx(3.75) + + @pytest.mark.parametrize( + "metrics", + [ + {}, + {"invoices": {}}, + {"invoices": {"text_extraction": {}}}, + {"invoices": {"text_extraction": {"time_taken(s)": None}}}, + {"invoices": {"text_extraction": "not-a-dict"}}, + {"invoices": "not-a-dict"}, + # bool is an int subclass; True must not count as 1 second + {"invoices": {"text_extraction": {"time_taken(s)": True}}}, + ], + ) + def test_missing_or_malformed_timing_yields_zero(self, metrics): + """A plugin that reports no duration must not fabricate one.""" + from file_processing.structure_tool_task import _agentic_extraction_seconds + + assert _agentic_extraction_seconds(metrics) == 0.0 + + def test_all_agentic_run_reports_the_file_metric(self): + """The agentic-only path has no legacy pipeline, so it starts empty.""" + from file_processing.structure_tool_task import _merge_agentic_metrics + + structured_output = {"output": {}, "metadata": {"agentic_only": True}} + _merge_agentic_metrics(structured_output, self._agentic(4.0)) + + metrics = structured_output["metrics"] + assert metrics["_file"]["text_extraction"]["time_taken(s)"] == pytest.approx(4.0) + + def test_mixed_run_adds_to_the_legacy_duration(self): + """The under-reporting case: both extractors ran, both must count.""" + from file_processing.structure_tool_task import _merge_agentic_metrics + + structured_output = { + "metrics": {"_file": {"text_extraction": {"time_taken(s)": 1.0}}} + } + _merge_agentic_metrics(structured_output, self._agentic(2.5)) + + total = structured_output["metrics"]["_file"]["text_extraction"]["time_taken(s)"] + assert total == pytest.approx(3.5) + + def test_per_prompt_metrics_land_beside_the_legacy_ones(self): + """Same shape LegacyExecutor._run_table_extraction uses.""" + from file_processing.structure_tool_task import _merge_agentic_metrics + + structured_output = {"metrics": {"field_a": {"llm": {"time_taken(s)": 2.0}}}} + _merge_agentic_metrics(structured_output, self._agentic(1.0)) + + metrics = structured_output["metrics"] + assert metrics["field_a"] == {"llm": {"time_taken(s)": 2.0}} + assert metrics["invoices"]["table_extraction"] == { + "text_extraction": {"time_taken(s)": 1.0} + } + + def test_no_agentic_prompts_changes_nothing(self): + """A purely legacy run must not grow an empty metrics dict.""" + from file_processing.structure_tool_task import _merge_agentic_metrics + + structured_output = {"output": {}} + _merge_agentic_metrics(structured_output, {}) + + assert "metrics" not in structured_output + + def test_executor_reporting_no_timing_leaves_the_bucket_absent(self): + """Per-prompt metrics still surface; a zero duration is not written.""" + from file_processing.structure_tool_task import _merge_agentic_metrics + + structured_output = {} + _merge_agentic_metrics( + structured_output, {"invoices": {"table_rows": {"count": 12}}} + ) + + metrics = structured_output["metrics"] + assert metrics["invoices"]["table_extraction"] == {"table_rows": {"count": 12}} + assert "_file" not in metrics From 23ed7b1ef0056061d975b4fe110e67463bda8d7a Mon Sep 17 00:00:00 2001 From: Athul Date: Fri, 4 Sep 2026 00:30:24 +0530 Subject: [PATCH 6/7] UN-2771 Keep a prompt named _file out of the reserved metrics bucket MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit _file is reserved for whole-document metrics, which is why the extraction timing is nested there rather than sitting at the top level beside user-defined output names. The agentic merge wrote per-prompt metrics keyed by prompt name into that same dict, so an agentic_table prompt actually named _file would land its table_extraction metrics inside the reserved bucket, and the extraction total would then write into the same object — conflating the two namespaces the bucket exists to keep apart. Skip the per-prompt entry for that name and log it. The prompt's extraction duration still counts toward the file-level total; only the per-prompt entry is dropped, since there is nowhere collision-free to put it. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01VdHng2GqQpAxBbQs8Ks5Ej --- .../file_processing/structure_tool_task.py | 11 +++++++++++ workers/tests/test_agentic_operations.py | 19 +++++++++++++++++++ 2 files changed, 30 insertions(+) diff --git a/workers/file_processing/structure_tool_task.py b/workers/file_processing/structure_tool_task.py index 1577cdad5a..d13c90b9ee 100644 --- a/workers/file_processing/structure_tool_task.py +++ b/workers/file_processing/structure_tool_task.py @@ -242,6 +242,17 @@ def _merge_agentic_metrics( metrics = structured_output.setdefault("metrics", {}) for prompt_name, prompt_metrics in agentic_metrics.items(): + if prompt_name == PSKeys.FILE: + # `_file` is reserved for whole-document metrics. A prompt with that + # name would land its per-prompt metrics in the same object the + # extraction total below writes into, conflating the two namespaces + # the reserved bucket exists to keep apart. Its duration still counts. + logger.warning( + "Agentic prompt named %r collides with the reserved file-level " + "metrics namespace; its per-prompt metrics are omitted", + prompt_name, + ) + continue metrics.setdefault(prompt_name, {}).update({"table_extraction": prompt_metrics}) extraction_seconds = _agentic_extraction_seconds(agentic_metrics) diff --git a/workers/tests/test_agentic_operations.py b/workers/tests/test_agentic_operations.py index 3167866e25..3e2c810482 100644 --- a/workers/tests/test_agentic_operations.py +++ b/workers/tests/test_agentic_operations.py @@ -368,3 +368,22 @@ def test_executor_reporting_no_timing_leaves_the_bucket_absent(self): metrics = structured_output["metrics"] assert metrics["invoices"]["table_extraction"] == {"table_rows": {"count": 12}} assert "_file" not in metrics + + def test_prompt_named_file_does_not_invade_the_reserved_bucket(self): + """`_file` is reserved; a prompt with that name must not land in it. + + Its extraction duration still counts toward the file-level total — + only the per-prompt entry is dropped, because there is nowhere + collision-free to put it. + """ + from file_processing.structure_tool_task import _merge_agentic_metrics + + structured_output = {} + _merge_agentic_metrics( + structured_output, + {"_file": {"text_extraction": {"time_taken(s)": 3.0}}}, + ) + + file_bucket = structured_output["metrics"]["_file"] + assert file_bucket["text_extraction"] == {"time_taken(s)": 3.0} + assert "table_extraction" not in file_bucket From 3e2d7ea6b1f20af7030f08ee267d6fa56d310086 Mon Sep 17 00:00:00 2001 From: Athul Date: Fri, 4 Sep 2026 14:17:25 +0530 Subject: [PATCH 7/7] UN-2771 Rehome a prompt named _file instead of dropping its metrics Skipping the entry kept the reserved bucket clean but silently lost every other count the prompt reported. Merging it in loses the separation the reserved bucket exists for. Neither is acceptable, so the prompt moves aside instead: its metrics are reported under "_file (prompt)". The reserved key contains no space, so the two can never alias, and nothing the executor reported is discarded. The prompt's extraction duration still counts toward the file-level total as before. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01VdHng2GqQpAxBbQs8Ks5Ej --- .../file_processing/structure_tool_task.py | 19 +++++++----- workers/tests/test_agentic_operations.py | 30 +++++++++++++------ 2 files changed, 32 insertions(+), 17 deletions(-) diff --git a/workers/file_processing/structure_tool_task.py b/workers/file_processing/structure_tool_task.py index d13c90b9ee..4b2f02bfcc 100644 --- a/workers/file_processing/structure_tool_task.py +++ b/workers/file_processing/structure_tool_task.py @@ -242,18 +242,21 @@ def _merge_agentic_metrics( metrics = structured_output.setdefault("metrics", {}) for prompt_name, prompt_metrics in agentic_metrics.items(): - if prompt_name == PSKeys.FILE: - # `_file` is reserved for whole-document metrics. A prompt with that - # name would land its per-prompt metrics in the same object the - # extraction total below writes into, conflating the two namespaces - # the reserved bucket exists to keep apart. Its duration still counts. + key = prompt_name + if key == PSKeys.FILE: + # `_file` is reserved for whole-document metrics, so a prompt with + # that name gets a de-collided key. Merging it into the reserved + # bucket would conflate per-prompt and file-level figures; dropping + # it would lose every other count the prompt reported. The space + # cannot appear in the reserved key, so the two can never alias. + key = f"{prompt_name} (prompt)" logger.warning( "Agentic prompt named %r collides with the reserved file-level " - "metrics namespace; its per-prompt metrics are omitted", + "metrics namespace; reporting its metrics under %r instead", prompt_name, + key, ) - continue - metrics.setdefault(prompt_name, {}).update({"table_extraction": prompt_metrics}) + metrics.setdefault(key, {}).update({"table_extraction": prompt_metrics}) extraction_seconds = _agentic_extraction_seconds(agentic_metrics) if not extraction_seconds: diff --git a/workers/tests/test_agentic_operations.py b/workers/tests/test_agentic_operations.py index 3e2c810482..4d0fec4977 100644 --- a/workers/tests/test_agentic_operations.py +++ b/workers/tests/test_agentic_operations.py @@ -369,21 +369,33 @@ def test_executor_reporting_no_timing_leaves_the_bucket_absent(self): assert metrics["invoices"]["table_extraction"] == {"table_rows": {"count": 12}} assert "_file" not in metrics - def test_prompt_named_file_does_not_invade_the_reserved_bucket(self): - """`_file` is reserved; a prompt with that name must not land in it. + def test_prompt_named_file_gets_a_de_collided_key(self): + """`_file` is reserved, so a prompt with that name moves aside. - Its extraction duration still counts toward the file-level total — - only the per-prompt entry is dropped, because there is nowhere - collision-free to put it. + Neither outcome the two namespaces could otherwise produce is + acceptable: merging into the reserved bucket conflates per-prompt with + file-level figures, and dropping the entry loses every other metric the + prompt reported. It is rehomed instead, and nothing is lost. """ from file_processing.structure_tool_task import _merge_agentic_metrics structured_output = {} _merge_agentic_metrics( structured_output, - {"_file": {"text_extraction": {"time_taken(s)": 3.0}}}, + {"_file": {"text_extraction": {"time_taken(s)": 3.0}, "rows": 12}}, ) - file_bucket = structured_output["metrics"]["_file"] - assert file_bucket["text_extraction"] == {"time_taken(s)": 3.0} - assert "table_extraction" not in file_bucket + metrics = structured_output["metrics"] + # The reserved bucket carries file-level metrics only... + assert metrics["_file"] == {"text_extraction": {"time_taken(s)": 3.0}} + # ...and the prompt keeps everything it reported, under its own key. + assert metrics["_file (prompt)"]["table_extraction"] == { + "text_extraction": {"time_taken(s)": 3.0}, + "rows": 12, + } + + def test_de_collided_key_cannot_alias_the_reserved_one(self): + """The rehomed key is distinct from the reserved key by construction.""" + from executor.executors.constants import PromptServiceConstants as PSKeys + + assert " " not in PSKeys.FILE