mirror of
https://github.com/NousResearch/hermes-agent.git
synced 2026-04-25 00:51:20 +00:00
Replaces the serial for-loop in tick() with ThreadPoolExecutor so all jobs due in a single tick run concurrently. A slow job no longer blocks others from executing, fixing silent job skipping (issue #9086). Thread safety: - Session/delivery env vars migrated from os.environ to ContextVars (gateway/session_context.py) so parallel jobs can't clobber each other's delivery targets. Each thread gets its own copied context. - jobs.json read-modify-write cycles (advance_next_run, mark_job_run) protected by threading.Lock to prevent concurrent save clobber. - send_message_tool reads delivery vars via get_session_env() for ContextVar-aware resolution with os.environ fallback. Configuration: - cron.max_parallel_jobs in config.yaml (null = unbounded, 1 = serial) - HERMES_CRON_MAX_PARALLEL env var override Based on PR #9169 by @VenomMoth1. Fixes #9086
154 lines
5.9 KiB
Python
154 lines
5.9 KiB
Python
"""
|
|
Session-scoped context variables for the Hermes gateway.
|
|
|
|
Replaces the previous ``os.environ``-based session state
|
|
(``HERMES_SESSION_PLATFORM``, ``HERMES_SESSION_CHAT_ID``, etc.) with
|
|
Python's ``contextvars.ContextVar``.
|
|
|
|
**Why this matters**
|
|
|
|
The gateway processes messages concurrently via ``asyncio``. When two
|
|
messages arrive at the same time the old code did:
|
|
|
|
os.environ["HERMES_SESSION_THREAD_ID"] = str(context.source.thread_id)
|
|
|
|
Because ``os.environ`` is *process-global*, Message A's value was
|
|
silently overwritten by Message B before Message A's agent finished
|
|
running. Background-task notifications and tool calls therefore routed
|
|
to the wrong thread.
|
|
|
|
``contextvars.ContextVar`` values are *task-local*: each ``asyncio``
|
|
task (and any ``run_in_executor`` thread it spawns) gets its own copy,
|
|
so concurrent messages never interfere.
|
|
|
|
**Backward compatibility**
|
|
|
|
The public helper ``get_session_env(name, default="")`` mirrors the old
|
|
``os.getenv("HERMES_SESSION_*", ...)`` calls. Existing tool code only
|
|
needs to replace the import + call site:
|
|
|
|
# before
|
|
import os
|
|
platform = os.getenv("HERMES_SESSION_PLATFORM", "")
|
|
|
|
# after
|
|
from gateway.session_context import get_session_env
|
|
platform = get_session_env("HERMES_SESSION_PLATFORM", "")
|
|
"""
|
|
|
|
from contextvars import ContextVar
|
|
from typing import Any
|
|
|
|
# Sentinel to distinguish "never set in this context" from "explicitly set to empty".
|
|
# When a contextvar holds _UNSET, we fall back to os.environ (CLI/cron compat).
|
|
# When it holds "" (after clear_session_vars resets it), we return "" — no fallback.
|
|
_UNSET: Any = object()
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Per-task session variables
|
|
# ---------------------------------------------------------------------------
|
|
|
|
_SESSION_PLATFORM: ContextVar = ContextVar("HERMES_SESSION_PLATFORM", default=_UNSET)
|
|
_SESSION_CHAT_ID: ContextVar = ContextVar("HERMES_SESSION_CHAT_ID", default=_UNSET)
|
|
_SESSION_CHAT_NAME: ContextVar = ContextVar("HERMES_SESSION_CHAT_NAME", default=_UNSET)
|
|
_SESSION_THREAD_ID: ContextVar = ContextVar("HERMES_SESSION_THREAD_ID", default=_UNSET)
|
|
_SESSION_USER_ID: ContextVar = ContextVar("HERMES_SESSION_USER_ID", default=_UNSET)
|
|
_SESSION_USER_NAME: ContextVar = ContextVar("HERMES_SESSION_USER_NAME", default=_UNSET)
|
|
_SESSION_KEY: ContextVar = ContextVar("HERMES_SESSION_KEY", default=_UNSET)
|
|
|
|
# Cron auto-delivery vars — set per-job in run_job() so concurrent jobs
|
|
# don't clobber each other's delivery targets.
|
|
_CRON_AUTO_DELIVER_PLATFORM: ContextVar = ContextVar("HERMES_CRON_AUTO_DELIVER_PLATFORM", default=_UNSET)
|
|
_CRON_AUTO_DELIVER_CHAT_ID: ContextVar = ContextVar("HERMES_CRON_AUTO_DELIVER_CHAT_ID", default=_UNSET)
|
|
_CRON_AUTO_DELIVER_THREAD_ID: ContextVar = ContextVar("HERMES_CRON_AUTO_DELIVER_THREAD_ID", default=_UNSET)
|
|
|
|
_VAR_MAP = {
|
|
"HERMES_SESSION_PLATFORM": _SESSION_PLATFORM,
|
|
"HERMES_SESSION_CHAT_ID": _SESSION_CHAT_ID,
|
|
"HERMES_SESSION_CHAT_NAME": _SESSION_CHAT_NAME,
|
|
"HERMES_SESSION_THREAD_ID": _SESSION_THREAD_ID,
|
|
"HERMES_SESSION_USER_ID": _SESSION_USER_ID,
|
|
"HERMES_SESSION_USER_NAME": _SESSION_USER_NAME,
|
|
"HERMES_SESSION_KEY": _SESSION_KEY,
|
|
"HERMES_CRON_AUTO_DELIVER_PLATFORM": _CRON_AUTO_DELIVER_PLATFORM,
|
|
"HERMES_CRON_AUTO_DELIVER_CHAT_ID": _CRON_AUTO_DELIVER_CHAT_ID,
|
|
"HERMES_CRON_AUTO_DELIVER_THREAD_ID": _CRON_AUTO_DELIVER_THREAD_ID,
|
|
}
|
|
|
|
|
|
def set_session_vars(
|
|
platform: str = "",
|
|
chat_id: str = "",
|
|
chat_name: str = "",
|
|
thread_id: str = "",
|
|
user_id: str = "",
|
|
user_name: str = "",
|
|
session_key: str = "",
|
|
) -> list:
|
|
"""Set all session context variables and return reset tokens.
|
|
|
|
Call ``clear_session_vars(tokens)`` in a ``finally`` block to restore
|
|
the previous values when the handler exits.
|
|
|
|
Returns a list of ``Token`` objects (one per variable) that can be
|
|
passed to ``clear_session_vars``.
|
|
"""
|
|
tokens = [
|
|
_SESSION_PLATFORM.set(platform),
|
|
_SESSION_CHAT_ID.set(chat_id),
|
|
_SESSION_CHAT_NAME.set(chat_name),
|
|
_SESSION_THREAD_ID.set(thread_id),
|
|
_SESSION_USER_ID.set(user_id),
|
|
_SESSION_USER_NAME.set(user_name),
|
|
_SESSION_KEY.set(session_key),
|
|
]
|
|
return tokens
|
|
|
|
|
|
def clear_session_vars(tokens: list) -> None:
|
|
"""Mark session context variables as explicitly cleared.
|
|
|
|
Sets all variables to ``""`` so that ``get_session_env`` returns an empty
|
|
string instead of falling back to (potentially stale) ``os.environ``
|
|
values. The *tokens* argument is accepted for API compatibility with
|
|
callers that saved the return value of ``set_session_vars``, but the
|
|
actual clearing uses ``var.set("")`` rather than ``var.reset(token)``
|
|
to ensure the "explicitly cleared" state is distinguishable from
|
|
"never set" (which holds the ``_UNSET`` sentinel).
|
|
"""
|
|
for var in (
|
|
_SESSION_PLATFORM,
|
|
_SESSION_CHAT_ID,
|
|
_SESSION_CHAT_NAME,
|
|
_SESSION_THREAD_ID,
|
|
_SESSION_USER_ID,
|
|
_SESSION_USER_NAME,
|
|
_SESSION_KEY,
|
|
):
|
|
var.set("")
|
|
|
|
|
|
def get_session_env(name: str, default: str = "") -> str:
|
|
"""Read a session context variable by its legacy ``HERMES_SESSION_*`` name.
|
|
|
|
Drop-in replacement for ``os.getenv("HERMES_SESSION_*", default)``.
|
|
|
|
Resolution order:
|
|
1. Context variable (set by the gateway for concurrency-safe access).
|
|
If the variable was explicitly set (even to ``""``) via
|
|
``set_session_vars`` or ``clear_session_vars``, that value is
|
|
returned — **no fallback to os.environ**.
|
|
2. ``os.environ`` (only when the context variable was never set in
|
|
this context — i.e. CLI, cron scheduler, and test processes that
|
|
don't use ``set_session_vars`` at all).
|
|
3. *default*
|
|
"""
|
|
import os
|
|
|
|
var = _VAR_MAP.get(name)
|
|
if var is not None:
|
|
value = var.get()
|
|
if value is not _UNSET:
|
|
return value
|
|
# Fall back to os.environ for CLI, cron, and test compatibility
|
|
return os.getenv(name, default)
|