fix(cli): persist one-shot resumed-session turns (Bot Chat bot-to-bot messages)

Bot Mode's bot-to-bot send (`hermes -p <bot> chat --in ~ -c "Bot Chat"
--create-if-missing -Q -q "..."`) runs one turn and exits. When the turn's
in-loop transcript flush failed transiently (state.db write-lock contention
with a multiplex gateway), the one-shot path had no end-of-run durable
retry: the reply reached stdout and agent.log while the resumed titled
session's stored history never changed (#88583). The interactive CLI is
immune — it retries the flush on the next persist point and finalizes the
row on quit — but every one-shot exit path lacked both.

Fix the whole class with cli._flush_one_shot_session_store():

- final _persist_session retry at one-shot exit (idempotent — per-message
  persisted-marker stamps mean already-written turns are not re-written)
- drain queued async token-accounting deltas
- end_session(..., "cli_close") so resumed/created titled session rows no
  longer dangle open forever after one-shot runs

Wired into _finalize_single_query (quiet -Q -q AND human -q paths, ahead
of memory-provider shutdown so nothing later can lose the turn) and into
the kanban SIGTERM handler before os._exit(0), which skips atexit and the
SessionDB token-drain hook entirely (same gap class as PR #50881).

Handed-off sessions (#88234) and persistence-isolated forks
(_persist_disabled) are skipped.

Fixes #88583

🤖 Generated with Hermes Agent

cli.py

@@ -1378,9 +1378,74 @@ def _notify_single_query_session_finalize(cli, *, reason: str = "shutdown") -> N
         _single_query_finalize_attempted_session_ids.add(session_id)
 
 
+def _flush_one_shot_session_store(cli) -> None:
+    """Durably flush + finalize the one-shot session row before process exit.
+
+    The quiet/one-shot ``-q`` / ``-Q`` paths (including resume-or-create of a
+    titled session via ``-c <name> --create-if-missing``, the Bot Mode
+    bot-to-bot send) get exactly ONE turn and then exit. The interactive CLI
+    finalizes its session row on quit (``end_session(..., "cli_close")``) and
+    every later turn retries a transiently-failed transcript flush; the
+    one-shot path had neither, so:
+
+    - a turn whose in-loop ``_flush_messages_to_session_db`` failed under
+      write-lock contention (e.g. a busy multiplex gateway sharing state.db)
+      was silently lost — the reply reached stdout and agent.log but the
+      resumed session's stored history never changed (#88583);
+    - the resumed/created titled session row was left dangling open
+      (``ended_at``/``end_reason`` NULL) on every one-shot exit;
+    - queued async token-accounting deltas relied on interpreter-exit hooks,
+      which the kanban SIGTERM path's ``os._exit(0)`` skips entirely.
+
+    Idempotent and best-effort: ``_persist_session`` dedupes via the
+    per-message ``_DB_PERSISTED_MARKER`` stamps (already-written turns are
+    not re-written) and ``end_session`` no-ops on an already-ended row.
+    Sessions handed off to the gateway are owned by the gateway process and
+    are left strictly alone (#88234).
+    """
+    agent = getattr(cli, "agent", None)
+    if agent is None:
+        return
+    session_id = getattr(agent, "session_id", None) or getattr(cli, "session_id", None)
+    if not session_id or session_id in _handed_off_session_ids:
+        return
+    if getattr(agent, "_persist_disabled", False):
+        return
+    # Retry persistence for any rows the in-turn flush failed to write.
+    # ``cli.conversation_history`` holds the resumed history's live dicts, so
+    # passing it keeps restored messages identity-skipped even when the failed
+    # first flush never got to stamp them.
+    try:
+        msgs = getattr(agent, "_session_messages", None)
+        if isinstance(msgs, list) and msgs and hasattr(agent, "_persist_session"):
+            agent._persist_session(
+                msgs, getattr(cli, "conversation_history", None)
+            )
+    except Exception:
+        logger.debug("one-shot final session persist retry failed", exc_info=True)
+    db = getattr(agent, "_session_db", None) or getattr(cli, "_session_db", None)
+    if db is None:
+        return
+    try:
+        db.flush_token_counts()
+    except Exception:
+        logger.debug("one-shot token-count drain failed", exc_info=True)
+    try:
+        db.end_session(session_id, "cli_close")
+    except Exception:
+        logger.debug("one-shot end_session failed", exc_info=True)
+
+
 def _finalize_single_query(cli) -> None:
     """Close one-shot CLI resources before releasing the active session lease."""
     try:
+        # Durable flush FIRST: memory-provider shutdown inside _run_cleanup
+        # can issue aux-LLM calls, and nothing after it may fail in a way
+        # that loses the turn (#88583).
+        try:
+            _flush_one_shot_session_store(cli)
+        except Exception:
+            logger.debug("one-shot session store flush failed", exc_info=True)
         _notify_single_query_session_finalize(cli)
         _run_cleanup(notify_session_finalize=False)
     finally:
@@ -20056,7 +20121,15 @@ def main(
                     # Cancel any pre-existing alarm to avoid colliding with
                     # caller-installed timers.
                     _sig_mod.signal(_sig_mod.SIGALRM, lambda *_: os._exit(0))
-                    _sig_mod.alarm(2)
+                    _sig_mod.alarm(5)
+            except Exception:
+                pass
+            # os._exit(0) skips atexit AND SessionDB's token-drain hook, so
+            # flush + finalize the session store here or the worker's turn
+            # (and its usage deltas) never become durable (#88583 / #50881
+            # class). Best-effort under the SIGALRM deadman above.
+            try:
+                _flush_one_shot_session_store(cli)
             except Exception:
                 pass
             try:

tests/cli/test_oneshot_resumed_session_persist.py

@@ -0,0 +1,202 @@
+"""Regression tests for #88583 — one-shot resumed-session turns must persist.
+
+Bot Mode's bot-to-bot send (``hermes -p <bot> chat --in ~ -c "Bot Chat"
+--create-if-missing -Q -q "..."``) runs exactly one turn and exits. The
+receiving agent replied, the CLI banner said it resumed the titled session,
+but nothing landed in state.db when the turn's in-loop transcript flush
+failed transiently (write-lock contention with a multiplex gateway sharing
+state.db): the one-shot path had no end-of-run durable retry and never
+finalized the session row, unlike the interactive CLI (which retries on the
+next turn and ends the session with ``cli_close`` on quit).
+
+The fix routes every one-shot exit (quiet ``-Q -q``, human ``-q``, and the
+kanban SIGTERM path) through ``cli._flush_one_shot_session_store``: a final
+``_persist_session`` retry (idempotent via the per-message persisted
+markers), a token-count drain, and ``end_session(..., "cli_close")``.
+"""
+
+from __future__ import annotations
+
+import os
+import tempfile
+from pathlib import Path
+from types import SimpleNamespace
+from unittest.mock import patch
+
+import pytest
+
+import cli as cli_mod
+
+
+@pytest.fixture(autouse=True)
+def _reset_finalize_state(monkeypatch):
+    monkeypatch.setattr(cli_mod, "_single_query_finalize_attempted_session_ids", set())
+    monkeypatch.setattr(cli_mod, "_handed_off_session_ids", set())
+    monkeypatch.setattr(cli_mod, "_cleanup_done", False, raising=False)
+
+
+def _make_agent(session_db, session_id="oneshot-88583"):
+    """Real AIAgent bound to a real temp SessionDB (test_860_dedup pattern)."""
+    with patch.dict(os.environ, {"OPENROUTER_API_KEY": "test-key"}):
+        from run_agent import AIAgent
+
+        agent = AIAgent(
+            api_key="test-key",
+            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 _fake_cli(agent):
+    return SimpleNamespace(
+        agent=agent,
+        session_id=agent.session_id,
+        conversation_history=[],
+        _session_db=agent._session_db,
+        _release_active_session=lambda: None,
+    )
+
+
+class TestOneShotDurableFlush:
+    """#88583: the one-shot exit path must retry persistence and finalize."""
+
+    def test_finalize_single_query_persists_unflushed_turn(self, monkeypatch):
+        """A turn whose in-loop flush failed must still reach state.db.
+
+        Simulates the reported failure: run_conversation produced the turn's
+        messages in memory (``_session_messages``) but the transcript flush
+        never landed (transient write-lock loss). Without the fix,
+        ``_finalize_single_query`` performs no durable write and the turn
+        evaporates — this test fails.
+        """
+        from hermes_state import SessionDB
+
+        with tempfile.TemporaryDirectory() as tmpdir:
+            db = SessionDB(db_path=Path(tmpdir) / "state.db")
+            try:
+                agent = _make_agent(db)
+                # The turn as run_conversation left it: in memory, un-stamped,
+                # never written (the in-loop flush failed transiently).
+                agent._session_messages = [
+                    {"role": "user", "content": "Message from 🤖 worker: hello, remember this"},
+                    {"role": "assistant", "content": "ack — noted."},
+                ]
+                assert db.get_messages(agent.session_id) == []
+
+                fake = _fake_cli(agent)
+                monkeypatch.setattr(cli_mod, "_run_cleanup", lambda **kw: None)
+                monkeypatch.setattr(
+                    cli_mod, "_notify_single_query_session_finalize", lambda _c: None
+                )
+
+                cli_mod._finalize_single_query(fake)
+
+                rows = db.get_messages(agent.session_id)
+                assert [r["role"] for r in rows] == ["user", "assistant"], (
+                    "one-shot exit must durably flush the turn to state.db "
+                    f"(#88583); got rows: {rows}"
+                )
+                assert "remember this" in rows[0]["content"]
+            finally:
+                db.close()
+
+    def test_finalize_single_query_ends_session_row(self, monkeypatch):
+        """The resumed/created one-shot session row is finalized on exit."""
+        from hermes_state import SessionDB
+
+        with tempfile.TemporaryDirectory() as tmpdir:
+            db = SessionDB(db_path=Path(tmpdir) / "state.db")
+            try:
+                agent = _make_agent(db)
+                agent._session_messages = [
+                    {"role": "user", "content": "hi"},
+                    {"role": "assistant", "content": "hello"},
+                ]
+                fake = _fake_cli(agent)
+                monkeypatch.setattr(cli_mod, "_run_cleanup", lambda **kw: None)
+                monkeypatch.setattr(
+                    cli_mod, "_notify_single_query_session_finalize", lambda _c: None
+                )
+
+                cli_mod._finalize_single_query(fake)
+
+                sess = db.get_session(agent.session_id)
+                assert sess["ended_at"] is not None
+                assert sess["end_reason"] == "cli_close"
+            finally:
+                db.close()
+
+    def test_flush_is_idempotent_for_already_persisted_turns(self, monkeypatch):
+        """A turn the in-loop flush already wrote is not duplicated."""
+        from hermes_state import SessionDB
+
+        with tempfile.TemporaryDirectory() as tmpdir:
+            db = SessionDB(db_path=Path(tmpdir) / "state.db")
+            try:
+                agent = _make_agent(db)
+                messages = [
+                    {"role": "user", "content": "hi"},
+                    {"role": "assistant", "content": "hello"},
+                ]
+                # Normal happy path: the in-loop flush already persisted.
+                agent._flush_messages_to_session_db(messages, [])
+                assert len(db.get_messages(agent.session_id)) == 2
+                agent._session_messages = messages
+
+                cli_mod._flush_one_shot_session_store(_fake_cli(agent))
+
+                rows = db.get_messages(agent.session_id)
+                assert len(rows) == 2, f"duplicate rows written: {rows}"
+            finally:
+                db.close()
+
+    def test_flush_skips_handed_off_sessions(self):
+        """A session handed off to the gateway is owned there (#88234)."""
+        from hermes_state import SessionDB
+
+        with tempfile.TemporaryDirectory() as tmpdir:
+            db = SessionDB(db_path=Path(tmpdir) / "state.db")
+            try:
+                agent = _make_agent(db)
+                agent._session_messages = [
+                    {"role": "user", "content": "hi"},
+                    {"role": "assistant", "content": "hello"},
+                ]
+                cli_mod._handed_off_session_ids.add(agent.session_id)
+
+                cli_mod._flush_one_shot_session_store(_fake_cli(agent))
+
+                assert db.get_messages(agent.session_id) == []
+                assert db.get_session(agent.session_id)["ended_at"] is None
+            finally:
+                db.close()
+
+    def test_flush_skips_persist_disabled_agents(self):
+        """Persistence-isolated forks must never write the canonical store."""
+        from hermes_state import SessionDB
+
+        with tempfile.TemporaryDirectory() as tmpdir:
+            db = SessionDB(db_path=Path(tmpdir) / "state.db")
+            try:
+                agent = _make_agent(db)
+                agent._persist_disabled = True
+                agent._session_messages = [
+                    {"role": "user", "content": "curator harness turn"},
+                ]
+
+                cli_mod._flush_one_shot_session_store(_fake_cli(agent))
+
+                assert db.get_messages(agent.session_id) == []
+            finally:
+                db.close()
+
+    def test_flush_survives_missing_agent(self):
+        cli_mod._flush_one_shot_session_store(SimpleNamespace(agent=None))
+        cli_mod._flush_one_shot_session_store(SimpleNamespace())