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..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,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. 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. 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..6a32e41ee152 --- /dev/null +++ b/sdk/agentserver/azure-ai-agentserver-responses/tests/unit/test_failed_response_history.py @@ -0,0 +1,286 @@ +# 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 + + +# --------------------------------------------------------------------------- +# 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: the envelope is the single source of truth for the status +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +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_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") + + assert await store.get_history_item_ids(None, "conv-1", limit=100) == [] + + +@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) == []