feat(a2a): client tools config-gated — disabled unless enabled (−561 tok/call on unconfigured installs) (#97421)

* feat(a2a): outbound client tools are config-gated — served only when a2a_agents configured, inbound platform enabled, or A2A_PORT set (-561 tok/call on unconfigured installs)

* ci: retrigger after runner startup_failure on rerun attempt

plugins/platforms/a2a/tools.py

@@ -582,8 +582,42 @@ _HANDLERS = {
 }
 
 
+def _a2a_tools_available() -> bool:
+    """check_fn for the outbound client tools: serve them ONLY when the
+    operator has opted into A2A somehow — peers configured under
+    ``a2a_agents`` in config.yaml, or the inbound platform enabled
+    (a peer-reachable Hermes plausibly dials back).
+
+    Maintainer-directed (#95681): these registered unconditionally, so
+    every session on every install paid ~561 tok/call for tools whose
+    only possible output without config is 'no peers configured'. A2A is
+    unrelated to Bot Mode (bots talk over gateway RPCs) — for most
+    installs this toolset is foreign-agent plumbing they never enabled.
+    Config adds mid-session surface at the next compaction (#97073).
+    """
+    cfg = {}
+    try:
+        cfg = _load_config()
+        if cfg.get("a2a_agents"):
+            return True
+    except Exception:  # noqa: BLE001
+        pass
+    try:
+        import os as _os
+
+        if _os.getenv("A2A_PORT"):
+            return True
+        platforms = cfg.get("platforms") or {}
+        a2a_cfg = platforms.get("a2a") or {}
+        if isinstance(a2a_cfg, dict) and a2a_cfg.get("enabled"):
+            return True
+    except Exception:  # noqa: BLE001
+        pass
+    return False
+
+
 def register_tools(ctx) -> None:
-    """Register the client tools in the ``a2a`` toolset."""
+    """Register the client tools in the ``a2a`` toolset (config-gated)."""
     for name, schema in _SCHEMAS.items():
         function_schema = schema["function"]
         ctx.register_tool(
@@ -593,4 +627,5 @@ def register_tools(ctx) -> None:
             handler=_HANDLERS[name],
             description=function_schema["description"],
             emoji="\U0001f9e9",  # puzzle piece
+            check_fn=_a2a_tools_available,
         )

tests/plugins/test_a2a_schema_registration.py

@@ -12,6 +12,14 @@ from tools.registry import ToolRegistry
 def test_a2a_call_schema_round_trips_through_tool_describe(monkeypatch):
     registry = ToolRegistry()
 
+    # The client tools are config-gated now (test_a2a_tools_gate.py):
+    # open the gate the way a real install would — configure a peer.
+    monkeypatch.setattr(
+        a2a_tools,
+        "_load_config",
+        lambda: {"a2a_agents": {"peer": {"url": "http://localhost:9999"}}},
+    )
+
     class _Context:
         def register_tool(self, name, toolset, schema, handler, **kwargs):
             registry.register(

tests/plugins/test_a2a_tools_gate.py

@@ -0,0 +1,65 @@
+"""a2a client tools config gate (#95681, maintainer-directed).
+
+The 5 outbound a2a_* tools registered unconditionally — every session on
+every install paid ~561 tok/call for a toolset whose only possible output
+without config is "no peers configured". A2A is NOT the Bot Mode
+mechanism (bots talk over gateway RPCs); it is opt-in foreign-agent
+plumbing. Gate: serve only when a2a_agents is non-empty, the inbound
+platform is enabled, or A2A_PORT is set. Fail closed.
+"""
+import os
+import sys
+import unittest
+from unittest.mock import patch
+
+sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", ".."))
+
+import plugins.platforms.a2a.tools as a2at
+
+
+class TestA2AToolsGate(unittest.TestCase):
+    def setUp(self):
+        os.environ.pop("A2A_PORT", None)
+
+    def _avail(self, cfg):
+        with patch.object(a2at, "_load_config", return_value=cfg):
+            return a2at._a2a_tools_available()
+
+    def test_unconfigured_install_serves_nothing(self):
+        self.assertFalse(self._avail({}))
+        self.assertFalse(self._avail({"a2a_agents": {}}))
+
+    def test_peers_configured_serves(self):
+        self.assertTrue(self._avail({"a2a_agents": {"r": {"url": "http://x"}}}))
+
+    def test_inbound_platform_enabled_serves(self):
+        self.assertTrue(self._avail({"platforms": {"a2a": {"enabled": True}}}))
+
+    def test_a2a_port_env_serves(self):
+        os.environ["A2A_PORT"] = "9999"
+        try:
+            self.assertTrue(self._avail({}))
+        finally:
+            os.environ.pop("A2A_PORT", None)
+
+    def test_config_crash_fails_closed(self):
+        with patch.object(a2at, "_load_config", side_effect=RuntimeError("boom")):
+            self.assertFalse(a2at._a2a_tools_available())
+
+    def test_all_five_tools_carry_the_gate(self):
+        """Every a2a_* registration must pass the check_fn — a sixth tool
+        added without it would silently reopen the hole."""
+        seen = {}
+
+        class Ctx:
+            def register_tool(self, name, **kw):
+                seen[name] = kw.get("check_fn")
+
+        a2at.register_tools(Ctx())
+        self.assertEqual(len(seen), 5, sorted(seen))
+        for name, fn in seen.items():
+            self.assertIs(fn, a2at._a2a_tools_available, name)
+
+
+if __name__ == "__main__":
+    unittest.main()