LCORE-1572: Integrate conversation compaction into the query flow (#1796)

* LCORE-1572: add conversation compaction and wire it into /v1/query

Introduce runtime conversation compaction (Option A): once a conversation
approaches the model's context window, lightspeed-stack summarizes older
turns and owns the LLM context itself instead of letting Llama Stack reload
the full history.

- src/utils/conversation_compaction.py: apply_compaction() async generator
  and apply_compaction_blocking() wrapper. Holds a per-conversation lock
  (R11), estimates tokens (LCORE-1569), partitions and summarizes old turns
  (LCORE-1570), writes the summary into the conversation as a marker item,
  and rebuilds the request as explicit input (summaries + recent verbatim
  turns + new query). Marker items track the boundary; the conversation_id
  is preserved and the full history stays in Llama Stack items for audit.
- models/common/responses/responses_api_params.py: omit_conversation flag so
  the conversation parameter is dropped from the request body in compacted
  mode while remaining on the object for identity.
- configuration.py: AppConfig.compaction accessor.
- app/endpoints/query.py: apply compaction after preparing params; in
  compacted mode store the completed turn against the original user query
  (the conversation parameter is no longer sent, so Llama Stack does not
  persist the turn automatically).

Background: the spec's original marker-keeps-conversation-parameter approach
was found unimplementable on llama-stack 0.6.0, which always reloads the full
conversation history when the conversation parameter is set. This restores
the spike's original explicit-input approach.

* LCORE-1572: unit tests for conversation compaction core and /v1/query

Cover marker detection and boundary selection, explicit-input assembly, the
trigger threshold, the disabled / no-context-window / existing-marker /
triggered paths of apply_compaction, the streaming CompactionStartedEvent
ordering, and compacted-turn storage.

* LCORE-1572: apply conversation compaction in the A2A endpoint

The A2A executor uses the same prepare_responses_params + Responses API
flow as /v1/query and persists conversation_id for multi-turn contexts, so
it accumulates context and must compact too.

- Run apply_compaction_blocking before responses.create (A2A is not a
  browser SSE stream, so no progress event is emitted).
- In compacted mode, persist the completed turn from the response.completed
  stream event, since the conversation parameter is no longer sent and Llama
  Stack therefore does not store the turn automatically.

* LCORE-1572: apply conversation compaction in the streaming_query endpoint

Stream /v1/streaming_query through the compaction-aware path only when the
conversation actually compacts, so non-compacting requests are unaffected
(byte-for-byte the existing flow, including HTTP error handling).

- conversation_compaction: add needs_compaction_path(), a cheap pre-stream
  predicate (no LLM, no lock) that is true only when the conversation already
  has a summary marker or would trigger a new compaction.
- streaming_query: when the predicate is true, stream via the new
  generate_response_with_compaction(), which emits the compaction progress
  event before the summarization LLM call (R12) and creates the response
  inside the stream, surfacing create-time errors as SSE error events.
  generate_response gains emit_start/compacted parameters and, in compacted
  mode, appends the completed turn to the conversation (the conversation
  parameter is not sent, so Llama Stack does not store it automatically).
- a2a: silence too-many-lines after the earlier compaction wiring.

* LCORE-1572: tests for the streaming compaction gate

Cover needs_compaction_path: disabled, existing-marker, over-threshold, and
under-threshold — the gate that keeps non-compacting requests on the
unchanged streaming path.

* LCORE-1572: apply conversation compaction in the /v1/responses endpoint

/v1/responses is the OpenAI-compatible Responses API, so compaction is
silent: no custom SSE event is injected (preserving wire compatibility) and
create-time error handling is unchanged. Summarization runs before the
response is created, on both the streaming and non-streaming paths.

- responses_endpoint_handler: run apply_compaction_blocking before the
  streaming/non-streaming split, gated to stateful single-conversation
  requests (store=True, a conversation present, no previous_response_id).
- ResponsesContext: carry compacted_original_input so the finalization can
  store the turn against the original user input.
- _append_previous_response_turn: generalized to also append the turn in
  compacted mode (the conversation parameter is dropped, so Llama Stack does
  not store the turn automatically) using the original input.

* LCORE-1572: tests for /v1/responses compacted-turn storage

Verify _append_previous_response_turn stores the turn against the original
input in compacted mode, and stores nothing when store is disabled.

* LCORE-1572: update spec doc to the as-built compaction design

Revise R10, R12, the architecture flow, the changed-request-flow section, and
the implementation guidance to match what was built: in compacted mode
lightspeed-stack builds explicit input and omits the Llama Stack conversation
parameter (which always reloads full history), preserving conversation_id and
the full item history. Record the redesign and the four affected endpoints
(query, streaming_query, A2A, /v1/responses) in a new Changelog section.

* LCORE-1572: fix needs_compaction_path docstring (pydocstyle D400)

* LCORE-1572: build compacted input as typed messages (silence Pydantic warning)

The explicit compacted input was assembled as plain dicts, which produced
PydanticSerializationUnexpectedValue warnings when ResponsesApiParams was
dumped (its input field is typed ResponseInput). Build the summary, recent
verbatim, and query items as typed OpenAIResponseMessage objects instead.

Verified end-to-end against a live stack: the serializer warning is gone and
compaction still triggers, preserves conversation identity, and recalls
earlier context correctly.

* LCORE-1572: raise instead of assert on the drained compaction result

apply_compaction_blocking asserted that the generator yielded a result. Under
python -O asserts are stripped, so the guard would vanish and a None result
could propagate to callers. Replace it with an explicit None check that raises
RuntimeError.

Clears a GitHub code-scanning (CodeQL) "use of assert" finding. The repository's
Bandit configuration skips B101, so this only surfaced via code scanning, not
the Bandit CI job.

* LCORE-1572: wire persisted recursive fold (R3) via the summary cache

Make the conversation summary cache the preferred source of truth for
compaction summaries and the home of the persisted recursive fold.

- apply_compaction / apply_compaction_blocking gain cache + user_id +
  skip_user_id_check. Summaries are read from the cache (get_summaries) and each
  new chunk is written to it (store_summary); the Llama Stack marker texts remain
  an authoritative fallback when no persisting cache is configured (marker-only
  mode, additive summaries, no fold).
- When the persisted summaries themselves exceed the threshold, they are folded
  via recursively_resummarize and the fold is persisted with replace_summaries,
  so it is computed once and reused rather than recomputed per request (R3).
- configured_conversation_cache() resolves the configured cache (or None) for
  the endpoints.
- Wired into /v1/query, /v1/streaming_query, and /v1/responses. The A2A executor
  stays marker-only: it has no resolved user_id for the (user_id, conversation_id)
  cache key.

Adds 7 unit tests: cache-preferred reads, store-on-compaction, fold trigger and
persistence, no-fold-without-cache, marker fallback, and the cache resolver.

* LCORE-1572: address CodeRabbit review — list-form input tokens + clarity rename

- Count tokens for list-form ResponseInput (e.g. /v1/responses), not only the
  string form, so compaction is not skipped on large item-list inputs that
  could otherwise still hit HTTP 413. Adds _estimate_response_input_tokens and a
  regression test.
- Rename CompactionResult.summarized to compacted: the flag means "served in
  compacted / explicit-input mode" (set whenever the conversation has any
  summary, reused or fresh), not "a summary was created this request". The old
  name caused reviewer confusion about turn-persistence gating, which is correct
  as written.

* LCORE-1572: persist compacted streaming turns with structure (CodeRabbit #4)

In compacted mode the streaming endpoint persisted the completed turn as
flattened strings via append_turn_to_conversation, dropping attachments and
non-text output items, and double-storing for shield-blocked requests. Persist
the structured turn instead:

- Capture the response's structured output items onto TurnSummary.output_items
  (set at response.completed, and to the refusal item on a shield block).
- generate_response now takes original_input and persists via store_compacted_turn
  with the original input plus structured output items, matching the /v1/query
  and A2A paths.
- The shield-blocked branch no longer stores the turn when the conversation
  parameter was omitted (compacted mode); generate_response stores it once with
  the correct original input, avoiding the duplicate refusal turn.

Adds tests for the structured compacted persistence and the shield dedup
(compacted and non-compacted).

* LCORE-1572: do not initialize the conversation cache when compaction is disabled

configured_conversation_cache() is evaluated eagerly as a call argument in the
query endpoint, so it ran on every request and accessed
configuration.conversation_cache unconditionally — forcing the (SQLite) cache to
initialize even when compaction is disabled. On configurations whose cache file
could not be opened that raised and returned HTTP 500, which failed the e2e
suites (where compaction is off). Return None without touching the cache when
compaction is disabled; the cache is only used by compaction on this path.

Adds a regression test.

* LCORE-1572: address CodeRabbit round 2 (compacted-mode persistence edges)

Follow-ups to the streaming-persistence work, all for non-happy-path terminals in
compacted mode (conversation parameter omitted), so the persisted turn uses the
original user input + structured output rather than the explicit rewrite or
flattened strings:

- /v1/responses: shield-blocked turns persist against compacted_original_input,
  not api_params.input (the explicit rewrite).
- streaming: interrupted (CancelledError) turns thread original_input through the
  interrupt callback and persist structured items, fixing the wrong-input storage
  and the cast(str, input) break on list inputs.
- streaming: capture output_items on response.failed / response.incomplete
  terminals too, not only response.completed, so compacted persistence keeps
  partial output.
- TurnSummary.output_items typed as list[OpenAIResponseOutput] instead of list[Any].

Also documents that disabling compaction mid-conversation on an already-compacted
conversation reverts it to full-history replay (unsupported transition); the
enabled flag stays a full off-switch (CodeRabbit E, declined by design).

Adds unit tests for the blocked /responses path, the interrupted compacted path,
and output_items capture on a failed terminal.

* LCORE-1572: document the disable-after-compaction limitation in the spec doc (CodeRabbit E)

* LCORE-1572: document as-built divergences in spec doc (cache source-of-truth, persisted fold)

The spec still described the earlier design (cache as a parallel/best-effort
layer, markers as the summary source). Update Summary storage, Additive
summarization, and Changed request flow to the as-built design, and add a
Changelog entry: the cache is the preferred source of truth for summaries (marker
texts as fallback + audit/boundary), the recursive fold is persisted via
replace_summaries (in-memory fold rejected), A2A is marker-only, and the
enabled flag stays a full off-switch.

* LCORE-1572: fix line-too-long (C0301) in interrupted-turn test docstring

* LCORE-1572: harden disabled-cache regression test to fail on eager cache access (CodeRabbit)

* LCORE-1572: ref-count per-conversation lock + extract apply_compaction helpers (review)

Addresses two inline review nits from tisnik on the LCORE-1572 PR.

Per-conversation lock cleanup (R11):
  Replace the bare ``dict[str, asyncio.Lock]`` registry with a ref-counted
  ``_LockEntry`` and an ``@asynccontextmanager`` helper guarded by a registry
  mutex. Entries are removed once the last waiter exits, so the registry no
  longer grows unbounded with the set of conversation_ids ever seen by the
  process. Adds tests for serialization, deletion-after-last-release,
  entry-kept-while-waiters-queued, and cleanup-on-cancellation.

apply_compaction refactor:
  Extract five helpers — ``_load_compaction_state``, ``_estimate_total_tokens``,
  ``_persist_new_summary_chunk``, ``_maybe_persist_fold``, ``_compacted_result``
  — leaving the orchestrating generator linear and roughly one screen long.
  The state-loading, token-estimation, persistence-side-effects, and result-
  building concerns are now each named and individually testable.

* LCORE-1572: tighten typed-item handling in compaction helpers (review)

Addresses asimurka's review nit about the dual dict-or-model branches in
``_verbatim_input_message`` and the surrounding token-estimator helpers.

Llama Stack's ``client.conversations.items.list`` returns items as typed
Pydantic models (the ``ItemListResponse`` discriminated union). The dict
branches in ``is_message_item``, ``extract_message_text``,
``estimate_conversation_tokens``, ``format_conversation_for_summary`` and
``_verbatim_input_message`` were defensive code for a shape that never
arrives from production code paths — they only kept the dict-using test
fixtures alive.

Drop the dict branches and tighten the docstrings to state the typed-item
contract. Update the compaction test fixtures (``_msg``, ``_marker``) to
return ``OpenAIResponseMessage`` instances instead of dicts. Remove the
token-estimator and compaction tests that explicitly asserted dict-shape
acceptance; replace with single tests verifying that dicts are now ignored.

* LCORE-1572: soften R12 doc on silent /v1/responses compaction (review)

Addresses asimurka's review note: emitting a compaction event on the
``/v1/responses`` endpoint would itself be spec-compliant under the
OpenResponses extension-events convention, so framing silent compaction as a
forced choice for "wire compatibility" overstated the constraint. Reword R12
and the changelog entry to acknowledge the spec-compliant option and to frame
silent as the *initial* choice, kept to preserve drop-in compatibility with
clients written against the upstream OpenAI Responses API; emitting the event
on this endpoint is left open as a follow-up. Lightspeed's own clients can
already use ``/v1/streaming_query`` to receive the event.

Author: max-svistunov

docs/design/conversation-compaction/conversation-compaction.md

@@ -1,5 +1,7 @@
 """Handler for A2A (Agent-to-Agent) protocol endpoints using Responses API."""
 
+# pylint: disable=too-many-lines
+
 import asyncio
 import json
 import uuid
@@ -32,7 +34,7 @@
 from llama_stack_api.openai_responses import (
     OpenAIResponseObjectStream,
 )
-from llama_stack_client import APIConnectionError
+from llama_stack_client import APIConnectionError, AsyncLlamaStackClient
 from starlette.responses import Response, StreamingResponse
 
 from a2a_storage import A2AContextStore, A2AStorageFactory
@@ -45,13 +47,18 @@
 from constants import MEDIA_TYPE_EVENT_STREAM
 from log import get_logger
 from models.api.requests import QueryRequest
+from models.common.responses.types import ResponseInput
 from models.config import Action
+from utils.conversation_compaction import (
+    apply_compaction_blocking,
+    store_compacted_turn,
+)
 from utils.mcp_headers import McpHeaders, mcp_headers_dependency
 from utils.responses import (
     extract_text_from_response_item,
     prepare_responses_params,
 )
-from utils.suid import normalize_conversation_id
+from utils.suid import normalize_conversation_id, to_llama_stack_conversation_id
 from version import __version__
 
 logger = get_logger(__name__)
@@ -336,6 +343,19 @@ async def _process_task_streaming(  # pylint: disable=too-many-locals
                 store=True,
                 request_headers=self.request_headers,
             )
+            # Compact the conversation if it is approaching the context window
+            # limit. A2A is not a browser SSE stream, so no progress event is
+            # emitted; the blocking variant summarizes inline before the call.
+            # No conversation cache is passed: the A2A executor has no resolved
+            # user_id for the (user_id, conversation_id) cache key, so A2A runs
+            # in marker-only mode (additive summaries, no persisted fold).
+            compaction = await apply_compaction_blocking(
+                client,
+                responses_params,
+                configuration.inference,
+                configuration.compaction,
+            )
+            responses_params = compaction.params
             # Stream response from LLM using the Responses API
             stream = await client.responses.create(**responses_params.model_dump())
         except APIConnectionError as e:
@@ -392,9 +412,16 @@ async def _process_task_streaming(  # pylint: disable=too-many-locals
             )
         )
 
-        # Process stream using generator and aggregator pattern
+        # Process stream using generator and aggregator pattern. In compacted
+        # mode the conversation parameter is not sent, so the turn is stored
+        # explicitly once the response completes (see _convert_stream_to_events).
         async for a2a_event in self._convert_stream_to_events(
-            stream, task_id, context_id, conversation_id
+            stream,
+            task_id,
+            context_id,
+            conversation_id,
+            client,
+            compaction.original_input if compaction.compacted else None,
         ):
             aggregator.process_event(a2a_event)
             await event_queue.enqueue_event(a2a_event)
@@ -414,12 +441,14 @@ async def _process_task_streaming(  # pylint: disable=too-many-locals
                 final=True,
             )
 
-    async def _convert_stream_to_events(  # pylint: disable=too-many-branches,too-many-locals
+    async def _convert_stream_to_events(  # pylint: disable=too-many-branches,too-many-locals,too-many-arguments,too-many-positional-arguments
         self,
         stream: AsyncIterator[OpenAIResponseObjectStream],
         task_id: str,
         context_id: str,
         conversation_id: Optional[str],
+        client: Optional[AsyncLlamaStackClient] = None,
+        compacted_original_input: Optional[ResponseInput] = None,
     ) -> AsyncIterator[Any]:
         """Convert Responses API stream chunks to A2A events.
 
@@ -508,6 +537,20 @@ async def _convert_stream_to_events(  # pylint: disable=too-many-branches,too-ma
 
                 if response_obj:
                     output = getattr(response_obj, "output", [])
+                    # In compacted mode the conversation parameter was not sent,
+                    # so persist this turn ourselves to keep the recent-turn
+                    # buffer and audit history intact for the next request.
+                    if (
+                        compacted_original_input is not None
+                        and client is not None
+                        and conversation_id
+                    ):
+                        await store_compacted_turn(
+                            client,
+                            to_llama_stack_conversation_id(conversation_id),
+                            compacted_original_input,
+                            output,
+                        )
                     a2a_parts = _convert_responses_content_to_a2a_parts(output)
                     if not a2a_parts and final_text:
                         a2a_parts = [Part(root=TextPart(text=final_text))]

src/app/endpoints/a2a.py

@@ -39,8 +39,14 @@
 from models.api.responses.successful import QueryResponse
 from models.common.moderation import ShieldModerationResult
 from models.common.responses.responses_api_params import ResponsesApiParams
+from models.common.responses.types import ResponseInput
 from models.common.turn_summary import TurnSummary
 from models.config import Action
+from utils.conversation_compaction import (
+    apply_compaction_blocking,
+    configured_conversation_cache,
+    store_compacted_turn,
+)
 from utils.conversations import append_turn_items_to_conversation
 from utils.endpoints import (
     check_configuration_loaded,
@@ -196,6 +202,20 @@ async def query_endpoint_handler(
         inline_rag_context=inline_rag_context.context_text,
     )
 
+    # Compact the conversation if it is approaching the context window limit.
+    # When compaction is active, params carry explicit input and the
+    # conversation parameter is dropped (lightspeed-stack owns the context).
+    compaction = await apply_compaction_blocking(
+        client,
+        responses_params,
+        configuration.inference,
+        configuration.compaction,
+        cache=configured_conversation_cache(),
+        user_id=user_id,
+        skip_user_id_check=_skip_userid_check,
+    )
+    responses_params = compaction.params
+
     # Handle Azure token refresh if needed
     if (
         responses_params.model.startswith("azure")
@@ -207,7 +227,11 @@ async def query_endpoint_handler(
 
     # Retrieve response using Responses API
     turn_summary = await retrieve_response(
-        client, responses_params, moderation_result, endpoint_path
+        client,
+        responses_params,
+        moderation_result,
+        endpoint_path,
+        original_input=compaction.original_input if compaction.compacted else None,
     )
 
     if moderation_result.decision == "passed":
@@ -282,6 +306,7 @@ async def retrieve_response(
     responses_params: ResponsesApiParams,
     moderation_result: ShieldModerationResult,
     endpoint_path: str = "",
+    original_input: Optional[ResponseInput] = None,
 ) -> TurnSummary:
     """
     Retrieve response from LLMs and agents.
@@ -294,17 +319,28 @@ async def retrieve_response(
         client: The AsyncLlamaStackClient to use for the request.
         responses_params: The Responses API parameters.
         moderation_result: The moderation result.
+        endpoint_path: The request path, for metrics/telemetry.
+        original_input: Set only in compacted mode (LCORE-1572). It is the new
+            user query before the explicit-input rewrite. When provided, the
+            turn is appended to the conversation here, because the conversation
+            parameter is no longer passed to Llama Stack and so the turn is not
+            stored automatically.
 
     Returns:
     -------
         TurnSummary: Summary of the LLM response content
     """
     response: Optional[OpenAIResponseObject] = None
+    # In compacted mode, the new turn must be stored against the original user
+    # query, not the explicit summaries-plus-recent input we send to inference.
+    turn_input = (
+        original_input if original_input is not None else responses_params.input
+    )
     if moderation_result.decision == "blocked":
         await append_turn_items_to_conversation(
             client,
             responses_params.conversation,
-            responses_params.input,
+            turn_input,
             [moderation_result.refusal_response],
         )
         return TurnSummary(
@@ -331,6 +367,16 @@ async def retrieve_response(
         error_response = handle_known_apistatus_errors(e, responses_params.model)
         raise HTTPException(**error_response.model_dump()) from e
 
+    # In compacted mode, store the completed turn ourselves (the conversation
+    # parameter was not sent, so Llama Stack did not persist it).
+    if original_input is not None:
+        await store_compacted_turn(
+            client,
+            responses_params.conversation,
+            original_input,
+            response.output,
+        )
+
     vector_store_ids = extract_vector_store_ids_from_tools(responses_params.tools)
     rag_id_mapping = configuration.rag_id_mapping
     return build_turn_summary(

src/app/endpoints/query.py

@@ -62,8 +62,13 @@
 from models.common.moderation import ShieldModerationBlocked
 from models.common.responses.contexts import ResponsesContext
 from models.common.responses.responses_api_params import ResponsesApiParams
+from models.common.responses.types import ResponseInput
 from models.common.turn_summary import TurnSummary
 from models.config import Action
+from utils.conversation_compaction import (
+    apply_compaction_blocking,
+    configured_conversation_cache,
+)
 from utils.conversations import append_turn_items_to_conversation
 from utils.endpoints import (
     check_configuration_loaded,
@@ -225,10 +230,18 @@ async def _persist_blocked_response_turn(
     """
     if api_params.store:
         moderation_result = cast(ShieldModerationBlocked, context.moderation_result)
+        # In compacted mode the conversation parameter was dropped and
+        # api_params.input is the explicit-input rewrite, so persist the turn
+        # against the original user input instead (LCORE-1572).
+        user_input = (
+            context.compacted_original_input
+            if context.compacted_original_input is not None
+            else api_params.input
+        )
         await append_turn_items_to_conversation(
             client=context.client,
             conversation_id=api_params.conversation,
-            user_input=api_params.input,
+            user_input=user_input,
             llm_output=[moderation_result.refusal_response],
         )
 
@@ -238,14 +251,30 @@ async def _append_previous_response_turn(
     context: ResponsesContext,
     output: Sequence[OpenAIResponseOutput],
 ) -> None:
-    """Append response output when continuing from a previous response id.
+    """Append the completed turn when Llama Stack did not store it automatically.
+
+    Llama Stack stores the turn itself only when the conversation parameter is
+    sent. Two cases bypass that and require an explicit append: continuing from
+    a ``previous_response_id``, and conversation compaction (LCORE-1572), where
+    the conversation parameter is dropped in favor of explicit input. In the
+    compaction case the turn is stored against the original user input (before
+    the explicit-input rewrite), carried on the context.
 
     Args:
         api_params: Responses API parameters containing conversation details.
         context: Request-scoped Responses API context.
         output: Final output items from the Responses API object.
     """
-    if api_params.store and api_params.previous_response_id:
+    if not api_params.store:
+        return
+    if context.compacted_original_input is not None:
+        await append_turn_items_to_conversation(
+            context.client,
+            api_params.conversation,
+            context.compacted_original_input,
+            output,
+        )
+    elif api_params.previous_response_id:
         await append_turn_items_to_conversation(
             context.client,
             api_params.conversation,
@@ -337,7 +366,7 @@ async def responses_endpoint_handler(
     check_configuration_loaded(configuration)
     started_at = datetime.now(UTC)
     rh_identity_context = get_rh_identity_context(request)
-    user_id, _, _, token = auth
+    user_id, _, skip_userid_check, token = auth
 
     await check_mcp_auth(configuration, mcp_headers, token, request.headers)
 
@@ -436,6 +465,32 @@ async def responses_endpoint_handler(
         )
 
     api_params = ResponsesApiParams.model_validate(updated_request.model_dump())
+
+    # Compact the conversation if it is approaching the context window limit.
+    # /v1/responses is OpenAI-compatible, so compaction is silent (no custom SSE
+    # event): summarization happens before the response is created, and the turn
+    # is appended explicitly afterward (the conversation parameter is dropped).
+    # Only stateful single-conversation requests are eligible.
+    compacted_original_input: Optional[ResponseInput] = None
+    if (
+        configuration.compaction.enabled
+        and api_params.store
+        and api_params.conversation
+        and not api_params.previous_response_id
+    ):
+        compaction = await apply_compaction_blocking(
+            client,
+            api_params,
+            configuration.inference,
+            configuration.compaction,
+            cache=configured_conversation_cache(),
+            user_id=user_id,
+            skip_user_id_check=skip_userid_check,
+        )
+        api_params = compaction.params
+        if compaction.compacted:
+            compacted_original_input = compaction.original_input
+
     context = ResponsesContext(
         client=client,
         auth=auth,
@@ -449,6 +504,7 @@ async def responses_endpoint_handler(
         user_agent=_get_user_agent(request),
         endpoint_path=endpoint_path,
         generate_topic_summary=updated_request.generate_topic_summary,
+        compacted_original_input=compacted_original_input,
     )
     response_handler = (
         handle_streaming_response