Fix shared conversation collaboration and govern agent model picker

application/single_app/app_settings_cache.py

@@ -14,6 +14,7 @@
 from datetime import datetime, timedelta
 from typing import Callable
 
+from azure.core import MatchConditions
 from azure.core.exceptions import AzureError
 from azure.cosmos import ContainerProxy
 from redis import Redis
@@ -49,6 +50,7 @@
 CACHE_VERSION_READ_TTL_SECONDS = 15
 COSMOS_CACHE_ENTRY_DOC_TYPE = 'app_cache_entry'
 COSMOS_CACHE_ENTRY_PREFIX = 'app_cache_entry:'
+STREAM_SESSION_APPEND_MAX_ATTEMPTS = 5
 update_settings_cache = None
 get_settings_cache = None
 get_app_settings_cache_version = None
@@ -375,13 +377,18 @@
 
 def _initialize_stream_session_cache_fallback(cache_key, metadata, ttl_seconds=None, log_event_func=None):
     expiration_timestamp = _get_expiration_timestamp(ttl_seconds)
+    existing_events = _get_cosmos_cache_entry(
+        _get_stream_session_events_key(cache_key),
+        log_event_func=log_event_func,
+    )
+    events_payload = list(existing_events) if isinstance(existing_events, list) else []
     with _app_cache_lock:
         APP_STREAM_SESSION_METADATA[cache_key] = {
             'value': copy.deepcopy(metadata or {}),
             'expires_at': expiration_timestamp,
         }
         APP_STREAM_SESSION_EVENTS[cache_key] = {
-            'value': [],
+            'value': events_payload,
             'expires_at': expiration_timestamp,
         }
     _set_cosmos_cache_entry(
@@ -390,12 +397,30 @@
         ttl_seconds=ttl_seconds,
         log_event_func=log_event_func,
     )
-    _set_cosmos_cache_entry(
-        _get_stream_session_events_key(cache_key),
-        [],
-        ttl_seconds=ttl_seconds,
-        log_event_func=log_event_func,
-    )
+    if existing_events is None:
+        now = datetime.utcnow()
+        expires_at = None
+        if ttl_seconds is not None:
+            expires_at = now + timedelta(seconds=max(int(ttl_seconds), 0))
+        try:
+            _get_cosmos_cache_container().create_item({
+                'id': _build_cosmos_cache_doc_id(_get_stream_session_events_key(cache_key)),
+                'type': COSMOS_CACHE_ENTRY_DOC_TYPE,
+                'cache_key': _get_stream_session_events_key(cache_key),
+                'payload': [],
+                'created_at': now.isoformat(),
+                'updated_at': now.isoformat(),
+                'expires_at': _serialize_datetime(expires_at),
+            })
+        except AzureError as ex:
+            if getattr(ex, 'status_code', None) != 409:
+                _logger.warning("[ASC] Cosmos stream event initialization failed for %s: %s", cache_key, ex)
+                if callable(log_event_func):
+                    log_event_func(
+                        "[ASC] Cosmos stream event initialization failed.",
+                        extra={'cache_key': cache_key, 'error': str(ex)},
+                        level=logging.WARNING,
+                    )
 
 
 def _set_stream_session_meta_fallback(cache_key, metadata, ttl_seconds=None, log_event_func=None):
@@ -430,37 +455,94 @@
 
 def _append_stream_session_event_fallback(cache_key, event_text, ttl_seconds=None, log_event_func=None):
     expiration_timestamp = _get_expiration_timestamp(ttl_seconds)
+    cache_entry_key = _get_stream_session_events_key(cache_key)
+    doc_id = _build_cosmos_cache_doc_id(cache_entry_key)
+    container = _get_cosmos_cache_container()
+    events_payload = None
+
+    for _ in range(STREAM_SESSION_APPEND_MAX_ATTEMPTS):
+        now = datetime.utcnow()
+        expires_at = None
+        if ttl_seconds is not None:
+            expires_at = now + timedelta(seconds=max(int(ttl_seconds), 0))
+        try:
+            document = container.read_item(item=doc_id, partition_key=doc_id)
+            events_payload = list(document.get('payload') or [])
+            events_payload.append(event_text)
+            document['payload'] = events_payload
+            document['updated_at'] = now.isoformat()
+            document['expires_at'] = _serialize_datetime(expires_at)
+            container.replace_item(
+                item=doc_id,
+                body=document,
+                etag=document.get('_etag'),
+                match_condition=MatchConditions.IfNotModified,
+            )
+        except AzureError as ex:
+            status_code = getattr(ex, 'status_code', None)
+            if status_code in {404, 409, 412}:
+                if status_code == 404:
+                    try:
+                        container.create_item({
+                            'id': doc_id,
+                            'type': COSMOS_CACHE_ENTRY_DOC_TYPE,
+                            'cache_key': cache_entry_key,
+                            'payload': [event_text],
+                            'created_at': now.isoformat(),
+                            'updated_at': now.isoformat(),
+                            'expires_at': _serialize_datetime(expires_at),
+                        })
+                    except AzureError as create_error:
+                        if getattr(create_error, 'status_code', None) in {409, 412}:
+                            continue
+                        ex = create_error
+                    else:
+                        events_payload = [event_text]
+                        break
+                else:
+                    continue
+            _logger.warning("[ASC] Cosmos stream event append failed for %s: %s", cache_key, ex)
+            if callable(log_event_func):
+                log_event_func(
+                    "[ASC] Cosmos stream event append failed.",
+                    extra={'cache_key': cache_key, 'error': str(ex), 'status_code': status_code},
+                    level=logging.WARNING,
+                )
+            break
+        else:
+            break
+    else:
+        events_payload = None
+
+    if not isinstance(events_payload, list):
+        with _app_cache_lock:
+            entry = APP_STREAM_SESSION_EVENTS.get(cache_key)
+            events_payload = list(entry.get('value') or []) if entry and not _is_expired(entry) else []
+            events_payload.append(event_text)
+
     with _app_cache_lock:
-        entry = APP_STREAM_SESSION_EVENTS.get(cache_key)
-        if _is_expired(entry):
-            entry = {'value': [], 'expires_at': expiration_timestamp}
-            APP_STREAM_SESSION_EVENTS[cache_key] = entry
-        entry['value'].append(event_text)
-        if expiration_timestamp is not None:
-            entry['expires_at'] = expiration_timestamp
-        events_payload = list(entry.get('value') or [])
-    _set_cosmos_cache_entry(
-        _get_stream_session_events_key(cache_key),
-        events_payload,
-        ttl_seconds=ttl_seconds,
-        log_event_func=log_event_func,
-    )
+        APP_STREAM_SESSION_EVENTS[cache_key] = {
+            'value': events_payload,
+            'expires_at': expiration_timestamp,
+        }
 
 
 def _get_stream_session_events_fallback(cache_key, start_index=0, log_event_func=None):
     start = int(start_index or 0)
+    cached = _get_cosmos_cache_entry(
+        _get_stream_session_events_key(cache_key),
+        log_event_func=log_event_func,
+    )
+    if cached is not None:
+        return list(cached[start:])
+
     with _app_cache_lock:
         entry = APP_STREAM_SESSION_EVENTS.get(cache_key)
         if entry and not _is_expired(entry):
             return list((entry.get('value') or [])[start:])
         if entry:
             APP_STREAM_SESSION_EVENTS.pop(cache_key, None)
-
-    cached = _get_cosmos_cache_entry(
-        _get_stream_session_events_key(cache_key),
-        log_event_func=log_event_func,
-    )
-    return list((cached or [])[start:])
+    return []
 
 
 def _delete_stream_session_cache_fallback(cache_key, log_event_func=None):
@@ -651,9 +733,7 @@
         def initialize_stream_session_cache_redis(cache_key, metadata, ttl_seconds=None):
             try:
                 metadata_key = get_stream_session_metadata_key(cache_key)
-                events_key = get_stream_session_events_key(cache_key)
                 pipeline = redis_client.pipeline()
-                pipeline.delete(events_key)
                 pipeline.set(metadata_key, json.dumps(metadata))
                 if ttl_seconds is not None:
                     pipeline.expire(metadata_key, int(ttl_seconds))
@@ -949,12 +1029,32 @@
             _set_ttl_cached_version(APP_GOVERNANCE_SHARED_VERSION_CACHE, fallback_version)
             return fallback_version
 
-        initialize_stream_session_cache = initialize_stream_session_cache_mem
-        set_stream_session_meta = set_stream_session_meta_mem
-        get_stream_session_meta = get_stream_session_meta_mem
-        append_stream_session_event = append_stream_session_event_mem
-        get_stream_session_events = get_stream_session_events_mem
-        delete_stream_session_cache = delete_stream_session_cache_mem
+        # Stream sessions must be visible across application workers. The Cosmos
+        # fallback preserves that contract when Redis is intentionally disabled.
+        initialize_stream_session_cache = lambda cache_key, metadata, ttl_seconds=None: (
+            _initialize_stream_session_cache_fallback(
+                cache_key, metadata, ttl_seconds=ttl_seconds, log_event_func=log_event,
+            )
+        )
+        set_stream_session_meta = lambda cache_key, metadata, ttl_seconds=None: (
+            _set_stream_session_meta_fallback(
+                cache_key, metadata, ttl_seconds=ttl_seconds, log_event_func=log_event,
+            )
+        )
+        get_stream_session_meta = lambda cache_key: _get_stream_session_meta_fallback(
+            cache_key, log_event_func=log_event,
+        )
+        append_stream_session_event = lambda cache_key, event_text, ttl_seconds=None: (
+            _append_stream_session_event_fallback(
+                cache_key, event_text, ttl_seconds=ttl_seconds, log_event_func=log_event,
+            )
+        )
+        get_stream_session_events = lambda cache_key, start_index=0: _get_stream_session_events_fallback(
+            cache_key, start_index=start_index, log_event_func=log_event,
+        )
+        delete_stream_session_cache = lambda cache_key: _delete_stream_session_cache_fallback(
+            cache_key, log_event_func=log_event,
+        )
         get_user_ui_settings_cache = get_user_ui_settings_cache_mem
         set_user_ui_settings_cache = set_user_ui_settings_cache_mem
         delete_user_ui_settings_cache = delete_user_ui_settings_cache_mem

application/single_app/config.py

@@ -98,7 +98,7 @@
 EXECUTOR_TYPE = 'thread'
 EXECUTOR_MAX_WORKERS = 30
 SESSION_TYPE = 'filesystem'
-VERSION = "0.261.047"
+VERSION = "0.261.048"
 IS_DEVELOPMENT = is_development_env_enabled()
 
 # Opt-out for deployments where App Service Easy Auth is active but the platform

application/single_app/route_backend_agents.py

@@ -57,7 +57,7 @@
     log_agent_deletion,
     log_general_admin_action,
 )
-from functions_governance import ensure_governance_access, upsert_item_policy
+from functions_governance import ensure_governance_access, filter_governed_model_endpoints, upsert_item_policy
 from functions_model_endpoint_identity_header import build_model_endpoint_identity_headers
 
 bpa = Blueprint('admin_agents', __name__)
@@ -2152,6 +2152,12 @@ def orchestration_settings():
 def build_combined_model_endpoints(settings, user_id=None, group_id=None):
     endpoints = []
     global_endpoints = settings.get("model_endpoints", []) or []
+    if user_id:
+        global_endpoints = filter_governed_model_endpoints(
+            user_id,
+            global_endpoints,
+            "governance_global_endpoints",
+        )
     for endpoint in global_endpoints:
         enriched = dict(endpoint)
         enriched["scope"] = "global"
@@ -2164,7 +2170,11 @@ def build_combined_model_endpoints(settings, user_id=None, group_id=None):
         if allow_group_custom_endpoints:
             try:
                 ensure_governance_access("governance_group_endpoints", user_id)
-                group_endpoints = get_group_model_endpoints(group_id)
+                group_endpoints = filter_governed_model_endpoints(
+                    user_id,
+                    get_group_model_endpoints(group_id),
+                    "governance_group_endpoints",
+                )
                 for endpoint in group_endpoints:
                     enriched = dict(endpoint)
                     enriched["scope"] = "group"
@@ -2177,7 +2187,11 @@ def build_combined_model_endpoints(settings, user_id=None, group_id=None):
             try:
                 ensure_governance_access("governance_user_endpoints", user_id)
                 user_settings = get_user_settings(user_id)
-                personal = user_settings.get("settings", {}).get("personal_model_endpoints", [])
+                personal = filter_governed_model_endpoints(
+                    user_id,
+                    user_settings.get("settings", {}).get("personal_model_endpoints", []),
+                    "governance_user_endpoints",
+                )
                 for endpoint in personal:
                     enriched = dict(endpoint)
                     enriched["scope"] = "user"

application/single_app/route_backend_collaboration.py

@@ -168,6 +168,11 @@ def publish(self, event_payload):
         with self._condition:
             self._condition.notify_all()
 
+    def get_event_cursor(self):
+        """Return the next event index before the caller reads persisted history."""
+        self.initialize()
+        return len(app_settings_cache.get_stream_session_events(self.cache_key, start_index=0) or [])
+
     def iter_events(self, start_index=0):
         self.initialize()
         next_index = max(int(start_index or 0), 0)
@@ -309,18 +314,17 @@ def _collaboration_events_for_viewer(events, viewer_user_id, conversation_id):
                 messages = hydrate_m365_pending_action_cards(
                     [payload['message']], viewer_user_id, conversation_id,
                 )
-            except (M365PolicyError, PermissionError, AzureError, ValueError) as error:
+            except Exception as error:
                 log_event(
                     '[COLLABORATION] Shared Microsoft 365 action cards could not be projected.',
                     extra={'conversation_id': conversation_id, 'exception_type': type(error).__name__},
                     level=logging.WARNING,
                 )
-                failure = _build_collaboration_event(
-                    conversation_id, 'collaboration.m365.pending_action',
-                    {'error': 'm365_pending_actions_unavailable'},
-                )
-                yield f'data: {json.dumps(failure)}\n\n'
-                return
+                # M365 card projection is optional. Preserve the collaboration
+                # message event so one viewer's Graph failure cannot terminate
+                # their live conversation stream.
+                yield event_text
+                continue
             event = {**event, 'payload': {**payload, 'message': messages[0]}}
             yield f'data: {json.dumps(make_json_serializable(event))}\n\n'
         else:
@@ -1348,10 +1352,11 @@ def get_collaboration_messages_api(conversation_id):
                 conversation_doc,
                 allow_pending=True,
             )
+            event_cursor = COLLABORATION_EVENT_REGISTRY.get_session(conversation_id).get_event_cursor()
             messages = [serialize_collaboration_message(doc) for doc in list_collaboration_messages(conversation_id)]
             attach_generated_file_approval_state(messages, current_user['user_id'])
             messages = hydrate_m365_pending_action_cards(messages, current_user['user_id'], conversation_id)
-            return jsonify({'messages': messages}), 200
+            return jsonify({'messages': messages, 'event_cursor': event_cursor}), 200
         except CosmosResourceNotFoundError:
             return jsonify({'error': 'Collaborative conversation not found'}), 404
         except PermissionError as exc:

application/single_app/route_backend_conversations.py

@@ -73,7 +73,7 @@
 def normalize_chat_type(conversation_item):
     chat_type = conversation_item.get('chat_type')
     if chat_type:
-        if chat_type == 'personal':
+        if chat_type in ('personal', 'new'):
             conversation_item['chat_type'] = 'personal_single_user'
             return conversation_item['chat_type'], True
         return chat_type, False

application/single_app/static/js/agents_common.js

@@ -689,7 +689,7 @@ export function getAvailableModels({ apimEnabled, settings, agent }) {
 	let selectedModel = null;
 	const endpoints = Array.isArray(settings?.model_endpoints) ? settings.model_endpoints : [];
 	const multiEndpointEnabled = (settings && settings.enable_multi_model_endpoints) || endpoints.length > 0;
-	if (multiEndpointEnabled && endpoints.length) {
+	if (multiEndpointEnabled) {
 		const agentType = (agent && agent.agent_type) ? agent.agent_type : 'local';
 		endpoints.forEach(endpoint => {
 			if (!endpoint || endpoint.enabled === false) return;