From d9eb121d12882fa7b856ffb62d06b3d23cb4237f Mon Sep 17 00:00:00 2001 From: Kevin Veen-Birkenbach Date: Mon, 9 Nov 2026 06:53:55 +0000 Subject: [PATCH] feat(orchestrator): S5 Tasks with choices, Gantt, estimates and burn charts MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Requirement 001 phase 6, S5: - Tasks tab: every pending ask, rule request, operator row and choice card shows under "waits on Operator"; a choice request goes out on a fixed channel button, post and call with an agent multi-select, and a click on an option sends a passkey-signed answer the agent checks with operator_message. The unsigned /api/decisions//choose route is gone, and a decision, choice answer or choice recipient whose name holds "*" is refused, so a signed answer never reaches every agent. - Gantt view: one bar per task from start to end (running to now), one arrow per explicit, branch-base and queue-order edge, hatched forecast bars (always, points only, real only) from the median duration or points times median hours per point. - rasci.py set --after/--drop-after stores task dependencies and refuses unknown tasks and cycles. - Estimation toggle (No estimate default, Story points) is stored in the hub settings; burndown and burnup per whole register, list filter or one agent or line, counting tasks or summing complexity with the unestimated tasks as their own line. - New skill kvb-org-queue-estimate: "Request estimates" (story points mode, passkey-signed board action) or the skill sends exactly one newsletter listing every unestimated task with its RASCI parties; a party name that is no plain name shows as "unnamed party". - Business value per task or in bulk is stored only with a verified passkey assertion over the exact ids and values; WSJF uses it instead of the median. The task list sorts by every decision-50 mode both ways. - e2e: the walk's pending ask lives an hour instead of 60 s, and the CDP and BiDi drivers retry removing the browser profile while the browser exits (ENOTEMPTY). Evidence: make test 709 OK, ruff and markdownlint clean (/tmp/s5-make-test-fix-2.log); e2e 26 OK in Chromium and Firefox (/tmp/s5-e2e-fix-2.log); MEDUSA 50 files, an independent skeptic rejected four holes (unsigned choose route, unsigned estimates request, free-text party names, "*" recipients incl. normalizer bypasses), all fixed and re-verified, the 8 remaining pattern hits accepted, rescan 0 open. 🎯 Fix: 92 % 💥 Break: 4 % 💸 Cost: 100 % ⏱️ Time: – 🛡️ Security: 100 % 🏆 Score: 88 🚦 Gate: ✅ 90/5, cost and time <= 100, security 100 % Co-Authored-By: Claude Opus 5.5 --- README.md | 6 +- docs/requirements/001-agent-orchestrator.md | 26 +- orchestrator/client/signing.py | 8 +- orchestrator/kvb_orchestrator/approving.py | 13 +- orchestrator/kvb_orchestrator/asking.py | 29 +- orchestrator/kvb_orchestrator/choosing.py | 100 ++++ orchestrator/kvb_orchestrator/handler.py | 10 +- orchestrator/kvb_orchestrator/hub.py | 3 +- orchestrator/kvb_orchestrator/mcp.py | 14 +- orchestrator/kvb_orchestrator/messaging.py | 41 +- orchestrator/kvb_orchestrator/settings.py | 5 + .../kvb_orchestrator/skill_scripts.py | 1 + orchestrator/kvb_orchestrator/tasking.py | 8 +- orchestrator/kvb_orchestrator/tasks.py | 128 ++++- orchestrator/kvb_orchestrator/view.py | 10 +- orchestrator/templates/index.html.j2 | 6 + orchestrator/templates/nav.js | 4 +- orchestrator/templates/page-choices.js | 72 +++ orchestrator/templates/page-constants.js | 2 +- orchestrator/templates/page-inbox.js | 6 +- orchestrator/templates/page-screens.js | 3 +- orchestrator/templates/page-settings.js | 1 + orchestrator/templates/page-tasks-gantt.js | 216 +++++++++ orchestrator/templates/page-tasks-prio.js | 42 +- orchestrator/templates/page-tasks.js | 30 +- orchestrator/templates/plan.js | 176 +++++++ orchestrator/templates/style-tasks.css | 35 ++ skills/kvb-org-queue-estimate/SKILL.md | 24 + .../scripts/estimate.py | 94 ++++ skills/kvb-org-rasci/SKILL.md | 8 +- skills/kvb-org-rasci/scripts/rasci.py | 51 +- skills/kvb-org-wsjf/SKILL.md | 2 +- skills/kvb-think-choice/SKILL.md | 7 + tests/e2e/bidi.mjs | 2 +- tests/e2e/cdp.mjs | 2 +- tests/e2e/test_orchestrator_e2e.py | 449 +++++++++++++++++- tests/hub_support.py | 2 + tests/test_estimate.py | 86 ++++ tests/test_hub_asks.py | 22 +- tests/test_hub_board.py | 2 +- tests/test_hub_choices.py | 173 +++++++ tests/test_hub_inbox.py | 7 +- tests/test_hub_nav.py | 10 +- tests/test_hub_page.py | 8 +- tests/test_hub_plan.py | 150 ++++++ tests/test_hub_settings.py | 16 + tests/test_hub_state.py | 2 +- tests/test_hub_tasks.py | 149 +++++- tests/test_rasci.py | 49 ++ 49 files changed, 2162 insertions(+), 148 deletions(-) create mode 100644 orchestrator/kvb_orchestrator/choosing.py create mode 100644 orchestrator/templates/page-choices.js create mode 100644 orchestrator/templates/page-tasks-gantt.js create mode 100644 orchestrator/templates/plan.js create mode 100644 skills/kvb-org-queue-estimate/SKILL.md create mode 100755 skills/kvb-org-queue-estimate/scripts/estimate.py create mode 100644 tests/test_estimate.py create mode 100644 tests/test_hub_choices.py create mode 100644 tests/test_hub_plan.py diff --git a/README.md b/README.md index 623e55f..6481e8a 100644 --- a/README.md +++ b/README.md @@ -87,6 +87,7 @@ flowchart LR think --> think_choice[kvb-think-choice] org --> org_radio[kvb-org-radio] org --> org_queue[kvb-org-queue] + org --> org_queue_estimate[kvb-org-queue-estimate] org --> org_update[kvb-org-update] agent --> agent_optimize[kvb-agent-optimize] agent --> agent_drain[kvb-agent-drain] @@ -194,6 +195,7 @@ flowchart LR org_resume[kvb-org-resume] radio[kvb-org-radio] queue[kvb-org-queue] + queue_estimate[kvb-org-queue-estimate] org_update[kvb-org-update] end subgraph cat_agent [agent] @@ -411,11 +413,11 @@ The hub page loads nothing from the network: its icons come from Bootstrap Icons The Agents view "Hierarchy" draws the council's formal offices as a collapsible org chart: Operator, chair, dimensions with their line representatives, organisers (`cluster/`) and presidents (`work/`); the chair in every graph is the holder of the `chair` office, not a fixed name. Appointing, handing over or recalling an office there asks for the passkey signature; the hub writes a job (office, signed holder, successor session, work; no command) into `~/.local/state/kvb-orchestrator/offices/jobs`, and the `kvb-orchestrator-offices.path` user unit runs `client/offices.py` on the host, which verifies the signature, refuses a council directory others can write, a link on the path or a holder that changed since the signature. For a takeover or recall it moves the office aside and removes it only while the claim there still names the signed holder; an appoint or takeover then runs a plain `council.py claim ` for the successor by argv. council.py itself has no flag that overrides a live holder. The same view holds the operator's agent order as a drag list with up and down buttons: `rasci.py agents order` stores it in `agent-order.json` beside the register, where it breaks ties of equal task priority and orders the chair's grants without touching any task's `--prio` pin. -The Tasks tab (key 4, `#tasks/list`, `#tasks/board`, `#tasks/rasci`) shows the `kvb-org-rasci` register as a list, a board by status and a RASCI matrix; queue jobs from `council/queue.json` appear as "waiting for a grant", and the filter "Waiting for operator" is the inbox, which the Inbox entry under Stream opens. A task's sheet pins its priority (`rasci.py set --prio `) or sets and drops the operator's WSJF estimate (`rasci.py estimate Operator --value --complexity ` or `--drop`) after a passkey signature; the hub writes an argv-only job, and the runner, started with `--rasci --council `, verifies it like an operator command and runs `rasci.py` from `/` without a shell, killing it after 30 s. Agent order stays untouched. The install copies the fresh runner unit and restarts the runner when it is enabled, and only while the installed `council.py`, `nofollow.py` and `rasci.py` match this tree: otherwise it disables the runner and asks for `make install`. The runner unit pins `PATH` to `/usr/bin:/bin`, so a signed command names any other program by its full path; a runner started without `--rasci` and `--council` keeps running signed commands and refuses task changes, and so does one whose council is a symlink or another uid's. Each worktree in the tree file may carry `"flow": {"pull": [...], "push": [...], "source": "file"}` or `null`, which the Tree, the Chord edges, the detail panel and the matching task show; a malformed flow shows as `invalid`. Worktrees and tasks with open work (uncommitted, unpushed, unmerged, task not done) carry a warning bar. +The Tasks tab (key 4, `#tasks/list`, `#tasks/board`, `#tasks/rasci`) shows the `kvb-org-rasci` register as a list, a board by status and a RASCI matrix; queue jobs from `council/queue.json` appear as "waiting for a grant", and the filter "Waits on Operator" is the inbox, which the Inbox entry under Stream opens: every pending review, ask, rule request, operator row and choice card, with no Choices column of its own. "Request choices" has one button per channel, post and call to the agents ticked under "Agents", newsletter, broadcast, pan-pan, Sécurité and mayday to all, that sends the fixed request "send me your open decisions as choices" without a text field, one passkey assertion over every recipient (`/api/choices/sign`, then `/api/choices`). Agents answer with `decision_request`, shown as choice cards, and a click on an option is the passkey-signed answer: after the 3 s undo the hub verifies the signature and posts the answer to the asking agent, whose `decision_wait` returns it as `message` and `answer` for `operator_message`. A task's sheet pins its priority (`rasci.py set --prio `) or sets and drops the operator's WSJF estimate (`rasci.py estimate Operator --value --complexity ` or `--drop`) after a passkey signature; the hub writes an argv-only job, and the runner, started with `--rasci --council `, verifies it like an operator command and runs `rasci.py` from `/` without a shell, killing it after 30 s. Agent order stays untouched. The install copies the fresh runner unit and restarts the runner when it is enabled, and only while the installed `council.py`, `nofollow.py` and `rasci.py` match this tree: otherwise it disables the runner and asks for `make install`. The runner unit pins `PATH` to `/usr/bin:/bin`, so a signed command names any other program by its full path; a runner started without `--rasci` and `--council` keeps running signed commands and refuses task changes, and so does one whose council is a symlink or another uid's. Each worktree in the tree file may carry `"flow": {"pull": [...], "push": [...], "source": "file"}` or `null`, which the Tree, the Chord edges, the detail panel and the matching task show; a malformed flow shows as `invalid`. Worktrees and tasks with open work (uncommitted, unpushed, unmerged, task not done) carry a warning bar. Without reinstalling the skills, `make orchestrator-install` installs, builds and starts only the hub, `make orchestrator-check` checks the installed hub (image, container, logs, HTTP, pinned keys, hook, runner, tree timer, disk; exit 1 on a failure), `make orchestrator-pin CREDENTIAL=` pins the passkey that "Register passkey" shows, `make orchestrator-runner` enables the runner service, and `make orchestrator-hook` registers the PermissionRequest hook in `~/.claude/settings.json`, replacing an earlier Agent Orchestrator entry and keeping the file mode, after a copy to `settings.json.before-kvb-orchestrator-hook` (Claude Code reads `~/.claude/settings.local.json` only in sessions started in the home directory). Agents that reach the Docker socket are root-equivalent: until the agent VMs of core requirement 093 isolate them, the signatures do not hold against such an agent. `make watch-check` shows whether the resource watch hook is stuck (running hooks, holders of the kept queue history, age of the stored stage). -For approvals the hub is only a relay: every approval of an ask, a staged commit or an operator command is signed end to end with a WebAuthn passkey held in KeePassXC (through KeePassXC-Browser, relying party `localhost`, origin `http://localhost:`; ES256, EdDSA and RS256, so a hardware key can be pinned beside it). To register, open the hub at `http://localhost:/` (WebAuthn needs the domain; on `127.0.0.1` the page links there), choose "Register passkey" in the settings and run the host command the page shows, `python3 ~/.local/share/kvb-orchestrator/client/approval.py pin --port --approvers ~/.config/kvb-orchestrator/approvers.json --credential ''`, in a terminal outside every sandbox: it pins the key in `~/.config/kvb-orchestrator/approvers.json` with mode 0600, a file no sandboxed agent can write and the hub mounts read-only; the hub never pins. Each approval, or one bulk selection, is one signature over the hashes of every item, a nonce and an expiry; each auto-approval rule is signed once. In the inbox, an operator command run with "Run" shows the runner's output in a console under its row until "Close". "Deny" on an ask, an operator row or a rule request signs the denial with its channel and optional reason (a hash no approval shares) and, after the 3 s undo, tells the requesting agent by post or by the radio class picked, never mayday; the hub verifies that signature against the pinned keys before it stores or sends anything, the hook passes the reason on only when it verifies the signature too, and a denied operator row ends as `stopped`. Agents ask for an auto-approval rule with the MCP tool `rule_request` (the rules form's limits; a request stays open for the ask lifetime of one hour, at most 20 in all, and 3 per agent name, which only spreads the queue since MCP names no caller); its row shows a dry run over the last 7 days, and only "Accept" with the passkey stores it. The hub stores no rule, of any kind, whose passkey assertion it cannot verify. The executors verify it themselves against the pinned keys and fail closed: the runner (`client/runner.py`, the `kvb-orchestrator-runner` user service) runs only signed commands, each signature once, recorded in `~/.local/state/kvb-orchestrator-approvals/runner.json`; the PermissionRequest hook (`client/permission_hook.py --port --timeout --approvers ~/.config/kvb-orchestrator/approvers.json --replay ~/.local/state/kvb-orchestrator-approvals/hook.json`) allows only a signed item that matches its own tool, command, cwd and session, else it leaves the decision to Claude Code's own prompt; the commit skills run `approval.py verify` on the staged tree before committing, else they ask in the terminal. With no key pinned, every executor refuses; decisions and review rejections stay unsigned. A signed rule never covers a command holding `;`, `&`, `|`, a backtick, `$`, `<`, `>`, parentheses or a line break: such a command always asks. The hub page is trusted while it asks for a signature or shows the blob to pin: whoever controls the hub container can show one item and have another signed, or hand over a key of its own. Any local process, sandboxed agents included, can also send the unsigned operator actions (decision choices, the priority order, settings that add no stream origin, review rejections, the bare ask denial `/api/asks//deny`, console dismissals) with the hub's guard headers, and the container reads the host `/tmp` for the review pages in `/tmp/kvb-git-diff`. All three close with the agent VMs of core requirement 093, like the Docker-socket gap. +For approvals the hub is only a relay: every approval of an ask, a staged commit or an operator command is signed end to end with a WebAuthn passkey held in KeePassXC (through KeePassXC-Browser, relying party `localhost`, origin `http://localhost:`; ES256, EdDSA and RS256, so a hardware key can be pinned beside it). To register, open the hub at `http://localhost:/` (WebAuthn needs the domain; on `127.0.0.1` the page links there), choose "Register passkey" in the settings and run the host command the page shows, `python3 ~/.local/share/kvb-orchestrator/client/approval.py pin --port --approvers ~/.config/kvb-orchestrator/approvers.json --credential ''`, in a terminal outside every sandbox: it pins the key in `~/.config/kvb-orchestrator/approvers.json` with mode 0600, a file no sandboxed agent can write and the hub mounts read-only; the hub never pins. Each approval, or one bulk selection, is one signature over the hashes of every item, a nonce and an expiry; each auto-approval rule is signed once. In the inbox, an operator command run with "Run" shows the runner's output in a console under its row until "Close". "Deny" on an ask, an operator row or a rule request signs the denial with its channel and optional reason (a hash no approval shares) and, after the 3 s undo, tells the requesting agent by post or by the radio class picked, never mayday; the hub verifies that signature against the pinned keys before it stores or sends anything, the hook passes the reason on only when it verifies the signature too, and a denied operator row ends as `stopped`. Agents ask for an auto-approval rule with the MCP tool `rule_request` (the rules form's limits; a request stays open for the ask lifetime of one hour, at most 20 in all, and 3 per agent name, which only spreads the queue since MCP names no caller); its row shows a dry run over the last 7 days, and only "Accept" with the passkey stores it. The hub stores no rule, of any kind, whose passkey assertion it cannot verify. The executors verify it themselves against the pinned keys and fail closed: the runner (`client/runner.py`, the `kvb-orchestrator-runner` user service) runs only signed commands, each signature once, recorded in `~/.local/state/kvb-orchestrator-approvals/runner.json`; the PermissionRequest hook (`client/permission_hook.py --port --timeout --approvers ~/.config/kvb-orchestrator/approvers.json --replay ~/.local/state/kvb-orchestrator-approvals/hook.json`) allows only a signed item that matches its own tool, command, cwd and session, else it leaves the decision to Claude Code's own prompt; the commit skills run `approval.py verify` on the staged tree before committing, else they ask in the terminal. With no key pinned, every executor refuses; review rejections and a decision a rule takes stay unsigned. A signed rule never covers a command holding `;`, `&`, `|`, a backtick, `$`, `<`, `>`, parentheses or a line break: such a command always asks. The hub page is trusted while it asks for a signature or shows the blob to pin: whoever controls the hub container can show one item and have another signed, or hand over a key of its own. Any local process, sandboxed agents included, can also send the unsigned operator actions (the bare decision choice `/api/decisions//choose`, which sends no answer, the priority order, settings that add no stream origin, review rejections, the bare ask denial `/api/asks//deny`, console dismissals) with the hub's guard headers, and the container reads the host `/tmp` for the review pages in `/tmp/kvb-git-diff`. All three close with the agent VMs of core requirement 093, like the Docker-socket gap. Restart your agent afterwards so it loads the new skills. diff --git a/docs/requirements/001-agent-orchestrator.md b/docs/requirements/001-agent-orchestrator.md index ce9308e..d7a0c3a 100644 --- a/docs/requirements/001-agent-orchestrator.md +++ b/docs/requirements/001-agent-orchestrator.md @@ -619,11 +619,11 @@ flowchart TD ### Phase 6: Tasks with asks and choices (S5) -- [ ] Every pending ask, rule request, operator row and choice card shows in +- [x] Every pending ask, rule request, operator row and choice card shows in the Tasks tab under "waits on Operator"; no Choices column exists. -- [ ] Each channel button sends the fixed choice request with no text field; +- [x] Each channel button sends the fixed choice request with no text field; `post` and `call` take a multi-select of agents. -- [ ] A click on a choice option sends a passkey-signed answer the agent can +- [x] A click on a choice option sends a passkey-signed answer the agent can check with `operator_message`. ### Phase 6: Firewall (S6) @@ -656,31 +656,31 @@ flowchart TD ### Phase 6: Gantt, estimates, burndown and burnup (S5) -- [ ] `rasci.py set --after ` stores the dependency and +- [x] `rasci.py set --after ` stores the dependency and `--drop-after ` removes it; an unknown task and a cycle are refused. -- [ ] The Gantt view draws one bar per task from start to end, a running +- [x] The Gantt view draws one bar per task from start to end, a running task to now, and one arrow per explicit, branch-base and queue-order edge of a fixture register and queue. -- [ ] The forecast selector offers `always` (default), `points only` and +- [x] The forecast selector offers `always` (default), `points only` and `real only`; a hatched bar is as long as the median duration of done tasks (no estimate) or points times median hours per point (story points). -- [ ] The estimation toggle defaults to `No estimate`, is stored in the hub +- [x] The estimation toggle defaults to `No estimate`, is stored in the hub settings and survives a reload. -- [ ] Burndown and burnup offer the scopes whole register (default), list +- [x] Burndown and burnup offer the scopes whole register (default), list filter within the time window, and one agent or line; with no estimate they count tasks, with story points they sum complexity and draw the unestimated tasks as a line of their own. -- [ ] "Request estimates" shows only in story points mode; a click or +- [x] "Request estimates" shows only in story points mode; a click or `/kvb-org-queue-estimate` sends exactly one newsletter to all agents that lists every unestimated task with every party holding a RASCI role in it; a state build or a mode switch sends none. -- [ ] A per-task or bulk business value change is stored only with a +- [x] A per-task or bulk business value change is stored only with a verified passkey assertion over the exact task ids and values; a forged, replayed or altered batch changes nothing, and WSJF uses the operator value instead of the median. -- [ ] The task list sorts by each mode of decision 50 in both directions. +- [x] The task list sorts by each mode of decision 50 in both directions. ### Tests @@ -700,7 +700,7 @@ flowchart TD rename-once timeline, squelch scheduler, face generator, settings writer (forged, replayed, unlisted file), token sums, German text scan. - - [ ] S5 Gantt and estimates: dependency edges and cycle refusal, forecast lengths, burn + - [x] S5 Gantt and estimates: dependency edges and cycle refusal, forecast lengths, burn sums per mode and scope, newsletter body and its single send, bulk business value signature (forged, replayed, altered), sort orders. - **Integration** @@ -725,6 +725,6 @@ flowchart TD - [ ] Phase 6: no page scrollbar on any view, agent and node popups, choice request and signed answer, Firewall settings diff and signature, Resources charts, each at 1920, 1366 and 390 px. - - [ ] S5: Gantt with arrows, burndown and burnup, the estimation toggle, + - [x] S5: Gantt with arrows, burndown and burnup, the estimation toggle, "Request estimates" and a signed bulk business value change, each at 1920, 1366 and 390 px. diff --git a/orchestrator/client/signing.py b/orchestrator/client/signing.py index 260fd44..5579537 100644 --- a/orchestrator/client/signing.py +++ b/orchestrator/client/signing.py @@ -152,8 +152,9 @@ def task_argv(task: object, argv: object) -> list[str]: 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 ` or - `estimate Operator --drop`. + `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. @@ -167,7 +168,8 @@ def task_argv(task: object, argv: object) -> list[str]: 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])) - estimated = len(rest) == 4 and rest[0] == "--value" and rest[1] in FIBONACCI and rest[2] == "--complexity" and rest[3] in FIBONACCI + 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"])): return argv raise Refused("the argv is no task priority or WSJF estimate change of this task") diff --git a/orchestrator/kvb_orchestrator/approving.py b/orchestrator/kvb_orchestrator/approving.py index 68ccf7e..64dc5ca 100644 --- a/orchestrator/kvb_orchestrator/approving.py +++ b/orchestrator/kvb_orchestrator/approving.py @@ -4,6 +4,7 @@ import secrets import time from .asking import ask_state +from .choosing import answer_signed from .cleanup import cleanup_request from .denying import DENIABLE, DENY_KEYS, denial_note, request_status, request_words from .errors import Conflict, Missing, ToolError @@ -35,7 +36,8 @@ def inbox_item(raw: dict, label: str, title: str, gate: str, subject: dict | Non label: the chosen option of a decision, else empty. title: what the stream logs for the item. gate: the review gate, else empty. - subject: what an approval signs; None for a decision. + subject: what an approval signs; for a decision the answer post to + the asking agent. requester: the requesting agent's name for a denial, else empty. Returns: @@ -105,14 +107,15 @@ class Approving: Returns: The proof to relay, None when nothing needs a signature. The hub verifies the signature of a denial itself, since it acts on it; - of an approval it checks the shape, and the executors verify. + of an approval it checks the shape, and the executors verify, + the hub itself for a choice answer when it sends it. """ if not signed: if proof is not None: raise ToolError("proof must be null when no item needs a signature") return None if not isinstance(proof, dict): - raise ToolError("not signed: an approval of an ask, review or operator command needs the passkey signature") + raise ToolError("not signed: an approval of an ask, review, choice or operator command needs the passkey signature") expect(proof, ("payload", "assertion")) try: payload = signing.checked_approval(proof["payload"], now) @@ -162,6 +165,8 @@ class Approving: record, changed = self.asks.decide(item["id"], {"status": "allowed" if approve else "denied", "proof": proof}) elif item["kind"] == "review": record, changed = self.reviews.decide(item["id"], {"status": "approved" if approve else "rejected", "proof": proof}) + elif item["signed"] is not None: + return self.answer_choice(item, proof) else: record, changed = self.decisions.decide(item["id"], {"status": "decided", "choice": item["label"]}) if record is None: @@ -218,7 +223,7 @@ class Approving: if kind_name == "rule": return inbox_item(raw, "", f"Rule {request_words(record)}", "", request_signed(record), self.requester(raw, record["agent"])) title = {"review": record.get("title"), "ask": f"{record.get('tool')} {record.get('command')}", "decision": record.get("question")}[kind_name] - subject = {"review": review_signed, "ask": ask_signed}[kind_name](record) if kind_name != "decision" else None + subject = {"review": review_signed, "ask": ask_signed}[kind_name](record) if kind_name != "decision" else answer_signed(record, label) return inbox_item(raw, label, title, record.get("gate", ""), subject, self.requester(raw, record["session"] if kind_name == "ask" else "")) def checked(self, raw: object) -> list[dict]: diff --git a/orchestrator/kvb_orchestrator/asking.py b/orchestrator/kvb_orchestrator/asking.py index 8128294..49ec1c5 100644 --- a/orchestrator/kvb_orchestrator/asking.py +++ b/orchestrator/kvb_orchestrator/asking.py @@ -2,6 +2,7 @@ from __future__ import annotations import time +from .choosing import EVERYONE, answer_text from .errors import Conflict, Missing, ToolError from .fields import ( COMMAND_LIMIT, @@ -35,7 +36,13 @@ def ask_row(ask: dict, names: dict[str, str]) -> dict: def decision_state(decision: dict) -> dict: - return {key: decision[key] for key in ("id", "status", "choice", "decided")} + """Returns: + The status and choice, with the id and text of the operator's + passkey-signed answer when a click chose it, else None for both. + """ + message = decision.get("message") + answer = None if message is None else answer_text(decision, decision["choice"]) + return {**{key: decision[key] for key in ("id", "status", "choice", "decided")}, "message": message, "answer": answer} def decision_row(decision: dict) -> dict: @@ -87,20 +94,7 @@ class Asking: if not changed: raise Conflict(ask_state(decided, time.time())) return ask_state(decided, time.time()) - if kind_name != "decisions" or verb != "choose": - raise Missing(verb) - payload = decode(body) - expect(payload, ("label",)) - label = field(payload, "label") - record = self.decisions.get(item_id) - if record is None: - raise Missing(item_id) - if label not in [option["label"] for option in record["options"]]: - raise ToolError(f"label must be one of the options of decision {item_id}") - decided, changed = self.decisions.decide(item_id, {"status": "decided", "choice": label}) - if not changed: - raise Conflict(decision_state(decided)) - return decision_state(decided) + raise Missing(verb) def decision_request(self, arguments: dict) -> dict: expect(arguments, ("question", "options", "agent", "recommended")) @@ -114,7 +108,10 @@ class Asking: recommended = arguments["recommended"] if not isinstance(recommended, str) or recommended.strip() not in ["", *(option["label"] for option in parsed)]: raise ToolError("recommended must be one of the option labels or an empty string") - fields = {"question": question, "options": parsed, "agent": field(arguments, "agent").strip(), "recommended": recommended.strip()} + asker = field(arguments, "agent").strip() + if EVERYONE in asker: + raise ToolError(f"agent names the asking agent, never a name with {EVERYONE}") + fields = {"question": question, "options": parsed, "agent": asker, "recommended": recommended.strip()} record = self.decisions.add({**fields, "status": "pending", "choice": None, "decided": None, "created_ts": time.time()}) self.automate("decision", record) return {"id": record["id"], "status": record["status"], "url": self.url()} diff --git a/orchestrator/kvb_orchestrator/choosing.py b/orchestrator/kvb_orchestrator/choosing.py new file mode 100644 index 0000000..a5b1038 --- /dev/null +++ b/orchestrator/kvb_orchestrator/choosing.py @@ -0,0 +1,100 @@ +from __future__ import annotations + +from .council import agent +from .errors import ToolError +from .fields import decode, expect, field +from .messaging import DIRECT, EVERYONE, TO_ALL +from .signed import signing + +CHOICE_REQUEST = "send me your open decisions as choices" +CHANNELS = {**DIRECT, **TO_ALL} +CHOICE_KEYS = ("class", "to") +RECIPIENT_LIMIT = 40 +ANSWER_CLASS = "post" +QUESTION_SHOWN = 200 + + +def answer_text(decision: dict, label: str) -> str: + question = " ".join(decision["question"].split())[:QUESTION_SHOWN] + return f"Choice for decision {decision['id']} ({question}): {label}" + + +def answer_signed(decision: dict, label: str) -> dict: + """Returns: + The post to the asking agent that a click on the option label + signs: the operator's answer operator_message proves. + """ + if EVERYONE in decision["agent"]: + raise ToolError(f"a choice answer goes to the asking agent, never to a name with {EVERYONE}") + return signing.message_item(agent(decision["agent"]), ANSWER_CLASS, answer_text(decision, label), "", "", "") + + +def recipients(payload: dict) -> list[str]: + """Args: + payload: the page's class and to: 1 to RECIPIENT_LIMIT distinct + agent names for post and call, [EVERYONE] for a class to all. + """ + cls, names = payload["class"], payload["to"] + if not isinstance(cls, str) or cls not in CHANNELS: + raise ToolError(f"class must be one of {', '.join(CHANNELS)}") + if not isinstance(names, list) or not all(isinstance(name, str) for name in names): + raise ToolError("to must be a list of agent names") + if cls in TO_ALL: + if names != [EVERYONE]: + raise ToolError(f"a choice request to all names to [{EVERYONE}]") + return names + targets = [field({"to": name}, "to").strip() for name in names] + if not 1 <= len(targets) <= RECIPIENT_LIMIT or any(EVERYONE in target for target in targets): + raise ToolError(f"a post or a call names 1 to {RECIPIENT_LIMIT} agents, never a name with {EVERYONE}") + if len({agent(target) for target in targets}) != len(targets): + raise ToolError("to must name distinct agents") + return targets + + +class Choosing: + def choice_items(self, payload: dict) -> list[dict]: + return [signing.message_item(target, payload["class"], CHOICE_REQUEST, "", "", "") for target in recipients(payload)] + + def sign_choices(self, body: bytes) -> dict: + payload = decode(body) + expect(payload, CHOICE_KEYS) + return self.challenge_for(self.choice_items(payload)) + + def request_choices(self, body: bytes) -> dict: + """Returns: + The ids of the stored requests, one per recipient, once one + passkey assertion signed each of them; nothing is stored or sent + before every item verified. + """ + payload = decode(body) + expect(payload, (*CHOICE_KEYS, "proof")) + items = self.choice_items(payload) + for item in items: + self.operator_proven(payload["proof"], item) + records = [self.stored_message(item, payload["proof"], None) for item in items] + runs = [self.radioed(record, CHANNELS[payload["class"]]) for record in records] + return {"ids": [record["id"] for record in records], "ok": all(run["code"] == 0 for run in runs), "runs": runs} + + def answer_choice(self, item: dict, proof: dict) -> str: + """Args: + item: a checked decision item whose signed is answer_signed(). + proof: the relayed signature; the hub sends the answer itself, + so it verifies the signature here, once. + + Returns: + The outcome words for the stream. + """ + record = self.decisions.get(item["id"]) + if record is None: + return "unknown" + if record["status"] != "pending": + return f"already {record['status']}" + self.operator_proven(proof, item["signed"]) + message = self.stored_message(item["signed"], proof, None) + decided, changed = self.decisions.decide(item["id"], {"status": "decided", "choice": item["label"], "message": message["id"]}) + if not changed: + self.messages.remove(message["id"]) + return f"already {decided['status']}" + run = self.radioed(message, CHANNELS[ANSWER_CLASS]) + where = f"{ANSWER_CLASS} to {message['address']}" + return f"decided · {where}" if run["code"] == 0 else f"decided · {where} failed ({run['code']})" diff --git a/orchestrator/kvb_orchestrator/handler.py b/orchestrator/kvb_orchestrator/handler.py index 40abc94..c272289 100644 --- a/orchestrator/kvb_orchestrator/handler.py +++ b/orchestrator/kvb_orchestrator/handler.py @@ -17,7 +17,7 @@ TTS_ROUTE = "/api/tts" WAV_TYPE = "audio/wav" STT_ROUTE = "/api/stt" ROUTE = re.compile(r"/reviews/([0-9a-f]{16})/(page|files|approve|reject)") -ITEM = re.compile(r"/api/(asks|decisions)/([0-9a-f]{16})(?:/(allow|deny|choose))?") +ITEM = re.compile(r"/api/(asks|decisions)/([0-9a-f]{16})(?:/(allow|deny))?") ENTRY = re.compile(r"/api/stream/([0-9]{1,12})") BOARD = re.compile(r"/api/board/([a-z-]{1,32})") UPLOADS = "/api/uploads" @@ -48,8 +48,10 @@ VOICE_ITEM = re.compile(r"/api/voice/([0-9a-f]{16})(\.wav)?") MESSAGE_SIGN_ROUTE = "/api/messages/sign" BOARD_SIGN_ROUTE = "/api/board/sign" SETTINGS_SIGN_ROUTE = "/api/settings/sign" +CHOICES_ROUTE = "/api/choices" +CHOICES_SIGN_ROUTE = "/api/choices/sign" WAV_POSTS = (STT_ROUTE, VOICE_ROUTE) -ACTION_POSTS = ("/api/rules", "/api/nl", SETTINGS_ROUTE, TTS_ROUTE, STT_ROUTE, SIGN_ROUTE, RULE_SIGN_ROUTE, CLEANUP_ROUTE, OFFICES_ROUTE, TASKS_ROUTE, VOICE_ROUTE, MESSAGE_SIGN_ROUTE, SETTINGS_SIGN_ROUTE) +ACTION_POSTS = ("/api/rules", "/api/nl", SETTINGS_ROUTE, TTS_ROUTE, STT_ROUTE, SIGN_ROUTE, RULE_SIGN_ROUTE, CLEANUP_ROUTE, OFFICES_ROUTE, TASKS_ROUTE, VOICE_ROUTE, MESSAGE_SIGN_ROUTE, SETTINGS_SIGN_ROUTE, CHOICES_ROUTE, CHOICES_SIGN_ROUTE) class Handler(Exchange): @@ -182,6 +184,10 @@ class Handler(Exchange): return lambda body: self.answer(lambda: hub.voice_notes.submit(body), HTTPStatus.ACCEPTED) if path == MESSAGE_SIGN_ROUTE: return lambda body: self.answer(lambda: hub.sign_message(body), HTTPStatus.OK) + if path == CHOICES_SIGN_ROUTE: + return lambda body: self.answer(lambda: hub.sign_choices(body), HTTPStatus.OK) + if path == CHOICES_ROUTE: + return lambda body: self.answer(lambda: hub.request_choices(body), HTTPStatus.CREATED) return None def posted(self, path: str) -> bool: diff --git a/orchestrator/kvb_orchestrator/hub.py b/orchestrator/kvb_orchestrator/hub.py index 4ed47a1..e9d49af 100644 --- a/orchestrator/kvb_orchestrator/hub.py +++ b/orchestrator/kvb_orchestrator/hub.py @@ -11,6 +11,7 @@ from .approving import Approving from .artifacts import Touched from .asking import Asking from .batches import Pending +from .choosing import Choosing from .cleanup import Cleaning, Spool from .denying import Denying from .engines import Engines, Speech @@ -62,7 +63,7 @@ def environment(templates: Path) -> jinja2.Environment: @dataclass(frozen=True) -class Hub(Mcp, Reviewing, Asking, Messaging, Approving, Denying, Requesting, Cleaning, Offices, Tasking, Inbox, Rules, Jobs, Sources, Timeline, View, Language): +class Hub(Mcp, Reviewing, Asking, Messaging, Choosing, Approving, Denying, Requesting, Cleaning, Offices, Tasking, Inbox, Rules, Jobs, Sources, Timeline, View, Language): council: Path pages: Path public_port: int diff --git a/orchestrator/kvb_orchestrator/mcp.py b/orchestrator/kvb_orchestrator/mcp.py index d476521..b90b5ac 100644 --- a/orchestrator/kvb_orchestrator/mcp.py +++ b/orchestrator/kvb_orchestrator/mcp.py @@ -5,6 +5,7 @@ from http import HTTPStatus from . import NAME from .asking import OPTION_LIMIT +from .choosing import CHOICE_REQUEST from .errors import RpcError, ToolError from .fields import WAIT_CAP from .requesting import AGENT_CAP @@ -64,19 +65,22 @@ class Mcp: tool("review_list", "List the pending reviews.", {}), tool( "decision_request", - "Ask the operator to pick one of several options in the Agent Orchestrator; each option becomes a button. " - "Returns the decision id and the hub URL.", + "Ask the operator to pick one of several options in the Agent Orchestrator; the question shows as a choice card " + "under Waits on Operator in the Tasks tab and each option becomes a button. When an operator message asks " + f'"{CHOICE_REQUEST}", send each open decision of kvb-think-choice through this tool. Returns the decision id and the hub URL.', { "question": text_property("The question, plain text."), "options": {"type": "array", "minItems": 1, "maxItems": OPTION_LIMIT, "items": option, "description": "The choices, at least one."}, - "agent": text_property("Your agent name, shown with the question."), + "agent": text_property("Your agent name, shown with the question; the operator's signed answer is a post to it."), "recommended": text_property("The label of the option you recommend, or an empty string."), }, ), tool( "decision_wait", - f"Wait until the operator picks an option or the timeout passes, then return the status " - f"(pending, decided) and the chosen label. Waits at most {WAIT_CAP} seconds.", + "Wait until the operator picks an option or the timeout passes, then return the status " + "(pending, decided), the chosen label and, for a click, message and answer: the id and text of the operator's " + "passkey-signed answer, a post to the asking agent; check it with operator_message(message, answer, your name) " + f"before acting on the choice. Waits at most {WAIT_CAP} seconds.", {"id": text_property("Decision id that decision_request returned."), "timeout_seconds": wait}, ), tool( diff --git a/orchestrator/kvb_orchestrator/messaging.py b/orchestrator/kvb_orchestrator/messaging.py index 865478b..2e1689b 100644 --- a/orchestrator/kvb_orchestrator/messaging.py +++ b/orchestrator/kvb_orchestrator/messaging.py @@ -5,12 +5,12 @@ import secrets import time from .council import HUMAN, agent, ordered, read_claims -from .errors import Missing, ToolError +from .errors import Conflict, Missing, ToolError from .fields import NAME_LIMIT, ORDER_NAME, decode, expect, field, message_text, single from .records import write_atomic from .register import REGISTER, agent_order, read_register from .signed import message_signed, signing -from .skill_scripts import NIGHT, RADIO, RASCI +from .skill_scripts import ESTIMATE, NIGHT, RADIO, RASCI NIGHTS = { "drain": (["drain"], "Operator ran kvb-org-drain in the Agent Orchestrator: run kvb-agent-drain; verify with night.py status."), @@ -23,7 +23,7 @@ TO_ALL = {"newsletter": ["post"], **{cls: ["send", cls] for cls in ("broadcast", EVERYONE = "*" DENIED = "Denied by the operator." MESSAGE_KEYS = ("to", "text", "upload", "class", "voice") -SIGNED_BOARD = ("clear-mayday", *NIGHTS) +SIGNED_BOARD = ("clear-mayday", *NIGHTS, "estimates") def chosen(payload: dict, choices: dict[str, list[str]]) -> list[str]: @@ -142,17 +142,31 @@ class Messaging: item, _, _ = self.composed(payload, {**DIRECT, **TO_ALL}) return self.challenge_for([item]) + def stored_message(self, item: dict, proof: dict, upload: dict | None) -> dict: + """Args: + item: the message item a pinned key signed. + proof: that signature, kept so operator_message can prove it. + upload: the attached upload or None. + + Returns: + The stored record, addressed to the radio address the claims + give the recipient. + """ + target = item["to"] + claims = read_claims(self.council) + address = next((claim["address"] for claim in claims if claim["agent"] == target and claim["address"]), target) + fields = {"to": target, "address": address, "text": item["text"], "upload": upload, "voice": item["voice"] or None, "voice_sha256": item["voice_sha256"], "class": item["class"]} + return self.messages.add({**fields, "proof": proof, "created_ts": time.time()}) + + def radioed(self, record: dict, command: list[str]) -> dict: + targets = [] if record["to"] == EVERYONE else ["--to", record["address"]] + return self.scripts.run(RADIO, [*command, *targets, "--session", HUMAN, f"[kvb-orchestrator msg {record['id']}] {self.relayed(record)}"]) + def relay(self, payload: dict, choices: dict[str, list[str]]) -> dict: with self.voice_notes.held(payload["voice"]): item, command, upload = self.composed(payload, choices) - proof = self.operator_proven(payload["proof"], item) - target = item["to"] - claims = read_claims(self.council) - address = next((claim["address"] for claim in claims if claim["agent"] == target and claim["address"]), target) - fields = {"to": target, "address": address, "text": item["text"], "upload": upload, "voice": item["voice"] or None, "voice_sha256": item["voice_sha256"], "class": item["class"]} - record = self.messages.add({**fields, "proof": proof, "created_ts": time.time()}) - targets = [] if target == EVERYONE else ["--to", address] - run = self.scripts.run(RADIO, [*command, *targets, "--session", HUMAN, f"[kvb-orchestrator msg {record['id']}] {self.relayed(record)}"]) + record = self.stored_message(item, self.operator_proven(payload["proof"], item), upload) + run = self.radioed(record, command) return {"id": record["id"], "ok": run["code"] == 0, "runs": [run]} def message(self, body: bytes) -> dict: @@ -191,6 +205,11 @@ class Messaging: if unknown: raise ToolError(f"unknown agents: {', '.join(unknown)[:300]}; the register and the council know {', '.join(known)[:600] or 'none'}") runs = [self.scripts.run(RASCI, ["agents", "order", ",".join(names)])] + elif action == "estimates": + estimation = self.settings.get()["estimation"] + if estimation != "points": + raise Conflict({"estimation": estimation, "reason": "estimates are requested only in story points mode"}) + runs = [self.scripts.run(ESTIMATE, ["request", "--session", HUMAN])] elif action in NIGHTS: command, text = NIGHTS[action] runs = [self.scripts.run(NIGHT, [*command, "--session", HUMAN])] diff --git a/orchestrator/kvb_orchestrator/settings.py b/orchestrator/kvb_orchestrator/settings.py index 0316225..a82f08e 100644 --- a/orchestrator/kvb_orchestrator/settings.py +++ b/orchestrator/kvb_orchestrator/settings.py @@ -14,6 +14,7 @@ from .fields import FIELDS, NAME_LIMIT, expect from .records import read_json, write_atomic THEMES = ("sky", "light", "dark", "custom") +ESTIMATIONS = ("none", "points") TONES = ("mayday", "review", "direct", "radio") COLOR = re.compile(r"#[0-9a-f]{6}") CLOCK = re.compile(r"(?:[01][0-9]|2[0-3]):[0-5][0-9]") @@ -44,6 +45,7 @@ SETTINGS = { "graph": {"axis": "none", "yaw": 30.0, "pitch": 20.0, "zoom": 1.0, "pan_x": 0.0, "pan_y": 0.0}, "sky": {"azimuth": 180.0, "altitude": 0.0, "fov": 180.0}, }, + "estimation": "none", } SPACE_AXES = ("none", "time", "spawn") SPACE_LIMITS = { @@ -189,6 +191,8 @@ def valid_settings(raw: object) -> dict: raise ToolError(f"theme must be one of {', '.join(THEMES)}") if not isinstance(data["streams"], list) or len(data["streams"]) > STREAMS_LIMIT: raise ToolError(f"streams must be a list of at most {STREAMS_LIMIT} entries") + if not isinstance(data["estimation"], str) or data["estimation"] not in ESTIMATIONS: + raise ToolError(f"estimation must be one of {', '.join(ESTIMATIONS)}") return { "latitude": number_in(data["latitude"], "latitude", -90.0, 90.0), "longitude": number_in(data["longitude"], "longitude", -180.0, 180.0), @@ -201,6 +205,7 @@ def valid_settings(raw: object) -> dict: "streams": [stream_entry(entry, index) for index, entry in enumerate(data["streams"])], "speech": speech_setting(data["speech"]), "space": space_setting(data["space"]), + "estimation": data["estimation"], } diff --git a/orchestrator/kvb_orchestrator/skill_scripts.py b/orchestrator/kvb_orchestrator/skill_scripts.py index 2eb9c46..d606d0b 100644 --- a/orchestrator/kvb_orchestrator/skill_scripts.py +++ b/orchestrator/kvb_orchestrator/skill_scripts.py @@ -13,6 +13,7 @@ RADIO = "kvb-org-radio/scripts/radio.py" RASCI = "kvb-org-rasci/scripts/rasci.py" NIGHT = "kvb-org-suspend/scripts/night.py" OPERATOR_SCRIPT = "kvb-org-operator/scripts/operator.py" +ESTIMATE = "kvb-org-queue-estimate/scripts/estimate.py" class Runner(Protocol): diff --git a/orchestrator/kvb_orchestrator/tasking.py b/orchestrator/kvb_orchestrator/tasking.py index b28c8a5..0180cba 100644 --- a/orchestrator/kvb_orchestrator/tasking.py +++ b/orchestrator/kvb_orchestrator/tasking.py @@ -9,7 +9,7 @@ from .register import REGISTER, read_register, status_of from .signed import signing from .tasks import editable -CHANGE_LIMIT = 10 +CHANGE_LIMIT = 100 CHANGE_KEYS = ("kind", "id", "argv") @@ -25,7 +25,9 @@ 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 or WSJF estimate. + 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. Returns: The signed items, each for an open register task outside @@ -60,7 +62,7 @@ class Tasking: items = self.task_changes(payload["items"]) proof = self.covered(payload["proof"], items, time.time()) for item in items: - self.still_signed(proof, item) + self.operator_proven(proof, item) ids = [] 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} diff --git a/orchestrator/kvb_orchestrator/tasks.py b/orchestrator/kvb_orchestrator/tasks.py index e680e1a..15c9af2 100644 --- a/orchestrator/kvb_orchestrator/tasks.py +++ b/orchestrator/kvb_orchestrator/tasks.py @@ -3,8 +3,8 @@ from __future__ import annotations import itertools import statistics -from .council import agent -from .records import iso_epoch, shown +from .council import HUMAN, agent +from .records import epoch, iso_epoch, shown from .register import ENDED, OPERATOR, parties, register_row, status_of from .signed import signing from .tree import inside @@ -13,11 +13,13 @@ ENDED_SHOWN = 100 OPEN_SHOWN = 300 QUEUE_SHOWN = 100 PARTIES_SHOWN = 40 +AFTER_SHOWN = 40 TEXT_SHOWN = 500 FIBONACCI = (1, 2, 3, 5, 8, 13, 21, 34, 55, 89) LETTERS = "RASCI" HOTFIX = "hotfix/" QUEUE = "queue/" +HOUR = 3600 def kind_of(task: str) -> str: @@ -36,23 +38,42 @@ def fibonacci(value: object) -> bool: def estimates_of(row: dict) -> dict[str, dict]: + """Returns: + Each party's Fibonacci business value and complexity; the + Operator's complexity is None where it recorded a business value + alone. + """ raw = row.get("estimates") found = {} for party, entry in itertools.islice(raw.items(), PARTIES_SHOWN) if isinstance(raw, dict) else (): - if isinstance(entry, dict) and fibonacci(entry.get("value")) and fibonacci(entry.get("complexity")): - found[str(party)[:TEXT_SHOWN]] = {"value": entry["value"], "complexity": entry["complexity"]} + alone = party == HUMAN and isinstance(entry, dict) and "complexity" not in entry + if isinstance(entry, dict) and fibonacci(entry.get("value")) and (alone or fibonacci(entry.get("complexity"))): + found[str(party)[:TEXT_SHOWN]] = {"value": entry["value"], "complexity": entry.get("complexity")} return found +def business_value(estimates: dict[str, dict]) -> float | None: + if HUMAN in estimates: + return estimates[HUMAN]["value"] + return statistics.median(entry["value"] for entry in estimates.values()) if estimates else None + + +def points(estimates: dict[str, dict]) -> float | None: + sized = [entry["complexity"] for entry in estimates.values() if entry["complexity"] is not None] + return statistics.median(sized) if sized else None + + def wsjf(estimates: dict[str, dict]) -> dict | None: """Returns: - Median business value, median complexity and their quotient, the - same reading as rasci.py's wsjf(); None without an estimate. + Business value, median complexity and their quotient, the same + reading as rasci.py's wsjf(): the Operator's business value where + it gave one, else the median; None while no estimate carries a + complexity. """ - if not estimates: + complexity = points(estimates) + if complexity is None: return None - value = statistics.median(entry["value"] for entry in estimates.values()) - complexity = statistics.median(entry["complexity"] for entry in estimates.values()) + value = business_value(estimates) return {"value": value, "complexity": complexity, "score": round(value / complexity, 2)} @@ -96,6 +117,11 @@ def located(row: dict, trees: list[dict], open_task: bool) -> dict: return {"worktree": tree["path"], "worktree_branch": tree["branch"], "repo": tree["repo"], "flow": tree["flow"], "open_work": [*own, *tree["open_work"]]} +def after_of(row: dict) -> list[str]: + raw = row.get("after") + return [other for other in raw[:AFTER_SHOWN] if isinstance(other, str) and len(other) <= TEXT_SHOWN] if isinstance(raw, list) else [] + + def task_row(task: str, row: dict, trees: list[dict]) -> dict: estimates = estimates_of(row) ended = status_of(row) in ENDED @@ -108,6 +134,12 @@ def task_row(task: str, row: dict, trees: list[dict]) -> dict: "prio": prio_of(row), "estimates": estimates, "wsjf": wsjf(estimates), + "value": business_value(estimates), + "points": points(estimates), + "start": iso_epoch(row.get("start")), + "end": iso_epoch(row.get("end")) if ended else None, + "after": after_of(row), + "base": shown(row.get("base"))[:TEXT_SHOWN], "roles": roles_of(row), "queue": None, **located(row, trees, not ended), @@ -134,6 +166,12 @@ def queue_row(entry: dict, created: float, trees: list[dict]) -> dict: "prio": None, "estimates": {}, "wsjf": None, + "value": None, + "points": None, + "start": (epoch(entry.get("granted")) or created) if entry.get("status") == "active" else None, + "end": None, + "after": [], + "base": "-", "roles": {}, "queue": {"kind": text["kind"], "parent": text["task"], "enqueued": created}, **located(entry, trees, False), @@ -165,3 +203,75 @@ def task_rows(tasks: dict[str, dict], queue: list[tuple[dict, float]], scan: dic *(queue_row(entry, created, trees) for entry, created in jobs), *(task_row(task, tasks[task], trees) for task in ended[:ENDED_SHOWN]), ] + + +def register_ranks(tasks: dict[str, dict], agents: list[str]) -> dict[str, int]: + """Returns: + Each open task's index in rasci.py's ranked(): pin, WSJF, the + worker's place in the operator's agent order, then the name. + """ + order = {name: index for index, name in enumerate(agents)} + + def key(task: str) -> tuple: + worker = tasks[task].get("worker") + return (*rank(task, tasks[task])[:4], order.get(worker, len(order)) if isinstance(worker, str) else len(order), task) + + current = sorted((task for task, row in tasks.items() if status_of(row) not in ENDED), key=key) + return {task: index for index, task in enumerate(current)} + + +def queue_chain(queue: list[tuple[dict, float]], tasks: dict[str, dict], agents: list[str], now: float) -> list[tuple[str, bool]]: + """Args: + queue: the kvb-org-queue entries with their enqueue time. + tasks: the register rows by task. + agents: the operator's agent order. + now: epoch seconds. + + Returns: + The entry ids in grant order, each with whether it waits: the + active ones by grant time, then the waiting ones as queue.py's + ordered() sorts them (hotfix first, then the register rank less one + per whole hour waited, then the holder's agent order, then the + enqueue sequence). + """ + ranks = register_ranks(tasks, agents) + order = {name: index for index, name in enumerate(agents)} + + def waiting_key(pair: tuple[dict, float]) -> tuple: + entry, created = pair + task, name, seq = entry.get("task"), entry.get("name"), entry.get("seq") + hours = int((now - created) // HOUR) + place = ranks[task] if isinstance(task, str) and task in ranks else len(ranks) + holder = order[name] if isinstance(name, str) and name in order else len(order) + return (not entry.get("hotfix"), place - hours, holder, seq if isinstance(seq, int) and not isinstance(seq, bool) else 0) + + active = sorted((pair for pair in queue if pair[0].get("status") == "active"), key=lambda pair: epoch(pair[0].get("granted")) or pair[1]) + waiting = sorted((pair for pair in queue if pair[0].get("status") == "waiting"), key=waiting_key) + return [(shown(entry.get("id"))[:TEXT_SHOWN], entry.get("status") == "waiting") for entry, _ in [*active, *waiting]] + + +def branch_of(row: dict) -> str: + return row["branch"] if row["branch"] != "-" else row["worktree_branch"] + + +def task_edges(rows: list[dict], chain: list[tuple[str, bool]]) -> list[dict]: + """Args: + rows: the shown task rows of task_rows(). + chain: queue_chain(). + + Returns: + One {"from", "to", "kind"} edge per predecessor among the shown + rows: kind after for an explicit `rasci.py set --after`, base for a + task whose base branch is another task's branch, queue for each + waiting queue job and the job ahead of it. + """ + shown_tasks = {row["task"] for row in rows} + edges = [{"from": other, "to": row["task"], "kind": "after"} for row in rows for other in row["after"] if other in shown_tasks and other != row["task"]] + branches: dict[str, list[str]] = {} + for row in rows: + if row["kind"] != "queue" and branch_of(row): + branches.setdefault(branch_of(row), []).append(row["task"]) + edges += [{"from": other, "to": row["task"], "kind": "base"} for row in rows if row["kind"] != "queue" for other in branches.get(row["base"], []) if other != row["task"]] + jobs = [(f"{QUEUE}{job}", waits) for job, waits in chain] + edges += [{"from": ahead, "to": job, "kind": "queue"} for (ahead, _), (job, waits) in itertools.pairwise(jobs) if waits and {ahead, job} <= shown_tasks] + return edges diff --git a/orchestrator/kvb_orchestrator/view.py b/orchestrator/kvb_orchestrator/view.py index cd2ff70..5637fea 100644 --- a/orchestrator/kvb_orchestrator/view.py +++ b/orchestrator/kvb_orchestrator/view.py @@ -36,7 +36,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 task_rows +from .tasks import queue_chain, task_edges, task_rows RADIO_WINDOW = 86400 JOBS_SHOWN = 20 @@ -103,6 +103,9 @@ class View: people = {node["id"]: {"name": node["label"], "now": directory.now(node["id"]) or node["label"], "names": node["names"]} for node in drawn["nodes"] if node["kind"] != "subagent"} settings = self.settings.get() inbox = self.inbox(now, tasks, directory) + queue = queue_entries(self.council) + order = stored_order(self.council / AGENT_ORDER) + rows = task_rows(tasks, queue, self.tree.get()) return { "settings": settings, "media": media_origins(settings), @@ -128,8 +131,9 @@ class View: "messages": [message_row(message) for message in self.messages.listed() if begin <= message["created_ts"] <= end], "operator": operator_rows(tasks), "register": register_rows(tasks), - "tasks": task_rows(tasks, queue_entries(self.council), self.tree.get()), - "agent_order": agent_order(tasks, claims, stored_order(self.council / AGENT_ORDER)), + "tasks": rows, + "task_edges": task_edges(rows, queue_chain(queue, tasks, order, now)), + "agent_order": agent_order(tasks, claims, order), "radio": radio, "claims": claims, "grants": grant_rows, diff --git a/orchestrator/templates/index.html.j2 b/orchestrator/templates/index.html.j2 index 4a6db0d..de073f7 100644 --- a/orchestrator/templates/index.html.j2 +++ b/orchestrator/templates/index.html.j2 @@ -453,6 +453,8 @@ pre.notes { margin: 0; padding: 8px 10px; background: var(--sunken); border-radi {% include "popup.js" %} {% include "faces.js" %} + +{% include "plan.js" %} ", "")) self.assertEqual((state["register"][0]["title"], state["radio"][0]["from"]), ("", "")) - def test_the_index_scripts_parse_and_embed_the_ambience_voice_nav_popup_faces_and_space_modules(self): + def test_the_index_scripts_parse_and_embed_the_ambience_voice_nav_popup_faces_plan_and_space_modules(self): html = self.index() - for module in (AMBIENCE, VOICE, NAV, POPUP, FACES, SPACE): + for module in (AMBIENCE, VOICE, NAV, POPUP, FACES, PLAN, SPACE): with self.subTest(module=module.name): self.assertIn(module.read_text(encoding="utf-8").strip(), html) scripts = re.findall(r'', html, re.DOTALL) @@ -74,7 +74,7 @@ class TestHubPage(HubCase): page.write_text("\n;\n".join(scripts), encoding="utf-8") done = subprocess.run(["node", "--check", str(page)], capture_output=True, text=True, timeout=60, check=False) self.assertEqual(done.returncode, 0, done.stderr) - for module in (AMBIENCE, VOICE, NAV, POPUP, FACES, SPACE): + for module in (AMBIENCE, VOICE, NAV, POPUP, FACES, PLAN, SPACE): with self.subTest(module=module.name): done = subprocess.run(["node", "--check", str(module)], capture_output=True, text=True, timeout=60, check=False) self.assertEqual(done.returncode, 0, done.stderr) diff --git a/tests/test_hub_plan.py b/tests/test_hub_plan.py new file mode 100644 index 0000000..5c0c833 --- /dev/null +++ b/tests/test_hub_plan.py @@ -0,0 +1,150 @@ +from __future__ import annotations + +import json +import subprocess +import unittest + +from hub_support import PLAN + +HOUR = 3600 +NOW = 100 * HOUR +PLAN_PROBE = """ +const P = require(process.argv[1]); +const H = 3600; +const now = 100 * H; +const row = (task, fields) => ({task, kind: "task", status: "queued", ended: false, start: null, end: null, points: null, wsjf: null, value: null, worker: "-", title: "-", roles: {}, ...fields}); +const rows = [ + row("work/a", {status: "done", ended: true, start: 80 * H, end: 82 * H, points: 2, worker: "Skills"}), + row("work/b", {status: "done", ended: true, start: 84 * H, end: 88 * H, points: 2, roles: {Stage: "R"}}), + row("work/c", {status: "done", ended: true, start: 90 * H, end: 91 * H}), + row("work/run", {status: "running", start: 95 * H, points: 5, worker: "Stage"}), + row("work/next", {points: 3}), + row("work/tail", {}), + row("work/stop", {status: "stopped", ended: true, start: 85 * H, end: 86 * H, points: 1}), + row("queue/x", {kind: "queue", status: "awaiting"}), + row("operator/push", {kind: "command", status: "done", ended: true, end: 70 * H}), +]; +const edges = [ + {from: "work/run", to: "work/next", kind: "after"}, + {from: "work/next", to: "work/tail", kind: "after"}, + {from: "work/c", to: "work/tail", kind: "base"}, + {from: "work/run", to: "queue/x", kind: "queue"}, + {from: "work/gone", to: "work/tail", kind: "base"}, +]; +const loop = [row("loop/a", {}), row("loop/b", {})]; +const looped = [{from: "loop/a", to: "loop/b", kind: "base"}, {from: "loop/b", to: "loop/a", kind: "queue"}]; +const lay = (estimation, forecast) => P.bars(rows, edges, {now, estimation, forecast}).map((bar) => [bar.task, bar.start / H, bar.end / H, bar.forecast, bar.dot]); +const sorts = [ + row("t1", {wsjf: {score: 2}, value: 8, points: 3, status: "running", start: 5, worker: "Bravo", title: "Zulu"}), + row("t2", {status: "done", start: 3, end: 9}), + row("t3", {wsjf: {score: 5}, value: 3, points: 1, worker: "alpha", title: "Alpha"}), + row("t4", {wsjf: {score: 2}, value: 13, points: 8, status: "blocked", start: 1, end: 4, worker: "Charlie", title: "mike"}), +]; +const context = {filtered: [rows[0], rows[7]], since: 90 * H, until: 99 * H, now, holders: {"identity/oidc": "Stage"}}; +const scope = (choice) => { + const found = P.scoped(rows, choice, context); + return {tasks: found.rows.map((entry) => entry.task), from: found.from / H, to: found.to / H}; +}; +const series = (estimation) => P.burn(P.scoped(rows, "all", context).rows, {estimation, from: 70 * H, to: now}).points.map((point) => [point.t / H, point.scope, point.done, point.remaining, point.unestimated]); +console.log(JSON.stringify({ + options: [P.FORECASTS, P.ESTIMATIONS, P.SORTS.map(([id]) => id), P.SERIES], + medians: P.medians(rows), + median: [P.median([]), P.median([3, 1, 2]), P.median([4, 1, 3, 2])], + bars: {noneAlways: lay("none", "always"), pointsAlways: lay("points", "always"), pointsOnly: lay("points", "points"), noneOnly: lay("none", "points"), real: lay("points", "real")}, + loop: P.bars(loop, looped, {now, estimation: "none", forecast: "always"}).map((bar) => bar.task), + empty: P.bars([row("work/new", {}), row("work/gone", {status: "done", ended: true})], [], {now, estimation: "none", forecast: "always"}), + scopes: {all: scope("all"), filter: scope("filter"), agent: scope("agent:Stage"), line: scope("line:identity/oidc"), none: scope("agent:Nobody")}, + burn: {none: series("none"), points: series("points")}, + sorted: Object.fromEntries(P.SORTS.map(([id]) => [id, [P.sortRows(sorts, id, false).map((entry) => entry.task), P.sortRows(sorts, id, true).map((entry) => entry.task)]])), + unknown: P.sortRows(sorts, "nope", false).map((entry) => entry.task), +})); +""" + + +class TestPlanModule(unittest.TestCase): + @classmethod + def setUpClass(cls): + done = subprocess.run(["node", "-e", PLAN_PROBE, str(PLAN)], capture_output=True, text=True, timeout=60, check=True) + cls.data = json.loads(done.stdout) + + def test_the_selectors_offer_the_forecasts_estimations_sort_modes_and_series(self): + forecasts, estimations, sorts, series = self.data["options"] + self.assertEqual(forecasts, [["always", "always"], ["points", "points only"], ["real", "real only"]]) + self.assertEqual(estimations, [["none", "No estimate"], ["points", "Story points"]]) + self.assertEqual(sorts, ["prio", "wsjf", "value", "complexity", "status", "start", "end", "agent", "title"]) + self.assertEqual(series, ["scope", "done", "remaining", "unestimated"]) + + def test_medians_take_done_tasks_only_and_the_points_of_those_with_points(self): + self.assertEqual(self.data["medians"], {"duration": 2 * HOUR, "perPoint": 1.5 * HOUR}) + self.assertEqual(self.data["median"], [None, 2, 2.5]) + + def test_without_estimate_a_forecast_is_as_long_as_the_median_duration_behind_its_predecessors(self): + self.assertEqual(self.data["bars"]["noneAlways"], [ + ["work/a", 80, 82, False, False], + ["work/b", 84, 88, False, False], + ["work/c", 90, 91, False, False], + ["work/run", 95, 100, False, False], + ["work/next", 100, 102, True, False], + ["work/tail", 102, 104, True, False], + ["work/stop", 85, 86, False, False], + ["queue/x", 100, 102, True, False], + ["operator/push", 70, 70, False, True], + ]) + + def test_with_story_points_a_forecast_is_points_times_the_median_hours_per_point(self): + bars = {task: rest for task, *rest in self.data["bars"]["pointsAlways"]} + self.assertEqual(bars["work/next"], [100, 104.5, True, False]) + self.assertEqual(bars["work/tail"], [104.5, 106.5, True, False]) + + def test_points_only_draws_a_dot_behind_the_predecessors_of_a_task_without_estimate(self): + only = {task: rest for task, *rest in self.data["bars"]["pointsOnly"]} + self.assertEqual((only["work/next"], only["work/tail"], only["queue/x"]), ([100, 104.5, True, False], [104.5, 104.5, True, True], [100, 100, True, True])) + none = {task: rest for task, *rest in self.data["bars"]["noneOnly"]} + self.assertEqual((none["work/next"], none["work/tail"]), ([100, 100, True, True], [100, 100, True, True])) + + def test_real_only_draws_no_forecast(self): + self.assertEqual([task for task, *_ in self.data["bars"]["real"]], ["work/a", "work/b", "work/c", "work/run", "work/stop", "operator/push"]) + self.assertFalse(any(bar[3] for bar in self.data["bars"]["real"])) + + def test_a_cycle_still_draws_every_task_no_done_task_means_a_dot_and_an_ended_task_without_times_no_bar(self): + self.assertEqual(self.data["loop"], ["loop/a", "loop/b"]) + self.assertEqual(self.data["empty"], [{"task": "work/new", "start": NOW, "end": NOW, "forecast": True, "dot": True}]) + + def test_the_scopes_are_the_whole_register_the_filter_in_its_window_and_one_agent_or_line(self): + work = ["work/a", "work/b", "work/c", "work/run", "work/next", "work/tail", "work/stop"] + scopes = self.data["scopes"] + self.assertEqual(scopes["all"], {"tasks": work, "from": 80, "to": 100}) + self.assertEqual(scopes["filter"], {"tasks": ["work/a"], "from": 90, "to": 99}) + self.assertEqual(scopes["agent"], {"tasks": ["work/b", "work/run"], "from": 84, "to": 100}) + self.assertEqual(scopes["line"], scopes["agent"]) + self.assertEqual(scopes["none"], {"tasks": [], "from": 76, "to": 100}) + + def test_without_estimate_the_burn_counts_tasks(self): + self.assertEqual(self.data["burn"]["none"], [ + [70, 2, 0, 2, 0], [80, 3, 0, 3, 0], [82, 3, 1, 2, 0], [84, 4, 1, 3, 0], [85, 5, 1, 4, 0], [86, 4, 1, 3, 0], + [88, 4, 2, 2, 0], [90, 5, 2, 3, 0], [91, 5, 3, 2, 0], [95, 6, 3, 3, 0], [100, 6, 3, 3, 0], + ]) + + def test_with_story_points_the_burn_sums_complexity_and_counts_the_unestimated_apart(self): + burn = self.data["burn"]["points"] + self.assertEqual(burn[0], [70, 3, 0, 3, 1]) + self.assertEqual(burn[-1], [100, 12, 4, 8, 1]) + self.assertEqual([point[4] for point in burn], [1, 1, 1, 1, 1, 1, 1, 2, 1, 1, 1]) + + def test_the_list_sorts_by_every_mode_in_both_directions_with_missing_keys_last(self): + self.assertEqual(self.data["sorted"], { + "prio": [["t1", "t2", "t3", "t4"], ["t4", "t3", "t2", "t1"]], + "wsjf": [["t1", "t4", "t3", "t2"], ["t3", "t1", "t4", "t2"]], + "value": [["t3", "t1", "t4", "t2"], ["t4", "t1", "t3", "t2"]], + "complexity": [["t3", "t1", "t4", "t2"], ["t4", "t1", "t3", "t2"]], + "status": [["t3", "t1", "t4", "t2"], ["t2", "t4", "t1", "t3"]], + "start": [["t4", "t2", "t1", "t3"], ["t1", "t2", "t4", "t3"]], + "end": [["t4", "t2", "t1", "t3"], ["t2", "t4", "t1", "t3"]], + "agent": [["t3", "t1", "t4", "t2"], ["t4", "t1", "t3", "t2"]], + "title": [["t3", "t4", "t2", "t1"], ["t1", "t2", "t4", "t3"]], + }) + self.assertEqual(self.data["unknown"], ["t1", "t2", "t3", "t4"]) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_hub_settings.py b/tests/test_hub_settings.py index cbd9739..1f0ed8b 100644 --- a/tests/test_hub_settings.py +++ b/tests/test_hub_settings.py @@ -150,6 +150,22 @@ class TestHubSettings(HubCase): self.assertEqual(status, 200, body) self.assertEqual(json.loads(body)["settings"]["speech"], {"on": False, "classes": ["broadcast", "mayday", "security"], "muted": ["Operator", "name:Docs", "s1"]}) + def test_the_estimation_mode_defaults_to_no_estimate_is_stored_and_survives_a_reload(self): + self.assertEqual((settings.SETTINGS["estimation"], self.snapshot()["settings"]["estimation"]), ("none", "none")) + good = self.guard(self.host) + for wrong in ("Story points", "", None, 1, ["points"]): + with self.subTest(wrong=wrong): + self.assertEqual(self.post_json("/api/settings", {**settings.SETTINGS, "estimation": wrong}, good)[0], 400) + self.assertEqual(self.post_json("/api/settings", {key: value for key, value in settings.SETTINGS.items() if key != "estimation"}, good)[0], 400) + status, _, body = self.save({**self.snapshot()["settings"], "estimation": "points"}, self.pin()) + self.assertEqual((status, json.loads(body)["settings"]["estimation"]), (200, "points")) + self.assertEqual(json.loads((self.state / "settings.json").read_text(encoding="utf-8"))["estimation"], "points") + self.assertEqual(settings.Settings(self.state / "settings.json").get()["estimation"], "points") + self.assertEqual(self.snapshot()["settings"]["estimation"], "points") + older = {key: value for key, value in self.settings_payload(STREAMS).items() if key != "estimation"} + (self.state / "settings.json").write_text(json.dumps(older), encoding="utf-8") + self.assertEqual(settings.Settings(self.state / "settings.json").get()["estimation"], "none") + def test_settings_written_before_the_radio_filter_keep_their_values(self): older = {key: value for key, value in self.settings_payload(STREAMS).items() if key != "speech"} (self.state / "settings.json").write_text(json.dumps(older), encoding="utf-8") diff --git a/tests/test_hub_state.py b/tests/test_hub_state.py index 8ad6aea..fddd2bb 100644 --- a/tests/test_hub_state.py +++ b/tests/test_hub_state.py @@ -151,7 +151,7 @@ class TestHubState(HubCase): set(state), { "now", "since", "until", "first", "alarm_since", "reviews", "asks", "decisions", "messages", "operator", - "register", "tasks", "agent_order", "radio", "claims", "grants", "night", "silence", "graph", + "register", "tasks", "task_edges", "agent_order", "radio", "claims", "grants", "night", "silence", "graph", "inbox", "pending", "rules", "approvers", "jobs", "views", "llm", "people", "settings", "media", "speech", "badges", "kinds", }, diff --git a/tests/test_hub_tasks.py b/tests/test_hub_tasks.py index 52c4fa8..1807a4d 100644 --- a/tests/test_hub_tasks.py +++ b/tests/test_hub_tasks.py @@ -1,5 +1,6 @@ from __future__ import annotations +import importlib.util import json import os import signal @@ -10,19 +11,31 @@ from pathlib import Path from unittest import mock from hub_case import HubCase -from hub_support import runner, signing, verifier -from kvb_orchestrator import skill_scripts, tasks, tree +from hub_support import REPO_ROOT, iso, key, runner, signing, verifier +from kvb_orchestrator import settings, skill_scripts, tasks, tree BROKEN = {"pull": [], "push": [], "source": "invalid"} FLOW = {"pull": ["origin/main", "store-i18n"], "push": ["main", "production"], "source": "file"} PRIO = ["set", "work/feature", "--prio", "3"] ESTIMATE = ["estimate", "work/feature", "Operator", "--value", "8", "--complexity", "3"] +QUEUE_SCRIPT = REPO_ROOT / "skills" / "kvb-org-queue" / "scripts" / "queue.py" def change(argv: list, task: str = "work/feature") -> dict: return {"kind": "task", "id": task, "argv": argv} +def valued(task: str, value: str) -> dict: + return change(["estimate", task, "Operator", "--value", value], task) + + +def load_queue(): + spec = importlib.util.spec_from_file_location("queue_for_the_hub_chain", QUEUE_SCRIPT) + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + return module + + class TestHubTasks(HubCase): def places(self) -> tuple[str, str, str]: root = self.write_tree() @@ -307,6 +320,138 @@ class TestHubTasks(HubCase): self.assertEqual((len(rows[0]["estimates"]), len(rows[0]["roles"])), (2, 2)) self.assertEqual(max(len(row["task"]) for row in self.snapshot()["tasks"]), tasks.TEXT_SHOWN) + def plan_register(self, now: float) -> tuple[dict, dict]: + register = { + "work/base": {"status": "done", "branch": "feature/base", "start": iso(now - 7200), "end": iso(now - 3600)}, + "work/child": {"status": "queued", "base": "feature/base"}, + "work/next": {"status": "queued", "after": ["work/child", "work/gone", "work/next", 7]}, + "work/side": {"status": "running", "branch": "side", "base": "feature/base", "start": iso(now - 600)}, + "work/pinned": {"status": "queued", "prio": 1}, + } + entries = [ + {"id": "deploy-1", "seq": 1, "kind": "deploy", "task": "work/side", "name": "Stage", "hotfix": None, "status": "active", "enqueued": now - 200, "granted": now - 100, "cmd": "make deploy"}, + {"id": "gate-2", "seq": 2, "kind": "gate", "task": "work/next", "name": "Docs", "hotfix": None, "status": "waiting", "enqueued": now - 50, "granted": None, "cmd": "make test"}, + {"id": "gate-3", "seq": 3, "kind": "gate", "task": "work/pinned", "name": "Docs", "hotfix": None, "status": "waiting", "enqueued": now - 40, "granted": None, "cmd": "make test"}, + {"id": "hot-4", "seq": 4, "kind": "gate", "task": "work/next", "name": "Docs", "hotfix": "db", "status": "waiting", "enqueued": now - 30, "granted": None, "cmd": "make test"}, + {"id": "old-5", "seq": 5, "kind": "gate", "task": "work/next", "name": "Stage", "hotfix": None, "status": "waiting", "enqueued": now - 7210, "granted": None, "cmd": "make test"}, + ] + return register, {"seq": 5, "entries": entries} + + def test_edges_come_from_explicit_dependencies_branch_bases_and_the_queue_order(self): + now = time.time() + register, queue = self.plan_register(now) + self.write_council({"rasci.json": register, "queue.json": queue}) + state = self.snapshot() + self.assertEqual(state["task_edges"], [ + {"from": "work/child", "to": "work/next", "kind": "after"}, + {"from": "work/base", "to": "work/child", "kind": "base"}, + {"from": "work/base", "to": "work/side", "kind": "base"}, + {"from": "queue/deploy-1", "to": "queue/hot-4", "kind": "queue"}, + {"from": "queue/hot-4", "to": "queue/gate-3", "kind": "queue"}, + {"from": "queue/gate-3", "to": "queue/old-5", "kind": "queue"}, + {"from": "queue/old-5", "to": "queue/gate-2", "kind": "queue"}, + ]) + rows = {row["task"]: row for row in state["tasks"]} + self.assertEqual((rows["work/next"]["after"], rows["work/child"]["base"], rows["work/pinned"]["base"]), (["work/child", "work/gone", "work/next"], "feature/base", "-")) + self.assertAlmostEqual(rows["work/base"]["start"], now - 7200, places=3) + self.assertAlmostEqual(rows["work/base"]["end"], now - 3600, places=3) + self.assertIsNone(rows["work/side"]["end"]) + self.assertEqual((rows["queue/deploy-1"]["start"], rows["queue/gate-2"]["start"]), (now - 100, None)) + + def test_the_queue_chain_follows_the_grant_order_of_queue_py(self): + now = time.time() + register, queue = self.plan_register(now) + entries = [(entry, entry["enqueued"]) for entry in queue["entries"]] + chain = tasks.queue_chain(entries, register, [], now) + self.assertEqual(chain, [("deploy-1", False), ("hot-4", True), ("gate-3", True), ("old-5", True), ("gate-2", True)]) + self.assertEqual(tasks.register_ranks(register, []), {"work/pinned": 0, "work/child": 1, "work/next": 2, "work/side": 3}) + waiting = [entry for entry in queue["entries"] if entry["status"] == "waiting"] + queue_py = load_queue() + for agents, first in (([], "gate-3"), (["Stage", "Docs"], "old-5")): + with self.subTest(agents=agents): + ranks = tasks.register_ranks(register, agents) + expected = [entry["id"] for entry in queue_py.ordered(waiting, ranks, {name: index for index, name in enumerate(agents)}, now)] + found = [job for job, waits in tasks.queue_chain(entries, register, agents, now) if waits] + self.assertEqual(found, expected) + self.assertEqual(found[:2], ["hot-4", first]) + + def test_a_bulk_business_value_is_stored_only_with_one_verified_assertion_over_the_exact_ids_and_values(self): + register = { + "work/a": {"status": "running", "estimates": {"Skills": {"value": 3, "complexity": 2}, "Docs": {"value": 5, "complexity": 2}}}, + "work/b": {"status": "queued", "estimates": {"Docs": {"value": 2, "complexity": 1}}}, + } + self.write_council({"rasci.json": register}) + before = (self.council / "rasci.json").read_bytes() + signer = self.pin() + batch = [valued("work/a", "21"), valued("work/b", "21")] + proof = self.signed(batch, signer) + good = self.guard(self.host) + for items, relayed, reason in ( + ([batch[0], valued("work/b", "34")], proof, "list every approved item"), + ([valued("work/b", "21"), valued("work/a", "34")], proof, "list every approved item"), + ([*batch, valued("work/c", "21")], proof, "not found"), + (batch, key(-7, "stranger").proof(proof["payload"], self.origin), "not signed"), + (batch, {**proof, "assertion": {**proof["assertion"], "signature": proof["assertion"]["signature"][::-1]}}, "not signed"), + (batch, None, "not signed"), + ): + with self.subTest(items=items, reason=reason): + status, _, body = self.post_json("/api/tasks", {"items": items, "proof": relayed}, good) + self.assertIn(status, (400, 404), body) + self.assertIn(reason, body.decode()) + self.assertEqual([job for job in self.snapshot()["jobs"] if job["kind"] == "task"], []) + status, _, body = self.post_json("/api/tasks", {"items": batch, "proof": proof}, good) + self.assertEqual(status, 202, body) + self.assertEqual(len(json.loads(body)["jobs"]), 2) + for items in (batch, batch[:1], batch[::-1]): + with self.subTest(replayed=items): + status, _, body = self.post_json("/api/tasks", {"items": items, "proof": proof}, good) + self.assertEqual(status, 400, body) + self.assertIn("already used", body.decode()) + self.assertEqual((self.council / "rasci.json").read_bytes(), before) + self.run_runner() + stored = json.loads((self.council / "rasci.json").read_text(encoding="utf-8")) + self.assertEqual({task: row["estimates"]["Operator"]["value"] for task, row in stored.items()}, {"work/a": 21, "work/b": 21}) + self.assertFalse(any("complexity" in row["estimates"]["Operator"] for row in stored.values())) + rows = {row["task"]: row for row in self.snapshot()["tasks"]} + self.assertEqual(rows["work/a"]["wsjf"], {"value": 21, "complexity": 2, "score": 10.5}) + self.assertEqual(rows["work/b"]["wsjf"], {"value": 21, "complexity": 1, "score": 21.0}) + self.assertEqual((rows["work/a"]["value"], rows["work/a"]["points"], rows["work/a"]["estimates"]["Operator"]), (21, 2, {"value": 21, "complexity": None})) + self.assertEqual(self.runner_jobs(), []) + tampered_fields = {"kind": "task", "task": "work/a", "argv": ["estimate", "work/a", "Operator", "--value", "34"], "command": "x", "directory": "", "sha256": "", "status": "queued", "proof": proof} + tampered = self.httpd.hub.jobs.add({**tampered_fields, "code": None, "output": "", "created_ts": time.time(), "started": None, "finished": None}) + replayed = self.httpd.hub.jobs.add({**tampered_fields, "argv": batch[0]["argv"], "code": None, "output": "", "created_ts": time.time(), "started": None, "finished": None}) + self.run_runner() + self.assertIn("does not list this item", self.httpd.hub.jobs.get(tampered["id"])["output"]) + self.assertIn("already used", self.httpd.hub.jobs.get(replayed["id"])["output"]) + self.assertEqual(json.loads((self.council / "rasci.json").read_text(encoding="utf-8")), stored) + + def test_estimates_go_out_only_on_an_explicit_request_in_story_points_mode_once_per_request(self): + self.write_council({"rasci.json": {"work/a": {"status": "queued", "roles": {"Skills": "RA"}}}}) + good = self.guard(self.host) + signer = self.pin() + self.snapshot() + self.index() + status, _, body = self.post_json("/api/board/estimates", {"proof": self.board_proof("estimates", signer)}, good) + self.assertEqual(status, 409, body) + self.assertEqual(self.post_json("/api/board/estimates", {}, self.client())[0], 403) + status, _, body = self.post_json("/api/settings", {**self.snapshot()["settings"], "estimation": "points", "proof": None}, good) + self.assertEqual((status, json.loads(body)["settings"]["estimation"]), (200, "points")) + self.snapshot() + self.index() + self.assertEqual(self.post_json("/api/board/estimates", {}, good)[0], 400) + self.assertEqual(self.post_json("/api/board/estimates", {"proof": None}, good)[0], 400) + self.assertEqual(self.scripts.calls, []) + proof = self.board_proof("estimates", signer) + status, _, body = self.post_json("/api/board/estimates", {"proof": proof}, good) + self.assertEqual(status, 200, body) + self.assertTrue(json.loads(body)["ok"]) + self.assertEqual(self.post_json("/api/board/estimates", {"proof": proof}, good)[0], 400) + self.assertEqual(self.scripts.calls, [(skill_scripts.ESTIMATE, ["request", "--session", "Operator"])]) + self.snapshot() + self.assertEqual(self.post_json("/api/settings", {**self.snapshot()["settings"], "estimation": "none", "proof": None}, good)[0], 200) + self.assertEqual(len(self.scripts.calls), 1) + self.assertEqual(settings.Settings(self.state / "settings.json").get()["estimation"], "none") + if __name__ == "__main__": unittest.main() diff --git a/tests/test_rasci.py b/tests/test_rasci.py index ca49fcb..37a2544 100644 --- a/tests/test_rasci.py +++ b/tests/test_rasci.py @@ -162,6 +162,53 @@ class TestRasci(unittest.TestCase): self._run("set", "work/side", "--base", "") self.assertFalse(any(line.startswith("base:") for line in self._out("get", "work/side"))) + def _after(self) -> dict[str, list[str]]: + rows = json.loads((self.root / "rasci.json").read_text(encoding="utf-8")) + return {task: row["after"] for task, row in rows.items() if "after" in row} + + def test_after_stores_and_drops_a_dependency_and_refuses_unknown_tasks_and_cycles(self): + for task in ("work/a", "work/b", "work/c"): + self._out("set", task) + self._out("set", "work/b", "--after", "work/a") + self._out("set", "work/c", "--after", "work/b", "--after", "work/a", "--after", "work/b") + self.assertEqual(self._after(), {"work/b": ["work/a"], "work/c": ["work/b", "work/a"]}) + self.assertIn("after: work/b", self._out("get", "work/c")) + before = (self.root / "rasci.json").read_bytes() + for args, reason in ( + (("set", "work/a", "--after", "work/c"), "closes a cycle"), + (("set", "work/a", "--after", "work/b", "--title", "kept out"), "closes a cycle"), + (("set", "work/a", "--after", "work/a"), "cannot follow itself"), + (("set", "work/a", "--after", "work/nope"), "no task of the register"), + (("set", "work/new", "--after", "work/nope"), "no task of the register"), + (("set", "work/a", "--after", "Bad Task"), "invalid task"), + ): + with self.subTest(args=args): + result = self._run(*args) + self.assertEqual(result.returncode, 2, result.stdout) + self.assertIn(reason, result.stderr) + self.assertEqual((self.root / "rasci.json").read_bytes(), before) + self._out("set", "work/c", "--drop-after", "work/b") + self.assertEqual(self._after(), {"work/b": ["work/a"], "work/c": ["work/a"]}) + self._out("set", "work/b", "--drop-after", "work/a") + self._out("set", "work/a", "--after", "work/b") + self.assertEqual(self._after(), {"work/a": ["work/b"], "work/c": ["work/a"]}) + + def test_the_operator_business_value_replaces_the_median_and_may_stand_alone(self): + self._run("estimate", "work/a", "One", "--value", "8", "--complexity", "2") + self._run("estimate", "work/a", "Two", "--value", "13", "--complexity", "3") + self._out("estimate", "work/a", "Operator", "--value", "34") + row = json.loads((self.root / "rasci.json").read_text(encoding="utf-8"))["work/a"] + self.assertNotIn("complexity", row["estimates"]["Operator"]) + self.assertEqual(self._out("wsjf")[2], "| 1 | work/a | - | 34 | 2.5 | 13.6 | One, Operator, Two | queued |") + self.assertIn("estimate Operator: business value 34, complexity -", self._out("get", "work/a")) + self._out("estimate", "work/a", "Operator", "--value", "21", "--complexity", "8") + self._out("estimate", "work/a", "Operator", "--value", "5") + row = json.loads((self.root / "rasci.json").read_text(encoding="utf-8"))["work/a"] + self.assertEqual((row["estimates"]["Operator"]["value"], row["estimates"]["Operator"]["complexity"]), (5, 8)) + self.assertEqual(self._out("wsjf")[2], "| 1 | work/a | - | 5 | 3 | 1.67 | One, Operator, Two | queued |") + self._out("estimate", "work/b", "Operator", "--value", "8") + self.assertEqual(self._out("wsjf")[3:], ["| 2 | work/b | - | - | - | - | Operator | queued |", "unestimated: work/b"]) + def test_done_tasks_show_only_with_all(self): self._run("set", "work/a", "--status", "done", "--end", "2026-11-01T12:00:00+00:00") self._run("role", "work/a", "Skills", "RA") @@ -257,6 +304,8 @@ class TestRasci(unittest.TestCase): ("set", "work/a", "--artifact", "no-kind"), ("estimate", "work/a", "One", "--value", "4", "--complexity", "2"), ("estimate", "work/a", "One", "--value", "8"), + ("estimate", "work/a", "Operator", "--complexity", "8"), + ("estimate", "work/a", "Operator", "--value", "4"), ): with self.subTest(args=args): self.assertEqual(self._run(*args).returncode, 2)