From 51a4bcb11caa6699917dcc6e188627d6c77c8713 Mon Sep 17 00:00:00 2001 From: Paul Lizer Date: Fri, 7 Aug 2026 17:58:15 -0400 Subject: [PATCH] Queue exhaustive tabular runs directly from CSV sources --- application/single_app/config.py | 2 +- application/single_app/route_backend_chats.py | 650 +++++++++++++----- .../test_tabular_row_orchestration_scale.py | 162 ++++- 3 files changed, 635 insertions(+), 179 deletions(-) diff --git a/application/single_app/config.py b/application/single_app/config.py index fc2c4d43..4efbe596 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.132" +VERSION = "0.250.133" IS_DEVELOPMENT = is_development_env_enabled() SESSION_COOKIE_SAMESITE = os.getenv('SESSION_COOKIE_SAMESITE', 'Lax') diff --git a/application/single_app/route_backend_chats.py b/application/single_app/route_backend_chats.py index 9a77ff0c..9e293e7e 100644 --- a/application/single_app/route_backend_chats.py +++ b/application/single_app/route_backend_chats.py @@ -64,6 +64,7 @@ import io import inspect import json +import math import mimetypes import os import app_settings_cache @@ -5841,6 +5842,210 @@ def _build_tabular_generated_output_query_descriptor( return descriptor +def _build_direct_tabular_generated_output_source(user_question, file_contexts, user_id, conversation_id, settings): + """Build a replayable full-CSV source descriptor without requiring a prior tool page.""" + generated_output_requested = question_requests_tabular_generated_output(user_question) + hierarchical_analysis_requested = question_requests_tabular_hierarchical_analysis(user_question) + durable_task_type = _get_tabular_generated_output_task_type( + generated_output_requested, + hierarchical_analysis_requested, + settings, + ) + analysis_only_requested = durable_task_type == TABULAR_RUN_TASK_HIERARCHICAL_ANALYSIS + combined_requested = durable_task_type == TABULAR_RUN_TASK_COMBINED + if not generated_output_requested and not analysis_only_requested: + return None + if hierarchical_analysis_requested and not generated_output_requested and not analysis_only_requested: + return None + + normalized_contexts = dedupe_tabular_file_contexts(file_contexts) + if len(normalized_contexts) != 1: + return None + + file_context = normalized_contexts[0] + file_name = str(file_context.get('file_name') or '').strip() + if not file_name.lower().endswith('.csv'): + return None + + from semantic_kernel_plugins.tabular_processing_plugin import TabularProcessingPlugin + + source_hint = str(file_context.get('source_hint') or 'workspace').strip().lower() or 'workspace' + group_id = file_context.get('group_id') + public_workspace_id = file_context.get('public_workspace_id') + tabular_plugin = TabularProcessingPlugin() + query_expression = 'index == index' + container_name = None + blob_path = None + storage_locator = file_context.get('storage_locator') if isinstance(file_context.get('storage_locator'), dict) else {} + if storage_locator.get('container') and storage_locator.get('blob_path'): + container_name = str(storage_locator.get('container') or '').strip() + blob_path = str(storage_locator.get('blob_path') or '').strip() + else: + container_name, blob_path = tabular_plugin._resolve_blob_location_with_fallback( + user_id, + conversation_id, + file_name, + source_hint, + group_id=group_id, + public_workspace_id=public_workspace_id, + ) + + query_result = tabular_plugin._query_csv_data_in_bounded_chunks( + container_name, + blob_path, + file_name, + query_expression, + return_columns=None, + start_row=0, + max_rows=1, + ) + result_payload = json.loads(str(query_result or '{}')) + row_count = _safe_int(result_payload.get('total_matches')) + if row_count <= 0: + raise ValueError('Direct tabular durable source had no CSV rows to process') + + internal_metadata = getattr(query_result, 'internal_metadata', {}) or {} + source_descriptor = internal_metadata.get('tabular_generated_export_source') or {} + source_authorization = internal_metadata.get('tabular_source_authorization') or {} + if not source_descriptor: + raise ValueError('Direct tabular durable source descriptor could not be created') + + batch_budget = _get_tabular_generated_output_batch_budget(settings) + source_descriptor = dict(source_descriptor) + source_descriptor.update({ + 'expected_row_count': row_count, + 'batch_max_rows': batch_budget['max_rows'], + 'batch_max_chars': batch_budget['max_chars'], + }) + output_format = get_tabular_generated_output_format(user_question) or 'md' + queued_output_format = 'md' if analysis_only_requested else output_format + return { + 'file_context': file_context, + 'source_candidate': { + 'function_name': 'query_tabular_data', + 'filename': file_name, + 'selected_sheet': '', + 'source_parameters': { + 'source': source_hint, + 'group_id': group_id, + 'public_workspace_id': public_workspace_id, + 'query_expression': query_expression, + 'return_columns': None, + }, + 'source_descriptor': source_descriptor, + 'source_authorization': source_authorization, + 'rows': [], + 'row_count': 0, + 'total_matches': row_count, + 'full_result_available': False, + 'page_count': 0, + 'diagnostics': [{ + 'function_name': 'direct_source_descriptor', + 'file_name': file_name, + 'total_matches': row_count, + 'full_result_available': False, + }], + }, + 'source_descriptor': source_descriptor, + 'task_type': durable_task_type, + 'analysis_objective': user_question if analysis_only_requested or combined_requested else None, + 'output_format': queued_output_format, + 'row_count': row_count, + 'batch_count_estimate': max(1, math.ceil(row_count / max(batch_budget['max_rows'], 1))), + 'analysis_only_requested': analysis_only_requested, + 'combined_requested': combined_requested, + } + + +def maybe_queue_direct_tabular_generated_output( + user_question, + file_contexts, + user_id, + conversation_id, + gpt_model, + settings, + thought_callback=None, + model_context=None, + cancel_requested=None, + 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 + + 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 + + def _build_tabular_generated_output_source_authorization(source_candidate): exact_source_authorization = (source_candidate or {}).get('source_authorization') or {} if ( @@ -12089,6 +12294,39 @@ def execute_source(source): if not file_context: raise ValueError('Authorized tabular source context is unavailable') + direct_generated_output = maybe_queue_direct_tabular_generated_output( + user_question=user_question, + file_contexts=[file_context], + user_id=user_id, + conversation_id=conversation_id, + gpt_model=gpt_model, + settings=settings, + thought_callback=publish_post_processing_thought, + model_context=model_context, + cancel_requested=cancel_requested, + request_correlation_id=request_correlation_id, + ) + if direct_generated_output: + generated_outputs.append(direct_generated_output) + system_messages.append({ + 'role': 'system', + 'content': _build_tabular_generated_output_system_message(direct_generated_output), + }) + return { + 'summary': ( + 'Queued a durable exhaustive tabular generated-output run from the authorized CSV source. ' + 'The final artifact will be published after checkpointed background processing completes.' + ), + 'evidence': [], + 'citations': [], + 'generated_artifacts': [direct_generated_output], + 'coverage': { + 'tool_call_count': 0, + 'execution_mode': 'direct_durable_source', + 'direct_source_backed': True, + }, + } + baseline_invocation_count = len( plugin_logger.get_invocations_for_conversation( user_id, @@ -17371,57 +17609,71 @@ def record_tabular_post_processing_thought(thought_payload): detail=f"files={tabular_filenames_str}; mode={tabular_execution_mode}", ) - tabular_analysis, streamed_tabular_tool_thoughts = asyncio.run(run_tabular_analysis_with_thought_tracking( - user_question=user_message, - tabular_filenames=workspace_tabular_files, - tabular_file_contexts=workspace_tabular_file_contexts, - user_id=user_id, - conversation_id=conversation_id, - gpt_model=gpt_model, - settings=settings, - source_hint=tabular_source_hint, - group_id=effective_active_group_id if tabular_source_hint == 'group' else None, - public_workspace_id=effective_active_public_workspace_id if tabular_source_hint == 'public' else None, - execution_mode=tabular_execution_mode, - thought_tracker=thought_tracker, - model_context=tabular_model_context, - )) - tabular_invocations = get_new_plugin_invocations( - plugin_logger.get_invocations_for_conversation(user_id, conversation_id, limit=1000), - baseline_tabular_invocation_count - ) + tabular_analysis = None + streamed_tabular_tool_thoughts = [] + tabular_invocations = [] tabular_related_document_summary = '' - tabular_related_document_stats = augment_tabular_invocations_with_related_document_evidence( - tabular_invocations, + tabular_generated_output = maybe_queue_direct_tabular_generated_output( user_message, + workspace_tabular_file_contexts, user_id, - conversation_id=conversation_id, + conversation_id, + gpt_model, + settings, + thought_callback=record_tabular_post_processing_thought, + model_context=tabular_model_context, ) - if tabular_related_document_stats.get('augmented_row_count'): - tabular_related_document_summary = build_tabular_related_document_evidence_summary( + if not tabular_generated_output: + tabular_analysis, streamed_tabular_tool_thoughts = asyncio.run(run_tabular_analysis_with_thought_tracking( + user_question=user_message, + tabular_filenames=workspace_tabular_files, + tabular_file_contexts=workspace_tabular_file_contexts, + user_id=user_id, + conversation_id=conversation_id, + gpt_model=gpt_model, + settings=settings, + source_hint=tabular_source_hint, + group_id=effective_active_group_id if tabular_source_hint == 'group' else None, + public_workspace_id=effective_active_public_workspace_id if tabular_source_hint == 'public' else None, + execution_mode=tabular_execution_mode, + thought_tracker=thought_tracker, + model_context=tabular_model_context, + )) + tabular_invocations = get_new_plugin_invocations( + plugin_logger.get_invocations_for_conversation(user_id, conversation_id, limit=1000), + baseline_tabular_invocation_count + ) + tabular_related_document_stats = augment_tabular_invocations_with_related_document_evidence( tabular_invocations, + user_message, + user_id, + conversation_id=conversation_id, ) - if not streamed_tabular_tool_thoughts: - tabular_thought_payloads = get_tabular_tool_thought_payloads(tabular_invocations) - for thought_content, thought_detail in tabular_thought_payloads: + if tabular_related_document_stats.get('augmented_row_count'): + tabular_related_document_summary = build_tabular_related_document_evidence_summary( + tabular_invocations, + ) + if not streamed_tabular_tool_thoughts: + tabular_thought_payloads = get_tabular_tool_thought_payloads(tabular_invocations) + for thought_content, thought_detail in tabular_thought_payloads: + thought_tracker.add_thought('tabular_analysis', thought_content, thought_detail) + tabular_status_thought_payloads = get_tabular_status_thought_payloads( + tabular_invocations, + analysis_succeeded=bool(tabular_analysis), + ) + for thought_content, thought_detail in tabular_status_thought_payloads: thought_tracker.add_thought('tabular_analysis', thought_content, thought_detail) - tabular_status_thought_payloads = get_tabular_status_thought_payloads( - tabular_invocations, - analysis_succeeded=bool(tabular_analysis), - ) - for thought_content, thought_detail in tabular_status_thought_payloads: - thought_tracker.add_thought('tabular_analysis', thought_content, thought_detail) - tabular_generated_output = asyncio.run(maybe_create_tabular_generated_output( - user_question=user_message, - invocations=tabular_invocations, - gpt_model=gpt_model, - settings=settings, - conversation_id=conversation_id, - thought_callback=record_tabular_post_processing_thought, - user_id=user_id, - model_context=tabular_model_context, - )) + tabular_generated_output = asyncio.run(maybe_create_tabular_generated_output( + user_question=user_message, + invocations=tabular_invocations, + gpt_model=gpt_model, + settings=settings, + conversation_id=conversation_id, + thought_callback=record_tabular_post_processing_thought, + user_id=user_id, + model_context=tabular_model_context, + )) if tabular_generated_output: generated_tabular_outputs_list.append(tabular_generated_output) generated_analysis_artifacts_list.append(tabular_generated_output) @@ -17720,54 +17972,72 @@ def record_tabular_post_processing_thought(thought_payload): detail=f"files={chat_tabular_filenames_str}; mode={chat_tabular_execution_mode}", ) - chat_tabular_analysis, streamed_chat_tabular_tool_thoughts = asyncio.run(run_tabular_analysis_with_thought_tracking( - user_question=user_message, - tabular_filenames=chat_tabular_files, - user_id=user_id, - conversation_id=conversation_id, - gpt_model=gpt_model, - settings=settings, - source_hint="chat", - execution_mode=chat_tabular_execution_mode, - thought_tracker=thought_tracker, - model_context=tabular_model_context, - )) - chat_tabular_invocations = get_new_plugin_invocations( - plugin_logger.get_invocations_for_conversation(user_id, conversation_id, limit=1000), - baseline_tabular_invocation_count - ) + chat_tabular_analysis = None + streamed_chat_tabular_tool_thoughts = [] + chat_tabular_invocations = [] chat_tabular_related_document_summary = '' - chat_tabular_related_document_stats = augment_tabular_invocations_with_related_document_evidence( - chat_tabular_invocations, + chat_tabular_file_contexts = [ + build_tabular_file_context(file_name, source_hint='chat') + for file_name in chat_tabular_files + ] + chat_tabular_generated_output = maybe_queue_direct_tabular_generated_output( user_message, + chat_tabular_file_contexts, user_id, - conversation_id=conversation_id, + conversation_id, + gpt_model, + settings, + thought_callback=record_tabular_post_processing_thought, + model_context=tabular_model_context, ) - if chat_tabular_related_document_stats.get('augmented_row_count'): - chat_tabular_related_document_summary = build_tabular_related_document_evidence_summary( + if not chat_tabular_generated_output: + chat_tabular_analysis, streamed_chat_tabular_tool_thoughts = asyncio.run(run_tabular_analysis_with_thought_tracking( + user_question=user_message, + tabular_filenames=chat_tabular_files, + user_id=user_id, + conversation_id=conversation_id, + gpt_model=gpt_model, + settings=settings, + source_hint="chat", + execution_mode=chat_tabular_execution_mode, + thought_tracker=thought_tracker, + model_context=tabular_model_context, + )) + chat_tabular_invocations = get_new_plugin_invocations( + plugin_logger.get_invocations_for_conversation(user_id, conversation_id, limit=1000), + baseline_tabular_invocation_count + ) + chat_tabular_related_document_stats = augment_tabular_invocations_with_related_document_evidence( + chat_tabular_invocations, + user_message, + user_id, + conversation_id=conversation_id, + ) + if chat_tabular_related_document_stats.get('augmented_row_count'): + chat_tabular_related_document_summary = build_tabular_related_document_evidence_summary( + chat_tabular_invocations, + ) + if not streamed_chat_tabular_tool_thoughts: + chat_tabular_thought_payloads = get_tabular_tool_thought_payloads(chat_tabular_invocations) + for thought_content, thought_detail in chat_tabular_thought_payloads: + thought_tracker.add_thought('tabular_analysis', thought_content, thought_detail) + chat_tabular_status_thought_payloads = get_tabular_status_thought_payloads( chat_tabular_invocations, + analysis_succeeded=bool(chat_tabular_analysis), ) - if not streamed_chat_tabular_tool_thoughts: - chat_tabular_thought_payloads = get_tabular_tool_thought_payloads(chat_tabular_invocations) - for thought_content, thought_detail in chat_tabular_thought_payloads: + for thought_content, thought_detail in chat_tabular_status_thought_payloads: thought_tracker.add_thought('tabular_analysis', thought_content, thought_detail) - chat_tabular_status_thought_payloads = get_tabular_status_thought_payloads( - chat_tabular_invocations, - analysis_succeeded=bool(chat_tabular_analysis), - ) - for thought_content, thought_detail in chat_tabular_status_thought_payloads: - thought_tracker.add_thought('tabular_analysis', thought_content, thought_detail) - chat_tabular_generated_output = asyncio.run(maybe_create_tabular_generated_output( - user_question=user_message, - invocations=chat_tabular_invocations, - gpt_model=gpt_model, - settings=settings, - conversation_id=conversation_id, - thought_callback=record_tabular_post_processing_thought, - user_id=user_id, - model_context=tabular_model_context, - )) + chat_tabular_generated_output = asyncio.run(maybe_create_tabular_generated_output( + user_question=user_message, + invocations=chat_tabular_invocations, + gpt_model=gpt_model, + settings=settings, + conversation_id=conversation_id, + thought_callback=record_tabular_post_processing_thought, + user_id=user_id, + model_context=tabular_model_context, + )) if chat_tabular_generated_output: generated_tabular_outputs_list.append(chat_tabular_generated_output) generated_analysis_artifacts_list.append(chat_tabular_generated_output) @@ -21192,62 +21462,74 @@ def record_and_publish_streaming_thought(thought_payload): detail=f"files={tabular_filenames_str}; mode={tabular_execution_mode}" ) - tabular_analysis, streamed_tabular_tool_thoughts = asyncio.run(run_tabular_analysis_with_thought_tracking( - user_question=user_message, - tabular_filenames=workspace_tabular_files, - tabular_file_contexts=workspace_tabular_file_contexts, - user_id=user_id, - conversation_id=conversation_id, - gpt_model=gpt_model, - settings=settings, - source_hint=tabular_source_hint, - group_id=effective_active_group_id if tabular_source_hint == 'group' else None, - public_workspace_id=effective_active_public_workspace_id if tabular_source_hint == 'public' else None, - execution_mode=tabular_execution_mode, - thought_tracker=thought_tracker, - live_thought_callback=publish_live_plugin_thought, - model_context=tabular_model_context, - )) - tabular_invocations = get_new_plugin_invocations( - plugin_logger.get_invocations_for_conversation(user_id, conversation_id, limit=1000), - baseline_tabular_invocation_count - ) + tabular_analysis = None + streamed_tabular_tool_thoughts = [] + tabular_invocations = [] tabular_related_document_summary = '' - tabular_related_document_stats = augment_tabular_invocations_with_related_document_evidence( - tabular_invocations, + tabular_generated_output = maybe_queue_direct_tabular_generated_output( user_message, + workspace_tabular_file_contexts, user_id, conversation_id=conversation_id, + thought_callback=record_and_publish_streaming_thought, + model_context=tabular_model_context, ) - if tabular_related_document_stats.get('augmented_row_count'): - tabular_related_document_summary = build_tabular_related_document_evidence_summary( + if not tabular_generated_output: + tabular_analysis, streamed_tabular_tool_thoughts = asyncio.run(run_tabular_analysis_with_thought_tracking( + user_question=user_message, + tabular_filenames=workspace_tabular_files, + tabular_file_contexts=workspace_tabular_file_contexts, + user_id=user_id, + conversation_id=conversation_id, + gpt_model=gpt_model, + settings=settings, + source_hint=tabular_source_hint, + group_id=effective_active_group_id if tabular_source_hint == 'group' else None, + public_workspace_id=effective_active_public_workspace_id if tabular_source_hint == 'public' else None, + execution_mode=tabular_execution_mode, + thought_tracker=thought_tracker, + live_thought_callback=publish_live_plugin_thought, + model_context=tabular_model_context, + )) + tabular_invocations = get_new_plugin_invocations( + plugin_logger.get_invocations_for_conversation(user_id, conversation_id, limit=1000), + baseline_tabular_invocation_count + ) + tabular_related_document_stats = augment_tabular_invocations_with_related_document_evidence( tabular_invocations, + user_message, + user_id, + conversation_id=conversation_id, ) - debug_print( - "[STREAMING][Tabular SK] Completed workspace tabular analysis | " - f"analysis_returned={bool(tabular_analysis)} | new_invocations={len(tabular_invocations)}" - ) - if not streamed_tabular_tool_thoughts: - tabular_thought_payloads = get_tabular_tool_thought_payloads(tabular_invocations) - for thought_content, thought_detail in tabular_thought_payloads: + if tabular_related_document_stats.get('augmented_row_count'): + tabular_related_document_summary = build_tabular_related_document_evidence_summary( + tabular_invocations, + ) + debug_print( + "[STREAMING][Tabular SK] Completed workspace tabular analysis | " + f"analysis_returned={bool(tabular_analysis)} | new_invocations={len(tabular_invocations)}" + ) + if not streamed_tabular_tool_thoughts: + tabular_thought_payloads = get_tabular_tool_thought_payloads(tabular_invocations) + for thought_content, thought_detail in tabular_thought_payloads: + yield emit_thought('tabular_analysis', thought_content, thought_detail) + tabular_status_thought_payloads = get_tabular_status_thought_payloads( + tabular_invocations, + analysis_succeeded=bool(tabular_analysis), + ) + for thought_content, thought_detail in tabular_status_thought_payloads: yield emit_thought('tabular_analysis', thought_content, thought_detail) - tabular_status_thought_payloads = get_tabular_status_thought_payloads( - tabular_invocations, - analysis_succeeded=bool(tabular_analysis), - ) - for thought_content, thought_detail in tabular_status_thought_payloads: - yield emit_thought('tabular_analysis', thought_content, thought_detail) - tabular_generated_output = asyncio.run(maybe_create_tabular_generated_output( - user_question=user_message, - invocations=tabular_invocations, - gpt_model=gpt_model, - settings=settings, - conversation_id=conversation_id, - thought_callback=record_and_publish_streaming_thought, - user_id=user_id, - model_context=tabular_model_context, - )) + tabular_generated_output = asyncio.run(maybe_create_tabular_generated_output( + user_question=user_message, + invocations=tabular_invocations, + gpt_model=gpt_model, + settings=settings, + conversation_id=conversation_id, + thought_callback=record_and_publish_streaming_thought, + user_id=user_id, + model_context=tabular_model_context, + )) if tabular_generated_output: generated_tabular_outputs_list.append(tabular_generated_output) generated_analysis_artifacts_list.append(tabular_generated_output) @@ -21558,59 +21840,77 @@ def record_and_publish_streaming_thought(thought_payload): detail=f"files={chat_tabular_filenames_str}; mode={chat_tabular_execution_mode}" ) - chat_tabular_analysis, streamed_chat_tabular_tool_thoughts = asyncio.run(run_tabular_analysis_with_thought_tracking( - user_question=user_message, - tabular_filenames=chat_tabular_files, - user_id=user_id, - conversation_id=conversation_id, - gpt_model=gpt_model, - settings=settings, - source_hint="chat", - execution_mode=chat_tabular_execution_mode, - thought_tracker=thought_tracker, - live_thought_callback=publish_live_plugin_thought, - model_context=tabular_model_context, - )) - chat_tabular_invocations = get_new_plugin_invocations( - plugin_logger.get_invocations_for_conversation(user_id, conversation_id, limit=1000), - baseline_tabular_invocation_count - ) + chat_tabular_analysis = None + streamed_chat_tabular_tool_thoughts = [] + chat_tabular_invocations = [] chat_tabular_related_document_summary = '' - chat_tabular_related_document_stats = augment_tabular_invocations_with_related_document_evidence( - chat_tabular_invocations, + chat_tabular_file_contexts = [ + build_tabular_file_context(file_name, source_hint='chat') + for file_name in chat_tabular_files + ] + chat_tabular_generated_output = maybe_queue_direct_tabular_generated_output( user_message, + chat_tabular_file_contexts, user_id, - conversation_id=conversation_id, + conversation_id, + gpt_model, + settings, + thought_callback=record_and_publish_streaming_thought, + model_context=tabular_model_context, ) - if chat_tabular_related_document_stats.get('augmented_row_count'): - chat_tabular_related_document_summary = build_tabular_related_document_evidence_summary( + if not chat_tabular_generated_output: + chat_tabular_analysis, streamed_chat_tabular_tool_thoughts = asyncio.run(run_tabular_analysis_with_thought_tracking( + user_question=user_message, + tabular_filenames=chat_tabular_files, + user_id=user_id, + conversation_id=conversation_id, + gpt_model=gpt_model, + settings=settings, + source_hint="chat", + execution_mode=chat_tabular_execution_mode, + thought_tracker=thought_tracker, + live_thought_callback=publish_live_plugin_thought, + model_context=tabular_model_context, + )) + chat_tabular_invocations = get_new_plugin_invocations( + plugin_logger.get_invocations_for_conversation(user_id, conversation_id, limit=1000), + baseline_tabular_invocation_count + ) + chat_tabular_related_document_stats = augment_tabular_invocations_with_related_document_evidence( chat_tabular_invocations, + user_message, + user_id, + conversation_id=conversation_id, ) - debug_print( - "[STREAMING][Chat Tabular SK] Completed chat-uploaded tabular analysis | " - f"analysis_returned={bool(chat_tabular_analysis)} | new_invocations={len(chat_tabular_invocations)}" - ) - if not streamed_chat_tabular_tool_thoughts: - chat_tabular_thought_payloads = get_tabular_tool_thought_payloads(chat_tabular_invocations) - for thought_content, thought_detail in chat_tabular_thought_payloads: + if chat_tabular_related_document_stats.get('augmented_row_count'): + chat_tabular_related_document_summary = build_tabular_related_document_evidence_summary( + chat_tabular_invocations, + ) + debug_print( + "[STREAMING][Chat Tabular SK] Completed chat-uploaded tabular analysis | " + f"analysis_returned={bool(chat_tabular_analysis)} | new_invocations={len(chat_tabular_invocations)}" + ) + if not streamed_chat_tabular_tool_thoughts: + chat_tabular_thought_payloads = get_tabular_tool_thought_payloads(chat_tabular_invocations) + for thought_content, thought_detail in chat_tabular_thought_payloads: + yield emit_thought('tabular_analysis', thought_content, thought_detail) + chat_tabular_status_thought_payloads = get_tabular_status_thought_payloads( + chat_tabular_invocations, + analysis_succeeded=bool(chat_tabular_analysis), + ) + for thought_content, thought_detail in chat_tabular_status_thought_payloads: yield emit_thought('tabular_analysis', thought_content, thought_detail) - chat_tabular_status_thought_payloads = get_tabular_status_thought_payloads( - chat_tabular_invocations, - analysis_succeeded=bool(chat_tabular_analysis), - ) - for thought_content, thought_detail in chat_tabular_status_thought_payloads: - yield emit_thought('tabular_analysis', thought_content, thought_detail) - chat_tabular_generated_output = asyncio.run(maybe_create_tabular_generated_output( - user_question=user_message, - invocations=chat_tabular_invocations, - gpt_model=gpt_model, - settings=settings, - conversation_id=conversation_id, - thought_callback=record_and_publish_streaming_thought, - user_id=user_id, - model_context=tabular_model_context, - )) + chat_tabular_generated_output = asyncio.run(maybe_create_tabular_generated_output( + user_question=user_message, + invocations=chat_tabular_invocations, + gpt_model=gpt_model, + settings=settings, + conversation_id=conversation_id, + thought_callback=record_and_publish_streaming_thought, + user_id=user_id, + model_context=tabular_model_context, + )) if chat_tabular_generated_output: generated_tabular_outputs_list.append(chat_tabular_generated_output) generated_analysis_artifacts_list.append(chat_tabular_generated_output) diff --git a/functional_tests/test_tabular_row_orchestration_scale.py b/functional_tests/test_tabular_row_orchestration_scale.py index 6f6a55cc..df762b99 100644 --- a/functional_tests/test_tabular_row_orchestration_scale.py +++ b/functional_tests/test_tabular_row_orchestration_scale.py @@ -1,8 +1,8 @@ # test_tabular_row_orchestration_scale.py """ Functional test for scalable per-row tabular orchestration. -Version: 0.250.132 -Implemented in: 0.250.060; generated CSV formula safety in 0.250.065; generated file export routing in 0.250.072; source descriptor generalization in 0.250.127; unified durable run contract in 0.250.128; hierarchical analysis in 0.250.129; combined analysis and export in 0.250.130; scale validation in 0.250.132 +Version: 0.250.133 +Implemented in: 0.250.060; generated CSV formula safety in 0.250.065; generated file export routing in 0.250.072; source descriptor generalization in 0.250.127; unified durable run contract in 0.250.128; hierarchical analysis in 0.250.129; combined analysis and export in 0.250.130; scale validation in 0.250.132; direct source-backed exhaustive queueing in 0.250.133 This test ensures generated exports preserve source identity and row order while enforcing one stable output schema across independently generated batches. @@ -22,7 +22,7 @@ import uuid from datetime import datetime, timezone from pathlib import Path -from types import SimpleNamespace +from types import ModuleType, SimpleNamespace from collections import Counter from typing import Any, Dict, Optional @@ -270,6 +270,33 @@ def _load_generated_output_router(route_dependencies): return namespace['maybe_create_tabular_generated_output'] +def _load_direct_source_queue_helpers(route_dependencies): + """Load direct source-backed queue helpers with focused dependency stubs.""" + module_tree = ast.parse(CHAT_ROUTE.read_text(encoding='utf-8'), filename=str(CHAT_ROUTE)) + helper_names = { + '_build_direct_tabular_generated_output_source', + 'maybe_queue_direct_tabular_generated_output', + } + selected_nodes = [ + node + for node in module_tree.body + if isinstance(node, ast.FunctionDef) and node.name in helper_names + ] + if len(selected_nodes) != len(helper_names): + raise AssertionError('Missing direct source-backed queue helper implementation') + + namespace = dict(route_dependencies) + namespace.setdefault('TABULAR_RUN_TASK_HIERARCHICAL_ANALYSIS', 'hierarchical_analysis') + namespace.setdefault('TABULAR_RUN_TASK_COMBINED', 'combined') + namespace.setdefault('json', json) + namespace.setdefault('math', math) + namespace.setdefault('inspect', SimpleNamespace(isawaitable=lambda value: False)) + namespace.setdefault('asyncio', SimpleNamespace(run=lambda value: value)) + extracted_module = ast.Module(body=selected_nodes, type_ignores=[]) + exec(compile(extracted_module, str(CHAT_ROUTE), 'exec'), namespace) + return namespace + + def _build_query_invocation(start_row, row_count, total_matches=300, source_etag='etag-source-7'): class InvocationPayload(dict): pass @@ -1297,6 +1324,134 @@ class MixedSourceCancellationError(Exception): assert output_metadata['task_type'] == 'combined' +def test_direct_source_backed_csv_queue_bypasses_tool_paging(): + """Explicit exhaustive CSV prompts queue directly from one authorized source blob.""" + original_module = sys.modules.get('semantic_kernel_plugins.tabular_processing_plugin') + fake_module = ModuleType('semantic_kernel_plugins.tabular_processing_plugin') + + class FakePluginResult(str): + def __new__(cls, value, internal_metadata=None): + instance = super().__new__(cls, value) + instance.internal_metadata = internal_metadata or {} + return instance + + class FakeTabularProcessingPlugin: + def _resolve_blob_location_with_fallback(self, user_id, conversation_id, filename, source, group_id=None, public_workspace_id=None): + assert user_id == 'user-1' + assert conversation_id == 'conversation-1' + assert filename == 'bank_treasury_operations_dataset-3000.csv' + assert source == 'workspace' + assert group_id is None + assert public_workspace_id is None + return 'user-documents', 'user-1/bank_treasury_operations_dataset-3000.csv' + + def _query_csv_data_in_bounded_chunks(self, container_name, blob_path, filename, query_expression, return_columns, start_row, max_rows): + assert container_name == 'user-documents' + assert blob_path == 'user-1/bank_treasury_operations_dataset-3000.csv' + assert filename == 'bank_treasury_operations_dataset-3000.csv' + assert query_expression == 'index == index' + assert return_columns is None + assert start_row == 0 + assert max_rows == 1 + return FakePluginResult( + json.dumps({'total_matches': 3000, 'data': [{'transaction_id': 'BT-000001'}]}), + internal_metadata={ + 'tabular_generated_export_source': { + 'version': 1, + 'kind': 'query_tabular_data', + 'source_function': 'query_tabular_data', + 'source': 'workspace', + 'scope_id': None, + 'container': container_name, + 'blob_path': blob_path, + 'blob_etag': 'etag-3000', + 'filename': filename, + 'query_expression': query_expression, + 'return_columns': return_columns, + 'expected_row_count': 3000, + }, + 'tabular_source_authorization': { + 'source': 'workspace', + 'scope_id': None, + 'container': container_name, + 'blob_path': blob_path, + 'blob_etag': 'etag-3000', + }, + }, + ) + + fake_module.TabularProcessingPlugin = FakeTabularProcessingPlugin + sys.modules['semantic_kernel_plugins.tabular_processing_plugin'] = fake_module + queued_runs = [] + thought_payloads = [] + + 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: 'combined' if generated and analysis else None, + 'question_requests_tabular_generated_output': lambda question: True, + 'question_requests_tabular_hierarchical_analysis': lambda question: True, + '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) or { + 'id': 'direct-run-3000', + 'task_type': kwargs.get('task_type'), + 'output_format': kwargs.get('output_format'), + 'row_count': kwargs['source_descriptor']['expected_row_count'], + 'batch_count': 50, + }, + 'build_background_tabular_generated_output_metadata': lambda run: { + 'background_export': True, + 'export_run_id': run['id'], + 'task_type': run['task_type'], + 'output_format': run['output_format'], + 'row_count': run['row_count'], + }, + 'build_tabular_post_processing_activity_payload': lambda *args, **kwargs: { + 'phase': kwargs.get('phase'), + }, + 'logging': logging, + 'log_event': lambda *args, **kwargs: None, + }) + + output_metadata = helpers['maybe_queue_direct_tabular_generated_output']( + user_question='For each row, answer each question, generate a CSV, and summarize risk patterns.', + file_contexts=[{ + 'file_name': 'bank_treasury_operations_dataset-3000.csv', + 'source_hint': 'workspace', + }], + user_id='user-1', + conversation_id='conversation-1', + gpt_model='test-model', + settings={}, + thought_callback=lambda payload: thought_payloads.append(payload), + model_context={'endpoint_id': 'model-1'}, + ) + 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 len(queued_runs) == 1 + queued_run = queued_runs[0] + assert queued_run['row_batches'] is None + assert queued_run['task_type'] == 'combined' + assert queued_run['output_format'] == 'csv' + assert queued_run['analysis_objective'].startswith('For each row') + assert queued_run['source_descriptor']['expected_row_count'] == 3000 + assert queued_run['source_descriptor']['batch_max_rows'] == 60 + assert output_metadata['background_export'] is True + assert output_metadata['task_type'] == 'combined' + assert thought_payloads + assert 'run_id=direct-run-3000' in thought_payloads[0]['detail'] + + 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() @@ -2237,6 +2392,7 @@ def main(): test_filter_rows_pages_queue_full_3000_row_source_replay, 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_hierarchical_analysis_routing_requires_feature_flag, test_non_replayable_filter_rows_reports_explicit_failure, test_streaming_finalizer_writes_30000_rows_in_bounded_chunks,