diff --git a/AGENTS.md b/AGENTS.md index 9737ceed5f8c..b3bd8032aac0 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -39,12 +39,17 @@ hermes-agent/ │ ├── platforms/ # Adapter per platform (telegram, discord, slack, whatsapp, │ │ # homeassistant, signal, matrix, mattermost, email, sms, │ │ # dingtalk, wecom, weixin, feishu, qqbot, bluebubbles, -│ │ # webhook, api_server, ...). See ADDING_A_PLATFORM.md. +│ │ # yuanbao, webhook, api_server, ...). See ADDING_A_PLATFORM.md. │ └── builtin_hooks/ # Extension point for always-registered gateway hooks (none shipped) ├── plugins/ # Plugin system (see "Plugins" section below) │ ├── memory/ # Memory-provider plugins (honcho, mem0, supermemory, ...) │ ├── context_engine/ # Context-engine plugins -│ └── / # Dashboard, image-gen, disk-cleanup, examples, ... +│ ├── kanban/ # Multi-agent board dispatcher + worker plugin +│ ├── hermes-achievements/ # Gamified achievement tracking +│ ├── observability/ # Metrics / traces / logs plugin +│ ├── image_gen/ # Image-generation providers +│ └── / # disk-cleanup, example-dashboard, google_meet, platforms, +│ # spotify, strike-freedom-cockpit, ... ├── optional-skills/ # Heavier/niche skills shipped but NOT active by default ├── skills/ # Built-in skills bundled with the repo ├── ui-tui/ # Ink (React) terminal UI — `hermes --tui` @@ -55,7 +60,7 @@ hermes-agent/ ├── environments/ # RL training environments (Atropos) ├── scripts/ # run_tests.sh, release.py, auxiliary scripts ├── website/ # Docusaurus docs site -└── tests/ # Pytest suite (~15k tests across ~700 files as of Apr 2026) +└── tests/ # Pytest suite (~17k tests across ~900 files as of May 2026) ``` **User config:** `~/.hermes/config.yaml` (settings), `~/.hermes/.env` (API keys only). @@ -314,9 +319,9 @@ registry.register( ) ``` -**2. Add to `toolsets.py`** — either `_HERMES_CORE_TOOLS` (all platforms) or a new toolset. +**2. Add to `toolsets.py`** — either `_HERMES_CORE_TOOLS` (all platforms) or a new toolset. **This step is required:** auto-discovery imports the tool and registers its schema, but the tool is only *exposed to an agent* if its name appears in a toolset. `_HERMES_CORE_TOOLS` is not dead code — it's the default bundle every platform's base toolset inherits from. -Auto-discovery: any `tools/*.py` file with a top-level `registry.register()` call is imported automatically — no manual import list to maintain. +Auto-discovery: any `tools/*.py` file with a top-level `registry.register()` call is imported automatically — no manual import list to maintain. Wiring into a toolset is still a deliberate, manual step. The registry handles schema collection, dispatch, availability checking, and error wrapping. All handlers MUST return a JSON string. @@ -338,6 +343,22 @@ The registry handles schema collection, dispatch, availability checking, and err section is handled automatically by the deep-merge and does NOT require a version bump. +### Top-level `config.yaml` sections (non-exhaustive): + +`model`, `agent`, `terminal`, `compression`, `display`, `stt`, `tts`, +`memory`, `security`, `delegation`, `smart_model_routing`, `checkpoints`, +`auxiliary`, `curator`, `skills`, `gateway`, `logging`, `cron`, `profiles`, +`plugins`, `honcho`. + +`auxiliary` holds per-task overrides for side-LLM work (curator, vision, +embedding, title generation, session_search, etc.) — each task can pin +its own provider/model/base_url/max_tokens/reasoning_effort. See +`agent/auxiliary_client.py::_resolve_auto` for resolution order. + +`curator` holds the background skill-maintenance config — +`enabled`, `interval_hours`, `min_idle_hours`, `stale_after_days`, +`archive_after_days`, `backup` (nested). + ### .env variables (SECRETS ONLY — API keys, tokens, passwords): 1. Add to `OPTIONAL_ENV_VARS` in `hermes_cli/config.py` with metadata: ```python @@ -544,11 +565,176 @@ niche skills belong in `optional-skills/`. ### SKILL.md frontmatter -Standard fields: `name`, `description`, `version`, `platforms` -(OS-gating list: `[macos]`, `[linux, macos]`, ...), +Standard fields: `name`, `description`, `version`, `author`, `license`, +`platforms` (OS-gating list: `[macos]`, `[linux, macos]`, ...), `metadata.hermes.tags`, `metadata.hermes.category`, -`metadata.hermes.config` (config.yaml settings the skill needs — stored -under `skills.config.`, prompted during setup, injected at load time). +`metadata.hermes.related_skills`, `metadata.hermes.config` (config.yaml +settings the skill needs — stored under `skills.config.`, prompted +during setup, injected at load time). + +Top-level `tags:` and `category:` are also accepted and mirrored from +`metadata.hermes.*` by the loader. + +--- + +## Toolsets + +All toolsets are defined in `toolsets.py` as a single `TOOLSETS` dict. +Each platform's adapter picks a base toolset (e.g. Telegram uses +`"messaging"`); `_HERMES_CORE_TOOLS` is the default bundle most +platforms inherit from. + +Current toolset keys: `browser`, `clarify`, `code_execution`, `cronjob`, +`debugging`, `delegation`, `discord`, `discord_admin`, `feishu_doc`, +`feishu_drive`, `file`, `homeassistant`, `image_gen`, `kanban`, `memory`, +`messaging`, `moa`, `rl`, `safe`, `search`, `session_search`, `skills`, +`spotify`, `terminal`, `todo`, `tts`, `video`, `vision`, `web`, `yuanbao`. + +Enable/disable per platform via `hermes tools` (the curses UI) or the +`tools..enabled` / `tools..disabled` lists in +`config.yaml`. + +--- + +## Delegation (`delegate_task`) + +`tools/delegate_tool.py` spawns a subagent with an isolated +context + terminal session. Synchronous: the parent waits for the +child's summary before continuing its own loop — if the parent is +interrupted, the child is cancelled. + +Two shapes: + +- **Single:** pass `goal` (+ optional `context`, `toolsets`). +- **Batch (parallel):** pass `tasks: [...]` — each gets its own subagent + running concurrently. Concurrency is capped by + `delegation.max_concurrent_children` (default 3). + +Roles: + +- `role="leaf"` (default) — focused worker. Cannot call `delegate_task`, + `clarify`, `memory`, `send_message`, `execute_code`. +- `role="orchestrator"` — retains `delegate_task` so it can spawn its + own workers. Gated by `delegation.orchestrator_enabled` (default true) + and bounded by `delegation.max_spawn_depth` (default 2). + +Key config knobs (under `delegation:` in `config.yaml`): +`max_concurrent_children`, `max_spawn_depth`, `child_timeout_seconds`, +`orchestrator_enabled`, `subagent_auto_approve`, `inherit_mcp_toolsets`, +`max_iterations`. + +Synchronicity rule: delegate_task is **not** durable. For long-running +work that must outlive the current turn, use `cronjob` or +`terminal(background=True, notify_on_complete=True)` instead. + +--- + +## Curator (skill lifecycle) + +Background skill-maintenance system that tracks usage on agent-created +skills and auto-archives stale ones. Users never lose skills; archives +go to `~/.hermes/skills/.archive/` and are restorable. + +- **Core:** `agent/curator.py` (review loop, auto-transitions, LLM review + prompt) + `agent/curator_backup.py` (pre-run tar.gz snapshots). +- **CLI:** `hermes_cli/curator.py` wires `hermes curator ` where + verbs are: `status`, `run`, `pause`, `resume`, `pin`, `unpin`, + `archive`, `restore`, `prune`, `backup`, `rollback`. +- **Telemetry:** `tools/skill_usage.py` owns the sidecar + `~/.hermes/skills/.usage.json` — per-skill `use_count`, `view_count`, + `patch_count`, `last_activity_at`, `state` (active / stale / + archived), `pinned`. + +Invariants: +- Curator only touches skills with `created_by: "agent"` provenance — + bundled + hub-installed skills are off-limits. +- Never deletes; max destructive action is archive. +- Pinned skills are exempt from every auto-transition and from the + LLM review pass. +- `skill_manage(action="delete")` refuses pinned skills; patch/edit/ + write_file/remove_file go through so the agent can keep improving + pinned skills. + +Config section (`curator:` in `config.yaml`): +`enabled`, `interval_hours`, `min_idle_hours`, `stale_after_days`, +`archive_after_days`, `backup.*`. + +Full user-facing docs: `website/docs/user-guide/features/curator.md`. + +--- + +## Cron (scheduled jobs) + +`cron/jobs.py` (job store) + `cron/scheduler.py` (tick loop). Agents +schedule jobs via the `cronjob` tool; users via `hermes cron ` +(`list`, `add`, `edit`, `pause`, `resume`, `run`, `remove`) or the +`/cron` slash command. + +Supported schedule formats: +- Duration: `"30m"`, `"2h"`, `"1d"` +- "every" phrase: `"every 2h"`, `"every monday 9am"` +- 5-field cron expression: `"0 9 * * *"` +- ISO timestamp (one-shot): `"2026-06-01T09:00:00Z"` + +Per-job fields include `skills` (load specific skills), `model` / +`provider` overrides, `script` (pre-run data-collection script whose +stdout is injected into the prompt; `no_agent=True` turns the script +into the entire job), `context_from` (chain job A's last output into +job B's prompt), `workdir` (run in a specific directory with its +`AGENTS.md`/`CLAUDE.md` loaded), and multi-platform delivery. + +Hardening invariants: +- **3-minute hard interrupt** on cron sessions — runaway agent loops + cannot monopolize the scheduler. +- Catchup window: half the job's period, clamped to 120s–2h. +- Grace window: 120s for one-shot jobs whose fire time was missed. +- File lock at `~/.hermes/cron/.tick.lock` prevents duplicate ticks + across processes. +- Cron sessions pass `skip_memory=True` by default; memory providers + intentionally do not run during cron. + +Cron deliveries are **not** mirrored into the target gateway session — +they land in their own cron session with a header/footer frame so the +main conversation's message-role alternation stays intact. + +--- + +## Kanban (multi-agent work queue) + +Durable SQLite-backed board that lets multiple profiles / workers +collaborate on shared tasks. Users drive it via `hermes kanban `; +workers spawned by the dispatcher drive it via a dedicated `kanban_*` +toolset so their schema footprint is zero when they're not inside a +kanban task. + +- **CLI:** `hermes_cli/kanban.py` wires `hermes kanban` with verbs + `init`, `create`, `list` (alias `ls`), `show`, `assign`, `link`, + `unlink`, `comment`, `complete`, `block`, `unblock`, `archive`, + `tail`, plus less-commonly-used `watch`, `stats`, `runs`, `log`, + `assignees`, `heartbeat`, `notify-*`, `dispatch`, `daemon`, `gc`. +- **Worker toolset:** `tools/kanban_tools.py` exposes `kanban_show`, + `kanban_complete`, `kanban_block`, `kanban_heartbeat`, `kanban_comment`, + `kanban_create`, `kanban_link` — gated by `HERMES_KANBAN_TASK` so + the schema only appears for processes actually running as a worker. +- **Dispatcher:** long-lived loop that (default every 60s) reclaims + stale claims, promotes ready tasks, atomically claims, and spawns + assigned profiles. Runs **inside the gateway** by default via + `kanban.dispatch_in_gateway: true`. +- **Plugin assets:** `plugins/kanban/dashboard/` (web UI) + + `plugins/kanban/systemd/` (`hermes-kanban-dispatcher.service` for + standalone dispatcher deployment). + +Isolation model: +- **Board** is the hard boundary — workers are spawned with + `HERMES_KANBAN_BOARD` pinned in their env so they can't see other + boards. +- **Tenant** is a soft namespace *within* a board — one specialist + fleet can serve multiple businesses with workspace-path + memory-key + isolation. +- After ~5 consecutive spawn failures on the same task the dispatcher + auto-blocks it to prevent spin loops. + +Full user-facing docs: `website/docs/user-guide/features/kanban.md`. --- diff --git a/acp_adapter/session.py b/acp_adapter/session.py index d6dace66b4ea..61d06432a715 100644 --- a/acp_adapter/session.py +++ b/acp_adapter/session.py @@ -466,17 +466,10 @@ class SessionManager: except Exception: logger.debug("Failed to update ACP session metadata", exc_info=True) - # Replace stored messages with current history. - db.clear_messages(state.session_id) - for msg in state.history: - db.append_message( - session_id=state.session_id, - role=msg.get("role", "user"), - content=msg.get("content"), - tool_name=msg.get("tool_name") or msg.get("name"), - tool_calls=msg.get("tool_calls"), - tool_call_id=msg.get("tool_call_id"), - ) + # Replace stored messages with current history atomically so a + # mid-rewrite failure rolls back and the previously persisted + # conversation is preserved (salvaged from #13675). + db.replace_messages(state.session_id, state.history) except Exception: logger.warning("Failed to persist ACP session %s", state.session_id, exc_info=True) diff --git a/agent/auxiliary_client.py b/agent/auxiliary_client.py index 4c706748a0b0..54a1d63a7fea 100644 --- a/agent/auxiliary_client.py +++ b/agent/auxiliary_client.py @@ -259,10 +259,12 @@ _PROVIDERS_WITHOUT_VISION: frozenset = frozenset({ "kimi-coding-cn", }) -# OpenRouter app attribution headers (base — always sent) +# OpenRouter app attribution headers (base — always sent). +# `X-Title` is the canonical attribution header OpenRouter's dashboard +# reads; the previous `X-OpenRouter-Title` label was not recognized there. _OR_HEADERS_BASE = { "HTTP-Referer": "https://hermes-agent.nousresearch.com", - "X-OpenRouter-Title": "Hermes Agent", + "X-Title": "Hermes Agent", "X-OpenRouter-Categories": "productivity,cli-agent", } @@ -567,7 +569,12 @@ class _CodexCompletionsAdapter: # API allows it. pass else: - effort = reasoning_cfg.get("effort", "medium") + # Truthy-only check mirrors agent/transports/codex.py + # build_kwargs(): falsy values (None, "", 0) fall back + # to the default rather than being forwarded to the + # Codex backend, which rejects e.g. {"effort": null} + # with a 400. + effort = reasoning_cfg.get("effort") or "medium" # Codex backend rejects "minimal"; clamp to "low" to # match the main-agent Codex transport behavior. if effort == "minimal": @@ -1643,6 +1650,39 @@ def _is_payment_error(exc: Exception) -> bool: return False +def _is_rate_limit_error(exc: Exception) -> bool: + """Detect rate-limit errors that warrant provider fallback. + + Returns True for HTTP 429 errors whose message indicates rate limiting + (as opposed to billing/quota exhaustion, which _is_payment_error handles). + Also catches OpenAI SDK RateLimitError instances that may not set + .status_code on the exception object. + """ + status = getattr(exc, "status_code", None) + err_lower = str(exc).lower() + + # OpenAI SDK's RateLimitError sometimes omits .status_code — + # detect by class name so we don't miss these. (PR #8023 pattern) + if type(exc).__name__ == "RateLimitError": + return True + + if status == 429: + # Distinguish rate-limit from billing: billing keywords are handled + # by _is_payment_error, everything else on 429 is a rate limit. + if any(kw in err_lower for kw in ( + "rate limit", "rate_limit", "too many requests", + "try again", "retry after", "resets in", + )): + return True + # Generic 429 without billing keywords = likely a rate limit + if not any(kw in err_lower for kw in ( + "credits", "insufficient funds", "billing", + "payment required", "can only afford", + )): + return True + return False + + def _is_connection_error(exc: Exception) -> bool: """Detect connection/network errors that warrant provider fallback. @@ -3127,8 +3167,14 @@ def _resolve_task_provider_model( if task: # Config.yaml is the primary source for per-task overrides. - if cfg_base_url: + if cfg_base_url and cfg_api_key: + # Both base_url and api_key explicitly set → custom endpoint. return "custom", resolved_model, cfg_base_url, cfg_api_key, resolved_api_mode + if cfg_base_url and cfg_provider and cfg_provider != "auto": + # base_url set without api_key but with a known provider — use + # the provider so it can resolve credentials from env vars + # (e.g. OPENROUTER_API_KEY) instead of locking into "custom". + return cfg_provider, resolved_model, cfg_base_url, None, resolved_api_mode if cfg_provider and cfg_provider != "auto": return cfg_provider, resolved_model, None, None, resolved_api_mode @@ -3529,7 +3575,7 @@ def call_llm( except Exception as retry_err: # If the max_tokens retry also hits a payment or connection # error, fall through to the fallback chain below. - if not (_is_payment_error(retry_err) or _is_connection_error(retry_err)): + if not (_is_payment_error(retry_err) or _is_connection_error(retry_err) or _is_rate_limit_error(retry_err)): raise first_err = retry_err @@ -3612,13 +3658,27 @@ def call_llm( # Codex/OAuth tokens that authenticate but whose endpoint is down, # and providers the user never configured that got picked up by # the auto-detection chain. - should_fallback = _is_payment_error(first_err) or _is_connection_error(first_err) + # + # ── Rate-limit fallback (#13579) ───────────────────────────── + # When the provider returns a 429 rate-limit (not billing), fall + # back to an alternative provider instead of exhausting retries + # against the same rate-limited endpoint. + should_fallback = ( + _is_payment_error(first_err) + or _is_connection_error(first_err) + or _is_rate_limit_error(first_err) + ) # Only try alternative providers when the user didn't explicitly # configure this task's provider. Explicit provider = hard constraint; # auto (the default) = best-effort fallback chain. (#7559) is_auto = resolved_provider in ("auto", "", None) if should_fallback and is_auto: - reason = "payment error" if _is_payment_error(first_err) else "connection error" + if _is_payment_error(first_err): + reason = "payment error" + elif _is_rate_limit_error(first_err): + reason = "rate limit" + else: + reason = "connection error" logger.info("Auxiliary %s: %s on %s (%s), trying fallback", task or "call", reason, resolved_provider, first_err) fb_client, fb_model, fb_label = _try_payment_fallback( @@ -3821,7 +3881,7 @@ async def async_call_llm( except Exception as retry_err: # If the max_tokens retry also hits a payment or connection # error, fall through to the fallback chain below. - if not (_is_payment_error(retry_err) or _is_connection_error(retry_err)): + if not (_is_payment_error(retry_err) or _is_connection_error(retry_err) or _is_rate_limit_error(retry_err)): raise first_err = retry_err @@ -3890,11 +3950,20 @@ async def async_call_llm( return _validate_llm_response( await retry_client.chat.completions.create(**retry_kwargs), task) - # ── Payment / connection fallback (mirrors sync call_llm) ───── - should_fallback = _is_payment_error(first_err) or _is_connection_error(first_err) + # ── Payment / connection / rate-limit fallback (mirrors sync call_llm) ── + should_fallback = ( + _is_payment_error(first_err) + or _is_connection_error(first_err) + or _is_rate_limit_error(first_err) + ) is_auto = resolved_provider in ("auto", "", None) if should_fallback and is_auto: - reason = "payment error" if _is_payment_error(first_err) else "connection error" + if _is_payment_error(first_err): + reason = "payment error" + elif _is_rate_limit_error(first_err): + reason = "rate limit" + else: + reason = "connection error" logger.info("Auxiliary %s (async): %s on %s (%s), trying fallback", task or "call", reason, resolved_provider, first_err) fb_client, fb_model, fb_label = _try_payment_fallback( diff --git a/agent/context_compressor.py b/agent/context_compressor.py index f9111f96004f..20f35fed5f14 100644 --- a/agent/context_compressor.py +++ b/agent/context_compressor.py @@ -993,15 +993,39 @@ The user has requested that this compaction PRIORITISE preserving all informatio return None @staticmethod - def _with_summary_prefix(summary: str) -> str: - """Normalize summary text to the current compaction handoff format.""" + def _strip_summary_prefix(summary: str) -> str: + """Return summary body without the current or legacy handoff prefix.""" text = (summary or "").strip() - for prefix in (LEGACY_SUMMARY_PREFIX, SUMMARY_PREFIX): + for prefix in (SUMMARY_PREFIX, LEGACY_SUMMARY_PREFIX): if text.startswith(prefix): - text = text[len(prefix):].lstrip() - break + return text[len(prefix):].lstrip() + return text + + @classmethod + def _with_summary_prefix(cls, summary: str) -> str: + """Normalize summary text to the current compaction handoff format.""" + text = cls._strip_summary_prefix(summary) return f"{SUMMARY_PREFIX}\n{text}" if text else SUMMARY_PREFIX + @staticmethod + def _is_context_summary_content(content: Any) -> bool: + text = _content_text_for_contains(content).lstrip() + return text.startswith(SUMMARY_PREFIX) or text.startswith(LEGACY_SUMMARY_PREFIX) + + @classmethod + def _find_latest_context_summary( + cls, + messages: List[Dict[str, Any]], + start: int, + end: int, + ) -> tuple[Optional[int], str]: + """Find the newest handoff summary inside a compression window.""" + for idx in range(end - 1, start - 1, -1): + content = messages[idx].get("content") + if cls._is_context_summary_content(content): + return idx, cls._strip_summary_prefix(_content_text_for_contains(content)) + return None, "" + # ------------------------------------------------------------------ # Tool-call / tool-result pair integrity helpers # ------------------------------------------------------------------ @@ -1308,6 +1332,15 @@ The user has requested that this compaction PRIORITISE preserving all informatio return messages turns_to_summarize = messages[compress_start:compress_end] + summary_idx, summary_body = self._find_latest_context_summary( + messages, + compress_start, + compress_end, + ) + if summary_idx is not None: + if summary_body and not self._previous_summary: + self._previous_summary = summary_body + turns_to_summarize = messages[summary_idx + 1:compress_end] if not self.quiet_mode: logger.info( @@ -1385,6 +1418,19 @@ The user has requested that this compaction PRIORITISE preserving all informatio # Merge the summary into the first tail message instead # of inserting a standalone message that breaks alternation. _merge_summary_into_tail = True + + # When the summary lands as a standalone role="user" message, + # weak models read the verbatim "## Active Task" quote of a past + # user request as fresh input (#11475, #14521). Append the explicit + # end marker — the same one used in the merge-into-tail path — so + # the model has a clear "summary above, not new input" signal. + if not _merge_summary_into_tail and summary_role == "user": + summary = ( + summary + + "\n\n--- END OF CONTEXT SUMMARY — " + "respond to the message below, not the summary above ---" + ) + if not _merge_summary_into_tail: compressed.append({"role": summary_role, "content": summary}) diff --git a/agent/error_classifier.py b/agent/error_classifier.py index 67feaa4304f5..419a984b75e6 100644 --- a/agent/error_classifier.py +++ b/agent/error_classifier.py @@ -55,6 +55,7 @@ class FailoverReason(enum.Enum): thinking_signature = "thinking_signature" # Anthropic thinking block sig invalid long_context_tier = "long_context_tier" # Anthropic "extra usage" tier gate oauth_long_context_beta_forbidden = "oauth_long_context_beta_forbidden" # Anthropic OAuth subscription rejects 1M context beta — disable beta and retry + llama_cpp_grammar_pattern = "llama_cpp_grammar_pattern" # llama.cpp json-schema-to-grammar rejects regex escapes in `pattern` / `format` — strip from tools and retry # Catch-all unknown = "unknown" # Unclassifiable — retry with backoff @@ -470,6 +471,31 @@ def classify_api_error( should_compress=False, ) + # llama.cpp's ``json-schema-to-grammar`` converter (used by its OAI + # server to build GBNF tool-call parsers) rejects regex escape classes + # like ``\d``/``\w``/``\s`` and most ``format`` values. MCP servers + # routinely emit ``"pattern": "\\d{4}-\\d{2}-\\d{2}"`` for date/phone/ + # email params. llama.cpp surfaces this as HTTP 400 with one of a few + # recognizable phrases; on match we strip ``pattern``/``format`` from + # ``self.tools`` in the retry loop and retry once. Cloud providers are + # unaffected — they accept these keywords and we never hit this branch. + if ( + status_code == 400 + and ( + "error parsing grammar" in error_msg + or "json-schema-to-grammar" in error_msg + or ( + "unable to generate parser" in error_msg + and "template" in error_msg + ) + ) + ): + return _result( + FailoverReason.llama_cpp_grammar_pattern, + retryable=True, + should_compress=False, + ) + # ── 2. HTTP status code classification ────────────────────────── if status_code is not None: diff --git a/agent/i18n.py b/agent/i18n.py new file mode 100644 index 000000000000..98d7ebce9a58 --- /dev/null +++ b/agent/i18n.py @@ -0,0 +1,230 @@ +"""Lightweight internationalization (i18n) for Hermes static user-facing messages. + +Scope (thin slice, by design): only the highest-impact static strings shown +to the user by Hermes itself -- approval prompts, a handful of gateway slash +command replies, restart-drain notices. Agent-generated output, log lines, +error tracebacks, tool outputs, and slash-command descriptions all stay in +English. + +Catalog files live under ``locales/.yaml`` at the repo root. Each +catalog is a flat dict keyed by dotted paths (e.g. ``approval.choose`` or +``gateway.approval_expired``). Missing keys fall back to English; if English +is missing too, the key path itself is returned so a broken catalog never +crashes the agent. + +Usage:: + + from agent.i18n import t + print(t("approval.choose_long")) # current lang + print(t("gateway.draining", count=3)) # {count} formatted + print(t("approval.choose_long", lang="zh")) # explicit override + +Language resolution order: + 1. Explicit ``lang=`` argument passed to :func:`t` + 2. ``HERMES_LANGUAGE`` environment variable (for tests / quick override) + 3. ``display.language`` from config.yaml + 4. ``"en"`` (baseline) + +Supported languages: en, zh, ja, de, es. Unknown values fall back to en. +""" + +from __future__ import annotations + +import logging +import os +import threading +from functools import lru_cache +from pathlib import Path +from typing import Any + +logger = logging.getLogger(__name__) + +SUPPORTED_LANGUAGES: tuple[str, ...] = ("en", "zh", "ja", "de", "es") +DEFAULT_LANGUAGE = "en" + +# Accept a few natural aliases so users who type "chinese" / "zh-CN" / "jp" +# get the right catalog instead of silently falling back to English. +_LANGUAGE_ALIASES: dict[str, str] = { + "english": "en", "en-us": "en", "en-gb": "en", + "chinese": "zh", "mandarin": "zh", "zh-cn": "zh", "zh-tw": "zh", "zh-hans": "zh", "zh-hant": "zh", + "japanese": "ja", "jp": "ja", "ja-jp": "ja", + "german": "de", "deutsch": "de", "de-de": "de", + "spanish": "es", "español": "es", "espanol": "es", "es-es": "es", "es-mx": "es", +} + +_catalog_cache: dict[str, dict[str, str]] = {} +_catalog_lock = threading.Lock() + + +def _locales_dir() -> Path: + """Return the directory containing locale YAML files. + + Lives next to the repo root so both the bundled install and editable + checkouts find it without PYTHONPATH gymnastics. + """ + # agent/i18n.py -> agent/ -> repo root + return Path(__file__).resolve().parent.parent / "locales" + + +def _normalize_lang(value: Any) -> str: + """Normalize a user-supplied language value to a supported code. + + Accepts supported codes directly, common aliases (``chinese`` -> ``zh``), + and case-insensitive regional tags (``zh-CN`` -> ``zh``). Returns the + default language for unknown values. + """ + if not isinstance(value, str): + return DEFAULT_LANGUAGE + key = value.strip().lower() + if not key: + return DEFAULT_LANGUAGE + if key in SUPPORTED_LANGUAGES: + return key + if key in _LANGUAGE_ALIASES: + return _LANGUAGE_ALIASES[key] + # Try stripping a region suffix (e.g. "pt-br" -> "pt" won't be supported, + # but "zh-CN" -> "zh" will). + base = key.split("-", 1)[0] + if base in SUPPORTED_LANGUAGES: + return base + return DEFAULT_LANGUAGE + + +def _load_catalog(lang: str) -> dict[str, str]: + """Load and flatten one locale YAML file into a dotted-key dict. + + YAML files can be nested for human readability; this produces the flat + key space :func:`t` expects. Cached per-language for the process. + """ + with _catalog_lock: + cached = _catalog_cache.get(lang) + if cached is not None: + return cached + + path = _locales_dir() / f"{lang}.yaml" + if not path.is_file(): + logger.debug("i18n catalog missing for %s at %s", lang, path) + with _catalog_lock: + _catalog_cache[lang] = {} + return {} + + try: + import yaml # PyYAML is already a hermes dependency + with path.open("r", encoding="utf-8") as f: + raw = yaml.safe_load(f) or {} + except Exception as exc: + logger.warning("Failed to load i18n catalog %s: %s", path, exc) + with _catalog_lock: + _catalog_cache[lang] = {} + return {} + + flat: dict[str, str] = {} + _flatten_into(raw, "", flat) + with _catalog_lock: + _catalog_cache[lang] = flat + return flat + + +def _flatten_into(node: Any, prefix: str, out: dict[str, str]) -> None: + if isinstance(node, dict): + for key, value in node.items(): + child_key = f"{prefix}.{key}" if prefix else str(key) + _flatten_into(value, child_key, out) + elif isinstance(node, str): + out[prefix] = node + # Non-string, non-dict leaves are ignored -- catalogs are text-only. + + +@lru_cache(maxsize=1) +def _config_language_cached() -> str | None: + """Read ``display.language`` from config.yaml once per process. + + Cached because ``t()`` is called in hot paths (every approval prompt, + every gateway reply) and re-reading YAML each call would be wasteful. + ``reset_language_cache()`` clears this when config changes at runtime + (e.g. after the setup wizard). + """ + try: + from hermes_cli.config import load_config + cfg = load_config() + lang = (cfg.get("display") or {}).get("language") + if lang: + return _normalize_lang(lang) + except Exception as exc: + logger.debug("Could not read display.language from config: %s", exc) + return None + + +def reset_language_cache() -> None: + """Invalidate cached language resolution and catalogs. + + Call after :func:`hermes_cli.config.save_config` if a running process + needs to pick up a changed ``display.language`` without restart. + """ + _config_language_cached.cache_clear() + with _catalog_lock: + _catalog_cache.clear() + + +def get_language() -> str: + """Resolve the active language using env > config > default order.""" + env_lang = os.environ.get("HERMES_LANGUAGE") + if env_lang: + return _normalize_lang(env_lang) + cfg_lang = _config_language_cached() + if cfg_lang: + return cfg_lang + return DEFAULT_LANGUAGE + + +def t(key: str, lang: str | None = None, **format_kwargs: Any) -> str: + """Translate a dotted key to the active language. + + Parameters + ---------- + key + Dotted path into the catalog, e.g. ``"approval.choose_long"``. + lang + Explicit language override. Takes precedence over env + config. + **format_kwargs + ``str.format`` substitution arguments (``t("gateway.drain", count=3)`` + expects a catalog entry with a ``{count}`` placeholder). + + Returns + ------- + The translated string, or the English fallback if the key is missing in + the target language, or the bare key if English is also missing. + """ + target = _normalize_lang(lang) if lang else get_language() + catalog = _load_catalog(target) + value = catalog.get(key) + + if value is None and target != DEFAULT_LANGUAGE: + # Fall through to English rather than showing a key path to the user. + value = _load_catalog(DEFAULT_LANGUAGE).get(key) + + if value is None: + # Last-ditch: return the key itself. A broken catalog should not + # crash anything; it just looks ugly until someone fixes it. + logger.debug("i18n miss: key=%r lang=%r", key, target) + value = key + + if format_kwargs: + try: + return value.format(**format_kwargs) + except (KeyError, IndexError, ValueError) as exc: + logger.warning( + "i18n format failed for key=%r lang=%r kwargs=%r: %s", + key, target, format_kwargs, exc, + ) + return value + return value + + +__all__ = [ + "SUPPORTED_LANGUAGES", + "DEFAULT_LANGUAGE", + "t", + "get_language", + "reset_language_cache", +] diff --git a/agent/prompt_builder.py b/agent/prompt_builder.py index 8494a70eef2a..2f00020cc1ce 100644 --- a/agent/prompt_builder.py +++ b/agent/prompt_builder.py @@ -513,6 +513,12 @@ PLATFORM_HINTS = { "image and is the WRONG path. Bare Unicode emoji in text is also not a substitute " "— when a sticker is the right response, use yb_send_sticker." ), + "api_server": ( + "You're responding through an API server. The rendering layer is unknown — " + "assume plain text. No markdown formatting (no asterisks, bullets, headers, " + "code fences). Treat this like a conversation, not a document. Keep responses " + "brief and natural." + ), } # --------------------------------------------------------------------------- diff --git a/agent/think_scrubber.py b/agent/think_scrubber.py new file mode 100644 index 000000000000..44ddcacff704 --- /dev/null +++ b/agent/think_scrubber.py @@ -0,0 +1,386 @@ +"""Stateful scrubber for reasoning/thinking blocks in streamed assistant text. + +``run_agent._strip_think_blocks`` is regex-based and correct for a complete +string, but when it runs *per-delta* in ``_fire_stream_delta`` it destroys +the state that downstream consumers (CLI ``_stream_delta``, gateway +``GatewayStreamConsumer._filter_and_accumulate``) rely on. + +Concretely, when MiniMax-M2.7 streams + + delta1 = "" + delta2 = "Let me check their config" + delta3 = "" + +the per-delta regex erases delta1 entirely (case 2: unterminated-open at +boundary matches ``^...``), so the downstream state machine never +sees the open tag, treats delta2 as regular content, and leaks reasoning +to the user. Consumers that don't run their own state machine (ACP, +api_server, TTS) never had any defence at all — they just emitted +whatever survived the upstream regex. + +This module centralises the tag-suppression state machine at the +upstream layer so every stream_delta_callback sees text that has +already had reasoning blocks removed. Partial tags at delta +boundaries are held back until the next delta resolves them, and +end-of-stream flushing surfaces any held-back prose that turned out +not to be a real tag. + +Usage:: + + scrubber = StreamingThinkScrubber() + for delta in stream: + visible = scrubber.feed(delta) + if visible: + emit(visible) + tail = scrubber.flush() # at end of stream + if tail: + emit(tail) + +The scrubber is re-entrant per agent instance. Call ``reset()`` at +the top of each new turn so a hung block from an interrupted prior +stream cannot taint the next turn's output. + +Tag variants handled (case-insensitive): + ````, ````, ````, ````, + ````. + +Block-boundary rule for opens: an opening tag is only treated as a +reasoning-block opener when it appears at the start of the stream, +after a newline (optionally followed by whitespace), or when only +whitespace has been emitted on the current line. This prevents prose +that *mentions* the tag name (e.g. ``"use tags here"``) from +being incorrectly suppressed. Closed pairs (``X``) are +always suppressed regardless of boundary; a closed pair is an +intentional, bounded construct. +""" + +from __future__ import annotations + +from typing import Tuple + +__all__ = ["StreamingThinkScrubber"] + + +class StreamingThinkScrubber: + """Stateful scrubber for streaming reasoning/thinking blocks. + + State machine: + - ``_in_block``: True while inside an opened block, waiting for + a close tag. All text inside is discarded. + - ``_buf``: held-back partial-tag tail. Emitted / discarded on + the next ``feed()`` call or by ``flush()``. + - ``_last_emitted_ended_newline``: True iff the most recent + emission to the consumer ended with ``\\n``, or nothing has + been emitted yet (start-of-stream counts as a boundary). Used + to decide whether an open tag at buffer position 0 is at a + block boundary. + """ + + _OPEN_TAG_NAMES: Tuple[str, ...] = ( + "think", + "thinking", + "reasoning", + "thought", + "REASONING_SCRATCHPAD", + ) + + # Materialise literal tag strings so the hot path does string + # operations, not regex compilation per feed(). + _OPEN_TAGS: Tuple[str, ...] = tuple(f"<{name}>" for name in _OPEN_TAG_NAMES) + _CLOSE_TAGS: Tuple[str, ...] = tuple(f"" for name in _OPEN_TAG_NAMES) + + # Pre-compute the longest tag (for partial-tag hold-back bound). + _MAX_TAG_LEN: int = max(len(tag) for tag in _OPEN_TAGS + _CLOSE_TAGS) + + def __init__(self) -> None: + self._in_block: bool = False + self._buf: str = "" + self._last_emitted_ended_newline: bool = True + + def reset(self) -> None: + """Reset all state. Call at the top of every new turn.""" + self._in_block = False + self._buf = "" + self._last_emitted_ended_newline = True + + def feed(self, text: str) -> str: + """Feed one delta; return the scrubbed visible portion. + + May return an empty string when the entire delta is reasoning + content or is being held back pending resolution of a partial + tag at the boundary. + """ + if not text: + return "" + buf = self._buf + text + self._buf = "" + out: list[str] = [] + + while buf: + if self._in_block: + # Hunt for the earliest close tag. + close_idx, close_len = self._find_first_tag( + buf, self._CLOSE_TAGS, + ) + if close_idx == -1: + # No close yet — hold back a potential partial + # close-tag prefix; discard everything else. + held = self._max_partial_suffix(buf, self._CLOSE_TAGS) + self._buf = buf[-held:] if held else "" + return "".join(out) + # Found close: discard block content + tag, continue. + buf = buf[close_idx + close_len:] + self._in_block = False + else: + # Priority 1 — closed X pair anywhere in + # buf. Closed pairs are always an intentional, + # bounded construct (even mid-line prose containing + # an open/close pair is almost certainly a model + # leaking reasoning inline), so no boundary gating. + pair = self._find_earliest_closed_pair(buf) + # Priority 2 — unterminated open tag at a block + # boundary. Boundary-gated so prose that mentions + # '' isn't over-stripped. + open_idx, open_len = self._find_open_at_boundary( + buf, out, + ) + + # Pick whichever match comes earliest in the buffer. + if pair is not None and ( + open_idx == -1 or pair[0] <= open_idx + ): + start_idx, end_idx = pair + preceding = buf[:start_idx] + if preceding: + preceding = self._strip_orphan_close_tags(preceding) + if preceding: + out.append(preceding) + self._last_emitted_ended_newline = ( + preceding.endswith("\n") + ) + buf = buf[end_idx:] + continue + + if open_idx != -1: + # Unterminated open at boundary — emit preceding, + # enter block, continue loop with remainder. + preceding = buf[:open_idx] + if preceding: + preceding = self._strip_orphan_close_tags(preceding) + if preceding: + out.append(preceding) + self._last_emitted_ended_newline = ( + preceding.endswith("\n") + ) + self._in_block = True + buf = buf[open_idx + open_len:] + continue + + # No resolvable tag structure in buf. Hold back any + # partial-tag prefix at the tail so a split tag + # across deltas isn't missed, then emit the rest. + held = self._max_partial_suffix(buf, self._OPEN_TAGS) + held_close = self._max_partial_suffix( + buf, self._CLOSE_TAGS, + ) + held = max(held, held_close) + if held: + emit_text = buf[:-held] + self._buf = buf[-held:] + else: + emit_text = buf + self._buf = "" + if emit_text: + emit_text = self._strip_orphan_close_tags(emit_text) + if emit_text: + out.append(emit_text) + self._last_emitted_ended_newline = ( + emit_text.endswith("\n") + ) + return "".join(out) + + return "".join(out) + + def flush(self) -> str: + """End-of-stream flush. + + If still inside an unterminated block, held-back content is + discarded — leaking partial reasoning is worse than a + truncated answer. Otherwise the held-back partial-tag tail is + emitted verbatim (it turned out not to be a real tag prefix). + """ + if self._in_block: + self._buf = "" + self._in_block = False + return "" + tail = self._buf + self._buf = "" + if not tail: + return "" + tail = self._strip_orphan_close_tags(tail) + if tail: + self._last_emitted_ended_newline = tail.endswith("\n") + return tail + + # ── internal helpers ─────────────────────────────────────────────── + + @staticmethod + def _find_first_tag( + buf: str, tags: Tuple[str, ...], + ) -> Tuple[int, int]: + """Return (earliest_index, tag_length) over *tags*, or (-1, 0). + + Case-insensitive match. + """ + buf_lower = buf.lower() + best_idx = -1 + best_len = 0 + for tag in tags: + idx = buf_lower.find(tag.lower()) + if idx != -1 and (best_idx == -1 or idx < best_idx): + best_idx = idx + best_len = len(tag) + return best_idx, best_len + + def _find_earliest_closed_pair(self, buf: str): + """Return (start_idx, end_idx) of the earliest closed pair, else None. + + A closed pair is ``...`` of any variant. Matches are + case-insensitive and non-greedy (the closest close tag after + an open tag wins), matching the regex ``.*?`` + semantics of ``_strip_think_blocks`` case 1. When two tag + variants could both match, the one whose open tag appears + earlier wins. + """ + buf_lower = buf.lower() + best: "tuple[int, int] | None" = None + for open_tag, close_tag in zip(self._OPEN_TAGS, self._CLOSE_TAGS): + open_lower = open_tag.lower() + close_lower = close_tag.lower() + open_idx = buf_lower.find(open_lower) + if open_idx == -1: + continue + close_idx = buf_lower.find( + close_lower, open_idx + len(open_lower), + ) + if close_idx == -1: + continue + end_idx = close_idx + len(close_lower) + if best is None or open_idx < best[0]: + best = (open_idx, end_idx) + return best + + def _find_open_at_boundary( + self, buf: str, already_emitted: list[str], + ) -> Tuple[int, int]: + """Return the earliest block-boundary open-tag (idx, len). + + Returns (-1, 0) if no boundary-legal opener is present. + """ + buf_lower = buf.lower() + best_idx = -1 + best_len = 0 + for tag in self._OPEN_TAGS: + tag_lower = tag.lower() + search_start = 0 + while True: + idx = buf_lower.find(tag_lower, search_start) + if idx == -1: + break + if self._is_block_boundary(buf, idx, already_emitted): + if best_idx == -1 or idx < best_idx: + best_idx = idx + best_len = len(tag) + break # first boundary hit for this tag is enough + search_start = idx + 1 + return best_idx, best_len + + def _is_block_boundary( + self, buf: str, idx: int, already_emitted: list[str], + ) -> bool: + """True iff position *idx* in *buf* is a block boundary. + + A block boundary is: + - buf position 0 AND the most recent emission ended with + a newline (or nothing has been emitted yet) + - any position whose preceding text on the current line + (since the last newline in buf) is whitespace-only, AND + if there is no newline in the preceding buf portion, the + most recent prior emission ended with a newline + """ + if idx == 0: + # Check whether the last already-emitted chunk in THIS + # feed() call ended with a newline, otherwise fall back + # to the cross-feed flag. + if already_emitted: + return already_emitted[-1].endswith("\n") + return self._last_emitted_ended_newline + preceding = buf[:idx] + last_nl = preceding.rfind("\n") + if last_nl == -1: + # No newline in buf before the tag — boundary only if the + # prior emission ended with a newline AND everything since + # is whitespace. + if already_emitted: + prior_newline = already_emitted[-1].endswith("\n") + else: + prior_newline = self._last_emitted_ended_newline + return prior_newline and preceding.strip() == "" + # Newline present — text between it and the tag must be + # whitespace-only. + return preceding[last_nl + 1:].strip() == "" + + @classmethod + def _max_partial_suffix( + cls, buf: str, tags: Tuple[str, ...], + ) -> int: + """Return the longest buf-suffix that is a prefix of any tag. + + Only prefixes strictly shorter than the tag itself count + (full-length suffixes are the tag and are handled as matches, + not held-back partials). Case-insensitive. + """ + if not buf: + return 0 + buf_lower = buf.lower() + max_check = min(len(buf_lower), cls._MAX_TAG_LEN - 1) + for i in range(max_check, 0, -1): + suffix = buf_lower[-i:] + for tag in tags: + tag_lower = tag.lower() + if len(tag_lower) > i and tag_lower.startswith(suffix): + return i + return 0 + + @classmethod + def _strip_orphan_close_tags(cls, text: str) -> str: + """Remove any close tags from *text* (orphan-close handling). + + An orphan close tag has no matching open in the current + scrubber state; it's always noise, stripped with any trailing + whitespace so the surrounding prose flows naturally. + """ + if " set: """Return plugin command names (without slash prefix) for dispatch matching.""" try: - from hermes_cli.plugins import get_plugin_manager - return set(get_plugin_manager()._plugin_commands.keys()) + from hermes_cli.plugins import get_plugin_commands + return set(get_plugin_commands().keys()) except Exception: return set() @@ -2145,7 +2145,10 @@ class HermesCLI: elif CLI_CONFIG.get("max_turns"): # Backwards compat: root-level max_turns self.max_turns = CLI_CONFIG["max_turns"] elif os.getenv("HERMES_MAX_ITERATIONS"): - self.max_turns = int(os.getenv("HERMES_MAX_ITERATIONS")) + try: + self.max_turns = int(os.getenv("HERMES_MAX_ITERATIONS", "")) + except (TypeError, ValueError): + self.max_turns = 90 else: self.max_turns = 90 @@ -7659,6 +7662,10 @@ class HermesCLI: ): self.session_id = self.agent.session_id self._pending_title = None + # Manual /compress replaces conversation_history with a new + # compressed handoff for the child session. Persist it from + # offset 0 so resume can recover the continuation after exit. + self.agent._flush_messages_to_session_db(self.conversation_history, None) new_tokens = estimate_request_tokens_rough( self.conversation_history, system_prompt=_sys_prompt, diff --git a/cron/scheduler.py b/cron/scheduler.py index c49370352c1a..c17c1fa46f85 100644 --- a/cron/scheduler.py +++ b/cron/scheduler.py @@ -114,12 +114,20 @@ from cron.jobs import get_due_jobs, mark_job_run, save_job_output, advance_next_ # locally for audit. SILENT_MARKER = "[SILENT]" -# Resolve Hermes home directory (respects HERMES_HOME override) -_hermes_home = get_hermes_home() +# Backward-compatible module override used by tests and emergency monkeypatches. +_hermes_home: Path | None = None -# File-based lock prevents concurrent ticks from gateway + daemon + systemd timer -_LOCK_DIR = _hermes_home / "cron" -_LOCK_FILE = _LOCK_DIR / ".tick.lock" + +def _get_hermes_home() -> Path: + """Resolve Hermes home dynamically while preserving test monkeypatch hooks.""" + return _hermes_home or get_hermes_home() + + +def _get_lock_paths() -> tuple[Path, Path]: + """Resolve cron lock paths at call time so profile/env changes are honored.""" + hermes_home = _get_hermes_home() + lock_dir = hermes_home / "cron" + return lock_dir, lock_dir / ".tick.lock" def _resolve_origin(job: dict) -> Optional[dict]: @@ -597,7 +605,7 @@ def _run_job_script(script_path: str) -> tuple[bool, str]: """ from hermes_constants import get_hermes_home - scripts_dir = get_hermes_home() / "scripts" + scripts_dir = _get_hermes_home() / "scripts" scripts_dir.mkdir(parents=True, exist_ok=True) scripts_dir_resolved = scripts_dir.resolve() @@ -1058,9 +1066,9 @@ def run_job(job: dict) -> tuple[bool, str, str, Optional[str]]: # changes take effect without a gateway restart. from dotenv import load_dotenv try: - load_dotenv(str(_hermes_home / ".env"), override=True, encoding="utf-8") + load_dotenv(str(_get_hermes_home() / ".env"), override=True, encoding="utf-8") except UnicodeDecodeError: - load_dotenv(str(_hermes_home / ".env"), override=True, encoding="latin-1") + load_dotenv(str(_get_hermes_home() / ".env"), override=True, encoding="latin-1") delivery_target = _resolve_delivery_target(job) if delivery_target: @@ -1078,7 +1086,7 @@ def run_job(job: dict) -> tuple[bool, str, str, Optional[str]]: _cfg = {} try: import yaml - _cfg_path = str(_hermes_home / "config.yaml") + _cfg_path = str(_get_hermes_home() / "config.yaml") if os.path.exists(_cfg_path): with open(_cfg_path) as _f: _cfg = yaml.safe_load(_f) or {} @@ -1112,7 +1120,7 @@ def run_job(job: dict) -> tuple[bool, str, str, Optional[str]]: if prefill_file: pfpath = Path(prefill_file).expanduser() if not pfpath.is_absolute(): - pfpath = _hermes_home / pfpath + pfpath = _get_hermes_home() / pfpath if pfpath.exists(): try: with open(pfpath, "r", encoding="utf-8") as _pf: @@ -1436,12 +1444,13 @@ def tick(verbose: bool = True, adapters=None, loop=None) -> int: Returns: Number of jobs executed (0 if another tick is already running) """ - _LOCK_DIR.mkdir(parents=True, exist_ok=True) + lock_dir, lock_file = _get_lock_paths() + lock_dir.mkdir(parents=True, exist_ok=True) # Cross-platform file locking: fcntl on Unix, msvcrt on Windows lock_fd = None try: - lock_fd = open(_LOCK_FILE, "w") + lock_fd = open(lock_file, "w") if fcntl: fcntl.flock(lock_fd, fcntl.LOCK_EX | fcntl.LOCK_NB) elif msvcrt: diff --git a/gateway/config.py b/gateway/config.py index fa64b9046d4d..2e0e3276b7b2 100644 --- a/gateway/config.py +++ b/gateway/config.py @@ -845,6 +845,16 @@ def load_gateway_config() -> GatewayConfig: ): if yaml_key in allow_mentions_cfg and not os.getenv(env_key): os.environ[env_key] = str(allow_mentions_cfg[yaml_key]).lower() + # reply_to_mode: top-level preferred, falls back to extra.reply_to_mode + # YAML 1.1 parses bare 'off' as boolean False — coerce to string "off". + _discord_extra = discord_cfg.get("extra") if isinstance(discord_cfg.get("extra"), dict) else {} + _discord_rtm = ( + discord_cfg["reply_to_mode"] if "reply_to_mode" in discord_cfg + else _discord_extra.get("reply_to_mode") + ) + if _discord_rtm is not None and not os.getenv("DISCORD_REPLY_TO_MODE"): + _rtm_str = "off" if _discord_rtm is False else str(_discord_rtm).lower() + os.environ["DISCORD_REPLY_TO_MODE"] = _rtm_str # Bridge top-level require_mention to Telegram when the telegram: section # does not already provide one. Users often write "require_mention: true" @@ -881,6 +891,16 @@ def load_gateway_config() -> GatewayConfig: os.environ["TELEGRAM_REACTIONS"] = str(telegram_cfg["reactions"]).lower() if "proxy_url" in telegram_cfg and not os.getenv("TELEGRAM_PROXY"): os.environ["TELEGRAM_PROXY"] = str(telegram_cfg["proxy_url"]).strip() + # reply_to_mode: top-level preferred, falls back to extra.reply_to_mode + # YAML 1.1 parses bare 'off' as boolean False — coerce to string "off". + _telegram_extra = telegram_cfg.get("extra") if isinstance(telegram_cfg.get("extra"), dict) else {} + _telegram_rtm = ( + telegram_cfg["reply_to_mode"] if "reply_to_mode" in telegram_cfg + else _telegram_extra.get("reply_to_mode") + ) + if _telegram_rtm is not None and not os.getenv("TELEGRAM_REPLY_TO_MODE"): + _rtm_str = "off" if _telegram_rtm is False else str(_telegram_rtm).lower() + os.environ["TELEGRAM_REPLY_TO_MODE"] = _rtm_str allowed_users = telegram_cfg.get("allow_from") if allowed_users is not None and not os.getenv("TELEGRAM_ALLOWED_USERS"): if isinstance(allowed_users, list): diff --git a/gateway/platforms/api_server.py b/gateway/platforms/api_server.py index 230859023b7d..b46075433161 100644 --- a/gateway/platforms/api_server.py +++ b/gateway/platforms/api_server.py @@ -2,8 +2,8 @@ OpenAI-compatible API server platform adapter. Exposes an HTTP server with endpoints: -- POST /v1/chat/completions — OpenAI Chat Completions format (stateless; opt-in session continuity via X-Hermes-Session-Id header) -- POST /v1/responses — OpenAI Responses API format (stateful via previous_response_id) +- POST /v1/chat/completions — OpenAI Chat Completions format (stateless; opt-in session continuity via X-Hermes-Session-Id header; opt-in long-term memory scoping via X-Hermes-Session-Key header) +- POST /v1/responses — OpenAI Responses API format (stateful via previous_response_id; X-Hermes-Session-Key supported) - GET /v1/responses/{response_id} — Retrieve a stored response - DELETE /v1/responses/{response_id} — Delete a stored response - GET /v1/models — lists hermes-agent as an available model @@ -698,6 +698,71 @@ class APIServerAdapter(BasePlatformAdapter): status=401, ) + # ------------------------------------------------------------------ + # Session header helpers + # ------------------------------------------------------------------ + + # Soft length cap for session identifiers. Headers are bounded in + # aggregate by aiohttp (``client_max_size`` / default 8 KiB per + # header), but we impose a tighter limit on the session headers so a + # caller can't burn memory by passing a multi-kilobyte "session key". + # 256 chars is well above any realistic stable channel identifier + # (e.g. ``agent:main:webui:dm:user-42``) while staying small enough + # that the sanitized form is safe to pass into Honcho / state.db. + _MAX_SESSION_HEADER_LEN = 256 + + def _parse_session_key_header( + self, request: "web.Request" + ) -> tuple[Optional[str], Optional["web.Response"]]: + """Extract and validate the ``X-Hermes-Session-Key`` header. + + The session key is a stable per-channel identifier that scopes + long-term memory (e.g. Honcho sessions) across transcripts. It + is independent of ``X-Hermes-Session-Id``: callers may send + either, both, or neither. + + Returns ``(session_key, None)`` on success (with an empty/absent + header yielding ``None`` for the key), or ``(None, error_response)`` + on validation failure. + + Security: like session continuation, accepting a caller-supplied + memory scope requires API-key authentication so that an + unauthenticated client on a local-only server can't inject itself + into another user's long-term memory scope by guessing a key. + """ + raw = request.headers.get("X-Hermes-Session-Key", "").strip() + if not raw: + return None, None + + if not self._api_key: + logger.warning( + "X-Hermes-Session-Key rejected: no API key configured. " + "Set API_SERVER_KEY to enable long-term memory scoping." + ) + return None, web.json_response( + _openai_error( + "X-Hermes-Session-Key requires API key authentication. " + "Configure API_SERVER_KEY to enable this feature." + ), + status=403, + ) + + # Reject control characters that could enable header injection on + # the echo path. + if re.search(r'[\r\n\x00]', raw): + return None, web.json_response( + {"error": {"message": "Invalid session key", "type": "invalid_request_error"}}, + status=400, + ) + + if len(raw) > self._MAX_SESSION_HEADER_LEN: + return None, web.json_response( + {"error": {"message": "Session key too long", "type": "invalid_request_error"}}, + status=400, + ) + + return raw, None + # ------------------------------------------------------------------ # Session DB helper # ------------------------------------------------------------------ @@ -728,6 +793,7 @@ class APIServerAdapter(BasePlatformAdapter): tool_progress_callback=None, tool_start_callback=None, tool_complete_callback=None, + gateway_session_key: Optional[str] = None, ) -> Any: """ Create an AIAgent instance using the gateway's runtime config. @@ -736,6 +802,13 @@ class APIServerAdapter(BasePlatformAdapter): base_url, etc. from config.yaml / env vars. Toolsets are resolved from config.yaml platform_toolsets.api_server (same as all other gateway platforms), falling back to the hermes-api-server default. + + ``gateway_session_key`` is a stable per-channel identifier supplied + by the client (via ``X-Hermes-Session-Key``). Unlike ``session_id`` + which scopes the short-term transcript and rotates on /new, this + key is meant to persist across transcripts so long-term memory + providers (e.g. Honcho) can scope their per-chat state correctly + — matching the semantics of the native gateway's ``session_key``. """ from run_agent import AIAgent from gateway.run import _resolve_runtime_agent_kwargs, _resolve_gateway_model, _load_gateway_config, GatewayRunner @@ -771,6 +844,7 @@ class APIServerAdapter(BasePlatformAdapter): session_db=self._ensure_session_db(), fallback_model=fallback_model, reasoning_config=reasoning_config, + gateway_session_key=gateway_session_key, ) return agent @@ -854,6 +928,7 @@ class APIServerAdapter(BasePlatformAdapter): "run_stop": True, "tool_progress_events": True, "session_continuity_header": "X-Hermes-Session-Id", + "session_key_header": "X-Hermes-Session-Key", "cors": bool(self._cors_origins), }, "endpoints": { @@ -925,6 +1000,15 @@ class APIServerAdapter(BasePlatformAdapter): status=400, ) + # Allow caller to scope long-term memory (e.g. Honcho) with a + # stable per-channel identifier via X-Hermes-Session-Key. This + # is independent of X-Hermes-Session-Id: the key persists across + # transcripts while the id rotates when the caller starts a new + # transcript (i.e. /new semantics). See _parse_session_key_header. + gateway_session_key, key_err = self._parse_session_key_header(request) + if key_err is not None: + return key_err + # Allow caller to continue an existing session by passing X-Hermes-Session-Id. # When provided, history is loaded from state.db instead of from the request body. # @@ -1059,11 +1143,13 @@ class APIServerAdapter(BasePlatformAdapter): tool_start_callback=_on_tool_start, tool_complete_callback=_on_tool_complete, agent_ref=agent_ref, + gateway_session_key=gateway_session_key, )) return await self._write_sse_chat_completion( request, completion_id, model_name, created, _stream_q, agent_task, agent_ref, session_id=session_id, + gateway_session_key=gateway_session_key, ) # Non-streaming: run the agent (with optional Idempotency-Key) @@ -1073,6 +1159,7 @@ class APIServerAdapter(BasePlatformAdapter): conversation_history=history, ephemeral_system_prompt=system_prompt, session_id=session_id, + gateway_session_key=gateway_session_key, ) idempotency_key = request.headers.get("Idempotency-Key") @@ -1122,11 +1209,17 @@ class APIServerAdapter(BasePlatformAdapter): }, } - return web.json_response(response_data, headers={"X-Hermes-Session-Id": session_id}) + response_headers = { + "X-Hermes-Session-Id": result.get("session_id", session_id), + } + if gateway_session_key: + response_headers["X-Hermes-Session-Key"] = gateway_session_key + return web.json_response(response_data, headers=response_headers) async def _write_sse_chat_completion( self, request: "web.Request", completion_id: str, model: str, created: int, stream_q, agent_task, agent_ref=None, session_id: str = None, + gateway_session_key: str = None, ) -> "web.StreamResponse": """Write real streaming SSE from agent's stream_delta_callback queue. @@ -1149,6 +1242,8 @@ class APIServerAdapter(BasePlatformAdapter): sse_headers.update(cors) if session_id: sse_headers["X-Hermes-Session-Id"] = session_id + if gateway_session_key: + sse_headers["X-Hermes-Session-Key"] = gateway_session_key response = web.StreamResponse(status=200, headers=sse_headers) await response.prepare(request) @@ -1272,6 +1367,7 @@ class APIServerAdapter(BasePlatformAdapter): conversation: Optional[str], store: bool, session_id: str, + gateway_session_key: Optional[str] = None, ) -> "web.StreamResponse": """Write an SSE stream for POST /v1/responses (OpenAI Responses API). @@ -1314,6 +1410,8 @@ class APIServerAdapter(BasePlatformAdapter): sse_headers.update(cors) if session_id: sse_headers["X-Hermes-Session-Id"] = session_id + if gateway_session_key: + sse_headers["X-Hermes-Session-Key"] = gateway_session_key response = web.StreamResponse(status=200, headers=sse_headers) await response.prepare(request) @@ -1763,6 +1861,11 @@ class APIServerAdapter(BasePlatformAdapter): if auth_err: return auth_err + # Long-term memory scope header (see chat_completions for details). + gateway_session_key, key_err = self._parse_session_key_header(request) + if key_err is not None: + return key_err + # Parse request body try: body = await request.json() @@ -1914,6 +2017,7 @@ class APIServerAdapter(BasePlatformAdapter): tool_start_callback=_on_tool_start, tool_complete_callback=_on_tool_complete, agent_ref=agent_ref, + gateway_session_key=gateway_session_key, )) response_id = f"resp_{uuid.uuid4().hex[:28]}" @@ -1934,6 +2038,7 @@ class APIServerAdapter(BasePlatformAdapter): conversation=conversation, store=store, session_id=session_id, + gateway_session_key=gateway_session_key, ) async def _compute_response(): @@ -1942,6 +2047,7 @@ class APIServerAdapter(BasePlatformAdapter): conversation_history=conversation_history, ephemeral_system_prompt=instructions, session_id=session_id, + gateway_session_key=gateway_session_key, ) idempotency_key = request.headers.get("Idempotency-Key") @@ -2016,7 +2122,10 @@ class APIServerAdapter(BasePlatformAdapter): if conversation: self._response_store.set_conversation(conversation, response_id) - return web.json_response(response_data) + response_headers = {"X-Hermes-Session-Id": session_id} + if gateway_session_key: + response_headers["X-Hermes-Session-Key"] = gateway_session_key + return web.json_response(response_data, headers=response_headers) # ------------------------------------------------------------------ # GET / DELETE response endpoints @@ -2338,6 +2447,7 @@ class APIServerAdapter(BasePlatformAdapter): tool_start_callback=None, tool_complete_callback=None, agent_ref: Optional[list] = None, + gateway_session_key: Optional[str] = None, ) -> tuple: """ Create an agent and run a conversation in a thread executor. @@ -2360,6 +2470,7 @@ class APIServerAdapter(BasePlatformAdapter): tool_progress_callback=tool_progress_callback, tool_start_callback=tool_start_callback, tool_complete_callback=tool_complete_callback, + gateway_session_key=gateway_session_key, ) if agent_ref is not None: agent_ref[0] = agent @@ -2374,6 +2485,12 @@ class APIServerAdapter(BasePlatformAdapter): "output_tokens": getattr(agent, "session_completion_tokens", 0) or 0, "total_tokens": getattr(agent, "session_total_tokens", 0) or 0, } + # Include the effective session ID in the result so callers + # (e.g. X-Hermes-Session-Id header) can track compression- + # triggered session rotations. (#16938) + _eff_sid = getattr(agent, "session_id", session_id) + if isinstance(_eff_sid, str) and _eff_sid: + result["session_id"] = _eff_sid return result, usage return await loop.run_in_executor(None, _run) @@ -2453,6 +2570,11 @@ class APIServerAdapter(BasePlatformAdapter): if auth_err: return auth_err + # Long-term memory scope header (see chat_completions for details). + gateway_session_key, key_err = self._parse_session_key_header(request) + if key_err is not None: + return key_err + # Enforce concurrency limit if len(self._run_streams) >= self._MAX_CONCURRENT_RUNS: return web.json_response( @@ -2561,6 +2683,7 @@ class APIServerAdapter(BasePlatformAdapter): session_id=session_id, stream_delta_callback=_text_cb, tool_progress_callback=event_cb, + gateway_session_key=gateway_session_key, ) self._active_run_agents[run_id] = agent def _run_sync(): @@ -2661,7 +2784,14 @@ class APIServerAdapter(BasePlatformAdapter): if hasattr(task, "add_done_callback"): task.add_done_callback(self._background_tasks.discard) - return web.json_response({"run_id": run_id, "status": "started"}, status=202) + response_headers = ( + {"X-Hermes-Session-Key": gateway_session_key} if gateway_session_key else {} + ) + return web.json_response( + {"run_id": run_id, "status": "started"}, + status=202, + headers=response_headers, + ) async def _handle_get_run(self, request: "web.Request") -> "web.Response": """GET /v1/runs/{run_id} — return pollable run status for external UIs.""" diff --git a/gateway/platforms/base.py b/gateway/platforms/base.py index 4d611fdaa536..5c2bbf96aa88 100644 --- a/gateway/platforms/base.py +++ b/gateway/platforms/base.py @@ -2675,10 +2675,18 @@ class BasePlatformAdapter(ABC): mode = os.getenv("HERMES_HUMAN_DELAY_MODE", "off").lower() if mode == "off": return 0.0 - min_ms = int(os.getenv("HERMES_HUMAN_DELAY_MIN_MS", "800")) - max_ms = int(os.getenv("HERMES_HUMAN_DELAY_MAX_MS", "2500")) if mode == "natural": min_ms, max_ms = 800, 2500 + return random.uniform(min_ms / 1000.0, max_ms / 1000.0) + # custom mode — tolerate malformed env vars instead of crashing. + try: + min_ms = int(os.getenv("HERMES_HUMAN_DELAY_MIN_MS", "800")) + except (TypeError, ValueError): + min_ms = 800 + try: + max_ms = int(os.getenv("HERMES_HUMAN_DELAY_MAX_MS", "2500")) + except (TypeError, ValueError): + max_ms = 2500 return random.uniform(min_ms / 1000.0, max_ms / 1000.0) async def _process_message_background(self, event: MessageEvent, session_key: str) -> None: diff --git a/gateway/platforms/feishu.py b/gateway/platforms/feishu.py index ac920bab69df..e1528b9bca09 100644 --- a/gateway/platforms/feishu.py +++ b/gateway/platforms/feishu.py @@ -153,6 +153,9 @@ _MARKDOWN_HINT_RE = re.compile( r"(^#{1,6}\s)|(^\s*[-*]\s)|(^\s*\d+\.\s)|(^\s*---+\s*$)|(```)|(`[^`\n]+`)|(\*\*[^*\n].+?\*\*)|(~~[^~\n].+?~~)|(.+?)|(\*[^*\n]+\*)|(\[[^\]]+\]\([^)]+\))|(^>\s)", re.MULTILINE, ) +# Detect markdown tables: a line starting with | followed by a separator line. +# Feishu post-type 'md' elements do not render tables, so we force text mode. +_MARKDOWN_TABLE_RE = re.compile(r"^\|.*\|\n\|[-|: ]+\|", re.MULTILINE) _MARKDOWN_LINK_RE = re.compile(r"\[([^\]]+)\]\(([^)]+)\)") _MARKDOWN_FENCE_OPEN_RE = re.compile(r"^```([^\n`]*)\s*$") _MARKDOWN_FENCE_CLOSE_RE = re.compile(r"^```\s*$") @@ -3862,47 +3865,50 @@ class FeishuAdapter(BasePlatformAdapter): and self-sent bot event filtering. Populates ``_bot_open_id`` and ``_bot_name`` from /open-apis/bot/v3/info - (no extra scopes required beyond the tenant access token). Falls back to - the application info endpoint for ``_bot_name`` only when the first probe - doesn't return it. Each field is hydrated independently — a value already - supplied via env vars (FEISHU_BOT_OPEN_ID / FEISHU_BOT_USER_ID / - FEISHU_BOT_NAME) is preserved and skips its probe. + (no extra scopes required beyond the tenant access token). The probe + always runs when a client is available so stale env vars from app/bot + migrations do not break group @mention gating. Falls back to the + application info endpoint for ``_bot_name`` only when the first probe + doesn't return it. If the probe fails, env-provided values are preserved. """ if not self._client: return - if self._bot_open_id and self._bot_name: - # Everything the self-send filter and precise mention gate need is - # already in place; nothing to probe. - return # Primary probe: /open-apis/bot/v3/info — returns bot_name + open_id, no # extra scopes required. This is the same endpoint the onboarding wizard # uses via probe_bot(). - if not self._bot_open_id or not self._bot_name: - try: - req = ( - BaseRequest.builder() - .http_method(HttpMethod.GET) - .uri("/open-apis/bot/v3/info") - .token_types({AccessTokenType.TENANT}) - .build() - ) - resp = await asyncio.to_thread(self._client.request, req) - content = getattr(getattr(resp, "raw", None), "content", None) - if content: - payload = json.loads(content) - parsed = _parse_bot_response(payload) or {} - open_id = (parsed.get("bot_open_id") or "").strip() - bot_name = (parsed.get("bot_name") or "").strip() - if open_id and not self._bot_open_id: - self._bot_open_id = open_id - if bot_name and not self._bot_name: - self._bot_name = bot_name - except Exception: - logger.debug( - "[Feishu] /bot/v3/info probe failed during hydration", - exc_info=True, - ) + try: + req = ( + BaseRequest.builder() + .http_method(HttpMethod.GET) + .uri("/open-apis/bot/v3/info") + .token_types({AccessTokenType.TENANT}) + .build() + ) + resp = await asyncio.to_thread(self._client.request, req) + content = getattr(getattr(resp, "raw", None), "content", None) + if content: + payload = json.loads(content) + parsed = _parse_bot_response(payload) or {} + open_id = (parsed.get("bot_open_id") or "").strip() + bot_name = (parsed.get("bot_name") or "").strip() + if open_id: + if self._bot_open_id and self._bot_open_id != open_id: + logger.warning( + "[Feishu] FEISHU_BOT_OPEN_ID is stale; using /bot/v3/info open_id for group @mention gating." + ) + self._bot_open_id = open_id + if bot_name: + if self._bot_name and self._bot_name != bot_name: + logger.info( + "[Feishu] FEISHU_BOT_NAME differs from /bot/v3/info; using hydrated bot name for group @mention gating." + ) + self._bot_name = bot_name + except Exception: + logger.debug( + "[Feishu] /bot/v3/info probe failed during hydration", + exc_info=True, + ) # Fallback probe for _bot_name only: application info endpoint. Needs # admin:app.info:readonly or application:application:self_manage scope, @@ -3947,7 +3953,14 @@ class FeishuAdapter(BasePlatformAdapter): if isinstance(seen_data, list): entries: Dict[str, float] = {str(item).strip(): 0.0 for item in seen_data if str(item).strip()} elif isinstance(seen_data, dict): - entries = {k: float(v) for k, v in seen_data.items() if isinstance(k, str) and k.strip()} + entries = {} + for key, value in seen_data.items(): + if not isinstance(key, str) or not key.strip(): + continue + try: + entries[key] = float(value) + except (TypeError, ValueError): + continue else: return # Filter out TTL-expired entries (entries saved with ts=0.0 are treated as immortal @@ -3992,6 +4005,12 @@ class FeishuAdapter(BasePlatformAdapter): # ========================================================================= def _build_outbound_payload(self, content: str) -> tuple[str, str]: + # Feishu post-type 'md' elements do not render markdown tables; sending + # table content as post causes the message to appear blank on the client. + # Force plain text for anything that looks like a markdown table. + if _MARKDOWN_TABLE_RE.search(content): + text_payload = {"text": content} + return "text", json.dumps(text_payload, ensure_ascii=False) if _MARKDOWN_HINT_RE.search(content): return "post", _build_markdown_post_payload(content) text_payload = {"text": content} @@ -4087,7 +4106,15 @@ class FeishuAdapter(BasePlatformAdapter): content=payload, uuid_value=str(uuid.uuid4()), ) - request = self._build_create_message_request("chat_id", body) + # Detect whether chat_id is a user open_id (DM) or a chat_id (group). + # Feishu API expects receive_id_type="open_id" for user DMs (ou_ prefix) + # and receive_id_type="chat_id" for group chats (oc_ prefix, which IS + # the chat_id format — see https://open.feishu.cn/document/). + if chat_id.startswith("ou_"): + receive_id_type = "open_id" + else: + receive_id_type = "chat_id" + request = self._build_create_message_request(receive_id_type, body) return await asyncio.to_thread(self._client.im.v1.message.create, request) @staticmethod diff --git a/gateway/platforms/helpers.py b/gateway/platforms/helpers.py index 8db7af7eba90..673beeac9b45 100644 --- a/gateway/platforms/helpers.py +++ b/gateway/platforms/helpers.py @@ -222,33 +222,37 @@ class ThreadParticipationTracker: def __init__(self, platform_name: str, max_tracked: int = 500): self._platform = platform_name self._max_tracked = max_tracked - self._threads: set = self._load() + self._threads: dict[str, None] = { + str(thread_id): None for thread_id in self._load() + } def _state_path(self) -> Path: from hermes_constants import get_hermes_home return get_hermes_home() / f"{self._platform}_threads.json" - def _load(self) -> set: + def _load(self) -> list[str]: path = self._state_path() if path.exists(): try: - return set(json.loads(path.read_text(encoding="utf-8"))) + data = json.loads(path.read_text(encoding="utf-8")) + if isinstance(data, list): + return [str(thread_id) for thread_id in data] except Exception: pass - return set() + return [] def _save(self) -> None: path = self._state_path() thread_list = list(self._threads) if len(thread_list) > self._max_tracked: thread_list = thread_list[-self._max_tracked:] - self._threads = set(thread_list) + self._threads = {thread_id: None for thread_id in thread_list} atomic_json_write(path, thread_list, indent=None) def mark(self, thread_id: str) -> None: """Mark *thread_id* as participated and persist.""" if thread_id not in self._threads: - self._threads.add(thread_id) + self._threads[thread_id] = None self._save() def __contains__(self, thread_id: str) -> bool: diff --git a/gateway/platforms/telegram_network.py b/gateway/platforms/telegram_network.py index b099adc50e05..8fe4c2809345 100644 --- a/gateway/platforms/telegram_network.py +++ b/gateway/platforms/telegram_network.py @@ -185,10 +185,13 @@ async def _query_doh_provider( async def discover_fallback_ips() -> list[str]: """Auto-discover Telegram API IPs via DNS-over-HTTPS. - Resolves api.telegram.org through Google and Cloudflare DoH, collects all - unique IPs, and excludes the system-DNS-resolved IP (which is presumably - unreachable on this network). Falls back to a hardcoded seed list when DoH - is also unavailable. + Resolves api.telegram.org through Google and Cloudflare DoH and returns all + unique A records. IPs that match the local system resolver are kept rather + than excluded: in many networks the system-DNS IP is the most reliable path + to api.telegram.org and a transient primary-path failure should be retried + against the same address via the IP-rewrite path before the seed list is + consulted (#14520). Falls back to a hardcoded seed list only when DoH + yields no usable answers. """ async with httpx.AsyncClient(timeout=httpx.Timeout(_DOH_TIMEOUT)) as client: doh_tasks = [_query_doh_provider(client, p) for p in _DOH_PROVIDERS] @@ -203,11 +206,11 @@ async def discover_fallback_ips() -> list[str]: if isinstance(r, list): doh_ips.extend(r) - # Deduplicate preserving order, exclude system-DNS IPs + # Deduplicate preserving order seen: set[str] = set() candidates: list[str] = [] for ip in doh_ips: - if ip not in seen and ip not in system_ips: + if ip not in seen: seen.add(ip) candidates.append(ip) @@ -219,7 +222,7 @@ async def discover_fallback_ips() -> list[str]: return validated logger.info( - "DoH discovery yielded no new IPs (system DNS: %s); using seed fallback IPs %s", + "DoH discovery yielded no usable IPs (system DNS: %s); using seed fallback IPs %s", ", ".join(system_ips) or "unknown", ", ".join(_SEED_FALLBACK_IPS), ) diff --git a/gateway/platforms/wecom.py b/gateway/platforms/wecom.py index 873284de7961..c93a8fe3d65b 100644 --- a/gateway/platforms/wecom.py +++ b/gateway/platforms/wecom.py @@ -1015,6 +1015,8 @@ class WeComAdapter(BasePlatformAdapter): if not aes_key: raise ValueError("aes_key is required") + # WeCom doesn't pad base64 keys; add padding if needed + aes_key = aes_key + '=' * ((4 - len(aes_key) % 4) % 4) key = base64.b64decode(aes_key) if len(key) != 32: raise ValueError(f"Invalid WeCom AES key length: expected 32 bytes, got {len(key)}") diff --git a/gateway/run.py b/gateway/run.py index ebfd2731fe18..ed3bd47b96b3 100644 --- a/gateway/run.py +++ b/gateway/run.py @@ -39,6 +39,7 @@ from typing import Dict, Optional, Any, List, Union # gateway is a long-running daemon, so its boot cost matters less than # preserving the established test-patch surface. from agent.account_usage import fetch_account_usage, render_account_usage_lines +from agent.i18n import t from hermes_cli.config import cfg_get # --- Agent cache tuning --------------------------------------------------- @@ -93,153 +94,6 @@ def _telegramize_command_mentions(text: str, platform: Any) -> str: _AUTO_CONTINUE_FRESHNESS_SECS_DEFAULT = 60 * 60 -# --- Stale-code self-check ------------------------------------------------ -# Long-running gateway processes that survive an ``hermes update`` keep the -# old ``hermes_cli.config`` (and friends) cached in ``sys.modules``. When -# the updated tool files on disk then try to ``from hermes_cli.config -# import cfg_get`` (added in PR #17304), the import resolves against the -# already-loaded stale module object and raises ``ImportError`` — see -# Issue #17648. Rather than papering over the import failure site-by-site -# in every tool file, detect the stale state centrally and auto-restart -# so the gateway reloads with fresh code. -# -# The signal we use is ``git rev-parse HEAD`` — the only thing ``hermes -# update`` moves that is NOT moved by agent-driven file edits. Earlier -# revisions of this check compared file mtimes across a sentinel set -# (run_agent.py, gateway/run.py, ...), but that produced false positives -# whenever the agent edited its own source files during a session: -# mtime jumps, stale-check fires, gateway restarts, user must retype. -# See the conversation at PR # for the motivating incident. -# -# The legacy mtime sentinels are kept ONLY as a last-resort fallback for -# non-git installs (pip install from wheel, sparse clones with no .git -# dir). In those environments ``hermes update`` is not a supported path, -# so the check effectively no-ops — which is the safe behavior: better -# to ship one broken import than to restart on every agent-edit. -_STALE_CODE_SENTINELS: tuple[str, ...] = ( - "hermes_cli/config.py", - "hermes_cli/__init__.py", - "run_agent.py", - "gateway/run.py", - "pyproject.toml", -) - -# Cache git HEAD reads across consecutive messages so a chat burst doesn't -# spawn one subprocess per message. 5s is long enough to collapse a burst -# and short enough that the real post-update detection still fires within -# the user's perceived "next message" window. -_GIT_SHA_CACHE_TTL_SECS = 5.0 - - -def _read_git_head_sha(repo_root: Path) -> Optional[str]: - """Return the git HEAD SHA for ``repo_root``, or None if unavailable. - - Reads ``.git/HEAD`` directly (and follows one level of ref) instead - of shelling out to ``git`` — cheaper, no subprocess tax, works on - gateway hosts that don't have a ``git`` binary on PATH. Returns - None for non-git installs (no ``.git`` dir) or any I/O error; callers - treat None as "can't tell" and skip the check. - - Supports the three layouts we care about: - 1. Main checkout: ``/.git/`` is a directory. - 2. Git worktree: ``/.git`` is a file ``gitdir: `` that - points at ``
/.git/worktrees//``. The worktree's - gitdir has HEAD + index but NOT refs/heads/ — those live in - the main checkout, and ``/commondir`` points - at the main ``.git``. We search both locations for refs. - 3. Packed refs: ``refs/heads/`` is absent on disk but - listed in ``/packed-refs``. - """ - try: - git_dir = repo_root / ".git" - # Worktrees store ``.git`` as a file pointing at gitdir: - if git_dir.is_file(): - try: - content = git_dir.read_text().strip() - if content.startswith("gitdir:"): - git_dir = Path(content.split(":", 1)[1].strip()) - if not git_dir.is_absolute(): - git_dir = (repo_root / git_dir).resolve() - except OSError: - return None - if not git_dir.is_dir(): - return None - - # Figure out the "common" git dir — the one that owns shared refs. - # For a worktree, commondir points at it (relative path, resolve - # against git_dir). For a main checkout, common_dir == git_dir. - common_dir = git_dir - commondir_file = git_dir / "commondir" - if commondir_file.is_file(): - try: - rel = commondir_file.read_text().strip() - candidate = (git_dir / rel).resolve() if rel else git_dir - if candidate.is_dir(): - common_dir = candidate - except OSError: - pass - - head_path = git_dir / "HEAD" - if not head_path.is_file(): - return None - head_content = head_path.read_text().strip() - - if head_content.startswith("ref:"): - # Symbolic ref — follow one level (e.g. ref: refs/heads/main). - # Worktree-local refs (bisect, rebase-merge state) live under - # git_dir; shared refs (refs/heads/*, refs/tags/*) live under - # common_dir. Try git_dir first, then common_dir. - ref_rel = head_content.split(":", 1)[1].strip() - for base in (git_dir, common_dir) if git_dir != common_dir else (git_dir,): - ref_path = base / ref_rel - if ref_path.is_file(): - try: - sha = ref_path.read_text().strip() - except OSError: - continue - if sha: - return sha - # Packed refs fallback — always stored in the common dir. - packed = common_dir / "packed-refs" - if packed.is_file(): - try: - for line in packed.read_text().splitlines(): - line = line.strip() - if not line or line.startswith("#") or line.startswith("^"): - continue - parts = line.split(None, 1) - if len(parts) == 2 and parts[1] == ref_rel: - return parts[0] or None - except OSError: - return None - return None - - # Detached HEAD — content is the SHA directly. - return head_content or None - except Exception: - return None - - -def _compute_repo_mtime(repo_root: Path) -> float: - """Return the newest mtime across the stale-code sentinel files. - - Legacy fallback used only for non-git installs (``.git`` missing). - Missing files are ignored (they may not exist on older checkouts). - Returns 0.0 if no sentinel file is readable — treat that as "can't - tell", which downstream callers interpret as "not stale" to avoid - false-positive restart loops. - """ - newest = 0.0 - for rel in _STALE_CODE_SENTINELS: - try: - st = (repo_root / rel).stat() - except (OSError, FileNotFoundError): - continue - if st.st_mtime > newest: - newest = st.st_mtime - return newest - - def _coerce_gateway_timestamp(value: Any) -> Optional[float]: """Best-effort conversion of stored gateway timestamps to epoch seconds. @@ -1107,13 +961,6 @@ class GatewayRunner: _stop_task: Optional[asyncio.Task] = None _session_model_overrides: Dict[str, Dict[str, str]] = {} _session_reasoning_overrides: Dict[str, Dict[str, Any]] = {} - # Stale-code self-check defaults (see _detect_stale_code()). Class-level - # so tests that construct GatewayRunner via ``object.__new__`` without - # running __init__ don't crash when _handle_message reads these. - _boot_wall_time: float = 0.0 - _boot_repo_mtime: float = 0.0 - _boot_git_sha: Optional[str] = None - _stale_code_restart_triggered: bool = False def __init__(self, config: Optional[GatewayConfig] = None): global _gateway_runner_ref @@ -1122,30 +969,6 @@ class GatewayRunner: self._warn_if_docker_media_delivery_is_risky() _gateway_runner_ref = _weakref.ref(self) - # Boot-time snapshot used by the stale-code self-check. Captured - # before any work happens so post-update file writes are guaranteed - # to have newer mtimes. See _detect_stale_code() / Issue #17648. - try: - self._boot_wall_time: float = time.time() - self._repo_root_for_staleness: Path = Path(__file__).resolve().parent.parent - self._boot_git_sha: Optional[str] = _read_git_head_sha( - self._repo_root_for_staleness, - ) - self._boot_repo_mtime: float = _compute_repo_mtime( - self._repo_root_for_staleness, - ) - except Exception: - self._boot_wall_time = 0.0 - self._repo_root_for_staleness = Path(".") - self._boot_git_sha = None - self._boot_repo_mtime = 0.0 - self._stale_code_notified: set[str] = set() - self._stale_code_restart_triggered: bool = False - # Cached current-SHA read, refreshed at most every - # _GIT_SHA_CACHE_TTL_SECS so bursty chats don't hammer the filesystem. - self._cached_current_sha: Optional[str] = self._boot_git_sha - self._cached_current_sha_at: float = self._boot_wall_time - # Load ephemeral config from config.yaml / env vars. # Both are injected at API-call time only and never persisted. self._prefill_messages = self._load_prefill_messages() @@ -2853,101 +2676,6 @@ class GatewayRunner: task.add_done_callback(self._background_tasks.discard) return True - def _current_git_sha_cached(self) -> Optional[str]: - """Return the current HEAD SHA, cached for _GIT_SHA_CACHE_TTL_SECS. - - A bursty chat (user mashes "hello?" three times) would otherwise - re-read ``.git/HEAD`` on every message. Caching collapses that - into a single read and still re-checks within the user's - perceived "next message" window. - """ - now = time.time() - if ( - self._cached_current_sha is not None - and (now - self._cached_current_sha_at) < _GIT_SHA_CACHE_TTL_SECS - ): - return self._cached_current_sha - try: - sha = _read_git_head_sha(self._repo_root_for_staleness) - except Exception: - sha = None - self._cached_current_sha = sha - self._cached_current_sha_at = now - return sha - - def _detect_stale_code(self) -> bool: - """Return True if the git HEAD moved since this process booted. - - A gateway that survives ``hermes update`` (manual SIGTERM never - escalated, systemd restart race, detached-process respawn failed, - etc.) keeps pre-update modules cached in ``sys.modules``. Later - imports of names added post-update — e.g. ``cfg_get`` from PR - #17304 — raise ImportError against the stale module object (see - Issue #17648). - - We compare the git HEAD SHA at boot to the current SHA on disk. - ``hermes update`` always moves HEAD forward via ``git pull``; - agent file edits (the agent patching ``run_agent.py`` or - ``gateway/run.py`` during a self-dev session) never move HEAD. - That makes SHA comparison free of the false-positive class that - the old mtime check suffered from — the agent can edit any file - without triggering a phantom restart. - - Returns False when: - - the boot SHA is unavailable (non-git install, first call - during partial init, etc.); we can't tell and refuse to loop - - the current SHA matches the boot SHA - - reading the current SHA fails for any reason - """ - if not self._boot_wall_time: - return False - if not self._boot_git_sha: - # Non-git install. ``hermes update`` is git-based, so a - # non-git install can't experience the stale-modules class - # this check exists to catch. Return False — no check, no - # false positives. (If we ever ship a pip-install update - # path, we'd add a persistent update marker here and compare - # its timestamp to self._boot_wall_time.) - return False - try: - current = self._current_git_sha_cached() - except Exception: - return False - if not current: - return False - return current != self._boot_git_sha - - def _trigger_stale_code_restart(self) -> None: - """Idempotently kick off a graceful restart after stale-code detection. - - Runs at most once per process. The restart request goes through - the normal drain path so in-flight agent turns finish before the - process exits; the service manager (systemd / launchd / detached - profile watcher) then respawns with fresh code. On manual - ``hermes gateway run`` installs without a supervisor, the - process exits and the user must restart by hand — but they get a - user-visible message telling them so. - """ - if self._stale_code_restart_triggered: - return - self._stale_code_restart_triggered = True - current_sha = None - try: - current_sha = self._current_git_sha_cached() - except Exception: - pass - logger.warning( - "Stale-code self-check: git HEAD moved since gateway boot " - "(boot=%s, current=%s) — requesting graceful restart. " - "See Issue #17648.", - (self._boot_git_sha or "?")[:12], - (current_sha or "?")[:12], - ) - try: - self.request_restart(detached=False, via_service=True) - except Exception as exc: - logger.error("Stale-code restart request failed: %s", exc) - async def start(self) -> bool: """ Start the gateway and all configured platform adapters. @@ -3906,7 +3634,17 @@ class GatewayRunner: return out def _ready_nonempty() -> bool: - """Cheap probe: is there a ready+assigned+unclaimed task on ANY board?""" + """Cheap probe: is there at least one ready+assigned+unclaimed + task on ANY board whose assignee maps to a real Hermes profile + (i.e. one the dispatcher would actually spawn for)? + + Tasks assigned to control-plane lanes (e.g. ``orion-cc``, + ``orion-research``) are pulled by terminals via + ``claim_task`` directly and never spawnable, so a queue full + of those is "correctly idle", not "stuck". Filtering them out + here keeps the stuck-warn fire only on real failures (broken + PATH, missing venv, credential loss for a real Hermes profile). + """ try: boards = _kb.list_boards(include_archived=False) except Exception: @@ -3916,12 +3654,7 @@ class GatewayRunner: conn = None try: conn = _kb.connect(board=slug) - row = conn.execute( - "SELECT 1 FROM tasks " - "WHERE status = 'ready' AND assignee IS NOT NULL " - " AND claim_lock IS NULL LIMIT 1" - ).fetchone() - if row is not None: + if _kb.has_spawnable_ready(conn): return True except Exception: continue @@ -4873,27 +4606,6 @@ class GatewayRunner: """ source = event.source - # Stale-code self-check (Issue #17648). A gateway that survives - # ``hermes update`` keeps old modules cached in sys.modules; the - # first inbound message is our earliest safe chance to detect - # this and restart gracefully before we dispatch to the agent - # and hit ImportError on freshly-added names (e.g. cfg_get). - # Idempotent — runs the real check at most once per message, and - # request_restart() no-ops after the first call. - try: - if self._detect_stale_code(): - self._trigger_stale_code_restart() - # Acknowledge to the user so they don't see a silent - # drop; the gateway will be back up in a moment via the - # service manager / profile-watcher respawn. - return ( - "⟳ Gateway code was updated in the background — " - "restarting this gateway so your next message runs " - "on the new code. Please retry in a moment." - ) - except Exception as _stale_exc: - logger.debug("Stale-code self-check failed: %s", _stale_exc) - # Internal events (e.g. background-process completion notifications) # are system-generated and must skip user authorization. is_internal = bool(getattr(event, "internal", False)) @@ -5020,10 +4732,12 @@ class GatewayRunner: response_text = raw if response_text: response_path = _hermes_home / ".update_response" + prompt_path = _hermes_home / ".update_prompt.json" try: tmp = response_path.with_suffix(".tmp") tmp.write_text(response_text) tmp.replace(response_path) + prompt_path.unlink(missing_ok=True) except OSError as e: logger.warning("Failed to write update response: %s", e) return f"✗ Failed to send response to update process: {e}" @@ -5038,10 +4752,12 @@ class GatewayRunner: # The slash command then falls through to normal dispatch. if _recognized_cmd: response_path = _hermes_home / ".update_response" + prompt_path = _hermes_home / ".update_prompt.json" try: tmp = response_path.with_suffix(".tmp") tmp.write_text("") tmp.replace(response_path) + prompt_path.unlink(missing_ok=True) logger.info( "Recognized /%s during pending update prompt for %s; " "cancelled prompt with default and dispatching command", @@ -7662,7 +7378,7 @@ class GatewayRunner: if self._restart_requested or self._draining: count = self._running_agent_count() if count: - return f"⏳ Draining {count} active agent(s) before restart..." + return t("gateway.draining", count=count) return EphemeralReply("⏳ Gateway restart already in progress...") # Save the requester's routing info so the new gateway process can @@ -7714,7 +7430,7 @@ class GatewayRunner: else: self.request_restart(detached=True, via_service=False) if active_agents: - return f"⏳ Draining {active_agents} active agent(s) before restart..." + return t("gateway.draining", count=active_agents) return EphemeralReply("♻ Restarting gateway. If you aren't notified within 60 seconds, restart from the console with `hermes gateway restart`.") def _is_stale_restart_redelivery(self, event: MessageEvent) -> bool: @@ -7862,6 +7578,7 @@ class GatewayRunner: from hermes_cli.model_switch import ( switch_model as _switch_model, parse_model_flags, list_authenticated_providers, + list_picker_providers, ) from hermes_cli.providers import get_label @@ -7916,7 +7633,7 @@ class GatewayRunner: if has_picker: try: - providers = list_authenticated_providers( + providers = list_picker_providers( current_provider=current_provider, current_base_url=current_base_url, current_model=current_model, @@ -8384,7 +8101,7 @@ class GatewayRunner: if lower in ("clear", "stop", "done"): had = mgr.has_goal() mgr.clear() - return "✓ Goal cleared." if had else "No active goal." + return t("gateway.goal_cleared") if had else t("gateway.no_active_goal") # Otherwise — treat the remaining text as the new goal. try: @@ -9602,7 +9319,7 @@ class GatewayRunner: try: user_config: dict = _load_gateway_config() except Exception as e: - return f"⚠️ Could not read config.yaml: {e}" + return t("gateway.config_read_failed", error=e) effective = resolve_footer_config(user_config, platform_key) @@ -9635,7 +9352,7 @@ class GatewayRunner: atomic_yaml_write(config_path, user_config) except Exception as e: logger.warning("Failed to save runtime_footer.enabled: %s", e) - return f"⚠️ Could not save config: {e}" + return t("gateway.config_save_failed", error=e) state = "ON" if new_state else "OFF" example = "" @@ -11073,7 +10790,7 @@ class GatewayRunner: if not has_blocking_approval(session_key): if session_key in self._pending_approvals: self._pending_approvals.pop(session_key) - return "⚠️ Approval expired (agent is no longer waiting). Ask the agent to try again." + return t("gateway.approval_expired") return "No pending command to approve." # Parse args: support "all", "all session", "all always", "session", "always" @@ -11488,12 +11205,13 @@ class GatewayRunner: f"or type your answer directly.", metadata=metadata, ) + # Keep the prompt marker on disk until the user + # answers. If the gateway restarts mid-prompt, the + # next watcher can recover by re-forwarding it from + # disk. Duplicate sends in the same process are + # still suppressed by _update_prompt_pending. self._update_prompt_pending[session_key] = True - # Remove the prompt file so it isn't re-read on the - # next poll cycle. The update process only needs # .update_response to continue — it doesn't re-check - # .update_prompt.json while waiting. - prompt_path.unlink(missing_ok=True) logger.info("Forwarded update prompt to %s: %s", session_key, prompt_text[:80]) except (json.JSONDecodeError, OSError) as e: logger.debug("Failed to read update prompt: %s", e) diff --git a/gateway/session.py b/gateway/session.py index 16de296e0e7f..be393e48e6fc 100644 --- a/gateway/session.py +++ b/gateway/session.py @@ -1276,8 +1276,9 @@ class SessionStore: # Also write legacy JSONL (keeps existing tooling working during transition) transcript_path = self.get_transcript_path(session_id) - with open(transcript_path, "a", encoding="utf-8") as f: - f.write(json.dumps(message, ensure_ascii=False) + "\n") + with self._lock: + with open(transcript_path, "a", encoding="utf-8") as f: + f.write(json.dumps(message, ensure_ascii=False) + "\n") def rewrite_transcript(self, session_id: str, messages: List[Dict[str, Any]]) -> None: """Replace the entire transcript for a session with new messages. diff --git a/hermes_cli/claw.py b/hermes_cli/claw.py index f6e2521eb01f..5f9d728252dd 100644 --- a/hermes_cli/claw.py +++ b/hermes_cli/claw.py @@ -235,6 +235,9 @@ def _scan_workspace_state(source_dir: Path) -> list[tuple[Path, str]]: """ findings: list[tuple[Path, str]] = [] + if not source_dir.exists(): + return findings + # Direct state files in the root for name in ("todo.json", "sessions", "logs"): candidate = source_dir / name @@ -243,7 +246,12 @@ def _scan_workspace_state(source_dir: Path) -> list[tuple[Path, str]]: findings.append((candidate, f"Root {kind}: {name}")) # State files inside workspace directories - for child in sorted(source_dir.iterdir()): + try: + children = sorted(source_dir.iterdir()) + except OSError: + return findings + + for child in children: if not child.is_dir() or child.name.startswith("."): continue # Check for workspace-like subdirectories diff --git a/hermes_cli/config.py b/hermes_cli/config.py index 425eb2db0787..aa978b60dc58 100644 --- a/hermes_cli/config.py +++ b/hermes_cli/config.py @@ -781,6 +781,11 @@ DEFAULT_CONFIG = { "inline_diffs": True, # Show inline diff previews for write actions (write_file, patch, skill_manage) "show_cost": False, # Show $ cost in the status bar (off by default) "skin": "default", + # UI language for static user-facing messages (approval prompts, a + # handful of gateway slash-command replies). Does NOT affect agent + # responses, log lines, tool outputs, or slash-command descriptions. + # Supported: en, zh, ja, de, es. Unknown values fall back to en. + "language": "en", # TUI busy indicator style: kaomoji (default), emoji, unicode (braille # spinner), or ascii. Live-swappable via `/indicator