Fix live user message metadata during streaming

application/single_app/config.py

@@ -96,7 +96,7 @@
 EXECUTOR_TYPE = 'thread'
 EXECUTOR_MAX_WORKERS = 30
 SESSION_TYPE = 'filesystem'
-VERSION = "0.250.201"
+VERSION = "0.250.202"
 IS_DEVELOPMENT = is_development_env_enabled()
 
 SESSION_COOKIE_SAMESITE = os.getenv('SESSION_COOKIE_SAMESITE', 'Lax')

application/single_app/functions_chat_stream_events.py

@@ -0,0 +1,37 @@
+# functions_chat_stream_events.py
+
+import json
+from typing import Any, Dict
+
+
+USER_MESSAGE_PERSISTED_EVENT_TYPE = "user_message_persisted"
+
+
+def build_user_message_persisted_stream_payload(
+    conversation_id: str,
+    user_message_id: str,
+) -> Dict[str, Any]:
+    """Build the SSE payload that acknowledges durable user-message storage."""
+    normalized_conversation_id = str(conversation_id or "").strip()
+    normalized_user_message_id = str(user_message_id or "").strip()
+    if not normalized_conversation_id or not normalized_user_message_id:
+        raise ValueError("conversation_id and user_message_id are required")
+
+    return {
+        "type": USER_MESSAGE_PERSISTED_EVENT_TYPE,
+        "conversation_id": normalized_conversation_id,
+        "user_message_id": normalized_user_message_id,
+        "message_persisted": True,
+    }
+
+
+def build_user_message_persisted_stream_event(
+    conversation_id: str,
+    user_message_id: str,
+) -> str:
+    """Serialize a user-message persistence acknowledgement as an SSE event."""
+    payload = build_user_message_persisted_stream_payload(
+        conversation_id,
+        user_message_id,
+    )
+    return f"data: {json.dumps(payload)}\n\n"

application/single_app/route_backend_chats.py

@@ -116,6 +116,7 @@
 from functions_global_agents import get_global_agents
 from functions_group_agents import get_group_agents
 from functions_personal_agents import get_personal_agents
+from functions_chat_stream_events import build_user_message_persisted_stream_event
 from functions_source_review import (
     build_deep_research_ledger,
     build_deep_research_ledger_markdown,
@@ -15020,6 +15021,13 @@ def execute_document_action_chat_request(
             'metadata': user_metadata,
         })
         cosmos_messages_container.upsert_item(user_message_doc)
+        if callable(publish_background_event):
+            publish_background_event(
+                build_user_message_persisted_stream_event(
+                    conversation_id,
+                    user_message_id,
+                )
+            )
 
         try:
             document_action_activity_context = {
@@ -15880,6 +15888,11 @@ def generate_image_from_proposal():
     @login_required
     @user_required
     def chat_api():
+        publish_background_event = getattr(
+            g,
+            'chat_publish_background_event',
+            None,
+        )
         try:
             request_start_time = time.time()
             settings = get_settings()
@@ -16932,6 +16945,13 @@ def result_requires_message_reload(result: Any) -> bool:
                 # Note: Message-level chat_type will be updated after document search
 
                 cosmos_messages_container.upsert_item(user_message_doc)
+                if callable(publish_background_event):
+                    publish_background_event(
+                        build_user_message_persisted_stream_event(
+                            conversation_id,
+                            user_message_id,
+                        )
+                    )
 
                 # Log chat activity for real-time tracking
                 try:
@@ -20099,7 +20119,7 @@ def normalize_legacy_chat_payload(payload):
                 'blocked': payload.get('blocked', False),
             })
 
-        def generate_compatibility_response():
+        def generate_compatibility_response(publish_background_event=None):
             """Bridge legacy JSON chat handling into a terminal SSE event for parity cases."""
             try:
                 g.conversation_id = finalized_conversation_id
@@ -20122,6 +20142,7 @@ def generate_compatibility_response():
                     }
                     yield f"data: {json.dumps(image_request_event)}\n\n"
 
+                g.chat_publish_background_event = publish_background_event
                 legacy_result = chat_api()
                 legacy_response = legacy_result
                 status_code = 200
@@ -21205,6 +21226,10 @@ def build_streaming_capability_usage():
                     }
 
                     cosmos_messages_container.upsert_item(user_message_doc)
+                    yield build_user_message_persisted_stream_event(
+                        conversation_id,
+                        user_message_id,
+                    )
                     debug_print(
                         f"[STREAMING] Saved user message {user_message_id} | thread_id={current_user_thread_id} | previous_thread_id={previous_thread_id}"
                     )

application/single_app/route_backend_collaboration.py

@@ -45,6 +45,10 @@
     update_personal_collaboration_title,
 )
 from functions_conversation_cache import bump_conversation_cache_version
+from functions_chat_stream_events import (
+    USER_MESSAGE_PERSISTED_EVENT_TYPE,
+    build_user_message_persisted_stream_event,
+)
 from functions_group import assert_group_role, check_group_status_allows_operation, find_group_by_id, require_active_group
 from functions_image_messages import decode_image_content, get_complete_image_content, is_blob_backed_image_message, is_external_image_url
 from functions_message_masking import (
@@ -1449,6 +1453,10 @@ def stream_collaboration_message_api(conversation_id):
 
             def generate_stream():
                 try:
+                    yield build_user_message_persisted_stream_event(
+                        conversation_id,
+                        serialized_user_message.get('id'),
+                    )
                     internal_stream_view = current_app.view_functions.get('chat_stream_api')
                     if not callable(internal_stream_view):
                         yield _serialize_stream_error(
@@ -1508,6 +1516,9 @@ def transform_event_block(event_block):
                                     conversation_id=conversation_id,
                                 )
 
+                            if stream_payload.get('type') == USER_MESSAGE_PERSISTED_EVENT_TYPE:
+                                return None
+
                             if not stream_payload.get('done'):
                                 return normalized_event_block + '\n\n'
 

application/single_app/static/js/chat/chat-collaboration.js

@@ -5,6 +5,7 @@ import {
     getCollaborativeTagSuggestions,
     getGeneratedImageProposalSourceMessageId,
     groupGeneratedImageProposalMessages,
+    setUserMessageStreamingActionsDisabled,
     updateSendButtonVisibility,
     updateUserMessageId,
     userInput,
@@ -1074,6 +1075,12 @@ function handleConversationEvent(eventEnvelope = {}) {
         const decoratedMessage = decorateReplyMessage(payload.message);
         cacheCollaborationMessage(payload.message);
         if (reconcilePendingCollaborativeUserMessage(payload.message)) {
+            const messageKind = String(
+                payload.message.message_kind || payload.message.metadata?.message_kind || ''
+            ).trim();
+            if (senderUserId === getCurrentUserId() && messageKind !== 'ai_request') {
+                setUserMessageStreamingActionsDisabled(payload.message.id, false);
+            }
             if (shouldClearNotifications) {
                 void markCollaborationConversationRead(eventEnvelope.conversation_id || payload.message.conversation_id, {
                     suppressErrorToast: true,
@@ -1308,6 +1315,7 @@ async function sendCollaborativeMessage(messageText, tempMessageId = null) {
         if (!reconcilePendingCollaborativeUserMessage(payload.message, tempMessageId)) {
             renderCollaborationMessage(decorateReplyMessage(payload.message), { isNewMessage: true });
         }
+        setUserMessageStreamingActionsDisabled(payload.message.id, false);
     }
 
     setTypingState(false, { force: true });