LCORE-3788: keep a turn a shield blocked on /v1/responses out of what compaction sends to the model

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
+    ]

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)

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,

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}"

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,