From 17065dfaad88c970932520cb8ac7cae1652b6f00 Mon Sep 17 00:00:00 2001 From: witbrock Date: Mon, 17 Aug 2026 16:04:50 +0200 Subject: [PATCH 1/2] fix: make MCP proxy acquisition loop-neutral JVNAUTOSCI-2639 --- .../internal_mcp/arxiv_proxy_mcp.py | 7 +- .../integrations/internal_mcp/catalogue.py | 33 +++++++- .../internal_mcp/github_proxy_mcp.py | 7 +- .../internal_mcp/jira_proxy_mcp.py | 7 +- .../internal_mcp/linkedin_proxy_mcp.py | 7 +- ...talogue_download_paper_prefers_finalise.py | 8 ++ ...est_internal_mcp_arxiv_download_gateway.py | 44 +++++++++++ .../backend/test_internal_mcp_github_tools.py | 4 +- .../test_internal_mcp_proxy_lifecycle.py | 79 +++++++++++++++++++ 9 files changed, 181 insertions(+), 15 deletions(-) create mode 100644 tests/backend/test_internal_mcp_proxy_lifecycle.py diff --git a/src/backend/integrations/internal_mcp/arxiv_proxy_mcp.py b/src/backend/integrations/internal_mcp/arxiv_proxy_mcp.py index 9db8e859e..ea98bfd66 100644 --- a/src/backend/integrations/internal_mcp/arxiv_proxy_mcp.py +++ b/src/backend/integrations/internal_mcp/arxiv_proxy_mcp.py @@ -5,10 +5,10 @@ from __future__ import annotations -import asyncio import hashlib import logging import os +import threading import time from dataclasses import dataclass from pathlib import Path @@ -952,14 +952,15 @@ def inspect_cached_arxiv_artifacts( # Singleton instance _proxy_instance: Optional[ArxivMCPProxy] = None -_proxy_lock = asyncio.Lock() +# Catalogue calls can run on fresh event loops, so construction is loop-neutral. +_proxy_lock = threading.Lock() async def get_arxiv_proxy() -> ArxivMCPProxy: """Get or create the global arXiv proxy instance.""" global _proxy_instance - async with _proxy_lock: + with _proxy_lock: if _proxy_instance is None: storage_path = resolve_arxiv_cache_root() config = ArxivProxyConfig(storage_path=storage_path) diff --git a/src/backend/integrations/internal_mcp/catalogue.py b/src/backend/integrations/internal_mcp/catalogue.py index f3ca12344..a210e0c71 100644 --- a/src/backend/integrations/internal_mcp/catalogue.py +++ b/src/backend/integrations/internal_mcp/catalogue.py @@ -4830,7 +4830,38 @@ async def _async_download(): preferred_filename=kwargs.get("filename"), ) if not isinstance(stored, Mapping): - proxy = await get_arxiv_proxy() + try: + proxy = await get_arxiv_proxy() + except ArxivProxyError: + raise + except Exception as exc: + cause_preview = f"{type(exc).__name__}: {exc}"[:500] + return make_error_response( + "arxiv_acquisition_unavailable", + ( + "The arXiv acquisition adapter could not be initialised " + "before the provider was invoked." + ), + details={ + "arxiv_id": arxiv_id, + "exception_type": type(exc).__name__, + "cause_preview": cause_preview, + "acquisition_stage": "proxy_initialisation", + "provider_invoked": False, + "cache_state": cache_diagnostics.get("cache_state"), + "cache_diagnostics": cache_diagnostics, + "recommended_recovery_action": cache_recovery_action, + "partial_cache_recovery_attempted": ( + partial_cache_recovery_attempted + ), + "partial_cache_markdown_deleted": ( + partial_cache_markdown_deleted + ), + "partial_cache_recovery_error": ( + partial_cache_recovery_error + ), + }, + ) stored = await proxy.download_paper( arxiv_id=arxiv_id, filename=kwargs.get("filename"), diff --git a/src/backend/integrations/internal_mcp/github_proxy_mcp.py b/src/backend/integrations/internal_mcp/github_proxy_mcp.py index 99d232795..ed1c68c5d 100644 --- a/src/backend/integrations/internal_mcp/github_proxy_mcp.py +++ b/src/backend/integrations/internal_mcp/github_proxy_mcp.py @@ -6,11 +6,11 @@ from __future__ import annotations -import asyncio import logging import os import shlex import tempfile +import threading from dataclasses import dataclass from pathlib import Path from typing import Any, Dict, Mapping, Optional @@ -207,14 +207,15 @@ def _build_github_config() -> GitHubProxyConfig: _proxy_instance: Optional[GitHubMCPProxy] = None -_proxy_lock = asyncio.Lock() +# Catalogue calls can run on fresh event loops, so construction is loop-neutral. +_proxy_lock = threading.Lock() async def get_github_proxy() -> GitHubMCPProxy: """Get or create the global GitHub MCP proxy instance.""" global _proxy_instance - async with _proxy_lock: + with _proxy_lock: if _proxy_instance is None: config = _build_github_config() _proxy_instance = GitHubMCPProxy(config) diff --git a/src/backend/integrations/internal_mcp/jira_proxy_mcp.py b/src/backend/integrations/internal_mcp/jira_proxy_mcp.py index 9b2cff8b6..203173ad2 100644 --- a/src/backend/integrations/internal_mcp/jira_proxy_mcp.py +++ b/src/backend/integrations/internal_mcp/jira_proxy_mcp.py @@ -7,10 +7,10 @@ from __future__ import annotations -import asyncio import logging import os import sys +import threading from dataclasses import dataclass from pathlib import Path from typing import Any, Dict, Mapping, Optional, cast @@ -318,14 +318,15 @@ def _build_jira_config() -> JiraProxyConfig: # Singleton instance _proxy_instance: Optional[JiraMCPProxy] = None -_proxy_lock = asyncio.Lock() +# Catalogue calls can run on fresh event loops, so construction is loop-neutral. +_proxy_lock = threading.Lock() async def get_jira_proxy() -> JiraMCPProxy: """Get or create the global Jira MCP proxy instance.""" global _proxy_instance - async with _proxy_lock: + with _proxy_lock: if _proxy_instance is None: config = _build_jira_config() _proxy_instance = JiraMCPProxy(config) diff --git a/src/backend/integrations/internal_mcp/linkedin_proxy_mcp.py b/src/backend/integrations/internal_mcp/linkedin_proxy_mcp.py index 1e91716bc..34fb23f70 100644 --- a/src/backend/integrations/internal_mcp/linkedin_proxy_mcp.py +++ b/src/backend/integrations/internal_mcp/linkedin_proxy_mcp.py @@ -7,10 +7,10 @@ from __future__ import annotations import ast -import asyncio import json import logging import os +import threading from dataclasses import dataclass from pathlib import Path from typing import Any, Dict, Optional @@ -267,7 +267,8 @@ def _build_linkedin_config() -> LinkedInProxyConfig: _proxy_instance: Optional[LinkedInMCPProxy] = None -_proxy_lock = asyncio.Lock() +# Catalogue calls can run on fresh event loops, so construction is loop-neutral. +_proxy_lock = threading.Lock() async def get_linkedin_proxy() -> LinkedInMCPProxy: @@ -275,7 +276,7 @@ async def get_linkedin_proxy() -> LinkedInMCPProxy: global _proxy_instance - async with _proxy_lock: + with _proxy_lock: if _proxy_instance is None: config = _build_linkedin_config() _proxy_instance = LinkedInMCPProxy(config) diff --git a/tests/backend/test_catalogue_download_paper_prefers_finalise.py b/tests/backend/test_catalogue_download_paper_prefers_finalise.py index 472496799..10fa90ac9 100644 --- a/tests/backend/test_catalogue_download_paper_prefers_finalise.py +++ b/tests/backend/test_catalogue_download_paper_prefers_finalise.py @@ -1,6 +1,14 @@ from pathlib import Path from typing import Any, cast +import pytest + + +@pytest.fixture(autouse=True) +def _isolated_blob_store(monkeypatch, tmp_path) -> None: + monkeypatch.setenv("VON_BLOB_STORE_BACKEND", "local") + monkeypatch.setenv("VON_BLOB_STORE_LOCAL_ROOT", str(tmp_path / "blob_store")) + def test_download_paper_prefers_finalise_when_cached_and_authenticated( monkeypatch, tmp_path diff --git a/tests/backend/test_internal_mcp_arxiv_download_gateway.py b/tests/backend/test_internal_mcp_arxiv_download_gateway.py index 180d068c1..4daa80aa2 100644 --- a/tests/backend/test_internal_mcp_arxiv_download_gateway.py +++ b/tests/backend/test_internal_mcp_arxiv_download_gateway.py @@ -2,12 +2,20 @@ from __future__ import annotations +import pytest + from src.backend.integrations.internal_mcp.catalogue import build_default_catalogue from src.backend.integrations.internal_mcp.gateway import InternalMCPGateway from src.backend.integrations.internal_mcp.schemas import validate_payload from src.backend.integrations.internal_mcp.transport import InternalMCPTransport +@pytest.fixture(autouse=True) +def _isolated_blob_store(monkeypatch, tmp_path) -> None: + monkeypatch.setenv("VON_BLOB_STORE_BACKEND", "local") + monkeypatch.setenv("VON_BLOB_STORE_LOCAL_ROOT", str(tmp_path / "blob_store")) + + def _build_gateway() -> InternalMCPGateway: return InternalMCPGateway( catalogue=build_default_catalogue(), @@ -66,3 +74,39 @@ async def _fake_get_proxy(): assert cache_diagnostics.get("cached_markdown_path") == str(markdown_path) assert cache_diagnostics.get("partial_cache_without_pdf") is True _assert_schema_conformance(gateway, "download_paper", payload) + + +@pytest.mark.asyncio +async def test_download_paper_gateway_types_failure_before_provider_invocation( + monkeypatch, tmp_path +) -> None: + cache_dir = tmp_path / "arxiv_cache" + cache_dir.mkdir(parents=True, exist_ok=True) + monkeypatch.setenv("ARXIV_CACHE_PATH", str(cache_dir)) + monkeypatch.setattr( + "src.backend.security.access_control.get_effective_user_concept_id", + lambda: "#V#user_test", + ) + + async def _failed_get_proxy(): + raise RuntimeError("adapter initialisation failed") + + monkeypatch.setattr( + "src.backend.integrations.internal_mcp.arxiv_proxy_mcp.get_arxiv_proxy", + _failed_get_proxy, + ) + + gateway = _build_gateway() + payload = gateway.invoke("download_paper", {"arxiv_id": "2608.00001"}).payload + + assert payload.get("success") is False + assert payload.get("error_code") == "arxiv_acquisition_unavailable" + details = payload.get("error_details") or {} + assert details.get("acquisition_stage") == "proxy_initialisation" + assert details.get("provider_invoked") is False + assert details.get("exception_type") == "RuntimeError" + assert details.get("arxiv_id") == "2608.00001" + assert str(details.get("cause_preview") or "").startswith("RuntimeError:") + assert len(str(details.get("cause_preview") or "")) <= 500 + assert details.get("recommended_recovery_action") == "download_from_source" + _assert_schema_conformance(gateway, "download_paper", payload) diff --git a/tests/backend/test_internal_mcp_github_tools.py b/tests/backend/test_internal_mcp_github_tools.py index 201323178..8fcf78947 100644 --- a/tests/backend/test_internal_mcp_github_tools.py +++ b/tests/backend/test_internal_mcp_github_tools.py @@ -2,7 +2,7 @@ from __future__ import annotations -import asyncio +import threading from pathlib import Path import pytest @@ -332,7 +332,7 @@ async def test_github_proxy_initialisation_does_not_log_config_arguments( monkeypatch.setattr(runtime_env, "get_project_root", lambda: tmp_path) monkeypatch.setattr(github_proxy_mcp, "_proxy_instance", None) - monkeypatch.setattr(github_proxy_mcp, "_proxy_lock", asyncio.Lock()) + monkeypatch.setattr(github_proxy_mcp, "_proxy_lock", threading.Lock()) caplog.set_level("INFO") proxy = await github_proxy_mcp.get_github_proxy() diff --git a/tests/backend/test_internal_mcp_proxy_lifecycle.py b/tests/backend/test_internal_mcp_proxy_lifecycle.py new file mode 100644 index 000000000..f052a7051 --- /dev/null +++ b/tests/backend/test_internal_mcp_proxy_lifecycle.py @@ -0,0 +1,79 @@ +"""Cross-execution lifecycle tests for internal MCP proxy acquisition.""" + +from __future__ import annotations + +import asyncio +import concurrent.futures +import importlib +import threading +import time +from pathlib import Path +from types import SimpleNamespace +from typing import Any + +import pytest + + +@pytest.mark.parametrize( + ("module_name", "builder_name", "proxy_class_name"), + [ + ( + "src.backend.integrations.internal_mcp.arxiv_proxy_mcp", + "resolve_arxiv_cache_root", + "ArxivMCPProxy", + ), + ( + "src.backend.integrations.internal_mcp.github_proxy_mcp", + "_build_github_config", + "GitHubMCPProxy", + ), + ( + "src.backend.integrations.internal_mcp.jira_proxy_mcp", + "_build_jira_config", + "JiraMCPProxy", + ), + ( + "src.backend.integrations.internal_mcp.linkedin_proxy_mcp", + "_build_linkedin_config", + "LinkedInMCPProxy", + ), + ], +) +def test_proxy_acquisition_is_safe_across_fresh_event_loops( + monkeypatch: pytest.MonkeyPatch, + module_name: str, + builder_name: str, + proxy_class_name: str, +) -> None: + module = importlib.import_module(module_name) + caller_count = 8 + start_barrier = threading.Barrier(caller_count) + + class _Proxy: + def __init__(self, config: Any) -> None: + self.config = config + self.usable = True + # Keep construction in flight so callers from independent loops + # contend on the public getter rather than running sequentially. + time.sleep(0.05) + + monkeypatch.setattr(module, "_proxy_instance", None) + monkeypatch.setattr(module, proxy_class_name, _Proxy) + if module_name.endswith("arxiv_proxy_mcp"): + monkeypatch.setattr(module, builder_name, lambda: Path("test-arxiv-cache")) + else: + config = SimpleNamespace(command="test", args=["server"]) + monkeypatch.setattr(module, builder_name, lambda: config) + + get_proxy = getattr(module, f"get_{module_name.rsplit('.', 1)[-1].split('_', 1)[0]}_proxy") + + def _invoke_from_fresh_loop() -> Any: + start_barrier.wait(timeout=2.0) + return asyncio.run(asyncio.wait_for(get_proxy(), timeout=2.0)) + + with concurrent.futures.ThreadPoolExecutor(max_workers=caller_count) as executor: + futures = [executor.submit(_invoke_from_fresh_loop) for _ in range(caller_count)] + proxies = [future.result(timeout=3.0) for future in futures] + + assert len(proxies) == caller_count + assert all(isinstance(proxy, _Proxy) and proxy.usable for proxy in proxies) From 7cbba79beb20ffa81b7adfea8a59f1f8d08ece17 Mon Sep 17 00:00:00 2001 From: witbrock Date: Mon, 17 Aug 2026 16:18:56 +0200 Subject: [PATCH 2/2] fix: make paper workflow recovery truthful JVNAUTOSCI-2640 --- .../von_workflow_language_manual.md | 1 + .../workflows/durable/durable_executor.py | 7 +- src/backend/workflows/engine.py | 7 +- ...r_representation_workflow_seed_bundle.json | 74 +++++--- .../workflows/terminal_success_contracts.py | 32 +++- .../test_durable_workflow_executor_safety.py | 51 ++++++ ...presentation_workflow_vontology_service.py | 172 +++++++++++++++--- ...st_workflow_definition_identity_service.py | 35 ++++ .../test_workflow_execution_contracts.py | 56 ++++++ 9 files changed, 384 insertions(+), 51 deletions(-) diff --git a/docs/engineering/von_workflow_language_manual.md b/docs/engineering/von_workflow_language_manual.md index e7af0dd45..c105bea93 100644 --- a/docs/engineering/von_workflow_language_manual.md +++ b/docs/engineering/von_workflow_language_manual.md @@ -888,6 +888,7 @@ terminal-success contracts, or typed route maps. - plan-state items support `pending|in_progress|blocked|done` statuses, bounded checkpoint history, periodic summary snapshots, resumable cursor snapshots, and resume telemetry; - completion gates MUST fail closed before terminal success when required plan items or required context keys are not satisfied; - terminal-success contracts MAY declare which terminal statuses count as success for a workflow boundary and which execution-summary fields MUST be present before a parent turn or workflow may treat the child workflow as successful; +- a terminal-success contract MAY declare one `failed_terminal_error_code` for failure-like terminal states when the workflow has a stable aggregate non-success reason; runtimes MUST otherwise retain the generic `workflow_failed_terminal_state` fallback; - when a workflow declares a terminal-success contract, runtimes and parent completion gates MUST fail closed if terminal status or other contracted summary fields are absent or violate the contract; - launch input contracts MAY map invocation-context values into workflow context keys before the initial state executes; - launch input contracts MUST remain declarative, so reusable extractors such as quoted-text extraction or workflow-ID list extraction are configured in metadata rather than hard-coded for specific workflow IDs. diff --git a/src/backend/workflows/durable/durable_executor.py b/src/backend/workflows/durable/durable_executor.py index 30221a5e8..fa30a59b5 100644 --- a/src/backend/workflows/durable/durable_executor.py +++ b/src/backend/workflows/durable/durable_executor.py @@ -50,6 +50,7 @@ mark_workflow_plan_state_resume, ) from ..trace_model import WorkflowExecutionTrace +from ..terminal_success_contracts import resolve_workflow_failed_terminal_error from ..trace_store import insert_workflow_execution_trace from .authority_snapshot_attestation import ( DURABLE_EXECUTED_WORKFLOW_DEFINITION_IDENTITY_KEY, @@ -1270,7 +1271,11 @@ def _complete_with_gate(final_state: str) -> DurableWorkflowResult: control_signal = get_last_control_signal(context) if workflow_final_state_is_failure_like(current_state): - error = "workflow_failed_terminal_state" + error = resolve_workflow_failed_terminal_error( + contract=(definition.metadata or {}).get( + "terminal_success_contract" + ) + ) trace.finish_failed(error) return _build_result( completed=False, diff --git a/src/backend/workflows/engine.py b/src/backend/workflows/engine.py index 9dc13bb71..893a85e5e 100644 --- a/src/backend/workflows/engine.py +++ b/src/backend/workflows/engine.py @@ -61,6 +61,7 @@ build_progress_facts_for_step, ) from .trace_model import WorkflowExecutionTrace +from .terminal_success_contracts import resolve_workflow_failed_terminal_error logger = logging.getLogger(__name__) @@ -2500,7 +2501,11 @@ def _run_in_execution_scope( transitions=transitions, trace=trace, final_state=current_state, - error="workflow_failed_terminal_state", + error=resolve_workflow_failed_terminal_error( + contract=(definition.metadata or {}).get( + "terminal_success_contract" + ) + ), ) if control_signal == WORKFLOW_CONTROL_SIGNAL_RETURN: return self._complete_with_gate( diff --git a/src/backend/workflows/repo_seed_bundles/paper_representation_workflow_seed_bundle.json b/src/backend/workflows/repo_seed_bundles/paper_representation_workflow_seed_bundle.json index 22f12a898..fad61615c 100644 --- a/src/backend/workflows/repo_seed_bundles/paper_representation_workflow_seed_bundle.json +++ b/src/backend/workflows/repo_seed_bundles/paper_representation_workflow_seed_bundle.json @@ -2,7 +2,7 @@ "family_id": "paper_representation_workflow_seed_bundle", "managed_by": "paper_representation_workflow_vontology_service", "schema_version": "repo_seed_workflow_bundle.v1", - "seed_version": "22", + "seed_version": "24", "known_legacy_authority_payload_sha256_by_seed_version": { "#V#arxiv_paper_representation_workflow": { "18": [ @@ -10,11 +10,25 @@ ], "21": [ "73d2d1570351c56cfb98fbecb409a8dfc306cade830ff73b40c5d916868a6cb9" + ], + "22": [ + "3e60cdfac1882da2b3d226d9e08df7e1cb92ef668e40d7191b73c5a4e5301e4c" + ], + "23": [ + "af03d9242cd12fb0a463091a2cbb95629da3c4075ce9202f005036f3b52af981", + "c7579667b7fe62cdff7f7c77efa1270612509ed1851a0967bd7c45f6c960e81c" ] }, "#V#source_neutral_paper_reference_ingestion_workflow": { "19": [ "c87c88b27cc9c78a7313822784890d7afba4ef57bdf5581d3c51aad8995dbbc0" + ], + "22": [ + "40b7b1d2649f3ac7e4f6339822858e63daccbab6a137b490d51ecebdd6411eb9" + ], + "23": [ + "7940a52c0bd65dcffd0838c6aefff7b2d538665e40ae7b1527cbc320172f679c", + "85b5a8a7700645f21b322a6258861be251f8a37fba87910089b102afa7dcaf31" ] } }, @@ -1622,7 +1636,7 @@ "workflow_id": "#V#arxiv_paper_representation_workflow", "display_name": "Arxiv Paper Representation Workflow", "description": "Canonical arXiv wrapper workflow that normalises an arXiv source, fetches authoritative arXiv metadata, acquires or finalises the paper artefact when available, delegates to the scholarly-paper workflow, and fails closed on incomplete metadata representation.", - "content": "Represent an arXiv paper by extracting a canonical arXiv identifier, fetching authoritative arXiv metadata, deciding whether acquisition is required, invoking download_paper when needed, trying a direct PDF URL import as optional enrichment after provider failure, then delegating to the general scholarly-paper workflow with arXiv verification even when external PDF acquisition is skipped.", + "content": "Represent an arXiv paper by extracting a canonical arXiv identifier, fetching authoritative arXiv metadata, deciding whether acquisition is required, invoking download_paper when needed, trying a direct PDF URL import as optional enrichment when primary acquisition is unavailable or fails, then delegating to the general scholarly-paper workflow with arXiv verification even when the PDF remains unavailable.", "text_relations": [ { "predicate": "#V#hasWorkflowDiscoveryExemplarsJson", @@ -1641,7 +1655,7 @@ }, { "predicate": "#V#hasWorkflowRequiredEffectsContractJson", - "text": "{\"contract_id\":\"arxiv_paper_representation_readback\",\"required_effects\":[{\"description\":\"Require verified paper concept read-back for each arXiv target supplied to the workflow turn; source/PDF artefact acquisition is optional enrichment when external providers fail.\",\"effect_id\":\"arxiv_paper_representation\",\"effect_type\":\"scholarly_representation\",\"failed_failure_code\":\"arxiv_paper_representation_failed\",\"missing_failure_code\":\"arxiv_paper_representation_not_executed\",\"not_executed_reason\":\"No verified arXiv paper representation read-back was observed for this target.\",\"not_satisfied_reason\":\"ArXiv paper representation read-back did not verify for this target.\",\"required_payload_fields\":[\"paper_concept_id\"],\"required_tools\":[\"scholarly_paper.verify_representation\"],\"targets_extractor\":\"arxiv_id_list\",\"targets_source_expressions\":[\"selected_workflow_trace.expected_outcome_contract_state.fields.summary\",\"completion_report.turn_expected_outcome_contract_state.fields.summary\",\"workflow_discovery_result.discovery_query_input\"],\"wrong_target_failure_code\":\"arxiv_paper_representation_wrong_target\"}],\"schema_version\":\"workflow_required_effects_contract.v1\"}", + "text": "{\"contract_id\":\"arxiv_paper_representation_readback\",\"required_effects\":[{\"description\":\"Require verified paper concept read-back for each arXiv target supplied to the workflow turn; source/PDF artefact acquisition is optional enrichment when primary acquisition is unavailable or fails.\",\"effect_id\":\"arxiv_paper_representation\",\"effect_type\":\"scholarly_representation\",\"failed_failure_code\":\"arxiv_paper_representation_failed\",\"missing_failure_code\":\"arxiv_paper_representation_not_executed\",\"not_executed_reason\":\"No verified arXiv paper representation read-back was observed for this target.\",\"not_satisfied_reason\":\"ArXiv paper representation read-back did not verify for this target.\",\"required_payload_fields\":[\"paper_concept_id\"],\"required_tools\":[\"scholarly_paper.verify_representation\"],\"targets_extractor\":\"arxiv_id_list\",\"targets_source_expressions\":[\"selected_workflow_trace.expected_outcome_contract_state.fields.summary\",\"completion_report.turn_expected_outcome_contract_state.fields.summary\",\"workflow_discovery_result.discovery_query_input\"],\"wrong_target_failure_code\":\"arxiv_paper_representation_wrong_target\"}],\"schema_version\":\"workflow_required_effects_contract.v1\"}", "lang": "en-NZ" }, { @@ -2281,14 +2295,12 @@ "expected": true }, { - "key": "last_action_outputs.result.error_details.provider", - "kind": "context_value_equals", - "value": "external_third_party" - }, - { - "key": "last_action_outputs.result.error_details.external_provider", - "kind": "context_value_equals", - "value": "arxiv-mcp-server" + "key": "last_action_outputs.result.error_details.recommended_recovery_action", + "kind": "context_value_in", + "values": [ + "download_from_source", + "reacquire_pdf_from_source" + ] } ], "kind": "all" @@ -2334,11 +2346,6 @@ "concept_id": "#V#workflow_mapping_tool_field_arxiv_paper_representation_workflow_download_or_finalise_result_computer_file_copy_concept_id_to_file_copy_concept_id", "context_key": "file_copy_concept_id", "tool_output_field": "result.computer_file_copy_concept_id" - }, - { - "concept_id": "#V#workflow_mapping_tool_field_arxiv_paper_representation_workflow_download_or_finalise_result_error_details_to_arxiv_mcp_failure_details", - "context_key": "arxiv_mcp_failure_details", - "tool_output_field": "result.error_details" } ], "writes_context_keys": [ @@ -2378,8 +2385,11 @@ "template": "{failure_kind}", "variables": { "failure_kind": { - "default": "external_arxiv_mcp_failed", - "value_from_context": "last_action_outputs.result.error_details.external_provider_failure_kind" + "default": "paper_acquisition_failed", + "value_from_context_options": [ + "last_action_outputs.result.error_code", + "last_action_outputs.result.error_details.external_provider_failure_kind" + ] } } } @@ -2471,7 +2481,7 @@ [ { "key": "pdf_acquisition_status", - "value": "skipped" + "value": "unavailable" }, { "key": "pdf_acquisition_optional", @@ -2718,7 +2728,7 @@ "workflow_id": "#V#source_neutral_paper_reference_ingestion_workflow", "display_name": "Source Neutral Paper Reference Ingestion Workflow", "description": "Canonical source-neutral entry workflow for ingesting scholarly paper references supplied as arXiv identifiers or URLs, DOI/source URLs, uploaded/file-copy concepts, pasted metadata, or mixed reference sets without assuming a mail source, sender profile, or English prompt wording.", - "content": "Normalise caller-supplied scholarly paper references into paper_reference_set.v1, preserve source/provenance context, optionally preview the list without mutation, then fan out each item to the represented item-ingestion workflow with allow-partial outcomes. Source-specific representation remains delegated to existing arXiv, scholarly metadata, and file-copy paper workflows.", + "content": "Normalise caller-supplied scholarly paper references into paper_reference_set.v1, preserve source/provenance context, optionally preview the list without mutation, then fan out each item to the represented item-ingestion workflow. A non-preview run completes only when at least one item succeeds; mixed results preserve item-level failures. Source-specific representation remains delegated to existing arXiv, scholarly metadata, and file-copy paper workflows.", "text_relations": [ { "predicate": "#V#hasWorkflowDiscoveryExemplarsJson", @@ -2732,12 +2742,12 @@ }, { "predicate": "#V#hasWorkflowTerminalSuccessContractJson", - "text": "{\"schema_version\":\"workflow_terminal_success_contract.v1\",\"success_statuses\":[\"completed\"],\"require_completed_true\":true,\"require_final_state\":true,\"require_terminal_status\":true,\"required_summary_fields\":[\"workflow_id\",\"terminal_status\",\"final_state\",\"completed\",\"paper_reference_item_count\",\"for_each_success_count\",\"for_each_error_count\"]}", + "text": "{\"schema_version\":\"workflow_terminal_success_contract.v1\",\"success_statuses\":[\"completed\"],\"require_completed_true\":true,\"require_final_state\":true,\"require_terminal_status\":true,\"required_summary_fields\":[\"workflow_id\",\"terminal_status\",\"final_state\",\"completed\",\"paper_reference_item_count\",\"for_each_success_count\",\"for_each_error_count\"],\"failed_terminal_error_code\":\"paper_reference_ingestion_no_items_succeeded\"}", "lang": "en-NZ" }, { "predicate": "#V#hasWorkflowRequiredEffectsContractJson", - "text": "{\"schema_version\":\"workflow_required_effects_contract.v1\",\"contract_id\":\"source_neutral_paper_reference_ingestion\",\"required_effects\":[{\"effect_id\":\"normalised_paper_reference_set\",\"effect_type\":\"workflow_context_projection\",\"postcondition_strategy\":\"execution_observed\",\"description\":\"Normalise the request into paper_reference_set.v1 while preserving source_context provenance fields.\",\"required_payload_fields\":[\"normalised_paper_reference_set\",\"paper_reference_items\"],\"missing_failure_code\":\"paper_reference_set_not_normalised\",\"failed_failure_code\":\"paper_reference_set_normalisation_failed\",\"required_tools\":[\"paper_reference.normalise_reference_set\"]},{\"effect_id\":\"paper_reference_item_delegation\",\"effect_type\":\"workflow_delegation\",\"postcondition_strategy\":\"execution_observed\",\"description\":\"Delegate each non-preview item to the represented source-specific paper workflow for its reference kind.\",\"required_payload_fields\":[\"iteration_results\"],\"required_tools\":[\"workflow_control.for_each\",\"workflow_invoke_subworkflow\"],\"missing_failure_code\":\"paper_reference_items_not_dispatched\",\"failed_failure_code\":\"paper_reference_item_dispatch_failed\"},{\"effect_id\":\"paper_reference_item_outcomes\",\"effect_type\":\"workflow_result_projection\",\"postcondition_strategy\":\"execution_observed\",\"description\":\"Return per-item outcomes, preserving partial success and typed failures.\",\"required_payload_fields\":[\"for_each_success_count\",\"for_each_error_count\",\"iteration_results\"],\"missing_failure_code\":\"paper_reference_item_outcomes_missing\",\"failed_failure_code\":\"paper_reference_item_outcomes_failed\",\"required_tools\":[\"workflow_control.for_each\"]}]}", + "text": "{\"schema_version\":\"workflow_required_effects_contract.v1\",\"contract_id\":\"source_neutral_paper_reference_ingestion\",\"required_effects\":[{\"effect_id\":\"normalised_paper_reference_set\",\"effect_type\":\"workflow_context_projection\",\"postcondition_strategy\":\"execution_observed\",\"description\":\"Normalise the request into paper_reference_set.v1 while preserving source_context provenance fields.\",\"required_payload_fields\":[\"normalised_paper_reference_set\",\"paper_reference_items\"],\"missing_failure_code\":\"paper_reference_set_not_normalised\",\"failed_failure_code\":\"paper_reference_set_normalisation_failed\",\"required_tools\":[\"paper_reference.normalise_reference_set\"]},{\"effect_id\":\"paper_reference_item_delegation\",\"effect_type\":\"workflow_delegation\",\"postcondition_strategy\":\"execution_observed\",\"description\":\"Attempt represented source-specific execution for each non-preview item and retain every item outcome.\",\"required_payload_fields\":[\"iteration_results\"],\"required_tools\":[\"workflow_control.for_each\",\"workflow_invoke_subworkflow\"],\"missing_failure_code\":\"paper_reference_items_not_dispatched\",\"failed_failure_code\":\"paper_reference_item_dispatch_failed\"},{\"effect_id\":\"paper_reference_item_outcomes\",\"effect_type\":\"workflow_result_projection\",\"postcondition_strategy\":\"execution_observed\",\"description\":\"Return per-item outcomes, preserving partial success and typed failures; a non-empty execution with zero successful items is a failed ingestion outcome.\",\"required_payload_fields\":[\"for_each_success_count\",\"for_each_error_count\",\"iteration_results\"],\"missing_failure_code\":\"paper_reference_item_outcomes_missing\",\"failed_failure_code\":\"paper_reference_item_outcomes_failed\",\"required_tools\":[\"workflow_control.for_each\"]}]}", "lang": "en-NZ" }, { @@ -2915,7 +2925,25 @@ "state_id": "dispatch_reference_items", "action_id": "workflow_control.for_each", "execution_mode": "deterministic", - "next_state": "completed", + "conditional_transitions": [ + { + "condition_spec": { + "key": "for_each_success_count", + "kind": "context_compare", + "operator": "gt", + "value": 0 + }, + "reason": "paper_reference_items_produced_success", + "to_state": "completed" + }, + { + "condition_spec": { + "kind": "always" + }, + "reason": "paper_reference_items_produced_no_success", + "to_state": "failed" + } + ], "static_input_bindings": [ [ "workflow_id", diff --git a/src/backend/workflows/terminal_success_contracts.py b/src/backend/workflows/terminal_success_contracts.py index fa4884d6d..ffcab8c78 100644 --- a/src/backend/workflows/terminal_success_contracts.py +++ b/src/backend/workflows/terminal_success_contracts.py @@ -88,7 +88,7 @@ def normalise_workflow_terminal_success_contract( "completed", ] - return { + contract = { "schema_version": WORKFLOW_TERMINAL_SUCCESS_CONTRACT_SCHEMA_VERSION, "success_statuses": success_statuses, "required_summary_fields": required_summary_fields, @@ -109,6 +109,35 @@ def normalise_workflow_terminal_success_contract( default=False, ), } + if "failed_terminal_error_code" in value: + failed_terminal_error_code_value = value.get("failed_terminal_error_code") + if not isinstance(failed_terminal_error_code_value, str): + raise ValueError( + "workflow_terminal_success_contract_failed_terminal_error_code_invalid" + ) + failed_terminal_error_code = failed_terminal_error_code_value.strip() + if not failed_terminal_error_code: + raise ValueError( + "workflow_terminal_success_contract_failed_terminal_error_code_invalid" + ) + contract["failed_terminal_error_code"] = failed_terminal_error_code + return contract + + +def resolve_workflow_failed_terminal_error( + *, + contract: Mapping[str, Any] | None, + fallback: str = "workflow_failed_terminal_state", +) -> str: + """Resolve the workflow-authored code for an already-proven failed terminal.""" + + if isinstance(contract, Mapping) and isinstance( + contract.get("failed_terminal_error_code"), str + ): + error_code = contract["failed_terminal_error_code"].strip() + if error_code: + return error_code + return _normalise_text(fallback) or "workflow_failed_terminal_state" def _summary_field_present(summary: Mapping[str, Any], field_name: str) -> bool: @@ -222,4 +251,5 @@ def evaluate_workflow_terminal_success_contract( "WORKFLOW_TERMINAL_SUCCESS_EVALUATION_SCHEMA_VERSION", "evaluate_workflow_terminal_success_contract", "normalise_workflow_terminal_success_contract", + "resolve_workflow_failed_terminal_error", ] diff --git a/tests/backend/test_durable_workflow_executor_safety.py b/tests/backend/test_durable_workflow_executor_safety.py index 9a127cac3..6e182366e 100644 --- a/tests/backend/test_durable_workflow_executor_safety.py +++ b/tests/backend/test_durable_workflow_executor_safety.py @@ -150,6 +150,57 @@ def test_durable_executor_reports_explicit_failed_terminal_as_failure() -> None: ) +def test_durable_executor_uses_authored_failed_terminal_error_code() -> None: + failed_state = "#V#workflow_step_example_workflow_failed" + definition = WorkflowDefinition( + workflow_id="#V#durable_failed_terminal_probe", + initial_state=failed_state, + states={ + failed_state: WorkflowStateSpec(state_id=failed_state, terminal=True), + }, + termination_states=(failed_state,), + metadata={ + "terminal_success_contract": { + "failed_terminal_error_code": "example_workflow_no_items_succeeded" + } + }, + ) + manager = MagicMock() + manager.get_instance.return_value = _build_instance(definition.workflow_id) + manager.is_cancelled.return_value = False + manager.extend_lock.return_value = True + manager.checkpoint.return_value = True + + executor = DurableWorkflowExecutor( + registry=ActionRegistry(), + instance_manager=manager, + ) + with ( + patch( + "src.backend.languagemodels.llm_interface.get_llm_client", + return_value=MagicMock(), + ), + patch( + "src.backend.languagemodels.llm_interface.get_active_model_name", + return_value="test-model", + ), + ): + result = executor.run_durable( + "instance-1", + definition, + resume_from_checkpoint=False, + ) + + assert result.completed is False + assert result.error == "example_workflow_no_items_succeeded" + assert result.data[WORKFLOW_RESULT_ENVELOPE_KEY]["diagnostics"]["error"] == ( + "example_workflow_no_items_succeeded" + ) + assert manager.checkpoint.call_args_list[-1].kwargs["error"] == ( + "example_workflow_no_items_succeeded" + ) + + def test_durable_executor_atomically_pauses_at_successor_checkpoint() -> None: registry = ActionRegistry() register_control_flow_actions(registry, definition_loader=lambda _workflow_id: None) diff --git a/tests/backend/test_paper_representation_workflow_vontology_service.py b/tests/backend/test_paper_representation_workflow_vontology_service.py index c6d77d8a9..2942b9501 100644 --- a/tests/backend/test_paper_representation_workflow_vontology_service.py +++ b/tests/backend/test_paper_representation_workflow_vontology_service.py @@ -480,6 +480,35 @@ def test_bootstrap_materialises_paper_representation_workflow_family( SOURCE_NEUTRAL_PAPER_REFERENCE_ITEM_INGESTION_WORKFLOW_ID ) assert source_dispatch_action.inputs["success_policy"] == "allow_partial" + source_dispatch_transitions = { + transition.reason: transition + for transition in source_neutral_definition.states[ + source_dispatch_state_id + ].transitions + } + assert source_dispatch_transitions[ + "paper_reference_items_produced_success" + ].condition_spec == { + "key": "for_each_success_count", + "kind": "context_compare", + "operator": "gt", + "value": 0, + } + assert source_dispatch_transitions[ + "paper_reference_items_produced_success" + ].to_state == authority_service._step_concept_id( + workflow_id=SOURCE_NEUTRAL_PAPER_REFERENCE_INGESTION_WORKFLOW_ID, + state_id="completed", + ) + assert source_dispatch_transitions[ + "paper_reference_items_produced_no_success" + ].to_state == authority_service._step_concept_id( + workflow_id=SOURCE_NEUTRAL_PAPER_REFERENCE_INGESTION_WORKFLOW_ID, + state_id="failed", + ) + assert source_neutral_definition.metadata["terminal_success_contract"][ + "failed_terminal_error_code" + ] == "paper_reference_ingestion_no_items_succeeded" source_required_effects = source_neutral_definition.metadata.get( "required_effects_contract" ) @@ -1150,14 +1179,19 @@ def test_bootstrap_materialises_paper_representation_workflow_family( ) download_state = arxiv_definition.states[download_state_id] assert download_state.metadata["retry_policy"]["max_attempts"] == 2 + assert all( + mapping["context_key"] != "arxiv_mcp_failure_details" + for mapping in download_state.metadata["tool_output_context_mappings"] + ) download_transitions = { transition.reason: transition for transition in download_state.transitions } assert download_transitions["download_or_finalise_succeeded"].to_state == ( delegate_state_id ) - assert download_transitions["on_failure"].to_state == build_pdf_import_state_id - assert download_transitions["on_failure"].condition_spec == { + recovery_transition = download_transitions["on_failure"] + assert recovery_transition.to_state == build_pdf_import_state_id + assert recovery_transition.condition_spec == { "conditions": [ { "expected": True, @@ -1165,23 +1199,31 @@ def test_bootstrap_materialises_paper_representation_workflow_family( "kind": "context_flag", }, { - "key": "last_action_outputs.result.error_details.provider", - "kind": "context_value_equals", - "value": "external_third_party", - }, - { - "key": "last_action_outputs.result.error_details.external_provider", - "kind": "context_value_equals", - "value": "arxiv-mcp-server", + "key": ( + "last_action_outputs.result.error_details." + "recommended_recovery_action" + ), + "kind": "context_value_in", + "values": ["download_from_source", "reacquire_pdf_from_source"], }, ], "kind": "all", } + assert "finalise_cached_pdf" not in recovery_transition.condition_spec[ + "conditions" + ][1]["values"] build_pdf_import_action = arxiv_definition.states[ build_pdf_import_state_id ].actions[0] assert build_pdf_import_action.action_id == "workflow_control.context_template" assert build_pdf_import_action.inputs["assignments"][0]["key"] == "arxiv_pdf_url" + import_reason_variable = build_pdf_import_action.inputs["assignments"][2][ + "variables" + ]["failure_kind"] + assert import_reason_variable["value_from_context_options"] == [ + "last_action_outputs.result.error_code", + "last_action_outputs.result.error_details.external_provider_failure_kind", + ] import_pdf_action = arxiv_definition.states[import_pdf_state_id].actions[0] assert import_pdf_action.action_id == "import_url_file_copy" assert import_pdf_action.inputs["url"] == { @@ -1218,7 +1260,7 @@ def test_bootstrap_materialises_paper_representation_workflow_family( ].actions[0] assert record_pdf_import_skipped_action.action_id == "workflow_control.context_set" assert record_pdf_import_skipped_action.inputs.get("assignments")[:3] == [ - {"key": "pdf_acquisition_status", "value": "skipped"}, + {"key": "pdf_acquisition_status", "value": "unavailable"}, {"key": "pdf_acquisition_optional", "value": True}, { "key": "pdf_acquisition_failure_stage", @@ -1599,6 +1641,52 @@ def test_source_neutral_paper_reference_workflow_fans_out_mixed_references_with_ assert errors[0]["error"] == "paper_reference_kind_unsupported" +def test_source_neutral_paper_reference_workflow_fails_when_no_item_succeeds( + _reset_mock_db: Any, + monkeypatch: pytest.MonkeyPatch, +) -> None: + bootstrap_canonical_paper_representation_workflows() + parent_definition, item_definition = _load_source_neutral_test_definitions() + registry, calls = _build_source_neutral_execution_registry( + parent_definition=parent_definition, + item_definition=item_definition, + monkeypatch=monkeypatch, + ) + + result = WorkflowExecutor(registry=registry, max_transitions=10).run( + parent_definition, + environment=WorkflowEnvironment( + llm_client=None, + user_namespace=_LIVE_ARXIV_ACCEPTANCE_NAMESPACE, + user_concept_id=_LIVE_ARXIV_ACCEPTANCE_USER_ID, + org_concept_id=_LIVE_ARXIV_ACCEPTANCE_ORG_ID, + ), + data={ + "paper_references": [ + { + "reference_kind": "unsupported_reference", + "source_uri": "urn:example:not-a-paper", + } + ] + }, + ) + + assert result.completed is False + assert result.final_state.endswith("_failed") + assert result.error == "paper_reference_ingestion_no_items_succeeded" + assert result.result_envelope is not None + assert result.result_envelope["diagnostics"]["error"] == ( + "paper_reference_ingestion_no_items_succeeded" + ) + assert result.data["paper_reference_item_count"] == 1 + assert result.data["for_each_success_count"] == 0 + assert result.data["for_each_error_count"] == 1 + assert result.data["iteration_results"][0]["error"] == ( + "paper_reference_kind_unsupported" + ) + assert calls == [] + + def test_source_neutral_paper_reference_workflow_preview_mode_does_not_delegate( _reset_mock_db: Any, monkeypatch: pytest.MonkeyPatch, @@ -1889,7 +1977,7 @@ def test_bootstrap_seed_version_refresh_repairs_old_arxiv_launch_contract( for row in marker_rows if isinstance(row.get("text"), str) ] - assert any(payload.get("seed_version") == "22" for payload in marker_payloads) + assert any(payload.get("seed_version") == "24" for payload in marker_payloads) refreshed_definition = load_workflow_definition_from_vontology( ARXIV_PAPER_REPRESENTATION_WORKFLOW_ID @@ -2084,9 +2172,50 @@ def test_arxiv_launch_contract_resolves_deictic_paper_ids_from_prior_context( @pytest.mark.parametrize("import_succeeds", [True, False]) -def test_arxiv_workflow_routes_external_mcp_failure_to_workflow_url_import( +@pytest.mark.parametrize( + ("download_error_code", "download_error_details", "expected_import_reason"), + [ + ( + "external_arxiv_mcp_download_failed", + { + "provider": "external_third_party", + "external_provider": "arxiv-mcp-server", + "external_provider_operation": "download_paper", + "external_provider_failure_kind": "provider_error", + "external_provider_retryable": True, + "recovery_hint": "import_pdf_url", + "recommended_recovery_action": "download_from_source", + }, + "external_arxiv_mcp_download_failed", + ), + ( + "arxiv_acquisition_unavailable", + { + "acquisition_stage": "proxy_initialisation", + "provider_invoked": False, + "exception_type": "RuntimeError", + "recommended_recovery_action": "download_from_source", + }, + "arxiv_acquisition_unavailable", + ), + ( + "arxiv_acquisition_unavailable", + { + "acquisition_stage": "proxy_initialisation", + "provider_invoked": False, + "exception_type": "RuntimeError", + "recommended_recovery_action": "reacquire_pdf_from_source", + }, + "arxiv_acquisition_unavailable", + ), + ], +) +def test_arxiv_workflow_routes_recoverable_acquisition_failure_to_url_import( _reset_mock_db: Any, import_succeeds: bool, + download_error_code: str, + download_error_details: dict[str, Any], + expected_import_reason: str, ) -> None: bootstrap_canonical_paper_representation_workflows() registry_factory._resolve_subworkflow_definition.cache_clear() @@ -2158,19 +2287,12 @@ def _download_failure(request: WorkflowActionRequest) -> WorkflowActionResult: download_calls.append(dict(request.inputs)) return WorkflowActionResult( status="failed", - error="external_arxiv_mcp_download_failed", + error=download_error_code, outputs={ "result": { "success": False, - "error_code": "external_arxiv_mcp_download_failed", - "error_details": { - "provider": "external_third_party", - "external_provider": "arxiv-mcp-server", - "external_provider_operation": "download_paper", - "external_provider_failure_kind": "provider_error", - "external_provider_retryable": True, - "recovery_hint": "import_pdf_url", - }, + "error_code": download_error_code, + "error_details": dict(download_error_details), } }, ) @@ -2287,11 +2409,11 @@ def _verify_representation( ) assert verify_calls[0]["require_file_copy"] is False assert result.data["arxiv_pdf_url"] == "https://arxiv.org/pdf/2406.15341.pdf" - assert result.data["arxiv_pdf_import_reason"] == "provider_error" + assert result.data["arxiv_pdf_import_reason"] == expected_import_reason if import_succeeds: assert "pdf_acquisition_status" not in result.data else: - assert result.data["pdf_acquisition_status"] == "skipped" + assert result.data["pdf_acquisition_status"] == "unavailable" assert result.data["pdf_acquisition_optional"] is True assert result.data["pdf_acquisition_failure_stage"] == ( "import_arxiv_pdf_from_url" diff --git a/tests/backend/test_workflow_definition_identity_service.py b/tests/backend/test_workflow_definition_identity_service.py index 29540c73c..2256918d1 100644 --- a/tests/backend/test_workflow_definition_identity_service.py +++ b/tests/backend/test_workflow_definition_identity_service.py @@ -904,6 +904,41 @@ def test_validate_contract_rejects_invalid_terminal_success_contract() -> None: ) +def test_validate_contract_rejects_invalid_failed_terminal_error_code() -> None: + definition = WorkflowDefinition( + workflow_id="#V#terminal_success_contract_invalid_error_code_workflow", + initial_state="done", + states={ + "done": WorkflowStateSpec( + state_id="done", + actions=(WorkflowActionInvocation(action_id="mark.done"),), + terminal=True, + ), + }, + termination_states=("done",), + metadata={ + "terminal_success_contract": { + "schema_version": "workflow_terminal_success_contract.v1", + "failed_terminal_error_code": {"not": "a string"}, + } + }, + ) + + validation = validate_workflow_definition_contract(definition=definition) + + assert validation["valid"] is False + assert validation["terminal_success_contract_issues"] == [ + { + "scope": "workflow", + "reason_code": ( + "workflow_terminal_success_contract_" + "failed_terminal_error_code_invalid" + ), + } + ] + + + def test_validate_contract_rejects_invalid_required_effects_contract() -> None: definition = WorkflowDefinition( workflow_id="#V#required_effects_contract_invalid_workflow", diff --git a/tests/backend/test_workflow_execution_contracts.py b/tests/backend/test_workflow_execution_contracts.py index 4275b52f1..bef169bf7 100644 --- a/tests/backend/test_workflow_execution_contracts.py +++ b/tests/backend/test_workflow_execution_contracts.py @@ -202,6 +202,62 @@ def test_workflow_executor_does_not_complete_explicit_failed_concept_state() -> assert result.result_envelope["terminal_status"] == "failed" +def test_workflow_executor_uses_authored_failed_terminal_error_code() -> None: + failed_state = "#V#workflow_step_example_workflow_failed" + definition = WorkflowDefinition( + workflow_id="#V#example_workflow", + initial_state=failed_state, + states={ + failed_state: WorkflowStateSpec(state_id=failed_state, terminal=True), + }, + termination_states=(failed_state,), + metadata={ + "terminal_success_contract": { + "failed_terminal_error_code": "example_workflow_no_items_succeeded" + } + }, + ) + + result = WorkflowExecutor(registry=ActionRegistry(), max_transitions=3).run( + definition, + environment=WorkflowEnvironment(llm_client=None), + data={}, + ) + + assert result.completed is False + assert result.error == "example_workflow_no_items_succeeded" + assert result.result_envelope is not None + assert result.result_envelope["diagnostics"]["error"] == ( + "example_workflow_no_items_succeeded" + ) + + +def test_workflow_executor_ignores_malformed_failed_terminal_error_code() -> None: + failed_state = "#V#workflow_step_example_workflow_failed" + definition = WorkflowDefinition( + workflow_id="#V#example_workflow", + initial_state=failed_state, + states={ + failed_state: WorkflowStateSpec(state_id=failed_state, terminal=True), + }, + termination_states=(failed_state,), + metadata={ + "terminal_success_contract": { + "failed_terminal_error_code": {"not": "a string"} + } + }, + ) + + result = WorkflowExecutor(registry=ActionRegistry(), max_transitions=3).run( + definition, + environment=WorkflowEnvironment(llm_client=None), + data={}, + ) + + assert result.completed is False + assert result.error == "workflow_failed_terminal_state" + + def test_failed_terminal_state_takes_precedence_over_return_signal() -> None: failed_state = "#V#workflow_step_returning_workflow_failed" definition = WorkflowDefinition(