diff --git a/application/single_app/config.py b/application/single_app/config.py index 4efbe596..bfafe596 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.133" +VERSION = "0.250.136" 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 3a15a4e7..4bdacffb 100644 --- a/application/single_app/functions_settings.py +++ b/application/single_app/functions_settings.py @@ -1003,6 +1003,7 @@ def get_settings(use_cosmos=False, include_source=False): 'tabular_durable_run_confirmation_threshold_batches': 75, 'tabular_generated_output_chunk_model_mode': 'current', 'tabular_generated_output_chunk_model_deployment': '', + 'tabular_generated_output_model_validation_auto_retries': 3, 'enable_multi_agent_orchestration': False, 'enable_mixed_source_development_telemetry': False, 'enable_mixed_source_manifest': False, diff --git a/application/single_app/functions_tabular_generated_exports.py b/application/single_app/functions_tabular_generated_exports.py index bc332ab1..b2cd4e07 100644 --- a/application/single_app/functions_tabular_generated_exports.py +++ b/application/single_app/functions_tabular_generated_exports.py @@ -77,17 +77,69 @@ TABULAR_EXPORT_DEFAULT_INLINE_MAX_BATCHES = 75 TABULAR_EXPORT_DEFAULT_INLINE_MAX_ROWS = 500 TABULAR_EXPORT_DEFAULT_BATCH_RETRY_ATTEMPTS = 2 +TABULAR_EXPORT_DEFAULT_MODEL_VALIDATION_AUTO_RETRIES = 3 TABULAR_EXPORT_DEFAULT_LEASE_SECONDS = 300 -TABULAR_EXPORT_DEFAULT_STALE_SECONDS = 420 +TABULAR_EXPORT_DEFAULT_STALE_SECONDS = 900 TABULAR_EXPORT_DEFAULT_SCAN_LIMIT = 5 TABULAR_EXPORT_DEFAULT_MAX_TRANSIENT_FAILURES = 20 -TABULAR_EXPORT_DEFAULT_BATCH_CONCURRENCY = 3 -TABULAR_EXPORT_MAX_BATCH_CONCURRENCY = 5 +TABULAR_EXPORT_DEFAULT_BATCH_CONCURRENCY = 16 +TABULAR_EXPORT_HIGH_BATCH_CONCURRENCY = 64 +TABULAR_EXPORT_MAX_BATCH_CONCURRENCY = 128 +TABULAR_EXPORT_HIGH_CONCURRENCY_BATCH_THRESHOLD = 128 +TABULAR_EXPORT_MAX_CONCURRENCY_BATCH_THRESHOLD = 256 TABULAR_EXPORT_DEFAULT_BATCH_TIMEOUT_SECONDS = 300 TABULAR_EXPORT_FINAL_SPOOL_MAX_MEMORY_BYTES = 1024 * 1024 TABULAR_EXPORT_DEFAULT_SOURCE_CHUNK_ROWS = 1000 TABULAR_EXPORT_DEFAULT_SOURCE_BATCH_ROWS = 50 TABULAR_EXPORT_DEFAULT_SOURCE_BATCH_CHARS = 60000 +TABULAR_EXPORT_MAX_SOURCE_BATCH_ROWS = 500 +TABULAR_EXPORT_MAX_SOURCE_BATCH_CHARS = 720000 +TABULAR_EXPORT_DEFAULT_CONTEXT_TOKEN_LIMIT = 128000 +TABULAR_EXPORT_DEFAULT_OUTPUT_TOKEN_LIMIT = 65536 +TABULAR_EXPORT_DEFAULT_INPUT_TOKEN_RATIO = 0.5 +TABULAR_EXPORT_LARGE_CONTEXT_INPUT_TOKEN_RATIO = 0.3 +TABULAR_EXPORT_DEFAULT_OUTPUT_TOKEN_RATIO = 0.6 +TABULAR_EXPORT_LARGE_CONTEXT_TOKEN_THRESHOLD = 500000 +TABULAR_EXPORT_INPUT_TOKEN_SOFT_CAP = 180000 +TABULAR_EXPORT_PROMPT_TOKEN_RESERVE = 4096 +TABULAR_EXPORT_APPROXIMATE_CHARS_PER_TOKEN = 4.0 +TABULAR_EXPORT_DEFAULT_OUTPUT_EXPANSION_RATIO = 1.5 +TABULAR_EXPORT_MODEL_CONTEXT_LIMIT_FIELDS = ( + 'inputTokenLimit', + 'input_token_limit', + 'maxInputTokens', + 'max_input_tokens', + 'contextWindow', + 'context_window', + 'maxContextTokens', + 'max_context_tokens', + 'contextLength', + 'context_length', +) +TABULAR_EXPORT_MODEL_OUTPUT_LIMIT_FIELDS = ( + 'outputTokenLimit', + 'output_token_limit', + 'maxOutputTokens', + 'max_output_tokens', + 'responseLength', + 'response_length', + 'maxCompletionTokens', + 'max_completion_tokens', + 'maxTokens', + 'max_tokens', +) +TABULAR_EXPORT_MODEL_LIMIT_CONTAINER_FIELDS = ('tokenLimits', 'token_limits', 'limits') +TABULAR_EXPORT_MODEL_IDENTIFIER_FIELDS = ( + 'id', + 'modelId', + 'model_id', + 'model_deployment', + 'modelName', + 'model_name', + 'deploymentName', + 'deployment', + 'name', +) TABULAR_ANALYSIS_DEFAULT_REDUCE_FAN_IN = 25 TABULAR_ANALYSIS_MAX_REDUCE_FAN_IN = 50 TABULAR_ANALYSIS_SUMMARY_MAX_CHARS = 24000 @@ -132,6 +184,19 @@ 'worker exiting', 'worker restart', ) +TABULAR_EXPORT_MODEL_VALIDATION_RETRYABLE_MESSAGE_MARKERS = ( + 'failed validation', + 'schema mismatch', + 'schema drift', + 'source row token mismatch', + 'did not return the required', + 'returned no content after tool errors', + 'returned no content after workbook tool errors', + 'returned no content', + 'response did not contain valid structured_rows', + 'was not a valid compact json analysis summary', + 'was not a valid json object', +) TABULAR_EXPORT_INPUT_ROW_NUMBER_FIELD = '__simplechat_source_row_number' TABULAR_EXPORT_INPUT_ROW_IDENTITY_FIELD = '__simplechat_source_row_identity' TABULAR_EXPORT_INPUT_ROW_TOKEN_FIELD = '__simplechat_source_row_token' @@ -186,6 +251,37 @@ def _settings_int(settings, key, default, minimum=None, maximum=None): return _safe_int((settings or {}).get(key, default), default=default, minimum=minimum, maximum=maximum) +def _settings_float(settings, key, default, minimum=None, maximum=None): + parsed_value = _safe_float((settings or {}).get(key, default), default=default) + if minimum is not None: + parsed_value = max(minimum, parsed_value) + if maximum is not None: + parsed_value = min(maximum, parsed_value) + return parsed_value + + +def _resolve_tabular_batch_concurrency(settings, batch_count): + configured_concurrency = (settings or {}).get('tabular_generated_output_batch_concurrency') + if configured_concurrency not in (None, ''): + return _safe_int( + configured_concurrency, + default=TABULAR_EXPORT_DEFAULT_BATCH_CONCURRENCY, + minimum=1, + maximum=TABULAR_EXPORT_MAX_BATCH_CONCURRENCY, + ) + + normalized_batch_count = _safe_int(batch_count, minimum=1) + if normalized_batch_count <= 4: + return normalized_batch_count + if normalized_batch_count < TABULAR_EXPORT_DEFAULT_BATCH_CONCURRENCY: + return 4 + if normalized_batch_count < TABULAR_EXPORT_HIGH_CONCURRENCY_BATCH_THRESHOLD: + return TABULAR_EXPORT_DEFAULT_BATCH_CONCURRENCY + if normalized_batch_count < TABULAR_EXPORT_MAX_CONCURRENCY_BATCH_THRESHOLD: + return TABULAR_EXPORT_HIGH_BATCH_CONCURRENCY + return TABULAR_EXPORT_MAX_BATCH_CONCURRENCY + + def _normalize_tabular_run_task_type(task_type): normalized_task_type = str(task_type or '').strip().lower() if normalized_task_type in TABULAR_RUN_TASK_TYPES: @@ -242,6 +338,14 @@ def _is_retryable_export_error_message(error_message): return any(marker in normalized_message for marker in TABULAR_EXPORT_RETRYABLE_MESSAGE_MARKERS) +def _is_retryable_model_validation_error_message(error_message): + normalized_message = str(error_message or '').lower() + return any( + marker in normalized_message + for marker in TABULAR_EXPORT_MODEL_VALIDATION_RETRYABLE_MESSAGE_MARKERS + ) + + def _is_retryable_export_error(exc): status_code = _exception_status_code(exc) if status_code in TABULAR_EXPORT_RETRYABLE_STATUS_CODES: @@ -256,6 +360,13 @@ def _is_retryable_export_error(exc): return _is_retryable_export_error_message(exc) +def _is_retryable_model_validation_error(exc): + for candidate in _iter_exception_chain(exc): + if _is_retryable_model_validation_error_message(candidate): + return True + return _is_retryable_model_validation_error_message(exc) + + def _sanitize_file_base_name(file_name): base_name = os.path.splitext(str(file_name or '').strip())[0] normalized_base_name = re.sub(r'[^A-Za-z0-9._-]+', '_', base_name).strip('._') @@ -1305,13 +1416,13 @@ def _stage_tabular_generated_output_source(run, settings): source_descriptor.get('batch_max_rows'), default=TABULAR_EXPORT_DEFAULT_SOURCE_BATCH_ROWS, minimum=1, - maximum=100, + maximum=TABULAR_EXPORT_MAX_SOURCE_BATCH_ROWS, ) max_batch_chars = _safe_int( source_descriptor.get('batch_max_chars'), default=TABULAR_EXPORT_DEFAULT_SOURCE_BATCH_CHARS, minimum=6000, - maximum=120000, + maximum=TABULAR_EXPORT_MAX_SOURCE_BATCH_CHARS, ) source_blob_client = _get_versioned_source_blob_client(source_descriptor) resume_source_row = _safe_int(run.get('source_scan_row_count')) @@ -1592,6 +1703,274 @@ def _resolve_tabular_chunk_model_selection(gpt_model, settings, model_context=No return configured_deployment, {} +def _normalize_tabular_model_identifier(value): + return re.sub(r'[^a-z0-9]+', '-', str(value or '').strip().lower()).strip('-') + + +def _get_tabular_model_record_identifiers(model_record): + if not isinstance(model_record, dict): + return set() + + identifiers = { + _normalize_tabular_model_identifier(model_record.get(field_name)) + for field_name in TABULAR_EXPORT_MODEL_IDENTIFIER_FIELDS + if model_record.get(field_name) + } + for alias in model_record.get('aliases') or []: + normalized_alias = _normalize_tabular_model_identifier(alias) + if normalized_alias: + identifiers.add(normalized_alias) + return {identifier for identifier in identifiers if identifier} + + +def _read_tabular_model_token_limit(model_record, field_names): + if not isinstance(model_record, dict): + return None + + containers = [model_record] + containers.extend( + model_record.get(container_name) + for container_name in TABULAR_EXPORT_MODEL_LIMIT_CONTAINER_FIELDS + if isinstance(model_record.get(container_name), dict) + ) + for container in containers: + for field_name in field_names: + value = _safe_int(container.get(field_name)) + if value > 0: + return value + return None + + +def _iter_configured_tabular_model_records(settings): + settings = settings or {} + gpt_model_settings = settings.get('gpt_model') + if isinstance(gpt_model_settings, dict): + for model_record in gpt_model_settings.get('selected') or []: + if isinstance(model_record, dict): + yield model_record + + for endpoint in settings.get('model_endpoints') or []: + if not isinstance(endpoint, dict): + continue + for model_record in endpoint.get('models') or []: + if isinstance(model_record, dict): + yield model_record + + +def _load_tabular_model_limit_catalog(): + catalog_path = os.path.join( + os.path.dirname(__file__), + 'static', + 'json', + 'model_capabilities.json', + ) + try: + with open(catalog_path, 'r', encoding='utf-8') as catalog_file: + catalog = json.load(catalog_file) + except (OSError, json.JSONDecodeError): + return [] + return [ + model_record + for model_record in catalog.get('models') or [] + if isinstance(model_record, dict) + ] if isinstance(catalog, dict) else [] + + +def _resolve_tabular_model_token_limits(gpt_model, settings, model_context=None, catalog_records=None): + chunk_gpt_model, chunk_model_context = _resolve_tabular_chunk_model_selection( + gpt_model, + settings, + model_context=model_context, + ) + chunk_model_context = chunk_model_context if isinstance(chunk_model_context, dict) else {} + requested_identifiers = { + _normalize_tabular_model_identifier(identifier) + for identifier in ( + chunk_gpt_model, + chunk_model_context.get('model_id'), + chunk_model_context.get('model_deployment'), + ) + if identifier + } + candidate_groups = [ + ('context', [chunk_model_context]), + ('configured', list(_iter_configured_tabular_model_records(settings))), + ( + 'catalog', + list(catalog_records) if catalog_records is not None else _load_tabular_model_limit_catalog(), + ), + ] + context_token_limit = None + output_token_limit = None + limit_sources = [] + for source_name, model_records in candidate_groups: + for model_record in model_records: + if not isinstance(model_record, dict): + continue + record_identifiers = _get_tabular_model_record_identifiers(model_record) + if requested_identifiers and not requested_identifiers.intersection(record_identifiers): + continue + requested_identifiers.update(record_identifiers) + prior_context_token_limit = context_token_limit + prior_output_token_limit = output_token_limit + if context_token_limit is None: + context_token_limit = _read_tabular_model_token_limit( + model_record, + TABULAR_EXPORT_MODEL_CONTEXT_LIMIT_FIELDS, + ) + if output_token_limit is None: + output_token_limit = _read_tabular_model_token_limit( + model_record, + TABULAR_EXPORT_MODEL_OUTPUT_LIMIT_FIELDS, + ) + supplied_limit = ( + context_token_limit != prior_context_token_limit + or output_token_limit != prior_output_token_limit + ) + if supplied_limit and source_name not in limit_sources: + limit_sources.append(source_name) + if context_token_limit and output_token_limit: + break + if context_token_limit and output_token_limit: + break + + return { + 'model': chunk_gpt_model, + 'context_token_limit': context_token_limit or TABULAR_EXPORT_DEFAULT_CONTEXT_TOKEN_LIMIT, + 'output_token_limit': output_token_limit or TABULAR_EXPORT_DEFAULT_OUTPUT_TOKEN_LIMIT, + 'source': '+'.join(limit_sources) if limit_sources else 'fallback', + } + + +def _build_model_aware_source_batch_budget( + gpt_model, + settings, + model_context=None, + task_type=TABULAR_RUN_TASK_STRUCTURED_EXPORT, + user_question=None, + catalog_records=None, +): + settings = settings or {} + token_limits = _resolve_tabular_model_token_limits( + gpt_model, + settings, + model_context=model_context, + catalog_records=catalog_records, + ) + context_token_limit = _safe_int(token_limits.get('context_token_limit'), minimum=1) + output_token_limit = _safe_int(token_limits.get('output_token_limit'), minimum=1) + input_ratio = _settings_float( + settings, + 'tabular_generated_output_input_token_ratio', + TABULAR_EXPORT_DEFAULT_INPUT_TOKEN_RATIO, + minimum=0.1, + maximum=0.8, + ) + if context_token_limit > TABULAR_EXPORT_LARGE_CONTEXT_TOKEN_THRESHOLD: + input_ratio = min( + input_ratio, + _settings_float( + settings, + 'tabular_generated_output_large_context_input_token_ratio', + TABULAR_EXPORT_LARGE_CONTEXT_INPUT_TOKEN_RATIO, + minimum=0.1, + maximum=0.5, + ), + ) + input_token_budget = int(context_token_limit * input_ratio) + if context_token_limit > TABULAR_EXPORT_LARGE_CONTEXT_TOKEN_THRESHOLD: + input_token_budget = min( + input_token_budget, + _settings_int( + settings, + 'tabular_generated_output_input_token_soft_cap', + TABULAR_EXPORT_INPUT_TOKEN_SOFT_CAP, + minimum=16000, + maximum=400000, + ), + ) + question_token_reserve = math.ceil(len(str(user_question or '')) / TABULAR_EXPORT_APPROXIMATE_CHARS_PER_TOKEN) + input_token_budget = max( + input_token_budget - TABULAR_EXPORT_PROMPT_TOKEN_RESERVE - question_token_reserve, + 1500, + ) + output_token_budget = max( + int(output_token_limit * _settings_float( + settings, + 'tabular_generated_output_output_token_ratio', + TABULAR_EXPORT_DEFAULT_OUTPUT_TOKEN_RATIO, + minimum=0.1, + maximum=0.9, + )), + 1000, + ) + input_bound_chars = int(input_token_budget * TABULAR_EXPORT_APPROXIMATE_CHARS_PER_TOKEN) + max_batch_chars = input_bound_chars + if _normalize_tabular_run_task_type(task_type) != TABULAR_RUN_TASK_HIERARCHICAL_ANALYSIS: + output_expansion_ratio = _settings_float( + settings, + 'tabular_generated_output_output_expansion_ratio', + TABULAR_EXPORT_DEFAULT_OUTPUT_EXPANSION_RATIO, + minimum=0.5, + maximum=5.0, + ) + output_bound_chars = int( + output_token_budget + * TABULAR_EXPORT_APPROXIMATE_CHARS_PER_TOKEN + / output_expansion_ratio + ) + max_batch_chars = min(max_batch_chars, output_bound_chars) + max_batch_chars = _safe_int( + max_batch_chars, + minimum=6000, + maximum=TABULAR_EXPORT_MAX_SOURCE_BATCH_CHARS, + ) + configured_max_chars = settings.get('tabular_generated_output_max_batch_chars') + if configured_max_chars not in (None, ''): + max_batch_chars = min( + max_batch_chars, + _safe_int( + configured_max_chars, + default=TABULAR_EXPORT_DEFAULT_SOURCE_BATCH_CHARS, + minimum=6000, + maximum=TABULAR_EXPORT_MAX_SOURCE_BATCH_CHARS, + ), + ) + + scaled_batch_rows = math.ceil( + TABULAR_EXPORT_DEFAULT_SOURCE_BATCH_ROWS + * max_batch_chars + / TABULAR_EXPORT_DEFAULT_SOURCE_BATCH_CHARS + ) + max_batch_rows = _safe_int( + scaled_batch_rows, + minimum=1, + maximum=TABULAR_EXPORT_MAX_SOURCE_BATCH_ROWS, + ) + configured_max_rows = settings.get('tabular_generated_output_max_batch_rows') + if configured_max_rows not in (None, ''): + max_batch_rows = min( + max_batch_rows, + _safe_int( + configured_max_rows, + default=TABULAR_EXPORT_DEFAULT_SOURCE_BATCH_ROWS, + minimum=1, + maximum=TABULAR_EXPORT_MAX_SOURCE_BATCH_ROWS, + ), + ) + + return { + 'max_rows': max_batch_rows, + 'max_chars': max_batch_chars, + 'context_token_limit': context_token_limit, + 'output_token_limit': output_token_limit, + 'input_token_budget': input_token_budget, + 'output_token_budget': output_token_budget, + 'limit_source': token_limits.get('source'), + 'model': token_limits.get('model'), + } + + def _build_chat_service(gpt_model, settings, model_context=None): chunk_gpt_model, chunk_model_context = _resolve_tabular_chunk_model_selection( gpt_model, @@ -2278,7 +2657,27 @@ def _is_stale_queued_run(run, settings): def _is_retryable_failed_run(run): status = str((run or {}).get('status') or '').strip().lower() - return status == TABULAR_EXPORT_STATUS_FAILED and _is_retryable_export_error_message((run or {}).get('last_error')) + last_error = (run or {}).get('last_error') + return status == TABULAR_EXPORT_STATUS_FAILED and ( + _is_retryable_export_error_message(last_error) + or _is_retryable_model_validation_error_message(last_error) + ) + + +def _is_auto_retry_exhausted(run): + return bool((run or {}).get('auto_retry_exhausted')) + + +def _can_auto_retry_failed_run(run, settings=None): + if not _is_retryable_failed_run(run) or _is_auto_retry_exhausted(run): + return False + return _safe_int((run or {}).get('transient_failure_count')) < _settings_int( + settings or {}, + 'tabular_generated_output_max_transient_failures', + TABULAR_EXPORT_DEFAULT_MAX_TRANSIENT_FAILURES, + minimum=1, + maximum=100, + ) def _scheduler_candidate_reason(run, settings): @@ -2294,7 +2693,7 @@ def _scheduler_candidate_reason(run, settings): return 'running heartbeat is stale' return None if status == TABULAR_EXPORT_STATUS_FAILED: - if _is_retryable_failed_run(run): + if _can_auto_retry_failed_run(run, settings or {}): return 'failed run has retryable error' return None return None @@ -2314,7 +2713,7 @@ def _query_scheduler_candidates_by_status(status, scan_limit, settings): query = ( "SELECT " "c.id, c.user_id, c.status, c.created_at, c.updated_at, c.last_heartbeat_at, " - "c.next_attempt_at, c.last_error, c.transient_failure_count " + "c.next_attempt_at, c.last_error, c.transient_failure_count, c.auto_retry_exhausted " "FROM c WHERE c.type = @type AND c.status = @status " "ORDER BY c.updated_at ASC" ) @@ -2657,9 +3056,14 @@ def append_generated_artifact(artifact, fallback_file_name, fallback_output_form 'retry_delay_seconds': status_detail.get('retry_delay_seconds'), 'estimated_remaining_seconds': run.get('estimated_remaining_seconds'), 'estimated_total_seconds': run.get('estimated_total_seconds'), + 'rows_per_minute': run.get('rows_per_minute'), + 'batch_concurrency': _safe_int(run.get('batch_concurrency')), + 'effective_batch_concurrency': _safe_int(run.get('effective_batch_concurrency')), 'mismatch_count': _safe_int(run.get('mismatch_count')), 'retry_count': _safe_int(run.get('retry_count')), 'transient_failure_count': _safe_int(run.get('transient_failure_count')), + 'auto_retry_exhausted': bool(run.get('auto_retry_exhausted')), + 'last_retry_category': run.get('last_retry_category'), 'manual_resume_count': _safe_int(run.get('manual_resume_count')), 'next_attempt_at': run.get('next_attempt_at'), 'can_resume': can_resume, @@ -2802,6 +3206,8 @@ def resume_tabular_generated_output_run(user_id, run_id): 'next_attempt_at': now, 'last_message': 'Manual resume queued; export will continue from completed checkpoints', 'transient_failure_count': 0, + 'auto_retry_exhausted': False, + 'last_retry_category': 'manual_resume', 'manual_resume_count': _safe_int(run.get('manual_resume_count')) + 1, 'last_manual_resume_at': now, }) @@ -3002,17 +3408,7 @@ def _try_claim_run(user_id, run_id, settings): status = str(run.get('status') or '').strip().lower() if status in TABULAR_EXPORT_TERMINAL_STATUSES: - retryable_failed_run = ( - status == TABULAR_EXPORT_STATUS_FAILED - and _is_retryable_export_error_message(run.get('last_error')) - and _safe_int(run.get('transient_failure_count')) < _settings_int( - settings, - 'tabular_generated_output_max_transient_failures', - TABULAR_EXPORT_DEFAULT_MAX_TRANSIENT_FAILURES, - minimum=1, - maximum=100, - ) - ) + retryable_failed_run = status == TABULAR_EXPORT_STATUS_FAILED and _can_auto_retry_failed_run(run, settings) if not retryable_failed_run: return None if status == TABULAR_EXPORT_STATUS_RUNNING and not _is_stale_running_run(run, settings): @@ -3083,19 +3479,40 @@ def _mark_run_failed(run, error_message): return run -def _mark_run_retryable(run, error_message, settings): - transient_failure_count = _safe_int(run.get('transient_failure_count')) + 1 - max_transient_failures = _settings_int( +def _get_auto_retry_limit_for_category(settings, retry_category): + if retry_category == 'model_validation': + return _settings_int( + settings, + 'tabular_generated_output_model_validation_auto_retries', + TABULAR_EXPORT_DEFAULT_MODEL_VALIDATION_AUTO_RETRIES, + minimum=0, + maximum=10, + ) + return _settings_int( settings, 'tabular_generated_output_max_transient_failures', TABULAR_EXPORT_DEFAULT_MAX_TRANSIENT_FAILURES, minimum=1, maximum=100, ) - if transient_failure_count > max_transient_failures: + + +def _mark_run_retryable(run, error_message, settings, retry_category='transient'): + normalized_retry_category = str(retry_category or 'transient').strip().lower() or 'transient' + transient_failure_count = _safe_int(run.get('transient_failure_count')) + 1 + max_auto_retries = _get_auto_retry_limit_for_category(settings, normalized_retry_category) + if transient_failure_count > max_auto_retries: + exhausted_message = ( + 'Max automatic model-output retry attempts exceeded; last error: ' + if normalized_retry_category == 'model_validation' + else 'Max transient retry attempts exceeded; last error: ' + ) + run['auto_retry_exhausted'] = True + run['last_retry_category'] = normalized_retry_category + run['last_auto_retry_exhausted_at'] = _now_iso() return _mark_run_failed( run, - f'Max transient retry attempts exceeded; last error: {error_message}', + f'{exhausted_message}{error_message}', ) now = _now_utc() @@ -3109,8 +3526,14 @@ def _mark_run_retryable(run, error_message, settings): 'lease_holder_id': None, 'lease_expires_at': None, 'last_error': str(error_message or 'Transient background export error')[:1000], - 'last_message': 'Background structured export will resume after a transient connection error', + 'last_message': ( + 'Background structured export will retry after model-output validation failed' + if normalized_retry_category == 'model_validation' + else 'Background structured export will resume after a transient connection error' + ), 'transient_failure_count': transient_failure_count, + 'last_retry_category': normalized_retry_category, + 'auto_retry_exhausted': False, 'next_attempt_at': next_attempt_at, }) try: @@ -3128,7 +3551,8 @@ def _mark_run_retryable(run, error_message, settings): 'processed_rows': run.get('processed_rows'), 'row_count': run.get('row_count'), 'transient_failure_count': transient_failure_count, - 'max_transient_failures': max_transient_failures, + 'max_transient_failures': max_auto_retries, + 'retry_category': normalized_retry_category, 'next_attempt_at': next_attempt_at, 'error': str(error_message or '')[:1000], }, @@ -3137,47 +3561,91 @@ def _mark_run_retryable(run, error_message, settings): return run -def _update_run_progress(run, completed_batches, processed_rows, batch_rows, batch_elapsed_seconds, mismatch_count=0): - now = _now_utc() - started_at = str(run.get('started_at') or '').strip() - elapsed_seconds = 0.0 - if started_at: - try: - started_time = datetime.fromisoformat(started_at) - if started_time.tzinfo is None: - started_time = started_time.replace(tzinfo=timezone.utc) - elapsed_seconds = max((now - started_time).total_seconds(), 0.0) - except ValueError: - elapsed_seconds = 0.0 +def _is_schema_discovery_progress_window(run, completed_batches, window_batch_count): + return ( + _safe_int(completed_batches) == 1 + and _safe_int(window_batch_count) == 1 + and _safe_int((run or {}).get('batch_count')) > 1 + and _safe_int((run or {}).get('batch_concurrency')) > 1 + ) + +def _calculate_window_throughput( + run, + processed_rows, + window_rows, + window_elapsed_seconds, + completed_at, +): + recent_windows = list(run.get('recent_progress_windows') or [])[-9:] + normalized_window_rows = _safe_int(window_rows) + normalized_window_seconds = max(_safe_float(window_elapsed_seconds), 0.0) + if normalized_window_rows > 0 and normalized_window_seconds > 0: + recent_windows.append({ + 'row_count': normalized_window_rows, + 'elapsed_seconds': round(normalized_window_seconds, 3), + 'completed_at': completed_at.isoformat(), + }) + + sampled_rows = sum(_safe_int(window.get('row_count')) for window in recent_windows) + sampled_seconds = sum( + max(_safe_float(window.get('elapsed_seconds')), 0.0) + for window in recent_windows + ) + rows_per_minute = None + estimated_total_seconds = None + estimated_remaining_seconds = None + if sampled_rows > 0 and sampled_seconds > 0: + rows_per_second = sampled_rows / sampled_seconds + row_count = _safe_int(run.get('row_count')) + rows_per_minute = round(rows_per_second * 60, 2) + estimated_total_seconds = round(row_count / rows_per_second, 1) + estimated_remaining_seconds = round( + max(row_count - _safe_int(processed_rows), 0) / rows_per_second, + 1, + ) + + return { + 'recent_progress_windows': recent_windows, + 'rows_per_minute': rows_per_minute, + 'estimated_total_seconds': estimated_total_seconds, + 'estimated_remaining_seconds': estimated_remaining_seconds, + } + + +def _update_run_progress( + run, + completed_batches, + processed_rows, + window_rows, + window_elapsed_seconds, + window_batch_count, + mismatch_count=0, +): + now = _now_utc() active_processing_seconds = max(_safe_float(run.get('active_processing_seconds')), 0.0) - active_processing_seconds += max(_safe_float(batch_elapsed_seconds), 0.0) + active_processing_seconds += max(_safe_float(window_elapsed_seconds), 0.0) + batch_count = _safe_int(run.get('batch_count')) recent_batches = list(run.get('recent_batches') or [])[-9:] recent_batches.append({ 'batch_number': completed_batches, - 'row_count': _safe_int(batch_rows), - 'elapsed_seconds': round(_safe_float(batch_elapsed_seconds), 3), + 'batch_count': _safe_int(window_batch_count), + 'row_count': _safe_int(window_rows), + 'elapsed_seconds': round(_safe_float(window_elapsed_seconds), 3), 'completed_at': now.isoformat(), }) - - batch_count = _safe_int(run.get('batch_count')) - estimated_total_seconds = None - estimated_remaining_seconds = None - if completed_batches > 0 and batch_count > 0: - recent_elapsed_values = [ - _safe_float(batch.get('elapsed_seconds')) - for batch in recent_batches - if _safe_float(batch.get('elapsed_seconds')) > 0 - ] - if recent_elapsed_values: - seconds_per_batch = sum(recent_elapsed_values) / len(recent_elapsed_values) - elif active_processing_seconds > 0: - seconds_per_batch = active_processing_seconds / completed_batches - else: - seconds_per_batch = elapsed_seconds / completed_batches - estimated_total_seconds = round(seconds_per_batch * batch_count, 1) - estimated_remaining_seconds = round(seconds_per_batch * max(batch_count - completed_batches, 0), 1) - + is_schema_discovery_window = _is_schema_discovery_progress_window( + run, + completed_batches, + window_batch_count, + ) + throughput = _calculate_window_throughput( + run, + processed_rows, + 0 if is_schema_discovery_window else window_rows, + 0 if is_schema_discovery_window else window_elapsed_seconds, + now, + ) run.update({ 'completed_batches': completed_batches, 'processed_rows': processed_rows, @@ -3185,13 +3653,12 @@ def _update_run_progress(run, completed_batches, processed_rows, batch_rows, bat 'updated_at': now.isoformat(), 'last_heartbeat_at': now.isoformat(), 'active_processing_seconds': round(active_processing_seconds, 3), - 'estimated_total_seconds': estimated_total_seconds, - 'estimated_remaining_seconds': estimated_remaining_seconds, + 'effective_batch_concurrency': _safe_int(window_batch_count), 'mismatch_count': _safe_int(run.get('mismatch_count')) + _safe_int(mismatch_count), 'last_message': f"Processed structured export batch {completed_batches} of {batch_count}", + 'recent_batches': recent_batches, }) - run['recent_batches'] = recent_batches - + run.update(throughput) return _replace_claimed_run(run) @@ -3217,6 +3684,9 @@ def _log_progress_if_due(run, last_logged_at): 'row_count': run.get('row_count'), 'progress_percent': progress_percent, 'estimated_remaining_seconds': run.get('estimated_remaining_seconds'), + 'rows_per_minute': run.get('rows_per_minute'), + 'batch_concurrency': run.get('batch_concurrency'), + 'effective_batch_concurrency': run.get('effective_batch_concurrency'), 'mismatch_count': run.get('mismatch_count'), }, debug_only=True, @@ -4042,63 +4512,102 @@ def _build_passthrough_batch_results(run, batch_requests): def _advance_run_progress_for_window(run, batch_results, completed_batches, processed_rows, window_start, window_end): + window_results = [] + window_rows = 0 + window_mismatch_count = 0 for batch_number in range(window_start, window_end + 1): batch_result = batch_results.get(batch_number) if not batch_result: break + window_results.append(batch_result) completed_batches = batch_number - processed_rows += _safe_int(batch_result.get('batch_row_count')) + batch_rows = _safe_int(batch_result.get('batch_row_count')) + processed_rows += batch_rows + window_rows += batch_rows mismatch_count = _safe_int(batch_result.get('mismatch_count')) + window_mismatch_count += mismatch_count if mismatch_count: run['retry_count'] = _safe_int(run.get('retry_count')) + max(mismatch_count - 1, 0) + if window_results: + generated_elapsed_seconds = [ + max(_safe_float(batch_result.get('elapsed_seconds')), 0.0) + for batch_result in window_results + if not batch_result.get('from_checkpoint') + ] run = _update_run_progress( run, completed_batches, processed_rows, - batch_result.get('batch_row_count'), - batch_result.get('elapsed_seconds'), - mismatch_count=mismatch_count, + window_rows, + max(generated_elapsed_seconds, default=0.0), + len(window_results), + mismatch_count=window_mismatch_count, ) return run, completed_batches, processed_rows def _advance_analysis_map_progress_for_window(run, batch_results, completed_batches, processed_rows, window_start, window_end): + window_results = [] + window_rows = 0 for batch_number in range(window_start, window_end + 1): batch_result = batch_results.get(batch_number) if not batch_result: break + window_results.append(batch_result) completed_batches = batch_number - processed_rows += _safe_int(batch_result.get('batch_row_count')) + batch_rows = _safe_int(batch_result.get('batch_row_count')) + processed_rows += batch_rows + window_rows += batch_rows + if window_results: + generated_elapsed_seconds = [ + max(_safe_float(batch_result.get('elapsed_seconds')), 0.0) + for batch_result in window_results + if not batch_result.get('from_checkpoint') + ] run = _update_analysis_map_progress( run, completed_batches, processed_rows, - batch_result.get('batch_row_count'), - batch_result.get('elapsed_seconds'), + window_rows, + max(generated_elapsed_seconds, default=0.0), + len(window_results), ) return run, completed_batches, processed_rows -def _update_analysis_map_progress(run, completed_batches, processed_rows, batch_rows, batch_elapsed_seconds): +def _update_analysis_map_progress( + run, + completed_batches, + processed_rows, + window_rows, + window_elapsed_seconds, + window_batch_count, +): now = _now_utc() active_processing_seconds = max(_safe_float(run.get('active_processing_seconds')), 0.0) - active_processing_seconds += max(_safe_float(batch_elapsed_seconds), 0.0) + active_processing_seconds += max(_safe_float(window_elapsed_seconds), 0.0) batch_count = _safe_int(run.get('batch_count')) - estimated_total_seconds = None - estimated_remaining_seconds = None - if completed_batches > 0 and batch_count > 0 and active_processing_seconds > 0: - seconds_per_batch = active_processing_seconds / completed_batches - estimated_total_seconds = round(seconds_per_batch * batch_count, 1) - estimated_remaining_seconds = round(seconds_per_batch * max(batch_count - completed_batches, 0), 1) - recent_batches = list(run.get('recent_batches') or [])[-9:] recent_batches.append({ 'batch_number': completed_batches, - 'row_count': _safe_int(batch_rows), - 'elapsed_seconds': round(_safe_float(batch_elapsed_seconds), 3), + 'batch_count': _safe_int(window_batch_count), + 'row_count': _safe_int(window_rows), + 'elapsed_seconds': round(_safe_float(window_elapsed_seconds), 3), 'completed_at': now.isoformat(), 'phase': 'map', }) + is_schema_discovery_window = _is_schema_discovery_progress_window( + run, + completed_batches, + window_batch_count, + ) + throughput = _calculate_window_throughput( + run, + processed_rows, + 0 if is_schema_discovery_window else window_rows, + 0 if is_schema_discovery_window else window_elapsed_seconds, + now, + ) run.update({ 'completed_batches': completed_batches, 'processed_rows': processed_rows, @@ -4107,11 +4616,11 @@ def _update_analysis_map_progress(run, completed_batches, processed_rows, batch_ 'updated_at': now.isoformat(), 'last_heartbeat_at': now.isoformat(), 'active_processing_seconds': round(active_processing_seconds, 3), - 'estimated_total_seconds': estimated_total_seconds, - 'estimated_remaining_seconds': estimated_remaining_seconds, + 'effective_batch_concurrency': _safe_int(window_batch_count), 'last_message': f'Analyzed tabular chunk {completed_batches} of {batch_count}', 'recent_batches': recent_batches, }) + run.update(throughput) return _replace_claimed_run(run) @@ -4422,13 +4931,17 @@ def process_tabular_generated_output_run(run_id, user_id): minimum=1, maximum=5, ) - batch_concurrency = _settings_int( + batch_concurrency = _resolve_tabular_batch_concurrency( settings, - 'tabular_generated_output_batch_concurrency', - TABULAR_EXPORT_DEFAULT_BATCH_CONCURRENCY, - minimum=1, - maximum=TABULAR_EXPORT_MAX_BATCH_CONCURRENCY, + run.get('batch_count'), ) + if _safe_int(run.get('batch_concurrency')) != batch_concurrency: + run.update({ + 'batch_concurrency': batch_concurrency, + 'updated_at': _now_iso(), + 'last_heartbeat_at': _now_iso(), + }) + run = _replace_claimed_run(run) stale_seconds = _settings_int( settings, 'tabular_generated_output_stale_seconds', @@ -4581,7 +5094,9 @@ def process_tabular_generated_output_run(run_id, user_id): return _read_run(normalized_user_id, normalized_run_id) except Exception as exc: if _is_retryable_export_error(exc): - return _mark_run_retryable(run, exc, settings) + return _mark_run_retryable(run, exc, settings, retry_category='transient') + if _is_retryable_model_validation_error(exc): + return _mark_run_retryable(run, exc, settings, retry_category='model_validation') return _mark_run_failed(run, exc) @@ -4656,35 +5171,20 @@ def queue_tabular_generated_output_run( staged_char_count = 0 staged_batch_count = 0 staged_chunk_row_counts = [] + model_batch_budget = _build_model_aware_source_batch_budget( + gpt_model, + settings, + model_context=model_context, + task_type=normalized_task_type, + user_question=user_question, + ) if source_descriptor: staged_row_count = _safe_int(source_descriptor.get('expected_row_count')) if staged_row_count <= 0: raise ValueError('Source query descriptor must include the expected row count') - source_descriptor['batch_max_rows'] = _safe_int( - source_descriptor.get('batch_max_rows'), - default=_settings_int( - settings, - 'tabular_generated_output_max_batch_rows', - TABULAR_EXPORT_DEFAULT_SOURCE_BATCH_ROWS, - minimum=1, - maximum=100, - ), - minimum=1, - maximum=100, - ) - source_descriptor['batch_max_chars'] = _safe_int( - source_descriptor.get('batch_max_chars'), - default=_settings_int( - settings, - 'tabular_generated_output_max_batch_chars', - TABULAR_EXPORT_DEFAULT_SOURCE_BATCH_CHARS, - minimum=6000, - maximum=120000, - ), - minimum=6000, - maximum=120000, - ) + source_descriptor['batch_max_rows'] = model_batch_budget['max_rows'] + source_descriptor['batch_max_chars'] = model_batch_budget['max_chars'] staged_batch_count = max( 1, math.ceil(staged_row_count / source_descriptor['batch_max_rows']), @@ -4765,6 +5265,7 @@ def queue_tabular_generated_output_run( 'processed_rows': 0, 'output_schema': None, 'source_descriptor': source_descriptor or None, + 'batch_budget': model_batch_budget, 'source_authorization': source_authorization or None, 'source_staging_complete': not bool(source_descriptor), 'source_staged_rows': 0 if source_descriptor else staged_row_count, @@ -4779,6 +5280,7 @@ def queue_tabular_generated_output_run( 'mismatch_count': 0, 'retry_count': 0, 'recent_batches': [], + 'recent_progress_windows': [], 'analysis_phase': 'queued' if _is_tabular_analysis_task(normalized_task_type) else None, 'active_processing_seconds': 0, 'last_message': ( @@ -4813,6 +5315,11 @@ def queue_tabular_generated_output_run( 'batch_count': staged_batch_count, 'total_chunk_count': staged_batch_count, 'staged_input_char_count': staged_char_count, + 'batch_max_rows': model_batch_budget.get('max_rows'), + 'batch_max_chars': model_batch_budget.get('max_chars'), + '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'), 'source_backed': bool(source_descriptor), 'submitted_to_executor': submitted, }, diff --git a/application/single_app/route_backend_chats.py b/application/single_app/route_backend_chats.py index 3beff797..66813572 100644 --- a/application/single_app/route_backend_chats.py +++ b/application/single_app/route_backend_chats.py @@ -17628,12 +17628,12 @@ def record_tabular_post_processing_thought(thought_payload): tabular_invocations = [] tabular_related_document_summary = '' tabular_generated_output = maybe_queue_direct_tabular_generated_output( - user_message, - workspace_tabular_file_contexts, - user_id, - conversation_id, - gpt_model, - settings, + user_question=user_message, + file_contexts=workspace_tabular_file_contexts, + user_id=user_id, + conversation_id=conversation_id, + gpt_model=gpt_model, + settings=settings, thought_callback=record_tabular_post_processing_thought, model_context=tabular_model_context, ) @@ -17995,12 +17995,12 @@ def record_tabular_post_processing_thought(thought_payload): for file_name in chat_tabular_files ] chat_tabular_generated_output = maybe_queue_direct_tabular_generated_output( - user_message, - chat_tabular_file_contexts, - user_id, - conversation_id, - gpt_model, - settings, + user_question=user_message, + file_contexts=chat_tabular_file_contexts, + user_id=user_id, + conversation_id=conversation_id, + gpt_model=gpt_model, + settings=settings, thought_callback=record_tabular_post_processing_thought, model_context=tabular_model_context, ) @@ -21479,10 +21479,12 @@ def record_and_publish_streaming_thought(thought_payload): tabular_invocations = [] tabular_related_document_summary = '' tabular_generated_output = maybe_queue_direct_tabular_generated_output( - user_message, - workspace_tabular_file_contexts, - user_id, + user_question=user_message, + file_contexts=workspace_tabular_file_contexts, + user_id=user_id, conversation_id=conversation_id, + gpt_model=gpt_model, + settings=settings, thought_callback=record_and_publish_streaming_thought, model_context=tabular_model_context, ) @@ -21861,12 +21863,12 @@ def record_and_publish_streaming_thought(thought_payload): for file_name in chat_tabular_files ] chat_tabular_generated_output = maybe_queue_direct_tabular_generated_output( - user_message, - chat_tabular_file_contexts, - user_id, - conversation_id, - gpt_model, - settings, + user_question=user_message, + file_contexts=chat_tabular_file_contexts, + user_id=user_id, + conversation_id=conversation_id, + gpt_model=gpt_model, + settings=settings, thought_callback=record_and_publish_streaming_thought, model_context=tabular_model_context, ) diff --git a/application/single_app/static/js/chat/chat-messages.js b/application/single_app/static/js/chat/chat-messages.js index 4a3ed30a..f1207321 100644 --- a/application/single_app/static/js/chat/chat-messages.js +++ b/application/single_app/static/js/chat/chat-messages.js @@ -4420,6 +4420,9 @@ function renderReplyQuoteHtml(fullMessageObject = null) { const manualResumeCount = Number.parseInt(outputMetadata?.manual_resume_count, 10); const retryDelaySeconds = Number.parseInt(outputMetadata?.retry_delay_seconds, 10); const estimatedRemainingSeconds = Number.parseInt(outputMetadata?.estimated_remaining_seconds, 10); + const rowsPerMinute = Number.parseFloat(outputMetadata?.rows_per_minute); + const batchConcurrency = Number.parseInt(outputMetadata?.batch_concurrency, 10); + const effectiveBatchConcurrency = Number.parseInt(outputMetadata?.effective_batch_concurrency, 10); const taskType = String(outputMetadata?.task_type || '').trim().toLowerCase(); const analysisPhase = String(outputMetadata?.analysis_phase || '').trim().toLowerCase(); const progressPercent = calculateGeneratedOutputProgress(outputMetadata); @@ -4488,6 +4491,19 @@ function renderReplyQuoteHtml(fullMessageObject = null) { if (Number.isFinite(processedChunkCount) && Number.isFinite(totalChunkCount) && totalChunkCount > processedChunkCount) { detailParts.push(`Remaining chunks: ${(totalChunkCount - processedChunkCount).toLocaleString()}`); } + if (Number.isFinite(rowsPerMinute) && rowsPerMinute > 0) { + detailParts.push(`Throughput: ${rowsPerMinute.toLocaleString(undefined, { maximumFractionDigits: 1 })} rows/min`); + } + if (Number.isFinite(batchConcurrency) && batchConcurrency > 0) { + const concurrencyLabel = ( + Number.isFinite(effectiveBatchConcurrency) + && effectiveBatchConcurrency > 0 + && effectiveBatchConcurrency !== batchConcurrency + ) + ? `${effectiveBatchConcurrency.toLocaleString()} of ${batchConcurrency.toLocaleString()}` + : batchConcurrency.toLocaleString(); + detailParts.push(`Model concurrency: ${concurrencyLabel}`); + } if (outputMetadata?.waiting_for_retry) { const nextAttempt = formatGeneratedOutputTimestamp(outputMetadata?.next_attempt_at); diff --git a/application/single_app/static/json/model_capabilities.json b/application/single_app/static/json/model_capabilities.json index ec547036..d44d19ad 100644 --- a/application/single_app/static/json/model_capabilities.json +++ b/application/single_app/static/json/model_capabilities.json @@ -2,7 +2,7 @@ "$schema": "https://simplechat.local/schemas/model-capabilities.schema.json", "schemaVersion": 1, "lastUpdated": "2026-08-04", - "description": "Initial SimpleChat model capability catalog for multimodal and coding feature selection. This file is data-only and is not wired into runtime behavior yet.", + "description": "SimpleChat model capability catalog. Capability flags remain data-only; optional model token-limit fields are consumed by durable tabular batch planning when present.", "capabilityFields": { "processesText": "Accepts text input.", "generatesText": "Produces text output.", diff --git a/docs/explanation/features/TABULAR_BACKGROUND_GENERATED_EXPORTS.md b/docs/explanation/features/TABULAR_BACKGROUND_GENERATED_EXPORTS.md index 00221181..97e2fba6 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.241.064** +Updated through version: **0.250.136** ## Overview @@ -25,7 +25,7 @@ The feature supports large spreadsheet-driven analysis, including workbooks that - Chat and workflow tabular generated-output requests continue to use the existing inline path for smaller exports. - Oversized structured exports are queued with `queue_tabular_generated_output_run(...)`. -- Input row batches are staged as a single blob-backed JSON payload. +- Version-pinned CSV sources are replayed into bounded per-batch input checkpoints without model pagination. - Each completed model batch is checkpointed as an output blob. - Cosmos stores compact run metadata, progress counts, retry state, and final artifact metadata. - The background scheduler claims queued runs with optimistic status updates and resumes from checkpointed output batches. @@ -45,8 +45,13 @@ The feature supports large spreadsheet-driven analysis, including workbooks that - `tabular_generated_output_max_batch_rows` - `tabular_generated_output_max_batch_chars` - `tabular_generated_output_batch_concurrency` +- `tabular_generated_output_input_token_ratio` +- `tabular_generated_output_large_context_input_token_ratio` +- `tabular_generated_output_input_token_soft_cap` +- `tabular_generated_output_output_token_ratio` +- `tabular_generated_output_output_expansion_ratio` -If settings are absent, conservative defaults are used. +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. ### File Structure @@ -62,11 +67,12 @@ Users continue requesting tabular structured output in chat or workflows. For sm When a workflow/document analysis request also creates a full generated tabular export, the generated export is presented as the primary deliverable. The analysis layer may still attach a supporting CSV preview, but redundant analysis JSON and Markdown artifacts are suppressed so they do not compete with the full generated export card. -The progress card displays current status, completed checkpoint counts, processed row counts, estimated remaining time, scheduled retry time, retry-due state, transient retry count, manual continuation count, last update time, and heartbeat time when available. +The progress card displays current status, completed checkpoint counts, processed row counts, wall-clock rows per minute, model concurrency, estimated remaining time, scheduled retry time, retry-due state, transient retry count, manual continuation count, last update time, and heartbeat time when available. ## Testing and Validation - Functional regression: `functional_tests/test_tabular_background_generated_exports.py` +- Scale and performance regression: `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. @@ -75,7 +81,10 @@ The progress card displays current status, completed checkpoint counts, processe - The request only stages durable input and queues work for oversized exports. - Phase 3 batch packing compacts generated-export prompt payloads, removes internal tabular helper fields from staged model input, avoids duplicating row-linked document excerpts as synthetic attachment text, and packs rows by configurable row and character budgets. -- Phase 4 bounded concurrency lets the background worker generate a small configurable window of model batches in parallel while checkpointing successful batches and advancing public progress only in contiguous batch order. +- Model-aware packing targets 50% of ordinary model input capacity and 60% of output capacity. Context windows above 500,000 tokens use a lower 30% input ratio and a default 180,000-token soft input cap. +- 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. - 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. @@ -83,6 +92,8 @@ The progress card displays current status, completed checkpoint counts, processe ## Known Limitations - Background runs still depend on configured background scheduler capacity and available Azure OpenAI throughput. +- One durable run is still claimed by one application worker; App Service scale-out does not shard a single run across workers. +- Completion time remains proportional to LLM-generated output volume and model generation speed. Higher batching and concurrency improve throughput but do not guarantee a fixed completion time. - Completion appears through status polling or on the next chat reload; no push notification is added in this version. - Manual continuation applies to retryable failures, stale running leases, queued retries whose retry time has passed, and stale queued runs; hard validation failures remain terminal. @@ -92,3 +103,4 @@ The progress card displays current status, completed checkpoint counts, processe - `application/single_app/config.py` was updated to version **0.241.059** for Phase 3 compact batch packing. - `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. diff --git a/functional_tests/test_tabular_row_orchestration_scale.py b/functional_tests/test_tabular_row_orchestration_scale.py index d8ede1ba..1708fa55 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.133 -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 +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 This test ensures generated exports preserve source identity and row order while enforcing one stable output schema across independently generated batches. @@ -20,7 +20,7 @@ import re import sys import uuid -from datetime import datetime, timezone +from datetime import datetime, timedelta, timezone from pathlib import Path from types import ModuleType, SimpleNamespace from collections import Counter @@ -87,6 +87,26 @@ '_raise_if_tabular_export_canceled', '_replace_claimed_run', } +RETRY_FUNCTIONS = { + '_safe_int', + '_settings_int', + '_is_retryable_export_error_message', + '_is_retryable_model_validation_error_message', + '_is_retryable_failed_run', + '_is_auto_retry_exhausted', + '_can_auto_retry_failed_run', + '_mark_run_failed', + '_get_auto_retry_limit_for_category', + '_mark_run_retryable', +} +RETRY_CONSTANTS = { + 'TABULAR_EXPORT_STATUS_FAILED', + 'TABULAR_EXPORT_STATUS_QUEUED', + 'TABULAR_EXPORT_DEFAULT_MAX_TRANSIENT_FAILURES', + 'TABULAR_EXPORT_DEFAULT_MODEL_VALIDATION_AUTO_RETRIES', + 'TABULAR_EXPORT_RETRYABLE_MESSAGE_MARKERS', + 'TABULAR_EXPORT_MODEL_VALIDATION_RETRYABLE_MESSAGE_MARKERS', +} LEGACY_MIGRATION_FUNCTIONS = { '_normalize_source_identity_label', '_select_source_row_identity', @@ -137,6 +157,25 @@ 'TABULAR_ANALYSIS_MAX_FINDINGS', 'TABULAR_ANALYSIS_MAX_NOTABLE_ROWS', } +PERFORMANCE_FUNCTIONS = { + '_safe_int', + '_safe_float', + '_settings_int', + '_settings_float', + '_resolve_tabular_batch_concurrency', + '_normalize_tabular_run_task_type', + '_resolve_tabular_chunk_model_selection', + '_normalize_tabular_model_identifier', + '_get_tabular_model_record_identifiers', + '_read_tabular_model_token_limit', + '_iter_configured_tabular_model_records', + '_load_tabular_model_limit_catalog', + '_resolve_tabular_model_token_limits', + '_build_model_aware_source_batch_budget', + '_is_schema_discovery_progress_window', + '_calculate_window_throughput', + '_advance_run_progress_for_window', +} def _load_contract_helpers(): @@ -630,6 +669,55 @@ class TabularExportLeaseLostError(RuntimeError): return namespace +def _load_retry_helpers(): + module_tree = ast.parse(EXPORT_MODULE.read_text(encoding='utf-8'), filename=str(EXPORT_MODULE)) + selected_nodes = [] + found_functions = set() + found_constants = set() + + for node in module_tree.body: + if isinstance(node, ast.FunctionDef) and node.name in RETRY_FUNCTIONS: + selected_nodes.append(node) + found_functions.add(node.name) + elif isinstance(node, ast.Assign): + assigned_names = { + target.id + for target in node.targets + if isinstance(target, ast.Name) + } + if assigned_names & RETRY_CONSTANTS: + selected_nodes.append(node) + found_constants.update(assigned_names & RETRY_CONSTANTS) + + missing_functions = RETRY_FUNCTIONS - found_functions + missing_constants = RETRY_CONSTANTS - found_constants + if missing_functions or missing_constants: + raise AssertionError( + f'Missing retry helpers: functions={sorted(missing_functions)}, constants={sorted(missing_constants)}' + ) + + stored_run = {} + + def replace_claimed_run(run): + stored_run.clear() + stored_run.update(run) + return dict(stored_run) + + namespace = { + 'timedelta': timedelta, + 'logging': logging, + 'TabularExportLeaseLostError': RuntimeError, + '_now_iso': lambda: '2026-08-09T16:00:00+00:00', + '_now_utc': lambda: datetime(2026, 8, 9, 16, 0, 0, tzinfo=timezone.utc), + '_replace_claimed_run': replace_claimed_run, + '_read_run': lambda user_id, run_id: dict(stored_run), + 'log_event': lambda *args, **kwargs: None, + } + extracted_module = ast.Module(body=selected_nodes, type_ignores=[]) + exec(compile(extracted_module, str(EXPORT_MODULE), 'exec'), namespace) + return namespace, stored_run + + def _load_legacy_migration_helper(aggregate_batches): module_tree = ast.parse(EXPORT_MODULE.read_text(encoding='utf-8'), filename=str(EXPORT_MODULE)) selected_nodes = [ @@ -895,6 +983,185 @@ def query_items(self, query, parameters, enable_cross_partition_query): return namespace['_query_scheduler_candidates_by_status'], run_container +def _load_performance_helpers(progress_updates=None): + module_tree = ast.parse(EXPORT_MODULE.read_text(encoding='utf-8'), filename=str(EXPORT_MODULE)) + selected_nodes = [] + found_functions = set() + for node in module_tree.body: + if isinstance(node, ast.FunctionDef) and node.name in PERFORMANCE_FUNCTIONS: + selected_nodes.append(node) + found_functions.add(node.name) + elif isinstance(node, ast.Assign): + assigned_names = { + target.id + for target in node.targets + if isinstance(target, ast.Name) + } + if any( + name.startswith('TABULAR_EXPORT_') or name.startswith('TABULAR_RUN_TASK_') + for name in assigned_names + ): + selected_nodes.append(node) + + missing_functions = PERFORMANCE_FUNCTIONS - found_functions + if missing_functions: + raise AssertionError(f'Missing performance helper functions: {sorted(missing_functions)}') + + namespace = { + '__file__': str(EXPORT_MODULE), + 'json': json, + 'math': math, + 'os': os, + 're': re, + } + extracted_module = ast.Module(body=selected_nodes, type_ignores=[]) + exec(compile(extracted_module, str(EXPORT_MODULE), 'exec'), namespace) + + progress_updates = progress_updates if progress_updates is not None else [] + + def update_progress( + run, + completed_batches, + processed_rows, + window_rows, + window_elapsed_seconds, + window_batch_count, + mismatch_count=0, + ): + progress_updates.append({ + 'completed_batches': completed_batches, + 'processed_rows': processed_rows, + 'window_rows': window_rows, + 'window_elapsed_seconds': window_elapsed_seconds, + 'window_batch_count': window_batch_count, + 'mismatch_count': mismatch_count, + }) + return run + + namespace['_update_run_progress'] = update_progress + return namespace + + +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() + build_budget = helpers['_build_model_aware_source_batch_budget'] + + fallback_budget = build_budget( + 'unlisted-model', + {}, + model_context={'model_id': 'unlisted-model'}, + ) + assert fallback_budget['limit_source'] == 'fallback' + assert fallback_budget['max_chars'] == 104856 + assert fallback_budget['max_rows'] == 88 + + catalog_records = [{ + 'id': 'large-context-model', + 'tokenLimits': { + 'inputTokenLimit': 1000000, + 'outputTokenLimit': 200000, + }, + }] + structured_budget = build_budget( + 'large-context-model', + {}, + model_context={'model_id': 'large-context-model'}, + catalog_records=catalog_records, + ) + assert structured_budget['limit_source'] == 'catalog' + assert structured_budget['context_token_limit'] == 1000000 + assert structured_budget['output_token_limit'] == 200000 + assert structured_budget['input_token_budget'] == 175904 + assert structured_budget['output_token_budget'] == 120000 + assert structured_budget['max_chars'] == 320000 + assert structured_budget['max_rows'] == 267 + + analysis_budget = build_budget( + 'large-context-model', + {}, + model_context={'model_id': 'large-context-model'}, + task_type='hierarchical_analysis', + catalog_records=catalog_records, + ) + assert analysis_budget['max_chars'] == 703616 + assert analysis_budget['max_rows'] == 500 + + custom_deployment_budget = build_budget( + 'prod-west-chat', + { + 'gpt_model': { + 'selected': [{ + 'deploymentName': 'prod-west-chat', + 'modelName': 'large-context-model', + }], + }, + }, + catalog_records=catalog_records, + ) + assert custom_deployment_budget['limit_source'] == 'catalog' + assert custom_deployment_budget['context_token_limit'] == 1000000 + assert custom_deployment_budget['output_token_limit'] == 200000 + + +def test_dynamic_concurrency_and_parallel_window_eta(): + """Large runs use 16 calls and ETA measures rows per parallel wall-clock window.""" + progress_updates = [] + helpers = _load_performance_helpers(progress_updates) + resolve_concurrency = helpers['_resolve_tabular_batch_concurrency'] + assert resolve_concurrency({}, 1) == 1 + assert resolve_concurrency({}, 10) == 4 + assert resolve_concurrency({}, 100) == 16 + assert resolve_concurrency({}, 128) == 64 + assert resolve_concurrency({}, 256) == 128 + assert resolve_concurrency({'tabular_generated_output_batch_concurrency': 96}, 1000) == 96 + is_schema_window = helpers['_is_schema_discovery_progress_window'] + assert is_schema_window({'batch_count': 909, 'batch_concurrency': 128}, 1, 1) is True + assert is_schema_window({'batch_count': 909, 'batch_concurrency': 128}, 129, 128) is False + assert is_schema_window({'batch_count': 1, 'batch_concurrency': 1}, 1, 1) is False + + throughput = helpers['_calculate_window_throughput']( + {'row_count': 30000}, + processed_rows=528, + window_rows=528, + window_elapsed_seconds=155, + completed_at=datetime(2026, 8, 9, tzinfo=timezone.utc), + ) + assert math.isclose(throughput['rows_per_minute'], 204.39, abs_tol=0.01) + assert throughput['estimated_total_seconds'] == 8806.8 + assert throughput['estimated_remaining_seconds'] == 8651.8 + + run = {'retry_count': 0} + updated_run, completed_batches, processed_rows = helpers['_advance_run_progress_for_window']( + run, + { + 1: {'batch_row_count': 33, 'elapsed_seconds': 150, 'mismatch_count': 0}, + 2: {'batch_row_count': 33, 'elapsed_seconds': 149, 'mismatch_count': 2}, + 3: { + 'batch_row_count': 33, + 'elapsed_seconds': 0.01, + 'mismatch_count': 0, + 'from_checkpoint': True, + }, + }, + completed_batches=0, + processed_rows=0, + window_start=1, + window_end=3, + ) + assert updated_run['retry_count'] == 1 + assert completed_batches == 3 + assert processed_rows == 99 + assert progress_updates == [{ + 'completed_batches': 3, + 'processed_rows': 99, + 'window_rows': 99, + 'window_elapsed_seconds': 150.0, + 'window_batch_count': 3, + 'mismatch_count': 2, + }] + + def test_source_identity_and_order_contract(): """Every row receives a canonical ordinal and preserves its source identifier.""" helpers = _load_contract_helpers() @@ -1512,6 +1779,87 @@ def _resolve_blob_location_with_fallback(self, *args, **kwargs): ) +def test_direct_source_backed_queue_call_sites_use_required_keywords(): + """Every direct queue call site passes required arguments explicitly.""" + module_tree = ast.parse(CHAT_ROUTE.read_text(encoding='utf-8'), filename=str(CHAT_ROUTE)) + direct_queue_calls = [ + call + for call in ast.walk(module_tree) + if isinstance(call, ast.Call) + and isinstance(call.func, ast.Name) + and call.func.id == 'maybe_queue_direct_tabular_generated_output' + ] + assert len(direct_queue_calls) >= 4 + + required_keyword_names = { + 'user_question', + 'file_contexts', + 'user_id', + 'conversation_id', + 'gpt_model', + 'settings', + } + for call in direct_queue_calls: + keyword_names = { + keyword.arg + for keyword in call.keywords + if keyword.arg + } + assert required_keyword_names <= keyword_names + + +def test_model_validation_failures_auto_retry_then_manual_continue(): + """Model-output validation failures auto retry briefly, then remain manually resumable.""" + helpers, stored_run = _load_retry_helpers() + settings = { + 'tabular_generated_output_model_validation_auto_retries': 3, + } + run = { + 'id': 'validation-run', + 'user_id': 'user-1', + 'conversation_id': 'conversation-1', + 'status': 'running', + 'completed_batches': 1, + 'processed_rows': 33, + 'batch_count': 91, + 'row_count': 3000, + '_etag': 'etag-1', + } + validation_error = ValueError( + 'Background structured export batch 2/91 failed validation: returned 0 object(s) for 33 input row(s).' + ) + + first_retry = helpers['_mark_run_retryable']( + dict(run), + validation_error, + settings, + retry_category='model_validation', + ) + assert first_retry['status'] == 'queued' + assert first_retry['transient_failure_count'] == 1 + assert first_retry['last_retry_category'] == 'model_validation' + assert first_retry['auto_retry_exhausted'] is False + assert first_retry['next_attempt_at'] + + exhausted_run = dict(first_retry) + exhausted_run.update({ + 'status': 'running', + 'transient_failure_count': 3, + '_etag': 'etag-2', + }) + exhausted_retry = helpers['_mark_run_retryable']( + exhausted_run, + validation_error, + settings, + retry_category='model_validation', + ) + assert exhausted_retry['status'] == 'failed' + assert exhausted_retry['auto_retry_exhausted'] is True + assert exhausted_retry['last_retry_category'] == 'model_validation' + assert helpers['_is_retryable_failed_run'](stored_run) is True + assert helpers['_can_auto_retry_failed_run'](stored_run, settings) is False + + def test_hierarchical_analysis_routing_requires_feature_flag(): """Lane C queueing stays behind the feature flag until scale hardening completes.""" candidate_helpers = _load_candidate_helpers() @@ -2443,6 +2791,8 @@ def test_route_queues_replayable_pages_and_suppresses_summary_fallback(): def main(): """Run focused row-orchestration contract checks.""" tests = [ + test_model_aware_batch_budget_uses_safe_token_limits, + test_dynamic_concurrency_and_parallel_window_eta, test_source_identity_and_order_contract, test_generated_batch_schema_contract, test_durable_runner_enforces_row_contract, @@ -2454,6 +2804,8 @@ def main(): test_filter_rows_pages_queue_combined_analysis_and_export_run, test_direct_source_backed_csv_queue_bypasses_tool_paging, test_direct_source_backed_queue_failure_falls_back_without_stream_abort, + test_direct_source_backed_queue_call_sites_use_required_keywords, + test_model_validation_failures_auto_retry_then_manual_continue, test_hierarchical_analysis_routing_requires_feature_flag, test_non_replayable_filter_rows_reports_explicit_failure, test_streaming_finalizer_writes_30000_rows_in_bounded_chunks, diff --git a/ui_tests/test_chat_background_generated_export_status.py b/ui_tests/test_chat_background_generated_export_status.py index 5518f2b0..d02063a8 100644 --- a/ui_tests/test_chat_background_generated_export_status.py +++ b/ui_tests/test_chat_background_generated_export_status.py @@ -1,8 +1,8 @@ # test_chat_background_generated_export_status.py """ UI test for chat background generated export status cards. -Version: 0.250.131 -Implemented in: 0.241.046; cancellation in 0.250.060; automatic-only refresh in 0.250.061; combined progress and large-run confirmation in 0.250.131 +Version: 0.250.136 +Implemented in: 0.241.046; cancellation in 0.250.060; automatic-only refresh in 0.250.061; combined progress and large-run confirmation in 0.250.131; throughput and concurrency status in 0.250.136 This test ensures queued tabular generated exports render progress in chat and turn into a downloadable artifact when complete or a visible canceled state. @@ -280,6 +280,9 @@ def test_chat_combined_background_status_shows_reduce_progress(playwright) -> No "failed_chunk_count": 0, "progress_percent": 80, "estimated_remaining_seconds": 120, + "rows_per_minute": 1200.5, + "batch_concurrency": 16, + "effective_batch_concurrency": 16, "background_export": True, }, }, @@ -326,6 +329,9 @@ def test_chat_combined_background_status_shows_reduce_progress(playwright) -> No processed_chunk_count: 48, progress_percent: 80, estimated_remaining_seconds: 120, + rows_per_minute: 1200.5, + batch_concurrency: 16, + effective_batch_concurrency: 16, file_name: 'combined-output.csv', output_format: 'csv', source_file_name: 'large-source.csv' @@ -346,6 +352,8 @@ def test_chat_combined_background_status_shows_reduce_progress(playwright) -> No expect(message.get_by_text("Remaining batches: 12")).to_be_visible() expect(message.get_by_text("Remaining chunks: 12")).to_be_visible() expect(message.get_by_text("Estimated remaining: 2m")).to_be_visible() + expect(message.get_by_text("Throughput: 1,200.5 rows/min")).to_be_visible() + expect(message.get_by_text("Model concurrency: 16")).to_be_visible() assert page_errors == [] finally: context.close()