From 7047c8c37165cb0cda5741554375f5fda64ad5b9 Mon Sep 17 00:00:00 2001 From: handsdiff <239876380+handsdiff@users.noreply.github.com> Date: Thu, 23 Apr 2026 23:02:29 -0400 Subject: [PATCH] feat(agent): add actor runtime state --- cron/scheduler.py | 17 + gateway/agent_actor.py | 584 ++++++++++++++++++ gateway/run.py | 103 ++- gateway/session.py | 13 + gateway/session_context.py | 15 + hermes_cli/tools_config.py | 1 + hermes_state.py | 491 ++++++++++++++- tests/gateway/test_agent_actor.py | 222 +++++++ tests/gateway/test_routing_context_user_id.py | 9 + tests/test_hermes_state.py | 84 ++- tests/tools/test_registry.py | 1 + tests/tools/test_self_state_tool.py | 124 ++++ tests/tools/test_send_message_tool.py | 32 + tools/self_state_tool.py | 501 +++++++++++++++ tools/send_message_tool.py | 74 +++ toolsets.py | 12 +- 16 files changed, 2273 insertions(+), 10 deletions(-) create mode 100644 gateway/agent_actor.py create mode 100644 tests/gateway/test_agent_actor.py create mode 100644 tests/tools/test_self_state_tool.py create mode 100644 tools/self_state_tool.py diff --git a/cron/scheduler.py b/cron/scheduler.py index 3eba1ee102c6..2fbeddabd35e 100644 --- a/cron/scheduler.py +++ b/cron/scheduler.py @@ -371,6 +371,23 @@ def _deliver_result(job: dict, content: str, adapters=None, loop=None) -> Option delivery_errors.append(msg) continue + try: + from gateway.agent_actor import evaluate_send_message_policy + + decision = evaluate_send_message_policy( + target_platform=platform_name, + target_chat_id=chat_id, + target_thread_id=thread_id or "", + message=cleaned_delivery_content, + ) + if not decision.allowed: + msg = f"delivery blocked by {decision.policy}: {decision.reason}" + logger.warning("Job '%s': %s", job["id"], msg) + delivery_errors.append(msg) + continue + except Exception as e: + logger.debug("Job '%s': outbound policy check failed open: %s", job["id"], e) + # Prefer the live adapter when the gateway is running — this supports E2EE # rooms (e.g. Matrix) where the standalone HTTP path cannot encrypt. runtime_adapter = (adapters or {}).get(platform) diff --git a/gateway/agent_actor.py b/gateway/agent_actor.py new file mode 100644 index 000000000000..b38c52e77d3f --- /dev/null +++ b/gateway/agent_actor.py @@ -0,0 +1,584 @@ +"""Agent-level runtime state for gateway sessions. + +Sessions remain transcript windows. This module adds a small actor layer above +them: per-platform identity, append-only events, active directives, structured +state packets, and outbound policy checks. +""" + +from __future__ import annotations + +import json +import os +import re +from dataclasses import dataclass +from typing import Any, Dict, Iterable, Optional + +from gateway.session_context import get_session_env + + +_PUBLIC_PLATFORM_NAMES = { + "discord", + "slack", + "telegram", + "matrix", + "mattermost", + "feishu", + "dingtalk", + "wecom", + "wecom_callback", + "weixin", + "qqbot", +} + +_AUTONOMOUS_SOURCE_NAMES = { + "cron", + "hub", + "webhook", + "homeassistant", + "api_server", +} + +_STOP_WORD_RE = re.compile( + r"\b(stop|turn\s+this\s+off|turn\s+off|disable|pause|shut\s+off|kill)\b", + re.IGNORECASE, +) +_PUBLIC_BROADCAST_RE = re.compile( + r"\b(post(?:ing)?|broadcast(?:ing)?|send(?:ing)?|digest|market|alpha|cron|general|channel|public)\b", + re.IGNORECASE, +) + + +@dataclass +class GateDecision: + allowed: bool + reason: str = "" + policy: str = "" + event_id: str = "" + + def to_dict(self) -> Dict[str, Any]: + return { + "allowed": self.allowed, + "reason": self.reason, + "policy": self.policy, + "event_id": self.event_id, + } + + +def _platform_value(value: Any) -> str: + if hasattr(value, "value"): + return str(value.value) + return str(value or "") + + +def _source_to_payload(source: Any) -> Dict[str, Any]: + try: + return source.to_dict() + except Exception: + return {} + + +def _preview(text: Any, limit: int = 500) -> str: + s = str(text or "").replace("\r", " ").replace("\n", " ").strip() + if len(s) <= limit: + return s + return s[: limit - 3].rstrip() + "..." + + +def _load_db(db=None): + if db is not None: + return db, False + from hermes_state import SessionDB + + return SessionDB(), True + + +def _split_identifier_csv(raw: str) -> set[str]: + return {part.strip() for part in str(raw or "").split(",") if part.strip()} + + +def owner_user_ids_for_platform(platform: str) -> set[str]: + """Return owner user IDs for a trusted platform. + + The provisioner-owned identity should eventually be emitted as env/config. + Until then, SOUL.md is the generated local source available on agent VMs. + Keep this narrowly scoped: only the explicit "Your owner" block is parsed. + """ + platform = str(platform or "").strip().lower() + ids: set[str] = set() + ids.update(_split_identifier_csv(os.getenv("GATEWAY_OWNER_USER_IDS", ""))) + if platform: + prefix = platform.upper() + ids.update(_split_identifier_csv(os.getenv(f"{prefix}_OWNER_USER_ID", ""))) + ids.update(_split_identifier_csv(os.getenv(f"{prefix}_OWNER_USER_IDS", ""))) + if ids or platform != "discord": + return ids + + try: + from hermes_constants import get_hermes_home + + soul = get_hermes_home() / "SOUL.md" + text = soul.read_text(encoding="utf-8", errors="replace") + except Exception: + return ids + + owner_match = re.search(r"(?ms)^## Your owner\b(?P.*?)(?:^## |\Z)", text) + if not owner_match: + return ids + body = owner_match.group("body") + for pattern in ( + r"Discord:\s*`?@?[^`\n]*`?\s*\(user_id\s*`?(\d{5,})`?\)", + r"\buser_id\s*`?(\d{5,})`?", + ): + ids.update(re.findall(pattern, body)) + return ids + + +def infer_platform_authority(source: Any) -> str: + """Best-effort v1 authority without cross-platform person unification.""" + platform = _platform_value(getattr(source, "platform", "")) + user_id = str(getattr(source, "user_id", "") or "") + chat_type = str(getattr(source, "chat_type", "") or "") + if not platform or not user_id: + return "system" + check_ids = {user_id} + if "@" in user_id: + check_ids.add(user_id.split("@", 1)[0]) + owner_ids = owner_user_ids_for_platform(platform) + + platform_allowlist_env = { + "telegram": "TELEGRAM_ALLOWED_USERS", + "discord": "DISCORD_ALLOWED_USERS", + "whatsapp": "WHATSAPP_ALLOWED_USERS", + "slack": "SLACK_ALLOWED_USERS", + "signal": "SIGNAL_ALLOWED_USERS", + "email": "EMAIL_ALLOWED_USERS", + "sms": "SMS_ALLOWED_USERS", + "mattermost": "MATTERMOST_ALLOWED_USERS", + "matrix": "MATRIX_ALLOWED_USERS", + "dingtalk": "DINGTALK_ALLOWED_USERS", + "feishu": "FEISHU_ALLOWED_USERS", + "wecom": "WECOM_ALLOWED_USERS", + "wecom_callback": "WECOM_CALLBACK_ALLOWED_USERS", + "weixin": "WEIXIN_ALLOWED_USERS", + "bluebubbles": "BLUEBUBBLES_ALLOWED_USERS", + "qqbot": "QQ_ALLOWED_USERS", + "hub": "HUB_ALLOWED_USERS", + } + if owner_ids and check_ids & owner_ids: + return "owner" + platform_allow_all_env = f"{platform.upper()}_ALLOW_ALL_USERS" + if os.getenv(platform_allow_all_env, "").strip().lower() in {"1", "true", "yes"}: + return "user" + + allowed_raw = ",".join( + part + for part in ( + os.getenv(platform_allowlist_env.get(platform, ""), ""), + os.getenv("GATEWAY_ALLOWED_USERS", ""), + ) + if part + ) + allowed = {p.strip() for p in allowed_raw.split(",") if p.strip()} + if "*" in allowed: + return "trusted" + if allowed and check_ids & allowed: + return "trusted" + if chat_type == "dm": + return "known" + return "user" + + +def resolve_identity(db, source: Any, authority: str = "") -> str: + platform = _platform_value(getattr(source, "platform", "")) + user_id = str(getattr(source, "user_id", "") or "") + if not platform or not user_id: + return "" + authority = authority or infer_platform_authority(source) + return db.upsert_agent_identity( + platform=platform, + platform_user_id=user_id, + display_name=str(getattr(source, "user_name", "") or ""), + authority=authority, + payload={ + "chat_id": str(getattr(source, "chat_id", "") or ""), + "chat_type": str(getattr(source, "chat_type", "") or ""), + "thread_id": str(getattr(source, "thread_id", "") or ""), + }, + ) + + +def record_inbound_event( + db, + *, + source: Any, + session_id: str, + session_key: str, + text: str, + message_id: str = "", + platform_update_id: str = "", + authority: str = "", +) -> tuple[str, str]: + """Record one inbound MessageEvent and return (event_id, person_id).""" + person_id = resolve_identity(db, source, authority=authority) + platform = _platform_value(getattr(source, "platform", "")) + chat_type = str(getattr(source, "chat_type", "") or "") + event_id = db.append_agent_event( + event_type="inbound", + event_subtype="message", + status="received", + session_id=session_id, + session_key=session_key, + actor_id="main", + actor_kind="user" if person_id else "system", + source=platform, + person_id=person_id, + sender_user_id=str(getattr(source, "user_id", "") or ""), + sender_name=str(getattr(source, "user_name", "") or ""), + chat_type=chat_type, + audience_type="private" if chat_type == "dm" else "shared", + platform=platform, + platform_chat_id=str(getattr(source, "chat_id", "") or ""), + platform_thread_id=str(getattr(source, "thread_id", "") or ""), + platform_message_id=str(message_id or ""), + platform_update_id=str(platform_update_id or ""), + content=_preview(text, 1000), + payload={ + "source": _source_to_payload(source), + "authority": authority or infer_platform_authority(source), + }, + ) + return event_id, person_id + + +def detect_public_broadcast_stop_directive(text: str) -> Optional[Dict[str, Any]]: + """Conservative v1 detector for owner/trusted stop-broadcast directives.""" + if not text: + return None + if not _STOP_WORD_RE.search(text): + return None + if not _PUBLIC_BROADCAST_RE.search(text): + return None + return { + "text": _preview(text, 1000), + "target": "public_broadcasts", + "behavior": "suppress", + "reason": "trusted sender asked the agent to stop/disable public broadcast behavior", + } + + +def maybe_record_directive_from_inbound( + db, + *, + source: Any, + session_id: str, + session_key: str, + inbound_event_id: str, + person_id: str, + text: str, + authority: str, +) -> Optional[str]: + """Persist a durable directive extracted from an inbound trusted message.""" + if authority not in {"trusted", "owner"}: + return None + directive = detect_public_broadcast_stop_directive(text) + if not directive: + return None + platform = _platform_value(getattr(source, "platform", "")) + return db.create_or_replace_agent_directive( + directive_scope="actor", + directive_key="public-broadcast-suppression", + directive_type="suppress_public_broadcasts", + payload={**directive, "source_event_id": inbound_event_id}, + session_id=session_id, + session_key="", + actor_id="main", + issuer_person_id=person_id, + issuer_platform=platform, + issuer_user_id=str(getattr(source, "user_id", "") or ""), + priority=100, + ) + + +def _format_directive_line(directive: Dict[str, Any]) -> str: + payload = directive.get("payload") or {} + text = payload.get("text") or directive.get("directive_type") or directive.get("directive_key") + return f"- {directive.get('directive_type')}: {_preview(text, 220)}" + + +def _format_runtime_event_line(event: Dict[str, Any]) -> str: + who = event.get("person_id") or event.get("sender_name") or event.get("sender_user_id") or "unknown" + target = event.get("platform_chat_id") or "unknown" + session = event.get("session_key") or "unknown-session" + return ( + f"- {event.get('event_type') or '?'}:{event.get('event_subtype') or '?'} " + f"{event.get('status') or 'unknown'} platform={event.get('platform') or event.get('source') or '?'} " + f"chat={target} chat_type={event.get('chat_type') or '?'} " + f"person={who} session={session} content={_preview(event.get('content'), 180)!r}" + ) + + +def _visible_recent_events(db, *, authority: str, session_key: str, person_id: str) -> list[Dict[str, Any]]: + if authority in {"owner", "trusted"}: + return db.list_recent_agent_events(limit=8) + events = db.list_recent_agent_events(session_key=session_key, limit=5) + if person_id: + by_person = db.list_recent_agent_events(person_id=person_id, limit=5) + events = list({event["event_id"]: event for event in events + by_person}.values()) + return events[:8] + + +def build_state_packet( + db, + *, + source: Any, + session_id: str, + session_key: str, + inbound_event_id: str = "", + person_id: str = "", + authority: str = "", +) -> str: + """Build an ephemeral structured system prefix for this exact event.""" + platform = _platform_value(getattr(source, "platform", "")) + chat_id = str(getattr(source, "chat_id", "") or "") + chat_type = str(getattr(source, "chat_type", "") or "") + user_id = str(getattr(source, "user_id", "") or "") + user_name = str(getattr(source, "user_name", "") or "") + authority = authority or infer_platform_authority(source) + person_id = person_id or (f"{platform}:{user_id}" if platform and user_id else "") + + try: + directives = db.list_active_agent_directives(actor_id="main", limit=8) + except Exception: + directives = [] + try: + recent_events = _visible_recent_events( + db, + authority=authority, + session_key=session_key, + person_id=person_id, + ) + except Exception: + recent_events = [] + + lines = [ + "## Agent Runtime State", + "", + "This packet is ephemeral runtime state, not conversation history.", + "Use it for behavior and policy. Do not reveal private cross-session details unless the current sender is authorized and disclosure is needed.", + "", + "**Current Sender:**", + f"- person_id: {person_id or 'unknown'}", + f"- platform_user_id: {user_id or 'unknown'}", + f"- display_name: {user_name or 'unknown'}", + f"- authority: {authority}", + "", + "**Current Audience:**", + f"- platform: {platform or 'unknown'}", + f"- chat_id: {chat_id or 'unknown'}", + f"- chat_type: {chat_type or 'unknown'}", + f"- session_key: {session_key}", + f"- inbound_event_id: {inbound_event_id or 'unknown'}", + ] + + if directives: + lines.extend(["", "**Active Agent Directives:**"]) + lines.extend(_format_directive_line(d) for d in directives) + else: + lines.extend(["", "**Active Agent Directives:**", "- none"]) + + if recent_events: + lines.extend(["", "**Recent Runtime Events:**"]) + lines.extend(_format_runtime_event_line(event) for event in recent_events) + else: + lines.extend(["", "**Recent Runtime Events:**", "- none"]) + + lines.extend([ + "", + "**Broadcast Policy Reminder:**", + "- A cron, Hub, webhook, or other autonomous inbound must not be rebroadcast to a public channel unless the current event contains explicit trusted authorization.", + "- If an action is blocked by policy, explain the causal source instead of trying another public target.", + ]) + return "\n".join(lines) + + +def _iter_directive_payloads(db) -> Iterable[Dict[str, Any]]: + try: + directives = db.list_active_agent_directives( + actor_id="main", + directive_type="suppress_public_broadcasts", + limit=20, + ) + except Exception: + directives = [] + for directive in directives: + yield directive.get("payload") or {} + + +def _event_content_for_current_context(db) -> str: + event_id = get_session_env("HERMES_AGENT_EVENT_ID", "") + if not event_id: + return "" + try: + event = db.get_agent_event(event_id) + except Exception: + event = None + return str((event or {}).get("content") or "") + + +def _is_publicish_cross_session_target( + *, + target_platform: str, + target_chat_id: str, + source_platform: str, + source_chat_id: str, +) -> bool: + if target_platform not in _PUBLIC_PLATFORM_NAMES: + return False + if not source_platform or target_platform != source_platform: + return True + return bool(target_chat_id and source_chat_id and target_chat_id != source_chat_id) + + +def evaluate_send_message_policy( + *, + target_platform: str, + target_chat_id: str, + target_thread_id: str = "", + message: str = "", + db=None, +) -> GateDecision: + """Gate model/tool initiated cross-channel sends.""" + db, should_close = _load_db(db) + try: + source_platform = get_session_env("HERMES_SESSION_PLATFORM", "").strip().lower() + source_chat_id = get_session_env("HERMES_SESSION_CHAT_ID", "").strip() + source_session_key = get_session_env("HERMES_SESSION_KEY", "").strip() + target_platform = str(target_platform or "").strip().lower() + target_chat_id = str(target_chat_id or "").strip() + target_thread_id = str(target_thread_id or "").strip() + message = str(message or "") + risky_public_target = _is_publicish_cross_session_target( + target_platform=target_platform, + target_chat_id=target_chat_id, + source_platform=source_platform, + source_chat_id=source_chat_id, + ) + + active_stop = any(_iter_directive_payloads(db)) + if active_stop and risky_public_target: + event_id = db.append_agent_event( + event_type="outbound", + event_subtype="send_message", + status="blocked", + session_key=source_session_key, + actor_id="main", + actor_kind="tool", + source=source_platform or "tool", + person_id=get_session_env("HERMES_AGENT_PERSON_ID", ""), + sender_user_id=get_session_env("HERMES_SESSION_USER_ID", ""), + sender_name=get_session_env("HERMES_SESSION_USER_NAME", ""), + parent_event_id=get_session_env("HERMES_AGENT_EVENT_ID", ""), + platform=target_platform, + platform_chat_id=target_chat_id, + platform_thread_id=target_thread_id, + tool_name="send_message", + content=_preview(message, 1000), + payload={"decision": "deny", "source_platform": source_platform}, + ) + return GateDecision( + allowed=False, + reason="Active directive suppresses public/cross-session broadcasts.", + policy="suppress_public_broadcasts", + event_id=event_id, + ) + + inbound_content = _event_content_for_current_context(db) + cron_like = inbound_content.lower().startswith("cronjob response:") + autonomous_source = source_platform in _AUTONOMOUS_SOURCE_NAMES + if risky_public_target and (cron_like or autonomous_source): + event_id = db.append_agent_event( + event_type="outbound", + event_subtype="send_message", + status="blocked", + session_key=source_session_key, + actor_id="main", + actor_kind="tool", + source=source_platform or "tool", + person_id=get_session_env("HERMES_AGENT_PERSON_ID", ""), + sender_user_id=get_session_env("HERMES_SESSION_USER_ID", ""), + sender_name=get_session_env("HERMES_SESSION_USER_NAME", ""), + parent_event_id=get_session_env("HERMES_AGENT_EVENT_ID", ""), + platform=target_platform, + platform_chat_id=target_chat_id, + platform_thread_id=target_thread_id, + tool_name="send_message", + content=_preview(message, 1000), + payload={ + "decision": "deny", + "source_platform": source_platform, + "cron_like": cron_like, + "autonomous_source": autonomous_source, + }, + ) + return GateDecision( + allowed=False, + reason="Autonomous cron/Hub/webhook context cannot rebroadcast to public/cross-session targets without explicit trusted authorization.", + policy="autonomous_public_rebroadcast_guard", + event_id=event_id, + ) + + return GateDecision(allowed=True) + finally: + if should_close: + db.close() + + +def record_send_message_outbound( + *, + target_platform: str, + target_chat_id: str, + target_thread_id: str = "", + message: str = "", + status: str = "succeeded", + result: Optional[Dict[str, Any]] = None, +) -> None: + """Best-effort event log write after an allowed send attempt.""" + db = None + try: + from hermes_state import SessionDB + + db = SessionDB() + db.append_agent_event( + event_type="outbound", + event_subtype="send_message", + status=status, + session_key=get_session_env("HERMES_SESSION_KEY", ""), + actor_id="main", + actor_kind="tool", + source=get_session_env("HERMES_SESSION_PLATFORM", "tool"), + person_id=get_session_env("HERMES_AGENT_PERSON_ID", ""), + sender_user_id=get_session_env("HERMES_SESSION_USER_ID", ""), + sender_name=get_session_env("HERMES_SESSION_USER_NAME", ""), + parent_event_id=get_session_env("HERMES_AGENT_EVENT_ID", ""), + platform=target_platform, + platform_chat_id=target_chat_id, + platform_thread_id=target_thread_id, + tool_name="send_message", + content=_preview(message, 1000), + payload={"result": result or {}}, + ) + except Exception: + pass + finally: + if db is not None: + db.close() + + +def blocked_tool_result(decision: GateDecision) -> str: + return json.dumps({ + "success": False, + "blocked": True, + "policy": decision.policy, + "reason": decision.reason, + "event_id": decision.event_id, + }, ensure_ascii=False) diff --git a/gateway/run.py b/gateway/run.py index ccc022c3c72b..4a8ef6cb9dba 100644 --- a/gateway/run.py +++ b/gateway/run.py @@ -1072,6 +1072,14 @@ def _classify_source_kind(self, source) -> str: except Exception: pass + try: + from gateway.agent_actor import infer_platform_authority + + if infer_platform_authority(source) == "owner": + return "owner" + except Exception: + pass + try: cfg = getattr(self, "config", None) get_hc = getattr(cfg, "get_home_channel", None) if cfg is not None else None @@ -4063,9 +4071,49 @@ async def _handle_message_with_agent(self, event, source, _quick_key: str, run_g # Build session context context = build_session_context(source, self.config, session_entry) + + inbound_event_id = "" + inbound_person_id = "" + inbound_authority = "" + if self._session_db is not None: + try: + from gateway.agent_actor import ( + build_state_packet, + infer_platform_authority, + maybe_record_directive_from_inbound, + record_inbound_event, + ) + + inbound_authority = infer_platform_authority(source) + inbound_event_id, inbound_person_id = record_inbound_event( + self._session_db, + source=source, + session_id=session_entry.session_id, + session_key=session_key, + text=event.text or "", + message_id=getattr(event, "message_id", "") or "", + platform_update_id=str(getattr(event, "platform_update_id", "") or ""), + authority=inbound_authority, + ) + maybe_record_directive_from_inbound( + self._session_db, + source=source, + session_id=session_entry.session_id, + session_key=session_key, + inbound_event_id=inbound_event_id, + person_id=inbound_person_id, + text=event.text or "", + authority=inbound_authority, + ) + except Exception as _actor_exc: + logger.debug("Agent actor inbound recording failed: %s", _actor_exc) # Set session context variables for tools (task-local, concurrency-safe) - _session_env_tokens = self._set_session_env(context) + _session_env_tokens = self._set_session_env( + context, + agent_event_id=inbound_event_id, + person_id=inbound_person_id, + ) # Read privacy.redact_pii from config (re-read per message) _redact_pii = False @@ -4079,6 +4127,21 @@ async def _handle_message_with_agent(self, event, source, _quick_key: str, run_g # Build the context prompt to inject context_prompt = build_session_context_prompt(context, redact_pii=_redact_pii) + if self._session_db is not None: + try: + from gateway.agent_actor import build_state_packet + + context_prompt += "\n\n" + build_state_packet( + self._session_db, + source=source, + session_id=session_entry.session_id, + session_key=session_key, + inbound_event_id=inbound_event_id, + person_id=inbound_person_id, + authority=inbound_authority, + ) + except Exception as _actor_exc: + logger.debug("Agent actor state packet failed: %s", _actor_exc) # If the previous session expired and was auto-reset, prepend a notice # so the agent knows this is a fresh conversation (not an intentional /reset). @@ -4787,6 +4850,33 @@ async def _handle_message_with_agent(self, event, source, _quick_key: str, run_g last_prompt_tokens=agent_result.get("last_prompt_tokens", 0), ) + if response and response.strip() != "NO_REPLY" and self._session_db is not None: + try: + self._session_db.append_agent_event( + event_type="outbound", + event_subtype="reply", + status="prepared", + session_id=session_entry.session_id, + session_key=session_key, + actor_id="main", + actor_kind="agent", + source=source.platform.value if source.platform else "", + person_id=inbound_person_id, + sender_user_id=str(source.user_id or ""), + sender_name=str(source.user_name or ""), + parent_event_id=inbound_event_id, + platform=source.platform.value if source.platform else "", + platform_chat_id=str(source.chat_id or ""), + platform_thread_id=str(source.thread_id or ""), + content=(response or "").replace("\r", " ").replace("\n", " ")[:1000], + payload={ + "api_calls": _api_calls, + "response_time_seconds": round(_response_time, 3), + }, + ) + except Exception as _actor_exc: + logger.debug("Agent actor outbound reply recording failed: %s", _actor_exc) + # Auto voice reply: send TTS audio before the text response _already_sent = bool(agent_result.get("already_sent")) if self._should_send_voice_reply(event, response, agent_messages, already_sent=_already_sent): @@ -8190,7 +8280,13 @@ async def _send_restart_notification(self) -> None: finally: notify_path.unlink(missing_ok=True) - def _set_session_env(self, context: SessionContext) -> list: + def _set_session_env( + self, + context: SessionContext, + *, + agent_event_id: str = "", + person_id: str = "", + ) -> list: """Set session context variables for the current async task. Uses ``contextvars`` instead of ``os.environ`` so that concurrent @@ -8204,10 +8300,13 @@ def _set_session_env(self, context: SessionContext) -> list: platform=context.source.platform.value, chat_id=context.source.chat_id, chat_name=context.source.chat_name or "", + chat_type=context.source.chat_type or "", thread_id=str(context.source.thread_id) if context.source.thread_id else "", user_id=str(context.source.user_id) if context.source.user_id else "", user_name=str(context.source.user_name) if context.source.user_name else "", session_key=context.session_key, + agent_event_id=agent_event_id, + person_id=person_id, ) def _clear_session_env(self, tokens: list) -> None: diff --git a/gateway/session.py b/gateway/session.py index ea3f174909de..2f8a2002cc7b 100644 --- a/gateway/session.py +++ b/gateway/session.py @@ -241,6 +241,19 @@ def build_session_context_prompt( if context.source.chat_topic: lines.append(f"**Channel Topic:** {context.source.chat_topic}") + lines.append("") + lines.append("**Runtime self-awareness:** Conversation sessions are isolated, but you have a shared runtime.") + lines.append( + "When the user asks what you are doing, why you posted or messaged something, " + "what sessions are active, or whether a cron or another session caused behavior, " + "call `self_state` before answering." + ) + lines.append( + "`self_state` is the first source for this runtime evidence. Do not inspect " + "~/.hermes files, session DBs, logs, or cron files with terminal as a substitute " + "until after `self_state` has been tried and found insufficient." + ) + # User identity. # In shared multi-user sessions (shared threads OR shared non-thread groups # when group_sessions_per_user=False), multiple users contribute to the same diff --git a/gateway/session_context.py b/gateway/session_context.py index 9dc051e3a2c4..e68d139d1d00 100644 --- a/gateway/session_context.py +++ b/gateway/session_context.py @@ -51,10 +51,13 @@ _SESSION_PLATFORM: ContextVar = ContextVar("HERMES_SESSION_PLATFORM", default=_UNSET) _SESSION_CHAT_ID: ContextVar = ContextVar("HERMES_SESSION_CHAT_ID", default=_UNSET) _SESSION_CHAT_NAME: ContextVar = ContextVar("HERMES_SESSION_CHAT_NAME", default=_UNSET) +_SESSION_CHAT_TYPE: ContextVar = ContextVar("HERMES_SESSION_CHAT_TYPE", default=_UNSET) _SESSION_THREAD_ID: ContextVar = ContextVar("HERMES_SESSION_THREAD_ID", default=_UNSET) _SESSION_USER_ID: ContextVar = ContextVar("HERMES_SESSION_USER_ID", default=_UNSET) _SESSION_USER_NAME: ContextVar = ContextVar("HERMES_SESSION_USER_NAME", default=_UNSET) _SESSION_KEY: ContextVar = ContextVar("HERMES_SESSION_KEY", default=_UNSET) +_AGENT_EVENT_ID: ContextVar = ContextVar("HERMES_AGENT_EVENT_ID", default=_UNSET) +_AGENT_PERSON_ID: ContextVar = ContextVar("HERMES_AGENT_PERSON_ID", default=_UNSET) # Cron auto-delivery vars — set per-job in run_job() so concurrent jobs # don't clobber each other's delivery targets. @@ -66,10 +69,13 @@ "HERMES_SESSION_PLATFORM": _SESSION_PLATFORM, "HERMES_SESSION_CHAT_ID": _SESSION_CHAT_ID, "HERMES_SESSION_CHAT_NAME": _SESSION_CHAT_NAME, + "HERMES_SESSION_CHAT_TYPE": _SESSION_CHAT_TYPE, "HERMES_SESSION_THREAD_ID": _SESSION_THREAD_ID, "HERMES_SESSION_USER_ID": _SESSION_USER_ID, "HERMES_SESSION_USER_NAME": _SESSION_USER_NAME, "HERMES_SESSION_KEY": _SESSION_KEY, + "HERMES_AGENT_EVENT_ID": _AGENT_EVENT_ID, + "HERMES_AGENT_PERSON_ID": _AGENT_PERSON_ID, "HERMES_CRON_AUTO_DELIVER_PLATFORM": _CRON_AUTO_DELIVER_PLATFORM, "HERMES_CRON_AUTO_DELIVER_CHAT_ID": _CRON_AUTO_DELIVER_CHAT_ID, "HERMES_CRON_AUTO_DELIVER_THREAD_ID": _CRON_AUTO_DELIVER_THREAD_ID, @@ -80,10 +86,13 @@ def set_session_vars( platform: str = "", chat_id: str = "", chat_name: str = "", + chat_type: str = "", thread_id: str = "", user_id: str = "", user_name: str = "", session_key: str = "", + agent_event_id: str = "", + person_id: str = "", ) -> list: """Set all session context variables and return reset tokens. @@ -97,10 +106,13 @@ def set_session_vars( _SESSION_PLATFORM.set(platform), _SESSION_CHAT_ID.set(chat_id), _SESSION_CHAT_NAME.set(chat_name), + _SESSION_CHAT_TYPE.set(chat_type), _SESSION_THREAD_ID.set(thread_id), _SESSION_USER_ID.set(user_id), _SESSION_USER_NAME.set(user_name), _SESSION_KEY.set(session_key), + _AGENT_EVENT_ID.set(agent_event_id), + _AGENT_PERSON_ID.set(person_id), ] return tokens @@ -120,10 +132,13 @@ def clear_session_vars(tokens: list) -> None: _SESSION_PLATFORM, _SESSION_CHAT_ID, _SESSION_CHAT_NAME, + _SESSION_CHAT_TYPE, _SESSION_THREAD_ID, _SESSION_USER_ID, _SESSION_USER_NAME, _SESSION_KEY, + _AGENT_EVENT_ID, + _AGENT_PERSON_ID, ): var.set("") diff --git a/hermes_cli/tools_config.py b/hermes_cli/tools_config.py index 7a9a598f950b..30bee15b94c5 100644 --- a/hermes_cli/tools_config.py +++ b/hermes_cli/tools_config.py @@ -60,6 +60,7 @@ ("skills", "📚 Skills", "list, view, manage"), ("todo", "📋 Task Planning", "todo"), ("memory", "💾 Memory", "persistent memory across sessions"), + ("self_state", "🧭 Self State", "inspect sessions, recent activity, and local crons"), ("session_search", "🔎 Session Search", "search past conversations"), ("clarify", "❓ Clarifying Questions", "clarify"), ("delegation", "👥 Task Delegation", "delegate_task"), diff --git a/hermes_state.py b/hermes_state.py index 0ea9815b5a15..cee0033acc00 100644 --- a/hermes_state.py +++ b/hermes_state.py @@ -21,6 +21,7 @@ import sqlite3 import threading import time +import uuid from pathlib import Path from hermes_constants import get_hermes_home from typing import Any, Callable, Dict, List, Optional, TypeVar @@ -31,7 +32,7 @@ DEFAULT_DB_PATH = get_hermes_home() / "state.db" -SCHEMA_VERSION = 8 +SCHEMA_VERSION = 9 SCHEMA_SQL = """ CREATE TABLE IF NOT EXISTS schema_version ( @@ -91,10 +92,115 @@ value TEXT ); +CREATE TABLE IF NOT EXISTS agent_identities ( + person_id TEXT PRIMARY KEY, + platform TEXT NOT NULL, + platform_user_id TEXT NOT NULL, + display_name TEXT, + authority TEXT, + payload_json TEXT, + first_seen_at REAL NOT NULL, + last_seen_at REAL NOT NULL, + UNIQUE(platform, platform_user_id) +); + +CREATE TABLE IF NOT EXISTS agent_events ( + seq INTEGER PRIMARY KEY AUTOINCREMENT, + event_id TEXT NOT NULL UNIQUE, + session_id TEXT, + session_key TEXT, + actor_id TEXT, + actor_kind TEXT, + source TEXT, + person_id TEXT, + sender_user_id TEXT, + sender_name TEXT, + chat_type TEXT, + audience_type TEXT, + event_type TEXT NOT NULL, + event_subtype TEXT, + status TEXT, + parent_event_id TEXT, + root_event_id TEXT, + correlation_id TEXT, + platform TEXT, + platform_chat_id TEXT, + platform_thread_id TEXT, + platform_message_id TEXT, + platform_update_id TEXT, + tool_name TEXT, + tool_call_id TEXT, + directive_id TEXT, + supersedes_directive_id TEXT, + content TEXT, + payload_json TEXT, + error_json TEXT, + created_at REAL NOT NULL, + updated_at REAL +); + +CREATE TABLE IF NOT EXISTS agent_event_links ( + from_event_id TEXT NOT NULL, + to_event_id TEXT NOT NULL, + link_type TEXT NOT NULL, + created_at REAL NOT NULL, + PRIMARY KEY (from_event_id, to_event_id, link_type) +); + +CREATE TABLE IF NOT EXISTS agent_directives ( + directive_id TEXT PRIMARY KEY, + directive_scope TEXT NOT NULL, + directive_key TEXT NOT NULL, + session_id TEXT, + session_key TEXT, + actor_id TEXT, + issuer_person_id TEXT, + issuer_platform TEXT, + issuer_user_id TEXT, + status TEXT NOT NULL, + active INTEGER NOT NULL DEFAULT 1, + created_event_id TEXT NOT NULL, + supersedes_directive_id TEXT, + superseded_by_directive_id TEXT, + directive_type TEXT, + priority INTEGER DEFAULT 0, + payload_json TEXT NOT NULL, + created_at REAL NOT NULL, + superseded_at REAL, + completed_at REAL +); + CREATE INDEX IF NOT EXISTS idx_sessions_source ON sessions(source); CREATE INDEX IF NOT EXISTS idx_sessions_parent ON sessions(parent_session_id); CREATE INDEX IF NOT EXISTS idx_sessions_started ON sessions(started_at DESC); CREATE INDEX IF NOT EXISTS idx_messages_session ON messages(session_id, timestamp); +CREATE INDEX IF NOT EXISTS idx_agent_identities_platform_user +ON agent_identities(platform, platform_user_id); +CREATE INDEX IF NOT EXISTS idx_agent_events_session_seq +ON agent_events(session_id, seq); +CREATE INDEX IF NOT EXISTS idx_agent_events_session_type_created +ON agent_events(session_id, event_type, created_at); +CREATE INDEX IF NOT EXISTS idx_agent_events_correlation +ON agent_events(correlation_id, seq); +CREATE INDEX IF NOT EXISTS idx_agent_events_parent +ON agent_events(parent_event_id); +CREATE INDEX IF NOT EXISTS idx_agent_events_root +ON agent_events(root_event_id, seq); +CREATE INDEX IF NOT EXISTS idx_agent_events_platform_message +ON agent_events(platform, platform_chat_id, platform_thread_id, platform_message_id); +CREATE INDEX IF NOT EXISTS idx_agent_events_tool_call +ON agent_events(tool_call_id); +CREATE INDEX IF NOT EXISTS idx_agent_events_directive +ON agent_events(directive_id); +CREATE INDEX IF NOT EXISTS idx_agent_events_person +ON agent_events(person_id, created_at); +CREATE INDEX IF NOT EXISTS idx_agent_event_links_to +ON agent_event_links(to_event_id, link_type); +CREATE INDEX IF NOT EXISTS idx_agent_directives_scope +ON agent_directives(directive_scope, directive_key, active); +CREATE UNIQUE INDEX IF NOT EXISTS idx_agent_directives_one_active +ON agent_directives(directive_scope, directive_key, COALESCE(session_key, ''), COALESCE(actor_id, '')) +WHERE active = 1; """ FTS_SQL = """ @@ -356,6 +462,11 @@ def _init_schema(self): except sqlite3.OperationalError: pass # Column already exists cursor.execute("UPDATE schema_version SET version = 8") + if current_version < 9: + # v9: additive agent-actor event log. Tables and indexes are + # created by SCHEMA_SQL above; the version bump records that + # this database supports the actor/directive APIs. + cursor.execute("UPDATE schema_version SET version = 9") # Unique title index — always ensure it exists (safe to run after migrations # since the title column is guaranteed to exist at this point) @@ -1019,6 +1130,383 @@ def _do(conn): return self._execute_write(_do) + # ========================================================================= + # Agent actor event log + # ========================================================================= + + @staticmethod + def _json_dumps_or_none(value: Any) -> Optional[str]: + if value is None: + return None + if isinstance(value, str): + return value + try: + return json.dumps(value, ensure_ascii=False, default=str) + except Exception: + return json.dumps(str(value), ensure_ascii=False) + + def upsert_agent_identity( + self, + *, + platform: str, + platform_user_id: str, + display_name: str = "", + authority: str = "", + payload: Optional[Dict[str, Any]] = None, + person_id: Optional[str] = None, + ) -> str: + """Create/update a per-platform identity and return its person_id.""" + platform = str(platform or "").strip() + platform_user_id = str(platform_user_id or "").strip() + if not platform or not platform_user_id: + raise ValueError("platform and platform_user_id are required") + person_id = person_id or f"{platform}:{platform_user_id}" + now = time.time() + payload_json = self._json_dumps_or_none(payload) + + def _do(conn): + conn.execute( + """INSERT INTO agent_identities + (person_id, platform, platform_user_id, display_name, authority, + payload_json, first_seen_at, last_seen_at) + VALUES (?, ?, ?, ?, ?, ?, ?, ?) + ON CONFLICT(platform, platform_user_id) DO UPDATE SET + display_name = excluded.display_name, + authority = excluded.authority, + payload_json = excluded.payload_json, + last_seen_at = excluded.last_seen_at""", + ( + person_id, + platform, + platform_user_id, + display_name, + authority, + payload_json, + now, + now, + ), + ) + row = conn.execute( + "SELECT person_id FROM agent_identities WHERE platform = ? AND platform_user_id = ?", + (platform, platform_user_id), + ).fetchone() + return row["person_id"] if isinstance(row, sqlite3.Row) else row[0] + + return self._execute_write(_do) + + def append_agent_event( + self, + *, + event_type: str, + event_subtype: str = "", + status: str = "", + event_id: Optional[str] = None, + session_id: str = "", + session_key: str = "", + actor_id: str = "main", + actor_kind: str = "agent", + source: str = "", + person_id: str = "", + sender_user_id: str = "", + sender_name: str = "", + chat_type: str = "", + audience_type: str = "", + parent_event_id: str = "", + root_event_id: str = "", + correlation_id: str = "", + platform: str = "", + platform_chat_id: str = "", + platform_thread_id: str = "", + platform_message_id: str = "", + platform_update_id: str = "", + tool_name: str = "", + tool_call_id: str = "", + directive_id: str = "", + supersedes_directive_id: str = "", + content: str = "", + payload: Optional[Dict[str, Any]] = None, + error: Optional[Dict[str, Any]] = None, + created_at: Optional[float] = None, + ) -> str: + """Append one durable agent-level event. Duplicate event_id is idempotent.""" + event_id = event_id or uuid.uuid4().hex + created_at = float(created_at or time.time()) + root_event_id = root_event_id or parent_event_id or event_id + payload_json = self._json_dumps_or_none(payload) + error_json = self._json_dumps_or_none(error) + + def _do(conn): + conn.execute( + """INSERT OR IGNORE INTO agent_events ( + event_id, session_id, session_key, actor_id, actor_kind, source, + person_id, sender_user_id, sender_name, chat_type, audience_type, + event_type, event_subtype, status, parent_event_id, root_event_id, + correlation_id, platform, platform_chat_id, platform_thread_id, + platform_message_id, platform_update_id, tool_name, tool_call_id, + directive_id, supersedes_directive_id, content, payload_json, + error_json, created_at, updated_at + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""", + ( + event_id, + session_id or None, + session_key or None, + actor_id or None, + actor_kind or None, + source or None, + person_id or None, + sender_user_id or None, + sender_name or None, + chat_type or None, + audience_type or None, + event_type, + event_subtype or None, + status or None, + parent_event_id or None, + root_event_id or None, + correlation_id or None, + platform or None, + platform_chat_id or None, + platform_thread_id or None, + platform_message_id or None, + platform_update_id or None, + tool_name or None, + tool_call_id or None, + directive_id or None, + supersedes_directive_id or None, + content or None, + payload_json, + error_json, + created_at, + created_at, + ), + ) + return event_id + + return self._execute_write(_do) + + def link_agent_events(self, from_event_id: str, to_event_id: str, link_type: str) -> None: + """Record a causal relationship between two agent events.""" + if not from_event_id or not to_event_id or not link_type: + return + + def _do(conn): + conn.execute( + """INSERT OR IGNORE INTO agent_event_links + (from_event_id, to_event_id, link_type, created_at) + VALUES (?, ?, ?, ?)""", + (from_event_id, to_event_id, link_type, time.time()), + ) + + self._execute_write(_do) + + def create_or_replace_agent_directive( + self, + *, + directive_scope: str, + directive_key: str, + directive_type: str, + payload: Dict[str, Any], + session_id: str = "", + session_key: str = "", + actor_id: str = "main", + issuer_person_id: str = "", + issuer_platform: str = "", + issuer_user_id: str = "", + created_event_id: str = "", + priority: int = 0, + ) -> str: + """Create an active directive, superseding any active equivalent.""" + directive_id = uuid.uuid4().hex + created_event_id = created_event_id or uuid.uuid4().hex + now = time.time() + payload_json = self._json_dumps_or_none(payload) or "{}" + + def _do(conn): + old = conn.execute( + """SELECT directive_id, created_event_id FROM agent_directives + WHERE directive_scope = ? AND directive_key = ? + AND COALESCE(session_key, '') = COALESCE(?, '') + AND COALESCE(actor_id, '') = COALESCE(?, '') + AND active = 1 + ORDER BY created_at DESC LIMIT 1""", + (directive_scope, directive_key, session_key or None, actor_id or None), + ).fetchone() + old_id = old["directive_id"] if old else None + old_event_id = old["created_event_id"] if old else None + + conn.execute( + """INSERT OR IGNORE INTO agent_events ( + event_id, session_id, session_key, actor_id, actor_kind, source, + person_id, sender_user_id, event_type, event_subtype, status, + platform, directive_id, supersedes_directive_id, content, + payload_json, created_at, updated_at + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""", + ( + created_event_id, + session_id or None, + session_key or None, + actor_id or None, + "user", + issuer_platform or None, + issuer_person_id or None, + issuer_user_id or None, + "directive", + "create" if old_id is None else "supersede", + "active", + issuer_platform or None, + directive_id, + old_id, + payload.get("text") if isinstance(payload, dict) else None, + payload_json, + now, + now, + ), + ) + + if old_id: + conn.execute( + """UPDATE agent_directives + SET active = 0, status = 'superseded', + superseded_by_directive_id = ?, superseded_at = ? + WHERE directive_id = ?""", + (directive_id, now, old_id), + ) + if old_event_id: + conn.execute( + """INSERT OR IGNORE INTO agent_event_links + (from_event_id, to_event_id, link_type, created_at) + VALUES (?, ?, 'supersedes', ?)""", + (created_event_id, old_event_id, now), + ) + + conn.execute( + """INSERT INTO agent_directives ( + directive_id, directive_scope, directive_key, session_id, + session_key, actor_id, issuer_person_id, issuer_platform, + issuer_user_id, status, active, created_event_id, + supersedes_directive_id, directive_type, priority, + payload_json, created_at + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, 'active', 1, ?, ?, ?, ?, ?, ?)""", + ( + directive_id, + directive_scope, + directive_key, + session_id or None, + session_key or None, + actor_id or None, + issuer_person_id or None, + issuer_platform or None, + issuer_user_id or None, + created_event_id, + old_id, + directive_type, + int(priority or 0), + payload_json, + now, + ), + ) + return directive_id + + return self._execute_write(_do) + + def list_active_agent_directives( + self, + *, + actor_id: str = "main", + directive_type: str = "", + limit: int = 20, + ) -> List[Dict[str, Any]]: + """Return active directives for state packet/action policy use.""" + with self._lock: + if directive_type: + cursor = self._conn.execute( + """SELECT * FROM agent_directives + WHERE active = 1 + AND COALESCE(actor_id, '') = COALESCE(?, '') + AND directive_type = ? + ORDER BY priority DESC, created_at DESC LIMIT ?""", + (actor_id or None, directive_type, int(limit)), + ) + else: + cursor = self._conn.execute( + """SELECT * FROM agent_directives + WHERE active = 1 + AND COALESCE(actor_id, '') = COALESCE(?, '') + ORDER BY priority DESC, created_at DESC LIMIT ?""", + (actor_id or None, int(limit)), + ) + rows = cursor.fetchall() + directives = [] + for row in rows: + item = dict(row) + if item.get("payload_json"): + try: + item["payload"] = json.loads(item["payload_json"]) + except Exception: + item["payload"] = {} + directives.append(item) + return directives + + def list_recent_agent_events( + self, + *, + event_type: str = "", + session_key: str = "", + person_id: str = "", + limit: int = 20, + ) -> List[Dict[str, Any]]: + """Return recent agent events ordered newest-first.""" + clauses = [] + params: List[Any] = [] + if event_type: + clauses.append("event_type = ?") + params.append(event_type) + if session_key: + clauses.append("session_key = ?") + params.append(session_key) + if person_id: + clauses.append("person_id = ?") + params.append(person_id) + where = ("WHERE " + " AND ".join(clauses)) if clauses else "" + params.append(int(limit)) + with self._lock: + cursor = self._conn.execute( + f"SELECT * FROM agent_events {where} ORDER BY seq DESC LIMIT ?", + params, + ) + rows = cursor.fetchall() + events = [] + for row in rows: + item = dict(row) + for key in ("payload_json", "error_json"): + if item.get(key): + try: + item[key[:-5]] = json.loads(item[key]) + except Exception: + pass + events.append(item) + return events + + def get_agent_event(self, event_id: str) -> Optional[Dict[str, Any]]: + """Fetch one agent event by id.""" + if not event_id: + return None + with self._lock: + row = self._conn.execute( + "SELECT * FROM agent_events WHERE event_id = ?", + (event_id,), + ).fetchone() + if not row: + return None + item = dict(row) + for key in ("payload_json", "error_json"): + if item.get(key): + try: + item[key[:-5]] = json.loads(item[key]) + except Exception: + pass + return item + def get_messages(self, session_id: str) -> List[Dict[str, Any]]: """Load all messages for a session, ordered by timestamp.""" with self._lock: @@ -1588,4 +2076,3 @@ def maybe_auto_prune_and_vacuum( result["error"] = str(exc) return result - diff --git a/tests/gateway/test_agent_actor.py b/tests/gateway/test_agent_actor.py new file mode 100644 index 000000000000..31f8c8429c93 --- /dev/null +++ b/tests/gateway/test_agent_actor.py @@ -0,0 +1,222 @@ +import json + +from gateway.agent_actor import ( + build_state_packet, + detect_public_broadcast_stop_directive, + evaluate_send_message_policy, + infer_platform_authority, + maybe_record_directive_from_inbound, + owner_user_ids_for_platform, + record_inbound_event, +) +from gateway.config import Platform +from gateway.session import SessionSource +from gateway.session_context import clear_session_vars, set_session_vars +from hermes_state import SessionDB + + +def test_detects_public_broadcast_stop_directive(): + directive = detect_public_broadcast_stop_directive("Is this a cron? Turn this off") + + assert directive is not None + assert directive["behavior"] == "suppress" + + +def test_state_packet_is_sender_scoped(tmp_path): + db = SessionDB(db_path=tmp_path / "state.db") + source = SessionSource( + platform=Platform.DISCORD, + chat_id="149", + chat_type="group", + user_id="141", + user_name="hands", + ) + event_id, person_id = record_inbound_event( + db, + source=source, + session_id="sid", + session_key="agent:main:discord:group:149:141", + text="hello", + authority="trusted", + ) + + packet = build_state_packet( + db, + source=source, + session_id="sid", + session_key="agent:main:discord:group:149:141", + inbound_event_id=event_id, + person_id=person_id, + authority="trusted", + ) + + assert "person_id: discord:141" in packet + assert "authority: trusted" in packet + assert "session_key: agent:main:discord:group:149:141" in packet + db.close() + + +def test_owner_authority_uses_generated_soul_owner_block(tmp_path, monkeypatch): + hermes_home = tmp_path / ".hermes" + hermes_home.mkdir() + (hermes_home / "SOUL.md").write_text( + "## Your owner\n" + "- Discord: `@handsdiff` (user_id `1417636184355766305`)\n" + "\n## Peer roster\n" + "- Someone else user_id `999999999999999999`\n", + encoding="utf-8", + ) + monkeypatch.setenv("HERMES_HOME", str(hermes_home)) + monkeypatch.setenv("DISCORD_ALLOW_ALL_USERS", "true") + monkeypatch.delenv("DISCORD_OWNER_USER_ID", raising=False) + monkeypatch.delenv("DISCORD_OWNER_USER_IDS", raising=False) + + source = SessionSource( + platform=Platform.DISCORD, + chat_id="1495468809216327702", + chat_type="group", + user_id="1417636184355766305", + user_name="hands", + ) + + assert owner_user_ids_for_platform("discord") == {"1417636184355766305"} + assert infer_platform_authority(source) == "owner" + + +def test_owner_state_packet_includes_recent_cross_session_events(tmp_path): + db = SessionDB(db_path=tmp_path / "state.db") + group_source = SessionSource( + platform=Platform.DISCORD, + chat_id="general", + chat_type="group", + user_id="141", + user_name="hands", + ) + record_inbound_event( + db, + source=group_source, + session_id="group-sid", + session_key="agent:main:discord:group:general:141", + text="hey from general", + authority="owner", + ) + dm_source = SessionSource( + platform=Platform.DISCORD, + chat_id="dm", + chat_type="dm", + user_id="141", + user_name="hands", + ) + dm_event_id, person_id = record_inbound_event( + db, + source=dm_source, + session_id="dm-sid", + session_key="agent:main:discord:dm:dm", + text="do you see general?", + authority="owner", + ) + + packet = build_state_packet( + db, + source=dm_source, + session_id="dm-sid", + session_key="agent:main:discord:dm:dm", + inbound_event_id=dm_event_id, + person_id=person_id, + authority="owner", + ) + + assert "authority: owner" in packet + assert "Recent Runtime Events" in packet + assert "hey from general" in packet + db.close() + + +def test_directive_blocks_public_cross_session_send(tmp_path): + db = SessionDB(db_path=tmp_path / "state.db") + source = SessionSource( + platform=Platform.DISCORD, + chat_id="general", + chat_type="group", + user_id="141", + user_name="hands", + ) + event_id, person_id = record_inbound_event( + db, + source=source, + session_id="sid", + session_key="agent:main:discord:group:general:141", + text="Is this a cron? Turn this off", + authority="trusted", + ) + maybe_record_directive_from_inbound( + db, + source=source, + session_id="sid", + session_key="agent:main:discord:group:general:141", + inbound_event_id=event_id, + person_id=person_id, + text="Is this a cron? Turn this off", + authority="trusted", + ) + + tokens = set_session_vars( + platform="hub", + chat_id="hub:speculator", + chat_type="dm", + user_id="speculator", + user_name="speculator", + session_key="agent:main:hub:dm:hub:speculator", + agent_event_id=event_id, + person_id="hub:speculator", + ) + try: + decision = evaluate_send_message_policy( + target_platform="discord", + target_chat_id="general", + message="Market digest", + db=db, + ) + finally: + clear_session_vars(tokens) + db.close() + + assert decision.allowed is False + assert decision.policy == "suppress_public_broadcasts" + + +def test_cron_like_hub_inbound_blocks_public_rebroadcast(tmp_path): + db = SessionDB(db_path=tmp_path / "state.db") + event_id = db.append_agent_event( + event_type="inbound", + event_subtype="message", + status="received", + session_key="agent:main:hub:dm:hub:speculator", + source="hub", + platform="hub", + platform_chat_id="hub:speculator", + content="Cronjob Response: synthetic market digest", + ) + tokens = set_session_vars( + platform="hub", + chat_id="hub:speculator", + chat_type="dm", + user_id="speculator", + user_name="speculator", + session_key="agent:main:hub:dm:hub:speculator", + agent_event_id=event_id, + person_id="hub:speculator", + ) + try: + decision = evaluate_send_message_policy( + target_platform="discord", + target_chat_id="1495468809216327702", + message="Synthetic market digest", + db=db, + ) + finally: + clear_session_vars(tokens) + db.close() + + assert decision.allowed is False + assert decision.policy == "autonomous_public_rebroadcast_guard" diff --git a/tests/gateway/test_routing_context_user_id.py b/tests/gateway/test_routing_context_user_id.py index 1686e4b098c1..c3dcdb380c4a 100644 --- a/tests/gateway/test_routing_context_user_id.py +++ b/tests/gateway/test_routing_context_user_id.py @@ -51,6 +51,15 @@ def test_discord_group_exposes_user_id(self): assert ctx["user_id"] == "1417636184355766305" assert ctx["platform"] == "discord" + def test_discord_group_owner_user_id_logs_owner_source_kind(self, monkeypatch): + from gateway.run import GatewayRunner + monkeypatch.setenv("DISCORD_OWNER_USER_ID", "1417636184355766305") + runner = _make_runner() + ctx = GatewayRunner._build_routing_context( + runner, _src("discord", user_id="1417636184355766305") + ) + assert ctx["source_kind"] == "owner" + def test_telegram_dm_exposes_user_id(self): from gateway.run import GatewayRunner runner = _make_runner() diff --git a/tests/test_hermes_state.py b/tests/test_hermes_state.py index f405cf8bd51b..78328ebc383d 100644 --- a/tests/test_hermes_state.py +++ b/tests/test_hermes_state.py @@ -1169,11 +1169,15 @@ def test_tables_exist(self, db): assert "sessions" in tables assert "messages" in tables assert "schema_version" in tables + assert "agent_events" in tables + assert "agent_event_links" in tables + assert "agent_directives" in tables + assert "agent_identities" in tables def test_schema_version(self, db): cursor = db._conn.execute("SELECT version FROM schema_version") version = cursor.fetchone()[0] - assert version == 8 + assert version == 9 def test_title_column_exists(self, db): """Verify the title column was created in the sessions table.""" @@ -1229,12 +1233,12 @@ def test_migration_from_v2(self, tmp_path): conn.commit() conn.close() - # Open with SessionDB — should migrate to v8 + # Open with SessionDB — should migrate to current schema migrated_db = SessionDB(db_path=db_path) # Verify migration cursor = migrated_db._conn.execute("SELECT version FROM schema_version") - assert cursor.fetchone()[0] == 8 + assert cursor.fetchone()[0] == 9 # Verify title column exists and is NULL for existing sessions session = migrated_db.get_session("existing") @@ -1255,6 +1259,79 @@ def test_migration_from_v2(self, tmp_path): migrated_db.close() +class TestAgentActorEventLog: + def test_identity_upsert_is_per_platform(self, db): + p1 = db.upsert_agent_identity( + platform="discord", + platform_user_id="123", + display_name="A", + authority="trusted", + ) + p2 = db.upsert_agent_identity( + platform="telegram", + platform_user_id="123", + display_name="A", + authority="trusted", + ) + + assert p1 == "discord:123" + assert p2 == "telegram:123" + assert p1 != p2 + + def test_append_and_fetch_agent_event(self, db): + event_id = db.append_agent_event( + event_id="evt-1", + event_type="inbound", + event_subtype="message", + status="received", + session_id="sid", + session_key="agent:main:discord:group:1:123", + platform="discord", + platform_chat_id="1", + sender_user_id="123", + person_id="discord:123", + content="hello", + payload={"authority": "trusted"}, + ) + + assert event_id == "evt-1" + event = db.get_agent_event("evt-1") + assert event["event_type"] == "inbound" + assert event["payload"]["authority"] == "trusted" + assert db.list_recent_agent_events(event_type="inbound")[0]["event_id"] == "evt-1" + + def test_directive_supersession(self, db): + first = db.create_or_replace_agent_directive( + directive_scope="actor", + directive_key="public-broadcast-suppression", + directive_type="suppress_public_broadcasts", + payload={"text": "stop posting digests"}, + actor_id="main", + issuer_person_id="discord:1", + issuer_platform="discord", + issuer_user_id="1", + ) + second = db.create_or_replace_agent_directive( + directive_scope="actor", + directive_key="public-broadcast-suppression", + directive_type="suppress_public_broadcasts", + payload={"text": "turn this off"}, + actor_id="main", + issuer_person_id="discord:1", + issuer_platform="discord", + issuer_user_id="1", + ) + + active = db.list_active_agent_directives(actor_id="main") + assert [d["directive_id"] for d in active] == [second] + old = db._conn.execute( + "SELECT active, superseded_by_directive_id FROM agent_directives WHERE directive_id = ?", + (first,), + ).fetchone() + assert old["active"] == 0 + assert old["superseded_by_directive_id"] == second + + class TestTitleUniqueness: """Tests for unique title enforcement and title-based lookups.""" @@ -1911,4 +1988,3 @@ def test_state_meta_survives_vacuum(self, db): assert marker is not None # Should parse as a float timestamp close to now. assert abs(float(marker) - time.time()) < 60 - diff --git a/tests/tools/test_registry.py b/tests/tools/test_registry.py index d015b483864a..a8d836fbc51a 100644 --- a/tests/tools/test_registry.py +++ b/tests/tools/test_registry.py @@ -308,6 +308,7 @@ def test_matches_previous_manual_builtin_tool_set(self): "tools.process_registry", "tools.rl_training_tool", "tools.send_message_tool", + "tools.self_state_tool", "tools.session_search_tool", "tools.skill_manager_tool", "tools.skills_tool", diff --git a/tests/tools/test_self_state_tool.py b/tests/tools/test_self_state_tool.py new file mode 100644 index 000000000000..5aa1cb45ced7 --- /dev/null +++ b/tests/tools/test_self_state_tool.py @@ -0,0 +1,124 @@ +import json +import sys +from types import SimpleNamespace + +from tools.self_state_tool import self_state_tool +from tools.registry import registry +from toolsets import resolve_toolset + + +def test_self_state_lists_sessions(tmp_path, monkeypatch): + monkeypatch.setenv("HERMES_HOME", str(tmp_path)) + sessions_dir = tmp_path / "sessions" + sessions_dir.mkdir() + (sessions_dir / "sessions.json").write_text(json.dumps({ + "agent:main:discord:group:123": { + "session_id": "20260423_120000_abcd", + "platform": "discord", + "chat_type": "group", + "display_name": "#general", + "updated_at": "2026-04-23T12:03:00", + "created_at": "2026-04-23T12:00:00", + "origin": { + "platform": "discord", + "chat_id": "123", + "chat_name": "#general", + "user_id": "u1", + "user_name": "hands", + }, + } + }), encoding="utf-8") + + result = self_state_tool(action="sessions") + + assert result["action"] == "sessions" + assert result["sessions"][0]["session_id"] == "20260423_120000_abcd" + assert result["sessions"][0]["platform"] == "discord" + assert result["sessions"][0]["chat_name"] == "#general" + + +def test_self_state_lists_local_crons(tmp_path, monkeypatch): + monkeypatch.setenv("HERMES_HOME", str(tmp_path)) + cron_dir = tmp_path / "cron" + cron_dir.mkdir() + (cron_dir / "jobs.json").write_text(json.dumps({ + "jobs": [{ + "id": "job-1", + "name": "Market monitor", + "enabled": True, + "deliver": "origin", + "schedule": {"display": "every 10m"}, + "origin": {"platform": "hub", "chat_id": "hub:sal"}, + "next_run": "2026-04-23T12:10:00", + }] + }), encoding="utf-8") + + result = self_state_tool(action="crons") + + assert result["action"] == "crons" + assert result["crons"][0]["id"] == "job-1" + assert result["crons"][0]["deliver"] == "origin" + assert result["crons"][0]["origin"]["chat_id"] == "hub:sal" + + +def test_self_state_recent_activity_uses_session_db(tmp_path, monkeypatch): + monkeypatch.setenv("HERMES_HOME", str(tmp_path)) + sessions_dir = tmp_path / "sessions" + sessions_dir.mkdir() + (sessions_dir / "sessions.json").write_text(json.dumps({ + "agent:main:hub:dm:hub:codex-cron": { + "session_id": "sid-1", + "platform": "hub", + "chat_type": "dm", + "updated_at": "2026-04-23T12:03:00", + "origin": { + "platform": "hub", + "chat_id": "hub:codex-cron", + "user_id": "codex-cron", + "user_name": "codex-cron", + }, + } + }), encoding="utf-8") + + class FakeDB: + def list_sessions_rich(self, **_kwargs): + return [{"id": "sid-1", "source": "hub", "last_active": 1}] + + def get_messages(self, session_id): + assert session_id == "sid-1" + return [ + { + "timestamp": 1, + "role": "assistant", + "content": "Posting an update to #general", + "tool_calls": [{"function": {"name": "send_message"}}], + } + ] + + def close(self): + pass + + monkeypatch.setitem(sys.modules, "hermes_state", SimpleNamespace(SessionDB=FakeDB)) + + result = self_state_tool(action="recent_activity", session_filter="codex-cron") + + assert result["activity"][0]["session_id"] == "sid-1" + assert result["activity"][0]["source"] == "hub:codex-cron" + assert result["activity"][0]["tool_calls"] == ["send_message"] + + +def test_self_state_registry_dispatch_returns_json_string(tmp_path, monkeypatch): + monkeypatch.setenv("HERMES_HOME", str(tmp_path)) + sessions_dir = tmp_path / "sessions" + sessions_dir.mkdir() + (sessions_dir / "sessions.json").write_text(json.dumps({}), encoding="utf-8") + + result = registry.dispatch("self_state", {"action": "sessions"}) + + assert isinstance(result, str) + assert json.loads(result)["action"] == "sessions" + + +def test_self_state_is_in_core_toolsets(): + assert "self_state" in resolve_toolset("hermes-cli") + assert "self_state" in resolve_toolset("hermes-discord") diff --git a/tests/tools/test_send_message_tool.py b/tests/tools/test_send_message_tool.py index 626179de19b7..050fffc58c49 100644 --- a/tests/tools/test_send_message_tool.py +++ b/tests/tools/test_send_message_tool.py @@ -65,6 +65,38 @@ def _ensure_slack_mock(monkeypatch): class TestSendMessageTool: + def test_policy_block_skips_send(self): + discord_cfg = SimpleNamespace(enabled=True, token="***", extra={}) + config = SimpleNamespace( + platforms={Platform.DISCORD: discord_cfg}, + get_home_channel=lambda _platform: None, + ) + + from gateway.agent_actor import GateDecision + + with patch("gateway.config.load_gateway_config", return_value=config), \ + patch("tools.interrupt.is_interrupted", return_value=False), \ + patch("gateway.agent_actor.evaluate_send_message_policy", return_value=GateDecision( + allowed=False, + policy="autonomous_public_rebroadcast_guard", + reason="blocked", + event_id="evt-blocked", + )), \ + patch("tools.send_message_tool._send_to_platform", new=AsyncMock(return_value={"success": True})) as send_mock: + result = json.loads( + send_message_tool( + { + "action": "send", + "target": "discord:1495468809216327702", + "message": "market digest", + } + ) + ) + + assert result["blocked"] is True + assert result["policy"] == "autonomous_public_rebroadcast_guard" + send_mock.assert_not_awaited() + def test_cron_duplicate_target_is_skipped_and_explained(self): home = SimpleNamespace(chat_id="-1001") config, _telegram_cfg = _make_config() diff --git a/tools/self_state_tool.py b/tools/self_state_tool.py new file mode 100644 index 000000000000..4508aba9b8cc --- /dev/null +++ b/tools/self_state_tool.py @@ -0,0 +1,501 @@ +#!/usr/bin/env python3 +"""Self-state introspection tool. + +Gives the agent a compact, factual view of its own runtime state across +isolated gateway sessions. This is intentionally read-only: it helps the +model diagnose "what am I doing?" without mutating crons, memory, or sessions. +""" + +from __future__ import annotations + +import json +import time +from pathlib import Path +from typing import Any, Dict, Iterable, List, Optional + +from hermes_constants import get_hermes_home + + +def _clamp_int(value: Any, default: int, minimum: int, maximum: int) -> int: + try: + parsed = int(value) + except (TypeError, ValueError): + parsed = default + return max(minimum, min(maximum, parsed)) + + +def _load_json(path: Path, default: Any) -> Any: + try: + if not path.exists(): + return default + return json.loads(path.read_text(encoding="utf-8")) + except Exception: + return default + + +def _format_ts(ts: Any) -> Optional[str]: + if ts in (None, ""): + return None + try: + return time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime(float(ts))) + except (TypeError, ValueError, OSError): + return str(ts) + + +def _preview(text: Any, limit: int = 240) -> str: + if text is None: + return "" + s = str(text).replace("\r", " ").replace("\n", " ").strip() + if len(s) <= limit: + return s + return s[: limit - 3].rstrip() + "..." + + +def _session_entries(limit: int) -> List[Dict[str, Any]]: + sessions_path = get_hermes_home() / "sessions" / "sessions.json" + data = _load_json(sessions_path, {}) + if not isinstance(data, dict): + return [] + + entries: List[Dict[str, Any]] = [] + for key, raw in data.items(): + if not isinstance(raw, dict): + continue + origin = raw.get("origin") if isinstance(raw.get("origin"), dict) else {} + entries.append({ + "session_key": key, + "session_id": raw.get("session_id", ""), + "platform": raw.get("platform") or origin.get("platform", ""), + "chat_type": raw.get("chat_type") or origin.get("chat_type", ""), + "chat_id": origin.get("chat_id", ""), + "chat_name": origin.get("chat_name") or raw.get("display_name", ""), + "user_id": origin.get("user_id", ""), + "user_name": origin.get("user_name", ""), + "thread_id": origin.get("thread_id"), + "updated_at": raw.get("updated_at", ""), + "created_at": raw.get("created_at", ""), + }) + + entries.sort(key=lambda e: str(e.get("updated_at") or ""), reverse=True) + return entries[:limit] + + +def _session_index_by_id() -> Dict[str, Dict[str, Any]]: + entries = _session_entries(1000) + return { + str(entry.get("session_id")): entry + for entry in entries + if entry.get("session_id") + } + + +def _recent_activity_from_db(limit: int, session_filter: str = "") -> List[Dict[str, Any]]: + try: + from hermes_state import SessionDB + except Exception: + return [] + + db = None + try: + db = SessionDB() + session_index = _session_index_by_id() + filter_text = session_filter.lower() + sessions = db.list_sessions_rich(limit=max(limit * 3, limit), include_children=True) + rows: List[Dict[str, Any]] = [] + for session in sessions: + session_id = session.get("id") or session.get("session_id") + if not session_id: + continue + index_entry = session_index.get(str(session_id), {}) + session_haystack = " ".join( + str(value or "") + for value in ( + session_id, + session.get("source", ""), + index_entry.get("session_key", ""), + index_entry.get("platform", ""), + index_entry.get("chat_id", ""), + index_entry.get("chat_name", ""), + index_entry.get("user_id", ""), + index_entry.get("user_name", ""), + ) + ).lower() + if filter_text and filter_text not in session_haystack: + continue + try: + messages = db.get_messages(session_id) + except Exception: + continue + for msg in messages[-8:]: + role = msg.get("role") + tool_calls = msg.get("tool_calls") or [] + tool_names = _tool_names(tool_calls) + tool_details = _tool_call_details(tool_calls) + content = msg.get("content") + if role not in ("user", "assistant", "tool") and not tool_names: + continue + row = { + "timestamp": _format_ts(msg.get("timestamp")), + "session_id": session_id, + "session_key": index_entry.get("session_key", ""), + "source": index_entry.get("chat_id") or session.get("source", ""), + "platform": index_entry.get("platform") or session.get("source", ""), + "user_name": index_entry.get("user_name", ""), + "role": role, + "tool_name": msg.get("tool_name"), + "tool_calls": tool_names, + "content_preview": _preview(content), + } + if tool_details: + row["tool_call_details"] = tool_details + rows.append(row) + rows.sort(key=lambda r: str(r.get("timestamp") or ""), reverse=True) + return rows[:limit] + except Exception: + return [] + finally: + if db is not None: + try: + db.close() + except Exception: + pass + + +def _tool_names(tool_calls: Any) -> List[str]: + if not isinstance(tool_calls, list): + return [] + names: List[str] = [] + for call in tool_calls: + if not isinstance(call, dict): + continue + name = call.get("name") + if not name and isinstance(call.get("function"), dict): + name = call["function"].get("name") + if name: + names.append(str(name)) + return names + + +def _tool_call_details(tool_calls: Any) -> List[Dict[str, Any]]: + if not isinstance(tool_calls, list): + return [] + details: List[Dict[str, Any]] = [] + for call in tool_calls: + if not isinstance(call, dict): + continue + function = call.get("function") if isinstance(call.get("function"), dict) else {} + name = call.get("name") or function.get("name") + if not name: + continue + raw_args = call.get("arguments") + if raw_args is None: + raw_args = function.get("arguments") + args_summary: Dict[str, Any] = {} + if isinstance(raw_args, str) and raw_args.strip(): + try: + parsed = json.loads(raw_args) + except (TypeError, ValueError): + args_summary["arguments_preview"] = _preview(raw_args) + else: + if isinstance(parsed, dict): + for key in ("action", "target", "message", "query", "job_id"): + if key in parsed: + args_summary[key] = _preview(parsed.get(key), 180) + else: + args_summary["arguments_preview"] = _preview(parsed) + elif isinstance(raw_args, dict): + for key in ("action", "target", "message", "query", "job_id"): + if key in raw_args: + args_summary[key] = _preview(raw_args.get(key), 180) + details.append({"name": str(name), **args_summary}) + return details + + +def _jobs_iter(data: Any) -> Iterable[Dict[str, Any]]: + if isinstance(data, list): + for item in data: + if isinstance(item, dict): + yield item + return + if isinstance(data, dict): + raw_jobs = data.get("jobs") + if isinstance(raw_jobs, list): + for item in raw_jobs: + if isinstance(item, dict): + yield item + else: + for item in data.values(): + if isinstance(item, dict): + yield item + + +def _cron_jobs(limit: int) -> List[Dict[str, Any]]: + jobs_path = get_hermes_home() / "cron" / "jobs.json" + data = _load_json(jobs_path, []) + jobs: List[Dict[str, Any]] = [] + for job in _jobs_iter(data): + origin = job.get("origin") if isinstance(job.get("origin"), dict) else {} + schedule = job.get("schedule") if isinstance(job.get("schedule"), dict) else {} + jobs.append({ + "id": job.get("id", ""), + "name": job.get("name", ""), + "enabled": bool(job.get("enabled", True)), + "deliver": job.get("deliver", "local"), + "origin": { + "platform": origin.get("platform"), + "chat_id": origin.get("chat_id"), + "chat_name": origin.get("chat_name"), + "thread_id": origin.get("thread_id"), + } if origin else None, + "schedule": job.get("schedule_display") or schedule.get("display") or job.get("schedule"), + "next_run": job.get("next_run"), + "last_run": job.get("last_run"), + }) + jobs.sort(key=lambda j: (not j.get("enabled", False), str(j.get("next_run") or ""))) + return jobs[:limit] + + +def _outbound_events(limit: int) -> List[Dict[str, Any]]: + rows = _recent_activity_from_db(limit=limit * 4) + events: List[Dict[str, Any]] = [] + for row in rows: + tool_calls = row.get("tool_calls") or [] + if "send_message" not in tool_calls: + continue + events.append({ + "timestamp": row.get("timestamp"), + "session_id": row.get("session_id"), + "source": row.get("source"), + "kind": "send_message_tool_call", + "content_preview": row.get("content_preview", ""), + "tool_calls": tool_calls, + "tool_call_details": row.get("tool_call_details", []), + }) + if len(events) >= limit: + break + if len(events) < limit: + events.extend(_mirror_events_from_jsonl(limit - len(events))) + events.sort(key=lambda r: str(r.get("timestamp") or ""), reverse=True) + return events[:limit] + + +def _active_directives(limit: int) -> List[Dict[str, Any]]: + try: + from hermes_state import SessionDB + except Exception: + return [] + db = None + try: + db = SessionDB() + rows = db.list_active_agent_directives(actor_id="main", limit=limit) + result = [] + for row in rows: + payload = row.get("payload") or {} + result.append({ + "directive_id": row.get("directive_id"), + "directive_type": row.get("directive_type"), + "directive_key": row.get("directive_key"), + "priority": row.get("priority"), + "created_at": _format_ts(row.get("created_at")), + "issuer": row.get("issuer_person_id") or row.get("issuer_user_id"), + "text": payload.get("text", ""), + }) + return result + except Exception: + return [] + finally: + if db is not None: + try: + db.close() + except Exception: + pass + + +def _agent_events(limit: int) -> List[Dict[str, Any]]: + try: + from hermes_state import SessionDB + except Exception: + return [] + db = None + try: + db = SessionDB() + rows = db.list_recent_agent_events(limit=limit) + return [ + { + "event_id": row.get("event_id"), + "event_type": row.get("event_type"), + "event_subtype": row.get("event_subtype"), + "status": row.get("status"), + "source": row.get("source"), + "person_id": row.get("person_id"), + "platform": row.get("platform"), + "chat_id": row.get("platform_chat_id"), + "session_key": row.get("session_key"), + "created_at": _format_ts(row.get("created_at")), + "content_preview": _preview(row.get("content")), + } + for row in rows + ] + except Exception: + return [] + finally: + if db is not None: + try: + db.close() + except Exception: + pass + + +def _mirror_events_from_jsonl(limit: int) -> List[Dict[str, Any]]: + if limit <= 0: + return [] + sessions_path = get_hermes_home() / "sessions" + entries = _session_entries(50) + events: List[Dict[str, Any]] = [] + for entry in entries: + session_id = entry.get("session_id") + if not session_id: + continue + transcript = sessions_path / f"{session_id}.jsonl" + if not transcript.exists(): + continue + try: + lines = transcript.read_text(encoding="utf-8").splitlines() + except Exception: + continue + for line in reversed(lines[-100:]): + try: + msg = json.loads(line) + except Exception: + continue + if not isinstance(msg, dict) or not msg.get("mirror"): + continue + events.append({ + "timestamp": msg.get("timestamp"), + "session_id": session_id, + "source": entry.get("platform", ""), + "kind": "delivery_mirror", + "mirror_source": msg.get("mirror_source", ""), + "content_preview": _preview(msg.get("content")), + }) + if len(events) >= limit: + return events + return events + + +def self_state_tool(action: str = "summary", limit: int = 10, session_filter: str = "", **_) -> Dict[str, Any]: + """Return a read-only snapshot of the agent's runtime state.""" + action = (action or "summary").strip().lower() + limit = _clamp_int(limit, default=10, minimum=1, maximum=50) + session_filter = str(session_filter or "").strip() + + if action == "sessions": + return { + "action": "sessions", + "sessions": _session_entries(limit), + "note": "These are separate conversation contexts for the same agent runtime.", + } + if action == "recent_activity": + return { + "action": "recent_activity", + "activity": _recent_activity_from_db(limit, session_filter=session_filter), + "note": "Use this to inspect what the agent recently saw or did across sessions.", + } + if action == "crons": + return { + "action": "crons", + "crons": _cron_jobs(limit), + "note": "This only lists cron jobs on this VM. Jobs delegated to other agents live on their VMs.", + } + if action == "outbounds": + return { + "action": "outbounds", + "outbounds": _outbound_events(limit), + "note": "Best-effort view based on persisted assistant/tool-call history.", + } + if action == "directives": + return { + "action": "directives", + "directives": _active_directives(limit), + "note": "Active agent-level directives are shared across isolated sessions.", + } + if action == "events": + return { + "action": "events", + "events": _agent_events(limit), + "note": "Recent agent-level runtime events across sessions.", + } + if action != "summary": + return { + "error": f"Unknown action {action!r}. Use summary, sessions, recent_activity, crons, outbounds, directives, or events.", + } + + return { + "action": "summary", + "sessions": _session_entries(min(limit, 10)), + "recent_activity": _recent_activity_from_db(min(limit, 12), session_filter=session_filter), + "crons": _cron_jobs(min(limit, 12)), + "outbounds": _outbound_events(min(limit, 8)), + "directives": _active_directives(min(limit, 8)), + "agent_events": _agent_events(min(limit, 8)), + "notes": [ + "Sessions are isolated for prompt context, but this tool provides a shared runtime view.", + "Remote agents' local crons are not visible here unless they report through your sessions.", + ], + } + + +SELF_STATE_SCHEMA = { + "name": "self_state", + "description": ( + "Inspect your own runtime across isolated sessions. Use this when the user asks " + "what you are doing, why you posted or messaged something, what sessions/channels " + "are active, what local crons exist, or whether recent behavior came from another " + "conversation context. Prefer this before terminal, session_search, logs, or cron " + "tools for runtime self-awareness. Read-only." + ), + "parameters": { + "type": "object", + "properties": { + "action": { + "type": "string", + "enum": ["summary", "sessions", "recent_activity", "crons", "outbounds", "directives", "events"], + "description": "Which self-state view to return. Use summary first unless you need a narrower view.", + "default": "summary", + }, + "limit": { + "type": "integer", + "description": "Maximum entries per section, 1-50.", + "default": 10, + }, + "session_filter": { + "type": "string", + "description": "Optional substring filter for session id/source/chat/user in recent_activity.", + }, + }, + "required": [], + }, +} + + +def _check_self_state_requirements() -> bool: + return get_hermes_home().exists() + + +from tools.registry import registry # noqa: E402 + +registry.register( + name="self_state", + toolset="self_state", + schema=SELF_STATE_SCHEMA, + handler=lambda args, **kw: json.dumps( + self_state_tool( + action=args.get("action", "summary"), + limit=args.get("limit", 10), + session_filter=args.get("session_filter", ""), + ), + ensure_ascii=False, + ), + check_fn=_check_self_state_requirements, + emoji="", +) diff --git a/tools/send_message_tool.py b/tools/send_message_tool.py index 19da4f55af8d..010e6bce5e30 100644 --- a/tools/send_message_tool.py +++ b/tools/send_message_tool.py @@ -169,6 +169,31 @@ def _handle_send(args): else: is_explicit = False + try: + from gateway.agent_actor import ( + blocked_tool_result, + evaluate_send_message_policy, + record_send_message_outbound, + ) + except Exception: + blocked_tool_result = None + evaluate_send_message_policy = None + record_send_message_outbound = None + + def _policy_decision(current_chat_id, current_message): + if not evaluate_send_message_policy: + return None + try: + return evaluate_send_message_policy( + target_platform=platform_name, + target_chat_id=current_chat_id, + target_thread_id=thread_id or "", + message=current_message, + ) + except Exception: + logger.debug("send_message policy evaluation failed", exc_info=True) + return None + # Resolve human-friendly channel names to numeric IDs if target_ref and not is_explicit: try: @@ -191,6 +216,19 @@ def _handle_send(args): if is_interrupted(): return tool_error("Interrupted") + policy_checked = False + if chat_id: + duplicate_skip = _maybe_skip_cron_duplicate_send(platform_name, chat_id, thread_id) + if duplicate_skip: + return json.dumps(duplicate_skip) + + if blocked_tool_result: + decision = _policy_decision(chat_id, message) + if decision and not decision.allowed: + return blocked_tool_result(decision) + if decision: + policy_checked = True + try: from gateway.config import load_gateway_config, Platform config = load_gateway_config() @@ -271,6 +309,29 @@ def _handle_send(args): if duplicate_skip: return json.dumps(duplicate_skip) + if not policy_checked and blocked_tool_result: + decision = _policy_decision(chat_id, cleaned_message) + if decision and not decision.allowed: + return blocked_tool_result(decision) + + try: + from gateway.agent_actor import ( + blocked_tool_result, + evaluate_send_message_policy, + record_send_message_outbound, + ) + + decision = evaluate_send_message_policy( + target_platform=platform_name, + target_chat_id=chat_id, + target_thread_id=thread_id or "", + message=cleaned_message, + ) + if not decision.allowed: + return blocked_tool_result(decision) + except Exception: + record_send_message_outbound = None + try: from model_tools import _run_async result = _run_async( @@ -286,6 +347,19 @@ def _handle_send(args): if used_home_channel and isinstance(result, dict) and result.get("success"): result["note"] = f"Sent to {platform_name} home channel (chat_id: {chat_id})" + if record_send_message_outbound: + try: + record_send_message_outbound( + target_platform=platform_name, + target_chat_id=chat_id, + target_thread_id=thread_id or "", + message=cleaned_message, + status="succeeded" if isinstance(result, dict) and result.get("success") else "failed", + result=result if isinstance(result, dict) else {"result": result}, + ) + except Exception: + pass + # Mirror the sent message into the target's gateway session if isinstance(result, dict) and result.get("success") and mirror_text: try: diff --git a/toolsets.py b/toolsets.py index f1dc7fca1c1e..c7b812cc59c8 100644 --- a/toolsets.py +++ b/toolsets.py @@ -48,6 +48,8 @@ "text_to_speech", # Planning & memory "todo", "memory", + # Runtime self-inspection across isolated sessions + "self_state", # Session history search "session_search", # Clarifying questions @@ -167,6 +169,12 @@ "tools": ["memory"], "includes": [] }, + + "self_state": { + "description": "Read-only runtime self-inspection across sessions, local crons, and recent activity", + "tools": ["self_state"], + "includes": [] + }, "session_search": { "description": "Search and recall past conversations with summarization", @@ -250,7 +258,7 @@ "browser_type", "browser_scroll", "browser_back", "browser_press", "browser_get_images", "browser_vision", "browser_console", "browser_cdp", - "todo", "memory", + "todo", "memory", "self_state", "session_search", "execute_code", "delegate_task", ], @@ -276,7 +284,7 @@ "browser_press", "browser_get_images", "browser_vision", "browser_console", "browser_cdp", # Planning & memory - "todo", "memory", + "todo", "memory", "self_state", # Session history search "session_search", # Code execution + delegation