Skip to content

Follow-up: keep direct tabular queue failures from aborting streams #1176

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
Paul Lizer (paullizer) merged 1 commit into from
Aug 8, 2026
+149 −77
Merged

Follow-up: keep direct tabular queue failures from aborting streams #1176

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
165 changes: 88 additions & 77 deletions application/single_app/route_backend_chats.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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):
Expand Down Expand Up @@ -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:
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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:
Expand Down Expand Up @@ -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}"
Expand Down
61 changes: 61 additions & 0 deletions functional_tests/test_tabular_row_orchestration_scale.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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()
Expand Down Expand Up @@ -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,
Expand Down