diff --git a/api/models.py b/api/models.py index 70c774db..54517a81 100644 --- a/api/models.py +++ b/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, diff --git a/tests/test_session_lost_response_regression.py b/tests/test_session_lost_response_regression.py index 842867e2..6ff50515 100644 --- a/tests/test_session_lost_response_regression.py +++ b/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" + +