fix(compression): a trailing held row without any stamp keeps the lease watermark

_held_watermark skipped trailing dicts that carried neither `_row_id` nor the persisted marker
when picking the "newest held row", assuming such a row is not durable. The TUI model-switch
marker is: server.py appends the bare dict to session["history"] and writes it with a plain
append_message, stamping nothing. With that row last in history a plain /compress capped the
archive at the previous stamped row, so the marker's durable row sat above the cap, was cloned
as a "concurrent append" AND inserted from the compacted set: two live copies (probe on the PR
head: rows 32 and 33 both the marker; origin/main keeps one).

Any trailing dict of unknown provenance now disables the cap (lease watermark, today's
behaviour) instead of being looked past. The regression case rides the existing
never-leaves-two-live-copies test as a third parameter; red on the previous predicate.

Review-fix on #120156.

(cherry picked from commit b114f564f0eddb3ccfe16123e364097747562928)

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:

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")