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)