diff --git a/application/single_app/config.py b/application/single_app/config.py index 9d6d17910..bfc800eaf 100644 --- a/application/single_app/config.py +++ b/application/single_app/config.py @@ -101,7 +101,7 @@ EXECUTOR_TYPE = 'thread' EXECUTOR_MAX_WORKERS = 30 SESSION_TYPE = 'filesystem' -VERSION = "0.261.305" +VERSION = "0.261.306" IS_DEVELOPMENT = is_development_env_enabled() # Opt-out for deployments where App Service Easy Auth is active but the platform diff --git a/application/single_app/functions_orchestration_recovery.py b/application/single_app/functions_orchestration_recovery.py index d001d0d63..c0c7a7b5f 100644 --- a/application/single_app/functions_orchestration_recovery.py +++ b/application/single_app/functions_orchestration_recovery.py @@ -1,7 +1,7 @@ # functions_orchestration_recovery.py """Execution leases and explicitly requested, checkpoint-only retry attempts. -Version: 0.261.304 +Version: 0.261.306 Retry publication is one transactional parent CAS + child create. It never replans, invokes an adapter, or changes plan-revision lineage. Terminal publication preserves an administrator's reply retraction; its probe @@ -18,11 +18,15 @@ A retry that could only resend requests a service declined is not offered. A run from the removed legacy contract is never retried, resumed or continued; only conversation deletion still reads it, to remove its saved data. +Conversation cleanup reports safe failure stages and hashed correlations without +logging provider text or changing the fail-closed deletion policy (0.261.306). """ import logging +import re import threading import uuid +from contextlib import contextmanager from copy import deepcopy from datetime import datetime, timedelta, timezone @@ -32,14 +36,14 @@ from content_screening.contracts import ScreeningError import functions_orchestration_runs as run_store -from functions_appinsights import log_event +from functions_appinsights import log_event, workflow_log_context from functions_orchestration_checkpoints import ( CHECKPOINT_VERSION, DEPENDENCY_STATE_FIELDS, LIFECYCLE_ID, OPTIONAL_STATE_FIELDS, STATE_FIELDS, CheckpointError, CheckpointStore, context_binding, context_state, effective_plan, fingerprint, orchestration_answer_message_id, restore_context, step_input_fingerprint, ) from functions_orchestration_plan_revisions import PlanRevisionError, read_revision_run -from functions_orchestration_output_store import build_output_cleanup_intent +from functions_orchestration_output_store import OutputError, build_output_cleanup_intent from functions_orchestration_registry import ( CAPABILITY_ACTION_INVOKE, CAPABILITY_WORKFLOW_RESULTS, admitted_export_pairs, external_effect_capability_ids, get_capability, @@ -1465,6 +1469,41 @@ def commit(self, step, result, input_fingerprint, *, reused=None): ) +def conversation_cleanup_failure_context(error, conversation_id, *, stage, run_id=None): + """Retain safe cleanup diagnostics without provider text or raw identifiers.""" + context = { + **workflow_log_context(conversation_id=conversation_id, run_id=run_id), + 'stage': stage, + 'error_type': type(error).__name__, + } + if isinstance(error, ( + RecoveryError, CheckpointError, PlanRevisionError, OutputError, + AnalysisWorkUnitConflictError, WorkflowResultIntegrityError, ResultUnavailableError, + )): + code = getattr(error, 'code', None) + if isinstance(code, str) and re.fullmatch(r'[a-z][a-z0-9_]{0,79}', code): + context['failure_code'] = code + status_code = getattr(error, 'status_code', None) + if type(status_code) is int and 400 <= status_code <= 599: + context['status_code'] = status_code + return context + + +@contextmanager +def _conversation_cleanup_stage(conversation_id, *, stage, run_id=None): + try: + yield + except Exception as error: + log_event( + '[ORCHESTRATION_RUNS] Conversation cleanup stage failed.', + extra=conversation_cleanup_failure_context( + error, conversation_id, stage=stage, run_id=run_id, + ), + level=logging.ERROR, exceptionTraceback=False, + ) + raise + + def cleanup_conversation_checkpoints( conversation_id, user_id, authorize, *, message_container=None, conversation_container=None, analysis_cleanup=None, analysis_fence=None, output_cleanup=None, retain_committed=False, @@ -1473,27 +1512,29 @@ def cleanup_conversation_checkpoints( if not callable(authorize) or authorize() is False: raise RecoveryError(code='not_found', status_code=404) if conversation_container is not None: - for _ in range(8): - conversation = conversation_container.read_item(item=conversation_id, partition_key=conversation_id) - if conversation.get('user_id') != user_id: - raise RecoveryError(code='not_found', status_code=404) - replacement = run_store._strip_cosmos_metadata(conversation) - replacement['orchestration_deleted'] = True - try: - conversation_container.replace_item( - item=conversation_id, body=replacement, etag=conversation['_etag'], - match_condition=MatchConditions.IfNotModified, - ) - break - except exceptions.CosmosAccessConditionFailedError: - continue - else: - raise CheckpointError() - rows = list(run_store.cosmos_orchestration_runs_container.query_items( - query='SELECT * FROM c WHERE c.conversation_id = @conversation_id AND c.user_id = @user_id', - parameters=[{'name': '@conversation_id', 'value': conversation_id}, {'name': '@user_id', 'value': user_id}], - partition_key=conversation_id, - )) + with _conversation_cleanup_stage(conversation_id, stage='conversation_fence'): + for _ in range(8): + conversation = conversation_container.read_item(item=conversation_id, partition_key=conversation_id) + if conversation.get('user_id') != user_id: + raise RecoveryError(code='not_found', status_code=404) + replacement = run_store._strip_cosmos_metadata(conversation) + replacement['orchestration_deleted'] = True + try: + conversation_container.replace_item( + item=conversation_id, body=replacement, etag=conversation['_etag'], + match_condition=MatchConditions.IfNotModified, + ) + break + except exceptions.CosmosAccessConditionFailedError: + continue + else: + raise CheckpointError() + with _conversation_cleanup_stage(conversation_id, stage='run_discovery'): + rows = list(run_store.cosmos_orchestration_runs_container.query_items( + query='SELECT * FROM c WHERE c.conversation_id = @conversation_id AND c.user_id = @user_id', + parameters=[{'name': '@conversation_id', 'value': conversation_id}, {'name': '@user_id', 'value': user_id}], + partition_key=conversation_id, + )) visited = set() cleanup_runs = [] for row in rows: @@ -1502,100 +1543,104 @@ def cleanup_conversation_checkpoints( visited.add(row['id']) if not run_store._is_run_record(row) or row.get('user_id') != user_id: continue - for _ in range(8): - if authorize() is False: - raise RecoveryError(code='not_found', status_code=404) - current = read_revision_run(row['id'], user_id, conversation_id, allow_legacy=True) - updates = { - 'checkpoints_deleted': True, 'execution_lease': None, - 'recovery_blocked_code': 'context_unavailable', - } - intent = None - if not is_legacy_plan(current.get('plan')): - intent = build_output_cleanup_intent(current, retain_committed=retain_committed) - if intent['output_ids'] and output_cleanup is not None and not callable(output_cleanup): - raise RecoveryError( - 'Generated-file cleanup is unavailable. The conversation was not deleted.', - code='output_cleanup_required', status_code=503, - ) - updates['output_cleanup'] = intent - elif current.get('render_output_ids') or 'output_cleanup' in current: - raise CheckpointError('checkpoint_invalid') - store = None - if type(current.get('checkpoint_version')) is int and current['checkpoint_version'] == CHECKPOINT_VERSION: - store = checkpoint_store(current, authorize) - # A lost parent acknowledgment must already have genuine deletion proof. - store.fence(deleted=True, allow_missing=True) - elif intent is not None and intent['output_ids']: - raise CheckpointError('checkpoint_invalid') - fence_publication(current, message_container) - try: - current = _replace(current, updates) - break - except exceptions.CosmosAccessConditionFailedError: - continue - else: - raise CheckpointError() - if current.get('latest_attempt_run_id') and current['latest_attempt_run_id'] not in visited: - rows.append(read_revision_run( - current['latest_attempt_run_id'], user_id, conversation_id, allow_legacy=True, - )) + with _conversation_cleanup_stage(conversation_id, stage='run_fence', run_id=row['id']): + for _ in range(8): + if authorize() is False: + raise RecoveryError(code='not_found', status_code=404) + current = read_revision_run(row['id'], user_id, conversation_id, allow_legacy=True) + updates = { + 'checkpoints_deleted': True, 'execution_lease': None, + 'recovery_blocked_code': 'context_unavailable', + } + intent = None + if not is_legacy_plan(current.get('plan')): + intent = build_output_cleanup_intent(current, retain_committed=retain_committed) + if intent['output_ids'] and output_cleanup is not None and not callable(output_cleanup): + raise RecoveryError( + 'Generated-file cleanup is unavailable. The conversation was not deleted.', + code='output_cleanup_required', status_code=503, + ) + updates['output_cleanup'] = intent + elif current.get('render_output_ids') or 'output_cleanup' in current: + raise CheckpointError('checkpoint_invalid') + store = None + if type(current.get('checkpoint_version')) is int and current['checkpoint_version'] == CHECKPOINT_VERSION: + store = checkpoint_store(current, authorize) + # A lost parent acknowledgment must already have genuine deletion proof. + store.fence(deleted=True, allow_missing=True) + elif intent is not None and intent['output_ids']: + raise CheckpointError('checkpoint_invalid') + fence_publication(current, message_container) + try: + current = _replace(current, updates) + break + except exceptions.CosmosAccessConditionFailedError: + continue + else: + raise CheckpointError() + if current.get('latest_attempt_run_id') and current['latest_attempt_run_id'] not in visited: + rows.append(read_revision_run( + current['latest_attempt_run_id'], user_id, conversation_id, allow_legacy=True, + )) cleanup_runs.append((current, store)) for index, (current, store) in enumerate(cleanup_runs): intent = None if is_legacy_plan(current.get('plan')) else current.get('output_cleanup') if intent is None or not intent['output_ids']: continue - if output_cleanup is None: - # Initialize output services only after every deletion fence and intent is durable. - from functions_orchestration_bootstrap import build_orchestration_cleanup_service - - output_cleanup = build_orchestration_cleanup_service(user_id, conversation_id).enroll_run_cleanup - enrolled = output_cleanup(current['id']) - expected = { - 'run_id': current['id'], 'enrollment_status': 'completed', - 'output_count': len(intent['output_ids']), 'retain_committed': intent['retain_committed'], - } - if ( - type(enrolled) is not dict or enrolled != expected - or type(enrolled.get('output_count')) is not int - or type(enrolled.get('retain_committed')) is not bool - ): - raise RecoveryError( - 'Generated-file cleanup could not be confirmed.', - code='output_cleanup_unconfirmed', status_code=503, - ) - confirmed = read_revision_run(current['id'], user_id, conversation_id) - if ( - confirmed.get('checkpoints_deleted') is not True - or build_output_cleanup_intent( - confirmed, retain_committed=intent['retain_committed'], - ) != {**intent, 'state': 'completed'} - ): - raise RecoveryError( - 'Generated-file cleanup could not be confirmed.', - code='output_cleanup_unconfirmed', status_code=503, - ) + with _conversation_cleanup_stage(conversation_id, stage='output_enrollment', run_id=current['id']): + if output_cleanup is None: + # Initialize output services only after every deletion fence and intent is durable. + from functions_orchestration_bootstrap import build_orchestration_cleanup_service + + output_cleanup = build_orchestration_cleanup_service(user_id, conversation_id).enroll_run_cleanup + enrolled = output_cleanup(current['id']) + expected = { + 'run_id': current['id'], 'enrollment_status': 'completed', + 'output_count': len(intent['output_ids']), 'retain_committed': intent['retain_committed'], + } + if ( + type(enrolled) is not dict or enrolled != expected + or type(enrolled.get('output_count')) is not int + or type(enrolled.get('retain_committed')) is not bool + ): + raise RecoveryError( + 'Generated-file cleanup could not be confirmed.', + code='output_cleanup_unconfirmed', status_code=503, + ) + confirmed = read_revision_run(current['id'], user_id, conversation_id) + if ( + confirmed.get('checkpoints_deleted') is not True + or build_output_cleanup_intent( + confirmed, retain_committed=intent['retain_committed'], + ) != {**intent, 'state': 'completed'} + ): + raise RecoveryError( + 'Generated-file cleanup could not be confirmed.', + code='output_cleanup_unconfirmed', status_code=503, + ) cleanup_runs[index] = (confirmed, store) # Enroll every owned run before any source-cleanup failure can stop this sweep. for current, store in cleanup_runs: - if authorize() is False: - raise RecoveryError(code='not_found', status_code=404) - if _retained_producer_steps(current): - if analysis_cleanup is None: - # Resolve private result I/O only for runs that actually planned Analyze. - from functions_workflow_result_store import ( - delete_orchestration_analysis_results, - fence_orchestration_analysis_result, - ) - analysis_cleanup = delete_orchestration_analysis_results - analysis_fence = analysis_fence or fence_orchestration_analysis_result - if analysis_fence is not None: - for step in _retained_producer_steps(current): - analysis_fence(user_id, conversation_id, current['id'], step['step_id']) + with _conversation_cleanup_stage(conversation_id, stage='retained_result_cleanup', run_id=current['id']): if authorize() is False: raise RecoveryError(code='not_found', status_code=404) - analysis_cleanup(user_id, conversation_id, current['id']) + if _retained_producer_steps(current): + if analysis_cleanup is None: + # Every current-plan step can persist a private result, not just Analyze. + from functions_workflow_result_store import ( + delete_orchestration_analysis_results, + fence_orchestration_analysis_result, + ) + analysis_cleanup = delete_orchestration_analysis_results + analysis_fence = analysis_fence or fence_orchestration_analysis_result + if analysis_fence is not None: + for step in _retained_producer_steps(current): + analysis_fence(user_id, conversation_id, current['id'], step['step_id']) + if authorize() is False: + raise RecoveryError(code='not_found', status_code=404) + analysis_cleanup(user_id, conversation_id, current['id']) if store is not None: - store.delete_payloads() + with _conversation_cleanup_stage(conversation_id, stage='checkpoint_payload_cleanup', run_id=current['id']): + store.delete_payloads() diff --git a/application/single_app/route_backend_conversations.py b/application/single_app/route_backend_conversations.py index 229460664..a4ccb9beb 100644 --- a/application/single_app/route_backend_conversations.py +++ b/application/single_app/route_backend_conversations.py @@ -97,7 +97,10 @@ from swagger_wrapper import swagger_route, get_auth_security from functions_activity_logging import log_conversation_creation, log_conversation_deletion, log_conversation_archival from functions_thoughts import archive_thoughts_for_conversation, delete_thoughts_for_conversation -from functions_orchestration_recovery import cleanup_conversation_checkpoints +from functions_orchestration_recovery import ( + cleanup_conversation_checkpoints, + conversation_cleanup_failure_context, +) from functions_orchestration_artifacts import is_retained_orchestration_file from functions_orchestration_external_configuration import ExternalConfigurationServiceError from functions_orchestration_external_identity import ExternalIdentityServiceError @@ -117,6 +120,7 @@ ) from functions_workflow_result_masking import message_asks_about_workflow_result, message_uses_workflow_result from functions_workflow_result_reader import WorkflowResultUnavailable, workflow_result_error_payload +from functions_workflow_result_store import WorkflowResultIntegrityError, WorkflowResultStorageUnavailableError from utils_cache import invalidate_personal_search_cache @@ -133,6 +137,54 @@ def _enroll_retained_orchestration_outputs(user_id, conversation_id, run_id): return build_orchestration_cleanup_service(user_id, conversation_id).enroll_run_cleanup(run_id) +def _conversation_delete_failure(error, conversation_id, *, stage, is_bulk=False): + """Map failures to trusted browser text and sanitized cleanup telemetry.""" + code = 'conversation_delete_failed' + message = 'The conversation could not be deleted. Please retry deletion.' + status_code = 500 + event = '[CONVERSATION_DELETE] Conversation deletion failed.' + if stage == 'm365_cancellation': + code = 'conversation_pending_actions_cleanup_failed' + message = 'Pending Microsoft 365 actions could not be stopped. The conversation was not deleted.' + status_code = 503 + event = '[CONVERSATION_DELETE] Unable to stop outgoing Microsoft 365 actions.' + elif stage in {'orchestration_cleanup', 'chat_analysis_cleanup'}: + code = 'conversation_execution_cleanup_failed' + message = ( + 'Execution data could not be removed. Please retry deletion. ' + 'If this continues, contact your administrator.' + ) + status_code = 503 + event = ( + '[ORCHESTRATION_RUNS] Conversation recovery cleanup failed.' + if stage == 'orchestration_cleanup' + else '[CONVERSATION_DELETE] Saved analysis cleanup failed.' + ) + if isinstance(error, WorkflowResultIntegrityError): + code = 'conversation_execution_integrity_failed' + message = ( + 'Saved execution data could not be verified for deletion. ' + 'Contact your administrator, then retry deletion.' + ) + elif isinstance(error, (AzureError, WorkflowResultStorageUnavailableError, OutputStorageError)): + code = 'conversation_execution_storage_unavailable' + message = ( + 'Execution data storage is unavailable. Please retry deletion. ' + 'If this continues, ask your administrator to check storage access.' + ) + log_event( + event, + extra={ + **conversation_cleanup_failure_context(error, conversation_id, stage=stage), + 'response_failure': code, + 'response_status_code': status_code, + 'is_bulk_operation': is_bulk, + }, + level=logging.ERROR, exceptionTraceback=False, + ) + return {'error': message, 'code': code}, status_code + + def normalize_chat_type(conversation_item): chat_type = conversation_item.get('chat_type') if chat_type: @@ -1786,22 +1838,19 @@ def delete_conversation(conversation_id): }), 404 except PermissionError: return jsonify({'error': 'Forbidden'}), 403 - except Exception as e: - return jsonify({ - "error": str(e) - }), 500 + except Exception as error: + payload, status_code = _conversation_delete_failure( + error, conversation_id, stage='authorization', + ) + return jsonify(payload), status_code try: cancel_m365_conversation_deliveries(conversation_id) except Exception as error: - log_event( - "[CONVERSATION_DELETE] Unable to stop outgoing Microsoft 365 actions.", - extra={"conversation_id": conversation_id, "exception_type": type(error).__name__}, - level=logging.ERROR, + payload, status_code = _conversation_delete_failure( + error, conversation_id, stage='m365_cancellation', ) - return jsonify({ - "error": "Pending Microsoft 365 actions could not be stopped. The conversation was not deleted.", - }), 503 + return jsonify(payload), status_code try: cleanup_conversation_checkpoints( @@ -1812,12 +1861,11 @@ def delete_conversation(conversation_id): output_cleanup=partial(_enroll_retained_orchestration_outputs, user_id, conversation_id), retain_committed=archiving_enabled, ) - except Exception as exc: - log_event( - '[ORCHESTRATION_RUNS] Conversation recovery cleanup failed.', - extra={'conversation_id': conversation_id, 'error_type': type(exc).__name__}, level=logging.ERROR, + except Exception as error: + payload, status_code = _conversation_delete_failure( + error, conversation_id, stage='orchestration_cleanup', ) - return jsonify({'error': 'Execution data could not be removed. Please retry deletion.'}), 503 + return jsonify(payload), status_code if archiving_enabled: archived_item = dict(conversation_item) @@ -1835,11 +1883,17 @@ def delete_conversation(conversation_id): ) message_query = f"SELECT * FROM c WHERE c.conversation_id = '{conversation_id}'" - results = list(cosmos_messages_container.query_items( - query=message_query, - partition_key=conversation_id - )) - cleanup_chat_analysis_conversation(conversation_id, conversation_item.get('user_id'), results) + try: + results = list(cosmos_messages_container.query_items( + query=message_query, + partition_key=conversation_id + )) + cleanup_chat_analysis_conversation(conversation_id, conversation_item.get('user_id'), results) + except Exception as error: + payload, status_code = _conversation_delete_failure( + error, conversation_id, stage='chat_analysis_cleanup', + ) + return jsonify(payload), status_code direct_messages = [message for message in results if not _is_retained_orchestration_file(message)] direct_message_ids = {message['id'] for message in direct_messages} @@ -1906,10 +1960,11 @@ def delete_conversation(conversation_id): ) bump_conversation_cache_version(user_id, reason="conversation_deleted") # TODO: Delete any facts that were stored with this conversation. - except Exception as e: - return jsonify({ - "error": str(e) - }), 500 + except Exception as error: + payload, status_code = _conversation_delete_failure( + error, conversation_id, stage='conversation_delete', + ) + return jsonify(payload), status_code return jsonify({ "success": True @@ -1938,17 +1993,27 @@ def delete_multiple_conversations(): success_count = 0 failed_ids = [] + failures = [] for conversation_id in conversation_ids: + stage = 'authorization' try: # Verify the conversation exists and belongs to the user try: conversation_item = _authorize_personal_conversation_read(user_id, conversation_id) except (LookupError, PermissionError): failed_ids.append(conversation_id) + failures.append({ + 'conversation_id': conversation_id, + 'error': 'Conversation not found or access denied.', + 'code': 'conversation_unavailable', + 'status_code': 404, + }) continue + stage = 'm365_cancellation' cancel_m365_conversation_deliveries(conversation_id) + stage = 'orchestration_cleanup' cleanup_conversation_checkpoints( conversation_id, user_id, lambda: _authorize_personal_conversation_read(user_id, conversation_id), @@ -1959,6 +2024,7 @@ def delete_multiple_conversations(): ) # Archive if enabled + stage = 'conversation_archive' if archiving_enabled: archived_item = dict(conversation_item) archived_item["archived_at"] = datetime.utcnow().isoformat() @@ -1975,6 +2041,7 @@ def delete_multiple_conversations(): ) # Get and archive messages if enabled + stage = 'chat_analysis_cleanup' message_query = f"SELECT * FROM c WHERE c.conversation_id = '{conversation_id}'" messages = list(cosmos_messages_container.query_items( query=message_query, @@ -1986,6 +2053,7 @@ def delete_multiple_conversations(): ] direct_message_ids = {message['id'] for message in direct_messages} + stage = 'message_cleanup' if not archiving_enabled: delete_blob_backed_chat_message_files(direct_messages, conversation=conversation_item) @@ -1999,6 +2067,7 @@ def delete_multiple_conversations(): cosmos_messages_container.delete_item(message['id'], partition_key=conversation_id) # Archive/delete thoughts for conversation + stage = 'thoughts_cleanup' if archiving_enabled: archive_thoughts_for_conversation(conversation_id, user_id) else: @@ -2017,6 +2086,7 @@ def delete_multiple_conversations(): ) # Delete the conversation + stage = 'conversation_delete' cosmos_conversations_container.delete_item( item=conversation_id, partition_key=conversation_id @@ -2025,12 +2095,15 @@ def delete_multiple_conversations(): success_count += 1 except Exception as error: - log_event( - "[CONVERSATION_DELETE] Conversation deletion failed.", - extra={"conversation_id": conversation_id, "exception_type": type(error).__name__}, - level=logging.ERROR, + payload, status_code = _conversation_delete_failure( + error, conversation_id, stage=stage, is_bulk=True, ) failed_ids.append(conversation_id) + failures.append({ + 'conversation_id': conversation_id, + **payload, + 'status_code': status_code, + }) if success_count: bump_conversation_cache_version(user_id, reason="conversations_bulk_deleted") @@ -2038,7 +2111,8 @@ def delete_multiple_conversations(): return jsonify({ "success": True, "deleted_count": success_count, - "failed_ids": failed_ids + "failed_ids": failed_ids, + "failures": failures, }), 200 @bp.route('/api/conversations//pin', methods=['POST']) diff --git a/docs/explanation/fixes/CONVERSATION_DELETE_HNS_CLEANUP_FIX.md b/docs/explanation/fixes/CONVERSATION_DELETE_HNS_CLEANUP_FIX.md new file mode 100644 index 000000000..07c3ae6fe --- /dev/null +++ b/docs/explanation/fixes/CONVERSATION_DELETE_HNS_CLEANUP_FIX.md @@ -0,0 +1,162 @@ +# Conversation deletion and HNS cleanup (v0.261.306) + +Fixed in version: **0.261.306** + +Application version tracking: `application/single_app/config.py`. +Depends on the shared HNS cleanup introduced in **0.261.305** and merged in +[#1726](https://github.com/microsoft/simplechat/pull/1726). + +## Issue and root cause + +Deleting a conversation could return HTTP 503 with +"Execution data could not be removed. Please retry deletion." +Production telemetry identified `WorkflowResultIntegrityError` in all 11 +investigated single-conversation failures. The latest correlated request listed +Blob Storage successfully, then failed before any Blob deletion. This was not +evidence of a Microsoft 365 sign-in failure. + +The storage account uses hierarchical namespace (HNS). Flat Blob listings can +contain explicitly marked directory objects alongside private execution-result +files. The old result-file-only validation rejected these legitimate directories. +Conversation deletion and workflow deletion both use +`WorkflowResultStore._delete_scoped_blobs()`, so the shared fix also repairs +conversation cleanup. + +The [workflow deletion fix](WORKFLOW_DELETE_HNS_DIRECTORY_FIX.md) documents +read-only verification of the HNS directory markers and the storage safeguards. +No production conversation, workflow, or Blob was manually deleted during this +investigation. + +## Technical changes + +### Scoped result cleanup + +The integrated storage fix handles both orchestration results and chat-analysis +results. Explicit directory identity, a known zero size, and the allowed hashed +path depth are required. A trailing slash alone is not proof of a directory. +Result files retain their path and identity-metadata validation. + +Files are deleted before directories. Directories are freshly verified and +deleted deepest-first with their current ETag and a non-recursive conditional +delete. Missing objects are retryable; unexpected objects, changed types, +conditional-delete races, and storage failures still block completion. +Scope-root and ancestor directories are not swept. + +Current-plan results are cleaned for every retained producer step, not only an +Analyze step. Cleanup does not rerun a workflow, retrieve original sources, or +require a fresh Microsoft 365 sign-in. The application still needs access to its +Cosmos DB and storage resources. + +### Safe diagnostics and browser responses + +`application/single_app/functions_orchestration_recovery.py` adds scoped failure +logging around conversation fencing, run discovery, run fencing, generated-output +enrollment, retained-result cleanup, and checkpoint-payload cleanup. It rethrows +the original exception. + +Diagnostics retain the exception class, controlled stage, hashed conversation +and run correlations, recognized application failure codes, and valid HTTP error +statuses. They exclude exception text, provider payloads, raw identifiers, and +tracebacks. Provider-specific codes, malformed application codes, boolean +statuses, and statuses outside 400-599 are omitted. + +`application/single_app/route_backend_conversations.py` returns trusted static +messages instead of provider errors: + +| Failure | Client code | Status | +| --- | --- | --- | +| Private execution-result integrity validation | `conversation_execution_integrity_failed` | 503 | +| Execution-data storage failure | `conversation_execution_storage_unavailable` | 503 | +| Other required execution cleanup | `conversation_execution_cleanup_failed` | 503 | +| Pending Microsoft 365 action cancellation | `conversation_pending_actions_cleanup_failed` | 503 | +| Other caught conversation deletion failure | `conversation_delete_failed` | 500 | + +Single deletion now also handles chat-analysis cleanup failures as safe JSON +errors. Bulk deletion preserves `success`, `deleted_count`, and `failed_ids`, +and adds a parallel `failures` array: + +```json +{ + "conversation_id": "requested-conversation", + "error": "Execution data storage is unavailable. Please retry deletion. If this continues, ask your administrator to check storage access.", + "code": "conversation_execution_storage_unavailable", + "status_code": 503 +} +``` + +The bulk endpoint still returns HTTP 200 for its aggregate result; each failure +contains its own status. Missing and foreign conversations receive identical +`conversation_unavailable` details with status 404, without starting cleanup. + +No routes, settings, authentication policies, or browser assets are added. + +### Cleanup boundaries and retries + +Conversation history is not purged while required execution-data cleanup remains +unconfirmed. Deletion fences and cleared execution leases intentionally persist +after a failure so late workers cannot republish results. A hidden +`assistant_artifact` publication tombstone may also be created; it is a fence, not +a new user-visible reply. + +Cleanup is not an atomic transaction across every storage object. Verified result +files or checkpoint payloads may already have been removed before a later failure. +The conversation and ordinary history remain available for a deletion retry. +Retained generated files keep their existing enrollment, conditional cleanup, +grace period, and archiving rules; direct message cleanup does not bypass them. + +## Recovery after deployment + +Deploy **0.261.306** or later for the shared storage fix and conversation-specific +diagnostics. The underlying valid-directory deletion repair is already present +in **0.261.305**. + +Retry **Delete** on the stuck conversation. A valid HNS-directory case requires +no manual data repair and no workflow restart. Failed cleanup fences remain in +effect while the retry finishes. + +If deletion still fails, correlate the failed request with +`[ORCHESTRATION_RUNS] Conversation cleanup stage failed.` and the route's cleanup +event. Inspect `sc_stage`, `sc_error_type`, `sc_failure_code` when present, and +the hashed conversation/run fields. Check application storage access for a +storage error; investigate unexpected objects for an integrity error. + +Do not bypass integrity validation, recursively delete broad prefixes, or remove +conversation history before cleanup is confirmed. + +This follow-up has not been deployed as part of the investigation. + +## Testing and validation + +`functional_tests/test_conversation_delete_hns_directory_cleanup.py` exercises +real authenticated single and bulk routes, recovery, and private result cleanup +with deterministic external storage doubles. Its 42 cases cover ordinary and +HNS listings, non-Analyze retained producers, committed and interrupted chat +analysis, invalid objects, storage failures, retry, ownership, partial bulk +results, generated-output enrollment, and telemetry sanitization. + +`functional_tests/test_conversations_read_ownership_authorization.py` now restores +its fake import dependencies after requests, import errors, and app setup errors. +This prevents the lightweight read fixture from replacing authentication modules +used by subsequent real cleanup tests. + +The combined result-store, deletion enrollment/bootstrap/output lifecycle, +initial-claim recovery, conversation ownership/deletion, saved-analysis, and +Microsoft 365 lifecycle selection passed **541 tests and 112 subtests**. +The reverse-order conversation selection passed **52 tests**. All three route +policy scripts passed **26 checks**. + +Local validation uses an ignored virtual environment with repository-pinned +Flask/Werkzeug and a compatible OpenSSL package. No global Python installation +or application dependency manifest was changed. Tests do not mutate production +data. + +## Before and after + +Before, legitimate HNS directories blocked conversation deletion and the generic +toast concealed the failing cleanup stage. After, verified directories are +removed safely, failed cleanup remains retryable without purging history, and +administrators can distinguish integrity, storage, and enrollment failures +without exposing private error details. + +See [Operate SimpleChat day to day](../../guides/admin-operate-simplechat.md) +for operational troubleshooting. diff --git a/docs/guides/admin-operate-simplechat.md b/docs/guides/admin-operate-simplechat.md index 3e8f6009c..71d8f33c3 100644 --- a/docs/guides/admin-operate-simplechat.md +++ b/docs/guides/admin-operate-simplechat.md @@ -80,6 +80,7 @@ Start with [Troubleshooting]({{ '/troubleshooting/' | relative_url }}). Query fa | You cannot tell which backend call failed | The investigation started from settings instead of telemetry | Query Application Insights failed requests, capture `operation_Id`, then pivot to exceptions. | | Horizontal scale causes inconsistent behavior | Shared cache or session assumptions were not ready | Review Redis-backed cache guidance in [Scale settings]({{ '/admin/scale/' | relative_url }}) before serious scale-out. | | File processing history grows without cleanup | File-processing logs are enabled without a retention routine | Use [Logging settings]({{ '/admin/operations/' | relative_url }}) cleanup controls and set an operating cadence. | +| Conversation deletion reports that execution data could not be removed | Required private-result cleanup failed; this does not by itself mean Microsoft 365 sign-in failed | Deploy 0.261.306 or later and retry **Delete**. If it persists, correlate the failed request with `sc_stage` and `sc_error_type` in Application Insights. Check storage access for storage failures and unexpected stored objects for integrity failures; do not bypass validation or purge history first. | ## Related diff --git a/functional_tests/test_conversation_delete_hns_directory_cleanup.py b/functional_tests/test_conversation_delete_hns_directory_cleanup.py new file mode 100644 index 000000000..e072799ac --- /dev/null +++ b/functional_tests/test_conversation_delete_hns_directory_cleanup.py @@ -0,0 +1,402 @@ +# test_conversation_delete_hns_directory_cleanup.py +""" +Conversation deletion with real checkpoint and private-result cleanup. +Version: 0.261.306 +Implemented in: 0.261.306 + +Authenticated single/bulk routes exercise hierarchical namespace directory +listings, ordinary Blob listings, retained generated-file enrollment, cleanup +retries, ownership boundaries, and sanitized Application Insights diagnostics. +Only external storage I/O is doubled; no model, source read, or deployment runs. +""" + +import hashlib +import importlib +from copy import deepcopy +from unittest.mock import Mock + +import pytest +from azure.core.exceptions import ResourceModifiedError, ServiceRequestError + +from test_orchestration_cleanup_bootstrap import cleanup_root # noqa: F401 +from test_orchestration_output_cleanup import cleanup_lifecycle, production_modules # noqa: F401 +from test_orchestration_output_deletion_pipeline import ( + advance_cleanup_grace, + assert_deletion_response, + cleanup_tick, + deletion_runtime, # noqa: F401 + request_deletion, +) + + +CONVERSATION_ID = "conversation-1" +RUN_ID = "run-1" +PRIVATE_DETAIL = "private-provider-detail https://storage.example.test/?sig=private-token" + + +@pytest.fixture +def result_deletion_runtime(deletion_runtime, monkeypatch): + runtime, world = deletion_runtime, deletion_runtime.world + results = importlib.import_module("functions_workflow_result_store") + recovery = importlib.import_module("functions_orchestration_recovery") + telemetry = importlib.import_module("functions_appinsights") + store = results.WorkflowResultStore( + container=world.results.container, + blob_client=world.results.blobs, + blob_container_name="private-results", + ) + monkeypatch.setattr(results, "_configured_result_store", lambda *args, **kwargs: store) + logger = Mock() + monkeypatch.setattr(telemetry, "get_appinsights_logger", lambda: logger) + record_error = runtime.routes.log_event + + def record_event(message, **kwargs): + record_error(message, **kwargs) + telemetry.log_event(message, **kwargs) + + monkeypatch.setattr(runtime.routes, "log_event", record_event) + monkeypatch.setattr(recovery, "log_event", record_event) + runtime.private_store = store + runtime.private_blobs = world.results.blobs + runtime.result_module = results + runtime.recovery = recovery + runtime.logger = logger + return runtime + + +def add_hns_directories(runtime): + """Place explicit directory entries before their child result files.""" + blobs = runtime.private_blobs + files = list(blobs.records) + directories = set() + blobs.hierarchical = True + for container, name in files: + parts = name.split("/") + scope_depth = 6 if parts[0] == "orchestration-analysis-results" else 5 + for depth in range(scope_depth + 1, len(parts)): + key = (container, "/".join(parts[:depth])) + if key in directories: + continue + blobs.add_directory( + key[1], container=container, metadata={"HDI_ISFOLDER": "TrUe"}, + ) + directories.add(key) + return directories + + +def seed_chat_result(runtime, *, interrupted=False): + runtime.private_store.save_chat( + "owner", CONVERSATION_ID, "analysis-message", {"findings": ["retained chat analysis"]}, + ) + if interrupted: + message = { + "id": "analysis-request", "conversation_id": CONVERSATION_ID, + "role": "user", "user_id": "owner", "content": "Analyze my document.", + "metadata": {"analysis_attempt_message_id": "analysis-message"}, + } + else: + message = { + "id": "analysis-message", "conversation_id": CONVERSATION_ID, + "role": "assistant", "content": "Historical analysis.", + "metadata": {"document_action": {"type": "analyze"}}, + } + runtime.world.messages.create_item(message) + + +def cleanup_failure(response, route): + body = response.get_json() + return body if route == "single" else body["failures"][0] + + +def assert_history_preserved(runtime, messages): + conversation = runtime.world.conversations.read_item(CONVERSATION_ID, CONVERSATION_ID) + run = runtime.world.runs.read_item(RUN_ID, CONVERSATION_ID) + guard = runtime.world.cleanup_guards.read_item("checkpoint:lifecycle", RUN_ID) + assert conversation["orchestration_deleted"] is True + assert run["checkpoints_deleted"] is True and run["execution_lease"] is None + assert guard["deleted"] is True and guard["token"] is None + publication_key = (CONVERSATION_ID, runtime.recovery._publication_id(RUN_ID)) + preserved = { + key: message for key, message in runtime.world.messages.items.items() + if key != publication_key + } + original = {key: message for key, message in messages.items() if key != publication_key} + assert preserved == original + publication = runtime.world.messages.items[publication_key] + assert publication["user_id"] == "owner" and publication["run_id"] == RUN_ID + assert publication["role"] == "assistant_artifact" and publication["content"] == "" + assert publication["token"] is None + assert publication["metadata"]["orchestration_publication_guard"] is True + + +def logged_events(runtime, *, stage=None): + events = [call.kwargs["extra"] for call in runtime.logger.log.call_args_list] + return events if stage is None else [event for event in events if event.get("sc_stage") == stage] + + +@pytest.mark.parametrize("route", ["single", "bulk"]) +@pytest.mark.parametrize("hierarchical", [False, True]) +@pytest.mark.parametrize("without_analyze", [False, True]) +def test_owned_results_and_directories_are_removed_before_history( + result_deletion_runtime, route, hierarchical, without_analyze, +): + runtime, world = result_deletion_runtime, result_deletion_runtime.world + output = world.run(world.prepare()) + if without_analyze: + run = world.runs.read_item(RUN_ID, CONVERSATION_ID) + run["plan"]["steps"] = [ + step for step in run["plan"]["steps"] if step["capability_id"] != "document_analyze" + ] + world.runs.upsert_item(run) + for step in run["plan"]["steps"]: + runtime.private_store.save_orchestration( + "owner", CONVERSATION_ID, RUN_ID, step["step_id"], {"reply": "Retained non-Analyze result."}, + ) + runtime.private_store.save_chat( + "another-owner", "another-conversation", "another-message", {"findings": ["outside scope"]}, + ) + directories = add_hns_directories(runtime) if hierarchical else set() + foreign = { + key: deepcopy(row) for key, row in runtime.private_blobs.records.items() + if key[1].startswith("chat-analysis-results/") + } + owned_keys = set(runtime.private_blobs.records) - set(foreign) + response = request_deletion(runtime, route) + assert_deletion_response(response, route, succeeded=True) + body = response.get_json() + if route == "bulk": + assert body["failures"] == [] + remaining = runtime.private_blobs.records + owned_directories = directories & owned_keys + assert not owned_keys.intersection(remaining) + assert all(remaining[key] == row for key, row in foreign.items()) + assert owned_directories.issubset(runtime.private_blobs.deletes) + assert runtime.private_blobs.deletes + deleted_files = [key for key in runtime.private_blobs.deletes if key not in directories] + assert runtime.private_blobs.deletes[:len(deleted_files)] == deleted_files + assert (CONVERSATION_ID, CONVERSATION_ID) not in world.conversations.items + parent = world.runs.read_item(RUN_ID, CONVERSATION_ID) + retained = world.runs.read_item(output["output_id"], CONVERSATION_ID) + assert parent["output_cleanup"]["state"] == "completed" + assert retained["state"] == "cancelled" and retained["cleanup_pending"] is True + assert world.blobs.deletes == 0 and world.blobs.data + advance_cleanup_grace(world, [output]) + cleaned = cleanup_tick(runtime) + assert cleaned["ok"] is True and cleaned["counts"]["outputs_processed"] == 1 + assert not world.blobs.data and world.blobs.deletes == 1 + assert len(world.render_calls) == 1 + + +@pytest.mark.parametrize("route", ["single", "bulk"]) +@pytest.mark.parametrize("interrupted", [False, True]) +def test_chat_analysis_directories_are_cleaned_for_committed_and_interrupted_results( + result_deletion_runtime, route, interrupted, +): + runtime = result_deletion_runtime + seed_chat_result(runtime, interrupted=interrupted) + directories = add_hns_directories(runtime) + response = request_deletion(runtime, route) + assert_deletion_response(response, route, succeeded=True) + assert not runtime.private_blobs.records + assert directories.issubset(runtime.private_blobs.deletes) + assert any(entry["prefix"].startswith("chat-analysis-results/") for entry in runtime.private_blobs.lists) + assert not runtime.world.messages.items + + +@pytest.mark.parametrize("route", ["single", "bulk"]) +@pytest.mark.parametrize("scope", ["orchestration", "chat"]) +@pytest.mark.parametrize("fault", ["stray_file", "unmarked_directory", "nonempty_directory", "misnamed_directory"]) +def test_real_unexpected_objects_keep_conversation_deletion_fail_closed( + result_deletion_runtime, route, scope, fault, +): + runtime = result_deletion_runtime + if scope == "chat": + seed_chat_result(runtime) + directories = add_hns_directories(runtime) + family = f"{scope}-analysis-results/" + directory = next(key for key in directories if key[1].startswith(family)) + if fault == "unmarked_directory": + runtime.private_blobs.records[directory]["metadata"] = {} + elif fault == "nonempty_directory": + runtime.private_blobs.records[directory]["data"] = b"Not a directory." + else: + bad_key = ( + directory[0], f"{directory[1].rsplit('/', 1)[0]}/" + f"{'unexpected.txt' if fault == 'stray_file' else 'unexpected-directory'}", + ) + bad_row = deepcopy(runtime.private_blobs.records[directory]) + if fault == "stray_file": + bad_row.update(data=b"Private result bytes.", metadata={}, directory=False) + runtime.private_blobs.records = {bad_key: bad_row, **runtime.private_blobs.records} + messages = deepcopy(runtime.world.messages.items) + response = request_deletion(runtime, route) + assert_deletion_response(response, route, succeeded=False) + detail = cleanup_failure(response, route) + assert detail["code"] == "conversation_execution_integrity_failed" + assert "administrator" in detail["error"] + assert_history_preserved(runtime, messages) + stage = "retained_result_cleanup" if scope == "orchestration" else "chat_analysis_cleanup" + events = logged_events(runtime, stage=stage) + assert events and events[-1]["sc_error_type"] == "WorkflowResultIntegrityError" + + +@pytest.mark.parametrize("route", ["single", "bulk"]) +@pytest.mark.parametrize("scope", ["orchestration", "chat"]) +@pytest.mark.parametrize("operation", ["listing", "conditional_delete"]) +def test_storage_failure_preserves_fences_and_a_retry_completes( + result_deletion_runtime, route, scope, operation, monkeypatch, +): + runtime = result_deletion_runtime + if scope == "chat": + seed_chat_result(runtime) + add_hns_directories(runtime) + blobs = runtime.private_blobs + original_container = blobs.get_container_client + original_blob = blobs.get_blob_client + family = f"{scope}-analysis-results/" + + def container_client(container): + client = original_container(container) + listing = client.list_blobs + + def fail_listing(*, name_starts_with, include): + if name_starts_with.startswith(family): + raise ServiceRequestError(PRIVATE_DETAIL) + return listing(name_starts_with=name_starts_with, include=include) + + monkeypatch.setattr(client, "list_blobs", fail_listing) + return client + + def blob_client(*, container, blob): + client = original_blob(container=container, blob=blob) + if blob.startswith(family): + monkeypatch.setattr(client, "delete_blob", Mock(side_effect=ResourceModifiedError(PRIVATE_DETAIL))) + return client + + with monkeypatch.context() as unavailable: + unavailable.setattr( + blobs, "get_container_client" if operation == "listing" else "get_blob_client", + container_client if operation == "listing" else blob_client, + ) + messages = deepcopy(runtime.world.messages.items) + failed = request_deletion(runtime, route) + assert_deletion_response(failed, route, succeeded=False) + detail = cleanup_failure(failed, route) + assert detail["code"] == "conversation_execution_storage_unavailable" + assert_history_preserved(runtime, messages) + body = failed.get_data(as_text=True) + events = logged_events(runtime) + assert PRIVATE_DETAIL not in body and "private-token" not in repr(events) + retried = request_deletion(runtime, route) + assert_deletion_response(retried, route, succeeded=True) + assert not runtime.private_blobs.records + assert not runtime.world.messages.items + + +@pytest.mark.parametrize("route", ["single", "bulk"]) +def test_unauthorized_actor_cannot_start_private_result_cleanup(result_deletion_runtime, route): + runtime, world = result_deletion_runtime, result_deletion_runtime.world + add_hns_directories(runtime) + before = deepcopy(( + world.conversations.items, world.runs.items, world.cleanup_guards.items, + world.messages.items, runtime.private_blobs.records, world.results.container.items, + )) + serializer = runtime.app.session_interface.get_signing_serializer(runtime.app) + cookie = serializer.dumps({"user": {"oid": "another-owner", "roles": ["Admin"]}}) + runtime.client.set_cookie(runtime.app.config["SESSION_COOKIE_NAME"], cookie) + response = request_deletion(runtime, route) + body = response.get_json() + if route == "single": + assert response.status_code == 403 + else: + assert response.status_code == 200 and body["failed_ids"] == [CONVERSATION_ID] + assert body["failures"][0]["code"] == "conversation_unavailable" + after = ( + world.conversations.items, world.runs.items, world.cleanup_guards.items, + world.messages.items, runtime.private_blobs.records, world.results.container.items, + ) + assert before == after + assert not runtime.private_blobs.lists and not runtime.private_blobs.deletes + + +def test_bulk_partial_failure_keeps_existing_fields_and_adds_safe_details( + result_deletion_runtime, +): + runtime, world = result_deletion_runtime, result_deletion_runtime.world + add_hns_directories(runtime) + runtime.private_blobs.list_error = ServiceRequestError(PRIVATE_DETAIL) + for conversation_id, owner in (("other-owned", "owner"), ("foreign", "another-owner")): + world.conversations.create_item({"id": conversation_id, "user_id": owner}) + world.messages.create_item({ + "id": f"{conversation_id}-message", "conversation_id": conversation_id, + "role": "user", "content": "Historical message.", + }) + foreign_messages = deepcopy({ + key: value for key, value in world.messages.items.items() if key[0] == "foreign" + }) + response = runtime.client.post("/api/delete_multiple_conversations", json={ + "conversation_ids": [CONVERSATION_ID, "other-owned", "foreign"], + }) + body = response.get_json() + assert response.status_code == 200 and body["success"] is True + assert body["deleted_count"] == 1 and body["failed_ids"] == [CONVERSATION_ID, "foreign"] + assert [item["conversation_id"] for item in body["failures"]] == body["failed_ids"] + assert body["failures"][0]["code"] == "conversation_execution_storage_unavailable" + assert body["failures"][0]["status_code"] == 503 + assert body["failures"][1]["code"] == "conversation_unavailable" + assert (CONVERSATION_ID, CONVERSATION_ID) in world.conversations.items + assert ("other-owned", "other-owned") not in world.conversations.items + assert ("foreign", "foreign") in world.conversations.items + assert all(world.messages.items[key] == row for key, row in foreign_messages.items()) + assert PRIVATE_DETAIL not in repr(body) and "private-token" not in repr(logged_events(runtime)) + + +@pytest.mark.parametrize("route", ["single", "bulk"]) +def test_cleanup_telemetry_retains_safe_code_status_stage_and_hashed_run( + result_deletion_runtime, route, monkeypatch, +): + runtime, world = result_deletion_runtime, result_deletion_runtime.world + world.run(world.prepare()) + failure = runtime.recovery.RecoveryError( + PRIVATE_DETAIL, code="output_cleanup_required", status_code=503, + ) + monkeypatch.setattr( + runtime.root, "build_orchestration_cleanup_service", Mock(side_effect=failure), + ) + response = request_deletion(runtime, route) + assert_deletion_response(response, route, succeeded=False) + enrollment = logged_events(runtime, stage="output_enrollment") + boundary = logged_events(runtime, stage="orchestration_cleanup") + assert len(enrollment) == len(boundary) == 1 + event = enrollment[0] + assert event["sc_error_type"] == "RecoveryError" + assert event["sc_failure_code"] == "output_cleanup_required" and event["sc_status_code"] == 503 + assert event["sc_conversation_id_hash"] == hashlib.sha256(CONVERSATION_ID.encode()).hexdigest() + assert event["sc_run_id_hash"] == hashlib.sha256(RUN_ID.encode()).hexdigest() + assert boundary[0]["sc_response_status_code"] == 503 + assert boundary[0]["sc_is_bulk_operation"] is (route == "bulk") + assert "sc_conversation_id" not in event and "sc_run_id" not in event + assert "private-token" not in repr(logged_events(runtime)) + assert all(call.kwargs["exc_info"] is False for call in runtime.logger.log.call_args_list) + runtime.logger.exception.assert_not_called() + body = response.get_data(as_text=True) + assert PRIVATE_DETAIL not in body + + +def test_cleanup_diagnostics_drop_provider_codes_invalid_status_and_malformed_codes( + result_deletion_runtime, +): + runtime = result_deletion_runtime + provider_error = ServiceRequestError(PRIVATE_DETAIL) + provider_error.code = "provider_private_detail" + provider_error.status_code = True + provider = runtime.recovery.conversation_cleanup_failure_context( + provider_error, CONVERSATION_ID, stage="retained_result_cleanup", run_id=RUN_ID, + ) + invalid = runtime.recovery.conversation_cleanup_failure_context( + runtime.recovery.RecoveryError(PRIVATE_DETAIL, code=PRIVATE_DETAIL, status_code=700), + CONVERSATION_ID, stage="retained_result_cleanup", + ) + assert "failure_code" not in provider and "status_code" not in provider + assert "failure_code" not in invalid and "status_code" not in invalid + assert PRIVATE_DETAIL not in repr((provider, invalid)) diff --git a/functional_tests/test_conversations_read_ownership_authorization.py b/functional_tests/test_conversations_read_ownership_authorization.py index a641d36a1..59ec7bb91 100644 --- a/functional_tests/test_conversations_read_ownership_authorization.py +++ b/functional_tests/test_conversations_read_ownership_authorization.py @@ -2,8 +2,9 @@ # test_conversations_read_ownership_authorization.py """ Functional test for personal conversation read authorization hardening. -Version: 0.250.101 +Version: 0.261.306 Implemented in: 0.241.011; 0.241.022; 0.241.032; 0.250.033; 0.250.035; 0.250.074; 0.250.101 +Deletion diagnostics import compatibility and scoped test imports added in: 0.261.306 This test ensures authenticated users can only read messages and images from their own personal conversations, and that foreign conversation reads fail with @@ -18,6 +19,7 @@ import os import sys import types +from unittest.mock import patch from flask import Flask, jsonify import werkzeug @@ -130,8 +132,8 @@ def _passthrough_decorator(*args, **kwargs): return lambda func: func -def _install_route_import_stubs(): - """Install lightweight module stubs so the route module imports in isolation.""" +def _route_import_stubs(): + """Build lightweight dependencies without mutating process-wide imports.""" stub_modules = {} config_module = types.ModuleType('config') @@ -157,6 +159,16 @@ def _install_route_import_stubs(): collaboration_module.serialize_collaboration_conversation = lambda item, *args, **kwargs: item stub_modules['functions_collaboration'] = collaboration_module + generated_documents_module = types.ModuleType('functions_collaboration_generated_documents') + generated_documents_module.collect_generated_documents = lambda *args, **kwargs: [] + generated_documents_module.can_download_generated_document = lambda *args, **kwargs: False + + def unexpected_generated_download(*args, **kwargs): + raise AssertionError('Conversation read tests unexpectedly entered generated-document downloads.') + + generated_documents_module.authorize_generated_document_download = unexpected_generated_download + stub_modules['functions_collaboration_generated_documents'] = generated_documents_module + auth_module = types.ModuleType('functions_authentication') auth_module.login_required = _passthrough_decorator auth_module.user_required = _passthrough_decorator @@ -229,6 +241,7 @@ def clear_conversation_unread(item): orchestration_recovery_module = types.ModuleType('functions_orchestration_recovery') orchestration_recovery_module.cleanup_conversation_checkpoints = lambda *args, **kwargs: None + orchestration_recovery_module.conversation_cleanup_failure_context = lambda *args, **kwargs: {} stub_modules['functions_orchestration_recovery'] = orchestration_recovery_module orchestration_artifacts_module = types.ModuleType('functions_orchestration_artifacts') @@ -291,6 +304,7 @@ def clear_conversation_unread(item): stub_modules['functions_debug'] = debug_module documents_module = types.ModuleType('functions_documents') + documents_module.build_document_download_response = lambda *args, **kwargs: None documents_module.delete_chat_upload_workspace_documents_for_conversation = lambda *args, **kwargs: None documents_module.serialize_chat_upload_workspace_documents_for_conversation = lambda *args, **kwargs: [] stub_modules['functions_documents'] = documents_module @@ -340,52 +354,96 @@ def clear_conversation_unread(item): utils_cache_module.invalidate_personal_search_cache = lambda *args, **kwargs: None stub_modules['utils_cache'] = utils_cache_module - for module_name, module in stub_modules.items(): - sys.modules[module_name] = module + return stub_modules def _load_route_backend_conversations_module(): - """Import the route module after installing lightweight dependency stubs.""" - _install_route_import_stubs() - if 'route_backend_conversations' in sys.modules: - del sys.modules['route_backend_conversations'] - return importlib.import_module('route_backend_conversations') + """Keep fake imports scoped to the request tests, including import failures.""" + imports = patch.dict(sys.modules, _route_import_stubs()) + imports.start() + try: + sys.modules.pop('route_backend_conversations', None) + module = importlib.import_module('route_backend_conversations') + except BaseException: + imports.stop() + raise + return module, imports.stop def build_test_app(test_user_id, conversation_items, message_items, blob_items=None): """Register the conversation routes with fake auth and fake Cosmos containers.""" - route_backend_conversations = _load_route_backend_conversations_module() - - conversation_container = FakeConversationContainer(conversation_items) - message_container = FakeMessageContainer(message_items) - cache_bumps = [] - - route_backend_conversations.cosmos_conversations_container = conversation_container - route_backend_conversations.cosmos_messages_container = message_container - route_backend_conversations.login_required = lambda func: func - route_backend_conversations.user_required = lambda func: func - route_backend_conversations.swagger_route = lambda **kwargs: (lambda func: func) - route_backend_conversations.get_auth_security = lambda: {} - route_backend_conversations.get_current_user_id = lambda: test_user_id - route_backend_conversations.debug_print = lambda *args, **kwargs: None - route_backend_conversations.filter_assistant_artifact_items = lambda items: items - route_backend_conversations.CosmosResourceNotFoundError = DummyNotFoundError - route_backend_conversations.bump_conversation_cache_version = ( - lambda user_id, reason=None: cache_bumps.append((user_id, reason)) or len(cache_bumps) - ) - route_backend_conversations.mark_chat_response_notifications_read_for_conversation = lambda *args, **kwargs: 0 - route_backend_conversations.CLIENTS = { - 'storage_account_office_docs_client': FakeBlobServiceClient(blob_items), - } + route_backend_conversations, restore = _load_route_backend_conversations_module() + try: + conversation_container = FakeConversationContainer(conversation_items) + message_container = FakeMessageContainer(message_items) + cache_bumps = [] + + route_backend_conversations.cosmos_conversations_container = conversation_container + route_backend_conversations.cosmos_messages_container = message_container + route_backend_conversations.login_required = lambda func: func + route_backend_conversations.user_required = lambda func: func + route_backend_conversations.swagger_route = lambda **kwargs: (lambda func: func) + route_backend_conversations.get_auth_security = lambda: {} + route_backend_conversations.get_current_user_id = lambda: test_user_id + route_backend_conversations.debug_print = lambda *args, **kwargs: None + route_backend_conversations.filter_assistant_artifact_items = lambda items: items + route_backend_conversations.CosmosResourceNotFoundError = DummyNotFoundError + route_backend_conversations.bump_conversation_cache_version = ( + lambda user_id, reason=None: cache_bumps.append((user_id, reason)) or len(cache_bumps) + ) + route_backend_conversations.mark_chat_response_notifications_read_for_conversation = lambda *args, **kwargs: 0 + route_backend_conversations.CLIENTS = { + 'storage_account_office_docs_client': FakeBlobServiceClient(blob_items), + } + + app = Flask(__name__) + app.config['TESTING'] = True + app.config['conversation_container'] = conversation_container + app.config['conversation_cache_bumps'] = cache_bumps + app.config['route_backend_conversations'] = route_backend_conversations + route_backend_conversations.register_route_backend_conversations(app) + except BaseException: + restore() + raise + return app, message_container, restore - app = Flask(__name__) - app.config['TESTING'] = True - app.config['conversation_container'] = conversation_container - app.config['conversation_cache_bumps'] = cache_bumps - app.config['route_backend_conversations'] = route_backend_conversations - route_backend_conversations.register_route_backend_conversations(app) - return app, message_container, (lambda: None) +def test_route_import_dependencies_are_restored_after_requests_and_setup_failures(): + """The legacy read fixture must not poison real deletion-module imports.""" + names = {*_route_import_stubs(), 'route_backend_conversations'} + original = {name: sys.modules.get(name) for name in names} + app, _, restore = build_test_app( + 'user-owner', [{'id': 'conversation-owner', 'user_id': 'user-owner'}], [], + ) + try: + with app.test_client() as client: + response = client.get('/api/get_messages?conversation_id=conversation-owner') + assert response.status_code == 200 + finally: + restore() + restored = {name: sys.modules.get(name) for name in names} + assert restored == original + + with patch.object(importlib, 'import_module', side_effect=ImportError('Injected route-import failure')): + try: + build_test_app('user-owner', [], []) + except ImportError: + pass + else: + raise AssertionError('The injected route-import failure was not propagated.') + restored_after_failure = {name: sys.modules.get(name) for name in names} + assert restored_after_failure == original + + with patch.object(Flask, 'add_url_rule', side_effect=RuntimeError('Injected app-setup failure')): + try: + build_test_app('user-owner', [], []) + except RuntimeError: + pass + else: + raise AssertionError('The injected app-setup failure was not propagated.') + restored_after_setup_failure = {name: sys.modules.get(name) for name in names} + assert restored_after_setup_failure == original + return True if __name__ == '__main__' else None def test_owner_can_read_messages(): @@ -837,6 +895,7 @@ def capture_log_event( if __name__ == '__main__': tests = [ + test_route_import_dependencies_are_restored_after_requests_and_setup_failures, test_owner_can_read_messages, test_foreign_messages_return_forbidden_before_query, test_missing_conversation_preserves_empty_message_history_response,