fix(kanban): only guard open pull requests

hermes_cli/config.py

@@ -1687,6 +1687,18 @@ def _ensure_hermes_home_managed(home: Path):
             "extra_body": {},
             "reasoning_effort": "",  # per-task thinking level: none|minimal|low|medium|high|xhigh|max|ultra (empty = provider default)
         },
+        # Batch planner — orders only selected Kanban cards that are likely
+        # to overlap in a repository or versioned knowledge base. Invoked by
+        # the dashboard's "Plan & take" bulk action; keep it cheap.
+        "kanban_batch_planner": {
+            "provider": "auto",
+            "model": "",
+            "base_url": "",
+            "api_key": "",
+            "timeout": 90,
+            "extra_body": {},
+            "reasoning_effort": "",
+        },
         # Profile describer — auto-generates a 1-2 sentence description
         # of what a profile is good at. Invoked by
         # ``hermes profile describe <name> --auto`` and the dashboard's
@@ -2786,6 +2798,12 @@ def _ensure_hermes_home_managed(home: Path):
         # assignee to any installed profile. When unset, falls back to the
         # default profile. A task never ends up with assignee=None.
         "default_assignee": "",
+        # Coarse maximum context size the Kanban decomposer should plan for
+        # per fresh worker. The estimate is deliberately approximate: semantic
+        # seams decide where work can split; this cap tells the decomposer how
+        # large the resulting pieces may be. Runtime handoff remains the
+        # safeguard when actual tool output exceeds the forecast.
+        "decomposer_context_budget_tokens": 150_000,
         # Per-profile concurrency cap (#21582). When set to a positive int,
         # no single profile can have more than N workers running at once,
         # even if the global max_in_progress / max_spawn caps would allow

hermes_cli/kanban_batch_take.py

@@ -0,0 +1,175 @@
+"""Plan and promote a selected Kanban batch without overlapping workers.
+
+The planner asks a low-cost auxiliary model only for ordering constraints
+between the chosen cards.  It never rewrites existing dependencies and uses
+``link_tasks`` for every proposed edge, so the database remains the source of
+truth for DAG validation.
+"""
+from __future__ import annotations
+
+import json
+import logging
+import re
+from dataclasses import dataclass
+from typing import Any, Optional
+
+from hermes_cli import kanban_db as kb
+
+logger = logging.getLogger(__name__)
+
+_SYSTEM_PROMPT = """You plan a selected batch of Kanban tasks. Serialize only
+CLEAR problems, not every touch. Add an ordering edge when two tasks would
+heavily rewrite the same code, configuration, or versioned knowledge base in
+different directions — that is the expensive merge-conflict case to prevent.
+Light overlap (each task adds a little in different places of the same file)
+merges trivially: keep those parallel. Judge by the volume and nature of the
+edits in the shared area: additive changes are peaceful, rewrites conflict.
+Do not invent dependencies for merely related tasks; preserve parallelism
+whenever possible.
+
+Return JSON only:
+{"edges":[{"before":"task-id","after":"task-id","reason":"short reason"}]}
+
+Rules: ids must come from the supplied tasks; before and after differ; use the
+fewest edges necessary; an empty edges list is valid."""
+_FENCE_RE = re.compile(r"^```(?:json)?\s*|\s*```$", re.MULTILINE)
+
+
+@dataclass
+class BatchTakeOutcome:
+    ok: bool
+    reason: str = ""
+    edges: list[dict[str, str]] | None = None
+    promoted: list[str] | None = None
+    waiting: list[str] | None = None
+    skipped: list[dict[str, str]] | None = None
+
+
+def _json(raw: str) -> Optional[dict[str, Any]]:
+    text = _FENCE_RE.sub("", (raw or "").strip())
+    start, end = text.find("{"), text.rfind("}")
+    if start < 0 or end <= start:
+        return None
+    try:
+        value = json.loads(text[start : end + 1])
+    except (ValueError, json.JSONDecodeError):
+        return None
+    return value if isinstance(value, dict) else None
+
+
+def _prompt(tasks: list[Any]) -> str:
+    rows = []
+    for task in tasks:
+        rows.append({
+            "id": task.id,
+            "title": (task.title or "")[:400],
+            "body": (task.body or "")[:3000],
+            "status": task.status,
+        })
+    return "Selected tasks:\n" + json.dumps(rows, ensure_ascii=False)
+
+
+def _incomplete_parents(conn: Any, task_id: str) -> bool:
+    row = conn.execute(
+        """SELECT 1 FROM task_links l JOIN tasks p ON p.id=l.parent_id
+           WHERE l.child_id=? AND p.status != 'done' LIMIT 1""",
+        (task_id,),
+    ).fetchone()
+    return row is not None
+
+
+def plan_and_take(task_ids: list[str], *, timeout: int = 90) -> BatchTakeOutcome:
+    """Add safe ordering edges, then promote immediately runnable tasks.
+
+    The operation is deliberately conservative: malformed/failed model output
+    changes nothing, and a graph edge rejected by the DB is reported rather
+    than bypassing cycle checks.
+    """
+    ids = list(dict.fromkeys(str(task_id) for task_id in task_ids if task_id))
+    if not ids:
+        return BatchTakeOutcome(False, "ids is required")
+    with kb.connect_closing() as conn:
+        tasks = []
+        skipped: list[dict[str, str]] = []
+        for task_id in ids:
+            task = kb.get_task(conn, task_id)
+            if task is None:
+                skipped.append({"id": task_id, "reason": "not found"})
+            elif task.status not in {"todo", "ready"}:
+                skipped.append({"id": task_id, "reason": f"status {task.status!r} cannot be batch-taken"})
+            else:
+                tasks.append(task)
+        if not tasks:
+            return BatchTakeOutcome(False, "no eligible tasks", skipped=skipped)
+
+        try:
+            from agent.auxiliary_client import call_llm
+            response = call_llm(
+                task="kanban_batch_planner",
+                messages=[
+                    {"role": "system", "content": _SYSTEM_PROMPT},
+                    {"role": "user", "content": _prompt(tasks)},
+                ],
+                temperature=0.1,
+                max_tokens=1500,
+                timeout=timeout,
+            )
+            raw = response.choices[0].message.content or ""
+        except Exception as exc:
+            logger.info("batch-take planner failed: %s", exc)
+            return BatchTakeOutcome(False, f"planner unavailable: {type(exc).__name__}", skipped=skipped)
+        parsed = _json(raw)
+        if (
+            parsed is None
+            or "edges" not in parsed
+            or not isinstance(parsed.get("edges"), list)
+        ):
+            return BatchTakeOutcome(False, "planner returned malformed JSON", skipped=skipped)
+
+        eligible = {task.id for task in tasks}
+        edges: list[dict[str, str]] = []
+        for item in parsed.get("edges", []):
+            if not isinstance(item, dict):
+                continue
+            before, after = item.get("before"), item.get("after")
+            if not isinstance(before, str) or not isinstance(after, str):
+                continue
+            if before == after or before not in eligible or after not in eligible:
+                continue
+            edge = {"before": before, "after": after, "reason": str(item.get("reason") or "overlapping work")[:300]}
+            if edge not in edges:
+                edges.append(edge)
+
+        # Apply planner edges one-by-one through the canonical DAG guard.
+        applied: list[dict[str, str]] = []
+        for edge in edges:
+            try:
+                kb.link_tasks(conn, edge["before"], edge["after"])
+                applied.append(edge)
+            except ValueError as exc:
+                skipped.append({"id": edge["after"], "reason": f"dependency rejected: {exc}"})
+
+        promoted: list[str] = []
+        waiting: list[str] = []
+        # Direct status updates mirror the dashboard bulk endpoint. A task
+        # with unfinished parents must remain todo; dispatcher's normal DAG
+        # promotion will move it to ready after those parents finish.
+        for task in tasks:
+            if _incomplete_parents(conn, task.id):
+                if task.status != "todo":
+                    with kb.write_txn(conn):
+                        conn.execute("UPDATE tasks SET status='todo' WHERE id=?", (task.id,))
+                waiting.append(task.id)
+            else:
+                if task.status != "ready":
+                    ok, reason = kb.promote_task(
+                        conn,
+                        task.id,
+                        actor="batch-planner",
+                        reason="batch take: no unfinished dependencies",
+                    )
+                    if not ok:
+                        skipped.append({"id": task.id, "reason": reason or "promotion refused"})
+                        continue
+                promoted.append(task.id)
+        return BatchTakeOutcome(True, edges=applied, promoted=promoted, waiting=waiting, skipped=skipped)

hermes_cli/kanban_db.py

@@ -5900,7 +5900,7 @@ def schedule_task(
 # for operators who want a tighter/looser probe cadence.
 DEFAULT_RATE_LIMIT_COOLDOWN_SECONDS = 300  # 5 minutes
 
-# Within this window a GitHub PR URL in a comment blocks re-spawn.
+# Within this window an open GitHub PR URL in a comment blocks re-spawn.
 _RESPAWN_GUARD_PR_WINDOW = 86400  # 24 hours
 
 # Pattern matching a GitHub PR URL in task comments.
@@ -5910,6 +5910,26 @@ def schedule_task(
 )
 
 
+def _is_open_github_pr(pr_url: str) -> bool:
+    """Return whether GitHub currently reports ``pr_url`` as open.
+
+    The respawn guard must never infer PR status from a URL alone: merged or
+    closed PRs are terminal and must not leave ready work parked for a day.
+    If GitHub cannot be queried (offline dispatcher, missing ``gh``, timeout,
+    or authentication error), fail open so a stale URL cannot block execution.
+    """
+    try:
+        result = subprocess.run(
+            ["gh", "pr", "view", pr_url, "--json", "state", "--jq", ".state"],
+            capture_output=True,
+            text=True,
+            timeout=10,
+        )
+    except (OSError, subprocess.TimeoutExpired):
+        return False
+    return result.returncode == 0 and result.stdout.strip().upper() == "OPEN"
+
+
 @dataclass
 class DispatchResult:
     """Outcome of a single ``dispatch`` pass."""
@@ -7172,9 +7192,10 @@ def check_respawn_guard(conn: sqlite3.Connection, task_id: str) -> Optional[str]
         arrives AFTER that completion — that's a deliberate re-run request.
 
     ``"active_pr"``
-        A GitHub PR URL appears in a recent task comment (within
-        ``_RESPAWN_GUARD_PR_WINDOW`` seconds).  A prior worker already
-        opened a PR; re-spawning risks a duplicate PR on the same task.
+        An open GitHub PR URL appears in a recent task comment (within
+        ``_RESPAWN_GUARD_PR_WINDOW`` seconds). A prior worker already opened
+        an active PR; re-spawning risks a duplicate PR on the same task.
+        Closed, merged, and unresolvable PR URLs do not guard the task.
 
     Stale / dead claim locks are NOT a guard reason — they are handled
     by ``release_stale_claims`` and ``detect_crashed_workers`` which
@@ -7256,13 +7277,16 @@ def check_respawn_guard(conn: sqlite3.Connection, task_id: str) -> Optional[str]
         if not requeued_after:
             return "recent_success"
 
-    # 4. GitHub PR URL in a recent comment — prior worker already opened a PR.
+    # 4. Open GitHub PR URL in a recent comment — a prior worker already
+    #    opened work that is still pending review. URL presence alone is not
+    #    enough: merged/closed PRs must release the task immediately.
     pr_cutoff = now - _RESPAWN_GUARD_PR_WINDOW
     for c in conn.execute(
         "SELECT body FROM task_comments WHERE task_id = ? AND created_at >= ?",
         (task_id, pr_cutoff),
     ).fetchall():
-        if c["body"] and _RESPAWN_GUARD_PR_URL_RE.search(c["body"]):
+        match = _RESPAWN_GUARD_PR_URL_RE.search(c["body"] or "")
+        if match and _is_open_github_pr(match.group(0)):
             return "active_pr"
 
     return None

hermes_cli/kanban_decompose.py

@@ -49,6 +49,10 @@
 logger = logging.getLogger(__name__)
 
 
+_DEFAULT_CONTEXT_BUDGET_TOKENS = 150_000
+_MIN_CONTEXT_BUDGET_TOKENS = 32_000
+
+
 _SYSTEM_PROMPT = """You are the Kanban decomposer for the Hermes Agent board.
 
 A user dropped a rough idea into the Triage column. Your job is to break it
@@ -59,6 +63,7 @@
   - The original task title and body
   - The list of available profiles (each with name + description)
   - The fallback "default_assignee" used when no profile fits
+  - A target maximum context budget for one fresh worker session
 
 Output a single JSON object with this exact shape:
 
@@ -70,20 +75,34 @@
         "title": "<concrete task title, imperative voice, <= 80 chars>",
         "body":  "<detailed spec for the worker on this child task>",
         "assignee": "<profile name from the roster, or null for default>",
-        "parents": [<int>, ...]
+        "parents": [<int>, ...],
+        "estimated_context_tokens": <rough integer estimate>
       },
       ...
     ]
   }
 
 Rules:
+  - The context-budget estimate is intentionally rough, not a promise. It
+    includes the worker's base prompt, task spec, likely repository/file/tool
+    output, and implementation conversation. Keep every child at or below the
+    supplied budget with sensible margin.
+  - First look for semantic seams. Split only at those seams; never create
+    arbitrary micro-tasks merely to hit a number.
+  - Among valid semantic splits, choose the FEWEST workers and the LARGEST
+    pieces that fit the budget. The objective is minimum handoffs and minimum
+    duplicate worker context, not maximum parallelism.
+  - If the whole task is plausibly within budget, return fanout=false. If a
+    semantic child would exceed budget, split that child further at its own
+    semantic seams before returning the graph.
   - "parents" is a list of INDICES (0-based) into this same "tasks" list,
     expressing actual data dependencies. Tasks with no parents run in
     PARALLEL. Tasks with parents wait until every parent completes.
-  - Prefer parallelism. If two tasks can be done independently, give
-    them no parents so the dispatcher fans them out at once.
-  - Use 2-6 tasks for normal work. Don't create 20 tiny tasks. Don't
-    cram everything into 1 task.
+  - Prefer parallelism only when it does not add a handoff or duplicate work.
+    If two tasks can be done independently, give them no parents so the
+    dispatcher can fan them out at once.
+  - Use 2-6 tasks for normal work. Don't create 20 tiny tasks. Do not cram
+    clearly over-budget work into 1 task.
   - Pick assignees from the roster by matching the task to the profile's
     DESCRIPTION (not just the name). When nothing matches well, use null
     and the system will route to the default_assignee.
@@ -98,7 +117,8 @@
     "rationale": "<one sentence>",
     "title": "<tightened title>",
     "body":  "<concrete spec for a single worker>",
-    "assignee": "<profile name from the roster, or null for default>"
+    "assignee": "<profile name from the roster, or null for default>",
+    "estimated_context_tokens": <rough integer estimate>
   }
 
 In that case the task stays as one work item, just with a tightened spec and
@@ -118,6 +138,8 @@
 {roster}
 
 Default assignee (used when no profile fits a task): {default_assignee}
+
+Maximum rough context budget per fresh worker: {context_budget_tokens:,} tokens
 """
 
 
@@ -214,6 +236,24 @@ def _resolve_default_assignee(cfg: dict) -> str:
         return "default"
 
 
+def _resolve_context_budget_tokens(cfg: dict) -> int:
+    """Resolve the decomposer's coarse per-worker context cap.
+
+    This is a planning guardrail, not a runtime token counter: actual context
+    varies with tool calls and model behavior. Keep a defensible lower bound so
+    malformed configuration cannot force pathological micro-task fanout.
+    """
+    kanban_cfg = cfg.get("kanban", {}) if isinstance(cfg, dict) else {}
+    raw = kanban_cfg.get("decomposer_context_budget_tokens")
+    if raw is None:
+        return _DEFAULT_CONTEXT_BUDGET_TOKENS
+    try:
+        budget = int(raw)
+    except (TypeError, ValueError):
+        return _DEFAULT_CONTEXT_BUDGET_TOKENS
+    return max(_MIN_CONTEXT_BUDGET_TOKENS, budget)
+
+
 def _build_roster() -> tuple[list[dict], set[str]]:
     """Return (roster_for_prompt, valid_assignee_names).
 
@@ -293,6 +333,7 @@ def decompose_task(
     cfg = _load_config()
     orchestrator = _resolve_orchestrator_profile(cfg)
     default_assignee = _resolve_default_assignee(cfg)
+    context_budget_tokens = _resolve_context_budget_tokens(cfg)
     kanban_cfg = cfg.get("kanban", {}) if isinstance(cfg, dict) else {}
     auto_promote = bool(kanban_cfg.get("auto_promote_children", True))
     roster, valid_names = _build_roster()
@@ -309,6 +350,7 @@ def decompose_task(
         body=_truncate(task.body or "(no body)", 4000),
         roster=_format_roster(roster),
         default_assignee=default_assignee,
+        context_budget_tokens=context_budget_tokens,
     )
 
     try: