diff --git a/agent/conversation_compression.py b/agent/conversation_compression.py
index 1e85ee1cde..334c31b20c 100644
--- a/agent/conversation_compression.py
+++ b/agent/conversation_compression.py
@@ -3616,19 +3616,18 @@ def _held_watermark(agent: Any, watermark: Optional[int], messages: list, verbat
     history and search, and the summary never saw them. Above the cap they take the concurrent-append path
     instead (cloned after the compacted set).
 
-    Only while the held history is a live prefix of the session: its newest durable row names its ``_row_id``
-    (one held without it could sit above the cap and be cloned beside its own carried copy), and that row is
-    still active (after another surface compacted, the held rows are archived and every live row would be
-    cloned beside the new summary).
+    Only while the held history is a live prefix of the session: its LAST row names its ``_row_id`` (a
+    trailing row of unknown provenance may be durable under the lease watermark without any stamp, e.g. the
+    TUI model-switch marker appended to history and written with a bare ``append_message``; capped below it,
+    the clone would land beside its own carried copy), and that row is still active (after another surface
+    compacted, the held rows are archived and every live row would be cloned beside the new summary).
     """
     if watermark is None:
         return None
-    from agent.context_compressor import _DB_PERSISTED_MARKER
-    rows = [*((m, False) for m in messages), *((m, True) for m in verbatim_tail or ())]
-    held = [m.get("_row_id") for m, _ in rows if isinstance(m, dict)]
+    rows = [*messages, *(verbatim_tail or ())]
+    held = [m.get("_row_id") for m in rows if isinstance(m, dict)]
     held = [rid for rid in held if isinstance(rid, int) and not isinstance(rid, bool) and rid > 0]
-    newest = next((m for m, kept in reversed(rows) if isinstance(m, dict)
-                   and (kept or "_row_id" in m or m.get(_DB_PERSISTED_MARKER))), None)
+    newest = next((m for m in reversed(rows) if isinstance(m, dict)), None)
     if newest is None or newest.get("_row_id") not in held or max(held) >= watermark:
         return watermark
     if agent._session_db.get_message_role(agent.session_id, max(held)) is None:
diff --git a/tests/agent/test_conversation_compression_manual.py b/tests/agent/test_conversation_compression_manual.py
index c55e6c65ab..95e694b915 100644
--- a/tests/agent/test_conversation_compression_manual.py
+++ b/tests/agent/test_conversation_compression_manual.py
@@ -240,17 +240,23 @@ def test_in_place_compress_keeps_turns_the_caller_never_held(session_db, raw):
     assert session_db.search_messages("vault 7741")
 
 
-@pytest.mark.parametrize("stale", ["compacted elsewhere", "newest rows without ids"])
+@pytest.mark.parametrize("stale", ["compacted elsewhere", "newest rows without ids", "trailing row without any stamp"])
 def test_in_place_compress_never_leaves_two_live_copies_of_a_row(session_db, stale):
     """Stopping the archive at the caller's newest row is only exact while the held history is a live prefix of the
-    session. After another surface compacted it, every live row is newer than the held (now archived) ones; and a
-    durable row held without its row id may sit above the stop. Either way the rows above it would be cloned beside
-    their own copies in the new transcript."""
+    session. After another surface compacted it, every live row is newer than the held (now archived) ones; a
+    durable row held without its row id may sit above the stop; and a trailing row held with neither a row id nor
+    the persisted marker can still be durable under the lease watermark (the TUI model-switch marker is appended
+    to history and written with a bare ``append_message``). Either way the rows above the stop would be cloned
+    beside their own copies in the new transcript."""
     agent, _ = _stored_agent(session_db, _exchanges(10))
     held = session_db.get_resume_conversations("sid")[0]
     if stale == "compacted elsewhere":
         other, _ = _stored_agent(session_db, [], create=False)
         assert _compress(other, session_db.get_messages_as_conversation("sid"), "").status == "compressed"
+    elif stale == "trailing row without any stamp":
+        marker = "[Model switched to test/other.]"
+        held.append({"role": "user", "content": marker, "display_kind": "model_switch"})
+        session_db.append_message("sid", "user", marker, display_kind="model_switch")
     else:
         for message in held[-2:]:
             message.pop("_row_id")