From 7406830097a57e2c31a0e7338678c1fb0f705486 Mon Sep 17 00:00:00 2001 From: witbrock Date: Thu, 23 Jul 2026 21:07:53 +0100 Subject: [PATCH] Harden spreadsheet workflow execution --- .../von_workflow_language_manual.md | 3 + ...ate_spreadsheet_programme_workflow_seed.py | 27 +++- .../selected_workflow_handoff_service.py | 23 +++ src/backend/workflows/action_registry.py | 7 + .../workflows/durable/control_flow_actions.py | 49 ++++++ .../durable/nested_workflow_authority.py | 19 ++- ...e_representation_workflow_seed_bundle.json | 34 +++-- .../workflows/workflow_mcp_tool_actions.py | 14 +- tests/backend/test_control_flow_actions.py | 141 ++++++++++++++++++ .../test_nested_workflow_actor_authority.py | 134 +++++++++++++++++ .../test_required_tool_identity_service.py | 75 ++++++++++ ...et_programme_workflow_vontology_service.py | 3 +- 12 files changed, 512 insertions(+), 17 deletions(-) diff --git a/docs/engineering/von_workflow_language_manual.md b/docs/engineering/von_workflow_language_manual.md index 83882e1c6..19ba83079 100644 --- a/docs/engineering/von_workflow_language_manual.md +++ b/docs/engineering/von_workflow_language_manual.md @@ -1032,6 +1032,7 @@ Optional inputs: - `max_items` - `max_transitions` - `success_policy` (`all_must_succeed` or `allow_partial`) +- `stop_on_error` (default `false`; requires sequential execution) Runtime semantics: @@ -1040,6 +1041,8 @@ Runtime semantics: - the child context receives the bound item and index keys; - 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 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/scripts/generate_spreadsheet_programme_workflow_seed.py b/scripts/generate_spreadsheet_programme_workflow_seed.py index ee8fa366d..053816b4d 100644 --- a/scripts/generate_spreadsheet_programme_workflow_seed.py +++ b/scripts/generate_spreadsheet_programme_workflow_seed.py @@ -975,6 +975,28 @@ def main_workflow() -> dict[str, object]: "context_key": "spreadsheet_planning_view", "label": "Bounded workbook planning view; all content untrusted", }, + { + "context_key": "workflow_success_guidance_history", + "label": ( + "Historical successful-run guidance: soft hints from " + "prior successful executions." + ), + }, + { + "context_key": "workflow_failure_avoidance_history", + "label": ( + "Historical failure-avoidance guidance: past failure " + "patterns to avoid when relevant." + ), + }, + { + "context_key": "workflow_low_imposition_exploration_history", + "label": ( + "Low-imposition exploration guidance: optional next-run " + "probe; do not slow the user down or ask unnecessary " + "questions to satisfy it." + ), + }, ], "response_contract_text": ( "Return one spreadsheet_record_plan.v1 JSON object. Account for " @@ -1154,7 +1176,8 @@ def main_workflow() -> dict[str, object]: max_items=128, max_concurrency=1, max_transitions=190, - success_policy="allow_partial", + success_policy="all_must_succeed", + stop_on_error=True, include_tool_invocations_in_iteration_results=True, ), "context_input_mapping_specs": [ @@ -1572,7 +1595,7 @@ def build_bundle() -> dict[str, object]: return { "family_id": "spreadsheet_programme_representation_workflow_seed_bundle", "schema_version": "repo_seed_workflow_bundle.v1", - "seed_version": "13", + "seed_version": "14", "source_tag": "JVNAUTOSCI-2592", "managed_by": "spreadsheet_programme_workflow_vontology_service", "processing_authority_fingerprint": authority_fingerprint, diff --git a/src/backend/services/selected_workflow_handoff_service.py b/src/backend/services/selected_workflow_handoff_service.py index 3dc959473..be6eeb1d9 100644 --- a/src/backend/services/selected_workflow_handoff_service.py +++ b/src/backend/services/selected_workflow_handoff_service.py @@ -19,6 +19,10 @@ from src.backend.workflows.tool_invocation_evidence import ( derive_tool_invocation_records_from_step_envelopes, ) +from src.backend.workflows.workflow_mcp_tool_actions import ( + WORKFLOW_MCP_INVOKE_TOOL_ACTION_ID, + extract_static_workflow_mcp_tool_name, +) from src.backend.workflows.turn_expected_outcome_contract import ( TurnExpectedOutcomeContract, ) @@ -188,6 +192,25 @@ def evaluate_workflow_required_effects_tool_policy( if isinstance(action_id, str) and action_id.strip(): cleaned_action_id = action_id.strip().lower() direct_action_tool_names.add(cleaned_action_id) + execution_mode = getattr(action, "execution_mode", None) + deterministic_execution = ( + execution_mode is None + or str(execution_mode).strip().lower() == "deterministic" + ) + if ( + cleaned_action_id == WORKFLOW_MCP_INVOKE_TOOL_ACTION_ID + and deterministic_execution + and not bool(getattr(action, "is_llm_step", False)) + ): + action_inputs = getattr(action, "inputs", None) + if isinstance(action_inputs, Mapping): + static_tool_name = ( + extract_static_workflow_mcp_tool_name(action_inputs) + ) + if static_tool_name: + direct_action_tool_names.add( + static_tool_name.lower() + ) if ( cleaned_action_id in _REQUIRED_EFFECTS_UNRESTRICTED_TOOL_SURFACE_ACTION_IDS diff --git a/src/backend/workflows/action_registry.py b/src/backend/workflows/action_registry.py index 8acb942bb..dbb94a685 100644 --- a/src/backend/workflows/action_registry.py +++ b/src/backend/workflows/action_registry.py @@ -66,6 +66,13 @@ 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) diff --git a/src/backend/workflows/durable/control_flow_actions.py b/src/backend/workflows/durable/control_flow_actions.py index e0f7c067e..5398584bc 100644 --- a/src/backend/workflows/durable/control_flow_actions.py +++ b/src/backend/workflows/durable/control_flow_actions.py @@ -623,6 +623,10 @@ def _handle(request: WorkflowActionRequest) -> WorkflowActionResult: 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, @@ -634,6 +638,36 @@ def _handle(request: WorkflowActionRequest) -> WorkflowActionResult: _coerce_for_each_max_concurrency(request.inputs.get("max_concurrency")), len(selected_items) or 1, ) + if stop_on_error and max_concurrency > 1: + return WorkflowActionResult( + status="failed", + error="for_each_stop_on_error_requires_sequential_execution", + outputs={ + "items_source": items_source or None, + "for_each_item_count": 0, + "for_each_selected_item_count": len(selected_items), + "for_each_unattempted_count": len(selected_items), + "for_each_item_limit": item_limit, + "for_each_max_concurrency": max_concurrency, + "for_each_success_policy": success_policy, + "for_each_stop_on_error": True, + "for_each_stopped_on_error": False, + "for_each_stopped_early": False, + "for_each_include_tool_invocations_in_iteration_results": ( + include_tool_invocations_in_iteration_results + ), + "for_each_success_count": 0, + "for_each_error_count": 0, + "for_each_partial_success": False, + "for_each_authority_resolution": ( + authority_resolution.to_projection() + ), + "iteration_results": [], + "iteration_errors": [], + "successful_results": [], + "invocations": [], + }, + ) def _execute_item(index: int, item: Any) -> dict[str, Any]: child_context = dict(request.data) @@ -666,6 +700,7 @@ def _execute_item(index: int, item: Any) -> dict[str, Any]: "index_context_key": index_context_key, "item_index": index, "success_policy": success_policy, + "stop_on_error": stop_on_error, "authority_resolution": authority_resolution.to_projection(), }, ) @@ -702,6 +737,8 @@ def _execute_item(index: int, item: Any) -> dict[str, Any]: if max_concurrency <= 1 or len(selected_items) <= 1: for index, item in enumerate(selected_items): indexed_results[index] = _execute_item(index, item) + if stop_on_error and not indexed_results[index]["completed"]: + break else: with ThreadPoolExecutor(max_workers=max_concurrency) as executor: future_to_index = { @@ -748,6 +785,9 @@ def _execute_item(index: int, item: Any) -> dict[str, Any]: success_count = len([item for item in iteration_results if item["completed"]]) error_count = len(iteration_results) - success_count + unattempted_count = len(selected_items) - len(iteration_results) + stopped_on_error = stop_on_error and error_count > 0 + stopped_early = stopped_on_error and unattempted_count > 0 successful_results = [ dict(result_payload) for item in iteration_results @@ -768,9 +808,14 @@ def _execute_item(index: int, item: Any) -> dict[str, Any]: outputs: Dict[str, Any] = { "items_source": items_source or None, "for_each_item_count": len(iteration_results), + "for_each_selected_item_count": len(selected_items), + "for_each_unattempted_count": unattempted_count, "for_each_item_limit": item_limit, "for_each_max_concurrency": max_concurrency, "for_each_success_policy": success_policy, + "for_each_stop_on_error": stop_on_error, + "for_each_stopped_on_error": stopped_on_error, + "for_each_stopped_early": stopped_early, "for_each_include_tool_invocations_in_iteration_results": ( include_tool_invocations_in_iteration_results ), @@ -809,7 +854,11 @@ def _execute_item(index: int, item: Any) -> dict[str, Any]: "max_concurrency": max_concurrency, "success_count": success_count, "error_count": error_count, + "unattempted_count": unattempted_count, "success_policy": success_policy, + "stop_on_error": stop_on_error, + "stopped_on_error": stopped_on_error, + "stopped_early": stopped_early, "include_tool_invocations_in_iteration_results": ( include_tool_invocations_in_iteration_results ), diff --git a/src/backend/workflows/durable/nested_workflow_authority.py b/src/backend/workflows/durable/nested_workflow_authority.py index 61fed37b1..b0c1a8503 100644 --- a/src/backend/workflows/durable/nested_workflow_authority.py +++ b/src/backend/workflows/durable/nested_workflow_authority.py @@ -229,6 +229,21 @@ def resolve_nested_workflow_definition( ) if actor_context.actor_scoped: + cache_key = ( + workflow_id_text, + actor_context.user_id, + actor_context.org_id, + actor_context.namespace, + ) + cached_resolution = environment._nested_workflow_resolution_cache.get( + cache_key + ) + if ( + isinstance(cached_resolution, NestedWorkflowDefinitionResolution) + and cached_resolution.success + ): + return cached_resolution + try: from .registry_factory import resolve_workflow_definition_from_authority @@ -276,13 +291,15 @@ def resolve_nested_workflow_definition( ), authority_diagnostics=authority_payload, ) - return NestedWorkflowDefinitionResolution( + 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 definition: WorkflowDefinition | None = None if fallback_loader is not None: diff --git a/src/backend/workflows/repo_seed_bundles/spreadsheet_programme_representation_workflow_seed_bundle.json b/src/backend/workflows/repo_seed_bundles/spreadsheet_programme_representation_workflow_seed_bundle.json index b368619cd..3f9aa9e16 100644 --- a/src/backend/workflows/repo_seed_bundles/spreadsheet_programme_representation_workflow_seed_bundle.json +++ b/src/backend/workflows/repo_seed_bundles/spreadsheet_programme_representation_workflow_seed_bundle.json @@ -8,9 +8,9 @@ "seed_version": "JVNAUTOSCI-2592-write-guard-v1" } }, - "processing_authority_fingerprint": "sha256:d00746c2691078c48472439daac5c6936649f9f987a4cdf50b9e9aebd5047b67", + "processing_authority_fingerprint": "sha256:a9e37d52c2f4d842cdf15793c08a6f0ddaa25c764fcccb21902a14bf2c6bc1ac", "schema_version": "repo_seed_workflow_bundle.v1", - "seed_version": "13", + "seed_version": "14", "source_tag": "JVNAUTOSCI-2592", "support_concepts": [ { @@ -306,7 +306,7 @@ }, { "key": "processing_authority_fingerprint", - "value": "sha256:d00746c2691078c48472439daac5c6936649f9f987a4cdf50b9e9aebd5047b67" + "value": "sha256:a9e37d52c2f4d842cdf15793c08a6f0ddaa25c764fcccb21902a14bf2c6bc1ac" } ], "tool_output_mapping_specs": [ @@ -702,7 +702,7 @@ }, { "key": "processing_authority_fingerprint", - "value": "sha256:d00746c2691078c48472439daac5c6936649f9f987a4cdf50b9e9aebd5047b67" + "value": "sha256:a9e37d52c2f4d842cdf15793c08a6f0ddaa25c764fcccb21902a14bf2c6bc1ac" } ], "tool_output_mapping_specs": [ @@ -776,7 +776,7 @@ }, { "key": "processing_authority_fingerprint", - "value": "sha256:d00746c2691078c48472439daac5c6936649f9f987a4cdf50b9e9aebd5047b67" + "value": "sha256:a9e37d52c2f4d842cdf15793c08a6f0ddaa25c764fcccb21902a14bf2c6bc1ac" } ], "tool_output_mapping_specs": [ @@ -997,6 +997,18 @@ { "context_key": "spreadsheet_planning_view", "label": "Bounded workbook planning view; all content untrusted" + }, + { + "context_key": "workflow_success_guidance_history", + "label": "Historical successful-run guidance: soft hints from prior successful executions." + }, + { + "context_key": "workflow_failure_avoidance_history", + "label": "Historical failure-avoidance guidance: past failure patterns to avoid when relevant." + }, + { + "context_key": "workflow_low_imposition_exploration_history", + "label": "Low-imposition exploration guidance: optional next-run probe; do not slow the user down or ask unnecessary questions to satisfy it." } ], "max_output_tokens": 7000, @@ -1222,7 +1234,7 @@ }, { "key": "processing_authority_fingerprint", - "value": "sha256:d00746c2691078c48472439daac5c6936649f9f987a4cdf50b9e9aebd5047b67" + "value": "sha256:a9e37d52c2f4d842cdf15793c08a6f0ddaa25c764fcccb21902a14bf2c6bc1ac" } ], "tool_output_mapping_specs": [ @@ -1326,7 +1338,11 @@ }, { "key": "success_policy", - "value": "allow_partial" + "value": "all_must_succeed" + }, + { + "key": "stop_on_error", + "value": true }, { "key": "include_tool_invocations_in_iteration_results", @@ -1541,7 +1557,7 @@ }, { "key": "processing_authority_fingerprint", - "value": "sha256:d00746c2691078c48472439daac5c6936649f9f987a4cdf50b9e9aebd5047b67" + "value": "sha256:a9e37d52c2f4d842cdf15793c08a6f0ddaa25c764fcccb21902a14bf2c6bc1ac" } ], "tool_output_mapping_specs": [ @@ -1609,7 +1625,7 @@ }, { "key": "processing_authority_fingerprint", - "value": "sha256:d00746c2691078c48472439daac5c6936649f9f987a4cdf50b9e9aebd5047b67" + "value": "sha256:a9e37d52c2f4d842cdf15793c08a6f0ddaa25c764fcccb21902a14bf2c6bc1ac" } ], "tool_output_mapping_specs": [ diff --git a/src/backend/workflows/workflow_mcp_tool_actions.py b/src/backend/workflows/workflow_mcp_tool_actions.py index 6460002fe..6a7f2ee86 100644 --- a/src/backend/workflows/workflow_mcp_tool_actions.py +++ b/src/backend/workflows/workflow_mcp_tool_actions.py @@ -40,13 +40,14 @@ WORKFLOW_MCP_INVOKE_TOOL_ACTION_ID = "workflow_mcp.invoke_tool" -_TOOL_NAME_INPUT_KEYS: tuple[str, ...] = ( +WORKFLOW_MCP_TOOL_NAME_INPUT_KEYS: tuple[str, ...] = ( "tool_name", "method_name", "mcp_tool", "mcp_method", "tool", ) +_TOOL_NAME_INPUT_KEYS = WORKFLOW_MCP_TOOL_NAME_INPUT_KEYS _PAYLOAD_INPUT_KEYS: tuple[str, ...] = ( "tool_arguments", "arguments", @@ -90,7 +91,11 @@ def _normalise_string_list(value: Any) -> tuple[str, ...]: return tuple(items) -def _extract_static_tool_name(inputs: Mapping[str, Any]) -> str | None: +def extract_static_workflow_mcp_tool_name( + inputs: Mapping[str, Any], +) -> str | None: + """Return the exact statically bound MCP method, if one is represented.""" + for key in _TOOL_NAME_INPUT_KEYS: value = inputs.get(key) if isinstance(value, str) and value.strip(): @@ -222,7 +227,7 @@ def validate_workflow_mcp_invocation_contract( return [] issues: list[dict[str, Any]] = [] - tool_name = _extract_static_tool_name(action_inputs) + tool_name = extract_static_workflow_mcp_tool_name(action_inputs) allowed_tools = _explicit_allowed_tool_names( inputs=action_inputs, state_metadata=state_metadata, @@ -336,7 +341,7 @@ def _handle_workflow_mcp_invoke_tool( request: WorkflowActionRequest, ) -> WorkflowActionResult: inputs = dict(request.inputs or {}) - requested_tool_name = _extract_static_tool_name(inputs) + requested_tool_name = extract_static_workflow_mcp_tool_name(inputs) if not requested_tool_name: return WorkflowActionResult( status="failed", @@ -578,6 +583,7 @@ def register_workflow_mcp_tool_actions(registry: ActionRegistry) -> None: __all__ = [ "WORKFLOW_MCP_INVOKE_TOOL_ACTION_ID", + "extract_static_workflow_mcp_tool_name", "register_workflow_mcp_tool_actions", "validate_workflow_mcp_invocation_contract", ] diff --git a/tests/backend/test_control_flow_actions.py b/tests/backend/test_control_flow_actions.py index 5035b23c7..9f0a560c0 100644 --- a/tests/backend/test_control_flow_actions.py +++ b/tests/backend/test_control_flow_actions.py @@ -322,6 +322,147 @@ def _conditionally_fail(request): ] +def test_for_each_action_stops_after_first_error_when_requested() -> None: + registry = ActionRegistry() + seen_items: list[str] = [] + + def _conditionally_fail(request): + item = request.data.get("current_item") + seen_items.append(item) + if item == "bad": + return WorkflowActionResult(status="failed", error="child_failed") + return WorkflowActionResult(outputs={"item_value": item}) + + registry.register( + ActionSpec(action_id="child.maybe_fail", handler=_conditionally_fail) + ) + definitions = { + "#V#child_each_fail_fast": _child_definition( + "#V#child_each_fail_fast", + "child.maybe_fail", + ), + } + register_control_flow_actions( + registry, + definition_loader=lambda workflow_id: definitions.get(workflow_id), + ) + + result = registry.execute( + WORKFLOW_CONTROL_ACTION_FOR_EACH_ID, + inputs={ + "workflow_id": "#V#child_each_fail_fast", + "items": ["good", "bad", "not-attempted"], + "max_concurrency": 1, + "success_policy": "all_must_succeed", + "stop_on_error": True, + }, + context={}, + env=WorkflowEnvironment(llm_client=None), + ) + + assert result.status == "failed" + assert result.error == "for_each_item_failed:#V#child_each_fail_fast" + assert seen_items == ["good", "bad"] + assert result.outputs.get("for_each_item_count") == 2 + assert result.outputs.get("for_each_selected_item_count") == 3 + assert result.outputs.get("for_each_unattempted_count") == 1 + assert result.outputs.get("for_each_success_count") == 1 + assert result.outputs.get("for_each_error_count") == 1 + assert result.outputs.get("for_each_stop_on_error") is True + assert result.outputs.get("for_each_stopped_on_error") is True + assert result.outputs.get("for_each_stopped_early") is True + assert [item["index"] for item in result.outputs["iteration_results"]] == [0, 1] + + +def test_for_each_action_default_continues_after_child_error() -> None: + registry = ActionRegistry() + seen_items: list[str] = [] + + def _conditionally_fail(request): + item = request.data.get("current_item") + seen_items.append(item) + if item == "bad": + return WorkflowActionResult(status="failed", error="child_failed") + return WorkflowActionResult(outputs={"item_value": item}) + + registry.register( + ActionSpec(action_id="child.maybe_fail", handler=_conditionally_fail) + ) + definitions = { + "#V#child_each_default": _child_definition( + "#V#child_each_default", + "child.maybe_fail", + ), + } + register_control_flow_actions( + registry, + definition_loader=lambda workflow_id: definitions.get(workflow_id), + ) + + result = registry.execute( + WORKFLOW_CONTROL_ACTION_FOR_EACH_ID, + inputs={ + "workflow_id": "#V#child_each_default", + "items": ["good", "bad", "still-attempted"], + "max_concurrency": 1, + "success_policy": "all_must_succeed", + }, + context={}, + env=WorkflowEnvironment(llm_client=None), + ) + + assert result.status == "failed" + assert seen_items == ["good", "bad", "still-attempted"] + assert result.outputs.get("for_each_item_count") == 3 + assert result.outputs.get("for_each_unattempted_count") == 0 + assert result.outputs.get("for_each_stop_on_error") is False + assert result.outputs.get("for_each_stopped_on_error") is False + assert result.outputs.get("for_each_stopped_early") is False + + +def test_for_each_action_rejects_fail_fast_with_concurrent_execution() -> None: + registry = ActionRegistry() + seen_items: list[str] = [] + + def _emit_item(request): + seen_items.append(request.data.get("current_item")) + return WorkflowActionResult( + outputs={"item_value": request.data.get("current_item")} + ) + + registry.register(ActionSpec(action_id="child.emit_item", handler=_emit_item)) + definitions = { + "#V#child_each_concurrent": _child_definition( + "#V#child_each_concurrent", + "child.emit_item", + ), + } + register_control_flow_actions( + registry, + definition_loader=lambda workflow_id: definitions.get(workflow_id), + ) + + result = registry.execute( + WORKFLOW_CONTROL_ACTION_FOR_EACH_ID, + inputs={ + "workflow_id": "#V#child_each_concurrent", + "items": ["A", "B"], + "max_concurrency": 2, + "stop_on_error": True, + }, + context={}, + env=WorkflowEnvironment(llm_client=None), + ) + + assert result.status == "failed" + assert result.error == "for_each_stop_on_error_requires_sequential_execution" + assert seen_items == [] + assert result.outputs.get("for_each_item_count") == 0 + assert result.outputs.get("for_each_selected_item_count") == 2 + assert result.outputs.get("for_each_unattempted_count") == 2 + assert result.outputs.get("for_each_stopped_on_error") is False + + # --------------------------------------------------------------------------- # context.set action tests (JVNAUTOSCI-1440) # --------------------------------------------------------------------------- diff --git a/tests/backend/test_nested_workflow_actor_authority.py b/tests/backend/test_nested_workflow_actor_authority.py index 931389ec4..04d51ffb6 100644 --- a/tests/backend/test_nested_workflow_actor_authority.py +++ b/tests/backend/test_nested_workflow_actor_authority.py @@ -349,3 +349,137 @@ def test_nested_resolution_rejects_wrong_definition_identity_without_fallback( assert resolution.success is False assert resolution.error_code == NESTED_WORKFLOW_DEFINITION_IDENTITY_MISMATCH fallback_loader.assert_not_called() + + +def test_nested_resolution_caches_success_for_same_environment_and_scope( + monkeypatch, +) -> None: + import src.backend.workflows.durable.registry_factory as registry_factory + + definition = _definition("child.cohort") + authority_resolver = MagicMock(return_value=_resolution(definition)) + fallback_loader = MagicMock(return_value=_definition("child.owner")) + monkeypatch.setattr( + registry_factory, + "resolve_workflow_definition_from_authority", + authority_resolver, + ) + environment = _environment(COHORT_ID, TRUSTED_ORG_ID) + + first = resolve_nested_workflow_definition( + workflow_id=WORKFLOW_ID, + environment=environment, + fallback_loader=fallback_loader, + ) + second = resolve_nested_workflow_definition( + workflow_id=WORKFLOW_ID, + environment=environment, + fallback_loader=fallback_loader, + ) + + assert first.success is True + assert second is first + authority_resolver.assert_called_once_with( + WORKFLOW_ID, + use_current_shared_registry=True, + register_authoritative_fallback=True, + actor_user_id=COHORT_ID, + actor_org_id=TRUSTED_ORG_ID, + ) + assert tuple(environment._nested_workflow_resolution_cache) == ( + ( + WORKFLOW_ID, + COHORT_ID, + TRUSTED_ORG_ID, + first.actor_context.namespace, + ), + ) + fallback_loader.assert_not_called() + + +def test_nested_resolution_cache_is_not_reused_across_actor_scopes( + monkeypatch, +) -> None: + import src.backend.workflows.durable.registry_factory as registry_factory + + owner_definition = _definition("child.owner") + cohort_definition = _definition("child.cohort") + authority_calls: list[tuple[object, object]] = [] + + def _resolve(_workflow_id: str, **kwargs): + actor_scope = ( + kwargs.get("actor_user_id"), + kwargs.get("actor_org_id"), + ) + authority_calls.append(actor_scope) + if actor_scope == (OWNER_ID, TRUSTED_ORG_ID): + return _resolution(owner_definition) + if actor_scope == (COHORT_ID, TRUSTED_ORG_ID): + return _resolution(cohort_definition) + return _resolution(None, error_code="workflow_concept_not_accessible") + + monkeypatch.setattr( + registry_factory, + "resolve_workflow_definition_from_authority", + _resolve, + ) + owner_environment = _environment(OWNER_ID, TRUSTED_ORG_ID) + cohort_environment = _environment(COHORT_ID, TRUSTED_ORG_ID) + + owner = resolve_nested_workflow_definition( + workflow_id=WORKFLOW_ID, + environment=owner_environment, + fallback_loader=None, + ) + cohort = resolve_nested_workflow_definition( + workflow_id=WORKFLOW_ID, + environment=cohort_environment, + fallback_loader=None, + ) + + assert owner.definition is owner_definition + assert cohort.definition is cohort_definition + assert authority_calls == [ + (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 + ) + + +def test_nested_resolution_does_not_cache_authority_failure( + monkeypatch, +) -> None: + import src.backend.workflows.durable.registry_factory as registry_factory + + definition = _definition("child.cohort") + authority_resolver = MagicMock( + side_effect=[ + _resolution(None, error_code="workflow_concept_not_accessible"), + _resolution(definition), + ] + ) + monkeypatch.setattr( + registry_factory, + "resolve_workflow_definition_from_authority", + authority_resolver, + ) + environment = _environment(COHORT_ID, TRUSTED_ORG_ID) + + denied = resolve_nested_workflow_definition( + workflow_id=WORKFLOW_ID, + environment=environment, + fallback_loader=None, + ) + accepted = resolve_nested_workflow_definition( + workflow_id=WORKFLOW_ID, + environment=environment, + fallback_loader=None, + ) + + assert denied.success is False + assert denied.error_code == "workflow_concept_not_accessible" + assert accepted.success is True + assert accepted.definition is definition + assert authority_resolver.call_count == 2 diff --git a/tests/backend/test_required_tool_identity_service.py b/tests/backend/test_required_tool_identity_service.py index 8a4bca817..103497015 100644 --- a/tests/backend/test_required_tool_identity_service.py +++ b/tests/backend/test_required_tool_identity_service.py @@ -2,6 +2,8 @@ from types import SimpleNamespace +import pytest + from src.backend.services.required_tool_identity_service import ( canonical_required_tool_key, required_tool_names_match, @@ -138,6 +140,79 @@ def test_selected_workflow_policy_matches_qualified_requirement_to_action() -> N assert policy["direct_action_tool_canonical_keys"] == ["fetch_concept"] +@pytest.mark.parametrize( + "input_key", + ("tool_name", "method_name", "mcp_tool", "mcp_method", "tool"), +) +def test_selected_workflow_policy_accepts_static_workflow_mcp_tool_binding( + input_key: str, +) -> None: + action = SimpleNamespace( + action_id="workflow_mcp.invoke_tool", + inputs={input_key: "read_file_copy"}, + is_llm_step=False, + llm_policy=None, + ) + workflow_def = SimpleNamespace( + metadata={ + "required_effects_contract": { + "required_effects": [ + {"required_tools": ["vontology:read_file_copy"]} + ] + } + }, + states={"run": SimpleNamespace(actions=[action])}, + ) + + policy = evaluate_workflow_required_effects_tool_policy(workflow_def) + + assert policy["ok"] is True + assert policy["unavailable_required_tools"] == [] + assert "read_file_copy" in policy["direct_action_tool_canonical_keys"] + + +@pytest.mark.parametrize( + ("inputs", "execution_mode", "is_llm_step"), + ( + ({"tool_name": "get_text_relations_summary"}, "deterministic", False), + ({}, "deterministic", False), + ( + {"tool_name": {"$context_key": "selected_tool_name"}}, + "deterministic", + False, + ), + ({"tool_name": "read_file_copy"}, "subworkflow", False), + ({"tool_name": "read_file_copy"}, "llm", True), + ), +) +def test_selected_workflow_policy_rejects_other_or_dynamic_workflow_mcp_tool( + inputs: dict[str, object], + execution_mode: str, + is_llm_step: bool, +) -> None: + action = SimpleNamespace( + action_id="workflow_mcp.invoke_tool", + inputs=inputs, + execution_mode=execution_mode, + is_llm_step=is_llm_step, + llm_policy=None, + ) + workflow_def = SimpleNamespace( + metadata={ + "required_effects_contract": { + "required_effects": [{"required_tools": ["read_file_copy"]}] + } + }, + states={"run": SimpleNamespace(actions=[action])}, + ) + + policy = evaluate_workflow_required_effects_tool_policy(workflow_def) + + assert policy["ok"] is False + assert policy["unavailable_required_tools"] == ["read_file_copy"] + assert policy["reason_code"] == "workflow_required_effect_tool_not_allowed" + + def test_runtime_and_step_evidence_use_the_shared_identity() -> None: invocations = [{"tool": "fetch_concept", "status": "ok", "payload": {}}] assert ( diff --git a/tests/backend/test_spreadsheet_programme_workflow_vontology_service.py b/tests/backend/test_spreadsheet_programme_workflow_vontology_service.py index b6902f574..28b74d384 100644 --- a/tests/backend/test_spreadsheet_programme_workflow_vontology_service.py +++ b/tests/backend/test_spreadsheet_programme_workflow_vontology_service.py @@ -168,7 +168,8 @@ def test_bootstrap_materialises_model_led_spreadsheet_workflow_family( for_each = _state(main, "materialise_records").actions[0] assert for_each.action_id == "workflow_control.for_each" assert for_each.inputs["workflow_id"] == SPREADSHEET_RECORD_ITEM_WORKFLOW_ID - assert for_each.inputs["success_policy"] == "allow_partial" + assert for_each.inputs["success_policy"] == "all_must_succeed" + assert for_each.inputs["stop_on_error"] is True assert for_each.inputs["max_items"] == 128 assert ( "spreadsheet_records"