Clearer saved-result error messages and repaired workflow test harnesses

application/single_app/config.py

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

application/single_app/functions_analysis_access.py

@@ -1,5 +1,9 @@
 # functions_analysis_access.py
-"""Current source access checks for saved Analyze results."""
+"""Source checks for Analyze inputs, and the error raised when an analysis can't be used.
+
+Only reading a document as an input checks it. A saved or generated result takes its access
+from the conversation, workflow or run that holds it, and keeps its sources as provenance.
+"""
 
 from collections import defaultdict
 import math
@@ -18,13 +22,123 @@
 
 ANALYSIS_SOURCE_ACCESS_VERSION = "analysis-source-access-v1"
 
+# Each code is classified by what its raise sites check, not by its name. A raise site whose
+# code is shared with another family passes ``family=`` instead. Every code raised in
+# application/single_app must be classified: functional_tests/
+# test_analysis_unavailable_message_families.py scans the raise sites and fails otherwise.
+ANALYSIS_UNAVAILABLE_SOURCE = "source"
+ANALYSIS_UNAVAILABLE_CONTAINER = "container"
+ANALYSIS_UNAVAILABLE_SAVED_RESULT = "saved_result"
+
+# An input read found a source document it can't read now, or one that changed while it was read.
+ANALYSIS_SOURCE_UNAVAILABLE_CODES = frozenset({
+    "analysis_source_manifest_changed",
+    "analysis_source_snapshot_changed",
+    "analysis_source_unavailable",
+    # Loop document re-authorization passes its WorkflowLoopInputError code through.
+    "workflow_loop_input_unavailable",
+    "workflow_loop_scope_forbidden",
+    "workflow_loop_source_changed",
+})
+
+# The conversation, message, workflow or run that holds the result is gone or isn't the caller's.
+ANALYSIS_CONTAINER_UNAVAILABLE_CODES = frozenset({
+    "analysis_conversation_changed",
+    "analysis_conversation_deleted",
+    "analysis_conversation_unavailable",
+    "analysis_conversation_version_missing",
+    "analysis_message_masked",
+    "analysis_message_unavailable",
+    "analysis_retry_changed",
+    "analysis_retry_unavailable",
+    "analysis_workflow_unavailable",
+    "generated_artifact_source_unavailable",
+    "native_compute_owner_unavailable",
+})
+
+# The stored result, or an earlier result, manifest, receipt, or loop or repeat state it
+# depends on, is missing, changed or invalid.
+ANALYSIS_SAVED_RESULT_UNAVAILABLE_CODES = frozenset({
+    "analysis_artifact_not_valid",
+    "analysis_artifact_unbound",
+    "analysis_lineage_invalid",
+    "analysis_native_identity_invalid",
+    "analysis_native_source_mismatch",
+    "analysis_producer_unavailable",
+    "analysis_result_unavailable",
+    "analysis_source_manifest_conflict",
+    "analysis_source_manifest_invalid",
+    "analysis_source_manifest_missing",
+    "generated_artifact_content_changed",
+    "generated_artifact_source_conflict",
+    "generated_artifact_source_unbound",
+    "generated_artifact_source_uncommitted",
+    "native_compute_identity_invalid",
+    "native_compute_producer_stopped",
+    "native_compute_producer_unavailable",
+    "native_compute_result_changed",
+    "native_compute_result_unavailable",
+    "workflow_body_export_invalid",
+    "workflow_body_export_scope_invalid",
+    "workflow_iteration_outcome_invalid",
+    "workflow_iteration_receipt_invalid",
+    "workflow_loop_identity_invalid",
+    "workflow_loop_item_changed",
+    "workflow_loop_item_invalid",
+    "workflow_loop_manifest_invalid",
+    "workflow_loop_manifest_unavailable",
+    "workflow_loop_source_receipt_invalid",
+    "workflow_repeat_admission_invalid",
+    "workflow_repeat_grant_invalid",
+    "workflow_repeat_head_invalid",
+    "workflow_repeat_identity_invalid",
+    "workflow_repeat_iteration_invalid",
+    "workflow_repeat_source_invalid",
+    "workflow_repeat_source_uncommitted",
+    "workflow_repeat_state_changed",
+    "workflow_repeat_state_invalid",
+    "workflow_repeat_state_uncommitted",
+    "workflow_repeat_transition_invalid",
+})
+
+ANALYSIS_UNAVAILABLE_MESSAGES = {
+    ANALYSIS_UNAVAILABLE_SOURCE: "A source document for this analysis is no longer available or has changed.",
+    ANALYSIS_UNAVAILABLE_CONTAINER: (
+        "The conversation, workflow or run that holds this result no longer exists or isn't available to you."
+    ),
+    ANALYSIS_UNAVAILABLE_SAVED_RESULT: (
+        "This saved result can't be used because something it depends on is missing or has changed."
+    ),
+}
+
+
+def analysis_unavailable_family(code):
+    """Return the message family for ``code``.
+
+    An unclassified code gets the saved-result wording, which blames neither a source
+    document nor the caller's access.
+    """
+    if isinstance(code, str):
+        if code in ANALYSIS_SOURCE_UNAVAILABLE_CODES:
+            return ANALYSIS_UNAVAILABLE_SOURCE
+        if code in ANALYSIS_CONTAINER_UNAVAILABLE_CODES:
+            return ANALYSIS_UNAVAILABLE_CONTAINER
+    return ANALYSIS_UNAVAILABLE_SAVED_RESULT
+
 
 class AnalysisResultUnavailable(PermissionError):
-    """A saved analysis cannot be read under the caller's current access."""
+    """An analysis input or saved result can't be used.
 
-    def __init__(self, code="analysis_source_unavailable"):
+    ``code`` names the check that failed. ``family`` picks the public wording: a source
+    document, the conversation, workflow or run that holds the result, or the saved result
+    itself. Pass ``family`` only when the code is shared with a different family.
+    """
+
+    def __init__(self, code="analysis_source_unavailable", *, family=None):
         self.code = code
-        super().__init__("This analysis is unavailable because its source access could not be confirmed.")
+        self.family = family if family in ANALYSIS_UNAVAILABLE_MESSAGES else analysis_unavailable_family(code)
+        self.public_message = ANALYSIS_UNAVAILABLE_MESSAGES[self.family]
+        super().__init__(self.public_message)
 
 
 def resolve_analysis_source_manifest(document_ids, user_id, *, resolver=None, **context):
@@ -50,7 +164,8 @@
         ):
             if strict_source_authority_enabled():
                 raise_source_authority_error(SourceAuthorityUnverifiedError())
-            raise AnalysisResultUnavailable("analysis_source_manifest_invalid")
+            # The current source lookup itself failed: an input read, not a saved manifest.
+            raise AnalysisResultUnavailable("analysis_source_manifest_invalid", family=ANALYSIS_UNAVAILABLE_SOURCE)
         if strict_source_authority_enabled():
             for source in resolved:
                 if source.get("authorization_status") not in {"authorized", "unresolved"}:
@@ -138,7 +253,7 @@
 def authorize_analysis_sources(user_id, sources, *, require_snapshot=False, resolver=None):
     """Authorize every contributor; a saved permission is never sufficient."""
     if not isinstance(user_id, str) or not user_id.strip():
-        raise AnalysisResultUnavailable()
+        raise AnalysisResultUnavailable("analysis_source_unavailable")
     snapshots = analysis_source_snapshot(sources)
     if not snapshots:
         raise AnalysisResultUnavailable("analysis_source_manifest_missing")
@@ -163,7 +278,7 @@
             fresh,
         )
         if comparison["authorization_failure_count"]:
-            raise AnalysisResultUnavailable()
+            raise AnalysisResultUnavailable("analysis_source_unavailable")
         fresh_by_id = {source["document_id"]: source for source in fresh}
         for source in original:
             current = fresh_by_id[source["document_id"]]

application/single_app/functions_document_analysis_checkpoints.py

@@ -94,7 +94,7 @@ def validate_sources(self, sources=None):
             require_snapshot=True,
         )
         if allowed is False:
-            raise AnalysisResultUnavailable()
+            raise AnalysisResultUnavailable('analysis_source_unavailable')
 
     def initialize(self, request, sources):
         assigned = analysis_source_snapshot(sources)

application/single_app/functions_document_analysis_results.py

@@ -8,7 +8,11 @@
 import unicodedata
 from copy import deepcopy
 
-from functions_analysis_access import AnalysisResultUnavailable, analysis_source_snapshot
+from functions_analysis_access import (
+    ANALYSIS_UNAVAILABLE_SOURCE,
+    AnalysisResultUnavailable,
+    analysis_source_snapshot,
+)
 from functions_analysis_deliverables import apply_analysis_calculations
 from functions_tabular_transformations import normalize_tabular_transformation_spec
 
@@ -198,7 +202,8 @@ def index_analysis_source_manifest(source_manifest, document_ids):
         if source['document_id'].strip() not in assigned_ids:
             continue
         if 'authorization_status' in source and source['authorization_status'] != 'authorized':
-            raise AnalysisResultUnavailable('analysis_source_manifest_invalid')
+            # An assigned document the current source lookup did not authorize can't be read now.
+            raise AnalysisResultUnavailable('analysis_source_manifest_invalid', family=ANALYSIS_UNAVAILABLE_SOURCE)
         snapshot = analysis_source_snapshot([source])[0]
         previous = snapshots.get(snapshot['document_id'])
         if previous is not None and previous != snapshot:

application/single_app/functions_generated_artifact_sources.py

@@ -28,7 +28,10 @@
 _HISTORY_SOURCE_ERRORS = (PermissionError, LookupError, ValueError, AzureError,
                           WorkflowResultStorageUnavailableError, RuntimeUnavailable, ScreeningError,
                           OutputConflictError, OutputStorageError)
-_UNAVAILABLE_HISTORY = "Saved workflow output is unavailable because current access could not be confirmed."
+_UNAVAILABLE_HISTORY = (
+    "Saved workflow output is unavailable because it or the workflow run it came from is missing, "
+    "has changed or can't be read right now."
+)
 _ORCHESTRATION_UNAVAILABLE_HISTORY = "Generated file is unavailable because current access or publication could not be confirmed."
 _RETAINED_SOURCE_KINDS = frozenset({"workflow_saved_output", ORCHESTRATION_ARTIFACT_KIND})
 _ORCHESTRATION_HISTORY_DENIAL_CODES = frozenset({

application/single_app/functions_saved_analysis.py

@@ -39,7 +39,10 @@
 MAX_ANALYSIS_PAGE_RECORDS = 100
 MAX_ANALYSIS_PAGE_BYTES = 256 * 1024
 MAX_ANALYSIS_DIAGNOSTIC_PAGE_BYTES = 65536
-UNAVAILABLE_ANALYSIS_MESSAGE = "This saved analysis is unavailable because its access could not be confirmed."
+UNAVAILABLE_ANALYSIS_MESSAGE = (
+    "This saved analysis is unavailable because it or something it depends on is missing, "
+    "has changed or can't be read right now."
+)
 UNVERIFIED_WORKFLOW_OUTPUT_MESSAGE = "This saved task output is unavailable because it could not be read or verified."
 
 
@@ -164,7 +167,7 @@ def _load_authorized_message(user_id, conversation_id, message_id):
             raise AnalysisResultUnavailable("analysis_conversation_deleted")
         message = get_collaboration_message(message_id)
     if message.get("conversation_id") != conversation_id or message.get("id") != message_id:
-        raise AnalysisResultUnavailable()
+        raise AnalysisResultUnavailable("analysis_message_unavailable")
     return message
 
 
@@ -199,7 +202,7 @@ def _load_authorized_workflow(user_id, binding):
         or not item or item.get("workflow_id") != workflow["id"]
         or item.get("run_id") != binding["run_id"] or item.get("task_id") != binding["task_id"]
     ):
-        raise AnalysisResultUnavailable()
+        raise AnalysisResultUnavailable("analysis_workflow_unavailable")
     if binding.get("execution_id") and any(
         item.get(key) != binding.get(key) for key in ("node_id", "execution_id", "iteration_path")
     ):

application/single_app/functions_workflow_artifacts.py

@@ -7,7 +7,7 @@
 from azure.core.exceptions import ResourceNotFoundError
 from azure.cosmos.exceptions import CosmosResourceNotFoundError
 
-from functions_analysis_access import AnalysisResultUnavailable
+from functions_analysis_access import ANALYSIS_UNAVAILABLE_SAVED_RESULT, AnalysisResultUnavailable
 from functions_appinsights import log_event
 from functions_generated_artifact_sources import generated_chat_artifact_address
 from functions_generated_file_exports import GeneratedFileExportRequest, build_generated_file_export
@@ -180,7 +180,10 @@ def load_workflow_artifact_binding(user_id, value, *, require_ready=True, for_pu
             "[SIMPLE_CHAT] Saved-output artifact source unavailable",
             {"exception_type": type(exc).__name__},
         )
-        raise AnalysisResultUnavailable("generated_artifact_source_unavailable") from exc
+        # The saved output's run records could not be loaded; the scope checks raise their own error.
+        raise AnalysisResultUnavailable(
+            "generated_artifact_source_unavailable", family=ANALYSIS_UNAVAILABLE_SAVED_RESULT,
+        ) from exc
 
 
 def _load_workflow_artifact_binding(user_id, value, *, require_ready, for_publication):

application/single_app/functions_workflow_flow_runner.py

@@ -5,7 +5,12 @@
 from copy import deepcopy
 
 from functions_workflow_bindings import WorkflowInputError
-from functions_analysis_access import AnalysisResultUnavailable
+from functions_analysis_access import (
+    ANALYSIS_UNAVAILABLE_CONTAINER,
+    ANALYSIS_UNAVAILABLE_SAVED_RESULT,
+    ANALYSIS_UNAVAILABLE_SOURCE,
+    AnalysisResultUnavailable,
+)
 from functions_workflow_definitions import workflow_output_kind_matches
 from functions_workflow_flow import MISSING, compile_workflow_flow, evaluate_predicate
 from functions_workflow_identity import canonical_digest, workflow_execution_id, workflow_node_identity
@@ -17,6 +22,18 @@
 )
 
 
+# Pause reasons for a loop item that could not be read, by AnalysisResultUnavailable.family.
+LOOP_ITEM_UNAVAILABLE_REASONS = {
+    ANALYSIS_UNAVAILABLE_SOURCE: "The current loop item's original source is no longer available.",
+    ANALYSIS_UNAVAILABLE_CONTAINER: (
+        "The conversation, workflow or run that holds the current loop item's input is no longer available."
+    ),
+    ANALYSIS_UNAVAILABLE_SAVED_RESULT: (
+        "The current loop item's saved input can't be used because something it depends on is missing or has changed."
+    ),
+}
+
+
 class WorkflowFlowRunner:
     def __init__(self, workflow, run_id, execution, task_results, *, actor_user_id, settings=None,
                  load_output=load_workflow_node_input, persist=persist_workflow_task_result):
@@ -457,9 +474,12 @@ def _for_each(self, node, region_id):
                     current = self.execution.store.journal_read("iteration", key)
                     if current["payload"].get("state") not in {"running", "completed", "skipped"}:
                         item_state = "failed"
-                except AnalysisResultUnavailable:
+                except AnalysisResultUnavailable as exc:
                     self.execution.pause_input(
-                        "The current loop item's original source is no longer available.", code="workflow_loop_source_unavailable",
+                        LOOP_ITEM_UNAVAILABLE_REASONS.get(
+                            getattr(exc, "family", None), LOOP_ITEM_UNAVAILABLE_REASONS[ANALYSIS_UNAVAILABLE_SOURCE],
+                        ),
+                        code="workflow_loop_source_unavailable",
                     )
                 except WorkflowInputError:
                     exports, item_state = {}, "failed"

application/single_app/functions_workflow_runner.py

@@ -11066,10 +11066,18 @@ def dispatch_task():
                     level=logging.WARNING,
                     exceptionTraceback=True,
                 )
+                # Keyed by AnalysisResultUnavailable.family (functions_analysis_access); a source is the default.
+                unavailable_error = {
+                    'container': 'A conversation, workflow or run that this task needs is no longer available.',
+                    'saved_result': (
+                        "A saved result this task needs can't be used because something it depends on "
+                        'is missing or has changed.'
+                    ),
+                }.get(getattr(safe_error, 'family', None), 'A required source is no longer available for this workflow run.')
                 task_error = (
                     str(safe_error) if isinstance(safe_error, (WorkflowContextBudgetError, WorkflowResultNotReadyError))
                     else safe_error.public_message if isinstance(safe_error, WorkflowInputError)
-                    else 'A required source is no longer available for this workflow run.' if isinstance(safe_error, AnalysisResultUnavailable)
+                    else unavailable_error if isinstance(safe_error, AnalysisResultUnavailable)
                     else {
                         'runner': (
                             'The task runner is unavailable, disabled, or no longer authorized. '
@@ -11249,10 +11257,20 @@ def save_analysis_section(bound_workflow, bound_run, bound_task, section, **_kwa
             except AnalysisResultUnavailable as exc:
                 # Raised when the task's own analysis could not confirm the sources it read
                 # while it ran, or its result provenance is malformed. Nothing was saved.
-                message = (
+                # The error's family (functions_analysis_access) picks the wording; a source is the default.
+                message = {
+                    'container': (
+                        'The analysis result was not saved because the conversation, workflow or run that '
+                        'holds it is no longer available. Dependent tasks were not run.'
+                    ),
+                    'saved_result': (
+                        'The analysis result was not saved because something it depends on is missing or '
+                        'has changed. Dependent tasks were not run.'
+                    ),
+                }.get(getattr(exc, 'family', None), (
                     'The analysis result was not saved because a document it read changed or became '
                     'unavailable while it ran. Dependent tasks were not run.'
-                )
+                ))
                 _save_workflow_task_run_item(
                     workflow, run_id, task, 'failed', attempt_count=attempt_count,
                     error=message, created_at=created_at, runner_audit=runner_audit,