diff --git a/adapters/cli/amplifier_loop_live_cli/bundle_astra.py b/adapters/cli/amplifier_loop_live_cli/bundle_astra.py index 0648202..c732da5 100644 --- a/adapters/cli/amplifier_loop_live_cli/bundle_astra.py +++ b/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) diff --git a/amplifier_module_loop_live/failures.py b/amplifier_module_loop_live/failures.py new file mode 100644 index 0000000..5a146d5 --- /dev/null +++ b/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.") diff --git a/amplifier_module_loop_live/orchestrator.py b/amplifier_module_loop_live/orchestrator.py index ac07d60..4460374 100644 --- a/amplifier_module_loop_live/orchestrator.py +++ b/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: diff --git a/tests/test_failures.py b/tests/test_failures.py new file mode 100644 index 0000000..6e228c1 --- /dev/null +++ b/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() diff --git a/tests/test_native_job_attribution.py b/tests/test_native_job_attribution.py index f634638..590bd5e 100644 --- a/tests/test_native_job_attribution.py +++ b/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)