From efa134b0e355698dd1636acced4f032c1d8e80dd Mon Sep 17 00:00:00 2001 From: Ryan Monroe <36248507+rm0nroe@users.noreply.github.com> Date: Mon, 10 Aug 2026 10:48:41 -0400 Subject: [PATCH] fix(gateway): make API runs durable and idempotent --- gateway/platforms/api_server.py | 432 ++++++++++++++++++++--- gateway/run_ledger.py | 299 ++++++++++++++++ tests/gateway/test_api_server.py | 21 +- tests/gateway/test_api_server_runs.py | 485 +++++++++++++++++++++++++- 4 files changed, 1188 insertions(+), 49 deletions(-) create mode 100644 gateway/run_ledger.py diff --git a/gateway/platforms/api_server.py b/gateway/platforms/api_server.py index 6799003ec8d95..e07785f6acbde 100644 --- a/gateway/platforms/api_server.py +++ b/gateway/platforms/api_server.py @@ -1427,6 +1427,12 @@ def __init__(self, config: PlatformConfig): # Active run agent/task references for stop support self._active_run_agents: Dict[str, Any] = {} self._active_run_tasks: Dict[str, "asyncio.Task"] = {} + # In-memory control objects are process-global; bind each one to the + # profile-local ledger that authorized its creation. + self._run_scopes: Dict[str, str] = {} + self._pending_run_ids: set[str] = set() + self._run_ledger_lock: Optional[asyncio.Lock] = None + self._run_status_lock = threading.RLock() # Stop is cooperative: the executor thread may outlive the HTTP request. self._stopping_run_ids: set[str] = set() # Pollable run status for dashboards and external control-plane UIs. @@ -2080,6 +2086,7 @@ def _http_route_table(self) -> List[tuple]: ("POST", "/api/jobs/{job_id}/resume", self._handle_resume_job), ("POST", "/api/jobs/{job_id}/run", self._handle_run_job), ("POST", "/v1/runs", self._handle_runs), + ("GET", "/v1/runs", self._handle_lookup_run), ("GET", "/v1/runs/{run_id}", self._handle_get_run), ("GET", "/v1/runs/{run_id}/events", self._handle_run_events), ("POST", "/v1/runs/{run_id}/approval", self._handle_run_approval), @@ -3092,6 +3099,8 @@ async def _handle_capabilities(self, request: "web.Request") -> "web.Response": if auth_err: return auth_err + from gateway.run_ledger import RETENTION_SECONDS + return web.json_response({ "object": "hermes.api_server.capabilities", "platform": "hermes-agent", @@ -3119,6 +3128,11 @@ async def _handle_capabilities(self, request: "web.Request") -> "web.Response": "run_status": True, "run_events_sse": True, "run_stop": True, + "run_idempotency": True, + "run_correlation_lookup": True, + "run_status_durable": True, + "run_stop_idempotent": True, + "run_status_retention_seconds": RETENTION_SECONDS, "run_approval_response": True, "tool_progress_events": True, "approval_events": True, @@ -3146,6 +3160,11 @@ async def _handle_capabilities(self, request: "web.Request") -> "web.Response": "chat_completions": {"method": "POST", "path": "/v1/chat/completions"}, "responses": {"method": "POST", "path": "/v1/responses"}, "runs": {"method": "POST", "path": "/v1/runs"}, + "run_lookup": { + "method": "GET", + "path": "/v1/runs", + "correlation_header": "Idempotency-Key", + }, "run_status": {"method": "GET", "path": "/v1/runs/{run_id}"}, "run_events": {"method": "GET", "path": "/v1/runs/{run_id}/events"}, "run_approval": {"method": "POST", "path": "/v1/runs/{run_id}/approval"}, @@ -6343,21 +6362,131 @@ def _run(): _RUN_STREAM_TTL = 300 # seconds before orphaned runs are swept _RUN_STATUS_TTL = 3600 # seconds to retain terminal run status for polling - def _set_run_status(self, run_id: str, status: str, **fields: Any) -> Dict[str, Any]: + @staticmethod + def _current_run_scope() -> str: + from hermes_constants import get_hermes_home + + return str(get_hermes_home().resolve()) + + def _active_run_ids_for_scope(self) -> set[str]: + scope = self._current_run_scope() + return { + run_id + for run_id in self._run_scopes + if self._run_scopes.get(run_id) == scope + and ( + run_id in self._active_run_agents + or run_id in self._pending_run_ids + or ( + run_id in self._active_run_tasks + and not self._active_run_tasks[run_id].done() + ) + ) + } + + def _update_run_status_memory( + self, run_id: str, status: str, **fields: Any + ) -> Dict[str, Any]: """Update pollable run status without exposing private agent objects.""" + from gateway.run_ledger import transition_allowed + + fields = dict(fields) now = time.time() - current = self._run_statuses.get(run_id, {}) - current.update({ - "object": "hermes.run", - "run_id": run_id, - "status": status, - "updated_at": now, - }) - current.setdefault("created_at", fields.pop("created_at", now)) - current.update(fields) - self._run_statuses[run_id] = current + with self._run_status_lock: + current = dict(self._run_statuses.get(run_id, {})) + if not transition_allowed(current.get("status"), status): + return current + current.update({ + "object": "hermes.run", + "run_id": run_id, + "status": status, + "updated_at": now, + }) + current.setdefault("created_at", fields.pop("created_at", now)) + current.update(fields) + self._run_statuses[run_id] = current + return current + + def _store_persisted_run_status( + self, run_id: str, persisted: Dict[str, Any] + ) -> Dict[str, Any]: + from gateway.run_ledger import transition_allowed + + with self._run_status_lock: + current = self._run_statuses.get(run_id, {}) + if not transition_allowed(current.get("status"), persisted.get("status")): + return current + self._run_statuses[run_id] = persisted + return persisted + + def _set_run_status(self, run_id: str, status: str, **fields: Any) -> Dict[str, Any]: + """Persist status from worker callbacks without breaking execution.""" + current = self._update_run_status_memory(run_id, status, **fields) + from gateway.run_ledger import update_run + + try: + persisted = update_run(run_id, status, **fields) + except Exception: + logger.exception("[api_server] durable status update failed for %s", run_id) + return current + if persisted is not None: + stored = self._store_persisted_run_status(run_id, persisted) + # Stop is an acceptance receipt. Preserve that response snapshot + # when the executor races to a terminal state during persistence. + if status != "stopping" or current.get("status") != "stopping": + current = stored return current + async def _set_run_status_async( + self, run_id: str, status: str, **fields: Any + ) -> Dict[str, Any]: + """Persist status without blocking the aiohttp event loop.""" + current = self._update_run_status_memory(run_id, status, **fields) + from gateway.run_ledger import update_run + + try: + persisted = await asyncio.to_thread(update_run, run_id, status, **fields) + except Exception: + logger.exception("[api_server] durable status update failed for %s", run_id) + return current + if persisted is not None: + stored = self._store_persisted_run_status(run_id, persisted) + if status != "stopping" or current.get("status") != "stopping": + current = stored + return current + + def _clear_run_memory(self, run_id: str) -> None: + task = self._active_run_tasks.get(run_id) + if task is not None and not task.done(): + return + self._run_statuses.pop(run_id, None) + self._run_streams.pop(run_id, None) + self._run_streams_created.pop(run_id, None) + self._run_approval_sessions.pop(run_id, None) + self._stopping_run_ids.discard(run_id) + self._run_scopes.pop(run_id, None) + self._pending_run_ids.discard(run_id) + + def _get_run_ledger_lock(self) -> asyncio.Lock: + if self._run_ledger_lock is None: + self._run_ledger_lock = asyncio.Lock() + return self._run_ledger_lock + + async def _maintain_run_ledger_locked(self) -> None: + """Reconcile ownership while reservation is excluded.""" + from gateway.run_ledger import purge_terminal_runs, recover_interrupted_runs + + active = self._active_run_ids_for_scope() + recovered = await asyncio.to_thread(recover_interrupted_runs, active) + purged = await asyncio.to_thread(purge_terminal_runs) + for run_id in (*recovered, *purged): + self._clear_run_memory(run_id) + + async def _maintain_run_ledger(self) -> None: + """Reconcile adapter ownership and enforce durable retention.""" + async with self._get_run_ledger_lock(): + await self._maintain_run_ledger_locked() + def _make_run_event_callback(self, run_id: str, loop: "asyncio.AbstractEventLoop"): """Return a tool_progress_callback that pushes structured events to the run's SSE queue.""" def _push(event: Dict[str, Any]) -> None: @@ -6538,6 +6667,32 @@ async def _handle_runs(self, request: "web.Request") -> "web.Response": if selection_error: return web.json_response(_openai_error(selection_error), status=400) + idempotency_key = request.headers.get("Idempotency-Key", "").strip() or None + if idempotency_key is not None and len(idempotency_key) > 255: + return web.json_response( + _openai_error( + "Idempotency-Key must be 255 characters or fewer.", + code="invalid_idempotency_key", + ), + status=400, + ) + + fingerprint = None + if idempotency_key is not None: + fingerprint_payload = { + "body": body, + "gateway_session_key": gateway_session_key, + } + fingerprint = hashlib.sha256( + json.dumps( + fingerprint_payload, + sort_keys=True, + separators=(",", ":"), + ensure_ascii=False, + default=str, + ).encode("utf-8") + ).hexdigest() + 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 @@ -6547,9 +6702,60 @@ async def _handle_runs(self, request: "web.Request") -> "web.Response": # approval for one run must not unblock another run's dangerous command. approval_session_key = run_id ephemeral_system_prompt = instructions + try: + from gateway.run_ledger import ( + IdempotencyConflictError, + reserve_run, + ) + + async with self._get_run_ledger_lock(): + await self._maintain_run_ledger_locked() + durable_status, created = await asyncio.to_thread( + reserve_run, + run_id=run_id, + idempotency_key=idempotency_key, + request_fingerprint=fingerprint, + data={ + "session_id": session_id, + "model": body.get("model", self._model_name), + }, + ) + if created: + self._run_scopes[run_id] = self._current_run_scope() + self._pending_run_ids.add(run_id) + except IdempotencyConflictError: + return web.json_response( + _openai_error( + "Idempotency-Key was already used for a different run request.", + code="idempotency_conflict", + ), + status=409, + ) + except Exception: + logger.exception("[api_server] durable run reservation failed") + return web.json_response( + _openai_error( + "Durable run storage is unavailable; no run was started.", + err_type="server_error", + code="run_storage_unavailable", + ), + status=503, + ) + + self._run_statuses[durable_status["run_id"]] = durable_status + response_headers = ( + {"X-Hermes-Session-Key": gateway_session_key} + if gateway_session_key + else {} + ) + if not created: + return web.json_response( + durable_status, status=202, headers=response_headers + ) + + created_at = float(durable_status["created_at"]) 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 @@ -6577,28 +6783,20 @@ def _text_cb(delta: Optional[str]) -> None: 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), - ) - # Background task outlives the HTTP response (and thus the middleware # profile scope). Capture now and re-enter inside the task/executor. request_profile = _api_request_profile.get() async def _run_and_close(): try: - self._set_run_status(run_id, "running") + await self._set_run_status_async(run_id, "running") if run_id in self._stopping_run_ids: _put_event_if_active({ "event": "run.cancelled", "run_id": run_id, "timestamp": time.time(), }) - self._set_run_status( + await self._set_run_status_async( run_id, "cancelled", last_event="run.cancelled", @@ -6723,7 +6921,7 @@ def _run_sync(): "run_id": run_id, "timestamp": time.time(), }) - self._set_run_status( + await self._set_run_status_async( run_id, "cancelled", last_event="run.cancelled", @@ -6739,7 +6937,7 @@ def _run_sync(): "timestamp": time.time(), "error": error_msg, }) - self._set_run_status( + await self._set_run_status_async( run_id, "failed", error=error_msg, @@ -6754,7 +6952,7 @@ def _run_sync(): "output": final_response, "usage": usage, }) - self._set_run_status( + await self._set_run_status_async( run_id, "completed", output=final_response, @@ -6762,7 +6960,7 @@ def _run_sync(): last_event="run.completed", ) except asyncio.CancelledError: - self._set_run_status( + await self._set_run_status_async( run_id, "cancelled", last_event="run.cancelled", @@ -6786,7 +6984,7 @@ def _run_sync(): # except-Exception branch below. logger.warning("Provider authentication failed for run=%s: %s", run_id, exc) error_msg = f"⚠️ Provider authentication failed: {exc}" - self._set_run_status( + await self._set_run_status_async( run_id, "failed", error=error_msg, @@ -6803,7 +7001,7 @@ def _run_sync(): pass except Exception as exc: logger.exception("[api_server] run %s failed", run_id) - self._set_run_status( + await self._set_run_status_async( run_id, "failed", error=_redact_api_error_text(exc), @@ -6843,6 +7041,7 @@ def _run_sync(): self._activate_admitted_request() task = asyncio.create_task(_run_and_close()) self._active_run_tasks[run_id] = task + self._pending_run_ids.discard(run_id) try: self._background_tasks.add(task) except TypeError: @@ -6850,14 +7049,51 @@ def _run_sync(): if hasattr(task, "add_done_callback"): task.add_done_callback(self._background_tasks.discard) - response_headers = ( - {"X-Hermes-Session-Key": gateway_session_key} if gateway_session_key else {} - ) - return web.json_response( - {"run_id": run_id, "status": "started"}, - status=202, - headers=response_headers, - ) + response_status = dict(durable_status) + response_status["status"] = "started" + return web.json_response(response_status, status=202, headers=response_headers) + + async def _handle_lookup_run(self, request: "web.Request") -> "web.Response": + """GET /v1/runs — resolve one run by its caller correlation key.""" + auth_err = self._check_auth(request) + if auth_err: + return auth_err + + idempotency_key = request.headers.get("Idempotency-Key", "").strip() + if not idempotency_key: + return web.json_response( + _openai_error( + "Idempotency-Key is required for run lookup.", + code="idempotency_key_required", + ), + status=400, + ) + try: + from gateway.run_ledger import ( + get_run_by_idempotency_key, + ) + + await self._maintain_run_ledger() + status = await asyncio.to_thread( + get_run_by_idempotency_key, idempotency_key + ) + except Exception: + logger.exception("[api_server] durable run lookup failed") + return web.json_response( + _openai_error( + "Durable run storage is unavailable.", + err_type="server_error", + code="run_storage_unavailable", + ), + status=503, + ) + if status is None: + return web.json_response( + _openai_error("Run not found", code="run_not_found"), + status=404, + ) + self._run_statuses[status["run_id"]] = status + return web.json_response(status) async def _handle_get_run(self, request: "web.Request") -> "web.Response": """GET /v1/runs/{run_id} — return pollable run status for external UIs.""" @@ -6866,12 +7102,27 @@ async def _handle_get_run(self, request: "web.Request") -> "web.Response": return auth_err run_id = request.match_info["run_id"] - status = self._run_statuses.get(run_id) + try: + from gateway.run_ledger import get_run + + await self._maintain_run_ledger() + status = await asyncio.to_thread(get_run, run_id) + except Exception: + logger.exception("[api_server] durable run status lookup failed") + return web.json_response( + _openai_error( + "Durable run storage is unavailable.", + err_type="server_error", + code="run_storage_unavailable", + ), + status=503, + ) if status is None: return web.json_response( _openai_error(f"Run not found: {run_id}", code="run_not_found"), status=404, ) + self._run_statuses[run_id] = status return web.json_response(status) async def _handle_run_events(self, request: "web.Request") -> "web.StreamResponse": @@ -6881,6 +7132,26 @@ async def _handle_run_events(self, request: "web.Request") -> "web.StreamRespons return auth_err run_id = request.match_info["run_id"] + try: + from gateway.run_ledger import get_run + + await self._maintain_run_ledger() + durable_status = await asyncio.to_thread(get_run, run_id) + except Exception: + logger.exception("[api_server] durable run event lookup failed") + return web.json_response( + _openai_error( + "Durable run storage is unavailable.", + err_type="server_error", + code="run_storage_unavailable", + ), + status=503, + ) + if durable_status is None: + return web.json_response( + _openai_error(f"Run not found: {run_id}", code="run_not_found"), + status=404, + ) # Allow subscribing slightly before the run is registered (race condition window) for _ in range(20): @@ -6933,7 +7204,21 @@ async def _handle_run_approval(self, request: "web.Request") -> "web.Response": return auth_err run_id = request.match_info["run_id"] - status = self._run_statuses.get(run_id) + try: + from gateway.run_ledger import get_run + + await self._maintain_run_ledger() + status = await asyncio.to_thread(get_run, run_id) + except Exception: + logger.exception("[api_server] durable run approval lookup failed") + return web.json_response( + _openai_error( + "Durable run storage is unavailable.", + err_type="server_error", + code="run_storage_unavailable", + ), + status=503, + ) if status is None: return web.json_response( _openai_error(f"Run not found: {run_id}", code="run_not_found"), @@ -6958,7 +7243,9 @@ async def _handle_run_approval(self, request: "web.Request") -> "web.Response": status=400, ) - approval_session_key = self._run_approval_sessions.get(run_id) + approval_session_key = None + if self._run_scopes.get(run_id) == self._current_run_scope(): + approval_session_key = self._run_approval_sessions.get(run_id) if not approval_session_key: return web.json_response( _openai_error( @@ -6993,7 +7280,9 @@ async def _handle_run_approval(self, request: "web.Request") -> "web.Response": status=409, ) - self._set_run_status(run_id, "running", last_event="approval.responded") + await self._set_run_status_async( + run_id, "running", last_event="approval.responded" + ) q = self._run_streams.get(run_id) if q is not None: try: @@ -7021,15 +7310,61 @@ async def _handle_stop_run(self, request: "web.Request") -> "web.Response": return auth_err run_id = request.match_info["run_id"] - agent = self._active_run_agents.get(run_id) - task = self._active_run_tasks.get(run_id) + scope_matches = self._run_scopes.get(run_id) == self._current_run_scope() + agent = self._active_run_agents.get(run_id) if scope_matches else None + task = self._active_run_tasks.get(run_id) if scope_matches else None + try: + from gateway.run_ledger import ( + TERMINAL_STATUSES, + get_run, + ) + + await self._maintain_run_ledger() + status = await asyncio.to_thread(get_run, run_id) + except Exception: + logger.exception("[api_server] durable run stop lookup failed") + if agent is None and task is None: + return web.json_response( + _openai_error( + "Durable run storage is unavailable.", + err_type="server_error", + code="run_storage_unavailable", + ), + status=503, + ) + status = self._run_statuses.get(run_id) if scope_matches else None + if status is None: + if agent is None and task is None: + return web.json_response( + _openai_error(f"Run not found: {run_id}", code="run_not_found"), + status=404, + ) + now = time.time() + status = { + "object": "hermes.run", + "run_id": run_id, + "status": "running", + "session_id": run_id, + "idempotency_key": None, + "created_at": now, + "updated_at": now, + } + self._run_statuses[run_id] = status + if status.get("status") in TERMINAL_STATUSES: + return web.json_response(status) if agent is None and task is None: - return web.json_response(_openai_error(f"Run not found: {run_id}", code="run_not_found"), status=404) + # Recovery classified the dead owner before this lookup. Return the + # exact durable state instead of hiding a known run behind 404. + return web.json_response(status) + + if run_id in self._stopping_run_ids: + return web.json_response(status) - self._set_run_status(run_id, "stopping", last_event="run.stopping") self._stopping_run_ids.add(run_id) + # Safety action precedes fallible persistence: a broken ledger must + # never leave an executor running after an acknowledged stop request. if agent is not None: try: request_hard_interrupt(agent, "Stop requested via API") @@ -7044,7 +7379,11 @@ async def _handle_stop_run(self, request: "web.Request") -> "web.Response": agent, source="api_server_run_stop" ) - return web.json_response({"run_id": run_id, "status": "stopping"}) + status = await self._set_run_status_async( + run_id, "stopping", last_event="run.stopping" + ) + + return web.json_response(status) async def _sweep_orphaned_runs(self) -> None: """Periodically expire transport buffers and terminal status records.""" @@ -7088,7 +7427,8 @@ def _sweep_orphaned_runs_once(self, now: Optional[float] = None) -> None: stale_statuses = [ run_id for run_id, status in list(self._run_statuses.items()) - if status.get("status") in {"completed", "failed", "cancelled"} + if status.get("status") + in {"completed", "failed", "cancelled", "interrupted"} and now - float(status.get("updated_at", 0) or 0) > self._RUN_STATUS_TTL ] for run_id in stale_statuses: diff --git a/gateway/run_ledger.py b/gateway/run_ledger.py new file mode 100644 index 0000000000000..255d1b84a5df3 --- /dev/null +++ b/gateway/run_ledger.py @@ -0,0 +1,299 @@ +"""Durable correlation and lifecycle ledger for ``/v1/runs``.""" + +from __future__ import annotations + +import json +import os +import sqlite3 +import threading +import time +from contextlib import contextmanager +from typing import Any, Dict, Iterable, Iterator, List, Optional, Tuple + +from hermes_constants import get_hermes_home + + +TERMINAL_STATUSES = frozenset({"completed", "failed", "cancelled", "interrupted"}) +ACTIVE_STATUSES = frozenset({"queued", "running", "waiting_for_approval", "stopping"}) +RETENTION_SECONDS = 24 * 60 * 60 +_ALLOWED_TRANSITIONS = { + "queued": frozenset({"running", "stopping", *TERMINAL_STATUSES}), + "running": frozenset({"waiting_for_approval", "stopping", *TERMINAL_STATUSES}), + "waiting_for_approval": frozenset({"running", "stopping", *TERMINAL_STATUSES}), + "stopping": TERMINAL_STATUSES, +} + + +class IdempotencyConflictError(RuntimeError): + """The same key was reused for a materially different request.""" + + +_lock = threading.RLock() + + +def transition_allowed(current: Optional[str], requested: str) -> bool: + """Return whether a lifecycle update is monotonic.""" + if current in TERMINAL_STATUSES: + return False + if current is None or current == requested: + return True + return requested in _ALLOWED_TRANSITIONS.get(current, frozenset()) + + +def _db_path(): + return get_hermes_home() / "state.db" + + +def _connect() -> sqlite3.Connection: + path = _db_path() + path.parent.mkdir(parents=True, exist_ok=True) + conn = sqlite3.connect(path, timeout=10) + conn.row_factory = sqlite3.Row + try: + from hermes_state import apply_wal_with_fallback + + apply_wal_with_fallback(conn, db_label="state.db (run_ledger)") + conn.execute("PRAGMA busy_timeout=10000") + conn.execute( + """CREATE TABLE IF NOT EXISTS api_runs ( + run_id TEXT PRIMARY KEY, + idempotency_key TEXT UNIQUE, + request_fingerprint TEXT, + status TEXT NOT NULL, + data TEXT NOT NULL, + owner_pid INTEGER, + owner_started_at INTEGER, + created_at REAL NOT NULL, + updated_at REAL NOT NULL + )""" + ) + conn.execute( + "CREATE INDEX IF NOT EXISTS idx_api_runs_updated_at " + "ON api_runs(status, updated_at)" + ) + conn.commit() + except Exception: + conn.close() + raise + return conn + + +@contextmanager +def _transaction() -> Iterator[sqlite3.Connection]: + with _lock: + conn = _connect() + try: + conn.execute("BEGIN IMMEDIATE") + yield conn + conn.commit() + except Exception: + conn.rollback() + raise + finally: + conn.close() + + +def _owner_stamp() -> tuple[int, Optional[int]]: + pid = os.getpid() + try: + from gateway.status import get_process_start_time + + return pid, get_process_start_time(pid) + except Exception: + return pid, None + + +def _owner_is_live(pid: Any, started_at: Any) -> bool: + if not pid: + return False + try: + pid = int(pid) + from gateway.status import _pid_exists, get_process_start_time + + if not _pid_exists(pid): + return False + if started_at is None: + return pid == os.getpid() + current = get_process_start_time(pid) + return current is not None and int(current) == int(started_at) + except Exception: + # Inability to prove owner death must not rewrite live state. + return True + + +def _decode(row: Optional[sqlite3.Row]) -> Optional[Dict[str, Any]]: + if row is None: + return None + data = json.loads(row["data"]) + data["run_id"] = row["run_id"] + data["status"] = row["status"] + data["idempotency_key"] = row["idempotency_key"] + data["created_at"] = row["created_at"] + data["updated_at"] = row["updated_at"] + return data + + +def reserve_run( + *, + run_id: str, + idempotency_key: Optional[str], + request_fingerprint: Optional[str], + data: Dict[str, Any], +) -> Tuple[Dict[str, Any], bool]: + """Atomically reserve a run identity before executor dispatch.""" + now = time.time() + pid, started_at = _owner_stamp() + with _transaction() as conn: + if idempotency_key: + existing = conn.execute( + "SELECT * FROM api_runs WHERE idempotency_key = ?", + (idempotency_key,), + ).fetchone() + if existing is not None: + if existing["request_fingerprint"] != request_fingerprint: + raise IdempotencyConflictError(idempotency_key) + return _decode(existing), False + + record = dict(data) + record.update({ + "object": "hermes.run", + "run_id": run_id, + "status": "queued", + "idempotency_key": idempotency_key, + "created_at": now, + "updated_at": now, + }) + conn.execute( + """INSERT INTO api_runs + (run_id, idempotency_key, request_fingerprint, status, data, + owner_pid, owner_started_at, created_at, updated_at) + VALUES (?, ?, ?, 'queued', ?, ?, ?, ?, ?)""", + ( + run_id, + idempotency_key, + request_fingerprint, + json.dumps(record, sort_keys=True, separators=(",", ":"), default=str), + pid, + started_at, + now, + now, + ), + ) + return record, True + + +def get_run(run_id: str) -> Optional[Dict[str, Any]]: + with _lock: + conn = _connect() + try: + return _decode( + conn.execute( + "SELECT * FROM api_runs WHERE run_id = ?", (run_id,) + ).fetchone() + ) + finally: + conn.close() + + +def get_run_by_idempotency_key(key: str) -> Optional[Dict[str, Any]]: + with _lock: + conn = _connect() + try: + return _decode( + conn.execute( + "SELECT * FROM api_runs WHERE idempotency_key = ?", (key,) + ).fetchone() + ) + finally: + conn.close() + + +def update_run(run_id: str, status: str, **fields: Any) -> Optional[Dict[str, Any]]: + """Persist one lifecycle transition; terminal rows cannot regress.""" + now = time.time() + with _transaction() as conn: + row = conn.execute( + "SELECT * FROM api_runs WHERE run_id = ?", (run_id,) + ).fetchone() + if row is None: + return None + if not transition_allowed(row["status"], status): + return _decode(row) + data = _decode(row) or {} + data.update(fields) + data.update({"status": status, "updated_at": now}) + terminal = status in TERMINAL_STATUSES + conn.execute( + """UPDATE api_runs + SET status = ?, data = ?, owner_pid = ?, owner_started_at = ?, updated_at = ? + WHERE run_id = ?""", + ( + status, + json.dumps(data, sort_keys=True, separators=(",", ":"), default=str), + None if terminal else row["owner_pid"], + None if terminal else row["owner_started_at"], + now, + run_id, + ), + ) + return data + + +def recover_interrupted_runs(active_run_ids: Optional[Iterable[str]] = None) -> List[str]: + """Classify rows that have no live owner and adapter task.""" + now = time.time() + recovered: List[str] = [] + active = None if active_run_ids is None else set(active_run_ids) + with _transaction() as conn: + placeholders = ",".join("?" for _ in ACTIVE_STATUSES) + rows = conn.execute( + f"SELECT * FROM api_runs WHERE status IN ({placeholders})", + tuple(ACTIVE_STATUSES), + ).fetchall() + for row in rows: + if _owner_is_live(row["owner_pid"], row["owner_started_at"]) and ( + active is None or row["run_id"] in active + ): + continue + data = _decode(row) or {} + data.update({ + "status": "interrupted", + "last_event": "run.interrupted", + "error": "Gateway restarted before the run reached a terminal state.", + "updated_at": now, + }) + conn.execute( + """UPDATE api_runs + SET status = 'interrupted', data = ?, owner_pid = NULL, + owner_started_at = NULL, updated_at = ? + WHERE run_id = ?""", + ( + json.dumps( + data, sort_keys=True, separators=(",", ":"), default=str + ), + now, + row["run_id"], + ), + ) + recovered.append(row["run_id"]) + return recovered + + +def purge_terminal_runs(now: Optional[float] = None) -> List[str]: + """Delete terminal correlation rows after the advertised retention window.""" + cutoff = (time.time() if now is None else now) - RETENTION_SECONDS + with _transaction() as conn: + placeholders = ",".join("?" for _ in TERMINAL_STATUSES) + rows = conn.execute( + f"SELECT run_id FROM api_runs " + f"WHERE status IN ({placeholders}) AND updated_at < ?", + (*TERMINAL_STATUSES, cutoff), + ).fetchall() + run_ids = [row["run_id"] for row in rows] + if run_ids: + delete_placeholders = ",".join("?" for _ in run_ids) + conn.execute( + f"DELETE FROM api_runs WHERE run_id IN ({delete_placeholders})", + tuple(run_ids), + ) + return run_ids diff --git a/tests/gateway/test_api_server.py b/tests/gateway/test_api_server.py index 57ac52dbe4b00..a25df3e6cc693 100644 --- a/tests/gateway/test_api_server.py +++ b/tests/gateway/test_api_server.py @@ -608,6 +608,16 @@ async def test_stop_run_reaps_owned_processes(self, adapter, monkeypatch): agent = MagicMock() _publish_turn_process_ownership(agent, "run-stop-sess") + from gateway.run_ledger import reserve_run, update_run + + reserve_run( + run_id="run_x", + idempotency_key=None, + request_fingerprint=None, + data={"session_id": "run-stop-sess", "model": "test"}, + ) + update_run("run_x", "running") + adapter._run_scopes["run_x"] = adapter._current_run_scope() adapter._active_run_agents["run_x"] = agent request = MagicMock() @@ -870,9 +880,19 @@ async def test_capabilities_advertises_plugin_safe_contract(self, adapter): assert data["features"]["chat_completions"] is True assert data["features"]["run_status"] is True assert data["features"]["run_events_sse"] is True + assert data["features"]["run_idempotency"] is True + assert data["features"]["run_correlation_lookup"] is True + assert data["features"]["run_status_durable"] is True + assert data["features"]["run_stop_idempotent"] is True + assert data["features"]["run_status_retention_seconds"] == 86400 assert data["features"]["model_options"] is True assert data["features"]["session_continuity_header"] == "X-Hermes-Session-Id" assert data["endpoints"]["run_status"]["path"] == "/v1/runs/{run_id}" + assert data["endpoints"]["run_lookup"] == { + "method": "GET", + "path": "/v1/runs", + "correlation_header": "Idempotency-Key", + } assert data["endpoints"]["model_options"] == {"method": "GET", "path": "/api/model/options"} assert data["endpoints"]["skills"] == {"method": "GET", "path": "/v1/skills"} assert data["endpoints"]["toolsets"] == {"method": "GET", "path": "/v1/toolsets"} @@ -2865,4 +2885,3 @@ def __init__(self, **kwargs): ) adapter._create_agent(session_id="another-session", gateway_session_key="stable-chan-1") assert captured[1]["model"] == "minimax/minimax-m3" - diff --git a/tests/gateway/test_api_server_runs.py b/tests/gateway/test_api_server_runs.py index 59ce03154ff57..ac85c4b138803 100644 --- a/tests/gateway/test_api_server_runs.py +++ b/tests/gateway/test_api_server_runs.py @@ -9,6 +9,7 @@ """ import asyncio +import sqlite3 import threading import time from unittest.mock import MagicMock, patch @@ -66,6 +67,7 @@ def _create_runs_app(adapter: APIServerAdapter) -> web.Application: app = web.Application(middlewares=mws) app["api_server_adapter"] = adapter app.router.add_post("/v1/runs", adapter._handle_runs) + app.router.add_get("/v1/runs", adapter._handle_lookup_run) app.router.add_get("/v1/runs/{run_id}", adapter._handle_get_run) app.router.add_get("/v1/runs/{run_id}/events", adapter._handle_run_events) app.router.add_post("/v1/runs/{run_id}/approval", adapter._handle_run_approval) @@ -120,6 +122,177 @@ def auth_adapter(): class TestStartRun: + @pytest.mark.asyncio + async def test_reservation_is_serialized_with_stale_snapshot_recovery( + self, monkeypatch, tmp_path + ): + monkeypatch.setenv("HERMES_HOME", str(tmp_path)) + adapter = _make_adapter() + app = _create_runs_app(adapter) + from gateway import run_ledger + + real_recover = run_ledger.recover_interrupted_runs + real_reserve = run_ledger.reserve_run + recovery_calls = 0 + second_recovery_entered = threading.Event() + allow_second_recovery = threading.Event() + first_row_reserved = threading.Event() + release_first_reservation = threading.Event() + + def controlled_recover(active_run_ids=None): + nonlocal recovery_calls + recovery_calls += 1 + if recovery_calls == 2: + second_recovery_entered.set() + allow_second_recovery.wait(timeout=2) + return real_recover(active_run_ids) + + def controlled_reserve(**kwargs): + result = real_reserve(**kwargs) + if kwargs.get("idempotency_key") == "serialized-a": + first_row_reserved.set() + release_first_reservation.wait(timeout=2) + return result + + monkeypatch.setattr(run_ledger, "recover_interrupted_runs", controlled_recover) + monkeypatch.setattr(run_ledger, "reserve_run", controlled_reserve) + 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 + first = asyncio.create_task( + cli.post( + "/v1/runs", + json={"input": "first"}, + headers={"Idempotency-Key": "serialized-a"}, + ) + ) + assert await asyncio.to_thread(first_row_reserved.wait, 1) + second = asyncio.create_task( + cli.post( + "/v1/runs", + json={"input": "second"}, + headers={"Idempotency-Key": "serialized-b"}, + ) + ) + await asyncio.sleep(0.05) + assert not second_recovery_entered.is_set() + release_first_reservation.set() + first_response = await first + assert await asyncio.to_thread(second_recovery_entered.wait, 1) + allow_second_recovery.set() + second_response = await second + assert first_response.status == 202 + assert second_response.status == 202 + first_run_id = (await first_response.json())["run_id"] + lookup = await cli.get(f"/v1/runs/{first_run_id}") + assert (await lookup.json())["status"] != "interrupted" + + @pytest.mark.asyncio + async def test_durable_reservation_does_not_block_event_loop( + self, monkeypatch, tmp_path + ): + monkeypatch.setenv("HERMES_HOME", str(tmp_path)) + adapter = _make_adapter() + app = _create_runs_app(adapter) + from gateway import run_ledger + + real_reserve = run_ledger.reserve_run + + reserve_entered = threading.Event() + release_reserve = threading.Event() + + def slow_reserve(**kwargs): + reserve_entered.set() + release_reserve.wait(timeout=2) + return real_reserve(**kwargs) + + monkeypatch.setattr(run_ledger, "reserve_run", slow_reserve) + 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 + post_task = asyncio.create_task( + cli.post("/v1/runs", json={"input": "hello"}) + ) + assert await asyncio.to_thread(reserve_entered.wait, 1) + started = time.monotonic() + await asyncio.sleep(0.02) + elapsed = time.monotonic() - started + assert elapsed < 0.5 + release_reserve.set() + response = await post_task + assert response.status == 202 + + @pytest.mark.asyncio + async def test_idempotency_key_conflict_does_not_start_second_run( + self, monkeypatch, tmp_path + ): + monkeypatch.setenv("HERMES_HOME", str(tmp_path)) + adapter = _make_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": "aic-conflict-398"} + + first = await cli.post( + "/v1/runs", json={"input": "first"}, headers=headers + ) + conflict = await cli.post( + "/v1/runs", json={"input": "changed"}, headers=headers + ) + + assert first.status == 202 + assert conflict.status == 409 + data = await conflict.json() + assert data["error"]["code"] == "idempotency_conflict" + assert mock_create.call_count == 1 + + @pytest.mark.asyncio + async def test_same_idempotency_key_recovers_original_run(self, monkeypatch, tmp_path): + """A lost 202 can be retried without dispatching a duplicate run.""" + monkeypatch.setenv("HERMES_HOME", str(tmp_path)) + adapter = _make_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": "aic-submit-398"} + payload = {"input": "hello", "session_id": "aic-session"} + + first = await cli.post("/v1/runs", json=payload, headers=headers) + second = await cli.post("/v1/runs", json=payload, headers=headers) + + assert first.status == 202 + assert second.status == 202 + first_data = await first.json() + second_data = await second.json() + assert second_data["run_id"] == first_data["run_id"] + assert second_data["session_id"] == "aic-session" + assert second_data["idempotency_key"] == "aic-submit-398" + assert mock_create.call_count == 1 + @pytest.mark.asyncio async def test_start_returns_202(self, adapter): app = _create_runs_app(adapter) @@ -266,6 +439,240 @@ async def test_start_passes_request_model_provider_options_to_create_agent(self, class TestRunStatus: + @pytest.mark.asyncio + async def test_cross_profile_cannot_read_or_stop_cached_run( + self, monkeypatch, tmp_path + ): + home_a = tmp_path / "profile-a" + home_b = tmp_path / "profile-b" + monkeypatch.setenv("HERMES_HOME", str(home_a)) + adapter = _make_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 + response = await cli.post( + "/v1/runs", + json={"input": "profile a"}, + headers={"Idempotency-Key": "shared-profile-key"}, + ) + run_id = (await response.json())["run_id"] + for _ in range(40): + if adapter._run_statuses[run_id]["status"] == "completed": + break + await asyncio.sleep(0.05) + + monkeypatch.setenv("HERMES_HOME", str(home_b)) + status_response = await cli.get(f"/v1/runs/{run_id}") + stop_response = await cli.post(f"/v1/runs/{run_id}/stop") + assert status_response.status == 404 + assert stop_response.status == 404 + + @pytest.mark.asyncio + async def test_missing_adapter_task_reconciles_live_owner_row( + self, monkeypatch, tmp_path + ): + monkeypatch.setenv("HERMES_HOME", str(tmp_path)) + from gateway.run_ledger import reserve_run, update_run + + reserve_run( + run_id="run_no_task", + idempotency_key="aic-no-task-398", + request_fingerprint="fingerprint", + data={"session_id": "no-task", "model": "test"}, + ) + update_run("run_no_task", "running") + adapter = _make_adapter() + app = _create_runs_app(adapter) + async with TestClient(TestServer(app)) as cli: + lookup = await cli.get( + "/v1/runs", headers={"Idempotency-Key": "aic-no-task-398"} + ) + assert lookup.status == 200 + assert (await lookup.json())["status"] == "interrupted" + + @pytest.mark.asyncio + async def test_terminal_retention_purges_expired_correlation( + self, monkeypatch, tmp_path + ): + monkeypatch.setenv("HERMES_HOME", str(tmp_path)) + from gateway.run_ledger import RETENTION_SECONDS, reserve_run, update_run + + reserve_run( + run_id="run_expired", + idempotency_key="aic-expired-398", + request_fingerprint="fingerprint", + data={"session_id": "expired", "model": "test"}, + ) + update_run("run_expired", "completed") + with sqlite3.connect(tmp_path / "state.db") as conn: + conn.execute( + "UPDATE api_runs SET updated_at = ? WHERE run_id = ?", + (time.time() - RETENTION_SECONDS - 1, "run_expired"), + ) + + adapter = _make_adapter() + app = _create_runs_app(adapter) + async with TestClient(TestServer(app)) as cli: + lookup = await cli.get( + "/v1/runs", headers={"Idempotency-Key": "aic-expired-398"} + ) + assert lookup.status == 404 + + def test_terminal_status_cannot_regress(self, monkeypatch, tmp_path): + monkeypatch.setenv("HERMES_HOME", str(tmp_path)) + from gateway.run_ledger import get_run, reserve_run, update_run + + reserve_run( + run_id="run_terminal", + idempotency_key="aic-terminal-398", + request_fingerprint="fingerprint", + data={"session_id": "terminal-session", "model": "test"}, + ) + update_run("run_terminal", "completed", output="done") + update_run("run_terminal", "running", last_event="late.event") + + status = get_run("run_terminal") + assert status["status"] == "completed" + assert status["output"] == "done" + assert status.get("last_event") != "late.event" + + def test_terminal_same_status_update_is_byte_immutable( + self, monkeypatch, tmp_path + ): + monkeypatch.setenv("HERMES_HOME", str(tmp_path)) + from gateway.run_ledger import get_run, reserve_run, update_run + + reserve_run( + run_id="run_terminal_same", + idempotency_key="aic-terminal-same-398", + request_fingerprint="fingerprint", + data={"session_id": "terminal-same", "model": "test"}, + ) + update_run("run_terminal_same", "completed", output="original") + before = get_run("run_terminal_same") + with sqlite3.connect(tmp_path / "state.db") as conn: + durable_before = conn.execute( + "SELECT * FROM api_runs WHERE run_id = ?", ("run_terminal_same",) + ).fetchone() + + update_run( + "run_terminal_same", + "completed", + output="changed", + last_event="late.completed", + ) + with sqlite3.connect(tmp_path / "state.db") as conn: + durable_after = conn.execute( + "SELECT * FROM api_runs WHERE run_id = ?", ("run_terminal_same",) + ).fetchone() + assert durable_after == durable_before + assert get_run("run_terminal_same") == before + + adapter = _make_adapter() + adapter._run_statuses["run_terminal_same"] = dict(before) + adapter._set_run_status( + "run_terminal_same", + "completed", + output="memory changed", + last_event="late.memory.completed", + ) + assert adapter._run_statuses["run_terminal_same"] == before + + def test_stopping_status_cannot_regress_to_active(self, monkeypatch, tmp_path): + monkeypatch.setenv("HERMES_HOME", str(tmp_path)) + from gateway.run_ledger import get_run, reserve_run, update_run + + reserve_run( + run_id="run_stopping", + idempotency_key="aic-stopping-398", + request_fingerprint="fingerprint", + data={"session_id": "stopping-session", "model": "test"}, + ) + update_run("run_stopping", "running") + update_run("run_stopping", "stopping") + update_run("run_stopping", "waiting_for_approval") + update_run("run_stopping", "running") + + assert get_run("run_stopping")["status"] == "stopping" + + @pytest.mark.asyncio + async def test_restart_marks_orphaned_run_interrupted(self, monkeypatch, tmp_path): + monkeypatch.setenv("HERMES_HOME", str(tmp_path)) + from gateway.run_ledger import reserve_run, update_run + + reserve_run( + run_id="run_orphaned", + idempotency_key="aic-orphaned-398", + request_fingerprint="fingerprint", + data={"session_id": "orphaned-session", "model": "test"}, + ) + update_run("run_orphaned", "running") + with sqlite3.connect(tmp_path / "state.db") as conn: + conn.execute( + "UPDATE api_runs SET owner_pid = ?, owner_started_at = ? WHERE run_id = ?", + (999_999_999, 1, "run_orphaned"), + ) + + restarted_adapter = _make_adapter() + restarted_app = _create_runs_app(restarted_adapter) + async with TestClient(TestServer(restarted_app)) as cli: + lookup = await cli.get( + "/v1/runs", headers={"Idempotency-Key": "aic-orphaned-398"} + ) + + assert lookup.status == 200 + data = await lookup.json() + assert data["run_id"] == "run_orphaned" + assert data["status"] == "interrupted" + assert data["last_event"] == "run.interrupted" + + @pytest.mark.asyncio + async def test_correlation_lookup_survives_adapter_restart(self, monkeypatch, tmp_path): + monkeypatch.setenv("HERMES_HOME", str(tmp_path)) + headers = {"Idempotency-Key": "aic-restart-398"} + first_adapter = _make_adapter() + first_app = _create_runs_app(first_adapter) + + async with TestClient(TestServer(first_app)) as cli: + with patch.object(first_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 + response = await cli.post( + "/v1/runs", + json={"input": "hello", "session_id": "restart-session"}, + headers=headers, + ) + run_id = (await response.json())["run_id"] + for _ in range(40): + if first_adapter._run_statuses[run_id]["status"] == "completed": + break + await asyncio.sleep(0.05) + + restarted_adapter = _make_adapter() + restarted_app = _create_runs_app(restarted_adapter) + async with TestClient(TestServer(restarted_app)) as cli: + status_response = await cli.get(f"/v1/runs/{run_id}") + assert status_response.status == 200 + assert (await status_response.json())["status"] == "completed" + + lookup = await cli.get("/v1/runs", headers=headers) + assert lookup.status == 200 + data = await lookup.json() + assert data["run_id"] == run_id + assert data["status"] == "completed" + assert data["session_id"] == "restart-session" + assert data["idempotency_key"] == "aic-restart-398" + @pytest.mark.asyncio async def test_status_reflects_explicit_session_id(self, adapter): app = _create_runs_app(adapter) @@ -409,6 +816,18 @@ async def test_approval_resolve_all_is_scoped_to_target_run(self, auth_adapter): class TestRunLifecycleSweep: + def test_interrupted_status_is_swept_from_memory(self, adapter): + now = time.time() + adapter._run_statuses["run_interrupted"] = { + "run_id": "run_interrupted", + "status": "interrupted", + "updated_at": now - adapter._RUN_STATUS_TTL - 1, + } + + adapter._sweep_orphaned_runs_once(now) + + assert "run_interrupted" not in adapter._run_statuses + @pytest.mark.asyncio async def test_expired_live_run_drops_transport_but_keeps_control_state(self, adapter): @@ -424,7 +843,7 @@ async def test_expired_live_run_drops_transport_but_keeps_control_state(self, ad start_resp = await cli.post("/v1/runs", json={"input": "hello"}) assert start_resp.status == 202 run_id = (await start_resp.json())["run_id"] - assert agent_ready.wait(timeout=3.0) + assert await asyncio.to_thread(agent_ready.wait, 3.0) task = adapter._active_run_tasks[run_id] assert isinstance(task, asyncio.Task) @@ -477,6 +896,68 @@ async def test_expired_live_run_drops_transport_but_keeps_control_state(self, ad class TestStopRun: + @pytest.mark.asyncio + async def test_storage_failure_cannot_prevent_active_interrupt( + self, monkeypatch, tmp_path + ): + monkeypatch.setenv("HERMES_HOME", str(tmp_path)) + adapter = _make_adapter() + app = _create_runs_app(adapter) + async with TestClient(TestServer(app)) as cli: + with patch.object(adapter, "_create_agent") as mock_create: + mock_agent, ready, interrupted = _make_slow_agent() + mock_create.return_value = mock_agent + response = await cli.post("/v1/runs", json={"input": "hello"}) + run_id = (await response.json())["run_id"] + assert await asyncio.to_thread(ready.wait, 3) + + monkeypatch.setattr( + "gateway.run_ledger.update_run", + MagicMock(side_effect=sqlite3.OperationalError("disk full")), + ) + stop_response = await cli.post(f"/v1/runs/{run_id}/stop") + + assert stop_response.status == 200 + assert interrupted.wait(timeout=1) + assert (await stop_response.json())["status"] == "stopping" + + @pytest.mark.asyncio + async def test_stop_terminal_run_is_idempotent_after_restart( + self, monkeypatch, tmp_path + ): + monkeypatch.setenv("HERMES_HOME", str(tmp_path)) + first_adapter = _make_adapter() + first_app = _create_runs_app(first_adapter) + async with TestClient(TestServer(first_app)) as cli: + with patch.object(first_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 + response = await cli.post( + "/v1/runs", + json={"input": "hello"}, + headers={"Idempotency-Key": "aic-stop-terminal-398"}, + ) + run_id = (await response.json())["run_id"] + for _ in range(40): + if first_adapter._run_statuses[run_id]["status"] == "completed": + break + await asyncio.sleep(0.05) + + restarted_adapter = _make_adapter() + restarted_app = _create_runs_app(restarted_adapter) + async with TestClient(TestServer(restarted_app)) as cli: + first_stop = await cli.post(f"/v1/runs/{run_id}/stop") + second_stop = await cli.post(f"/v1/runs/{run_id}/stop") + + assert first_stop.status == 200 + assert second_stop.status == 200 + assert (await first_stop.json())["status"] == "completed" + assert (await second_stop.json())["run_id"] == run_id + @pytest.mark.asyncio async def test_stop_keeps_uncooperative_executor_tracked_until_exit(self, adapter): """Cancelling an asyncio wrapper must not hide its live executor thread.""" @@ -503,7 +984,7 @@ def _run_conversation(*_args, **_kwargs): resp = await cli.post("/v1/runs", json={"input": "hello"}) run_id = (await resp.json())["run_id"] - assert started.wait(timeout=3) + assert await asyncio.to_thread(started.wait, 3) stop_resp = await cli.post(f"/v1/runs/{run_id}/stop") assert stop_resp.status == 200