fix(tools): unify LangChain tool names with MCP — fix doc/tool drift from #2237

workspace/adapter_base.py

@@ -421,8 +421,8 @@ async def _common_setup(self, config: AdapterConfig) -> SetupResult:
         from coordinator import get_children, get_parent_context, build_children_description
         from prompt import build_system_prompt, get_peer_capabilities, get_platform_instructions
         from builtin_tools.approval import request_approval
-        from builtin_tools.delegation import delegate_to_workspace, check_delegation_status
-        from builtin_tools.memory import commit_memory, search_memory
+        from builtin_tools.delegation import delegate_task, delegate_task_async, check_task_status
+        from builtin_tools.memory import commit_memory, recall_memory
         from builtin_tools.sandbox import run_code
 
         platform_url = os.environ.get("PLATFORM_URL", "http://host.docker.internal:8080")
@@ -455,8 +455,19 @@ async def _common_setup(self, config: AdapterConfig) -> SetupResult:
                     seen_skill_ids.add(skill.metadata.id)
         logger.info(f"Loaded {len(loaded_skills)} skills: {[s.metadata.id for s in loaded_skills]}")
 
-        # Assemble tools: 6 core + skill tools
-        all_tools = [delegate_to_workspace, check_delegation_status, request_approval, commit_memory, search_memory, run_code]
+        # Assemble tools: 7 core + skill tools.
+        # Tool naming is unified across MCP and LangChain runtimes so the
+        # platform-injected docs (get_a2a_instructions, get_hma_instructions)
+        # name tools that actually exist in both worlds:
+        #   delegate_task        — sync, returns peer's response text
+        #   delegate_task_async  — async, returns task_id
+        #   check_task_status    — poll an async delegation
+        #   commit_memory        — write to HMA with scope
+        #   recall_memory        — read from HMA (LOCAL+TEAM+GLOBAL)
+        all_tools = [
+            delegate_task, delegate_task_async, check_task_status,
+            request_approval, commit_memory, recall_memory, run_code,
+        ]
         for skill in loaded_skills:
             all_tools.extend(skill.tools)
 

workspace/builtin_tools/delegation.py

@@ -2,7 +2,7 @@
 
 Delegations are non-blocking: the tool fires the A2A request in the background
 and returns immediately with a task_id. The agent can check status anytime via
-check_delegation_status, or just continue working and check later.
+check_task_status, or just continue working and check later.
 
 When the delegate responds, the result is stored and the agent is notified
 via a status update.
@@ -44,7 +44,7 @@ class DelegationStatus(str, Enum):
     # The reply will arrive via the platform's stitch path when the
     # peer finishes its current work. The LLM should WAIT, not retry,
     # and definitely not fall back to doing the work itself — see the
-    # check_delegation_status docstring for the prompt-side guidance.
+    # check_task_status docstring for the prompt-side guidance.
     QUEUED = "queued"
     COMPLETED = "completed"
     FAILED = "failed"
@@ -110,7 +110,7 @@ async def _record_delegation_on_platform(task_id: str, target_workspace_id: str,
     Best-effort POST to /workspaces/<self>/delegations/record. The agent still
     fires A2A directly for speed + OTEL propagation, but the platform's
     GET /delegations endpoint now mirrors the same set an agent's local
-    check_delegation_status sees.
+    check_task_status sees.
     """
     try:
         async with httpx.AsyncClient(timeout=10) as client:
@@ -129,11 +129,11 @@ async def _record_delegation_on_platform(task_id: str, target_workspace_id: str,
 async def _refresh_queued_from_platform(task_id: str) -> bool:
     """Lazy-refresh a QUEUED delegation's local state from the platform.
 
-    Called by check_delegation_status when local status is QUEUED. The
+    Called by check_task_status when local status is QUEUED. The
     platform's drain stitch (a2a_queue.go) updates the delegate_result
     activity_logs row when a queued delegation eventually completes,
     but it has no callback to this runtime — without this lazy refresh,
-    the LLM polling check_delegation_status would see "queued" forever
+    the LLM polling check_task_status would see "queued" forever
     even after the platform has the result.
 
     Returns True if the local delegation was updated to a terminal state
@@ -215,7 +215,7 @@ async def _execute_delegation(task_id: str, workspace_id: str, task: str):
     delegation.status = DelegationStatus.IN_PROGRESS
 
     # #64: register on the platform so GET /workspaces/<self>/delegations
-    # sees the same set as check_delegation_status. Best-effort — platform
+    # sees the same set as check_task_status. Best-effort — platform
     # unreachability must not block the actual A2A delegation.
     await _record_delegation_on_platform(task_id, workspace_id, task)
 
@@ -286,7 +286,7 @@ async def _execute_delegation(task_id: str, workspace_id: str, task: str):
                     # accepted the request but the peer's runtime is
                     # mid-task. Platform-side drain will deliver the
                     # reply asynchronously. Mark QUEUED locally so
-                    # check_delegation_status can surface that state
+                    # check_task_status can surface that state
                     # to the LLM with explicit "wait, don't bypass"
                     # guidance. Do NOT mark FAILED — the request is
                     # alive in the platform's queue, not lost.
@@ -371,22 +371,49 @@ async def _execute_delegation(task_id: str, workspace_id: str, task: str):
 
 
 @tool
-async def delegate_to_workspace(
+async def delegate_task(
+    workspace_id: str,
+    task: str,
+) -> str:
+    """Delegate a task to a peer workspace via A2A and WAIT for the response.
+
+    Synchronous variant — blocks until the peer replies (or the platform's
+    A2A round-trip times out). Use this for QUICK questions and small
+    sub-tasks where you can afford to wait inline.
+
+    For longer-running work (research, synthesis, multi-minute jobs), use
+    delegate_task_async + check_task_status instead so you don't hold this
+    workspace busy waiting.
+
+    Args:
+        workspace_id: The ID of the target workspace to delegate to.
+        task: The task description to send to the peer.
+
+    Returns:
+        The peer's response text directly (or a "DELEGATION FAILED" string
+        prefixed with recovery guidance if the round-trip errored).
+    """
+    from a2a_tools import tool_delegate_task
+    return await tool_delegate_task(workspace_id, task)
+
+
+@tool
+async def delegate_task_async(
     workspace_id: str,
     task: str,
 ) -> dict:
     """Delegate a task to a peer workspace via A2A protocol (non-blocking).
 
     Sends the task in the background and returns immediately with a task_id.
-    Use check_delegation_status to poll for the result, or continue working
+    Use check_task_status to poll for the result, or continue working
     and check later. The delegate works independently.
 
     Args:
         workspace_id: The ID of the target workspace to delegate to.
         task: The task description to send to the peer.
 
     Returns:
-        A dict with task_id and status="delegated". Use check_delegation_status(task_id) to get results.
+        A dict with task_id and status="delegated". Use check_task_status(task_id) to get results.
     """
     task_id = str(uuid.uuid4())
 
@@ -417,12 +444,12 @@ async def delegate_to_workspace(
         "success": True,
         "task_id": task_id,
         "status": "delegated",
-        "message": f"Task delegated to {workspace_id}. Use check_delegation_status('{task_id}') to get the result when ready.",
+        "message": f"Task delegated to {workspace_id}. Use check_task_status('{task_id}') to get the result when ready.",
     }
 
 
 @tool
-async def check_delegation_status(
+async def check_task_status(
     task_id: str = "",
 ) -> dict:
     """Check the status of a delegated task, or list all active delegations.
@@ -434,7 +461,7 @@ async def check_delegation_status(
       processing a prior task. The reply WILL arrive — the platform's
       drain re-dispatches when the peer is free. This tool transparently
       polls the platform for the eventual outcome on each call, so
-      keep polling check_delegation_status periodically and you'll see
+      keep polling check_task_status periodically and you'll see
       the status flip to "completed" / "failed" automatically.
       Do NOT retry the delegation. Do NOT do the work yourself.
       Acknowledge to the user that the peer is busy and will reply,
@@ -445,7 +472,7 @@ async def check_delegation_status(
       yourself if status is "failed", never if status is "queued".
 
     Args:
-        task_id: The task_id returned by delegate_to_workspace. If empty, lists all delegations.
+        task_id: The task_id returned by delegate_task_async. If empty, lists all delegations.
 
     Returns:
         Status and result (if completed) of the delegation.

workspace/builtin_tools/memory.py

@@ -8,7 +8,7 @@
 RBAC enforcement
 ----------------
 ``commit_memory`` requires the ``"memory.write"`` action.
-``search_memory`` requires the ``"memory.read"`` action.
+``recall_memory`` requires the ``"memory.read"`` action.
 Roles are read from ``config.yaml`` under ``rbac.roles`` (default: operator).
 
 Audit trail
@@ -188,7 +188,7 @@ async def commit_memory(content: str, scope: str = "LOCAL") -> dict:
 
 
 @tool
-async def search_memory(query: str = "", scope: str = "") -> dict:
+async def recall_memory(query: str = "", scope: str = "") -> dict:
     """Search stored memories.
 
     Args:

workspace/coordinator.py

@@ -81,7 +81,7 @@ def build_children_description(children: list[dict]) -> str:
         children,
         heading="## Your Team (sub-workspaces you coordinate)",
         instruction=(
-            "Use the `delegate_to_workspace` tool to send tasks to the chosen member. "
+            "Use the `delegate_task_async` tool to send tasks to the chosen member. "
             "Only delegate to members listed above."
         ),
     )
@@ -92,7 +92,7 @@ def build_children_description(children: list[dict]) -> str:
             "",
             "### Coordination Rules — MANDATORY",
             "1. You are a COORDINATOR. Your ONLY job is to delegate and synthesize. NEVER do the work yourself.",
-            "2. For EVERY task, use `delegate_to_workspace` to send it to the appropriate team member(s). "
+            "2. For EVERY task, use `delegate_task_async` to send it to the appropriate team member(s). "
             "Do this BEFORE writing any analysis, code, or research yourself.",
             "3. If a task spans multiple members, delegate to ALL of them in parallel and aggregate results.",
             "4. If ALL members are offline/paused, tell the caller which members are unavailable. "
@@ -120,7 +120,7 @@ async def route_task_to_team(
         task: The task description to route.
         preferred_member_id: Optional — directly delegate to this member.
     """
-    from builtin_tools.delegation import delegate_to_workspace as delegate
+    from builtin_tools.delegation import delegate_task_async as delegate
 
     children = await get_children()
     decision = build_team_routing_payload(

workspace/main.py

@@ -337,11 +337,13 @@ def _on_skill_reload(updated_skill):
                 # Rebuild the agent's tool list from updated skills
                 if hasattr(adapter, "all_tools") and hasattr(adapter, "system_prompt"):
                     from builtin_tools.approval import request_approval
-                    from builtin_tools.delegation import delegate_to_workspace
-                    from builtin_tools.memory import commit_memory, search_memory
+                    from builtin_tools.delegation import delegate_task, delegate_task_async, check_task_status
+                    from builtin_tools.memory import commit_memory, recall_memory
                     from builtin_tools.sandbox import run_code
-                    base_tools = [delegate_to_workspace, request_approval,
-                                  commit_memory, search_memory, run_code]
+                    base_tools = [
+                        delegate_task, delegate_task_async, check_task_status,
+                        request_approval, commit_memory, recall_memory, run_code,
+                    ]
                     skill_tools = []
                     for sk in adapter.loaded_skills:
                         skill_tools.extend(sk.tools)

workspace/policies/routing.py

@@ -64,7 +64,7 @@ def build_team_routing_payload(
         "action": "choose_member",
         "message": (
             f"You have {len(members)} team members. "
-            "Choose the best one for this task and call delegate_to_workspace with their ID."
+            "Choose the best one for this task and call delegate_task_async with their ID."
         ),
         "task": task,
         "members": members,

workspace/shared_runtime.py

@@ -140,7 +140,7 @@ def build_peer_section(
     *,
     heading: str = "## Your Peers (workspaces you can delegate to)",
     instruction: str = (
-        "Use the `delegate_to_workspace` tool to send tasks to peers. "
+        "Use the `delegate_task_async` tool to send tasks to peers. "
         "Only delegate to peers listed above."
     ),
 ) -> str:

workspace/tests/conftest.py

@@ -113,10 +113,12 @@ def _make_tools_mocks():
     tools_mod.__path__ = []  # Make it a proper package
 
     tools_delegation_mod = ModuleType("builtin_tools.delegation")
-    tools_delegation_mod.delegate_to_workspace = MagicMock()
-    tools_delegation_mod.delegate_to_workspace.name = "delegate_to_workspace"
-    tools_delegation_mod.check_delegation_status = MagicMock()
-    tools_delegation_mod.check_delegation_status.name = "check_delegation_status"
+    tools_delegation_mod.delegate_task = MagicMock()
+    tools_delegation_mod.delegate_task.name = "delegate_task"
+    tools_delegation_mod.delegate_task_async = MagicMock()
+    tools_delegation_mod.delegate_task_async.name = "delegate_task_async"
+    tools_delegation_mod.check_task_status = MagicMock()
+    tools_delegation_mod.check_task_status.name = "check_task_status"
 
     tools_approval_mod = ModuleType("builtin_tools.approval")
     tools_approval_mod.request_approval = MagicMock()
@@ -125,8 +127,8 @@ def _make_tools_mocks():
     tools_memory_mod = ModuleType("builtin_tools.memory")
     tools_memory_mod.commit_memory = MagicMock()
     tools_memory_mod.commit_memory.name = "commit_memory"
-    tools_memory_mod.search_memory = MagicMock()
-    tools_memory_mod.search_memory.name = "search_memory"
+    tools_memory_mod.recall_memory = MagicMock()
+    tools_memory_mod.recall_memory.name = "recall_memory"
 
     tools_sandbox_mod = ModuleType("builtin_tools.sandbox")
     tools_sandbox_mod.run_code = MagicMock()

workspace/tests/test_coordinator_routing.py

@@ -28,7 +28,7 @@ async def test_route_task_to_team_delegates_preferred_member(monkeypatch):
 
     delegate = MagicMock()
     delegate.ainvoke = AsyncMock(return_value={"ok": True})
-    monkeypatch.setattr(sys.modules["builtin_tools.delegation"], "delegate_to_workspace", delegate)
+    monkeypatch.setattr(sys.modules["builtin_tools.delegation"], "delegate_task_async", delegate)
 
     result = await coordinator.route_task_to_team(
         "Do the thing",
@@ -58,4 +58,4 @@ def test_build_children_description_reuses_shared_renderer():
     assert "## Your Team (sub-workspaces you coordinate)" in description
     assert "**Alpha** (id: `child-1`, status: online)" in description
     assert "Skills: research" in description
-    assert "delegate_to_workspace" in description
+    assert "delegate_task_async" in description