Make paper recovery and failure outcomes truthful

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.

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)

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"),

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)

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)

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,15 +267,16 @@ 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:
     """Get or create the singleton LinkedIn MCP proxy instance."""
 
     global _proxy_instance
 
-    async with _proxy_lock:
+    with _proxy_lock:
         if _proxy_instance is None:
             config = _build_linkedin_config()
             _proxy_instance = LinkedInMCPProxy(config)

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,

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(