Add Microsoft 365 actions, approvals, and delegated workflow execution

application/single_app/agent_logging_chat_completion.py

@@ -1,8 +1,13 @@
-
+# agent_logging_chat_completion.py
+from contextlib import aclosing
 import json
 import logging
 from pydantic import Field
 from semantic_kernel.agents import ChatCompletionAgent
+from functions_m365_agent_continuation import (
+    m365_agent_continuation,
+    m365_agent_stream_continuation,
+)
 from functions_appinsights import log_event
 import datetime
 import re
@@ -131,6 +136,7 @@
         """
         return []  # Plugin invocation logger handles this now
 
+    @m365_agent_continuation
     async def invoke(self, *args, **kwargs):
         # Clear previous tool invocations
         self.tool_invocations = []
@@ -206,6 +212,12 @@
                 }
             )
 
+    @m365_agent_stream_continuation
+    async def invoke_stream(self, *args, **kwargs):
+        async with aclosing(super().invoke_stream(*args, **kwargs)) as stream:
+            async for response in stream:
+                yield response
+
     def _capture_tool_invocations_simplified(self, args, response):
         """
         SIMPLIFIED: Basic fallback citation capture.

application/single_app/app.py

@@ -97,6 +97,24 @@
 from route_backend_collaboration import register_route_backend_collaboration
 from route_backend_data_management import register_route_backend_data_management
 from route_backend_msgraph_pending_actions import register_route_backend_msgraph_pending_actions
+from route_backend_m365 import configure_m365_routes, register_route_backend_m365
+from functions_m365_approvals import configure_m365_approvals
+from functions_m365_connections import configure_m365_connection_authorization
+from functions_m365_execution import configure_m365_execution, validate_m365_workflow_context
+from functions_m365_file_runtime import configure_m365_file_runtime
+from functions_m365_request_resume import queue_approved_chat
+from functions_m365_runtime import (
+    authorize_m365_conversation_audit,
+    complete_m365_request,
+    configure_m365_history_runtime,
+    configure_m365_pending_delivery_runtime,
+    resolve_m365_action_config,
+    resolve_m365_action_selection,
+    resolve_m365_audit_conversation_id,
+    resolve_m365_workflow_binding,
+    validate_m365_approval_decision,
+    validate_m365_workflow_execution,
+)
 from route_inbound_mcp import register_route_inbound_mcp
 from route_enhanced_citations import register_enhanced_citations_routes
 from plugin_validation_endpoint import plugin_validation_admin_bp, plugin_validation_bp
@@ -1296,6 +1314,34 @@
 
 # ------------------- API MS Graph Pending Action Routes -
 register_route_blueprint('backend_msgraph_pending_actions', register_route_backend_msgraph_pending_actions, user_required_blueprint)
+configure_m365_approvals(decision_validator=validate_m365_approval_decision)
+configure_m365_execution(
+    workflow_validator=validate_m365_workflow_execution,
+    action_config_resolver=resolve_m365_action_config,
+    workflow_binding_resolver=resolve_m365_workflow_binding,
+    action_selection_resolver=resolve_m365_action_selection,
+)
+configure_m365_connection_authorization(validate_m365_workflow_context)
+configure_m365_routes(
+    conversation_authorizer=authorize_m365_conversation_audit,
+    decision_callback=queue_approved_chat,
+    audit_conversation_resolver=resolve_m365_audit_conversation_id,
+)
+configure_m365_history_runtime()
+configure_m365_file_runtime()
+configure_m365_pending_delivery_runtime(app.test_request_context)
+register_route_blueprint('backend_m365', register_route_backend_m365, user_required_blueprint)
+
+
+@app.after_request
+def finalize_m365_json_request(response):
+    if response.is_json:
+        payload = response.get_json()
+        success = response.status_code < 400 and isinstance(payload, dict) and not (
+            payload.get("error") or payload.get("pending") or payload.get("success") is False
+        )
+        complete_m365_request(success=success)
+    return response
 
 # ------------------- API Documents Routes ---------------
 register_route_blueprint('backend_documents', register_route_backend_documents, user_required_blueprint)

application/single_app/background_tasks.py

@@ -12,7 +12,7 @@
 
 from azure.core import MatchConditions
 
-from config import cosmos_settings_container, exceptions
+from config import cosmos_m365_execution_runs_container, cosmos_settings_container
 from functions_appinsights import log_event
 from functions_control_center import (
     calculate_next_control_center_auto_refresh_run,
@@ -52,7 +52,27 @@
     update_group_workflow_runtime_fields,
 )
 from functions_settings import get_settings, is_group_workflows_enabled_for_group, update_settings
-from functions_workflow_runner import create_workflow_run_id, run_group_workflow, run_personal_workflow
+from functions_m365_workflow_binding import (
+    M365_ACTIVE_STATES,
+    workflow_result_is_waiting,
+    workflow_result_runtime_status,
+)
+from functions_m365_approvals import get_m365_approval_service
+from functions_m365_connections import configure_m365_connection_authorization, get_m365_connection_service
+from functions_m365_continuations import resume_pending_workflows
+from functions_m365_execution import configure_m365_execution, validate_m365_workflow_context
+from functions_m365_file_runtime import configure_m365_file_runtime
+from functions_m365_runtime import (
+    configure_m365_pending_delivery_runtime,
+    load_current_workflow,
+    resolve_m365_action_config,
+    resolve_m365_action_selection,
+    resolve_m365_workflow_binding,
+    validate_m365_approval_decision,
+    validate_m365_workflow_execution,
+)
+from functions_workflow_runner import _get_workflow_runner_app, create_workflow_run_id, run_group_workflow, run_personal_workflow
+from functions_m365_pending_delivery import dispatch_due_m365_deliveries
 
 
 def _get_lock_holder_id():
@@ -527,10 +547,83 @@ def run_cosmos_throughput_autoscale_loop():
         time.sleep(sleep_seconds)
 
 
+def check_m365_workflow_continuations_once():
+    approval_service = get_m365_approval_service()
+    approval_service.decision_validator = validate_m365_approval_decision
+    configure_m365_execution(
+        workflow_validator=validate_m365_workflow_execution,
+        action_config_resolver=resolve_m365_action_config,
+        workflow_binding_resolver=resolve_m365_workflow_binding,
+        action_selection_resolver=resolve_m365_action_selection,
+    )
+    configure_m365_connection_authorization(validate_m365_workflow_context)
+    configure_m365_file_runtime()
+    configure_m365_pending_delivery_runtime(_get_workflow_runner_app().test_request_context)
+    dispatch_due_m365_deliveries()
+
+    def can_resume(job, approval):
+        workflow = load_current_workflow(job["workflow_ref"])
+        return (
+            workflow.get("active_run_id") == job.get("run_id")
+            and workflow.get("status") in M365_ACTIVE_STATES
+            and workflow.get("m365_run_as_user_id") == job.get("user_id")
+            and (approval is None or approval.get("subject_user_id") == job.get("user_id"))
+        )
+
+    def connection_ready(job):
+        from config import TENANT_ID
+        connection = get_m365_connection_service().current_connection(job["user_id"], TENANT_ID)
+        if not connection or connection.get("status") != "connected":
+            return False
+        granted = {scope.rsplit("/", 1)[-1].lower() for scope in connection.get("authorized_scopes") or []}
+        return all(scope.rsplit("/", 1)[-1].lower() in granted for scope in job.get("required_scopes") or [])
+
+    def execute(job):
+        workflow = load_current_workflow(job["workflow_ref"])
+        settings = get_settings()
+        group_id = workflow.get("group_id")
+        if group_id:
+            if not is_group_workflows_enabled_for_group(settings, group_id):
+                raise PermissionError("Group workflows are no longer enabled for this group.")
+            lock_name = f"group_workflow_run_{group_id}_{workflow['id']}"
+        else:
+            if not settings.get("allow_user_workflows", False):
+                raise PermissionError("Personal workflows are no longer enabled.")
+            lock_name = f"workflow_run_{workflow['id']}"
+        lock = acquire_distributed_task_lock(lock_name, lease_seconds=900)
+        if not lock:
+            raise RuntimeError("The workflow is already executing.")
+        try:
+            runner = run_group_workflow if group_id else run_personal_workflow
+            result = runner(
+                workflow, trigger_source="m365_approval",
+                actor_user_id=job.get("actor_user_id"), run_id=job["run_id"],
+            )
+            updates = dict(result.get("workflow_updates") or {})
+            updates["status"] = workflow_result_runtime_status(result)
+            if not workflow_result_is_waiting(result):
+                updates["next_run_at"] = compute_next_run_at(
+                    workflow, from_time=datetime.now(timezone.utc),
+                )
+            if group_id:
+                update_group_workflow_runtime_fields(group_id, workflow["id"], updates)
+            else:
+                update_personal_workflow_runtime_fields(workflow["user_id"], workflow["id"], updates)
+            return result
+        finally:
+            release_distributed_task_lock(lock)
+
+    return resume_pending_workflows(
+        cosmos_m365_execution_runs_container, approval_service,
+        execute=execute, can_resume=can_resume, log_event=log_event,
+        connection_ready=connection_ready,
+    )
+
+
 def check_due_workflows_once():
     """Execute scheduled personal and group workflows that are due."""
     settings = get_settings()
-    results = []
+    results = check_m365_workflow_continuations_once()
 
     if settings.get('allow_user_workflows', False):
         due_workflows = get_due_personal_workflows(limit=20)
@@ -549,6 +642,8 @@ def check_due_workflows_once():
                 refreshed_workflow = get_personal_workflow(user_id, workflow_id)
                 if not refreshed_workflow:
                     continue
+                if refreshed_workflow.get('status') in M365_ACTIVE_STATES:
+                    continue
                 trigger_type = str(refreshed_workflow.get('trigger_type') or '').strip().lower()
                 if trigger_type not in {'interval', 'file_sync'} or not refreshed_workflow.get('is_enabled', False):
                     continue
@@ -584,8 +679,9 @@ def check_due_workflows_once():
                     run_id=active_run_id,
                 )
                 update_fields = dict(result.get('workflow_updates') or {})
-                update_fields['status'] = 'idle'
-                update_fields['next_run_at'] = compute_next_run_at(refreshed_workflow, from_time=datetime.now(timezone.utc))
+                update_fields['status'] = workflow_result_runtime_status(result)
+                if not workflow_result_is_waiting(result):
+                    update_fields['next_run_at'] = compute_next_run_at(refreshed_workflow, from_time=datetime.now(timezone.utc))
                 update_personal_workflow_runtime_fields(user_id, workflow_id, update_fields)
                 results.append({'scope': 'personal', 'workflow_id': workflow_id, 'success': bool(result.get('success'))})
             except Exception as exc:
@@ -633,6 +729,8 @@ def check_due_workflows_once():
                 refreshed_workflow = get_group_workflow(group_id, workflow_id)
                 if not refreshed_workflow:
                     continue
+                if refreshed_workflow.get('status') in M365_ACTIVE_STATES:
+                    continue
                 trigger_type = str(refreshed_workflow.get('trigger_type') or '').strip().lower()
                 if trigger_type not in {'interval', 'file_sync'} or not refreshed_workflow.get('is_enabled', False):
                     continue
@@ -668,8 +766,9 @@ def check_due_workflows_once():
                     run_id=active_run_id,
                 )
                 update_fields = dict(result.get('workflow_updates') or {})
-                update_fields['status'] = 'idle'
-                update_fields['next_run_at'] = compute_next_run_at(refreshed_workflow, from_time=datetime.now(timezone.utc))
+                update_fields['status'] = workflow_result_runtime_status(result)
+                if not workflow_result_is_waiting(result):
+                    update_fields['next_run_at'] = compute_next_run_at(refreshed_workflow, from_time=datetime.now(timezone.utc))
                 update_group_workflow_runtime_fields(group_id, workflow_id, update_fields)
                 results.append({'scope': 'group', 'group_id': group_id, 'workflow_id': workflow_id, 'success': bool(result.get('success'))})
             except Exception as exc:

application/single_app/config.py

@@ -35,7 +35,7 @@
 import pandas
 from functions_latest_features_nav import is_development_env_enabled
 from functions_appinsights import log_event
-from functions_azure_endpoint_validation import validate_azure_blob_endpoint
+from functions_azure_endpoint_validation import validate_configured_chat_blob_endpoint
 
 from functions_environment import load_simplechat_dotenv
 from flask import (
@@ -98,7 +98,7 @@
 EXECUTOR_TYPE = 'thread'
 EXECUTOR_MAX_WORKERS = 30
 SESSION_TYPE = 'filesystem'
-VERSION = "0.261.028"
+VERSION = "0.261.032"
 IS_DEVELOPMENT = is_development_env_enabled()
 
 # Opt-out for deployments where App Service Easy Auth is active but the platform
@@ -537,7 +537,11 @@ def build_enhanced_citations_blob_service_client(settings):
         blob_endpoint = str(settings.get("office_docs_storage_account_blob_endpoint") or "").strip()
         if not blob_endpoint:
             raise ValueError("Enhanced Citations blob endpoint is required for managed identity authentication.")
-        safe_blob_endpoint = validate_azure_blob_endpoint(blob_endpoint)
+        # Endpoint ownership is deployment configuration, not a file/action argument.
+        safe_blob_endpoint = validate_configured_chat_blob_endpoint(
+            blob_endpoint,
+            CUSTOM_BLOB_STORAGE_URL_VALUE if AZURE_ENVIRONMENT == "custom" else "",
+        )
         # codeql[py/full-ssrf]
         return BlobServiceClient(account_url=safe_blob_endpoint, credential=DefaultAzureCredential())
 
@@ -1024,6 +1028,20 @@ def _create_container_if_not_exists_with_conflict_recovery(*args, **kwargs):
     default_ttl=-1
 )
 
+cosmos_m365_connections_container_name = "m365_connections"
+cosmos_m365_connections_container = cosmos_database.create_container_if_not_exists(
+    id=cosmos_m365_connections_container_name,
+    partition_key=PartitionKey(path="/user_id"),
+    default_ttl=-1,
+)
+
+cosmos_m365_execution_runs_container_name = "m365_execution_runs"
+cosmos_m365_execution_runs_container = cosmos_database.create_container_if_not_exists(
+    id=cosmos_m365_execution_runs_container_name,
+    partition_key=PartitionKey(path="/user_id"),
+    default_ttl=-1,
+)
+
 cosmos_thoughts_container_name = "thoughts"
 cosmos_thoughts_container = cosmos_database.create_container_if_not_exists(
     id=cosmos_thoughts_container_name,