diff --git a/application/single_app/config.py b/application/single_app/config.py index 179e25894..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.006" +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 72a38c29b..42199ae80 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 @@ -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', @@ -195,6 +204,27 @@ 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 +# 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, ' + '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()) @@ -397,6 +427,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) @@ -422,6 +453,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'} @@ -436,7 +470,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 ' @@ -479,32 +518,52 @@ 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: - 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 = 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), 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: @@ -512,7 +571,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) @@ -530,8 +589,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 @@ -543,24 +626,198 @@ 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 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 + + +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: + """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: @@ -950,7 +1207,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..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, @@ -158,6 +159,8 @@ normalize_json_artifact_payload, normalize_generated_output_format, normalize_xml_artifact_payload, + resolve_pending_generated_file_format, + select_prior_turn_action_citations, serialize_generated_json, serialize_generated_xml, ) @@ -2257,6 +2260,116 @@ def _has_generated_tabular_csv_output(generated_outputs): return has_generated_tabular_csv_output(generated_outputs) +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 ' + '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 _read_agent_citation_artifact_payloads(conversation_id, artifact_ids): + if not artifact_ids: + return {} + 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, + )) + return build_message_artifact_payload_map(artifact_documents) + + +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: + return [] + + try: + _authorize_personal_conversation_access(normalized_user_id, normalized_conversation_id) + assistant_messages = _read_recent_assistant_messages( + normalized_conversation_id, + _resolve_prior_turn_history_window(settings), + ) + 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', + { + '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 +2386,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 +2403,12 @@ 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, + settings=get_settings(), + ), + 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 new file mode 100644 index 000000000..eba77f64d --- /dev/null +++ b/functional_tests/test_generated_csv_uses_authorized_action_rows.py @@ -0,0 +1,600 @@ +# 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.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 +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. + 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, a clarification turn publishes nothing, and an +answered clarification resumes the original request using the earlier rows. +""" + +import json +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, + 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' + +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}, +}] +# 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' +) + +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_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 _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...') + 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_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_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, + ] + + 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)