Fix private action credential recovery in shared conversations

application/single_app/agent_delegation_runtime.py

@@ -563,7 +563,7 @@
         return {"citations": _delegation_records_citations(records), "usage": _delegation_records_usage(records)}
 
     async def _run(self, messages, *, streaming=False, **kwargs):
-        from functions_action_credentials import _continuation, credential_continuation
+        from functions_action_credentials import _continuation, capture_chat_origin, credential_continuation
         from functions_action_credential_continuation import invoke_credential_agent, stream_credential_agent
 
         if self.prevent_replay and self._invocations and self.budget.attempts:
@@ -609,6 +609,7 @@
                     "mode": "chat", "checkpoint_id": self.credential_checkpoint_id,
                     "agent_reference": deepcopy(self.reference),
                     "construction_profile": getattr(selected, "construction_profile", "agent-core"),
+                    **capture_chat_origin(self.identity.user_id, self.identity.conversation_id),
                 }
             if streaming:
                 with agent_execution(frame), credential_continuation(descriptor):

application/single_app/config.py

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

application/single_app/functions_action_credential_continuation.py

@@ -20,7 +20,7 @@
 from agent_execution_context import current_agent_execution
 from functions_action_manifest import get_action_origin
 import functions_action_credential_state as state
-from functions_action_credentials import _continuation
+from functions_action_credentials import _continuation, authorize_conversation
 
 
 _journal = ContextVar("action_credential_journal", default=None)
@@ -61,6 +61,8 @@
         journal.deferred.add(_call_id.get())
         raise journal.pending
     try:
+        if (_continuation.get() or {}).get("resume"):
+            authorize_conversation(journal.user_id, journal.conversation_id)
         await next(context)
     except Exception as error:
         pending = find_credential_stop(error)

application/single_app/functions_action_credential_conversations.py

@@ -0,0 +1,108 @@
+# functions_action_credential_conversations.py
+"""Credential-specific conversation routing; storage and policies come from bootstrap."""
+
+from azure.cosmos.exceptions import CosmosResourceNotFoundError
+
+from action_credential_contracts import ActionCredentialError
+from cosmos_document_utils import normalize_cosmos_document
+
+
+class CredentialConversations:
+    def __init__(self, *, personal_container, shared_container, message_container,
+                 authorize_personal, authorize_shared, authorize_orchestration):
+        self.personal = personal_container
+        self.shared = shared_container
+        self.messages = message_container
+        self.authorize_personal = authorize_personal
+        self.authorize_shared = authorize_shared
+        self.authorize_orchestration = authorize_orchestration
+
+    @staticmethod
+    def _read(container, item_id, partition):
+        try:
+            return normalize_cosmos_document(container.read_item(item=item_id, partition_key=partition))
+        except CosmosResourceNotFoundError:
+            return None
+
+    @staticmethod
+    def _linked(source, shared):
+        return bool(
+            source
+            and source.get("conversation_kind") == "collaboration_source"
+            and source.get("collaboration_conversation_id") == shared["id"]
+            and source.get("user_id") == shared.get("created_by_user_id")
+            and source.get("id") in {shared["id"], shared.get("source_conversation_id")}
+        )
+
+    def resolve(self, user_id, conversation_id):
+        if not isinstance(conversation_id, str) or not conversation_id.strip() or not user_id:
+            raise LookupError("Conversation not found.")
+        source = self._read(self.personal, conversation_id, conversation_id)
+        shared_id = (source or {}).get("collaboration_conversation_id")
+        if source is not None and not shared_id:
+            self.authorize_personal(user_id, conversation_id)
+            return source
+        shared_id = shared_id or conversation_id
+        shared = self._read(self.shared, shared_id, shared_id)
+        if shared is None or shared.get("status") != "active":
+            raise LookupError("Conversation not found.")
+        self.authorize_shared(user_id, shared)
+        if source is not None and not self._linked(source, shared):
+            raise LookupError("Conversation not found.")
+        execution_ids = []
+        source_id = shared.get("source_conversation_id")
+        if source_id:
+            backing = source if source and source["id"] == source_id else self._read(self.personal, source_id, source_id)
+            if not self._linked(backing, shared):
+                raise LookupError("Conversation not found.")
+            execution_ids.append(source_id)
+        if shared.get("created_by_user_id") == user_id:
+            backing = source if source and source["id"] == shared_id else self._read(self.personal, shared_id, shared_id)
+            if backing is not None:
+                if not self._linked(backing, shared):
+                    raise LookupError("Conversation not found.")
+                self.authorize_orchestration(shared, user_id)
+                execution_ids.append(shared_id)
+        return {
+            "id": conversation_id,
+            "credential_visible_conversation_id": shared_id,
+            "credential_conversation_kind": "collaborative",
+            "credential_execution_ids": tuple(dict.fromkeys(execution_ids)),
+        }
+
+    def chat_origin(self, user_id, conversation_id, message_id):
+        context = self.resolve(user_id, conversation_id)
+        visible_id = context.get("credential_visible_conversation_id")
+        if not visible_id or not message_id or conversation_id == visible_id:
+            raise ActionCredentialError("credential_continuation_unavailable")
+        message = self._read(self.messages, message_id, visible_id)
+        metadata = (message or {}).get("metadata") or {}
+        if (
+            not message or message.get("conversation_id") != visible_id
+            or (metadata.get("sender") or {}).get("user_id") != user_id
+            or metadata.get("source_conversation_id") != conversation_id
+            or message.get("message_kind") != "ai_request"
+            or message.get("is_deleted") or metadata.get("is_deleted")
+        ):
+            raise ActionCredentialError("credential_continuation_unavailable")
+        return {"visible_conversation_id": visible_id, "collaboration_message_id": message_id}
+
+    def authorize_request(self, user_id, item):
+        context = self.resolve(user_id, item["conversation_id"])
+        if context.get("credential_conversation_kind") != "collaborative":
+            return context
+        if item["conversation_id"] not in context["credential_execution_ids"]:
+            raise LookupError("Conversation not found.")
+        descriptor = item.get("continuation") or {}
+        if descriptor.get("mode") == "chat":
+            origin = self.chat_origin(user_id, item["conversation_id"], descriptor.get("collaboration_message_id"))
+            if origin["visible_conversation_id"] != descriptor.get("visible_conversation_id"):
+                raise ActionCredentialError("credential_continuation_unavailable")
+        else:
+            if descriptor.get("mode") != "orchestration" or item["conversation_id"] != context["credential_visible_conversation_id"]:
+                raise ActionCredentialError("credential_continuation_unavailable")
+            shared = self._read(self.shared, context["credential_visible_conversation_id"], context["credential_visible_conversation_id"])
+            if shared is None:
+                raise LookupError("Conversation not found.")
+            self.authorize_orchestration(shared, user_id)
+        return context

application/single_app/functions_action_credential_runtime.py

@@ -2,6 +2,7 @@
 """Web/scheduler composition of credential services after runtime initialization."""
 
 from action_credential_contracts import ActionCredentialError
+from functions_action_credential_conversations import CredentialConversations
 import functions_action_credentials as credentials
 import functions_action_credential_continuation as continuation
 
@@ -16,6 +17,69 @@ def configure_action_credential_runtime():
     import functions_workspace_identities as identities
     import route_backend_chats
     from content_screening import access
+    from flask import has_request_context, request
+
+    def authorize_shared(user_id, shared):
+        # Collaboration initializes storage; only initialized runtime callbacks load it.
+        from functions_collaboration import assert_user_can_participate_in_collaboration_conversation
+
+        return assert_user_can_participate_in_collaboration_conversation(user_id, shared, read_only=True)
+
+    def authorize_orchestration(shared, user_id):
+        from functions_orchestration_collaboration import authorize_shared_orchestration
+
+        return authorize_shared_orchestration(shared, user_id, read_only=True)
+
+    from config import cosmos_collaboration_conversations_container, cosmos_collaboration_messages_container
+
+    conversations = CredentialConversations(
+        personal_container=route_backend_chats.cosmos_conversations_container,
+        shared_container=cosmos_collaboration_conversations_container,
+        message_container=cosmos_collaboration_messages_container,
+        authorize_personal=route_backend_chats._authorize_personal_conversation_access,
+        authorize_shared=authorize_shared, authorize_orchestration=authorize_orchestration,
+    )
+
+    def capture_chat_origin(user_id, conversation_id):
+        data = request.get_json(silent=True) if has_request_context() else None
+        message_id = data.get("m365_collaboration_message_id") if isinstance(data, dict) else None
+        return conversations.chat_origin(user_id, conversation_id, message_id) if message_id else {}
+
+    def publish_chat_result(user_id, item, saved):
+        context = conversations.authorize_request(user_id, item)
+        if context.get("credential_conversation_kind") != "collaborative":
+            return saved
+        from functions_collaboration import (
+            create_collaboration_message_notifications, get_collaboration_conversation,
+            mirror_source_message_to_collaboration, publish_collaboration_event,
+            serialize_collaboration_conversation, serialize_collaboration_message,
+        )
+
+        visible_id = context["credential_visible_conversation_id"]
+        shared = get_collaboration_conversation(visible_id)
+        authorize_shared(user_id, shared)
+        question = conversations._read(
+            conversations.messages, item["continuation"]["collaboration_message_id"], visible_id,
+        )
+        sender = (question["metadata"] or {})["sender"]
+        message, shared, _created = mirror_source_message_to_collaboration(
+            shared, saved, sender, reply_to_message_id=question["id"],
+            extra_metadata={"source_thought_user_id": user_id},
+            message_id="shared_" + saved["id"],
+        )
+        if not message:
+            raise ActionCredentialError("credential_publication_unavailable", 503)
+        authorize_shared(user_id, shared)
+        create_collaboration_message_notifications(shared, message)
+        serialized = serialize_collaboration_message(message)
+        publish_collaboration_event(visible_id, {
+            "conversation_id": visible_id, "event_type": "collaboration.message.created",
+            "payload": {
+                "conversation": serialize_collaboration_conversation(shared, current_user_id=user_id),
+                "message": serialized,
+            },
+        })
+        return serialized
 
     def resolve_manifest(user_id, action_ref):
         scope_type, scope_id, action_id = action_ref.split(":", 2)
@@ -38,14 +102,17 @@ def update_identity(*args, **kwargs):
 
     credentials.configure_action_credential_services(credentials.CredentialServices(
         current_user_id=functions_authentication.get_current_user_id_or_none,
-        authorize_conversation=route_backend_chats._authorize_personal_conversation_access,
+        authorize_conversation=conversations.resolve,
         resolve_manifest=resolve_manifest,
         read_identity=identities.get_workspace_identity,
         list_identities=identities.list_workspace_identities,
         identity_supports_usage=identities.identity_supports_usage,
         read_identity_auth=identities.get_workspace_identity_auth,
         create_identity=identities.create_workspace_identity,
         update_identity=update_identity,
+        authorize_request_context=conversations.authorize_request,
+        capture_chat_origin=capture_chat_origin,
+        publish_chat_result=publish_chat_result,
     ))
     continuation.configure_action_credential_continuation_services(continuation.CredentialContinuationServices(
         collect_sources=access.current_request_source_provenance,

application/single_app/functions_action_credentials.py

@@ -34,6 +34,9 @@ class CredentialServices:
     read_identity_auth: Callable
     create_identity: Callable
     update_identity: Callable
+    authorize_request_context: Callable | None = None
+    capture_chat_origin: Callable | None = None
+    publish_chat_result: Callable | None = None
 
 
 def configure_action_credential_services(services):
@@ -74,6 +77,20 @@ def authorize_conversation(user_id, conversation_id):
     return credential_services().authorize_conversation(user_id, conversation_id)
 
 
+def capture_chat_origin(user_id, conversation_id):
+    callback = _services.capture_chat_origin if _services is not None else None
+    return callback(user_id, conversation_id) if callback else {}
+
+
+def request_conversation(item):
+    callback = credential_services().authorize_request_context
+    return callback(item["user_id"], item) if callback else authorize_conversation(item["user_id"], item["conversation_id"])
+
+
+def pending_execution_ids(context, conversation_id):
+    return context.get("credential_execution_ids", (conversation_id,))
+
+
 def authorized_manifest(user_id, action_ref):
     return credential_services().resolve_manifest(user_id, action_ref)
 
@@ -106,18 +123,22 @@ def safe_identities(user_id, requirements):
 
 def safe_request(item, manifest):
     user_id = item["user_id"]
+    context = request_conversation(item)
     requirements = [
         {"slot": slot, "auth_type": auth_type,
          "label": "Proxy sign-in" if slot == "proxy" else "YAMCS sign-in"}
         for slot, auth_type in item["requirements"].items()
     ]
     result = {
-        "id": item["id"], "etag": item["_etag"], "conversation_id": item["conversation_id"],
+        "id": item["id"], "etag": item["_etag"],
+        "conversation_id": context.get("credential_visible_conversation_id", item["conversation_id"]),
         "action_name": manifest.get("displayName") or manifest.get("name") or "YAMCS",
         "destination_label": destination_label(manifest), "reason": item["reason"],
         "requirements": requirements, "identities": safe_identities(user_id, requirements),
         "status": "pending" if item["status"] == "saving" else item["status"],
     }
+    if context.get("credential_conversation_kind") == "collaborative":
+        result["conversation_kind"] = "collaborative"
     if item["status"] == "saving":
         result["etag"] = item["submission_etag"]
     for key in ("run_id", "turn_id"):
@@ -128,7 +149,7 @@ def safe_request(item, manifest):
 
 def authorize_request(user_id, request_id):
     item = state.read(user_id, request_id, record_type="credential_request")
-    authorize_conversation(user_id, item["conversation_id"])
+    request_conversation(item)
     manifest = authorized_manifest(user_id, item["action_ref"])
     if credential_fingerprint(manifest) != item["fingerprint"]:
         raise ActionCredentialError("action_destination_changed")

application/single_app/functions_collaboration.py

@@ -336,7 +336,9 @@ def ensure_collaboration_user_state_for_participant(
     return state_doc
 
 
-def _bootstrap_collaboration_user_state_from_participant(conversation_doc, participant_record, invited_by_user_id=None):
+def _bootstrap_collaboration_user_state_from_participant(
+    conversation_doc, participant_record, invited_by_user_id=None, *, persist=True,
+):
     if not isinstance(participant_record, dict):
         return None
 
@@ -350,7 +352,8 @@ def _bootstrap_collaboration_user_state_from_participant(conversation_doc, parti
         or (conversation_doc or {}).get('created_at')
         or utc_now_iso()
     )
-    state_doc = ensure_collaboration_user_state_for_participant(
+    builder = ensure_collaboration_user_state_for_participant if persist else build_collaboration_user_state
+    state_doc = builder(
         conversation_doc,
         participant_record,
         role=participant_record.get('role') or MEMBERSHIP_ROLE_MEMBER,
@@ -363,7 +366,8 @@ def _bootstrap_collaboration_user_state_from_participant(conversation_doc, parti
         state_doc['responded_at'] = participant_record.get('responded_at')
     if participant_record.get('removed_at'):
         state_doc['removed_at'] = participant_record.get('removed_at')
-    cosmos_collaboration_user_state_container.upsert_item(state_doc)
+    if persist:
+        cosmos_collaboration_user_state_container.upsert_item(state_doc)
     return state_doc
 
 
@@ -1646,7 +1650,7 @@ def list_group_collaboration_conversations_for_user(user_id):
     return conversations
 
 
-def assert_user_can_view_collaboration_conversation(user_id, conversation_doc, allow_pending=False):
+def assert_user_can_view_collaboration_conversation(user_id, conversation_doc, allow_pending=False, *, read_only=False):
     if not is_collaboration_conversation(conversation_doc):
         raise LookupError('Collaboration conversation not found')
 
@@ -1702,6 +1706,7 @@ def assert_user_can_view_collaboration_conversation(user_id, conversation_doc, a
                 conversation_doc,
                 participant,
                 invited_by_user_id=conversation_doc.get('created_by_user_id'),
+                persist=not read_only,
             )
 
         if user_state is None:
@@ -1733,11 +1738,12 @@ def assert_user_can_view_collaboration_conversation(user_id, conversation_doc, a
     raise PermissionError('Unsupported collaboration conversation type')
 
 
-def assert_user_can_participate_in_collaboration_conversation(user_id, conversation_doc):
+def assert_user_can_participate_in_collaboration_conversation(user_id, conversation_doc, *, read_only=False):
     access_context = assert_user_can_view_collaboration_conversation(
         user_id,
         conversation_doc,
         allow_pending=False,
+        read_only=read_only,
     )
 
     if is_group_collaboration_conversation(conversation_doc):

application/single_app/functions_orchestration_collaboration.py

@@ -71,14 +71,16 @@ def shared_conversation(conversation_id):
         return None
 
 
-def authorize_shared_orchestration(collaboration, user_id):
+def authorize_shared_orchestration(collaboration, user_id, *, read_only=False):
     """Require a current participant who started the conversation; return the backing fields."""
     from functions_collaboration import (
         assert_user_can_participate_in_collaboration_conversation,
         is_group_collaboration_conversation,
     )
 
-    assert_user_can_participate_in_collaboration_conversation(user_id, collaboration)
+    assert_user_can_participate_in_collaboration_conversation(
+        user_id, collaboration, **({"read_only": True} if read_only else {}),
+    )
     if str(collaboration.get('created_by_user_id') or '').strip() != str(user_id or '').strip():
         raise SharedOrchestrationError()
     return {