fix(clients): hide compaction carriers across surfaces

agent/compaction_display.py

@@ -0,0 +1,47 @@
+"""Client-facing projection helpers for model-only compaction carriers."""
+
+from __future__ import annotations
+
+from typing import Any, Dict, Optional
+
+from agent.context_compressor import (
+    ContextCompressor,
+    is_compaction_summary_message,
+)
+
+
+_COMPACTION_INTERNAL_FIELDS = (
+    "tool_calls",
+    "finish_reason",
+    "reasoning",
+    "reasoning_content",
+    "reasoning_details",
+    "codex_reasoning_items",
+    "codex_message_items",
+)
+
+
+def project_compaction_message_for_display(
+    message: Dict[str, Any],
+) -> Optional[Dict[str, Any]]:
+    """Return authentic transcript content, or ``None`` for a pure handoff.
+
+    Model-facing recovery history retains the complete carrier. Display
+    projections instead remove the handoff, inherited tool state, and internal
+    reasoning while preserving any real prior-tail content or live user ask
+    embedded in the carrier.
+    """
+    if not isinstance(message, dict):
+        return None
+    if not is_compaction_summary_message(message):
+        return message.copy()
+
+    projected = ContextCompressor._strip_context_summary_handoff_message(message)
+    if projected is None:
+        return None
+
+    projected = projected.copy()
+    for key in _COMPACTION_INTERNAL_FIELDS:
+        projected.pop(key, None)
+    projected.pop("display_kind", None)
+    return projected

gateway/platforms/api_server.py

@@ -441,29 +441,23 @@ def _project_client_message(message: Dict[str, Any]) -> Dict[str, Any]:
     summary delimiter. Tool calls are dropped from both shapes because a
     carrier's inherited calls are historical context, not live client output.
     """
-    projected = message.copy()
+    from agent.compaction_display import project_compaction_message_for_display
-    if not _is_compressed_summary_message(projected):
-        return projected
 
-    from agent.context_compressor import ContextCompressor
+    projected = project_compaction_message_for_display(message)
-
+    if projected is None:
-    unwrapped = ContextCompressor._strip_context_summary_handoff_message(
+        projected = message.copy()
-        projected
+        for internal_key in (
-    )
+            "tool_calls",
-    for internal_key in (
+            "finish_reason",
-        "tool_calls",
+            "reasoning",
-        "finish_reason",
+            "reasoning_content",
-        "reasoning",
+            "reasoning_details",
-        "reasoning_content",
+            "codex_reasoning_items",
-    ):
+            "codex_message_items",
-        projected.pop(internal_key, None)
+        ):
-    if unwrapped is None:
+            projected.pop(internal_key, None)
         projected["content"] = ""
         projected["display_kind"] = "hidden"
-        return projected
-
-    projected["content"] = unwrapped.get("content")
-    projected.pop("display_kind", None)
     return projected
 
 

gateway/run.py

@@ -59,6 +59,7 @@ from agent.conversation_compression import (
     PREFLIGHT_COMPRESSION_STATUS_TEMPLATE,
 )
 from agent.conversation_loop import INTERRUPT_WAITING_FOR_MODEL_PREFIX
+from agent.compaction_display import project_compaction_message_for_display
 from agent.i18n import t
 from agent.interrupt_compat import request_hard_interrupt
 from agent.turn_context import (
@@ -23308,8 +23309,11 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
         last_assistant = None
         try:
             for message in reversed(await self._session_db.get_messages(session_id)):
-                if message.get("role") == "assistant" and message.get("content"):
+                if message.get("role") != "assistant":
-                    last_assistant = str(message.get("content"))
+                    continue
+                projected = project_compaction_message_for_display(message)
+                if projected is not None and projected.get("content"):
+                    last_assistant = str(projected.get("content"))
                     break
         except Exception:
             last_assistant = None

hermes_state.py

@@ -54,6 +54,7 @@ from hermes_state_common import (  # noqa: F401  (re-exported for back-compat)
     _FTS_CJK_TRIGGERS,
     _FTS_TRIGGERS,
     _LISTABLE_CHILD_SQL,
+    _PREVIEW_ELIGIBLE_SQL,
     _PREVIEW_RAW_SELECT,
     _RESET_END_REASONS,
     _RESET_END_REASONS_SQL,
@@ -8878,6 +8879,7 @@ class SessionDB(SessionSearchMixin, SessionSchemaMixin, SessionPortabilityMixin)
                         (SELECT {_PREVIEW_RAW_SELECT}
                          FROM messages m
                          WHERE m.session_id = s.id AND m.role = 'user' AND m.content IS NOT NULL
+                           AND {_PREVIEW_ELIGIBLE_SQL}
                          ORDER BY m.timestamp, m.id LIMIT 1),
                         ''
                     ) AS _preview_raw,
@@ -8901,6 +8903,7 @@ class SessionDB(SessionSearchMixin, SessionSchemaMixin, SessionPortabilityMixin)
                         (SELECT {_PREVIEW_RAW_SELECT}
                          FROM messages m
                          WHERE m.session_id = s.id AND m.role = 'user' AND m.content IS NOT NULL
+                           AND {_PREVIEW_ELIGIBLE_SQL}
                          ORDER BY m.timestamp, m.id LIMIT 1),
                         ''
                     ) AS _preview_raw,
@@ -8940,6 +8943,7 @@ class SessionDB(SessionSearchMixin, SessionSchemaMixin, SessionPortabilityMixin)
                         (SELECT {_PREVIEW_RAW_SELECT}
                          FROM messages m
                          WHERE m.session_id = s.id AND m.role = 'user' AND m.content IS NOT NULL
+                           AND {_PREVIEW_ELIGIBLE_SQL}
                          ORDER BY m.timestamp, m.id LIMIT 1),
                         ''
                     ) AS _preview_raw,
@@ -12709,6 +12713,7 @@ class SessionDB(SessionSearchMixin, SessionSchemaMixin, SessionPortabilityMixin)
                             (SELECT {_PREVIEW_RAW_SELECT}
                              FROM messages m
                              WHERE m.session_id = s.id AND m.role = 'user' AND m.content IS NOT NULL
+                               AND {_PREVIEW_ELIGIBLE_SQL}
                              ORDER BY m.timestamp, m.id LIMIT 1),
                             ''
                         ) AS _preview_raw,
@@ -12739,6 +12744,7 @@ class SessionDB(SessionSearchMixin, SessionSchemaMixin, SessionPortabilityMixin)
                             (SELECT {_PREVIEW_RAW_SELECT}
                              FROM messages m
                              WHERE m.session_id = s.id AND m.role = 'user' AND m.content IS NOT NULL
+                               AND {_PREVIEW_ELIGIBLE_SQL}
                              ORDER BY m.timestamp, m.id LIMIT 1),
                             ''
                         ) AS _preview_raw,

hermes_state_common.py

@@ -13,6 +13,13 @@ from agent.skill_commands import (
     SKILL_SCAFFOLD_SQL_LIKE,
     describe_skill_invocation,
 )
+from agent.context_compressor import (
+    LEGACY_SUMMARY_PREFIX,
+    SUMMARY_PREFIX,
+    _MERGED_PRIOR_CONTEXT_HEADER,
+    _MERGED_SUMMARY_DELIMITER,
+    _SUMMARY_END_MARKER,
+)
 
 
 # Session preview = the head of the first user message, shown wherever a
@@ -52,12 +59,90 @@ _PREVIEW_CONTENT_SQL = "REPLACE(REPLACE(m.content, X'0A', ' '), X'0D', ' ')"
 _PREVIEW_SCAFFOLDED_SQL = f"m.content LIKE '{SKILL_SCAFFOLD_SQL_LIKE}'"
 
 
+def _sql_literal(text: str) -> str:
+    return "'" + text.replace("'", "''") + "'"
+
+
+_SQL_WHITESPACE = "CHAR(9) || CHAR(10) || CHAR(13) || CHAR(32)"
+
+
+def _sql_ltrim_whitespace(expression: str) -> str:
+    return f"LTRIM({expression}, {_SQL_WHITESPACE})"
+
+
+def _sql_trim_whitespace(expression: str) -> str:
+    return f"TRIM({expression}, {_SQL_WHITESPACE})"
+
+
+def _sql_starts_with(expression: str, prefixes: tuple[str, ...]) -> str:
+    trimmed = _sql_ltrim_whitespace(expression)
+    checks = [
+        f"SUBSTR({trimmed}, 1, {len(prefix)}) = {_sql_literal(prefix)}"
+        for prefix in prefixes
+    ]
+    return "(" + " OR ".join(checks) + ")"
+
+
+# Current and historical long-form prefixes share this complete introduction;
+# their stale-item guidance diverges only after it. Matching the whole intro
+# avoids treating an ordinary user message that merely starts with the short
+# bracketed label as a compaction carrier.
+_PREVIEW_LONG_FORM_PREFIX = SUMMARY_PREFIX.split("Do NOT answer", 1)[0]
+_PREVIEW_SUMMARY_PREFIXES = (
+    _PREVIEW_LONG_FORM_PREFIX,
+    LEGACY_SUMMARY_PREFIX,
+)
+_PREVIEW_STANDALONE_SUMMARY_SQL = _sql_starts_with(
+    "m.content", _PREVIEW_SUMMARY_PREFIXES
+)
+_PREVIEW_MERGED_AFTER_SQL = (
+    f"SUBSTR(m.content, INSTR(m.content, {_sql_literal(_MERGED_SUMMARY_DELIMITER)})"
+    f" + {len(_MERGED_SUMMARY_DELIMITER)})"
+)
+_PREVIEW_MERGED_SUMMARY_SQL = (
+    f"(INSTR(m.content, {_sql_literal(_MERGED_SUMMARY_DELIMITER)}) > 0"
+    f" AND {_sql_starts_with(_PREVIEW_MERGED_AFTER_SQL, _PREVIEW_SUMMARY_PREFIXES)})"
+)
+_PREVIEW_MERGED_PRIOR_SQL = _sql_trim_whitespace(
+    f"SUBSTR(m.content, 1, INSTR(m.content, {_sql_literal(_MERGED_SUMMARY_DELIMITER)}) - 1)"
+)
+_PREVIEW_MERGED_PRIOR_LTRIMMED_SQL = _sql_ltrim_whitespace(
+    _PREVIEW_MERGED_PRIOR_SQL
+)
+_PREVIEW_MERGED_PRIOR_UNWRAPPED_SQL = (
+    f"CASE WHEN SUBSTR({_PREVIEW_MERGED_PRIOR_LTRIMMED_SQL}, 1,"
+    f" {len(_MERGED_PRIOR_CONTEXT_HEADER)}) = {_sql_literal(_MERGED_PRIOR_CONTEXT_HEADER)}"
+    f" THEN {_sql_ltrim_whitespace(f'SUBSTR({_PREVIEW_MERGED_PRIOR_LTRIMMED_SQL}, {len(_MERGED_PRIOR_CONTEXT_HEADER) + 1})')}"
+    f" ELSE {_PREVIEW_MERGED_PRIOR_SQL} END"
+)
+_PREVIEW_FORCE_USER_REMAINDER_SQL = (
+    f"SUBSTR(m.content, INSTR(m.content, {_sql_literal(_SUMMARY_END_MARKER)})"
+    f" + {len(_SUMMARY_END_MARKER)})"
+)
+
+# Session preview subqueries select their first eligible user-authored content.
+# Pure compaction rows are ineligible; force-user-leading and merged carriers
+# remain eligible only when authentic content survives the wire boundary.
+_PREVIEW_ELIGIBLE_SQL = (
+    f"((NOT {_PREVIEW_STANDALONE_SUMMARY_SQL} AND NOT {_PREVIEW_MERGED_SUMMARY_SQL})"
+    f" OR ({_PREVIEW_STANDALONE_SUMMARY_SQL}"
+    f" AND INSTR(m.content, {_sql_literal(_SUMMARY_END_MARKER)}) > 0"
+    f" AND LENGTH({_sql_trim_whitespace(_PREVIEW_FORCE_USER_REMAINDER_SQL)}) > 0)"
+    f" OR ({_PREVIEW_MERGED_SUMMARY_SQL}"
+    f" AND LENGTH({_sql_trim_whitespace(_PREVIEW_MERGED_PRIOR_UNWRAPPED_SQL)}) > 0))"
+)
+
+
 # The shared ``_preview_raw`` SELECT expression, interpolated by every listing
 # query. A scaffolded row gets a wider excerpt: the whole message while it fits
 # the budget, else head + tail (where the typed instruction lands) spliced
 # around SKILL_EXCERPT_JOINT.
 _PREVIEW_RAW_SELECT = (
-    f"CASE WHEN {_PREVIEW_SCAFFOLDED_SQL}"
+    f"CASE WHEN {_PREVIEW_STANDALONE_SUMMARY_SQL}"
+    f" THEN {_PREVIEW_FORCE_USER_REMAINDER_SQL}"
+    f" WHEN {_PREVIEW_MERGED_SUMMARY_SQL}"
+    f" THEN {_PREVIEW_MERGED_PRIOR_UNWRAPPED_SQL}"
+    f" WHEN {_PREVIEW_SCAFFOLDED_SQL}"
     f" AND LENGTH(m.content) > {_PREVIEW_SCAFFOLD_WINDOW * 2}"
     f" THEN SUBSTR({_PREVIEW_CONTENT_SQL}, 1, {_PREVIEW_SCAFFOLD_WINDOW})"
     f" || '{SKILL_EXCERPT_JOINT}'"
@@ -73,6 +158,7 @@ def _shape_preview(raw: Any) -> str:
     text = str(raw or "").strip()
     if not text:
         return ""
+    text = text.replace("\n", " ").replace("\r", " ")
     described = describe_skill_invocation(text)
     text = described if described is not None else text.split(SKILL_EXCERPT_JOINT)[0]
     if len(text) > _PREVIEW_MAX_CHARS:

hermes_state_portability.py

@@ -16,6 +16,7 @@ from typing import Any, Dict, List, Optional
 from agent.skill_commands import SKILL_SCAFFOLD_SQL_LIKE
 from hermes_state_common import (
     SCHEMA_SQL,
+    _PREVIEW_ELIGIBLE_SQL,
     _PREVIEW_RAW_SELECT,
     _shape_preview,
     _sql_session_last_active,
@@ -107,6 +108,7 @@ class SessionPortabilityMixin:
                     (SELECT {_PREVIEW_RAW_SELECT}
                      FROM messages m
                      WHERE m.session_id = s.id AND m.role = 'user' AND m.content IS NOT NULL
+                       AND {_PREVIEW_ELIGIBLE_SQL}
                      ORDER BY m.timestamp, m.id LIMIT 1),
                     ''
                 ) AS _preview_raw,
@@ -191,6 +193,7 @@ class SessionPortabilityMixin:
                     (SELECT {_PREVIEW_RAW_SELECT}
                      FROM messages m
                      WHERE m.session_id = s.id AND m.role = 'user' AND m.content IS NOT NULL
+                       AND {_PREVIEW_ELIGIBLE_SQL}
                      ORDER BY m.timestamp, m.id LIMIT 1),
                     ''
                 ) AS _preview_raw,

tests/gateway/test_api_server_compaction_projection.py

@@ -5,6 +5,7 @@ from __future__ import annotations
 from aiohttp.test_utils import TestClient, TestServer
 import pytest
 
+from agent.compaction_display import project_compaction_message_for_display
 from agent.context_compressor import (
     COMPRESSED_SUMMARY_METADATA_KEY,
     HISTORICAL_TASK_HEADING,
@@ -79,6 +80,9 @@ class TestMessageProjection:
                 tool_calls=[{"id": "stale"}],
                 reasoning="internal compression reasoning",
                 reasoning_content="internal compression reasoning",
+                reasoning_details=[{"type": "reasoning.summary", "summary": "internal"}],
+                codex_reasoning_items=[{"type": "reasoning", "id": "internal"}],
+                codex_message_items=[{"type": "message", "id": "internal"}],
             )
         )
 
@@ -88,6 +92,9 @@ class TestMessageProjection:
         assert "finish_reason" not in projected
         assert "reasoning" not in projected
         assert "reasoning_content" not in projected
+        assert "reasoning_details" not in projected
+        assert "codex_reasoning_items" not in projected
+        assert "codex_message_items" not in projected
 
     def test_merged_carrier_preserves_only_real_prior_content(self):
         projected = APIServerAdapter._message_response(
@@ -125,11 +132,16 @@ class TestMessageProjection:
         assert projected["content"] == [{"type": "text", "text": REAL_USER}]
 
     def test_real_message_that_mentions_marker_text_is_untouched(self):
-        content = "please explain the string [CONTEXT COMPACTION] in this bug report"
+        message = _row(
-        projected = APIServerAdapter._message_response(_row("user", content))
+            "user",
+            "please explain the string [CONTEXT COMPACTION] in this bug report",
+            tool_calls=[{"id": "real"}],
+            reasoning="real provider payload",
+        )
+        projected = project_compaction_message_for_display(message)
 
-        assert projected["content"] == content
+        assert projected == message
-        assert "display_kind" not in projected
+        assert projected is not message
 
     def test_unrelated_hidden_message_is_not_reclassified_as_compaction(self):
         message = _row("assistant", "ordinary hidden control row", display_kind="hidden")

tests/gateway/test_telegram_topic_mode.py

@@ -10,6 +10,13 @@ from unittest.mock import AsyncMock, MagicMock
 
 import pytest
 
+from agent.context_compressor import (
+    HISTORICAL_TASK_HEADING,
+    SUMMARY_PREFIX,
+    _MERGED_PRIOR_CONTEXT_HEADER,
+    _MERGED_SUMMARY_DELIMITER,
+    _SUMMARY_END_MARKER,
+)
 from hermes_state import SessionDB
 from gateway.config import GatewayConfig, HomeChannel, Platform, PlatformConfig
 from gateway.platforms.base import MessageEvent
@@ -161,6 +168,67 @@ def _make_runner(session_db=None):
     return runner
 
 
+@pytest.mark.asyncio
+async def test_topic_restore_quote_never_exposes_compaction_scaffolding(tmp_path):
+    db = SessionDB(db_path=tmp_path / "state.db")
+    db.enable_telegram_topic_mode(chat_id="208214988", user_id="208214988")
+    db.create_session(
+        session_id="restorable",
+        source="telegram",
+        user_id="208214988",
+    )
+    db.set_session_title("restorable", "Browser control")
+    db.append_message("restorable", "assistant", "real completed answer")
+    summary = (
+        f"{SUMMARY_PREFIX}\n\n"
+        f"{HISTORICAL_TASK_HEADING}\nold work\n\n"
+        f"{_SUMMARY_END_MARKER}"
+    )
+    db.append_message("restorable", "assistant", summary)
+    runner = _make_runner(session_db=db)
+
+    result = await runner._restore_telegram_topic_session(
+        _make_event("/topic restorable", thread_id="17585"),
+        "restorable",
+    )
+
+    assert "Last Hermes message:\nreal completed answer" in result
+    assert "CONTEXT COMPACTION" not in result
+    assert "Historical Task Snapshot" not in result
+    db.close()
+
+
+@pytest.mark.asyncio
+async def test_topic_restore_quote_unwraps_merged_assistant_carrier(tmp_path):
+    db = SessionDB(db_path=tmp_path / "state.db")
+    db.enable_telegram_topic_mode(chat_id="208214988", user_id="208214988")
+    db.create_session(
+        session_id="restorable",
+        source="telegram",
+        user_id="208214988",
+    )
+    carrier = (
+        f"{_MERGED_PRIOR_CONTEXT_HEADER}\n"
+        "real completed answer\n\n"
+        f"{_MERGED_SUMMARY_DELIMITER}\n\n"
+        f"{SUMMARY_PREFIX}\n\n"
+        f"{HISTORICAL_TASK_HEADING}\nold work\n\n"
+        f"{_SUMMARY_END_MARKER}"
+    )
+    db.append_message("restorable", "assistant", carrier)
+    runner = _make_runner(session_db=db)
+
+    result = await runner._restore_telegram_topic_session(
+        _make_event("/topic restorable", thread_id="17585"),
+        "restorable",
+    )
+
+    assert "Last Hermes message:\nreal completed answer" in result
+    assert "PRIOR CONTEXT" not in result
+    assert "CONTEXT COMPACTION" not in result
+    db.close()
+
+
 @pytest.mark.asyncio
 @pytest.mark.parametrize("thread_id", [None, "1"])
 async def test_internal_root_telegram_dm_event_bypasses_topic_lobby(

tests/test_session_skill_previews.py

@@ -12,6 +12,14 @@ shaper directly, so the CASE expression and the Python side are covered together
 
 import pytest
 
+from agent.context_compressor import (
+    HISTORICAL_TASK_HEADING,
+    SUMMARY_PREFIX,
+    _HISTORICAL_SUMMARY_PREFIXES,
+    _MERGED_PRIOR_CONTEXT_HEADER,
+    _MERGED_SUMMARY_DELIMITER,
+    _SUMMARY_END_MARKER,
+)
 import agent.skill_commands as skill_commands
 import tools.skills_tool as skills_tool
 from hermes_state import SessionDB
@@ -80,8 +88,6 @@ class TestSkillPreview:
         (row,) = db.list_sessions_rich(limit=10)
         assert row["preview"] == "/work"
 
-
-
     def test_rewind_picker_shows_the_typed_instruction(
         self, db, tmp_path, monkeypatch
     ):
@@ -94,6 +100,92 @@ class TestSkillPreview:
         assert entry["preview"] == "/work — fix the title leak"
 
 
+class TestCompactionPreview:
+    def test_literal_marker_text_is_still_a_real_user_preview(self, db):
+        message = "[CONTEXT COMPACTION — REFERENCE ONLY] what does this label mean?"
+        db.create_session(session_id="s1", source="cli", model="m")
+        db.append_message("s1", role="user", content=message)
+
+        (row,) = db.list_sessions_rich(limit=10)
+
+        assert row["preview"].startswith("[CONTEXT COMPACTION — REFERENCE ONLY]")
+        assert row["preview"] != ""
+
+    def test_pure_compaction_row_cannot_become_session_preview(self, db):
+        summary = (
+            f"{SUMMARY_PREFIX}\n\n"
+            f"{HISTORICAL_TASK_HEADING}\nold work\n\n"
+            f"{_SUMMARY_END_MARKER}"
+        )
+        db.create_session(session_id="s1", source="cli", model="m")
+        db.append_message("s1", role="user", content=summary)
+        db.append_message("s1", role="user", content="test the browser controller")
+
+        (row,) = db.list_sessions_rich(limit=10)
+
+        assert row["preview"] == "test the browser controller"
+
+    def test_historical_compaction_row_cannot_become_session_preview(self, db):
+        summary = (
+            f"{_HISTORICAL_SUMMARY_PREFIXES[-1]}\n\n"
+            f"{HISTORICAL_TASK_HEADING}\nold work\n\n"
+            f"{_SUMMARY_END_MARKER}\n\n"
+        )
+        db.create_session(session_id="s1", source="cli", model="m")
+        db.append_message("s1", role="user", content=summary)
+        db.append_message("s1", role="user", content="test the browser controller")
+
+        (row,) = db.list_sessions_rich(limit=10)
+
+        assert row["preview"] == "test the browser controller"
+
+    def test_force_user_leading_compaction_preview_preserves_live_ask(self, db):
+        carrier = (
+            f"{SUMMARY_PREFIX}\n\n"
+            f"{HISTORICAL_TASK_HEADING}\nold work\n\n"
+            f"{_SUMMARY_END_MARKER}\n\n"
+            "test the browser controller"
+        )
+        db.create_session(session_id="s1", source="cli", model="m")
+        db.append_message("s1", role="user", content=carrier)
+
+        (row,) = db.list_sessions_rich(limit=10)
+
+        assert row["preview"] == "test the browser controller"
+
+    def test_merged_compaction_preview_preserves_prior_user_content(self, db):
+        carrier = (
+            f"{_MERGED_PRIOR_CONTEXT_HEADER}\n"
+            "test the browser controller\n\n"
+            f"{_MERGED_SUMMARY_DELIMITER}\n\n"
+            f"{SUMMARY_PREFIX}\n\n"
+            f"{HISTORICAL_TASK_HEADING}\nold work\n\n"
+            f"{_SUMMARY_END_MARKER}"
+        )
+        db.create_session(session_id="s1", source="cli", model="m")
+        db.append_message("s1", role="user", content=carrier)
+
+        (row,) = db.list_sessions_rich(limit=10)
+
+        assert row["preview"] == "test the browser controller"
+
+    def test_empty_merged_carrier_does_not_block_later_user_preview(self, db):
+        carrier = (
+            f"{_MERGED_PRIOR_CONTEXT_HEADER}\n\n"
+            f"{_MERGED_SUMMARY_DELIMITER}\n\n"
+            f"{SUMMARY_PREFIX}\n\n"
+            f"{HISTORICAL_TASK_HEADING}\nold work\n\n"
+            f"{_SUMMARY_END_MARKER}"
+        )
+        db.create_session(session_id="s1", source="cli", model="m")
+        db.append_message("s1", role="user", content=carrier)
+        db.append_message("s1", role="user", content="test the browser controller")
+
+        (row,) = db.list_sessions_rich(limit=10)
+
+        assert row["preview"] == "test the browser controller"
+
+
 class TestSkillScaffoldedSessionLookup:
     """Backing queries for `hermes sessions retitle-skills`."""
 

tests/test_tui_gateway_server.py

@@ -2582,6 +2582,83 @@ def test_history_to_messages_preserves_tool_calls_for_resume_display():
     ]
 
 
+def test_history_to_messages_drops_pure_compaction_scaffolding():
+    from agent.context_compressor import (
+        HISTORICAL_TASK_HEADING,
+        SUMMARY_PREFIX,
+        _SUMMARY_END_MARKER,
+    )
+
+    summary = (
+        f"{SUMMARY_PREFIX}\n\n"
+        f"{HISTORICAL_TASK_HEADING}\nold work\n\n"
+        f"{_SUMMARY_END_MARKER}"
+    )
+
+    assert server._history_to_messages(
+        [
+            {"role": "user", "content": summary},
+            {"role": "assistant", "content": "real answer"},
+        ]
+    ) == [{"role": "assistant", "text": "real answer"}]
+
+
+def test_history_to_messages_preserves_live_ask_without_compaction_scaffolding():
+    from agent.context_compressor import (
+        HISTORICAL_TASK_HEADING,
+        SUMMARY_PREFIX,
+        _SUMMARY_END_MARKER,
+    )
+
+    carrier = (
+        f"{SUMMARY_PREFIX}\n\n"
+        f"{HISTORICAL_TASK_HEADING}\nold work\n\n"
+        f"{_SUMMARY_END_MARKER}\n\n"
+        "test the browser controller"
+    )
+
+    assert server._history_to_messages(
+        [
+            {
+                "role": "user",
+                "content": carrier,
+                "tool_calls": [{"id": "stale"}],
+                "reasoning": "internal compaction reasoning",
+            }
+        ]
+    ) == [{"role": "user", "text": "test the browser controller"}]
+
+
+def test_history_to_messages_unwraps_merged_assistant_carrier():
+    from agent.context_compressor import (
+        HISTORICAL_TASK_HEADING,
+        SUMMARY_PREFIX,
+        _MERGED_PRIOR_CONTEXT_HEADER,
+        _MERGED_SUMMARY_DELIMITER,
+        _SUMMARY_END_MARKER,
+    )
+
+    carrier = (
+        f"{_MERGED_PRIOR_CONTEXT_HEADER}\n"
+        "real completed answer\n\n"
+        f"{_MERGED_SUMMARY_DELIMITER}\n\n"
+        f"{SUMMARY_PREFIX}\n\n"
+        f"{HISTORICAL_TASK_HEADING}\nold work\n\n"
+        f"{_SUMMARY_END_MARKER}"
+    )
+
+    assert server._history_to_messages(
+        [
+            {
+                "role": "assistant",
+                "content": carrier,
+                "tool_calls": [{"id": "stale"}],
+                "reasoning_details": [{"summary": "internal"}],
+            }
+        ]
+    ) == [{"role": "assistant", "text": "real completed answer"}]
+
+
 def test_history_to_messages_ships_full_tool_args():
     # This is the display projection. `context` is an 80-char preview for
     # collapsed row titles. A renderer that shows the full call (the expanded

tui_gateway/server.py

@@ -35,6 +35,7 @@ from hermes_cli.env_loader import load_hermes_dotenv
 from utils import is_truthy_value
 from tools.environments.local import hermes_subprocess_env
 from agent.replay_cleanup import sanitize_replay_history
+from agent.compaction_display import project_compaction_message_for_display
 from agent.skill_commands import describe_skill_invocation
 from agent.conversation_loop import INTERRUPT_WAITING_FOR_MODEL_PREFIX
 from tui_gateway import git_probe
@@ -7673,6 +7674,9 @@ def _history_to_messages(history: list[dict]) -> list[dict]:
     for m in history:
         if not isinstance(m, dict):
             continue
+        m = project_compaction_message_for_display(m)
+        if m is None:
+            continue
         role = m.get("role")
         if role not in {"user", "assistant", "tool", "system"}:
             continue