fix(webui): address greptile P1 - snapshot-first save, safer recovery gate

api/models.py

@@ -2379,30 +2379,56 @@ def _sync_sidecar_from_state_db_if_newer(session) -> bool:
         session,
         state_messages=state_messages,
     )
+    # The reconciler is append-only: when state.db genuinely advances the
+    # transcript (new assistant/tool output that the lost stream never wrote
+    # back), that shows up as MORE rows than the sidecar. A merged length that
+    # is not greater than the sidecar means there is nothing new to recover, so
+    # leave the sidecar untouched rather than rewriting it in place.
     if len(merged_messages) <= sidecar_count:
         return False
-
-    # Avoid claiming a stream is still active after we have incorporated rows
-    # from state.db that are newer than the pending turn. The stream may still
-    # lack a journal terminal event, but the durable transcript is no longer the
-    # stale sidecar; clearing pending state lets refresh/loadSession render the
-    # reconciled conversation instead of treating it as an unfinished run forever.
-    session.messages = merged_messages
-    session.context_messages = reconciled_state_db_messages_for_session(
+    merged_context = reconciled_state_db_messages_for_session(
         session,
         prefer_context=True,
         state_messages=state_messages,
     )
+
+    # Persist FIRST on a private snapshot so a failed write never mutates the
+    # shared (and cached) Session object. Mutating ``session`` before the save
+    # succeeds means a raising ``save()`` would leave the in-memory object with
+    # cleared active_stream_id/pending fields while the on-disk sidecar still
+    # holds the unfinished stream — later get_session() reads in this process
+    # would then skip recovery state that still exists durably. (greptile P1:
+    # save failure mutates cache) Only after the durable write succeeds do we
+    # copy the reconciled state back onto the live object.
+    try:
+        snapshot = Session.load(sid)
+    except Exception:
+        snapshot = None
+    if snapshot is None:
+        logger.debug("state.db newer-sidecar sync: snapshot load failed for session %s", sid, exc_info=True)
+        return False
+    snapshot.messages = merged_messages
+    snapshot.context_messages = merged_context
+    snapshot.active_stream_id = None
+    snapshot.pending_user_message = None
+    snapshot.pending_attachments = []
+    snapshot.pending_started_at = None
+    snapshot.pending_user_source = None
+    try:
+        snapshot.save(touch_updated_at=True)
+    except Exception:
+        logger.debug("state.db newer-sidecar sync save failed for session %s", sid, exc_info=True)
+        return False
+
+    # Durable write succeeded — now it is safe to reflect the reconciled state
+    # on the shared/cached object the caller is holding.
+    session.messages = merged_messages
+    session.context_messages = merged_context
     session.active_stream_id = None
     session.pending_user_message = None
     session.pending_attachments = []
     session.pending_started_at = None
     session.pending_user_source = None
-    try:
-        session.save(touch_updated_at=True)
-    except Exception:
-        logger.debug("state.db newer-sidecar sync save failed for session %s", sid, exc_info=True)
-        return False
     logger.info(
         "Session %s: synced sidecar from newer state.db transcript (%d -> %d messages)",
         sid,

tests/test_session_lost_response_regression.py

@@ -870,3 +870,111 @@ def test_get_session_does_not_sync_while_stream_is_still_live(monkeypatch):
     assert loaded.active_stream_id == stream_id
     assert loaded.pending_user_message == "new request"
     assert [m["content"] for m in loaded.messages] == ["old question", "old answer"]
+
+
+def test_sync_save_failure_does_not_mutate_session(monkeypatch):
+    """If the durable snapshot save raises, the live object must stay untouched.
+
+    The sync mutates the shared/cached Session only AFTER persisting a private
+    snapshot. A failing save must leave active_stream_id/pending fields intact
+    so later reads still see the unfinished stream that is still on disk.
+    Tests the helper directly to isolate it from _repair_stale_pending, which
+    has its own (separate) dead-stream pending cleanup. (greptile P1)
+    """
+    sid = "state_sync_save_fail_sid"
+    stream_id = "stream_save_fail_no_done"
+    s = Session(
+        session_id=sid,
+        title="Save failure",
+        messages=[
+            {"role": "user", "content": "old question", "timestamp": 100.0},
+            {"role": "assistant", "content": "old answer", "timestamp": 101.0},
+        ],
+        active_stream_id=stream_id,
+        pending_user_message="new request",
+        pending_started_at=102.0,
+    )
+    s.save()
+
+    state_messages = [
+        {"role": "user", "content": "old question", "timestamp": 100.0},
+        {"role": "assistant", "content": "old answer", "timestamp": 101.0},
+        {"role": "user", "content": "new request", "timestamp": 102.0},
+        {"role": "assistant", "content": "recovered text", "timestamp": 103.0},
+    ]
+    monkeypatch.setattr(
+        models,
+        "get_state_db_session_summary",
+        lambda sid_arg, profile=None: {"message_count": len(state_messages), "last_message_at": 103.0},
+    )
+    monkeypatch.setattr(
+        models,
+        "get_state_db_session_messages",
+        lambda sid_arg, **kwargs: list(state_messages),
+    )
+
+    # Force the snapshot save to fail.
+    def boom(self, *args, **kwargs):
+        raise OSError("disk full")
+
+    monkeypatch.setattr(Session, "save", boom)
+
+    result = models._sync_sidecar_from_state_db_if_newer(s)
+
+    # Sync reports failure and the live object is NOT mutated.
+    assert result is False
+    assert s.active_stream_id == stream_id
+    assert s.pending_user_message == "new request"
+    assert [m["content"] for m in s.messages] == ["old question", "old answer"]
+
+
+def test_sync_persists_recovered_state_db_tail_when_stream_dead(monkeypatch):
+    """Happy path: dead stream + newer state.db tail is written back durably.
+
+    Mutates the live object only after the snapshot save succeeds, and the
+    reconciled transcript reaches both memory and disk.
+    """
+    sid = "state_sync_ok_sid"
+    stream_id = "stream_dead_no_done"
+    s = Session(
+        session_id=sid,
+        title="Recover tail",
+        messages=[
+            {"role": "user", "content": "old question", "timestamp": 100.0},
+            {"role": "assistant", "content": "old answer", "timestamp": 101.0},
+        ],
+        active_stream_id=stream_id,
+        pending_user_message="new request",
+        pending_started_at=102.0,
+    )
+    s.save()
+
+    state_messages = [
+        {"role": "user", "content": "old question", "timestamp": 100.0},
+        {"role": "assistant", "content": "old answer", "timestamp": 101.0},
+        {"role": "user", "content": "new request", "timestamp": 102.0},
+        {"role": "assistant", "content": "recovered tail", "timestamp": 103.0},
+    ]
+    monkeypatch.setattr(
+        models,
+        "get_state_db_session_summary",
+        lambda sid_arg, profile=None: {"message_count": len(state_messages), "last_message_at": 103.0},
+    )
+    monkeypatch.setattr(
+        models,
+        "get_state_db_session_messages",
+        lambda sid_arg, **kwargs: list(state_messages),
+    )
+
+    result = models._sync_sidecar_from_state_db_if_newer(s)
+
+    assert result is True
+    assert s.active_stream_id is None
+    assert s.pending_user_message is None
+    assert [m["content"] for m in s.messages] == [m["content"] for m in state_messages]
+
+    reloaded = Session.load(sid)
+    assert reloaded.active_stream_id is None
+    assert reloaded.messages[-1]["content"] == "recovered tail"
+
+