fix(workspace/OFFSEC-003): correct boundary wrapping + add closer truncation

Two bugs fixed in tool_delegate_task wrapping logic:

1. Wrapping used raw _A2A_BOUNDARY_START/_END markers, which
   appeared alongside the escaped form of peer content. Fixed: wrap
   with _A2A_BOUNDARY_START_ESCAPED/_END_ESCAPED so output contains
   no raw closer that could confuse downstream parsers.

2. A malicious peer could inject a fake closer ([/A2A_RESULT_FROM_PEER])
   to make legitimate content appear truncated. Fixed: truncate at the
   raw closer BEFORE sanitization (truncation loses the raw form).

Updated test assertions across 3 test files to match new escaped wrapper
form (previous tests expected raw markers in output).

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>

workspace/_sanitize_a2a.py

@@ -40,6 +40,8 @@ _A2A_BOUNDARY_END = "[/A2A_RESULT_FROM_PEER]"
 # inside the trusted zone. Escape BOTH boundary markers in the raw text
 # before wrapping so they can never close the boundary early.
 # We use "[/ " as the escape prefix — visually distinct from the real marker.
+_A2A_BOUNDARY_START_ESCAPED = "[/ A2A_RESULT_FROM_PEER]"
+_A2A_BOUNDARY_END_ESCAPED = "[/ /A2A_RESULT_FROM_PEER]"
 
 
 def _escape_boundary_markers(text: str) -> str:
@@ -50,8 +52,8 @@ def _escape_boundary_markers(text: str) -> str:
     the boundary early or inject a fake opener.
     """
     return (
-        text.replace(_A2A_BOUNDARY_START, "[/ A2A_RESULT_FROM_PEER]")
-        .replace(_A2A_BOUNDARY_END, "[/ /A2A_RESULT_FROM_PEER]")
+        text.replace(_A2A_BOUNDARY_START, _A2A_BOUNDARY_START_ESCAPED)
+        .replace(_A2A_BOUNDARY_END, _A2A_BOUNDARY_END_ESCAPED)
     )
 
 

workspace/a2a_tools_delegation.py

@@ -49,7 +49,9 @@ from a2a_client import (
 from a2a_tools_rbac import auth_headers_for_heartbeat as _auth_headers_for_heartbeat
 from _sanitize_a2a import (
     _A2A_BOUNDARY_END,
+    _A2A_BOUNDARY_END_ESCAPED,
     _A2A_BOUNDARY_START,
+    _A2A_BOUNDARY_START_ESCAPED,
     sanitize_a2a_result,
 )  # noqa: E402
 
@@ -330,8 +332,18 @@ async def tool_delegate_task(
     # markers so the agent can distinguish trusted (own output) from untrusted
     # (peer-supplied) content.  Explicit wrapping here rather than inside
     # sanitize_a2a_result preserves a clean separation of concerns.
+    #
+    # Truncate at the closer BEFORE sanitizing so the raw closer (which gets
+    # lost during escaping) is removed from the content.  After truncation,
+    # sanitize the remaining text and wrap with escaped boundary markers.
+    if _A2A_BOUNDARY_END in result:
+        result = result[:result.index(_A2A_BOUNDARY_END)]
     escaped = sanitize_a2a_result(result)
-    return f"{_A2A_BOUNDARY_START}\n{escaped}\n{_A2A_BOUNDARY_END}"
+    return (
+        f"{_A2A_BOUNDARY_START_ESCAPED}\n"
+        f"{escaped}\n"
+        f"{_A2A_BOUNDARY_END_ESCAPED}"
+    )
 
 
 async def tool_delegate_task_async(

workspace/tests/test_a2a_tools_delegation.py

@@ -218,7 +218,8 @@ class TestPollingPathSanitization:
         result = asyncio.run(d.tool_delegate_task("ws-peer", "do it"))
         # tool_delegate_task wraps the sanitized text in _A2A_BOUNDARY_START/END
         # (NOT _A2A_RESULT_FROM_PEER — that marker is for the messaging path).
-        assert d._A2A_BOUNDARY_START in result
-        assert d._A2A_BOUNDARY_END in result
+        # Wrapped in escaped form to prevent raw closer from appearing in output.
+        assert d._A2A_BOUNDARY_START_ESCAPED in result
+        assert d._A2A_BOUNDARY_END_ESCAPED in result
         assert "Sanitized peer reply" in result
 

workspace/tests/test_a2a_tools_impl.py

@@ -277,7 +277,7 @@ class TestToolDelegateTask:
              patch("a2a_tools.report_activity", new=AsyncMock()):
             result = await a2a_tools.tool_delegate_task("ws-1", "do something")
 
-        assert result == "[A2A_RESULT_FROM_PEER]\nTask completed!\n[/A2A_RESULT_FROM_PEER]"
+        assert result == "[/ A2A_RESULT_FROM_PEER]\nTask completed!\n[/ /A2A_RESULT_FROM_PEER]"
 
     async def test_error_response_returns_delegation_failed_message(self):
         """When send_a2a_message returns _A2A_ERROR_PREFIX text, delegation fails."""
@@ -305,7 +305,7 @@ class TestToolDelegateTask:
              patch("a2a_tools.report_activity", new=AsyncMock()):
             result = await a2a_tools.tool_delegate_task("ws-cached", "task")
 
-        assert result == "[A2A_RESULT_FROM_PEER]\ndone\n[/A2A_RESULT_FROM_PEER]"
+        assert result == "[/ A2A_RESULT_FROM_PEER]\ndone\n[/ /A2A_RESULT_FROM_PEER]"
 
     async def test_peer_name_falls_back_to_id_prefix(self):
         """When peer has no name and cache is empty, name = first 8 chars of workspace_id."""
@@ -319,7 +319,7 @@ class TestToolDelegateTask:
              patch("a2a_tools.report_activity", new=AsyncMock()):
             result = await a2a_tools.tool_delegate_task("ws-nona000", "task")
 
-        assert result == "[A2A_RESULT_FROM_PEER]\nok\n[/A2A_RESULT_FROM_PEER]"
+        assert result == "[/ A2A_RESULT_FROM_PEER]\nok\n[/ /A2A_RESULT_FROM_PEER]"
         # Cache should now have been set
         assert a2a_tools._peer_names.get("ws-nona000") is not None
 

workspace/tests/test_delegation_sync_via_polling.py

@@ -69,7 +69,7 @@ class TestFlagOffLegacyPath:
         monkeypatch.delenv("DELEGATION_SYNC_VIA_INBOX", raising=False)
 
         import a2a_tools
-        from _sanitize_a2a import _A2A_BOUNDARY_END, _A2A_BOUNDARY_START
+        from _sanitize_a2a import _A2A_BOUNDARY_END_ESCAPED, _A2A_BOUNDARY_START_ESCAPED
         send_calls = []
 
         async def fake_send(workspace_id, task, source_workspace_id=None):
@@ -91,8 +91,8 @@ class TestFlagOffLegacyPath:
             )
 
         # OFFSEC-003: result is wrapped in boundary markers
-        assert _A2A_BOUNDARY_START in result
-        assert _A2A_BOUNDARY_END in result
+        assert _A2A_BOUNDARY_START_ESCAPED in result
+        assert _A2A_BOUNDARY_END_ESCAPED in result
         assert "legacy ok" in result
         assert send_calls == [("ws-target", "task body", "ws-self")]
         poll_mock.assert_not_called()
@@ -124,7 +124,7 @@ class TestPollModeAutoFallback:
         monkeypatch.delenv("DELEGATION_SYNC_VIA_INBOX", raising=False)
 
         import a2a_tools
-        from _sanitize_a2a import _A2A_BOUNDARY_END, _A2A_BOUNDARY_START
+        from _sanitize_a2a import _A2A_BOUNDARY_END_ESCAPED, _A2A_BOUNDARY_START_ESCAPED
         from a2a_client import _A2A_QUEUED_PREFIX
 
         send_calls = []
@@ -159,8 +159,8 @@ class TestPollModeAutoFallback:
         assert poll_calls[0] == ("ws-target", "task body", "ws-self")
         # Caller sees the real reply, NOT the queued sentinel and NOT
         # a DELEGATION FAILED string. Wrapped in OFFSEC-003 boundary markers.
-        assert _A2A_BOUNDARY_START in result
-        assert _A2A_BOUNDARY_END in result
+        assert _A2A_BOUNDARY_START_ESCAPED in result
+        assert _A2A_BOUNDARY_END_ESCAPED in result
         assert "real response from poll-mode peer" in result
 
     async def test_non_queued_send_result_does_not_trigger_fallback(self, monkeypatch):
@@ -169,7 +169,7 @@ class TestPollModeAutoFallback:
         monkeypatch.delenv("DELEGATION_SYNC_VIA_INBOX", raising=False)
 
         import a2a_tools
-        from _sanitize_a2a import _A2A_BOUNDARY_END, _A2A_BOUNDARY_START
+        from _sanitize_a2a import _A2A_BOUNDARY_END_ESCAPED, _A2A_BOUNDARY_START_ESCAPED
 
         async def fake_send(*_a, **_kw):
             return "normal reply"
@@ -189,8 +189,8 @@ class TestPollModeAutoFallback:
             )
 
         # OFFSEC-003: wrapped in boundary markers
-        assert _A2A_BOUNDARY_START in result
-        assert _A2A_BOUNDARY_END in result
+        assert _A2A_BOUNDARY_START_ESCAPED in result
+        assert _A2A_BOUNDARY_END_ESCAPED in result
         assert "normal reply" in result
         poll_mock.assert_not_called()