LCORE-4593: redact the query, replayed turns and summaries of compacted-mode requests

docs/user_doc/shields_guide.md

@@ -95,6 +95,22 @@ for a complete example.
 
 Invalid regex patterns are rejected at configuration load time.
 
+In a [compacted conversation](conversation_compaction.md), Lightspeed Core
+Stack builds the model input and stores the turn itself. It applies the
+redaction shields selected for the request to the new query there, so that the
+model receives and the conversation stores the same redacted query as outside
+compacted mode. The user messages and the summaries it replays from the
+conversation, and the user messages it hands to the summarizer, go through the
+same rules each time they are sent, so a replacement must not itself match a
+rule.
+
+Not covered: answers of the model are stored and replayed as the model
+produced them, in both modes, and the summarizer gets them as stored. Text
+that is already stored is not rewritten, so
+`GET /v1/conversations/{conversation_id}` returns it as stored. The recursive
+re-summarization sends the stored summaries as they are, and `/v1/responses`
+replays a compacted conversation as stored.
+
 ## granite_guardian
 
 IBM Granite Guardian screening with configurable risk definitions, thresholds,

src/app/endpoints/a2a.py

@@ -76,6 +76,7 @@
 from utils.pydantic_ai_helpers import build_agent, captured_output_items
 from utils.query import extract_provider_and_model_from_model_id
 from utils.responses import prepare_responses_params
+from utils.shields import request_redactor
 from utils.suid import normalize_conversation_id
 from version import __version__
 
@@ -241,6 +242,7 @@ async def _compact_a2a_request(
         responses_params,
         configuration.inference,
         configuration.compaction,
+        redact=request_redactor(configuration.shields),
     )
     return compaction.params, PendingTurn.for_request(
         client, compaction.params, compaction.original_input

src/app/endpoints/query.py

@@ -60,7 +60,7 @@
     maybe_get_topic_summary,
     prepare_responses_params,
 )
-from utils.shields import validate_shield_ids_override
+from utils.shields import request_redactor, validate_shield_ids_override
 from utils.suid import normalize_conversation_id
 from utils.types import Responses
 from utils.vector_search import build_rag_context
@@ -239,6 +239,7 @@ async def _handle_query_with_tracing(
         responses_params,
         configuration.inference,
         configuration.compaction,
+        redact=request_redactor(configuration.shields, query_request.shield_ids),
         cache=configured_conversation_cache(),
         user_id=user_id,
         skip_user_id_check=_skip_userid_check,

src/app/endpoints/streaming_query.py

@@ -81,7 +81,7 @@
     extract_vector_store_ids_from_tools,
     prepare_responses_params,
 )
-from utils.shields import validate_shield_ids_override
+from utils.shields import request_redactor, validate_shield_ids_override
 from utils.streaming_sse import (
     http_exception_stream_event,
     stream_compaction_event,
@@ -431,6 +431,9 @@ async def generate_response_with_compaction(
                 configuration.inference,
                 configuration.compaction,
                 emit_events=True,
+                redact=request_redactor(
+                    configuration.shields, context.query_request.shield_ids
+                ),
                 cache=configured_conversation_cache(),
                 user_id=context.user_id,
                 skip_user_id_check=context.skip_userid_check,

src/utils/conversation_compaction.py

@@ -80,6 +80,7 @@
     extract_message_text,
     is_message_item,
 )
+from utils.types import TextRedactor
 
 logger = get_logger(__name__)
 
@@ -184,7 +185,7 @@ class CompactionResult:
             or the cache), so the ``conversation`` parameter is omitted. This is
             True whether the summary was created this request or reused from a
             prior one. Drives ``context_status``.
-        original_input: The new user query exactly as it arrived (before the
+        original_input: The new user query after redaction, if any (before the
             explicit-input rewrite). Populated only in compacted mode (where
             ``compacted`` is True); ``None`` otherwise. In compacted mode the
             caller hands it to ``utils.pending_turn.PendingTurn``, which
@@ -438,6 +439,33 @@ def _query_input_message(original_input: ResponseInput) -> list[Any]:
     return list(original_input)
 
 
+def _redact_user_messages(
+    items: list[Any], redact: Optional[TextRedactor]
+) -> list[Any]:
+    """Return conversation items with the text of their user messages redacted.
+
+    User messages are the ones :func:`_verbatim_input_message` replays with
+    the user role; other messages are left as stored, as the redaction
+    capability leaves them. A message the redaction changes is replaced by a
+    text-only copy and every other item is the object that was passed in, so
+    the stored conversation is not rewritten. Without a redaction the items
+    are returned as they are.
+    """
+    if redact is None:
+        return items
+    redacted_items: list[Any] = []
+    for item in items:
+        replacement = item
+        role = getattr(item, "role", "user")
+        if is_message_item(item) and role not in ("system", "developer", "assistant"):
+            text = extract_message_text(item)
+            redacted_text = redact(text)
+            if redacted_text != text:
+                replacement = OpenAIResponseMessage(role="user", content=redacted_text)
+        redacted_items.append(replacement)
+    return redacted_items
+
+
 def _build_explicit_input(
     summaries: list[str],
     recent_items: list[Any],
@@ -701,6 +729,7 @@ async def apply_compaction(  # pylint: disable=too-many-arguments,too-many-posit
     compaction_config: CompactionConfiguration,
     emit_events: bool = False,
     encoding_name: str = DEFAULT_ENCODING_NAME,
+    redact: Optional[TextRedactor] = None,
     cache: Optional[Cache] = None,
     user_id: str = "",
     skip_user_id_check: bool = False,
@@ -725,6 +754,13 @@ async def apply_compaction(  # pylint: disable=too-many-arguments,too-many-posit
         compaction_config: Compaction tuning (enabled, threshold, buffer, ...).
         emit_events: Whether to yield CompactionStartedEvent before summarizing.
         encoding_name: tiktoken encoding name for estimation/summarization.
+        redact: The redaction of the shields the request selects
+            (``utils.shields.request_redactor``), or ``None``. It is applied
+            to the new query, so that a compacted turn sends and stores the
+            text the redaction capability sends outside compacted mode. It
+            is also applied to the summaries and the user turns replayed from
+            the conversation and to the user turns handed to the summarizer,
+            which may have been stored unredacted.
         cache: Conversation cache, the preferred summary store and the home of
             the persisted recursive fold. ``None`` (or a non-persisting backend)
             falls back to marker-only summaries with no folding.
@@ -755,6 +791,10 @@ async def apply_compaction(  # pylint: disable=too-many-arguments,too-many-posit
         summaries, cached_summaries, recent_items = _load_compaction_state(
             items, cache, user_id, conversation_id, skip_user_id_check
         )
+        # An item list (the /v1/responses form) is left alone: that endpoint
+        # passes no redaction.
+        if redact is not None and isinstance(original_input, str):
+            original_input = redact(original_input)
 
         context_window = resolve_context_window(model, inference_config)
         estimated = _estimate_total_tokens(
@@ -778,7 +818,7 @@ async def apply_compaction(  # pylint: disable=too-many-arguments,too-many-posit
                 summary = await summarize_chunk(
                     client,
                     model,
-                    old_items,
+                    _redact_user_messages(old_items, redact),
                     summarized_through_turn=_covered_item_count(items, keep_items),
                     encoding_name=encoding_name,
                 )
@@ -813,6 +853,13 @@ async def apply_compaction(  # pylint: disable=too-many-arguments,too-many-posit
             yield CompactionResult(params, compacted=False)
             return
 
+        if redact is not None:
+            # Summaries and replayed user turns go out as user messages and may
+            # have been stored unredacted. Not earlier: the kept turns are found
+            # among the stored items by identity (_covered_item_count).
+            summaries = [redact(text) for text in summaries]
+            recent_items = _redact_user_messages(recent_items, redact)
+
         # Compacted mode: lightspeed owns the context. Build explicit input and
         # stop passing the conversation parameter to inference.
         yield _compacted_result(params, summaries, recent_items, original_input)
@@ -824,6 +871,7 @@ async def apply_compaction_blocking(  # pylint: disable=too-many-arguments,too-m
     inference_config: InferenceConfiguration,
     compaction_config: CompactionConfiguration,
     encoding_name: str = DEFAULT_ENCODING_NAME,
+    redact: Optional[TextRedactor] = None,
     cache: Optional[Cache] = None,
     user_id: str = "",
     skip_user_id_check: bool = False,
@@ -842,6 +890,7 @@ async def apply_compaction_blocking(  # pylint: disable=too-many-arguments,too-m
         compaction_config,
         emit_events=False,
         encoding_name=encoding_name,
+        redact=redact,
         cache=cache,
         user_id=user_id,
         skip_user_id_check=skip_user_id_check,

src/utils/shields.py

@@ -36,6 +36,7 @@
 from pydantic_ai_lightspeed.capabilities.redaction._capability import (
     PiiRedactionCapability,
 )
+from pydantic_ai_lightspeed.capabilities.redaction.core import redact_text
 from utils.agents.error_handler import map_agent_inference_error
 from utils.input_sanitization import sanitize_input
 from utils.otel_tracing import (
@@ -44,6 +45,7 @@
     set_span_attributes,
     shield_span_attributes,
 )
+from utils.types import TextRedactor
 
 logger = get_logger(__name__)
 tracer = trace.get_tracer(__name__)
@@ -215,3 +217,43 @@ def get_shields_for_request(
         raise HTTPException(**response.model_dump())
 
     return [shield for shield in shields if shield.name in requested]
+
+
+def request_redactor(
+    shields: list[ShieldConfiguration],
+    shield_ids: Optional[list[str]] = None,
+) -> Optional[TextRedactor]:
+    """Build the text redaction of the shields a request selects.
+
+    The rules of the selected redaction shields are applied one after another
+    in configuration order, the order in which their capabilities run on an
+    agent. It is for the places where lightspeed-stack builds or stores text
+    that the redaction capability does not see.
+
+    Parameters:
+        shields: Configured LCS shields.
+        shield_ids: The shield selection of the request, with the meaning it
+            has for ``get_shields_for_request``.
+
+    Returns:
+        A function that redacts a text, or ``None`` when the selection holds
+        no redaction rule.
+
+    Raises:
+        HTTPException: 404 if shield_ids names a shield that is not configured.
+
+    """
+    patterns = [
+        compiled
+        for shield in get_shields_for_request(shields, shield_ids)
+        if isinstance(shield.config, RedactionConfig)
+        for compiled in shield.config.compiled_patterns
+    ]
+    if not patterns:
+        return None
+
+    def redact(text: str) -> str:
+        """Apply the selected redaction rules to a text."""
+        return redact_text(text, patterns).content
+
+    return redact

src/utils/types.py

@@ -1,5 +1,6 @@
 """Common types for the project."""
 
+from collections.abc import Callable
 from re import Pattern
 from typing import Any, TypeVar, cast
 
@@ -9,6 +10,9 @@
 
 CompiledPatterns = list[tuple[Pattern[str], str]]
 
+TextRedactor = Callable[[str], str]
+"""Rewrites a text with the redaction rules selected for a request."""
+
 T = TypeVar("T")