ouyangbo/hermes-agent
1
0
Fork 0
Code Issues Pull Requests Actions 12 Packages Projects Releases Wiki Activity

refactor(gateway/run, slash_commands): dispatch tables, helper unification and hand-reviewed comment compaction

Browse Source 
run.py:
- built-in adapter creation: 9-branch if/elif -> _BUILTIN_ADAPTERS table
- idle slash-command routing: 35 `if canonical == ...` branches -> _gateway_idle_command_handlers()
- shared helpers: _send_command_ack (4 sites), _command_origin_for_source (2), _session_entry_for_manager
  (goal/heartbeat), _toggle_adapter_auto_tts_set (2), _load_env_or_agent_cfg_timeout (2), _float_env reuse (2),
  _resolve_session_key_or_none (3), _running_agent_ids (4), _schedule_rename_from_title_thread (2),
  _write_runtime_status_quiet (5), _AUTO_RESET_CONTEXT_NOTES/_auto_reset_reason_text
- ruff SIM102/SIM103/SIM105/SIM108/SIM118 + F401 across gateway/ (semantics re-reviewed; sqlite Row
  `.keys()` and side-effecting assignments kept)
- two hand-reviewed comment/docstring compaction passes (AST-identical, rationale kept)

slash_commands.py:
- /model: typed path and picker callback shared one 200-line commit block -> _perform_model_switch +
  _commit_model_switch
- comment/docstring compaction (AST-identical)
This commit is contained in:
Teknium
2026-09-02 10:43:26 -07:00
parent ffd628c3ee
commit aed6720dab
21 changed files with 3705 additions and 6377 deletions
Download Patch File Download Diff File Expand all files Collapse all files
gateway/browser_control_artifacts.py
+5 -12
View File
@@ -60,6 +60,7 @@ import time
from dataclasses import dataclass
from pathlib import Path
from typing import Any, Callable, Optional
import contextlib
logger = logging.getLogger(__name__)
@@ -338,10 +339,8 @@ class ArtifactStore:
except Exception:
with self._lock:
self._entries.pop(artifact_id, None)
try:
with contextlib.suppress(Exception):
temp.unlink(missing_ok=True)
except Exception:
pass
raise
return receipt
@@ -397,10 +396,8 @@ class ArtifactStore:
for artifact_id, entry in list(self._entries.items()):
if entry.receipt.expires_at <= now:
self._entries.pop(artifact_id, None)
try:
with contextlib.suppress(OSError):
entry.path.unlink(missing_ok=True)
except OSError:
pass
removed += 1
for temp in self._root.glob(f"*{_TEMP_SUFFIX}"):
try:
@@ -435,10 +432,8 @@ class ArtifactStore:
raise ArtifactNotFound(f"unknown artifact {artifact_id!r}")
if entry.receipt.expires_at <= now:
self._entries.pop(artifact_id, None)
try:
with contextlib.suppress(OSError):
path.unlink(missing_ok=True)
except OSError:
pass
raise ArtifactExpired(f"artifact {artifact_id!r} expired")
if entry.receipt.scope_key != scope_key:
raise ArtifactScopeMismatch(
@@ -450,10 +445,8 @@ class ArtifactStore:
for artifact_id, entry in list(self._entries.items()):
if entry.receipt.expires_at <= now:
self._entries.pop(artifact_id, None)
try:
with contextlib.suppress(OSError):
entry.path.unlink(missing_ok=True)
except OSError:
pass
def _artifact_path(self, artifact_id: str) -> Path:
"""Resolve a minted id strictly inside the controlled root."""
gateway/config.py
+3 -6
View File
@@ -25,6 +25,7 @@ from gateway.shutdown_watchdog import (
DEFAULT_LOOP_WATCHDOG_TIMEOUT_S,
)
from utils import is_truthy_value
import contextlib
logger = logging.getLogger(__name__)
@@ -2062,10 +2063,8 @@ def _env_int_extra(extra: Dict[str, Any], key: str, env: str) -> None:
"""Set an int extra from env; a non-integer value is silently ignored."""
raw = _getenv_str(env)
if raw:
try:
with contextlib.suppress(ValueError):
extra[key] = int(raw)
except ValueError:
pass
def _env_home_channel(
@@ -2627,10 +2626,8 @@ def _apply_env_overrides(config: GatewayConfig) -> None:
for env, attr in (("SESSION_IDLE_MINUTES", "idle_minutes"), ("SESSION_RESET_HOUR", "at_hour")):
raw = getenv(env)
if raw:
try:
with contextlib.suppress(ValueError):
setattr(config.default_reset_policy, attr, int(raw))
except ValueError:
pass
_enable_plugin_platforms_from_env(config)
gateway/drain_control.py
+2 -3
View File
@@ -42,6 +42,7 @@ from typing import Any, Optional
from hermes_constants import get_hermes_home
from utils import atomic_json_write
import contextlib
_log = logging.getLogger(__name__)
@@ -71,10 +72,8 @@ def current_instantiation_epoch() -> str:
which disables the epoch check downstream — never fail-closed.
"""
boot_id = ""
try:
with contextlib.suppress(OSError):
boot_id = Path("/proc/sys/kernel/random/boot_id").read_text(encoding="utf-8").strip()
except OSError:
pass
pid1_start = ""
try:
gateway/hosted_room_links.py
+2 -3
View File
@@ -21,6 +21,7 @@ from gateway.hosted_room_peer import (
TransportSecurity,
validate_room_link_url,
)
import contextlib
MAX_LINKS = 512
@@ -164,10 +165,8 @@ def save_room_link(db_path: Path | str, link: StoredRoomLink) -> None:
db_path, record=link.as_record(), max_links=MAX_LINKS
)
if os.name == "posix":
try:
with contextlib.suppress(OSError):
Path(db_path).chmod(0o600)
except OSError:
pass
def mark_room_link_status(
gateway/kanban_watchers.py
+6 -11
View File
@@ -21,6 +21,7 @@ from pathlib import Path
from typing import Any, Callable, Optional
from agent.i18n import t
import contextlib
# Match the logger run.py uses (logging.getLogger(__name__) where __name__ ==
# "gateway.run") so extracted log records keep their original logger name.
@@ -164,10 +165,8 @@ def _release_singleton_lock(handle) -> None:
_release_file_lock(handle)
except Exception:
pass
try:
with contextlib.suppress(Exception):
handle.close()
except Exception:
pass
def _wake_scope_id(adapter: Any, sub: dict) -> Optional[str]:
@@ -341,7 +340,7 @@ class GatewayKanbanWatchersMixin:
)
active_platforms = {
getattr(platform, "value", str(platform)).lower()
for platform in self.adapters.keys()
for platform in self.adapters
}
# Widen to every platform any secondary profile has live,
# not just the default profile's. This is only a coarse
@@ -359,7 +358,7 @@ class GatewayKanbanWatchersMixin:
for _profile_adapter_map in getattr(self, "_profile_adapters", {}).values():
active_platforms.update(
getattr(platform, "value", str(platform)).lower()
for platform in _profile_adapter_map.keys()
for platform in _profile_adapter_map
)
if not active_platforms:
logger.debug("kanban notifier: no connected adapters; skipping tick")
@@ -1606,10 +1605,8 @@ class GatewayKanbanWatchersMixin:
return None
finally:
if conn is not None:
try:
with contextlib.suppress(Exception):
conn.close()
except Exception:
pass
def _tick_once() -> "list[tuple[str, Optional[object]]]":
"""Run one dispatch_once per board. Returns (slug, result) pairs.
@@ -1664,10 +1661,8 @@ class GatewayKanbanWatchersMixin:
continue
finally:
if conn is not None:
try:
with contextlib.suppress(Exception):
conn.close()
except Exception:
pass
return False
# Auto-decompose: turn fresh triage tasks into ready workgraphs
gateway/memory_monitor.py
+3 -6
View File
@@ -37,6 +37,7 @@ import sys
import threading
import time
from typing import Optional
import contextlib
logger = logging.getLogger(__name__)
@@ -205,10 +206,8 @@ def stop_memory_monitoring(timeout: float = 2.0) -> None:
return
# Final snapshot before teardown so "last RSS" is always in the log.
try:
with contextlib.suppress(Exception):
log_memory_usage(prefix="shutdown")
except Exception:
pass
_stop_event.set()
thread = _monitor_thread
@@ -216,10 +215,8 @@ def stop_memory_monitoring(timeout: float = 2.0) -> None:
_stop_event = None
# Join outside the lock so a stuck log call can't deadlock shutdown.
try:
with contextlib.suppress(Exception):
thread.join(timeout=timeout)
except Exception:
pass
logger.info("[MEMORY] Periodic memory monitoring stopped")
gateway/pairing.py
+4 -9
View File
@@ -39,6 +39,7 @@ from hermes_constants import (
get_hermes_home,
)
from utils import atomic_replace
import contextlib
logger = logging.getLogger(__name__)
@@ -304,20 +305,16 @@ def _sync_live_adapter_allowlist_remove(platform: str, user_id: str) -> None:
if _adapter_platform_name(adapter) != platform_name:
continue
if hasattr(adapter, "_allow_from"):
try:
with contextlib.suppress(Exception):
adapter._allow_from = _purge_allowlist_entries(
set(adapter._allow_from or ()), platform_name, user_id
)
except Exception:
pass
extra = getattr(getattr(adapter, "config", None), "extra", None)
if isinstance(extra, dict) and "allow_from" in extra:
try:
with contextlib.suppress(Exception):
extra["allow_from"] = _purge_allowlist_entries(
extra.get("allow_from"), platform_name, user_id
)
except Exception:
pass
def _sync_allowlist_remove(platform: str, user_id: str) -> None:
@@ -426,10 +423,8 @@ def _secure_write(path: Path, data: str) -> None:
except OSError:
pass # Windows doesn't support chmod the same way
except BaseException:
try:
with contextlib.suppress(OSError):
os.unlink(tmp_path)
except OSError:
pass
raise
gateway/relay/adapter.py
+1 -3
View File
@@ -338,9 +338,7 @@ class RelayAdapter(BasePlatformAdapter):
platform = self._platform_by_chat.get(str(chat_id))
if platform is None:
platform = getattr(desc, "platform", None)
if self._slack_unfurl_hints(platform):
return False
return True
return not self._slack_unfurl_hints(platform)
def prefers_fresh_final_streaming(
self,
gateway/relay/media.py
-1
View File
@@ -36,7 +36,6 @@ import mimetypes
import os
import tempfile
import urllib.error
import urllib.parse
import urllib.request
from pathlib import Path
from typing import Optional
gateway/relay/ws_transport.py
+2 -3
View File
@@ -29,6 +29,7 @@ from gateway.platforms.base import MessageEvent, MessageType
from gateway.session import SessionSource
from gateway.relay.descriptor import CapabilityDescriptor
from gateway.relay.transport import InboundHandler
import contextlib
logger = logging.getLogger(__name__)
@@ -77,10 +78,8 @@ def _env_disconnect_budget_s() -> float:
budget = 5.0
raw = os.getenv("HERMES_GATEWAY_ADAPTER_DISCONNECT_TIMEOUT", "").strip()
if raw:
try:
with contextlib.suppress(ValueError):
budget = max(0.0, float(raw))
except ValueError:
pass
return budget
gateway/restart_loop_guard.py
+2 -3
View File
@@ -29,6 +29,7 @@ import time
from typing import List, Optional
from hermes_constants import get_hermes_home
import contextlib
logger = logging.getLogger("gateway.run")
@@ -119,10 +120,8 @@ def record_restart_interrupted_boot(
def clear() -> None:
"""Remove the persisted boot log (used on clean shutdown / by tests)."""
try:
with contextlib.suppress(OSError):
_state_path().unlink(missing_ok=True)
except OSError:
pass
def check_and_record(
gateway/run.py
+3074 -5278
View File
File diff suppressed because it is too large Load Diff
gateway/session.py
+3 -4
View File
@@ -104,6 +104,7 @@ from .whatsapp_identity import (
)
from utils import atomic_replace
from agent.turn_context import extract_api_content_sidecar
import contextlib
# Session keys/ids flow into filesystem paths downstream (e.g.
# ``sessions_dir / f"{session_id}.json"`` in hermes_state, request-dump
@@ -1378,7 +1379,7 @@ class SessionStore:
once it expires, one caller reopens while concurrent callers keep
using the JSONL fallback.
"""
from hermes_state import SessionDB, _default_db_path, get_shared_session_db
from hermes_state import _default_db_path, get_shared_session_db
path = Path(db_path) if db_path is not None else Path(_default_db_path())
def _open():
@@ -2547,10 +2548,8 @@ class SessionStore:
return
try:
origin_json = None
try:
with contextlib.suppress(Exception):
origin_json = json.dumps(source.to_dict())
except Exception:
pass
recorder(
session_id,
source=source.platform.value,
gateway/session_db_recovery.py
+3 -6
View File
@@ -8,6 +8,7 @@ import weakref
from dataclasses import dataclass
from pathlib import Path
from typing import Any, Callable
import contextlib
_INITIAL_RETRY_DELAY_SECONDS = 1.0
@@ -137,10 +138,8 @@ class RecoverableHandleCache:
close_rejected = self._close_rejected if stale else None
if stale:
if close_rejected is not None:
try:
with contextlib.suppress(Exception):
close_rejected(handle)
except Exception:
pass
return None
_publish_health(self._health_source, path, "ok")
if was_unavailable and on_recovered is not None:
@@ -157,10 +156,8 @@ class RecoverableHandleCache:
self.handles.clear()
self._unavailable.clear()
for handle in handles:
try:
with contextlib.suppress(Exception):
close(handle)
except Exception:
pass
with _health_lock:
states = _health_states.get(self._health_source)
if states is not None:
gateway/shutdown_forensics.py
+5 -12
View File
@@ -30,6 +30,7 @@ from gateway.restart import (
DEFAULT_GATEWAY_CRON_DRAIN_TIMEOUT,
resolve_systemd_timeout_stop_sec,
)
import contextlib
_SIGNAL_NAME_BY_NUM: Dict[int, str] = {}
@@ -91,10 +92,8 @@ def _proc_summary(pid: int) -> Dict[str, Any]:
summary["state"] = state
ppid = _read_proc_field(pid, "PPid")
if ppid is not None:
try:
with contextlib.suppress(ValueError):
summary["ppid"] = int(ppid)
except ValueError:
pass
uid = _read_proc_field(pid, "Uid")
if uid is not None:
# "real effective saved fs"
@@ -149,10 +148,8 @@ def snapshot_shutdown_context(received_signal: Any = None) -> Dict[str, Any]:
# Load average — high load points the finger at "something else
# crushing the box" rather than "external killer".
try:
with contextlib.suppress(OSError, AttributeError):
ctx["loadavg_1m"] = os.getloadavg()[0]
except (OSError, AttributeError):
pass
# /proc/self/status TracerPid: nonzero means a debugger / strace is
# attached. Useful when "phantom SIGKILL" turns out to be a manual
@@ -268,17 +265,13 @@ def spawn_async_diagnostic(
close_fds=True,
)
except (FileNotFoundError, OSError):
try:
with contextlib.suppress(OSError):
os.close(fd)
except OSError:
pass
return None
finally:
# Subprocess inherited the fd; we can drop our handle.
try:
with contextlib.suppress(OSError):
os.close(fd)
except OSError:
pass
return proc.pid
gateway/shutdown_watchdog.py
+7 -18
View File
@@ -38,6 +38,7 @@ from typing import Any, Callable, Dict, Optional
from gateway.restart import GATEWAY_SERVICE_RESTART_EXIT_CODE
from hermes_constants import get_hermes_home
from utils import atomic_json_write
import contextlib
logger = logging.getLogger(__name__)
@@ -175,7 +176,7 @@ def start_loop_liveness_watchdog(
if stop_event.is_set():
return
try:
with contextlib.suppress(Exception):
logger.critical(
"Gateway event loop missed %d consecutive liveness probes; "
"dumping all thread stacks and exiting with code %d so the "
@@ -183,8 +184,6 @@ def start_loop_liveness_watchdog(
strikes,
exit_code,
)
except Exception:
pass
try:
faulthandler.dump_traceback(all_threads=True)
except Exception:
@@ -406,21 +405,17 @@ def arm_shutdown_watchdog(
target = dump_path if dump_path is not None else get_shutdown_watchdog_dump_path()
_write_watchdog_dump(target, delay_s=delay, snapshot=snapshot)
try:
with contextlib.suppress(Exception):
logger.critical(
"Shutdown watchdog fired after %.0fs — forcing process exit "
"(asyncio drain path appears wedged; see %s)",
delay,
target,
)
except Exception:
pass
for stream in (sys.stdout, sys.stderr):
try:
with contextlib.suppress(Exception):
stream.flush()
except Exception:
pass
# Mirror _exit_after_graceful_shutdown: release PID file + runtime
# lock BEFORE the log drain (locks must never be stranded), then
# drain the async log queue so the logger.critical above actually
@@ -470,10 +465,8 @@ async def _tick_socket_handler(
except Exception:
pass
finally:
try:
with contextlib.suppress(Exception):
writer.close()
except Exception:
pass
async def loop_heartbeat_forever(
@@ -638,12 +631,8 @@ async def loop_heartbeat_forever(
finally:
if tick_server is not None:
tick_server.close()
try:
with contextlib.suppress(Exception):
await tick_server.wait_closed()
except Exception:
pass
if tick_socket_path is not None:
try:
with contextlib.suppress(Exception):
tick_socket_path.unlink(missing_ok=True)
except Exception:
pass
gateway/slash_commands.py
+548 -898
View File
File diff suppressed because it is too large Load Diff
gateway/status.py
+23 -72
View File
@@ -29,6 +29,7 @@ from pathlib import Path
from hermes_constants import get_hermes_home, _get_platform_default_hermes_home
from typing import Any, Callable, NamedTuple, Optional
from utils import atomic_json_write
import contextlib
if sys.platform == "win32":
import msvcrt
@@ -659,9 +660,7 @@ def _command_line_belongs_to_profile(command: str, profile_home: Path) -> bool:
# absence is not disqualifying — only a conflicting explicit value is.
if "--profile " in command_lc or " -p " in command_lc:
return False
if "hermes_home=" in command_lc and f"hermes_home={home_lc}" not in command_lc:
return False
return True
return not ("hermes_home=" in command_lc and f"hermes_home={home_lc}" not in command_lc)
def _record_matches_live_gateway_pid(
@@ -688,11 +687,7 @@ def _record_matches_live_gateway_pid(
if live_cmdline:
if not looks_like_gateway_runtime_command_line(live_cmdline):
return False
if expected_home is not None and not _command_line_belongs_to_profile(
live_cmdline, expected_home
):
return False
return True
return not (expected_home is not None and not _command_line_belongs_to_profile(live_cmdline, expected_home))
return _record_looks_like_gateway(record)
@@ -877,14 +872,10 @@ def _cleanup_invalid_pid_path(pid_path: Path, *, cleanup_stale: bool) -> None:
if not cleanup_stale:
return
_clear_running_pid_cache()
try:
with contextlib.suppress(Exception):
pid_path.unlink(missing_ok=True)
except Exception:
pass
try:
with contextlib.suppress(Exception):
_get_gateway_lock_path(pid_path).unlink(missing_ok=True)
except Exception:
pass
def _write_gateway_lock_record(handle) -> None:
@@ -892,10 +883,8 @@ def _write_gateway_lock_record(handle) -> None:
handle.truncate()
json.dump(_build_pid_record(), handle)
handle.flush()
try:
with contextlib.suppress(OSError):
os.fsync(handle.fileno())
except OSError:
pass
def _try_acquire_file_lock(handle) -> bool:
@@ -1089,10 +1078,8 @@ def release_gateway_runtime_lock() -> None:
return
_gateway_lock_handle = None
_release_file_lock(handle)
try:
with contextlib.suppress(OSError):
handle.close()
except OSError:
pass
_clear_running_pid_cache()
@@ -1127,10 +1114,8 @@ def is_gateway_runtime_lock_active(lock_path: Optional[Path] = None) -> bool:
# session that ran as root. The parent directory owner can unlink
# files even when they don't own them, so remove the stale lock
# and report inactive — the new process will create a fresh one.
try:
with contextlib.suppress(OSError):
resolved_lock_path.unlink()
except OSError:
pass
return False
try:
if _try_acquire_file_lock(handle):
@@ -1138,10 +1123,8 @@ def is_gateway_runtime_lock_active(lock_path: Optional[Path] = None) -> bool:
return False
return True
finally:
try:
with contextlib.suppress(OSError):
handle.close()
except OSError:
pass
def _strict_path_exists(path: Path, label: str) -> bool:
@@ -1170,10 +1153,8 @@ def _is_gateway_runtime_lock_active_strict(lock_path: Path) -> bool:
except OSError as exc:
raise RuntimeError(f"gateway runtime lock probe failed: {exc}") from exc
finally:
try:
with contextlib.suppress(OSError):
handle.close()
except OSError:
pass
def write_pid_file() -> None:
@@ -1195,10 +1176,8 @@ def write_pid_file() -> None:
f.write(record)
_clear_running_pid_cache()
except Exception:
try:
with contextlib.suppress(OSError):
path.unlink(missing_ok=True)
except OSError:
pass
raise
@@ -1358,13 +1337,7 @@ def runtime_status_pid_is_live(record: Optional[dict[str, Any]]) -> bool:
return False
recorded_start = (record or {}).get("start_time")
current_start = _get_process_start_time(pid)
if (
recorded_start is not None
and current_start is not None
and current_start != recorded_start
):
return False
return True
return not (recorded_start is not None and current_start is not None and current_start != recorded_start)
def parse_active_agents(raw: Any) -> int:
@@ -1673,10 +1646,8 @@ def acquire_scoped_lock(scope: str, identity: str, metadata: Optional[dict[str,
# stale. This happens when a previous process was killed between
# O_CREAT|O_EXCL and the subsequent json.dump() (e.g. DNS failure
# during rapid Slack reconnect retries).
try:
with contextlib.suppress(OSError):
lock_path.unlink(missing_ok=True)
except OSError:
pass
if existing:
try:
existing_pid = int(existing["pid"])
@@ -1772,10 +1743,8 @@ def acquire_scoped_lock(scope: str, identity: str, metadata: Optional[dict[str,
except OSError:
pass
else:
try:
with contextlib.suppress(OSError):
tombstone.unlink(missing_ok=True)
except OSError:
pass
else:
return False, existing
@@ -1787,10 +1756,8 @@ def acquire_scoped_lock(scope: str, identity: str, metadata: Optional[dict[str,
with os.fdopen(fd, "w", encoding="utf-8") as handle:
json.dump(record, handle)
except Exception:
try:
with contextlib.suppress(OSError):
lock_path.unlink(missing_ok=True)
except OSError:
pass
raise
return True, None
@@ -1807,10 +1774,8 @@ def release_scoped_lock(scope: str, identity: str) -> None:
# start_time equality: on-disk null vs a live fingerprint (macOS/psutil
# timing) would otherwise leave the lock stuck across Discord/Telegram
# reconnects (#81468). start_time only guards PID reuse for *other* PIDs.
try:
with contextlib.suppress(OSError):
lock_path.unlink(missing_ok=True)
except OSError:
pass
def release_all_scoped_locks(
@@ -1924,17 +1889,13 @@ def _consume_pid_marker_for_self(
target_start_time = record.get(start_time_field)
written_at = record.get("written_at") or ""
except (KeyError, TypeError, ValueError):
try:
with contextlib.suppress(OSError):
path.unlink(missing_ok=True)
except OSError:
pass
return False
if _marker_is_stale(written_at, ttl_s):
try:
with contextlib.suppress(OSError):
path.unlink(missing_ok=True)
except OSError:
pass
return False
# Cross-profile guard (#29092): new markers explicitly name the verified
@@ -1977,10 +1938,8 @@ def _consume_pid_marker_for_self(
else:
matches = True
try:
with contextlib.suppress(OSError):
path.unlink(missing_ok=True)
except OSError:
pass
return matches
@@ -2050,10 +2009,8 @@ def consume_takeover_marker_for_self() -> bool:
def clear_takeover_marker(target_home: Optional[Path] = None) -> None:
"""Remove the takeover marker unconditionally. Safe to call repeatedly."""
try:
with contextlib.suppress(OSError):
_get_takeover_marker_path(target_home).unlink(missing_ok=True)
except OSError:
pass
def _validated_scoped_lock_gateway_owner(
@@ -2420,19 +2377,15 @@ def planned_stop_marker_targets_self() -> bool:
written_at = record.get("written_at") or ""
except (KeyError, TypeError, ValueError):
# Malformed marker can never match anyone — drop it.
try:
with contextlib.suppress(OSError):
path.unlink(missing_ok=True)
except OSError:
pass
return False
if _marker_is_stale(written_at, _PLANNED_STOP_MARKER_TTL_S):
# A marker this old is past its useful life regardless of target —
# clean it up so it cannot crash-loop a freshly booted gateway.
try:
with contextlib.suppress(OSError):
path.unlink(missing_ok=True)
except OSError:
pass
return False
our_pid = os.getpid()
@@ -2455,10 +2408,8 @@ def planned_stop_marker_targets_self() -> bool:
def clear_planned_stop_marker() -> None:
"""Remove the planned-stop marker unconditionally."""
try:
with contextlib.suppress(OSError):
_get_planned_stop_marker_path().unlink(missing_ok=True)
except OSError:
pass
def get_running_pid(
gateway/sticker_cache.py
+2 -3
View File
@@ -11,6 +11,7 @@ import time
from typing import Optional
from hermes_cli.config import get_hermes_home
import contextlib
CACHE_PATH = get_hermes_home() / "sticker_cache.json"
@@ -42,10 +43,8 @@ def _save_cache(cache: dict) -> None:
os.fsync(f.fileno())
os.replace(tmp_path, str(CACHE_PATH))
except BaseException:
try:
with contextlib.suppress(OSError):
os.unlink(tmp_path)
except OSError:
pass
raise
gateway/stream_consumer.py
+7 -20
View File
@@ -38,6 +38,7 @@ from gateway.response_filters import (
is_intentional_silence_response as _is_intentional_silence_response,
is_partial_silence_marker as _is_partial_silence_marker,
)
import contextlib
logger = logging.getLogger("gateway.stream_consumer")
@@ -685,9 +686,7 @@ class GatewayStreamConsumer:
return True
# A segment break / commentary may have delivered the final text
# earlier in the turn under a different record.
if self.has_delivered_text(final_text):
return True
return False
return bool(self.has_delivered_text(final_text))
def has_delivered_text(self, text: str) -> bool:
"""Return True if *text* was already delivered as visible chat content."""
@@ -752,10 +751,8 @@ class GatewayStreamConsumer:
race conditions with pending deltas or other queue items.
"""
loop = None
try:
with contextlib.suppress(RuntimeError):
loop = asyncio.get_running_loop()
except RuntimeError:
pass
if not self._use_native_streaming:
# No native stream to close — return resolved future
@@ -774,10 +771,7 @@ class GatewayStreamConsumer:
# Create a future that run() will resolve after processing.
# cancelled_flag is retained for backward compatibility with callers
# (run.py sets it on timeout) but the handler always finalizes regardless.
if loop:
boundary_future = loop.create_future()
else:
boundary_future = concurrent.futures.Future()
boundary_future = loop.create_future() if loop else concurrent.futures.Future()
cancelled_flag = {"cancelled": False}
self._queue.put((_APPROVAL_BOUNDARY, boundary_future, cancelled_flag))
@@ -854,10 +848,8 @@ class GatewayStreamConsumer:
"""
if flush_event is None:
return
try:
with contextlib.suppress(Exception):
flush_event.set()
except Exception:
pass
def _reset_segment_state(self, *, preserve_no_edit: bool = False) -> None:
if preserve_no_edit and self._message_id == "__no_edit__":
@@ -1040,10 +1032,7 @@ class GatewayStreamConsumer:
# Resolve future so approval callback knows the result
if boundary_future is not None:
try:
if isinstance(boundary_future, asyncio.Future):
if not boundary_future.done():
boundary_future.set_result(boundary_ok)
elif isinstance(boundary_future, concurrent.futures.Future):
if isinstance(boundary_future, (asyncio.Future, concurrent.futures.Future)):
if not boundary_future.done():
boundary_future.set_result(boundary_ok)
except Exception:
@@ -1995,14 +1984,12 @@ class GatewayStreamConsumer:
# _final_response_sent itself; this handler owns the flags.
_best_effort_ok = False
if self._accumulated and self._message_id:
try:
with contextlib.suppress(Exception):
_best_effort_ok = bool(
await self._send_or_edit(
self._accumulated, finalize=True, is_turn_final=False,
)
)
except Exception:
pass
elif self._message_id is None:
# Native draft path deliberately keeps _message_id=None, so
# the best-effort edit above never runs for it — the stream
gateway/streaming_tts_consumer.py
+3 -6
View File
@@ -33,6 +33,7 @@ import threading
from typing import Any, Dict, Optional
from gateway.platforms.base import AudioFormat, StreamingTTSHandle
import contextlib
logger = logging.getLogger("gateway.streaming_tts_consumer")
@@ -300,16 +301,12 @@ class StreamingTTSConsumer:
else:
logger.debug("streaming TTS _ABORT sentinel could not be enqueued")
if self._handle is not None and not self._handle.aborted:
try:
with contextlib.suppress(Exception):
self._loop.call_soon_threadsafe(asyncio.create_task, self._safe_abort(reason))
except Exception:
pass
async def wait_complete(self, timeout: float = 10.0) -> bool:
"""Wait for the drain task to finish. Returns True only on full success."""
if self._task is not None:
try:
with contextlib.suppress(asyncio.CancelledError, Exception):
await asyncio.wait_for(asyncio.shield(self._task), timeout=timeout)
except (asyncio.CancelledError, Exception):
pass
return self._completed