Merge remote-tracking branch 'origin/main' into bb/gui

# Conflicts:
#       apps/dashboard/package-lock.json
#       apps/dashboard/package.json
#       apps/dashboard/src/components/BottomPickSheet.tsx
#       apps/dashboard/src/hooks/useBelowBreakpoint.ts
#       gateway/platforms/telegram.py
#       hermes_cli/gateway.py
#       hermes_cli/web_server.py
#       nix/web.nix
#       scripts/install.ps1
#       tests/gateway/test_telegram_thread_fallback.py
#       tui_gateway/server.py

plugins/kanban/dashboard/dist/index.js

@@ -51,6 +51,24 @@
     return str;
   }
 
+  // ``fetchJSON`` throws ``Error("<status>: <raw body>")`` on non-2xx, and
+  // FastAPI bodies look like ``{"detail":"<message>"}``.  Pull the
+  // human-readable message out so banners/toasts don't have to leak HTTP
+  // plumbing at the user (e.g. ``409: {"detail":"…"}``).  See #26744.
+  function parseApiErrorMessage(err) {
+    const raw = (err && err.message) ? String(err.message) : String(err || "");
+    const m = raw.match(/^(\d{3}):\s*(.*)$/s);
+    const body = m ? m[2] : raw;
+    try {
+      const parsed = JSON.parse(body);
+      if (parsed && typeof parsed.detail === "string") return parsed.detail;
+      if (parsed && parsed.detail && typeof parsed.detail.message === "string") {
+        return parsed.detail.message;
+      }
+    } catch (_e) { /* not JSON — fall through to raw body */ }
+    return body || raw;
+  }
+
   // Order matches BOARD_COLUMNS in plugin_api.py.
   const COLUMN_ORDER = ["triage", "todo", "ready", "running", "blocked", "done"];
   // English fallback dictionaries — used when the i18n catalog is missing
@@ -83,6 +101,12 @@
     completion_blocked_hallucination: "⚠ Completion blocked — phantom card ids",
     suspected_hallucinated_references: "⚠ Prose referenced phantom card ids",
   };
+  const FALLBACK_TRASH = {
+    label: "Trash",
+    title: "Drag a card here to permanently delete it",
+    confirm: "Permanently delete this task? This cannot be undone.",
+    dropHint: "Drop to delete",
+  };
   const DIAGNOSTIC_EVENT_KIND_KEYS = {
     completion_blocked_hallucination: "completionBlockedHallucination",
     suspected_hallucinated_references: "suspectedHallucinatedReferences",
@@ -331,10 +355,12 @@
         const under = document.elementFromPoint(ev.clientX, ev.clientY);
         proxy.style.display = "";
         const col = under && under.closest && under.closest("[data-kanban-column]");
-        if (col !== lastTarget) {
+        const trash = under && under.closest && under.closest("[data-kanban-trash]");
+        const target = col || trash;
+        if (target !== lastTarget) {
           if (lastTarget) lastTarget.classList.remove("hermes-kanban-column--drop");
-          if (col) col.classList.add("hermes-kanban-column--drop");
-          lastTarget = col;
+          if (target) target.classList.add("hermes-kanban-column--drop");
+          lastTarget = target;
         }
       }
       function up() {
@@ -344,10 +370,18 @@
         if (lastTarget) {
           lastTarget.classList.remove("hermes-kanban-column--drop");
           const status = lastTarget.getAttribute("data-kanban-column");
-          lastTarget.dispatchEvent(new CustomEvent("hermes-kanban:drop", {
-            detail: { taskId, status },
-            bubbles: true,
-          }));
+          const isTrash = lastTarget.hasAttribute("data-kanban-trash");
+          if (isTrash) {
+            lastTarget.dispatchEvent(new CustomEvent("hermes-kanban:delete", {
+              detail: { taskId },
+              bubbles: true,
+            }));
+          } else if (status) {
+            lastTarget.dispatchEvent(new CustomEvent("hermes-kanban:drop", {
+              detail: { taskId, status },
+              bubbles: true,
+            }));
+          }
         }
         proxy.remove();
       }
@@ -413,7 +447,7 @@
 
   function KanbanPage() {
     const { t } = useI18n();
-    const [board, setBoard] = useState(() => readSelectedBoard() || "default");
+    const [board, setBoard] = useState(() => readSelectedBoard() || null);
     const [boardList, setBoardList] = useState([]);      // [{slug, name, counts, ...}]
     const [showNewBoard, setShowNewBoard] = useState(false);
 
@@ -494,11 +528,16 @@
       return SDK.fetchJSON(withBoard(`${API}/boards`, board))
         .then(function (data) {
           const boards = (data && data.boards) || [];
+          const storedBoard = readSelectedBoard();
           setBoardList(boards);
+          if (!storedBoard && !board && data && data.current) {
+            setBoard(data.current);
+            return;
+          }
           // If the stored slug isn't in the list any longer (board was
           // deleted in the CLI while dashboard was open), fall back to
           // default so the UI doesn't hang on a 404.
-          if (board !== "default" && !boards.find(function (b) { return b.slug === board; })) {
+          if (board && board !== "default" && !boards.find(function (b) { return b.slug === board; })) {
             setBoard("default");
             writeSelectedBoard("default");
           }
@@ -633,7 +672,7 @@
         headers: { "Content-Type": "application/json" },
         body: JSON.stringify(patch),
       }).catch(function (err) {
-        setError(tx(t, "moveFailed", "Move failed: ") + (err.message || err));
+        setError(tx(t, "moveFailed", "Move failed: ") + parseApiErrorMessage(err));
         loadBoard();
       });
     }, [loadBoard, board, t]);
@@ -873,6 +912,32 @@
       });
     }, [board, loadBoardList, switchBoard]);
 
+   const deleteTask = useCallback(function (taskId) {
+     if (!window.confirm(tx(t, "trash.confirm", FALLBACK_TRASH.confirm))) return Promise.resolve();
+     return SDK.fetchJSON(`${API}/tasks/${encodeURIComponent(taskId)}`, {
+       method: "DELETE",
+     }).then(function () {
+       loadBoard();
+       setSelectedIds(function (prev) {
+         const next = new Set(prev);
+         next.delete(taskId);
+         return next;
+       });
+     }).catch(function (e) { setError(String(e.message || e)); });
+   }, [board, loadBoard, t]);
+
+    const deleteSelected = useCallback(function (count) {
+      if (selectedIds.size === 0) return Promise.resolve();
+      if (!window.confirm(tx(t, "trash.confirmMany", "Permanently delete {n} selected tasks? This cannot be undone.", { n: count }))) return Promise.resolve();
+      const ids = Array.from(selectedIds);
+      setSelectedIds(new Set());
+      return Promise.all(ids.map(function (id) {
+        return SDK.fetchJSON(`${API}/tasks/${encodeURIComponent(id)}`, { method: "DELETE" });
+      })).then(function () {
+        loadBoard();
+      }).catch(function (e) { setError(String(e.message || e)); });
+    }, [selectedIds, board, loadBoard, t]);
+
     // --- render -------------------------------------------------------------
     if (loading && !boardData) {
       return h("div", { className: "p-8 text-sm text-muted-foreground" },
@@ -927,13 +992,14 @@
           },
           onRefresh: loadBoard,
         }),
-        selectedIds.size > 0 ? h(BulkActionBar, {
-          count: selectedIds.size,
-          assignees: (boardData && boardData.assignees) || [],
-          onApply: applyBulk,
-          onClear: clearSelected,
-          onSelectAllVisible: selectAllVisible,
-        }) : null,
+       selectedIds.size > 0 ? h(BulkActionBar, {
+         count: selectedIds.size,
+         assignees: (boardData && boardData.assignees) || [],
+         onApply: applyBulk,
+         onClear: clearSelected,
+         onSelectAllVisible: selectAllVisible,
+         onDelete: deleteSelected,
+       }) : null,
         error ? h("div", { className: "text-xs text-destructive px-2" }, error) : null,
         h(BoardColumns, {
           board: filteredBoard,
@@ -948,6 +1014,7 @@
           selectAllInColumn,
           onMove: moveTask,
           onMoveSelected: moveSelected,
+          onDelete: deleteTask,
           onOpen: setSelectedTaskId,
           onCreate: createTask,
           allTasks: boardData.columns.reduce(function (acc, c) { return acc.concat(c.tasks); }, []),
@@ -1588,10 +1655,12 @@
                   saveSettings({ auto_decompose: !!e.target.checked });
                 },
               }),
-              settings.auto_decompose ? "Auto (default)" : "Manual",
+              "Auto-decompose triage tasks",
             ),
             h("div", { className: "text-[10px] text-muted-foreground" },
-              "When on, the dispatcher decomposes new triage tasks automatically."),
+              settings.auto_decompose
+                ? "The dispatcher decomposes new triage tasks automatically."
+                : "Triage tasks stay in triage until you click ⚗ Decompose."),
           ),
         ) : h("div", { className: "text-xs text-muted-foreground" },
           "Loading…"),
@@ -2002,6 +2071,14 @@
         size: "sm",
         title: "Archive selected tasks. They disappear from the default board view but remain in the database.",
       }, tx(t, "archive", "Archive")),
+      h(Button, {
+        onClick: function () {
+          props.onDelete(props.count);
+        },
+        size: "sm",
+        variant: "destructive",
+        title: "Permanently delete selected tasks. This cannot be undone.",
+      }, tx(t, "delete", "Delete")),
       h("div", { className: "hermes-kanban-bulk-priority",
                  title: "Set priority on selected tasks. Higher = claimed first." },
         h(Input, {
@@ -2066,6 +2143,65 @@
     );
   }
 
+  // -------------------------------------------------------------------------
+  // Trash Drop Zone
+  // -------------------------------------------------------------------------
+
+  function TrashDropZone(props) {
+    const { t } = useI18n();
+    const [dragOver, setDragOver] = useState(false);
+    const zoneRef = useRef(null);
+
+    useEffect(function () {
+      if (!zoneRef.current) return undefined;
+      const el = zoneRef.current;
+      function onTouchDelete(e) {
+        const taskId = e.detail && e.detail.taskId;
+        if (taskId && props.onDelete) props.onDelete(taskId);
+      }
+      el.addEventListener("hermes-kanban:delete", onTouchDelete);
+      return function () { el.removeEventListener("hermes-kanban:delete", onTouchDelete); };
+    }, [props.onDelete]);
+
+    const handleDragOver = function (e) {
+      e.preventDefault();
+      e.dataTransfer.dropEffect = "move";
+      if (!dragOver) setDragOver(true);
+    };
+    const handleDragLeave = function () { setDragOver(false); };
+    const handleDrop = function (e) {
+      e.preventDefault();
+      setDragOver(false);
+      const taskId = e.dataTransfer.getData(MIME_TASK);
+      if (!taskId) return;
+      if (props.selectedIds && props.selectedIds.has(taskId) && props.selectedIds.size > 1) {
+        if (window.confirm(tx(t, "trash.confirmMany", "Permanently delete {n} selected tasks? This cannot be undone.", { n: props.selectedIds.size }))) {
+          const ids = Array.from(props.selectedIds);
+          Promise.all(ids.map(function (id) { return props.onDelete(id); })).catch(function () {});
+        }
+      } else {
+        props.onDelete(taskId);
+      }
+    };
+
+    return h("div", {
+      ref: zoneRef,
+      "data-kanban-trash": "true",
+      className: cn(
+        "hermes-kanban-trash",
+        dragOver ? "hermes-kanban-trash--drop" : "",
+        props.draggingTaskId ? "hermes-kanban-trash--active" : "",
+      ),
+      onDragOver: handleDragOver,
+      onDragLeave: handleDragLeave,
+      onDrop: handleDrop,
+    },
+      h("span", { className: "hermes-kanban-trash-icon" }, "🗑️"),
+      h("span", { className: "hermes-kanban-trash-label" },
+        tx(t, "trash.dropHint", FALLBACK_TRASH.dropHint)),
+    );
+  }
+
   // -------------------------------------------------------------------------
   // Columns
   // -------------------------------------------------------------------------
@@ -2099,6 +2235,11 @@
           allTasks: props.allTasks,
         });
       }),
+      h(TrashDropZone, {
+        draggingTaskId: props.draggingTaskId,
+        selectedIds: props.selectedIds,
+        onDelete: props.onDelete,
+      }),
     );
   }
 
@@ -2586,6 +2727,11 @@
     const [data, setData] = useState(null);
     const [loading, setLoading] = useState(true);
     const [err, setErr] = useState(null);
+    // Surface PATCH failures (e.g. 409 "parent not done") right next to
+    // the drawer's action row — without it, the drawer's only error
+    // surface (``err``) is hidden behind the loaded ``data`` and the
+    // Ready/Block/Complete buttons feel like no-ops.  See #26744.
+    const [patchErr, setPatchErr] = useState(null);
     const [newComment, setNewComment] = useState("");
     const [editing, setEditing] = useState(false);
     // Home-channel notification toggles. homeChannels is the list of platforms
@@ -2597,7 +2743,7 @@
 
     const load = useCallback(function () {
       return SDK.fetchJSON(withBoard(`${API}/tasks/${encodeURIComponent(props.taskId)}`, boardSlug))
-        .then(function (d) { setData(d); setErr(null); })
+        .then(function (d) { setData(d); setErr(null); setPatchErr(null); })
         .catch(function (e) { setErr(String(e.message || e)); })
         .finally(function () { setLoading(false); });
     }, [props.taskId, boardSlug]);
@@ -2641,11 +2787,13 @@
       }
       const finalPatch = withCompletionSummary(patch, 1);
       if (!finalPatch) return Promise.resolve();
+      setPatchErr(null);
       return SDK.fetchJSON(withBoard(`${API}/tasks/${encodeURIComponent(props.taskId)}`, boardSlug), {
         method: "PATCH",
         headers: { "Content-Type": "application/json" },
         body: JSON.stringify(finalPatch),
-      }).then(function () { load(); props.onRefresh(); });
+      }).then(function () { load(); props.onRefresh(); })
+        .catch(function (e) { setPatchErr(parseApiErrorMessage(e)); });
     };
 
     // Triage specifier — calls the auxiliary LLM to flesh out a rough

plugins/kanban/dashboard/dist/style.css

@@ -63,13 +63,18 @@
 /* ---- Columns layout -------------------------------------------------- */
 
 .hermes-kanban-columns {
-  display: grid;
-  grid-template-columns: repeat(auto-fit, minmax(260px, 1fr));
+  display: flex;
   gap: 0.75rem;
   align-items: start;
+  overflow-x: auto;
+  scrollbar-width: none;
+}
+.hermes-kanban-columns::-webkit-scrollbar {
+  display: none;
 }
 
 .hermes-kanban-column {
+  flex: 0 0 280px;
   display: flex;
   flex-direction: column;
   background: color-mix(in srgb, var(--color-card) 85%, transparent);
@@ -1493,3 +1498,44 @@
   font-size: 0.7rem;
   cursor: pointer;
 }
+
+/* ---- Trash drop zone ------------------------------------------------- */
+
+.hermes-kanban-trash {
+  display: flex;
+  flex-direction: column;
+  align-items: center;
+  justify-content: center;
+  gap: 0.35rem;
+  padding: 0.75rem 0.5rem;
+  border: 2px dashed var(--color-border);
+  border-radius: var(--radius);
+  background: color-mix(in srgb, var(--color-card) 85%, transparent);
+  color: var(--color-muted-foreground);
+  font-size: 0.75rem;
+  min-height: 80px;
+  opacity: 0.5;
+  transition: opacity 120ms ease, border-color 120ms ease, background-color 120ms ease;
+  user-select: none;
+  pointer-events: none;
+}
+
+.hermes-kanban-trash--active {
+  opacity: 1;
+  pointer-events: auto;
+}
+
+.hermes-kanban-trash--drop {
+  border-color: var(--color-destructive, #d14a4a);
+  background: color-mix(in srgb, var(--color-destructive, #d14a4a) 8%, var(--color-card));
+  color: var(--color-destructive, #d14a4a);
+}
+
+.hermes-kanban-trash-icon {
+  font-size: 1.25rem;
+  line-height: 1;
+}
+
+.hermes-kanban-trash-label {
+  font-weight: 500;
+}

plugins/kanban/dashboard/plugin_api.py

@@ -49,6 +49,7 @@ from fastapi import APIRouter, HTTPException, Query, WebSocket, WebSocketDisconn
 from pydantic import BaseModel, Field
 
 from hermes_cli import kanban_db
+from hermes_cli import kanban_diagnostics as kd
 
 log = logging.getLogger(__name__)
 
@@ -129,8 +130,14 @@ def _conn(board: Optional[str] = None):
 
 # Columns shown by the dashboard, in left-to-right order. "archived" is
 # available via a filter toggle rather than a visible column.
+#
+# Keep this in sync with kanban_db.VALID_STATUSES.  In particular,
+# ``scheduled`` is a first-class waiting column used for time-based follow-ups;
+# if it is omitted here, the board-level fallback below mis-buckets scheduled
+# tasks into ``todo`` and makes the dashboard look like the Scheduled column
+# disappeared.
 BOARD_COLUMNS: list[str] = [
-    "triage", "todo", "ready", "running", "blocked", "done",
+    "triage", "todo", "scheduled", "ready", "running", "blocked", "review", "done",
 ]
 
 
@@ -347,6 +354,12 @@ def get_board(
     tenant: Optional[str] = Query(None, description="Filter to a single tenant"),
     include_archived: bool = Query(False),
     board: Optional[str] = Query(None, description="Kanban board slug (omit for current)"),
+    workflow_template_id: Optional[str] = Query(
+        None, description="Restrict to tasks using this workflow template id",
+    ),
+    current_step_key: Optional[str] = Query(
+        None, description="Restrict to tasks at this workflow step key",
+    ),
 ):
     """Return the full board grouped by status column.
 
@@ -361,7 +374,11 @@ def get_board(
     conn = _conn(board=board)
     try:
         tasks = kanban_db.list_tasks(
-            conn, tenant=tenant, include_archived=include_archived
+            conn,
+            tenant=tenant,
+            include_archived=include_archived,
+            workflow_template_id=workflow_template_id,
+            current_step_key=current_step_key,
         )
         # Pre-fetch link counts per task (cheap: one query).
         link_counts: dict[str, dict[str, int]] = {}
@@ -472,10 +489,29 @@ def get_board(
 # ---------------------------------------------------------------------------
 
 @router.get("/tasks/{task_id}")
-def get_task(task_id: str, board: Optional[str] = Query(None)):
+def get_task(
+    task_id: str,
+    board: Optional[str] = Query(None),
+    run_state_type: Optional[str] = Query(
+        None, description="With run_state_name: filter runs by column 'status' or 'outcome'",
+    ),
+    run_state_name: Optional[str] = Query(
+        None, description="With run_state_type: exact value for that run column",
+    ),
+):
     board = _resolve_board(board)
     conn = _conn(board=board)
     try:
+        if (run_state_type is None) ^ (run_state_name is None):
+            raise HTTPException(
+                status_code=400,
+                detail="run_state_type and run_state_name must be passed together or omitted",
+            )
+        if run_state_type is not None and run_state_type not in ("status", "outcome"):
+            raise HTTPException(
+                status_code=400,
+                detail="run_state_type must be 'status' or 'outcome'",
+            )
         task = kanban_db.get_task(conn, task_id)
         if task is None:
             raise HTTPException(status_code=404, detail=f"task {task_id} not found")
@@ -496,7 +532,15 @@ def get_task(task_id: str, board: Optional[str] = Query(None)):
             "comments": [_comment_dict(c) for c in kanban_db.list_comments(conn, task_id)],
             "events": [_event_dict(e) for e in kanban_db.list_events(conn, task_id)],
             "links": _links_for(conn, task_id),
-            "runs": [_run_dict(r) for r in kanban_db.list_runs(conn, task_id)],
+            "runs": [
+                _run_dict(r)
+                for r in kanban_db.list_runs(
+                    conn,
+                    task_id,
+                    state_type=run_state_type,
+                    state_name=run_state_name,
+                )
+            ],
         }
     finally:
         conn.close()
@@ -617,10 +661,12 @@ def update_task(task_id: str, payload: UpdateTaskBody, board: Optional[str] = Qu
                 )
             elif s == "blocked":
                 ok = kanban_db.block_task(conn, task_id, reason=payload.block_reason)
+            elif s == "scheduled":
+                ok = kanban_db.schedule_task(conn, task_id, reason=payload.block_reason)
             elif s == "ready":
-                # Re-open a blocked task, or just an explicit status set.
+                # Re-open a blocked/scheduled task, or just an explicit status set.
                 current = kanban_db.get_task(conn, task_id)
-                if current and current.status == "blocked":
+                if current and current.status in ("blocked", "scheduled"):
                     ok = kanban_db.unblock_task(conn, task_id)
                 else:
                     # Direct status write for drag-drop (todo -> ready etc).
@@ -632,11 +678,28 @@ def update_task(task_id: str, payload: UpdateTaskBody, board: Optional[str] = Qu
                     status_code=400,
                     detail="Cannot set status to 'running' directly; use the dispatcher/claim path",
                 )
-            elif s in {"todo", "triage"}:
+            elif s in ("todo", "triage", "scheduled"):
                 ok = _set_status_direct(conn, task_id, s)
             else:
                 raise HTTPException(status_code=400, detail=f"unknown status: {s}")
             if not ok:
+                # For ``ready``, name the blocking parent(s) so the dashboard
+                # can render an actionable toast instead of a silent no-op.
+                # See #26744.
+                if s == "ready":
+                    blockers = _parents_blocking_ready(conn, task_id)
+                    if blockers:
+                        names = ", ".join(
+                            f"{p['title']!r} ({p['id']}, status={p['status']})"
+                            for p in blockers
+                        )
+                        raise HTTPException(
+                            status_code=409,
+                            detail=(
+                                f"Cannot move to 'ready': blocked by parent(s) "
+                                f"not done — {names}"
+                            ),
+                        )
                 raise HTTPException(
                     status_code=409,
                     detail=f"status transition to {s!r} not valid from current state",
@@ -684,6 +747,46 @@ def update_task(task_id: str, payload: UpdateTaskBody, board: Optional[str] = Qu
         conn.close()
 
 
+# ---------------------------------------------------------------------------
+# DELETE /tasks/:id
+# ---------------------------------------------------------------------------
+
+@router.delete("/tasks/{task_id}")
+def delete_task(task_id: str, board: Optional[str] = Query(None)):
+    board = _resolve_board(board)
+    conn = _conn(board=board)
+    try:
+        ok = kanban_db.delete_task(conn, task_id)
+        if not ok:
+            raise HTTPException(status_code=404, detail=f"task {task_id} not found")
+        return {"deleted": True, "task_id": task_id}
+    finally:
+        conn.close()
+
+
+def _parents_blocking_ready(
+    conn: sqlite3.Connection, task_id: str,
+) -> list:
+    """Return parent rows (``id``, ``title``, ``status``) that aren't ``done``
+    and therefore prevent ``task_id`` from being promoted to ``ready``.
+
+    Used to enrich the 409 response from :func:`update_task` so the
+    dashboard can show an actionable toast (#26744) instead of a silent
+    no-op.  Returns ``[]`` when nothing blocks the transition (e.g. no
+    parents, or all parents already done).
+    """
+    rows = conn.execute(
+        "SELECT t.id, t.title, t.status FROM tasks t "
+        "JOIN task_links l ON l.parent_id = t.id "
+        "WHERE l.child_id = ? AND t.status != 'done'",
+        (task_id,),
+    ).fetchall()
+    return [
+        {"id": r["id"], "title": r["title"], "status": r["status"]}
+        for r in rows
+    ]
+
+
 def _set_status_direct(
     conn: sqlite3.Connection, task_id: str, new_status: str,
 ) -> bool:
@@ -722,6 +825,10 @@ def _set_status_direct(
                 return False
 
         was_running = prev["status"] == "running"
+        reopening_satisfied_parent = (
+            prev["status"] in {"done", "archived"}
+            and new_status not in {"done", "archived"}
+        )
 
         cur = conn.execute(
             "UPDATE tasks SET status = ?, "
@@ -745,6 +852,37 @@ def _set_status_direct(
             "VALUES (?, ?, 'status', ?, ?)",
             (task_id, run_id, json.dumps({"status": new_status}), int(time.time())),
         )
+        if reopening_satisfied_parent:
+            # A parent leaving done/archived invalidates any direct child that
+            # was sitting in ready solely because that parent used to satisfy
+            # the dependency gate. Demote those children immediately so the
+            # dashboard does not keep advertising stale-ready work.
+            for row in conn.execute(
+                "SELECT child_id FROM task_links WHERE parent_id = ? ORDER BY child_id",
+                (task_id,),
+            ).fetchall():
+                child_id = row["child_id"]
+                demoted = conn.execute(
+                    "UPDATE tasks SET status = 'todo' "
+                    "WHERE id = ? AND status = 'ready'",
+                    (child_id,),
+                )
+                if demoted.rowcount == 1:
+                    conn.execute(
+                        "INSERT INTO task_events (task_id, kind, payload, created_at) "
+                        "VALUES (?, 'status', ?, ?)",
+                        (
+                            child_id,
+                            json.dumps(
+                                {
+                                    "status": "todo",
+                                    "reason": "parent_reopened",
+                                    "parent": task_id,
+                                }
+                            ),
+                            int(time.time()),
+                        ),
+                    )
     # If we re-opened something, children may have gone stale.
     if new_status in {"done", "ready"}:
         kanban_db.recompute_ready(conn)
@@ -868,11 +1006,23 @@ def bulk_update(payload: BulkTaskBody, board: Optional[str] = Query(None)):
                         ok = kanban_db.block_task(conn, tid)
                     elif s == "ready":
                         cur = kanban_db.get_task(conn, tid)
-                        if cur and cur.status == "blocked":
+                        if cur and cur.status in ("blocked", "scheduled"):
                             ok = kanban_db.unblock_task(conn, tid)
                         else:
                             ok = _set_status_direct(conn, tid, "ready")
-                    elif s in {"todo", "running", "triage"}:
+                    elif s == "running":
+                        entry.update(
+                            ok=False,
+                            error=(
+                                "Cannot set status to 'running' directly; "
+                                "use the dispatcher/claim path"
+                            ),
+                        )
+                        results.append(entry)
+                        continue
+                    elif s == "scheduled":
+                        ok = kanban_db.schedule_task(conn, tid)
+                    elif s in {"todo", "triage"}:
                         ok = _set_status_direct(conn, tid, s)
                     else:
                         entry.update(ok=False, error=f"unknown status {s!r}")
@@ -950,7 +1100,7 @@ def list_diagnostics(
         if severity:
             filtered: dict[str, list[dict]] = {}
             for tid, dl in diags_by_task.items():
-                keep = [d for d in dl if d.get("severity") == severity]
+                keep = [d for d in dl if kd.severity_at_or_above(d.get("severity"), severity)]
                 if keep:
                     filtered[tid] = keep
             diags_by_task = filtered
@@ -998,6 +1148,168 @@ def list_diagnostics(
         conn.close()
 
 
+
+# ---------------------------------------------------------------------------
+# Worker visibility — cross-task active-worker list and per-run inspection
+# ---------------------------------------------------------------------------
+
+try:
+    import psutil as _psutil
+except ImportError:
+    _psutil = None  # type: ignore[assignment]
+
+
+@router.get("/workers/active")
+def list_active_workers(
+    board: Optional[str] = Query(None, description="Kanban board slug (omit for current)"),
+):
+    """Return every currently-running worker on the board.
+
+    A worker is a ``task_runs`` row whose ``ended_at`` is NULL and whose
+    ``worker_pid`` is non-NULL, belonging to a task with ``status='running'``.
+
+    Returns ``{workers: [...], count: N, checked_at: <epoch>}``.  Each
+    worker entry carries enough context for the dashboard to link back to
+    its task without a second round-trip.
+    """
+    board = _resolve_board(board)
+    conn = _conn(board=board)
+    try:
+        rows = conn.execute(
+            """
+            SELECT
+                r.id          AS run_id,
+                r.task_id,
+                t.title       AS task_title,
+                t.status      AS task_status,
+                t.assignee    AS task_assignee,
+                r.profile,
+                r.worker_pid,
+                r.started_at,
+                r.claim_lock,
+                r.claim_expires,
+                r.last_heartbeat_at,
+                r.max_runtime_seconds
+            FROM task_runs r
+            JOIN tasks t ON t.id = r.task_id
+            WHERE r.ended_at IS NULL
+              AND r.worker_pid IS NOT NULL
+              AND t.status = 'running'
+            ORDER BY r.started_at ASC
+            """,
+        ).fetchall()
+        workers = [
+            {
+                "run_id": row["run_id"],
+                "task_id": row["task_id"],
+                "task_title": row["task_title"],
+                "task_status": row["task_status"],
+                "task_assignee": row["task_assignee"],
+                "profile": row["profile"],
+                "worker_pid": row["worker_pid"],
+                "started_at": row["started_at"],
+                "claim_lock": row["claim_lock"],
+                "claim_expires": row["claim_expires"],
+                "last_heartbeat_at": row["last_heartbeat_at"],
+                "max_runtime_seconds": row["max_runtime_seconds"],
+            }
+            for row in rows
+        ]
+        return {"workers": workers, "count": len(workers), "checked_at": int(time.time())}
+    finally:
+        conn.close()
+
+
+@router.get("/runs/{run_id}")
+def get_run_endpoint(
+    run_id: int,
+    board: Optional[str] = Query(None, description="Kanban board slug (omit for current)"),
+):
+    """Direct lookup of a ``task_runs`` row by its integer id.
+
+    Returns ``{run: {...}}`` using the same serialisation as the
+    per-task run history embedded in ``GET /tasks/{task_id}``.
+    404 when no such run exists.
+    """
+    board = _resolve_board(board)
+    conn = _conn(board=board)
+    try:
+        r = kanban_db.get_run(conn, run_id)
+        if r is None:
+            raise HTTPException(status_code=404, detail=f"run {run_id} not found")
+        return {"run": _run_dict(r)}
+    finally:
+        conn.close()
+
+
+@router.get("/runs/{run_id}/inspect")
+def inspect_run_endpoint(
+    run_id: int,
+    board: Optional[str] = Query(None, description="Kanban board slug (omit for current)"),
+):
+    """Live PID stats for a run's worker process via psutil.
+
+    If the run has already ended, or has no recorded ``worker_pid``,
+    returns ``{alive: false}`` with a human-readable ``reason``.
+
+    When the process is live, returns CPU, memory, thread count, fd count,
+    status, create_time, and cmdline.  ``access_denied`` is set when the
+    OS refuses inspection rather than raising a 500.
+
+    psutil availability: if psutil is not installed the endpoint still
+    works but ``alive`` is always returned as ``false`` with
+    ``reason="psutil not available"``.
+    """
+    board = _resolve_board(board)
+    conn = _conn(board=board)
+    try:
+        r = kanban_db.get_run(conn, run_id)
+        if r is None:
+            raise HTTPException(status_code=404, detail=f"run {run_id} not found")
+    finally:
+        conn.close()
+
+    if r.ended_at is not None:
+        return {"run_id": run_id, "alive": False, "reason": "run already ended"}
+    if r.worker_pid is None:
+        return {"run_id": run_id, "alive": False, "reason": "no worker_pid recorded"}
+
+    pid = r.worker_pid
+
+    if _psutil is None:
+        return {"run_id": run_id, "alive": False, "pid": pid, "reason": "psutil not available"}
+
+    try:
+        proc = _psutil.Process(pid)
+        info = proc.as_dict(attrs=[
+            "cpu_percent", "memory_info", "num_threads",
+            "status", "create_time", "cmdline",
+        ])
+        # num_fds is POSIX-only; skip gracefully on Windows.
+        try:
+            num_fds = proc.num_fds()
+        except AttributeError:
+            num_fds = None
+        mem = info.get("memory_info")
+        return {
+            "run_id": run_id,
+            "alive": True,
+            "pid": pid,
+            "cpu_percent": info.get("cpu_percent"),
+            "memory_rss_bytes": mem.rss if mem else None,
+            "memory_vms_bytes": mem.vms if mem else None,
+            "num_threads": info.get("num_threads"),
+            "num_fds": num_fds,
+            "status": info.get("status"),
+            "create_time": info.get("create_time"),
+            "cmdline": info.get("cmdline"),
+        }
+    except _psutil.NoSuchProcess:
+        return {"run_id": run_id, "alive": False, "pid": pid, "reason": "process not found"}
+    except _psutil.AccessDenied:
+        return {"run_id": run_id, "alive": True, "pid": pid, "error": "access denied"}
+
+
 # ---------------------------------------------------------------------------
 # Recovery actions — reclaim a running claim, reassign to a new profile
 # ---------------------------------------------------------------------------
@@ -1207,6 +1519,15 @@ def _configured_home_channels() -> list[dict]:
     return result
 
 
+def _active_profile_name() -> str:
+    """Return the current Hermes profile name for notify-sub ownership."""
+    try:
+        from hermes_cli.profiles import get_active_profile_name
+        return get_active_profile_name() or "default"
+    except Exception:
+        return "default"
+
+
 def _home_sub_matches(sub: dict, home: dict) -> bool:
     """True if a notify_subs row corresponds to the given home channel."""
     return (
@@ -1278,6 +1599,7 @@ def subscribe_home(task_id: str, platform: str, board: Optional[str] = Query(Non
             platform=platform,
             chat_id=home["chat_id"],
             thread_id=home["thread_id"] or None,
+            notifier_profile=_active_profile_name(),
         )
         return {"ok": True, "task_id": task_id, "home_channel": home}
     finally:
@@ -1701,6 +2023,7 @@ class OrchestrationSettingsBody(BaseModel):
     orchestrator_profile: Optional[str] = None
     default_assignee: Optional[str] = None
     auto_decompose: Optional[bool] = None
+    auto_promote_children: Optional[bool] = None
 
 
 @router.get("/orchestration")
@@ -1716,6 +2039,7 @@ def get_orchestration_settings():
     explicit_orch = (kanban_cfg.get("orchestrator_profile") or "").strip()
     explicit_default = (kanban_cfg.get("default_assignee") or "").strip()
     auto_decompose = bool(kanban_cfg.get("auto_decompose", True))
+    auto_promote_children = bool(kanban_cfg.get("auto_promote_children", True))
 
     # Resolve fallbacks the same way the decomposer does.
     resolved_orch = explicit_orch
@@ -1738,6 +2062,7 @@ def get_orchestration_settings():
         "orchestrator_profile": explicit_orch,
         "default_assignee": explicit_default,
         "auto_decompose": auto_decompose,
+        "auto_promote_children": auto_promote_children,
         "resolved_orchestrator_profile": resolved_orch,
         "resolved_default_assignee": resolved_default,
         "active_profile": active_default,
@@ -1803,6 +2128,9 @@ def set_orchestration_settings(payload: OrchestrationSettingsBody):
     if payload.auto_decompose is not None:
         kanban_section["auto_decompose"] = bool(payload.auto_decompose)
 
+    if payload.auto_promote_children is not None:
+        kanban_section["auto_promote_children"] = bool(payload.auto_promote_children)
+
     try:
         save_config(cfg)
     except Exception as exc:

plugins/model-providers/azure-foundry/__init__.py

@@ -1,4 +1,4 @@
-"""Azure AI Foundry provider profile.
+"""Microsoft Foundry provider profile.
 
 Azure Foundry exposes an OpenAI-compatible endpoint; users supply their own
 base URL at setup since endpoints are per-resource.
@@ -11,7 +11,7 @@ azure_foundry = ProviderProfile(
     name="azure-foundry",
     aliases=("azure", "azure-ai-foundry", "azure-ai"),
     display_name="Azure Foundry",
-    description="Azure AI Foundry — OpenAI-compatible endpoint (user-supplied base URL)",
+    description="Microsoft Foundry - OpenAI-compatible endpoint (user-supplied base URL)",
     signup_url="https://ai.azure.com/",
     env_vars=("AZURE_FOUNDRY_API_KEY", "AZURE_FOUNDRY_BASE_URL"),
     base_url="",  # per-resource; user provides at setup

plugins/model-providers/azure-foundry/plugin.yaml

@@ -1,5 +1,5 @@
 name: azure-foundry-provider
 kind: model-provider
 version: 1.0.0
-description: Azure AI Foundry
+description: Microsoft Foundry
 author: Nous Research

plugins/platforms/google_chat/oauth.py

@@ -586,7 +586,8 @@ def revoke(email: Optional[str] = None) -> None:
                 f"https://oauth2.googleapis.com/revoke?token={creds.token}",
                 method="POST",
                 headers={"Content-Type": "application/x-www-form-urlencoded"},
-            )
+            ),
+            timeout=15,
         )
         print("Token revoked with Google.")
     except Exception as exc:

plugins/web/xai/__init__.py

@@ -0,0 +1,14 @@
+"""xAI web search plugin — bundled, auto-loaded.
+
+Mirrors the ``plugins/web/brave_free/`` layout: ``provider.py`` holds the
+provider class, ``__init__.py::register(ctx)`` registers an instance.
+"""
+
+from __future__ import annotations
+
+from plugins.web.xai.provider import XAIWebSearchProvider
+
+
+def register(ctx) -> None:
+    """Register the xAI Web Search provider with the plugin context."""
+    ctx.register_web_search_provider(XAIWebSearchProvider())

plugins/web/xai/plugin.yaml

@@ -0,0 +1,7 @@
+name: web-xai
+version: 1.0.0
+description: "xAI Web Search — search the web via Grok's agentic web_search tool (Responses API). Requires xAI Grok OAuth (via `hermes auth`) or XAI_API_KEY (https://x.ai)."
+author: NousResearch
+kind: backend
+provides_web_providers:
+  - xai

plugins/web/xai/provider.py

@@ -0,0 +1,560 @@
+"""xAI Web Search — plugin form.
+
+Routes ``web_search`` tool calls through xAI's agentic Web Search tool
+(server-side ``web_search`` on the Responses API). Grok runs the actual
+searching and page-browsing server-side; we ask it to return the top
+results as structured JSON so we can hand back the same
+``{title, url, description, position}`` rows every other Hermes web
+provider produces.
+
+Reference: https://docs.x.ai/developers/tools/web-search
+
+Config keys this provider responds to::
+
+    web:
+      search_backend: "xai"           # explicit per-capability
+      backend: "xai"                  # shared fallback
+
+Optional knobs (under ``web.xai`` in ``config.yaml``)::
+
+    web:
+      xai:
+        model: "grok-4.3"             # reasoning model required by web_search
+        allowed_domains: ["x.ai"]     # max 5 — mutually exclusive with excluded_domains
+        excluded_domains: ["bad.com"] # max 5 — mutually exclusive with allowed_domains
+        timeout: 90                   # seconds (default 90)
+
+Auth: reuses :func:`tools.xai_http.resolve_xai_http_credentials`, which
+prefers Hermes-managed xAI Grok OAuth (via ``hermes auth``) and falls back
+to ``XAI_API_KEY`` (resolved through ``~/.hermes/.env``, then
+``os.environ``).
+"""
+
+from __future__ import annotations
+
+import json
+import logging
+import re
+from typing import Any, Dict, List, Optional
+
+from agent.web_search_provider import WebSearchProvider
+from tools.xai_http import (
+    has_xai_credentials,
+    hermes_xai_user_agent,
+    resolve_xai_http_credentials,
+)
+
+logger = logging.getLogger(__name__)
+
+DEFAULT_MODEL = "grok-4.3"
+DEFAULT_TIMEOUT = 90
+_MAX_DOMAIN_FILTERS = 5  # xAI hard cap on allowed_domains / excluded_domains
+
+# Match the JSON object Grok is asked to emit. Tolerates leading/trailing
+# prose since reasoning models occasionally narrate before the JSON block
+# even when explicitly asked not to.
+_JSON_BLOCK_RE = re.compile(r"\{[\s\S]*\}", re.MULTILINE)
+
+
+# ---------------------------------------------------------------------------
+# Config
+# ---------------------------------------------------------------------------
+
+
+def _load_xai_web_config() -> Dict[str, Any]:
+    """Read ``web.xai`` from config.yaml (returns {} on miss)."""
+    try:
+        from hermes_cli.config import load_config
+
+        cfg = load_config()
+        web_section = cfg.get("web") if isinstance(cfg, dict) else None
+        xai_section = web_section.get("xai") if isinstance(web_section, dict) else None
+        return xai_section if isinstance(xai_section, dict) else {}
+    except Exception as exc:  # noqa: BLE001
+        logger.debug("Could not load web.xai config: %s", exc)
+        return {}
+
+
+def _coerce_domain_list(value: Any) -> List[str]:
+    """Coerce a config value to a clean list of <=5 domain strings."""
+    if not isinstance(value, list):
+        return []
+    cleaned: List[str] = []
+    for item in value:
+        if isinstance(item, str) and item.strip():
+            cleaned.append(item.strip())
+        if len(cleaned) >= _MAX_DOMAIN_FILTERS:
+            break
+    return cleaned
+
+
+# ---------------------------------------------------------------------------
+# Provider
+# ---------------------------------------------------------------------------
+
+
+class XAIWebSearchProvider(WebSearchProvider):
+    """Search-only provider backed by xAI's agentic Web Search tool.
+
+    Sends a structured prompt to Grok with ``tools=[{"type": "web_search"}]``
+    enabled and asks it to return the top *limit* results as JSON. Falls
+    back to the Responses API ``citations`` list if Grok ignores the JSON
+    schema instruction (rare for grok-4.3 but cheap insurance).
+
+    No extract capability — pair with Firecrawl / Tavily / Exa for
+    ``web_extract`` if you need page content.
+
+    Trust model
+    -----------
+    Unlike index-backed providers (Brave / Tavily / Exa) which return
+    verbatim search-engine results, this backend is an LLM in a trench
+    coat: Grok decides which URLs to surface, generates the titles and
+    descriptions itself, and is influenced by the *content of the query*.
+    A maliciously crafted query (e.g. injected via untrusted upstream
+    input the agent picked up) can in principle steer Grok into emitting
+    attacker-chosen URLs. Callers that pipe untrusted text directly into
+    ``web_search`` should treat returned URLs the same way they would
+    treat any model-generated link — validate before fetching.
+    """
+
+    @property
+    def name(self) -> str:
+        return "xai"
+
+    @property
+    def display_name(self) -> str:
+        return "xAI Web Search (Grok)"
+
+    def is_available(self) -> bool:
+        """Cheap availability probe — env var OR auth-store has OAuth tokens.
+
+        Delegates to :func:`tools.xai_http.has_xai_credentials`, which is
+        deliberately *not* the same as :func:`resolve_xai_http_credentials`:
+        it never triggers OAuth token refresh or acquires the auth-store
+        lock. The ABC contract requires this method to be safe to call on
+        every ``hermes tools`` repaint and at tool-registration time.
+        Token freshness / refresh is handled inside :meth:`search`.
+        """
+        return has_xai_credentials()
+
+    def supports_search(self) -> bool:
+        return True
+
+    def supports_extract(self) -> bool:
+        return False
+
+    def supports_crawl(self) -> bool:
+        return False
+
+    # -- Search -----------------------------------------------------------
+
+    def search(self, query: str, limit: int = 5) -> Dict[str, Any]:
+        """Execute a Grok-backed web search.
+
+        Returns ``{"success": True, "data": {"web": [{title, url, description, position}, ...]}}``
+        on success, ``{"success": False, "error": str}`` on failure.
+        """
+        try:
+            from tools.interrupt import is_interrupted
+
+            if is_interrupted():
+                return {"success": False, "error": "Interrupted"}
+        except Exception:  # noqa: BLE001 — interrupt module is best-effort
+            pass
+
+        creds = resolve_xai_http_credentials()
+        api_key = str(creds.get("api_key") or "").strip()
+        base_url = str(creds.get("base_url") or "https://api.x.ai/v1").strip().rstrip("/")
+        if not api_key:
+            return {
+                "success": False,
+                "error": (
+                    "No xAI credentials found. Run `hermes auth` to sign in with "
+                    "xAI Grok OAuth, or set XAI_API_KEY."
+                ),
+            }
+
+        # Clamp limit to the same range the caller (web_search_tool) accepts,
+        # so we don't silently downgrade explicit limits. Grok happily
+        # produces longer lists; cost scales linearly with the requested
+        # count via reasoning tokens, but that's the caller's call to make.
+        try:
+            limit = int(limit)
+        except (TypeError, ValueError):
+            limit = 5
+        limit = max(1, min(limit, 100))
+
+        cfg = _load_xai_web_config()
+        model = cfg.get("model") if isinstance(cfg.get("model"), str) else DEFAULT_MODEL
+        model = model.strip() or DEFAULT_MODEL
+
+        try:
+            timeout = float(cfg.get("timeout", DEFAULT_TIMEOUT))
+        except (TypeError, ValueError):
+            timeout = DEFAULT_TIMEOUT
+
+        allowed = _coerce_domain_list(cfg.get("allowed_domains"))
+        excluded = _coerce_domain_list(cfg.get("excluded_domains"))
+        if allowed and excluded:
+            # xAI explicitly rejects this combo — surface a clear error
+            # rather than a 400 from the API.
+            return {
+                "success": False,
+                "error": (
+                    "web.xai.allowed_domains and web.xai.excluded_domains "
+                    "cannot both be set (xAI restriction)."
+                ),
+            }
+
+        web_search_tool: Dict[str, Any] = {"type": "web_search"}
+        if allowed:
+            web_search_tool["filters"] = {"allowed_domains": allowed}
+        elif excluded:
+            web_search_tool["filters"] = {"excluded_domains": excluded}
+
+        prompt = self._build_prompt(query, limit)
+
+        payload: Dict[str, Any] = {
+            "model": model,
+            "input": [{"role": "user", "content": prompt}],
+            "tools": [web_search_tool],
+            # Drop inline citation markdown — we want the JSON block clean,
+            # and we read URLs from annotations / citations separately.
+            "include": ["no_inline_citations"],
+        }
+
+        headers = {
+            "Authorization": f"Bearer {api_key}",
+            "Content-Type": "application/json",
+            "User-Agent": hermes_xai_user_agent(),
+        }
+
+        try:
+            import httpx
+        except ImportError:
+            return {
+                "success": False,
+                "error": "httpx is not installed (required for xAI web search)",
+            }
+
+        logger.info(
+            "xAI web search via %s: '%s' (limit=%d, model=%s)",
+            base_url, query, limit, model,
+        )
+
+        # Two-attempt loop: if the first call returns 401 and our creds came
+        # from the OAuth path, force-refresh the token once and retry. This
+        # closes two gaps the proactive resolver check doesn't cover:
+        # (1) opaque (non-JWT) access tokens — `_xai_access_token_is_expiring`
+        #     can't decode them and returns False, so refresh never fires
+        #     until the server hands us a 401.
+        # (2) mid-window revocation — admin revoke, refresh-token rotation,
+        #     or clock skew can produce 401s on a token whose JWT `exp` claim
+        #     is still in the future.
+        # Env-var (`XAI_API_KEY`) credentials skip the retry entirely — we
+        # can't refresh those and an immediate retry would just burn quota.
+        is_oauth_path = (creds.get("provider") == "xai-oauth")
+        resp = None
+        for attempt in range(2):
+            try:
+                resp = httpx.post(
+                    f"{base_url}/responses",
+                    headers=headers,
+                    json=payload,
+                    timeout=timeout,
+                )
+                resp.raise_for_status()
+                break
+            except httpx.HTTPStatusError as exc:
+                status = exc.response.status_code if exc.response is not None else 0
+                if status == 401 and attempt == 0 and is_oauth_path:
+                    logger.info(
+                        "xAI web search got 401 on first attempt; forcing OAuth "
+                        "refresh and retrying once.",
+                    )
+                    try:
+                        refreshed = resolve_xai_http_credentials(force_refresh=True)
+                        refreshed_key = str(refreshed.get("api_key") or "").strip()
+                        if refreshed_key and refreshed_key != api_key:
+                            api_key = refreshed_key
+                            headers["Authorization"] = f"Bearer {api_key}"
+                            continue
+                        # Refresh returned the same (or empty) token — no point
+                        # in retrying. Fall through to the error return below.
+                    except Exception as refresh_exc:  # noqa: BLE001
+                        logger.warning(
+                            "xAI web search OAuth refresh after 401 failed: %s",
+                            refresh_exc,
+                        )
+                body = ""
+                try:
+                    body = exc.response.text[:300] if exc.response is not None else ""
+                except Exception:
+                    body = ""
+                logger.warning("xAI web search HTTP %d: %s", status, body)
+                return {
+                    "success": False,
+                    "error": f"xAI web search returned HTTP {status}: {body}".rstrip(),
+                }
+            except httpx.RequestError as exc:
+                logger.warning("xAI web search request error: %s", exc)
+                return {"success": False, "error": f"Could not reach xAI: {exc}"}
+
+        if resp is None:
+            # Defensive — both attempts somehow exited the loop without resp.
+            return {"success": False, "error": "xAI web search produced no response"}
+
+        try:
+            data = resp.json()
+        except Exception as exc:  # noqa: BLE001
+            logger.warning("xAI web search bad JSON: %s", exc)
+            return {
+                "success": False,
+                "error": "Could not parse xAI Responses API reply as JSON",
+            }
+
+        # xAI's Responses surface sometimes returns HTTP 200 with an error
+        # envelope (model overloaded, content-policy refusal, etc.). Without
+        # this check, ``_extract_results`` would silently produce an empty
+        # list and we'd report success-with-no-rows — masking a real failure
+        # the agent should see and decide whether to retry.
+        api_error = data.get("error") if isinstance(data, dict) else None
+        if isinstance(api_error, dict):
+            err_msg = (
+                api_error.get("message")
+                or api_error.get("code")
+                or "unknown error"
+            )
+            logger.warning("xAI web search returned error envelope: %s", err_msg)
+            return {"success": False, "error": f"xAI returned an error: {err_msg}"}
+
+        web_results = self._extract_results(data, limit=limit)
+        if not web_results:
+            # Successful call, just no usable rows — return success with an
+            # empty list so the model can decide whether to retry. Matches
+            # what brave-free / exa do when the upstream API returns 0 hits.
+            return {"success": True, "data": {"web": []}}
+
+        return {"success": True, "data": {"web": web_results}}
+
+    # -- Prompt + parsing -------------------------------------------------
+
+    @staticmethod
+    def _build_prompt(query: str, limit: int) -> str:
+        """Compose the prompt that asks Grok to act as a search engine.
+
+        We deliberately ask for a JSON object (not bare array) so we can
+        match it cheaply with ``_JSON_BLOCK_RE``; we explicitly forbid
+        prose, markdown fences, and inline-citation links to keep the
+        payload parseable.
+        """
+        return (
+            "Use the web_search tool to find current information for the query below, "
+            "then respond with ONLY a single JSON object — no prose, no markdown "
+            "fences, no inline citation links — matching this exact schema:\n\n"
+            '{"results": [{"title": "string", "url": "string", '
+            '"description": "1-2 sentence summary"}]}\n\n'
+            f'Return at most {limit} results, ordered by relevance, with absolute '
+            "https:// URLs. If no usable results exist, return "
+            '{"results": []}.\n\n'
+            f"Query: {query}"
+        )
+
+    @classmethod
+    def _extract_results(
+        cls,
+        response_data: Dict[str, Any],
+        *,
+        limit: int,
+    ) -> List[Dict[str, Any]]:
+        """Pull a ``[{title, url, description, position}, ...]`` list out of a
+        Responses-API reply.
+
+        Strategy:
+
+        1. Walk ``output[*].content[*].text`` for ``output_text`` blocks and
+           try to parse the first JSON object that has a ``results`` list.
+        2. If the JSON path fails, fall back to the message annotations
+           (``url_citation`` entries) — every annotation carries a URL and
+           a ``title`` (citation number); we pair those URLs with surrounding
+           text from the message body as a best-effort description.
+        """
+        text_blocks, annotations = cls._collect_output_text(response_data)
+
+        # Primary path: parse the JSON object Grok was asked for.
+        for block in text_blocks:
+            parsed = cls._try_parse_json_results(block, limit=limit)
+            if parsed:
+                return parsed
+
+        # Secondary path: derive results from message annotations + raw text.
+        # Only short-circuit when annotations actually yielded usable rows;
+        # otherwise fall through to the citations list. (xAI currently only
+        # emits ``url_citation`` annotations, but future annotation types
+        # would silently produce an empty result set if we returned here
+        # unconditionally — masking real data in ``citations``.)
+        if annotations:
+            joined_text = "\n".join(text_blocks)
+            annotation_results = cls._results_from_annotations(
+                annotations, joined_text, limit=limit,
+            )
+            if annotation_results:
+                return annotation_results
+
+        # Last-ditch: raw citations list (no titles or descriptions).
+        citations = response_data.get("citations") or []
+        if isinstance(citations, list):
+            return [
+                {
+                    "title": "",
+                    "url": str(u),
+                    "description": "",
+                    "position": i + 1,
+                }
+                for i, u in enumerate(citations[:limit])
+                if isinstance(u, str) and u.strip()
+            ]
+
+        return []
+
+    @staticmethod
+    def _collect_output_text(
+        response_data: Dict[str, Any],
+    ) -> tuple[List[str], List[Dict[str, Any]]]:
+        """Return (text_blocks, annotations) extracted from ``response.output``."""
+        text_blocks: List[str] = []
+        annotations: List[Dict[str, Any]] = []
+        output = response_data.get("output")
+        if not isinstance(output, list):
+            return text_blocks, annotations
+
+        for item in output:
+            if not isinstance(item, dict) or item.get("type") != "message":
+                continue
+            content = item.get("content")
+            if not isinstance(content, list):
+                continue
+            for chunk in content:
+                if not isinstance(chunk, dict) or chunk.get("type") != "output_text":
+                    continue
+                text = chunk.get("text")
+                if isinstance(text, str) and text.strip():
+                    text_blocks.append(text)
+                chunk_annotations = chunk.get("annotations")
+                if isinstance(chunk_annotations, list):
+                    for ann in chunk_annotations:
+                        if isinstance(ann, dict):
+                            annotations.append(ann)
+        return text_blocks, annotations
+
+    @staticmethod
+    def _try_parse_json_results(
+        text: str,
+        *,
+        limit: int,
+    ) -> Optional[List[Dict[str, Any]]]:
+        """Parse a JSON object with a ``results`` array out of ``text``.
+
+        Returns the normalized result list on success, ``None`` when the
+        block has no valid JSON object or no ``results`` key. Tolerates
+        leading/trailing prose because reasoning models sometimes prefix a
+        short narration even when told not to.
+        """
+        # Try the whole string first — cheapest path when Grok obeys.
+        candidates = [text]
+        match = _JSON_BLOCK_RE.search(text)
+        if match and match.group(0) != text:
+            candidates.append(match.group(0))
+
+        for candidate in candidates:
+            try:
+                parsed = json.loads(candidate)
+            except (json.JSONDecodeError, ValueError):
+                continue
+            if not isinstance(parsed, dict):
+                continue
+            results = parsed.get("results")
+            if not isinstance(results, list):
+                continue
+            normalized: List[Dict[str, Any]] = []
+            for row in results[:limit]:
+                if not isinstance(row, dict):
+                    continue
+                url = str(row.get("url", "")).strip()
+                if not url:
+                    continue
+                normalized.append(
+                    {
+                        "title": str(row.get("title", "")).strip(),
+                        "url": url,
+                        "description": str(row.get("description", "")).strip(),
+                        # Renumber from the kept results, not the raw input
+                        # index, so a dropped malformed row doesn't leave a
+                        # gap in the positions handed back to the agent.
+                        "position": len(normalized) + 1,
+                    }
+                )
+            if normalized:
+                return normalized
+        return None
+
+    @staticmethod
+    def _results_from_annotations(
+        annotations: List[Dict[str, Any]],
+        joined_text: str,
+        *,
+        limit: int,
+    ) -> List[Dict[str, Any]]:
+        """Best-effort fallback when JSON parsing fails.
+
+        Uses each ``url_citation`` annotation's ``url`` (the citation
+        title is just the integer label, so we don't surface it) and
+        slices ~200 characters of surrounding text as the description.
+        """
+        seen: set[str] = set()
+        results: List[Dict[str, Any]] = []
+        for ann in annotations:
+            if ann.get("type") != "url_citation":
+                continue
+            url = str(ann.get("url", "")).strip()
+            if not url or url in seen:
+                continue
+            seen.add(url)
+
+            description = ""
+            start = ann.get("start_index")
+            end = ann.get("end_index")
+            if isinstance(start, int) and isinstance(end, int) and 0 <= start < end <= len(joined_text):
+                window_start = max(0, start - 200)
+                description = joined_text[window_start:start].strip()
+                if len(description) > 200:
+                    description = description[-200:].strip()
+
+            results.append(
+                {
+                    "title": "",
+                    "url": url,
+                    "description": description,
+                    "position": len(results) + 1,
+                }
+            )
+            if len(results) >= limit:
+                break
+        return results
+
+    # -- Setup picker -----------------------------------------------------
+
+    def get_setup_schema(self) -> Dict[str, Any]:
+        # Auth resolution is delegated to the shared ``xai_grok`` post_setup
+        # hook (same one image_gen.xai and tts.xai use) so users see the
+        # familiar OAuth-or-API-key prompt for every xAI service.
+        return {
+            "name": "xAI Web Search (Grok)",
+            "badge": "paid",
+            "tag": (
+                "Agentic web search via Grok's web_search tool — uses xAI "
+                "Grok OAuth or XAI_API_KEY."
+            ),
+            "env_vars": [],
+            "post_setup": "xai_grok",
+        }