diff --git a/gateway/kanban_watchers.py b/gateway/kanban_watchers.py index cad31e993e3fa..17b65e1110f5b 100644 --- a/gateway/kanban_watchers.py +++ b/gateway/kanban_watchers.py @@ -1111,6 +1111,12 @@ class GatewayKanbanWatchersMixin: ) stale_timeout_seconds = 0 + # kanban.reconcile_orphans (config.yaml, default true): each tick, + # requeue 'running' cards whose claim bookkeeping is broken (no + # valid claim, dead/gone worker) — the zombie-card reconciliation + # pass. Set false to keep orphans frozen for manual forensics. + reconcile_orphans = bool(kanban_cfg.get("reconcile_orphans", True)) + # Read kanban.default_assignee — fallback profile for tasks # created without an explicit assignee (e.g. via the dashboard). # When set, the dispatcher applies it to unassigned ready tasks @@ -1247,6 +1253,7 @@ class GatewayKanbanWatchersMixin: stale_timeout_seconds=stale_timeout_seconds, default_assignee=default_assignee, max_in_progress_per_profile=max_in_progress_per_profile, + reconcile_orphans=reconcile_orphans, ) except sqlite3.DatabaseError as exc: if _is_corrupt_board_db_error(exc): diff --git a/hermes_cli/config_defaults.py b/hermes_cli/config_defaults.py index 67e6d3f9801d4..2c557f3cfc96e 100644 --- a/hermes_cli/config_defaults.py +++ b/hermes_cli/config_defaults.py @@ -2354,6 +2354,12 @@ DEFAULT_CONFIG = { # worker process (if still running host-locally) is terminated # before the reclaim. 0 disables stale detection entirely. "dispatch_stale_timeout_seconds": 14400, + # Orphaned-card reconciliation: each dispatcher tick, requeue + # 'running' cards whose claim bookkeeping is broken (claim_lock or + # claim_expires NULL with a dead/gone worker) — zombies invisible + # to the TTL/crash/stale recovery paths. Set false to keep orphans + # frozen for manual forensics. + "reconcile_orphans": True, }, # execute_code settings — controls the tool used for programmatic tool calls. diff --git a/hermes_cli/kanban_db.py b/hermes_cli/kanban_db.py index 113e34842ec54..c4bb7caf9482b 100644 --- a/hermes_cli/kanban_db.py +++ b/hermes_cli/kanban_db.py @@ -6794,6 +6794,10 @@ class DispatchResult: reclaimed: int = 0 promoted: int = 0 + reconciled_orphans: list[str] = field(default_factory=list) + """Task ids requeued by :func:`reconcile_orphaned_running` this tick — + ``running`` cards whose claim bookkeeping was broken (no valid claim, + dead/gone worker). See the reconciliation pass for details.""" spawned: list[tuple[str, str, str]] = field(default_factory=list) """List of ``(task_id, assignee, workspace_path)`` triples.""" skipped_unassigned: list[str] = field(default_factory=list) @@ -7433,6 +7437,96 @@ def detect_stale_running( return reclaimed +def reconcile_orphaned_running( + conn: sqlite3.Connection, +) -> list[str]: + """Reconcile ``running`` cards whose claim bookkeeping is broken. + + Tracked-state vs. reality divergence: a task can sit in + ``status='running'`` with ``claim_lock IS NULL`` or ``claim_expires IS + NULL`` (crash mid-claim, manual SQL, DB restore). None of the other + recovery paths ever touch such a card — ``release_stale_claims`` + requires a non-NULL ``claim_expires``, ``detect_crashed_workers`` + requires a host-local claim_lock + worker_pid, and + ``detect_stale_running`` is disabled by default — so the card shows + Running forever (a zombie). + + This pass finds those orphans, requeues them to ``ready`` with an + explanatory comment, closes any leaked run, and appends a + ``reconciled`` event. If the orphan row still records a live PID on + this host, requeueing is deferred to a later tick so we never spawn a + duplicate beside a possibly-alive worker. + + Returns the list of reconciled task ids. Safe to call every tick. + + Idea from openai/symphony's tracker reconciliation (Apache-2.0). + """ + now = int(time.time()) + reconciled: list[str] = [] + rows = conn.execute( + "SELECT id, claim_lock, claim_expires, worker_pid FROM tasks " + "WHERE status = 'running' " + " AND (claim_lock IS NULL OR claim_expires IS NULL)" + ).fetchall() + for row in rows: + tid = row["id"] + pid = row["worker_pid"] + if pid and _pid_alive(pid): + # The recorded worker may still be doing real work — never + # requeue beside a live process. Retry next tick. + _log.debug( + "kanban reconcile: task %s has broken claim bookkeeping but " + "pid %s is alive on this host — deferring", tid, pid, + ) + continue + with write_txn(conn): + cur = conn.execute( + "UPDATE tasks SET status = 'ready', claim_lock = NULL, " + "claim_expires = NULL, worker_pid = NULL, " + "last_heartbeat_at = NULL " + "WHERE id = ? AND status = 'running' " + " AND claim_lock IS ? AND claim_expires IS ?", + (tid, row["claim_lock"], row["claim_expires"]), + ) + if cur.rowcount != 1: + continue + payload = { + "reason": "orphaned_running", + "claim_lock": row["claim_lock"], + "claim_expires": ( + int(row["claim_expires"]) + if row["claim_expires"] is not None else None + ), + "worker_pid": int(pid) if pid else None, + "now": now, + } + run_id = _end_run( + conn, tid, + outcome="reclaimed", status="reclaimed", + error="orphaned running card (broken claim bookkeeping)", + metadata=payload, + ) + # Inline comment INSERT — add_comment opens its own write_txn + # and would raise on nesting (see write_txn pitfalls). + conn.execute( + "INSERT INTO task_comments (task_id, author, body, created_at) " + "VALUES (?, ?, ?, ?)", + ( + tid, "dispatcher", + "reconciliation: card was 'running' with no valid claim " + "(dead/gone worker) — requeued to ready", + now, + ), + ) + _append_event(conn, tid, "reconciled", payload, run_id=run_id) + reconciled.append(tid) + _log.info( + "kanban reconcile: requeued orphaned running task %s " + "(claim_lock=%r, worker_pid=%r)", tid, row["claim_lock"], pid, + ) + return reconciled + + def _error_fingerprint(error_text: str) -> str: """Normalize an error message for grouping identical failures. @@ -8214,6 +8308,7 @@ def dispatch_once( board: Optional[str] = None, default_assignee: Optional[str] = None, max_in_progress_per_profile: Optional[int] = None, + reconcile_orphans: bool = True, ) -> DispatchResult: """Run one dispatcher tick under the board's single-writer lock. @@ -8248,6 +8343,7 @@ def dispatch_once( board=board, default_assignee=default_assignee, max_in_progress_per_profile=max_in_progress_per_profile, + reconcile_orphans=reconcile_orphans, ) with _dispatch_tick_lock(db_path) as held: if not held: @@ -8264,6 +8360,7 @@ def dispatch_once( board=board, default_assignee=default_assignee, max_in_progress_per_profile=max_in_progress_per_profile, + reconcile_orphans=reconcile_orphans, ) # Still under the dispatch lock: opportunistically truncate the WAL # at a coarse interval so it cannot grow unbounded between restarts. @@ -8284,6 +8381,7 @@ def _dispatch_once_locked( board: Optional[str] = None, default_assignee: Optional[str] = None, max_in_progress_per_profile: Optional[int] = None, + reconcile_orphans: bool = True, ) -> DispatchResult: """Run one dispatcher tick. @@ -8319,6 +8417,11 @@ def _dispatch_once_locked( result = DispatchResult() result.reclaimed = release_stale_claims(conn) + if reconcile_orphans: + # Orphaned-card reconciliation: requeue 'running' cards whose claim + # bookkeeping is broken (no valid claim, dead/gone worker) that the + # TTL/crash/stale paths can never see. See reconcile_orphaned_running. + result.reconciled_orphans = reconcile_orphaned_running(conn) result.stale = detect_stale_running( conn, stale_timeout_seconds=stale_timeout_seconds, ) diff --git a/tests/gateway/test_kanban_reconcile_orphans.py b/tests/gateway/test_kanban_reconcile_orphans.py new file mode 100644 index 0000000000000..27c33c61f898b --- /dev/null +++ b/tests/gateway/test_kanban_reconcile_orphans.py @@ -0,0 +1,178 @@ +"""Tests: orphaned-card reconciliation for the kanban dispatcher. + +Tracked-state vs. reality divergence: a task can sit in ``status='running'`` +with broken claim bookkeeping — ``claim_lock IS NULL`` or ``claim_expires IS +NULL`` (crash mid-claim, manual SQL, DB restore, partial migration). None of +the existing recovery paths ever touch such a card: + +- ``release_stale_claims`` requires ``claim_expires IS NOT NULL``; +- ``detect_crashed_workers`` requires a host-local ``claim_lock`` prefix and + a recorded ``worker_pid``; +- ``detect_stale_running`` is disabled by default (``stale_timeout=0``). + +Result: a zombie card that shows Running forever. ``reconcile_orphaned_running`` +is the reconciliation pass: it finds those orphans, requeues them to ``ready`` +with an explanatory note, and logs a ``reconciled`` event. Wired into +``dispatch_once`` each tick, gated by ``kanban.reconcile_orphans`` (config.yaml, +default on) at the gateway watcher layer. + +Inspired by openai/symphony's tracker reconciliation (Apache-2.0), idea-level. +""" + +from __future__ import annotations + +import subprocess +from pathlib import Path + +import pytest + +from hermes_cli import kanban_db as kb + + +@pytest.fixture +def kanban_home(tmp_path, monkeypatch): + home = tmp_path / ".hermes" + home.mkdir() + monkeypatch.setenv("HERMES_HOME", str(home)) + monkeypatch.setenv("HERMES_KANBAN_HOME", str(home)) + monkeypatch.setenv("HERMES_KANBAN_CRASH_GRACE_SECONDS", "0") + monkeypatch.setattr(Path, "home", lambda: tmp_path) + db_path = kb.kanban_db_path(board="default") + kb._INITIALIZED_PATHS.discard(str(db_path.resolve())) + kb.init_db() + return home + + +@pytest.fixture +def conn(kanban_home): + with kb.connect() as c: + yield c + + +def _orphan_running(conn, tid, *, claim_lock=None, claim_expires=None, + worker_pid=None): + """Force a task into running with (partially) broken claim bookkeeping.""" + conn.execute( + "UPDATE tasks SET status='running', claim_lock=?, claim_expires=?, " + "worker_pid=? WHERE id=?", + (claim_lock, claim_expires, worker_pid, tid), + ) + conn.commit() + + +class TestReconcileOrphanedRunning: + def test_null_claim_lock_orphan_requeued(self, conn): + """running + claim_lock NULL → requeued to ready with a note.""" + tid = kb.create_task(conn, title="zombie", assignee="w") + _orphan_running(conn, tid) + + reconciled = kb.reconcile_orphaned_running(conn) + + assert reconciled == [tid] + row = conn.execute( + "SELECT status, claim_lock, claim_expires, worker_pid " + "FROM tasks WHERE id=?", (tid,), + ).fetchone() + assert row["status"] == "ready" + assert row["claim_lock"] is None + assert row["claim_expires"] is None + assert row["worker_pid"] is None + + def test_null_claim_expires_orphan_requeued(self, conn): + """running + claim_lock set but claim_expires NULL is also invisible + to release_stale_claims — reconciliation must catch it.""" + host = kb._claimer_id().split(":", 1)[0] + tid = kb.create_task(conn, title="half-claim", assignee="w") + _orphan_running(conn, tid, claim_lock=f"{host}:dead") + + reconciled = kb.reconcile_orphaned_running(conn) + + assert reconciled == [tid] + assert conn.execute( + "SELECT status FROM tasks WHERE id=?", (tid,) + ).fetchone()["status"] == "ready" + + def test_reconciled_event_and_note_logged(self, conn): + tid = kb.create_task(conn, title="zombie", assignee="w") + _orphan_running(conn, tid) + + kb.reconcile_orphaned_running(conn) + + events = kb.list_events(conn, tid) + recon = [e for e in events if e.kind == "reconciled"] + assert len(recon) == 1 + assert recon[0].payload["reason"] == "orphaned_running" + comments = kb.list_comments(conn, tid) + assert any("reconcil" in (c.body or "").lower() for c in comments) + + def test_healthy_running_task_untouched(self, conn): + """A properly claimed running task is NOT an orphan.""" + tid = kb.create_task(conn, title="healthy", assignee="w") + kb.claim_task(conn, tid) + + assert kb.reconcile_orphaned_running(conn) == [] + assert conn.execute( + "SELECT status FROM tasks WHERE id=?", (tid,) + ).fetchone()["status"] == "running" + + def test_live_worker_pid_defers_reconcile(self, conn): + """If the orphan row still records a live PID on this host, don't + requeue beside a possibly-alive worker — defer to the next tick.""" + tid = kb.create_task(conn, title="maybe-alive", assignee="w") + sleeper = subprocess.Popen(["sleep", "30"]) + try: + _orphan_running(conn, tid, worker_pid=sleeper.pid) + assert kb.reconcile_orphaned_running(conn) == [] + assert conn.execute( + "SELECT status FROM tasks WHERE id=?", (tid,) + ).fetchone()["status"] == "running" + finally: + sleeper.terminate() + sleeper.wait() + + def test_dead_worker_pid_orphan_requeued(self, conn): + """Orphan with a recorded but dead PID is reconciled.""" + tid = kb.create_task(conn, title="dead-pid", assignee="w") + dead = subprocess.Popen(["true"]) + dead.wait() + _orphan_running(conn, tid, worker_pid=dead.pid) + + assert kb.reconcile_orphaned_running(conn) == [tid] + + def test_non_running_statuses_ignored(self, conn): + for status in ("todo", "ready", "blocked", "done"): + tid = kb.create_task(conn, title=f"s-{status}", assignee="w") + conn.execute( + "UPDATE tasks SET status=?, claim_lock=NULL, " + "claim_expires=NULL WHERE id=?", (status, tid), + ) + conn.commit() + assert kb.reconcile_orphaned_running(conn) == [] + + +class TestDispatchOnceReconciles: + def test_dispatch_once_reconciles_orphans(self, conn): + tid = kb.create_task(conn, title="zombie", assignee="w") + _orphan_running(conn, tid) + + result = kb.dispatch_once(conn, spawn_fn=lambda *a, **k: (True, ""), + dry_run=True) + + assert tid in result.reconciled_orphans + assert conn.execute( + "SELECT status FROM tasks WHERE id=?", (tid,) + ).fetchone()["status"] == "ready" + + def test_dispatch_once_reconcile_can_be_disabled(self, conn): + """kanban.reconcile_orphans=false plumbs through as + reconcile_orphans=False and skips the pass.""" + tid = kb.create_task(conn, title="zombie", assignee="w") + _orphan_running(conn, tid) + + result = kb.dispatch_once(conn, spawn_fn=lambda *a, **k: (True, ""), + dry_run=True, reconcile_orphans=False) + + assert result.reconciled_orphans == [] + assert conn.execute( + "SELECT status FROM tasks WHERE id=?", (tid,) + ).fetchone()["status"] == "running" diff --git a/tests/tools/test_mcp_identity_header.py b/tests/tools/test_mcp_identity_header.py new file mode 100644 index 0000000000000..ac197ad59acf5 --- /dev/null +++ b/tests/tools/test_mcp_identity_header.py @@ -0,0 +1,283 @@ +"""Tests for the per-server MCP identity header (``identity_header``). + +An optional per-server config key in ``mcp_servers`` attaches a static or +profile-derived identity header to that server's HTTP/SSE transport +requests: + + mcp_servers: + remote_api: + url: "https://my-mcp-server.example.com/mcp" + identity_header: + name: "X-User-Id" + value_from: "static" # or "profile" + value: "alice" # required for value_from: static + +Covers: + +1. ``_resolve_identity_header`` helper — static mode, profile mode, + validation failures (warn + ignore, never break the server). + +2. HTTP (new SDK ``streamable_http_client``) path attaches the header to + the user-owned ``httpx.AsyncClient`` when configured, and not otherwise. + +3. Explicit per-server ``headers`` with the same name win over the + identity header (no silent override of user config). + +4. stdio servers: ``identity_header`` is warn-and-ignore (headers don't + exist on stdio transports). +""" + +from __future__ import annotations + +import asyncio +import logging +from unittest.mock import MagicMock, patch + +import pytest + + +# --------------------------------------------------------------------------- +# _resolve_identity_header helper +# --------------------------------------------------------------------------- + + +class TestResolveIdentityHeader: + def test_returns_none_when_unset(self): + from tools.mcp_tool import _resolve_identity_header + + assert _resolve_identity_header("srv", {}) is None + assert _resolve_identity_header("srv", {"url": "https://x"}) is None + + def test_static_mode_returns_name_value(self): + from tools.mcp_tool import _resolve_identity_header + + result = _resolve_identity_header("srv", { + "identity_header": { + "name": "X-User-Id", + "value_from": "static", + "value": "alice", + }, + }) + assert result == ("X-User-Id", "alice") + + def test_static_is_default_value_from(self): + from tools.mcp_tool import _resolve_identity_header + + result = _resolve_identity_header("srv", { + "identity_header": {"name": "X-User-Id", "value": "bob"}, + }) + assert result == ("X-User-Id", "bob") + + def test_profile_mode_uses_active_profile_name(self): + from tools.mcp_tool import _resolve_identity_header + + with patch( + "hermes_cli.profiles.get_active_profile_name", + return_value="workbot", + ): + result = _resolve_identity_header("srv", { + "identity_header": { + "name": "X-Hermes-Profile", + "value_from": "profile", + }, + }) + assert result == ("X-Hermes-Profile", "workbot") + + def test_missing_name_warns_and_returns_none(self, caplog): + from tools.mcp_tool import _resolve_identity_header + + with caplog.at_level(logging.WARNING): + result = _resolve_identity_header("srv", { + "identity_header": {"value": "alice"}, + }) + assert result is None + assert any("identity_header" in r.message for r in caplog.records) + + def test_static_missing_value_warns_and_returns_none(self, caplog): + from tools.mcp_tool import _resolve_identity_header + + with caplog.at_level(logging.WARNING): + result = _resolve_identity_header("srv", { + "identity_header": {"name": "X-User-Id"}, + }) + assert result is None + assert any("identity_header" in r.message for r in caplog.records) + + def test_unknown_value_from_warns_and_returns_none(self, caplog): + from tools.mcp_tool import _resolve_identity_header + + with caplog.at_level(logging.WARNING): + result = _resolve_identity_header("srv", { + "identity_header": { + "name": "X-User-Id", + "value_from": "per_call", + "value": "x", + }, + }) + assert result is None + assert any("identity_header" in r.message for r in caplog.records) + + def test_non_dict_config_warns_and_returns_none(self, caplog): + from tools.mcp_tool import _resolve_identity_header + + with caplog.at_level(logging.WARNING): + result = _resolve_identity_header("srv", { + "identity_header": "X-User-Id: alice", + }) + assert result is None + assert any("identity_header" in r.message for r in caplog.records) + + +# --------------------------------------------------------------------------- +# HTTP transport — header attached to httpx.AsyncClient +# --------------------------------------------------------------------------- + + +def _drive_http(server, config): + """Run ``_run_http`` with the SDK boundary mocked out, capturing the + kwargs passed to ``httpx.AsyncClient``. Mirrors the pattern in + ``test_mcp_client_cert.py``. + """ + from tools.mcp_tool import MCPServerTask + + captured: dict = {} + + class DummyAsyncClient: + def __init__(self, **kwargs): + captured.update(kwargs) + + async def __aenter__(self): + return self + + async def __aexit__(self, *a): + return False + + class DummyTransportCtx: + async def __aenter__(self): + return MagicMock(), MagicMock(), (lambda: None) + + async def __aexit__(self, *a): + return False + + class DummySession: + def __init__(self, *args, **kwargs): + pass + + async def __aenter__(self): + return self + + async def __aexit__(self, *a): + return False + + async def initialize(self): + return None + + async def _discover_tools(self): + self._shutdown_event.set() + + async def _drive(): + with patch("tools.mcp_tool._MCP_HTTP_AVAILABLE", True), \ + patch("tools.mcp_tool._MCP_NEW_HTTP", True), \ + patch("httpx.AsyncClient", DummyAsyncClient), \ + patch("tools.mcp_tool.streamable_http_client", + return_value=DummyTransportCtx()), \ + patch("tools.mcp_tool.ClientSession", DummySession), \ + patch.object(MCPServerTask, "_discover_tools", _discover_tools): + await server._run_http(config) + + asyncio.run(_drive()) + return captured + + +class TestHTTPIdentityHeader: + def test_header_attached_when_configured(self): + from tools.mcp_tool import MCPServerTask + + server = MCPServerTask("remote") + captured = _drive_http(server, { + "url": "https://example.com/mcp", + "identity_header": { + "name": "X-User-Id", + "value": "alice", + }, + }) + headers = captured.get("headers") or {} + assert headers.get("X-User-Id") == "alice" + + def test_header_absent_when_not_configured(self): + from tools.mcp_tool import MCPServerTask + + server = MCPServerTask("remote") + captured = _drive_http(server, { + "url": "https://example.com/mcp", + }) + headers = captured.get("headers") or {} + assert not any(k.lower() == "x-user-id" for k in headers) + + def test_explicit_header_with_same_name_wins(self): + """A user-set per-server header of the same name (any casing) is + not overridden by the identity header.""" + from tools.mcp_tool import MCPServerTask + + server = MCPServerTask("remote") + captured = _drive_http(server, { + "url": "https://example.com/mcp", + "headers": {"x-user-id": "explicit-wins"}, + "identity_header": { + "name": "X-User-Id", + "value": "alice", + }, + }) + headers = captured.get("headers") or {} + assert headers.get("x-user-id") == "explicit-wins" + assert "X-User-Id" not in headers + + def test_profile_mode_header_attached(self): + from tools.mcp_tool import MCPServerTask + + server = MCPServerTask("remote") + with patch( + "hermes_cli.profiles.get_active_profile_name", + return_value="workbot", + ): + captured = _drive_http(server, { + "url": "https://example.com/mcp", + "identity_header": { + "name": "X-Hermes-Profile", + "value_from": "profile", + }, + }) + headers = captured.get("headers") or {} + assert headers.get("X-Hermes-Profile") == "workbot" + + +# --------------------------------------------------------------------------- +# stdio transport — identity_header is warn-and-ignore +# --------------------------------------------------------------------------- + + +class TestStdioIdentityHeader: + def test_stdio_warns_and_ignores(self, caplog): + """identity_header on a stdio server logs a warning and does not + break the transport path (headers don't exist on stdio).""" + from tools.mcp_tool import MCPServerTask + + server = MCPServerTask("local") + + async def _drive(): + # Force the SDK-unavailable fast path so no subprocess spawns; + # the warning must fire before the availability check. + with patch("tools.mcp_tool._MCP_AVAILABLE", False): + await server._run_stdio({ + "command": "echo", + "identity_header": {"name": "X-User-Id", "value": "a"}, + }) + + with caplog.at_level(logging.WARNING): + with pytest.raises(ImportError): + asyncio.run(_drive()) + + assert any( + "identity_header" in r.message and "stdio" in r.message + for r in caplog.records + ) diff --git a/tools/mcp_tool.py b/tools/mcp_tool.py index 993d9a13c80f5..9b0909c937a9a 100644 --- a/tools/mcp_tool.py +++ b/tools/mcp_tool.py @@ -37,6 +37,11 @@ Example config:: url: "https://my-mcp-server.example.com/mcp" headers: Authorization: "Bearer sk-..." + identity_header: # optional per-user identity header attached + name: "X-User-Id" # to this server's HTTP/SSE requests + value_from: "static" # "static" (default) or "profile" + value: "alice" # required for static; profile mode uses the + # active Hermes profile name timeout: 180 skip_preflight: true # bypass the content-type probe for a valid # Streamable HTTP endpoint that answers HEAD/GET @@ -1181,6 +1186,81 @@ def _resolve_client_cert(server_name: str, config: dict): return cert_path +def _resolve_identity_header(server_name: str, config: dict): + """Resolve the optional per-server ``identity_header`` config. + + Config shape (in the server's ``mcp_servers`` entry):: + + identity_header: + name: "X-User-Id" + value_from: "static" # or "profile"; default: static + value: "alice" # required when value_from is static + + Returns a ``(header_name, header_value)`` tuple, or ``None`` when the + key is unset or invalid. Invalid configs warn and are ignored — an + identity header must never break the server connection. ``profile`` + mode resolves the value to the active Hermes profile name once at + connect time; there is no per-call mutation. + """ + raw = config.get("identity_header") + if raw is None: + return None + if not isinstance(raw, dict): + logger.warning( + "MCP server '%s': identity_header must be a mapping with " + "'name' and 'value'/'value_from' keys (got %s) — ignoring", + server_name, type(raw).__name__, + ) + return None + name = raw.get("name") + if not isinstance(name, str) or not name.strip(): + logger.warning( + "MCP server '%s': identity_header requires a non-empty " + "'name' — ignoring", server_name, + ) + return None + value_from = (raw.get("value_from") or "static").strip().lower() + if value_from == "static": + value = raw.get("value") + if not isinstance(value, str) or not value.strip(): + logger.warning( + "MCP server '%s': identity_header with value_from: static " + "requires a non-empty string 'value' — ignoring", + server_name, + ) + return None + return (name.strip(), value) + if value_from == "profile": + from hermes_cli.profiles import get_active_profile_name + return (name.strip(), get_active_profile_name()) + logger.warning( + "MCP server '%s': identity_header value_from must be 'static' or " + "'profile' (got %r) — ignoring", server_name, value_from, + ) + return None + + +def _apply_identity_header(server_name: str, config: dict, headers: dict) -> dict: + """Merge the resolved identity header into ``headers`` (in place). + + An explicit per-server ``headers`` entry with the same name (any + casing) wins — the identity header never silently overrides user + config. + """ + resolved = _resolve_identity_header(server_name, config) + if resolved is None: + return headers + name, value = resolved + if any(key.lower() == name.lower() for key in headers): + logger.debug( + "MCP server '%s': identity_header '%s' already set via explicit " + "headers config — keeping the explicit value", server_name, name, + ) + return headers + headers[name] = value + return headers + + def _format_connect_error(exc: BaseException) -> str: """Render nested MCP connection errors into an actionable short message.""" @@ -2376,6 +2456,13 @@ class MCPServerTask: async def _run_stdio(self, config: dict): """Run the server using stdio transport.""" + if config.get("identity_header") is not None: + # Headers don't exist on stdio transports — warn and ignore so a + # copy-pasted HTTP config block doesn't silently mislead. + logger.warning( + "MCP server '%s': identity_header is only supported on " + "HTTP/SSE transports — ignored for stdio servers", self.name, + ) if not _MCP_AVAILABLE: raise ImportError( f"MCP server '{self.name}' requires the 'mcp' Python SDK, but " @@ -2760,6 +2847,9 @@ class MCPServerTask: url = config["url"] headers = dict(config.get("headers") or {}) + # Optional per-user identity header (config-gated; static or + # profile-derived). Explicit headers of the same name win. + headers = _apply_identity_header(self.name, config, headers) # Some MCP servers require MCP-Protocol-Version on the initial # initialize request and reject session-less POSTs otherwise. # Seed it as a client-level default, but treat user overrides as diff --git a/website/docs/user-guide/features/kanban.md b/website/docs/user-guide/features/kanban.md index b1f16cfc269ae..8e5859a3bed44 100644 --- a/website/docs/user-guide/features/kanban.md +++ b/website/docs/user-guide/features/kanban.md @@ -1023,6 +1023,7 @@ Every transition appends a row to `task_events`. Each row carries an optional `r | `crashed` | `{pid, claimer}` | Worker PID no longer alive but TTL hadn't expired yet. | | `timed_out` | `{pid, elapsed_seconds, limit_seconds, sigkill}` | `max_runtime_seconds` exceeded; dispatcher SIGTERM'd (then SIGKILL'd after 5 s grace) and re-queued. | | `stale` | `{elapsed_seconds, last_heartbeat_at, heartbeat_age_seconds, timeout_seconds, pid, terminated}` | Task ran longer than `kanban.dispatch_stale_timeout_seconds` (default 4 h) AND no `kanban_heartbeat` arrived in the last hour. Dispatcher SIGTERM'd the host-local worker (if any), reset the task to `ready` for re-dispatch. Does NOT tick the failure counter (stale is dispatcher-side absence detection, not a worker fault). Workers running long operations should call `kanban_heartbeat` at least once an hour to avoid this. | +| `reconciled` | `{reason, claim_lock, claim_expires, worker_pid}` | Orphaned-card reconciliation: the card was `running` with broken claim bookkeeping (`claim_lock` or `claim_expires` NULL — crash mid-claim, manual SQL, DB restore) and no live worker, so none of the TTL/crash/stale paths could ever recover it. The dispatcher requeued it to `ready` with an explanatory comment. Gated by `kanban.reconcile_orphans` in config.yaml (default `true`). | | `respawn_guarded` | `{reason}` | Dispatcher refused to re-spawn this ready task this tick. Reasons: `blocker_auth` (last failure was a quota/auth/429 error — wait for the rate window to reset), `recent_success` (a completed run happened in the last hour — wait for review before re-running), `active_pr` (a GitHub PR URL appears in a recent comment — a prior worker already opened a PR). The task stays in `ready`; the next tick gets another chance to spawn. If the underlying condition persists, the normal `consecutive_failures` circuit breaker will auto-block via `gave_up` after `failure_limit` failures. | | `spawn_failed` | `{error, failures}` | One spawn attempt failed (missing PATH, workspace unmountable, …). Counter increments; task returns to `ready` for retry. | | `protocol_violation` | `{pid, claimer, exit_code, protocol_violation}` | Worker exited successfully while the task was still `running`, usually because it answered without calling `kanban_complete` or `kanban_block`. Emitted on every violation (the payload's `protocol_violation: true` marker is copied into the run metadata and feeds the violation-only retry budget). Below the budget — up to `_PROTOCOL_VIOLATION_FAILURE_LIMIT` (default 3) *consecutive* violations, per-task `max_retries` overriding — the task simply returns to `ready` for another attempt; when the streak reaches the bound the dispatcher also emits `gave_up` and auto-blocks. | diff --git a/website/docs/user-guide/features/mcp.md b/website/docs/user-guide/features/mcp.md index e9902c7914c3c..f55e36211508d 100644 --- a/website/docs/user-guide/features/mcp.md +++ b/website/docs/user-guide/features/mcp.md @@ -312,6 +312,25 @@ mcp_servers: You can also keep the cert and key fully separate via `client_cert` (combined PEM) plus an explicit `client_key`. Paths support `~` expansion; a missing file raises a clear, server-scoped error rather than an opaque TLS handshake failure. +## Per-user identity header + +Remote HTTP/SSE MCP servers that key behavior on a caller identity (per-user rate limits, audit trails, multi-tenant routing) can be sent an identity header on every request via `identity_header`: + +```yaml +mcp_servers: + team_api: + url: "https://mcp.team.example.com/mcp" + identity_header: + name: "X-User-Id" + value_from: "static" # "static" (default) or "profile" + value: "alice" # required for static +``` + +- `value_from: static` sends the literal `value` from config.yaml. +- `value_from: profile` sends the active Hermes profile name, resolved once at connect time — useful when multiple profiles on one machine talk to the same server and it needs to tell them apart. + +An explicit entry in the server's `headers` mapping with the same name (any casing) always wins; the identity header never overrides your own header config. Invalid `identity_header` blocks are warned about and ignored — they never block the server from connecting. On stdio servers the key is ignored with a warning (stdio transports have no headers). + ## Basic configuration reference Hermes reads MCP config from `~/.hermes/config.yaml` under `mcp_servers`. @@ -327,6 +346,7 @@ Hermes reads MCP config from `~/.hermes/config.yaml` under `mcp_servers`. | `headers` | mapping | HTTP headers for remote servers | | `client_cert` | string \| list | Client certificate for mTLS — a combined PEM path, or `[cert, key]` / `[cert, key, password]` | | `client_key` | string | Client private-key PEM path (when separate from `client_cert`) | +| `identity_header` | mapping | Optional per-user identity header for HTTP/SSE servers — `{name, value_from: static\|profile, value}` | | `timeout` | number | Tool call timeout | | `connect_timeout` | number | Initial connection timeout (also bounds the MCP `initialize` handshake) | | `idle_timeout_seconds` | number | Recycle a stdio server after this many seconds without a tool call (`0` = never, default). The server restarts transparently on the next tool call. |