diff --git a/gateway/platforms/api_server.py b/gateway/platforms/api_server.py index bfe3d3558648..ce744d455686 100644 --- a/gateway/platforms/api_server.py +++ b/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) diff --git a/hermes_cli/config.py b/hermes_cli/config.py index 060373555f0b..2f3ba635af04 100644 --- a/hermes_cli/config.py +++ b/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). diff --git a/tests/gateway/test_api_server.py b/tests/gateway/test_api_server.py index aa48e5323daa..608be79ab059 100644 --- a/tests/gateway/test_api_server.py +++ b/tests/gateway/test_api_server.py @@ -2046,6 +2046,273 @@ async def test_previous_response_id_stores_full_agent_transcript_once(self, adap assert stored_history.count(first_history[0]) == 1 assert stored_history.count({"role": "user", "content": "Now add 1 more"}) == 1 + @pytest.mark.asyncio + async def test_previous_response_id_stores_compressed_transcript_directly(self, adapter): + """After compression, stored history is the compressed transcript, not prior + compressed.""" + prior_history = [ + {"role": "user", "content": "What is 1+1?"}, + {"role": "assistant", "content": "2"}, + ] * 10 # 20 messages — enough to simulate a long conversation + adapter._response_store.put( + "resp_prev", + { + "response": {"id": "resp_prev", "status": "completed"}, + "conversation_history": list(prior_history), + "session_id": "api-test-session", + }, + ) + + compressed_history = [ + # Compressed transcript starts with summary, NOT with prior[0] + {"role": "user", "content": "[Compressed summary of earlier conversation]"}, + {"role": "user", "content": "Now add 1 more"}, + {"role": "assistant", "content": "3"}, + ] + + app = _create_app(adapter) + async with TestClient(TestServer(app)) as cli: + with patch.object(adapter, "_run_agent", new_callable=AsyncMock) as mock_run: + mock_run.return_value = ( + { + "final_response": "3", + "messages": list(compressed_history), + "_compressed": True, + "api_calls": 1, + }, + {"input_tokens": 0, "output_tokens": 0, "total_tokens": 0}, + ) + resp = await cli.post( + "/v1/responses", + json={ + "model": "hermes-agent", + "input": "Now add 1 more", + "previous_response_id": "resp_prev", + }, + ) + assert resp.status == 200 + data = await resp.json() + + stored = adapter._response_store.get(data["id"]) + stored_history = stored["conversation_history"] + # Must NOT contain the original prior_history messages + for msg in prior_history: + assert msg not in stored_history, ( + f"Prior history message leaked into stored compressed transcript: {msg}" + ) + # Must contain the compressed transcript + assert stored_history == compressed_history + + @pytest.mark.asyncio + async def test_rotation_compression_exercises_detection_and_persists_rotated_session_id( + self, adapter + ): + """Fake-agent rotation: _run_agent detects session_id change, sets + _compressed, and the Responses handler persists the rotated session_id + so subsequent previous_response_id chaining loads the compressed child.""" + prior_history = [ + {"role": "user", "content": "What is 1+1?"}, + {"role": "assistant", "content": "2"}, + ] * 10 + + adapter._response_store.put( + "resp_prev_rot", + { + "response": {"id": "resp_prev_rot", "status": "completed"}, + "conversation_history": list(prior_history), + "session_id": "api-test-session", + }, + ) + + compressed_history = [ + {"role": "user", "content": "[Compressed summary]"}, + {"role": "user", "content": "Now add 1 more"}, + {"role": "assistant", "content": "3"}, + ] + + # Fake agent whose session_id was rotated by compression + mock_agent = MagicMock() + mock_agent.session_id = "rotated-child-session" + mock_agent._last_compaction_in_place = False + mock_agent.session_prompt_tokens = 0 + mock_agent.session_completion_tokens = 0 + mock_agent.session_total_tokens = 0 + mock_agent.run_conversation.return_value = { + "final_response": "3", + "messages": list(compressed_history), + "api_calls": 1, + } + + app = _create_app(adapter) + async with TestClient(TestServer(app)) as cli: + with patch.object(adapter, "_create_agent", return_value=mock_agent): + resp = await cli.post( + "/v1/responses", + json={ + "model": "hermes-agent", + "input": "Now add 1 more", + "previous_response_id": "resp_prev_rot", + }, + ) + assert resp.status == 200 + data = await resp.json() + + # The detection path in _run_agent must have set _compressed + # _run_agent mutates the result dict in place -- verify via stored data + stored = adapter._response_store.get(data["id"]) + assert stored is not None + + # Stored history must be the compressed transcript, not prior + compressed + stored_history = stored["conversation_history"] + for msg in prior_history: + assert msg not in stored_history + assert stored_history == compressed_history + + # Stored session_id must be the rotated child, not the original + assert stored["session_id"] == "rotated-child-session" + + # Response header must also reflect the rotated session + assert resp.headers.get("X-Hermes-Session-Id") == "rotated-child-session" + + @pytest.mark.asyncio + async def test_inplace_compression_exercises_detection_and_persists_compressed_history( + self, adapter + ): + """Fake-agent in-place: _run_agent detects _last_compaction_in_place, + sets _compressed, and the Responses handler persists compressed history + WITHOUT rotating the session_id.""" + prior_history = [ + {"role": "user", "content": "What is 1+1?"}, + {"role": "assistant", "content": "2"}, + ] * 10 + + adapter._response_store.put( + "resp_prev_inplace", + { + "response": {"id": "resp_prev_inplace", "status": "completed"}, + "conversation_history": list(prior_history), + "session_id": "api-test-session", + }, + ) + + compressed_history = [ + {"role": "user", "content": "[Compressed summary in-place]"}, + {"role": "user", "content": "Continue"}, + {"role": "assistant", "content": "42"}, + ] + + # Fake agent: in-place compaction, session_id unchanged + mock_agent = MagicMock() + mock_agent.session_id = "api-test-session" # same as input -- no rotation + mock_agent._last_compaction_in_place = True + mock_agent.session_prompt_tokens = 0 + mock_agent.session_completion_tokens = 0 + mock_agent.session_total_tokens = 0 + mock_agent.run_conversation.return_value = { + "final_response": "42", + "messages": list(compressed_history), + "api_calls": 1, + } + + app = _create_app(adapter) + async with TestClient(TestServer(app)) as cli: + with patch.object(adapter, "_create_agent", return_value=mock_agent): + resp = await cli.post( + "/v1/responses", + json={ + "model": "hermes-agent", + "input": "Continue", + "previous_response_id": "resp_prev_inplace", + }, + ) + assert resp.status == 200 + data = await resp.json() + + stored = adapter._response_store.get(data["id"]) + assert stored is not None + + # Stored history must be the compressed transcript + stored_history = stored["conversation_history"] + for msg in prior_history: + assert msg not in stored_history + assert stored_history == compressed_history + + # Session_id must NOT change for in-place compaction + assert stored["session_id"] == "api-test-session" + assert resp.headers.get("X-Hermes-Session-Id") == "api-test-session" + + @pytest.mark.asyncio + async def test_chained_rotation_propagates_effective_session_id(self, adapter): + """Two-request chain: first request triggers rotation, second request + loads history using the rotated session_id stored by the first.""" + # First request -- no previous_response_id, establishes the conversation + mock_agent_1 = MagicMock() + mock_agent_1.session_id = "child-session-after-rotation" + mock_agent_1._last_compaction_in_place = False + mock_agent_1.session_prompt_tokens = 0 + mock_agent_1.session_completion_tokens = 0 + mock_agent_1.session_total_tokens = 0 + compressed_msg_1 = [ + {"role": "user", "content": "[Summary of turn 1]"}, + {"role": "assistant", "content": "Hello back"}, + ] + mock_agent_1.run_conversation.return_value = { + "final_response": "Hello back", + "messages": list(compressed_msg_1), + "api_calls": 1, + } + + app = _create_app(adapter) + async with TestClient(TestServer(app)) as cli: + with patch.object(adapter, "_create_agent", return_value=mock_agent_1): + resp1 = await cli.post( + "/v1/responses", + json={"model": "hermes-agent", "input": "Hello"}, + ) + assert resp1.status == 200 + data1 = await resp1.json() + response_id_1 = data1["id"] + + # Verify the rotated session_id was persisted + stored_1 = adapter._response_store.get(response_id_1) + assert stored_1["session_id"] == "child-session-after-rotation" + assert stored_1["conversation_history"] == compressed_msg_1 + + # Second request -- chains via previous_response_id + # _run_agent is called with the session_id from the stored response + mock_agent_2 = MagicMock() + mock_agent_2.session_id = "child-session-after-rotation" + mock_agent_2._last_compaction_in_place = False + mock_agent_2.session_prompt_tokens = 0 + mock_agent_2.session_completion_tokens = 0 + mock_agent_2.session_total_tokens = 0 + mock_agent_2.run_conversation.return_value = { + "final_response": "Goodbye", + "messages": [ + {"role": "user", "content": "[Summary of turn 1]"}, + {"role": "assistant", "content": "Hello back"}, + {"role": "user", "content": "Goodbye"}, + {"role": "assistant", "content": "See you!"}, + ], + "api_calls": 1, + } + + with patch.object(adapter, "_create_agent", return_value=mock_agent_2): + resp2 = await cli.post( + "/v1/responses", + json={ + "model": "hermes-agent", + "input": "Goodbye", + "previous_response_id": response_id_1, + }, + ) + assert resp2.status == 200 + + # The second _run_agent call must receive the rotated session_id + # (from the stored response), not the original request session_id + call_kwargs = mock_agent_2.run_conversation.call_args.kwargs + # conversation_history loaded from store should be the compressed transcript + assert call_kwargs["conversation_history"] == compressed_msg_1 + @pytest.mark.asyncio async def test_previous_response_id_outputs_only_current_turn_items(self, adapter): """Response output must not replay previous tool artifacts."""