Release WI (v0.51.628): no ghost messages after edit/retry/undo (#4772, closes #4767)

Release WI (v0.51.628): no ghost messages after edit/retry/undo (#4772, closes #4767)

CHANGELOG.md

@@ -3,6 +3,12 @@
 
 ## [Unreleased]
 
+## [v0.51.628] — 2026-06-24 — Release WI (no ghost messages after edit/retry/undo)
+
+### Fixed
+
+- **Editing a message (or `/retry` / `/undo`) no longer resurrects the replaced messages — and stops feeding them back into the agent's context.** After an edit/retry/undo, switching away and back used to bring the original pre-edit message and its reply back into the transcript (and into the model's context window, polluting the conversation). The append-only state.db merge now ADVANCES the truncation watermark to the new turn instead of clearing it, and persists the original truncate cutoff (`truncation_boundary`) separately so cold-load/crash recovery can tell a legitimate prefix from a deleted suffix — instead of guessing. Covers editing an older message that leaves several later turns, same-second edits, and the recovery window where the new turn is saved but its reply only exists in state.db. The `#2914` truncate-to-empty block and `#3346` repeated-identical-turn handling are preserved. Thanks @AlexeyDsov. (#4772, closes #4767)
+
 ## [v0.51.627] — 2026-06-24 — Release WH (Tasks cron APIs survive a shadowing top-level `cron` package)
 
 ### Fixed

api/models.py

@@ -428,12 +428,15 @@ def _append_recovered_pending_turn(session, *, timestamp: int | None = None) ->
         recovered['attachments'] = list(session.pending_attachments)
     session.messages.append(recovered)
     _append_recovered_turn_to_context(session, recovered)
-    # The new user turn is now committed to messages (#3831): retire a positive
-    # truncation watermark from a prior retry/undo/edit so it can't freeze at the
-    # old edit boundary and drop these post-edit turns on a later empty-sidecar
-    # reconcile. None, never 0.0 (the truncate-to-empty sentinel, #2914).
+    # The new user turn is now committed to messages (#3831): advance the
+    # truncation watermark to the new message's timestamp so that
+    # merge_session_messages_append_only() still filters out replaced
+    # pre-edit rows from state.db whose timestamps fall below the boundary.
+    # The merge's sidecar_advanced_past_watermark guard allows state.db rows
+    # newer than the watermark, so post-edit turns are not dropped.
+    # Never 0.0 (the truncate-to-empty sentinel, #2914).
     if getattr(session, 'truncation_watermark', None):
-        session.truncation_watermark = None
+        session.truncation_watermark = recovered_ts
     return recovered
 
 
@@ -643,6 +646,7 @@ class Session:
                  context_length=None, threshold_tokens=None,
                  last_prompt_tokens=None,
                  truncation_watermark=None,
+                 truncation_boundary=None,
                  gateway_routing=None, gateway_routing_history=None,
                  llm_title_generated: bool=False,
                  manual_title: bool=False,
@@ -693,6 +697,7 @@ class Session:
         self.threshold_tokens = threshold_tokens
         self.last_prompt_tokens = last_prompt_tokens
         self.truncation_watermark = truncation_watermark
+        self.truncation_boundary = truncation_boundary
         self.gateway_routing = gateway_routing if isinstance(gateway_routing, dict) else None
         self.gateway_routing_history = gateway_routing_history if isinstance(gateway_routing_history, list) else []
         self.llm_title_generated = bool(llm_title_generated)
@@ -762,6 +767,7 @@ class Session:
             'compression_anchor_details', 'context_engine_state',
             'context_length', 'threshold_tokens', 'last_prompt_tokens',
             'truncation_watermark',
+            'truncation_boundary',
             'gateway_routing', 'gateway_routing_history', 'llm_title_generated', 'manual_title',
             'parent_session_id',
             'worktree_path', 'worktree_branch', 'worktree_repo_root', 'worktree_created_at',
@@ -5106,31 +5112,74 @@ def merge_session_messages_append_only(
     state_messages: list,
     *,
     truncation_watermark=None,
+    truncation_boundary=None,
 ) -> list:
-    """Merge sidecar/context and state.db messages without deleting local rows."""
+    """Merge sidecar/context and state.db messages without deleting local rows.
+
+    ``truncation_boundary``: the original truncate cutoff — the
+    timestamp of the last message kept by the truncate operation.  When the
+    watermark is later advanced (new turn committed), this boundary is preserved
+    so the empty-sidecar recovery can distinguish a legitimate prefix from a
+    deleted suffix instead of guessing by dropping one turn pair.
+    """
     sidecar_messages = list(sidecar_messages or [])
     state_messages = list(state_messages or [])
     watermark_timestamp = _message_timestamp_as_float({"timestamp": truncation_watermark})
     if not state_messages:
         return sidecar_messages
     if not sidecar_messages:
-        if watermark_timestamp is not None:
-            filtered = [
-                msg for msg in state_messages
-                if (
-                    (timestamp := _message_timestamp_as_float(msg)) is not None
-                    and timestamp <= watermark_timestamp
-                )
-            ]
-        else:
+        if watermark_timestamp is None:
+            # No watermark — keep everything, just dedup.
             filtered = state_messages
+        elif watermark_timestamp == 0:
+            # Truncate-to-empty sentinel (#2914) — block all replay.
+            return []
+        else:
+            # Positive watermark after edit/retry/undo (#4767).  Without a
+            # sidecar there's no seen_content_keys to check against, so we
+            # reconstruct the correct transcript from state.db alone.
+            #
+            # `at_or_after` (ts >= watermark) is legitimate POST-EDIT content
+            # ONLY when the watermark was ADVANCED strictly past the original
+            # truncate cutoff — i.e. a new turn was committed after the edit, so
+            # truncation_boundary (the original cutoff) is strictly below the
+            # advanced watermark.  In that state we keep the legitimate prefix
+            # (ts <= boundary) plus the post-edit tail (ts >= watermark) and drop
+            # the deleted (boundary, watermark) suffix.
+            #
+            # In every OTHER state the content above the watermark is the deleted
+            # suffix, NOT post-edit content, so keeping it would resurrect deleted
+            # turns (the exact data-loss this fix exists to kill):
+            #   * boundary == watermark — just truncated, no new turn committed
+            #     yet (e.g. crash/cold-load with metadata-vs-sidecar divergence);
+            #   * boundary is None — legacy session saved before this field
+            #     existed.  In the pre-#4767 model committing a turn CLEARED the
+            #     watermark to None, so a persisted positive watermark always
+            #     meant "frozen at cutoff, not advanced".
+            # For all of those, fall back to the conservative pre-#4767 filter
+            # `ts <= watermark`, which never resurrects a deleted suffix.
+            boundary_ts = _message_timestamp_as_float({"timestamp": truncation_boundary})
+            if boundary_ts is not None and boundary_ts < watermark_timestamp:
+                pre_legitimate = [
+                    m for m in state_messages
+                    if (ts := _message_timestamp_as_float(m)) is not None
+                    and ts <= boundary_ts
+                ]
+                at_or_after = [
+                    m for m in state_messages
+                    if (ts := _message_timestamp_as_float(m)) is not None
+                    and ts >= watermark_timestamp
+                ]
+                filtered = pre_legitimate + at_or_after
+            else:
+                filtered = [
+                    m for m in state_messages
+                    if (ts := _message_timestamp_as_float(m)) is not None
+                    and ts <= watermark_timestamp
+                ]
+
         # Deduplicate true duplicates (same role, content, exact timestamp)
         # without collapsing legitimately-repeated identical turns (#3346).
-        # Note: rows whose timestamps were mutated by compaction/recovery to
-        # microsecond-different values will not be folded — only byte-identical
-        # timestamps are treated as the same message.  This is intentional;
-        # collapsing same-second distinct turns would be worse than retaining
-        # a compaction-restamped duplicate.
         seen = set()
         seen_messages = {}
         deduped = []
@@ -5186,6 +5235,18 @@ def merge_session_messages_append_only(
     sidecar_visible_lookup = _build_visible_duplicate_lookup(sidecar_visible_keys)
     state_replay_idx = 0
     skipped_state_visible_counts = {}
+    # Loop-invariant: a session whose original truncate cutoff (truncation_boundary)
+    # is strictly below the watermark is genuinely ADVANCED (a new turn was
+    # committed after the edit). In that state post-watermark state.db rows are
+    # legitimate post-edit content, even when the sidecar's newest row only
+    # EQUALS the watermark (the post-edit user is checkpointed but its assistant
+    # reply exists only in state.db). Conservative for boundary None / == watermark.
+    boundary_ts = _message_timestamp_as_float({"timestamp": truncation_boundary})
+    watermark_advanced_by_boundary = (
+        watermark_timestamp is not None
+        and boundary_ts is not None
+        and boundary_ts < watermark_timestamp
+    )
     for msg in state_messages:
         timestamp = _message_timestamp_as_float(msg)
         key = _session_message_merge_key(msg)
@@ -5222,10 +5283,33 @@ def merge_session_messages_append_only(
         # rows once the session moves forward past the edit boundary. Once the
         # sidecar's own max timestamp is beyond the watermark (the session has
         # advanced), allow state rows newer than the sidecar tail to merge.
+        #
+        # The sidecar's max timestamp can also EQUAL the watermark when the new
+        # post-edit USER turn has been checkpointed into the sidecar (its
+        # timestamp == the advanced watermark) but its ASSISTANT reply exists
+        # only in state.db (recovery before the sidecar tail advances). In that
+        # state truncation_boundary < watermark proves the session is genuinely
+        # advanced, so the post-watermark state-only reply is legitimate
+        # post-edit content and must merge through (not be dropped as a replaced
+        # tail). The conservative skip still applies for boundary is None and
+        # boundary == watermark (not-advanced / legacy).
+        #
+        # CRITICAL: the boundary-advanced signal may only bypass the skip AFTER
+        # state replay has consumed the sidecar's visible checkpoint
+        # (state_replay_idx >= len(sidecar_visible_sequence)). A deleted suffix
+        # row with ts > watermark that appears in state.db BEFORE the edited
+        # checkpoint must still be skipped — otherwise the advanced signal would
+        # resurrect it. The sidecar-max-timestamp signal needs no such gate (a
+        # sidecar tail beyond the watermark is itself proof the checkpoint has
+        # advanced).
+        checkpoint_consumed = state_replay_idx >= len(sidecar_visible_sequence)
         sidecar_advanced_past_watermark = (
             watermark_timestamp is not None
-            and max_sidecar_timestamp is not None
-            and max_sidecar_timestamp > watermark_timestamp
+            and (
+                (max_sidecar_timestamp is not None
+                 and max_sidecar_timestamp > watermark_timestamp)
+                or (watermark_advanced_by_boundary and checkpoint_consumed)
+            )
         )
         if (
             watermark_timestamp is not None
@@ -5252,6 +5336,25 @@ def merge_session_messages_append_only(
             and _session_message_content_key(msg) not in seen_content_keys
         ):
             continue
+        # Same-second edit: if timestamp equals the watermark and the message
+        # content is not in the sidecar, it's a replaced message edited at the
+        # same second — skip it.  The edited version (same timestamp, different
+        # content) is in the sidecar and survives this check.
+        #
+        # Only apply the same-second guard to user messages.  An assistant reply
+        # (or tool message) at the same second as the watermark is a legitimate
+        # post-edit recovery row — the sidecar holds only the edited user
+        # checkpoint, so the assistant reply's content won't be in it and would
+        # be silently dropped without this role guard.
+        if (
+            watermark_timestamp is not None
+            and timestamp is not None
+            and timestamp == watermark_timestamp
+            and key not in seen_message_keys
+            and _session_message_content_key(msg) not in seen_content_keys
+            and str(msg.get("role", "")).lower() == "user"
+        ):
+            continue
         # Check for true duplicates using full-precision timestamp (#3346).
         # Must run before the merge-key guards so that legitimately distinct
         # sub-second messages with the same second-level merge key are not
@@ -5309,30 +5412,50 @@ def merge_session_messages_append_only(
             and timestamp is not None
             and timestamp <= max_sidecar_timestamp
         ):
-            # Legacy key within sidecar timestamp range.  Normally skip — the
-            # sidecar already has this message.  Exception: if the state.db
-            # message has tool_calls that DIFFER from the sidecar version
-            # (same content_key but different dedup_key because tool_calls
-            # differ), preserve it — distinct tool_calls must not be collapsed.
-            _tc = msg.get("tool_calls")
-            if _tc:
-                _ck = _session_message_content_key(msg)
-                if _ck in seen_content_keys and dedup_key not in seen_dedup_keys:
-                    pass  # different tool_calls from sidecar — preserve
+            # When a truncation watermark is active and the sidecar holds only
+            # the edited user checkpoint, state.db may contain an assistant/tool
+            # reply at the same timestamp that is NOT in the sidecar.  This
+            # block would normally skip it ("sidecar already has this message"),
+            # but the sidecar doesn't — it's a genuine state-only recovery row.
+            # Let it through (CORE-B, #4767).
+            #
+            # Only AFTER the sidecar's visible checkpoint has been consumed
+            # (checkpoint_consumed) — a same-second row appearing in state.db
+            # BEFORE the edited user replay is a deleted/replaced row, not the
+            # post-edit reply, and must stay skipped.
+            if (
+                watermark_timestamp is not None
+                and timestamp == watermark_timestamp
+                and checkpoint_consumed
+                and str(msg.get("role", "")).lower() != "user"
+                and _session_message_content_key(msg) not in seen_content_keys
+            ):
+                pass  # fall through to append below
+            else:
+                # Legacy key within sidecar timestamp range.  Normally skip — the
+                # sidecar already has this message.  Exception: if the state.db
+                # message has tool_calls that DIFFER from the sidecar version
+                # (same content_key but different dedup_key because tool_calls
+                # differ), preserve it — distinct tool_calls must not be collapsed.
+                _tc = msg.get("tool_calls")
+                if _tc:
+                    _ck = _session_message_content_key(msg)
+                    if _ck in seen_content_keys and dedup_key not in seen_dedup_keys:
+                        pass  # different tool_calls from sidecar — preserve
+                    else:
+                        _merge_session_display_metadata(merged_by_message_key.get(key), msg)
+                        continue
                 else:
+                    if msg.get("role") == "user" and _session_message_content_key(msg) not in seen_content_keys:
+                        if _insert_state_message_chronologically(merged_messages, msg):
+                            seen_message_keys.add(key)
+                            seen_dedup_keys.add(dedup_key)
+                            seen_content_keys.add(_session_message_content_key(msg))
+                            seen_visible_keys.add(visible_key)
+                            _remember_merged_message(msg)
+                        continue
                     _merge_session_display_metadata(merged_by_message_key.get(key), msg)
                     continue
-            else:
-                if msg.get("role") == "user" and _session_message_content_key(msg) not in seen_content_keys:
-                    if _insert_state_message_chronologically(merged_messages, msg):
-                        seen_message_keys.add(key)
-                        seen_dedup_keys.add(dedup_key)
-                        seen_content_keys.add(_session_message_content_key(msg))
-                        seen_visible_keys.add(visible_key)
-                        _remember_merged_message(msg)
-                    continue
-                _merge_session_display_metadata(merged_by_message_key.get(key), msg)
-                continue
         seen_message_keys.add(key)
         seen_dedup_keys.add(dedup_key)
         seen_content_keys.add(_session_message_content_key(msg))
@@ -5383,6 +5506,7 @@ def reconciled_state_db_messages_for_session(
         local_messages,
         state_messages,
         truncation_watermark=getattr(session, "truncation_watermark", None),
+        truncation_boundary=getattr(session, "truncation_boundary", None),
     )
 
 

api/routes.py

@@ -6059,6 +6059,7 @@ def _limited_webui_messages_for_display(session, state_db_messages) -> list:
         sidecar_messages,
         state_db_messages,
         truncation_watermark=getattr(session, "truncation_watermark", None),
+        truncation_boundary=getattr(session, "truncation_boundary", None),
     )
 
 
@@ -6117,6 +6118,7 @@ def _webui_sidecar_lineage_messages_for_display(session, *, max_hops: int = 20)
             merged,
             getattr(segment, "messages", []) or [],
             truncation_watermark=getattr(segment, "truncation_watermark", None),
+            truncation_boundary=getattr(segment, "truncation_boundary", None),
         )
     return merge_session_messages_append_only(
         merged,
@@ -6143,6 +6145,7 @@ def _merged_session_messages_for_display(session, cli_messages=None) -> list:
                     sidecar_messages,
                     cli_messages,
                     truncation_watermark=getattr(session, "truncation_watermark", None),
+                    truncation_boundary=getattr(session, "truncation_boundary", None),
                 )
             merged_messages = []
             seen_message_keys = set()
@@ -9548,6 +9551,7 @@ def handle_get(handler, parsed) -> bool:
                         _webui_sidecar_lineage_messages_for_display(s),
                         state_db_messages,
                         truncation_watermark=getattr(s, "truncation_watermark", None),
+                        truncation_boundary=getattr(s, "truncation_boundary", None),
                     )
                     _all_msgs = _merged_webui_lineage_messages_for_display(s, _all_msgs)
             else:
@@ -10894,6 +10898,7 @@ def handle_post(handler, parsed) -> bool:
                 context_length=getattr(session, "context_length", None),
                 threshold_tokens=getattr(session, "threshold_tokens", None),
                 truncation_watermark=getattr(session, "truncation_watermark", None),
+                truncation_boundary=getattr(session, "truncation_boundary", None),
                 # context_messages is the authoritative model-facing prefix — must be
                 # deepcopied so the duplicate has its own independent context that won't
                 # be mutated when the original session's context changes (#2914).
@@ -11459,8 +11464,11 @@ def handle_post(handler, parsed) -> bool:
             try:
                 from api.session_ops import _truncation_watermark_for
                 s.truncation_watermark = _truncation_watermark_for(s.messages)
+                # Persist the original truncate cutoff.
+                s.truncation_boundary = s.truncation_watermark
             except Exception:
                 s.truncation_watermark = 0.0
+                s.truncation_boundary = 0.0
             s.save()
             logger.info(
                 "truncate %s: messages %d→%d, context_messages %d→%d, watermark=%.2f",
@@ -16243,15 +16251,15 @@ def _checkpoint_user_message_for_eager_session_save(s, msg: str, attachments, st
     if attachments:
         user_msg["attachments"] = list(attachments)
     s.messages.append(user_msg)
-    # The new user turn is now committed to messages (#3831): a positive
-    # truncation watermark from a prior retry/undo/edit has been superseded and
-    # must retire, else it freezes at the old edit boundary and later drops these
-    # post-edit turns on an empty-sidecar reconcile. Safe here (not at chat-start)
-    # because the row is durably in messages, so the merge's max-sidecar guard now
-    # suppresses the replaced tail without the watermark. Cleared to None — never
-    # 0.0, which is the truncate-to-empty sentinel (#2914).
+    # The new user turn is now committed to messages (#3831): advance the
+    # truncation watermark to the new message's timestamp so that
+    # merge_session_messages_append_only() still filters out replaced
+    # pre-edit rows from state.db whose timestamps fall below the boundary.
+    # The merge's sidecar_advanced_past_watermark guard (models.py:5172)
+    # allows state.db rows newer than the watermark, so post-edit turns
+    # are not dropped. Never 0.0 (the truncate-to-empty sentinel, #2914).
     if getattr(s, "truncation_watermark", None):
-        s.truncation_watermark = None
+        s.truncation_watermark = user_msg.get("timestamp") or time.time()
 
 
 def _is_default_or_empty_session_title(title) -> bool:

api/session_ops.py

@@ -116,6 +116,9 @@ def retry_last(session_id: str) -> dict[str, Any]:
             removed_count = len(history) - last_user_idx
             s.messages = history[:last_user_idx]
             s.truncation_watermark = _truncation_watermark_for(s.messages)
+            # Persist the original truncate cutoff so empty-sidecar recovery
+            # can distinguish legitimate prefix from deleted suffix.
+            s.truncation_boundary = s.truncation_watermark
             if isinstance(getattr(s, 'context_messages', None), list) and s.context_messages:
                 truncated_context = _truncate_at_last_user(s.context_messages)
                 if truncated_context is not None:
@@ -156,6 +159,8 @@ def undo_last(session_id: str) -> dict[str, Any]:
             removed_count = len(history) - last_user_idx
             s.messages = history[:last_user_idx]
             s.truncation_watermark = _truncation_watermark_for(s.messages)
+            # Persist the original truncate cutoff.
+            s.truncation_boundary = s.truncation_watermark
             if isinstance(getattr(s, 'context_messages', None), list) and s.context_messages:
                 truncated_context = _truncate_at_last_user(s.context_messages)
                 if truncated_context is not None:

api/streaming.py

@@ -4505,21 +4505,29 @@ def _stream_writeback_can_supersede_recovery_marker(session, msg_text):
     return False
 
 
-def _retire_truncation_watermark_after_commit(session) -> None:
-    """Clear a positive truncation watermark once a new user turn is committed
+def _advance_truncation_watermark_after_commit(session) -> None:
+    """Advance a positive truncation watermark once a new user turn is committed
     to ``session.messages`` (#3831).
 
     retry/undo/Edit set a positive watermark to suppress the *replaced* tail from
     the append-only state.db merge; Session.save() deliberately never auto-clears
-    it (#2914). But nothing retired it when the user then sent a NEW turn, so it
-    froze at the old edit boundary and later dropped those post-edit turns on an
-    empty-sidecar reconcile. Once the new turn is durably in messages the merge's
-    max-sidecar guard suppresses the replaced tail without the watermark, so it is
-    safe — and necessary — to retire it here. Cleared to None, never 0.0 (the
-    truncate-to-empty sentinel that must keep blocking replay, #2914).
+    it (#2914). Once the new turn is durably in messages we advance the watermark
+    to the newest user message timestamp so that state.db rows newer than the
+    watermark are still merged in, while the replaced pre-edit tail remains
+    filtered. Never 0.0 (the truncate-to-empty sentinel that must keep blocking
+    replay, #2914).
     """
-    if getattr(session, 'truncation_watermark', None):
-        session.truncation_watermark = None
+    if not getattr(session, 'truncation_watermark', None):
+        return
+    messages = getattr(session, 'messages', None) or []
+    # Walk backwards to find the newest user message timestamp
+    for msg in reversed(messages):
+        if isinstance(msg, dict) and msg.get('role') == 'user':
+            ts = msg.get('timestamp')
+            if isinstance(ts, (int, float)) and ts > 0:
+                session.truncation_watermark = float(ts)
+                return
+    session.truncation_watermark = time.time()
 
 
 def _merge_display_messages_after_agent_result(previous_display, previous_context, result_messages, msg_text, source: str = "webui"):
@@ -5183,14 +5191,14 @@ def _materialize_pending_user_turn_before_error(session) -> bool:
             for e in ctx[-8:]
         ):
             ctx.append({k: v for k, v in recovered.items() if k != 'timestamp'})
-    # The new user turn is now committed to messages (#3831): retire a positive
-    # truncation watermark left over from a prior retry/undo/edit so it cannot
-    # freeze at the old edit boundary and later drop these post-edit turns on an
-    # empty-sidecar reconcile. Cleared to None — never 0.0 (the truncate-to-empty
-    # sentinel, #2914). Safe here because the row is durably in messages, so the
-    # merge's max-sidecar guard suppresses the replaced tail without the watermark.
+    # The new user turn is now committed to messages (#3831): advance a positive
+    # truncation watermark left over from a prior retry/undo/edit so that
+    # merge_session_messages_append_only() still filters out replaced pre-edit
+    # rows from state.db. The merge's sidecar_advanced_past_watermark guard
+    # allows state.db rows newer than the watermark, so post-edit turns are not
+    # dropped. Never 0.0 (the truncate-to-empty sentinel, #2914).
     if getattr(session, 'truncation_watermark', None):
-        session.truncation_watermark = None
+        session.truncation_watermark = float(recovered_ts)
     return True
 
 
@@ -7538,7 +7546,7 @@ def _run_agent_streaming(
                     msg_text,
                     source=getattr(s, 'pending_user_source', None) or 'webui',
                 )
-                _retire_truncation_watermark_after_commit(s)  # #3831
+                _advance_truncation_watermark_after_commit(s)  # #3831
                 # Strip XML tool-call blocks from assistant message content.
                 # DeepSeek and some other providers emit <function_calls>...</function_calls>
                 # in the raw response text; this must be removed before the content is
@@ -7831,7 +7839,7 @@ def _run_agent_streaming(
                                     msg_text,
                                     source=getattr(s, 'pending_user_source', None) or 'webui',
                                 )
-                                _retire_truncation_watermark_after_commit(s)  # #3831
+                                _advance_truncation_watermark_after_commit(s)  # #3831
                                 # Skip the error block — jump directly to the
                                 # normal post-result persistence path by
                                 # leaving _assistant_added truthy (set below).
@@ -8862,7 +8870,7 @@ def _run_agent_streaming(
                                     msg_text,
                                     source=getattr(s, 'pending_user_source', None) or 'webui',
                                 )
-                                _retire_truncation_watermark_after_commit(s)  # #3831
+                                _advance_truncation_watermark_after_commit(s)  # #3831
                                 s.save()
                         logger.info('[webui] self-heal (except path): retry succeeded')
                         return  # skip error emission

api/webui_session_db.py

@@ -52,6 +52,7 @@ _METADATA_FIELDS = frozenset(
         "threshold_tokens",
         "last_prompt_tokens",
         "truncation_watermark",
+        "truncation_boundary",
         "gateway_routing",
         "gateway_routing_history",
         "llm_title_generated",

tests/test_core_data_loss_cases.py

@@ -0,0 +1,532 @@
+"""Reproduction tests for data-loss cases found in PR review.
+
+Empty-sidecar recovery resurrects deleted pre-edit turns when editing
+an older message with ≥2 later turns.
+
+Same-second recovery can silently drop the legitimate post-edit
+assistant reply.
+
+Both are regression tests — they should FAIL against the current code
+(commit 297c88fed7) and PASS after the fix.
+"""
+from __future__ import annotations
+
+import api.models as models
+import api.webui_session_db as webui_db
+
+
+def _msg(role: str, content: str, ts: float) -> dict:
+    return {"role": role, "content": content, "timestamp": ts}
+
+
+# ─── Resurrected deleted turns ───────────────────────────────────────────────
+
+
+def test_core_a_empty_sidecar_resurrects_deleted_turns():
+    """Editing an older message with ≥2 later turns resurrects
+    the deleted suffix on empty-sidecar recovery."""
+    state = [
+        _msg("user", "original prompt", 100.0),
+        _msg("assistant", "original reply", 101.0),
+        _msg("user", "deleted question 1", 200.0),
+        _msg("assistant", "deleted answer 1", 201.0),
+        _msg("user", "deleted question 2", 300.0),
+        _msg("assistant", "deleted answer 2", 302.0),
+        _msg("user", "edited prompt", 400.0),
+        _msg("assistant", "post-edit reply", 401.0),
+    ]
+
+    merged = models.merge_session_messages_append_only(
+        [],  # empty sidecar (cold reload)
+        state,
+        truncation_watermark=400.0,
+        truncation_boundary=101.0,
+    )
+
+    contents = [m["content"] for m in merged]
+
+    assert "original prompt" in contents
+    assert "original reply" in contents
+    assert "edited prompt" in contents
+    assert "post-edit reply" in contents
+
+    assert "deleted question 1" not in contents, (
+        f"Deleted turn @200 resurrected! Contents: {contents}"
+    )
+    assert "deleted answer 1" not in contents, (
+        f"Deleted turn @201 resurrected! Contents: {contents}"
+    )
+    assert "deleted question 2" not in contents, (
+        f"Deleted turn @300 resurrected! Contents: {contents}"
+    )
+    assert "deleted answer 2" not in contents, (
+        f"Deleted turn @302 resurrected! Contents: {contents}"
+    )
+
+    expected = [
+        "original prompt", "original reply",
+        "edited prompt", "post-edit reply",
+    ]
+    assert contents == expected, (
+        f"Expected {expected}, got {contents}"
+    )
+
+
+def test_core_a_single_deleted_turn_still_works():
+    """Sanity check: an advanced watermark with the original cutoff persisted
+    as truncation_boundary keeps the prefix + post-edit tail and drops the one
+    deleted turn."""
+    state = [
+        _msg("user", "first msg", 50.0),
+        _msg("assistant", "first reply", 51.0),
+        _msg("user", "original pre-edit", 100.0),
+        _msg("assistant", "original reply", 101.0),
+        _msg("user", "edited/new turn", 200.0),
+        _msg("assistant", "post-edit reply", 201.0),
+    ]
+
+    # Original cutoff kept through the first reply (@51); a new turn was then
+    # committed (@200), advancing the watermark. boundary (51) < watermark (200).
+    merged = models.merge_session_messages_append_only(
+        [], state, truncation_watermark=200.0, truncation_boundary=51.0
+    )
+
+    contents = [m["content"] for m in merged]
+    expected = [
+        "first msg", "first reply", "edited/new turn", "post-edit reply"
+    ]
+    assert contents == expected, (
+        f"Sanity: expected {expected}, got {contents}"
+    )
+
+
+def test_core_a_not_advanced_watermark_equals_boundary_does_not_resurrect():
+    """Just-truncated state (boundary == watermark, no new turn committed yet):
+    everything above the watermark is the deleted suffix and must NOT be kept.
+
+    Reachable via crash/cold-load metadata-vs-sidecar divergence — the exact
+    scenario the empty-sidecar branch serves. Opus gate found that the prior
+    code swept the suffix into `at_or_after` here, resurrecting deleted turns
+    (the same data-loss class this fix exists to kill)."""
+    state = [
+        _msg("user", "u1", 50.0),
+        _msg("assistant", "a1", 51.0),
+        _msg("user", "deleted-u2", 100.0),
+        _msg("assistant", "deleted-a2", 101.0),
+        _msg("user", "deleted-u3", 150.0),
+        _msg("assistant", "deleted-a3", 151.0),
+    ]
+
+    merged = models.merge_session_messages_append_only(
+        [], state, truncation_watermark=51.0, truncation_boundary=51.0
+    )
+
+    contents = [m["content"] for m in merged]
+    assert contents == ["u1", "a1"], (
+        f"watermark==boundary must keep only ts<=watermark, got {contents}"
+    )
+
+
+def test_core_a_legacy_none_boundary_frozen_watermark_does_not_resurrect():
+    """Legacy session (truncation_boundary is None) with a positive watermark:
+    in the pre-#4767 model a persisted positive watermark always meant
+    'frozen at cutoff' (committing a turn cleared it to None), so the safe
+    behavior is the conservative ts<=watermark filter — no resurrection, and
+    the legitimately-kept first turn survives."""
+    state = [
+        _msg("user", "u1", 50.0),
+        _msg("assistant", "a1", 51.0),
+        _msg("user", "deleted-u2", 100.0),
+        _msg("assistant", "deleted-a2", 101.0),
+        _msg("user", "deleted-u3", 150.0),
+        _msg("assistant", "deleted-a3", 151.0),
+    ]
+
+    merged = models.merge_session_messages_append_only(
+        [], state, truncation_watermark=51.0  # boundary defaults to None
+    )
+
+    contents = [m["content"] for m in merged]
+    assert contents == ["u1", "a1"], (
+        f"legacy None-boundary frozen watermark must keep only ts<=watermark "
+        f"(no resurrection, keep u1), got {contents}"
+    )
+
+
+def test_advanced_sidecar_at_watermark_keeps_state_only_post_edit_reply():
+    """Non-empty sidecar whose newest row EQUALS the watermark (the post-edit
+    user turn is checkpointed but its assistant reply exists only in state.db)
+    must keep the state-only post-edit reply when the session is genuinely
+    advanced (truncation_boundary < watermark).
+
+    Opus/Codex gate found that the `sidecar_advanced_past_watermark` guard only
+    treated `max_sidecar_timestamp > watermark` as advanced, so a reply at
+    ts > watermark was dropped whenever the sidecar tail merely EQUALLED the
+    watermark — silently losing the legitimate post-edit assistant reply on
+    restore/full reload. The boundary < watermark signal now also marks the
+    session advanced."""
+    # boundary @51 (original cutoff), watermark advanced to @200 (new user turn),
+    # sidecar holds the prefix + the committed edited user @200 (== watermark),
+    # state.db additionally holds the post-edit assistant reply @201 (state-only).
+    sidecar = [
+        _msg("user", "first", 50.0),
+        _msg("assistant", "first reply", 51.0),
+        _msg("user", "edited prompt", 200.0),
+    ]
+    state = [
+        _msg("user", "first", 50.0),
+        _msg("assistant", "first reply", 51.0),
+        _msg("user", "edited prompt", 200.0),
+        _msg("assistant", "post-edit reply", 201.0),
+    ]
+
+    merged = models.merge_session_messages_append_only(
+        sidecar, state, truncation_watermark=200.0, truncation_boundary=51.0
+    )
+
+    contents = [m["content"] for m in merged]
+    assert "post-edit reply" in contents, (
+        f"State-only post-edit assistant reply was DROPPED! Contents: {contents}"
+    )
+    assert contents == ["first", "first reply", "edited prompt", "post-edit reply"], (
+        f"Expected full advanced transcript, got {contents}"
+    )
+
+
+def test_advanced_does_not_resurrect_stale_post_watermark_row_before_checkpoint():
+    """The boundary<watermark advanced signal must NOT resurrect a deleted
+    suffix row with ts > watermark that appears in state.db BEFORE the edited
+    checkpoint row.
+
+    Codex final-gate found that making `sidecar_advanced_past_watermark`
+    globally true (whenever boundary < watermark) let a stale ts>watermark row
+    bypass the skip before state replay reached the sidecar's edited checkpoint,
+    resurrecting deleted turns. The bypass is now gated on the checkpoint having
+    been consumed (state_replay_idx >= len(sidecar_visible_sequence))."""
+    # sidecar ends at the edited user @200 (== watermark); state.db lists a
+    # deleted future suffix (@300/@301) BEFORE the appended edited turn.
+    sidecar = [
+        _msg("user", "first", 50.0),
+        _msg("assistant", "first reply", 51.0),
+        _msg("user", "edited prompt", 200.0),
+    ]
+    state = [
+        _msg("user", "first", 50.0),
+        _msg("assistant", "first reply", 51.0),
+        _msg("user", "deleted future prompt", 300.0),
+        _msg("assistant", "deleted future answer", 301.0),
+        _msg("user", "edited prompt", 200.0),
+    ]
+
+    merged = models.merge_session_messages_append_only(
+        sidecar, state, truncation_watermark=200.0, truncation_boundary=51.0
+    )
+
+    contents = [m["content"] for m in merged]
+    assert "deleted future prompt" not in contents, (
+        f"Stale post-watermark row resurrected before checkpoint! {contents}"
+    )
+    assert "deleted future answer" not in contents, (
+        f"Stale post-watermark row resurrected before checkpoint! {contents}"
+    )
+
+
+# ─── Same-second assistant reply dropped ──────────────────────────────────────
+
+
+def test_core_b_same_second_assistant_reply_dropped():
+    """Same-second equality guard drops legitimate post-edit
+    assistant reply when it shares the timestamp with the edited user message."""
+    T = 100.0
+    sidecar = [_msg("user", "edited prompt", T)]
+    state = [
+        _msg("user", "edited prompt", T),
+        _msg("assistant", "post-edit reply", T),
+    ]
+
+    merged = models.merge_session_messages_append_only(
+        sidecar, state, truncation_watermark=T
+    )
+
+    contents = [m["content"] for m in merged]
+
+    assert "edited prompt" in contents, (
+        f"Edited user message missing! Contents: {contents}"
+    )
+    assert "post-edit reply" in contents, (
+        f"Same-second assistant reply was DROPPED! Contents: {contents}"
+    )
+
+
+def test_core_b_same_second_replaced_user_still_filtered():
+    """Sanity check: same-second guard still filters the replaced
+    pre-edit user message (same timestamp, different content)."""
+    T = 100.0
+    sidecar = [_msg("user", "edited prompt", T)]
+    state = [
+        _msg("user", "original pre-edit prompt", T),
+        _msg("user", "edited prompt", T),
+        _msg("assistant", "post-edit reply", T),
+    ]
+
+    merged = models.merge_session_messages_append_only(
+        sidecar, state, truncation_watermark=T
+    )
+
+    contents = [m["content"] for m in merged]
+
+    assert "original pre-edit prompt" not in contents, (
+        f"Sanity: replaced user message leaked! Contents: {contents}"
+    )
+    assert "edited prompt" in contents
+    assert "post-edit reply" in contents
+
+
+def test_core_b_same_second_empty_sidecar_assistant_reply():
+    """Same-second assistant reply in empty-sidecar
+    recovery path with truncation_boundary — must survive."""
+    T = 100.0
+    state = [
+        _msg("user", "first msg", 50.0),
+        _msg("assistant", "first reply", 51.0),
+        _msg("user", "edited prompt", T),
+        _msg("assistant", "post-edit reply", T),
+    ]
+
+    merged = models.merge_session_messages_append_only(
+        [], state, truncation_watermark=T,
+        truncation_boundary=51.0,
+    )
+
+    contents = [m["content"] for m in merged]
+
+    assert "first msg" in contents
+    assert "first reply" in contents
+    assert "edited prompt" in contents
+    assert "post-edit reply" in contents, (
+        f"Same-second assistant reply dropped! Contents: {contents}"
+    )
+
+
+# ─── Persistence: save/load round-trip ────────────────────────────────────────
+
+
+def test_truncation_boundary_survives_save_load(monkeypatch, tmp_path):
+    """truncation_boundary must be persisted to JSON and restored on load."""
+    session_dir = tmp_path / "sessions"
+    session_dir.mkdir(parents=True, exist_ok=True)
+    monkeypatch.setattr(models, "SESSION_DIR", session_dir)
+    monkeypatch.setattr(models, "SESSION_INDEX_FILE", session_dir / "_index.json")
+    models.SESSIONS.clear()
+
+    sid = "boundary_save_load"
+    session = models.Session(
+        session_id=sid,
+        messages=[_msg("user", "hello", 100.0)],
+        truncation_watermark=200.0,
+        truncation_boundary=101.0,
+    )
+    session.save()
+
+    with models.LOCK:
+        models.SESSIONS.pop(sid, None)
+
+    loaded = models.Session.load(sid)
+    assert loaded.truncation_boundary == 101.0, (
+        f"truncation_boundary lost after save/load: got {loaded.truncation_boundary}"
+    )
+    assert loaded.truncation_watermark == 200.0
+
+
+def test_truncation_boundary_none_survives_save_load(monkeypatch, tmp_path):
+    """When truncation_boundary is None, it must survive save/load as None."""
+    session_dir = tmp_path / "sessions"
+    session_dir.mkdir(parents=True, exist_ok=True)
+    monkeypatch.setattr(models, "SESSION_DIR", session_dir)
+    monkeypatch.setattr(models, "SESSION_INDEX_FILE", session_dir / "_index.json")
+    models.SESSIONS.clear()
+
+    sid = "boundary_none_save"
+    session = models.Session(
+        session_id=sid,
+        messages=[_msg("user", "hello", 100.0)],
+        truncation_watermark=None,
+        truncation_boundary=None,
+    )
+    session.save()
+
+    with models.LOCK:
+        models.SESSIONS.pop(sid, None)
+
+    loaded = models.Session.load(sid)
+    assert getattr(loaded, "truncation_boundary", None) is None
+
+
+# ─── reconciled_state_db_messages_for_session passes boundary ─────────────────
+
+
+def test_reconciled_passes_truncation_boundary(monkeypatch, tmp_path):
+    """reconciled_state_db_messages_for_session must pass truncation_boundary
+    to merge_session_messages_append_only so empty-sidecar recovery uses it."""
+    session_dir = tmp_path / "sessions"
+    session_dir.mkdir(parents=True, exist_ok=True)
+    monkeypatch.setattr(models, "SESSION_DIR", session_dir)
+    monkeypatch.setattr(models, "SESSION_INDEX_FILE", session_dir / "_index.json")
+    models.SESSIONS.clear()
+
+    sid = "reconciled_boundary"
+    session = models.Session(
+        session_id=sid,
+        messages=[],  # empty sidecar (crash recovery)
+        truncation_watermark=200.0,
+        truncation_boundary=51.0,
+    )
+    session.save()
+
+    state_db = [
+        _msg("user", "kept", 50.0),
+        _msg("assistant", "kept reply", 51.0),
+        _msg("user", "deleted 1", 100.0),
+        _msg("assistant", "deleted answer 1", 101.0),
+        _msg("user", "deleted 2", 150.0),
+        _msg("assistant", "deleted answer 2", 151.0),
+        _msg("user", "new turn", 200.0),
+        _msg("assistant", "new reply", 201.0),
+    ]
+
+    monkeypatch.setattr(
+        models,
+        "get_state_db_session_messages",
+        lambda sid: state_db,
+    )
+
+    reconciled = models.reconciled_state_db_messages_for_session(session)
+    contents = [m["content"] for m in reconciled]
+
+    assert "deleted 1" not in contents, (
+        f"reconciled leaked deleted turn: {contents}"
+    )
+    assert "deleted 2" not in contents, (
+        f"reconciled leaked deleted turn: {contents}"
+    )
+    assert "kept" in contents
+    assert "kept reply" in contents
+    assert "new turn" in contents
+    assert "new reply" in contents
+
+
+# ─── webui_session_db _METADATA_FIELDS includes boundary ──────────────────────
+
+
+def test_webui_session_db_metadata_fields_includes_boundary():
+    """webui_session_db._METADATA_FIELDS must include truncation_boundary
+    so it's recognized as a metadata field (not leaked into extra)."""
+    assert "truncation_boundary" in webui_db._METADATA_FIELDS, (
+        "webui_session_db._METADATA_FIELDS missing truncation_boundary"
+    )
+
+
+# ─── Route-level CORE-A: default /api/session full reload must not resurrect ──
+
+
+def test_core_a_route_full_session_load_does_not_resurrect_deleted_turns(tmp_path):
+    """The default GET /api/session full reload (no msg_limit) must pass the
+    persisted truncation_boundary through to the append-only merge.
+
+    Regression guard for the call-site gap: merge_session_messages_append_only
+    was fixed to honor truncation_boundary, but the default full-load call site
+    (api/routes.py handle_get '/api/session') still passed only
+    truncation_watermark, so a normal full reload fell back to the old
+    'drop one turn pair' heuristic and resurrected deleted suffix turns when an
+    older message was edited leaving >=2 later turns. This drives the real
+    endpoint end-to-end so the boundary must reach the merge.
+    """
+    import json
+    from io import BytesIO
+    from urllib.parse import urlparse
+
+    import api.routes as routes
+    from api.models import Session
+
+    state = [
+        _msg("user", "original prompt", 100.0),
+        _msg("assistant", "original reply", 101.0),
+        _msg("user", "deleted question 1", 200.0),
+        _msg("assistant", "deleted answer 1", 201.0),
+        _msg("user", "deleted question 2", 300.0),
+        _msg("assistant", "deleted answer 2", 302.0),
+        _msg("user", "edited prompt", 400.0),
+        _msg("assistant", "post-edit reply", 401.0),
+    ]
+
+    class _Handler:
+        def __init__(self):
+            self.status = None
+            self.response_headers = []
+            self.wfile = BytesIO()
+
+        def send_response(self, status):
+            self.status = status
+
+        def send_header(self, k, v):
+            self.response_headers.append((k, v))
+
+        def end_headers(self):
+            pass
+
+        def payload(self):
+            return json.loads(self.wfile.getvalue().decode() or "{}")
+
+    sess_dir = tmp_path / "sessions"
+    sess_dir.mkdir()
+    orig_dir, orig_index = models.SESSION_DIR, models.SESSION_INDEX_FILE
+    models.SESSION_DIR = sess_dir
+    models.SESSION_INDEX_FILE = sess_dir / "_index.json"
+    models.SESSIONS.clear()
+
+    saved = {
+        "get_state_db_session_messages": getattr(routes, "get_state_db_session_messages", None),
+        "_session_visible_to_active_profile": getattr(routes, "_session_visible_to_active_profile", None),
+        "_clear_stale_stream_state": getattr(routes, "_clear_stale_stream_state", None),
+        "_resolve_effective_session_model_for_display": getattr(routes, "_resolve_effective_session_model_for_display", None),
+        "_resolve_effective_session_model_provider_for_display": getattr(routes, "_resolve_effective_session_model_provider_for_display", None),
+    }
+    try:
+        s = Session(
+            session_id="corea_route",
+            messages=[],
+            truncation_watermark=400.0,
+            truncation_boundary=101.0,
+        )
+        s.save()
+        routes.get_state_db_session_messages = lambda sid, profile=None: list(state)
+        routes._session_visible_to_active_profile = lambda profile, handler: True
+        routes._clear_stale_stream_state = lambda s: None
+        routes._resolve_effective_session_model_for_display = lambda s: getattr(s, "model", None)
+        routes._resolve_effective_session_model_provider_for_display = lambda s: getattr(s, "model_provider", None)
+
+        h = _Handler()
+        routes.handle_get(
+            h, urlparse("/api/session?session_id=corea_route&messages=1&resolve_model=0")
+        )
+        contents = [
+            m.get("content")
+            for m in h.payload().get("session", {}).get("messages", [])
+        ]
+
+        assert h.status == 200, f"unexpected status {h.status}"
+        assert "deleted question 1" not in contents, (
+            f"Deleted turn resurrected via /api/session full reload: {contents}"
+        )
+        assert "deleted answer 1" not in contents, (
+            f"Deleted turn resurrected via /api/session full reload: {contents}"
+        )
+        assert "edited prompt" in contents and "post-edit reply" in contents, (
+            f"Post-edit turn missing from full reload: {contents}"
+        )
+    finally:
+        for name, val in saved.items():
+            if val is not None:
+                setattr(routes, name, val)
+        models.SESSION_DIR, models.SESSION_INDEX_FILE = orig_dir, orig_index
+        models.SESSIONS.clear()

tests/test_issue3831_watermark_clear.py

@@ -8,18 +8,22 @@ it froze at the old edit boundary. A frozen watermark then dropped post-watermar
 state.db rows whenever the sidecar was later reconstructed empty
 (recovery/reconcile), permanently losing the turns sent after the edit.
 
-Fix: retire a *positive* watermark to ``None`` once the new user turn is
-COMMITTED to ``session.messages`` (the success-merge, eager-checkpoint, and
-recovery/cold-load commit points) — NOT at chat-start. Clearing at chat-start
-was unsafe: in deferred mode (the default) the new user row isn't in messages
-yet, so a merge in that window would resurrect the replaced tail (the
-max-sidecar guard hasn't risen past the old boundary). Once the row is committed,
-``max_sidecar_timestamp`` rises past the replaced tail and the merge suppresses
-it without the watermark, so retiring it is both safe and necessary.
+Fix: advance a *positive* watermark to the newest user message timestamp once
+the new user turn is COMMITTED to ``session.messages`` (the success-merge,
+eager-checkpoint, and recovery/cold-load commit points) — NOT at chat-start.
+Clearing at chat-start was unsafe: in deferred mode (the default) the new user
+row isn't in messages yet, so a merge in that window would resurrect the replaced
+tail (the max-sidecar guard hasn't risen past the old boundary). Once the row is
+committed, ``max_sidecar_timestamp`` rises past the replaced tail and the merge
+suppresses it without the watermark, so advancing it is both safe and necessary.
 
-``None`` is distinct from ``0.0``: ``0.0`` is the truncate-to-empty sentinel
-(#2914) that must keep blocking all state replay, so the clear is falsy-gated and
-never touches ``0.0``.
+Advancing to the new message timestamp (instead of clearing to None) keeps the
+merge's watermark filter active for replaced pre-edit rows from state.db whose
+timestamps fall below the boundary, while the sidecar_advanced_past_watermark
+guard allows newer state.db rows to merge in.
+
+``0.0`` is the truncate-to-empty sentinel (#2914) that must keep blocking all
+state replay, so the advance is falsy-gated and never touches ``0.0``.
 """
 import api.models as models
 import api.streaming as streaming
@@ -42,75 +46,117 @@ class _FakeSession:
         self.pending_started_at = None
 
 
-# --- The fix: a committed new turn retires a stale positive watermark ---------
+# --- The fix: a committed new turn advances a stale positive watermark --------
 
-def test_retire_helper_clears_positive_watermark():
+def test_advance_helper_advances_positive_watermark():
+    """After a new user turn is committed, the watermark advances to the
+    newest user message timestamp — NOT cleared to None."""
     s = _FakeSession(100.0)  # stale, from a prior retry/undo/edit
     s.messages = _rows(("user", "new turn", 200))
-    streaming._retire_truncation_watermark_after_commit(s)
-    assert s.truncation_watermark is None
+    streaming._advance_truncation_watermark_after_commit(s)
+    assert s.truncation_watermark == 200.0
 
 
-def test_retire_helper_leaves_zero_watermark_untouched():
+def test_advance_helper_leaves_zero_watermark_untouched():
     """0.0 is the truncate-to-empty sentinel (#2914), not a stale boundary — the
     falsy guard must leave it alone so #2914 replay-blocking is preserved."""
     s = _FakeSession(0.0)
     s.messages = _rows(("user", "new turn", 200))
-    streaming._retire_truncation_watermark_after_commit(s)
+    streaming._advance_truncation_watermark_after_commit(s)
     assert s.truncation_watermark == 0.0
 
 
-def test_retire_helper_noop_when_unset():
+def test_advance_helper_noop_when_unset():
     s = _FakeSession(None)
-    streaming._retire_truncation_watermark_after_commit(s)
+    streaming._advance_truncation_watermark_after_commit(s)
     assert s.truncation_watermark is None
 
 
-def test_recovery_commit_clears_watermark():
+def test_advance_helper_uses_current_time_when_no_timestamp():
+    """When the newest user message has no timestamp, the helper falls back to
+    time.time()."""
+    s = _FakeSession(100.0)
+    s.messages = _rows(("user", "new turn", None))
+    streaming._advance_truncation_watermark_after_commit(s)
+    assert s.truncation_watermark is not None
+    assert s.truncation_watermark > 100.0  # advanced past the old watermark
+
+
+def test_advance_helper_picks_newest_user_timestamp():
+    """When there are multiple user messages, the helper picks the newest one."""
+    s = _FakeSession(50.0)
+    s.messages = _rows(
+        ("user", "first", 100),
+        ("assistant", "reply", 101),
+        ("user", "second", 200),
+        ("assistant", "reply2", 201),
+    )
+    streaming._advance_truncation_watermark_after_commit(s)
+    assert s.truncation_watermark == 200.0
+
+
+def test_recovery_commit_advances_watermark():
     """The cold-load / recovery commit path (_append_recovered_pending_turn)
-    retires a stale positive watermark when it materializes the pending turn."""
+    advances a stale positive watermark when it materializes the pending turn."""
     s = _FakeSession(100.0)
     s.pending_user_message = "recovered new turn"
     s.pending_attachments = None
     models._append_recovered_pending_turn(s, timestamp=200)
-    assert s.truncation_watermark is None
+    assert s.truncation_watermark == 200
     # The pending turn was committed to messages.
     assert any(m.get("content") == "recovered new turn" for m in s.messages)
 
 
-# --- The effect: cleared (None) watermark stops dropping post-edit turns -------
+# --- The effect: advanced watermark still merges post-edit turns --------------
 
-def test_cleared_watermark_keeps_post_edit_state_rows_with_empty_sidecar():
-    """Once the watermark is cleared (None), an empty-sidecar reconcile keeps the
-    genuinely-new post-edit turns instead of dropping them (the #3831 outcome)."""
-    state = _rows(
+def test_advanced_watermark_keeps_post_edit_state_rows():
+    """With a sidecar whose max timestamp exceeds the watermark (the real
+    post-edit state after the assistant reply was appended), post-edit state.db
+    rows are merged in because sidecar_advanced_past_watermark is True."""
+    sidecar = _rows(
         ("user", "q1", 50),
         ("assistant", "a1", 100),     # old edit boundary
-        ("user", "q2-new", 200),      # sent AFTER the retry/edit
-        ("assistant", "a2-new", 300),
+        ("user", "q2-new", 200),      # new message after edit (watermark=200)
+        ("assistant", "a2-new", 250), # assistant reply — pushes sidecar past watermark
+    )
+    state = _rows(
+        ("user", "q1", 50),
+        ("assistant", "a1", 100),
+        ("user", "q2-new", 200),
+        ("assistant", "a2-new", 250),
     )
     merged = models.merge_session_messages_append_only(
-        [], state, truncation_watermark=None
+        sidecar, state, truncation_watermark=200.0
     )
+    # max_sidecar_timestamp=250 > watermark=200 → sidecar_advanced_past_watermark=True
+    # → watermark filter does not block a2-new
     assert [m["content"] for m in merged] == ["q1", "a1", "q2-new", "a2-new"]
 
 
 # --- The guardrails: the fix must NOT regress #2914/#3102 ---------------------
 
 def test_active_watermark_still_filters_replaced_tail_empty_sidecar():
-    """A still-active (positive) watermark with an empty sidecar still suppresses
-    rows above the boundary — the replaced/edited tail (#2914). The fix only
-    changes WHEN the watermark is retired, never the merge semantics."""
+    """A positive watermark (advanced to the new user-turn timestamp) with an
+    empty sidecar must suppress the replaced pre-edit tail while keeping
+    post-edit rows (#4767 / CORE finding #2).
+
+    A genuinely-advanced session persists the original cutoff as
+    ``truncation_boundary`` (@100, the last kept reply before the replaced tail)
+    alongside the advanced watermark (@200, the new committed turn). The
+    reconstruction keeps ts <= boundary plus ts >= watermark and drops the
+    replaced (boundary, watermark) suffix."""
     state = _rows(
         ("user", "q1", 50),
         ("assistant", "a1", 100),
-        ("user", "replaced-q2", 200),   # above watermark -> filtered
-        ("assistant", "replaced-a2", 300),
+        ("user", "replaced-q2", 150),   # replaced by the edit -> filtered
+        ("assistant", "replaced-a2", 160),  # replaced by the edit -> filtered
+        ("user", "edited-q2", 200),     # new user turn (watermark advanced here)
+        ("assistant", "a2-new", 300),   # post-edit reply -> kept
     )
     merged = models.merge_session_messages_append_only(
-        [], state, truncation_watermark=100.0
+        [], state, truncation_watermark=200.0, truncation_boundary=100.0
     )
-    assert [m["content"] for m in merged] == ["q1", "a1"]
+    assert [m["content"] for m in merged] == ["q1", "a1", "edited-q2", "a2-new"]
 
 
 def test_zero_watermark_still_blocks_all_replay_empty_sidecar():
@@ -126,19 +172,18 @@ def test_zero_watermark_still_blocks_all_replay_empty_sidecar():
     assert merged == []
 
 
-# --- Inline commit-path coverage (greptile): the two sites that inline the -----
-# --- falsy-gated clear instead of calling the helper must behave identically. --
+# --- Inline commit-path coverage: the two sites that inline the advance --------
 
-def test_error_path_materialize_clears_positive_watermark():
+def test_error_path_materialize_advances_positive_watermark():
     """The error/cancel materialization path (_materialize_pending_user_turn_before_error)
-    inlines the watermark clear. #3831 was triggered on recovery/reconcile after a
-    crash, so the error path is precisely the failure mode — lock it down: a pending
-    user turn committed on the error path retires a stale positive watermark."""
+    inlines the watermark advance. A pending user turn committed on the error
+    path advances a stale positive watermark to the recovered timestamp."""
     s = _FakeSession(100.0)  # stale, from a prior retry/undo/edit
     s.pending_user_message = "new turn after edit"
     appended = streaming._materialize_pending_user_turn_before_error(s)
     assert appended is True
-    assert s.truncation_watermark is None
+    assert s.truncation_watermark is not None
+    assert s.truncation_watermark >= 100.0  # advanced past old watermark
     assert any(m.get("content") == "new turn after edit" for m in s.messages)
 
 
@@ -151,17 +196,18 @@ def test_error_path_materialize_preserves_zero_sentinel():
     assert s.truncation_watermark == 0.0
 
 
-def test_eager_checkpoint_clears_positive_watermark():
+def test_eager_checkpoint_advances_positive_watermark():
     """The eager first-turn checkpoint path (_checkpoint_user_message_for_eager_session_save
-    in routes.py) inlines the same clear. A committed user turn retires a stale
-    positive watermark; the 0.0 sentinel is preserved."""
+    in routes.py) inlines the same advance. A committed user turn advances a
+    stale positive watermark to the new message timestamp; the 0.0 sentinel is
+    preserved."""
     import api.routes as routes
 
     s = _FakeSession(100.0)
     routes._checkpoint_user_message_for_eager_session_save(
         s, "eager new turn", None, started_at=200.0
     )
-    assert s.truncation_watermark is None
+    assert s.truncation_watermark == 200.0
     assert any(m.get("content") == "eager new turn" for m in s.messages)
 
     s0 = _FakeSession(0.0)
@@ -170,3 +216,28 @@ def test_eager_checkpoint_clears_positive_watermark():
     )
     assert s0.truncation_watermark == 0.0
 
+
+# --- Advanced watermark filters pre-edit state.db rows -------------------------
+
+def test_advanced_watermark_filters_pre_edit_state_db_rows():
+    """With the watermark advanced to the new message timestamp, pre-edit state.db
+    rows whose timestamps are below the watermark AND whose content is not in the
+    sidecar are still filtered out."""
+    sidecar = _rows(
+        ("user", "square", 200),
+        ("assistant", "360", 201),
+    )
+    state_db = _rows(
+        ("user", "triangle", 100),    # pre-edit, below watermark, not in sidecar
+        ("assistant", "180", 101),    # pre-edit assistant, below watermark
+        ("user", "square", 200),      # in sidecar
+        ("assistant", "360", 201),    # in sidecar
+    )
+    merged = models.merge_session_messages_append_only(
+        sidecar, state_db, truncation_watermark=200.0
+    )
+    contents = [m["content"] for m in merged]
+    assert "triangle" not in contents
+    assert "180" not in contents
+    assert "square" in contents
+    assert "360" in contents

tests/test_session_duplicate_fields.py

@@ -73,6 +73,16 @@ def test_duplicate_copies_truncation_watermark():
         "Duplicate must copy truncation_watermark"
 
 
+def test_duplicate_copies_truncation_boundary():
+    """truncation_boundary must be copied alongside truncation_watermark so
+    empty-sidecar recovery in the duplicate can distinguish legitimate prefix
+    from deleted suffix."""
+    block, _ = _extract_duplicate_block()
+    ctor = _find_session_ctor(block)
+    assert _has_field(ctor, 'truncation_boundary'), \
+        "Duplicate must copy truncation_boundary"
+
+
 def test_duplicate_copies_context_messages():
     """context_messages is the authoritative model-facing prefix. Without it,
     the duplicate's agent context diverges from what the user sees."""
@@ -419,6 +429,16 @@ def test_branch_does_not_copy_truncation_watermark():
         "truncation_watermark should NOT be copied in branch"
 
 
+def test_branch_does_not_copy_truncation_boundary():
+    """truncation_boundary must NOT be copied in branch — same reasoning as
+    truncation_watermark: the branch is a fresh message slice with no truncate
+    history of its own."""
+    block, _ = _extract_branch_block()
+    ctor = _find_session_ctor(block, 'branch')
+    assert not _has_field(ctor, 'truncation_boundary'), \
+        "truncation_boundary should NOT be copied in branch"
+
+
 def test_branch_does_not_copy_usage_counters():
     """Branch starts a new conversation — token counters must NOT carry over."""
     block, _ = _extract_branch_block()

tests/test_watermark_advance_after_edit.py

@@ -0,0 +1,386 @@
+"""End-to-end regression tests for the edit/session-switch watermark bug.
+
+Scenario: user edits a message, sends a new turn, switches to another session,
+then switches back — the original (pre-edit) message must NOT reappear from
+state.db because the truncation_watermark was advanced (not cleared to None)
+when the new turn was committed.
+
+This covers the exact reproduction path:
+1. Send message A ("triangle")
+2. Edit message A to "square" (truncate + re-send)
+3. Send message B ("light speed") — watermark advances to B's timestamp
+4. Switch session (evict from LRU cache)
+5. Switch back (reload from disk, merge sidecar + state.db)
+6. Assert: "triangle" is NOT in the merged transcript
+"""
+from __future__ import annotations
+
+import json
+from io import BytesIO
+from types import SimpleNamespace
+
+import api.models as models
+import api.routes as routes
+from api.models import Session
+
+
+def _msg(role: str, content: str, ts: float, mid: str) -> dict:
+    return {"id": mid, "role": role, "content": content, "timestamp": ts}
+
+
+def _setup_session_dir(monkeypatch, tmp_path):
+    """Configure models to use a temp session directory."""
+    session_dir = tmp_path / "sessions"
+    session_dir.mkdir(parents=True, exist_ok=True)
+    monkeypatch.setattr(models, "SESSION_DIR", session_dir)
+    monkeypatch.setattr(models, "SESSION_INDEX_FILE", session_dir / "_index.json")
+    models.SESSIONS.clear()
+
+
+def _post_truncate(monkeypatch, sid, keep_count):
+    """Call POST /api/session/truncate via the real route handler."""
+    import io
+
+    class _JSONHandler:
+        def __init__(self, body_bytes: bytes):
+            self.status = None
+            self.response_headers = []
+            self.rfile = BytesIO(body_bytes)
+            self.headers = {"Content-Length": str(len(body_bytes))}
+            self.wfile = io.BytesIO()
+
+        def send_response(self, status):
+            self.status = status
+
+        def send_header(self, key, value):
+            self.response_headers.append((key, value))
+
+        def end_headers(self):
+            pass
+
+        def payload(self):
+            return json.loads(self.wfile.getvalue().decode("utf-8"))
+
+    body_bytes = json.dumps({"session_id": sid, "keep_count": keep_count}).encode()
+    monkeypatch.setattr(routes, "_check_csrf", lambda handler: True)
+    handler = _JSONHandler(body_bytes)
+    routes.handle_post(handler, SimpleNamespace(path="/api/session/truncate"))
+    return {"status": handler.status, "payload": handler.payload()}
+
+
+def _checkpoint_user(s, msg, started_at, monkeypatch):
+    """Call the eager checkpoint path (simulates sending a new message)."""
+    monkeypatch.setattr(routes, "_check_csrf", lambda handler: True)
+    routes._checkpoint_user_message_for_eager_session_save(
+        s, msg, None, started_at=started_at
+    )
+
+
+def test_edit_then_new_turn_watermark_advances_not_clears(monkeypatch, tmp_path):
+    """After edit → new turn, truncation_watermark should advance to the new
+    message's timestamp (NOT be cleared to None)."""
+    _setup_session_dir(monkeypatch, tmp_path)
+
+    sid = "edit_watermark_advance"
+    # Real edit scenario: message 1 + reply kept, message 2 edited
+    session = Session(
+        session_id=sid,
+        messages=[
+            _msg("user", "first msg", 50.0, "u0"),
+            _msg("assistant", "first reply", 51.0, "a0"),
+            _msg("user", "triangle", 100.0, "u1"),
+            _msg("assistant", "180°", 101.0, "a1"),
+        ],
+    )
+    session.save()
+
+    # Step 1: Truncate (edit) — keep 2 messages (first exchange)
+    captured = _post_truncate(monkeypatch, sid, 2)
+    assert captured["status"] == 200
+
+    loaded = Session.load(sid)
+    assert loaded.truncation_watermark == 51.0  # positive watermark (not 0.0)
+
+    # Step 2: Send new message "square" — checkpoint advances watermark
+    _checkpoint_user(loaded, "square", started_at=200.0, monkeypatch=monkeypatch)
+    loaded.save()
+
+    reloaded = Session.load(sid)
+    # Watermark should be advanced to 200.0, NOT None
+    assert reloaded.truncation_watermark is not None, (
+        "truncation_watermark was cleared to None instead of advanced"
+    )
+    assert reloaded.truncation_watermark == 200.0
+
+
+def test_edit_then_new_turn_then_reload_filters_pre_edit_state_db(monkeypatch, tmp_path):
+    """After edit → new turn → reload from disk, pre-edit state.db rows must
+    NOT appear in the merged transcript."""
+    _setup_session_dir(monkeypatch, tmp_path)
+
+    sid = "edit_reload_filter"
+    # Real edit scenario: first exchange kept, second message edited
+    session = Session(
+        session_id=sid,
+        messages=[
+            _msg("user", "first msg", 50.0, "u0"),
+            _msg("assistant", "first reply", 51.0, "a0"),
+            _msg("user", "triangle", 100.0, "u1"),
+            _msg("assistant", "180°", 101.0, "a1"),
+        ],
+    )
+    session.save()
+
+    # Step 1: Truncate (edit) — keep 2 messages (first exchange)
+    captured = _post_truncate(monkeypatch, sid, 2)
+    assert captured["status"] == 200
+
+    # Step 2: Send new message "square"
+    loaded = Session.load(sid)
+    _checkpoint_user(loaded, "square", started_at=200.0, monkeypatch=monkeypatch)
+    # Simulate assistant reply being appended to sidecar
+    loaded.messages.append(_msg("assistant", "360°", 201.0, "a2"))
+    loaded.save()
+
+    # Step 3: Simulate state.db containing both original and edited messages
+    state_db = [
+        _msg("user", "first msg", 50.0, "state-u0"),
+        _msg("assistant", "first reply", 51.0, "state-a0"),
+        _msg("user", "triangle", 100.0, "state-u1"),
+        _msg("assistant", "180°", 101.0, "state-a1"),
+        _msg("user", "square", 200.0, "state-u2"),
+        _msg("assistant", "360°", 201.0, "state-a2"),
+    ]
+
+    # Step 4: Reload from disk (simulates session switch → switch back)
+    # Evict from cache to force disk reload
+    with models.LOCK:
+        models.SESSIONS.pop(sid, None)
+
+    reloaded = Session.load(sid)
+    assert reloaded.truncation_watermark is not None
+    assert reloaded.truncation_watermark == 200.0
+
+    # Step 5: Merge sidecar + state.db (what GET /api/session does)
+    merged = models.merge_session_messages_append_only(
+        reloaded.messages,
+        state_db,
+        truncation_watermark=reloaded.truncation_watermark,
+    )
+
+    contents = [m["content"] for m in merged]
+    # "triangle" must NOT leak through
+    assert "triangle" not in contents, (
+        f"Pre-edit 'triangle' leaked from state.db! Contents: {contents}"
+    )
+    assert "180°" not in contents, (
+        f"Pre-edit assistant reply leaked from state.db! Contents: {contents}"
+    )
+    # "square" must be present
+    assert "square" in contents
+    assert "360°" in contents
+
+
+def test_edit_then_two_turns_then_reload_all_visible(monkeypatch, tmp_path):
+    """After edit → two new turns → reload, all post-edit turns are visible
+    and pre-edit rows are filtered."""
+    _setup_session_dir(monkeypatch, tmp_path)
+
+    sid = "edit_two_turns"
+    # Real edit scenario: first exchange kept, second message edited
+    session = Session(
+        session_id=sid,
+        messages=[
+            _msg("user", "first msg", 50.0, "u0"),
+            _msg("assistant", "first reply", 51.0, "a0"),
+            _msg("user", "original", 100.0, "u1"),
+            _msg("assistant", "original reply", 101.0, "a1"),
+        ],
+    )
+    session.save()
+
+    # Step 1: Truncate (edit) — keep 2 messages
+    captured = _post_truncate(monkeypatch, sid, 2)
+    assert captured["status"] == 200
+
+    # Step 2: Send edited message
+    loaded = Session.load(sid)
+    _checkpoint_user(loaded, "edited", started_at=200.0, monkeypatch=monkeypatch)
+    loaded.save()
+
+    # Step 3: Simulate assistant reply
+    loaded.messages.append(_msg("assistant", "edited reply", 201.0, "a2"))
+    loaded.save()
+
+    # Step 4: Send second new turn
+    _checkpoint_user(loaded, "second question", started_at=300.0, monkeypatch=monkeypatch)
+    # Simulate assistant reply to second question
+    loaded.messages.append(_msg("assistant", "second reply", 301.0, "a3"))
+    loaded.save()
+
+    # Step 5: Simulate state.db with all messages
+    state_db = [
+        _msg("user", "first msg", 50.0, "state-u0"),
+        _msg("assistant", "first reply", 51.0, "state-a0"),
+        _msg("user", "original", 100.0, "state-u1"),
+        _msg("assistant", "original reply", 101.0, "state-a1"),
+        _msg("user", "edited", 200.0, "state-u2"),
+        _msg("assistant", "edited reply", 201.0, "state-a2"),
+        _msg("user", "second question", 300.0, "state-u3"),
+        _msg("assistant", "second reply", 301.0, "state-a3"),
+    ]
+
+    # Step 6: Reload from disk
+    with models.LOCK:
+        models.SESSIONS.pop(sid, None)
+
+    reloaded = Session.load(sid)
+    assert reloaded.truncation_watermark is not None
+    assert reloaded.truncation_watermark == 300.0
+
+    # Step 7: Merge
+    merged = models.merge_session_messages_append_only(
+        reloaded.messages,
+        state_db,
+        truncation_watermark=reloaded.truncation_watermark,
+    )
+
+    contents = [m["content"] for m in merged]
+    # Pre-edit rows filtered
+    assert "original" not in contents
+    assert "original reply" not in contents
+    # Post-edit rows present
+    assert "edited" in contents
+    assert "edited reply" in contents
+    assert "second question" in contents
+    assert "second reply" in contents
+
+
+def test_retry_then_new_turn_watermark_advances(monkeypatch, tmp_path):
+    """After /retry → new turn, watermark should advance (not clear to None)."""
+    _setup_session_dir(monkeypatch, tmp_path)
+
+    from api.session_ops import retry_last
+
+    sid = "retry_watermark_advance"
+    session = Session(
+        session_id=sid,
+        messages=[
+            _msg("user", "first", 100.0, "u1"),
+            _msg("assistant", "reply", 101.0, "a1"),
+            _msg("user", "second", 200.0, "u2"),
+            _msg("assistant", "reply2", 201.0, "a2"),
+        ],
+    )
+    session.save()
+
+    # Step 1: Retry — truncates to before "second"
+    retry_last(sid)
+    loaded = Session.load(sid)
+    assert loaded.truncation_watermark == 101.0
+    assert [m["content"] for m in loaded.messages] == ["first", "reply"]
+
+    # Step 2: Send new message
+    _checkpoint_user(loaded, "new question", started_at=300.0, monkeypatch=monkeypatch)
+    loaded.save()
+
+    reloaded = Session.load(sid)
+    assert reloaded.truncation_watermark is not None
+    assert reloaded.truncation_watermark == 300.0
+
+
+def test_undo_then_new_turn_watermark_advances(monkeypatch, tmp_path):
+    """After /undo → new turn, watermark should advance (not clear to None)."""
+    _setup_session_dir(monkeypatch, tmp_path)
+
+    from api.session_ops import undo_last
+
+    sid = "undo_watermark_advance"
+    session = Session(
+        session_id=sid,
+        messages=[
+            _msg("user", "first", 100.0, "u1"),
+            _msg("assistant", "reply", 101.0, "a1"),
+            _msg("user", "second", 200.0, "u2"),
+            _msg("assistant", "reply2", 201.0, "a2"),
+        ],
+    )
+    session.save()
+
+    # Step 1: Undo — removes "second" and "reply2"
+    undo_last(sid)
+    loaded = Session.load(sid)
+    assert loaded.truncation_watermark == 101.0
+    assert [m["content"] for m in loaded.messages] == ["first", "reply"]
+
+    # Step 2: Send new message
+    _checkpoint_user(loaded, "new question", started_at=300.0, monkeypatch=monkeypatch)
+    loaded.save()
+
+    reloaded = Session.load(sid)
+    assert reloaded.truncation_watermark is not None
+    assert reloaded.truncation_watermark == 300.0
+
+
+def test_zero_watermark_preserved_through_new_turn(monkeypatch, tmp_path):
+    """The 0.0 truncate-to-empty sentinel (#2914) must survive a new turn —
+    the falsy guard prevents advancing it."""
+    _setup_session_dir(monkeypatch, tmp_path)
+
+    sid = "zero_watermark_preserved"
+    session = Session(
+        session_id=sid,
+        messages=[],
+        truncation_watermark=0.0,
+    )
+    session.save()
+
+    loaded = Session.load(sid)
+    assert loaded.truncation_watermark == 0.0
+
+    # Send new message — 0.0 must NOT be advanced
+    _checkpoint_user(loaded, "new msg", started_at=100.0, monkeypatch=monkeypatch)
+    loaded.save()
+
+    reloaded = Session.load(sid)
+    assert reloaded.truncation_watermark == 0.0, (
+        "0.0 truncate-to-empty sentinel was advanced — #2914 regression!"
+    )
+
+
+def test_empty_sidecar_advanced_watermark_no_ghost_no_data_loss():
+    """Empty sidecar + advanced watermark must keep post-edit rows without
+    resurrecting stale pre-edit ghosts (#4767).
+
+    Reproduction: cold reload / crash recovery before sidecar persists.
+    state.db has the full lineage including replaced pre-edit rows.
+
+    A genuinely-advanced session persists the original cutoff as
+    ``truncation_boundary`` (here @51, the last kept reply) while the watermark
+    is advanced to the new committed turn (@200). The reconstruction keeps the
+    legitimate prefix (ts <= boundary) plus the post-edit tail (ts >= watermark)
+    and drops the replaced (boundary, watermark) suffix. Passing the boundary is
+    required — an advanced watermark with NO boundary is the unreachable
+    legacy/frozen state (see the not-advanced regression tests in
+    test_core_data_loss_cases.py)."""
+    state = [
+        {"role": "user", "content": "first msg", "timestamp": 50},
+        {"role": "assistant", "content": "first reply", "timestamp": 51},
+        {"role": "user", "content": "original pre-edit", "timestamp": 100},
+        {"role": "assistant", "content": "original reply", "timestamp": 101},
+        {"role": "user", "content": "edited/new turn", "timestamp": 200},
+        {"role": "assistant", "content": "post-edit reply", "timestamp": 201},
+    ]
+    merged = models.merge_session_messages_append_only(
+        [], state, truncation_watermark=200.0, truncation_boundary=51.0
+    )
+    contents = [m["content"] for m in merged]
+    wanted = ["first msg", "first reply", "edited/new turn", "post-edit reply"]
+    assert contents == wanted, (
+        f"empty-sidecar + advanced watermark broken: got {contents}, wanted {wanted}"
+    )
+    # Ghost must NOT appear
+    assert "original pre-edit" not in contents
+    assert "original reply" not in contents
+    # Post-edit reply must NOT be dropped
+    assert "post-edit reply" in contents