Fix regressions from lazy session creation changes

Fixes on record

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

• #20363 fix: resolve lazy session creation regressions (#18370 fallout)

  merged to main

  files: cli.py, gateway/run.py, hermes_state.py, tests/test_lazy_session_regressions.py, tests/test_tui_gateway_server.py, tui_gateway/server.py

  Show the diff

  • cli.py

    @@ -940,6 +940,18 @@ def _run_state_db_auto_maintenance(session_db) -> None:
             except Exception as _prune_exc:
                 logger.debug("Ghost session prune skipped: %s", _prune_exc)
     
    +        # One-time finalize of orphaned compression continuations (#20001).
    +        try:
    +            if not session_db.get_meta("orphaned_compression_finalize_v1"):
    +                finalized = session_db.finalize_orphaned_compression_sessions()
    +                session_db.set_meta("orphaned_compression_finalize_v1", "1")
    +                if finalized:
    +                    logger.info(
    +                        "Finalized %d orphaned compression sessions", finalized
    +                    )
    +        except Exception as _finalize_exc:
    +            logger.debug("Orphan compression finalize skipped: %s", _finalize_exc)
    +
             cfg = (_load_full_config().get("sessions") or {})
             if not cfg.get("auto_prune", False):
                 return

  • gateway/run.py

    @@ -939,6 +939,52 @@ def _format_gateway_process_notification(evt: dict) -> "str | None":
     _gateway_runner_ref: _weakref.ref = lambda: None
     
     
    +def _normalize_empty_agent_response(
    +    agent_result: dict,
    +    response: str,
    +    *,
    +    history_len: int = 0,
    +) -> str:
    +    """Normalize empty/None agent responses into user-facing messages.
    +
    +    Consolidates the existing ``failed`` handler and adds a catch-all for
    +    the case where the agent did work (api_calls > 0) but returned no text.
    +    Fix for #18765.
    +    """
    +    if response:
    +        return response
    +
    +    if agent_result.get("failed"):
    +        error_detail = agent_result.get("error", "unknown error")
    +        error_str = str(error_detail).lower()
    +        is_context_failure = any(
    +            p in error_str
    +            for p in ("context", "token", "too large", "too long", "exceed", "payload")
    +        ) or ("400" in error_str and history_len > 50)
    +        if is_context_failure:
    +            return (
    +                "⚠️ Session too large for the model's context window.\n"
    +                "Use /compact to compress the conversation, or "
    +                "/reset to start fresh."
    +            )
    +        return (
    +            f"The request failed: {str(error_detail)[:300]}\n"
    +            "Try again or use /reset to start a fresh session."
    +        )
    +
    +    api_calls = int(agent_result.get("api_calls", 0) or 0)
    +    if api_calls > 0 and not agent_result.get("interrupted"):
    +        if agent_result.get("partial"):
    +            err = agent_result.get("error", "processing incomplete")
    +            return f"⚠️ Processing stopped: {str(err)[:200]}. Try again."
    +        return (
    +            "⚠️ Processing completed but no response was generated. "
    +            "This may be a transient error — try sending your message again."
    +        )
    +
    +    return response
    +
    +
     class GatewayRunner:
         """
         Main gateway controller.
    @@ -6439,33 +6485,11 @@ async def _handle_message_with_agent(self, event, source, _quick_key: str, run_g
                             session_key, _e,
                         )
     
    -            # Surface error details when the agent failed silently (final_response=None)
    -            if not response and agent_result.get("failed"):
    -                error_detail = agent_result.get("error", "unknown error")
    -                error_str = str(error_detail).lower()
    -
    -                # Detect context-overflow failures and give specific guidance.
    -                # Generic 400 "Error" from Anthropic with large sessions is the
    -                # most common cause of this (#1630).
    -                _is_ctx_fail = any(p in error_str for p in (
    -                    "context", "token", "too large", "too long",
    -                    "exceed", "payload",
    -                )) or (
    -                    "400" in error_str
    -                    and len(history) > 50
    -                )
    -
    -                if _is_ctx_fail:
    -                    response = (
    -                        "⚠️ Session too large for the model's context window.\n"
    -                        "Use /compact to compress the conversation, or "
    -                        "/reset to start fresh."
    -                    )
    -                else:
    -                    response = (
    -                        f"The request failed: {str(error_detail)[:300]}\n"
    -                        "Try again or use /reset to start a fresh session."
    -                    )
    +            # Normalize empty responses: surface errors, partial failures, and
    +            # the case where agent did work but returned no text. Fix for #18765.
    +            response = _normalize_empty_agent_response(
    +                agent_result, response, history_len=len(history),
    +            )
     
                 # If the agent's session_id changed during compression, update
                 # session_entry so transcript writes below go to the right session.

  • hermes_state.py

    @@ -718,6 +718,45 @@ def _do(conn):
                     self._remove_session_files(sessions_dir, sid)
             return len(removed_ids)
     
    +    def finalize_orphaned_compression_sessions(self) -> int:
    +        """Mark orphaned compression continuation sessions as ended.
    +
    +        Targets child sessions that were never finalized: parent is ended
    +        with reason='compression', child has messages but no end_reason/ended_at
    +        and api_call_count=0.  Non-destructive: preserves all messages and sets
    +        end_reason='orphaned_compression'.  Fix for #20001.
    +        """
    +        cutoff = time.time() - 604800  # 7 days
    +
    +        def _do(conn):
    +            now = time.time()
    +            result = conn.execute(
    +                """
    +                UPDATE sessions
    +                SET ended_at = ?,
    +                    end_reason = 'orphaned_compression'
    +                WHERE api_call_count = 0
    +                  AND end_reason IS NULL
    +                  AND ended_at IS NULL
    +                  AND started_at < ?
    +                  AND parent_session_id IS NOT NULL
    +                  AND EXISTS (
    +                      SELECT 1 FROM sessions p
    +                      WHERE p.id = sessions.parent_session_id
    +                        AND p.end_reason = 'compression'
    +                        AND p.ended_at IS NOT NULL
    +                  )
    +                  AND EXISTS (
    +                      SELECT 1 FROM messages m
    +                      WHERE m.session_id = sessions.id
    +                  )
    +                """,
    +                (now, cutoff),
    +            )
    +            return result.rowcount
    +
    +        return self._execute_write(_do) or 0
    +
         def get_session(self, session_id: str) -> Optional[Dict[str, Any]]:
             """Get a session by ID."""
             with self._lock:

  • tests/test_lazy_session_regressions.py

    @@ -0,0 +1,608 @@
    +"""Reproduction tests for #18370 fallout: lazy session creation regressions.
    +
    +Tests cover:
    +1. Bug #20001 — _finalize_session() uses stale session_key after compression rotation
    +2. Bug #20001 — _sync_session_key_after_compress called post-run_conversation
    +3. Bug #19029 — pending_title ValueError leaves title wedged
    +4. Bug #18765 — gateway surfaces null response when agent did work
    +5. Prune — finalize_orphaned_compression_sessions catches ghost continuations
    +"""
    +
    +import threading
    +import time
    +import types
    +from pathlib import Path
    +from unittest.mock import MagicMock, patch
    +
    +import pytest
    +
    +
    +# ===========================================================================
    +# Helpers
    +# ===========================================================================
    +
    +def _make_session_db(tmp_path):
    +    """Create a real SessionDB for integration-style tests."""
    +    from hermes_state import SessionDB
    +    db_path = tmp_path / "test_state.db"
    +    return SessionDB(db_path=db_path)
    +
    +
    +def _tui_session(agent=None, session_key="session-key-old", **extra):
    +    """Minimal TUI gateway session dict matching server._sessions values."""
    +    return {
    +        "agent": agent if agent is not None else types.SimpleNamespace(session_id=session_key),
    +        "session_key": session_key,
    +        "history": [],
    +        "history_lock": threading.Lock(),
    +        "history_version": 0,
    +        "running": False,
    +        "attached_images": [],
    +        "image_counter": 0,
    +        "cols": 80,
    +        "slash_worker": None,
    +        "show_reasoning": False,
    +        "tool_progress_mode": "all",
    +        "pending_title": None,
    +        **extra,
    +    }
    +
    +
    +# ===========================================================================
    +# Bug #20001: _finalize_session uses stale session_key
    +# ===========================================================================
    +
    +class TestFinalizeSessionUsesAgentSessionId:
    +    """After compression rotates agent.session_id, _finalize_session()
    +    must call end_session() on the NEW (current) session_id, not the stale
    +    session_key stored in the session dict."""
    +
    +    def test_finalize_targets_agent_session_id_not_stale_key(self, tmp_path):
    +        """Reproduction: agent.session_id rotated by compression, but
    +        session['session_key'] still holds old value. _finalize_session()
    +        should end the agent's current session."""
    +        from tui_gateway import server
    +
    +        db = _make_session_db(tmp_path)
    +
    +        # Create two sessions: parent (already ended by compression) and continuation
    +        db.create_session(session_id="parent-session", source="tui", model="test")
    +        db.end_session("parent-session", "compression")
    +
    +        db.create_session(
    +            session_id="continuation-session",
    +            source="tui",
    +            model="test",
    +            parent_session_id="parent-session",
    +        )
    +        # Continuation is NOT ended — this is the bug state
    +
    +        # Agent has rotated to continuation session
    +        agent = types.SimpleNamespace(
    +            session_id="continuation-session",
    +            commit_memory_session=lambda h: None,
    +        )
    +
    +        # Session dict still holds stale key (the bug condition)
    +        session = _tui_session(
    +            agent=agent,
    +            session_key="parent-session",
    +            history=[{"role": "user", "content": "hello"}],
    +        )
    +
    +        # Monkeypatch _get_db to return our test DB
    +        with patch.object(server, "_get_db", return_value=db):
    +            with patch.object(server, "_notify_session_boundary", lambda *a: None):
    +                server._finalize_session(session, end_reason="tui_close")
    +
    +        # The continuation session should be ended
    +        continuation = db.get_session("continuation-session")
    +        assert continuation["ended_at"] is not None, (
    +            "_finalize_session should end the agent's current session (continuation), "
    +            "not the already-ended parent"
    +        )
    +        assert continuation["end_reason"] == "tui_close"
    +
    +    def test_finalize_fallback_to_session_key_when_agent_is_none(self, tmp_path):
    +        """When agent is None (e.g. session never fully initialized),
    +        _finalize_session falls back to session_key."""
    +        from tui_gateway import server
    +
    +        db = _make_session_db(tmp_path)
    +        db.create_session(session_id="orphan-key", source="tui", model="test")
    +
    +        session = _tui_session(agent=None, session_key="orphan-key")
    +
    +        with patch.object(server, "_get_db", return_value=db):
    +            with patch.object(server, "_notify_session_boundary", lambda *a: None):
    +                server._finalize_session(session, end_reason="tui_close")
    +
    +        row = db.get_session("orphan-key")
    +        assert row["ended_at"] is not None
    +        assert row["end_reason"] == "tui_close"
    +
    +
    +# ===========================================================================
    +# Bug #20001: _sync_session_key_after_compress post-run_conversation
    +# ===========================================================================
    +
    +class TestSyncSessionKeyAfterAutoCompress:
    +    """When auto-compression fires inside run_conversation(), the post-turn
    +    code in _run_prompt_submit must call _sync_session_key_after_compress
    +    to update session_key for downstream consumers (title, goals, etc.)."""
    +
    +    def test_session_key_synced_after_run_conversation_with_compression(self, monkeypatch):
    +        """Simulate: run_conversation() internally compresses and rotates
    +        agent.session_id. After it returns, session['session_key'] must match."""
    +        from tui_gateway import server
    +
    +        class _CompressingAgent:
    +            """Agent that simulates compression-driven session_id rotation."""
    +            def __init__(self):
    +                self.session_id = "pre-compress-key"
    +                self._cached_system_prompt = ""
    +
    +            def run_conversation(self, prompt, conversation_history=None, stream_callback=None):
    +                # Simulate what _compress_context does: rotate session_id
    +                self.session_id = "post-compress-key"
    +                return {
    +                    "final_response": "done",
    +                    "messages": [
    +                        {"role": "user", "content": prompt},
    +                        {"role": "assistant", "content": "done"},
    +                    ],
    +                }
    +
    +        agent = _CompressingAgent()
    +        session = _tui_session(agent=agent, session_key="pre-compress-key")
    +
    +        # Track if _sync_session_key_after_compress was called
    +        sync_calls = []
    +        original_sync = server._sync_session_key_after_compress
    +
    +        def _tracking_sync(sid, sess, **kwargs):
    +            sync_calls.append((sid, sess.get("session_key")))
    +            # Just update the key directly (skip approval routing etc.)
    +            new_id = getattr(sess.get("agent"), "session_id", None) or ""
    +            if new_id and new_id != sess.get("session_key"):
    +                sess["session_key"] = new_id
    +
    +        monkeypatch.setattr(server, "_sync_session_key_after_compress", _tracking_sync)
    +        monkeypatch.setattr(server, "_emit", lambda *a, **kw: None)
    +        monkeypatch.setattr(server, "make_stream_renderer", lambda cols: None)
    +        monkeypatch.setattr(server, "render_message", lambda raw, cols: None)
    +
    +        # Use _ImmediateThread pattern to run synchronously
    +        class _ImmediateThread:
    +            def __init__(self, target=None, daemon=None, **kw):
    +                self._target = target
    +            def start(self):
    +                self._target()
    +
    +        server._sessions["test-sid"] = session
    +        monkeypatch.setattr(server.threading, "Thread", _ImmediateThread)
    +
    +        try:
    +            server.handle_request({
    +                "id": "1",
    +                "method": "prompt.submit",
    +                "params": {"session_id": "test-sid", "text": "hello"},
    +            })
    +
    +            # Sync should have been called
    +            assert len(sync_calls) > 0, (
    +                "_sync_session_key_after_compress must be called after run_conversation "
    +                "to pick up compression-driven session_id rotation"
    +            )
    +
    +            # session_key should now match agent.session_id
    +            assert session["session_key"] == "post-compress-key", (
    +                "session_key must be updated to match agent.session_id after compression"
    +            )
    +        finally:
    +            server._sessions.pop("test-sid", None)
    +
    +
    +# ===========================================================================
    +# Bug #19029: pending_title ValueError wedge
    +# ===========================================================================
    +
    +class TestPendingTitleValueError:
    +    """When set_session_title raises ValueError (duplicate/invalid title),
    +    pending_title must be cleared — not left wedged forever."""
    +
    +    def test_valueerror_clears_pending_title(self, monkeypatch):
    +        """ValueError from set_session_title should drop pending_title."""
    +        from tui_gateway import server
    +
    +        mock_db = MagicMock()
    +        mock_db.set_session_title.side_effect = ValueError("duplicate title")
    +
    +        class _Agent:
    +            session_id = "test-session"
    +            _cached_system_prompt = ""
    +            def run_conversation(self, prompt, **kw):
    +                return {
    +                    "final_response": "ok",
    +                    "messages": [{"role": "assistant", "content": "ok"}],
    +                }
    +
    +        session = _tui_session(
    +            agent=_Agent(),
    +            session_key="test-session",
    +            pending_title="My Title",
    +        )
    +
    +        monkeypatch.setattr(server, "_get_db", lambda: mock_db)
    +        monkeypatch.setattr(server, "_emit", lambda *a, **kw: None)
    +        monkeypatch.setattr(server, "make_stream_renderer", lambda cols: None)
    +        monkeypatch.setattr(server, "render_message", lambda raw, cols: None)
    +        monkeypatch.setattr(
    +            server, "_sync_session_key_after_compress", lambda *a, **kw: None
    +        )
    +
    +        class _ImmediateThread:
    +            def __init__(self, target=None, daemon=None, **kw):
    +                self._target = target
    +            def start(self):
    +                self._target()
    +
    +        server._sessions["sid"] = session
    +        monkeypatch.setattr(server.threading, "Thread", _ImmediateThread)
    +
    +        try:
    +            server.handle_request({
    +                "id": "1",
    +                "method": "prompt.submit",
    +                "params": {"session_id": "sid", "text": "hello"},
    +            })
    +
    +            # pending_title should be cleared on ValueError, not left wedged
    +            assert session.get("pending_title") is None, (
    +                "ValueError from set_session_title must clear pending_title "
    +                "so auto-title can take over"
    +            )
    +        finally:
    +            server._sessions.pop("sid", None)
    +
    +    def test_other_exception_keeps_pending_title_for_retry(self, monkeypatch):
    +        """Non-ValueError exceptions should keep pending_title for retry."""
    +        from tui_gateway import server
    +
    +        mock_db = MagicMock()
    +        mock_db.set_session_title.side_effect = RuntimeError("transient DB lock")
    +
    +        class _Agent:
    +            session_id = "test-session"
    +            _cached_system_prompt = ""
    +            def run_conversation(self, prompt, **kw):
    +                return {
    +                    "final_response": "ok",
    +                    "messages": [{"role": "assistant", "content": "ok"}],
    +                }
    +
    +        session = _tui_session(
    +            agent=_Agent(),
    +            session_key="test-session",
    +            pending_title="My Title",
    +        )
    +
    +        monkeypatch.setattr(server, "_get_db", lambda: mock_db)
    +        monkeypatch.setattr(server, "_emit", lambda *a, **kw: None)
    +        monkeypatch.setattr(server, "make_stream_renderer", lambda cols: None)
    +        monkeypatch.setattr(server, "render_message", lambda raw, cols: None)
    +        monkeypatch.setattr(
    +            server, "_sync_session_key_after_compress", lambda *a, **kw: None
    +        )
    +
    +        class _ImmediateThread:
    +            def __init__(self, target=None, daemon=None, **kw):
    +                self._target = target
    +            def start(self):
    +                self._target()
    +
    +        server._sessions["sid"] = session
    +        monkeypatch.setattr(server.threading, "Thread", _ImmediateThread)
    +
    +        try:
    +            server.handle_request({
    +                "id": "1",
    +                "method": "prompt.submit",
    +                "params": {"session_id": "sid", "text": "hello"},
    +            })
    +
    +            # Non-ValueError should keep pending_title for retry
    +            assert session.get("pending_title") == "My Title", (
    +                "Non-ValueError exceptions should keep pending_title intact "
    +                "for retry on next turn"
    +            )
    +        finally:
    +            server._sessions.pop("sid", None)
    +
    +
    +# ===========================================================================
    +# Bug #18765: Gateway surfaces null response
    +# ===========================================================================
    +
    +class TestGatewaySurfacesNullResponse:
    +    """When the agent does work (api_calls > 0) but returns no final_response,
    +    the gateway must surface an error to the user instead of silently sending
    +    nothing. Tests exercise the production _normalize_empty_agent_response helper."""
    +
    +    def test_partial_response_surfaces_error(self):
    +        """Agent returns partial=True with no response → user sees error."""
    +        from gateway.run import _normalize_empty_agent_response
    +
    +        agent_result = {
    +            "final_response": None,
    +            "api_calls": 5,
    +            "partial": True,
    +            "interrupted": False,
    +            "error": "Model generated invalid tool call: nonexistent_tool",
    +        }
    +
    +        response = agent_result.get("final_response") or ""
    +        response = _normalize_empty_agent_response(
    +            agent_result, response, history_len=10,
    +        )
    +
    +        assert response != "", "Null response with api_calls>0 must be surfaced"
    +        assert "nonexistent_tool" in response
    +
    +    def test_interrupted_response_stays_empty(self):
    +        """Interrupted agent → response stays empty (platform handles UX)."""
    +        from gateway.run import _normalize_empty_agent_response
    +
    +        agent_result = {
    +            "final_response": None,
    +            "api_calls": 3,
    +            "partial": False,
    +            "interrupted": True,
    +        }
    +
    +        response = agent_result.get("final_response") or ""
    +        response = _normalize_empty_agent_response(
    +            agent_result, response, history_len=10,
    +        )
    +
    +        assert response == "", "Interrupted turns should not get synthetic responses"
    +
    +    def test_failed_context_overflow(self):
    +        """Agent failed with context overflow → specific guidance message."""
    +        from gateway.run import _normalize_empty_agent_response
    +
    +        agent_result = {
    +            "final_response": None,
    +            "api_calls": 0,
    +            "failed": True,
    +            "error": "400 Bad Request: context length exceeded",
    +        }
    +
    +        response = agent_result.get("final_response") or ""
    +        response = _normalize_empty_agent_response(
    +            agent_result, response, history_len=60,
    +        )
    +
    +        assert "context window" in response
    +        assert "/compact" in response
    +
    +    def test_failed_generic_error(self):
    +        """Agent failed with non-context error → generic error message."""
    +        from gateway.run import _normalize_empty_agent_response
    +
    +        agent_result = {
    +            "final_response": None,
    +            "api_calls": 0,
    +            "failed": True,
    +            "error": "500 Internal Server Error",
    +        }
    +
    +        response = agent_result.get("final_response") or ""
    +        response = _normalize_empty_agent_response(
    +            agent_result, response, history_len=5,
    +        )
    +
    +        assert "500 Internal Server Error" in response
    +        assert "/reset" in response
    +
    +    def test_nonempty_response_passes_through(self):
    +        """Non-empty response is returned unchanged."""
    +        from gateway.run import _normalize_empty_agent_response
    +
    +        agent_result = {"final_response": "Hello!", "api_calls": 1}
    +        response = "Hello!"
    +        result = _normalize_empty_agent_response(
    +            agent_result, response, history_len=5,
    +        )
    +
    +        assert result == "Hello!"
    +
    +
    +# ===========================================================================
    +# Prune: finalize_orphaned_compression_sessions
    +# ===========================================================================
    +
    +class TestFinalizeOrphanedCompressionSessions:
    +    """The prune migration marks ghost compression continuations as ended."""
    +
    +    def test_marks_ghost_continuation_with_compression_parent(self, tmp_path):
    +        """Ghost session with compression-ended parent + messages → finalized."""
    +        db = _make_session_db(tmp_path)
    +
    +        # Parent session (ended by compression — this is the key condition)
    +        db.create_session(session_id="parent", source="tui", model="test")
    +        db.end_session("parent", "compression")
    +
    +        # Ghost continuation (has messages, never finalized)
    +        db.create_session(
    +            session_id="ghost-cont",
    +            source="tui",
    +            model="test",
    +            parent_session_id="parent",
    +        )
    +        db.append_message("ghost-cont", role="user", content="hello")
    +        db.append_message("ghost-cont", role="assistant", content="hi")
    +
    +        # Make it old enough (fake started_at)
    +        db._execute_write(
    +            lambda conn: conn.execute(
    +                "UPDATE sessions SET started_at = ? WHERE id = ?",
    +                (time.time() - 800000, "ghost-cont"),  # ~9 days old
    +            )
    +        )
    +
    +        count = db.finalize_orphaned_compression_sessions()
    +        assert count == 1
    +
    +        session = db.get_session("ghost-cont")
    +        assert session["ended_at"] is not None
    +        assert session["end_reason"] == "orphaned_compression"
    +
    +    def test_skips_session_without_parent(self, tmp_path):
    +        """Ghost session without parent_session_id is NOT a compression
    +        continuation — should not be touched by this prune."""
    +        db = _make_session_db(tmp_path)
    +
    +        db.create_session(session_id="ghost-notitle", source="tui", model="test")
    +        db.append_message("ghost-notitle", role="user", content="test")
    +
    +        db._execute_write(
    +            lambda conn: conn.execute(
    +                "UPDATE sessions SET started_at = ? WHERE id = ?",
    +                (time.time() - 800000, "ghost-notitle"),
    +            )
    +        )
    +
    +        count = db.finalize_orphaned_compression_sessions()
    +        assert count == 0
    +
    +    def test_skips_recent_sessions(self, tmp_path):
    +        """Sessions younger than 7 days are not touched."""
    +        db = _make_session_db(tmp_path)
    +
    +        # Create parent first to satisfy FK constraint
    +        db.create_session(session_id="some-parent", source="tui", model="test")
    +        db.create_session(
    +            session_id="recent",
    +            source="tui",
    +            model="test",
    +            parent_session_id="some-parent",
    +        )
    +        db.append_message("recent", role="user", content="hello")
    +        # started_at is now() — within 7 days
    +
    +        count = db.finalize_orphaned_compression_sessions()
    +        assert count == 0
    +
    +    def test_skips_sessions_with_end_reason(self, tmp_path):
    +        """Properly finalized sessions (even without api_call_count) are skipped."""
    +        db = _make_session_db(tmp_path)
    +
    +        # Create parent first to satisfy FK constraint
    +        db.create_session(session_id="parent", source="tui", model="test")
    +        db.end_session("parent", "compression")
    +
    +        db.create_session(
    +            session_id="already-ended",
    +            source="tui",
    +            model="test",
    +            parent_session_id="parent",
    +        )
    +        db.append_message("already-ended", role="user", content="hello")
    +        db.end_session("already-ended", "user_exit")
    +
    +        db._execute_write(
    +            lambda conn: conn.execute(
    +                "UPDATE sessions SET started_at = ? WHERE id = ?",
    +                (time.time() - 800000, "already-ended"),
    +            )
    +        )
    +
    +        count = db.finalize_orphaned_compression_sessions()
    +        assert count == 0
    +
    +    def test_skips_session_with_non_compression_parent(self, tmp_path):
    +        """Child session whose parent was NOT ended by compression should
    +        not be touched — it's not from the compression continuation path."""
    +        db = _make_session_db(tmp_path)
    +
    +        # Parent ended by user_exit, not compression
    +        db.create_session(session_id="parent", source="tui", model="test")
    +        db.end_session("parent", "user_exit")
    +
    +        db.create_session(
    +            session_id="child",
    +            source="tui",
    +            model="test",
    +            parent_session_id="parent",
    +        )
    +        db.append_message("child", role="user", content="hello")
    +
    +        db._execute_write(
    +            lambda conn: conn.execute(
    +                "UPDATE sessions SET started_at = ? WHERE id = ?",
    +                (time.time() - 800000, "child"),
    +            )
    +        )
    +
    +        count = db.finalize_orphaned_compression_sessions()
    +        assert count == 0
    +
    +    def test_skips_sessions_without_messages(self, tmp_path):
    +        """Empty sessions (no messages) are NOT targeted by this prune —
    +        those are handled by prune_empty_ghost_sessions()."""
    +        db = _make_session_db(tmp_path)
    +
    +        # Create parent first to satisfy FK constraint
    +        db.create_session(session_id="parent", source="tui", model="test")
    +        db.end_session("parent", "compression")
    +
    +        db.create_session(
    +            session_id="empty-ghost",
    +            source="tui",
    +            model="test",
    +            parent_session_id="parent",
    +        )
    +        # No messages appended
    +
    +        db._execute_write(
    +            lambda conn: conn.execute(
    +                "UPDATE sessions SET started_at = ? WHERE id = ?",
    +                (time.time() - 800000, "empty-ghost"),
    +            )
    +        )
    +
    +        count = db.finalize_orphaned_compression_sessions()
    +        assert count == 0
    +
    +    def test_titled_ghost_with_parent_is_caught(self, tmp_path):
    +        """Ghost continuation that HAS a title (propagated from parent by
    +        _compress_context) is still caught via parent with end_reason='compression'."""
    +        db = _make_session_db(tmp_path)
    +
    +        # Create parent first — ended by compression
    +        db.create_session(session_id="parent", source="tui", model="test")
    +        db.set_session_title("parent", "Chat")
    +        db.end_session("parent", "compression")
    +
    +        db.create_session(
    +            session_id="titled-ghost",
    +            source="tui",
    +            model="test",
    +            parent_session_id="parent",
    +        )
    +        db.set_session_title("titled-ghost", "Chat (2)")
    +        db.append_message("titled-ghost", role="user", content="continued...")
    +
    +        db._execute_write(
    +            lambda conn: conn.execute(
    +                "UPDATE sessions SET started_at = ? WHERE id = ?",
    +                (time.time() - 800000, "titled-ghost"),
    +            )
    +        )
    +
    +        count = db.finalize_orphaned_compression_sessions()
    +        assert count == 1
    +
    +        session = db.get_session("titled-ghost")
    +        assert session["end_reason"] == "orphaned_compression"

  • tests/test_tui_gateway_server.py

    @@ -921,56 +921,70 @@ def set_session_title(self, _key, _title):
     
     
     def test_session_create_drops_pending_title_on_valueerror(monkeypatch):
    -    unblock_agent = threading.Event()
    -
    -    class _FakeWorker:
    -        def __init__(self, key, model):
    -            self.key = key
    -
    -        def close(self):
    -            return None
    +    """When set_session_title raises ValueError during post-message title flush,
    +    pending_title should be dropped (non-retryable). Updated for post-#18370
    +    lazy session creation where title is applied post-first-message.
    +    """
     
    -    class _FakeAgent:
    +    class _Agent:
    +        session_id = "test-session"
             model = "x"
             provider = "openrouter"
             base_url = ""
             api_key = ""
    +        _cached_system_prompt = ""
     
    -    class _FakeDB:
    -        def create_session(self, _key, source="tui", model=None):
    -            return None
    +        def run_conversation(self, prompt, **kw):
    +            return {
    +                "final_response": "ok",
    +                "messages": [{"role": "assistant", "content": "ok"}],
    +            }
     
    +    class _FakeDB:
             def set_session_title(self, _key, _title):
                 raise ValueError("Title already in use")
     
    -    def _make_agent(_sid, _key):
    -        unblock_agent.wait(timeout=2.0)
    -        return _FakeAgent()
    -
    -    monkeypatch.setattr(server, "_make_agent", _make_agent)
    -    monkeypatch.setattr(server, "_SlashWorker", _FakeWorker)
    -    monkeypatch.setattr(server, "_get_db", lambda: _FakeDB())
    -    monkeypatch.setattr(server, "_session_info", lambda _a: {"model": "x"})
    -    monkeypatch.setattr(server, "_probe_credentials", lambda _a: None)
    -    monkeypatch.setattr(server, "_wire_callbacks", lambda _sid: None)
    -    monkeypatch.setattr(server, "_emit", lambda *a, **kw: None)
    +    class _ImmediateThread:
    +        def __init__(self, target=None, daemon=None, **kw):
    +            self._target = target
     
    -    import tools.approval as _approval
    +        def start(self):
    +            self._target()
     
    -    monkeypatch.setattr(_approval, "register_gateway_notify", lambda key, cb: None)
    -    monkeypatch.setattr(_approval, "load_permanent_allowlist", lambda: None)
    +    agent = _Agent()
    +    session = {
    +        "agent": agent,
    +        "session_key": "test-session",
    +        "history": [],
    +        "history_lock": threading.Lock(),
    +        "history_version": 0,
    +        "running": False,
    +        "attached_images": [],
    +        "image_counter": 0,
    +        "cols": 80,
    +        "slash_worker": None,
    +        "show_reasoning": False,
    +        "tool_progress_mode": "all",
    +        "pending_title": "duplicate title",
    +    }
     
    -    resp = server.handle_request(
    -        {"id": "1", "method": "session.create", "params": {"cols": 80}}
    +    server._sessions["sid"] = session
    +    monkeypatch.setattr(server, "_get_db", lambda: _FakeDB())
    +    monkeypatch.setattr(server, "_emit", lambda *a, **kw: None)
    +    monkeypatch.setattr(server, "make_stream_renderer", lambda cols: None)
    +    monkeypatch.setattr(server, "render_message", lambda raw, cols: None)
    +    monkeypatch.setattr(
    +        server, "_sync_session_key_after_compress", lambda *a, **kw: None
         )
    -    sid = resp["result"]["session_id"]
    -    session = server._sessions[sid]
    -    session["pending_title"] = "duplicate title"
    -    unblock_agent.set()
    -    session["agent_ready"].wait(timeout=2.0)
    +    monkeypatch.setattr(server.threading, "Thread", _ImmediateThread)
     
    -    assert session["pending_title"] is None
    -    server._sessions.pop(sid, None)
    +    try:
    +        server.handle_request(
    +            {"id": "1", "method": "prompt.submit", "params": {"session_id": "sid", "text": "hello"}}
    +        )
    +        assert session["pending_title"] is None
    +    finally:
    +        server._sessions.pop("sid", None)
     
     
     def test_config_set_yolo_toggles_session_scope():

  • tui_gateway/server.py

    @@ -304,12 +304,14 @@ def _finalize_session(session: dict | None, end_reason: str = "tui_close") -> No
         _notify_session_boundary("on_session_finalize", session_id)
     
         # Mark session ended in DB so it doesn't linger as a ghost row in /resume.
    -    # Adapted from #18283 (luyao618) and #18299 (Bartok9).
    -    if session_key:
    +    # Use session_id (from agent.session_id) not session_key — after compression,
    +    # session_key may be stale (the ended parent) while session_id is the live
    +    # continuation. Fix for #20001.
    +    if session_id:
             try:
                 db = _get_db()
                 if db is not None:
    -                db.end_session(session_key, end_reason)
    +                db.end_session(session_id, end_reason)
             except Exception:
                 pass
     
    @@ -1175,7 +1177,13 @@ def _compress_session_history(
         return len(history) - len(compressed), usage
     
     
    -def _sync_session_key_after_compress(sid: str, session: dict) -> None:
    +def _sync_session_key_after_compress(
    +    sid: str,
    +    session: dict,
    +    *,
    +    clear_pending_title: bool = True,
    +    restart_slash_worker: bool = True,
    +) -> None:
         """Re-anchor session_key when AIAgent._compress_context rotates session_id.
     
         AIAgent._compress_context ends the current SessionDB session and creates
    @@ -1184,7 +1192,14 @@ def _sync_session_key_after_compress(sid: str, session: dict) -> None:
         approval routing, slash worker init, DB title/history lookups, yolo
         state).  Without this sync, those operations would target the ended
         parent session while the agent writes to the new continuation session.
    -    Mirrors HermesCLI._manual_compress's session_id sync.
    +
    +    Policy flags:
    +        clear_pending_title: True for manual /compress (title belongs to old
    +            session). False for post-turn auto-compression (preserve user
    +            intent so pending_title can be applied to the continuation).
    +        restart_slash_worker: True for manual /compress and post-turn
    +            auto-compression (worker holds stale session key). False only
    +            if the caller manages the worker lifecycle separately.
         """
         agent = session.get("agent")
         new_session_id = getattr(agent, "session_id", None) or ""
    @@ -1229,11 +1244,13 @@ def _sync_session_key_after_compress(sid: str, session: dict) -> None:
             # don't keep targeting the ended row.
             session["session_key"] = new_session_id
     
    -    session["pending_title"] = None
    -    try:
    -        _restart_slash_worker(session)
    -    except Exception:
    -        pass
    +    if clear_pending_title:
    +        session["pending_title"] = None
    +    if restart_slash_worker:
    +        try:
    +            _restart_slash_worker(session)
    +        except Exception:
    +            pass
     
     
     def _get_usage(agent) -> dict:
    @@ -2965,6 +2982,17 @@ def _stream(delta):
                                     "History changed during this turn — the response above is visible "
                                     "but was not saved to session history."
                                 )
    +
    +                # If auto-compression fired inside run_conversation(), agent.session_id
    +                # may have rotated. Sync session_key before downstream title/goal/finalize
    +                # handling uses it. Preserve pending_title (user intent) so it can be
    +                # applied to the continuation. Restart slash worker so subsequent
    +                # worker-backed commands (/title etc.) target the live session.
    +                # Fix for #20001.
    +                _sync_session_key_after_compress(
    +                    sid, session, clear_pending_title=False, restart_slash_worker=True,
    +                )
    +
                     raw = result.get("final_response", "")
                     status = (
                         "interrupted"
    @@ -3042,11 +3070,21 @@ def _stream(delta):
                 if _pending and status == "complete":
                     _pdb = _get_db()
                     if _pdb:
    +                    _session_key = session.get("session_key") or sid
                         try:
    -                        if _pdb.set_session_title(session.get("session_key") or sid, _pending):
    +                        if _pdb.set_session_title(_session_key, _pending):
                                 session["pending_title"] = None
    +                    except ValueError as exc:
    +                        # Invalid/duplicate title — non-retryable, drop it.
    +                        # Auto-title will take over. Fix for #19029.
    +                        session["pending_title"] = None
    +                        logger.info(
    +                            "Dropping pending title for session %s: %s",
    +                            _session_key, exc,
    +                        )
                         except Exception:
    -                        pass  # Best effort — auto-title will handle it below
    +                        # Transient DB failure — keep pending_title for retry.
    +                        pass
     
                 if (
                     status == "complete"