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

Merge simp/r3-06-D into simp/r3-06

Browse Source
This commit is contained in:
Teknium
2026-09-02 20:08:12 -07:00
parent 24aeb15600 5e0aa62754
commit 85155cc0b7
12 changed files with 499 additions and 1135 deletions
Download Patch File Download Diff File Expand all files Collapse all files
agent/lsp/workspace.py
+36 -62
View File
@@ -1,11 +1,10 @@
"""Workspace and project-root resolution for LSP.
1. **Workspace gate** — LSP only runs when the cwd (or the edited file) sits
inside a git worktree. Files outside any git root never trigger LSP,
which keeps gateway users on user-home cwd's from spawning daemons.
2. **nearest_root** — the per-server project-root walk: up from a start path
looking for marker files (``pyproject.toml``, ``Cargo.toml``, ...),
optionally bailing if an exclude marker shows up first.
1. **Workspace gate** — LSP only runs when the cwd (or the edited file) sits inside a git
worktree, so gateway users on user-home cwd's never spawn daemons.
2. **nearest_root** — the per-server project-root walk: up from a start path looking for marker
files (``pyproject.toml``, ``Cargo.toml``, ...), optionally bailing if an exclude marker
shows up first.
"""
from __future__ import annotations
@@ -16,8 +15,7 @@ from typing import Iterable, Iterator, Optional, Tuple
logger = logging.getLogger("agent.lsp.workspace")
# Cache: start dir → (worktree_root, is_git) so repeated calls don't re-stat.
# Cleared on shutdown.
# Cache: start dir → (worktree_root, is_git) so repeated calls don't re-stat. Cleared on shutdown.
_workspace_cache: dict = {}
# Walk cap: the deepest reasonable monorepo is well under 64 levels; bounds a
@@ -26,11 +24,8 @@ _MAX_WALK = 64
def normalize_path(path: str) -> str:
"""Expand ``~``, make absolute, collapse ``.``/``..``.
Symlinks are deliberately NOT resolved — some servers (rust-analyzer's
Cargo workspace identity) care, and we want the path the user typed.
"""
"""Expand ``~``, make absolute, collapse ``.``/``..``. Symlinks are deliberately NOT resolved —
some servers (rust-analyzer's Cargo workspace identity) care, and we want the path the user typed."""
return os.path.abspath(os.path.expanduser(path))
@@ -62,11 +57,9 @@ def find_git_worktree(start: str) -> Optional[str]:
start_path = _start_dir(start)
if start_path is None:
return None
cached = _workspace_cache.get(str(start_path))
if cached is not None:
return cached[0]
for cur in _walk_up(start_path):
try:
if (cur / ".git").exists():
@@ -74,30 +67,24 @@ def find_git_worktree(start: str) -> Optional[str]:
_workspace_cache[str(start_path)] = (resolved, True)
return resolved
except OSError:
# Permission error on a parent dir — bail out cleanly.
break
break # permission error on a parent dir — bail out cleanly
_workspace_cache[str(start_path)] = (None, False)
return None
def is_inside_workspace(path: str, workspace_root: str) -> bool:
"""True iff ``path`` is inside (or equal to) ``workspace_root``.
Symlinks are not resolved: a symlink pointing outside still counts as
outside, matching servers that reject didOpen for unrelated files.
"""
"""True iff ``path`` is inside (or equal to) ``workspace_root``. Symlinks are not resolved: a
symlink pointing outside still counts as outside, matching servers that reject didOpen for
unrelated files."""
p = normalize_path(path)
root = normalize_path(workspace_root)
if p == root:
return True
# commonpath handles case-insensitive filesystems on macOS/Windows.
try:
common = os.path.commonpath([p, root])
return os.path.commonpath([p, root]) == root
except ValueError:
# Different drives on Windows.
return False
return common == root
return False # different drives on Windows
def nearest_root(
@@ -109,51 +96,42 @@ def nearest_root(
) -> Optional[str]:
"""Walk up from ``start`` for the directory containing the first matched marker.
Returns ``None`` past ``ceiling`` (or the filesystem root), or when an
exclude marker is found first — the server is gated off for that file
(e.g. typescript skips deno projects when ``deno.json`` precedes
``package.json``). Marker names are exact filenames — no globs.
Returns ``None`` past ``ceiling`` (or the filesystem root), or when an exclude marker is found
first — the server is gated off for that file (e.g. typescript skips deno projects when
``deno.json`` precedes ``package.json``). Marker names are exact filenames — no globs.
"""
start_path = _start_dir(start)
if start_path is None:
return None
ceiling_path = Path(normalize_path(ceiling)) if ceiling else None
markers_list = list(markers)
excludes_list = list(excludes) if excludes else []
def present(cur: Path, names: list) -> bool:
for name in names:
try:
if (cur / name).exists():
return True
except OSError:
continue
return False
for cur in _walk_up(start_path):
# Excludes are checked before markers at each level.
for exc in excludes_list:
try:
if (cur / exc).exists():
return None
except OSError:
continue
for marker in markers_list:
try:
if (cur / marker).exists():
return str(cur)
except OSError:
continue
if present(cur, excludes_list):
return None
if present(cur, markers_list):
return str(cur)
if ceiling_path is not None and cur == ceiling_path:
return None
return None
def resolve_workspace_for_file(
file_path: str,
*,
cwd: Optional[str] = None,
) -> Tuple[Optional[str], bool]:
"""Return ``(workspace_root, gated_in)`` for a file.
The cwd's worktree wins when the file is inside it; otherwise the file's
own worktree is the fallback anchor (monorepos / unrelated checkouts).
``(None, False)`` when neither is in a git worktree.
"""
cwd = cwd or os.getcwd()
cwd_root = find_git_worktree(cwd)
def resolve_workspace_for_file(file_path: str, *, cwd: Optional[str] = None) -> Tuple[Optional[str], bool]:
"""Return ``(workspace_root, gated_in)`` for a file. The cwd's worktree wins when the file is
inside it; otherwise the file's own worktree is the fallback anchor (monorepos / unrelated
checkouts). ``(None, False)`` when neither is in a git worktree."""
cwd_root = find_git_worktree(cwd or os.getcwd())
if cwd_root is not None and is_inside_workspace(file_path, cwd_root):
return cwd_root, True
file_root = find_git_worktree(file_path)
@@ -168,10 +146,6 @@ def clear_cache() -> None:
__all__ = [
"find_git_worktree",
"is_inside_workspace",
"nearest_root",
"normalize_path",
"resolve_workspace_for_file",
"find_git_worktree", "is_inside_workspace", "nearest_root", "normalize_path", "resolve_workspace_for_file",
"clear_cache",
]
agent/monitoring/__init__.py
+1 -6
View File
@@ -14,9 +14,4 @@ from . import emitter, events
emit = emitter.emit
get_emitter = emitter.get_emitter
__all__ = [
"emitter",
"events",
"emit",
"get_emitter",
]
__all__ = ["emitter", "events", "emit", "get_emitter"]
agent/monitoring/cron_health.py
+16 -45
View File
@@ -9,11 +9,7 @@ from datetime import datetime
from typing import Any, Callable, Optional
from agent.monitoring.events import CronExecutionEvent
from agent.monitoring.gateway_health import (
GatewayMetric,
_contains_any,
_safe_instance_id,
)
from agent.monitoring.gateway_health import GatewayMetric, _contains_any, _safe_instance_id
from cron.jobs import (
_compute_grace_seconds,
get_catch_up_occurrence_count,
@@ -27,9 +23,7 @@ from hermes_time import now as _now
logger = logging.getLogger(__name__)
_KNOWN_STATUSES = {"claimed", "running", "completed", "failed", "unknown"}
_KNOWN_SOURCES = {"builtin", "direct", "external"}
_KNOWN_DELIVERY_OUTCOMES = {
"delivered", "failed", "suppressed", "suppressed_acked", "not_configured",
}
_KNOWN_DELIVERY_OUTCOMES = {"delivered", "failed", "suppressed", "suppressed_acked", "not_configured"}
_TERMINAL_STATUSES = {"completed", "failed", "unknown"}
@@ -43,11 +37,10 @@ class CronHealthSnapshot:
_job_key = _safe_instance_id
_AUTH_RE = re.compile(
r"\b(?:authentication|authenticated|authenticate|authorization|authorized|authorize"
r"|unauthorized|forbidden|bearer|401|403)\b|\b(?:access|api|refresh) token\b"
r"\b(?:authentication|authenticated|authenticate|authorization|authorized|authorize|unauthorized|forbidden|bearer|401|403)\b"
r"|\b(?:access|api|refresh) token\b"
)
# Ordered (predicate, class) rules; first match wins. Auth uses word boundaries so
# "oauth"/"tokenizer"/"HTTP 4015" do not false-positive.
_CRON_ERROR_RULES: tuple[tuple[Callable[[str], bool], str], ...] = (
@@ -86,9 +79,7 @@ def _duration_ms(record: dict[str, Any]) -> Optional[int]:
return max(0, duration)
def project_execution_event(
record: dict[str, Any], *, delivery_outcome: Optional[str] = None
) -> CronExecutionEvent:
def project_execution_event(record: dict[str, Any], *, delivery_outcome: Optional[str] = None) -> CronExecutionEvent:
status = str(record.get("status") or "unknown").lower()
source = str(record.get("source") or "unknown").lower()
outcome = str(delivery_outcome).lower() if delivery_outcome is not None else None
@@ -98,26 +89,17 @@ def project_execution_event(
# Unknown non-empty sources are bucketed as "external" (not dropped to unknown).
source=source if source in _KNOWN_SOURCES or source == "unknown" else "external",
duration_ms=_duration_ms(record),
delivery_outcome=(
outcome if outcome in _KNOWN_DELIVERY_OUTCOMES else None
),
error_class=(
classify_cron_error(record.get("error"))
if status in {"failed", "unknown"}
else None
),
delivery_outcome=outcome if outcome in _KNOWN_DELIVERY_OUTCOMES else None,
error_class=classify_cron_error(record.get("error")) if status in {"failed", "unknown"} else None,
)
def emit_execution_state(
record: Optional[dict[str, Any]], *, delivery_outcome: Optional[str] = None
) -> None:
def emit_execution_state(record: Optional[dict[str, Any]], *, delivery_outcome: Optional[str] = None) -> None:
"""Best-effort lifecycle emit; terminal states synchronously cross the queue barrier."""
if not record:
return
try:
from agent.monitoring import emitter
event = project_execution_event(record, delivery_outcome=delivery_outcome)
target = emitter.get_emitter()
target.emit(event)
@@ -137,8 +119,7 @@ def _is_overdue(job: dict[str, Any], now: datetime) -> bool:
try:
if next_run.tzinfo is None and now.tzinfo is not None:
next_run = next_run.replace(tzinfo=now.tzinfo)
lateness = (now - next_run).total_seconds()
return lateness > _compute_grace_seconds(schedule)
return (now - next_run).total_seconds() > _compute_grace_seconds(schedule)
except (TypeError, ValueError):
return False
@@ -146,11 +127,7 @@ def _is_overdue(job: dict[str, Any], now: datetime) -> bool:
def _job_metrics(metrics: list[GatewayMetric]) -> None:
enabled = [job for job in load_jobs() if job.get("enabled", True)]
metrics.append(GatewayMetric("hermes.cron.jobs.enabled", len(enabled), {}))
metrics.append(GatewayMetric(
"hermes.cron.jobs.overdue",
sum(1 for job in enabled if _is_overdue(job, _now())),
{},
))
metrics.append(GatewayMetric("hermes.cron.jobs.overdue", sum(1 for job in enabled if _is_overdue(job, _now())), {}))
def _freshness_metric(name: str, reader: Callable[[], Optional[float]]) -> Callable[[list[GatewayMetric]], None]:
@@ -158,6 +135,7 @@ def _freshness_metric(name: str, reader: Callable[[], Optional[float]]) -> Calla
value = reader()
if value is not None:
metrics.append(GatewayMetric(name, max(0.0, float(value)), {}))
return build
@@ -168,15 +146,11 @@ def _single_metric(name: str, reader: Callable[[], Any]) -> Callable[[list[Gatew
# Each group is independently fail-open so one unavailable source never hides the rest.
# Readers are wrapped in lambdas so monkeypatching this module's names still takes effect.
_METRIC_GROUPS: tuple[tuple[Callable[[list[GatewayMetric]], None], str], ...] = (
(_freshness_metric("hermes.cron.scheduler.heartbeat_age_seconds", lambda: get_ticker_heartbeat_age()),
"cron freshness metric unavailable"),
(_freshness_metric("hermes.cron.scheduler.last_success_age_seconds", lambda: get_ticker_success_age()),
"cron freshness metric unavailable"),
(_single_metric("hermes.cron.scheduler.catch_up_occurrences", lambda: get_catch_up_occurrence_count()),
"cron catch-up metric unavailable"),
(_freshness_metric("hermes.cron.scheduler.heartbeat_age_seconds", lambda: get_ticker_heartbeat_age()), "cron freshness metric unavailable"),
(_freshness_metric("hermes.cron.scheduler.last_success_age_seconds", lambda: get_ticker_success_age()), "cron freshness metric unavailable"),
(_single_metric("hermes.cron.scheduler.catch_up_occurrences", lambda: get_catch_up_occurrence_count()), "cron catch-up metric unavailable"),
(_job_metrics, "cron job metrics unavailable"),
(_single_metric("hermes.cron.jobs.running", lambda: len(get_running_job_ids())),
"cron running-job metric unavailable"),
(_single_metric("hermes.cron.jobs.running", lambda: len(get_running_job_ids())), "cron running-job metric unavailable"),
)
@@ -191,9 +165,6 @@ def build_cron_health_snapshot() -> CronHealthSnapshot:
__all__ = [
"CronHealthSnapshot",
"build_cron_health_snapshot",
"classify_cron_error",
"emit_execution_state",
"CronHealthSnapshot", "build_cron_health_snapshot", "classify_cron_error", "emit_execution_state",
"project_execution_event",
]
agent/monitoring/emitter.py
+13 -36
View File
@@ -1,11 +1,9 @@
"""Monitoring emitter: fire-and-forget queue + background dispatcher.
The single seam between producers (gateway status hooks, diagnostic log handler)
and consumers (OTLP streamers). Hot-path invariant: ``emit()`` MUST return in
O(microseconds), MUST NOT block on disk/network, and MUST NEVER raise into the
caller — a monitoring failure is logged locally and dropped. On a full queue the
*oldest* event is dropped. A daemon thread fans batches out to fail-isolated
subscribers. Nothing is persisted: monitoring is an egress path, not a store.
"""Monitoring emitter: fire-and-forget queue + background dispatcher — the single seam between
producers (gateway status hooks, diagnostic log handler) and consumers (OTLP streamers).
Hot-path invariant: ``emit()`` MUST return in O(microseconds), MUST NOT block on disk/network, and
MUST NEVER raise into the caller — a monitoring failure is logged locally and dropped. On a full
queue the *oldest* event is dropped. A daemon thread fans batches out to fail-isolated
subscribers. Nothing is persisted: monitoring is an egress path, not a store.
"""
from __future__ import annotations
@@ -14,6 +12,7 @@ import logging
import queue
import threading
import time
from contextlib import suppress
from typing import Any, Dict, Optional
logger = logging.getLogger(__name__)
@@ -67,9 +66,7 @@ class MonitoringEmitter:
with self._lock:
if self._started:
return
self._thread = threading.Thread(
target=self._run, name="hermes-monitoring-dispatch", daemon=True
)
self._thread = threading.Thread(target=self._run, name="hermes-monitoring-dispatch", daemon=True)
self._thread.start()
self._started = True
@@ -106,10 +103,8 @@ class MonitoringEmitter:
self._enabled = True
def unsubscribe(self, callback) -> None:
try:
with suppress(ValueError):
self._subscribers.remove(callback)
except ValueError:
pass
if not self._subscribers:
self._enabled = False
@@ -118,28 +113,17 @@ class MonitoringEmitter:
"""Wait boundedly for queued and in-flight batches to finish dispatch."""
if timeout <= 0:
return
finished = threading.Event()
def _wait_for_completion() -> None:
self._q.join()
finished.set()
waiter = threading.Thread(
target=_wait_for_completion,
name="hermes-monitoring-flush",
daemon=True,
)
waiter.start()
threading.Thread(target=_wait_for_completion, name="hermes-monitoring-flush", daemon=True).start()
finished.wait(timeout=timeout)
def stats(self) -> Dict[str, int]:
return {
"queued": self._q.qsize(),
"dispatched": self._dispatched,
"dropped": self._dropped,
"subscribers": len(self._subscribers),
}
return {"queued": self._q.qsize(), "dispatched": self._dispatched, "dropped": self._dropped, "subscribers": len(self._subscribers)}
def close(self) -> None:
self._stop.set()
@@ -175,16 +159,9 @@ def reset_emitter_for_tests(emitter: Optional[MonitoringEmitter] = None) -> None
global _EMITTER
with _EMITTER_LOCK:
if _EMITTER is not None and emitter is not _EMITTER:
try:
with suppress(Exception):
_EMITTER.close()
except Exception:
pass
_EMITTER = emitter
__all__ = [
"MonitoringEmitter",
"get_emitter",
"emit",
"reset_emitter_for_tests",
]
__all__ = ["MonitoringEmitter", "get_emitter", "emit", "reset_emitter_for_tests"]
agent/monitoring/events.py
+1 -11
View File
@@ -24,9 +24,7 @@ class _MonitoringEvent:
@dataclass(slots=True)
class GatewayHealthEvent(_MonitoringEvent):
"""Content-free gateway health snapshot or lifecycle event."""
EVENT: ClassVar[str] = "gateway_health"
name: str
gateway_state: Optional[str] = None
old_state: Optional[str] = None
@@ -49,9 +47,7 @@ class GatewayHealthEvent(_MonitoringEvent):
@dataclass(slots=True)
class GatewayDiagnosticEvent(_MonitoringEvent):
"""Redacted gateway diagnostic event for operator-owned observability."""
EVENT: ClassVar[str] = "gateway_diagnostic"
name: str
subsystem: str
error_class: str = "unknown"
@@ -69,9 +65,7 @@ class GatewayDiagnosticEvent(_MonitoringEvent):
@dataclass(slots=True)
class CronExecutionEvent(_MonitoringEvent):
"""Content-free durable cron execution lifecycle projection."""
EVENT: ClassVar[str] = "cron_execution"
status: str
job_key: str
source: str = "unknown"
@@ -81,8 +75,4 @@ class CronExecutionEvent(_MonitoringEvent):
ts_ns: int = field(default_factory=time.time_ns)
__all__ = [
"GatewayHealthEvent",
"GatewayDiagnosticEvent",
"CronExecutionEvent",
]
__all__ = ["GatewayHealthEvent", "GatewayDiagnosticEvent", "CronExecutionEvent"]
agent/monitoring/gateway_health.py
+55 -111
View File
@@ -60,10 +60,8 @@ _GATEWAY_ERROR_RULES: tuple[tuple[Callable[[str], bool], str], ...] = (
(_contains_any("auth", "token", "unauthorized", "forbidden", "401", "403"), "auth_failed"),
(lambda s: "rate" in s and "limit" in s, "rate_limited"),
(_contains_any("timeout", "timed out"), "timeout"),
(_contains_any(
"network", "connection", "dns", "socket", "connect call failed",
"failed to connect", "cannot connect", "unreachable", "name resolution",
), "network_error"),
(_contains_any("network", "connection", "dns", "socket", "connect call failed", "failed to connect", "cannot connect",
"unreachable", "name resolution"), "network_error"),
(_contains_any("config", "missing", "invalid"), "invalid_config"),
(_contains_any("startup"), "startup_failed"),
(_contains_any("fatal"), "platform_fatal"),
@@ -75,9 +73,7 @@ def classify_gateway_error(raw: Any) -> str:
return next((label for match, label in _GATEWAY_ERROR_RULES if match(s)), "unknown")
def classify_exit_reason(
raw: Any, *, state: Any, restart_requested: bool
) -> Optional[str]:
def classify_exit_reason(raw: Any, *, state: Any, restart_requested: bool) -> Optional[str]:
"""Reduce free-form shutdown text to a bounded operational class."""
if restart_requested:
return "restart_requested"
@@ -128,33 +124,34 @@ def platform_for_subsystem(subsystem: str) -> Optional[str]:
return (subsystem.split(".", 1)[1] or None) if subsystem.startswith("platform.") else None
def _parse_active_agents(raw: Any) -> int:
def _gateway_status(name: str, fallback: Callable[[], Any], /, **kwargs: Any) -> Any:
"""Prefer ``gateway.status.<name>`` (the runtime-status contract); fall back to the local approximation."""
try:
from gateway.status import parse_active_agents
return parse_active_agents(raw)
import gateway.status as status
return getattr(status, name)(**kwargs)
except Exception:
try:
return max(0, int(raw))
except (TypeError, ValueError):
return 0
return fallback()
def _int_or_zero(raw: Any) -> int:
try:
return max(0, int(raw))
except (TypeError, ValueError):
return 0
def _parse_active_agents(raw: Any) -> int:
return _gateway_status("parse_active_agents", lambda: _int_or_zero(raw), raw=raw)
def _derive_busy(gateway_running: bool, gateway_state: Any, active_agents: Any) -> bool:
try:
from gateway.status import derive_gateway_busy
return derive_gateway_busy(
gateway_running=gateway_running, gateway_state=gateway_state, active_agents=active_agents
)
except Exception:
return bool(gateway_running and gateway_state == "running" and _parse_active_agents(active_agents) > 0)
fallback = lambda: bool(gateway_running and gateway_state == "running" and _parse_active_agents(active_agents) > 0) # noqa: E731
return _gateway_status("derive_gateway_busy", fallback, gateway_running=gateway_running, gateway_state=gateway_state, active_agents=active_agents)
def _derive_drainable(gateway_running: bool, gateway_state: Any) -> bool:
try:
from gateway.status import derive_gateway_drainable
return derive_gateway_drainable(gateway_running=gateway_running, gateway_state=gateway_state)
except Exception:
return bool(gateway_running and gateway_state == "running")
fallback = lambda: bool(gateway_running and gateway_state == "running") # noqa: E731
return _gateway_status("derive_gateway_drainable", fallback, gateway_running=gateway_running, gateway_state=gateway_state)
def _base_attrs(*, install_id: str, version: str, supervision_mode: str) -> Dict[str, str]:
@@ -179,13 +176,13 @@ def _platforms_of(runtime: Optional[dict[str, Any]]) -> dict[str, Any]:
return raw if isinstance(raw, dict) else {}
def _platform_error_code(pdata: dict[str, Any]) -> str:
# classify_* is idempotent on its own labels, so error_class == error_code downstream.
return classify_gateway_error(pdata.get("error_code") or pdata.get("error_message"))
def build_gateway_health_snapshot(
runtime: Optional[dict[str, Any]],
*,
gateway_running: bool,
profile: str,
install_id: str,
version: str,
runtime: Optional[dict[str, Any]], *, gateway_running: bool, profile: str, install_id: str, version: str,
supervision_mode: str = "unknown",
) -> GatewayHealthSnapshot:
"""Convert gateway_state.json-compatible runtime state into P0 signals."""
@@ -196,7 +193,6 @@ def build_gateway_health_snapshot(
drainable = _derive_drainable(gateway_running, gateway_state)
platforms = _platforms_of(runtime)
base = _base_attrs(install_id=install_id, version=version, supervision_mode=supervision_mode)
metrics: list[GatewayMetric] = [
_metric("hermes.gateway.up", 1 if gateway_running else 0, base),
_metric("hermes.gateway.active_agents", active_agents, base),
@@ -205,14 +201,12 @@ def build_gateway_health_snapshot(
_metric("hermes.gateway.restart_requested", 1 if runtime.get("restart_requested") else 0, base),
_metric("hermes.gateway.state", 1, base, **{"hermes.gateway.state": gateway_state}),
]
fatal_count = 0
events: list[GatewayHealthEvent | GatewayDiagnosticEvent] = []
for platform, pdata in platforms.items():
pdata = pdata if isinstance(pdata, dict) else {}
state = _bounded_state(pdata.get("state"), allowed=_KNOWN_PLATFORM_STATES)
# classify_* is idempotent on its own labels, so error_class == error_code here.
error_code = classify_gateway_error(pdata.get("error_code") or pdata.get("error_message"))
error_code = _platform_error_code(pdata)
is_degraded = state in _FATAL_PLATFORM_STATES
if is_degraded:
fatal_count += 1
@@ -223,29 +217,14 @@ def build_gateway_health_snapshot(
))
if is_degraded:
events.append(GatewayDiagnosticEvent(
name="platform.fatal",
subsystem=f"platform.{platform}",
platform=str(platform),
error_code=error_code,
error_class=error_code,
profile=profile,
version=version,
name="platform.fatal", subsystem=f"platform.{platform}", platform=str(platform),
error_code=error_code, error_class=error_code, profile=profile, version=version,
severity="error" if state == "fatal" else "warning",
))
events.insert(0, GatewayHealthEvent(
name="gateway.health_snapshot",
gateway_state=gateway_state,
active_agents=active_agents,
gateway_busy=busy,
gateway_drainable=drainable,
platform_count=len(platforms),
fatal_platform_count=fatal_count,
profile=profile,
install_id=install_id,
version=version,
supervision_mode=supervision_mode,
pid=_coerce_pid(runtime.get("pid")),
name="gateway.health_snapshot", gateway_state=gateway_state, active_agents=active_agents, gateway_busy=busy,
gateway_drainable=drainable, platform_count=len(platforms), fatal_platform_count=fatal_count, profile=profile,
install_id=install_id, version=version, supervision_mode=supervision_mode, pid=_coerce_pid(runtime.get("pid")),
))
return GatewayHealthSnapshot(metrics=metrics, events=events)
@@ -278,31 +257,18 @@ def _lifecycle_events(
def health(name: str) -> GatewayHealthEvent:
return GatewayHealthEvent(
name=name,
gateway_state=new_state,
old_state=old_state,
new_state=new_state,
exit_reason=classify_exit_reason(
current.get("exit_reason"), state=new_state, restart_requested=restart_requested
),
restart_requested=restart_requested,
active_agents=_parse_active_agents(current.get("active_agents", 0)),
profile=profile,
version=version,
pid=_coerce_pid(current.get("pid")),
name=name, gateway_state=new_state, old_state=old_state, new_state=new_state,
exit_reason=classify_exit_reason(current.get("exit_reason"), state=new_state, restart_requested=restart_requested),
restart_requested=restart_requested, active_agents=_parse_active_agents(current.get("active_agents", 0)),
profile=profile, version=version, pid=_coerce_pid(current.get("pid")),
)
out: list[GatewayHealthEvent | GatewayDiagnosticEvent] = [health("gateway.lifecycle")]
if new_state == "startup_failed":
error_class = classify_gateway_error(current.get("exit_reason") or "startup_failed")
out.append(GatewayDiagnosticEvent(
name="gateway.startup_failed",
subsystem="gateway",
error_class=error_class,
error_code=error_class,
profile=profile,
version=version,
severity="error",
name="gateway.startup_failed", subsystem="gateway", error_class=error_class, error_code=error_class,
profile=profile, version=version, severity="error",
))
if new_state == "stopped":
out.append(health("gateway.exit"))
@@ -323,30 +289,20 @@ def _platform_events(
new_state = _optional_state(pdata.get("state"), allowed=_KNOWN_PLATFORM_STATES)
if old_state == new_state or not new_state:
continue
error_code = classify_gateway_error(pdata.get("error_code") or pdata.get("error_message"))
error_code = _platform_error_code(pdata)
common: dict[str, Any] = dict(
subsystem=f"platform.{platform}",
platform=str(platform),
error_code=error_code,
error_class=error_code,
profile=profile,
version=version,
severity="error" if new_state in {"fatal", "failed", "error"} else "warning",
subsystem=f"platform.{platform}", platform=str(platform), error_code=error_code, error_class=error_code,
profile=profile, version=version, severity="error" if new_state in {"fatal", "failed", "error"} else "warning",
)
out.append(GatewayDiagnosticEvent(
name="platform.state_change", old_state=old_state, new_state=new_state, **common
))
out.append(GatewayDiagnosticEvent(name="platform.state_change", old_state=old_state, new_state=new_state, **common))
if new_state in _FATAL_PLATFORM_STATES:
out.append(GatewayDiagnosticEvent(name="platform.fatal", **common))
return out
def emit_runtime_status_transition(previous: Optional[dict[str, Any]], current: dict[str, Any]) -> None:
"""Emit immediate content-free gateway events for runtime status changes.
Called by gateway.status.write_runtime_status after persisting the new status.
Fully fail-open: failures never affect gateway status writes.
"""
"""Emit immediate content-free gateway events for runtime status changes. Called by
gateway.status.write_runtime_status after persisting; fully fail-open."""
try:
ctx = dict(profile=_safe_profile(), version=_safe_version())
for ev in _lifecycle_events(previous, current, **ctx) + _platform_events(previous, current, **ctx):
@@ -373,33 +329,21 @@ class GatewayDiagnosticLogHandler(logging.Handler):
def emit(self, record: logging.LogRecord) -> None:
try:
if record.levelno < logging.WARNING:
return
if not (record.name == "gateway" or record.name.startswith("gateway.")):
if record.levelno < logging.WARNING or not (record.name == "gateway" or record.name.startswith("gateway.")):
return
subsystem = subsystem_for_logger(record.name)
error_class = classify_gateway_error(record.getMessage())
event = GatewayDiagnosticEvent(
name=f"gateway.log.{record.levelname.lower()}",
subsystem=subsystem,
source_logger=source_logger_for_export(record.name),
platform=platform_for_subsystem(subsystem),
error_class=error_class,
error_code=error_class,
profile=self.profile,
version=self.version,
emitter.get_emitter().emit(GatewayDiagnosticEvent(
name=f"gateway.log.{record.levelname.lower()}", subsystem=subsystem,
source_logger=source_logger_for_export(record.name), platform=platform_for_subsystem(subsystem),
error_class=error_class, error_code=error_class, profile=self.profile, version=self.version,
severity=record.levelname.lower(),
)
emitter.get_emitter().emit(event)
))
except Exception:
logger.debug("gateway diagnostic emit failed", exc_info=True)
__all__ = [
"GatewayMetric",
"GatewayHealthSnapshot",
"GatewayDiagnosticLogHandler",
"build_gateway_health_snapshot",
"classify_gateway_error",
"source_logger_for_export",
"GatewayMetric", "GatewayHealthSnapshot", "GatewayDiagnosticLogHandler",
"build_gateway_health_snapshot", "classify_gateway_error", "source_logger_for_export",
]
agent/monitoring/gateway_health_export.py
+33 -96
View File
@@ -1,9 +1,8 @@
"""Gateway Health & Diagnostics OTLP export runtime.
Emits operator-owned gateway service-health metrics plus narrow redacted
diagnostic events. Deliberately in-process and fail-open so it works under
systemd, launchd, s6, containers, tmux, nohup, or a plain shell without a
sidecar/watchdog dependency.
Emits operator-owned gateway service-health metrics plus narrow redacted diagnostic events.
Deliberately in-process and fail-open so it works under systemd, launchd, s6, containers,
tmux, nohup, or a plain shell without a sidecar/watchdog dependency.
"""
from __future__ import annotations
@@ -12,6 +11,7 @@ import importlib
import logging
import os
import threading
from contextlib import suppress
from dataclasses import dataclass
from typing import Any, Callable, Dict, Optional
@@ -47,21 +47,11 @@ _METRICS_SDK = (
)
# Every gauge the runtime snapshot can emit MUST be listed here or it is silently dropped.
_OBSERVABLE_METRIC_NAMES = (
"hermes.gateway.up",
"hermes.gateway.state",
"hermes.gateway.active_agents",
"hermes.gateway.busy",
"hermes.gateway.drainable",
"hermes.gateway.restart_requested",
"hermes.gateway.background_work",
"hermes.gateway.background_delegations",
"hermes.platform.up",
"hermes.platform.degraded",
"hermes.cron.scheduler.heartbeat_age_seconds",
"hermes.cron.scheduler.last_success_age_seconds",
"hermes.cron.scheduler.catch_up_occurrences",
"hermes.cron.jobs.enabled",
"hermes.cron.jobs.running",
"hermes.gateway.up", "hermes.gateway.state", "hermes.gateway.active_agents", "hermes.gateway.busy",
"hermes.gateway.drainable", "hermes.gateway.restart_requested", "hermes.gateway.background_work",
"hermes.gateway.background_delegations", "hermes.platform.up", "hermes.platform.degraded",
"hermes.cron.scheduler.heartbeat_age_seconds", "hermes.cron.scheduler.last_success_age_seconds",
"hermes.cron.scheduler.catch_up_occurrences", "hermes.cron.jobs.enabled", "hermes.cron.jobs.running",
"hermes.cron.jobs.overdue",
)
@@ -88,40 +78,29 @@ class GatewayHealthExportRuntime:
if self.thread is not None:
self.thread.join(timeout=0.25)
if self.log_handler is not None:
try:
with suppress(Exception):
logging.getLogger().removeHandler(self.log_handler)
except Exception:
pass
# Producers are stopped; drain queued/in-flight events BEFORE detaching subscribers
# so the terminal lifecycle event cannot race exporter shutdown. Bounded, fail-open.
subscribers = [item for item in (self.streamer, self.log_streamer) if item is not None]
try:
with suppress(Exception):
bus = emitter.get_emitter()
bus.flush(timeout=1.0)
for sub in subscribers:
bus.unsubscribe(sub)
except Exception:
pass
# Network flush/close runs under one bounded daemon-thread deadline so it can
# never delay gateway teardown indefinitely.
closeables = subscribers + ([self.metric_provider] if self.metric_provider is not None else [])
def _close() -> None:
for item in closeables:
try:
with suppress(Exception):
item.shutdown()
except Exception:
pass
if closeables:
worker = threading.Thread(
target=_close, name="hermes-gateway-health-export-shutdown", daemon=True
)
worker = threading.Thread(target=_close, name="hermes-gateway-health-export-shutdown", daemon=True)
worker.start()
worker.join(timeout=2.0)
self.streamer = self.log_streamer = self.metric_provider = self.thread = self.stop_event = None
@@ -172,18 +151,13 @@ def _read_gateway_snapshot(config: Dict[str, Any]):
except Exception:
runtime = {}
return build_gateway_health_snapshot(
runtime,
gateway_running=True,
profile=_profile(),
install_id=_install_id(config),
version=_version(),
runtime, gateway_running=True, profile=_profile(), install_id=_install_id(config), version=_version(),
supervision_mode=_supervision_mode(),
)
def _read_cron_snapshot():
from agent.monitoring.cron_health import build_cron_health_snapshot
return build_cron_health_snapshot()
@@ -197,27 +171,20 @@ def _count(failure_msg: str, module: str, read: Callable[[Any], Any]) -> int:
def _read_background_work_count() -> int:
"""Live background/subagent work that ``active_agents`` deliberately does NOT include.
``active_agents`` counts foreground turns + in-flight cron + API runs; backgrounded
``delegate_task`` subagents, ``terminal(background=true)`` processes and kanban workers
are tracked only by the scale-to-zero guard, so without this a peer churning through
subagents shows ``active_agents=0``. TASK-granular: a fan-out batch of N contributes N
(real concurrent load), unlike the pool's one-slot-per-batch accounting. Content-free.
"""
"""Live background/subagent work that ``active_agents`` (foreground turns + in-flight cron + API
runs) deliberately does NOT include: backgrounded ``delegate_task`` subagents,
``terminal(background=true)`` processes, kanban workers. TASK-granular: a fan-out batch of N
contributes N (real concurrent load), unlike the pool's one-slot-per-batch accounting."""
return (
_count("background-work async-delegation count failed", "tools.async_delegation",
lambda m: m.active_task_count())
_count("background-work async-delegation count failed", "tools.async_delegation", lambda m: m.active_task_count())
+ _count("background-work process-registry count failed", "tools.process_registry",
lambda m: m.process_registry.count_running())
)
def _read_background_delegations_count() -> int:
"""Live async delegation UNITS (dispatch/pool slots): a batch counts ONE regardless of
fan-out width, matching the pool's capacity accounting — so operators can see slot
pressure (alert vs ``max_concurrent_children``) alongside ``background_work``'s real load.
Delegations only; terminal/kanban work is already folded into ``background_work``."""
"""Live async delegation UNITS (pool slots): a batch counts ONE regardless of fan-out width, so
operators see slot pressure (vs ``max_concurrent_children``) alongside ``background_work``."""
return _count("background-delegations count failed", "tools.async_delegation", lambda m: m.active_count())
@@ -233,20 +200,14 @@ def _read_runtime_snapshot(config: Dict[str, Any]):
):
gateway_snapshot.metrics.append(GatewayMetric(name=name, value=read(), attributes=base))
except Exception as exc:
logger.warning(
"background-work snapshot unavailable; metric not exported (error_type=%s)",
type(exc).__name__,
)
logger.warning("background-work snapshot unavailable; metric not exported (error_type=%s)", type(exc).__name__)
logger.debug("background-work snapshot traceback", exc_info=True)
try:
cron_snapshot = _read_cron_snapshot()
except Exception as exc:
# Cron telemetry silently dropping out is a release-relevant regression: WARN with only
# the exception *type* (the message could carry paths); exc_info stays on DEBUG.
logger.warning(
"cron health snapshot unavailable; cron telemetry not exported (error_type=%s)",
type(exc).__name__,
)
logger.warning("cron health snapshot unavailable; cron telemetry not exported (error_type=%s)", type(exc).__name__)
logger.debug("cron health snapshot traceback", exc_info=True)
return gateway_snapshot
gateway_snapshot.metrics.extend(cron_snapshot.metrics)
@@ -268,20 +229,18 @@ def _start_metric_provider(config: Dict[str, Any], sdk: Dict[str, Any]) -> Any:
exporter = sdk["OTLPMetricExporter"](**_exporter_kwargs(config, "metrics"))
interval_ms = max(5, int(gh.get("export_interval_seconds", 60))) * 1000
reader = sdk["PeriodicExportingMetricReader"](exporter, export_interval_millis=interval_ms)
provider = sdk["MeterProvider"](
metric_readers=[reader], resource=_resource(config, sdk, "gateway_health")
)
provider = sdk["MeterProvider"](metric_readers=[reader], resource=_resource(config, sdk, "gateway_health"))
meter = provider.get_meter("hermes.gateway.health")
Observation = sdk["Observation"]
def callback(name: str):
def _cb(_options=None):
try:
snapshot = _read_runtime_snapshot(config)
return [Observation(m.value, m.attributes) for m in snapshot.metrics if m.name == name]
return [Observation(m.value, m.attributes) for m in _read_runtime_snapshot(config).metrics if m.name == name]
except Exception:
logger.debug("gateway metric callback failed", exc_info=True)
return []
return _cb
for metric_name in _OBSERVABLE_METRIC_NAMES:
@@ -290,12 +249,7 @@ def _start_metric_provider(config: Dict[str, Any], sdk: Dict[str, Any]) -> Any:
_SEVERITY_NAMES = {
"critical": "FATAL",
"fatal": "FATAL",
"error": "ERROR",
"info": "INFO",
"information": "INFO",
"debug": "DEBUG",
"critical": "FATAL", "fatal": "FATAL", "error": "ERROR", "info": "INFO", "information": "INFO", "debug": "DEBUG",
}
@@ -309,9 +263,7 @@ class GatewayDiagnosticLogStreamer(EmitterStreamer):
def __init__(self, config: Dict[str, Any], sdk: Dict[str, Any]):
self._provider = sdk["LoggerProvider"](resource=_resource(config, sdk, "gateway_diagnostics"))
self._processor = sdk["BatchLogRecordProcessor"](
sdk["OTLPLogExporter"](**_exporter_kwargs(config, "logs"))
)
self._processor = sdk["BatchLogRecordProcessor"](sdk["OTLPLogExporter"](**_exporter_kwargs(config, "logs")))
self._provider.add_log_record_processor(self._processor)
self._logger = self._provider.get_logger(_DEFAULT_DIAGNOSTIC_SCOPE)
self._sdk = sdk
@@ -328,13 +280,9 @@ class GatewayDiagnosticLogStreamer(EmitterStreamer):
source_logger = source_logger_for_export(ev.get("source_logger"))
otel_logger = self._provider.get_logger(source_logger) if source_logger is not None else self._logger
otel_logger.emit(sdk["LogRecord"](
timestamp=ev.get("ts_ns"),
trace_id=sdk["INVALID_TRACE_ID"],
span_id=sdk["INVALID_SPAN_ID"],
trace_flags=sdk["TraceFlags"].DEFAULT,
severity_text=str(ev.get("severity") or "warning").upper(),
severity_number=_severity_number(sdk, ev.get("severity")),
body=redact_bounded("gateway diagnostic"),
timestamp=ev.get("ts_ns"), trace_id=sdk["INVALID_TRACE_ID"], span_id=sdk["INVALID_SPAN_ID"],
trace_flags=sdk["TraceFlags"].DEFAULT, severity_text=str(ev.get("severity") or "warning").upper(),
severity_number=_severity_number(sdk, ev.get("severity")), body=redact_bounded("gateway diagnostic"),
attributes=_diagnostic_log_attributes(ev),
))
self.exported += 1
@@ -376,18 +324,12 @@ def start_gateway_health_export(config: Dict[str, Any]) -> GatewayHealthExportRu
diagnostics_on = gh.get("diagnostic_events_enabled", True)
runtime = GatewayHealthExportRuntime(enabled=True, reason="enabled")
sdk: Optional[Dict[str, Any]] = None
if metrics_on or diagnostics_on:
try:
sdk = _require_metrics_sdk(prompt=False)
except Exception:
logger.warning(
"monitoring.gateway_health_export.enabled but OTLP SDK is unavailable; "
"install 'hermes-agent[otlp]'",
exc_info=True,
)
logger.warning("monitoring.gateway_health_export.enabled but OTLP SDK is unavailable; install 'hermes-agent[otlp]'", exc_info=True)
return GatewayHealthExportRuntime(enabled=False, reason="otlp_unavailable")
if metrics_on and sdk is not None:
try:
runtime.metric_provider = _start_metric_provider(config, sdk)
@@ -395,7 +337,6 @@ def start_gateway_health_export(config: Dict[str, Any]) -> GatewayHealthExportRu
logger.warning("gateway health OTLP metrics failed to start", exc_info=True)
runtime.shutdown()
return GatewayHealthExportRuntime(enabled=False, reason="metrics_start_failed")
if diagnostics_on and sdk is not None:
try:
runtime.streamer = otlp_exporter.start_streaming(config, event_filter=_gateway_health_event)
@@ -408,7 +349,6 @@ def start_gateway_health_export(config: Dict[str, Any]) -> GatewayHealthExportRu
logger.debug("gateway diagnostic OTLP export failed to start", exc_info=True)
runtime.shutdown()
return GatewayHealthExportRuntime(enabled=False, reason="diagnostics_start_failed")
try:
runtime.log_handler = _attach_log_handler(config)
except Exception:
@@ -423,7 +363,4 @@ def start_gateway_health_export(config: Dict[str, Any]) -> GatewayHealthExportRu
return runtime
__all__ = [
"GatewayHealthExportRuntime",
"start_gateway_health_export",
]
__all__ = ["GatewayHealthExportRuntime", "start_gateway_health_export"]
agent/monitoring/otlp_exporter.py
+50 -106
View File
@@ -1,18 +1,10 @@
"""Export monitoring events to an OpenTelemetry Collector over OTLP/HTTP.
Maps gateway monitoring events to OTel spans and sends them to the operator
configured ``monitoring.export.otlp`` endpoint (no default destination ships).
Also hosts the OTLP plumbing shared with ``gateway_health_export`` (SDK loading,
header resolution, resource attributes, endpoint mapping).
* The OTel SDK is an optional extra (``pip install hermes-agent[otlp]``),
imported lazily so it is only required when export is actually used.
* ``headers_env`` maps a header name to an environment variable name; values
are read from the environment at export time and never logged or stored.
* The continuous subscriber runs in the emitter's dispatcher thread and is
fail-isolated, so an export error cannot affect the gateway. The
``event_filter`` seam keeps future planes sharing the emitter from silently
riding along on this exporter.
Maps gateway monitoring events to OTel spans for the operator-configured ``monitoring.export.otlp``
endpoint (no default destination ships) and hosts the OTLP plumbing shared with
``gateway_health_export``. The OTel SDK is an optional extra (``hermes-agent[otlp]``) imported
lazily; ``headers_env`` values are read at export time and never logged or stored. The streaming
subscriber runs fail-isolated on the emitter thread; ``event_filter`` keeps other planes off it.
"""
from __future__ import annotations
@@ -21,6 +13,7 @@ import importlib
import logging
import os
import re
from contextlib import suppress
from typing import Any, Callable, Dict, Iterable, List, Optional
from agent.monitoring.gateway_health import _safe_instance_id
@@ -34,49 +27,39 @@ class OTLPUnavailable(RuntimeError):
# ── SDK loading ──────────────────────────────────────────────────────────────
_SDK_SYMBOLS: Dict[str, str] = {
"TracerProvider": "opentelemetry.sdk.trace",
"BatchSpanProcessor": "opentelemetry.sdk.trace.export",
"Resource": "opentelemetry.sdk.resources",
"OTLPSpanExporter": "opentelemetry.exporter.otlp.proto.http.trace_exporter",
"SpanKind": "opentelemetry.trace",
"OTLPLogExporter": "opentelemetry.exporter.otlp.proto.http._log_exporter",
"OTLPMetricExporter": "opentelemetry.exporter.otlp.proto.http.metric_exporter",
"Observation": "opentelemetry.metrics",
"INVALID_SPAN_ID": "opentelemetry.trace",
"INVALID_TRACE_ID": "opentelemetry.trace",
"TraceFlags": "opentelemetry.trace",
"LogRecord": "opentelemetry._logs",
"SeverityNumber": "opentelemetry._logs.severity",
"LoggerProvider": "opentelemetry.sdk._logs",
"BatchLogRecordProcessor": "opentelemetry.sdk._logs.export",
"MeterProvider": "opentelemetry.sdk.metrics",
"PeriodicExportingMetricReader": "opentelemetry.sdk.metrics.export",
_SDK_MODULES: Dict[str, tuple[str, ...]] = {
"opentelemetry.sdk.trace": ("TracerProvider",),
"opentelemetry.sdk.trace.export": ("BatchSpanProcessor",),
"opentelemetry.sdk.resources": ("Resource",),
"opentelemetry.exporter.otlp.proto.http.trace_exporter": ("OTLPSpanExporter",),
"opentelemetry.exporter.otlp.proto.http._log_exporter": ("OTLPLogExporter",),
"opentelemetry.exporter.otlp.proto.http.metric_exporter": ("OTLPMetricExporter",),
"opentelemetry.trace": ("SpanKind", "INVALID_SPAN_ID", "INVALID_TRACE_ID", "TraceFlags"),
"opentelemetry.metrics": ("Observation",),
"opentelemetry._logs": ("LogRecord",),
"opentelemetry._logs.severity": ("SeverityNumber",),
"opentelemetry.sdk._logs": ("LoggerProvider",),
"opentelemetry.sdk._logs.export": ("BatchLogRecordProcessor",),
"opentelemetry.sdk.metrics": ("MeterProvider",),
"opentelemetry.sdk.metrics.export": ("PeriodicExportingMetricReader",),
}
_SDK_SYMBOLS: Dict[str, str] = {name: module for module, names in _SDK_MODULES.items() for name in names}
_SPAN_SDK = ("TracerProvider", "BatchSpanProcessor", "Resource", "OTLPSpanExporter", "SpanKind")
def _require_sdk(
names: Iterable[str] = _SPAN_SDK, *, auto_install: bool = True, prompt: bool = True
) -> Dict[str, Any]:
def _require_sdk(names: Iterable[str] = _SPAN_SDK, *, auto_install: bool = True, prompt: bool = True) -> Dict[str, Any]:
"""Import the named OTel SDK symbols, lazily installing the extra on first use.
Routes through tools.lazy_deps (feature 'export.otlp') — gated by
security.allow_lazy_installs and TTY-prompted (``prompt=False`` from
non-interactive contexts). Any lazy-install failure falls through to the
import attempt, which raises OTLPUnavailable with a manual install hint.
Routes through tools.lazy_deps (feature 'export.otlp') — gated by security.allow_lazy_installs
and TTY-prompted (``prompt=False`` from non-interactive contexts). Any lazy-install failure
falls through to the import attempt, which raises OTLPUnavailable with a manual install hint.
"""
if auto_install:
try:
with suppress(Exception):
from tools.lazy_deps import ensure as _lazy_ensure
_lazy_ensure("export.otlp", prompt=prompt)
except Exception:
pass
try:
return {
name: getattr(importlib.import_module(_SDK_SYMBOLS[name]), name)
for name in names
}
return {name: getattr(importlib.import_module(_SDK_SYMBOLS[name]), name) for name in names}
except Exception as e: # ImportError or partial install
raise OTLPUnavailable(
"OTLP export requires the optional dependency. Install with:\n"
@@ -122,15 +105,8 @@ def _signal_endpoint(endpoint: str, signal: str) -> str:
_RESOURCE_ATTRIBUTE_KEYS = frozenset({
"service.name",
"service.namespace",
"service.version",
"service.instance.id",
"deployment.environment.name",
"cloud.provider",
"cloud.platform",
"cloud.region",
"telemetry.scope",
"service.name", "service.namespace", "service.version", "service.instance.id",
"deployment.environment.name", "cloud.provider", "cloud.platform", "cloud.region", "telemetry.scope",
})
_SAFE_RESOURCE_VALUE = re.compile(r"^[A-Za-z0-9._:/-]{1,128}$")
@@ -158,12 +134,9 @@ def _safe_resource_attributes(raw: Any) -> Dict[str, str]:
return attrs
def _runtime_resource_attributes(
config: Dict[str, Any], *, telemetry_scope: str
) -> Dict[str, str]:
def _runtime_resource_attributes(config: Dict[str, Any], *, telemetry_scope: str) -> Dict[str, str]:
"""Build the safe OTLP resource shared by spans, metrics and diagnostic logs."""
gh = _monitoring_section(config, "gateway_health_export")
attrs = _safe_resource_attributes(gh.get("resource_attributes"))
attrs = _safe_resource_attributes(_monitoring_section(config, "gateway_health_export").get("resource_attributes"))
attrs["service.name"] = "hermes-gateway"
attrs["service.instance.id"] = _safe_instance_id(_install_id(config))
attrs["telemetry.scope"] = telemetry_scope
@@ -177,8 +150,7 @@ def build_exporter(config: Dict[str, Any]):
endpoint = otlp.get("endpoint")
if not endpoint:
raise ValueError("monitoring.export.otlp.endpoint is not set")
headers = _resolve_headers(otlp.get("headers_env"))
return sdk["OTLPSpanExporter"](endpoint=endpoint, headers=headers or None)
return sdk["OTLPSpanExporter"](endpoint=endpoint, headers=_resolve_headers(otlp.get("headers_env")) or None)
def _resource_attributes(config: Dict[str, Any]) -> Dict[str, str]:
@@ -187,8 +159,7 @@ def _resource_attributes(config: Dict[str, Any]) -> Dict[str, str]:
def _make_provider(config: Dict[str, Any]):
sdk = _require_sdk()
resource = sdk["Resource"].create(_resource_attributes(config))
provider = sdk["TracerProvider"](resource=resource)
provider = sdk["TracerProvider"](resource=sdk["Resource"].create(_resource_attributes(config)))
processor = sdk["BatchSpanProcessor"](build_exporter(config))
provider.add_span_processor(processor)
return provider, processor
@@ -197,16 +168,12 @@ def _make_provider(config: Dict[str, Any]):
# ── event -> span attribute mapping ──────────────────────────────────────────
# Per-kind attribute allowlists: everything else (profile, install_id, ...) never egresses.
_KEEP_BY_KIND: Dict[str, tuple[str, ...]] = {
"gateway_health": ("name", "gateway_state", "old_state", "new_state",
"exit_reason", "restart_requested", "active_agents",
"gateway_busy", "gateway_drainable", "platform_count",
"fatal_platform_count", "version",
"supervision_mode", "pid"),
"gateway_diagnostic": ("name", "subsystem", "error_class", "error_code",
"platform", "old_state", "new_state",
"version", "severity"),
"cron_execution": ("status", "job_key", "source", "duration_ms",
"delivery_outcome", "error_class"),
"gateway_health": (
"name", "gateway_state", "old_state", "new_state", "exit_reason", "restart_requested", "active_agents",
"gateway_busy", "gateway_drainable", "platform_count", "fatal_platform_count", "version", "supervision_mode", "pid",
),
"gateway_diagnostic": ("name", "subsystem", "error_class", "error_code", "platform", "old_state", "new_state", "version", "severity"),
"cron_execution": ("status", "job_key", "source", "duration_ms", "delivery_outcome", "error_class"),
}
@@ -234,9 +201,7 @@ def export_batch(provider, batch: List[Dict[str, Any]]) -> int:
n = 0
for ev in batch:
try:
name = f"hermes.{ev.get('event', 'event')}"
span = tracer.start_span(name, attributes=_span_attrs(ev))
span.end()
tracer.start_span(f"hermes.{ev.get('event', 'event')}", attributes=_span_attrs(ev)).end()
n += 1
except Exception:
logger.debug("OTLP span map failed", exc_info=True)
@@ -246,36 +211,24 @@ def export_batch(provider, batch: List[Dict[str, Any]]) -> int:
# ── continuous streaming subscribers ─────────────────────────────────────────
class EmitterStreamer:
"""Base for emitter subscribers owning an OTel provider + batch processor.
Register with ``emitter.subscribe(streamer)``. Fail-isolated by the emitter.
"""
Register with ``emitter.subscribe(streamer)``. Fail-isolated by the emitter."""
_provider: Any
_processor: Any
exported: int = 0
def shutdown(self) -> None:
try:
with suppress(Exception):
from agent.monitoring.emitter import get_emitter
get_emitter().unsubscribe(self)
except Exception:
pass
try:
with suppress(Exception):
self._processor.force_flush()
self._provider.shutdown()
except Exception:
pass
class OTLPStreamer(EmitterStreamer):
"""A live subscriber that pushes each emitter batch to OTLP as spans."""
def __init__(
self,
config: Dict[str, Any],
*,
event_filter: Optional[Callable[[Dict[str, Any]], bool]] = None,
):
def __init__(self, config: Dict[str, Any], *, event_filter: Optional[Callable[[Dict[str, Any]], bool]] = None):
self._provider, self._processor = _make_provider(config)
self._event_filter = event_filter
self.exported = 0
@@ -303,16 +256,13 @@ def is_enabled(config: Dict[str, Any]) -> bool:
def start_streaming(
config: Dict[str, Any],
*,
event_filter: Optional[Callable[[Dict[str, Any]], bool]] = None,
config: Dict[str, Any], *, event_filter: Optional[Callable[[Dict[str, Any]], bool]] = None,
) -> Optional[OTLPStreamer]:
"""If OTLP is enabled, attach a streamer to the singleton emitter.
``event_filter`` scopes the exporter to its plane. Startup is non-interactive:
a configured-but-missing SDK is lazily installed once (prompt=False, gated by
security.allow_lazy_installs); if it still can't load, log and no-op — never
raise into startup.
``event_filter`` scopes the exporter to its plane. Startup is non-interactive: a
configured-but-missing SDK is lazily installed once (prompt=False, gated by
security.allow_lazy_installs); if it still can't load, log and no-op — never raise into startup.
"""
if not is_enabled(config):
return None
@@ -329,11 +279,5 @@ def start_streaming(
__all__ = [
"OTLPUnavailable",
"OTLPStreamer",
"build_exporter",
"export_batch",
"is_available",
"is_enabled",
"start_streaming",
"OTLPUnavailable", "OTLPStreamer", "build_exporter", "export_batch", "is_available", "is_enabled", "start_streaming",
]
agent/monitoring/policy.py
+1 -5
View File
@@ -25,11 +25,9 @@ def ensure_install_id(config: Dict[str, Any]) -> str:
existing = (mon or {}).get("install_id") if isinstance(mon, dict) else None
if isinstance(existing, str) and existing.strip():
return existing
minted = str(uuid.uuid4())
try:
from hermes_cli.config import load_config, save_config
fresh = load_config()
if isinstance(fresh, dict):
slot = fresh.setdefault("monitoring", {})
@@ -46,6 +44,4 @@ def ensure_install_id(config: Dict[str, Any]) -> str:
return minted
__all__ = [
"ensure_install_id",
]
__all__ = ["ensure_install_id"]
agent/monitoring/redaction.py
+7 -22
View File
@@ -1,11 +1,9 @@
"""Redaction applied to monitoring data before egress.
One unconditional scrub, no modes, no knobs. Every string that leaves the
process passes through ``redact_for_export``: secrets first (wraps
``agent/redact.py::redact_sensitive_text(force=True)`` plus bearer/token
shapes, failing CLOSED so a broken redactor never emits the raw string), then
One unconditional scrub, no modes, no knobs. Every string that leaves the process passes
through ``redact_for_export``: secrets first (``agent/redact.py::redact_sensitive_text(force=True)``
plus bearer/token shapes, failing CLOSED so a broken redactor never emits the raw string), then
PII (e-mail, phone, UUID-shaped ids -> ``[email]`` / ``[phone]`` / ``[id]``).
There is deliberately no setting to weaken this.
"""
from __future__ import annotations
@@ -15,9 +13,7 @@ from typing import Any, Optional
# ── secret shapes (belt-and-suspenders on top of agent/redact.py) ───────────
_BEARER_RE = re.compile(r"\bBearer\s+[A-Za-z0-9._~+\-/]+=*", re.IGNORECASE)
_TOKEN_RE = re.compile(
r"\b(xox[baprs]-[A-Za-z0-9-]+|sk-[A-Za-z0-9_-]{8,}|gh[pousr]_[A-Za-z0-9_]{8,})\b"
)
_TOKEN_RE = re.compile(r"\b(xox[baprs]-[A-Za-z0-9-]+|sk-[A-Za-z0-9_-]{8,}|gh[pousr]_[A-Za-z0-9_]{8,})\b")
_SECRET_LITERAL_RE = re.compile(r"\*{3,}")
_BEARER_RESIDUE_RE = re.compile(r"\bBearer\s+\[[^\]]+\]", re.IGNORECASE)
@@ -27,9 +23,7 @@ _EMAIL_RE = re.compile(r"[A-Za-z0-9._%+\-]+@[A-Za-z0-9.\-]+\.[A-Za-z]{2,}")
_PHONE_RE = re.compile(
r"(?<!\w)(?:\+?\d{1,3}[\s.\-]?)?(?:\(\d{2,4}\)[\s.\-]?)?\d{3}[\s.\-]?\d{3,4}(?:[\s.\-]?\d{2,4})?(?!\w)"
)
_UUID_RE = re.compile(
r"\b[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12}\b"
)
_UUID_RE = re.compile(r"\b[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12}\b")
UNAVAILABLE = "[redaction-unavailable]"
@@ -58,13 +52,7 @@ def redact_for_export(text: Optional[str]) -> Optional[str]:
return out
def redact_bounded(
raw: Any,
*,
limit: int = 500,
empty: str = "[redacted]",
unavailable: str = UNAVAILABLE,
) -> str:
def redact_bounded(raw: Any, *, limit: int = 500, empty: str = "[redacted]", unavailable: str = UNAVAILABLE) -> str:
"""Redact ``str(raw or "")`` and length-bound it; ``empty`` replaces an empty
result, ``unavailable`` is returned if redaction itself raises."""
try:
@@ -73,7 +61,4 @@ def redact_bounded(
return unavailable
__all__ = [
"redact_for_export",
"redact_bounded",
]
__all__ = ["redact_for_export", "redact_bounded"]
agent/outbound_webhooks.py
+55 -154
View File
@@ -1,31 +1,10 @@
"""Outbound webhook notifications.
Reads ``hooks.outbound:`` from config.yaml and registers notify-only callbacks on
the plugin hook manager, so every ``invoke_hook()`` site can push lifecycle events
to external HTTP endpoints. Outbound mirror of ``gateway/platforms/webhook.py``.
* Delivery is fire-and-forget through a bounded queue and one daemon worker
thread; callbacks serialize, enqueue, and return ``None`` immediately, so a
target can never block a tool call or influence agent flow.
* Payloads are HMAC-SHA256 signed (``X-Hermes-Signature-256: sha256=<hex>`` over
the raw body) when a secret is configured.
* No consent prompt (no code runs on this machine); ``HERMES_SAFE_MODE=1`` still
skips registration. Registration is idempotent.
Config::
hooks:
outbound:
- url: https://ci.example.com/hermes-events
events: [on_session_end, subagent_stop]
secret_env: HERMES_OUTBOUND_WEBHOOK_SECRET # or inline ``secret``
matcher: "terminal|delegate_task" # pre/post_tool_call only
timeout: 10 # seconds, clamped to [1, 60]
name: ci-notify
POST body: ``{hook_event_name, profile, tool_name, tool_input, session_id, cwd,
extra, delivery_id, timestamp}``. Headers: ``Content-Type``, ``User-Agent``,
``X-Hermes-Event``, ``X-Hermes-Delivery``, ``X-Hermes-Signature-256`` (if secret).
"""Outbound webhooks: ``hooks.outbound`` entries (url, events, secret_env|secret, matcher for
pre/post_tool_call, timeout clamped to [1, 60], name) -> notify-only callbacks on the plugin hook
manager, so every ``invoke_hook()`` site can POST lifecycle events (mirror of
``gateway/platforms/webhook.py``). Fire-and-forget through a bounded queue + one daemon worker,
so a target can never block a tool call or influence agent flow. HMAC-SHA256 signed
(``X-Hermes-Signature-256: sha256=<hex>`` over the raw body) when a secret is configured;
``HERMES_SAFE_MODE=1`` skips registration; registration is idempotent.
"""
from __future__ import annotations
@@ -68,9 +47,7 @@ QUEUE_MAX_SIZE = 256
_registered: Set[Tuple[str, str, str]] = set()
_registered_lock = threading.Lock()
_delivery_queue: "queue.Queue[Optional[Dict[str, Any]]]" = queue.Queue(
maxsize=QUEUE_MAX_SIZE
)
_delivery_queue: "queue.Queue[Optional[Dict[str, Any]]]" = queue.Queue(maxsize=QUEUE_MAX_SIZE)
_worker_lock = threading.Lock()
_worker: Optional[threading.Thread] = None
@@ -78,9 +55,7 @@ _worker: Optional[threading.Thread] = None
@dataclass
class WebhookTarget(_ToolMatcherMixin):
"""Parsed and validated representation of one ``hooks.outbound`` entry."""
_MATCHER_KIND = "outbound webhook"
url: str
events: List[str]
name: str = ""
@@ -97,29 +72,20 @@ class WebhookTarget(_ToolMatcherMixin):
# --- Public API -----------------------------------------------------------------
def register_from_config(cfg: Optional[Dict[str, Any]]) -> List[WebhookTarget]:
"""Register every configured outbound webhook on the plugin manager.
Malformed ``hooks.outbound`` means zero targets — never raises. Returns the
targets that ended up wired (deduplicated across repeat calls).
"""
"""Register every configured outbound webhook on the plugin manager. Malformed ``hooks.outbound``
means zero targets — never raises. Returns the targets that ended up wired (deduplicated)."""
if not isinstance(cfg, dict):
return []
from utils import env_var_enabled
if env_var_enabled("HERMES_SAFE_MODE"):
logger.info("HERMES_SAFE_MODE=1 — outbound webhook registration skipped")
return []
targets = iter_configured_targets(cfg)
if not targets:
return []
from hermes_cli.plugins import get_plugin_manager
manager = get_plugin_manager()
home_key = _home_key()
registered: List[WebhookTarget] = []
with _registered_lock:
for target in targets:
@@ -128,19 +94,15 @@ def register_from_config(cfg: Optional[Dict[str, Any]]) -> List[WebhookTarget]:
key = (home_key, event, target.url)
if key in _registered:
continue
manager._hooks.setdefault(event, []).append(
_make_callback(event, target)
)
manager._hooks.setdefault(event, []).append(_make_callback(event, target))
_registered.add(key)
wired_any = True
logger.info(
"outbound webhook registered: %s -> %s (matcher=%s, "
"timeout=%ds)",
"outbound webhook registered: %s -> %s (matcher=%s, timeout=%ds)",
event, target.label, target.matcher, target.timeout,
)
if wired_any:
registered.append(target)
return registered
@@ -149,9 +111,14 @@ def iter_configured_targets(cfg: Optional[Dict[str, Any]]) -> List[WebhookTarget
if not isinstance(cfg, dict):
return []
hooks_cfg = cfg.get("hooks")
return _parse_outbound_block(
hooks_cfg.get("outbound") if isinstance(hooks_cfg, dict) else None
)
raw = hooks_cfg.get("outbound") if isinstance(hooks_cfg, dict) else None
if raw is None:
return []
if not isinstance(raw, list):
logger.warning("hooks.outbound must be a list of webhook targets; got %s", type(raw).__name__)
return []
targets = (_parse_single_target(i, entry) for i, entry in enumerate(raw))
return [t for t in targets if t is not None]
def flush(timeout: float = 5.0) -> bool:
@@ -167,13 +134,10 @@ def flush(timeout: float = 5.0) -> bool:
def re_register_config_hooks() -> None:
"""Re-register outbound webhooks after a plugin force-reload cleared ``_hooks``.
Only the current home's idempotence keys are cleared so a force-reload in one
profile cannot invalidate another profile's still-live registration.
"""
"""Re-register outbound webhooks after a plugin force-reload cleared ``_hooks``. Only the
current home's idempotence keys are cleared so a force-reload in one profile cannot
invalidate another profile's still-live registration."""
from hermes_cli.config import load_config
_forget_home_registrations(_registered, _registered_lock)
register_from_config(load_config())
@@ -192,19 +156,6 @@ def reset_for_tests() -> None:
# --- Config parsing -------------------------------------------------------------
def _parse_outbound_block(raw: Any) -> List[WebhookTarget]:
if raw is None:
return []
if not isinstance(raw, list):
logger.warning(
"hooks.outbound must be a list of webhook targets; got %s",
type(raw).__name__,
)
return []
targets = (_parse_single_target(i, entry) for i, entry in enumerate(raw))
return [t for t in targets if t is not None]
def _parse_single_target(index: int, raw: Any) -> Optional[WebhookTarget]:
from hermes_cli.plugins import VALID_HOOKS
@@ -214,7 +165,6 @@ def _parse_single_target(index: int, raw: Any) -> Optional[WebhookTarget]:
if not isinstance(raw, dict):
warn(" must be a mapping with 'url' and 'events' keys; got %s", type(raw).__name__)
return None
url = raw.get("url")
if not isinstance(url, str) or not url.strip():
warn(" is missing a non-empty 'url'")
@@ -225,22 +175,18 @@ def _parse_single_target(index: int, raw: Any) -> Optional[WebhookTarget]:
return None
if url.lower().startswith("http://"):
warn(".url uses plain http:// — payloads (including tool inputs) travel unencrypted. Prefer https.")
events_raw = raw.get("events")
valid_list = ", ".join(sorted(VALID_HOOKS))
if not isinstance(events_raw, list) or not events_raw:
warn(" needs a non-empty 'events' list (valid: %s)", valid_list)
return None
events: List[str] = []
events: List[str] = [ev for ev in events_raw if ev in VALID_HOOKS]
for ev in events_raw:
if ev in VALID_HOOKS:
events.append(ev)
else:
if ev not in VALID_HOOKS:
warn(": unknown event %r ignored (valid: %s)", ev, valid_list)
if not events:
warn(" has no valid events — skipped")
return None
matcher = raw.get("matcher")
if matcher is not None and not isinstance(matcher, str):
warn(".matcher must be a string regex; ignoring")
@@ -248,23 +194,16 @@ def _parse_single_target(index: int, raw: Any) -> Optional[WebhookTarget]:
if matcher is not None and not any(e in _TOOL_SCOPED_EVENTS for e in events):
warn(".matcher=%r will be ignored — matcher is only honored for pre_tool_call / post_tool_call.", matcher)
matcher = None
timeout_raw = raw.get("timeout", DEFAULT_TIMEOUT_SECONDS)
try:
timeout = int(timeout_raw)
except (TypeError, ValueError):
warn(".timeout must be an int (got %r); using default %ds", timeout_raw, DEFAULT_TIMEOUT_SECONDS)
timeout = DEFAULT_TIMEOUT_SECONDS
timeout = max(1, min(timeout, MAX_TIMEOUT_SECONDS))
name = raw.get("name")
return WebhookTarget(
url=url,
events=events,
name=name.strip() if isinstance(name, str) else "",
secret=_resolve_secret(index, raw),
matcher=matcher,
timeout=timeout,
url=url, events=events, name=name.strip() if isinstance(name, str) else "", secret=_resolve_secret(index, raw),
matcher=matcher, timeout=max(1, min(timeout, MAX_TIMEOUT_SECONDS)),
)
@@ -276,8 +215,8 @@ def _resolve_secret(index: int, raw: Dict[str, Any]) -> Optional[str]:
if value:
return value
logger.warning(
"hooks.outbound[%d].secret_env=%r is not set in the environment "
"— deliveries will be UNSIGNED", index, secret_env.strip(),
"hooks.outbound[%d].secret_env=%r is not set in the environment — deliveries will be UNSIGNED",
index, secret_env.strip(),
)
return None
secret = raw.get("secret")
@@ -297,8 +236,7 @@ def _make_callback(event: str, target: WebhookTarget):
body = _serialize_payload(event, kwargs, delivery_id)
except Exception: # a bad payload must not hurt the loop
logger.warning(
"outbound webhook payload serialization failed (event=%s "
"target=%s)", event, target.label, exc_info=True,
"outbound webhook payload serialization failed (event=%s target=%s)", event, target.label, exc_info=True,
)
return None
_enqueue(_build_delivery(event, target, body, delivery_id))
@@ -309,48 +247,30 @@ def _make_callback(event: str, target: WebhookTarget):
return _callback
def _serialize_payload(
event: str, kwargs: Dict[str, Any], delivery_id: str,
) -> bytes:
"""Render the POST body: shell-hooks stdin shape plus delivery metadata.
``delivery_id`` (also the ``X-Hermes-Delivery`` header) and ``timestamp`` live
inside the HMAC-signed body, so they double as replay protection.
"""
def _serialize_payload(event: str, kwargs: Dict[str, Any], delivery_id: str) -> bytes:
"""Render the POST body: shell-hooks stdin shape plus delivery metadata. ``delivery_id``
(also the ``X-Hermes-Delivery`` header) and ``timestamp`` live inside the HMAC-signed
body, so they double as replay protection."""
# Profile resolved at fire time so a multiplexed gateway's receivers can tell which profile emitted.
from hermes_cli.profiles import get_active_profile_name
payload = {
"hook_event_name": event,
"profile": get_active_profile_name(),
**_payload_fields(kwargs),
"delivery_id": delivery_id,
"timestamp": _utc_now_iso(),
"hook_event_name": event, "profile": get_active_profile_name(), **_payload_fields(kwargs),
"delivery_id": delivery_id, "timestamp": _utc_now_iso(),
}
return json.dumps(payload, ensure_ascii=False, default=str).encode("utf-8")
def _build_delivery(
event: str, target: WebhookTarget, body: bytes, delivery_id: str,
) -> Dict[str, Any]:
def _build_delivery(event: str, target: WebhookTarget, body: bytes, delivery_id: str) -> Dict[str, Any]:
headers = {
"Content-Type": "application/json",
"User-Agent": "Hermes-Agent-Outbound-Webhook",
"X-Hermes-Event": event,
"X-Hermes-Delivery": delivery_id,
"Content-Type": "application/json", "User-Agent": "Hermes-Agent-Outbound-Webhook",
"X-Hermes-Event": event, "X-Hermes-Delivery": delivery_id,
}
if target.secret:
digest = hmac.new(
target.secret.encode("utf-8"), body, hashlib.sha256
).hexdigest()
digest = hmac.new(target.secret.encode("utf-8"), body, hashlib.sha256).hexdigest()
headers["X-Hermes-Signature-256"] = f"sha256={digest}"
return {
"url": target.url,
"label": target.label,
"event": event,
"body": body,
"headers": headers,
"timeout": target.timeout,
"url": target.url, "label": target.label, "event": event,
"body": body, "headers": headers, "timeout": target.timeout,
}
@@ -360,8 +280,8 @@ def _enqueue(delivery: Dict[str, Any]) -> None:
_delivery_queue.put_nowait(delivery)
except queue.Full:
logger.warning(
"outbound webhook queue full (%d pending) — dropping %s event "
"for %s", QUEUE_MAX_SIZE, delivery["event"], delivery["label"],
"outbound webhook queue full (%d pending) — dropping %s event for %s",
QUEUE_MAX_SIZE, delivery["event"], delivery["label"],
)
@@ -372,13 +292,10 @@ def _ensure_worker() -> None:
with _worker_lock:
if _worker is not None and _worker.is_alive():
return
_worker = threading.Thread(
target=_worker_loop, name="outbound-webhooks", daemon=True,
)
_worker = threading.Thread(target=_worker_loop, name="outbound-webhooks", daemon=True)
_worker.start()
# Daemon worker: a short-lived process could exit right after enqueuing
# on_session_end. Drain at interpreter shutdown, bounded so a dead
# endpoint can only delay exit, never hang it.
# Daemon worker: a short-lived process could exit right after enqueuing on_session_end.
# Drain at interpreter shutdown, bounded so a dead endpoint can only delay exit, never hang it.
atexit.register(flush, timeout=5.0)
@@ -411,49 +328,33 @@ _opener = urlrequest.build_opener(_NoRedirectHandler)
def _deliver(delivery: Dict[str, Any]) -> None:
"""POST with bounded retries: retry on connection errors and 5xx; 4xx and 3xx are final."""
event, label = delivery["event"], delivery["label"]
last_error = ""
for attempt in range(1, MAX_DELIVERY_ATTEMPTS + 1):
req = urlrequest.Request(
delivery["url"],
data=delivery["body"],
headers=delivery["headers"],
method="POST",
)
req = urlrequest.Request(delivery["url"], data=delivery["body"], headers=delivery["headers"], method="POST")
try:
with _opener.open(req, timeout=delivery["timeout"]) as resp:
status = getattr(resp, "status", 200)
if 200 <= status < 300:
logger.debug(
"outbound webhook delivered: %s -> %s (HTTP %d)",
delivery["event"], delivery["label"], status,
)
logger.debug("outbound webhook delivered: %s -> %s (HTTP %d)", event, label, status)
return
last_error = f"HTTP {status}"
except urlerror.HTTPError as exc:
last_error = f"HTTP {exc.code}"
if 300 <= exc.code < 400:
logger.warning(
"outbound webhook target redirected (event=%s target=%s): "
"%s -> %s — redirects are not followed; update the "
"configured url", delivery["event"], delivery["label"],
last_error, exc.headers.get("Location", "?"),
"outbound webhook target redirected (event=%s target=%s): %s -> %s — redirects are not "
"followed; update the configured url", event, label, last_error, exc.headers.get("Location", "?"),
)
return
if 400 <= exc.code < 500:
logger.warning(
"outbound webhook rejected (event=%s target=%s): %s — "
"not retrying", delivery["event"], delivery["label"],
last_error,
)
logger.warning("outbound webhook rejected (event=%s target=%s): %s — not retrying", event, label, last_error)
return
except Exception as exc:
last_error = str(exc) or type(exc).__name__
if attempt < MAX_DELIVERY_ATTEMPTS:
time.sleep(RETRY_BACKOFF_SECONDS * attempt)
logger.warning(
"outbound webhook delivery failed after %d attempt(s) (event=%s "
"target=%s): %s",
MAX_DELIVERY_ATTEMPTS, delivery["event"], delivery["label"], last_error,
"outbound webhook delivery failed after %d attempt(s) (event=%s target=%s): %s",
MAX_DELIVERY_ATTEMPTS, event, label, last_error,
)
agent/proxy_sources/iron_proxy.py
+231 -481
View File
File diff suppressed because it is too large Load Diff