263 lines
11 KiB
Python
263 lines
11 KiB
Python
"""Bot-relay JSON-RPC handlers — the gateway side of cross-connection A2A.
|
|
|
|
Connections ARE the peer set: every gateway the Desktop holds a socket to
|
|
(local, remote URL, SSH, Hermes Cloud, docker) must be able to find every
|
|
other connection's agents and message them. The Desktop is the relay — it
|
|
owns every socket — and these four methods are the door it uses on EACH
|
|
connected gateway:
|
|
|
|
- ``bot_relay.roster.sync`` — Desktop pushes the union roster of agents on
|
|
the OTHER connections into this gateway's ``bot_relay/roster.json``, so
|
|
``message_agent`` can resolve cross-connection targets and Bot Chat
|
|
prompts list them (capability-epoch refresh picks up changes).
|
|
- ``bot_relay.outbox.drain`` — Desktop collects envelopes queued here by
|
|
``message_agent`` for targets on other connections.
|
|
- ``bot_relay.deliver`` — Desktop hands an envelope to the TARGET
|
|
gateway; this method runs the same one-turn Bot Chat delivery local DMs
|
|
use and returns the reply text.
|
|
- ``bot_relay.reply`` — Desktop writes the reply (or a delivery
|
|
error) back on the SENDER gateway; the waiter spawned at send time picks
|
|
it up and wakes the sending agent via the standard completion path.
|
|
|
|
Storage/validation plumbing lives in ``tools/bot_relay.py``. Handlers are
|
|
rebound onto server.py's globals at install time (see method_ctx.py) and may
|
|
reference server module globals (``_ok``, ``_err``) not imported here.
|
|
"""
|
|
|
|
from .method_ctx import HandlerRegistry
|
|
|
|
_registry = HandlerRegistry()
|
|
method = _registry.method
|
|
|
|
|
|
@method("bot_relay.roster.sync")
|
|
def _(rid, params: dict) -> dict:
|
|
"""Replace this gateway's view of agents on OTHER connections.
|
|
|
|
Params: ``agents`` — list of rows ``{profile, handle, connection_id,
|
|
connection_label?, title?, description?}``. Rows failing validation are
|
|
dropped, not fatal. Result: ``{count}`` (accepted rows).
|
|
"""
|
|
try:
|
|
import os
|
|
from pathlib import Path
|
|
|
|
from tools.bot_relay import write_remote_roster
|
|
|
|
home = Path(os.getenv("HERMES_HOME") or os.path.expanduser("~/.hermes"))
|
|
root = home.parent.parent if home.parent.name == "profiles" else home
|
|
count = write_remote_roster(root, params.get("agents"))
|
|
return _ok(rid, {"count": count})
|
|
except Exception as e:
|
|
return _err(rid, 5090, str(e))
|
|
|
|
|
|
@method("bot_relay.outbox.drain")
|
|
def _(rid, params: dict) -> dict:
|
|
"""Claim every pending cross-connection envelope queued on this gateway.
|
|
|
|
Claimed envelopes move to ``claimed/`` atomically, so concurrent drains
|
|
(two Desktop windows) can't double-deliver. Result: ``{envelopes}``.
|
|
"""
|
|
try:
|
|
import os
|
|
from pathlib import Path
|
|
|
|
from tools.bot_relay import claim_pending_envelopes
|
|
|
|
home = Path(os.getenv("HERMES_HOME") or os.path.expanduser("~/.hermes"))
|
|
root = home.parent.parent if home.parent.name == "profiles" else home
|
|
return _ok(rid, {"envelopes": claim_pending_envelopes(root)})
|
|
except Exception as e:
|
|
return _err(rid, 5091, str(e))
|
|
|
|
|
|
@method("bot_relay.deliver")
|
|
def _(rid, params: dict) -> dict:
|
|
"""Deliver a relayed DM into a profile's Bot Chat ON THIS GATEWAY.
|
|
|
|
Params: ``profile`` (target on this install), ``message`` (already
|
|
attribution-prefixed by the sender gateway). Runs the same one-turn
|
|
``hermes -p <profile> chat -c "Bot Chat"`` transport local DMs use and
|
|
returns ``{reply}`` — the target agent's response text. Blocking by
|
|
design (the Desktop calls it from its relay worker, off any UI path;
|
|
the RPC pool keeps it off the WS reader thread).
|
|
"""
|
|
import os
|
|
import subprocess
|
|
import tempfile
|
|
from pathlib import Path
|
|
|
|
profile = str(params.get("profile") or "").strip()
|
|
message = str(params.get("message") or "").strip()
|
|
if not profile or not message:
|
|
return _err(rid, 4090, "profile and message required")
|
|
try:
|
|
from tools.bot_mode_dm import MESSAGE_MAX_CHARS
|
|
from tools.bot_relay import acquire_turn_lock, local_delivery_command
|
|
|
|
if len(message) > MESSAGE_MAX_CHARS + 200: # + attribution headroom
|
|
return _err(rid, 4091, "message too long")
|
|
|
|
home = Path(os.getenv("HERMES_HOME") or os.path.expanduser("~/.hermes"))
|
|
root = home.parent.parent if home.parent.name == "profiles" else home
|
|
known = {"default"}
|
|
profiles_dir = root / "profiles"
|
|
if profiles_dir.is_dir():
|
|
known.update(c.name for c in profiles_dir.iterdir() if c.is_dir())
|
|
resolved = "default" if profile.lower() == "hermes" else profile
|
|
if resolved not in known:
|
|
return _err(rid, 4092, f"no profile '{profile}' on this gateway")
|
|
|
|
# #100523: when THIS gateway already hosts the target's Bot Chat live
|
|
# (the Desktop has it open), the subprocess transport is fenced out by
|
|
# the single-owner lease ("already has a live owner") and the payload
|
|
# is dropped. Land the DM in the live session as a normal user turn
|
|
# via prompt.submit instead — same choke point the composer uses, so
|
|
# role alternation, persistence and streaming all behave as a typed
|
|
# message would. (Nested per method_ctx rebinding.)
|
|
def _live_bot_chat_sid(profile_name: str) -> str:
|
|
from tools.bot_mode_probe import BOT_CHAT_TITLE
|
|
|
|
live_home = _profile_home(profile_name)
|
|
want_home = str(live_home) if live_home is not None else None
|
|
for live_sid, record in list(_sessions.items()):
|
|
if not isinstance(record, dict):
|
|
continue
|
|
if (record.get("profile_home") or None) != want_home:
|
|
continue
|
|
key = _session_lookup_key(record, fallback=live_sid)
|
|
if _session_live_title(record, key) == BOT_CHAT_TITLE:
|
|
return live_sid
|
|
return ""
|
|
|
|
live_sid = _live_bot_chat_sid(resolved)
|
|
if live_sid:
|
|
# queued=True: a teammate's DM runs as the NEXT turn. It must never
|
|
# interrupt or steer a turn already in flight (the default busy
|
|
# mode does); hundreds of arrivals simply queue in arrival order.
|
|
submitted = _methods["prompt.submit"](rid, {"session_id": live_sid, "text": message, "queued": True})
|
|
if "error" in submitted:
|
|
return submitted
|
|
return _ok(
|
|
rid,
|
|
{"reply": f"Delivered into @{resolved}'s open Bot Chat; the reply will appear there."},
|
|
)
|
|
|
|
fd, tmp = tempfile.mkstemp(prefix="hermes-relay-dm-", suffix=".txt", text=True)
|
|
try:
|
|
with os.fdopen(fd, "w", encoding="utf-8") as f:
|
|
f.write(message)
|
|
# Per-profile turn lock (#93091): serialize with any other
|
|
# delivery turn into this profile (relay or local message_agent).
|
|
# The lock covers only the turn execution window. Worst-case
|
|
# handler hold is lock wait (bot_mode.turn_wait_seconds, default
|
|
# 120s) + the 600s turn timeout below — doubled when the retry
|
|
# policy grants one bounded re-run — so clients calling
|
|
# bot_relay.deliver must tolerate ~1320s before assuming failure.
|
|
with acquire_turn_lock(root, resolved):
|
|
proc = subprocess.run(
|
|
local_delivery_command(resolved, tmp),
|
|
capture_output=True,
|
|
text=True,
|
|
encoding="utf-8",
|
|
errors="replace",
|
|
timeout=600,
|
|
)
|
|
if proc.returncode != 0:
|
|
# Retry session policy (#93091 item 5): transient classes
|
|
# re-run the SAME session once; context_overflow also
|
|
# re-runs the same session — the retried turn's pre-API
|
|
# compaction pass (agent/conversation_loop.py) compacts
|
|
# the over-threshold Bot Chat transcript first, which is
|
|
# the sanctioned compression lever (no fresh session is
|
|
# ever minted). Auth/quota/config classes never retry.
|
|
from tools.bot_failure_reasons import (
|
|
RETRY_NONE,
|
|
classify_agent_error,
|
|
retry_action,
|
|
)
|
|
|
|
first_detail = (proc.stderr or proc.stdout or "").strip()[-500:]
|
|
if retry_action(classify_agent_error(first_detail)) != RETRY_NONE:
|
|
proc = subprocess.run(
|
|
local_delivery_command(resolved, tmp),
|
|
capture_output=True,
|
|
text=True,
|
|
encoding="utf-8",
|
|
errors="replace",
|
|
timeout=600,
|
|
)
|
|
finally:
|
|
try:
|
|
os.unlink(tmp)
|
|
except OSError:
|
|
pass
|
|
if proc.returncode != 0:
|
|
from tools.bot_failure_reasons import classify_agent_error
|
|
|
|
detail = (proc.stderr or proc.stdout or "").strip()[-500:]
|
|
return _err(
|
|
rid,
|
|
5092,
|
|
f"delivery turn failed: {detail or proc.returncode}",
|
|
data={"reason": classify_agent_error(detail)},
|
|
)
|
|
return _ok(rid, {"reply": (proc.stdout or "").strip()})
|
|
except subprocess.TimeoutExpired:
|
|
return _err(rid, 5093, "delivery turn timed out")
|
|
except Exception as e:
|
|
# 'target_busy' extends the #93091 item-1 structured refusal enum.
|
|
if getattr(e, "reason", "") == "target_busy":
|
|
return _err(rid, 5096, str(e))
|
|
return _err(rid, 5094, str(e))
|
|
|
|
|
|
@method("bot_relay.reply")
|
|
def _(rid, params: dict) -> dict:
|
|
"""Write a relayed reply (or delivery error) for a sender-side waiter.
|
|
|
|
Params: ``id`` (envelope id), ``reply`` and/or ``error``, optional
|
|
``reason`` (typed failure code, see ``tools.bot_failure_reasons``).
|
|
"""
|
|
envelope_id = str(params.get("id") or "").strip()
|
|
if not envelope_id:
|
|
return _err(rid, 4093, "id required")
|
|
try:
|
|
import os
|
|
from pathlib import Path
|
|
|
|
from tools.bot_relay import write_reply
|
|
|
|
home = Path(os.getenv("HERMES_HOME") or os.path.expanduser("~/.hermes"))
|
|
root = home.parent.parent if home.parent.name == "profiles" else home
|
|
write_reply(
|
|
root,
|
|
envelope_id,
|
|
reply=str(params.get("reply") or ""),
|
|
error=str(params.get("error") or ""),
|
|
reason=str(params.get("reason") or ""),
|
|
)
|
|
return _ok(rid, {"ok": True})
|
|
except ValueError as e:
|
|
return _err(rid, 4094, str(e))
|
|
except Exception as e:
|
|
return _err(rid, 5095, str(e))
|
|
|
|
|
|
def register(server) -> None:
|
|
_registry.install(server)
|
|
from . import methods_groups
|
|
|
|
server._LONG_HANDLERS = server._LONG_HANDLERS | methods_groups.LONG_HANDLERS
|
|
server.get_hosted_room_service = methods_groups.get_hosted_room_service
|
|
server._WORKER_UNAVAILABLE = methods_groups._WORKER_UNAVAILABLE
|
|
server._profile_name = methods_groups._profile_name
|
|
server._requested_profile = methods_groups._requested_profile
|
|
server._api_server_key = methods_groups._api_server_key
|
|
server._room_link_run_storage_durable = (
|
|
methods_groups._room_link_run_storage_durable
|
|
)
|
|
methods_groups.bind_server(server)
|
|
methods_groups.register(server)
|