Fix conversation context and orchestration execution reliability

application/single_app/background_tasks.py

@@ -61,9 +61,9 @@
     workflow_result_is_waiting,
     workflow_result_runtime_status,
 )
-from functions_m365_approvals import get_m365_approval_service
+from functions_m365_approvals import M365PolicyError, 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_continuations import M365ContinuationNotStarted, 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 (
@@ -572,7 +572,13 @@
         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"])
+        try:
+            workflow = load_current_workflow(job["workflow_ref"])
+        except M365PolicyError as error:
+            if error.code != "m365_workflow_missing":
+                raise
+            # Only this initial read proves no workflow/Graph effects were entered.
+            raise M365ContinuationNotStarted() from error
         settings = get_settings()
         group_id = workflow.get("group_id")
         if group_id:

application/single_app/config.py

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

application/single_app/functions_appinsights.py

@@ -102,6 +102,7 @@
 LOGGER_EXTERNAL_EVENT_MESSAGE = "[SIMPLE_CHAT_EXTERNAL_EVENT]"
 MAX_EXTERNAL_EVENT_STRING_LENGTH = 256
 LOGGER_SAFE_TEXT_MAX_LENGTH = 1024
+LOGGER_APPLICATION_VERSION_MAX_LENGTH = 64
 # Diagnostic keys whose sanitized text may be retained; everything else stays a length.
 LOGGER_SAFE_TEXT_KEYS = frozenset({
     "attempt",
@@ -131,6 +132,7 @@
 LOGGER_WORKFLOW_CODE_KEYS = frozenset({
     "validationcode", "validationrule", "responsefailure", "executioncode", "outputcode", "durablestatus",
     "failurecode", "authorityreason", "capabilityid", "outputformat", "producercapabilityid",
+    "executionpath", "outcome",
 })
 
 
@@ -146,6 +148,46 @@
     }
 
 
+def _is_application_version(value: Any) -> bool:
+    """Accept bounded three-part decimal versions without regex backtracking."""
+    if not isinstance(value, str) or len(value) > LOGGER_APPLICATION_VERSION_MAX_LENGTH:
+        return False
+    parts = value.split(".", 2)
+    return len(parts) == 3 and all(part.isdecimal() for part in parts)
+
+
+def log_chat_execution_outcome(
+    *, execution_path, status, outcome, conversation_id, turn_id=None, run_id=None,
+    failure=None, application_version=None, attempt_index=None, message_saved=None,
+):
+    """Record business outcomes independently of a successful SSE/HTTP transport."""
+    if execution_path not in {"direct", "orchestration"}:
+        raise ValueError("Invalid chat execution path.")
+    outcomes = {"completed", "partial", "failed", "waiting", "cancelled"}
+    if outcome not in outcomes or status not in outcomes:
+        raise ValueError("Invalid chat execution outcome.")
+    failure = failure if isinstance(failure, dict) else {}
+    fields = {
+        **workflow_log_context(
+            conversation_id=conversation_id, turn_id=turn_id, run_id=run_id,
+            step_id=failure.get("step_id"),
+        ),
+        "execution_path": execution_path, "status": status, "outcome": outcome,
+        "failure_code": failure.get("code"), "capability_id": failure.get("capability_id"),
+    }
+    if _is_application_version(application_version):
+        fields["application_version"] = application_version
+    if type(attempt_index) is int and attempt_index > 0:
+        fields["attempt_index"] = attempt_index
+    if type(message_saved) is bool:
+        fields["message_saved"] = message_saved
+    log_event(
+        "[CHAT_EXECUTION] Execution outcome recorded.",
+        extra=fields,
+        level=logging.WARNING if outcome in {"failed", "partial"} else logging.INFO,
+    )
+
+
 def _format_message(message: Any, message_args: Optional[Tuple[Any, ...]] = None) -> str:
     """Support legacy printf-style rendering while preserving plain strings."""
     message_text = str(message)
@@ -281,6 +323,9 @@
                 elif diagnostic_key in LOGGER_WORKFLOW_CODE_KEYS:
                     if re.fullmatch(r"[a-z][a-z0-9_]{0,79}", value):
                         logger_extra[normalized_key] = value
+                elif diagnostic_key == "applicationversion":
+                    if _is_application_version(value):
+                        logger_extra[normalized_key] = value
                 elif _is_safe_log_text_key(key):
                     logger_extra[normalized_key] = (
                         sanitize_log_message(value)[:LOGGER_SAFE_TEXT_MAX_LENGTH]

application/single_app/functions_m365_continuations.py

@@ -2,15 +2,51 @@
 """Conditional delivery of persisted Microsoft 365 workflow continuations."""
 
 from datetime import datetime, timedelta, timezone
+import hashlib
 import logging
 
 from azure.core import MatchConditions
 from azure.cosmos.exceptions import CosmosHttpResponseError, CosmosResourceNotFoundError
 
+from functions_m365_context import M365PolicyError
+
 
 DECIDED_STATES = frozenset({"approved", "denied", "expired", "invalidated", "revoked"})
 
 
+class M365ContinuationNotStarted(M365PolicyError):
+    """The executor verified deletion before entering any workflow execution."""
+
+    def __init__(self):
+        super().__init__("m365_workflow_missing", "The workflow no longer exists.")
+
+
+def _log_continuation(log_event, candidate, *, code, stage, level, error=None):
+    properties = {
+        "request_id_hash": hashlib.sha256(str(candidate.get("id") or "").encode("utf-8")).hexdigest(),
+        "failure_code": code,
+        "stage": stage,
+    }
+    if error is not None:
+        properties["error_type"] = type(error).__name__
+    log_event(
+        "[MS_GRAPH_PLUGIN] Workflow continuation outcome.",
+        level=level, extra=properties,
+    )
+
+
+def _fail_deleted_workflow(jobs, candidate, now):
+    stopped = {
+        **candidate, "status": "failed", "failure_code": "workflow_deleted",
+        "completed_at": now.isoformat(),
+    }
+    stopped.pop("resume_lease_expires_at", None)
+    jobs.replace_item(
+        candidate["id"], body=stopped,
+        etag=candidate["_etag"], match_condition=MatchConditions.IfNotModified,
+    )
+
+
 def resume_pending_workflows(
     jobs, approvals, *, execute, can_resume, log_event, connection_ready=None, limit=25, clock=None,
 ):
@@ -27,84 +63,136 @@
     )
     outcomes = []
     for candidate in candidates:
-        if candidate.get("status") == "resuming":
-            lease = candidate.get("resume_lease_expires_at")
-            if lease and datetime.fromisoformat(lease) > now:
-                continue
-            # Uncertain execution is not an instruction to repeat external mutations.
-            candidate["status"] = "recovery_required"
-            candidate["recovery_reason"] = "A continuation worker stopped before acknowledging its result."
-            jobs.replace_item(
-                candidate["id"], body=candidate,
-                etag=candidate["_etag"], match_condition=MatchConditions.IfNotModified,
-            )
-            log_event(
-                "[MS_GRAPH_PLUGIN] Workflow continuation requires recovery.",
-                level=logging.WARNING,
-                extra={"request_id": candidate["id"]},
-            )
-            continue
-        if candidate.get("status") == "awaiting_sign_in":
-            if connection_ready is None or not connection_ready(candidate):
+        stage = "preflight"
+        try:
+            if candidate.get("status") == "resuming":
+                lease = candidate.get("resume_lease_expires_at")
+                if lease and datetime.fromisoformat(lease) > now:
+                    continue
+                # An expired claim is uncertain even if its workflow was later deleted.
+                recovering = {
+                    **candidate, "status": "recovery_required",
+                    "recovery_reason": "A continuation worker stopped before acknowledging its result.",
+                }
+                stage = "recover_expired_claim"
+                jobs.replace_item(
+                    candidate["id"], body=recovering,
+                    etag=candidate["_etag"], match_condition=MatchConditions.IfNotModified,
+                )
+                _log_continuation(
+                    log_event, candidate, code="continuation_recovery_required",
+                    stage=stage, level=logging.WARNING,
+                )
                 continue
-            approval = None
-        else:
+
             try:
-                approval = approvals.get_approval(
-                    candidate["approval_id"], candidate["user_id"],
-                )
-            except (CosmosResourceNotFoundError, LookupError):
-                log_event(
-                    "[MS_GRAPH_PLUGIN] Continuation approval is unavailable.",
-                    level=logging.WARNING, extra={"request_id": candidate["id"]},
+                # Check the workflow even when approval/sign-in is still pending.
+                eligible = can_resume(candidate, None)
+            except M365PolicyError as error:
+                if error.code != "m365_workflow_missing":
+                    raise
+                stage = "terminalize_deleted_workflow"
+                _fail_deleted_workflow(jobs, candidate, now)
+                _log_continuation(
+                    log_event, candidate, code="workflow_deleted",
+                    stage=stage, level=logging.INFO,
                 )
                 continue
-            if approval.get("status") not in DECIDED_STATES:
+            if not eligible:
                 continue
-        if not can_resume(candidate, approval):
-            continue
-        claimed = {
-            **candidate,
-            "status": "resuming",
-            "resume_lease_expires_at": (now + timedelta(minutes=15)).isoformat(),
-        }
-        try:
+            if candidate.get("status") == "awaiting_sign_in":
+                if connection_ready is None or not connection_ready(candidate):
+                    continue
+            else:
+                try:
+                    approval = approvals.get_approval(
+                        candidate["approval_id"], candidate["user_id"],
+                    )
+                except (CosmosResourceNotFoundError, LookupError) as error:
+                    _log_continuation(
+                        log_event, candidate, code="continuation_approval_unavailable",
+                        stage=stage, level=logging.WARNING, error=error,
+                    )
+                    continue
+                if approval.get("status") not in DECIDED_STATES:
+                    continue
+                try:
+                    eligible = can_resume(candidate, approval)
+                except M365PolicyError as error:
+                    if error.code != "m365_workflow_missing":
+                        raise
+                    stage = "terminalize_deleted_workflow"
+                    _fail_deleted_workflow(jobs, candidate, now)
+                    _log_continuation(
+                        log_event, candidate, code="workflow_deleted",
+                        stage=stage, level=logging.INFO,
+                    )
+                    continue
+                if not eligible:
+                    continue
+            claimed = {
+                **candidate,
+                "status": "resuming",
+                "resume_lease_expires_at": (now + timedelta(minutes=15)).isoformat(),
+            }
+            stage = "claim"
             claimed = jobs.replace_item(
                 candidate["id"], body=claimed,
                 etag=candidate["_etag"], match_condition=MatchConditions.IfNotModified,
             )
-        except CosmosHttpResponseError as error:
-            if error.status_code == 412:
+            stage = "execute"
+            try:
+                result = execute(claimed)
+            except M365ContinuationNotStarted:
+                stage = "terminalize_deleted_workflow"
+                _fail_deleted_workflow(jobs, claimed, now)
+                _log_continuation(
+                    log_event, claimed, code="workflow_deleted",
+                    stage=stage, level=logging.INFO,
+                )
                 continue
-            raise
-        try:
-            result = execute(claimed)
-        except Exception:
-            log_event(
-                "[MS_GRAPH_PLUGIN] Workflow continuation failed.",
-                level=logging.ERROR,
-                extra={"request_id": claimed["id"]},
-                exceptionTraceback=True,
-            )
-            current = jobs.read_item(claimed["id"], partition_key=claimed["user_id"])
-            if current.get("_etag") == claimed["_etag"]:
-                current["status"] = "recovery_required"
+            except Exception as error:
+                _log_continuation(
+                    log_event, claimed, code="continuation_execution_failed",
+                    stage=stage, level=logging.ERROR, error=error,
+                )
+                stage = "recover_failed_execution"
+                recovering = {
+                    **claimed, "status": "recovery_required",
+                    "recovery_reason": "The continuation failed after execution may have started.",
+                }
                 jobs.replace_item(
-                    current["id"], body=current,
-                    etag=current["_etag"], match_condition=MatchConditions.IfNotModified,
+                    claimed["id"], body=recovering,
+                    etag=claimed["_etag"], match_condition=MatchConditions.IfNotModified,
                 )
-            raise
-        current = jobs.read_item(claimed["id"], partition_key=claimed["user_id"])
-        if current.get("_etag") == claimed["_etag"]:
+                continue
+            stage = "acknowledge"
+            current = dict(claimed)
             if result.get("durable_execution") is True and result.get("pending") is True:
                 current["status"] = "running"
                 current.pop("resume_lease_expires_at", None)
             else:
                 current["status"] = "completed" if result.get("success") else "failed"
                 current["completed_at"] = now.isoformat()
-            jobs.replace_item(
-                current["id"], body=current,
-                etag=current["_etag"], match_condition=MatchConditions.IfNotModified,
+            try:
+                jobs.replace_item(
+                    current["id"], body=current,
+                    etag=claimed["_etag"], match_condition=MatchConditions.IfNotModified,
+                )
+            except CosmosHttpResponseError as error:
+                if error.status_code != 412:
+                    raise
+                # The executor may have saved a new approval or its own checkpoint.
+                _log_continuation(
+                    log_event, candidate, code="continuation_conflict",
+                    stage=stage, level=logging.INFO, error=error,
+                )
+            outcomes.append({"request_id": claimed["id"], "result": result})
+        except Exception as error:
+            conflict = isinstance(error, CosmosHttpResponseError) and error.status_code in {404, 412}
+            _log_continuation(
+                log_event, candidate,
+                code="continuation_conflict" if conflict else "continuation_processing_failed",
+                stage=stage, level=logging.INFO if conflict else logging.ERROR, error=error,
             )
-        outcomes.append({"request_id": claimed["id"], "result": result})
     return outcomes