diff --git a/application/single_app/config.py b/application/single_app/config.py index 1bfe253f0..5caf0c223 100644 --- a/application/single_app/config.py +++ b/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 diff --git a/application/single_app/content_screening/access.py b/application/single_app/content_screening/access.py index f34c89d15..744835af8 100644 --- a/application/single_app/content_screening/access.py +++ b/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. diff --git a/application/single_app/functions_document_provenance.py b/application/single_app/functions_document_provenance.py index 2d7b3211b..1db04b787 100644 --- a/application/single_app/functions_document_provenance.py +++ b/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", diff --git a/application/single_app/functions_saved_analysis.py b/application/single_app/functions_saved_analysis.py index 7d4ee7a05..f63589e90 100644 --- a/application/single_app/functions_saved_analysis.py +++ b/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 []] diff --git a/application/single_app/functions_workflow_execution_history.py b/application/single_app/functions_workflow_execution_history.py index 901af0ece..891a9da72 100644 --- a/application/single_app/functions_workflow_execution_history.py +++ b/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 = { diff --git a/application/single_app/functions_workflow_flow_runner.py b/application/single_app/functions_workflow_flow_runner.py index 60f2f6e82..18a2224cd 100644 --- a/application/single_app/functions_workflow_flow_runner.py +++ b/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" diff --git a/application/single_app/functions_workflow_inspection.py b/application/single_app/functions_workflow_inspection.py index 43b340798..e2b8b54bb 100644 --- a/application/single_app/functions_workflow_inspection.py +++ b/application/single_app/functions_workflow_inspection.py @@ -5,9 +5,8 @@ import math import re -from functions_analysis_access import ( - AnalysisResultUnavailable, authorize_analysis_sources, resolve_analysis_source_manifest, -) +from content_screening.contracts import ScreeningError +from functions_analysis_access import AnalysisResultUnavailable, resolve_analysis_source_manifest from functions_document_analysis_results import normalize_analysis_options from functions_workflow_bindings import WorkflowInputError, authorize_workflow_reference from functions_workflow_definitions import ( @@ -28,6 +27,10 @@ "metadata", "alerts", "alert_settings", "document_actions", "publication", "publication_options", }) _DIGEST = re.compile(r"[a-f0-9]{64}\Z") +# A missing, inaccessible or held authored source is marked, not raised. +_UNAVAILABLE_SOURCE_ERRORS = ( + AnalysisResultUnavailable, WorkflowInputError, PermissionError, LookupError, ValueError, ScreeningError, +) class WorkflowFlowUnsupported(ValueError): @@ -94,8 +97,12 @@ def _action(value): return result -def _row(label, value): - return {"label": label, "value": value} +def _row(label, value, source=None): + row = {"label": label, "value": value} + if source is not None: + # Internal only: removed before a page is returned. + row["_source"] = source + return row def _label(value, fallback): @@ -309,18 +316,33 @@ def _selection(workflow, node, tasks): iterable = node["iterable"] rows = [_row("Iterable", {name: iterable[name] for name in ("kind", "name", "filters", "content", "selection") if name in iterable})] - rows.extend(_row("Document", value) for value in iterable.get("documents", [])) - rows.extend(_row("Workspace", value) for value in iterable.get("scopes", [])) + rows.extend(_row("Document", value, { + "kind": "document", "document_id": value["document_id"], "scope_type": value["scope_type"], + "scope_id": value.get("scope_id") or workflow["user_id"], + }) for value in iterable.get("documents", [])) + rows.extend(_row("Workspace", value, { + "kind": "scope", "scope_type": value["scope_type"], "scope_id": value.get("scope_id") or workflow["user_id"], + }) for value in iterable.get("scopes", [])) return rows if node["kind"] not in {"task", "region"}: return [] if node["kind"] == "region" and node["id"] != workflow["flow"]["id"]: return [] task = tasks[node["task_id"]] if node["kind"] == "task" else None - rows = [_row(reference["name"], reference) for reference in _references(workflow, task)] + rows = [ + _row(reference["name"], reference, {"kind": "reference", "reference": reference}) + for reference in _references(workflow, task) + ] action = (task if task is not None else workflow).get("document_action") or {} if not isinstance(action, dict): raise WorkflowDefinitionError("Document selection configuration must be an object.") + # Nothing is read for "none", and a current-item action reads its loop's frozen item. + reads_selection = action.get("type", "none") != "none" and action.get("target_mode") != "current_item" + selected = { + "kind": "selected_document", "doc_scope": action.get("doc_scope", "all"), + "active_group_ids": action.get("active_group_ids", []), + "active_public_workspace_ids": action.get("active_public_workspace_id", []), + } for name, label in ( ("document_ids", "Selected document"), ("right_document_ids", "Comparison document"), ("active_group_ids", "Source group"), ("active_public_workspace_id", "Source public workspace"), @@ -328,9 +350,21 @@ def _selection(workflow, node, tasks): values = action.get(name, []) if not isinstance(values, list) or len(values) > 5000: raise WorkflowDefinitionError("Document selections must be bounded identifier lists.") - rows.extend(_row(label, {name: _text(value)}) for value in values) + for value in values: + value = _text(value) + source = None + if reads_selection: + source = ( + {**selected, "document_id": value} if name.endswith("document_ids") + else {"kind": "scope", "scope_type": "group" if name == "active_group_ids" else "public", "scope_id": value} + ) + rows.append(_row(label, {name: value}, source)) if action.get("left_document_id"): - rows.append(_row("Comparison source", {"document_id": _text(action["left_document_id"])})) + value = _text(action["left_document_id"]) + rows.append(_row( + "Comparison source", {"document_id": value}, + {**selected, "document_id": value} if reads_selection else None, + )) return rows @@ -363,9 +397,95 @@ def _json_size(value): raise WorkflowDefinitionError("Workflow inspection requires bounded finite JSON.") from exc +class _SourceAvailability: + """Marks authored sources this reader can't use right now; it never denies the definition. + + A definition takes its access from its workflow, which the route already + checked. Only the source-selection rows on the requested page are looked up, + and a missing, inaccessible or held source gets ``"available": false`` on its + row. Storage failures still fail the request so an outage isn't shown as an + unavailable source. + """ + + def __init__(self, workflow, reader_user_id): + self.workflow = workflow + self.reader_user_id = reader_user_id + + def _scope(self, source): + if source["scope_type"] == "personal": + return source["scope_id"] == self.reader_user_id + try: + return _default_authorize_scope( + {"scope_type": source["scope_type"], "scope_id": source["scope_id"]}, + actor_user_id=self.reader_user_id, + ) is not False + except _UNAVAILABLE_SOURCE_ERRORS: + return False + + def _reference(self, source): + try: + authorize_workflow_reference(self.workflow, source["reference"], actor_user_id=self.reader_user_id) + except _UNAVAILABLE_SOURCE_ERRORS: + return False + return True + + def _documents(self, sources): + first = sources[0] + identifiers = list(dict.fromkeys(source["document_id"] for source in sources)) + if first["kind"] == "document": + scope, scope_id = first["scope_type"], first["scope_id"] + arguments = { + "doc_scope": scope, "active_group_ids": [scope_id] if scope == "group" else [], + "active_public_workspace_ids": [scope_id] if scope == "public" else [], + } + else: + scope = scope_id = None + arguments = {name: first[name] for name in ("doc_scope", "active_group_ids", "active_public_workspace_ids")} + try: + manifest = resolve_analysis_source_manifest(identifiers, self.reader_user_id, **arguments) + except _UNAVAILABLE_SOURCE_ERRORS: + return {} + return { + entry.get("document_id"): entry.get("authorization_status") == "authorized" and ( + scope is None or (entry.get("scope") == scope and entry.get("scope_id") == scope_id) + ) + for entry in manifest if isinstance(entry, dict) + } + + def annotate(self, rows, sources): + batches = {} + for index, source in enumerate(sources): + if source is not None and source["kind"] in {"document", "selected_document"}: + key = canonical_digest({name: value for name, value in source.items() if name != "document_id"}) + batches.setdefault(key, []).append(index) + available = {} + for indices in batches.values(): + found = self._documents([sources[index] for index in indices]) + for index in indices: + available[index] = found.get(sources[index]["document_id"], False) + marked = [] + for index, (row, source) in enumerate(zip(rows, sources)): + if source is not None: + usable = ( + self._scope(source) if source["kind"] == "scope" + else self._reference(source) if source["kind"] == "reference" + else available[index] + ) + if not usable: + row = {**row, "value": {**row["value"], "available": False}} + marked.append(row) + return marked + + def workflow_flow_inspection(workflow, *, source_kind="saved", run_id=None, snapshot_sha256=None, - node_id=None, section=None, revision=None, cursor=None, limit=50): - """Pure projection. Callers must first authorize this exact definition source.""" + node_id=None, section=None, revision=None, cursor=None, limit=50, + reader_user_id=None): + """Pure projection. Callers must first authorize this exact definition source. + + With ``reader_user_id``, the source-selection rows on the returned page mark + authored sources that reader can't use right now as unavailable. A missing, + inaccessible or held source never denies the definition or its topology. + """ if not isinstance(workflow, dict) or type(workflow.get("definition_version")) is not int or workflow["definition_version"] != 3: raise WorkflowFlowUnsupported() if source_kind not in {"saved", "draft", "run"} or (source_kind == "run") != (run_id is not None): @@ -407,9 +527,13 @@ def workflow_flow_inspection(workflow, *, source_kind="saved", run_id=None, snap rows = _detail_rows(workflow, compiled, node_id, section) if offset > len(rows) or cursor and offset == len(rows): raise ValueError("The detail cursor exceeds this definition section.") + page = [dict(row) for row in rows[offset:offset + limit]] + sources = [row.pop("_source", None) for row in page] + if reader_user_id is not None and any(source is not None for source in sources): + page = _SourceAvailability(workflow, reader_user_id).annotate(page, sources) result = {**scope, "items": [], "total_count": len(rows), "next_cursor": None} used = _json_size(result) + 1024 - for row in rows[offset:offset + limit]: + for row in page: size = _json_size(row) + 1 if size + _json_size({**scope, "total_count": len(rows), "next_cursor": None, "items": []}) + 1024 > FLOW_DETAIL_MAX_BYTES: raise WorkflowFlowDetailTooLarge() @@ -442,70 +566,6 @@ def preview_workflow_flow(definition, *, user_id, group_id=None, **selectors): return workflow_flow_inspection(workflow, source_kind="draft", **selectors) -def authorize_workflow_flow_sources(workflow, *, reader_user_id): - """Recheck declared source metadata, without expanding queries or reading results. - - Whole-run result authorization traverses all historical execution lineage. - Definition-only inspection instead checks its authored source boundaries; - execution overlays retain the existing exact payload/lineage authorization. - """ - if not isinstance(workflow, dict) or type(workflow.get("definition_version")) is not int or workflow["definition_version"] != 3: - raise WorkflowFlowUnsupported() - compiled = compile_workflow_flow(workflow) - try: - for reference in _references(workflow): - authorize_workflow_reference(workflow, reference, actor_user_id=reader_user_id) - sources = {} - scopes = {} - for entry in compiled["nodes"].values(): - node = entry["node"] - if node["kind"] != "for_each": - continue - iterable = node["iterable"] - for value in iterable.get("documents", []): - source = { - "document_id": value["document_id"], "scope": value["scope_type"], - "scope_id": value.get("scope_id") or workflow["user_id"], - } - sources[canonical_digest(source)] = source - for value in iterable.get("scopes", []): - scope = {"scope_type": value["scope_type"], "scope_id": value.get("scope_id") or workflow["user_id"]} - scopes[canonical_digest(scope)] = scope - for scope in scopes.values(): - if scope["scope_type"] == "personal" and scope["scope_id"] != reader_user_id: - raise PermissionError - if _default_authorize_scope(scope, actor_user_id=reader_user_id) is False: - raise PermissionError - if sources: - authorize_analysis_sources(reader_user_id, list(sources.values())) - for task in [workflow, *compiled["tasks"]]: - action = task.get("document_action") or {} - if not isinstance(action, dict): - raise WorkflowDefinitionError("Document selection configuration must be an object.") - if action.get("type", "none") == "none" or action.get("target_mode") == "current_item": - continue - for scope_type, field in (("group", "active_group_ids"), ("public", "active_public_workspace_id")): - for scope_id in action.get(field, []): - if _default_authorize_scope( - {"scope_type": scope_type, "scope_id": scope_id}, actor_user_id=reader_user_id, - ) is False: - raise PermissionError - identifiers = list(dict.fromkeys([ - *action.get("document_ids", []), *action.get("right_document_ids", []), - *([action["left_document_id"]] if action.get("left_document_id") else []), - ])) - if identifiers: - manifest = resolve_analysis_source_manifest( - identifiers, reader_user_id, doc_scope=action.get("doc_scope", "all"), - active_group_ids=action.get("active_group_ids", []), - active_public_workspace_ids=action.get("active_public_workspace_id", []), - ) - if any(source.get("authorization_status") != "authorized" for source in manifest): - raise AnalysisResultUnavailable() - except (WorkflowInputError, PermissionError, LookupError) as exc: - raise AnalysisResultUnavailable() from exc - - def workflow_run_flow_inspection(workflow, run_id, *, reader_user_id, **selectors): """Read only after the route has proved current workflow/run/scope access.""" store = workflow_runtime_store(workflow, run_id) @@ -522,7 +582,7 @@ def workflow_run_flow_inspection(workflow, run_id, *, reader_user_id, **selector raise WorkflowRuntimeConflict("workflow_definition_changed") if type(snapshot.get("definition_version")) is not int or snapshot["definition_version"] != 3: raise WorkflowFlowUnsupported() - authorize_workflow_flow_sources(snapshot, reader_user_id=reader_user_id) return workflow_flow_inspection( - snapshot, source_kind="run", run_id=run_id, snapshot_sha256=control["snapshot_ref"]["sha256"], **selectors, + snapshot, source_kind="run", run_id=run_id, snapshot_sha256=control["snapshot_ref"]["sha256"], + reader_user_id=reader_user_id, **selectors, ) diff --git a/application/single_app/functions_workflow_iterations.py b/application/single_app/functions_workflow_iterations.py index 4ba3788e6..8eaf38428 100644 --- a/application/single_app/functions_workflow_iterations.py +++ b/application/single_app/functions_workflow_iterations.py @@ -247,6 +247,7 @@ def frozen_item_receipt(manifest, reference, item): def _authorize_frozen_document(workflow, item, reader_user_id): + """Input check for a document item read as a loop input: access, not held, same frozen version.""" from functions_workflow_loop_inputs import WorkflowLoopInputError, reauthorize_workflow_loop_document try: @@ -343,6 +344,4 @@ def authorize_iteration_path(workflow, run_id, identity, *, reader_user_id, rece source_resolver=source_resolver, store=store, ) authorization.walk(dependencies) - if authorization.access()["source_snapshot_changed"]: - raise AnalysisResultUnavailable("analysis_source_snapshot_changed") return verified diff --git a/application/single_app/functions_workflow_loop_history.py b/application/single_app/functions_workflow_loop_history.py index 807fd4d81..e70dc6bea 100644 --- a/application/single_app/functions_workflow_loop_history.py +++ b/application/single_app/functions_workflow_loop_history.py @@ -141,7 +141,14 @@ def workflow_loop_items_page(workflow, run_id, execution_id, *, reader_user_id, items = [] for index in range(offset, min(manifest["count"], offset + limit)): item = read_frozen_item(workflow, run_id, manifest, index) - current = load_frozen_item_value(workflow, run_id, manifest, item, reader_user_id=reader_user_id) + if item["kind"] == "document": + # A frozen document item is provenance in history; it is not re-checked against the document. + document = item.get("document") if isinstance(item.get("document"), dict) else {} + label = str(document.get("file_name") or f"Item {index + 1}")[:256] + else: + # Proves the frozen record still matches its saved source collection. + load_frozen_item_value(workflow, run_id, manifest, item, reader_user_id=reader_user_id) + label = f"Item {index + 1}" outcome = store.journal_read("iteration", [execution_id, item["item_id"]]) payload = (outcome or {}).get("payload") or {} path = identity["iteration_path"] + [{ @@ -154,8 +161,7 @@ def workflow_loop_items_page(workflow, run_id, execution_id, *, reader_user_id, executions.append(child_id) projected = { "item_id": item["item_id"], "index": index, "iteration_path": path, - "label": str(current["value"].get("file_name") or f"Item {index + 1}")[:256] - if item["kind"] == "document" else f"Item {index + 1}", + "label": label, "state": payload.get("state", "queued"), "execution_ids": executions, } if len(json.dumps(items + [projected], ensure_ascii=True).encode("ascii")) > 240 * 1024: diff --git a/application/single_app/functions_workflow_node_results.py b/application/single_app/functions_workflow_node_results.py index ee102b39c..745b284c3 100644 --- a/application/single_app/functions_workflow_node_results.py +++ b/application/single_app/functions_workflow_node_results.py @@ -6,7 +6,7 @@ from collections.abc import Mapping from copy import deepcopy -from functions_analysis_access import AnalysisResultUnavailable, analysis_source_snapshot, authorize_analysis_sources +from functions_analysis_access import AnalysisResultUnavailable, analysis_source_snapshot from functions_workflow_bindings import WorkflowInputError from functions_workflow_identity import canonical_digest, workflow_node_identity from functions_workflow_limits import WORKFLOW_MAX_EXECUTION_ADMISSIONS @@ -59,24 +59,32 @@ def read_consumed_inputs(manifest, load_section): class WorkflowLineageAuthorization: - """One bounded iterative graph for results, mixed paths and temporal state proofs.""" + """One bounded iterative graph for results, mixed paths and temporal state proofs. + + The walk proves identity, lineage and integrity: each result belongs to this + workflow and run, its hashes and references match, and its consumed-input + receipts chain correctly. A saved result takes its access from its workflow + and run, so contributing sources are collected as provenance and are never + re-resolved for the reader. Frozen document items are provenance here too; + reading one as an input checks it (``load_frozen_item_value``). + ``source_resolver`` is accepted for existing callers and is not used. + """ def __init__(self, workflow, run_id, *, reader_user_id=None, load_result=load_workflow_node_result, source_resolver=None, include_sources=True, source_callback=None, store=None): self.workflow, self.run_id = workflow, run_id self.reader_user_id = reader_user_id or workflow["user_id"] - self.load_result, self.source_resolver = load_result, source_resolver + self.load_result = load_result self.include_sources, self.source_callback = include_sources, source_callback self._store = store self._compiled = None self.loaded = OrderedDict() self.active, self.visited = set(), set() - self.sources, self.source_ids, self.source_batch = {}, set(), [] - self.changed = False + self.sources, self.source_ids = {}, set() self._recording = None self._proof_cache = None # Reuse immutable graph structure only within this active worker request; - # source, document and publication authority is never cached. + # publication destination authority is never cached. from functions_workflow_execution import current_workflow_execution execution = current_workflow_execution() @@ -115,16 +123,6 @@ def load(self, identity, reference): self.loaded.move_to_end(key) return self.loaded[key] - def _flush_sources(self): - if not self.source_batch: - return - checked = authorize_analysis_sources(self.reader_user_id, self.source_batch, resolver=self.source_resolver) - self.changed |= checked["source_snapshot_changed"] - if self.source_callback is not None: - for source in self.source_batch: - self.source_callback(source) - self.source_batch.clear() - def source_seen(self, source): source = analysis_source_snapshot([source])[0] if self._recording is not None: @@ -135,16 +133,8 @@ def source_seen(self, source): self.source_ids.add(digest) if self.include_sources: self.sources[digest] = source - self.source_batch.append(source) - if len(self.source_batch) >= 100: - self._flush_sources() - - def _document_access(self, item): - from functions_workflow_iterations import _authorize_frozen_document - - _authorize_frozen_document(self.workflow, item, self.reader_user_id) - if self._recording is not None: - self._recording["documents"].append(item) + if self.source_callback is not None: + self.source_callback(source) def _publication_access(self, publication): from functions_artifact_publication import authorize_publication_status_read @@ -163,13 +153,11 @@ def _cached_children(self, key, proof): entry, _ = cached for source in entry["sources"]: self.source_seen(source) - for item in entry["documents"]: - self._document_access(item) for publication in entry["publications"]: self._publication_access(publication) yield from entry["dependencies"] return - entry = {"dependencies": [], "sources": [], "documents": [], "publications": []} + entry = {"dependencies": [], "sources": [], "publications": []} children = iter(self._children(proof)) while True: self._recording = entry if cache is not None else None @@ -192,9 +180,10 @@ def _cached_children(self, key, proof): cache["bytes"] -= removed_size def access(self): - self._flush_sources() + # source_snapshot_changed stays in the shape shared with saved analysis + # readers; a saved result is never compared with its sources' current versions. return { - "source_count": len(self.source_ids), "source_snapshot_changed": self.changed, + "source_count": len(self.source_ids), "source_snapshot_changed": False, "sources": list(self.sources.values()) if self.include_sources else None, } @@ -218,7 +207,6 @@ def walk(self, roots): continue self.active.add(child_key) pending.append((child_key, iter(self._cached_children(child_key, proof)))) - self._flush_sources() def _children(self, proof): kind, *arguments = proof @@ -359,7 +347,7 @@ def _children(self, proof): if kind == "frozen_item" and item["item_id"] != arguments[2]: raise AnalysisResultUnavailable("workflow_loop_item_invalid") if item["kind"] == "document": - self._document_access(item) + # The frozen item is provenance; a body that reads it as input checks it then. self.source_seen(item["source"]) elif kind == "frozen_item": receipt = manifest["source_receipt"] @@ -426,7 +414,7 @@ def authorize_workflow_node_result_read( class AuthorizedWorkflowRecordInput: - """A source-authorized immutable record range, not a preview or access token.""" + """A lineage-verified immutable record range, not a preview or access token.""" def __init__(self, workflow, run_id, identity, reference, *, output_name="authoritative", reader_user_id=None, allow_partial=False, load_result=load_workflow_node_result, @@ -506,8 +494,6 @@ def _authorize(self): if getattr(self, "receipt", {}).get("repeat_state"): authorization.walk([("receipt", self.receipt)]) access = authorization.access() - if access["source_snapshot_changed"] and not self.inspection: - raise AnalysisResultUnavailable("analysis_source_snapshot_changed") return manifest, access def _load(self, reference): @@ -586,8 +572,6 @@ def load_workflow_node_input( load_result=load_result, source_resolver=source_resolver, ) _require_completed_result(manifest, allow_partial=allow_partial) - if access["source_snapshot_changed"]: - raise AnalysisResultUnavailable("analysis_source_snapshot_changed") validation = manifest.get("workflow_validation") or {} if validation.get("eligible") is not True: raise ValueError("The producer's output contract is not eligible.") @@ -648,5 +632,6 @@ def load_workflow_node_input( "consumed_result": value_receipt, "provenance": manifest.get("provenance") or {}, "coverage": manifest.get("coverage") or {}, "validation": manifest.get("validation") or {}, "kind": output["kind"], "value": output["value"], - "source_snapshot_changed": access["source_snapshot_changed"], + # A fixed field keeps saved input digests stable; sources are not re-checked. + "source_snapshot_changed": False, }, ensure_ascii=False, allow_nan=False, sort_keys=True), receipt diff --git a/application/single_app/functions_workflow_repeat_execution.py b/application/single_app/functions_workflow_repeat_execution.py index 6939e67b2..0fb0c4361 100644 --- a/application/single_app/functions_workflow_repeat_execution.py +++ b/application/single_app/functions_workflow_repeat_execution.py @@ -92,7 +92,10 @@ def run_repeat_until(flow, node, region_id): try: initial, initial_ref = prepare_repeat_state(flow, node, identity) except AnalysisResultUnavailable: - execution.pause_input("The Repeat state's original sources are no longer available.", code="workflow_repeat_source_unavailable") + execution.pause_input( + "The Repeat state's input is no longer available or could not be verified. Saved originals are retained.", + code="workflow_repeat_source_unavailable", + ) except (WorkflowInputError, WorkflowResultNotReadyError, ValueError): execution.pause_input( "The required initial Repeat state did not satisfy its contract. Saved originals are retained.", @@ -157,8 +160,6 @@ def run_repeat_until(flow, node, region_id): flow.workflow, flow.run_id, reader_user_id=flow.actor_user_id, store=store, load_result=execution.load_result, ) before = authorization.authorize_repeat(identity, admission["before_state_ref"]) - if authorization.access()["source_snapshot_changed"]: - raise AnalysisResultUnavailable("analysis_source_snapshot_changed") execution.set_node( None, node["body"]["id"], iteration_path=path, iteration_inputs=parent_receipts + [iteration_receipt], ) @@ -182,7 +183,7 @@ def run_repeat_until(flow, node, region_id): except AnalysisResultUnavailable: execution.set_node(node, region_id, iteration_path=parent_path, iteration_inputs=parent_receipts) execution.pause_input( - "The Repeat state's original sources are no longer available. Saved originals are retained.", + "The Repeat state's input is no longer available or could not be verified. Saved originals are retained.", code="workflow_repeat_source_unavailable", ) except (WorkflowInputError, WorkflowResultNotReadyError, ValueError): diff --git a/application/single_app/functions_workflow_repeat_history.py b/application/single_app/functions_workflow_repeat_history.py index 66ed1f833..d2dc4b7d1 100644 --- a/application/single_app/functions_workflow_repeat_history.py +++ b/application/single_app/functions_workflow_repeat_history.py @@ -101,7 +101,8 @@ def workflow_repeat_iterations_page(workflow, run_id, execution_id, *, reader_us return { "iterations": items, "total_count": through, "next_cursor": next_cursor, "repeat_execution_id": execution_id, "repeat": repeat_summary(head), - "source_snapshot_changed": authorization.access()["source_snapshot_changed"], + # Retained for older clients; saved rounds are never compared with current sources. + "source_snapshot_changed": False, } @@ -159,5 +160,6 @@ def workflow_repeat_state_page(workflow, run_id, execution_id, iteration, *, rea return { **result, "states": states, "available": True, "total_count": len(declarations), "next_cursor": _next_cursor(scope, offset + len(states), len(declarations)), - "partial": state["partial"], "source_snapshot_changed": authorization.access()["source_snapshot_changed"], + # Retained for older clients; saved state is never compared with current sources. + "partial": state["partial"], "source_snapshot_changed": False, } diff --git a/application/single_app/functions_workflow_repeat_state.py b/application/single_app/functions_workflow_repeat_state.py index ac57d7e58..aeedc766e 100644 --- a/application/single_app/functions_workflow_repeat_state.py +++ b/application/single_app/functions_workflow_repeat_state.py @@ -316,8 +316,6 @@ def current_repeat_state(workflow, run_id, path, source, *, reader_user_id, stor workflow, run_id, reader_user_id=reader_user_id, load_result=load_result, store=store, ) state = authorization.authorize_repeat(identity, admission["before_state_ref"]) - if authorization.access()["source_snapshot_changed"]: - raise AnalysisResultUnavailable("analysis_source_snapshot_changed") slot = state["slots"].get(source["state_name"]) if slot is None: raise WorkflowInputError("The requested Repeat state slot is not declared.") @@ -384,13 +382,11 @@ def save(value): receipt = deepcopy(outputs.get(declaration["next"])) if receipt is None: raise WorkflowInputError("A required next-state output did not finish.") - manifest, access = authorize_workflow_node_result_read( + manifest, _ = authorize_workflow_node_result_read( flow.workflow, flow.run_id, receipt["producer"], receipt["result_ref"], reader_user_id=flow.actor_user_id, load_result=execution.load_result, ) _require_completed_result(manifest, allow_partial=contract["allow_partial"]) - if access["source_snapshot_changed"]: - raise AnalysisResultUnavailable("analysis_source_snapshot_changed") descriptor = manifest["outputs"].get(receipt["output_name"]) or {} if descriptor.get("result_ref") != receipt["output_ref"] or descriptor.get("kind") != contract["kind"]: raise WorkflowInputError("Repeat state must preserve its exact declared output kind.") @@ -494,8 +490,6 @@ def prepare_repeat_grant(store, control, gate, *, actor_user_id): workflow, store.identity["run_id"], reader_user_id=reader, store=store, ) authorization.authorize_repeat(identity, head["current_state_ref"]) - if authorization.access()["source_snapshot_changed"]: - raise AnalysisResultUnavailable("analysis_source_snapshot_changed") return row, { **head, "state": "running", "batch_number": head["batch_number"] + 1, "batch_start_iteration": head["next_iteration"], "batch_usage": 0, diff --git a/application/single_app/functions_workflow_reporting.py b/application/single_app/functions_workflow_reporting.py index 234c81f8f..6dde2b63f 100644 --- a/application/single_app/functions_workflow_reporting.py +++ b/application/single_app/functions_workflow_reporting.py @@ -646,8 +646,6 @@ def finish(self, reply, *, mode, page_count, reduction_levels, supported=None, r partial = any(reader.metadata()["accepted_subset_only"] for reader in self.inputs) if partial: reply += "\n\n**Partial saved result:** only the explicitly accepted subset is described; unresolved work remains." - if any(reader.access.get("source_snapshot_changed") for reader in self.inputs): - reply += "\n\n**Source snapshot changed:** this explanation describes the saved revisions, not current source contents." consumption = { "input_kind": "workflow_records", "version": WORKFLOW_RECORD_REPORT_VERSION, "mode": mode, "record_count": self.record_count, "page_count": page_count, diff --git a/application/single_app/functions_workflow_result_reader.py b/application/single_app/functions_workflow_result_reader.py index 2838447e7..726eaf56e 100644 --- a/application/single_app/functions_workflow_result_reader.py +++ b/application/single_app/functions_workflow_result_reader.py @@ -3,10 +3,12 @@ The reader is Flask-free. It point-reads the workflow and the run in the requester's own partition, so someone else's run reads exactly like a missing -one. It authorizes the whole run the way run history does, binds the result to -a digest over its included outputs, and returns a public descriptor. Follow up -also asks for bounded excerpts; those carry only a task label, a kind and text, -never store references or identifiers. +one. It verifies the identity and lineage of every stored task result, binds the +result to a digest over its included outputs, and returns a public descriptor. +A stored result takes its access from that workflow and run: the documents its +tasks used are provenance and are never re-checked. Follow up also asks for +bounded excerpts; those carry only a task label, a kind and text, never store +references or identifiers. Structured (v3) runs are closed for now: their outputs need an exact node, execution and attempt selector. @@ -26,6 +28,7 @@ from azure.core.exceptions import AzureError, ResourceNotFoundError from content_screening.contracts import ScreeningError +from functions_analysis_access import AnalysisResultUnavailable from functions_appinsights import log_event from functions_workflow_result_masking import WORKFLOW_RESULT_VERSION from functions_workflow_result_store import ( @@ -80,7 +83,7 @@ "workflow_result_private_only": (403, "Workflow results can only be used in your own private chats."), "workflow_result_not_found": (404, "This workflow result is unavailable. The run may have been removed."), "workflow_result_access_denied": ( - 403, "This workflow result is unavailable because access to one of its sources could not be confirmed.", + 403, "This workflow result is unavailable because access to it could not be confirmed.", ), "workflow_result_changed": ( 409, @@ -424,6 +427,9 @@ def workflow_result_digest(workflow_id, run_id, status, eligible_rows): def _authorization_code(error): if isinstance(error, WorkflowResultUnavailable): return error.code + if isinstance(error, AnalysisResultUnavailable): + # Lineage or integrity: the stored result itself doesn't check out. + return "workflow_result_invalid" if isinstance(error, ScreeningError): return "workflow_result_storage_unavailable" if getattr(error, "retryable", False) else "workflow_result_access_denied" if isinstance(error, PermissionError): @@ -734,8 +740,8 @@ def read_workflow_result( if expected_sha256 is not None and expected_sha256 != digest: raise _closed("workflow_result_changed", "digest") - # The whole run, including every stored task result and its analysis sources, - # is authorized exactly as run history is. The rows are the digest's rows. + # Every stored task result in the run is verified for identity and lineage, and + # the run takes its access from the point reads above. The rows are the digest's rows. _guarded("authorize", lambda: authorize_workflow_run_read( workflow, run_id, reader_user_id=user_id, result_items=rows, load_result=loader, source_resolver=source_resolver, @@ -803,7 +809,7 @@ def _build_excerpts(result, workflow, run_id, eligible, manifests, loader, page_ source_resolver=source_resolver, bounded=True, allow_partial=allow_partial, )) if not isinstance(saved_input, SavedAnalysisInput): - # No analysis sources to verify against: read it like any other output. + # Not a saved analysis to explain: read it like any other output. plain_rows.append(row) continue key = saved_input.context.get("result_sha256") diff --git a/application/single_app/functions_workflow_result_store.py b/application/single_app/functions_workflow_result_store.py index ec3687c01..820f1db66 100644 --- a/application/single_app/functions_workflow_result_store.py +++ b/application/single_app/functions_workflow_result_store.py @@ -4,8 +4,9 @@ Version: 0.261.127 Implemented in: 0.261.106 -Callers must authorize the current workflow/chat/orchestration context and -contributing sources before using this module. Storage additionally binds every +Callers must authorize the current workflow/chat/orchestration context before +using this module; a saved workflow result takes its access from its workflow and +run, not from its contributing sources. Storage additionally binds every object to its personal user/group, workflow, run, and task; its original chat owner, conversation, and real assistant message; or its original owner, conversation, real orchestration run, and step. It does not interpret the result contract or diff --git a/application/single_app/functions_workflow_results.py b/application/single_app/functions_workflow_results.py index a350bc54e..a5a8cba08 100644 --- a/application/single_app/functions_workflow_results.py +++ b/application/single_app/functions_workflow_results.py @@ -9,7 +9,6 @@ from functions_analysis_access import ( AnalysisResultUnavailable, analysis_source_snapshot, - authorize_analysis_sources, ) from functions_artifact_publication_readiness import public_publication_status from functions_workflow_result_store import ( @@ -359,7 +358,14 @@ def authorize_workflow_task_result_read( workflow, run_id, task_id, reference, *, reader_user_id=None, manifest=None, load_result=load_workflow_task_result, source_resolver=None, **selectors, ): - """Recheck contributors of this result and every actually consumed ancestor.""" + """Verify this result and every actually consumed ancestor; their sources are provenance. + + A saved result takes its access from its workflow and run, which the caller has + already authorized. This proves identity, lineage and integrity only: contract + versions, producer identities, hashes and consumed-input receipts. Contributing + sources are returned so callers can carry ``analysis_access`` forward; they are + never re-resolved for the reader. ``source_resolver`` is accepted and unused. + """ if selectors: from functions_workflow_identity import workflow_node_identity from functions_workflow_node_results import authorize_workflow_node_result_read @@ -447,19 +453,12 @@ def visit(current, current_task_id, current_ref): loaded[(str(task_id), reference["sha256"])] = (dict(reference), root) visit(root, task_id, reference) snapshots = analysis_source_snapshot(sources) - access = ( - authorize_analysis_sources( - reader_user_id if reader_user_id is not None else workflow.get("user_id"), - snapshots, resolver=source_resolver, - ) - if snapshots else {"source_count": 0, "source_snapshot_changed": False} - ) - return root, {**access, "sources": snapshots} + return root, {"source_count": len(snapshots), "source_snapshot_changed": False, "sources": snapshots} def authorize_workflow_run_read(workflow, run_id, *, reader_user_id=None, result_items=None, load_result=load_workflow_task_result, source_resolver=None): - """Guard history/activity with every stored task result, without a UI item cap.""" + """Verify the identity and lineage of every stored task result in a run, without a UI item cap.""" structured_run = False if result_items is None: # These are already scope-authorized workflow/run reads. Query only task @@ -898,7 +897,8 @@ def load_workflow_task_input(workflow, run_id, task_id, reference, "provenance": manifest.get("provenance") or {}, "coverage": manifest.get("coverage") or {}, "validation": public_analysis_validation(manifest.get("validation")) if access["source_count"] else manifest.get("validation") or {}, - "source_snapshot_changed": access["source_snapshot_changed"], + # A fixed field keeps saved input digests stable; sources are not re-checked. + "source_snapshot_changed": False, "kind": output["kind"], "value": output["value"], **({ diff --git a/application/single_app/functions_workflow_runner.py b/application/single_app/functions_workflow_runner.py index 3526f478d..27739c9d4 100644 --- a/application/single_app/functions_workflow_runner.py +++ b/application/single_app/functions_workflow_runner.py @@ -27,6 +27,7 @@ get_bearer_token_provider, ) from flask import Flask, g, has_request_context, session +from content_screening.access import isolate_request_source_fence from functions_m365_approvals import M365ApprovalRequired, M365PolicyError from functions_workflow_alert_safety import sanitize_workflow_alert_decision from m365_interaction import M365_AUTH_INTERACTION_CODES, M365SignInRequired @@ -10762,7 +10763,7 @@ def raise_if_cancelled(): checkpoint_ref = summary.get('result_ref') if checkpoint_ref: try: - _, source_access = authorize_workflow_task_result_read( + authorize_workflow_task_result_read( workflow, run_id, task_id, checkpoint_ref, reader_user_id=actor_id, **(durable.selectors(attempt=summary['producer']['attempt']) if structured_definition else {}), ) @@ -10770,8 +10771,6 @@ def raise_if_cancelled(): if not structured_definition: raise durable._pause(task_unit_key, checkpoint_ref['sha256']) - if structured_definition and source_access.get('source_snapshot_changed'): - durable._pause(task_unit_key, checkpoint_ref['sha256']) task_results.append(completed_task) validation = saved_result.get('workflow_validation') or { 'eligible': completed_task.get('status') == 'succeeded', @@ -10827,6 +10826,9 @@ def raise_if_cancelled(): raise_if_cancelled() attempt_count = attempt_index + 1 task_stage = 'runner' + # Each attempt gets its own model fence: only the documents this task reads as + # inputs can block its model calls, never an earlier task's sources. + source_fence = isolate_request_source_fence() try: if task.get('publication') is not None: task_stage = 'publication' @@ -11095,6 +11097,8 @@ def dispatch_task(): title=str(task.get('name') or f"Task {task['order']}"), status='running', ) + finally: + source_fence.restore() if task_result is not None: # Persistence is outside the invocation retry loop: a failed write @@ -11243,7 +11247,12 @@ def save_analysis_section(bound_workflow, bound_run, bound_task, section, **_kwa error=task_error, ) except AnalysisResultUnavailable as exc: - message = 'Saved task output was withheld because its source access could not be confirmed.' + # 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 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, diff --git a/application/single_app/functions_workflow_runtime.py b/application/single_app/functions_workflow_runtime.py index 7a38920c7..513888f14 100644 --- a/application/single_app/functions_workflow_runtime.py +++ b/application/single_app/functions_workflow_runtime.py @@ -361,18 +361,16 @@ def _project_runtime_run(services, workflow, run_id, control, *, result=None, at def workflow_runtime_status(workflow, run_id, *, reader_user_id): - from functions_workflow_results import authorize_workflow_run_read - services = _services(workflow) run = services["runs"].read_item(item=run_id, partition_key=services["partition"]) if run.get("workflow_id") != workflow["id"] or run.get("durable_execution") is not True: raise LookupError("Durable workflow run not found.") - authorize_workflow_run_read(workflow, run_id, reader_user_id=reader_user_id) + # The run takes its access from the workflow the caller already read; its sources are not re-checked. store = workflow_runtime_store(workflow, run_id) control = store.read() gate = control.get("gate") or {} if gate.get("publication"): - # Polling UI has the same source and destination boundary as exact result inspection. + # Polling UI has the same lineage and destination boundary as exact result inspection. from functions_artifact_publication import authorize_publication_status_read from functions_workflow_execution_history import authorize_execution_payload diff --git a/application/single_app/functions_workflow_structured_execution.py b/application/single_app/functions_workflow_structured_execution.py index 92e66f854..bbe37c57a 100644 --- a/application/single_app/functions_workflow_structured_execution.py +++ b/application/single_app/functions_workflow_structured_execution.py @@ -291,8 +291,8 @@ def authorize_iteration(self): ) except AnalysisResultUnavailable: self.pause_input( - "The Repeat state's original sources are no longer available. Saved originals are retained.", - code="workflow_repeat_source_unavailable", + "This Repeat round's saved state could not be verified. Saved originals are retained.", + code="workflow_repeat_state_invalid", ) def _attempt(self, attempt, **fields): diff --git a/application/single_app/route_backend_workflows.py b/application/single_app/route_backend_workflows.py index 32182947f..5787da21b 100644 --- a/application/single_app/route_backend_workflows.py +++ b/application/single_app/route_backend_workflows.py @@ -111,7 +111,7 @@ from functions_workflow_runtime_store import RuntimeUnavailable, WorkflowRuntimeConflict from functions_workflow_execution_history import workflow_execution_history, workflow_execution_result_page from functions_workflow_inspection import ( - WorkflowFlowDetailTooLarge, WorkflowFlowUnsupported, authorize_workflow_flow_sources, preview_workflow_flow, + WorkflowFlowDetailTooLarge, WorkflowFlowUnsupported, preview_workflow_flow, workflow_flow_inspection, workflow_run_flow_inspection, ) from functions_workflow_node_results import WorkflowRecordPageTooLarge @@ -209,6 +209,9 @@ def _workflow_task_result_page_response(workflow, run_record, task_id, get_item) workflow, run_id, task_id, result_ref, offset=offset, limit=limit, ) return jsonify({**page, 'output_name': output_name}) + except AnalysisResultUnavailable: + # Lineage or integrity; the result takes its access from the workflow and run above. + return jsonify({'error': 'The saved result or requested page is unavailable.'}), 409 except PermissionError: return jsonify({'error': 'Access to this task result is not allowed.'}), 403 except LookupError: @@ -418,8 +421,9 @@ def _workflow_flow_response(workflow_id=None, run_id=None, *, group=False, previ raise WorkflowFlowUnsupported() response = workflow_run_flow_inspection(workflow, run_id, reader_user_id=user_id, **selectors) else: - authorize_workflow_flow_sources(workflow, reader_user_id=user_id) - response = workflow_flow_inspection(workflow, **selectors) + # The definition takes its access from the workflow; an unavailable + # authored source is marked on its own row instead of denying the definition. + response = workflow_flow_inspection(workflow, reader_user_id=user_id, **selectors) return jsonify(response) except WorkflowFlowDetailTooLarge as exc: return jsonify({'error': exc.public_message, 'code': exc.code}), 413 @@ -431,8 +435,8 @@ def _workflow_flow_response(workflow_id=None, run_id=None, *, group=False, previ return jsonify({'error': exc.public_message, 'code': 'invalid_workflow_definition'}), 400 except WorkflowRuntimeConflict as exc: return jsonify({'error': exc.public_message, 'code': exc.code}), 409 - except (PermissionError, AnalysisResultUnavailable): - return jsonify({'error': 'Current access to this workflow definition or its sources could not be confirmed.'}), 403 + except PermissionError: + return jsonify({'error': 'Current access to this workflow could not be confirmed.'}), 403 except (LookupError, CosmosResourceNotFoundError): return jsonify({'error': 'The selected workflow definition, run or structural node was not found.'}), 404 except (ValueError, TypeError, RecursionError): @@ -517,8 +521,11 @@ def _workflow_execution_history_response(workflow_id, run_id, *, group=False, ki return jsonify(response) except WorkflowRuntimeConflict as exc: return jsonify({'error': exc.public_message, 'code': exc.code}), 409 - except (PermissionError, AnalysisResultUnavailable): - return jsonify({'error': 'Current access to this execution or its contributing sources could not be confirmed.'}), 403 + except AnalysisResultUnavailable: + # Lineage or integrity: a saved record doesn't check out. Sources are never re-checked here. + return jsonify({'error': 'This saved execution record could not be verified.'}), 409 + except PermissionError: + return jsonify({'error': 'Current access to this execution could not be confirmed.'}), 403 except (LookupError, CosmosResourceNotFoundError): return jsonify({'error': 'Workflow execution or attempt not found.'}), 404 except WorkflowRecordPageTooLarge as exc: @@ -1133,7 +1140,8 @@ def _stream_workflow_activity(user_id, conversation_id='', workflow_id='', run_i user_id, conversation_id=conversation_id, workflow_id=workflow_id, run_id=run_id, ) except AnalysisResultUnavailable: - yield 'event: error\ndata: {"error":"Workflow source access is no longer available.","code":"source_access_denied"}\n\n' + # A saved record that fails its lineage check; source documents are never re-checked here. + yield 'event: error\ndata: {"error":"Workflow activity is no longer available.","code":"activity_unavailable"}\n\n' return payload = json.dumps(snapshot, default=str, sort_keys=True) @@ -1166,7 +1174,8 @@ def _stream_group_workflow_activity(user_id, group_id, conversation_id='', workf user_id, group_id, conversation_id=conversation_id, workflow_id=workflow_id, run_id=run_id, ) except AnalysisResultUnavailable: - yield 'event: error\ndata: {"error":"Workflow source access is no longer available.","code":"source_access_denied"}\n\n' + # A saved record that fails its lineage check; source documents are never re-checked here. + yield 'event: error\ndata: {"error":"Workflow activity is no longer available.","code":"activity_unavailable"}\n\n' return payload = json.dumps(snapshot, default=str, sort_keys=True) diff --git a/application/v2_ui/src/components/workflows/WorkflowDefinitionInspector.tsx b/application/v2_ui/src/components/workflows/WorkflowDefinitionInspector.tsx index 4dee4c08a..f1f80e28d 100644 --- a/application/v2_ui/src/components/workflows/WorkflowDefinitionInspector.tsx +++ b/application/v2_ui/src/components/workflows/WorkflowDefinitionInspector.tsx @@ -17,6 +17,12 @@ const sectionLabels: Record = { outputs: 'Declared outputs', state: 'Authored Repeat state', selection: 'Source selection', }; +/** The server marks an authored source this reader can't use now; the definition still opens. */ +function isUnavailableSource(value: InspectionJson): boolean { + return Boolean(value) && typeof value === 'object' && !Array.isArray(value) && + (value as Record).available === false; +} + function DetailValue({ value, onSelectNode }: { value: InspectionJson; onSelectNode: (id: string) => void }) { if (isFlowBinding(value)) { const source = value.source; @@ -140,7 +146,12 @@ export function WorkflowDefinitionInspector({
{details.items.map((item, index) =>
{item.label}
-
+
+ {isUnavailableSource(item.value) ?

+ Not available to you right now. It may have been removed, be held for review, or be outside your access. The workflow definition is unchanged. +

: null} + +
)}
{!details.items.length ?

No entries are declared in this section.

: null} diff --git a/application/v2_ui/src/components/workflows/WorkflowExecutionHistory.tsx b/application/v2_ui/src/components/workflows/WorkflowExecutionHistory.tsx index a97c35265..d41602237 100644 --- a/application/v2_ui/src/components/workflows/WorkflowExecutionHistory.tsx +++ b/application/v2_ui/src/components/workflows/WorkflowExecutionHistory.tsx @@ -770,7 +770,6 @@ function RepeatStatePages({ scope, workflowId, runId, executionId, iteration, ph {page.metadata?.partial ?

This state retains accepted partial data. Its coverage and limitations are not cleared by a later round or manual continuation.

: null} - {page.metadata?.sourceSnapshotChanged ?

Source snapshots have changed. These are retained saved versions, with access checked on every read.

: null}