diff --git a/docs/specs/002-python-hosting-channels.md b/docs/specs/002-python-hosting-channels.md index 337dcef1f92..a432cfa122b 100644 --- a/docs/specs/002-python-hosting-channels.md +++ b/docs/specs/002-python-hosting-channels.md @@ -215,7 +215,7 @@ The Responses package provides the helper-first surface for OpenAI Responses-sha - `messages_from_responses_input(input) -> list[Message]` - `responses_to_run(body) -> AgentRunArgs` -- `responses_session_id(body) -> str | None` +- `responses_session_id(body) -> tuple[str, bool] | tuple[None, None]` - `create_response_id() -> str` `responses_to_run(...)` returns values corresponding to `Agent.run(...)`: @@ -233,23 +233,36 @@ It excludes protocol transport/session fields from `options` and remaps known Re `responses_session_id(...)` returns: - `previous_response_id` when present (`resp_*`); -- otherwise `conversation_id` when present (`conv_*`); -- otherwise `None`. +- otherwise the id from `conversation` when present as a string or `{ "id": ... }` object (`conv_*`); +- otherwise the id from deprecated `conversation_id`, with a deprecation warning; +- otherwise `(None, None)`. + +The returned boolean identifies conversation-based continuation. Supplying more +than one continuation mechanism is invalid. The helper only extracts the candidate key. App code decides whether to trust and use that key. ### Response helpers -- `responses_from_run(result, *, response_id, session_id=None) -> dict[str, Any]` -- `responses_from_streaming_run(stream, *, response_id, session_id=None) -> AsyncIterator[str]` +- `responses_from_run(result, *, response_id, conversation_id=None) -> dict[str, Any]` +- `responses_from_streaming_run(stream, *, response_id, conversation_id=None) -> AsyncIterator[str]` `responses_from_run(...)` renders a full Responses JSON payload from an `AgentResponse`. It renders the full set of -OpenAI Responses output item types supported by Agent Framework content. +OpenAI Responses output item types supported by Agent Framework content. Transport status is read from the nested raw +Responses representation when available, rather than from free-form `AgentResponse.additional_properties`, where providers +may flatten user metadata. Failed transport responses preserve the structured error from that same raw representation. +A native Responses usage object from that raw representation is preserved when it validates against the installed OpenAI +SDK; raw and Agent Framework counters are never merged. Without valid raw usage, `UsageDetails` counters map only to their +matching Responses fields and the installed SDK validates the resulting version-specific detail shape. Missing counters +are not cross-filled or rendered as invented zeros. An absent total alone is derived as the sum of the known input and +output counts. Explicit zero values are preserved, while usage whose detail or total relationships cannot form a consistent +Responses shape is omitted. `responses_from_streaming_run(...)` renders Server-Sent Event strings for a `ResponseStream`. It emits a created event, -text deltas, and a completed event. The final completed payload is produced through `responses_from_run(...)`; the helper -also preserves the model id observed on streaming updates when the finalized `AgentResponse` no longer carries raw model -metadata. +text deltas, and a terminal event matching the final `completed`, `incomplete`, or `failed` status. A stream that finalizes +with another status is rejected and rendered as `response.failed`. The final payload is produced through +`responses_from_run(...)`; the helper also preserves the model id observed on streaming updates when the finalized +`AgentResponse` no longer carries raw model metadata. ## `agent-framework-hosting-a2a` @@ -451,7 +464,8 @@ state = AgentState(create_agent) @app.post("/responses", response_model=None) async def responses(body: dict = Body(...)) -> JSONResponse | StreamingResponse: run = responses_to_run(body) - candidate_session_id = responses_session_id(body) + candidate_session_id, is_conversation_id = responses_session_id(body) + conversation_id = candidate_session_id if is_conversation_id else None response_id = create_response_id() # Verify this caller owns candidate_session_id before loading it. @@ -468,16 +482,16 @@ async def responses(body: dict = Body(...)) -> JSONResponse | StreamingResponse: async for event in responses_from_streaming_run( stream, response_id=response_id, - session_id=candidate_session_id, + conversation_id=conversation_id, ): yield event - await state.set_session(response_id, session) + await state.set_session(conversation_id or response_id, session) return StreamingResponse(events(), media_type="text/event-stream") result = await target.run(run["messages"], session=session, options=run["options"]) - await state.set_session(response_id, session) - return JSONResponse(responses_from_run(result, response_id=response_id, session_id=candidate_session_id)) + await state.set_session(conversation_id or response_id, session) + return JSONResponse(responses_from_run(result, response_id=response_id, conversation_id=conversation_id)) ``` ## Validation diff --git a/python/packages/hosting-responses/README.md b/python/packages/hosting-responses/README.md index cf97edd7ef1..936bfe1d883 100644 --- a/python/packages/hosting-responses/README.md +++ b/python/packages/hosting-responses/README.md @@ -7,8 +7,8 @@ This package provides the Responses-specific conversion layer: - `responses_to_run(...)` — convert a Responses request body into Agent Framework run values. - `responses_session_id(...)` — return `(session_id, is_conversation_id)` for a - prior `resp_*` response id or `conv_*` conversation id, or `(None, None)` when - neither is present. + prior `resp_*` response id or the `conv_*` id from the official `conversation` + field, or `(None, None)` when neither is present. - `create_conversation_id(...)` — mint a Responses-shaped conversation id. - `create_response_id(...)` — mint a Responses-shaped response id. - `responses_from_run(...)` — convert an `AgentResponse` into a @@ -16,6 +16,18 @@ This package provides the Responses-specific conversion layer: - `responses_from_streaming_run(...)` — convert an Agent Framework `ResponseStream` into Responses-compatible SSE events. +Final streaming events match the rendered response status: +`response.completed`, `response.incomplete`, or `response.failed`. Finalizing a +stream with a nonterminal status produces `response.failed`. Response status is +read from the raw transport representation rather than free-form agent metadata, +and failed transport responses preserve their structured error. A valid native +Responses usage object is preserved before considering Agent Framework counters; +the two sources are never merged. Otherwise, counters map only from matching +Agent Framework fields and the installed OpenAI SDK schema validates the shape. +Missing counters never borrow from another field or become invented zeros; an +absent total alone is derived from known input and output counts. Usage that +cannot form a consistent Responses shape is omitted. + FastAPI/Starlette/Django/Azure Functions code owns route registration, authentication, status codes, response construction, and background work. @@ -56,8 +68,13 @@ async def responses(body: dict = Body(...)) -> JSONResponse: `previous_response_id` identifies an immutable continuation snapshot: multiple requests may branch from it and store their results under distinct new response -ids. `conversation_id` is a mutable head instead; only one caller should -advance it at a time. These helpers do not provide per-conversation locking. +ids. `conversation` accepts either a conversation id string or an `{"id": ...}` +object and identifies a mutable head; only one caller should advance it at a +time. Supplying both mechanisms is invalid. + +The former `conversation_id` request field remains available only as a +deprecated fallback when neither standard mechanism is present. These helpers +do not provide per-conversation locking. `AgentState` lives in [`agent-framework-hosting`](https://pypi.org/project/agent-framework-hosting/). 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 2ced53a629e..3536fc83d52 100644 --- a/python/packages/hosting-responses/agent_framework_hosting_responses/_parsing.py +++ b/python/packages/hosting-responses/agent_framework_hosting_responses/_parsing.py @@ -17,7 +17,7 @@ import time import uuid import warnings -from collections.abc import AsyncIterator, Mapping, Sequence +from collections.abc import AsyncIterator, Iterator, Mapping, Sequence from typing import Any, cast from agent_framework import AgentResponse, AgentResponseUpdate, ChatOptions, Content, Message, ResponseStream @@ -36,11 +36,13 @@ ResponseOutputMessage, ResponseOutputText, ) +from openai.types.responses.response_usage import ResponseUsage from pydantic import TypeAdapter, ValidationError from ._feature_usage import FeatureIndex _RESPONSE_OUTPUT_ITEM_ADAPTER: TypeAdapter[Any] = TypeAdapter(ResponseOutputItem) +_RESPONSE_USAGE_ADAPTER: TypeAdapter[Any] = TypeAdapter(ResponseUsage) # OpenAI Responses field name → Agent Framework ChatOptions field name. _RESPONSES_OPTION_REMAP = { @@ -49,7 +51,17 @@ } # Fields the Responses transport owns; they are consumed separately and must # not also appear in options. -_RESPONSES_RUN_TRANSPORT_KEYS = frozenset({"input", "stream", "previous_response_id", "conversation_id"}) +_RESPONSES_RUN_TRANSPORT_KEYS = frozenset({ + "input", + "stream", + "previous_response_id", + "conversation", + "conversation_id", +}) +_RESPONSES_CONTINUATION_KEYS = ("previous_response_id", "conversation", "conversation_id") +_RESPONSES_INPUT_MESSAGE_ROLES = frozenset({"user", "assistant", "system", "developer"}) +_RESPONSES_STATUSES = frozenset({"completed", "failed", "in_progress", "cancelled", "queued", "incomplete"}) +_STREAMING_TERMINAL_STATUSES = frozenset({"completed", "failed", "incomplete"}) def _content_from_input_item(item: Mapping[str, Any]) -> Content: @@ -62,26 +74,37 @@ def _content_from_input_item(item: Mapping[str, Any]) -> Content: """ item_type = item.get("type") if item_type in ("input_text", "output_text", "text"): - return Content.from_text(text=str(item.get("text", ""))) + text = item.get("text") + if not isinstance(text, str) or not text: + raise ValueError(f"{item_type} requires a non-empty string `text`") + return Content.from_text(text=text) if item_type == "input_image": image_url: Any = item.get("image_url") if isinstance(image_url, Mapping): image_url = cast("Mapping[str, Any]", image_url).get("url") - if not isinstance(image_url, str): - raise ValueError("input_image requires `image_url`") + if not isinstance(image_url, str) or not image_url: + raise ValueError("input_image requires a non-empty string `image_url`") return Content.from_uri(uri=image_url, media_type="image/*") if item_type == "input_file": - if (uri := item.get("file_url")) and isinstance(uri, str): - return Content.from_uri(uri=uri, media_type=item.get("mime_type")) - if file_id := item.get("file_id"): - return Content(type="hosted_file", file_id=str(file_id)) - raise ValueError("input_file requires `file_url` or `file_id`") + file_url = item.get("file_url") + if file_url is not None: + if not isinstance(file_url, str) or not file_url: + raise ValueError("input_file `file_url` must be a non-empty string") + return Content.from_uri(uri=file_url, media_type=item.get("mime_type")) + file_id = item.get("file_id") + if file_id is not None: + if not isinstance(file_id, str) or not file_id: + raise ValueError("input_file `file_id` must be a non-empty string") + return Content(type="hosted_file", file_id=file_id) + raise ValueError("input_file requires a non-empty string `file_url` or `file_id`") raise ValueError(f"Unsupported Responses input content type: {item_type!r}") def messages_from_responses_input(value: Any) -> list[Message]: """Translate ``input`` (string or list of items) into :class:`Message` objects.""" if isinstance(value, str): + if not value: + raise ValueError("`input` must be a non-empty string or list") return [Message("user", [Content.from_text(text=value)])] if not isinstance(value, list) or not value: raise ValueError("`input` must be a non-empty string or list") @@ -101,12 +124,20 @@ def flush() -> None: item_map = cast("Mapping[str, Any]", item) if item_map.get("type") == "message": flush() - role = str(item_map.get("role") or "user") - content: Any = item_map.get("content") or [] + role = item_map.get("role") + if not isinstance(role, str) or role not in _RESPONSES_INPUT_MESSAGE_ROLES: + raise ValueError("message `role` must be one of `user`, `assistant`, `system`, or `developer`") + if "content" not in item_map: + raise ValueError("message requires non-empty `content`") + content: Any = item_map["content"] parts: list[Content] if isinstance(content, str): + if not content: + raise ValueError("message `content` must not be empty") parts = [Content.from_text(text=content)] elif isinstance(content, list): + if not content: + raise ValueError("message `content` must not be empty") parts = [] for content_item in cast("list[Any]", content): if not isinstance(content_item, Mapping): @@ -145,29 +176,63 @@ def responses_session_id(body: Mapping[str, Any]) -> tuple[str, bool] | tuple[No body: OpenAI Responses-shaped request body. Returns: - The session id, if present, and whether it came from ``conversation_id``. + The session id, if present, and whether it came from ``conversation``. The flag is ``None`` when no session id is present. + + Raises: + ValueError: If a continuation field is malformed or conflicts with + another continuation mechanism. """ - previous_response_id = body.get("previous_response_id") - if isinstance(previous_response_id, str) and previous_response_id and not previous_response_id.startswith("resp_"): - warnings.warn( - "`previous_response_id` does not use the OpenAI Responses `resp_` prefix; " - "continuing with the supplied value.", - UserWarning, - stacklevel=2, - ) - conversation_id = body.get("conversation_id") - if isinstance(conversation_id, str) and conversation_id and not conversation_id.startswith("conv_"): + return _responses_session_id(body, warn=True) + + +def _responses_session_id( + body: Mapping[str, Any], + *, + warn: bool, +) -> tuple[str, bool] | tuple[None, None]: + supplied_keys = [key for key in _RESPONSES_CONTINUATION_KEYS if body.get(key) is not None] + if len(supplied_keys) > 1: + raise ValueError("`previous_response_id`, `conversation`, and `conversation_id` are mutually exclusive") + if not supplied_keys: + return None, None + + key = supplied_keys[0] + value = body[key] + if key == "previous_response_id": + if not isinstance(value, str) or not value: + raise ValueError("`previous_response_id` must be a non-empty string") + if warn and not value.startswith("resp_"): + warnings.warn( + "`previous_response_id` does not use the OpenAI Responses `resp_` prefix; " + "continuing with the supplied value.", + UserWarning, + stacklevel=3, + ) + return value, False + + if key == "conversation": + if isinstance(value, Mapping): + value = cast("Mapping[str, Any]", value).get("id") + if not isinstance(value, str) or not value: + raise ValueError("`conversation` must be a non-empty string or an object with a non-empty string `id`") + else: + if not isinstance(value, str) or not value: + raise ValueError("`conversation_id` must be a non-empty string") + if warn: + warnings.warn( + "`conversation_id` is deprecated; use the OpenAI Responses `conversation` field instead.", + DeprecationWarning, + stacklevel=3, + ) + + if warn and not value.startswith("conv_"): warnings.warn( - "`conversation_id` does not use the OpenAI Responses `conv_` prefix; continuing with the supplied value.", + f"`{key}` does not use the OpenAI Responses `conv_` prefix; continuing with the supplied value.", UserWarning, - stacklevel=2, + stacklevel=3, ) - if isinstance(previous_response_id, str) and previous_response_id: - return previous_response_id, False - if isinstance(conversation_id, str) and conversation_id: - return conversation_id, True - return None, None + return value, True def responses_to_run(body: Mapping[str, Any]) -> AgentRunArgs: @@ -180,9 +245,10 @@ def responses_to_run(body: Mapping[str, Any]) -> AgentRunArgs: Arguments corresponding to ``Agent.run``. Raises: - ValueError: If the request body has invalid ``input``. + ValueError: If the request body has invalid ``input`` or continuation fields. """ mark_feature_used(FeatureIndex.HOSTING_RESPONSES) + _responses_session_id(body, warn=False) messages = messages_from_responses_input(body.get("input")) options: dict[str, Any] = {} for key, value in body.items(): @@ -216,24 +282,196 @@ def responses_from_run( Responses-compatible JSON payload. """ mark_feature_used(FeatureIndex.HOSTING_RESPONSES) - output_items = _result_to_output_items(result, status="completed") + status = _status_from_result(result) + output_items = _result_to_output_items(result, status=status) response_kwargs: dict[str, Any] = { "id": response_id, "object": "response", "created_at": int(time.time()), - "status": "completed", + "status": status, "model": _model_from_result(result), "output": output_items, "parallel_tool_calls": False, "tool_choice": "auto", "tools": [], - "metadata": {}, } + if (metadata := _metadata_from_result(result)) is not None: + response_kwargs["metadata"] = metadata + if (usage := _usage_from_result(result)) is not None: + response_kwargs["usage"] = usage + if (error := _response_field_from_result(result, "error")) is not None: + response_kwargs["error"] = error + if (incomplete_details := _incomplete_details_from_result(result, status=status)) is not None: + response_kwargs["incomplete_details"] = incomplete_details if conversation_id is not None: response_kwargs["conversation"] = {"id": conversation_id} return _response_payload(OpenAIResponse(**response_kwargs)) +def _status_from_result(result: AgentResponse[Any]) -> str: + transport_status = _response_field_from_result(result, "status") + if transport_status is not None: + if not isinstance(transport_status, str) or transport_status not in _RESPONSES_STATUSES: + raise ValueError(f"AgentResponse status is not a valid Responses status: {transport_status!r}") + return transport_status + + finish_reason = getattr(result.finish_reason, "value", result.finish_reason) + if finish_reason in ("length", "content_filter"): + return "incomplete" + if result.continuation_token is not None: + return "in_progress" + return "completed" + + +def _metadata_from_result(result: AgentResponse[Any]) -> dict[str, str] | None: + metadata = _response_field_from_result(result, "metadata", allow_additional_properties=True) + if metadata is None: + return None + if not isinstance(metadata, Mapping): + raise ValueError("AgentResponse metadata must be an object with string keys and values") + metadata_dict = dict(cast("Mapping[Any, Any]", metadata)) + if not all(isinstance(key, str) and isinstance(value, str) for key, value in metadata_dict.items()): + raise ValueError("AgentResponse metadata must be an object with string keys and values") + return cast("dict[str, str]", metadata_dict) + + +def _usage_from_result(result: AgentResponse[Any]) -> Any | None: + if (raw_usage := _raw_response_usage_from_result(result)) is not None: + return raw_usage + + usage_details = result.usage_details + if not usage_details: + return None + + input_tokens = _usage_count(usage_details, "input_token_count") + output_tokens = _usage_count(usage_details, "output_token_count") + total_tokens = _usage_count(usage_details, "total_token_count") + cached_tokens = _usage_count(usage_details, "cache_read_input_token_count") + cache_write_tokens = _usage_count(usage_details, "cache_creation_input_token_count") + reasoning_tokens = _usage_count(usage_details, "reasoning_output_token_count") + if input_tokens is None or output_tokens is None: + return None + + expected_total = input_tokens + output_tokens + cache_details = [count for count in (cached_tokens, cache_write_tokens) if count is not None] + if sum(cache_details) > input_tokens: + return None + if reasoning_tokens is not None and reasoning_tokens > output_tokens: + return None + if total_tokens is not None and total_tokens != expected_total: + return None + if total_tokens is None: + total_tokens = expected_total + + input_tokens_details: dict[str, int] = {} + if cached_tokens is not None: + input_tokens_details["cached_tokens"] = cached_tokens + if cache_write_tokens is not None: + input_tokens_details["cache_write_tokens"] = cache_write_tokens + output_tokens_details: dict[str, int] = {} + if reasoning_tokens is not None: + output_tokens_details["reasoning_tokens"] = reasoning_tokens + return _validated_response_usage({ + "input_tokens": input_tokens, + "input_tokens_details": input_tokens_details, + "output_tokens": output_tokens, + "output_tokens_details": output_tokens_details, + "total_tokens": total_tokens, + }) + + +def _raw_response_usage_from_result(result: AgentResponse[Any]) -> Any | None: + for source in _raw_response_sources(result.raw_representation): + if isinstance(source, Mapping): + source_map = cast("Mapping[str, Any]", source) + response_object = source_map.get("object") + usage = source_map.get("usage") + else: + response_object = getattr(source, "object", None) + usage = getattr(source, "usage", None) + if response_object == "response" and usage is not None: + return _validated_response_usage(usage) + return None + + +def _validated_response_usage(value: Any) -> Any | None: + try: + _RESPONSE_USAGE_ADAPTER.validate_python(value) + except ValidationError: + return None + return value + + +def _usage_count(usage_details: Mapping[str, Any], key: str) -> int | None: + value = usage_details.get(key) + if value is None: + return None + if isinstance(value, bool) or not isinstance(value, int) or value < 0: + raise ValueError(f"AgentResponse usage_details `{key}` must be a non-negative integer") + return value + + +def _incomplete_details_from_result(result: AgentResponse[Any], *, status: str) -> Any | None: + incomplete_details = _response_field_from_result(result, "incomplete_details", allow_additional_properties=True) + if incomplete_details is not None: + return incomplete_details + if status != "incomplete": + return None + + finish_reason = getattr(result.finish_reason, "value", result.finish_reason) + if finish_reason == "length": + return {"reason": "max_output_tokens"} + if finish_reason == "content_filter": + return {"reason": "content_filter"} + return None + + +def _response_field_from_result( + result: AgentResponse[Any], + key: str, + *, + allow_additional_properties: bool = False, +) -> Any | None: + for source in _raw_response_sources(result.raw_representation): + if isinstance(source, Mapping): + value = cast("Mapping[str, Any]", source).get(key) + else: + value = getattr(source, key, None) + if value is not None: + return value + if allow_additional_properties and key in result.additional_properties: + return result.additional_properties[key] + return None + + +def _raw_response_sources(raw: Any) -> Iterator[Any]: + """Traverse raw response wrappers, preferring the latest streaming update.""" + stack = [raw] + seen: set[int] = set() + while stack: + source = stack.pop() + if source is None or id(source) in seen: + continue + seen.add(id(source)) + + if isinstance(source, Sequence) and not isinstance(source, (str, bytes, bytearray)): + stack.extend(cast("Sequence[Any]", source)) + continue + + yield source + if isinstance(source, Mapping): + source_map = cast("Mapping[str, Any]", source) + nested_raw = source_map.get("raw_representation") + response = source_map.get("response") + else: + nested_raw = getattr(source, "raw_representation", None) + response = getattr(source, "response", None) + if nested_raw is not None: + stack.append(nested_raw) + if response is not None: + stack.append(response) + + def _model_from_update(update: AgentResponseUpdate) -> str | None: """Best-effort model id from one streamed update's raw representation. @@ -305,19 +543,15 @@ def _output_to_output_items(output: Any, *, status: str) -> list[ResponseOutputI def _messages_to_output_items(messages: Sequence[Any], *, status: str) -> list[ResponseOutputItem]: output_items: list[ResponseOutputItem] = [] - message_contents: list[Content] = [] for message in messages: if not isinstance(message, Message): - if message_contents: - output_items.extend(_contents_to_output_items(message_contents, status=status)) - message_contents.clear() output_items.extend(_output_to_output_items(message, status=status)) continue - message_contents.extend(message.contents) - - if message_contents: - output_items.extend(_contents_to_output_items(message_contents, status=status)) + message_output_items = _contents_to_output_items(message.contents, status=status) + if message.role != "assistant" and any(_raw_type(item) == "message" for item in message_output_items): + raise ValueError(f"Responses output messages require the `assistant` role; received {message.role!r}") + output_items.extend(message_output_items) return output_items @@ -408,8 +642,9 @@ def flush_message() -> None: flush_message() output_items.append(_function_approval_response_output_item(content)) case "data" | "uri" | "hosted_file": - flush_message() - output_items.append(_media_content_output_item(content, status=status)) + raise ValueError( + f"Responses output has no standard representation for standalone {content.type!r} content" + ) case "error": message_content.append(ResponseOutputText(type="output_text", text=str(content), annotations=[])) case _: @@ -662,22 +897,6 @@ def _function_approval_response_output_item(content: Content) -> ResponseOutputI }) -def _media_content_output_item(content: Content, *, status: str) -> ResponseOutputItem: - parts = _content_parts_to_input_items([content]) - if parts: - return cast( - ResponseOutputItem, - ResponseFunctionToolCallOutputItem( - id=f"content_{uuid.uuid4().hex}", - type="function_call_output", - call_id=f"content_{uuid.uuid4().hex}", - output=parts, - status=_message_status(status), # type: ignore[arg-type] - ), - ) - return _text_output_items(json.dumps(content.to_dict(), default=str), status=status)[0] - - def _content_parts_to_input_items(contents: Sequence[Content] | None) -> list[Any]: if not contents: return [] @@ -920,10 +1139,17 @@ async def responses_from_streaming_run( # (see `_model_from_update`), so prefer the model observed on the # stream's own chunks over `responses_from_run`'s "agent" fallback. payload["model"] = model + status = payload.get("status") + if not isinstance(status, str) or status not in _STREAMING_TERMINAL_STATUSES: + raise ValueError( + f"Response stream finalized with unsupported status {status!r}; " + "expected `completed`, `incomplete`, or `failed`" + ) + terminal_event_type = f"response.{status}" yield _sse_event( - "response.completed", + terminal_event_type, { - "type": "response.completed", + "type": terminal_event_type, "response": payload, }, ) diff --git a/python/packages/hosting-responses/tests/hosting_responses/test_http_round_trip.py b/python/packages/hosting-responses/tests/hosting_responses/test_http_round_trip.py index 2772ca7b0c0..6d3a7149bf1 100644 --- a/python/packages/hosting-responses/tests/hosting_responses/test_http_round_trip.py +++ b/python/packages/hosting-responses/tests/hosting_responses/test_http_round_trip.py @@ -213,6 +213,30 @@ async def test_invalid_input_returns_400_not_500(self) -> None: assert response.status_code == 400 assert "input" in response.json()["detail"] + async def test_conflicting_continuation_mechanisms_return_400(self) -> None: + app = _build_app(_StubAgent()) + response = await _post( + app, + { + "input": "hello", + "previous_response_id": "resp_1", + "conversation": "conv_1", + }, + ) + + assert response.status_code == 400 + assert "mutually exclusive" in response.json()["detail"] + + async def test_missing_message_role_returns_400(self) -> None: + app = _build_app(_StubAgent()) + response = await _post( + app, + {"input": [{"type": "message", "content": "hello"}]}, + ) + + assert response.status_code == 400 + assert "role" in response.json()["detail"] + class TestStreamingRoundTrip: async def test_stream_emits_created_delta_and_completed_events(self) -> None: @@ -277,13 +301,13 @@ async def test_previous_response_id_supports_independent_branches(self) -> None: assert agent.session_turn_counts_seen == [1, 2, 2, 3, 3] - async def test_conversation_id_preserves_session_across_turns(self) -> None: + async def test_conversation_preserves_session_across_turns(self) -> None: agent = _StubAgent() app = _build_app(agent) - turn1 = await _post(app, {"input": "hi", "conversation_id": "conv_stable"}) + turn1 = await _post(app, {"input": "hi", "conversation": "conv_stable"}) assert turn1.status_code == 200 - turn2 = await _post(app, {"input": "still there?", "conversation_id": "conv_stable"}) + turn2 = await _post(app, {"input": "still there?", "conversation": {"id": "conv_stable"}}) assert turn2.status_code == 200 assert agent.session_ids_seen == ["conv_stable", "conv_stable"] 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 9377cc03eb1..765694ec437 100644 --- a/python/packages/hosting-responses/tests/hosting_responses/test_parsing.py +++ b/python/packages/hosting-responses/tests/hosting_responses/test_parsing.py @@ -7,10 +7,12 @@ import json import warnings from collections.abc import AsyncIterator, Sequence +from types import SimpleNamespace from typing import cast import pytest -from agent_framework import AgentResponse, AgentResponseUpdate, Content, Message, ResponseStream +from agent_framework import AgentResponse, AgentResponseUpdate, Content, Message, ResponseStream, UsageDetails +from openai.types.responses.response_usage import InputTokensDetails, ResponseUsage from agent_framework_hosting_responses import ( create_conversation_id, @@ -28,6 +30,19 @@ def _sse_payload(event: str) -> dict[str, object]: return cast("dict[str, object]", json.loads(data_line.removeprefix("data: "))) +def _native_usage_payload() -> dict[str, object]: + input_tokens_details = {"cached_tokens": 1} + if "cache_write_tokens" in InputTokensDetails.model_fields: + input_tokens_details["cache_write_tokens"] = 0 + return { + "input_tokens": 7, + "input_tokens_details": input_tokens_details, + "output_tokens": 2, + "output_tokens_details": {"reasoning_tokens": 0}, + "total_tokens": 9, + } + + class TestMessagesFromResponsesInput: def test_string_input_becomes_single_user_message(self) -> None: msgs = messages_from_responses_input("hello") @@ -41,6 +56,11 @@ def test_input_text_items_collapse_into_one_user_message(self) -> None: assert msgs[0].role == "user" assert msgs[0].text == "a b" + @pytest.mark.parametrize("text", [None, "", 42]) + def test_text_items_require_non_empty_string_text(self, text: object) -> None: + with pytest.raises(ValueError, match="non-empty string `text`"): + messages_from_responses_input([{"type": "input_text", "text": text}]) + def test_message_envelope_with_string_content(self) -> None: msgs = messages_from_responses_input([ {"type": "message", "role": "system", "content": "be brief"}, @@ -67,6 +87,23 @@ def test_message_envelope_rejects_invalid_content_shape(self) -> None: with pytest.raises(ValueError, match="content.*string or list"): messages_from_responses_input([{"type": "message", "role": "user", "content": 42}]) + @pytest.mark.parametrize("role", [None, "", "moderator", 42]) + def test_message_envelope_requires_supported_role(self, role: object) -> None: + with pytest.raises(ValueError, match="message `role`"): + messages_from_responses_input([{"type": "message", "role": role, "content": "hello"}]) + + @pytest.mark.parametrize( + "item", + [ + {"type": "message", "role": "user"}, + {"type": "message", "role": "user", "content": ""}, + {"type": "message", "role": "user", "content": []}, + ], + ) + def test_message_envelope_requires_non_empty_content(self, item: dict[str, object]) -> None: + with pytest.raises(ValueError, match="content"): + messages_from_responses_input([item]) + def test_input_file_via_url(self) -> None: msgs = messages_from_responses_input([ {"type": "input_file", "file_url": "https://example.com/report.pdf", "mime_type": "application/pdf"} @@ -110,6 +147,10 @@ def test_empty_list_raises(self) -> None: with pytest.raises(ValueError, match="non-empty"): messages_from_responses_input([]) + def test_empty_string_raises(self) -> None: + with pytest.raises(ValueError, match="non-empty"): + messages_from_responses_input("") + def test_non_string_non_list_raises(self) -> None: with pytest.raises(ValueError): messages_from_responses_input(42) # type: ignore[arg-type] @@ -132,28 +173,56 @@ def test_create_response_id_shape(self) -> None: assert response_id.startswith("resp_") - def test_responses_session_id_prefers_previous_response(self) -> None: - assert responses_session_id({"previous_response_id": "resp_1", "conversation_id": "conv_1"}) == ( - "resp_1", - False, - ) - def test_responses_session_id_valid_ids_do_not_warn(self) -> None: with warnings.catch_warnings(): warnings.simplefilter("error") assert responses_session_id({"previous_response_id": "resp_1"}) == ("resp_1", False) - assert responses_session_id({"conversation_id": "conv_1"}) == ("conv_1", True) + assert responses_session_id({"conversation": "conv_1"}) == ("conv_1", True) + assert responses_session_id({"conversation": {"id": "conv_2"}}) == ("conv_2", True) def test_responses_session_id_warns_for_nonstandard_previous_response_id(self) -> None: with pytest.warns(UserWarning, match="previous_response_id.*resp_"): assert responses_session_id({"previous_response_id": "custom-response"}) == ("custom-response", False) - def test_responses_session_id_warns_for_nonstandard_conversation_id(self) -> None: - with pytest.warns(UserWarning, match="conversation_id.*conv_"): - assert responses_session_id({"conversation_id": "custom-conversation"}) == ("custom-conversation", True) - - def test_responses_session_id_uses_conversation_id(self) -> None: - assert responses_session_id({"conversation_id": "conv_1"}) == ("conv_1", True) + def test_responses_session_id_warns_for_nonstandard_conversation(self) -> None: + with pytest.warns(UserWarning, match="conversation.*conv_"): + assert responses_session_id({"conversation": "custom-conversation"}) == ("custom-conversation", True) + + def test_responses_session_id_accepts_deprecated_conversation_id_alone(self) -> None: + with pytest.warns(DeprecationWarning, match="conversation_id.*deprecated.*conversation"): + assert responses_session_id({"conversation_id": "conv_legacy"}) == ("conv_legacy", True) + + @pytest.mark.parametrize( + "body", + [ + {"previous_response_id": "resp_1", "conversation": "conv_1"}, + {"previous_response_id": "resp_1", "conversation_id": "conv_1"}, + {"conversation": "conv_1", "conversation_id": "conv_1"}, + ], + ) + def test_responses_session_id_rejects_conflicting_continuation_mechanisms( + self, + body: dict[str, object], + ) -> None: + with pytest.raises(ValueError, match="mutually exclusive"): + responses_session_id(body) + + @pytest.mark.parametrize( + "body", + [ + {"previous_response_id": ""}, + {"previous_response_id": 42}, + {"conversation": ""}, + {"conversation": {}}, + {"conversation": {"id": ""}}, + {"conversation": {"id": 42}}, + {"conversation_id": ""}, + {"conversation_id": 42}, + ], + ) + def test_responses_session_id_rejects_invalid_continuation_values(self, body: dict[str, object]) -> None: + with pytest.raises(ValueError, match="non-empty"): + responses_session_id(body) def test_responses_session_id_returns_none_when_absent(self) -> None: assert responses_session_id({"input": "hi"}) == (None, None) @@ -162,8 +231,7 @@ def test_responses_to_run_returns_messages_options_and_stream(self) -> None: run = responses_to_run({ "input": "hi", "stream": True, - "previous_response_id": "resp_1", - "conversation_id": "conv_1", + "conversation": {"id": "conv_1"}, "max_output_tokens": 32, "model": "gpt-x", }) @@ -175,6 +243,14 @@ def test_responses_to_run_returns_messages_options_and_stream(self) -> None: assert run["stream"] is True assert run["options"] == {"max_tokens": 32, "model": "gpt-x"} + def test_responses_to_run_rejects_conflicting_continuation_mechanisms(self) -> None: + with pytest.raises(ValueError, match="mutually exclusive"): + responses_to_run({ + "input": "hi", + "previous_response_id": "resp_1", + "conversation": "conv_1", + }) + def test_responses_from_run_returns_response_payload(self) -> None: result = AgentResponse( messages=Message(role="assistant", contents=[Content.from_text("hello")]), @@ -187,6 +263,51 @@ def test_responses_from_run_returns_response_payload(self) -> None: assert payload["model"] == "test-model" assert payload["output"][0]["content"][0]["text"] == "hello" + def test_responses_from_run_preserves_message_boundaries(self) -> None: + result = AgentResponse( + messages=[ + Message(role="assistant", contents=[Content.from_text("first")]), + Message(role="assistant", contents=[Content.from_text("second")]), + ] + ) + + payload = responses_from_run(result, response_id="resp_new") + + assert [item["content"][0]["text"] for item in payload["output"]] == ["first", "second"] + + def test_responses_from_run_rejects_non_assistant_message_role(self) -> None: + result = AgentResponse(messages=Message(role="user", contents=[Content.from_text("hello")])) + + with pytest.raises(ValueError, match="require.*assistant.*user"): + responses_from_run(result, response_id="resp_new") + + def test_responses_from_run_preserves_tool_role_function_result(self) -> None: + result = AgentResponse( + messages=Message( + role="tool", + contents=[Content.from_function_result("call_1", result="sunny")], + ) + ) + + payload = responses_from_run(result, response_id="resp_new") + + assert len(payload["output"]) == 1 + assert payload["output"][0]["type"] == "function_call_output" + assert payload["output"][0]["call_id"] == "call_1" + assert payload["output"][0]["output"] == [{"type": "input_text", "text": "sunny"}] + assert payload["output"][0]["status"] == "completed" + + def test_responses_from_run_rejects_standalone_media(self) -> None: + result = AgentResponse( + messages=Message( + role="assistant", + contents=[Content.from_uri("https://example.com/cat.png", media_type="image/png")], + ) + ) + + with pytest.raises(ValueError, match="no standard representation.*uri"): + responses_from_run(result, response_id="resp_new") + def test_responses_from_run_preserves_multimodal_output_items(self) -> None: result = AgentResponse( messages=Message( @@ -226,6 +347,370 @@ def test_responses_from_run_preserves_multimodal_output_items(self) -> None: ] assert output[3]["content"][0]["text"] == "done" + def test_responses_from_run_preserves_status_metadata_and_usage(self) -> None: + result = AgentResponse( + messages=Message(role="assistant", contents=[Content.from_text("truncated")]), + finish_reason="length", + usage_details={ + "input_token_count": 10, + "output_token_count": 4, + "total_token_count": 14, + "cache_read_input_token_count": 3, + "cache_creation_input_token_count": 1, + "reasoning_output_token_count": 2, + }, + additional_properties={"metadata": {"tenant": "contoso"}}, + ) + + payload = responses_from_run(result, response_id="resp_new") + + assert payload["status"] == "incomplete" + assert payload["incomplete_details"] == {"reason": "max_output_tokens"} + assert payload["metadata"] == {"tenant": "contoso"} + assert payload["usage"] == { + "input_tokens": 10, + "input_tokens_details": {"cached_tokens": 3, "cache_write_tokens": 1}, + "output_tokens": 4, + "output_tokens_details": {"reasoning_tokens": 2}, + "total_tokens": 14, + } + assert payload["output"][0]["status"] == "incomplete" + + def test_responses_from_run_uses_raw_response_fields_as_fallback(self) -> None: + earlier_response = SimpleNamespace(status="completed") + terminal_response = SimpleNamespace( + object="response", + status="failed", + metadata={"source": "raw"}, + error={"code": "server_error", "message": "provider failed"}, + usage=_native_usage_payload(), + ) + raw = [ + SimpleNamespace(raw_representation=SimpleNamespace(response=earlier_response)), + SimpleNamespace(raw_representation=SimpleNamespace(response=terminal_response)), + ] + result = AgentResponse( + messages=Message(role="assistant", contents=[Content.from_text("partial")]), + raw_representation=raw, + ) + + payload = responses_from_run(result, response_id="resp_new") + + assert payload["status"] == "failed" + assert payload["metadata"] == {"source": "raw"} + assert payload["error"] == {"code": "server_error", "message": "provider failed"} + assert payload["usage"]["total_tokens"] == 9 + + def test_responses_from_run_prefers_valid_raw_usage_without_merging_af_counts(self) -> None: + raw_usage = ResponseUsage.model_validate(_native_usage_payload()) + result = AgentResponse( + messages=Message(role="assistant", contents=[Content.from_text("hello")]), + usage_details=cast("UsageDetails", {"input_token_count": True}), + raw_representation=SimpleNamespace(object="response", usage=raw_usage), + ) + + payload = responses_from_run(result, response_id="resp_new") + + assert payload["usage"] == raw_usage.model_dump(mode="json", exclude_none=True) + + def test_responses_from_run_falls_back_to_complete_af_usage_when_raw_usage_is_invalid(self) -> None: + result = AgentResponse( + messages=Message(role="assistant", contents=[Content.from_text("hello")]), + usage_details={ + "input_token_count": 5, + "output_token_count": 3, + "cache_read_input_token_count": 2, + "cache_creation_input_token_count": 1, + "reasoning_output_token_count": 1, + }, + raw_representation=SimpleNamespace(object="response", usage={"input_tokens": 99}), + ) + + payload = responses_from_run(result, response_id="resp_new") + + assert payload["usage"]["input_tokens"] == 5 + assert payload["usage"]["total_tokens"] == 8 + + def test_responses_from_run_omits_invalid_raw_and_partial_af_usage(self) -> None: + result = AgentResponse( + messages=Message(role="assistant", contents=[Content.from_text("hello")]), + usage_details={"input_token_count": 5}, + raw_representation=SimpleNamespace(object="response", usage={"input_tokens": 99}), + ) + + payload = responses_from_run(result, response_id="resp_new") + + assert "usage" not in payload + + def test_responses_from_run_does_not_treat_lookalike_raw_usage_as_responses_usage(self) -> None: + result = AgentResponse( + messages=Message(role="assistant", contents=[Content.from_text("hello")]), + usage_details={"input_token_count": 5}, + raw_representation=SimpleNamespace(usage=_native_usage_payload()), + ) + + payload = responses_from_run(result, response_id="resp_new") + + assert "usage" not in payload + + def test_responses_from_run_does_not_treat_user_metadata_status_as_transport_status(self) -> None: + raw_response = SimpleNamespace(status="completed", metadata={"status": "gold"}) + result = AgentResponse( + messages=Message(role="assistant", contents=[Content.from_text("hello")]), + additional_properties={"status": "gold"}, + raw_representation=SimpleNamespace(raw_representation=raw_response), + ) + + payload = responses_from_run(result, response_id="resp_new") + + assert payload["status"] == "completed" + assert payload["metadata"] == {"status": "gold"} + + def test_responses_from_run_rejects_invalid_metadata(self) -> None: + result = AgentResponse( + messages=Message(role="assistant", contents=[Content.from_text("hello")]), + additional_properties={"metadata": {"attempt": 1}}, + ) + + with pytest.raises(ValueError): + responses_from_run(result, response_id="resp_new") + + @pytest.mark.parametrize( + ("usage_details", "expected_usage"), + [ + ( + { + "input_token_count": 5, + "output_token_count": 3, + "cache_read_input_token_count": 2, + "cache_creation_input_token_count": 1, + "reasoning_output_token_count": 1, + }, + { + "input_tokens": 5, + "input_tokens_details": {"cached_tokens": 2, "cache_write_tokens": 1}, + "output_tokens": 3, + "output_tokens_details": {"reasoning_tokens": 1}, + "total_tokens": 8, + }, + ), + ( + { + "input_token_count": 0, + "output_token_count": 0, + "total_token_count": 0, + "cache_read_input_token_count": 0, + "cache_creation_input_token_count": 0, + "reasoning_output_token_count": 0, + }, + { + "input_tokens": 0, + "input_tokens_details": {"cached_tokens": 0, "cache_write_tokens": 0}, + "output_tokens": 0, + "output_tokens_details": {"reasoning_tokens": 0}, + "total_tokens": 0, + }, + ), + ], + ) + def test_responses_from_run_maps_complete_usage_without_cross_filling( + self, + usage_details: UsageDetails, + expected_usage: dict[str, object], + ) -> None: + result = AgentResponse( + messages=Message(role="assistant", contents=[Content.from_text("hello")]), + usage_details=usage_details, + ) + + payload = responses_from_run(result, response_id="resp_new") + + assert payload["usage"] == expected_usage + + @pytest.mark.parametrize( + "usage_details", + [ + pytest.param({"input_token_count": 2}, id="input-only"), + pytest.param({"output_token_count": 3}, id="output-only"), + pytest.param({"total_token_count": 5}, id="total-only"), + pytest.param( + { + "cache_read_input_token_count": 1, + "cache_creation_input_token_count": 1, + "reasoning_output_token_count": 1, + }, + id="details-only", + ), + pytest.param({"input_token_count": 2, "output_token_count": 3}, id="parents-only"), + pytest.param( + { + "output_token_count": 0, + "cache_read_input_token_count": 0, + "cache_creation_input_token_count": 0, + "reasoning_output_token_count": 0, + }, + id="missing-input", + ), + pytest.param( + { + "input_token_count": 0, + "cache_read_input_token_count": 0, + "cache_creation_input_token_count": 0, + "reasoning_output_token_count": 0, + }, + id="missing-output", + ), + pytest.param( + { + "input_token_count": 0, + "output_token_count": 0, + "cache_creation_input_token_count": 0, + "reasoning_output_token_count": 0, + }, + id="missing-cache-read", + ), + pytest.param( + { + "input_token_count": 0, + "output_token_count": 0, + "cache_read_input_token_count": 0, + "cache_creation_input_token_count": 0, + }, + id="missing-reasoning", + ), + ], + ) + def test_responses_from_run_omits_partial_usage( + self, + usage_details: UsageDetails, + ) -> None: + result = AgentResponse( + messages=Message(role="assistant", contents=[Content.from_text("hello")]), + usage_details=usage_details, + ) + + payload = responses_from_run(result, response_id="resp_new") + + assert "usage" not in payload + + def test_responses_from_run_uses_installed_sdk_usage_detail_schema(self) -> None: + result = AgentResponse( + messages=Message(role="assistant", contents=[Content.from_text("hello")]), + usage_details={ + "input_token_count": 5, + "output_token_count": 1, + "cache_read_input_token_count": 1, + "reasoning_output_token_count": 0, + }, + ) + + payload = responses_from_run(result, response_id="resp_new") + + if "cache_write_tokens" in InputTokensDetails.model_fields: + assert "usage" not in payload + else: + assert payload["usage"] == { + "input_tokens": 5, + "input_tokens_details": {"cached_tokens": 1}, + "output_tokens": 1, + "output_tokens_details": {"reasoning_tokens": 0}, + "total_tokens": 6, + } + + @pytest.mark.parametrize( + "usage_details", + [ + pytest.param( + { + "input_token_count": 5, + "output_token_count": 1, + "total_token_count": 6, + "cache_read_input_token_count": 100, + "cache_creation_input_token_count": 0, + "reasoning_output_token_count": 0, + }, + id="provider-exclusive-cache-count", + ), + pytest.param( + { + "input_token_count": 5, + "output_token_count": 1, + "total_token_count": 6, + "cache_read_input_token_count": 3, + "cache_creation_input_token_count": 3, + "reasoning_output_token_count": 0, + }, + id="combined-cache-details-exceed-input", + ), + pytest.param( + { + "input_token_count": 1, + "output_token_count": 1, + "total_token_count": 2, + "cache_read_input_token_count": 0, + "cache_creation_input_token_count": 0, + "reasoning_output_token_count": 2, + }, + id="reasoning-exceeds-output", + ), + pytest.param( + { + "input_token_count": 2, + "output_token_count": 3, + "total_token_count": 4, + "cache_read_input_token_count": 0, + "cache_creation_input_token_count": 0, + "reasoning_output_token_count": 0, + }, + id="total-under-counts", + ), + pytest.param( + { + "input_token_count": 2, + "output_token_count": 3, + "total_token_count": 6, + "cache_read_input_token_count": 0, + "cache_creation_input_token_count": 0, + "reasoning_output_token_count": 0, + }, + id="total-over-counts", + ), + ], + ) + def test_responses_from_run_omits_usage_that_cannot_map_consistently( + self, + usage_details: UsageDetails, + ) -> None: + result = AgentResponse( + messages=Message(role="assistant", contents=[Content.from_text("hello")]), + usage_details=usage_details, + ) + + payload = responses_from_run(result, response_id="resp_new") + + assert "usage" not in payload + + @pytest.mark.parametrize( + "key", + [ + "input_token_count", + "output_token_count", + "total_token_count", + "cache_read_input_token_count", + "cache_creation_input_token_count", + "reasoning_output_token_count", + ], + ) + @pytest.mark.parametrize("count", [-1, True, 1.5]) + def test_responses_from_run_rejects_invalid_usage_count(self, key: str, count: object) -> None: + result = AgentResponse( + messages=Message(role="assistant", contents=[Content.from_text("hello")]), + usage_details=cast("UsageDetails", {key: count}), + ) + + with pytest.raises(ValueError, match="non-negative integer"): + responses_from_run(result, response_id="resp_new") + def test_responses_from_run_maps_conversation_id(self) -> None: result = AgentResponse(messages=Message(role="assistant", contents=[Content.from_text("hello")])) @@ -295,6 +780,97 @@ async def updates() -> AsyncIterator[AgentResponseUpdate]: assert error["message"] == "upstream blew up" assert "partial" in events[-1] + async def test_responses_from_streaming_run_preserves_final_metadata_and_usage(self) -> None: + async def updates() -> AsyncIterator[AgentResponseUpdate]: + yield AgentResponseUpdate(contents=[Content.from_text("hello")], role="assistant") + + def finalizer(items: Sequence[AgentResponseUpdate]) -> AgentResponse: + response = AgentResponse.from_updates(items) + response.usage_details = UsageDetails( + input_token_count=5, + output_token_count=1, + total_token_count=6, + cache_read_input_token_count=0, + cache_creation_input_token_count=0, + reasoning_output_token_count=0, + ) + response.additional_properties["metadata"] = {"source": "stream"} + return response + + stream = ResponseStream(updates(), finalizer=finalizer) + + events = [event async for event in responses_from_streaming_run(stream, response_id="resp_new")] + payload = _sse_payload(events[-1]) + response = cast("dict[str, object]", payload["response"]) + + assert response["metadata"] == {"source": "stream"} + usage = cast("dict[str, object]", response["usage"]) + assert usage["total_tokens"] == 6 + + @pytest.mark.parametrize("status", ["completed", "incomplete", "failed"]) + async def test_responses_from_streaming_run_emits_matching_terminal_event(self, status: str) -> None: + async def updates() -> AsyncIterator[AgentResponseUpdate]: + yield AgentResponseUpdate(contents=[Content.from_text("hello")], role="assistant") + + def finalizer(items: Sequence[AgentResponseUpdate]) -> AgentResponse: + response = AgentResponse.from_updates(items) + response.raw_representation = SimpleNamespace(status=status) + return response + + stream = ResponseStream(updates(), finalizer=finalizer) + + events = [event async for event in responses_from_streaming_run(stream, response_id="resp_new")] + payload = _sse_payload(events[-1]) + response = cast("dict[str, object]", payload["response"]) + + assert events[-1].startswith(f"event: response.{status}") + assert payload["type"] == f"response.{status}" + assert response["status"] == status + + async def test_responses_from_streaming_run_preserves_failed_transport_error(self) -> None: + async def updates() -> AsyncIterator[AgentResponseUpdate]: + yield AgentResponseUpdate(contents=[Content.from_text("partial")], role="assistant") + + def finalizer(items: Sequence[AgentResponseUpdate]) -> AgentResponse: + response = AgentResponse.from_updates(items) + response.raw_representation = SimpleNamespace( + response=SimpleNamespace( + status="failed", + error={"code": "server_error", "message": "provider failed"}, + ) + ) + return response + + stream = ResponseStream(updates(), finalizer=finalizer) + + events = [event async for event in responses_from_streaming_run(stream, response_id="resp_new")] + payload = _sse_payload(events[-1]) + response = cast("dict[str, object]", payload["response"]) + + assert events[-1].startswith("event: response.failed") + assert response["error"] == {"code": "server_error", "message": "provider failed"} + + @pytest.mark.parametrize("status", ["in_progress", "queued"]) + async def test_responses_from_streaming_run_rejects_nonterminal_final_status(self, status: str) -> None: + async def updates() -> AsyncIterator[AgentResponseUpdate]: + yield AgentResponseUpdate(contents=[Content.from_text("partial")], role="assistant") + + def finalizer(items: Sequence[AgentResponseUpdate]) -> AgentResponse: + response = AgentResponse.from_updates(items) + response.raw_representation = SimpleNamespace(status=status) + return response + + stream = ResponseStream(updates(), finalizer=finalizer) + + events = [event async for event in responses_from_streaming_run(stream, response_id="resp_new")] + payload = _sse_payload(events[-1]) + response = cast("dict[str, object]", payload["response"]) + error = cast("dict[str, object]", response["error"]) + + assert events[-1].startswith("event: response.failed") + assert response["status"] == "failed" + assert f"unsupported status {status!r}" in cast(str, error["message"]) + async def test_responses_from_streaming_run_emits_failed_when_finalizer_raises(self) -> None: async def updates() -> AsyncIterator[AgentResponseUpdate]: yield AgentResponseUpdate(contents=[Content.from_text("partial")], role="assistant") diff --git a/python/samples/04-hosting/af-hosting/local_responses/README.md b/python/samples/04-hosting/af-hosting/local_responses/README.md index 995eacda897..d3539c1d0e9 100644 --- a/python/samples/04-hosting/af-hosting/local_responses/README.md +++ b/python/samples/04-hosting/af-hosting/local_responses/README.md @@ -31,10 +31,10 @@ What the route demonstrates: caller continue from any earlier response, not just the latest one — so every response id needs to stay independently resolvable, not just the most recent. -- Treats an unknown `conversation_id` as a request to create a new local +- Treats an unknown id supplied through `conversation` as a request to create a new local session. Your app can choose a stricter policy, such as requiring a separate API to create new conversations before callers can continue them. -- Explicitly advances a supplied `conversation_id` after each completed run. +- Explicitly advances a conversation supplied through `conversation` after each completed run. A conversation id is a mutable head, so only one caller should advance it at a time. The sample and `AgentState` do not provide that locking; production apps must serialize writers or use optimistic concurrency. These requests @@ -55,10 +55,10 @@ to callers, add authentication and authorization at the infrastructure layer, the FastAPI app layer, or inside the route body. Session continuation deserves particular care: treat `previous_response_id` and -`conversation_id` as untrusted request values, authorize the caller before +`conversation` as untrusted request values, authorize the caller before loading or storing a session for those ids, and partition any durable session store by tenant/user as appropriate for your application. Also coordinate -writers for each stable `conversation_id`; this sample does not do so out of +writers for each stable conversation id; this sample does not do so out of the box. ## Run diff --git a/python/samples/04-hosting/af-hosting/local_responses/app.py b/python/samples/04-hosting/af-hosting/local_responses/app.py index 63c06c69986..54ac3856c48 100644 --- a/python/samples/04-hosting/af-hosting/local_responses/app.py +++ b/python/samples/04-hosting/af-hosting/local_responses/app.py @@ -18,12 +18,12 @@ layer, the FastAPI app layer, or inside the route body. Session continuation deserves particular care: treat ``previous_response_id`` -and ``conversation_id`` as untrusted request values, authorize the caller +and ``conversation`` as untrusted request values, authorize the caller before loading or storing a session for those ids, and partition durable session storage by tenant/user as appropriate for your application. See ``README.md#production-readiness``. -Unknown ``conversation_id`` values create a new local session in this sample. +Unknown ids supplied through ``conversation`` create a new local session in this sample. Your app can choose a different policy, such as requiring a separate API to create new conversations before callers can continue them. @@ -134,7 +134,7 @@ async def responses(body: dict[str, Any] = Body(...)) -> JSONResponse | Streamin target = await state.get_target() lookup_id = session_id or response_id - # An unknown `conversation_id` becomes a new session here. Production apps + # An unknown id supplied through `conversation` becomes a new session here. Production apps # can choose to require a separate "create conversation" API instead. session = await state.get_or_create_session(lookup_id) if run["stream"]: diff --git a/python/samples/04-hosting/af-hosting/local_responses_harness/README.md b/python/samples/04-hosting/af-hosting/local_responses_harness/README.md index ddf6f5c8cba..af8aafedb58 100644 --- a/python/samples/04-hosting/af-hosting/local_responses_harness/README.md +++ b/python/samples/04-hosting/af-hosting/local_responses_harness/README.md @@ -44,9 +44,9 @@ What the route demonstrates (identical to `local_responses/`): the session. OpenAI's `previous_response_id` rotates every turn *by design* — it lets a caller continue from any earlier response, not just the latest one — so every response id needs to stay independently resolvable. -- Treats an unknown `conversation_id` as a request to create a new local +- Treats an unknown id supplied through `conversation` as a request to create a new local session. Your app can choose a stricter policy. -- Explicitly advances a supplied `conversation_id` after each completed run. A +- Explicitly advances a conversation supplied through `conversation` after each completed run. A conversation id is a mutable head, so production apps should serialize writers or use optimistic concurrency; the sample stores the updated session only under that stable conversation id. @@ -64,7 +64,7 @@ to callers, add authentication and authorization at the infrastructure layer, the FastAPI app layer, or inside the route body. Session continuation deserves particular care: treat `previous_response_id` and -`conversation_id` as untrusted request values, authorize the caller before +`conversation` as untrusted request values, authorize the caller before loading or storing a session for those ids, and partition any durable session store by tenant/user as appropriate for your application. diff --git a/python/samples/04-hosting/af-hosting/local_responses_harness/app.py b/python/samples/04-hosting/af-hosting/local_responses_harness/app.py index 3fa65e37833..b9bef6cab0b 100644 --- a/python/samples/04-hosting/af-hosting/local_responses_harness/app.py +++ b/python/samples/04-hosting/af-hosting/local_responses_harness/app.py @@ -36,12 +36,12 @@ layer, the FastAPI app layer, or inside the route body. Session continuation deserves particular care: treat ``previous_response_id`` -and ``conversation_id`` as untrusted request values, authorize the caller +and ``conversation`` as untrusted request values, authorize the caller before loading or storing a session for those ids, and partition durable session storage by tenant/user as appropriate for your application. See ``README.md#production-readiness``. -Unknown ``conversation_id`` values create a new local session in this sample. +Unknown ids supplied through ``conversation`` create a new local session in this sample. Your app can choose a different policy, such as requiring a separate API to create new conversations before callers can continue them. @@ -175,7 +175,7 @@ async def responses(body: dict[str, Any] = Body(...)) -> JSONResponse | Streamin target = await state.get_target() lookup_id = session_id or response_id - # An unknown `conversation_id` becomes a new session here. Production apps + # An unknown id supplied through `conversation` becomes a new session here. Production apps # can choose to require a separate "create conversation" API instead. session = await state.get_or_create_session(lookup_id) if run["stream"]: diff --git a/python/samples/04-hosting/af-hosting/local_responses_workflow/README.md b/python/samples/04-hosting/af-hosting/local_responses_workflow/README.md index d394ec92d6c..88aca57eef1 100644 --- a/python/samples/04-hosting/af-hosting/local_responses_workflow/README.md +++ b/python/samples/04-hosting/af-hosting/local_responses_workflow/README.md @@ -9,7 +9,7 @@ This sample shows the helper-first hosting shape for a local workflow: `response_id -> checkpoint_id` cursor used to continue from a previous response. - Continuation is intentionally limited to `previous_response_id`; this sample - rejects `conversation_id` continuity with HTTP 400. + rejects `conversation` continuity with HTTP 400. The workflow writes a slogan with one Foundry-backed writer agent and a small deterministic formatter executor. That keeps the sample focused on native @@ -60,8 +60,8 @@ uv run python call_server.py '{"topic": "electric SUV", "style": "playful", "aud The script sends a follow-up using the first response id as `previous_response_id`, so the workflow restores the prior checkpoint before -running the next turn. It deliberately does not send `conversation_id`, because -this sample rejects `conversation_id` continuation. +running the next turn. It deliberately does not send `conversation`, because +this sample rejects `conversation` continuation. > This sample uses local file storage under `storage/` for both workflow > checkpoints and checkpoint cursors. The checkpoint bucket names are hashed diff --git a/python/samples/04-hosting/af-hosting/local_responses_workflow/app.py b/python/samples/04-hosting/af-hosting/local_responses_workflow/app.py index dc49915d9c1..b7396881a3e 100644 --- a/python/samples/04-hosting/af-hosting/local_responses_workflow/app.py +++ b/python/samples/04-hosting/af-hosting/local_responses_workflow/app.py @@ -17,7 +17,7 @@ layer, the FastAPI app layer, or inside the route body. This sample demonstrates continuation with ``previous_response_id`` only. It -rejects ``conversation_id`` continuity with HTTP 400. Treat every +rejects ``conversation`` continuity with HTTP 400. Treat every ``previous_response_id`` as an untrusted request value, authorize the caller before restoring or storing a checkpoint cursor for that id, and partition durable checkpoint/cursor storage by tenant/user as appropriate for your @@ -230,13 +230,13 @@ async def responses(body: dict[str, Any] = Body(...)) -> JSONResponse: # noqa: raise HTTPException(status_code=400, detail=str(exc)) from exc # This sample demonstrates only Responses `previous_response_id` - # continuation, so reject `conversation_id` instead of treating it as a + # continuation, so reject `conversation` instead of treating it as a # checkpoint cursor. previous_response_id, is_conversation_id = responses_session_id(body) if is_conversation_id: raise HTTPException( status_code=400, - detail="This server supports previous_response_id continuation only; conversation_id is not implemented.", + detail="This server supports previous_response_id continuation only; conversation is not implemented.", ) response_id = create_response_id() diff --git a/python/samples/04-hosting/af-hosting/local_responses_workflow/call_server.py b/python/samples/04-hosting/af-hosting/local_responses_workflow/call_server.py index 418708c8396..6c7cbdfbe83 100644 --- a/python/samples/04-hosting/af-hosting/local_responses_workflow/call_server.py +++ b/python/samples/04-hosting/af-hosting/local_responses_workflow/call_server.py @@ -4,7 +4,7 @@ Posts to ``/responses`` using the standard ``openai`` SDK. This client demonstrates the sample's only supported continuation mode: -``previous_response_id``. It deliberately does not send ``conversation_id``, +``previous_response_id``. It deliberately does not send ``conversation``, which the sample server rejects. Start the server first (in another shell):: @@ -38,7 +38,7 @@ def main() -> None: print(f"Response ID: {response.id}") # Continue with the returned response id. The server sample rejects - # `conversation_id` continuity. + # `conversation` continuity. follow_up = client.responses.create( input=FOLLOW_UP, previous_response_id=response.id,