From 66da160c7cd0fb9c8412434c0966c81e918efe7d Mon Sep 17 00:00:00 2001 From: Jose Alvarez Date: Fri, 16 Oct 2026 17:29:29 +0200 Subject: [PATCH 1/2] Python: Sanitize unexpected streaming response failures --- .../_responses.py | 216 +++++++++++++----- .../foundry_hosting/tests/test_responses.py | 97 ++++++-- .../_parsing.py | 25 +- .../tests/hosting_responses/test_parsing.py | 87 ++++++- 4 files changed, 344 insertions(+), 81 deletions(-) diff --git a/python/packages/foundry_hosting/agent_framework_foundry_hosting/_responses.py b/python/packages/foundry_hosting/agent_framework_foundry_hosting/_responses.py index 001ebb3f98b..80afe6e0c99 100644 --- a/python/packages/foundry_hosting/agent_framework_foundry_hosting/_responses.py +++ b/python/packages/foundry_hosting/agent_framework_foundry_hosting/_responses.py @@ -53,6 +53,7 @@ from agent_framework._mcp import _MCP_TOOL_RESULT_HOST_PAYLOAD_KEY # pyright: ignore[reportPrivateUsage] from agent_framework._telemetry import mark_feature_used from agent_framework.exceptions import AgentFrameworkException +from agent_framework.observability import OBSERVABILITY_SETTINGS from azure.ai.agentserver.core import get_request_context from azure.ai.agentserver.responses import ( ResponseContext, @@ -142,11 +143,44 @@ _HOSTED_PROVIDER_USAGE_KEY = "_foundry_provider_usage" _PENDING_CONTAINER_FILE_CITATIONS_KEY = "_foundry_pending_container_file_citations" _CONTAINER_FILE_CITATIONS_KEY = "container_file_citations" +_PUBLIC_STREAM_FAILURE_MESSAGE = "An internal server error occurred." _HistorySource = Literal["agent_server", "agent", "service"] _AGENT_SOURCE_UNSET = object() +class _UnexpectedResponseFailure(Exception): + """Wrap an unexpected response failure so public serialization can redact it.""" + + def __init__(self, error: Exception) -> None: + super().__init__(_PUBLIC_STREAM_FAILURE_MESSAGE) + self.error = error + + +def _log_response_failure(phase: str, response_id: str, exc: BaseException) -> None: + """Log a correlated hosted response failure under the sensitive-data policy.""" + diagnostic_exception = exc.error if isinstance(exc, _UnexpectedResponseFailure) else exc + if OBSERVABILITY_SETTINGS.SENSITIVE_DATA_ENABLED: + logger.error( + "Hosted Responses failure (phase=%s, response_id=%s, error_type=%s).", + phase, + response_id, + type(diagnostic_exception).__name__, + exc_info=( + type(diagnostic_exception), + diagnostic_exception, + diagnostic_exception.__traceback__, + ), + ) + else: + logger.error( + "Hosted Responses failure (phase=%s, response_id=%s, error_type=%s).", + phase, + response_id, + type(diagnostic_exception).__name__, + ) + + @dataclass(frozen=True) class _ContainerFileCitation: container_id: str @@ -325,6 +359,17 @@ def _agent_response_updates(response: AgentResponse[Any], response_id: str) -> l _T = TypeVar("_T") +async def _redact_unexpected_stream(stream: AsyncIterable[_T]) -> AsyncGenerator[_T]: + """Forward a stream while marking unexpected iteration failures for redaction.""" + try: + async for item in stream: + yield item + except _UnexpectedResponseFailure: + raise + except Exception as exc: + raise _UnexpectedResponseFailure(exc) from None + + async def _await_before_signal( operation: Callable[[], Awaitable[_T]], *signals: asyncio.Event ) -> tuple[bool, _T | None]: @@ -1110,8 +1155,13 @@ async def _handle_response( await prepare_response_options(hosted_request, self._prepare_options) validate_request_options(hosted_request.options) except Exception as exc: - logger.error("Failed to prepare hosted Responses request", exc_info=(type(exc), exc, exc.__traceback__)) - for event in self._emit_failure(response_event_stream, None, exc): + for event in self._emit_failure( + context.response_id, + "request preparation", + response_event_stream, + None, + exc, + ): yield event return @@ -1165,8 +1215,13 @@ async def _handle_prepared_response( except AgentFrameworkException as ex: consent_errors_to_emit = consent_url_from_error(ex) if consent_errors_to_emit is None or len(consent_errors_to_emit) == 0: - logger.error("Failed to prepare agent: %s", ex, exc_info=(type(ex), ex, ex.__traceback__)) - for event in self._emit_failure(response_event_stream, None, ex): + for event in self._emit_failure( + context.response_id, + "agent initialization", + response_event_stream, + None, + _UnexpectedResponseFailure(ex), + ): yield event return @@ -1185,14 +1240,26 @@ async def _handle_prepared_response( f"OAuth consent request for tool '{invalid_consent.name}' must include an allowed safe HTTPS " "consent link." ) - logger.error("%s", validation_error) - for event in self._emit_failure(response_event_stream, None, validation_error): + for event in self._emit_failure( + context.response_id, + "OAuth consent validation", + response_event_stream, + None, + validation_error, + public_message=str(validation_error), + ): yield event return if request.get("store") is False: + validation_error = ValueError("OAuth consent continuation requires store=true.") for event in self._emit_failure( - response_event_stream, None, ValueError("OAuth consent continuation requires store=true.") + context.response_id, + "OAuth consent validation", + response_event_stream, + None, + validation_error, + public_message=str(validation_error), ): yield event return @@ -1227,11 +1294,13 @@ async def _handle_prepared_response( if context.conversation_id is not None: await session_storage.set(context.conversation_id, session) except Exception as save_error: - logger.error( - "Failed to persist the Agent Framework session for OAuth consent", - exc_info=(type(save_error), save_error, save_error.__traceback__), - ) - for event in self._emit_failure(response_event_stream, None, save_error): + for event in self._emit_failure( + context.response_id, + "OAuth consent session persistence", + response_event_stream, + None, + _UnexpectedResponseFailure(save_error), + ): yield event return @@ -1314,11 +1383,16 @@ async def _handle_prepared_response( else: yield response_event_stream.emit_completed(usage=tracker.usage) except Exception as ex: - logger.error("Failed to produce response for agent", exc_info=(type(ex), ex, ex.__traceback__)) for event in tracker.close(): yield event - for event in self._emit_failure(response_event_stream, tracker, ex): + for event in self._emit_failure( + context.response_id, + "response production", + response_event_stream, + tracker, + ex, + ): yield event async def _load_request_messages( @@ -1548,7 +1622,7 @@ async def _handle_inner_agent( with suppress(BaseException): await request_messages_task if isinstance(ex, Exception): - logger.error("Failed to prepare state storage: %s", ex, exc_info=(type(ex), ex, ex.__traceback__)) + _log_response_failure("state storage preparation", context.response_id, ex) raise request_failure: Exception | None = None @@ -1650,11 +1724,13 @@ async def _handle_inner_agent( emitted_output_count=output_count, ) async with aclosing(responses): - async for response, output_count in responses: + async for response, output_count in _redact_unexpected_stream(responses): for update in _agent_response_updates(response, context.response_id): if context.shutdown.is_set() or cancellation_signal.is_set(): break - async for event in tracker.handle_update(update, approval_storage=approval_storage): + async for event in _redact_unexpected_stream( + tracker.handle_update(update, approval_storage=approval_storage) + ): yield event if context.shutdown.is_set() or cancellation_signal.is_set(): continue @@ -1665,20 +1741,28 @@ async def _handle_inner_agent( if self._resilient_background: yield response_event_stream.checkpoint() else: - inner_stream = agent.run(stream=True, **run_kwargs) # type: ignore[reportUnknownMemberType] + try: + inner_stream = agent.run(stream=True, **run_kwargs) # type: ignore[reportUnknownMemberType] + except Exception as ex: + raise _UnexpectedResponseFailure(ex) from None updates = _SignalledIterator(inner_stream, context.shutdown, cancellation_signal) async with aclosing(updates): - async for update in updates: + async for update in _redact_unexpected_stream(updates): if not stored and any( content.type in ("function_approval_request", "oauth_consent_request") or content.user_input_request for content in update.contents ): raise ValueError("Approval and user-input continuation requires store=true.") - async for event in tracker.handle_update(update, approval_storage=approval_storage): + async for event in _redact_unexpected_stream( + tracker.handle_update(update, approval_storage=approval_storage) + ): yield event if inner_stream is not None and isinstance(updates, _SignalledIterator) and not updates.signalled: - final = await inner_stream.get_final_response() + try: + final = await inner_stream.get_final_response() + except Exception as ex: + raise _UnexpectedResponseFailure(ex) from None if final.continuation_token is not None and final.finish_reason is None: raise RuntimeError( "The inner agent returned an unfinished provider response; " @@ -1689,10 +1773,7 @@ async def _handle_inner_agent( raise except Exception as ex: request_failure = ex - logger.error( - "Failed to produce response for agent", - exc_info=(type(ex), ex, ex.__traceback__), - ) + _log_response_failure("agent execution", context.response_id, ex) finally: if configuration.hosted_history: session.state.pop(_HOSTED_RESPONSES_HISTORY_SOURCE_ID, None) @@ -1709,7 +1790,7 @@ async def _handle_inner_agent( "The agent's chat client stored this turn server-side despite store=False for the inner model. " "Configure the client to honor store=False, or use history_source='service' with store=true." ) - logger.error("%s", misconfigured) + _log_response_failure("stored output validation", context.response_id, misconfigured) if request_failure is None and not request_interrupted: request_failure = misconfigured if ( @@ -1768,13 +1849,16 @@ async def _handle_inner_agent( message = "Failed to persist the Agent Framework session after an agent failure" else: message = "Failed to persist the Agent Framework session after a successful request" - logger.error(message, exc_info=(type(save_error), save_error, save_error.__traceback__)) + _log_response_failure(message, context.response_id, save_error) if request_failure is not None and save_failure is not None: - raise RuntimeError( + combined_failure = RuntimeError( f"Agent request failed: {str(request_failure) or type(request_failure).__name__}; " f"session persistence also failed: {str(save_failure) or type(save_failure).__name__}" ) + if isinstance(request_failure, _UnexpectedResponseFailure): + raise _UnexpectedResponseFailure(combined_failure) from None + raise combined_failure elif request_failure is not None: raise request_failure elif save_failure is not None: @@ -2062,11 +2146,14 @@ async def _handle_inner_workflow( logger.debug( "Serving recovery request with no prior workflow checkpoint; replaying original input" ) - run_stream = agent.run( - input_messages, - stream=True, - checkpoint_storage=checkpoint_storage, - ) + try: + run_stream = agent.run( + input_messages, + stream=True, + checkpoint_storage=checkpoint_storage, + ) + except Exception as exc: + raise _UnexpectedResponseFailure(exc) from None else: # Determine the latest checkpoint (if any) so we can resume the # workflow's prior state for this turn. The directory is keyed by @@ -2107,17 +2194,21 @@ async def _handle_inner_workflow( # ``run(input_messages, ...)`` call may contain ``function_call_output`` # items (carried as FunctionResult/FunctionApprovalResponse content) # that fulfill them via :meth:`WorkflowAgent._process_pending_requests`. - restore_iter = _SignalledIterator( - agent.run( + try: + restore_stream = agent.run( stream=True, checkpoint_id=latest_checkpoint.checkpoint_id, checkpoint_storage=restore_checkpoint_storage, - ), + ) + except Exception as exc: + raise _UnexpectedResponseFailure(exc) from None + restore_iter = _SignalledIterator( + restore_stream, context.shutdown, cancellation_signal, ) async with aclosing(restore_iter): - async for _ in restore_iter: + async for _ in _redact_unexpected_stream(restore_iter): pass if restore_iter.signalled: if context.shutdown.is_set(): @@ -2130,11 +2221,14 @@ async def _handle_inner_workflow( if cancellation_signal.is_set(): return - run_stream = agent.run( - input_messages, - stream=True, - checkpoint_storage=checkpoint_storage, - ) + try: + run_stream = agent.run( + input_messages, + stream=True, + checkpoint_storage=checkpoint_storage, + ) + except Exception as exc: + raise _UnexpectedResponseFailure(exc) from None workflow_name = agent.workflow.name @@ -2167,7 +2261,7 @@ def snapshot_response( stamp=latest_checkpoint_id if self._resilient_background else None, ) async with aclosing(main_iter): - async for update in main_iter: + async for update in _redact_unexpected_stream(main_iter): if self._resilient_background: # Every update before this one belongs to the stamped checkpoint (or an # earlier one), so the output so far can be snapshotted against it. If the @@ -2176,7 +2270,9 @@ def snapshot_response( for event in snapshot_response(main_iter.stamp): yield event - async for event in tracker.handle_update(update, approval_storage=approval_storage): + async for event in _redact_unexpected_stream( + tracker.handle_update(update, approval_storage=approval_storage) + ): yield event # Cancellation needs no extra action here (the loop above already stopped); shutdown # does, but only if it's what actually stopped the loop, not a natural completion. @@ -2187,8 +2283,8 @@ def snapshot_response( # recovery after this point does not replay the last superstep. for event in snapshot_response(await latest_checkpoint_id()): yield event - except Exception: - logger.exception("Failed to produce response for workflow agent") + except Exception as exc: + _log_response_failure("workflow execution", context.response_id, exc) raise async def _resume_workflow_from_checkpoint( @@ -2223,26 +2319,36 @@ async def _resume_workflow_from_checkpoint( @staticmethod def _emit_failure( + response_id: str, + phase: str, response_event_stream: ResponseEventStream, tracker: _OutputItemTracker | None, ex: BaseException, + *, + public_message: str | None = None, ) -> Generator[ResponseStreamEvent]: - """Yield a terminal ``response.failed`` event for ``ex``. + """Log ``ex`` and yield a terminal ``response.failed`` event. Drains any in-progress streaming output item first so the resulting - SSE stream stays well-formed, then emits ``response.failed`` carrying - the exception's message (falling back to the exception type name when - ``str(ex)`` is empty). Any error raised while draining the tracker is - logged and otherwise ignored so that the original failure is always - what the client sees. + SSE stream stays well-formed, then emits ``response.failed`` carrying a + stable public message unless the caller explicitly supplies a safe + validation message. Any error raised while draining the tracker is + logged and otherwise ignored so that the original failure remains the + terminal failure. """ + _log_response_failure(phase, response_id, ex) if tracker is not None: try: yield from tracker.close() - except Exception: - logger.exception("Error while closing streaming tracker after failure") + except Exception as close_error: + _log_response_failure("failed tracker cleanup", response_id, close_error) tracker.discard_pending_container_file_citations() - message = str(ex) or type(ex).__name__ + if public_message is not None: + message = public_message + elif isinstance(ex, _UnexpectedResponseFailure): + message = _PUBLIC_STREAM_FAILURE_MESSAGE + else: + message = str(ex) or type(ex).__name__ yield response_event_stream.emit_failed(message=message, usage=tracker.usage if tracker is not None else None) diff --git a/python/packages/foundry_hosting/tests/test_responses.py b/python/packages/foundry_hosting/tests/test_responses.py index b3f033acd60..f930b302cea 100644 --- a/python/packages/foundry_hosting/tests/test_responses.py +++ b/python/packages/foundry_hosting/tests/test_responses.py @@ -60,6 +60,7 @@ ) from agent_framework._mcp import _MCP_TOOL_RESULT_HOST_PAYLOAD_KEY # pyright: ignore[reportPrivateUsage] from agent_framework.ag_ui import AgentFrameworkAgent, InMemoryAGUIThreadSnapshotStore +from agent_framework.observability import OBSERVABILITY_SETTINGS from agent_framework.openai import OpenAIChatClient, OpenAIChatOptions, OpenAIContinuationToken from azure.ai.agentserver.core import FoundryAgentRequestContext, get_request_context from azure.ai.agentserver.responses import ( @@ -2631,7 +2632,7 @@ async def collect_events() -> list[Any]: context.shutdown.set() if stage == "submit": events = await asyncio.wait_for(task, timeout=3) - assert "cannot safely retry" in _failure_message(events) + assert _failure_message(events) == "An internal server error occurred." exit_for_recovery.assert_not_awaited() else: with pytest.raises(ResponseExitForRecovery): @@ -2971,7 +2972,7 @@ async def collect(response_id: str) -> list[Any]: failed = await collect("failed-dispatch") blocked = await collect("retry-after-failure") - assert "provider timed out" in _failure_message(failed) + assert _failure_message(failed) == "An internal server error occurred." assert "in-flight turn" in _failure_message(blocked) agent.run.assert_called_once() head = await store.get(conversation_id) @@ -3191,7 +3192,10 @@ async def test_client_that_ignores_disabled_storage_fails_without_saving_session response = await _post(server, input_text="first") assert response.json()["status"] == "failed" - assert "stored this turn server-side" in caplog.text + assert "stored this turn server-side" not in caplog.text + assert "phase=stored output validation" in caplog.text + assert f"response_id={response.json()['id']}" in caplog.text + assert "error_type=RuntimeError" in caplog.text assert await store.get(response.json()["id"]) is None async def test_per_service_call_persistence_preserves_function_loop_history(self) -> None: @@ -3331,9 +3335,10 @@ def failing_run(*_args: Any, **_kwargs: Any) -> ResponseStream[AgentResponseUpda assert event_types.count("response.failed") == 1 assert "response.completed" not in event_types error = failed_events[0]["data"]["response"]["error"] - assert "agent failed" in error["message"] - assert "session storage is full" in error["message"] - assert "Failed to produce response for agent" in caplog.text + assert error["message"] == "An internal server error occurred." + assert "agent failed" not in str(failed_events) + assert "session storage is full" not in str(failed_events) + assert "phase=agent execution" in caplog.text assert "Failed to persist the Agent Framework session after an agent failure" in caplog.text async def test_cancellation_is_preserved_when_best_effort_save_fails( @@ -3936,7 +3941,7 @@ async def test_computer_result_without_screenshot_fails_response(self, stream: b body = resp.json() assert body["status"] == "failed" error = body.get("error") or {} - assert error.get("message") == "A computer result requires a call_id and screenshot." + assert error.get("message") == "An internal server error occurred." async def test_computer_items_with_provider_ids_use_valid_host_ids(self) -> None: call_id = "provider-computer-call" @@ -4502,7 +4507,7 @@ async def test_divergent_or_shrinking_text_snapshot_fails_stream(self, replaceme assert deltas == ["Hello world"] assert done["data"]["text"] == "Hello world" assert events[-1]["event"] == "response.failed" - assert "divergent replacement text snapshot" in events[-1]["data"]["response"]["error"]["message"] + assert events[-1]["data"]["response"]["error"]["message"] == "An internal server error occurred." async def test_container_file_citation_is_consistent_across_stream_events(self) -> None: resp = await _post( @@ -7808,7 +7813,11 @@ async def test_configured_origin_allowlist_rejects_connect_time_consent( assert body["status"] == "failed" assert not any(item["type"] == "oauth_consent_request" for item in body["output"]) - assert "must include an allowed safe HTTPS consent link" in caplog.text + assert "must include an allowed safe HTTPS consent link" in body["error"]["message"] + assert "must include an allowed safe HTTPS consent link" not in caplog.text + assert "phase=OAuth consent validation" in caplog.text + assert f"response_id={body['id']}" in caplog.text + assert "error_type=ValueError" in caplog.text agent.run.assert_not_called() async def test_non_streaming_consent_error_emits_oauth_output_item(self) -> None: @@ -8247,13 +8256,12 @@ async def test_workflow_agent_content_filter_marks_response_incomplete(self) -> class TestResponseFailedSurfacing: """Tests that exceptions raised by the hosted agent are converted into - terminal ``response.failed`` events carrying the exception message, - rather than propagating as 5xx HTTP errors or being replaced by the - orchestrator's generic ``"An internal server error occurred."`` - fallback. + terminal ``response.failed`` events carrying a stable public message, + rather than propagating as 5xx HTTP errors or exposing exception details. """ async def test_non_streaming_run_failure_emits_response_failed(self) -> None: + private_error_detail = "non-stream-secret at C:\\sensitive\\non_stream.py" agent = _make_agent( response=AgentResponse(messages=[Message(role="assistant", contents=[Content.from_text("hi")])]) ) @@ -8261,7 +8269,7 @@ async def test_non_streaming_run_failure_emits_response_failed(self) -> None: def run_failure(*args: Any, **kwargs: Any) -> ResponseStream[AgentResponseUpdate, AgentResponse]: del args, kwargs return ResponseStream( - _raising_updates("non-stream kaboom"), + _raising_updates(private_error_detail), finalizer=AgentResponse.from_updates, ) @@ -8274,12 +8282,15 @@ def run_failure(*args: Any, **kwargs: Any) -> ResponseStream[AgentResponseUpdate body = resp.json() assert body["status"] == "failed" error: dict[str, Any] = body.get("error") or {} - assert error.get("message") == "non-stream kaboom" + assert error.get("message") == "An internal server error occurred." + assert private_error_detail not in resp.text async def test_streaming_run_failure_emits_response_failed(self) -> None: + private_error_detail = "stream-secret at C:\\sensitive\\stream.py" + async def _raise_stream() -> AsyncIterator[AgentResponseUpdate]: yield AgentResponseUpdate(contents=[Content.from_text("partial ")], role="assistant") - raise RuntimeError("stream kaboom") + raise RuntimeError(private_error_detail) agent = _RawAgentMock() agent.id = "test-agent" @@ -8318,7 +8329,9 @@ def run_streaming(*args: Any, **kwargs: Any) -> ResponseStream[AgentResponseUpda assert len(failed) == 1 response_payload: dict[str, Any] = failed[0]["data"].get("response") or {} error: dict[str, Any] = response_payload.get("error") or {} - assert error.get("message") == "stream kaboom" + assert error.get("message") == "An internal server error occurred." + assert response_payload["id"] == events[0]["data"]["response"]["id"] + assert private_error_detail not in resp.text async def test_streaming_run_failure_drains_pending_output_item(self) -> None: """If a streaming output item was open when the failure happens, the @@ -8411,11 +8424,12 @@ async def test_workflow_agent_run_failure_emits_response_failed(self) -> None: agent path. """ workflow_agent = _build_text_workflow_agent("ignored") + private_error_detail = "workflow-secret at C:\\sensitive\\workflow.py" def run_failure(*args: Any, **kwargs: Any) -> ResponseStream[AgentResponseUpdate, AgentResponse]: del args, kwargs return ResponseStream( - _raising_updates("workflow kaboom"), + _raising_updates(private_error_detail), finalizer=AgentResponse.from_updates, ) @@ -8430,7 +8444,52 @@ def run_failure(*args: Any, **kwargs: Any) -> ResponseStream[AgentResponseUpdate body = resp.json() assert body["status"] == "failed" error: dict[str, Any] = body.get("error") or {} - assert error.get("message") == "workflow kaboom" + assert error.get("message") == "An internal server error occurred." + assert private_error_detail not in resp.text + + @pytest.mark.parametrize("sensitive_data", [False, True]) + async def test_failure_diagnostics_respect_sensitive_data( + self, + monkeypatch: pytest.MonkeyPatch, + caplog: pytest.LogCaptureFixture, + sensitive_data: bool, + ) -> None: + private_error_detail = "foundry-diagnostic-secret at C:\\sensitive\\foundry.py" + monkeypatch.setattr(OBSERVABILITY_SETTINGS, "enable_instrumentation", True) + monkeypatch.setattr(OBSERVABILITY_SETTINGS, "enable_sensitive_data", sensitive_data) + agent = _make_agent( + response=AgentResponse(messages=[Message(role="assistant", contents=[Content.from_text("unused")])]) + ) + + def run_failure(*args: Any, **kwargs: Any) -> ResponseStream[AgentResponseUpdate, AgentResponse]: + del args, kwargs + return ResponseStream( + _raising_updates(private_error_detail), + finalizer=AgentResponse.from_updates, + ) + + agent.run = MagicMock(side_effect=run_failure) + server = _make_server(agent) + + with caplog.at_level(logging.ERROR, logger="agent_framework_foundry_hosting._responses"): + resp = await _post(server, input_text="hello", stream=True) + + body = _parse_sse_events(resp.text)[-1]["data"]["response"] + assert body["status"] == "failed" + assert body["error"]["message"] == "An internal server error occurred." + assert private_error_detail not in resp.text + assert f"response_id={body['id']}" in caplog.text + assert "phase=response production" in caplog.text + assert "error_type=RuntimeError" in caplog.text + assert (private_error_detail in caplog.text) is sensitive_data + records = [ + record + for record in caplog.records + if record.name == "agent_framework_foundry_hosting._responses" + and "phase=response production" in record.getMessage() + ] + assert len(records) == 1 + assert bool(records[0].exc_info) is sensitive_data # endregion diff --git a/python/packages/hosting-responses/agent_framework_hosting_responses/_parsing.py b/python/packages/hosting-responses/agent_framework_hosting_responses/_parsing.py index 0c00c438804..88f4655e3ac 100644 --- a/python/packages/hosting-responses/agent_framework_hosting_responses/_parsing.py +++ b/python/packages/hosting-responses/agent_framework_hosting_responses/_parsing.py @@ -14,6 +14,7 @@ from __future__ import annotations import json +import logging import time import uuid import warnings @@ -31,6 +32,7 @@ get_namespaced_value, ) from agent_framework._telemetry import mark_feature_used +from agent_framework.observability import OBSERVABILITY_SETTINGS from agent_framework_hosting import AgentRunArgs from openai.types.responses import ( Response as OpenAIResponse, @@ -51,8 +53,11 @@ from ._feature_usage import FeatureIndex +logger = logging.getLogger(__name__) + _MODEL_OUTPUT_KIND_KEY = "model_output_kind" _MODEL_OUTPUT_REFUSAL = "refusal" +_PUBLIC_STREAM_FAILURE_MESSAGE = "An internal server error occurred." _RESPONSE_OUTPUT_ITEM_ADAPTER: TypeAdapter[Any] = TypeAdapter(ResponseOutputItem) _RESPONSE_USAGE_ADAPTER: TypeAdapter[Any] = TypeAdapter(ResponseUsage) @@ -76,6 +81,23 @@ _STREAMING_TERMINAL_STATUSES = frozenset({"completed", "failed", "incomplete"}) +def _log_stream_failure(response_id: str, exc: Exception) -> None: + """Log a correlated stream failure under the framework's sensitive-data policy.""" + if OBSERVABILITY_SETTINGS.SENSITIVE_DATA_ENABLED: + logger.error( + "Responses stream failed (phase=stream processing, response_id=%s, error_type=%s).", + response_id, + type(exc).__name__, + exc_info=(type(exc), exc, exc.__traceback__), + ) + else: + logger.error( + "Responses stream failed (phase=stream processing, response_id=%s, error_type=%s).", + response_id, + type(exc).__name__, + ) + + def _is_refusal_text_content(content: Content) -> bool: return ( content.type == "text" @@ -1495,6 +1517,7 @@ def apply_streamed_message_ids(response: dict[str, Any]) -> None: terminal_event_type = f"response.{status}" yield event(terminal_event_type, response=payload) except Exception as exc: + _log_stream_failure(response_id, exc) for stream_event in close_message("incomplete"): yield stream_event partial_contents = [ @@ -1513,7 +1536,7 @@ def apply_streamed_message_ids(response: dict[str, Any]) -> None: "metadata": {}, "error": { "code": "server_error", - "message": str(exc), + "message": _PUBLIC_STREAM_FAILURE_MESSAGE, }, } if conversation_id is not None: diff --git a/python/packages/hosting-responses/tests/hosting_responses/test_parsing.py b/python/packages/hosting-responses/tests/hosting_responses/test_parsing.py index 21b560b8143..2d9276beff5 100644 --- a/python/packages/hosting-responses/tests/hosting_responses/test_parsing.py +++ b/python/packages/hosting-responses/tests/hosting_responses/test_parsing.py @@ -5,6 +5,7 @@ from __future__ import annotations import json +import logging import warnings from collections.abc import AsyncIterator, Sequence from types import SimpleNamespace @@ -20,6 +21,7 @@ ResponseStream, UsageDetails, ) +from agent_framework.observability import OBSERVABILITY_SETTINGS from openai.types.responses.response_usage import InputTokensDetails, ResponseUsage import agent_framework_hosting_responses._parsing as parsing_module @@ -1082,7 +1084,7 @@ async def updates() -> AsyncIterator[AgentResponseUpdate]: assert deltas == ["Hello world"] assert done["text"] == "Hello world" assert failed["type"] == "response.failed" - assert "divergent replacement text snapshot" in cast(str, error["message"]) + assert error == {"code": "server_error", "message": "An internal server error occurred."} async def test_responses_from_streaming_run_preserves_mixed_text_and_refusal_parts(self) -> None: marker = {"model_output_kind": "refusal"} @@ -1437,9 +1439,11 @@ async def updates() -> AsyncIterator[AgentResponseUpdate]: assert output[0]["code"] == "print('done')" async def test_responses_from_streaming_run_emits_failed_when_iteration_raises(self) -> None: + private_error_detail = "iterator-secret at C:\\sensitive\\iterator.py" + async def updates() -> AsyncIterator[AgentResponseUpdate]: yield AgentResponseUpdate(contents=[Content.from_text("partial")], role="assistant") - raise RuntimeError("upstream blew up") + raise RuntimeError(private_error_detail) stream = ResponseStream(updates(), finalizer=AgentResponse.from_updates) @@ -1460,9 +1464,12 @@ async def updates() -> AsyncIterator[AgentResponseUpdate]: error = cast("dict[str, object]", response["error"]) assert payload["type"] == "response.failed" assert response["status"] == "failed" + assert response["id"] == "resp_new" assert response["conversation"] == {"id": "conv_1"} - assert error["message"] == "upstream blew up" + assert error["code"] == "server_error" + assert error["message"] == "An internal server error occurred." assert "partial" in events[-1] + assert private_error_detail not in "".join(events) async def test_failed_stream_preserves_partial_marked_refusal(self) -> None: async def updates() -> AsyncIterator[AgentResponseUpdate]: @@ -1576,14 +1583,16 @@ def finalizer(items: Sequence[AgentResponseUpdate]) -> AgentResponse: assert events[-1].startswith("event: response.failed") assert response["status"] == "failed" - assert f"unsupported status {status!r}" in cast(str, error["message"]) + assert error == {"code": "server_error", "message": "An internal server error occurred."} async def test_responses_from_streaming_run_emits_failed_when_finalizer_raises(self) -> None: + private_error_detail = "finalizer-secret at C:\\sensitive\\finalizer.py" + async def updates() -> AsyncIterator[AgentResponseUpdate]: yield AgentResponseUpdate(contents=[Content.from_text("partial")], role="assistant") def finalizer(items: Sequence[AgentResponseUpdate]) -> AgentResponse: - raise RuntimeError("finalizer blew up") + raise RuntimeError(private_error_detail) stream = ResponseStream(updates(), finalizer=finalizer) @@ -1596,4 +1605,70 @@ def finalizer(items: Sequence[AgentResponseUpdate]) -> AgentResponse: response = cast("dict[str, object]", payload["response"]) error = cast("dict[str, object]", response["error"]) assert response["status"] == "failed" - assert error["message"] == "finalizer blew up" + assert error == {"code": "server_error", "message": "An internal server error occurred."} + assert private_error_detail not in "".join(events) + + async def test_responses_from_streaming_run_sanitizes_projection_failures( + self, + monkeypatch: pytest.MonkeyPatch, + ) -> None: + private_error_detail = "projection-secret at C:\\sensitive\\projection.py" + + def fail_projection(content: Content) -> tuple[str, str] | None: + del content + raise RuntimeError(private_error_detail) + + monkeypatch.setattr(parsing_module, "_code_interpreter_projection_key", fail_projection) + + async def updates() -> AsyncIterator[AgentResponseUpdate]: + yield AgentResponseUpdate(contents=[Content.from_text("partial")], role="assistant") + yield AgentResponseUpdate( + contents=[Content.from_function_call("call_1", "lookup", arguments={})], + role="assistant", + ) + + stream = ResponseStream(updates(), finalizer=AgentResponse.from_updates) + + events = [event async for event in responses_from_streaming_run(stream, response_id="resp_projection")] + + assert _sse_types(events)[-1] == "response.failed" + assert "response.output_text.delta" in _sse_types(events) + payload = _sse_payload(events[-1]) + response = cast("dict[str, object]", payload["response"]) + assert response["id"] == "resp_projection" + assert response["error"] == {"code": "server_error", "message": "An internal server error occurred."} + assert "partial" in events[-1] + assert private_error_detail not in "".join(events) + + @pytest.mark.parametrize("sensitive_data", [False, True]) + async def test_responses_from_streaming_run_failure_diagnostics_respect_sensitive_data( + self, + monkeypatch: pytest.MonkeyPatch, + caplog: pytest.LogCaptureFixture, + sensitive_data: bool, + ) -> None: + private_error_detail = "diagnostic-secret at C:\\sensitive\\diagnostic.py" + monkeypatch.setattr(OBSERVABILITY_SETTINGS, "enable_instrumentation", True) + monkeypatch.setattr(OBSERVABILITY_SETTINGS, "enable_sensitive_data", sensitive_data) + + async def updates() -> AsyncIterator[AgentResponseUpdate]: + yield AgentResponseUpdate(contents=[Content.from_text("partial")], role="assistant") + raise RuntimeError(private_error_detail) + + stream = ResponseStream(updates(), finalizer=AgentResponse.from_updates) + + with caplog.at_level(logging.ERROR, logger="agent_framework_hosting_responses._parsing"): + events = [event async for event in responses_from_streaming_run(stream, response_id="resp_diagnostics")] + + assert private_error_detail not in "".join(events) + assert "response_id=resp_diagnostics" in caplog.text + assert "error_type=RuntimeError" in caplog.text + assert (private_error_detail in caplog.text) is sensitive_data + records = [ + record + for record in caplog.records + if record.name == "agent_framework_hosting_responses._parsing" + and "response_id=resp_diagnostics" in record.getMessage() + ] + assert len(records) == 1 + assert bool(records[0].exc_info) is sensitive_data From 4862c66b90402aa0b3ad5b6f6a9e60776d52dd07 Mon Sep 17 00:00:00 2001 From: Jose Alvarez Date: Fri, 16 Oct 2026 17:41:16 +0200 Subject: [PATCH 2/2] Python: Preserve vetted hosting validation messages --- .../_responses.py | 293 ++++++++++-------- .../foundry_hosting/tests/test_responses.py | 37 ++- 2 files changed, 192 insertions(+), 138 deletions(-) diff --git a/python/packages/foundry_hosting/agent_framework_foundry_hosting/_responses.py b/python/packages/foundry_hosting/agent_framework_foundry_hosting/_responses.py index 80afe6e0c99..09692723734 100644 --- a/python/packages/foundry_hosting/agent_framework_foundry_hosting/_responses.py +++ b/python/packages/foundry_hosting/agent_framework_foundry_hosting/_responses.py @@ -149,17 +149,28 @@ _AGENT_SOURCE_UNSET = object() -class _UnexpectedResponseFailure(Exception): - """Wrap an unexpected response failure so public serialization can redact it.""" +class _PublicResponseFailure(Exception): + """Mark a vetted failure message as safe for client-visible serialization.""" - def __init__(self, error: Exception) -> None: - super().__init__(_PUBLIC_STREAM_FAILURE_MESSAGE) + def __init__(self, error: Exception, public_message: str | None = None) -> None: + super().__init__(public_message or str(error) or type(error).__name__) self.error = error + self.public_message = public_message or str(error) or type(error).__name__ + + +async def _set_agent_session(store: SessionStore, session_id: str, session: AgentSession) -> None: + """Persist a session while preserving the established public conflict contract.""" + try: + await store.set(session_id, session) + except RuntimeError as exc: + if str(exc) == "Another request advanced this agent session; reload before writing.": + raise _PublicResponseFailure(exc) from None + raise def _log_response_failure(phase: str, response_id: str, exc: BaseException) -> None: """Log a correlated hosted response failure under the sensitive-data policy.""" - diagnostic_exception = exc.error if isinstance(exc, _UnexpectedResponseFailure) else exc + diagnostic_exception = exc.error if isinstance(exc, _PublicResponseFailure) else exc if OBSERVABILITY_SETTINGS.SENSITIVE_DATA_ENABLED: logger.error( "Hosted Responses failure (phase=%s, response_id=%s, error_type=%s).", @@ -302,7 +313,7 @@ def _validate_checkpoint_context_id(context_id: str) -> None: or os.path.isabs(context_id) or os.path.splitdrive(context_id)[0] ): - raise RuntimeError(f"Invalid context id: {context_id!r}") + raise _PublicResponseFailure(RuntimeError(f"Invalid context id: {context_id!r}")) def _is_hosted_responses_history_sentinel(provider: ContextProvider) -> bool: @@ -319,10 +330,12 @@ def _is_hosted_responses_history_sentinel(provider: ContextProvider) -> bool: def _reject_busy_conversation(session: AgentSession) -> None: if session.state.get(_HOSTED_CONVERSATION_CLAIM_KEY) is not None: - raise RuntimeError( - "A service-backed conversation already has an in-flight turn. Wait for it to finish; " - "if it failed or was cancelled after dispatch, start a new conversation rather than " - "reusing a possibly changed provider thread." + raise _PublicResponseFailure( + RuntimeError( + "A service-backed conversation already has an in-flight turn. Wait for it to finish; " + "if it failed or was cancelled after dispatch, start a new conversation rather than " + "reusing a possibly changed provider thread." + ) ) @@ -359,17 +372,6 @@ def _agent_response_updates(response: AgentResponse[Any], response_id: str) -> l _T = TypeVar("_T") -async def _redact_unexpected_stream(stream: AsyncIterable[_T]) -> AsyncGenerator[_T]: - """Forward a stream while marking unexpected iteration failures for redaction.""" - try: - async for item in stream: - yield item - except _UnexpectedResponseFailure: - raise - except Exception as exc: - raise _UnexpectedResponseFailure(exc) from None - - async def _await_before_signal( operation: Callable[[], Awaitable[_T]], *signals: asyncio.Event ) -> tuple[bool, _T | None]: @@ -1133,15 +1135,21 @@ async def _handle_response( terminal_event: ResponseStreamEvent | None = None try: - scope = FoundryRequestScope.from_context( - self.config, get_request_context(), local_session_id=context.response_id - ) + try: + scope = FoundryRequestScope.from_context( + self.config, get_request_context(), local_session_id=context.response_id + ) + except (TypeError, ValueError, RuntimeError) as exc: + raise _PublicResponseFailure(exc) from None if self._agent_source is None: raise RuntimeError("The hosted agent source is not configured.") agent = await resolve_agent(self._agent_source) - configuration = self._configuration or _validate_agent_configuration( - agent, self._history_source, self._host_options, background_source=self._background_source - ) + try: + configuration = self._configuration or _validate_agent_configuration( + agent, self._history_source, self._host_options, background_source=self._background_source + ) + except (TypeError, ValueError, RuntimeError) as exc: + raise _PublicResponseFailure(exc) from None if self._configuration is None: _initialize_agent_history(agent, configuration) if configuration.workflow: @@ -1149,11 +1157,17 @@ async def _handle_response( hosted_request: HostedResponseRequest | None = None if configuration.workflow: if self._prepare_options is not None: - raise ValueError("prepare_options is only supported for regular agents.") + raise _PublicResponseFailure(ValueError("prepare_options is only supported for regular agents.")) else: - hosted_request = HostedResponseRequest(request, context, scope, response_run_options(request)) + try: + hosted_request = HostedResponseRequest(request, context, scope, response_run_options(request)) + except (TypeError, ValueError) as exc: + raise _PublicResponseFailure(exc) from None await prepare_response_options(hosted_request, self._prepare_options) - validate_request_options(hosted_request.options) + try: + validate_request_options(hosted_request.options) + except (TypeError, ValueError) as exc: + raise _PublicResponseFailure(exc) from None except Exception as exc: for event in self._emit_failure( context.response_id, @@ -1220,7 +1234,7 @@ async def _handle_prepared_response( "agent initialization", response_event_stream, None, - _UnexpectedResponseFailure(ex), + ex, ): yield event return @@ -1245,8 +1259,7 @@ async def _handle_prepared_response( "OAuth consent validation", response_event_stream, None, - validation_error, - public_message=str(validation_error), + _PublicResponseFailure(validation_error), ): yield event return @@ -1258,8 +1271,7 @@ async def _handle_prepared_response( "OAuth consent validation", response_event_stream, None, - validation_error, - public_message=str(validation_error), + _PublicResponseFailure(validation_error), ): yield event return @@ -1275,9 +1287,11 @@ async def _handle_prepared_response( session = await session_storage.get(session_load_id) if session_load_id is not None else None if session is None: if previous_response_id is not None and context.conversation_id is None: - raise RuntimeError( - "Cannot find an existing agent session for " - f"previous_response_id={previous_response_id}." + raise _PublicResponseFailure( + RuntimeError( + "Cannot find an existing agent session for " + f"previous_response_id={previous_response_id}." + ) ) session = agent.create_session() if context.conversation_id is not None: @@ -1286,20 +1300,22 @@ async def _handle_prepared_response( if session.service_session_id is not None and session.state.get( _HOSTED_SOURCE_CONVERSATION_KEY ): - raise ValueError("A service-managed downstream conversation cannot be forked.") + raise _PublicResponseFailure( + ValueError("A service-managed downstream conversation cannot be forked.") + ) session.state.pop(_HOSTED_SOURCE_CONVERSATION_KEY, None) if context.conversation_id is not None: session.state[_HOSTED_SOURCE_CONVERSATION_KEY] = context.conversation_id - await session_storage.set(context.response_id, session) + await _set_agent_session(session_storage, context.response_id, session) if context.conversation_id is not None: - await session_storage.set(context.conversation_id, session) + await _set_agent_session(session_storage, context.conversation_id, session) except Exception as save_error: for event in self._emit_failure( context.response_id, "OAuth consent session persistence", response_event_stream, None, - _UnexpectedResponseFailure(save_error), + save_error, ): yield event return @@ -1471,15 +1487,20 @@ async def _handle_inner_agent( conversation_already_committed = False try: if isinstance(agent, RawAgent): - validate_default_transport_options( - agent.default_options, - allow_agent_store=self._history_source == "agent" and stored, - ) + try: + validate_default_transport_options( + agent.default_options, + allow_agent_store=self._history_source == "agent" and stored, + ) + except (TypeError, ValueError) as exc: + raise _PublicResponseFailure(exc) from None if not stored: if not isinstance(agent, RawAgent): - raise RuntimeError( - "store=false cannot guarantee that a custom agent will disable inner storage; " - "use a RawAgent or store=true." + raise _PublicResponseFailure( + RuntimeError( + "store=false cannot guarantee that a custom agent will disable inner storage; " + "use a RawAgent or store=true." + ) ) continuation_defaults = [ name @@ -1503,9 +1524,11 @@ async def _handle_inner_agent( and (provider.store_inputs or provider.store_context_messages or provider.store_outputs) for provider in agent.context_providers ): - raise RuntimeError( - "store=false cannot prevent an external HistoryProvider from persisting responses. " - "Use an InMemoryHistoryProvider or store=true." + raise _PublicResponseFailure( + RuntimeError( + "store=false cannot prevent an external HistoryProvider from persisting responses. " + "Use an InMemoryHistoryProvider or store=true." + ) ) if self._history_source == "agent": stores_by_default = getattr(cast(Any, agent).client, "STORES_BY_DEFAULT", None) @@ -1561,10 +1584,14 @@ async def _handle_inner_agent( ) if session is None: if context.is_recovery and provider_background: - raise RuntimeError("Provider background was interrupted before its continuation token was stored.") + raise _PublicResponseFailure( + RuntimeError("Provider background was interrupted before its continuation token was stored.") + ) if stored and previous_response_id is not None and context.conversation_id is None: - raise RuntimeError( - f"Cannot find an existing agent session for previous_response_id={previous_response_id}." + raise _PublicResponseFailure( + RuntimeError( + f"Cannot find an existing agent session for previous_response_id={previous_response_id}." + ) ) session = agent.create_session() provider_state = session.state.get(_HOSTED_PROVIDER_STATE_KEY) @@ -1583,16 +1610,20 @@ async def _handle_inner_agent( if not conversation_already_committed and ( head is None or head.state.get(_HOSTED_CONVERSATION_CLAIM_KEY) != context.response_id ): - raise RuntimeError( - "Cannot recover provider background: the service-backed conversation claim " - "is no longer held by this response." + raise _PublicResponseFailure( + RuntimeError( + "Cannot recover provider background: the service-backed conversation claim " + "is no longer held by this response." + ) ) else: _reject_busy_conversation(session) if not stored and self._history_source == "agent" and session.service_session_id is not None: - raise ValueError( - "store=false cannot continue agent-managed downstream service history; " - "start a new one-shot request or use store=true." + raise _PublicResponseFailure( + ValueError( + "store=false cannot continue agent-managed downstream service history; " + "start a new one-shot request or use store=true." + ) ) if ( not context.is_recovery @@ -1602,17 +1633,21 @@ async def _handle_inner_agent( or cast(Mapping[str, Any], provider_state).get("completed") is not True ) ): - raise ValueError("A provider background response must complete before the next turn.") + raise _PublicResponseFailure( + ValueError("A provider background response must complete before the next turn.") + ) if previous_response_id is not None and context.conversation_id is None and not context.is_recovery: if session.service_session_id is not None and session.state.get(_HOSTED_SOURCE_CONVERSATION_KEY): - raise ValueError("A service-managed downstream conversation cannot be forked.") + raise _PublicResponseFailure( + ValueError("A service-managed downstream conversation cannot be forked.") + ) if ( stored and self._history_source in ("service", "agent") and session.service_session_id is not None and session.state.get(_HOSTED_SERVICE_CHILD_KEY) ): - raise ValueError("A service-managed downstream response cannot be forked.") + raise _PublicResponseFailure(ValueError("A service-managed downstream response cannot be forked.")) except BaseException as ex: # Session preparation failed (or the request was cancelled / the stream closed — # neither of which is an Exception). Cancel and drain the in-flight message-loading @@ -1670,7 +1705,7 @@ async def _handle_inner_agent( run_kwargs["options"] = chat_options elif are_options_set: if self._unsupported_options == "error": - raise TypeError("The hosted agent does not accept caller runtime options.") + raise _PublicResponseFailure(TypeError("The hosted agent does not accept caller runtime options.")) if self._unsupported_options == "warn": logger.warning("Agent doesn't support runtime options. They will be ignored.") @@ -1680,7 +1715,7 @@ async def _handle_inner_agent( raise RuntimeError("Service history requires agent session storage.") session.state[_HOSTED_SERVICE_CHILD_KEY] = context.response_id try: - await session_storage.set(previous_response_id, session) + await _set_agent_session(session_storage, previous_response_id, session) finally: session.state.pop(_HOSTED_SERVICE_CHILD_KEY, None) session.state.pop(_HOSTED_SOURCE_CONVERSATION_KEY, None) @@ -1701,7 +1736,7 @@ async def _handle_inner_agent( # mutate its linear thread; a losing turn never reaches agent.run(). session.state[_HOSTED_CONVERSATION_CLAIM_KEY] = context.response_id try: - await session_storage.set(context.conversation_id, session) + await _set_agent_session(session_storage, context.conversation_id, session) finally: session.state.pop(_HOSTED_CONVERSATION_CLAIM_KEY, None) @@ -1724,13 +1759,11 @@ async def _handle_inner_agent( emitted_output_count=output_count, ) async with aclosing(responses): - async for response, output_count in _redact_unexpected_stream(responses): + async for response, output_count in responses: for update in _agent_response_updates(response, context.response_id): if context.shutdown.is_set() or cancellation_signal.is_set(): break - async for event in _redact_unexpected_stream( - tracker.handle_update(update, approval_storage=approval_storage) - ): + async for event in tracker.handle_update(update, approval_storage=approval_storage): yield event if context.shutdown.is_set() or cancellation_signal.is_set(): continue @@ -1741,28 +1774,20 @@ async def _handle_inner_agent( if self._resilient_background: yield response_event_stream.checkpoint() else: - try: - inner_stream = agent.run(stream=True, **run_kwargs) # type: ignore[reportUnknownMemberType] - except Exception as ex: - raise _UnexpectedResponseFailure(ex) from None + inner_stream = agent.run(stream=True, **run_kwargs) # type: ignore[reportUnknownMemberType] updates = _SignalledIterator(inner_stream, context.shutdown, cancellation_signal) async with aclosing(updates): - async for update in _redact_unexpected_stream(updates): + async for update in updates: if not stored and any( content.type in ("function_approval_request", "oauth_consent_request") or content.user_input_request for content in update.contents ): raise ValueError("Approval and user-input continuation requires store=true.") - async for event in _redact_unexpected_stream( - tracker.handle_update(update, approval_storage=approval_storage) - ): + async for event in tracker.handle_update(update, approval_storage=approval_storage): yield event if inner_stream is not None and isinstance(updates, _SignalledIterator) and not updates.signalled: - try: - final = await inner_stream.get_final_response() - except Exception as ex: - raise _UnexpectedResponseFailure(ex) from None + final = await inner_stream.get_final_response() if final.continuation_token is not None and final.finish_reason is None: raise RuntimeError( "The inner agent returned an unfinished provider response; " @@ -1792,7 +1817,7 @@ async def _handle_inner_agent( ) _log_response_failure("stored output validation", context.response_id, misconfigured) if request_failure is None and not request_interrupted: - request_failure = misconfigured + request_failure = _PublicResponseFailure(misconfigured) if ( self._history_source == "service" and stored @@ -1825,7 +1850,7 @@ async def _handle_inner_agent( if stored and session_storage is not None and not stored_output_violation and final_provider_state: if context.conversation_id is not None: session.state[_HOSTED_SOURCE_CONVERSATION_KEY] = context.conversation_id - await session_storage.set(context.response_id, session) + await _set_agent_session(session_storage, context.response_id, session) if ( context.conversation_id is not None and not conversation_already_committed @@ -1840,7 +1865,7 @@ async def _handle_inner_agent( if provider_background: session.state.pop(_HOSTED_PROVIDER_STATE_KEY, None) session.state[_HOSTED_CONVERSATION_COMMITTED_KEY] = context.response_id - await session_storage.set(context.conversation_id, session) + await _set_agent_session(session_storage, context.conversation_id, session) except Exception as save_error: save_failure = save_error if request_interrupted: @@ -1852,13 +1877,10 @@ async def _handle_inner_agent( _log_response_failure(message, context.response_id, save_error) if request_failure is not None and save_failure is not None: - combined_failure = RuntimeError( + raise RuntimeError( f"Agent request failed: {str(request_failure) or type(request_failure).__name__}; " f"session persistence also failed: {str(save_failure) or type(save_failure).__name__}" ) - if isinstance(request_failure, _UnexpectedResponseFailure): - raise _UnexpectedResponseFailure(combined_failure) from None - raise combined_failure elif request_failure is not None: raise request_failure elif save_failure is not None: @@ -1892,8 +1914,10 @@ async def proceed(*, has_token: bool) -> bool: await context.exit_for_recovery() if has_token: raise RuntimeError("Provider background recovery requires resilient_background=True.") - raise RuntimeError( - "Provider background submission stopped before its token was saved; cannot safely retry it." + raise _PublicResponseFailure( + RuntimeError( + "Provider background submission stopped before its token was saved; cannot safely retry it." + ) ) if cancellation_signal.is_set(): raise RuntimeError("Provider background was interrupted without a client cancellation.") @@ -1960,10 +1984,14 @@ async def pending_outputs() -> AsyncGenerator[tuple[AgentResponse[Any], int]]: if context.is_recovery: saved = session.state.get(_HOSTED_PROVIDER_STATE_KEY) if not isinstance(saved, Mapping): - raise RuntimeError("Cannot recover a provider background job without its stored continuation token.") + raise _PublicResponseFailure( + RuntimeError("Cannot recover a provider background job without its stored continuation token.") + ) saved_payload = cast(Mapping[str, Any], saved) if saved_payload.get("outer_response_id") != context.response_id: - raise RuntimeError("Cannot recover a provider background job without its stored continuation token.") + raise _PublicResponseFailure( + RuntimeError("Cannot recover a provider background job without its stored continuation token.") + ) token = saved_payload.get("continuation_token") if not isinstance(token, Mapping): raise RuntimeError("The stored provider continuation token is invalid.") @@ -2146,14 +2174,11 @@ async def _handle_inner_workflow( logger.debug( "Serving recovery request with no prior workflow checkpoint; replaying original input" ) - try: - run_stream = agent.run( - input_messages, - stream=True, - checkpoint_storage=checkpoint_storage, - ) - except Exception as exc: - raise _UnexpectedResponseFailure(exc) from None + run_stream = agent.run( + input_messages, + stream=True, + checkpoint_storage=checkpoint_storage, + ) else: # Determine the latest checkpoint (if any) so we can resume the # workflow's prior state for this turn. The directory is keyed by @@ -2175,8 +2200,11 @@ async def _handle_inner_workflow( if latest_checkpoint is None and previous_response_id is not None: # A previous_response_id must have a prior workflow checkpoint to resume from - raise RuntimeError( - f"Cannot find an existing workflow checkpoint for previous_response_id={previous_response_id}." + raise _PublicResponseFailure( + RuntimeError( + "Cannot find an existing workflow checkpoint for " + f"previous_response_id={previous_response_id}." + ) ) if latest_checkpoint is not None: @@ -2194,21 +2222,17 @@ async def _handle_inner_workflow( # ``run(input_messages, ...)`` call may contain ``function_call_output`` # items (carried as FunctionResult/FunctionApprovalResponse content) # that fulfill them via :meth:`WorkflowAgent._process_pending_requests`. - try: - restore_stream = agent.run( + restore_iter = _SignalledIterator( + agent.run( stream=True, checkpoint_id=latest_checkpoint.checkpoint_id, checkpoint_storage=restore_checkpoint_storage, - ) - except Exception as exc: - raise _UnexpectedResponseFailure(exc) from None - restore_iter = _SignalledIterator( - restore_stream, + ), context.shutdown, cancellation_signal, ) async with aclosing(restore_iter): - async for _ in _redact_unexpected_stream(restore_iter): + async for _ in restore_iter: pass if restore_iter.signalled: if context.shutdown.is_set(): @@ -2221,14 +2245,11 @@ async def _handle_inner_workflow( if cancellation_signal.is_set(): return - try: - run_stream = agent.run( - input_messages, - stream=True, - checkpoint_storage=checkpoint_storage, - ) - except Exception as exc: - raise _UnexpectedResponseFailure(exc) from None + run_stream = agent.run( + input_messages, + stream=True, + checkpoint_storage=checkpoint_storage, + ) workflow_name = agent.workflow.name @@ -2261,7 +2282,7 @@ def snapshot_response( stamp=latest_checkpoint_id if self._resilient_background else None, ) async with aclosing(main_iter): - async for update in _redact_unexpected_stream(main_iter): + async for update in main_iter: if self._resilient_background: # Every update before this one belongs to the stamped checkpoint (or an # earlier one), so the output so far can be snapshotted against it. If the @@ -2270,9 +2291,7 @@ def snapshot_response( for event in snapshot_response(main_iter.stamp): yield event - async for event in _redact_unexpected_stream( - tracker.handle_update(update, approval_storage=approval_storage) - ): + async for event in tracker.handle_update(update, approval_storage=approval_storage): yield event # Cancellation needs no extra action here (the loop above already stopped); shutdown # does, but only if it's what actually stopped the loop, not a natural completion. @@ -2345,10 +2364,10 @@ def _emit_failure( tracker.discard_pending_container_file_citations() if public_message is not None: message = public_message - elif isinstance(ex, _UnexpectedResponseFailure): - message = _PUBLIC_STREAM_FAILURE_MESSAGE + elif isinstance(ex, _PublicResponseFailure): + message = ex.public_message else: - message = str(ex) or type(ex).__name__ + message = _PUBLIC_STREAM_FAILURE_MESSAGE yield response_event_stream.emit_failed(message=message, usage=tracker.usage if tracker is not None else None) @@ -2526,7 +2545,9 @@ def _consume_text_update(self, text: str, merge_mode: Literal["append", "replace if not previous: self._accumulated[:] = [text] return text - raise RuntimeError("Cannot project a divergent replacement text snapshot onto an append-only Responses stream.") + raise _PublicResponseFailure( + RuntimeError("Cannot project a divergent replacement text snapshot onto an append-only Responses stream.") + ) async def handle_update( self, @@ -2765,7 +2786,7 @@ async def handle( elif content.type == "computer_tool_result": if not content.call_id or content.screenshot is None: - raise ValueError("A computer result requires a call_id and screenshot.") + raise _PublicResponseFailure(ValueError("A computer result requires a call_id and screenshot.")) for event in self._close(): yield event if content.id and IdGenerator.is_valid(content.id)[0]: @@ -3349,7 +3370,10 @@ async def _item_to_message( if item["type"] == "mcp_approval_request": if approval_storage is not None: - function_approval_request_content = await approval_storage.load_approval_request(item["id"]) + try: + function_approval_request_content = await approval_storage.load_approval_request(item["id"]) + except (KeyError, ValueError) as exc: + raise _PublicResponseFailure(exc) from None else: raise ValueError("ApprovalStorage is required to load approval request.") return Message( @@ -3359,9 +3383,12 @@ async def _item_to_message( if item["type"] == "mcp_approval_response": if approval_storage is not None: - function_approval_request_content = await approval_storage.load_approval_request( - item["approval_request_id"] - ) + try: + function_approval_request_content = await approval_storage.load_approval_request( + item["approval_request_id"] + ) + except (KeyError, ValueError) as exc: + raise _PublicResponseFailure(exc) from None else: raise ValueError("ApprovalStorage is required to load approval request.") return Message( diff --git a/python/packages/foundry_hosting/tests/test_responses.py b/python/packages/foundry_hosting/tests/test_responses.py index f930b302cea..2f4bc2d48c5 100644 --- a/python/packages/foundry_hosting/tests/test_responses.py +++ b/python/packages/foundry_hosting/tests/test_responses.py @@ -116,6 +116,7 @@ _OPENAI_HTTPX = cast(Any, import_module(DefaultAsyncHttpxClient.__mro__[1].__module__.partition(".")[0])) _PRIVATE_ERROR_DETAIL = "test-token-value at /srv/private/tool.py" +_SESSION_STORAGE_ERROR_DETAIL = "session-storage-secret at C:\\sensitive\\session_store.py" @pytest.fixture(scope="session", autouse=True) @@ -568,7 +569,7 @@ def __init__(self) -> None: async def set(self, session_id: str, session: AgentSession) -> None: del session_id, session self.set_attempts += 1 - raise OSError("session storage is full") + raise OSError(_SESSION_STORAGE_ERROR_DETAIL) class _ConflictingConversationStore(SessionStore): @@ -1763,6 +1764,31 @@ async def test_hook_cannot_reintroduce_host_owned_storage_or_identity(self) -> N assert "host-controlled" in response.json()["error"]["message"] agent.run.assert_not_called() + async def test_prepare_options_failure_is_sanitized( + self, + caplog: pytest.LogCaptureFixture, + ) -> None: + private_error_detail = "prepare-options-secret at C:\\sensitive\\prepare_options.py" + + def prepare_options(_request: Any, _options: dict[str, Any]) -> dict[str, Any]: + raise RuntimeError(private_error_detail) + + agent = _make_agent() + server = _make_server(agent, prepare_options=prepare_options) + + with caplog.at_level(logging.ERROR, logger="agent_framework_foundry_hosting._responses"): + response = await _post_json(server, {"input": "one shot", "store": False}) + + body = response.json() + assert body["status"] == "failed" + assert body["error"]["message"] == "An internal server error occurred." + assert private_error_detail not in response.text + assert private_error_detail not in caplog.text + assert "phase=request preparation" in caplog.text + assert f"response_id={body['id']}" in caplog.text + assert "error_type=RuntimeError" in caplog.text + agent.run.assert_not_called() + async def test_nested_extra_body_cannot_override_unstored_openai_request(self) -> None: outbound: list[dict[str, Any]] = [] @@ -2632,7 +2658,7 @@ async def collect_events() -> list[Any]: context.shutdown.set() if stage == "submit": events = await asyncio.wait_for(task, timeout=3) - assert _failure_message(events) == "An internal server error occurred." + assert "cannot safely retry" in _failure_message(events) exit_for_recovery.assert_not_awaited() else: with pytest.raises(ResponseExitForRecovery): @@ -3337,7 +3363,8 @@ def failing_run(*_args: Any, **_kwargs: Any) -> ResponseStream[AgentResponseUpda error = failed_events[0]["data"]["response"]["error"] assert error["message"] == "An internal server error occurred." assert "agent failed" not in str(failed_events) - assert "session storage is full" not in str(failed_events) + assert _SESSION_STORAGE_ERROR_DETAIL not in str(failed_events) + assert _SESSION_STORAGE_ERROR_DETAIL not in caplog.text assert "phase=agent execution" in caplog.text assert "Failed to persist the Agent Framework session after an agent failure" in caplog.text @@ -3941,7 +3968,7 @@ async def test_computer_result_without_screenshot_fails_response(self, stream: b body = resp.json() assert body["status"] == "failed" error = body.get("error") or {} - assert error.get("message") == "An internal server error occurred." + assert error.get("message") == "A computer result requires a call_id and screenshot." async def test_computer_items_with_provider_ids_use_valid_host_ids(self) -> None: call_id = "provider-computer-call" @@ -4507,7 +4534,7 @@ async def test_divergent_or_shrinking_text_snapshot_fails_stream(self, replaceme assert deltas == ["Hello world"] assert done["data"]["text"] == "Hello world" assert events[-1]["event"] == "response.failed" - assert events[-1]["data"]["response"]["error"]["message"] == "An internal server error occurred." + assert "divergent replacement text snapshot" in events[-1]["data"]["response"]["error"]["message"] async def test_container_file_citation_is_consistent_across_stream_events(self) -> None: resp = await _post(