Session transcript persistence and signal-handling gaps

Fixes on record

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

• #46071 fix(agent): persist repaired-turn responses

  merged to main

  files: run_agent.py, tests/run_agent/test_compression_persistence.py, tests/run_agent/test_identity_flush.py

  Show the diff

  • run_agent.py

    @@ -1548,9 +1548,10 @@ def _repair_message_sequence(self, messages: List[Dict]) -> int:
         def _flush_messages_to_session_db(self, messages: List[Dict], conversation_history: List[Dict] = None):
             """Persist any un-flushed messages to the SQLite session store.
     
    -        Uses _last_flushed_db_idx to track which messages have already been
    -        written, so repeated calls (from multiple exit paths) only write
    -        truly new messages — preventing the duplicate-write bug (#860).
    +        Uses per-session message identity tracking so repeated calls (from
    +        multiple exit paths) only write truly new messages — preventing the
    +        duplicate-write bug (#860) without relying on positional slices that
    +        can drift after message-sequence repair.
             """
             if not self._session_db:
                 return
    @@ -1559,14 +1560,41 @@ def _flush_messages_to_session_db(self, messages: List[Dict], conversation_histo
                 # Retry row creation if the earlier attempt failed transiently.
                 if not self._session_db_created:
                     self._ensure_db_session()
    -            start_idx = len(conversation_history) if conversation_history else 0
    -            # Guard against the flush cursor overshooting the message list.
    -            # This can happen when repair_message_sequence compacts the list
    -            # (merging consecutive users, dropping stray tools) after the
    -            # cursor was set.  Fall back to start_idx so we don't skip
    -            # persisting the assistant/tool chain (#44837).
    -            flush_from = max(start_idx, min(self._last_flushed_db_idx, len(messages)))
    -            for msg in messages[flush_from:]:
    +            # Positional flushing used to slice at
    +            # max(len(conversation_history), _last_flushed_db_idx). That
    +            # assumes the live `messages` list is the original history plus a
    +            # new tail. repair_message_sequence can shrink/merge the history
    +            # copy before the final flush, making len(conversation_history)
    +            # larger than len(messages); the slice is then empty and delivered
    +            # assistant responses never reach state.db (#46053).
    +            #
    +            # Track object identities instead. `messages` is a shallow copy of
    +            # `conversation_history`, so history dicts are skipped by identity,
    +            # and new dicts appended during this turn are written once even if
    +            # repair compacts the list around them.
    +            current_session_id = getattr(self, "session_id", None)
    +            flushed_session_id = getattr(self, "_flushed_db_message_session_id", None)
    +            if flushed_session_id != current_session_id or self._last_flushed_db_idx == 0:
    +                self._flushed_db_message_ids = set()
    +                self._flushed_db_message_session_id = current_session_id
    +            flushed_ids = getattr(self, "_flushed_db_message_ids", None)
    +            if not isinstance(flushed_ids, set):
    +                flushed_ids = set()
    +                self._flushed_db_message_ids = flushed_ids
    +            history_ids = {
    +                id(item) for item in (conversation_history or [])
    +                if isinstance(item, dict)
    +            }
    +
    +            for msg in messages:
    +                if not isinstance(msg, dict):
    +                    continue
    +                msg_id = id(msg)
    +                if msg_id in flushed_ids:
    +                    continue
    +                if msg_id in history_ids:
    +                    flushed_ids.add(msg_id)
    +                    continue
                     role = msg.get("role", "unknown")
                     content = msg.get("content")
                     # Persist multimodal tool results as their text summary only —
    @@ -1605,6 +1633,7 @@ def _flush_messages_to_session_db(self, messages: List[Dict], conversation_histo
                         codex_reasoning_items=msg.get("codex_reasoning_items") if role == "assistant" else None,
                         codex_message_items=msg.get("codex_message_items") if role == "assistant" else None,
                     )
    +                flushed_ids.add(msg_id)
                 self._last_flushed_db_idx = len(messages)
             except Exception as e:
                 logger.warning("Session DB append_message failed: %s", e)

  • tests/run_agent/test_compression_persistence.py

    @@ -101,7 +101,7 @@ def test_flush_after_compression_with_long_history(self):
                 )
     
         def test_flush_with_stale_history_loses_messages(self):
    -        """Demonstrates the bug condition: stale conversation_history causes data loss."""
    +        """Stale conversation_history no longer causes data loss."""
             from hermes_state import SessionDB
     
             with tempfile.TemporaryDirectory() as tmpdir:
    @@ -120,17 +120,14 @@ def test_flush_with_stale_history_loses_messages(self):
                     {"role": "assistant", "content": "continuing..."},
                 ]
     
    -            # Bug: passing a conversation_history longer than compressed messages
    +            # Stale history longer than messages: the old positional flush
    +            # sliced past the end and dropped both messages (#46053).
                 stale_history = [{"role": "user", "content": f"msg{i}"} for i in range(100)]
                 agent._flush_messages_to_session_db(compressed, stale_history)
     
                 rows = db.get_messages("new-session")
    -            # With the stale history, flush_from = max(100, 0) = 100
    -            # But compressed only has 2 entries → messages[100:] = empty
    -            assert len(rows) == 0, (
    -                "Expected 0 messages with stale conversation_history "
    -                "(this test verifies the bug condition exists)"
    -            )
    +            assert len(rows) == 2
    +            assert [row["content"] for row in rows] == ["summary", "continuing..."]
     
     
     # ---------------------------------------------------------------------------

  • tests/run_agent/test_identity_flush.py

    @@ -0,0 +1,150 @@
    +"""Regression tests for identity-based SessionDB flushing (#46053)."""
    +
    +import os
    +import tempfile
    +from pathlib import Path
    +from unittest.mock import patch
    +
    +SESSION_ID = "test-identity-flush"
    +
    +
    +def _make_agent(session_db, session_id=SESSION_ID):
    +    with patch.dict(os.environ, {"OPENROUTER_API_KEY": "test-key"}):
    +        from run_agent import AIAgent
    +
    +        agent = AIAgent(
    +            api_key="[redacted:secret-assignment]",
    +            base_url="https://openrouter.ai/api/v1",
    +            model="test/model",
    +            quiet_mode=True,
    +            session_db=session_db,
    +            session_id=session_id,
    +            skip_context_files=True,
    +            skip_memory=True,
    +        )
    +    agent._ensure_db_session()
    +    return agent
    +
    +
    +def _contents(db, session_id=SESSION_ID):
    +    return [row["content"] for row in db.get_messages(session_id)]
    +
    +
    +class TestIdentityFlush:
    +    def test_repair_shrunk_messages_below_history_length_still_persists_assistant(self):
    +        """When repair shortens messages below conversation_history, don't slice empty."""
    +        from hermes_state import SessionDB
    +
    +        with tempfile.TemporaryDirectory() as tmpdir:
    +            db = SessionDB(db_path=Path(tmpdir) / "t.db")
    +            try:
    +                agent = _make_agent(db)
    +
    +                # Simulate history already loaded from state.db.
    +                history = [{"role": "user", "content": f"u{i}"} for i in range(6)]
    +                for msg in history:
    +                    db.append_message(
    +                        session_id=SESSION_ID,
    +                        role=msg["role"],
    +                        content=msg["content"],
    +                    )
    +
    +                # repair_message_sequence merged the six history rows into one
    +                # dict before this turn appended the new user/assistant pair.
    +                messages = [
    +                    {"role": "user", "content": "\n\n".join(f"u{i}" for i in range(6))},
    +                    {"role": "user", "content": "new question"},
    +                    {"role": "assistant", "content": "new answer"},
    +                ]
    +                assert len(history) > len(messages)
    +
    +                # The old positional flush computed flush_from >= len(messages)
    +                # and dropped the assistant. Identity flush persists new dicts.
    +                agent._last_flushed_db_idx = len(history)
    +                agent._flush_messages_to_session_db(messages, history)
    +
    +                contents = _contents(db)
    +                assert "new question" in contents
    +                assert "new answer" in contents
    +            finally:
    +                db.close()
    +
    +    def test_overlapping_turn_stale_cursor_does_not_drop_assistant(self):
    +        """A stale cached-agent cursor must not suppress this turn's new dicts."""
    +        from hermes_state import SessionDB
    +
    +        with tempfile.TemporaryDirectory() as tmpdir:
    +            db = SessionDB(db_path=Path(tmpdir) / "t.db")
    +            try:
    +                agent = _make_agent(db)
    +                history = [
    +                    {"role": "user", "content": "old question"},
    +                    {"role": "assistant", "content": "old answer"},
    +                ]
    +                for msg in history:
    +                    db.append_message(
    +                        session_id=SESSION_ID,
    +                        role=msg["role"],
    +                        content=msg["content"],
    +                    )
    +
    +                messages = history + [
    +                    {"role": "user", "content": "current question"},
    +                    {"role": "assistant", "content": "current answer"},
    +                ]
    +                agent._last_flushed_db_idx = len(messages) + 10
    +                agent._flush_messages_to_session_db(messages, history)
    +
    +                assert _contents(db) == [
    +                    "old question",
    +                    "old answer",
    +                    "current question",
    +                    "current answer",
    +                ]
    +            finally:
    +                db.close()
    +
    +    def test_repeated_flush_same_turn_writes_once(self):
    +        """Identity tracking preserves #860 same-turn dedup behavior."""
    +        from hermes_state import SessionDB
    +
    +        with tempfile.TemporaryDirectory() as tmpdir:
    +            db = SessionDB(db_path=Path(tmpdir) / "t.db")
    +            try:
    +                agent = _make_agent(db)
    +                messages = [{"role": "user", "content": "q"}]
    +
    +                agent._flush_messages_to_session_db(messages, [])
    +                messages.append({"role": "assistant", "content": "a"})
    +                agent._flush_messages_to_session_db(messages, [])
    +                agent._flush_messages_to_session_db(messages, [])
    +
    +                assert _contents(db) == ["q", "a"]
    +            finally:
    +                db.close()
    +
    +    def test_cursor_reset_starts_new_turn_identity_window(self):
    +        """Gateway resets _last_flushed_db_idx=0 before a cached-agent turn."""
    +        from hermes_state import SessionDB
    +
    +        with tempfile.TemporaryDirectory() as tmpdir:
    +            db = SessionDB(db_path=Path(tmpdir) / "t.db")
    +            try:
    +                agent = _make_agent(db)
    +                first_turn = [
    +                    {"role": "user", "content": "q1"},
    +                    {"role": "assistant", "content": "a1"},
    +                ]
    +                agent._flush_messages_to_session_db(first_turn, [])
    +
    +                history = [dict(m) for m in first_turn]
    +                second_turn = history + [
    +                    {"role": "user", "content": "q2"},
    +                    {"role": "assistant", "content": "a2"},
    +                ]
    +                agent._last_flushed_db_idx = 0
    +                agent._flush_messages_to_session_db(second_turn, history)
    +
    +                assert _contents(db) == ["q1", "a1", "q2", "a2"]
    +            finally:
    +                db.close()

• #50004 fix(cli): persist interrupted session transcript on SIGHUP/SIGTERM/window-close

  merged to main

  files: cli.py, tests/cli/test_cli_shutdown_memory_messages.py

• #50299 fix(session): finalize owned SQLite session rows on AIAgent.close() (#12029)

  merged to main

  files: agent/agent_init.py, agent/background_review.py, gateway/run.py, run_agent.py, tests/gateway/test_busy_session_ack.py, tests/run_agent/test_background_review.py, tests/tools/test_zombie_process_cleanup.py

  Show the diff

  • agent/agent_init.py

    @@ -1096,6 +1096,12 @@ def init_agent(
         agent._parent_session_id = parent_session_id
         agent._last_flushed_db_idx = 0  # tracks DB-write cursor to prevent duplicate writes
         agent._session_db_created = False  # DB row deferred to run_conversation()
    +    # Most agents own their session row and should finalize it on close().
    +    # Some temporary helper agents (manual compression / session-hygiene /
    +    # background-review forks) rotate or share the session forward to a
    +    # continuation row that must remain open after the helper is torn down;
    +    # those callers explicitly set this flag to False.
    +    agent._end_session_on_close = True
         agent._session_init_model_config = {
             "max_iterations": agent.max_iterations,
             "reasoning_config": reasoning_config,

  • agent/background_review.py

    @@ -575,6 +575,13 @@ def _bg_review_auto_deny(command, description, **kwargs):
                 # if a future code path bypasses the cache.
                 review_agent.session_start = agent.session_start
                 review_agent.session_id = agent.session_id
    +            # The fork shares the parent's live session_id (pinned above for
    +            # prefix-cache parity). It is single-lifecycle and calls close()
    +            # right after this run_conversation(); without opting out, close()
    +            # would finalize the parent's still-active session row mid
    +            # conversation (the review fires every ~10 turns). Leave session
    +            # finalization to the real owner (CLI close / gateway reset / cron).
    +            review_agent._end_session_on_close = False
                 # Never let the review fork compress. It shares the parent's
                 # session_id, so if it won a compression race it would rotate the
                 # parent into a NEW child that the gateway never adopts (the fork

  • gateway/run.py

    @@ -4686,6 +4686,27 @@ def _finalize_shutdown_agents(self, active_agents: Dict[str, Any]) -> None:
                     pass
                 self._cleanup_agent_resources(agent)
     
    +    def _should_emit_long_running_notification(
    +        self,
    +        session_key: Optional[str],
    +        agent: Any,
    +        executor_task: Optional[Any],
    +    ) -> bool:
    +        """Only emit the heartbeat while this task still owns the live run.
    +
    +        Guards against a stale ``running: delegate_task`` heartbeat outliving the
    +        run that started it: stop once the executor finishes, the agent is gone,
    +        or the session key has been rebound to a different live agent (e.g. the
    +        user sent ``/new`` and a fresh agent took the slot mid-run, #12029).
    +        """
    +        if agent is None:
    +            return False
    +        if executor_task is not None and executor_task.done():
    +            return False
    +        if session_key and self._running_agents.get(session_key) is not agent:
    +            return False
    +        return True
    +
         def _cleanup_agent_resources(self, agent: Any) -> None:
             """Best-effort cleanup for temporary or cached agent instances."""
             if agent is None:
    @@ -9194,6 +9215,13 @@ async def _handle_message_with_agent(self, event, source, _quick_key: str, run_g
                                         session_id=session_entry.session_id,
                                     )
                                     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).
    +                                    _hyg_agent._end_session_on_close = False
                                         _hyg_agent._print_fn = lambda *a, **kw: None
     
                                         loop = asyncio.get_running_loop()
    @@ -16274,6 +16302,20 @@ async def _notify_long_running():
                 _heartbeat_msg_id: Optional[str] = None
                 while True:
                     await asyncio.sleep(_NOTIFY_INTERVAL)
    +                # Stop heartbeating once this run no longer owns the session
    +                # slot or the executor has finished — otherwise a stale
    +                # "running: delegate_task" bubble can outlive the run that
    +                # spawned it (#12029). _executor_task is a closure var bound
    +                # just after this task is scheduled; tolerate the brief window
    +                # before then (the first wake is _NOTIFY_INTERVAL away anyway).
    +                try:
    +                    _exec_ref = _executor_task
    +                except NameError:
    +                    _exec_ref = None
    +                if not self._should_emit_long_running_notification(
    +                    session_key, agent_holder[0], _exec_ref
    +                ):
    +                    break
                     _elapsed_mins = int((time.time() - _notify_start) // 60)
                     # Include agent activity context if available. Default
                     # heartbeat is terse: elapsed + current tool. Verbose

  • run_agent.py

    @@ -3250,6 +3250,22 @@ def close(self) -> None:
             except Exception:
                 pass
     
    +        # 7. Finalize the owned SQLite session row unless this agent is only a
    +        # temporary helper that deliberately handed session ownership forward
    +        # (manual compression helpers that rotate to a continuation session_id,
    +        # or background-review forks that share the live parent's session_id and
    +        # must leave it open). end_session() is first-reason-wins and no-ops on
    +        # an already-ended row, so this never clobbers a 'compression' /
    +        # 'cron_complete' / 'cli_close' reason set by an earlier terminal path.
    +        try:
    +            if getattr(self, "_end_session_on_close", True):
    +                session_db = getattr(self, "_session_db", None)
    +                session_id = getattr(self, "session_id", None)
    +                if session_db and session_id:
    +                    session_db.end_session(session_id, "agent_close")
    +        except Exception:
    +            pass
    +
         def _hydrate_todo_store(self, history: List[Dict[str, Any]]) -> None:
             """
             Recover todo state from conversation history.

  • tests/gateway/test_busy_session_ack.py

    @@ -715,3 +715,62 @@ async def test_queue_mode_hint_points_to_interrupt(self, tmp_path, monkeypatch):
             assert "/busy interrupt" in content
             # Must NOT tell the user to /busy queue when they're already on queue.
             assert "/busy queue" not in content
    +
    +
    +class TestLongRunningNotificationOwnership:
    +    """The long-running heartbeat must stop once its run no longer owns the
    +    session slot or the executor finished — otherwise a stale
    +    'running: delegate_task' bubble outlives the run that spawned it (#12029).
    +    """
    +
    +    def test_notification_stops_after_session_ownership_moves(self):
    +        from gateway.run import GatewayRunner
    +
    +        runner = object.__new__(GatewayRunner)
    +        runner._running_agents = {}
    +
    +        original_agent = MagicMock()
    +        replacement_agent = MagicMock()
    +        runner._running_agents["sess"] = replacement_agent
    +
    +        assert runner._should_emit_long_running_notification(
    +            "sess", original_agent, executor_task=None
    +        ) is False
    +
    +    def test_notification_stops_after_executor_finishes(self):
    +        from gateway.run import GatewayRunner
    +
    +        runner = object.__new__(GatewayRunner)
    +        agent = MagicMock()
    +        runner._running_agents = {"sess": agent}
    +
    +        done_task = MagicMock()
    +        done_task.done.return_value = True
    +
    +        assert runner._should_emit_long_running_notification(
    +            "sess", agent, executor_task=done_task
    +        ) is False
    +
    +    def test_notification_stops_when_agent_is_gone(self):
    +        from gateway.run import GatewayRunner
    +
    +        runner = object.__new__(GatewayRunner)
    +        runner._running_agents = {}
    +
    +        assert runner._should_emit_long_running_notification(
    +            "sess", None, executor_task=None
    +        ) is False
    +
    +    def test_notification_continues_for_live_active_run(self):
    +        from gateway.run import GatewayRunner
    +
    +        runner = object.__new__(GatewayRunner)
    +        agent = MagicMock()
    +        runner._running_agents = {"sess": agent}
    +
    +        live_task = MagicMock()
    +        live_task.done.return_value = False
    +
    +        assert runner._should_emit_long_running_notification(
    +            "sess", agent, executor_task=live_task
    +        ) is True

  • tests/run_agent/test_background_review.py

    @@ -76,6 +76,50 @@ def close(self):
         ]
     
     
    +def test_background_review_fork_opts_out_of_session_finalization(monkeypatch):
    +    """The review fork shares the parent's live session_id, so it must set
    +    ``_end_session_on_close = False``. Otherwise close() (now finalizing owned
    +    session rows) would end the still-active parent session mid-conversation
    +    every time the review fires (~every 10 turns). Regression for #12029.
    +    """
    +    seen = {}
    +
    +    class FakeReviewAgent:
    +        def __init__(self, **kwargs):
    +            self._session_messages = []
    +            # Default matches AIAgent.__init__ (agent_init.py): owns its row.
    +            self._end_session_on_close = True
    +
    +        def __setattr__(self, name, value):
    +            object.__setattr__(self, name, value)
    +            if name == "_end_session_on_close":
    +                seen["end_session_on_close"] = value
    +
    +        def run_conversation(self, **kwargs):
    +            # By the time the fork runs, the opt-out must already be applied.
    +            seen["at_run_time"] = self._end_session_on_close
    +
    +        def shutdown_memory_provider(self):
    +            pass
    +
    +        def close(self):
    +            pass
    +
    +    monkeypatch.setattr(run_agent_module, "AIAgent", FakeReviewAgent)
    +    monkeypatch.setattr(run_agent_module.threading, "Thread", ImmediateThread)
    +
    +    agent = _bare_agent()
    +
    +    AIAgent._spawn_background_review(
    +        agent,
    +        messages_snapshot=[{"role": "user", "content": "hello"}],
    +        review_memory=True,
    +    )
    +
    +    assert seen.get("end_session_on_close") is False
    +    assert seen.get("at_run_time") is False
    +
    +
     def test_background_review_summarizer_receives_captured_messages_after_close(monkeypatch):
         """The action summarizer must see review messages even after close cleanup.
     

  • tests/tools/test_zombie_process_cleanup.py

    @@ -155,6 +155,59 @@ def test_close_propagates_to_children(self):
                 child_2.close.assert_called_once()
                 assert agent._active_children == []
     
    +    def test_close_ends_owned_session_row(self):
    +        """close() finalizes the agent's owned SQLite session row."""
    +        from unittest.mock import MagicMock, patch
    +
    +        with patch("run_agent.AIAgent.__init__", return_value=None):
    +            from run_agent import AIAgent
    +            agent = AIAgent.__new__(AIAgent)
    +            agent.session_id = "test-close-session-row"
    +            agent._active_children = []
    +            agent._active_children_lock = threading.Lock()
    +            agent.client = None
    +            agent._end_session_on_close = True
    +            agent._session_db = MagicMock()
    +
    +            agent.close()
    +
    +            agent._session_db.end_session.assert_called_once_with(
    +                "test-close-session-row", "agent_close"
    +            )
    +
    +    def test_close_skips_session_end_for_forwarded_continuation_agents(self):
    +        """Helper agents that handed session ownership forward opt out."""
    +        from unittest.mock import MagicMock, patch
    +
    +        with patch("run_agent.AIAgent.__init__", return_value=None):
    +            from run_agent import AIAgent
    +            agent = AIAgent.__new__(AIAgent)
    +            agent.session_id = "test-close-forwarded-session"
    +            agent._active_children = []
    +            agent._active_children_lock = threading.Lock()
    +            agent.client = None
    +            agent._end_session_on_close = False
    +            agent._session_db = MagicMock()
    +
    +            agent.close()
    +
    +            agent._session_db.end_session.assert_not_called()
    +
    +    def test_close_session_end_noops_without_session_db(self):
    +        """close() is a no-op for session finalization when no DB is wired in."""
    +        from unittest.mock import patch
    +
    +        with patch("run_agent.AIAgent.__init__", return_value=None):
    +            from run_agent import AIAgent
    +            agent = AIAgent.__new__(AIAgent)
    +            agent.session_id = "test-close-no-db"
    +            agent._active_children = []
    +            agent._active_children_lock = threading.Lock()
    +            agent.client = None
    +            # No _session_db / _end_session_on_close attributes at all —
    +            # getattr defaults must keep close() from raising.
    +            agent.close()  # must not raise
    +
         def test_close_survives_partial_failures(self):
             """close() continues cleanup even if one step fails."""
             from unittest.mock import patch

• #56279 fix(agent): persist recovered final responses at the finalize_turn chokepoint

  merged to main

  files: agent/turn_finalizer.py, scripts/release.py, tests/agent/test_turn_finalizer_final_response_persistence.py