diff --git a/acp_adapter/server.py b/acp_adapter/server.py index f664c47070e..76f71ea7cb9 100644 --- a/acp_adapter/server.py +++ b/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 diff --git a/agent/context_compressor.py b/agent/context_compressor.py index d249f9898b6..9eaee872e35 100644 --- a/agent/context_compressor.py +++ b/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: diff --git a/tests/acp/test_server.py b/tests/acp/test_server.py index e60f3e8feb7..3333dc967e3 100644 --- a/tests/acp/test_server.py +++ b/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.""" diff --git a/tests/agent/test_compressed_summary_metadata.py b/tests/agent/test_compressed_summary_metadata.py index dbbf0f0903d..29c4765d892 100644 --- a/tests/agent/test_compressed_summary_metadata.py +++ b/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