refactor(moa): co-locate guidance peel with attach, add round-trip contract

_peel_moa_guidance hand-implemented the inverse of moa_loop's
_attach_reference_guidance from a different module — a drifting separator
or shape would make the peel silently no-op and put the last cache
breakpoint on the turn-varying guidance block (the #72626 bug class).
Move the inverse into moa_loop.peel_reference_guidance directly adjacent
to the attach, keep a thin wrapper in conversation_loop, and pin the
contract with a round-trip test over all three attach shapes.

Also fix the empty-list residue: peeling a guidance-only content part now
drops the whole message (mirroring the appended-user-message shape)
instead of leaving an empty-content user turn behind.

agent/conversation_loop.py

@@ -847,39 +847,13 @@ def _peel_moa_guidance(
 ) -> List[Dict[str, Any]]:
     """Remove MoA reference guidance previously attached by ``_attach_reference_guidance``.
 
-    Redecoration must run on the base transcript so the last cache breakpoint
-    does not land on the turn-varying guidance block; callers then rebase via
-    ``rebase_prepared_request`` (#72626).
+    Thin wrapper over :func:`agent.moa_loop.peel_reference_guidance` (kept
+    adjacent to the attach so the forward/inverse shapes evolve together).
+    Lazy import mirrors the module's other moa_loop touchpoints.
     """
-    if not guidance or not messages:
-        return messages
-    guidance_text = str(guidance)
-    last = messages[-1]
-    if not isinstance(last, dict) or last.get("role") != "user":
-        return messages
-    content = last.get("content")
-    if content == guidance_text:
-        return list(messages[:-1])
-    suffix = "\n\n" + guidance_text
-    if isinstance(content, str) and content.endswith(suffix):
-        peeled = dict(last)
-        peeled["content"] = content[: -len(suffix)]
-        return [*messages[:-1], peeled]
-    if isinstance(content, list) and content:
-        last_part = content[-1]
-        if isinstance(last_part, dict) and last_part.get("type", "text") == "text":
-            text = last_part.get("text") or ""
-            if text == suffix or text == guidance_text:
-                peeled = dict(last)
-                peeled["content"] = list(content[:-1])
-                return [*messages[:-1], peeled]
-            if text.endswith(suffix):
-                new_part = dict(last_part)
-                new_part["text"] = text[: -len(suffix)]
-                peeled = dict(last)
-                peeled["content"] = [*content[:-1], new_part]
-                return [*messages[:-1], peeled]
-    return messages
+    from agent.moa_loop import peel_reference_guidance
+
+    return peel_reference_guidance(messages, guidance)
 
 
 def _redecorate_prompt_cache_for_provider(

agent/moa_loop.py

@@ -1337,6 +1337,63 @@ def _attach_reference_guidance(agg_messages: list[dict[str, Any]], guidance: str
     agg_messages.append({"role": "user", "content": guidance})
 
 
+def peel_reference_guidance(
+    messages: list[dict[str, Any]],
+    guidance: Any,
+) -> list[dict[str, Any]]:
+    """Remove reference guidance previously attached by ``_attach_reference_guidance``.
+
+    Exact inverse of the three attach shapes above (string merge, trailing
+    text part, appended user message) — kept adjacent so the two evolve
+    together; a drifting separator or shape would make the peel silently
+    no-op and let a cache breakpoint land on the turn-varying guidance
+    block (the bug class #72626 fixes).
+
+    Used by the failover redecoration chokepoint: redecoration must run on
+    the base transcript so the last cache breakpoint does not land on the
+    guidance; callers then rebase via ``rebase_prepared_request``.
+
+    Returns a new list (input list and its messages are not mutated).
+    """
+    if not guidance or not messages:
+        return messages
+    guidance_text = str(guidance)
+    last = messages[-1]
+    if not isinstance(last, dict) or last.get("role") != "user":
+        return messages
+    content = last.get("content")
+    if content == guidance_text:
+        # Attach shape (c): guidance was appended as its own user message.
+        return list(messages[:-1])
+    suffix = "\n\n" + guidance_text
+    if isinstance(content, str) and content.endswith(suffix):
+        # Attach shape (a): merged into a trailing string user turn.
+        peeled = dict(last)
+        peeled["content"] = content[: -len(suffix)]
+        return [*messages[:-1], peeled]
+    if isinstance(content, list) and content:
+        last_part = content[-1]
+        if isinstance(last_part, dict) and last_part.get("type", "text") == "text":
+            text = last_part.get("text") or ""
+            if text == suffix or text == guidance_text:
+                # Attach shape (b): guidance rode as its own trailing part.
+                peeled = dict(last)
+                peeled["content"] = list(content[:-1])
+                if not peeled["content"]:
+                    # The guidance part was the only content — mirror the
+                    # string shape (c) and drop the whole message rather
+                    # than leaving an empty-content user turn behind.
+                    return list(messages[:-1])
+                return [*messages[:-1], peeled]
+            if text.endswith(suffix):
+                new_part = dict(last_part)
+                new_part["text"] = text[: -len(suffix)]
+                peeled = dict(last)
+                peeled["content"] = [*content[:-1], new_part]
+                return [*messages[:-1], peeled]
+    return messages
+
+
 class MoAChatCompletions:
     """OpenAI-chat-compatible facade where the aggregator is the acting model."""
 

tests/agent/test_failover_identity.py

@@ -410,3 +410,62 @@ class TestRedecoratePromptCacheOnPolicyChange:
         # redecorated (stripped) messages, not the stale decorated list.
         assert _count_cache_markers(new_prepared["messages"]) == 0
         assert new_prepared["messages"] == out
+
+
+class TestPeelReferenceGuidanceRoundTrip:
+    """peel must invert every attach shape — the two live adjacent in
+    moa_loop.py precisely so this contract can't drift silently."""
+
+    _GUIDANCE = "[Mixture of Agents reference context]\nAdvice body."
+
+    def _round_trip(self, base):
+        import copy
+
+        from agent.moa_loop import _attach_reference_guidance, peel_reference_guidance
+
+        attached = copy.deepcopy(base)
+        _attach_reference_guidance(attached, self._GUIDANCE)
+        assert attached != base, "attach must change the transcript"
+        return peel_reference_guidance(attached, self._GUIDANCE)
+
+    def test_string_merge_shape(self):
+        base = [
+            {"role": "system", "content": "sys"},
+            {"role": "user", "content": "task"},
+        ]
+        assert self._round_trip(base) == base
+
+    def test_list_part_shape(self):
+        base = [
+            {"role": "system", "content": "sys"},
+            {
+                "role": "user",
+                "content": [{"type": "text", "text": "task", "cache_control": {"type": "ephemeral"}}],
+            },
+        ]
+        assert self._round_trip(base) == base
+
+    def test_appended_user_message_shape(self):
+        # No trailing user turn — attach appends a separate user message.
+        base = [
+            {"role": "system", "content": "sys"},
+            {"role": "user", "content": "task"},
+            {"role": "assistant", "content": "done"},
+        ]
+        assert self._round_trip(base) == base
+
+    def test_guidance_only_part_drops_message_not_empty_residue(self):
+        # If the guidance part is the only content left after peeling, the
+        # whole message goes — an empty-content user turn must never remain.
+        from agent.moa_loop import peel_reference_guidance
+
+        messages = [
+            {"role": "user", "content": "task"},
+            {"role": "assistant", "content": "ok"},
+            {
+                "role": "user",
+                "content": [{"type": "text", "text": self._GUIDANCE}],
+            },
+        ]
+        peeled = peel_reference_guidance(messages, self._GUIDANCE)
+        assert peeled == messages[:-1]