Azure · banibrata-de · Sep 17, 2026 · Sep 18, 2026 · Sep 18, 2026 · Sep 18, 2026
@@ -2,6 +2,11 @@
 
 ## 2.2.0b2 (Unreleased)
 
+### Features Added
+
+- Added optional response metadata to `ResponseEventStream.emit_failed`. Invalid failure metadata is logged and
+  omitted so metadata validation cannot suppress the original failure response.
+
 ### Bugs Fixed
 
 - Scoped durable multi-turn task IDs with `FOUNDRY_AGENT_SESSION_GUID` when

@@ -0,0 +1,10 @@
+# Copyright (c) Microsoft Corporation.
+# Licensed under the MIT license.
+"""Shared constraints for Responses metadata."""
+
+# These limits are defined by the Responses Metadata contract generated in
+# models._generated.types.Metadata. Keep runtime validation centralized here
+# so a future contract revision requires one implementation change.
+MAX_METADATA_KEYS = 16
+MAX_METADATA_KEY_LENGTH = 64
+MAX_METADATA_VALUE_LENGTH = 512
@@ -15,6 +15,11 @@
     PLATFORM_ERROR_TAG,
 )
 from .._id_generator import IdGenerator
+from .._metadata_constraints import (
+    MAX_METADATA_KEYS,
+    MAX_METADATA_KEY_LENGTH,
+    MAX_METADATA_VALUE_LENGTH,
+)
 from .._options import ResponsesServerOptions
 
 from .. import models as _public_models
@@ -108,25 +113,26 @@ def validate_create_response(request: _public_models.CreateResponse) -> None:
 
     # B22: model is optional — resolved to default in normalize_create_response()
 
-    # Metadata constraints: ≤16 keys, key ≤64 chars, value ≤512 chars
     metadata = request.get("metadata")
     if isinstance(metadata, Mapping):
-        if len(metadata) > 16:
+        if len(metadata) > MAX_METADATA_KEYS:
             raise RequestValidationError(
-                "metadata must have at most 16 key-value pairs",
+                f"metadata must have at most {MAX_METADATA_KEYS} key-value pairs",
                 code="invalid_request",
                 param="metadata",
             )
         for key, value in metadata.items():
-            if isinstance(key, str) and len(key) > 64:
+            if isinstance(key, str) and len(key) > MAX_METADATA_KEY_LENGTH:
                 raise RequestValidationError(
-                    f"metadata key '{key[:64]}...' exceeds maximum length of 64 characters",
+                    f"metadata key '{key[:MAX_METADATA_KEY_LENGTH]}...' exceeds maximum length of "
+                    f"{MAX_METADATA_KEY_LENGTH} characters",
                     code="invalid_request",
                     param="metadata",
                 )
-            if isinstance(value, str) and len(value) > 512:
+            if isinstance(value, str) and len(value) > MAX_METADATA_VALUE_LENGTH:
                 raise RequestValidationError(
-                    f"metadata value for key '{key}' exceeds maximum length of 512 characters",
+                    f"metadata value for key '{key}' exceeds maximum length of "
+                    f"{MAX_METADATA_VALUE_LENGTH} characters",
                     code="invalid_request",
                     param="metadata",
                 )

@@ -4,7 +4,8 @@
 
 from __future__ import annotations
 
-from collections.abc import MutableMapping
+import logging
+from collections.abc import Mapping, MutableMapping
 from copy import deepcopy
 from datetime import datetime, timezone
 from typing import Any, Iterator, Sequence, cast
@@ -13,6 +14,11 @@
 
 
 from .._id_generator import IdGenerator
+from .._metadata_constraints import (
+    MAX_METADATA_KEYS,
+    MAX_METADATA_KEY_LENGTH,
+    MAX_METADATA_VALUE_LENGTH,
+)
 from . import _internals
 from ._builders import (
     OutputItemBuilder,
@@ -35,6 +41,7 @@
 # Event types whose payload is a full Response snapshot.
 # Lifecycle events nest under a "response" key on the wire.
 _RESPONSE_SNAPSHOT_EVENT_TYPES = _internals._RESPONSE_SNAPSHOT_EVENT_TYPES  # pylint: disable=protected-access
+_LOGGER = logging.getLogger(__name__)
 
 
 def _resolve_conversation_param(raw: Any) -> str | None:
@@ -75,6 +82,43 @@ def _require_wire_dict(obj: Any, field_name: str) -> dict[str, Any]:
     return obj
 
 
+def _merge_response_metadata(response: dict[str, Any], metadata: Mapping[str, str]) -> None:
+    """Merge validated public metadata into a response envelope.
+
+    :param response: Mutable response envelope receiving the metadata.
+    :type response: dict[str, ~typing.Any]
+    :param metadata: Public metadata values to validate and merge.
+    :type metadata: ~collections.abc.Mapping[str, str]
+    """
+    current = response.get("metadata")
+    if current is None:
+        merged: dict[str, str] = {}
+    elif isinstance(current, dict):
+        merged = dict(current)
+    else:
+        raise TypeError("response metadata must be a mapping")
+
+    for key, value in metadata.items():
+        if not isinstance(key, str):
+            raise TypeError(f"metadata keys must be str, got {type(key).__name__}")
+        if not isinstance(value, str):
+            raise TypeError(f"metadata values must be str, got {type(value).__name__}")
+        if len(key) > MAX_METADATA_KEY_LENGTH:
+            raise ValueError(
+                f"metadata key exceeds the {MAX_METADATA_KEY_LENGTH}-character limit: "
+                f"{key[:MAX_METADATA_KEY_LENGTH]}..."
+            )
+        if len(value) > MAX_METADATA_VALUE_LENGTH:
+            raise ValueError(
+                f"metadata value for key '{key}' exceeds the {MAX_METADATA_VALUE_LENGTH}-character limit"
+            )
+        merged[key] = value
+
+    if len(merged) > MAX_METADATA_KEYS:
+        raise ValueError(f"response metadata must have at most {MAX_METADATA_KEYS} key-value pairs")
+    response["metadata"] = merged
+
+
 class _MutableResponseDict(dict[str, Any]):
     """Mutable response wire payload with legacy attribute access."""
 
@@ -331,6 +375,7 @@ def emit_failed(
         *,
         code: str = "server_error",
         message: str = "An internal server error occurred.",
+        metadata: Mapping[str, str] | None = None,
         usage: response_models.ResponseUsage | None = None,
     ) -> response_models.ResponseFailedEvent:
         """Emit a ``response.failed`` terminal lifecycle event.
@@ -339,6 +384,9 @@ def emit_failed(
         :keyword type code: str | ~azure.ai.agentserver.responses.models.ResponseErrorCode
         :keyword message: Human-readable error message.
         :keyword type message: str
+        :keyword metadata: Optional public response metadata to merge into the terminal response. Invalid
+            metadata is logged and omitted so it cannot suppress the original failure.
+        :keyword type metadata: ~collections.abc.Mapping[str, str] | None
         :keyword usage: Optional usage statistics to attach to the response.
         :keyword type usage: ~azure.ai.agentserver.responses.models.ResponseUsage | None
         :returns: The emitted event model instance.
@@ -350,6 +398,15 @@ def emit_failed(
             "code": _internals.enum_value(code),
             "message": message,
         }
+        if metadata is not None:
+            try:
+                _merge_response_metadata(self._response, metadata)
+            except (TypeError, ValueError) as exc:
+                _LOGGER.warning(
+                    "Ignoring invalid metadata supplied to emit_failed; the response failure will be emitted "
+                    "without it: %s",
+                    exc,
+                )
         self._set_terminal_fields(usage=usage)
         return cast(
             "response_models.ResponseFailedEvent",

@@ -8,9 +8,9 @@
 from collections.abc import Iterator, MutableMapping
 from typing import Any
 
+from .._metadata_constraints import MAX_METADATA_KEYS, MAX_METADATA_VALUE_LENGTH
+
 _RESERVED_KEY = "_internal_metadata"
-_MAX_METADATA_KEYS = 16
-_MAX_VALUE_LEN = 512
 
 
 class _ResponseInternalMetadataView(MutableMapping[str, Any]):
@@ -52,20 +52,20 @@ def _store(self, value: dict[str, Any]) -> None:
             separators=(",", ":"),
             sort_keys=True,
         )
-        if len(encoded) > _MAX_VALUE_LEN:
+        if len(encoded) > MAX_METADATA_VALUE_LENGTH:
             raise ValueError(
                 f"internal_metadata encodes to {len(encoded)} chars, exceeding the "
-                f"{_MAX_VALUE_LEN}-char limit of the response metadata value"
+                f"{MAX_METADATA_VALUE_LENGTH}-char limit of the response metadata value"
             )
 
         if not isinstance(metadata, dict):
             metadata = {}
             self._response["metadata"] = metadata
         projected_key_count = len(metadata) + (0 if _RESERVED_KEY in metadata else 1)
-        if projected_key_count > _MAX_METADATA_KEYS:
+        if projected_key_count > MAX_METADATA_KEYS:
             raise ValueError(
                 f"cannot add internal_metadata: response metadata already has "
-                f"{len(metadata)} keys (limit {_MAX_METADATA_KEYS})"
+                f"{len(metadata)} keys (limit {MAX_METADATA_KEYS})"
             )
         metadata[_RESERVED_KEY] = encoded
 

@@ -8,7 +8,7 @@
 
 from starlette.testclient import TestClient
 
-from azure.ai.agentserver.responses import ResponsesAgentServerHost
+from azure.ai.agentserver.responses import ResponseEventStream, ResponsesAgentServerHost
 from tests._helpers import poll_until
 
 
@@ -582,6 +582,36 @@ async def _events():
     ), f"S-015: synthesised terminal must set status to 'failed', got {payload.get('status')!r}"
 
 
+def test_sync_failed_terminal_preserves_public_metadata() -> None:
+    async def _failed_handler(request: Any, context: Any, cancellation_signal: asyncio.Event):
+        async def _events():
+            stream = ResponseEventStream(response_id=context.response_id, request=request)
+            yield stream.emit_created()
+            yield stream.emit_in_progress()
+            yield stream.emit_failed(
+                code="budget_exceeded",
+                message="Budget exceeded",
+                metadata={"costControl": '{"x-ms-budget-cause":"budget_exceeded"}'},
+            )
+
+        return _events()
+
+    app = ResponsesAgentServerHost()
+    app.response_handler(_failed_handler)
+    client = TestClient(app)
+
+    response = client.post(
+        "/responses",
+        json={"model": "gpt-4o-mini", "input": "hello", "stream": False, "store": True, "background": False},
+    )
+
+    assert response.status_code == 200
+    payload = response.json()
+    assert payload["status"] == "failed"
+    assert payload["error"]["code"] == "budget_exceeded"
+    assert payload["metadata"]["costControl"] == '{"x-ms-budget-cause":"budget_exceeded"}'
+
+
 # ══════════════════════════════════════════════════════════
 # Phase 5 — Task 5.1: FR-006 / FR-007 first-event contract first-event contract tests
 # ══════════════════════════════════════════════════════════

@@ -417,6 +417,39 @@ def test_streaming__post_creation_error_yields_response_failed_not_error_event()
     ), f"Standalone 'error' event must not appear after response.created. Events: {event_types}"
 
 
+def test_streaming_failed_terminal_preserves_public_metadata() -> None:
+    async def _failed_handler(request: Any, context: Any, cancellation_signal: asyncio.Event):
+        async def _events():
+            stream = ResponseEventStream(response_id=context.response_id, request=request)
+            yield stream.emit_created()
+            yield stream.emit_in_progress()
+            yield stream.emit_failed(
+                code="budget_exceeded",
+                message="Budget exceeded",
+                metadata={"costControl": '{"x-ms-budget-cause":"budget_exceeded"}'},
+            )
+
+        return _events()
+
+    app = ResponsesAgentServerHost()
+    app.response_handler(_failed_handler)
+    client = TestClient(app)
+
+    with client.stream(
+        "POST",
+        "/responses",
+        json={"model": "gpt-4o-mini", "input": "hello", "stream": True, "store": True, "background": False},
+    ) as response:
+        assert response.status_code == 200
+        events = _collect_stream_events(response)
+
+    failed = next(event for event in events if event["type"] == "response.failed")
+    payload = failed["data"]["response"]
+    assert payload["status"] == "failed"
+    assert payload["error"]["code"] == "budget_exceeded"
+    assert payload["metadata"]["costControl"] == '{"x-ms-budget-cause":"budget_exceeded"}'
+
+
 # ══════════════════════════════════════════════════════════
 # Task 4.1 — _process_handler_events pipeline contract tests
 # ══════════════════════════════════════════════════════════

@@ -4,6 +4,7 @@
 
 from __future__ import annotations
 
+import logging
 from typing import cast
 
 import pytest
@@ -170,7 +171,10 @@ def test_event_stream_builder__emit_completed_accepts_usage_and_sets_terminal_fi
 
 
 def test_event_stream_builder__emit_failed_accepts_error_and_usage() -> None:
-    stream = ResponseEventStream(response_id="resp_builder_failed_params")
+    stream = ResponseEventStream(
+        response_id="resp_builder_failed_params",
+        request={"metadata": {"request": "preserved"}},
+    )
     stream.emit_created(status="in_progress")
 
     usage = ResponseUsage(
@@ -181,18 +185,58 @@ def test_event_stream_builder__emit_failed_accepts_error_and_usage() -> None:
         total_tokens=9,
     )
 
-    failed = stream.emit_failed(code="server_error", message="boom", usage=usage)
+    failed = stream.emit_failed(
+        code="server_error",
+        message="boom",
+        metadata={"failure": "details"},
+        usage=usage,
+    )
 
     assert isinstance(failed, dict)
     assert failed["type"] == "response.failed"
     assert failed["response"]["status"] == "failed"
     assert failed["response"]["error"]["code"] == "server_error"
     assert failed["response"]["error"]["message"] == "boom"
+    assert failed["response"]["metadata"] == {
+        "request": "preserved",
+        "failure": "details",
+    }
     assert failed["response"]["usage"]["total_tokens"] == 9
     assert failed["response"]["usage"]["input_tokens_details"]["cache_write_tokens"] == 2
     assert failed["response"].get("completed_at") is None
 
 
+def test_event_stream_builder__emit_failed_ignores_invalid_metadata(caplog: pytest.LogCaptureFixture) -> None:
+    stream = ResponseEventStream(
+        response_id="resp_builder_failed_metadata",
+        request={"metadata": {"request": "preserved"}},
+    )
+    stream.emit_created(status="in_progress")
+
+    with caplog.at_level(logging.WARNING):
+        failed = stream.emit_failed(metadata={"invalid": 1})  # type: ignore[dict-item]
+
+    assert failed["response"]["status"] == "failed"
+    assert failed["response"]["metadata"] == {"request": "preserved"}
+    assert "Ignoring invalid metadata supplied to emit_failed" in caplog.text
+    assert "metadata values must be str" in caplog.text
+
+
+def test_event_stream_builder__emit_failed_ignores_metadata_over_contract_limit(
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    stream = ResponseEventStream(response_id="resp_builder_failed_metadata_limit")
+    stream.emit_created(status="in_progress")
+
+    with caplog.at_level(logging.WARNING):
+        failed = stream.emit_failed(metadata={f"k{index}": "value" for index in range(17)})
+
+    assert failed["response"]["status"] == "failed"
+    assert "metadata" not in failed["response"]
+    assert "Ignoring invalid metadata supplied to emit_failed" in caplog.text
+    assert "at most 16" in caplog.text
+
+
 def test_event_stream_builder__emit_incomplete_accepts_reason_and_usage() -> None:
     stream = ResponseEventStream(response_id="resp_builder_incomplete_params")
     stream.emit_created(status="in_progress")