From 5d32cc8396e87d213cc0635d81a67e51e4cdc260 Mon Sep 17 00:00:00 2001 From: Claude Date: Sun, 26 Jul 2026 16:50:54 +0000 Subject: [PATCH] feat(cli): drive Buzz workflows from amy (trigger, run, approve/deny) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Switch the agent-support-channel prototype from the speculative agent-job kinds (43001-43006, reserved with no upstream builder) to Buzz's real, source-confirmed workflow primitive: 30620 def / 46020 trigger / 46001-46007 lifecycle / 46010 approval gate / 46030-46031 grant-deny (pinned against buzz-relay's command_executor.rs). This bakes the human-approval gate into the protocol — anyone in the channel can drive a run, but a human grants before anything is pushed. - commons WorkflowRunAggregator folds trigger + lifecycle + grant/deny into per-run state (TRIGGERED/RUNNING/AWAITING_APPROVAL/APPROVED/COMPLETED/ FAILED/DENIED), correlating by run id (= trigger event id = approval token); 8-case test. - cli `amy buzz workflow` — trigger/list/show/approve/deny plus the `run` runner: per new trigger it does the agent work in a fresh worktree+branch, posts the 46010 gate, and on a later poll runs --on-approve (push + PR) and emits 46005 completed; a deny discards the worktree (run is DENIED). On a real Buzz relay the relay executes the workflow YAML; self-hosted on geode there is no engine, so amy is the runner and emits the lifecycle events itself (documented divergence). Two store realities, both verified against geode: decisions are fetched by author (quartz's store serves #d only for addressable kinds, and 46030/46031 are regular), and the runner is restart-safe (runs at the gate are rebuilt from the deterministic run id). Also hardens the exec helper against a broken pipe when --exec doesn't read stdin, and fixes worktree teardown to run git against the owning repo. End-to-end headless harness (cli/tests/buzz/workflow-loop.sh) covers the grant and deny paths through an embedded geode relay: 14/14 green. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_011mApqAbr8vkLC7gUDjavu6 --- cli/README.md | 29 + .../2026-07-25-buzz-agent-support-channel.md | 51 +- .../com/vitorpamplona/amethyst/cli/Main.kt | 3 + .../amethyst/cli/commands/BuzzCommands.kt | 2 + .../cli/commands/BuzzWorkflowCommands.kt | 572 ++++++++++++++++++ cli/tests/.gitignore | 1 + cli/tests/buzz/workflow-loop.sh | 148 +++++ .../commons/model/buzz/WorkflowRuns.kt | 220 +++++++ .../model/buzz/WorkflowRunAggregatorTest.kt | 131 ++++ 9 files changed, 1155 insertions(+), 2 deletions(-) create mode 100644 cli/src/main/kotlin/com/vitorpamplona/amethyst/cli/commands/BuzzWorkflowCommands.kt create mode 100755 cli/tests/buzz/workflow-loop.sh create mode 100644 commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/model/buzz/WorkflowRuns.kt create mode 100644 commons/src/commonTest/kotlin/com/vitorpamplona/amethyst/commons/model/buzz/WorkflowRunAggregatorTest.kt diff --git a/cli/README.md b/cli/README.md index 8da2c5c92c..59a969c37c 100644 --- a/cli/README.md +++ b/cli/README.md @@ -620,6 +620,35 @@ and `commons` aggregator the app uses. > upstream builder; the tag layout (`e`/`h`/`p`/`status`) is a best-effort model and will be > reconciled once Buzz implements the protocol. See > [`cli/plans/2026-07-25-buzz-agent-support-channel.md`](plans/2026-07-25-buzz-agent-support-channel.md). + +#### Buzz workflows (source-confirmed — the human-approval primitive) + +Where agent-jobs are speculative, **workflows are Buzz's real structured-work primitive**: the +command kinds (30620 definition, 46020 trigger, 46030/46031 grant/deny) are pinned against +buzz-relay's Rust `command_executor.rs`. A run pauses on a **human-approval gate** and only ships +after someone grants it — exactly the "anyone can drive, but a human gates the merge" model. + +On a real Buzz relay the *relay* parses the workflow YAML and executes it. Self-hosted on geode +there is no workflow engine, so **`amy` is the runner** and emits the lifecycle events itself — a +documented divergence. The **run id is the trigger's event id and doubles as the approval token**, +so a grant's `d` tag equals the run id (no separate token bookkeeping). Because quartz's event store +serves `#d` only for addressable kinds, decisions (regular kind 46030/46031) are fetched **by +author** — every 46010 gate names its approver in a `p` tag — and matched to their run by the token. + +| Command | What it does | +| --- | --- | +| `amy buzz workflow trigger RELAY WFID --task TEXT --channel GID` | Trigger a run (kind:46020). Prints the `run_id` (= the trigger event id = the approval token). | +| `amy buzz workflow list RELAY --channel GID [--timeout SECS]` | List a channel's runs, folded to state (TRIGGERED/RUNNING/AWAITING_APPROVAL/APPROVED/COMPLETED/FAILED/DENIED), awaiting-approval first. | +| `amy buzz workflow show RELAY RUNID [--timeout SECS]` | Show one run's folded state + lifecycle (resolves the channel from the trigger, then folds it). | +| `amy buzz workflow approve RELAY RUNID [--note N]` | Grant a run's approval gate (kind:46030, `d`=run id). Resumes the paused run. | +| `amy buzz workflow deny RELAY RUNID [--note N]` | Deny a run's approval gate (kind:46031). The run is terminal (DENIED); the runner discards the unshipped work. | +| `amy buzz workflow run RELAY --exec CMD --channel GID --approver NPUB [--on-approve CMD] [--worktree REPODIR] [--base-ref REF] [--accept-from npub,…] [--poll SECS] [--once]` | Run the **runner**. Per new trigger: emits triggered (46001) → step-started (46002) → runs `sh -c CMD` inside a fresh `git worktree`+branch (task on stdin; `BUZZ_RUN/CHANNEL/RELAY/AGENT/REQUESTER/BRANCH/WORKTREE/BASE_REF` in env) → step-completed (46003) → posts the **approval gate** (46010, addressed to `--approver`). On a later poll, when the approver publishes a grant it runs `--on-approve CMD` (the push + open-PR step) and emits completed (46005, carrying the PR url); a deny discards the worktree. Restart-safe: runs still at the gate are rebuilt from the run id on startup. | + +> **Permissions (same three-layer model as jobs).** Buzz gates *who can trigger/approve* +> (`--approver`, `--accept-from`); the exec credential bounds *what the agent can touch* (keep the +> git token PR-only); GitHub branch protection keeps *merge off the agent's path*. The approval gate +> adds a fourth: a human must grant before anything is pushed. See +> [`cli/plans/2026-07-25-buzz-agent-support-channel.md`](plans/2026-07-25-buzz-agent-support-channel.md). > > **Permissions.** Buzz scopes by identity, not capability flags. `--accept-from` is the > intake gate; what the agent can do to a repo is bounded by the credentials you give diff --git a/cli/plans/2026-07-25-buzz-agent-support-channel.md b/cli/plans/2026-07-25-buzz-agent-support-channel.md index 03b66aa8f0..a64dc8fd7a 100644 --- a/cli/plans/2026-07-25-buzz-agent-support-channel.md +++ b/cli/plans/2026-07-25-buzz-agent-support-channel.md @@ -190,10 +190,57 @@ Key files: routes `amethyst/.../navigation/routes/Routes.kt`; render dispatch ingest `model/LocalCache.kt` (~L4780-4855); subscription `.../relayGroup/datasource/RelayGroupFilterBuilders.kt`. +## Pivot — jobs → workflows (2026-07-26) + +The 43001-43006 job prototype above proved the *shape* (drive an agent from a shared channel, +worktree-isolate, PR-only, merge-on-GitHub), but those kinds are **reserved/speculative** with no +upstream builder. Buzz's **real, source-confirmed** structured-work primitive is the **workflow** +family — the command kinds are pinned against buzz-relay's Rust `command_executor.rs`: + +- **30620** workflow definition, **46020** trigger, **46001-46007** run/step lifecycle, +- **46010** approval-requested gate, **46030 / 46031** grant / deny. + +So the driving surface switched to workflows. What that buys over jobs: a **first-class +human-approval gate** (46010 → 46030/46031) baked into the protocol — the exact "anyone in the +channel can drive, but a human gates the merge" model the goal asks for — rather than relying on +GitHub branch-protection alone. + +**Divergence (documented):** on a real Buzz relay the *relay* parses the workflow YAML and executes +it, signing the lifecycle + approval events. Self-hosted on geode there is no workflow engine, so +**`amy` is the runner** (`amy buzz workflow run`) and emits the lifecycle events itself. The command +events (30620/46020/46030/46031) stay faithful to Buzz; only the lifecycle *content* shape is +Amethyst's (Buzz leaves it relay-defined). + +**Correlation:** the **run id is the trigger's event id and doubles as the approval token**, so a +grant's `d` tag equals the run id — no separate token bookkeeping. Two store realities shaped the +wire handling, both verified against geode: +- quartz's `SQLiteEventStore` routes every `#d` filter to the addressable `d_tag` column (NULL for a + regular kind like 46030), so **decisions are fetched by author** — every 46010 gate names its + approver in a `p` tag — and matched to their run by the token the aggregator reads off the event. +- The runner is **restart-safe**: runs still at the gate (AWAITING_APPROVAL / APPROVED / DENIED) are + rebuilt into the in-flight map from the run id on startup (worktree path + branch are + deterministic), so a decision arriving in a later poll — or a fresh `--once` process — still + resolves. The relay is the source of truth, not the in-memory map. + +**Landed (CLI + commons):** +- `commons/.../model/buzz/WorkflowRuns.kt` — `WorkflowRunAggregator` folds trigger + lifecycle + + grant/deny into per-run state (`WorkflowRunAggregatorTest`, 8 cases). +- `cli/.../commands/BuzzWorkflowCommands.kt` — `trigger` / `list` / `show` / `approve` / `deny` and + the **`run`** runner (agent work → 46010 gate → on grant runs `--on-approve` → 46005 completed; a + deny discards the worktree, run is DENIED). Wired into `amy buzz workflow`. +- `cli/tests/buzz/workflow-loop.sh` — end-to-end headless harness (alice triggers, bot runs, + carol approves/denies) through embedded geode; 14/14 green, including the deny path and + worktree cleanup. + +The jobs code stays for now (the board already ships) but the workflow path is the one that matches +Buzz upstream; the mobile P1 approvals-inbox below is really the 46010/46030/46031 surface. + ## Follow-ups -1. Reconcile 43001-43006 with Buzz upstream once it defines the job protocol. -2. Wire the P0 mobile screens (approvals inbox + jobs board) on top of `BuzzJobAggregator`. +1. Reconcile 43001-43006 with Buzz upstream once it defines the job protocol (or retire the job + path in favor of workflows). +2. Wire the P0 mobile screens (approvals inbox + jobs board) on top of `BuzzJobAggregator` / + `WorkflowRunAggregator`. 3. ✅ **Done** — a reference `--exec` wrapper (`tools/buzz-agent/agent-exec.sh` + README) runs the coding agent in the job worktree, commits, pushes the feature branch, opens a PR with a PR-only token, and prints the URL as the job result — with the branch-protection + token-scope diff --git a/cli/src/main/kotlin/com/vitorpamplona/amethyst/cli/Main.kt b/cli/src/main/kotlin/com/vitorpamplona/amethyst/cli/Main.kt index f7d936f3aa..1a07da75cf 100644 --- a/cli/src/main/kotlin/com/vitorpamplona/amethyst/cli/Main.kt +++ b/cli/src/main/kotlin/com/vitorpamplona/amethyst/cli/Main.kt @@ -824,6 +824,9 @@ private fun printUsage() { | buzz personas [--relays R,R] list my kind-30175 personas | buzz job request RELAY file an agent job (kind-43001) | buzz job list/show/cancel RELAY … track agent jobs (43001-43006) + | buzz workflow trigger RELAY WFID … trigger a Buzz workflow run (kind-46020) + | buzz workflow run RELAY --exec CMD … run the workflow runner (agent work → approval gate) + | buzz workflow approve/deny RELAY RUNID grant/deny a run's approval gate (46030/46031) | buzz agent serve RELAY --exec CMD run a parallel backlog scheduler (worktree-isolated) | |Marmot (MLS group messaging): diff --git a/cli/src/main/kotlin/com/vitorpamplona/amethyst/cli/commands/BuzzCommands.kt b/cli/src/main/kotlin/com/vitorpamplona/amethyst/cli/commands/BuzzCommands.kt index 19aa4502f1..8d1f37a498 100644 --- a/cli/src/main/kotlin/com/vitorpamplona/amethyst/cli/commands/BuzzCommands.kt +++ b/cli/src/main/kotlin/com/vitorpamplona/amethyst/cli/commands/BuzzCommands.kt @@ -81,6 +81,7 @@ object BuzzCommands { |amy buzz dm hide RELAY CHANNEL hide a DM from my sidebar (kind-41012) |amy buzz dm add-member RELAY CHANNEL PUBKEY add a member to a group DM (kind-41011) |amy buzz job … file/list/show/cancel agent jobs (43001-43006) + |amy buzz workflow … trigger/run/approve Buzz workflows (30620/46020/46030) |amy buzz agent serve RELAY --exec CMD run a parallel backlog scheduler (worktree-isolated) """.trimMargin() @@ -101,6 +102,7 @@ object BuzzCommands { "personas" to { rest -> personas(dataDir, rest) }, "dm" to { rest -> dm(dataDir, rest) }, "job" to { rest -> BuzzJobCommands.dispatch(dataDir, rest) }, + "workflow" to { rest -> BuzzWorkflowCommands.dispatch(dataDir, rest) }, "agent" to { rest -> BuzzAgentCommands.dispatch(dataDir, rest) }, ), ) diff --git a/cli/src/main/kotlin/com/vitorpamplona/amethyst/cli/commands/BuzzWorkflowCommands.kt b/cli/src/main/kotlin/com/vitorpamplona/amethyst/cli/commands/BuzzWorkflowCommands.kt new file mode 100644 index 0000000000..834082f3c3 --- /dev/null +++ b/cli/src/main/kotlin/com/vitorpamplona/amethyst/cli/commands/BuzzWorkflowCommands.kt @@ -0,0 +1,572 @@ +/* + * Copyright (c) 2025 Vitor Pamplona + * + * Permission is hereby granted, free of charge, to any person obtaining a copy of + * this software and associated documentation files (the "Software"), to deal in + * the Software without restriction, including without limitation the rights to use, + * copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the + * Software, and to permit persons to whom the Software is furnished to do so, + * subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in all + * copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS + * FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR + * COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN + * AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION + * WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. + */ +package com.vitorpamplona.amethyst.cli.commands + +import com.vitorpamplona.amethyst.cli.Args +import com.vitorpamplona.amethyst.cli.Context +import com.vitorpamplona.amethyst.cli.DataDir +import com.vitorpamplona.amethyst.cli.Output +import com.vitorpamplona.amethyst.commons.model.buzz.WorkflowRun +import com.vitorpamplona.amethyst.commons.model.buzz.WorkflowRunAggregator +import com.vitorpamplona.amethyst.commons.model.buzz.WorkflowRunPayload +import com.vitorpamplona.amethyst.commons.model.buzz.WorkflowRunState +import com.vitorpamplona.quartz.buzz.workflow.ApprovalDenyEvent +import com.vitorpamplona.quartz.buzz.workflow.ApprovalGrantEvent +import com.vitorpamplona.quartz.buzz.workflow.WorkflowApprovalRequestedEvent +import com.vitorpamplona.quartz.buzz.workflow.WorkflowCancelledEvent +import com.vitorpamplona.quartz.buzz.workflow.WorkflowCompletedEvent +import com.vitorpamplona.quartz.buzz.workflow.WorkflowFailedEvent +import com.vitorpamplona.quartz.buzz.workflow.WorkflowStepCompletedEvent +import com.vitorpamplona.quartz.buzz.workflow.WorkflowStepStartedEvent +import com.vitorpamplona.quartz.buzz.workflow.WorkflowTriggerEvent +import com.vitorpamplona.quartz.buzz.workflow.WorkflowTriggeredEvent +import com.vitorpamplona.quartz.buzz.workflow.workflowChannel +import com.vitorpamplona.quartz.nip01Core.core.Event +import com.vitorpamplona.quartz.nip01Core.core.HexKey +import com.vitorpamplona.quartz.nip01Core.core.isValid +import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter +import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl +import com.vitorpamplona.quartz.nip01Core.signers.EventTemplate +import com.vitorpamplona.quartz.nip19Bech32.decodePublicKeyAsHexOrNull +import kotlinx.coroutines.Dispatchers +import kotlinx.coroutines.async +import kotlinx.coroutines.coroutineScope +import kotlinx.coroutines.delay +import kotlinx.coroutines.withContext +import kotlinx.serialization.encodeToString +import kotlinx.serialization.json.Json +import java.io.File + +/** + * `amy buzz workflow …` — Buzz's **source-confirmed** structured-work + human-approval primitive + * (kinds 30620 def, 46020 trigger, 46010 approval-requested, 46030/46031 grant/deny, 46001-46007 + * lifecycle), replacing the speculative agent-job protocol (43001-43006). + * + * On a real Buzz relay the RELAY parses the workflow YAML, runs the steps, and signs the + * lifecycle + approval events. Self-hosted on geode there is no workflow engine, so **`amy` is + * the runner** (`workflow run`) and emits the lifecycle events itself — a documented divergence. + * The approval gate is faithful: the runner does the agent's work, pauses on a 46010 (addressed to + * an approver), and only pushes/opens the PR after a human publishes a 46030 grant (46031 = deny). + * + * Correlation is simple: the **run id is the trigger's event id, and it doubles as the approval + * token**, so an `ApprovalGrant`'s `d` tag equals the run id. Run/step folding is the shared + * [WorkflowRunAggregator] in `commons`. + */ +object BuzzWorkflowCommands { + private val json = Json { ignoreUnknownKeys = true } + + private val USAGE = + """ + |amy buzz workflow trigger RELAY WFID --task TEXT --channel GID start a run (kind-46020) + |amy buzz workflow list RELAY --channel GID [--timeout SECS] list runs + their state + |amy buzz workflow show RELAY RUNID [--timeout SECS] one run's lifecycle + |amy buzz workflow approve RELAY RUNID [--note N] grant the approval gate (46030) + |amy buzz workflow deny RELAY RUNID [--note N] deny the approval gate (46031) + |amy buzz workflow run RELAY --exec CMD --channel GID run the workflow runner + | --approver NPUB [--on-approve CMD] [--worktree REPODIR] agent work → 46010 gate → + | [--base-ref REF] [--accept-from npub,…] [--poll SECS] [--once] on grant: --on-approve → 46005 + """.trimMargin() + + suspend fun dispatch( + dataDir: DataDir, + tail: Array, + ): Int = + route( + "buzz workflow", + tail, + USAGE, + mapOf( + "trigger" to { rest -> trigger(dataDir, rest) }, + "list" to { rest -> list(dataDir, rest) }, + "show" to { rest -> show(dataDir, rest) }, + "approve" to { rest -> decide(dataDir, rest, grant = true) }, + "deny" to { rest -> decide(dataDir, rest, grant = false) }, + "run" to { rest -> run(dataDir, rest) }, + ), + ) + + // ---- requester side ------------------------------------------------------ + + /** `workflow trigger RELAY WFID --task TEXT --channel GID` → publishes a kind-46020. */ + private suspend fun trigger( + dataDir: DataDir, + rest: Array, + ): Int { + val args = Args(rest) + val usage = "buzz workflow trigger RELAY WFID --task TEXT --channel GID" + val relayUrl = args.positionalOrNull(0) ?: return Output.error("bad_args", usage) + val wfId = args.positionalOrNull(1) ?: return Output.error("bad_args", usage) + val relay = normalizeGroupRelay(relayUrl) ?: return Output.error("bad_args", "invalid relay url: $relayUrl") + val task = args.flag("task")?.takeIf { it.isNotBlank() } ?: return Output.error("bad_args", "pass --task TEXT") + val channel = args.flag("channel") ?: return Output.error("bad_args", "pass --channel GID") + args.rejectUnknown("task", "channel") + + Context.open(dataDir).use { ctx -> + ctx.prepare() + val content = json.encodeToString(WorkflowRunPayload(task = task, workflow = wfId)) + val signed = ctx.signer.sign(WorkflowTriggerEvent.build(wfId, content) { workflowChannel(channel) }) + val ack = ctx.publish(signed, setOf(relay)) + RawEventSupport.publishGuard(ack, signed.id)?.let { return it } + Output.emit( + mapOf( + "run_id" to signed.id, // the trigger id IS the run id and the approval token + "workflow" to wfId, + "channel" to channel, + "relay" to relay.url, + "published" to ack.values.any { it.accepted }, + ), + ) + return 0 + } + } + + /** `workflow approve|deny RELAY RUNID [--note N]` → publishes a kind-46030/46031 with `d` = run id. */ + private suspend fun decide( + dataDir: DataDir, + rest: Array, + grant: Boolean, + ): Int { + val args = Args(rest) + val verb = if (grant) "approve" else "deny" + val usage = "buzz workflow $verb RELAY RUNID [--note N]" + val relayUrl = args.positionalOrNull(0) ?: return Output.error("bad_args", usage) + val runId = args.positionalOrNull(1) ?: return Output.error("bad_args", usage) + val relay = normalizeGroupRelay(relayUrl) ?: return Output.error("bad_args", "invalid relay url: $relayUrl") + val note = args.flag("note") ?: "" + args.rejectUnknown("note") + + Context.open(dataDir).use { ctx -> + ctx.prepare() + val template = if (grant) ApprovalGrantEvent.build(runId, note) else ApprovalDenyEvent.build(runId, note) + val signed = ctx.signer.sign(template) + val ack = ctx.publish(signed, setOf(relay)) + RawEventSupport.publishGuard(ack, signed.id)?.let { return it } + Output.emit(mapOf("event_id" to signed.id, "kind" to signed.kind, "run_id" to runId, "decision" to verb, "published" to ack.values.any { it.accepted })) + return 0 + } + } + + private suspend fun list( + dataDir: DataDir, + rest: Array, + ): Int { + val args = Args(rest) + val usage = "buzz workflow list RELAY --channel GID [--timeout SECS]" + val relayUrl = args.positionalOrNull(0) ?: return Output.error("bad_args", usage) + val relay = normalizeGroupRelay(relayUrl) ?: return Output.error("bad_args", "invalid relay url: $relayUrl") + val channel = args.flag("channel") ?: return Output.error("bad_args", "pass --channel GID") + val timeoutSecs = args.flag("timeout")?.toLongOrNull() ?: 8 + args.rejectUnknown("channel", "timeout") + + Context.open(dataDir).use { ctx -> + ctx.prepare() + val runs = WorkflowRunAggregator.byPriority(fetchRuns(ctx, relay, channel, timeoutSecs)) + Output.emit(mapOf("relay" to relay.url, "count" to runs.size, "runs" to runs.map { it.toRow() })) + return 0 + } + } + + private suspend fun show( + dataDir: DataDir, + rest: Array, + ): Int { + val args = Args(rest) + val usage = "buzz workflow show RELAY RUNID [--timeout SECS]" + val relayUrl = args.positionalOrNull(0) ?: return Output.error("bad_args", usage) + val runId = args.positionalOrNull(1) ?: return Output.error("bad_args", usage) + val relay = normalizeGroupRelay(relayUrl) ?: return Output.error("bad_args", "invalid relay url: $relayUrl") + val timeoutSecs = args.flag("timeout")?.toLongOrNull() ?: 8 + args.rejectUnknown("timeout") + + Context.open(dataDir).use { ctx -> + ctx.prepare() + // The run id is the trigger's id; the lifecycle events carry the run id only in their JSON + // `content` (+ an `h` channel tag), so we can't query them by `#e`/`#d`. Resolve the channel + // from the trigger, then fold the whole channel — the same path `list` takes. + val trigger = + ctx + .drain(mapOf(relay to listOf(Filter(kinds = listOf(WorkflowTriggerEvent.KIND), ids = listOf(runId)))), timeoutSecs * 1000, pendingOnAuthRequired = true) + .map { it.second } + .filterIsInstance() + .firstOrNull { it.id == runId } + ?: return Output.error("not_found", "no workflow trigger $runId on ${relay.url}") + val channel = trigger.tags.workflowChannel() ?: return Output.error("not_found", "trigger $runId has no channel") + val run = + fetchRuns(ctx, relay, channel, timeoutSecs).firstOrNull { it.runId == runId } + ?: return Output.error("not_found", "no run $runId on ${relay.url}") + Output.emit(run.toRow() + mapOf("relay" to relay.url)) + return 0 + } + } + + /** + * All workflow events on a channel, folded into runs. Two-phase: the trigger + lifecycle events + * are scoped by the channel `h` tag, but the approval grant/deny events (46030/46031) reference + * their run only by a `d` tag (= the token = the run id) — and quartz's shared event store routes + * every `#d` filter to the addressable `d_tag` column, which is NULL for a regular kind like + * 46030, so they can't be fetched by `#d`. Instead we fetch them by **author** — every 46010 gate + * names its approver in a `p` tag, so those are exactly the keys that can sign a decision — and + * the aggregator matches each decision to its run by the `d`-tag token it reads off the event. + */ + private suspend fun fetchRuns( + ctx: Context, + relay: NormalizedRelayUrl, + channel: String, + timeoutSecs: Long, + ): List { + val base = + ctx + .drain(mapOf(relay to listOf(Filter(kinds = listOf(WorkflowTriggerEvent.KIND) + LIFECYCLE_KINDS, tags = mapOf("h" to listOf(channel))))), timeoutSecs * 1000, pendingOnAuthRequired = true) + .map { it.second } + val approvers = + base + .filterIsInstance() + .mapNotNull { it.approver() } + .distinct() + val decisions = + if (approvers.isEmpty()) { + emptyList() + } else { + ctx + .drain(mapOf(relay to listOf(Filter(kinds = DECISION_KINDS, authors = approvers))), timeoutSecs * 1000, pendingOnAuthRequired = true) + .map { it.second } + } + return WorkflowRunAggregator.aggregate(base + decisions) + } + + private fun WorkflowRun.toRow(): Map = + mapOf( + "run_id" to runId, + "state" to state.name.lowercase(), + "workflow" to workflowId, + "channel" to channel, + "task" to task, + "requester" to requester, + "pending_approver" to pendingApprover, + "approval_token" to if (state == WorkflowRunState.AWAITING_APPROVAL) approvalToken else null, + "result" to result, + "error" to error, + "last_step" to lastStep, + "created_at" to createdAt, + "updated_at" to updatedAt, + ) + + // ---- runner -------------------------------------------------------------- + + private class AwaitingRun( + val channel: String, + val requester: HexKey?, + val worktree: String?, + val branch: String?, + ) + + /** The worktree path/branch for a run are deterministic from its id, so a restarted runner + * (or a `--once` resolve pass in a fresh process) can rebuild the [AwaitingRun] it lost. */ + private fun worktreeDirFor(runId: HexKey): File = File(System.getProperty("java.io.tmpdir"), "buzz-runs/${runId.take(12)}") + + private fun branchFor(runId: HexKey): String = "claude/run-${runId.take(12)}" + + private suspend fun run( + dataDir: DataDir, + rest: Array, + ): Int { + val args = Args(rest) + val relayUrl = args.positionalOrNull(0) ?: return Output.error("bad_args", USAGE) + val relay = normalizeGroupRelay(relayUrl) ?: return Output.error("bad_args", "invalid relay url: $relayUrl") + val exec = args.flag("exec") ?: return Output.error("bad_args", "pass --exec CMD (the agent's work step)") + val channel = args.flag("channel") ?: return Output.error("bad_args", "pass --channel GID") + val approverInput = args.flag("approver") ?: return Output.error("bad_args", "pass --approver NPUB (who signs off the gate)") + val approver = + decodePublicKeyAsHexOrNull(approverInput.trim())?.takeIf { it.isValid() } + ?: return Output.error("bad_args", "invalid --approver key: $approverInput") + val onApprove = args.flag("on-approve") // push + open PR; runs in the worktree after grant + val worktreeBase = args.flag("worktree") + val baseRef = args.flag("base-ref") ?: "HEAD" + val once = args.bool("once") + val pollSecs = args.flag("poll")?.toLongOrNull() ?: 5 + val timeoutSecs = args.flag("timeout")?.toLongOrNull() ?: 8 + val acceptFrom = + args + .flag("accept-from") + ?.split(",") + ?.mapNotNull { it.trim().ifBlank { null } } + ?.map { + decodePublicKeyAsHexOrNull(it)?.takeIf { hex -> hex.isValid() } ?: return Output.error("bad_args", "invalid --accept-from key: $it") + }?.toSet() + args.rejectUnknown("exec", "channel", "approver", "on-approve", "worktree", "base-ref", "once", "poll", "timeout", "accept-from") + + if (worktreeBase != null && !File(worktreeBase).isDirectory) return Output.error("bad_args", "--worktree is not a directory: $worktreeBase") + + Context.open(dataDir).use { ctx -> + ctx.prepare() + val me = ctx.identity.pubKeyHex + val started = mutableSetOf() // triggers we've begun + val awaiting = mutableMapOf() // runId -> worktree while at the gate + val decided = mutableSetOf() + + // Seed from existing runs so a restart doesn't re-run finished work. A run the runner has + // already carried to a terminal outcome (COMPLETED/FAILED/CANCELLED) is done. A run still + // needing the runner to act — parked at the gate (AWAITING_APPROVAL), granted-but-not-yet + // -shipped (APPROVED), or denied-but-its-worktree-still-around (DENIED) — is rebuilt into + // `awaiting` from its run id (the worktree/branch are deterministic) so a decision arriving + // in a later poll (or a fresh `--once` process) still resolves. The in-memory `awaiting` + // map is a cache, not the source of truth — the relay is. + fetchRuns(ctx, relay, channel, timeoutSecs).forEach { runv -> + if (runv.state != WorkflowRunState.TRIGGERED) started.add(runv.runId) + val ch = runv.channel + when (runv.state) { + WorkflowRunState.COMPLETED, WorkflowRunState.FAILED, WorkflowRunState.CANCELLED -> + decided.add(runv.runId) + WorkflowRunState.AWAITING_APPROVAL, WorkflowRunState.APPROVED, WorkflowRunState.DENIED -> + if (ch != null) { + awaiting[runv.runId] = + AwaitingRun( + channel = ch, + requester = runv.requester, + worktree = worktreeBase?.let { worktreeDirFor(runv.runId).absolutePath }, + branch = worktreeBase?.let { branchFor(runv.runId) }, + ) + } + WorkflowRunState.TRIGGERED, WorkflowRunState.RUNNING -> Unit + } + } + + if (!once) { + Output.emit(mapOf("running" to me, "relay" to relay.url, "channel" to channel, "approver" to approver, "seeded" to started.size)) + System.err.println("[workflow] runner up on ${relay.url} #$channel — gate → $approver — Ctrl-C to stop") + } + + val summary = mutableListOf>() + while (true) { + // 1. Start new triggers (agent work → 46010 gate). + fetchRuns(ctx, relay, channel, timeoutSecs) + .filter { it.state == WorkflowRunState.TRIGGERED && it.runId !in started } + .filter { acceptFrom == null || it.requester in acceptFrom } + .forEach { runv -> + started.add(runv.runId) + val a = startRun(ctx, relay, me, runv, exec, approver, worktreeBase, baseRef, timeoutSecs) + if (a != null) awaiting[runv.runId] = a else decided.add(runv.runId) + summary.add(mapOf("run_id" to runv.runId, "stage" to if (a != null) "awaiting_approval" else "failed")) + } + + // 2. Resolve gates whose decision has arrived. Grants/denies reference the run only by + // a `d`-tag token, which quartz's store can't serve via `#d` on a regular kind, so we + // fetch the approver's decisions by author and match the token to a run at the gate. + val decisions = + if (awaiting.isEmpty()) { + emptyList() + } else { + ctx.drain(mapOf(relay to listOf(Filter(kinds = DECISION_KINDS, authors = listOf(approver)))), timeoutSecs * 1000, pendingOnAuthRequired = true).map { it.second } + } + decisions.forEach { d -> + val (runId, granted) = + when (d) { + is ApprovalGrantEvent -> (d.tokenHash() ?: return@forEach) to true + is ApprovalDenyEvent -> (d.tokenHash() ?: return@forEach) to false + else -> return@forEach + } + val a = awaiting[runId] ?: return@forEach + if (runId in decided) return@forEach + decided.add(runId) + awaiting.remove(runId) + resolve(ctx, relay, runId, a, granted, onApprove, worktreeBase, timeoutSecs) + summary.add(mapOf("run_id" to runId, "stage" to if (granted) "completed" else "denied")) + if (!once) System.err.println("[workflow] ${if (granted) "granted → shipped" else "denied"} run ${runId.take(12)}…") + } + + if (once) { + Output.emit(mapOf("relay" to relay.url, "handled" to summary.size, "runs" to summary)) + return 0 + } + delay(pollSecs * 1000) + } + } + @Suppress("UNREACHABLE_CODE") + return 0 + } + + /** Emit 46001/46002, run --exec in a worktree, emit 46003 + the 46010 gate. Null if the work failed. */ + private suspend fun startRun( + ctx: Context, + relay: NormalizedRelayUrl, + me: HexKey, + runv: WorkflowRun, + exec: String, + approver: HexKey, + worktreeBase: String?, + baseRef: String, + timeoutSecs: Long, + ): AwaitingRun? { + val channel = runv.channel ?: return null + val runId = runv.runId + emit(ctx, relay, WorkflowTriggeredEvent.build(channel, payload(WorkflowRunPayload(run = runId, workflow = runv.workflowId, task = runv.task)))) + emit(ctx, relay, WorkflowStepStartedEvent.build(channel, payload(WorkflowRunPayload(run = runId, step = "build")))) + + var workdir: String? = null + var branch: String? = null + if (worktreeBase != null) { + branch = branchFor(runId) + val wt = worktreeDirFor(runId) + wt.parentFile?.mkdirs() + git(worktreeBase, "worktree", "prune") + wt.deleteRecursively() + val add = git(worktreeBase, "worktree", "add", "-B", branch, wt.absolutePath, baseRef) + if (add.exit != 0) { + emit(ctx, relay, WorkflowFailedEvent.build(channel, payload(WorkflowRunPayload(run = runId, error = "worktree: ${add.err.take(500)}")))) + return null + } + workdir = wt.absolutePath + } + + val env = + buildMap { + put("BUZZ_RUN", runId) + put("BUZZ_CHANNEL", channel) + put("BUZZ_RELAY", relay.url) + put("BUZZ_AGENT", me) + runv.requester?.let { put("BUZZ_REQUESTER", it) } + branch?.let { put("BUZZ_BRANCH", it) } + workdir?.let { put("BUZZ_WORKTREE", it) } + put("BUZZ_BASE_REF", baseRef) + } + val work = exec(exec, runv.task ?: "", env, workdir) + if (work.exit != 0) { + emit(ctx, relay, WorkflowFailedEvent.build(channel, payload(WorkflowRunPayload(run = runId, error = (work.err.ifBlank { work.out }).take(MAX))))) + return null + } + emit(ctx, relay, WorkflowStepCompletedEvent.build(channel, payload(WorkflowRunPayload(run = runId, step = "build")))) + // Pause on the approval gate — a human reviews the work before it ships. + emit(ctx, relay, WorkflowApprovalRequestedEvent.build(channel, approver, payload(WorkflowRunPayload(run = runId, note = work.out.take(1000))))) + return AwaitingRun(channel, runv.requester, workdir, branch) + } + + /** On grant: run --on-approve (push + PR) and emit 46005 with its output. On deny: just discard. */ + private suspend fun resolve( + ctx: Context, + relay: NormalizedRelayUrl, + runId: HexKey, + a: AwaitingRun, + granted: Boolean, + onApprove: String?, + worktreeBase: String?, + timeoutSecs: Long, + ) { + try { + // A deny (46031) is itself the terminal signal — the aggregator folds it to DENIED — so the + // runner only discards the unshipped work (the `finally` removes the worktree); emitting a + // competing 46007 cancelled would just race the deny for "newest terminal". + if (!granted) return + val prUrl = + if (onApprove != null) { + val env = + buildMap { + put("BUZZ_RUN", runId) + a.branch?.let { put("BUZZ_BRANCH", it) } + a.worktree?.let { put("BUZZ_WORKTREE", it) } + } + val r = exec(onApprove, "", env, a.worktree) + if (r.exit != 0) { + emit(ctx, relay, WorkflowFailedEvent.build(a.channel, payload(WorkflowRunPayload(run = runId, error = "on-approve: ${(r.err.ifBlank { r.out }).take(MAX)}")))) + return + } + r.out.trim().takeIf { it.isNotBlank() } + } else { + null + } + emit(ctx, relay, WorkflowCompletedEvent.build(a.channel, payload(WorkflowRunPayload(run = runId, pr = prUrl)))) + } finally { + // The worktree lives under the runner's tmpdir but *belongs* to the repo at worktreeBase, + // so `git worktree remove` must run against that repo, not the worktree's own parent path. + if (a.worktree != null && worktreeBase != null) { + git(worktreeBase, "worktree", "remove", "--force", a.worktree) + git(worktreeBase, "worktree", "prune") + } + } + } + + // ---- helpers ------------------------------------------------------------- + + private fun payload(p: WorkflowRunPayload): String = json.encodeToString(p) + + private suspend fun emit( + ctx: Context, + relay: NormalizedRelayUrl, + template: EventTemplate, + ) { + val signed = ctx.signer.sign(template) + ctx.publish(signed, setOf(relay)) + } + + private class ExecResult( + val exit: Int, + val out: String, + val err: String, + ) + + private suspend fun exec( + cmd: String, + input: String, + env: Map, + workdir: String?, + ): ExecResult = + withContext(Dispatchers.IO) { + val pb = ProcessBuilder("sh", "-c", cmd) + workdir?.let { pb.directory(File(it)) } + pb.environment().putAll(env) + val proc = pb.start() + coroutineScope { + val out = async { proc.inputStream.readBytes().decodeToString() } + val err = async { proc.errorStream.readBytes().decodeToString() } + // Feed the task on stdin, but a command that never reads it (e.g. `printf …`) exits + // and closes the pipe first — tolerate the resulting broken pipe rather than abort. + runCatching { proc.outputStream.use { it.write(input.encodeToByteArray()) } } + proc.waitFor() + ExecResult(proc.exitValue(), out.await(), err.await()) + } + } + + private suspend fun git( + dir: String, + vararg gitArgs: String, + ): ExecResult = + withContext(Dispatchers.IO) { + val proc = ProcessBuilder(listOf("git", "-C", dir) + gitArgs).start() + coroutineScope { + val out = async { proc.inputStream.readBytes().decodeToString() } + val err = async { proc.errorStream.readBytes().decodeToString() } + proc.waitFor() + ExecResult(proc.exitValue(), out.await(), err.await()) + } + } + + private const val MAX = 60_000 + private val LIFECYCLE_KINDS = + listOf( + WorkflowTriggeredEvent.KIND, + WorkflowStepStartedEvent.KIND, + WorkflowStepCompletedEvent.KIND, + WorkflowApprovalRequestedEvent.KIND, + WorkflowCompletedEvent.KIND, + WorkflowFailedEvent.KIND, + WorkflowCancelledEvent.KIND, + ) + private val DECISION_KINDS = listOf(ApprovalGrantEvent.KIND, ApprovalDenyEvent.KIND) +} diff --git a/cli/tests/.gitignore b/cli/tests/.gitignore index db90bb8311..77d0da270e 100644 --- a/cli/tests/.gitignore +++ b/cli/tests/.gitignore @@ -8,3 +8,4 @@ sync/state-sync-deletions/ blossom/state-blossom-live/ git/state-git-nip34/ buzz/state-job-loop/ +buzz/state-workflow-loop/ diff --git a/cli/tests/buzz/workflow-loop.sh b/cli/tests/buzz/workflow-loop.sh new file mode 100755 index 0000000000..784fea690b --- /dev/null +++ b/cli/tests/buzz/workflow-loop.sh @@ -0,0 +1,148 @@ +#!/usr/bin/env bash +# +# workflow-loop.sh — self-contained headless test for the Buzz workflow loop. +# +# Buzz's *source-confirmed* structured-work + human-approval primitive (kinds +# 30620 def / 46020 trigger / 46030-46031 grant-deny / 46001-46007 lifecycle), +# driven end to end through an embedded relay (`amy serve`, i.e. geode — no +# external binary). Three `amy` accounts: +# +# alice = requester (triggers the workflow, kind-46020) +# bot = the runner (`amy buzz workflow run`; does agent work, posts the +# 46010 approval gate, then on grant runs --on-approve → 46005) +# carol = the approver (publishes 46030 grant / 46031 deny) +# +# Because the run id IS the trigger event id (and doubles as the approval +# token), carol can grant/deny without any extra token bookkeeping. +# +# 1. alice triggers → run_id +# 2. bot run --once → agent work → 46010 gate (state=awaiting_approval) +# 3. carol approves → 46030 (d = run_id) +# 4. bot run --once → resolves the gate → --on-approve → 46005 completed +# 5. alice show → WorkflowRunAggregator folds it to state=completed + PR +# 6. deny path: a second run, carol denies → state=denied (work discarded) +# +# Usage: ./workflow-loop.sh [--port N] [--no-build] +# +set -uo pipefail + +SCRIPT_DIR="$(cd -- "$(dirname -- "${BASH_SOURCE[0]}")" && pwd)" +REPO_ROOT="$(cd -- "$SCRIPT_DIR/../../.." && pwd)" +STATE_DIR="$SCRIPT_DIR/state-workflow-loop" +AMY_BIN="$REPO_ROOT/cli/build/install/amy/bin/amy" + +PORT=7788 +BUILD=1 +while [[ $# -gt 0 ]]; do + case "$1" in + --port) PORT="$2"; shift 2 ;; + --no-build) BUILD=0; shift ;; + *) echo "unknown arg: $1" >&2; exit 2 ;; + esac +done +RELAY="ws://127.0.0.1:$PORT" +CHANNEL="3f2504e0-4f89-41d3-9a0c-0305e82c3301" +WFID="agent-build" + +if [[ $BUILD -eq 1 ]]; then + echo "> building amy…" >&2 + (cd "$REPO_ROOT" && ./gradlew -q :cli:installDist) || { echo "build failed" >&2; exit 1; } +fi +[[ -x "$AMY_BIN" ]] || { echo "amy not built at $AMY_BIN (drop --no-build)" >&2; exit 1; } + +rm -rf "$STATE_DIR"; mkdir -p "$STATE_DIR" +REQ_HOME="$STATE_DIR/alice"; BOT_HOME="$STATE_DIR/bot"; APV_HOME="$STATE_DIR/carol" +mkdir -p "$REQ_HOME" "$BOT_HOME" "$APV_HOME" +RELAY_LOG="$STATE_DIR/relay.log" +PASS=0; FAIL=0 +RELAY_PID="" +cleanup() { [[ -n "$RELAY_PID" ]] && kill "$RELAY_PID" 2>/dev/null; } +trap cleanup EXIT + +run_req() { HOME="$REQ_HOME" "$AMY_BIN" --account alice --secret-backend plaintext --json "$@" 2>/dev/null; } +run_bot() { HOME="$BOT_HOME" "$AMY_BIN" --account bot --secret-backend plaintext --json "$@" 2>/dev/null; } +run_apv() { HOME="$APV_HOME" "$AMY_BIN" --account carol --secret-backend plaintext --json "$@" 2>/dev/null; } +jkey() { + python3 -c ' +import sys, json +v = json.load(sys.stdin).get("'"$1"'", "") +print(str(v).lower() if isinstance(v, bool) else v)' +} +check() { # check