From bc09fe483987a572b4260623c14df78a2003d86c Mon Sep 17 00:00:00 2001 From: kshitij Date: Thu, 6 Aug 2026 03:05:14 +0530 Subject: [PATCH 1/2] 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 --- agent/delegation_context.py | 68 ++- agent/skill_utils.py | 22 +- cron/scheduler.py | 27 ++ model_tools.py | 13 + tests/cron/test_cron_kanban_env_isolation.py | 424 +++++++++++++++++++ tools/kanban_tools.py | 19 +- 6 files changed, 567 insertions(+), 6 deletions(-) create mode 100644 tests/cron/test_cron_kanban_env_isolation.py diff --git a/agent/delegation_context.py b/agent/delegation_context.py index 41fe7f56ac171..9b8bce97597a9 100644 --- a/agent/delegation_context.py +++ b/agent/delegation_context.py @@ -5,11 +5,16 @@ 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 @@ 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 diff --git a/agent/skill_utils.py b/agent/skill_utils.py index a302c6981a478..0454f3f7e946e 100644 --- a/agent/skill_utils.py +++ b/agent/skill_utils.py @@ -289,10 +289,12 @@ def skill_matches_platform(frontmatter: Dict[str, Any]) -> 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: diff --git a/cron/scheduler.py b/cron/scheduler.py index f2ab51919a6e7..47f0f26abb201 100644 --- a/cron/scheduler.py +++ b/cron/scheduler.py @@ -49,6 +49,10 @@ 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 _heartbeat_run_claim_if_due(): 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: diff --git a/model_tools.py b/model_tools.py index 0f1290258985e..a4ed0c20c5771 100644 --- a/model_tools.py +++ b/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 diff --git a/tests/cron/test_cron_kanban_env_isolation.py b/tests/cron/test_cron_kanban_env_isolation.py new file mode 100644 index 0000000000000..8ab24b6a00cf1 --- /dev/null +++ b/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" + ) diff --git a/tools/kanban_tools.py b/tools/kanban_tools.py index 76f8902790367..fdf846d7353a4 100644 --- a/tools/kanban_tools.py +++ b/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 From 1e404390622cd430f0afb384a8a9c26e83a50c6f Mon Sep 17 00:00:00 2001 From: kshitij Date: Thu, 6 Aug 2026 03:11:33 +0530 Subject: [PATCH 2/2] test(cron): close a blind spot in the kanban env drift guard The AST invariant only matched `env["HERMES_KANBAN_X"] = ...` subscript assignments, so a future dispatcher var added via `env.update({...})`, `env.setdefault(...)`, or an annotated subscript would have slipped past the guard and leaked into cron sessions unnoticed. None of those shapes exist in _default_spawn today; this is about the guard staying trustworthy as that function evolves. Verified by injecting an unregistered var into _default_spawn one shape at a time and requiring the guard to fail: subscript assign, annotated assign, update(dict literal), setdefault(literal), and update(kwarg) are all detected. Source restored byte-identical after probing. tests/cron/ 410 passed; ruff clean. --- tests/cron/test_cron_kanban_env_isolation.py | 40 +++++++++++++++----- 1 file changed, 31 insertions(+), 9 deletions(-) diff --git a/tests/cron/test_cron_kanban_env_isolation.py b/tests/cron/test_cron_kanban_env_isolation.py index 8ab24b6a00cf1..24be753e0c13b 100644 --- a/tests/cron/test_cron_kanban_env_isolation.py +++ b/tests/cron/test_cron_kanban_env_isolation.py @@ -396,16 +396,38 @@ def test_every_dispatcher_kanban_var_is_identity_gated(): 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): + # env["HERMES_KANBAN_X"] = ... and the annotated form + if isinstance(node, (ast.Assign, ast.AnnAssign)): + targets = node.targets if isinstance(node, ast.Assign) else [node.target] + for target in 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) + # env.update({"HERMES_KANBAN_X": ...}) / env.setdefault("HERMES_KANBAN_X", ...) + elif isinstance(node, ast.Call): + func = ast.unparse(node.func) + if func not in ("env.update", "env.setdefault"): continue - if ast.unparse(target.value) != "env": - continue - key = ast.unparse(target.slice).strip("\"'") - if key.startswith("HERMES_KANBAN_"): - injected.add(key) + literals = [] + for arg in node.args: + if isinstance(arg, ast.Dict): + literals.extend( + k for k in arg.keys if isinstance(k, ast.Constant) + ) + elif isinstance(arg, ast.Constant): + literals.append(arg) + for kw in node.keywords: + if kw.arg and kw.arg.startswith("HERMES_KANBAN_"): + injected.add(kw.arg) + for lit in literals: + if isinstance(lit.value, str) and lit.value.startswith( + "HERMES_KANBAN_" + ): + injected.add(lit.value) assert injected, "failed to parse dispatcher kanban env injection"