Preserve safe manager failure details and original cause

adapters/cli/amplifier_loop_live_cli/bundle_astra.py

@@ -55,13 +55,13 @@ def _validate_native_items(self, items):
         from .computer_results import validate_computer_outputs
         return validate_computer_outputs(items)
 
-    async def _native_response(self, params):
+    async def _native_response(self, params, **kwargs):
         # The provider owns native eligibility and its own request context.
         # Bind the loop's existing attribution only at the actual native wire
         # boundary, so its durable job ledger can retain original async calls.
         token = LOOP_NATIVE_REQUEST.set(self if NATIVE_REQUEST.get() is self else None)
         try:
-            return await super()._native_response(params)
+            return await super()._native_response(params, **kwargs)
         finally:
             LOOP_NATIVE_REQUEST.reset(token)
 

amplifier_module_loop_live/failures.py

@@ -0,0 +1,62 @@
+"""Bounded public turn failures; exception payloads remain private causes."""
+
+from amplifier_core.llm_errors import (
+    AuthenticationError,
+    ContentFilterError,
+    ContextLengthError,
+    InvalidRequestError,
+    LLMError,
+    LLMTimeoutError,
+    ProviderUnavailableError,
+    RateLimitError,
+)
+
+
+def turn_failure(error, stage="manager_turn"):
+    """Describe a failure without copying SDK bodies, prompts, paths or keys.
+
+    A provider attribute identifies a provider-related check, not proof that a
+    request reached the service. Earlier tools may already have changed state.
+    """
+    category, message = "unknown", "Manager turn failed. Inspect saved details before continuing."
+    origin = error.__traceback__
+    while origin is not None and origin.tb_next is not None:
+        origin = origin.tb_next
+    module = origin.tb_frame.f_globals.get("__name__", "") if origin else ""
+    local_context = (module.startswith("amplifier_module_context_")
+                     or module == "amplifier_module_loop_streaming")
+    if isinstance(error, ContextLengthError):
+        category = "context_limit"
+        if local_context:
+            stage = "context_preparation"
+            message = "Local context preparation could not fit the required content within the input budget. Required content was not discarded."
+        else:
+            message = "The request could not fit within the context budget."
+    elif isinstance(error, AuthenticationError):
+        category, message = "authentication", "The selected provider rejected its credentials."
+    elif isinstance(error, RateLimitError):
+        category, message = "rate_limit", "The selected provider's rate limit was reached."
+    elif isinstance(error, ContentFilterError):
+        category, message = "content_filter", "The selected provider stopped the request under its content policy."
+    elif isinstance(error, InvalidRequestError):
+        category, message = "invalid_request", "The selected provider could not accept the request format."
+    elif isinstance(error, LLMTimeoutError):
+        category, message = "provider_timeout", "The provider request timed out; its outcome may be unknown."
+    elif isinstance(error, ProviderUnavailableError):
+        category, message = "provider_unavailable", "The selected provider is unavailable."
+    if isinstance(error, LLMError) and error.provider and stage != "context_preparation":
+        stage = "provider_request"
+    # Class names from arbitrary third-party exceptions are not trusted text.
+    kind = next((cls.__name__ for cls in type(error).__mro__
+                 if cls.__module__ in {"builtins", "amplifier_core.llm_errors"}), "Exception")
+    return {"error_type": kind, "error_category": category, "error_stage": stage,
+            "error_message": message, "effects": "not_rolled_back", "replayed": False,
+            "retryable": isinstance(error, LLMError) and error.retryable is True}
+
+
+class ManagerTurnError(RuntimeError):
+    """Safe public message with the original exception available via __cause__."""
+
+    def __init__(self, failure):
+        self.failure = failure
+        super().__init__(failure["error_message"] + " Work was not automatically replayed; earlier effects were not rolled back.")

amplifier_module_loop_live/orchestrator.py

@@ -20,6 +20,7 @@
 from .host import HostAdapter, AttachmentContext
 from .scope import HOST_ADAPTER
 from .dispatch import ApprovedDispatch
+from .failures import ManagerTurnError, turn_failure
 
 _WAKE = "[loop-live inbox wake]"
 _GOAL_SCOPE = ContextVar("live_goal_scope", default=None)
@@ -192,6 +193,7 @@ async def turn(command):
             activation = coordinator.get_capability("live.activation")
             activation_token = None
             owner_token = LIVE_OWNER.set(self)
+            stage = "turn_setup"
             try:
                 if activation:
                     activation_token = activation.bind(command.activation)
@@ -202,13 +204,14 @@ async def turn(command):
                 if command.attachments:
                     turn_context = AttachmentContext(context, self._text(command), self.host.content(command, coordinator))
                 turn_context = InputContext(turn_context, self._text(command), command)
+                stage = "manager_turn"
                 result = await self._execute_guarded_goal(
                     self._text(command), turn_context, providers, tools, hooks, coordinator)
                 await runtime.inbox.put(("bundle_turn", (result, None)))
             except asyncio.CancelledError:
                 raise
             except Exception as exc:
-                await runtime.inbox.put(("bundle_turn", (None, type(exc).__name__)))
+                await runtime.inbox.put(("bundle_turn", (None, (exc, turn_failure(exc, stage)))))
             finally:
                 if activation and activation_token is not None:
                     activation.reset(activation_token)
@@ -286,9 +289,11 @@ async def turn(command):
                     active = None
                     result, error = value
                     if error:
-                        await runtime.emit("generation.failed", error_type=error)
-                        await runtime.emit("provider.error", error_type=error)
-                        raise RuntimeError("Manager turn failed; no automatic replay")
+                        cause, failure = error
+                        await runtime.emit("generation.failed", **failure)
+                        # The base engine owns provider-specific events. A local
+                        # context/setup failure is not evidence of provider failure.
+                        raise ManagerTurnError(failure) from cause
                     last = result or last
                     checkpoint = coordinator.get_capability("live.checkpoint")
                     if checkpoint:

tests/test_failures.py

@@ -0,0 +1,59 @@
+import asyncio
+import json
+from types import SimpleNamespace
+from unittest.mock import AsyncMock
+
+import pytest
+from amplifier_core import HookResult
+from amplifier_core.llm_errors import AuthenticationError, ContextLengthError, LLMTimeoutError
+from amplifier_module_loop_live.failures import ManagerTurnError, turn_failure
+from amplifier_module_loop_live.orchestrator import BundleLiveOrchestrator
+from amplifier_module_loop_live.runtime import Runtime
+
+
+@pytest.mark.parametrize('error,category,stage', [
+    (ContextLengthError('secret prompt and base64 data'), 'context_limit', 'manager_turn'),
+    (ContextLengthError('private request', provider='fixture'), 'context_limit', 'provider_request'),
+    (AuthenticationError('Authorization: Bearer secret-key', provider='fixture'), 'authentication', 'provider_request'),
+    (LLMTimeoutError('https://secret:password@example.com', provider='fixture'), 'provider_timeout', 'provider_request'),
+    (ValueError('private local file and tool arguments'), 'unknown', 'manager_turn'),
+])
+def test_failure_uses_safe_taxonomy_not_exception_payload(error, category, stage):
+    failure = turn_failure(error)
+    assert failure['error_category'] == category
+    assert failure['error_stage'] == stage
+    assert failure['effects'] == 'not_rolled_back' and failure['replayed'] is False
+    assert str(error) not in json.dumps(failure)
+
+
+async def test_local_failure_keeps_cause_and_identity_without_provider_error_or_replay():
+    runtime = Runtime()
+    loop = BundleLiveOrchestrator({})
+    from amplifier_module_context_simple import SimpleContextManager
+    budget_context = SimpleContextManager({})
+    budget_context._request_hard_fit = True
+    try:
+        budget_context._check_retained_budget([{'role': 'user', 'content': 'secret-key ' * 100}], 1)
+    except ContextLengthError as error:
+        cause = error
+    loop._execute_guarded_goal = AsyncMock(side_effect=cause)
+    context = SimpleNamespace(get_messages=AsyncMock(return_value=[]))
+    capabilities = {'live.runtime': runtime}
+    coordinator = SimpleNamespace(get_capability=capabilities.get,
+        register_capability=capabilities.__setitem__, config={})
+    hooks = SimpleNamespace(register=lambda *a, **kw: None,
+        emit=AsyncMock(return_value=HookResult()))
+    provider = SimpleNamespace(name='fixture', config={})
+    with pytest.raises(ManagerTurnError) as captured:
+        await asyncio.wait_for(loop.execute('user request', context, {'fixture': provider}, {}, hooks, coordinator), 3)
+    assert captured.value.__cause__ is cause
+    assert 'secret-key' not in str(captured.value)
+    failed = next(e for e in runtime.events if e['type'] == 'generation.failed')
+    delivered = next(e for e in runtime.events if e['type'] == 'input.delivered')
+    assert failed['input_ids'] == [delivered['input_id']]
+    assert failed['error_type'] == 'ContextLengthError'
+    assert failed['error_stage'] == 'context_preparation'
+    assert 'provider.error' not in [e['type'] for e in runtime.events]
+    assert 'generation.finished' not in [e['type'] for e in runtime.events]
+    assert 'secret-key' not in json.dumps(runtime.events)
+    loop._execute_guarded_goal.assert_awaited_once()

tests/test_native_job_attribution.py

@@ -263,3 +263,42 @@ async def test_selection_preserves_existing_instance_wrapper_and_ordinary_cleanu
     assert await loop.root_provider.provider.complete(None) == "wrapped"
     assert len(cleanups) == 1 and cleanups[0].__name__ == "close"
     await cleanups[0]()
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("options", [{}, {"timeout": 7.25}, {"timeout": None}])
+@pytest.mark.parametrize("outcome", ["success", "error", "cancel"])
+async def test_native_boundary_forwards_options_and_restores_scope(monkeypatch, options, outcome):
+    from amplifier_module_provider_openai.native import NativeResponsesProvider
+
+    provider = BundleAstraProvider.wrap(
+        OpenAIProvider(api_key="fixture", config={"default_model": "gpt-6-astra"})
+    )
+    params, result, previous = {}, object(), object()
+    seen = []
+
+    async def native_response(self, received, **kwargs):
+        seen.append((self, received, kwargs, LOOP_REQUEST.get()))
+        if outcome == "error":
+            raise ValueError("fixture failure")
+        if outcome == "cancel":
+            raise asyncio.CancelledError
+        return result
+
+    monkeypatch.setattr(NativeResponsesProvider, "_native_response", native_response)
+    loop_token = LOOP_REQUEST.set(previous)
+    provider_token = PROVIDER_REQUEST.set(provider)
+    try:
+        if outcome == "success":
+            assert await provider._native_response(params, **options) is result
+        else:
+            with pytest.raises(ValueError if outcome == "error" else asyncio.CancelledError):
+                await provider._native_response(params, **options)
+        assert len(seen) == 1
+        assert seen[0][0] is provider and seen[0][1] is params
+        assert seen[0][2] == options and seen[0][3] is provider
+        assert LOOP_REQUEST.get() is previous
+        assert PROVIDER_REQUEST.get() is provider
+    finally:
+        LOOP_REQUEST.reset(loop_token)
+        PROVIDER_REQUEST.reset(provider_token)