From 9a68f45346b8aa2fd8e9a620909b0d4c6a28ab91 Mon Sep 17 00:00:00 2001 From: Paul Lizer Date: Fri, 14 Aug 2026 15:59:45 -0400 Subject: [PATCH] fix: generate exhaustive row-level markdown outputs --- application/single_app/config.py | 2 +- .../functions_tabular_generated_exports.py | 173 +++++++++++++++++- .../functions_tabular_orchestration.py | 131 ++++++++++++- application/single_app/route_backend_chats.py | 56 +++++- .../static/js/chat/chat-messages.js | 7 + .../TABULAR_DURABLE_ARTIFACT_LIFECYCLE_FIX.md | 2 + .../TABULAR_EXHAUSTIVE_ROW_MARKDOWN_FIX.md | 76 ++++++++ docs/explanation/release_notes.md | 10 + ...st_tabular_line_terminology_routing_fix.py | 82 ++++++--- ...tabular_phase3_public_schema_projection.py | 109 ++++++++++- ...t_tabular_phase5_artifact_set_lifecycle.py | 5 +- ...est_tabular_phase8_ui_telemetry_rollout.py | 5 +- ...ular_queue_run_output_schema_end_to_end.py | 101 +++++++++- .../test_tabular_row_orchestration_scale.py | 10 +- ..._tabular_search_analyze_artifact_matrix.py | 37 +++- ...chat_background_generated_export_status.py | 142 ++++++++++++-- 16 files changed, 871 insertions(+), 77 deletions(-) create mode 100644 docs/explanation/fixes/TABULAR_EXHAUSTIVE_ROW_MARKDOWN_FIX.md diff --git a/application/single_app/config.py b/application/single_app/config.py index 94954c66f..e4527b682 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.200" +VERSION = "0.250.201" IS_DEVELOPMENT = is_development_env_enabled() SESSION_COOKIE_SAMESITE = os.getenv('SESSION_COOKIE_SAMESITE', 'Lax') diff --git a/application/single_app/functions_tabular_generated_exports.py b/application/single_app/functions_tabular_generated_exports.py index 44031a5ae..8209b626e 100644 --- a/application/single_app/functions_tabular_generated_exports.py +++ b/application/single_app/functions_tabular_generated_exports.py @@ -6,6 +6,7 @@ import csv import heapq import hashlib +import html import io import json import logging @@ -282,6 +283,8 @@ TABULAR_ANALYSIS_SUMMARY_MAX_CHARS = 24000 TABULAR_ANALYSIS_MAX_FINDINGS = 12 TABULAR_ANALYSIS_MAX_NOTABLE_ROWS = 25 +TABULAR_ROW_ANALYSIS_MAX_QUESTIONS = 20 +TABULAR_ROW_ANALYSIS_ANSWER_ESTIMATED_CHARS = 180 TABULAR_EXPORT_SUMMARY_MAX_FIELDS = 25 TABULAR_EXPORT_SUMMARY_MAX_VALUES_PER_FIELD = 5 TABULAR_EXPORT_SUMMARY_AGGREGATE_MAX_VALUES = 25 @@ -1872,10 +1875,15 @@ def _sanitize_file_base_name(file_name): def _build_generated_file_name(source_file_name, output_format): timestamp_suffix = datetime.utcnow().strftime('%Y%m%d_%H%M%S') - normalized_extension = normalize_generated_output_format(output_format) + normalized_extension = _normalize_tabular_artifact_format(output_format) return f"{_sanitize_file_base_name(source_file_name)}_generated_{timestamp_suffix}.{normalized_extension}" +def _build_row_analysis_file_name(source_file_name): + timestamp_suffix = datetime.utcnow().strftime('%Y%m%d_%H%M%S') + return f"{_sanitize_file_base_name(source_file_name)}_row_analysis_{timestamp_suffix}.md" + + def _build_analysis_file_name(source_file_name): timestamp_suffix = datetime.utcnow().strftime('%Y%m%d_%H%M%S') return f"{_sanitize_file_base_name(source_file_name)}_analysis_{timestamp_suffix}.md" @@ -1894,6 +1902,17 @@ def _serialize_generated_output_value(value): return neutralize_csv_spreadsheet_formula(value) +def _escape_markdown_text(value): + """Render untrusted values as literal Markdown text, not active markup.""" + normalized_value = str(value or '').replace('\r\n', '\n').replace('\r', '\n').replace('\t', ' ') + escaped_lines = [] + for line in normalized_value.split('\n'): + escaped_line = html.escape(line, quote=False).replace('\\', '\\\\') + escaped_line = re.sub(r'([`*_[\]{}()#+\-.!|>])', r'\\\1', escaped_line) + escaped_lines.append(escaped_line) + return '\n'.join(escaped_lines) + + def _sanitize_generated_xml_tag_name(value, fallback_value='Field'): normalized_name = re.sub(r'[^A-Za-z0-9_.-]+', '_', str(value or '').strip()).strip('._-') if not normalized_name: @@ -2056,6 +2075,58 @@ def _normalize_generated_batch_entries( return ordered_entries, output_schema +def _validate_exhaustive_row_analysis_entries(entries, output_schema): + public_fields = [ + str(field_name or '').strip().lower() + for field_name in list(output_schema or []) + if str(field_name or '').strip() + and not is_analysis_internal_lineage_field(field_name) + ] + exact_row_fields = ( + public_fields == ['row_analysis'] + or any(re.fullmatch(r'answer_\d+', field_name) for field_name in public_fields) + ) + if not exact_row_fields: + return + if public_fields != ['row_analysis']: + expected_fields = [f'answer_{field_index}' for field_index in range(1, len(public_fields) + 1)] + if public_fields != expected_fields: + raise ValueError( + f'Exhaustive row analysis fields must be consecutive answer_1 through answer_N; got {public_fields}' + ) + for row_index, entry in enumerate(entries or [], start=1): + for field_name in public_fields: + if (entry or {}).get(field_name) in (None, '', [], {}): + raise ValueError( + f'Generated exhaustive row analysis left {field_name} empty at row {row_index}' + ) + + +def _validate_exhaustive_row_analysis_contract(row_analysis_mode, questions, public_output_schema): + if str(row_analysis_mode or '').strip().lower() != 'exhaustive': + return + normalized_questions = [ + str(question or '').strip() + for question in list(questions or [])[:TABULAR_ROW_ANALYSIS_MAX_QUESTIONS] + if str(question or '').strip() + ] + normalized_public_schema = [ + str(field_name or '').strip().lower() + for field_name in list(public_output_schema or []) + if str(field_name or '').strip() + and not is_analysis_internal_lineage_field(field_name) + ] + expected_schema = ( + [f'answer_{question_index}' for question_index in range(1, len(normalized_questions) + 1)] + if normalized_questions + else ['row_analysis'] + ) + if normalized_public_schema != expected_schema: + raise ValueError( + 'Exhaustive row analysis questions do not match the persisted public output schema' + ) + + def _generated_entry_has_source_position_conflict(source_row, generated_entry): source_row_number = _safe_int(source_row.get(TABULAR_EXPORT_INPUT_ROW_NUMBER_FIELD), default=0, minimum=0) source_row_identity = str(source_row.get(TABULAR_EXPORT_INPUT_ROW_IDENTITY_FIELD) or '').strip() @@ -3259,6 +3330,15 @@ def _build_combined_chunk_prompt(run, batch_rows, batch_number, batch_count, out if model_output_schema else '' ) + answer_field_line = ( + 'Fields named answer_N map to the Nth question in the user instructions. ' + 'Answer every answer_N field independently and do not combine or omit questions.\n' + if any( + re.fullmatch(r'answer_\d+', str(field_name or '').strip().lower()) + for field_name in model_output_schema + ) + else '' + ) return ( 'Transform and analyze the bounded tabular chunk below for the user.\n\n' f'User structured-output instructions:\n{user_question}\n\n' @@ -3266,6 +3346,7 @@ def _build_combined_chunk_prompt(run, batch_rows, batch_number, batch_count, out 'Return ONLY a valid JSON object with exactly these top-level fields: structured_rows, analysis_summary.\n' f'structured_rows must be an array of exactly {len(batch_rows)} object(s), one per input row, in the same order.\n' f'{output_schema_line}' + f'{answer_field_line}' f'Each structured row must copy {TABULAR_EXPORT_INPUT_ROW_TOKEN_FIELD} exactly from the matching input row. ' f'Do not include {TABULAR_EXPORT_INPUT_ROW_NUMBER_FIELD} or {TABULAR_EXPORT_INPUT_ROW_IDENTITY_FIELD} in structured rows.\n' 'Do not drop, merge, summarize, or cap structured rows. If a requested field cannot be derived, include it with null or an empty string.\n' @@ -3740,6 +3821,15 @@ def _build_batch_prompt( if model_output_schema else '' ) + answer_field_line = ( + 'Fields named answer_N map to the Nth question in the user instructions. ' + 'Answer every answer_N field independently and do not combine or omit questions.\n' + if any( + re.fullmatch(r'answer_\d+', str(field_name or '').strip().lower()) + for field_name in model_output_schema + ) + else '' + ) return ( 'Transform the tabular input rows below into structured output for the user.\n\n' @@ -3747,6 +3837,7 @@ def _build_batch_prompt( 'Return ONLY a valid JSON array.\n' f'Return exactly {len(batch_rows)} JSON object(s), one per input row, in the same order.\n' f'{output_schema_line}' + f'{answer_field_line}' f'Copy {TABULAR_EXPORT_INPUT_ROW_TOKEN_FIELD} exactly from each input row into its matching output object. ' f'The {TABULAR_EXPORT_INPUT_ROW_NUMBER_FIELD} and {TABULAR_EXPORT_INPUT_ROW_IDENTITY_FIELD} fields are internal; ' 'do not include those two fields in generated objects.\n' @@ -5220,6 +5311,7 @@ async def _generate_batch_entries( expected_output_schema or output_schema, transformation_spec=transformation_spec, ) + _validate_exhaustive_row_analysis_entries(normalized_entries, output_schema) semantic_validation_counts = {} semantic_validation_attempts = [] if transformation_spec: @@ -6213,6 +6305,7 @@ async def _generate_combined_chunk_result( expected_output_schema or output_schema, transformation_spec=transformation_spec, ) + _validate_exhaustive_row_analysis_entries(normalized_entries, output_schema) candidate_checkpoint = None if transformation_spec and semantic_mode in {'shadow', 'active'}: candidate_checkpoint = _load_tabular_semantic_candidate_checkpoint( @@ -6842,6 +6935,12 @@ def _normalize_tabular_run_planner_metadata(planner_metadata): 'execution_state': str(planner_metadata.get('execution_state') or '').strip().lower()[:40], 'durable_task_type': _normalize_tabular_run_task_type(planner_metadata.get('durable_task_type')), 'reason_code': str(planner_metadata.get('reason_code') or '').strip().lower()[:80], + 'row_analysis_mode': str(planner_metadata.get('row_analysis_mode') or '').strip().lower()[:40], + 'row_analysis_questions': [ + str(question or '').strip()[:500] + for question in list(planner_metadata.get('row_analysis_questions') or [])[:TABULAR_ROW_ANALYSIS_MAX_QUESTIONS] + if str(question or '').strip() + ], 'execution_group_id': str(planner_metadata.get('execution_group_id') or '').strip()[:128], 'source_coverage_summary': _build_planner_source_coverage_summary( planner_metadata.get('source_coverage'), @@ -8282,6 +8381,12 @@ def _write_ordered_output_stream(run, output_stream): csv_writer.writeheader() elif output_format == 'xml': output_stream.write('\n\n') + elif output_format == 'md': + output_stream.write('# Row-by-Row Tabular Analysis\n\n') + output_stream.write( + f"Source file: {run.get('source_file_name') or 'unknown file'} \n" + f"Rows: {expected_row_count:,}\n\n" + ) else: output_stream.write('[\n') @@ -8326,6 +8431,31 @@ def _write_ordered_output_stream(run, output_stream): }) elif output_format == 'xml': _write_generated_xml_row(output_stream, public_entry) + elif output_format == 'md': + source_row_identity = _normalize_analysis_text( + ordered_entry.get(TABULAR_EXPORT_OUTPUT_ROW_IDENTITY_FIELD), + max_chars=200, + ) + identity_suffix = f": {source_row_identity}" if source_row_identity else '' + output_stream.write( + f"## Row {source_row_number}{_escape_markdown_text(identity_suffix)}\n\n" + ) + row_questions = list((run or {}).get('row_analysis_questions') or []) + for field_index, field_name in enumerate(public_output_schema, start=1): + question_label = ( + _normalize_analysis_text(row_questions[field_index - 1], max_chars=500) + if field_index <= len(row_questions) + else str(field_name or '').replace('_', ' ').strip().title() + ) + field_value = _escape_markdown_text( + _serialize_generated_output_value(public_entry.get(field_name)) + ) + output_stream.write( + f"{field_index}. **{_escape_markdown_text(question_label)}**\n\n" + ) + for value_line in field_value.split('\n'): + output_stream.write(f" {value_line}\n") + output_stream.write('\n') else: if written_row_count: output_stream.write(',\n') @@ -8336,7 +8466,7 @@ def _write_ordered_output_stream(run, output_stream): if output_format == 'xml': output_stream.write('\n') - elif output_format != 'csv': + elif output_format not in {'csv', 'md'}: output_stream.write('\n]\n') if written_row_count != expected_row_count: raise ValueError( @@ -9070,7 +9200,11 @@ def _build_public_artifact_projection(artifact): def _publish_structured_export_artifact(run, descriptor=None): descriptor = descriptor if isinstance(descriptor, dict) else {} member_id = str(descriptor.get('member_id') or _get_structured_artifact_member_id(run)).strip() - output_format = normalize_generated_output_format(descriptor.get('format') or run.get('output_format')) + output_format = _normalize_tabular_artifact_format( + descriptor.get('format') or run.get('output_format') + ) + if output_format not in {'csv', 'json', 'xml', 'md'}: + raise ValueError(f'Unsupported tabular structured artifact format: {output_format}') generated_file_name = run.get('generated_file_name') or _build_generated_file_name( run.get('source_file_name'), output_format, @@ -9148,7 +9282,7 @@ def _publish_structured_export_artifacts(run): artifacts = [] first_summary = '' first_entry_count = None - first_output_format = normalize_generated_output_format(run.get('output_format')) + first_output_format = _normalize_tabular_artifact_format(run.get('output_format')) first_file_name = run.get('generated_file_name') or _build_generated_file_name( run.get('source_file_name'), first_output_format, @@ -10623,6 +10757,11 @@ def process_tabular_generated_output_run(run_id, user_id): run = _migrate_legacy_tabular_export_run(run) if run.get('source_descriptor') and not run.get('source_staging_complete'): run = _stage_tabular_generated_output_source(run, settings) + _validate_exhaustive_row_analysis_contract( + run.get('row_analysis_mode'), + run.get('row_analysis_questions'), + _get_tabular_run_public_output_schema(run), + ) retry_attempts = _settings_int( settings, @@ -10940,6 +11079,8 @@ def queue_tabular_generated_output_run( normalized_analysis_objective = str(user_question or '').strip() if normalized_task_type == TABULAR_RUN_TASK_HIERARCHICAL_ANALYSIS: generated_file_name = _build_analysis_file_name(source_file_name) + elif normalized_output_format == 'md': + generated_file_name = _build_row_analysis_file_name(source_file_name) else: generated_file_name = _build_generated_file_name(source_file_name, normalized_output_format) analysis_generated_file_name = ( @@ -10973,6 +11114,28 @@ def queue_tabular_generated_output_run( task_type=normalized_task_type, user_question=user_question, ) + row_analysis_mode = str(tabular_planner_metadata.get('row_analysis_mode') or '').strip().lower() + row_analysis_questions = list(tabular_planner_metadata.get('row_analysis_questions') or []) + _validate_exhaustive_row_analysis_contract( + row_analysis_mode, + row_analysis_questions, + contract_public_output_schema, + ) + if row_analysis_mode == 'exhaustive': + question_count = max(1, len(row_analysis_questions)) + estimated_output_chars_per_row = max( + 600, + question_count * TABULAR_ROW_ANALYSIS_ANSWER_ESTIMATED_CHARS, + ) + output_char_budget = max( + 1000, + _safe_int(model_batch_budget.get('output_token_budget'), minimum=1) + * TABULAR_EXPORT_APPROXIMATE_CHARS_PER_TOKEN, + ) + model_batch_budget['max_rows'] = min( + _safe_int(model_batch_budget.get('max_rows'), minimum=1), + max(1, output_char_budget // estimated_output_chars_per_row), + ) chunk_gpt_model, chunk_model_context = _resolve_tabular_chunk_model_selection( gpt_model, settings, @@ -11141,6 +11304,8 @@ def queue_tabular_generated_output_run( 'tabular_planner_metadata': tabular_planner_metadata, 'task_type': normalized_task_type, 'analysis_objective': normalized_analysis_objective, + 'row_analysis_mode': row_analysis_mode, + 'row_analysis_questions': row_analysis_questions, 'user_id': normalized_user_id, 'conversation_id': normalized_conversation_id, 'status': TABULAR_EXPORT_STATUS_QUEUED, diff --git a/application/single_app/functions_tabular_orchestration.py b/application/single_app/functions_tabular_orchestration.py index 298a91360..e49da4c52 100644 --- a/application/single_app/functions_tabular_orchestration.py +++ b/application/single_app/functions_tabular_orchestration.py @@ -4,9 +4,12 @@ import hashlib import json import os +import re from typing import Mapping from functions_analysis_deliverables import ( + ANALYSIS_ARTIFACT_ROLE_PRIMARY_ANALYSIS, + ANALYSIS_ARTIFACT_ROLE_REQUESTED_OUTPUT, ANALYSIS_DELIVERABLE_EVENT_FINALIZED, ANALYSIS_DELIVERABLE_EVENT_PLANNED, ANALYSIS_ORDERING_NOT_APPLICABLE, @@ -18,6 +21,7 @@ ANALYSIS_VALIDATION_PROFILE_ARTIFACT_SET, ANALYSIS_VALIDATION_PROFILE_EXACT_ROWS_SCHEMA, ANALYSIS_VALIDATION_PROFILE_EXACT_ROWS_SCHEMA_AND_RULES, + build_analysis_deliverable_artifact, build_analysis_deliverable_contract, emit_analysis_deliverable_contract_event, ) @@ -323,11 +327,76 @@ def question_requests_tabular_hierarchical_analysis(user_question): ) +def question_requests_tabular_exhaustive_row_output(user_question): + """Return True when the user requires one narrative result per source row.""" + normalized_question = str(user_question or "").strip().lower() + if not normalized_question: + return False + + row_output_markers = ( + "for each row", + "for every row", + "for each line", + "for every line", + "line by line", + "row by row", + "each line item", + "each row individually", + "each line individually", + "individually for each row", + "individually for each line", + "one answer per row", + "one answer per line", + "one result per row", + "one result per line", + "one output per row", + "one output per line", + "one markdown section per row", + "one markdown section per line", + ) + return any(marker in normalized_question for marker in row_output_markers) + + +def extract_tabular_row_analysis_questions(user_question, max_questions=20): + """Extract an ordered, bounded question list for exact-row narrative output.""" + question_text = str(user_question or "").strip() + if not question_text: + return [] + + marker_match = re.search( + r"(?is)\b(?:questions?\s+(?:are|is)\s+as\s+follows|answer\s+(?:the\s+)?following\s+questions?)\s*:?", + question_text, + ) + candidate_text = question_text[marker_match.end():] if marker_match else question_text + line_candidates = [] + for raw_line in candidate_text.splitlines(): + normalized_line = re.sub(r"^\s*(?:[-*•]|\d+[.)])\s*", "", raw_line).strip() + if normalized_line: + line_candidates.append(normalized_line) + if len(line_candidates) >= 2: + return line_candidates[:max_questions] + + split_candidates = re.split( + r"(?i)(?=\b(?:what|why|how|when|where|who|which|does|do|is|are|can|could|should|would|concerns?)\b)", + candidate_text, + ) + questions = [] + for candidate in split_candidates: + normalized_candidate = re.sub(r"\s+", " ", candidate).strip(" :-\t\r\n") + if not normalized_candidate: + continue + questions.append(normalized_candidate[:500]) + if len(questions) >= max_questions: + break + return questions + + def get_tabular_generated_output_task_type( generated_output_requested, hierarchical_analysis_requested, settings, action_mode=None, + exhaustive_row_output_requested=False, ): """Map request intent to the existing durable generated-output task type.""" hierarchical_analysis_enabled = settings_flag_enabled( @@ -336,6 +405,10 @@ def get_tabular_generated_output_task_type( False, ) analysis_required = str(action_mode or "").strip().lower() == "analyze" + if exhaustive_row_output_requested and hierarchical_analysis_enabled: + return TABULAR_RUN_TASK_COMBINED if analysis_required else TABULAR_RUN_TASK_STRUCTURED_EXPORT + if exhaustive_row_output_requested: + return None if generated_output_requested and analysis_required: return TABULAR_RUN_TASK_COMBINED if generated_output_requested and hierarchical_analysis_requested and hierarchical_analysis_enabled: @@ -575,13 +648,29 @@ def plan_tabular_request( structured_output_formats = get_tabular_generated_output_formats(user_question) generated_output_requested = question_requests_tabular_generated_output(user_question) hierarchical_analysis_requested = question_requests_tabular_hierarchical_analysis(user_question) + exhaustive_row_output_requested = question_requests_tabular_exhaustive_row_output(user_question) + exhaustive_narrative_row_output_requested = bool( + exhaustive_row_output_requested and not structured_output_formats + ) + row_analysis_questions = ( + extract_tabular_row_analysis_questions(user_question) + if exhaustive_narrative_row_output_requested + else [] + ) durable_task_type = get_tabular_generated_output_task_type( generated_output_requested, hierarchical_analysis_requested, settings, action_mode=normalized_action_mode, + exhaustive_row_output_requested=exhaustive_narrative_row_output_requested, + ) + output_format = ( + structured_output_formats[0] + if structured_output_formats + else "md" + if exhaustive_narrative_row_output_requested + else None ) - output_format = structured_output_formats[0] if structured_output_formats else None execution_contract = durable_task_type or TABULAR_EXECUTION_CONTRACT_FOREGROUND_AGGREGATE source_coverage = _build_source_coverage(normalized_contexts) execution_group_id = _build_execution_group_id( @@ -642,16 +731,42 @@ def plan_tabular_request( if transformation_spec else ANALYSIS_TRANSFORMATION_MODE_SEMANTIC ) + exact_row_output_requested = generated_output_requested or exhaustive_narrative_row_output_requested validation_profile = ( ANALYSIS_VALIDATION_PROFILE_EXACT_ROWS_SCHEMA_AND_RULES - if generated_output_requested and transformation_spec + if exact_row_output_requested and transformation_spec else ANALYSIS_VALIDATION_PROFILE_EXACT_ROWS_SCHEMA - if generated_output_requested + if exact_row_output_requested else ANALYSIS_VALIDATION_PROFILE_ARTIFACT_SET ) + row_analysis_output_schema = [ + f"answer_{question_index}" + for question_index in range(1, len(row_analysis_questions) + 1) + ] or (["row_analysis"] if exhaustive_narrative_row_output_requested else []) + requested_artifacts = None + if exhaustive_narrative_row_output_requested: + requested_artifacts = [] + request_order = 0 + if normalized_action_mode == "analyze": + requested_artifacts.append(build_analysis_deliverable_artifact( + "analysis-summary", + ANALYSIS_ARTIFACT_ROLE_PRIMARY_ANALYSIS, + "md", + required=True, + request_order=request_order, + )) + request_order += 1 + requested_artifacts.append(build_analysis_deliverable_artifact( + "row-analysis-md", + ANALYSIS_ARTIFACT_ROLE_REQUESTED_OUTPUT, + "md", + required=True, + request_order=request_order, + )) deliverable_contract = build_analysis_deliverable_contract( action_mode=action_mode, requested_output_formats=requested_output_formats, + requested_artifacts=requested_artifacts, analysis_required=( normalized_action_mode == "analyze" or durable_task_type in { @@ -660,18 +775,19 @@ def plan_tabular_request( } ), public_output_schema=( - output_hints.get("public_output_schema") + row_analysis_output_schema + or output_hints.get("public_output_schema") or output_hints.get("output_schema") or [] ), row_cardinality=( ANALYSIS_ROW_CARDINALITY_ONE_PER_SOURCE_ROW - if generated_output_requested + if exact_row_output_requested else ANALYSIS_ROW_CARDINALITY_NOT_APPLICABLE ), ordering=( ANALYSIS_ORDERING_SOURCE_ORDER - if generated_output_requested + if exact_row_output_requested else ANALYSIS_ORDERING_NOT_APPLICABLE ), transformation_mode=transformation_mode, @@ -689,6 +805,9 @@ def plan_tabular_request( "durable_task_type": durable_task_type, "generated_output_requested": generated_output_requested, "hierarchical_analysis_requested": hierarchical_analysis_requested, + "exhaustive_row_output_requested": exhaustive_narrative_row_output_requested, + "row_analysis_mode": "exhaustive" if exhaustive_narrative_row_output_requested else "summary", + "row_analysis_questions": row_analysis_questions, "requested_output_formats": requested_output_formats, "output_format": output_format, "action_mode": normalized_action_mode, diff --git a/application/single_app/route_backend_chats.py b/application/single_app/route_backend_chats.py index dd1df3ca5..f89d50ec1 100644 --- a/application/single_app/route_backend_chats.py +++ b/application/single_app/route_backend_chats.py @@ -63,6 +63,7 @@ build_tabular_legacy_post_tool_fallback_decision as _shared_build_tabular_legacy_post_tool_fallback_decision, get_tabular_generated_output_format as _shared_get_tabular_generated_output_format, get_tabular_generated_output_task_type as _shared_get_tabular_generated_output_task_type, + question_requests_tabular_exhaustive_row_output as _shared_question_requests_tabular_exhaustive_row_output, question_requests_tabular_generated_output as _shared_question_requests_tabular_generated_output, question_requests_tabular_hierarchical_analysis as _shared_question_requests_tabular_hierarchical_analysis, settings_flag_enabled as _shared_settings_flag_enabled, @@ -5016,16 +5017,28 @@ def question_requests_tabular_hierarchical_analysis(user_question): return _shared_question_requests_tabular_hierarchical_analysis(user_question) +def question_requests_tabular_exhaustive_row_output(user_question): + """Return True when the prompt requires one narrative result per source row.""" + return _shared_question_requests_tabular_exhaustive_row_output(user_question) + + def _settings_flag_enabled(settings, key, default=False): return _shared_settings_flag_enabled(settings, key, default=default) -def _get_tabular_generated_output_task_type(generated_output_requested, hierarchical_analysis_requested, settings, action_mode=None): +def _get_tabular_generated_output_task_type( + generated_output_requested, + hierarchical_analysis_requested, + settings, + action_mode=None, + exhaustive_row_output_requested=False, +): return _shared_get_tabular_generated_output_task_type( generated_output_requested, hierarchical_analysis_requested, settings, action_mode=action_mode, + exhaustive_row_output_requested=exhaustive_row_output_requested, ) @@ -5760,23 +5773,38 @@ 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, action_mode=None): +def _build_direct_tabular_generated_output_source( + user_question, + file_contexts, + user_id, + conversation_id, + settings, + action_mode=None, + planner_metadata=None, +): """Build a replayable full-tabular 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) + exhaustive_row_output_requested = question_requests_tabular_exhaustive_row_output(user_question) durable_task_type = _get_tabular_generated_output_task_type( generated_output_requested, hierarchical_analysis_requested, settings, action_mode=action_mode, + exhaustive_row_output_requested=exhaustive_row_output_requested, ) analysis_only_requested = durable_task_type == TABULAR_RUN_TASK_HIERARCHICAL_ANALYSIS combined_requested = durable_task_type == TABULAR_RUN_TASK_COMBINED if generated_output_requested and str(action_mode or '').strip().lower() == 'analyze' and not durable_task_type: return None - if not generated_output_requested and not analysis_only_requested: + if not generated_output_requested and not analysis_only_requested and not exhaustive_row_output_requested: return None - if hierarchical_analysis_requested and not generated_output_requested and not analysis_only_requested: + if ( + hierarchical_analysis_requested + and not generated_output_requested + and not analysis_only_requested + and not exhaustive_row_output_requested + ): return None normalized_contexts = dedupe_tabular_file_contexts(file_contexts) @@ -5920,7 +5948,11 @@ def _build_direct_tabular_generated_output_source(user_question, file_contexts, ) ), }) - output_format = get_tabular_generated_output_format(user_question) or 'md' + output_format = ( + str((planner_metadata or {}).get('output_format') or '').strip().lower() + or 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, @@ -5958,6 +5990,7 @@ def _build_direct_tabular_generated_output_source(user_question, file_contexts, '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, + 'exhaustive_row_output_requested': exhaustive_row_output_requested, } @@ -6014,6 +6047,7 @@ def emit_direct_parity_event(event_name, planner_result=None, metrics=None, dime conversation_id, settings, action_mode=planner_action_mode, + planner_metadata=planner_metadata, ) if not direct_source: emit_direct_parity_event( @@ -6044,6 +6078,10 @@ def emit_direct_parity_event(event_name, planner_result=None, metrics=None, dime planner_metadata=planner_metadata, ) background_metadata = build_background_tabular_generated_output_metadata(background_run) + actual_batch_count = ( + _safe_int(background_metadata.get('batch_count')) + or direct_source['batch_count_estimate'] + ) accepted_parity_result = parity_result if callable(parity_result_builder): accepted_parity_result = parity_result_builder( @@ -6061,7 +6099,7 @@ def emit_direct_parity_event(event_name, planner_result=None, metrics=None, dime metrics={ 'source_count': len(file_contexts or []), 'row_count': direct_source.get('row_count'), - 'batch_count_estimate': direct_source.get('batch_count_estimate'), + 'batch_count_estimate': actual_batch_count, }, ) emit_direct_parity_event( @@ -6082,7 +6120,7 @@ def emit_direct_parity_event(event_name, planner_result=None, metrics=None, dime '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" + f"rows={direct_source['row_count']}; batches={actual_batch_count}; checkpointed=true" ), 'activity': build_tabular_post_processing_activity_payload( 'tabular.generated_output', @@ -6092,7 +6130,7 @@ def emit_direct_parity_event(event_name, planner_result=None, metrics=None, dime output_format=direct_source['output_format'], file_name=direct_source['source_candidate'].get('filename'), batch_index=0, - batch_count=direct_source['batch_count_estimate'], + batch_count=actual_batch_count, ), } maybe_callback_result = thought_callback(thought_payload) @@ -6105,7 +6143,7 @@ def emit_direct_parity_event(event_name, planner_result=None, metrics=None, dime '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'], + 'batch_count_estimate': actual_batch_count, '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'), diff --git a/application/single_app/static/js/chat/chat-messages.js b/application/single_app/static/js/chat/chat-messages.js index 224c9104b..c35488d27 100644 --- a/application/single_app/static/js/chat/chat-messages.js +++ b/application/single_app/static/js/chat/chat-messages.js @@ -4311,6 +4311,13 @@ function renderReplyQuoteHtml(fullMessageObject = null) { } function getGeneratedAnalysisArtifactTitle(outputMetadata, outputFormat) { + const artifactId = String(outputMetadata?.artifact_id || outputMetadata?.member_id || '').trim().toLowerCase(); + if (artifactId === 'analysis-summary') { + return 'Analyze Markdown summary'; + } + if (artifactId === 'row-analysis-md') { + return 'Row-by-row Markdown output'; + } const capability = String(outputMetadata?.capability || '').trim().toLowerCase(); if (capability === 'analyze') { return `Analyze ${outputFormat.toUpperCase()} artifact`; diff --git a/docs/explanation/fixes/TABULAR_DURABLE_ARTIFACT_LIFECYCLE_FIX.md b/docs/explanation/fixes/TABULAR_DURABLE_ARTIFACT_LIFECYCLE_FIX.md index a8ba162bd..d21252893 100644 --- a/docs/explanation/fixes/TABULAR_DURABLE_ARTIFACT_LIFECYCLE_FIX.md +++ b/docs/explanation/fixes/TABULAR_DURABLE_ARTIFACT_LIFECYCLE_FIX.md @@ -20,6 +20,8 @@ The intended output contract is: - Search exhaustive analysis without an explicit format: one primary Markdown analysis artifact. - Analyze exhaustive analysis without an explicit format: one primary Markdown analysis artifact. +> **Superseded for explicit row-by-row prompts in 0.250.201:** Search now publishes one exhaustive row Markdown artifact, while Analyze publishes a summary Markdown artifact plus an exhaustive row Markdown sibling. Aggregate whole-dataset analysis remains summary-only. + ## Root Cause Analysis The production failures had seven related causes: diff --git a/docs/explanation/fixes/TABULAR_EXHAUSTIVE_ROW_MARKDOWN_FIX.md b/docs/explanation/fixes/TABULAR_EXHAUSTIVE_ROW_MARKDOWN_FIX.md new file mode 100644 index 000000000..5ac69ff11 --- /dev/null +++ b/docs/explanation/fixes/TABULAR_EXHAUSTIVE_ROW_MARKDOWN_FIX.md @@ -0,0 +1,76 @@ +# TABULAR EXHAUSTIVE ROW MARKDOWN FIX + +Fixed in version: **0.250.201** + +## Issue Description + +Durable Search and Analyze runs correctly read all 200 rows and published Markdown, but the Markdown contained only 12 row findings. The artifact still reported `Rows analyzed: 200`, which described input coverage rather than output cardinality. + +For the customer prompt requesting eight individual answers for every line, the required outputs are: + +- **Search:** one exhaustive Markdown file containing all 200 rows and all eight answers per row. +- **Analyze:** one concise Markdown analysis summary plus one exhaustive Markdown file containing all 200 rows and all eight answers per row. + +Aggregate prompts such as analyzing all rows for themes or risks continue to produce the bounded summary artifact. + +## Root Cause Analysis + +The no-format row request was routed to the hierarchical summary lane. That lane intentionally: + +- asks each chunk for `summary`, `findings`, `counts`, and `notable_rows`; +- caps findings at 12 and notable rows at 25; +- recursively reduces chunk summaries; +- renders only the bounded reduced fields into Markdown. + +It guaranteed that all source rows were read, but never guaranteed one output entry per row. The 200-row run happened to use one chunk, and the model returned 12 findings, exactly matching the server cap. + +## Technical Details + +### Files Modified + +- `application/single_app/functions_tabular_orchestration.py` +- `application/single_app/route_backend_chats.py` +- `application/single_app/functions_tabular_generated_exports.py` +- `application/single_app/static/js/chat/chat-messages.js` +- `application/single_app/config.py` +- Related functional and Playwright tests + +### Code Changes Summary + +- Added explicit per-row narrative intent detection, separate from aggregate whole-dataset analysis. +- Extracted the user's ordered, bounded question list and planned `answer_1` through `answer_N` fields. +- Routed no-format per-row Search to `structured_export(md)` and Analyze to `combined(md)`. +- Planned distinct artifact IDs so Analyze can require two same-format siblings: `analysis-summary` and `row-analysis-md`. +- Reused the durable exact-row checkpoint engine, including source tokens, exact row counts, stable schemas, source order, retries, and idempotent publication. +- Sized batches using estimated narrative output per row so a 200-row/eight-question request is split into multiple bounded model calls. +- Required every expected answer field to be non-empty and consecutive before checkpoint acceptance. +- Added an ordered Markdown stream finalizer that writes every source row and original question label, then verifies the final row count. +- Escaped source identities, user question labels, and model-generated answers as literal Markdown text. +- Added distinct UI titles for the summary and row-by-row Markdown artifacts. +- Preserved the existing hierarchical summary lane for aggregate prompts and the existing CSV/JSON/XML behavior for explicit structured formats. + +## Validation + +- Exact customer prompt extracts all eight questions and plans: + - Search: `row-analysis-md`. + - Analyze: `analysis-summary` plus `row-analysis-md`. +- A deterministic 200-row test streams four checkpoints into Markdown and asserts: + - 200 row headings; + - each of eight questions appears 200 times; + - the final row and final eighth answer are present; + - 1,600 answers are represented; + - missing answers and malformed answer sequences fail validation. +- Adversarial row identities, questions, and answers verify active Markdown links and raw HTML are escaped. +- Queue-level testing verifies output-aware batching creates multiple batches and persists exact schemas. +- Full 70-test tabular scale coverage passes, including 100,000-row planning and idempotent publication. +- Full background-card UI suite passes, including Search's one Markdown card and Analyze's two distinct Markdown cards. + +## Impact Analysis + +- "All rows analyzed" now means all rows are also present in the exhaustive deliverable when the user asks for individual per-row output. +- Summary and exhaustive output remain separate concerns, preventing concise analysis from truncating row deliverables. +- Large requests remain bounded and checkpointed instead of attempting one unbounded model response or in-memory artifact. + +## Related Version Updates + +- `application/single_app/config.py` was updated to version **0.250.201**. diff --git a/docs/explanation/release_notes.md b/docs/explanation/release_notes.md index 4d48cfe0d..bc5115933 100644 --- a/docs/explanation/release_notes.md +++ b/docs/explanation/release_notes.md @@ -2,6 +2,16 @@ For feature-focused and fix-focused drill-downs by version, see [Features by Version](/explanation/features/) and [Fixes by Version](/explanation/fixes/). +### **(v0.250.201)** + +#### Bug Fixes + +* **Exhaustive Row-by-Row Markdown Output** + * Fixed line-by-line Markdown analysis reading every source row but publishing only 12 summarized findings because the previous hierarchical lane intentionally bounded findings and notable rows. + * Search now produces one exhaustive Markdown artifact containing every source row and every requested answer; Analyze produces a concise Markdown summary plus a separate exhaustive row-by-row Markdown artifact. + * Exact-row Markdown uses ordered checkpoints, output-aware batching, consecutive answer-field validation, non-empty answer enforcement, final row-count/source-order checks, and literal Markdown escaping for untrusted content. + * (Ref: `functions_tabular_orchestration.py`, `functions_tabular_generated_exports.py`, `route_backend_chats.py`, `TABULAR_EXHAUSTIVE_ROW_MARKDOWN_FIX.md`) + ### **(v0.250.200)** #### Bug Fixes diff --git a/functional_tests/test_tabular_line_terminology_routing_fix.py b/functional_tests/test_tabular_line_terminology_routing_fix.py index 8d598083e..4743cc892 100644 --- a/functional_tests/test_tabular_line_terminology_routing_fix.py +++ b/functional_tests/test_tabular_line_terminology_routing_fix.py @@ -2,8 +2,8 @@ # test_tabular_line_terminology_routing_fix.py """ Functional test for tabular exhaustive per-line terminology routing. -Version: 0.250.199 -Implemented in: 0.250.197; updated in 0.250.199 +Version: 0.250.201 +Implemented in: 0.250.197; updated in 0.250.199 and 0.250.201 A customer reported that a prompt phrased as "For each line in this document, I need eight questions answered... Go line by line and make sure all eight @@ -41,7 +41,7 @@ if str(APP_ROOT) not in sys.path: sys.path.insert(0, str(APP_ROOT)) -IMPLEMENTED_VERSION = "0.250.197" +IMPLEMENTED_VERSION = "0.250.201" CUSTOMER_PROMPT = ( "For each line in this document, I need eight questions answered. I want " @@ -83,8 +83,10 @@ def get_requested_artifact_formats(prompt): _install_lightweight_planner_dependency_stubs() from functions_tabular_orchestration import ( # noqa: E402 + extract_tabular_row_analysis_questions, get_tabular_generated_output_task_type, plan_tabular_request, + question_requests_tabular_exhaustive_row_output, question_requests_tabular_generated_output, question_requests_tabular_hierarchical_analysis, ) @@ -106,46 +108,80 @@ def test_line_phrasing_is_recognized_as_hierarchical_analysis_intent(): assert question_requests_tabular_hierarchical_analysis(row_prompt) is True, row_prompt -def test_customer_prompt_routes_to_durable_hierarchical_analysis_for_analyze_and_search(): - """The exact reported prompt must resolve to the hierarchical_analysis task - type for both Analyze and Search action modes once the feature default is - active, and must fall back to no durable routing when the flag is off - (reproducing the reported bug).""" - print("Testing customer prompt routes to durable hierarchical analysis...") +def test_customer_prompt_routes_to_exact_row_markdown_for_analyze_and_search(): + """The exact prompt must preserve every row while keeping Analyze's summary sibling.""" + print("Testing customer prompt routes to exact-row Markdown...") assert_app_version_at_least(IMPLEMENTED_VERSION) generated_output_requested = question_requests_tabular_generated_output(CUSTOMER_PROMPT) hierarchical_analysis_requested = question_requests_tabular_hierarchical_analysis(CUSTOMER_PROMPT) + exhaustive_row_output_requested = question_requests_tabular_exhaustive_row_output(CUSTOMER_PROMPT) assert hierarchical_analysis_requested is True + assert exhaustive_row_output_requested is True + assert len(extract_tabular_row_analysis_questions(CUSTOMER_PROMPT)) == 8 active_settings = {"enable_tabular_hierarchical_analysis": True} assert get_tabular_generated_output_task_type( - generated_output_requested, hierarchical_analysis_requested, active_settings, action_mode="analyze" - ) == "hierarchical_analysis" + generated_output_requested, + hierarchical_analysis_requested, + active_settings, + action_mode="analyze", + exhaustive_row_output_requested=exhaustive_row_output_requested, + ) == "combined" assert get_tabular_generated_output_task_type( - generated_output_requested, hierarchical_analysis_requested, active_settings, action_mode="search" - ) == "hierarchical_analysis" - for action_mode in ("analyze", "search"): + generated_output_requested, + hierarchical_analysis_requested, + active_settings, + action_mode="search", + exhaustive_row_output_requested=exhaustive_row_output_requested, + ) == "structured_export" + expected_plans = { + "search": { + "task_type": "structured_export", + "artifact_ids": ["row-analysis-md"], + "analysis_required": False, + }, + "analyze": { + "task_type": "combined", + "artifact_ids": ["analysis-summary", "row-analysis-md"], + "analysis_required": True, + }, + } + for action_mode, expected in expected_plans.items(): plan = plan_tabular_request( CUSTOMER_PROMPT, [{"file_name": "simple_financial_review_test_200.csv", "document_id": "doc-1"}], action_mode=action_mode, settings=active_settings, ) - assert plan["durable_task_type"] == "hierarchical_analysis" - assert plan["deliverable_contract"]["analysis_required"] is True - assert [ - artifact["format"] - for artifact in plan["deliverable_contract"]["requested_artifacts"] - ] == ["md"] + contract = plan["deliverable_contract"] + assert plan["durable_task_type"] == expected["task_type"] + assert plan["output_format"] == "md" + assert plan["row_analysis_mode"] == "exhaustive" + assert len(plan["row_analysis_questions"]) == 8 + assert contract["analysis_required"] is expected["analysis_required"] + assert [artifact["artifact_id"] for artifact in contract["requested_artifacts"]] == expected["artifact_ids"] + assert [artifact["format"] for artifact in contract["requested_artifacts"]] == ["md"] * len(expected["artifact_ids"]) + assert contract["public_output_schema"] == [f"answer_{index}" for index in range(1, 9)] + assert contract["row_cardinality"] == "one_per_source_row" + assert contract["ordering"] == "source_order" + assert contract["validation_profile"] == "exact_rows_schema" # Reproduce the reported bug: with the flag off, no durable task type is selected. disabled_settings = {"enable_tabular_hierarchical_analysis": False} assert get_tabular_generated_output_task_type( - generated_output_requested, hierarchical_analysis_requested, disabled_settings, action_mode="analyze" + generated_output_requested, + hierarchical_analysis_requested, + disabled_settings, + action_mode="analyze", + exhaustive_row_output_requested=exhaustive_row_output_requested, ) is None assert get_tabular_generated_output_task_type( - generated_output_requested, hierarchical_analysis_requested, disabled_settings, action_mode="search" + generated_output_requested, + hierarchical_analysis_requested, + disabled_settings, + action_mode="search", + exhaustive_row_output_requested=exhaustive_row_output_requested, ) is None @@ -219,7 +255,7 @@ def test_enable_tabular_hierarchical_analysis_defaults_active(): if __name__ == "__main__": tests = [ test_line_phrasing_is_recognized_as_hierarchical_analysis_intent, - test_customer_prompt_routes_to_durable_hierarchical_analysis_for_analyze_and_search, + test_customer_prompt_routes_to_exact_row_markdown_for_analyze_and_search, test_enable_tabular_hierarchical_analysis_defaults_active, ] results = [] diff --git a/functional_tests/test_tabular_phase3_public_schema_projection.py b/functional_tests/test_tabular_phase3_public_schema_projection.py index c5f619573..df883856c 100644 --- a/functional_tests/test_tabular_phase3_public_schema_projection.py +++ b/functional_tests/test_tabular_phase3_public_schema_projection.py @@ -2,8 +2,8 @@ # test_tabular_phase3_public_schema_projection.py """ Functional test for Phase 3 public schema projection and passthrough safety. -Version: 0.250.182 -Implemented in: 0.250.173; request-order and unchanged-copy guard compatibility updated in 0.250.182 +Version: 0.250.201 +Implemented in: 0.250.173; request-order and unchanged-copy guard compatibility updated in 0.250.182; exhaustive Markdown updated in 0.250.201 This test ensures generated tabular artifacts expose only the persisted public schema while retaining internal checkpoint lineage, and that raw row passthrough @@ -11,8 +11,10 @@ """ import ast +import html import io import json +import re import sys import traceback from pathlib import Path @@ -63,6 +65,9 @@ def load_tabular_export_namespace(checkpoint_rows): module_tree = ast.parse(source, filename=str(TABULAR_EXPORTS)) function_names = { "_safe_int", + "_normalize_analysis_text", + "_escape_markdown_text", + "_validate_exhaustive_row_analysis_entries", "_serialize_generated_output_value", "_sanitize_generated_xml_tag_name", "_write_generated_xml_row", @@ -89,6 +94,7 @@ def load_tabular_export_namespace(checkpoint_rows): "build_safe_csv_headers": build_safe_csv_headers, "csv": __import__("csv"), "escape_xml_text": escape_xml_text, + "html": html, "io": io, "is_analysis_internal_lineage_field": lambda field_name: str(field_name or "").strip() in { "source_row_number", @@ -198,6 +204,104 @@ def test_public_projection_drives_csv_json_xml_and_preview(): assert_false("source_row_number" in preview_rows[0], "preview lineage leakage") +def test_exact_row_markdown_serializes_all_200_rows_and_eight_answers(): + print("Testing exhaustive row-by-row Markdown serialization...") + assert_app_version_at_least("0.250.201") + questions = [ + "What [is this](javascript:alert(1)) and what is it trying to accomplish?", + "Why are we doing it?", + "What value does it produce?", + "What resources are identified or implied?", + "What is the timeline or schedule?", + "What happens if we stop?", + "Does this appear reasonable? Concerns / duplication / measurable outcomes", + "What information is missing to assess this activity?", + ] + answer_fields = [f"answer_{index}" for index in range(1, 9)] + checkpoint_rows = {} + for batch_number in range(1, 5): + batch_start = ((batch_number - 1) * 50) + 1 + checkpoint_rows[f"batch-{batch_number}"] = [ + { + "source_row_number": row_number, + "source_row_identity": f"FRI-{row_number:03d}", + **{ + answer_field: f"FRI-{row_number:03d} answer {answer_index}" + for answer_index, answer_field in enumerate(answer_fields, start=1) + }, + } + for row_number in range(batch_start, batch_start + 50) + ] + checkpoint_rows["batch-1"][0]["source_row_identity"] = "FRI-001 " + checkpoint_rows["batch-1"][0]["answer_1"] = "[Open](javascript:alert(1)) " + + namespace = load_tabular_export_namespace(checkpoint_rows) + run = { + "user_id": "user-1", + "conversation_id": "conversation-1", + "id": "run-row-markdown", + "batch_count": 4, + "row_count": 200, + "output_format": "md", + "source_file_name": "financial_review.csv", + "output_schema": ["source_row_number", "source_row_identity", *answer_fields], + "public_output_schema": answer_fields, + "lineage_schema": ["source_row_number", "source_row_identity"], + "internal_checkpoint_schema": ["source_row_number", "source_row_identity", *answer_fields], + "row_analysis_questions": questions, + } + markdown_stream = io.StringIO() + + all_checkpoint_rows = [ + row + for batch_rows in checkpoint_rows.values() + for row in batch_rows + ] + namespace["_validate_exhaustive_row_analysis_entries"]( + all_checkpoint_rows, + run["output_schema"], + ) + invalid_rows = [dict(row) for row in all_checkpoint_rows] + invalid_rows[-1]["answer_8"] = "" + try: + namespace["_validate_exhaustive_row_analysis_entries"]( + invalid_rows, + run["output_schema"], + ) + except ValueError as exc: + assert_true("answer_8 empty at row 200" in str(exc), "empty final answer rejection") + else: + raise AssertionError("An empty row answer was accepted") + try: + namespace["_validate_exhaustive_row_analysis_entries"]( + all_checkpoint_rows, + ["source_row_number", "source_row_identity", "answer_1", "answer_3"], + ) + except ValueError as exc: + assert_true("consecutive answer_1 through answer_N" in str(exc), "answer sequence rejection") + else: + raise AssertionError("A skipped answer field sequence was accepted") + + written_row_count = namespace["_write_ordered_output_stream"](run, markdown_stream) + markdown = markdown_stream.getvalue() + + assert_equal(written_row_count, 200, "Markdown written row count") + assert_equal(len(re.findall(r"(?m)^## Row \d+", markdown)), 200, "Markdown row headings") + for answer_index, question in enumerate(questions, start=1): + escaped_question = namespace["_escape_markdown_text"](question) + assert_equal( + markdown.count(f"{answer_index}. **{escaped_question}**"), + 200, + f"question {answer_index} occurrence count", + ) + assert_true(r"## Row 1: FRI\-001 <script>alert\(1\)</script>" in markdown, "escaped first row present") + assert_true("## Row 200: FRI\\-200" in markdown, "last row present") + assert_true("FRI\\-200 answer 8" in markdown, "last row final answer present") + assert_false("](javascript:" in markdown, "active Markdown links excluded") + assert_false("