diff --git a/cli/README.md b/cli/README.md index e0723b78c6..63263e96f5 100644 --- a/cli/README.md +++ b/cli/README.md @@ -614,7 +614,7 @@ and `commons` aggregator the app uses. | `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. | +| `amy buzz agent serve RELAY --exec CMD [--channel GID] [--accept-from npub,…] [--accept-from-channel] [--parallel N] [--worktree REPODIR] [--base-ref REF] [--branch-prefix P] [--claim-untargeted] [--poll SECS] [--exec-timeout SECS] [--no-progress] [--dry-run] [--once]` | Run a backlog **scheduler**. Watches a channel's REQUESTED jobs, orders them by the group's upvotes (kind-7 likes), and runs up to `--parallel N` at once — each in its own `git worktree`+branch (`--worktree REPODIR`, off `--base-ref`, named ``) so concurrent runs never collide. Per job: accept (43002) → progress (43003) → runs `sh -c CMD` inside the worktree (task text on stdin; `BUZZ_JOB_ID/REQUESTER/CHANNEL/RELAY/AGENT/UPVOTES/BRANCH/WORKTREE/BASE_REF` in env) → result (43004) or error (43006). Intake gate: `--accept-from` (explicit npubs) and/or `--accept-from-channel` (the channel's kind-39002 member roster). `--parallel > 1` requires `--worktree`. The exec commits/pushes its branch and opens the PR; **merge stays on GitHub, never here.** | > **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 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 4335f7b747..ae7a760705 100644 --- a/cli/plans/2026-07-25-buzz-agent-support-channel.md +++ b/cli/plans/2026-07-25-buzz-agent-support-channel.md @@ -6,10 +6,36 @@ ## 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`. +Give the Amethyst team a **shared feature-request channel** where anyone can drive work: the +team debates and files requests, an AI coding agent — Claude Code running as *this* Anthropic +account — **manages the backlog by itself and works items in parallel**, over a self-hosted +[`block/buzz`](https://github.com/block/buzz) workspace. Every request, upvote, and result is +a signed, audited Nostr event the whole room sees. This is **not** a 1:1 chat with the bot. + +Interaction model (decided): +- **Anyone in the channel can drive** a work stream — no propose-and-confirm gate; a member's + job request is auto-accepted and scheduled (**full auto from intake**). +- The bot **owns a stack**: it orders the backlog by the group's upvotes and runs up to N in + parallel, each isolated in its own git worktree/branch. +- **The only human gate is the merge, and it happens on GitHub** (branch protection + review) — + never inside Amy or the channel. The agent opens PRs; it can never merge or damage `main`. + +### Can this live in Amy? Yes — Amy is the scheduler, the coding agent is `--exec`. + +A clean three-way split, no separate project needed for the team-on-a-box case: +- **Amy** owns the Buzz side: watch the backlog, order by upvotes, dispatch up to `--parallel N`, + isolate each job in a worktree/branch, report status as job events. Reuses everything already + built (relay client, job models, `BuzzJobAggregator`, the responder, subprocess spawning, the + long-running `serve` pattern). Decision logic lives in `commons` (pure/testable); git + + process I/O lives in the `cli` command — so Amy stays a thin assembly layer. +- **`--exec`** is the coding agent (Claude Code via buzz-acp / Goose / a script) Amy spawns per + job. Not a new project — an existing tool. It runs inside the job's worktree (`BUZZ_BRANCH`, + `BUZZ_WORKTREE` exported), commits, pushes the branch, opens the PR; its stdout is the result. +- **GitHub** owns review + merge, entirely outside the loop. + +Graduate to a separate service only if you outgrow one host (hosted, multi-tenant, a web +dashboard, a cross-machine worker fleet) — and even then Amy/`quartz`/`commons` stay the library +underneath. ## Why Buzz is the right substrate (and what it is NOT) @@ -78,16 +104,25 @@ Thin assembly over quartz job models + a shared aggregator; no protocol logic in - **`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. +- **`amy buzz agent serve RELAY --exec CMD`** (`BuzzAgentCommands.kt`) — the **backlog + scheduler**. Watches a channel's REQUESTED jobs, orders them by `BuzzJobAggregator.byPriority` + (upvotes desc, oldest-first tiebreak), and runs up to `--parallel N` at once — each in its own + `git worktree` + branch (`--worktree REPODIR`, off `--base-ref`, named ``) + so concurrent runs never collide (`--parallel > 1` requires `--worktree`; worktree add/remove + is mutex-serialized, the agent work runs concurrently). Per job: 43002 accept → 43003 progress + → `sh -c CMD` inside the worktree (task text on stdin; `BUZZ_JOB_ID/REQUESTER/CHANNEL/RELAY/ + AGENT/UPVOTES/BRANCH/WORKTREE/BASE_REF` in env) → 43004 result or 43006 error. Intake gate: + `--accept-from` (explicit npubs) and/or `--accept-from-channel` (the channel's kind-39002 + member roster — "anyone in the channel drives"). `--dry-run`, `--once`, `--claim-untargeted`, + `--exec-timeout` for testing/ops. This is where Claude Code plugs in: `--exec` runs the agent, + which opens the PR and echoes the URL as the result. +- **Upvote priority** (`BuzzJobs.kt`): `BuzzJobAggregator` folds kind-7 likes (distinct reactors, + dislikes excluded) targeting a job into `JobView.upvotes`; `byPriority` orders the backlog. The + group reprioritizes the stack just by reacting. -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. +Guardrails restated in the command's KDoc: `--accept-from` / `--accept-from-channel` is the +Buzz-layer intake gate; repo blast radius is the `--exec` credential (PR-only) + branch +protection, not Buzz. Merge is never done here — only on GitHub. ### Schema caveat 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 3da5a0a6d4..f7d936f3aa 100644 --- a/cli/src/main/kotlin/com/vitorpamplona/amethyst/cli/Main.kt +++ b/cli/src/main/kotlin/com/vitorpamplona/amethyst/cli/Main.kt @@ -824,7 +824,7 @@ 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 agent serve RELAY --exec CMD run an agent job-responder loop + | buzz agent serve RELAY --exec CMD run a parallel backlog scheduler (worktree-isolated) | |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 index 95e820e236..8e6965c089 100644 --- a/cli/src/main/kotlin/com/vitorpamplona/amethyst/cli/commands/BuzzAgentCommands.kt +++ b/cli/src/main/kotlin/com/vitorpamplona/amethyst/cli/commands/BuzzAgentCommands.kt @@ -24,6 +24,7 @@ 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.JobAcceptedEvent @@ -33,32 +34,46 @@ 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.filters.Filter import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl import com.vitorpamplona.quartz.nip01Core.signers.EventTemplate import com.vitorpamplona.quartz.nip19Bech32.decodePublicKeyAsHexOrNull +import com.vitorpamplona.quartz.nip29RelayGroups.metadata.GroupMembersEvent import kotlinx.coroutines.Dispatchers +import kotlinx.coroutines.coroutineScope import kotlinx.coroutines.delay +import kotlinx.coroutines.launch +import kotlinx.coroutines.supervisorScope +import kotlinx.coroutines.sync.Mutex +import kotlinx.coroutines.sync.Semaphore +import kotlinx.coroutines.sync.withLock +import kotlinx.coroutines.sync.withPermit import kotlinx.coroutines.withContext +import java.io.File 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). + * `amy buzz agent …` — the AGENT side of the Buzz job protocol: a headless SCHEDULER that + * manages a shared backlog by itself. It watches a channel's job requests (kind-43001), + * orders them by the group's upvotes, and runs up to `--parallel N` at a time — each in its + * own git worktree/branch so concurrent runs never clobber each other — reporting every step + * back as accept/progress/result/error events (43002/43003/43004/43006) the whole room sees. * - * 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. + * This is the "shared channel where the team drives an AI to build Amethyst" model: + * **anyone in the channel files a job, the bot works them autonomously in parallel, and the + * only human gate left is the merge — which happens on GitHub, never here.** Point `--exec` + * at a coding agent (Claude Code via buzz-acp, a Goose/Codex wrapper, or a script): the job's + * task text is piped to its stdin, and it runs inside a fresh worktree whose branch is + * exported as `BUZZ_BRANCH`. The agent commits + pushes that branch and opens the PR; its + * stdout (e.g. the PR URL) 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`. + * PERMISSIONS — Buzz authorizes by identity, not capability flags, so this is only as safe as: + * 1. INTAKE — `--accept-from` / `--accept-from-channel` gate WHO the bot obeys (the channel + * roster). Without either, it answers anyone who can post to the relay. + * 2. BLAST RADIUS — what `--exec` can DO to a repo is bounded by the credentials you give it, + * NOT by Buzz. Each job gets its own branch off `--base-ref`; the exec's git token should + * only open PRs on feature branches (never merge, never force-push), and `main` must be + * branch-protected. 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]. @@ -66,18 +81,23 @@ import java.util.concurrent.TimeUnit object BuzzAgentCommands { private val USAGE = """ - |amy buzz agent serve RELAY --exec CMD run a job-responder loop + |amy buzz agent serve RELAY --exec CMD run a backlog scheduler | [--channel GID] only handle jobs scoped to this channel - | [--accept-from npub,npub] allowlist of requester keys (STRONGLY advised) + | [--accept-from npub,npub] allowlist of requester keys + | [--accept-from-channel] obey any member of --channel (kind-39002 roster) | [--claim-untargeted] also handle jobs with no `p` target + | [--parallel N] run up to N jobs at once (default 1) + | [--worktree REPODIR] base git repo; each job gets its own worktree+branch + | [--base-ref REF] branch base for worktrees (default HEAD) + | [--branch-prefix P] job branch prefix (default claude/job-) | [--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 + | [--no-progress] [--dry-run] [--once] """.trimMargin() + private val worktreeMutex = Mutex() // git worktree add/remove touch shared repo metadata + suspend fun dispatch( dataDir: DataDir, tail: Array, @@ -91,6 +111,22 @@ object BuzzAgentCommands { ), ) + private class Opts( + val relay: NormalizedRelayUrl, + val exec: String?, + val channel: String?, + val claimUntargeted: Boolean, + val postProgress: Boolean, + val dryRun: Boolean, + val parallel: Int, + val pollSecs: Long, + val timeoutSecs: Long, + val execTimeoutSecs: Long, + val worktreeBase: String?, + val baseRef: String, + val branchPrefix: String, + ) + private suspend fun serve( dataDir: DataDir, rest: Array, @@ -105,9 +141,14 @@ object BuzzAgentCommands { val claimUntargeted = args.bool("claim-untargeted") val postProgress = !args.bool("no-progress") val once = args.bool("once") + val parallel = args.flag("parallel")?.toIntOrNull()?.coerceAtLeast(1) ?: 1 val pollSecs = args.flag("poll")?.toLongOrNull() ?: 5 val timeoutSecs = args.flag("timeout")?.toLongOrNull() ?: 8 val execTimeoutSecs = args.flag("exec-timeout")?.toLongOrNull() ?: 1800 + val worktreeBase = args.flag("worktree") + val baseRef = args.flag("base-ref") ?: "HEAD" + val branchPrefix = args.flag("branch-prefix") ?: "claude/job-" + val fromChannel = args.bool("accept-from-channel") val acceptFrom = args .flag("accept-from") @@ -116,12 +157,17 @@ object BuzzAgentCommands { ?.map { decodePublicKeyAsHexOrNull(it)?.takeIf { hex -> hex.isValid() } ?: return Output.error("bad_args", "invalid --accept-from key (npub or 64-char hex): $it") - }?.toSet() + }?.toMutableSet() args.rejectUnknown( "exec", "channel", "accept-from", + "accept-from-channel", "claim-untargeted", + "parallel", + "worktree", + "base-ref", + "branch-prefix", "poll", "exec-timeout", "timeout", @@ -130,60 +176,157 @@ object BuzzAgentCommands { "once", ) + // Parallel runs share one working tree unless each gets its own worktree — that's a + // guaranteed clobber. Require --worktree once concurrency is on. + if (parallel > 1 && worktreeBase == null) { + return Output.error("bad_args", "--parallel > 1 needs --worktree REPODIR so concurrent jobs don't clobber one working tree") + } + if (fromChannel && channel == null) { + return Output.error("bad_args", "--accept-from-channel needs --channel GID") + } + if (worktreeBase != null && !File(worktreeBase).isDirectory) { + return Output.error("bad_args", "--worktree is not a directory: $worktreeBase") + } + + val opts = + Opts( + relay, + exec, + channel, + claimUntargeted, + postProgress, + dryRun, + parallel, + pollSecs, + timeoutSecs, + execTimeoutSecs, + worktreeBase, + baseRef, + branchPrefix, + ) + 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. + if (worktreeBase != null && git(worktreeBase, "rev-parse", "--git-dir").exit != 0) { + return Output.error("bad_args", "--worktree is not a git repo: $worktreeBase") + } + + // Resolve the intake allowlist: explicit --accept-from ∪ the channel's kind-39002 + // member roster (when --accept-from-channel). Null = obey anyone (no gate). + val allow: Set? = + if (fromChannel) { + val members = channelMembers(ctx, relay, channel!!, timeoutSecs) + (acceptFrom ?: mutableSetOf()).apply { addAll(members) } + } else { + acceptFrom + } + + // Seed the handled set so a restart doesn't re-run work already picked up. 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") - } + if (once) return runOnce(ctx, me, opts, allow, handled) - 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) - } + Output.emit( + mapOf( + "serving" to me, + "relay" to relay.url, + "channel" to channel, + "exec" to (exec ?: "(dry-run)"), + "parallel" to parallel, + "worktree" to worktreeBase, + "accept_from" to allow?.toList(), + "already_handled" to handled.size, + ), + ) + System.err.println("[agent] serving as $me on ${relay.url} — parallel=$parallel — Ctrl-C to stop") + runForever(ctx, me, opts, allow, handled) } @Suppress("UNREACHABLE_CODE") return 0 } + /** One pass: launch every pending job (throttled to --parallel), wait for all, emit a summary. */ + private suspend fun runOnce( + ctx: Context, + me: HexKey, + opts: Opts, + allow: Set?, + handled: MutableSet, + ): Int { + val pending = selectPending(ctx, me, opts, allow, handled) + val done = mutableListOf>() + val doneMutex = Mutex() + coroutineScope { + val sem = Semaphore(opts.parallel) + pending.forEach { job -> + handled.add(job.jobId) + launch { + sem.withPermit { + val r = handle(ctx, me, opts, job) + doneMutex.withLock { done.add(r) } + } + } + } + } + Output.emit(mapOf("relay" to opts.relay.url, "handled" to done.size, "jobs" to done)) + return 0 + } + + /** The long-running loop: keep the in-flight count at ≤ --parallel, launching by priority. */ + private suspend fun runForever( + ctx: Context, + me: HexKey, + opts: Opts, + allow: Set?, + handled: MutableSet, + ) { + supervisorScope { + val inflight = mutableSetOf() + val mutex = Mutex() + while (true) { + val busy = mutex.withLock { inflight.toSet() } + val free = opts.parallel - busy.size + if (free > 0) { + val pending = selectPending(ctx, me, opts, allow, handled + busy).take(free) + pending.forEach { job -> + handled.add(job.jobId) + mutex.withLock { inflight.add(job.jobId) } + launch { + try { + val r = handle(ctx, me, opts, job) + System.err.println("[agent] ${r["state"]} job ${job.jobId.take(12)}… (${job.upvotes} upvotes)") + } finally { + mutex.withLock { inflight.remove(job.jobId) } + } + } + } + } + delay(opts.pollSecs * 1000) + } + } + } + + /** REQUESTED jobs targeting me, from an allowed requester, not already taken — priority-ordered. */ + private suspend fun selectPending( + ctx: Context, + me: HexKey, + opts: Opts, + allow: Set?, + exclude: Set, + ): List = + BuzzJobAggregator.byPriority( + BuzzJobCommands + .fetchJobs(ctx, opts.relay, opts.channel, opts.timeoutSecs) + .filter { it.state == JobState.REQUESTED && it.jobId !in exclude } + .filter { targetedAtMe(it, me, opts.claimUntargeted) } + .filter { allow == null || it.requester in allow }, + ) + /** 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, @@ -196,47 +339,97 @@ object BuzzAgentCommands { else -> false } - /** Accept → (progress) → run --exec → result/error. Returns a summary row. */ + /** Accept → (worktree) → (progress) → run --exec → result/error. Returns a summary row. */ private suspend fun handle( ctx: Context, - relay: NormalizedRelayUrl, me: HexKey, + opts: Opts, 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")) + publish(ctx, opts.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")) + if (opts.dryRun) { + publish(ctx, opts.relay, JobResultEvent.build(job.jobId, "[dry-run] would run: ${opts.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) + // Each job gets its own worktree+branch off base-ref so N run without collision. + val short = job.jobId.take(12) + val branch = opts.branchPrefix + short + var workdir: String? = null + var worktreePath: String? = null + if (opts.worktreeBase != null) { + val wt = File(System.getProperty("java.io.tmpdir"), "buzz-worktrees/$short") + worktreePath = wt.absolutePath + val add = + worktreeMutex.withLock { + wt.parentFile?.mkdirs() + // Clear anything a crashed prior run left behind for this exact job id. + git(opts.worktreeBase, "worktree", "prune") + wt.deleteRecursively() + git(opts.worktreeBase, "worktree", "add", "-b", branch, worktreePath, opts.baseRef) + } + if (add.exit != 0) { + publish(ctx, opts.relay, JobErrorEvent.build(job.jobId, "worktree setup failed: ${add.stderr.take(MAX_BODY)}", channel, "error")) + return mapOf("job_id" to job.jobId, "state" to "failed", "error" to "worktree") } - 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) + workdir = worktreePath } + + try { + if (opts.postProgress) { + publish(ctx, opts.relay, JobProgressEvent.build(job.jobId, "working on $branch…", 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", opts.relay.url) + put("BUZZ_AGENT", me) + put("BUZZ_UPVOTES", job.upvotes.toString()) + if (opts.worktreeBase != null) { + put("BUZZ_BRANCH", branch) + put("BUZZ_WORKTREE", worktreePath!!) + put("BUZZ_BASE_REF", opts.baseRef) + } + } + val run = runExec(opts.exec!!, job.request ?: "", env, opts.execTimeoutSecs, workdir) + + return if (run.exit == 0) { + val body = run.stdout.ifBlank { "(no output)" } + publish(ctx, opts.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, "branch" to if (opts.worktreeBase != null) branch else null) + } else { + val body = (run.stderr.ifBlank { run.stdout }).ifBlank { "exited ${run.exit}" } + publish(ctx, opts.relay, JobErrorEvent.build(job.jobId, body.take(MAX_BODY), channel, "error")) + mapOf("job_id" to job.jobId, "state" to "failed", "exit" to run.exit) + } + } finally { + // Drop the worktree; the branch stays in the base repo (the exec pushed it). + if (worktreePath != null) { + worktreeMutex.withLock { git(opts.worktreeBase!!, "worktree", "remove", "--force", worktreePath) } + } + } + } + + /** Latest kind-39002 roster for the channel → its member pubkeys. Empty if none served. */ + private suspend fun channelMembers( + ctx: Context, + relay: NormalizedRelayUrl, + channel: String, + timeoutSecs: Long, + ): Set { + val filter = Filter(kinds = listOf(GroupMembersEvent.KIND), tags = mapOf("d" to listOf(channel))) + return ctx + .drain(mapOf(relay to listOf(filter)), timeoutSecs * 1000, pendingOnAuthRequired = true) + .map { it.second } + .filterIsInstance() + .maxByOrNull { it.createdAt } + ?.members() + ?.toSet() + .orEmpty() } private class ExecResult( @@ -245,15 +438,17 @@ object BuzzAgentCommands { val stderr: String, ) - /** Run `sh -c CMD`, piping [input] to its stdin and exporting [env]; capture both streams. */ + /** Run `sh -c CMD` in [workdir], piping [input] to stdin and exporting [env]; capture both streams. */ private suspend fun runExec( cmd: String, input: String, env: Map, timeoutSecs: Long, + 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() proc.outputStream.use { it.write(input.encodeToByteArray()) } @@ -271,6 +466,19 @@ object BuzzAgentCommands { ExecResult(proc.exitValue(), out, err) } + /** Run `git -C dir args…`, capturing exit + both streams. */ + private suspend fun git( + dir: String, + vararg gitArgs: String, + ): ExecResult = + withContext(Dispatchers.IO) { + val proc = ProcessBuilder(listOf("git", "-C", dir) + gitArgs).start() + val out = proc.inputStream.readBytes().decodeToString() + val err = proc.errorStream.readBytes().decodeToString() + proc.waitFor() + ExecResult(proc.exitValue(), out, err) + } + private suspend fun publish( ctx: Context, relay: NormalizedRelayUrl, 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 656e944946..19aa4502f1 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,7 +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 agent serve RELAY --exec CMD run an agent job-responder loop + |amy buzz agent serve RELAY --exec CMD run a parallel backlog scheduler (worktree-isolated) """.trimMargin() suspend fun dispatch( 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 index edd6f8e399..e2b6cd6c00 100644 --- a/cli/src/main/kotlin/com/vitorpamplona/amethyst/cli/commands/BuzzJobCommands.kt +++ b/cli/src/main/kotlin/com/vitorpamplona/amethyst/cli/commands/BuzzJobCommands.kt @@ -33,6 +33,7 @@ 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 +import com.vitorpamplona.quartz.nip25Reactions.ReactionEvent /** * `amy buzz job …` — the requester side of the Buzz agent-job protocol (kinds @@ -162,11 +163,12 @@ object BuzzJobCommands { Context.open(dataDir).use { ctx -> ctx.prepare() - // Fetch the request by id plus every reply that references it via `e`. + // Fetch the request by id, every reply that references it via `e`, and every + // upvote (kind-7 reaction) targeting it. val filters = listOf( Filter(kinds = JOB_KINDS, ids = listOf(jobId)), - Filter(kinds = JOB_REPLY_KINDS, tags = mapOf("e" to listOf(jobId))), + Filter(kinds = JOB_REPLY_KINDS + ReactionEvent.KIND, tags = mapOf("e" to listOf(jobId))), ) val events = ctx @@ -212,7 +214,11 @@ object BuzzJobCommands { } } - /** Drain every job kind (optionally channel-scoped) and fold via the shared aggregator. */ + /** + * Drain every job kind (optionally channel-scoped) plus, when a channel is given, its + * kind-7 upvotes, and fold via the shared aggregator. Upvotes are only fetched with a + * channel scope — a bare kind-7 query would pull the relay's entire reaction firehose. + */ internal suspend fun fetchJobs( ctx: Context, relay: NormalizedRelayUrl, @@ -220,10 +226,14 @@ object BuzzJobCommands { timeoutSecs: Long, ): List { val tags = channel?.let { mapOf("h" to listOf(it)) } - val filter = Filter(kinds = JOB_KINDS, tags = tags) + val filters = + buildList { + add(Filter(kinds = JOB_KINDS, tags = tags)) + if (channel != null) add(Filter(kinds = listOf(ReactionEvent.KIND), tags = tags)) + } val events = ctx - .drain(mapOf(relay to listOf(filter)), timeoutSecs * 1000, pendingOnAuthRequired = true) + .drain(mapOf(relay to filters), timeoutSecs * 1000, pendingOnAuthRequired = true) .map { it.second } return BuzzJobAggregator.aggregate(events) } @@ -236,6 +246,7 @@ object BuzzJobCommands { "agent" to agent, "channel" to channel, "request" to request, + "upvotes" to upvotes, "progress_updates" to progressUpdates, "last_progress" to lastProgress, "result" to result, diff --git a/cli/tests/buzz/job-loop.sh b/cli/tests/buzz/job-loop.sh index 617a2d445e..972684455f 100755 --- a/cli/tests/buzz/job-loop.sh +++ b/cli/tests/buzz/job-loop.sh @@ -96,6 +96,20 @@ check "requester" _ "$(echo "$SHOW" | jkey requester)" "$ALICE" check "agent" _ "$(echo "$SHOW" | jkey agent)" "$BOT" contains "result carries exec stdout with piped stdin" "$(echo "$SHOW" | jkey result)" "PR opened for: hello from alice" +echo "> 5. parallel scheduler: 3 jobs, --parallel 3, one worktree+branch each" >&2 +REPO="$STATE_DIR/repo"; mkdir -p "$REPO" +git -C "$REPO" -c init.defaultBranch=main init -q +git -C "$REPO" -c user.email=a@b.c -c user.name=t commit -q --allow-empty -m init +for n in one two three; do run_req buzz job request "$RELAY" "parallel $n" --agent "$BOT" >/dev/null; done +PAR=$(run_agent buzz agent serve "$RELAY" \ + --exec 'printf "on %s: " "$BUZZ_BRANCH"; cat' \ + --accept-from "$ALICE" --parallel 3 --worktree "$REPO" --base-ref main --once) +check "parallel scheduler handled 3" _ "$(echo "$PAR" | jkey handled)" "3" +BRANCHES=$(git -C "$REPO" branch --list 'claude/job-*' | wc -l | tr -d ' ') +check "created one branch per job" _ "$BRANCHES" "3" +LEFTOVER=$(git -C "$REPO" worktree list | sed 1d | wc -l | tr -d ' ') +check "worktrees cleaned up after runs" _ "$LEFTOVER" "0" + echo "" >&2 echo "> RESULTS: $PASS passed, $FAIL failed" >&2 [[ $FAIL -eq 0 ]] && exit 0 || exit 1 diff --git a/commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/model/buzz/BuzzJobs.kt b/commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/model/buzz/BuzzJobs.kt index 53a2aaf760..fd9e069b71 100644 --- a/commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/model/buzz/BuzzJobs.kt +++ b/commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/model/buzz/BuzzJobs.kt @@ -29,6 +29,7 @@ import com.vitorpamplona.quartz.buzz.jobs.JobRequestEvent 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.nip25Reactions.ReactionEvent /** The lifecycle state of one Buzz agent job, folded from its 43001-43006 events. */ enum class JobState { @@ -79,6 +80,12 @@ data class JobView( val error: String?, /** The 43005 cancel reason, when [state] is CANCELLED. */ val cancelReason: String?, + /** + * Distinct channel members who upvoted this job — a NIP-25 like (kind-7, content not `-`) + * whose `e` tag targets the job's request id. The group's priority signal: a scheduler + * orders the backlog by this, newest-first as a tiebreaker. + */ + val upvotes: Int, /** The request timestamp, or the earliest correlated event when the request is absent. */ val createdAt: Long, /** The timestamp of the most recent event in the thread. */ @@ -101,10 +108,21 @@ object BuzzJobAggregator { fun aggregate(events: List): List { if (events.isEmpty()) return emptyList() - // Correlate every event to a job id: the request is its own id; a reply carries + val distinct = events.distinctBy { it.id } + + // Upvotes: distinct authors of a NIP-25 like (kind-7, content not `-`) per targeted + // event id. Counting distinct pubkeys stops one member inflating priority by spamming. + val upvoters = HashMap>() + distinct.forEach { e -> + if (e is ReactionEvent && e.content != ReactionEvent.DISLIKE) { + e.originalPost().forEach { target -> upvoters.getOrPut(target) { mutableSetOf() }.add(e.pubKey) } + } + } + + // Correlate every job event to a job id: the request is its own id; a reply carries // the request id in its `e` tag. Replies we can't correlate (no `e`) are dropped. val byJob = LinkedHashMap>() - events.distinctBy { it.id }.forEach { e -> + distinct.forEach { e -> val jobId = when (e) { is JobRequestEvent -> e.id @@ -119,13 +137,14 @@ object BuzzJobAggregator { } return byJob - .map { (jobId, thread) -> fold(jobId, thread) } + .map { (jobId, thread) -> fold(jobId, thread, upvoters[jobId]?.size ?: 0) } .sortedByDescending { it.updatedAt } } private fun fold( jobId: HexKey, thread: List, + upvotes: Int, ): JobView { val request = thread.filterIsInstance().maxByOrNull { it.createdAt } val accepted = thread.filterIsInstance().maxByOrNull { it.createdAt } @@ -166,8 +185,16 @@ object BuzzJobAggregator { result = result?.result(), error = error?.error(), cancelReason = cancel?.reason()?.ifBlank { null }, + upvotes = upvotes, createdAt = request?.createdAt ?: thread.minOf { it.createdAt }, updatedAt = thread.maxOf { it.createdAt }, ) } + + /** + * Backlog ordering for a scheduler: most-upvoted first (the group's priority signal), + * oldest-first as the tiebreaker so an un-upvoted item still drains FIFO. Applies to the + * caller-provided set (typically the REQUESTED jobs targeting the agent). + */ + fun byPriority(jobs: List): List = jobs.sortedWith(compareByDescending { it.upvotes }.thenBy { it.createdAt }) } diff --git a/commons/src/commonTest/kotlin/com/vitorpamplona/amethyst/commons/model/buzz/BuzzJobAggregatorTest.kt b/commons/src/commonTest/kotlin/com/vitorpamplona/amethyst/commons/model/buzz/BuzzJobAggregatorTest.kt index 880d654e93..a3e320ac92 100644 --- a/commons/src/commonTest/kotlin/com/vitorpamplona/amethyst/commons/model/buzz/BuzzJobAggregatorTest.kt +++ b/commons/src/commonTest/kotlin/com/vitorpamplona/amethyst/commons/model/buzz/BuzzJobAggregatorTest.kt @@ -27,6 +27,7 @@ import com.vitorpamplona.quartz.buzz.jobs.JobProgressEvent import com.vitorpamplona.quartz.buzz.jobs.JobRequestEvent import com.vitorpamplona.quartz.buzz.jobs.JobResultEvent import com.vitorpamplona.quartz.nip01Core.core.Event +import com.vitorpamplona.quartz.nip25Reactions.ReactionEvent import kotlin.test.Test import kotlin.test.assertEquals import kotlin.test.assertNull @@ -78,6 +79,14 @@ class BuzzJobAggregatorTest { return JobCancelEvent("5".repeat(64), requester, t.createdAt, t.tags, t.content, "sig") } + // A NIP-25 like targeting a job's request id — the group's upvote signal. + private fun like( + reactor: String, + target: String = jobId, + id: String, + content: String = "+", + ): ReactionEvent = ReactionEvent(id, reactor, 1050, arrayOf(arrayOf("e", target)), content, "sig") + @Test fun emptyInput() { assertEquals(emptyList(), BuzzJobAggregator.aggregate(emptyList())) @@ -172,6 +181,48 @@ class BuzzJobAggregatorTest { assertEquals(requester, job.requester) } + @Test + fun upvotesCountDistinctReactorsIgnoringDislikes() { + val bob = "b".repeat(64) + val carol = "c".repeat(64) + val events = + listOf( + request(), + like(bob, id = "e1".padEnd(64, '0')), + like(carol, id = "e2".padEnd(64, '0')), + like(bob, id = "e3".padEnd(64, '0')), // same reactor again → not double-counted + like(carol, id = "e4".padEnd(64, '0'), content = "-"), // dislike → excluded + ) + val job = BuzzJobAggregator.aggregate(events).single() + assertEquals(2, job.upvotes) + } + + @Test + fun byPriorityOrdersByUpvotesThenAge() { + val jobA = "a".repeat(64) // older, 0 upvotes + val jobB = "b".repeat(64) // newer, 2 upvotes + val jobC = "c".repeat(64) // oldest, 0 upvotes + + fun req( + id: String, + at: Long, + ): JobRequestEvent { + val t = JobRequestEvent.build("task $id", channel, agent, at) + return JobRequestEvent(id, requester, t.createdAt, t.tags, t.content, "sig") + } + val events = + listOf( + req(jobA, 2000), + req(jobB, 3000), + req(jobC, 1000), + like("1".repeat(64), target = jobB, id = "u1".padEnd(64, '0')), + like("2".repeat(64), target = jobB, id = "u2".padEnd(64, '0')), + ) + val ordered = BuzzJobAggregator.byPriority(BuzzJobAggregator.aggregate(events)) + // B first (2 upvotes), then the 0-upvote jobs oldest-first: C (1000) before A (2000). + assertEquals(listOf(jobB, jobC, jobA), ordered.map { it.jobId }) + } + @Test fun multipleJobsSortedByUpdatedAtDesc() { val otherId = "9".repeat(64)