diff --git a/agent/codex_responses_adapter.py b/agent/codex_responses_adapter.py
index bce372ebb5da..dd37afaeafd0 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/agent/codex_runtime.py b/agent/codex_runtime.py
index 91c2af3e995f..de0f351c8f12 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 = {}
@@ -670,6 +682,22 @@ def run_codex_app_server_turn(
exc_info=True,
)
+ # Keep Hermes' configured MCPs scoped to the Codex app-server runtime
+ # while using the upstream full event bridge for progress/interim text.
+ 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,
+ )
+
# Bridge codex JSON-RPC notifications (item/started, item/completed,
# item/agentMessage/delta, ...) into Hermes' gateway UI callbacks
# (tool_progress_callback, _fire_stream_delta,
@@ -679,6 +707,7 @@ def run_codex_app_server_turn(
# Supersedes the narrower item/started-only bridge from #38835.
agent._codex_session = CodexAppServerSession(
cwd=cwd,
+ extra_args=codex_extra_args,
approval_callback=approval_callback,
request_routing=_ServerRequestRouting(
auto_approve_exec=auto_approve_requests,
@@ -691,8 +720,28 @@ 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")
+
+ # 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/agent/transports/codex_app_server_session.py b/agent/transports/codex_app_server_session.py
index 16c6905c07a5..7643ac90e4ad 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.
@@ -204,6 +209,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 +219,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 +252,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",
@@ -369,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
@@ -378,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/cli.py b/cli.py
index 05dc6d012905..fcddd5aa2e13 100644
--- a/cli.py
+++ b/cli.py
@@ -1322,6 +1322,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:
@@ -1862,17 +1874,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()
@@ -1900,13 +1908,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:
@@ -4235,7 +4237,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/authz_mixin.py b/gateway/authz_mixin.py
index 884a60948c9c..4e2f1d77bbd4 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 cc80cda40155..00cfa7e12561 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/gateway/run.py b/gateway/run.py
index 9184c22b4125..504a0e99def1 100644
--- a/gateway/run.py
+++ b/gateway/run.py
@@ -3396,16 +3396,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 7eba80e50242..09d52b854813 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 2a25217160c0..ec7aa1e4efa9 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
@@ -803,6 +804,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,
@@ -857,7 +866,9 @@ def _too_large(path: Path, rel_name: str) -> bool:
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
@@ -887,12 +898,15 @@ def _too_large(path: Path, rel_name: str) -> bool:
# 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():
@@ -907,12 +921,22 @@ def _too_large(path: Path, rel_name: str) -> bool:
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)
@@ -920,12 +944,14 @@ def _too_large(path: Path, rel_name: str) -> bool:
# 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)
@@ -1041,6 +1067,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"
@@ -1154,18 +1278,33 @@ 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,
reverse=True,
)
+ if len(dirs) <= keep:
+ return 0
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/codex_runtime_plugin_migration.py b/hermes_cli/codex_runtime_plugin_migration.py
index 4b30d3ebf261..219c20b275b6 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 06bff58e2e1d..92dcac98d6d7 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/hermes_cli/config.py b/hermes_cli/config.py
index 70dfa2fda77a..310c02015b63 100644
--- a/hermes_cli/config.py
+++ b/hermes_cli/config.py
@@ -2997,6 +2997,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,
# Durable delivery-obligation ledger: final agent responses are
# recorded in state.db around the platform send, and a gateway that
# died between finalize and platform ACK redelivers the stored
@@ -3192,6 +3193,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
@@ -3229,6 +3232,7 @@ def _ensure_hermes_home_managed(home: Path):
# ``hermes update`` behaviour.
"updates": {
+ "branch": "main",
# Pre-update safety backup — ONE consolidated mechanism, three modes:
#
# quick (default) — snapshot critical small state files (pairing
diff --git a/hermes_cli/gateway.py b/hermes_cli/gateway.py
index 5b33d7646e44..a20ef8990df3 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
@@ -3979,23 +3980,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 526fa9345161..17aced528b64 100644
--- a/hermes_cli/main.py
+++ b/hermes_cli/main.py
@@ -8832,7 +8832,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):
@@ -9843,7 +9855,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
@@ -14280,6 +14292,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",
@@ -14902,6 +14944,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_cli/oneshot.py b/hermes_cli/oneshot.py
index e0c1337698ce..7853ec3844cb 100644
--- a/hermes_cli/oneshot.py
+++ b/hermes_cli/oneshot.py
@@ -458,6 +458,14 @@ def _run_agent(
agent.close()
except Exception:
logging.debug("oneshot agent cleanup failed", exc_info=True)
+ # Top-level ``hermes -z`` skips the normal CLI process cleanup. Stop
+ # process-global MCP transports before closing the database handle.
+ try:
+ from tools.mcp_tool import shutdown_mcp_servers
+
+ shutdown_mcp_servers()
+ except BaseException:
+ logging.debug("oneshot MCP cleanup failed", exc_info=True)
# agent.close() calls session_db.end_session() but leaves the connection
# open; close it here to checkpoint the WAL before os._exit skips
# finalizers.
diff --git a/hermes_state.py b/hermes_state.py
index 8d8ff6bc0213..cce9e825a5e4 100644
--- a/hermes_state.py
+++ b/hermes_state.py
@@ -28,6 +28,7 @@
from agent.memory_manager import sanitize_context
from agent.message_sanitization import _sanitize_surrogates
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__)
@@ -196,15 +197,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.
@@ -1026,6 +1032,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
@@ -1179,20 +1187,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])
@@ -1631,7 +1657,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
@@ -1668,7 +1694,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:
@@ -1699,9 +1728,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) "
@@ -1914,13 +1945,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
)
@@ -1930,6 +1968,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()
@@ -6740,6 +6783,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,
@@ -6786,6 +6830,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)
@@ -6982,6 +7029,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]:
@@ -7523,6 +7636,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.
@@ -7538,9 +7669,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.
"""
@@ -7634,6 +7764,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,
@@ -7657,7 +7788,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")
@@ -7671,10 +7807,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
@@ -7692,7 +7860,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 "",
@@ -7704,6 +7872,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/plugins/platforms/discord/adapter.py b/plugins/platforms/discord/adapter.py
index c31b928ad847..742fd9a4fb70 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/run_agent.py b/run_agent.py
index 6c13f737c861..02cce9aad96b 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)
@@ -3665,6 +3679,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/session_policy.py b/session_policy.py
new file mode 100644
index 000000000000..7fdd681c035a
--- /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/agent/test_codex_app_server_event_bridge.py b/tests/agent/test_codex_app_server_event_bridge.py
index 05032c1c88e4..6e72783f56ee 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/agent/test_codex_responses_adapter.py b/tests/agent/test_codex_responses_adapter.py
index eb690a9e3b82..7a8cad8de578 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
diff --git a/tests/agent/transports/test_codex_app_server_session.py b/tests/agent/transports/test_codex_app_server_session.py
index 956709e9ef4e..907a5bb5c1ea 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:
@@ -960,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(
diff --git a/tests/cli/test_single_query_session_finalize.py b/tests/cli/test_single_query_session_finalize.py
index 3041c03dbfe1..47dd5dd01054 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/gateway/test_agent_cache.py b/tests/gateway/test_agent_cache.py
index 853a9538f5d0..d2f1b60fb23c 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.
diff --git a/tests/gateway/test_discord_bot_auth_bypass.py b/tests/gateway/test_discord_bot_auth_bypass.py
index 71be4edfb6cc..a2fdacd580d2 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 6b0c4c32753d..a14407b574af 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
diff --git a/tests/hermes_cli/test_active_sessions.py b/tests/hermes_cli/test_active_sessions.py
index 560803dc852e..0b851632733b 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 cbac29eae503..691a95763349 100644
--- a/tests/hermes_cli/test_backup.py
+++ b/tests/hermes_cli/test_backup.py
@@ -1627,6 +1627,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 b8b0244531a0..a310004927de 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_codex_runtime_plugin_migration.py b/tests/hermes_cli/test_codex_runtime_plugin_migration.py
index fc6df86c852c..d1a6b95aac74 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/hermes_cli/test_gateway_service.py b/tests/hermes_cli/test_gateway_service.py
index 7041ab9aee4a..b4d4e0086e86 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
@@ -2598,6 +2599,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/hermes_cli/test_tui_resume_flow.py b/tests/hermes_cli/test_tui_resume_flow.py
index 4bcd3a6119fd..22eaea5e4a2b 100644
--- a/tests/hermes_cli/test_tui_resume_flow.py
+++ b/tests/hermes_cli/test_tui_resume_flow.py
@@ -1635,6 +1635,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
diff --git a/tests/run_agent/test_codex_app_server_integration.py b/tests/run_agent/test_codex_app_server_integration.py
index 199ff5d56cb1..737b4bcfaac0 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
@@ -378,6 +406,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
@@ -786,4 +848,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
-
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 000000000000..49ed59264293
--- /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 == []
diff --git a/tests/test_hermes_state.py b/tests/test_hermes_state.py
index dd6ecbc55182..d9bb588c6d91 100644
--- a/tests/test_hermes_state.py
+++ b/tests/test_hermes_state.py
@@ -4877,12 +4877,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()
@@ -6265,8 +6269,168 @@ 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()
# =========================================================================
# Lone-surrogate persistence
# =========================================================================
@@ -6358,4 +6522,3 @@ def test_session_title_survives_lone_surrogate(self, db):
db.create_session("s1", source="cli")
assert db.set_session_title("s1", "title \ud835 bad") is True
assert db.get_session("s1")["title"] == "title \ufffd bad"
-
diff --git a/website/docs/user-guide/sessions.md b/website/docs/user-guide/sessions.md
index daece81f4829..2d802b251ff0 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