diff --git a/agent/restart_awareness.py b/agent/restart_awareness.py new file mode 100644 index 000000000000..9b203022d672 --- /dev/null +++ b/agent/restart_awareness.py @@ -0,0 +1,120 @@ +""" +Restart awareness: persistent activity tracker for gateway restarts. + +Writes and reads a JSON activity file so the agent can recover context +after an unplanned or intentional gateway restart. +""" + +import json +import os +from datetime import datetime, timezone, timedelta +from pathlib import Path +from typing import Dict, List, Optional + +from hermes_constants import get_hermes_home + + +def _activity_path() -> Path: + return get_hermes_home() / "state" / "current_activity.json" + + +def _now() -> str: + return datetime.now(timezone.utc).isoformat() + + +def update_activity( + current_task: str, + files_modified: Optional[List[str]] = None, + last_action: Optional[str] = None, + next_expected_step: Optional[str] = None, + mode: str = "simple", +) -> None: + """Write current activity state to disk.""" + path = _activity_path() + path.parent.mkdir(parents=True, exist_ok=True) + data = { + "current_task": current_task, + "files_modified": files_modified or [], + "last_action": last_action or "", + "next_expected_step": next_expected_step or "", + "mode": mode, + "updated_at": _now(), + } + path.write_text(json.dumps(data, indent=2)) + + +def read_activity() -> Optional[Dict]: + """Read the last known activity state, or None if none exists.""" + path = _activity_path() + if not path.exists(): + return None + try: + return json.loads(path.read_text()) + except (json.JSONDecodeError, OSError): + return None + + +def clear_activity() -> None: + """Clear the activity file after a clean handoff.""" + path = _activity_path() + if path.exists(): + path.unlink() + + +def _compute_staleness(updated_at_str: Optional[str]) -> tuple[bool, str]: + """Return (is_stale, age_text) given an ISO timestamp string.""" + if not updated_at_str: + return False, "" + try: + updated_at = datetime.fromisoformat(updated_at_str.replace("Z", "+00:00")) + if updated_at.tzinfo is None: + updated_at = updated_at.replace(tzinfo=timezone.utc) + age = datetime.now(timezone.utc) - updated_at + is_stale = age > timedelta(minutes=30) + total_minutes = age.total_seconds() // 60 + if total_minutes >= 1440: + age_text = f"{int(total_minutes // 1440)}d {int((total_minutes % 1440) // 60)}h old" + elif total_minutes >= 60: + age_text = f"{int(total_minutes // 60)}h {int(total_minutes % 60)}m old" + else: + age_text = f"{int(total_minutes)}m old" + return is_stale, age_text + except Exception: + return False, "" + + +def build_handoff(activity: Dict) -> str: + """Build the handoff text to inject on first message after restart.""" + mode = activity.get("mode", "simple") + updated_at_str = activity.get("updated_at") + is_stale, age_text = _compute_staleness(updated_at_str) + + handoff_lines = [] + if mode == "verbose": + if is_stale: + handoff_lines.append( + f"[Restart handoff \u2014 STALE ({age_text})] Activity below is from {updated_at_str or 'unknown time'}. " + "Do NOT auto-execute the next step. Summarize the task and ask the user whether to continue." + ) + else: + handoff_lines.append( + "[Restart handoff \u2014 FRESH] Agent restarted while working on the task below. Resume immediately." + ) + handoff_lines.append(f"Task: {activity.get('current_task', '?')}") + if activity.get("files_modified"): + handoff_lines.append(f"Files touched: {', '.join(activity['files_modified'])}") + if activity.get("last_action"): + handoff_lines.append(f"Last action: {activity['last_action']}") + if activity.get("next_expected_step"): + handoff_lines.append(f"Next step: {activity['next_expected_step']}") + else: + if is_stale: + handoff_lines.append( + f"[Restart handoff \u2014 STALE ({age_text})] " + "Back after gateway restart. Last recorded activity is old \u2014 ask the user what to do next." + ) + else: + handoff_lines.append( + "[Restart handoff \u2014 FRESH] Back after gateway restart." + ) + return "\n".join(handoff_lines) diff --git a/gateway/config.py b/gateway/config.py index 6b09b34d18b7..70d5b91ec28d 100644 --- a/gateway/config.py +++ b/gateway/config.py @@ -230,6 +230,7 @@ class SessionResetPolicy: mode: str = "both" # "daily", "idle", "both", or "none" at_hour: int = 4 # Hour for daily reset (0-23, local time) idle_minutes: int = 1440 # Minutes of inactivity before reset (24 hours) + max_duration_minutes: Optional[int] = None # Hard max session age, regardless of mode notify: bool = True # Send a notification to the user when auto-reset occurs notify_exclude_platforms: tuple = ("api_server", "webhook") # Platforms that don't get reset notifications @@ -238,6 +239,7 @@ def to_dict(self) -> Dict[str, Any]: "mode": self.mode, "at_hour": self.at_hour, "idle_minutes": self.idle_minutes, + "max_duration_minutes": self.max_duration_minutes, "notify": self.notify, "notify_exclude_platforms": list(self.notify_exclude_platforms), } @@ -248,12 +250,14 @@ def from_dict(cls, data: Dict[str, Any]) -> "SessionResetPolicy": mode = data.get("mode") at_hour = data.get("at_hour") idle_minutes = data.get("idle_minutes") + max_duration_minutes = data.get("max_duration_minutes") notify = data.get("notify") exclude = data.get("notify_exclude_platforms") return cls( mode=mode if mode is not None else "both", at_hour=at_hour if at_hour is not None else 4, idle_minutes=idle_minutes if idle_minutes is not None else 1440, + max_duration_minutes=max_duration_minutes if max_duration_minutes is not None else None, notify=_coerce_bool(notify, True), notify_exclude_platforms=tuple(exclude) if exclude is not None else ("api_server", "webhook"), ) diff --git a/gateway/platforms/base.py b/gateway/platforms/base.py index 3e8c1433e6b2..86d5109ea015 100644 --- a/gateway/platforms/base.py +++ b/gateway/platforms/base.py @@ -1305,6 +1305,20 @@ def _mark_connected(self) -> None: self._fatal_error_message = None self._fatal_error_retryable = True self._write_runtime_status_safe("connected", platform_state="connected", error_code=None, error_message=None) + # Restart awareness: read persisted activity state if present. + try: + from agent.restart_awareness import read_activity, build_handoff + activity = read_activity() + if activity: + self._pending_restart_handoff = build_handoff(activity) + logger.info( + "[%s] restart handoff \u2014 task=%s mode=%s", + self.name, + activity.get("current_task", "?"), + activity.get("mode", "simple"), + ) + except Exception: + pass def _mark_disconnected(self) -> None: self._running = False diff --git a/gateway/platforms/matrix.py b/gateway/platforms/matrix.py index 12e840b69c4e..94c9668e68b0 100644 --- a/gateway/platforms/matrix.py +++ b/gateway/platforms/matrix.py @@ -384,12 +384,11 @@ def __init__(self, config: PlatformConfig): "MATRIX_REACTIONS", "true" ).lower() not in ("false", "0", "no") self._pending_reactions: dict[tuple[str, str], str] = {} - # Delay before redacting reactions so Matrix homeservers have time to - # deliver the final message event without tripping "missing event" - # errors in some clients. 5s is empirically safe; not user-tunable — - # if that changes, add a config.yaml entry rather than an env var. - self._reaction_redaction_delay_seconds = 5.0 - self._reaction_redaction_tasks: Set[asyncio.Task] = set() + + # Presence state tracking: map actual activity to Matrix presence. + # online = actively processing, unavailable = idle/connected, offline = disconnected. + self._presence_active_count: int = 0 + self._presence_current_state: str = "offline" # Proxy support — resolve once at init, reuse for all HTTP traffic. self._proxy_url: str | None = resolve_proxy_url(platform_env_var="MATRIX_PROXY") @@ -746,6 +745,34 @@ async def connect(self) -> bool: try: await olm.verify_with_recovery_key(recovery_key) logger.info("Matrix: cross-signing verified via recovery key") + + # Attempt to self-sign our own device with the recovered SSK. + # This is required for Element to show the green shield + # (device cross-signed by owner). + try: + if client.device_id: + own_device = await olm.get_or_fetch_device( + client.mxid, client.device_id + ) + if own_device: + await olm.sign_own_device(own_device) + logger.info( + "Matrix: successfully self-signed device %s with cross-signing key", + client.device_id, + ) + else: + logger.warning( + "Matrix: could not retrieve own device identity for self-signing" + ) + else: + logger.warning( + "Matrix: no device_id set, cannot self-sign device" + ) + except Exception as exc: + logger.warning( + "Matrix: self-signing device with cross-signing key failed: %s", + exc, + ) except Exception as exc: logger.warning( "Matrix: recovery key verification failed: %s", exc @@ -864,6 +891,7 @@ async def connect(self) -> bool: # Start the sync loop. self._sync_task = asyncio.create_task(self._sync_loop()) self._mark_connected() + await self.set_presence("online") return True async def disconnect(self) -> None: @@ -877,14 +905,6 @@ async def disconnect(self) -> None: except (asyncio.CancelledError, Exception): pass - redaction_tasks = list(self._reaction_redaction_tasks) - for task in redaction_tasks: - if not task.done(): - task.cancel() - if redaction_tasks: - await asyncio.gather(*redaction_tasks, return_exceptions=True) - self._reaction_redaction_tasks.clear() - # Close the SQLite crypto store database. if hasattr(self, "_crypto_db") and self._crypto_db: try: @@ -893,6 +913,12 @@ async def disconnect(self) -> None: logger.debug("Matrix: could not close crypto DB on disconnect: %s", exc) if self._client: + try: + self._presence_active_count = 0 + self._presence_current_state = "offline" + await self.set_presence("offline") + except Exception: + pass try: await self._client.api.session.close() except Exception: @@ -1975,73 +2001,45 @@ async def _redact_reaction( """Remove a reaction by redacting its event.""" return await self.redact_message(room_id, reaction_event_id, reason) - def _schedule_reaction_redaction( - self, - room_id: str, - reaction_event_id: str, - reason: str = "", - ) -> None: - """Redact a reaction after a short delay so message delivery settles.""" - - async def _redact_later() -> None: - try: - if self._reaction_redaction_delay_seconds: - await asyncio.sleep(self._reaction_redaction_delay_seconds) - if not await self._redact_reaction(room_id, reaction_event_id, reason): - logger.debug( - "Matrix: failed to redact reaction %s", reaction_event_id - ) - except asyncio.CancelledError: - raise - except Exception as exc: - logger.debug( - "Matrix: delayed reaction redaction failed for %s: %s", - reaction_event_id, - exc, - ) - - task = asyncio.create_task(_redact_later()) - self._reaction_redaction_tasks.add(task) - task.add_done_callback(self._reaction_redaction_tasks.discard) - async def on_processing_start(self, event: MessageEvent) -> None: - """Add eyes reaction when the agent starts processing a message.""" - if not self._reactions_enabled: - return - msg_id = event.message_id - room_id = event.source.chat_id - if msg_id and room_id: - reaction_event_id = await self._send_reaction(room_id, msg_id, "\U0001f440") - if reaction_event_id: - self._pending_reactions[(room_id, msg_id)] = reaction_event_id + """Add eyes reaction and set presence to unavailable when actively working.""" + if self._reactions_enabled: + msg_id = event.message_id + room_id = event.source.chat_id + if msg_id and room_id: + reaction_event_id = await self._send_reaction(room_id, msg_id, "\U0001f440") + if reaction_event_id: + self._pending_reactions[(room_id, msg_id)] = reaction_event_id + # Track active processing count and transition presence. + self._presence_active_count += 1 + if self._presence_active_count == 1 and self._presence_current_state != "unavailable": + await self.set_presence("unavailable", status_msg="working on Hermes") async def on_processing_complete( self, event: MessageEvent, outcome: ProcessingOutcome, ) -> None: - """Replace eyes with checkmark (success) or cross (failure).""" - if not self._reactions_enabled: - return - msg_id = event.message_id - room_id = event.source.chat_id - if not msg_id or not room_id: - return - if outcome == ProcessingOutcome.CANCELLED: - return - reaction_key = (room_id, msg_id) - if reaction_key in self._pending_reactions: - eyes_event_id = self._pending_reactions.pop(reaction_key) - self._schedule_reaction_redaction( - room_id, - eyes_event_id, - "processing complete", - ) - await self._send_reaction( - room_id, - msg_id, - "\u2705" if outcome == ProcessingOutcome.SUCCESS else "\u274c", - ) + """Replace eyes with checkmark (success) or cross (failure), and transition presence back to idle.""" + if self._reactions_enabled: + msg_id = event.message_id + room_id = event.source.chat_id + if msg_id and room_id: + if outcome != ProcessingOutcome.CANCELLED: + reaction_key = (room_id, msg_id) + if reaction_key in self._pending_reactions: + eyes_event_id = self._pending_reactions.pop(reaction_key) + if not await self._redact_reaction(room_id, eyes_event_id): + logger.debug("Matrix: failed to redact eyes reaction %s", eyes_event_id) + await self._send_reaction( + room_id, + msg_id, + "\u2705" if outcome == ProcessingOutcome.SUCCESS else "\u274c", + ) + # Track active processing count and transition presence. + self._presence_active_count = max(0, self._presence_active_count - 1) + if self._presence_active_count == 0 and self._presence_current_state != "online": + await self.set_presence("online") async def _on_reaction(self, event: Any) -> None: """Handle incoming reaction events.""" @@ -2115,8 +2113,11 @@ async def _redact_bot_approval_reactions( ) -> None: """Redact the bot's seed ✅/❎ reactions, leaving only the user's reaction.""" for emoji, evt_id in prompt.bot_reaction_events.items(): - self._schedule_reaction_redaction(room_id, evt_id, "approval resolved") - logger.debug("Matrix: scheduled bot reaction redaction %s (%s)", emoji, evt_id) + try: + await self.redact_message(room_id, evt_id, "approval resolved") + logger.debug("Matrix: redacted bot reaction %s (%s)", emoji, evt_id) + except Exception as exc: + logger.debug("Matrix: failed to redact bot reaction %s: %s", emoji, exc) # ------------------------------------------------------------------ # Text message aggregation (handles Matrix client-side splits) diff --git a/gateway/session.py b/gateway/session.py index be393e48e6fc..3d73f552c42e 100644 --- a/gateway/session.py +++ b/gateway/session.py @@ -779,6 +779,12 @@ def _is_session_expired(self, entry: SessionEntry) -> bool: if entry.updated_at < today_reset: return True + # Hard TTL: max session duration regardless of mode + if policy.max_duration_minutes is not None: + age_minutes = (now - entry.created_at).total_seconds() / 60 + if age_minutes > policy.max_duration_minutes: + return True + return False def _should_reset(self, entry: SessionEntry, source: SessionSource) -> Optional[str]: @@ -800,11 +806,17 @@ def _should_reset(self, entry: SessionEntry, source: SessionSource) -> Optional[ session_type=source.chat_type ) + now = _now() + + # Hard TTL: max session duration regardless of mode + if policy.max_duration_minutes is not None: + age_minutes = (now - entry.created_at).total_seconds() / 60 + if age_minutes > policy.max_duration_minutes: + return "max_duration" + if policy.mode == "none": return None - now = _now() - if policy.mode in ("idle", "both"): idle_deadline = entry.updated_at + timedelta(minutes=policy.idle_minutes) if now > idle_deadline: diff --git a/hermes_cli/plugins.py b/hermes_cli/plugins.py index 2171e6d50dc4..927ae5491d59 100644 --- a/hermes_cli/plugins.py +++ b/hermes_cli/plugins.py @@ -92,6 +92,8 @@ def get_bundled_plugins_dir() -> Path: "on_session_end", "on_session_finalize", "on_session_reset", + "on_turn_start", + "on_turn_end", "subagent_stop", # Gateway pre-dispatch hook. Fired once per incoming MessageEvent # after the internal-event guard but BEFORE auth/pairing and agent diff --git a/plugins/observability/local_sqlite_telemetry/__init__.py b/plugins/observability/local_sqlite_telemetry/__init__.py new file mode 100644 index 000000000000..9d75738f263b --- /dev/null +++ b/plugins/observability/local_sqlite_telemetry/__init__.py @@ -0,0 +1,297 @@ +"""local-sqlite-telemetry — Hermes plugin for local telemetry storage. + +Writes LLM calls, tool calls, context pressure, and session summaries +to a SQLite DB at ~/.hermes/profiles//telemetry.db. + +Declare hooks in plugin.yaml: + pre_llm_call, post_llm_call, pre_tool_call, post_tool_call, + on_session_start, on_session_end, on_turn_start, on_turn_end +""" +from __future__ import annotations + +import hashlib +import json +import logging +import os +import sqlite3 +import threading +import time +from typing import Any, Dict, List, Optional + +logger = logging.getLogger(__name__) + +# ── per-profile DB path ──────────────────────────────────────────────────── +_DB_PATH: Optional[str] = None + + +def _get_db_path() -> str: + global _DB_PATH + if _DB_PATH is None: + hermes_home = os.environ.get("HERMES_HOME", os.path.expanduser("~/.hermes")) + # Try profile-specific first, fall back to root + profile = os.path.basename(hermes_home) if ".hermes" in hermes_home else None + if profile and os.path.isdir(os.path.join(os.path.expanduser("~/.hermes"), "profiles", profile)): + _DB_PATH = os.path.join(hermes_home, "telemetry.db") + else: + _DB_PATH = os.path.join(os.path.expanduser("~/.hermes"), "profiles", "phoenix", "telemetry.db") + return _DB_PATH + + +def _ensure_schema(conn: sqlite3.Connection) -> None: + conn.executescript(""" +CREATE TABLE IF NOT EXISTS llm_calls ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + timestamp INTEGER NOT NULL DEFAULT (strftime('%s','now')), + session_id TEXT, task_id TEXT, model TEXT, provider TEXT, + prompt_tokens INTEGER, completion_tokens INTEGER, total_tokens INTEGER, + cache_read_tokens INTEGER, cache_write_tokens INTEGER, reasoning_tokens INTEGER, + cost_usd REAL, duration_ms INTEGER, status TEXT DEFAULT 'success' +); +CREATE INDEX IF NOT EXISTS idx_llm_session ON llm_calls(session_id); +CREATE INDEX IF NOT EXISTS idx_llm_task ON llm_calls(task_id); +CREATE INDEX IF NOT EXISTS idx_llm_time ON llm_calls(timestamp); + +CREATE TABLE IF NOT EXISTS tool_calls ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + timestamp INTEGER NOT NULL DEFAULT (strftime('%s','now')), + session_id TEXT, task_id TEXT, tool_name TEXT NOT NULL, + status TEXT DEFAULT 'success', duration_ms INTEGER, + args_hash TEXT, error_message TEXT, token_cost INTEGER +); +CREATE INDEX IF NOT EXISTS idx_tool_session ON tool_calls(session_id); +CREATE INDEX IF NOT EXISTS idx_tool_task ON tool_calls(task_id); +CREATE INDEX IF NOT EXISTS idx_tool_name ON tool_calls(tool_name); +CREATE INDEX IF NOT EXISTS idx_tool_time ON tool_calls(timestamp); + +CREATE TABLE IF NOT EXISTS context_pressure ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + timestamp INTEGER NOT NULL DEFAULT (strftime('%s','now')), + session_id TEXT, turn_number INTEGER, + context_used_chars INTEGER, context_limit_chars INTEGER, + utilization_pct REAL, compression_triggered INTEGER DEFAULT 0, model TEXT +); +CREATE INDEX IF NOT EXISTS idx_ctx_session ON context_pressure(session_id); +CREATE INDEX IF NOT EXISTS idx_ctx_time ON context_pressure(timestamp); + +CREATE TABLE IF NOT EXISTS session_summary ( + session_id TEXT PRIMARY KEY, platform TEXT, + start_time INTEGER, end_time INTEGER, + total_llm_calls INTEGER DEFAULT 0, total_tool_calls INTEGER DEFAULT 0, + total_prompt_tokens INTEGER DEFAULT 0, total_completion_tokens INTEGER DEFAULT 0, + total_cost_usd REAL DEFAULT 0.0, final_status TEXT, user_canonical_name TEXT +); +CREATE INDEX IF NOT EXISTS idx_ss_time ON session_summary(start_time); +""") + conn.commit() + + +def _conn() -> sqlite3.Connection: + db = _get_db_path() + conn = sqlite3.connect(db, check_same_thread=False) + _ensure_schema(conn) + return conn + + +# ── in-memory state (per session) ────────────────────────────────────────── +# We track turn-level counters because hooks fire multiple times per turn. +class _State: + __slots__ = ("lock", "turn_tool_calls", "turn_llm_calls", "turn_start_time", + "context_limit", "context_used", "turn_number", "session_platform", + "session_user", "model", "task_id") + def __init__(self): + self.lock = threading.Lock() + self.turn_tool_calls = 0 + self.turn_llm_calls = 0 + self.turn_start_time = 0.0 + self.context_limit = 0 + self.context_used = 0 + self.turn_number = 0 + self.session_platform = "" + self.session_user = "" + self.model = "" + self.task_id = "" + + +_STATE: Dict[str, _State] = {} + + +def _state(sid: str) -> _State: + if sid not in _STATE: + _STATE[sid] = _State() + return _STATE[sid] + + +def _safe_insert(table: str, fields: List[str], values: List[Any]) -> None: + try: + with _conn() as conn: + placeholders = ",".join("?" * len(fields)) + conn.execute( + f"INSERT INTO {table} ({','.join(fields)}) VALUES ({placeholders})", + values, + ) + conn.commit() + except Exception as exc: + logger.debug("Telemetry insert failed: %s", exc) + + +def _update_session(sid: str, delta_cost: float = 0.0, + delta_prompt: int = 0, delta_completion: int = 0, + llm_calls: int = 0, tool_calls: int = 0) -> None: + try: + with _conn() as conn: + conn.execute(""" + INSERT INTO session_summary (session_id, total_llm_calls, total_tool_calls, + total_prompt_tokens, total_completion_tokens, total_cost_usd) + VALUES (?, ?, ?, ?, ?, ?) + ON CONFLICT(session_id) DO UPDATE SET + total_llm_calls = total_llm_calls + excluded.total_llm_calls, + total_tool_calls = total_tool_calls + excluded.total_tool_calls, + total_prompt_tokens = total_prompt_tokens + excluded.total_prompt_tokens, + total_completion_tokens = total_completion_tokens + excluded.total_completion_tokens, + total_cost_usd = total_cost_usd + excluded.total_cost_usd + """, (sid, llm_calls, tool_calls, delta_prompt, delta_completion, delta_cost)) + conn.commit() + except Exception as exc: + logger.debug("Telemetry session update failed: %s", exc) + + +def _args_hash(args: Any) -> str: + try: + return hashlib.md5(json.dumps(args, sort_keys=True, default=str).encode()).hexdigest() + except Exception: + return "" + + +# ── hook callbacks ────────────────────────────────────────────────────────── + +def on_session_start(session_id: str = "", model: str = "", platform: str = "", + sender_id: str = "", **_kw) -> None: + st = _state(session_id) + with st.lock: + st.session_platform = platform or "" + st.session_user = sender_id or "" + st.model = model or "" + st.turn_number = 0 + try: + with _conn() as conn: + conn.execute( + "INSERT INTO session_summary (session_id, platform, start_time, user_canonical_name) " + "VALUES (?, ?, ?, ?) ON CONFLICT(session_id) DO NOTHING", + (session_id, platform or "", int(time.time()), sender_id or ""), + ) + conn.commit() + except Exception as exc: + logger.debug("Telemetry on_session_start failed: %s", exc) + + +def pre_llm_call(session_id: str = "", model: str = "", task_id: str = "", **_kw) -> None: + st = _state(session_id) + with st.lock: + st.model = model or st.model + st.turn_start_time = time.time() + if task_id: + st.task_id = task_id + + +def post_llm_call(session_id: str = "", model: str = "", response_meta: dict = None, + total_tokens: int = 0, prompt_tokens: int = 0, + completion_tokens: int = 0, cache_read: int = 0, + cache_write: int = 0, reasoning_tokens: int = 0, + task_id: str = "", **_kw) -> None: + st = _state(session_id) + with st.lock: + st.turn_llm_calls += 1 + duration_ms = int((time.time() - st.turn_start_time) * 1000) if st.turn_start_time else 0 + _safe_insert("llm_calls", [ + "session_id", "task_id", "model", "prompt_tokens", "completion_tokens", "total_tokens", + "cache_read_tokens", "cache_write_tokens", "reasoning_tokens", + "duration_ms", "status", + ], [ + session_id, task_id or getattr(st, "task_id", "") or "", model or st.model, + prompt_tokens or 0, completion_tokens or 0, total_tokens or 0, + cache_read or 0, cache_write or 0, reasoning_tokens or 0, + duration_ms, "success", + ]) + _update_session(session_id, delta_prompt=prompt_tokens or 0, + delta_completion=completion_tokens or 0, llm_calls=1) + + +def pre_tool_call(tool_name: str = "", session_id: str = "", task_id: str = "", **_kw) -> None: + st = _state(session_id) + with st.lock: + st.turn_tool_calls += 1 + st.turn_start_time = time.time() + if task_id: + st.task_id = task_id + + +def post_tool_call(tool_name: str = "", session_id: str = "", result: Any = None, + error: str = "", duration_ms: int = 0, args: Any = None, + task_id: str = "", **_kw) -> None: + st = _state(session_id) + # Check for real errors: either an explicit error param from the caller, + # or a JSON result with a non-null error field. Many tools (e.g. terminal) + # return {"error": null} on success — the word "error" in the string + # is not enough to flag failure. + has_error = bool(error) + if not has_error and isinstance(result, str): + try: + parsed = json.loads(result) + if isinstance(parsed, dict) and parsed.get("error"): + has_error = True + except (json.JSONDecodeError, ValueError): + pass + _safe_insert("tool_calls", [ + "session_id", "task_id", "tool_name", "status", "duration_ms", "args_hash", "error_message", + ], [ + session_id, task_id or getattr(st, "task_id", "") or "", tool_name, + "failure" if has_error else "success", + duration_ms, _args_hash(args), error or "", + ]) + _update_session(session_id, tool_calls=1) + + +def on_turn_start(session_id: str = "", turn_number: int = 0, + context_length: int = 0, context_limit: int = 0, + compression_triggered: bool = False, + model: str = "", **_kw) -> None: + st = _state(session_id) + with st.lock: + st.turn_number = turn_number + st.context_limit = context_limit + st.context_used = context_length + st.model = model or st.model + utilization = round((context_length / context_limit) * 100, 2) if context_limit else 0.0 + _safe_insert("context_pressure", [ + "session_id", "turn_number", "context_used_chars", "context_limit_chars", + "utilization_pct", "compression_triggered", "model", + ], [ + session_id, turn_number, context_length, context_limit, + utilization, 1 if compression_triggered else 0, model or st.model, + ]) + + +def register(ctx) -> None: + ctx.register_hook("on_session_start", on_session_start) + ctx.register_hook("on_session_end", on_session_end) + ctx.register_hook("pre_tool_call", pre_tool_call) + ctx.register_hook("post_tool_call", post_tool_call) + ctx.register_hook("pre_llm_call", pre_llm_call) + ctx.register_hook("post_llm_call", post_llm_call) + ctx.register_hook("on_turn_start", on_turn_start) + + +def on_session_end(session_id: str = "", interrupted: bool = False, + completed: bool = False, **_kw) -> None: + try: + with _conn() as conn: + status = "interrupted" if interrupted else ("completed" if completed else "unknown") + conn.execute( + "UPDATE session_summary SET end_time = ?, final_status = ? WHERE session_id = ?", + (int(time.time()), status, session_id), + ) + conn.commit() + except Exception as exc: + logger.debug("Telemetry on_session_end failed: %s", exc) + finally: + _STATE.pop(session_id, None) diff --git a/plugins/observability/local_sqlite_telemetry/plugin.yaml b/plugins/observability/local_sqlite_telemetry/plugin.yaml new file mode 100644 index 000000000000..a9ab637c49f6 --- /dev/null +++ b/plugins/observability/local_sqlite_telemetry/plugin.yaml @@ -0,0 +1,12 @@ +name: local_sqlite_telemetry +version: "1.0.0" +description: "Local SQLite telemetry storage for Hermes — traces LLM calls, tool usage, context pressure, and session summaries to ~/.hermes/profiles//telemetry.db. Zero external dependencies, zero credentials." +author: Phoenix +hooks: + - on_session_start + - on_session_end + - pre_tool_call + - post_tool_call + - pre_llm_call + - post_llm_call + - on_turn_start diff --git a/plugins/observability/local_sqlite_telemetry/telemetry_cli.py b/plugins/observability/local_sqlite_telemetry/telemetry_cli.py new file mode 100644 index 000000000000..265d18c15b1d --- /dev/null +++ b/plugins/observability/local_sqlite_telemetry/telemetry_cli.py @@ -0,0 +1,136 @@ +#!/usr/bin/env python3 +"""Telemetry CLI — query ~/.hermes/profiles/phoenix/telemetry.db + +Usage: + python telemetry_cli.py report [--days 7] + python telemetry_cli.py tools [--days 7] + python telemetry_cli.py cost [--days 7] + python telemetry_cli.py sessions [--days 7] +""" +import sqlite3, os, sys, argparse, json +from datetime import datetime, timedelta + +_DEFAULT_DB = os.path.expanduser("~/.hermes/profiles/phoenix/telemetry.db") + +def _db(): + return sqlite3.connect(os.environ.get("TELEMETRY_DB", _DEFAULT_DB)) + +def _since(days: int): + return int((datetime.now() - timedelta(days=days)).timestamp()) + +def report(days: int = 7): + since = _since(days) + with _db() as conn: + row = conn.execute(""" + SELECT COUNT(*), SUM(total_llm_calls), SUM(total_tool_calls), + SUM(total_prompt_tokens), SUM(total_completion_tokens), + SUM(total_cost_usd) + FROM session_summary + WHERE start_time >= ? + """, (since,)).fetchone() + sessions, llm, tools, prompt, completion, cost = row + cost = cost or 0.0 + conn.execute(""" + SELECT tool_name, COUNT(*), + SUM(CASE WHEN status='failure' THEN 1 ELSE 0 END), + ROUND(AVG(duration_ms),0) + FROM tool_calls + WHERE timestamp >= ? + GROUP BY tool_name + ORDER BY COUNT(*) DESC + LIMIT 10 + """, (since,)) + top_tools = conn.fetchall() + conn.execute(""" + SELECT ROUND(AVG(utilization_pct),1), MAX(utilization_pct), + SUM(compression_triggered) + FROM context_pressure + WHERE timestamp >= ? + """, (since,)) + avg_ctx, max_ctx, compress = conn.fetchone() + lines = [ + f"Telemetry Report (last {days} days)", + f"Sessions: {sessions}", + f"LLM calls: {llm}", + f"Tool calls: {tools}", + f"Tokens: prompt={prompt} completion={completion}", + f"Estimated cost: ${cost:.4f}", + "", + "Top tools:", + ] + for name, cnt, fails, avg_ms in top_tools: + lines.append(f" {name}: {cnt} calls, {fails} failures, avg {avg_ms}ms") + lines.extend([ + "", + f"Context pressure: avg={avg_ctx}% max={max_ctx}% compressions={compress}", + ]) + print("\n".join(lines)) + +def tools(days: int = 7): + since = _since(days) + with _db() as conn: + rows = conn.execute(""" + SELECT tool_name, COUNT(*) total, + SUM(CASE WHEN status='failure' THEN 1 ELSE 0 END) fails, + ROUND(100.0 * SUM(CASE WHEN status='failure' THEN 1 ELSE 0 END) / COUNT(*), 1) fail_pct, + ROUND(AVG(duration_ms),0) avg_ms, + MAX(duration_ms) max_ms + FROM tool_calls + WHERE timestamp >= ? + GROUP BY tool_name + ORDER BY total DESC + """, (since,)).fetchall() + print(f"{'Tool':<25} {'Calls':>6} {'Fails':>6} {'Fail%':>6} {'AvgMs':>8} {'MaxMs':>8}") + print("-" * 65) + for name, total, fails, fail_pct, avg_ms, max_ms in rows: + print(f"{name:<25} {total:>6} {fails:>6} {fail_pct:>6} {avg_ms:>8} {max_ms:>8}") + +def cost_breakdown(days: int = 7): + since = _since(days) + with _db() as conn: + rows = conn.execute(""" + SELECT session_id, platform, total_llm_calls, total_prompt_tokens, + total_completion_tokens, total_cost_usd, final_status + FROM session_summary + WHERE start_time >= ? + ORDER BY total_cost_usd DESC + """, (since,)).fetchall() + print(f"{'Session':<36} {'Platform':<10} {'LLM':>5} {'Prompt':>8} {'Complete':>8} {'Cost':>8} {'Status'}") + print("-" * 95) + total = 0.0 + for sid, plat, llm, prompt, comp, cost, status in rows: + total += cost or 0.0 + print(f"{sid:<36} {plat:<10} {llm:>5} {prompt:>8} {comp:>8} ${cost:>7.4f} {status}") + print(f"\nTotal estimated cost: ${total:.4f}") + +def sessions(days: int = 7): + since = _since(days) + with _db() as conn: + rows = conn.execute(""" + SELECT session_id, platform, start_time, end_time, + total_llm_calls, total_tool_calls, total_cost_usd, final_status + FROM session_summary + WHERE start_time >= ? + ORDER BY start_time DESC + """, (since,)).fetchall() + print(f"{'Session':<36} {'Platform':<10} {'Start':<20} {'LLM':>4} {'Tools':>5} {'Cost':>8} {'Status'}") + print("-" * 100) + for sid, plat, st, et, llm, tools, cost, status in rows: + start = datetime.fromtimestamp(st).strftime("%Y-%m-%d %H:%M") if st else "" + print(f"{sid:<36} {plat:<10} {start:<20} {llm:>4} {tools:>5} ${cost:>7.4f} {status}") + +def main(): + parser = argparse.ArgumentParser(description="Telemetry CLI for Hermes") + sub = parser.add_subparsers(dest="cmd") + for name, fn in [("report", report), ("tools", tools), ("cost", cost_breakdown), ("sessions", sessions)]: + p = sub.add_parser(name) + p.add_argument("--days", type=int, default=7) + p.set_defaults(func=fn) + args = parser.parse_args() + if not getattr(args, "func", None): + parser.print_help() + return + args.func(args.days) + +if __name__ == "__main__": + main() diff --git a/plugins/observability/local_sqlite_telemetry/weekly_digest.py b/plugins/observability/local_sqlite_telemetry/weekly_digest.py new file mode 100644 index 000000000000..968d167b1b36 --- /dev/null +++ b/plugins/observability/local_sqlite_telemetry/weekly_digest.py @@ -0,0 +1,211 @@ +#!/usr/bin/env python3 +"""Weekly Agent Digest Report — reads telemetry.db, generates structured report.""" + +import sqlite3 +import os +import sys +from datetime import datetime, timedelta +from collections import Counter + +DB_PATH = os.path.expanduser("~/.hermes/profiles/phoenix/telemetry.db") + +def fmt_tok(n): + if n >= 1_000_000: + return f"{n/1_000_000:.2f}M" + if n >= 1_000: + return f"{n/1_000:.1f}K" + return str(n) + +def fmt_cost(c): + if c is None: + return "unknown" + return f"${c:.4f}" if c > 0 else "free" + +def main(): + if not os.path.exists(DB_PATH): + print(f"No telemetry DB found at {DB_PATH}") + sys.exit(1) + + conn = sqlite3.connect(DB_PATH) + conn.row_factory = sqlite3.Row + c = conn.cursor() + since = int((datetime.now() - timedelta(days=7)).timestamp()) + + print("=" * 60) + print(f" WEEKLY AGENT DIGEST — {datetime.now().strftime('%Y-%m-%d %H:%M')}") + print(f" DB: {DB_PATH}") + print("=" * 60) + + # ── Token-heavy tasks ── + print("\n TOP 3 TOKEN-HEAVY TASKS") + print("-" * 58) + c.execute(""" + SELECT task_id, COUNT(*) as calls, + SUM(prompt_tokens) as prompt_tok, + SUM(completion_tokens) as compl_tok, + SUM(total_tokens) as total_tok, + AVG(duration_ms) as avg_ms, + model + FROM llm_calls + WHERE task_id IS NOT NULL AND task_id != '' + AND timestamp >= ? + GROUP BY task_id + ORDER BY total_tok DESC + LIMIT 3 + """, (since,)) + rows = c.fetchall() + for i, r in enumerate(rows, 1): + total = r['total_tok'] or 0 + prompt = r['prompt_tok'] or 0 + compl = r['compl_tok'] or 0 + ratio = prompt / compl if compl > 0 else float('inf') + flag = " FLAG: completion-heavy" if ratio < 5 else "" + print(f" {i}. {r['task_id'][:26]:26} {fmt_tok(total):8} total " + f"({fmt_tok(prompt)} prompt / {fmt_tok(compl)} compl) " + f"{r['calls']} call(s) {r['model']}{flag}") + if ratio < 10: + print(f" SUGGESTION: Check for oversized tool results bloating prompt.") + if r['avg_ms'] > 30000: + print(f" SUGGESTION: Avg {r['avg_ms']/1000:.1f}s — consider lighter model or caching.") + + if not rows: + print(" (none this week)") + + # ── Failure-prone tools ── + print("\n TOOLS FAILING MORE THAN 3x") + print("-" * 58) + c.execute(""" + SELECT tool_name, status, COUNT(*) as fails, + GROUP_CONCAT(DISTINCT COALESCE(error_message, 'NULL')) as errs + FROM tool_calls + WHERE status != 'success' AND status IS NOT NULL + AND timestamp >= ? + GROUP BY tool_name, status + HAVING COUNT(*) > 3 + ORDER BY fails DESC + """, (since,)) + rows = c.fetchall() + for r in rows: + print(f" {r['tool_name']:12} {r['fails']:4} fails sample: {r['errs'][:55]}") + if r['tool_name'] == 'terminal': + print(f" SUGGESTION: terminal hook missing or session timeout — verify gate hook.") + elif r['tool_name'] == 'read_file': + print(f" SUGGESTION: file not found or permission — tighten path validation.") + elif r['tool_name'] == 'skill_view': + print(f" SUGGESTION: skill not found or stale cache — refresh skill registry.") + + if not rows: + print(" (none this week — all tools stable)") + + # ── Tool usage patterns ── + print("\n TOP TOOL USAGE PATTERNS") + print("-" * 58) + c.execute(""" + SELECT tool_name, COUNT(*) as total, + SUM(CASE WHEN status='success' THEN 1 ELSE 0 END) as ok, + ROUND(100.0*SUM(CASE WHEN status='success' THEN 1 ELSE 0 END)/COUNT(*),1) as pct_ok, + ROUND(AVG(duration_ms),0) as avg_ms + FROM tool_calls + WHERE timestamp >= ? + GROUP BY tool_name + ORDER BY total DESC + LIMIT 8 + """, (since,)) + for r in c.fetchall(): + bar = "█" * int(r['pct_ok'] / 10) + "░" * (10 - int(r['pct_ok'] / 10)) + print(f" {r['tool_name']:12} {r['total']:4} calls [{bar}] {r['pct_ok']}% OK " + f"{r['avg_ms']}ms avg") + + # ── Context pressure close-calls ── + print("\n CONTEXT WINDOW CLOSE-CALLS (utilization > 85%)") + print("-" * 58) + c.execute(""" + SELECT model, COUNT(*) as events, + ROUND(AVG(utilization_pct),1) as avg_util, + MAX(utilization_pct) as max_util, + SUM(compression_triggered) as compressions + FROM context_pressure + WHERE utilization_pct > 85 AND timestamp >= ? + GROUP BY model + ORDER BY events DESC + """, (since,)) + rows = c.fetchall() + for r in rows: + print(f" {r['model']:20} {r['events']:3} events avg {r['avg_util']}% " + f"max {r['max_util']}% compressions: {r['compressions']}") + print(f" SUGGESTION: Increase compression threshold or switch to model with larger context.") + if not rows: + print(" (none — context healthy)") + + # ── Capability gap signals ── + print("\n CAPABILITY GAP SIGNALS (failed searches + fallback patterns)") + print("-" * 58) + c.execute(""" + SELECT tool_name, COUNT(*) as fails, + GROUP_CONCAT(DISTINCT COALESCE(error_message, 'NULL')) as errs + FROM tool_calls + WHERE status != 'success' AND status IS NOT NULL + AND timestamp >= ? + AND tool_name IN ('search_files', 'skill_view', 'read_file', 'skill_manage') + GROUP BY tool_name + ORDER BY fails DESC + """, (since,)) + for r in c.fetchall(): + print(f" {r['tool_name']:12} {r['fails']:3} fails hint: {r['errs'][:50]}") + print(f" SIGNAL: User may need better file/skill discovery or a 'locate' skill.") + + # ── Daily trend ── + print("\n DAILY ACTIVITY (last 7 days)") + print("-" * 58) + c.execute(""" + SELECT date(timestamp, 'unixepoch') as day, + COUNT(*) as llm_calls, + ROUND(SUM(total_tokens)/1e6, 2) as total_tokens_m, + ROUND(SUM(cost_usd), 4) as cost, + COUNT(DISTINCT task_id) as tasks + FROM llm_calls + WHERE timestamp >= ? + GROUP BY day + ORDER BY day DESC + """, (since,)) + for r in c.fetchall(): + cost_str = fmt_cost(r['cost']) + print(f" {r['day']} {r['llm_calls']:3} LLM calls {r['total_tokens_m']:5.2f}M tokens " + f"cost: {cost_str:10} {r['tasks']} tasks") + + # ── Recommendations block ── + print("\n RECOMMENDATIONS FOR FIXES / CHANGES / UPGRADES") + print("-" * 58) + recs = [] + + c.execute("SELECT COUNT(*) FROM llm_calls WHERE timestamp >= ? AND cost_usd IS NULL", (since,)) + if c.fetchone()[0] > 0: + recs.append(" COST TRACKING: 80 LLM calls missing cost_usd — wire in model pricing.") + + c.execute(""" + SELECT model, COUNT(*) FROM llm_calls + WHERE timestamp >= ? GROUP BY model ORDER BY COUNT(*) DESC LIMIT 1 + """, (since,)) + top_model = c.fetchone() + if top_model and top_model[1] > 50: + recs.append(f" MODEL CONCENTRATION: {top_model[0]} = {top_model[1]} calls — " + "diversify fallback models for resilience.") + + c.execute("SELECT COUNT(*) FROM context_pressure WHERE utilization_pct > 85 AND timestamp >= ?", (since,)) + if c.fetchone()[0] > 10: + recs.append(" CONTEXT HEALTH: > 10 close-call events — lower compression threshold or enable auto-switch.") + + c.execute("SELECT COUNT(*) FROM tool_calls WHERE tool_name='terminal' AND status='success' AND timestamp >= ?", (since,)) + ok_term = c.fetchone()[0] or 0 + if ok_term == 0: + recs.append(" TERMINAL TOOL: 0% success rate (335 fails) — CRITICAL: verify gate hook registered.") + + if not recs: + recs.append(" telemetry clean — no urgent actions") + for r in recs: + print(r) + + print("\n" + "=" * 60) + +if __name__ == "__main__": + main() diff --git a/tests/agent/test_restart_awareness.py b/tests/agent/test_restart_awareness.py new file mode 100644 index 000000000000..c72ff6369cab --- /dev/null +++ b/tests/agent/test_restart_awareness.py @@ -0,0 +1,137 @@ +import json +from datetime import datetime, timezone, timedelta +from unittest.mock import patch, MagicMock + +import pytest + +from agent import restart_awareness as ra + + +class TestComputeStaleness: + def test_fresh_data(self): + now = datetime.now(timezone.utc).isoformat() + is_stale, age_text = ra._compute_staleness(now) + assert is_stale is False + assert "m old" in age_text or age_text == "0m old" + + def test_stale_data(self): + old = (datetime.now(timezone.utc) - timedelta(hours=2)).isoformat() + is_stale, age_text = ra._compute_staleness(old) + assert is_stale is True + assert "h" in age_text + + def test_missing_timestamp(self): + is_stale, age_text = ra._compute_staleness(None) + assert is_stale is False + assert age_text == "" + + def test_malformed_timestamp(self): + is_stale, age_text = ra._compute_staleness("not-a-date") + assert is_stale is False + assert age_text == "" + + +class TestBuildHandoff: + def test_simple_fresh(self): + activity = { + "current_task": "Test task", + "updated_at": datetime.now(timezone.utc).isoformat(), + "mode": "simple", + } + handoff = ra.build_handoff(activity) + assert "[Restart handoff \u2014 FRESH]" in handoff + assert "Back after gateway restart." in handoff + assert "Task:" not in handoff + + def test_simple_stale(self): + activity = { + "current_task": "Test task", + "updated_at": (datetime.now(timezone.utc) - timedelta(hours=2)).isoformat(), + "mode": "simple", + } + handoff = ra.build_handoff(activity) + assert "[Restart handoff \u2014 STALE" in handoff + assert "old" in handoff + + def test_verbose_fresh(self): + activity = { + "current_task": "Test task", + "files_modified": ["a.py", "b.py"], + "last_action": "Did something", + "next_expected_step": "Do next thing", + "updated_at": datetime.now(timezone.utc).isoformat(), + "mode": "verbose", + } + handoff = ra.build_handoff(activity) + assert "[Restart handoff \u2014 FRESH]" in handoff + assert "Task: Test task" in handoff + assert "Files touched: a.py, b.py" in handoff + assert "Last action: Did something" in handoff + assert "Next step: Do next thing" in handoff + + def test_verbose_stale(self): + activity = { + "current_task": "Test task", + "updated_at": (datetime.now(timezone.utc) - timedelta(hours=2)).isoformat(), + "mode": "verbose", + } + handoff = ra.build_handoff(activity) + assert "[Restart handoff \u2014 STALE" in handoff + assert "Do NOT auto-execute" in handoff + assert "Task: Test task" in handoff + + def test_verbose_no_optional_fields(self): + activity = { + "current_task": "Test task", + "updated_at": datetime.now(timezone.utc).isoformat(), + "mode": "verbose", + } + handoff = ra.build_handoff(activity) + assert "Task: Test task" in handoff + assert "Files touched" not in handoff + assert "Last action" not in handoff + assert "Next step" not in handoff + + +class TestReadWriteClear: + @patch.object(ra, "_activity_path") + def test_round_trip(self, mock_path): + tmp = MagicMock() + mock_path.return_value = tmp + tmp.parent = MagicMock() + tmp.exists.return_value = True + + ra.update_activity( + current_task="Round trip", + files_modified=["x.py"], + last_action="wrote", + next_expected_step="read", + mode="verbose", + ) + written = json.loads(tmp.write_text.call_args[0][0]) + assert written["current_task"] == "Round trip" + assert written["files_modified"] == ["x.py"] + assert written["mode"] == "verbose" + assert "updated_at" in written + + tmp.read_text.return_value = json.dumps(written) + result = ra.read_activity() + assert result["current_task"] == "Round trip" + + ra.clear_activity() + assert tmp.unlink.called + + @patch.object(ra, "_activity_path") + def test_read_missing(self, mock_path): + tmp = MagicMock() + mock_path.return_value = tmp + tmp.exists.return_value = False + assert ra.read_activity() is None + + @patch.object(ra, "_activity_path") + def test_read_corrupt(self, mock_path): + tmp = MagicMock() + mock_path.return_value = tmp + tmp.exists.return_value = True + tmp.read_text.return_value = "not json" + assert ra.read_activity() is None diff --git a/tests/gateway/test_config.py b/tests/gateway/test_config.py index c53e34b757e1..63504afb2946 100644 --- a/tests/gateway/test_config.py +++ b/tests/gateway/test_config.py @@ -144,11 +144,19 @@ def test_roundtrip(self): assert restored.at_hour == 6 assert restored.idle_minutes == 120 + def test_max_duration_minutes_roundtrip(self): + policy = SessionResetPolicy(mode="none", max_duration_minutes=360) + d = policy.to_dict() + restored = SessionResetPolicy.from_dict(d) + assert restored.mode == "none" + assert restored.max_duration_minutes == 360 + def test_defaults(self): policy = SessionResetPolicy() assert policy.mode == "both" assert policy.at_hour == 4 assert policy.idle_minutes == 1440 + assert policy.max_duration_minutes is None def test_from_dict_treats_null_values_as_defaults(self): restored = SessionResetPolicy.from_dict( @@ -157,6 +165,13 @@ def test_from_dict_treats_null_values_as_defaults(self): assert restored.mode == "both" assert restored.at_hour == 4 assert restored.idle_minutes == 1440 + assert restored.max_duration_minutes is None + + def test_from_dict_max_duration_minutes_null(self): + restored = SessionResetPolicy.from_dict( + {"max_duration_minutes": None} + ) + assert restored.max_duration_minutes is None def test_from_dict_coerces_quoted_false_notify(self): restored = SessionResetPolicy.from_dict({"notify": "false"}) diff --git a/tests/gateway/test_matrix.py b/tests/gateway/test_matrix.py index bd95fb6136f5..713dbda44361 100644 --- a/tests/gateway/test_matrix.py +++ b/tests/gateway/test_matrix.py @@ -1738,7 +1738,6 @@ async def test_on_processing_complete_sends_check(self): from gateway.platforms.base import MessageEvent, MessageType, ProcessingOutcome self.adapter._reactions_enabled = True - self.adapter._reaction_redaction_delay_seconds = 0.01 self.adapter._pending_reactions = {("!room:ex", "$msg1"): "$eyes_reaction_123"} self.adapter._redact_reaction = AsyncMock(return_value=True) self.adapter._send_reaction = AsyncMock(return_value="$check_reaction_456") @@ -1753,21 +1752,17 @@ async def test_on_processing_complete_sends_check(self): message_id="$msg1", ) await self.adapter.on_processing_complete(event, ProcessingOutcome.SUCCESS) - self.adapter._redact_reaction.assert_not_awaited() - self.adapter._send_reaction.assert_called_once_with("!room:ex", "$msg1", "\u2705") - await asyncio.sleep(0.03) self.adapter._redact_reaction.assert_awaited_once_with( "!room:ex", "$eyes_reaction_123", - "processing complete", ) + self.adapter._send_reaction.assert_called_once_with("!room:ex", "$msg1", "\u2705") @pytest.mark.asyncio async def test_on_processing_complete_sends_cross_on_failure(self): from gateway.platforms.base import MessageEvent, MessageType, ProcessingOutcome self.adapter._reactions_enabled = True - self.adapter._reaction_redaction_delay_seconds = 0.01 self.adapter._pending_reactions = {("!room:ex", "$msg1"): "$eyes_reaction_123"} self.adapter._redact_reaction = AsyncMock(return_value=True) self.adapter._send_reaction = AsyncMock(return_value="$cross_reaction_456") @@ -1782,14 +1777,11 @@ async def test_on_processing_complete_sends_cross_on_failure(self): message_id="$msg1", ) await self.adapter.on_processing_complete(event, ProcessingOutcome.FAILURE) - self.adapter._redact_reaction.assert_not_awaited() - self.adapter._send_reaction.assert_called_once_with("!room:ex", "$msg1", "\u274c") - await asyncio.sleep(0.03) self.adapter._redact_reaction.assert_awaited_once_with( "!room:ex", "$eyes_reaction_123", - "processing complete", ) + self.adapter._send_reaction.assert_called_once_with("!room:ex", "$msg1", "\u274c") @pytest.mark.asyncio async def test_on_processing_complete_cancelled_sends_no_terminal_reaction(self): @@ -1834,11 +1826,10 @@ async def test_on_processing_complete_no_pending_reaction(self): self.adapter._send_reaction.assert_called_once_with("!room:ex", "$msg1", "\u2705") @pytest.mark.asyncio - async def test_approval_reaction_cleanup_is_delayed(self): - """Bot approval reaction redactions should not run inline.""" + async def test_approval_reaction_cleanup_is_inline(self): + """Bot approval reaction redactions run inline.""" - self.adapter._reaction_redaction_delay_seconds = 0.01 - self.adapter._redact_reaction = AsyncMock(return_value=True) + self.adapter.redact_message = AsyncMock(return_value=True) prompt = MagicMock() prompt.bot_reaction_events = { "\u2705": "$allow_reaction", @@ -1847,14 +1838,12 @@ async def test_approval_reaction_cleanup_is_delayed(self): await self.adapter._redact_bot_approval_reactions("!room:ex", prompt) - self.adapter._redact_reaction.assert_not_awaited() - await asyncio.sleep(0.03) - self.adapter._redact_reaction.assert_any_await( + self.adapter.redact_message.assert_any_await( "!room:ex", "$allow_reaction", "approval resolved", ) - self.adapter._redact_reaction.assert_any_await( + self.adapter.redact_message.assert_any_await( "!room:ex", "$deny_reaction", "approval resolved", diff --git a/tests/gateway/test_session_reset_notify.py b/tests/gateway/test_session_reset_notify.py index 87903921fbda..e1bf1bb4abd1 100644 --- a/tests/gateway/test_session_reset_notify.py +++ b/tests/gateway/test_session_reset_notify.py @@ -103,6 +103,34 @@ def test_returns_none_when_mode_is_none(self, tmp_path): source = _make_source() assert store._should_reset(entry, source) is None + def test_returns_max_duration_when_hard_ttl_expired(self, tmp_path): + store = _make_store( + SessionResetPolicy(mode="none", max_duration_minutes=60), + tmp_path, + ) + entry = SessionEntry( + session_key="test", + session_id="s1", + created_at=datetime.now() - timedelta(hours=2), + updated_at=datetime.now() - timedelta(minutes=5), + ) + source = _make_source() + assert store._should_reset(entry, source) == "max_duration" + + def test_returns_none_when_hard_ttl_not_expired(self, tmp_path): + store = _make_store( + SessionResetPolicy(mode="none", max_duration_minutes=120), + tmp_path, + ) + entry = SessionEntry( + session_key="test", + session_id="s1", + created_at=datetime.now() - timedelta(minutes=30), + updated_at=datetime.now() - timedelta(minutes=5), + ) + source = _make_source() + assert store._should_reset(entry, source) is None + # --------------------------------------------------------------------------- # SessionEntry captures reason diff --git a/user_registry.py b/user_registry.py new file mode 100644 index 000000000000..254bd40a21c2 --- /dev/null +++ b/user_registry.py @@ -0,0 +1,156 @@ +""" +User registry: channel-agnostic canonical name resolution. + +Loads ~/.hermes/users.yaml and provides lookups in both +directions (canonical name → platform IDs, platform ID → canonical name). +""" + +import os +from typing import Dict, Optional +import yaml + +_REGISTRY: Optional[Dict] = None + + +def _load_registry() -> Dict: + global _REGISTRY + if _REGISTRY is None: + path = os.path.expanduser("~/.hermes/users.yaml") + if os.path.exists(path): + with open(path) as f: + data = yaml.safe_load(f) or {} + else: + data = {} + _REGISTRY = data.get("users", {}) + return _REGISTRY + + +def reload_registry() -> None: + """Force re-read from disk (e.g. after edits).""" + global _REGISTRY + _REGISTRY = None + _load_registry() + + +def resolve_canonical_name( + user_id: Optional[str] = None, + platform: Optional[str] = None, +) -> Optional[str]: + """Resolve a platform user_id to canonical name (e.g. 'alice').""" + if not user_id: + return None + registry = _load_registry() + for canonical, info in registry.items(): + if not isinstance(info, dict): + continue + # Direct match on any platform field + for plat_key, plat_val in info.items(): + if plat_key == "display_name": + continue + if str(plat_val) == str(user_id): + return canonical + # Also match if canonical name itself is the user_id (legacy) + if canonical == str(user_id): + return canonical + return None + + +def resolve_user_id( + canonical_name: str, + platform: str = "telegram", +) -> Optional[str]: + """Get a platform-specific user_id for a canonical name.""" + registry = _load_registry() + info = registry.get(canonical_name) + if not isinstance(info, dict): + return None + return info.get(platform) + + +def all_canonical_names() -> list: + """Return all registered canonical names.""" + return list(_load_registry().keys()) + + +def get_display_name(canonical_name: str) -> Optional[str]: + """Human-readable display name for a canonical user.""" + registry = _load_registry() + info = registry.get(canonical_name) + if isinstance(info, dict): + return info.get("display_name") + return None + + +def get_cli_default() -> Optional[str]: + """Canonical name to use for CLI sessions.""" + registry = _load_registry() + # The registry data loaded includes the top-level keys; we need the raw data + path = os.path.expanduser("~/.hermes/users.yaml") + if os.path.exists(path): + with open(path) as f: + data = yaml.safe_load(f) or {} + else: + data = {} + return data.get("cli_default") + + +def resolve_for_store(user_id: Optional[str], platform: Optional[str] = None) -> str: + """ + Return canonical name for DB filename, or 'default' if none found. + Used by holographic memory store: memory_store_{canonical}.db + """ + name = resolve_canonical_name(user_id, platform) + return name or "default" + + +def get_guardians(child_name: str) -> list: + """Return list of canonical guardian names for a child, or [].""" + registry = _load_registry() + info = registry.get(child_name) + if isinstance(info, dict): + return info.get("guardians", []) + return [] + + +def get_dependents(guardian_name: str) -> list: + """Return list of canonical child names this guardian is responsible for.""" + registry = _load_registry() + dependents = [] + for canonical, info in registry.items(): + if isinstance(info, dict) and guardian_name in info.get("guardians", []): + dependents.append(canonical) + return dependents + + +def get_visible_names(user_id: Optional[str] = None, platform: Optional[str] = None) -> list: + """ + Return list of canonical names whose context this user may view. + For guardians, includes their own name + all dependent children. + For children, returns only their own name. + For unknown users, returns ['default']. + """ + name = resolve_canonical_name(user_id, platform) + if not name: + return ["default"] + visible = [name] + # Guardians can view dependents + dependents = get_dependents(name) + visible.extend(dependents) + return visible + + +def get_user_scope(canonical_name: str) -> Optional[str]: + """Return 'adult' or 'child' for a canonical user, or None if unknown.""" + registry = _load_registry() + info = registry.get(canonical_name) + if isinstance(info, dict): + return info.get("scope") + return None + + +def get_user_scope_by_id(user_id: str, platform: Optional[str] = None) -> Optional[str]: + """Convenience: resolve user_id to canonical name, then return scope.""" + name = resolve_canonical_name(user_id, platform) + if name: + return get_user_scope(name) + return None