feat(kanban): talk to a running worker without a restart

A running worker now polls its comment thread and folds new operator notes
into the live turn via the OUT-OF-BAND steer channel (list_comments_after +
a heartbeat-driven bridge, watermarked so history isn't re-injected and the
worker's own notes are skipped). No block→comment→unblock dance. Desktop's
composer sends notes live ("delivered within a few seconds") with "Requeue
with note" as the restart option and a help tooltip.

apps/desktop/src/plugins/kanban/drawer.tsx

@@ -279,7 +279,23 @@ function AssigneeMenu({
 // Mirrors the review pane's commit-message field: one row tall to start
 // (button-height), CSS field-sizing grows it with content, button hugs the
 // bottom edge as it grows.
-function CommentComposer({ onSubmit, pending }: { onSubmit: (body: string) => void; pending: boolean }) {
+//
+// On a RUNNING task the worker polls its comment thread and folds new notes
+// into the live turn (OUT-OF-BAND steer), so a plain note reaches the agent
+// mid-run within a few seconds — no block/unblock dance. `onRequeue` is the
+// heavier option: post the note AND reclaim so the task restarts from scratch
+// with the note in context (use when the current run has gone off the rails).
+function CommentComposer({
+  onRequeue,
+  onSubmit,
+  pending,
+  running
+}: {
+  onRequeue?: (body: string) => void
+  onSubmit: (body: string) => void
+  pending: boolean
+  running?: boolean
+}) {
   const [body, setBody] = useState('')
 
   const submit = () => {
@@ -291,31 +307,53 @@ function CommentComposer({ onSubmit, pending }: { onSubmit: (body: string) => vo
     }
   }
 
+  const requeue = () => {
+    const trimmed = body.trim()
+
+    if (trimmed && !pending && onRequeue) {
+      onRequeue(trimmed)
+      setBody('')
+    }
+  }
+
   return (
-    <div className="relative">
-      <Textarea
-        className="field-sizing-content max-h-40 min-h-0 resize-none pr-[5rem]"
-        onChange={event => setBody(event.target.value)}
-        onKeyDown={event => {
-          if (event.key === 'Enter' && !event.shiftKey) {
-            event.preventDefault()
-            submit()
-          }
-        }}
-        placeholder="Add a comment…"
-        rows={1}
-        size="sm"
-        value={body}
-      />
-      <Button
-        className="absolute top-1 right-1"
-        disabled={!body.trim() || pending}
-        onClick={submit}
-        size="xs"
-        variant="secondary"
-      >
-        Comment
-      </Button>
+    <div className="flex flex-col gap-1.5">
+      <div className="relative">
+        <Textarea
+          className={cn('field-sizing-content max-h-40 min-h-0 resize-none', running ? 'pr-[3.5rem]' : 'pr-[5rem]')}
+          onChange={event => setBody(event.target.value)}
+          onKeyDown={event => {
+            if (event.key === 'Enter' && !event.shiftKey) {
+              event.preventDefault()
+              submit()
+            }
+          }}
+          placeholder={running ? 'Message the running worker…' : 'Add a comment…'}
+          rows={1}
+          size="sm"
+          value={body}
+        />
+        <Button
+          className="absolute top-1 right-1"
+          disabled={!body.trim() || pending}
+          onClick={submit}
+          size="xs"
+          variant="secondary"
+        >
+          {running ? 'Send' : 'Comment'}
+        </Button>
+      </div>
+      {running && onRequeue && (
+        <div className="flex items-center justify-between gap-2">
+          <span className="text-[0.625rem] leading-tight text-(--ui-text-quaternary)">
+            Delivered to the running worker within a few seconds.
+          </span>
+          <Button className="shrink-0" disabled={!body.trim() || pending} onClick={requeue} size="xs" variant="outline">
+            <Codicon name="debug-restart" size="0.7rem" />
+            Requeue with note
+          </Button>
+        </div>
+      )}
     </div>
   )
 }
@@ -434,7 +472,8 @@ function AttachmentsSection({
 // Rough effort estimate via the auxiliary (auto-routed) model. Tokens +
 // complexity, never dollars — providers don't report cost reliably. Gated
 // behind an explicit click + disclaimer since it makes a model call. The
-// control keeps a stable footprint (spinner swaps in place) so nothing jumps.
+// control keeps a stable footprint (spinner swaps in place) so there's no
+// layout jump when it runs.
 function EstimateSection({ id }: { id: string }) {
   const [result, setResult] = useState<null | TaskEstimate>(null)
 
@@ -587,6 +626,21 @@ export function TaskDrawer({
     onSuccess: invalidate
   })
 
+  // "Note & requeue" for a running task: post the note, then reclaim so the
+  // dispatcher re-runs it with the note in the worker's context — the one-click
+  // replacement for the block → comment → unblock dance.
+  const requeueMut = useMutation({
+    mutationFn: async (body: string) => {
+      await addComment(id!, body)
+      await reclaimTask(id!)
+    },
+    onError: err => host.notify({ kind: 'error', message: errText(err) }),
+    onSuccess: () => {
+      host.notify({ kind: 'info', message: 'Note posted — worker requeued' })
+      invalidate()
+    }
+  })
+
   const uploadMut = useMutation({
     mutationFn: async (file: File) =>
       uploadAttachment(id!, {
@@ -776,7 +830,22 @@ export function TaskDrawer({
               </Section>
             )}
 
-            <Section label={`Comments · ${detail.comments.length}`}>
+            <Section
+              action={
+                <Tip
+                  label={
+                    running
+                      ? 'This task is running. Your note is folded into the worker’s current turn within a few seconds — no block/unblock dance. “Requeue with note” instead restarts the task from scratch with your note in context.'
+                      : 'Comments are added to the task thread. When a worker picks the task up it reads them as part of its context.'
+                  }
+                >
+                  <span className="grid size-5 place-items-center rounded text-(--ui-text-quaternary) hover:text-(--ui-text-secondary)">
+                    <Codicon name="question" size="0.8rem" />
+                  </span>
+                </Tip>
+              }
+              label={`Comments · ${detail.comments.length}`}
+            >
               {detail.comments.length > 0 && (
                 <ul className="flex flex-col gap-2">
                   {detail.comments.map(comment => (
@@ -790,7 +859,12 @@ export function TaskDrawer({
                   ))}
                 </ul>
               )}
-              <CommentComposer onSubmit={body => commentMut.mutate(body)} pending={commentMut.isPending} />
+              <CommentComposer
+                onRequeue={body => requeueMut.mutate(body)}
+                onSubmit={body => commentMut.mutate(body)}
+                pending={commentMut.isPending || requeueMut.isPending}
+                running={running}
+              />
             </Section>
 
             {detail.events.length > 0 && (

hermes_cli/kanban_db.py

@@ -3580,6 +3580,33 @@ def list_comments(conn: sqlite3.Connection, task_id: str) -> list[Comment]:
     ]
 
 
+def list_comments_after(
+    conn: sqlite3.Connection, task_id: str, *, after_id: int = 0
+) -> list[Comment]:
+    """Return comments on ``task_id`` with ``id > after_id`` (ascending).
+
+    Keyed on the monotonic rowid rather than ``created_at`` so a same-second
+    burst can't be skipped. Used by the live worker bridge to fold new
+    operator notes into a running task without a restart (see
+    ``tools.kanban_tools.inject_new_comments_from_env``).
+    """
+    rows = conn.execute(
+        "SELECT id, task_id, author, body, created_at FROM task_comments "
+        "WHERE task_id = ? AND id > ? ORDER BY id ASC",
+        (task_id, int(after_id)),
+    ).fetchall()
+    return [
+        Comment(
+            id=r["id"],
+            task_id=r["task_id"],
+            author=r["author"],
+            body=r["body"],
+            created_at=r["created_at"],
+        )
+        for r in rows
+    ]
+
+
 # ---------------------------------------------------------------------------
 # Attachments
 # ---------------------------------------------------------------------------

run_agent.py

@@ -3540,8 +3540,14 @@ class AIAgent:
         self._last_activity_desc = desc
         if os.environ.get("HERMES_KANBAN_TASK"):
             try:
-                from tools.kanban_tools import heartbeat_current_worker_from_env
+                from tools.kanban_tools import (
+                    heartbeat_current_worker_from_env,
+                    inject_new_comments_from_env,
+                )
                 heartbeat_current_worker_from_env()
+                # Fold any new operator notes into the running turn (OUT-OF-BAND
+                # steer) so the user can talk to a live task without a restart.
+                inject_new_comments_from_env(self)
             except Exception:
                 # Never let the bridge break the agent loop.  The function
                 # already swallows exceptions internally; this outer guard

tests/hermes_cli/test_kanban_comment_queries.py

@@ -0,0 +1,54 @@
+"""Comment-watermark queries in kanban_db.
+
+``list_comments_after`` backs the live worker bridge: it returns only comments
+newer than a cursor so a running worker folds in new operator notes without
+re-reading the whole thread.
+"""
+
+from __future__ import annotations
+
+import sys
+from pathlib import Path
+
+import pytest
+
+_WORKTREE = Path(__file__).resolve().parents[2]
+if str(_WORKTREE) not in sys.path:
+    sys.path.insert(0, str(_WORKTREE))
+
+from hermes_cli import kanban_db as kb
+
+
+@pytest.fixture
+def fresh_home(tmp_path, monkeypatch):
+    home = tmp_path / "hermes_home"
+    home.mkdir()
+    monkeypatch.setenv("HERMES_HOME", str(home))
+    monkeypatch.setattr(Path, "home", lambda: tmp_path)
+    for var in ("HERMES_KANBAN_DB", "HERMES_KANBAN_WORKSPACES_ROOT", "HERMES_KANBAN_HOME", "HERMES_KANBAN_BOARD"):
+        monkeypatch.delenv(var, raising=False)
+    try:
+        import hermes_constants
+        hermes_constants._cached_default_hermes_root = None  # type: ignore[attr-defined]
+    except Exception:
+        pass
+    kb._INITIALIZED_PATHS.clear()
+    return home
+
+
+def test_list_comments_after_cursor(fresh_home):
+    conn = kb.connect()
+    try:
+        tid = kb.create_task(conn, title="chat")
+        c1 = kb.add_comment(conn, tid, author="alice", body="first")
+        c2 = kb.add_comment(conn, tid, author="bob", body="second")
+
+        assert [c.id for c in kb.list_comments_after(conn, tid, after_id=0)] == [c1, c2]
+
+        newer = kb.list_comments_after(conn, tid, after_id=c1)
+        assert [c.id for c in newer] == [c2]
+        assert newer[0].body == "second"
+
+        assert kb.list_comments_after(conn, tid, after_id=c2) == []
+    finally:
+        conn.close()

tests/tools/test_kanban_comment_injection.py

@@ -0,0 +1,124 @@
+"""Live operator-note injection into a running kanban worker.
+
+``tools.kanban_tools.inject_new_comments_from_env`` polls the worker's task
+for comments added *after* the run started and folds them into the live turn
+via the agent's OUT-OF-BAND steer channel — so a user can talk to a running
+task without the block→comment→unblock dance or a restart.
+
+Verifies: no-op off a worker, watermark seeding (history isn't re-injected),
+new comments steer, and own-authored comments are skipped.
+"""
+
+from __future__ import annotations
+
+import sys
+from pathlib import Path
+
+import pytest
+
+_WORKTREE = Path(__file__).resolve().parents[2]
+if str(_WORKTREE) not in sys.path:
+    sys.path.insert(0, str(_WORKTREE))
+
+from hermes_cli import kanban_db as kb
+import tools.kanban_tools as kt
+
+
+class FakeAgent:
+    def __init__(self):
+        self.steers: list[str] = []
+
+    def steer(self, text: str) -> bool:
+        self.steers.append(text)
+        return True
+
+
+@pytest.fixture
+def worker_home(tmp_path, monkeypatch):
+    home = tmp_path / "hermes_home"
+    home.mkdir()
+    monkeypatch.setenv("HERMES_HOME", str(home))
+    monkeypatch.setattr(Path, "home", lambda: tmp_path)
+    for var in ("HERMES_KANBAN_DB", "HERMES_KANBAN_WORKSPACES_ROOT", "HERMES_KANBAN_HOME", "HERMES_KANBAN_BOARD"):
+        monkeypatch.delenv(var, raising=False)
+    try:
+        import hermes_constants
+        hermes_constants._cached_default_hermes_root = None  # type: ignore[attr-defined]
+    except Exception:
+        pass
+    kb._INITIALIZED_PATHS.clear()
+    # Reset module-level poll state so tests don't leak into each other.
+    kt._comment_watermark.clear()
+    kt._comment_poll_last_attempt = 0.0
+    return home
+
+
+def _unthrottle():
+    """Bypass the inter-poll rate limit for deterministic tests."""
+    kt._comment_poll_last_attempt = 0.0
+
+
+def test_noop_without_worker_env(worker_home, monkeypatch):
+    monkeypatch.delenv("HERMES_KANBAN_TASK", raising=False)
+    agent = FakeAgent()
+    assert kt.inject_new_comments_from_env(agent) is False
+    assert agent.steers == []
+
+
+def test_seed_then_inject_new_comment(worker_home, monkeypatch):
+    conn = kb.connect()
+    try:
+        tid = kb.create_task(conn, title="live task")
+        kb.add_comment(conn, tid, author="desktop", body="pre-existing note")
+    finally:
+        conn.close()
+
+    monkeypatch.setenv("HERMES_KANBAN_TASK", tid)
+    monkeypatch.setenv("HERMES_PROFILE", "worker-bot")
+    agent = FakeAgent()
+
+    # First poll seeds the watermark past the existing thread — no injection.
+    _unthrottle()
+    assert kt.inject_new_comments_from_env(agent) is False
+    assert agent.steers == []
+
+    conn = kb.connect()
+    try:
+        kb.add_comment(conn, tid, author="desktop", body="actually use the v2 API")
+    finally:
+        conn.close()
+
+    _unthrottle()
+    assert kt.inject_new_comments_from_env(agent) is True
+    assert len(agent.steers) == 1
+    assert "v2 API" in agent.steers[0]
+
+    # Watermark advanced — a re-poll with no new comments injects nothing.
+    _unthrottle()
+    assert kt.inject_new_comments_from_env(agent) is False
+    assert len(agent.steers) == 1
+
+
+def test_skips_own_authored_comments(worker_home, monkeypatch):
+    conn = kb.connect()
+    try:
+        tid = kb.create_task(conn, title="echo guard")
+    finally:
+        conn.close()
+
+    monkeypatch.setenv("HERMES_KANBAN_TASK", tid)
+    monkeypatch.setenv("HERMES_PROFILE", "worker-bot")
+    agent = FakeAgent()
+
+    _unthrottle()
+    kt.inject_new_comments_from_env(agent)  # seed
+
+    conn = kb.connect()
+    try:
+        kb.add_comment(conn, tid, author="worker-bot", body="i did a thing")
+    finally:
+        conn.close()
+
+    _unthrottle()
+    assert kt.inject_new_comments_from_env(agent) is False
+    assert agent.steers == []

tools/kanban_tools.py

@@ -320,6 +320,85 @@ def heartbeat_current_worker_from_env() -> bool:
         return False
 
 
+# Live operator-note injection: poll the worker's task for new comments and
+# fold them into the running agent via the OUT-OF-BAND steer channel, so a user
+# can "talk to" a running kanban task without the block → comment → unblock
+# dance (or a restart). Rate-limited on its own (tighter than the 60s heartbeat
+# so notes land within a few seconds), watermarked per task id.
+_COMMENT_POLL_MIN_INTERVAL_SECONDS = 6.0
+_comment_poll_last_attempt: float = 0.0
+# task_id -> highest comment id already seen (seeded on first poll so history
+# already present in build_worker_context isn't re-injected).
+_comment_watermark: dict[str, int] = {}
+
+
+def inject_new_comments_from_env(agent: Any) -> bool:
+    """Fold new operator comments on the current worker's task into ``agent``.
+
+    Best-effort and self-gating: no-op unless this process is a kanban worker
+    (``HERMES_KANBAN_TASK`` set) and ``agent`` exposes ``steer``. Returns True
+    if a steer was injected, else False. Never raises into the agent loop.
+
+    The first poll only *seeds* the watermark to the newest existing comment —
+    those are already in the worker's context — so only comments added after
+    the run started are injected. The worker's own authored comments (matched
+    by ``HERMES_PROFILE``) are skipped to avoid echoing itself.
+    """
+    tid = os.environ.get("HERMES_KANBAN_TASK")
+    if not tid or agent is None or not hasattr(agent, "steer"):
+        return False
+    global _comment_poll_last_attempt
+    import time as _time
+    now = _time.monotonic()
+    if (now - _comment_poll_last_attempt) < _COMMENT_POLL_MIN_INTERVAL_SECONDS:
+        return False
+    _comment_poll_last_attempt = now
+
+    seen = _comment_watermark.get(tid)
+    try:
+        kb, conn = _connect()
+        try:
+            rows = kb.list_comments_after(conn, tid, after_id=seen or 0)
+        finally:
+            try:
+                conn.close()
+            except Exception:
+                pass
+    except Exception:
+        logger.debug("comment-inject: bridge failed", exc_info=True)
+        return False
+
+    if seen is None:
+        # First poll for this task: seed past the existing thread, inject nothing.
+        _comment_watermark[tid] = max((c.id for c in rows), default=0)
+        return False
+    if not rows:
+        return False
+
+    # Advance the watermark past everything we just read (including our own
+    # notes) so nothing is re-injected next poll.
+    _comment_watermark[tid] = max(c.id for c in rows)
+
+    own = (os.environ.get("HERMES_PROFILE") or "").strip()
+    fresh = [c for c in rows if (c.author or "").strip() != own and (c.body or "").strip()]
+    if not fresh:
+        return False
+
+    lines = [f"- {c.author or 'operator'}: {c.body.strip()}" for c in fresh]
+    note = (
+        "New note"
+        + ("s" if len(fresh) > 1 else "")
+        + " on your kanban task from the operator (delivered mid-run). "
+        + "Take it into account for the work you're doing right now:\n"
+        + "\n".join(lines)
+    )
+    try:
+        return bool(agent.steer(note))
+    except Exception:
+        logger.debug("comment-inject: steer failed", exc_info=True)
+        return False
+
+
 def _ok(**fields: Any) -> str:
     return json.dumps({"ok": True, **fields})