feat(kanban): one workflow definition for columns, traits and transitions

Groundwork for user-defined columns and workflow per board (#54818).

- hermes_cli/kanban_workflow.py: Column/Workflow + DEFAULT_WORKFLOW, today's
  board written down as data (column order, traits, default kernel-event
  edges, manual move allow-list). Not consulted by the kernel yet.
- VALID_STATUSES, dashboard BOARD_COLUMNS, the kanban_list status enum and
  CLI status icons now derive from it. Fixes drift: the tool enum lacked
  scheduled/review; CLI icons lacked triage/review.
- GET /workflow on the dashboard plugin API.
- Tests pin DEFAULT_WORKFLOW to real behavior: every default event edge vs.
  its kernel verb, and all 64 PATCH src->dst moves vs. can_move.
This commit is contained in:
Ben Barclay
2026-09-24 19:58:52 +10:00
committed by kshitij
parent 82732650a1
commit 96db175da7
7 changed files with 467 additions and 12 deletions

View File

@@ -29,6 +29,7 @@ from dataclasses import dataclass
from pathlib import Path
from typing import Any, Iterable, Optional
from hermes_cli import kanban_workflow
from toolsets import get_toolset_names
_log = logging.getLogger(__name__)
@@ -103,7 +104,8 @@ def _git_out(cwd: Path, *args: str, timeout: int = 30) -> Optional[str]:
# --- Constants ---
VALID_STATUSES = {"triage", "todo", "scheduled", "ready", "running", "blocked", "review", "done", "archived"}
# Derived from the one workflow definition (``kanban_workflow``); a mutable copy for legacy callers.
VALID_STATUSES = set(kanban_workflow.DEFAULT_STATUSES)
VALID_INITIAL_STATUSES = {"running", "blocked"}
# Typed block reasons (routing in ``_route_block``); ``None`` = legacy un-typed.

View File

@@ -9,11 +9,9 @@ import time
from typing import Any, Callable, Iterable, Optional
from hermes_cli import kanban_db as kb
from hermes_cli import kanban_workflow
_STATUS_ICONS = {
"todo": "◻", "ready": "▶", "running": "●", "scheduled": "⏱",
"blocked": "⊘", "done": "✓", "archived": "—",
}
_STATUS_ICONS = {c.key: c.icon for c in kanban_workflow.DEFAULT_WORKFLOW} | {kanban_workflow.ARCHIVED: "—"}
_TASK_DICT_FIELDS = (
"id", "title", "body", "assignee", "status", "priority", "tenant",

View File

@@ -0,0 +1,199 @@
"""Kanban board workflow: the one definition of columns, traits and transitions.
Groundwork for user-defined columns and workflow per board (issue #54818). Target model —
``tasks.status`` holds a board-defined column key; kernel behavior attaches to
column *traits*, and kernel events resolve to a column through per-column *edges*
with a board-wide default, so an automatic transition never has to guess.
Phase 0 (this module): ``DEFAULT_WORKFLOW`` writes today's hardcoded behavior down
as data, and the scattered status copies (``VALID_STATUSES``, ``BOARD_COLUMNS``, the
agent-tool enum, CLI icons) derive from it. The kernel does not consult traits or
edges yet; ``tests/hermes_cli/test_kanban_workflow.py`` pins them to the kernel's real
behavior so the later phases that switch the kernel over cannot drift silently.
Pure data + stdlib only: ``kanban_db`` imports this module, never the reverse.
"""
from __future__ import annotations
from dataclasses import dataclass, field
from types import MappingProxyType
from typing import Iterator, Mapping, Optional
# ``archived`` is outside every workflow: a filter toggle, never a board column.
ARCHIVED = "archived"
# --- Traits: kernel behavior a column opts into ------------------------------------------
DECOMPOSE = "decompose" # specifier/decomposer input (today: triage)
WAIT_PARENTS = "wait_parents" # parked until every parent is satisfied (todo)
HOLD_TIME = "hold_time" # time gate, not dispatchable (scheduled)
DISPATCH_IMPLEMENT = "dispatch_implement" # dispatcher claims for the assignee (ready)
CLAIMED = "claimed" # a worker holds the card (running) — Phase 1 turns
# this into a claim fact instead of a column
HOLD_HUMAN = "hold_human" # waiting on a person / external blocker (blocked)
DISPATCH_REVIEW = "dispatch_review" # dispatcher claims for the reviewer (review)
TERMINAL = "terminal" # satisfies a child's parent gate (done)
TRAITS = frozenset({
DECOMPOSE, WAIT_PARENTS, HOLD_TIME, DISPATCH_IMPLEMENT, CLAIMED, HOLD_HUMAN, DISPATCH_REVIEW, TERMINAL,
})
# --- Kernel events that move a card without a human choosing the target -----------------
EV_CLAIM = "claim" # dispatcher claims the card
EV_COMPLETE = "complete" # worker/human completes
EV_BLOCK = "block" # worker/human blocks
EV_SCHEDULE = "schedule" # parked on a time gate
EV_REVIEW = "review" # implementer hands off for review
EV_CHANGES = "changes" # reviewer requests changes
EV_PARENTS_DONE = "parents_done" # last open parent is satisfied
EVENTS = frozenset({EV_CLAIM, EV_COMPLETE, EV_BLOCK, EV_SCHEDULE, EV_REVIEW, EV_CHANGES, EV_PARENTS_DONE})
@dataclass(frozen=True)
class Column:
"""One board column. ``key`` is stable (stored in ``tasks.status``); ``label`` is display-only."""
key: str
label: str
icon: str = "?" # single-glyph CLI marker
traits: frozenset = frozenset()
# Offered as a drag/menu target in board UIs. False for columns a card normally
# reaches only through a verb with extra input (reviewer, wake time) or the kernel.
drag_target: bool = True
# Per-column event edges; an event missing here falls back to ``Workflow.defaults``.
on: Mapping[str, str] = field(default_factory=lambda: MappingProxyType({}))
@dataclass(frozen=True)
class Workflow:
"""Columns (board order), default event edges, and the manual move allow-list."""
columns: tuple
defaults: Mapping[str, str]
# Manual moves (drag, PATCH status) a human may request: ``src -> {dst}``. Archive is
# always allowed and not listed. Allowed != guaranteed: verbs still apply their own
# gates (parents open, completion evidence).
manual: Mapping[str, frozenset]
def __iter__(self) -> Iterator[Column]:
return iter(self.columns)
def keys(self) -> tuple:
"""Column keys in board order (never includes ``archived``)."""
return tuple(c.key for c in self.columns)
def column(self, key: str) -> Optional[Column]:
return next((c for c in self.columns if c.key == key), None)
def keys_with(self, trait: str) -> tuple:
"""Keys of every column carrying ``trait``, in board order."""
if trait not in TRAITS:
raise ValueError(f"unknown kanban workflow trait {trait!r}")
return tuple(c.key for c in self.columns if trait in c.traits)
def on_event(self, key: str, event: str) -> str:
"""Column a card in ``key`` moves to when the kernel emits ``event``."""
if event not in EVENTS:
raise ValueError(f"unknown kanban workflow event {event!r}")
col = self.column(key)
if col is not None and event in col.on:
return col.on[event]
return self.defaults[event]
def can_move(self, src: str, dst: str) -> bool:
"""True when a human may request ``src -> dst`` (archive always allowed)."""
if dst == ARCHIVED:
return True
return dst in self.manual.get(src, frozenset())
def to_dict(self) -> dict:
"""JSON shape served by the dashboard's ``GET /workflow``."""
return {
"columns": [
{"key": c.key, "label": c.label, "icon": c.icon, "traits": sorted(c.traits),
"drag_target": c.drag_target, "on": dict(c.on)}
for c in self.columns
],
"defaults": dict(self.defaults),
"manual": {src: sorted(dsts) for src, dsts in self.manual.items()},
"archived": ARCHIVED,
}
def validate(self) -> None:
"""Raise ``ValueError`` on an inconsistent workflow (unknown keys, dead-end events)."""
keys = self.keys()
if len(set(keys)) != len(keys):
raise ValueError(f"duplicate column keys: {keys}")
if ARCHIVED in keys:
raise ValueError(f"{ARCHIVED!r} is reserved and cannot be a column")
known = set(keys)
for c in self.columns:
unknown = c.traits - TRAITS
if unknown:
raise ValueError(f"column {c.key!r}: unknown traits {sorted(unknown)}")
for ev, dst in c.on.items():
if ev not in EVENTS or dst not in known:
raise ValueError(f"column {c.key!r}: bad edge {ev!r} -> {dst!r}")
missing = EVENTS - set(self.defaults)
if missing:
raise ValueError(f"no default column for events {sorted(missing)}")
for ev, dst in self.defaults.items():
if dst not in known:
raise ValueError(f"default edge {ev!r} -> unknown column {dst!r}")
for src, dsts in self.manual.items():
if src not in known or not set(dsts) <= known:
raise ValueError(f"manual moves from {src!r} reference unknown columns")
if not self.keys_with(TERMINAL):
raise ValueError("workflow needs at least one terminal column")
def _col(key: str, label: str, icon: str, *traits: str, drag_target: bool = True) -> Column:
return Column(key=key, label=label, icon=icon, traits=frozenset(traits), drag_target=drag_target)
def _frozen_manual(table: Mapping[str, tuple]) -> Mapping[str, frozenset]:
return MappingProxyType({src: frozenset(dsts) for src, dsts in table.items()})
# Today's board, written down. Column order = dashboard order. ``drag_target=False``
# mirrors the desktop's LOCKED_COLUMNS (review needs a reviewer, scheduled a wake time,
# running is kernel-only). The manual table is the measured PATCH /tasks/{id} matrix on
# a parentless task; ``test_kanban_workflow.py`` re-measures it against the live code.
DEFAULT_WORKFLOW = Workflow(
columns=(
_col("triage", "Triage", "◇", DECOMPOSE),
_col("todo", "Todo", "◻", WAIT_PARENTS),
_col("scheduled", "Scheduled", "⏱", HOLD_TIME, drag_target=False),
_col("ready", "Ready", "▶", DISPATCH_IMPLEMENT),
_col("running", "Running", "●", CLAIMED, drag_target=False),
_col("blocked", "Blocked", "⊘", HOLD_HUMAN),
_col("review", "Review", "◎", DISPATCH_REVIEW, drag_target=False),
_col("done", "Done", "✓", TERMINAL),
),
defaults=MappingProxyType({
EV_CLAIM: "running",
EV_COMPLETE: "done",
EV_BLOCK: "blocked",
EV_SCHEDULE: "scheduled",
EV_REVIEW: "review",
EV_CHANGES: "ready",
EV_PARENTS_DONE: "ready",
}),
manual=_frozen_manual({
"triage": ("todo", "ready"),
"todo": ("triage", "scheduled", "ready"),
"scheduled": ("triage", "todo", "ready"),
"ready": ("triage", "todo", "scheduled", "blocked", "review", "done"),
"running": ("triage", "todo", "scheduled", "ready", "blocked", "review", "done"),
"blocked": ("triage", "todo", "scheduled", "ready", "done"),
# Known quirk kept for zero behavior change: review -> todo lands in ``ready``
# (reopen_review_task ignores the requested target). Phase 1 fixes it.
"review": ("triage", "todo", "ready", "done"),
"done": ("triage", "todo", "ready"),
}),
)
DEFAULT_WORKFLOW.validate()
# Every value ``tasks.status`` can hold under the default workflow.
DEFAULT_STATUSES = frozenset(DEFAULT_WORKFLOW.keys()) | {ARCHIVED}

View File

@@ -29,6 +29,7 @@ from fastapi.responses import FileResponse
from pydantic import BaseModel, Field
from hermes_cli import kanban_db
from hermes_cli import kanban_workflow
from hermes_cli.web_read_coalescing import coalesced_read
from hermes_cli import kanban_db_connect as kbc
from hermes_cli import kanban_db_notify as kbn
@@ -164,9 +165,9 @@ def _errors_to_500(prefix: str) -> Iterator[None]:
# --- Serialization helpers --------------------------------------------------
# Dashboard columns, left-to-right ("archived" is a filter toggle, not a column). Keep in
# sync with kanban_db.VALID_STATUSES — a status missing here gets mis-bucketed into ``todo``.
BOARD_COLUMNS: list[str] = ["triage", "todo", "scheduled", "ready", "running", "blocked", "review", "done"]
# Dashboard columns, left-to-right ("archived" is a filter toggle, not a column), from the
# one workflow definition. A status missing here gets mis-bucketed into ``todo``.
BOARD_COLUMNS: list[str] = list(kanban_workflow.DEFAULT_WORKFLOW.keys())
_CARD_SUMMARY_PREVIEW_CHARS = 200
@@ -1615,6 +1616,14 @@ class OrchestrationSettingsBody(BaseModel):
_PROFILE_SETTINGS = ("orchestrator_profile", "default_assignee")
@router.get("/workflow")
def get_workflow():
"""Board columns (order, label, icon, traits, drag target), default event edges and
the manual move allow-list. Every board uses the default workflow today; per-board
workflows (``board.json``) arrive in a later phase behind this same shape."""
return kanban_workflow.DEFAULT_WORKFLOW.to_dict()
@router.get("/orchestration")
def get_orchestration_settings():
"""Current orchestration knobs from config.yaml plus the resolved effective

View File

@@ -0,0 +1,151 @@
"""Kanban workflow definition: shape, derived copies, and kernel-event edges.
``DEFAULT_WORKFLOW`` is data the kernel does not consult yet (Phase 0 of #54818).
These tests pin it to what the kernel actually does, so a later phase that switches
the kernel over to reading the workflow cannot change behavior unnoticed, and a
kernel change that forgets the workflow fails here.
"""
from __future__ import annotations
from dataclasses import replace
from pathlib import Path
from types import MappingProxyType
import pytest
from hermes_cli import kanban_db as kb
from hermes_cli import kanban_db_connect as kbc
from hermes_cli import kanban_output
from hermes_cli import kanban_workflow as kw
from tools import kanban_tools_schemas
W = kw.DEFAULT_WORKFLOW
# --- shape + single source of truth ------------------------------------------------------
def test_default_workflow_is_valid_and_excludes_archived():
W.validate()
assert kw.ARCHIVED not in W.keys()
def test_status_copies_derive_from_the_workflow():
assert kb.VALID_STATUSES == set(W.keys()) | {kw.ARCHIVED}
# CLI: every listable status has a glyph (was missing triage + review).
assert set(kanban_output._STATUS_ICONS) == kb.VALID_STATUSES
def test_agent_tool_status_enum_covers_every_status():
# Was missing scheduled + review, so an agent could not filter on them.
schema = next(s for s in _all_tool_schemas() if s["name"] == "kanban_list")
enum = schema["parameters"]["properties"]["status"]["enum"]
assert set(enum) == kb.VALID_STATUSES
def _all_tool_schemas():
for value in vars(kanban_tools_schemas).values():
if isinstance(value, dict) and "name" in value and "parameters" in value:
yield value
def test_every_trait_and_event_is_used_by_the_default():
# A trait or event no default column uses has no behavior to pin — it would be dead data.
for trait in kw.TRAITS:
assert W.keys_with(trait), trait
assert set(W.defaults) == kw.EVENTS
with pytest.raises(ValueError):
W.keys_with("nope")
def test_on_event_prefers_column_edge_over_default():
custom = replace(
W,
columns=tuple(replace(c, on=MappingProxyType({kw.EV_COMPLETE: "review"})) if c.key == "ready" else c
for c in W.columns),
)
custom.validate()
assert custom.on_event("ready", kw.EV_COMPLETE) == "review"
assert custom.on_event("todo", kw.EV_COMPLETE) == "done"
with pytest.raises(ValueError):
custom.on_event("ready", "nope")
def test_archive_is_always_a_manual_move():
assert all(W.can_move(k, kw.ARCHIVED) for k in W.keys())
assert not any(W.can_move(k, "running") for k in W.keys())
@pytest.mark.parametrize("mutate, message", [
(lambda w: replace(w, columns=w.columns + (w.columns[0],)), "duplicate"),
(lambda w: replace(w, columns=w.columns + (kw.Column(key=kw.ARCHIVED, label="A"),)), "reserved"),
(lambda w: replace(w, defaults=MappingProxyType({k: v for k, v in w.defaults.items()
if k != kw.EV_BLOCK})), "no default"),
(lambda w: replace(w, defaults=MappingProxyType({**w.defaults, kw.EV_BLOCK: "gone"})), "unknown column"),
(lambda w: replace(w, manual=MappingProxyType({"ready": frozenset({"gone"})})), "manual"),
(lambda w: replace(w, columns=tuple(replace(c, traits=frozenset()) if c.key == "done" else c
for c in w.columns)), "terminal"),
(lambda w: replace(w, columns=tuple(replace(c, traits=frozenset({"bogus"})) if c.key == "done" else c
for c in w.columns)), "unknown traits"),
])
def test_validate_rejects_inconsistent_workflows(mutate, message):
with pytest.raises(ValueError, match=message):
mutate(W).validate()
# --- kernel events land where the workflow says --------------------------------------------
@pytest.fixture
def conn(tmp_path: Path):
db = kbc.connect(tmp_path / "kanban.db")
try:
yield db
finally:
db.close()
def _status(conn, task_id: str) -> str:
return kb.get_task(conn, task_id).status
def test_edge_claim(conn):
tid = kb.create_task(conn, title="t", assignee="builder")
assert kb.claim_task(conn, tid) is not None
assert _status(conn, tid) == W.on_event("ready", kw.EV_CLAIM)
def test_edge_complete_block_schedule(conn):
for event, act in (
(kw.EV_COMPLETE, lambda tid: kb.complete_task(conn, tid, summary="s")),
(kw.EV_BLOCK, lambda tid: kb.block_task(conn, tid, reason="r")),
(kw.EV_SCHEDULE, lambda tid: kb.schedule_task(conn, tid, reason="r")),
):
tid = kb.create_task(conn, title=event, assignee="builder")
assert kb.claim_task(conn, tid) is not None
assert act(tid)
assert _status(conn, tid) == W.on_event("running", event), event
def test_edge_review_and_changes(conn):
tid = kb.create_task(conn, title="t", assignee="builder")
assert kb.claim_task(conn, tid) is not None
assert kb.request_review(conn, tid, summary="s", reviewer="reviewer")
assert _status(conn, tid) == W.on_event("running", kw.EV_REVIEW)
review = kb.claim_review_task(conn, tid, claimer="reviewer:1")
assert review is not None
ok, _ = kb.request_changes(conn, tid, reason="fix it", expected_run_id=review.current_run_id)
assert ok
assert _status(conn, tid) == W.on_event("running", kw.EV_CHANGES)
def test_edge_parents_done(conn):
parent = kb.create_task(conn, title="p", assignee="builder")
child = kb.create_task(conn, title="c", assignee="builder", parents=[parent])
waiting = _status(conn, child)
assert waiting in W.keys_with(kw.WAIT_PARENTS)
assert kb.complete_task(conn, parent, summary="s")
kb.recompute_ready(conn)
assert _status(conn, child) == W.on_event(waiting, kw.EV_PARENTS_DONE)

View File

@@ -0,0 +1,97 @@
"""Manual move matrix: ``DEFAULT_WORKFLOW.manual`` vs. the live ``PATCH /tasks/{id}``.
Every ``src -> dst`` pair is requested through the dashboard API on a parentless task.
The workflow allow-list must agree with what the server accepts, in both directions.
Also covers ``GET /workflow`` and ``GET /board`` column order.
"""
from __future__ import annotations
import importlib.util
import sys
from pathlib import Path
import pytest
from fastapi import FastAPI
from fastapi.testclient import TestClient
from hermes_cli import kanban_db as kb
from hermes_cli import kanban_db_connect as kbc
from hermes_cli import kanban_workflow as kw
W = kw.DEFAULT_WORKFLOW
# Accepted moves whose card lands somewhere other than the requested column.
# review -> todo goes through reopen_review_task, which re-gates to ``ready``.
_KNOWN_REDIRECTS = {("review", "todo"): "ready"}
def _load_plugin_router():
plugin_file = Path(__file__).resolve().parents[2] / "plugins" / "kanban" / "dashboard" / "plugin_api.py"
spec = importlib.util.spec_from_file_location("hermes_dashboard_plugin_kanban_workflow_test", plugin_file)
mod = importlib.util.module_from_spec(spec)
sys.modules[spec.name] = mod
spec.loader.exec_module(mod)
return mod.router
@pytest.fixture
def client(tmp_path, monkeypatch):
home = tmp_path / ".hermes"
home.mkdir()
monkeypatch.setenv("HERMES_HOME", str(home))
monkeypatch.setattr(Path, "home", lambda: tmp_path)
kb.init_db()
app = FastAPI()
app.include_router(_load_plugin_router(), prefix="/k")
return TestClient(app)
def _task_in(client, src: str) -> str:
tid = client.post("/k/tasks", json={"title": src, "assignee": "a", "triage": src == "triage"}).json()["task"]["id"]
with kbc.connect_closing() as conn:
if src == "todo":
assert client.patch(f"/k/tasks/{tid}", json={"status": "todo"}).status_code == 200
elif src == "running":
assert kb.claim_task(conn, tid) is not None
elif src == "blocked":
assert kb.block_task(conn, tid, reason="x")
elif src == "scheduled":
assert kb.schedule_task(conn, tid, reason="x")
elif src == "review":
assert kb.request_review(conn, tid, summary="s", force=True)
elif src == "done":
assert kb.complete_task(conn, tid, summary="s", force=True)
assert kb.get_task(conn, tid).status == src
return tid
@pytest.mark.parametrize("src", W.keys())
def test_manual_moves_match_server(client, src):
mismatches = []
for dst in (*W.keys(), kw.ARCHIVED):
if dst == src:
continue
tid = _task_in(client, src)
# ``done`` needs completion evidence; that is a verb gate, not a transition rule.
body = {"status": dst, **({"summary": "s"} if dst == "done" else {})}
r = client.patch(f"/k/tasks/{tid}", json=body)
with kbc.connect_closing() as conn:
landed = kb.get_task(conn, tid).status
accepted = r.status_code == 200
if accepted != W.can_move(src, dst):
mismatches.append(f"{src}->{dst}: server {r.status_code}, workflow can_move={W.can_move(src, dst)}")
elif accepted and landed != _KNOWN_REDIRECTS.get((src, dst), dst):
mismatches.append(f"{src}->{dst}: landed in {landed}")
assert not mismatches, "\n".join(mismatches)
def test_get_workflow(client):
r = client.get("/k/workflow")
assert r.status_code == 200
assert r.json() == W.to_dict()
def test_board_columns_follow_workflow_order(client):
names = [c["name"] for c in client.get("/k/board").json()["columns"]]
assert names == list(W.keys())

View File

@@ -3,6 +3,8 @@ from __future__ import annotations
from typing import Any
from hermes_cli.kanban_workflow import ARCHIVED, DEFAULT_WORKFLOW
_DESC_TASK_ID_DEFAULT = (
"Task id. If omitted, defaults to HERMES_KANBAN_TASK from the env — the "
"task the dispatcher spawned you to work on. That default only exists for "
@@ -76,10 +78,7 @@ KANBAN_LIST_SCHEMA = _schema(
"assignee": _prop("string", "Optional assignee/profile filter."),
"status": {
"type": "string",
"enum": [
"triage", "todo", "ready", "running",
"blocked", "done", "archived",
],
"enum": [*DEFAULT_WORKFLOW.keys(), ARCHIVED],
"description": "Optional task status filter.",
},
"tenant": _prop("string", "Optional tenant/project namespace filter."),