From 278735180f9f868adc552dc5cddf478c054ba0e7 Mon Sep 17 00:00:00 2001
From: eavanvalkenburg <github@vanvalkenburg.eu>
Date: Mon, 31 Aug 2026 14:08:01 +0200
Subject: [PATCH 1/6] Python: align Responses hosting parser

Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com>
---
 docs/specs/002-python-hosting-channels.md     |  25 +-
 python/packages/hosting-responses/README.md   |  13 +-
 .../_parsing.py                               | 263 ++++++++++++++----
 .../hosting_responses/test_http_round_trip.py |  30 +-
 .../tests/hosting_responses/test_parsing.py   | 239 ++++++++++++++--
 .../af-hosting/local_responses/README.md      |   8 +-
 .../af-hosting/local_responses/app.py         |   6 +-
 .../local_responses_harness/README.md         |   6 +-
 .../af-hosting/local_responses_harness/app.py |   6 +-
 .../local_responses_workflow/README.md        |   6 +-
 .../local_responses_workflow/app.py           |   6 +-
 .../local_responses_workflow/call_server.py   |   4 +-
 12 files changed, 499 insertions(+), 113 deletions(-)

diff --git a/docs/specs/002-python-hosting-channels.md b/docs/specs/002-python-hosting-channels.md
index 337dcef1f92..60c303b728c 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,15 +233,19 @@ 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.
@@ -451,7 +455,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 +473,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..ab17e6061b9 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
@@ -56,8 +56,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..b69c612d8d1 100644
--- a/python/packages/hosting-responses/agent_framework_hosting_responses/_parsing.py
+++ b/python/packages/hosting-responses/agent_framework_hosting_responses/_parsing.py
@@ -49,7 +49,16 @@
 }
 # 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"})
 
 
 def _content_from_input_item(item: Mapping[str, Any]) -> Content:
@@ -62,26 +71,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 +121,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 +173,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 +242,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 +279,126 @@ 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 (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:
+    explicit_status = _response_field_from_result(result, "status")
+    if explicit_status is not None:
+        if not isinstance(explicit_status, str) or explicit_status not in _RESPONSES_STATUSES:
+            raise ValueError(f"AgentResponse status is not a valid Responses status: {explicit_status!r}")
+        return explicit_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")
+    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:
+    usage_details = result.usage_details
+    if usage_details is None:
+        return _response_field_from_result(result, "usage")
+    if not usage_details:
+        return None
+
+    input_tokens = _usage_count(usage_details, "input_token_count")
+    if input_tokens is None:
+        raise ValueError("AgentResponse usage_details requires `input_token_count` for Responses conversion")
+    output_tokens = _usage_count(usage_details, "output_token_count")
+    if output_tokens is None:
+        raise ValueError("AgentResponse usage_details requires `output_token_count` for Responses conversion")
+    total_tokens = _usage_count(usage_details, "total_token_count")
+    cached_tokens = _usage_count(usage_details, "cache_read_input_token_count") or 0
+    cache_write_tokens = _usage_count(usage_details, "cache_creation_input_token_count") or 0
+    reasoning_tokens = _usage_count(usage_details, "reasoning_output_token_count") or 0
+    usage: dict[str, Any] = {
+        "input_tokens": input_tokens,
+        "input_tokens_details": {
+            "cached_tokens": cached_tokens,
+            "cache_write_tokens": cache_write_tokens,
+        },
+        "output_tokens": output_tokens,
+        "output_tokens_details": {"reasoning_tokens": reasoning_tokens},
+        "total_tokens": total_tokens if total_tokens is not None else input_tokens + output_tokens,
+    }
+    return usage
+
+
+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")
+    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) -> Any | None:
+    if key in result.additional_properties:
+        return result.additional_properties[key]
+
+    raw = result.raw_representation
+    for source in (raw, getattr(raw, "raw_representation", None)):
+        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
+    return None
+
+
 def _model_from_update(update: AgentResponseUpdate) -> str | None:
     """Best-effort model id from one streamed update's raw representation.
 
@@ -305,19 +470,14 @@ 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))
+        if message.role != "assistant":
+            raise ValueError(f"Responses output messages require the `assistant` role; received {message.role!r}")
+        output_items.extend(_contents_to_output_items(message.contents, status=status))
 
     return output_items
 
@@ -408,8 +568,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 +823,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 []
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..d189c07d076 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,11 @@
 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 agent_framework_hosting_responses import (
     create_conversation_id,
@@ -41,6 +42,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 +73,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 +133,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 +159,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 +217,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 +229,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 +249,35 @@ 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_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 +317,98 @@ 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:
+        raw = SimpleNamespace(
+            status="failed",
+            metadata={"source": "raw"},
+            usage={
+                "input_tokens": 7,
+                "input_tokens_details": {"cached_tokens": 1, "cache_write_tokens": 0},
+                "output_tokens": 2,
+                "output_tokens_details": {"reasoning_tokens": 0},
+                "total_tokens": 9,
+            },
+        )
+        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["usage"]["total_tokens"] == 9
+
+    @pytest.mark.parametrize(
+        "additional_properties",
+        [
+            {"status": "unknown"},
+            {"metadata": {"attempt": 1}},
+        ],
+    )
+    def test_responses_from_run_rejects_invalid_response_fields(
+        self,
+        additional_properties: dict[str, object],
+    ) -> None:
+        result = AgentResponse(
+            messages=Message(role="assistant", contents=[Content.from_text("hello")]),
+            additional_properties=additional_properties,
+        )
+
+        with pytest.raises(ValueError):
+            responses_from_run(result, response_id="resp_new")
+
+    @pytest.mark.parametrize(
+        ("usage_details", "message"),
+        [
+            ({"output_token_count": 1}, "input_token_count"),
+            ({"input_token_count": 1}, "output_token_count"),
+            ({"input_token_count": -1, "output_token_count": 1}, "non-negative"),
+        ],
+    )
+    def test_responses_from_run_rejects_unrepresentable_usage(
+        self,
+        usage_details: UsageDetails,
+        message: str,
+    ) -> None:
+        result = AgentResponse(
+            messages=Message(role="assistant", contents=[Content.from_text("hello")]),
+            usage_details=usage_details,
+        )
+
+        with pytest.raises(ValueError, match=message):
+            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 +478,30 @@ 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,
+            )
+            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
+
     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,

From 31505a4331abb892bbd9700f4f75cc186c8795b8 Mon Sep 17 00:00:00 2001
From: eavanvalkenburg <github@vanvalkenburg.eu>
Date: Mon, 31 Aug 2026 14:10:11 +0200
Subject: [PATCH 2/6] Python: preserve tool-role Responses output items

Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com>
---
 .../_parsing.py                                  |  5 +++--
 .../tests/hosting_responses/test_parsing.py      | 16 ++++++++++++++++
 2 files changed, 19 insertions(+), 2 deletions(-)

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 b69c612d8d1..8eaac2e6a84 100644
--- a/python/packages/hosting-responses/agent_framework_hosting_responses/_parsing.py
+++ b/python/packages/hosting-responses/agent_framework_hosting_responses/_parsing.py
@@ -475,9 +475,10 @@ def _messages_to_output_items(messages: Sequence[Any], *, status: str) -> list[R
         if not isinstance(message, Message):
             output_items.extend(_output_to_output_items(message, status=status))
             continue
-        if message.role != "assistant":
+        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(_contents_to_output_items(message.contents, status=status))
+        output_items.extend(message_output_items)
 
     return output_items
 
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 d189c07d076..3511be7dc13 100644
--- a/python/packages/hosting-responses/tests/hosting_responses/test_parsing.py
+++ b/python/packages/hosting-responses/tests/hosting_responses/test_parsing.py
@@ -267,6 +267,22 @@ def test_responses_from_run_rejects_non_assistant_message_role(self) -> None:
         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(

From 32b856f7b8cbaa084a05649485ccca54231d3959 Mon Sep 17 00:00:00 2001
From: eavanvalkenburg <github@vanvalkenburg.eu>
Date: Mon, 31 Aug 2026 15:18:18 +0200
Subject: [PATCH 3/6] Python: fix Responses terminal semantics

Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com>
---
 docs/specs/002-python-hosting-channels.md     |  12 +-
 python/packages/hosting-responses/README.md   |   6 +
 .../_parsing.py                               |  81 +++++++++----
 .../tests/hosting_responses/test_parsing.py   | 112 ++++++++++++++----
 4 files changed, 164 insertions(+), 47 deletions(-)

diff --git a/docs/specs/002-python-hosting-channels.md b/docs/specs/002-python-hosting-channels.md
index 60c303b728c..c1d9bda072e 100644
--- a/docs/specs/002-python-hosting-channels.md
+++ b/docs/specs/002-python-hosting-channels.md
@@ -248,12 +248,16 @@ The helper only extracts the candidate key. App code decides whether to trust an
 - `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. Partial `UsageDetails` values are accepted; missing required Responses input/output counters are
+rendered as zero, and a missing total is derived from those counters.
 
 `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`
 
diff --git a/python/packages/hosting-responses/README.md b/python/packages/hosting-responses/README.md
index ab17e6061b9..cb6f00ce4c3 100644
--- a/python/packages/hosting-responses/README.md
+++ b/python/packages/hosting-responses/README.md
@@ -16,6 +16,12 @@ 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 partial usage values render absent required token counters as zero.
+
 FastAPI/Starlette/Django/Azure Functions code owns route registration,
 authentication, status codes, response construction, and background work.
 
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 8eaac2e6a84..ac592b63531 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
@@ -59,6 +59,7 @@
 _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:
@@ -304,11 +305,11 @@ def responses_from_run(
 
 
 def _status_from_result(result: AgentResponse[Any]) -> str:
-    explicit_status = _response_field_from_result(result, "status")
-    if explicit_status is not None:
-        if not isinstance(explicit_status, str) or explicit_status not in _RESPONSES_STATUSES:
-            raise ValueError(f"AgentResponse status is not a valid Responses status: {explicit_status!r}")
-        return explicit_status
+    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"):
@@ -319,7 +320,7 @@ def _status_from_result(result: AgentResponse[Any]) -> str:
 
 
 def _metadata_from_result(result: AgentResponse[Any]) -> dict[str, str] | None:
-    metadata = _response_field_from_result(result, "metadata")
+    metadata = _response_field_from_result(result, "metadata", allow_additional_properties=True)
     if metadata is None:
         return None
     if not isinstance(metadata, Mapping):
@@ -333,16 +334,12 @@ def _metadata_from_result(result: AgentResponse[Any]) -> dict[str, str] | None:
 def _usage_from_result(result: AgentResponse[Any]) -> Any | None:
     usage_details = result.usage_details
     if usage_details is None:
-        return _response_field_from_result(result, "usage")
+        return _response_field_from_result(result, "usage", allow_additional_properties=True)
     if not usage_details:
         return None
 
-    input_tokens = _usage_count(usage_details, "input_token_count")
-    if input_tokens is None:
-        raise ValueError("AgentResponse usage_details requires `input_token_count` for Responses conversion")
-    output_tokens = _usage_count(usage_details, "output_token_count")
-    if output_tokens is None:
-        raise ValueError("AgentResponse usage_details requires `output_token_count` for Responses conversion")
+    input_tokens = _usage_count(usage_details, "input_token_count") or 0
+    output_tokens = _usage_count(usage_details, "output_token_count") or 0
     total_tokens = _usage_count(usage_details, "total_token_count")
     cached_tokens = _usage_count(usage_details, "cache_read_input_token_count") or 0
     cache_write_tokens = _usage_count(usage_details, "cache_creation_input_token_count") or 0
@@ -370,7 +367,7 @@ def _usage_count(usage_details: Mapping[str, Any], key: str) -> int | None:
 
 
 def _incomplete_details_from_result(result: AgentResponse[Any], *, status: str) -> Any | None:
-    incomplete_details = _response_field_from_result(result, "incomplete_details")
+    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":
@@ -384,21 +381,52 @@ def _incomplete_details_from_result(result: AgentResponse[Any], *, status: str)
     return None
 
 
-def _response_field_from_result(result: AgentResponse[Any], key: str) -> Any | None:
-    if key in result.additional_properties:
-        return result.additional_properties[key]
-
-    raw = result.raw_representation
-    for source in (raw, getattr(raw, "raw_representation", 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.
 
@@ -1066,10 +1094,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_parsing.py b/python/packages/hosting-responses/tests/hosting_responses/test_parsing.py
index 3511be7dc13..76daadaa2d3 100644
--- a/python/packages/hosting-responses/tests/hosting_responses/test_parsing.py
+++ b/python/packages/hosting-responses/tests/hosting_responses/test_parsing.py
@@ -363,7 +363,8 @@ def test_responses_from_run_preserves_status_metadata_and_usage(self) -> None:
         assert payload["output"][0]["status"] == "incomplete"
 
     def test_responses_from_run_uses_raw_response_fields_as_fallback(self) -> None:
-        raw = SimpleNamespace(
+        earlier_response = SimpleNamespace(status="completed")
+        terminal_response = SimpleNamespace(
             status="failed",
             metadata={"source": "raw"},
             usage={
@@ -374,6 +375,10 @@ def test_responses_from_run_uses_raw_response_fields_as_fallback(self) -> None:
                 "total_tokens": 9,
             },
         )
+        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,
@@ -385,44 +390,70 @@ def test_responses_from_run_uses_raw_response_fields_as_fallback(self) -> None:
         assert payload["metadata"] == {"source": "raw"}
         assert payload["usage"]["total_tokens"] == 9
 
-    @pytest.mark.parametrize(
-        "additional_properties",
-        [
-            {"status": "unknown"},
-            {"metadata": {"attempt": 1}},
-        ],
-    )
-    def test_responses_from_run_rejects_invalid_response_fields(
-        self,
-        additional_properties: dict[str, object],
-    ) -> None:
+    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=additional_properties,
+            additional_properties={"metadata": {"attempt": 1}},
         )
 
         with pytest.raises(ValueError):
             responses_from_run(result, response_id="resp_new")
 
     @pytest.mark.parametrize(
-        ("usage_details", "message"),
+        ("usage_details", "expected_counts"),
         [
-            ({"output_token_count": 1}, "input_token_count"),
-            ({"input_token_count": 1}, "output_token_count"),
-            ({"input_token_count": -1, "output_token_count": 1}, "non-negative"),
+            (
+                {"output_token_count": 3},
+                {"input_tokens": 0, "output_tokens": 3, "total_tokens": 3},
+            ),
+            (
+                {"input_token_count": 2},
+                {"input_tokens": 2, "output_tokens": 0, "total_tokens": 2},
+            ),
+            (
+                {"total_token_count": 5},
+                {"input_tokens": 0, "output_tokens": 0, "total_tokens": 5},
+            ),
         ],
     )
-    def test_responses_from_run_rejects_unrepresentable_usage(
+    def test_responses_from_run_zero_fills_partial_usage(
         self,
         usage_details: UsageDetails,
-        message: str,
+        expected_counts: dict[str, int],
     ) -> None:
         result = AgentResponse(
             messages=Message(role="assistant", contents=[Content.from_text("hello")]),
             usage_details=usage_details,
         )
 
-        with pytest.raises(ValueError, match=message):
+        payload = responses_from_run(result, response_id="resp_new")
+
+        usage = cast("dict[str, object]", payload["usage"])
+        assert {key: usage[key] for key in expected_counts} == expected_counts
+        assert usage["input_tokens_details"] == {"cached_tokens": 0, "cache_write_tokens": 0}
+        assert usage["output_tokens_details"] == {"reasoning_tokens": 0}
+
+    @pytest.mark.parametrize("count", [-1, True])
+    def test_responses_from_run_rejects_invalid_usage_count(self, count: object) -> None:
+        result = AgentResponse(
+            messages=Message(role="assistant", contents=[Content.from_text("hello")]),
+            usage_details=cast("UsageDetails", {"input_token_count": 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:
@@ -518,6 +549,47 @@ def finalizer(items: Sequence[AgentResponseUpdate]) -> AgentResponse:
         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
+
+    @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")

From 8e8d4629623f7428e987dae4aeeab9fd8980c87d Mon Sep 17 00:00:00 2001
From: eavanvalkenburg <github@vanvalkenburg.eu>
Date: Mon, 31 Aug 2026 15:30:03 +0200
Subject: [PATCH 4/6] Python: keep partial Responses usage consistent

Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com>
---
 docs/specs/002-python-hosting-channels.md     |  3 +-
 python/packages/hosting-responses/README.md   |  3 +-
 .../_parsing.py                               | 19 +++++-
 .../tests/hosting_responses/test_parsing.py   | 58 +++++++++++++++++--
 4 files changed, 74 insertions(+), 9 deletions(-)

diff --git a/docs/specs/002-python-hosting-channels.md b/docs/specs/002-python-hosting-channels.md
index c1d9bda072e..b577132f7e6 100644
--- a/docs/specs/002-python-hosting-channels.md
+++ b/docs/specs/002-python-hosting-channels.md
@@ -251,7 +251,8 @@ The helper only extracts the candidate key. App code decides whether to trust an
 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. Partial `UsageDetails` values are accepted; missing required Responses input/output counters are
-rendered as zero, and a missing total is derived from those counters.
+derived from their detail counters or rendered as zero, and a missing total is derived from the input/output counters.
+Explicit counters that would undercount their details or total are invalid.
 
 `responses_from_streaming_run(...)` renders Server-Sent Event strings for a `ResponseStream`. It emits a created event,
 text deltas, and a terminal event matching the final `completed`, `incomplete`, or `failed` status. A stream that finalizes
diff --git a/python/packages/hosting-responses/README.md b/python/packages/hosting-responses/README.md
index cb6f00ce4c3..ce4c3221355 100644
--- a/python/packages/hosting-responses/README.md
+++ b/python/packages/hosting-responses/README.md
@@ -20,7 +20,8 @@ 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 partial usage values render absent required token counters as zero.
+and partial usage values derive absent parent counters from their details or
+render them as zero. Explicitly inconsistent usage counters are rejected.
 
 FastAPI/Starlette/Django/Azure Functions code owns route registration,
 authentication, status codes, response construction, and background work.
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 ac592b63531..1d27dd27742 100644
--- a/python/packages/hosting-responses/agent_framework_hosting_responses/_parsing.py
+++ b/python/packages/hosting-responses/agent_framework_hosting_responses/_parsing.py
@@ -338,12 +338,25 @@ def _usage_from_result(result: AgentResponse[Any]) -> Any | None:
     if not usage_details:
         return None
 
-    input_tokens = _usage_count(usage_details, "input_token_count") or 0
-    output_tokens = _usage_count(usage_details, "output_token_count") or 0
+    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") or 0
     cache_write_tokens = _usage_count(usage_details, "cache_creation_input_token_count") or 0
     reasoning_tokens = _usage_count(usage_details, "reasoning_output_token_count") or 0
+    if input_tokens is None:
+        input_tokens = max(cached_tokens, cache_write_tokens)
+    elif cached_tokens > input_tokens or cache_write_tokens > input_tokens:
+        raise ValueError("AgentResponse input token details must not exceed `input_token_count`")
+    if output_tokens is None:
+        output_tokens = reasoning_tokens
+    elif reasoning_tokens > output_tokens:
+        raise ValueError("AgentResponse reasoning token count must not exceed `output_token_count`")
+    minimum_total = input_tokens + output_tokens
+    if total_tokens is None:
+        total_tokens = minimum_total
+    elif total_tokens < minimum_total:
+        raise ValueError("AgentResponse `total_token_count` must not be less than input plus output tokens")
     usage: dict[str, Any] = {
         "input_tokens": input_tokens,
         "input_tokens_details": {
@@ -352,7 +365,7 @@ def _usage_from_result(result: AgentResponse[Any]) -> Any | None:
         },
         "output_tokens": output_tokens,
         "output_tokens_details": {"reasoning_tokens": reasoning_tokens},
-        "total_tokens": total_tokens if total_tokens is not None else input_tokens + output_tokens,
+        "total_tokens": total_tokens,
     }
     return usage
 
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 76daadaa2d3..4db263ff220 100644
--- a/python/packages/hosting-responses/tests/hosting_responses/test_parsing.py
+++ b/python/packages/hosting-responses/tests/hosting_responses/test_parsing.py
@@ -413,26 +413,46 @@ def test_responses_from_run_rejects_invalid_metadata(self) -> None:
             responses_from_run(result, response_id="resp_new")
 
     @pytest.mark.parametrize(
-        ("usage_details", "expected_counts"),
+        ("usage_details", "expected_counts", "cached_tokens", "reasoning_tokens"),
         [
             (
                 {"output_token_count": 3},
                 {"input_tokens": 0, "output_tokens": 3, "total_tokens": 3},
+                0,
+                0,
             ),
             (
                 {"input_token_count": 2},
                 {"input_tokens": 2, "output_tokens": 0, "total_tokens": 2},
+                0,
+                0,
             ),
             (
                 {"total_token_count": 5},
                 {"input_tokens": 0, "output_tokens": 0, "total_tokens": 5},
+                0,
+                0,
+            ),
+            (
+                {"output_token_count": 1, "cache_read_input_token_count": 3},
+                {"input_tokens": 3, "output_tokens": 1, "total_tokens": 4},
+                3,
+                0,
+            ),
+            (
+                {"input_token_count": 1, "reasoning_output_token_count": 2},
+                {"input_tokens": 1, "output_tokens": 2, "total_tokens": 3},
+                0,
+                2,
             ),
         ],
     )
-    def test_responses_from_run_zero_fills_partial_usage(
+    def test_responses_from_run_normalizes_partial_usage(
         self,
         usage_details: UsageDetails,
         expected_counts: dict[str, int],
+        cached_tokens: int,
+        reasoning_tokens: int,
     ) -> None:
         result = AgentResponse(
             messages=Message(role="assistant", contents=[Content.from_text("hello")]),
@@ -443,8 +463,38 @@ def test_responses_from_run_zero_fills_partial_usage(
 
         usage = cast("dict[str, object]", payload["usage"])
         assert {key: usage[key] for key in expected_counts} == expected_counts
-        assert usage["input_tokens_details"] == {"cached_tokens": 0, "cache_write_tokens": 0}
-        assert usage["output_tokens_details"] == {"reasoning_tokens": 0}
+        assert usage["input_tokens_details"] == {"cached_tokens": cached_tokens, "cache_write_tokens": 0}
+        assert usage["output_tokens_details"] == {"reasoning_tokens": reasoning_tokens}
+
+    @pytest.mark.parametrize(
+        ("usage_details", "message"),
+        [
+            (
+                {"input_token_count": 2, "cache_read_input_token_count": 3},
+                "input token details",
+            ),
+            (
+                {"output_token_count": 1, "reasoning_output_token_count": 2},
+                "reasoning token count",
+            ),
+            (
+                {"input_token_count": 2, "output_token_count": 3, "total_token_count": 4},
+                "total_token_count",
+            ),
+        ],
+    )
+    def test_responses_from_run_rejects_inconsistent_usage(
+        self,
+        usage_details: UsageDetails,
+        message: str,
+    ) -> None:
+        result = AgentResponse(
+            messages=Message(role="assistant", contents=[Content.from_text("hello")]),
+            usage_details=usage_details,
+        )
+
+        with pytest.raises(ValueError, match=message):
+            responses_from_run(result, response_id="resp_new")
 
     @pytest.mark.parametrize("count", [-1, True])
     def test_responses_from_run_rejects_invalid_usage_count(self, count: object) -> None:

From b0d5432228051a99885be22bf0222852bc34a57d Mon Sep 17 00:00:00 2001
From: eavanvalkenburg <github@vanvalkenburg.eu>
Date: Tue, 1 Sep 2026 09:23:47 +0200
Subject: [PATCH 5/6] Python: preserve Responses usage provenance

Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com>

Copilot-Session: 1fa98222-11d1-4116-99db-32de50d72dee
---
 docs/specs/002-python-hosting-channels.md     |   8 +-
 python/packages/hosting-responses/README.md   |   7 +-
 .../_parsing.py                               |  37 +--
 .../tests/hosting_responses/test_parsing.py   | 253 ++++++++++++++----
 4 files changed, 236 insertions(+), 69 deletions(-)

diff --git a/docs/specs/002-python-hosting-channels.md b/docs/specs/002-python-hosting-channels.md
index b577132f7e6..90f3f932bbc 100644
--- a/docs/specs/002-python-hosting-channels.md
+++ b/docs/specs/002-python-hosting-channels.md
@@ -250,9 +250,11 @@ The helper only extracts the candidate key. App code decides whether to trust an
 `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. 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. Partial `UsageDetails` values are accepted; missing required Responses input/output counters are
-derived from their detail counters or rendered as zero, and a missing total is derived from the input/output counters.
-Explicit counters that would undercount their details or total are invalid.
+may flatten user metadata. Failed transport responses preserve the structured error from that same raw representation.
+Usage is emitted only when `UsageDetails` explicitly supplies every Responses-required input, output, cache-read,
+cache-creation, and reasoning counter. 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 terminal event matching the final `completed`, `incomplete`, or `failed` status. A stream that finalizes
diff --git a/python/packages/hosting-responses/README.md b/python/packages/hosting-responses/README.md
index ce4c3221355..8ac09ccecd1 100644
--- a/python/packages/hosting-responses/README.md
+++ b/python/packages/hosting-responses/README.md
@@ -20,8 +20,11 @@ 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 partial usage values derive absent parent counters from their details or
-render them as zero. Explicitly inconsistent usage counters are rejected.
+and failed transport responses preserve their structured error. Usage is emitted
+only when Agent Framework reports every Responses-required input, output, cache,
+and reasoning counter. 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.
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 1d27dd27742..0002d1c991b 100644
--- a/python/packages/hosting-responses/agent_framework_hosting_responses/_parsing.py
+++ b/python/packages/hosting-responses/agent_framework_hosting_responses/_parsing.py
@@ -297,6 +297,8 @@ def responses_from_run(
         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:
@@ -341,22 +343,27 @@ def _usage_from_result(result: AgentResponse[Any]) -> Any | 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") or 0
-    cache_write_tokens = _usage_count(usage_details, "cache_creation_input_token_count") or 0
-    reasoning_tokens = _usage_count(usage_details, "reasoning_output_token_count") or 0
-    if input_tokens is None:
-        input_tokens = max(cached_tokens, cache_write_tokens)
-    elif cached_tokens > input_tokens or cache_write_tokens > input_tokens:
-        raise ValueError("AgentResponse input token details must not exceed `input_token_count`")
-    if output_tokens is None:
-        output_tokens = reasoning_tokens
-    elif reasoning_tokens > output_tokens:
-        raise ValueError("AgentResponse reasoning token count must not exceed `output_token_count`")
-    minimum_total = input_tokens + output_tokens
+    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
+        or cached_tokens is None
+        or cache_write_tokens is None
+        or reasoning_tokens is None
+    ):
+        return None
+
+    expected_total = input_tokens + output_tokens
+    if (
+        cached_tokens + cache_write_tokens > input_tokens
+        or reasoning_tokens > output_tokens
+        or (total_tokens is not None and total_tokens != expected_total)
+    ):
+        return None
     if total_tokens is None:
-        total_tokens = minimum_total
-    elif total_tokens < minimum_total:
-        raise ValueError("AgentResponse `total_token_count` must not be less than input plus output tokens")
+        total_tokens = expected_total
     usage: dict[str, Any] = {
         "input_tokens": input_tokens,
         "input_tokens_details": {
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 4db263ff220..a285fe8cc70 100644
--- a/python/packages/hosting-responses/tests/hosting_responses/test_parsing.py
+++ b/python/packages/hosting-responses/tests/hosting_responses/test_parsing.py
@@ -367,6 +367,7 @@ def test_responses_from_run_uses_raw_response_fields_as_fallback(self) -> None:
         terminal_response = SimpleNamespace(
             status="failed",
             metadata={"source": "raw"},
+            error={"code": "server_error", "message": "provider failed"},
             usage={
                 "input_tokens": 7,
                 "input_tokens_details": {"cached_tokens": 1, "cache_write_tokens": 0},
@@ -388,6 +389,7 @@ def test_responses_from_run_uses_raw_response_fields_as_fallback(self) -> None:
 
         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_does_not_treat_user_metadata_status_as_transport_status(self) -> None:
@@ -413,46 +415,122 @@ def test_responses_from_run_rejects_invalid_metadata(self) -> None:
             responses_from_run(result, response_id="resp_new")
 
     @pytest.mark.parametrize(
-        ("usage_details", "expected_counts", "cached_tokens", "reasoning_tokens"),
+        ("usage_details", "expected_usage"),
         [
             (
-                {"output_token_count": 3},
-                {"input_tokens": 0, "output_tokens": 3, "total_tokens": 3},
-                0,
-                0,
+                {
+                    "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": 2},
-                {"input_tokens": 2, "output_tokens": 0, "total_tokens": 2},
-                0,
-                0,
+                {
+                    "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,
+                },
             ),
-            (
-                {"total_token_count": 5},
-                {"input_tokens": 0, "output_tokens": 0, "total_tokens": 5},
-                0,
-                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",
             ),
-            (
-                {"output_token_count": 1, "cache_read_input_token_count": 3},
-                {"input_tokens": 3, "output_tokens": 1, "total_tokens": 4},
-                3,
-                0,
+            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",
             ),
-            (
-                {"input_token_count": 1, "reasoning_output_token_count": 2},
-                {"input_tokens": 1, "output_tokens": 2, "total_tokens": 3},
-                0,
-                2,
+            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,
+                    "reasoning_output_token_count": 0,
+                },
+                id="missing-cache-creation",
+            ),
+            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_normalizes_partial_usage(
+    def test_responses_from_run_omits_partial_usage(
         self,
         usage_details: UsageDetails,
-        expected_counts: dict[str, int],
-        cached_tokens: int,
-        reasoning_tokens: int,
     ) -> None:
         result = AgentResponse(
             messages=Message(role="assistant", contents=[Content.from_text("hello")]),
@@ -461,46 +539,97 @@ def test_responses_from_run_normalizes_partial_usage(
 
         payload = responses_from_run(result, response_id="resp_new")
 
-        usage = cast("dict[str, object]", payload["usage"])
-        assert {key: usage[key] for key in expected_counts} == expected_counts
-        assert usage["input_tokens_details"] == {"cached_tokens": cached_tokens, "cache_write_tokens": 0}
-        assert usage["output_tokens_details"] == {"reasoning_tokens": reasoning_tokens}
+        assert "usage" not in payload
 
     @pytest.mark.parametrize(
-        ("usage_details", "message"),
+        "usage_details",
         [
-            (
-                {"input_token_count": 2, "cache_read_input_token_count": 3},
-                "input token 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",
             ),
-            (
-                {"output_token_count": 1, "reasoning_output_token_count": 2},
-                "reasoning token 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",
             ),
-            (
-                {"input_token_count": 2, "output_token_count": 3, "total_token_count": 4},
-                "total_token_count",
+            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_rejects_inconsistent_usage(
+    def test_responses_from_run_omits_usage_that_cannot_map_consistently(
         self,
         usage_details: UsageDetails,
-        message: str,
     ) -> None:
         result = AgentResponse(
             messages=Message(role="assistant", contents=[Content.from_text("hello")]),
             usage_details=usage_details,
         )
 
-        with pytest.raises(ValueError, match=message):
-            responses_from_run(result, response_id="resp_new")
+        payload = responses_from_run(result, response_id="resp_new")
 
-    @pytest.mark.parametrize("count", [-1, True])
-    def test_responses_from_run_rejects_invalid_usage_count(self, count: object) -> None:
+        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", {"input_token_count": count}),
+            usage_details=cast("UsageDetails", {key: count}),
         )
 
         with pytest.raises(ValueError, match="non-negative integer"):
@@ -585,6 +714,9 @@ def finalizer(items: Sequence[AgentResponseUpdate]) -> AgentResponse:
                 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
@@ -619,6 +751,29 @@ def finalizer(items: Sequence[AgentResponseUpdate]) -> AgentResponse:
         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]:

From 39c6c1e11b003b7287948bf52d3311b4ebb70e84 Mon Sep 17 00:00:00 2001
From: eavanvalkenburg <github@vanvalkenburg.eu>
Date: Tue, 1 Sep 2026 11:41:20 +0200
Subject: [PATCH 6/6] Python: preserve native Responses usage

Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com>

Copilot-Session: 1fa98222-11d1-4116-99db-32de50d72dee
---
 docs/specs/002-python-hosting-channels.md     |  10 +-
 python/packages/hosting-responses/README.md   |  12 +-
 .../_parsing.py                               |  69 +++++++----
 .../tests/hosting_responses/test_parsing.py   | 108 +++++++++++++++---
 4 files changed, 152 insertions(+), 47 deletions(-)

diff --git a/docs/specs/002-python-hosting-channels.md b/docs/specs/002-python-hosting-channels.md
index 90f3f932bbc..a432cfa122b 100644
--- a/docs/specs/002-python-hosting-channels.md
+++ b/docs/specs/002-python-hosting-channels.md
@@ -251,10 +251,12 @@ The helper only extracts the candidate key. App code decides whether to trust an
 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.
-Usage is emitted only when `UsageDetails` explicitly supplies every Responses-required input, output, cache-read,
-cache-creation, and reasoning counter. 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.
+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 terminal event matching the final `completed`, `incomplete`, or `failed` status. A stream that finalizes
diff --git a/python/packages/hosting-responses/README.md b/python/packages/hosting-responses/README.md
index 8ac09ccecd1..936bfe1d883 100644
--- a/python/packages/hosting-responses/README.md
+++ b/python/packages/hosting-responses/README.md
@@ -20,11 +20,13 @@ 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. Usage is emitted
-only when Agent Framework reports every Responses-required input, output, cache,
-and reasoning counter. 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.
+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.
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 0002d1c991b..3536fc83d52 100644
--- a/python/packages/hosting-responses/agent_framework_hosting_responses/_parsing.py
+++ b/python/packages/hosting-responses/agent_framework_hosting_responses/_parsing.py
@@ -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 = {
@@ -334,9 +336,10 @@ def _metadata_from_result(result: AgentResponse[Any]) -> dict[str, str] | None:
 
 
 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 usage_details is None:
-        return _response_field_from_result(result, "usage", allow_additional_properties=True)
     if not usage_details:
         return None
 
@@ -346,35 +349,57 @@ def _usage_from_result(result: AgentResponse[Any]) -> Any | None:
     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
-        or cached_tokens is None
-        or cache_write_tokens is None
-        or reasoning_tokens is None
-    ):
+    if input_tokens is None or output_tokens is None:
         return None
 
     expected_total = input_tokens + output_tokens
-    if (
-        cached_tokens + cache_write_tokens > input_tokens
-        or reasoning_tokens > output_tokens
-        or (total_tokens is not None and total_tokens != expected_total)
-    ):
+    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
-    usage: dict[str, Any] = {
+
+    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": {
-            "cached_tokens": cached_tokens,
-            "cache_write_tokens": cache_write_tokens,
-        },
+        "input_tokens_details": input_tokens_details,
         "output_tokens": output_tokens,
-        "output_tokens_details": {"reasoning_tokens": reasoning_tokens},
+        "output_tokens_details": output_tokens_details,
         "total_tokens": total_tokens,
-    }
-    return usage
+    })
+
+
+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:
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 a285fe8cc70..765694ec437 100644
--- a/python/packages/hosting-responses/tests/hosting_responses/test_parsing.py
+++ b/python/packages/hosting-responses/tests/hosting_responses/test_parsing.py
@@ -12,6 +12,7 @@
 
 import pytest
 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,
@@ -29,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")
@@ -365,16 +379,11 @@ def test_responses_from_run_preserves_status_metadata_and_usage(self) -> None:
     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={
-                "input_tokens": 7,
-                "input_tokens_details": {"cached_tokens": 1, "cache_write_tokens": 0},
-                "output_tokens": 2,
-                "output_tokens_details": {"reasoning_tokens": 0},
-                "total_tokens": 9,
-            },
+            usage=_native_usage_payload(),
         )
         raw = [
             SimpleNamespace(raw_representation=SimpleNamespace(response=earlier_response)),
@@ -392,6 +401,58 @@ def test_responses_from_run_uses_raw_response_fields_as_fallback(self) -> None:
         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(
@@ -508,15 +569,6 @@ def test_responses_from_run_maps_complete_usage_without_cross_filling(
                 },
                 id="missing-cache-read",
             ),
-            pytest.param(
-                {
-                    "input_token_count": 0,
-                    "output_token_count": 0,
-                    "cache_read_input_token_count": 0,
-                    "reasoning_output_token_count": 0,
-                },
-                id="missing-cache-creation",
-            ),
             pytest.param(
                 {
                     "input_token_count": 0,
@@ -541,6 +593,30 @@ def test_responses_from_run_omits_partial_usage(
 
         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",
         [