fix(a2a): preserve async delegation and message idempotency

[tuantran1231]

What does this PR do?

Related Issue

Fixes #

Type of Change

• 🐛 Bug fix (non-breaking change that fixes an issue)
• ✨ New feature (non-breaking change that adds functionality)
• 🔒 Security fix
• 📝 Documentation update
• ✅ Tests (adding or improving test coverage)
• ♻️ Refactor (no behavior change)
• 🎯 New skill (bundled or hub)

Changes Made

How to Test

Checklist

Code

• I've read the Contributing Guide
• My commit messages follow Conventional Commits (fix(scope):, feat(scope):, etc.)
• I searched for existing PRs to make sure this isn't a duplicate
• My PR contains only changes related to this fix/feature (no unrelated commits)
• I've run pytest tests/ -q and all tests pass
• I've added tests for my changes (required for bug fixes, strongly encouraged for features)
• I've tested on my platform:

Documentation & Housekeeping

• I've updated relevant documentation (README, docs/, docstrings) — or N/A
• I've updated cli-config.yaml.example if I added/changed config keys — or N/A
• I've updated CONTRIBUTING.md or AGENTS.md if I changed architecture or workflows — or N/A
• I've considered cross-platform impact (Windows, macOS) per the compatibility guide — or N/A
• I've updated tool descriptions/schemas if I changed tool behavior — or N/A

For New Skills

• This skill is broadly useful to most users (if bundled) — see Contributing Guide
• SKILL.md follows the standard format (frontmatter, trigger conditions, steps, pitfalls)
• No external dependencies that aren't already available (prefer stdlib, curl, existing Hermes tools)
• I've tested the skill end-to-end: hermes --toolsets skills -q "Use the X skill to do Y"

Screenshots / Logs

[Enough1122]

AI code review — automated review for reference; please use your judgment.

Two unrelated changes in one PR: (a) forwarded/cross-profile tasks run off the HTTP worker and message/send returns WORKING immediately (plugins/platforms/a2a/adapter.py — new _forward_in_background at :637, the async _rpc_message_send path at :704-726, reworked forwarded branch at :566-586); (b) a messageId idempotency lookup so a retried message returns the existing task instead of dispatching twice (adapter.py:535-546, protocol.py:185-193 and get_by_message_id at :382). The TaskStore/set_state guard (rec["state"] not in TERMINAL_STATES, protocol.py:333) and _finalize_task's finally: self._pop_pending make the async plumbing consistent, and the state machine transitions themselves look right. Two issues, both about the change being only half-applied.

1. message/send is now async for every agent, but this repo's own outbound client cannot consume that — adapter.py:704-726 returns STATE_WORKING with an empty reply as soon as the task is scheduled, for the local path as well as the forwarded one (main blocked in _await_reply and returned the finished Task). The Hermes A2A client in the same plugin reads the answer straight out of the message/send response and never polls tasks/get: tools.py:121-132 (_reply_text_from_result walks artifacts then status.message), used by tools.py:180-190 (a2a_call) and tools.py:242-249 (_call_peer_sync). Against a peer running this build, a2a_call now answers [peer · context X · working] + (no text reply), and a2a_orchestrate's first/best modes treat that empty body as a success (tools.py:282-287 only reject replies starting with Error: and otherwise rank by length). The PR updated the server-side tests to poll (tests/plugins/test_a2a_phase23.py:532, tests/plugins/test_a2a_plugin.py:1406) but no client-side test covers the pairing. Either teach _send_task to poll tasks/get on a non-terminal state (bounded by the peer timeout, then report the task id), or scope the async return to the forwarded branch and keep the blocking reply for agent.get("local", True). As it stands the change breaks Hermes-to-Hermes delegation for every agent, not just the forwarded ones it was written for. (blocker)
2. A2A_PEER_TOKENS separator flipped from : to = with no migration and stale docs — plugins/platforms/a2a/security.py:35 now splits on =, so an alice:tok1 value — the format every other surface still produces and documents — no longer yields any pair (":" never satisfies if "=" in pair). _parse_peer_tokens then returns {}, authenticate() finds no peer match and returns None (401 for every peer), and with no token of any kind the server refuses to widen off loopback (security.py:81). Still advertising the old format after this PR: plugins/platforms/a2a/plugin.yaml:43-45 ("'alice:tok1,bob:tok2'", prompt "name:token, comma-separated"), the interactive setup wizard at plugins/platforms/a2a/init.py:65-69 (writes A2A_PEER_TOKENS in name:token form), README.md:60 and :76, DESIGN.md:99, and get_peer_tokens' own docstring at security.py:236. Accept both separators for a deprecation window (e.g. split on = when present, else on :), or update the wizard, plugin.yaml prompt, both docs and the docstring in the same PR — otherwise every existing deployment silently drops peer identity on upgrade. (blocker)

Minor: the idempotency check is a check-then-act across two calls — get_by_message_id at adapter.py:540 and tasks.create at :548 both take self.tasks._lock separately, so two concurrent retries carrying the same messageId can both miss and create two tasks; a create-time uniqueness check inside the store lock (or setdefault on a message_id index) would close it. Minor: the idempotent return path skips security.audit("inbound", ...) and _register_inline_push (adapter.py:546-559), so a retried message leaves no audit line and cannot re-register a push config.