From 6866c6465ebd52049283b6e41a1c0cbcefea3799 Mon Sep 17 00:00:00 2001 From: JR Razmus Date: Wed, 15 Jul 2026 17:57:51 -0500 Subject: [PATCH 01/10] fix: bound Hermes session state growth safely --- cli.py | 38 +-- gateway/run.py | 13 +- hermes_cli/active_sessions.py | 33 ++- hermes_cli/backup.py | 143 ++++++++++- hermes_cli/config.py | 4 + hermes_cli/gateway.py | 44 ++-- hermes_cli/main.py | 145 ++++++++++- hermes_state.py | 233 ++++++++++++++++-- session_policy.py | 110 +++++++++ .../cli/test_single_query_session_finalize.py | 55 +++++ tests/hermes_cli/test_active_sessions.py | 31 +++ tests/hermes_cli/test_backup.py | 133 ++++++++++ tests/hermes_cli/test_cmd_update.py | 27 +- tests/hermes_cli/test_gateway_service.py | 49 ++++ tests/test_hermes_state.py | 172 ++++++++++++- website/docs/user-guide/sessions.md | 62 ++++- 16 files changed, 1206 insertions(+), 86 deletions(-) create mode 100644 session_policy.py diff --git a/cli.py b/cli.py index 8a8fdf6f5ff57..6550e7c3f7835 100644 --- a/cli.py +++ b/cli.py @@ -1230,6 +1230,18 @@ def _notify_single_query_session_finalize(cli, *, reason: str = "shutdown") -> N def _finalize_single_query(cli) -> None: """Close one-shot CLI resources before releasing the active session lease.""" try: + agent = getattr(cli, "agent", None) + session_id = getattr(agent, "session_id", None) or getattr( + cli, "session_id", None + ) + session_db = getattr(cli, "_session_db", None) or getattr( + agent, "_session_db", None + ) + if session_db is not None and session_id: + try: + session_db.end_session(session_id, "agent_close") + except Exception: + pass _notify_single_query_session_finalize(cli) _run_cleanup(notify_session_finalize=False) finally: @@ -1770,17 +1782,13 @@ def _cleanup_worktree(info: Dict[str, str] = None) -> None: def _run_state_db_auto_maintenance(session_db) -> None: """Call ``SessionDB.maybe_auto_prune_and_vacuum`` using current config. - Reads the ``sessions:`` section from config.yaml via - :func:`hermes_cli.config.load_config` (the authoritative loader that - deep-merges DEFAULT_CONFIG, so unmigrated configs still get default - values). Honours ``auto_prune`` / ``retention_days`` / - ``vacuum_after_prune`` / ``min_interval_hours``, and delegates to the - DB. Never raises — maintenance must never block interactive startup. + Uses the profile-specific policy loaded by SessionDB and delegates the + maintenance operation to the database. Never raises — maintenance must + never block interactive startup. """ if session_db is None: return try: - from hermes_cli.config import load_config as _load_full_config from hermes_constants import get_hermes_home as _get_hermes_home _hermes_home_maint = _get_hermes_home() @@ -1808,13 +1816,7 @@ def _run_state_db_auto_maintenance(session_db) -> None: except Exception as _finalize_exc: logger.debug("Orphan compression finalize skipped: %s", _finalize_exc) - cfg = (_load_full_config().get("sessions") or {}) - if not cfg.get("auto_prune", False): - return - session_db.maybe_auto_prune_and_vacuum( - retention_days=int(cfg.get("retention_days", 90)), - min_interval_hours=int(cfg.get("min_interval_hours", 24)), - vacuum=bool(cfg.get("vacuum_after_prune", True)), + session_db.maybe_auto_maintenance( sessions_dir=_hermes_home_maint / "sessions", ) except Exception as exc: @@ -4142,7 +4144,13 @@ def _claim_active_session(self, surface: str = "cli", *, stderr: bool = False) - ) except Exception as exc: logger.warning("Failed to claim active session slot: %s", exc) - return True + if stderr: + print("Hermes could not verify active-session ownership.", file=sys.stderr) + else: + self._console_print( + "[bold red]Hermes could not verify active-session ownership.[/]" + ) + return False if message: if stderr: print(message, file=sys.stderr) diff --git a/gateway/run.py b/gateway/run.py index f05d89540b5e8..3ce1082268438 100644 --- a/gateway/run.py +++ b/gateway/run.py @@ -3122,16 +3122,9 @@ def __init__(self, config: Optional[GatewayConfig] = None): # but never raised. if self._session_db is not None: try: - from hermes_cli.config import load_config as _load_full_config - _sess_cfg = (_load_full_config().get("sessions") or {}) - if _sess_cfg.get("auto_prune", False): - # Construction-time, before the loop serves traffic; sync DB is fine. - self._session_db._db.maybe_auto_prune_and_vacuum( - retention_days=int(_sess_cfg.get("retention_days", 90)), - min_interval_hours=int(_sess_cfg.get("min_interval_hours", 24)), - vacuum=bool(_sess_cfg.get("vacuum_after_prune", True)), - sessions_dir=self.config.sessions_dir, - ) + self._session_db._db.maybe_auto_maintenance( + sessions_dir=self.config.sessions_dir, + ) except Exception as exc: logger.debug("state.db auto-maintenance skipped: %s", exc) diff --git a/hermes_cli/active_sessions.py b/hermes_cli/active_sessions.py index 7eba80e50242c..09d52b854813d 100644 --- a/hermes_cli/active_sessions.py +++ b/hermes_cli/active_sessions.py @@ -13,6 +13,7 @@ import time import uuid from dataclasses import dataclass +from contextlib import contextmanager from pathlib import Path from typing import Any, Optional @@ -142,17 +143,21 @@ def __exit__(self, exc_type, exc, tb): self._fh = None -def _read_entries(path: Path) -> list[dict[str, Any]]: +def _read_entries(path: Path, *, strict: bool = False) -> list[dict[str, Any]]: try: with open(path, "r", encoding="utf-8") as fh: data = json.load(fh) except FileNotFoundError: return [] - except Exception: + except Exception as exc: + if strict: + raise RuntimeError("active session registry is unreadable") from exc logger.warning("Ignoring corrupt active session registry at %s", path) return [] entries = data.get("entries") if isinstance(data, dict) else data if not isinstance(entries, list): + if strict: + raise RuntimeError("active session registry has an invalid shape") return [] return [entry for entry in entries if isinstance(entry, dict)] @@ -240,19 +245,11 @@ def try_acquire_active_session( ) -> tuple[Optional[ActiveSessionLease], Optional[str]]: """Acquire an active-session slot. - Returns ``(lease, None)`` on success. When the cap is disabled, the lease is - a no-op object so callers can unconditionally call ``release()``. + Returns ``(lease, None)`` on success. A disabled cap still records ownership + so offline maintenance can distinguish live and abandoned sessions. """ max_sessions = resolve_max_concurrent_sessions(config) lease_id = uuid.uuid4().hex - if max_sessions is None: - return ActiveSessionLease( - lease_id=lease_id, - session_id=session_id, - surface=surface, - enabled=False, - ), None - now = time.time() entry = { "lease_id": lease_id, @@ -276,7 +273,7 @@ def try_acquire_active_session( if pruned: logger.info("Pruned %d stale active session lease(s)", pruned) active_count = len(entries) - if active_count >= max_sessions: + if max_sessions is not None and active_count >= max_sessions: _write_entries(state_path, entries) logger.info( "Active session limit reached: active=%d max=%d surface=%s", @@ -355,3 +352,13 @@ def active_session_registry_snapshot() -> list[dict[str, Any]]: entries = _prune_dead(_read_entries(state_path)) _write_entries(state_path, entries) return entries + + +@contextmanager +def locked_active_session_registry(): + """Hold the registry lock and yield live entries, failing closed on damage.""" + state_path = _state_path() + with _FileLock(_lock_path()): + entries = _prune_dead(_read_entries(state_path, strict=True)) + _write_entries(state_path, entries) + yield entries diff --git a/hermes_cli/backup.py b/hermes_cli/backup.py index 737bea1509a25..76a733fbc2aec 100644 --- a/hermes_cli/backup.py +++ b/hermes_cli/backup.py @@ -9,6 +9,7 @@ """ import json +import hashlib import logging import os import shutil @@ -790,6 +791,14 @@ def _quick_snapshot_root(hermes_home: Optional[Path] = None) -> Path: return home / _QUICK_SNAPSHOTS_DIR +def _sha256_file(path: Path) -> str: + digest = hashlib.sha256() + with open(path, "rb") as handle: + for chunk in iter(lambda: handle.read(1024 * 1024), b""): + digest.update(chunk) + return digest.hexdigest() + + def create_quick_snapshot( label: Optional[str] = None, hermes_home: Optional[Path] = None, @@ -811,7 +820,9 @@ def create_quick_snapshot( snap_dir = root / snap_id snap_dir.mkdir(parents=True, exist_ok=True) - manifest: Dict[str, int] = {} # rel_path -> file size + manifest: Dict[str, int] = {} + digests: Dict[str, str] = {} + capture_errors: List[str] = [] for rel in _QUICK_STATE_FILES: src = home / rel @@ -839,12 +850,15 @@ def create_quick_snapshot( # snapshot time) is captured consistently. if sub.suffix == ".db": if not _safe_copy_db(sub, dst): + capture_errors.append(sub_rel) continue else: shutil.copy2(sub, dst) manifest[sub_rel] = dst.stat().st_size + digests[sub_rel] = _sha256_file(dst) except (OSError, PermissionError) as exc: logger.warning("Could not snapshot %s: %s", sub_rel, exc) + capture_errors.append(sub_rel) continue if not src.is_file(): @@ -856,12 +870,22 @@ def create_quick_snapshot( try: if src.suffix == ".db": if not _safe_copy_db(src, dst): + capture_errors.append(rel) continue else: shutil.copy2(src, dst) manifest[rel] = dst.stat().st_size + digests[rel] = _sha256_file(dst) except (OSError, PermissionError) as exc: logger.warning("Could not snapshot %s: %s", rel, exc) + capture_errors.append(rel) + + if capture_errors: + logger.error( + "State snapshot %s is incomplete; preserving it without a manifest", + snap_id, + ) + return None if not manifest: shutil.rmtree(snap_dir, ignore_errors=True) @@ -869,12 +893,14 @@ def create_quick_snapshot( # Write manifest meta = { + "manifest_version": 2, "id": snap_id, "timestamp": ts, "label": label, "file_count": len(manifest), "total_size": sum(manifest.values()), "files": manifest, + "sha256": digests, } with open(snap_dir / "manifest.json", "w", encoding="utf-8") as f: json.dump(meta, f, indent=2) @@ -990,6 +1016,104 @@ def restore_quick_snapshot( return restored > 0 +def verify_quick_snapshot( + snapshot_id: str, + hermes_home: Optional[Path] = None, +) -> tuple[bool, str]: + """Verify that a quick snapshot is complete, contained, and readable.""" + home = hermes_home or get_hermes_home() + root = _quick_snapshot_root(home) + + if not snapshot_id or "/" in snapshot_id or "\\" in snapshot_id or snapshot_id in (".", ".."): + return False, "invalid snapshot id" + + snap_dir = root / snapshot_id + try: + if snap_dir.is_symlink() or snap_dir.resolve().parent != root.resolve(): + return False, "snapshot path escapes snapshot root" + except OSError: + return False, "snapshot path is unreadable" + if not snap_dir.is_dir(): + return False, "snapshot directory is missing" + + manifest_path = snap_dir / "manifest.json" + if manifest_path.is_symlink() or not manifest_path.is_file(): + return False, "manifest is missing or unsafe" + try: + with open(manifest_path, encoding="utf-8") as f: + meta = json.load(f) + except (OSError, json.JSONDecodeError): + return False, "manifest is unreadable" + + files = meta.get("files") if isinstance(meta, dict) else None + if not isinstance(files, dict) or not files: + return False, "manifest has no files" + if meta.get("id") != snapshot_id: + return False, "manifest id does not match directory" + if meta.get("file_count") != len(files): + return False, "manifest file count does not match" + manifest_version = meta.get("manifest_version", 1) + if ( + isinstance(manifest_version, bool) + or not isinstance(manifest_version, int) + or manifest_version not in {1, 2} + ): + return False, "snapshot manifest version is invalid" + digests = meta.get("sha256", {}) + if manifest_version >= 2 and ( + not isinstance(digests, dict) or set(digests) != set(files) + ): + return False, "snapshot digest manifest is incomplete" + + total_size = 0 + for rel, expected_size in files.items(): + if not isinstance(rel, str) or not rel or not isinstance(expected_size, int) or expected_size < 0: + return False, "manifest contains an invalid file entry" + rel_path = Path(rel) + if rel_path.is_absolute() or ".." in rel_path.parts: + return False, f"manifest path is unsafe: {rel}" + + path = snap_dir / rel_path + try: + if path.is_symlink() or path.resolve().relative_to(snap_dir.resolve()) != rel_path: + return False, f"snapshot file is unsafe: {rel}" + except (OSError, ValueError): + return False, f"snapshot file escapes snapshot root: {rel}" + if not path.is_file(): + return False, f"snapshot file is missing: {rel}" + try: + actual_size = path.stat().st_size + except OSError: + return False, f"snapshot file is unreadable: {rel}" + if actual_size != expected_size: + return False, f"snapshot file size does not match: {rel}" + total_size += actual_size + if manifest_version >= 2: + expected_digest = digests.get(rel) + try: + actual_digest = _sha256_file(path) + except OSError: + return False, f"snapshot file cannot be hashed: {rel}" + if not isinstance(expected_digest, str) or actual_digest != expected_digest: + return False, f"snapshot file digest does not match: {rel}" + + if path.suffix == ".db": + try: + conn = sqlite3.connect(f"file:{path}?mode=ro&immutable=1", uri=True) + result = conn.execute("PRAGMA quick_check").fetchall() + conn.close() + except sqlite3.Error: + return False, f"snapshot database is unreadable: {rel}" + if result != [("ok",)]: + return False, f"snapshot database failed integrity check: {rel}" + + if meta.get("total_size") != total_size: + return False, "manifest total size does not match" + if (home / "state.db").is_file() and "state.db" not in files: + return False, "snapshot is missing live state.db" + return True, "ok" + + # Relative path of the cron job database inside HERMES_HOME. Kept in sync with # the entry in ``_QUICK_STATE_FILES`` and with ``cron/jobs.py``'s ``JOBS_FILE``. _CRON_JOBS_REL = "cron/jobs.json" @@ -1099,10 +1223,12 @@ def restore_cron_jobs_if_emptied( def _prune_quick_snapshots(root: Path, keep: int = _QUICK_DEFAULT_KEEP) -> int: - """Remove oldest quick snapshots beyond the keep limit. Returns count deleted.""" + """Remove verified old quick snapshots beyond the keep limit.""" if not root.exists(): return 0 + keep = max(1, keep) + dirs = sorted( (d for d in root.iterdir() if d.is_dir()), key=lambda d: d.name, @@ -1110,7 +1236,18 @@ def _prune_quick_snapshots(root: Path, keep: int = _QUICK_DEFAULT_KEEP) -> int: ) deleted = 0 - for d in dirs[keep:]: + verified_kept = 0 + for d in dirs: + try: + verified, reason = verify_quick_snapshot(d.name, hermes_home=root.parent) + except Exception as exc: + verified, reason = False, f"verification error ({type(exc).__name__})" + if not verified: + logger.warning("Preserving unverified snapshot %s: %s", d.name, reason) + continue + if verified_kept < keep: + verified_kept += 1 + continue try: shutil.rmtree(d) deleted += 1 diff --git a/hermes_cli/config.py b/hermes_cli/config.py index 45cee20712323..b10db8616b448 100644 --- a/hermes_cli/config.py +++ b/hermes_cli/config.py @@ -2898,6 +2898,7 @@ def _ensure_hermes_home_managed(home: Path): # Gateway settings — control how messaging platforms (Telegram, Discord, # Slack, etc.) deliver agent-produced files as native attachments. "gateway": { + "launchd_wrapper": None, # Seconds the gateway waits for a single messaging platform to finish # connecting during startup (and on reconnect). Discord in particular # can blow past the old fixed 30s when an account has many slash @@ -3071,6 +3072,8 @@ def _ensure_hermes_home_managed(home: Path): # How many days of ended-session history to keep. Matches the # default of ``hermes sessions prune``. "retention_days": 90, + "retention_days_by_source": {}, + "trigram_enabled": True, # VACUUM after a prune that actually deleted rows. SQLite does not # reclaim disk space on DELETE — freed pages are just reused on # subsequent INSERTs — so without VACUUM the file stays bloated @@ -3108,6 +3111,7 @@ def _ensure_hermes_home_managed(home: Path): # ``hermes update`` behaviour. "updates": { + "branch": "main", # Run a full ``hermes backup``-style zip of HERMES_HOME before every # ``hermes update``. Backups land in ``/backups/`` and # can be restored with ``hermes import ``. Off by default: diff --git a/hermes_cli/gateway.py b/hermes_cli/gateway.py index 7b9817eadd8fb..ec11b2234ac68 100644 --- a/hermes_cli/gateway.py +++ b/hermes_cli/gateway.py @@ -15,6 +15,7 @@ import sys import textwrap import time +from xml.sax.saxutils import escape as xml_escape from dataclasses import dataclass from pathlib import Path @@ -3907,23 +3908,38 @@ def generate_launchd_plist() -> str: ) ) - # Build ProgramArguments array, including --profile when using a named profile - prog_args = [ - f"{python_path}", - "-m", - "hermes_cli.main", - ] + raw_config = read_raw_config() + gateway_config = raw_config.get("gateway", {}) if isinstance(raw_config, dict) else {} + launchd_wrapper = ( + gateway_config.get("launchd_wrapper") + if isinstance(gateway_config, dict) + else None + ) + wrapper_path = None + if launchd_wrapper is not None: + if not isinstance(launchd_wrapper, str) or not launchd_wrapper.strip(): + raise ValueError("gateway.launchd_wrapper must be a non-empty absolute path") + wrapper_path = Path(launchd_wrapper).expanduser() + if ( + not wrapper_path.is_absolute() + or not wrapper_path.is_file() + or not os.access(wrapper_path, os.X_OK) + ): + raise ValueError( + "gateway.launchd_wrapper must name an absolute executable file" + ) + + prog_args = [python_path, "-m", "hermes_cli.main"] if profile_arg: - for part in profile_arg.split(): - prog_args.append(f"{part}") + prog_args.extend(profile_arg.split()) prog_args.extend( - [ - "gateway", - "run", - "--replace", - ] + ["gateway", "run", "--replace"] + ) + if wrapper_path is not None: + prog_args.insert(0, str(wrapper_path)) + prog_args_xml = "\n ".join( + f"{xml_escape(str(argument))}" for argument in prog_args ) - prog_args_xml = "\n ".join(prog_args) return f""" diff --git a/hermes_cli/main.py b/hermes_cli/main.py index 45ce0a7a8c264..dd92a69e48253 100644 --- a/hermes_cli/main.py +++ b/hermes_cli/main.py @@ -8511,7 +8511,19 @@ def _resolve_update_branch(args) -> str: ``--branch`` (check path, git-update path, ZIP-fallback path) agrees on the same answer. """ - return (getattr(args, "branch", None) or "main").strip() or "main" + explicit = getattr(args, "branch", None) + if explicit is not None: + return explicit.strip() or "main" + try: + from hermes_cli.config import read_raw_config + + updates = read_raw_config().get("updates", {}) + configured = updates.get("branch") if isinstance(updates, dict) else None + if isinstance(configured, str) and configured.strip(): + return configured.strip() + except Exception: + pass + return "main" def _cmd_update_check(branch: str = "main", *, branch_explicit: bool = False): @@ -9451,7 +9463,7 @@ def cmd_update(args): branch = _resolve_update_branch(args) _cmd_update_check( branch=branch, - branch_explicit=bool(getattr(args, "branch", None)), + branch_explicit=bool(getattr(args, "branch", None)) or branch != "main", ) return @@ -13837,6 +13849,36 @@ def _add_session_filter_args(p, default_older_help): help="Also delete archived sessions (excluded by default)", ) + sessions_finalize_stale = sessions_subparsers.add_parser( + "finalize-stale", + help="Offline repair for abandoned open CLI sessions", + ) + sessions_finalize_stale.add_argument("--source", required=True, choices=["cli"]) + sessions_finalize_stale.add_argument( + "--older-than", + required=True, + type=float, + help="Minimum age in days; must be at least 1", + ) + sessions_finalize_stale.add_argument("--limit", type=int, default=50_000) + sessions_finalize_stale.add_argument("--apply", action="store_true") + sessions_finalize_stale.add_argument( + "--offline", + action="store_true", + help="Confirm all gateways and other Hermes CLI processes are stopped", + ) + sessions_finalize_stale.add_argument( + "--yes", "-y", action="store_true", help="Skip confirmation" + ) + + sessions_drop_trigram = sessions_subparsers.add_parser( + "drop-trigram", + help="Remove the optional trigram search index after disabling it in config", + ) + sessions_drop_trigram.add_argument( + "--yes", "-y", action="store_true", help="Skip confirmation" + ) + sessions_archive = sessions_subparsers.add_parser( "archive", help="Bulk-archive (soft-hide) sessions matching filters — no deletion", @@ -14459,6 +14501,105 @@ def _export_one(session_id: str): else: print(f"Session '{args.session_id}' not found.") + elif action == "finalize-stale": + if args.older_than < 1: + print("Error: --older-than must be at least 1 day.") + return + candidates = db.list_stale_open_sessions( + source=args.source, + older_than_days=args.older_than, + limit=args.limit, + ) + if not candidates: + print("No stale open sessions match.") + return + print( + f"{len(candidates)} open {args.source} session(s) have no activity " + f"for at least {args.older_than:g} day(s)." + ) + for row in candidates[:100]: + print( + f" {row['id']} last_active={row['last_active']:.0f} " + f"{row['message_count']} msgs" + ) + if len(candidates) > 100: + print(f" ... {len(candidates) - 100} more") + if not args.apply: + print("Dry run — pass --apply after stopping other Hermes processes.") + return + if not args.offline: + print( + "Refusing to apply without --offline. Stop all Hermes gateways " + "and other Hermes CLI processes first." + ) + return + if not args.yes and not _confirm_prompt( + f"Finalize these {len(candidates)} session(s) as stale? [y/N] " + ): + print("Cancelled.") + return + from hermes_cli.active_sessions import locked_active_session_registry + from hermes_cli.gateway import find_gateway_pids + + gateway_pids = find_gateway_pids( + exclude_pids={os.getpid()}, all_profiles=True + ) + if gateway_pids: + print("Refusing to finalize while a Hermes gateway is running.") + return + candidate_ids = {row["id"] for row in candidates} + try: + with locked_active_session_registry() as active_entries: + leased_ids = { + str(entry.get("session_id") or "") for entry in active_entries + } + if candidate_ids & leased_ids: + print("Refusing to finalize sessions with live CLI leases.") + return + revalidated = db.list_stale_open_sessions( + source=args.source, + older_than_days=args.older_than, + limit=args.limit, + ) + safe_ids = [ + row["id"] + for row in revalidated + if row["id"] in candidate_ids + and not db.get_compression_lock_holder(row["id"]) + ] + if len(safe_ids) != len(candidate_ids): + print( + "Refusing to finalize because ownership or activity " + "changed after the preview." + ) + return + count = db.finalize_stale_open_sessions( + session_ids=safe_ids, + older_than_days=args.older_than, + ) + except RuntimeError: + print("Refusing to finalize because lease ownership is unreadable.") + return + print(f"Finalized {count} stale session(s).") + + elif action == "drop-trigram": + if db.session_policy.trigram_enabled: + print( + "Refusing to remove trigram while sessions.trigram_enabled is true." + ) + return + if not args.yes and not _confirm_prompt( + "Remove the derived trigram search index? [y/N] " + ): + print("Cancelled.") + return + removed = db.drop_trigram_index() + print( + "Removed trigram search index." + if removed + else "Trigram search index is already absent." + ) + elif action in ("prune", "archive"): from hermes_cli.session_filters import ( build_prune_filters, diff --git a/hermes_state.py b/hermes_state.py index 34923ebc4f23c..57e6a9acd258d 100644 --- a/hermes_state.py +++ b/hermes_state.py @@ -27,6 +27,7 @@ from agent.memory_manager import sanitize_context from hermes_constants import get_hermes_home +from session_policy import SessionPolicy, load_session_policy from typing import Any, Callable, Dict, List, Optional, Tuple, TypeVar logger = logging.getLogger(__name__) @@ -184,15 +185,20 @@ def _delete_delegate_children(conn, parent_ids: List[str]) -> List[str]: _wal_fallback_warned_paths: set[str] = set() _wal_fallback_warned_lock = threading.Lock() -_FTS_TRIGGERS = ( +_BASE_FTS_TRIGGERS = ( "messages_fts_insert", "messages_fts_delete", "messages_fts_update", +) + +_TRIGRAM_FTS_TRIGGERS = ( "messages_fts_trigram_insert", "messages_fts_trigram_delete", "messages_fts_trigram_update", ) +_FTS_TRIGGERS = _BASE_FTS_TRIGGERS + _TRIGRAM_FTS_TRIGGERS + def _set_last_init_error(msg: Optional[str]) -> None: """Record (or clear) the most recent state.db init failure. @@ -1012,6 +1018,8 @@ class SessionDB: def __init__(self, db_path: Path = None, read_only: bool = False): self.db_path = db_path or DEFAULT_DB_PATH self.read_only = read_only + self.session_policy = load_session_policy(self.db_path.parent) + self._trigram_enabled = self.session_policy.trigram_enabled self._lock = threading.Lock() self._write_count = 0 @@ -1159,20 +1167,38 @@ def _sqlite_supports_fts5(self, cursor: sqlite3.Cursor) -> bool: return False @staticmethod - def _drop_fts_triggers(cursor: sqlite3.Cursor) -> None: - for trigger in _FTS_TRIGGERS: + def _drop_fts_triggers( + cursor: sqlite3.Cursor, + *, + include_trigram: bool = True, + ) -> None: + triggers = _FTS_TRIGGERS if include_trigram else _BASE_FTS_TRIGGERS + for trigger in triggers: + try: + cursor.execute(f"DROP TRIGGER IF EXISTS {trigger}") + except sqlite3.OperationalError: + pass + + @staticmethod + def _drop_trigram_fts_triggers(cursor: sqlite3.Cursor) -> None: + for trigger in _TRIGRAM_FTS_TRIGGERS: try: cursor.execute(f"DROP TRIGGER IF EXISTS {trigger}") except sqlite3.OperationalError: pass @staticmethod - def _fts_trigger_count(cursor: sqlite3.Cursor) -> int: - placeholders = ",".join("?" for _ in _FTS_TRIGGERS) + def _fts_trigger_count( + cursor: sqlite3.Cursor, + *, + include_trigram: bool = True, + ) -> int: + triggers = _FTS_TRIGGERS if include_trigram else _BASE_FTS_TRIGGERS + placeholders = ",".join("?" for _ in triggers) row = cursor.execute( f"SELECT COUNT(*) FROM sqlite_master " f"WHERE type = 'trigger' AND name IN ({placeholders})", - _FTS_TRIGGERS, + triggers, ).fetchone() return int(row[0] if not isinstance(row, sqlite3.Row) else row[0]) @@ -1533,7 +1559,7 @@ def _init_schema(self): # backfills, index changes tied to a specific version step) stay # in a version-gated chain. Column additions are handled by # _reconcile_columns() above and no longer need entries here. - if current_version < 10 and SCHEMA_VERSION == 10: + if current_version < 10 and SCHEMA_VERSION == 10 and self._trigram_enabled: # v10: trigram FTS5 table for CJK/substring search. The # virtual table + triggers are created unconditionally via # FTS_TRIGRAM_SQL below, but existing rows need a one-time @@ -1570,7 +1596,10 @@ def _init_schema(self): # FTS_TRIGRAM_SQL, then backfill every message row. Fixes #16751. if fts5_available: self._drop_fts_triggers(cursor) - for _tbl in ("messages_fts", "messages_fts_trigram"): + _fts_tables = ["messages_fts"] + if self._trigram_enabled: + _fts_tables.append("messages_fts_trigram") + for _tbl in _fts_tables: try: cursor.execute(f"DROP TABLE IF EXISTS {_tbl}") except sqlite3.OperationalError as exc: @@ -1601,9 +1630,11 @@ def _init_schema(self): "COALESCE(tool_calls, '') " "FROM messages" ) - trigram_ok = self._ensure_fts_schema( - cursor, "messages_fts_trigram", FTS_TRIGRAM_SQL - ) + trigram_ok = False + if self._trigram_enabled: + trigram_ok = self._ensure_fts_schema( + cursor, "messages_fts_trigram", FTS_TRIGRAM_SQL + ) if trigram_ok: cursor.execute( "INSERT INTO messages_fts_trigram(rowid, content) " @@ -1723,13 +1754,20 @@ def _init_schema(self): # FTS5 setup. Run the DDL even when the virtual table exists so # CREATE TRIGGER IF NOT EXISTS repairs trigger-only degradation from # an earlier no-FTS5 runtime. - triggers_need_repair = self._fts_trigger_count(cursor) < len(_FTS_TRIGGERS) + triggers_need_repair = self._fts_trigger_count( + cursor, + include_trigram=self._trigram_enabled, + ) < ( + len(_FTS_TRIGGERS) + if self._trigram_enabled + else len(_BASE_FTS_TRIGGERS) + ) self._fts_enabled = self._ensure_fts_schema(cursor, "messages_fts", FTS_SQL) # Trigram FTS5 for CJK/substring search. This is optional relative # to the main FTS table; if it cannot be created, CJK search falls # back to LIKE. - if self._fts_enabled: + if self._fts_enabled and self._trigram_enabled: trigram_enabled = self._ensure_fts_schema( cursor, "messages_fts_trigram", FTS_TRIGRAM_SQL ) @@ -1739,6 +1777,11 @@ def _init_schema(self): cursor, include_trigram=trigram_enabled, ) + elif not self._trigram_enabled: + self._drop_trigram_fts_triggers(cursor) + self._trigram_available = False + if triggers_need_repair and self._fts_enabled: + self._rebuild_fts_indexes(cursor, include_trigram=False) self._conn.commit() @@ -6163,6 +6206,7 @@ def _prune_filter_where( *, started_before: Optional[float] = None, started_after: Optional[float] = None, + ended_before: Optional[float] = None, source: Optional[str] = None, title_like: Optional[str] = None, end_reason: Optional[str] = None, @@ -6209,6 +6253,9 @@ def _prune_filter_where( if started_after is not None: clauses.append("s.started_at >= ?") params.append(started_after) + if ended_before is not None: + clauses.append("s.ended_at < ?") + params.append(ended_before) if source: clauses.append("s.source = ?") params.append(source) @@ -6405,6 +6452,72 @@ def _do(conn): self._remove_session_files(sessions_dir, sid) return count + def list_stale_open_sessions( + self, + *, + source: str, + older_than_days: float, + limit: int = 50_000, + ) -> List[Dict[str, Any]]: + if source != "cli" or older_than_days <= 0: + raise ValueError("only stale cli sessions with a positive age are supported") + cutoff = time.time() - older_than_days * 86400 + bounded_limit = max(1, min(int(limit), 50_000)) + last_active_sql = ( + "COALESCE((SELECT MAX(m.timestamp) FROM messages m " + "WHERE m.session_id = s.id), s.started_at)" + ) + with self._lock: + rows = self._conn.execute( + f"SELECT s.id, s.source, s.started_at, s.message_count, " + f"{last_active_sql} AS last_active " + "FROM sessions s " + "WHERE s.source = ? AND s.ended_at IS NULL AND s.archived = 0 " + f"AND {last_active_sql} < ? " + "ORDER BY last_active ASC LIMIT ?", + (source, cutoff, bounded_limit), + ).fetchall() + return [dict(row) for row in rows] + + def finalize_stale_open_sessions( + self, + *, + session_ids: List[str], + older_than_days: float, + ) -> int: + if older_than_days <= 0: + raise ValueError("a positive older_than_days is required") + unique_ids = list(dict.fromkeys(session_ids))[:50_000] + if not unique_ids: + return 0 + cutoff = time.time() - older_than_days * 86400 + finalized = {"count": 0} + + def _do(conn): + for offset in range(0, len(unique_ids), 500): + batch = unique_ids[offset : offset + 500] + placeholders = ",".join("?" for _ in batch) + rows = conn.execute( + "SELECT s.id, COALESCE((SELECT MAX(m.timestamp) FROM messages m " + "WHERE m.session_id = s.id), s.started_at) AS last_active " + "FROM sessions s " + f"WHERE s.id IN ({placeholders}) AND s.source = 'cli' " + "AND s.ended_at IS NULL AND s.archived = 0 " + "AND COALESCE((SELECT MAX(m.timestamp) FROM messages m " + "WHERE m.session_id = s.id), s.started_at) < ?", + [*batch, cutoff], + ).fetchall() + for row in rows: + cursor = conn.execute( + "UPDATE sessions SET ended_at = ?, end_reason = ? " + "WHERE id = ? AND ended_at IS NULL", + (row["last_active"], "stale_repair", row["id"]), + ) + finalized["count"] += cursor.rowcount + return finalized["count"] + + return int(self._execute_write(_do)) + # ── Meta key/value (for scheduler bookkeeping) ── def get_meta(self, key: str) -> Optional[str]: @@ -6946,6 +7059,24 @@ def _fts_table_exists(self, name: str) -> bool: except sqlite3.OperationalError: return False + def drop_trigram_index(self) -> bool: + removed = {"value": False} + + def _do(conn): + row = conn.execute( + "SELECT 1 FROM sqlite_master WHERE type = 'table' " + "AND name = 'messages_fts_trigram'" + ).fetchone() + self._drop_trigram_fts_triggers(conn.cursor()) + if row is not None: + conn.execute("DROP TABLE messages_fts_trigram") + removed["value"] = True + return removed["value"] + + self._execute_write(_do) + self._trigram_available = False + return removed["value"] + def optimize_fts(self) -> int: """Merge fragmented FTS5 b-tree segments into one per index. @@ -6961,9 +7092,8 @@ def optimize_fts(self) -> int: speed. It is complementary to VACUUM: ``optimize`` compacts the FTS index internally, then VACUUM returns the freed pages to the OS. - Skips any FTS table that does not exist (e.g. the trigram index when - disabled via ``HERMES_DISABLE_FTS_TRIGRAM`` or not yet created), so - it is safe to call unconditionally. + Skips any FTS table that does not exist, so it is safe to call + unconditionally. Returns the number of FTS indexes that were optimized. """ @@ -7026,6 +7156,7 @@ def vacuum(self) -> int: def maybe_auto_prune_and_vacuum( self, retention_days: int = 90, + retention_days_by_source: Optional[Dict[str, float]] = None, min_interval_hours: int = 24, vacuum: bool = True, sessions_dir: Optional[Path] = None, @@ -7049,7 +7180,12 @@ def maybe_auto_prune_and_vacuum( - ``"vacuumed"`` (bool) — true if VACUUM ran - ``"error"`` (str, optional) — present only on failure """ - result: Dict[str, Any] = {"skipped": False, "pruned": 0, "vacuumed": False} + result: Dict[str, Any] = { + "skipped": False, + "pruned": 0, + "pruned_by_source": {}, + "vacuumed": False, + } try: # Skip if another process/call did maintenance recently. last_raw = self.get_meta("last_auto_prune") @@ -7063,10 +7199,42 @@ def maybe_auto_prune_and_vacuum( except (TypeError, ValueError): pass # corrupt meta; treat as no prior run - pruned = self.prune_sessions( - older_than_days=retention_days, - sessions_dir=sessions_dir, - ) + source_policy: Dict[str, float] = {} + for raw_source, raw_days in (retention_days_by_source or {}).items(): + source = str(raw_source).strip().lower() + try: + days = float(raw_days) + except (TypeError, ValueError): + continue + if source and days > 0: + source_policy[source] = days + if source_policy: + with self._lock: + sources = [ + row["source"] + for row in self._conn.execute( + "SELECT DISTINCT source FROM sessions WHERE source IS NOT NULL" + ).fetchall() + ] + pruned = 0 + for source in sources: + source_days = source_policy.get(source.lower(), retention_days) + source_pruned = self.prune_sessions( + older_than_days=None, + source=source, + sessions_dir=sessions_dir, + archived=False, + ended_before=now - source_days * 86400, + ) + result["pruned_by_source"][source] = source_pruned + pruned += source_pruned + else: + pruned = self.prune_sessions( + older_than_days=None, + sessions_dir=sessions_dir, + archived=False, + ended_before=now - retention_days * 86400, + ) result["pruned"] = pruned # Only VACUUM if we actually freed rows — VACUUM on a tight DB @@ -7084,7 +7252,7 @@ def maybe_auto_prune_and_vacuum( if pruned > 0: logger.info( - "state.db auto-maintenance: pruned %d session(s) older than %d days%s", + "state.db auto-maintenance: pruned %d session(s) using retention policy %s%s", pruned, retention_days, " + VACUUM" if result["vacuumed"] else "", @@ -7096,6 +7264,27 @@ def maybe_auto_prune_and_vacuum( return result + def maybe_auto_maintenance( + self, + *, + sessions_dir: Optional[Path] = None, + ) -> Dict[str, Any]: + policy: SessionPolicy = self.session_policy + if not policy.auto_prune: + return { + "skipped": True, + "pruned": 0, + "pruned_by_source": {}, + "vacuumed": False, + } + return self.maybe_auto_prune_and_vacuum( + retention_days=policy.retention_days, + retention_days_by_source=dict(policy.retention_days_by_source), + min_interval_hours=policy.min_interval_hours, + vacuum=policy.vacuum_after_prune, + sessions_dir=sessions_dir, + ) + # ── Handoff (cross-platform session transfer) ────────────────────────── # # State machine: diff --git a/session_policy.py b/session_policy.py new file mode 100644 index 0000000000000..7fdd681c035a2 --- /dev/null +++ b/session_policy.py @@ -0,0 +1,110 @@ +from __future__ import annotations + +import logging +import os +from dataclasses import dataclass, field +from pathlib import Path +from typing import Dict, Mapping + +import yaml + + +logger = logging.getLogger(__name__) + + +@dataclass(frozen=True) +class SessionPolicy: + trigram_enabled: bool = True + auto_prune: bool = False + retention_days: float = 90.0 + retention_days_by_source: Mapping[str, float] = field(default_factory=dict) + vacuum_after_prune: bool = True + min_interval_hours: float = 24.0 + + +def _positive_number(value, default: float) -> float: + try: + parsed = float(value) + except (TypeError, ValueError): + return default + return parsed if parsed > 0 else default + + +def _source_retention(value) -> Dict[str, float]: + if not isinstance(value, dict): + return {} + result: Dict[str, float] = {} + for raw_source, raw_days in value.items(): + source = str(raw_source or "").strip().lower() + if not source: + continue + try: + days = float(raw_days) + except (TypeError, ValueError): + continue + if days > 0: + result[source] = days + return result + + +def _strict_bool(value, default: bool) -> bool: + if isinstance(value, bool): + return value + if isinstance(value, str): + normalized = value.strip().lower() + if normalized in {"true", "yes", "on", "1"}: + return True + if normalized in {"false", "no", "off", "0"}: + return False + return default + + +def load_session_policy(hermes_home: Path) -> SessionPolicy: + sessions = {} + config_path = Path(hermes_home) / "config.yaml" + try: + if config_path.exists(): + loaded = yaml.safe_load(config_path.read_text(encoding="utf-8")) or {} + if not isinstance(loaded, dict): + raise ValueError("config root is not a mapping") + if "sessions" in loaded and not isinstance(loaded["sessions"], dict): + raise ValueError("sessions config is not a mapping") + sessions = loaded.get("sessions") or {} + except Exception as exc: + logger.warning( + "Could not read session policy from %s (%s)", + config_path, + type(exc).__name__, + ) + return SessionPolicy( + trigram_enabled=False, + auto_prune=False, + vacuum_after_prune=False, + ) + + trigram_enabled = ( + _strict_bool(sessions["trigram_enabled"], False) + if "trigram_enabled" in sessions + else True + ) + if os.environ.get("HERMES_DISABLE_FTS_TRIGRAM", "").strip().lower() in { + "1", "true", "yes", "on", + }: + trigram_enabled = False + + return SessionPolicy( + trigram_enabled=trigram_enabled, + auto_prune=_strict_bool(sessions.get("auto_prune"), False), + retention_days=_positive_number(sessions.get("retention_days"), 90.0), + retention_days_by_source=_source_retention( + sessions.get("retention_days_by_source") + ), + vacuum_after_prune=( + _strict_bool(sessions["vacuum_after_prune"], False) + if "vacuum_after_prune" in sessions + else True + ), + min_interval_hours=_positive_number( + sessions.get("min_interval_hours"), 24.0 + ), + ) diff --git a/tests/cli/test_single_query_session_finalize.py b/tests/cli/test_single_query_session_finalize.py index 3041c03dbfe15..47dd5dd010545 100644 --- a/tests/cli/test_single_query_session_finalize.py +++ b/tests/cli/test_single_query_session_finalize.py @@ -55,6 +55,61 @@ def cleanup(**kwargs): assert calls == ["finalize", "cleanup", "release"] +def test_finalize_single_query_ends_sqlite_session_before_cleanup(tmp_path, monkeypatch): + from hermes_state import SessionDB + + db = SessionDB(db_path=tmp_path / "state.db") + db.create_session(session_id="one-shot", source="cli") + calls = [] + fake_agent = SimpleNamespace(session_id="one-shot", platform="cli") + fake_cli = SimpleNamespace( + agent=fake_agent, + session_id="one-shot", + _session_db=db, + _release_active_session=lambda: calls.append("release"), + ) + monkeypatch.setattr( + cli, + "_notify_single_query_session_finalize", + lambda _cli: calls.append("finalize"), + ) + monkeypatch.setattr(cli, "_run_cleanup", lambda **_kwargs: calls.append("cleanup")) + + cli._finalize_single_query(fake_cli) + row = db.get_session("one-shot") + + assert row["ended_at"] is not None + assert row["end_reason"] == "agent_close" + assert calls == ["finalize", "cleanup", "release"] + db.close() + + +def test_finalize_single_query_ends_sqlite_session_when_cleanup_fails(tmp_path, monkeypatch): + from hermes_state import SessionDB + + db = SessionDB(db_path=tmp_path / "state.db") + db.create_session(session_id="one-shot", source="cli") + fake_cli = SimpleNamespace( + agent=SimpleNamespace(session_id="one-shot", platform="cli"), + _session_db=db, + _release_active_session=lambda: None, + ) + monkeypatch.setattr(cli, "_notify_single_query_session_finalize", lambda _cli: None) + monkeypatch.setattr( + cli, + "_run_cleanup", + lambda **_kwargs: (_ for _ in ()).throw(RuntimeError("cleanup failed")), + ) + + with pytest.raises(RuntimeError, match="cleanup failed"): + cli._finalize_single_query(fake_cli) + + row = db.get_session("one-shot") + assert row["ended_at"] is not None + assert row["end_reason"] == "agent_close" + db.close() + + def test_finalize_single_query_runs_cleanup_when_finalize_hook_fails(monkeypatch): calls = [] fake_agent = SimpleNamespace(session_id="agent-session", platform="cli") diff --git a/tests/hermes_cli/test_active_sessions.py b/tests/hermes_cli/test_active_sessions.py index 560803dc852ef..0b851632733bd 100644 --- a/tests/hermes_cli/test_active_sessions.py +++ b/tests/hermes_cli/test_active_sessions.py @@ -76,6 +76,37 @@ def test_active_session_lease_blocks_until_release(tmp_path, monkeypatch): assert active_sessions.active_session_registry_snapshot() == [] +def test_unlimited_sessions_still_record_ownership(tmp_path, monkeypatch): + home = tmp_path / ".hermes" + monkeypatch.setenv("HERMES_HOME", str(home)) + lease, message = active_sessions.try_acquire_active_session( + session_id="tracked-unlimited", + surface="cli", + config={}, + ) + assert message is None + assert lease is not None and lease.enabled is True + assert [ + entry["session_id"] + for entry in active_sessions.active_session_registry_snapshot() + ] == ["tracked-unlimited"] + lease.release() + assert active_sessions.active_session_registry_snapshot() == [] + + +def test_locked_registry_fails_closed_on_corrupt_state(tmp_path, monkeypatch): + import pytest + + home = tmp_path / ".hermes" + runtime = home / "runtime" + runtime.mkdir(parents=True) + (runtime / "active_sessions.json").write_text("{") + monkeypatch.setenv("HERMES_HOME", str(home)) + with pytest.raises(RuntimeError, match="unreadable"): + with active_sessions.locked_active_session_registry(): + pass + + def test_active_session_registry_prunes_dead_pids(tmp_path, monkeypatch): home = tmp_path / ".hermes" monkeypatch.setenv("HERMES_HOME", str(home)) diff --git a/tests/hermes_cli/test_backup.py b/tests/hermes_cli/test_backup.py index 17832746ca3aa..36779c427e05c 100644 --- a/tests/hermes_cli/test_backup.py +++ b/tests/hermes_cli/test_backup.py @@ -1512,6 +1512,139 @@ def test_manual_prune(self, hermes_home): assert deleted == 7 assert len(list_quick_snapshots(hermes_home=hermes_home)) == 3 + def test_verify_snapshot(self, hermes_home): + from hermes_cli.backup import create_quick_snapshot, verify_quick_snapshot + snap_id = create_quick_snapshot(hermes_home=hermes_home) + assert verify_quick_snapshot(snap_id, hermes_home=hermes_home) == (True, "ok") + + def test_verify_rejects_missing_file(self, hermes_home): + from hermes_cli.backup import create_quick_snapshot, verify_quick_snapshot + snap_id = create_quick_snapshot(hermes_home=hermes_home) + (hermes_home / "state-snapshots" / snap_id / "config.yaml").unlink() + ok, reason = verify_quick_snapshot(snap_id, hermes_home=hermes_home) + assert ok is False + assert "missing" in reason + + def test_verify_rejects_size_mismatch(self, hermes_home): + from hermes_cli.backup import create_quick_snapshot, verify_quick_snapshot + snap_id = create_quick_snapshot(hermes_home=hermes_home) + (hermes_home / "state-snapshots" / snap_id / "config.yaml").write_text("changed") + ok, reason = verify_quick_snapshot(snap_id, hermes_home=hermes_home) + assert ok is False + assert "size" in reason + + def test_verify_rejects_same_size_corruption(self, hermes_home): + from hermes_cli.backup import create_quick_snapshot, verify_quick_snapshot + snap_id = create_quick_snapshot(hermes_home=hermes_home) + config = hermes_home / "state-snapshots" / snap_id / "config.yaml" + original = config.read_bytes() + config.write_bytes(b"x" * len(original)) + ok, reason = verify_quick_snapshot(snap_id, hermes_home=hermes_home) + assert ok is False + assert "digest" in reason + + def test_verify_rejects_manifest_traversal(self, hermes_home): + from hermes_cli.backup import create_quick_snapshot, verify_quick_snapshot + snap_id = create_quick_snapshot(hermes_home=hermes_home) + manifest = hermes_home / "state-snapshots" / snap_id / "manifest.json" + meta = json.loads(manifest.read_text()) + meta["files"] = {"../config.yaml": 1} + meta["sha256"] = {"../config.yaml": "0" * 64} + meta["file_count"] = 1 + meta["total_size"] = 1 + manifest.write_text(json.dumps(meta)) + ok, reason = verify_quick_snapshot(snap_id, hermes_home=hermes_home) + assert ok is False + assert "unsafe" in reason + + @pytest.mark.parametrize("version", ("invalid", False, 1.5, 3)) + def test_verify_rejects_malformed_manifest_version(self, hermes_home, version): + from hermes_cli.backup import create_quick_snapshot, verify_quick_snapshot + snap_id = create_quick_snapshot(hermes_home=hermes_home) + manifest = hermes_home / "state-snapshots" / snap_id / "manifest.json" + meta = json.loads(manifest.read_text()) + meta["manifest_version"] = version + manifest.write_text(json.dumps(meta)) + ok, reason = verify_quick_snapshot(snap_id, hermes_home=hermes_home) + assert ok is False + assert "version" in reason + + def test_prune_preserves_snapshot_when_verifier_raises( + self, hermes_home, monkeypatch + ): + import hermes_cli.backup as backup_mod + + old = backup_mod.create_quick_snapshot(label="old", hermes_home=hermes_home) + backup_mod.create_quick_snapshot(label="new", hermes_home=hermes_home) + real_verify = backup_mod.verify_quick_snapshot + + def raising_verify(snapshot_id, hermes_home=None): + if snapshot_id == old: + raise OSError("unreadable") + return real_verify(snapshot_id, hermes_home=hermes_home) + + monkeypatch.setattr(backup_mod, "verify_quick_snapshot", raising_verify) + assert backup_mod.prune_quick_snapshots(keep=1, hermes_home=hermes_home) == 0 + assert (hermes_home / "state-snapshots" / old).exists() + + def test_verify_rejects_corrupt_database(self, hermes_home): + from hermes_cli.backup import create_quick_snapshot, verify_quick_snapshot + snap_id = create_quick_snapshot(hermes_home=hermes_home) + db_path = hermes_home / "state-snapshots" / snap_id / "state.db" + db_path.write_bytes(b"not a sqlite database") + manifest = hermes_home / "state-snapshots" / snap_id / "manifest.json" + meta = json.loads(manifest.read_text()) + meta["files"]["state.db"] = db_path.stat().st_size + meta["total_size"] = sum(meta["files"].values()) + manifest.write_text(json.dumps(meta)) + ok, reason = verify_quick_snapshot(snap_id, hermes_home=hermes_home) + assert ok is False + assert "database" in reason or "digest" in reason + + def test_prune_preserves_unverified_snapshot(self, hermes_home): + from hermes_cli.backup import create_quick_snapshot, prune_quick_snapshots + bad_id = create_quick_snapshot(label="bad", hermes_home=hermes_home) + (hermes_home / "state-snapshots" / bad_id / "config.yaml").unlink() + for label in ("newer-a", "newer-b"): + create_quick_snapshot(label=label, hermes_home=hermes_home) + deleted = prune_quick_snapshots(keep=2, hermes_home=hermes_home) + assert deleted == 0 + assert (hermes_home / "state-snapshots" / bad_id).exists() + + def test_unverified_newest_does_not_displace_verified_recovery_point( + self, hermes_home + ): + from hermes_cli.backup import create_quick_snapshot, prune_quick_snapshots + oldest = create_quick_snapshot(label="a-oldest", hermes_home=hermes_home) + newest_verified = create_quick_snapshot( + label="m-newest-verified", hermes_home=hermes_home + ) + newest_bad = create_quick_snapshot(label="z-newest-bad", hermes_home=hermes_home) + (hermes_home / "state-snapshots" / newest_bad / "config.yaml").unlink() + assert prune_quick_snapshots(keep=1, hermes_home=hermes_home) == 1 + assert not (hermes_home / "state-snapshots" / oldest).exists() + assert (hermes_home / "state-snapshots" / newest_verified).exists() + assert (hermes_home / "state-snapshots" / newest_bad).exists() + + def test_failed_state_db_copy_never_produces_verified_manifest( + self, hermes_home, monkeypatch + ): + import hermes_cli.backup as backup_mod + + real_copy = backup_mod._safe_copy_db + + def fail_state_db(src, dst): + if src.name == "state.db": + return False + return real_copy(src, dst) + + monkeypatch.setattr(backup_mod, "_safe_copy_db", fail_state_db) + snap_id = backup_mod.create_quick_snapshot(hermes_home=hermes_home) + assert snap_id is None + snapshots = list((hermes_home / "state-snapshots").iterdir()) + assert len(snapshots) == 1 + assert not (snapshots[0] / "manifest.json").exists() + def test_snapshot_includes_pairing_directories(self, hermes_home): """Pairing JSONs live outside state.db — snapshot must capture them recursively (generic + per-platform) so approved-user lists survive diff --git a/tests/hermes_cli/test_cmd_update.py b/tests/hermes_cli/test_cmd_update.py index b1458cbd3c2f9..d19ffcd1cb7b8 100644 --- a/tests/hermes_cli/test_cmd_update.py +++ b/tests/hermes_cli/test_cmd_update.py @@ -7,7 +7,7 @@ import pytest -from hermes_cli.main import cmd_update, PROJECT_ROOT +from hermes_cli.main import cmd_update, PROJECT_ROOT, _resolve_update_branch def _make_run_side_effect(branch="main", verify_ok=True, commit_count="0"): @@ -40,6 +40,31 @@ def mock_args(): return SimpleNamespace() +def test_update_branch_uses_config_when_cli_is_absent(monkeypatch): + monkeypatch.setattr( + "hermes_cli.config.read_raw_config", + lambda: {"updates": {"branch": "jr-stable"}}, + ) + assert _resolve_update_branch(SimpleNamespace(branch=None)) == "jr-stable" + + +def test_update_branch_cli_overrides_config(monkeypatch): + monkeypatch.setattr( + "hermes_cli.config.read_raw_config", + lambda: {"updates": {"branch": "jr-stable"}}, + ) + assert _resolve_update_branch(SimpleNamespace(branch="upstream-test")) == "upstream-test" + + +@pytest.mark.parametrize( + "config", + ({}, {"updates": []}, {"updates": {"branch": " "}}, {"updates": {"branch": 3}}), +) +def test_update_branch_invalid_config_falls_back_to_main(monkeypatch, config): + monkeypatch.setattr("hermes_cli.config.read_raw_config", lambda: config) + assert _resolve_update_branch(SimpleNamespace(branch=None)) == "main" + + # --------------------------------------------------------------------------- # Managed-uv compatibility for tests that patch shutil.which # --------------------------------------------------------------------------- diff --git a/tests/hermes_cli/test_gateway_service.py b/tests/hermes_cli/test_gateway_service.py index ef563c34b49a5..514835f3ed19d 100644 --- a/tests/hermes_cli/test_gateway_service.py +++ b/tests/hermes_cli/test_gateway_service.py @@ -1,6 +1,7 @@ """Tests for gateway service management helpers.""" import os +import plistlib import subprocess from pathlib import Path from types import SimpleNamespace @@ -2590,6 +2591,54 @@ def test_launchd_plist_includes_profile(self, tmp_path, monkeypatch): assert "--profile" in plist assert "mybot" in plist + def test_launchd_plist_wraps_full_command(self, tmp_path, monkeypatch): + wrapper = tmp_path / "gate & reliability" + wrapper.write_text("#!/bin/sh\nexec \"$@\"\n") + wrapper.chmod(0o755) + monkeypatch.setattr( + gateway_cli, + "read_raw_config", + lambda: {"gateway": {"launchd_wrapper": str(wrapper)}}, + ) + plist = plistlib.loads(gateway_cli.generate_launchd_plist().encode()) + arguments = plist["ProgramArguments"] + assert arguments[0] == str(wrapper) + assert arguments[1:4] == [gateway_cli.get_python_path(), "-m", "hermes_cli.main"] + assert arguments[-3:] == ["gateway", "run", "--replace"] + + def test_launchd_plist_wrapper_preserves_named_profile(self, tmp_path, monkeypatch): + wrapper = tmp_path / "gate" + wrapper.write_text("#!/bin/sh\nexec \"$@\"\n") + wrapper.chmod(0o755) + profile_dir = tmp_path / ".hermes" / "profiles" / "mybot" + profile_dir.mkdir(parents=True) + monkeypatch.setattr(Path, "home", lambda: tmp_path) + monkeypatch.setenv("HERMES_HOME", str(profile_dir)) + monkeypatch.setattr(gateway_cli, "get_hermes_home", lambda: profile_dir) + monkeypatch.setattr( + gateway_cli, + "read_raw_config", + lambda: {"gateway": {"launchd_wrapper": str(wrapper)}}, + ) + arguments = plistlib.loads( + gateway_cli.generate_launchd_plist().encode() + )["ProgramArguments"] + assert arguments[:2] == [str(wrapper), gateway_cli.get_python_path()] + assert arguments[4:6] == ["--profile", "mybot"] + + @pytest.mark.parametrize("configured", ("relative", "", None)) + def test_launchd_plist_invalid_configured_wrapper_fails_closed( + self, tmp_path, monkeypatch, configured + ): + value = str(tmp_path / "missing") if configured is None else configured + monkeypatch.setattr( + gateway_cli, + "read_raw_config", + lambda: {"gateway": {"launchd_wrapper": value}}, + ) + with pytest.raises(ValueError, match="launchd_wrapper"): + gateway_cli.generate_launchd_plist() + def test_launchd_plist_supports_aqua_and_background_sessions(self): # macOS 26+ only loads the agent in non-Aqua sessions when the plist # opts into Background as well (issue #23387). diff --git a/tests/test_hermes_state.py b/tests/test_hermes_state.py index eb6aaf279ea3a..77a29c9f2d6c2 100644 --- a/tests/test_hermes_state.py +++ b/tests/test_hermes_state.py @@ -4682,12 +4682,16 @@ def _boom(): class TestAutoMaintenance: def _make_old_ended(self, db, sid: str, days_old: int = 100): - """Create a session that is ended and was started `days_old` days ago.""" + """Create a session that ended ``days_old`` days ago.""" db.create_session(session_id=sid, source="cli") db.end_session(sid, end_reason="done") db._conn.execute( - "UPDATE sessions SET started_at = ? WHERE id = ?", - (time.time() - days_old * 86400, sid), + "UPDATE sessions SET started_at = ?, ended_at = ? WHERE id = ?", + ( + time.time() - days_old * 86400, + time.time() - days_old * 86400, + sid, + ), ) db._conn.commit() @@ -6070,3 +6074,165 @@ def test_offset_without_limit_pages(self, db): self._seed(db, n=5) rows = db.get_messages("s1", offset=3) assert [m["content"] for m in rows] == ["msg-3", "msg-4"] +def test_trigram_policy_disables_growth_and_explicit_drop_is_idempotent(tmp_path): + db_path = tmp_path / "state.db" + enabled = SessionDB(db_path=db_path) + enabled.create_session(session_id="s1", source="cli") + enabled.append_message("s1", role="user", content="大别山 project alpha") + assert enabled._fts_table_exists("messages_fts_trigram") is True + enabled.close() + + (tmp_path / "config.yaml").write_text( + "sessions:\n trigram_enabled: false\n", + encoding="utf-8", + ) + disabled = SessionDB(db_path=db_path) + assert disabled._trigram_available is False + trigger_count = disabled._conn.execute( + "SELECT COUNT(*) FROM sqlite_master WHERE type='trigger' " + "AND name LIKE 'messages_fts_trigram_%'" + ).fetchone()[0] + assert trigger_count == 0 + before = disabled._conn.execute( + "SELECT COUNT(*) FROM messages_fts_trigram" + ).fetchone()[0] + disabled.append_message("s1", role="assistant", content="new unindexed text") + after = disabled._conn.execute( + "SELECT COUNT(*) FROM messages_fts_trigram" + ).fetchone()[0] + assert after == before + assert len(disabled.search_messages("alpha")) == 1 + assert len(disabled.search_messages("大别山")) == 1 + assert disabled.drop_trigram_index() is True + assert disabled.drop_trigram_index() is False + disabled.close() + + reopened = SessionDB(db_path=db_path) + assert reopened._fts_table_exists("messages_fts_trigram") is False + assert reopened._trigram_available is False + assert len(reopened.search_messages("alpha")) == 1 + assert len(reopened.search_messages("大别山")) == 1 + reopened.close() + + +def test_session_policy_coerces_quoted_booleans_and_fails_closed(tmp_path): + from session_policy import load_session_policy + + config_path = tmp_path / "config.yaml" + config_path.write_text( + "sessions:\n" + " trigram_enabled: \"false\"\n" + " auto_prune: \"false\"\n" + " vacuum_after_prune: \"false\"\n", + encoding="utf-8", + ) + policy = load_session_policy(tmp_path) + assert policy.trigram_enabled is False + assert policy.auto_prune is False + assert policy.vacuum_after_prune is False + + config_path.write_text( + "sessions:\n" + " trigram_enabled: invalid\n" + " auto_prune: invalid\n" + " vacuum_after_prune: invalid\n", + encoding="utf-8", + ) + policy = load_session_policy(tmp_path) + assert policy.trigram_enabled is False + assert policy.auto_prune is False + assert policy.vacuum_after_prune is False + + config_path.write_text("sessions: [", encoding="utf-8") + policy = load_session_policy(tmp_path) + assert policy.trigram_enabled is False + assert policy.auto_prune is False + assert policy.vacuum_after_prune is False + + +def test_source_retention_uses_global_fallback_and_preserves_protected_rows(tmp_path): + db = SessionDB(db_path=tmp_path / "state.db") + now = time.time() + for session_id, source, age_days, ended, archived in ( + ("cron-old", "cron", 10, True, False), + ("cron-new", "cron", 2, True, False), + ("discord-kept", "discord", 100, True, False), + ("unknown-old", "custom", 100, True, False), + ("recently-ended-long", "custom", 100, True, False), + ("active-old", "cron", 30, False, False), + ("archived-old", "cron", 30, True, True), + ): + db.create_session(session_id=session_id, source=source) + db._conn.execute( + "UPDATE sessions SET started_at=?, ended_at=?, archived=? WHERE id=?", + ( + now - age_days * 86400, + now - age_days * 86400 if ended else None, + int(archived), + session_id, + ), + ) + db._conn.commit() + db._conn.execute( + "UPDATE sessions SET ended_at=? WHERE id='recently-ended-long'", + (now - 86400,), + ) + db._conn.commit() + + result = db.maybe_auto_prune_and_vacuum( + retention_days=90, + retention_days_by_source={"cron": 7, "discord": 120}, + min_interval_hours=0, + vacuum=False, + ) + + assert result["pruned"] == 2 + assert result["pruned_by_source"]["cron"] == 1 + assert result["pruned_by_source"]["discord"] == 0 + assert result["pruned_by_source"]["custom"] == 1 + assert db.get_session("cron-old") is None + assert db.get_session("unknown-old") is None + for session_id in ( + "cron-new", + "discord-kept", + "recently-ended-long", + "active-old", + "archived-old", + ): + assert db.get_session(session_id) is not None + db.close() + + +def test_finalize_stale_open_sessions_is_scoped_and_idempotent(tmp_path): + db = SessionDB(db_path=tmp_path / "state.db") + now = time.time() + for session_id, source, age_days, archived, ended in ( + ("stale-cli", "cli", 3, False, False), + ("recent-cli", "cli", 0.25, False, False), + ("archived-cli", "cli", 3, True, False), + ("stale-discord", "discord", 3, False, False), + ("ended-cli", "cli", 3, False, True), + ): + db.create_session(session_id=session_id, source=source) + timestamp = now - age_days * 86400 + db.append_message(session_id, role="user", content=session_id, timestamp=timestamp) + db._conn.execute( + "UPDATE sessions SET started_at=?, archived=?, ended_at=? WHERE id=?", + (timestamp, int(archived), timestamp if ended else None, session_id), + ) + db._conn.commit() + + candidates = db.list_stale_open_sessions(source="cli", older_than_days=2) + assert [row["id"] for row in candidates] == ["stale-cli"] + assert db.finalize_stale_open_sessions( + session_ids=["stale-cli"], older_than_days=2 + ) == 1 + assert db.finalize_stale_open_sessions( + session_ids=["stale-cli"], older_than_days=2 + ) == 0 + repaired = db.get_session("stale-cli") + assert repaired["ended_at"] is not None + assert repaired["end_reason"] == "stale_repair" + assert db.get_session("archived-cli")["ended_at"] is None + assert db.get_session("stale-discord")["ended_at"] is None + db.close() diff --git a/website/docs/user-guide/sessions.md b/website/docs/user-guide/sessions.md index daece81f48292..2d802b251ff05 100644 --- a/website/docs/user-guide/sessions.md +++ b/website/docs/user-guide/sessions.md @@ -483,6 +483,22 @@ delete them too. Pruning only deletes **ended** sessions (sessions that have been explicitly ended or auto-reset). Active sessions are never pruned. ::: +### Finalize Stale Open Sessions + +If an older Hermes version exited a one-shot CLI run without recording its end, +repair those rows before pruning them. This command is deliberately dry-run by +default and requires both a source and a minimum age: + +```bash +hermes sessions finalize-stale --source cli --older-than 2 +hermes sessions finalize-stale --source cli --older-than 2 --apply --offline --yes +``` + +Stop other Hermes CLI processes before applying the repair. The command ignores +ended and archived sessions, marks matching rows with `end_reason=stale_repair`, +and can be repeated safely. It does not delete messages; a later, separately +reviewed `sessions prune` command performs deletion. + ### Bulk-Archive Sessions If you want sessions out of your listings without deleting anything, @@ -504,6 +520,26 @@ archive your entire history. Archived sessions are hidden from `hermes sessions list` and `/resume` but remain in the database and can be unarchived from the Desktop/Dashboard session list. +### Drop the Trigram Search Index + +The optional trigram FTS index accelerates longer CJK substring searches but can +be substantially larger than the normal full-text index. Disable it in config, +stop Hermes writers, then remove the derived index: + +```yaml +sessions: + trigram_enabled: false +``` + +```bash +hermes sessions drop-trigram --yes +hermes sessions optimize +``` + +Normal English full-text search is unchanged. CJK substring searches fall back +to a slower `LIKE` query. The removal command refuses to run while the config +setting is enabled and is idempotent. + ### Session Statistics ```bash @@ -681,6 +717,7 @@ Key tables in `state.db`: - **sessions** — session metadata (id, source, user_id, model, title, timestamps, token counts). Titles have a unique index (NULL titles allowed, only non-NULL must be unique). - **messages** — full message history (role, content, tool_calls, tool_name, token_count) - **messages_fts** — FTS5 virtual table for full-text search across message content +- **messages_fts_trigram** — optional FTS5 trigram index for longer CJK substring search; absent when `sessions.trigram_enabled` is false ## Session Expiry and Cleanup @@ -688,7 +725,7 @@ Key tables in `state.db`: - Gateway sessions auto-reset based on the configured reset policy - Before reset, the agent saves memories and skills from the expiring session -- Opt-in auto-pruning: when `sessions.auto_prune` is `true`, ended sessions older than `sessions.retention_days` (default 90) are pruned at CLI/gateway startup +- Opt-in auto-pruning: when `sessions.auto_prune` is `true`, ended sessions use `sessions.retention_days_by_source` when their source is listed and `sessions.retention_days` as the fallback - After a prune that actually removed rows, `state.db` is `VACUUM`ed to reclaim disk space (SQLite does not shrink the file on plain DELETE) - Pruning runs at most once per `sessions.min_interval_hours` (default 24); the last-run timestamp is tracked inside `state.db` itself so it's shared across every Hermes process in the same `HERMES_HOME` @@ -698,11 +735,21 @@ Default is **off** — session history is valuable for `session_search` recall, sessions: auto_prune: true # opt in — default is false retention_days: 90 # keep ended sessions this many days - vacuum_after_prune: true # reclaim disk space after a pruning sweep + retention_days_by_source: + cron: 7 + cli: 30 + discord: 90 + trigram_enabled: false # omit the larger optional CJK substring index + vacuum_after_prune: false # reuse freed pages; compact explicitly offline min_interval_hours: 24 # don't re-run the sweep more often than this ``` -Active sessions are never auto-pruned, regardless of age. +Active and archived sessions are never auto-pruned, regardless of age. Sources +not listed in `retention_days_by_source` use the global `retention_days` value. +For high-volume installations, leave `vacuum_after_prune` false so a daily +prune does not rewrite the entire database. Run `hermes sessions optimize` +offline after a large one-time cleanup when disk space needs to be returned to +the operating system. ### Manual Cleanup @@ -716,6 +763,15 @@ hermes sessions delete # Export before pruning (backup) hermes sessions export backup.jsonl hermes sessions prune --older-than 30 --yes + +# Repair leaked one-shot rows before a scoped prune +hermes sessions finalize-stale --source cli --older-than 2 +hermes sessions finalize-stale --source cli --older-than 2 --apply --offline --yes +hermes sessions prune --source cli --older-than 30 --dry-run + +# Remove the optional trigram index after disabling it in config +hermes sessions drop-trigram --yes +hermes sessions optimize ``` :::tip From 6ae6d83015001fb34feb00095b681584a7c80cd9 Mon Sep 17 00:00:00 2001 From: JR Razmus Date: Wed, 15 Jul 2026 19:30:46 -0500 Subject: [PATCH 02/10] fix: skip redundant snapshot verification --- hermes_cli/backup.py | 2 ++ 1 file changed, 2 insertions(+) diff --git a/hermes_cli/backup.py b/hermes_cli/backup.py index 76a733fbc2aec..2a2dabd1187bc 100644 --- a/hermes_cli/backup.py +++ b/hermes_cli/backup.py @@ -1234,6 +1234,8 @@ def _prune_quick_snapshots(root: Path, keep: int = _QUICK_DEFAULT_KEEP) -> int: key=lambda d: d.name, reverse=True, ) + if len(dirs) <= keep: + return 0 deleted = 0 verified_kept = 0 From f1e722fee7ec915e841619e12770342346bec967 Mon Sep 17 00:00:00 2001 From: JR Razmus Date: Wed, 15 Jul 2026 20:37:36 -0500 Subject: [PATCH 03/10] fix: finalize top-level oneshot sessions --- hermes_cli/oneshot.py | 27 +++- tests/hermes_cli/test_tui_resume_flow.py | 155 +++++++++++++++++++++++ 2 files changed, 180 insertions(+), 2 deletions(-) diff --git a/hermes_cli/oneshot.py b/hermes_cli/oneshot.py index e7db628a7c1de..7ef50e9872b57 100644 --- a/hermes_cli/oneshot.py +++ b/hermes_cli/oneshot.py @@ -422,8 +422,31 @@ def _run_agent( agent.stream_delta_callback = None agent.tool_gen_callback = None - result = agent.run_conversation(prompt) - return (result.get("final_response") or "", result) + try: + result = agent.run_conversation(prompt) + return (result.get("final_response") or "", result) + finally: + # Top-level ``hermes -z`` deliberately bypasses ``cli.py``, including + # its normal ``_run_cleanup`` path. Close the agent here so its owned + # SQLite session is finalized and per-session resources are released + # even when the provider/tool turn raises. Then stop process-global MCP + # transports before closing the database handle. Every step is + # best-effort so cleanup can never mask the actual model result/error. + try: + agent.close() + except Exception: + pass + try: + from tools.mcp_tool import shutdown_mcp_servers + + shutdown_mcp_servers() + except BaseException: + pass + try: + if session_db is not None: + session_db.close() + except Exception: + pass def _oneshot_clarify_callback(question: str, choices=None) -> str: diff --git a/tests/hermes_cli/test_tui_resume_flow.py b/tests/hermes_cli/test_tui_resume_flow.py index 7c83ff2a5029d..97627eb39b9d2 100644 --- a/tests/hermes_cli/test_tui_resume_flow.py +++ b/tests/hermes_cli/test_tui_resume_flow.py @@ -886,6 +886,161 @@ def mod(name, **attrs): assert captured["prompt"] == "recall this" +def test_oneshot_closes_agent_and_session_db_after_success(monkeypatch): + """Top-level ``hermes -z`` owns and must finalize its agent lifecycle.""" + from hermes_cli.oneshot import _run_agent + + events = [] + + class FakeSessionDB: + def close(self): + events.append("db.close") + + session_db = FakeSessionDB() + + class FakeAgent: + def __init__(self, **_kwargs): + self.suppress_status_output = False + self.stream_delta_callback = object() + self.tool_gen_callback = object() + + def run_conversation(self, _prompt): + events.append("run") + return {"final_response": "ok", "failed": False, "partial": False} + + def close(self): + events.append("agent.close") + + def mod(name, **attrs): + module = types.ModuleType(name) + for key, value in attrs.items(): + setattr(module, key, value) + return module + + monkeypatch.setattr( + "hermes_cli.oneshot._create_session_db_for_oneshot", lambda: session_db + ) + monkeypatch.setitem(sys.modules, "run_agent", mod("run_agent", AIAgent=FakeAgent)) + monkeypatch.setitem( + sys.modules, + "hermes_cli.config", + mod("hermes_cli.config", load_config=lambda: {"model": {"default": "m"}}), + ) + monkeypatch.setitem( + sys.modules, + "hermes_cli.models", + mod("hermes_cli.models", detect_provider_for_model=lambda *_args, **_kwargs: None), + ) + monkeypatch.setitem( + sys.modules, + "hermes_cli.runtime_provider", + mod( + "hermes_cli.runtime_provider", + resolve_runtime_provider=lambda **_kwargs: { + "api_key": "k", + "base_url": "u", + "provider": "p", + "api_mode": "chat_completions", + "credential_pool": None, + }, + ), + ) + monkeypatch.setitem( + sys.modules, + "hermes_cli.tools_config", + mod("hermes_cli.tools_config", _get_platform_tools=lambda *_args, **_kwargs: set()), + ) + monkeypatch.setitem( + sys.modules, + "tools.mcp_tool", + mod( + "tools.mcp_tool", + shutdown_mcp_servers=lambda: events.append("mcp.shutdown"), + ), + ) + + text, result = _run_agent("hello") + + assert text == "ok" + assert not result.get("failed") + assert events == ["run", "agent.close", "mcp.shutdown", "db.close"] + + +def test_oneshot_closes_agent_and_session_db_after_failure(monkeypatch): + """Cleanup also runs when the one-shot provider/tool turn raises.""" + import pytest + + import hermes_cli.oneshot as oneshot_mod + + events = [] + + class FakeSessionDB: + def close(self): + events.append("db.close") + + class FakeAgent: + def __init__(self, **_kwargs): + self.suppress_status_output = False + self.stream_delta_callback = object() + self.tool_gen_callback = object() + + def run_conversation(self, _prompt): + events.append("run") + raise RuntimeError("boom") + + def close(self): + events.append("agent.close") + + monkeypatch.setattr(oneshot_mod, "_create_session_db_for_oneshot", FakeSessionDB) + monkeypatch.setattr(oneshot_mod, "_normalize_toolsets", lambda _value=None: []) + monkeypatch.setattr(oneshot_mod, "get_fallback_chain", lambda _cfg: []) + monkeypatch.setitem( + sys.modules, + "run_agent", + types.SimpleNamespace(AIAgent=FakeAgent), + ) + monkeypatch.setitem( + sys.modules, + "hermes_cli.config", + types.SimpleNamespace(load_config=lambda: {"model": {"default": "m"}}), + ) + monkeypatch.setitem( + sys.modules, + "hermes_cli.models", + types.SimpleNamespace(detect_provider_for_model=lambda *_args, **_kwargs: None), + ) + monkeypatch.setitem( + sys.modules, + "hermes_cli.runtime_provider", + types.SimpleNamespace( + resolve_runtime_provider=lambda **_kwargs: { + "api_key": "k", + "base_url": "u", + "provider": "p", + "api_mode": "chat_completions", + "credential_pool": None, + } + ), + ) + monkeypatch.setitem( + sys.modules, + "hermes_cli.tools_config", + types.SimpleNamespace(_get_platform_tools=lambda *_args, **_kwargs: set()), + ) + monkeypatch.setitem( + sys.modules, + "tools.mcp_tool", + types.SimpleNamespace( + shutdown_mcp_servers=lambda: events.append("mcp.shutdown") + ), + ) + + with pytest.raises(RuntimeError, match="boom"): + oneshot_mod._run_agent("hello") + + assert events == ["run", "agent.close", "mcp.shutdown", "db.close"] + + def test_launch_tui_exports_model_provider_and_toolsets(monkeypatch, main_mod): captured = {} active_path_during_call = None From 3538f52cb2b4fab4d2dc0b8b177e5d7ee18744f4 Mon Sep 17 00:00:00 2001 From: JR Razmus Date: Fri, 17 Jul 2026 08:49:12 -0500 Subject: [PATCH 04/10] fix: scope Hermes MCPs to its Codex runtime --- agent/codex_runtime.py | 15 +++++++ agent/transports/codex_app_server_session.py | 6 ++- hermes_cli/codex_runtime_plugin_migration.py | 41 +++++++++++++++++++ hermes_cli/codex_runtime_switch.py | 9 ++-- .../test_codex_app_server_session.py | 20 +++++++++ .../test_codex_runtime_plugin_migration.py | 32 +++++++++++++++ .../test_codex_app_server_integration.py | 35 +++++++++++++++- 7 files changed, 152 insertions(+), 6 deletions(-) diff --git a/agent/codex_runtime.py b/agent/codex_runtime.py index c1e46e8a46042..6f22cde0bf942 100644 --- a/agent/codex_runtime.py +++ b/agent/codex_runtime.py @@ -396,8 +396,23 @@ def _on_codex_event(note: dict) -> None: except Exception: logger.debug("codex tool-progress callback raised", exc_info=True) + codex_extra_args: list[str] = [] + try: + from hermes_cli.config import load_config + from hermes_cli.codex_runtime_plugin_migration import ( + build_runtime_mcp_enable_args, + ) + + codex_extra_args = build_runtime_mcp_enable_args(load_config()) + except Exception: + logger.debug( + "codex app-server: failed to build scoped MCP overrides", + exc_info=True, + ) + agent._codex_session = CodexAppServerSession( cwd=cwd, + extra_args=codex_extra_args, approval_callback=approval_callback, request_routing=_ServerRequestRouting( auto_approve_exec=auto_approve_requests, diff --git a/agent/transports/codex_app_server_session.py b/agent/transports/codex_app_server_session.py index e7600f3be0dd5..b1dbc17d46fce 100644 --- a/agent/transports/codex_app_server_session.py +++ b/agent/transports/codex_app_server_session.py @@ -204,6 +204,7 @@ def __init__( cwd: Optional[str] = None, codex_bin: str = "codex", codex_home: Optional[str] = None, + extra_args: Optional[list[str]] = None, permission_profile: Optional[str] = None, approval_callback: Optional[Callable[..., str]] = None, on_event: Optional[Callable[[dict], None]] = None, @@ -213,6 +214,7 @@ def __init__( self._cwd = cwd or os.getcwd() self._codex_bin = codex_bin self._codex_home = codex_home + self._extra_args = list(extra_args or []) self._permission_profile = ( permission_profile or _HERMES_TO_CODEX_PERMISSION_PROFILE.get( os.environ.get("HERMES_TERMINAL_SECURITY_MODE", "auto"), @@ -245,7 +247,9 @@ def ensure_started(self) -> str: return self._thread_id if self._client is None: self._client = self._client_factory( - codex_bin=self._codex_bin, codex_home=self._codex_home + codex_bin=self._codex_bin, + codex_home=self._codex_home, + extra_args=self._extra_args, ) self._client.initialize( client_name="hermes", diff --git a/hermes_cli/codex_runtime_plugin_migration.py b/hermes_cli/codex_runtime_plugin_migration.py index 4b30d3ebf261d..219c20b275b64 100644 --- a/hermes_cli/codex_runtime_plugin_migration.py +++ b/hermes_cli/codex_runtime_plugin_migration.py @@ -614,6 +614,7 @@ def migrate( discover_plugins: bool = True, default_permission_profile: Optional[str] = ":workspace", expose_hermes_tools: bool = True, + enable_mcp_by_default: bool = True, ) -> MigrationReport: """Translate Hermes mcp_servers config + Codex curated plugins into ~/.codex/config.toml. @@ -640,6 +641,11 @@ def migrate( memory, skills, etc.) as an MCP server in ~/.codex/config.toml so the codex subprocess can call back into Hermes for tools codex doesn't have built in. Set False to opt out. + enable_mcp_by_default: when False, write Hermes-managed MCP entries + disabled in the shared Codex config. Hermes' own app-server + subprocess enables only those entries with scoped ``-c`` + overrides, preventing unrelated Codex Desktop tasks from + spawning and retaining Hermes MCP subprocesses. """ report = MigrationReport(dry_run=dry_run) codex_home = codex_home or Path.home() / ".codex" @@ -662,6 +668,8 @@ def migrate( ) continue translated[str(name)] = out + if not enable_mcp_by_default: + out["enabled"] = False if skipped: report.skipped_keys_per_server[str(name)] = skipped report.migrated.append(str(name)) @@ -695,6 +703,8 @@ def migrate( # and is launched on demand by codex (stdio MCP). if expose_hermes_tools: translated["hermes-tools"] = _build_hermes_tools_mcp_entry() + if not enable_mcp_by_default: + translated["hermes-tools"]["enabled"] = False if "hermes-tools" not in report.migrated: report.migrated.append("hermes-tools") @@ -755,3 +765,34 @@ def migrate( except Exception as exc: report.errors.append(f"could not write {target}: {exc}") return report + + +def build_runtime_mcp_enable_args( + hermes_config: dict, + *, + expose_hermes_tools: bool = True, +) -> list[str]: + """Return Codex ``-c`` overrides for MCPs owned by a Hermes runtime. + + The shared ``~/.codex/config.toml`` keeps these entries disabled so + Codex Desktop tasks do not inherit Hermes' process-heavy MCP surface. + A Hermes-owned app-server process opts back into the enabled subset. + """ + names: set[str] = set() + servers = (hermes_config or {}).get("mcp_servers") or {} + if isinstance(servers, dict): + for raw_name, cfg in servers.items(): + if not isinstance(cfg, dict) or cfg.get("enabled") is False: + continue + translated, _skipped = _translate_one_server(str(raw_name), cfg) + if translated is not None: + names.add(str(raw_name)) + if expose_hermes_tools: + names.add("hermes-tools") + + args: list[str] = [] + for name in sorted(names): + args.extend( + ["-c", f"mcp_servers.{_quote_key(name)}.enabled=true"] + ) + return args diff --git a/hermes_cli/codex_runtime_switch.py b/hermes_cli/codex_runtime_switch.py index 06bff58e2e1d7..92dcac98d6d74 100644 --- a/hermes_cli/codex_runtime_switch.py +++ b/hermes_cli/codex_runtime_switch.py @@ -206,13 +206,14 @@ def _check_binary_cached() -> tuple[bool, Optional[str]]: if ok: msg_lines.append(f"codex CLI: {ver}") # Auto-migrate Hermes' MCP servers + Codex's installed curated - # plugins into ~/.codex/config.toml so the spawned codex subprocess - # sees the same tool surface AND can call back into Hermes for - # browser/web/delegate_task/vision/memory tools (#7 fix). + # plugins into ~/.codex/config.toml. Hermes-owned MCP entries are + # disabled in this shared config so unrelated Codex Desktop tasks do + # not spawn them; Hermes' own app-server process enables the scoped + # subset with per-process config overrides. # Failures are non-fatal — the runtime change still proceeds. try: from hermes_cli.codex_runtime_plugin_migration import migrate - mig_report = migrate(config) + mig_report = migrate(config, enable_mcp_by_default=False) # Tools/MCP servers (excluding the hermes-tools callback, # which is internal plumbing — surface separately). user_servers = [ diff --git a/tests/agent/transports/test_codex_app_server_session.py b/tests/agent/transports/test_codex_app_server_session.py index 6c620ec595196..c1c29ee30f1dd 100644 --- a/tests/agent/transports/test_codex_app_server_session.py +++ b/tests/agent/transports/test_codex_app_server_session.py @@ -113,6 +113,26 @@ def make_session(client: FakeClient, **kwargs) -> CodexAppServerSession: ) +def test_session_forwards_scoped_codex_args_to_client_factory(): + client = FakeClient() + captured = {} + + def factory(**kwargs): + captured.update(kwargs) + return client + + session = CodexAppServerSession( + cwd="/tmp", + extra_args=["-c", "mcp_servers.hermes-tools.enabled=true"], + client_factory=factory, + ) + session.ensure_started() + + assert captured["extra_args"] == [ + "-c", "mcp_servers.hermes-tools.enabled=true" + ] + + # ---- choice mapping ---- class TestApprovalChoiceMapping: diff --git a/tests/hermes_cli/test_codex_runtime_plugin_migration.py b/tests/hermes_cli/test_codex_runtime_plugin_migration.py index fc6df86c852c3..d1a6b95aac747 100644 --- a/tests/hermes_cli/test_codex_runtime_plugin_migration.py +++ b/tests/hermes_cli/test_codex_runtime_plugin_migration.py @@ -9,6 +9,7 @@ MIGRATION_MARKER, MIGRATION_END_MARKER, _build_hermes_tools_mcp_entry, + build_runtime_mcp_enable_args, _format_toml_value, _looks_like_test_tempdir, _strip_existing_managed_block, @@ -309,6 +310,37 @@ def test_preserves_unrelated_section_after_managed_block(self): # ---- end-to-end migrate(, expose_hermes_tools=False) ---- class TestMigrate: + def test_shared_config_can_disable_hermes_mcp_defaults(self, tmp_path): + migrate( + {"mcp_servers": {"custom": {"command": "custom-mcp"}}}, + codex_home=tmp_path, + discover_plugins=False, + default_permission_profile=None, + expose_hermes_tools=True, + enable_mcp_by_default=False, + ) + text = (tmp_path / "config.toml").read_text() + assert "[mcp_servers.custom]" in text + assert "[mcp_servers.hermes-tools]" in text + assert text.count("enabled = false") == 2 + + def test_runtime_overrides_enable_only_valid_enabled_hermes_mcps(self): + args = build_runtime_mcp_enable_args( + { + "mcp_servers": { + "alpha": {"command": "alpha-mcp"}, + "disabled": {"command": "disabled-mcp", "enabled": False}, + "broken": {"description": "no transport"}, + "quoted.name": {"command": "quoted-mcp"}, + } + } + ) + assert args == [ + "-c", "mcp_servers.alpha.enabled=true", + "-c", "mcp_servers.hermes-tools.enabled=true", + "-c", 'mcp_servers."quoted.name".enabled=true', + ] + def test_no_servers_no_plugins_no_perms_writes_placeholder(self, tmp_path): report = migrate({}, codex_home=tmp_path, discover_plugins=False, diff --git a/tests/run_agent/test_codex_app_server_integration.py b/tests/run_agent/test_codex_app_server_integration.py index c88d45c483af1..5fd4ca2984f98 100644 --- a/tests/run_agent/test_codex_app_server_integration.py +++ b/tests/run_agent/test_codex_app_server_integration.py @@ -377,6 +377,40 @@ def fake_run_turn(self, user_input: str, **kwargs): assert captured["cwd"] == str(tmp_path) + def test_codex_session_receives_scoped_hermes_mcp_overrides(self, monkeypatch): + from agent.transports.codex_app_server_session import ( + CodexAppServerSession, TurnResult, + ) + + captured = {} + + def fake_init(self, **kwargs): + captured.update(kwargs) + self._thread_id = "thread-stub-1" + + def fake_run_turn(self, user_input: str, **kwargs): + return TurnResult( + final_text="ok", + projected_messages=[{"role": "assistant", "content": "ok"}], + turn_id="turn-stub-1", + thread_id="thread-stub-1", + ) + + monkeypatch.setattr(CodexAppServerSession, "__init__", fake_init) + monkeypatch.setattr(CodexAppServerSession, "run_turn", fake_run_turn) + monkeypatch.setattr( + "hermes_cli.codex_runtime_plugin_migration.build_runtime_mcp_enable_args", + lambda config: ["-c", "mcp_servers.hermes-tools.enabled=true"], + ) + + agent = _make_codex_agent() + with patch.object(agent, "_spawn_background_review", return_value=None): + agent.run_conversation("hi") + + assert captured["extra_args"] == [ + "-c", "mcp_servers.hermes-tools.enabled=true" + ] + def _capture_routing_agent(self, monkeypatch): """Build a codex agent with a CodexAppServerSession stub that captures the request_routing passed at construction time, so we can assert how @@ -769,4 +803,3 @@ def fake_run_turn(self, user_input, **kwargs): assert "on_event" in captured_init and captured_init["on_event"] is not None assert ("tool.started", "exec_command", "pytest") in events - From b7d7cfd0ecd761366f23d159ac7559037702a837 Mon Sep 17 00:00:00 2001 From: JR Razmus Date: Sat, 18 Jul 2026 16:42:29 -0500 Subject: [PATCH 05/10] fix: close Codex runtime on agent teardown --- run_agent.py | 8 +++++ .../test_codex_app_server_lifecycle.py | 33 +++++++++++++++++++ 2 files changed, 41 insertions(+) create mode 100644 tests/run_agent/test_codex_app_server_lifecycle.py diff --git a/run_agent.py b/run_agent.py index bcacec3909b69..34b5e22133b5f 100644 --- a/run_agent.py +++ b/run_agent.py @@ -3601,6 +3601,14 @@ def close(self) -> None: except Exception: pass + try: + codex_session = getattr(self, "_codex_session", None) + if codex_session is not None: + self._codex_session = None + codex_session.close() + except Exception: + pass + # 6. Free conversation history. Mirrors _release_evicted_agent_soft's # soft-eviction clear — close() is the hard teardown for true session # boundaries (/new, /reset, session expiry), so the message list won't diff --git a/tests/run_agent/test_codex_app_server_lifecycle.py b/tests/run_agent/test_codex_app_server_lifecycle.py new file mode 100644 index 0000000000000..49ed592642938 --- /dev/null +++ b/tests/run_agent/test_codex_app_server_lifecycle.py @@ -0,0 +1,33 @@ +import threading + +from run_agent import AIAgent + + +class _FakeCodexSession: + def __init__(self): + self.close_calls = 0 + + def close(self): + self.close_calls += 1 + + +def test_agent_close_releases_codex_app_server_session(monkeypatch): + agent = AIAgent.__new__(AIAgent) + agent.session_id = "test-codex-lifecycle" + agent.client = None + agent._active_children_lock = threading.Lock() + agent._active_children = set() + agent._end_session_on_close = False + agent._session_messages = ["retained"] + codex_session = _FakeCodexSession() + agent._codex_session = codex_session + + monkeypatch.setattr("run_agent.cleanup_vm", lambda _task_id: None) + monkeypatch.setattr("run_agent.cleanup_browser", lambda _task_id: None) + + agent.close() + agent.close() + + assert codex_session.close_calls == 1 + assert agent._codex_session is None + assert agent._session_messages == [] From 28d21c08e63d8bc9cc009c2147ad45dbb3b038ce Mon Sep 17 00:00:00 2001 From: JR Razmus Date: Tue, 21 Jul 2026 07:14:01 -0500 Subject: [PATCH 06/10] fix: bound replayed Responses call IDs --- agent/codex_responses_adapter.py | 25 +++++++++-- tests/agent/test_codex_responses_adapter.py | 48 +++++++++++++++++++++ 2 files changed, 70 insertions(+), 3 deletions(-) diff --git a/agent/codex_responses_adapter.py b/agent/codex_responses_adapter.py index bce372ebb5da6..dd37afaeafd0f 100644 --- a/agent/codex_responses_adapter.py +++ b/agent/codex_responses_adapter.py @@ -294,6 +294,25 @@ def _responses_tools(tools: Optional[List[Dict[str, Any]]] = None) -> Optional[L # ids (msg_...) stay well under this cap and are worth keeping for # prefix-cache hits. Drop only the oversized ones on replay. _MAX_RESPONSES_ITEM_ID_LENGTH = 64 +_MAX_RESPONSES_CALL_ID_LENGTH = 64 + + +def _normalize_responses_call_id(value: str) -> str: + """Return a stable Responses-compatible tool call id. + + Codex app-server can mint MCP call ids longer than the Responses API's + 64-character input limit. Background review replays those calls through + the Responses transport, so normalize only at this outbound boundary. + The stable suffix keeps matching function_call/function_call_output pairs + aligned without mutating the stored conversation or live app-server ids. + """ + call_id = value.strip() + if len(call_id) <= _MAX_RESPONSES_CALL_ID_LENGTH: + return call_id + + digest = hashlib.sha256(call_id.encode("utf-8", errors="replace")).hexdigest()[:16] + prefix_length = _MAX_RESPONSES_CALL_ID_LENGTH - len(digest) - 1 + return f"{call_id[:prefix_length]}_{digest}" def _normalize_responses_message_status(value: Any, *, default: str = "completed") -> str: @@ -633,7 +652,7 @@ def _preflight_codex_input_items( normalized.append( { "type": "function_call", - "call_id": call_id.strip(), + "call_id": _normalize_responses_call_id(call_id), "name": name.strip(), "arguments": arguments, } @@ -674,7 +693,7 @@ def _preflight_codex_input_items( normalized.append( { "type": "function_call_output", - "call_id": call_id.strip(), + "call_id": _normalize_responses_call_id(call_id), "output": cleaned if cleaned else "", } ) @@ -685,7 +704,7 @@ def _preflight_codex_input_items( normalized.append( { "type": "function_call_output", - "call_id": call_id.strip(), + "call_id": _normalize_responses_call_id(call_id), "output": output, } ) diff --git a/tests/agent/test_codex_responses_adapter.py b/tests/agent/test_codex_responses_adapter.py index eb690a9e3b82d..7a8cad8de578c 100644 --- a/tests/agent/test_codex_responses_adapter.py +++ b/tests/agent/test_codex_responses_adapter.py @@ -307,6 +307,54 @@ def test_preflight_codex_api_kwargs_drops_oversized_message_id_end_to_end(): assert "id" not in message_item +def test_preflight_codex_input_items_normalizes_oversized_paired_call_ids(): + oversized_call_id = "codex_mcp__node_repl_js_" + ("a" * 42) + assert len(oversized_call_id) == 66 + + raw_items = [ + { + "type": "function_call", + "call_id": oversized_call_id, + "name": "mcp.node_repl.js", + "arguments": "{}", + }, + { + "type": "function_call_output", + "call_id": oversized_call_id, + "output": "ok", + }, + ] + + first = _preflight_codex_input_items(raw_items) + second = _preflight_codex_input_items(raw_items) + + normalized_call_id = first[0]["call_id"] + assert normalized_call_id == first[1]["call_id"] + assert normalized_call_id == second[0]["call_id"] + assert normalized_call_id != oversized_call_id + assert len(normalized_call_id) == 64 + + +def test_preflight_codex_input_items_keeps_short_paired_call_ids(): + raw_items = [ + { + "type": "function_call", + "call_id": "call_short", + "name": "terminal", + "arguments": "{}", + }, + { + "type": "function_call_output", + "call_id": "call_short", + "output": "ok", + }, + ] + + items = _preflight_codex_input_items(raw_items) + + assert [item["call_id"] for item in items] == ["call_short", "call_short"] + + # --------------------------------------------------------------------------- # _preflight_codex_api_kwargs — built-in (provider-executed) tools must pass # through validation. Regression guard for the xAI native web_search From 9ea06b0a4b8476412a5e36c5c37b1b5d7c7a9649 Mon Sep 17 00:00:00 2001 From: JR Razmus Date: Tue, 21 Jul 2026 09:04:08 -0500 Subject: [PATCH 07/10] fix: preserve Discord threaded front door --- gateway/authz_mixin.py | 17 ++ gateway/config.py | 4 + plugins/platforms/discord/adapter.py | 156 ++++++++++++++++-- tests/gateway/test_discord_bot_auth_bypass.py | 14 ++ tests/gateway/test_discord_free_response.py | 98 +++++++++++ 5 files changed, 279 insertions(+), 10 deletions(-) diff --git a/gateway/authz_mixin.py b/gateway/authz_mixin.py index 884a60948c9c9..4e2f1d77bbd48 100644 --- a/gateway/authz_mixin.py +++ b/gateway/authz_mixin.py @@ -465,6 +465,23 @@ def _is_user_authorized(self, source: SessionSource) -> bool: # In multiplex gateways, route to the per-profile PairingStore so each # profile's whitelist is isolated; falls back to the global store when # the source has no profile or the profile isn't registered. + if getattr(source, "is_bot", False): + bot_allowlist_raw = ( + os.getenv("DISCORD_ALLOWED_BOTS", "") + or os.getenv("DISCORD_ALLOWED_BOT_USERS", "") + if source.platform == Platform.DISCORD + else "" + ) + if bot_allowlist_raw: + allowed_bot_ids = { + part.strip() for part in bot_allowlist_raw.split(",") if part.strip() + } + if "*" in allowed_bot_ids or user_id in allowed_bot_ids: + return True + allow_bots_var = platform_allow_bots_map.get(source.platform) + if allow_bots_var and os.getenv(allow_bots_var, "none").lower().strip() in {"mentions", "all"}: + return True + platform_name = source.platform.value if source.platform else "" pairing_store = self._pairing_store_for(source) if pairing_store is not None and pairing_store.is_approved(platform_name, user_id): diff --git a/gateway/config.py b/gateway/config.py index cc80cda40155d..00cfa7e12561e 100644 --- a/gateway/config.py +++ b/gateway/config.py @@ -1461,6 +1461,10 @@ def _merge_platform_map(source_platforms: Any) -> None: bridged["allowed_topics"] = platform_cfg["allowed_topics"] if "free_response_channels" in platform_cfg: bridged["free_response_channels"] = platform_cfg["free_response_channels"] + if plat == Platform.DISCORD and "auto_thread_free_response" in platform_cfg: + bridged["auto_thread_free_response"] = platform_cfg["auto_thread_free_response"] + if plat == Platform.DISCORD and "self_message_channels" in platform_cfg: + bridged["self_message_channels"] = platform_cfg["self_message_channels"] if "mention_patterns" in platform_cfg: bridged["mention_patterns"] = platform_cfg["mention_patterns"] if "exclusive_bot_mentions" in platform_cfg: diff --git a/plugins/platforms/discord/adapter.py b/plugins/platforms/discord/adapter.py index c31b928ad8473..742fd9a4fb703 100644 --- a/plugins/platforms/discord/adapter.py +++ b/plugins/platforms/discord/adapter.py @@ -25,7 +25,7 @@ import time from collections import defaultdict from contextlib import suppress -from typing import Callable, Dict, List, Optional, Any, Tuple +from typing import Awaitable, Callable, Dict, List, Optional, Any, Tuple, cast from agent.async_utils import ( consume_detached_task_result as _consume_background_task_result, @@ -1259,19 +1259,28 @@ def _discord_message_admission( return False, False elif self._dedup.contains(message_id): return False, False - if message.author == self._client.user: + self_authored = message.author == self._client.user + if self_authored and not self._discord_self_message_allowed(message): return False, False if message.type not in {discord.MessageType.default, discord.MessageType.reply}: return False, False - role_authorized = False + role_authorized = self_authored if getattr(message.author, "bot", False): - allow_bots = os.getenv("DISCORD_ALLOW_BOTS", "none").lower().strip() - if allow_bots == "none": - return False, False - if allow_bots == "mentions" and not self._self_is_explicitly_mentioned(message): - return False, False - if ( + allowed_bot_ids = self._discord_allowed_bot_users() + author_id = str(getattr(message.author, "id", "")) + explicitly_allowed = ( + self_authored + or "*" in allowed_bot_ids + or author_id in allowed_bot_ids + ) + if not explicitly_allowed: + allow_bots = os.getenv("DISCORD_ALLOW_BOTS", "none").lower().strip() + if allow_bots == "none": + return False, False + if allow_bots == "mentions" and not self._self_is_explicitly_mentioned(message): + return False, False + if not self_authored and ( self._discord_bots_require_inline_mention() and not self._self_is_raw_mentioned(message) ): @@ -5707,6 +5716,11 @@ def _discord_free_response_channels(self) -> set: raw = self.config.extra.get("free_response_channels") if raw is None: raw = os.getenv("DISCORD_FREE_RESPONSE_CHANNELS", "") + return self._discord_csv_id_set(raw) + + @staticmethod + def _discord_csv_id_set(raw: Any) -> set: + """Normalize a list/scalar/CSV Discord ID setting into a set.""" if isinstance(raw, list): return {str(part).strip() for part in raw if str(part).strip()} # Coerce non-list scalars (str/int/float) to str before splitting. @@ -5826,6 +5840,52 @@ def _discord_channel_keys_from_channel( return keys + def _discord_allowed_bot_users(self) -> set: + """Return bot user IDs explicitly allowed to trigger Discord sessions.""" + raw = self.config.extra.get("allowed_bot_users") + if raw is None: + raw = self.config.extra.get("allowed_bots") + if raw is None: + raw = os.getenv("DISCORD_ALLOWED_BOTS", "") or os.getenv( + "DISCORD_ALLOWED_BOT_USERS", "" + ) + return self._discord_csv_id_set(raw) + + @staticmethod + def _discord_truthy(raw: Any, *, default: bool = False) -> bool: + """Parse bool-like Discord config values.""" + if raw is None: + return default + if isinstance(raw, str): + return raw.strip().lower() in {"true", "1", "yes", "on"} + return bool(raw) + + def _discord_auto_thread_free_response(self) -> bool: + """Return whether free-response root messages should get auto-threads.""" + configured = self.config.extra.get("auto_thread_free_response") + if configured is None: + configured = os.getenv("DISCORD_AUTO_THREAD_FREE_RESPONSE") + return self._discord_truthy(configured, default=False) + + def _discord_self_message_channels(self) -> set: + """Return top-level Discord channel IDs where self messages are accepted.""" + raw = self.config.extra.get("self_message_channels") + if raw is None: + raw = os.getenv("DISCORD_SELF_MESSAGE_CHANNELS", "") + return self._discord_csv_id_set(raw) + + def _discord_self_message_allowed(self, message: Any) -> bool: + """Return True for opt-in top-level self-authored dispatch messages.""" + dm_channel = getattr(discord, "DMChannel", ()) + thread_channel = getattr(discord, "Thread", ()) + if isinstance(message.channel, dm_channel) or isinstance(message.channel, thread_channel): + return False + allowed = self._discord_self_message_channels() + if not allowed: + return False + channel_id = str(getattr(message.channel, "id", "")) + return "*" in allowed or channel_id in allowed + def _discord_thread_require_mention(self) -> bool: """Return whether thread participation requires @mention to follow up. @@ -6216,6 +6276,66 @@ def _derive_auto_thread_name(self, content: str) -> str: thread_name = thread_name[:77] + "..." return thread_name + @staticmethod + def _manual_job_dispatch_start_message(content: str) -> Optional[str]: + """Return the visible thread-starter message for job dispatch prompts.""" + text = content or "" + if "manual-job-application-dispatch" not in text: + return None + if "one automated job application" not in text.lower(): + return None + + def _line_value(label: str) -> Optional[str]: + match = re.search(rf"(?im)^\s*{re.escape(label)}:\s*(.+?)\s*$", text) + if not match: + return None + value = re.sub(r"\s+", " ", match.group(1)).strip() + return value[:160] if value else None + + company = _line_value("Company") + title = _line_value("Title") + if not company or not title: + queue_id = _line_value("Queue ID") + if queue_id: + parts = queue_id.split("|") + if not company and parts: + company = parts[0].strip() + if not title and len(parts) > 1: + title = parts[1].strip() + + if company and title: + subject = f"{company}: {title}" + elif company or title: + subject = company or title + else: + subject = "this job" + return f"Starting application process for {subject}.\nI'll report the result in this thread." + + async def _send_manual_job_dispatch_thread_start(self, thread: Any, content: str) -> None: + """Post a visible first message in auto-created job-dispatch threads.""" + starter = self._manual_job_dispatch_start_message(content) + if not starter: + return + send = getattr(thread, "send", None) + if not callable(send): + logger.debug( + "[%s] Cannot post job-dispatch starter; thread %s has no send()", + self.name, + getattr(thread, "id", "unknown"), + ) + return + try: + result = send(starter) + if hasattr(result, "__await__"): + await cast(Awaitable[Any], result) + except Exception as exc: + logger.warning( + "[%s] Failed to post job-dispatch starter in thread %s: %s", + self.name, + getattr(thread, "id", "unknown"), + exc, + ) + async def _auto_create_thread(self, message: 'DiscordMessage') -> Optional[Any]: """Create a thread from a user message for auto-threading. @@ -7162,7 +7282,9 @@ async def _handle_message( if not is_thread and not isinstance(message.channel, discord.DMChannel): no_thread_channels_raw = os.getenv("DISCORD_NO_THREAD_CHANNELS", "") no_thread_channels = {ch.strip() for ch in no_thread_channels_raw.split(",") if ch.strip()} - skip_thread = bool(channel_keys & no_thread_channels) or is_free_channel + skip_thread = bool(channel_keys & no_thread_channels) or ( + is_free_channel and not self._discord_auto_thread_free_response() + ) auto_thread = os.getenv("DISCORD_AUTO_THREAD", "true").lower() in {"true", "1", "yes"} is_reply_message = getattr(message, "type", None) == discord.MessageType.reply if auto_thread and not skip_thread and not is_voice_linked_channel and not is_reply_message: @@ -7183,6 +7305,7 @@ async def _handle_message( # event is dropped before it can trigger a second agent run. # Fixes #51057. self._dedup.is_duplicate(str(thread.id)) + await self._send_manual_job_dispatch_thread_start(thread, normalized_content) else: # Auto-threading is the configured routing target for this # message; if it fails we must NOT silently fall back to an @@ -9349,8 +9472,21 @@ def _apply_yaml_config(yaml_cfg: dict, discord_cfg: dict) -> dict | None: os.environ["DISCORD_FREE_RESPONSE_CHANNELS"] = str(frc) if "auto_thread" in discord_cfg and not os.getenv("DISCORD_AUTO_THREAD"): os.environ["DISCORD_AUTO_THREAD"] = str(discord_cfg["auto_thread"]).lower() + if "auto_thread_free_response" in discord_cfg and not os.getenv( + "DISCORD_AUTO_THREAD_FREE_RESPONSE" + ): + os.environ["DISCORD_AUTO_THREAD_FREE_RESPONSE"] = str( + discord_cfg["auto_thread_free_response"] + ).lower() if "reactions" in discord_cfg and not os.getenv("DISCORD_REACTIONS"): os.environ["DISCORD_REACTIONS"] = str(discord_cfg["reactions"]).lower() + if "allow_bots" in discord_cfg and not os.getenv("DISCORD_ALLOW_BOTS"): + os.environ["DISCORD_ALLOW_BOTS"] = str(discord_cfg["allow_bots"]).lower() + allowed_bots = discord_cfg.get("allowed_bot_users", discord_cfg.get("allowed_bots")) + if allowed_bots is not None and not os.getenv("DISCORD_ALLOWED_BOTS"): + if isinstance(allowed_bots, list): + allowed_bots = ",".join(str(v) for v in allowed_bots) + os.environ["DISCORD_ALLOWED_BOTS"] = str(allowed_bots) seeded_extra = {} backfill_cfg = discord_cfg.get("missed_message_backfill") if isinstance(backfill_cfg, dict): diff --git a/tests/gateway/test_discord_bot_auth_bypass.py b/tests/gateway/test_discord_bot_auth_bypass.py index 71be4edfb6ccf..a2fdacd580d21 100644 --- a/tests/gateway/test_discord_bot_auth_bypass.py +++ b/tests/gateway/test_discord_bot_auth_bypass.py @@ -28,6 +28,8 @@ def _isolate_discord_env(monkeypatch): """ for var in ( "DISCORD_ALLOW_BOTS", + "DISCORD_ALLOWED_BOTS", + "DISCORD_ALLOWED_BOT_USERS", "DISCORD_ALLOWED_USERS", "DISCORD_ALLOWED_ROLES", "DISCORD_ALLOW_ALL_USERS", @@ -109,6 +111,18 @@ def test_discord_bot_authorized_when_allow_bots_all(monkeypatch): assert runner._is_user_authorized(source) is True +def test_discord_bot_authorized_when_explicitly_allowlisted(monkeypatch): + """DISCORD_ALLOWED_BOTS authorizes a specific bot without opening all bot senders.""" + runner = _make_bare_runner() + + monkeypatch.setenv("DISCORD_ALLOW_BOTS", "none") + monkeypatch.setenv("DISCORD_ALLOWED_BOTS", "999888777") + monkeypatch.setenv("DISCORD_ALLOWED_USERS", "100200300") + + assert runner._is_user_authorized(_make_discord_bot_source("999888777")) is True + assert runner._is_user_authorized(_make_discord_bot_source("111222333")) is False + + def test_discord_bot_NOT_authorized_when_allow_bots_none(monkeypatch): """DISCORD_ALLOW_BOTS=none (default) must still reject bots that aren't in DISCORD_ALLOWED_USERS — preserves the original security behavior. diff --git a/tests/gateway/test_discord_free_response.py b/tests/gateway/test_discord_free_response.py index 6b0c4c32753dd..a14407b574af1 100644 --- a/tests/gateway/test_discord_free_response.py +++ b/tests/gateway/test_discord_free_response.py @@ -88,6 +88,11 @@ def __init__(self, channel_id: int = 1, name: str = "thread", parent=None, guild self.parent_id = getattr(parent, "id", None) self.guild = getattr(parent, "guild", None) or SimpleNamespace(name=guild_name) self.topic = None + self.sent_messages = [] + + async def send(self, content): + self.sent_messages.append(content) + return SimpleNamespace(id=len(self.sent_messages), content=content) def history(self, *, limit, before, after=None, oldest_first=None): async def _iter(): @@ -116,6 +121,8 @@ def adapter(monkeypatch): "DISCORD_HISTORY_BACKFILL", "DISCORD_HISTORY_BACKFILL_LIMIT", "DISCORD_ALLOW_BOTS", + "DISCORD_AUTO_THREAD_FREE_RESPONSE", + "DISCORD_SELF_MESSAGE_CHANNELS", ): monkeypatch.delenv(_var, raising=False) @@ -685,6 +692,97 @@ async def test_discord_free_response_channel_skips_auto_thread(adapter, monkeypa assert event.source.chat_type == "group" +@pytest.mark.asyncio +async def test_discord_free_response_channel_can_auto_thread_when_enabled(adapter, monkeypatch): + """Opt-in free-response channels can still auto-thread root messages.""" + monkeypatch.setenv("DISCORD_REQUIRE_MENTION", "true") + monkeypatch.setenv("DISCORD_FREE_RESPONSE_CHANNELS", "789") + monkeypatch.setenv("DISCORD_AUTO_THREAD_FREE_RESPONSE", "true") + monkeypatch.delenv("DISCORD_AUTO_THREAD", raising=False) # default true + + thread = FakeThread(channel_id=790, name="job thread", parent=FakeTextChannel(channel_id=789)) + adapter._auto_create_thread = AsyncMock(return_value=thread) + + message = make_message( + channel=FakeTextChannel(channel_id=789), + content="apply to this job", + ) + + await adapter._handle_message(message) + + adapter._auto_create_thread.assert_awaited_once() + adapter.handle_message.assert_awaited_once() + event = adapter.handle_message.await_args.args[0] + assert event.text == "apply to this job" + assert event.source.chat_type == "thread" + assert event.source.thread_id == "790" + assert thread.sent_messages == [] + + +@pytest.mark.asyncio +async def test_discord_job_dispatch_auto_thread_posts_visible_start(adapter, monkeypatch): + monkeypatch.setenv("DISCORD_REQUIRE_MENTION", "true") + monkeypatch.setenv("DISCORD_FREE_RESPONSE_CHANNELS", "789") + monkeypatch.setenv("DISCORD_AUTO_THREAD_FREE_RESPONSE", "true") + monkeypatch.delenv("DISCORD_AUTO_THREAD", raising=False) + + thread = FakeThread(channel_id=790, name="job thread", parent=FakeTextChannel(channel_id=789)) + adapter._auto_create_thread = AsyncMock(return_value=thread) + + message = make_message( + channel=FakeTextChannel(channel_id=789), + content=( + "Use the `manual-job-application-dispatch` skill for one automated job application.\n\n" + "Queue ID: dutchie|staff software engineer, front-end|https://example.com|linkedin\n" + "Company: Dutchie\n" + "Title: Staff Software Engineer, Front-End\n" + "ATS: linkedin\n" + "URL: https://example.com\n\n" + "Report result in this thread.\n" + "Why it passed: manual CAPTCHA/bot-security retry" + ), + ) + + await adapter._handle_message(message) + + assert thread.sent_messages == [ + "Starting application process for Dutchie: Staff Software Engineer, Front-End.\n" + "I'll report the result in this thread." + ] + adapter.handle_message.assert_awaited_once() + event = adapter.handle_message.await_args.args[0] + assert event.source.chat_type == "thread" + assert event.source.thread_id == "790" + + +def test_discord_manual_job_dispatch_start_message_falls_back_to_queue_id(adapter): + message = DiscordAdapter._manual_job_dispatch_start_message( + "Use the `manual-job-application-dispatch` skill for one automated job application.\n" + "Queue ID: code metal|principal frontend engineer|https://example.com|linkedin" + ) + + assert message == ( + "Starting application process for code metal: principal frontend engineer.\n" + "I'll report the result in this thread." + ) + + +def test_discord_manual_job_dispatch_start_message_ignores_other_messages(adapter): + assert DiscordAdapter._manual_job_dispatch_start_message("apply to this job") is None + + +def test_discord_self_message_allowed_only_for_configured_top_level_channel(adapter, monkeypatch): + monkeypatch.setenv("DISCORD_SELF_MESSAGE_CHANNELS", "789") + + assert adapter._discord_self_message_allowed( + make_message(channel=FakeTextChannel(channel_id=789), content="dispatch") + ) + assert not adapter._discord_self_message_allowed( + make_message(channel=FakeTextChannel(channel_id=456), content="dispatch") + ) + assert not adapter._discord_self_message_allowed( + make_message(channel=FakeThread(channel_id=790, parent=FakeTextChannel(channel_id=789)), content="reply") + ) @pytest.mark.asyncio From 6e4cc3dd984f7183efe1152cc0d8a3403ad0e023 Mon Sep 17 00:00:00 2001 From: JR Razmus Date: Fri, 24 Jul 2026 07:11:46 -0500 Subject: [PATCH 08/10] fix: retire idle Codex app servers --- agent/codex_runtime.py | 16 +++++++++ run_agent.py | 14 ++++++++ .../test_codex_app_server_event_bridge.py | 34 ++++++++++++++++++- tests/gateway/test_agent_cache.py | 19 +++++++++++ 4 files changed, 82 insertions(+), 1 deletion(-) diff --git a/agent/codex_runtime.py b/agent/codex_runtime.py index c0a02f4cefe0c..276d5ee25faec 100644 --- a/agent/codex_runtime.py +++ b/agent/codex_runtime.py @@ -587,6 +587,18 @@ def on_event(note: dict) -> None: if not isinstance(note, dict): return method = note.get("method") or "" + if not isinstance(method, str) or not method: + return + touch = getattr(agent, "_touch_activity", None) + if callable(touch): + try: + touch(f"codex app-server event: {method}") + except Exception: + logger.debug( + "_touch_activity raised for codex app-server event %s", + method, + exc_info=True, + ) params = note.get("params") or {} if not isinstance(params, dict): params = {} @@ -708,6 +720,10 @@ def run_codex_app_server_turn( # standard run_conversation() flow (line ~11823) before the early # return reaches us. Do NOT append again — that would duplicate. + touch = getattr(agent, "_touch_activity", None) + if callable(touch): + touch("codex app-server turn started") + try: turn = agent._codex_session.run_turn(user_input=user_message) except Exception as exc: diff --git a/run_agent.py b/run_agent.py index d315640b68d53..02cce9aad96ba 100644 --- a/run_agent.py +++ b/run_agent.py @@ -3577,6 +3577,8 @@ def release_clients(self) -> None: We DO close: - OpenAI/httpx client pool (big chunk of held memory + sockets; the rebuilt agent gets a fresh client anyway) + - Codex app-server session (subprocess tree; the rebuilt agent + resumes from durable session state with a fresh client) - Active child subagents (per-turn artefacts; safe to drop) Safe to call multiple times. Distinct from close() — which is the @@ -3600,6 +3602,18 @@ def release_clients(self) -> None: except Exception: pass + # Close the Codex app-server subprocess tree. Soft cache eviction + # drops this AIAgent instance permanently, so retaining its transport + # cannot help a later resume; it only leaks the app-server and MCP + # children until gateway shutdown. + try: + codex_session = getattr(self, "_codex_session", None) + if codex_session is not None: + self._codex_session = None + codex_session.close() + except Exception: + pass + # Close the OpenAI/httpx client to release sockets immediately. try: client = getattr(self, "client", None) diff --git a/tests/agent/test_codex_app_server_event_bridge.py b/tests/agent/test_codex_app_server_event_bridge.py index 05032c1c88e4e..6e72783f56ee7 100644 --- a/tests/agent/test_codex_app_server_event_bridge.py +++ b/tests/agent/test_codex_app_server_event_bridge.py @@ -17,7 +17,7 @@ import json from types import SimpleNamespace -from unittest.mock import MagicMock +from unittest.mock import MagicMock, call import pytest @@ -39,6 +39,7 @@ def _make_stub_agent() -> SimpleNamespace: _emit_interim_assistant_message=MagicMock( name="_emit_interim_assistant_message" ), + _touch_activity=MagicMock(name="_touch_activity"), ) @@ -482,6 +483,35 @@ def test_show_commentary_off_suppresses_interim(self): class TestBridgeRobustness: + def test_valid_notifications_refresh_agent_activity(self): + agent = _make_stub_agent() + bridge = make_codex_app_server_event_bridge(agent) + + bridge({"method": "turn/started", "params": {}}) + bridge(_item_started({ + "type": "commandExecution", "id": "exec-live", "command": "ls", + })) + bridge({"method": "item/commandExecution/outputDelta", + "params": {"delta": "still working"}}) + + assert agent._touch_activity.call_args_list == [ + call("codex app-server event: turn/started"), + call("codex app-server event: item/started"), + call( + "codex app-server event: item/commandExecution/outputDelta" + ), + ] + + def test_invalid_notifications_do_not_refresh_agent_activity(self): + agent = _make_stub_agent() + bridge = make_codex_app_server_event_bridge(agent) + + bridge(None) # type: ignore[arg-type] + bridge({}) + bridge({"method": 123, "params": {}}) + + agent._touch_activity.assert_not_called() + def test_non_dict_notification_is_ignored(self): agent = _make_stub_agent() bridge = make_codex_app_server_event_bridge(agent) @@ -608,6 +638,7 @@ def close(self): context_compressor=None, event_callback=None, _session_db=None, + _touch_activity=MagicMock(), ) codex_runtime.run_codex_app_server_turn( @@ -625,6 +656,7 @@ def close(self): assert callable(captured["on_event"]), ( "on_event must be the bridge callable, not None or a sentinel" ) + agent._touch_activity.assert_any_call("codex app-server turn started") # And the bridge must actually drive the agent's callbacks when # fed a representative notification. diff --git a/tests/gateway/test_agent_cache.py b/tests/gateway/test_agent_cache.py index 853a9538f5d0b..d2f1b60fb23c9 100644 --- a/tests/gateway/test_agent_cache.py +++ b/tests/gateway/test_agent_cache.py @@ -1463,6 +1463,25 @@ def test_release_clients_closes_llm_client(self): # Post-release: client reference is dropped (memory freed). assert agent.client is None + def test_release_clients_closes_codex_app_server_session(self): + """Soft eviction must retire the subprocess transport it owns.""" + from run_agent import AIAgent + + agent = AIAgent( + model="anthropic/claude-sonnet-4", api_key="test", + base_url="https://openrouter.ai/api/v1", provider="openrouter", + max_iterations=5, quiet_mode=True, + skip_context_files=True, skip_memory=True, + ) + codex_session = MagicMock() + agent._codex_session = codex_session + + agent.release_clients() + agent.release_clients() + + codex_session.close.assert_called_once_with() + assert agent._codex_session is None + def test_close_vs_release_full_teardown_difference(self, monkeypatch): """close() tears down task state; release_clients() does not. From 6dbb1d58fff2d7c720f736ca0eebf87a68bf1354 Mon Sep 17 00:00:00 2001 From: JR Razmus Date: Fri, 31 Jul 2026 07:11:30 -0500 Subject: [PATCH 09/10] fix: allow long Codex post-tool reasoning --- agent/transports/codex_app_server_session.py | 12 ++- .../test_codex_app_server_session.py | 77 +++++++++++++++++++ 2 files changed, 86 insertions(+), 3 deletions(-) diff --git a/agent/transports/codex_app_server_session.py b/agent/transports/codex_app_server_session.py index b43662d3e3ad6..7643ac90e4ad9 100644 --- a/agent/transports/codex_app_server_session.py +++ b/agent/transports/codex_app_server_session.py @@ -48,6 +48,11 @@ # enough to surface a config/provider/auth diagnostic. _STDERR_TAIL_LINES = 12 +# Reasoning-heavy Codex models can legitimately stay quiet for several minutes +# after a tool result. Keep this below the 10-minute outer turn deadline so a +# truly wedged app-server still retires early without interrupting healthy work. +_POST_TOOL_QUIET_TIMEOUT_SECONDS = 300.0 + # Permission profile mapping mirrors the docstring in PR proposal: # Hermes' tools.terminal.security_mode → Codex's permissions profile id. @@ -373,7 +378,7 @@ def run_turn( *, turn_timeout: float = 600.0, notification_poll_timeout: float = 0.25, - post_tool_quiet_timeout: float = 90.0, + post_tool_quiet_timeout: float = _POST_TOOL_QUIET_TIMEOUT_SECONDS, ) -> TurnResult: """Send a user message and block until turn/completed, while forwarding server-initiated approval requests and projecting items @@ -382,8 +387,9 @@ def run_turn( post_tool_quiet_timeout: if codex emits a tool completion and then goes quiet for this many seconds without emitting another item or `turn/completed`, fast-fail and mark the session for retirement. - Mirrors openclaw beta.8's post-tool completion watchdog (#81697) - so a wedged codex doesn't burn the full turn deadline. + Keeps a wedged codex from burning the full turn deadline while leaving + enough room for reasoning-heavy models to legitimately spend several + minutes processing a tool result before their next event. """ # Pre-create the result so startup failures (codex subprocess can't # spawn, initialize handshake rejects, thread/start blows up) surface diff --git a/tests/agent/transports/test_codex_app_server_session.py b/tests/agent/transports/test_codex_app_server_session.py index 2fe3f3e592a00..907a5bb5c1ea3 100644 --- a/tests/agent/transports/test_codex_app_server_session.py +++ b/tests/agent/transports/test_codex_app_server_session.py @@ -980,6 +980,83 @@ def test_post_tool_quiet_watchdog_trips_and_retires(self): # Confirm we issued turn/interrupt to free codex compute assert any(method == "turn/interrupt" for (method, _) in client.requests) + def test_default_post_tool_watchdog_allows_long_codex_reasoning(self): + """A healthy Codex turn may reason for more than 90 seconds after a + tool result before emitting its final response. + + Keep the watchdog bounded, but do not interrupt a live turn at the + old 90-second threshold tuned for shorter requests. + """ + clock = [0.0] + + class DelayedFinalClient(FakeClient): + def __init__(self) -> None: + super().__init__() + self.tool_delivered = False + self.final_delivered = False + + def take_notification(self, timeout: float = 0.0): + if not self.tool_delivered: + self.tool_delivered = True + return { + "method": "item/completed", + "params": { + "item": { + "type": "commandExecution", + "id": "ex1", + "command": "echo hi", + "cwd": "/tmp", + "status": "completed", + "aggregatedOutput": "hi", + "exitCode": 0, + "commandActions": [], + }, + "threadId": "t", + "turnId": "tu1", + }, + } + if clock[0] < 120.0: + clock[0] += 30.0 + return None + if not self.final_delivered: + self.final_delivered = True + return { + "method": "item/completed", + "params": { + "item": { + "type": "agentMessage", + "id": "m1", + "text": "finished after reasoning", + }, + "threadId": "t", + "turnId": "tu1", + }, + } + return { + "method": "turn/completed", + "params": { + "threadId": "t", + "turn": { + "id": "tu1", + "status": "completed", + "error": None, + }, + }, + } + + client = DelayedFinalClient() + session = make_session(client) + with patch.object(session_mod.time, "monotonic", side_effect=lambda: clock[0]): + result = session.run_turn("tool then long reasoning") + + assert result.final_text == "finished after reasoning" + assert result.error is None + assert result.interrupted is False + assert result.should_retire is False + assert not any( + method == "turn/interrupt" for method, _ in client.requests + ) + def test_post_tool_watchdog_uses_monotonic_clock(self): client = FakeClient() client.queue_notification( From 4d60b9733b0a2dcb3e733a5281a43a38b8233fc0 Mon Sep 17 00:00:00 2001 From: JR Razmus Date: Sun, 2 Aug 2026 07:08:06 -0500 Subject: [PATCH 10/10] fix: honor configured Codex turn timeout --- agent/codex_runtime.py | 18 +++++++++++- .../test_codex_app_server_integration.py | 28 +++++++++++++++++++ 2 files changed, 45 insertions(+), 1 deletion(-) diff --git a/agent/codex_runtime.py b/agent/codex_runtime.py index 276d5ee25faec..de0f351c8f12b 100644 --- a/agent/codex_runtime.py +++ b/agent/codex_runtime.py @@ -724,8 +724,24 @@ def run_codex_app_server_turn( if callable(touch): touch("codex app-server turn started") + # The app-server transport has its own wall-clock deadline, separate from + # the cron inactivity watchdog. Honor the same supported per-provider / + # per-model request timeout used by the other inference transports so a + # configured long-reasoning allowance is not silently capped at the + # transport's 600-second default. + from hermes_cli.timeouts import get_provider_request_timeout + + turn_timeout = get_provider_request_timeout( + getattr(agent, "provider", ""), + getattr(agent, "model", None), + ) + turn_kwargs = {"turn_timeout": turn_timeout} if turn_timeout is not None else {} + try: - turn = agent._codex_session.run_turn(user_input=user_message) + turn = agent._codex_session.run_turn( + user_input=user_message, + **turn_kwargs, + ) except Exception as exc: logger.exception("codex app-server turn failed") # Crash → unconditionally drop the session so the next turn diff --git a/tests/run_agent/test_codex_app_server_integration.py b/tests/run_agent/test_codex_app_server_integration.py index 9b1bc5a111ccc..737b4bcfaac0f 100644 --- a/tests/run_agent/test_codex_app_server_integration.py +++ b/tests/run_agent/test_codex_app_server_integration.py @@ -73,6 +73,34 @@ def test_api_mode_is_codex_app_server(self): class TestRunConversationCodexPath: + def test_configured_provider_timeout_reaches_app_server(self, monkeypatch): + observed = {} + + def fake_run_turn(self, user_input: str, **kwargs): + observed.update(kwargs) + return TurnResult( + final_text="done", + projected_messages=[{"role": "assistant", "content": "done"}], + turn_id="turn-timeout-1", + thread_id="thread-timeout-1", + ) + + monkeypatch.setattr(CodexAppServerSession, "run_turn", fake_run_turn) + monkeypatch.setattr( + CodexAppServerSession, "ensure_started", lambda self: "thread-timeout-1" + ) + monkeypatch.setattr( + "hermes_cli.timeouts.get_provider_request_timeout", + lambda provider, model: 900.0, + ) + + agent = _make_codex_agent() + with patch.object(agent, "_spawn_background_review", return_value=None): + result = agent.run_conversation("hello") + + assert result["completed"] is True + assert observed["turn_timeout"] == 900.0 + def test_run_conversation_returns_codex_shape(self, fake_session): agent = _make_codex_agent() # No background review fork during tests