feat(cli): buzz agent-job loop — file/track jobs + a driving responder

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 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_011mApqAbr8vkLC7gUDjavu6
2026-10-05 19:28:25 +00:00 · 2026-07-25 17:18:30 +00:00
parent a488c589f5
commit 635e785753
11 changed files with 1157 additions and 0 deletions
@@ -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 <text> [--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)

@@ -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 🆕. |
@@ -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.
@@ -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 <text>              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
@@ -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<String>,
+    ): Int =
+        route(
+            "buzz agent",
+            tail,
+            USAGE,
+            mapOf(
+                "serve" to { rest -> serve(dataDir, rest) },
+            ),
+        )
+
+    private suspend fun serve(
+        dataDir: DataDir,
+        rest: Array<String>,
+    ): 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<HexKey>()
+            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<Map<String, Any?>>()
+            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<String, Any?> {
+        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<String, String>,
+        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<out Event>,
+    ) {
+        val signed = ctx.signer.sign(template)
+        ctx.publish(signed, setOf(relay))
+    }
+
+    private const val MAX_BODY = 60_000
+}
@@ -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) },
            ),
        )

@@ -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 <text>            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<String>,
+    ): 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 <text> [--agent PUBKEY] [--channel GID]` → publishes a kind-43001. */
+    private suspend fun request(
+        dataDir: DataDir,
+        rest: Array<String>,
+    ): Int {
+        val args = Args(rest)
+        val usage = "buzz job request RELAY <text> [--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<String>,
+    ): 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<String>,
+    ): 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<String>,
+    ): 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<JobView> {
+        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<String, Any?> =
+        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)
+}
@@ -7,3 +7,4 @@ relaygroup/state-relaygroup-headless/
 sync/state-sync-deletions/
 blossom/state-blossom-live/
 git/state-git-nip34/
+buzz/state-job-loop/
@@ -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 <label> <condition-desc> <actual> <expected>
+    if [[ "$3" == "$4" ]]; then echo "  ✓ PASS $1 ($3)" >&2; PASS=$((PASS+1));
+    else echo "  ✗ FAIL $1: expected [$4] got [$3]" >&2; FAIL=$((FAIL+1)); fi
+}
+contains() { # contains <label> <haystack> <needle>
+    if [[ "$2" == *"$3"* ]]; then echo "  ✓ PASS $1" >&2; PASS=$((PASS+1));
+    else echo "  ✗ FAIL $1: [$2] has no [$3]" >&2; FAIL=$((FAIL+1)); fi
+}
+
+run_req init >/dev/null; run_agent init >/dev/null
+ALICE=$(run_req whoami | jkey hex)
+BOT=$(run_agent whoami | jkey hex)
+echo "> alice=$ALICE bot=$BOT" >&2
+
+HOME="$AGENT_HOME" "$AMY_BIN" --account bot --secret-backend plaintext serve --port "$PORT" >"$RELAY_LOG" 2>&1 &
+RELAY_PID=$!
+for _ in $(seq 1 60); do grep -q "relay up" "$RELAY_LOG" && break; sleep 0.5; done
+grep -q "relay up" "$RELAY_LOG" || { echo "relay never came up" >&2; cat "$RELAY_LOG" >&2; exit 1; }
+
+echo "> 1. alice files a job targeting the bot" >&2
+JOB=$(run_req buzz job request "$RELAY" "hello from alice" --agent "$BOT")
+JOB_ID=$(echo "$JOB" | jkey job_id)
+check "request published" _ "$(echo "$JOB" | jkey published)" "true"
+[[ -n "$JOB_ID" ]] && { echo "  ✓ PASS got job id $JOB_ID" >&2; PASS=$((PASS+1)); } || { echo "  ✗ FAIL no job id" >&2; FAIL=$((FAIL+1)); }
+
+echo "> 2. UNAUTHORIZED responder (allowlist excludes alice) handles nothing" >&2
+UNAUTH=$(run_agent buzz agent serve "$RELAY" --exec 'cat' --accept-from "$BOT" --once)
+check "allowlist blocks non-listed requester" _ "$(echo "$UNAUTH" | jkey handled)" "0"
+
+echo "> 3. authorized responder handles it, piping task text through --exec" >&2
+AUTH=$(run_agent buzz agent serve "$RELAY" --exec 'printf "PR opened for: "; cat' --accept-from "$ALICE" --once)
+check "authorized responder handled 1" _ "$(echo "$AUTH" | jkey handled)" "1"
+
+echo "> 4. alice reads back the folded job" >&2
+SHOW=$(run_req buzz job show "$RELAY" "$JOB_ID")
+check "state" _ "$(echo "$SHOW" | jkey state)" "completed"
+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 "" >&2
+echo "> RESULTS: $PASS passed, $FAIL failed" >&2
+[[ $FAIL -eq 0 ]] && exit 0 || exit 1
@@ -0,0 +1,173 @@
+/*
+ * 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.commons.model.buzz
+
+import androidx.compose.runtime.Immutable
+import com.vitorpamplona.quartz.buzz.jobs.JobAcceptedEvent
+import com.vitorpamplona.quartz.buzz.jobs.JobCancelEvent
+import com.vitorpamplona.quartz.buzz.jobs.JobErrorEvent
+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.nip01Core.core.HexKey
+
+/** The lifecycle state of one Buzz agent job, folded from its 43001-43006 events. */
+enum class JobState {
+    /** Only the 43001 request is present — no agent has picked it up yet. */
+    REQUESTED,
+
+    /** An agent published a 43002 acceptance. */
+    ACCEPTED,
+
+    /** At least one 43003 progress ping has arrived after acceptance. */
+    IN_PROGRESS,
+
+    /** Terminal success — a 43004 result. */
+    COMPLETED,
+
+    /** Terminal failure — a 43006 error. */
+    FAILED,
+
+    /** Terminal cancellation — a 43005 cancel. */
+    CANCELLED,
+}
+
+/**
+ * One agent job, correlated from its request and every reply that references it.
+ *
+ * A job is identified by the id of its [JobRequestEvent] (kind 43001). Every reply
+ * (accepted/progress/result/error/cancel) points back to that id via its `e` tag, so
+ * the whole thread folds into a single record regardless of how many relays served it.
+ */
+@Immutable
+data class JobView(
+    val jobId: HexKey,
+    /** The request author — who asked for the work (or a `p` counterparty when the request event is absent). */
+    val requester: HexKey?,
+    /** The target/responding agent — the request `p` tag, else whoever signed the agent-side replies. */
+    val agent: HexKey?,
+    /** The `h` channel UUID the job is scoped to, if any. */
+    val channel: String?,
+    /** The task description (43001 content); null when the request event wasn't in the input set. */
+    val request: String?,
+    val state: JobState,
+    /** The most recent 43003 progress message, if any. */
+    val lastProgress: String?,
+    val progressUpdates: Int,
+    /** The 43004 result payload, when [state] is COMPLETED. */
+    val result: String?,
+    /** The 43006 error message, when [state] is FAILED. */
+    val error: String?,
+    /** The 43005 cancel reason, when [state] is CANCELLED. */
+    val cancelReason: String?,
+    /** 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. */
+    val updatedAt: Long,
+) {
+    val isTerminal: Boolean get() = state == JobState.COMPLETED || state == JobState.FAILED || state == JobState.CANCELLED
+}
+
+/**
+ * Folds a flat list of Buzz agent-job events (kinds 43001-43006, as dispatched by
+ * `EventFactory`) into per-job [JobView] records. Pure and platform-agnostic so both the
+ * `amy` CLI and a future mobile Jobs board share one correlation + state machine.
+ *
+ * SCHEMA CAVEAT: the 43001-43006 protocol is *reserved* in Buzz (no upstream builder);
+ * the tag layout this reads (`e` request reference, `h` channel, `p` counterparty,
+ * `status` token) is Quartz's best-effort model and must be reconciled once Buzz
+ * implements the job protocol. See [JobRequestEvent].
+ */
+object BuzzJobAggregator {
+    fun aggregate(events: List<Event>): List<JobView> {
+        if (events.isEmpty()) return emptyList()
+
+        // Correlate every 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<HexKey, MutableList<Event>>()
+        events.distinctBy { it.id }.forEach { e ->
+            val jobId =
+                when (e) {
+                    is JobRequestEvent -> e.id
+                    is JobAcceptedEvent -> e.jobRequest()
+                    is JobProgressEvent -> e.jobRequest()
+                    is JobResultEvent -> e.jobRequest()
+                    is JobErrorEvent -> e.jobRequest()
+                    is JobCancelEvent -> e.jobRequest()
+                    else -> null
+                } ?: return@forEach
+            byJob.getOrPut(jobId) { mutableListOf() }.add(e)
+        }
+
+        return byJob
+            .map { (jobId, thread) -> fold(jobId, thread) }
+            .sortedByDescending { it.updatedAt }
+    }
+
+    private fun fold(
+        jobId: HexKey,
+        thread: List<Event>,
+    ): JobView {
+        val request = thread.filterIsInstance<JobRequestEvent>().maxByOrNull { it.createdAt }
+        val accepted = thread.filterIsInstance<JobAcceptedEvent>().maxByOrNull { it.createdAt }
+        val progress = thread.filterIsInstance<JobProgressEvent>().sortedBy { it.createdAt }
+        val result = thread.filterIsInstance<JobResultEvent>().maxByOrNull { it.createdAt }
+        val error = thread.filterIsInstance<JobErrorEvent>().maxByOrNull { it.createdAt }
+        val cancel = thread.filterIsInstance<JobCancelEvent>().maxByOrNull { it.createdAt }
+
+        // Among terminal outcomes, the newest one wins (a result can supersede a cancel and
+        // vice-versa depending on arrival order — trust the latest signed timestamp).
+        val terminal =
+            listOfNotNull(
+                result?.let { JobState.COMPLETED to it.createdAt },
+                error?.let { JobState.FAILED to it.createdAt },
+                cancel?.let { JobState.CANCELLED to it.createdAt },
+            ).maxByOrNull { it.second }?.first
+
+        val state =
+            terminal
+                ?: when {
+                    progress.isNotEmpty() -> JobState.IN_PROGRESS
+                    accepted != null -> JobState.ACCEPTED
+                    else -> JobState.REQUESTED
+                }
+
+        val requester = request?.pubKey ?: cancel?.pubKey ?: accepted?.requester() ?: result?.requester()
+        val agent = request?.target() ?: accepted?.pubKey ?: result?.pubKey ?: error?.pubKey ?: progress.lastOrNull()?.pubKey
+
+        return JobView(
+            jobId = jobId,
+            requester = requester,
+            agent = agent,
+            channel = request?.channel() ?: accepted?.channel() ?: result?.channel(),
+            request = request?.request(),
+            state = state,
+            lastProgress = progress.lastOrNull()?.content,
+            progressUpdates = progress.size,
+            result = result?.result(),
+            error = error?.error(),
+            cancelReason = cancel?.reason()?.ifBlank { null },
+            createdAt = request?.createdAt ?: thread.minOf { it.createdAt },
+            updatedAt = thread.maxOf { it.createdAt },
+        )
+    }
+}
@@ -0,0 +1,186 @@
+/*
+ * 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.commons.model.buzz
+
+import com.vitorpamplona.quartz.buzz.jobs.JobAcceptedEvent
+import com.vitorpamplona.quartz.buzz.jobs.JobCancelEvent
+import com.vitorpamplona.quartz.buzz.jobs.JobErrorEvent
+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 kotlin.test.Test
+import kotlin.test.assertEquals
+import kotlin.test.assertNull
+import kotlin.test.assertTrue
+
+class BuzzJobAggregatorTest {
+    private val requester = "a".repeat(64)
+    private val agent = "f".repeat(64)
+    private val channel = "3f2504e0-4f89-41d3-9a0c-0305e82c3301"
+    private val jobId = "1".repeat(64)
+
+    // Events are built via the quartz template and constructed directly with an explicit
+    // id/author/createdAt (no signing) — mirroring the quartz job-event unit tests. The
+    // aggregator only reads typed accessors, so unsigned fixtures exercise it faithfully.
+    private fun request(
+        id: String = jobId,
+        createdAt: Long = 1000,
+    ): JobRequestEvent {
+        val t = JobRequestEvent.build("fix the login bug", channel, agent, createdAt)
+        return JobRequestEvent(id, requester, t.createdAt, t.tags, t.content, "sig")
+    }
+
+    private fun accepted(createdAt: Long = 1010): JobAcceptedEvent {
+        val t = JobAcceptedEvent.build(jobId, channel, requester, "on it", createdAt)
+        return JobAcceptedEvent("2".repeat(64), agent, t.createdAt, t.tags, t.content, "sig")
+    }
+
+    private fun progress(
+        msg: String,
+        createdAt: Long,
+        id: String,
+    ): JobProgressEvent {
+        val t = JobProgressEvent.build(jobId, msg, channel, "running", createdAt)
+        return JobProgressEvent(id, agent, t.createdAt, t.tags, t.content, "sig")
+    }
+
+    private fun result(createdAt: Long = 1040): JobResultEvent {
+        val t = JobResultEvent.build(jobId, "opened PR #123", channel, requester, "completed", createdAt)
+        return JobResultEvent("4".repeat(64), agent, t.createdAt, t.tags, t.content, "sig")
+    }
+
+    private fun error(createdAt: Long = 1040): JobErrorEvent {
+        val t = JobErrorEvent.build(jobId, "build failed", channel, "error", createdAt)
+        return JobErrorEvent("6".repeat(64), agent, t.createdAt, t.tags, t.content, "sig")
+    }
+
+    private fun cancel(createdAt: Long = 1035): JobCancelEvent {
+        val t = JobCancelEvent.build(jobId, "changed my mind", channel, createdAt)
+        return JobCancelEvent("5".repeat(64), requester, t.createdAt, t.tags, t.content, "sig")
+    }
+
+    @Test
+    fun emptyInput() {
+        assertEquals(emptyList(), BuzzJobAggregator.aggregate(emptyList()))
+    }
+
+    @Test
+    fun fullHappyPath() {
+        val events: List<Event> =
+            listOf(
+                request(),
+                accepted(),
+                progress("cloning", 1020, "a1".padEnd(64, '0')),
+                progress("running tests", 1030, "a2".padEnd(64, '0')),
+                result(),
+            )
+        val jobs = BuzzJobAggregator.aggregate(events)
+        assertEquals(1, jobs.size)
+        val job = jobs.single()
+        assertEquals(jobId, job.jobId)
+        assertEquals(requester, job.requester)
+        assertEquals(agent, job.agent)
+        assertEquals(channel, job.channel)
+        assertEquals("fix the login bug", job.request)
+        assertEquals(JobState.COMPLETED, job.state)
+        assertTrue(job.isTerminal)
+        assertEquals("opened PR #123", job.result)
+        assertEquals("running tests", job.lastProgress)
+        assertEquals(2, job.progressUpdates)
+        assertEquals(1000, job.createdAt)
+        assertEquals(1040, job.updatedAt)
+    }
+
+    @Test
+    fun requestOnlyIsRequested() {
+        val job = BuzzJobAggregator.aggregate(listOf(request())).single()
+        assertEquals(JobState.REQUESTED, job.state)
+        assertTrue(!job.isTerminal)
+        assertNull(job.result)
+        assertEquals(0, job.progressUpdates)
+    }
+
+    @Test
+    fun acceptedWithoutProgressIsAccepted() {
+        val job = BuzzJobAggregator.aggregate(listOf(request(), accepted())).single()
+        assertEquals(JobState.ACCEPTED, job.state)
+    }
+
+    @Test
+    fun progressWithoutTerminalIsInProgress() {
+        val job =
+            BuzzJobAggregator
+                .aggregate(listOf(request(), accepted(), progress("working", 1020, "a1".padEnd(64, '0'))))
+                .single()
+        assertEquals(JobState.IN_PROGRESS, job.state)
+    }
+
+    @Test
+    fun errorIsFailed() {
+        val job = BuzzJobAggregator.aggregate(listOf(request(), accepted(), error())).single()
+        assertEquals(JobState.FAILED, job.state)
+        assertEquals("build failed", job.error)
+    }
+
+    @Test
+    fun cancelIsCancelled() {
+        val job = BuzzJobAggregator.aggregate(listOf(request(), cancel())).single()
+        assertEquals(JobState.CANCELLED, job.state)
+        assertEquals("changed my mind", job.cancelReason)
+    }
+
+    @Test
+    fun newestTerminalWins() {
+        // A cancel at 1035 followed by a result at 1040: the later result is authoritative.
+        val job = BuzzJobAggregator.aggregate(listOf(request(), cancel(1035), result(1040))).single()
+        assertEquals(JobState.COMPLETED, job.state)
+
+        // Reversed timing: a result at 1030 then a cancel at 1050 → CANCELLED.
+        val job2 = BuzzJobAggregator.aggregate(listOf(request(), result(1030), cancel(1050))).single()
+        assertEquals(JobState.CANCELLED, job2.state)
+    }
+
+    @Test
+    fun orphanRepliesWithoutRequestStillFold() {
+        // The 43001 request wasn't served to us, only the agent's result. We still surface
+        // the job, keyed by the `e`-referenced request id, with the request text unknown.
+        val job = BuzzJobAggregator.aggregate(listOf(accepted(), result())).single()
+        assertEquals(jobId, job.jobId)
+        assertNull(job.request)
+        assertEquals(JobState.COMPLETED, job.state)
+        assertEquals(agent, job.agent)
+        // requester is recovered from the reply's `p` tag when the request event is absent.
+        assertEquals(requester, job.requester)
+    }
+
+    @Test
+    fun multipleJobsSortedByUpdatedAtDesc() {
+        val otherId = "9".repeat(64)
+        val otherReq = JobRequestEvent.build("second task", channel, agent, 2000)
+        val other = JobRequestEvent(otherId, requester, otherReq.createdAt, otherReq.tags, otherReq.content, "sig")
+
+        val jobs = BuzzJobAggregator.aggregate(listOf(request(), result(), other))
+        assertEquals(2, jobs.size)
+        // `request()`+`result()` updates at 1040; `other` at 2000 → other is newest, first.
+        assertEquals(otherId, jobs.first().jobId)
+    }
+}