Add Microsoft 365 actions, approvals, and delegated workflow execution

application/single_app/agent_logging_chat_completion.py

@@ -1,22 +1,34 @@
-
+# agent_logging_chat_completion.py
+from contextlib import aclosing
+from copy import deepcopy
 import json
 import logging
 from pydantic import Field
 from semantic_kernel.agents import ChatCompletionAgent
+from semantic_kernel.const import DEFAULT_SERVICE_NAME
+from semantic_kernel.functions import KernelArguments
+from functions_m365_agent_continuation import (
+    m365_agent_continuation,
+    m365_agent_stream_continuation,
+)
 from functions_appinsights import log_event
+from functions_model_capabilities import ModelTokenBudget, ModelTokenBudgetError
+from functions_model_budget_runtime import prepare_model_execution_settings
 import datetime
 import re
 
 
 class LoggingChatCompletionAgent(ChatCompletionAgent):
     display_name: str | None = Field(default=None)
     default_agent: bool = Field(default=False)
+    is_global: bool = Field(default=False)
     tool_invocations: list = Field(default_factory=list)
     deployment_name: str | None = Field(default=None)
     azure_endpoint: str | None = Field(default=None)
     api_version: str | None = Field(default=None)
+    model_token_budget: ModelTokenBudget | None = Field(default=None, exclude=True)
 
-    def __init__(self, *args, display_name=None, default_agent=False, deployment_name=None, azure_endpoint=None, api_version=None, **kwargs):
+    def __init__(self, *args, display_name=None, default_agent=False, deployment_name=None, azure_endpoint=None, api_version=None, model_token_budget=None, **kwargs):
         # Remove these from kwargs so the base class doesn't see them
         kwargs.pop('display_name', None)
         kwargs.pop('default_agent', None)
@@ -29,8 +41,58 @@
         self.deployment_name = deployment_name
         self.azure_endpoint = azure_endpoint
         self.api_version = api_version
+        self.model_token_budget = model_token_budget
         # tool_invocations is now properly declared as a Pydantic field
 
+    def _merge_arguments(self, override_args):
+        base = self.arguments if self.arguments is not None else KernelArguments()
+        values = dict(base)
+        execution_settings = deepcopy(base.execution_settings or {})
+        if override_args is not None:
+            values.update(override_args)
+            overrides = deepcopy(override_args.execution_settings or {})
+            if self.model_token_budget is not None and self.service is not None:
+                service_id = self.service.service_id
+                if set(overrides) - {service_id, DEFAULT_SERVICE_NAME}:
+                    raise ModelTokenBudgetError(
+                        "model_context_invalid", "Select an agent bound to the requested model service."
+                    )
+                default = overrides.pop(DEFAULT_SERVICE_NAME, None)
+                if default is not None and service_id not in overrides:
+                    default.service_id = service_id
+                    overrides[service_id] = default
+            execution_settings.update(overrides)
+        return KernelArguments(settings=execution_settings, **values)
+
+    async def _get_chat_completion_service_and_settings(self, kernel, arguments):
+        service, settings = await super()._get_chat_completion_service_and_settings(kernel, arguments)
+        if self.model_token_budget is None:
+            return service, settings
+        if (
+            self.service is not None and service is not self.service
+            or self.deployment_name is not None and service.ai_model_id != self.deployment_name
+        ):
+            raise ModelTokenBudgetError(
+                "model_context_invalid", "The selected service does not match this agent's model budget."
+            )
+        override_model = getattr(settings, "ai_model_id", None)
+        if override_model and override_model != service.ai_model_id:
+            raise ModelTokenBudgetError(
+                "model_context_invalid", "Select a model with matching budget metadata instead of overriding its request identifier."
+            )
+        try:
+            settings, _ = prepare_model_execution_settings(
+                settings, self.model_token_budget,
+                tools_enabled=bool(kernel.get_full_list_of_function_metadata()),
+            )
+        except ModelTokenBudgetError as error:
+            log_event(
+                "[SK_LOADER] Agent model budget configuration is invalid.",
+                extra={"agent_id": self.id, "code": error.code}, level=logging.ERROR,
+            )
+            raise
+        return service, settings
+
     def log_tool_execution(self, tool_name, arguments=None, result=None):
         """Manual method to log tool executions. Can be called by plugins."""
         tool_citation = {
@@ -131,6 +193,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 +269,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
@@ -1302,6 +1320,34 @@ def list_semantic_kernel_plugins():
 
 # ------------------- 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():
@@ -295,6 +315,8 @@ def check_retention_policy_once():
     from functions_retention_policy import execute_retention_policy
 
     try:
+        # Retention can run before the workflow scheduler initializes delivery.
+        configure_m365_pending_delivery_runtime(_get_workflow_runner_app().test_request_context)
         results = execute_retention_policy(manual_execution=False)
         if results.get('success'):
             print(
@@ -527,10 +549,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 +644,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 +681,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 +731,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 +768,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: