Kanban board env mutation causing concurrency bugs

Fixes on record

Merged to main. A stamp counts the issues a fix is linked to.

• #39074 fix(kanban): isolate board override per concurrent call (CLI + dashboard)

  merged to main

  files: hermes_cli/kanban.py, hermes_cli/kanban_db.py, plugins/kanban/dashboard/plugin_api.py, scripts/release.py, tests/hermes_cli/test_kanban_cli.py

  Show the diff

  • hermes_cli/kanban.py

    @@ -15,6 +15,7 @@
     from __future__ import annotations
     
     import argparse
    +import contextlib
     import json
     import os
     import shlex
    @@ -884,16 +885,7 @@ def kanban_command(args: argparse.Namespace) -> int:
         # keeps the patch small and inherits the exact same resolution the
         # dispatcher uses for workers — consistency is a feature here.
         board_override = getattr(args, "board", None)
    -    prev_board_env = os.environ.get("HERMES_KANBAN_BOARD")
    -    restore_board_env = False
    -
    -    def _restore_board_env() -> None:
    -        if not restore_board_env:
    -            return
    -        if prev_board_env is None:
    -            os.environ.pop("HERMES_KANBAN_BOARD", None)
    -        else:
    -            os.environ["HERMES_KANBAN_BOARD"] = prev_board_env
    +    board_scope = contextlib.nullcontext()
         if board_override:
             try:
                 normed = kb._normalize_board_slug(board_override)
    @@ -912,8 +904,7 @@ def _restore_board_env() -> None:
                     file=sys.stderr,
                 )
                 return 1
    -        os.environ["HERMES_KANBAN_BOARD"] = normed
    -        restore_board_env = True
    +        board_scope = kb.scoped_current_board(normed)
     
         # Auto-initialize the DB before dispatching any subcommand. init_db
         # is idempotent, so running it every invocation is cheap (one
    @@ -922,66 +913,62 @@ def _restore_board_env() -> None:
         # HERMES_HOME. Previously only `init` and `daemon` triggered
         # schema creation; `create` / `list` / every other command would
         # error out on a fresh install.
    -    try:
    -        kb.init_db()
    -    except Exception as exc:
    -        print(f"kanban: could not initialize database: {exc}", file=sys.stderr)
    -        _restore_board_env()
    -        return 1
    +    with board_scope:
    +        try:
    +            kb.init_db()
    +        except Exception as exc:
    +            print(f"kanban: could not initialize database: {exc}", file=sys.stderr)
    +            return 1
     
    -    handlers = {
    -        "init":     _cmd_init,
    -        "create":   _cmd_create,
    -        "swarm":    _cmd_swarm,
    -        "list":     _cmd_list,
    -        "ls":       _cmd_list,
    -        "show":     _cmd_show,
    -        "assign":   _cmd_assign,
    -        "reclaim":  _cmd_reclaim,
    -        "reassign": _cmd_reassign,
    -        "diagnostics": _cmd_diagnostics,
    -        "diag":     _cmd_diagnostics,
    -        "link":     _cmd_link,
    -        "unlink":   _cmd_unlink,
    -        "claim":    _cmd_claim,
    -        "comment":  _cmd_comment,
    -        "complete": _cmd_complete,
    -        "edit":     _cmd_edit,
    -        "block":    _cmd_block,
    -        "schedule": _cmd_schedule,
    -        "unblock":  _cmd_unblock,
    -        "promote":  _cmd_promote,
    -        "archive":  _cmd_archive,
    -        "tail":     _cmd_tail,
    -        "dispatch": _cmd_dispatch,
    -        "daemon":   _cmd_daemon,
    -        "watch":    _cmd_watch,
    -        "stats":    _cmd_stats,
    -        "log":      _cmd_log,
    -        "runs":     _cmd_runs,
    -        "heartbeat": _cmd_heartbeat,
    -        "assignees": _cmd_assignees,
    -        "notify-subscribe":   _cmd_notify_subscribe,
    -        "notify-list":        _cmd_notify_list,
    -        "notify-unsubscribe": _cmd_notify_unsubscribe,
    -        "context":  _cmd_context,
    -        "specify":  _cmd_specify,
    -        "decompose":  _cmd_decompose,
    -        "gc":       _cmd_gc,
    -    }
    -    handler = handlers.get(action)
    -    if not handler:
    -        print(f"kanban: unknown action {action!r}", file=sys.stderr)
    -        _restore_board_env()
    -        return 2
    -    try:
    -        return int(handler(args) or 0)
    -    except (ValueError, RuntimeError) as exc:
    -        print(f"kanban: {exc}", file=sys.stderr)
    -        _restore_board_env()
    -        return 1
    -    finally:
    -        _restore_board_env()
    +        handlers = {
    +            "init":     _cmd_init,
    +            "create":   _cmd_create,
    +            "swarm":    _cmd_swarm,
    +            "list":     _cmd_list,
    +            "ls":       _cmd_list,
    +            "show":     _cmd_show,
    +            "assign":   _cmd_assign,
    +            "reclaim":  _cmd_reclaim,
    +            "reassign": _cmd_reassign,
    +            "diagnostics": _cmd_diagnostics,
    +            "diag":     _cmd_diagnostics,
    +            "link":     _cmd_link,
    +            "unlink":   _cmd_unlink,
    +            "claim":    _cmd_claim,
    +            "comment":  _cmd_comment,
    +            "complete": _cmd_complete,
    +            "edit":     _cmd_edit,
    +            "block":    _cmd_block,
    +            "schedule": _cmd_schedule,
    +            "unblock":  _cmd_unblock,
    +            "promote":  _cmd_promote,
    +            "archive":  _cmd_archive,
    +            "tail":     _cmd_tail,
    +            "dispatch": _cmd_dispatch,
    +            "daemon":   _cmd_daemon,
    +            "watch":    _cmd_watch,
    +            "stats":    _cmd_stats,
    +            "log":      _cmd_log,
    +            "runs":     _cmd_runs,
    +            "heartbeat": _cmd_heartbeat,
    +            "assignees": _cmd_assignees,
    +            "notify-subscribe":   _cmd_notify_subscribe,
    +            "notify-list":        _cmd_notify_list,
    +            "notify-unsubscribe": _cmd_notify_unsubscribe,
    +            "context":  _cmd_context,
    +            "specify":  _cmd_specify,
    +            "decompose":  _cmd_decompose,
    +            "gc":       _cmd_gc,
    +        }
    +        handler = handlers.get(action)
    +        if not handler:
    +            print(f"kanban: unknown action {action!r}", file=sys.stderr)
    +            return 2
    +        try:
    +            return int(handler(args) or 0)
    +        except (ValueError, RuntimeError) as exc:
    +            print(f"kanban: {exc}", file=sys.stderr)
    +            return 1
     
     
     # ---------------------------------------------------------------------------

  • hermes_cli/kanban_db.py

    @@ -83,6 +83,7 @@
     import threading
     import logging
     import time
    +from contextvars import ContextVar, Token
     from dataclasses import dataclass, field
     from pathlib import Path
     from typing import Any, Iterable, Optional
    @@ -222,6 +223,20 @@ def _resolve_rate_limit_cooldown_seconds() -> int:
     # ---------------------------------------------------------------------------
     
     DEFAULT_BOARD = "default"
    +_CURRENT_BOARD_OVERRIDE: ContextVar[str | None] = ContextVar(
    +    "hermes_kanban_current_board_override",
    +    default=None,
    +)
    +
    +
    [redacted:email]
    +def scoped_current_board(slug: str):
    +    """Temporarily pin the active board for the current context only."""
    +    token: [redacted:secret-assignment] | None] = _CURRENT_BOARD_OVERRIDE.set(slug)
    +    try:
    +        yield
    +    finally:
    +        _CURRENT_BOARD_OVERRIDE.reset(token)
     
     # Slug validator: lowercase alphanumerics, digits, hyphens; 1–64 chars.
     # Strict enough to stop traversal (`..`) and embedded path separators, loose
    @@ -305,6 +320,15 @@ def get_current_board() -> str:
         with a best-effort warning — the dispatcher must never crash because a
         user hand-edited a file or removed a board directory.
         """
    +    scoped = (_CURRENT_BOARD_OVERRIDE.get() or "").strip()
    +    if scoped:
    +        try:
    +            normed = _normalize_board_slug(scoped)
    +            if normed and board_exists(normed):
    +                return normed
    +        except ValueError:
    +            pass
    +
         env = os.environ.get("HERMES_KANBAN_BOARD", "").strip()
         if env:
             try:

  • plugins/kanban/dashboard/plugin_api.py

    @@ -38,7 +38,6 @@
     import asyncio
     import json
     import logging
    -import os
     import sqlite3
     import time
     from dataclasses import asdict
    @@ -1620,10 +1619,12 @@ def specify_task_endpoint(
         """
         board = _resolve_board(board)
         # Pin the board for the duration of this call so the specifier module
    -    # (which calls ``kb.connect()`` with no args) hits the right DB.
    -    prev_env = os.environ.get("HERMES_KANBAN_BOARD")
    -    try:
    -        os.environ["HERMES_KANBAN_BOARD"] = board or kanban_db.DEFAULT_BOARD
    +    # (which calls ``kb.connect()`` with no args) hits the right DB. Use a
    +    # context-local override rather than mutating the process-global
    +    # HERMES_KANBAN_BOARD env var — this endpoint runs in FastAPI's
    +    # threadpool, so two concurrent requests for different boards would
    +    # otherwise race on the shared env var and cross-write (issue #38323).
    +    with kanban_db.scoped_current_board(board or kanban_db.DEFAULT_BOARD):
             # Import lazily so a missing auxiliary client at import time
             # doesn't break plugin load.
             from hermes_cli import kanban_specify  # noqa: WPS433 (intentional)
    @@ -1632,11 +1633,6 @@ def specify_task_endpoint(
                 task_id,
                 author=(payload.author or None),
             )
    -    finally:
    -        if prev_env is None:
    -            os.environ.pop("HERMES_KANBAN_BOARD", None)
    -        else:
    -            os.environ["HERMES_KANBAN_BOARD"] = prev_env
     
         return {
             "ok": bool(outcome.ok),
    @@ -2233,19 +2229,16 @@ def decompose_task_endpoint(
         can take minutes on reasoning models.
         """
         board = _resolve_board(board)
    -    prev_env = os.environ.get("HERMES_KANBAN_BOARD")
    -    try:
    -        os.environ["HERMES_KANBAN_BOARD"] = board or kanban_db.DEFAULT_BOARD
    +    # Context-local board pin (see specify endpoint above): this sync
    +    # endpoint runs in FastAPI's threadpool, so mutating the process-global
    +    # HERMES_KANBAN_BOARD env var would let concurrent requests for
    +    # different boards race and cross-write (issue #38323).
    +    with kanban_db.scoped_current_board(board or kanban_db.DEFAULT_BOARD):
             from hermes_cli import kanban_decompose  # noqa: WPS433 (intentional)
             outcome = kanban_decompose.decompose_task(
                 task_id,
                 author=(payload.author or None),
             )
    -    finally:
    -        if prev_env is None:
    -            os.environ.pop("HERMES_KANBAN_BOARD", None)
    -        else:
    -            os.environ["HERMES_KANBAN_BOARD"] = prev_env
     
         return {
             "ok": bool(outcome.ok),

  • scripts/release.py

    @@ -47,6 +47,7 @@
     AUTHOR_MAP = {
         "[redacted:email]": "stremtec",
         "[redacted:email]": "solaitken",
    +    "[redacted:email]": "worlldz",
         "[redacted:email]": "vladkvlchk",
         "[email protected]": "ThyFriendlyFox",
         "[redacted:email]": "truenorth-lj",

  • tests/hermes_cli/test_kanban_cli.py

    @@ -5,6 +5,7 @@
     import argparse
     import json
     import os
    +import threading
     from pathlib import Path
     
     import pytest
    @@ -263,6 +264,52 @@ def test_run_slash_link_unlink(kanban_home):
         assert "Unlinked" in kc.run_slash(f"unlink {ta} {tb}")
     
     
    +def test_board_override_is_isolated_per_concurrent_call(kanban_home, monkeypatch):
    +    kb.create_board("alpha")
    +    kb.create_board("beta")
    +
    +    parser = argparse.ArgumentParser(prog="hermes", add_help=False)
    +    sub = parser.add_subparsers(dest="command")
    +    kc.build_parser(sub)
    +
    +    barrier = threading.Barrier(2)
    +    original_init_db = kb.init_db
    +
    +    def slow_init_db(*args, **kwargs):
    +        try:
    +            barrier.wait(timeout=5)
    +        except threading.BrokenBarrierError:
    +            pass
    +        return original_init_db(*args, **kwargs)
    +
    +    monkeypatch.setattr(kb, "init_db", slow_init_db)
    +
    +    failures: list[str] = []
    +
    +    def worker(board: str, title: str) -> None:
    +        args = parser.parse_args(["kanban", "--board", board, "create", title])
    +        rc = kc.kanban_command(args)
    +        if rc != 0:
    +            failures.append(f"{board}:{rc}")
    +
    +    t1 = threading.Thread(target=worker, args=("alpha", "alpha-task"))
    +    t2 = threading.Thread(target=worker, args=("beta", "beta-task"))
    +    t1.start()
    +    t2.start()
    +    t1.join()
    +    t2.join()
    +
    +    assert failures == []
    +
    +    with kb.connect_closing(board="alpha") as conn:
    +        alpha_titles = [row.title for row in kb.list_tasks(conn, limit=100)]
    +    with kb.connect_closing(board="beta") as conn:
    +        beta_titles = [row.title for row in kb.list_tasks(conn, limit=100)]
    +
    +    assert alpha_titles == ["alpha-task"]
    +    assert beta_titles == ["beta-task"]
    +
    +
     # ---------------------------------------------------------------------------
     # Integration with the COMMAND_REGISTRY
     # ---------------------------------------------------------------------------