diff --git a/docs/engineering/von_workflow_language_manual.md b/docs/engineering/von_workflow_language_manual.md index 19ba83079..a2e2a307a 100644 --- a/docs/engineering/von_workflow_language_manual.md +++ b/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. diff --git a/src/backend/workflows/action_registry.py b/src/backend/workflows/action_registry.py index dbb94a685..ca983fb86 100644 --- a/src/backend/workflows/action_registry.py +++ b/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 diff --git a/src/backend/workflows/durable/control_flow_actions.py b/src/backend/workflows/durable/control_flow_actions.py index 5398584bc..9f934825d 100644 --- a/src/backend/workflows/durable/control_flow_actions.py +++ b/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, diff --git a/src/backend/workflows/durable/durable_executor.py b/src/backend/workflows/durable/durable_executor.py index 902ab3759..a6e44ce2b 100644 --- a/src/backend/workflows/durable/durable_executor.py +++ b/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: diff --git a/src/backend/workflows/durable/nested_workflow_authority.py b/src/backend/workflows/durable/nested_workflow_authority.py index b0c1a8503..0e24f20a5 100644 --- a/src/backend/workflows/durable/nested_workflow_authority.py +++ b/src/backend/workflows/durable/nested_workflow_authority.py @@ -12,6 +12,7 @@ from __future__ import annotations +from contextlib import nullcontext from dataclasses import dataclass from typing import Any, Callable, Mapping @@ -19,7 +20,7 @@ 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: diff --git a/src/backend/workflows/durable/subworkflow_actions.py b/src/backend/workflows/durable/subworkflow_actions.py index 9fe25777b..0935cc7c4 100644 --- a/src/backend/workflows/durable/subworkflow_actions.py +++ b/src/backend/workflows/durable/subworkflow_actions.py @@ -1027,6 +1027,7 @@ def _handle(request: WorkflowActionRequest) -> WorkflowActionResult: workflow_id=child_workflow_id, environment=request.environment, fallback_loader=definition_loader, + execution_scope=request.execution_scope, ) definition = authority_resolution.definition if definition is None: @@ -1099,6 +1100,7 @@ def _handle(request: WorkflowActionRequest) -> WorkflowActionResult: environment=request.environment, data=child_inputs, trace=child_trace, + _execution_scope=request.execution_scope, ) invocation_event: Dict[str, Any] = { diff --git a/src/backend/workflows/engine.py b/src/backend/workflows/engine.py index 8a28b18c0..b5ab9c49a 100644 --- a/src/backend/workflows/engine.py +++ b/src/backend/workflows/engine.py @@ -17,6 +17,7 @@ WORKFLOW_ACTION_OUTCOME_UNKNOWN, WorkflowActionResult, WorkflowEnvironment, + WorkflowExecutionScope, normalise_action_outcome, ) from .metadata_validation import ( @@ -505,6 +506,7 @@ def execute_workflow_step_invocation( resolved_inputs: MutableMapping[str, Any], context: Dict[str, Any], env: WorkflowEnvironment, + execution_scope: WorkflowExecutionScope, trace: WorkflowExecutionTrace | None = None, ) -> WorkflowActionResult: if action.execution_mode == WORKFLOW_STEP_EXECUTION_MODE_LLM: @@ -542,6 +544,7 @@ def execute_workflow_step_invocation( if isinstance(workflow_state_metadata, Mapping) else None ), + execution_scope=execution_scope, ) return execute_llm_step(request) @@ -567,6 +570,7 @@ def execute_workflow_step_invocation( workflow_id=workflow_id, workflow_state_id=workflow_state_id, workflow_state_metadata=workflow_state_metadata, + execution_scope=execution_scope, ) @@ -1906,6 +1910,7 @@ def _execute_action_with_runtime_policies( resolved_inputs: Dict[str, Any], context: Dict[str, Any], environment: WorkflowEnvironment, + execution_scope: WorkflowExecutionScope, trace: WorkflowExecutionTrace | None, approval_gate: Mapping[str, Any] | None, idempotency_policy: Mapping[str, Any] | None, @@ -1960,6 +1965,7 @@ def _execute_action_with_runtime_policies( resolved_inputs=resolved_inputs, context=context, env=environment, + execution_scope=execution_scope, trace=trace, ) if trace is not None: @@ -2142,6 +2148,7 @@ def _execute_state_actions( state_support: _WorkflowStateRuntimeSupport, context: Dict[str, Any], environment: WorkflowEnvironment, + execution_scope: WorkflowExecutionScope, trace: WorkflowExecutionTrace | None, ) -> bool: approval_blocked = False @@ -2188,6 +2195,7 @@ def _execute_state_actions( resolved_inputs=resolved_inputs, context=context, environment=environment, + execution_scope=execution_scope, trace=trace, approval_gate=state_support.approval_gate, idempotency_policy=state_support.idempotency_policy, @@ -2306,6 +2314,30 @@ def run( environment: WorkflowEnvironment, data: Dict[str, Any] | None = None, trace: WorkflowExecutionTrace | None = None, + _execution_scope: WorkflowExecutionScope | None = None, + ) -> WorkflowResult: + owns_execution_scope = _execution_scope is None + execution_scope = _execution_scope or WorkflowExecutionScope() + try: + return self._run_in_execution_scope( + definition, + environment=environment, + data=data, + trace=trace, + execution_scope=execution_scope, + ) + finally: + if owns_execution_scope: + execution_scope.close() + + def _run_in_execution_scope( + self, + definition: WorkflowDefinition, + *, + environment: WorkflowEnvironment, + data: Dict[str, Any] | None, + trace: WorkflowExecutionTrace | None, + execution_scope: WorkflowExecutionScope, ) -> WorkflowResult: context: Dict[str, Any] = data or {} clear_control_signal_context(context) @@ -2378,6 +2410,7 @@ def run( state_support=state_support, context=context, environment=environment, + execution_scope=execution_scope, trace=trace, ) except ValueError as exc: diff --git a/tests/backend/test_control_flow_actions.py b/tests/backend/test_control_flow_actions.py index 9f0a560c0..885ff8b60 100644 --- a/tests/backend/test_control_flow_actions.py +++ b/tests/backend/test_control_flow_actions.py @@ -420,6 +420,37 @@ def _conditionally_fail(request): assert result.outputs.get("for_each_stopped_early") is False +def test_for_each_action_rejects_stop_on_error_with_partial_success() -> None: + registry = ActionRegistry() + + def _unexpected_definition_load(_workflow_id): + raise AssertionError("invalid fail-fast policy must fail before child lookup") + + register_control_flow_actions( + registry, + definition_loader=_unexpected_definition_load, + ) + + result = registry.execute( + WORKFLOW_CONTROL_ACTION_FOR_EACH_ID, + inputs={ + "workflow_id": "#V#child_each_invalid_policy", + "items": ["A", "B"], + "max_concurrency": 1, + "success_policy": "allow_partial", + "stop_on_error": True, + }, + context={}, + env=WorkflowEnvironment(llm_client=None), + ) + + assert result.status == "failed" + assert result.error == "for_each_stop_on_error_requires_all_must_succeed" + assert result.outputs.get("for_each_item_count") == 0 + assert result.outputs.get("for_each_success_policy") == "allow_partial" + assert result.outputs.get("for_each_stop_on_error") is True + + def test_for_each_action_rejects_fail_fast_with_concurrent_execution() -> None: registry = ActionRegistry() seen_items: list[str] = [] diff --git a/tests/backend/test_nested_workflow_actor_authority.py b/tests/backend/test_nested_workflow_actor_authority.py index 04d51ffb6..8bdac47a2 100644 --- a/tests/backend/test_nested_workflow_actor_authority.py +++ b/tests/backend/test_nested_workflow_actor_authority.py @@ -1,5 +1,7 @@ from __future__ import annotations +from concurrent.futures import ThreadPoolExecutor +from threading import Event, Lock from unittest.mock import MagicMock from src.backend.workflows.action_registry import ( @@ -7,6 +9,7 @@ ActionSpec, WorkflowActionResult, WorkflowEnvironment, + WorkflowExecutionScope, ) from src.backend.workflows.durable.control_flow_actions import ( register_control_flow_actions, @@ -25,6 +28,7 @@ from src.backend.workflows.engine import ( WorkflowActionInvocation, WorkflowDefinition, + WorkflowExecutor, WorkflowStateSpec, ) from src.backend.workflows.execution_contracts import ( @@ -365,16 +369,19 @@ def test_nested_resolution_caches_success_for_same_environment_and_scope( authority_resolver, ) environment = _environment(COHORT_ID, TRUSTED_ORG_ID) + execution_scope = WorkflowExecutionScope() first = resolve_nested_workflow_definition( workflow_id=WORKFLOW_ID, environment=environment, fallback_loader=fallback_loader, + execution_scope=execution_scope, ) second = resolve_nested_workflow_definition( workflow_id=WORKFLOW_ID, environment=environment, fallback_loader=fallback_loader, + execution_scope=execution_scope, ) assert first.success is True @@ -386,7 +393,7 @@ def test_nested_resolution_caches_success_for_same_environment_and_scope( actor_user_id=COHORT_ID, actor_org_id=TRUSTED_ORG_ID, ) - assert tuple(environment._nested_workflow_resolution_cache) == ( + assert execution_scope.nested_workflow_resolution_cache_keys() == ( ( WORKFLOW_ID, COHORT_ID, @@ -425,16 +432,19 @@ def _resolve(_workflow_id: str, **kwargs): ) owner_environment = _environment(OWNER_ID, TRUSTED_ORG_ID) cohort_environment = _environment(COHORT_ID, TRUSTED_ORG_ID) + execution_scope = WorkflowExecutionScope() owner = resolve_nested_workflow_definition( workflow_id=WORKFLOW_ID, environment=owner_environment, fallback_loader=None, + execution_scope=execution_scope, ) cohort = resolve_nested_workflow_definition( workflow_id=WORKFLOW_ID, environment=cohort_environment, fallback_loader=None, + execution_scope=execution_scope, ) assert owner.definition is owner_definition @@ -443,9 +453,7 @@ def _resolve(_workflow_id: str, **kwargs): (OWNER_ID, TRUSTED_ORG_ID), (COHORT_ID, TRUSTED_ORG_ID), ] - assert owner_environment._nested_workflow_resolution_cache is not ( - cohort_environment._nested_workflow_resolution_cache - ) + assert len(execution_scope.nested_workflow_resolution_cache_keys()) == 2 def test_nested_resolution_does_not_cache_authority_failure( @@ -466,16 +474,19 @@ def test_nested_resolution_does_not_cache_authority_failure( authority_resolver, ) environment = _environment(COHORT_ID, TRUSTED_ORG_ID) + execution_scope = WorkflowExecutionScope() denied = resolve_nested_workflow_definition( workflow_id=WORKFLOW_ID, environment=environment, fallback_loader=None, + execution_scope=execution_scope, ) accepted = resolve_nested_workflow_definition( workflow_id=WORKFLOW_ID, environment=environment, fallback_loader=None, + execution_scope=execution_scope, ) assert denied.success is False @@ -483,3 +494,135 @@ def test_nested_resolution_does_not_cache_authority_failure( assert accepted.success is True assert accepted.definition is definition assert authority_resolver.call_count == 2 + + +def test_nested_resolution_single_flights_concurrent_same_key( + monkeypatch, +) -> None: + import src.backend.workflows.durable.registry_factory as registry_factory + + definition = _definition("child.cohort") + authority_load_started = Event() + release_authority_load = Event() + second_lock_requested = Event() + authority_call_lock = Lock() + authority_call_count = 0 + + def _resolve(_workflow_id: str, **_kwargs): + nonlocal authority_call_count + with authority_call_lock: + authority_call_count += 1 + authority_load_started.set() + assert release_authority_load.wait(timeout=2) + return _resolution(definition) + + monkeypatch.setattr( + registry_factory, + "resolve_workflow_definition_from_authority", + _resolve, + ) + environment = _environment(COHORT_ID, TRUSTED_ORG_ID) + execution_scope = WorkflowExecutionScope() + original_lock_resolver = execution_scope.nested_workflow_resolution_lock + lock_request_count = 0 + lock_request_count_guard = Lock() + + def _observe_lock_request(key): + nonlocal lock_request_count + with lock_request_count_guard: + lock_request_count += 1 + if lock_request_count == 2: + second_lock_requested.set() + return original_lock_resolver(key) + + monkeypatch.setattr( + execution_scope, + "nested_workflow_resolution_lock", + _observe_lock_request, + ) + + def _load(): + return resolve_nested_workflow_definition( + workflow_id=WORKFLOW_ID, + environment=environment, + fallback_loader=None, + execution_scope=execution_scope, + ) + + with ThreadPoolExecutor(max_workers=2) as executor: + first = executor.submit(_load) + assert authority_load_started.wait(timeout=2) + second = executor.submit(_load) + assert second_lock_requested.wait(timeout=2) + release_authority_load.set() + resolutions = (first.result(timeout=2), second.result(timeout=2)) + + assert authority_call_count == 1 + assert all(resolution.success for resolution in resolutions) + assert resolutions[0].definition is definition + assert resolutions[1].definition is definition + + +def test_reused_environment_rechecks_authority_for_each_top_level_run( + monkeypatch, +) -> None: + import src.backend.workflows.durable.registry_factory as registry_factory + + registry = ActionRegistry() + registry.register( + ActionSpec( + action_id="child.cohort", + handler=lambda _request: WorkflowActionResult( + outputs={"authority_probe": "cohort"} + ), + ) + ) + child_definition = _definition("child.cohort") + authority_resolver = MagicMock( + side_effect=[ + _resolution(child_definition), + _resolution(None, error_code="workflow_concept_not_accessible"), + ] + ) + monkeypatch.setattr( + registry_factory, + "resolve_workflow_definition_from_authority", + authority_resolver, + ) + fallback_loader = MagicMock(return_value=child_definition) + register_subworkflow_actions(registry, definition_loader=fallback_loader) + child_invocation = WorkflowActionInvocation( + action_id=WORKFLOW_SUBWORKFLOW_ACTION_ID, + inputs={ + "workflow_id": WORKFLOW_ID, + "__parent_workflow_id": "#V#parent", + "__parent_state_id": "start", + }, + ) + parent_definition = WorkflowDefinition( + workflow_id="#V#parent", + initial_state="start", + states={ + "start": WorkflowStateSpec( + state_id="start", + # The first top-level run invokes the same child twice and must + # share one actor-authority resolution. The second top-level + # run reuses the environment but must create a fresh scope. + actions=(child_invocation, child_invocation), + terminal=True, + ) + }, + termination_states=("start",), + ) + environment = _environment(COHORT_ID, TRUSTED_ORG_ID) + executor = WorkflowExecutor(registry=registry) + + first = executor.run(parent_definition, environment=environment, data={}) + second = executor.run(parent_definition, environment=environment, data={}) + + assert first.completed is True + assert first.data["result"]["authority_probe"] == "cohort" + assert second.completed is False + assert "workflow_concept_not_accessible" in str(second.error) + assert authority_resolver.call_count == 2 + fallback_loader.assert_not_called()