diff --git a/gateway/platforms/api_server.py b/gateway/platforms/api_server.py index 30ed5a81081b5..1957d80edf046 100644 --- a/gateway/platforms/api_server.py +++ b/gateway/platforms/api_server.py @@ -1440,6 +1440,15 @@ def __init__(self, config: PlatformConfig): self._stopping_run_ids: set[str] = set() # Pollable run status for dashboards and external control-plane UIs. self._run_statuses: Dict[str, Dict[str, Any]] = {} + # Idempotent run admission: caller key -> request fingerprint + run id. + # This is process-local like run status itself and is swept with the + # corresponding terminal status record. + self._run_idempotency: Dict[str, Dict[str, Any]] = {} + # Admission for /v1/runs is serialized so the idempotency check and + # registration are atomic across the session-history reload. Without it, + # two same-key requests arriving in that await window both pass the + # empty check and both start a run (two handles for one operation). + self._run_idempotency_lock: "asyncio.Lock" = asyncio.Lock() # Active approval session key for each run_id. The approval core # resolves requests by session key, while API clients address the # in-flight run by run_id. @@ -6686,12 +6695,6 @@ async def _handle_runs(self, request: "web.Request") -> "web.Response": if key_err is not None: return key_err - # Enforce concurrency limit (shared across all agent-serving - # endpoints; configurable via gateway.api_server.max_concurrent_runs). - limited = self._concurrency_limited_response() - if limited is not None: - return limited - try: body = await request.json() except Exception: @@ -6765,52 +6768,126 @@ async def _handle_runs(self, request: "web.Request") -> "web.Response": if selection_error: return web.json_response(_openai_error(selection_error), status=400) - run_id = f"run_{uuid.uuid4().hex}" - session_id = session_id or run_id - # Approval queues gate host-side tool execution and must be isolated - # per API run. Client-provided session IDs and memory session keys are - # conversation/memory scopes, not authorization namespaces: multiple - # concurrent runs can intentionally share them, and resolving an - # approval for one run must not unblock another run's dangerous command. - approval_session_key = run_id - ephemeral_system_prompt = instructions - loop = asyncio.get_running_loop() - q: "asyncio.Queue[Optional[Dict]]" = asyncio.Queue() - created_at = time.time() - self._run_streams[run_id] = q - self._run_streams_created[run_id] = created_at - self._run_approval_sessions[run_id] = approval_session_key - - event_cb = self._make_run_event_callback(run_id, loop) - - def _put_event_if_active(event: Optional[Dict]) -> None: - """Enqueue only while this run still owns live transport state.""" - if self._run_streams.get(run_id) is q: - q.put_nowait(event) - - # Also wire stream_delta_callback so message.delta events flow through. - def _text_cb(delta: Optional[str]) -> None: - if delta is None: - return - if run_id not in self._run_streams: - return - try: - loop.call_soon_threadsafe(_put_event_if_active, { - "event": "message.delta", - "run_id": run_id, - "timestamp": time.time(), - "delta": delta, - }) - except Exception: - pass + # A replay must resolve even while the original run consumes the last + # concurrency slot. Check idempotency before enforcing the limit so a + # caller recovering from a lost 202 response receives the original + # run id instead of an unrelated 429. + idempotency_key = request.headers.get("Idempotency-Key", "").strip() + if len(idempotency_key) > 255 or re.search(r"[\r\n\x00]", idempotency_key): + return web.json_response( + _openai_error( + "Idempotency-Key must be at most 255 characters and contain no control newlines", + code="invalid_idempotency_key", + ), + status=400, + ) + async with self._run_idempotency_lock: + idempotency_fingerprint = "" + if idempotency_key: + idempotency_fingerprint = _make_request_fingerprint( + body, + [ + "input", + "session_id", + "instructions", + "conversation_history", + "previous_response_id", + "model", + "provider", + "model_options", + ], + ) + previous = self._run_idempotency.get(idempotency_key) + if previous is not None: + if previous.get("fingerprint") != idempotency_fingerprint: + return web.json_response( + _openai_error( + "Idempotency-Key was already used with a different request", + code="idempotency_conflict", + ), + status=409, + ) + previous_run_id = str(previous.get("run_id") or "") + previous_status = self._run_statuses.get(previous_run_id) + if previous_status is not None: + return web.json_response( + { + "run_id": previous_run_id, + "status": previous_status.get("status", "queued"), + "replayed": True, + }, + status=202, + ) + # Status retention is the public lifetime of a run handle. + # If it has expired, let the key start a fresh run. + self._run_idempotency.pop(idempotency_key, None) + + # Enforce concurrency only for a genuinely new run. + limited = self._concurrency_limited_response() + if limited is not None: + return limited + + # Native session callers expect continuity from session_id alone. + # Explicit history and previous_response_id still win. The durable + # turn lease reloads once more after a contended wait, closing the + # race where another process appends after this snapshot. + if not conversation_history and session_id and not previous_response_id: + conversation_history = await self._conversation_history_for_session( + str(session_id) + ) - self._set_run_status( - run_id, - "queued", - created_at=created_at, - session_id=session_id, - model=body.get("model", self._model_name), - ) + run_id = f"run_{uuid.uuid4().hex}" + session_id = session_id or run_id + # Approval queues gate host-side tool execution and must be isolated + # per API run. Client-provided session IDs and memory session keys are + # conversation/memory scopes, not authorization namespaces: multiple + # concurrent runs can intentionally share them, and resolving an + # approval for one run must not unblock another run's dangerous command. + approval_session_key = run_id + ephemeral_system_prompt = instructions + loop = asyncio.get_running_loop() + q: "asyncio.Queue[Optional[Dict]]" = asyncio.Queue() + created_at = time.time() + self._run_streams[run_id] = q + self._run_streams_created[run_id] = created_at + self._run_approval_sessions[run_id] = approval_session_key + + event_cb = self._make_run_event_callback(run_id, loop) + + def _put_event_if_active(event: Optional[Dict]) -> None: + """Enqueue only while this run still owns live transport state.""" + if self._run_streams.get(run_id) is q: + q.put_nowait(event) + + # Also wire stream_delta_callback so message.delta events flow through. + def _text_cb(delta: Optional[str]) -> None: + if delta is None: + return + if run_id not in self._run_streams: + return + try: + loop.call_soon_threadsafe(_put_event_if_active, { + "event": "message.delta", + "run_id": run_id, + "timestamp": time.time(), + "delta": delta, + }) + except Exception: + pass + + self._set_run_status( + run_id, + "queued", + created_at=created_at, + session_id=session_id, + model=body.get("model", self._model_name), + ) + if idempotency_key: + self._run_idempotency[idempotency_key] = { + "fingerprint": idempotency_fingerprint, + "run_id": run_id, + "created_at": created_at, + } # Background task outlives the HTTP response (and thus the middleware # profile scope). Capture now and re-enter inside the task/executor. @@ -7089,7 +7166,7 @@ def _run_sync(): {"X-Hermes-Session-Key": gateway_session_key} if gateway_session_key else {} ) return web.json_response( - {"run_id": run_id, "status": "started"}, + {"run_id": run_id, "status": "started", "replayed": False}, status=202, headers=response_headers, ) @@ -7388,6 +7465,10 @@ def _sweep_orphaned_runs_once(self, now: Optional[float] = None) -> None: ] for run_id in stale_statuses: self._run_statuses.pop(run_id, None) + live_run_ids = set(self._run_statuses) + for key, record in list(self._run_idempotency.items()): + if record.get("run_id") not in live_run_ids: + self._run_idempotency.pop(key, None) # ------------------------------------------------------------------ # BasePlatformAdapter interface diff --git a/hermes_cli/subcommands/peer.py b/hermes_cli/subcommands/peer.py index 13d10637a6a0e..387ef75c64f1d 100644 --- a/hermes_cli/subcommands/peer.py +++ b/hermes_cli/subcommands/peer.py @@ -7,6 +7,8 @@ hermes peer add spark --url http://spark.lan:8377 --key hermes peer dm spark "Message from 🤖 dixie (@dixie): disk status?" hermes peer dm spark/researcher "..." # named profile (multiplexed peer) + hermes peer run spark --idempotency-key ticket-123 < /tmp/long-task.txt + hermes peer status spark run_abc123 ``dm`` resolves the remote agent's canonical "Bot Chat" session (by title, creating it when missing), runs ONE synchronous agent turn over the peer's @@ -15,6 +17,10 @@ ``hermes -p chat --in ~ -c "Bot Chat" ...`` bot-messaging command, so the Bot Mode protocol composes over it unchanged. +``run`` starts the same canonical-session turn through the asynchronous Runs +API and returns a ``run_id`` immediately. ``status`` polls that handle without +holding the original HTTP connection open. Use this pair for long turns. + Design notes: - No new server surface: the peer's stock api_server is the transport. - Peer labels/URLs live in config.yaml (``bot_peers``); the peer's @@ -33,6 +39,7 @@ import urllib.error import urllib.parse import urllib.request +import uuid BOT_CHAT_TITLE = "Bot Chat" _PEER_NAME_RE = re.compile(r"^[a-z0-9][a-z0-9_-]{0,63}$") @@ -76,17 +83,28 @@ def _peer_secret(name: str) -> str: return (os.environ.get(env_name) or "").strip() -def _request(url: str, key: str, *, method: str = "GET", body: dict | None = None, timeout: int = LIST_TIMEOUT_S) -> dict: +def _request( + url: str, + key: str, + *, + method: str = "GET", + body: dict | None = None, + timeout: int = LIST_TIMEOUT_S, + headers: dict[str, str] | None = None, +) -> dict: data = json.dumps(body).encode("utf-8") if body is not None else None + request_headers = { + "Authorization": f"Bearer {key}", + "Content-Type": "application/json", + "User-Agent": "hermes-peer-dm", + } + if headers: + request_headers.update(headers) req = urllib.request.Request( url, data=data, method=method, - headers={ - "Authorization": f"Bearer {key}", - "Content-Type": "application/json", - "User-Agent": "hermes-peer-dm", - }, + headers=request_headers, ) with urllib.request.urlopen(req, timeout=timeout) as resp: # noqa: S310 — user-registered peer URL payload = resp.read().decode("utf-8", "replace") @@ -157,6 +175,28 @@ def _http_error_detail(exc: urllib.error.HTTPError) -> str: return str(exc) +def _resolve_peer_target(target: str) -> tuple[str, str | None, dict, str]: + """Resolve a registered target to ``(name, profile, config, key)``.""" + peer_name, profile = _parse_target(target) + peer = _load_peers().get(peer_name) + if not isinstance(peer, dict) or not peer.get("url"): + raise LookupError(f"No peer named '{peer_name}'. Run: hermes peer list") + key = _peer_secret(peer_name) + if not key: + raise PermissionError( + f"No API key for peer '{peer_name}'. Set it: hermes peer add {peer_name} " + f"--url --key (or add {_peer_key_env(peer_name)}= to ~/.hermes/.env)" + ) + return peer_name, profile, peer, key + + +def _message_from_args(args) -> str: + message = (getattr(args, "message", None) or "").strip() + if not message and not sys.stdin.isatty(): + message = sys.stdin.read().strip() + return message + + def cmd_peer(args) -> int: action = getattr(args, "peer_action", None) @@ -209,33 +249,109 @@ def cmd_peer(args) -> int: print(f"{name}\t{entry.get('url', '?')}\t[{has_key}]{note}") return 0 - if action == "dm": + if action in {"dm", "run", "status"}: try: - peer_name, profile = _parse_target(args.target) + peer_name, profile, peer, key = _resolve_peer_target(args.target) except ValueError as exc: print(str(exc), file=sys.stderr) return 2 - peers = _load_peers() - peer = peers.get(peer_name) - if not isinstance(peer, dict) or not peer.get("url"): - print(f"No peer named '{peer_name}'. Run: hermes peer list", file=sys.stderr) - return 1 - key = _peer_secret(peer_name) - if not key: - print( - f"No API key for peer '{peer_name}'. Set it: hermes peer add {peer_name} " - f"--url --key (or add {_peer_key_env(peer_name)}= to ~/.hermes/.env)", - file=sys.stderr, - ) + except (LookupError, PermissionError) as exc: + print(str(exc), file=sys.stderr) return 1 - message = (args.message or "").strip() - if not message and not sys.stdin.isatty(): - message = sys.stdin.read().strip() + + base = _base_url(peer, profile) + + if action == "status": + run_id = (getattr(args, "run_id", None) or "").strip() + if not run_id: + print("Run ID required.", file=sys.stderr) + return 2 + try: + result = _request( + f"{base}/v1/runs/{urllib.parse.quote(run_id, safe='')}", + key, + ) + except urllib.error.HTTPError as exc: + print( + f"Peer '{peer_name}' rejected the request (HTTP {exc.code}): {_http_error_detail(exc)}", + file=sys.stderr, + ) + return 1 + except (urllib.error.URLError, TimeoutError, OSError, RuntimeError) as exc: + print(f"Could not reach peer '{peer_name}': {exc}", file=sys.stderr) + return 1 + + payload = {"peer": peer_name, "profile": profile, **result} + if getattr(args, "json", False): + print(json.dumps(payload)) + else: + print(f"{run_id}: {result.get('status', 'unknown')}") + if result.get("output"): + print(result["output"]) + elif result.get("error"): + print(result["error"], file=sys.stderr) + return 0 + + message = _message_from_args(args) if not message: print("Message required (argument or stdin).", file=sys.stderr) return 2 - base = _base_url(peer, profile) + if action == "run": + idempotency_key = ( + getattr(args, "idempotency_key", None) or f"peer-{uuid.uuid4().hex}" + ).strip() + if ( + not idempotency_key + or len(idempotency_key) > 255 + or re.search(r"[\r\n\x00]", idempotency_key) + ): + print( + "Idempotency key must be 1-255 characters without control newlines.", + file=sys.stderr, + ) + return 2 + try: + session_id = _ensure_bot_chat(base, key) + result = _request( + f"{base}/v1/runs", + key, + method="POST", + body={"input": message, "session_id": session_id}, + headers={"Idempotency-Key": idempotency_key}, + ) + except urllib.error.HTTPError as exc: + print( + f"Peer '{peer_name}' rejected the request (HTTP {exc.code}): {_http_error_detail(exc)}", + file=sys.stderr, + ) + return 1 + except (urllib.error.URLError, TimeoutError, OSError, RuntimeError) as exc: + print(f"Could not reach peer '{peer_name}': {exc}", file=sys.stderr) + return 1 + + run_id = str(result.get("run_id") or "") + if not run_id: + print(f"Peer '{peer_name}' did not return a run ID.", file=sys.stderr) + return 1 + payload = { + "peer": peer_name, + "profile": profile, + "session_id": session_id, + "run_id": run_id, + "status": result.get("status") or "started", + "idempotency_key": idempotency_key, + "replayed": bool(result.get("replayed", False)), + } + if getattr(args, "json", False): + print(json.dumps(payload)) + else: + replay = " (replayed)" if payload["replayed"] else "" + print(f"{run_id}: {payload['status']}{replay}") + print(f"session_id: {session_id}") + print(f"idempotency_key: {idempotency_key}") + return 0 + try: session_id = _ensure_bot_chat(base, key) result = _request( @@ -285,6 +401,8 @@ def build_peer_parser(subparsers) -> None: " hermes peer list\n" ' hermes peer dm spark "Message from 🤖 dixie (@dixie): disk status?"\n' ' hermes peer dm spark/researcher "..." # named profile on a multiplexed peer\n' + " hermes peer run spark --idempotency-key ticket-123 < long-task.txt\n" + " hermes peer status spark run_abc123\n" " hermes peer remove spark\n" "\n" "Exit codes: 0 ok, 1 delivery/peer error, 2 usage error." @@ -308,8 +426,45 @@ def build_peer_parser(subparsers) -> None: "dm", help="Message an agent on a peer gateway and print its reply", ) - dm_p.add_argument("target", help=" or / (named profile on a multiplexed peer)") - dm_p.add_argument("message", nargs="?", default=None, help="Message text (or stdin)") - dm_p.add_argument("--json", action="store_true", default=False, help="Emit a JSON result") + dm_p.add_argument( + "target", help=" or / (named profile on a multiplexed peer)" + ) + dm_p.add_argument( + "message", nargs="?", default=None, help="Message text (or stdin)" + ) + dm_p.add_argument( + "--json", action="store_true", default=False, help="Emit a JSON result" + ) + + run_p = peer_sub.add_parser( + "run", + help="Start a long peer turn asynchronously and return its run ID", + ) + run_p.add_argument( + "target", help=" or / (named profile on a multiplexed peer)" + ) + run_p.add_argument( + "message", nargs="?", default=None, help="Message text (or stdin)" + ) + run_p.add_argument( + "--idempotency-key", + default=None, + help="Stable retry key (generated when omitted)", + ) + run_p.add_argument( + "--json", action="store_true", default=False, help="Emit a JSON result" + ) + + status_p = peer_sub.add_parser( + "status", + help="Read the status and final output of an asynchronous peer run", + ) + status_p.add_argument( + "target", help=" or / (named profile on a multiplexed peer)" + ) + status_p.add_argument("run_id", help="Run ID returned by 'hermes peer run'") + status_p.add_argument( + "--json", action="store_true", default=False, help="Emit a JSON result" + ) parser.set_defaults(func=cmd_peer) diff --git a/tests/gateway/test_api_server_runs.py b/tests/gateway/test_api_server_runs.py index d3edc9d151277..aaf76b4ecdff3 100644 --- a/tests/gateway/test_api_server_runs.py +++ b/tests/gateway/test_api_server_runs.py @@ -12,7 +12,7 @@ import asyncio import threading import time -from unittest.mock import MagicMock, patch +from unittest.mock import AsyncMock, MagicMock, patch import pytest from aiohttp import web @@ -147,6 +147,163 @@ async def test_start_returns_202(self, adapter): assert status["status"] in {"queued", "running", "completed"} assert status["object"] == "hermes.run" + @pytest.mark.asyncio + async def test_start_replays_same_idempotency_key_without_second_run(self, adapter): + app = _create_runs_app(adapter) + async with TestClient(TestServer(app)) as cli: + with patch.object(adapter, "_create_agent") as mock_create: + mock_agent = MagicMock() + mock_agent.run_conversation.return_value = {"final_response": "done"} + mock_agent.session_prompt_tokens = 0 + mock_agent.session_completion_tokens = 0 + mock_agent.session_total_tokens = 0 + mock_create.return_value = mock_agent + headers = {"Idempotency-Key": "peer-ticket-123"} + + first = await cli.post( + "/v1/runs", json={"input": "hello"}, headers=headers + ) + second = await cli.post( + "/v1/runs", json={"input": "hello"}, headers=headers + ) + first_data = await first.json() + second_data = await second.json() + + assert first.status == second.status == 202 + assert second_data["run_id"] == first_data["run_id"] + assert second_data["replayed"] is True + + for _ in range(40): + status_resp = await cli.get(f"/v1/runs/{first_data['run_id']}") + status = await status_resp.json() + if status["status"] == "completed": + break + await asyncio.sleep(0.05) + + mock_create.assert_called_once() + + @pytest.mark.asyncio + async def test_start_serializes_concurrent_same_idempotency_key(self, adapter): + """Concurrent requests with the same Idempotency-Key must not both start a run. + + The idempotency check and registration are separated by an ``await`` + (the session-history reload). Two same-key requests arriving in that + window would both pass the empty check and both create a run, leaving + two handles for one logical operation. The admission must serialize so + the loser resolves to the winner's run_id and no duplicate agent runs. + """ + app = _create_runs_app(adapter) + async with TestClient(TestServer(app)) as cli: + + async def _slow_history(session_id): + await asyncio.sleep(0.05) + return [] + + with ( + patch.object(adapter, "_conversation_history_for_session", new=_slow_history), + patch.object(adapter, "_create_agent") as mock_create, + ): + mock_agent = MagicMock() + mock_agent.run_conversation.return_value = {"final_response": "done"} + mock_agent.session_prompt_tokens = 0 + mock_agent.session_completion_tokens = 0 + mock_agent.session_total_tokens = 0 + mock_create.return_value = mock_agent + headers = {"Idempotency-Key": "peer-concurrent-777"} + + async def _post(): + return await cli.post( + "/v1/runs", + json={"input": "hello", "session_id": "bot-chat"}, + headers=headers, + ) + + first_resp, second_resp = await asyncio.gather(_post(), _post()) + first_data = await first_resp.json() + second_data = await second_resp.json() + + assert first_resp.status == second_resp.status == 202 + assert second_data["run_id"] == first_data["run_id"] + assert second_data["replayed"] is True + + for _ in range(40): + status_resp = await cli.get(f"/v1/runs/{first_data['run_id']}") + status = await status_resp.json() + if status["status"] == "completed": + break + await asyncio.sleep(0.05) + + mock_create.assert_called_once() + + @pytest.mark.asyncio + async def test_start_rejects_idempotency_key_reuse_for_different_request( + self, adapter + ): + app = _create_runs_app(adapter) + async with TestClient(TestServer(app)) as cli: + with patch.object(adapter, "_create_agent") as mock_create: + mock_agent = MagicMock() + mock_agent.run_conversation.return_value = {"final_response": "done"} + mock_agent.session_prompt_tokens = 0 + mock_agent.session_completion_tokens = 0 + mock_agent.session_total_tokens = 0 + mock_create.return_value = mock_agent + headers = {"Idempotency-Key": "peer-ticket-123"} + + first = await cli.post( + "/v1/runs", json={"input": "hello"}, headers=headers + ) + second = await cli.post( + "/v1/runs", json={"input": "different"}, headers=headers + ) + second_data = await second.json() + + assert first.status == 202 + assert second.status == 409 + assert second_data["error"]["code"] == "idempotency_conflict" + + @pytest.mark.asyncio + async def test_start_loads_history_for_existing_session_id(self, adapter): + app = _create_runs_app(adapter) + history = [ + {"role": "user", "content": "earlier"}, + {"role": "assistant", "content": "context"}, + ] + load_history = AsyncMock(return_value=history) + async with TestClient(TestServer(app)) as cli: + with ( + patch.object( + adapter, + "_conversation_history_for_session", + new=load_history, + ), + patch.object(adapter, "_create_agent") as mock_create, + ): + mock_agent = MagicMock() + mock_agent.run_conversation.return_value = {"final_response": "done"} + mock_agent.session_prompt_tokens = 0 + mock_agent.session_completion_tokens = 0 + mock_agent.session_total_tokens = 0 + mock_create.return_value = mock_agent + + resp = await cli.post( + "/v1/runs", + json={"input": "continue", "session_id": "bot-chat"}, + ) + data = await resp.json() + for _ in range(40): + status_resp = await cli.get(f"/v1/runs/{data['run_id']}") + status = await status_resp.json() + if status["status"] == "completed": + break + await asyncio.sleep(0.05) + + load_history.assert_awaited_once_with("bot-chat") + assert ( + mock_agent.run_conversation.call_args.kwargs["conversation_history"] + == history + ) + @pytest.mark.asyncio async def test_start_binds_chat_id_for_delegation_wake_target(self, adapter): """/v1/runs must bind the raw session id as the api_server chat_id diff --git a/tests/hermes_cli/test_peer_cmd.py b/tests/hermes_cli/test_peer_cmd.py index a3e4d15ae37e3..4c1f9d031f573 100644 --- a/tests/hermes_cli/test_peer_cmd.py +++ b/tests/hermes_cli/test_peer_cmd.py @@ -92,6 +92,8 @@ def test_dm_unknown_peer_and_missing_key(monkeypatch): class _FakePeer(BaseHTTPRequestHandler): sessions: list = [] chats: list = [] + runs: list = [] + run_idempotency_keys: list = [] auth_seen: list = [] def _json(self, payload, status=200): @@ -104,6 +106,14 @@ def _json(self, payload, status=200): def do_GET(self): type(self).auth_seen.append(self.headers.get("Authorization", "")) + if self.path == "/v1/runs/run_1": + return self._json({ + "object": "hermes.run", + "run_id": "run_1", + "status": "completed", + "session_id": "bc_existing", + "output": "async reply from the other machine", + }) if self.path.startswith("/api/sessions"): data = [{"id": s, "title": "Bot Chat"} for s in type(self).sessions] return self._json({"object": "list", "data": data}) @@ -122,12 +132,23 @@ def do_POST(self): if self.path.startswith("/api/sessions/") and self.path.endswith("/chat"): type(self).chats.append(body.get("message")) + return self._json({ + "object": "hermes.session.chat.completion", + "session_id": "bc_1", + "message": { + "role": "assistant", + "content": "reply from the other machine", + }, + }) + + if self.path == "/v1/runs": + type(self).runs.append(body) + type(self).run_idempotency_keys.append( + self.headers.get("Idempotency-Key", "") + ) return self._json( - { - "object": "hermes.session.chat.completion", - "session_id": "bc_1", - "message": {"role": "assistant", "content": "reply from the other machine"}, - } + {"run_id": "run_1", "status": "started", "replayed": False}, + 202, ) return self._json({"error": {"message": "not found"}}, 404) @@ -140,6 +161,8 @@ def log_message(self, *args): # noqa: D102 — silence test server logging def fake_peer_server(): _FakePeer.sessions = [] _FakePeer.chats = [] + _FakePeer.runs = [] + _FakePeer.run_idempotency_keys = [] _FakePeer.auth_seen = [] server = HTTPServer(("127.0.0.1", 0), _FakePeer) thread = threading.Thread(target=server.serve_forever, daemon=True) @@ -186,3 +209,58 @@ def test_dm_reuses_existing_bot_chat(monkeypatch, capsys, fake_peer_server): assert payload["reply"] == "reply from the other machine" # No new session was created — the existing canonical chat was reused. assert _FakePeer.sessions == ["bc_existing"] + + +def test_run_starts_async_turn_with_canonical_session_and_idempotency( + monkeypatch, capsys, fake_peer_server +): + _FakePeer.sessions = ["bc_existing"] + monkeypatch.setattr( + peer_cmd, "_load_peers", lambda: {"spark": {"url": fake_peer_server}} + ) + monkeypatch.setattr(peer_cmd, "_peer_secret", lambda name: "secret-key-123456") + + rc = peer_cmd.cmd_peer( + SimpleNamespace( + peer_action="run", + target="spark", + message="long task", + idempotency_key="ticket-123", + json=True, + ) + ) + + assert rc == 0 + payload = json.loads(capsys.readouterr().out) + assert payload == { + "peer": "spark", + "profile": None, + "session_id": "bc_existing", + "run_id": "run_1", + "status": "started", + "idempotency_key": "ticket-123", + "replayed": False, + } + assert _FakePeer.runs == [{"input": "long task", "session_id": "bc_existing"}] + assert _FakePeer.run_idempotency_keys == ["ticket-123"] + + +def test_status_reads_async_run_output(monkeypatch, capsys, fake_peer_server): + monkeypatch.setattr( + peer_cmd, "_load_peers", lambda: {"spark": {"url": fake_peer_server}} + ) + monkeypatch.setattr(peer_cmd, "_peer_secret", lambda name: "secret-key-123456") + + rc = peer_cmd.cmd_peer( + SimpleNamespace( + peer_action="status", + target="spark", + run_id="run_1", + json=True, + ) + ) + + assert rc == 0 + payload = json.loads(capsys.readouterr().out) + assert payload["status"] == "completed" + assert payload["output"] == "async reply from the other machine" diff --git a/website/docs/reference/cli-commands.md b/website/docs/reference/cli-commands.md index 7d434df6dbfeb..82b8cc723678f 100644 --- a/website/docs/reference/cli-commands.md +++ b/website/docs/reference/cli-commands.md @@ -447,6 +447,8 @@ hermes send --list telegram # filter by platform hermes peer add --url http://host:port --key hermes peer list hermes peer dm [/] "message" +hermes peer run [/] --idempotency-key "message" +hermes peer status [/] hermes peer remove ``` @@ -466,6 +468,8 @@ its `/p//` mirror). | `add --url [--key ] [--note TEXT]` | Register or update a peer. The URL goes to `config.yaml` (`bot_peers`); the key is stored as `HERMES_PEER__KEY` in `~/.hermes/.env`. | | `list` | List peers and whether each has a key configured. | | `dm [/] [message]` | Message the peer agent's canonical Bot Chat and print the reply (`--json` for machine-readable output; message falls back to stdin). | +| `run [/] [message]` | Start a long canonical Bot Chat turn asynchronously and return its `run_id`, session ID, and idempotency key (`--json` supported). Reuse `--idempotency-key` when retrying the same request. | +| `status [/] ` | Poll an asynchronous peer run and print its final output when complete (`--json` supported). | | `remove ` | Remove a peer from the registry (the `.env` key entry is left in place). | When at least one peer is registered, the Bot Mode messaging protocol diff --git a/website/docs/user-guide/bot-mode.md b/website/docs/user-guide/bot-mode.md index 8b46b208add5f..7072bba3db258 100644 --- a/website/docs/user-guide/bot-mode.md +++ b/website/docs/user-guide/bot-mode.md @@ -117,10 +117,18 @@ hermes peer add spark --url http://spark.lan:8377 --key hermes peer list hermes peer dm spark < /tmp/dm.txt # message body from a file (nothing shell-interpreted) hermes peer dm spark/researcher < /tmp/dm.txt # named profile on a multiplexed peer +hermes peer run spark --idempotency-key ticket-123 < /tmp/long-task.txt +hermes peer status spark run_abc123 ``` `hermes peer dm` delivers into the remote agent's canonical Bot Chat over the peer's existing API server, runs one agent turn there, and prints the reply on stdout — the exact cross-machine twin of the local `hermes -p chat` command. +Use `peer dm` only for short queries and receipts because it holds one HTTP +connection until the turn finishes. For a long turn, `peer run` returns a +`run_id` immediately; poll it with `peer status`. The run inherits the +canonical Bot Chat transcript, and a stable `--idempotency-key` makes a retry +return the original run instead of starting duplicate work. + Once a peer is registered, the messaging protocol taught to every Bot Chat (`agent.bot_mode_protocol`) automatically includes the peer roster and the `hermes peer dm` pattern — so **your bots learn on their own** that teammates exist on other machines and how to reach them. Registering or removing a peer refreshes each Bot Chat's protocol on its next message (capability epoch). Requirements: the peer machine runs the `api_server` gateway platform with a strong `API_SERVER_KEY`; reachability is your network's business (LAN, Tailscale, VPN). The key is a credential and lives in `~/.hermes/.env` as `HERMES_PEER__KEY`; peer names/URLs live in `config.yaml` under `bot_peers`. diff --git a/website/docs/user-guide/features/api-server.md b/website/docs/user-guide/features/api-server.md index ccba76e104d1c..29ef618019236 100644 --- a/website/docs/user-guide/features/api-server.md +++ b/website/docs/user-guide/features/api-server.md @@ -355,6 +355,17 @@ Create a new agent run. Returns a `run_id` that can be used to subscribe to prog Runs accept a simple `input` string and optional `session_id`, `instructions`, `conversation_history`, or `previous_response_id`. When `session_id` is provided, Hermes surfaces it in the run status so external UIs can correlate runs with their own conversation IDs. +When `session_id` identifies an existing Hermes session and no explicit +`conversation_history` or `previous_response_id` is supplied, the run loads +that session's active transcript. Session turn leases serialize concurrent +writers and refresh the transcript after a contended wait. + +Clients may send an `Idempotency-Key` header when starting a run. Repeating +the same request with the same key returns the original `run_id`; reusing the +key with a different request returns `409 idempotency_conflict`. Keys and run +statuses share the same process-local retention window, so durable workflows +should still store the returned `run_id` or use an external task record. + ### GET /v1/runs/\{run_id\} Poll the current run state. This is useful for dashboards that need status without holding an SSE connection open, or for UIs that reconnect after navigation.