""" A2A protocol helpers — Agent Card construction, JSON-RPC framing, task store, and disk-backed conversation persistence. Wire shape follows A2A Protocol v1.0 (JSON-RPC 2.0 binding over HTTP): - Agent Card served at GET /.well-known/agent-card.json (canonical v1.0; legacy agent.json also answers) - Tasks via POST {jsonrpc:"2.0", method:"message/send", params:{...}} - Streaming via ``message/stream`` → SSE; events are StreamResponse objects discriminated by member presence (``statusUpdate`` / ``artifactUpdate``), stream closure signals the terminal state (no ``final`` field in v1.0) - Task states / message roles are v1.0 SCREAMING_SNAKE_CASE enums - Parts are the v1.0 unified shape ({"text": ..., "mediaType": ...}), discriminated by member presence (no ``kind`` field) - Push notification configs carry ``configId`` + ``createdAt`` and can be passed inline in ``message/send`` via configuration.taskPushNotificationConfig We deliberately implement the subset of A2A needed for text task exchange with stdlib only (no a2a-sdk). ``extract_text`` stays tolerant of v0.3 peers. """ from __future__ import annotations import json import copy import os import threading import time import uuid from collections import OrderedDict, defaultdict, deque from concurrent.futures import Future from datetime import datetime, timezone from pathlib import Path from typing import Any, Optional PROTOCOL_VERSION = "1.0" # A2A v1.0 task lifecycle states. STATE_SUBMITTED = "TASK_STATE_SUBMITTED" STATE_WORKING = "TASK_STATE_WORKING" STATE_INPUT_REQUIRED = "TASK_STATE_INPUT_REQUIRED" STATE_AUTH_REQUIRED = "TASK_STATE_AUTH_REQUIRED" STATE_COMPLETED = "TASK_STATE_COMPLETED" STATE_FAILED = "TASK_STATE_FAILED" STATE_CANCELED = "TASK_STATE_CANCELED" STATE_REJECTED = "TASK_STATE_REJECTED" TERMINAL_STATES = frozenset({STATE_COMPLETED, STATE_FAILED, STATE_CANCELED, STATE_REJECTED}) # A2A v1.0 message roles. ROLE_USER = "ROLE_USER" ROLE_AGENT = "ROLE_AGENT" # The agent starts its reply with this marker when it needs clarification from # the peer before it can complete the task; the adapter maps such replies to # TASK_STATE_INPUT_REQUIRED (marker stripped, text in status.message). INPUT_REQUIRED_MARKER = "[INPUT_REQUIRED]" # JSON-RPC / A2A error codes. # -32001..-32003 are A2A spec-defined and used only with their spec semantics. # Custom errors live at -32050..-32059 (JSON-RPC implementation-defined server # error space, clear of the A2A-reserved block). ERR_PARSE = -32700 ERR_INVALID_PARAMS = -32602 ERR_METHOD_NOT_FOUND = -32601 ERR_TASK_NOT_FOUND = -32001 # A2A spec: TaskNotFoundError ERR_TASK_NOT_CANCELABLE = -32002 # A2A spec: TaskNotCancelableError ERR_PUSH_NOT_SUPPORTED = -32003 # A2A spec: PushNotificationNotSupportedError ERR_UNAUTHORIZED = -32050 ERR_RATE_LIMITED = -32051 ERR_UNTRUSTED_PEER = -32052 # Maximum turns an A2A conversation can have before anti-loop kicks in. # Default 5, configurable via A2A_MAX_PINGPONG_TURNS env (max 20). _DEFAULT_MAX_PINGPONG = 5 _HARD_MAX_PINGPONG = 20 def max_pingpong_turns() -> int: try: v = int(os.getenv("A2A_MAX_PINGPONG_TURNS", str(_DEFAULT_MAX_PINGPONG))) return max(1, min(v, _HARD_MAX_PINGPONG)) except (ValueError, TypeError): return _DEFAULT_MAX_PINGPONG def now_iso() -> str: """ISO 8601 UTC timestamp with millisecond precision (A2A v1.0).""" return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%S.%f")[:-3] + "Z" # -------------------------------------------------------------------------- # Agent Card (v1.0) # -------------------------------------------------------------------------- def build_agent_card( *, name: str, url: str, description: str, skills: Optional[list[dict]] = None, streaming: bool = False, push_notifications: bool = False, auth_required: bool = False, tenant: str = "", ) -> dict: """Construct an A2A v1.0 Agent Card document. ``tenant`` is the optional v1.0 multi-tenancy routing key advertised on AgentInterface. When present, clients MUST echo it in request params. """ iface: dict[str, Any] = { "url": url, "protocolBinding": "JSONRPC", "protocolVersion": PROTOCOL_VERSION, } if tenant: iface["tenant"] = tenant card: dict[str, Any] = { "name": name, "description": description, "url": url, # convenience for pre-1.0 clients; canonical is supportedInterfaces "version": "1.0.0", "provider": { "organization": os.getenv("A2A_PROVIDER_ORG", "Hermes Agent"), "url": os.getenv("A2A_PROVIDER_URL", "") or url, }, "supportedInterfaces": [iface], "capabilities": { "streaming": streaming, "pushNotifications": push_notifications, "stateTransitionHistory": False, "extendedAgentCard": False, }, "defaultInputModes": ["text/plain"], "defaultOutputModes": ["text/plain"], "skills": skills or [], } if auth_required: card["securitySchemes"] = { "bearer": {"type": "http", "scheme": "bearer"} } card["security"] = [{"bearer": []}] return card def skills_from_toolsets(toolsets: "list[str] | dict[str, list[str]] | None") -> list[dict]: """Derive A2A skill descriptors from the agent's toolsets. Accepts either a plain list of toolset names, or a mapping of toolset name → tool names (built from the live tool registry for dynamic Agent Cards — tool names become tags so peers can match tasks to us). """ skills = [] if isinstance(toolsets, dict): for ts_name in sorted(toolsets.keys()): tool_names = [str(t) for t in (toolsets[ts_name] or [])] skills.append({ "id": f"toolset.{ts_name}", "name": ts_name, "description": f"Hermes '{ts_name}' capabilities", "tags": [ts_name] + tool_names[:10], }) else: for ts in sorted(set(toolsets or [])): skills.append({ "id": f"toolset.{ts}", "name": ts, "description": f"Hermes '{ts}' capabilities", "tags": [ts], }) if not skills: skills.append({ "id": "general", "name": "general", "description": "General-purpose conversational agent", "tags": ["general"], }) return skills # -------------------------------------------------------------------------- # JSON-RPC framing # -------------------------------------------------------------------------- def jsonrpc_result(req_id: Any, result: Any) -> dict: return {"jsonrpc": "2.0", "id": req_id, "result": result} def jsonrpc_error(req_id: Any, code: int, message: str) -> dict: return {"jsonrpc": "2.0", "id": req_id, "error": {"code": code, "message": message}} def send_message_response(payload: dict) -> dict: """A2A v1.0 SendMessageResponse oneof wrapper. The JSON-RPC ``SendMessage`` result is not a bare Task/Message; it is a wrapper containing exactly one of ``task`` or ``message``. Legacy methods still return bare payloads for compatibility. """ if isinstance(payload, dict) and payload.get("status") and payload.get("id"): return {"task": payload} return {"message": payload} def unwrap_send_message_response(result: Any) -> Any: """Return the Task/Message inside a v1.0 response, or pass legacy through.""" if isinstance(result, dict): if isinstance(result.get("task"), dict): return result["task"] if isinstance(result.get("message"), dict): return result["message"] return result def stream_task(task: dict) -> dict: """v1.0 StreamResponse with a task member.""" return {"task": task} def stream_message(message: dict) -> dict: """v1.0 StreamResponse with a message member.""" return {"message": message} def new_task_id() -> str: return "task-" + uuid.uuid4().hex[:16] def new_context_id() -> str: return "ctx-" + uuid.uuid4().hex[:16] def text_part(text: str) -> dict: """Build a v1.0 text Part (member-presence discriminated, no ``kind``).""" return {"text": text, "mediaType": "text/plain"} def file_part(url: str = "", raw: str = "", filename: str = "", media_type: str = "application/octet-stream") -> dict: """Build a v1.0 file Part. Either ``url`` (file reference) or ``raw`` (base64-encoded bytes) must be provided. Discrimination is by member presence — no ``kind`` field. """ part: dict[str, Any] = {"mediaType": media_type} if filename: part["filename"] = filename if url: part["url"] = url elif raw: part["raw"] = raw return part def data_part(data: Any, media_type: str = "application/json") -> dict: """Build a v1.0 data Part (structured data, no ``kind`` field).""" return {"data": data, "mediaType": media_type} def text_message(role: str, text: str, context_id: str = "") -> dict: """Build an A2A v1.0 Message with a single text Part.""" msg: dict[str, Any] = { "role": role, # ROLE_USER | ROLE_AGENT "parts": [text_part(text)], "messageId": uuid.uuid4().hex, } if context_id: msg["contextId"] = context_id return msg def message_with_parts(role: str, parts: list[dict], context_id: str = "") -> dict: """Build an A2A v1.0 Message with arbitrary Parts (text, file, data).""" msg: dict[str, Any] = { "role": role, "parts": parts, "messageId": uuid.uuid4().hex, } if context_id: msg["contextId"] = context_id return msg def extract_text(message_or_params: dict) -> str: """Pull concatenated text from an A2A Message / Task-result / params payload. v1.0 Parts carry a ``text`` member directly; v0.3 used ``kind: "text"`` and some pre-0.3 peers used ``type``. All three shapes put the payload in ``part["text"]``, so presence of a string ``text`` member is the test. File and data Parts are rendered into the text stream so the agent sees them: file Parts with a URL include the URL and filename; data Parts include their JSON-serialised content. Raw (base64) file Parts are noted but not decoded (the agent can't act on binary inline). """ msg = message_or_params.get("message", message_or_params) parts = msg.get("parts", []) if isinstance(msg, dict) else [] chunks = [] for part in parts: if not isinstance(part, dict): continue # v1.0 text part (member-presence discrimination) txt = part.get("text") if isinstance(txt, str): chunks.append(txt) continue # v0.3 compatibility: kind == "text" if part.get("kind") == "text" and isinstance(part.get("text"), str): chunks.append(part["text"]) continue # v1.0 file part with URL url = part.get("url") if isinstance(url, str) and url: fname = part.get("filename") or part.get("name") or "" mtype = part.get("mediaType") or part.get("mimeType") or "" label = f"[file: {fname}]" if fname else "[file]" chunks.append(f"{label} {url}" + (f" ({mtype})" if mtype else "")) continue # v0.3 file part with nested file.fileWithUri v03_file = part.get("file") if isinstance(v03_file, dict) and isinstance(v03_file.get("fileWithUri"), str): uri = v03_file["fileWithUri"] fname = v03_file.get("name") or "" mtype = v03_file.get("mimeType") or "" label = f"[file: {fname}]" if fname else "[file]" chunks.append(f"{label} {uri}" + (f" ({mtype})" if mtype else "")) continue # v1.0 file part with raw bytes (base64) — note but don't decode if isinstance(part.get("raw"), str): fname = part.get("filename") or "" mtype = part.get("mediaType") or "" label = f"[file: {fname}]" if fname else "[file]" size_note = f"{len(part['raw'])} bytes base64-encoded" chunks.append(f"{label} {size_note}" + (f" ({mtype})" if mtype else "")) continue # v1.0 data part — include JSON content data = part.get("data") if data is not None: try: rendered = json.dumps(data, ensure_ascii=False, default=str) except (TypeError, ValueError): rendered = str(data) mtype = part.get("mediaType") or "application/json" chunks.append(f"[data ({mtype})]\n{rendered}") continue # v0.3 data part: kind == "data" if part.get("kind") == "data" and part.get("data") is not None: try: rendered = json.dumps(part["data"], ensure_ascii=False, default=str) except (TypeError, ValueError): rendered = str(part["data"]) chunks.append(f"[data]\n{rendered}") continue return "\n".join(chunks).strip() def extract_context_id(params: dict) -> str: """v1.0 puts contextId inside the Message; tolerate legacy top-level.""" msg = params.get("message") or {} ctx = "" if isinstance(msg, dict): ctx = str(msg.get("contextId") or "") return ctx or str(params.get("contextId") or "") def build_task( task_id: str, context_id: str, state: str, agent_text: str = "", *, created_at: str = "", ) -> dict: """Build an A2A v1.0 Task object for a message/send result. ``created_at`` is accepted for call-site compatibility but not serialized — the A2A v1.0 ``Task`` proto (``lf.a2a.v1.Task``) has no ``createdAt`` or ``lastModified`` field. Strict ProtoJSON parsers (e.g. a2a-sdk 1.1.0) reject unknown fields, so we must not include them. The spec's §5.6.1 timestamp-format example mentions them but they are not in the proto. """ now = now_iso() task: dict[str, Any] = { "id": task_id, "contextId": context_id, "status": {"state": state, "timestamp": now}, } if agent_text: task["status"]["message"] = text_message(ROLE_AGENT, agent_text, context_id) if state == STATE_COMPLETED: task["artifacts"] = [{ "artifactId": uuid.uuid4().hex, "parts": [text_part(agent_text)], }] return task # -------------------------------------------------------------------------- # Streaming (v1.0 StreamResponse events) # -------------------------------------------------------------------------- def status_update(task_id: str, context_id: str, state: str, text: str = "") -> dict: """v1.0 StreamResponse with a statusUpdate member.""" status: dict[str, Any] = {"state": state, "timestamp": now_iso()} if text: status["message"] = text_message(ROLE_AGENT, text, context_id) return {"statusUpdate": {"taskId": task_id, "contextId": context_id, "status": status}} def artifact_update(task_id: str, context_id: str, text: str) -> dict: """v1.0 StreamResponse with an artifactUpdate member.""" return { "artifactUpdate": { "taskId": task_id, "contextId": context_id, "artifact": { "artifactId": uuid.uuid4().hex, "parts": [text_part(text)], }, } } def sse_data(payload: dict, req_id: Any = None) -> str: """Encode one StreamResponse as a JSON-RPC-wrapped SSE data frame. A2A v1.0 §9.4 requires each SSE frame to be a full JSON-RPC response: ``{"jsonrpc":"2.0","id":,"result":{StreamResponse}}``. Emitting a bare StreamResponse (the REST binding shape) breaks JSON-RPC clients that expect the envelope, including the official a2a-sdk. """ if req_id is not None: envelope = jsonrpc_result(req_id, payload) else: envelope = payload # legacy/fallback — no envelope return f"data: {json.dumps(envelope, ensure_ascii=False)}\n\n" def sse_done() -> str: """SSE stream-closure marker — a comment, not a parseable data frame. A2A v1.0 signals terminal state by closing the stream. Emitting ``data: {}`` causes JSON-RPC clients to try parsing an empty response and fail. An SSE comment line (``: done``) is ignored by all SSE parsers. """ return ": done\n\n" # -------------------------------------------------------------------------- # Anti-loop ping-pong protection (per-adapter instance) # -------------------------------------------------------------------------- class TurnTracker: """Counts inbound turns per context_id to stop infinite agent↔agent loops. A "turn" is one inbound message/send from a peer. When the count exceeds max_pingpong_turns(), the adapter rejects further messages for that context. """ _TTL = 3600 # prune contexts idle longer than 1 hour def __init__(self) -> None: self._counts: dict[str, int] = defaultdict(int) self._timestamps: dict[str, float] = {} self._lock = threading.Lock() def track(self, context_id: str) -> int: """Increment and return the turn count; prunes stale contexts.""" with self._lock: now = time.time() stale = [cid for cid, ts in self._timestamps.items() if now - ts > self._TTL] for cid in stale: self._counts.pop(cid, None) self._timestamps.pop(cid, None) self._counts[context_id] += 1 self._timestamps[context_id] = now return self._counts[context_id] def reset(self, context_id: str) -> None: """Reset turn count for a context (e.g. after explicit cancel).""" with self._lock: self._counts.pop(context_id, None) self._timestamps.pop(context_id, None) # -------------------------------------------------------------------------- # Rate limiting (sliding window per authenticated peer identity) # -------------------------------------------------------------------------- _RATE_LIMIT_DEFAULT = 60 # requests per minute _RATE_WINDOW = 60.0 # seconds def _rate_limit_per_minute() -> int: try: return max(1, int(os.getenv("A2A_RATE_LIMIT", str(_RATE_LIMIT_DEFAULT)))) except (ValueError, TypeError): return _RATE_LIMIT_DEFAULT class RateLimiter: """Sliding-window request limiter, one bucket per authenticated identity.""" def __init__(self) -> None: self._buckets: dict[str, deque[float]] = defaultdict(deque) self._lock = threading.Lock() def allow(self, identity: str) -> bool: with self._lock: limit = _rate_limit_per_minute() now = time.time() bucket = self._buckets[identity] while bucket and now - bucket[0] > _RATE_WINDOW: bucket.popleft() if len(bucket) >= limit: return False bucket.append(now) return True # -------------------------------------------------------------------------- # Metrics collection # -------------------------------------------------------------------------- # Module-level singleton shared by the inbound adapter and the outbound client # tools so /metrics and a2a_list report both directions. Not persisted. class Metrics: """Simple counters for A2A operations.""" def __init__(self) -> None: self.inbound_total = 0 self.outbound_total = 0 self.streams_started = 0 self.push_sent = 0 self.push_failed = 0 self.tasks_completed = 0 self.tasks_failed = 0 self.anti_loop_triggers = 0 self.rate_limit_triggers = 0 self._start_time = time.time() # Rolling latency tracking (last 100 completed inbound tasks) self._latencies: deque[float] = deque(maxlen=100) def record_latency(self, seconds: float) -> None: self._latencies.append(seconds) def avg_latency(self) -> float: if not self._latencies: return 0.0 return sum(self._latencies) / len(self._latencies) def snapshot(self) -> dict[str, Any]: uptime = time.time() - self._start_time return { "uptime_seconds": round(uptime, 1), "inbound_total": self.inbound_total, "outbound_total": self.outbound_total, "streams_started": self.streams_started, "push_sent": self.push_sent, "push_failed": self.push_failed, "tasks_completed": self.tasks_completed, "tasks_failed": self.tasks_failed, "anti_loop_triggers": self.anti_loop_triggers, "rate_limit_triggers": self.rate_limit_triggers, "avg_latency_ms": round(self.avg_latency() * 1000, 1), } metrics = Metrics() # -------------------------------------------------------------------------- # Task store — pending AND completed tasks (queryable via tasks/get, tasks/list) # -------------------------------------------------------------------------- class TaskStore: """In-memory store of A2A tasks, kept after completion for tasks/get. Records carry the routed agent slug and tenant. All read/write helpers accept optional scope values and return not-found when the task exists but is not visible in that scope, satisfying the spec's authorization scoping rule. """ _MAX_TERMINAL = 500 def __init__(self) -> None: self._tasks: "OrderedDict[str, dict[str, Any]]" = OrderedDict() self._watchers: dict[str, list[Future]] = {} self._lock = threading.Lock() @staticmethod def _in_scope(rec: dict, agent_slug: str = "", tenant: str = "") -> bool: if agent_slug and rec.get("agent_slug", "") != agent_slug: return False if tenant and rec.get("tenant", "") != tenant: return False return True def create(self, task_id: str, context_id: str, peer: str, agent_slug: str = "", tenant: str = "") -> dict: rec = { "task_id": task_id, "context_id": context_id, "peer": peer, "agent_slug": agent_slug or "", "tenant": tenant or "", "state": STATE_SUBMITTED, "reply": "", "created_at": time.time(), "created_iso": now_iso(), "push_url": "", "push_config_id": "", } with self._lock: self._tasks[task_id] = rec return dict(rec) def set_state(self, task_id: str, state: str) -> None: with self._lock: rec = self._tasks.get(task_id) if rec and rec["state"] not in TERMINAL_STATES: rec["state"] = state def set_push_config(self, task_id: str, url: str, agent_slug: str = "", tenant: str = "") -> Optional[dict]: """Attach a push notification config; returns the stored config or None.""" with self._lock: rec = self._tasks.get(task_id) if not rec or not self._in_scope(rec, agent_slug, tenant): return None rec["push_url"] = url rec["push_config_id"] = "cfg-" + uuid.uuid4().hex[:12] return self._push_config_view(rec) @staticmethod def _push_config_view(rec: dict) -> dict: """Build the JSON-RPC result for a push notification config.""" return { "configId": rec.get("push_config_id") or "", "taskId": rec["task_id"], "createdAt": rec.get("created_iso", ""), "pushNotificationConfig": {"url": rec.get("push_url") or ""}, } def get_push_config(self, task_id: str, config_id: str = "", agent_slug: str = "", tenant: str = "") -> Optional[dict]: with self._lock: rec = self._tasks.get(task_id) if not rec or not self._in_scope(rec, agent_slug, tenant) or not rec.get("push_url"): return None if config_id and rec.get("push_config_id") != config_id: return None return self._push_config_view(rec) def list_push_configs(self, task_id: str, agent_slug: str = "", tenant: str = "") -> list[dict]: with self._lock: rec = self._tasks.get(task_id) if not rec or not self._in_scope(rec, agent_slug, tenant) or not rec.get("push_url"): return [] return [self._push_config_view(rec)] def delete_push_config(self, task_id: str, config_id: str = "", agent_slug: str = "", tenant: str = "") -> bool: with self._lock: rec = self._tasks.get(task_id) if not rec or not self._in_scope(rec, agent_slug, tenant) or not rec.get("push_url"): return False if config_id and rec.get("push_config_id") != config_id: return False rec["push_url"] = "" rec["push_config_id"] = "" return True def pop_push_url(self, task_id: str) -> str: with self._lock: rec = self._tasks.get(task_id) if not rec: return "" url, rec["push_url"] = rec["push_url"], "" return url def get(self, task_id: str, agent_slug: str = "", tenant: str = "") -> Optional[dict]: with self._lock: rec = self._tasks.get(task_id) if not rec or not self._in_scope(rec, agent_slug, tenant): return None return dict(rec) def complete(self, task_id: str, state: str, reply: str = "") -> Optional[dict]: """Transition a task to a terminal state. Idempotent.""" watchers: list[Future] = [] with self._lock: rec = self._tasks.get(task_id) if not rec or rec["state"] in TERMINAL_STATES: return None rec["state"] = state rec["reply"] = reply rec["completed_at"] = time.time() watchers = self._watchers.pop(task_id, []) self._trim_locked() out = dict(rec) for fut in watchers: if not fut.done(): fut.set_result((state, reply)) return out def watch(self, task_id: str, agent_slug: str = "", tenant: str = "") -> Optional[Future]: with self._lock: rec = self._tasks.get(task_id) if not rec or not self._in_scope(rec, agent_slug, tenant): return None fut: Future = Future() if rec["state"] in TERMINAL_STATES: fut.set_result((rec["state"], rec.get("reply", ""))) else: self._watchers.setdefault(task_id, []).append(fut) return fut def list( self, context_id: str = "", state: str = "", page_size: int = 50, offset: int = 0, agent_slug: str = "", tenant: str = "", with_total: bool = False, ): """Filtered task page (newest first). Historical API returns ``(records, next_offset)``. v1.0 ListTasks needs ``totalSize``, so callers can opt into ``(records, next_offset, total)``. """ page_size = max(1, min(int(page_size or 50), 100)) with self._lock: recs = [dict(r) for r in reversed(self._tasks.values())] if agent_slug or tenant: recs = [r for r in recs if self._in_scope(r, agent_slug, tenant)] if context_id: recs = [r for r in recs if r["context_id"] == context_id] if state: recs = [r for r in recs if r["state"] == state] total = len(recs) page = recs[offset:offset + page_size] next_offset = offset + page_size if offset + page_size < total else 0 if with_total: return page, next_offset, total return page, next_offset def fail_orphans(self, timeout_seconds: int = 300) -> list[str]: with self._lock: now = time.time() stale = [ tid for tid, rec in self._tasks.items() if rec["state"] not in TERMINAL_STATES and now - rec["created_at"] > timeout_seconds ] failed = [] for tid in stale: if self.complete(tid, STATE_FAILED, "[task orphaned — no reply produced]"): failed.append(tid) return failed def _trim_locked(self) -> None: terminal = [tid for tid, rec in self._tasks.items() if rec["state"] in TERMINAL_STATES] excess = len(terminal) - self._MAX_TERMINAL for tid in terminal[:max(0, excess)]: self._tasks.pop(tid, None) @staticmethod def to_task(rec: dict, history_length: Optional[int] = None, include_artifacts: bool = True) -> dict: """Render a stored record as an A2A v1.0 Task object.""" task = build_task( rec["task_id"], rec["context_id"], rec["state"], rec.get("reply", ""), created_at=rec.get("created_iso", ""), ) if not include_artifacts: task.pop("artifacts", None) if history_length == 0: task.pop("history", None) return copy.deepcopy(task) # -------------------------------------------------------------------------- # Conversation persistence (outside the context-compaction pipeline) # -------------------------------------------------------------------------- def _conv_dir() -> Path: try: from hermes_constants import get_hermes_home base = Path(get_hermes_home()) except Exception: base = Path(os.path.expanduser("~/.hermes")) return base / "a2a_conversations" def _safe_name(context_id: str) -> str: return "".join(c for c in (context_id or "default") if c.isalnum() or c in "-_") or "default" def persist_message(context_id: str, role: str, text: str, task_id: str = "") -> None: """Append one message to the context's on-disk conversation log.""" try: d = _conv_dir() d.mkdir(parents=True, exist_ok=True) rec = {"ts": time.time(), "role": role, "text": text, "task_id": task_id} with (d / f"{_safe_name(context_id)}.jsonl").open("a", encoding="utf-8") as fh: fh.write(json.dumps(rec, ensure_ascii=False) + "\n") except Exception: pass def load_conversation(context_id: str, limit: int = 50) -> list[dict]: """Load the last *limit* messages for a context (empty list if none).""" path = _conv_dir() / f"{_safe_name(context_id)}.jsonl" if not path.exists(): return [] out: list[dict] = [] try: with path.open("r", encoding="utf-8") as fh: for line in fh: line = line.strip() if not line: continue try: out.append(json.loads(line)) except json.JSONDecodeError: continue except Exception: return [] return out[-limit:] def list_conversations() -> list[str]: """Return known context-ids that have persisted conversations.""" d = _conv_dir() if not d.exists(): return [] return sorted(p.stem for p in d.glob("*.jsonl"))