fix(acp): flag replayed compaction summaries via _meta

A context-compaction handoff is persisted as an ordinary history message
but is not a real turn. The ACP history replay streamed it as a bare
user/agent message chunk, dropping the in-process _compressed_summary
marker, so ACP frontends (editors, vscode-hermes) rendered the entire
handoff as a regular message.

Tag replayed summary chunks under _meta.hermes (ACP's extensibility
channel), covering all three persistence shapes the compressor emits:

- standalone role="user" handoff -> compactionSummary: true
- standalone role="assistant" handoff (alternation-driven role pick)
  -> compactionSummary: true
- merge-into-tail message (preserved tail content + appended summary)
  -> containsCompactionSummary: true, a distinct key so clients that
  collapse standalone summaries cannot hide the preserved real content

Detection honors the in-process metadata flag and falls back to a new
ContextCompressor.classify_summary_content() content classifier
(standalone/merged/None), so it also works for a DB-reloaded session
that lost the in-memory flag. _is_context_summary_content is now a thin
wrapper over the classifier, keeping existing callers unchanged.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

acp_adapter/server.py

@@ -74,6 +74,10 @@ from acp_adapter.permissions import make_approval_callback
 from acp_adapter.provenance import session_provenance_meta
 from acp_adapter.session import SessionManager, SessionState, _expand_acp_enabled_toolsets
 from acp_adapter.tools import build_tool_complete, build_tool_start
+from agent.context_compressor import (
+    COMPRESSED_SUMMARY_METADATA_KEY,
+    ContextCompressor,
+)
 from tools.approval import (
     reset_hermes_interactive_context,
     set_hermes_interactive_context,
@@ -969,11 +973,49 @@ class HermesACPAgent(acp.Agent):
                 return text
         return ""
 
+    @staticmethod
+    def _history_summary_meta(message: dict[str, Any], text: str) -> dict[str, Any] | None:
+        """Build the ``_meta`` payload for a replayed compaction summary.
+
+        Compaction summaries are persisted as ordinary history messages —
+        standalone handoffs under ``role="user"`` OR ``role="assistant"``
+        (the compressor picks whichever role keeps alternation valid), and
+        merge-into-tail messages where the summary is appended after the
+        first preserved tail message's real content. Without a wire flag,
+        ACP frontends render all of these as ordinary turns.
+
+        Two distinct keys under ``_meta.hermes`` (ACP's extensibility
+        channel), so clients cannot accidentally hide real content:
+
+        * ``compactionSummary: true`` — the entire chunk is the handoff
+          summary. Safe to restyle or collapse wholesale.
+        * ``containsCompactionSummary: true`` — a merged-tail message: real
+          preserved turn content followed by the summary. Clients may style
+          it, but collapsing the whole chunk would hide the preserved
+          content, hence the separate key.
+
+        Detection honors the in-process ``_compressed_summary`` flag and
+        falls back to content classification, so it also works for a
+        DB-reloaded session that lost the in-memory flag.
+        """
+        kind = ContextCompressor.classify_summary_content(text)
+        if kind is None and message.get(COMPRESSED_SUMMARY_METADATA_KEY):
+            # Flagged in-process but content didn't classify (e.g. future
+            # prefix drift): treat as a standalone summary — the flag is only
+            # ever set on summary-bearing messages.
+            kind = "standalone"
+        if kind == "standalone":
+            return {"hermes": {"compactionSummary": True}}
+        if kind == "merged":
+            return {"hermes": {"containsCompactionSummary": True}}
+        return None
+
     @staticmethod
     def _history_message_update(
         *,
         role: str,
         text: str,
+        field_meta: dict[str, Any] | None = None,
     ) -> UserMessageChunk | AgentMessageChunk | None:
         """Build an ACP history replay update for a user/assistant message."""
         block = TextContentBlock(type="text", text=text)
@@ -981,11 +1023,13 @@ class HermesACPAgent(acp.Agent):
             return UserMessageChunk(
                 session_update="user_message_chunk",
                 content=block,
+                field_meta=field_meta,
             )
         if role == "assistant":
             return AgentMessageChunk(
                 session_update="agent_message_chunk",
                 content=block,
+                field_meta=field_meta,
             )
         return None
 
@@ -1056,7 +1100,11 @@ class HermesACPAgent(acp.Agent):
             if role == "user":
                 text = self._history_message_text(message)
                 if text:
-                    update = self._history_message_update(role=role, text=text)
+                    update = self._history_message_update(
+                        role=role,
+                        text=text,
+                        field_meta=self._history_summary_meta(message, text),
+                    )
                     if update is not None and not await _send(update):
                         return
                 continue
@@ -1068,7 +1116,11 @@ class HermesACPAgent(acp.Agent):
 
                 text = self._history_message_text(message)
                 if text:
-                    update = self._history_message_update(role=role, text=text)
+                    update = self._history_message_update(
+                        role=role,
+                        text=text,
+                        field_meta=self._history_summary_meta(message, text),
+                    )
                     if update is not None and not await _send(update):
                         return
 

agent/context_compressor.py

@@ -3011,7 +3011,29 @@ This compaction should PRIORITISE preserving all information related to the focu
         return f"{SUMMARY_PREFIX}\n{text}" if text else SUMMARY_PREFIX
 
     @staticmethod
-    def _is_context_summary_content(content: Any) -> bool:
+    def _starts_with_summary_prefix(text: str) -> bool:
+        """Return True if *text* begins with any known handoff prefix."""
+        if text.startswith(SUMMARY_PREFIX) or text.startswith(LEGACY_SUMMARY_PREFIX):
+            return True
+        return any(text.startswith(p) for p in _HISTORICAL_SUMMARY_PREFIXES)
+
+    @classmethod
+    def classify_summary_content(cls, content: Any) -> Optional[str]:
+        """Classify how *content* relates to a compaction summary.
+
+        Returns:
+            ``"standalone"``: the entire message IS a compaction handoff
+            (current, legacy, or historical prefix at the start). Frontends
+            may restyle/collapse the whole message as a summary.
+
+            ``"merged"``: a merge-into-tail message — real preserved turn
+            content wrapped under ``_MERGED_PRIOR_CONTEXT_HEADER``, followed by
+            ``_MERGED_SUMMARY_DELIMITER`` and the summary body. The message
+            *contains* a summary but is not only a summary; collapsing the
+            whole message would hide the preserved content.
+
+            ``None``: no compaction summary detected.
+        """
         text = _content_text_for_contains(content).lstrip()
         # Merge-into-tail summaries wrap prior tail content before the summary,
         # so the handoff prefix lands after _MERGED_SUMMARY_DELIMITER rather than
@@ -3019,10 +3041,13 @@ This compaction should PRIORITISE preserving all information related to the focu
         # (auto-focus skip, carry-forward summary find, last-real-user anchor)
         # mistake a merged summary message for a real user turn.
         if _MERGED_SUMMARY_DELIMITER in text:
-            text = text.split(_MERGED_SUMMARY_DELIMITER, 1)[1].lstrip()
-        if text.startswith(SUMMARY_PREFIX) or text.startswith(LEGACY_SUMMARY_PREFIX):
-            return True
-        return any(text.startswith(p) for p in _HISTORICAL_SUMMARY_PREFIXES)
+            after = text.split(_MERGED_SUMMARY_DELIMITER, 1)[1].lstrip()
+            return "merged" if cls._starts_with_summary_prefix(after) else None
+        return "standalone" if cls._starts_with_summary_prefix(text) else None
+
+    @classmethod
+    def _is_context_summary_content(cls, content: Any) -> bool:
+        return cls.classify_summary_content(content) is not None
 
     @staticmethod
     def _has_compressed_summary_metadata(message: Any) -> bool:

tests/acp/test_server.py

@@ -422,6 +422,126 @@ class TestSessionOps:
         assert "Search results" in tool_updates[1].content[0].content.text
         assert "cli.py:42" in tool_updates[1].content[0].content.text
 
+    @pytest.mark.asyncio
+    async def test_load_session_flags_compaction_summary_on_replayed_user_chunk(self, agent):
+        """A replayed compaction summary must carry _meta.hermes.compactionSummary.
+
+        The handoff is stored role="user" but is not a real user turn; without
+        the flag on the wire, ACP frontends render the whole summary as a user
+        message. Detection falls back to content, so this holds even for a
+        DB-reloaded session that lost the in-process metadata flag.
+        """
+        from agent.context_compressor import SUMMARY_PREFIX
+
+        mock_conn = MagicMock(spec=acp.Client)
+        mock_conn.session_update = AsyncMock()
+        agent._conn = mock_conn
+
+        summary_text = SUMMARY_PREFIX + "\n\n## Active Task\nDo the thing."
+        new_resp = await agent.new_session(cwd="/tmp")
+        state = agent.session_manager.get_session(new_resp.session_id)
+        state.history = [
+            {"role": "user", "content": summary_text},
+            {"role": "user", "content": "wait 5s and reply ok"},
+        ]
+
+        mock_conn.session_update.reset_mock()
+        await agent.load_session(cwd="/tmp", session_id=new_resp.session_id)
+        await asyncio.sleep(0)
+        await asyncio.sleep(0)
+
+        user_chunks = [
+            call.kwargs["update"]
+            for call in mock_conn.session_update.await_args_list
+            if isinstance(call.kwargs.get("update"), UserMessageChunk)
+        ]
+        assert len(user_chunks) == 2
+        # First user chunk is the summary → flagged; second is a real turn → not.
+        assert user_chunks[0].field_meta == {"hermes": {"compactionSummary": True}}
+        assert user_chunks[1].field_meta is None
+
+    @pytest.mark.asyncio
+    async def test_load_session_flags_compaction_summary_on_replayed_assistant_chunk(self, agent):
+        """The compressor can emit a standalone summary with role="assistant"
+        (whichever role keeps alternation valid), so the assistant replay
+        branch must flag it too — not just the user branch.
+        """
+        from agent.context_compressor import SUMMARY_PREFIX
+
+        mock_conn = MagicMock(spec=acp.Client)
+        mock_conn.session_update = AsyncMock()
+        agent._conn = mock_conn
+
+        summary_text = SUMMARY_PREFIX + "\n\n## Active Task\nDo the thing."
+        new_resp = await agent.new_session(cwd="/tmp")
+        state = agent.session_manager.get_session(new_resp.session_id)
+        state.history = [
+            {"role": "assistant", "content": summary_text},
+            {"role": "user", "content": "continue"},
+            {"role": "assistant", "content": "on it"},
+        ]
+
+        mock_conn.session_update.reset_mock()
+        await agent.load_session(cwd="/tmp", session_id=new_resp.session_id)
+        await asyncio.sleep(0)
+        await asyncio.sleep(0)
+
+        agent_chunks = [
+            call.kwargs["update"]
+            for call in mock_conn.session_update.await_args_list
+            if isinstance(call.kwargs.get("update"), AgentMessageChunk)
+        ]
+        assert len(agent_chunks) == 2
+        assert agent_chunks[0].field_meta == {"hermes": {"compactionSummary": True}}
+        assert agent_chunks[1].field_meta is None
+
+    @pytest.mark.asyncio
+    async def test_load_session_flags_merged_tail_summary_as_contains_not_standalone(self, agent):
+        """A merge-into-tail message carries real preserved content plus the
+        summary. It must be flagged containsCompactionSummary — NOT
+        compactionSummary — so a client that collapses standalone summaries
+        cannot hide the preserved turn content.
+        """
+        from agent.context_compressor import (
+            _MERGED_PRIOR_CONTEXT_HEADER,
+            _MERGED_SUMMARY_DELIMITER,
+            _SUMMARY_END_MARKER,
+            SUMMARY_PREFIX,
+        )
+
+        mock_conn = MagicMock(spec=acp.Client)
+        mock_conn.session_update = AsyncMock()
+        agent._conn = mock_conn
+
+        merged_text = (
+            _MERGED_PRIOR_CONTEXT_HEADER
+            + "\nplease fix the login bug"
+            + "\n\n" + _MERGED_SUMMARY_DELIMITER + "\n\n"
+            + SUMMARY_PREFIX + "\n\n## Active Task\nFix login."
+            + "\n\n" + _SUMMARY_END_MARKER
+        )
+        new_resp = await agent.new_session(cwd="/tmp")
+        state = agent.session_manager.get_session(new_resp.session_id)
+        state.history = [
+            {"role": "user", "content": merged_text},
+            {"role": "assistant", "content": "looking at it"},
+        ]
+
+        mock_conn.session_update.reset_mock()
+        await agent.load_session(cwd="/tmp", session_id=new_resp.session_id)
+        await asyncio.sleep(0)
+        await asyncio.sleep(0)
+
+        user_chunks = [
+            call.kwargs["update"]
+            for call in mock_conn.session_update.await_args_list
+            if isinstance(call.kwargs.get("update"), UserMessageChunk)
+        ]
+        assert len(user_chunks) == 1
+        assert user_chunks[0].field_meta == {
+            "hermes": {"containsCompactionSummary": True}
+        }
+
     @pytest.mark.asyncio
     async def test_load_session_replays_native_plan_for_persisted_todo_tool(self, agent):
         """Persisted todo tool results should rebuild Zed's native plan panel."""

tests/agent/test_compressed_summary_metadata.py

@@ -97,3 +97,63 @@ class TestMetadataFlagNeverReachesWire:
             isinstance(m, dict) and m.get(COMPRESSED_SUMMARY_METADATA_KEY)
             for m in out
         )
+
+
+class TestClassifySummaryContent:
+    """classify_summary_content distinguishes standalone handoffs from
+    merge-into-tail messages so wire consumers (ACP replay) can flag them
+    differently — collapsing a merged message would hide the preserved
+    tail content that precedes the summary."""
+
+    def test_standalone_summary(self):
+        from agent.context_compressor import SUMMARY_PREFIX
+
+        content = SUMMARY_PREFIX + "\n## Active Task\nstuff"
+        assert ContextCompressor.classify_summary_content(content) == "standalone"
+        assert ContextCompressor._is_context_summary_content(content) is True
+
+    def test_legacy_and_historical_prefixes_are_standalone(self):
+        from agent.context_compressor import (
+            LEGACY_SUMMARY_PREFIX,
+            _HISTORICAL_SUMMARY_PREFIXES,
+        )
+
+        assert ContextCompressor.classify_summary_content(
+            LEGACY_SUMMARY_PREFIX + " body"
+        ) == "standalone"
+        for prefix in _HISTORICAL_SUMMARY_PREFIXES:
+            assert ContextCompressor.classify_summary_content(
+                prefix + " body"
+            ) == "standalone"
+
+    def test_merged_tail_summary(self):
+        from agent.context_compressor import (
+            SUMMARY_PREFIX,
+            _MERGED_PRIOR_CONTEXT_HEADER,
+            _MERGED_SUMMARY_DELIMITER,
+            _SUMMARY_END_MARKER,
+        )
+
+        merged = (
+            _MERGED_PRIOR_CONTEXT_HEADER + "\n"
+            "old tail content\n\n"
+            + _MERGED_SUMMARY_DELIMITER + "\n\n"
+            + SUMMARY_PREFIX + "\nBODY\n\n"
+            + _SUMMARY_END_MARKER
+        )
+        assert ContextCompressor.classify_summary_content(merged) == "merged"
+        assert ContextCompressor._is_context_summary_content(merged) is True
+
+    def test_plain_messages_classify_none(self):
+        assert ContextCompressor.classify_summary_content("just a question") is None
+        assert ContextCompressor.classify_summary_content("") is None
+        assert ContextCompressor.classify_summary_content(None) is None
+
+    def test_delimiter_without_summary_prefix_is_none(self):
+        """A message merely quoting the merged delimiter (e.g. a user pasting
+        logs) is not a summary unless a handoff prefix follows it."""
+        from agent.context_compressor import _MERGED_SUMMARY_DELIMITER
+
+        content = "look at this:\n" + _MERGED_SUMMARY_DELIMITER + "\nnot a summary"
+        assert ContextCompressor.classify_summary_content(content) is None
+        assert ContextCompressor._is_context_summary_content(content) is False