Stop re-checking saved workflow results against their source documents (screening layer 3a)

application/single_app/config.py

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

application/single_app/content_screening/access.py

@@ -917,6 +917,54 @@ def current_request_source_provenance():
     return deepcopy(list(sources.values()))
 
 
+class _IsolatedSourceFence:
+    """The caller's request-local fence, set aside until ``restore`` puts it back."""
+
+    def __init__(self):
+        current = _SOURCE_AUTHORITY_SCOPE.get()
+        self._token = _SOURCE_AUTHORITY_SCOPE.set({"failure": None}) if current is not None else None
+        flask = import_module("flask")
+        self._previous = None
+        if flask.has_request_context():
+            self._previous = (
+                getattr(flask.g, "content_screening_error", None),
+                dict(getattr(flask.g, "content_screening_sources", {}) or {}),
+            )
+            flask.g.content_screening_error = None
+            flask.g.content_screening_sources = {}
+        self._restored = False
+
+    def restore(self):
+        if self._restored:
+            return
+        self._restored = True
+        flask = import_module("flask")
+        if self._previous is not None and flask.has_request_context():
+            flask.g.content_screening_error, flask.g.content_screening_sources = self._previous
+        if self._token is not None:
+            _SOURCE_AUTHORITY_SCOPE.reset(self._token)
+
+    def __enter__(self):
+        return self
+
+    def __exit__(self, *_exc_info):
+        self.restore()
+        return False
+
+
+def isolate_request_source_fence():
+    """Start an operation with an empty model fence; ``restore()`` returns the caller's.
+
+    The fence keeps the first source failure and every document read during a
+    request, and a guarded model call raises that failure or re-checks those
+    documents first. A workflow run executes all of its tasks in one request
+    context, so without this a task's input reads would block, or be re-checked
+    before, a later task that only reads saved results. Inside the isolated span
+    the operation's own input reads still block its own model calls.
+    """
+    return _IsolatedSourceFence()
+
+
 def guard_chat_service(service, *, source_validator=None):
     """Guard each SK provider call, including automatic tool-call continuations.
 

application/single_app/functions_document_provenance.py

@@ -622,8 +622,11 @@ def _authorized_workflow(user_id, scope_type, scope_id, workflow_id, settings, u
     return _load_scope_workflow(scope_type, scope_id, workflow_id)
 
 
-def _authorized_run(user_id, scope_type, scope_id, workflow, workflow_id, run_id):
-    """Return the origin run only when it belongs to the workflow and this reader may read it."""
+def _authorized_run(scope_type, scope_id, workflow, workflow_id, run_id):
+    """Return the origin run only when it belongs to the workflow this reader may open.
+
+    A run takes its access from its workflow; the documents it used are not re-checked.
+    """
     if not run_id or not isinstance(workflow, Mapping):
         return None
     try:
@@ -633,9 +636,6 @@ def _authorized_run(user_id, scope_type, scope_id, workflow, workflow_id, run_id
             run = import_module("functions_personal_workflows").get_personal_workflow_run(scope_id, run_id)
         if not isinstance(run, Mapping) or run.get("workflow_id") != workflow_id:
             return None
-        import_module("functions_workflow_results").authorize_workflow_run_read(
-            workflow, run_id, reader_user_id=user_id,
-        )
     except _EXPECTED_ACCESS_ERRORS:
         return None
     except Exception as exc:
@@ -660,7 +660,7 @@ def _workflow_origin_summary(user_id, origin, settings, user_roles):
         "href": workflow_origin_href(scope["type"], scope["id"], workflow_id),
     }
     run_id = origin.get("run_id")
-    run = _authorized_run(user_id, scope["type"], scope["id"], workflow, workflow_id, run_id)
+    run = _authorized_run(scope["type"], scope["id"], workflow, workflow_id, run_id)
     if run is not None:
         summary["href"] = workflow_origin_href(scope["type"], scope["id"], workflow_id, run_id)
         started_at = run.get("started_at")
@@ -812,7 +812,7 @@ def _workflow_filter(user_id, workflow_id, run_id, group_id, settings, user_role
             scope = ("group", group_id)
     if scope is None:
         raise DocumentOriginFilterError(ORIGIN_NOT_FOUND_MESSAGE, 404)
-    if run_id is not None and _authorized_run(user_id, scope[0], scope[1], workflow, workflow_id, run_id) is None:
+    if run_id is not None and _authorized_run(scope[0], scope[1], workflow, workflow_id, run_id) is None:
         raise DocumentOriginFilterError(ORIGIN_NOT_FOUND_MESSAGE, 404)
     conditions = [
         "c.origin.kind = @origin_kind",

application/single_app/functions_saved_analysis.py

@@ -40,6 +40,7 @@
 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."
+UNVERIFIED_WORKFLOW_OUTPUT_MESSAGE = "This saved task output is unavailable because it could not be read or verified."
 
 
 def _authorize_conversation(user_id, conversation_id):
@@ -874,6 +875,8 @@ def load_saved_analysis(
         else:
             loader = workflow_loader or _workflow_load
         load = lambda ref: loader(workflow, binding["run_id"], binding["task_id"], ref, **selectors)
+        # A workflow result takes its access from its workflow and run (checked above);
+        # this proves its lineage and returns its sources as provenance only.
         manifest, checked = authorize_workflow_task_result_read(
             workflow, binding["run_id"], binding["task_id"], reference,
             reader_user_id=user_id, load_result=loader, source_resolver=source_resolver,
@@ -1844,7 +1847,13 @@ def sanitize_saved_analysis_messages(
 
 
 def sanitize_workflow_analysis_history(workflow, run_record, user_id, *, items=None, result_reader=None):
-    """Keep task previews and activity from bypassing the workflow result reader."""
+    """Keep task previews and activity from bypassing the workflow result reader.
+
+    Previews take their access from the workflow and run the caller already read.
+    Each stored result is still checked for identity, lineage and integrity; a
+    result that fails that check, or can't be read from storage, is withheld.
+    Its contributing documents are never re-checked.
+    """
     if not isinstance(run_record, Mapping):
         return run_record, items, True
     read = result_reader or authorize_workflow_task_result_read
@@ -1907,17 +1916,17 @@ def redact(task):
         redacted = deepcopy(task)
         for field in ("output_summary", "response_preview", "reply"):
             if field in redacted:
-                redacted[field] = UNAVAILABLE_ANALYSIS_MESSAGE
+                redacted[field] = UNVERIFIED_WORKFLOW_OUTPUT_MESSAGE
         if redacted.get("error"):
-            redacted["error"] = UNAVAILABLE_ANALYSIS_MESSAGE
+            redacted["error"] = UNVERIFIED_WORKFLOW_OUTPUT_MESSAGE
         redacted["analysis_access_available"] = False
         return redacted
 
     run = deepcopy(run_record)
-    run["response_preview"] = UNAVAILABLE_ANALYSIS_MESSAGE
+    run["response_preview"] = UNVERIFIED_WORKFLOW_OUTPUT_MESSAGE
     run["analysis_access_available"] = False
     if run.get("error"):
-        run["error"] = UNAVAILABLE_ANALYSIS_MESSAGE
+        run["error"] = UNVERIFIED_WORKFLOW_OUTPUT_MESSAGE
     for field in ("reply", "analysis_result", "analysis_coverage", "generated_analysis_artifacts", "generated_tabular_outputs"):
         run.pop(field, None)
     run["task_results"] = [redact(task) for task in run.get("task_results") or []]

application/single_app/functions_workflow_execution_history.py

@@ -1,10 +1,14 @@
 # functions_workflow_execution_history.py
-"""Authorized safe projections of the schema-2 execution journal."""
+"""Authorized safe projections of the schema-2 execution journal.
 
-from functions_analysis_access import AnalysisResultUnavailable, authorize_analysis_sources, build_analysis_access
+The caller has already authorized the workflow and run. Projections prove the
+identity and lineage of what they show; the documents an execution read are
+provenance and are not re-checked here.
+"""
+
+from functions_analysis_access import build_analysis_access
 from functions_workflow_flow import compile_workflow_flow
 from functions_workflow_identity import normalize_workflow_iteration_path, workflow_execution_id, workflow_node_identity
-from functions_workflow_inspection import authorize_workflow_flow_sources
 from functions_workflow_journal import public_workflow_journal_entry
 from functions_workflow_limits import WORKFLOW_MAX_EXECUTION_ADMISSIONS
 from functions_workflow_node_results import (
@@ -35,8 +39,8 @@ def authorize_execution_payload(workflow, run_id, payload, *, reader_user_id, au
             authorization.authorize_repeat(head["identity"], reference)
     references = payload.get("reference_sources") or []
     if references:
-        policy = build_analysis_access(references)
-        authorize_analysis_sources(reader_user_id, policy["sources"])
+        # Validate the stored reference receipts' shape; the references themselves are provenance.
+        build_analysis_access(references)
     summary = payload.get("workflow_result") or {}
     if summary.get("result_ref"):
         authorization.authorize_result(summary["producer"], summary["result_ref"])
@@ -69,14 +73,11 @@ def workflow_execution_history(workflow, run_id, *, reader_user_id, kind="execut
         selected_id = workflow_execution_id(workflow, run_id, node_id, path)
         row = store.journal_read("execution", selected_id)
         if row is None:
-            authorize_workflow_flow_sources(workflow, reader_user_id=reader_user_id)
             # No payload exists to supply receipts: prove membership from the
             # frozen items and sealed admissions, never from the requested path alone.
             authorization.walk([("path", {
                 "node_id": node_id, "execution_id": selected_id, "iteration_path": path,
             }, None)])
-            if authorization.access()["source_snapshot_changed"]:
-                raise AnalysisResultUnavailable("analysis_source_snapshot_changed")
             return {"executions": [], "next_cursor": None, "total_count": 0}
         payload = row["payload"]
         expected = {

application/single_app/functions_workflow_flow_runner.py

@@ -182,13 +182,11 @@ def resolve(self, bindings, *, condition=None, stream_collections=False, metadat
                     self.partial |= (summary.get("workflow_validation") or {}).get("status") == "accepted_partial"
                     continue
                 if metadata_only:
-                    manifest, access = authorize_workflow_node_result_read(
+                    manifest, _ = authorize_workflow_node_result_read(
                         self.workflow, self.run_id, summary["producer"], summary["result_ref"],
                         reader_user_id=self.actor_user_id, load_result=self.execution.load_result,
                     )
                     _require_completed_result(manifest, allow_partial=binding["allow_partial"])
-                    if access["source_snapshot_changed"]:
-                        raise AnalysisResultUnavailable("analysis_source_snapshot_changed")
                     if (manifest.get("workflow_validation") or {}).get("eligible") is not True:
                         raise WorkflowInputError("The selected output is not eligible.")
                     descriptor = manifest.get("outputs", {}).get(name)
@@ -417,7 +415,10 @@ def _for_each(self, node, region_id):
         except WorkflowLoopInputError as exc:
             self.execution.pause_input(exc.public_message, code=getattr(exc, "code", "workflow_loop_input_unavailable"))
         except AnalysisResultUnavailable:
-            self.execution.pause_input("The loop's original input sources are no longer available.", code="workflow_loop_source_unavailable")
+            self.execution.pause_input(
+                "The loop's input is no longer available or could not be verified.",
+                code="workflow_loop_source_unavailable",
+            )
         for index in range(state["next_index"], manifest["count"]):
             self.execution.set_node(node, region_id, iteration_path=parent_path, iteration_inputs=parent_receipts)
             self.execution.check()
@@ -537,12 +538,10 @@ def write():
             "collect", write, inputs={"source": identity, "manifest_ref": reference, "contract": node["output_contract"]},
             replay_safe=True,
         )
-        manifest, access = authorize_workflow_node_result_read(
+        manifest, _ = authorize_workflow_node_result_read(
             self.workflow, self.run_id, summary["producer"], summary["result_ref"],
             reader_user_id=self.actor_user_id, load_result=self.execution.load_result,
         )
-        if access["source_snapshot_changed"]:
-            self.execution.pause_input("A collected source changed. The saved records were retained.", code="workflow_loop_source_changed")
         validation = manifest["workflow_validation"]
         state = "completed" if validation["eligible"] else validation["status"]
         self.partial |= validation["status"] == "accepted_partial"