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..09692723734 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,55 @@ _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 _PublicResponseFailure(Exception): + """Mark a vetted failure message as safe for client-visible serialization.""" + + 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, _PublicResponseFailure) 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 @@ -268,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: @@ -285,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." + ) ) @@ -1088,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: @@ -1104,14 +1157,25 @@ 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: - 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 +1229,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, + ex, + ): yield event return @@ -1185,14 +1254,24 @@ 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, + _PublicResponseFailure(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, + _PublicResponseFailure(validation_error), ): yield event return @@ -1208,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: @@ -1219,19 +1300,23 @@ 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: - 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, + save_error, + ): yield event return @@ -1314,11 +1399,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( @@ -1397,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 @@ -1429,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) @@ -1487,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) @@ -1509,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 @@ -1528,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 @@ -1548,7 +1657,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 @@ -1596,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.") @@ -1606,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) @@ -1627,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) @@ -1689,10 +1798,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,9 +1815,9 @@ 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 + request_failure = _PublicResponseFailure(misconfigured) if ( self._history_source == "service" and stored @@ -1744,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 @@ -1759,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: @@ -1768,7 +1874,7 @@ 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( @@ -1808,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.") @@ -1876,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.") @@ -2088,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: @@ -2187,8 +2302,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 +2338,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, _PublicResponseFailure): + message = ex.public_message + else: + message = _PUBLIC_STREAM_FAILURE_MESSAGE yield response_event_stream.emit_failed(message=message, usage=tracker.usage if tracker is not None else None) @@ -2420,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, @@ -2659,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]: @@ -3243,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( @@ -3253,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 b3f033acd60..2f4bc2d48c5 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 ( @@ -115,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) @@ -567,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): @@ -1762,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]] = [] @@ -2971,7 +2998,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 +3218,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 +3361,11 @@ 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_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 async def test_cancellation_is_preserved_when_best_effort_save_fails( @@ -7808,7 +7840,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 +8283,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 +8296,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 +8309,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 +8356,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 +8451,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 +8471,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