Stop re-checking workflow run sources in run history and after a task saves

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

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

application/single_app/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 @@
     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 @@
     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.")

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"],
         }):

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 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 _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 @@
     ):
         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 @@
     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 @@
         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 @@
         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 @@
         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 @@
     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):