diff --git a/application/single_app/route_backend_chats.py b/application/single_app/route_backend_chats.py index 9e293e7e..3beff797 100644 --- a/application/single_app/route_backend_chats.py +++ b/application/single_app/route_backend_chats.py @@ -5970,80 +5970,96 @@ def maybe_queue_direct_tabular_generated_output( request_correlation_id=None, ): """Queue an exhaustive CSV-backed generated-output run directly from an authorized source.""" - direct_source = _build_direct_tabular_generated_output_source( - user_question, - file_contexts, - user_id, - conversation_id, - settings, - ) - if not direct_source: - return None + try: + direct_source = _build_direct_tabular_generated_output_source( + user_question, + file_contexts, + user_id, + conversation_id, + settings, + ) + if not direct_source: + return None - raise_if_mixed_source_cancelled( - cancel_requested, - 'export', - request_correlation_id=request_correlation_id, - ) - background_run = queue_tabular_generated_output_run( - user_id=user_id, - conversation_id=conversation_id, - user_question=user_question, - source_candidate=direct_source['source_candidate'], - output_format=direct_source['output_format'], - row_batches=None, - gpt_model=gpt_model, - settings=settings, - model_context=model_context, - source_descriptor=direct_source['source_descriptor'], - task_type=direct_source.get('task_type') or None, - analysis_objective=direct_source.get('analysis_objective'), - ) - background_metadata = build_background_tabular_generated_output_metadata(background_run) - if callable(thought_callback): - output_label = str(direct_source['output_format'] or 'json').upper() - if direct_source.get('combined_requested'): - title = 'Queued exhaustive tabular analysis and export from the selected CSV source' - elif direct_source.get('analysis_only_requested'): - title = 'Queued exhaustive tabular analysis from the selected CSV source' - else: - title = f'Queued exhaustive {output_label} export from the selected CSV source' - thought_payload = { - 'step_type': 'tabular_analysis', - 'content': title, - 'detail': ( - f"run_id={background_metadata.get('export_run_id')}; " - f"rows={direct_source['row_count']}; batches~={direct_source['batch_count_estimate']}; checkpointed=true" - ), - 'activity': build_tabular_post_processing_activity_payload( - 'tabular.generated_output', - title, - 'running', - phase='queued', - output_format=direct_source['output_format'], - file_name=direct_source['source_candidate'].get('filename'), - batch_index=0, - batch_count=direct_source['batch_count_estimate'], - ), - } - maybe_callback_result = thought_callback(thought_payload) - if inspect.isawaitable(maybe_callback_result): - asyncio.run(maybe_callback_result) + raise_if_mixed_source_cancelled( + cancel_requested, + 'export', + request_correlation_id=request_correlation_id, + ) + background_run = queue_tabular_generated_output_run( + user_id=user_id, + conversation_id=conversation_id, + user_question=user_question, + source_candidate=direct_source['source_candidate'], + output_format=direct_source['output_format'], + row_batches=None, + gpt_model=gpt_model, + settings=settings, + model_context=model_context, + source_descriptor=direct_source['source_descriptor'], + task_type=direct_source.get('task_type') or None, + analysis_objective=direct_source.get('analysis_objective'), + ) + background_metadata = build_background_tabular_generated_output_metadata(background_run) + if callable(thought_callback): + output_label = str(direct_source['output_format'] or 'json').upper() + if direct_source.get('combined_requested'): + title = 'Queued exhaustive tabular analysis and export from the selected CSV source' + elif direct_source.get('analysis_only_requested'): + title = 'Queued exhaustive tabular analysis from the selected CSV source' + else: + title = f'Queued exhaustive {output_label} export from the selected CSV source' + thought_payload = { + 'step_type': 'tabular_analysis', + 'content': title, + 'detail': ( + f"run_id={background_metadata.get('export_run_id')}; " + f"rows={direct_source['row_count']}; batches~={direct_source['batch_count_estimate']}; checkpointed=true" + ), + 'activity': build_tabular_post_processing_activity_payload( + 'tabular.generated_output', + title, + 'running', + phase='queued', + output_format=direct_source['output_format'], + file_name=direct_source['source_candidate'].get('filename'), + batch_index=0, + batch_count=direct_source['batch_count_estimate'], + ), + } + maybe_callback_result = thought_callback(thought_payload) + if inspect.isawaitable(maybe_callback_result): + asyncio.run(maybe_callback_result) - log_event( - '[TABULAR_GENERATED_OUTPUT] Queued direct source-backed generated output run', - { - 'conversation_id': conversation_id, - 'source_file_name': direct_source['source_candidate'].get('filename'), - 'row_count': direct_source['row_count'], - 'batch_count_estimate': direct_source['batch_count_estimate'], - 'task_type': direct_source.get('task_type') or 'structured_export', - 'output_format': direct_source['output_format'], - 'export_run_id': background_metadata.get('export_run_id'), - }, - level=logging.INFO, - ) - return background_metadata + log_event( + '[TABULAR_GENERATED_OUTPUT] Queued direct source-backed generated output run', + { + 'conversation_id': conversation_id, + 'source_file_name': direct_source['source_candidate'].get('filename'), + 'row_count': direct_source['row_count'], + 'batch_count_estimate': direct_source['batch_count_estimate'], + 'task_type': direct_source.get('task_type') or 'structured_export', + 'output_format': direct_source['output_format'], + 'export_run_id': background_metadata.get('export_run_id'), + }, + level=logging.INFO, + ) + return background_metadata + except MixedSourceCancellationError: + raise + except Exception as exc: + log_event( + '[TABULAR_GENERATED_OUTPUT] Direct source-backed generated output queueing skipped', + { + 'conversation_id': conversation_id, + 'file_count': len(file_contexts or []), + 'error_type': exc.__class__.__name__, + 'error': str(exc)[:500], + }, + level=logging.WARNING, + exceptionTraceback=True, + ) + return None def _build_tabular_generated_output_source_authorization(source_candidate): @@ -16197,7 +16213,6 @@ def result_requires_message_reload(result: Any) -> bool: debug_print(f"Error retrieving group details: {e}") if 'workspace_search' in user_metadata: user_metadata['workspace_search']['group_name'] = None - import traceback traceback.print_exc() if effective_document_scope == 'public' and effective_active_public_workspace_id: @@ -17199,7 +17214,6 @@ def result_requires_message_reload(result: Any) -> bool: except Exception as e: debug_print(f"Error retrieving group name for chat context: {e}") user_metadata['chat_context']['group_name'] = None - import traceback traceback.print_exc() elif message_chat_type == 'public': # For public chat, add workspace information if available from document selection @@ -19669,7 +19683,6 @@ def stream_cancel_requested(): debug_print(f"[DEBUG] user_enable_agents={user_enable_agents}") except Exception as e: debug_print(f"Error loading user settings: {e}") - import traceback traceback.print_exc() # Streaming does not support image generation @@ -20421,7 +20434,6 @@ def build_streaming_capability_usage(): except Exception as e: debug_print(f"Error retrieving group details: {e}") user_metadata['workspace_search']['group_name'] = None - import traceback traceback.print_exc() if effective_document_scope == 'public' and effective_active_public_workspace_id: @@ -22465,7 +22477,6 @@ def finalize_cancelled_agent_stream_response(): continue raise except Exception as stream_error: - import traceback plugin_logger_cb.deregister_callbacks(callback_key) debug_print( f"[STREAMING][Plugin Callback] Deregistered callback after streaming error for key={callback_key}" diff --git a/functional_tests/test_tabular_row_orchestration_scale.py b/functional_tests/test_tabular_row_orchestration_scale.py index df762b99..d8ede1ba 100644 --- a/functional_tests/test_tabular_row_orchestration_scale.py +++ b/functional_tests/test_tabular_row_orchestration_scale.py @@ -292,6 +292,7 @@ def _load_direct_source_queue_helpers(route_dependencies): namespace.setdefault('math', math) namespace.setdefault('inspect', SimpleNamespace(isawaitable=lambda value: False)) namespace.setdefault('asyncio', SimpleNamespace(run=lambda value: value)) + namespace.setdefault('MixedSourceCancellationError', type('MixedSourceCancellationError', (Exception,), {})) extracted_module = ast.Module(body=selected_nodes, type_ignores=[]) exec(compile(extracted_module, str(CHAT_ROUTE), 'exec'), namespace) return namespace @@ -1452,6 +1453,65 @@ def _query_csv_data_in_bounded_chunks(self, container_name, blob_path, filename, assert 'run_id=direct-run-3000' in thought_payloads[0]['detail'] +def test_direct_source_backed_queue_failure_falls_back_without_stream_abort(): + """Direct queue failures return None so existing tabular analysis fallback can continue.""" + original_module = sys.modules.get('semantic_kernel_plugins.tabular_processing_plugin') + fake_module = ModuleType('semantic_kernel_plugins.tabular_processing_plugin') + + class FakeTabularProcessingPlugin: + def _resolve_blob_location_with_fallback(self, *args, **kwargs): + raise RuntimeError('simulated credential challenge') + + fake_module.TabularProcessingPlugin = FakeTabularProcessingPlugin + sys.modules['semantic_kernel_plugins.tabular_processing_plugin'] = fake_module + logged_events = [] + queued_runs = [] + + try: + helpers = _load_direct_source_queue_helpers({ + '_safe_int': lambda value: int(value or 0), + '_get_tabular_generated_output_batch_budget': lambda settings=None: { + 'max_rows': 60, + 'max_chars': 60000, + }, + '_get_tabular_generated_output_task_type': lambda generated, analysis, settings: None, + 'question_requests_tabular_generated_output': lambda question: True, + 'question_requests_tabular_hierarchical_analysis': lambda question: False, + 'get_tabular_generated_output_format': lambda question: 'csv', + 'dedupe_tabular_file_contexts': lambda contexts=None: list(contexts or []), + 'raise_if_mixed_source_cancelled': lambda *args, **kwargs: None, + 'queue_tabular_generated_output_run': lambda **kwargs: queued_runs.append(kwargs), + 'build_background_tabular_generated_output_metadata': lambda run: run, + 'build_tabular_post_processing_activity_payload': lambda *args, **kwargs: {}, + 'logging': logging, + 'log_event': lambda *args, **kwargs: logged_events.append((args, kwargs)), + }) + + output_metadata = helpers['maybe_queue_direct_tabular_generated_output']( + user_question='For each row, answer each question and generate a CSV.', + file_contexts=[{ + 'file_name': 'interior_resource_operations_dataset-30000.csv', + 'source_hint': 'workspace', + }], + user_id='user-1', + conversation_id='conversation-1', + gpt_model='test-model', + settings={}, + ) + finally: + if original_module is None: + sys.modules.pop('semantic_kernel_plugins.tabular_processing_plugin', None) + else: + sys.modules['semantic_kernel_plugins.tabular_processing_plugin'] = original_module + + assert output_metadata is None + assert queued_runs == [] + assert any( + args and args[0] == '[TABULAR_GENERATED_OUTPUT] Direct source-backed generated output queueing skipped' + for args, _kwargs in logged_events + ) + + def test_hierarchical_analysis_routing_requires_feature_flag(): """Lane C queueing stays behind the feature flag until scale hardening completes.""" candidate_helpers = _load_candidate_helpers() @@ -2393,6 +2453,7 @@ def main(): test_filter_rows_pages_queue_hierarchical_analysis_run, test_filter_rows_pages_queue_combined_analysis_and_export_run, test_direct_source_backed_csv_queue_bypasses_tool_paging, + test_direct_source_backed_queue_failure_falls_back_without_stream_abort, test_hierarchical_analysis_routing_requires_feature_flag, test_non_replayable_filter_rows_reports_explicit_failure, test_streaming_finalizer_writes_30000_rows_in_bounded_chunks,