Fix HNS result cleanup and workflow deletion recovery

application/single_app/config.py

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

application/single_app/functions_mixed_source_orchestration.py

@@ -140,6 +140,7 @@
     "canceled_source_count",
     "citation_count",
     "completed_source_count",
+    "completion_tokens",
     "duplicate_evidence_count",
     "engine_call_count",
     "evidence_omitted_count",
@@ -165,6 +166,7 @@
     "unexpected_evidence_count",
     "unsupported_source_count",
     "unresolved_source_count",
+    "xml_schema_source_count",
 })
 MIXED_SOURCE_TELEMETRY_DIMENSIONS = frozenset({
     "cancellation_phase",

application/single_app/functions_saved_analysis.py

@@ -19,7 +19,7 @@
 from functions_workflow_context import WorkflowContextBudgetError, calculate_workflow_context_budget
 from functions_workflow_identity import normalize_workflow_iteration_path
 from functions_workflow_result_masking import message_uses_workflow_result, withhold_workflow_result_message
-from functions_workflow_result_store import WorkflowResultStorageUnavailableError, _quota_bytes
+from functions_workflow_result_store import AnalysisWorkUnitConflictError, WorkflowResultStorageUnavailableError, _quota_bytes
 from functions_workflow_runtime_store import WorkflowRuntimeConflict
 from functions_workflow_results import (
     ANALYSIS_SOURCE_ACCESS_VERSION,
@@ -1927,7 +1927,10 @@
                         bound_workflow or workflow, producer["run_id"], producer.get("task_id"), reference,
                         reader_user_id=user_id, **selectors,
                     )
-                except (PermissionError, LookupError, ValueError, AzureError, WorkflowResultStorageUnavailableError, WorkflowRuntimeConflict) as exc:
+                except (
+                    PermissionError, LookupError, ValueError, AzureError, AnalysisWorkUnitConflictError,
+                    WorkflowResultStorageUnavailableError, WorkflowRuntimeConflict,
+                ) as exc:
                     log_event(
                         "[DOCUMENT_ANALYSIS] Workflow analysis preview withheld.",
                         extra={"run_id": run_record.get("id"), "task_id": task_id, "error_type": type(exc).__name__},

application/single_app/functions_workflow_result_store.py

@@ -249,6 +249,21 @@
     return prefix
 
 
+def _is_hierarchical_directory(properties):
+    """Return True for an ADLS Gen2 (hierarchical namespace) directory entry.
+
+    Flat Blob listings on HNS-enabled accounts also yield zero-byte directory
+    objects for every path segment; they carry no result payload or result-specific metadata.
+    """
+    metadata = getattr(properties, "metadata", None) or {}
+    metadata_is_folder = isinstance(metadata, Mapping) and any(
+        isinstance(key, str) and key.lower() == "hdi_isfolder" and str(value).strip().lower() == "true"
+        for key, value in metadata.items()
+    )
+    resource_type = str(getattr(properties, "resource_type", "") or "").lower()
+    return metadata_is_folder or resource_type == "directory"
+
+
 def _blob_name(identity, reference):
     if identity["scope_type"] == "chat":
         return f"{_chat_blob_prefix(identity)}{reference['sha256']}.json"
@@ -1505,10 +1520,26 @@
             prefix = _orchestration_blob_prefix(scope)
         else:
             prefix = _run_blob_prefix(scope)
+        if is_orchestration and "run_id" not in scope:
+            directory_pattern = r"[0-9a-f]{64}(?:/[0-9a-f]{64})?"
+        elif is_chat and "message_id" in scope:
+            directory_pattern = None
+        else:
+            directory_pattern = r"[0-9a-f]{64}"
+        directories = []
         container = self.blob_client.get_container_client(self.blob_container_name)
         for properties in container.list_blobs(name_starts_with=prefix, include=["metadata"]):
             name = properties.name
             suffix = name[len(prefix):] if isinstance(name, str) and name.startswith(prefix) else ""
+            if _is_hierarchical_directory(properties):
+                directory_suffix = suffix[:-1] if suffix.endswith("/") else suffix
+                if (
+                    directory_pattern is None or getattr(properties, "size", None) != 0
+                    or not re.fullmatch(directory_pattern, directory_suffix)
+                ):
+                    raise WorkflowResultIntegrityError("Unexpected object in the private result prefix.")
+                directories.append(name)
+                continue
             if is_orchestration and "run_id" not in scope:
                 if not re.fullmatch(r"[0-9a-f]{64}/[0-9a-f]{64}/[0-9a-f]{64}\.json", suffix):
                     raise WorkflowResultIntegrityError("Unexpected object in the private result prefix.")
@@ -1566,6 +1597,17 @@
                 )
             except ResourceNotFoundError:
                 continue
+        # Directories are removed only after their files, deepest first. Each one is
+        # re-read so the conditional delete uses its current ETag and type.
+        for name in sorted(set(directories), key=lambda value: (-value.rstrip("/").count("/"), value)):
+            blob = self.blob_client.get_blob_client(container=self.blob_container_name, blob=name)
+            try:
+                current = blob.get_blob_properties()
+                if not _is_hierarchical_directory(current) or getattr(current, "size", None) != 0:
+                    raise WorkflowResultIntegrityError("Unexpected object in the private result prefix.")
+                blob.delete_blob(etag=current.etag, match_condition=MatchConditions.IfNotModified)
+            except ResourceNotFoundError:
+                continue
 
     def delete_run_results(self, workflow, run_id):
         """Stream all private records and run-prefix blobs, including interrupted writes.

application/single_app/route_backend_workflows.py

@@ -91,7 +91,12 @@
     validate_url_access_request,
 )
 from functions_workflow_runner import _workflow_task_run_item_id, create_workflow_run_id, run_group_workflow, run_personal_workflow
-from functions_workflow_result_store import WorkflowResultStorageUnavailableError, read_workflow_task_result_page
+from functions_workflow_result_store import (
+    AnalysisWorkUnitConflictError,
+    WorkflowResultIntegrityError,
+    WorkflowResultStorageUnavailableError,
+    read_workflow_task_result_page,
+)
 from functions_workflow_definitions import (
     WorkflowDefinitionConflict,
     WorkflowDefinitionError,
@@ -134,9 +139,51 @@
 WORKFLOW_INSTRUCTION_FIELD_LIMIT = 6000
 
 
+WORKFLOW_DELETE_PENDING_MESSAGE = (
+    'This workflow is being deleted, so its runs can no longer be cancelled. '
+    'Delete the workflow again to finish cleanup.'
+)
+WORKFLOW_RUNTIME_MISSING_MESSAGE = (
+    'This workflow run can no longer be cancelled because its runtime state was removed. '
+    'If you were deleting the workflow, delete it again to finish cleanup.'
+)
+# Cleanup failures that leave the workflow retryable; anything else still surfaces as a 500.
+WORKFLOW_DELETE_FAILURE_ERRORS = (
+    AnalysisWorkUnitConflictError,
+    AzureError,
+    RuntimeUnavailable,
+    WorkflowResultIntegrityError,
+    WorkflowResultStorageUnavailableError,
+    WorkflowRuntimeConflict,
+)
+WORKFLOW_DELETE_FAILED_MESSAGE = (
+    'The workflow could not be fully deleted. Try deleting it again; '
+    'if the problem continues, contact an administrator.'
+)
+
+
 class WorkflowCancellationConflictError(RuntimeError):
     """Raised when a workflow run cannot transition into cancellation."""
 
+    def __init__(self, public_message='Workflow run cannot be cancelled in its current state.'):
+        self.public_message = public_message
+        super().__init__(public_message)
+
+
+def _workflow_delete_failure_response(exc, workflow_id, *, workspace_type, group_id=None):
+    log_event(
+        '[WORKFLOW_ROUTES] Workflow delete failed',
+        extra={
+            'workflow_id': workflow_id,
+            'workspace_type': workspace_type,
+            'group_id': group_id or '',
+            'error_type': type(exc).__name__,
+        },
+        level=logging.ERROR,
+        exceptionTraceback=True,
+    )
+    return jsonify({'error': WORKFLOW_DELETE_FAILED_MESSAGE}), 500
+
 
 def _normalize_identifier(value):
     return str(value or '').strip()
@@ -241,6 +288,8 @@ def _request_workflow_run_cancellation(
     workflow_id = _normalize_identifier(workflow.get('id'))
     active_run_id = _normalize_identifier(workflow.get('active_run_id'))
     target_run_id = _normalize_identifier(run_id) or active_run_id
+    if workflow.get('deleting') is True or _normalize_identifier(workflow.get('status')).lower() == 'deleting':
+        raise WorkflowCancellationConflictError(WORKFLOW_DELETE_PENDING_MESSAGE)
     if not target_run_id:
         raise WorkflowCancellationConflictError('No active workflow run is available to cancel.')
     if active_run_id and active_run_id != target_run_id:
@@ -252,7 +301,13 @@ def _request_workflow_run_cancellation(
     if not run_record and target_run_id != active_run_id:
         raise LookupError('Workflow run not found.')
     if run_record and run_record.get('durable_execution') is True:
-        cancelled = cancel_durable_workflow_run(workflow, target_run_id, actor_user_id=requested_by)
+        try:
+            cancelled = cancel_durable_workflow_run(workflow, target_run_id, actor_user_id=requested_by)
+        except WorkflowRuntimeConflict as exc:
+            # A deleted runtime control usually means an earlier delete already fenced this run.
+            if exc.code == 'not_found':
+                raise WorkflowCancellationConflictError(WORKFLOW_RUNTIME_MISSING_MESSAGE) from exc
+            raise WorkflowCancellationConflictError(exc.public_message) from exc
         safe_run = {key: cancelled['run'].get(key) for key in ('id', 'workflow_id', 'status', 'durable_execution', 'runtime')}
         return _workflow_definition_response(cancelled['workflow']), safe_run
 
@@ -1840,7 +1895,10 @@ def delete_user_workflow(workflow_id):
         if not workflow:
             return jsonify({'error': 'Workflow not found.'}), 404
 
-        deleted = delete_personal_workflow(user_id, workflow_id)
+        try:
+            deleted = delete_personal_workflow(user_id, workflow_id)
+        except WORKFLOW_DELETE_FAILURE_ERRORS as exc:
+            return _workflow_delete_failure_response(exc, workflow_id, workspace_type='personal')
         if not deleted:
             return jsonify({'error': 'Workflow not found.'}), 404
 
@@ -1901,7 +1959,7 @@ def cancel_active_user_workflow_run(workflow_id):
             return jsonify({'error': 'Workflow run not found.'}), 404
         except WorkflowCancellationConflictError as exc:
             logging.exception("Workflow cancellation conflict while cancelling active user workflow run.")
-            return jsonify({'error': 'Workflow run cannot be cancelled in its current state.'}), 409
+            return jsonify({'error': exc.public_message}), 409
 
         return jsonify({'success': True, 'workflow': updated_workflow, 'run': run_record}), 202
 
@@ -1935,14 +1993,14 @@ def cancel_user_workflow_run(workflow_id, run_id):
                 user_id,
             )
             return jsonify({'error': 'Workflow run not found.'}), 404
-        except WorkflowCancellationConflictError:
+        except WorkflowCancellationConflictError as exc:
             logging.exception(
                 "WorkflowCancellationConflictError while cancelling personal workflow run. workflow_id=%s run_id=%s user_id=%s",
                 sanitize_log_message(workflow_id),
                 sanitize_log_message(run_id),
                 user_id,
             )
-            return jsonify({'error': 'Workflow run cancellation conflict.'}), 409
+            return jsonify({'error': exc.public_message}), 409
 
         return jsonify({'success': True, 'workflow': updated_workflow, 'run': run_record}), 202
 
@@ -2284,7 +2342,10 @@ def delete_group_workflow_route(workflow_id):
         if not workflow:
             return jsonify({'error': 'Workflow not found.'}), 404
 
-        deleted = delete_group_workflow(group_id, workflow_id)
+        try:
+            deleted = delete_group_workflow(group_id, workflow_id)
+        except WORKFLOW_DELETE_FAILURE_ERRORS as exc:
+            return _workflow_delete_failure_response(exc, workflow_id, workspace_type='group', group_id=group_id)
         if not deleted:
             return jsonify({'error': 'Workflow not found.'}), 404
 
@@ -2379,7 +2440,7 @@ def cancel_active_group_workflow_run(workflow_id):
             logging.exception('Group workflow run cancellation failed: run not found.', exc_info=exc)
             return jsonify({'error': 'Workflow run not found.'}), 404
         except WorkflowCancellationConflictError as exc:
-            return jsonify({'error': str(exc)}), 409
+            return jsonify({'error': exc.public_message}), 409
 
         return jsonify({'success': True, 'workflow': updated_workflow, 'run': run_record}), 202
 
@@ -2422,7 +2483,7 @@ def cancel_group_workflow_run(workflow_id, run_id):
             return jsonify({'error': 'Run not found.'}), 404
         except WorkflowCancellationConflictError as exc:
             logging.exception('Group workflow run cancellation conflict.')
-            return jsonify({'error': 'Unable to cancel run in its current state.'}), 409
+            return jsonify({'error': exc.public_message}), 409
 
         return jsonify({'success': True, 'workflow': updated_workflow, 'run': run_record}), 202
 

docs/explanation/features/WORKFLOW_DURABLE_EXECUTION.md

@@ -2,7 +2,7 @@
 
 Implemented in version: **0.261.111**
 
-Updated in version: **0.261.120**.
+Updated in version: **0.261.305**.
 
 Application version tracking: `application/single_app/config.py`.
 
@@ -182,6 +182,12 @@ records are excluded from ordinary task-item lists.
 Private lifecycle tombstones survive run cleanup. Deletion removes result
 payloads without allowing a late worker to recreate the deleted run.
 
+Since **0.261.305**, private cleanup also handles explicitly marked, empty HNS
+directories, deleting them after their result files. If cleanup fails after a
+workflow is marked for deletion, retry **Delete** to finish it; **Cancel** cannot
+restore a tombstoned runtime. Deletion-fenced saved outputs are withheld from
+history. See [Workflow deletion recovery](../fixes/WORKFLOW_DELETE_HNS_DIRECTORY_FIX.md).
+
 ## Runtime API
 
 Use the corresponding personal or group prefix: