fix(agent): guard merged assistant compaction handoffs

Treat a merged assistant-role summary carrier as the driving reference handoff when it immediately follows a completed assistant stop. Its preserved prose and stale tool_calls are assistant continuity, not a fresh live user request.

Keep legitimate in-flight behavior unchanged when there is no completed stop, a real user turn follows, or a distinct later assistant tool-call row continues the loop.

Extends the #80622 active-turn guard for the merged-carrier shape reported under #42768.

agent/context_compressor.py

@@ -7978,8 +7978,25 @@ def reference_handoff_would_drive_next_model_call(
     for index, message in enumerate(messages):
         if not is_compaction_summary_message(message):
             continue
-        if _handoff_carries_live_user_content(message):
-            # Embedded live ask — this row is not a sole-handoff driver.
+        previous = messages[index - 1] if index > 0 else None
+        merged_completed_assistant = (
+            isinstance(message, dict)
+            and message.get("role") == "assistant"
+            and ContextCompressor.classify_summary_content(
+                message.get("content")
+            )
+            == "merged"
+            and isinstance(previous, dict)
+            and previous.get("role") == "assistant"
+            and previous.get("finish_reason") == "stop"
+        )
+        if (
+            _handoff_carries_live_user_content(message)
+            and not merged_completed_assistant
+        ):
+            # Embedded live ask — this row is not a sole-handoff driver. A
+            # merged assistant carrier after a completed stop preserves the
+            # assistant's own prose/tool_calls, not a fresh user request.
             continue
         last_driving_handoff = index
 

tests/agent/test_reference_handoff_active_turn.py

@@ -14,8 +14,11 @@ from agent.context_compressor import (
     COMPRESSED_SUMMARY_HAS_USER_TURN_KEY,
     COMPRESSED_SUMMARY_METADATA_KEY,
     COMPRESSION_CONTINUATION_USER_CONTENT,
+    ContextCompressor,
     HISTORICAL_TASK_HEADING,
     SUMMARY_PREFIX,
+    _MERGED_PRIOR_CONTEXT_HEADER,
+    _MERGED_SUMMARY_DELIMITER,
     _SUMMARY_END_MARKER,
     is_compaction_summary_message,
     is_user_originated_turn,
@@ -39,6 +42,24 @@ def _standalone_handoff(task: str = "finish the already-done refactor") -> dict:
     }
 
 
+def _merged_assistant_carrier() -> dict:
+    """Production merge-into-tail shape: assistant role/tool_calls survive."""
+    return {
+        "role": "assistant",
+        "content": (
+            f"{_MERGED_PRIOR_CONTEXT_HEADER}\n"
+            "Refactor complete.\n\n"
+            f"{_MERGED_SUMMARY_DELIMITER}\n"
+            f"{SUMMARY_PREFIX}\n{HISTORICAL_TASK_HEADING}\n"
+            "User asked: 'finish the already-done refactor'\n\n"
+            f"{_SUMMARY_END_MARKER}"
+        ),
+        "tool_calls": [{"id": "stale-c1", "function": {"name": "terminal"}}],
+        COMPRESSED_SUMMARY_METADATA_KEY: True,
+        COMPRESSED_SUMMARY_HAS_USER_TURN_KEY: True,
+    }
+
+
 class TestReferenceHandoffWouldDriveNextModelCall:
     def test_standalone_handoff_alone_drives(self):
         messages = [_standalone_handoff()]
@@ -76,6 +97,70 @@ class TestReferenceHandoffWouldDriveNextModelCall:
         ]
         assert reference_handoff_would_drive_next_model_call(messages) is False
 
+    def test_merged_assistant_carrier_after_completed_stop_drives(self):
+        """Completed assistant prose/tool_calls on the carrier are not a user ask."""
+        messages = [
+            {"role": "user", "content": "please finish the refactor"},
+            {
+                "role": "assistant",
+                "content": "Refactor complete.",
+                "finish_reason": "stop",
+            },
+            _merged_assistant_carrier(),
+        ]
+        assert reference_handoff_would_drive_next_model_call(messages) is True
+
+    def test_merged_assistant_carrier_without_completed_stop_stays_in_flight(self):
+        messages = [
+            {"role": "user", "content": "please finish the refactor"},
+            _merged_assistant_carrier(),
+        ]
+        assert reference_handoff_would_drive_next_model_call(messages) is False
+
+    def test_standalone_assistant_handoff_after_stop_uses_existing_guard(self):
+        handoff = _standalone_handoff()
+        handoff["role"] = "assistant"
+        messages = [
+            {
+                "role": "assistant",
+                "content": "Refactor complete.",
+                "finish_reason": "stop",
+            },
+            handoff,
+        ]
+        assert ContextCompressor.classify_summary_content(handoff["content"]) == (
+            "standalone"
+        )
+        assert reference_handoff_would_drive_next_model_call(messages) is True
+
+    def test_real_user_after_merged_assistant_carrier_does_not_drive(self):
+        messages = [
+            {
+                "role": "assistant",
+                "content": "Refactor complete.",
+                "finish_reason": "stop",
+            },
+            _merged_assistant_carrier(),
+            {"role": "user", "content": "start a different task"},
+        ]
+        assert reference_handoff_would_drive_next_model_call(messages) is False
+
+    def test_distinct_tool_call_after_merged_carrier_stays_in_flight(self):
+        messages = [
+            {
+                "role": "assistant",
+                "content": "Refactor complete.",
+                "finish_reason": "stop",
+            },
+            _merged_assistant_carrier(),
+            {
+                "role": "assistant",
+                "content": None,
+                "tool_calls": [{"id": "live-c2", "function": {"name": "terminal"}}],
+            },
+        ]
+        assert reference_handoff_would_drive_next_model_call(messages) is False
+
     def test_embedded_remainder_after_end_marker_does_not_drive(self):
         messages = [
             {