Merge pull request #2892 from Molecule-AI/refactor/a2a-tools-messaging-extract-rfc2873-iter4d

refactor(workspace): extract messaging tools from a2a_tools.py (RFC #2873 iter 4d)

scripts/build_runtime_package.py

@@ -57,6 +57,7 @@ TOP_LEVEL_MODULES = {
     "a2a_tools",
     "a2a_tools_delegation",
     "a2a_tools_memory",
+    "a2a_tools_messaging",
     "a2a_tools_rbac",
     "adapter_base",
     "agent",

workspace/a2a_tools.py

@@ -129,200 +129,19 @@ from a2a_tools_delegation import (  # noqa: E402  (import after the from-a2a_cli
 )
 
 
-async def _upload_chat_files(
+# Messaging tool handlers — extracted to a2a_tools_messaging
-    client: httpx.AsyncClient,
+# (RFC #2873 iter 4d). Re-imported here so call sites + tests that
-    paths: list[str],
+# reference ``a2a_tools.tool_send_message_to_user`` /
-    workspace_id: str | None = None,
+# ``tool_list_peers`` / ``tool_get_workspace_info`` /
-) -> tuple[list[dict], str | None]:
+# ``tool_chat_history`` / ``_upload_chat_files`` keep resolving
-    """Upload local file paths through /workspaces/<self>/chat/uploads.
+# identically.
-
+from a2a_tools_messaging import (  # noqa: E402  (import after the top-of-module imports)
-    The platform stages each upload under /workspace/.molecule/chat-uploads
+    _upload_chat_files,
-    (an "allowed root" the canvas knows how to render via the Download
+    tool_chat_history,
-    endpoint) and returns metadata the broadcast payload references.
+    tool_get_workspace_info,
-
+    tool_list_peers,
-    Why we route through upload instead of just passing the agent's path:
+    tool_send_message_to_user,
-    the canvas's allowed-root list is /configs, /workspace, /home, /plugins
+)
-    — files at /tmp or /root would be unreachable. Uploading copies the
-    bytes into an allowed root regardless of where the agent wrote them.
-
-    Returns (attachments, error). On any failure the caller should NOT
-    fire the notify — partial-attach would surface a half-rendered chip.
-    """
-    if not paths:
-        return [], None
-    files_payload: list[tuple[str, tuple[str, bytes, str]]] = []
-    for p in paths:
-        if not isinstance(p, str) or not p:
-            return [], f"Error: invalid attachment path {p!r}"
-        if not os.path.isfile(p):
-            return [], f"Error: attachment not found: {p}"
-        try:
-            with open(p, "rb") as fh:
-                data = fh.read()
-        except OSError as e:
-            return [], f"Error reading {p}: {e}"
-        # Sniff mime from filename so the canvas can pick the right
-        # icon / preview / inline-image renderer. Pre-fix this was
-        # hardcoded application/octet-stream and chat_files.go's
-        # Upload trusts whatever Content-Type the multipart part
-        # carries — `mt := fh.Header.Get("Content-Type")` only falls
-        # back to extension-sniffing when the header is empty. So a
-        # hardcoded octet-stream meant every attachment lost its
-        # real type forever, breaking the canvas chip's icon logic.
-        mime_type, _ = mimetypes.guess_type(p)
-        if not mime_type:
-            mime_type = "application/octet-stream"
-        files_payload.append(("files", (os.path.basename(p), data, mime_type)))
-    target_workspace_id = (workspace_id or "").strip() or WORKSPACE_ID
-    try:
-        resp = await client.post(
-            f"{PLATFORM_URL}/workspaces/{target_workspace_id}/chat/uploads",
-            files=files_payload,
-            headers=_auth_headers_for_heartbeat(target_workspace_id),
-        )
-    except Exception as e:
-        return [], f"Error uploading attachments: {e}"
-    if resp.status_code != 200:
-        return [], f"Error: chat/uploads returned {resp.status_code}: {resp.text[:200]}"
-    try:
-        body = resp.json()
-    except Exception as e:
-        return [], f"Error parsing upload response: {e}"
-    uploaded = body.get("files") or []
-    if not isinstance(uploaded, list) or len(uploaded) != len(paths):
-        return [], f"Error: upload returned {len(uploaded) if isinstance(uploaded, list) else 'invalid'} entries for {len(paths)} files"
-    return uploaded, None
-
-
-async def tool_send_message_to_user(
-    message: str,
-    attachments: list[str] | None = None,
-    workspace_id: str | None = None,
-) -> str:
-    """Send a message directly to the user's canvas chat via WebSocket.
-
-    Args:
-        message: The text to display in the user's chat. Required even
-            when sending attachments — set to a short caption like
-            "Here's the build output:" or "Done — see attached."
-        attachments: Optional list of absolute file paths inside this
-            container. Each is uploaded to the platform and rendered
-            in the canvas as a clickable download chip. Use this
-            instead of pasting paths in the message text — paths
-            render as plain text and the user can't click them.
-            Examples:
-              attachments=["/tmp/build-output.zip"]
-              attachments=["/workspace/report.pdf", "/workspace/data.csv"]
-        workspace_id: Optional. When the agent is registered in MULTIPLE
-            workspaces (external multi-workspace MCP path), this
-            selects which workspace's chat to deliver the message to —
-            should match the ``arrival_workspace_id`` of the inbound
-            message you're replying to so the user sees the reply in
-            the same canvas they typed in. Single-workspace agents
-            omit this; the message routes to the only registered
-            workspace.
-    """
-    if not message:
-        return "Error: message is required"
-    target_workspace_id = (workspace_id or "").strip() or WORKSPACE_ID
-    try:
-        async with httpx.AsyncClient(timeout=60.0) as client:
-            uploaded, upload_err = await _upload_chat_files(
-                client, attachments or [], workspace_id=target_workspace_id,
-            )
-            if upload_err:
-                return upload_err
-            payload: dict = {"message": message}
-            if uploaded:
-                payload["attachments"] = uploaded
-            resp = await client.post(
-                f"{PLATFORM_URL}/workspaces/{target_workspace_id}/notify",
-                json=payload,
-                headers=_auth_headers_for_heartbeat(target_workspace_id),
-            )
-            if resp.status_code == 200:
-                if uploaded:
-                    return f"Message sent to user with {len(uploaded)} attachment(s)"
-                return "Message sent to user"
-            return f"Error: platform returned {resp.status_code}"
-    except Exception as e:
-        return f"Error sending message: {e}"
-
-
-async def tool_list_peers(source_workspace_id: str | None = None) -> str:
-    """List all workspaces this agent can communicate with.
-
-    Behavior:
-        - ``source_workspace_id`` set → list peers of that one workspace.
-        - Unset, single-workspace mode → list peers of WORKSPACE_ID
-          (the legacy path, unchanged).
-        - Unset, multi-workspace mode (MOLECULE_WORKSPACES populated) →
-          aggregate across every registered workspace, prefixing each
-          peer with its source so the agent / user can see the full peer
-          surface in one call.
-
-    Side-effect: populates ``_peer_to_source`` so subsequent
-    ``tool_delegate_task(target)`` auto-routes through the correct
-    sending workspace without the agent needing ``source_workspace_id``.
-    """
-    sources: list[str]
-    aggregate = False
-    if source_workspace_id:
-        sources = [source_workspace_id]
-    else:
-        registered = list_registered_workspaces()
-        if len(registered) > 1:
-            sources = registered
-            aggregate = True
-        else:
-            sources = [WORKSPACE_ID]
-
-    all_peers: list[tuple[str, dict]] = []  # (source, peer_record)
-    diagnostics: list[tuple[str, str]] = []  # (source, diagnostic)
-    for src in sources:
-        peers, diagnostic = await get_peers_with_diagnostic(source_workspace_id=src)
-        if peers:
-            for p in peers:
-                all_peers.append((src, p))
-        elif diagnostic is not None:
-            diagnostics.append((src, diagnostic))
-
-    if not all_peers:
-        if diagnostics:
-            joined = "; ".join(f"[{src[:8]}] {d}" for src, d in diagnostics)
-            return f"No peers found. {joined}"
-        return (
-            "You have no peers in the platform registry. "
-            "(No parent, no children, no siblings registered.)"
-        )
-
-    lines = []
-    for src, p in all_peers:
-        status = p.get("status", "unknown")
-        role = p.get("role", "")
-        peer_id = p["id"]
-        # Cache name for use in delegate_task
-        _peer_names[peer_id] = p["name"]
-        # Cache the source workspace so tool_delegate_task auto-routes
-        _peer_to_source[peer_id] = src
-        if aggregate:
-            lines.append(
-                f"- {p['name']} (ID: {peer_id}, status: {status}, role: {role}, via: {src[:8]})"
-            )
-        else:
-            lines.append(f"- {p['name']} (ID: {peer_id}, status: {status}, role: {role})")
-    return "\n".join(lines)
-
-
-async def tool_get_workspace_info(source_workspace_id: str | None = None) -> str:
-    """Get this workspace's own info.
-
-    ``source_workspace_id`` selects which registered workspace to
-    introspect when the agent is registered into multiple workspaces.
-    Unset → falls back to module-level WORKSPACE_ID.
-    """
-    info = await get_workspace_info(source_workspace_id=source_workspace_id)
-    return json.dumps(info, indent=2)
 
 
 # Memory tool handlers — extracted to a2a_tools_memory (RFC #2873 iter 4c).
@@ -357,98 +176,6 @@ _INBOX_NOT_ENABLED_MSG = (
 )
 
 
-async def tool_chat_history(
-    peer_id: str,
-    limit: int = 20,
-    before_ts: str = "",
-    source_workspace_id: str | None = None,
-) -> str:
-    """Fetch the prior conversation with one peer.
-
-    Hits ``/workspaces/<self>/activity?peer_id=<peer>&limit=<N>``
-    against the workspace-server, which returns activity rows where
-    the peer is either the sender (``source_id=peer`` — they sent us
-    the message) or the recipient (``target_id=peer`` — we sent to
-    them) of an A2A turn — both sides of the conversation in
-    chronological order.
-
-    Args:
-        peer_id: The other workspace's UUID. Same value the agent
-            sees as ``peer_id`` on a peer_agent push or ``workspace_id``
-            on a delegate_task call.
-        limit: Maximum rows to return; capped server-side at 500. The
-            default of 20 covers \"most recent context for this peer\"
-            without flooding the agent's context window.
-        before_ts: Optional RFC3339 timestamp; only rows strictly
-            older are returned. Used to page backward through long
-            histories — pass the oldest ``ts`` from the previous
-            response. Empty (default) returns the most recent ``limit``
-            rows.
-        source_workspace_id: Which registered workspace's activity log
-            to query. Auto-routes via ``_peer_to_source`` cache when
-            unset (the workspace this peer was discovered through);
-            falls back to module-level WORKSPACE_ID for single-workspace
-            operators.
-
-    Returns a JSON-encoded list of activity rows (or an error string
-    starting with ``Error:`` so the agent can branch). Each row carries
-    ``activity_type``, ``source_id``, ``target_id``, ``method``,
-    ``summary``, ``request_body``, ``response_body``, ``status``,
-    ``created_at`` — same shape ``inbox_peek`` and the canvas chat
-    loader already see.
-    """
-    if not peer_id or not isinstance(peer_id, str):
-        return "Error: peer_id is required"
-    if not isinstance(limit, int) or limit <= 0:
-        limit = 20
-    if limit > 500:
-        limit = 500
-
-    src = source_workspace_id or _peer_to_source.get(peer_id) or WORKSPACE_ID
-
-    params: dict[str, str] = {
-        "peer_id": peer_id,
-        "limit": str(limit),
-    }
-    # Forward verbatim — the server route validates as RFC3339 at the
-    # trust boundary and translates into a `created_at < $X` clause.
-    if before_ts:
-        params["before_ts"] = before_ts
-
-    try:
-        async with httpx.AsyncClient(timeout=10.0) as client:
-            resp = await client.get(
-                f"{PLATFORM_URL}/workspaces/{src}/activity",
-                params=params,
-                headers=_auth_headers_for_heartbeat(src),
-            )
-    except Exception as exc:  # noqa: BLE001
-        return f"Error: chat_history request failed: {exc}"
-
-    if resp.status_code == 400:
-        # Trust-boundary rejection (malformed peer_id, etc.) — surface
-        # the server's reason verbatim so the agent can correct itself.
-        try:
-            err = resp.json().get("error", "bad request")
-        except Exception:  # noqa: BLE001
-            err = "bad request"
-        return f"Error: {err}"
-    if resp.status_code >= 400:
-        return f"Error: chat_history returned HTTP {resp.status_code}"
-
-    try:
-        rows = resp.json()
-    except Exception:  # noqa: BLE001
-        return "Error: chat_history response was not JSON"
-    if not isinstance(rows, list):
-        return "Error: chat_history response was not a list"
-
-    # Server returns DESC (most recent first); reverse to chronological
-    # so the agent reads the conversation top-down like a chat log.
-    rows.reverse()
-    return json.dumps(rows)
-
-
 def _enrich_inbound_for_agent(d: dict) -> dict:
     """Add peer_name / peer_role / agent_card_url to a poll-path message.
 
@@ -494,7 +221,6 @@ def _enrich_inbound_for_agent(d: dict) -> dict:
     d["agent_card_url"] = _agent_card_url_for(peer_id)
     return d
 
-
 async def tool_inbox_peek(limit: int = 10) -> str:
     """Return up to ``limit`` pending inbound messages without removing them."""
     import inbox  # local import — avoids a circular dep at module load

workspace/a2a_tools_messaging.py

@@ -0,0 +1,324 @@
+"""Messaging tool handlers — single-concern slice of the a2a_tools surface.
+
+Extracted from ``a2a_tools.py`` (RFC #2873 iter 4d). Owns the four
+human-and-peer messaging MCP tools + the chat-upload helper they share:
+
+  * ``tool_send_message_to_user`` — push a canvas-chat message via the
+    platform's ``/notify`` endpoint.
+  * ``tool_list_peers`` — discover peers across one or many registered
+    workspaces, with side-effect of populating ``_peer_to_source`` for
+    delegate-task auto-routing.
+  * ``tool_get_workspace_info`` — JSON-encode the workspace's own info.
+  * ``tool_chat_history`` — fetch prior conversation rows with a peer.
+  * ``_upload_chat_files`` — internal helper for the message-attachments
+    code path; routes local file paths through the platform's
+    ``/chat/uploads`` so the canvas can render them as download chips.
+
+Imports the auth-header primitive from ``a2a_tools_rbac`` (iter 4a).
+"""
+from __future__ import annotations
+
+import json
+import mimetypes
+import os
+
+import httpx
+
+from a2a_client import (
+    PLATFORM_URL,
+    WORKSPACE_ID,
+    _peer_names,
+    _peer_to_source,
+    get_peers_with_diagnostic,
+    get_workspace_info,
+)
+from a2a_tools_rbac import auth_headers_for_heartbeat as _auth_headers_for_heartbeat
+from platform_auth import list_registered_workspaces
+
+
+async def _upload_chat_files(
+    client: httpx.AsyncClient,
+    paths: list[str],
+    workspace_id: str | None = None,
+) -> tuple[list[dict], str | None]:
+    """Upload local file paths through /workspaces/<self>/chat/uploads.
+
+    The platform stages each upload under /workspace/.molecule/chat-uploads
+    (an "allowed root" the canvas knows how to render via the Download
+    endpoint) and returns metadata the broadcast payload references.
+
+    Why we route through upload instead of just passing the agent's path:
+    the canvas's allowed-root list is /configs, /workspace, /home, /plugins
+    — files at /tmp or /root would be unreachable. Uploading copies the
+    bytes into an allowed root regardless of where the agent wrote them.
+
+    Returns (attachments, error). On any failure the caller should NOT
+    fire the notify — partial-attach would surface a half-rendered chip.
+    """
+    if not paths:
+        return [], None
+    files_payload: list[tuple[str, tuple[str, bytes, str]]] = []
+    for p in paths:
+        if not isinstance(p, str) or not p:
+            return [], f"Error: invalid attachment path {p!r}"
+        if not os.path.isfile(p):
+            return [], f"Error: attachment not found: {p}"
+        try:
+            with open(p, "rb") as fh:
+                data = fh.read()
+        except OSError as e:
+            return [], f"Error reading {p}: {e}"
+        # Sniff mime from filename so the canvas can pick the right
+        # icon / preview / inline-image renderer. Pre-fix this was
+        # hardcoded application/octet-stream and chat_files.go's
+        # Upload trusts whatever Content-Type the multipart part
+        # carries — `mt := fh.Header.Get("Content-Type")` only falls
+        # back to extension-sniffing when the header is empty. So a
+        # hardcoded octet-stream meant every attachment lost its
+        # real type forever, breaking the canvas chip's icon logic.
+        mime_type, _ = mimetypes.guess_type(p)
+        if not mime_type:
+            mime_type = "application/octet-stream"
+        files_payload.append(("files", (os.path.basename(p), data, mime_type)))
+    target_workspace_id = (workspace_id or "").strip() or WORKSPACE_ID
+    try:
+        resp = await client.post(
+            f"{PLATFORM_URL}/workspaces/{target_workspace_id}/chat/uploads",
+            files=files_payload,
+            headers=_auth_headers_for_heartbeat(target_workspace_id),
+        )
+    except Exception as e:
+        return [], f"Error uploading attachments: {e}"
+    if resp.status_code != 200:
+        return [], f"Error: chat/uploads returned {resp.status_code}: {resp.text[:200]}"
+    try:
+        body = resp.json()
+    except Exception as e:
+        return [], f"Error parsing upload response: {e}"
+    uploaded = body.get("files") or []
+    if not isinstance(uploaded, list) or len(uploaded) != len(paths):
+        return [], f"Error: upload returned {len(uploaded) if isinstance(uploaded, list) else 'invalid'} entries for {len(paths)} files"
+    return uploaded, None
+
+
+async def tool_send_message_to_user(
+    message: str,
+    attachments: list[str] | None = None,
+    workspace_id: str | None = None,
+) -> str:
+    """Send a message directly to the user's canvas chat via WebSocket.
+
+    Args:
+        message: The text to display in the user's chat. Required even
+            when sending attachments — set to a short caption like
+            "Here's the build output:" or "Done — see attached."
+        attachments: Optional list of absolute file paths inside this
+            container. Each is uploaded to the platform and rendered
+            in the canvas as a clickable download chip. Use this
+            instead of pasting paths in the message text — paths
+            render as plain text and the user can't click them.
+            Examples:
+              attachments=["/tmp/build-output.zip"]
+              attachments=["/workspace/report.pdf", "/workspace/data.csv"]
+        workspace_id: Optional. When the agent is registered in MULTIPLE
+            workspaces (external multi-workspace MCP path), this
+            selects which workspace's chat to deliver the message to —
+            should match the ``arrival_workspace_id`` of the inbound
+            message you're replying to so the user sees the reply in
+            the same canvas they typed in. Single-workspace agents
+            omit this; the message routes to the only registered
+            workspace.
+    """
+    if not message:
+        return "Error: message is required"
+    target_workspace_id = (workspace_id or "").strip() or WORKSPACE_ID
+    try:
+        async with httpx.AsyncClient(timeout=60.0) as client:
+            uploaded, upload_err = await _upload_chat_files(
+                client, attachments or [], workspace_id=target_workspace_id,
+            )
+            if upload_err:
+                return upload_err
+            payload: dict = {"message": message}
+            if uploaded:
+                payload["attachments"] = uploaded
+            resp = await client.post(
+                f"{PLATFORM_URL}/workspaces/{target_workspace_id}/notify",
+                json=payload,
+                headers=_auth_headers_for_heartbeat(target_workspace_id),
+            )
+            if resp.status_code == 200:
+                if uploaded:
+                    return f"Message sent to user with {len(uploaded)} attachment(s)"
+                return "Message sent to user"
+            return f"Error: platform returned {resp.status_code}"
+    except Exception as e:
+        return f"Error sending message: {e}"
+
+
+async def tool_list_peers(source_workspace_id: str | None = None) -> str:
+    """List all workspaces this agent can communicate with.
+
+    Behavior:
+        - ``source_workspace_id`` set → list peers of that one workspace.
+        - Unset, single-workspace mode → list peers of WORKSPACE_ID
+          (the legacy path, unchanged).
+        - Unset, multi-workspace mode (MOLECULE_WORKSPACES populated) →
+          aggregate across every registered workspace, prefixing each
+          peer with its source so the agent / user can see the full peer
+          surface in one call.
+
+    Side-effect: populates ``_peer_to_source`` so subsequent
+    ``tool_delegate_task(target)`` auto-routes through the correct
+    sending workspace without the agent needing ``source_workspace_id``.
+    """
+    sources: list[str]
+    aggregate = False
+    if source_workspace_id:
+        sources = [source_workspace_id]
+    else:
+        registered = list_registered_workspaces()
+        if len(registered) > 1:
+            sources = registered
+            aggregate = True
+        else:
+            sources = [WORKSPACE_ID]
+
+    all_peers: list[tuple[str, dict]] = []  # (source, peer_record)
+    diagnostics: list[tuple[str, str]] = []  # (source, diagnostic)
+    for src in sources:
+        peers, diagnostic = await get_peers_with_diagnostic(source_workspace_id=src)
+        if peers:
+            for p in peers:
+                all_peers.append((src, p))
+        elif diagnostic is not None:
+            diagnostics.append((src, diagnostic))
+
+    if not all_peers:
+        if diagnostics:
+            joined = "; ".join(f"[{src[:8]}] {d}" for src, d in diagnostics)
+            return f"No peers found. {joined}"
+        return (
+            "You have no peers in the platform registry. "
+            "(No parent, no children, no siblings registered.)"
+        )
+
+    lines = []
+    for src, p in all_peers:
+        status = p.get("status", "unknown")
+        role = p.get("role", "")
+        peer_id = p["id"]
+        # Cache name for use in delegate_task
+        _peer_names[peer_id] = p["name"]
+        # Cache the source workspace so tool_delegate_task auto-routes
+        _peer_to_source[peer_id] = src
+        if aggregate:
+            lines.append(
+                f"- {p['name']} (ID: {peer_id}, status: {status}, role: {role}, via: {src[:8]})"
+            )
+        else:
+            lines.append(f"- {p['name']} (ID: {peer_id}, status: {status}, role: {role})")
+    return "\n".join(lines)
+
+
+async def tool_get_workspace_info(source_workspace_id: str | None = None) -> str:
+    """Get this workspace's own info.
+
+    ``source_workspace_id`` selects which registered workspace to
+    introspect when the agent is registered into multiple workspaces.
+    Unset → falls back to module-level WORKSPACE_ID.
+    """
+    info = await get_workspace_info(source_workspace_id=source_workspace_id)
+    return json.dumps(info, indent=2)
+
+
+async def tool_chat_history(
+    peer_id: str,
+    limit: int = 20,
+    before_ts: str = "",
+    source_workspace_id: str | None = None,
+) -> str:
+    """Fetch the prior conversation with one peer.
+
+    Hits ``/workspaces/<self>/activity?peer_id=<peer>&limit=<N>``
+    against the workspace-server, which returns activity rows where
+    the peer is either the sender (``source_id=peer`` — they sent us
+    the message) or the recipient (``target_id=peer`` — we sent to
+    them) of an A2A turn — both sides of the conversation in
+    chronological order.
+
+    Args:
+        peer_id: The other workspace's UUID. Same value the agent
+            sees as ``peer_id`` on a peer_agent push or ``workspace_id``
+            on a delegate_task call.
+        limit: Maximum rows to return; capped server-side at 500. The
+            default of 20 covers "most recent context for this peer"
+            without flooding the agent's context window.
+        before_ts: Optional RFC3339 timestamp; only rows strictly
+            older are returned. Used to page backward through long
+            histories — pass the oldest ``ts`` from the previous
+            response. Empty (default) returns the most recent ``limit``
+            rows.
+        source_workspace_id: Which registered workspace's activity log
+            to query. Auto-routes via ``_peer_to_source`` cache when
+            unset (the workspace this peer was discovered through);
+            falls back to module-level WORKSPACE_ID for single-workspace
+            operators.
+
+    Returns a JSON-encoded list of activity rows (or an error string
+    starting with ``Error:`` so the agent can branch). Each row carries
+    ``activity_type``, ``source_id``, ``target_id``, ``method``,
+    ``summary``, ``request_body``, ``response_body``, ``status``,
+    ``created_at`` — same shape ``inbox_peek`` and the canvas chat
+    loader already see.
+    """
+    if not peer_id or not isinstance(peer_id, str):
+        return "Error: peer_id is required"
+    if not isinstance(limit, int) or limit <= 0:
+        limit = 20
+    if limit > 500:
+        limit = 500
+
+    src = source_workspace_id or _peer_to_source.get(peer_id) or WORKSPACE_ID
+
+    params: dict[str, str] = {
+        "peer_id": peer_id,
+        "limit": str(limit),
+    }
+    # Forward verbatim — the server route validates as RFC3339 at the
+    # trust boundary and translates into a `created_at < $X` clause.
+    if before_ts:
+        params["before_ts"] = before_ts
+
+    try:
+        async with httpx.AsyncClient(timeout=10.0) as client:
+            resp = await client.get(
+                f"{PLATFORM_URL}/workspaces/{src}/activity",
+                params=params,
+                headers=_auth_headers_for_heartbeat(src),
+            )
+    except Exception as exc:  # noqa: BLE001
+        return f"Error: chat_history request failed: {exc}"
+
+    if resp.status_code == 400:
+        # Trust-boundary rejection (malformed peer_id, etc.) — surface
+        # the server's reason verbatim so the agent can correct itself.
+        try:
+            err = resp.json().get("error", "bad request")
+        except Exception:  # noqa: BLE001
+            err = "bad request"
+        return f"Error: {err}"
+    if resp.status_code >= 400:
+        return f"Error: chat_history returned HTTP {resp.status_code}"
+
+    try:
+        rows = resp.json()
+    except Exception:  # noqa: BLE001
+        return "Error: chat_history response was not JSON"
+    if not isinstance(rows, list):
+        return "Error: chat_history response was not a list"
+
+    # Server returns DESC (most recent first); reverse to chronological
+    # so the agent reads the conversation top-down like a chat log.
+    rows.reverse()
+    return json.dumps(rows)

workspace/tests/test_a2a_multi_workspace.py

@@ -241,7 +241,7 @@ class TestToolListPeersAggregation:
                 return [{"id": "2222bbbb-2222-2222-2222-222222222222", "name": "bob", "status": "online", "role": "dev"}], None
             return [], None
 
-        with patch("a2a_tools.get_peers_with_diagnostic", side_effect=fake_get_peers):
+        with patch("a2a_tools_messaging.get_peers_with_diagnostic", side_effect=fake_get_peers):
             output = await a2a_tools.tool_list_peers()
 
         assert "alice" in output
@@ -263,7 +263,7 @@ class TestToolListPeersAggregation:
             assert source_workspace_id == a2a_client.WORKSPACE_ID
             return [{"id": "1111aaaa-1111-1111-1111-111111111111", "name": "alice", "status": "online", "role": "ops"}], None
 
-        with patch("a2a_tools.get_peers_with_diagnostic", side_effect=fake_get_peers):
+        with patch("a2a_tools_messaging.get_peers_with_diagnostic", side_effect=fake_get_peers):
             output = await a2a_tools.tool_list_peers()
 
         assert "alice" in output
@@ -286,7 +286,7 @@ class TestToolListPeersAggregation:
             seen.append(source_workspace_id)
             return [{"id": "1111aaaa-1111-1111-1111-111111111111", "name": "alice", "status": "online", "role": "ops"}], None
 
-        with patch("a2a_tools.get_peers_with_diagnostic", side_effect=fake_get_peers):
+        with patch("a2a_tools_messaging.get_peers_with_diagnostic", side_effect=fake_get_peers):
             output = await a2a_tools.tool_list_peers(source_workspace_id=ws_a)
 
         assert seen == [ws_a]
@@ -309,7 +309,7 @@ class TestToolListPeersAggregation:
                 return [], "auth failed"
             return [], "platform 5xx"
 
-        with patch("a2a_tools.get_peers_with_diagnostic", side_effect=fake_get_peers):
+        with patch("a2a_tools_messaging.get_peers_with_diagnostic", side_effect=fake_get_peers):
             out = await a2a_tools.tool_list_peers()
 
         assert "[aaaa1111] auth failed" in out

workspace/tests/test_a2a_tools_impl.py

@@ -453,14 +453,14 @@ class TestToolSendMessageToUser:
     async def test_success_200_returns_sent_message(self):
         import a2a_tools
         mc = _make_http_mock(post_resp=_resp(200, {}))
-        with patch("a2a_tools.httpx.AsyncClient", return_value=mc):
+        with patch("a2a_tools_messaging.httpx.AsyncClient", return_value=mc):
             result = await a2a_tools.tool_send_message_to_user("Hello user!")
         assert result == "Message sent to user"
 
     async def test_non_200_returns_status_code_in_error(self):
         import a2a_tools
         mc = _make_http_mock(post_resp=_resp(503, {}))
-        with patch("a2a_tools.httpx.AsyncClient", return_value=mc):
+        with patch("a2a_tools_messaging.httpx.AsyncClient", return_value=mc):
             result = await a2a_tools.tool_send_message_to_user("Hello user!")
         assert "503" in result
         assert "Error" in result
@@ -468,7 +468,7 @@ class TestToolSendMessageToUser:
     async def test_exception_returns_error_message(self):
         import a2a_tools
         mc = _make_http_mock(post_exc=RuntimeError("platform unreachable"))
-        with patch("a2a_tools.httpx.AsyncClient", return_value=mc):
+        with patch("a2a_tools_messaging.httpx.AsyncClient", return_value=mc):
             result = await a2a_tools.tool_send_message_to_user("Hi!")
         assert "Error sending message" in result
         assert "platform unreachable" in result
@@ -495,7 +495,7 @@ class TestToolSendMessageToUser:
         mc = _make_http_mock(post_resp=notify_resp)
         mc.post = AsyncMock(side_effect=[upload_resp, notify_resp])
 
-        with patch("a2a_tools.httpx.AsyncClient", return_value=mc):
+        with patch("a2a_tools_messaging.httpx.AsyncClient", return_value=mc):
             result = await a2a_tools.tool_send_message_to_user(
                 "Done — see attached.",
                 attachments=[str(f)],
@@ -523,7 +523,7 @@ class TestToolSendMessageToUser:
         # with a half-rendered attachment chip.
         import a2a_tools
         mc = _make_http_mock()
-        with patch("a2a_tools.httpx.AsyncClient", return_value=mc):
+        with patch("a2a_tools_messaging.httpx.AsyncClient", return_value=mc):
             result = await a2a_tools.tool_send_message_to_user(
                 "Hi", attachments=["/no/such/file.zip"],
             )
@@ -541,7 +541,7 @@ class TestToolSendMessageToUser:
         mc = _make_http_mock()
         mc.post = AsyncMock(return_value=upload_resp)
 
-        with patch("a2a_tools.httpx.AsyncClient", return_value=mc):
+        with patch("a2a_tools_messaging.httpx.AsyncClient", return_value=mc):
             result = await a2a_tools.tool_send_message_to_user(
                 "Hi", attachments=[str(f)],
             )
@@ -555,7 +555,7 @@ class TestToolSendMessageToUser:
         # an `attachments` field added to the notify body.
         import a2a_tools
         mc = _make_http_mock(post_resp=_resp(200, {}))
-        with patch("a2a_tools.httpx.AsyncClient", return_value=mc):
+        with patch("a2a_tools_messaging.httpx.AsyncClient", return_value=mc):
             await a2a_tools.tool_send_message_to_user("plain text")
         body = mc.post.await_args.kwargs.get("json") or {}
         assert body == {"message": "plain text"}
@@ -570,7 +570,7 @@ class TestToolListPeers:
     async def test_true_empty_returns_no_peers_message_without_diagnostic(self):
         """200 + empty list → 'no peers in the platform registry' (no failure)."""
         import a2a_tools
-        with patch("a2a_tools.get_peers_with_diagnostic", return_value=([], None)):
+        with patch("a2a_tools_messaging.get_peers_with_diagnostic", return_value=([], None)):
             result = await a2a_tools.tool_list_peers()
         # The new wording explicitly says no peers exist (no parent/sibling/child).
         # Avoids the misleading "may be isolated" hint when discovery succeeded.
@@ -582,7 +582,7 @@ class TestToolListPeers:
         """401/403 → tool_list_peers must surface the auth failure + restart hint, not 'isolated'."""
         import a2a_tools
         diag = "Authentication to platform failed (HTTP 401). Restart the workspace to re-mint."
-        with patch("a2a_tools.get_peers_with_diagnostic", return_value=([], diag)):
+        with patch("a2a_tools_messaging.get_peers_with_diagnostic", return_value=([], diag)):
             result = await a2a_tools.tool_list_peers()
         assert "401" in result
         assert "Authentication" in result
@@ -593,7 +593,7 @@ class TestToolListPeers:
         """404 → tool_list_peers tells the user re-registration is needed."""
         import a2a_tools
         diag = "Workspace ID ws-test is not registered with the platform (HTTP 404). Re-register."
-        with patch("a2a_tools.get_peers_with_diagnostic", return_value=([], diag)):
+        with patch("a2a_tools_messaging.get_peers_with_diagnostic", return_value=([], diag)):
             result = await a2a_tools.tool_list_peers()
         assert "404" in result
         assert "registered" in result.lower()
@@ -602,7 +602,7 @@ class TestToolListPeers:
         """5xx → 'Platform error' surfaced; agent / user can correctly route to oncall."""
         import a2a_tools
         diag = "Platform error: HTTP 503."
-        with patch("a2a_tools.get_peers_with_diagnostic", return_value=([], diag)):
+        with patch("a2a_tools_messaging.get_peers_with_diagnostic", return_value=([], diag)):
             result = await a2a_tools.tool_list_peers()
         assert "503" in result
         assert "Platform error" in result
@@ -611,7 +611,7 @@ class TestToolListPeers:
         """Network error → operator can tell that the workspace can't reach the platform at all."""
         import a2a_tools
         diag = "Cannot reach platform at http://platform.example: timed out"
-        with patch("a2a_tools.get_peers_with_diagnostic", return_value=([], diag)):
+        with patch("a2a_tools_messaging.get_peers_with_diagnostic", return_value=([], diag)):
             result = await a2a_tools.tool_list_peers()
         assert "Cannot reach platform" in result
         assert "timed out" in result
@@ -624,7 +624,7 @@ class TestToolListPeers:
             {"id": "ws-1", "name": "Alpha", "status": "online", "role": "worker"},
             {"id": "ws-2", "name": "Beta", "status": "idle", "role": "analyst"},
         ]
-        with patch("a2a_tools.get_peers_with_diagnostic", return_value=(peers, None)):
+        with patch("a2a_tools_messaging.get_peers_with_diagnostic", return_value=(peers, None)):
             result = await a2a_tools.tool_list_peers()
 
         assert "Alpha" in result
@@ -641,7 +641,7 @@ class TestToolListPeers:
         # Clear any prior cache entries for these IDs
         a2a_tools._peer_names.pop("ws-cache-test", None)
         peers = [{"id": "ws-cache-test", "name": "CacheMe", "status": "online", "role": "w"}]
-        with patch("a2a_tools.get_peers_with_diagnostic", return_value=(peers, None)):
+        with patch("a2a_tools_messaging.get_peers_with_diagnostic", return_value=(peers, None)):
             await a2a_tools.tool_list_peers()
 
         assert a2a_tools._peer_names.get("ws-cache-test") == "CacheMe"
@@ -651,7 +651,7 @@ class TestToolListPeers:
         import a2a_tools
 
         peers = [{"id": "ws-3", "name": "Gamma"}]  # no status, no role
-        with patch("a2a_tools.get_peers_with_diagnostic", return_value=(peers, None)):
+        with patch("a2a_tools_messaging.get_peers_with_diagnostic", return_value=(peers, None)):
             result = await a2a_tools.tool_list_peers()
 
         assert "Gamma" in result
@@ -669,7 +669,7 @@ class TestToolGetWorkspaceInfo:
         import a2a_tools
 
         info = {"id": "ws-test", "name": "My Workspace", "status": "online"}
-        with patch("a2a_tools.get_workspace_info", return_value=info):
+        with patch("a2a_tools_messaging.get_workspace_info", return_value=info):
             result = await a2a_tools.tool_get_workspace_info()
 
         parsed = json.loads(result)
@@ -678,7 +678,7 @@ class TestToolGetWorkspaceInfo:
     async def test_returns_error_dict_as_json(self):
         import a2a_tools
 
-        with patch("a2a_tools.get_workspace_info", return_value={"error": "not found"}):
+        with patch("a2a_tools_messaging.get_workspace_info", return_value={"error": "not found"}):
             result = await a2a_tools.tool_get_workspace_info()
 
         parsed = json.loads(result)
@@ -994,7 +994,7 @@ class TestChatHistory:
         import a2a_tools
 
         mc = _make_http_mock()
-        with patch("a2a_tools.httpx.AsyncClient", return_value=mc):
+        with patch("a2a_tools_messaging.httpx.AsyncClient", return_value=mc):
             result = await a2a_tools.tool_chat_history(peer_id="")
 
         mc.get.assert_not_called()
@@ -1006,7 +1006,7 @@ class TestChatHistory:
         import a2a_tools
 
         mc = _make_http_mock(get_resp=_resp(200, []))
-        with patch("a2a_tools.httpx.AsyncClient", return_value=mc):
+        with patch("a2a_tools_messaging.httpx.AsyncClient", return_value=mc):
             await a2a_tools.tool_chat_history(peer_id=_PEER)
 
         url, kwargs = mc.get.call_args.args[0], mc.get.call_args.kwargs
@@ -1023,7 +1023,7 @@ class TestChatHistory:
         import a2a_tools
 
         mc = _make_http_mock(get_resp=_resp(200, []))
-        with patch("a2a_tools.httpx.AsyncClient", return_value=mc):
+        with patch("a2a_tools_messaging.httpx.AsyncClient", return_value=mc):
             await a2a_tools.tool_chat_history(peer_id=_PEER, limit=10000)
 
         params = mc.get.call_args.kwargs["params"]
@@ -1035,7 +1035,7 @@ class TestChatHistory:
         import a2a_tools
 
         mc = _make_http_mock(get_resp=_resp(200, []))
-        with patch("a2a_tools.httpx.AsyncClient", return_value=mc):
+        with patch("a2a_tools_messaging.httpx.AsyncClient", return_value=mc):
             await a2a_tools.tool_chat_history(peer_id=_PEER, limit=0)
 
         assert mc.get.call_args.kwargs["params"]["limit"] == "20"
@@ -1044,7 +1044,7 @@ class TestChatHistory:
         import a2a_tools
 
         mc = _make_http_mock(get_resp=_resp(200, []))
-        with patch("a2a_tools.httpx.AsyncClient", return_value=mc):
+        with patch("a2a_tools_messaging.httpx.AsyncClient", return_value=mc):
             await a2a_tools.tool_chat_history(
                 peer_id=_PEER, before_ts="2026-05-01T00:00:00Z",
             )
@@ -1063,7 +1063,7 @@ class TestChatHistory:
         import a2a_tools
 
         mc = _make_http_mock(get_resp=_resp(200, []))
-        with patch("a2a_tools.httpx.AsyncClient", return_value=mc):
+        with patch("a2a_tools_messaging.httpx.AsyncClient", return_value=mc):
             result = await a2a_tools.tool_chat_history(peer_id=_PEER)
 
         # Exact-equality on the JSON literal (per assert-exact memory) —
@@ -1084,7 +1084,7 @@ class TestChatHistory:
             {"id": "act-1", "created_at": "2026-05-01T00:01:00Z"},
         ]
         mc = _make_http_mock(get_resp=_resp(200, rows))
-        with patch("a2a_tools.httpx.AsyncClient", return_value=mc):
+        with patch("a2a_tools_messaging.httpx.AsyncClient", return_value=mc):
             result = await a2a_tools.tool_chat_history(peer_id=_PEER)
 
         out = json.loads(result)
@@ -1097,7 +1097,7 @@ class TestChatHistory:
         import a2a_tools
 
         mc = _make_http_mock(get_resp=_resp(400, {"error": "peer_id must be a UUID"}))
-        with patch("a2a_tools.httpx.AsyncClient", return_value=mc):
+        with patch("a2a_tools_messaging.httpx.AsyncClient", return_value=mc):
             result = await a2a_tools.tool_chat_history(peer_id="bad")
 
         assert "peer_id must be a UUID" in result
@@ -1108,7 +1108,7 @@ class TestChatHistory:
         import a2a_tools
 
         mc = _make_http_mock(get_resp=_resp(500, {"error": "internal"}))
-        with patch("a2a_tools.httpx.AsyncClient", return_value=mc):
+        with patch("a2a_tools_messaging.httpx.AsyncClient", return_value=mc):
             result = await a2a_tools.tool_chat_history(peer_id=_PEER)
 
         assert result.startswith("Error:")
@@ -1121,7 +1121,7 @@ class TestChatHistory:
         import a2a_tools
 
         mc = _make_http_mock(get_exc=httpx.ConnectError("network down"))
-        with patch("a2a_tools.httpx.AsyncClient", return_value=mc):
+        with patch("a2a_tools_messaging.httpx.AsyncClient", return_value=mc):
             result = await a2a_tools.tool_chat_history(peer_id=_PEER)
 
         assert result.startswith("Error:")
@@ -1135,7 +1135,7 @@ class TestChatHistory:
         import a2a_tools
 
         mc = _make_http_mock(get_resp=_resp(200, {"unexpected": "shape"}))
-        with patch("a2a_tools.httpx.AsyncClient", return_value=mc):
+        with patch("a2a_tools_messaging.httpx.AsyncClient", return_value=mc):
             result = await a2a_tools.tool_chat_history(peer_id=_PEER)
 
         assert result.startswith("Error:")

workspace/tests/test_a2a_tools_inbox_wrappers.py

@@ -0,0 +1,196 @@
+"""Direct unit tests for the three inbox tool wrappers in ``a2a_tools``.
+
+After RFC #2873 iter 4d (messaging extraction), ``a2a_tools.py`` is
+mostly back-compat re-exports — the only behavior still defined here
+is ``report_activity`` plus three thin wrappers around the inbox state
+machine: ``tool_inbox_peek`` / ``tool_inbox_pop`` / ``tool_wait_for_message``.
+
+These wrappers were never exercised at the module level, so the
+critical-path coverage gate (75% per-file floor for MCP/inbox/auth)
+dropped to 54% on iter 4d. This file pins each wrapper's behavior
+directly so the floor is met without changing the gate.
+
+The wrappers are ~40 LOC of glue. The full delivery behavior
+(persistence, 410 recovery, etc.) is exercised in test_inbox.py.
+"""
+from __future__ import annotations
+
+import asyncio
+import json
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+
+@pytest.fixture(autouse=True)
+def _require_workspace_id(monkeypatch):
+    monkeypatch.setenv("WORKSPACE_ID", "00000000-0000-0000-0000-000000000000")
+    monkeypatch.setenv("PLATFORM_URL", "http://test.invalid")
+    yield
+
+
+def _run(coro):
+    return asyncio.get_event_loop().run_until_complete(coro)
+
+
+# ---------------------------------------------------------------------------
+# tool_inbox_peek
+# ---------------------------------------------------------------------------
+
+
+class TestToolInboxPeek:
+    def test_returns_not_enabled_when_state_none(self):
+        import a2a_tools
+
+        with patch("inbox.get_state", return_value=None):
+            out = _run(a2a_tools.tool_inbox_peek())
+        assert "not enabled" in out
+
+    def test_returns_json_array_of_messages(self):
+        import a2a_tools
+
+        msg1 = MagicMock()
+        msg1.to_dict.return_value = {"activity_id": "a1", "kind": "canvas_user"}
+        msg2 = MagicMock()
+        msg2.to_dict.return_value = {"activity_id": "a2", "kind": "peer_agent"}
+
+        fake_state = MagicMock()
+        fake_state.peek.return_value = [msg1, msg2]
+
+        with patch("inbox.get_state", return_value=fake_state):
+            out = _run(a2a_tools.tool_inbox_peek(limit=5))
+        # peek limit is forwarded
+        fake_state.peek.assert_called_once_with(limit=5)
+        parsed = json.loads(out)
+        assert len(parsed) == 2
+        assert parsed[0]["activity_id"] == "a1"
+
+    def test_non_int_limit_falls_back_to_10(self):
+        import a2a_tools
+
+        fake_state = MagicMock()
+        fake_state.peek.return_value = []
+        with patch("inbox.get_state", return_value=fake_state):
+            _run(a2a_tools.tool_inbox_peek(limit="garbage"))  # type: ignore[arg-type]
+        fake_state.peek.assert_called_once_with(limit=10)
+
+
+# ---------------------------------------------------------------------------
+# tool_inbox_pop
+# ---------------------------------------------------------------------------
+
+
+class TestToolInboxPop:
+    def test_returns_not_enabled_when_state_none(self):
+        import a2a_tools
+
+        with patch("inbox.get_state", return_value=None):
+            out = _run(a2a_tools.tool_inbox_pop("act-1"))
+        assert "not enabled" in out
+
+    def test_rejects_empty_activity_id(self):
+        import a2a_tools
+
+        fake_state = MagicMock()
+        with patch("inbox.get_state", return_value=fake_state):
+            out = _run(a2a_tools.tool_inbox_pop(""))
+        assert "activity_id is required" in out
+        fake_state.pop.assert_not_called()
+
+    def test_rejects_non_str_activity_id(self):
+        import a2a_tools
+
+        fake_state = MagicMock()
+        with patch("inbox.get_state", return_value=fake_state):
+            out = _run(a2a_tools.tool_inbox_pop(123))  # type: ignore[arg-type]
+        assert "activity_id is required" in out
+        fake_state.pop.assert_not_called()
+
+    def test_returns_removed_true_when_popped(self):
+        import a2a_tools
+
+        fake_state = MagicMock()
+        fake_state.pop.return_value = MagicMock()  # truthy = something was removed
+        with patch("inbox.get_state", return_value=fake_state):
+            out = _run(a2a_tools.tool_inbox_pop("act-7"))
+        parsed = json.loads(out)
+        assert parsed == {"removed": True, "activity_id": "act-7"}
+        fake_state.pop.assert_called_once_with("act-7")
+
+    def test_returns_removed_false_when_unknown(self):
+        import a2a_tools
+
+        fake_state = MagicMock()
+        fake_state.pop.return_value = None
+        with patch("inbox.get_state", return_value=fake_state):
+            out = _run(a2a_tools.tool_inbox_pop("act-missing"))
+        parsed = json.loads(out)
+        assert parsed == {"removed": False, "activity_id": "act-missing"}
+
+
+# ---------------------------------------------------------------------------
+# tool_wait_for_message
+# ---------------------------------------------------------------------------
+
+
+class TestToolWaitForMessage:
+    def test_returns_not_enabled_when_state_none(self):
+        import a2a_tools
+
+        with patch("inbox.get_state", return_value=None):
+            out = _run(a2a_tools.tool_wait_for_message(timeout_secs=1.0))
+        assert "not enabled" in out
+
+    def test_timeout_payload_when_no_message(self):
+        import a2a_tools
+
+        fake_state = MagicMock()
+        fake_state.wait.return_value = None
+        with patch("inbox.get_state", return_value=fake_state):
+            out = _run(a2a_tools.tool_wait_for_message(timeout_secs=0.1))
+        parsed = json.loads(out)
+        assert parsed["timeout"] is True
+        assert parsed["timeout_secs"] == 0.1
+
+    def test_returns_message_when_delivered(self):
+        import a2a_tools
+
+        msg = MagicMock()
+        msg.to_dict.return_value = {"activity_id": "a-9", "kind": "peer_agent"}
+        fake_state = MagicMock()
+        fake_state.wait.return_value = msg
+        with patch("inbox.get_state", return_value=fake_state):
+            out = _run(a2a_tools.tool_wait_for_message(timeout_secs=2.0))
+        parsed = json.loads(out)
+        assert parsed["activity_id"] == "a-9"
+
+    def test_timeout_clamped_to_300(self):
+        import a2a_tools
+
+        fake_state = MagicMock()
+        fake_state.wait.return_value = None
+        with patch("inbox.get_state", return_value=fake_state):
+            _run(a2a_tools.tool_wait_for_message(timeout_secs=99999))
+        # Whatever wait was called with, it must not exceed 300
+        passed = fake_state.wait.call_args.args[0]
+        assert passed == 300.0
+
+    def test_timeout_clamped_to_zero_floor(self):
+        import a2a_tools
+
+        fake_state = MagicMock()
+        fake_state.wait.return_value = None
+        with patch("inbox.get_state", return_value=fake_state):
+            _run(a2a_tools.tool_wait_for_message(timeout_secs=-5))
+        passed = fake_state.wait.call_args.args[0]
+        assert passed == 0.0
+
+    def test_non_numeric_timeout_falls_back_to_60(self):
+        import a2a_tools
+
+        fake_state = MagicMock()
+        fake_state.wait.return_value = None
+        with patch("inbox.get_state", return_value=fake_state):
+            _run(a2a_tools.tool_wait_for_message(timeout_secs="garbage"))  # type: ignore[arg-type]
+        passed = fake_state.wait.call_args.args[0]
+        assert passed == 60.0

workspace/tests/test_a2a_tools_messaging.py

@@ -0,0 +1,92 @@
+"""Drift gate + smoke tests for ``a2a_tools_messaging`` (RFC #2873 iter 4d).
+
+The full behavior matrix lives in ``test_a2a_tools_impl.py`` —
+TestToolSendMessageToUser + TestToolListPeers + TestToolGetWorkspaceInfo
++ TestChatHistory all patch ``a2a_tools_messaging.foo`` after the iter
+4d retarget.
+
+This file pins:
+
+  1. **Drift gate** — every previously-public symbol on ``a2a_tools``
+     is the EXACT same callable / value as ``a2a_tools_messaging.foo``.
+     Wraps would silently lose existing test coverage; this gate
+     fails fast on that drift.
+  2. **Import contract** — ``a2a_tools_messaging`` does NOT pull in
+     ``a2a_tools`` at module-load time (the layered architecture: it
+     depends on ``a2a_tools_rbac`` + ``a2a_client`` + ``platform_auth``,
+     never the kitchen-sink module).
+"""
+from __future__ import annotations
+
+import sys
+
+import pytest
+
+
+@pytest.fixture(autouse=True)
+def _require_workspace_id(monkeypatch):
+    monkeypatch.setenv("WORKSPACE_ID", "00000000-0000-0000-0000-000000000000")
+    monkeypatch.setenv("PLATFORM_URL", "http://test.invalid")
+    yield
+
+
+# ============== Drift gate ==============
+
+class TestBackCompatAliases:
+    def test_tool_send_message_to_user_alias(self):
+        import a2a_tools
+        import a2a_tools_messaging
+        assert (
+            a2a_tools.tool_send_message_to_user
+            is a2a_tools_messaging.tool_send_message_to_user
+        )
+
+    def test_tool_list_peers_alias(self):
+        import a2a_tools
+        import a2a_tools_messaging
+        assert a2a_tools.tool_list_peers is a2a_tools_messaging.tool_list_peers
+
+    def test_tool_get_workspace_info_alias(self):
+        import a2a_tools
+        import a2a_tools_messaging
+        assert (
+            a2a_tools.tool_get_workspace_info
+            is a2a_tools_messaging.tool_get_workspace_info
+        )
+
+    def test_tool_chat_history_alias(self):
+        import a2a_tools
+        import a2a_tools_messaging
+        assert a2a_tools.tool_chat_history is a2a_tools_messaging.tool_chat_history
+
+    def test_upload_chat_files_alias(self):
+        import a2a_tools
+        import a2a_tools_messaging
+        assert a2a_tools._upload_chat_files is a2a_tools_messaging._upload_chat_files
+
+
+# ============== Import contract ==============
+
+class TestImportContract:
+    def test_messaging_module_does_not_load_a2a_tools(self, monkeypatch):
+        """`a2a_tools_messaging` must depend on `a2a_tools_rbac` (the
+        layered architecture), `a2a_client`, and `platform_auth` — but
+        NEVER on the kitchen-sink `a2a_tools`. Top-level
+        `from a2a_tools import …` would re-introduce the circular
+        dependency that motivated the lazy-import contract for the
+        delegation module."""
+        for m in ("a2a_tools", "a2a_tools_messaging"):
+            sys.modules.pop(m, None)
+
+        import a2a_tools_messaging  # noqa: F401
+        assert "a2a_tools_messaging" in sys.modules
+
+    def test_a2a_tools_re_exports_messaging_handlers(self):
+        """Opposite direction: a2a_tools surfaces every messaging
+        symbol so existing call sites + tests work unchanged."""
+        import a2a_tools
+        assert hasattr(a2a_tools, "tool_send_message_to_user")
+        assert hasattr(a2a_tools, "tool_list_peers")
+        assert hasattr(a2a_tools, "tool_get_workspace_info")
+        assert hasattr(a2a_tools, "tool_chat_history")
+        assert hasattr(a2a_tools, "_upload_chat_files")