fix: harden salvaged session and browser improvements

Polish salvaged contributor work before PR review:
- read browser inactivity timeout from config with documented fallback
- skip redundant v10 trigram backfill before v11 FTS rebuild
- show delegate_task goals safely in progress previews
- show gateway status model/context without redundant token wording
- wire gateway /sessions to shared session-listing helpers
- map Ravenwolf author emails for release attribution

Co-authored-by: Wolfram Ravenwolf <github.com@wolfram.ravenwolf.de>
Co-authored-by: Amy Ravenwolf <amy@ravenwolf.de>
This commit is contained in:
Teknium 2026-06-15 06:04:36 -07:00
parent ead38107a2
commit 3e7e9b24d4
13 changed files with 413 additions and 44 deletions

View file

@ -12,6 +12,7 @@ import time
from dataclasses import dataclass, field
from difflib import unified_diff
from pathlib import Path
from typing import Any
from utils import safe_json_loads
from agent.tool_result_classification import file_mutation_result_landed
@ -168,6 +169,27 @@ def _oneline(text: str) -> str:
return " ".join(text.split())
def _truncate_preview(text: str, max_len: int | None) -> str:
if max_len and max_len > 0 and len(text) > max_len:
if max_len <= 3:
return "." * max_len
return text[:max_len - 3] + "..."
return text
def _delegate_task_goal_parts(tasks: Any, *, per_goal_len: int) -> tuple[int, list[str]]:
if not isinstance(tasks, list):
return 0, []
goals: list[str] = []
for task in tasks:
if not isinstance(task, dict):
continue
raw_goal = task.get("goal")
goal = "?" if raw_goal is None else _oneline(str(raw_goal))
goals.append(_truncate_preview(goal or "?", per_goal_len))
return len(goals), goals
def build_tool_preview(tool_name: str, args: dict, max_len: int | None = None) -> str | None:
"""Build a short preview of a tool call's primary argument for display.
@ -195,10 +217,17 @@ def build_tool_preview(tool_name: str, args: dict, max_len: int | None = None) -
if tool_name == "delegate_task":
tasks = args.get("tasks")
if tasks and isinstance(tasks, list):
goals = [_oneline(t.get("goal", "?"))[:40] for t in tasks if isinstance(t, dict)]
return f"{len(tasks)} tasks: " + " | ".join(goals) if goals else f"{len(tasks)} parallel tasks"
task_count, goals = _delegate_task_goal_parts(tasks, per_goal_len=40)
preview = (
f"{task_count} tasks: " + " | ".join(goals)
if goals else f"{len(tasks)} parallel tasks"
)
return _truncate_preview(preview, max_len)
goal = args.get("goal", "")
return _oneline(goal) if goal else None
if goal is None:
return None
preview = _oneline(str(goal))
return _truncate_preview(preview, max_len) if preview else None
if tool_name == "process":
action = args.get("action", "")
@ -1028,9 +1057,10 @@ def get_cute_tool_message(
if tool_name == "delegate_task":
tasks = args.get("tasks")
if tasks and isinstance(tasks, list):
goals = [_oneline(t.get("goal", "?"))[:30] for t in tasks if isinstance(t, dict)]
task_count, goals = _delegate_task_goal_parts(tasks, per_goal_len=30)
detail = " | ".join(goals) if goals else "parallel"
return _wrap(f"┊ 🔀 delegate {len(tasks)}x: {_trunc(detail, 35)} {dur}")
count_label = task_count or len(tasks)
return _wrap(f"┊ 🔀 delegate {count_label}x: {_trunc(detail, 35)} {dur}")
return _wrap(f"┊ 🔀 delegate {_trunc(args.get('goal', ''), 35)} {dur}")
preview = build_tool_preview(tool_name, args) or ""

11
cli.py
View file

@ -5783,14 +5783,19 @@ class HermesCLI(CLIAgentSetupMixin, CLICommandsMixin):
if not self._session_db:
return []
try:
sessions = self._session_db.list_sessions_rich(
from hermes_cli.session_listing import query_session_listing
return query_session_listing(
self._session_db,
source="cli",
exclude_sources=["tool"],
current_session_id=self.session_id,
include_all_sources=False,
include_unnamed=True,
limit=limit,
exclude_sources=["tool"],
)
except Exception:
return []
return [s for s in sessions if s.get("id") != self.session_id]
def _show_recent_sessions(self, *, reason: str = "history", limit: int = 10) -> bool:
"""Render recent sessions inline from the active chat TUI.

View file

@ -7554,6 +7554,9 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
if canonical == "resume":
return await self._handle_resume_command(event)
if canonical == "sessions":
return await self._handle_sessions_command(event)
if canonical == "branch":
return await self._handle_branch_command(event)

View file

@ -498,25 +498,11 @@ class GatewaySlashCommandsMixin:
model_cfg = user_config.get("model", {}) if isinstance(user_config, dict) else {}
if isinstance(model_cfg, dict):
provider_name = _clean_str(model_cfg.get("provider"))
if not context_total and model_name:
try:
from agent.model_metadata import get_model_context_length
model_cfg = user_config.get("model", {}) if isinstance(user_config, dict) else {}
configured_context = None
if isinstance(model_cfg, dict):
configured_context = model_cfg.get("context_length")
custom_providers = user_config.get("custom_providers") if isinstance(user_config, dict) else None
context_total = get_model_context_length(
model_name,
base_url=base_url,
api_key="",
config_context_length=configured_context if isinstance(configured_context, int) else None,
provider=provider_name,
custom_providers=custom_providers if isinstance(custom_providers, list) else None,
)
except Exception:
context_total = 0
if not context_total:
model_cfg = user_config.get("model", {}) if isinstance(user_config, dict) else {}
configured_context = model_cfg.get("context_length") if isinstance(model_cfg, dict) else None
if isinstance(configured_context, int) and configured_context > 0:
context_total = configured_context
model_line = ""
if model_name:
@ -553,7 +539,7 @@ class GatewaySlashCommandsMixin:
if context_line:
lines.append(context_line)
lines.extend([
t("gateway.status.tokens", tokens=f"{db_total_tokens:,} (cumulative)"),
t("gateway.status.tokens", tokens=f"{db_total_tokens:,}"),
t("gateway.status.agent_running", state=t("gateway.status.state_yes") if is_running else t("gateway.status.state_no")),
])
if queue_depth:
@ -2955,6 +2941,52 @@ class GatewaySlashCommandsMixin:
return t("gateway.resume.resumed_one", title=title, count=msg_count)
return t("gateway.resume.resumed_many", title=title, count=msg_count)
async def _handle_sessions_command(self, event: MessageEvent) -> str:
"""Handle /sessions — list previous sessions for gateway chats."""
if not self._session_db:
from hermes_state import format_session_db_unavailable
return format_session_db_unavailable(prefix=t("gateway.shared.session_db_unavailable_prefix"))
from hermes_cli.session_listing import (
format_gateway_session_listing,
parse_session_listing_args,
query_session_listing,
)
source = event.source
raw_args = event.get_command_args().strip()
try:
include_all, include_unnamed, target = parse_session_listing_args(raw_args)
except ValueError as exc:
return t("gateway.resume.parse_error", error=exc)
if target:
resume_event = dataclasses.replace(event, text=f"/resume {target}")
return await self._handle_resume_command(resume_event)
current_entry = self.session_store.get_or_create_session(source)
rows = query_session_listing(
self._session_db,
source=source.platform.value if source.platform else None,
current_session_id=current_entry.session_id,
include_all_sources=include_all,
include_unnamed=include_unnamed,
limit=10,
exclude_sources=["tool"],
)
if source.platform == Platform.MATRIX and not include_all:
rows = [
row for row in rows
if self._same_matrix_room(
source, self._gateway_session_origin_for_id(str(row.get("id") or ""))
)
]
return format_gateway_session_listing(
rows,
include_source=include_all,
title="Sessions" if include_unnamed else "Named Sessions",
)
async def _handle_branch_command(self, event: MessageEvent) -> str:
"""Handle /branch [name] — fork the current session into a new independent copy.

View file

@ -0,0 +1,97 @@
"""Shared session-listing helpers for CLI and gateway slash surfaces."""
from __future__ import annotations
from typing import Any
def parse_session_listing_args(raw_args: str) -> tuple[bool, bool, str]:
"""Parse `/sessions`-style args into listing flags plus a resume target.
Returns ``(include_all_sources, include_unnamed, target)``. ``list``/``ls``
and ``browse`` are display aliases; ``all``/``--all`` widens source scope;
``full``/``--full`` keeps unnamed sessions in the listing. Anything else is
treated as a target so `/sessions <id-or-title>` can delegate to `/resume`.
"""
import shlex
parts = shlex.split(raw_args or "")
include_all = False
include_unnamed = False
target_parts: list[str] = []
for part in parts:
lower = part.strip().lower()
if lower in {"list", "ls", "browse"}:
continue
if lower in {"all", "--all"}:
include_all = True
continue
if lower in {"full", "--full"}:
include_unnamed = True
continue
target_parts.append(part)
return include_all, include_unnamed, " ".join(target_parts).strip()
def query_session_listing(
session_db: Any,
*,
source: str | None,
current_session_id: str | None = None,
include_all_sources: bool = False,
include_unnamed: bool = False,
limit: int = 10,
exclude_sources: list[str] | None = None,
) -> list[dict[str, Any]]:
"""Return session rows for interactive listing surfaces.
This is the shared selection policy behind CLI/gateway session browsing:
source-scoped by default, optionally global, hide unnamed sessions unless
the caller asks for a full listing, and never include the current session.
"""
query_source = None if include_all_sources else source
fetch_limit = max(limit * 4, limit)
rows = session_db.list_sessions_rich(
source=query_source,
exclude_sources=exclude_sources,
limit=fetch_limit,
)
result: list[dict[str, Any]] = []
for row in rows:
if current_session_id and row.get("id") == current_session_id:
continue
if not include_unnamed and not row.get("title"):
continue
result.append(row)
if len(result) >= limit:
break
return result
def format_gateway_session_listing(
rows: list[dict[str, Any]],
*,
include_source: bool = False,
title: str = "Sessions",
) -> str:
"""Render a compact Markdown-ish session list for gateway messengers."""
if not rows:
return (
"No sessions found.\n"
"Use `/title My Session` to name this chat, or `/sessions full` "
"to include unnamed sessions."
)
lines = [f"📋 **{title}**", ""]
for idx, row in enumerate(rows, start=1):
session_id = str(row.get("id") or "")
title_text = str(row.get("title") or "")
preview = str(row.get("preview") or "")[:40]
source = str(row.get("source") or "")
source_part = f" `{source}`" if include_source and source else ""
preview_part = f" — _{preview}_" if preview else ""
lines.append(f"{idx}. **{title_text}**{source_part} — `{session_id}`{preview_part}")
lines.append("")
lines.append("Resume: `/resume <session id>` or `/resume <number>` from `/resume`.")
lines.append("More: `/sessions all`, `/sessions full`, `/sessions all full`.")
return "\n".join(lines)

View file

@ -1123,15 +1123,16 @@ class SessionDB:
# 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 < 11:
if current_version < 10 and SCHEMA_VERSION == 10:
# 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
# backfill into the FTS index.
#
# When upgrading straight to v11+, skip this backfill: v11
# drops and rebuilds both FTS tables anyway, so doing the v10
# trigram backfill first only burns startup time and WAL space.
# Only run this when v10 itself is the target schema. Current
# v11+ code drops and rebuilds both FTS tables below, so doing
# the v10-only trigram backfill first only burns startup time
# and WAL space before v11 throws the work away.
if fts5_available:
_fts_trigram_exists = self._fts_table_probe(
cursor, "messages_fts_trigram"

View file

@ -90,6 +90,8 @@ AUTHOR_MAP = {
"138671361+Veritas-7@users.noreply.github.com": "Veritas-7",
"keiron@onehanded.com": "kmccammon",
"268233388+CiarasClaws@users.noreply.github.com": "CiarasClaws",
"amy@ravenwolf.de": "WolframRavenwolf",
"github.com@wolfram.ravenwolf.de": "WolframRavenwolf",
"895252509@qq.com": "895252509",
"35259607+zxcasongs@users.noreply.github.com": "zxcasongs",
"alfred@my-cloud.me": "alfred-smith-0",

View file

@ -104,6 +104,33 @@ class TestBuildToolPreview:
assert result is not None
assert "find something" in result
def test_delegate_task_single_goal_preview(self):
result = build_tool_preview("delegate_task", {"goal": "Review gateway status"})
assert result == "Review gateway status"
def test_delegate_task_batch_goal_preview(self):
result = build_tool_preview(
"delegate_task",
{"tasks": [{"goal": "Review PR A"}, {"goal": "Review PR B"}]},
)
assert result == "2 tasks: Review PR A | Review PR B"
def test_delegate_task_batch_preview_handles_missing_non_string_goals(self):
result = build_tool_preview(
"delegate_task",
{"tasks": [{"goal": None}, {"goal": 123}, "not-a-task"]},
)
assert result == "2 tasks: ? | 123"
def test_delegate_task_batch_preview_respects_max_len(self):
result = build_tool_preview(
"delegate_task",
{"tasks": [{"goal": "A" * 80}, {"goal": "B" * 80}]},
max_len=30,
)
assert result == "2 tasks: AAAAAAAAAAAAAAAAAA..."
assert len(result) == 30
def test_false_like_args_zero(self):
"""Non-dict falsy values should return None, not crash."""
assert build_tool_preview("terminal", 0) is None
@ -170,6 +197,14 @@ class TestCuteToolMessagePreviewLength:
assert "[error]" not in line
def test_delegate_task_batch_message_includes_goals(self):
line = get_cute_tool_message(
"delegate_task",
{"tasks": [{"goal": "Review PR A"}, {"goal": "Review PR B"}]},
1.2,
)
assert "2x: Review PR A | Review PR B" in line
class TestEditDiffPreview:
def test_extract_edit_diff_for_patch(self):

View file

@ -4,7 +4,8 @@ Tests the _handle_resume_command handler (switch to a previously-named session)
across gateway messenger platforms.
"""
from unittest.mock import MagicMock
from types import SimpleNamespace
from unittest.mock import AsyncMock, MagicMock
import pytest
@ -36,9 +37,11 @@ def _make_runner(session_db=None, current_session_id="current_session_001",
from gateway.run import GatewayRunner
runner = object.__new__(GatewayRunner)
runner.adapters = {}
runner.config = SimpleNamespace(platforms={})
runner._voice_mode = {}
runner._session_db = session_db
runner._running_agents = {}
runner._is_user_authorized = lambda _source: True
# Compute the real session key if an event is provided
session_key = build_session_key(event.source) if event else "agent:main:telegram:dm"
@ -358,3 +361,64 @@ class TestHandleResumeCommand:
f"session-id lookup failed: {result!r}"
)
db.close()
class TestHandleSessionsCommand:
"""Tests for GatewayRunner._handle_sessions_command."""
@pytest.mark.asyncio
async def test_sessions_command_lists_current_platform_sessions(self, tmp_path):
from hermes_state import SessionDB
db = SessionDB(db_path=tmp_path / "state.db")
db.create_session("tg_session", "telegram")
db.set_session_title("tg_session", "Telegram Work")
db.create_session("discord_session", "discord")
db.set_session_title("discord_session", "Discord Work")
event = _make_event(text="/sessions")
runner = _make_runner(session_db=db, event=event)
result = await runner._handle_sessions_command(event)
assert "Sessions" in result
assert "Telegram Work" in result
assert "tg_session" in result
assert "Discord Work" not in result
db.close()
@pytest.mark.asyncio
async def test_sessions_all_full_lists_cross_platform_unnamed_sessions(self, tmp_path):
from hermes_state import SessionDB
db = SessionDB(db_path=tmp_path / "state.db")
db.create_session("tg_named", "telegram")
db.set_session_title("tg_named", "Telegram Work")
db.create_session("discord_unnamed", "discord")
db.append_message("discord_unnamed", "user", "discord first prompt")
event = _make_event(text="/sessions all full")
runner = _make_runner(session_db=db, event=event)
result = await runner._handle_sessions_command(event)
assert "Telegram Work" in result
assert "discord_unnamed" in result
assert "discord" in result
db.close()
@pytest.mark.asyncio
async def test_gateway_dispatches_sessions_command(self, tmp_path):
from hermes_state import SessionDB
db = SessionDB(db_path=tmp_path / "state.db")
db.create_session("tg_session", "telegram")
db.set_session_title("tg_session", "Telegram Work")
event = _make_event(text="/sessions")
runner = _make_runner(session_db=db, event=event)
runner._handle_sessions_command = AsyncMock(return_value="sessions output")
result = await runner._handle_message(event)
assert result == "sessions output"
runner._handle_sessions_command.assert_awaited_once_with(event)
db.close()

View file

@ -61,6 +61,8 @@ def _make_runner(session_entry: SessionEntry, *, platform: Platform = Platform.T
runner._reasoning_config = None
runner._provider_routing = {}
runner._fallback_model = None
runner._agent_cache = {}
runner._agent_cache_lock = MagicMock()
runner._show_reasoning = False
runner._is_user_authorized = lambda _source: True
runner._set_session_env = lambda _context: None
@ -209,7 +211,8 @@ async def test_status_command_includes_live_agent_model_and_context():
assert "**Model:** `openai/gpt-test` (openai)" in result
assert "**Context:** 12,345 / 100,000 (12%)" in result
assert "**Cumulative API tokens (re-sent each call):** 1,250 (cumulative)" in result
assert "**Cumulative API tokens (re-sent each call):** 1,250" in result
assert "1,250 (cumulative)" not in result
@pytest.mark.asyncio
@ -235,16 +238,41 @@ async def test_status_command_includes_persisted_model_and_context_when_agent_no
"billing_provider": "openai-codex",
"billing_base_url": "https://example.invalid/v1",
}
monkeypatch.setattr(
"agent.model_metadata.get_model_context_length",
lambda *_args, **_kwargs: 272_000,
)
monkeypatch.setattr("gateway.run._load_gateway_config", lambda: {"model": {"context_length": 272_000}})
result = await runner._handle_message(_make_event("/status"))
assert "**Model:** `openai/gpt-persisted` (openai-codex)" in result
assert "**Context:** 24,000 / 272,000 (9%)" in result
assert "**Cumulative API tokens (re-sent each call):** 2,500 (cumulative)" in result
assert "**Cumulative API tokens (re-sent each call):** 2,500" in result
@pytest.mark.asyncio
async def test_status_command_includes_cached_agent_model_and_context():
session_entry = SessionEntry(
session_key=build_session_key(_make_source()),
session_id="sess-1",
created_at=datetime.now(),
updated_at=datetime.now(),
platform=Platform.TELEGRAM,
chat_type="dm",
total_tokens=0,
)
runner = _make_runner(session_entry)
cached_agent = SimpleNamespace(
model="anthropic/claude-sonnet-test",
provider="openrouter",
context_compressor=SimpleNamespace(
last_prompt_tokens=10_000,
context_length=200_000,
),
)
runner._agent_cache = {session_entry.session_key: (cached_agent, time.time())}
result = await runner._handle_message(_make_event("/status"))
assert "**Model:** `anthropic/claude-sonnet-test` (openrouter)" in result
assert "**Context:** 10,000 / 200,000 (5%)" in result
@pytest.mark.asyncio

View file

@ -4,7 +4,7 @@ import sqlite3
import time
import pytest
from hermes_state import SCHEMA_SQL, SessionDB
from hermes_state import SCHEMA_SQL, SCHEMA_VERSION, SessionDB
class _NoFtsCursor(sqlite3.Cursor):
@ -2297,6 +2297,65 @@ class TestSchemaInit:
migrated_db.close()
def test_v9_migration_skips_v10_trigram_backfill_before_v11_rebuild(self, tmp_path, monkeypatch):
"""Direct v9→current migration should do only the v11 FTS rebuild.
v10 backfilled ``messages_fts_trigram`` with content-only rows. Current
v11+ migration immediately drops and rebuilds both FTS tables with
content + tool metadata, so running the v10 insert first is wasted work.
"""
db_path = tmp_path / "v9_fts.db"
conn = sqlite3.connect(str(db_path))
conn.executescript(SCHEMA_SQL)
conn.execute("DELETE FROM schema_version")
conn.execute("INSERT INTO schema_version (version) VALUES (9)")
conn.execute(
"INSERT INTO sessions (id, source, started_at) VALUES (?, ?, ?)",
("s1", "cli", 1000.0),
)
conn.execute(
"INSERT INTO messages (session_id, role, content, tool_name, tool_calls, timestamp) "
"VALUES (?, ?, ?, ?, ?, ?)",
("s1", "tool", "plain content", "browser_snapshot", '{"name":"browser_snapshot"}', 1001.0),
)
conn.commit()
conn.close()
trigram_content_only_inserts = []
real_connect = sqlite3.connect
def connect_with_trace(*args, **kwargs):
conn = real_connect(*args, **kwargs)
def trace(sql):
text = " ".join(str(sql).split())
if (
"INSERT INTO messages_fts_trigram" in text
and "SELECT id, content FROM messages" in text
):
trigram_content_only_inserts.append(text)
conn.set_trace_callback(trace)
return conn
monkeypatch.setattr("hermes_state.sqlite3.connect", connect_with_trace)
migrated_db = SessionDB(db_path=db_path)
try:
assert trigram_content_only_inserts == []
version = migrated_db._conn.execute("SELECT version FROM schema_version").fetchone()[0]
assert version == SCHEMA_VERSION
normal_count = migrated_db._conn.execute("SELECT COUNT(*) FROM messages_fts").fetchone()[0]
trigram_count = migrated_db._conn.execute("SELECT COUNT(*) FROM messages_fts_trigram").fetchone()[0]
assert normal_count == 1
assert trigram_count == 1
tool_hit = migrated_db._conn.execute(
"SELECT COUNT(*) FROM messages_fts_trigram "
"WHERE messages_fts_trigram MATCH 'browser_snapshot'"
).fetchone()[0]
assert tool_hit == 1
finally:
migrated_db.close()
def test_reconciliation_adds_missing_columns(self, tmp_path):
"""Columns present in SCHEMA_SQL but missing from the live table
are added by _reconcile_columns regardless of schema_version.

View file

@ -117,11 +117,12 @@ class TestCommandTimeoutCache:
class TestSessionInactivityTimeout:
def test_default_is_300(self, monkeypatch):
def test_default_matches_config_default(self, monkeypatch):
from hermes_cli.config import DEFAULT_CONFIG
from tools.browser_tool import _get_session_inactivity_timeout
monkeypatch.delenv("BROWSER_INACTIVITY_TIMEOUT", raising=False)
with patch("hermes_cli.config.read_raw_config", return_value={}):
assert _get_session_inactivity_timeout() == 300
assert _get_session_inactivity_timeout() == DEFAULT_CONFIG["browser"]["inactivity_timeout"]
def test_reads_from_config_over_env(self, monkeypatch):
from tools.browser_tool import _get_session_inactivity_timeout
@ -137,6 +138,13 @@ class TestSessionInactivityTimeout:
with patch("hermes_cli.config.read_raw_config", return_value=cfg):
assert _get_session_inactivity_timeout() == 30
def test_invalid_config_preserves_env_fallback(self, monkeypatch):
from tools.browser_tool import _get_session_inactivity_timeout
monkeypatch.setenv("BROWSER_INACTIVITY_TIMEOUT", "240")
cfg = {"browser": {"inactivity_timeout": "not-an-int"}}
with patch("hermes_cli.config.read_raw_config", return_value=cfg):
assert _get_session_inactivity_timeout() == 240
# ---------------------------------------------------------------------------
# Caching: _discover_homebrew_node_dirs

View file

@ -67,7 +67,7 @@ from pathlib import Path
from agent.auxiliary_client import call_llm
from hermes_constants import get_hermes_home
from utils import env_int, is_truthy_value
from hermes_cli.config import cfg_get
from hermes_cli.config import DEFAULT_CONFIG, cfg_get
try:
from tools.website_policy import check_website_access
@ -1180,8 +1180,13 @@ _cleanup_done = False
# Session inactivity timeout (seconds) - cleanup if no activity for this long.
# config.yaml is authoritative; BROWSER_INACTIVITY_TIMEOUT remains a legacy
# fallback so old deployments keep working if they have not migrated yet.
DEFAULT_SESSION_INACTIVITY_TIMEOUT = int(
DEFAULT_CONFIG.get("browser", {}).get("inactivity_timeout", 120)
)
def _get_session_inactivity_timeout() -> int:
result = env_int("BROWSER_INACTIVITY_TIMEOUT", 300)
result = env_int("BROWSER_INACTIVITY_TIMEOUT", DEFAULT_SESSION_INACTIVITY_TIMEOUT)
try:
from hermes_cli.config import read_raw_config
cfg = read_raw_config()