Gateway compression and hygiene transcript rotation/persistence issues

Fixes on record

Merged to main. A stamp counts the issues a fix is linked to.

• #48584 fix(compression): preserve messages across session rotation (flush-before-rotate + /compress rewrite guard)

  merged to main

  files: agent/conversation_compression.py, cli.py, gateway/slash_commands.py

  Show the diff

  • agent/conversation_compression.py

    @@ -512,6 +512,16 @@ def _release_lock() -> None:
                 old_title = agent._session_db.get_session_title(agent.session_id)
                 # Trigger memory extraction on the old session before it rotates.
                 agent.commit_memory_session(messages)
    +            # Flush any un-persisted messages from the current turn to the
    +            # old session *before* rotating.  compress_context() can be
    +            # called mid-turn (auto-compress when context exceeds threshold)
    +            # at a point when _flush_messages_to_session_db() has not yet
    +            # run.  Without this, messages generated during the current turn
    +            # are silently lost on session rotation (#47202).
    +            try:
    +                agent._flush_messages_to_session_db(messages)
    +            except Exception:
    +                pass  # best-effort — don't block compression on a flush error
                 agent._session_db.end_session(agent.session_id, "compression")
                 old_session_id = agent.session_id
                 agent.session_id = f"{datetime.now().strftime('%Y%m%d_%H%M%S')}_{uuid.uuid4().hex[:6]}"

  • cli.py

    @@ -5957,6 +5957,18 @@ def new_session(self, silent=False, title=None):
     
             old_session_id = self.session_id
             if self._session_db and old_session_id:
    +            # Flush any un-persisted messages from the current turn to the
    +            # old session *before* rotating.  /new can be called mid-turn
    +            # when _flush_messages_to_session_db() has not yet run — without
    +            # this, messages generated during the current turn are silently
    +            # lost on session rotation (#47202).
    +            if self.agent:
    +                try:
    +                    self.agent._flush_messages_to_session_db(
    +                        self.conversation_history
    +                    )
    +                except Exception:
    +                    pass  # best-effort
                 try:
                     self._session_db.end_session(old_session_id, "new_session")
                 except Exception:

  • gateway/slash_commands.py

    @@ -2588,14 +2588,29 @@ async def _handle_compress_command(self, event: MessageEvent) -> str:
                     # session_id for the continuation.  Write the compressed messages
                     # into the NEW session so the original history stays searchable.
                     new_session_id = tmp_agent.session_id
    -                if new_session_id != session_entry.session_id:
    +                rotated = new_session_id != session_entry.session_id
    +                if rotated:
                         session_entry.session_id = new_session_id
                         self.session_store._save()
                         self._sync_telegram_topic_binding(
                             source, session_entry, reason="compress-command",
                         )
     
    -                self.session_store.rewrite_transcript(new_session_id, compressed)
    +                # Only rewrite the transcript when rotation actually produced a
    +                # NEW session id. If _compress_context could not rotate (e.g.
    +                # _session_db unavailable, or the DB split raised), session_id
    +                # is unchanged and rewrite_transcript() would DELETE the
    +                # original messages and replace them with only the compressed
    +                # summary — permanent data loss (#44794, #39704). In that case
    +                # leave the original transcript intact.
    +                if rotated:
    +                    self.session_store.rewrite_transcript(new_session_id, compressed)
    +                else:
    +                    logger.warning(
    +                        "Manual /compress: session rotation did not occur "
    +                        "(session_id unchanged) — preserving original transcript "
    +                        "instead of overwriting it (#44794)."
    +                    )
                     # Reset stored token count — transcript changed, old value is stale
                     self.session_store.update_session(
                         session_entry.session_key, last_prompt_tokens=0

• #49925 fix(gateway): preserve transcript when hygiene auto-compress can't rotate (#21301)

  merged to main

  files: gateway/run.py, tests/gateway/test_session_hygiene.py

• #52628 fix(gateway): pass session_db to compress temp agents so persistence works

  merged to main

  files: gateway/run.py, gateway/slash_commands.py, tests/gateway/test_compress_command.py, tests/gateway/test_compress_focus.py, tests/gateway/test_compress_plugin_engine.py

  Show the diff

  • gateway/run.py

    @@ -9550,14 +9550,15 @@ async def _handle_message_with_agent(self, event, source, _quick_key: str, run_g
                                         skip_memory=True,
                                         enabled_toolsets=["memory"],
                                         session_id=session_entry.session_id,
    +                                    session_db=self._session_db,
                                     )
                                     try:
                                         # The hygiene agent rotates the session
                                         # forward to a continuation id that becomes
                                         # the gateway session's live row. It must
    -                                    # never finalize on close() (today it has no
    -                                    # session_db so close() no-ops, but this
    -                                    # guards a future where one is wired in).
    +                                    # never finalize on close() — close() would
    +                                    # end the newly rotated session the gateway
    +                                    # entry now points at.
                                         _hyg_agent._end_session_on_close = False
                                         _hyg_agent._print_fn = lambda *a, **kw: None
     

  • gateway/slash_commands.py

    @@ -2818,9 +2818,14 @@ async def _handle_compress_command(self, event: MessageEvent) -> str:
                     skip_memory=True,
                     enabled_toolsets=["memory"],
                     session_id=session_entry.session_id,
    +                session_db=self._session_db,
                 )
                 try:
                     tmp_agent._print_fn = lambda *a, **kw: None
    +                # Prevent close() from ending the newly rotated session —
    +                # the gateway session entry now points at the new id and
    +                # must remain open for the next user turn.
    +                tmp_agent._end_session_on_close = False
     
                     # Estimate with system prompt + tool schemas included so the
                     # figure reflects real request pressure, not a transcript-only

  • tests/gateway/test_compress_command.py

    @@ -54,6 +54,7 @@ def _make_runner(history: list[dict[str, str]]):
         runner.session_store.rewrite_transcript = MagicMock()
         runner.session_store.update_session = MagicMock()
         runner.session_store._save = MagicMock()
    +    runner._session_db = None
         return runner
     
     
    @@ -247,3 +248,60 @@ def _estimate(messages, **_kwargs):
         assert "intact" in result
         agent_instance.shutdown_memory_provider.assert_called_once()
         agent_instance.close.assert_called_once()
    +
    +
    [redacted:email]
    +async def test_compress_command_passes_session_db_and_persists_rotated_session():
    +    """session_db must be wired into the /compress temp agent so that
    +    _compress_context can actually rotate the session and persist the
    +    compressed transcript — without it compression is a silent no-op."""
    +    history = _make_history()
    +    compressed = [
    +        history[0],
    +        {"role": "assistant", "content": "compressed summary"},
    +        history[-1],
    +    ]
    +    runner = _make_runner(history)
    +    runner._session_db = object()
    +    agent_instance = MagicMock()
    +    agent_instance.shutdown_memory_provider = MagicMock()
    +    agent_instance.close = MagicMock()
    +    agent_instance._cached_system_prompt = ""
    +    agent_instance.tools = None
    +    agent_instance.context_compressor.has_content_to_compress.return_value = True
    +    agent_instance.compression_in_place = False
    +    agent_instance.session_id = "sess-1"
    +
    +    def _compress(messages, *_args, **_kwargs):
    +        agent_instance.session_id = "sess-2"
    +        return compressed, ""
    +
    +    agent_instance._compress_context.side_effect = _compress
    +
    +    def _estimate(messages, **_kwargs):
    +        if messages == history:
    +            return 100
    +        if messages == compressed:
    +            return 60
    +        raise AssertionError(f"unexpected transcript: {messages!r}")
    +
    +    with (
    +        patch("gateway.run._resolve_runtime_agent_kwargs", return_value={"api_key": "***"}),
    +        patch("gateway.run._resolve_gateway_model", return_value="test-model"),
    +        patch("run_agent.AIAgent", return_value=agent_instance) as mock_agent_cls,
    +        patch("agent.model_metadata.estimate_request_tokens_rough", side_effect=_estimate),
    +    ):
    +        result = await runner._handle_compress_command(_make_event())
    +
    +    assert "Compressed:" in result
    +    mock_agent_cls.assert_called_once()
    +    assert mock_agent_cls.call_args.kwargs["session_db"] is runner._session_db
    +    runner.session_store._save.assert_called_once()
    +    runner.session_store.rewrite_transcript.assert_called_once_with(
    +        "sess-2", compressed
    +    )
    +    runner.session_store.update_session.assert_called_once_with(
    +        build_session_key(_make_source()), last_prompt_tokens=0
    +    )
    +    agent_instance.shutdown_memory_provider.assert_called_once()
    +    agent_instance.close.assert_called_once()

  • tests/gateway/test_compress_focus.py

    @@ -54,6 +54,7 @@ def _make_runner(history: list[dict[str, str]]):
         runner.session_store.rewrite_transcript = MagicMock()
         runner.session_store.update_session = MagicMock()
         runner.session_store._save = MagicMock()
    +    runner._session_db = None
         return runner
     
     

  • tests/gateway/test_compress_plugin_engine.py

    @@ -101,6 +101,7 @@ def _make_runner(history: list[dict[str, str]]):
         runner.session_store.rewrite_transcript = MagicMock()
         runner.session_store.update_session = MagicMock()
         runner.session_store._save = MagicMock()
    +    runner._session_db = None
         return runner
     
     

• #52658 feat(compression): flip in_place default to True + guard fix (#38763) [salvage #51959]

  merged to main

  files: gateway/run.py, gateway/slash_commands.py, hermes_cli/config.py, scripts/release.py, tests/agent/test_compression_concurrent_fork.py, tests/agent/test_compression_logging_session_context.py, tests/agent/test_compression_rotation_state.py, tests/gateway/test_compression_concurrent_sessions.py, tests/gateway/test_session_hygiene.py, tests/run_agent/test_compression_boundary_hook.py, tests/run_agent/test_in_place_compaction.py

• #60981 fix(gateway): compact hygiene transcripts in place (#60947)

  merged to main

  files: gateway/run.py, scripts/release.py, tests/gateway/test_session_hygiene.py