diff --git a/application/single_app/config.py b/application/single_app/config.py index bfafe596..20b79401 100644 --- a/application/single_app/config.py +++ b/application/single_app/config.py @@ -96,7 +96,7 @@ EXECUTOR_TYPE = 'thread' EXECUTOR_MAX_WORKERS = 30 SESSION_TYPE = 'filesystem' -VERSION = "0.250.136" +VERSION = "0.250.137" IS_DEVELOPMENT = is_development_env_enabled() SESSION_COOKIE_SAMESITE = os.getenv('SESSION_COOKIE_SAMESITE', 'Lax') diff --git a/application/single_app/functions_settings.py b/application/single_app/functions_settings.py index 4bdacffb..b31d277a 100644 --- a/application/single_app/functions_settings.py +++ b/application/single_app/functions_settings.py @@ -86,6 +86,18 @@ ADMIN_SETTINGS_NESTED_SECRET_FIELDS = ( "web_search_agent.other_settings.azure_ai_foundry.client_secret", ) +TABULAR_GENERATION_BACKEND_SETTING_KEYS = { + 'tabular_background_handoff_mode', + 'enable_tabular_generation_plan', + 'tabular_generation_plan_mode', + 'enable_tabular_compact_response_protocol', + 'enable_tabular_completion_driven_checkpointing', + 'enable_tabular_rolling_worker_pool', + 'enable_tabular_independent_batch_retries', + 'tabular_generation_checkpoint_writer_concurrency', + 'tabular_generation_heartbeat_seconds', + 'tabular_generation_systemic_failure_threshold', +} PUBLIC_WORKSPACE_DISPLAY_NAME_DEFAULT = "Public Workspace" PUBLIC_WORKSPACE_DISPLAY_NAME_PLURAL_DEFAULT = "Public Workspaces" PUBLIC_WORKSPACE_DISPLAY_NAME_MAX_LENGTH = 32 @@ -1004,6 +1016,16 @@ def get_settings(use_cosmos=False, include_source=False): 'tabular_generated_output_chunk_model_mode': 'current', 'tabular_generated_output_chunk_model_deployment': '', 'tabular_generated_output_model_validation_auto_retries': 3, + 'tabular_background_handoff_mode': 'legacy', + 'enable_tabular_generation_plan': False, + 'tabular_generation_plan_mode': 'off', + 'enable_tabular_compact_response_protocol': False, + 'enable_tabular_completion_driven_checkpointing': False, + 'enable_tabular_rolling_worker_pool': False, + 'enable_tabular_independent_batch_retries': False, + 'tabular_generation_checkpoint_writer_concurrency': 1, + 'tabular_generation_heartbeat_seconds': 30, + 'tabular_generation_systemic_failure_threshold': 0.5, 'enable_multi_agent_orchestration': False, 'enable_mixed_source_development_telemetry': False, 'enable_mixed_source_manifest': False, @@ -2771,6 +2793,8 @@ def sanitize_settings_for_user(full_settings: dict) -> dict: continue if k == 'agents_page_promoted_popular_agents': continue + if k in TABULAR_GENERATION_BACKEND_SETTING_KEYS: + continue if any(term in k.lower() for term in sensitive_terms): continue if k in ('model_endpoints', 'personal_model_endpoints') and isinstance(v, list): diff --git a/application/single_app/functions_tabular_generated_exports.py b/application/single_app/functions_tabular_generated_exports.py index b2cd4e07..53085c3e 100644 --- a/application/single_app/functions_tabular_generated_exports.py +++ b/application/single_app/functions_tabular_generated_exports.py @@ -53,6 +53,11 @@ TABULAR_EXPORT_RUN_TYPE = 'tabular_generated_output_run' TABULAR_EXPORT_CONTRACT_VERSION = 3 +TABULAR_GENERATION_CONTRACT_VERSION = 1 +TABULAR_RESPONSE_PROTOCOL_OBJECT_V1 = 'object-v1' +TABULAR_EXECUTOR_MODE_FIXED_WINDOW = 'fixed-window-v1' +TABULAR_ROLLOUT_PLANNER_MODES = {'off', 'shadow', 'active'} +TABULAR_ROLLOUT_HANDOFF_MODES = {'legacy', 'server', 'constrained_model'} TABULAR_RUN_TASK_STRUCTURED_EXPORT = 'structured_export' TABULAR_RUN_TASK_HIERARCHICAL_ANALYSIS = 'hierarchical_analysis' TABULAR_RUN_TASK_COMBINED = 'combined' @@ -149,6 +154,9 @@ TABULAR_EXPORT_SUMMARY_MAX_VALUES_PER_FIELD = 5 TABULAR_EXPORT_SUMMARY_AGGREGATE_MAX_VALUES = 25 TABULAR_EXPORT_PROGRESS_LOG_INTERVAL_SECONDS = 30 +TABULAR_GENERATION_DEFAULT_HEARTBEAT_SECONDS = 30 +TABULAR_GENERATION_DEFAULT_CHECKPOINT_WRITER_CONCURRENCY = 1 +TABULAR_GENERATION_DEFAULT_SYSTEMIC_FAILURE_THRESHOLD = 0.5 TABULAR_EXPORT_SCHEDULER_STATUSES = ( TABULAR_EXPORT_STATUS_QUEUED, TABULAR_EXPORT_STATUS_RUNNING, @@ -260,6 +268,188 @@ def _settings_float(settings, key, default, minimum=None, maximum=None): return parsed_value +def _settings_mode(settings, key, default, allowed_modes): + normalized_mode = str((settings or {}).get(key, default) or default).strip().lower() + if normalized_mode in allowed_modes: + return normalized_mode + return default + + +def _normalize_tabular_generation_rollout_settings(settings): + settings = settings or {} + return { + 'tabular_background_handoff_mode': _settings_mode( + settings, + 'tabular_background_handoff_mode', + 'legacy', + TABULAR_ROLLOUT_HANDOFF_MODES, + ), + 'tabular_generation_plan_mode': _settings_mode( + settings, + 'tabular_generation_plan_mode', + 'off', + TABULAR_ROLLOUT_PLANNER_MODES, + ), + 'enable_tabular_generation_plan': _settings_bool( + settings, + 'enable_tabular_generation_plan', + False, + ), + 'enable_tabular_compact_response_protocol': _settings_bool( + settings, + 'enable_tabular_compact_response_protocol', + False, + ), + 'enable_tabular_completion_driven_checkpointing': _settings_bool( + settings, + 'enable_tabular_completion_driven_checkpointing', + False, + ), + 'enable_tabular_rolling_worker_pool': _settings_bool( + settings, + 'enable_tabular_rolling_worker_pool', + False, + ), + 'enable_tabular_independent_batch_retries': _settings_bool( + settings, + 'enable_tabular_independent_batch_retries', + False, + ), + 'tabular_generation_checkpoint_writer_concurrency': _settings_int( + settings, + 'tabular_generation_checkpoint_writer_concurrency', + TABULAR_GENERATION_DEFAULT_CHECKPOINT_WRITER_CONCURRENCY, + minimum=1, + maximum=16, + ), + 'tabular_generation_heartbeat_seconds': _settings_int( + settings, + 'tabular_generation_heartbeat_seconds', + TABULAR_GENERATION_DEFAULT_HEARTBEAT_SECONDS, + minimum=5, + maximum=300, + ), + 'tabular_generation_systemic_failure_threshold': _settings_float( + settings, + 'tabular_generation_systemic_failure_threshold', + TABULAR_GENERATION_DEFAULT_SYSTEMIC_FAILURE_THRESHOLD, + minimum=0.0, + maximum=1.0, + ), + } + + +def _sync_tabular_generation_contract_fields(run): + if not isinstance(run, dict): + return run + + batch_count = _safe_int(run.get('batch_count')) + completed_batches = _safe_int(run.get('completed_batches')) + processed_rows = _safe_int(run.get('processed_rows')) + planned_batch_count = _safe_int(run.get('planned_batch_count')) + if planned_batch_count <= 0 and batch_count: + planned_batch_count = batch_count + + completed_batch_count = _safe_int(run.get('completed_batch_count')) + if completed_batch_count <= 0 and completed_batches: + completed_batch_count = completed_batches + + highest_contiguous_batch = _safe_int(run.get('highest_contiguous_batch')) + if highest_contiguous_batch <= 0 and completed_batches: + highest_contiguous_batch = completed_batches + + checkpointed_row_count = _safe_int(run.get('checkpointed_row_count')) + if checkpointed_row_count <= 0 and processed_rows: + checkpointed_row_count = processed_rows + + run.setdefault('generation_contract_version', TABULAR_GENERATION_CONTRACT_VERSION) + run.setdefault('response_protocol_version', TABULAR_RESPONSE_PROTOCOL_OBJECT_V1) + run.setdefault('executor_mode', TABULAR_EXECUTOR_MODE_FIXED_WINDOW) + run.setdefault('plan_blob_path', None) + run.setdefault('plan_hash', None) + run['planned_batch_count'] = planned_batch_count + run['completed_batch_count'] = completed_batch_count + run['highest_contiguous_batch'] = highest_contiguous_batch + run['active_batch_count'] = _safe_int(run.get('active_batch_count')) + run['retry_wait_batch_count'] = _safe_int(run.get('retry_wait_batch_count')) + run['exhausted_batch_count'] = _safe_int(run.get('exhausted_batch_count')) + run['checkpointed_row_count'] = checkpointed_row_count + run.setdefault('generation_started_at', run.get('started_at')) + run.setdefault('generation_completed_at', None) + return run + + +def _build_generation_progress_contract_fields(run, completed_batches, processed_rows): + batch_count = _safe_int((run or {}).get('batch_count')) + normalized_completed_batches = _safe_int(completed_batches) + normalized_processed_rows = _safe_int(processed_rows) + planned_batch_count = _safe_int((run or {}).get('planned_batch_count')) + if planned_batch_count <= 0 and batch_count: + planned_batch_count = batch_count + return { + 'planned_batch_count': planned_batch_count, + 'completed_batch_count': normalized_completed_batches, + 'highest_contiguous_batch': normalized_completed_batches, + 'active_batch_count': 0, + 'retry_wait_batch_count': _safe_int((run or {}).get('retry_wait_batch_count')), + 'exhausted_batch_count': _safe_int((run or {}).get('exhausted_batch_count')), + 'checkpointed_row_count': normalized_processed_rows, + } + + +def _extract_tabular_response_usage(result): + def read_usage_value(source, field_names): + if source is None: + return None + for field_name in field_names: + if isinstance(source, dict): + value = source.get(field_name) + else: + value = getattr(source, field_name, None) + parsed_value = _safe_int(value, default=0) + if parsed_value: + return parsed_value + return None + + first_message = result[0] if result else None + usage_sources = [] + for source in ( + getattr(first_message, 'metadata', None), + getattr(first_message, 'usage', None), + getattr(getattr(first_message, 'inner_content', None), 'usage', None), + ): + if source is not None: + usage_sources.append(source) + if isinstance(source, dict): + for nested_key in ('usage', 'token_usage', 'tokenUsage'): + nested_source = source.get(nested_key) + if nested_source is not None: + usage_sources.append(nested_source) + + usage = { + 'input_token_count': None, + 'output_token_count': None, + 'total_token_count': None, + } + for source in usage_sources: + if usage['input_token_count'] is None: + usage['input_token_count'] = read_usage_value( + source, + ('prompt_tokens', 'input_tokens', 'promptTokens', 'inputTokens'), + ) + if usage['output_token_count'] is None: + usage['output_token_count'] = read_usage_value( + source, + ('completion_tokens', 'output_tokens', 'completionTokens', 'outputTokens'), + ) + if usage['total_token_count'] is None: + usage['total_token_count'] = read_usage_value( + source, + ('total_tokens', 'totalTokens'), + ) + return usage + + def _resolve_tabular_batch_concurrency(settings, batch_count): configured_concurrency = (settings or {}).get('tabular_generated_output_batch_concurrency') if configured_concurrency not in (None, ''): @@ -1042,16 +1232,6 @@ def _parse_generated_json_object(response_content): return parsed_entries[0] -def _truncate_response_preview(response_content, max_chars=400): - cleaned = _clean_generated_json_code_fence(response_content) - normalized = re.sub(r'\s+', ' ', cleaned).strip() - if not normalized: - return '' - if len(normalized) <= max_chars: - return normalized - return f"{normalized[:max_chars]}..." - - def _dump_generated_output_json(value): return json.dumps(value, default=str, ensure_ascii=False, separators=(',', ':')) @@ -1531,7 +1711,7 @@ def _stage_tabular_generated_output_source(run, settings): def _migrate_legacy_tabular_export_run(run): if _safe_int(run.get('contract_version')) >= TABULAR_EXPORT_CONTRACT_VERSION: - return run + return _sync_tabular_generation_contract_fields(run) batch_count = _safe_int(run.get('batch_count')) expected_row_count = _safe_int(run.get('row_count')) @@ -1618,6 +1798,9 @@ def _migrate_legacy_tabular_export_run(run): now = _now_iso() run.update({ 'contract_version': TABULAR_EXPORT_CONTRACT_VERSION, + 'generation_contract_version': TABULAR_GENERATION_CONTRACT_VERSION, + 'response_protocol_version': TABULAR_RESPONSE_PROTOCOL_OBJECT_V1, + 'executor_mode': TABULAR_EXECUTOR_MODE_FIXED_WINDOW, 'task_type': _normalize_tabular_run_task_type(run.get('task_type')), 'analysis_objective': str(run.get('analysis_objective') or '').strip(), 'total_chunk_count': batch_count, @@ -1627,6 +1810,17 @@ def _migrate_legacy_tabular_export_run(run): 'input_blob_path': None, 'completed_batches': 0, 'processed_rows': 0, + 'planned_batch_count': batch_count, + 'completed_batch_count': 0, + 'highest_contiguous_batch': 0, + 'active_batch_count': 0, + 'retry_wait_batch_count': 0, + 'exhausted_batch_count': 0, + 'checkpointed_row_count': 0, + 'generation_started_at': run.get('started_at'), + 'generation_completed_at': None, + 'plan_blob_path': None, + 'plan_hash': None, 'output_schema': None, 'regenerate_legacy_output_checkpoints': False, 'updated_at': now, @@ -2014,6 +2208,15 @@ async def _generate_batch_entries( raw_response_content = '' mismatch_count = 0 last_validation_error = None + last_attempt_metrics = { + 'input_char_count': len(batch_prompt), + 'response_char_count': 0, + 'model_latency_seconds': None, + 'validation_seconds': None, + 'input_token_count': None, + 'output_token_count': None, + 'total_token_count': None, + } timeout_seconds = max( _safe_float( batch_timeout_seconds, @@ -2037,18 +2240,31 @@ async def _generate_batch_entries( execution_settings = AzureChatPromptExecutionSettings(service_id='tabular-generated-output-background') try: + model_started_at = time.monotonic() result = await asyncio.wait_for( chat_service.get_chat_message_contents(chat_history, execution_settings), timeout=timeout_seconds, ) + model_latency_seconds = time.monotonic() - model_started_at except asyncio.TimeoutError as exc: raise TimeoutError( f'Background structured export batch {batch_number}/{total_batches} ' f'timed out after {timeout_seconds:g} seconds.' ) from exc raw_response_content = result[0].content if result and result[0].content else '' + usage = _extract_tabular_response_usage(result) + validation_started_at = time.monotonic() parsed_entries = _parse_generated_json_entries(raw_response_content) if raw_response_content else None parsed_entry_count = len(parsed_entries) if parsed_entries is not None else 0 + last_attempt_metrics = { + 'input_char_count': len(batch_prompt), + 'response_char_count': len(raw_response_content), + 'model_latency_seconds': round(model_latency_seconds, 3), + 'validation_seconds': None, + 'input_token_count': usage.get('input_token_count'), + 'output_token_count': usage.get('output_token_count'), + 'total_token_count': usage.get('total_token_count'), + } if parsed_entries is not None and parsed_entry_count == len(batch_rows): try: normalized_entries, output_schema = _normalize_generated_batch_entries( @@ -2056,14 +2272,20 @@ async def _generate_batch_entries( parsed_entries, expected_output_schema=expected_output_schema, ) - return normalized_entries, mismatch_count, output_schema + last_attempt_metrics['validation_seconds'] = round( + time.monotonic() - validation_started_at, + 3, + ) + return normalized_entries, mismatch_count, output_schema, last_attempt_metrics except ValueError as exc: last_validation_error = str(exc) + last_attempt_metrics['validation_seconds'] = round(time.monotonic() - validation_started_at, 3) mismatch_count += 1 log_event( '[TABULAR_GENERATED_OUTPUT] Background export batch attempt mismatch', { + 'event_name': 'batch_validated', 'run_id': run_id, 'batch_number': batch_number, 'batch_count': total_batches, @@ -2072,7 +2294,11 @@ async def _generate_batch_entries( 'parsed_row_count': parsed_entry_count, 'validation_error': last_validation_error, 'response_char_count': len(raw_response_content), - 'response_preview': _truncate_response_preview(raw_response_content), + 'model_latency_seconds': last_attempt_metrics.get('model_latency_seconds'), + 'validation_seconds': last_attempt_metrics.get('validation_seconds'), + 'input_token_count': last_attempt_metrics.get('input_token_count'), + 'output_token_count': last_attempt_metrics.get('output_token_count'), + 'total_token_count': last_attempt_metrics.get('total_token_count'), }, debug_only=True, ) @@ -2099,9 +2325,11 @@ async def _generate_batch_entries_for_window( expected_output_schema, batch_timeout_seconds, ): + queued_at = time.monotonic() async with semaphore: + queue_wait_seconds = time.monotonic() - queued_at batch_started_at = time.monotonic() - batch_entries, mismatch_count, output_schema = await _generate_batch_entries( + batch_entries, mismatch_count, output_schema, attempt_metrics = await _generate_batch_entries( chat_service, user_question, batch_request['rows'], @@ -2114,12 +2342,42 @@ async def _generate_batch_entries_for_window( expected_output_schema=expected_output_schema, batch_timeout_seconds=batch_timeout_seconds, ) + elapsed_seconds = time.monotonic() - batch_started_at + log_event( + '[TABULAR_GENERATED_OUTPUT] Background export batch model completed', + { + 'event_name': 'batch_model_completed', + 'run_id': run_id, + 'batch_number': batch_request['batch_number'], + 'batch_count': total_batches, + 'row_count': len(batch_entries), + 'queue_wait_seconds': round(queue_wait_seconds, 3), + 'elapsed_seconds': round(elapsed_seconds, 3), + 'model_latency_seconds': attempt_metrics.get('model_latency_seconds'), + 'validation_seconds': attempt_metrics.get('validation_seconds'), + 'input_char_count': attempt_metrics.get('input_char_count'), + 'response_char_count': attempt_metrics.get('response_char_count'), + 'input_token_count': attempt_metrics.get('input_token_count'), + 'output_token_count': attempt_metrics.get('output_token_count'), + 'total_token_count': attempt_metrics.get('total_token_count'), + 'mismatch_count': mismatch_count, + }, + debug_only=True, + ) return { 'batch_number': batch_request['batch_number'], 'batch_entries': batch_entries, 'batch_summary': _build_generated_batch_summary(batch_entries), 'batch_row_count': len(batch_entries), - 'elapsed_seconds': time.monotonic() - batch_started_at, + 'elapsed_seconds': elapsed_seconds, + 'queue_wait_seconds': queue_wait_seconds, + 'model_latency_seconds': attempt_metrics.get('model_latency_seconds'), + 'validation_seconds': attempt_metrics.get('validation_seconds'), + 'input_char_count': attempt_metrics.get('input_char_count'), + 'response_char_count': attempt_metrics.get('response_char_count'), + 'input_token_count': attempt_metrics.get('input_token_count'), + 'output_token_count': attempt_metrics.get('output_token_count'), + 'total_token_count': attempt_metrics.get('total_token_count'), 'mismatch_count': mismatch_count, 'output_schema': output_schema, } @@ -2219,13 +2477,13 @@ async def _generate_analysis_chunk_summary( log_event( '[TABULAR_GENERATED_OUTPUT] Background analysis chunk attempt mismatch', { + 'event_name': 'batch_validated', 'run_id': run.get('id'), 'batch_number': batch_number, 'batch_count': total_batches, 'attempt_number': attempt_number, 'row_count': len(batch_rows), 'response_char_count': len(raw_response_content), - 'response_preview': _truncate_response_preview(raw_response_content), }, debug_only=True, ) @@ -2370,6 +2628,7 @@ async def _generate_combined_chunk_result( log_event( '[TABULAR_GENERATED_OUTPUT] Background combined chunk attempt mismatch', { + 'event_name': 'batch_validated', 'run_id': run.get('id'), 'batch_number': batch_number, 'batch_count': total_batches, @@ -2377,7 +2636,6 @@ async def _generate_combined_chunk_result( 'expected_row_count': len(batch_rows), 'validation_error': last_validation_error, 'response_char_count': len(raw_response_content), - 'response_preview': _truncate_response_preview(raw_response_content), }, debug_only=True, ) @@ -2512,6 +2770,7 @@ async def _generate_analysis_reduce_summary( log_event( '[TABULAR_GENERATED_OUTPUT] Background analysis reduce attempt mismatch', { + 'event_name': 'batch_validated', 'run_id': run.get('id'), 'level_number': level_number, 'node_number': node_number, @@ -2519,7 +2778,6 @@ async def _generate_analysis_reduce_summary( 'attempt_number': attempt_number, 'input_summary_count': len(summaries), 'response_char_count': len(raw_response_content), - 'response_preview': _truncate_response_preview(raw_response_content), }, debug_only=True, ) @@ -2958,6 +3216,7 @@ def _build_run_status_detail(run, settings, retryable_failure, can_resume): def _build_run_public_status(run, settings=None): if not isinstance(run, dict): return None + run = _sync_tabular_generation_contract_fields(run) batch_count = _safe_int(run.get('batch_count')) completed_batches = _safe_int(run.get('completed_batches')) @@ -3030,6 +3289,15 @@ def append_generated_artifact(artifact, fallback_file_name, fallback_output_form 'processed_rows': processed_rows, 'batch_count': batch_count, 'completed_batches': completed_batches, + 'generation_contract_version': _safe_int(run.get('generation_contract_version')), + 'response_protocol_version': run.get('response_protocol_version'), + 'planned_batch_count': _safe_int(run.get('planned_batch_count')), + 'completed_batch_count': _safe_int(run.get('completed_batch_count')), + 'highest_contiguous_batch': _safe_int(run.get('highest_contiguous_batch')), + 'active_batch_count': _safe_int(run.get('active_batch_count')), + 'retry_wait_batch_count': _safe_int(run.get('retry_wait_batch_count')), + 'exhausted_batch_count': _safe_int(run.get('exhausted_batch_count')), + 'checkpointed_row_count': _safe_int(run.get('checkpointed_row_count')), 'total_chunk_count': total_chunk_count, 'processed_chunk_count': processed_chunk_count, 'failed_chunk_count': failed_chunk_count, @@ -3041,6 +3309,8 @@ def append_generated_artifact(artifact, fallback_file_name, fallback_output_form 'progress_percent': progress_percent, 'created_at': run.get('created_at'), 'started_at': run.get('started_at'), + 'generation_started_at': run.get('generation_started_at'), + 'generation_completed_at': run.get('generation_completed_at'), 'updated_at': run.get('updated_at'), 'completed_at': run.get('completed_at'), 'last_heartbeat_at': run.get('last_heartbeat_at'), @@ -3424,9 +3694,11 @@ def _try_claim_run(user_id, run_id, settings): run.update({ 'status': TABULAR_EXPORT_STATUS_RUNNING, 'started_at': run.get('started_at') or now.isoformat(), + 'generation_started_at': run.get('generation_started_at') or now.isoformat(), 'attempt_started_at': now.isoformat(), 'updated_at': now.isoformat(), 'completed_at': None, + 'generation_completed_at': None, 'last_heartbeat_at': now.isoformat(), 'lease_holder_id': _lease_holder_id(), 'lease_generation': _safe_int(run.get('lease_generation')) + 1, @@ -3434,6 +3706,7 @@ def _try_claim_run(user_id, run_id, settings): 'next_attempt_at': None, 'last_message': 'Background structured export is running', }) + _sync_tabular_generation_contract_fields(run) try: return _replace_run(run) except Exception as exc: @@ -3454,6 +3727,7 @@ def _mark_run_failed(run, error_message): 'updated_at': now, 'completed_at': now, 'last_heartbeat_at': now, + 'active_batch_count': 0, 'last_error': str(error_message or 'Unknown error')[:1000], 'last_message': 'Background structured export failed', }) @@ -3525,6 +3799,7 @@ def _mark_run_retryable(run, error_message, settings, retry_category='transient' 'last_heartbeat_at': now.isoformat(), 'lease_holder_id': None, 'lease_expires_at': None, + 'active_batch_count': 0, 'last_error': str(error_message or 'Transient background export error')[:1000], 'last_message': ( 'Background structured export will retry after model-output validation failed' @@ -3658,6 +3933,7 @@ def _update_run_progress( 'last_message': f"Processed structured export batch {completed_batches} of {batch_count}", 'recent_batches': recent_batches, }) + run.update(_build_generation_progress_contract_fields(run, completed_batches, processed_rows)) run.update(throughput) return _replace_claimed_run(run) @@ -3837,6 +4113,7 @@ def _complete_run(run): 'status': TABULAR_EXPORT_STATUS_COMPLETED, 'updated_at': now, 'completed_at': now, + 'generation_completed_at': now, 'last_heartbeat_at': now, 'processed_rows': output_entry_count, 'completed_batches': _safe_int(run.get('batch_count')), @@ -3848,6 +4125,11 @@ def _complete_run(run): 'final_artifact': _build_artifact_metadata(uploaded_message, generated_file_name, output_format), 'estimated_remaining_seconds': 0, }) + run.update(_build_generation_progress_contract_fields( + run, + run.get('batch_count'), + output_entry_count, + )) run = _replace_claimed_run(run) log_event( '[TABULAR_GENERATED_OUTPUT] Background export completed', @@ -3859,6 +4141,10 @@ def _complete_run(run): 'output_format': output_format, 'row_count': output_entry_count, 'batch_count': run.get('batch_count'), + 'completed_batch_count': run.get('completed_batch_count'), + 'checkpointed_row_count': run.get('checkpointed_row_count'), + 'generation_contract_version': run.get('generation_contract_version'), + 'response_protocol_version': run.get('response_protocol_version'), 'artifact_message_id': uploaded_message.get('id'), 'generated_file_name': uploaded_message.get('file_name') or generated_file_name, }, @@ -3979,6 +4265,7 @@ def _complete_analysis_run(run, final_summary): 'status': TABULAR_EXPORT_STATUS_COMPLETED, 'updated_at': now, 'completed_at': now, + 'generation_completed_at': now, 'last_heartbeat_at': now, 'analysis_phase': 'completed', 'processed_rows': _safe_int(final_summary.get('row_count'), default=_safe_int(run.get('row_count'))), @@ -3992,6 +4279,11 @@ def _complete_analysis_run(run, final_summary): 'final_artifact': _build_artifact_metadata(uploaded_message, generated_file_name, 'md'), 'estimated_remaining_seconds': 0, }) + run.update(_build_generation_progress_contract_fields( + run, + run.get('batch_count'), + run.get('processed_rows'), + )) run = _replace_claimed_run(run) log_event( '[TABULAR_GENERATED_OUTPUT] Background tabular analysis completed', @@ -4072,6 +4364,7 @@ def _complete_combined_analysis_run(run, final_summary): 'status': TABULAR_EXPORT_STATUS_COMPLETED, 'updated_at': now, 'completed_at': now, + 'generation_completed_at': now, 'last_heartbeat_at': now, 'analysis_phase': 'completed', 'processed_rows': _safe_int(final_summary.get('row_count'), default=_safe_int(run.get('row_count'))), @@ -4089,6 +4382,11 @@ def _complete_combined_analysis_run(run, final_summary): 'final_artifact': structured_artifact or analysis_artifact, 'estimated_remaining_seconds': 0, }) + run.update(_build_generation_progress_contract_fields( + run, + run.get('batch_count'), + run.get('processed_rows'), + )) run = _replace_claimed_run(run) log_event( '[TABULAR_GENERATED_OUTPUT] Background combined tabular run completed', @@ -4338,6 +4636,7 @@ def _checkpoint_generated_batch_results(run, generated_results): run.get('id'), batch_number, ) + checkpoint_started_at = time.monotonic() try: _upload_json_blob( output_blob_path, @@ -4383,10 +4682,35 @@ def _checkpoint_generated_batch_results(run, generated_results): 'generated_output_summary': 'true', }, ) + checkpoint_seconds = time.monotonic() - checkpoint_started_at + log_event( + '[TABULAR_GENERATED_OUTPUT] Background export batch checkpointed', + { + 'event_name': 'batch_checkpointed', + 'run_id': run.get('id'), + 'conversation_id': run.get('conversation_id'), + 'user_id': run.get('user_id'), + 'batch_number': batch_number, + 'batch_row_count': generated_result['batch_row_count'], + 'checkpoint_seconds': round(checkpoint_seconds, 3), + 'response_protocol_version': run.get('response_protocol_version'), + 'plan_hash_present': bool(run.get('plan_hash')), + }, + debug_only=True, + ) batch_results[batch_number] = { 'batch_number': batch_number, 'batch_row_count': generated_result['batch_row_count'], 'elapsed_seconds': generated_result['elapsed_seconds'], + 'queue_wait_seconds': generated_result.get('queue_wait_seconds'), + 'model_latency_seconds': generated_result.get('model_latency_seconds'), + 'validation_seconds': generated_result.get('validation_seconds'), + 'input_char_count': generated_result.get('input_char_count'), + 'response_char_count': generated_result.get('response_char_count'), + 'input_token_count': generated_result.get('input_token_count'), + 'output_token_count': generated_result.get('output_token_count'), + 'total_token_count': generated_result.get('total_token_count'), + 'checkpoint_seconds': checkpoint_seconds, 'mismatch_count': generated_result['mismatch_count'], 'from_checkpoint': False, } @@ -4620,6 +4944,7 @@ def _update_analysis_map_progress( 'last_message': f'Analyzed tabular chunk {completed_batches} of {batch_count}', 'recent_batches': recent_batches, }) + run.update(_build_generation_progress_contract_fields(run, completed_batches, processed_rows)) run.update(throughput) return _replace_claimed_run(run) @@ -5178,6 +5503,7 @@ def queue_tabular_generated_output_run( task_type=normalized_task_type, user_question=user_question, ) + rollout_settings = _normalize_tabular_generation_rollout_settings(settings) if source_descriptor: staged_row_count = _safe_int(source_descriptor.get('expected_row_count')) @@ -5236,6 +5562,10 @@ def queue_tabular_generated_output_run( 'id': run_id, 'type': TABULAR_EXPORT_RUN_TYPE, 'contract_version': TABULAR_EXPORT_CONTRACT_VERSION, + 'generation_contract_version': TABULAR_GENERATION_CONTRACT_VERSION, + 'response_protocol_version': TABULAR_RESPONSE_PROTOCOL_OBJECT_V1, + 'executor_mode': TABULAR_EXECUTOR_MODE_FIXED_WINDOW, + 'generation_rollout_settings': rollout_settings, 'task_type': normalized_task_type, 'analysis_objective': normalized_analysis_objective, 'user_id': normalized_user_id, @@ -5262,6 +5592,17 @@ def queue_tabular_generated_output_run( 'failed_chunk_count': 0, 'chunk_manifest': chunk_manifest, 'completed_batches': 0, + 'planned_batch_count': staged_batch_count, + 'completed_batch_count': 0, + 'highest_contiguous_batch': 0, + 'active_batch_count': 0, + 'retry_wait_batch_count': 0, + 'exhausted_batch_count': 0, + 'checkpointed_row_count': 0, + 'generation_started_at': None, + 'generation_completed_at': None, + 'plan_blob_path': None, + 'plan_hash': None, 'processed_rows': 0, 'output_schema': None, 'source_descriptor': source_descriptor or None, @@ -5320,6 +5661,16 @@ def queue_tabular_generated_output_run( 'batch_input_token_budget': model_batch_budget.get('input_token_budget'), 'batch_output_token_budget': model_batch_budget.get('output_token_budget'), 'model_limit_source': model_batch_budget.get('limit_source'), + 'generation_contract_version': TABULAR_GENERATION_CONTRACT_VERSION, + 'response_protocol_version': TABULAR_RESPONSE_PROTOCOL_OBJECT_V1, + 'executor_mode': TABULAR_EXECUTOR_MODE_FIXED_WINDOW, + 'planner_mode': rollout_settings.get('tabular_generation_plan_mode'), + 'compact_protocol_enabled': rollout_settings.get('enable_tabular_compact_response_protocol'), + 'completion_checkpointing_enabled': rollout_settings.get( + 'enable_tabular_completion_driven_checkpointing' + ), + 'rolling_pool_enabled': rollout_settings.get('enable_tabular_rolling_worker_pool'), + 'independent_retries_enabled': rollout_settings.get('enable_tabular_independent_batch_retries'), 'source_backed': bool(source_descriptor), 'submitted_to_executor': submitted, }, diff --git a/docs/explanation/features/TABULAR_BACKGROUND_GENERATED_EXPORTS.md b/docs/explanation/features/TABULAR_BACKGROUND_GENERATED_EXPORTS.md index 97e2fba6..63c686d5 100644 --- a/docs/explanation/features/TABULAR_BACKGROUND_GENERATED_EXPORTS.md +++ b/docs/explanation/features/TABULAR_BACKGROUND_GENERATED_EXPORTS.md @@ -2,7 +2,7 @@ Implemented in version: **0.241.046** -Updated through version: **0.250.136** +Updated through version: **0.250.137** ## Overview @@ -32,6 +32,7 @@ The feature supports large spreadsheet-driven analysis, including workbooks that - Users can manually continue resumable failed or stale runs from the existing checkpoints without restarting completed batches. - Queued retry runs whose retry time has already passed are surfaced as resumable so deployments without active scheduler loops still give users a recovery action. - Run status includes safe user-facing status detail, checkpoint summaries, retry timing, heartbeat state, and continuation availability. +- Phase 1 acceleration groundwork adds additive generation contract fields, legacy-off rollout gates, safe batch latency/token telemetry, and deterministic fake model/storage harnesses without changing fixed-window execution behavior. ### API Endpoints @@ -50,8 +51,19 @@ The feature supports large spreadsheet-driven analysis, including workbooks that - `tabular_generated_output_input_token_soft_cap` - `tabular_generated_output_output_token_ratio` - `tabular_generated_output_output_expansion_ratio` +- `tabular_background_handoff_mode` +- `enable_tabular_generation_plan` +- `tabular_generation_plan_mode` +- `enable_tabular_compact_response_protocol` +- `enable_tabular_completion_driven_checkpointing` +- `enable_tabular_rolling_worker_pool` +- `enable_tabular_independent_batch_retries` +- `tabular_generation_checkpoint_writer_concurrency` +- `tabular_generation_heartbeat_seconds` +- `tabular_generation_systemic_failure_threshold` If a fixed concurrency is not configured, runs use up to 4, 16, 64, or 128 concurrent model calls according to the actual staged batch count. Model-aware source batching uses selected-model metadata, local `model_capabilities.json` token-limit fields when present, and bounded fallback limits otherwise. +The Phase 1 rollout settings default to legacy behavior and are copied into new run records for stable future rollouts. Backend-only rollout settings are filtered from sanitized non-admin frontend settings payloads. ### File Structure @@ -73,6 +85,7 @@ The progress card displays current status, completed checkpoint counts, processe - Functional regression: `functional_tests/test_tabular_background_generated_exports.py` - Scale and performance regression: `functional_tests/test_tabular_row_orchestration_scale.py` +- Phase 1 baseline and fake harness coverage: `functional_tests/test_tabular_row_orchestration_scale.py` - Functional regression for workflow/document-action presentation: `functional_tests/test_document_analysis_lossless_artifacts.py` - UI regression: `ui_tests/test_chat_background_generated_export_status.py` - Compile validation covers the modified Python modules. @@ -85,6 +98,7 @@ The progress card displays current status, completed checkpoint counts, processe - Adaptive concurrency uses up to 4 calls for small runs, 16 for medium runs, 64 for large runs, and 128 for runs with at least 256 staged batches. An explicit administrator setting overrides the adaptive tier. - Each parallel window checkpoints successful output batches before advancing public progress in contiguous order. - Progress is persisted once per completed parallel window. ETA uses recent wall-clock rows per minute rather than summing concurrent model-call durations as serial work. +- Phase 1 telemetry separates safe model-call, validation, and checkpoint timing metrics where the current executor can observe them. Validation mismatch logs record counts and timings only, not generated response previews. - Background processing writes each completed batch before moving on, allowing the run to resume after worker restarts. - The run status API returns compact metadata only, not source rows or generated batch content. - User-facing status details are derived from run metadata instead of displaying raw backend errors in the progress card. @@ -104,3 +118,4 @@ The progress card displays current status, completed checkpoint counts, processe - `application/single_app/config.py` was updated to version **0.241.060** for Phase 4 bounded batch concurrency. - `application/single_app/config.py` was updated to version **0.241.064** for generated export artifact presentation cleanup. - `application/single_app/config.py` was updated to version **0.250.136** for model-aware batch sizing, adaptive LLM concurrency, and parallel wall-clock ETA. +- `application/single_app/config.py` was updated to version **0.250.137** for Phase 1 acceleration baseline contracts, rollout controls, privacy-safe telemetry, and fake model/storage harnesses. diff --git a/docs/explanation/features/TABULAR_LLM_GENERATION_ACCELERATION_PHASE_1_BASELINE.md b/docs/explanation/features/TABULAR_LLM_GENERATION_ACCELERATION_PHASE_1_BASELINE.md new file mode 100644 index 00000000..f7c04def --- /dev/null +++ b/docs/explanation/features/TABULAR_LLM_GENERATION_ACCELERATION_PHASE_1_BASELINE.md @@ -0,0 +1,102 @@ +# Tabular LLM Generation Acceleration Phase 1 Baseline + +Implemented in version: **0.250.137** + +## Overview + +Phase 1 establishes additive run contracts, rollout controls, and safe observability for later tabular LLM generation acceleration work. It does not change the current fixed-window executor, object response protocol, retry behavior, checkpoint timing, or final artifact format. + +## Purpose + +The goal is to measure and compare future planner, compact protocol, completion-driven checkpoint, rolling scheduler, and independent retry changes without changing public semantics first. + +## Dependencies + +- Azure Cosmos DB container: `tabular_export_runs`, partitioned by `/user_id` +- Azure Blob Storage personal chat artifacts container +- Existing background tabular generated output runner in `functions_tabular_generated_exports.py` +- Existing settings sanitization in `functions_settings.py` +- Functional scale fixtures in `functional_tests/test_tabular_row_orchestration_scale.py` + +## Technical Specifications + +### Additive Run Fields + +New runs now carry compact aggregate fields in addition to the legacy `completed_batches` and `processed_rows` fields: + +- `generation_contract_version` +- `response_protocol_version` +- `plan_blob_path` +- `plan_hash` +- `planned_batch_count` +- `completed_batch_count` +- `highest_contiguous_batch` +- `active_batch_count` +- `retry_wait_batch_count` +- `exhausted_batch_count` +- `checkpointed_row_count` +- `generation_started_at` +- `generation_completed_at` + +Old run documents are defaulted in memory when loaded or migrated. No per-batch arrays are added to the Cosmos run document. +The existing export `contract_version` remains stable; Phase 1 uses `generation_contract_version` for the additive acceleration fields so current runs are not forced through legacy regeneration. + +### Rollout Controls + +The following settings default to legacy-off behavior: + +- `tabular_background_handoff_mode`: `legacy` +- `enable_tabular_generation_plan`: `False` +- `tabular_generation_plan_mode`: `off` +- `enable_tabular_compact_response_protocol`: `False` +- `enable_tabular_completion_driven_checkpointing`: `False` +- `enable_tabular_rolling_worker_pool`: `False` +- `enable_tabular_independent_batch_retries`: `False` +- `tabular_generation_checkpoint_writer_concurrency`: `1` +- `tabular_generation_heartbeat_seconds`: `30` +- `tabular_generation_systemic_failure_threshold`: `0.5` + +These are backend rollout controls. Sanitized user settings responses filter them unless a later UI phase intentionally exposes a safe subset. + +### Observability + +Existing `[TABULAR_GENERATED_OUTPUT]` telemetry now includes safe event markers and metrics for generated export batches: + +- `batch_validated` +- `batch_model_completed` +- `batch_checkpointed` + +Metrics include queue wait, model latency, validation time, checkpoint time, response character count, input character count, and provider token counts when available. Validation mismatch telemetry no longer logs generated response previews. + +## Baseline Status + +Local validation baseline for this implementation: + +| Check | Result | +| --- | --- | +| Python compile for changed backend and test modules | Passed | +| Focused tabular row orchestration scale suite | Passed | +| Deterministic fake model out-of-order completion harness | Passed | +| Fake storage success and injected upload failure harness | Passed | +| Sanitized settings source coverage for backend-only rollout keys | Passed | +| Existing custom logo sanitizer runtime test | Blocked locally by missing optional `olefile` dependency before reaching sanitizer assertions | +| Existing multi-endpoint sanitized notice test | Passed | + +Live 300-row, 3,000-row, and 30,000-row model baselines still need to be captured in an environment with configured model credentials and the production-like selected deployment. Those live measurements should record model latency, validation time, checkpoint time, finalization time, tokens or characters, retry rates, concurrency, rows per minute, and foreground handoff timing before Phase 2 or Phase 3 behavior rollout. + +## Testing and Validation + +- `python -m py_compile application\single_app\functions_tabular_generated_exports.py application\single_app\functions_settings.py functional_tests\test_tabular_row_orchestration_scale.py` +- `python functional_tests\test_tabular_row_orchestration_scale.py` +- `python functional_tests\test_custom_logo_sanitization_fix.py` (blocked locally by missing optional `olefile` dependency) +- `python functional_tests\test_chat_multi_endpoint_notice_template_fallback.py` + +## Known Limitations + +- This phase adds measurement and compatibility contracts only. It does not reduce elapsed generation time by itself. +- Live performance numbers are environment-dependent and were not generated by local fake harness tests. +- Later phases must continue to preserve old-run resume compatibility by honoring the recorded protocol and executor mode. + +## Related Version Updates + +- `application/single_app/config.py` was updated to version **0.250.137** for Phase 1 baseline contracts, rollout controls, privacy-safe telemetry, and deterministic fake harnesses. \ No newline at end of file diff --git a/functional_tests/test_tabular_row_orchestration_scale.py b/functional_tests/test_tabular_row_orchestration_scale.py index 1708fa55..3105683f 100644 --- a/functional_tests/test_tabular_row_orchestration_scale.py +++ b/functional_tests/test_tabular_row_orchestration_scale.py @@ -1,8 +1,8 @@ # test_tabular_row_orchestration_scale.py """ Functional test for scalable per-row tabular orchestration. -Version: 0.250.136 -Implemented in: 0.250.060; generated CSV formula safety in 0.250.065; generated file export routing in 0.250.072; source descriptor generalization in 0.250.127; unified durable run contract in 0.250.128; hierarchical analysis in 0.250.129; combined analysis and export in 0.250.130; scale validation in 0.250.132; direct source-backed exhaustive queueing in 0.250.133; direct queue call-site hardening in 0.250.134; model-validation auto retry in 0.250.135; model-aware parallel throughput in 0.250.136 +Version: 0.250.137 +Implemented in: 0.250.060; generated CSV formula safety in 0.250.065; generated file export routing in 0.250.072; source descriptor generalization in 0.250.127; unified durable run contract in 0.250.128; hierarchical analysis in 0.250.129; combined analysis and export in 0.250.130; scale validation in 0.250.132; direct source-backed exhaustive queueing in 0.250.133; direct queue call-site hardening in 0.250.134; model-validation auto retry in 0.250.135; model-aware parallel throughput in 0.250.136; Phase 1 acceleration contracts and observability in 0.250.137 This test ensures generated exports preserve source identity and row order while enforcing one stable output schema across independently generated batches. @@ -30,6 +30,7 @@ REPO_ROOT = Path(__file__).resolve().parents[1] APP_ROOT = REPO_ROOT / 'application' / 'single_app' EXPORT_MODULE = REPO_ROOT / 'application' / 'single_app' / 'functions_tabular_generated_exports.py' +SETTINGS_MODULE = REPO_ROOT / 'application' / 'single_app' / 'functions_settings.py' CHAT_ROUTE = REPO_ROOT / 'application' / 'single_app' / 'route_backend_chats.py' SIMPLECHAT_OPERATIONS = REPO_ROOT / 'application' / 'single_app' / 'functions_simplechat_operations.py' CSV_QUERY_MODULE = REPO_ROOT / 'application' / 'single_app' / 'functions_tabular_csv_query.py' @@ -111,6 +112,7 @@ '_normalize_source_identity_label', '_select_source_row_identity', '_prepare_tabular_source_rows', + '_sync_tabular_generation_contract_fields', '_migrate_legacy_tabular_export_run', } FAILURE_FUNCTIONS = { @@ -160,8 +162,14 @@ PERFORMANCE_FUNCTIONS = { '_safe_int', '_safe_float', + '_settings_bool', '_settings_int', '_settings_float', + '_settings_mode', + '_normalize_tabular_generation_rollout_settings', + '_sync_tabular_generation_contract_fields', + '_build_generation_progress_contract_fields', + '_extract_tabular_response_usage', '_resolve_tabular_batch_concurrency', '_normalize_tabular_run_task_type', '_resolve_tabular_chunk_model_selection', @@ -748,6 +756,9 @@ def safe_int(value, default=0, minimum=None, maximum=None): 'uuid': uuid, 'math': math, 'TABULAR_EXPORT_CONTRACT_VERSION': 3, + 'TABULAR_GENERATION_CONTRACT_VERSION': 1, + 'TABULAR_RESPONSE_PROTOCOL_OBJECT_V1': 'object-v1', + 'TABULAR_EXECUTOR_MODE_FIXED_WINDOW': 'fixed-window-v1', 'TABULAR_RUN_TASK_STRUCTURED_EXPORT': 'structured_export', 'TABULAR_RUN_TASK_HIERARCHICAL_ANALYSIS': 'hierarchical_analysis', 'TABULAR_RUN_TASK_COMBINED': 'combined', @@ -998,7 +1009,12 @@ def _load_performance_helpers(progress_updates=None): if isinstance(target, ast.Name) } if any( - name.startswith('TABULAR_EXPORT_') or name.startswith('TABULAR_RUN_TASK_') + name.startswith('TABULAR_EXPORT_') + or name.startswith('TABULAR_RUN_TASK_') + or name.startswith('TABULAR_GENERATION_') + or name.startswith('TABULAR_RESPONSE_') + or name.startswith('TABULAR_EXECUTOR_') + or name.startswith('TABULAR_ROLLOUT_') for name in assigned_names ): selected_nodes.append(node) @@ -1042,6 +1058,63 @@ def update_progress( return namespace +class FakeTabularModelHarness: + """Chat-service compatible fake model that releases calls in a caller-chosen order.""" + + def __init__(self, responses): + self.responses = list(responses) + self.calls = [] + self.completed_batches = [] + self._release_events = {} + + async def get_chat_message_contents(self, chat_history, execution_settings): + del chat_history, execution_settings + call_index = len(self.calls) + response = dict(self.responses[call_index]) + batch_number = response.get('batch_number') + release_event = asyncio.Event() + self.calls.append(batch_number) + self._release_events[batch_number] = release_event + await release_event.wait() + if response.get('exception'): + raise response['exception'] + self.completed_batches.append(batch_number) + usage = response.get('usage') or {} + return [SimpleNamespace( + content=response.get('content', '[]'), + metadata={'usage': usage}, + )] + + def release_batch(self, batch_number): + self._release_events[batch_number].set() + + +class FakeTabularStorageHarness: + """In-memory JSON blob harness with injectable upload and download failures.""" + + def __init__(self): + self.blobs = {} + self.metadata = {} + self.upload_failures = set() + self.download_failures = set() + + def upload_json_blob(self, path, payload, metadata=None, overwrite=True): + if path in self.upload_failures: + raise RuntimeError(f'Injected upload failure for {path}') + if not overwrite and path in self.blobs: + raise FileExistsError(path) + self.blobs[path] = payload + self.metadata[path] = dict(metadata or {}) + + def download_json_blob(self, path): + if path in self.download_failures: + raise RuntimeError(f'Injected download failure for {path}') + return self.blobs[path] + + def blob_exists(self, path): + return path in self.blobs + + def test_model_aware_batch_budget_uses_safe_token_limits(): """Batch planning uses output limits and caps very large input contexts.""" helpers = _load_performance_helpers() @@ -1162,6 +1235,169 @@ def test_dynamic_concurrency_and_parallel_window_eta(): }] +def test_phase_one_generation_contract_fields_are_additive_and_compact(): + """Phase 1 mirrors legacy progress fields without per-batch Cosmos arrays.""" + helpers = _load_performance_helpers() + sync_fields = helpers['_sync_tabular_generation_contract_fields'] + progress_fields = helpers['_build_generation_progress_contract_fields'] + + old_run = { + 'id': 'old-run', + 'batch_count': 909, + 'completed_batches': 3, + 'processed_rows': 99, + 'started_at': '2026-08-09T00:00:00+00:00', + } + synced_run = sync_fields(old_run) + + assert synced_run['generation_contract_version'] == 1 + assert synced_run['response_protocol_version'] == 'object-v1' + assert synced_run['executor_mode'] == 'fixed-window-v1' + assert synced_run['planned_batch_count'] == 909 + assert synced_run['completed_batch_count'] == 3 + assert synced_run['highest_contiguous_batch'] == 3 + assert synced_run['checkpointed_row_count'] == 99 + assert synced_run['plan_blob_path'] is None + assert synced_run['plan_hash'] is None + assert 'completed_batch_list' not in synced_run + + fields = progress_fields({'batch_count': 909}, completed_batches=7, processed_rows=231) + assert fields == { + 'planned_batch_count': 909, + 'completed_batch_count': 7, + 'highest_contiguous_batch': 7, + 'active_batch_count': 0, + 'retry_wait_batch_count': 0, + 'exhausted_batch_count': 0, + 'checkpointed_row_count': 231, + } + serialized_run_document = json.dumps(synced_run, separators=(',', ':'), ensure_ascii=False).encode('utf-8') + assert len(serialized_run_document) < 2048 + + +def test_phase_one_rollout_gates_default_to_legacy_and_stay_backend_only(): + """Later-phase rollout gates default off and are filtered from user settings.""" + helpers = _load_performance_helpers() + normalize_rollout = helpers['_normalize_tabular_generation_rollout_settings'] + + defaults = normalize_rollout({}) + assert defaults == { + 'tabular_background_handoff_mode': 'legacy', + 'tabular_generation_plan_mode': 'off', + 'enable_tabular_generation_plan': False, + 'enable_tabular_compact_response_protocol': False, + 'enable_tabular_completion_driven_checkpointing': False, + 'enable_tabular_rolling_worker_pool': False, + 'enable_tabular_independent_batch_retries': False, + 'tabular_generation_checkpoint_writer_concurrency': 1, + 'tabular_generation_heartbeat_seconds': 30, + 'tabular_generation_systemic_failure_threshold': 0.5, + } + overridden = normalize_rollout({ + 'tabular_background_handoff_mode': 'server', + 'tabular_generation_plan_mode': 'shadow', + 'enable_tabular_generation_plan': 'true', + 'enable_tabular_compact_response_protocol': 'yes', + 'enable_tabular_completion_driven_checkpointing': '1', + 'enable_tabular_rolling_worker_pool': 'on', + 'enable_tabular_independent_batch_retries': True, + 'tabular_generation_checkpoint_writer_concurrency': 99, + 'tabular_generation_heartbeat_seconds': 1, + 'tabular_generation_systemic_failure_threshold': 2, + }) + assert overridden['tabular_background_handoff_mode'] == 'server' + assert overridden['tabular_generation_plan_mode'] == 'shadow' + assert overridden['enable_tabular_generation_plan'] is True + assert overridden['enable_tabular_compact_response_protocol'] is True + assert overridden['enable_tabular_completion_driven_checkpointing'] is True + assert overridden['enable_tabular_rolling_worker_pool'] is True + assert overridden['enable_tabular_independent_batch_retries'] is True + assert overridden['tabular_generation_checkpoint_writer_concurrency'] == 16 + assert overridden['tabular_generation_heartbeat_seconds'] == 5 + assert overridden['tabular_generation_systemic_failure_threshold'] == 1.0 + + settings_source = SETTINGS_MODULE.read_text(encoding='utf-8') + for setting_key in defaults: + assert f"'{setting_key}'" in settings_source + assert 'TABULAR_GENERATION_BACKEND_SETTING_KEYS' in settings_source + assert 'if k in TABULAR_GENERATION_BACKEND_SETTING_KEYS' in settings_source + + +def test_phase_one_observability_uses_safe_metrics_not_response_content(): + """Telemetry records usage counts and excludes generated response previews.""" + helpers = _load_performance_helpers() + usage = helpers['_extract_tabular_response_usage']([ + SimpleNamespace(metadata={ + 'usage': { + 'prompt_tokens': 123, + 'completion_tokens': 45, + 'total_tokens': 168, + }, + }), + ]) + + assert usage == { + 'input_token_count': 123, + 'output_token_count': 45, + 'total_token_count': 168, + } + export_source = EXPORT_MODULE.read_text(encoding='utf-8') + assert 'response_preview' not in export_source + assert 'batch_model_completed' in export_source + assert 'batch_checkpointed' in export_source + assert 'model_latency_seconds' in export_source + assert 'checkpoint_seconds' in export_source + + +def test_phase_one_fake_harnesses_control_completion_order_and_storage_failures(): + """Reusable fakes let later phases force stragglers, usage counts, and blob failures.""" + async def complete_out_of_order(): + model = FakeTabularModelHarness([ + { + 'batch_number': 1, + 'content': '[{"answer":"one"}]', + 'usage': {'prompt_tokens': 10, 'completion_tokens': 3}, + }, + {'batch_number': 2, 'content': '[{"answer":"two"}]'}, + {'batch_number': 3, 'content': '[{"answer":"three"}]'}, + ]) + tasks = [ + asyncio.create_task(model.get_chat_message_contents(None, None)) + for _ in range(3) + ] + await asyncio.sleep(0) + model.release_batch(2) + await asyncio.sleep(0) + model.release_batch(3) + await asyncio.sleep(0) + model.release_batch(1) + results = await asyncio.gather(*tasks) + return model, results + + model, results = asyncio.run(complete_out_of_order()) + assert model.calls == [1, 2, 3] + assert model.completed_batches == [2, 3, 1] + assert results[0][0].metadata['usage']['prompt_tokens'] == 10 + + storage = FakeTabularStorageHarness() + storage.upload_json_blob( + 'output/batch_000002.json', + [{'source_row_number': 2, 'answer': 'two'}], + metadata={'batch_number': 2}, + overwrite=False, + ) + assert storage.blob_exists('output/batch_000002.json') is True + assert storage.download_json_blob('output/batch_000002.json')[0]['answer'] == 'two' + assert storage.metadata['output/batch_000002.json']['batch_number'] == 2 + storage.upload_failures.add('output/batch_000003.json') + try: + storage.upload_json_blob('output/batch_000003.json', [], overwrite=False) + except RuntimeError as exc: + assert 'Injected upload failure' in str(exc) + else: + raise AssertionError('Injected storage failures must be observable') + + def test_source_identity_and_order_contract(): """Every row receives a canonical ordinal and preserves its source identifier.""" helpers = _load_contract_helpers() @@ -2628,6 +2864,9 @@ def test_legacy_run_migration_tokenizes_inputs_and_resets_outputs(): manifest_page_path = 'user-1/conversation-1/generated/tabular_runs/legacy-run/manifest/chunks/page_000001.json' assert migrated_run['contract_version'] == 3 + assert migrated_run['generation_contract_version'] == 1 + assert migrated_run['response_protocol_version'] == 'object-v1' + assert migrated_run['executor_mode'] == 'fixed-window-v1' assert migrated_run['task_type'] == 'structured_export' assert migrated_run['analysis_objective'] == '' assert migrated_run['total_chunk_count'] == 2 @@ -2635,7 +2874,13 @@ def test_legacy_run_migration_tokenizes_inputs_and_resets_outputs(): assert migrated_run['failed_chunk_count'] == 0 assert migrated_run['chunk_manifest']['page_count'] == 1 assert migrated_run['completed_batches'] == 0 + assert migrated_run['planned_batch_count'] == 2 + assert migrated_run['completed_batch_count'] == 0 + assert migrated_run['highest_contiguous_batch'] == 0 + assert migrated_run['checkpointed_row_count'] == 0 assert migrated_run['processed_rows'] == 0 + assert migrated_run['plan_blob_path'] is None + assert migrated_run['plan_hash'] is None assert migrated_run['output_schema'] is None assert migrated_run['regenerate_legacy_output_checkpoints'] is False assert migrated_run['input_blob_path'] is None @@ -2793,6 +3038,10 @@ def main(): tests = [ test_model_aware_batch_budget_uses_safe_token_limits, test_dynamic_concurrency_and_parallel_window_eta, + test_phase_one_generation_contract_fields_are_additive_and_compact, + test_phase_one_rollout_gates_default_to_legacy_and_stay_backend_only, + test_phase_one_observability_uses_safe_metrics_not_response_content, + test_phase_one_fake_harnesses_control_completion_order_and_storage_failures, test_source_identity_and_order_contract, test_generated_batch_schema_contract, test_durable_runner_enforces_row_contract,