feat(kanban): fire kanban_worktree_created hook after git worktree add

agent/shell_hooks.py

@@ -2,7 +2,8 @@
 callbacks on the plugin hook manager, so every ``invoke_hook()`` site dispatches to the scripts.
 Wire: stdin JSON ``{hook_event_name, tool_name, tool_input, session_id, cwd, extra}``; optional stdout
 JSON ``{"decision"|"action": "block"|"modify", ...}`` / ``{"context": ...}`` via ``_parse_response``.
-Exit code 2 blocks a ``pre_tool_call`` even without JSON (Claude-Code / Cursor). Fail open unless ``fail_closed``."""
+Exit code 2 blocks a blocking-capable event (``pre_tool_call``, ``kanban_worktree_created``) even without JSON
+(Claude-Code / Cursor). Fail open unless ``fail_closed``."""
 
 from __future__ import annotations
 
@@ -41,7 +42,8 @@
 # Exit code that signals "block this action" independent of stdout (Claude Code / Cursor).
 BLOCK_EXIT_CODE = 2
 # Events whose block directive is honored downstream; exit-2 blocking and fail_closed only apply here.
-_BLOCKING_EVENTS = frozenset({"pre_tool_call"})
+# kanban_worktree_created: the dispatcher removes the just-created worktree and fails the card.
+_BLOCKING_EVENTS = frozenset({"pre_tool_call", "kanban_worktree_created"})
 _TOOL_EVENTS = frozenset({"pre_tool_call", "post_tool_call"})
 _STDERR_MESSAGE_LIMIT = 400
 _TRUTHY = {"1", "true", "yes", "on"}
@@ -438,10 +440,18 @@ def _block_message(primary: Any, secondary: Any) -> str:
 _PRE_TOOL_DIALECTS = (("action", "message", "reason", "args"), ("decision", "reason", "message", "tool_input"))
 
 
-def _parse_pre_tool_call(data: Dict[str, Any]) -> Optional[Dict[str, Any]]:
+def _parse_block_directive(data: Dict[str, Any]) -> Optional[Dict[str, Any]]:
+    """The block half of the dialects — the only directive ``kanban_worktree_created`` consumes."""
     for verb, primary, secondary, _ in _PRE_TOOL_DIALECTS:
         if data.get(verb) == "block":
             return {"action": "block", "message": _block_message(data.get(primary), data.get(secondary))}
+    return None
+
+
+def _parse_pre_tool_call(data: Dict[str, Any]) -> Optional[Dict[str, Any]]:
+    block = _parse_block_directive(data)
+    if block is not None:
+        return block
     for verb, _, _, payload in _PRE_TOOL_DIALECTS:
         if data.get(verb) == "modify" and isinstance(data.get(payload), dict):
             return {"action": "modify", "args": data[payload]}
@@ -471,7 +481,10 @@ def _parse_context(data: Dict[str, Any]) -> Optional[Dict[str, Any]]:
     return {"context": context} if isinstance(context, str) and context.strip() else None
 
 
-_RESPONSE_PARSERS: Dict[str, Callable[[Dict[str, Any]], Optional[Dict[str, Any]]]] = {"pre_tool_call": _parse_pre_tool_call, "pre_verify": _parse_pre_verify}
+_RESPONSE_PARSERS: Dict[str, Callable[[Dict[str, Any]], Optional[Dict[str, Any]]]] = {
+    "pre_tool_call": _parse_pre_tool_call, "pre_verify": _parse_pre_verify,
+    "kanban_worktree_created": _parse_block_directive,
+}
 
 
 def _parse_response(event: str, stdout: str) -> Optional[Dict[str, Any]]:

hermes_cli/hooks.py

@@ -169,6 +169,11 @@ def _mtime_drift(shell_hooks, spec, entry) -> tuple[bool | None, str, str]:
         "response_model": "claude-haiku-4-5", "usage": {"input_tokens": 256, "output_tokens": 12},
         "assistant_content_chars": 40, "assistant_tool_call_count": 0, "error": None, "error_type": None,
     },
+    "kanban_worktree_created": {
+        "task_id": "test-task", "board": "default", "profile_name": "default",
+        "worktree_path": "/tmp/repo/.worktrees/test-task", "repo_root": "/tmp/repo",
+        "branch": "wt/test-task",
+    },
     "subagent_stop": {
         "parent_session_id": "parent-sess", "child_role": None,
         "child_summary": "Synthetic summary for hooks test", "child_status": "completed",
@@ -332,7 +337,15 @@ def _doctor_one(spec, shell_hooks) -> int:
             rc = result.get("returncode")
             elapsed = result.get("elapsed_seconds", 0)
             stdout = (result.get("stdout") or "").strip()
-            if not stdout:
+            parsed = result.get("parsed")
+            if isinstance(parsed, dict) and parsed.get("action") == "block":
+                # Exit 2 (or block JSON) on a blocking-capable event: the gate fired. Expected for a
+                # script that checks real paths the synthetic payload cannot provide (e.g. a
+                # kanban_worktree_created seeder looking for /tmp/repo/.env); not an error, but not
+                # "observer-only" either — say what the dispatcher would have received.
+                print(f"      ✓ returned a block on the synthetic payload (exit={rc}, {elapsed}s): "
+                      f"{_truncate(str(parsed.get('message') or ''), 120)}")
+            elif not stdout:
                 print(f"      ✓ ran clean with empty stdout "
                       f"(exit={rc}, {elapsed}s) — hook is observer-only")
             else:

hermes_cli/kanban.py

@@ -30,6 +30,7 @@
 from hermes_cli.kanban_boards import _dispatch_boards
 from hermes_cli.kanban_ops import (
     _cmd_daemon, _kanban_config, _cmd_dispatch, _cmd_gc, _cmd_repair, _cmd_tail, _cmd_watch,
+    _register_config_hooks,
 )
 from hermes_cli.kanban_parser import build_parser  # noqa: F401  (re-exported: hermes_cli.main, run_slash)
 
@@ -740,6 +741,9 @@ def _cmd_claim(args: argparse.Namespace) -> int:
                 return _err(f"no such task: {args.task_id}")
             return _err(f"cannot claim {args.task_id}: status={existing.status} "
                         f"lock={existing.claim_lock or '(none)'}")
+        # A worktree card fires kanban_worktree_created from resolve_workspace; the
+        # hooks: block has to be wired first or a configured seed script never runs.
+        _register_config_hooks(args)
         workspace = kbw.resolve_workspace(task)
         kbw.set_workspace_path(conn, task.id, str(workspace))
     print(f"Claimed {task.id}\nWorkspace: {workspace}")

hermes_cli/kanban_db_workspace.py

@@ -482,8 +482,15 @@ def _repo_root_for_worktree_target(path: Path) -> Optional[Path]:
         current = current.parent
 
 
-def _ensure_git_worktree(repo_root: Path, target: Path, branch_name: str) -> None:
-    """Materialize ``target`` as a linked git worktree under ``repo_root``."""
+def _ensure_git_worktree(
+    repo_root: Path, target: Path, branch_name: str, *,
+    task_id: Optional[str] = None, board: Optional[str] = None,
+) -> None:
+    """Materialize ``target`` as a linked git worktree under ``repo_root``.
+
+    A worktree created here fires ``kanban_worktree_created`` before this
+    returns (see :func:`_fire_worktree_created_hook`); the idempotent early
+    return for an already-materialized checkout does not."""
     target = target.expanduser()
     repo_common = _git_common_dir(repo_root)
     if target.exists() and repo_common is not None and _path_key(_git_common_dir(target)) == _path_key(repo_common):
@@ -499,12 +506,107 @@ def _ensure_git_worktree(repo_root: Path, target: Path, branch_name: str) -> Non
         raise RuntimeError(
             f"git worktree add failed for {target} on branch {branch_name}: {stderr}"
         )
+    _fire_worktree_created_hook(repo_root, target, branch_name, task_id=task_id, board=board)
+
+
+#: The card reason travels through ``_record_task_failure``
+#: (hermes_cli/kanban_db_dispatch.py), which keeps ``error[:500]``. The sentence
+#: saying whether a retry is safe is therefore placed BEFORE the hook's own text
+#: and before git's removal error; these caps only bound the tail (a hook that
+#: dumps its whole stderr, git's full refusal) so the exception stays readable.
+_HOOK_MESSAGE_LIMIT = 300
+_REMOVAL_DETAIL_LIMIT = 120
+
+
+def _discard_unseeded_worktree(repo_root: Path, target: Path) -> str:
+    """Remove a worktree whose ``kanban_worktree_created`` hook blocked; describe the outcome.
+
+    Removal is part of the failure, not a courtesy. The raise it accompanies
+    lands in ``dispatch_once``'s workspace guard, which records a ``spawn_failed``
+    and RELEASES the claim, so the card is retried on the next tick. Left in
+    place, the half-seeded tree would make ``_ensure_git_worktree``'s idempotent
+    early return fire on that retry: no ``git worktree add``, no hook, and a
+    worker spawned into the unseeded checkout the block existed to prevent.
+    ``--force`` is required: a hook that failed halfway leaves untracked files
+    and git refuses to remove a dirty worktree. The branch is kept — it may
+    have existed before this call, and the retry's ``worktree add <target>
+    <branch>`` works on it once the tree is gone.
+
+    If the removal itself fails the tree stays and the next retry WILL spawn
+    into it; nothing here can prevent that, so the sentence says to remove it
+    by hand and names why git refused.
+    """
+    try:
+        result = _git(repo_root, "worktree", "remove", "--force", str(target), timeout=60)
+    except Exception as exc:  # git missing, timeout, ...
+        detail = str(exc)
+    else:
+        if result.returncode == 0:
+            return "the unseeded worktree was removed, so a retry re-creates and re-seeds it"
+        detail = (result.stderr or result.stdout or "").strip()
+    return (
+        "the unseeded worktree could NOT be removed — remove it by hand before the retry "
+        f"({detail[:_REMOVAL_DETAIL_LIMIT]})"
+    )
+
+
+def _fire_worktree_created_hook(
+    repo_root: Path, target: Path, branch_name: str, *,
+    task_id: Optional[str], board: Optional[str],
+) -> None:
+    """Fire ``kanban_worktree_created`` for a worktree created just now; a block fails the card.
+
+    A linked worktree is a fresh checkout: tracked files and nothing else, so
+    everything git ignores — for most real projects the ``.env`` the application
+    cannot start without — is absent by definition. Which secrets a project
+    needs, and where they live, is deployment knowledge, so this is the seam for
+    it: a plugin (``ctx.register_hook``) or a ``hooks:`` entry in config.yaml
+    (the shell bridge in ``agent.shell_hooks``) seeds the checkout before any
+    worker sees it.
+
+    Unlike the other ``kanban_*`` hooks this one is a gate, not an observer. It
+    runs synchronously on the caller's thread (``invoke_hook`` does not bound
+    ``kanban_*`` callbacks; a shell hook enforces its own ``timeout``), and a
+    ``{"action": "block", "message"}`` result — the directive shape
+    ``pre_tool_call`` uses, which the shell bridge derives from exit code 2 or
+    ``fail_closed`` — discards the worktree and raises. Raising is deliberate:
+    the hook exists only because somebody installed it, so its failure is a
+    configuration error they want on the card now, not a confusing application
+    failure later inside a checkout missing exactly what the hook was there to
+    provide. ``invoke_hook`` logs and skips a callback that RAISES, so plugins
+    must return the directive. With no subscriber nothing changes.
+    """
+    from hermes_cli.lifecycle import invoke_hook
+
+    try:
+        results = invoke_hook(
+            "kanban_worktree_created", task_id=task_id, board=board or _kb.get_current_board(),
+            profile_name=_kb._hook_profile_name(), worktree_path=str(target),
+            repo_root=str(repo_root), branch=branch_name,
+        )
+    except Exception as exc:
+        discarded = _discard_unseeded_worktree(repo_root, target)
+        raise RuntimeError(
+            f"kanban_worktree_created hook could not run for {target}; {discarded}; "
+            f"cause: {str(exc)[:_HOOK_MESSAGE_LIMIT]}"
+        ) from exc
+    block = next((r for r in results if isinstance(r, dict) and r.get("action") == "block"), None)
+    if block is None:
+        return
+    discarded = _discard_unseeded_worktree(repo_root, target)
+    message = str(block.get("message") or "hook returned a block directive without a message")
+    raise RuntimeError(
+        f"kanban_worktree_created hook blocked the worktree {target}; {discarded}; "
+        f"hook: {message[:_HOOK_MESSAGE_LIMIT]}"
+    )
 
 
-def _anchored_worktree(repo_root: Path, task_id: str, branch_name: str) -> tuple[Path, str]:
+def _anchored_worktree(
+    repo_root: Path, task_id: str, branch_name: str, *, board: Optional[str] = None,
+) -> tuple[Path, str]:
     """Materialize the canonical ``<repo>/.worktrees/<task-id>`` worktree."""
     target = repo_root / ".worktrees" / task_id
-    _ensure_git_worktree(repo_root, target, branch_name)
+    _ensure_git_worktree(repo_root, target, branch_name, task_id=task_id, board=board)
     return target, branch_name
 
 
@@ -537,7 +639,7 @@ def _resolve_worktree_workspace(task: Task, *, board: Optional[str] = None) -> t
                 f"task {task.id} has workspace_kind=worktree but board "
                 f"{board_slug!r} default_workdir {board_default!r} is not inside a git repo"
             )
-        return _anchored_worktree(repo_root, task.id, branch_name)
+        return _anchored_worktree(repo_root, task.id, branch_name, board=board)
 
     requested = Path(task.workspace_path).expanduser()
     if not requested.is_absolute():
@@ -560,23 +662,23 @@ def _resolve_worktree_workspace(task: Task, *, board: Optional[str] = None) -> t
         if fallback_root is not None:
             fallback = fallback_root / ".worktrees" / task.id
             if _path_key(fallback.resolve(strict=False)) != _path_key(requested_resolved):
-                _ensure_git_worktree(fallback_root, fallback, branch_name)
+                _ensure_git_worktree(fallback_root, fallback, branch_name, task_id=task.id, board=board)
                 return fallback.resolve(strict=False), branch_name
         # No repo to anchor a fallback on (or the occupied path IS this task's
         # own canonical worktree): keep the legacy reuse rather than fail dispatch.
         return requested_resolved, actual_branch or branch_name
 
     repo_root = _git_toplevel(requested)
     if repo_root is not None and _path_key(requested_resolved) == _path_key(repo_root):
-        return _anchored_worktree(repo_root, task.id, branch_name)
+        return _anchored_worktree(repo_root, task.id, branch_name, board=board)
 
     repo_root = _repo_root_for_worktree_target(requested.parent)
     if repo_root is None:
         raise ValueError(
             f"task {task.id} worktree path {task.workspace_path!r} is not inside a git repo "
             "and does not point at a git repo root"
         )
-    _ensure_git_worktree(repo_root, requested, branch_name)
+    _ensure_git_worktree(repo_root, requested, branch_name, task_id=task.id, board=board)
     return requested, branch_name
 
 

hermes_cli/kanban_ops.py

@@ -7,6 +7,7 @@
 
 import argparse
 import json
+import logging
 import os
 import sys
 import time
@@ -57,6 +58,29 @@ def tick():
     return _poll_loop(args.interval, tick)
 
 
+def _register_config_hooks(args: argparse.Namespace) -> None:
+    """Wire config.yaml's ``hooks:`` block onto the plugin manager for a standalone kanban entry point.
+
+    ``hermes chat`` / ``gateway run`` / ``cron run`` do this in ``_prepare_agent_startup``
+    (hermes_cli/main.py) and the gateway-hosted dispatcher in
+    ``GatewayStartupMixin._register_config_hooks``; ``hermes kanban`` is not an agent
+    command, so ``claim``, ``dispatch`` and ``daemon --force`` must do it themselves before
+    they resolve a workspace — otherwise a ``kanban_worktree_created`` shell hook the
+    operator configured never sees the worktree these paths create, and the unseeded tree
+    is then reused by every later tick. Python plugin hooks need nothing here
+    (``invoke_hook`` discovers plugins lazily). Consent is the bridge's own: the allowlist,
+    ``--accept-hooks``, ``HERMES_ACCEPT_HOOKS``, ``hooks_auto_accept``, or the TTY prompt.
+    Never raises: a broken hooks block must not take the dispatcher down.
+    """
+    try:
+        from hermes_cli.config import load_config
+        from agent.shell_hooks import register_from_config
+
+        register_from_config(load_config(), accept_hooks=bool(getattr(args, "accept_hooks", False)))
+    except Exception:
+        logging.getLogger(__name__).debug("shell-hook registration failed for hermes kanban", exc_info=True)
+
+
 def _cmd_dispatch(args: argparse.Namespace) -> int:
     # Honour kanban.default_assignee, kanban.max_in_progress,
     # kanban.max_in_progress_per_profile and kanban.max_spawn with the same
@@ -81,6 +105,8 @@ def _cmd_dispatch(args: argparse.Namespace) -> int:
     except Exception:
         default_assignee = max_in_progress_per_profile = max_in_progress = None
         max_spawn = getattr(args, "max", None)
+    if not args.dry_run:  # a dry run creates no workspace, so there is nothing for a hook to see
+        _register_config_hooks(args)
     with kbc.connect_closing() as conn:
         res = kbd.dispatch_once(
             conn,
@@ -181,6 +207,7 @@ def _cmd_daemon(args: argparse.Namespace) -> int:
     # Init before printing "started" so the DB path is right and init errors
     # surface immediately.
     kb.init_db()
+    _register_config_hooks(args)
 
     pidfile = getattr(args, "pidfile", None)
     if pidfile:

hermes_cli/plugins.py

@@ -160,6 +160,14 @@ def _install_plugin_debug_handler(force: bool = False) -> None:
     # spawn; completed/blocked fire in the WORKER (or whichever process drove it). Kwargs: task_id,
     # board, assignee, run_id, profile_name; completed adds summary, blocked adds reason.
     "kanban_task_claimed", "kanban_task_completed", "kanban_task_blocked",
+    # kanban_worktree_created: the one kanban GATE (hermes_cli.kanban_db_workspace). Fires synchronously
+    # right after `git worktree add` succeeds and BEFORE the worker spawns (dispatcher, or `hermes kanban
+    # claim`) so an operator can seed what a fresh checkout lacks (gitignored .env files, caches). Kwargs:
+    # task_id, board, profile_name, worktree_path, repo_root, branch. Return None to accept; {"action":
+    # "block", "message"} (shell bridge: exit 2 / fail_closed) makes the caller remove the worktree and
+    # fail the card with the message; the retry re-creates and re-fires. A RAISING callback is logged and
+    # skipped like any other hook, so plugins must return the directive. Not timeout-bounded (kanban_*).
+    "kanban_worktree_created",
     # Kanban worker/mutation/tick observers; returns ignored; fire sites short-circuit on
     # has_hook(). Kwargs: task_id, profile_name, board, assignee, run_id plus, per hook:
     # worker_spawned (DISPATCHER, after PID persisted, inside the dispatch lock — stay fast):

tests/hermes_cli/test_hooks_cli.py

@@ -201,6 +201,21 @@ def test_unallowlisted_script_is_not_executed(self, tmp_path):
         assert "not allowlisted" in out.lower()
         assert "skipped JSON smoke test" in out
 
+    def test_block_on_synthetic_payload_is_reported_as_a_block(self, tmp_path):
+        """An allowlisted gate that exits 2 on doctor's synthetic payload (the documented
+        ``kanban_worktree_created`` seeder finds no ``/tmp/repo/.env``) must be reported as the
+        block the dispatcher would receive, not as a clean observer-only run."""
+        script = _hook_script(
+            tmp_path, "#!/usr/bin/env bash\necho 'no .env in /tmp/repo' >&2\nexit 2\n",
+        )
+        shell_hooks._record_approval("kanban_worktree_created", str(script))
+        cfg = {"hooks": {"kanban_worktree_created": [{"command": str(script)}]}}
+        with patch("hermes_cli.config.load_config", return_value=cfg):
+            out = _run(SimpleNamespace(hooks_action="doctor"))
+
+        assert "returned a block on the synthetic payload (exit=2" in out
+        assert "no .env in /tmp/repo" in out
+        assert "observer-only" not in out
 
 def test_print_run_result_shows_decision_for_error_and_timeout():
     """A failing hook's decision must be printed, not hidden by early returns (#115968).