diff --git a/docs/2-handbook/swarm/archetypes.md b/docs/2-handbook/swarm/archetypes.md index d3cee7d40..5b99025f0 100644 --- a/docs/2-handbook/swarm/archetypes.md +++ b/docs/2-handbook/swarm/archetypes.md @@ -143,6 +143,28 @@ hyperhiveArchetypes.triage.meta.flows = [ nothing. - Custom agents have no archetype and get no archetype flows. +### On the board + +The notification router card on `/swarm` lists each archetype flow under +**archetype flows**, tagged with its archetype and with the agents it applies to +now, drawn as a graph with hit counts like the other flows. The flow follows +the archetype: an edit to the archetype reaches the card and every agent +within about 10 seconds, and the card has no `edit` button. + +- **Switch off for the archetype.** The `on` switch on the card turns the flow + off for every agent of the archetype, as for a built-in. It adds the key + `archetype:/` to the router's list of switched-off flows. +- **Switch off for one agent.** Each agent on the card has its own checkbox. + Clearing it adds `archetype:/@` to the same list; + the other agents keep the flow. +- **Edit a copy.** `duplicate` opens the new-flow editor on a copy of the flow + as a custom flow. The copy sends to the agent the archetype has when it has + just one, and to nobody until you name an agent when it has more. A custom + flow can also request a review, which an archetype's flow can't. Switch the + archetype's flow off to replace it with your copy. +- A key for a flow or agent that no longer exists stays in the list and does nothing, so + editing an archetype never breaks the stored flows. + ## How the list stays current The swarm controller asks the builder for the list when the queue connects and diff --git a/docs/2-handbook/swarm/ui.md b/docs/2-handbook/swarm/ui.md index e51da3da5..9617094b4 100644 --- a/docs/2-handbook/swarm/ui.md +++ b/docs/2-handbook/swarm/ui.md @@ -10,16 +10,16 @@ domain, covers host-level detail for one hive. The swarm UI is a set of pages. Each one is a card list or a table you open in place, and every one reads from [`swarm-controller`](../../../swarm-controller/README.md): -| route | what | -| ---------------------------------------------- | -------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | -| `/` | the start page: the swarm's name over a digital-rain backdrop with the slug (`hyperhive / we are the wired`, its shaded gradients pulsing in a wave unless reduced motion is on) at the foot, plus a paperclip offering to set up an agent while no `ruth` exists, or her hive declares her destroyed (it opens the create form with `ruth` filled in and the `ruth` archetype picked); the swarm name in the header links here | -| `/agents` | every agent as a card that opens in place (status, terminal, subagents, and settings with permissions, internet, allowed domains and accounts); create agents (with their permissions, optionally from an [archetype](archetypes.md)), link forge, matrix and GitHub accounts | -| `/swarm` | what belongs to the whole swarm: a folded **hives** card with the hive directory and each hive's last reported status, a folded **egress** card with the swarm-wide allowed and denied domains, and a folded, **beta** **notification router** card (with a warning banner) holding the router's flows: a switch for each built-in, every flow drawn as a graph, and an editor for custom flows. `/notifications` redirects here | -| `/agents//terminal` | one agent's live terminal | -| `/agents//subagents//terminal` | one of that agent's subagents' live terminal | -| `/reports` | three folded cards, each read only once opened: the **alerts** the swarm's Grafana holds now, the firing ones, with a button for the pending, no-data and error ones and a link to Grafana's own alert list; read through the controller with a Viewer service-account token, and a cross-repo **issues** list, and a cross-repo **pull requests** list | -| `/jobs` | the controller's job graph, where agent creation and credential mints show progress, and under it the swarm builder's recent builds (agent config PRs, and agents built from `internal/meta` before a deploy), with what nix printed when one failed, and the review classifier's recent runs, a try-out on a real PR, and its settings and keys | -| `/messages` | the swarm-wide message feed: every message between agents, and to them from the operator, reminders and compaction, with a history window and a filter on sender or recipient; a folded **broadcast** card above the feed sends one message to more than one agent at once, and a folded **schedules** card holds the scheduled prompts: add, edit, pause, resume and remove | +| route | what | +| ---------------------------------------------- | --------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | +| `/` | the start page: the swarm's name over a digital-rain backdrop with the slug (`hyperhive / we are the wired`, its shaded gradients pulsing in a wave unless reduced motion is on) at the foot, plus a paperclip offering to set up an agent while no `ruth` exists, or her hive declares her destroyed (it opens the create form with `ruth` filled in and the `ruth` archetype picked); the swarm name in the header links here | +| `/agents` | every agent as a card that opens in place (status, terminal, subagents, and settings with permissions, internet, allowed domains and accounts); create agents (with their permissions, optionally from an [archetype](archetypes.md)), link forge, matrix and GitHub accounts | +| `/swarm` | what belongs to the whole swarm: a folded **hives** card with the hive directory and each hive's last reported status, a folded **egress** card with the swarm-wide allowed and denied domains, and a folded, **beta** **notification router** card (with a warning banner) holding the router's flows: a switch for each built-in, a card for each flow an archetype carries, every flow drawn as a graph, and an editor for custom flows. `/notifications` redirects here | +| `/agents//terminal` | one agent's live terminal | +| `/agents//subagents//terminal` | one of that agent's subagents' live terminal | +| `/reports` | three folded cards, each read only once opened: the **alerts** the swarm's Grafana holds now, the firing ones, with a button for the pending, no-data and error ones and a link to Grafana's own alert list; read through the controller with a Viewer service-account token, and a cross-repo **issues** list, and a cross-repo **pull requests** list | +| `/jobs` | the controller's job graph, where agent creation and credential mints show progress, and under it the swarm builder's recent builds (agent config PRs, and agents built from `internal/meta` before a deploy), with what nix printed when one failed, and the review classifier's recent runs, a try-out on a real PR, and its settings and keys | +| `/messages` | the swarm-wide message feed: every message between agents, and to them from the operator, reminders and compaction, with a history window and a filter on sender or recipient; a folded **broadcast** card above the feed sends one message to more than one agent at once, and a folded **schedules** card holds the scheduled prompts: add, edit, pause, resume and remove | An agent created here or with `swarmctl agent create` starts `paused`; set it `up` from its card. A card shows **login needed** when the agent's claude has no diff --git a/docs/4-implementation/swarm/notify-router.md b/docs/4-implementation/swarm/notify-router.md index bf47a25fb..6efdfc280 100644 --- a/docs/4-implementation/swarm/notify-router.md +++ b/docs/4-implementation/swarm/notify-router.md @@ -285,10 +285,34 @@ no archetype. The router reads the `agent-archetypes` record and the `agent-archetype` key of each agent in its roster at most every 10 seconds, when an event arrives, and runs each archetype's valid, enabled flows after the built-ins and the document's flows, for the archetype's agents. The router logs a flow it can't use and leaves it out. A read that fails keeps the archetype flows it had. -Each runs under the id `:`, so its todos' source is -`router::`. A recipient not in the roster gets nothing, the same as a named agent that doesn't exist. +Each runs under its key, `archetype:/`, which is its flow id in the log, the hit counts and a todo's source (`router:archetype:/`). A recipient not in the roster gets nothing, the same as a named agent that doesn't exist. A todo key is the dedupe key as for any flow: the same key from the operator's document replaces the archetype's todo. +**Switching an archetype flow off.** The document's `disabled_builtins` takes +two more kinds of entry besides built-in ids: an archetype flow's key +(`archetype:/`) switches it off for every member, and the key +followed by `@` for that member only. The check accepts any key of that +shape and doesn't ask whether the flow or agent exists, so an edit to an +archetype can't make the stored document fail it; the router ignores a key that +names nothing. The router applies the list to the archetype flows it has read +(`archetype_flows::enabled`) before it runs them, and a flow with no member left +isn't run. A controller that predates this refuses these keys with a 400, as +it refuses any id that's no built-in. + +**On the board.** `GET /api/notify/flows` also returns `archetype_flows`: +`[{archetype, key, flow, members}]`, one per enabled archetype flow that passes +the archetype check and has an agent, with the flow as the archetype declares it +(a `todo` to `"members"`). The controller reads the archetype records and the +`agent-archetype` keys itself; a read that fails leaves the list empty and the +document still comes back. The page draws a card for each, with the `on` switch (the +key), a checkbox per member (the key with `@`) and `duplicate`, which +opens the editor on `?from=`. The copy replaces `"members"` with the one +member's name, or with an empty agent name when there are two or more, since the +operator's document refuses `members`. Hit counts come from the dry run, which +replays the archetype flows (as the router reads them now, less the document's +switches) next to the document's, under their keys. A router that predates this answers a dry run +with no entry for them, and the page shows zero hits. + **Deploying in any order.** The controller, the router and the UI can be on different hosts. A router that predates `request_reviewer` reads a document holding such a node (a custom flow with one) as one it can't read and keeps the diff --git a/docs/4-implementation/web-ui/swarm-ui.md b/docs/4-implementation/web-ui/swarm-ui.md index 01c376c0c..223b2836a 100644 --- a/docs/4-implementation/web-ui/swarm-ui.md +++ b/docs/4-implementation/web-ui/swarm-ui.md @@ -381,6 +381,17 @@ it starts from on the left, the conditions after them, and the todo or review request each branch ends in. A box cuts a long line of text with an ellipsis; hover over the box to read all of it. +Between the built-ins and the custom flows, **archetype flows** lists the flows +archetypes carry ([how an archetype gets them](../../2-handbook/swarm/archetypes.md#notification-flows)), +one card each, from `archetype_flows` in `GET /api/notify/flows`. The card shows +the archetype's name, an on/off switch, a checkbox for each agent the flow +applies to, and **duplicate**; it has no **edit**, since the flow follows its +archetype. The switch writes the key `archetype:/` into the +document's `disabled_builtins`, and a checkbox the same key followed by `@`. +**duplicate** opens the editor on a draft copy, sent to the archetype's one +agent, or to no agent yet when it has more. A controller that predates +`archetype_flows` leaves the group out. + A built-in flow has a **duplicate to edit** button and a custom flow has **edit** and **duplicate** buttons. **create flow**, at the top of the page beside the refresh interval, starts a custom flow with just a source. Creating and diff --git a/frontend/packages/swarm-ui/src/pages/notifications/FlowDraft.ts b/frontend/packages/swarm-ui/src/pages/notifications/FlowDraft.ts index 642adf43e..f56715541 100644 --- a/frontend/packages/swarm-ui/src/pages/notifications/FlowDraft.ts +++ b/frontend/packages/swarm-ui/src/pages/notifications/FlowDraft.ts @@ -368,12 +368,20 @@ function savedIds(view: FlowsView): string[] { } // The draft the editor opens for a flow not saved yet: an empty one, or a copy -// of the built-in or custom flow `from`. Nothing is sent until it is saved, so +// of the built-in, custom or archetype flow `from` (an archetype flow by its key). Nothing is sent until it is saved, so // the id is a placeholder (see `savedAs`). Null when `from` names no flow. export function seedFlow(view: FlowsView, from: string | null): Flow | null { const taken = savedIds(view); if (from === null) return { ...newFlow("", taken), id: NEW_FLOW_ID }; - const source = [...view.builtins, ...view.flows].find((f) => f.id === from); + const archetype = (view.archetype_flows ?? []).find((a) => a.key === from); + // An archetype's "members" has no meaning in the operator's document: the + // copy sends to the one member there is, or to nobody until one is named. + const source = archetype + ? withMembersAs( + archetype.flow, + archetype.members.length === 1 ? archetype.members[0] : "", + ) + : [...view.builtins, ...view.flows].find((f) => f.id === from); if (!source) return null; return { ...duplicateFlow(source, `${source.name} (copy)`, taken), @@ -453,3 +461,16 @@ export function saveOutcome(status: number): SaveOutcome { if (status === 400) return "invalid"; return "error"; } + +// `flow` with each todo to "members" (an archetype's) sent to the agent `who` +// instead, which is what the editor's recipients can hold. +export function withMembersAs(flow: Flow, who: string): Flow { + return { + ...flow, + nodes: flow.nodes.map((n) => + n.kind === "todo" && (n.to as unknown) === "members" + ? { ...n, to: { agent: who } } + : n, + ), + }; +} diff --git a/frontend/packages/swarm-ui/src/pages/notifications/FlowTypes.ts b/frontend/packages/swarm-ui/src/pages/notifications/FlowTypes.ts index b467417f7..022a2e91b 100644 --- a/frontend/packages/swarm-ui/src/pages/notifications/FlowTypes.ts +++ b/frontend/packages/swarm-ui/src/pages/notifications/FlowTypes.ts @@ -36,10 +36,29 @@ export interface FlowsView { disabled_builtins: string[]; builtins: Flow[]; flows: Flow[]; + // The flows archetypes carry. Absent from a controller that predates them. + archetype_flows?: ArchetypeFlow[]; // Absent from a controller that predates it. schema?: FlowSchema; } +// A flow an archetype carries, with the agents of that archetype. `key` is +// what names it in `disabled_builtins` (off for every member); `key@agent` +// is off for that one member. In `flow` a todo's recipient may be the string +// "members", which the editor's `Recipients` does not model: use +// `withMembersAs` (FlowDraft.ts) before drawing or editing it. +export interface ArchetypeFlow { + archetype: string; + key: string; + flow: Flow; + members: string[]; +} + +// The `disabled_builtins` entry that switches `key` off for one member. +export function memberKey(key: string, agent: string): string { + return `${key}@${agent}`; +} + export function describeRecipients(to: Recipients): string { return "field" in to ? `{${to.field}}` : to.agent; } diff --git a/frontend/packages/swarm-ui/src/pages/notifications/NotificationsPage.css b/frontend/packages/swarm-ui/src/pages/notifications/NotificationsPage.css index 9a324d132..f9983e429 100644 --- a/frontend/packages/swarm-ui/src/pages/notifications/NotificationsPage.css +++ b/frontend/packages/swarm-ui/src/pages/notifications/NotificationsPage.css @@ -281,3 +281,11 @@ .notify-failed { color: var(--red); } + +.notify-flow-agents { + display: flex; + flex-wrap: wrap; + align-items: center; + gap: 0.4em 1em; + margin-top: 0.4em; +} diff --git a/frontend/packages/swarm-ui/src/pages/notifications/NotificationsPage.tsx b/frontend/packages/swarm-ui/src/pages/notifications/NotificationsPage.tsx index 058d29efe..acc00e223 100644 --- a/frontend/packages/swarm-ui/src/pages/notifications/NotificationsPage.tsx +++ b/frontend/packages/swarm-ui/src/pages/notifications/NotificationsPage.tsx @@ -4,7 +4,9 @@ // every few seconds. // // A flow draws as its graph, read-only. A built-in flow has a switch and can -// be duplicated; a custom flow is edited on its own page (). +// be duplicated; an archetype's flow has the same switch, one for each of its +// agents, and can be duplicated but not edited, since it follows its archetype; +// a custom flow is edited on its own page (). // A new flow and a duplicate both open that page on a draft: nothing is created // until it is saved there. import { useState } from "preact/hooks"; @@ -22,9 +24,9 @@ import { import { fetchJson, usePolled } from "../../ui/polled/usePolled.js"; import { Deliveries, DeliveriesSkeleton } from "./Deliveries.js"; import { FlowListSkeleton } from "./FlowListSkeleton.js"; -import { NEW_FLOW_PATH } from "./FlowDraft.js"; +import { NEW_FLOW_PATH, withMembersAs } from "./FlowDraft.js"; import { FlowGraph } from "./FlowGraph.js"; -import { withBuiltin, type FlowsView } from "./FlowTypes.js"; +import { memberKey, withBuiltin, type FlowsView } from "./FlowTypes.js"; import { DELIVERIES_URL, DRY_RUN_URL, @@ -72,6 +74,8 @@ export function NotificationRouter() { // Switch one built-in on or off: the whole document goes back with the // revision this page read (0 for none), so a change made elsewhere since is refused // rather than overwritten. + // The switches of built-ins and of archetype flows are one list, see + // `ArchetypeFlow`. async function setBuiltin(id: string, enabled: boolean) { if (!view) return; setPending(true); @@ -161,6 +165,55 @@ export function NotificationRouter() { ); })} + {(view.archetype_flows ?? []).length > 0 ? ( +

archetype flows

+ ) : null} + {(view.archetype_flows ?? []).map((af) => { + const on = !view.disabled_builtins.includes(af.key); + return ( + +
+ {af.flow.name} + + archetype {af.archetype} + + setBuiltin(af.key, checked)} + /> + + + +
+
+ agents: + {af.members.map((agent) => ( + + setBuiltin(memberKey(af.key, agent), checked) + } + /> + ))} +
+ + + ); + })}

custom flows

{view.flows.length === 0 ?

no custom flows

: null} {view.flows.map((flow) => ( diff --git a/frontend/packages/swarm-ui/test/flow-draft.test.mjs b/frontend/packages/swarm-ui/test/flow-draft.test.mjs index aaec07481..f88a91d81 100644 --- a/frontend/packages/swarm-ui/test/flow-draft.test.mjs +++ b/frontend/packages/swarm-ui/test/flow-draft.test.mjs @@ -387,6 +387,47 @@ test("seedFlow opens an empty or copied draft under the placeholder id", () => { assert.equal(view.flows.length, 1, "nothing is added to the view"); }); +test("seedFlow copies an archetype flow by its key, sending to its one member", () => { + const members = { + ...flow(), + id: "new-issues", + name: "New issues", + nodes: flow().nodes.map((n) => + n.kind === "todo" ? { ...n, to: "members" } : n, + ), + }; + const archetype = (agents) => ({ + revision: 1, + disabled_builtins: [], + builtins: [], + flows: [], + archetype_flows: [ + { + archetype: "triage", + key: "archetype:triage/new-issues", + flow: members, + members: agents, + }, + ], + }); + const recipients = (f) => + f.nodes.filter((n) => n.kind === "todo").map((n) => n.to); + const one = seedFlow(archetype(["alice"]), "archetype:triage/new-issues"); + assert.equal(one.id, NEW_FLOW_ID); + assert.equal(one.name, "New issues (copy)"); + assert.ok(recipients(one).length > 0, "control: the flow has a todo"); + assert.ok(recipients(one).every((to) => to.agent === "alice")); + // Two members: the operator names the agent in the editor. + const two = seedFlow( + archetype(["alice", "bob"]), + "archetype:triage/new-issues", + ); + assert.ok(recipients(two).every((to) => to.agent === "")); + assert.equal(seedFlow(archetype(["alice"]), "archetype:triage/gone"), null); + // The archetype's own flow keeps "members". + assert.ok(recipients(members).every((to) => to === "members")); +}); + test("savedAs takes the id from the name, clear of every saved id", () => { const view = { revision: 1, diff --git a/frontend/packages/swarm-ui/test/flow-types.test.mjs b/frontend/packages/swarm-ui/test/flow-types.test.mjs index ffe8b7514..3b2cd8ef6 100644 --- a/frontend/packages/swarm-ui/test/flow-types.test.mjs +++ b/frontend/packages/swarm-ui/test/flow-types.test.mjs @@ -3,6 +3,7 @@ import assert from "node:assert/strict"; import { test } from "node:test"; import { describeNode, + memberKey, kindValues, ordinal, withBuiltin, @@ -114,3 +115,15 @@ test("describeNode says what a reviewer request and an unknown kind are", () => ); assert.match(describeNode({ id: "x", kind: "future" }), /future/); }); + +test("memberKey names a flow's switch for one agent", () => { + assert.equal( + memberKey("archetype:triage/new-issues", "alice"), + "archetype:triage/new-issues@alice", + ); + // The per-agent key joins the same list as a built-in's id. + assert.deepEqual( + withBuiltin(["closed"], memberKey("archetype:t/f", "a"), false), + ["archetype:t/f@a", "closed"], + ); +}); diff --git a/swarm-controller/src/agent_archetype.rs b/swarm-controller/src/agent_archetype.rs index ac3abfff0..22b226ca1 100644 --- a/swarm-controller/src/agent_archetype.rs +++ b/swarm-controller/src/agent_archetype.rs @@ -130,6 +130,24 @@ pub(crate) async fn of_agents( Ok(found) } +/// The archetype of every agent that has one, by agent name. +pub(crate) async fn all( + client: &async_nats::Client, +) -> anyhow::Result> { + use futures_util::TryStreamExt as _; + + let store = agent_archetype::open_or_create(client).await?; + let mut keys = store.keys().await?; + let mut found = std::collections::BTreeMap::new(); + while let Some(agent) = keys.try_next().await? { + let (archetype, _, _) = read_or_default::>(&store, &agent).await?; + if let Some(archetype) = archetype { + found.insert(agent, archetype); + } + } + Ok(found) +} + /// Delete a destroyed agent's key, so the next agent of that name is not /// built from this one's archetype. A failure is logged and does not stop the /// destroy, as in `agent_permissions::clear_on_destroy`. diff --git a/swarm-controller/src/notify_flows.rs b/swarm-controller/src/notify_flows.rs index e0ddbaacb..9c47b90fe 100644 --- a/swarm-controller/src/notify_flows.rs +++ b/swarm-controller/src/notify_flows.rs @@ -20,11 +20,14 @@ use axum::extract::rejection::JsonRejection; use axum::http::StatusCode; use problem_details::ProblemDetails; use serde::{Deserialize, Serialize}; +use std::collections::{BTreeMap, BTreeSet}; + +use swarm_queue_client::archetypes::{self, SourceRecord, flows_of}; use swarm_wire::notify_flows::{self, DecodeError, Document, Flow}; use utoipa::ToSchema; use crate::routes::{AppState, error_problem, wanted_error_status}; -use crate::schedules::{queue, read_with, refused, write_revision}; +use crate::schedules::{queue, read_or_default, read_with, refused, write_revision}; /// How many times a losing writer re-reads and re-applies, as in /// `wanted::MAX_ATTEMPTS`. @@ -36,17 +39,38 @@ pub(crate) struct FlowsView { /// The revision of the stored document, to send back in a `PUT`; `0` when /// none is stored. revision: u64, - /// The ids of the built-in flows that are switched off. + /// The ids of the built-in flows that are switched off, and the keys of the + /// archetype flows that are. disabled_builtins: Vec, /// The built-in flows this build ships, all of them, switched off or not. builtins: Vec, /// The custom flows. flows: Vec, + /// The flows archetypes carry, each with the agents it runs for. They follow + /// their archetype, so they are not in the document; a `PUT` switches one off + /// with its `key` in `disabled_builtins`. An older controller's answer has + /// none. + archetype_flows: Vec, /// What this build accepts: source events, fields, node kinds and limits, /// for an editor to offer. An older controller's answer has none. schema: notify_flows::Schema, } +/// One flow an archetype carries. +#[derive(Debug, Clone, PartialEq, Serialize, ToSchema)] +pub(crate) struct ArchetypeFlowView { + /// The archetype's name. + archetype: String, + /// What names the flow in `disabled_builtins` (off for every member), in the + /// router's log and in a todo's source. `@` in + /// `disabled_builtins` switches it off for one member. + key: String, + /// The flow as the archetype declares it, under its own `id`. + flow: Flow, + /// The agents of the archetype, by name. + members: Vec, +} + /// The body of a `PUT`: the whole document. #[derive(Debug, Clone, Deserialize, ToSchema)] #[serde(deny_unknown_fields)] @@ -64,12 +88,13 @@ pub(crate) struct PutFlows { } impl FlowsView { - fn of(doc: Document, revision: u64) -> Self { + fn of(doc: Document, revision: u64, archetype_flows: Vec) -> Self { Self { revision, disabled_builtins: doc.disabled_builtins, builtins: notify_flows::builtins(), flows: doc.flows, + archetype_flows, schema: notify_flows::schema(), } } @@ -152,12 +177,53 @@ pub(crate) async fn get_flows( Stored::Nothing => Ok(Json(FlowsView::of( Document::default(), revision.unwrap_or(0), + archetype_flows(&client).await, + ))), + Stored::Document(doc) => Ok(Json(FlowsView::of( + doc, + revision.unwrap_or(0), + archetype_flows(&client).await, ))), - Stored::Document(doc) => Ok(Json(FlowsView::of(doc, revision.unwrap_or(0)))), Stored::Unreadable(e) => Err(unreadable(&e)), } } +/// The flows the archetypes carry, with their agents. A read that fails is +/// logged and leaves the list empty: the document is still served. +async fn archetype_flows(client: &async_nats::Client) -> Vec { + let found = async { + let refs = crate::agent_archetype::all(client).await?; + let mut records = BTreeMap::new(); + if let Some(store) = archetypes::open(client).await? { + let sources: BTreeSet<&str> = refs.values().map(|r| r.source.as_str()).collect(); + for source in sources { + let (record, _, _) = read_or_default::(&store, source).await?; + records.insert(source.to_owned(), record); + } + } + anyhow::Ok(flows_of(&refs, &records).0) + } + .await; + match found { + Ok(found) => found + .into_iter() + .map(|f| ArchetypeFlowView { + key: f.key(), + archetype: f.archetype, + flow: f.flow, + members: f.members, + }) + .collect(), + Err(e) => { + tracing::warn!( + error = %format!("{e:#}"), + "notify flows: reading the archetype flows failed; none shown" + ); + Vec::new() + } + } +} + pub(crate) fn unreadable(why: &DecodeError) -> ProblemDetails { error_problem( StatusCode::CONFLICT, @@ -239,7 +305,11 @@ pub(crate) async fn put_flows( custom = doc.flows.len(), "set the notification router's flows" ); - return Ok(Json(FlowsView::of(doc, written))); + return Ok(Json(FlowsView::of( + doc, + written, + archetype_flows(&client).await, + ))); } } Err(queue_problem(&anyhow::anyhow!( @@ -465,4 +535,105 @@ mod tests { .await .expect("{number} is a field"); } + + /// The archetypes' flows are listed with their agents; their keys go in the + /// disabled list, a key for a flow that is gone is kept, and a key of no + /// valid shape is refused. + #[tokio::test] + async fn archetype_flows_are_listed_and_their_keys_are_saved() { + use swarm_queue_client::agent_archetype::ArchetypeRef; + use swarm_queue_client::archetypes::{ArchetypeMeta, BUILTIN_SOURCE, SourceRecord}; + + let nats = TestNats::start(); + let client = nats.client().await; + let state = crate::agent_limits::tests::state(&client); + + let mut members = custom("new-issues"); + members.nodes[1].kind = NodeKind::Todo(TodoSink { + to: Recipients::Members, + op: Op::Upsert, + key: "forge:{{ repo }}#{{ number }}:triage".to_owned(), + prefix: false, + text: "triage {{ repo }}#{{ number }}".to_owned(), + }); + let record = SourceRecord { + source: BUILTIN_SOURCE.to_owned(), + archetypes: [ + ( + "triage".to_owned(), + ArchetypeMeta { + description: "d".to_owned(), + flows: vec![members.clone()], + ..ArchetypeMeta::default() + }, + ), + ( + "reviewer".to_owned(), + ArchetypeMeta { + description: "d".to_owned(), + ..ArchetypeMeta::default() + }, + ), + ] + .into(), + refused: BTreeMap::default(), + }; + let records = swarm_queue_client::archetypes::open_or_create(&client) + .await + .expect("records"); + records + .put(BUILTIN_SOURCE, serde_json::to_vec(&record).unwrap().into()) + .await + .expect("put the record"); + let choices = swarm_queue_client::agent_archetype::open_or_create(&client) + .await + .expect("choices"); + for (agent, archetype) in [ + ("alice", "triage"), + ("bob", "triage"), + ("carol", "reviewer"), + ] { + let choice = ArchetypeRef { + source: BUILTIN_SOURCE.to_owned(), + name: archetype.to_owned(), + }; + choices + .put(agent, serde_json::to_vec(&choice).unwrap().into()) + .await + .expect("put the choice"); + } + + let Json(view) = get_flows(State(state.clone())).await.expect("get"); + assert_eq!(view.archetype_flows.len(), 1, "{:?}", view.archetype_flows); + let listed = &view.archetype_flows[0]; + assert_eq!(listed.key, "archetype:triage/new-issues"); + assert_eq!(listed.archetype, "triage"); + assert_eq!(listed.flow, members); + assert_eq!(listed.members, ["alice", "bob"]); + + // Off for everyone, off for one agent, and a key of a flow that is gone. + let keys = [ + "archetype:triage/new-issues", + "archetype:triage/new-issues@alice", + "archetype:gone/flow", + ]; + let Json(saved) = put_flows(State(state.clone()), Ok(body(0, &keys, vec![]))) + .await + .expect("the keys are saved"); + assert_eq!(saved.disabled_builtins, keys); + assert_eq!( + saved.archetype_flows.len(), + 1, + "a PUT answers with them too" + ); + for bad in ["archetype:triage", "archetype:/x", "archetype:a/b@", "nope"] { + let refused = put_flows( + State(state.clone()), + Ok(body(saved.revision, &[bad], vec![])), + ) + .await + .expect_err(bad); + assert_eq!(refused.status, Some(StatusCode::BAD_REQUEST), "{bad}"); + } + } } diff --git a/swarm-notify-router/src/api.rs b/swarm-notify-router/src/api.rs index e260e43c3..c717c422f 100644 --- a/swarm-notify-router/src/api.rs +++ b/swarm-notify-router/src/api.rs @@ -46,7 +46,7 @@ pub async fn serve(client: async_nats::Client) { tokio::spawn(async move { let Some(reply) = request.reply else { return }; let answer = if dry_run { - dry_run_answer(&js, &request.payload).await + dry_run_answer(&client, &js, &request.payload).await } else { deliveries_answer(&js, &request.payload).await }; @@ -91,7 +91,32 @@ async fn deliveries_answer(js: &jetstream::Context, payload: &[u8]) -> Vec { ) } -async fn dry_run_answer(js: &jetstream::Context, payload: &[u8]) -> Vec { +/// The archetypes' flows as they run under `document`'s switches. A bucket that +/// cannot be read leaves them out of the counts. +async fn archetype_flows( + client: &async_nats::Client, + js: &jetstream::Context, + document: &swarm_wire::notify_flows::Document, +) -> Vec { + let found = async { + let roster = crate::roster_of(js).await.map_err(|e| format!("{e:#}"))?; + crate::archetype_flows::load(client, &roster).await + } + .await; + match found { + Ok(found) => crate::archetype_flows::enabled(&found, &document.disabled_builtins), + Err(why) => { + tracing::warn!(%why, "notify router: a dry run without the archetype flows"); + Vec::new() + } + } +} + +async fn dry_run_answer( + client: &async_nats::Client, + js: &jetstream::Context, + payload: &[u8], +) -> Vec { body( async { let request: DryRunRequest = serde_json::from_slice(payload)?; @@ -100,6 +125,7 @@ async fn dry_run_answer(js: &jetstream::Context, payload: &[u8]) -> Vec { .check() .map_err(|why| anyhow::anyhow!("the document fails its check: {why}"))?; let (flows, _) = request.document.active(); + let archetype = archetype_flows(client, js, &request.document).await; let mut events: Vec<(String, Vec)> = log::newest(js, forge_events::STREAM, DRY_RUN_EVENTS) .await? @@ -109,7 +135,7 @@ async fn dry_run_answer(js: &jetstream::Context, payload: &[u8]) -> Vec { }) .collect(); events.reverse(); - Ok(dry_run::run(&flows, &events)) + Ok(dry_run::run(&flows, &archetype, &events)) } .await, ) diff --git a/swarm-notify-router/src/archetype_flows.rs b/swarm-notify-router/src/archetype_flows.rs index 3f1ebaf90..e42c117fd 100644 --- a/swarm-notify-router/src/archetype_flows.rs +++ b/swarm-notify-router/src/archetype_flows.rs @@ -6,6 +6,10 @@ //! `agent-archetype` key (`swarm_queue_client::agent_archetype`) names that //! archetype. //! +//! Each runs under its key (`archetype_flow_key`), and the document's +//! `disabled_builtins` switches it off for the archetype or for one member +//! ([`enabled`]). +//! //! The router reads both buckets again every [`REFRESH`], for the agents of the //! roster only, so an agent that is not in the swarm is no member. Each read it //! cannot make or decode leaves the flows it already has; before any read @@ -16,7 +20,7 @@ use std::time::{Duration, Instant}; use async_nats::jetstream::kv; use swarm_queue_client::agent_archetype::{self, Read}; -use swarm_queue_client::archetypes::{self, SourceRecord}; +use swarm_queue_client::archetypes::{self, SourceRecord, flows_of}; use swarm_wire::agent_archetype::ArchetypeRef; use crate::flow::ArchetypeFlow; @@ -119,49 +123,63 @@ async fn read( Ok((refs, records)) } -/// The flows to run: each flow of each archetype that has an agent, with those -/// agents, and why any flow was left out. A flow's id is prefixed with its -/// archetype's name, so it cannot be a built-in's or the document's, and no two -/// archetypes' collide. +/// The flows to run, see [`flows_of`], each under its key +/// (`swarm_wire::notify_flows::archetype_flow_key`) as its id: it cannot be a built-in's or the +/// document's, and no two archetypes' collide. fn build( refs: &BTreeMap, records: &BTreeMap, ) -> (Vec, Vec) { - let mut agents: BTreeMap<(&str, &str), Vec> = BTreeMap::new(); - for (agent, r) in refs { - agents - .entry((r.source.as_str(), r.name.as_str())) - .or_default() - .push(agent.clone()); - } - let mut active = Vec::new(); - let mut skipped = Vec::new(); - for ((source, name), members) in agents { - let Some(meta) = records - .get(source) - .and_then(|record| record.archetypes.get(name)) - else { - continue; - }; - let mut ids = BTreeSet::new(); - for flow in meta.flows.iter().filter(|f| f.enabled) { - if !ids.insert(flow.id.as_str()) { - skipped.push(format!("{name}: flow {}: its id is used twice", flow.id)); - } else if let Err(why) = flow.validate_archetype() { - skipped.push(format!("{name}: flow {}: {why}", flow.id)); - } else { - let mut flow = flow.clone(); - flow.id = format!("{name}:{}", flow.id); - active.push(ArchetypeFlow { - flow, - members: members.clone(), - }); + let (found, skipped) = flows_of(refs, records); + let active = found + .into_iter() + .map(|found| { + let mut flow = found.flow.clone(); + flow.id = found.key(); + ArchetypeFlow { + flow, + members: found.members, } - } - } + }) + .collect(); (active, skipped) } +/// `flows` less what `disabled`, the document's `disabled_builtins`, switches +/// off: a flow whose key is in it, and a member whose per-agent key is. A flow +/// left with no member is dropped. A key naming nothing there is ignored. +pub fn enabled(flows: &[ArchetypeFlow], disabled: &[String]) -> Vec { + flows + .iter() + .filter(|a| !disabled.contains(&a.flow.id)) + .filter_map(|a| { + let members: Vec = a + .members + .iter() + .filter(|m| !disabled.contains(&format!("{}@{m}", a.flow.id))) + .cloned() + .collect(); + (!members.is_empty()).then(|| ArchetypeFlow { + flow: a.flow.clone(), + members, + }) + }) + .collect() +} + +/// The flows the archetypes carry right now for `roster`, read from the +/// buckets. A bucket that cannot be read is an error. +pub async fn load( + client: &async_nats::Client, + roster: &BTreeSet, +) -> Result, String> { + let (refs, records) = read(client, roster).await?; + let (active, skipped) = build(&refs, &records); + for why in &skipped { + tracing::error!(%why, "notify router: skipping an archetype flow"); + } + Ok(active) +} #[cfg(test)] mod tests { use swarm_queue_client::archetypes::ArchetypeMeta; @@ -271,8 +289,8 @@ mod tests { assert_eq!( members(&built), [ - ("reviewer:new-issues", &carol[..]), - ("triage:new-issues", &alice_bob[..]) + ("archetype:reviewer/new-issues", &carol[..]), + ("archetype:triage/new-issues", &alice_bob[..]) ] ); } @@ -299,7 +317,7 @@ mod tests { .iter() .map(|a| a.flow.id.as_str()) .collect::>(), - ["triage:good"] + ["archetype:triage/good"] ); assert_eq!(built.1.len(), 2, "{:?}", built.1); assert!(built.1[0].contains("named"), "{:?}", built.1); @@ -356,9 +374,91 @@ mod tests { assert_eq!( members(&(active, skipped)), [( - "triage:new-issues", + "archetype:triage/new-issues", &["alice".to_owned(), "bob".to_owned()][..] )] ); } + + fn two_flows() -> Vec { + let records = record(&[ + ("triage", vec![flow("new-issues", Recipients::Members)]), + ("reviewer", vec![flow("new-issues", Recipients::Members)]), + ]); + build( + &refs(&[ + ("alice", "triage"), + ("bob", "triage"), + ("carol", "reviewer"), + ]), + &records, + ) + .0 + } + + fn who(flows: &[ArchetypeFlow]) -> Vec<(String, Vec<&str>)> { + flows + .iter() + .map(|a| { + ( + a.flow.id.clone(), + a.members.iter().map(String::as_str).collect(), + ) + }) + .collect() + } + + fn off(keys: &[&str]) -> Vec { + keys.iter().map(|k| (*k).to_owned()).collect() + } + + /// A flow's key switches it off for its archetype, the key with `@agent` + /// for that agent alone, and a key for something that is gone does nothing. + #[test] + fn the_disabled_list_switches_off_a_flow_or_one_agent_of_it() { + let flows = two_flows(); + // Control: nothing off. + assert_eq!( + who(&enabled(&flows, &[])), + [ + ("archetype:reviewer/new-issues".to_owned(), vec!["carol"]), + ( + "archetype:triage/new-issues".to_owned(), + vec!["alice", "bob"] + ), + ] + ); + // The archetype's key: the other archetype's flow of the same id stays. + assert_eq!( + who(&enabled(&flows, &off(&["archetype:triage/new-issues"]))), + [("archetype:reviewer/new-issues".to_owned(), vec!["carol"])] + ); + // One agent's key: the other members still get the flow. + assert_eq!( + who(&enabled( + &flows, + &off(&["archetype:triage/new-issues@alice"]) + )), + [ + ("archetype:reviewer/new-issues".to_owned(), vec!["carol"]), + ("archetype:triage/new-issues".to_owned(), vec!["bob"]), + ] + ); + // Every member off drops the flow; stale keys and built-in ids change nothing. + assert_eq!( + who(&enabled( + &flows, + &off(&[ + "archetype:reviewer/new-issues@carol", + "archetype:triage/gone", + "archetype:triage/new-issues@dave", + "closed" + ]) + )), + [( + "archetype:triage/new-issues".to_owned(), + vec!["alice", "bob"] + )] + ); + } } diff --git a/swarm-notify-router/src/dry_run.rs b/swarm-notify-router/src/dry_run.rs index d0c7dd14d..596aabef9 100644 --- a/swarm-notify-router/src/dry_run.rs +++ b/swarm-notify-router/src/dry_run.rs @@ -12,12 +12,18 @@ use swarm_wire::notify_flows::Flow; use swarm_wire::router_api::{DryRun, FlowHits, NodeHits}; use crate::event::{Delivery, Snapshot, fields}; +use crate::flow::ArchetypeFlow; use crate::{flow, route}; -/// `flows` over `events`, oldest first, each a forge event type and body. -/// An event that names no issue or PR is not counted. -pub fn run(flows: &[Flow], events: &[(String, Vec)]) -> DryRun { - let mut hits: Vec> = flows +/// `flows` and the archetypes' `archetype` flows over `events`, oldest first, +/// each a forge event type and body. An event that names no issue or PR is not +/// counted. +pub fn run(flows: &[Flow], archetype: &[ArchetypeFlow], events: &[(String, Vec)]) -> DryRun { + let all: Vec<&Flow> = flows + .iter() + .chain(archetype.iter().map(|a| &a.flow)) + .collect(); + let mut hits: Vec> = all .iter() .map(|f| f.nodes.iter().map(|n| (n.id.as_str(), 0)).collect()) .collect(); @@ -31,19 +37,19 @@ pub fn run(flows: &[Flow], events: &[(String, Vec)]) -> DryRun { let key = delivery.state_key(); let before = snapshots.remove(&key).unwrap_or_default(); let event_fields = fields(event_type, &delivery, &before); - for (flow, counts) in flows.iter().zip(&mut hits) { + for (flow, counts) in all.iter().zip(&mut hits) { for id in flow::reached(flow, event_type, &event_fields) { if let Some(count) = counts.get_mut(id) { *count += 1; } } } - let routed = route::route(event_type, &delivery, &before, flows, &[], |_| true); + let routed = route::route(event_type, &delivery, &before, flows, archetype, |_| true); snapshots.insert(key, routed.next); } DryRun { events: run, - flows: flows + flows: all .iter() .zip(hits) .map(|(flow, counts)| FlowHits { @@ -102,7 +108,7 @@ mod tests { ("issues".to_owned(), body("closed", 5, &[])), ("issues".to_owned(), b"not json".to_vec()), ]; - let r = run(&flows, &events); + let r = run(&flows, &[], &events); assert_eq!(r.events, 4); // The source sees all three assign events, the match and the sink both // assigned ones, though the second adds nobody. @@ -160,7 +166,7 @@ mod tests { edges: vec![edge("source", "atlas")], }; let assign = ("issue_assign".to_owned(), body("assigned", 5, &["atlas"])); - let r = run(&[flow], &[assign.clone(), assign]); + let r = run(&[flow], &[], &[assign.clone(), assign]); assert_eq!(count(&r, "added", "source"), 2); assert_eq!(count(&r, "added", "atlas"), 1); } @@ -173,11 +179,43 @@ mod tests { ..Document::default() }; let (flows, _) = doc.active(); - let r = run(&flows, &[]); + let r = run(&flows, &[], &[]); assert_eq!(r.events, 0); let listed: Vec<&str> = r.flows.iter().map(|f| f.flow.as_str()).collect(); assert!(!listed.contains(&"closed")); assert!(listed.contains(&"assigned")); assert_eq!(listed.len(), builtins().len() - 1); } + + /// An archetype's flow is counted next to the document's, under its key. + #[test] + fn an_archetype_flow_is_counted_under_its_key() { + let (flows, _) = Document::default().active(); + let mut flow = builtins().remove(0); + flow.id = "archetype:triage/new".to_owned(); + flow.nodes[2].kind = + swarm_wire::notify_flows::NodeKind::Todo(swarm_wire::notify_flows::TodoSink { + to: swarm_wire::notify_flows::Recipients::Members, + op: swarm_wire::router_todo::Op::Upsert, + key: "forge:{{ repo }}#{{ number }}:triage".to_owned(), + prefix: false, + text: "triage".to_owned(), + }); + let archetype = [ArchetypeFlow { + flow, + members: vec!["alice".to_owned()], + }]; + let events = vec![("issue_assign".to_owned(), body("assigned", 5, &["atlas"]))]; + let r = run(&flows, &archetype, &events); + assert_eq!(count(&r, "archetype:triage/new", "source"), 1); + assert_eq!(count(&r, "archetype:triage/new", "sink"), 1); + assert_eq!(count(&r, "assigned", "sink"), 1, "the document's flow too"); + // Control: with no archetype flow it has no entry. + assert!( + run(&flows, &[], &events) + .flows + .iter() + .all(|f| f.flow != "archetype:triage/new") + ); + } } diff --git a/swarm-notify-router/src/main.rs b/swarm-notify-router/src/main.rs index 87c36d984..e3db39f03 100644 --- a/swarm-notify-router/src/main.rs +++ b/swarm-notify-router/src/main.rs @@ -239,12 +239,14 @@ impl Router { self.flows.refresh(&self.js).await; let roster = self.roster().await.map_err(Skip::Retry)?; self.archetype_flows.refresh(&self.nats, &roster).await; + let archetype_flows = + archetype_flows::enabled(&self.archetype_flows.active, &self.flows.disabled); let routed = route::route( &event_type, &delivery, &snapshot, &self.flows.active, - &self.archetype_flows.active, + &archetype_flows, |a| roster.contains(a), ); let flows: BTreeSet<&str> = routed @@ -364,32 +366,37 @@ impl Router { { return Ok(agents.clone()); } - let bucket = swarm_wire::agent_status::BUCKET; - let stream = self - .js - .get_stream(format!("KV_{bucket}")) - .await - .context("opening the agent-status bucket")?; - let subjects: Vec<(String, usize)> = stream - .info_with_subjects(format!("$KV.{bucket}.>")) - .await - .context("listing agent-status keys")? - .try_collect() - .await - .context("listing agent-status keys")?; - let prefix = format!("$KV.{bucket}."); - let agents: BTreeSet = subjects - .into_iter() - .filter_map(|(subject, _)| { - let key = subject.strip_prefix(&prefix)?; - swarm_wire::agent_status::split_key(key).map(|(_, agent)| agent.to_owned()) - }) - .collect(); + let agents = roster_of(&self.js).await?; self.roster = Some((Instant::now(), agents.clone())); Ok(agents) } } +/// The agents of this swarm, read from the agent-status bucket. +pub(crate) async fn roster_of(js: &jetstream::Context) -> Result> { + let bucket = swarm_wire::agent_status::BUCKET; + let stream = js + .get_stream(format!("KV_{bucket}")) + .await + .context("opening the agent-status bucket")?; + let subjects: Vec<(String, usize)> = stream + .info_with_subjects(format!("$KV.{bucket}.>")) + .await + .context("listing agent-status keys")? + .try_collect() + .await + .context("listing agent-status keys")?; + let prefix = format!("$KV.{bucket}."); + let agents: BTreeSet = subjects + .into_iter() + .filter_map(|(subject, _)| { + let key = subject.strip_prefix(&prefix)?; + swarm_wire::agent_status::split_key(key).map(|(_, agent)| agent.to_owned()) + }) + .collect(); + Ok(agents) +} + /// Why an event was not handled. enum Skip { /// It can never be handled; dropped for good. diff --git a/swarm-notify-router/src/route.rs b/swarm-notify-router/src/route.rs index fbd575668..7280a9cdb 100644 --- a/swarm-notify-router/src/route.rs +++ b/swarm-notify-router/src/route.rs @@ -738,4 +738,85 @@ mod tests { ] ); } + + /// An archetype edit changes what its member receives at the next read of + /// the archetype record, with no flow document stored at all. + #[tokio::test] + async fn an_archetype_edit_changes_what_a_member_receives() { + use swarm_queue_client::agent_archetype::{self, ArchetypeRef}; + use swarm_queue_client::archetypes::{self, ArchetypeMeta, SourceRecord}; + + let nats = crate::test_nats::TestNats::start(); + let client = nats.client().await; + let records = archetypes::open_or_create(&client).await.expect("records"); + let choices = agent_archetype::open_or_create(&client) + .await + .expect("choices"); + let choice = ArchetypeRef { + source: "hyperhive".to_owned(), + name: "triage".to_owned(), + }; + choices + .put("alice", serde_json::to_vec(&choice).unwrap().into()) + .await + .expect("put the choice"); + let roster = std::collections::BTreeSet::from(["alice".to_owned()]); + + let publish = |text: &'static str| { + let records = records.clone(); + async move { + let mut flow = members_flow(); + flow.id = "new".to_owned(); + if let NodeKind::Todo(sink) = &mut flow.nodes[2].kind { + sink.text = text.to_owned(); + } + let record = SourceRecord { + source: "hyperhive".to_owned(), + archetypes: [( + "triage".to_owned(), + ArchetypeMeta { + description: "d".to_owned(), + flows: vec![flow], + ..ArchetypeMeta::default() + }, + )] + .into(), + refused: std::collections::BTreeMap::default(), + }; + records + .put("hyperhive", serde_json::to_vec(&record).unwrap().into()) + .await + .expect("put the record"); + } + }; + let received = || async { + let flows = crate::archetype_flows::load(&client, &roster) + .await + .expect("a read"); + let routed = super::route( + "issue_assign", + &delivery("assigned", &[]), + &Snapshot::default(), + &[], + &flows, + |_| true, + ); + routed + .actions + .iter() + .map(|a| (a.agent.clone(), a.op.summary.clone())) + .collect::>() + }; + + publish("first wording").await; + assert_eq!( + received().await, + [("alice".to_owned(), "first wording".to_owned())] + ); + publish("second wording").await; + assert_eq!( + received().await, + [("alice".to_owned(), "second wording".to_owned())] + ); + } } diff --git a/swarm-notify-router/src/stored.rs b/swarm-notify-router/src/stored.rs index 7aa640961..263f542a4 100644 --- a/swarm-notify-router/src/stored.rs +++ b/swarm-notify-router/src/stored.rs @@ -41,6 +41,9 @@ pub enum Read { /// The flows in force, and which revision of the key they came from. pub struct Flows { pub active: Vec, + /// The document's `disabled_builtins`, which also switches off archetype + /// flows ([`crate::archetype_flows::enabled`]). + pub disabled: Vec, /// The revision `active` was read from; `None` for the built-ins alone. revision: Option, /// The last problem logged, so a standing one is logged once. @@ -53,6 +56,7 @@ impl Flows { pub fn builtins() -> Self { Self { active: Document::default().active().0, + disabled: Vec::new(), revision: None, problem: None, checked: None, @@ -70,6 +74,7 @@ impl Flows { ); } self.active = Document::default().active().0; + self.disabled.clear(); self.problem = None; } Read::Document { revision, .. } if self.revision == Some(revision) => {} @@ -92,6 +97,7 @@ impl Flows { "notify router: running the flows document" ); self.active = active; + self.disabled = doc.disabled_builtins; self.revision = Some(revision); self.problem = None; } @@ -285,4 +291,22 @@ mod tests { }); assert_eq!(ids(&f), BUILTIN_IDS); } + + /// The document's disabled list is kept whole, archetype keys among the + /// built-in ids, and goes with the document: a deleted one clears it. + #[test] + fn the_disabled_list_includes_archetype_keys_and_goes_with_the_document() { + let mut f = Flows::builtins(); + assert!(f.disabled.is_empty()); + f.apply(at(1, doc(&["closed", "archetype:triage/new@alice"], &[]))); + assert_eq!(f.disabled, ["closed", "archetype:triage/new@alice"]); + f.apply(junk(2, b"not json")); + assert_eq!( + f.disabled.len(), + 2, + "a document it cannot read changes nothing" + ); + f.apply(Read::Absent); + assert!(f.disabled.is_empty()); + } } diff --git a/swarm-queue-client/src/archetypes.rs b/swarm-queue-client/src/archetypes.rs index 03de4b7e9..bceeadbc6 100644 --- a/swarm-queue-client/src/archetypes.rs +++ b/swarm-queue-client/src/archetypes.rs @@ -9,11 +9,12 @@ //! result here, so the create screen lists archetypes from a value that a //! create would accept. -use std::collections::BTreeMap; +use std::collections::{BTreeMap, BTreeSet}; +use swarm_wire::agent_archetype::ArchetypeRef; use swarm_wire::agent_inference::AgentInference; use swarm_wire::agent_limits::AgentCaps; -use swarm_wire::notify_flows::Flow; +use swarm_wire::notify_flows::{Flow, archetype_flow_key}; /// The KV bucket archetype sources are published into, keyed by source name. pub const BUCKET: &str = "agent-archetypes"; @@ -70,6 +71,67 @@ pub struct ArchetypeMeta { pub flows: Vec, } +/// One flow an archetype carries, with the agents of that archetype. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct ArchetypeFlow { + pub archetype: String, + /// As the archetype declares it, under its own `id`. + pub flow: Flow, + /// The agents whose archetype this is, by name. + pub members: Vec, +} + +impl ArchetypeFlow { + /// The name the flow goes by, see [`archetype_flow_key`]. + #[must_use] + pub fn key(&self) -> String { + archetype_flow_key(&self.archetype, &self.flow.id) + } +} + +/// The flows to run: each enabled flow of each archetype that has an agent +/// in `refs` (agent to archetype), with those agents, and why any flow was +/// left out. A flow that fails [`Flow::validate_archetype`] or repeats an id +/// of its archetype is left out. +#[must_use] +pub fn flows_of( + refs: &BTreeMap, + records: &BTreeMap, +) -> (Vec, Vec) { + let mut agents: BTreeMap<(&str, &str), Vec> = BTreeMap::new(); + for (agent, r) in refs { + agents + .entry((r.source.as_str(), r.name.as_str())) + .or_default() + .push(agent.clone()); + } + let mut active = Vec::new(); + let mut skipped = Vec::new(); + for ((source, name), members) in agents { + let Some(meta) = records + .get(source) + .and_then(|record| record.archetypes.get(name)) + else { + continue; + }; + let mut ids = BTreeSet::new(); + for flow in meta.flows.iter().filter(|f| f.enabled) { + if !ids.insert(flow.id.as_str()) { + skipped.push(format!("{name}: flow {}: its id is used twice", flow.id)); + } else if let Err(why) = flow.validate_archetype() { + skipped.push(format!("{name}: flow {}: {why}", flow.id)); + } else { + active.push(ArchetypeFlow { + archetype: name.to_owned(), + flow: flow.clone(), + members: members.clone(), + }); + } + } + } + (active, skipped) +} + /// Open the archetype bucket for writing, creating it if nothing has yet. /// /// **Controller-side only.** `history: 1`: a reader wants the current value. diff --git a/swarm-wire/src/notify_flows.rs b/swarm-wire/src/notify_flows.rs index 0ea5d4a8f..8c47a0991 100644 --- a/swarm-wire/src/notify_flows.rs +++ b/swarm-wire/src/notify_flows.rs @@ -27,6 +27,41 @@ pub const BUCKET: &str = "notify-flows"; /// The one key in [`BUCKET`]. pub const KEY: &str = "all"; +/// The key an archetype's flow `flow` of `archetype` goes by: in the +/// document's `disabled_builtins` it switches the flow off for every member, +/// and it names the flow in the router's log, a todo's source and the hit +/// counts. `flow` is the `id` the archetype declares. +#[must_use] +pub fn archetype_flow_key(archetype: &str, flow: &str) -> String { + format!("archetype:{archetype}/{flow}") +} + +/// [`archetype_flow_key`] followed by `@agent`: in `disabled_builtins` it +/// switches the flow off for that one member. +#[must_use] +pub fn archetype_member_key(archetype: &str, flow: &str, agent: &str) -> String { + format!("{}@{agent}", archetype_flow_key(archetype, flow)) +} + +/// Whether `key` has the shape of [`archetype_flow_key`] or +/// [`archetype_member_key`]. Whether such a flow, or agent, exists is not +/// asked: a key for one that is gone is ignored, so editing an archetype +/// cannot make the stored document fail its check. +#[must_use] +pub fn is_archetype_key(key: &str) -> bool { + let Some(rest) = key.strip_prefix("archetype:") else { + return false; + }; + let Some((archetype, rest)) = rest.split_once('/') else { + return false; + }; + let (flow, agent) = match rest.split_once('@') { + Some((flow, agent)) => (flow, Some(agent)), + None => (rest, None), + }; + is_id(archetype) && is_id(flow) && agent.is_none_or(is_id) +} + /// The document layout this build reads and writes. Version 1 wrote a todo's /// key and text with `{field}` placeholders; version 2 writes them as Jinja /// templates (`{{ field }}`). [`Document::decode`] converts a version 1 @@ -158,7 +193,8 @@ pub struct Document { /// leaves the document alone. #[serde(default = "version")] pub version: u32, - /// The ids of the built-in flows that are switched off. + /// The ids of the built-in flows that are switched off, and the keys of the + /// archetype flows that are ([`archetype_flow_key`], [`archetype_member_key`]). #[serde(default)] pub disabled_builtins: Vec, /// The flows the operator defined. @@ -338,9 +374,9 @@ impl Document { } let builtin = builtins(); for id in &self.disabled_builtins { - if !builtin.iter().any(|b| b.id == *id) { + if !builtin.iter().any(|b| b.id == *id) && !is_archetype_key(id) { return Err(format!( - "disabled_builtins names {id}, which is no built-in flow" + "disabled_builtins names {id}, which is no built-in flow and no archetype flow key" )); } } @@ -1802,4 +1838,35 @@ mod tests { assert!(f.validate_archetype().is_err(), "{key} {text}"); } } + + /// The keys that switch an archetype's flow off join `disabled_builtins`; + /// one that names nothing there is still accepted, one of no valid shape is not. + #[test] + fn archetype_flow_keys_are_accepted_in_the_disabled_list() { + let flow = archetype_flow_key("triage", "new-issues"); + let member = archetype_member_key("triage", "new-issues", "alice"); + assert_eq!(flow, "archetype:triage/new-issues"); + assert_eq!(member, "archetype:triage/new-issues@alice"); + let with = |keys: &[&str]| Document { + disabled_builtins: keys.iter().map(|k| (*k).to_owned()).collect(), + ..Document::default() + }; + assert_eq!(with(&[&flow, &member, "closed"]).check(), Ok(())); + assert_eq!(with(&["archetype:gone/flow"]).check(), Ok(()), "stale"); + for bad in [ + "archetype:triage", + "archetype:/x", + "archetype:a/", + "archetype:a/b@", + "archetype:a b/c", + "archetype:a/b/c", + "triage/new-issues", + "no-such-builtin", + ] { + assert!(with(&[bad]).check().is_err(), "{bad}"); + assert!(!is_archetype_key(bad), "{bad}"); + } + // A stale key leaves the built-ins as they were. + assert_eq!(with(&[&flow]).active().0.len(), builtins().len()); + } }