fix(session): preserve sidecar truncation boundary

api/models.py

@@ -415,6 +415,7 @@ class Session:
                  context_engine_state=None,
                  context_length=None, threshold_tokens=None,
                  last_prompt_tokens=None,
+                 truncation_watermark=None,
                  gateway_routing=None, gateway_routing_history=None,
                  llm_title_generated: bool=False,
                 parent_session_id: str=None,
@@ -461,6 +462,7 @@ class Session:
         self.context_length = context_length
         self.threshold_tokens = threshold_tokens
         self.last_prompt_tokens = last_prompt_tokens
+        self.truncation_watermark = truncation_watermark
         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)
@@ -490,6 +492,18 @@ class Session:
     def path(self):
         return SESSION_DIR / f'{self.session_id}.json'
 
+    def _maybe_clear_truncation_watermark(self) -> None:
+        watermark = _message_timestamp_as_float({"timestamp": self.truncation_watermark})
+        if watermark is None:
+            return
+        max_message_timestamp = None
+        for msg in self.messages or []:
+            timestamp = _message_timestamp_as_float(msg)
+            if timestamp is not None:
+                max_message_timestamp = timestamp if max_message_timestamp is None else max(max_message_timestamp, timestamp)
+        if max_message_timestamp is not None and max_message_timestamp > watermark:
+            self.truncation_watermark = None
+
     def save(self, touch_updated_at: bool = True, skip_index: bool = False) -> None:
         # ── #1558 P0 guard ──────────────────────────────────────────────
         # Refuse to save a session that was loaded with metadata_only=True.
@@ -510,6 +524,7 @@ class Session:
             )
         if touch_updated_at:
             self.updated_at = time.time()
+        self._maybe_clear_truncation_watermark()
         # Write metadata fields first so load_metadata_only() can read them
         # without parsing the full messages array (which may be 400KB+).
         # Fields are listed in the order they should appear in the JSON file.
@@ -525,6 +540,7 @@ class Session:
             'context_engine', 'compression_anchor_engine', 'compression_anchor_mode',
             'compression_anchor_details', 'context_engine_state',
             'context_length', 'threshold_tokens', 'last_prompt_tokens',
+            'truncation_watermark',
             'gateway_routing', 'gateway_routing_history', 'llm_title_generated',
             'parent_session_id',
             'worktree_path', 'worktree_branch', 'worktree_repo_root', 'worktree_created_at',
@@ -3285,13 +3301,27 @@ def state_db_delta_after_context(sidecar_context: list, state_messages: list) ->
     return state_messages[best_len:]
 
 
-def merge_session_messages_append_only(sidecar_messages: list, state_messages: list) -> list:
+def merge_session_messages_append_only(
+    sidecar_messages: list,
+    state_messages: list,
+    *,
+    truncation_watermark=None,
+) -> list:
     """Merge sidecar/context and state.db messages without deleting local rows."""
     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:
+            return [
+                msg for msg in state_messages
+                if (
+                    (timestamp := _message_timestamp_as_float(msg)) is not None
+                    and timestamp <= watermark_timestamp
+                )
+            ]
         return state_messages
 
     merged_messages = []
@@ -3336,6 +3366,13 @@ def merge_session_messages_append_only(sidecar_messages: list, state_messages: l
                     skipped_state_visible_counts.get(matched_visible_key, 0) + 1
                 )
             continue
+        if (
+            watermark_timestamp is not None
+            and timestamp is not None
+            and timestamp > watermark_timestamp
+            and key not in seen_message_keys
+        ):
+            continue
         if max_sidecar_timestamp is not None and timestamp is not None and timestamp <= max_sidecar_timestamp:
             if key in seen_message_keys:
                 continue
@@ -3392,7 +3429,11 @@ def reconciled_state_db_messages_for_session(
         state_messages = get_state_db_session_messages(getattr(session, 'session_id', None))
     if prefer_context and local_messages:
         state_messages = state_db_delta_after_context(local_messages, state_messages)
-    return merge_session_messages_append_only(local_messages, state_messages)
+    return merge_session_messages_append_only(
+        local_messages,
+        state_messages,
+        truncation_watermark=getattr(session, "truncation_watermark", None),
+    )
 
 
 def get_cli_session_messages(sid) -> list: