mara/hyperhive
mara/hyperhive
Watch
0
0
Fork
You've already forked hyperhive
 0
Code Activity

archetype flows behave like the built-ins: switch off per archetype or per agent, board cards, duplicate

Browse source 
The document's disabled list takes archetype:<archetype>/<flow> (off for every member) and the same with @<agent> (off for one member); a key naming nothing is ignored. GET /api/notify/flows lists the archetype flows with their members, the board draws a card for each with a switch, a checkbox per agent and duplicate, and the dry run counts their hits. The flows keep following the archetype.
This commit is contained in:
atlas 2026-11-11 19:08:19 +01:00 • committed by mara
commit 68c24643f6
20 changed files with 908 additions and 102 deletions
Download patch file Download diff file Expand all files Collapse all files

22
docs/2-handbook/swarm/archetypes.md
View file

@ -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:<archetype>/<flow id>` 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:<archetype>/<flow id>@<agent>` 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

20
docs/2-handbook/swarm/ui.md
View file

@ -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/<name>/terminal` | one agent's live terminal |
| `/agents/<name>/subagents/<subagent>/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/<name>/terminal` | one agent's live terminal |
| `/agents/<name>/subagents/<subagent>/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

28
docs/4-implementation/swarm/notify-router.md
View file

@ -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 `<archetype>:<id>`, so its todos' source is
`router:<archetype>:<id>`. A recipient not in the roster gets nothing, the same as a named agent that doesn't exist.
Each runs under its key, `archetype:<archetype>/<id>`, which is its flow id in the log, the hit counts and a todo's source (`router:archetype:<archetype>/<id>`). 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:<archetype>/<id>`) switches it off for every member, and the key
followed by `@<agent>` 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 `@<agent>`) and `duplicate`, which
opens the editor on `?from=<key>`. 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

11
docs/4-implementation/web-ui/swarm-ui.md
View file

@ -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:<archetype>/<flow id>` into the
document's `disabled_builtins`, and a checkbox the same key followed by `@<agent>`.
**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

25
frontend/packages/swarm-ui/src/pages/notifications/FlowDraft.ts
View file

@ -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,
),
};
}

19
frontend/packages/swarm-ui/src/pages/notifications/FlowTypes.ts
View file

@ -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;
}

8
frontend/packages/swarm-ui/src/pages/notifications/NotificationsPage.css
View file

@ -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;
}

59
frontend/packages/swarm-ui/src/pages/notifications/NotificationsPage.tsx
View file

@ -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 (<FlowEditorPage>).
// 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 (<FlowEditorPage>).
// 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() {
</Card>
);
})}
{(view.archetype_flows ?? []).length > 0 ? (
<h3>archetype flows</h3>
) : null}
{(view.archetype_flows ?? []).map((af) => {
const on = !view.disabled_builtins.includes(af.key);
return (
<Card key={af.key} class={on ? "" : "notify-off"}>
<div class="notify-flow-head">
<span class="notify-flow-name">{af.flow.name}</span>
<span class="notify-flow-state">
archetype {af.archetype}
</span>
<CheckboxField
id={`archetype-${af.key}`}
label={on ? "on" : "off"}
checked={on}
disabled={pending}
onChange={(checked) => setBuiltin(af.key, checked)}
/>
<ActionRow class="notify-flow-actions">
<Button onClick={() => draftFrom(af.key)}>duplicate</Button>
</ActionRow>
</div>
<div class="notify-flow-agents">
<span>agents:</span>
{af.members.map((agent) => (
<CheckboxField
key={agent}
id={`archetype-${af.key}-${agent}`}
label={agent}
checked={
!view.disabled_builtins.includes(
memberKey(af.key, agent),
)
}
disabled={pending}
onChange={(checked) =>
setBuiltin(memberKey(af.key, agent), checked)
}
/>
))}
</div>
<FlowGraph
flow={withMembersAs(af.flow, `members of ${af.archetype}`)}
hits={hitsOf(shown, af.key)}
/>
</Card>
);
})}
<h3>custom flows</h3>
{view.flows.length === 0 ? <p>no custom flows</p> : null}
{view.flows.map((flow) => (

41
frontend/packages/swarm-ui/test/flow-draft.test.mjs
View file

@ -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,

13
frontend/packages/swarm-ui/test/flow-types.test.mjs
View file

@ -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"],
);
});

18
swarm-controller/src/agent_archetype.rs
View file

@ -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<std::collections::BTreeMap<String, ArchetypeRef>> {
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::<Option<ArchetypeRef>>(&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`.

181
swarm-controller/src/notify_flows.rs
View file

@ -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<String>,
/// The built-in flows this build ships, all of them, switched off or not.
builtins: Vec<Flow>,
/// The custom flows.
flows: Vec<Flow>,
/// 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<ArchetypeFlowView>,
/// 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. `<key>@<agent>` 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<String>,
}
/// 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<ArchetypeFlowView>) -> 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<ArchetypeFlowView> {
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::<SourceRecord>(&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}");
}
}
}

32
swarm-notify-router/src/api.rs
View file

@ -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<u8> {
)
}
async fn dry_run_answer(js: &jetstream::Context, payload: &[u8]) -> Vec<u8> {
/// 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<crate::flow::ArchetypeFlow> {
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<u8> {
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<u8> {
.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<u8>)> =
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<u8> {
})
.collect();
events.reverse();
Ok(dry_run::run(&flows, &events))
Ok(dry_run::run(&flows, &archetype, &events))
}
.await,
)

180
swarm-notify-router/src/archetype_flows.rs
View file

@ -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<String, ArchetypeRef>,
records: &BTreeMap<String, SourceRecord>,
) -> (Vec<ArchetypeFlow>, Vec<String>) {
let mut agents: BTreeMap<(&str, &str), Vec<String>> = 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<ArchetypeFlow> {
flows
.iter()
.filter(|a| !disabled.contains(&a.flow.id))
.filter_map(|a| {
let members: Vec<String> = 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<String>,
) -> Result<Vec<ArchetypeFlow>, 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::<Vec<_>>(),
["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<ArchetypeFlow> {
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<String> {
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"]
)]
);
}
}

58
swarm-notify-router/src/dry_run.rs
View file

@ -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<u8>)]) -> DryRun {
let mut hits: Vec<BTreeMap<&str, u64>> = 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<u8>)]) -> DryRun {
let all: Vec<&Flow> = flows
.iter()
.chain(archetype.iter().map(|a| &a.flow))
.collect();
let mut hits: Vec<BTreeMap<&str, u64>> = 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<u8>)]) -> 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")
);
}
}

51
swarm-notify-router/src/main.rs
View file

@ -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<String> = 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<BTreeSet<String>> {
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<String> = 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.

81
swarm-notify-router/src/route.rs
View file

@ -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::<Vec<_>>()
};
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())]
);
}
}

24
swarm-notify-router/src/stored.rs
View file

@ -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<Flow>,
/// The document's `disabled_builtins`, which also switches off archetype
/// flows ([`crate::archetype_flows::enabled`]).
pub disabled: Vec<String>,
/// The revision `active` was read from; `None` for the built-ins alone.
revision: Option<u64>,
/// 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());
}
}

66
swarm-queue-client/src/archetypes.rs
View file

@ -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<Flow>,
}
/// 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<String>,
}
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<String, ArchetypeRef>,
records: &BTreeMap<String, SourceRecord>,
) -> (Vec<ArchetypeFlow>, Vec<String>) {
let mut agents: BTreeMap<(&str, &str), Vec<String>> = 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.

73
swarm-wire/src/notify_flows.rs
View file

@ -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<String>,
/// 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());
}
}