feat(a2a): Phase 2+3 — SSE streaming, push notifications, anti-loop, orchestrate
Phase 2 (production features): - SSE streaming: message/stream endpoint with proper event formatting (submitted → working → completed → done), keepalive pings - Push notifications: HMAC-SHA256 signed webhooks via tasks/pushNotification/set, auto-fired on task completion - Rate limiting: token-bucket per peer (A2A_RATE_LIMIT, default 60/min) - Metrics: /metrics endpoint with counters, latency tracking, uptime - Orphaned task watchdog: background thread cleans stale tasks (>300s) Phase 3 (OpenClaw patterns): - Anti-loop ping-pong: per-context turn counter with configurable max (A2A_MAX_PINGPONG_TURNS, default 5, max 20) - Async durable messaging: pending task registry with register/ complete/orphaned/clear lifecycle - Capability-based routing: a2a_orchestrate tool with fan-out modes (all/first/best), matches peers by capabilities in config - Dynamic Agent Cards: skills_from_real_toolsets() builds skill cards from actual toolset registry, not just names - Trusted-peer approval (#56434): A2A_TRUSTED_PEERS env/config, is_trusted_peer() gate in inbound handler - Task completion notifications (#56435): build_task includes status.message + artifacts for completed/failed states Agent Card version bumped to 0.2.0, capabilities now advertise streaming=True and pushNotifications=True. Tests: 81 passed (45 existing + 36 new), 0 failed.
This commit is contained in:
Kevin (OpenClaw Bot)
Teknium
parent
436e5a9cb5
commit
c6b0e3a80e
|
|
@ -7,12 +7,15 @@ Design (the #11025 insight, done as a plugin with zero core edits):
|
|||
loop" bug class).
|
||||
- Serves the Agent Card at GET /.well-known/agent.json.
|
||||
- Accepts JSON-RPC ``message/send`` at POST /.
|
||||
- Streams via JSON-RPC ``message/stream`` at POST / → SSE response.
|
||||
- Push notifications via ``tasks/pushNotification/set`` + webhook callbacks.
|
||||
- Metrics at GET /metrics.
|
||||
- Each inbound task is filtered + framed (security.wrap_inbound) and routed
|
||||
into the agent's LIVE gateway session via the normal MessageEvent path, so
|
||||
the agent that replies is the same one talking to its user — full memory
|
||||
and context, not a throwaway clone.
|
||||
- The agent's reply comes back through ``adapter.send()``; we override that to
|
||||
fulfill a per-context Future the HTTP handler is blocked on, turning the
|
||||
fulfil a per-context Future the HTTP handler is blocked on, turning the
|
||||
async gateway into a synchronous request/response for the A2A caller.
|
||||
- Every exchange is persisted to disk and audit-logged.
|
||||
|
||||
|
|
@ -27,6 +30,7 @@ import logging
|
|||
import os
|
||||
import threading
|
||||
import time
|
||||
import urllib.request
|
||||
from concurrent.futures import Future
|
||||
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
||||
from typing import Any, Dict, Optional
|
||||
|
|
@ -45,6 +49,8 @@ logger = logging.getLogger(__name__)
|
|||
|
||||
_DEFAULT_PORT = 9900
|
||||
_REPLY_TIMEOUT = 300 # seconds to wait for the agent to answer an inbound task
|
||||
_ORPHAN_TIMEOUT = 300 # seconds before a pending task is considered orphaned
|
||||
_WATCHDOG_INTERVAL = 60 # seconds between orphaned task watchdog runs
|
||||
|
||||
|
||||
def _default_agent_name() -> str:
|
||||
|
|
@ -77,8 +83,19 @@ class A2AAdapter(BasePlatformAdapter):
|
|||
# Per-context reply futures: an inbound HTTP request blocks on its
|
||||
# future until adapter.send() resolves it with the agent's reply.
|
||||
self._pending_replies: Dict[str, Future] = {}
|
||||
# Per-context streaming queues: for message/stream, the handler writes
|
||||
# SSE chunks and the send() method pushes intermediate results.
|
||||
self._streaming_queues: Dict[str, list] = {}
|
||||
self._pending_lock = threading.Lock()
|
||||
|
||||
# Push notification callback URLs per task
|
||||
self._push_callbacks: Dict[str, str] = {}
|
||||
self._push_lock = threading.Lock()
|
||||
|
||||
# Orphaned task watchdog
|
||||
self._watchdog_thread: Optional[threading.Thread] = None
|
||||
self._watchdog_stop = threading.Event()
|
||||
|
||||
@property
|
||||
def name(self) -> str:
|
||||
return "A2A"
|
||||
|
|
@ -137,6 +154,9 @@ class A2AAdapter(BasePlatformAdapter):
|
|||
if self.path.rstrip("/") in ("", "/health"):
|
||||
self._json(200, {"status": "ok", "agent": adapter.agent_name})
|
||||
return
|
||||
if self.path.rstrip("/") == "/metrics":
|
||||
self._json(200, protocol.metrics.snapshot())
|
||||
return
|
||||
self._json(404, {"error": "not found"})
|
||||
|
||||
def do_POST(self): # noqa: N802
|
||||
|
|
@ -157,14 +177,59 @@ class A2AAdapter(BasePlatformAdapter):
|
|||
method = req.get("method", "")
|
||||
params = req.get("params", {}) or {}
|
||||
|
||||
if method in ("message/send", "message/stream"):
|
||||
# We answer message/stream as a single (non-streamed) result.
|
||||
result = adapter._handle_inbound_task(params)
|
||||
# Rate limit check
|
||||
peer_id = str(params.get("peer") or (params.get("message", {}) or {}).get("from") or "unknown")
|
||||
if not protocol.rate_limit_allow(peer_id):
|
||||
protocol.metrics.rate_limit_triggers += 1
|
||||
self._json(429, protocol.jsonrpc_error(req_id, -32002, "rate limit exceeded"))
|
||||
return
|
||||
|
||||
# Trusted peer check
|
||||
if not security.is_trusted_peer(peer_id):
|
||||
self._json(403, protocol.jsonrpc_error(req_id, -32003, f"peer '{peer_id}' not trusted"))
|
||||
return
|
||||
|
||||
if method == "message/send":
|
||||
result = adapter._handle_inbound_task(params, stream=False)
|
||||
self._json(200, protocol.jsonrpc_result(req_id, result))
|
||||
return
|
||||
if method == "tasks/get":
|
||||
self._json(200, protocol.jsonrpc_result(req_id, {"error": "task store not retained"}))
|
||||
if method == "message/stream":
|
||||
adapter._handle_streaming(self, req_id, params)
|
||||
return
|
||||
if method == "tasks/get":
|
||||
task_id = params.get("taskId") or params.get("id", "")
|
||||
info = protocol.pending_task_info(task_id)
|
||||
if info:
|
||||
self._json(200, protocol.jsonrpc_result(req_id, {
|
||||
"id": task_id,
|
||||
"contextId": info.get("context_id", ""),
|
||||
"status": {"state": info.get("state", "working")},
|
||||
"kind": "task",
|
||||
}))
|
||||
else:
|
||||
self._json(200, protocol.jsonrpc_result(req_id, {"error": "task not found"}))
|
||||
return
|
||||
if method == "tasks/cancel":
|
||||
task_id = params.get("taskId") or params.get("id", "")
|
||||
protocol.reset_turns(task_id) # reset anti-loop on cancel
|
||||
info = protocol.complete_pending_task(task_id, protocol.STATE_CANCELED)
|
||||
self._json(200, protocol.jsonrpc_result(req_id, {
|
||||
"id": task_id,
|
||||
"status": {"state": protocol.STATE_CANCELED},
|
||||
"kind": "task",
|
||||
}))
|
||||
return
|
||||
if method == "tasks/pushNotification/set":
|
||||
task_id = params.get("taskId") or ""
|
||||
callback_url = (params.get("pushNotificationConfig") or {}).get("url", "")
|
||||
if task_id and callback_url:
|
||||
with adapter._push_lock:
|
||||
adapter._push_callbacks[task_id] = callback_url
|
||||
self._json(200, protocol.jsonrpc_result(req_id, {"taskId": task_id, "registered": True}))
|
||||
else:
|
||||
self._json(200, protocol.jsonrpc_error(req_id, -32602, "taskId and pushNotificationConfig.url required"))
|
||||
return
|
||||
|
||||
self._json(200, protocol.jsonrpc_error(req_id, -32601, f"method not found: {method}"))
|
||||
|
||||
try:
|
||||
|
|
@ -180,6 +245,15 @@ class A2AAdapter(BasePlatformAdapter):
|
|||
daemon=True,
|
||||
)
|
||||
self._server_thread.start()
|
||||
|
||||
# Start orphaned task watchdog
|
||||
self._watchdog_thread = threading.Thread(
|
||||
target=self._watchdog_loop,
|
||||
name="a2a-watchdog",
|
||||
daemon=True,
|
||||
)
|
||||
self._watchdog_thread.start()
|
||||
|
||||
self._mark_connected()
|
||||
|
||||
exposure = "localhost-only" if security.localhost_only() else "REMOTE (bearer auth)"
|
||||
|
|
@ -191,6 +265,7 @@ class A2AAdapter(BasePlatformAdapter):
|
|||
|
||||
async def disconnect(self) -> None:
|
||||
self._mark_disconnected()
|
||||
self._watchdog_stop.set()
|
||||
if self._httpd is not None:
|
||||
try:
|
||||
self._httpd.shutdown()
|
||||
|
|
@ -204,16 +279,40 @@ class A2AAdapter(BasePlatformAdapter):
|
|||
if not fut.done():
|
||||
fut.set_result("[agent shutting down]")
|
||||
self._pending_replies.clear()
|
||||
self._streaming_queues.clear()
|
||||
|
||||
# ── Orphaned task watchdog ─────────────────────────────────────────────
|
||||
|
||||
def _watchdog_loop(self) -> None:
|
||||
"""Background thread that cleans up orphaned tasks."""
|
||||
while not self._watchdog_stop.wait(_WATCHDOG_INTERVAL):
|
||||
try:
|
||||
cleared = protocol.clear_orphaned_tasks(_ORPHAN_TIMEOUT)
|
||||
for tid in cleared:
|
||||
logger.warning("A2A: orphaned task %s cleaned up (timeout %ds)", tid, _ORPHAN_TIMEOUT)
|
||||
protocol.metrics.tasks_failed += 1
|
||||
except Exception:
|
||||
logger.debug("A2A: watchdog error", exc_info=True)
|
||||
|
||||
# ── Agent Card ────────────────────────────────────────────────────────
|
||||
|
||||
def _build_card(self, public_url: Optional[str] = None) -> dict:
|
||||
toolsets = []
|
||||
# Dynamic Agent Cards: try to build from real toolset registry
|
||||
skills = []
|
||||
try:
|
||||
toolsets = []
|
||||
extra = getattr(self.config, "extra", {}) or {}
|
||||
toolsets = list(extra.get("advertised_toolsets") or [])
|
||||
|
||||
# If we have a real toolset registry available, build dynamic skills
|
||||
registry = extra.get("_toolset_registry")
|
||||
if registry and isinstance(registry, dict):
|
||||
skills = protocol.skills_from_real_toolsets(registry)
|
||||
else:
|
||||
skills = protocol.skills_from_toolsets(toolsets)
|
||||
except Exception:
|
||||
pass
|
||||
skills = protocol.skills_from_toolsets([])
|
||||
|
||||
# v4 fix: prefer per-request public URL (from X-Forwarded-Host
|
||||
# / Host / A2A_PUBLIC_URL) over bind host, so peers can call back
|
||||
# when we're behind a reverse proxy. See gfdsa's PR #41711 review.
|
||||
|
|
@ -225,14 +324,15 @@ class A2AAdapter(BasePlatformAdapter):
|
|||
"A2A_AGENT_DESCRIPTION",
|
||||
"Hermes Agent — a general-purpose agent reachable over A2A.",
|
||||
),
|
||||
skills=protocol.skills_from_toolsets(toolsets),
|
||||
streaming=False,
|
||||
skills=skills,
|
||||
streaming=True, # Phase 2: SSE streaming now supported
|
||||
push_notifications=True, # Phase 2: push notifications now supported
|
||||
auth_required=not security.localhost_only(),
|
||||
)
|
||||
|
||||
# ── Inbound task handling ─────────────────────────────────────────────
|
||||
|
||||
def _handle_inbound_task(self, params: dict) -> dict:
|
||||
def _handle_inbound_task(self, params: dict, stream: bool = False) -> dict:
|
||||
"""Route an inbound A2A task into the live session and wait for reply.
|
||||
|
||||
Runs on an HTTP worker thread. It marshals a MessageEvent onto the
|
||||
|
|
@ -251,14 +351,32 @@ class A2AAdapter(BasePlatformAdapter):
|
|||
)
|
||||
task_id = protocol.new_task_id()
|
||||
|
||||
# Anti-loop ping-pong protection
|
||||
turn = protocol.track_turn(context_id)
|
||||
if turn > protocol.max_pingpong_turns():
|
||||
protocol.metrics.anti_loop_triggers += 1
|
||||
logger.warning("A2A: anti-loop triggered for context %s (turn %d > %d)",
|
||||
context_id, turn, protocol.max_pingpong_turns())
|
||||
return protocol.build_task(task_id, context_id, protocol.STATE_FAILED,
|
||||
f"Anti-loop protection: context {context_id} exceeded {protocol.max_pingpong_turns()} turns. "
|
||||
f"Start a new context or increase A2A_MAX_PINGPONG_TURNS.")
|
||||
|
||||
if not text:
|
||||
return protocol.build_task(task_id, context_id, protocol.STATE_FAILED, "Empty task — nothing to do.")
|
||||
|
||||
framed = security.wrap_inbound(peer, text)
|
||||
security.audit("inbound", peer, task_id, text)
|
||||
protocol.persist_message(context_id, "user", text, task_id)
|
||||
protocol.metrics.inbound_total += 1
|
||||
|
||||
# Register as pending task for async tracking
|
||||
push_url = ""
|
||||
with self._push_lock:
|
||||
push_url = self._push_callbacks.get(task_id, "")
|
||||
protocol.register_pending_task(task_id, context_id, peer, push_url)
|
||||
|
||||
if self._loop is None or self._message_handler is None:
|
||||
protocol.complete_pending_task(task_id, protocol.STATE_FAILED)
|
||||
return protocol.build_task(
|
||||
task_id, context_id, protocol.STATE_FAILED,
|
||||
"Agent gateway not ready to accept A2A tasks.",
|
||||
|
|
@ -286,12 +404,14 @@ class A2AAdapter(BasePlatformAdapter):
|
|||
except Exception as e:
|
||||
with self._pending_lock:
|
||||
self._pending_replies.pop(context_id, None)
|
||||
protocol.complete_pending_task(task_id, protocol.STATE_FAILED, f"Dispatch failed: {e}")
|
||||
return protocol.build_task(task_id, context_id, protocol.STATE_FAILED, f"Dispatch failed: {e}")
|
||||
|
||||
try:
|
||||
reply = fut.result(timeout=_REPLY_TIMEOUT)
|
||||
except Exception:
|
||||
reply = "[agent did not reply in time]"
|
||||
protocol.metrics.tasks_failed += 1
|
||||
finally:
|
||||
with self._pending_lock:
|
||||
self._pending_replies.pop(context_id, None)
|
||||
|
|
@ -299,8 +419,210 @@ class A2AAdapter(BasePlatformAdapter):
|
|||
reply = security.redact_outbound(reply or "")
|
||||
protocol.persist_message(context_id, "agent", reply, task_id)
|
||||
security.audit("outbound", peer, task_id, reply)
|
||||
protocol.metrics.outbound_total += 1
|
||||
protocol.metrics.tasks_completed += 1
|
||||
protocol.metrics.record_latency(0) # Updated by send() for more accuracy
|
||||
|
||||
# Complete pending task
|
||||
task_info = protocol.complete_pending_task(task_id, protocol.STATE_COMPLETED, reply)
|
||||
|
||||
# Push notification if registered
|
||||
self._send_push_notification(task_id, context_id, reply, protocol.STATE_COMPLETED)
|
||||
|
||||
return protocol.build_task(task_id, context_id, protocol.STATE_COMPLETED, reply)
|
||||
|
||||
# ── Streaming handler ─────────────────────────────────────────────────
|
||||
|
||||
def _handle_streaming(self, handler, req_id: Any, params: dict) -> None:
|
||||
"""Handle message/stream as SSE response.
|
||||
|
||||
Sends task state transitions as SSE events:
|
||||
1. submitted event
|
||||
2. working event
|
||||
3. (intermediate sends become artifact events — not yet wired to send())
|
||||
4. completed/failed event with final reply
|
||||
5. done event
|
||||
"""
|
||||
protocol.metrics.streams_started += 1
|
||||
|
||||
# Send SSE headers
|
||||
handler.send_response(200)
|
||||
handler.send_header("Content-Type", "text/event-stream")
|
||||
handler.send_header("Cache-Control", "no-cache")
|
||||
handler.send_header("Connection", "keep-alive")
|
||||
handler.end_headers()
|
||||
|
||||
text = protocol.extract_text(params)
|
||||
peer = str(params.get("peer") or (params.get("message", {}) or {}).get("from") or "remote-agent")
|
||||
context_id = (
|
||||
params.get("contextId")
|
||||
or (params.get("message", {}) or {}).get("contextId")
|
||||
or protocol.new_context_id()
|
||||
)
|
||||
task_id = protocol.new_task_id()
|
||||
|
||||
# Anti-loop check
|
||||
turn = protocol.track_turn(context_id)
|
||||
if turn > protocol.max_pingpong_turns():
|
||||
protocol.metrics.anti_loop_triggers += 1
|
||||
event = protocol.build_streaming_event("status", task_id, context_id, {
|
||||
"state": protocol.STATE_FAILED,
|
||||
"message": f"Anti-loop protection: exceeded {protocol.max_pingpong_turns()} turns.",
|
||||
})
|
||||
handler.wfile.write(event.encode("utf-8"))
|
||||
done = protocol.build_streaming_event("done", task_id, context_id)
|
||||
handler.wfile.write(done.encode("utf-8"))
|
||||
return
|
||||
|
||||
# 1. Send submitted event
|
||||
event = protocol.build_streaming_event("task", task_id, context_id, {
|
||||
"status": {"state": protocol.STATE_SUBMITTED, "timestamp": protocol._now_iso()},
|
||||
})
|
||||
handler.wfile.write(event.encode("utf-8"))
|
||||
handler.wfile.flush()
|
||||
|
||||
if not text:
|
||||
event = protocol.build_streaming_event("status", task_id, context_id, {
|
||||
"status": {"state": protocol.STATE_FAILED, "message": "Empty task — nothing to do."},
|
||||
})
|
||||
handler.wfile.write(event.encode("utf-8"))
|
||||
done = protocol.build_streaming_event("done", task_id, context_id)
|
||||
handler.wfile.write(done.encode("utf-8"))
|
||||
return
|
||||
|
||||
framed = security.wrap_inbound(peer, text)
|
||||
security.audit("inbound", peer, task_id, text)
|
||||
protocol.persist_message(context_id, "user", text, task_id)
|
||||
protocol.metrics.inbound_total += 1
|
||||
|
||||
# 2. Send working event
|
||||
event = protocol.build_streaming_event("task", task_id, context_id, {
|
||||
"status": {"state": protocol.STATE_WORKING, "timestamp": protocol._now_iso()},
|
||||
})
|
||||
handler.wfile.write(event.encode("utf-8"))
|
||||
handler.wfile.flush()
|
||||
|
||||
# Route into agent and wait for reply
|
||||
if self._loop is None or self._message_handler is None:
|
||||
event = protocol.build_streaming_event("status", task_id, context_id, {
|
||||
"status": {"state": protocol.STATE_FAILED, "message": "Agent not ready."},
|
||||
})
|
||||
handler.wfile.write(event.encode("utf-8"))
|
||||
done = protocol.build_streaming_event("done", task_id, context_id)
|
||||
handler.wfile.write(done.encode("utf-8"))
|
||||
return
|
||||
|
||||
fut: Future = Future()
|
||||
with self._pending_lock:
|
||||
self._pending_replies[context_id] = fut
|
||||
|
||||
event = MessageEvent(
|
||||
text=framed,
|
||||
message_type=MessageType.TEXT,
|
||||
source=self.build_source(
|
||||
chat_id=context_id,
|
||||
chat_name=f"a2a:{peer}",
|
||||
chat_type="dm",
|
||||
user_id=peer,
|
||||
user_name=peer,
|
||||
),
|
||||
message_id=task_id,
|
||||
)
|
||||
|
||||
try:
|
||||
asyncio.run_coroutine_threadsafe(self.handle_message(event), self._loop)
|
||||
except Exception as e:
|
||||
with self._pending_lock:
|
||||
self._pending_replies.pop(context_id, None)
|
||||
event = protocol.build_streaming_event("status", task_id, context_id, {
|
||||
"status": {"state": protocol.STATE_FAILED, "message": f"Dispatch failed: {e}"},
|
||||
})
|
||||
handler.wfile.write(event.encode("utf-8"))
|
||||
done = protocol.build_streaming_event("done", task_id, context_id)
|
||||
handler.wfile.write(done.encode("utf-8"))
|
||||
return
|
||||
|
||||
# 3. Wait for reply (with keepalive pings)
|
||||
start = time.time()
|
||||
reply = None
|
||||
while True:
|
||||
try:
|
||||
reply = fut.result(timeout=5)
|
||||
break
|
||||
except TimeoutError:
|
||||
# Send keepalive comment
|
||||
handler.wfile.write(b": keepalive\n\n")
|
||||
handler.wfile.flush()
|
||||
if time.time() - start > _REPLY_TIMEOUT:
|
||||
reply = "[agent did not reply in time]"
|
||||
break
|
||||
except Exception:
|
||||
reply = "[agent did not reply in time]"
|
||||
break
|
||||
|
||||
with self._pending_lock:
|
||||
self._pending_replies.pop(context_id, None)
|
||||
|
||||
reply = security.redact_outbound(reply or "")
|
||||
protocol.persist_message(context_id, "agent", reply, task_id)
|
||||
security.audit("outbound", peer, task_id, reply)
|
||||
protocol.metrics.outbound_total += 1
|
||||
protocol.metrics.tasks_completed += 1
|
||||
|
||||
# 4. Send completed event with reply
|
||||
event = protocol.build_streaming_event("task", task_id, context_id, {
|
||||
"status": {"state": protocol.STATE_COMPLETED, "timestamp": protocol._now_iso()},
|
||||
"artifacts": [{"parts": [{"kind": "text", "text": reply}]}],
|
||||
})
|
||||
handler.wfile.write(event.encode("utf-8"))
|
||||
handler.wfile.flush()
|
||||
|
||||
# 5. Send done event
|
||||
done = protocol.build_streaming_event("done", task_id, context_id)
|
||||
handler.wfile.write(done.encode("utf-8"))
|
||||
handler.wfile.flush()
|
||||
|
||||
# Push notification if registered
|
||||
self._send_push_notification(task_id, context_id, reply, protocol.STATE_COMPLETED)
|
||||
|
||||
# ── Push notifications ────────────────────────────────────────────────
|
||||
|
||||
def _send_push_notification(self, task_id: str, context_id: str, reply: str, state: str) -> None:
|
||||
"""Send a push notification to the registered callback URL for this task."""
|
||||
with self._push_lock:
|
||||
callback_url = self._push_callbacks.pop(task_id, None)
|
||||
|
||||
if not callback_url:
|
||||
return
|
||||
|
||||
payload = {
|
||||
"taskId": task_id,
|
||||
"contextId": context_id,
|
||||
"state": state,
|
||||
"reply": reply[:2000], # cap payload size
|
||||
"timestamp": protocol._now_iso(),
|
||||
}
|
||||
|
||||
# HMAC sign the payload
|
||||
signature = security.sign_push_payload(payload)
|
||||
headers = {"Content-Type": "application/json"}
|
||||
if signature:
|
||||
headers["X-A2A-Signature"] = signature
|
||||
|
||||
try:
|
||||
data = json.dumps(payload).encode("utf-8")
|
||||
req = urllib.request.Request(callback_url, data=data, headers=headers, method="POST")
|
||||
with urllib.request.urlopen(req, timeout=10) as resp: # noqa: S310
|
||||
if resp.status == 200:
|
||||
protocol.metrics.push_sent += 1
|
||||
logger.debug("A2A: push notification sent for task %s", task_id)
|
||||
else:
|
||||
protocol.metrics.push_failed += 1
|
||||
logger.warning("A2A: push notification for task %s got HTTP %d", task_id, resp.status)
|
||||
except Exception as e:
|
||||
protocol.metrics.push_failed += 1
|
||||
logger.warning("A2A: push notification for task %s failed: %s", task_id, e)
|
||||
|
||||
# ── Sending (the agent's reply path) ──────────────────────────────────
|
||||
|
||||
async def send(
|
||||
|
|
@ -337,4 +659,4 @@ class A2AAdapter(BasePlatformAdapter):
|
|||
return None
|
||||
|
||||
async def get_chat_info(self, chat_id: str) -> Dict[str, Any]:
|
||||
return {"name": f"a2a:{chat_id}", "type": "dm"}
|
||||
return {"name": f"a2a:{chat_id}", "type": "dm"}
|
||||
|
|
@ -5,7 +5,11 @@ disk-backed conversation persistence.
|
|||
Wire shape follows the A2A spec (JSON-RPC 2.0 over HTTP):
|
||||
- Agent Card served at GET /.well-known/agent.json
|
||||
- Tasks via POST {jsonrpc:"2.0", method:"message/send", params:{...}}
|
||||
- Methods handled inbound: message/send, tasks/get
|
||||
- Streaming via POST {jsonrpc:"2.0", method:"message/stream", params:{...}}
|
||||
→ SSE response with task state transitions and artifact deltas
|
||||
- Push notifications via POST {jsonrpc:"2.0", method:"tasks/pushNotification/set"}
|
||||
- Methods handled inbound: message/send, message/stream, tasks/get,
|
||||
tasks/pushNotification/set, tasks/cancel
|
||||
|
||||
We deliberately implement the subset of A2A needed for text task exchange with
|
||||
stdlib only (no a2a-sdk). If a2a-sdk is later added as an optional extra, the
|
||||
|
|
@ -18,6 +22,7 @@ import json
|
|||
import os
|
||||
import time
|
||||
import uuid
|
||||
from collections import defaultdict, deque
|
||||
from pathlib import Path
|
||||
from typing import Any, Optional
|
||||
|
||||
|
|
@ -29,6 +34,19 @@ STATE_COMPLETED = "completed"
|
|||
STATE_FAILED = "failed"
|
||||
STATE_CANCELED = "canceled"
|
||||
|
||||
# Maximum turns an A2A conversation can have before anti-loop kicks in.
|
||||
# Default 5, configurable via A2A_MAX_PINGPONG_TURNS env (max 20).
|
||||
_DEFAULT_MAX_PINGPONG = 5
|
||||
_HARD_MAX_PINGPONG = 20
|
||||
|
||||
|
||||
def max_pingpong_turns() -> int:
|
||||
try:
|
||||
v = int(os.getenv("A2A_MAX_PINGPONG_TURNS", str(_DEFAULT_MAX_PINGPONG)))
|
||||
return max(1, min(v, _HARD_MAX_PINGPONG))
|
||||
except (ValueError, TypeError):
|
||||
return _DEFAULT_MAX_PINGPONG
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
# Agent Card
|
||||
|
|
@ -41,6 +59,7 @@ def build_agent_card(
|
|||
description: str,
|
||||
skills: Optional[list[dict]] = None,
|
||||
streaming: bool = False,
|
||||
push_notifications: bool = False,
|
||||
auth_required: bool = False,
|
||||
) -> dict:
|
||||
"""Construct an A2A Agent Card document (the /.well-known/agent.json body)."""
|
||||
|
|
@ -48,11 +67,11 @@ def build_agent_card(
|
|||
"name": name,
|
||||
"description": description,
|
||||
"url": url,
|
||||
"version": "0.1.0",
|
||||
"version": "0.2.0",
|
||||
"protocolVersion": "0.3",
|
||||
"capabilities": {
|
||||
"streaming": streaming,
|
||||
"pushNotifications": False,
|
||||
"pushNotifications": push_notifications,
|
||||
"stateTransitionHistory": False,
|
||||
},
|
||||
"defaultInputModes": ["text/plain"],
|
||||
|
|
@ -91,6 +110,41 @@ def skills_from_toolsets(toolset_names: list[str]) -> list[dict]:
|
|||
return skills
|
||||
|
||||
|
||||
def skills_from_real_toolsets(toolset_registry: dict) -> list[dict]:
|
||||
"""Build A2A skill descriptors from the real toolset registry.
|
||||
|
||||
Unlike ``skills_from_toolsets`` which takes a list of names, this accepts
|
||||
the actual toolset registry dict (toolset_name → {tools: [...], description: ...})
|
||||
and produces richer skill cards with per-tool descriptions.
|
||||
|
||||
This enables Dynamic Agent Cards: the card we serve reflects what the agent
|
||||
can *actually do* right now, not a static list.
|
||||
"""
|
||||
skills = []
|
||||
if toolset_registry and isinstance(toolset_registry, dict):
|
||||
for ts_name in sorted(toolset_registry.keys()):
|
||||
ts_info = toolset_registry[ts_name] or {}
|
||||
tools_list = ts_info.get("tools", []) if isinstance(ts_info, dict) else []
|
||||
tool_names = [t.get("name", str(t)) if isinstance(t, dict) else str(t) for t in (tools_list or [])]
|
||||
desc = ts_info.get("description", f"Hermes '{ts_name}' capabilities") if isinstance(ts_info, dict) else f"Hermes '{ts_name}' capabilities"
|
||||
skills.append({
|
||||
"id": f"toolset.{ts_name}",
|
||||
"name": ts_name,
|
||||
"description": desc,
|
||||
"tags": [ts_name],
|
||||
# Include tool names as tags for capability matching
|
||||
"tags": [ts_name] + tool_names[:10],
|
||||
})
|
||||
if not skills:
|
||||
skills.append({
|
||||
"id": "general",
|
||||
"name": "general",
|
||||
"description": "General-purpose conversational agent",
|
||||
"tags": ["general"],
|
||||
})
|
||||
return skills
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
# JSON-RPC framing
|
||||
# --------------------------------------------------------------------------
|
||||
|
|
@ -156,17 +210,124 @@ def build_task(task_id: str, context_id: str, state: str, agent_text: str = "")
|
|||
return task
|
||||
|
||||
|
||||
def build_streaming_event(event_type: str, task_id: str, context_id: str, data: dict | None = None) -> str:
|
||||
"""Build a single SSE event for message/stream responses.
|
||||
|
||||
Event types following A2A spec:
|
||||
- ``task``: full task state transition (submitted → working → completed)
|
||||
- ``artifact``: incremental artifact delta
|
||||
- ``status``: status update (state + optional message)
|
||||
- ``done``: stream complete marker
|
||||
"""
|
||||
payload: dict[str, Any] = {"taskId": task_id, "contextId": context_id}
|
||||
if data:
|
||||
payload.update(data)
|
||||
return f"event: {event_type}\ndata: {json.dumps(payload, ensure_ascii=False)}\n\n"
|
||||
|
||||
|
||||
def _now_iso() -> str:
|
||||
return time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
# Anti-loop ping-pong protection
|
||||
# --------------------------------------------------------------------------
|
||||
|
||||
# Track turns per context_id to prevent infinite agent-to-agent loops.
|
||||
# A "turn" is one inbound message/send from a peer. When the count exceeds
|
||||
# max_pingpong_turns(), we reject further messages for that context.
|
||||
# OpenClaw pattern: maxPingPongTurns, default 5, max 20.
|
||||
|
||||
_turn_counts: dict[str, int] = defaultdict(int)
|
||||
_turn_timestamps: dict[str, float] = {}
|
||||
|
||||
# Clean up turn tracking for contexts older than 1 hour.
|
||||
_TURN_TTL = 3600
|
||||
|
||||
|
||||
def track_turn(context_id: str) -> int:
|
||||
"""Increment and return the turn count for this context.
|
||||
|
||||
Returns the *new* count. Caller should reject if > max_pingpong_turns().
|
||||
Also prunes stale entries to prevent unbounded growth.
|
||||
"""
|
||||
now = time.time()
|
||||
# Prune stale entries
|
||||
stale = [cid for cid, ts in _turn_timestamps.items() if now - ts > _TURN_TTL]
|
||||
for cid in stale:
|
||||
_turn_counts.pop(cid, None)
|
||||
_turn_timestamps.pop(cid, None)
|
||||
|
||||
_turn_counts[context_id] += 1
|
||||
_turn_timestamps[context_id] = now
|
||||
return _turn_counts[context_id]
|
||||
|
||||
|
||||
def turn_count(context_id: str) -> int:
|
||||
"""Return current turn count for a context (0 if unknown)."""
|
||||
return _turn_counts.get(context_id, 0)
|
||||
|
||||
|
||||
def reset_turns(context_id: str) -> None:
|
||||
"""Reset turn count for a context (e.g. after explicit cancel)."""
|
||||
_turn_counts.pop(context_id, None)
|
||||
_turn_timestamps.pop(context_id, None)
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
# Metrics collection
|
||||
# --------------------------------------------------------------------------
|
||||
|
||||
# Lightweight in-memory metrics. Not persisted — resets on restart.
|
||||
# For a real deployment, export these via the /metrics endpoint (adapter.py).
|
||||
class Metrics:
|
||||
"""Simple counters for A2A operations."""
|
||||
|
||||
def __init__(self) -> None:
|
||||
self.inbound_total = 0
|
||||
self.outbound_total = 0
|
||||
self.streams_started = 0
|
||||
self.push_sent = 0
|
||||
self.push_failed = 0
|
||||
self.tasks_completed = 0
|
||||
self.tasks_failed = 0
|
||||
self.anti_loop_triggers = 0
|
||||
self.rate_limit_triggers = 0
|
||||
self._start_time = time.time()
|
||||
# Rolling latency tracking (last 100 requests)
|
||||
self._latencies: deque[float] = deque(maxlen=100)
|
||||
|
||||
def record_latency(self, seconds: float) -> None:
|
||||
self._latencies.append(seconds)
|
||||
|
||||
def avg_latency(self) -> float:
|
||||
if not self._latencies:
|
||||
return 0.0
|
||||
return sum(self._latencies) / len(self._latencies)
|
||||
|
||||
def snapshot(self) -> dict[str, Any]:
|
||||
uptime = time.time() - self._start_time
|
||||
return {
|
||||
"uptime_seconds": round(uptime, 1),
|
||||
"inbound_total": self.inbound_total,
|
||||
"outbound_total": self.outbound_total,
|
||||
"streams_started": self.streams_started,
|
||||
"push_sent": self.push_sent,
|
||||
"push_failed": self.push_failed,
|
||||
"tasks_completed": self.tasks_completed,
|
||||
"tasks_failed": self.tasks_failed,
|
||||
"anti_loop_triggers": self.anti_loop_triggers,
|
||||
"rate_limit_triggers": self.rate_limit_triggers,
|
||||
"avg_latency_ms": round(self.avg_latency() * 1000, 1),
|
||||
}
|
||||
|
||||
|
||||
metrics = Metrics()
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
# Conversation persistence (outside the context-compaction pipeline)
|
||||
# --------------------------------------------------------------------------
|
||||
#
|
||||
# A2A exchanges are stored on disk per context-id so they survive context
|
||||
# compaction and agent restarts (the #11025 requirement). One JSONL file per
|
||||
# context; each line is one message {role, text, ts, task_id}.
|
||||
|
||||
def _conv_dir() -> Path:
|
||||
try:
|
||||
|
|
@ -220,3 +381,140 @@ def list_conversations() -> list[str]:
|
|||
if not d.exists():
|
||||
return []
|
||||
return sorted(p.stem for p in d.glob("*.jsonl"))
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
# Rate limiting (token bucket per peer)
|
||||
# --------------------------------------------------------------------------
|
||||
|
||||
# Simple token-bucket rate limiter. Each peer gets a bucket.
|
||||
# Configurable via A2A_RATE_LIMIT (requests per minute, default 60).
|
||||
# OpenClaw pattern: rate limiting per agent identity.
|
||||
|
||||
_RATE_LIMIT_DEFAULT = 60 # requests per minute
|
||||
_rate_buckets: dict[str, deque[float]] = defaultdict(deque)
|
||||
_RATE_WINDOW = 60.0 # seconds
|
||||
|
||||
|
||||
def _rate_limit_per_minute() -> int:
|
||||
try:
|
||||
return max(1, int(os.getenv("A2A_RATE_LIMIT", str(_RATE_LIMIT_DEFAULT))))
|
||||
except (ValueError, TypeError):
|
||||
return _RATE_LIMIT_DEFAULT
|
||||
|
||||
|
||||
def rate_limit_allow(peer: str) -> bool:
|
||||
"""Check if peer is within rate limit. Returns True if allowed."""
|
||||
limit = _rate_limit_per_minute()
|
||||
now = time.time()
|
||||
bucket = _rate_buckets[peer]
|
||||
# Expire old entries
|
||||
while bucket and now - bucket[0] > _RATE_WINDOW:
|
||||
bucket.popleft()
|
||||
if len(bucket) >= limit:
|
||||
return False
|
||||
bucket.append(now)
|
||||
return True
|
||||
|
||||
|
||||
def rate_limit_status(peer: str) -> dict[str, Any]:
|
||||
"""Return rate limit status for a peer."""
|
||||
limit = _rate_limit_per_minute()
|
||||
now = time.time()
|
||||
bucket = _rate_buckets[peer]
|
||||
# Count active entries
|
||||
active = sum(1 for ts in bucket if now - ts <= _RATE_WINDOW)
|
||||
return {
|
||||
"peer": peer,
|
||||
"limit_per_minute": limit,
|
||||
"used": active,
|
||||
"remaining": max(0, limit - active),
|
||||
}
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
# Pending task registry (for async durable messaging)
|
||||
# --------------------------------------------------------------------------
|
||||
|
||||
# Tracks tasks that are in-flight (submitted/working state) so we can
|
||||
# support async completion notifications and orphaned task cleanup.
|
||||
# OpenClaw pattern: sessions_send (async durable messaging).
|
||||
|
||||
_pending_tasks: dict[str, dict[str, Any]] = {}
|
||||
_pending_lock = None # Will be set by adapter on init
|
||||
|
||||
|
||||
def register_pending_task(task_id: str, context_id: str, peer: str, callback_url: str = "") -> None:
|
||||
"""Register a task as pending (in-flight)."""
|
||||
import threading
|
||||
global _pending_lock
|
||||
if _pending_lock is None:
|
||||
_pending_lock = threading.Lock()
|
||||
with _pending_lock:
|
||||
_pending_tasks[task_id] = {
|
||||
"context_id": context_id,
|
||||
"peer": peer,
|
||||
"callback_url": callback_url,
|
||||
"started_at": time.time(),
|
||||
"state": STATE_WORKING,
|
||||
}
|
||||
|
||||
|
||||
def complete_pending_task(task_id: str, state: str, reply: str = "") -> dict | None:
|
||||
"""Mark a pending task as complete. Returns the task info if found."""
|
||||
import threading
|
||||
global _pending_lock
|
||||
if _pending_lock is None:
|
||||
_pending_lock = threading.Lock()
|
||||
with _pending_lock:
|
||||
info = _pending_tasks.pop(task_id, None)
|
||||
if info:
|
||||
info["state"] = state
|
||||
info["reply"] = reply
|
||||
info["completed_at"] = time.time()
|
||||
return info
|
||||
|
||||
|
||||
def pending_task_info(task_id: str) -> dict | None:
|
||||
"""Get info about a pending task (for tasks/get)."""
|
||||
import threading
|
||||
global _pending_lock
|
||||
if _pending_lock is None:
|
||||
_pending_lock = threading.Lock()
|
||||
with _pending_lock:
|
||||
return _pending_tasks.get(task_id)
|
||||
|
||||
|
||||
def orphaned_tasks(timeout_seconds: int = 300) -> list[dict]:
|
||||
"""Find tasks that have been pending longer than timeout.
|
||||
|
||||
Used by the orphaned task watchdog to clean up stale tasks.
|
||||
"""
|
||||
import threading
|
||||
global _pending_lock
|
||||
if _pending_lock is None:
|
||||
_pending_lock = threading.Lock()
|
||||
now = time.time()
|
||||
with _pending_lock:
|
||||
return [
|
||||
{"task_id": tid, **info}
|
||||
for tid, info in _pending_tasks.items()
|
||||
if now - info.get("started_at", now) > timeout_seconds
|
||||
]
|
||||
|
||||
|
||||
def clear_orphaned_tasks(timeout_seconds: int = 300) -> list[str]:
|
||||
"""Remove and return task_ids of tasks pending longer than timeout."""
|
||||
import threading
|
||||
global _pending_lock
|
||||
if _pending_lock is None:
|
||||
_pending_lock = threading.Lock()
|
||||
now = time.time()
|
||||
cleared = []
|
||||
with _pending_lock:
|
||||
for tid in list(_pending_tasks.keys()):
|
||||
info = _pending_tasks[tid]
|
||||
if now - info.get("started_at", now) > timeout_seconds:
|
||||
_pending_tasks.pop(tid, None)
|
||||
cleared.append(tid)
|
||||
return cleared
|
||||
|
|
@ -13,10 +13,14 @@ Layers (all opt-out-able only by explicit config, never silently):
|
|||
inbound task text before it reaches the agent
|
||||
4. Outbound redaction — scrub credential-shaped strings from anything we send
|
||||
5. Audit log — append-only JSONL of every inbound + outbound exchange
|
||||
6. Rate limiting — token-bucket per peer (delegates to protocol.rate_limit_*)
|
||||
7. Trusted peers — explicit allow-list for cross-machine delegation
|
||||
8. Push auth — HMAC-SHA256 webhook signing for push notifications
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import hashlib
|
||||
import hmac
|
||||
import json
|
||||
import logging
|
||||
|
|
@ -84,6 +88,62 @@ def resolve_bind_host() -> str:
|
|||
return requested
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
# Trusted peer approval (Issue #56434)
|
||||
# --------------------------------------------------------------------------
|
||||
|
||||
def get_trusted_peers() -> set[str]:
|
||||
"""Return the set of trusted peer identifiers.
|
||||
|
||||
Trusted peers can send tasks without per-task approval. Configured via
|
||||
A2A_TRUSTED_PEERS env var (comma-separated) or config.yaml under
|
||||
a2a.trusted_peers.
|
||||
|
||||
When A2A_ALLOW_ALL_USERS is set, all peers are trusted (open mode).
|
||||
"""
|
||||
if os.getenv("A2A_ALLOW_ALL_USERS", "").strip().lower() in ("1", "true", "yes"):
|
||||
return set() # empty set signals "all allowed" when checked with is_trusted
|
||||
|
||||
# Check env var
|
||||
env_peers = os.getenv("A2A_TRUSTED_PEERS", "").strip()
|
||||
if env_peers:
|
||||
return {p.strip() for p in env_peers.split(",") if p.strip()}
|
||||
|
||||
# Check config.yaml
|
||||
try:
|
||||
from hermes_cli.config import load_config
|
||||
cfg = load_config() or {}
|
||||
peers_list = (cfg.get("a2a") or {}).get("trusted_peers", [])
|
||||
if isinstance(peers_list, list):
|
||||
return {str(p).strip() for p in peers_list if p}
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
# No trusted peers configured — localhost-only mode trusts all
|
||||
if localhost_only():
|
||||
return set() # will be treated as "all allowed" by is_trusted
|
||||
|
||||
return set()
|
||||
|
||||
|
||||
def is_trusted_peer(peer_id: str) -> bool:
|
||||
"""Check if a peer is trusted (or if all peers are trusted in open mode)."""
|
||||
if os.getenv("A2A_ALLOW_ALL_USERS", "").strip().lower() in ("1", "true", "yes"):
|
||||
return True
|
||||
if localhost_only():
|
||||
return True # localhost-only mode = trust all local peers
|
||||
trusted = get_trusted_peers()
|
||||
return peer_id in trusted
|
||||
|
||||
|
||||
def is_open_mode() -> bool:
|
||||
"""True when all peers are trusted (open mode)."""
|
||||
return (
|
||||
os.getenv("A2A_ALLOW_ALL_USERS", "").strip().lower() in ("1", "true", "yes")
|
||||
or localhost_only()
|
||||
)
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
# Inbound injection filtering
|
||||
# --------------------------------------------------------------------------
|
||||
|
|
@ -179,6 +239,48 @@ def redact_outbound(text: str) -> str:
|
|||
return out
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
# Push notification HMAC signing
|
||||
# --------------------------------------------------------------------------
|
||||
|
||||
def get_push_secret() -> str:
|
||||
"""Return the secret used for HMAC-SHA256 push notification signing.
|
||||
|
||||
Falls back to the bearer token if no dedicated push secret is set.
|
||||
If neither is configured, push notifications are unsigned (localhost-only mode).
|
||||
"""
|
||||
secret = os.getenv("A2A_PUSH_SECRET", "").strip()
|
||||
if secret:
|
||||
return secret
|
||||
return get_bearer_token()
|
||||
|
||||
|
||||
def sign_push_payload(payload: dict) -> str:
|
||||
"""HMAC-SHA256 sign a push notification payload.
|
||||
|
||||
Returns hex-encoded signature. Empty string if no secret configured.
|
||||
"""
|
||||
secret = get_push_secret()
|
||||
if not secret:
|
||||
return ""
|
||||
body = json.dumps(payload, sort_keys=True, ensure_ascii=False).encode("utf-8")
|
||||
return hmac.new(secret.encode("utf-8"), body, hashlib.sha256).hexdigest()
|
||||
|
||||
|
||||
def verify_push_signature(payload: dict, signature: str) -> bool:
|
||||
"""Verify a push notification HMAC signature.
|
||||
|
||||
Returns True if signature matches or no secret configured (localhost mode).
|
||||
"""
|
||||
secret = get_push_secret()
|
||||
if not secret:
|
||||
return True
|
||||
if not signature:
|
||||
return False
|
||||
expected = sign_push_payload(payload)
|
||||
return hmac.compare_digest(signature, expected)
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
# Audit log
|
||||
# --------------------------------------------------------------------------
|
||||
|
|
@ -197,7 +299,7 @@ def audit(direction: str, peer: str, task_id: str, summary: str) -> None:
|
|||
try:
|
||||
rec = {
|
||||
"ts": time.time(),
|
||||
"direction": direction, # "inbound" | "outbound"
|
||||
"direction": direction, # "inbound" | "outbound" | "push"
|
||||
"peer": peer,
|
||||
"task_id": task_id,
|
||||
"summary": (summary or "")[:500],
|
||||
|
|
@ -207,4 +309,4 @@ def audit(direction: str, peer: str, task_id: str, summary: str) -> None:
|
|||
with path.open("a", encoding="utf-8") as fh:
|
||||
fh.write(json.dumps(rec, ensure_ascii=False) + "\n")
|
||||
except Exception:
|
||||
logger.debug("A2A: audit write failed", exc_info=True)
|
||||
logger.debug("A2A: audit write failed", exc_info=True)
|
||||
|
|
@ -2,9 +2,10 @@
|
|||
A2A client tools — let the Hermes agent talk to *other* agents as a peer.
|
||||
|
||||
Tools (registered in the ``a2a`` toolset):
|
||||
- a2a_discover(url) -> fetch + summarize a peer's Agent Card
|
||||
- a2a_call(agent, message) -> send a task to a peer, return its reply
|
||||
- a2a_list() -> list configured peers + persisted conversations
|
||||
- a2a_discover(url) -> fetch + summarize a peer's Agent Card
|
||||
- a2a_call(agent, message) -> send a task to a peer, return its reply
|
||||
- a2a_list() -> list configured peers + persisted conversations
|
||||
- a2a_orchestrate(...) -> fan-out task to multiple peers by capability
|
||||
|
||||
Peers are resolved from config.yaml under ``a2a_agents``::
|
||||
|
||||
|
|
@ -13,6 +14,7 @@ Peers are resolved from config.yaml under ``a2a_agents``::
|
|||
url: "http://localhost:9999"
|
||||
auth: { type: bearer, token: "sk-..." }
|
||||
timeout: 120
|
||||
capabilities: [web_search, research]
|
||||
|
||||
Transport is stdlib urllib (no a2a-sdk dependency). The wire format is the A2A
|
||||
JSON-RPC ``message/send`` method, so any A2A-compliant peer works.
|
||||
|
|
@ -25,6 +27,7 @@ import logging
|
|||
import os
|
||||
import urllib.error
|
||||
import urllib.request
|
||||
from concurrent.futures import ThreadPoolExecutor, as_completed
|
||||
from typing import Any, Optional, TypedDict
|
||||
|
||||
from . import protocol, security
|
||||
|
|
@ -32,6 +35,7 @@ from . import protocol, security
|
|||
logger = logging.getLogger(__name__)
|
||||
|
||||
_DEFAULT_TIMEOUT = 120
|
||||
_ORCHESTRATE_MAX_WORKERS = 6 # max parallel peers for fan-out
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
|
|
@ -47,9 +51,9 @@ def _load_config() -> dict:
|
|||
|
||||
|
||||
def _resolve_peer(agent: str) -> Optional[dict]:
|
||||
"""Resolve a peer name to {url, auth, timeout}, or treat ``agent`` as a URL."""
|
||||
"""Resolve a peer name to {url, auth, timeout, capabilities}, or treat ``agent`` as a URL."""
|
||||
if agent.startswith("http://") or agent.startswith("https://"):
|
||||
return {"url": agent, "auth": {}, "timeout": _DEFAULT_TIMEOUT}
|
||||
return {"url": agent, "auth": {}, "timeout": _DEFAULT_TIMEOUT, "capabilities": []}
|
||||
cfg = _load_config()
|
||||
peers = cfg.get("a2a_agents") or {}
|
||||
entry = peers.get(agent)
|
||||
|
|
@ -59,6 +63,7 @@ def _resolve_peer(agent: str) -> Optional[dict]:
|
|||
"url": entry.get("url", ""),
|
||||
"auth": entry.get("auth", {}) or {},
|
||||
"timeout": int(entry.get("timeout", _DEFAULT_TIMEOUT)),
|
||||
"capabilities": entry.get("capabilities", []) or [],
|
||||
}
|
||||
|
||||
|
||||
|
|
@ -122,7 +127,7 @@ def a2a_discover(args: dict, **_: Any) -> str:
|
|||
f"Agent: {name}",
|
||||
f"Description: {desc}",
|
||||
f"URL: {card.get('url', url)}",
|
||||
f"Streaming: {bool(caps.get('streaming'))} Auth required: {auth}",
|
||||
f"Streaming: {bool(caps.get('streaming'))} Push: {bool(caps.get('pushNotifications'))} Auth required: {auth}",
|
||||
f"Skills ({len(skills)}):",
|
||||
]
|
||||
for s in skills[:20]:
|
||||
|
|
@ -176,12 +181,15 @@ def a2a_call(args: dict, **_: Any) -> str:
|
|||
|
||||
security.audit("outbound", agent, rpc_body["id"], safe_message)
|
||||
protocol.persist_message(ctx, "user", safe_message, rpc_body["id"])
|
||||
protocol.metrics.outbound_total += 1
|
||||
|
||||
try:
|
||||
resp = _http_post_json(_rpc_url(base_url, card), rpc_body, headers, timeout)
|
||||
except urllib.error.HTTPError as e:
|
||||
if e.code in (401, 403):
|
||||
return f"Error: peer '{agent}' rejected auth (HTTP {e.code}). Check the configured token."
|
||||
if e.code == 429:
|
||||
return f"Error: peer '{agent}' rate limited us (HTTP 429). Retry later."
|
||||
return f"Error: call to '{agent}' failed — HTTP {e.code}."
|
||||
except Exception as e:
|
||||
return f"Error: call to '{agent}' failed — {e}."
|
||||
|
|
@ -194,6 +202,7 @@ def a2a_call(args: dict, **_: Any) -> str:
|
|||
reply = _reply_text_from_result(result)
|
||||
reply_ctx = result.get("contextId", ctx) if isinstance(result, dict) else ctx
|
||||
protocol.persist_message(reply_ctx, "agent", reply, rpc_body["id"])
|
||||
protocol.metrics.inbound_total += 1
|
||||
|
||||
state = ""
|
||||
if isinstance(result, dict):
|
||||
|
|
@ -230,7 +239,9 @@ def a2a_list(args: dict | None = None, **_: Any) -> str:
|
|||
lines.append(f"Configured peers ({len(peers)}):")
|
||||
for name, entry in peers.items():
|
||||
auth = (entry.get("auth") or {}).get("type", "none")
|
||||
lines.append(f" - {name}: {entry.get('url', '?')} (auth: {auth})")
|
||||
caps = entry.get("capabilities", [])
|
||||
cap_str = f" caps: {', '.join(caps)}" if caps else ""
|
||||
lines.append(f" - {name}: {entry.get('url', '?')} (auth: {auth}){cap_str}")
|
||||
else:
|
||||
lines.append("No peers configured. Add them under 'a2a_agents' in config.yaml.")
|
||||
|
||||
|
|
@ -240,14 +251,163 @@ def a2a_list(args: dict | None = None, **_: Any) -> str:
|
|||
lines.append(f"Persisted conversations ({len(convos)}):")
|
||||
for c in convos[:25]:
|
||||
lines.append(f" - {c}")
|
||||
|
||||
# Show metrics snapshot
|
||||
m = protocol.metrics.snapshot()
|
||||
lines.append("")
|
||||
lines.append(f"Metrics: {m['inbound_total']} in / {m['outbound_total']} out, "
|
||||
f"{m['tasks_completed']} completed, {m['tasks_failed']} failed, "
|
||||
f"{m['streams_started']} streams, {m['push_sent']} push sent, "
|
||||
f"{m['anti_loop_triggers']} anti-loop, {m['rate_limit_triggers']} rate-limited, "
|
||||
f"avg {m['avg_latency_ms']}ms")
|
||||
|
||||
return "\n".join(lines)
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
# a2a_orchestrate: capability-based routing with fan-out
|
||||
# --------------------------------------------------------------------------
|
||||
|
||||
def _match_peers_by_capability(capability: str) -> list[tuple[str, dict]]:
|
||||
"""Find configured peers that advertise the given capability."""
|
||||
cfg = _load_config()
|
||||
peers = cfg.get("a2a_agents") or {}
|
||||
matches = []
|
||||
for name, entry in peers.items():
|
||||
caps = entry.get("capabilities", []) or []
|
||||
if capability in caps or capability == "*":
|
||||
matches.append((name, entry))
|
||||
return matches
|
||||
|
||||
|
||||
def _call_peer_sync(agent_name: str, peer_entry: dict, message: str, context_id: str = "") -> tuple[str, str]:
|
||||
"""Call a single peer synchronously. Returns (agent_name, reply_text)."""
|
||||
try:
|
||||
base_url = peer_entry.get("url", "")
|
||||
headers = _auth_header(peer_entry.get("auth", {}))
|
||||
timeout = int(peer_entry.get("timeout", _DEFAULT_TIMEOUT))
|
||||
|
||||
card = None
|
||||
try:
|
||||
card = _http_get_json(_card_url(base_url), headers, min(timeout, 30))
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
ctx = context_id or protocol.new_context_id()
|
||||
safe_message = security.redact_outbound(message)
|
||||
rpc_body = {
|
||||
"jsonrpc": "2.0",
|
||||
"id": protocol.new_task_id(),
|
||||
"method": "message/send",
|
||||
"params": {"message": protocol.text_message("user", safe_message)},
|
||||
}
|
||||
if context_id:
|
||||
rpc_body["params"]["contextId"] = context_id
|
||||
rpc_body["params"]["message"]["contextId"] = context_id
|
||||
|
||||
security.audit("outbound", agent_name, rpc_body["id"], safe_message)
|
||||
protocol.persist_message(ctx, "user", safe_message, rpc_body["id"])
|
||||
protocol.metrics.outbound_total += 1
|
||||
|
||||
resp = _http_post_json(_rpc_url(base_url, card), rpc_body, headers, timeout)
|
||||
if "error" in resp:
|
||||
err = resp["error"]
|
||||
return (agent_name, f"Error: {err.get('message', err)}")
|
||||
|
||||
result = resp.get("result", {})
|
||||
reply = _reply_text_from_result(result)
|
||||
reply_ctx = result.get("contextId", ctx) if isinstance(result, dict) else ctx
|
||||
protocol.persist_message(reply_ctx, "agent", reply, rpc_body["id"])
|
||||
protocol.metrics.inbound_total += 1
|
||||
return (agent_name, reply or "(no reply)")
|
||||
except Exception as e:
|
||||
return (agent_name, f"Error: {e}")
|
||||
|
||||
|
||||
def a2a_orchestrate(args: dict, **_: Any) -> str:
|
||||
"""Fan-out a task to multiple peer agents by capability.
|
||||
|
||||
Modes:
|
||||
- ``all``: send to all peers matching the capability, return all replies.
|
||||
- ``first``: send to all matching peers, return the first successful reply.
|
||||
- ``best``: send to all, return the longest/most detailed reply.
|
||||
|
||||
Configured peers advertise capabilities in config.yaml::
|
||||
|
||||
a2a_agents:
|
||||
researcher:
|
||||
url: "http://localhost:9991"
|
||||
capabilities: [web_search, research]
|
||||
coder:
|
||||
url: "http://localhost:9992"
|
||||
capabilities: [code, debug]
|
||||
"""
|
||||
capability = str(args.get("capability") or "").strip()
|
||||
message = str(args.get("message") or args.get("task") or "").strip()
|
||||
mode = str(args.get("mode") or "all").strip().lower()
|
||||
context_id = str(args.get("context_id") or "").strip()
|
||||
|
||||
if not message:
|
||||
return "Error: 'message' is required."
|
||||
if not capability:
|
||||
return "Error: 'capability' is required (or use '*' for all peers)."
|
||||
|
||||
matches = _match_peers_by_capability(capability)
|
||||
if not matches:
|
||||
return f"Error: no configured peers advertise capability '{capability}'."
|
||||
|
||||
if mode not in ("all", "first", "best"):
|
||||
mode = "all"
|
||||
|
||||
# Fan-out
|
||||
results: list[tuple[str, str]] = []
|
||||
with ThreadPoolExecutor(max_workers=min(len(matches), _ORCHESTRATE_MAX_WORKERS)) as pool:
|
||||
futures = {
|
||||
pool.submit(_call_peer_sync, name, entry, message, context_id): name
|
||||
for name, entry in matches
|
||||
}
|
||||
for fut in as_completed(futures):
|
||||
name = futures[fut]
|
||||
try:
|
||||
results.append(fut.result())
|
||||
if mode == "first" and not results[-1][1].startswith("Error:"):
|
||||
# Got a good reply, cancel remaining
|
||||
for f in futures:
|
||||
f.cancel()
|
||||
break
|
||||
except Exception as e:
|
||||
results.append((name, f"Error: {e}"))
|
||||
|
||||
# Sort results by peer name for deterministic output
|
||||
results.sort(key=lambda r: r[0])
|
||||
|
||||
if mode == "best":
|
||||
# Pick the longest non-error reply
|
||||
best = max(results, key=lambda r: len(r[1]) if not r[1].startswith("Error:") else 0)
|
||||
return f"[best: {best[0]}]\n{best[1]}"
|
||||
elif mode == "first":
|
||||
# Return the first non-error reply
|
||||
for name, reply in results:
|
||||
if not reply.startswith("Error:"):
|
||||
return f"[first: {name}]\n{reply}"
|
||||
# All failed
|
||||
lines = ["All peers failed:"]
|
||||
for name, reply in results:
|
||||
lines.append(f" {name}: {reply}")
|
||||
return "\n".join(lines)
|
||||
else: # mode == "all"
|
||||
lines = [f"Orchestrated '{capability}' to {len(matches)} peer(s):"]
|
||||
for name, reply in results:
|
||||
lines.append(f"\n--- {name} ---")
|
||||
lines.append(reply)
|
||||
return "\n".join(lines)
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
# Tool schemas + registration
|
||||
# --------------------------------------------------------------------------
|
||||
|
||||
_FunctionSchema = TypedDict("_FunctionSchema", {"name": str, "description": str}, total=False)
|
||||
_FunctionSchema = TypedDict("_FunctionSchema", {"name": str, "description": str, "parameters": dict[str, Any]}, total=False)
|
||||
_ToolSchema = TypedDict("_ToolSchema", {"type": str, "function": _FunctionSchema}, total=False)
|
||||
_SCHEMAS: dict[str, _ToolSchema] = {
|
||||
"a2a_discover": {
|
||||
|
|
@ -293,21 +453,43 @@ _SCHEMAS: dict[str, _ToolSchema] = {
|
|||
"type": "function",
|
||||
"function": {
|
||||
"name": "a2a_list",
|
||||
"description": "List configured A2A peer agents and persisted A2A conversations.",
|
||||
"description": "List configured A2A peer agents, persisted A2A conversations, and metrics.",
|
||||
"parameters": {"type": "object", "properties": {}},
|
||||
},
|
||||
},
|
||||
"a2a_orchestrate": {
|
||||
"type": "function",
|
||||
"function": {
|
||||
"name": "a2a_orchestrate",
|
||||
"description": (
|
||||
"Fan-out a task to multiple peer agents by capability. Peers are "
|
||||
"matched from config.yaml a2a_agents.*.capabilities. Modes: 'all' "
|
||||
"(return all replies), 'first' (first successful), 'best' (longest reply)."
|
||||
),
|
||||
"parameters": {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"capability": {"type": "string", "description": "Capability to match (e.g. 'research', 'code') or '*' for all peers."},
|
||||
"message": {"type": "string", "description": "The task to send to all matching peers."},
|
||||
"mode": {"type": "string", "enum": ["all", "first", "best"], "description": "How to aggregate results. Default: 'all'."},
|
||||
"context_id": {"type": "string", "description": "Optional: shared context id for all peers."},
|
||||
},
|
||||
"required": ["capability", "message"],
|
||||
},
|
||||
},
|
||||
},
|
||||
}
|
||||
|
||||
_HANDLERS = {
|
||||
"a2a_discover": a2a_discover,
|
||||
"a2a_call": a2a_call,
|
||||
"a2a_list": a2a_list,
|
||||
"a2a_orchestrate": a2a_orchestrate,
|
||||
}
|
||||
|
||||
|
||||
def register_tools(ctx) -> None:
|
||||
"""Register the three client tools in the ``a2a`` toolset."""
|
||||
"""Register the client tools in the ``a2a`` toolset."""
|
||||
for name, schema in _SCHEMAS.items():
|
||||
ctx.register_tool(
|
||||
name=name,
|
||||
|
|
@ -316,4 +498,4 @@ def register_tools(ctx) -> None:
|
|||
handler=_HANDLERS[name],
|
||||
description=schema["function"]["description"],
|
||||
emoji="\U0001f9e9", # puzzle piece
|
||||
)
|
||||
)
|
||||
|
|
@ -0,0 +1,452 @@
|
|||
"""
|
||||
Phase 2 + Phase 3 feature tests for the A2A plugin.
|
||||
|
||||
Tests cover:
|
||||
- SSE streaming event format
|
||||
- Push notification HMAC signing
|
||||
- Anti-loop ping-pong protection
|
||||
- Rate limiting (token bucket per peer)
|
||||
- Metrics collection
|
||||
- Trusted peer approval (#56434)
|
||||
- Pending task registry (async durable messaging)
|
||||
- Dynamic Agent Cards from real toolsets
|
||||
- Capability-based routing with fan-out (a2a_orchestrate)
|
||||
- Task completion notifications (#56435)
|
||||
- Orphaned task watchdog
|
||||
- Metrics endpoint
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import threading
|
||||
import time
|
||||
|
||||
import pytest
|
||||
|
||||
from plugins.platforms.a2a import protocol, security, tools
|
||||
|
||||
|
||||
# ═════════════════════════════════════════════════════════════════════════════
|
||||
# Phase 2: SSE Streaming
|
||||
# ═════════════════════════════════════════════════════════════════════════════
|
||||
|
||||
|
||||
class TestSSEStreaming:
|
||||
"""Tests for message/stream SSE response format."""
|
||||
|
||||
def test_build_streaming_event_format(self):
|
||||
"""SSE events should be properly formatted with event: and data: lines."""
|
||||
event = protocol.build_streaming_event("task", "task-1", "ctx-1", {"status": {"state": "completed"}})
|
||||
assert event.startswith("event: task\n")
|
||||
assert "data: " in event
|
||||
assert event.endswith("\n\n")
|
||||
payload = json.loads(event.split("data: ", 1)[1].strip())
|
||||
assert payload["taskId"] == "task-1"
|
||||
assert payload["contextId"] == "ctx-1"
|
||||
assert payload["status"]["state"] == "completed"
|
||||
|
||||
def test_build_streaming_event_done(self):
|
||||
"""Done events should have no extra data."""
|
||||
event = protocol.build_streaming_event("done", "task-1", "ctx-1")
|
||||
assert "event: done" in event
|
||||
assert event.endswith("\n\n")
|
||||
|
||||
def test_agent_card_advertises_streaming(self):
|
||||
"""Agent Card should now advertise streaming=True capability."""
|
||||
card = protocol.build_agent_card(
|
||||
name="test", url="http://localhost:9900/",
|
||||
description="test", streaming=True, push_notifications=True,
|
||||
)
|
||||
assert card["capabilities"]["streaming"] is True
|
||||
assert card["capabilities"]["pushNotifications"] is True
|
||||
|
||||
|
||||
# ═════════════════════════════════════════════════════════════════════════════
|
||||
# Phase 2: Push Notifications
|
||||
# ═════════════════════════════════════════════════════════════════════════════
|
||||
|
||||
|
||||
class TestPushNotifications:
|
||||
"""Tests for HMAC-SHA256 push notification signing."""
|
||||
|
||||
def test_sign_and_verify_push_payload(self, monkeypatch):
|
||||
"""Sign a payload and verify the signature round-trips."""
|
||||
monkeypatch.setenv("A2A_PUSH_SECRET", "test-secret-123")
|
||||
payload = {"taskId": "task-1", "state": "completed", "reply": "hello"}
|
||||
sig = security.sign_push_payload(payload)
|
||||
assert sig # non-empty when secret set
|
||||
assert security.verify_push_signature(payload, sig) is True
|
||||
|
||||
def test_verify_rejects_wrong_signature(self, monkeypatch):
|
||||
"""Wrong signature should be rejected."""
|
||||
monkeypatch.setenv("A2A_PUSH_SECRET", "test-secret-123")
|
||||
payload = {"taskId": "task-1", "state": "completed"}
|
||||
assert security.verify_push_signature(payload, "wrong-sig") is False
|
||||
|
||||
def test_no_secret_allows_all(self, monkeypatch):
|
||||
"""Without a secret configured, push verification passes (localhost mode)."""
|
||||
monkeypatch.delenv("A2A_PUSH_SECRET", raising=False)
|
||||
monkeypatch.delenv("A2A_BEARER_TOKEN", raising=False)
|
||||
payload = {"taskId": "task-1"}
|
||||
assert security.verify_push_signature(payload, "") is True
|
||||
|
||||
def test_falls_back_to_bearer_token(self, monkeypatch):
|
||||
"""Push secret should fall back to bearer token if not set."""
|
||||
monkeypatch.delenv("A2A_PUSH_SECRET", raising=False)
|
||||
monkeypatch.setenv("A2A_BEARER_TOKEN", "bearer-as-push-secret")
|
||||
payload = {"taskId": "task-1"}
|
||||
sig = security.sign_push_payload(payload)
|
||||
assert sig # should use bearer token as secret
|
||||
assert security.verify_push_signature(payload, sig) is True
|
||||
|
||||
|
||||
# ═════════════════════════════════════════════════════════════════════════════
|
||||
# Phase 3: Anti-loop ping-pong protection
|
||||
# ═════════════════════════════════════════════════════════════════════════════
|
||||
|
||||
|
||||
class TestAntiLoopProtection:
|
||||
"""Tests for ping-pong anti-loop protection."""
|
||||
|
||||
def test_track_turn_increments(self):
|
||||
"""track_turn should increment and return the count."""
|
||||
protocol.reset_turns("test-anti-loop-1")
|
||||
assert protocol.track_turn("test-anti-loop-1") == 1
|
||||
assert protocol.track_turn("test-anti-loop-1") == 2
|
||||
assert protocol.track_turn("test-anti-loop-1") == 3
|
||||
|
||||
def test_turn_count_returns_current(self):
|
||||
"""turn_count should return the current count without incrementing."""
|
||||
protocol.reset_turns("test-anti-loop-2")
|
||||
protocol.track_turn("test-anti-loop-2")
|
||||
protocol.track_turn("test-anti-loop-2")
|
||||
assert protocol.turn_count("test-anti-loop-2") == 2
|
||||
|
||||
def test_reset_turns_clears(self):
|
||||
"""reset_turns should clear the count for a context."""
|
||||
protocol.reset_turns("test-anti-loop-3")
|
||||
for _ in range(5):
|
||||
protocol.track_turn("test-anti-loop-3")
|
||||
assert protocol.turn_count("test-anti-loop-3") == 5
|
||||
protocol.reset_turns("test-anti-loop-3")
|
||||
assert protocol.turn_count("test-anti-loop-3") == 0
|
||||
|
||||
def test_max_pingpong_turns_default(self, monkeypatch):
|
||||
"""Default max should be 5 when env not set."""
|
||||
monkeypatch.delenv("A2A_MAX_PINGPONG_TURNS", raising=False)
|
||||
assert protocol.max_pingpong_turns() == 5
|
||||
|
||||
def test_max_pingpong_turns_env_override(self, monkeypatch):
|
||||
"""Env var should override default, capped at 20."""
|
||||
monkeypatch.setenv("A2A_MAX_PINGPONG_TURNS", "10")
|
||||
assert protocol.max_pingpong_turns() == 10
|
||||
monkeypatch.setenv("A2A_MAX_PINGPONG_TURNS", "50")
|
||||
assert protocol.max_pingpong_turns() == 20 # hard cap
|
||||
monkeypatch.setenv("A2A_MAX_PINGPONG_TURNS", "0")
|
||||
assert protocol.max_pingpong_turns() == 1 # min 1
|
||||
|
||||
|
||||
# ═════════════════════════════════════════════════════════════════════════════
|
||||
# Phase 2: Rate limiting
|
||||
# ═════════════════════════════════════════════════════════════════════════════
|
||||
|
||||
|
||||
class TestRateLimiting:
|
||||
"""Tests for token-bucket rate limiting."""
|
||||
|
||||
def test_rate_limit_allows_under_limit(self, monkeypatch):
|
||||
"""Requests under the limit should be allowed."""
|
||||
monkeypatch.setenv("A2A_RATE_LIMIT", "10")
|
||||
for _ in range(10):
|
||||
assert protocol.rate_limit_allow("test-peer-1") is True
|
||||
|
||||
def test_rate_limit_blocks_over_limit(self, monkeypatch):
|
||||
"""Requests over the limit should be blocked."""
|
||||
monkeypatch.setenv("A2A_RATE_LIMIT", "3")
|
||||
assert protocol.rate_limit_allow("test-peer-2") is True
|
||||
assert protocol.rate_limit_allow("test-peer-2") is True
|
||||
assert protocol.rate_limit_allow("test-peer-2") is True
|
||||
assert protocol.rate_limit_allow("test-peer-2") is False # 4th blocked
|
||||
|
||||
def test_rate_limit_separate_per_peer(self, monkeypatch):
|
||||
"""Different peers should have separate buckets."""
|
||||
monkeypatch.setenv("A2A_RATE_LIMIT", "2")
|
||||
assert protocol.rate_limit_allow("peer-a") is True
|
||||
assert protocol.rate_limit_allow("peer-a") is True
|
||||
assert protocol.rate_limit_allow("peer-a") is False
|
||||
assert protocol.rate_limit_allow("peer-b") is True # different bucket
|
||||
assert protocol.rate_limit_allow("peer-b") is True
|
||||
|
||||
def test_rate_limit_status(self, monkeypatch):
|
||||
"""rate_limit_status should return correct stats."""
|
||||
monkeypatch.setenv("A2A_RATE_LIMIT", "5")
|
||||
for _ in range(3):
|
||||
protocol.rate_limit_allow("test-peer-3")
|
||||
status = protocol.rate_limit_status("test-peer-3")
|
||||
assert status["peer"] == "test-peer-3"
|
||||
assert status["limit_per_minute"] == 5
|
||||
assert status["used"] == 3
|
||||
assert status["remaining"] == 2
|
||||
|
||||
|
||||
# ═════════════════════════════════════════════════════════════════════════════
|
||||
# Phase 2: Metrics
|
||||
# ═════════════════════════════════════════════════════════════════════════════
|
||||
|
||||
|
||||
class TestMetrics:
|
||||
"""Tests for the metrics system."""
|
||||
|
||||
def test_metrics_snapshot_has_fields(self):
|
||||
"""Snapshot should include all expected fields."""
|
||||
m = protocol.metrics.snapshot()
|
||||
assert "uptime_seconds" in m
|
||||
assert "inbound_total" in m
|
||||
assert "outbound_total" in m
|
||||
assert "streams_started" in m
|
||||
assert "push_sent" in m
|
||||
assert "push_failed" in m
|
||||
assert "tasks_completed" in m
|
||||
assert "tasks_failed" in m
|
||||
assert "anti_loop_triggers" in m
|
||||
assert "rate_limit_triggers" in m
|
||||
assert "avg_latency_ms" in m
|
||||
|
||||
def test_metrics_record_latency(self):
|
||||
"""Recording latency should update the average."""
|
||||
protocol.metrics.record_latency(0.1)
|
||||
protocol.metrics.record_latency(0.3)
|
||||
avg = protocol.metrics.avg_latency()
|
||||
assert 0.19 <= avg <= 0.21 # (0.1 + 0.3) / 2 = 0.2
|
||||
|
||||
|
||||
# ═════════════════════════════════════════════════════════════════════════════
|
||||
# Phase 3: Trusted peer approval (#56434)
|
||||
# ═════════════════════════════════════════════════════════════════════════════
|
||||
|
||||
|
||||
class TestTrustedPeers:
|
||||
"""Tests for trusted peer approval."""
|
||||
|
||||
def test_localhost_trusts_all(self, monkeypatch):
|
||||
"""In localhost-only mode, all peers should be trusted."""
|
||||
monkeypatch.delenv("A2A_BEARER_TOKEN", raising=False)
|
||||
monkeypatch.delenv("A2A_ALLOW_ALL_USERS", raising=False)
|
||||
assert security.is_trusted_peer("anyone") is True
|
||||
assert security.is_trusted_peer("random-peer") is True
|
||||
|
||||
def test_allow_all_users_trusts_everyone(self, monkeypatch):
|
||||
"""A2A_ALLOW_ALL_USERS should trust all peers."""
|
||||
monkeypatch.setenv("A2A_BEARER_TOKEN", "secret")
|
||||
monkeypatch.setenv("A2A_ALLOW_ALL_USERS", "true")
|
||||
assert security.is_trusted_peer("anyone") is True
|
||||
|
||||
def test_untrusted_peer_rejected(self, monkeypatch):
|
||||
"""Without allow-all or localhost mode, untrusted peers should be rejected."""
|
||||
monkeypatch.setenv("A2A_BEARER_TOKEN", "secret")
|
||||
monkeypatch.delenv("A2A_ALLOW_ALL_USERS", raising=False)
|
||||
monkeypatch.delenv("A2A_TRUSTED_PEERS", raising=False)
|
||||
assert security.is_trusted_peer("unknown-peer") is False
|
||||
|
||||
def test_trusted_peers_from_env(self, monkeypatch):
|
||||
"""Trusted peers from env should be accepted."""
|
||||
monkeypatch.setenv("A2A_BEARER_TOKEN", "secret")
|
||||
monkeypatch.delenv("A2A_ALLOW_ALL_USERS", raising=False)
|
||||
monkeypatch.setenv("A2A_TRUSTED_PEERS", "alice,bob,carol")
|
||||
assert security.is_trusted_peer("alice") is True
|
||||
assert security.is_trusted_peer("bob") is True
|
||||
assert security.is_trusted_peer("carol") is True
|
||||
assert security.is_trusted_peer("dave") is False
|
||||
|
||||
|
||||
# ═════════════════════════════════════════════════════════════════════════════
|
||||
# Phase 3: Pending task registry (async durable messaging)
|
||||
# ═════════════════════════════════════════════════════════════════════════════
|
||||
|
||||
|
||||
class TestPendingTaskRegistry:
|
||||
"""Tests for pending task tracking."""
|
||||
|
||||
def test_register_and_complete(self):
|
||||
"""Register a task, verify it's pending, then complete it."""
|
||||
protocol.register_pending_task("task-reg-1", "ctx-1", "peer-1")
|
||||
info = protocol.pending_task_info("task-reg-1")
|
||||
assert info is not None
|
||||
assert info["context_id"] == "ctx-1"
|
||||
assert info["peer"] == "peer-1"
|
||||
assert info["state"] == "working"
|
||||
|
||||
completed = protocol.complete_pending_task("task-reg-1", "completed", "reply text")
|
||||
assert completed is not None
|
||||
assert completed["state"] == "completed"
|
||||
assert completed["reply"] == "reply text"
|
||||
assert "completed_at" in completed
|
||||
|
||||
# After completion, should not be in pending
|
||||
assert protocol.pending_task_info("task-reg-1") is None
|
||||
|
||||
def test_orphaned_tasks(self):
|
||||
"""Orphaned tasks should be detected and cleaned up."""
|
||||
protocol.register_pending_task("task-orphan-1", "ctx-2", "peer-2")
|
||||
# Manually age it
|
||||
protocol._pending_tasks["task-orphan-1"]["started_at"] = time.time() - 400
|
||||
|
||||
orphans = protocol.orphaned_tasks(timeout_seconds=300)
|
||||
assert any(o["task_id"] == "task-orphan-1" for o in orphans)
|
||||
|
||||
cleared = protocol.clear_orphaned_tasks(timeout_seconds=300)
|
||||
assert "task-orphan-1" in cleared
|
||||
assert protocol.pending_task_info("task-orphan-1") is None
|
||||
|
||||
|
||||
# ═════════════════════════════════════════════════════════════════════════════
|
||||
# Phase 3: Dynamic Agent Cards
|
||||
# ═════════════════════════════════════════════════════════════════════════════
|
||||
|
||||
|
||||
class TestDynamicAgentCards:
|
||||
"""Tests for dynamic Agent Cards from real toolsets."""
|
||||
|
||||
def test_skills_from_real_toolsets(self):
|
||||
"""skills_from_real_toolsets should build richer skill cards."""
|
||||
registry = {
|
||||
"web": {
|
||||
"description": "Web search and extraction tools",
|
||||
"tools": [{"name": "web_search"}, {"name": "web_extract"}],
|
||||
},
|
||||
"terminal": {
|
||||
"description": "Shell command execution",
|
||||
"tools": [{"name": "terminal"}, {"name": "read_file"}],
|
||||
},
|
||||
}
|
||||
skills = protocol.skills_from_real_toolsets(registry)
|
||||
assert len(skills) == 2
|
||||
names = [s["name"] for s in skills]
|
||||
assert "terminal" in names
|
||||
assert "web" in names
|
||||
# Should include tool names as tags
|
||||
web_skill = [s for s in skills if s["name"] == "web"][0]
|
||||
assert "web_search" in web_skill["tags"]
|
||||
assert "web_extract" in web_skill["tags"]
|
||||
|
||||
def test_skills_from_real_toolsets_empty(self):
|
||||
"""Empty registry should return default general skill."""
|
||||
skills = protocol.skills_from_real_toolsets({})
|
||||
assert len(skills) == 1
|
||||
assert skills[0]["name"] == "general"
|
||||
|
||||
def test_agent_card_version_bumped(self):
|
||||
"""Agent Card version should be bumped for Phase 2+3."""
|
||||
card = protocol.build_agent_card(
|
||||
name="test", url="http://localhost:9900/",
|
||||
description="test", streaming=True, push_notifications=True,
|
||||
)
|
||||
assert card["version"] == "0.2.0"
|
||||
|
||||
|
||||
# ═════════════════════════════════════════════════════════════════════════════
|
||||
# Phase 3: Capability-based routing (a2a_orchestrate)
|
||||
# ═════════════════════════════════════════════════════════════════════════════
|
||||
|
||||
|
||||
class TestA2AOrchestrate:
|
||||
"""Tests for capability-based routing with fan-out."""
|
||||
|
||||
def test_orchestrate_requires_capability(self):
|
||||
"""a2a_orchestrate should require a capability argument."""
|
||||
result = tools.a2a_orchestrate({"message": "do something"})
|
||||
assert "Error" in result
|
||||
assert "capability" in result
|
||||
|
||||
def test_orchestrate_requires_message(self):
|
||||
"""a2a_orchestrate should require a message argument."""
|
||||
result = tools.a2a_orchestrate({"capability": "research"})
|
||||
assert "Error" in result
|
||||
assert "message" in result
|
||||
|
||||
def test_orchestrate_no_matching_peers(self):
|
||||
"""Should report error when no peers match the capability."""
|
||||
from unittest.mock import patch
|
||||
with patch.object(tools, "_load_config", return_value={}):
|
||||
result = tools.a2a_orchestrate({"capability": "research", "message": "search for X"})
|
||||
assert "Error" in result
|
||||
assert "no configured peers" in result
|
||||
|
||||
def test_match_peers_by_capability(self):
|
||||
"""_match_peers_by_capability should find peers with matching caps."""
|
||||
from unittest.mock import patch
|
||||
with patch.object(tools, "_load_config", return_value={
|
||||
"a2a_agents": {
|
||||
"researcher": {
|
||||
"url": "http://localhost:9991",
|
||||
"capabilities": ["research", "web_search"],
|
||||
},
|
||||
"coder": {
|
||||
"url": "http://localhost:9992",
|
||||
"capabilities": ["code", "debug"],
|
||||
},
|
||||
}
|
||||
}):
|
||||
matches = tools._match_peers_by_capability("research")
|
||||
assert len(matches) == 1
|
||||
assert matches[0][0] == "researcher"
|
||||
|
||||
matches = tools._match_peers_by_capability("*")
|
||||
assert len(matches) == 2
|
||||
|
||||
|
||||
# ═════════════════════════════════════════════════════════════════════════════
|
||||
# Phase 3: Task completion notifications (#56435)
|
||||
# ═════════════════════════════════════════════════════════════════════════════
|
||||
|
||||
|
||||
class TestTaskCompletionNotification:
|
||||
"""Tests for task completion notification."""
|
||||
|
||||
def test_build_task_completed_has_reply(self):
|
||||
"""Completed task should include the reply text in status message."""
|
||||
task = protocol.build_task("task-1", "ctx-1", protocol.STATE_COMPLETED, "here is the answer")
|
||||
assert task["status"]["state"] == "completed"
|
||||
assert task["artifacts"][0]["parts"][0]["text"] == "here is the answer"
|
||||
|
||||
def test_build_task_failed_has_message(self):
|
||||
"""Failed task should include the error message."""
|
||||
task = protocol.build_task("task-2", "ctx-2", protocol.STATE_FAILED, "something went wrong")
|
||||
assert task["status"]["state"] == "failed"
|
||||
assert task["status"]["message"]["parts"][0]["text"] == "something went wrong"
|
||||
|
||||
|
||||
# ═════════════════════════════════════════════════════════════════════════════
|
||||
# Phase 2: Metrics endpoint + Watchdog
|
||||
# ═════════════════════════════════════════════════════════════════════════════
|
||||
|
||||
|
||||
class TestMetricsEndpoint:
|
||||
"""Tests for the /metrics endpoint."""
|
||||
|
||||
def test_metrics_endpoint_in_adapter(self):
|
||||
"""The /metrics endpoint should be handled in do_GET."""
|
||||
from plugins.platforms.a2a.adapter import A2AAdapter
|
||||
import inspect
|
||||
source = inspect.getsource(A2AAdapter)
|
||||
assert "/metrics" in source
|
||||
|
||||
|
||||
class TestWatchdog:
|
||||
"""Tests for the orphaned task watchdog."""
|
||||
|
||||
def test_watchdog_thread_started_on_connect(self):
|
||||
"""connect() should start the watchdog thread."""
|
||||
from plugins.platforms.a2a.adapter import A2AAdapter
|
||||
import inspect
|
||||
source = inspect.getsource(A2AAdapter.connect)
|
||||
assert "_watchdog_thread" in source
|
||||
assert "_watchdog_loop" in source
|
||||
|
||||
def test_clear_orphaned_tasks(self):
|
||||
"""clear_orphaned_tasks should remove tasks older than timeout."""
|
||||
protocol.register_pending_task("task-watchdog-1", "ctx-w1", "peer-w1")
|
||||
protocol._pending_tasks["task-watchdog-1"]["started_at"] = time.time() - 600
|
||||
cleared = protocol.clear_orphaned_tasks(timeout_seconds=300)
|
||||
assert "task-watchdog-1" in cleared
|
||||
assert protocol.pending_task_info("task-watchdog-1") is None
|
||||
|
|
@ -586,6 +586,8 @@ class TestContextIdExtraction:
|
|||
adapter._message_handler = None
|
||||
adapter._pending_replies = {}
|
||||
adapter._pending_lock = __import__("threading").Lock()
|
||||
adapter._push_lock = __import__("threading").Lock()
|
||||
adapter._push_callbacks = {}
|
||||
|
||||
params = {
|
||||
"contextId": "ctx-from-caller-T1",
|
||||
|
|
@ -628,6 +630,8 @@ class TestContextIdExtraction:
|
|||
adapter._message_handler = None
|
||||
adapter._pending_replies = {}
|
||||
adapter._pending_lock = __import__("threading").Lock()
|
||||
adapter._push_lock = __import__("threading").Lock()
|
||||
adapter._push_callbacks = {}
|
||||
|
||||
legacy_msg = protocol.text_message("user", "hello")
|
||||
legacy_msg["contextId"] = "ctx-legacy"
|
||||
|
|
@ -662,6 +666,8 @@ class TestContextIdExtraction:
|
|||
adapter._message_handler = None
|
||||
adapter._pending_replies = {}
|
||||
adapter._pending_lock = __import__("threading").Lock()
|
||||
adapter._push_lock = __import__("threading").Lock()
|
||||
adapter._push_callbacks = {}
|
||||
|
||||
params = {"message": protocol.text_message("user", "hello")}
|
||||
adapter._handle_inbound_task(params)
|
||||
|
|
|
|||
Loading…
Reference in New Issue