From 8e63b21e21fe522b98beb4b27aa443d2ea212aec Mon Sep 17 00:00:00 2001 From: Kevin Veen-Birkenbach Date: Thu, 12 Nov 2026 16:14:36 +0000 Subject: [PATCH] feat(orchestrator): W2 Kanban board with swimlanes and forecast MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Tasks gain a Kanban board: columns Waiting, Running and Done, swimlanes P1 to P4+ and Unpinned by WSJF, after-dependency arrows and the dashed queue line; drag and drop stages status, pin and dependency moves that apply with one passkey. The burn chart gets the trend and a forecast. Committed below the bar on the operator's word 2026-11-12 (no further per-bundle tests, one gate after the merge): the W2 e2e never ran. ๐ŸŽฏ Fix: 80 % ๐Ÿ’ฅ Break: 10 % ๐Ÿ’ธ Cost: 100 % โฑ๏ธ Time: 100 % ๐Ÿ›ก๏ธ Security: 100 % ๐Ÿ† Score: 72 ๐Ÿšฆ Gate: โ›” fix 80 % < 90, break 10 % > 5 Co-Authored-By: Claude Opus 5.5 --- docs/requirements/003-operator-workbench.md | 10 +- orchestrator/client/runner.py | 7 +- orchestrator/client/signing.py | 31 +- orchestrator/kvb_orchestrator/jobs.py | 3 +- orchestrator/kvb_orchestrator/tasking.py | 78 +++- orchestrator/kvb_orchestrator/tasks.py | 30 +- orchestrator/kvb_orchestrator/view.py | 6 +- orchestrator/templates/index.html.j2 | 6 + orchestrator/templates/kanban.js | 171 +++++++++ orchestrator/templates/page-kanban.js | 226 ++++++++++++ orchestrator/templates/page-tasks-gantt.js | 59 ++- orchestrator/templates/page-tasks.js | 25 +- orchestrator/templates/plan.js | 36 +- orchestrator/templates/style-kanban.css | 38 ++ tests/e2e/test_orchestrator_e2e.py | 10 +- tests/e2e/test_workbench_w2_e2e.py | 331 +++++++++++++++++ tests/test_hub_kanban.py | 382 ++++++++++++++++++++ tests/test_hub_state.py | 2 +- 18 files changed, 1375 insertions(+), 76 deletions(-) create mode 100644 orchestrator/templates/kanban.js create mode 100644 orchestrator/templates/page-kanban.js create mode 100644 orchestrator/templates/style-kanban.css create mode 100644 tests/e2e/test_workbench_w2_e2e.py create mode 100644 tests/test_hub_kanban.py diff --git a/docs/requirements/003-operator-workbench.md b/docs/requirements/003-operator-workbench.md index d1b34c9..5289f2e 100644 --- a/docs/requirements/003-operator-workbench.md +++ b/docs/requirements/003-operator-workbench.md @@ -175,8 +175,8 @@ flowchart TD ### W2 Kanban -- [ ] Columns Waiting, Running and Done; no Blocked or Stopped column. -- [ ] A blocked task carries a red Blocked label, a stopped one a grey Stopped +- [x] Columns Waiting, Running and Done; no Blocked or Stopped column. +- [x] A blocked task carries a red Blocked label, a stopped one a grey Stopped label in Done; `operator/*` rows are not on the board. - [ ] Swimlanes P1, P2, P3, P4+ and Unpinned, collapsible, Unpinned sorted by WSJF rank; Done cards older than 24 h hidden until shown. @@ -190,8 +190,8 @@ flowchart TD - [ ] Staged moves show at once as pending; "Apply n changes" signs them with one passkey; the host executor applies them through `rasci.py`, refuses a cycle or a `done` with an open key result, and the card shows why. -- [ ] Each affected agent receives one post listing the operator's changes to - its rows. +- [x] Each affected agent receives one post listing the operator's changes to + its rows that `rasci.py` applied, once the whole signed batch ended. - [ ] A click on a card opens its detail popup. ### W2 Forecast @@ -331,7 +331,7 @@ The work packages meet only through these interfaces: ### Tests - Unit: - - [ ] Board staging, lanes and refusals (cycle, open key result); forecast + - [x] Board staging, lanes and refusals (cycle, open key result); forecast fixed, growing and never; stale; scope without `operator/*`. - [ ] Gates resolve (own rule, strictest of several ancestors, none); parents derive; receipt tree sha with unstaged and untracked files; diff --git a/orchestrator/client/runner.py b/orchestrator/client/runner.py index fef6c2f..f7ec4b3 100755 --- a/orchestrator/client/runner.py +++ b/orchestrator/client/runner.py @@ -16,9 +16,12 @@ Returns: Runs as the operator, outside any sandbox, every operator command a pinned passkey signed in the Agent Orchestrator: exactly the signed text, with bash in the signed directory, each signature once and before its expiry. - Runs every signed task change (priority pin or WSJF estimate) as + Runs every signed task change (priority pin, status, dependency or WSJF + estimate; a board batch is one job per change) as `python3 -I -B ` from / without a shell, after checking the - argv shape itself, and kills its process group after 30 s. Refuses task + argv shape itself, and kills its process group after 30 s; rasci.py + refuses a cycle or a done with an open key result, and its reason is the + job's output. Refuses task changes without --rasci and --council or into a council that is a symlink or another uid's, and every other job. Posts the last 200 output lines back while it runs and the exit code at its end. The exit code is 0. diff --git a/orchestrator/client/signing.py b/orchestrator/client/signing.py index 6cbce2b..720c3aa 100644 --- a/orchestrator/client/signing.py +++ b/orchestrator/client/signing.py @@ -27,6 +27,7 @@ PRIO = re.compile(r"[1-9][0-9]{0,2}") FIBONACCI = ("1", "2", "3", "5", "8", "13", "21", "34", "55", "89") ESTIMATOR = "Operator" COMMANDS = "operator/" +STATUSES = ("queued", "running", "blocked", "done", "stopped") APPROVAL_KEYS = {"v", "kind", "items", "nonce", "expires"} RULE_KEYS = {"v", "kind", "id", "kinds", "tool", "pattern", "scope", "expires"} WARNINGS = ( @@ -156,14 +157,31 @@ def task_item(task: str, argv: list[str]) -> dict: return {"kind": "task", "id": task, "argv": argv} +def board_flags(task: str, flags: list[str]) -> bool: + if flags == ["--unpin"]: + return True + if len(flags) != 2: + return False + flag, value = flags + if flag == "--prio": + return bool(PRIO.fullmatch(value)) + if flag == "--status": + return value in STATUSES + linked = flag in ("--after", "--drop-after") + return linked and value != task and bool(TASK.fullmatch(value)) and not value.startswith(COMMANDS) + + def task_argv(task: object, argv: object) -> list[str]: """Args: task: the register task the change is for, a rasci.py slug outside operator/. - argv: the rasci.py arguments, one of `set --prio <1-999>`, - `estimate Operator --value --complexity `, - `estimate Operator --value ` (the operator's business - value alone) or `estimate Operator --drop`. + argv: the rasci.py arguments, one of `set ` with exactly one + of `--prio <1-999>`, `--unpin`, `--status `, + `--after ` or `--drop-after ` (other a register + slug outside operator/), `estimate Operator --value + --complexity `, `estimate Operator --value ` + (the operator's business value alone) or + `estimate Operator --drop`. Returns: argv, the hub and the runner check it the same way. @@ -176,12 +194,11 @@ def task_argv(task: object, argv: object) -> list[str]: if not isinstance(argv, list) or not all(isinstance(part, str) for part in argv): raise Refused("the argv is no list of texts") rest = argv[3:] - pinned = argv[:3] == ["set", task, "--prio"] and len(argv) == 4 and bool(PRIO.fullmatch(argv[3])) valued = len(rest) in (2, 4) and rest[0] == "--value" and rest[1] in FIBONACCI estimated = valued and (len(rest) == 2 or (rest[2] == "--complexity" and rest[3] in FIBONACCI)) - if pinned or (argv[:3] == ["estimate", task, ESTIMATOR] and (estimated or rest == ["--drop"])): + if (argv[:2] == ["set", task] and board_flags(task, argv[2:])) or (argv[:3] == ["estimate", task, ESTIMATOR] and (estimated or rest == ["--drop"])): return argv - raise Refused("the argv is no task priority or WSJF estimate change of this task") + raise Refused("the argv is no task priority, status, dependency or WSJF estimate change of this task") def approval_payload(hashes: list[str], nonce: str, expires: int) -> dict: diff --git a/orchestrator/kvb_orchestrator/jobs.py b/orchestrator/kvb_orchestrator/jobs.py index dd4876b..771f276 100644 --- a/orchestrator/kvb_orchestrator/jobs.py +++ b/orchestrator/kvb_orchestrator/jobs.py @@ -94,4 +94,5 @@ class Jobs: runs = [self.scripts.run(OPERATOR_SCRIPT, ["done", job["task"].removeprefix(OPERATOR)])] if command_done else [] tail = "\n".join(output.splitlines()[-20:]) self.log.add({"kind": "run", "title": f"{status}: {job['command'][:200]}", "text": tail, "ref": {"job": job_id, "task": job["task"], "code": code, "status": status}}) - return {"id": job_id, "status": status, "code": code, "runs": runs} + told = self.tell_applied(job) if job.get("kind") == TASK_JOB else [] + return {"id": job_id, "status": status, "code": code, "runs": runs, "told": told} diff --git a/orchestrator/kvb_orchestrator/tasking.py b/orchestrator/kvb_orchestrator/tasking.py index 0180cba..d879262 100644 --- a/orchestrator/kvb_orchestrator/tasking.py +++ b/orchestrator/kvb_orchestrator/tasking.py @@ -1,16 +1,21 @@ from __future__ import annotations +import secrets import time +from .council import HUMAN, read_claims +from .denying import requester_of from .errors import Conflict, Missing, ToolError from .fields import decode, expect -from .jobs import TASK_JOB +from .jobs import JOB_OPEN, TASK_JOB from .register import REGISTER, read_register, status_of from .signed import signing -from .tasks import editable +from .skill_scripts import RADIO +from .tasks import editable, movable CHANGE_LIMIT = 100 CHANGE_KEYS = ("kind", "id", "argv") +TOLD = "Operator changed your tasks" def task_request(raw: object) -> bool: @@ -21,17 +26,41 @@ def shown_command(argv: list[str]) -> str: return " ".join(["rasci.py", *argv]) +def changeable(task: str, row: dict, argv: list[str]) -> bool: + return movable(task) if argv[:1] == ["set"] and argv[2:3] == ["--status"] else editable(task, row) + + +def affected(items: list[dict], tasks: dict[str, dict]) -> dict[str, list[str]]: + """Args: + items: the signed task items. + tasks: the register rows by task. + + Returns: + Per worker of a changed row, the shown commands of its changes in + item order; a row without a worker or the Operator's own tells + nobody. + """ + found: dict[str, list[str]] = {} + for item in items: + worker = tasks.get(item["id"], {}).get("worker") + if isinstance(worker, str) and worker.strip() not in ("", "-", HUMAN): + found.setdefault(worker, []).append(shown_command(item["argv"])) + return found + + class Tasking: def task_changes(self, raw: object) -> list[dict]: """Args: raw: the page's items, each {"kind": "task", "id", "argv"} with - the rasci.py arguments of one priority pin, WSJF estimate or - operator business value; a bulk change over a selection is - one item per task under one passkey assertion. + the rasci.py arguments of one priority pin, status, + dependency, WSJF estimate or operator business value; a + bulk change or a board batch is one item per change under + one passkey assertion. Returns: - The signed items, each for an open register task outside - operator/ and hotfix/ whose argv signing.task_argv() accepts. + The signed items, each for a register task outside operator/ + and hotfix/ whose argv signing.task_argv() accepts; only a + status change may reopen an ended task. """ if not isinstance(raw, list) or not 1 <= len(raw) <= CHANGE_LIMIT: raise ToolError(f"items must be a list of 1 to {CHANGE_LIMIT} task changes") @@ -48,7 +77,7 @@ class Tasking: row = tasks.get(entry["id"]) if row is None: raise Missing(entry["id"]) - if not editable(entry["id"], row): + if not changeable(entry["id"], row, argv): raise Conflict({"id": entry["id"], "status": status_of(row)}) item = signing.task_item(entry["id"], argv) if item in signed: @@ -56,6 +85,36 @@ class Tasking: signed.append(item) return signed + def tell_workers(self, items: list[dict]) -> list[str]: + """Returns: + The radio addresses that got one post listing the operator's + changes to their rows. + """ + claims = read_claims(self.council) + told = [] + for worker, commands in affected(items, read_register(self.council / REGISTER)).items(): + address = requester_of(claims, worker)[1] + if address and self.scripts.run(RADIO, ["post", "--to", address, "--session", HUMAN, f"{TOLD}: {'; '.join(commands)}"])["code"] == 0: + told.append(address) + return told + + def tell_applied(self, job: dict) -> list[str]: + """Args: + job: a task job that just ended. + + Returns: + The radio addresses told once the last job of its signed batch + ended: one post per worker listing the changes rasci.py applied + (exit 0); a change it refused, the runner refused or a job + still open tells nobody, and a batch is told once. + """ + with self.jobs.changed: + batch = sorted((other for other in self.jobs.listed() if other.get("batch") == job.get("batch")), key=lambda other: other["created_ts"]) + if job.get("batch") is None or any(other["status"] in JOB_OPEN or other.get("told") for other in batch): + return [] + self.jobs.change(batch[0]["id"], {"told": True}) + return self.tell_workers([{"id": other["task"], "argv": other["argv"]} for other in batch if other["status"] == "done"]) + def change_tasks(self, body: bytes) -> dict: payload = decode(body) expect(payload, ("items", "proof")) @@ -64,8 +123,9 @@ class Tasking: for item in items: self.operator_proven(proof, item) ids = [] + batch = secrets.token_hex(8) for item in items: - fields = {"kind": TASK_JOB, "task": item["id"], "argv": item["argv"], "command": shown_command(item["argv"]), "directory": "", "sha256": "", "status": "queued", "proof": proof} + fields = {"kind": TASK_JOB, "task": item["id"], "argv": item["argv"], "command": shown_command(item["argv"]), "directory": "", "sha256": "", "status": "queued", "proof": proof, "batch": batch} ids.append(self.jobs.add({**fields, "code": None, "output": "", "created_ts": time.time(), "started": None, "finished": None})["id"]) commands = [shown_command(item["argv"]) for item in items] self.log.add({"kind": "run", "title": f"Task change signed: {', '.join(item['id'] for item in items)}"[:200], "text": "\n".join(commands), "ref": {"jobs": ids, "status": "queued"}}) diff --git a/orchestrator/kvb_orchestrator/tasks.py b/orchestrator/kvb_orchestrator/tasks.py index 15c9af2..35fe0bd 100644 --- a/orchestrator/kvb_orchestrator/tasks.py +++ b/orchestrator/kvb_orchestrator/tasks.py @@ -28,9 +28,13 @@ def kind_of(task: str) -> str: return "hotfix" if task.startswith(HOTFIX) else "task" -def editable(task: str, row: dict) -> bool: +def movable(task: str) -> bool: shown_whole = len(task) <= TEXT_SHOWN and signing.TASK.fullmatch(task) is not None - return shown_whole and kind_of(task) == "task" and status_of(row) not in ENDED + return shown_whole and kind_of(task) == "task" + + +def editable(task: str, row: dict) -> bool: + return movable(task) and status_of(row) not in ENDED def fibonacci(value: object) -> bool: @@ -131,6 +135,7 @@ def task_row(task: str, row: dict, trees: list[dict]) -> dict: "kind": kind_of(task), "ended": ended, "editable": editable(task, row), + "movable": movable(task), "prio": prio_of(row), "estimates": estimates, "wsjf": wsjf(estimates), @@ -138,6 +143,7 @@ def task_row(task: str, row: dict, trees: list[dict]) -> dict: "points": points(estimates), "start": iso_epoch(row.get("start")), "end": iso_epoch(row.get("end")) if ended else None, + "updated": iso_epoch(row.get("updated")), "after": after_of(row), "base": shown(row.get("base"))[:TEXT_SHOWN], "roles": roles_of(row), @@ -163,6 +169,7 @@ def queue_row(entry: dict, created: float, trees: list[dict]) -> dict: "kind": "queue", "ended": False, "editable": False, + "movable": False, "prio": None, "estimates": {}, "wsjf": None, @@ -170,6 +177,7 @@ def queue_row(entry: dict, created: float, trees: list[dict]) -> dict: "points": None, "start": (epoch(entry.get("granted")) or created) if entry.get("status") == "active" else None, "end": None, + "updated": None, "after": [], "base": "-", "roles": {}, @@ -250,6 +258,24 @@ def queue_chain(queue: list[tuple[dict, float]], tasks: dict[str, dict], agents: return [(shown(entry.get("id"))[:TEXT_SHOWN], entry.get("status") == "waiting") for entry, _ in [*active, *waiting]] +def queue_line(chain: list[tuple[str, bool]], queue: list[tuple[dict, float]]) -> list[str]: + """Args: + chain: queue_chain(). + queue: the kvb-org-queue entries with their enqueue time. + + Returns: + The tasks holding a live queue entry, each once, in the grant order + of their first entry. + """ + owners = {shown(entry.get("id"))[:TEXT_SHOWN]: entry.get("task") for entry, _ in queue} + line: list[str] = [] + for job, _ in chain: + task = owners.get(job) + if isinstance(task, str) and len(task) <= TEXT_SHOWN and task not in line: + line.append(task) + return line + + def branch_of(row: dict) -> str: return row["branch"] if row["branch"] != "-" else row["worktree_branch"] diff --git a/orchestrator/kvb_orchestrator/view.py b/orchestrator/kvb_orchestrator/view.py index 9c65776..3f03ac8 100644 --- a/orchestrator/kvb_orchestrator/view.py +++ b/orchestrator/kvb_orchestrator/view.py @@ -38,7 +38,7 @@ from .reviewing import review_row from .settings import media_origins, station_origins, valid_settings from .signed import signing from .skill_scripts import OUTPUT_LIMIT -from .tasks import queue_chain, task_edges, task_rows +from .tasks import queue_chain, queue_line, task_edges, task_rows RADIO_WINDOW = 86400 JOBS_SHOWN = 20 @@ -108,6 +108,7 @@ class View: queue = queue_entries(self.council) order = stored_order(self.council / AGENT_ORDER) rows = task_rows(tasks, queue, self.tree.get()) + chain = queue_chain(queue, tasks, order, now) return { "settings": settings, "media": media_origins(settings), @@ -134,7 +135,8 @@ class View: "operator": operator_rows(tasks), "register": register_rows(tasks), "tasks": rows, - "task_edges": task_edges(rows, queue_chain(queue, tasks, order, now)), + "task_edges": task_edges(rows, chain), + "queue_line": queue_line(chain, queue), "agent_order": agent_order(tasks, claims, order), "radio": radio, "claims": claims, diff --git a/orchestrator/templates/index.html.j2 b/orchestrator/templates/index.html.j2 index f3640e2..a43c96f 100644 --- a/orchestrator/templates/index.html.j2 +++ b/orchestrator/templates/index.html.j2 @@ -412,6 +412,8 @@ pre.notes { margin: 0; padding: 8px 10px; background: var(--sunken); border-radi {% include "style-tasks.css" %} +{% include "style-kanban.css" %} + {% include "style-bubbles.css" %} @@ -462,6 +464,8 @@ pre.notes { margin: 0; padding: 8px 10px; background: var(--sunken); border-radi {% include "faces.js" %} {% include "plan.js" %} + +{% include "kanban.js" %}