Saved chat, Analyze and orchestration results take their container's access (screening layer 3b)

application/single_app/config.py

@@ -101,7 +101,7 @@
 EXECUTOR_TYPE = 'thread'
 EXECUTOR_MAX_WORKERS = 30
 SESSION_TYPE = 'filesystem'
-VERSION = "0.261.231"
+VERSION = "0.261.232"
 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

@@ -14,10 +14,13 @@
 An injected ``units_reader(reference, subject)`` supplies canonical units.
 
 ``assert_evidence_available(evidence, user_id=None, *, metadata_reader=None,
-cached=False, strict_errors=False)`` checks known provenance on every reuse. Set ``cached=True`` for
-historical results: an old unversioned context cannot borrow a new clearance.
+cached=False, strict_errors=False)`` checks known provenance when evidence is read
+as an input. Set ``cached=True`` for evidence carried within one operation: an old
+unversioned context cannot borrow a new clearance. Generated results and stored
+conversation history are never passed here; they take their access from their
+container (the conversation, workflow or orchestration run).
 
-Server-owned retained-result operations opt into typed authority failures with
+Server-owned input reads opt into typed authority failures with
 ``strict_errors=True`` or ``strict_source_authority()``. A headless owner that
 catches errors keeps the latter scope around the entire source/model decision.
 
@@ -637,7 +640,7 @@
 
 
 def assert_evidence_available(evidence, user_id=None, *, metadata_reader=None, cached=False, strict_errors=False):
-    """Revalidate known workspace provenance before using saved/model evidence."""
+    """Revalidate known workspace provenance before evidence is used as an input."""
     if type(strict_errors) is not bool:
         raise TypeError("The source authority error policy must be server-owned.")
     if strict_errors:
@@ -842,7 +845,14 @@
 
 
 def public_history_messages(messages, user_id=None):
-    """Withhold unavailable source material without deleting ordinary chat text."""
+    """Return conversation history without rechecking where generated content came from.
+
+    Assistant and tool messages, and the citations stored with them, take their access
+    from the conversation the caller already authorized, so their sources are never
+    checked again. A file or image message backed by a workspace document is an input
+    read: only that message is replaced when the reader can no longer open the document
+    or it is held.
+    """
     # This read boundary already has conversation authorization. Keep the chat
     # adapter out of the document contracts' import/bootstrap dependency chain.
     from functions_chat_content_checks import strip_private_chat_checks
@@ -862,7 +872,6 @@
 
                 message = refresh_checked_message(message)
             refreshed = refresh_workspace_attachment(message, user_id)
-            assert_evidence_available(refreshed, user_id, cached=True)
         except (ScreeningError, LookupError, PermissionError):
             if request_context:
                 flask.g.content_screening_error = previous_error
@@ -882,8 +891,10 @@
             })
             safe_messages.append(safe)
         else:
-            # This dispatcher handles per-file denials itself. Operational failures
-            # must escape rather than become an unrelated document-screening hold.
+            # Generated files keep their container checks (conversation, workflow or
+            # orchestration run) and lose only private storage bindings here; their
+            # source documents are not rechecked. Operational failures must escape
+            # rather than become an unrelated document-screening hold.
             from functions_generated_artifact_sources import sanitize_generated_artifact_history
 
             safe_messages.append(strip_private_chat_checks(

application/single_app/functions_generated_artifact_sources.py

@@ -131,7 +131,11 @@
 
 
 def sanitize_generated_artifact_history(message, user_id):
-    """Reauthorize saved-output cards and strip private bindings from history."""
+    """Apply each generated file's container access and strip private bindings from history.
+
+    A saved-output card or file message stays visible while its conversation, workflow or
+    orchestration run allows it. The documents it was generated from are never rechecked.
+    """
     metadata = message.get("metadata") or {}
     orchestration_history = (
         is_orchestration_artifact_source(metadata.get("generated_artifact_source"))
@@ -154,7 +158,7 @@
         return message
 
     def authorize(conversation_id, message_id):
-        # Reuse the complete conversation/approval/source/screening boundary for saved cards.
+        # Reuse the conversation, approval and publication boundary for saved cards.
         from route_enhanced_citations import _get_authorized_chat_artifact_message
 
         request_context = has_request_context()

application/single_app/functions_orchestration_executor.py

@@ -879,12 +879,18 @@ def _run_dependency_step(
         manifest = _dependency_source_manifest(context, document_ids, settings, cancel_probe)
         if named_sources:
             by_id = {source['document_id']: source for source in manifest}
+            named_by = readers['sources'].reference.producer
+            # Sources named earlier in this attempt must still be the version that step read.
+            # A list reused from an earlier run or attempt only selects documents: they pass
+            # the input check above and are read at their current version.
+            same_attempt = (named_by.run_id, named_by.attempt_index) == (producer.run_id, producer.attempt_index)
+            fields = ('scope', 'scope_id', 'source_version', 'source_revision') if same_attempt else ('scope', 'scope_id')
             for source in named_sources:
                 current = by_id.get(source['document_id'], {})
-                if any(
-                    current.get(field) != source.get(field)
-                    for field in ('scope', 'scope_id', 'source_version', 'source_revision')
-                ) or (source.get('content_sha256') is not None and current.get('content_sha256') != source['content_sha256']):
+                if any(current.get(field) != source.get(field) for field in fields) or (
+                    same_attempt and source.get('content_sha256') is not None
+                    and current.get('content_sha256') != source['content_sha256']
+                ):
                     raise ResultUnavailableError('result_source_snapshot_changed')
         scoped = _dependency_adapter_context(context, manifest, step, input_fingerprint)
         binding_scope = getattr(context, 'step_model_scope', None)

application/single_app/functions_orchestration_native_results.py

@@ -498,7 +498,8 @@ def _sources(step, context, service, *, require_current, require_replay=False):
         ):
             raise NativeOrchestrationBridgeError("native_selection_unsupported")
     snapshots = analysis_source_snapshot(manifest)
-    service.access.authorize_sources(snapshots, require_snapshot=require_current)
+    # The native engine reads the uploaded tabular file itself, so this is an input read.
+    service.access.authorize_input_sources(snapshots, require_snapshot=require_current)
     return deepcopy(manifest), snapshots
 
 

application/single_app/functions_orchestration_output_store.py

@@ -37,7 +37,7 @@
     "waiting", "rendering", "retry_scheduled", "completed", "failed", "cancelled",
 })
 OUTPUT_UNAVAILABLE_MESSAGES = {
-    "output_access_denied": "This file is unavailable because current source access could not be confirmed.",
+    "output_access_denied": "This file is unavailable because access to its conversation could not be confirmed.",
     "output_screening_hold": "This file is unavailable while its source is under review.",
     "output_source_unavailable": "This file is unavailable because its retained source is missing or no longer readable.",
     "output_source_changed": "This file is unavailable because its source no longer matches the retained version.",

application/single_app/functions_orchestration_recovery.py

@@ -997,7 +997,12 @@ def request_cancellation(run_id, user_id, conversation_id, authorize, *, analysi
 
 
 def _validate_payload_sources(payload, context, settings, user_id):
-    """Recheck every retained result a checkpoint references, and its current sources."""
+    """Recheck every retained result a checkpoint references.
+
+    Each result's producer, lineage integrity and external sources are checked again.
+    The documents those results were produced from are not reread; a resumed step that
+    reads documents itself checks them as inputs when it runs.
+    """
     state = payload.get('state') or {}
     if getattr(context, 'plan_contract_version', None) != 2 or state.get('plan_contract_version') != 2:
         raise CheckpointError('checkpoint_invalid')

application/single_app/functions_orchestration_rendering.py

@@ -74,7 +74,7 @@
 
 
 MAX_OUTPUT_BYTES = 500 * 1024 * 1024
-# A render re-proves its claim, run state, capability admission and source access on this
+# A render re-proves its claim, run state, capability admission and result lineage on this
 # cadence rather than on every block it writes. Publication boundaries still check in full.
 RENDER_FULL_CHECK_INTERVAL_SECONDS = 5.0
 _REQUEST_FIELDS = frozenset({"output_format", "profile", "columns", "title", "sheet_name"})
@@ -738,7 +738,7 @@ def _public_records(self, run_id):
                 yield public_output(record), record
 
     def list_public_outputs(self, run_id):
-        """Current per-file visibility; a source denial cannot hide its siblings."""
+        """Current per-file state; one unavailable file cannot hide its siblings."""
         return [projection for projection, _ in self._public_records(run_id)]
 
     def claim_due(self, output_id, *, worker_id=None):

application/single_app/functions_orchestration_results.py

@@ -42,7 +42,7 @@
     validate_record,
     validate_result_role,
 )
-from functions_orchestration_source_access import authorize_orchestration_sources
+from functions_orchestration_source_access import verify_orchestration_input_sources
 from functions_workflow_collections import (
     CollectionWriteBudget,
     RecordTreeWriter,
@@ -90,14 +90,16 @@ def __post_init__(self):
 
 
 class OrchestrationResultAccess:
-    """Server-owned owner/run readers and authorized, screening-aware source I/O.
+    """Server-owned owner/run readers, plus screening-aware source I/O for input reads.
 
     ``read_conversation(conversation_id)`` and ``read_run(run_id)`` return current
-    server records, not request payloads. ``source_resolver(ids, **scope)`` has the
-    existing resolve_authorized_source_manifest protocol. ``source_metadata_reader``
-    has content_screening.access's metadata_reader protocol and must enforce the
-    current actor's document/workspace access. Source-free results need neither
-    source callback. Original-owner, same-conversation reuse is the v1 scope.
+    server records, not request payloads. A retained result takes its access from them
+    alone: the documents it was produced from are never rechecked when it is read.
+    ``source_resolver(ids, **scope)`` has the existing resolve_authorized_source_manifest
+    protocol. ``source_metadata_reader`` has content_screening.access's metadata_reader
+    protocol and must enforce the current actor's document/workspace access. Both are
+    used only by ``authorize_input_sources``, when a step reads uploaded documents
+    itself. Original-owner, same-conversation reuse is the v1 scope.
     """
 
     def __init__(
@@ -165,7 +167,11 @@ def authorize_producer(self, producer, *, for_write=False):
         ):
             raise ResultUnavailableError("result_attempt_stopped")
 
-    def authorize_sources(self, sources, *, require_snapshot):
+    def authorize_input_sources(self, sources, *, require_snapshot):
+        """Input reads only: check uploaded documents a step is about to read itself.
+
+        Retained results never call this; their access comes from ``authorize_producer``.
+        """
         if not sources:
             return {"source_count": 0, "source_snapshot_changed": False}
         if self.source_resolver is None or self.source_metadata_reader is None:
@@ -188,7 +194,7 @@ def resolve(document_ids, **scope):
                             raise ResultUnavailableError("result_source_snapshot_changed")
             return fresh
 
-        checked = authorize_orchestration_sources(
+        checked = verify_orchestration_input_sources(
             self.user_id, sources, require_snapshot=require_snapshot, resolver=resolve,
             metadata_reader=self.source_metadata_reader,
         )
@@ -445,17 +451,22 @@ def _load_manifest(self, reference):
         return manifest
 
     def _authorize_lineage(self, producer, lineage, *, for_write=False, force_current=False, active=None, visited=None):
+        """Check the producer chain and lineage integrity, never the source documents.
+
+        A retained result takes its access from its conversation and run. Its recorded
+        sources are provenance: they bound which source and evidence items its outputs
+        may name, but they are not read again. External (agent, action, memory and web)
+        sources keep their own current checks.
+        """
         self.access.authorize_producer(producer, for_write=for_write)
         parents = _lineage(lineage)
         current = force_current or lineage["source_policy"] == "current"
-        checked = self.access.authorize_sources(lineage["sources"], require_snapshot=current)
         source_snapshots = {
             _source_key(source): {canonical_bytes(source)} for source in lineage["sources"]
         }
-        external_changed, external_sources = self.access.authorize_external_sources(
+        changed, external_sources = self.access.authorize_external_sources(
             producer, lineage.get("external_sources", []), require_snapshot=current,
         )
-        changed = checked["source_snapshot_changed"] or external_changed
         active = set() if active is None else active
         visited = {} if visited is None else visited
         for parent in parents:
@@ -653,6 +664,11 @@ def recover_task_result(self, *, producer, input_fingerprint):
         return task
 
     def open_result(self, reference, *, allow_partial=False, require_current_sources=False):
+        """Open a retained result after its producer and lineage checks.
+
+        ``require_current_sources`` applies the current-version rule to external sources
+        only. The documents a result was produced from are provenance and are not reread.
+        """
         if type(reference) is not ResultRef or type(require_current_sources) is not bool:
             raise ResultContractError("result_reference_untrusted")
         reference.completeness.require_readable(allow_partial=allow_partial)

application/single_app/functions_orchestration_source_access.py

@@ -1,10 +1,13 @@
 # functions_orchestration_source_access.py
-"""Strict current-authority source services for retained orchestration results.
+"""Strict current-authority source services for orchestration input reads.
 
-Version: 0.261.127
+Version: 0.261.232
 
-Owners bind these runtime callbacks, not a browser-selectable policy. Legacy
-screening/search callers outside the explicit scope retain their old behavior.
+Owners bind these runtime callbacks, not a browser-selectable policy. They are used
+only when a step reads uploaded documents as inputs. Retained orchestration results
+take their access from the conversation and run that produced them and never call
+back into these services. Legacy screening/search callers outside the explicit scope
+retain their old behavior.
 """
 
 from content_screening.access import (
@@ -48,10 +51,15 @@ def read_orchestration_source_metadata(document_id, user_id, group_id=None, publ
     )
 
 
-def authorize_orchestration_sources(
+def verify_orchestration_input_sources(
     user_id, sources, *, require_snapshot, resolver, metadata_reader,
 ):
-    """Check retained lineage with an explicit server-owned strict authority boundary."""
+    """Input reads only: confirm a step may still read the uploaded documents it selected.
+
+    The reader must still have access, no document may be held, and with
+    ``require_snapshot`` each document must still be the version the step started from.
+    Never call this for a retained result: those take their access from their container.
+    """
     def resolve(document_ids, **scope):
         try:
             return resolver(document_ids, **scope)