diff --git a/application/single_app/functions_collaboration.py b/application/single_app/functions_collaboration.py index a3fcd46f5..a6d87f66f 100644 --- a/application/single_app/functions_collaboration.py +++ b/application/single_app/functions_collaboration.py @@ -34,7 +34,11 @@ translate_image_proposal_source_metadata, utc_now_iso, ) -from functions_activity_logging import log_chat_activity +from functions_activity_logging import ( + log_chat_activity, + log_conversation_archival, + log_conversation_deletion, +) from functions_appinsights import log_event from functions_conversation_cache import bump_conversation_cache_version, invalidate_conversation_cache_for_item from functions_documents import sync_chat_upload_workspace_document_sharing_for_collaboration @@ -46,7 +50,11 @@ ) from functions_message_artifacts import filter_assistant_artifact_items from functions_notifications import create_collaboration_message_notification -from functions_thoughts import delete_thoughts_for_conversation, get_thoughts_for_message +from functions_thoughts import ( + archive_thoughts_for_conversation, + delete_thoughts_for_conversation, + get_thoughts_for_message, +) PERSONAL_COLLABORATION_MANAGER_ROLES = { @@ -2236,79 +2244,375 @@ def leave_personal_collaboration_conversation(conversation_id, current_user_id, return conversation_doc, participant, promoted_participant -def _delete_source_personal_conversation(conversation_doc, current_user_id): - source_conversation_id = str((conversation_doc or {}).get('source_conversation_id') or '').strip() +def _archive_collaboration_item(item, archive_container, record_type, retention_deletion): + archived_item = deepcopy(item) + archived_item['archived_at'] = datetime.now(timezone.utc).isoformat() + archived_item['collaboration_record_type'] = record_type + if retention_deletion: + archived_item['archived_by_retention_policy'] = True + archive_container.upsert_item(archived_item) + + +def _delete_blob_backed_collaboration_files(messages): + # Local import avoids the functions_simplechat_operations collaboration dependency cycle. + from functions_simplechat_operations import delete_blob_backed_chat_message_files + + return delete_blob_backed_chat_message_files(messages, raise_on_error=True) + + +def _delete_item_if_present(container, item_id, partition_key): + try: + container.delete_item( + item=item_id, + partition_key=partition_key, + ) + except CosmosResourceNotFoundError: + log_event( + '[CollaborationRetention] Record was already deleted', + extra={ + 'item_id': item_id, + 'partition_key': partition_key, + }, + debug_only=True, + ) + + +def _collaboration_retention_identity(conversation_doc): + scope = ( + conversation_doc.get('scope') + if isinstance((conversation_doc or {}).get('scope'), dict) + else {} + ) + return ( + str((conversation_doc or {}).get('chat_type') or '').strip(), + str((conversation_doc or {}).get('created_by_user_id') or '').strip(), + str(scope.get('group_id') or '').strip(), + ) + + +def _read_collaboration_conversation_for_retention(selected_conversation_doc): + conversation_id = str((selected_conversation_doc or {}).get('id') or '').strip() + try: + live_conversation_doc = get_collaboration_conversation(conversation_id) + except CosmosResourceNotFoundError: + return { + 'id': conversation_id, + 'title': (selected_conversation_doc or {}).get('title', 'Untitled'), + 'updated_at': (selected_conversation_doc or {}).get('updated_at'), + 'already_deleted': True, + } + + if ( + live_conversation_doc.get('updated_at') + != (selected_conversation_doc or {}).get('updated_at') + or _collaboration_retention_identity(live_conversation_doc) + != _collaboration_retention_identity(selected_conversation_doc) + ): + return None + + return live_conversation_doc + + +def _cleanup_collaboration_thoughts(conversation_id, user_ids, archiving_enabled): + for user_id in { + str(raw_user_id or '').strip() + for raw_user_id in user_ids + if str(raw_user_id or '').strip() + }: + if archiving_enabled: + archive_thoughts_for_conversation( + conversation_id, + user_id, + raise_on_error=True, + ) + else: + delete_thoughts_for_conversation( + conversation_id, + user_id, + raise_on_error=True, + ) + + +def _cleanup_linked_collaboration_source( + conversation_doc, + source_field, + source_container, + source_messages_container, + archiving_enabled, + retention_deletion, + expected_user_id=None, + additional_thought_user_ids=None, +): + source_conversation_id = str((conversation_doc or {}).get(source_field) or '').strip() if not source_conversation_id: - return + return None try: - source_conversation = cosmos_conversations_container.read_item( + source_conversation = source_container.read_item( item=source_conversation_id, partition_key=source_conversation_id, ) except CosmosResourceNotFoundError: - return + return None - if str(source_conversation.get('user_id') or '').strip() != str(current_user_id or '').strip(): - return + normalized_expected_user_id = str(expected_user_id or '').strip() + if normalized_expected_user_id and str(source_conversation.get('user_id') or '').strip() != normalized_expected_user_id: + return None if str(source_conversation.get('collaboration_conversation_id') or '').strip() != str(conversation_doc.get('id') or '').strip(): - return + return None - message_query = 'SELECT * FROM c WHERE c.conversation_id = @conversation_id' - source_messages = list(cosmos_messages_container.query_items( - query=message_query, + source_messages = list(source_messages_container.query_items( + query='SELECT * FROM c WHERE c.conversation_id = @conversation_id', parameters=[{'name': '@conversation_id', 'value': source_conversation_id}], partition_key=source_conversation_id, )) - for message_doc in source_messages: - cosmos_messages_container.delete_item( - item=message_doc.get('id'), - partition_key=source_conversation_id, + if archiving_enabled: + _archive_collaboration_item( + source_conversation, + cosmos_archived_conversations_container, + 'linked_source_conversation', + retention_deletion, + ) + else: + _delete_blob_backed_collaboration_files(source_messages) + + for source_message in source_messages: + if archiving_enabled: + _archive_collaboration_item( + source_message, + cosmos_archived_messages_container, + 'linked_source_message', + retention_deletion, + ) + _delete_item_if_present( + source_messages_container, + source_message.get('id'), + source_conversation_id, ) - delete_thoughts_for_conversation(source_conversation_id, current_user_id) - cosmos_conversations_container.delete_item( - item=source_conversation_id, - partition_key=source_conversation_id, + source_user_id = str(source_conversation.get('user_id') or '').strip() + _cleanup_collaboration_thoughts( + source_conversation_id, + [source_user_id, *(additional_thought_user_ids or [])], + archiving_enabled, + ) + _delete_item_if_present( + source_container, + source_conversation_id, + source_conversation_id, + ) + invalidate_conversation_cache_for_item( + source_conversation, + reason='collaboration_source_deleted', ) + return source_conversation -def _delete_source_group_conversation(conversation_doc, current_user_id): - source_conversation_id = str((conversation_doc or {}).get('legacy_source_conversation_id') or '').strip() - if not source_conversation_id: - return +def _delete_source_personal_conversation( + conversation_doc, + current_user_id, + archiving_enabled=False, + retention_deletion=False, +): + return _cleanup_linked_collaboration_source( + conversation_doc, + 'source_conversation_id', + cosmos_conversations_container, + cosmos_messages_container, + archiving_enabled, + retention_deletion, + expected_user_id=current_user_id, + ) - try: - source_conversation = cosmos_group_conversations_container.read_item( - item=source_conversation_id, - partition_key=source_conversation_id, + +def _delete_source_group_conversation( + conversation_doc, + current_user_id, + archiving_enabled=False, + retention_deletion=False, +): + return _cleanup_linked_collaboration_source( + conversation_doc, + 'legacy_source_conversation_id', + cosmos_group_conversations_container, + cosmos_group_messages_container, + archiving_enabled, + retention_deletion, + expected_user_id=current_user_id, + ) + + +def _delete_collaboration_conversation_records( + conversation_doc, + workspace_type, + archiving_enabled, + retention_deletion, + expected_source_user_id=None, +): + conversation_id = str((conversation_doc or {}).get('id') or '').strip() + if not conversation_id: + raise ValueError('conversation_id is required') + + if retention_deletion: + live_conversation_doc = _read_collaboration_conversation_for_retention( + conversation_doc ) - except CosmosResourceNotFoundError: - return + if live_conversation_doc is None: + return None + if live_conversation_doc.get('already_deleted'): + return live_conversation_doc + conversation_doc = live_conversation_doc + else: + conversation_doc = get_collaboration_conversation(conversation_id) - if str(source_conversation.get('user_id') or '').strip() != str(current_user_id or '').strip(): - return - if str(source_conversation.get('collaboration_conversation_id') or '').strip() != str(conversation_doc.get('id') or '').strip(): - return + if is_personal_collaboration_conversation(conversation_doc): + revocation_conversation_doc = deepcopy(conversation_doc) + revocation_conversation_doc['accepted_participant_ids'] = [] + sync_chat_upload_workspace_document_sharing_for_collaboration( + revocation_conversation_doc + ) - message_query = 'SELECT * FROM c WHERE c.conversation_id = @conversation_id' - source_messages = list(cosmos_group_messages_container.query_items( - query=message_query, - parameters=[{'name': '@conversation_id', 'value': source_conversation_id}], - partition_key=source_conversation_id, + state_docs = list(cosmos_collaboration_user_state_container.query_items( + query='SELECT * FROM c WHERE c.conversation_id = @conversation_id', + parameters=[{'name': '@conversation_id', 'value': conversation_id}], + enable_cross_partition_query=True, )) - for message_doc in source_messages: - cosmos_group_messages_container.delete_item( - item=message_doc.get('id'), - partition_key=source_conversation_id, + if archiving_enabled: + archived_conversation_doc = deepcopy(conversation_doc) + archived_conversation_doc['collaboration_user_states'] = state_docs + _archive_collaboration_item( + archived_conversation_doc, + cosmos_archived_conversations_container, + 'collaboration_conversation', + retention_deletion, + ) + log_conversation_archival( + user_id=conversation_doc.get('created_by_user_id'), + conversation_id=conversation_id, + title=conversation_doc.get('title', 'Untitled'), + workspace_type=workspace_type, + context=conversation_doc.get('context', []), + tags=conversation_doc.get('tags', []), + group_id=str((conversation_doc.get('scope') or {}).get('group_id') or '').strip() or None, + additional_context={ + 'conversation_kind': 'collaboration', + 'deletion_reason': 'retention_policy' if retention_deletion else 'user_request', + }, ) - delete_thoughts_for_conversation(source_conversation_id, current_user_id) - cosmos_group_conversations_container.delete_item( - item=source_conversation_id, - partition_key=source_conversation_id, + messages = list(cosmos_collaboration_messages_container.query_items( + query='SELECT * FROM c WHERE c.conversation_id = @conversation_id', + parameters=[{'name': '@conversation_id', 'value': conversation_id}], + partition_key=conversation_id, + )) + if not archiving_enabled: + _delete_blob_backed_collaboration_files(messages) + + for message_doc in messages: + if archiving_enabled: + _archive_collaboration_item( + message_doc, + cosmos_archived_messages_container, + 'collaboration_message', + retention_deletion, + ) + _delete_item_if_present( + cosmos_collaboration_messages_container, + message_doc.get('id'), + conversation_id, + ) + + thought_user_ids = [conversation_doc.get('created_by_user_id')] + thought_user_ids.extend( + participant.get('user_id') + for participant in list(conversation_doc.get('participants', []) or []) + if isinstance(participant, dict) + ) + source_thought_user_ids = list(thought_user_ids) + source_thought_user_ids.extend( + str((message_doc.get('metadata') or {}).get('source_thought_user_id') or '').strip() + for message_doc in messages + if isinstance(message_doc, dict) + and isinstance(message_doc.get('metadata'), dict) + ) + _cleanup_collaboration_thoughts( + conversation_id, + thought_user_ids, + archiving_enabled, + ) + + for state_doc in state_docs: + _delete_item_if_present( + cosmos_collaboration_user_state_container, + state_doc.get('id'), + state_doc.get('user_id'), + ) + + _cleanup_linked_collaboration_source( + conversation_doc, + 'source_conversation_id', + cosmos_conversations_container, + cosmos_messages_container, + archiving_enabled, + retention_deletion, + expected_user_id=expected_source_user_id, + additional_thought_user_ids=source_thought_user_ids, + ) + _cleanup_linked_collaboration_source( + conversation_doc, + 'legacy_source_conversation_id', + cosmos_group_conversations_container, + cosmos_group_messages_container, + archiving_enabled, + retention_deletion, + expected_user_id=expected_source_user_id, + additional_thought_user_ids=source_thought_user_ids, + ) + + log_conversation_deletion( + user_id=conversation_doc.get('created_by_user_id'), + conversation_id=conversation_id, + title=conversation_doc.get('title', 'Untitled'), + workspace_type=workspace_type, + context=conversation_doc.get('context', []), + tags=conversation_doc.get('tags', []), + is_archived=archiving_enabled, + is_bulk_operation=retention_deletion, + group_id=str((conversation_doc.get('scope') or {}).get('group_id') or '').strip() or None, + additional_context={ + 'conversation_kind': 'collaboration', + 'deletion_reason': 'retention_policy' if retention_deletion else 'user_request', + }, + ) + _delete_item_if_present( + cosmos_collaboration_conversations_container, + conversation_id, + conversation_id, + ) + invalidate_conversation_cache_for_item( + conversation_doc, + reason='collaboration_deleted', + ) + return { + 'id': conversation_id, + 'title': conversation_doc.get('title', 'Untitled'), + 'updated_at': conversation_doc.get('updated_at'), + } + + +def delete_collaboration_conversation_for_retention( + conversation_doc, + workspace_type, + archiving_enabled, +): + """Delete a policy-selected collaboration conversation and all linked records.""" + return _delete_collaboration_conversation_records( + conversation_doc, + workspace_type, + archiving_enabled, + retention_deletion=True, ) @@ -2329,43 +2633,13 @@ def delete_personal_collaboration_conversation(conversation_id, current_user_id) if current_user_id not in set(conversation_doc.get('owner_user_ids', []) or []): raise PermissionError('Only conversation owners can delete this shared conversation') - if is_personal_collaboration_conversation(conversation_doc): - revocation_conversation_doc = deepcopy(conversation_doc) - revocation_conversation_doc['accepted_participant_ids'] = [] - sync_chat_upload_workspace_document_sharing_for_collaboration(revocation_conversation_doc) - - message_query = 'SELECT * FROM c WHERE c.conversation_id = @conversation_id' - messages = list(cosmos_collaboration_messages_container.query_items( - query=message_query, - parameters=[{'name': '@conversation_id', 'value': conversation_id}], - partition_key=conversation_id, - )) - for message_doc in messages: - cosmos_collaboration_messages_container.delete_item( - item=message_doc.get('id'), - partition_key=conversation_id, - ) - - state_query = 'SELECT * FROM c WHERE c.conversation_id = @conversation_id' - state_docs = list(cosmos_collaboration_user_state_container.query_items( - query=state_query, - parameters=[{'name': '@conversation_id', 'value': conversation_id}], - enable_cross_partition_query=True, - )) - for state_doc in state_docs: - cosmos_collaboration_user_state_container.delete_item( - item=state_doc.get('id'), - partition_key=state_doc.get('user_id'), - ) - - _delete_source_personal_conversation(conversation_doc, current_user_id) - if is_group_collaboration_conversation(conversation_doc): - _delete_source_group_conversation(conversation_doc, current_user_id) - cosmos_collaboration_conversations_container.delete_item( - item=conversation_id, - partition_key=conversation_id, + _delete_collaboration_conversation_records( + conversation_doc, + workspace_type='group' if is_group_collaboration_conversation(conversation_doc) else 'personal', + archiving_enabled=False, + retention_deletion=False, + expected_source_user_id=current_user_id, ) - invalidate_conversation_cache_for_item(conversation_doc, reason="collaboration_deleted") return conversation_doc diff --git a/application/single_app/functions_group.py b/application/single_app/functions_group.py index 94ce8a075..bab68f7f6 100644 --- a/application/single_app/functions_group.py +++ b/application/single_app/functions_group.py @@ -42,6 +42,10 @@ def create_group(name, description): ], "pendingUsers": [], "disable_file_downloads": False, + "retention_policy": { + "conversation_retention_days": "default", + "document_retention_days": "default" + }, "createdDate": now_str, "modifiedDate": now_str } diff --git a/application/single_app/functions_retention_policy.py b/application/single_app/functions_retention_policy.py index b538fc4b5..7e345ab62 100644 --- a/application/single_app/functions_retention_policy.py +++ b/application/single_app/functions_retention_policy.py @@ -6,11 +6,12 @@ This module handles automated deletion of aged conversations and documents based on configurable retention policies for personal, group, and public workspaces. -Version: 0.237.005 +Version: 0.250.103 Implemented in: 0.234.067 Updated in: 0.236.012 - Fixed race condition handling for NotFound errors during deletion Updated in: 0.237.004 - Fixed critical bug where conversations with null/undefined last_activity_at were deleted regardless of age Updated in: 0.237.005 - Fixed field name: use last_updated (actual field) instead of last_activity_at (non-existent) +Updated in: 0.250.103 - Applied retention by conversation ownership across current, legacy, and collaboration stores """ from config import * @@ -20,12 +21,20 @@ from functions_documents import delete_document, delete_document_chunks from functions_simplechat_operations import delete_blob_backed_chat_message_files from functions_activity_logging import log_conversation_deletion, log_conversation_archival +from functions_collaboration import delete_collaboration_conversation_for_retention +from functions_conversation_cache import invalidate_conversation_cache_for_item from functions_notifications import create_notification, create_group_notification, create_public_workspace_notification +from functions_thoughts import archive_thoughts_for_conversation, delete_thoughts_for_conversation from functions_debug import debug_print from functions_appinsights import log_event from datetime import datetime, timezone, timedelta +GROUP_SINGLE_USER_CHAT_TYPES = {'group', 'group-single-user', 'group_single_user'} +PERSONAL_MULTI_USER_CHAT_TYPE = 'personal_multi_user' +GROUP_MULTI_USER_CHAT_TYPE = 'group_multi_user' + + def get_all_user_settings(): """ Get all user settings from Cosmos DB. @@ -127,6 +136,351 @@ def resolve_retention_value(value, workspace_type, retention_type, settings=None return 'none' +def _parse_retention_timestamp(value): + """Return a timezone-aware activity timestamp or None when the value is invalid.""" + if not isinstance(value, str) or not value.strip(): + return None + + normalized_value = value.strip() + if normalized_value.endswith('Z'): + normalized_value = f"{normalized_value[:-1]}+00:00" + + try: + parsed_value = datetime.fromisoformat(normalized_value) + except ValueError: + return None + + if parsed_value.tzinfo is None: + return parsed_value.replace(tzinfo=timezone.utc) + return parsed_value.astimezone(timezone.utc) + + +def _get_primary_group_id(conversation_item): + """Return the group that governs a single-user conversation.""" + for context_item in list((conversation_item or {}).get('context', []) or []): + if not isinstance(context_item, dict): + continue + if context_item.get('type') != 'primary' or context_item.get('scope') != 'group': + continue + return str(context_item.get('id') or '').strip() + return str((conversation_item or {}).get('group_id') or '').strip() + + +def _is_group_single_user_conversation(conversation_item): + chat_type = str((conversation_item or {}).get('chat_type') or '').strip().lower() + return chat_type in GROUP_SINGLE_USER_CHAT_TYPES or bool(_get_primary_group_id(conversation_item)) + + +def _is_converted_conversation_source(conversation_item): + return bool(str((conversation_item or {}).get('collaboration_conversation_id') or '').strip()) + + +def _is_aged_conversation(conversation_item, timestamp_field, cutoff_date): + activity_at = _parse_retention_timestamp((conversation_item or {}).get(timestamp_field)) + return activity_at is not None and activity_at < cutoff_date + + +def _build_group_scope_query(timestamp_field): + return f""" + SELECT * FROM c + WHERE ( + c.group_id = @scope_id + OR ( + IS_ARRAY(c.context) + AND EXISTS( + SELECT VALUE context_item + FROM context_item IN c.context + WHERE context_item.type = 'primary' + AND context_item.scope = 'group' + AND context_item.id = @scope_id + ) + ) + ) + AND IS_DEFINED(c.{timestamp_field}) + AND IS_STRING(c.{timestamp_field}) + AND c.{timestamp_field} < @cutoff_date + """ + + +def _build_conversation_retention_sources( + workspace_type, + cutoff_iso, + user_id=None, + group_id=None, + public_workspace_id=None, +): + """Describe the backing stores governed by a workspace retention policy.""" + if workspace_type == 'personal': + return [ + { + 'name': 'personal_single_user', + 'container': cosmos_conversations_container, + 'messages_container': cosmos_messages_container, + 'timestamp_field': 'last_updated', + 'query': """ + SELECT * FROM c + WHERE c.user_id = @scope_id + AND IS_DEFINED(c.last_updated) + AND IS_STRING(c.last_updated) + AND c.last_updated < @cutoff_date + """, + 'parameters': [ + {'name': '@scope_id', 'value': user_id}, + {'name': '@cutoff_date', 'value': cutoff_iso}, + ], + 'matches_scope': lambda item: ( + not _is_group_single_user_conversation(item) + and not _is_converted_conversation_source(item) + ), + }, + { + 'name': PERSONAL_MULTI_USER_CHAT_TYPE, + 'container': cosmos_collaboration_conversations_container, + 'timestamp_field': 'updated_at', + 'query': """ + SELECT * FROM c + WHERE c.chat_type = @chat_type + AND c.created_by_user_id = @scope_id + AND IS_DEFINED(c.updated_at) + AND IS_STRING(c.updated_at) + AND c.updated_at < @cutoff_date + """, + 'parameters': [ + {'name': '@chat_type', 'value': PERSONAL_MULTI_USER_CHAT_TYPE}, + {'name': '@scope_id', 'value': user_id}, + {'name': '@cutoff_date', 'value': cutoff_iso}, + ], + 'matches_scope': lambda item: ( + item.get('chat_type') == PERSONAL_MULTI_USER_CHAT_TYPE + and item.get('created_by_user_id') == user_id + ), + 'collaboration': True, + }, + ] + + if workspace_type == 'group': + group_parameters = [ + {'name': '@scope_id', 'value': group_id}, + {'name': '@cutoff_date', 'value': cutoff_iso}, + ] + return [ + { + 'name': 'legacy_group', + 'container': cosmos_group_conversations_container, + 'messages_container': cosmos_group_messages_container, + 'timestamp_field': 'last_updated', + 'query': _build_group_scope_query('last_updated'), + 'parameters': group_parameters, + 'matches_scope': lambda item: ( + _get_primary_group_id(item) == group_id + and not _is_converted_conversation_source(item) + ), + }, + { + 'name': 'group_single_user', + 'container': cosmos_conversations_container, + 'messages_container': cosmos_messages_container, + 'timestamp_field': 'last_updated', + 'query': _build_group_scope_query('last_updated'), + 'parameters': group_parameters, + 'matches_scope': lambda item: ( + _is_group_single_user_conversation(item) + and _get_primary_group_id(item) == group_id + and not _is_converted_conversation_source(item) + ), + }, + { + 'name': GROUP_MULTI_USER_CHAT_TYPE, + 'container': cosmos_collaboration_conversations_container, + 'timestamp_field': 'updated_at', + 'query': """ + SELECT * FROM c + WHERE c.chat_type = @chat_type + AND c.scope.group_id = @scope_id + AND IS_DEFINED(c.updated_at) + AND IS_STRING(c.updated_at) + AND c.updated_at < @cutoff_date + """, + 'parameters': [ + {'name': '@chat_type', 'value': GROUP_MULTI_USER_CHAT_TYPE}, + {'name': '@scope_id', 'value': group_id}, + {'name': '@cutoff_date', 'value': cutoff_iso}, + ], + 'matches_scope': lambda item: ( + item.get('chat_type') == GROUP_MULTI_USER_CHAT_TYPE + and str((item.get('scope') or {}).get('group_id') or '').strip() == group_id + ), + 'collaboration': True, + }, + ] + + return [ + { + 'name': 'public', + 'container': cosmos_public_conversations_container, + 'messages_container': cosmos_public_messages_container, + 'timestamp_field': 'last_updated', + 'query': """ + SELECT * FROM c + WHERE c.public_workspace_id = @scope_id + AND IS_DEFINED(c.last_updated) + AND IS_STRING(c.last_updated) + AND c.last_updated < @cutoff_date + """, + 'parameters': [ + {'name': '@scope_id', 'value': public_workspace_id}, + {'name': '@cutoff_date', 'value': cutoff_iso}, + ], + 'matches_scope': lambda item: item.get('public_workspace_id') == public_workspace_id, + }, + ] + + +def _delete_standard_conversation_for_retention( + conversation_item, + source, + workspace_type, + archiving_enabled, + cutoff_date, +): + """Archive and delete a non-collaboration conversation and its dependent records.""" + conversation_id = conversation_item.get('id') + conversation_title = conversation_item.get('title', 'Untitled') + container = source['container'] + messages_container = source['messages_container'] + + selected_conversation_item = conversation_item + try: + live_conversation_item = container.read_item( + item=conversation_id, + partition_key=conversation_id, + ) + except CosmosResourceNotFoundError: + return { + 'id': conversation_id, + 'title': conversation_title, + source['timestamp_field']: selected_conversation_item.get( + source['timestamp_field'] + ), + 'already_deleted': True, + } + + if ( + not source['matches_scope'](live_conversation_item) + or not _is_aged_conversation( + live_conversation_item, + source['timestamp_field'], + cutoff_date, + ) + ): + return None + + if ( + live_conversation_item.get(source['timestamp_field']) + != selected_conversation_item.get(source['timestamp_field']) + ): + return None + + conversation_item = live_conversation_item + + if archiving_enabled: + archived_item = dict(conversation_item) + archived_item['archived_at'] = datetime.now(timezone.utc).isoformat() + archived_item['archived_by_retention_policy'] = True + archived_item['retention_source'] = source['name'] + cosmos_archived_conversations_container.upsert_item(archived_item) + + log_conversation_archival( + user_id=conversation_item.get('user_id'), + conversation_id=conversation_id, + title=conversation_title, + workspace_type=workspace_type, + context=conversation_item.get('context', []), + tags=conversation_item.get('tags', []), + group_id=_get_primary_group_id(conversation_item) or None, + public_workspace_id=conversation_item.get('public_workspace_id'), + additional_context={'deletion_reason': 'retention_policy'}, + ) + + messages = list(messages_container.query_items( + query='SELECT * FROM c WHERE c.conversation_id = @conversation_id', + parameters=[{'name': '@conversation_id', 'value': conversation_id}], + partition_key=conversation_id, + )) + + if not archiving_enabled: + delete_blob_backed_chat_message_files(messages, raise_on_error=True) + + for message_item in messages: + if archiving_enabled: + archived_message = dict(message_item) + archived_message['archived_at'] = datetime.now(timezone.utc).isoformat() + archived_message['archived_by_retention_policy'] = True + archived_message['retention_source'] = source['name'] + cosmos_archived_messages_container.upsert_item(archived_message) + + try: + messages_container.delete_item( + item=message_item.get('id'), + partition_key=conversation_id, + ) + except CosmosResourceNotFoundError: + debug_print( + f"[RetentionPolicy] Message {message_item.get('id')} was already deleted" + ) + + thought_user_id = conversation_item.get('user_id') + if archiving_enabled: + archive_thoughts_for_conversation( + conversation_id, + thought_user_id, + raise_on_error=True, + ) + else: + delete_thoughts_for_conversation( + conversation_id, + thought_user_id, + raise_on_error=True, + ) + + log_conversation_deletion( + user_id=conversation_item.get('user_id'), + conversation_id=conversation_id, + title=conversation_title, + workspace_type=workspace_type, + context=conversation_item.get('context', []), + tags=conversation_item.get('tags', []), + is_archived=archiving_enabled, + is_bulk_operation=True, + group_id=_get_primary_group_id(conversation_item) or None, + public_workspace_id=conversation_item.get('public_workspace_id'), + additional_context={ + 'deletion_reason': 'retention_policy', + 'retention_source': source['name'], + }, + ) + + try: + container.delete_item( + item=conversation_id, + partition_key=conversation_id, + ) + except CosmosResourceNotFoundError: + debug_print( + f"[RetentionPolicy] Conversation {conversation_id} was already deleted" + ) + + invalidate_conversation_cache_for_item( + conversation_item, + reason='retention_policy_deleted', + ) + return { + 'id': conversation_id, + 'title': conversation_title, + source['timestamp_field']: conversation_item.get(source['timestamp_field']), + } + + def execute_retention_policy(workspace_scopes=None, manual_execution=False): """ Execute retention policy for specified workspace scopes. @@ -489,7 +843,7 @@ def process_public_retention(): def delete_aged_conversations(retention_days, workspace_type='personal', user_id=None, group_id=None, public_workspace_id=None): """ - Delete conversations that exceed the retention period based on last_updated. + Delete conversations governed by a workspace policy across all backing stores. Args: retention_days (int): Number of days to retain conversations @@ -503,171 +857,108 @@ def delete_aged_conversations(retention_days, workspace_type='personal', user_id """ settings = get_settings() archiving_enabled = settings.get('enable_conversation_archiving', False) - - # Determine which container to use - if workspace_type == 'group': - container = cosmos_group_conversations_container - partition_field = 'group_id' - partition_value = group_id - elif workspace_type == 'public': - container = cosmos_public_conversations_container - partition_field = 'public_workspace_id' - partition_value = public_workspace_id - else: - container = cosmos_conversations_container - partition_field = 'user_id' - partition_value = user_id - - # Calculate cutoff date cutoff_date = datetime.now(timezone.utc) - timedelta(days=retention_days) cutoff_iso = cutoff_date.isoformat() - - # Query for aged conversations - # ONLY delete conversations that have a valid last_updated that is older than the cutoff - # Conversations with null/undefined last_updated should be SKIPPED (not deleted) - # This prevents accidentally deleting new conversations that haven't had their timestamp set - query = f""" - SELECT c.id, c.title, c.last_updated, c.{partition_field} - FROM c - WHERE c.{partition_field} = @partition_value - AND IS_DEFINED(c.last_updated) - AND NOT IS_NULL(c.last_updated) - AND c.last_updated < @cutoff_date - """ - - parameters = [ - {"name": "@partition_value", "value": partition_value}, - {"name": "@cutoff_date", "value": cutoff_iso} - ] - - debug_print(f"Querying aged conversations: workspace_type={workspace_type}, partition_field={partition_field}, partition_value={partition_value}, cutoff_date={cutoff_iso}, retention_days={retention_days}") - - try: - aged_conversations = list(container.query_items( - query=query, - parameters=parameters, - enable_cross_partition_query=True - )) - debug_print(f"Found {len(aged_conversations)} aged conversations for {workspace_type} workspace") - except Exception as query_error: - log_event("delete_aged_conversations_query_error", {"error": str(query_error), "workspace_type": workspace_type, "partition_value": partition_value}) - debug_print(f"Error querying aged conversations for {workspace_type} (partition_value={partition_value}): {query_error}") - return {'count': 0, 'details': []} - deleted_details = [] - - for conv in aged_conversations: + sources = _build_conversation_retention_sources( + workspace_type, + cutoff_iso, + user_id=user_id, + group_id=group_id, + public_workspace_id=public_workspace_id, + ) + + for source in sources: try: - conversation_id = conv.get('id') - conversation_title = conv.get('title', 'Untitled') - - # Read full conversation for archiving/logging + candidates = list(source['container'].query_items( + query=source['query'], + parameters=source['parameters'], + enable_cross_partition_query=True, + )) + except Exception as query_error: + log_event( + "[RetentionPolicy] Conversation query failed", + { + 'error': str(query_error), + 'workspace_type': workspace_type, + 'retention_source': source['name'], + }, + ) + debug_print( + f"[RetentionPolicy] Failed querying {source['name']}: {query_error}" + ) + continue + + aged_conversations = [ + candidate + for candidate in candidates + if source['matches_scope'](candidate) + and _is_aged_conversation( + candidate, + source['timestamp_field'], + cutoff_date, + ) + ] + debug_print( + f"[RetentionPolicy] Found {len(aged_conversations)} aged " + f"{source['name']} conversations for {workspace_type}" + ) + + for conversation_item in aged_conversations: + conversation_id = conversation_item.get('id', 'unknown') try: - conversation_item = container.read_item( - item=conversation_id, - partition_key=conversation_id + if source.get('collaboration'): + deletion_detail = delete_collaboration_conversation_for_retention( + conversation_item, + workspace_type=workspace_type, + archiving_enabled=archiving_enabled, + ) + else: + deletion_detail = _delete_standard_conversation_for_retention( + conversation_item, + source, + workspace_type, + archiving_enabled, + cutoff_date, + ) + + if deletion_detail is None: + debug_print( + f"[RetentionPolicy] Skipped {conversation_id} because it " + "changed after selection" + ) + continue + + deleted_details.append(deletion_detail) + debug_print( + f"[RetentionPolicy] Deleted {source['name']} conversation " + f"{conversation_id}" ) except CosmosResourceNotFoundError: - # Conversation was already deleted (race condition) - this is fine, skip to next - debug_print(f"Conversation {conversation_id} already deleted (not found during read), skipping") + debug_print( + f"[RetentionPolicy] Conversation {conversation_id} was already deleted" + ) deleted_details.append({ 'id': conversation_id, - 'title': conversation_title, - 'last_updated': conv.get('last_updated'), - 'already_deleted': True + 'title': conversation_item.get('title', 'Untitled'), + source['timestamp_field']: conversation_item.get(source['timestamp_field']), + 'already_deleted': True, }) - continue - - # Archive if enabled - if archiving_enabled: - archived_item = dict(conversation_item) - archived_item["archived_at"] = datetime.now(timezone.utc).isoformat() - archived_item["archived_by_retention_policy"] = True - cosmos_archived_conversations_container.upsert_item(archived_item) - - log_conversation_archival( - user_id=conversation_item.get('user_id'), - conversation_id=conversation_id, - title=conversation_title, - workspace_type=workspace_type, - context=conversation_item.get('context', []), - tags=conversation_item.get('tags', []), - group_id=conversation_item.get('group_id'), - public_workspace_id=conversation_item.get('public_workspace_id') + except Exception as deletion_error: + log_event( + "[RetentionPolicy] Conversation deletion failed", + { + 'error': str(deletion_error), + 'conversation_id': conversation_id, + 'workspace_type': workspace_type, + 'retention_source': source['name'], + }, ) - - # Delete messages - - if workspace_type == 'group': - messages_container = cosmos_group_messages_container - elif workspace_type == 'public': - messages_container = cosmos_public_messages_container - else: - messages_container = cosmos_messages_container - - message_query = f"SELECT * FROM c WHERE c.conversation_id = @conversation_id" - message_params = [{"name": "@conversation_id", "value": conversation_id}] - - messages = list(messages_container.query_items( - query=message_query, - parameters=message_params, - partition_key=conversation_id - )) - - if not archiving_enabled and workspace_type == 'personal': - delete_blob_backed_chat_message_files(messages) - - for msg in messages: - if archiving_enabled: - archived_msg = dict(msg) - archived_msg["archived_at"] = datetime.now(timezone.utc).isoformat() - archived_msg["archived_by_retention_policy"] = True - cosmos_archived_messages_container.upsert_item(archived_msg) - - try: - messages_container.delete_item(msg['id'], partition_key=conversation_id) - except CosmosResourceNotFoundError: - # Message was already deleted - this is fine, continue - debug_print(f"Message {msg['id']} already deleted (not found), skipping") - - # Log deletion - log_conversation_deletion( - user_id=conversation_item.get('user_id'), - conversation_id=conversation_id, - title=conversation_title, - workspace_type=workspace_type, - context=conversation_item.get('context', []), - tags=conversation_item.get('tags', []), - is_archived=archiving_enabled, - is_bulk_operation=True, - group_id=conversation_item.get('group_id'), - public_workspace_id=conversation_item.get('public_workspace_id'), - additional_context={'deletion_reason': 'retention_policy'} - ) - - # Delete conversation - try: - container.delete_item( - item=conversation_id, - partition_key=conversation_id + debug_print( + f"[RetentionPolicy] Failed deleting {conversation_id}: " + f"{deletion_error}" ) - except CosmosResourceNotFoundError: - # Conversation was already deleted after we read it (race condition) - this is fine - debug_print(f"Conversation {conversation_id} already deleted (not found during delete)") - - deleted_details.append({ - 'id': conversation_id, - 'title': conversation_title, - 'last_updated': conv.get('last_updated') - }) - - debug_print(f"Deleted conversation {conversation_id} ({conversation_title}) due to retention policy") - - except Exception as e: - conv_id = conv.get('id', 'unknown') if conv else 'unknown' - log_event("delete_aged_conversations_deletion_error", {"error": str(e), "conversation_id": conv_id, "workspace_type": workspace_type}) - debug_print(f"Error deleting conversation {conv_id}: {e}") - + return { 'count': len(deleted_details), 'details': deleted_details diff --git a/application/single_app/functions_simplechat_operations.py b/application/single_app/functions_simplechat_operations.py index 0210b3133..2e3aa49f5 100644 --- a/application/single_app/functions_simplechat_operations.py +++ b/application/single_app/functions_simplechat_operations.py @@ -1274,44 +1274,49 @@ def upload_generated_analysis_artifact_for_user( ) -def delete_blob_backed_chat_message_files(messages: Iterable[Dict[str, Any]]) -> int: +def delete_blob_backed_chat_message_files( + messages: Iterable[Dict[str, Any]], + raise_on_error: bool = False, +) -> int: """Delete blob-backed chat files referenced by the provided message documents.""" - blob_service_client = CLIENTS.get("storage_account_office_docs_client") - if not blob_service_client: - return 0 - - deleted_count = 0 - deleted_targets = set() - + blob_targets = [] + seen_targets = set() for message in messages or []: if not isinstance(message, dict): continue - if str(message.get("file_content_source") or "").strip().lower() != "blob": continue blob_container = str(message.get("blob_container") or "").strip() blob_path = str(message.get("blob_path") or "").strip() - if not blob_container or not blob_path: - continue - target = (blob_container, blob_path) - if target in deleted_targets: + if not blob_container or not blob_path or target in seen_targets: continue + seen_targets.add(target) + blob_targets.append(target) + + blob_service_client = CLIENTS.get("storage_account_office_docs_client") + if not blob_service_client: + if raise_on_error and blob_targets: + raise RuntimeError("Blob storage client is unavailable for chat file cleanup") + return 0 + + deleted_count = 0 + cleanup_errors = [] + for blob_container, blob_path in blob_targets: try: blob_client = blob_service_client.get_blob_client( container=blob_container, blob=blob_path, ) if not blob_client.exists(): - deleted_targets.add(target) continue blob_client.delete_blob() - deleted_targets.add(target) deleted_count += 1 except Exception as exc: + cleanup_errors.append((blob_container, blob_path, exc)) log_event( "[SimpleChat] Failed to delete blob-backed chat file", { @@ -1322,6 +1327,11 @@ def delete_blob_backed_chat_message_files(messages: Iterable[Dict[str, Any]]) -> debug_only=True, ) + if cleanup_errors and raise_on_error: + raise RuntimeError( + f"Failed to delete {len(cleanup_errors)} blob-backed chat file(s)" + ) from cleanup_errors[0][2] + return deleted_count diff --git a/application/single_app/functions_thoughts.py b/application/single_app/functions_thoughts.py index 91df9b981..21cd793e1 100644 --- a/application/single_app/functions_thoughts.py +++ b/application/single_app/functions_thoughts.py @@ -189,7 +189,7 @@ def get_pending_thoughts(conversation_id, user_id, message_id=None): return [] -def get_thoughts_for_conversation(conversation_id, user_id): +def get_thoughts_for_conversation(conversation_id, user_id, raise_on_error=False): """Return all thoughts for a conversation.""" try: query = ( @@ -208,13 +208,19 @@ def get_thoughts_for_conversation(conversation_id, user_id): return results except Exception as e: log_event(f"get_thoughts_for_conversation failed: {e}", level="WARNING") + if raise_on_error: + raise return [] -def archive_thoughts_for_conversation(conversation_id, user_id): +def archive_thoughts_for_conversation(conversation_id, user_id, raise_on_error=False): """Copy all thoughts for a conversation to the archive container, then delete originals.""" try: - thoughts = get_thoughts_for_conversation(conversation_id, user_id) + thoughts = get_thoughts_for_conversation( + conversation_id, + user_id, + raise_on_error=raise_on_error, + ) for thought in thoughts: archived = dict(thought) archived['archived_at'] = datetime.now(timezone.utc).isoformat() @@ -227,12 +233,18 @@ def archive_thoughts_for_conversation(conversation_id, user_id): ) except Exception as e: log_event(f"archive_thoughts_for_conversation failed: {e}", level="WARNING") + if raise_on_error: + raise -def delete_thoughts_for_conversation(conversation_id, user_id): +def delete_thoughts_for_conversation(conversation_id, user_id, raise_on_error=False): """Delete all thoughts for a conversation.""" try: - thoughts = get_thoughts_for_conversation(conversation_id, user_id) + thoughts = get_thoughts_for_conversation( + conversation_id, + user_id, + raise_on_error=raise_on_error, + ) for thought in thoughts: cosmos_thoughts_container.delete_item( item=thought['id'], @@ -240,6 +252,8 @@ def delete_thoughts_for_conversation(conversation_id, user_id): ) except Exception as e: log_event(f"delete_thoughts_for_conversation failed: {e}", level="WARNING") + if raise_on_error: + raise def delete_thoughts_for_message(message_id, user_id): diff --git a/docs/explanation/fixes/v0.250.103/RETENTION_POLICY_CONVERSATION_SCOPE_COVERAGE_FIX.md b/docs/explanation/fixes/v0.250.103/RETENTION_POLICY_CONVERSATION_SCOPE_COVERAGE_FIX.md new file mode 100644 index 000000000..a4416be49 --- /dev/null +++ b/docs/explanation/fixes/v0.250.103/RETENTION_POLICY_CONVERSATION_SCOPE_COVERAGE_FIX.md @@ -0,0 +1,87 @@ +# Retention Policy Conversation Scope Coverage Fix + +## Header Information + +- **Issue:** [#1054](https://github.com/microsoft/simplechat/issues/1054) +- **Fixed in version:** **0.250.103** +- **Area:** Conversation retention for personal, group, and collaborative chats + +### Issue Description + +Conversation retention previously inferred policy ownership from the Cosmos container. That no longer matched the conversation storage model: current private group conversations live in `conversations`, collaborative conversations live in `collaboration_conversations`, and legacy group conversations remain in `group_conversations`. + +### Root Cause + +The retention query selected personal records only by `user_id` and group records only by `group_id` in the legacy group container. It did not account for conversation type, collaboration scope, collaboration creator, the `updated_at` activity field, or hidden source records created during conversation conversion. + +## Technical Details + +### Policy Ownership Matrix + +| Conversation type | Storage | Governing policy | Activity field | +|---|---|---|---| +| Personal single-user | `conversations` | Creator's personal policy | `last_updated` | +| Group single-user | `conversations` | Primary group's policy | `last_updated` | +| Personal multi-user | `collaboration_conversations` | Creator's personal policy | `updated_at` | +| Group multi-user | `collaboration_conversations` | `scope.group_id` policy | `updated_at` | +| Legacy group | `group_conversations` | Group policy | `last_updated` | + +Converted source records linked through `collaboration_conversation_id` are excluded from independent policy selection. The collaboration cleanup removes the linked source record, preventing duplicate deletion and counting. + +### Files Modified + +| File | Change | +|---|---| +| `application/single_app/functions_retention_policy.py` | Selects all conversation shapes under the correct personal or group policy, validates activity timestamps, excludes converted sources, and dispatches cleanup by storage type. | +| `application/single_app/functions_collaboration.py` | Adds retention-aware collaboration cleanup for archival, messages, user state, linked sources, thoughts, blob-backed files, activity logs, race handling, and cache invalidation. | +| `application/single_app/functions_group.py` | Persists both new-group retention values as `"default"`. | +| `application/single_app/functions_simplechat_operations.py` | Adds strict blob cleanup mode so retention stops safely when generated files cannot be removed. | +| `application/single_app/functions_thoughts.py` | Adds strict thought cleanup mode so retention can preserve source metadata for retry after a dependency failure. | +| `functional_tests/test_retention_policy_conversation_scope_coverage.py` | Covers the policy matrix and cleanup side effects. | +| `application/single_app/config.py` | Updates the application version to `0.250.103`. | + +### Cleanup Behavior + +- Destructive retention removes collaboration messages, per-user state, linked personal or legacy group sources, linked source messages, thoughts, and blob-backed generated files. +- Archival retention copies collaboration and linked source conversations/messages into the existing archive containers before deleting active records. +- Personal collaboration deletion revokes automatic chat-upload document sharing. +- Conversation-list caches are invalidated for the collaboration participants and linked source owner. +- Missing or malformed activity timestamps are skipped. +- Already-deleted records are treated as successful race outcomes instead of retention failures. +- Standard and collaboration records are reread immediately before cleanup; records whose activity timestamp or governing scope changed after selection are skipped. +- Dependency cleanup failures leave the live conversation and its record metadata available for a later retention retry. + +### Compatibility + +Existing groups without a `retention_policy` object continue to resolve missing values through organization defaults. Only newly created groups persist the intended explicit `"default"` state. + +## Validation + +### Test Coverage + +`functional_tests/test_retention_policy_conversation_scope_coverage.py` verifies: + +- Every row in the conversation policy matrix. +- Explicit custom values, organization defaults, and `none`. +- Old, new, missing, null, non-string, and malformed timestamps. +- Converted source deduplication. +- Destructive cleanup and archival cleanup. +- Blob-backed file cleanup, user-state removal, linked-source removal, thought cleanup, activity logging, and cache invalidation. +- Already-deleted race handling. +- Concurrent activity revalidation and cleanup-failure preservation. +- Explicit defaults on newly created groups. + +### Before and After + +| Scenario | Before | After | +|---|---|---| +| Private group conversation | Processed by personal policy | Processed only by primary group policy | +| Personal collaboration | Not processed | Processed by creator's personal policy | +| Group collaboration | Not processed | Processed by scoped group policy | +| Converted source | Could be selected independently | Removed only through collaboration cleanup | +| Invalid activity timestamp | Container query determined behavior | Parsed and skipped safely | +| New group defaults | Missing from stored group document | Both values persisted as `"default"` | + +### User Experience + +Administrators can rely on configured personal and group retention periods across current and legacy conversation models. Shared conversations no longer remain indefinitely, and private group conversations no longer follow a creator's personal retention setting. diff --git a/docs/explanation/release_notes.md b/docs/explanation/release_notes.md index 3e70c72bf..ed1e373fa 100644 --- a/docs/explanation/release_notes.md +++ b/docs/explanation/release_notes.md @@ -13,6 +13,12 @@ For feature-focused and fix-focused drill-downs by version, see [Features by Ver #### Bug Fixes +* **Retention Coverage Across Group and Collaborative Conversations** + * Group-scoped private conversations now follow their primary group's retention policy instead of the creator's personal policy. + * Personal and group collaborative conversations now use their correct governing policy and activity timestamp, while linked conversion sources are cleaned once without duplicate counting. + * Collaboration cleanup now covers messages, per-user state, linked sources, blob-backed files, thoughts, activity logs, and conversation caches; new groups also persist explicit default retention values. + * (Ref: Closes #1054, `functions_retention_policy.py`, `functions_collaboration.py`, `functions_group.py`, `RETENTION_POLICY_CONVERSATION_SCOPE_COVERAGE_FIX.md`) + * **Custom Databricks-Prefixed Action Discovery** * Fixed action type discovery so custom plugin types such as `databricks_table_dscmo` no longer inherit the built-in Databricks discovery defaults. * Custom Databricks-prefixed plugin types now stay on the standard plugin configuration path and visual treatment unless their type is exactly `databricks` or `databricks_table`. diff --git a/functional_tests/test_retention_policy_conversation_scope_coverage.py b/functional_tests/test_retention_policy_conversation_scope_coverage.py new file mode 100644 index 000000000..91cf13ac6 --- /dev/null +++ b/functional_tests/test_retention_policy_conversation_scope_coverage.py @@ -0,0 +1,690 @@ +# test_retention_policy_conversation_scope_coverage.py +""" +Functional test for retention policy conversation scope coverage. +Version: 0.250.103 +Implemented in: 0.250.103 + +This test verifies the retention ownership matrix, timestamp safeguards, +collaboration cleanup, archival behavior, race handling, and new-group defaults. +""" + +import ast +import copy +from datetime import datetime, timezone +import os +import sys +import types +import uuid + + +REPO_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) +APP_ROOT = os.path.join(REPO_ROOT, 'application', 'single_app') +RETENTION_FILE = os.path.join(APP_ROOT, 'functions_retention_policy.py') +COLLABORATION_FILE = os.path.join(APP_ROOT, 'functions_collaboration.py') +GROUP_FILE = os.path.join(APP_ROOT, 'functions_group.py') + + +class FakeCosmosResourceNotFoundError(Exception): + """Minimal Cosmos not-found exception for isolated behavior tests.""" + + +class FakeContainer: + """In-memory Cosmos container with the operations used by cleanup helpers.""" + + def __init__(self, items=None): + self.items = { + item['id']: copy.deepcopy(item) + for item in (items or []) + } + self.deleted = [] + + def read_item(self, item=None, partition_key=None): + if item not in self.items: + raise FakeCosmosResourceNotFoundError(item) + return copy.deepcopy(self.items[item]) + + def query_items( + self, + query=None, + parameters=None, + partition_key=None, + enable_cross_partition_query=False, + ): + parameter_map = { + parameter['name']: parameter['value'] + for parameter in (parameters or []) + } + conversation_id = parameter_map.get('@conversation_id') + results = list(self.items.values()) + if conversation_id: + results = [ + item + for item in results + if item.get('conversation_id') == conversation_id + ] + return copy.deepcopy(results) + + def upsert_item(self, item): + persisted_item = copy.deepcopy(item) + persisted_item['_etag'] = str(uuid.uuid4()) + self.items[item['id']] = persisted_item + return copy.deepcopy(persisted_item) + + def replace_item( + self, + item=None, + body=None, + etag=None, + match_condition=None, + ): + if item not in self.items: + raise FakeCosmosResourceNotFoundError(item) + if etag and self.items[item].get('_etag') != etag: + conflict = RuntimeError('etag conflict') + conflict.status_code = 412 + raise conflict + return self.upsert_item(body) + + def create_item(self, item): + return self.upsert_item(item) + + def delete_item( + self, + item=None, + partition_key=None, + etag=None, + match_condition=None, + ): + if item not in self.items: + raise FakeCosmosResourceNotFoundError(item) + if etag and self.items[item].get('_etag') != etag: + conflict = RuntimeError('etag conflict') + conflict.status_code = 412 + raise conflict + self.deleted.append((item, partition_key)) + del self.items[item] + + +def load_source_members(path, function_names, assignment_names=None, namespace=None): + """Load selected source members without importing the application's config.""" + with open(path, 'r', encoding='utf-8') as handle: + tree = ast.parse(handle.read(), filename=path) + + selected_nodes = [] + assignment_names = set(assignment_names or []) + for node in tree.body: + if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)): + if node.name in function_names: + selected_nodes.append(node) + continue + if not isinstance(node, ast.Assign): + continue + assigned_names = { + target.id + for target in node.targets + if isinstance(target, ast.Name) + } + if assigned_names & assignment_names: + selected_nodes.append(node) + + loaded_namespace = dict(namespace or {}) + loaded_namespace.setdefault('__builtins__', __builtins__) + module = ast.Module(body=selected_nodes, type_ignores=[]) + exec(compile(module, path, 'exec'), loaded_namespace) + return loaded_namespace + + +def build_retention_namespace(): + container_names = ( + 'cosmos_conversations_container', + 'cosmos_messages_container', + 'cosmos_group_conversations_container', + 'cosmos_group_messages_container', + 'cosmos_collaboration_conversations_container', + 'cosmos_public_conversations_container', + 'cosmos_public_messages_container', + ) + namespace = { + 'datetime': datetime, + 'timezone': timezone, + } + namespace.update({ + container_name: object() + for container_name in container_names + }) + return load_source_members( + RETENTION_FILE, + { + 'resolve_retention_value', + '_parse_retention_timestamp', + '_get_primary_group_id', + '_is_group_single_user_conversation', + '_is_converted_conversation_source', + '_is_aged_conversation', + '_build_group_scope_query', + '_build_conversation_retention_sources', + }, + assignment_names={ + 'GROUP_SINGLE_USER_CHAT_TYPES', + 'PERSONAL_MULTI_USER_CHAT_TYPE', + 'GROUP_MULTI_USER_CHAT_TYPE', + }, + namespace=namespace, + ) + + +def test_policy_matrix_and_retention_values(): + """Verify each conversation shape maps to exactly one governing policy.""" + namespace = build_retention_namespace() + cutoff = datetime(2026, 1, 1, tzinfo=timezone.utc) + personal_sources = { + source['name']: source + for source in namespace['_build_conversation_retention_sources']( + 'personal', + cutoff.isoformat(), + user_id='user-1', + ) + } + group_sources = { + source['name']: source + for source in namespace['_build_conversation_retention_sources']( + 'group', + cutoff.isoformat(), + group_id='group-1', + ) + } + + personal_single = { + 'id': 'personal-single', + 'user_id': 'user-1', + 'chat_type': 'personal_single_user', + 'last_updated': '2025-01-01T00:00:00Z', + } + group_single = { + 'id': 'group-single', + 'user_id': 'user-1', + 'chat_type': 'group-single-user', + 'last_updated': '2025-01-01T00:00:00Z', + 'context': [ + {'type': 'primary', 'scope': 'group', 'id': 'group-1'}, + ], + } + personal_multi = { + 'id': 'personal-multi', + 'chat_type': 'personal_multi_user', + 'created_by_user_id': 'user-1', + 'updated_at': '2025-01-01T00:00:00Z', + } + group_multi = { + 'id': 'group-multi', + 'chat_type': 'group_multi_user', + 'created_by_user_id': 'user-1', + 'updated_at': '2025-01-01T00:00:00Z', + 'scope': {'type': 'group', 'group_id': 'group-1'}, + } + legacy_group = { + 'id': 'legacy-group', + 'group_id': 'group-1', + 'last_updated': '2025-01-01T00:00:00Z', + } + + assert personal_sources['personal_single_user']['matches_scope'](personal_single) + assert not personal_sources['personal_single_user']['matches_scope'](group_single) + assert personal_sources['personal_multi_user']['matches_scope'](personal_multi) + assert group_sources['group_single_user']['matches_scope'](group_single) + assert group_sources['group_multi_user']['matches_scope'](group_multi) + assert group_sources['legacy_group']['matches_scope'](legacy_group) + + converted_personal = dict(personal_single) + converted_personal['collaboration_conversation_id'] = 'personal-multi' + converted_group = dict(legacy_group) + converted_group['collaboration_conversation_id'] = 'group-multi' + assert not personal_sources['personal_single_user']['matches_scope']( + converted_personal + ) + assert not group_sources['legacy_group']['matches_scope'](converted_group) + + is_aged = namespace['_is_aged_conversation'] + assert is_aged(personal_single, 'last_updated', cutoff) + assert is_aged(personal_multi, 'updated_at', cutoff) + assert not is_aged( + {'last_updated': '2027-01-01T00:00:00Z'}, + 'last_updated', + cutoff, + ) + for invalid_value in (None, '', 'not-a-date', 123): + assert not is_aged( + {'last_updated': invalid_value}, + 'last_updated', + cutoff, + ) + + resolve_value = namespace['resolve_retention_value'] + settings = { + 'default_retention_conversation_personal': '30', + 'default_retention_conversation_group': '90', + } + assert resolve_value('default', 'personal', 'conversation', settings) == 30 + assert resolve_value(None, 'group', 'conversation', settings) == 90 + assert resolve_value('7', 'personal', 'conversation', settings) == 7 + assert resolve_value('none', 'group', 'conversation', settings) == 'none' + return True + + +def build_collaboration_cleanup_namespace( + collaboration_item, + collaboration_messages, + state_items, + source_item, + source_messages, + source_kind='personal', +): + """Create isolated cleanup dependencies and capture all side effects.""" + collaboration_container = FakeContainer([collaboration_item]) + collaboration_message_container = FakeContainer(collaboration_messages) + state_container = FakeContainer(state_items) + personal_source_container = FakeContainer( + [source_item] if source_kind == 'personal' else [] + ) + personal_source_messages = FakeContainer( + source_messages if source_kind == 'personal' else [] + ) + group_source_container = FakeContainer( + [source_item] if source_kind == 'group' else [] + ) + group_source_messages = FakeContainer( + source_messages if source_kind == 'group' else [] + ) + archived_conversations = FakeContainer() + archived_messages = FakeContainer() + effects = { + 'blob_message_ids': [], + 'archived_thoughts': [], + 'deleted_thoughts': [], + 'sharing_syncs': [], + 'cache_invalidations': [], + 'archival_logs': [], + 'deletion_logs': [], + } + + namespace = { + 'deepcopy': copy.deepcopy, + 'datetime': datetime, + 'timezone': timezone, + 'CosmosResourceNotFoundError': FakeCosmosResourceNotFoundError, + 'PERSONAL_MULTI_USER_CHAT_TYPE': 'personal_multi_user', + 'GROUP_MULTI_USER_CHAT_TYPE': 'group_multi_user', + 'cosmos_collaboration_conversations_container': collaboration_container, + 'cosmos_collaboration_messages_container': collaboration_message_container, + 'cosmos_collaboration_user_state_container': state_container, + 'cosmos_conversations_container': personal_source_container, + 'cosmos_messages_container': personal_source_messages, + 'cosmos_group_conversations_container': group_source_container, + 'cosmos_group_messages_container': group_source_messages, + 'cosmos_archived_conversations_container': archived_conversations, + 'cosmos_archived_messages_container': archived_messages, + 'log_event': lambda *args, **kwargs: None, + 'log_conversation_archival': ( + lambda **kwargs: effects['archival_logs'].append(copy.deepcopy(kwargs)) + ), + 'log_conversation_deletion': ( + lambda **kwargs: effects['deletion_logs'].append(copy.deepcopy(kwargs)) + ), + '_delete_blob_backed_collaboration_files': ( + lambda messages: effects['blob_message_ids'].extend( + message.get('id') + for message in messages + ) + ), + 'archive_thoughts_for_conversation': ( + lambda conversation_id, user_id, **kwargs: effects['archived_thoughts'].append( + (conversation_id, user_id) + ) + ), + 'delete_thoughts_for_conversation': ( + lambda conversation_id, user_id, **kwargs: effects['deleted_thoughts'].append( + (conversation_id, user_id) + ) + ), + 'sync_chat_upload_workspace_document_sharing_for_collaboration': ( + lambda conversation: effects['sharing_syncs'].append( + copy.deepcopy(conversation) + ) + ), + 'invalidate_conversation_cache_for_item': ( + lambda item, reason: effects['cache_invalidations'].append( + (item.get('id'), reason) + ) + ), + 'is_personal_collaboration_conversation': ( + lambda item: item.get('chat_type') == 'personal_multi_user' + ), + 'get_collaboration_conversation': ( + lambda conversation_id: collaboration_container.read_item( + item=conversation_id, + partition_key=conversation_id, + ) + ), + } + loaded = load_source_members( + COLLABORATION_FILE, + { + '_archive_collaboration_item', + '_delete_item_if_present', + '_collaboration_retention_identity', + '_read_collaboration_conversation_for_retention', + '_cleanup_collaboration_thoughts', + '_cleanup_linked_collaboration_source', + '_delete_collaboration_conversation_records', + 'delete_collaboration_conversation_for_retention', + }, + namespace=namespace, + ) + loaded['effects'] = effects + loaded['containers'] = { + 'collaboration': collaboration_container, + 'collaboration_messages': collaboration_message_container, + 'states': state_container, + 'personal_source': personal_source_container, + 'personal_source_messages': personal_source_messages, + 'group_source': group_source_container, + 'group_source_messages': group_source_messages, + 'archived_conversations': archived_conversations, + 'archived_messages': archived_messages, + } + return loaded + + +def test_destructive_collaboration_cleanup_and_race_handling(): + """Verify destructive retention removes collaboration and linked source data.""" + conversation_id = 'personal-multi-1' + source_id = 'personal-source-1' + collaboration_item = { + 'id': conversation_id, + 'title': 'Shared personal conversation', + 'chat_type': 'personal_multi_user', + 'created_by_user_id': 'owner-1', + 'updated_at': '2025-01-01T00:00:00Z', + 'source_conversation_id': source_id, + 'accepted_participant_ids': ['owner-1', 'member-1'], + 'participants': [ + {'user_id': 'owner-1'}, + {'user_id': 'member-1'}, + ], + 'owner_user_ids': ['owner-1'], + 'context': [], + 'tags': [], + } + source_item = { + 'id': source_id, + 'title': 'Hidden source', + 'user_id': 'owner-1', + 'collaboration_conversation_id': conversation_id, + } + namespace = build_collaboration_cleanup_namespace( + collaboration_item, + [{ + 'id': 'collaboration-message-1', + 'conversation_id': conversation_id, + 'file_content_source': 'blob', + 'blob_container': 'generated', + 'blob_path': 'collaboration/file.txt', + }], + [ + { + 'id': 'owner-state', + 'conversation_id': conversation_id, + 'user_id': 'owner-1', + }, + { + 'id': 'member-state', + 'conversation_id': conversation_id, + 'user_id': 'member-1', + }, + ], + source_item, + [{ + 'id': 'source-message-1', + 'conversation_id': source_id, + 'file_content_source': 'blob', + 'blob_container': 'generated', + 'blob_path': 'source/file.txt', + }], + ) + + detail = namespace['delete_collaboration_conversation_for_retention']( + collaboration_item, + workspace_type='personal', + archiving_enabled=False, + ) + containers = namespace['containers'] + effects = namespace['effects'] + + assert detail['id'] == conversation_id + assert not containers['collaboration'].items + assert not containers['collaboration_messages'].items + assert not containers['states'].items + assert not containers['personal_source'].items + assert not containers['personal_source_messages'].items + assert set(effects['blob_message_ids']) == { + 'collaboration-message-1', + 'source-message-1', + } + assert effects['sharing_syncs'][0]['accepted_participant_ids'] == [] + assert (conversation_id, 'owner-1') in effects['deleted_thoughts'] + assert (conversation_id, 'member-1') in effects['deleted_thoughts'] + assert (source_id, 'owner-1') in effects['deleted_thoughts'] + assert (source_id, 'member-1') in effects['deleted_thoughts'] + assert {item_id for item_id, _ in effects['cache_invalidations']} == { + conversation_id, + source_id, + } + assert effects['deletion_logs'][0]['is_bulk_operation'] is True + + race_detail = namespace['delete_collaboration_conversation_for_retention']( + collaboration_item, + workspace_type='personal', + archiving_enabled=False, + ) + assert race_detail['already_deleted'] is True + return True + + +def test_collaboration_archival_covers_group_source_records(): + """Verify archival preserves collaboration and linked legacy group records.""" + conversation_id = 'group-multi-1' + source_id = 'legacy-group-source-1' + collaboration_item = { + 'id': conversation_id, + 'title': 'Shared group conversation', + 'chat_type': 'group_multi_user', + 'created_by_user_id': 'owner-1', + 'updated_at': '2025-01-01T00:00:00Z', + 'legacy_source_conversation_id': source_id, + 'participants': [{'user_id': 'owner-1'}], + 'scope': {'type': 'group', 'group_id': 'group-1'}, + 'context': [], + 'tags': [], + } + source_item = { + 'id': source_id, + 'title': 'Legacy group source', + 'user_id': 'owner-1', + 'group_id': 'group-1', + 'collaboration_conversation_id': conversation_id, + } + namespace = build_collaboration_cleanup_namespace( + collaboration_item, + [{ + 'id': 'group-collaboration-message', + 'conversation_id': conversation_id, + }], + [{ + 'id': 'group-owner-state', + 'conversation_id': conversation_id, + 'user_id': 'owner-1', + }], + source_item, + [{ + 'id': 'legacy-group-message', + 'conversation_id': source_id, + }], + source_kind='group', + ) + + namespace['delete_collaboration_conversation_for_retention']( + collaboration_item, + workspace_type='group', + archiving_enabled=True, + ) + containers = namespace['containers'] + effects = namespace['effects'] + + assert set(containers['archived_conversations'].items) == { + conversation_id, + source_id, + } + assert set(containers['archived_messages'].items) == { + 'group-collaboration-message', + 'legacy-group-message', + } + archived_conversation = containers['archived_conversations'].items[ + conversation_id + ] + assert archived_conversation['collaboration_user_states'][0]['id'] == ( + 'group-owner-state' + ) + assert not effects['blob_message_ids'] + assert effects['archival_logs'][0]['workspace_type'] == 'group' + assert effects['archival_logs'][0]['group_id'] == 'group-1' + assert (conversation_id, 'owner-1') in effects['archived_thoughts'] + assert (source_id, 'owner-1') in effects['archived_thoughts'] + return True + + +def test_collaboration_revalidation_and_cleanup_failure_safety(): + """Verify changed conversations are skipped and failed cleanup preserves records.""" + conversation_id = 'personal-multi-safety' + collaboration_item = { + 'id': conversation_id, + 'title': 'Safety test', + 'chat_type': 'personal_multi_user', + 'created_by_user_id': 'owner-1', + 'updated_at': '2025-01-01T00:00:00Z', + 'participants': [{'user_id': 'owner-1'}], + 'accepted_participant_ids': ['owner-1'], + 'context': [], + 'tags': [], + } + namespace = build_collaboration_cleanup_namespace( + collaboration_item, + [{ + 'id': 'safety-message', + 'conversation_id': conversation_id, + 'file_content_source': 'blob', + 'blob_container': 'generated', + 'blob_path': 'safety/file.txt', + }], + [{ + 'id': 'safety-state', + 'conversation_id': conversation_id, + 'user_id': 'owner-1', + }], + {'id': 'unused-source', 'user_id': 'owner-1'}, + [], + ) + containers = namespace['containers'] + + changed_item = copy.deepcopy( + containers['collaboration'].items[conversation_id] + ) + changed_item['updated_at'] = '2026-01-02T00:00:00Z' + containers['collaboration'].upsert_item(changed_item) + skipped_detail = namespace['delete_collaboration_conversation_for_retention']( + collaboration_item, + workspace_type='personal', + archiving_enabled=False, + ) + assert skipped_detail is None + assert conversation_id in containers['collaboration'].items + assert 'safety-message' in containers['collaboration_messages'].items + + containers['collaboration'].upsert_item(collaboration_item) + namespace['_delete_blob_backed_collaboration_files'] = ( + lambda messages: (_ for _ in ()).throw(RuntimeError('blob cleanup failed')) + ) + try: + namespace['delete_collaboration_conversation_for_retention']( + collaboration_item, + workspace_type='personal', + archiving_enabled=False, + ) + raise AssertionError('Expected blob cleanup failure') + except RuntimeError as error: + assert str(error) == 'blob cleanup failed' + + assert conversation_id in containers['collaboration'].items + assert 'safety-message' in containers['collaboration_messages'].items + assert 'safety-state' in containers['states'].items + return True + + +def test_new_groups_persist_default_retention_values(): + """Verify create_group writes explicit default retention settings.""" + group_container = FakeContainer() + cache_reasons = [] + namespace = load_source_members( + GROUP_FILE, + {'create_group'}, + namespace={ + 'uuid': uuid, + 'datetime': datetime, + 'DEFAULT_WORKSPACE_HERO_COLOR': '#0d6efd', + 'functions_authentication': types.SimpleNamespace( + get_current_user_info=lambda: { + 'userId': 'owner-1', + 'email': 'owner@example.com', + 'displayName': 'Owner User', + } + ), + 'cosmos_groups_container': group_container, + 'bump_chat_bootstrap_global_cache_version': ( + lambda reason: cache_reasons.append(reason) + ), + }, + ) + + created_group = namespace['create_group']('Retention Group', 'Test group') + assert created_group['retention_policy'] == { + 'conversation_retention_days': 'default', + 'document_retention_days': 'default', + } + assert group_container.items[created_group['id']]['retention_policy'] == ( + created_group['retention_policy'] + ) + assert cache_reasons == ['group_created'] + return True + + +if __name__ == '__main__': + tests = [ + test_policy_matrix_and_retention_values, + test_destructive_collaboration_cleanup_and_race_handling, + test_collaboration_archival_covers_group_source_records, + test_collaboration_revalidation_and_cleanup_failure_safety, + test_new_groups_persist_default_retention_values, + ] + results = [] + for test in tests: + try: + test() + print(f"PASS: {test.__name__}") + results.append(True) + except Exception as error: + print(f"FAIL: {test.__name__}: {error}") + import traceback + traceback.print_exc() + results.append(False) + + print(f"{sum(results)}/{len(results)} retention scope tests passed") + sys.exit(0 if all(results) else 1)