fix(tabular): union passthrough CSV schema and keep assistant text visible

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')

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:

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 @@
     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 @@
         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 @@
     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 @@
         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 @@
                 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 @@
         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 @@
         '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,

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,
     ]