diff --git a/crates/plugin-manager/src/lib.rs b/crates/plugin-manager/src/lib.rs index 63886dd3..72c2f776 100644 --- a/crates/plugin-manager/src/lib.rs +++ b/crates/plugin-manager/src/lib.rs @@ -69,6 +69,8 @@ pub fn bundled_manifests() -> Vec { .expect("projects manifest"), serde_json::from_str(include_str!("../../../src/bundled/agents/manifest.json")) .expect("agents manifest"), + serde_json::from_str(include_str!("../../../src/bundled/workflows/manifest.json")) + .expect("workflows manifest"), ] } fn is_bundled(id: &str) -> bool { diff --git a/dev/relay-broker.mjs b/dev/relay-broker.mjs index 4df3a188..d3203a30 100644 --- a/dev/relay-broker.mjs +++ b/dev/relay-broker.mjs @@ -1,3 +1,12 @@ +import { + validateWorkflowEvent, + WORKFLOW_KINDS, +} from "../src/features/workflows/protocol.ts"; +import { + workflowRunsPath, + workflowReadText, +} from "../src/features/workflows/http.ts"; +import { readReceiptText } from "../src/features/relay/receipt.ts"; import { decodeAgentObserver } from "./agent-observer.mjs"; import { observerGeneration } from "../src/features/agents/observer.ts"; import { @@ -358,6 +367,11 @@ export function relayBrokerPlugin({ ) return json(res, 403, { error: "Origin rejected" }); const url = new URL(req.url, origin); + // Own cancellation before awaiting the request body, signing or dispatch. + const cancel = new AbortController(); + const release = () => cancel.abort(); + res.once("close", release); + if (res.destroyed) release(); try { if (url.pathname === "/api/relay/register" && req.method === "POST") { let raw = ""; @@ -518,18 +532,20 @@ export function relayBrokerPlugin({ }); } } - if (route === "/api/relay/session" && req.method === "GET") + if (route === "/api/relay/session" && req.method === "GET") { return json(res, 200, { viewer, ...(await getAuthority(relay)), relayUrl: relay, - writeKinds: [7, 9], + writeKinds: [7, 9, ...WORKFLOW_KINDS], + workflowReads: true, sidebarPreferences: true, readState: true, agentLibrary: true, live: true, agentActivity: true, }); + } if ( [ "/api/relay/stream-retry", @@ -734,6 +750,7 @@ export function relayBrokerPlugin({ "/api/relay/claim", "/api/relay/accept-policy", "/api/relay/gifs", + "/api/relay/workflow-runs", ].includes(route) || req.method !== "POST" ) @@ -750,6 +767,17 @@ export function relayBrokerPlugin({ } catch { return json(res, 400, { error: "Filter body is not JSON" }); } + let workflowPath; + if (route === "/api/relay/workflow-runs") { + try { + workflowPath = workflowRunsPath(filters); + } catch { + return json(res, 400, { + error: "Invalid workflow read", + sent: false, + }); + } + } const profile = route === "/api/relay/profile"; const claim = route === "/api/relay/claim"; const policy = route === "/api/relay/accept-policy"; @@ -848,8 +876,23 @@ export function relayBrokerPlugin({ const signing = route === "/api/relay/sign"; const publishing = route === "/api/relay/publish"; if (signing || publishing) { - if (!validMessageTemplate(filters)) + if (![7, 9].includes(filters?.kind)) { + try { + validateWorkflowEvent( + { ...filters, pubkey: signing ? viewer : filters.pubkey }, + viewer, + ); + } catch { + cancel.signal.throwIfAborted(); + return json(res, 400, { + error: "Workflow operation unavailable or invalid", + sent: false, + }); + } + } else if (!validMessageTemplate(filters)) return json(res, 400, { error: "Message rejected" }); + // Never sign or publish after the requesting browser has left. + cancel.signal.throwIfAborted(); if (signing) { const started = performance.now(); const event = finalizeEvent( @@ -874,6 +917,7 @@ export function relayBrokerPlugin({ !claim && !policy && !gifs && + !workflowPath && !readPublishing && !snapshot && !validFilters(filters) @@ -882,15 +926,18 @@ export function relayBrokerPlugin({ const gifSearchPath = gifs ? await getGifSearchPath(relay) : null; if (gifs && !gifSearchPath) return json(res, 404, { error: "GIF search is unavailable" }); - const upstreamPath = gifs - ? gifSearchPath - : profile || publishing || readPublishing - ? "/events" - : claim - ? "/api/invites/claim" - : policy - ? "/api/invites/accept-policy" - : "/query"; + const upstreamPath = + workflowPath ?? + (gifs + ? gifSearchPath + : profile || publishing || readPublishing + ? "/events" + : claim + ? "/api/invites/claim" + : policy + ? "/api/invites/accept-policy" + : "/query"); + const method = workflowPath ? "GET" : "POST"; if (inflight >= MAX_INFLIGHT) return json(res, 429, { error: "Query concurrency limit", @@ -899,121 +946,120 @@ export function relayBrokerPlugin({ inflight++; try { const lane = admissions(relay, viewer).api; - const body = JSON.stringify(filters); - // A browser that gave up (the client's ten-second deadline) must also release - // this upstream request, or hung requests exhaust the inflight budget. - const cancel = new AbortController(); - const release = () => cancel.abort(); - res.once("close", release); + const body = workflowPath ? undefined : JSON.stringify(filters); const admissionStart = performance.now(); let connectsBefore, upstreamStart; let response; - try { - const requestSignal = AbortSignal.any([ - cancel.signal, - AbortSignal.timeout(UPSTREAM_TIMEOUT_MS), - ]); - response = await admittedApiRequest( - lane, - () => { - // Auth freshness and network timings begin at dispatch, not queue entry. - requestSignal.throwIfAborted(); - timings.push( - `admission;dur=${(performance.now() - admissionStart).toFixed(2)}`, - ); - const authStart = performance.now(); - const auth = finalizeEvent( - { - kind: 27235, - created_at: Math.floor(Date.now() / 1000), - content: "", - tags: [ - ["u", `${relay}${upstreamPath}`], - ["method", "POST"], - [ - "payload", - createHash("sha256").update(body).digest("hex"), - ], - ["nonce", randomBytes(16).toString("hex")], - ], - }, - key, - ); + const requestSignal = AbortSignal.any([ + cancel.signal, + AbortSignal.timeout(UPSTREAM_TIMEOUT_MS), + ]); + response = await admittedApiRequest( + lane, + () => { + // Auth freshness and network timings begin at dispatch, not queue entry. + requestSignal.throwIfAborted(); + timings.push( + `admission;dur=${(performance.now() - admissionStart).toFixed(2)}`, + ); + const authStart = performance.now(); + const auth = finalizeEvent( + { + kind: 27235, + created_at: Math.floor(Date.now() / 1000), + content: "", + tags: [ + ["u", `${relay}${upstreamPath}`], + ["method", method], + ...(body === undefined + ? [] + : [ + [ + "payload", + createHash("sha256").update(body).digest("hex"), + ], + ]), + ["nonce", randomBytes(16).toString("hex")], + ], + }, + key, + ); + timings.push( + `auth;dur=${(performance.now() - authStart).toFixed(2)}`, + ); + connectsBefore = upstream.connects(); + upstreamStart = performance.now(); + return fetchUpstream(`${relay}${upstreamPath}`, { + method, + headers: { + "Content-Type": "application/json", + Authorization: + "Nostr " + + Buffer.from(JSON.stringify(auth)).toString("base64"), + }, + body, + redirect: "error", + signal: requestSignal, + }).then((response) => { timings.push( - `auth;dur=${(performance.now() - authStart).toFixed(2)}`, + `ttfb;dur=${(performance.now() - upstreamStart).toFixed(2)}`, ); - connectsBefore = upstream.connects(); - upstreamStart = performance.now(); - return fetchUpstream(`${relay}${upstreamPath}`, { - method: "POST", - headers: { - "Content-Type": "application/json", - Authorization: - "Nostr " + - Buffer.from(JSON.stringify(auth)).toString("base64"), - }, - body, - redirect: "error", - signal: requestSignal, - }).then((response) => { - timings.push( - `ttfb;dur=${(performance.now() - upstreamStart).toFixed(2)}`, - ); - return response; - }); - }, - requestSignal, - route === "/api/relay/query" && - req.headers["x-buzz-read-priority"] === "background" - ? "background" - : "foreground", - ); - const text = - snapshot && response.ok - ? await readSnapshotText(response) - : await response.text(); - // The relay's own service time separates server work from network time. - const relayMs = Number( - response.headers.get("x-envoy-upstream-service-time"), - ); - timings.push( - ...upstream.connectTiming(connectsBefore), - ...(Number.isFinite(relayMs) && - response.headers.has("x-envoy-upstream-service-time") - ? [`relay;dur=${relayMs}`] - : []), - `upstream;dur=${(performance.now() - upstreamStart).toFixed(2)}`, - ); - res.setHeader("Server-Timing", timings.join(", ")); - stats.queries++; - if (!response.ok) { - stats.errors++; - let failure; - try { - failure = apiFailure(response.status, JSON.parse(text)); - } catch { - failure = apiFailure(response.status, undefined); - } - return json(res, response.status, failure); - } - if (profile) { - const receipt = JSON.parse(text); - if ( - receipt.event_id !== filters.id || - typeof receipt.accepted !== "boolean" - ) - return json(res, 502, { - error: "Profile publication could not be confirmed", - }); + return response; + }); + }, + requestSignal, + route === "/api/relay/query" && + req.headers["x-buzz-read-priority"] === "background" + ? "background" + : "foreground", + ); + const text = + snapshot && response.ok + ? await readSnapshotText(response) + : workflowPath && response.ok + ? await workflowReadText(response) + : publishing && response.ok + ? await readReceiptText(response) + : await response.text(); + // The relay's own service time separates server work from network time. + const relayMs = Number( + response.headers.get("x-envoy-upstream-service-time"), + ); + timings.push( + ...upstream.connectTiming(connectsBefore), + ...(Number.isFinite(relayMs) && + response.headers.has("x-envoy-upstream-service-time") + ? [`relay;dur=${relayMs}`] + : []), + `upstream;dur=${(performance.now() - upstreamStart).toFixed(2)}`, + ); + res.setHeader("Server-Timing", timings.join(", ")); + stats.queries++; + if (!response.ok) { + stats.errors++; + let failure; + try { + failure = apiFailure(response.status, JSON.parse(text)); + } catch { + failure = apiFailure(response.status, undefined); } - res.writeHead(200, { - "Content-Type": "application/json", - "Cache-Control": "no-store", - }); - res.end(text); - } finally { - res.off("close", release); + return json(res, response.status, failure); } + if (profile) { + const receipt = JSON.parse(text); + if ( + receipt.event_id !== filters.id || + typeof receipt.accepted !== "boolean" + ) + return json(res, 502, { + error: "Profile publication could not be confirmed", + }); + } + res.writeHead(200, { + "Content-Type": "application/json", + "Cache-Control": "no-store", + }); + res.end(text); } finally { inflight--; } @@ -1041,6 +1087,8 @@ export function relayBrokerPlugin({ if (isConnectFailure(error)) return json(res, 502, { error: "Relay unreachable", sent: false }); json(res, 500, { error: "Local relay broker failed" }); + } finally { + res.off("close", release); } }); }, diff --git a/dev/workflow-broker.test.mjs b/dev/workflow-broker.test.mjs new file mode 100644 index 00000000..3e29dfc2 --- /dev/null +++ b/dev/workflow-broker.test.mjs @@ -0,0 +1,375 @@ +import { createServer } from "node:http"; +import { setTimeout as delay } from "node:timers/promises"; +import { expect, it, vi } from "vitest"; +import { finalizeEvent, getPublicKey, verifyEvent } from "nostr-tools"; +import { relayBrokerPlugin } from "./relay-broker.mjs"; +import { connectBrokerTransport } from "../src/features/relay/transport.ts"; +import { WORKFLOW_READ_BYTES } from "../src/features/workflows/http.ts"; + +vi.mock("nostr-tools", async (importOriginal) => { + const actual = await importOriginal(); + return { ...actual, finalizeEvent: vi.fn(actual.finalizeEvent) }; +}); + +const id = "11111111-1111-4111-8111-111111111111"; +const runId = "22222222-2222-4222-8222-222222222222"; +const cursor = { before: "2026-09-12T14:44:19.123456+00:00", beforeId: runId }; +async function harness( + respond = () => Response.json({ runs: [], next: null }), + metadata, +) { + const key = new Uint8Array(32); + key[31] = 8; + const viewer = getPublicKey(key), + calls = [], + logs = []; + let handler; + const server = createServer((req, res) => { + if (!req.headers.origin) req.headers.origin = `http://${req.headers.host}`; + void handler(req, res); + }); + await relayBrokerPlugin({ + relayUrl: "https://a.workflow.test", + communityAliases: JSON.stringify({ secondary: "https://b.workflow.test" }), + identity: () => key, + ...(metadata ? {} : { authority: async () => ({ relayAuthor: viewer }) }), + upstreamFetch: async (url, init) => { + if (init.headers.Accept === "application/nostr+json") { + expect(init.redirect).toBe("error"); + const data = await metadata(String(url), viewer, init); + if (data instanceof Error) throw data; + return data instanceof Response ? data : Response.json(data); + } + const auth = JSON.parse( + Buffer.from(init.headers.Authorization.slice(6), "base64").toString(), + ); + expect(verifyEvent(auth)).toBe(true); + expect(auth.pubkey).toBe(viewer); + const call = { url: String(url), init, auth }; + calls.push(call); + return respond(call); + }, + }).configureServer({ + httpServer: server, + config: { + logger: { + info() {}, + error(text) { + logs.push(text); + }, + }, + }, + middlewares: { + use(fn) { + handler = fn; + }, + }, + }); + await new Promise((resolve) => server.listen(0, "127.0.0.1", resolve)); + const base = `http://127.0.0.1:${server.address().port}`; + return { + base, + viewer, + calls, + logs, + post: (route, body, headers = {}) => + fetch(`${base}/api/relay/${route}`, { + method: "POST", + headers: { "Content-Type": "application/json", ...headers }, + body: JSON.stringify(body), + }), + async close() { + server.closeAllConnections(); + await new Promise((resolve) => server.close(resolve)); + }, + }; +} +const signal = () => new AbortController().signal; +it("real broker scoped history signs exact GET path/cursor and captured principal without startup reads or workflow writes", async () => { + const h = await harness(); + try { + const first = await connectBrokerTransport(h.base); + const other = await connectBrokerTransport(h.base, undefined, "secondary"); + expect(h.calls).toHaveLength(0); + expect(first.writer.kinds).toEqual([7, 9, 30620, 46020, 5]); + await first.workflows.runs(id, cursor, signal()); + await other.workflows.runs(id, undefined, signal()); + await first.workflows.runs(id, undefined, signal()); + expect(h.calls.map((call) => call.url)).toEqual([ + `https://a.workflow.test/workflows/${id}/runs?limit=20&before=2026-09-12T14%3A44%3A19.123456%2B00%3A00&before_id=${runId}`, + `https://b.workflow.test/workflows/${id}/runs?limit=20`, + `https://a.workflow.test/workflows/${id}/runs?limit=20`, + ]); + for (const { url, init, auth } of h.calls) { + expect(init.method).toBe("GET"); + expect(init.body).toBeUndefined(); + expect(init.redirect).toBe("error"); + expect(auth.tags).toContainEqual(["u", url]); + expect(auth.tags).toContainEqual(["method", "GET"]); + expect(auth.tags.some(([name]) => name === "payload")).toBe(false); + expect(auth.tags.some(([name]) => name === "nonce")).toBe(true); + } + expect(new Set(h.calls.map(({ auth }) => auth.id)).size).toBe(3); + } finally { + await h.close(); + } +}); +it("broker rejects arbitrary targets, cursors, limits and untrusted origin before upstream dispatch", async () => { + const h = await harness(); + try { + for (const body of [ + null, + { id: "../secret" }, + { id, limit: 100 }, + { id, url: "https://evil.test" }, + { id, cursor: { before: cursor.before } }, + { id, cursor: { ...cursor, before: "not a date" } }, + { id, cursor: { ...cursor, beforeId: "../" } }, + { id, cursor: null }, + ]) { + expect((await h.post("workflow-runs", body)).status).toBe(400); + } + expect( + (await h.post("workflow-approvals", { id, runId: "../" })).status, + ).toBe(404); + expect( + (await h.post("workflow-runs", { id }, { Origin: "https://evil.test" })) + .status, + ).toBe(403); + expect(h.calls).toHaveLength(0); + } finally { + await h.close(); + } +}); +it("workflow quota gates the existing query lane while another community remains independent", async () => { + const h = await harness(({ url }) => + url.startsWith("https://a.") + ? Response.json( + { error: "rate-limited: quota exceeded; retry in 0s" }, + { status: 429 }, + ) + : Response.json([]), + ); + try { + const transport = await connectBrokerTransport(h.base); + await expect( + transport.workflows.runs(id, undefined, signal()), + ).rejects.toMatchObject({ status: 429, retryAfterMs: 1000 }); + await expect( + transport.query([{ kinds: [0], limit: 1 }]), + ).rejects.toMatchObject({ status: 429 }); + expect(h.calls).toHaveLength(1); + const other = await connectBrokerTransport(h.base, undefined, "secondary"); + await other.query([{ kinds: [0], limit: 1 }]); + expect(h.calls).toHaveLength(2); + } finally { + await h.close(); + } +}); +it("workflow 404 remains unavailable, over-budget body is cancelled without leaking bytes", async () => { + let large = false, + cancelled = false; + const h = await harness(() => + large + ? new Response( + new ReadableStream({ + start(controller) { + controller.enqueue( + new TextEncoder().encode( + "PRIVATE".repeat(Math.ceil(WORKFLOW_READ_BYTES / 7) + 1), + ), + ); + }, + cancel() { + cancelled = true; + }, + }), + ) + : Response.json({ error: "PRIVATE missing workflow" }, { status: 404 }), + ); + try { + const transport = await connectBrokerTransport(h.base); + await expect( + transport.workflows.runs(id, undefined, signal()), + ).rejects.toMatchObject({ status: 404 }); + large = true; + await expect( + transport.workflows.runs(id, undefined, signal()), + ).rejects.toThrow(); + expect(cancelled).toBe(true); + expect(h.logs.join(" ")).not.toContain("PRIVATE"); + } finally { + await h.close(); + } +}); +it("closing workflow interest aborts the actual broker upstream request", async () => { + const h = await harness( + ({ init }) => + new Promise((_resolve, reject) => { + init.signal.addEventListener( + "abort", + () => reject(init.signal.reason), + { once: true }, + ); + }), + ); + try { + const transport = await connectBrokerTransport(h.base), + cancel = new AbortController(); + const pending = transport.workflows.runs(id, undefined, cancel.signal); + const rejection = expect(pending).rejects.toThrow(); + for (let i = 0; i < 100 && !h.calls.length; i++) await delay(5); + expect(h.calls).toHaveLength(1); + cancel.abort(); + await rejection; + for (let i = 0; i < 100 && !h.calls[0].init.signal.aborted; i++) + await delay(5); + expect(h.calls[0].init.signal.aborted).toBe(true); + } finally { + await h.close(); + } +}); + +const existingBackend = (_url, viewer) => ({ self: viewer }); +const yaml = + "name: Local test\nenabled: false\ntrigger:\n on: message_posted\nsteps:\n - id: wait\n action: delay\n duration: 1s\n"; +const template = (kind = 30620) => ({ + kind, + created_at: Math.floor(Date.now() / 1000), + content: kind === 30620 ? yaml : "", + tags: [ + ["h", runId], + ["d", id], + ], +}); +it("existing backend signs only canonical workflow sign/publish with exact own events and unchanged receipts", async () => { + const h = await harness(({ init }) => { + const event = JSON.parse(init.body); + expect(verifyEvent(event)).toBe(true); + return Response.json({ + accepted: true, + event_id: event.id, + message: "workflow-result", + }); + }, existingBackend); + try { + const t = await connectBrokerTransport(h.base); + expect(t.writer.kinds).toEqual([7, 9, 30620, 46020, 5]); + for (const input of [ + template(), + template(46020), + { + ...template(5), + tags: [ + ["h", runId], + ["a", `30620:${h.viewer}:${id}`], + ], + }, + ]) { + const event = await t.writer.sign(input, signal()); + expect(verifyEvent(event)).toBe(true); + expect(event.pubkey).toBe(h.viewer); + expect(event.kind).toBe(input.kind); + expect(event.content).toBe(input.content); + expect(event.tags).toEqual(input.tags); + expect(await t.writer.publish(event, signal())).toBe("workflow-result"); + expect(JSON.parse(h.calls.at(-1).init.body)).toEqual( + JSON.parse(JSON.stringify(event)), + ); + } + expect(h.calls).toHaveLength(3); + } finally { + await h.close(); + } +}); +const invalidCommands = [ + ["null", () => null], + ["admin deletion", () => ({ ...template(), kind: 9005 })], + [ + "legacy name", + () => ({ + ...template(), + tags: [ + ["h", runId], + ["d", "name"], + ], + }), + ], + [ + "webhook", + () => ({ + ...template(), + content: yaml.replace("message_posted", "webhook"), + }), + ], + [ + "event-target deletion", + () => ({ + ...template(5), + tags: [ + ["h", runId], + ["e", "a".repeat(64)], + ], + }), + ], + [ + "numeric alias", + (viewer) => ({ + ...template(5), + tags: [ + ["h", runId], + ["a", `030620:${viewer}:${id}`], + ], + }), + ], + [ + "other owner", + () => ({ + ...template(5), + tags: [ + ["h", runId], + ["a", `30620:${"a".repeat(64)}:${id}`], + ], + }), + ], + [ + "extra tag", + () => ({ + ...template(), + tags: [...template().tags, ["p", "a".repeat(64)]], + }), + ], +]; +it.each(invalidCommands)( + "existing-backend broker rejects %s before signing or upstream writes", + async (_name, input) => { + const h = await harness(() => { + throw new Error("must not dispatch writes"); + }, existingBackend); + try { + const signaturesBefore = finalizeEvent.mock.calls.length; + expect((await h.post("sign", input(h.viewer))).status).toBe(400); + expect(finalizeEvent.mock.calls).toHaveLength(signaturesBefore); + expect(h.calls).toHaveLength(0); + } finally { + await h.close(); + } + }, +); +it("existing-backend broker rejects forged commands before upstream writes", async () => { + const h = await harness(() => { + throw new Error("must not dispatch writes"); + }, existingBackend); + try { + const own = await (await h.post("sign", template())).json(); + expect( + (await h.post("publish", { ...own, content: "tampered" })).status, + ).toBe(400); + expect( + (await h.post("publish", { ...own, pubkey: "a".repeat(64) })).status, + ).toBe(400); + expect(h.calls).toHaveLength(0); + } finally { + await h.close(); + } +}); diff --git a/docs/workflows.md b/docs/workflows.md new file mode 100644 index 00000000..3b3924b8 --- /dev/null +++ b/docs/workflows.md @@ -0,0 +1,53 @@ +# Workflows plugin + +The bundled page owns the editor and drafts. The existing relay session owns +configuration reads and commands through its reader and durable outbox; the dev +broker owns authentication and the fixed run-history route. The relay executes +workflows. No separate connection, cache, outbox, scheduler or backend changes. +See the [capability contract](../src/features/workflows/types.ts). + +## Scope + +- Channel-scoped saved configurations; new drafts start disabled. +- Form editing for message/reaction triggers and Send Message/Delay actions. + Other definitions stay in YAML; opening them does not rewrite their contents. +- Save with the original owner/channel/UUID and signed `expected-revision`. + Warn on broad message or schedule activation, not ordinary enabled edits. +- Confirmed deletion request, manual run, and on-demand run/trace history in + 20-row pages with the relay's exact `(before,beforeId)` cursor. +- No approval UI, webhook-secret handling, lifecycle negotiation, alternative + signed-host adapter or plugin command-replay API. Webhook-trigger saves are + blocked at both the editor and signing boundary, including raw YAML. + +## Recovery and limits + +The outbox journals intent/signature before publication. A verified echo does +not replace the result-bearing receipt. Restored commands never run automatically; +result text stays ephemeral, outside the journal. Workflow intents cannot be replayed +through generic Outbox Retry either; inspection and dismissal remain available. + +**Check saved configuration** resolves an unknown save only when a fresh verified +head matches its owner, channel, UUID and exact signed revision. Missing/different +heads retain the draft for review. Dismissal clears the notice and editor lock +only after durable dismissal; it neither undoes nor repeats a command. Unknown +runs stay unknown: only a returned run ID identifies a requested run. + +Saving a configured enabled flag does not prove runtime activation or cancellation. +Legacy deletion can retain a visible definition; accepted delivery is not proof +of runtime cleanup. These backend limitations are displayed, not repaired here. + +Reads begin on UI interest and stop on unmount or access loss; no background poll. +Transient socket recovery cancels stale reads but retains the draft and active +HTTP receipt correlation; refresh checks current data. Actual access loss purges +private snapshots before callbacks. Drafts are editor-local, +not durable, and never move between viewers or communities. The broker preserves +same-origin checks, signature validation, captured principal quotas, cancellation, +fixed upstream paths and bounded history/receipt bodies. + +## Validation boundary + +Offline regressions cover the editor, exact-save recovery, uncertain commands, +session isolation, broker authorization and history. This is not live acceptance: +create → edit → run → inspect against an unchanged backend and packaged-native +acceptance remain unverified. Live credentials, native launch and real workflow +writes require separate consent; no backend work belongs to this plugin PR. diff --git a/package.json b/package.json index 2532f3c1..232989c1 100644 --- a/package.json +++ b/package.json @@ -57,7 +57,8 @@ "react-markdown": "10.1.0", "remark-breaks": "4.0.0", "remark-gfm": "4.0.1", - "virtua": "0.51.0" + "virtua": "0.51.0", + "yaml": "2.8.3" }, "devDependencies": { "@biomejs/biome": "2.5.12", diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index d64a91f9..a857eee7 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -80,6 +80,9 @@ importers: virtua: specifier: 0.51.0 version: 0.51.0(patch_hash=14ff685d9bc68b34d1a7c6b0042d6000e4b07a9746088fc3842c89c4a995ca9d)(react-dom@19.2.8(react@19.2.8))(react@19.2.8) + yaml: + specifier: 2.8.3 + version: 2.8.3 devDependencies: '@biomejs/biome': specifier: 2.5.12 @@ -104,7 +107,7 @@ importers: version: 19.2.7(@types/react@19.2.18) '@vitejs/plugin-react': specifier: 6.1.1 - version: 6.1.1(vite@8.2.2(@types/node@24.13.3)(jiti@2.7.0)) + version: 6.1.1(vite@8.2.2(@types/node@24.13.3)(jiti@2.7.0)(yaml@2.8.3)) postcss: specifier: 8.5.28 version: 8.5.28 @@ -119,10 +122,10 @@ importers: version: 7.16.0 vite: specifier: 8.2.2 - version: 8.2.2(@types/node@24.13.3)(jiti@2.7.0) + version: 8.2.2(@types/node@24.13.3)(jiti@2.7.0)(yaml@2.8.3) vitest: specifier: 4.1.11 - version: 4.1.11(@types/node@24.13.3)(vite@8.2.2(@types/node@24.13.3)(jiti@2.7.0)) + version: 4.1.11(@types/node@24.13.3)(vite@8.2.2(@types/node@24.13.3)(jiti@2.7.0)(yaml@2.8.3)) packages: @@ -1598,6 +1601,11 @@ packages: engines: {node: '>=8'} hasBin: true + yaml@2.8.3: + resolution: {integrity: sha512-AvbaCLOO2Otw/lW5bmh9d/WEdcDFdQp2Z2ZUH3pX9U2ihyUY0nvLv7J6TrWowklRGPYbB/IuIMfYgxaCPg5Bpg==} + engines: {node: '>= 14.6'} + hasBin: true + zwitch@2.0.4: resolution: {integrity: sha512-bXE4cR/kVZhKZX/RjPEflHaKVhUVl85noU3v6b8apfQEc1x4A+zBxjZ4lN8LqGd6WZ3dl98pY4o717VFmoPp+A==} @@ -2047,10 +2055,10 @@ snapshots: '@ungap/structured-clone@1.4.0': {} - '@vitejs/plugin-react@6.1.1(vite@8.2.2(@types/node@24.13.3)(jiti@2.7.0))': + '@vitejs/plugin-react@6.1.1(vite@8.2.2(@types/node@24.13.3)(jiti@2.7.0)(yaml@2.8.3))': dependencies: '@rolldown/pluginutils': 1.0.1 - vite: 8.2.2(@types/node@24.13.3)(jiti@2.7.0) + vite: 8.2.2(@types/node@24.13.3)(jiti@2.7.0)(yaml@2.8.3) '@vitest/expect@4.1.11': dependencies: @@ -2061,13 +2069,13 @@ snapshots: chai: 6.2.2 tinyrainbow: 3.1.1 - '@vitest/mocker@4.1.11(vite@8.2.2(@types/node@24.13.3)(jiti@2.7.0))': + '@vitest/mocker@4.1.11(vite@8.2.2(@types/node@24.13.3)(jiti@2.7.0)(yaml@2.8.3))': dependencies: '@vitest/spy': 4.1.11 estree-walker: 3.0.3 magic-string: 0.30.21 optionalDependencies: - vite: 8.2.2(@types/node@24.13.3)(jiti@2.7.0) + vite: 8.2.2(@types/node@24.13.3)(jiti@2.7.0)(yaml@2.8.3) '@vitest/pretty-format@4.1.11': dependencies: @@ -2964,7 +2972,7 @@ snapshots: react: 19.2.8 react-dom: 19.2.8(react@19.2.8) - vite@8.2.2(@types/node@24.13.3)(jiti@2.7.0): + vite@8.2.2(@types/node@24.13.3)(jiti@2.7.0)(yaml@2.8.3): dependencies: lightningcss: 1.33.0 picomatch: 4.0.7 @@ -2975,11 +2983,12 @@ snapshots: '@types/node': 24.13.3 fsevents: 2.3.3 jiti: 2.7.0 + yaml: 2.8.3 - vitest@4.1.11(@types/node@24.13.3)(vite@8.2.2(@types/node@24.13.3)(jiti@2.7.0)): + vitest@4.1.11(@types/node@24.13.3)(vite@8.2.2(@types/node@24.13.3)(jiti@2.7.0)(yaml@2.8.3)): dependencies: '@vitest/expect': 4.1.11 - '@vitest/mocker': 4.1.11(vite@8.2.2(@types/node@24.13.3)(jiti@2.7.0)) + '@vitest/mocker': 4.1.11(vite@8.2.2(@types/node@24.13.3)(jiti@2.7.0)(yaml@2.8.3)) '@vitest/pretty-format': 4.1.11 '@vitest/runner': 4.1.11 '@vitest/snapshot': 4.1.11 @@ -2996,7 +3005,7 @@ snapshots: tinyexec: 1.3.1 tinyglobby: 0.2.17 tinyrainbow: 3.1.1 - vite: 8.2.2(@types/node@24.13.3)(jiti@2.7.0) + vite: 8.2.2(@types/node@24.13.3)(jiti@2.7.0)(yaml@2.8.3) why-is-node-running: 2.3.0 optionalDependencies: '@types/node': 24.13.3 @@ -3008,4 +3017,6 @@ snapshots: siginfo: 2.0.0 stackback: 0.0.2 + yaml@2.8.3: {} + zwitch@2.0.4: {} diff --git a/src/app/pages.integration.test.mjs b/src/app/pages.integration.test.mjs index f9b0f0f5..73823f3e 100644 --- a/src/app/pages.integration.test.mjs +++ b/src/app/pages.integration.test.mjs @@ -29,7 +29,7 @@ test("the app runtime exposes ready bundled pages and removes them on disable", services = createServices(); assert.deepEqual(services.pages.snapshot(), []); await settle(); - assert.equal(services.pages.snapshot().length, 3); + assert.equal(services.pages.snapshot().length, 4); await vi.waitFor(() => assert.equal(services.conversation.tools.snapshot().length, 2), ); @@ -110,7 +110,7 @@ test("the app runtime exposes ready bundled pages and removes them on disable", .some((panel) => panel.pluginId === "buzz.bestie"), false, ); - assert.equal(services.pages.snapshot().length, 3); + assert.equal(services.pages.snapshot().length, 4); await services.plugins.change("enable", "buzz.bestie"); // Management completion is not activation completion; Cordis still owns import/disposal barriers. await vi.waitFor(() => @@ -146,6 +146,33 @@ test("the app runtime exposes ready bundled pages and removes them on disable", ); assert.equal(services.relay.snapshot().session, session); assert.ok(session.agentLibrary); + const workflows = services.pages + .snapshot() + .find((page) => page.pluginId === "buzz.workflows"); + assert.equal(workflows.title, "Workflows"); + assert.equal(workflows.layout, "workspace"); + assert.match( + renderToStaticMarkup(createElement(workflows.component)), + /Connect to a community/, + ); + await services.plugins.change("disable", "buzz.workflows"); + assert.equal( + services.pages + .snapshot() + .some((page) => page.pluginId === "buzz.workflows"), + false, + ); + assert.equal(services.relay.snapshot().session, session); + assert.ok(session.workflows); + await services.plugins.change("enable", "buzz.workflows"); + await vi.waitFor(() => + assert.ok( + services.pages + .snapshot() + .some((page) => page.pluginId === "buzz.workflows"), + ), + ); + await services.plugins.change("disable", "buzz.workflows"); await services.plugins.change("disable", "buzz.channels"); const [projects] = services.pages.snapshot(); assert.equal(services.pages.snapshot().length, 1); diff --git a/src/bundled/channels/OutboxStatus.tsx b/src/bundled/channels/OutboxStatus.tsx index a6091060..a52b6003 100644 --- a/src/bundled/channels/OutboxStatus.tsx +++ b/src/bundled/channels/OutboxStatus.tsx @@ -1,3 +1,4 @@ +import { isWorkflowOperation } from "../../features/workflows/protocol"; import { useState, useSyncExternalStore } from "react"; import type { RelayProfiler } from "../../features/relay/profiling"; import { RelayTimings } from "./RelayTimings"; @@ -43,11 +44,15 @@ export function OutboxStatus({ }[item.delivery] } {" "} - {(item.delivery === "failed" || item.delivery === "unknown") && ( - - )}{" "} + {!isWorkflowOperation(item.event) && + (item.delivery === "failed" || item.delivery === "unknown") && ( + + )}{" "} {item.delivery !== "sending" && ( + + + + + + ); +} diff --git a/src/bundled/workflows/WorkflowChannel.tsx b/src/bundled/workflows/WorkflowChannel.tsx new file mode 100644 index 00000000..30af788f --- /dev/null +++ b/src/bundled/workflows/WorkflowChannel.tsx @@ -0,0 +1,465 @@ +import { + useEffect, + useCallback, + useRef, + useState, + useSyncExternalStore, +} from "react"; +import type { + WorkflowCapability, + WorkflowDefinition, +} from "../../features/workflows/types"; +import { Button } from "../../shared/design-system/ui/Button"; +import { ConfirmAction } from "./ConfirmAction"; +import { WorkflowEditor } from "./WorkflowEditor"; +import { WorkflowOperations } from "./WorkflowOperations"; +import { WorkflowRuns } from "./WorkflowRuns"; +import { exactSaveReadback } from "./editor-model"; +import { DEFAULT_FORM_STATE, formStateToYaml } from "./workflowFormTypes"; +import { readWorkflowDocumentFields } from "./workflowYamlDocument"; +import { useWorkflowView } from "./useWorkflowView"; + +type Draft = { + original: WorkflowDefinition | undefined; + yaml: string; + initial: string; + operationId?: string; +}; + +export function WorkflowChannel({ + capability, + channelId, + channelName, + viewer, + onDraftRiskChange, +}: { + capability: WorkflowCapability; + channelId: string; + channelName: string; + viewer: string; + onDraftRiskChange?: (atRisk: boolean) => void; +}) { + const { snapshot, refresh } = useWorkflowView( + useCallback( + () => capability.definitions(channelId), + [capability, channelId], + ), + ); + const operations = useSyncExternalStore( + capability.operations.subscribe, + capability.operations.snapshot, + capability.operations.snapshot, + ); + const submission = useRef(null); + const [draft, setDraft] = useState(null); + const [pendingSelection, setPendingSelection] = useState< + WorkflowDefinition | "new" | "close" | null + >(null); + const [confirmDelete, setConfirmDelete] = useState(false); + const [error, setError] = useState(null); + const [readRuns, setReadRuns] = useState(false); + const operation = draft?.operationId + ? operations.find((item) => item.eventId === draft.operationId) + : undefined; + const ownOperations = operations.filter( + (item) => item.workflow.channelId === channelId, + ); + const busy = + !!draft?.operationId && (!operation || operation.outcome === "pending"); + const readonly = !!draft?.original && draft.original.owner !== viewer; + const dirty = !!draft && draft.yaml !== draft.initial; + const atRisk = dirty || !!draft?.operationId; + const unresolvedWrite = ownOperations.some( + (item) => + (item.outcome === "pending" || item.outcome === "unknown") && + (draft?.original + ? item.workflow.id === draft.original.id && + item.workflow.owner === draft.original.owner + : item.action === "save"), + ); + useEffect(() => { + onDraftRiskChange?.(atRisk); + return () => onDraftRiskChange?.(false); + }, [atRisk, onDraftRiskChange]); + useEffect(() => { + if (!atRisk) return; + const warn = (event: BeforeUnloadEvent) => event.preventDefault(); + window.addEventListener("beforeunload", warn); + return () => window.removeEventListener("beforeunload", warn); + }, [atRisk]); + const open = (next: WorkflowDefinition | "new" | "close") => { + const yaml = + next === "new" + ? formStateToYaml({ ...DEFAULT_FORM_STATE, name: "Untitled workflow" }) + : next === "close" + ? "" + : next.yaml; + setDraft( + next === "close" + ? null + : { + original: typeof next === "string" ? undefined : next, + yaml, + initial: yaml, + }, + ); + submission.current = null; + setError(null); + setReadRuns(false); + setPendingSelection(null); + setConfirmDelete(false); + }; + const select = (next: WorkflowDefinition | "new" | "close") => { + if (atRisk) setPendingSelection(next); + else open(next); + }; + useEffect(() => { + if (operation?.eventId && operation.outcome === "succeeded") void refresh(); + }, [operation?.eventId, operation?.outcome, refresh]); + useEffect(() => { + if (!operation || !draft || snapshot?.status !== "ready") return; + const saved = exactSaveReadback(operation, snapshot.data.items); + if (saved) { + submission.current = null; + setDraft({ original: saved, yaml: saved.yaml, initial: saved.yaml }); + } + }, [operation, snapshot, draft]); + // A cleared/unavailable view withdraws the saved private definition from display. + // Unsaved user-authored drafts never become a second retained definition cache. + useEffect(() => { + if (snapshot?.status === "unavailable" || snapshot?.status === "idle") { + submission.current = null; + setDraft(null); + setPendingSelection(null); + setConfirmDelete(false); + setReadRuns(false); + setError(null); + } + }, [snapshot?.status]); + const dismiss = async (eventId: string) => { + await capability.operations.dismiss(eventId); + if (submission.current === eventId) submission.current = null; + setDraft((current) => { + if (current?.operationId !== eventId) return current; + const { operationId: _, ...retained } = current; + return retained; + }); + }; + const save = () => { + if ( + !draft || + submission.current || + busy || + unresolvedWrite || + readonly || + !capability.availability.save || + draft.operationId + ) + return; + try { + submission.current = "submitting"; + const operationId = capability.save({ + channelId, + yaml: draft.yaml, + ...(draft.original ? { existing: draft.original } : {}), + }); + submission.current = operationId; + setDraft({ ...draft, operationId }); + setError(null); + } catch (cause) { + submission.current = null; + setError( + cause instanceof Error + ? cause.message + : "The workflow could not be submitted. Your draft is retained.", + ); + } + }; + const remove = () => { + if ( + !draft?.original || + submission.current || + readonly || + unresolvedWrite || + !capability.availability.delete || + draft.operationId + ) + return; + try { + submission.current = "submitting"; + const operationId = capability.delete(draft.original); + submission.current = operationId; + setDraft({ ...draft, operationId }); + setConfirmDelete(false); + setError(null); + } catch (cause) { + submission.current = null; + setError( + cause instanceof Error + ? cause.message + : "Deletion could not be submitted. Your draft is retained.", + ); + } + }; + const trigger = () => { + if ( + !draft?.original || + submission.current || + dirty || + unresolvedWrite || + draft.operationId || + readonly || + !capability.availability.trigger + ) + return; + try { + submission.current = "submitting"; + submission.current = capability.trigger(draft.original); + setError(null); + } catch (cause) { + submission.current = null; + setError( + cause instanceof Error ? cause.message : "Run could not be submitted.", + ); + } + }; + useEffect(() => { + const active = operations.find( + (item) => item.eventId === submission.current, + ); + if ( + active?.action === "trigger" && + (active.outcome === "succeeded" || active.outcome === "rejected") + ) + submission.current = null; + }, [operations]); + let blocked: string | undefined; + if (!capability.availability.save) + blocked = "Saving is unavailable from this host."; + else if (unresolvedWrite && !draft?.operationId) + blocked = + "Check the saved configuration or review the unresolved request in Recent activity before continuing."; + else if (draft?.operationId) + blocked = + operation?.outcome === "succeeded" + ? operation.action === "delete" + ? "Deletion request accepted, not verified runtime deletion. The configuration may remain visible. Review Recent activity to continue." + : "Configuration saved; waiting for a readback of this exact revision. Check saved configuration or review the current version in Recent activity." + : operation?.outcome === "rejected" + ? "Request rejected. Your draft is retained; review the error before continuing." + : "Your draft is retained. Check the saved configuration or review the request in Recent activity to continue."; + if (!snapshot) return

Reading configurations…

; + return ( +
+
+

Saved configurations

+ + +
+

+ Configured activation may differ from the existing backend’s runtime + state. Saving a disabled configuration does not confirm that automatic + runs have stopped or cancel work already running. +

+ {snapshot.status === "loading" && ( +

Reading configurations…

+ )} + {snapshot.status === "idle" && ( +

Configurations cleared. Refresh to read again.

+ )} + {snapshot.status === "unavailable" && ( +

Workflow definitions are unavailable.

+ )} + {snapshot.status === "error" && ( +

+ {snapshot.error ?? + "Configurations could not be read. Use Refresh configurations to retry."} +

+ )} + {snapshot.data.partial && ( +

This is a bounded, partial list.

+ )} + {!capability.availability.save && ( +

+ Creating and saving workflows is unavailable from this host. You can + browse saved configurations, but cannot save changes here yet. +

+ )} + {snapshot.status === "ready" && !snapshot.data.items.length && ( +

+ No saved configurations returned for this channel. + {capability.availability.save && " Create a disabled draft to start."} +

+ )} +
    + {snapshot.data.items.map((definition) => { + const header = readWorkflowDocumentFields(definition.yaml); + return ( +
  • + + + {header.editable + ? header.enabled === false + ? "Configured disabled" + : "Configured enabled" + : "Unreadable configuration"} + {definition.owner !== viewer ? " · Read-only" : ""} + +
    • + ); + })} +
+ {draft && + snapshot.status !== "unavailable" && + snapshot.status !== "idle" && ( +
+
+

+ {draft.original ? "Workflow details" : "New workflow"} +

+ +
+ {draft.original && ( +
+ Configuration details +

+ Owner: {draft.original.owner} +

+

+ Revision: {draft.original.revision} +

+
+ )} + {readonly && ( +

+ This definition belongs to another identity. Only its author can + manage it here. +

+ )} + setDraft({ ...draft, yaml })} + onSave={save} + readOnly={readonly} + busy={busy} + locked={!!draft.operationId} + blocked={blocked} + /> +

+ Drafts stay in this editor only. Leaving the Workflows page or + reloading discards unsaved text, but does not cancel submitted + operations. +

+ {error && ( +

+ {error} +

+ )} + {operation?.outcome === "rejected" && ( + + )} + {draft.original && ( +
+ + {!readonly && ( + <> + + + + )} +
+ )} + {draft.original && !capability.availability.delete && !readonly && ( +

+ Delete requests are unavailable from this host. +

+ )} + {draft.original && readRuns && ( + + )} + {confirmDelete && ( + setConfirmDelete(false)} + /> + )} +
+ )} + + {pendingSelection && + snapshot.status !== "idle" && + snapshot.status !== "unavailable" && ( + open(pendingSelection)} + onCancel={() => setPendingSelection(null)} + /> + )} +
+ ); +} diff --git a/src/bundled/workflows/WorkflowEditor.tsx b/src/bundled/workflows/WorkflowEditor.tsx new file mode 100644 index 00000000..deb18fbe --- /dev/null +++ b/src/bundled/workflows/WorkflowEditor.tsx @@ -0,0 +1,200 @@ +import { Input } from "@base-ui/react/input"; +import { useId, useState } from "react"; +import { Button } from "../../shared/design-system/ui/Button"; +import { Switch } from "../../shared/design-system/ui/Switch"; +import { Tabs } from "../../shared/design-system/ui/Tabs"; +import { ConfirmAction } from "./ConfirmAction"; +import { WorkflowForm } from "./WorkflowForm"; +import { draftError, hasWebhookTrigger } from "./editor-model"; +import { getWorkflowActivationWarning } from "./workflowActivationWarning"; +import { + formStateToYaml, + yamlToFormState, + type WorkflowFormState, +} from "./workflowFormTypes"; +import { + readWorkflowDocumentFields, + yamlWithWorkflowEnabled, + yamlWithWorkflowName, +} from "./workflowYamlDocument"; + +export function WorkflowEditor({ + yaml, + initialYaml, + onChange, + onSave, + readOnly = false, + blocked, + busy = false, + locked = false, +}: { + yaml: string; + initialYaml?: string | undefined; + onChange: (yaml: string) => void; + onSave: () => void; + readOnly?: boolean; + blocked?: string | undefined; + busy?: boolean; + locked?: boolean; +}) { + const id = useId(); + const [mode, setMode] = useState<"form" | "yaml">(() => + yamlToFormState(yaml).ok ? "form" : "yaml", + ); + const [formDraft, setFormDraft] = useState(null); + const [formYaml, setFormYaml] = useState(yaml); + const [activating, setActivating] = useState(false); + const [modeError, setModeError] = useState(null); + const fields = readWorkflowDocumentFields(yaml); + const parsed = yamlToFormState(yaml); + // An incomplete form is still an editable draft. An external YAML change must + // be reparsed instead of reviving stale form state. + const form = + formYaml === yaml && formDraft + ? formDraft + : parsed.ok + ? parsed.state + : null; + const error = draftError(yaml); + const secretGate = hasWebhookTrigger(yaml) + ? "Webhook-trigger saves are unavailable until secure one-time-secret display is supported." + : undefined; + const unavailable = blocked || secretGate; + const mutateForm = (state: WorkflowFormState) => { + const next = formStateToYaml(state); + setFormDraft(state); + setFormYaml(next); + onChange(next); + }; + const changeHeader = ( + next: string | null, + patch: Partial, + ) => { + if (next === null) return; + if (form) { + setFormDraft({ ...form, ...patch }); + setFormYaml(next); + } + onChange(next); + }; + const warning = getWorkflowActivationWarning(yaml); + const submit = () => { + if (readOnly || busy || locked || unavailable || error) return; + const wasEnabled = + initialYaml !== undefined && + readWorkflowDocumentFields(initialYaml).enabled !== false; + if (fields.enabled !== false && !wasEnabled && warning) setActivating(true); + else onSave(); + }; + return ( +
+
+ + + changeHeader(yamlWithWorkflowEnabled(yaml, enabled), { enabled }) + } + /> +
+ { + if (next === "form" && !form) { + setModeError(parsed.ok ? null : parsed.error); + return; + } + setModeError(null); + setMode(next); + }} + /> + {modeError && ( +

+ {modeError} +

+ )} + {mode === "form" && form ? ( + + ) : ( +
+ +