From 175ce970a018c44bb9b2c08c605ad96b9ec453c9 Mon Sep 17 00:00:00 2001 From: Maxim Svistunov Date: Tue, 6 Oct 2026 14:26:30 +0200 Subject: [PATCH 1/3] LCORE-3788: add the mark of a turn a shield blocked A turn a shield blocked has to stay in the conversation, so the history is complete, and has to stay out of what compaction replays to the model. Nothing tells the two apart today: a blocked message is stored like any other. A conversation item has no metadata, so the mark is the item id. OGX keeps an id the caller supplies, and the ids it generates are msg_ followed by hex digits or a UUID. utils/blocked_turns.py holds the prefix (msg_blocked_), a generator for new ids, the predicate, a filter for the code that builds the model's context, and mark_blocked, which gives every message of a turn a new blocked id before it is stored. Nothing uses the module yet, so behaviour does not change. Tests: tests/unit/utils/test_blocked_turns.py, one test per function. The predicate is also checked on the type OGX lists items as. --- src/utils/blocked_turns.py | 100 +++++++++++++++++++++++++ tests/unit/utils/test_blocked_turns.py | 80 ++++++++++++++++++++ 2 files changed, 180 insertions(+) create mode 100644 src/utils/blocked_turns.py create mode 100644 tests/unit/utils/test_blocked_turns.py diff --git a/src/utils/blocked_turns.py b/src/utils/blocked_turns.py new file mode 100644 index 000000000..bb2065557 --- /dev/null +++ b/src/utils/blocked_turns.py @@ -0,0 +1,100 @@ +"""The mark of a turn a shield blocked (LCORE-3788). + +A turn a shield blocked is stored in the conversation, so that the history a +client reads is complete. It must not reach the model again. Conversation +compaction rebuilds the model's context from the stored items, so a blocked +text that is replayed or summarized there was not blocked, only delayed by one +turn. + +A conversation item has no field for metadata, so the mark is the item id: the +messages of a blocked turn are stored under an id that starts with +``BLOCKED_ITEM_ID_PREFIX``. OGX keeps an id the caller supplies. The ids it +generates itself are ``msg_`` followed by hexadecimal digits or a UUID, so they +never start with the prefix. + +Only turns stored with the mark are recognized. A blocked turn stored before +the mark existed looks like any other turn. +""" + +import secrets +from collections.abc import Sequence +from typing import Any, Final + +BLOCKED_ITEM_ID_PREFIX: Final[str] = "msg_blocked_" +"""Start of the id of every message that belongs to a blocked turn.""" + + +def new_blocked_item_id() -> str: + """Return a new id for a message of a blocked turn. + + Item ids are unique across all conversations in OGX, and storing an item + under an id that exists overwrites that item. So the id is random, with as + many random bytes as OGX uses for the ids it generates. + + Returns: + ``BLOCKED_ITEM_ID_PREFIX`` followed by 48 hexadecimal digits. + + """ + return f"{BLOCKED_ITEM_ID_PREFIX}{secrets.token_hex(24)}" + + +def is_blocked_item(item: Any) -> bool: + """Tell whether a stored conversation item belongs to a blocked turn. + + Parameters: + item: A conversation item, as OGX lists it. + + Returns: + True when the item has an id that starts with + ``BLOCKED_ITEM_ID_PREFIX``. + + """ + item_id = getattr(item, "id", None) + return isinstance(item_id, str) and item_id.startswith(BLOCKED_ITEM_ID_PREFIX) + + +def exclude_blocked_items(items: Sequence[Any]) -> list[Any]: + """Return the conversation items that do not belong to a blocked turn. + + This is for code that builds what the model reads. A client that reads the + conversation gets the blocked turns as well. + + A summary marker records how many stored items it covers, blocked ones + included. Apply this filter after the items were cut at that count, never + before. + + Parameters: + items: Conversation items, as stored. + + Returns: + The items that are not blocked: the same objects, in their original + order. + + """ + return [item for item in items if not is_blocked_item(item)] + + +def mark_blocked(items: Sequence[dict[str, Any]]) -> list[dict[str, Any]]: + """Mark the items of a turn as blocked, before they are stored. + + Every message gets a new blocked id, in place of the id it may have had: + the input of a request can carry ids the client chose. Items of other + types are left as they are; the model's context is built from messages + only. + + Parameters: + items: The items of the turn, as dictionaries ready to be stored. + + Returns: + New dictionaries, in the same order; the ones passed in are not + modified. + + """ + return [ + ( + {**item, "id": new_blocked_item_id()} + if item.get("type") == "message" + else dict(item) + ) + for item in items + ] diff --git a/tests/unit/utils/test_blocked_turns.py b/tests/unit/utils/test_blocked_turns.py new file mode 100644 index 000000000..6b7aaf66d --- /dev/null +++ b/tests/unit/utils/test_blocked_turns.py @@ -0,0 +1,80 @@ +"""Unit tests for the mark of a turn a shield blocked (LCORE-3788).""" + +from ogx_api.openai_responses import OpenAIResponseMessage +from ogx_client.models.open_ai_response_message11_variants import ( + OpenAIResponseMessage11Variants as ConversationItem, +) + +from utils.blocked_turns import ( + BLOCKED_ITEM_ID_PREFIX, + exclude_blocked_items, + is_blocked_item, + mark_blocked, + new_blocked_item_id, +) + + +def _msg(text: str, item_id: str | None = None) -> OpenAIResponseMessage: + """Build a stored user message, with the id it was stored under.""" + return OpenAIResponseMessage(role="user", content=text, id=item_id) + + +def test_new_blocked_item_id_is_prefixed_and_unique() -> None: + """Every id carries the prefix and 48 random hex characters, as OGX ids do.""" + first, second = new_blocked_item_id(), new_blocked_item_id() + + assert first.startswith(BLOCKED_ITEM_ID_PREFIX) + assert len(first) == len(BLOCKED_ITEM_ID_PREFIX) + 48 + assert first != second + + +def test_is_blocked_item_reads_the_id_prefix() -> None: + """An item is blocked when its id has the prefix, and only then.""" + blocked_id = new_blocked_item_id() + listed = ConversationItem.from_dict( + {"type": "message", "role": "user", "content": "x", "id": blocked_id} + ) + + assert is_blocked_item(_msg("x", blocked_id)) + assert is_blocked_item(listed) # the type OGX lists conversation items as + assert not is_blocked_item(_msg("x", "msg_0123456789abcdef")) + assert not is_blocked_item(_msg("x")) + assert not is_blocked_item({"type": "function_call"}) + + +def test_exclude_blocked_items_keeps_the_rest_as_they_are() -> None: + """The items that are left are the same objects, in the same order.""" + question, answer = _msg("question", "msg_01"), _msg("answer", "msg_02") + items = [ + question, + _msg("blocked question", new_blocked_item_id()), + _msg("refusal", new_blocked_item_id()), + answer, + ] + + kept = exclude_blocked_items(items) + + assert len(kept) == 2 + assert kept[0] is question + assert kept[1] is answer + + +def test_mark_blocked_gives_every_message_a_new_blocked_id() -> None: + """Messages get a blocked id of their own; other items and the input stay.""" + tool_output = {"type": "function_call_output", "call_id": "c1", "output": "o"} + items = [ + {"type": "message", "role": "user", "content": "q", "id": "msg_client"}, + tool_output, + {"type": "message", "role": "assistant", "content": "refusal"}, + ] + + marked = mark_blocked(items) + + assert marked[0]["id"].startswith(BLOCKED_ITEM_ID_PREFIX) + assert marked[2]["id"].startswith(BLOCKED_ITEM_ID_PREFIX) + assert marked[0]["id"] != marked[2]["id"] + assert marked[0]["content"] == "q" + assert marked[1] == tool_output + # the dicts handed in are not modified + assert items[0]["id"] == "msg_client" + assert "id" not in items[2] From 56a90d798d9024357b20f4af7a347ef8e1da0feb Mon Sep 17 00:00:00 2001 From: Maxim Svistunov Date: Tue, 6 Oct 2026 14:38:27 +0200 Subject: [PATCH 2/3] LCORE-3788: keep blocked turns out of what compaction sends to the model Compaction rebuilds the model's context from the stored conversation items. It skipped tool items and summary markers, and nothing else, so a turn a shield had blocked was replayed verbatim on the next compacted request, and folded into a summary once it left the buffer. _load_compaction_state now drops the items that carry the blocked mark from the recent items, which feed the token estimate, the partition, the summarizer and the explicit input. needs_compaction_path leaves them out of its estimate. The filter runs after the stored items were cut at the count the last marker records: that count stays an index into the stored items, so _covered_item_count gets the unfiltered list. A marked message is never read as a summary marker (is_marker_item). A blocked text with the marker prefix would otherwise be sent to the model as a summary and move the boundary of the recent turns. The conversation read therefore returns such a message, where it left out every user message with the prefix before; the docstring of exclude_marker_items names the exception. Nothing stores a turn with the mark yet. Tests, unit: blocked pairs before and after the boundary are absent from the explicit input (also fails with the filter ahead of the cut); a blocked text with the marker prefix is no summary, no boundary, and exclude_marker_items keeps it; the summarizer does not get a blocked turn and the marker says [covers:4], the raw index; the estimate of needs_compaction_path does not count it. All four fail without the change. Integration: the conversation read returns a blocked turn, and fails when the handler leaves marked items out. --- src/utils/conversation_compaction.py | 23 +++- .../test_compaction_conversations_api.py | 45 +++++++ .../utils/test_conversation_compaction.py | 126 +++++++++++++++++- 3 files changed, 188 insertions(+), 6 deletions(-) diff --git a/src/utils/conversation_compaction.py b/src/utils/conversation_compaction.py index cc1f95fad..ee0a21f7a 100644 --- a/src/utils/conversation_compaction.py +++ b/src/utils/conversation_compaction.py @@ -63,6 +63,7 @@ from models.common.turn_summary import ContextStatus from models.compaction import ConversationSummary from models.config import CompactionConfiguration, InferenceConfiguration +from utils.blocked_turns import exclude_blocked_items, is_blocked_item from utils.compaction import ( CallCounter, partition_conversation, @@ -214,8 +215,13 @@ def context_status(self) -> ContextStatus: def is_marker_item(item: Any) -> bool: - """Return True when *item* is a compaction summary marker message.""" - if not is_message_item(item): + """Return True when *item* is a compaction summary marker message. + + A message of a turn a shield blocked is never a marker, whatever its + text: a marker is read back as a summary and sets the boundary of the + recent turns (LCORE-3788). + """ + if not is_message_item(item) or is_blocked_item(item): return False return extract_message_text(item).startswith(MARKER_SENTINEL) @@ -339,7 +345,8 @@ def exclude_marker_items(items: Sequence[Any]) -> list[Any]: are recognized as well and no migration is needed. Markers are only ever written as user messages, so a message of any other role is kept even when its text starts with the sentinel. A message the user typed with the - sentinel at its start cannot be told apart from a marker and is left out. + sentinel at its start cannot be told apart from a marker and is left out, + unless it carries the blocked mark (:func:`is_marker_item`). Parameters: items: Conversation items, as stored. @@ -595,6 +602,10 @@ def _load_compaction_state( configured. The recent-verbatim boundary is always derived from marker position in the conversation items. + Turns a shield blocked are left out of the recent items, after the + boundary was applied to the stored items: they are neither estimated, + summarized nor replayed (LCORE-3788). + Returns ``(summaries, cached_summaries, recent_items)``. """ cached_summaries = _read_cached_summaries( @@ -605,7 +616,7 @@ def _load_compaction_state( if cached_summaries else _marker_summaries(items) ) - recent_items = _items_after_last_marker(items) + recent_items = exclude_blocked_items(_items_after_last_marker(items)) return summaries, cached_summaries, recent_items @@ -927,6 +938,8 @@ async def needs_compaction_path( return True context_window = resolve_context_window(params.model, inference_config) estimated = estimate_tokens(params.instructions or "", encoding_name) - estimated += estimate_conversation_tokens(items, encoding_name=encoding_name) + estimated += estimate_conversation_tokens( + exclude_blocked_items(items), encoding_name=encoding_name + ) estimated += _estimate_response_input_tokens(params.input, encoding_name) return _should_compact(estimated, context_window, compaction_config) diff --git a/tests/integration/endpoints/test_compaction_conversations_api.py b/tests/integration/endpoints/test_compaction_conversations_api.py index 5de78795f..db3ecb01d 100644 --- a/tests/integration/endpoints/test_compaction_conversations_api.py +++ b/tests/integration/endpoints/test_compaction_conversations_api.py @@ -46,6 +46,7 @@ marker, msg, ) +from utils.blocked_turns import new_blocked_item_id from utils.conversation_compaction import MARKER_SENTINEL NEW_QUERY = "What else can you help with?" @@ -260,6 +261,50 @@ async def test_stored_markers_of_both_shapes_are_not_returned( ] assert_marker_count(mock_conversation_store, CONV_ID_LLAMA, 2) + @pytest.mark.asyncio + async def test_blocked_turn_is_returned( + self, + test_config: AppConfig, + mock_ogx_client: AsyncMockType, + mock_conversation_store: InMemoryConversationStore, + non_admin_test_request: Request, + test_auth: AuthTuple, + patch_db_session: Session, + ) -> None: + """A turn a shield blocked is part of the history a client reads (LCORE-3788). + + Unlike a marker, the user did send it. Only what the model reads + leaves blocked turns out. + """ + _ = test_config + _ = mock_ogx_client + + create_existing_conversation(patch_db_session, test_auth[0]) + await mock_conversation_store.create( + conversation_id=CONV_ID_LLAMA, + items=[ + OpenAIResponseMessage( + role="user", + content="the blocked question", + id=new_blocked_item_id(), + ), + OpenAIResponseMessage( + role="assistant", content="the refusal", id=new_blocked_item_id() + ), + ], + ) + + response = await get_conversation_v1( + request=non_admin_test_request, + conversation_id=EXISTING_CONV_ID, + auth=test_auth, + ) + + assert _returned_messages(response) == [ + ("user", "the blocked question"), + ("assistant", "the refusal"), + ] + @pytest.mark.asyncio async def test_v2_history_holds_no_marker_after_a_compaction( self, diff --git a/tests/unit/utils/test_conversation_compaction.py b/tests/unit/utils/test_conversation_compaction.py index 9100d1271..1fd633b38 100644 --- a/tests/unit/utils/test_conversation_compaction.py +++ b/tests/unit/utils/test_conversation_compaction.py @@ -1,7 +1,7 @@ """Unit tests for runtime conversation compaction (LCORE-1572).""" # Tests exercise internal helpers directly. -# pylint: disable=protected-access +# pylint: disable=protected-access,too-many-lines import asyncio from types import SimpleNamespace @@ -20,6 +20,7 @@ from models.compaction import ConversationSummary from models.config import CompactionConfiguration, InferenceConfiguration from utils import conversation_compaction as cc +from utils.blocked_turns import new_blocked_item_id MODEL = "openai/gpt-5.6-luna" CATALOG_MODEL = "rhaiis/granite" @@ -44,6 +45,13 @@ def _marker_covering(covered_items: int, text: str) -> OpenAIResponseMessage: ) +def _blocked(role: str, text: str) -> OpenAIResponseMessage: + """Build a message item of a turn a shield blocked (LCORE-3788).""" + return OpenAIResponseMessage( + role=cast("Any", role), content=text, id=new_blocked_item_id() + ) + + def _params(input_text: str = "new question") -> ResponsesApiParams: return ResponsesApiParams( input=input_text, @@ -104,6 +112,21 @@ def test_is_marker_item() -> None: assert cc.is_marker_item({"type": "function_call"}) is False +def test_blocked_message_with_the_marker_prefix_is_not_a_marker() -> None: + """A blocked message is never read as a summary marker (LCORE-3788). + + Its text would be sent to the model as a summary, and the count in it + would move the boundary of the recent turns. + """ + forged = _blocked("user", f"{cc.MARKER_SENTINEL} [covers:0] blocked text") + items = [_msg("user", "old q"), _marker_covering(1, "summary"), forged] + + assert cc.is_marker_item(forged) is False + assert cc._marker_summaries(items) == ["summary"] + assert cc._items_after_last_marker(items) == [forged] + assert cc.exclude_marker_items(items) == [items[0], forged] + + def test_items_after_last_marker() -> None: """Markers without a covers count fall back to their position (pre-LCORE-4219).""" items = [ @@ -395,6 +418,84 @@ async def test_triggers_summarization_and_writes_marker(mocker: MockerFixture) - assert texts[-1] == "follow-up" +@pytest.mark.asyncio +async def test_blocked_turn_is_not_replayed(mocker: MockerFixture) -> None: + """A turn a shield blocked is left out of the explicit input (LCORE-3788).""" + items = [ + _msg("user", "old q"), + _blocked("user", "the blocked question"), + _blocked("assistant", "the refusal"), + _msg("user", "recent q"), + _msg("assistant", "recent a"), + # The count is an index into the stored items, blocked ones included. + _marker_covering(3, "the earlier conversation summary"), + _blocked("user", "a later blocked question"), + _blocked("assistant", "a later refusal"), + ] + mocker.patch.object( + cc, "get_all_conversation_items", mocker.AsyncMock(return_value=items) + ) + + result = await cc.apply_compaction_blocking( + client=mocker.AsyncMock(), + params=_params("brand new"), + inference_config=_inference(1_000_000), # huge: no new trigger + compaction_config=_compaction(), + ) + + assert [cast("OpenAIResponseMessage", m).content for m in result.params.input] == [ + "Summary of earlier conversation:\nthe earlier conversation summary", + "recent q", + "recent a", + "brand new", + ] + + +@pytest.mark.asyncio +async def test_blocked_turn_is_not_summarized(mocker: MockerFixture) -> None: + """A blocked turn is not sent to the summarizer, but counts as a stored item. + + The count a marker records is an index into the stored items, blocked + ones included: the turn kept verbatim here is the fifth stored item, so + the marker covers four, not the two that were summarized (LCORE-3788). + """ + items = [ + _msg("user", "q1 " * 50), + _msg("assistant", "a1 " * 50), + _blocked("user", "the blocked question"), + _blocked("assistant", "the refusal"), + _msg("user", "q2"), + _msg("assistant", "a2"), + ] + mocker.patch.object( + cc, "get_all_conversation_items", mocker.AsyncMock(return_value=items) + ) + client = mocker.AsyncMock() + client.responses.create.return_value = SimpleNamespace( + output=[SimpleNamespace(content=[SimpleNamespace(text="condensed")])] + ) + + result = await cc.apply_compaction_blocking( + client=client, + params=_params("follow-up"), + inference_config=_inference(200), # small window forces the trigger + compaction_config=_compaction(threshold_ratio=0.1, buffer_turns=1), + ) + + transcript = client.responses.create.await_args.kwargs["input"] + assert "q1" in transcript + assert "the blocked question" not in transcript + assert "the refusal" not in transcript + written = client.items.create.await_args.kwargs["add_items_request"].items + assert cc._split_marker_text(written[0]) == (4, "condensed") + assert [cast("OpenAIResponseMessage", m).content for m in result.params.input] == [ + "Summary of earlier conversation:\ncondensed", + "q2", + "a2", + "follow-up", + ] + + @pytest.mark.asyncio async def test_streaming_emits_event_before_summarizing(mocker: MockerFixture) -> None: """In streaming mode a CompactionStartedEvent precedes the summary result.""" @@ -529,6 +630,29 @@ async def test_needs_compaction_path_under_threshold(mocker: MockerFixture) -> N ) +@pytest.mark.asyncio +async def test_needs_compaction_path_does_not_count_blocked_turns( + mocker: MockerFixture, +) -> None: + """Text that is never sent to the model does not count towards the threshold.""" + mocker.patch.object( + cc, + "get_all_conversation_items", + mocker.AsyncMock( + return_value=[_blocked("user", "q " * 50), _blocked("assistant", "a " * 50)] + ), + ) + assert ( + await cc.needs_compaction_path( + mocker.AsyncMock(), + _params(), + _inference(100), # over the threshold if the blocked turn counted + _compaction(threshold_ratio=0.1), + ) + is False + ) + + # --- cache as source of truth + recursive fold (R3) --- From c528506842118f09480c47c133dcefab1bcf97c3 Mon Sep 17 00:00:00 2001 From: Maxim Svistunov Date: Tue, 6 Oct 2026 14:52:44 +0200 Subject: [PATCH 3/3] LCORE-3788: store a blocked turn marked, through the owner of the write /v1/responses stores the turn of a request a shield blocked: the input as it arrived and the refusal. On a compacted conversation the next request replayed both to the model, so the shield only delayed the text by one turn. append_turn_items_to_conversation takes blocked=False. When set, every message of the turn is stored under a blocked id. PendingTurn sets it in store_blocked and nowhere else, so a completed or an interrupted turn is stored as before. /v1/responses reaches store_blocked in every mode (compacted or not, with a previous response), so the endpoint itself does not change. Together with the filter of the previous commit, a blocked turn stays in the conversation and is neither replayed nor summarized. Turns that were stored before this commit carry no mark and are still replayed. Tests: - integration, test_turn_persistence.py: a request blocked by a shield and then a follow-up on the same compacted conversation, blocking and streaming. The store holds the blocked input and the refusal with ids starting msg_blocked_, the follow-up turn is not marked, and no item of the second request's input holds the blocked text or the refusal. With the filter taken out, the test fails on the request body. - unit: the write marks every message of a blocked turn; of the three ways the owner stores a turn, only the blocked one is marked. - three assertions in the /v1/responses unit tests follow the new argument. --- src/utils/conversations.py | 8 +++ src/utils/pending_turn.py | 19 +++++-- .../endpoints/test_turn_persistence.py | 54 ++++++++++++++++++- tests/unit/app/endpoints/test_responses.py | 5 +- tests/unit/utils/test_conversations.py | 23 ++++++++ tests/unit/utils/test_pending_turn.py | 23 ++++++++ 6 files changed, 125 insertions(+), 7 deletions(-) diff --git a/src/utils/conversations.py b/src/utils/conversations.py index 471a2e843..448077812 100644 --- a/src/utils/conversations.py +++ b/src/utils/conversations.py @@ -64,6 +64,7 @@ from models.common.responses.types import ResponseInput from models.common.turn_summary import ToolCallSummary, ToolResultSummary from models.database.conversations import UserTurn +from utils.blocked_turns import mark_blocked from utils.responses import parse_arguments_string type FunctionCallOutputPart = InputTextContent | InputImageContent | InputFileContent @@ -541,6 +542,8 @@ async def append_turn_items_to_conversation( conversation_id: str, user_input: ResponseInput, llm_output: Sequence[OpenAIResponseOutput], + *, + blocked: bool = False, ) -> None: """ Append a turn (user input + LLM output) to a conversation in LLS database. @@ -550,6 +553,9 @@ async def append_turn_items_to_conversation( conversation_id: The OGX conversation ID. user_input: User input text or list of ResponseItem. llm_output: Output from the LLM: a list of OpenAIResponseOutput. + blocked: Whether a shield blocked the turn. Its messages are then + stored with the blocked mark, which keeps them out of what + conversation compaction sends to the model (LCORE-3788). """ if isinstance(user_input, str): @@ -560,6 +566,8 @@ async def append_turn_items_to_conversation( items = [item.model_dump(exclude_none=True) for item in user_input] items.extend(item.model_dump(exclude_none=True) for item in llm_output) + if blocked: + items = mark_blocked(items) try: await client.items.create( conversation_id, diff --git a/src/utils/pending_turn.py b/src/utils/pending_turn.py index d1d192e77..f4bd64279 100644 --- a/src/utils/pending_turn.py +++ b/src/utils/pending_turn.py @@ -165,6 +165,10 @@ async def store_completed( async def store_blocked(self, refusal: OpenAIResponseMessage) -> bool: """Store the turn of a request a shield blocked. + The turn is stored with the blocked mark: it stays in the history, + and conversation compaction does not send it to the model on a later + request (LCORE-3788). + Parameters: refusal: The refusal message returned in place of an answer. @@ -175,7 +179,7 @@ async def store_blocked(self, refusal: OpenAIResponseMessage) -> bool: HTTPException: When the write fails. """ - return await self._store("blocked", [refusal]) + return await self._store("blocked", [refusal], blocked=True) async def store_interrupted(self, partial_response: str) -> bool: """Store the turn of a stream the client interrupted. @@ -227,7 +231,11 @@ def _settle(self, outcome: str) -> bool: return True async def _store( - self, outcome: str, output_items: Sequence[OpenAIResponseOutput] + self, + outcome: str, + output_items: Sequence[OpenAIResponseOutput], + *, + blocked: bool = False, ) -> bool: """Settle the turn and append it to the conversation. @@ -238,6 +246,7 @@ async def _store( Parameters: outcome: How the turn ended; recorded when this call settles it. output_items: The assistant's side of the turn. + blocked: Whether the turn is stored with the blocked mark. Returns: Whether this call wrote the turn. ``False`` when the turn was @@ -252,7 +261,11 @@ async def _store( return False try: await append_turn_items_to_conversation( - self.client, self.conversation_id, self.user_input, output_items + self.client, + self.conversation_id, + self.user_input, + output_items, + blocked=blocked, ) except BaseException: self.outcome = f"failed: {outcome}" diff --git a/tests/integration/endpoints/test_turn_persistence.py b/tests/integration/endpoints/test_turn_persistence.py index dcb933b45..fec5bb7be 100644 --- a/tests/integration/endpoints/test_turn_persistence.py +++ b/tests/integration/endpoints/test_turn_persistence.py @@ -53,7 +53,7 @@ from authentication.interface import AuthTuple from configuration import AppConfig from models.api.requests import QueryRequest, ResponsesRequest -from models.common.moderation import ShieldModerationBlocked +from models.common.moderation import ShieldModerationBlocked, ShieldModerationPassed from models.common.responses.contexts import ResponsesContext from models.common.responses.responses_api_params import ResponsesApiParams from models.common.turn_summary import TurnSummary @@ -78,6 +78,7 @@ mock_a2a_agent, msg, ) +from utils.blocked_turns import is_blocked_item from utils.pending_turn import TurnNotStoredError from utils.stream_interrupts import ( CancelStreamResult, @@ -87,6 +88,7 @@ from utils.token_estimator import extract_message_text NEW_QUERY = "What else can you help with?" +BLOCKED_QUERY = "A question the shield does not let through" REFUSAL = "Content blocked by safety shield" PARTIAL_ANSWER = "Ansible is" PREVIOUS_RESPONSE_ID = "resp_previous_turn" @@ -703,12 +705,13 @@ async def _send_response_request( stream: bool, store: bool = True, previous_response_id: Optional[str] = None, + query: str = NEW_QUERY, ) -> Any: """Send the new input on the stored conversation and read the answer.""" response = await responses_endpoint_handler( request=test_request, responses_request=ResponsesRequest( - input=NEW_QUERY, + input=query, model=TEST_MODEL, conversation=None if previous_response_id else EXISTING_CONV_ID, previous_response_id=previous_response_id, @@ -809,6 +812,53 @@ async def test_blocked_turn_is_stored_once( mock_conversation_store, stored_conversation() + _turn(REFUSAL) ) + @pytest.mark.asyncio + @pytest.mark.parametrize("stream", [False, True], ids=["blocking", "streaming"]) + async def test_blocked_turn_is_not_sent_to_the_model_by_the_next_request( + self, + stream: bool, + test_config: AppConfig, + mock_ogx_client: AsyncMockType, + mock_conversation_store: InMemoryConversationStore, + test_request: Request, + test_auth: AuthTuple, + patch_db_session: Session, + mocker: MockerFixture, + ) -> None: + """A blocked turn stays in the conversation and out of the next request. + + The first request is blocked by a shield, the second is an ordinary + follow-up on the same compacted conversation. The store holds the + blocked input and the refusal, marked as blocked, and the body of the + second request carries neither (LCORE-3788). + """ + create_existing_conversation(patch_db_session, test_auth[0]) + await _seed(test_config, mock_conversation_store, _compacted_conversation()) + mocker.patch( + "app.endpoints.responses.run_shield_moderation_v2", + side_effect=[_blocked(), ShieldModerationPassed()], + ) + + await _send_response_request( + test_request, test_auth, stream, query=BLOCKED_QUERY + ) + await _send_response_request(test_request, test_auth, stream) + + await _assert_stored( + mock_conversation_store, + _compacted_conversation() + + _turn(REFUSAL, BLOCKED_QUERY) + + _turn(DEFAULT_MODEL_RESPONSE), + ) + stored = await collect_items(mock_conversation_store, CONV_ID_LLAMA) + assert all(str(item.id).startswith("msg_blocked_") for item in stored[3:5]) + assert not any(is_blocked_item(item) for item in stored[5:]) + mock_ogx_client.responses.create.assert_awaited_once() + sent_input = mock_ogx_client.responses.create.await_args.kwargs["input"] + assert NEW_QUERY in str(sent_input[-1]) + assert not any(BLOCKED_QUERY in str(item) for item in sent_input) + assert not any(REFUSAL in str(item) for item in sent_input) + @pytest.mark.asyncio @pytest.mark.parametrize("stream", [False, True], ids=["blocking", "streaming"]) @pytest.mark.parametrize( diff --git a/tests/unit/app/endpoints/test_responses.py b/tests/unit/app/endpoints/test_responses.py index df9c1006c..e90f0b659 100644 --- a/tests/unit/app/endpoints/test_responses.py +++ b/tests/unit/app/endpoints/test_responses.py @@ -715,6 +715,7 @@ async def test_responses_blocked_with_conversation_appends_refusal( VALID_CONV_ID, responses_request.input, [mock_moderation.refusal_response], + blocked=True, ) assert isinstance(response, ResponsesResponse) payload = response.model_dump() @@ -3027,7 +3028,7 @@ async def test_append_previous_response_turn_compacted(mocker: MockerFixture) -> await _append_previous_response_turn(api_params, context, ["out"]) append.assert_awaited_once_with( - context.client, "conv_x", "the original query", ["out"] + context.client, "conv_x", "the original query", ["out"], blocked=False ) @@ -3081,5 +3082,5 @@ async def test_persist_blocked_response_turn_compacted(mocker: MockerFixture) -> await _persist_blocked_response_turn(api_params, context) append.assert_awaited_once_with( - context.client, "conv_x", "the original query", [refusal] + context.client, "conv_x", "the original query", [refusal], blocked=True ) diff --git a/tests/unit/utils/test_conversations.py b/tests/unit/utils/test_conversations.py index ae7c67052..3d4228800 100644 --- a/tests/unit/utils/test_conversations.py +++ b/tests/unit/utils/test_conversations.py @@ -27,6 +27,7 @@ from constants import DEFAULT_RAG_TOOL from models.common.turn_summary import ToolCallSummary from models.database.conversations import UserTurn +from utils.blocked_turns import is_blocked_item from utils.conversations import ( _build_tool_call_summary_from_item, _extract_text_from_content, @@ -916,6 +917,28 @@ async def test_appends_user_input_and_llm_output( assert items[1].type == "message" and items[1].role == "assistant" assert items[1].content == "I cannot help with that" + @pytest.mark.asyncio + async def test_blocked_turn_is_stored_with_the_blocked_mark( + self, mocker: MockerFixture + ) -> None: + """Every message of a blocked turn is stored under a blocked id (LCORE-3788).""" + mock_client = mocker.Mock() + mock_client.items.create = mocker.AsyncMock(return_value=None) + + await append_turn_items_to_conversation( + mock_client, + conversation_id="conv-123", + user_input=[ + OpenAIResponseMessage(role="user", content="Hello", id="msg_client") + ], + llm_output=[OpenAIResponseMessage(role="assistant", content="Refused")], + blocked=True, + ) + + stored = mock_client.items.create.call_args[1]["add_items_request"].items + assert [item.content for item in stored] == ["Hello", "Refused"] + assert all(is_blocked_item(item) for item in stored) + class TestAppendTurnToConversation: # pylint: disable=too-few-public-methods """Tests for append_turn_to_conversation function.""" diff --git a/tests/unit/utils/test_pending_turn.py b/tests/unit/utils/test_pending_turn.py index c7d393e07..71ec22c75 100644 --- a/tests/unit/utils/test_pending_turn.py +++ b/tests/unit/utils/test_pending_turn.py @@ -15,6 +15,7 @@ from models.common.responses.responses_api_params import ResponsesApiParams from models.config import CompactionConfiguration, InferenceConfiguration +from utils.blocked_turns import is_blocked_item from utils.conversation_compaction import MARKER_SENTINEL, apply_compaction_blocking from utils.pending_turn import PendingTurn, TurnNotStoredError, pending_turn from utils.token_estimator import extract_message_text @@ -35,6 +36,7 @@ class RecordingClient: # pylint: disable=too-few-public-methods def __init__(self, fail_with: Optional[Exception] = None) -> None: """Create the client, optionally one whose writes fail.""" self.stored: list[tuple[str, str, str]] = [] + self.marked_blocked: list[bool] = [] self.writes = 0 self._fail_with = fail_with self.items = self @@ -50,6 +52,9 @@ async def create( (conversation_id, str(item.role), extract_message_text(item)) for item in add_items_request.items ) + self.marked_blocked.extend( + is_blocked_item(item) for item in add_items_request.items + ) def _params( @@ -165,6 +170,24 @@ async def test_blocked_turn_is_stored(compacted: bool) -> None: assert client.stored == [USER, (CONVERSATION, "assistant", "blocked by a shield")] +@pytest.mark.asyncio +async def test_only_a_blocked_turn_is_stored_as_blocked() -> None: + """A blocked turn carries the mark that keeps it from the model (LCORE-3788). + + Both of its messages do, the input and the refusal. A turn that ended any + other way must not: it would be missing from the model's context. + """ + blocked, completed, interrupted = (RecordingClient() for _ in range(3)) + + await _turn(blocked, compacted=True).store_blocked(REFUSAL) + await _turn(completed, compacted=True).store_completed([ANSWER]) + await _turn(interrupted, compacted=True).store_interrupted("half an ans") + + assert blocked.marked_blocked == [True, True] + assert completed.marked_blocked == [False, False] + assert interrupted.marked_blocked == [False, False] + + @pytest.mark.asyncio @pytest.mark.parametrize("compacted", [False, True]) async def test_interrupted_turn_is_stored(compacted: bool) -> None: