fix(cron): don't let a cron job inherit a kanban worker's dispatcher identity

A kanban worker that fires a cron job in-process no longer leaks its task
identity into the cron agent.

The worker is a normal `hermes chat -q` CLI agent whose default toolset
includes `cronjob`, running with HERMES_KANBAN_TASK legitimately set in its
own environment. `cronjob(action="run")` calls run_one_job() -> run_job()
in that same process, so the cron AIAgent was misidentified as that worker:
kanban toolset force-added, kanban-worker protocol injected into its system
prompt, and kanban_complete defaulting task_id to $HERMES_KANBAN_TASK --
letting an unrelated cron job close the worker's task and overwrite real
results.

Fixed with a ContextVar (`non_dispatcher_owned_context`), not by clearing
os.environ. The env is process-global and shared with three concurrent
readers that all need the real values:

  * the worker's own claim heartbeat -- run_agent._touch_activity ->
    heartbeat_current_worker_from_env reads TASK/CLAIM_LOCK/RUN_ID, and the
    cron-run heartbeat thread drives it every 10s. Clearing them silently
    no-ops the heartbeat, so after DEFAULT_CLAIM_TTL_SECONDS (15 min) the
    dispatcher reclaims a task whose worker is still alive and re-dispatches
    it -- the same duplicate-work failure from the other direction.
  * the gateway's kanban watchers, which do their own HERMES_KANBAN_BOARD
    save/restore around a slow decompose_task() LLM call.
  * concurrent cron jobs, which take a *shared* read lock
    (_terminal_cwd_lock.acquire_read) and so interleave: job A clears, job B
    snapshots empty, A restores, B clears and its restore no-ops -- the
    worker's identity is destroyed permanently.

`is_dispatcher_owned_worker_context()` is now the single predicate every
HERMES_KANBAN_* identity gate consults before trusting those vars. It also
closes a pre-existing gap in agent/skill_utils.py, which read the vars
without consulting the delegate_task ContextVar at all; the `kanban` verdict
additionally bypasses _ENV_DETECT_CACHE, since a context-dependent answer
must not be memoized process-wide.

HERMES_KANBAN_BOARD/DB/WORKSPACES_ROOT are left untouched, so the #20074
board pin and the dispatcher's path overrides keep working.

Tests: 18 new, including thread-isolation, concurrent-cron-jobs, and an AST
invariant over _default_spawn that fails if the dispatcher gains a var that
is neither identity-gated nor explicitly classified behaviour-only. All six
mutations are caught, including one that reintroduces the os.environ clear.
tests/cron/ + kanban suites 440 passed; model_tools/skill_utils/boards 63
passed; ruff clean.

Reported and diagnosed by Geoff Friesen (#78961), who identified the symptom
and the exact gating mechanism.

Co-authored-by: Geoff Friesen <gfriesen1@users.noreply.github.com>

agent/delegation_context.py

@@ -5,11 +5,16 @@ HERMES_KANBAN_* variables in process env. delegate_task children run inside the
 same Python process, but they are not dispatcher-owned Kanban workers. This
 module lets code paths that resolve tool schemas or spawn subprocesses fail
 closed for delegated children without mutating global os.environ for the parent.
+
+Cron jobs need the same treatment for the same reason: ``cronjob(action="run")``
+executes ``run_job()`` in-process, so a cron agent fired from inside a Kanban
+worker would otherwise inherit that worker's dispatcher identity.
+``non_dispatcher_owned_context()`` covers both cases.
 """
 from __future__ import annotations
 
 from contextlib import contextmanager
-from contextvars import ContextVar
+from contextvars import ContextVar, Token
 from typing import Iterator, Mapping, MutableMapping
 
 _DELEGATED_CHILD_CONTEXT: ContextVar[bool] = ContextVar(
@@ -17,6 +22,16 @@ _DELEGATED_CHILD_CONTEXT: ContextVar[bool] = ContextVar(
     default=False,
 )
 
+# Set for any in-process execution that is NOT the dispatcher-owned worker even
+# though the worker's HERMES_KANBAN_* vars are legitimately in os.environ (cron
+# jobs fired via the `cronjob` tool).  Kept separate from
+# _DELEGATED_CHILD_CONTEXT so the delegate_task-specific behaviour attached to
+# that flag (subprocess env scrubbing, its own error strings) is unchanged.
+_NON_DISPATCHER_OWNED_CONTEXT: ContextVar[bool] = ContextVar(
+    "hermes_non_dispatcher_owned_context",
+    default=False,
+)
+
 DELEGATED_CHILD_ENV_MARKER = "HERMES_DELEGATED_CHILD_CONTEXT"
 
 KANBAN_ENV_KEYS: tuple[str, ...] = (
@@ -55,6 +70,57 @@ def is_delegated_child_context() -> bool:
     return bool(_DELEGATED_CHILD_CONTEXT.get())
 
 
+@contextmanager
+def non_dispatcher_owned_context() -> Iterator[None]:
+    """Mark in-process execution that does NOT own the dispatcher's Kanban task.
+
+    A Kanban worker is a normal CLI agent whose default toolset includes
+    ``cronjob``; ``cronjob(action="run")`` runs ``run_job()`` inside the worker's
+    own process, where ``HERMES_KANBAN_TASK`` is legitimately set.  Without this
+    marker the cron agent is misread as that worker: the kanban toolset is
+    force-added, the worker protocol is injected into its system prompt, and
+    ``kanban_complete`` defaults ``task_id`` to ``$HERMES_KANBAN_TASK`` — letting
+    an unrelated cron job close the worker's task and overwrite real results.
+
+    Scoped via ContextVar rather than by clearing ``os.environ``: the env is
+    process-global and shared with the worker's own claim heartbeat, the
+    gateway's Kanban watchers, and concurrent cron jobs on the parallel pool, so
+    mutating it would starve the worker's claim and race those readers.
+    """
+    token = _NON_DISPATCHER_OWNED_CONTEXT.set(True)
+    try:
+        yield
+    finally:
+        _NON_DISPATCHER_OWNED_CONTEXT.reset(token)
+
+
+def is_dispatcher_owned_worker_context() -> bool:
+    """Return True only when this execution owns the dispatcher's Kanban task.
+
+    The single predicate every ``HERMES_KANBAN_*`` identity gate should use
+    before trusting those vars.  False for delegate_task children and for cron
+    jobs fired in-process from a worker.
+    """
+    if _DELEGATED_CHILD_CONTEXT.get():
+        return False
+    return not _NON_DISPATCHER_OWNED_CONTEXT.get()
+
+
+def enter_non_dispatcher_owned_context() -> Token[bool]:
+    """Token-based form of :func:`non_dispatcher_owned_context`.
+
+    For callers whose scope is a long ``try`` with a matching ``finally`` rather
+    than a ``with`` block (``cron.scheduler.run_job``).  Pair with
+    :func:`exit_non_dispatcher_owned_context`.
+    """
+    return _NON_DISPATCHER_OWNED_CONTEXT.set(True)
+
+
+def exit_non_dispatcher_owned_context(token: Token[bool]) -> None:
+    """Restore the flag saved by :func:`enter_non_dispatcher_owned_context`."""
+    _NON_DISPATCHER_OWNED_CONTEXT.reset(token)
+
+
 def is_delegated_child_process_context() -> bool:
     """Return True in this process or a subprocess spawned by a child."""
     import os

agent/skill_utils.py

@@ -289,10 +289,12 @@ _ENV_DETECT_CACHE: Dict[str, bool] = {}
 def _detect_environment(env: str) -> bool:
     """Return True when the named runtime environment is currently active.
 
-    Cached per process. Unknown env names return True (fail-open: never hide a
-    skill because of a tag we don't understand).
+    Cached per process, EXCEPT ``kanban``: that verdict is context-dependent
+    (a delegate_task child or an in-process cron job sees the worker's
+    HERMES_KANBAN_* vars without owning them), so caching it process-wide would
+    freeze whichever context asked first and leak it to the others.
     """
-    if env in _ENV_DETECT_CACHE:
+    if env != "kanban" and env in _ENV_DETECT_CACHE:
         return _ENV_DETECT_CACHE[env]
 
     result = True
@@ -304,6 +306,20 @@ def _detect_environment(env: str) -> bool:
         # gate on (``tools/kanban_tools.py``) so the offer filter agrees with
         # tool availability.
         if os.getenv("HERMES_KANBAN_TASK") or os.getenv("HERMES_KANBAN_BOARD"):
+            # ...but only when this execution actually owns the dispatcher's
+            # task. A delegate_task child or a cron job fired in-process from a
+            # worker sees the worker's vars without being that worker.
+            try:
+                from agent.delegation_context import (
+                    is_dispatcher_owned_worker_context,
+                )
+
+                _owns_dispatcher_task = is_dispatcher_owned_worker_context()
+            except Exception:
+                _owns_dispatcher_task = True
+        else:
+            _owns_dispatcher_task = False
+        if _owns_dispatcher_task:
             result = True
         else:
             try:

cron/scheduler.py

@@ -49,6 +49,10 @@ from hermes_cli.config import (
 from hermes_cli.fallback_config import get_fallback_chain
 from hermes_time import now as _hermes_now
 from agent.interrupt_compat import request_hard_interrupt
+from agent.delegation_context import (
+    enter_non_dispatcher_owned_context,
+    exit_non_dispatcher_owned_context,
+)
 
 logger = logging.getLogger(__name__)
 
@@ -3123,12 +3127,33 @@ def run_job(
     # future writers.  Acquire itself can't leak (it either blocks or returns).
     _cron_session_var = _VAR_MAP["HERMES_CRON_SESSION"]
     _cron_session_token = None
+    _non_dispatcher_token = None
     try:
         # Scope cron approval policy to this job. Keep the token so the finally
         # restores the pre-job state instead of pinning an explicit empty value,
         # which would suppress the legacy os.environ fallback used by standalone
         # cron entrypoints and tests.
         _cron_session_token = _cron_session_var.set("1")
+
+        # Mark this job as NOT the dispatcher-owned kanban worker.
+        #
+        # A kanban worker is a normal `hermes chat -q` CLI agent whose default
+        # toolset includes `cronjob`, running with HERMES_KANBAN_TASK
+        # legitimately in its own env; `cronjob(action="run")` calls
+        # run_one_job() -> run_job() right here in that process.  Without this
+        # marker the cron agent is misread as that worker: the kanban toolset is
+        # force-added, the worker protocol is injected into its system prompt,
+        # and kanban_complete defaults task_id to $HERMES_KANBAN_TASK -- letting
+        # an unrelated cron job close the worker's task and overwrite real
+        # results.
+        #
+        # A ContextVar, NOT an os.environ clear: the env is process-global and
+        # shared with the worker's own claim heartbeat (run_agent._touch_activity
+        # -> heartbeat_current_worker_from_env, which would starve and let the
+        # dispatcher reclaim a live task), the gateway's kanban watchers, and
+        # concurrent cron jobs on the parallel pool.  contextvars.copy_context()
+        # at the run_conversation hop carries this into the agent thread.
+        _non_dispatcher_token = enter_non_dispatcher_owned_context()
         if _job_workdir:
             os.environ["TERMINAL_CWD"] = _job_workdir
             logger.info("Job '%s': using workdir %s", job_id, _job_workdir)
@@ -3777,6 +3802,8 @@ def run_job(
         clear_session_vars(_ctx_tokens)
         if _cron_session_token is not None:
             _cron_session_var.reset(_cron_session_token)
+        if _non_dispatcher_token is not None:
+            exit_non_dispatcher_owned_context(_non_dispatcher_token)
         for _var_name in _cron_delivery_vars:
             _VAR_MAP[_var_name].set("")
         if _session_db:

model_tools.py

@@ -54,6 +54,17 @@ def _is_delegated_child_context() -> bool:
         return False
 
 
+def _is_dispatcher_owned_worker() -> bool:
+    """False when HERMES_KANBAN_* is present but this execution does not own it
+    (delegate_task child, or a cron job fired in-process from a worker)."""
+    try:
+        from agent.delegation_context import is_dispatcher_owned_worker_context
+
+        return is_dispatcher_owned_worker_context()
+    except Exception:
+        return True
+
+
 # =============================================================================
 # Async Bridging  (single source of truth -- used by registry.dispatch too)
 # =============================================================================
@@ -342,6 +353,7 @@ def get_tool_definitions(
                 bool(os.environ.get("HERMES_KANBAN_TASK")),
                 bool(skip_tool_search_assembly),
                 _is_delegated_child_context(),
+                _is_dispatcher_owned_worker(),
                 profile_scope,
             )
         cached = _tool_defs_cache.get(cache_key) if cache_key is not None else None
@@ -391,6 +403,7 @@ def _compute_tool_definitions(
         if (
             os.environ.get("HERMES_KANBAN_TASK")
             and not _is_delegated_child_context()
+            and _is_dispatcher_owned_worker()
             and "kanban" not in effective_enabled_toolsets
         ):
             # Dispatcher-spawned workers are scoped by HERMES_KANBAN_TASK and

tests/cron/test_cron_kanban_env_isolation.py

@@ -0,0 +1,424 @@
+"""Cron sessions must not inherit a kanban worker's dispatcher identity.
+
+A cron job can be fired *in-process* from a kanban worker: the worker is a
+normal ``hermes chat -q`` CLI agent (its default toolset includes ``cronjob``)
+running with ``HERMES_KANBAN_TASK`` legitimately set in its own environment,
+and ``cronjob(action="run")`` calls ``run_one_job()`` -> ``run_job()`` in that
+same process.
+
+Without isolation the cron ``AIAgent`` is misidentified as that worker: the
+kanban toolset is force-added, the kanban-worker protocol is injected into its
+system prompt, and ``kanban_complete`` defaults ``task_id`` to
+``$HERMES_KANBAN_TASK`` — letting an unrelated cron job close the worker's task
+and overwrite real results.
+
+The isolation is a **ContextVar**, deliberately not an ``os.environ`` clear:
+``os.environ`` is process-global and shared with
+
+  * the worker's own claim heartbeat (``run_agent._touch_activity`` ->
+    ``heartbeat_current_worker_from_env``), which would starve and let the
+    dispatcher reclaim a task whose worker is still alive;
+  * the gateway's kanban watchers, which do their own board save/restore;
+  * concurrent cron jobs on the parallel pool, which take a *shared* read lock
+    and can interleave one another's snapshot/restore.
+
+So these tests assert both that the identity is hidden AND that the environment
+is left completely untouched.
+"""
+
+from __future__ import annotations
+
+import ast
+import os
+import threading
+
+import pytest
+
+
+@pytest.fixture(autouse=True)
+def _clear_kanban_detect_cache():
+    """`_detect_environment` memoizes per process; kanban is context-dependent."""
+    import agent.skill_utils as su
+
+    su._ENV_DETECT_CACHE.pop("kanban", None)
+    yield
+    su._ENV_DETECT_CACHE.pop("kanban", None)
+
+
+@pytest.fixture()
+def worker_env(monkeypatch):
+    """Simulate running inside a dispatcher-spawned kanban worker."""
+    monkeypatch.setenv("HERMES_KANBAN_TASK", "t_worker_real_task")
+    monkeypatch.setenv("HERMES_KANBAN_WORKSPACE", "/tmp/ws")
+    monkeypatch.setenv("HERMES_KANBAN_RUN_ID", "42")
+    monkeypatch.setenv("HERMES_KANBAN_CLAIM_LOCK", "lock-abc")
+    monkeypatch.setenv("HERMES_KANBAN_BOARD", "team-alpha")
+
+
+# ---------------------------------------------------------------------------
+# The predicate itself
+# ---------------------------------------------------------------------------
+
+class TestDispatcherOwnedPredicate:
+    def test_default_is_dispatcher_owned(self):
+        from agent.delegation_context import is_dispatcher_owned_worker_context
+
+        assert is_dispatcher_owned_worker_context() is True
+
+    def test_false_inside_non_dispatcher_context(self):
+        from agent.delegation_context import (
+            is_dispatcher_owned_worker_context,
+            non_dispatcher_owned_context,
+        )
+
+        with non_dispatcher_owned_context():
+            assert is_dispatcher_owned_worker_context() is False
+        assert is_dispatcher_owned_worker_context() is True
+
+    def test_token_form_restores(self):
+        from agent.delegation_context import (
+            enter_non_dispatcher_owned_context,
+            exit_non_dispatcher_owned_context,
+            is_dispatcher_owned_worker_context,
+        )
+
+        token = enter_non_dispatcher_owned_context()
+        assert is_dispatcher_owned_worker_context() is False
+        exit_non_dispatcher_owned_context(token)
+        assert is_dispatcher_owned_worker_context() is True
+
+    def test_nesting_restores_outer_value(self):
+        from agent.delegation_context import (
+            is_dispatcher_owned_worker_context,
+            non_dispatcher_owned_context,
+        )
+
+        with non_dispatcher_owned_context():
+            with non_dispatcher_owned_context():
+                assert is_dispatcher_owned_worker_context() is False
+            assert is_dispatcher_owned_worker_context() is False
+        assert is_dispatcher_owned_worker_context() is True
+
+    def test_delegated_child_still_not_dispatcher_owned(self, monkeypatch):
+        """The pre-existing delegate_task flag keeps its meaning."""
+        import agent.delegation_context as dc
+
+        token = dc._DELEGATED_CHILD_CONTEXT.set(True)
+        try:
+            assert dc.is_dispatcher_owned_worker_context() is False
+        finally:
+            dc._DELEGATED_CHILD_CONTEXT.reset(token)
+
+    def test_thread_isolation(self, worker_env):
+        """A ContextVar set in one thread must not leak into a sibling thread.
+
+        This is the property an os.environ clear cannot provide, and the reason
+        concurrent cron jobs can't corrupt each other.
+        """
+        from agent.delegation_context import (
+            is_dispatcher_owned_worker_context,
+            non_dispatcher_owned_context,
+        )
+
+        seen = {}
+        release = threading.Event()
+
+        def sibling():
+            seen["sibling"] = is_dispatcher_owned_worker_context()
+            release.set()
+
+        def job():
+            with non_dispatcher_owned_context():
+                seen["job"] = is_dispatcher_owned_worker_context()
+                t = threading.Thread(target=sibling)
+                t.start()
+                release.wait(5)
+                t.join(5)
+
+        t = threading.Thread(target=job)
+        t.start()
+        t.join(5)
+
+        assert seen["job"] is False, "job thread must be marked non-dispatcher"
+        assert seen["sibling"] is True, "sibling thread must be unaffected"
+
+
+# ---------------------------------------------------------------------------
+# The gates that consume it
+# ---------------------------------------------------------------------------
+
+class TestKanbanGatesRespectContext:
+    def test_task_tools_hidden_from_cron_agent(self, worker_env):
+        from agent.delegation_context import non_dispatcher_owned_context
+        from tools import kanban_tools
+
+        assert kanban_tools._check_kanban_mode() is True
+        with non_dispatcher_owned_context():
+            assert kanban_tools._check_kanban_mode() is False
+
+    def test_complete_does_not_default_to_worker_task(self, worker_env):
+        """The damage path: kanban_complete must not inherit the task id."""
+        from agent.delegation_context import non_dispatcher_owned_context
+        from tools import kanban_tools
+
+        assert kanban_tools._default_task_id(None) == "t_worker_real_task"
+        with non_dispatcher_owned_context():
+            assert kanban_tools._default_task_id(None) is None
+
+    def test_explicit_task_id_still_honoured(self, worker_env):
+        """Only the implicit default is suppressed, not an explicit argument."""
+        from agent.delegation_context import non_dispatcher_owned_context
+        from tools import kanban_tools
+
+        with non_dispatcher_owned_context():
+            assert kanban_tools._default_task_id("t_explicit") == "t_explicit"
+
+    def test_skill_environment_gate(self, worker_env):
+        from agent.delegation_context import non_dispatcher_owned_context
+        import agent.skill_utils as su
+
+        su._ENV_DETECT_CACHE.pop("kanban", None)
+        assert su._detect_environment("kanban") is True
+        with non_dispatcher_owned_context():
+            su._ENV_DETECT_CACHE.pop("kanban", None)
+            assert su._detect_environment("kanban") is False
+
+    def test_kanban_env_verdict_is_not_memoized(self, worker_env):
+        """`kanban` must bypass _ENV_DETECT_CACHE: caching it process-wide would
+        freeze whichever context asked first and leak it to the others."""
+        from agent.delegation_context import non_dispatcher_owned_context
+        import agent.skill_utils as su
+
+        su._ENV_DETECT_CACHE.pop("kanban", None)
+        assert su._detect_environment("kanban") is True
+        with non_dispatcher_owned_context():
+            # No manual cache clear here — the production code must not have
+            # cached the previous True.
+            assert su._detect_environment("kanban") is False
+        assert su._detect_environment("kanban") is True
+
+    def test_toolset_force_add_suppressed(self, worker_env):
+        from agent.delegation_context import non_dispatcher_owned_context
+        import model_tools
+
+        assert model_tools._is_dispatcher_owned_worker() is True
+        with non_dispatcher_owned_context():
+            assert model_tools._is_dispatcher_owned_worker() is False
+
+
+# ---------------------------------------------------------------------------
+# run_job wiring
+# ---------------------------------------------------------------------------
+
+class TestRunJobKanbanIsolation:
+    @staticmethod
+    def _install_stubs(monkeypatch, observed: dict, agent_cls=None):
+        import sys
+
+        import cron.scheduler as sched
+        from agent.delegation_context import is_dispatcher_owned_worker_context
+
+        class FakeAgent:
+            def __init__(self, **kwargs):
+                observed["dispatcher_owned_during_init"] = (
+                    is_dispatcher_owned_worker_context()
+                )
+                observed["kanban_env_during_init"] = {
+                    k: v for k, v in os.environ.items()
+                    if k.startswith("HERMES_KANBAN_")
+                }
+
+            def run_conversation(self, *_a, **_kw):
+                observed["dispatcher_owned_during_run"] = (
+                    is_dispatcher_owned_worker_context()
+                )
+                return {"final_response": "done", "messages": []}
+
+            def get_activity_summary(self):
+                return {"seconds_since_activity": 0.0}
+
+        fake_mod = type(sys)("run_agent")
+        fake_mod.AIAgent = agent_cls or FakeAgent
+        monkeypatch.setitem(sys.modules, "run_agent", fake_mod)
+
+        from hermes_cli import runtime_provider as _rtp
+
+        monkeypatch.setattr(
+            _rtp, "resolve_runtime_provider",
+            lambda **_kw: {
+                "provider": "test", "api_key": "k",
+                "base_url": "http://test.local",
+                "api_mode": "chat_completions",
+            },
+        )
+        monkeypatch.setattr(
+            sched, "_build_job_prompt", lambda job, prerun_script=None: "hi"
+        )
+        monkeypatch.setattr(sched, "_resolve_origin", lambda job: None)
+        monkeypatch.setattr(sched, "_resolve_delivery_target", lambda job: None)
+        monkeypatch.setattr(
+            sched, "_resolve_cron_enabled_toolsets", lambda job, cfg: None
+        )
+        monkeypatch.setenv("HERMES_CRON_TIMEOUT", "0")
+
+        import dotenv
+
+        monkeypatch.setattr(dotenv, "load_dotenv", lambda *_a, **_kw: True)
+
+    @staticmethod
+    def _job(job_id="kanban-iso"):
+        return {
+            "id": job_id, "name": "kanban-iso-job",
+            "workdir": None, "schedule_display": "manual",
+        }
+
+    def test_agent_runs_as_non_dispatcher(self, monkeypatch, worker_env):
+        import cron.scheduler as sched
+
+        observed: dict = {}
+        self._install_stubs(monkeypatch, observed)
+
+        success, *_ = sched.run_job(self._job())
+        assert success is True
+        assert observed["dispatcher_owned_during_init"] is False
+        assert observed["dispatcher_owned_during_run"] is False
+
+    def test_environment_is_left_untouched(self, monkeypatch, worker_env):
+        """The whole point of the ContextVar: os.environ must not be mutated, so
+        the worker's claim heartbeat and the gateway watchers keep working."""
+        import cron.scheduler as sched
+
+        before = {
+            k: v for k, v in os.environ.items() if k.startswith("HERMES_KANBAN_")
+        }
+        assert before, "fixture should have populated kanban env"
+
+        observed: dict = {}
+        self._install_stubs(monkeypatch, observed)
+
+        success, *_ = sched.run_job(self._job())
+        assert success is True
+
+        # Untouched DURING the job (the heartbeat thread reads it concurrently)...
+        assert observed["kanban_env_during_init"] == before
+        # ...and after.
+        after = {
+            k: v for k, v in os.environ.items() if k.startswith("HERMES_KANBAN_")
+        }
+        assert after == before
+
+    def test_context_reset_after_job(self, monkeypatch, worker_env):
+        import cron.scheduler as sched
+        from agent.delegation_context import is_dispatcher_owned_worker_context
+
+        observed: dict = {}
+        self._install_stubs(monkeypatch, observed)
+
+        sched.run_job(self._job("kanban-iso-reset"))
+        assert is_dispatcher_owned_worker_context() is True
+
+    def test_context_reset_even_when_job_raises(self, monkeypatch, worker_env):
+        import cron.scheduler as sched
+        from agent.delegation_context import is_dispatcher_owned_worker_context
+
+        class ExplodingAgent:
+            def __init__(self, **kwargs):
+                pass
+
+            def run_conversation(self, *_a, **_kw):
+                raise RuntimeError("boom")
+
+            def get_activity_summary(self):
+                return {"seconds_since_activity": 0.0}
+
+        observed: dict = {}
+        self._install_stubs(monkeypatch, observed, agent_cls=ExplodingAgent)
+
+        success, *_ = sched.run_job(self._job("kanban-iso-fail"))
+        assert success is False
+        assert is_dispatcher_owned_worker_context() is True
+        # And the env survived the failure too.
+        assert os.environ.get("HERMES_KANBAN_BOARD") == "team-alpha"
+
+    def test_concurrent_jobs_do_not_corrupt_worker_identity(
+        self, monkeypatch, worker_env
+    ):
+        """Two workdir-less jobs run concurrently on the parallel pool and take a
+        SHARED read lock, so they interleave. With an os.environ snapshot/clear/
+        restore this permanently destroyed the worker's identity; a ContextVar is
+        per-thread and cannot."""
+        import cron.scheduler as sched
+
+        before = {
+            k: v for k, v in os.environ.items() if k.startswith("HERMES_KANBAN_")
+        }
+        observed: dict = {}
+        self._install_stubs(monkeypatch, observed)
+
+        results = {}
+
+        def run(name):
+            ok, *_ = sched.run_job(self._job(f"kanban-iso-{name}"))
+            results[name] = ok
+
+        threads = [threading.Thread(target=run, args=(n,)) for n in ("a", "b")]
+        for t in threads:
+            t.start()
+        for t in threads:
+            t.join(60)
+
+        assert results == {"a": True, "b": True}
+        after = {
+            k: v for k, v in os.environ.items() if k.startswith("HERMES_KANBAN_")
+        }
+        assert after == before, "worker identity must survive concurrent cron jobs"
+
+
+# ---------------------------------------------------------------------------
+# Drift guard
+# ---------------------------------------------------------------------------
+
+def test_every_dispatcher_kanban_var_is_identity_gated():
+    """Invariant: every HERMES_KANBAN_* var the dispatcher injects is covered by
+    the canonical KANBAN_ENV_KEYS, so the delegate_task subprocess scrubber and
+    any future consumer stay in sync with ``_default_spawn``.
+
+    Fails loudly if a new dispatcher var is added without registering it.
+    """
+    import hermes_cli.kanban_db as kanban_db
+    from agent.delegation_context import KANBAN_ENV_KEYS
+
+    source = ast.parse(open(kanban_db.__file__, encoding="utf-8").read())
+    spawn = next(
+        node for node in ast.walk(source)
+        if isinstance(node, ast.FunctionDef) and node.name == "_default_spawn"
+    )
+
+    injected = set()
+    for node in ast.walk(spawn):
+        if not isinstance(node, ast.Assign):
+            continue
+        for target in node.targets:
+            if not isinstance(target, ast.Subscript):
+                continue
+            if ast.unparse(target.value) != "env":
+                continue
+            key = ast.unparse(target.slice).strip("\"'")
+            if key.startswith("HERMES_KANBAN_"):
+                injected.add(key)
+
+    assert injected, "failed to parse dispatcher kanban env injection"
+
+    # These are worker-behaviour knobs rather than board/task identity; they are
+    # intentionally not part of KANBAN_ENV_KEYS. Listed explicitly so adding a
+    # new var forces a decision instead of silently passing.
+    behaviour_only = {
+        "HERMES_KANBAN_BRANCH",
+        "HERMES_KANBAN_GOAL_MODE",
+        "HERMES_KANBAN_GOAL_MAX_TURNS",
+    }
+    uncovered = injected - set(KANBAN_ENV_KEYS) - behaviour_only
+    assert not uncovered, (
+        f"dispatcher injects {sorted(uncovered)} which is neither in "
+        "KANBAN_ENV_KEYS nor explicitly classified as behaviour-only"
+    )

tools/kanban_tools.py

@@ -71,6 +71,17 @@ def _is_delegated_child_context() -> bool:
         return False
 
 
+def _is_dispatcher_owned_worker() -> bool:
+    """False for delegate_task children AND for cron jobs fired in-process from
+    a worker — i.e. whenever HERMES_KANBAN_* is present but not ours."""
+    try:
+        from agent.delegation_context import is_dispatcher_owned_worker_context
+
+        return is_dispatcher_owned_worker_context()
+    except Exception:
+        return True
+
+
 def _reject_delegated_child_mutation(tool_name: str) -> Optional[str]:
     """Deny Kanban mutations from delegate_task children.
 
@@ -103,7 +114,7 @@ def _check_kanban_mode() -> bool:
     """
     if _is_delegated_child_context():
         return False
-    if os.environ.get("HERMES_KANBAN_TASK"):
+    if os.environ.get("HERMES_KANBAN_TASK") and _is_dispatcher_owned_worker():
         return True
     return _profile_has_kanban_toolset()
 
@@ -119,7 +130,7 @@ def _check_kanban_orchestrator_mode() -> bool:
     """
     if _is_delegated_child_context():
         return False
-    if os.environ.get("HERMES_KANBAN_TASK"):
+    if os.environ.get("HERMES_KANBAN_TASK") and _is_dispatcher_owned_worker():
         return False
     return _profile_has_kanban_toolset()
 
@@ -134,6 +145,10 @@ def _default_task_id(arg: Optional[str]) -> Optional[str]:
         return arg
     if _is_delegated_child_context():
         return None
+    if not _is_dispatcher_owned_worker():
+        # A cron job fired in-process from a worker must never inherit the
+        # worker's task id as an implicit default.
+        return None
     env_tid = os.environ.get("HERMES_KANBAN_TASK")
     return env_tid or None