[agentserver-responses] Exclude failed response inputs from replayable conversation history

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

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.

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
 from ..models import _generated as _generated_models
 
 
@@ -544,6 +545,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 +571,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 +590,33 @@ 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
+        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
+        :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 [])
+        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.
 

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.

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

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.