Fix orchestration runs failing on settings saves and unexplained file failures

application/single_app/config.py

@@ -101,7 +101,7 @@
 EXECUTOR_TYPE = 'thread'
 EXECUTOR_MAX_WORKERS = 30
 SESSION_TYPE = 'filesystem'
-VERSION = "0.261.302"
+VERSION = "0.261.303"
 IS_DEVELOPMENT = is_development_env_enabled()
 
 # Opt-out for deployments where App Service Easy Auth is active but the platform

application/single_app/functions_orchestration_artifacts.py

@@ -4,10 +4,15 @@
 The application owner registers an actor-scoped service factory at bootstrap.
 This module never discovers clients, settings, credentials, or source paths.
 The output row, not file-message metadata, authorizes visibility.
+A file's publication is authorized by the rendering service running its attempt on the
+same task, so a signed-in execution checks the file's sources with that sign-in instead of
+a fresh service that has none (0.261.303).
 """
 
 import re
 import uuid
+from contextlib import contextmanager
+from contextvars import ContextVar
 from copy import deepcopy
 
 from azure.core import MatchConditions
@@ -27,19 +32,47 @@
     parse_time,
     safe_identity,
 )
-from functions_orchestration_result_contracts import ProducerIdentity, ResultContractError, canonical_digest
+from functions_orchestration_result_contracts import (
+    EXTERNAL_SESSION_UNAVAILABLE_REASON, ProducerIdentity, ResultContractError, canonical_digest,
+)
 
 
 ORCHESTRATION_ARTIFACT_KIND = "orchestration_retained_output"
 ORCHESTRATION_ARTIFACT_VERSION = 1
 ORCHESTRATION_ARTIFACT_KEY_PREFIX = "orchestration-output:v1:"
+OUTPUT_SIGN_IN_REQUIRED = "output_sign_in_required"
 _DIGEST = re.compile(r"[a-f0-9]{64}\Z")
 _OUTPUT_ID = re.compile(r"orender_[a-f0-9]{64}\Z")
 _SOURCE_FIELDS = frozenset({
     "version", "kind", "producer", "output_id", "intent_id", "attempt_number",
     "source_digest", "spec_digest",
 })
 _service_factory = None
+# The rendering service running a file attempt on this task. Never shared across threads.
+_attempt_service = ContextVar("orchestration_render_attempt_service", default=None)
+
+
+@contextmanager
+def render_attempt_scope(service):
+    """Authorize this task's file publications with the service that is rendering them."""
+    token = _attempt_service.set(service)
+    try:
+        yield
+    finally:
+        _attempt_service.reset(token)
+
+
+def signed_in_session_required(error):
+    """Whether a refusal came only from having no signed-in session to check current access."""
+    seen = set()
+    current = error
+    while current is not None and id(current) not in seen:
+        seen.add(id(current))
+        for reason in (getattr(current, "code", None), getattr(current, "authority_reason", None)):
+            if type(reason) is str and reason in (EXTERNAL_SESSION_UNAVAILABLE_REASON, OUTPUT_SIGN_IN_REQUIRED):
+                return True
+        current = current.__cause__
+    return False
 
 
 def _external_authority_failure(error, *, include_configuration):
@@ -277,7 +310,16 @@
 def _bound_service(user_id, value):
     binding = validate_orchestration_artifact_binding(value)
     producer = binding["producer"]
-    if user_id != producer["user_id"] or _service_factory is None:
+    if user_id != producer["user_id"]:
+        raise OutputUnavailableError("output_access_denied")
+    attempt_service = _attempt_service.get()
+    if (
+        attempt_service is not None
+        and attempt_service.store.user_id == user_id
+        and attempt_service.store.conversation_id == producer["conversation_id"]
+    ):
+        return attempt_service, binding
+    if _service_factory is None:
         raise OutputUnavailableError("output_access_denied")
     return _conversation_service(user_id, producer["conversation_id"]), binding
 
@@ -311,7 +353,9 @@
         raise OutputStorageError() from exc
     except (PermissionError, LookupError, ValueError, TypeError, RuntimeError) as exc:
         _raise_typed_artifact_failure(exc)
-        raise OutputUnavailableError("output_source_unavailable") from exc
+        raise OutputUnavailableError(
+            OUTPUT_SIGN_IN_REQUIRED if signed_in_session_required(exc) else "output_source_unavailable",
+        ) from exc
     except (AttributeError, OSError) as exc:
         _raise_typed_artifact_failure(exc)
         raise

application/single_app/functions_orchestration_checkpoints.py

@@ -1,7 +1,8 @@
 # functions_orchestration_checkpoints.py
 """Private, immutable step-boundary checkpoints in the run-steps partition.
 
-Version: 0.261.139
+Version: 0.261.303
+Settings fingerprints ignore storage metadata and runtime state in: 0.261.303
 The lifecycle row fences every batch, including uncommitted chunks. It survives
 cleanup, so an old worker cannot recreate payloads after conversation deletion.
 Checkpoints record Gather / Reason / Render state only. A checkpoint written by the
@@ -26,6 +27,7 @@
     ProducerIdentity, ResultContractError, ResultRef, TaskResult, digest as validate_result_digest,
 )
 from functions_orchestration_result_runtime import encode_step_result
+from functions_settings_runtime_state import configuration_settings
 
 
 CHECKPOINT_VERSION = 1
@@ -100,8 +102,13 @@
 
 
 def _execution_settings_fingerprint(settings):
-    """The settings a saved attempt's work was approved and executed under."""
-    return fingerprint(settings)
+    """The settings a saved attempt's work was approved and executed under.
+
+    Storage metadata and the runtime state background tasks write are not policy. They
+    change on every save, so including them failed waiting runs and retries with
+    ``recovery_changed`` whenever anything saved settings.
+    """
+    return fingerprint(configuration_settings(settings) if isinstance(settings, dict) else settings)
 
 
 def _manifest_address(step_id, *, waiting=False, input_only=False):
@@ -252,6 +259,21 @@
     })
 
 
+def _ended_attempt_waits(tasks, run_id):
+    """Steps whose saved result is a wait that an earlier attempt left unfinished.
+
+    A checkpoint saves every result its attempt held, including the pending result of a step
+    that was still waiting. A new attempt runs that step again, so the old wait is not restored:
+    it has no outputs that anything could have read, and only its own attempt could finish it.
+    """
+    if type(run_id) is not str:
+        return set()
+    return {
+        name for name, task in tasks.items()
+        if task.status == 'pending' and not task.outputs and task.producer.run_id != run_id
+    }
+
+
 def restore_context(context, payload):
     state = payload.get('state')
     allowed = set(STATE_FIELDS) | set(OPTIONAL_STATE_FIELDS) | set(DEPENDENCY_STATE_FIELDS)
@@ -274,6 +296,7 @@
             raise CheckpointError('checkpoint_invalid')
         setattr(context, key, deepcopy(value))
     try:
+        ended_waits = set()
         for field, descriptor_type in (('task_results', TaskResult), ('result_aliases', ResultRef)):
             values = state.get(field)
             if type(values) is not dict:
@@ -282,8 +305,12 @@
             current = getattr(context, field)
             if field == 'task_results' and any(name != value.producer.step_id for name, value in restored.items()):
                 raise CheckpointError('checkpoint_invalid')
+            if field == 'task_results':
+                ended_waits = _ended_attempt_waits(restored, getattr(context, 'run_id', None))
             for name, value in restored.items():
-                if field == 'task_results' and name in getattr(context, '_failed_result_step_ids', ()):
+                if field == 'task_results' and (
+                    name in getattr(context, '_failed_result_step_ids', ()) or name in ended_waits
+                ):
                     continue
                 previous = current.get(name)
                 if previous is not None and previous != value:
@@ -308,6 +335,7 @@
                 (task is not None and task.status in ('complete', 'partial'))
                 or step_id in getattr(context, '_completed_result_step_ids', ())
                 or step_id in getattr(context, '_failed_result_step_ids', ())
+                or step_id in ended_waits
             ):
                 continue
             if step_id in context.pending_results and context.pending_results[step_id] != wait:

application/single_app/functions_orchestration_continuation.py

@@ -49,7 +49,7 @@
     lease_fields,
     reconcile_checkpoints,
 )
-from functions_orchestration_rendering import raise_output_read_infrastructure_failure
+from functions_orchestration_rendering import raise_output_read_infrastructure_failure, render_failure_code
 from functions_orchestration_result_contracts import TaskResult
 from functions_orchestration_result_runtime import decode_step_result, validate_task_outputs
 from functions_orchestration_schema import (
@@ -764,7 +764,9 @@ def reconcile_run_outputs(record, *, services, lease):
             pending[step_id] = deepcopy(row.get("wait") or wait)
         else:
             pending.pop(step_id, None)
-        failure = build_failure("step_failed", step_id=step_id, capability_id="render_file") if status == "failed" else None
+        failure = build_failure(
+            render_failure_code(fact.get("error_code")), step_id=step_id, capability_id="render_file",
+        ) if status == "failed" else None
         failures = [value for value in failures if value.get("step_id") != step_id]
         if failure is not None:
             failures.append(failure)

application/single_app/functions_orchestration_output_store.py

@@ -49,11 +49,31 @@
     "output_artifact_missing": "This file is unavailable because its committed artifact is missing.",
     "output_artifact_mismatch": "This file is unavailable because its artifact binding could not be verified.",
     "output_intent_invalid": "This file is unavailable because its artifact binding could not be verified.",
+    "output_sign_in_required": (
+        "This file is unavailable here because the results it uses can only be checked with your sign-in."
+    ),
 }
-# A file a time budget stopped says so instead of the generic failed or cancelled text.
+# A failed file says why it could not be created instead of the generic failed text. A file a
+# time budget stopped names that budget.
 OUTPUT_FAILURE_MESSAGES = {
     "output_step_time_limit": "This file was stopped because its step reached the time limit.",
     "output_deadline_exceeded": "This file was stopped because the run reached its time limit.",
+    "output_sign_in_required": (
+        "This file could not be created because the results it uses can only be checked with your "
+        "sign-in, which wasn't available where the file was prepared. Retry from this chat to create it."
+    ),
+    "output_access_denied": (
+        "This file could not be created because access to its conversation or to the results it uses "
+        "could not be confirmed."
+    ),
+    "output_source_unavailable": (
+        "This file could not be created because a result it uses is missing or no longer readable."
+    ),
+    "output_source_changed": (
+        "This file could not be created because a result it uses no longer matches the retained version."
+    ),
+    "output_screening_hold": "This file could not be created while a source it uses is under review.",
+    "output_capability_disabled": "This file could not be created under the current capability settings.",
 }
 _ID = re.compile(r"[A-Za-z0-9][A-Za-z0-9_.:-]{0,199}\Z")
 _OUTPUT_ID = re.compile(r"orender_[a-f0-9]{64}\Z")
@@ -169,6 +189,7 @@ def public_output(record, *, unavailable_code=None, withhold_details=False):
     if withhold_details or not available or record.get("error_code") in {
         "output_access_denied", "output_screening_hold", "output_deleted",
         "output_conversation_unavailable", "output_source_unavailable", "output_superseded",
+        "output_sign_in_required",
     }:
         descriptor = {}
     projected = {

application/single_app/functions_orchestration_recovery.py

@@ -1,7 +1,7 @@
 # functions_orchestration_recovery.py
 """Execution leases and explicitly requested, checkpoint-only retry attempts.
 
-Version: 0.261.293
+Version: 0.261.303
 Retry publication is one transactional parent CAS + child create. It never
 replans, invokes an adapter, or changes plan-revision lineage.
 Terminal publication preserves an administrator's reply retraction; its probe
@@ -11,6 +11,8 @@
 file admissions, because preparing the retry supersedes the parent's files.
 A retry runs again a step that finished without the chart or diagram the plan asked of it,
 and asks for confirmation when that step can have external effects (0.261.293).
+A wait left unfinished by an attempt that failed or was stopped no longer blocks a retry;
+the retry runs the waiting step again (0.261.303).
 A retry that could only resend requests a service declined is not offered.
 A run from the removed legacy contract is never retried, resumed or continued; only
 conversation deletion still reads it, to remove its saved data.
@@ -480,7 +482,9 @@ def recovery_projection(record):
         reason, message = 'legacy_no_checkpoints', 'This run has no durable checkpoints and cannot resume. Create a new plan.'
     elif _live(record):
         reason, message = 'execution_live', 'This attempt is still running. Wait for it to finish or stop it.'
-    elif any(step.get('status') == 'waiting' for step in steps):
+    elif record.get('status') not in _TERMINAL and any(step.get('status') == 'waiting' for step in steps):
+        # A wait only blocks retry while its attempt can still continue. Once the attempt has
+        # ended, the waiting step and the steps after it run again in the new attempt.
         reason, message = 'result_not_ready', build_failure('result_not_ready')['message']
     elif any(step.get('status') == 'running' for step in steps):
         reason, message = 'result_commit_unconfirmed', build_failure('result_commit_unconfirmed')['message']
@@ -1130,6 +1134,10 @@ def validate_resume(record, context, settings, authorize, *, source_run_id=None,
             saved = source_steps.get(step['step_id']) or {}
             if saved.get('status') == 'waiting':
                 if not allow_waiting:
+                    if source.get('status') in _TERMINAL:
+                        # The attempt that waited has ended, so nothing will finish this wait.
+                        # A new attempt runs the step again instead of reusing it.
+                        continue
                     raise CheckpointError('result_not_ready')
                 source_store = checkpoint_store(source, authorize)
                 if not source_store.has_manifest(step['step_id'], waiting=True):

application/single_app/functions_orchestration_rendering.py

@@ -4,6 +4,8 @@
 No producer, composition model, executor, route, or configuration owner is
 imported. Runtime owners supply current capability/admission checks and private
 transport. Approved deadlines and retry admissions survive worker replacement.
+A file whose sources could only be checked with a signed-in session that wasn't
+available fails as ``output_sign_in_required`` and says so (0.261.303).
 """
 
 import hashlib
@@ -38,17 +40,21 @@
 from functions_generated_export_registry import ASSEMBLED_DOCUMENT_PROFILE, resolve_generated_file_export_format
 from functions_generated_file_exports import build_generated_file_export
 from functions_orchestration_artifacts import (
+    OUTPUT_SIGN_IN_REQUIRED,
     OrchestrationArtifactTransport,
     assert_artifact_matches,
     binding_for_intent,
     committed_artifact_card,
     external_authority_failure,
     orchestration_artifact_file_extensions,
     output_storage_failure,
+    render_attempt_scope,
+    signed_in_session_required,
     validate_orchestration_artifact_binding,
 )
 from functions_orchestration_export_sources import open_orchestration_export_source
 from functions_orchestration_output_store import (
+    OUTPUT_FAILURE_MESSAGES,
     OUTPUT_UNAVAILABLE_MESSAGES,
     OrchestrationOutputStore,
     OutputClaim,
@@ -164,6 +170,8 @@ def _source_visibility_code(error):
             return "output_source_changed"
         if error.code in {"result_producer_unavailable", "result_attempt_mismatch", "result_attempt_stopped"}:
             return "output_source_unavailable"
+        if signed_in_session_required(error):
+            return OUTPUT_SIGN_IN_REQUIRED
         return "output_access_denied"
     if isinstance(error, PermissionError):
         return "output_access_denied"
@@ -252,6 +260,9 @@ def output_failure(exc):
         return authority[1], authority[2]
     if output_storage_failure(exc) is not None:
         return "output_storage_unavailable", True
+    if signed_in_session_required(exc):
+        # Automatic and file-only retries render in the background, which has no sign-in.
+        return OUTPUT_SIGN_IN_REQUIRED, False
     if isinstance(exc, DocumentHeldError):
         return "output_screening_hold", False
     if isinstance(exc, SourceAuthorityUnavailableError):
@@ -1019,7 +1030,10 @@ def render_attempt(self, output_id, *, claim=None, worker_id=None, stop=None, ob
         attempt = _RenderAttempt(self, stop=stop)
         outcome = failure = None
         try:
-            outcome = self._render_attempt(output_id, claim, worker_id, attempt)
+            # Publishing the file rechecks its sources with this service's identity, not a
+            # fresh service's, so a signed-in execution can publish what it rendered.
+            with render_attempt_scope(self):
+                outcome = self._render_attempt(output_id, claim, worker_id, attempt)
             return outcome
         except Exception as exc:
             failure = exc
@@ -1273,6 +1287,11 @@ def check():
             yield stream
 
 
+def render_failure_code(error_code):
+    """The step failure a failed file reports when it carries a reason of its own."""
+    return "file_sign_in_required" if error_code == OUTPUT_SIGN_IN_REQUIRED else "step_failed"
+
+
 def _render_step_result(output, record, *, build_step_result, build_failure):
     if output["state"] == "completed":
         result = build_step_result(
@@ -1288,7 +1307,10 @@ def _render_step_result(output, record, *, build_step_result, build_failure):
         result = build_step_result(status="cancelled", summary=output["message"])
     else:
         # A file its step's time budget stopped reports that budget, with its admin hint.
-        failure_code = "step_timeout" if output.get("error_code") == OutputStepTimeLimitError.code else "step_failed"
+        failure_code = (
+            "step_timeout" if output.get("error_code") == OutputStepTimeLimitError.code
+            else render_failure_code(output.get("error_code"))
+        )
         result = build_step_result(
             status="failed", summary=output["message"],
             failure=build_failure(failure_code), error=output["message"],
@@ -1496,9 +1518,13 @@ def execute_render_file(
             result["outputs"] = [] if uncertain_authority or output["state"] == "completed" else [output]
             result["output_error"] = {"code": code, "retryable": retryable}
             return result
+        message = (
+            OUTPUT_FAILURE_MESSAGES[code] if code == OUTPUT_SIGN_IN_REQUIRED
+            else "This file could not be created."
+        )
         result = build_step_result(
-            status="failed", summary="This file could not be created.",
-            failure=build_failure("step_failed"), error="This file could not be created.",
+            status="failed", summary=message,
+            failure=build_failure(render_failure_code(code)), error=message,
         )
         result["output_error"] = {"code": code, "retryable": retryable}
         return result