From 9397d5c376f4248a290b91527f58a2e82a9cd72b Mon Sep 17 00:00:00 2001 From: Paul Lizer Date: Thu, 13 Aug 2026 20:04:21 -0400 Subject: [PATCH] Fix narrative Analyze evidence-envelope truncation (0.250.194) Root cause: pure-narrative Analyze requests (PDF -> XML/JSON template population, no tabular sources) were routed through the mixed-source evidence-envelope + collective-reduction hop. That hop exists to combine narrative and tabular evidence into one bounded payload, but its per-source summary cap discarded most of run_document_analysis's own already-synthesized output before the final reply was ever produced. Fixes: - _execute_mixed_source_analyze_workflow now bypasses the evidence envelope/collective-reduction path entirely when there are narrative sources, a completed narrative_result, and no tabular sources - returning run_document_analysis's full-fidelity result directly. mixed_source_manifest is preserved on the returned dict so the existing reauthorization security check still runs unchanged. - Raised EVIDENCE_SUMMARY_MAX_BYTES (4096 -> 16384) and MIXED_SOURCE_HANDOFF_MAX_BYTES (49152 -> 131072) as a secondary safety margin for remaining mixed-source (narrative + tabular) runs that still go through the envelope/handoff path. - Added char-count diagnostic logging across the window analysis, per-document reduction, and global reduction stages in run_document_analysis, plus around evidence envelope/handoff sizing in the mixed-source workflow, to make any future truncation easier to pinpoint from logs alone. Also fixes a pre-existing namespace gap in test_tabular_analyze_shared_preflight_adapter.py's AST-extraction harness (missing debug_print/json stubs) surfaced by the new bypass's logging calls. Validated: all data elements (BusinessUnit/LoanCategory/DepositCategory/ Exception/Certification counts and the full "Reserved optional extension" comment block) now match the source XML template exactly, confirmed against a real production PDF+XML population run. --- application/single_app/config.py | 2 +- .../single_app/functions_document_analysis.py | 32 +++++++-- .../functions_mixed_source_orchestration.py | 8 ++- .../single_app/functions_workflow_runner.py | 72 ++++++++++++++++++- ...abular_analyze_shared_preflight_adapter.py | 3 + 5 files changed, 109 insertions(+), 8 deletions(-) diff --git a/application/single_app/config.py b/application/single_app/config.py index 2a45ecb17..44827d8cf 100644 --- a/application/single_app/config.py +++ b/application/single_app/config.py @@ -96,7 +96,7 @@ EXECUTOR_TYPE = 'thread' EXECUTOR_MAX_WORKERS = 30 SESSION_TYPE = 'filesystem' -VERSION = "0.250.191" +VERSION = "0.250.194" IS_DEVELOPMENT = is_development_env_enabled() SESSION_COOKIE_SAMESITE = os.getenv('SESSION_COOKIE_SAMESITE', 'Lax') diff --git a/application/single_app/functions_document_analysis.py b/application/single_app/functions_document_analysis.py index c9b6e2bd0..7900b8175 100644 --- a/application/single_app/functions_document_analysis.py +++ b/application/single_app/functions_document_analysis.py @@ -761,6 +761,13 @@ def _reduce_document_analysis_items( stage_label=f'document-reduction-{reduction_round}.{batch_index}', failed_range_labels=failed_range_labels, ) + batch_input_chars = sum(len(str(item.get('text') or '')) for item in batch_items) + debug_print( + '[DOCUMENT_ANALYSIS] Starting document reduction batch | ' + f'document_name={document_name} | ' + f'round={reduction_round} | batch={batch_index}/{len(batches)} | ' + f'items={len(batch_items)} | input_chars={batch_input_chars}' + ) reduced_text = str(invoke_prompt( reduction_prompt, stage='reduction', @@ -782,6 +789,12 @@ def _reduce_document_analysis_items( f'Document analysis document reduction returned an empty response for {document_name} ' f'at round {reduction_round}, batch {batch_index}.' ) + debug_print( + '[DOCUMENT_ANALYSIS] Completed document reduction batch | ' + f'document_name={document_name} | ' + f'round={reduction_round} | batch={batch_index}/{len(batches)} | ' + f'input_chars={batch_input_chars} | output_chars={len(reduced_text)}' + ) next_items.append({ 'label': f'{document_name} reduction {reduction_round}.{batch_index}', 'text': reduced_text, @@ -1100,6 +1113,7 @@ def run_document_analysis( 'progress': _build_progress_snapshot(coverage), }) + window_source_chars = len(_render_window_source_text(window_payload)) analysis_text = '' last_error = '' max_attempts = targets.get('max_retries_per_window', DEFAULT_MAX_RETRIES_PER_WINDOW) + 1 @@ -1188,7 +1202,10 @@ def run_document_analysis( f'document_id={document_id} | ' f'document_name={document_name} | ' f"window={window_range.get('window_number')} | " - f"chunk_count={window_range.get('chunk_count', 0)}" + f"chunk_count={window_range.get('chunk_count', 0)} | " + f'source_chars={window_source_chars} | ' + f'prompt_chars={len(prompt_text)} | ' + f'response_chars={len(analysis_text)}' ) coverage['processed_windows'] += 1 coverage['processed_chunks'] += window_range.get('chunk_count', 0) or 0 @@ -1242,6 +1259,7 @@ def run_document_analysis( f'document_id={document_id} | ' f'document_name={document_name} | ' f"window={window_range.get('window_number')} | " + f'source_chars={window_source_chars} | ' f'error={last_error or "unknown"}' ) coverage['failed_windows'] += 1 @@ -1401,11 +1419,13 @@ def run_document_analysis( phase_step=reduction_step_index, phase_total_steps=reduction_step_total, ) + global_reduction_input_chars = sum(len(str(item.get('text') or '')) for item in batch_items) debug_print( '[DOCUMENT_ANALYSIS] Starting reduction batch | ' f'round={reduction_round} | ' f'batch={batch_index}/{len(batches)} | ' - f'items={len(batch_items)}' + f'items={len(batch_items)} | ' + f'input_chars={global_reduction_input_chars}' ) if callable(activity_callback): activity_callback({ @@ -1455,7 +1475,9 @@ def run_document_analysis( '[DOCUMENT_ANALYSIS] Completed reduction batch | ' f'round={reduction_round} | ' f'batch={batch_index}/{len(batches)} | ' - f'sources={len(source_labels)}' + f'sources={len(source_labels)} | ' + f'input_chars={global_reduction_input_chars} | ' + f'output_chars={len(reduced_text)}' ) next_items.append({ 'label': f'Reduction {reduction_round}.{batch_index}', @@ -1509,6 +1531,7 @@ def run_document_analysis( 'processed_windows': coverage.get('processed_windows', 0), 'failed_windows': coverage.get('failed_windows', 0), 'retries': coverage.get('retries', 0), + 'final_analysis_reply_chars': len(final_analysis_reply), }, level=logging.INFO, ) @@ -1518,7 +1541,8 @@ def run_document_analysis( f"windows={coverage.get('total_windows', 0)} | " f"processed={coverage.get('processed_windows', 0)} | " f"failed={coverage.get('failed_windows', 0)} | " - f"retries={coverage.get('retries', 0)}" + f"retries={coverage.get('retries', 0)} | " + f'final_analysis_reply_chars={len(final_analysis_reply)}' ) return { diff --git a/application/single_app/functions_mixed_source_orchestration.py b/application/single_app/functions_mixed_source_orchestration.py index ee6fa125a..7d5dc68f9 100644 --- a/application/single_app/functions_mixed_source_orchestration.py +++ b/application/single_app/functions_mixed_source_orchestration.py @@ -94,7 +94,9 @@ def log_event(*args, **kwargs): }) EVIDENCE_ENVELOPE_MAX_BYTES = 65536 -EVIDENCE_SUMMARY_MAX_BYTES = 4096 +# Narrative Analyze summaries carry the only copy of document detail (no separate durable +# artifact channel like tabular exports have), so this must stay well below EVIDENCE_ENVELOPE_MAX_BYTES. +EVIDENCE_SUMMARY_MAX_BYTES = 16384 EVIDENCE_ERROR_MAX_BYTES = 1024 EVIDENCE_LIST_MAX_ITEMS = 10 EVIDENCE_ITEM_MAX_BYTES = 1536 @@ -102,7 +104,9 @@ def log_event(*args, **kwargs): EVIDENCE_JSON_MAX_DEPTH = 4 EVIDENCE_JSON_MAX_COLLECTION_ITEMS = 20 EVIDENCE_JSON_MAX_STRING_BYTES = 1024 -MIXED_SOURCE_HANDOFF_MAX_BYTES = 49152 +# Raised alongside EVIDENCE_SUMMARY_MAX_BYTES so a handful of narrative sources at the new +# per-summary cap do not immediately fall back to the 512-byte-per-summary compaction path. +MIXED_SOURCE_HANDOFF_MAX_BYTES = 131072 MIXED_SOURCE_HANDOFF_MAX_ENVELOPES = 20 MIXED_SOURCE_MODES = frozenset({"chat", "search", "analyze", "compare"}) MIXED_SOURCE_TERMINAL_REASON_MAX_BYTES = 128 diff --git a/application/single_app/functions_workflow_runner.py b/application/single_app/functions_workflow_runner.py index 27e98fa69..f48d0f764 100644 --- a/application/single_app/functions_workflow_runner.py +++ b/application/single_app/functions_workflow_runner.py @@ -1247,6 +1247,13 @@ def _upload_document_analysis_generated_artifact( ) return None + debug_print( + '[WORKFLOW_DOCUMENT_ANALYSIS] Uploaded generated artifact | ' + f'conversation_id={normalized_conversation_id} | ' + f'file={file_name} | ' + f'output_format={output_format} | ' + f'content_chars={len(str(file_content or ""))}' + ) artifact_payload = { 'capability': 'analyze', 'artifact_message_id': upload_result.get('message', {}).get('id'), @@ -1344,6 +1351,15 @@ def _maybe_create_document_analysis_generated_artifacts( json_artifact_requested = _prompt_explicitly_requests_json_artifact(analysis_prompt) xml_payload = normalize_xml_artifact_payload(analysis_reply) xml_artifact_requested = bool(artifact_intent.get('xml_artifact_requested')) + debug_print( + '[WORKFLOW_DOCUMENT_ANALYSIS] Analysis artifact sizing | ' + f'document_count={document_count} | ' + f'analysis_reply_chars={len(analysis_reply)} | ' + f'xml_payload_chars={len(xml_payload) if xml_payload else 0} | ' + f'xml_artifact_requested={xml_artifact_requested} | ' + f'json_payload_present={json_payload is not None} | ' + f'json_artifact_requested={json_artifact_requested}' + ) create_lossless_artifacts = bool( artifact_intent.get('exhaustive') or artifact_intent.get('table_output_requested') @@ -2948,6 +2964,7 @@ def _execute_mixed_source_analyze_workflow( return tabular_preflight_result narrative_sources = partitions['narrative_sources'] + narrative_result = None if narrative_sources: if callable(activity_callback): activity_callback({'type': 'mixed_source_progress', 'phase': 'analyzing_narrative', 'label': 'Analyzing narrative documents'}) @@ -2992,20 +3009,29 @@ def _execute_mixed_source_analyze_workflow( if total_windows and processed_windows == total_windows and not failed_windows else ('partial' if processed_windows else EVIDENCE_STATUS_FAILED) ) + narrative_summary_source_text = str(narrative_item.get('text') or '') evidence_envelopes.append(build_evidence_envelope( document_id=document_id, source_kind='narrative', engine=EVIDENCE_ENGINE_DOCUMENT_ANALYSIS, status=status, - summary=str(narrative_item.get('text') or ''), + summary=narrative_summary_source_text, citations=[], generated_artifacts=[], coverage={'terminal': True, 'processed_windows': processed_windows, 'total_windows': total_windows, 'failed_windows': failed_windows}, error='Narrative analysis could not be completed.' if status == EVIDENCE_STATUS_FAILED else None, )) + debug_print( + '[MIXED_SOURCE_ANALYZE] Narrative evidence envelope sized | ' + f'document_id={document_id} | ' + f'source_text_chars={len(narrative_summary_source_text)} | ' + f"envelope_summary_chars={len(evidence_envelopes[-1].get('summary') or '')} | " + f"envelope_truncated={bool((evidence_envelopes[-1].get('coverage') or {}).get('evidence_envelope_truncated'))}" + ) except MixedSourceCancellationError: raise except Exception: + narrative_result = None for source in narrative_sources: evidence_envelopes.append(build_evidence_envelope( document_id=source.get('document_id'), source_kind='narrative', @@ -3014,6 +3040,39 @@ def _execute_mixed_source_analyze_workflow( coverage={'terminal': True}, error='Narrative analysis could not be completed.', )) + if narrative_sources and narrative_result is not None and not partitions['tabular_sources']: + # Pure-narrative Analyze (no tabular sources): use run_document_analysis's own + # full-fidelity synthesis directly instead of the bounded evidence-envelope/collective- + # reduction hop below, which exists to combine different engine types (narrative + + # tabular) and caps each source's contribution far below what run_document_analysis + # already produced. + debug_print( + '[MIXED_SOURCE_ANALYZE] Pure narrative sources only | skipping bounded evidence handoff | ' + f'document_count={len(narrative_sources)} | ' + f"analysis_reply_chars={len(str(narrative_result.get('analysis_reply') or ''))}" + ) + if callable(activity_callback): + activity_callback({ + 'type': 'mixed_source_progress', + 'phase': 'complete', + 'label': 'Analysis complete', + 'status': EVIDENCE_STATUS_COMPLETED, + }) + return { + 'reply': narrative_result.get('reply') or narrative_result.get('analysis_reply'), + 'analysis_reply': narrative_result.get('analysis_reply'), + 'coverage': narrative_result.get('coverage') or {}, + 'documents': (narrative_result.get('coverage') or {}).get('documents') or [], + 'document_ids': narrative_result.get('document_ids') or [source.get('document_id') for source in narrative_sources], + 'raw_analysis_items': narrative_result.get('raw_analysis_items') or [], + 'document_analysis_items': narrative_result.get('document_analysis_items') or [], + 'analysis_intent': narrative_result.get('analysis_intent') or {}, + 'mixed_source_manifest': manifest, + 'mixed_source_evidence': [], + 'generated_tabular_outputs': [], + 'agent_citations': [], + } + tabular_sources = partitions['tabular_sources'] if tabular_sources: if callable(activity_callback): @@ -3084,6 +3143,12 @@ def _execute_mixed_source_analyze_workflow( telemetry_settings=settings, request_correlation_id=request_correlation_id, ) + debug_print( + '[MIXED_SOURCE_ANALYZE] Evidence handoff sized | ' + f'envelope_count={len(evidence_envelopes)} | ' + f'handoff_bytes={len(json.dumps(handoff, ensure_ascii=False, separators=(",", ":")).encode("utf-8"))} | ' + f"handoff_compacted={bool((handoff.get('mixed_source_coverage') or {}).get('handoff_compacted'))}" + ) mode_outcome = evaluate_mixed_source_mode_outcome( 'analyze', { @@ -3143,6 +3208,11 @@ def _execute_mixed_source_analyze_workflow( ) if not collective_reply: collective_reply = 'The selected sources could not be combined into a final analysis.' + debug_print( + '[MIXED_SOURCE_ANALYZE] Collective reduction completed | ' + f'evidence_envelope_count={len(evidence_envelopes)} | ' + f'collective_reply_chars={len(collective_reply)}' + ) emit_mixed_source_telemetry( settings, 'reduction', diff --git a/functional_tests/test_tabular_analyze_shared_preflight_adapter.py b/functional_tests/test_tabular_analyze_shared_preflight_adapter.py index 8b8c154b7..80e3ed78a 100644 --- a/functional_tests/test_tabular_analyze_shared_preflight_adapter.py +++ b/functional_tests/test_tabular_analyze_shared_preflight_adapter.py @@ -12,6 +12,7 @@ """ import ast +import json import logging import sys import time @@ -148,6 +149,8 @@ def fake_invoke_prompt(prompt, **kwargs): "emit_mixed_source_telemetry": lambda *args, **kwargs: False, "_build_mixed_source_analyze_reduction_prompt": lambda prompt, handoff: handoff["content"], "log_event": lambda *args, **kwargs: log_events.append({"args": args, "kwargs": kwargs}), + "debug_print": lambda *args, **kwargs: None, + "json": json, "logging": logging, "time": time, }