From 03fce990bbf0b4984c54eae7c793d2fb11acfcb1 Mon Sep 17 00:00:00 2001 From: Paul Lizer Date: Sat, 3 Oct 2026 02:58:50 -0400 Subject: [PATCH 1/2] Stop re-checking saved workflow results against their source documents Saved and generated workflow results now take their access from the workflow, run and group they belong to. Task and node results, structured execution history, For each and Repeat history, run history previews, chat Follow up on a workflow result and the document provenance run link no longer re-resolve the documents a run read; those are kept as provenance. Lineage, contract and hash integrity checks stay, as do the container checks and the input-time checks when a workflow reads an uploaded document. Each task attempt gets its own model fence. Flow inspection marks an unavailable authored source on its own selection row instead of denying the definition. Lineage failures answer 409, activity streams end with activity_unavailable, and source_snapshot_changed is always false. Version 0.261.231. Refs #1621 (layer 3a). Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- application/single_app/config.py | 2 +- .../single_app/content_screening/access.py | 48 ++ .../functions_document_provenance.py | 14 +- .../single_app/functions_saved_analysis.py | 19 +- .../functions_workflow_execution_history.py | 17 +- .../functions_workflow_flow_runner.py | 13 +- .../functions_workflow_inspection.py | 218 ++++-- .../functions_workflow_iterations.py | 3 +- .../functions_workflow_loop_history.py | 12 +- .../functions_workflow_node_results.py | 63 +- .../functions_workflow_repeat_execution.py | 9 +- .../functions_workflow_repeat_history.py | 6 +- .../functions_workflow_repeat_state.py | 8 +- .../functions_workflow_reporting.py | 2 - .../functions_workflow_result_reader.py | 22 +- .../functions_workflow_result_store.py | 5 +- .../single_app/functions_workflow_results.py | 24 +- .../single_app/functions_workflow_runner.py | 17 +- .../single_app/functions_workflow_runtime.py | 6 +- ...functions_workflow_structured_execution.py | 4 +- .../single_app/route_backend_workflows.py | 27 +- .../workflows/WorkflowDefinitionInspector.tsx | 13 +- .../workflows/WorkflowExecutionHistory.tsx | 2 - .../components/workflows/WorkflowFlowView.tsx | 2 +- .../v2_ui/src/lib/workflowExecutionHistory.ts | 13 +- application/v2_ui/src/lib/workflowResults.ts | 7 +- docs/admin/workflow.md | 4 +- .../CHAT_WORKFLOW_RESULTS_FOLLOW_UP.md | 31 +- .../features/CHAT_WORKFLOW_RESULT_DELIVERY.md | 5 +- .../features/CONTENT_SCREENING_FRAMEWORK.md | 2 + .../features/WORKFLOW_DURABLE_EXECUTION.md | 3 +- .../features/WORKFLOW_EXPLICIT_DATA_FLOW.md | 15 +- .../features/WORKFLOW_FLOW_INSPECTION.md | 20 +- .../features/WORKFLOW_FOR_EACH_COLLECT.md | 10 +- .../WORKFLOW_PUBLICATION_COMPLETION.md | 3 +- .../features/WORKFLOW_REPEAT_UNTIL.md | 4 + .../features/WORKFLOW_RESULT_READERS.md | 44 +- .../WORKFLOW_SAVED_OUTPUT_PUBLICATION.md | 11 +- .../WORKFLOW_STRUCTURED_CONTROL_FLOW.md | 12 +- ...ORKFLOW_SAVED_RESULT_SOURCE_RECHECK_FIX.md | 164 ++++ docs/explanation/release_notes.md | 13 + docs/guides/ask-about-workflow-results.md | 9 +- .../test_workflow_flow_inspection_policy.py | 55 +- ...st_analysis_workflow_source_inheritance.py | 76 +- .../test_document_provenance_origin.py | 42 +- .../test_support/workflow_results.py | 2 + .../test_workflow_collect_publication.py | 12 +- .../test_workflow_data_flow_execution.py | 16 +- .../test_workflow_flow_inspection.py | 53 +- .../test_workflow_history_source_access.py | 36 +- .../test_workflow_loop_native_analysis.py | 18 +- .../test_workflow_loop_reporting.py | 32 +- .../test_workflow_named_result_inputs.py | 42 +- .../test_workflow_repeat_publication.py | 19 +- .../test_workflow_repeat_recovery.py | 101 +-- .../test_workflow_result_followup.py | 23 +- .../test_workflow_result_masking.py | 19 +- .../test_workflow_result_reader.py | 29 +- .../test_workflow_result_routes.py | 31 +- ...workflow_saved_results_container_access.py | 732 ++++++++++++++++++ .../test_workflow_structured_edges.py | 26 +- .../test_workflow_structured_flow.py | 15 +- .../test_workflow_structured_publication.py | 42 +- ui_tests/fixtures/workflow_flow.py | 10 +- ui_tests/test_v2_workflow_flow_authoring.py | 6 +- ui_tests/test_v2_workflow_flow_inspection.py | 25 +- ui_tests/test_v2_workflow_repeat_until.py | 53 +- 67 files changed, 1895 insertions(+), 546 deletions(-) create mode 100644 docs/explanation/fixes/WORKFLOW_SAVED_RESULT_SOURCE_RECHECK_FIX.md create mode 100644 functional_tests/test_workflow_saved_results_container_access.py diff --git a/application/single_app/config.py b/application/single_app/config.py index 40772cb69..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.229" +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 7f8e8a237..aa585bc4b 100644 --- a/application/single_app/content_screening/access.py +++ b/application/single_app/content_screening/access.py @@ -919,6 +919,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}