fix(gateway): Responses API stores bloated context after compression

gateway/platforms/api_server.py

@@ -3096,6 +3096,7 @@ async def _write_sse_responses(
         store: bool,
         session_id: str,
         gateway_session_key: Optional[str] = None,
+        history_from_store: bool = False,
     ) -> "web.StreamResponse":
         """Write an SSE stream for POST /v1/responses (OpenAI Responses API).
 
@@ -3606,6 +3607,35 @@ async def _flush_batch() -> None:
                     result,
                     final_response_text,
                 )
+                # Persist compressed messages when compression occurred and
+                # history was loaded from response_store (same logic as the
+                # non-streaming path).
+                if history_from_store and isinstance(result, dict):
+                    _result_sid = result.get("session_id")
+                    _did_compress = bool(result.get("_compressed"))
+                    _rotated = bool(_result_sid and _result_sid != session_id)
+                    if _did_compress or _rotated:
+                        try:
+                            from hermes_cli.config import load_config
+                            _comp_cfg = load_config().get("compression", {})
+                            if _comp_cfg.get("persist_in_response_store", True):
+                                _agent_messages = result.get("messages")
+                                if isinstance(_agent_messages, list) and _agent_messages:
+                                    _mode = "in-place" if _did_compress and not _rotated else "rotation"
+                                    logger.info(
+                                        "Compression persisted in response_store (streaming, %s): "
+                                        "%d messages (was %d before compression)",
+                                        _mode,
+                                        len(_agent_messages),
+                                        len(conversation_history) + 1,
+                                    )
+                                    full_history = list(_agent_messages)
+                        except Exception as e:
+                            logger.warning(
+                                "Failed to persist compressed response_store snapshot "
+                                "(streaming): %s",
+                                e,
+                            )
                 _persist_response_snapshot(
                     completed_env,
                     conversation_history_snapshot=full_history,
@@ -3761,12 +3791,14 @@ async def _handle_responses(self, request: "web.Request") -> "web.Response":
                 logger.debug("Both conversation_history and previous_response_id provided; using conversation_history")
 
         stored_session_id = None
+        _history_from_store = False
         if not conversation_history and previous_response_id:
             stored = self._response_store.get(previous_response_id)
             if stored is None:
                 return web.json_response(_openai_error(f"Previous response not found: {previous_response_id}"), status=404)
             conversation_history = list(stored.get("conversation_history", []))
             stored_session_id = stored.get("session_id")
+            _history_from_store = True
             # If no instructions provided, carry forward from previous
             if instructions is None:
                 instructions = stored.get("instructions")
@@ -3869,6 +3901,7 @@ def _on_tool_complete(tool_call_id, function_name, function_args, function_resul
                 store=store,
                 session_id=session_id,
                 gateway_session_key=gateway_session_key,
+                history_from_store=_history_from_store,
             )
 
         async def _compute_response():
@@ -3921,6 +3954,40 @@ async def _compute_response():
             final_response,
         )
 
+        # If compression occurred during the agent run and the history was
+        # loaded from the response_store (not supplied explicitly by the
+        # client), persist the compressed messages instead of the original
+        # uncompressed chain.  This prevents repeated re-compression on
+        # subsequent requests in the same conversation.
+        _effective_session_id = session_id
+        if _history_from_store and isinstance(result, dict):
+            _result_sid = result.get("session_id")
+            _did_compress = bool(result.get("_compressed"))
+            _rotated = bool(_result_sid and _result_sid != session_id)
+            if _did_compress or _rotated:
+                try:
+                    from hermes_cli.config import load_config
+                    _comp_cfg = load_config().get("compression", {})
+                    if _comp_cfg.get("persist_in_response_store", True):
+                        _agent_messages = result.get("messages")
+                        if isinstance(_agent_messages, list) and _agent_messages:
+                            _mode = "in-place" if _did_compress and not _rotated else "rotation"
+                            logger.info(
+                                "Compression persisted in response_store (%s): "
+                                "%d messages (was %d before compression)",
+                                _mode,
+                                len(_agent_messages),
+                                len(conversation_history) + 1,
+                            )
+                            full_history = list(_agent_messages)
+                            if _rotated and _result_sid:
+                                _effective_session_id = _result_sid
+                except Exception as e:
+                    logger.warning(
+                        "Failed to persist compressed response_store snapshot: %s",
+                        e,
+                    )
+
         # Build output items from the current turn only.  AIAgent returns a
         # full transcript in result["messages"], while older/mocked paths may
         # return only the current turn suffix.
@@ -3951,14 +4018,14 @@ async def _compute_response():
                 "response": response_data,
                 "conversation_history": full_history,
                 "instructions": instructions,
-                "session_id": session_id,
+                "session_id": _effective_session_id,
             })
             # Update conversation mapping so the next request with the same
             # conversation name automatically chains to this response
             if conversation:
                 self._response_store.set_conversation(conversation, response_id)
 
-        response_headers = {"X-Hermes-Session-Id": session_id}
+        response_headers = {"X-Hermes-Session-Id": _effective_session_id}
         if gateway_session_key:
             response_headers["X-Hermes-Session-Key"] = gateway_session_key
         return web.json_response(response_data, headers=response_headers)
@@ -4315,7 +4382,16 @@ def _build_response_conversation_history(
         result: Dict[str, Any],
         final_response: Any,
     ) -> List[Dict[str, Any]]:
-        """Build the stored Responses transcript without duplicating history."""
+        """Build the stored Responses transcript without duplicating history.
+
+        When context compression occurs during a turn the agent returns a
+        compressed full transcript in ``result["messages"]`` (starting with a
+        summary) and sets ``result["_compressed"] = True``.  Because the
+        compressed transcript does not share the input ``conversation_history``
+        prefix, the normal turn-start detection fails and old code would
+        concatenate the uncompressed history on front, bloating the stored
+        context and re-triggering compression on every subsequent request.
+        """
         prior = list(conversation_history)
         current_user = {"role": "user", "content": user_message}
         agent_messages = result.get("messages") if isinstance(result, dict) else None
@@ -4329,6 +4405,16 @@ def _build_response_conversation_history(
             if turn_start:
                 return list(agent_messages)
 
+            # turn_start == 0: agent_messages does not start with prior.
+            # This can happen because compression rewrote the transcript
+            # (summary prefix replaces original history), OR because
+            # agent_messages only carries the current turn without prior.
+            # The ``_compressed`` flag (set by _run_agent after compaction)
+            # distinguishes — skip the concatenation and use the compressed
+            # transcript directly.
+            if result.get("_compressed"):
+                return list(agent_messages)
+
             full_history = prior
             full_history.append(current_user)
             full_history.extend(agent_messages)
@@ -4589,6 +4675,18 @@ def _run():
                     _eff_sid = getattr(agent, "session_id", session_id)
                     if isinstance(_eff_sid, str) and _eff_sid:
                         result["session_id"] = _eff_sid
+                    # Signal whether context compression occurred during this turn
+                    # so _build_response_conversation_history can skip the
+                    # prior-concatenation path and store the compressed transcript
+                    # directly.  Rotation mode changes agent.session_id; in-place
+                    # mode sets _last_compaction_in_place (see #38763).
+                    _compacted_in_place = bool(getattr(agent, "_last_compaction_in_place", False))
+                    _session_rotated = (
+                        isinstance(_eff_sid, str) and isinstance(session_id, str)
+                        and _eff_sid != session_id
+                    )
+                    if _compacted_in_place or _session_rotated:
+                        result["_compressed"] = True
                     return result, usage
                 finally:
                     clear_session_vars(tokens)

hermes_cli/config.py

@@ -1485,6 +1485,16 @@ def _ensure_hermes_home_managed(home: Path):
                                       # session_search and recoverable, not deleted.
                                       # Default False during rollout; will flip on
                                       # after live validation.
+        "persist_in_response_store": True,  # When True, if compression occurs during a
+                                      # /v1/responses request that loads history from the
+                                      # response_store (via previous_response_id or
+                                      # conversation name), the compressed messages are
+                                      # persisted as the stored conversation_history
+                                      # snapshot.  This prevents repeated re-compression
+                                      # on subsequent requests in the same chain.
+                                      # Only applies when the client does NOT supply an
+                                      # explicit conversation_history array.  Set to False
+                                      # to preserve legacy behavior (store uncompressed).
     },
 
     # Kanban subsystem (orchestrator workers + dispatcher-driven child tasks).