diff --git a/api/config.py b/api/config.py index adb4670a..e5fdc0fc 100644 --- a/api/config.py +++ b/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 diff --git a/api/models.py b/api/models.py index 867beb37..33e736b2 100644 --- a/api/models.py +++ b/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 = [] diff --git a/api/routes.py b/api/routes.py index 47a2b12e..5a0ae736 100644 --- a/api/routes.py +++ b/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: diff --git a/api/streaming.py b/api/streaming.py index 1ff71227..c206e032 100644 --- a/api/streaming.py +++ b/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, diff --git a/docs/pr-media/1406/eager-config-app-shell.png b/docs/pr-media/1406/eager-config-app-shell.png new file mode 100644 index 00000000..871741ea Binary files /dev/null and b/docs/pr-media/1406/eager-config-app-shell.png differ diff --git a/tests/test_session_save_mode.py b/tests/test_session_save_mode.py new file mode 100644 index 00000000..32759574 --- /dev/null +++ b/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"]