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..744835af8 100644 --- a/application/single_app/content_screening/access.py +++ b/application/single_app/content_screening/access.py @@ -56,7 +56,6 @@ HELD_STATES, content_fingerprint, hash_payload, - metadata_fingerprint, normalize_units, public_screening_summary, require_document_available, @@ -321,7 +320,7 @@ def _read_authorized_document( def _require_release_proof(document, container): - """A restored marker or stale metadata write cannot create a clearance.""" + """A restored marker cannot create a clearance; later metadata edits do not revoke one.""" _require_available_metadata(document) marker = document[SCREENING_FIELD] try: @@ -346,7 +345,6 @@ def _require_release_proof(document, container): or not isinstance(publication, Mapping) or publication.get("active_blob") != marker.get("active_blob") or publication.get("content_fingerprint") != marker.get("content_fingerprint") - or publication.get("metadata_fingerprint") != metadata_fingerprint(document) ): raise DocumentHeldError() @@ -919,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/content_screening/contracts.py b/application/single_app/content_screening/contracts.py index 2bfa89610..803e63089 100644 --- a/application/single_app/content_screening/contracts.py +++ b/application/single_app/content_screening/contracts.py @@ -9,6 +9,10 @@ SCHEMA_VERSION = 1 SCREENING_FIELD = "content_screening" +# Server-managed marker for workspace documents that SimpleChat generated. The +# "screening" prefix keeps it out of client payloads in both directions. +SCREENING_EXEMPTION_FIELD = "screening_exemption" +GENERATED_SCREENING_EXEMPTION = {"reason": "generated", "version": 1} SCOPE_TYPES = frozenset({"personal", "group", "public"}) AVAILABLE_STATES = frozenset({"cleared", "approved_with_flags"}) HELD_STATES = frozenset({ @@ -86,6 +90,13 @@ class ScreeningConflictError(ScreeningError): public_message = "The document or review changed. Refresh it before trying again." +class ScreeningNotRequiredError(ScreeningConflictError): + """A generated document is never enrolled, so scan jobs skip it.""" + + code = "screening_not_required" + public_message = "This document was generated by SimpleChat and is not screened." + + class DocumentHeldError(ScreeningError): code = "document_under_review" status_code = 409 @@ -126,6 +137,25 @@ def metadata_fingerprint(document: Mapping[str, Any]) -> str: return hash_payload({name: document.get(name) for name in CONTENT_METADATA_FIELDS}) +def generated_screening_exemption() -> dict: + """Return the marker that server code stores on a generated workspace document.""" + return dict(GENERATED_SCREENING_EXEMPTION) + + +def is_generated_screening_exempt(document: Mapping[str, Any]) -> bool: + """Only the exact server-written marker exempts a document; anything else stays screened.""" + if not isinstance(document, Mapping): + return False + exemption = document.get(SCREENING_EXEMPTION_FIELD) + return ( + isinstance(exemption, Mapping) + and set(exemption) == set(GENERATED_SCREENING_EXEMPTION) + and exemption.get("reason") == GENERATED_SCREENING_EXEMPTION["reason"] + and type(exemption.get("version")) is int + and exemption["version"] == GENERATED_SCREENING_EXEMPTION["version"] + ) + + def normalize_identifier(value: Any, field_name: str = "identifier") -> str: if not isinstance(value, (str, int)) or isinstance(value, bool): raise ScreeningValidationError(f"A valid {field_name} is required.") diff --git a/application/single_app/content_screening/jobs.py b/application/single_app/content_screening/jobs.py index 708090bc0..104e3ca54 100644 --- a/application/single_app/content_screening/jobs.py +++ b/application/single_app/content_screening/jobs.py @@ -24,13 +24,14 @@ ScreeningConfigurationError, ScreeningConflictError, ScreeningError, + ScreeningNotRequiredError, ScreeningPolicyRequiredError, ScreeningValidationError, Subject, content_fingerprint, document_is_available, hash_payload, - metadata_fingerprint, + is_generated_screening_exempt, normalize_identifier, normalize_units, subject_from_document, @@ -397,7 +398,7 @@ def _verify_ingestion_job(job, subject, scan_id, actor_id): def enqueue_document_scan(subject, actor_id, *, scan_id=None, repository=None): """Idempotently enroll an already-authorized ingestion/reprocessing scan. - Internal upload/metadata adapters must stage and authorize the exact scan + Internal upload/reprocessing adapters must stage and authorize the exact scan before calling this helper. It never grants workspace or evidence access, and must not be exposed as an alternative to role-checked create_scan_job. """ @@ -825,7 +826,6 @@ def _publication_matches_document(scan, document): and scan["units_ref"] == marker.get("canonical_ref") and scan.get("content_fingerprint") == marker.get("content_fingerprint") == publication.get("content_fingerprint") and scan.get("policy_fingerprint") == marker.get("policy_fingerprint") == hash_payload(scan.get("policy")) - and publication.get("metadata_fingerprint") == metadata_fingerprint(document) and publication.get("active_blob") == marker.get("active_blob") == active_blob and bool(scan.get("sanitized")) == bool(marker.get("sanitized")) ) @@ -886,6 +886,8 @@ def _hold_document(repository, job, item, owner, *, use_service=False): if expires_at is not None: raise _ScanLeaseBusy(expires_at) document = repository.read_document(subject) + if SCREENING_FIELD not in document and is_generated_screening_exempt(document): + raise ScreeningNotRequiredError() marker = document.get(SCREENING_FIELD, {}) if ( not isinstance(marker, dict) @@ -1162,19 +1164,12 @@ def _reconcile_scan_completion(repository, scan_id): if not _completed_scan(repository, {"scan_id": scan_id, "subject": scan["subject"]}): document = repository.read_document(Subject.from_dict(scan["subject"])) marker = document.get(SCREENING_FIELD) - successor = repository.get_scan(marker["scan_id"]) if isinstance(marker, dict) and marker.get("scan_id") else None inactive_review = ( scan["state"] == "pending_review" and isinstance(marker, dict) and marker.get("scan_id") and marker["scan_id"] != scan_id ) - if not inactive_review and ( - scan["state"] not in AVAILABLE_STATES or not successor - or successor.get("subject") != scan["subject"] or successor.get("trigger") != "metadata" - or (successor.get("previous_marker") or {}).get("scan_id") != scan_id - ): + if not inactive_review: return False - # A crash may leave the completed parent's acknowledgement pending - # after metadata has already queued a new, independently held scan. if repository.count("work_item", filters={ "scan_id": scan_id, "status": ["queued", "running", "retry"], }): diff --git a/application/single_app/content_screening/service.py b/application/single_app/content_screening/service.py index 5d66393d0..7cf8c93e9 100644 --- a/application/single_app/content_screening/service.py +++ b/application/single_app/content_screening/service.py @@ -26,11 +26,13 @@ ScreeningConfigurationError, ScreeningConflictError, ScreeningError, + ScreeningNotRequiredError, ScreeningPolicyRequiredError, ScreeningValidationError, Subject, content_fingerprint, hash_payload, + is_generated_screening_exempt, metadata_fingerprint, normalize_units, subject_from_document, @@ -138,11 +140,16 @@ def get_effective_policy(subject, *, repository=None, require_active=True): def document_requires_screening(document, settings=None, *, repository=None): - """Persisted enrollment always wins over the absence of checks for new uploads.""" + """Persisted enrollment always wins over the absence of checks for new uploads. + + A generated document carries a server-written exemption and is never enrolled. + """ if not isinstance(document, dict): raise ScreeningValidationError("The document metadata is unavailable.") if SCREENING_FIELD in document: return True + if is_generated_screening_exempt(document): + return False settings = _settings(settings) if ( settings.get("enable_content_screening") is not True @@ -229,19 +236,12 @@ def _set_document_state(repository, scan, state, **updates): ) -def _consume_publication_reservation(document, scan_id): - if document.get("group_id") and document.get("generated_artifact_publication_receipt_id"): - # Publication imports ingestion; resolve the receipt owner only at scan - # execution, never during service/configuration bootstrap. - from functions_artifact_publication import consume_artifact_publication_screening_scan - - consume_artifact_publication_screening_scan(document, scan_id) - - def begin_scan(subject, actor_id, *, repository=None, storage=None, job_id=None, scan_id=None, expected_document_etag=None, parent_scan_id=None): repository = _repository(repository) document = repository.read_document(subject) + if SCREENING_FIELD not in document and is_generated_screening_exempt(document): + raise ScreeningNotRequiredError() settings = _settings() if settings.get("enable_content_screening") is not True: raise ScreeningConfigurationError("Enable content screening before rescanning inspected content.") @@ -257,7 +257,6 @@ def begin_scan(subject, actor_id, *, repository=None, storage=None, job_id=None, ): raise ScreeningConflictError() scan_id = scan_id or uuid.uuid4().hex - _consume_publication_reservation(document, scan_id) existing = repository.get_scan(scan_id) if existing: if ( @@ -547,7 +546,6 @@ def inspect_scan(scan_id, units, actor_id, *, repository=None, storage=None, eng units = normalize_units(units) scan = _read_scan(repository, scan_id) subject = Subject.from_dict(scan["subject"]) - _consume_publication_reservation(repository.read_document(subject), scan_id) fingerprint = content_fingerprint(units) if scan.get("content_fingerprint") not in (None, fingerprint): raise ScreeningConflictError("A source change requires a new scan.") @@ -760,7 +758,6 @@ def finalize_publication_checkpoint(scan_id, *, repository=None): or marker.get("canonical_ref") != scan.get("units_ref") or marker.get("active_blob") != publication.get("active_blob") or publication.get("content_fingerprint") != scan.get("content_fingerprint") - or publication.get("metadata_fingerprint") != metadata_fingerprint(document) ): raise ScreeningConflictError() job_id = marker.get("job_id") or scan.get("job_id") @@ -774,7 +771,7 @@ def finalize_publication_checkpoint(scan_id, *, repository=None): published_at=scan.get("published_at") or _timestamp(), postprocess_pending=True, ) current = repository.read_document(subject) - if current.get(SCREENING_FIELD) != marker or metadata_fingerprint(current) != publication["metadata_fingerprint"]: + if current.get(SCREENING_FIELD) != marker: raise ScreeningConflictError() _log("publication_checkpoint_finalized", scan_id=scan_id, document_id=subject.document_id) return scan @@ -852,6 +849,8 @@ def heartbeat(): latest_scan = _read_scan(repository, scan_id) scan = _save_scan(repository, latest_scan, publication={ "active_blob": marker["active_blob"], + # Records the metadata released with this scan. Later metadata + # edits apply directly and do not revoke the release. "metadata_fingerprint": metadata_fingerprint({**current, **updates}), "content_fingerprint": scan["content_fingerprint"], }) @@ -1003,30 +1002,13 @@ def scan_existing_document(subject, actor_id, job_id=None, *, repository=None, s return inspect_scan(scan["id"], units, actor_id, repository=repository, storage=storage) -def queue_metadata_rescan(document, updates, actor_id, *, repository=None): - repository = _repository(repository) - if "file_name" in updates and Path(str(updates["file_name"])).suffix.lower() != Path(document["file_name"]).suffix.lower(): - raise ScreeningValidationError("Renaming a screened file cannot change its source format.") - subject = subject_from_document(document) - marker = document.get(SCREENING_FIELD) or {} +def validate_screened_metadata_update(document, updates): + """Metadata edits apply directly; a screened file keeps the source format it was inspected as.""" if ( - marker.get("state") == "pending_scan" and marker.get("origin_upload") is True - and marker.get("scan_id") and repository.get_scan(marker["scan_id"]) is None + SCREENING_FIELD in document and updates.get("file_name") is not None + and Path(str(updates["file_name"])).suffix.lower() != Path(str(document.get("file_name") or "")).suffix.lower() ): - repository.update_document(subject, updates, etag=document["_etag"]) - return None - scan = begin_scan( - subject, actor_id, repository=repository, - expected_document_etag=document["_etag"], - parent_scan_id=(document.get(SCREENING_FIELD) or {}).get("scan_id"), - ) - scan = _save_scan(repository, scan, trigger="metadata") - current = repository.read_document(subject) - repository.update_document(subject, updates, etag=current["_etag"]) - from content_screening.jobs import enqueue_document_scan - - enqueue_document_scan(subject, actor_id, scan_id=scan["id"], repository=repository) - return scan + raise ScreeningValidationError("Renaming a screened file cannot change its source format.") def reprocess_document(subject, actor_id, extraction_mode, *, repository=None, storage=None): diff --git a/application/single_app/functions_artifact_publication.py b/application/single_app/functions_artifact_publication.py index 5e0666b3b..864a83728 100644 --- a/application/single_app/functions_artifact_publication.py +++ b/application/single_app/functions_artifact_publication.py @@ -22,13 +22,12 @@ cosmos_user_documents_container, ) from content_screening.access import public_document_payload -from content_screening.contracts import SCREENING_FIELD, ScreeningError, hash_payload, subject_from_document +from content_screening.contracts import SCREENING_FIELD, ScreeningError, generated_screening_exemption from functions_appinsights import log_event from functions_artifact_publication_readiness import ( - PUBLICATION_BINDING, PUBLICATION_SCREENING_CONSUMPTION, PUBLICATION_SCREENING_RESERVATION, - SCREENING_BOOTSTRAP_REMEDY, inspect_publication_readiness, publication_binding_matches, - publication_handoff_observed, publication_processing_observation, public_publication_status, - publication_screening_reservation, + PUBLICATION_BINDING, PUBLICATION_SCREENING_HOLD_MESSAGE, inspect_publication_readiness, + publication_binding_matches, publication_handoff_observed, publication_processing_observation, + public_publication_status, ) from functions_collaboration import build_conversation_participation_context from functions_document_provenance import publication_origin_fields @@ -234,136 +233,28 @@ def _destination_document(container, receipt): return document -def _reservation_has_screening_history(document): - # Screening clients are resolved only for an enrolled publication operation. - from content_screening.repository import get_repository +def _publication_destination_admits_approval(document): + """Approval never overrides a screening hold on the destination itself. - repository = get_repository() - subject = subject_from_document(document) - if repository.get_scan(document[SCREENING_FIELD]["scan_id"]) is not None: - return True - return any( - repository.query( - kind, subject.scope_key, filters={"document_id": subject.document_id}, page_size=1, - )["items"] - for kind in ("scan", "review", "finding", "model_window", "event", "audit", "checkpoint", "work_item") - ) - - -def _enroll_screening_reservation(artifact, receipt, container): - """Capture proof only while preparing a new canonical, stored destination.""" - def enroll(current): - document = _destination_document(container, current) - proof = publication_screening_reservation(document) if document else None - if ( - proof is None or current.get("decision") - or PUBLICATION_SCREENING_CONSUMPTION in current - or "approval_queue" in (current.get("stages") or {}) - or _reservation_has_screening_history(document) - ): - raise ValueError(SCREENING_BOOTSTRAP_REMEDY) - if PUBLICATION_SCREENING_RESERVATION in current: - if current[PUBLICATION_SCREENING_RESERVATION] != proof: - raise ValueError(SCREENING_BOOTSTRAP_REMEDY) - return None - current[PUBLICATION_SCREENING_RESERVATION] = proof - return current - - return _receipt_change(artifact, receipt["id"], enroll)[0] - - -def _publication_approval_screening(receipt, document, *, settings=None, operation_id=None, queue_claimed=False): - """The bootstrap result admits only this receipt's approval-to-scan handoff.""" - if SCREENING_FIELD in document: - if (public_document_payload(document).get(SCREENING_FIELD) or {}).get("available") is True: - return "available" - elif PUBLICATION_SCREENING_RESERVATION in receipt or PUBLICATION_SCREENING_CONSUMPTION in receipt: - return None - if settings is None: - # Do not introduce settings/bootstrap initialization into readiness imports. - from functions_settings import get_settings - - settings = get_settings() + Generated destinations carry no screening marker, so approval proceeds. A destination + enrolled before generated content was exempt is approved only once its release is + available; otherwise the requester cancels and requests publication again. + """ if SCREENING_FIELD not in document: - if settings.get("enable_content_screening") is not True: - return "available" - from content_screening.service import document_requires_screening - - return None if document_requires_screening(document, settings) else "available" - if settings.get("enable_content_screening") is not True: - return None - proof = publication_screening_reservation(document) - stages = receipt.get("stages") or {} - if ( - proof is None or receipt.get(PUBLICATION_SCREENING_RESERVATION) != proof - or stages.get("create") != "complete" or stages.get("prepare") != "complete" - or "queue" in stages - or "approval_queue" in stages and not (queue_claimed and stages["approval_queue"] == "started") - ): - return None - consumption = receipt.get(PUBLICATION_SCREENING_CONSUMPTION) - decision = receipt.get("decision") - if PUBLICATION_SCREENING_CONSUMPTION in receipt: - if ( - not isinstance(consumption, dict) or not operation_id - or consumption.get("operation_id") != operation_id - or consumption.get("fingerprint") != proof["fingerprint"] - or "scan_id" in consumption - or decision and ( - decision.get("choice") != "approved" or decision.get("operation_id") != operation_id - ) - ): - return None - elif decision or queue_claimed: - return None - if _reservation_has_screening_history(document): - return None - return "bootstrap" + return True + return (public_document_payload(document).get(SCREENING_FIELD) or {}).get("available") is True -def artifact_publication_approval_available(document, *, settings=None, operation_id=None): - """Read fresh receipt and destination evidence without consuming the latch.""" +def artifact_publication_approval_available(document): + """Read fresh receipt and destination evidence before offering an approval.""" _artifact, receipt, _bound = read_artifact_publication_request(document) container = cosmos_group_documents_container if document.get("group_id") else cosmos_public_documents_container current = _destination_document(container, receipt) if current is None or current.get("_etag") != document.get("_etag"): return False - return _publication_approval_screening( - receipt, current, settings=settings, operation_id=operation_id, - ) is not None + return _publication_destination_admits_approval(current) -def consume_artifact_publication_screening_scan(document, scan_id): - """Write ahead of every scan entry; neither errors nor deletion unlatch it. - - A scan that races initial receipt preparation consumes an unverified - reservation too. Preparation cannot subsequently manufacture positive proof. - """ - artifact, receipt, _bound = read_artifact_publication_request(document) - subject = subject_from_document(document).to_dict() - - def consume(current): - proof = current.get(PUBLICATION_SCREENING_RESERVATION) - if proof is not None and (not isinstance(proof, dict) or proof.get("subject") != subject): - raise ValueError(SCREENING_BOOTSTRAP_REMEDY) - fingerprint = proof["fingerprint"] if proof else hash_payload({ - "subject": subject, "marker": document.get(SCREENING_FIELD), - }) - consumption = current.get(PUBLICATION_SCREENING_CONSUMPTION) - if PUBLICATION_SCREENING_CONSUMPTION in current: - if not isinstance(consumption, dict) or not consumption.get("operation_id"): - raise ValueError(SCREENING_BOOTSTRAP_REMEDY) - if proof and consumption.get("fingerprint") != fingerprint: - raise ValueError(SCREENING_BOOTSTRAP_REMEDY) - if "scan_id" in consumption: - return None - else: - consumption = {"operation_id": f"scan:{scan_id}", "fingerprint": fingerprint} - current[PUBLICATION_SCREENING_CONSUMPTION] = {**consumption, "scan_id": scan_id} - return current - - _receipt_change(artifact, receipt["id"], consume) - def _log_uncertain(stage, exc): log_event( @@ -627,7 +518,7 @@ def reauthorize(): if len(metadata.get(RECEIPTS_FIELD) or {}) >= MAX_ARTIFACT_PUBLICATION_REQUESTS and key not in metadata[RECEIPTS_FIELD]: raise ValueError("This artifact has reached its publication request limit.") reauthorize() - receipt, new_request = _receipt_change(artifact, key, lambda current: None if current else receipt) + receipt, _new_request = _receipt_change(artifact, key, lambda current: None if current else receipt) if receipt.get("completion_policy") != completion_policy or completion_policy is not None and receipt.get("source_receipt") != source_receipt: raise ValueError("The publication request is already bound to different inputs or a different policy.") document = _destination_document(container, receipt) @@ -635,7 +526,6 @@ def reauthorize(): return _completion_response(receipt, document) scope_args = {field: value for field, value in destination.items() if field != "workspace_scope"} scope = destination["workspace_scope"] - created_destination = False if document is None: reauthorize() if document is None and _stage(artifact, receipt, "create"): @@ -657,10 +547,12 @@ def reauthorize(): origin_fields = publication_origin_fields( user_id, artifact, source_receipt=source_receipt, destination=destination, ) + # A published artifact is generated content, so its destination is never screened. create_document( file_name=name, user_id=user_id, document_id=receipt["document_id"], num_file_chunks=0, status="Queued for processing" if scope == "personal" else "Pending approval", source_file_path=source_file_path, allow_deferred_xsd_source=scope != "personal", + screening_exemption=generated_screening_exemption(), **scope_args, **origin_fields, ) except OrchestrationFilePolicyError: @@ -668,7 +560,6 @@ def reauthorize(): except (AzureError, OSError, RuntimeError) as exc: _log_uncertain("create", exc) document = _destination_document(container, receipt) - created_destination = document is not None if document is None: return _publication_response(artifact, receipt, container) _stage(artifact, receipt, "create", complete=True) @@ -716,12 +607,6 @@ def reauthorize(): except (AzureError, OSError, RuntimeError) as exc: _log_uncertain("prepare", exc) document = _destination_document(container, receipt) - if ( - new_request and created_destination and scope == "group" and document - and document.get("generated_artifact_publication_receipt_id") == key - and SCREENING_FIELD in document - ): - receipt = _enroll_screening_reservation(artifact, receipt, container) if not document or document.get("generated_artifact_publication_receipt_id") != key: return _publication_response(artifact, receipt, container) _stage(artifact, receipt, "prepare", complete=True) @@ -1079,14 +964,13 @@ def bind(current): def decide_artifact_publication( user_id, document, choice, *, operation_guard=None, delete_destination=None, decision_link_url=None, - operation_id=None, ): """Use the destination's existing review role, while retaining a durable receipt outcome.""" require_generated_file_publication_allowed() try: return _decide_artifact_publication( user_id, document, choice, operation_guard=operation_guard, - delete_destination=delete_destination, decision_link_url=decision_link_url, operation_id=operation_id, + delete_destination=delete_destination, decision_link_url=decision_link_url, ) except OrchestrationFilePolicyError: raise @@ -1100,19 +984,17 @@ def decide_artifact_publication( def _decide_artifact_publication( user_id, document, choice, *, operation_guard=None, delete_destination=None, decision_link_url=None, - operation_id=None, ): require_generated_file_publication_allowed() with ExitStack() as resources: return _decide_artifact_publication_with_content( user_id, document, choice, resources=resources, operation_guard=operation_guard, - delete_destination=delete_destination, decision_link_url=decision_link_url, operation_id=operation_id, + delete_destination=delete_destination, decision_link_url=decision_link_url, ) def _decide_artifact_publication_with_content( user_id, document, choice, *, resources, operation_guard=None, delete_destination=None, decision_link_url=None, - operation_id=None, ): require_generated_file_publication_allowed() if choice not in {"approved", "rejected", "cancelled"}: @@ -1133,7 +1015,6 @@ def _decide_artifact_publication_with_content( receipt, _ = _receipt_change(artifact, binding["receipt_id"], lambda current: None) if not receipt or not publication_binding_matches(receipt, document): raise ValueError("The publication decision does not match its destination.") - operation_id = operation_id or (receipt.get("decision") or {}).get("operation_id") or str(uuid.uuid4()) destination = receipt["destination"] scope = destination["workspace_scope"] if scope not in {"group", "public"}: @@ -1189,24 +1070,13 @@ def record_decision(current): previous = current.get("decision") if previous and previous.get("choice") != choice: raise ValueError("A different publication decision already committed.") - bootstrap = False - if choice == "approved": - admission = _publication_approval_screening(current, current_document, operation_id=operation_id) - if admission is None: - raise ValueError(SCREENING_BOOTSTRAP_REMEDY) - bootstrap = admission == "bootstrap" - if bootstrap and PUBLICATION_SCREENING_CONSUMPTION not in current: - current[PUBLICATION_SCREENING_CONSUMPTION] = { - "operation_id": operation_id, - "fingerprint": current[PUBLICATION_SCREENING_RESERVATION]["fingerprint"], - } + if choice == "approved" and not _publication_destination_admits_approval(current_document): + raise ValueError(PUBLICATION_SCREENING_HOLD_MESSAGE) if previous: return None current["decision"] = { "choice": choice, "actor_user_id": user_id, "decided_at": datetime.now(timezone.utc).isoformat(), } - if bootstrap: - current["decision"]["operation_id"] = operation_id return current receipt, _ = _receipt_change(artifact, receipt["id"], record_decision) @@ -1227,10 +1097,8 @@ def record_decision(current): _authorize_destination(receipt["actor_user_id"], destination) receipt, _ = _receipt_change(artifact, receipt["id"], lambda saved: None) current = authorize_decision() - if _publication_approval_screening( - receipt, current, operation_id=operation_id, queue_claimed=True, - ) is None: - raise ValueError(SCREENING_BOOTSTRAP_REMEDY) + if not _publication_destination_admits_approval(current): + raise ValueError(PUBLICATION_SCREENING_HOLD_MESSAGE) queue_generated_document_processing( document_id=receipt["document_id"], owner_user_id=receipt["actor_user_id"], normalized_file_name=receipt["file_name"], file_content_bytes=source_bytes, **scope_args, diff --git a/application/single_app/functions_artifact_publication_readiness.py b/application/single_app/functions_artifact_publication_readiness.py index ae53b498f..a3c57af56 100644 --- a/application/single_app/functions_artifact_publication_readiness.py +++ b/application/single_app/functions_artifact_publication_readiness.py @@ -9,21 +9,17 @@ from azure.core import MatchConditions from azure.cosmos.exceptions import CosmosHttpResponseError -from content_screening.access import PROVENANCE_FIELD, assert_document_available +from content_screening.access import assert_document_available from content_screening.contracts import ( AVAILABLE_STATES, HELD_STATES, SCREENING_FIELD, DocumentHeldError, - hash_payload, subject_from_document, ) PUBLICATION_BINDING = "generated_artifact_publication_binding" PUBLICATION_PROCESSING = "generated_artifact_publication_processing" -PUBLICATION_SCREENING_RESERVATION = "screening_reservation" -PUBLICATION_SCREENING_CONSUMPTION = "screening_reservation_consumption" -SCREENING_BOOTSTRAP_REMEDY = ( - "This screening reservation cannot start publication. Cancel the pending publication request " - "and request publication again. If approval already committed, reconcile its existing handoff; " - "do not queue another copy." +PUBLICATION_SCREENING_HOLD_MESSAGE = ( + "This publication destination is held by content screening and cannot be approved. " + "Cancel the pending publication request and request publication again." ) PUBLICATION_STATUS_FIELDS = ( "version", "id", "document_id", "document_version", "destination", "completion_policy", @@ -32,58 +28,6 @@ ) -def publication_screening_reservation(document): - """Identify only a pristine origin reservation, never a scan or a clearance. - - The reserved UUID and original marker timestamp distinguish successive - reservations for the same subject. Cosmos resource identity, when present, - also binds the proof to the stored document instance rather than its name. - """ - marker = document.get(SCREENING_FIELD) - if not isinstance(marker, dict) or set(marker) != { - "schema_version", "state", "source_revision", "scan_id", "content_fingerprint", - "finding_count", "review_required", "generation", "origin_upload", "updated_at", - }: - return None - if ( - not document.get("group_id") or document.get("public_workspace_id") - or type(document.get("version")) is not int or document["version"] != 1 - or document.get("revision_family_id") not in (None, document.get("id")) - or not _current_revision(document) - or type(marker.get("schema_version")) is not int or marker["schema_version"] != 1 - or marker.get("state") != "pending_scan" or marker.get("source_revision") != "1" - or type(marker.get("generation")) is not int or marker["generation"] != 1 - or marker.get("origin_upload") is not True or marker.get("review_required") is not False - or type(marker.get("finding_count")) is not int or marker["finding_count"] != 0 - or marker.get("content_fingerprint") is not None - or not isinstance(marker.get("scan_id"), str) or not marker["scan_id"] - or not isinstance(marker.get("updated_at"), str) or not marker["updated_at"] - or PUBLICATION_PROCESSING in document or PROVENANCE_FIELD in document - ): - return None - if any(document.get(field) for field in ( - "blob_path", "archived_blob_path", "original_blob_path", "blob_etag", "blob_content_hash", - "source_ref", "canonical_ref", "units_ref", "result_ref", "active_manifest_id", - "active_content_manifest", "source_file_available", "enhanced_citations", - "content", "text", "chunks", "pages", "title", "abstract", "keywords", "authors", - "publication_date", "vision_analysis", - )): - return None - if any( - field in document and (type(document[field]) is not int or document[field] != 0) - for field in ("num_chunks", "num_file_chunks", "number_of_pages", "current_file_chunk") - ) or document.get("percentage_complete", 0) != 0: - return None - subject = subject_from_document(document).to_dict() - return { - "schema_version": 1, "subject": subject, - "fingerprint": hash_payload({ - "subject": subject, "marker": marker, - "resource_id": document.get("_rid"), "upload_date": document.get("upload_date"), - }), - } - - def public_publication_status(value): if not isinstance(value, dict) or value.get("version") != 1: raise ValueError("The publication completion status is unavailable.") @@ -267,6 +211,8 @@ def _index_count(receipt, document): def inspect_publication_readiness(receipt, document, *, available_reader=None, index_count=None, check_index=True): """Read native proof; never queue, approve, resume, index, or mutate a document.""" processing = publication_processing_observation(receipt, document) + # Generated destinations carry no screening marker, so screening is already complete + # for them; the destination access check below still applies. observation = {"processing": processing, "screening": "not_required", "index": "pending"} if not publication_binding_matches(receipt, document): return {**observation, "reason_code": "publication_revision_changed"} 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_documents.py b/application/single_app/functions_documents.py index 009623ea8..06ea28076 100644 --- a/application/single_app/functions_documents.py +++ b/application/single_app/functions_documents.py @@ -14,12 +14,14 @@ from azure.core import MatchConditions from azure.core.exceptions import AzureError, ResourceExistsError from content_screening.contracts import ( + SCREENING_EXEMPTION_FIELD, SCREENING_FIELD, DocumentHeldError, ScreeningConflictError, ScreeningError, ScreeningValidationError, document_is_available, + is_generated_screening_exempt, require_document_available, subject_from_document, ) @@ -30,7 +32,6 @@ read_available_document_bytes, ) from content_screening.extraction import ( - CONTENT_METADATA_FIELDS, current_extraction, is_publication, ) @@ -40,8 +41,8 @@ prepare_document_deletion, prepare_document_upload, process_screened_upload, - queue_metadata_rescan, reprocess_document, + validate_screened_metadata_update, ) from config import * from functions_appinsights import log_event @@ -2047,16 +2048,22 @@ def create_document( allow_deferred_xsd_source=False, origin=None, server_tags=None, + screening_exemption=None, ): """Create or version a document record. - ``origin`` and ``server_tags`` are server-only: callers pass them only from server-held - workflow or chat bindings, never from client input. Each version stores only its own origin. + ``origin``, ``server_tags`` and ``screening_exemption`` are server-only: callers pass them only + from server-held workflow, chat, or generated-content bindings, never from client input. Each + version stores only its own origin and exemption. """ if origin is not None: # Validated before any earlier revision is archived, so a malformed origin can never # leave the document family without a current version. origin = validate_origin(origin) + if screening_exemption is not None and not is_generated_screening_exempt( + {SCREENING_EXEMPTION_FIELD: screening_exemption} + ): + raise ScreeningValidationError("The screening exemption is invalid.") current_time = datetime.now(timezone.utc).strftime('%Y-%m-%dT%H:%M:%SZ') is_group = group_id is not None is_public_workspace = public_workspace_id is not None @@ -2364,6 +2371,9 @@ def create_document( "xsd_blob_etag": staged_xsd_source.get("blob_etag"), }) apply_document_provenance(document_metadata, origin=origin, server_tags=server_tags) + if screening_exemption is not None: + # Stored before the marker is chosen, so a generated version is never enrolled. + document_metadata[SCREENING_EXEMPTION_FIELD] = dict(screening_exemption) screening_marker = initial_document_marker(document_metadata) if screening_marker is not None: document_metadata[SCREENING_FIELD] = screening_marker @@ -3532,7 +3542,9 @@ def calculate_processing_percentage(doc_metadata): return max(final_pct, current_pct) def update_document(**kwargs): - if SCREENING_FIELD in kwargs: + # Screening state and every screening_* field, including the generated-content + # exemption, are server-managed and set only when a version is created. + if SCREENING_FIELD in kwargs or any(str(name).startswith("screening_") for name in kwargs): raise ScreeningValidationError("Screening state is managed by the review workflow.") if any(field_name in kwargs for field_name in ORIGIN_FIELD_NAMES): # Origin is written only when a version is created; no update path may set or change it. @@ -3652,29 +3664,13 @@ def update_document(**kwargs): kwargs["status"] = "Content review required" if marker.get("review_required") else "Content screening pending" if kwargs.get("percentage_complete") == 100: kwargs["percentage_complete"] = 99 - content_changed = any( - key in kwargs and kwargs[key] is not None and kwargs[key] != existing_document.get(key) - for key in CONTENT_METADATA_FIELDS - ) if ( - marker is not None and content_changed and capture is None + marker is not None and capture is None and not is_publication(subject_from_document(existing_document), marker.get("scan_id")) ): - metadata_updates = { - key: value for key, value in kwargs.items() - if key not in {"document_id", "user_id", "group_id", "public_workspace_id"} - and value is not None - } - if operation_guard is not None: - operation_guard() - queue_metadata_rescan(existing_document, metadata_updates, user_id) - if strict: - saved_document = cosmos_container.read_item(item=document_id, partition_key=document_id) - projection = sync_document_access_index_for_document_fail_open(saved_document, operation="document_updated") - if not projection.get("success"): - raise DocumentMutationPropagationError("Document saved, but its access projection could not be updated.") - return saved_document - return + # Metadata edits, including model-generated metadata, apply directly and never + # start a new hold. Only the inspected source format of the file stays fixed. + validate_screened_metadata_update(existing_document, kwargs) original_percentage = existing_document.get('percentage_complete', 0) # Store for comparison # 2. Apply updates from kwargs @@ -3753,7 +3749,11 @@ def update_document(**kwargs): # This happens regardless of 'update_occurred' flag because the *intent* from kwargs might trigger it, # even if the main doc update didn't happen (e.g., only percentage changed). # However, it's better to only do this if the relevant fields *actually* changed. - if (update_occurred or strict) and updated_fields_requiring_chunk_sync and marker is None: + # A held screened document has no released chunks; publication rebuilds them. + if ( + (update_occurred or strict) and updated_fields_requiring_chunk_sync + and (marker is None or document_is_available(existing_document)) + ): try: chunks_to_update = get_all_chunks( document_id, @@ -13249,6 +13249,10 @@ def propagate_tags_to_blob_metadata( or (public_workspace_id is not None and doc_item.get("public_workspace_id") != public_workspace_id) ): raise ScreeningConflictError() + if SCREENING_FIELD in doc_item: + # A screened release pins its blob by ETag and content hash. Rewriting the blob's + # metadata would make the released file unreadable, so tags stay in Cosmos and Search. + return if strict and not _has_persisted_blob_reference(doc_item) and not doc_item.get("enhanced_citations"): return storage_account_container_name, blob_path = get_document_blob_storage_info( diff --git a/application/single_app/functions_group_document_management.py b/application/single_app/functions_group_document_management.py index 4b2b1b7a3..df6e26507 100644 --- a/application/single_app/functions_group_document_management.py +++ b/application/single_app/functions_group_document_management.py @@ -285,11 +285,11 @@ def update_group_document_metadata( updated_fields={name: "[updated]" for name in changes} if SCREENING_FIELD in document else changes, file_type=document.get("file_type"), ) - queued = SCREENING_FIELD in document and any(document.get(name) != value for name, value in changes.items()) + # Metadata edits apply directly; they never start a new screening hold. return { - "message": "Metadata saved and queued for content screening." if queued else "Group document metadata updated.", + "message": "Group document metadata updated.", "document_id": document_id, "group_id": group_id, - "updated_fields": list(changes), "status": "queued" if queued else "updated", + "updated_fields": list(changes), "status": "updated", } diff --git a/application/single_app/functions_group_document_publication.py b/application/single_app/functions_group_document_publication.py index bcd1b3848..e533ff562 100644 --- a/application/single_app/functions_group_document_publication.py +++ b/application/single_app/functions_group_document_publication.py @@ -11,8 +11,7 @@ from functions_appinsights import log_event import functions_artifact_publication as publication from functions_artifact_publication_readiness import ( - PUBLICATION_BINDING, PUBLICATION_SCREENING_CONSUMPTION, SCREENING_BOOTSTRAP_REMEDY, - publication_handoff_observed, + PUBLICATION_BINDING, PUBLICATION_SCREENING_HOLD_MESSAGE, publication_handoff_observed, ) from functions_documents import delete_document_revision, select_current_documents from functions_group_document_access import ( @@ -53,31 +52,9 @@ def _has_publication(document): ) -def _bootstrap_retry_operation_id(document, receipt, user_id): - operation = document.get(GROUP_DOCUMENT_COLLABORATION_OPERATION) or {} - consumption = receipt.get(PUBLICATION_SCREENING_CONSUMPTION) or {} - if ( - isinstance(operation, dict) and isinstance(consumption, dict) - and operation.get("schema_version") == 1 and operation.get("action") == "approve_artifact" - and operation.get("phase") in {"executing", "repair"} - and operation.get("actor_user_id") == user_id - and operation.get("actor_group_id") == operation.get("source_group_id") == document.get("group_id") - and operation.get("document_id") == document.get("id") - and operation.get("document_version") == document.get("version") - and isinstance(operation.get("id"), str) and operation["id"] - and isinstance(operation.get("execution_token"), str) and operation["execution_token"] - and consumption.get("operation_id") == operation["id"] and "scan_id" not in consumption - and "approval_queue" not in (receipt.get("stages") or {}) - ): - return operation["id"] - return None - - -def _approval_available(document, settings, *, operation_id=None): +def _approval_available(document): try: - return publication.artifact_publication_approval_available( - document, settings=settings, operation_id=operation_id, - ) + return publication.artifact_publication_approval_available(document) except ScreeningError as error: log_event( "[DOCUMENTS] Publication screening admission could not be established.", @@ -90,7 +67,7 @@ def _approval_available(document, settings, *, operation_id=None): def group_publication_view(document, user_id, group_id, *, context): if not _has_publication(document): return None - _group, role, settings, supported = context + _group, role, _settings, supported = context status = "pending_approval" if group_document_approval_pending(document) else document.get("generated_artifact_promotion_status") if status not in {"pending_approval", "approved", "approval_failed", "rejected", "cancelled"}: status = "unavailable" @@ -99,9 +76,8 @@ def group_publication_view(document, user_id, group_id, *, context): try: _artifact, receipt, _bound = publication.read_artifact_publication_request(document) requester = receipt["actor_user_id"] - retry_id = _bootstrap_retry_operation_id(document, receipt, user_id) operation = document.get(GROUP_DOCUMENT_COLLABORATION_OPERATION) or {} - can_approve = operation.get("phase") != "executing" or retry_id is not None + can_approve = operation.get("phase") != "executing" decision = (receipt.get("decision") or {}).get("choice") if decision == "approved": status = "approved" if (receipt.get("stages") or {}).get("approval_queue") == "complete" else "approval_failed" @@ -115,7 +91,7 @@ def group_publication_view(document, user_id, group_id, *, context): actions.append("reject_artifact") if ( can_approve and is_current_group_document(document) - and _approval_available(document, settings, operation_id=retry_id) + and _approval_available(document) ): actions.append("approve_artifact") if requester == user_id: @@ -126,7 +102,7 @@ def group_publication_view(document, user_id, group_id, *, context): if ( can_approve and role in GROUP_DOCUMENT_MANAGER_ROLES and is_current_group_document(document) - and _approval_available(document, settings, operation_id=retry_id) + and _approval_available(document) ): actions.append("approve_artifact") elif decision == "rejected" and role in GROUP_DOCUMENT_MANAGER_ROLES: @@ -173,7 +149,7 @@ def decide_group_document_publication(user_id, group_id, document_id, action, pa require_payload(payload, {"expected_etag"}, {"expected_etag"}) validate_group_document_id(document_id) - group, role, settings, supported = _context(user_id, group_id) + group, role, _settings, supported = _context(user_id, group_id) if action not in PUBLICATION_ACTIONS or action not in supported: raise GroupDocumentCollaborationError("publication_forbidden", "Your current group role or status cannot make this decision.", 403) document = read_group_document_record(document_id) @@ -196,18 +172,17 @@ def decide_group_document_publication(user_id, group_id, document_id, action, pa raise GroupDocumentCollaborationError("manager_required", "A destination document manager must make this decision.", 403) if choice == "approved" and not is_current_group_document(document): raise GroupDocumentCollaborationError("revision_unavailable", "A historical publication request cannot be approved.", 409) - retry_id = _bootstrap_retry_operation_id(document, receipt, user_id) if choice == "approved" else None - if choice == "approved" and not _approval_available(document, settings, operation_id=retry_id): - raise GroupDocumentCollaborationError("publication_unavailable", SCREENING_BOOTSTRAP_REMEDY, 409) + if choice == "approved" and not _approval_available(document): + raise GroupDocumentCollaborationError("publication_unavailable", PUBLICATION_SCREENING_HOLD_MESSAGE, 409) previous_operation = _operation(document) if ( previous_operation and _unfinished(previous_operation) - and previous_operation.get("phase") in {"executing", "uncertain"} and not retry_id + and previous_operation.get("phase") in {"executing", "uncertain"} ): raise GroupDocumentCollaborationError("operation_busy", "Reconcile the existing document operation before making another decision.", 409) if previous_operation and previous_operation.get("phase") not in {"complete", None} and previous_operation.get("action") != action: raise GroupDocumentCollaborationError("operation_busy", "A different collaboration operation needs reconciliation.", 409) - operation_id = retry_id or str(uuid.uuid4()) + operation_id = str(uuid.uuid4()) execution_token = str(uuid.uuid4()) operation = { "schema_version": 1, "id": operation_id, "document_id": document_id, @@ -297,7 +272,6 @@ def cleanup_guard(document_id=None): publication.decide_artifact_publication( user_id, document, choice, operation_guard=guard, delete_destination=cleanup, decision_link_url=f"/v2/groups/{quote(group_id, safe='')}/documents?document_id={quote(document_id, safe='')}", - operation_id=operation_id, ) except Exception as error: log_event( diff --git a/application/single_app/functions_public_document_management.py b/application/single_app/functions_public_document_management.py index d953ab76a..e410c9940 100644 --- a/application/single_app/functions_public_document_management.py +++ b/application/single_app/functions_public_document_management.py @@ -291,11 +291,11 @@ def update_public_document_metadata( updated_fields={name: "[updated]" for name in changes} if SCREENING_FIELD in document else changes, file_type=document.get("file_type"), ) - queued = SCREENING_FIELD in document and any(document.get(name) != value for name, value in changes.items()) + # Metadata edits apply directly; they never start a new screening hold. return { - "message": "Metadata saved and queued for content screening." if queued else "Public document metadata updated.", + "message": "Public document metadata updated.", "document_id": document_id, "public_workspace_id": workspace_id, - "updated_fields": list(changes), "status": "queued" if queued else "updated", + "updated_fields": list(changes), "status": "updated", } diff --git a/application/single_app/functions_public_document_publication.py b/application/single_app/functions_public_document_publication.py index f828a43e6..74a9eb79d 100644 --- a/application/single_app/functions_public_document_publication.py +++ b/application/single_app/functions_public_document_publication.py @@ -2,11 +2,10 @@ """Scoped adapters for the existing durable artifact-publication lifecycle, targeting an explicit public workspace. -Public workspaces reuse the same shared artifact-publication engine, screening -bootstrap and consume-latch as groups; this module only substitutes the public -workspace scope for the group scope. There is no public share relationship yet -(that arrives with M3D), so this file carries only the generated-artifact -decision surface. +Public workspaces reuse the same shared artifact-publication engine as groups; +this module only substitutes the public workspace scope for the group scope. +There is no public share relationship yet (that arrives with M3D), so this file +carries only the generated-artifact decision surface. """ from urllib.parse import quote @@ -17,8 +16,7 @@ from functions_appinsights import log_event import functions_artifact_publication as publication from functions_artifact_publication_readiness import ( - PUBLICATION_SCREENING_CONSUMPTION, SCREENING_BOOTSTRAP_REMEDY, - publication_handoff_observed, + PUBLICATION_SCREENING_HOLD_MESSAGE, publication_handoff_observed, ) from functions_documents import delete_document_revision, select_current_documents from functions_public_document_access import ( @@ -61,31 +59,9 @@ def _has_publication(document): return public_document_has_publication(document) -def _bootstrap_retry_operation_id(document, receipt, user_id): - operation = document.get(PUBLIC_DOCUMENT_COLLABORATION_OPERATION) or {} - consumption = receipt.get(PUBLICATION_SCREENING_CONSUMPTION) or {} - if ( - isinstance(operation, dict) and isinstance(consumption, dict) - and operation.get("schema_version") == 1 and operation.get("action") == "approve_artifact" - and operation.get("phase") in {"executing", "repair"} - and operation.get("actor_user_id") == user_id - and operation.get("actor_public_workspace_id") == operation.get("source_public_workspace_id") == document.get("public_workspace_id") - and operation.get("document_id") == document.get("id") - and operation.get("document_version") == document.get("version") - and isinstance(operation.get("id"), str) and operation["id"] - and isinstance(operation.get("execution_token"), str) and operation["execution_token"] - and consumption.get("operation_id") == operation["id"] and "scan_id" not in consumption - and "approval_queue" not in (receipt.get("stages") or {}) - ): - return operation["id"] - return None - - -def _approval_available(document, settings, *, operation_id=None): +def _approval_available(document): try: - return publication.artifact_publication_approval_available( - document, settings=settings, operation_id=operation_id, - ) + return publication.artifact_publication_approval_available(document) except ScreeningError as error: log_event( "[DOCUMENTS] Publication screening admission could not be established.", @@ -98,7 +74,7 @@ def _approval_available(document, settings, *, operation_id=None): def public_publication_view(document, user_id, public_workspace_id, *, context): if not _has_publication(document): return None - _workspace, role, settings, supported = context + _workspace, role, _settings, supported = context status = "pending_approval" if public_document_approval_pending(document) else document.get("generated_artifact_promotion_status") if status not in {"pending_approval", "approved", "approval_failed", "rejected", "cancelled"}: status = "unavailable" @@ -107,9 +83,8 @@ def public_publication_view(document, user_id, public_workspace_id, *, context): try: _artifact, receipt, _bound = publication.read_artifact_publication_request(document) requester = receipt["actor_user_id"] - retry_id = _bootstrap_retry_operation_id(document, receipt, user_id) operation = document.get(PUBLIC_DOCUMENT_COLLABORATION_OPERATION) or {} - can_approve = operation.get("phase") != "executing" or retry_id is not None + can_approve = operation.get("phase") != "executing" decision = (receipt.get("decision") or {}).get("choice") if decision == "approved": status = "approved" if (receipt.get("stages") or {}).get("approval_queue") == "complete" else "approval_failed" @@ -123,7 +98,7 @@ def public_publication_view(document, user_id, public_workspace_id, *, context): actions.append("reject_artifact") if ( can_approve and is_current_public_document(document) - and _approval_available(document, settings, operation_id=retry_id) + and _approval_available(document) ): actions.append("approve_artifact") if requester == user_id: @@ -134,7 +109,7 @@ def public_publication_view(document, user_id, public_workspace_id, *, context): if ( can_approve and role in PUBLIC_DOCUMENT_MANAGER_ROLES and is_current_public_document(document) - and _approval_available(document, settings, operation_id=retry_id) + and _approval_available(document) ): actions.append("approve_artifact") elif decision == "rejected" and role in PUBLIC_DOCUMENT_MANAGER_ROLES: @@ -181,7 +156,7 @@ def decide_public_document_publication(user_id, public_workspace_id, document_id require_payload(payload, {"expected_etag"}, {"expected_etag"}) validate_public_document_id(document_id) - workspace, role, settings, supported = _context(user_id, public_workspace_id) + workspace, role, _settings, supported = _context(user_id, public_workspace_id) if action not in PUBLICATION_ACTIONS or action not in supported: raise PublicDocumentCollaborationError("publication_forbidden", "Your current workspace role or status cannot make this decision.", 403) document = read_public_document_record(document_id) @@ -207,18 +182,17 @@ def decide_public_document_publication(user_id, public_workspace_id, document_id raise PublicDocumentCollaborationError("manager_required", "A hosting workspace document manager must make this decision.", 403) if choice == "approved" and not is_current_public_document(document): raise PublicDocumentCollaborationError("revision_unavailable", "A historical publication request cannot be approved.", 409) - retry_id = _bootstrap_retry_operation_id(document, receipt, user_id) if choice == "approved" else None - if choice == "approved" and not _approval_available(document, settings, operation_id=retry_id): - raise PublicDocumentCollaborationError("publication_unavailable", SCREENING_BOOTSTRAP_REMEDY, 409) + if choice == "approved" and not _approval_available(document): + raise PublicDocumentCollaborationError("publication_unavailable", PUBLICATION_SCREENING_HOLD_MESSAGE, 409) previous_operation = _operation(document) if ( previous_operation and _unfinished(previous_operation) - and previous_operation.get("phase") in {"executing", "uncertain"} and not retry_id + and previous_operation.get("phase") in {"executing", "uncertain"} ): raise PublicDocumentCollaborationError("operation_busy", "Reconcile the existing document operation before making another decision.", 409) if previous_operation and previous_operation.get("phase") not in {"complete", None} and previous_operation.get("action") != action: raise PublicDocumentCollaborationError("operation_busy", "A different collaboration operation needs reconciliation.", 409) - operation_id = retry_id or str(uuid.uuid4()) + operation_id = str(uuid.uuid4()) execution_token = str(uuid.uuid4()) operation = { "schema_version": 1, "id": operation_id, "document_id": document_id, @@ -308,7 +282,6 @@ def cleanup_guard(document_id=None): publication.decide_artifact_publication( user_id, document, choice, operation_guard=guard, delete_destination=cleanup, decision_link_url=f"/v2/public/{quote(public_workspace_id, safe='')}/documents?document_id={quote(document_id, safe='')}", - operation_id=operation_id, ) except Exception as error: log_event( 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_simplechat_operations.py b/application/single_app/functions_simplechat_operations.py index b4beb5260..3295f4203 100644 --- a/application/single_app/functions_simplechat_operations.py +++ b/application/single_app/functions_simplechat_operations.py @@ -43,6 +43,7 @@ log_workflow_creation, ) from functions_appinsights import log_event +from content_screening.contracts import generated_screening_exemption from content_screening.service import prepare_document_upload from functions_citation_tracking import rebuild_conversation_used_documents from functions_authentication import ( @@ -2866,6 +2867,7 @@ def _upload_generated_document_for_current_user( num_file_chunks=0, status=initial_status, source_file_path=temp_file_path, + screening_exemption=generated_screening_exemption(), ) update_document( document_id=document_id, @@ -2881,6 +2883,7 @@ def _upload_generated_document_for_current_user( num_file_chunks=0, status=initial_status, source_file_path=temp_file_path, + screening_exemption=generated_screening_exemption(), ) update_document( document_id=document_id, 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 5159bdb22..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 @@ -11229,10 +11233,6 @@ def save_analysis_section(bound_workflow, bound_run, bound_task, section, **_kwa **persistence_options, ) result_summary = workflow_result_summary(manifest, result_ref) - authorize_workflow_task_result_read( - workflow, run_id, task_id, result_ref, manifest=manifest, reader_user_id=actor_id, - **(durable.selectors(attempt=attempt_count) if structured_definition else {}), - ) task_result['workflow_result'] = result_summary task_result['context_budget'] = context_budget _save_workflow_task_run_item( @@ -11247,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_group_documents.py b/application/single_app/route_backend_group_documents.py index fbb635430..2cc3f7dfd 100644 --- a/application/single_app/route_backend_group_documents.py +++ b/application/single_app/route_backend_group_documents.py @@ -633,7 +633,7 @@ def api_scoped_group_document_metadata(group_id, document_id): _group_management_query() changes = validate_metadata_changes(request.get_json(silent=True)) receipt = update_group_document_metadata(get_current_user_id(), group_id, document_id, changes) - return jsonify(receipt), 202 if receipt["status"] == "queued" else 200 + return jsonify(receipt), 200 @bp.route('/api/groups//documents/', methods=['DELETE']) @swagger_route(security=get_auth_security()) diff --git a/application/single_app/route_backend_public_document_management.py b/application/single_app/route_backend_public_document_management.py index d7a721148..909da6b97 100644 --- a/application/single_app/route_backend_public_document_management.py +++ b/application/single_app/route_backend_public_document_management.py @@ -113,7 +113,7 @@ def api_scoped_public_document_metadata(workspace_id, document_id): _public_management_query() changes = validate_metadata_changes(request.get_json(silent=True)) receipt = update_public_document_metadata(get_current_user_id(), workspace_id, document_id, changes) - return jsonify(receipt), 202 if receipt["status"] == "queued" else 200 + return jsonify(receipt), 200 @bp.route('/api/public-workspaces//documents/', methods=['DELETE']) @swagger_route(security=get_auth_security()) diff --git a/application/single_app/route_backend_workflows.py b/application/single_app/route_backend_workflows.py index 24f0180f1..5787da21b 100644 --- a/application/single_app/route_backend_workflows.py +++ b/application/single_app/route_backend_workflows.py @@ -99,7 +99,7 @@ ) from functions_workflow_editor import get_workflow_editor_options from functions_analysis_access import AnalysisResultUnavailable -from functions_workflow_results import authorize_workflow_run_read, authorize_workflow_task_result_read +from functions_workflow_results import authorize_workflow_task_result_read from functions_workflow_result_reader import WorkflowResultUnavailable, read_workflow_result, workflow_result_error_payload from functions_saved_analysis import sanitize_workflow_analysis_history from functions_workflow_runtime import ( @@ -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 @@ -150,26 +150,19 @@ def _normalize_bool(value): return bool(value) -def _workflow_definition_response(workflow, reader_user_id): - """Keep definitions editable without exposing an inaccessible run's cached text.""" +def _workflow_definition_response(workflow): + """Keep definitions editable without exposing a cached preview that no run owns.""" if not workflow.get('last_run_response_preview') and not workflow.get('last_run_error'): return workflow - access = 'available' - run_id = workflow.get('last_run_id') - if not run_id: - access = 'legacy_preview_unbound' - else: - try: - authorize_workflow_run_read(workflow, run_id, reader_user_id=reader_user_id) - except AnalysisResultUnavailable: - access = 'source_unavailable' - if access == 'available': + # A run's preview takes its access from the workflow the caller already read; + # it is not re-checked against the documents the run used. + if workflow.get('last_run_id'): return workflow return { **workflow, 'last_run_response_preview': '', 'last_run_error': '', - 'result_access': access, + 'result_access': 'legacy_preview_unbound', } @@ -216,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: @@ -258,7 +254,7 @@ def _request_workflow_run_cancellation( if run_record and run_record.get('durable_execution') is True: cancelled = cancel_durable_workflow_run(workflow, target_run_id, actor_user_id=requested_by) safe_run = {key: cancelled['run'].get(key) for key in ('id', 'workflow_id', 'status', 'durable_execution', 'runtime')} - return _workflow_definition_response(cancelled['workflow'], requested_by), safe_run + return _workflow_definition_response(cancelled['workflow']), safe_run run_status = _normalize_identifier((run_record or {}).get('status')).lower() if run_status in {'completed', 'completed_partial', 'failed', 'invalid', 'incomplete', 'skipped', 'cancelled', 'canceled'}: @@ -425,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 @@ -438,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): @@ -524,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: @@ -621,7 +621,7 @@ def _queue_workflow_response(workflow, user_id): queued = queue_durable_workflow_run( workflow, actor_user_id=user_id, request_id=data.get('request_id'), ) - queued['workflow'] = _workflow_definition_response(queued['workflow'], user_id) + queued['workflow'] = _workflow_definition_response(queued['workflow']) queued['run'] = {key: queued['run'].get(key) for key in ( 'id', 'workflow_id', 'status', 'success', 'durable_execution', 'started_at', 'completed_at', )} @@ -1005,8 +1005,6 @@ def _resolve_workflow_activity_context(user_id, conversation_id='', workflow_id= raise ValueError('The requested run does not belong to this workflow conversation.') thoughts = [] - if run_record and workflow: - authorize_workflow_run_read(workflow, run_record['id'], reader_user_id=user_id) if run_record and conversation_id and _normalize_identifier(run_record.get('assistant_message_id')): thoughts = get_thoughts_for_message( conversation_id, @@ -1092,8 +1090,6 @@ def _resolve_group_workflow_activity_context(user_id, group_id, conversation_id= ) thoughts = [] - if run_record and workflow: - authorize_workflow_run_read(workflow, run_record['id'], reader_user_id=user_id) if run_record and conversation_id and _normalize_identifier(run_record.get('assistant_message_id')): thoughts = get_thoughts_for_message( conversation_id, @@ -1144,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) @@ -1177,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) @@ -1687,7 +1685,7 @@ def draft_workflow_instructions(): def get_user_workflows(): user_id = get_current_user_id() return jsonify({'workflows': [ - _workflow_definition_response(workflow, user_id) + _workflow_definition_response(workflow) for workflow in get_personal_workflows(user_id) ]}) @@ -1824,7 +1822,7 @@ def save_user_workflow(): trigger_type=workflow.get('trigger_type'), ) - return jsonify({'success': True, 'workflow': _workflow_definition_response(workflow, user_id)}), 201 if is_create else 200 + return jsonify({'success': True, 'workflow': _workflow_definition_response(workflow)}), 201 if is_create else 200 @bp.route('/api/user/workflows/', methods=['DELETE']) @@ -1863,12 +1861,8 @@ def get_user_workflow_runs(workflow_id): if not workflow: return jsonify({'error': 'Workflow not found.'}), 404 + # Runs take their access from the workflow; one run's sources never hide the list. runs = list_personal_workflow_runs(user_id, workflow_id, limit=50) - try: - for run in runs: - authorize_workflow_run_read(workflow, run['id'], reader_user_id=user_id) - except AnalysisResultUnavailable: - return jsonify({'error': 'Run history is unavailable because source access could not be confirmed.'}), 403 return jsonify({ 'workflow_id': workflow_id, 'runs': [ @@ -1966,10 +1960,6 @@ def get_user_workflow_run_items(workflow_id, run_id): if not run_record or _normalize_identifier(run_record.get('workflow_id')) != _normalize_identifier(workflow_id): return jsonify({'error': 'Workflow run not found.'}), 404 - try: - authorize_workflow_run_read(workflow, run_id, reader_user_id=user_id) - except AnalysisResultUnavailable: - return jsonify({'error': 'Task results are unavailable because source access could not be confirmed.'}), 403 return jsonify({ 'workflow_id': workflow_id, 'run_id': run_id, @@ -2116,7 +2106,7 @@ def get_group_workflows_route(): try: group_id, _ = _resolve_group_workflow_request_group(user_id) return jsonify({'workflows': [ - _workflow_definition_response(workflow, user_id) + _workflow_definition_response(workflow) for workflow in get_group_workflows(group_id) ]}) except ValueError as exc: @@ -2267,7 +2257,7 @@ def save_group_workflow_route(): group_id=group_id, ) - return jsonify({'success': True, 'workflow': _workflow_definition_response(workflow, user_id)}), 201 if is_create else 200 + return jsonify({'success': True, 'workflow': _workflow_definition_response(workflow)}), 201 if is_create else 200 @bp.route('/api/group/workflows/', methods=['DELETE']) @@ -2326,12 +2316,8 @@ def get_group_workflow_runs_route(workflow_id): if not workflow: return jsonify({'error': 'Workflow not found.'}), 404 + # Runs take their access from the workflow; one run's sources never hide the list. runs = list_group_workflow_runs(group_id, workflow_id, limit=50) - try: - for run in runs: - authorize_workflow_run_read(workflow, run['id'], reader_user_id=user_id) - except AnalysisResultUnavailable: - return jsonify({'error': 'Run history is unavailable because source access could not be confirmed.'}), 403 return jsonify({ 'workflow_id': workflow_id, 'runs': [ @@ -2463,10 +2449,6 @@ def get_group_workflow_run_items_route(workflow_id, run_id): if not run_record or _normalize_identifier(run_record.get('workflow_id')) != _normalize_identifier(workflow_id): return jsonify({'error': 'Workflow run not found.'}), 404 - try: - authorize_workflow_run_read(workflow, run_id, reader_user_id=user_id) - except AnalysisResultUnavailable: - return jsonify({'error': 'Task results are unavailable because source access could not be confirmed.'}), 403 return jsonify({ 'workflow_id': workflow_id, 'run_id': run_id, diff --git a/application/single_app/templates/approvals.html b/application/single_app/templates/approvals.html index 216ede4da..5faee4767 100644 --- a/application/single_app/templates/approvals.html +++ b/application/single_app/templates/approvals.html @@ -15,12 +15,15 @@

- -
- Workspace knowledge holds use protected evidence and revision-bound decisions. - Open Content review to review or remediate screened documents, - including holds created before new scanning was disabled. + {% if settings.enable_content_screening %} + +
+ To review uploaded documents that content screening is holding, open + Content review.
+ {% endif %} + +
diff --git a/application/v2_ui/src/components/documents/DocumentExplorer.tsx b/application/v2_ui/src/components/documents/DocumentExplorer.tsx index 857ec68ba..42f74187b 100644 --- a/application/v2_ui/src/components/documents/DocumentExplorer.tsx +++ b/application/v2_ui/src/components/documents/DocumentExplorer.tsx @@ -1140,14 +1140,10 @@ function ScopedDocumentExplorer({ const changes = changedDocumentMetadata(target, draft); beginMutation('Saving metadata', 1); try { - const status = await captured.adapter.editMetadata(captured.targets[0], changes); + await captured.adapter.editMetadata(captured.targets[0], changes); if (!mounted.current) return; setDialog(null); - if (status === 'queued') { - setDocuments((current) => current.filter((document) => documentId(document) !== documentId(target))); - setInspectedId(null); - toast.info('Metadata saved. Screening is queued; the document remains unavailable until released.'); - } else toast.success('Metadata saved.'); + toast.success('Metadata saved.'); await refreshAll(); } catch (saveError) { if (mounted.current) { 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}