fix(sidebar): keep newer continuation visible over older snapshot

api/compression_anchor.py

@@ -68,6 +68,7 @@ def is_context_compression_marker(message):
         text.startswith("[context compaction")
         or text.startswith("context compaction")
         or text.startswith("[your active task list was preserved across context compression]")
+        or text.startswith("[session arc summary")
     )
 
 

api/models.py

@@ -2493,6 +2493,14 @@ def _prefer_fuller_snapshots_for_sidebar(sessions: list[dict]) -> list[dict]:
         if _sidebar_message_count(best_snapshot) <= best_visible_count:
             continue
 
+        newest_visible_ts = max(_session_sort_timestamp(session) for session in visible)
+        snapshot_ts = _session_sort_timestamp(best_snapshot)
+        # Keep the active continuation visible when it has newer activity than
+        # the archived snapshot. A fuller snapshot can still be older than a
+        # continuation that contains the latest turns after compression.
+        if newest_visible_ts > snapshot_ts:
+            continue
+
         snapshot_ids_to_show.add(str(best_snapshot.get('session_id')))
         continuation_ids_to_hide.update(
             str(session.get('session_id'))

api/routes.py

@@ -2434,6 +2434,43 @@ def _normalize_sidebar_source_flags(session: dict) -> dict:
     return normalized
 
 
+def _session_source_is_webui(session: dict) -> bool:
+    """Return True for state.db/sidebar rows that describe WebUI-origin sessions."""
+    if not isinstance(session, dict):
+        return False
+    for key in ("source_tag", "raw_source", "session_source", "source"):
+        if str(session.get(key) or "").strip().lower() == "webui":
+            return True
+    return False
+
+
+def _session_lineage_ids(session: dict) -> set[str]:
+    """Return known ids that identify one logical sidebar lineage."""
+    if not isinstance(session, dict):
+        return set()
+    ids: set[str] = set()
+    for key in ("session_id", "_lineage_root_id", "_lineage_tip_id"):
+        value = session.get(key)
+        if value:
+            ids.add(str(value))
+    return ids
+
+
+def _is_duplicate_webui_state_projection(session: dict, represented_webui_ids: set[str]) -> bool:
+    """Return True when a state.db row is only a duplicate WebUI-origin projection.
+
+    The "Show non-WebUI sessions" toggle should add external/agent-owned
+    conversations, not make WebUI compression continuations appear only when the
+    external-session bridge is enabled. WebUI-origin state.db rows are still
+    useful metadata sidecars, but if any id in their compression lineage is
+    already represented by WebUI session JSON, they should not be injected as an
+    additive external row.
+    """
+    if not _session_source_is_webui(session):
+        return False
+    return bool(_session_lineage_ids(session) & represented_webui_ids)
+
+
 CLI_VISIBLE_SESSION_CAP = 20
 
 
@@ -4495,9 +4532,17 @@ def handle_get(handler, parsed) -> bool:
                 # Apply the same CLI visibility semantics to imported local copies so
                 # low-value imported artifacts do not leak into the sidebar.
                 webui_sessions = [s for s in webui_sessions if is_cli_session_row_visible(s)]
-                webui_ids = {s["session_id"] for s in webui_sessions}
+                represented_webui_ids = set()
+                for s in webui_sessions:
+                    represented_webui_ids.update(_session_lineage_ids(s))
                 from api.models import _hide_from_default_sidebar as _cron_hide
-                deduped_cli = [s for s in cli if s["session_id"] not in webui_ids and is_cli_session_row_visible(s) and not _cron_hide(s)]
+                deduped_cli = [
+                    s for s in cli
+                    if s["session_id"] not in represented_webui_ids
+                    and not _is_duplicate_webui_state_projection(s, represented_webui_ids)
+                    and is_cli_session_row_visible(s)
+                    and not _cron_hide(s)
+                ]
             else:
                 diag.stage("filter_webui_sessions")
                 webui_sessions = [s for s in webui_sessions if not _is_cli_session_for_settings(s)]

api/streaming.py

@@ -2890,6 +2890,60 @@ def _merge_display_messages_after_agent_result(previous_display, previous_contex
     if not result_messages:
         return previous_display
 
+    # ── Backfill normal turns from previous_context that are missing from
+    # previous_display.  After context compression recovery, previous_context
+    # can contain user/assistant turns that were never rendered in the visible
+    # transcript (they were behind a compression marker). On the next
+    # append-only merge those turns sit inside the shared prefix and get
+    # stripped, leaving them permanently invisible.  Reinsert them now.
+    #
+    # Use display as the backbone to preserve visible order. Walk display in
+    # order and for each display message search for its identity in context
+    # at/after a cursor. Any context messages between the cursor and that
+    # match are context-only gaps that get spliced in before the display msg.
+    if previous_display and previous_context:
+        _display_id_set = {_message_identity(m) for m in previous_display}
+        _context_id_set = {_message_identity(m) for m in previous_context}
+        _has_context_only_turns = bool(_context_id_set - _display_id_set)
+        if _has_context_only_turns:
+            context_keys = [_message_identity(m) for m in previous_context]
+            _backfilled = []
+            _emitted = set()
+            _cursor = 0
+            for _dmsg in previous_display:
+                _dkey = _message_identity(_dmsg)
+                if _dkey is not None:
+                    _j = _cursor
+                    while _j < len(context_keys) and context_keys[_j] != _dkey:
+                        _j += 1
+                    if _j < len(context_keys):
+                        for _k in range(_cursor, _j):
+                            _ckey = context_keys[_k]
+                            _cmsg = previous_context[_k]
+                            if _ckey is not None and _ckey not in _emitted and not _is_context_compression_marker(_cmsg):
+                                _backfilled.append(copy.deepcopy(_cmsg))
+                                _emitted.add(_ckey)
+                        _cursor = _j + 1
+                if _dkey not in _emitted:
+                    _backfilled.append(_dmsg)
+                    if _dkey is not None:
+                        _emitted.add(_dkey)
+            while _cursor < len(context_keys):
+                _ckey = context_keys[_cursor]
+                _cmsg = previous_context[_cursor]
+                _cursor += 1
+                if _ckey is not None and _ckey not in _emitted and not _is_context_compression_marker(_cmsg):
+                    _backfilled.append(copy.deepcopy(_cmsg))
+                    _emitted.add(_ckey)
+            if len(_backfilled) > len(previous_display):
+                logger.debug(
+                    "Backfilled %d context-only turns into previous_display (was %d, now %d)",
+                    len(_backfilled) - len(previous_display),
+                    len(previous_display),
+                    len(_backfilled),
+                )
+                previous_display = _backfilled
+
     if _messages_have_prefix(result_messages, previous_context):
         candidates = result_messages[len(previous_context):]
         candidates = _strip_replayed_prefix(previous_display, candidates)

tests/test_context_message_dedup.py

@@ -173,3 +173,171 @@ def test_merge_display_messages_preserves_current_user_turn():
     # Current user message should use msg_text
     user_msgs = [m for m in merged if m.get("role") == "user"]
     assert any(m.get("content") == "next question" for m in user_msgs)
+
+
+def test_merge_display_backfill_preserves_visible_head_ordering():
+    """Display head must stay before hidden context-only middle turns.
+
+    A compacted session can have a visible transcript head that is absent from
+    model context, plus a later visible tail that is present in model context.
+    When model-only middle turns are restored, the merged order must be:
+
+        old visible head
+        hidden context-only middle turn(s)
+        current visible tail
+        new current turn
+    """
+    from api.streaming import _merge_display_messages_after_agent_result
+
+    previous_display = [
+        {"role": "user", "content": "visible head user turn"},
+        {"role": "assistant", "content": "visible head assistant turn"},
+        {"role": "user", "content": "visible tail user turn"},
+    ]
+    previous_context = [
+        {"role": "user", "content": "context-only middle user turn"},
+        {"role": "assistant", "content": "context-only middle assistant turn"},
+        {"role": "user", "content": "visible tail user turn"},
+    ]
+    result_messages = previous_context + [
+        {"role": "user", "content": "new follow-up user turn"},
+        {"role": "assistant", "content": "new follow-up assistant turn"},
+    ]
+    msg_text = "new follow-up user turn"
+
+    merged = _merge_display_messages_after_agent_result(
+        previous_display, previous_context, result_messages, msg_text
+    )
+
+    user_texts = [
+        m.get("content", "")
+        for m in merged
+        if isinstance(m, dict) and m.get("role") == "user"
+    ]
+
+    head_idx = next(i for i, t in enumerate(user_texts) if "visible head" in t)
+    middle_idx = next(i for i, t in enumerate(user_texts) if "context-only middle" in t)
+    tail_idx = next(i for i, t in enumerate(user_texts) if "visible tail" in t)
+    followup_idx = next(i for i, t in enumerate(user_texts) if "new follow-up" in t)
+
+    assert head_idx < middle_idx, f"Visible head must precede restored context middle; got indices {head_idx} vs {middle_idx}"
+    assert middle_idx < tail_idx, f"Restored context middle must precede visible tail; got indices {middle_idx} vs {tail_idx}"
+    assert tail_idx < followup_idx, f"Visible tail must precede new turn; got indices {tail_idx} vs {followup_idx}"
+
+
+def test_merge_display_backfills_context_only_turns_missing_from_display():
+    """Normal user/assistant turns present in previous_context but absent from
+    previous_display must be restored into the visible transcript.
+
+    This reproduces the generic bug where context compression recovery expands
+    previous_context with normal turns that never appear in previous_display.
+    A subsequent append-only merge skips over the shared context prefix, so
+    without backfill those turns remain permanently invisible in the WebUI.
+    """
+    from api.streaming import _merge_display_messages_after_agent_result
+
+    previous_display = [
+        {"role": "user", "content": "visible head user turn"},
+        {"role": "assistant", "content": "visible head assistant turn"},
+    ]
+    previous_context = [
+        {"role": "user", "content": "visible head user turn"},
+        {"role": "assistant", "content": "visible head assistant turn"},
+        {"role": "user", "content": "context-only middle user turn"},
+        {"role": "assistant", "content": "context-only middle assistant turn"},
+    ]
+    result_messages = previous_context + [
+        {"role": "user", "content": "new follow-up user turn"},
+        {"role": "assistant", "content": "new follow-up assistant turn"},
+    ]
+    msg_text = "new follow-up user turn"
+
+    merged = _merge_display_messages_after_agent_result(
+        previous_display, previous_context, result_messages, msg_text
+    )
+
+    merged_texts = [
+        (m.get("role"), _message_text_safe(m))
+        for m in merged
+        if isinstance(m, dict) and m.get("role") in ("user", "assistant")
+    ]
+
+    assert any(
+        "context-only middle user turn" in text
+        for role, text in merged_texts
+        if role == "user"
+    ), f"Missing context-only user turn from visible transcript; got: {merged_texts}"
+
+    assert any(
+        "context-only middle assistant turn" in text
+        for role, text in merged_texts
+        if role == "assistant"
+    ), f"Missing context-only assistant turn from visible transcript; got: {merged_texts}"
+
+    assert any(
+        "new follow-up user turn" in text
+        for role, text in merged_texts
+        if role == "user"
+    ), "New current turn should also be present"
+
+    head_idx = next(i for i, (r, t) in enumerate(merged_texts) if "visible head" in t)
+    middle_idx = next(i for i, (r, t) in enumerate(merged_texts) if "context-only middle" in t)
+    assert head_idx < middle_idx, f"Display head must come before backfilled context turn; got indices {head_idx} vs {middle_idx}"
+
+
+def test_merge_display_backfill_does_not_reintroduce_compression_markers():
+    """Context compression markers in previous_context that were intentionally
+    removed from previous_display must NOT be restored by the backfill logic."""
+    from api.streaming import _merge_display_messages_after_agent_result
+
+    previous_display = [
+        {"role": "user", "content": "first question"},
+        {"role": "assistant", "content": "first answer"},
+    ]
+    previous_context = [
+        {"role": "user", "content": "first question"},
+        {"role": "assistant", "content": "first answer"},
+        {"role": "assistant", "content": "[context compaction] prior messages summarized"},
+        {"role": "user", "content": "context-only middle user turn"},
+        {"role": "assistant", "content": "context-only middle assistant turn"},
+    ]
+    result_messages = previous_context + [
+        {"role": "user", "content": "next question"},
+        {"role": "assistant", "content": "next answer"},
+    ]
+    msg_text = "next question"
+
+    merged = _merge_display_messages_after_agent_result(
+        previous_display, previous_context, result_messages, msg_text
+    )
+
+    merged_texts = [
+        _message_text_safe(m)
+        for m in merged
+        if isinstance(m, dict) and m.get("role") == "assistant"
+    ]
+
+    assert not any(
+        "[context compaction]" in t for t in merged_texts
+    ), f"Compression marker should not be in visible display; got: {merged_texts}"
+
+    assert any(
+        "context-only middle user turn" in _message_text_safe(m)
+        for m in merged
+        if isinstance(m, dict) and m.get("role") == "user"
+    ), "Normal user turn from context should be backfilled"
+
+
+def _message_text_safe(msg):
+    """Extract plain text from a message content field (list or string)."""
+    if not isinstance(msg, dict):
+        return ""
+    content = msg.get("content", "")
+    if isinstance(content, str):
+        return content
+    if isinstance(content, list):
+        return " ".join(
+            part.get("text", "") for part in content
+            if isinstance(part, dict) and isinstance(part.get("text"), str)
+        )
+    return str(content or "")

tests/test_session_import_cli_fallback_model.py

@@ -14,12 +14,34 @@ proceeds with a sensible default rather than crashing.
 
 from __future__ import annotations
 
+import io
+import json
 from pathlib import Path
+from urllib.parse import urlparse
 
 REPO = Path(__file__).resolve().parents[1]
 ROUTES_PY = (REPO / "api" / "routes.py").read_text(encoding="utf-8")
 
 
+class _FakeHandler:
+    def __init__(self):
+        self.status = None
+        self.headers = {}
+        self.wfile = io.BytesIO()
+
+    def send_response(self, status):
+        self.status = status
+
+    def send_header(self, key, value):
+        self.headers[key] = value
+
+    def end_headers(self):
+        pass
+
+    def json_body(self):
+        return json.loads(self.wfile.getvalue().decode("utf-8"))
+
+
 def _extract_handler(name: str) -> str:
     """Return the source of the handler function `name` from api/routes.py."""
     marker = f"def {name}("
@@ -274,6 +296,99 @@ def test_merge_cli_sidebar_metadata_keeps_larger_sidecar_message_count():
     assert merged["message_count"] == 535
 
 
+def test_webui_state_projection_dedupes_by_lineage_root():
+    """WebUI-origin state.db projections should not be additive non-WebUI rows."""
+    import api.routes as routes
+
+    represented = {"root_sid"}
+    state_projection = {
+        "session_id": "tip_sid",
+        "source_tag": "webui",
+        "raw_source": "webui",
+        "session_source": "webui",
+        "_lineage_root_id": "root_sid",
+        "_lineage_tip_id": "tip_sid",
+    }
+
+    assert routes._is_duplicate_webui_state_projection(state_projection, represented) is True
+
+
+def test_external_state_projection_not_deduped_by_webui_source_guard():
+    """The WebUI-source guard must not hide real external conversations."""
+    import api.routes as routes
+
+    represented = {"root_sid"}
+    external_projection = {
+        "session_id": "tip_sid",
+        "source_tag": "telegram",
+        "raw_source": "telegram",
+        "session_source": "messaging",
+        "_lineage_root_id": "root_sid",
+        "_lineage_tip_id": "tip_sid",
+    }
+
+    assert routes._is_duplicate_webui_state_projection(external_projection, represented) is False
+
+
+def test_sessions_endpoint_suppresses_duplicate_webui_state_projection(monkeypatch):
+    """The /api/sessions merge should not add WebUI state.db lineage duplicates."""
+    import api.profiles as profiles
+    import api.routes as routes
+
+    monkeypatch.setattr(routes, "_reconcile_stale_stream_state_for_session_rows", lambda _sessions: False)
+    monkeypatch.setattr(routes, "load_settings", lambda: {"show_cli_sessions": True})
+    monkeypatch.setattr(profiles, "get_active_profile_name", lambda: "default")
+
+    webui_row = {
+        "session_id": "visible_tip",
+        "title": "Long Conversation",
+        "profile": "default",
+        "updated_at": 20,
+        "last_message_at": 20,
+        "source_tag": "webui",
+        "raw_source": "webui",
+        "session_source": "webui",
+        "_lineage_root_id": "root_sid",
+        "_lineage_tip_id": "visible_tip",
+    }
+    duplicate_webui_projection = {
+        "session_id": "state_projection_tip",
+        "title": "Long Conversation",
+        "profile": "default",
+        "updated_at": 30,
+        "last_message_at": 30,
+        "source_tag": "webui",
+        "raw_source": "webui",
+        "session_source": "webui",
+        "_lineage_root_id": "root_sid",
+        "_lineage_tip_id": "state_projection_tip",
+    }
+    external_projection = {
+        "session_id": "telegram_tip",
+        "title": "External Thread",
+        "profile": "default",
+        "updated_at": 10,
+        "last_message_at": 10,
+        "source_tag": "telegram",
+        "raw_source": "telegram",
+        "session_source": "messaging",
+        "_lineage_root_id": "root_sid",
+        "_lineage_tip_id": "telegram_tip",
+    }
+
+    monkeypatch.setattr(routes, "all_sessions", lambda diag=None: [webui_row])
+    monkeypatch.setattr(routes, "get_cli_sessions", lambda: [duplicate_webui_projection, external_projection])
+
+    handler = _FakeHandler()
+    routes.handle_get(handler, urlparse("http://example.com/api/sessions"))
+
+    assert handler.status == 200
+    session_ids = [row["session_id"] for row in handler.json_body()["sessions"]]
+    assert "visible_tip" in session_ids
+    assert "state_projection_tip" not in session_ids
+    assert "telegram_tip" in session_ids
+
+
 def test_messaging_session_loader_prefers_longer_sidecar_transcript():
     """Pin the /api/session invariant that repaired sidecars can be longer than state.db segments."""
     handler = _extract_handler("handle_get")

tests/test_session_index.py

@@ -423,8 +423,8 @@ def test_pre_compression_snapshot_hidden_from_active_sidebar_but_file_remains(mo
         parent_session_id="old_sid",
         updated_at=200.0,
     )
-    snapshot.save()
-    continuation.save()
+    snapshot.save(touch_updated_at=False)
+    continuation.save(touch_updated_at=False)
     monkeypatch.setattr(models, "_enrich_sidebar_lineage_metadata", lambda _sessions: None)
 
     rows = models.all_sessions()
@@ -452,16 +452,18 @@ def test_fuller_pre_compression_snapshot_replaces_shorter_visible_segment(monkey
         ],
         pre_compression_snapshot=True,
         updated_at=300.0,
+        last_message_at=300.0,
     )
     continuation = Session(
         session_id="short_child",
         title="Long Conversation",
         messages=[{"role": "user", "content": "first"}],
         parent_session_id="full_parent",
-        updated_at=400.0,
+        updated_at=250.0,
+        last_message_at=250.0,
     )
-    snapshot.save()
-    continuation.save()
+    snapshot.save(touch_updated_at=False)
+    continuation.save(touch_updated_at=False)
     monkeypatch.setattr(models, "_enrich_sidebar_lineage_metadata", lambda _sessions: None)
 
     rows = models.all_sessions()
@@ -471,6 +473,48 @@ def test_fuller_pre_compression_snapshot_replaces_shorter_visible_segment(monkey
     assert rows[0]["pre_compression_snapshot"] is True
 
 
+def test_newer_continuation_beats_older_fuller_snapshot(monkeypatch):
+    """Do not hide a newer continuation behind an older fuller snapshot.
+
+    Compression snapshots can have a higher message count while still being
+    older than the continuation that contains the latest user-visible turns.
+    The sidebar should keep the newer continuation visible in that case.
+    """
+    snapshot = Session(
+        session_id="older_full_parent",
+        title="Long Conversation",
+        messages=[
+            {"role": "user", "content": "first"},
+            {"role": "assistant", "content": "second"},
+            {"role": "user", "content": "third"},
+            {"role": "assistant", "content": "fourth"},
+        ],
+        pre_compression_snapshot=True,
+        updated_at=300.0,
+        last_message_at=300.0,
+    )
+    continuation = Session(
+        session_id="newer_short_child",
+        title="Long Conversation",
+        messages=[
+            {"role": "user", "content": "latest task"},
+            {"role": "assistant", "content": "latest result"},
+        ],
+        parent_session_id="older_full_parent",
+        updated_at=450.0,
+        last_message_at=450.0,
+    )
+    snapshot.save(touch_updated_at=False)
+    continuation.save(touch_updated_at=False)
+    monkeypatch.setattr(models, "_enrich_sidebar_lineage_metadata", lambda _sessions: None)
+
+    rows = models.all_sessions()
+
+    assert [row["session_id"] for row in rows] == ["newer_short_child"]
+    assert rows[0]["pre_compression_snapshot"] is False
+    assert rows[0]["message_count"] == 2
+
+
 def test_session_save_does_not_persist_metadata_message_count_hint():
     s = Session(
         session_id="sess_private_hint",