Merge #5274 (pr/5270-cli-webui-continuity) into stage-w1

api/models.py

@@ -5627,6 +5627,17 @@ def _session_message_merge_key(msg: dict):
     )
 
 
+def _session_messages_have_prefix(messages, prefix) -> bool:
+    messages = list(messages or [])
+    prefix = list(prefix or [])
+    if len(prefix) > len(messages):
+        return False
+    for idx, expected in enumerate(prefix):
+        if _session_message_merge_key(messages[idx]) != _session_message_merge_key(expected):
+            return False
+    return True
+
+
 _SESSION_MESSAGE_DISPLAY_METADATA_KEYS = (
     "_turnDuration",
     "_turnTps",
@@ -6457,23 +6468,39 @@ def reconciled_state_db_messages_for_session(
         state_messages = get_state_db_session_messages(getattr(session, 'session_id', None))
     if prefer_context and local_messages:
         if using_context_messages:
-            compressed_context = _context_messages_include_compression_marker(local_messages)
-            anchor_key = getattr(session, "compression_anchor_message_key", None)
-            if compressed_context:
-                if not anchor_key:
-                    logger.debug(
-                        "Compressed context for session %s has no compression anchor; using context_messages only",
-                        getattr(session, "session_id", None),
-                    )
-                    return list(local_messages)
-                anchor_index = _state_db_anchor_index(state_messages, anchor_key)
-                if anchor_index is None:
-                    logger.debug(
-                        "Compressed context for session %s has an unverifiable compression anchor; using context_messages only",
-                        getattr(session, "session_id", None),
-                    )
-                    return list(local_messages)
-                state_messages = list(state_messages or [])[anchor_index + 1 :]
+            sidecar_messages = getattr(session, 'messages', None) or []
+            if (
+                getattr(session, 'is_cli_session', False)
+                and not getattr(session, 'read_only', False)
+                and sidecar_messages
+                and len(sidecar_messages) > len(local_messages)
+                and _session_messages_have_prefix(sidecar_messages, local_messages)
+            ):
+                # A claimed CLI sidecar can carry a stale context prefix while the
+                # stitched CLI transcript already landed in session.messages. On the
+                # first WebUI follow-up, prefer that longer authoritative transcript
+                # unless context_messages intentionally diverged via compaction or
+                # another non-prefix transform.
+                local_messages = sidecar_messages
+                using_context_messages = False
+            if using_context_messages:
+                compressed_context = _context_messages_include_compression_marker(local_messages)
+                anchor_key = getattr(session, "compression_anchor_message_key", None)
+                if compressed_context:
+                    if not anchor_key:
+                        logger.debug(
+                            "Compressed context for session %s has no compression anchor; using context_messages only",
+                            getattr(session, "session_id", None),
+                        )
+                        return list(local_messages)
+                    anchor_index = _state_db_anchor_index(state_messages, anchor_key)
+                    if anchor_index is None:
+                        logger.debug(
+                            "Compressed context for session %s has an unverifiable compression anchor; using context_messages only",
+                            getattr(session, "session_id", None),
+                        )
+                        return list(local_messages)
+                    state_messages = list(state_messages or [])[anchor_index + 1 :]
         state_messages = state_db_delta_after_context(local_messages, state_messages)
     return merge_session_messages_append_only(
         local_messages,

api/routes.py

@@ -4418,7 +4418,7 @@ def _handle_session_anchor_scene(handler, body):
     return j(handler, {"ok": True, "message_index": idx, "message_ref": ref})
 
 
-def _get_or_materialize_session(sid: str):
+def _get_or_materialize_session(sid: str, *, refresh_cli_messages: bool = False):
     """Get a session, materializing from CLI/agent metadata if not in WebUI store.
 
     Mirrors the fallback logic in /api/session/archive (routes.py:~8530).
@@ -4437,6 +4437,21 @@ def _get_or_materialize_session(sid: str):
         # below (and the heuristic record-check would mis-trip on mock sessions).
         if getattr(s, "read_only", False):
             raise PermissionError("read-only imported session")
+        if refresh_cli_messages and getattr(s, "is_cli_session", False):
+            latest_messages = get_cli_session_messages(
+                sid,
+                profile=getattr(s, "profile", None),
+            )
+            current_messages = list(getattr(s, "messages", None) or [])
+            if (
+                latest_messages
+                and len(latest_messages) >= len(current_messages)
+                and _session_messages_have_prefix(latest_messages, current_messages)
+            ):
+                # Keep the stitched CLI transcript authoritative on the first
+                # WebUI continuation path without clobbering later divergent
+                # WebUI-owned turns.
+                s.messages = list(latest_messages)
         return s
     except KeyError:
         pass
@@ -7559,6 +7574,7 @@ from api.models import (
     _active_stream_ids,
     _merge_session_display_metadata,
     _session_message_merge_key,
+    _session_messages_have_prefix,
     _session_message_visible_key,
     _message_timestamp_as_float,
     _is_empty_partial_activity_message,
@@ -18116,7 +18132,7 @@ def _handle_chat_start(handler, body, diag=None):
             return bad(handler, str(e))
         diag.stage("get_session") if diag else None
         try:
-            s = _get_or_materialize_session(body["session_id"])
+            s = _get_or_materialize_session(body["session_id"], refresh_cli_messages=True)
         except KeyError:
             # No WebUI sidecar. If this is a foreign-origin session (CLI,
             # TUI, Desktop) with recoverable state.db messages, claim it by

tests/test_issue3975_cron_reply_materialization.py

@@ -47,7 +47,7 @@ def test_chat_start_materializes_cron_session_before_reply(monkeypatch, tmp_path
     def fail_get_session(_sid):
         raise KeyError(_sid)
 
-    def materialize(_sid):
+    def materialize(_sid, **_kwargs):
         captured["materialize_sid"] = _sid
         return materialized
 

tests/test_issue5270_cli_webui_continuity.py

@@ -0,0 +1,315 @@
+from __future__ import annotations
+
+import queue
+import sqlite3
+from collections import OrderedDict
+from pathlib import Path
+
+import pytest
+
+
+pytestmark = pytest.mark.requires_agent_modules
+
+
+CLI_PROMPT = "Reply with exactly CLI-ORIGIN-OK and nothing else."
+CLI_REPLY = "CLI-ORIGIN-OK"
+WEBUI_FOLLOWUP = "What exact string did you just reply with? Answer with only that string."
+
+
+def _make_cli_continuation_state_db(path: Path, *, parent_sid: str, child_sid: str, workspace: str) -> None:
+    conn = sqlite3.connect(path)
+    conn.execute(
+        """
+        CREATE TABLE sessions (
+            id TEXT PRIMARY KEY,
+            source TEXT,
+            title TEXT,
+            model TEXT,
+            cwd TEXT,
+            started_at REAL,
+            ended_at REAL,
+            end_reason TEXT,
+            parent_session_id TEXT,
+            message_count INTEGER
+        )
+        """
+    )
+    conn.execute(
+        """
+        CREATE TABLE messages (
+            id INTEGER PRIMARY KEY AUTOINCREMENT,
+            session_id TEXT,
+            role TEXT,
+            content TEXT,
+            timestamp REAL
+        )
+        """
+    )
+    conn.execute(
+        """
+        INSERT INTO sessions (
+            id, source, title, model, cwd, started_at, ended_at, end_reason, parent_session_id, message_count
+        ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
+        """,
+        (parent_sid, "cli", "CLI parent", "test-model", workspace, 1.0, 2.0, "cli_close", None, 2),
+    )
+    conn.execute(
+        """
+        INSERT INTO sessions (
+            id, source, title, model, cwd, started_at, ended_at, end_reason, parent_session_id, message_count
+        ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
+        """,
+        (child_sid, "cli", "CLI child", "test-model", workspace, 3.0, None, None, parent_sid, 0),
+    )
+    conn.execute(
+        "INSERT INTO messages (session_id, role, content, timestamp) VALUES (?, ?, ?, ?)",
+        (parent_sid, "user", CLI_PROMPT, 1.0),
+    )
+    conn.execute(
+        "INSERT INTO messages (session_id, role, content, timestamp) VALUES (?, ?, ?, ?)",
+        (parent_sid, "assistant", CLI_REPLY, 2.0),
+    )
+    conn.commit()
+    conn.close()
+
+
+def _install_cli_continuity_env(monkeypatch, tmp_path):
+    import api.config as config
+    import api.models as models
+    import api.profiles as profiles
+    import api.routes as routes
+    import api.streaming as streaming
+
+    session_dir = tmp_path / "sessions"
+    session_dir.mkdir()
+    index_file = session_dir / "_index.json"
+    state_db = tmp_path / "state.db"
+
+    monkeypatch.setattr(models, "SESSION_DIR", session_dir)
+    monkeypatch.setattr(models, "SESSION_INDEX_FILE", index_file)
+    monkeypatch.setattr(models, "SESSIONS", OrderedDict(), raising=False)
+    monkeypatch.setattr(models, "_active_state_db_path", lambda: state_db, raising=False)
+    monkeypatch.setattr(config, "SESSION_DIR", session_dir, raising=False)
+    monkeypatch.setattr(config, "SESSION_INDEX_FILE", index_file, raising=False)
+    monkeypatch.setattr(streaming, "SESSION_DIR", session_dir, raising=False)
+    monkeypatch.setattr(profiles, "get_active_hermes_home", lambda: tmp_path, raising=False)
+    monkeypatch.setattr(routes, "SESSION_INDEX_FILE", index_file, raising=False)
+
+    config.STREAMS.clear()
+    config.CANCEL_FLAGS.clear()
+    config.AGENT_INSTANCES.clear()
+    config.SESSION_AGENT_LOCKS.clear()
+    streaming.STREAMS.clear()
+    streaming.CANCEL_FLAGS.clear()
+    streaming.AGENT_INSTANCES.clear()
+    streaming.STREAM_PARTIAL_TEXT.clear()
+    streaming.STREAM_REASONING_TEXT.clear()
+    streaming.STREAM_LIVE_TOOL_CALLS.clear()
+
+    return config, models, routes, streaming, state_db
+
+
+def _capture_streaming_history(monkeypatch, config, streaming, session, *, stream_id: str, tmp_path: Path):
+    captured: dict[str, list] = {}
+
+    class FakeAgent:
+        def __init__(self, **_kwargs):
+            self.session_id = session.session_id
+            self.context_compressor = None
+            self.ephemeral_system_prompt = None
+            self.session_prompt_tokens = 0
+            self.session_completion_tokens = 0
+            self.session_estimated_cost_usd = None
+            self.session_cache_read_tokens = 0
+            self.session_cache_write_tokens = 0
+            self._last_error = None
+
+        def run_conversation(self, **kwargs):
+            history = list(kwargs.get("conversation_history") or [])
+            captured["conversation_history"] = history
+            return {
+                "completed": True,
+                "final_response": "ok",
+                "messages": history + [
+                    {"role": "user", "content": kwargs.get("persist_user_message", "")},
+                    {"role": "assistant", "content": "ok"},
+                ],
+            }
+
+        def interrupt(self, _message):
+            return None
+
+    monkeypatch.setattr(streaming, "_get_ai_agent", lambda: FakeAgent)
+    monkeypatch.setattr(streaming, "resolve_model_provider", lambda *args, **kwargs: ("test-model", None, None))
+    monkeypatch.setattr(streaming, "get_config", lambda: {})
+    monkeypatch.setattr(config, "get_config", lambda: {})
+    monkeypatch.setattr(config, "_resolve_cli_toolsets", lambda *args, **kwargs: [])
+
+    session.active_stream_id = stream_id
+    session.pending_user_message = WEBUI_FOLLOWUP
+    session.pending_started_at = 10.0
+    session.save(touch_updated_at=False)
+    config.STREAMS[stream_id] = queue.Queue()
+    try:
+        streaming._run_agent_streaming(
+            session_id=session.session_id,
+            msg_text=WEBUI_FOLLOWUP,
+            model="test-model",
+            workspace=str(tmp_path),
+            stream_id=stream_id,
+            attachments=[],
+        )
+    finally:
+        config.STREAMS.pop(stream_id, None)
+    return captured.get("conversation_history") or []
+
+
+def _history_contents(history):
+    return [msg.get("content") for msg in history if isinstance(msg, dict)]
+
+
+def test_first_webui_followup_receives_immediate_cli_assistant_context(monkeypatch, tmp_path):
+    config, models, routes, streaming, state_db = _install_cli_continuity_env(monkeypatch, tmp_path)
+    sid = "issue5270_cli_child_fresh"
+    _make_cli_continuation_state_db(
+        state_db,
+        parent_sid="issue5270_cli_parent_fresh",
+        child_sid=sid,
+        workspace=str(tmp_path),
+    )
+
+    session, reason = routes._claim_or_synthesize_cli_session(sid)
+
+    assert reason == "materialized"
+    assert session is not None
+    assert session.read_only is False
+
+    session.save(touch_updated_at=False)
+    models.SESSIONS[sid] = session
+    history = _capture_streaming_history(
+        monkeypatch,
+        config,
+        streaming,
+        session,
+        stream_id="stream-issue5270-fresh",
+        tmp_path=tmp_path,
+    )
+
+    assert _history_contents(history) == [CLI_PROMPT, CLI_REPLY]
+
+
+def test_already_claimed_cli_sidecar_still_sees_cli_prior_assistant_on_first_webui_turn(monkeypatch, tmp_path):
+    config, models, routes, streaming, state_db = _install_cli_continuity_env(monkeypatch, tmp_path)
+    sid = "issue5270_cli_child_claimed"
+    _make_cli_continuation_state_db(
+        state_db,
+        parent_sid="issue5270_cli_parent_claimed",
+        child_sid=sid,
+        workspace=str(tmp_path),
+    )
+
+    stale_sidecar = models.Session(
+        session_id=sid,
+        title="Claimed CLI child",
+        workspace=str(tmp_path),
+        model="test-model",
+        messages=[
+            {"role": "user", "content": CLI_PROMPT, "timestamp": 1.0},
+        ],
+        context_messages=[
+            {"role": "user", "content": CLI_PROMPT, "timestamp": 1.0},
+        ],
+        is_cli_session=True,
+        source_tag="cli",
+        raw_source="cli",
+        session_source="cli",
+        source_label="CLI",
+        read_only=False,
+    )
+    stale_sidecar.save(touch_updated_at=False)
+    models.SESSIONS[sid] = stale_sidecar
+
+    session = routes._get_or_materialize_session(sid, refresh_cli_messages=True)
+    history = _capture_streaming_history(
+        monkeypatch,
+        config,
+        streaming,
+        session,
+        stream_id="stream-issue5270-claimed",
+        tmp_path=tmp_path,
+    )
+
+    assert _history_contents(history) == [CLI_PROMPT, CLI_REPLY]
+
+
+def test_chat_start_refreshes_cli_messages_before_first_webui_turn(monkeypatch, tmp_path):
+    _config, models, routes, _streaming, _state_db = _install_cli_continuity_env(monkeypatch, tmp_path)
+
+    session = models.Session(
+        session_id="issue5270_cli_child_chat_start",
+        title="Claimed CLI child",
+        workspace=str(tmp_path),
+        model="test-model",
+        messages=[{"role": "user", "content": CLI_PROMPT, "timestamp": 1.0}],
+        context_messages=[{"role": "user", "content": CLI_PROMPT, "timestamp": 1.0}],
+        is_cli_session=True,
+        source_tag="cli",
+        raw_source="cli",
+        session_source="cli",
+        source_label="CLI",
+        read_only=False,
+    )
+
+    seen: dict[str, bool] = {}
+
+    def _fake_get_or_materialize_session(sid, *, refresh_cli_messages=False):
+        seen["refresh_cli_messages"] = refresh_cli_messages
+        assert sid == session.session_id
+        return session
+
+    def _fake_start_run(s, **kwargs):
+        assert s is session
+        assert seen["refresh_cli_messages"] is True
+        return {"ok": True}
+
+    monkeypatch.setattr(routes, "_get_or_materialize_session", _fake_get_or_materialize_session)
+    monkeypatch.setattr(routes, "_session_visible_to_active_profile", lambda *args, **kwargs: True)
+    monkeypatch.setattr(routes, "_resolve_chat_workspace_with_recovery", lambda *args, **kwargs: str(tmp_path))
+    monkeypatch.setattr(routes, "_read_profile_model_config", lambda *args, **kwargs: (None, None))
+    monkeypatch.setattr(
+        routes,
+        "_resolve_compatible_session_model_state",
+        lambda *args, **kwargs: ("test-model", None, "test-model"),
+    )
+    monkeypatch.setattr(routes, "_start_run", _fake_start_run)
+    monkeypatch.setattr(routes, "j", lambda _handler, payload, status=200: {"status": status, **payload})
+
+    response = routes._handle_chat_start(
+        None,
+        {
+            "session_id": session.session_id,
+            "message": WEBUI_FOLLOWUP,
+        },
+    )
+
+    assert seen["refresh_cli_messages"] is True
+    assert response["ok"] is True
+    assert response["status"] == 200
+
+
+def test_regular_cli_sessions_remain_writable_after_fix(monkeypatch, tmp_path):
+    _config, _models, routes, _streaming, state_db = _install_cli_continuity_env(monkeypatch, tmp_path)
+    sid = "issue5270_cli_child_writable"
+    _make_cli_continuation_state_db(
+        state_db,
+        parent_sid="issue5270_cli_parent_writable",
+        child_sid=sid,
+        workspace=str(tmp_path),
+    )
+
+    session, reason = routes._claim_or_synthesize_cli_session(sid)
+
+    assert reason == "materialized"
+    assert session is not None
+    assert session.read_only is False
+    assert session.is_cli_session is True

tests/test_session_active_profile_authorization.py

@@ -164,7 +164,7 @@ def test_chat_start_foreign_persisted_session_returns_404_before_start_run(monke
 
     monkeypatch.setattr(routes, "_check_csrf", lambda _handler: True)
     monkeypatch.setattr(routes, "read_body", lambda _handler: {"session_id": "chat_foreign", "message": "hello"})
-    monkeypatch.setattr(routes, "_get_or_materialize_session", lambda sid: persisted_foreign)
+    monkeypatch.setattr(routes, "_get_or_materialize_session", lambda sid, **_kwargs: persisted_foreign)
     monkeypatch.setattr(routes, "_get_active_profile_name", lambda: "default")
     monkeypatch.setattr(routes, "_start_run", lambda *_, **__: (_ for _ in ()).throw(AssertionError("_start_run should not run")))
 
@@ -185,7 +185,7 @@ def test_chat_start_body_profile_cannot_retag_visible_empty_session_without_acti
         "read_body",
         lambda _handler: {"session_id": "chat_empty", "message": "hello", "profile": "other"},
     )
-    monkeypatch.setattr(routes, "_get_or_materialize_session", lambda sid: empty_visible)
+    monkeypatch.setattr(routes, "_get_or_materialize_session", lambda sid, **_kwargs: empty_visible)
     monkeypatch.setattr(routes, "_get_active_profile_name", lambda: "default")
     monkeypatch.setattr(routes, "_resolve_chat_workspace_with_recovery", lambda *_args, **_kwargs: "/workspace")
     monkeypatch.setattr(routes, "_read_profile_model_config", lambda *_args, **_kwargs: (None, None))