From fbdb1006497b57c360beff33881c31a2a763b510 Mon Sep 17 00:00:00 2001 From: Prasanna Sankaran Date: Mon, 14 Sep 2026 18:26:14 -0700 Subject: [PATCH 1/3] Exclude failed response inputs from replayable conversation history AgentServer stores a response and its input items when it processes the initial response.created event. When the handler later fails the response, get_history_item_ids() still returned that response's input item IDs for every later request in the same conversation, and for requests chained through previous_response_id. The input that made the turn fail (for example a function_call_output with no matching call) was replayed into each subsequent request, which then failed the same way. Define the invariant on ResponseProviderProtocol and enforce it in the in-memory and file providers: a failed response contributes only the history it inherited; its own input and output items are excluded, before the history limit is applied. The failed response and its input items remain stored and retrievable through the response and input_items endpoints. The file store now records the response status in indexes.json on create and update so history resolution does not open every envelope; stores written before this change fall back to the envelope status. Fixes #48929 --- .../CHANGELOG.md | 8 + .../ai/agentserver/responses/store/_base.py | 9 + .../ai/agentserver/responses/store/_file.py | 54 +++- .../agentserver/responses/store/_history.py | 46 +++ .../ai/agentserver/responses/store/_memory.py | 38 ++- .../contract/test_failed_response_history.py | 249 +++++++++++++++ .../unit/test_failed_response_history.py | 294 ++++++++++++++++++ 7 files changed, 678 insertions(+), 20 deletions(-) create mode 100644 sdk/agentserver/azure-ai-agentserver-responses/azure/ai/agentserver/responses/store/_history.py create mode 100644 sdk/agentserver/azure-ai-agentserver-responses/tests/contract/test_failed_response_history.py create mode 100644 sdk/agentserver/azure-ai-agentserver-responses/tests/unit/test_failed_response_history.py diff --git a/sdk/agentserver/azure-ai-agentserver-responses/CHANGELOG.md b/sdk/agentserver/azure-ai-agentserver-responses/CHANGELOG.md index 1efe65234a02..682d8789faa7 100644 --- a/sdk/agentserver/azure-ai-agentserver-responses/CHANGELOG.md +++ b/sdk/agentserver/azure-ai-agentserver-responses/CHANGELOG.md @@ -4,6 +4,14 @@ ### Bugs Fixed +- A failed response no longer contributes its own input and output items to the + history resolved for later turns in the same conversation or chained through + `previous_response_id`. Previously the input that made a turn fail (for example a + `function_call_output` with no matching call) was replayed into every subsequent + request, which then failed the same way. The failed response and its input items + remain retrievable through `GET /responses/{id}` and `GET /responses/{id}/input_items`. + Applies to the in-memory and file response stores + ([#48929](https://github.com/Azure/azure-sdk-for-python/issues/48929)). - Scoped durable multi-turn task IDs with `FOUNDRY_AGENT_SESSION_GUID` when available, preventing recreated same-name sessions from colliding with task tombstones. Existing pre-rollout active chains remain resumable through a diff --git a/sdk/agentserver/azure-ai-agentserver-responses/azure/ai/agentserver/responses/store/_base.py b/sdk/agentserver/azure-ai-agentserver-responses/azure/ai/agentserver/responses/store/_base.py index 807982f4ce19..24113caf8098 100644 --- a/sdk/agentserver/azure-ai-agentserver-responses/azure/ai/agentserver/responses/store/_base.py +++ b/sdk/agentserver/azure-ai-agentserver-responses/azure/ai/agentserver/responses/store/_base.py @@ -167,6 +167,15 @@ async def get_history_item_ids( ) -> list[str]: """Get history item IDs for a conversation chain scope. + A response whose stored status is ``failed`` contributes neither its + own input items nor its output items to the resolved history: replaying + the input that made a turn fail would make every later turn in the + same conversation (or chained through ``previous_response_id``) fail in + the same way. The failed response's inherited history is still + contributed, and its stored items stay retrievable through + :meth:`get_input_items` for diagnostics. The exclusion is applied + before ``limit`` truncation. + :param previous_response_id: Optional response ID to chain history from. :type previous_response_id: str | None :param conversation_id: Optional conversation ID to scope history lookup. diff --git a/sdk/agentserver/azure-ai-agentserver-responses/azure/ai/agentserver/responses/store/_file.py b/sdk/agentserver/azure-ai-agentserver-responses/azure/ai/agentserver/responses/store/_file.py index db9fc24f9134..0e948bbb4f9d 100644 --- a/sdk/agentserver/azure-ai-agentserver-responses/azure/ai/agentserver/responses/store/_file.py +++ b/sdk/agentserver/azure-ai-agentserver-responses/azure/ai/agentserver/responses/store/_file.py @@ -82,6 +82,7 @@ from ..models._helpers import get_conversation_id from ._base import ResponseAlreadyExistsError, ResponseProviderProtocol, ResponseStoreCorruptionError +from ._history import is_replayable_status, normalize_status from ..models import _generated as _generated_models @@ -334,6 +335,7 @@ async def create_response( "input_item_ids": input_ids, "output_item_ids": output_ids, "history_item_ids": history_ids, + "status": normalize_status(response.get("status")), }, ) # (Spec 028) Best-effort removal of a legacy per-response @@ -401,7 +403,7 @@ async def update_response( # crash-ordering invariant as create_response). output_ids = self._store_output_items_unlocked(response) _atomic_write_json(target, self._pointerize_output(response_dict)) - self._update_indexes_unlocked(response_id, output_item_ids=output_ids) + self._update_indexes_unlocked(response_id, output_item_ids=output_ids, status=response.get("status")) async def delete_response(self, response_id: str, *, context: PlatformContext | None = None) -> None: """Soft-delete a stored response envelope by identifier. @@ -544,6 +546,9 @@ async def get_history_item_ids( - When ``conversation_id`` is set, iterates all non-deleted responses in that conversation and contributes their ``history_item_ids + input_item_ids + output_item_ids``. + - A ``failed`` response contributes only its ``history_item_ids``; + its own input and output items are excluded so the input that + made it fail is not replayed into later turns. - Both may be set; results are concatenated in the same order. - When over ``limit``, keeps the most recent N item IDs from the resolved chain, preserving chronological order in the returned slice. @@ -567,23 +572,14 @@ async def get_history_item_ids( resolved: list[str] = [] if previous_response_id is not None and not self._deleted_marker(previous_response_id).exists(): - indexes = _read_json_or_none(self._indexes_path(previous_response_id)) - if indexes is not None: - resolved.extend(indexes.get("history_item_ids") or []) - resolved.extend(indexes.get("input_item_ids") or []) - resolved.extend(indexes.get("output_item_ids") or []) + resolved.extend(self._replayable_item_ids_unlocked(previous_response_id)) if conversation_id is not None: conv_data = _read_json_or_none(self._conversation_path(conversation_id)) for rid in (conv_data or {}).get("response_ids", []): if self._deleted_marker(rid).exists(): continue - indexes = _read_json_or_none(self._indexes_path(rid)) - if indexes is None: - continue - resolved.extend(indexes.get("history_item_ids") or []) - resolved.extend(indexes.get("input_item_ids") or []) - resolved.extend(indexes.get("output_item_ids") or []) + resolved.extend(self._replayable_item_ids_unlocked(rid)) if limit <= 0: return [] @@ -595,6 +591,35 @@ async def get_history_item_ids( # Internal helpers (must be called with self._lock held) # ------------------------------------------------------------------ + def _replayable_item_ids_unlocked(self, response_id: str) -> list[str]: + """Return the item IDs one response contributes to replayable history. + + A ``failed`` response contributes only the history it inherited; its + own input and output items are excluded so the input that made it + fail is not replayed into later turns. The status is read from the + indexes file; stores written before the status was indexed fall back + to the persisted response envelope. + + :param response_id: The response identifier. + :type response_id: str + :returns: Ordered history + input + output item IDs, or history only + for a failed response. Empty when the response has no indexes. + :rtype: list[str] + """ + indexes = _read_json_or_none(self._indexes_path(response_id)) + if indexes is None: + return [] + resolved = list(indexes.get("history_item_ids") or []) + if "status" in indexes: + status = indexes.get("status") + else: + envelope = _read_json_or_none(self._response_path(response_id)) + status = envelope.get("status") if envelope is not None else None + if is_replayable_status(status): + resolved.extend(indexes.get("input_item_ids") or []) + resolved.extend(indexes.get("output_item_ids") or []) + return resolved + def _store_items_unlocked(self, items: Iterable[Any]) -> list[str]: """Persist items to the single global ``items/`` store. @@ -700,6 +725,7 @@ def _update_indexes_unlocked( input_item_ids: list[str] | None = None, output_item_ids: list[str] | None = None, history_item_ids: list[str] | None = None, + status: Any = None, ) -> None: """Merge the supplied id lists into the persisted indexes file. @@ -708,6 +734,8 @@ def _update_indexes_unlocked( :keyword input_item_ids: New input ids to overwrite. :keyword output_item_ids: New output ids to overwrite. :keyword history_item_ids: New history ids to overwrite. + :keyword status: New response status to record, used to decide + whether the response's items are replayable history. :rtype: None """ path = self._indexes_path(response_id) @@ -718,6 +746,8 @@ def _update_indexes_unlocked( current["output_item_ids"] = output_item_ids if history_item_ids is not None: current["history_item_ids"] = history_item_ids + if status is not None: + current["status"] = normalize_status(status) _atomic_write_json(path, current) def _add_response_to_conversation_unlocked(self, conversation_id: str, response_id: str) -> None: diff --git a/sdk/agentserver/azure-ai-agentserver-responses/azure/ai/agentserver/responses/store/_history.py b/sdk/agentserver/azure-ai-agentserver-responses/azure/ai/agentserver/responses/store/_history.py new file mode 100644 index 000000000000..0e376cf84e84 --- /dev/null +++ b/sdk/agentserver/azure-ai-agentserver-responses/azure/ai/agentserver/responses/store/_history.py @@ -0,0 +1,46 @@ +# Copyright (c) Microsoft Corporation. +# Licensed under the MIT license. +"""Shared history-resolution rules for response providers.""" + +from __future__ import annotations + +from typing import Any + +_NON_REPLAYABLE_STATUSES: frozenset[str] = frozenset({"failed"}) + + +def normalize_status(status: Any) -> str | None: + """Return *status* as the plain string the wire uses, or ``None`` when unset. + + Accepts the raw string stored on a response envelope as well as an enum + member whose ``value`` is that string, so providers persist and compare + the same representation regardless of how the status was produced. + + :param status: The stored response status, if any. + :type status: Any + :returns: The status string, or ``None``. + :rtype: str | None + """ + if status is None: + return None + return str(getattr(status, "value", status)) + + +def is_replayable_status(status: Any) -> bool: + """Return whether a response with *status* contributes its own items to history. + + Only a ``failed`` response is excluded: its input is what made the turn + fail, so replaying it would fail every later turn in the same conversation + or chain. Responses in any other state (including ``incomplete`` and + ``cancelled``) keep their items in history, as does a response whose + status is unknown. + + :param status: The stored response status, if any. + :type status: Any + :returns: ``True`` when the response's input and output items are replayable. + :rtype: bool + """ + normalized = normalize_status(status) + if normalized is None: + return True + return normalized not in _NON_REPLAYABLE_STATUSES diff --git a/sdk/agentserver/azure-ai-agentserver-responses/azure/ai/agentserver/responses/store/_memory.py b/sdk/agentserver/azure-ai-agentserver-responses/azure/ai/agentserver/responses/store/_memory.py index 37c6c37dbf75..059b735fbcab 100644 --- a/sdk/agentserver/azure-ai-agentserver-responses/azure/ai/agentserver/responses/store/_memory.py +++ b/sdk/agentserver/azure-ai-agentserver-responses/azure/ai/agentserver/responses/store/_memory.py @@ -16,6 +16,7 @@ from ..models._helpers import get_conversation_id from ..models.runtime import ResponseExecution, ResponseModeFlags, ResponseStatus, StreamEventRecord, _StreamReplayState from ._base import ResponseAlreadyExistsError, ResponseProviderProtocol +from ._history import is_replayable_status from ..models import _generated as _generated_models @@ -296,8 +297,11 @@ async def get_history_item_ids( Collects history, input, and output item IDs from the previous response chain and/or all responses within the given conversation. - When over *limit*, keeps the most recent N item IDs from the - resolved chain, preserving chronological order in the returned slice. + A ``failed`` response contributes only its inherited history; its own + input and output items are excluded so that the input that made it + fail is not replayed into later turns. When over *limit*, keeps the + most recent N item IDs from the resolved chain, preserving + chronological order in the returned slice. :param previous_response_id: Optional response ID to chain history from. :type previous_response_id: str | None @@ -318,18 +322,14 @@ async def get_history_item_ids( if entry is not None and not entry.deleted: # Resolve history chain for the previous response: # return historyItemIds + inputItemIds + outputItemIds of the previous response - resolved.extend(entry.history_item_ids or []) - resolved.extend(entry.input_item_ids or []) - resolved.extend(entry.output_item_ids or []) + resolved.extend(self._replayable_item_ids_unlocked(entry)) if conversation_id is not None: for response_id in self._conversation_responses.get(conversation_id, []): entry = self._entries.get(response_id) if entry is None or entry.deleted: continue - resolved.extend(entry.history_item_ids or []) - resolved.extend(entry.input_item_ids or []) - resolved.extend(entry.output_item_ids or []) + resolved.extend(self._replayable_item_ids_unlocked(entry)) if limit <= 0: return [] @@ -595,6 +595,28 @@ def _purge_expired_unlocked(self, *, now: datetime | None = None) -> int: return len(expired_ids) + @staticmethod + def _replayable_item_ids_unlocked(entry: _StoreEntry) -> list[str]: + """Return the item IDs one response contributes to replayable history. + + Must be called while holding ``self._lock``. + + A ``failed`` response contributes only the history it inherited; its + own input and output items are excluded so the input that made it + fail is not replayed into later turns. + + :param entry: The store entry to read. + :type entry: _StoreEntry + :returns: Ordered history + input + output item IDs, or history only for a failed response. + :rtype: list[str] + """ + resolved = list(entry.history_item_ids or []) + status = entry.response.get("status") if entry.response is not None else None + if is_replayable_status(status): + resolved.extend(entry.input_item_ids or []) + resolved.extend(entry.output_item_ids or []) + return resolved + def _store_output_items_unlocked(self, response: _generated_models.ResponseObject) -> list[str]: """Extract output items from a response, store them in the item store, and return their IDs. diff --git a/sdk/agentserver/azure-ai-agentserver-responses/tests/contract/test_failed_response_history.py b/sdk/agentserver/azure-ai-agentserver-responses/tests/contract/test_failed_response_history.py new file mode 100644 index 000000000000..4ddf3c583d58 --- /dev/null +++ b/sdk/agentserver/azure-ai-agentserver-responses/tests/contract/test_failed_response_history.py @@ -0,0 +1,249 @@ +# Copyright (c) Microsoft Corporation. +# Licensed under the MIT license. +"""A failed turn must not poison the conversation for later turns. + +End-to-end through ``ResponsesAgentServerHost``: a handler fails when its +input carries a ``function_call_output`` that matches no call. The input of +that failed turn must not be replayed into the history of later turns in +the same conversation, or of turns chained through ``previous_response_id``, +in synchronous, streaming and background modes. The failed response and its +input items remain retrievable for diagnostics. + +Regression coverage for https://github.com/Azure/azure-sdk-for-python/issues/48929. +""" + +from __future__ import annotations + +import asyncio +import json as _json +import time +from typing import Any +from uuid import uuid4 + +from starlette.testclient import TestClient + +from azure.ai.agentserver.responses import ResponsesAgentServerHost +from azure.ai.agentserver.responses.models._helpers import get_input_expanded +from azure.ai.agentserver.responses.streaming._event_stream import ResponseEventStream + +_BAD_CALL_ID = "call_that_does_not_exist" + + +# ════════════════════════════════════════════════════════════ +# Helpers +# ════════════════════════════════════════════════════════════ + + +class _RecordingHandler: + """Records the history each turn saw; fails a turn whose input carries an unmatched ``function_call_output``.""" + + def __init__(self) -> None: + self.histories: list[list[dict[str, Any]]] = [] + + def as_handler(self): + recorder = self + + async def handler(request: Any, context: Any, cancellation_signal: asyncio.Event): + history = [dict(item) for item in await context.get_history()] + recorder.histories.append(history) + input_items = [dict(item) for item in get_input_expanded(request)] + unmatched = any( + item.get("type") == "function_call_output" and item.get("call_id") == _BAD_CALL_ID + for item in input_items + ) + + async def _events(): + stream = ResponseEventStream(response_id=context.response_id, model=getattr(request, "model", None)) + yield stream.emit_created() + if unmatched: + yield stream.emit_failed( + code="invalid_request", + message=f"No tool call found for function call output with call_id {_BAD_CALL_ID}.", + ) + else: + yield stream.emit_completed() + + return _events() + + return handler + + +def _build_client() -> tuple[TestClient, _RecordingHandler]: + recorder = _RecordingHandler() + app = ResponsesAgentServerHost() + app.response_handler(recorder.as_handler()) + return TestClient(app), recorder + + +def _poison_input() -> list[dict[str, Any]]: + return [ + {"role": "user", "content": "Hello"}, + {"type": "function_call_output", "call_id": _BAD_CALL_ID, "output": "invalid output"}, + ] + + +def _collect_sse_events(response: Any) -> list[dict[str, Any]]: + events: list[dict[str, Any]] = [] + current_type: str | None = None + current_data: str | None = None + for line in response.iter_lines(): + if not line: + if current_type is not None: + events.append({"type": current_type, "data": _json.loads(current_data) if current_data else {}}) + current_type = None + current_data = None + continue + if line.startswith("event:"): + current_type = line.split(":", 1)[1].strip() + elif line.startswith("data:"): + current_data = line.split(":", 1)[1].strip() + if current_type is not None: + events.append({"type": current_type, "data": _json.loads(current_data) if current_data else {}}) + return events + + +def _post(client: TestClient, **body: Any) -> dict[str, Any]: + resp = client.post("/responses", json={"model": "test", **body}) + assert resp.status_code == 200, resp.text + return resp.json() + + +def _post_stream(client: TestClient, **body: Any) -> dict[str, Any]: + with client.stream("POST", "/responses", json={"model": "test", "stream": True, **body}) as resp: + assert resp.status_code == 200 + events = _collect_sse_events(resp) + terminal = [e for e in events if e["type"] in ("response.completed", "response.failed")] + assert terminal, [e["type"] for e in events] + return terminal[-1]["data"]["response"] + + +def _wait_terminal(client: TestClient, response_id: str, timeout: float = 5.0) -> dict[str, Any]: + deadline = time.monotonic() + timeout + while True: + body = client.get(f"/responses/{response_id}").json() + if body.get("status") in ("completed", "failed", "cancelled", "incomplete"): + return body + assert time.monotonic() < deadline, f"response {response_id} did not reach a terminal state: {body}" + time.sleep(0.05) + + +def _conv() -> str: + """Unique conversation id: the host's default file store is shared within a process.""" + return f"conv_{uuid4().hex}" + + +def _history_types(history: list[dict[str, Any]]) -> list[str]: + return [str(item.get("type")) for item in history] + + +# ════════════════════════════════════════════════════════════ +# conversation scope +# ════════════════════════════════════════════════════════════ + + +def test_failed_turn_input_is_not_replayed_into_conversation() -> None: + """Issue #48929 repro: a failed turn must not make the next valid turn in the conversation fail.""" + client, handler = _build_client() + conv = _conv() + + failed = _post(client, conversation=conv, input=_poison_input()) + assert failed["status"] == "failed" + assert failed["error"]["code"] == "invalid_request" + + ok = _post(client, conversation=conv, input="Hello, how are you?") + assert ok["status"] == "completed" + + # The second turn saw no history from the failed turn. + assert handler.histories[1] == [] + + # A third turn sees only the successful second turn. + third = _post(client, conversation=conv, input="And now?") + assert third["status"] == "completed" + assert _history_types(handler.histories[2]) == ["message"] + assert all(item.get("call_id") != _BAD_CALL_ID for item in handler.histories[2]) + + +def test_successful_turns_before_and_after_a_failure_are_kept() -> None: + client, handler = _build_client() + conv = _conv() + + first = _post(client, conversation=conv, input="first") + assert first["status"] == "completed" + failed = _post(client, conversation=conv, input=_poison_input()) + assert failed["status"] == "failed" + third = _post(client, conversation=conv, input="third") + assert third["status"] == "completed" + + # The third turn's history is exactly the first turn's input: no poison, nothing dropped. + history = handler.histories[2] + assert _history_types(history) == ["message"] + assert history[0]["role"] == "user" + assert all(item.get("call_id") != _BAD_CALL_ID for item in history) + + +def test_failed_response_and_its_input_items_remain_retrievable() -> None: + """Exclusion from history is not deletion: diagnostics still work.""" + client, _handler = _build_client() + conv = _conv() + + failed = _post(client, conversation=conv, input=_poison_input()) + response_id = failed["id"] + + stored = client.get(f"/responses/{response_id}") + assert stored.status_code == 200 + assert stored.json()["status"] == "failed" + + items = client.get(f"/responses/{response_id}/input_items").json() + types = [item.get("type") for item in items.get("data", [])] + assert "function_call_output" in types + + +# ════════════════════════════════════════════════════════════ +# previous_response_id scope +# ════════════════════════════════════════════════════════════ + + +def test_chaining_from_failed_response_does_not_replay_its_input() -> None: + client, handler = _build_client() + + ok = _post(client, input="first") + failed = _post(client, previous_response_id=ok["id"], input=_poison_input()) + assert failed["status"] == "failed" + + chained = _post(client, previous_response_id=failed["id"], input="Hello again") + assert chained["status"] == "completed" + + # Only the successful first turn is inherited through the failed response. + history = handler.histories[2] + assert _history_types(history) == ["message"] + assert all(item.get("call_id") != _BAD_CALL_ID for item in history) + + +# ════════════════════════════════════════════════════════════ +# streaming and background modes +# ════════════════════════════════════════════════════════════ + + +def test_streaming_failed_turn_input_is_not_replayed() -> None: + client, handler = _build_client() + conv = _conv() + + failed = _post_stream(client, conversation=conv, input=_poison_input()) + assert failed["status"] == "failed" + + ok = _post_stream(client, conversation=conv, input="Hello, how are you?") + assert ok["status"] == "completed" + assert handler.histories[1] == [] + + +def test_background_failed_turn_input_is_not_replayed() -> None: + client, handler = _build_client() + conv = _conv() + + started = _post(client, conversation=conv, background=True, input=_poison_input()) + failed = _wait_terminal(client, started["id"]) + assert failed["status"] == "failed" + + ok = _post(client, conversation=conv, input="Hello, how are you?") + assert ok["status"] == "completed" + assert handler.histories[1] == [] diff --git a/sdk/agentserver/azure-ai-agentserver-responses/tests/unit/test_failed_response_history.py b/sdk/agentserver/azure-ai-agentserver-responses/tests/unit/test_failed_response_history.py new file mode 100644 index 000000000000..7cae4e441d07 --- /dev/null +++ b/sdk/agentserver/azure-ai-agentserver-responses/tests/unit/test_failed_response_history.py @@ -0,0 +1,294 @@ +# Copyright (c) Microsoft Corporation. +# Licensed under the MIT license. +"""A failed response's own items must not be replayed as conversation history. + +When a turn fails (for example because its input carried a +``function_call_output`` with no matching call), the input that made it fail +must not be resolved into the history of later turns in the same conversation +or of turns chained through ``previous_response_id``. Otherwise every later +request fails the same way. The failed response's items stay retrievable +through ``get_input_items`` for diagnostics. + +The scenarios run against both ``InMemoryResponseProvider`` and +``FileResponseStore`` and assert identical results. +""" + +from __future__ import annotations + +import json +from enum import Enum +from pathlib import Path +from typing import Any, Callable + +import pytest + +from azure.ai.agentserver.responses.models import _generated as generated_models +from azure.ai.agentserver.responses.store._file import FileResponseStore +from azure.ai.agentserver.responses.store._history import is_replayable_status +from azure.ai.agentserver.responses.store._memory import InMemoryResponseProvider + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + + +def _response( + response_id: str, + *, + status: str = "completed", + output: list[dict[str, Any]] | None = None, + conversation_id: str | None = None, +) -> generated_models.ResponseObject: + payload: dict[str, Any] = { + "id": response_id, + "object": "response", + "output": output or [], + "store": True, + "status": status, + } + if conversation_id is not None: + payload["conversation"] = {"id": conversation_id} + return generated_models.ResponseObject(payload) + + +def _input_item(item_id: str, text: str = "hello") -> dict[str, Any]: + return { + "id": item_id, + "type": "message", + "role": "user", + "content": [{"type": "input_text", "text": text}], + } + + +def _output_item(item_id: str, text: str = "world") -> dict[str, Any]: + return { + "id": item_id, + "type": "message", + "role": "assistant", + "content": [{"type": "output_text", "text": text}], + } + + +def _make_provider_factories(tmp_path: Path) -> list[tuple[str, Callable[[], Any]]]: + return [ + ("memory", lambda: InMemoryResponseProvider()), + ("file", lambda: FileResponseStore(storage_dir=tmp_path / "store")), + ] + + +# --------------------------------------------------------------------------- +# Status rule +# --------------------------------------------------------------------------- + + +@pytest.mark.parametrize( + "status,expected", + [ + ("completed", True), + ("incomplete", True), + ("cancelled", True), + ("in_progress", True), + ("queued", True), + (None, True), + ("failed", False), + ], +) +def test_is_replayable_status(status: str | None, expected: bool) -> None: + assert is_replayable_status(status) is expected + + +class _Status(str, Enum): + COMPLETED = "completed" + FAILED = "failed" + + +def test_is_replayable_status_accepts_enum_members() -> None: + """A provider may hand in an enum member; ``str(member)`` is not its value on Python 3.11+.""" + assert is_replayable_status(_Status.COMPLETED) is True + assert is_replayable_status(_Status.FAILED) is False + + +@pytest.mark.asyncio +async def test_file_store_indexes_enum_status_as_its_value(tmp_path: Path) -> None: + store = FileResponseStore(storage_dir=tmp_path / "store") + await store.create_response( + _response("r_failed", status=_Status.FAILED, conversation_id="conv-1"), [_input_item("bad_in")], None + ) + indexes = json.loads( + store._indexes_path("r_failed").read_text(encoding="utf-8") + ) # pylint: disable=protected-access + assert indexes["status"] == "failed" + assert await store.get_history_item_ids(None, "conv-1", limit=100) == [] + + +# --------------------------------------------------------------------------- +# Conversation scope +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_failed_turn_input_and_output_excluded_from_conversation_history(tmp_path: Path) -> None: + """The failed turn's own items are skipped; earlier and later successful turns are kept in order.""" + for label, factory in _make_provider_factories(tmp_path): + provider = factory() + await provider.create_response( + _response("r_ok_1", output=[_output_item("ok1_out")], conversation_id="conv-1"), + [_input_item("ok1_in")], + None, + ) + # A turn whose input carried an unmatched function_call_output and failed. + await provider.create_response( + _response("r_failed", status="failed", output=[_output_item("bad_out")], conversation_id="conv-1"), + [_input_item("bad_in"), _input_item("bad_call_output")], + None, + ) + await provider.create_response( + _response("r_ok_2", output=[_output_item("ok2_out")], conversation_id="conv-1"), + [_input_item("ok2_in")], + None, + ) + + ids = await provider.get_history_item_ids(None, "conv-1", limit=100) + assert ids == ["ok1_in", "ok1_out", "ok2_in", "ok2_out"], label + + +@pytest.mark.asyncio +async def test_response_that_fails_after_creation_is_excluded_on_update(tmp_path: Path) -> None: + """The orchestrator persists the response before the terminal; a later ``failed`` update must take effect.""" + for label, factory in _make_provider_factories(tmp_path): + provider = factory() + await provider.create_response( + _response("r_1", status="in_progress", conversation_id="conv-1"), + [_input_item("in_1")], + None, + ) + # While in progress the input is part of the history. + assert await provider.get_history_item_ids(None, "conv-1", limit=100) == ["in_1"], label + + await provider.update_response(_response("r_1", status="failed", conversation_id="conv-1")) + assert await provider.get_history_item_ids(None, "conv-1", limit=100) == [], label + + # The stored response itself still reports the failure. + stored = await provider.get_response("r_1") + assert stored["status"] == "failed", label + + +@pytest.mark.asyncio +async def test_failed_response_input_items_remain_retrievable(tmp_path: Path) -> None: + """Exclusion from replayable history does not delete the stored items (diagnostics).""" + for label, factory in _make_provider_factories(tmp_path): + provider = factory() + await provider.create_response( + _response("r_failed", status="failed", conversation_id="conv-1"), + [_input_item("bad_in"), _input_item("bad_call_output")], + history_item_ids=["hist_1"], + ) + + items = await provider.get_input_items("r_failed", limit=100, ascending=True) + assert [item["id"] for item in items] == ["bad_in", "bad_call_output"], label + assert await provider.get_history_item_ids(None, "conv-1", limit=100) == ["hist_1"], label + + +# --------------------------------------------------------------------------- +# previous_response_id scope +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_chaining_from_failed_response_keeps_only_inherited_history(tmp_path: Path) -> None: + """``previous_response_id`` pointing at a failed response yields its inherited history only.""" + for label, factory in _make_provider_factories(tmp_path): + provider = factory() + await provider.create_response( + _response("r_prev", output=[_output_item("prev_out")]), + [_input_item("prev_in")], + None, + ) + chain = await provider.get_history_item_ids("r_prev", None, limit=100) + assert chain == ["prev_in", "prev_out"], label + + await provider.create_response( + _response("r_failed", status="failed", output=[_output_item("bad_out")]), + [_input_item("bad_in")], + history_item_ids=chain, + ) + + ids = await provider.get_history_item_ids("r_failed", None, limit=100) + assert ids == ["prev_in", "prev_out"], label + + +@pytest.mark.asyncio +async def test_successful_chain_through_failed_response_stays_clean(tmp_path: Path) -> None: + """A successful response chained after a failed one carries forward only clean history.""" + for label, factory in _make_provider_factories(tmp_path): + provider = factory() + await provider.create_response( + _response("r_failed", status="failed"), + [_input_item("bad_in")], + history_item_ids=["hist_1"], + ) + inherited = await provider.get_history_item_ids("r_failed", None, limit=100) + assert inherited == ["hist_1"], label + + await provider.create_response( + _response("r_next", output=[_output_item("next_out")]), + [_input_item("next_in")], + history_item_ids=inherited, + ) + ids = await provider.get_history_item_ids("r_next", None, limit=100) + assert ids == ["hist_1", "next_in", "next_out"], label + + +# --------------------------------------------------------------------------- +# Interaction with the history limit +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_failed_items_do_not_consume_history_limit(tmp_path: Path) -> None: + """Exclusion happens before truncation, so failed items never displace replayable ones.""" + for label, factory in _make_provider_factories(tmp_path): + provider = factory() + await provider.create_response( + _response("r_ok", output=[_output_item("ok_out")], conversation_id="conv-1"), + [_input_item("ok_in")], + None, + ) + await provider.create_response( + _response("r_failed", status="failed", conversation_id="conv-1"), + [_input_item("bad_1"), _input_item("bad_2"), _input_item("bad_3")], + None, + ) + + ids = await provider.get_history_item_ids(None, "conv-1", limit=2) + assert ids == ["ok_in", "ok_out"], label + + +# --------------------------------------------------------------------------- +# File store: stores written before the status was indexed +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_file_store_reads_status_from_envelope_when_indexes_predate_it(tmp_path: Path) -> None: + """Indexes written by older versions have no ``status``; the envelope decides instead.""" + store = FileResponseStore(storage_dir=tmp_path / "store") + await store.create_response( + _response("r_failed", status="failed", conversation_id="conv-1"), + [_input_item("bad_in")], + None, + ) + await store.create_response( + _response("r_ok", conversation_id="conv-1"), + [_input_item("ok_in")], + None, + ) + + # Simulate a pre-existing store by dropping the indexed status. + for response_id in ("r_failed", "r_ok"): + indexes_path = store._indexes_path(response_id) # pylint: disable=protected-access + indexes = json.loads(indexes_path.read_text(encoding="utf-8")) + assert indexes.pop("status") is not None + indexes_path.write_text(json.dumps(indexes), encoding="utf-8") + + assert await store.get_history_item_ids(None, "conv-1", limit=100) == ["ok_in"] From 5e80dedc6a1db5f8480734f666864a8561e1219e Mon Sep 17 00:00:00 2001 From: Prasanna Sankaran Date: Mon, 14 Sep 2026 19:05:49 -0700 Subject: [PATCH 2/3] Read the response status from the envelope in FileResponseStore Keeping a copy of the status in indexes.json created a second source of truth that could disagree with the envelope if the process stopped between the two atomic writes, and an update that omitted the status left the old indexed value in place. Read the status from the persisted envelope instead, which is written atomically, and drop the indexed copy. --- .../ai/agentserver/responses/store/_file.py | 22 +++----- .../unit/test_failed_response_history.py | 52 ++++++++----------- 2 files changed, 29 insertions(+), 45 deletions(-) diff --git a/sdk/agentserver/azure-ai-agentserver-responses/azure/ai/agentserver/responses/store/_file.py b/sdk/agentserver/azure-ai-agentserver-responses/azure/ai/agentserver/responses/store/_file.py index 0e948bbb4f9d..a42dd61eaf51 100644 --- a/sdk/agentserver/azure-ai-agentserver-responses/azure/ai/agentserver/responses/store/_file.py +++ b/sdk/agentserver/azure-ai-agentserver-responses/azure/ai/agentserver/responses/store/_file.py @@ -82,7 +82,7 @@ from ..models._helpers import get_conversation_id from ._base import ResponseAlreadyExistsError, ResponseProviderProtocol, ResponseStoreCorruptionError -from ._history import is_replayable_status, normalize_status +from ._history import is_replayable_status from ..models import _generated as _generated_models @@ -335,7 +335,6 @@ async def create_response( "input_item_ids": input_ids, "output_item_ids": output_ids, "history_item_ids": history_ids, - "status": normalize_status(response.get("status")), }, ) # (Spec 028) Best-effort removal of a legacy per-response @@ -403,7 +402,7 @@ async def update_response( # crash-ordering invariant as create_response). output_ids = self._store_output_items_unlocked(response) _atomic_write_json(target, self._pointerize_output(response_dict)) - self._update_indexes_unlocked(response_id, output_item_ids=output_ids, status=response.get("status")) + self._update_indexes_unlocked(response_id, output_item_ids=output_ids) async def delete_response(self, response_id: str, *, context: PlatformContext | None = None) -> None: """Soft-delete a stored response envelope by identifier. @@ -597,8 +596,9 @@ def _replayable_item_ids_unlocked(self, response_id: str) -> list[str]: A ``failed`` response contributes only the history it inherited; its own input and output items are excluded so the input that made it fail is not replayed into later turns. The status is read from the - indexes file; stores written before the status was indexed fall back - to the persisted response envelope. + persisted response envelope, which is the single source of truth for + it: the envelope is written atomically, so a crash can never leave + the status and the item indexes disagreeing. :param response_id: The response identifier. :type response_id: str @@ -610,11 +610,8 @@ def _replayable_item_ids_unlocked(self, response_id: str) -> list[str]: if indexes is None: return [] resolved = list(indexes.get("history_item_ids") or []) - if "status" in indexes: - status = indexes.get("status") - else: - envelope = _read_json_or_none(self._response_path(response_id)) - status = envelope.get("status") if envelope is not None else None + envelope = _read_json_or_none(self._response_path(response_id)) + status = envelope.get("status") if envelope is not None else None if is_replayable_status(status): resolved.extend(indexes.get("input_item_ids") or []) resolved.extend(indexes.get("output_item_ids") or []) @@ -725,7 +722,6 @@ def _update_indexes_unlocked( input_item_ids: list[str] | None = None, output_item_ids: list[str] | None = None, history_item_ids: list[str] | None = None, - status: Any = None, ) -> None: """Merge the supplied id lists into the persisted indexes file. @@ -734,8 +730,6 @@ def _update_indexes_unlocked( :keyword input_item_ids: New input ids to overwrite. :keyword output_item_ids: New output ids to overwrite. :keyword history_item_ids: New history ids to overwrite. - :keyword status: New response status to record, used to decide - whether the response's items are replayable history. :rtype: None """ path = self._indexes_path(response_id) @@ -746,8 +740,6 @@ def _update_indexes_unlocked( current["output_item_ids"] = output_item_ids if history_item_ids is not None: current["history_item_ids"] = history_item_ids - if status is not None: - current["status"] = normalize_status(status) _atomic_write_json(path, current) def _add_response_to_conversation_unlocked(self, conversation_id: str, response_id: str) -> None: diff --git a/sdk/agentserver/azure-ai-agentserver-responses/tests/unit/test_failed_response_history.py b/sdk/agentserver/azure-ai-agentserver-responses/tests/unit/test_failed_response_history.py index 7cae4e441d07..6a32e41ee152 100644 --- a/sdk/agentserver/azure-ai-agentserver-responses/tests/unit/test_failed_response_history.py +++ b/sdk/agentserver/azure-ai-agentserver-responses/tests/unit/test_failed_response_history.py @@ -108,19 +108,6 @@ def test_is_replayable_status_accepts_enum_members() -> None: assert is_replayable_status(_Status.FAILED) is False -@pytest.mark.asyncio -async def test_file_store_indexes_enum_status_as_its_value(tmp_path: Path) -> None: - store = FileResponseStore(storage_dir=tmp_path / "store") - await store.create_response( - _response("r_failed", status=_Status.FAILED, conversation_id="conv-1"), [_input_item("bad_in")], None - ) - indexes = json.loads( - store._indexes_path("r_failed").read_text(encoding="utf-8") - ) # pylint: disable=protected-access - assert indexes["status"] == "failed" - assert await store.get_history_item_ids(None, "conv-1", limit=100) == [] - - # --------------------------------------------------------------------------- # Conversation scope # --------------------------------------------------------------------------- @@ -265,30 +252,35 @@ async def test_failed_items_do_not_consume_history_limit(tmp_path: Path) -> None # --------------------------------------------------------------------------- -# File store: stores written before the status was indexed +# File store: the envelope is the single source of truth for the status # --------------------------------------------------------------------------- @pytest.mark.asyncio -async def test_file_store_reads_status_from_envelope_when_indexes_predate_it(tmp_path: Path) -> None: - """Indexes written by older versions have no ``status``; the envelope decides instead.""" +async def test_file_store_reads_status_from_envelope_not_indexes(tmp_path: Path) -> None: + """A stale or tampered indexes file cannot make a failed turn replayable again.""" store = FileResponseStore(storage_dir=tmp_path / "store") await store.create_response( - _response("r_failed", status="failed", conversation_id="conv-1"), - [_input_item("bad_in")], - None, - ) - await store.create_response( - _response("r_ok", conversation_id="conv-1"), - [_input_item("ok_in")], + _response("r_1", status="in_progress", conversation_id="conv-1"), + [_input_item("in_1")], None, ) + await store.update_response(_response("r_1", status="failed", conversation_id="conv-1")) + + # Simulate the worst case: the indexes file says nothing about the status (or lies about it). + indexes_path = store._indexes_path("r_1") # pylint: disable=protected-access + indexes = json.loads(indexes_path.read_text(encoding="utf-8")) + assert "status" not in indexes + indexes["status"] = "in_progress" + indexes_path.write_text(json.dumps(indexes), encoding="utf-8") - # Simulate a pre-existing store by dropping the indexed status. - for response_id in ("r_failed", "r_ok"): - indexes_path = store._indexes_path(response_id) # pylint: disable=protected-access - indexes = json.loads(indexes_path.read_text(encoding="utf-8")) - assert indexes.pop("status") is not None - indexes_path.write_text(json.dumps(indexes), encoding="utf-8") + assert await store.get_history_item_ids(None, "conv-1", limit=100) == [] - assert await store.get_history_item_ids(None, "conv-1", limit=100) == ["ok_in"] + +@pytest.mark.asyncio +async def test_file_store_enum_status_on_envelope(tmp_path: Path) -> None: + store = FileResponseStore(storage_dir=tmp_path / "store") + await store.create_response( + _response("r_failed", status=_Status.FAILED, conversation_id="conv-1"), [_input_item("bad_in")], None + ) + assert await store.get_history_item_ids(None, "conv-1", limit=100) == [] From 43dc68bc76753ba2047c0138b52cecc87f33a8be Mon Sep 17 00:00:00 2001 From: Prasanna Sankaran Date: Mon, 14 Sep 2026 19:17:06 -0700 Subject: [PATCH 3/3] Document that the hosted history endpoint owns the failed-response exclusion --- .../ai/agentserver/responses/store/_foundry_provider.py | 6 ++++++ 1 file changed, 6 insertions(+) diff --git a/sdk/agentserver/azure-ai-agentserver-responses/azure/ai/agentserver/responses/store/_foundry_provider.py b/sdk/agentserver/azure-ai-agentserver-responses/azure/ai/agentserver/responses/store/_foundry_provider.py index 2736942d0f23..19a264364cd8 100644 --- a/sdk/agentserver/azure-ai-agentserver-responses/azure/ai/agentserver/responses/store/_foundry_provider.py +++ b/sdk/agentserver/azure-ai-agentserver-responses/azure/ai/agentserver/responses/store/_foundry_provider.py @@ -418,6 +418,12 @@ async def get_history_item_ids( ) -> list[str]: """Retrieve the ordered list of item IDs that form the conversation history. + Resolution is delegated to the hosted ``history/item_ids`` endpoint, which is + responsible for applying the replayable-history rule documented on + :meth:`ResponseProviderProtocol.get_history_item_ids` (a ``failed`` response + contributes neither its input nor its output items). The returned IDs are + passed through unchanged; the client cannot tell which response an ID belongs to. + :param previous_response_id: The response whose prior turn should be the history anchor. :type previous_response_id: str | None :param conversation_id: An explicit conversation scope identifier, if available.