feat(orchestrator): W2 Kanban board with swimlanes and forecast

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 <noreply@anthropic.com>

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}

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"}})

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"]
 

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,