From 635e7857534d4d427b9b8d692b5a2d4f3bd61660 Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 25 Jul 2026 17:18:30 +0000 Subject: [PATCH 01/23] =?UTF-8?q?feat(cli):=20buzz=20agent-job=20loop=20?= =?UTF-8?q?=E2=80=94=20file/track=20jobs=20+=20a=20driving=20responder?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Prototype the "human drives an AI coding agent to develop Amethyst" support channel on top of the existing block/buzz integration, all through amy. - commons: BuzzJobAggregator (BuzzJobs.kt) — a pure, tested folder that correlates the Buzz agent-job kinds (43001-43006) by their `e` request reference into JobView records with a REQUESTED→ACCEPTED→IN_PROGRESS→ COMPLETED/FAILED/CANCELLED state machine (newest terminal wins). Shared so a future mobile Jobs board reuses one correlation path. 9 unit tests. - cli: `amy buzz job request|list|show|cancel` (requester side) and `amy buzz agent serve --exec CMD` (the responder loop): polls for REQUESTED jobs targeting my key, gates intake on `--accept-from` (allowlist) and `--channel`, then accepts (43002) → progress (43003) → runs `sh -c CMD` (task text on stdin; BUZZ_JOB_ID/REQUESTER/CHANNEL/RELAY/AGENT in env) → result (43004) or error (43006). Point `--exec` at a coding agent to drive it. - cli/tests/buzz/job-loop.sh — self-contained headless harness over an embedded `amy serve` relay; asserts the full loop AND the permission gate (an allowlist excluding the requester handles nothing). - Design doc cli/plans/2026-07-25-buzz-agent-support-channel.md: the three-layer permission model (Buzz scopes by identity, not capability flags — so "can't merge/destroy main" lives in GitHub branch protection + the `--exec` credential, not the relay), the MVP architecture, and the prioritized mobile app gap list (approvals inbox, jobs board, diff/PR review, …). Schema caveat: kinds 43001-43006 are reserved in Buzz with no upstream builder; the tag layout is Quartz's best-effort model, to be reconciled upstream. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_011mApqAbr8vkLC7gUDjavu6 --- cli/README.md | 14 + cli/ROADMAP.md | 1 + .../2026-07-25-buzz-agent-support-channel.md | 137 +++++++++ .../com/vitorpamplona/amethyst/cli/Main.kt | 3 + .../cli/commands/BuzzAgentCommands.kt | 284 ++++++++++++++++++ .../amethyst/cli/commands/BuzzCommands.kt | 4 + .../amethyst/cli/commands/BuzzJobCommands.kt | 253 ++++++++++++++++ cli/tests/.gitignore | 1 + cli/tests/buzz/job-loop.sh | 101 +++++++ .../amethyst/commons/model/buzz/BuzzJobs.kt | 173 +++++++++++ .../model/buzz/BuzzJobAggregatorTest.kt | 186 ++++++++++++ 11 files changed, 1157 insertions(+) create mode 100644 cli/plans/2026-07-25-buzz-agent-support-channel.md create mode 100644 cli/src/main/kotlin/com/vitorpamplona/amethyst/cli/commands/BuzzAgentCommands.kt create mode 100644 cli/src/main/kotlin/com/vitorpamplona/amethyst/cli/commands/BuzzJobCommands.kt create mode 100755 cli/tests/buzz/job-loop.sh create mode 100644 commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/model/buzz/BuzzJobs.kt create mode 100644 commons/src/commonTest/kotlin/com/vitorpamplona/amethyst/commons/model/buzz/BuzzJobAggregatorTest.kt diff --git a/cli/README.md b/cli/README.md index 00dd84a862..e0723b78c6 100644 --- a/cli/README.md +++ b/cli/README.md @@ -610,6 +610,20 @@ and `commons` aggregator the app uses. | `amy buzz dm open RELAY PUBKEY [PUBKEY…]` | Open (or re-surface) a DM with 1-8 people (kind:41010). The relay assigns the channel id and confirms via 41001. | | `amy buzz dm hide RELAY CHANNEL` | Hide a DM from my sidebar (kind:41012); re-opening it un-hides. | | `amy buzz dm add-member RELAY CHANNEL PUBKEY` | Add a member to an existing group DM (kind:41011). | +| `amy buzz job request RELAY [--agent PUBKEY] [--channel GID]` | File an agent job (kind:43001): ask an agent to do a task, optionally targeting an agent (`p`) and/or scoping to a channel (`h`). | +| `amy buzz job list RELAY [--channel GID] [--mine\|--assigned] [--limit N] [--timeout SECS]` | List jobs and their folded state (REQUESTED/ACCEPTED/IN_PROGRESS/COMPLETED/FAILED/CANCELLED). `--mine` = jobs I requested; `--assigned` = jobs targeting me. | +| `amy buzz job show RELAY JOBID [--timeout SECS]` | Show one job's full lifecycle (request + every reply, folded). | +| `amy buzz job cancel RELAY JOBID [--reason R] [--channel GID]` | Cancel a job (kind:43005). | +| `amy buzz agent serve RELAY --exec CMD [--channel GID] [--accept-from npub,…] [--claim-untargeted] [--poll SECS] [--exec-timeout SECS] [--no-progress] [--dry-run] [--once]` | Run an agent job-responder loop. Polls for REQUESTED jobs targeting my key, gated by `--accept-from` (allowlist) and `--channel`, then per job publishes accept (43002) → progress (43003) → runs `sh -c CMD` (task text on stdin; `BUZZ_JOB_ID/REQUESTER/CHANNEL/RELAY/AGENT` in env) → result (43004) or error (43006). Point `--exec` at a coding agent to drive it. | + +> **Agent-job schema is provisional.** Kinds 43001-43006 are *reserved* in Buzz with no +> 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). +> +> **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 +> `--exec` (keep its git token PR-only) and by branch-protecting `main` — not by Buzz. ### Concord Channels (encrypted communities) diff --git a/cli/ROADMAP.md b/cli/ROADMAP.md index e785b1e2e0..b8dffda594 100644 --- a/cli/ROADMAP.md +++ b/cli/ROADMAP.md @@ -60,6 +60,7 @@ Status legend: ✅ shipped · 📦 logic lives in `commons/`, needs a command · | NIP-25 reactions | ✅ in groups · 🆕 elsewhere | `marmot message react` covers MLS group reactions; outer-event reactions still pending. | | NIP-29 relay groups (`amy relaygroup`) | ✅ | `RelayGroupCommands` — list/browse/info/create/join/leave/message/edit/invite/put-user/remove-user against a host relay; kind:10009 joined-list kept in sync. | | Buzz workspaces (`amy buzz`) | ✅ | `BuzzCommands` — post/read the kind:40002 stream timeline, `attest` (offline NIP-OA), `console` (decrypt+aggregate kind:44200 turn metrics via the shared `AgentFleetAggregator`), `personas` (kind:30175). Join/leave reuse `amy relaygroup` (Buzz workspaces are NIP-29 groups). | +| Buzz agent jobs (`amy buzz job` / `agent serve`) | ✅ | `BuzzJobCommands` (request/list/show/cancel, kinds 43001-43006) + `BuzzAgentCommands` (`agent serve` responder loop: gate on `--accept-from`, run `--exec`, report accept/progress/result/error). Correlation + state via the shared `BuzzJobAggregator` in `commons`. Schema provisional (43001-43006 reserved upstream). See `cli/plans/2026-07-25-buzz-agent-support-channel.md`. | | NIP-51 lists (bookmarks, mute, follow sets) | 🆕 | `amethyst/model/nip51Lists/` | | NIP-57 zaps (send) | ✅ partial | `ZapCommand` — `zap user`/`zap event` build the kind:9734 request and fetch the BOLT11 (zap splits honored, one invoice per recipient); `--with NDEBIT` auto-pays through a CLINK debit pointer. Receipt (kind:9735) verification still 🆕. | | BOLT12 zaps (NIP-XX, kinds 9736/9737/10058) | ✅ partial | `Bolt12Commands` + `Bolt12SendCommands` over shared `commons` `Bolt12ZapActions` — `bolt12 decode` (offer/proof), `verify` (validate a kind:9736), `offer get/set` (kind:10058), and the two-step send `intent`→`zap` (out-of-band proof, since amy has no NWC rail). Interop harness + NWC-fetched proofs still 🆕. | diff --git a/cli/plans/2026-07-25-buzz-agent-support-channel.md b/cli/plans/2026-07-25-buzz-agent-support-channel.md new file mode 100644 index 0000000000..4335f7b747 --- /dev/null +++ b/cli/plans/2026-07-25-buzz-agent-support-channel.md @@ -0,0 +1,137 @@ +# Buzz-driven agent support channel for Amethyst + +**Date:** 2026-07-25 +**Status:** prototype landing (CLI) · mobile gaps scoped +**Owning module:** `cli/` (with a shared aggregator in `commons/`) + +## Goal + +Let the Amethyst team (and, gated, users) drive an AI coding agent — Claude Code +running as *this* Anthropic account — to develop Amethyst, over a self-hosted +[`block/buzz`](https://github.com/block/buzz) workspace. Every request and result is a +signed, audited Nostr event; the agent opens PRs but can never merge or damage `main`. + +## Why Buzz is the right substrate (and what it is NOT) + +Buzz is a self-hosted Nostr relay that acts as a workspace where humans and agents share +rooms; Amethyst already models ~78 of its kinds (`quartz/.../buzz/`) plus client UI (agent +console, workspaces, DMs, attestations — shipped in v1.13.0). Upstream, Buzz ships +`buzz-acp`, an ACP harness that already plugs **Claude Code** (and Goose/Codex) in as the +agent runner, and produces code as **NIP-34 patches / git diffs / PRs** — the same flow +this repo's `claude/*` branches already use. + +What already exists in-repo to build on: + +| Layer | Status | +|---|---| +| Workspace = a relay you own; channels/threads/canvas | app + `amy buzz post/read` | +| DMs to an agent key (open/hide/add-member/list) | app + `amy buzz dm …` | +| Agent authorization — NIP-OA owner attestation (virtual membership) | `AgentAttestationScreen` + `amy buzz attest` | +| Agent config — personas (30175), managed agents (30177), agent profiles (10100) | quartz models + persona editor | +| Cost/activity telemetry — turn metrics (44200), observer (24200) | Agent Console + `amy buzz console` | +| Code changes in the room — diff (40008), NIP-34 patches | rendered in chat | +| Human-in-the-loop gate — workflow approval (46010/46030/46031) | quartz models only, no UI | +| Structured jobs — 43001-43006 | quartz models + EventFactory dispatch; **no client surface (this plan)** | + +### The permission reality — the crux + +Buzz authorizes **by identity, not by capability flags**. Its entire vocabulary is coarse: +membership + `owner`/`admin`/`member` roles, NIP-OA conditions limited to a single `kind` +and `created_at` before/after bounds, and per-agent `respond_to` / `channel_add_policy` +gates. **There is no way in Buzz to express "may push but not merge" or "only this repo."** +So the constraints the task asks for live in **three layers**, and Buzz is only one: + +| Requirement | Enforced by | How | +|---|---|---| +| Can't merge/destroy `main` | **GitHub branch protection** (load-bearing) | Protect `main` (PR + review + green CI, no direct/force push, no branch delete). The agent runner's git credential can only open PRs on feature branches — never merge. | +| Can't use the agent to code other things | **Agent runtime + Buzz intake** | `--exec` checked out in `amethyst` only, scoped tools; persona system-prompt scopes the task; `--accept-from` allowlist = team npubs only. | +| Only the team can drive it | **Buzz** | Team npubs = relay members / the `--accept-from` allowlist. | +| Everything accountable | **Buzz** | Every request/progress/result is a signed event in the tenant's hash-chained audit log. | +| Human sign-off before risky actions | **Buzz workflow gate** | 46010 pause → 46030/46031 grant/deny by a designated approver key (two-signer; a run can't self-approve). | + +Honest blast radius: a Buzz-authorized agent key has member-level reach *on the relay* +only. Its reach into **code** is bounded entirely by the git credential handed to `--exec`. +Keep that credential minimal; branch protection is what actually stops a bad merge. + +## Architecture (MVP) + +1. **One `buzz-relay`** (Block's Rust relay — geode does NOT implement Buzz server + semantics: kind accept-list, `h`-scope, NIP-OA fallback, relay-signed metadata) = the + "Amethyst workspace" tenant. Team npubs enrolled as members; a maintainer is owner. +2. **One agent identity** = its own nostr key, authorized by a NIP-OA attestation the owner + issues (`amy buzz attest` / `AgentAttestationScreen`). On GitHub it authenticates with a + PR-only token; `main` is branch-protected. +3. **Intake:** a team member files a job in the `#build` channel (or DMs the agent). The + responder picks it up, runs a coding agent in an `amethyst` checkout, streams progress, + posts the result, and opens a PR on a `claude/*` branch. **Merge stays human.** +4. Optional **approval gate** (46010/46030/46031) for irreversible mid-run steps. + +## CLI prototype (this change) + +Thin assembly over quartz job models + a shared aggregator; no protocol logic in `cli/`. + +- **`commons/.../model/buzz/BuzzJobs.kt`** — `BuzzJobAggregator`, a pure, tested + (`BuzzJobAggregatorTest`, 9 cases) folder that correlates 43001-43006 events (by the + reply `e` → request id) into `JobView` records with a `JobState` machine + (REQUESTED→ACCEPTED→IN_PROGRESS→COMPLETED/FAILED/CANCELLED; newest terminal wins). Shared + so a future mobile Jobs board reuses one correlation path. +- **`amy buzz job request|list|show|cancel`** (`BuzzJobCommands.kt`) — the requester side: + file a 43001 (optional `--agent`, `--channel`), list/fold jobs (`--mine`/`--assigned`), + show one job's lifecycle, cancel (43005). +- **`amy buzz agent serve RELAY --exec CMD`** (`BuzzAgentCommands.kt`) — the responder loop. + Polls for REQUESTED jobs targeting my key, gates on `--accept-from` (allowlist) and + `--channel`, then per job: publishes 43002 accept → optional 43003 progress → runs + `sh -c CMD` (task text on stdin; `BUZZ_JOB_ID/REQUESTER/CHANNEL/RELAY/AGENT` in env) → + publishes 43004 result or 43006 error. `--dry-run`, `--once`, `--claim-untargeted`, + `--exec-timeout` for testing/ops. This is where Claude Code plugs in: `--exec 'claude -p'` + or a wrapper that runs the agent, opens the PR, and echoes the PR URL as the result. + +Guardrails restated in the command's KDoc: `--accept-from` is the Buzz-layer intake gate; +repo safety is the `--exec` credential + branch protection, not Buzz. + +### Schema caveat + +Kinds 43001-43006 are *reserved* in Buzz with no upstream builder; the tag layout +(`e`/`h`/`p`/`status`) is Quartz's best-effort model and must be reconciled once Buzz +implements the job protocol. The prototype is deliberately isolated so that reconciliation +touches only the quartz models + this aggregator. + +## Mobile app gaps (prioritized) + +The quartz layer + LocalCache ingest are complete for every kind; the app has **zero +create/interact surface** for the two kinds that define the workflow. Priorities: + +**P0 — the human-in-the-loop loop** +- **P0-1 Approvals inbox** — render 46010, publish 46030/46031 grant/deny (token-hash + correlation; 46010 is NIP-PL push-urgent). Build on `AgentConsoleViewModel` fetch pattern + + `ApprovalGrantEvent.build`/`ApprovalDenyEvent.build`. Size **M**. +- **P0-2 Jobs board** — create 43001, watch 43002-43006, see result, grouped by `e`. Reuse + the new `BuzzJobAggregator`; jobs already render as inline system rows + are subscribed in + `RelayGroupFilterBuilders`. Size **L**. +- **P0-3 Agent picker** — choose a target agent when filing a job/approval, from stored + 30177/10100 + the fleet list. Size **S**. + +**P1 — a credible agent-driving client** +- **P1-1 Diff/PR review surface** — upgrade read-only 40008 (`RenderBuzzDiff`) into a + full-screen per-file review whose approve action emits 46030/46031. Size **M**. +- **P1-2 Managed-agent (30177) editor** — clone `AgentPersonaEditScreen`. Size **M**. +- **P1-3 Persona `respond_to`/allowlist editing** — the safety gate for pointing a persona + at a support channel. Size **S–M**. +- **P1-4 Attestation persistence** — `BuzzHeldAttestations` is in-memory; survive restart. + Size **S–M**. + +**P2 — completeness**: agent-profile (10100) viewer; a stable "Agents" hub; +workflow-run timeline (46020 family, all stored, unrendered); turn-metric → job attribution. + +Key files: routes `amethyst/.../navigation/routes/Routes.kt`; render dispatch +`.../chats/feed/ChatMessageCompose.kt`; renderers `.../chats/feed/types/RenderBuzzNotes.kt`; +ingest `model/LocalCache.kt` (~L4780-4855); subscription +`.../relayGroup/datasource/RelayGroupFilterBuilders.kt`. + +## 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`. +3. A reference `--exec` wrapper that runs Claude Code, opens a PR with a PR-only token, and + returns the PR URL — plus a documented branch-protection + token-scope checklist. +4. Consider promoting the approval gate (46010) into the responder for irreversible steps. 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 5880237780..3da5a0a6d4 100644 --- a/cli/src/main/kotlin/com/vitorpamplona/amethyst/cli/Main.kt +++ b/cli/src/main/kotlin/com/vitorpamplona/amethyst/cli/Main.kt @@ -822,6 +822,9 @@ private fun printUsage() { | buzz attest AGENT [--kind K] issue a NIP-OA attestation (offline) | buzz console [--relays R,R] aggregate my kind-44200 agent turn metrics | 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 agent serve RELAY --exec CMD run an agent job-responder loop | |Marmot (MLS group messaging): | marmot key-package publish publish a fresh KeyPackage diff --git a/cli/src/main/kotlin/com/vitorpamplona/amethyst/cli/commands/BuzzAgentCommands.kt b/cli/src/main/kotlin/com/vitorpamplona/amethyst/cli/commands/BuzzAgentCommands.kt new file mode 100644 index 0000000000..95e820e236 --- /dev/null +++ b/cli/src/main/kotlin/com/vitorpamplona/amethyst/cli/commands/BuzzAgentCommands.kt @@ -0,0 +1,284 @@ +/* + * 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.JobState +import com.vitorpamplona.amethyst.commons.model.buzz.JobView +import com.vitorpamplona.quartz.buzz.jobs.JobAcceptedEvent +import com.vitorpamplona.quartz.buzz.jobs.JobErrorEvent +import com.vitorpamplona.quartz.buzz.jobs.JobProgressEvent +import com.vitorpamplona.quartz.buzz.jobs.JobResultEvent +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.normalizer.NormalizedRelayUrl +import com.vitorpamplona.quartz.nip01Core.signers.EventTemplate +import com.vitorpamplona.quartz.nip19Bech32.decodePublicKeyAsHexOrNull +import kotlinx.coroutines.Dispatchers +import kotlinx.coroutines.delay +import kotlinx.coroutines.withContext +import java.util.concurrent.TimeUnit + +/** + * `amy buzz agent …` — the AGENT side of the Buzz job protocol: a headless responder loop + * that turns an incoming job request (kind-43001 targeting my key) into a spawned command, + * and reports the outcome back as accept/progress/result/error events (43002/43003/43004/43006). + * + * This is the "drive this agent to build Amethyst" prototype: point `--exec` at a coding + * agent (e.g. `claude -p`, a Goose/Codex wrapper, or a build script). The job's task text is + * piped to the command's stdin; `BUZZ_JOB_ID`, `BUZZ_REQUESTER`, `BUZZ_CHANNEL`, and + * `BUZZ_RELAY` are exported into its environment; its stdout becomes the job result. + * + * PERMISSIONS — read this. Buzz authorizes by identity, not by capability flags, so this + * responder is only as safe as the two guardrails around it: + * 1. `--accept-from` is the intake gate: only listed requester keys are obeyed (your team's + * npubs). Without it the agent answers anyone who can post to the relay. + * 2. What `--exec` can DO to a repo is bounded entirely by the credentials/tooling you give + * that command — NOT by Buzz. Keep its git token scoped to open PRs on feature branches + * (never merge, never force-push), and branch-protect `main`. See + * `cli/plans/2026-07-25-buzz-agent-support-channel.md`. + * + * SCHEMA CAVEAT: kinds 43001-43006 are *reserved* in Buzz with no upstream builder; see + * [com.vitorpamplona.quartz.buzz.jobs.JobRequestEvent]. + */ +object BuzzAgentCommands { + private val USAGE = + """ + |amy buzz agent serve RELAY --exec CMD run a job-responder loop + | [--channel GID] only handle jobs scoped to this channel + | [--accept-from npub,npub] allowlist of requester keys (STRONGLY advised) + | [--claim-untargeted] also handle jobs with no `p` target + | [--poll SECS] poll interval (default 5) + | [--exec-timeout SECS] kill --exec after N seconds (default 1800; 0 = none) + | [--timeout SECS] per-fetch relay timeout (default 8) + | [--no-progress] don't post a 43003 "working" ping + | [--dry-run] accept + canned result, never run --exec + | [--once] drain currently-pending jobs, then exit + """.trimMargin() + + suspend fun dispatch( + dataDir: DataDir, + tail: Array, + ): Int = + route( + "buzz agent", + tail, + USAGE, + mapOf( + "serve" to { rest -> serve(dataDir, rest) }, + ), + ) + + private suspend fun serve( + 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 dryRun = args.bool("dry-run") + val exec = args.flag("exec") + if (exec == null && !dryRun) return Output.error("bad_args", "pass --exec CMD (or --dry-run to test without running anything)") + val channel = args.flag("channel") + val claimUntargeted = args.bool("claim-untargeted") + val postProgress = !args.bool("no-progress") + val once = args.bool("once") + val pollSecs = args.flag("poll")?.toLongOrNull() ?: 5 + val timeoutSecs = args.flag("timeout")?.toLongOrNull() ?: 8 + val execTimeoutSecs = args.flag("exec-timeout")?.toLongOrNull() ?: 1800 + 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 (npub or 64-char hex): $it") + }?.toSet() + args.rejectUnknown( + "exec", + "channel", + "accept-from", + "claim-untargeted", + "poll", + "exec-timeout", + "timeout", + "no-progress", + "dry-run", + "once", + ) + + Context.open(dataDir).use { ctx -> + ctx.prepare() + val me = ctx.identity.pubKeyHex + + // Seed the handled set from jobs already accepted/terminated so a restart doesn't + // re-run work. Any job past REQUESTED — someone (maybe a previous run of me) + // already picked it up — is treated as done for intake purposes. + val handled = mutableSetOf() + BuzzJobCommands.fetchJobs(ctx, relay, channel, timeoutSecs).forEach { job -> + if (job.state != JobState.REQUESTED) handled.add(job.jobId) + } + + if (!once) { + Output.emit( + mapOf( + "serving" to me, + "relay" to relay.url, + "channel" to channel, + "exec" to (exec ?: "(dry-run)"), + "accept_from" to acceptFrom?.toList(), + "poll_secs" to pollSecs, + "already_handled" to handled.size, + ), + ) + System.err.println("[agent] serving as $me on ${relay.url} — Ctrl-C to stop") + } + + val donePass = mutableListOf>() + while (true) { + val pending = + BuzzJobCommands + .fetchJobs(ctx, relay, channel, timeoutSecs) + .filter { it.state == JobState.REQUESTED && it.jobId !in handled } + .filter { targetedAtMe(it, me, claimUntargeted) } + .filter { acceptFrom == null || it.requester in acceptFrom } + .sortedBy { it.createdAt } + + for (job in pending) { + handled.add(job.jobId) // mark before working so a slow --exec isn't double-run + val outcome = handle(ctx, relay, me, job, exec, dryRun, postProgress, execTimeoutSecs) + if (once) donePass.add(outcome) else System.err.println("[agent] ${outcome["state"]} job ${job.jobId.take(12)}…") + } + + if (once) { + Output.emit(mapOf("relay" to relay.url, "handled" to donePass.size, "jobs" to donePass)) + return 0 + } + delay(pollSecs * 1000) + } + } + @Suppress("UNREACHABLE_CODE") + return 0 + } + + /** A REQUESTED job is mine to handle if it `p`-targets me, or has no target and I opted in. */ + private fun targetedAtMe( + job: JobView, + me: HexKey, + claimUntargeted: Boolean, + ): Boolean = + when (job.agent) { + me -> true + null -> claimUntargeted + else -> false + } + + /** Accept → (progress) → run --exec → result/error. Returns a summary row. */ + private suspend fun handle( + ctx: Context, + relay: NormalizedRelayUrl, + me: HexKey, + job: JobView, + exec: String?, + dryRun: Boolean, + postProgress: Boolean, + execTimeoutSecs: Long, + ): Map { + val channel = job.channel + publish(ctx, relay, JobAcceptedEvent.build(job.jobId, channel, job.requester, "picked up by amy")) + + if (dryRun) { + publish(ctx, relay, JobResultEvent.build(job.jobId, "[dry-run] would run: ${exec ?: "(none)"}", channel, job.requester, "completed")) + return mapOf("job_id" to job.jobId, "state" to "completed", "dry_run" to true) + } + if (postProgress) { + publish(ctx, relay, JobProgressEvent.build(job.jobId, "working…", channel, "running")) + } + + val env = + buildMap { + put("BUZZ_JOB_ID", job.jobId) + job.requester?.let { put("BUZZ_REQUESTER", it) } + channel?.let { put("BUZZ_CHANNEL", it) } + put("BUZZ_RELAY", relay.url) + put("BUZZ_AGENT", me) + } + val run = runExec(exec!!, job.request ?: "", env, execTimeoutSecs) + + return if (run.exit == 0) { + val body = run.stdout.ifBlank { "(no output)" } + publish(ctx, relay, JobResultEvent.build(job.jobId, body.take(MAX_BODY), channel, job.requester, "completed")) + mapOf("job_id" to job.jobId, "state" to "completed", "exit" to 0) + } else { + val body = (run.stderr.ifBlank { run.stdout }).ifBlank { "exited ${run.exit}" } + publish(ctx, relay, JobErrorEvent.build(job.jobId, body.take(MAX_BODY), channel, "error")) + mapOf("job_id" to job.jobId, "state" to "failed", "exit" to run.exit) + } + } + + private class ExecResult( + val exit: Int, + val stdout: String, + val stderr: String, + ) + + /** Run `sh -c CMD`, piping [input] to its stdin and exporting [env]; capture both streams. */ + private suspend fun runExec( + cmd: String, + input: String, + env: Map, + timeoutSecs: Long, + ): ExecResult = + withContext(Dispatchers.IO) { + val pb = ProcessBuilder("sh", "-c", cmd) + pb.environment().putAll(env) + val proc = pb.start() + proc.outputStream.use { it.write(input.encodeToByteArray()) } + // Read both pipes before waiting so a chatty child can't deadlock on a full buffer. + val out = proc.inputStream.readBytes().decodeToString() + val err = proc.errorStream.readBytes().decodeToString() + if (timeoutSecs > 0) { + if (!proc.waitFor(timeoutSecs, TimeUnit.SECONDS)) { + proc.destroyForcibly() + return@withContext ExecResult(124, out, "exec timed out after ${timeoutSecs}s") + } + } else { + proc.waitFor() + } + ExecResult(proc.exitValue(), out, err) + } + + private suspend fun publish( + ctx: Context, + relay: NormalizedRelayUrl, + template: EventTemplate, + ) { + val signed = ctx.signer.sign(template) + ctx.publish(signed, setOf(relay)) + } + + private const val MAX_BODY = 60_000 +} 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 60562864dc..656e944946 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 @@ -80,6 +80,8 @@ object BuzzCommands { |amy buzz dm open RELAY PUBKEY [PUBKEY…] open a DM with 1-8 people (kind-41010) |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 agent serve RELAY --exec CMD run an agent job-responder loop """.trimMargin() suspend fun dispatch( @@ -98,6 +100,8 @@ object BuzzCommands { "console" to { rest -> console(dataDir, rest) }, "personas" to { rest -> personas(dataDir, rest) }, "dm" to { rest -> dm(dataDir, rest) }, + "job" to { rest -> BuzzJobCommands.dispatch(dataDir, rest) }, + "agent" to { rest -> BuzzAgentCommands.dispatch(dataDir, rest) }, ), ) diff --git a/cli/src/main/kotlin/com/vitorpamplona/amethyst/cli/commands/BuzzJobCommands.kt b/cli/src/main/kotlin/com/vitorpamplona/amethyst/cli/commands/BuzzJobCommands.kt new file mode 100644 index 0000000000..edd6f8e399 --- /dev/null +++ b/cli/src/main/kotlin/com/vitorpamplona/amethyst/cli/commands/BuzzJobCommands.kt @@ -0,0 +1,253 @@ +/* + * 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.BuzzJobAggregator +import com.vitorpamplona.amethyst.commons.model.buzz.JobState +import com.vitorpamplona.amethyst.commons.model.buzz.JobView +import com.vitorpamplona.quartz.buzz.jobs.JobCancelEvent +import com.vitorpamplona.quartz.buzz.jobs.JobRequestEvent +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.nip19Bech32.decodePublicKeyAsHexOrNull + +/** + * `amy buzz job …` — the requester side of the Buzz agent-job protocol (kinds + * 43001-43006): ask an agent to do a task, list jobs, inspect one job's lifecycle, and + * cancel. The agent side (accept/progress/result/error, plus a driving loop) lives in + * [BuzzAgentCommands]. Job correlation + state folding is the shared, tested + * [BuzzJobAggregator] in `commons`, so this file stays a thin assembly layer. + * + * SCHEMA CAVEAT: 43001-43006 are *reserved* in Buzz with no upstream builder; the tag + * layout is Quartz's best-effort model (see [JobRequestEvent]). + */ +object BuzzJobCommands { + private val USAGE = + """ + |amy buzz job request RELAY file a job (kind-43001) + | [--agent PUBKEY] [--channel GID] target agent (p) / channel scope (h) + |amy buzz job list RELAY [--channel GID] list jobs and their state + | [--mine|--assigned] [--limit N] [--timeout SECS] + |amy buzz job show RELAY JOBID [--timeout SECS] show one job's full lifecycle + |amy buzz job cancel RELAY JOBID [--reason R] cancel a job (kind-43005) + | [--channel GID] + """.trimMargin() + + suspend fun dispatch( + dataDir: DataDir, + tail: Array, + ): Int = + route( + "buzz job", + tail, + USAGE, + mapOf( + "request" to { rest -> request(dataDir, rest) }, + "list" to { rest -> list(dataDir, rest) }, + "show" to { rest -> show(dataDir, rest) }, + "cancel" to { rest -> cancel(dataDir, rest) }, + ), + ) + + /** `buzz job request RELAY [--agent PUBKEY] [--channel GID]` → publishes a kind-43001. */ + private suspend fun request( + dataDir: DataDir, + rest: Array, + ): Int { + val args = Args(rest) + val usage = "buzz job request RELAY [--agent PUBKEY] [--channel GID]" + val relayUrl = args.positionalOrNull(0) ?: return Output.error("bad_args", usage) + val text = args.positionalOrNull(1) ?: return Output.error("bad_args", usage) + if (text.isBlank()) return Output.error("bad_args", "job text must not be blank") + val relay = normalizeGroupRelay(relayUrl) ?: return Output.error("bad_args", "invalid relay url: $relayUrl") + val channel = args.flag("channel") + val agent = + args.flag("agent")?.let { + decodePublicKeyAsHexOrNull(it.trim())?.takeIf { hex -> hex.isValid() } + ?: return Output.error("bad_args", "invalid agent public key (npub or 64-char hex): $it") + } + args.rejectUnknown("agent", "channel") + + Context.open(dataDir).use { ctx -> + ctx.prepare() + val signed = ctx.signer.sign(JobRequestEvent.build(text, channel, agent)) + val ack = ctx.publish(signed, setOf(relay)) + RawEventSupport.publishGuard(ack, signed.id)?.let { return it } + Output.emit( + mapOf( + "job_id" to signed.id, + "kind" to signed.kind, + "relay" to relay.url, + "agent" to agent, + "channel" to channel, + "published" to ack.values.any { it.accepted }, + ), + ) + return 0 + } + } + + /** + * `buzz job list RELAY [--channel GID] [--mine|--assigned] [--limit N] [--timeout SECS]` → + * drains the job kinds and folds them into per-job state. `--mine` keeps jobs I + * requested; `--assigned` keeps jobs targeting me; default shows both. + */ + private suspend fun list( + dataDir: DataDir, + rest: Array, + ): Int { + val args = Args(rest) + val usage = "buzz job list RELAY [--channel GID] [--mine|--assigned] [--limit N] [--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") + val mine = args.bool("mine") + val assigned = args.bool("assigned") + val limit = args.flag("limit")?.toIntOrNull() ?: 50 + val timeoutSecs = args.flag("timeout")?.toLongOrNull() ?: 8 + args.rejectUnknown("channel", "mine", "assigned", "limit", "timeout") + + Context.open(dataDir).use { ctx -> + ctx.prepare() + val me = ctx.identity.pubKeyHex + val jobs = + fetchJobs(ctx, relay, channel, timeoutSecs) + .filter { job -> + when { + mine && !assigned -> job.requester == me + assigned && !mine -> job.agent == me + else -> true + } + }.take(limit) + Output.emit(mapOf("relay" to relay.url, "count" to jobs.size, "jobs" to jobs.map { it.toRow() })) + return 0 + } + } + + /** `buzz job show RELAY JOBID [--timeout SECS]` → the full folded lifecycle of one job. */ + private suspend fun show( + dataDir: DataDir, + rest: Array, + ): Int { + val args = Args(rest) + val usage = "buzz job show RELAY JOBID [--timeout SECS]" + val relayUrl = args.positionalOrNull(0) ?: return Output.error("bad_args", usage) + val jobId = 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() + // Fetch the request by id plus every reply that references it via `e`. + val filters = + listOf( + Filter(kinds = JOB_KINDS, ids = listOf(jobId)), + Filter(kinds = JOB_REPLY_KINDS, tags = mapOf("e" to listOf(jobId))), + ) + val events = + ctx + .drain(mapOf(relay to filters), timeoutSecs * 1000, pendingOnAuthRequired = true) + .map { it.second } + val job = + BuzzJobAggregator.aggregate(events).firstOrNull { it.jobId == jobId } + ?: return Output.error("not_found", "no job $jobId on ${relay.url}") + Output.emit(job.toRow() + mapOf("relay" to relay.url)) + return 0 + } + } + + /** `buzz job cancel RELAY JOBID [--reason R] [--channel GID]` → publishes a kind-43005. */ + private suspend fun cancel( + dataDir: DataDir, + rest: Array, + ): Int { + val args = Args(rest) + val usage = "buzz job cancel RELAY JOBID [--reason R] [--channel GID]" + val relayUrl = args.positionalOrNull(0) ?: return Output.error("bad_args", usage) + val jobId = args.positionalOrNull(1) ?: return Output.error("bad_args", usage) + val relay = normalizeGroupRelay(relayUrl) ?: return Output.error("bad_args", "invalid relay url: $relayUrl") + val reason = args.flag("reason") ?: "" + val channel = args.flag("channel") + args.rejectUnknown("reason", "channel") + + Context.open(dataDir).use { ctx -> + ctx.prepare() + val signed = ctx.signer.sign(JobCancelEvent.build(jobId, reason, channel)) + 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, + "job_id" to jobId, + "relay" to relay.url, + "published" to ack.values.any { it.accepted }, + ), + ) + return 0 + } + } + + /** Drain every job kind (optionally channel-scoped) and fold via the shared aggregator. */ + internal suspend fun fetchJobs( + ctx: Context, + relay: NormalizedRelayUrl, + channel: String?, + timeoutSecs: Long, + ): List { + val tags = channel?.let { mapOf("h" to listOf(it)) } + val filter = Filter(kinds = JOB_KINDS, tags = tags) + val events = + ctx + .drain(mapOf(relay to listOf(filter)), timeoutSecs * 1000, pendingOnAuthRequired = true) + .map { it.second } + return BuzzJobAggregator.aggregate(events) + } + + private fun JobView.toRow(): Map = + mapOf( + "job_id" to jobId, + "state" to state.name.lowercase(), + "requester" to requester, + "agent" to agent, + "channel" to channel, + "request" to request, + "progress_updates" to progressUpdates, + "last_progress" to lastProgress, + "result" to result, + "error" to error, + "cancel_reason" to cancelReason, + "created_at" to createdAt, + "updated_at" to updatedAt, + ) + + internal val JOB_KINDS = (43001..43006).toList() + internal val JOB_REPLY_KINDS = (43002..43006).toList() + + /** The terminal states — jobs a responder should never re-handle. */ + internal val TERMINAL = setOf(JobState.COMPLETED, JobState.FAILED, JobState.CANCELLED) +} diff --git a/cli/tests/.gitignore b/cli/tests/.gitignore index 61f2fb29a9..db90bb8311 100644 --- a/cli/tests/.gitignore +++ b/cli/tests/.gitignore @@ -7,3 +7,4 @@ relaygroup/state-relaygroup-headless/ sync/state-sync-deletions/ blossom/state-blossom-live/ git/state-git-nip34/ +buzz/state-job-loop/ diff --git a/cli/tests/buzz/job-loop.sh b/cli/tests/buzz/job-loop.sh new file mode 100755 index 0000000000..617a2d445e --- /dev/null +++ b/cli/tests/buzz/job-loop.sh @@ -0,0 +1,101 @@ +#!/usr/bin/env bash +# +# job-loop.sh — self-contained headless test for the Buzz agent-job loop. +# +# Two `amy` accounts (alice = requester, bot = agent) talk through an embedded +# relay (`amy serve`, i.e. geode — no external binary). Exercises the whole +# drive-an-agent loop end to end: +# +# alice: buzz job request (kind-43001, targeting the bot) +# bot: buzz agent serve --once (accept 43002 → progress 43003 → +# run --exec → result 43004) +# alice: buzz job show (BuzzJobAggregator folds it to state=completed) +# +# It also proves the permission gate: a responder whose --accept-from allowlist +# excludes alice handles NOTHING; only an allowlisted requester is obeyed. +# +# Usage: ./job-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-job-loop" +AMY_BIN="$REPO_ROOT/cli/build/install/amy/bin/amy" + +PORT=7799 +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" + +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"; AGENT_HOME="$STATE_DIR/bot"; mkdir -p "$REQ_HOME" "$AGENT_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_agent() { HOME="$AGENT_HOME" "$AMY_BIN" --account bot --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