From 361fae679c273c9dda02337ea029e79ff07cee2b Mon Sep 17 00:00:00 2001 From: Paul Lizer Date: Wed, 19 Aug 2026 12:49:55 -0400 Subject: [PATCH] fix(tabular): union passthrough CSV schema and keep assistant text visible Passthrough CSV exports staged already-final rows, but _build_passthrough_batch_results derived the output schema from row 1 only and required every later row to match it exactly. Rows merged from multiple action results (or records with optional fields) failed with "Generated output schema mismatch at row N", were classified as model_validation retries even though no model runs on this path, and the run failed after exhausting its retries. queue_tabular_generated_output_run now unions column names across every staged passthrough batch and pins output_schema, public_output_schema, and internal_checkpoint_schema up front. _build_passthrough_batch_results pads sparse rows to that schema, falling back to a per-batch union for runs queued before this change. Model-generated runs still defer to batch-1 schema discovery. Also stop hiding the assistant response when a CSV artifact card renders. suppress_assistant_text exists for JSON/XML, whose payload is withheld from the streamed text and replaced with a placeholder. CSV narratives stream intact, so the card must append below them instead of replacing them. Added ASSISTANT_TEXT_SUPPRESSING_FORMATS and reused it for background structured exports; Markdown analysis artifacts keep suppression as a real handoff. Version: 0.260.005 --- application/single_app/config.py | 2 +- .../functions_generated_file_exports.py | 5 +- .../functions_tabular_generated_exports.py | 70 +++++- .../test_assistant_table_csv_artifact.py | 39 ++- ..._tabular_passthrough_heterogeneous_rows.py | 232 ++++++++++++++++++ 5 files changed, 338 insertions(+), 10 deletions(-) create mode 100644 functional_tests/test_tabular_passthrough_heterogeneous_rows.py diff --git a/application/single_app/config.py b/application/single_app/config.py index e3f9a3646..0f2645f31 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.260.003" +VERSION = "0.260.005" IS_DEVELOPMENT = is_development_env_enabled() SESSION_COOKIE_SAMESITE = os.getenv('SESSION_COOKIE_SAMESITE', 'Lax') diff --git a/application/single_app/functions_generated_file_exports.py b/application/single_app/functions_generated_file_exports.py index edd46aa24..5055da7dd 100644 --- a/application/single_app/functions_generated_file_exports.py +++ b/application/single_app/functions_generated_file_exports.py @@ -31,6 +31,9 @@ GENERATED_FILE_FORMAT_PDF, } SUPPORTED_GENERATED_EXPORT_FORMATS = {'csv', 'json', 'xml'} +# Only these payloads are withheld from the streamed assistant text, so only their cards +# replace it. CSV narratives stream intact and must stay visible above the artifact card. +ASSISTANT_TEXT_SUPPRESSING_FORMATS = {'json', 'xml'} GENERATED_FILE_PREVIEW_ROWS = 3 REQUESTED_ARTIFACT_FORMATS = ('csv', 'json', 'xml', 'md', 'docx', 'pdf') STRUCTURED_ARTIFACT_FORMAT_MARKERS = { @@ -597,7 +600,7 @@ def build_generated_file_artifact_metadata( 'file_name': uploaded_message.get('file_name') or generated_file_name, 'output_format': normalized_output_format, 'summary': str(export_payload.get('summary') or '').strip(), - 'suppress_assistant_text': normalized_output_format in {'csv', 'json', 'xml'}, + 'suppress_assistant_text': normalized_output_format in ASSISTANT_TEXT_SUPPRESSING_FORMATS, } row_count = export_payload.get('row_count') if isinstance(row_count, int) and row_count > 0: diff --git a/application/single_app/functions_tabular_generated_exports.py b/application/single_app/functions_tabular_generated_exports.py index 8209b626e..cd2076ca1 100644 --- a/application/single_app/functions_tabular_generated_exports.py +++ b/application/single_app/functions_tabular_generated_exports.py @@ -65,6 +65,7 @@ ) from functions_group import assert_group_role from functions_generated_file_exports import ( + ASSISTANT_TEXT_SUPPRESSING_FORMATS, normalize_generated_output_format, serialize_generated_json, serialize_generated_xml, @@ -2000,6 +2001,38 @@ def _prepare_tabular_source_rows(rows, start_row=0, token_namespace=''): return prepared_rows +def _extend_passthrough_output_field_names(field_names, rows): + """Union already-final passthrough row fields so sparse rows still share one schema.""" + ordered_fields = list(field_names or []) + seen_fields = set(ordered_fields) + for row in rows or []: + if not isinstance(row, dict): + continue + for field_name in row: + normalized_field = str(field_name) + if normalized_field in seen_fields or is_analysis_internal_lineage_field(normalized_field): + continue + seen_fields.add(normalized_field) + ordered_fields.append(normalized_field) + return ordered_fields + + +def _align_passthrough_rows_to_output_fields(rows, public_output_fields): + """Give every passthrough row the same field set before schema validation.""" + aligned_rows = [] + for row in rows or []: + if not isinstance(row, dict): + aligned_rows.append(row) + continue + aligned_row = { + TABULAR_EXPORT_INPUT_ROW_TOKEN_FIELD: row.get(TABULAR_EXPORT_INPUT_ROW_TOKEN_FIELD), + } + for field_name in public_output_fields: + aligned_row[field_name] = row.get(field_name, '') + aligned_rows.append(aligned_row) + return aligned_rows + + def _normalize_generated_batch_entries( source_rows, generated_entries, @@ -9270,7 +9303,7 @@ def _publish_structured_export_artifact(run, descriptor=None): generated_file_name, output_format, preview_rows=_build_structured_export_preview_rows(run_for_output), - suppress_assistant_text=True, + suppress_assistant_text=output_format in ASSISTANT_TEXT_SUPPRESSING_FORMATS, ) artifact_metadata['artifact_id'] = member_id artifact_metadata['member_id'] = member_id @@ -10214,9 +10247,14 @@ def _build_passthrough_batch_results(run, batch_requests): generated_results = [] for batch_request in batch_requests: batch_started_at = time.monotonic() + source_rows = batch_request['rows'] + public_output_fields = ( + _get_public_fields_from_output_schema(expected_output_schema) + or _extend_passthrough_output_field_names([], source_rows) + ) batch_entries, output_schema = _normalize_generated_batch_entries( - batch_request['rows'], - batch_request['rows'], + source_rows, + _align_passthrough_rows_to_output_fields(source_rows, public_output_fields), expected_output_schema=expected_output_schema, ) if not expected_output_schema: @@ -11179,6 +11217,7 @@ def queue_tabular_generated_output_run( passthrough_input_rows=passthrough_input_rows, ) retry_mode = _select_tabular_retry_mode(rollout_settings, executor_mode) + passthrough_output_fields = [] if source_descriptor: staged_row_count = _safe_int(source_descriptor.get('expected_row_count')) @@ -11262,6 +11301,11 @@ def queue_tabular_generated_output_run( start_row=staged_row_count, token_namespace=run_id, ) + if passthrough_input_rows: + passthrough_output_fields = _extend_passthrough_output_field_names( + passthrough_output_fields, + prepared_batch_rows, + ) _upload_json_blob( _input_blob_path(normalized_user_id, normalized_conversation_id, run_id, index), prepared_batch_rows, @@ -11280,6 +11324,15 @@ def queue_tabular_generated_output_run( if not staged_batch_count or not staged_row_count: raise ValueError('At least one source row is required for a background tabular export') + passthrough_output_schema = ( + [ + TABULAR_EXPORT_OUTPUT_ROW_NUMBER_FIELD, + TABULAR_EXPORT_OUTPUT_ROW_IDENTITY_FIELD, + ] + passthrough_output_fields + if passthrough_output_fields + else [] + ) + chunk_manifest = _write_chunk_manifest_for_run( normalized_user_id, normalized_conversation_id, @@ -11363,9 +11416,14 @@ def queue_tabular_generated_output_run( 'processed_rows': 0, # Only lock the schema up front when real output columns are already known; otherwise # defer to batch-1 discovery instead of validating against a lineage-only placeholder. - 'output_schema': contract_internal_checkpoint_schema if contract_public_output_schema else None, - 'public_output_schema': contract_public_output_schema, - 'internal_checkpoint_schema': contract_internal_checkpoint_schema, + # Passthrough rows are already final output, so their union schema is known at queue time. + 'output_schema': ( + contract_internal_checkpoint_schema + if contract_public_output_schema + else passthrough_output_schema or None + ), + 'public_output_schema': contract_public_output_schema or passthrough_output_fields, + 'internal_checkpoint_schema': contract_internal_checkpoint_schema or passthrough_output_schema, 'lineage_schema': [ TABULAR_EXPORT_OUTPUT_ROW_NUMBER_FIELD, TABULAR_EXPORT_OUTPUT_ROW_IDENTITY_FIELD, diff --git a/functional_tests/test_assistant_table_csv_artifact.py b/functional_tests/test_assistant_table_csv_artifact.py index 93996a06d..256305ccf 100644 --- a/functional_tests/test_assistant_table_csv_artifact.py +++ b/functional_tests/test_assistant_table_csv_artifact.py @@ -2,8 +2,8 @@ #!/usr/bin/env python3 """ Functional test for assistant-rendered table CSV artifacts. -Version: 0.250.178 -Implemented in: 0.241.050; non-tabular document CSV parsing in 0.250.065; generated file export framework in 0.250.072; updated in 0.250.073; linear fence parsing coverage in 0.250.112; version assertion compatibility updated in 0.250.172; Word function-result serialization restored in 0.250.178 +Version: 0.260.005 +Implemented in: 0.241.050; non-tabular document CSV parsing in 0.250.065; generated file export framework in 0.250.072; updated in 0.250.073; linear fence parsing coverage in 0.250.112; version assertion compatibility updated in 0.250.172; Word function-result serialization restored in 0.250.178; CSV assistant-text preservation in 0.260.005 This test ensures that explicit table-format requests with assistant-rendered tables, including CSV rows extracted from non-tabular documents, are converted @@ -1061,6 +1061,40 @@ def test_workflow_generated_file_artifacts_reuse_shared_contract(): ) +def test_csv_artifact_card_appends_below_assistant_response(): + """CSV narratives stream intact, so their card must never hide the assistant message.""" + print('Testing CSV artifact assistant-text preservation...') + assert_app_version_at_least('0.260.005') + + csv_metadata = build_generated_file_artifact_metadata( + { + 'capability': 'file_export', + 'file_name': 'generated_output_20260819_153747.csv', + 'output_format': 'csv', + 'row_count': 3, + 'summary': 'Generated CSV artifact.', + }, + {'message': {'id': 'artifact-csv', 'file_name': 'generated_output_20260819_153747.csv'}}, + 'conversation-1', + ) + assert_true( + csv_metadata['suppress_assistant_text'] is False, + 'Expected the CSV artifact card to append below the assistant response, not replace it.', + ) + + background_export_content = read_text(BACKGROUND_EXPORT_FILE) + assert_true( + 'suppress_assistant_text=output_format in ASSISTANT_TEXT_SUPPRESSING_FORMATS' in background_export_content, + 'Expected background structured exports to reuse the shared assistant-text suppression contract.', + ) + + chat_route_content = read_text(CHAT_ROUTE_FILE) + assert_true( + "suppress_streamed_file_payload = requested_streamed_file_format in {'json', 'xml'}" in chat_route_content, + 'Expected only JSON/XML payloads to be withheld from the streamed assistant text.', + ) + + def test_chat_route_wires_assistant_table_artifacts(): print('Testing chat route assistant-table artifact plumbing...') @@ -1177,6 +1211,7 @@ def run_tests() -> bool: test_universal_csv_request_variants_are_recognized, test_csv_schema_clarification_guidance_is_specific_and_resumable, test_workflow_generated_file_artifacts_reuse_shared_contract, + test_csv_artifact_card_appends_below_assistant_response, test_chat_route_wires_assistant_table_artifacts, ] diff --git a/functional_tests/test_tabular_passthrough_heterogeneous_rows.py b/functional_tests/test_tabular_passthrough_heterogeneous_rows.py new file mode 100644 index 000000000..0c9e0ae9e --- /dev/null +++ b/functional_tests/test_tabular_passthrough_heterogeneous_rows.py @@ -0,0 +1,232 @@ +# test_tabular_passthrough_heterogeneous_rows.py +#!/usr/bin/env python3 +""" +Functional test for heterogeneous passthrough rows in background CSV exports. +Version: 0.260.004 +Implemented in: 0.260.004 + +Chat and workflow CSV artifacts queue already-final rows with +passthrough_input_rows=True. When those rows come from more than one action +result -- or from records with optional fields -- the per-row field sets +differ. The background export used to derive its schema from the first row +only, hard-fail with "Generated output schema mismatch at row 2", requeue the +run as a model-validation failure, and eventually fail permanently even though +no model is involved in a passthrough run. + +This test calls the *real* queue_tabular_generated_output_run() and the *real* +_build_passthrough_batch_results() (extracted from +functions_tabular_generated_exports.py via AST, with only genuine I/O +boundaries stubbed) to ensure the union output schema is pinned at queue time +and sparse rows are padded instead of rejected. +""" + +import ast +import sys +from pathlib import Path + +ROOT = Path(__file__).resolve().parents[1] +APP_ROOT = ROOT / "application" / "single_app" +if str(APP_ROOT) not in sys.path: + sys.path.insert(0, str(APP_ROOT)) +if str(Path(__file__).resolve().parent) not in sys.path: + sys.path.insert(0, str(Path(__file__).resolve().parent)) + +from test_support.versioning import assert_app_version_at_least # noqa: E402 +from test_tabular_queue_run_output_schema_end_to_end import ( # noqa: E402 + EXPORT_MODULE, + _build_namespace, + _collect_local_closure, + _module_node_name, + _resolve_cross_module_names, +) + +IMPLEMENTED_VERSION = "0.260.004" + +LINEAGE_SCHEMA = ["source_row_number", "source_row_identity"] + +# Two action results merged into one chat CSV artifact: the first group carries a +# "facts" column, the second carries an entirely different telemetry envelope. +HETEROGENEOUS_ROWS = [ + {"name": "simulator", "facts": "1 instance online"}, + { + "application": "yamcs", + "schema_version": "1.0", + "runtime": "realtime", + "message_metadata": "BatteryVoltage1", + }, + {"name": "simulator", "facts": "still online", "runtime": "realtime"}, +] + +EXPECTED_PUBLIC_SCHEMA = [ + "name", + "facts", + "application", + "schema_version", + "runtime", + "message_metadata", +] + + +def _build_passthrough_namespace(): + """Extend the real queue-function namespace with the real passthrough helpers.""" + namespace, fake_container = _build_namespace() + tree = ast.parse(EXPORT_MODULE.read_text(encoding="utf-8"), filename=str(EXPORT_MODULE)) + included, unresolved = _collect_local_closure( + tree, + {"_build_passthrough_batch_results", "_prepare_tabular_source_rows"}, + ) + still_needed = {name for name in unresolved if name not in namespace} + resolved_cross_module, still_unresolved = _resolve_cross_module_names(still_needed) + namespace.update(resolved_cross_module) + if still_unresolved: + raise AssertionError( + f"Could not resolve required names for the passthrough checkpoint test: {sorted(still_unresolved)}" + ) + + included_names = set(included) + ordered_nodes = [node for node in tree.body if _module_node_name(node) in included_names] + module = ast.Module(body=ordered_nodes, type_ignores=[]) + ast.fix_missing_locations(module) + exec(compile(module, str(EXPORT_MODULE), "exec"), namespace) + return namespace, fake_container + + +def _queue_passthrough_run(namespace, row_batches): + return namespace["queue_tabular_generated_output_run"]( + user_id="user-1", + conversation_id="conversation-1", + user_question="generate a csv", + source_candidate={ + "filename": "generated_output_20260819_152402.csv", + "selected_sheet": "", + "source_authorization": {"source": "chat"}, + }, + output_format="csv", + row_batches=row_batches, + gpt_model="", + settings={}, + passthrough_input_rows=True, + ) + + +def test_queue_pins_union_schema_for_heterogeneous_passthrough_rows(): + """The union of every staged passthrough column must be locked before batch 1 runs.""" + assert_app_version_at_least(IMPLEMENTED_VERSION) + namespace, fake_container = _build_passthrough_namespace() + + run = _queue_passthrough_run(namespace, [HETEROGENEOUS_ROWS[:2], HETEROGENEOUS_ROWS[2:]]) + + assert fake_container.created_items, "the real queue function must create one Cosmos run item" + persisted_run = fake_container.created_items[0] + assert persisted_run["passthrough_input_rows"] is True + assert persisted_run["public_output_schema"] == EXPECTED_PUBLIC_SCHEMA, ( + "columns from every staged batch must be unioned, not taken from the first row; " + f"got {persisted_run['public_output_schema']!r}" + ) + assert persisted_run["output_schema"] == LINEAGE_SCHEMA + EXPECTED_PUBLIC_SCHEMA + assert persisted_run["internal_checkpoint_schema"] == LINEAGE_SCHEMA + EXPECTED_PUBLIC_SCHEMA + assert run["output_schema"] == persisted_run["output_schema"] + + +def test_model_backed_runs_still_defer_schema_discovery(): + """Non-passthrough runs must keep deferring to batch-1 schema discovery.""" + assert_app_version_at_least(IMPLEMENTED_VERSION) + namespace, fake_container = _build_passthrough_namespace() + + namespace["queue_tabular_generated_output_run"]( + user_id="user-1", + conversation_id="conversation-1", + user_question="summarize each row as a csv", + source_candidate={"filename": "source_rows.csv", "selected_sheet": ""}, + output_format="csv", + row_batches=[HETEROGENEOUS_ROWS], + gpt_model="gpt-5.6-luna", + settings={}, + passthrough_input_rows=False, + ) + + persisted_run = fake_container.created_items[0] + assert persisted_run["output_schema"] is None, ( + "model-generated runs must still discover their schema from batch 1; " + f"got {persisted_run['output_schema']!r}" + ) + assert persisted_run["public_output_schema"] == [] + + +def test_passthrough_checkpoints_pad_sparse_rows(): + """Sparse passthrough rows must be padded to the pinned schema, never rejected.""" + assert_app_version_at_least(IMPLEMENTED_VERSION) + namespace, _ = _build_passthrough_namespace() + run = _queue_passthrough_run(namespace, [HETEROGENEOUS_ROWS]) + + prepared_rows = namespace["_prepare_tabular_source_rows"]( + HETEROGENEOUS_ROWS, + start_row=0, + token_namespace=run["id"], + ) + generated_results = namespace["_build_passthrough_batch_results"]( + run, + [{"batch_number": 1, "rows": prepared_rows}], + ) + + assert len(generated_results) == 1 + batch_entries = generated_results[0]["batch_entries"] + assert generated_results[0]["output_schema"] == run["output_schema"] + assert len(batch_entries) == len(HETEROGENEOUS_ROWS) + for row_index, entry in enumerate(batch_entries, start=1): + assert list(entry) == run["output_schema"], ( + f"row {row_index} must be checkpointed with the exact pinned schema; got {list(entry)!r}" + ) + assert entry["source_row_number"] == row_index + + assert batch_entries[0]["facts"] == "1 instance online" + assert batch_entries[0]["application"] == "", "missing columns must be padded, not dropped" + assert batch_entries[1]["message_metadata"] == "BatteryVoltage1" + assert batch_entries[1]["facts"] == "" + assert batch_entries[2]["runtime"] == "realtime" + + +def test_passthrough_batch_unions_rows_when_schema_is_not_pinned(): + """Runs queued before the fix have no pinned schema and must still checkpoint cleanly.""" + assert_app_version_at_least(IMPLEMENTED_VERSION) + namespace, _ = _build_passthrough_namespace() + legacy_run = {"output_schema": None} + + prepared_rows = namespace["_prepare_tabular_source_rows"]( + HETEROGENEOUS_ROWS, + start_row=0, + token_namespace="legacy-run", + ) + generated_results = namespace["_build_passthrough_batch_results"]( + legacy_run, + [{"batch_number": 1, "rows": prepared_rows}], + ) + + output_schema = generated_results[0]["output_schema"] + assert output_schema == LINEAGE_SCHEMA + EXPECTED_PUBLIC_SCHEMA + for entry in generated_results[0]["batch_entries"]: + assert list(entry) == output_schema + + +if __name__ == "__main__": + tests = [ + test_queue_pins_union_schema_for_heterogeneous_passthrough_rows, + test_model_backed_runs_still_defer_schema_discovery, + test_passthrough_checkpoints_pad_sparse_rows, + test_passthrough_batch_unions_rows_when_schema_is_not_pinned, + ] + results = [] + for test in tests: + print(f"Running {test.__name__}...") + try: + test() + print(" passed") + results.append(True) + except Exception as exc: + print(f" failed: {exc}") + import traceback + + traceback.print_exc() + results.append(False) + print(f"Results: {sum(results)}/{len(results)} tests passed") + sys.exit(0 if all(results) else 1)