feat(kanban,mcp): orphaned-card reconciliation + per-server MCP identity header

Two small config-gated features:

1. Kanban orphaned-card reconciliation (kanban.reconcile_orphans, default
   true, config.yaml): a running card with broken claim bookkeeping
   (claim_lock or claim_expires NULL — crash mid-claim, manual SQL, DB
   restore) is invisible to all existing recovery paths
   (release_stale_claims requires claim_expires NOT NULL,
   detect_crashed_workers requires host-local lock + pid,
   detect_stale_running is config-disabled by default) and shows Running
   forever. New reconcile_orphaned_running() pass in kanban_db.py runs
   each dispatch_once tick: requeues orphans to ready with an explanatory
   comment, closes any leaked run, emits a 'reconciled' event, and defers
   when the recorded PID is still alive on this host (never requeue
   beside a live worker). Surfaced via DispatchResult.reconciled_orphans.

2. Per-server MCP identity header (mcp_servers.<name>.identity_header,
   config.yaml): optional {name, value_from: static|profile, value}
   mapping; the header is attached to that server's HTTP/SSE transport
   requests. 'static' sends the config value; 'profile' resolves the
   active Hermes profile name once at connect time (no per-call
   mutation). Explicit per-server headers of the same name (any casing)
   win. Invalid blocks warn-and-ignore; stdio servers warn-and-ignore.

Tests: tests/gateway/test_kanban_reconcile_orphans.py (9),
tests/tools/test_mcp_identity_header.py (13), all written first (RED)
then implemented (GREEN). No new HERMES_* env vars.

Inspired by: openai/symphony tracker reconciliation (Apache-2.0) +
Poke per-user MCP identity (idea-level).

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):

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.

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,
     )

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"

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
+        )

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

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. |

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. |