From 02eaed0b0b7aa7b33318ee396db7930012a1bc15 Mon Sep 17 00:00:00 2001 From: Paul Lizer Date: Wed, 19 Aug 2026 15:10:03 -0400 Subject: [PATCH 1/3] fix(csv): build CSV artifacts from authorized action rows A telemetry agent retrieved 900 BatteryVoltage1 samples and was asked for a CSV. The published artifact held 3 rows. A follow-up conversation published 2 rows of server discovery metadata. Three defects combined. 1. build_generated_file_export() let any assistant-rendered table outrank the authorized action rows, so the illustrative excerpt the model pasted replaced the 900-row dataset. CSV now prefers the authorized rows whenever the assistant table is an excerpt of them, matched on shared columns and cell values. A table the model composed itself stays authoritative. 2. CSV never received the publication contract JSON and XML already state, so the model reported that it could not attach files and pasted a sample instead. build_generated_file_output_guidance() now tells the model the server serializes and attaches the artifact, and forbids pasting a row sample in its place. The one-clarification rule is unchanged. 3. A turn whose reply was the schema clarification still published a CSV, assembled from whatever incidental discovery rows the turn produced. That turn now publishes nothing and waits for the user's answer. Version: 0.260.007 --- application/single_app/config.py | 2 +- .../functions_generated_file_exports.py | 88 +++++- ...nerated_csv_uses_authorized_action_rows.py | 283 ++++++++++++++++++ 3 files changed, 368 insertions(+), 5 deletions(-) create mode 100644 functional_tests/test_generated_csv_uses_authorized_action_rows.py diff --git a/application/single_app/config.py b/application/single_app/config.py index 179e25894..39d9a8a90 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.006" +VERSION = "0.260.007" 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 72a38c29b..00104480c 100644 --- a/application/single_app/functions_generated_file_exports.py +++ b/application/single_app/functions_generated_file_exports.py @@ -195,6 +195,20 @@ re.compile(r'\b(?:csv|json|xml|docx|word|pdf|spreadsheet)\b[\w\s,.:;\-/]{0,100}\b(?:export|download|file|format|copy)\b'), ) +# The one schema clarification build_csv_output_clarification_guidance() asks the model to send. +CSV_CLARIFICATION_REPLY_PATTERNS = ( + re.compile(r'\b(?:should|shall|does|do)\s+(?:each|every|the)\s+(?:csv\s+)?(?:row|record|line|entry)\b'), + re.compile(r'\b(?:which|what)\s+(?:columns?|fields?)\b'), + re.compile(r'\b(?:columns?|fields?)\s+(?:should|would|do)\s+(?:it|they|the\s+\w+)\s+include\b'), +) +CSV_CLARIFICATION_REPLY_MAX_CHARS = 500 +CSV_PUBLICATION_GUIDANCE = ( + 'The server serializes the authorized action results into the requested CSV artifact and ' + 'attaches the file after generation. Do not claim that you cannot create or attach files, ' + 'do not tell the user to copy or save rows manually, and do not paste a sample of the rows ' + 'in place of the artifact.' +) + def _normalize_question_for_artifact_detection(user_question: str) -> str: return re.sub(r'\s+', ' ', str(user_question or '').strip().casefold()) @@ -436,7 +450,12 @@ def build_generated_file_output_guidance( or get_requested_structured_artifact_format(user_question) ) if output_format == GENERATED_FILE_FORMAT_CSV: - return build_csv_output_clarification_guidance(user_question) + clarification_guidance = build_csv_output_clarification_guidance(user_question) + return ' '.join( + guidance_part + for guidance_part in (clarification_guidance, CSV_PUBLICATION_GUIDANCE) + if guidance_part + ) if output_format == 'json': return ( 'The user requested a downloadable JSON artifact. The server will validate and attach the file after ' @@ -494,17 +513,23 @@ def build_generated_file_export( ) if function_rows else {'allowed': False, 'reason_code': 'source_result_incomplete'} if output_format == GENERATED_FILE_FORMAT_CSV: - rows = assistant_rows or (function_rows if function_passthrough.get('allowed') else []) + if not assistant_rows and _assistant_reply_requests_clarification(assistant_text): + return None + use_function_rows = bool(function_rows) and bool(function_passthrough.get('allowed')) and ( + not assistant_rows + or _assistant_rows_sample_function_rows(assistant_rows, function_rows) + ) + rows = function_rows if use_function_rows else assistant_rows if not rows: return None - row_source = 'assistant response' if assistant_rows else 'structured function result' + row_source = 'structured function result' if use_function_rows else 'assistant response' return _build_generated_file_payload( output_format=output_format, file_content=build_assistant_table_csv(rows), rows=rows, row_source=row_source, assistant_content=assistant_text, - passthrough_reason_code=(None if assistant_rows else function_passthrough.get('reason_code')), + passthrough_reason_code=(function_passthrough.get('reason_code') if use_function_rows else None), ) if not assistant_text and not assistant_rows and not function_rows: @@ -563,6 +588,61 @@ def extract_authorized_function_result_rows(function_results: Optional[List[Dict return combined_rows +def _assistant_reply_requests_clarification(assistant_content: str) -> bool: + """Return whether the reply is the schema clarification rather than the answer.""" + normalized_content = _normalize_question_for_artifact_detection(assistant_content) + if '?' not in normalized_content or len(normalized_content) > CSV_CLARIFICATION_REPLY_MAX_CHARS: + return False + return any(pattern.search(normalized_content) for pattern in CSV_CLARIFICATION_REPLY_PATTERNS) + + +def _normalized_row_text_map(row: Dict[str, Any]) -> Dict[str, str]: + return { + _normalize_function_result_key(column_name): ( + '' if value is None else str(value).strip() + ) + for column_name, value in row.items() + } + + +def _assistant_rows_sample_function_rows( + assistant_rows: Sequence[Dict[str, Any]], + function_rows: Sequence[Dict[str, Any]], +) -> bool: + """Return whether the assistant table is an excerpt of the same authorized rows.""" + if not assistant_rows or len(assistant_rows) >= len(function_rows): + return False + + assistant_row_maps = [] + assistant_columns = [] + seen_columns = set() + for assistant_row in assistant_rows: + if not isinstance(assistant_row, dict): + return False + row_map = _normalized_row_text_map(assistant_row) + assistant_row_maps.append(row_map) + for column_name in row_map: + if column_name and column_name not in seen_columns: + seen_columns.add(column_name) + assistant_columns.append(column_name) + if not assistant_columns: + return False + + function_projections = set() + for function_row in function_rows: + if not isinstance(function_row, dict): + return False + row_map = _normalized_row_text_map(function_row) + if not seen_columns.issubset(row_map): + return False + function_projections.add(tuple(row_map[column_name] for column_name in assistant_columns)) + + return all( + tuple(row_map.get(column_name, '') for column_name in assistant_columns) in function_projections + for row_map in assistant_row_maps + ) + + def has_generated_file_output(existing_outputs: Optional[List[Dict[str, Any]]], output_format: str) -> bool: """Return whether an existing generated artifact already covers an output format.""" normalized_output_format = str(output_format or '').strip().lower() diff --git a/functional_tests/test_generated_csv_uses_authorized_action_rows.py b/functional_tests/test_generated_csv_uses_authorized_action_rows.py new file mode 100644 index 000000000..62d4cb073 --- /dev/null +++ b/functional_tests/test_generated_csv_uses_authorized_action_rows.py @@ -0,0 +1,283 @@ +# test_generated_csv_uses_authorized_action_rows.py +#!/usr/bin/env python3 +""" +Functional test for CSV artifacts built from authorized action rows. +Version: 0.260.007 +Implemented in: 0.260.007 + +Two customer conversations asked a telemetry agent to retrieve BatteryVoltage1 +and create a CSV. The agent retrieved 900 authorized samples, but the published +artifact held 3 rows in one conversation and 2 rows of server discovery metadata +in the other. + +Three defects combined: + 1. build_generated_file_export() let any assistant-rendered table outrank the + authorized action rows, so an illustrative excerpt replaced the dataset. + 2. CSV never received the publication contract JSON/XML already had, so the + model said it could not attach files and pasted a sample instead. + 3. A turn whose reply was the schema clarification still published a CSV, + built from whatever incidental discovery rows the turn happened to produce. + +This test ensures the authorized rows win over an excerpt, a deliberate +assistant table still wins, and a clarification turn publishes nothing. +""" + +import sys +import traceback +from pathlib import Path + +ROOT = Path(__file__).resolve().parents[1] +APP_DIR = ROOT / 'application' / 'single_app' +if str(APP_DIR) not in sys.path: + sys.path.insert(0, str(APP_DIR)) +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 functions_generated_file_exports import ( # noqa: E402 + build_generated_file_export, + build_generated_file_output_guidance, +) + +IMPLEMENTED_VERSION = '0.260.007' + +TELEMETRY_COLUMNS = ( + 'generation_time_utc', + 'reception_time_utc', + 'battery_voltage_1_v', + 'raw_value', + 'monitoring_result', + 'validity_status', +) +TELEMETRY_ROWS = [ + { + 'generation_time_utc': f'2026-08-19T15:37:{index % 60:02d}.371000+00:00', + 'reception_time_utc': f'2026-08-19T15:37:{index % 60:02d}.372000+00:00', + 'battery_voltage_1_v': 27 + (index % 5), + 'raw_value': 27 + (index % 5), + 'monitoring_result': 'CRITICAL', + 'validity_status': 'ACQUIRED', + } + for index in range(900) +] +TELEMETRY_FUNCTION_RESULTS = [{ + 'plugin_name': 'YamcsPlugin', + 'function_name': 'list_parameter_history', + 'success': True, + 'function_result': {'row_count': 900, 'rows': TELEMETRY_ROWS}, +}] +TELEMETRY_QUESTION = ( + 'grab BatteryVoltage1 over the last 15 minutes, provide high granularity and create a csv' +) + +DISCOVERY_FUNCTION_RESULTS = [ + { + 'plugin_name': 'YamcsPlugin', + 'function_name': 'list_instances', + 'success': True, + 'function_result': {'rows': [{'name': 'simulator', 'state': 'RUNNING'}]}, + }, + { + 'plugin_name': 'YamcsPlugin', + 'function_name': 'list_parameters', + 'success': True, + 'function_result': {'rows': [{'qualified_name': '/YSS/SIMULATOR/BatteryVoltage1', 'units': 'V'}]}, + }, +] +CLARIFICATION_REPLY = ( + 'Paul, should each CSV row represent an extracted BatteryVoltage1 telemetry sample, ' + 'and which columns should it include, for example: generation time, voltage (V), ' + 'monitoring result, and validity status?' +) + + +def assert_true(condition, message): + if not condition: + raise AssertionError(message) + + +def build_sample_excerpt_reply(excerpt_row_count=3): + """Rebuild the assistant reply that pasted real rows as an illustrative sample.""" + sample_lines = [','.join(TELEMETRY_COLUMNS)] + for row in TELEMETRY_ROWS[:excerpt_row_count]: + sample_lines.append(','.join(str(row[column]) for column in TELEMETRY_COLUMNS)) + rendered_sample = '\n'.join(sample_lines) + return ( + 'Paul, I retrieved **BatteryVoltage1** at 1 Hz. Samples: 900. Observed range 27 to 31 V.\n\n' + 'I do not have a file-creation or attachment capability in this session, so I cannot ' + 'deliver a downloadable CSV artifact. The CSV schema is:\n\n' + f'```\n{rendered_sample}\n```\n\n' + 'The archive response establishes the full 900-row dataset.\n' + ) + + +def test_pasted_sample_does_not_replace_authorized_rows(): + """An excerpt of the authorized rows must not shrink the published artifact.""" + print('Testing pasted-sample precedence...') + assert_app_version_at_least(IMPLEMENTED_VERSION) + + export_payload = build_generated_file_export( + TELEMETRY_QUESTION, + build_sample_excerpt_reply(), + function_results=TELEMETRY_FUNCTION_RESULTS, + ) + + assert_true(export_payload is not None, 'Expected the telemetry request to publish a CSV artifact.') + assert_true( + export_payload['row_count'] == len(TELEMETRY_ROWS), + f"Expected all {len(TELEMETRY_ROWS)} authorized rows; got {export_payload['row_count']}.", + ) + assert_true( + export_payload['row_source'] == 'structured function result', + 'Expected the authorized action rows to supply the artifact.', + ) + assert_true( + list(export_payload['preview_rows'][0]) == list(TELEMETRY_COLUMNS), + 'Expected the artifact to keep the telemetry column schema.', + ) + + +def test_deliberate_assistant_table_still_wins(): + """A table the model composed itself is not an excerpt and must stay authoritative.""" + print('Testing deliberate assistant table precedence...') + assert_app_version_at_least(IMPLEMENTED_VERSION) + + directory_results = [ + { + 'plugin_name': 'DirectoryPlugin', + 'function_name': 'list_people', + 'success': True, + 'function_result': {'value': [{'Name': 'Ada', 'Department': 'Engineering'}]}, + }, + { + 'plugin_name': 'DirectoryPlugin', + 'function_name': 'list_contractors', + 'success': True, + 'function_result': {'items': [{'Name': 'Grace', 'Department': 'Operations'}]}, + }, + ] + export_payload = build_generated_file_export( + 'create a combined CSV', + '| Name | Department |\n| --- | --- |\n| Assistant-selected | Finance |\n', + function_results=directory_results, + ) + + assert_true(export_payload is not None, 'Expected the combined CSV request to publish an artifact.') + assert_true( + export_payload['row_source'] == 'assistant response', + 'Expected a model-composed table to remain authoritative over action rows.', + ) + assert_true(export_payload['row_count'] == 1, 'Expected the model-composed row to be preserved.') + + +def test_partial_excerpt_of_larger_result_is_still_replaced(): + """A longer excerpt is still an excerpt, not a derived answer.""" + print('Testing multi-row excerpt precedence...') + assert_app_version_at_least(IMPLEMENTED_VERSION) + + export_payload = build_generated_file_export( + TELEMETRY_QUESTION, + build_sample_excerpt_reply(excerpt_row_count=25), + function_results=TELEMETRY_FUNCTION_RESULTS, + ) + + assert_true(export_payload is not None, 'Expected the telemetry request to publish a CSV artifact.') + assert_true( + export_payload['row_count'] == len(TELEMETRY_ROWS), + 'Expected a 25-row excerpt to be replaced by the full authorized result set.', + ) + + +def test_clarification_turn_publishes_no_artifact(): + """The turn that asks the schema clarification must not publish a CSV.""" + print('Testing clarification-turn suppression...') + assert_app_version_at_least(IMPLEMENTED_VERSION) + + export_payload = build_generated_file_export( + 'create a csv', + CLARIFICATION_REPLY, + function_results=DISCOVERY_FUNCTION_RESULTS, + ) + + assert_true( + export_payload is None, + 'Expected a clarifying question to publish no artifact instead of discovery metadata.', + ) + + +def test_answered_turn_still_publishes_action_rows(): + """Suppression must be limited to clarification replies, not ordinary summaries.""" + print('Testing normal reply still publishes action rows...') + assert_app_version_at_least(IMPLEMENTED_VERSION) + + export_payload = build_generated_file_export( + 'create a csv', + 'Retrieved the BatteryVoltage1 archive window and prepared the export.', + function_results=TELEMETRY_FUNCTION_RESULTS, + ) + + assert_true(export_payload is not None, 'Expected an ordinary reply to still publish the artifact.') + assert_true( + export_payload['row_count'] == len(TELEMETRY_ROWS), + 'Expected the full authorized result set for an ordinary reply.', + ) + + +def test_csv_guidance_states_the_publication_contract(): + """CSV must receive the same publication contract JSON and XML already state.""" + print('Testing CSV publication guidance parity...') + assert_app_version_at_least(IMPLEMENTED_VERSION) + + csv_guidance = build_generated_file_output_guidance('create a csv') + assert_true( + 'attaches the file after generation' in csv_guidance, + 'Expected CSV guidance to state that the server attaches the artifact.', + ) + assert_true( + 'cannot create or attach files' in csv_guidance, + 'Expected CSV guidance to forbid claiming files cannot be attached.', + ) + assert_true( + 'do not paste a sample of the rows' in csv_guidance, + 'Expected CSV guidance to forbid pasting a row sample instead of the artifact.', + ) + assert_true( + 'ask exactly one concise clarification' in csv_guidance, + 'Expected the existing one-clarification rule to survive.', + ) + assert_true( + build_generated_file_output_guidance('summarize the selected sources') == '', + 'Expected non-artifact requests to receive no generated-file guidance.', + ) + + +def run_tests() -> bool: + tests = [ + test_pasted_sample_does_not_replace_authorized_rows, + test_deliberate_assistant_table_still_wins, + test_partial_excerpt_of_larger_result_is_still_replaced, + test_clarification_turn_publishes_no_artifact, + test_answered_turn_still_publishes_action_rows, + test_csv_guidance_states_the_publication_contract, + ] + + results = [] + for test in tests: + print(f'\nRunning {test.__name__}...') + try: + test() + print(f'{test.__name__} passed') + results.append(True) + except Exception as exc: + print(f'{test.__name__} failed: {exc}') + traceback.print_exc() + results.append(False) + + passed = sum(1 for result in results if result) + print(f'\nResults: {passed}/{len(tests)} tests passed') + return all(results) + + +if __name__ == '__main__': + sys.exit(0 if run_tests() else 1) From fd8c56fc919b8106620c9a4bda3f9bdfa3952e8b Mon Sep 17 00:00:00 2001 From: Paul Lizer Date: Wed, 19 Aug 2026 15:26:52 -0400 Subject: [PATCH 2/3] fix(csv): reuse rows gathered in earlier turns and resume answered clarifications Follow-up to the same customer conversations. Even with the authorized rows winning, two requests still had nothing to export. Discovery calls diluted the dataset. A turn that ran list_instances, list_parameters, then two pages of list_parameter_history merged all four into one table with a Source action column, so the artifact carried lookup rows alongside the samples. Action rows are now grouped by action, so paged calls to one action stay one dataset, and a retrieval result that dwarfs the turn's other calls is used on its own. "create a csv" referring to earlier data had no rows. Action results were read from the current turn only, so a follow-up request exported nothing even though the conversation already held the full result set in its stored citations. When a turn produces no assistant rows and no action rows, the export now reaches back through recent assistant turns, rehydrates their full citation payloads from the stored artifact records, and uses the most recent turn that gathered rows. The reach-back is skipped whenever the turn gathered its own rows, so rows are never double counted, and the conversation is reauthorized before any message read. Answering the schema clarification did nothing. The reply that answers it does not mention CSV, so the request was not recognized and no artifact was published. An unanswered clarification now carries the pending CSV request forward to exactly the next user turn, unless that turn requests a different format. The passthrough contract established by the original request carries with it. Version: 0.260.008 --- application/single_app/config.py | 2 +- .../functions_generated_file_exports.py | 118 +++++++++--- application/single_app/route_backend_chats.py | 121 ++++++++++++ ...nerated_csv_uses_authorized_action_rows.py | 182 +++++++++++++++++- 4 files changed, 397 insertions(+), 26 deletions(-) diff --git a/application/single_app/config.py b/application/single_app/config.py index 39d9a8a90..8dd84f2b2 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.007" +VERSION = "0.260.008" 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 00104480c..50b624480 100644 --- a/application/single_app/functions_generated_file_exports.py +++ b/application/single_app/functions_generated_file_exports.py @@ -8,7 +8,7 @@ import re import tempfile from datetime import datetime -from typing import Any, Dict, Iterable, List, Optional, Sequence, Tuple +from typing import Any, Callable, Dict, Iterable, List, Optional, Sequence, Tuple from xml.etree import ElementTree from defusedxml import ElementTree as DefusedElementTree @@ -202,6 +202,13 @@ re.compile(r'\b(?:columns?|fields?)\s+(?:should|would|do)\s+(?:it|they|the\s+\w+)\s+include\b'), ) CSV_CLARIFICATION_REPLY_MAX_CHARS = 500 +# A retrieval action that dwarfs the turn's other calls is the payload; the rest are lookups. +DOMINANT_FUNCTION_RESULT_MIN_ROWS = 10 +DOMINANT_FUNCTION_RESULT_RATIO = 5 +ROW_SOURCE_ASSISTANT = 'assistant response' +ROW_SOURCE_CURRENT_ACTION = 'structured function result' +ROW_SOURCE_EARLIER_ACTION = 'earlier action result' +FUNCTION_RESULT_ROW_SOURCES = frozenset({ROW_SOURCE_CURRENT_ACTION, ROW_SOURCE_EARLIER_ACTION}) CSV_PUBLICATION_GUIDANCE = ( 'The server serializes the authorized action results into the requested CSV artifact and ' 'attaches the file after generation. Do not claim that you cannot create or attach files, ' @@ -411,6 +418,7 @@ def evaluate_generated_file_passthrough_eligibility( user_question: str, rows: Optional[Sequence[Dict[str, Any]]] = None, public_output_schema: Optional[Sequence[str]] = None, + carried_output_format: Optional[str] = None, ) -> Dict[str, Any]: """Return whether raw rows can satisfy a requested generated file contract.""" normalized_question = _normalize_question_for_artifact_detection(user_question) @@ -436,6 +444,9 @@ def evaluate_generated_file_passthrough_eligibility( return {'allowed': False, 'reason_code': 'derived_output_requires_transform'} if _question_requests_serialization(normalized_question): return {'allowed': True, 'reason_code': 'explicit_format_conversion'} + if carried_output_format: + # The format contract came from the earlier request this reply answers. + return {'allowed': True, 'reason_code': 'pending_format_clarification'} return {'allowed': False, 'reason_code': 'no_explicit_passthrough_contract'} @@ -498,18 +509,32 @@ def build_generated_file_export( user_question: str, assistant_content: str, function_results: Optional[List[Dict[str, Any]]] = None, + prior_function_results_loader: Optional[Callable[[], Optional[List[Dict[str, Any]]]]] = None, + pending_output_format: Optional[str] = None, ) -> Optional[Dict[str, Any]]: """Build a generated file payload from final assistant content and function-result evidence.""" - output_format = get_requested_generated_file_format(user_question) + output_format = get_requested_generated_file_format(user_question) or _normalize_pending_output_format( + pending_output_format, + ) if output_format not in GENERATED_FILE_FORMATS: return None + carried_output_format = ( + output_format if not get_requested_generated_file_format(user_question) else None + ) assistant_text = str(assistant_content or '').strip() assistant_rows = extract_assistant_table_entries(assistant_text) function_rows = extract_authorized_function_result_rows(function_results) + row_provenance = ROW_SOURCE_CURRENT_ACTION + if not assistant_rows and not function_rows and prior_function_results_loader is not None: + # This turn answered from evidence gathered earlier, so reuse those rows instead of none. + function_rows = extract_authorized_function_result_rows(prior_function_results_loader()) + if function_rows: + row_provenance = ROW_SOURCE_EARLIER_ACTION function_passthrough = evaluate_generated_file_passthrough_eligibility( user_question, rows=function_rows, + carried_output_format=carried_output_format, ) if function_rows else {'allowed': False, 'reason_code': 'source_result_incomplete'} if output_format == GENERATED_FILE_FORMAT_CSV: @@ -522,7 +547,7 @@ def build_generated_file_export( rows = function_rows if use_function_rows else assistant_rows if not rows: return None - row_source = 'structured function result' if use_function_rows else 'assistant response' + row_source = row_provenance if use_function_rows else ROW_SOURCE_ASSISTANT return _build_generated_file_payload( output_format=output_format, file_content=build_assistant_table_csv(rows), @@ -537,7 +562,7 @@ def build_generated_file_export( rows = function_rows if function_rows and function_passthrough.get('allowed') else assistant_rows if not assistant_text and not rows: return None - row_source = 'structured function result' if rows and rows is function_rows else 'assistant response' + row_source = row_provenance if rows and rows is function_rows else ROW_SOURCE_ASSISTANT title = _build_generated_file_title(output_format) if output_format == GENERATED_FILE_FORMAT_DOCX: file_content = _render_docx_file_export(title, assistant_text, rows, row_source) @@ -555,8 +580,32 @@ def build_generated_file_export( def extract_authorized_function_result_rows(function_results: Optional[List[Dict[str, Any]]]) -> List[Dict[str, Any]]: - """Return structured rows from successful current-turn non-tabular function results.""" - function_row_groups: List[Tuple[str, List[Dict[str, Any]]]] = [] + """Return structured rows from successful non-tabular action results.""" + function_row_groups = _collect_authorized_function_row_groups(function_results) + if not function_row_groups: + return [] + + dominant_rows = _select_dominant_function_row_group(function_row_groups) + if dominant_rows is not None: + return dominant_rows + if len(function_row_groups) == 1: + return function_row_groups[0][1] + + source_column = _get_function_result_source_column(function_row_groups) + combined_rows = [] + for function_label, rows in function_row_groups: + for row in rows: + normalized_row = dict(row) + normalized_row[source_column] = function_label + combined_rows.append(normalized_row) + return combined_rows + + +def _collect_authorized_function_row_groups( + function_results: Optional[List[Dict[str, Any]]], +) -> List[Tuple[str, List[Dict[str, Any]]]]: + """Group authorized rows by action, so paged calls to one action stay one dataset.""" + grouped_rows: Dict[str, List[Dict[str, Any]]] = {} for function_result in function_results or []: if not isinstance(function_result, dict): continue @@ -568,24 +617,49 @@ def extract_authorized_function_result_rows(function_results: Optional[List[Dict ) if not structured_rows: continue - function_row_groups.append(( + grouped_rows.setdefault( _get_function_result_label(function_result), - structured_rows, - )) + [], + ).extend(structured_rows) + return list(grouped_rows.items()) - if not function_row_groups: - return [] - if len(function_row_groups) == 1: - return function_row_groups[0][1] - source_column = _get_function_result_source_column(function_row_groups) - combined_rows = [] - for function_label, rows in function_row_groups: - for row in rows: - normalized_row = dict(row) - normalized_row[source_column] = function_label - combined_rows.append(normalized_row) - return combined_rows +def _select_dominant_function_row_group( + function_row_groups: Sequence[Tuple[str, List[Dict[str, Any]]]], +) -> Optional[List[Dict[str, Any]]]: + """Return the retrieval result set when the turn's other actions are only lookups.""" + if len(function_row_groups) < 2: + return None + + dominant_label, dominant_rows = max(function_row_groups, key=lambda group: len(group[1])) + if len(dominant_rows) < DOMINANT_FUNCTION_RESULT_MIN_ROWS: + return None + + supporting_row_count = sum( + len(rows) for label, rows in function_row_groups if label != dominant_label + ) + if len(dominant_rows) < supporting_row_count * DOMINANT_FUNCTION_RESULT_RATIO: + return None + return dominant_rows + + +def _normalize_pending_output_format(pending_output_format: Optional[str]) -> Optional[str]: + normalized_format = str(pending_output_format or '').strip().lower() + return normalized_format if normalized_format in GENERATED_FILE_FORMATS else None + + +def resolve_pending_generated_file_format( + user_question: str, + previous_assistant_content: str, +) -> Optional[str]: + """Carry one unanswered CSV schema clarification forward to the reply that answers it.""" + if not str(user_question or '').strip(): + return None + if get_requested_artifact_formats(user_question): + return None + if not _assistant_reply_requests_clarification(previous_assistant_content): + return None + return GENERATED_FILE_FORMAT_CSV def _assistant_reply_requests_clarification(assistant_content: str) -> bool: @@ -1030,7 +1104,7 @@ def _format_structured_cell(value: Any) -> str: def _build_structured_rows_heading(row_source: str) -> str: - if row_source == 'structured function result': + if row_source in FUNCTION_RESULT_ROW_SOURCES: return 'Structured function results' return 'Structured response rows' diff --git a/application/single_app/route_backend_chats.py b/application/single_app/route_backend_chats.py index 1587cd4ad..ff2608652 100644 --- a/application/single_app/route_backend_chats.py +++ b/application/single_app/route_backend_chats.py @@ -158,6 +158,7 @@ normalize_json_artifact_payload, normalize_generated_output_format, normalize_xml_artifact_payload, + resolve_pending_generated_file_format, serialize_generated_json, serialize_generated_xml, ) @@ -2257,6 +2258,115 @@ def _has_generated_tabular_csv_output(generated_outputs): return has_generated_tabular_csv_output(generated_outputs) +PRIOR_TURN_ACTION_RESULT_MESSAGE_LIMIT = 5 +PRIOR_TURN_ACTION_RESULT_ARTIFACT_LIMIT = 25 + + +def _read_recent_assistant_messages(conversation_id, message_limit): + query = ( + f'SELECT TOP {int(message_limit)} c.id, c.content, c.agent_citations FROM c ' + 'WHERE c.conversation_id = @conversation_id AND c.role = @role ' + 'ORDER BY c.timestamp DESC' + ) + return list(cosmos_messages_container.query_items( + query=query, + parameters=[ + {'name': '@conversation_id', 'value': conversation_id}, + {'name': '@role', 'value': 'assistant'}, + ], + partition_key=conversation_id, + )) + + +def _hydrate_assistant_message_agent_citations(conversation_id, assistant_message): + """Inflate one assistant turn's compact citations back to their full stored payloads.""" + compact_citations = assistant_message.get('agent_citations') + if not isinstance(compact_citations, list) or not compact_citations: + return [] + + artifact_ids = [] + for citation in compact_citations: + if not isinstance(citation, dict): + continue + artifact_id = str(citation.get('artifact_id') or '').strip() + if artifact_id and artifact_id not in artifact_ids: + artifact_ids.append(artifact_id) + if len(artifact_ids) >= PRIOR_TURN_ACTION_RESULT_ARTIFACT_LIMIT: + break + + if not artifact_ids: + return [citation for citation in compact_citations if isinstance(citation, dict)] + + artifact_documents = list(cosmos_messages_container.query_items( + query=( + 'SELECT * FROM c WHERE c.conversation_id = @conversation_id ' + 'AND (ARRAY_CONTAINS(@artifact_ids, c.id) ' + 'OR ARRAY_CONTAINS(@artifact_ids, c.parent_message_id))' + ), + parameters=[ + {'name': '@conversation_id', 'value': conversation_id}, + {'name': '@artifact_ids', 'value': artifact_ids}, + ], + partition_key=conversation_id, + )) + hydrated_messages = hydrate_agent_citations_from_artifacts( + [assistant_message], + build_message_artifact_payload_map(artifact_documents), + ) + return [ + citation + for citation in (hydrated_messages[0].get('agent_citations') or []) + if isinstance(citation, dict) + ] + + +def _load_prior_turn_function_results(user_id, conversation_id): + """Return full action results from the most recent earlier turn that gathered rows.""" + normalized_user_id = str(user_id or '').strip() + normalized_conversation_id = str(conversation_id or '').strip() + if not normalized_user_id or not normalized_conversation_id: + return [] + + try: + _authorize_personal_conversation_access(normalized_user_id, normalized_conversation_id) + assistant_messages = _read_recent_assistant_messages( + normalized_conversation_id, + PRIOR_TURN_ACTION_RESULT_MESSAGE_LIMIT, + ) + for assistant_message in assistant_messages: + hydrated_citations = _hydrate_assistant_message_agent_citations( + normalized_conversation_id, + assistant_message, + ) + if hydrated_citations: + return hydrated_citations + except Exception as exc: + log_event( + '[GENERATED_FILE_EXPORT] Could not reuse earlier action results', + { + 'conversation_id': normalized_conversation_id, + 'error': str(exc), + }, + debug_only=True, + ) + return [] + + +def _resolve_pending_generated_file_format(user_question, conversation_id, user_id): + """Return the artifact format still owed from an unanswered schema clarification.""" + try: + _authorize_personal_conversation_access(str(user_id or '').strip(), str(conversation_id or '').strip()) + recent_assistant_messages = _read_recent_assistant_messages(str(conversation_id or '').strip(), 1) + except Exception: + return None + if not recent_assistant_messages: + return None + return resolve_pending_generated_file_format( + user_question, + recent_assistant_messages[0].get('content') or '', + ) + + def maybe_create_generated_file_output( user_question, assistant_content, @@ -2273,6 +2383,12 @@ def maybe_create_generated_file_output( request_correlation_id=request_correlation_id, ) output_format = get_requested_generated_file_format(user_question) + if not output_format: + output_format = _resolve_pending_generated_file_format( + user_question, + conversation_id, + get_current_user_id(), + ) if not output_format: return None if output_format == 'csv' and _has_generated_tabular_csv_output(existing_outputs): @@ -2284,6 +2400,11 @@ def maybe_create_generated_file_output( user_question, assistant_content, function_results=function_results, + prior_function_results_loader=lambda: _load_prior_turn_function_results( + get_current_user_id(), + conversation_id, + ), + pending_output_format=output_format, ) if not export_payload: return None diff --git a/functional_tests/test_generated_csv_uses_authorized_action_rows.py b/functional_tests/test_generated_csv_uses_authorized_action_rows.py index 62d4cb073..f38fbb005 100644 --- a/functional_tests/test_generated_csv_uses_authorized_action_rows.py +++ b/functional_tests/test_generated_csv_uses_authorized_action_rows.py @@ -2,8 +2,8 @@ #!/usr/bin/env python3 """ Functional test for CSV artifacts built from authorized action rows. -Version: 0.260.007 -Implemented in: 0.260.007 +Version: 0.260.008 +Implemented in: 0.260.007; earlier-turn reach-back and clarification carry-forward in 0.260.008 Two customer conversations asked a telemetry agent to retrieve BatteryVoltage1 and create a CSV. The agent retrieved 900 authorized samples, but the published @@ -17,9 +17,14 @@ model said it could not attach files and pasted a sample instead. 3. A turn whose reply was the schema clarification still published a CSV, built from whatever incidental discovery rows the turn happened to produce. + 4. Discovery and lookup calls were merged into the retrieved dataset, adding + junk rows and a Source action column to the artifact. + 5. Nothing reached back to data an earlier turn already gathered, so a + follow-up "create a csv" and an answered clarification both had no rows. This test ensures the authorized rows win over an excerpt, a deliberate -assistant table still wins, and a clarification turn publishes nothing. +assistant table still wins, a clarification turn publishes nothing, and an +answered clarification resumes the original request using the earlier rows. """ import sys @@ -38,6 +43,8 @@ from functions_generated_file_exports import ( # noqa: E402 build_generated_file_export, build_generated_file_output_guidance, + extract_authorized_function_result_rows, + resolve_pending_generated_file_format, ) IMPLEMENTED_VERSION = '0.260.007' @@ -67,6 +74,33 @@ 'success': True, 'function_result': {'row_count': 900, 'rows': TELEMETRY_ROWS}, }] +# The real turn also ran two discovery calls before paging the history. +FULL_TURN_FUNCTION_RESULTS = [ + { + 'plugin_name': 'YamcsPlugin', + 'function_name': 'list_instances', + 'success': True, + 'function_result': {'rows': [{'name': 'simulator', 'state': 'RUNNING'}]}, + }, + { + 'plugin_name': 'YamcsPlugin', + 'function_name': 'list_parameters', + 'success': True, + 'function_result': {'rows': [{'qualified_name': '/YSS/SIMULATOR/BatteryVoltage1', 'units': 'V'}]}, + }, + { + 'plugin_name': 'YamcsPlugin', + 'function_name': 'list_parameter_history', + 'success': True, + 'function_result': {'rows': TELEMETRY_ROWS[:450]}, + }, + { + 'plugin_name': 'YamcsPlugin', + 'function_name': 'list_parameter_history', + 'success': True, + 'function_result': {'rows': TELEMETRY_ROWS[450:]}, + }, +] TELEMETRY_QUESTION = ( 'grab BatteryVoltage1 over the last 15 minutes, provide high granularity and create a csv' ) @@ -224,6 +258,142 @@ def test_answered_turn_still_publishes_action_rows(): ) +def test_discovery_calls_do_not_dilute_the_retrieved_dataset(): + """Lookup and discovery calls in the same turn must not become artifact rows.""" + print('Testing dominant action-result selection...') + assert_app_version_at_least(IMPLEMENTED_VERSION) + + rows = extract_authorized_function_result_rows(FULL_TURN_FUNCTION_RESULTS) + + assert_true( + len(rows) == len(TELEMETRY_ROWS), + f'Expected only the {len(TELEMETRY_ROWS)} retrieved samples; got {len(rows)}.', + ) + assert_true( + list(rows[0]) == list(TELEMETRY_COLUMNS), + 'Expected the telemetry schema without a Source action column from discovery calls.', + ) + + +def test_paged_calls_to_one_action_stay_one_dataset(): + """Two pages of the same action are one dataset, not two labeled groups.""" + print('Testing paged action-result grouping...') + assert_app_version_at_least(IMPLEMENTED_VERSION) + + paged_results = [ + { + 'plugin_name': 'YamcsPlugin', + 'function_name': 'list_parameter_history', + 'success': True, + 'function_result': {'rows': TELEMETRY_ROWS[:450]}, + }, + { + 'plugin_name': 'YamcsPlugin', + 'function_name': 'list_parameter_history', + 'success': True, + 'function_result': {'rows': TELEMETRY_ROWS[450:]}, + }, + ] + rows = extract_authorized_function_result_rows(paged_results) + + assert_true(len(rows) == len(TELEMETRY_ROWS), 'Expected both pages to be kept.') + assert_true( + 'Source action' not in rows[0], + 'Expected one action to stay one dataset without a Source action column.', + ) + + +def test_followup_request_reuses_earlier_turn_rows(): + """'create a csv' must reach back to data an earlier turn already gathered.""" + print('Testing earlier-turn reach-back...') + assert_app_version_at_least(IMPLEMENTED_VERSION) + + export_payload = build_generated_file_export( + 'create a csv', + 'Here is the CSV of the BatteryVoltage1 samples I retrieved.', + function_results=[], + prior_function_results_loader=lambda: FULL_TURN_FUNCTION_RESULTS, + ) + + assert_true(export_payload is not None, 'Expected the follow-up request to publish an artifact.') + assert_true( + export_payload['row_count'] == len(TELEMETRY_ROWS), + 'Expected the earlier turn\'s full result set to be reused.', + ) + assert_true( + export_payload['row_source'] == 'earlier action result', + 'Expected reused rows to declare their earlier-turn provenance.', + ) + + +def test_current_turn_rows_are_never_double_counted(): + """The reach-back must not run when the turn gathered its own rows.""" + print('Testing reach-back suppression when the turn has data...') + assert_app_version_at_least(IMPLEMENTED_VERSION) + + export_payload = build_generated_file_export( + 'create a csv', + 'Retrieved the archive window and prepared the export.', + function_results=FULL_TURN_FUNCTION_RESULTS, + prior_function_results_loader=lambda: FULL_TURN_FUNCTION_RESULTS, + ) + + assert_true(export_payload is not None, 'Expected the request to publish an artifact.') + assert_true( + export_payload['row_count'] == len(TELEMETRY_ROWS), + 'Expected current-turn rows only, with no earlier-turn rows appended.', + ) + assert_true( + export_payload['row_source'] == 'structured function result', + 'Expected current-turn provenance when the turn gathered its own rows.', + ) + + +def test_answering_the_clarification_publishes_the_csv(): + """Answering the schema clarification must resume the original CSV request.""" + print('Testing clarification answer carry-forward...') + assert_app_version_at_least(IMPLEMENTED_VERSION) + + pending_format = resolve_pending_generated_file_format( + 'yes, one row per sample', + CLARIFICATION_REPLY, + ) + assert_true(pending_format == 'csv', 'Expected the answered clarification to resume a CSV request.') + + export_payload = build_generated_file_export( + 'yes, one row per sample', + 'Understood, one row per sample.', + function_results=[], + prior_function_results_loader=lambda: FULL_TURN_FUNCTION_RESULTS, + pending_output_format=pending_format, + ) + + assert_true(export_payload is not None, 'Expected the answered clarification to publish the CSV.') + assert_true( + export_payload['row_count'] == len(TELEMETRY_ROWS), + 'Expected the answered clarification to export the already-gathered samples.', + ) + + +def test_pending_clarification_does_not_override_a_new_request(): + """A reply that asks for another format or follows no clarification carries nothing.""" + print('Testing pending clarification boundaries...') + assert_app_version_at_least(IMPLEMENTED_VERSION) + + assert_true( + resolve_pending_generated_file_format('actually make it a pdf', CLARIFICATION_REPLY) is None, + 'Expected a different requested format to cancel the pending CSV request.', + ) + assert_true( + resolve_pending_generated_file_format('yes, one row per sample', 'Here are the results.') is None, + 'Expected no pending request when the previous turn asked nothing.', + ) + assert_true( + resolve_pending_generated_file_format('', CLARIFICATION_REPLY) is None, + 'Expected an empty reply to carry no pending request.', + ) + + def test_csv_guidance_states_the_publication_contract(): """CSV must receive the same publication contract JSON and XML already state.""" print('Testing CSV publication guidance parity...') @@ -259,6 +429,12 @@ def run_tests() -> bool: test_partial_excerpt_of_larger_result_is_still_replaced, test_clarification_turn_publishes_no_artifact, test_answered_turn_still_publishes_action_rows, + test_discovery_calls_do_not_dilute_the_retrieved_dataset, + test_paged_calls_to_one_action_stay_one_dataset, + test_followup_request_reuses_earlier_turn_rows, + test_current_turn_rows_are_never_double_counted, + test_answering_the_clarification_publishes_the_csv, + test_pending_clarification_does_not_override_a_new_request, test_csv_guidance_states_the_publication_contract, ] From b6954f7888c3a3582cdfd3c2321712ae09cd1e20 Mon Sep 17 00:00:00 2001 From: Paul Lizer Date: Wed, 19 Aug 2026 15:55:20 -0400 Subject: [PATCH 3/3] fix(csv): scan stored citations before reading any earlier payload The reach-back read every artifact of a turn just to learn whether it held rows, which capped the window at 5 assistant messages and made a wider one expensive. It also took the first turn with any rows, so a newer memory or discovery call outranked the dataset a request actually meant. Compaction already leaves the answer on the message. A 54KB telemetry result is stored on the assistant message as a 677 byte citation that still carries row_count, the column names, and a remaining_items marker for the dropped tail. That is enough to decide which action holds the data without reading a single payload. Selection is now two phase. select_prior_turn_action_citations() scans the compact citations already returned by the message query, groups them by action, and picks the newest group whose recorded rows clear the dataset threshold, falling back to the most recent rows when no turn has a dataset. Only the winning group's artifacts are then fetched and expanded. The window now follows conversation_history_limit from admin settings rather than a fixed 5, so anything the model can still see in context is reachable by the exporter, capped at 40 messages. Selection is pure and lives with the other row-shape logic; the route keeps authorization and I/O. A guard test pins the compaction assumption so a change to the compaction limits fails loudly instead of silently degrading the scan. Version: 0.260.009 --- application/single_app/config.py | 2 +- .../functions_generated_file_exports.py | 103 +++++++++++++ application/single_app/route_backend_chats.py | 80 +++++----- ...nerated_csv_uses_authorized_action_rows.py | 145 +++++++++++++++++- 4 files changed, 289 insertions(+), 41 deletions(-) diff --git a/application/single_app/config.py b/application/single_app/config.py index 8dd84f2b2..e342eef4a 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.008" +VERSION = "0.260.009" 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 50b624480..42199ae80 100644 --- a/application/single_app/functions_generated_file_exports.py +++ b/application/single_app/functions_generated_file_exports.py @@ -101,6 +101,15 @@ 'output', 'payload', ) +FUNCTION_RESULT_ROW_COUNT_KEYS = ( + 'row_count', + 'returned_rows', + 'total_rows', + 'total_count', + 'record_count', +) +# Below this a stored action reads as a lookup, not the dataset a request is asking for. +PRIOR_TURN_SUBSTANTIVE_ROW_COUNT = 10 FUNCTION_RESULT_CONTROL_KEYS = { 'count', 'detail', @@ -643,6 +652,100 @@ def _select_dominant_function_row_group( return dominant_rows +def select_prior_turn_action_citations( + assistant_messages: Optional[Sequence[Dict[str, Any]]], + substantive_row_count: int = PRIOR_TURN_SUBSTANTIVE_ROW_COUNT, +) -> List[Dict[str, Any]]: + """Pick the newest stored action group holding a dataset, reading compact citations only. + + Callers pass assistant messages newest first and hydrate only the returned citations. + """ + fallback_citations: List[Dict[str, Any]] = [] + for assistant_message in assistant_messages or []: + if not isinstance(assistant_message, dict): + continue + compact_citations = assistant_message.get('agent_citations') + if not isinstance(compact_citations, list): + continue + + grouped_citations: Dict[str, Dict[str, Any]] = {} + for citation in compact_citations: + if not isinstance(citation, dict): + continue + estimated_rows = estimate_function_result_row_count(citation) + if not estimated_rows: + continue + action_label = str( + citation.get('function_name') or citation.get('plugin_name') or '', + ).strip() + action_group = grouped_citations.setdefault( + action_label, + {'citations': [], 'estimated_rows': 0}, + ) + action_group['citations'].append(citation) + action_group['estimated_rows'] += estimated_rows + + if not grouped_citations: + continue + best_group = max(grouped_citations.values(), key=lambda group: group['estimated_rows']) + if best_group['estimated_rows'] >= substantive_row_count: + return list(best_group['citations']) + if not fallback_citations: + fallback_citations = list(best_group['citations']) + return fallback_citations + + +def estimate_function_result_row_count(function_result: Optional[Dict[str, Any]]) -> int: + """Estimate an action's stored rows from its compacted citation, reading no payload.""" + if not isinstance(function_result, dict): + return 0 + if function_result.get('success') is False or _is_tabular_function_result(function_result): + return 0 + return _estimate_payload_row_count( + _parse_function_result_payload(function_result.get('function_result')), + ) + + +def _estimate_payload_row_count(payload: Any, depth: int = 0) -> int: + if depth > 4: + return 0 + if isinstance(payload, list): + estimated_rows = 0 + for item in payload: + # Compaction replaces the dropped tail of a list with this marker. + if isinstance(item, dict) and set(item) == {'remaining_items'}: + estimated_rows += _coerce_row_count(item.get('remaining_items')) + else: + estimated_rows += 1 + return estimated_rows + if not isinstance(payload, dict): + return 0 + + normalized_keys = {_normalize_function_result_key(key): key for key in payload} + for count_key in FUNCTION_RESULT_ROW_COUNT_KEYS: + matching_key = normalized_keys.get(_normalize_function_result_key(count_key)) + declared_row_count = _coerce_row_count(payload.get(matching_key)) if matching_key else 0 + if declared_row_count: + return declared_row_count + + for row_key in FUNCTION_RESULT_ROW_KEYS: + matching_key = normalized_keys.get(_normalize_function_result_key(row_key)) + if matching_key is None: + continue + estimated_rows = _estimate_payload_row_count(payload.get(matching_key), depth + 1) + if estimated_rows: + return estimated_rows + return 1 if _normalize_function_result_row(payload, is_data_row=False) else 0 + + +def _coerce_row_count(value: Any) -> int: + try: + row_count = int(value) + except (TypeError, ValueError): + return 0 + return row_count if row_count > 0 else 0 + + def _normalize_pending_output_format(pending_output_format: Optional[str]) -> Optional[str]: normalized_format = str(pending_output_format or '').strip().lower() return normalized_format if normalized_format in GENERATED_FILE_FORMATS else None diff --git a/application/single_app/route_backend_chats.py b/application/single_app/route_backend_chats.py index ff2608652..996343a5a 100644 --- a/application/single_app/route_backend_chats.py +++ b/application/single_app/route_backend_chats.py @@ -151,6 +151,7 @@ build_generated_file_export, build_generated_file_output_guidance, evaluate_generated_file_passthrough_eligibility, + estimate_function_result_row_count, get_generated_file_export_content, get_requested_generated_file_format, get_requested_structured_artifact_format, @@ -159,6 +160,7 @@ normalize_generated_output_format, normalize_xml_artifact_payload, resolve_pending_generated_file_format, + select_prior_turn_action_citations, serialize_generated_json, serialize_generated_xml, ) @@ -2258,10 +2260,21 @@ def _has_generated_tabular_csv_output(generated_outputs): return has_generated_tabular_csv_output(generated_outputs) -PRIOR_TURN_ACTION_RESULT_MESSAGE_LIMIT = 5 +PRIOR_TURN_ACTION_RESULT_MAX_MESSAGES = 40 PRIOR_TURN_ACTION_RESULT_ARTIFACT_LIMIT = 25 +def _resolve_prior_turn_history_window(settings=None): + """Reach back as far as the history the model itself can still see.""" + resolved_settings = settings if isinstance(settings, dict) else get_settings() + return _bounded_int( + resolved_settings.get('conversation_history_limit', 10), + default=10, + minimum=1, + maximum=PRIOR_TURN_ACTION_RESULT_MAX_MESSAGES, + ) + + def _read_recent_assistant_messages(conversation_id, message_limit): query = ( f'SELECT TOP {int(message_limit)} c.id, c.content, c.agent_citations FROM c ' @@ -2278,25 +2291,9 @@ def _read_recent_assistant_messages(conversation_id, message_limit): )) -def _hydrate_assistant_message_agent_citations(conversation_id, assistant_message): - """Inflate one assistant turn's compact citations back to their full stored payloads.""" - compact_citations = assistant_message.get('agent_citations') - if not isinstance(compact_citations, list) or not compact_citations: - return [] - - artifact_ids = [] - for citation in compact_citations: - if not isinstance(citation, dict): - continue - artifact_id = str(citation.get('artifact_id') or '').strip() - if artifact_id and artifact_id not in artifact_ids: - artifact_ids.append(artifact_id) - if len(artifact_ids) >= PRIOR_TURN_ACTION_RESULT_ARTIFACT_LIMIT: - break - +def _read_agent_citation_artifact_payloads(conversation_id, artifact_ids): if not artifact_ids: - return [citation for citation in compact_citations if isinstance(citation, dict)] - + return {} artifact_documents = list(cosmos_messages_container.query_items( query=( 'SELECT * FROM c WHERE c.conversation_id = @conversation_id ' @@ -2309,19 +2306,11 @@ def _hydrate_assistant_message_agent_citations(conversation_id, assistant_messag ], partition_key=conversation_id, )) - hydrated_messages = hydrate_agent_citations_from_artifacts( - [assistant_message], - build_message_artifact_payload_map(artifact_documents), - ) - return [ - citation - for citation in (hydrated_messages[0].get('agent_citations') or []) - if isinstance(citation, dict) - ] + return build_message_artifact_payload_map(artifact_documents) -def _load_prior_turn_function_results(user_id, conversation_id): - """Return full action results from the most recent earlier turn that gathered rows.""" +def _load_prior_turn_function_results(user_id, conversation_id, settings=None): + """Return the full action results an earlier turn already gathered in this conversation.""" normalized_user_id = str(user_id or '').strip() normalized_conversation_id = str(conversation_id or '').strip() if not normalized_user_id or not normalized_conversation_id: @@ -2331,15 +2320,29 @@ def _load_prior_turn_function_results(user_id, conversation_id): _authorize_personal_conversation_access(normalized_user_id, normalized_conversation_id) assistant_messages = _read_recent_assistant_messages( normalized_conversation_id, - PRIOR_TURN_ACTION_RESULT_MESSAGE_LIMIT, + _resolve_prior_turn_history_window(settings), ) - for assistant_message in assistant_messages: - hydrated_citations = _hydrate_assistant_message_agent_citations( - normalized_conversation_id, - assistant_message, - ) - if hydrated_citations: - return hydrated_citations + selected_citations = select_prior_turn_action_citations( + assistant_messages, + )[:PRIOR_TURN_ACTION_RESULT_ARTIFACT_LIMIT] + if not selected_citations: + return [] + + artifact_payload_map = _read_agent_citation_artifact_payloads( + normalized_conversation_id, + [ + str(citation.get('artifact_id') or '').strip() + for citation in selected_citations + if str(citation.get('artifact_id') or '').strip() + ], + ) + + resolved_citations = [] + for citation in selected_citations: + artifact_payload = artifact_payload_map.get(str(citation.get('artifact_id') or '')) + stored_citation = artifact_payload.get('citation') if isinstance(artifact_payload, dict) else None + resolved_citations.append(stored_citation if isinstance(stored_citation, dict) else citation) + return resolved_citations except Exception as exc: log_event( '[GENERATED_FILE_EXPORT] Could not reuse earlier action results', @@ -2403,6 +2406,7 @@ def maybe_create_generated_file_output( prior_function_results_loader=lambda: _load_prior_turn_function_results( get_current_user_id(), conversation_id, + settings=get_settings(), ), pending_output_format=output_format, ) diff --git a/functional_tests/test_generated_csv_uses_authorized_action_rows.py b/functional_tests/test_generated_csv_uses_authorized_action_rows.py index f38fbb005..eba77f64d 100644 --- a/functional_tests/test_generated_csv_uses_authorized_action_rows.py +++ b/functional_tests/test_generated_csv_uses_authorized_action_rows.py @@ -2,8 +2,8 @@ #!/usr/bin/env python3 """ Functional test for CSV artifacts built from authorized action rows. -Version: 0.260.008 -Implemented in: 0.260.007; earlier-turn reach-back and clarification carry-forward in 0.260.008 +Version: 0.260.009 +Implemented in: 0.260.007; earlier-turn reach-back and clarification carry-forward in 0.260.008; history-limit window and compact-citation scan in 0.260.009 Two customer conversations asked a telemetry agent to retrieve BatteryVoltage1 and create a CSV. The agent retrieved 900 authorized samples, but the published @@ -27,6 +27,7 @@ answered clarification resumes the original request using the earlier rows. """ +import json import sys import traceback from pathlib import Path @@ -43,8 +44,10 @@ from functions_generated_file_exports import ( # noqa: E402 build_generated_file_export, build_generated_file_output_guidance, + estimate_function_result_row_count, extract_authorized_function_result_rows, resolve_pending_generated_file_format, + select_prior_turn_action_citations, ) IMPLEMENTED_VERSION = '0.260.007' @@ -394,6 +397,141 @@ def test_pending_clarification_does_not_override_a_new_request(): ) +def _build_compact_citation(artifact_id, plugin_name, function_name, function_result): + """Compact a citation exactly the way the chat pipeline stores it on a message.""" + import types + + if 'functions_azure_maps' not in sys.modules: + # functions_message_artifacts only needs this for map payloads, and importing the real + # module pulls in config.py and a live Cosmos client. + azure_maps_stub = types.ModuleType('functions_azure_maps') + azure_maps_stub.refresh_azure_maps_citation_payload = lambda payload: payload + sys.modules['functions_azure_maps'] = azure_maps_stub + + from functions_message_artifacts import build_compact_agent_citation + + return build_compact_agent_citation( + { + 'plugin_name': plugin_name, + 'function_name': function_name, + 'success': True, + 'function_result': function_result, + }, + artifact_id=artifact_id, + ) + + +def test_row_count_estimate_survives_citation_compaction(): + """The cheap scan depends on compaction keeping a usable row signal on the message.""" + print('Testing compacted row-count estimate...') + assert_app_version_at_least(IMPLEMENTED_VERSION) + + compact_citation = _build_compact_citation( + 'm1_artifact_2', + 'YamcsPlugin', + 'list_parameter_history', + {'instance': 'simulator', 'row_count': 900, 'rows': TELEMETRY_ROWS}, + ) + + assert_true( + len(json.dumps(compact_citation)) < 4000, + 'Expected the stored citation to stay compact.', + ) + assert_true( + estimate_function_result_row_count(compact_citation) == len(TELEMETRY_ROWS), + 'Expected the compacted citation to still report its full row count.', + ) + + unlabeled_citation = _build_compact_citation( + 'm1_artifact_3', + 'YamcsPlugin', + 'list_parameter_history', + {'rows': TELEMETRY_ROWS}, + ) + assert_true( + estimate_function_result_row_count(unlabeled_citation) == len(TELEMETRY_ROWS), + 'Expected the truncated row list marker to restore the full count.', + ) + + +def test_reach_back_prefers_the_dataset_over_newer_lookups(): + """A newer memory or discovery call must not outrank the dataset a request means.""" + print('Testing prior-turn action selection...') + assert_app_version_at_least(IMPLEMENTED_VERSION) + + assistant_messages = [ + { + 'id': 'm3', + 'agent_citations': [_build_compact_citation( + 'm3_artifact_1', 'FactMemoryPlugin', 'get_facts', + {'facts': [{'key': 'name', 'value': 'Paul'}]}, + )], + }, + { + 'id': 'm2', + 'agent_citations': [_build_compact_citation( + 'm2_artifact_1', 'YamcsPlugin', 'list_instances', + {'rows': [{'name': 'simulator', 'state': 'RUNNING'}]}, + )], + }, + { + 'id': 'm1', + 'agent_citations': [ + _build_compact_citation( + 'm1_artifact_1', 'YamcsPlugin', 'list_instances', + {'rows': [{'name': 'simulator', 'state': 'RUNNING'}]}, + ), + _build_compact_citation( + 'm1_artifact_2', 'YamcsPlugin', 'list_parameter_history', + {'row_count': 450, 'rows': TELEMETRY_ROWS[:450]}, + ), + _build_compact_citation( + 'm1_artifact_3', 'YamcsPlugin', 'list_parameter_history', + {'row_count': 450, 'rows': TELEMETRY_ROWS[450:]}, + ), + ], + }, + ] + + selected_citations = select_prior_turn_action_citations(assistant_messages) + + assert_true(len(selected_citations) == 2, 'Expected both pages of the dataset action.') + assert_true( + [citation['artifact_id'] for citation in selected_citations] == ['m1_artifact_2', 'm1_artifact_3'], + 'Expected only the dataset artifacts to be selected for hydration.', + ) + assert_true( + all(citation['function_name'] == 'list_parameter_history' for citation in selected_citations), + 'Expected discovery and memory lookups to be skipped.', + ) + + +def test_reach_back_falls_back_to_the_only_rows_available(): + """When no turn holds a dataset, the most recent rows are still better than nothing.""" + print('Testing prior-turn fallback selection...') + assert_app_version_at_least(IMPLEMENTED_VERSION) + + assistant_messages = [ + {'id': 'm2', 'agent_citations': [_build_compact_citation( + 'm2_artifact_1', 'DirectoryPlugin', 'list_people', + {'rows': [{'Name': 'Ada'}, {'Name': 'Grace'}]}, + )]}, + {'id': 'm1', 'agent_citations': []}, + ] + + selected_citations = select_prior_turn_action_citations(assistant_messages) + + assert_true(len(selected_citations) == 1, 'Expected the only available action to be selected.') + assert_true( + selected_citations[0]['artifact_id'] == 'm2_artifact_1', + 'Expected the most recent available rows to be used.', + ) + assert_true( + select_prior_turn_action_citations([]) == [], + 'Expected an empty history to select nothing.', + ) + + def test_csv_guidance_states_the_publication_contract(): """CSV must receive the same publication contract JSON and XML already state.""" print('Testing CSV publication guidance parity...') @@ -435,6 +573,9 @@ def run_tests() -> bool: test_current_turn_rows_are_never_double_counted, test_answering_the_clarification_publishes_the_csv, test_pending_clarification_does_not_override_a_new_request, + test_row_count_estimate_survives_citation_compaction, + test_reach_back_prefers_the_dataset_over_newer_lookups, + test_reach_back_falls_back_to_the_only_rows_available, test_csv_guidance_states_the_publication_contract, ]