feat: add session save mode config

api/config.py

@@ -199,6 +199,10 @@ def _get_config_path() -> Path:
         return HOME / ".hermes" / "config.yaml"
 
 
+_WEBUI_SESSION_SAVE_MODES = {"deferred", "eager"}
+_DEFAULT_WEBUI_SESSION_SAVE_MODE = "deferred"
+
+
 def get_config() -> dict:
     """Return the cached config dict, loading from disk if needed."""
     if not _cfg_cache:
@@ -206,6 +210,28 @@ def get_config() -> dict:
     return _cfg_cache
 
 
+def get_webui_session_save_mode(config_data: dict | None = None) -> str:
+    """Return the validated first-turn session persistence mode.
+
+    ``deferred`` preserves the current first-turn sidecar behaviour: persist
+    pending_user_message/runtime fields before streaming, then merge the turn
+    after the agent finishes. ``eager`` additionally checkpoints the current
+    user turn into ``messages`` before launching the agent thread. Unknown
+    values fail closed to ``deferred`` so a typo never reintroduces eager disk
+    writes unexpectedly.
+    """
+    active_cfg = config_data if isinstance(config_data, dict) else cfg
+    webui_cfg = active_cfg.get("webui", {}) if isinstance(active_cfg, dict) else {}
+    if not isinstance(webui_cfg, dict):
+        return _DEFAULT_WEBUI_SESSION_SAVE_MODE
+    mode = webui_cfg.get("session_save_mode", _DEFAULT_WEBUI_SESSION_SAVE_MODE)
+    if isinstance(mode, str):
+        normalized = mode.strip().lower()
+        if normalized in _WEBUI_SESSION_SAVE_MODES:
+            return normalized
+    return _DEFAULT_WEBUI_SESSION_SAVE_MODE
+
+
 def reload_config() -> None:
     """Reload config.yaml from the active profile's directory."""
     global _cfg_mtime

api/models.py

@@ -629,18 +629,26 @@ def _apply_core_sync_or_error_marker(
     # prompt submitted just before a server restart, so materialize it before
     # clearing runtime stream state.
     if len(session.messages) != 0:
+        _pending_text = " ".join(str(session.pending_user_message or "").split())
+        _already_checkpointed = False
+        if _pending_text and session.messages:
+            _last_msg = session.messages[-1]
+            if isinstance(_last_msg, dict) and _last_msg.get('role') == 'user':
+                _last_text = " ".join(str(_last_msg.get('content') or "").split())
+                _already_checkpointed = _last_text == _pending_text
         _recovered_ts = int(time.time())
         if isinstance(session.pending_started_at, (int, float)) and session.pending_started_at > 0:
             _recovered_ts = int(session.pending_started_at)
-        recovered = {
-            'role': 'user',
-            'content': session.pending_user_message,
-            'timestamp': _recovered_ts,
-            '_recovered': True,
-        }
-        if session.pending_attachments:
-            recovered['attachments'] = list(session.pending_attachments)
-        session.messages.append(recovered)
+        if not _already_checkpointed:
+            recovered = {
+                'role': 'user',
+                'content': session.pending_user_message,
+                'timestamp': _recovered_ts,
+                '_recovered': True,
+            }
+            if session.pending_attachments:
+                recovered['attachments'] = list(session.pending_attachments)
+            session.messages.append(recovered)
         session.active_stream_id = None
         session.pending_user_message = None
         session.pending_attachments = []

api/routes.py

@@ -392,6 +392,7 @@ from api.config import (
     set_reasoning_display,
     set_reasoning_effort,
     create_stream_channel,
+    get_webui_session_save_mode,
 )
 from api.helpers import (
     require,
@@ -5056,6 +5057,68 @@ def _handle_background(handler, body):
     return j(handler, {"task_id": task_id, "stream_id": stream_id, "session_id": bg.session_id})
 
 
+def _checkpoint_user_message_for_eager_session_save(s, msg: str, attachments, started_at: float | None) -> None:
+    """Materialize the current user turn for eager first-turn persistence.
+
+    The streaming thread still receives ``pending_user_message`` so existing
+    cancel/recovery/final-merge paths keep their current contract. Eager mode
+    only adds a durable display-message checkpoint before the agent launches.
+    """
+    if not msg:
+        return
+    existing = list(getattr(s, "messages", None) or [])
+    if existing:
+        latest = existing[-1]
+        if isinstance(latest, dict) and latest.get("role") == "user":
+            latest_text = " ".join(str(latest.get("content") or "").split())
+            msg_text = " ".join(str(msg or "").split())
+            if latest_text == msg_text:
+                return
+    user_msg = {"role": "user", "content": msg}
+    if isinstance(started_at, (int, float)) and started_at > 0:
+        user_msg["timestamp"] = int(started_at)
+    if attachments:
+        user_msg["attachments"] = list(attachments)
+    s.messages.append(user_msg)
+
+
+def _prepare_chat_start_session_for_stream(
+    s,
+    *,
+    msg: str,
+    attachments,
+    workspace: str,
+    model: str,
+    model_provider,
+    stream_id: str,
+    started_at: float | None = None,
+):
+    """Persist chat-start state according to webui.session_save_mode.
+
+    ``deferred`` keeps the existing sidecar/WAL-backed behaviour: save pending
+    fields but leave the display transcript empty until the agent merges the
+    result. ``eager`` additionally writes the current user turn into messages so
+    a process restart immediately after /api/chat/start preserves the prompt as
+    a normal session message. Empty sessions are never saved here because this
+    helper only runs after a non-empty message is validated.
+    """
+    s.workspace = workspace
+    s.model = model
+    s.model_provider = model_provider
+    s.active_stream_id = stream_id
+    s.pending_user_message = msg
+    s.pending_attachments = attachments
+    s.pending_started_at = started_at if started_at is not None else time.time()
+    if get_webui_session_save_mode() == "eager":
+        _checkpoint_user_message_for_eager_session_save(
+            s,
+            msg,
+            attachments,
+            s.pending_started_at,
+        )
+    s.save()
+
+
 def _handle_chat_start(handler, body):
     try:
         require(body, "session_id")
@@ -5103,14 +5166,15 @@ def _handle_chat_start(handler, body):
         _clear_stale_stream_state(s)
     stream_id = uuid.uuid4().hex
     with _get_session_agent_lock(s.session_id):
-        s.workspace = workspace
-        s.model = model
-        s.model_provider = model_provider
-        s.active_stream_id = stream_id
-        s.pending_user_message = msg
-        s.pending_attachments = attachments
-        s.pending_started_at = time.time()
-        s.save()
+        _prepare_chat_start_session_for_stream(
+            s,
+            msg=msg,
+            attachments=attachments,
+            workspace=workspace,
+            model=model,
+            model_provider=model_provider,
+            stream_id=stream_id,
+        )
     set_last_workspace(workspace)
     stream = create_stream_channel()
     with STREAMS_LOCK:

api/streaming.py

@@ -1164,6 +1164,17 @@ def _find_current_user_turn(messages, msg_text):
     return fallback
 
 
+def _drop_checkpointed_current_user_from_context(messages, msg_text):
+    """Return model history without an eager-checkpointed current user turn."""
+    history = list(messages or [])
+    if not history:
+        return history
+    current_user_key = _message_identity({'role': 'user', 'content': msg_text})
+    if current_user_key and _message_identity(history[-1]) == current_user_key:
+        return history[:-1]
+    return history
+
+
 def _merge_display_messages_after_agent_result(previous_display, previous_context, result_messages, msg_text):
     """Keep UI transcript durable while allowing model context to compact.
 
@@ -1191,8 +1202,20 @@ def _merge_display_messages_after_agent_result(previous_display, previous_contex
 
     merged = previous_display[:]
     seen = {_message_identity(m) for m in merged}
+    current_user_key = _message_identity({'role': 'user', 'content': msg_text})
     for msg in candidates:
         key = _message_identity(msg)
+        if (
+            key is not None
+            and key == current_user_key
+            and merged
+            and _message_identity(merged[-1]) == key
+        ):
+            # Eager session-save mode can checkpoint the current user turn
+            # before the agent runs. When the agent returns that same user turn
+            # in result_messages, keep the durable checkpoint and append only
+            # the assistant/tool delta.
+            continue
         if _is_context_compression_marker(msg) and key is not None and key in seen:
             continue
         merged.append(copy.deepcopy(msg))
@@ -2059,7 +2082,10 @@ def _run_agent_streaming(
             # Truthy-check covers None, missing-attr, and 0 uniformly.
             _turn_started_at = _pending_started_at if _pending_started_at else time.time()
             _previous_messages = list(s.messages or [])
-            _previous_context_messages = list(_session_context_messages(s))
+            _previous_context_messages = _drop_checkpointed_current_user_from_context(
+                _session_context_messages(s),
+                msg_text,
+            )
             _pre_compression_count = getattr(
                 getattr(agent, 'context_compressor', None),
                 'compression_count', 0,

tests/test_session_save_mode.py

@@ -0,0 +1,132 @@
+"""Regression tests for config-driven first-turn session persistence (#1406)."""
+import json
+
+import pytest
+
+import api.config as config
+import api.models as models
+import api.routes as routes
+import api.streaming as streaming
+from api.models import Session, new_session
+
+
+@pytest.fixture(autouse=True)
+def _isolate_state(tmp_path, monkeypatch):
+    session_dir = tmp_path / "sessions"
+    session_dir.mkdir()
+    index_file = session_dir / "_index.json"
+    monkeypatch.setattr(models, "SESSION_DIR", session_dir)
+    monkeypatch.setattr(models, "SESSION_INDEX_FILE", index_file)
+    monkeypatch.setattr(config, "SESSION_INDEX_FILE", index_file, raising=False)
+    models.SESSIONS.clear()
+    config.STREAMS.clear()
+    config.CANCEL_FLAGS.clear()
+    config.AGENT_INSTANCES.clear()
+    config.SESSION_AGENT_LOCKS.clear()
+    monkeypatch.setattr(config, "cfg", {})
+    monkeypatch.setattr(config, "_cfg_cache", {})
+    yield session_dir
+    models.SESSIONS.clear()
+    config.STREAMS.clear()
+    config.CANCEL_FLAGS.clear()
+    config.AGENT_INSTANCES.clear()
+    config.SESSION_AGENT_LOCKS.clear()
+
+
+def test_session_save_mode_defaults_to_deferred_for_missing_config():
+    assert config.get_webui_session_save_mode({}) == "deferred"
+    assert config.get_webui_session_save_mode({"webui": {}}) == "deferred"
+
+
+@pytest.mark.parametrize("raw", ["bogus", "", None, 42, {"mode": "eager"}])
+def test_invalid_session_save_mode_falls_back_to_deferred(raw):
+    assert config.get_webui_session_save_mode({"webui": {"session_save_mode": raw}}) == "deferred"
+
+
+def test_eager_session_save_mode_is_accepted():
+    assert config.get_webui_session_save_mode({"webui": {"session_save_mode": "eager"}}) == "eager"
+
+
+def test_eager_mode_still_does_not_save_empty_new_sessions(_isolate_state, monkeypatch):
+    monkeypatch.setattr(config, "cfg", {"webui": {"session_save_mode": "eager"}})
+    s = new_session()
+    assert not s.path.exists(), "eager mode must not recreate empty Untitled session files"
+
+
+def test_deferred_chat_start_persists_pending_only_before_thread(_isolate_state, monkeypatch):
+    monkeypatch.setattr(config, "cfg", {"webui": {"session_save_mode": "deferred"}})
+    s = new_session(workspace=str(_isolate_state.parent))
+    routes._prepare_chat_start_session_for_stream(
+        s,
+        msg="hello deferred",
+        attachments=[],
+        workspace=str(_isolate_state.parent),
+        model=s.model,
+        model_provider=s.model_provider,
+        stream_id="stream_deferred",
+        started_at=123.0,
+    )
+    on_disk = json.loads(s.path.read_text(encoding="utf-8"))
+    assert on_disk["messages"] == []
+    assert on_disk["pending_user_message"] == "hello deferred"
+
+
+def test_eager_chat_start_checkpoints_first_user_message_before_thread(_isolate_state, monkeypatch):
+    monkeypatch.setattr(config, "cfg", {"webui": {"session_save_mode": "eager"}})
+    s = new_session(workspace=str(_isolate_state.parent))
+    routes._prepare_chat_start_session_for_stream(
+        s,
+        msg="hello eager",
+        attachments=[{"name": "note.txt", "path": "", "mime": "text/plain"}],
+        workspace=str(_isolate_state.parent),
+        model=s.model,
+        model_provider=s.model_provider,
+        stream_id="stream_eager",
+        started_at=456.0,
+    )
+    on_disk = json.loads(s.path.read_text(encoding="utf-8"))
+    assert [m["role"] for m in on_disk["messages"]] == ["user"]
+    assert on_disk["messages"][0]["content"] == "hello eager"
+    assert on_disk["messages"][0]["attachments"][0]["name"] == "note.txt"
+    assert on_disk["pending_user_message"] == "hello eager"
+
+
+def test_eager_wal_repair_does_not_duplicate_checkpointed_user_message(_isolate_state, monkeypatch):
+    s = Session(session_id="eager_repair", messages=[{"role": "user", "content": "survive"}])
+    s.pending_user_message = "survive"
+    s.active_stream_id = "dead_stream"
+    s.pending_started_at = 789.0
+    s.save()
+
+    repaired = models._repair_stale_pending(s)
+
+    assert repaired is True
+    user_messages = [m for m in s.messages if m.get("role") == "user" and m.get("content") == "survive"]
+    assert len(user_messages) == 1
+    assert s.pending_user_message is None
+    assert any(m.get("_error") for m in s.messages if m.get("role") == "assistant")
+
+
+def test_eager_checkpointed_user_is_removed_from_model_context():
+    context = streaming._drop_checkpointed_current_user_from_context(
+        [
+            {"role": "user", "content": "older"},
+            {"role": "assistant", "content": "prior"},
+            {"role": "user", "content": "current"},
+        ],
+        "current",
+    )
+    assert [m["content"] for m in context] == ["older", "prior"]
+
+
+def test_eager_checkpointed_user_is_not_duplicated_after_agent_result():
+    merged = streaming._merge_display_messages_after_agent_result(
+        previous_display=[{"role": "user", "content": "repeat me"}],
+        previous_context=[],
+        result_messages=[
+            {"role": "user", "content": "repeat me"},
+            {"role": "assistant", "content": "ok"},
+        ],
+        msg_text="repeat me",
+    )
+    assert [m["role"] for m in merged] == ["user", "assistant"]