Harden execution-scoped nested workflow authority

docs/engineering/von_workflow_language_manual.md

@@ -1032,7 +1032,8 @@ Optional inputs:
 - `max_items`
 - `max_transitions`
 - `success_policy` (`all_must_succeed` or `allow_partial`)
-- `stop_on_error` (default `false`; requires sequential execution)
+- `stop_on_error` (default `false`; requires sequential execution and
+  `success_policy=all_must_succeed`)
 
 Runtime semantics:
 
@@ -1042,6 +1043,7 @@ Runtime semantics:
 - per-item results are returned in `iteration_results`;
 - aggregate counts are returned in `for_each_success_count`, `for_each_error_count`, and `for_each_partial_success`;
 - when `stop_on_error=true` in sequential mode, iteration stops after the first failed child and reports the attempted prefix plus the unattempted count;
+- `stop_on_error=true` with `success_policy=allow_partial` fails before child lookup because partial success is incompatible with fail-fast outcome semantics;
 - `stop_on_error=true` with effective concurrency greater than one fails before child execution rather than silently weakening fail-fast semantics;
 - `all_must_succeed` returns failure when any child execution fails;
 - `allow_partial` returns success while preserving structured failure details.

src/backend/workflows/action_registry.py

@@ -13,6 +13,7 @@
 
 import logging
 from dataclasses import dataclass, field
+from threading import RLock
 from typing import Any, Callable, Dict, Mapping, Sequence
 
 from ..security.access_control import override_current_actor
@@ -47,6 +48,65 @@ def normalise_action_outcome(status: str | None) -> str:
     return WORKFLOW_ACTION_OUTCOME_UNKNOWN
 
 
+@dataclass
+class WorkflowExecutionScope:
+    """Ephemeral state shared only by one top-level workflow execution."""
+
+    _nested_workflow_resolution_cache: Dict[
+        tuple[str, str | None, str | None, str | None],
+        Any,
+    ] = field(default_factory=dict, init=False, repr=False, compare=False)
+    _nested_workflow_resolution_locks: Dict[
+        tuple[str, str | None, str | None, str | None],
+        RLock,
+    ] = field(default_factory=dict, init=False, repr=False, compare=False)
+    _cache_lock: RLock = field(
+        default_factory=RLock,
+        init=False,
+        repr=False,
+        compare=False,
+    )
+
+    def get_nested_workflow_resolution(
+        self,
+        key: tuple[str, str | None, str | None, str | None],
+    ) -> Any | None:
+        with self._cache_lock:
+            return self._nested_workflow_resolution_cache.get(key)
+
+    def cache_nested_workflow_resolution(
+        self,
+        key: tuple[str, str | None, str | None, str | None],
+        resolution: Any,
+    ) -> None:
+        with self._cache_lock:
+            self._nested_workflow_resolution_cache[key] = resolution
+
+    def nested_workflow_resolution_lock(
+        self,
+        key: tuple[str, str | None, str | None, str | None],
+    ) -> RLock:
+        """Return the execution-local single-flight lock for one cache key."""
+
+        with self._cache_lock:
+            lock = self._nested_workflow_resolution_locks.get(key)
+            if lock is None:
+                lock = RLock()
+                self._nested_workflow_resolution_locks[key] = lock
+            return lock
+
+    def nested_workflow_resolution_cache_keys(
+        self,
+    ) -> tuple[tuple[str, str | None, str | None, str | None], ...]:
+        with self._cache_lock:
+            return tuple(self._nested_workflow_resolution_cache)
+
+    def close(self) -> None:
+        with self._cache_lock:
+            self._nested_workflow_resolution_cache.clear()
+            self._nested_workflow_resolution_locks.clear()
+
+
 @dataclass(frozen=True)
 class WorkflowEnvironment:
     """Runtime dependencies available to workflow actions."""
@@ -66,13 +126,6 @@ class WorkflowEnvironment:
     user_concept_id: str | None = None
     org_concept_id: str | None = None
     step_callback: Callable[[Mapping[str, Any]], None] | None = None
-    # Ephemeral execution-local support state. This is deliberately excluded
-    # from construction, representation and comparison, and must never be
-    # copied into workflow context/data or durable instance records.
-    _nested_workflow_resolution_cache: Dict[
-        tuple[str, str | None, str | None, str | None],
-        Any,
-    ] = field(default_factory=dict, init=False, repr=False, compare=False)
 
 
 @dataclass(frozen=True)
@@ -91,6 +144,7 @@ class WorkflowActionRequest:
     workflow_id: str | None = None
     workflow_state_id: str | None = None
     workflow_state_metadata: Mapping[str, Any] | None = None
+    execution_scope: WorkflowExecutionScope | None = None
 
 
 @dataclass
@@ -335,6 +389,7 @@ def execute(
         workflow_id: str | None = None,
         workflow_state_id: str | None = None,
         workflow_state_metadata: Mapping[str, Any] | None = None,
+        execution_scope: WorkflowExecutionScope | None = None,
     ) -> WorkflowActionResult:
         action_target_id = str(action_id or "").strip()
         spec = self.resolve_action_spec(action_target_id)
@@ -388,6 +443,7 @@ def execute(
                     if isinstance(workflow_state_metadata, Mapping)
                     else None
                 ),
+                execution_scope=execution_scope,
             )
             # Durable execution carries the authenticated actor in the workflow
             # environment rather than a Flask request context. Bind that actor

src/backend/workflows/durable/control_flow_actions.py

@@ -434,6 +434,7 @@ def _handle(request: WorkflowActionRequest) -> WorkflowActionResult:
                 workflow_id=child_workflow_id,
                 environment=request.environment,
                 fallback_loader=definition_loader,
+                execution_scope=request.execution_scope,
             )
             child_definition = authority_resolution.definition
             if child_definition is None:
@@ -486,6 +487,7 @@ def _handle(request: WorkflowActionRequest) -> WorkflowActionResult:
                 environment=request.environment,
                 data=child_context,
                 trace=child_trace,
+                _execution_scope=request.execution_scope,
             )
 
             branch_result = {
@@ -580,11 +582,36 @@ def _handle(request: WorkflowActionRequest) -> WorkflowActionResult:
                 status="failed",
                 error="for_each_workflow_id_missing",
             )
+        success_policy = _normalise_for_each_success_policy(
+            request.inputs.get("success_policy")
+        )
+        stop_on_error = _coerce_bool_with_default(
+            request.inputs.get("stop_on_error"),
+            default=False,
+        )
+        if (
+            stop_on_error
+            and success_policy != WORKFLOW_FOR_EACH_SUCCESS_POLICY_ALL_MUST_SUCCEED
+        ):
+            return WorkflowActionResult(
+                status="failed",
+                error="for_each_stop_on_error_requires_all_must_succeed",
+                outputs={
+                    "for_each_success_policy": success_policy,
+                    "for_each_stop_on_error": True,
+                    "for_each_item_count": 0,
+                    "iteration_results": [],
+                    "iteration_errors": [],
+                    "successful_results": [],
+                    "invocations": [],
+                },
+            )
 
         authority_resolution = resolve_nested_workflow_definition(
             workflow_id=child_workflow_id,
             environment=request.environment,
             fallback_loader=definition_loader,
+            execution_scope=request.execution_scope,
         )
         child_definition = authority_resolution.definition
         if child_definition is None:
@@ -620,13 +647,6 @@ def _handle(request: WorkflowActionRequest) -> WorkflowActionResult:
         index_context_key = (
             _normalise_text(request.inputs.get("index_context_key")) or "index"
         )
-        success_policy = _normalise_for_each_success_policy(
-            request.inputs.get("success_policy")
-        )
-        stop_on_error = _coerce_bool_with_default(
-            request.inputs.get("stop_on_error"),
-            default=False,
-        )
         include_tool_invocations_in_iteration_results = _coerce_bool_with_default(
             request.inputs.get("include_tool_invocations_in_iteration_results"),
             default=True,
@@ -712,6 +732,7 @@ def _execute_item(index: int, item: Any) -> dict[str, Any]:
                 environment=request.environment,
                 data=child_context,
                 trace=child_trace,
+                _execution_scope=request.execution_scope,
             )
             child_invocations = _derive_child_step_invocations(
                 child_result,

src/backend/workflows/durable/durable_executor.py

@@ -26,6 +26,7 @@
 from ..action_registry import (
     ActionRegistry,
     WorkflowEnvironment,
+    WorkflowExecutionScope,
 )
 from ..trace_model import WorkflowExecutionTrace
 from ..execution_contracts import (
@@ -480,6 +481,8 @@ def run_durable(
         Returns:
             DurableWorkflowResult with execution outcome.
         """
+        execution_scope = WorkflowExecutionScope()
+
         def _retry_store_call(operation_name: str, operation):
             return run_with_transient_mongo_retry(
                 operation,
@@ -1141,6 +1144,7 @@ def _complete_with_gate(final_state: str) -> DurableWorkflowResult:
                     state_support=state_support,
                     context=context,
                     environment=environment,
+                    execution_scope=execution_scope,
                     trace=trace,
                 )
             except ValueError as exc:

src/backend/workflows/durable/nested_workflow_authority.py

@@ -12,14 +12,15 @@
 
 from __future__ import annotations
 
+from contextlib import nullcontext
 from dataclasses import dataclass
 from typing import Any, Callable, Mapping
 
 from ...services.namespace_service import (
     derive_actor_context_from_namespace,
     resolve_canonical_namespace,
 )
-from ..action_registry import WorkflowEnvironment
+from ..action_registry import WorkflowEnvironment, WorkflowExecutionScope
 from ..engine import WorkflowDefinition
 
 
@@ -208,6 +209,7 @@ def resolve_nested_workflow_definition(
     workflow_id: str,
     environment: WorkflowEnvironment,
     fallback_loader: Callable[[str], WorkflowDefinition | None] | None,
+    execution_scope: WorkflowExecutionScope | None = None,
 ) -> NestedWorkflowDefinitionResolution:
     """Resolve a child definition under the parent request's actor authority.
 
@@ -235,71 +237,85 @@ def resolve_nested_workflow_definition(
             actor_context.org_id,
             actor_context.namespace,
         )
-        cached_resolution = environment._nested_workflow_resolution_cache.get(
-            cache_key
+        resolution_lock = (
+            execution_scope.nested_workflow_resolution_lock(cache_key)
+            if execution_scope is not None
+            else nullcontext()
         )
-        if (
-            isinstance(cached_resolution, NestedWorkflowDefinitionResolution)
-            and cached_resolution.success
-        ):
-            return cached_resolution
-
-        try:
-            from .registry_factory import resolve_workflow_definition_from_authority
-
-            authority_resolution = resolve_workflow_definition_from_authority(
-                workflow_id_text,
-                use_current_shared_registry=True,
-                register_authoritative_fallback=True,
-                actor_user_id=actor_context.user_id,
-                actor_org_id=actor_context.org_id,
-            )
-        except Exception as exc:
-            return NestedWorkflowDefinitionResolution(
-                workflow_id=workflow_id_text,
-                definition=None,
-                actor_context=actor_context,
-                authority_source="vontology_actor_authority",
-                error_code=NESTED_WORKFLOW_AUTHORITY_RESOLUTION_FAILED,
-                authority_diagnostics={"exception_type": type(exc).__name__},
+        with resolution_lock:
+            cached_resolution = (
+                execution_scope.get_nested_workflow_resolution(cache_key)
+                if execution_scope is not None
+                else None
             )
-
-        definition = authority_resolution.definition
-        authority_payload = authority_resolution.to_dict()
-        identity_error = _definition_identity_error(
-            requested_workflow_id=workflow_id_text,
-            definition=definition,
-        )
-        if identity_error:
-            return NestedWorkflowDefinitionResolution(
-                workflow_id=workflow_id_text,
-                definition=None,
-                actor_context=actor_context,
-                authority_source="vontology_actor_authority",
-                error_code=identity_error,
-                authority_diagnostics=authority_payload,
+            if (
+                isinstance(cached_resolution, NestedWorkflowDefinitionResolution)
+                and cached_resolution.success
+            ):
+                return cached_resolution
+
+            try:
+                from .registry_factory import (
+                    resolve_workflow_definition_from_authority,
+                )
+
+                authority_resolution = resolve_workflow_definition_from_authority(
+                    workflow_id_text,
+                    use_current_shared_registry=True,
+                    register_authoritative_fallback=True,
+                    actor_user_id=actor_context.user_id,
+                    actor_org_id=actor_context.org_id,
+                )
+            except Exception as exc:
+                return NestedWorkflowDefinitionResolution(
+                    workflow_id=workflow_id_text,
+                    definition=None,
+                    actor_context=actor_context,
+                    authority_source="vontology_actor_authority",
+                    error_code=NESTED_WORKFLOW_AUTHORITY_RESOLUTION_FAILED,
+                    authority_diagnostics={"exception_type": type(exc).__name__},
+                )
+
+            definition = authority_resolution.definition
+            authority_payload = authority_resolution.to_dict()
+            identity_error = _definition_identity_error(
+                requested_workflow_id=workflow_id_text,
+                definition=definition,
             )
-        if definition is None:
-            return NestedWorkflowDefinitionResolution(
+            if identity_error:
+                return NestedWorkflowDefinitionResolution(
+                    workflow_id=workflow_id_text,
+                    definition=None,
+                    actor_context=actor_context,
+                    authority_source="vontology_actor_authority",
+                    error_code=identity_error,
+                    authority_diagnostics=authority_payload,
+                )
+            if definition is None:
+                return NestedWorkflowDefinitionResolution(
+                    workflow_id=workflow_id_text,
+                    definition=None,
+                    actor_context=actor_context,
+                    authority_source="vontology_actor_authority",
+                    error_code=(
+                        _normalise_text(authority_resolution.error_code)
+                        or NESTED_WORKFLOW_DEFINITION_NOT_FOUND
+                    ),
+                    authority_diagnostics=authority_payload,
+                )
+            resolution = NestedWorkflowDefinitionResolution(
                 workflow_id=workflow_id_text,
-                definition=None,
+                definition=definition,
                 actor_context=actor_context,
                 authority_source="vontology_actor_authority",
-                error_code=(
-                    _normalise_text(authority_resolution.error_code)
-                    or NESTED_WORKFLOW_DEFINITION_NOT_FOUND
-                ),
                 authority_diagnostics=authority_payload,
             )
-        resolution = NestedWorkflowDefinitionResolution(
-            workflow_id=workflow_id_text,
-            definition=definition,
-            actor_context=actor_context,
-            authority_source="vontology_actor_authority",
-            authority_diagnostics=authority_payload,
-        )
-        environment._nested_workflow_resolution_cache[cache_key] = resolution
-        return resolution
+            if execution_scope is not None:
+                execution_scope.cache_nested_workflow_resolution(
+                    cache_key,
+                    resolution,
+                )
+            return resolution
 
     definition: WorkflowDefinition | None = None
     if fallback_loader is not None: