diff --git a/hermes_cli/kanban.py b/hermes_cli/kanban.py index 4e975bb3e8d7..c6b3669023b5 100644 --- a/hermes_cli/kanban.py +++ b/hermes_cli/kanban.py @@ -2016,6 +2016,7 @@ def _cmd_dispatch(args: argparse.Namespace) -> int: "timed_out": res.timed_out, "stale": res.stale, "auto_blocked": res.auto_blocked, + "preflight_blocked": res.preflight_blocked, "promoted": res.promoted, "spawned": [ {"task_id": tid, "assignee": who, "workspace": ws} @@ -2038,6 +2039,9 @@ def _cmd_dispatch(args: argparse.Namespace) -> int: print(f"Auto-blocked: {len(res.auto_blocked)}") if res.auto_blocked: print(f" {', '.join(res.auto_blocked)}") + print(f"Preflight-blocked: {len(res.preflight_blocked)}") + if res.preflight_blocked: + print(f" {', '.join(res.preflight_blocked)}") print(f"Promoted: {res.promoted}") print(f"Spawned: {len(res.spawned)}") for tid, who, ws in res.spawned: @@ -2149,13 +2153,14 @@ def _on_tick(res): return did_work = ( res.reclaimed or res.crashed or res.timed_out or res.promoted - or res.spawned or res.auto_blocked or res.stale + or res.spawned or res.auto_blocked or res.stale or res.preflight_blocked ) if did_work: print( f"[{_fmt_ts(int(time.time()))}] " f"reclaimed={res.reclaimed} crashed={len(res.crashed)} " f"timed_out={len(res.timed_out)} stale={len(res.stale)} " + f"preflight_blocked={len(res.preflight_blocked)} " f"promoted={res.promoted} spawned={len(res.spawned)} " f"auto_blocked={len(res.auto_blocked)}", flush=True, diff --git a/hermes_cli/kanban_db.py b/hermes_cli/kanban_db.py index b0a076480ac3..2f7f7300ed21 100644 --- a/hermes_cli/kanban_db.py +++ b/hermes_cli/kanban_db.py @@ -3584,6 +3584,8 @@ class DispatchResult: """Task ids reclaimed because their worker PID disappeared.""" auto_blocked: list[str] = field(default_factory=list) """Task ids auto-blocked by the spawn-failure circuit breaker.""" + preflight_blocked: list[str] = field(default_factory=list) + """Task ids refused by pre-dispatch validation before worker spawn.""" timed_out: list[str] = field(default_factory=list) """Task ids whose workers exceeded ``max_runtime_seconds``.""" stale: list[str] = field(default_factory=list) @@ -4562,7 +4564,6 @@ def has_spawnable_ready(conn: sqlite3.Connection) -> bool: return True return False - def has_spawnable_review(conn: sqlite3.Connection) -> bool: """Return True iff there is at least one review+assigned+unclaimed task whose assignee maps to a real Hermes profile. @@ -4588,6 +4589,108 @@ def has_spawnable_review(conn: sqlite3.Connection) -> bool: return False + +def _strict_dispatch_preflight_enabled() -> bool: + """Return True when optional contract/workspace policy checks are enabled. + + Existing profile-existence filtering remains always on in the dispatcher. + These stricter checks are opt-in so lightweight boards are not forced into + a CAT-style task contract overnight. + """ + env = os.environ.get("HERMES_KANBAN_STRICT_DISPATCH_PREFLIGHT", "").strip().lower() + if env in {"1", "true", "yes", "on"}: + return True + if env in {"0", "false", "no", "off"}: + return False + try: + from hermes_cli.config import load_config + cfg = load_config() + return bool((cfg.get("kanban") or {}).get("strict_dispatch_preflight", False)) + except Exception: + return False + + +def _selection_packet_must_not_spawn(task: Task) -> Optional[str]: + text = f"{task.title or ''}\n{task.body or ''}".lower() + if "selection packet" in text or "scheduler-visible selection packet" in text: + return "selection packet is a planning artifact and must not auto-dispatch" + if "[selection" in text or "[packet" in text: + return "selection/packet task marker is non-runnable" + return None + + +def _validate_dispatch_contract(task: Task) -> Optional[str]: + """CAT-style task contract lint for strict dispatcher mode.""" + text = f"{task.title or ''}\n{task.body or ''}".lower() + required_any = { + "workspace mode": ("workspace mode", "workspace_mode", "workspace:"), + "write authority": ("write authority", "write_authority"), + "mutation authority": ( + "mutation authority", "kanban mutation", "cron mutation", "git mutation", + ), + "conflict boundary": ("conflict boundary", "max parallelism", "parallelism"), + } + missing = [label for label, needles in required_any.items() if not any(n in text for n in needles)] + if missing: + return "missing dispatch contract fields: " + ", ".join(missing) + return None + + +def _validate_workspace_preflight(task: Task, workspace: Path) -> Optional[str]: + kind = task.workspace_kind or "scratch" + if kind in {"dir", "worktree"} and not Path(str(workspace)).is_absolute(): + return f"workspace path is not absolute: {workspace}" + if kind == "dir" and not workspace.exists(): + return f"dir workspace does not exist: {workspace}" + if kind == "worktree": + if not workspace.exists(): + return f"worktree workspace does not exist: {workspace}" + if not (workspace / ".git").exists(): + return f"worktree workspace is not a git worktree: {workspace}" + try: + import subprocess + top = subprocess.run( + ["git", "-C", str(workspace), "rev-parse", "--show-toplevel"], + capture_output=True, text=True, timeout=5, + ) + if top.returncode != 0: + return f"worktree workspace is not inside a git checkout: {workspace}" + except Exception as exc: + return f"could not validate git worktree {workspace}: {exc}" + return None + + +def _workspace_conflict_error(conn: sqlite3.Connection, task: Task, workspace: Path) -> Optional[str]: + rows = conn.execute( + "SELECT id, title FROM tasks " + "WHERE status = 'running' AND id != ? AND workspace_path = ? LIMIT 3", + (task.id, str(workspace)), + ).fetchall() + if rows: + ids = ", ".join(str(r["id"]) for r in rows) + return f"workspace already has running task(s): {ids}" + return None + + +def _preflight_real_dispatch( + conn: sqlite3.Connection, task: Task, workspace: Path, *, strict: bool, +) -> Optional[str]: + """Fail-closed guard run after claim but before worker spawn.""" + selection_error = _selection_packet_must_not_spawn(task) + if selection_error: + return selection_error + conflict_error = _workspace_conflict_error(conn, task, workspace) + if conflict_error: + return conflict_error + if strict: + workspace_error = _validate_workspace_preflight(task, workspace) + if workspace_error: + return workspace_error + contract_error = _validate_dispatch_contract(task) + if contract_error: + return contract_error + return None + def dispatch_once( conn: sqlite3.Connection, *, @@ -4692,7 +4795,7 @@ def dispatch_once( "SELECT COUNT(*) FROM tasks WHERE status = 'running'" ).fetchone()[0] ) - + strict_preflight = _strict_dispatch_preflight_enabled() ready_rows = conn.execute( "SELECT id, assignee FROM tasks " "WHERE status = 'ready' AND claim_lock IS NULL " @@ -4786,6 +4889,21 @@ def dispatch_once( continue # Persist the resolved workspace path so the worker can cd there. set_workspace_path(conn, claimed.id, str(workspace)) + if spawn_fn is None: + preflight_error = _preflight_real_dispatch( + conn, claimed, workspace, strict=strict_preflight, + ) + if preflight_error: + auto = _record_spawn_failure( + conn, + claimed.id, + f"preflight: {preflight_error}", + failure_limit=failure_limit, + ) + result.preflight_blocked.append(claimed.id) + if auto: + result.auto_blocked.append(claimed.id) + continue _spawn = spawn_fn if spawn_fn is not None else _default_spawn try: # Back-compat: older spawn_fn signatures accept only diff --git a/tests/hermes_cli/test_kanban_db.py b/tests/hermes_cli/test_kanban_db.py index 6e338ffa8796..00603ee8dd4e 100644 --- a/tests/hermes_cli/test_kanban_db.py +++ b/tests/hermes_cli/test_kanban_db.py @@ -1046,6 +1046,91 @@ def fake_spawn(task, workspace): assert kb.get_task(conn, ready_a).status == "running" assert kb.get_task(conn, ready_b).status == "ready" +def test_dispatch_preflight_blocks_selection_packet_before_spawn(kanban_home): + with kb.connect() as conn: + t = kb.create_task( + conn, + title="Scheduler-visible selection packet", + body="Selection packet for future dispatcher; not executable.", + assignee="default", + ) + res = kb.dispatch_once(conn, failure_limit=1) + task = kb.get_task(conn, t) + + assert res.preflight_blocked == [t] + assert res.spawned == [] + assert task.status == "blocked" + assert "selection packet" in task.last_failure_error + + +def test_strict_dispatch_preflight_requires_contract(kanban_home, monkeypatch): + monkeypatch.setenv("HERMES_KANBAN_STRICT_DISPATCH_PREFLIGHT", "1") + with kb.connect() as conn: + t = kb.create_task(conn, title="implementation without contract", assignee="default") + res = kb.dispatch_once(conn, failure_limit=1) + task = kb.get_task(conn, t) + + assert res.preflight_blocked == [t] + assert res.spawned == [] + assert task.status == "blocked" + assert "missing dispatch contract fields" in task.last_failure_error + + +def test_strict_dispatch_preflight_requires_existing_worktree(kanban_home, monkeypatch, tmp_path): + monkeypatch.setenv("HERMES_KANBAN_STRICT_DISPATCH_PREFLIGHT", "1") + missing = tmp_path / ".worktrees" / "missing-task" + body = """ + Workspace mode: worktree + Write authority: repo docs only + Kanban mutation authority: comments only + Max parallelism/conflict boundary: one writer + """ + with kb.connect() as conn: + t = kb.create_task( + conn, + title="implementation with missing worktree", + body=body, + assignee="default", + workspace_kind="worktree", + workspace_path=str(missing), + ) + res = kb.dispatch_once(conn, failure_limit=1) + task = kb.get_task(conn, t) + + assert res.preflight_blocked == [t] + assert res.spawned == [] + assert task.status == "blocked" + assert "worktree workspace does not exist" in task.last_failure_error + + +def test_dispatch_preflight_blocks_duplicate_running_workspace(kanban_home, tmp_path): + workspace = tmp_path / "shared" + workspace.mkdir() + with kb.connect() as conn: + running = kb.create_task( + conn, + title="first writer", + assignee="default", + workspace_kind="dir", + workspace_path=str(workspace), + ) + assert kb.claim_task(conn, running) is not None + kb.set_workspace_path(conn, running, str(workspace)) + contender = kb.create_task( + conn, + title="second writer", + assignee="default", + workspace_kind="dir", + workspace_path=str(workspace), + ) + res = kb.dispatch_once(conn, failure_limit=1) + task = kb.get_task(conn, contender) + + assert res.preflight_blocked == [contender] + assert res.spawned == [] + assert task.status == "blocked" + assert "workspace already has running task" in task.last_failure_error + def test_dispatch_reclaims_stale_before_spawning(kanban_home): with kb.connect() as conn: