From 5b3a40adf4c031ed451afe2c81ad43f10b99a4e1 Mon Sep 17 00:00:00 2001 From: Stener1 Date: Fri, 6 Nov 2026 12:39:54 +0000 Subject: [PATCH 1/4] feat(box): a box keeps attention items, so a plugin's snooze follows you between computers (BRA-3274) Co-Authored-By: Claude Opus 5.5 --- .../verify-berthd/scripts/attention-live.sh | 72 +++ .../skills/verify-berthd/scripts/drive.mjs | 2 +- .../scenarios/attention-two-laptops-live.mjs | 168 +++++++ app/src/lib/attention.test.ts | 65 +++ app/src/lib/attention.ts | 42 ++ app/src/plugins/context.ts | 2 + cmd/berthd/main.go | 1 + docs/changelog.mdx | 7 + docs/reference/app-api.mdx | 1 + docs/reference/events.mdx | 4 + docs/reference/plugin-sdk.mdx | 46 ++ internal/box/api.go | 6 + internal/box/attention.go | 475 ++++++++++++++++++ internal/box/attention_test.go | 263 ++++++++++ internal/box/attentionapi.go | 156 ++++++ internal/box/turns.go | 1 + internal/box/upgrade.go | 5 + packages/plugin-sdk/src/index.ts | 28 ++ packages/plugin-sdk/src/types.ts | 67 +++ 19 files changed, 1410 insertions(+), 1 deletion(-) create mode 100644 .claude/skills/verify-berthd/scripts/attention-live.sh create mode 100644 .claude/skills/verify-berthd/scripts/scenarios/attention-two-laptops-live.mjs create mode 100644 app/src/lib/attention.test.ts create mode 100644 app/src/lib/attention.ts create mode 100644 internal/box/attention.go create mode 100644 internal/box/attention_test.go create mode 100644 internal/box/attentionapi.go diff --git a/.claude/skills/verify-berthd/scripts/attention-live.sh b/.claude/skills/verify-berthd/scripts/attention-live.sh new file mode 100644 index 000000000..3202daef3 --- /dev/null +++ b/.claude/skills/verify-berthd/scripts/attention-live.sh @@ -0,0 +1,72 @@ +#!/usr/bin/env bash +# attention-live.sh up|env|down: one box and two laptop agents of this +# checkout's build, as two computers paired with the same box, for the +# attention scenario (scenarios/attention-two-laptops-live.mjs). No agent +# CLI runs: the scenario starts a plain shell session itself. +# +# Its own ports, so it never meets dev-box.sh (7544/9377/9378) or the +# machine's Shipyard (1377-1379): the box on 127.0.0.1:7547, laptop A on +# 9471/9472 and laptop B on 9473/9474 (proxy/UI). `env` prints what +# drive.mjs needs; `down` stops it all and deletes the state. +set -euo pipefail +root=$(git -C "$(dirname "$0")" rev-parse --show-toplevel) +dir=${VERIFY_BOX_DIR:-/tmp/vb-attention} +addr=127.0.0.1:7547 +box_env=(BERTH_HOME="$dir/box-home" BERTH_USER_DIR="$dir/box-user" TMUX_TMPDIR="$dir/tmux") +lap_env() { echo BERTH_HOME="$dir/lap-$1-home" BERTH_USER_DIR="$dir/lap-$1-user" BERTH_PROXY_ADDR="127.0.0.1:$2" BERTH_UI_ADDR="127.0.0.1:$3"; } +box() { env "${box_env[@]}" "$root/bin/berthd" "$@"; } +# shellcheck disable=SC2046 +lap() { local who=$1 p=$2 u=$3; shift 3; env $(lap_env "$who" "$p" "$u") "$root/bin/berth" "$@"; } +tm() { env TMUX_TMPDIR="$dir/tmux" tmux -L berth "$@"; } + +case ${1:-} in +up) + for p in 7547 9471 9472 9473 9474; do + if ss -ltn | grep -q ":$p "; then echo "port $p is taken: is it up already? ($0 down)" >&2; exit 1; fi + done + mkdir -p "$dir"/{box-home,box-user,lap-a-home,lap-a-user,lap-b-home,lap-b-user,tmux} && chmod 700 "$dir/tmux" + make -C "$root" build >/dev/null + env "${box_env[@]}" nohup "$root/bin/berthd" serve --listen "$addr" >"$dir/berthd.log" 2>&1 & + echo $! >"$dir/berthd.pid" + until grep -q "berthd serving" "$dir/berthd.log"; do sleep 0.5; done + if grep -q "updated berth's hooks" "$dir/berthd.log"; then + echo "the box rewrote agent hooks in $HOME (berthd.log): put them back before going on" >&2 + exit 1 + fi + # Each laptop pairs with a code of its own, as two computers do. + lap a 9471 9472 pair "$(box pair --address "$addr" --json | jq -r .link)" --name devq >/dev/null + lap b 9473 9474 pair "$(box pair --address "$addr" --json | jq -r .link)" --name devq >/dev/null + repo=$dir/acme-shop + if [[ ! -d $repo/.git ]]; then + git init -q -b main "$repo" && echo "# acme shop" >"$repo/README.md" + git -C "$repo" add . && git -C "$repo" -c user.email=dev@acme.test -c user.name=acme commit -qm init + fi + box location add acme-shop "$repo" >/dev/null + # The CLI starts each agent; wait until both answer. + lap a 9471 9472 ui-token >/dev/null && lap b 9473 9474 ui-token >/dev/null + for u in 9472 9474; do + until ss -ltn | grep -q "127.0.0.1:$u "; do sleep 0.5; done + done + echo "up: box devq ($addr), laptop A http://127.0.0.1:9472, laptop B http://127.0.0.1:9474" + ;; +env) + # The tokens are secrets: export them, don't print them anywhere else. + echo "export VERIFY_BERTH_TOKEN=$(lap a 9471 9472 ui-token | jq -r .token) VERIFY_BERTH_TOKEN_B=$(lap b 9473 9474 ui-token | jq -r .token) VERIFY_AGENT_B=http://127.0.0.1:9474 TMUX_TMPDIR=$dir/tmux" + ;; +down) + [[ -f $dir/berthd.pid ]] && kill "$(cat "$dir/berthd.pid")" 2>/dev/null || true + tm kill-server 2>/dev/null || true + # The agents the CLI started: the ones on 9472/9474 with this state, no other. + for port in 9472 9474; do + for pid in $(ss -ltnpH "sport = :$port" 2>/dev/null | grep -o 'pid=[0-9]*' | cut -d= -f2 | sort -u); do + tr '\0' '\n' <"/proc/$pid/environ" | grep -q "^BERTH_HOME=$dir/lap-" && kill "$pid" + done + done + rm -rf "$dir" + echo "down: $dir removed" + ;; +*) + echo "usage: attention-live.sh up|env|down" >&2 + exit 2 + ;; +esac diff --git a/.claude/skills/verify-berthd/scripts/drive.mjs b/.claude/skills/verify-berthd/scripts/drive.mjs index 4917aae96..2871bc901 100755 --- a/.claude/skills/verify-berthd/scripts/drive.mjs +++ b/.claude/skills/verify-berthd/scripts/drive.mjs @@ -102,7 +102,7 @@ try { step(agent ? `open http://localhost:${port}/ on the agent at ${agent}` : `open http://localhost:${port}/?${params}`); await page.goto(`http://localhost:${port}/?${params}`); await expect(page.getByTestId("nav-home")).toBeVisible({ timeout: 20_000 }); - await scenario.run({ page, shot, step, expect, call }); + await scenario.run({ page, shot, step, expect, call, out }); } catch (err) { ok = false; failures.push(redact(`scenario failed: ${err.message?.split("\n").slice(0, 6).join("\n")}`)); diff --git a/.claude/skills/verify-berthd/scripts/scenarios/attention-two-laptops-live.mjs b/.claude/skills/verify-berthd/scripts/scenarios/attention-two-laptops-live.mjs new file mode 100644 index 000000000..7421f962f --- /dev/null +++ b/.claude/skills/verify-berthd/scripts/scenarios/attention-two-laptops-live.mjs @@ -0,0 +1,168 @@ +// Attention items on a real box, read from two computers +// (features/plugins.md, plugin-attention): a snooze made in a plugin on +// laptop A is what the same plugin reads on laptop B, though B's own plugin +// storage is empty; B resolving an item shows on A at once, from the box's +// event; and an open item settles on both when its session ends. +// +// On ../attention-live.sh (one box "devq", laptop A on 9472, laptop B on +// 9474): +// +// ../attention-live.sh up && eval "$(../attention-live.sh env)" +// node drive.mjs attention-two-laptops-live --port $PORT --agent http://127.0.0.1:9472 +// +// It starts a shell session "att-shell" in acme-shop and posts two asks +// about it, as an agent would. + +export const live = true; +export const params = { labs: "0" }; + +const BOX = "devq"; +const SESSION = "att-shell"; +const ASK = `session:${SESSION}`; +const DEPLOY = "acme/deploy-check"; +const UNTIL = "2030-01-02T09:00:00Z"; +const AGENT_B = process.env.VERIFY_AGENT_B ?? "http://127.0.0.1:9474"; +const TOKEN_B = process.env.VERIFY_BERTH_TOKEN_B; + +export const plugins = [ + { + id: "attention-check", + name: "Attention check", + source: ` +import { createElement as h, useEffect, useState } from "react"; +export default function (berth) { + const BOX = ${JSON.stringify(BOX)}; + const Screen = () => { + const [items, setItems] = useState(null); + const [error, setError] = useState(""); + const [local, setLocal] = useState(berth.storage.get("snoozed", "")); + const load = () => berth.attention.list(BOX).then(setItems, (e) => setError(e.message)); + useEffect(() => { + load(); + return berth.on("attention.*", load); + }, []); + const act = (key, change, note) => { + if (note) { + berth.storage.set("snoozed", note); + setLocal(note); + } + berth.attention.update(BOX, key, change).then(load, (e) => setError(e.message)); + }; + const line = (it) => + it.key + " · " + it.kind + " · " + it.status + (it.closed_reason ? " (" + it.closed_reason + ")" : "") + + " · " + (it.snoozed_until ? "snoozed until " + it.snoozed_until : "not snoozed") + " · rev " + it.revision + " · by " + it.by; + return h("div", { style: { padding: 16, display: "grid", gap: 8 } }, + h("p", null, "berth.attention is " + typeof berth.attention), + h("p", { "data-testid": "att-local" }, "This computer's plugin storage: " + (local || "nothing")), + error && h("p", { role: "alert" }, error), + h("ul", { "aria-label": "Attention items" }, (items ?? []).map((it) => + h("li", { key: it.key }, + h("span", { "data-testid": "att-row" }, line(it)), + it.title && h("span", null, " — " + it.title), + " ", + h("button", { onClick: () => act(it.key, { snoozed_until: ${JSON.stringify(UNTIL)}, revision: it.revision }, it.key) }, "Snooze " + it.key), + " ", + h("button", { onClick: () => act(it.key, { status: "closed", revision: it.revision }) }, "Resolve " + it.key), + ))), + items && h("p", null, items.length + " items on " + BOX), + ); + }; + berth.addScreen({ id: "attention-check", title: "Attention check", Component: Screen }); + berth.addSidebarItem({ id: "attention-check", title: "Attention check", icon: "Bell", screen: "attention-check" }); +} +`, + }, +]; + +const openScreen = async (page, expect) => { + await page.getByTestId("nav-more").click(); + await page.getByRole("menu").getByRole("menuitem", { name: "Attention check" }).click(); + await expect(page.getByText("berth.attention is object")).toBeVisible(); +}; + +const row = (page, key) => page.getByTestId("att-row").filter({ hasText: new RegExp(`^${key.replace(/[/#]/g, "\\$&")} ·`) }); + +export async function run({ page, shot, step, expect, call, out }) { + if (!TOKEN_B) throw new Error("VERIFY_BERTH_TOKEN_B is not set: eval \"$(attention-live.sh env)\""); + const box = (path, init) => call(`/v1/boxes/${BOX}/api${path}`, init); + const info = await box("/info"); + if (!info.capabilities?.includes("attention")) throw new Error(`${BOX} doesn't list the attention capability: ${info.capabilities}`); + + step(`start a shell session ${SESSION} in acme-shop, and post two asks about it as its agent would`); + await box("/sessions", { method: "POST", body: JSON.stringify({ location: "acme-shop", name: SESSION, command: "cat" }) }); + await box("/attention", { method: "POST", body: JSON.stringify({ session: SESSION, kind: "decision", title: "Ship the acme checkout on Friday?", options: [{ label: "Yes" }, { label: "Not yet" }], recommended: "Not yet" }) }); + await box("/attention", { method: "POST", body: JSON.stringify({ key: DEPLOY, session: SESSION, kind: "action", title: "Log in to the acme staging deploy" }) }); + + step("laptop A: open the plugin's screen; both asks are open and not snoozed"); + await openScreen(page, expect); + await expect(row(page, ASK)).toHaveText(`${ASK} · decision · open · not snoozed · rev 1 · by app`); + await expect(row(page, DEPLOY)).toHaveText(`${DEPLOY} · action · open · not snoozed · rev 1 · by app`); + await expect(page.getByTestId("att-local")).toHaveText("This computer's plugin storage: nothing"); + await shot("1-a-two-open-asks"); + + step(`laptop A: snooze ${ASK} until ${UNTIL} (the plugin also notes it in its own storage on A)`); + await page.getByRole("button", { name: `Snooze ${ASK}` }).click(); + await expect(row(page, ASK)).toHaveText(`${ASK} · decision · open · snoozed until ${UNTIL} · rev 2 · by app`); + await expect(page.getByTestId("att-local")).toHaveText(`This computer's plugin storage: ${ASK}`); + await shot("2-a-snoozed"); + + step("laptop B: another browser, its own laptop agent paired with the same box, its own plugin storage"); + const browser = page.context().browser(); + const ctxB = await browser.newContext({ viewport: { width: 1440, height: 900 }, colorScheme: "dark" }); + await ctxB.route(/^https?:\/\/(?:[^/]*\.)?(?:localhost|127\.0\.0\.1|\[::1\]):(?:1377|1378|1379)(?:\/|$)/, (r) => r.abort()); + const pageB = await ctxB.newPage(); + const errorsB = []; + pageB.on("console", (m) => m.type() === "error" && errorsB.push(m.text())); + pageB.on("pageerror", (e) => errorsB.push(e.message)); + await pageB.route("**/builtin-plugins/index.json", async (r) => { + const res = await r.fetch(); + const list = await res.json(); + await r.fulfill({ response: res, json: [...list, ...plugins.map((p) => ({ id: p.id, name: p.name, version: "0.0.0-verify", main: "index.js" }))] }); + }); + for (const p of plugins) await pageB.route(`**/builtin-plugins/${p.id}/index.js`, (r) => r.fulfill({ contentType: "text/javascript", body: p.source })); + const origin = new URL(page.url()).origin; + await pageB.goto(`${origin}/?${new URLSearchParams({ token: TOKEN_B, agent: AGENT_B, ...params })}`); + await expect(pageB.getByTestId("nav-home")).toBeVisible({ timeout: 20_000 }); + // B's screenshots go beside the driver's own. + const shotB = (label) => pageB.screenshot({ path: `${out}/${label}.png` }); + try { + await openScreen(pageB, expect); + step("laptop B: the snooze made on A is there, read from the box; B's own plugin storage has nothing"); + await expect(row(pageB, ASK)).toHaveText(`${ASK} · decision · open · snoozed until ${UNTIL} · rev 2 · by app`); + await expect(pageB.getByTestId("att-local")).toHaveText("This computer's plugin storage: nothing"); + await shotB("3-b-reads-the-snooze"); + + step(`laptop B: resolve ${ASK}; laptop A shows it closed without a reload, from the box's attention.updated event`); + await pageB.getByRole("button", { name: `Resolve ${ASK}` }).click(); + await expect(row(pageB, ASK)).toHaveText(`${ASK} · decision · closed (resolved) · snoozed until ${UNTIL} · rev 3 · by app`); + await expect(row(page, ASK)).toHaveText(`${ASK} · decision · closed (resolved) · snoozed until ${UNTIL} · rev 3 · by app`); + await shot("4-a-sees-b-resolve"); + + step(`a stale write is refused: A snoozes ${ASK} again from revision 2`); + const stale = await fetch(`${AGENT_B}/v1/boxes/${BOX}/api/attention/${encodeURIComponent(ASK)}`, { method: "PATCH", headers: { Authorization: `Bearer ${TOKEN_B}`, "Content-Type": "application/json" }, body: JSON.stringify({ revision: 2, snoozed_until: "" }) }); + if (stale.status !== 409) throw new Error(`a stale revision answered ${stale.status}, not 409`); + + step(`end session ${SESSION}: its open ask ${DEPLOY} settles on both laptops, as the box's own change`); + await box(`/sessions/${SESSION}`, { method: "DELETE" }); + await expect(row(page, DEPLOY)).toHaveText(`${DEPLOY} · action · closed (session_ended) · not snoozed · rev 2 · by berth`); + await expect(row(pageB, DEPLOY)).toHaveText(`${DEPLOY} · action · closed (session_ended) · not snoozed · rev 2 · by berth`); + await shot("5-a-session-ended"); + await shotB("6-b-session-ended"); + + step("second view: the box's API through laptop B says the same"); + const res = await fetch(`${AGENT_B}/v1/boxes/${BOX}/api/attention`, { headers: { Authorization: `Bearer ${TOKEN_B}` } }); + const items = await res.json(); + const ask = items.find((i) => i.key === ASK); + const deploy = items.find((i) => i.key === DEPLOY); + if (ask?.status !== "closed" || ask.closed_reason !== "resolved" || ask.snoozed_until !== UNTIL || deploy?.closed_reason !== "session_ended") { + throw new Error(`the box holds ${JSON.stringify(items)}`); + } + step(`GET attention on laptop B: ${ASK} closed (resolved), snoozed until ${UNTIL}; ${DEPLOY} closed (session_ended)`); + if (errorsB.length) throw new Error(`laptop B's console: ${errorsB.join(" | ")}`); + } catch (err) { + await shotB("failed-b").catch(() => {}); + throw err; + } finally { + await ctxB.close(); + } +} diff --git a/app/src/lib/attention.test.ts b/app/src/lib/attention.test.ts new file mode 100644 index 000000000..21fd751c0 --- /dev/null +++ b/app/src/lib/attention.test.ts @@ -0,0 +1,65 @@ +import assert from "node:assert/strict"; +import { test } from "node:test"; + +import type { Client } from "@/lib/api"; + +import { makeAttentionApi } from "./attention.ts"; + +// A client that records the box calls and answers from a script. +function fakeClient(answer: (method: string, path: string) => unknown) { + const calls: { box: string; method: string; path: string; body: unknown }[] = []; + const client = { + box: async (box: string, method: string, path: string, body?: unknown) => { + calls.push({ box, method, path, body }); + return answer(method, path); + }, + } as unknown as Client; + return { client, calls }; +} + +const item = { key: "devl/shop/wait", revision: 2, kind: "mark", status: "open", created: "", updated: "" }; + +test("a key goes to the box path-escaped, slashes and all", async () => { + const { client, calls } = fakeClient(() => item); + const api = makeAttentionApi(client); + await api.update("devl", "devl/shop/wait#3", { snoozed_until: "2026-11-07T08:00:00Z", revision: 1 }); + await api.get("devl", "devl/shop/wait#3"); + assert.deepEqual( + calls.map((c) => [c.box, c.method, c.path]), + [ + ["devl", "PATCH", "attention/devl%2Fshop%2Fwait%233"], + ["devl", "GET", "attention/devl%2Fshop%2Fwait%233"], + ], + ); + assert.deepEqual(calls[0].body, { snoozed_until: "2026-11-07T08:00:00Z", revision: 1 }); +}); + +test("list asks for one session's or the open items, and an empty answer is a list", async () => { + const { client, calls } = fakeClient(() => null); + const api = makeAttentionApi(client); + assert.deepEqual(await api.list("devl", { session: "shop claude", open: true }), []); + await api.list("devl"); + assert.deepEqual( + calls.map((c) => c.path), + ["attention?session=shop+claude&open=1", "attention"], + ); +}); + +test("get is undefined for a key with no item, and other errors still reject", async () => { + const notFound = Object.assign(new Error("no attention item with that key"), { status: 404, code: "not_found" }); + const offline = Object.assign(new Error("devl is offline"), { status: 502, code: "box_offline" }); + let next: Error = notFound; + const { client } = fakeClient(() => { + throw next; + }); + const api = makeAttentionApi(client); + assert.equal(await api.get("devl", "nope"), undefined); + next = offline; + await assert.rejects(api.get("devl", "nope"), /offline/); +}); + +test("post sends the ask as it is", async () => { + const { client, calls } = fakeClient(() => item); + await makeAttentionApi(client).post("devl", { session: "shop-claude", kind: "decision", title: "Which queue?", options: [{ label: "SQS" }], recommended: "SQS" }); + assert.deepEqual(calls[0], { box: "devl", method: "POST", path: "attention", body: { session: "shop-claude", kind: "decision", title: "Which queue?", options: [{ label: "SQS" }], recommended: "SQS" } }); +}); diff --git a/app/src/lib/attention.ts b/app/src/lib/attention.ts new file mode 100644 index 000000000..70ed6d5cd --- /dev/null +++ b/app/src/lib/attention.ts @@ -0,0 +1,42 @@ +import type { AttentionItem, AttentionPost, AttentionUpdate, BerthAttention } from "@berth/plugin"; + +import type { Client } from "@/lib/api"; + +// What needs a person, kept on each box (internal/box/attention.go), so a +// snooze or an answer made on one computer holds on every other. The key +// goes path-escaped: a plugin's keys may hold slashes. + +const itemPath = (key: string) => `attention/${encodeURIComponent(key)}`; + +// A box that holds no item at a key says so with 404 "not_found". +const missing = (err: unknown) => { + const e = err as { status?: number; code?: string } | null; + return e?.status === 404 && e.code === "not_found"; +}; + +export const attentionApi = { + list: async (c: Client, box: string, opts: { session?: string; open?: boolean } = {}) => { + const q = new URLSearchParams(); + if (opts.session) q.set("session", opts.session); + if (opts.open) q.set("open", "1"); + const qs = q.toString(); + return (await c.box(box, "GET", `attention${qs ? `?${qs}` : ""}`)) ?? []; + }, + get: (c: Client, box: string, key: string) => + c.box(box, "GET", itemPath(key)).catch((err: unknown) => { + if (missing(err)) return undefined; + throw err; + }), + post: (c: Client, box: string, item: AttentionPost) => c.box(box, "POST", "attention", item), + update: (c: Client, box: string, key: string, change: AttentionUpdate) => c.box(box, "PATCH", itemPath(key), change), +}; + +// makeAttentionApi is the store as plugins see it (berth.attention). +export function makeAttentionApi(client: Client): BerthAttention { + return { + list: (box, opts) => attentionApi.list(client, box, opts), + get: (box, key) => attentionApi.get(client, box, key), + post: (box, item) => attentionApi.post(client, box, item), + update: (box, key, change) => attentionApi.update(client, box, key, change), + }; +} diff --git a/app/src/plugins/context.ts b/app/src/plugins/context.ts index 33dd48176..69eb77b2b 100644 --- a/app/src/plugins/context.ts +++ b/app/src/plugins/context.ts @@ -3,6 +3,7 @@ import { createContext } from "react"; import { boxApi, type Client } from "@/lib/api"; import { attachApi, localPaths } from "@/lib/attachments"; +import { makeAttentionApi } from "@/lib/attention"; import { makeHelperApi } from "@/lib/helper"; import { makeLinearApi } from "@/lib/linear"; import { notify } from "@/lib/notify"; @@ -101,6 +102,7 @@ export function makeContext(id: string, client: Client): BerthPluginContext { prompts: makePromptsApi(), linear: makeLinearApi(client), helper: makeHelperApi(client, id), + attention: makeAttentionApi(client), orchestrate: { send: orchestrate.send, wait: orchestrate.wait, diff --git a/cmd/berthd/main.go b/cmd/berthd/main.go index d2f48d3f4..e90f17bcd 100644 --- a/cmd/berthd/main.go +++ b/cmd/berthd/main.go @@ -409,6 +409,7 @@ func serve(b boxHome, args []string) error { // A worktree that is removed, however it was, is one they forget. defer bx.Helpers.Watch(bus)() bx.AutoFix = &box.AutoFixStore{Path: filepath.Join(b.dir, "autofix.json")} + bx.Attention = &box.AttentionStore{Path: filepath.Join(b.dir, "attention.json")} // How agents start when what starts them names no driver // (agentdefaults.go): the app keeps it to the person's Labs setting. bx.AgentDefaults = &box.AgentDefaultsStore{Path: filepath.Join(b.dir, "agent-defaults.json")} diff --git a/docs/changelog.mdx b/docs/changelog.mdx index cc4ef4e5a..8e7220f36 100644 --- a/docs/changelog.mdx +++ b/docs/changelog.mdx @@ -61,6 +61,13 @@ again. branch, and asks git again only when the repository changes (or after a minute). +- **A plugin's snooze can follow you between computers.** Each box now keeps + a small list of what needs you: asks an agent posts, and marks such as + "snoozed until 9" or "resolved" on items a plugin builds. Every app paired + with the box reads the same list, so snoozing on one laptop holds on the + other. An item settles by itself when its session ends, and is forgotten + two weeks after it settles. Plugins reach it as `berth.attention`. + ## v0.3.31 - **Worktrees with submodules can be removed.** git refuses to remove any diff --git a/docs/reference/app-api.mdx b/docs/reference/app-api.mdx index 3888a0bb5..c45d4abe0 100644 --- a/docs/reference/app-api.mdx +++ b/docs/reference/app-api.mdx @@ -185,6 +185,7 @@ failed after it went out (the box may have acted on it). The box API: | `POST team` | a `TeamBundle` from the laptop (`id`, `name`, `org`, `commit`, `script`, `steps`, `github`, `files`: path → base64, `projects`: `id`, `repo`, `url`, `path`, `source`, `trust_hash`, `kit`, `init`, `env`) → `TeamStatus`. Starts the steps in a terminal; 409 while it runs. Boxes with the `team` capability | | `POST team/{id}/retry` | `{ from? }` → `TeamStatus` | | `GET shares`, `POST shares`, `DELETE shares/{id}` | public shares | +| `GET attention?session=S&open=1`, `POST attention`, `GET attention/{key}`, `PATCH attention/{key}` | [Attention items](/reference/plugin-sdk#attention): what needs a person, the same for every app paired with the box. `POST` takes `{ key?, kind, title, details?, options?, recommended?, session?, revision? }` and replaces the item at its key; `PATCH` takes `{ revision?, status?, answer?, closed_reason?, snoozed_until?, session? }` and makes a mark at a key with no item. Both answer with the `AttentionItem`; a `revision` that has gone by is refused (409). Path-escape the key (`%2F`, `%23`). Boxes that keep them list `attention` in `capabilities` | | `GET units`, `POST units`, `GET units/{name}`, `DELETE units/{name}`, `POST units/{name}/restart`, `GET units/{name}/log` | units | | `GET helpers` | `{ platform, helpers: HelperOnBox[] }`: the [helpers of plugins](/guides/plugins#give-a-plugin-a-helper) on the box, and the box's platform (`linux-arm64`). A `HelperOnBox` is `{ plugin, state, platform, name?, version?, sha256?, digest?, pid?, since?, restarts?, attach?, error?, attached? }`, with `state` one of `off`, `installing`, `starting`, `running`, `failed`; `digest` names the program and how it is run, for a laptop to compare with a manifest. Boxes that run helpers list `helpers` in `capabilities`; an older one answers 404 | | `GET helpers/{plugin}` | `HelperOnBox`; `state` is `off` for a plugin with no helper here | diff --git a/docs/reference/events.mdx b/docs/reference/events.mdx index def6c26af..ee8983296 100644 --- a/docs/reference/events.mdx +++ b/docs/reference/events.mdx @@ -116,6 +116,8 @@ events. | `run.gate_decided` | the same, with `step`, `approve`, `by` | | `browser.opened`, `browser.closed` | `location`, `name`, `path`; `closed` has `reason` (`idle`, `closed`, the guard, another worktree needing one, its worktree removed) | | `artifact.added`, `artifact.updated`, `artifact.removed` | `id`, `title`, `kind`, `location`, `name`, `path`, `version`, and `session` when an agent made it; `problem` when a rewrite of its file wasn't taken. A rewrite the box noticed has origin `watch`; a visual diff from `berthd shots compare` has origin `shots` and its `summary`. See [artifacts](/guides/artifacts) | +| `attention.posted`, `attention.updated` | `key`, `revision`, `kind`, `status`, and `session`, `closed_reason` and `snoozed_until` when set: an [attention item](/reference/plugin-sdk#attention) was posted or replaced, or answered, closed, reopened or snoozed. Never its title, details or answer: read the item for those | +| `attention.settled` | the same, with `status: "closed"` and `closed_reason: "session_ended"`, origin `berth`: an open item's session ended | | `browser.failed` | `location`, `name`, `path`, `reason`: `sandbox` (Chromium's sandbox can't start; see [agent browser](/guides/agent-browser#when-ubuntu-blocks-it)) or `error` | | `browser.navigated` | `location`, `name`, `path`, `tab`, `url`: the [agent browser](/guides/agent-browser) loaded a page (`tab` is `""` for the main tab) | | `browser.shot` | `location`, `name`, `path`, `tab`, `file`, `url` | @@ -206,6 +208,8 @@ exits non-zero refuses it with what it printed. Gates in a laptop's | `before:run.pr` | `run`, `path`, `branch`: a run about to push and open a pull request | | `before:artifact.add` | `location`, `name`, `path`, `title`, `kind`, `id` (when updating one) | | `before:artifact.remove` | `location`, `name`, `path`, `id` | +| `before:attention.post` | `key`, `kind`, `session`: posting an [attention item](/reference/plugin-sdk#attention) | +| `before:attention.update` | `key`, and `status` and `snoozed_until` when the change sets them | | `before:browser.open` | `location`, `name`, `path`, `tab`, `url`: an agent browser opening a page | | `before:browser.eval`, `before:browser.watch` | `location`, `name`, `path`, `tab`: running JavaScript in it, watching it live | | `before:browser.allow` | `origin`: letting agents' browsers load a public origin | diff --git a/docs/reference/plugin-sdk.mdx b/docs/reference/plugin-sdk.mdx index e7dbcfa01..356854f41 100644 --- a/docs/reference/plugin-sdk.mdx +++ b/docs/reference/plugin-sdk.mdx @@ -29,6 +29,7 @@ that gives `activate` its types. | `prompts` | [`BerthPrompts`](#prompts): the saved prompt library | | `linear` | [`BerthLinear`](#linear): the person's Linear issues, once they added a key | | `helper` | [`BerthHelper`](#helper): the plugin's own program on a box, for a plugin whose manifest carries one | +| `attention` | [`BerthAttention`](#attention): what needs a person, kept on each box so every computer reads the same | | `storage` | `get(key, fallback)`, `set(key, value)`: small JSON values kept on this computer, apart from other plugins' | ### Adding things @@ -304,6 +305,51 @@ helper writes there once it is listening. A request goes through the box as JSON, one answer for one request, so a helper's event stream can't be followed this way: read again on a timer while your view is on screen. +## attention + +What needs a person, kept on each box: an ask an agent posts (a decision, +an action, something to review), and a person's marks on items a plugin +builds from the box's own signals, such as a snooze or "resolved". The +plugin's `storage` is this computer's alone, so a snooze kept there is news +to the person's other laptop; kept here, every app paired with the box +reads it. ("Attention" is a working name: a box's inbox is already the +prompts it holds for an agent.) + +An item is named by its `key`. Posting to a key replaces the item there and +counts its `revision` on, so an agent replaces its own ask rather than +adding another; a post with a `session` and no key is that session's own, +`session:`. An item is `open`, `answered` or `closed`; the last two are +settled. An open item about a session closes by itself, with +`closed_reason: "session_ended"`, when the session ends, and a settled item +is forgotten 14 days later. + +| Call | What it does | +| --- | --- | +| `list(box, { session?, open? })` | `AttentionItem[]`, the most recently changed first: `{ key, revision, kind, title?, details?, options?, recommended?, status, answer?, closed_reason?, snoozed_until?, session?, by, created, updated, settled? }`. `by` is where the last change came from (`app`, `phone`, an agent's tool, `berth`) | +| `get(box, key)` | One item, or `undefined` when there is none at that key | +| `post(box, { key?, kind, title, details?, options?, recommended?, session?, revision? })` | Posts an ask, or replaces the one at its key: it opens again, its answer and snooze gone. `kind` is a short lowercase word of the poster's (`mark` is the box's own), `options` up to 12 `{ label, description? }` and `recommended` one of their labels. A `session` must be running (404 otherwise) | +| `update(box, key, { revision?, status?, answer?, closed_reason?, snoozed_until?, session? })` | Answers, closes, reopens or snoozes an item; what it leaves out stays. An `answer` alone makes it `answered`, a `closed_reason` alone closes it (closed without one is `resolved`), and `status: "open"` reopens it. `snoozed_until` is an ISO time, or `""` to end a snooze. At a key with no item it makes a mark (`kind: "mark"`), about `session` when given one | + +With `revision`, a post or update is refused (409) if the item has changed +since that revision, so you never overwrite a change you haven't seen. A +`before:attention.post` or `before:attention.update` [gate](/reference/events#gates) +can refuse one (403 `refused`). Each change is announced as +`attention.posted`, `attention.updated` or `attention.settled`, with the +key, revision and status and never the item's words: listen with +`berth.on("attention.*", …)` and read the item again. + +A box keeps up to 1000 items, the longest-settled going first to make +room. On a box whose `berthd` is from before this, every call rejects +(501 or 404): check that `info(box).capabilities` has `"attention"`. An +older Shipyard has no `berth.attention`: check before calling. The box +API is `GET /v1/attention?session=&open=1`, `POST /v1/attention`, and +`GET` and `PATCH /v1/attention/{key}` with the key path-escaped. + +```ts +// A snooze that follows the person between computers. +await berth.attention.update("devl", "devl/shop-claude/wait", { snoozed_until: new Date(Date.now() + 3_600_000).toISOString() }); +``` + ## React hooks From `@berth/plugin`, backed by the app's live state: diff --git a/internal/box/api.go b/internal/box/api.go index 2efa67ec8..763f5d42d 100644 --- a/internal/box/api.go +++ b/internal/box/api.go @@ -130,6 +130,9 @@ type Box struct { // Helpers runs the helper programs plugins bring (helpers.go); nil on a // box without them. Helpers *Helpers + // Attention keeps what needs a person, for every app paired with the + // box (attention.go); nil on a box without it. + Attention *AttentionStore } func (b *Box) own(path string) { @@ -279,6 +282,7 @@ func (b *Box) Mount(s *wire.Server) { b.mountBrowser(route) b.mountNotify(route) b.mountArtifacts(route) + b.mountAttention(route) b.mountHelpers(s, route) b.mountPairing(s, route) } @@ -668,6 +672,7 @@ func (b *Box) stopSessionsIn(from, dir string) { b.Events.Publish(events.Event{Type: "session.stopped", Box: b.Name, Origin: from, Data: map[string]any{"name": s.Name, "path": s.Dir}}) } } + b.attentionSessionsEnded(ctx) } func (b *Box) listSessions(w http.ResponseWriter, r *http.Request) error { @@ -924,6 +929,7 @@ func (b *Box) removeSession(w http.ResponseWriter, r *http.Request) error { return err } b.publish(r, "session.stopped", map[string]any{"name": name}) + b.attentionSessionsEnded(r.Context()) writeJSON(w, map[string]string{"removed": name}) return nil } diff --git a/internal/box/attention.go b/internal/box/attention.go new file mode 100644 index 000000000..887878b70 --- /dev/null +++ b/internal/box/attention.go @@ -0,0 +1,475 @@ +package box + +import ( + "encoding/json" + "errors" + "fmt" + "net/http" + "os" + "regexp" + "sort" + "strings" + "sync" + "time" + "unicode" + "unicode/utf8" + + "github.com/cosscom/shipyard/internal/statefile" +) + +// Attention items are what needs a person, kept on the box so every app +// and computer paired with it reads the same list: an ask an agent posts +// (a decision, an action, something to review), and a person's marks on +// items a plugin builds from the box's own signals (a waiting agent, a run's +// gate), such as a snooze or "resolved". Plugin storage is per computer, so +// without this a snooze on one laptop is news to the other. +// +// An item is named by its key. Posting to a key replaces what is there and +// counts its revision on, so an agent replaces its own ask instead of adding +// another; an agent's post without a key is keyed by its session. An item +// settles when it is answered or closed, or by itself when its session ends, +// and a settled item is forgotten 14 days later. +// +// "Attention" is a working name: the box's inbox already means prompts held +// for an agent (inbox.go). + +// AttentionItem is one item. +type AttentionItem struct { + Key string `json:"key"` + // Revision counts every change from 1, so a client can tell an item + // changed, and refuse to overwrite a change it hasn't seen. + Revision int `json:"revision"` + // Kind is what it is ("decision", "action", "review", "blocked", …, the + // poster's own word), or "mark" for a person's mark on an item built + // from signals. + Kind string `json:"kind"` + Title string `json:"title,omitempty"` + Details string `json:"details,omitempty"` + Options []AttentionOption `json:"options,omitempty"` + Recommended string `json:"recommended,omitempty"` + // Status is open, answered or closed; the last two are settled. + Status string `json:"status"` + Answer string `json:"answer,omitempty"` + ClosedReason string `json:"closed_reason,omitempty"` + SnoozedUntil time.Time `json:"snoozed_until,omitzero"` + // Session, when set, is the session it is about: it settles when that + // session ends. + Session string `json:"session,omitempty"` + // By is the origin of the last change ("app", "phone", an agent's tool, + // "berth" for the box itself). + By string `json:"by,omitempty"` + Created time.Time `json:"created"` + Updated time.Time `json:"updated"` + Settled time.Time `json:"settled,omitzero"` +} + +// AttentionOption is one of the answers an ask offers. +type AttentionOption struct { + Label string `json:"label"` + Description string `json:"description,omitempty"` +} + +// AttentionPost is a new ask, or one that replaces the ask at its key. +type AttentionPost struct { + Key string `json:"key,omitempty"` + Kind string `json:"kind"` + Title string `json:"title"` + Details string `json:"details,omitempty"` + Options []AttentionOption `json:"options,omitempty"` + Recommended string `json:"recommended,omitempty"` + Session string `json:"session,omitempty"` + // Revision, when set, is the revision the poster last read: the post is + // refused if the item has changed since (0 replaces whatever is there). + Revision int `json:"revision,omitempty"` +} + +// AttentionUpdate changes an item; what it leaves out stays as it is. An +// update to a key with no item makes a mark there. +type AttentionUpdate struct { + Revision int `json:"revision,omitempty"` + Status *string `json:"status,omitempty"` + Answer *string `json:"answer,omitempty"` + ClosedReason *string `json:"closed_reason,omitempty"` + // SnoozedUntil is an RFC 3339 time; "" ends a snooze. + SnoozedUntil *string `json:"snoozed_until,omitempty"` + // Session is the session a new mark is about. It is ignored on an item + // that exists. + Session string `json:"session,omitempty"` +} + +const ( + AttentionOpen = "open" + AttentionAnswered = "answered" + AttentionClosed = "closed" + + // attentionMark is the kind of a mark on an item the box doesn't hold. + attentionMark = "mark" + // attentionSessionEnded is the closed_reason of an item whose session + // ended while it was open. + attentionSessionEnded = "session_ended" +) + +// AttentionKeep is how long a settled item is kept. +const AttentionKeep = 14 * 24 * time.Hour + +const ( + maxAttentionItems = 1000 + maxAttentionKey = 300 + maxAttentionTitle = 300 + maxAttentionText = 16 << 10 + maxAttentionOptions = 12 + maxAttentionLabel = 200 + maxAttentionOptDesc = 1000 + maxAttentionSession = 200 +) + +var ( + errAttentionRevision = errors.New("the item changed since that revision; read it again") + errAttentionFull = fmt.Errorf("the box already holds %d open attention items; settle some first", maxAttentionItems) + attentionWord = regexp.MustCompile(`^[a-z][a-z0-9_-]{0,31}$`) +) + +// AttentionStore keeps the box's items in one state file. +type AttentionStore struct { + Path string + // Now is the clock; nil is time.Now. + Now func() time.Time + + mu sync.Mutex + items map[string]*AttentionItem + loaded bool +} + +func (s *AttentionStore) now() time.Time { + if s.Now != nil { + return s.Now().UTC() + } + return time.Now().UTC() +} + +// load reads the file once. It holds s.mu. An item kept past its time is +// left behind as it is read. +func (s *AttentionStore) load() { + if s.loaded { + return + } + s.loaded = true + s.items = map[string]*AttentionItem{} + b, err := os.ReadFile(s.Path) + if err != nil { + return + } + var all []AttentionItem + if json.Unmarshal(b, &all) != nil { + return + } + now := s.now() + for i := range all { + it := all[i] + if it.Key == "" || expired(it, now) { + continue + } + s.items[it.Key] = &it + } +} + +func (s *AttentionStore) save() error { + all := make([]AttentionItem, 0, len(s.items)) + for _, it := range s.items { + all = append(all, *it) + } + sort.Slice(all, func(i, j int) bool { return all[i].Key < all[j].Key }) + b, err := json.Marshal(all) + if err != nil { + return err + } + return statefile.Write(s.Path, b) +} + +// expired says an item is over and can be forgotten: settled more than +// AttentionKeep ago, or an open mark nobody has touched in that long whose +// snooze, if any, has run out. +func expired(it AttentionItem, now time.Time) bool { + cut := now.Add(-AttentionKeep) + if it.Status != AttentionOpen { + return !it.Settled.IsZero() && it.Settled.Before(cut) + } + return it.Kind == attentionMark && it.Session == "" && it.Updated.Before(cut) && !it.SnoozedUntil.After(now) +} + +// AttentionFilter narrows List. +type AttentionFilter struct { + Session string + // Open leaves settled items out. + Open bool +} + +// List returns the items, the most recently changed first. +func (s *AttentionStore) List(f AttentionFilter) []AttentionItem { + s.mu.Lock() + defer s.mu.Unlock() + s.load() + out := []AttentionItem{} + for _, it := range s.items { + if f.Session != "" && it.Session != f.Session || f.Open && it.Status != AttentionOpen { + continue + } + out = append(out, cloneAttention(*it)) + } + sort.Slice(out, func(i, j int) bool { + if !out[i].Updated.Equal(out[j].Updated) { + return out[i].Updated.After(out[j].Updated) + } + return out[i].Key < out[j].Key + }) + return out +} + +// Get returns one item. +func (s *AttentionStore) Get(key string) (AttentionItem, bool) { + s.mu.Lock() + defer s.mu.Unlock() + s.load() + it, ok := s.items[key] + if !ok { + return AttentionItem{}, false + } + return cloneAttention(*it), true +} + +// SessionKey is the key of a session's own ask. +func SessionKey(session string) string { return "session:" + session } + +// Post puts an ask at its key, replacing the one there: it opens again with +// a new revision, and its answer and snooze are gone. +func (s *AttentionStore) Post(p AttentionPost, by string) (AttentionItem, error) { + if p.Key == "" && p.Session != "" { + p.Key = SessionKey(p.Session) + } + if err := checkAttentionPost(p); err != nil { + return AttentionItem{}, err + } + s.mu.Lock() + defer s.mu.Unlock() + s.load() + now := s.now() + was, ok := s.items[p.Key] + if p.Revision != 0 && (!ok || was.Revision != p.Revision) { + return AttentionItem{}, httpError{http.StatusConflict, errAttentionRevision.Error()} + } + if !ok { + if err := s.makeRoom(); err != nil { + return AttentionItem{}, err + } + } + it := &AttentionItem{ + Key: p.Key, Revision: 1, Kind: p.Kind, Title: p.Title, Details: p.Details, Options: p.Options, + Recommended: p.Recommended, Status: AttentionOpen, Session: p.Session, By: by, Created: now, Updated: now, + } + if ok { + it.Revision, it.Created = was.Revision+1, was.Created + } + s.items[p.Key] = it + if err := s.save(); err != nil { + if ok { + s.items[p.Key] = was + } else { + delete(s.items, p.Key) + } + return AttentionItem{}, err + } + return cloneAttention(*it), nil +} + +// Update changes the item at key, or makes a mark there when there is +// none. made says it made one. +func (s *AttentionStore) Update(key string, u AttentionUpdate, by string) (it AttentionItem, made bool, err error) { + if err := checkAttentionKey(key); err != nil { + return AttentionItem{}, false, err + } + var until time.Time + if u.SnoozedUntil != nil && *u.SnoozedUntil != "" { + t, err := time.Parse(time.RFC3339, *u.SnoozedUntil) + if err != nil { + return AttentionItem{}, false, badRequest("snoozed_until must be an RFC 3339 time, or \"\" to end the snooze") + } + until = t.UTC() + } + if u.Status != nil && *u.Status != AttentionOpen && *u.Status != AttentionAnswered && *u.Status != AttentionClosed { + return AttentionItem{}, false, badRequest("status must be open, answered or closed") + } + if u.Answer != nil && len(*u.Answer) > maxAttentionText { + return AttentionItem{}, false, badRequest("an answer may be %d KB", maxAttentionText>>10) + } + if u.ClosedReason != nil && *u.ClosedReason != "" && !attentionWord.MatchString(*u.ClosedReason) { + return AttentionItem{}, false, badRequest("closed_reason is a short word: lowercase letters, digits, - and _") + } + if len(u.Session) > maxAttentionSession { + return AttentionItem{}, false, badRequest("that isn't a session name") + } + s.mu.Lock() + defer s.mu.Unlock() + s.load() + now := s.now() + was, ok := s.items[key] + if u.Revision != 0 && (!ok || was.Revision != u.Revision) { + return AttentionItem{}, false, httpError{http.StatusConflict, errAttentionRevision.Error()} + } + var next AttentionItem + if ok { + next = cloneAttention(*was) + } else { + if err := s.makeRoom(); err != nil { + return AttentionItem{}, false, err + } + next = AttentionItem{Key: key, Kind: attentionMark, Status: AttentionOpen, Session: u.Session, Created: now} + } + status := next.Status + switch { + case u.Status != nil: + status = *u.Status + case u.Answer != nil && *u.Answer != "": + status = AttentionAnswered + case u.ClosedReason != nil && *u.ClosedReason != "": + status = AttentionClosed + } + if status != next.Status { + next.Status = status + next.Settled = time.Time{} + if status != AttentionOpen { + next.Settled = now + } + } + if u.Answer != nil { + next.Answer = *u.Answer + } + if u.ClosedReason != nil { + next.ClosedReason = *u.ClosedReason + } + switch next.Status { + case AttentionOpen: + next.Answer, next.ClosedReason = "", "" + case AttentionAnswered: + next.ClosedReason = "" + case AttentionClosed: + if next.ClosedReason == "" { + next.ClosedReason = "resolved" + } + } + if u.SnoozedUntil != nil { + next.SnoozedUntil = until + } + next.Revision++ + next.By, next.Updated = by, now + s.items[key] = &next + if err := s.save(); err != nil { + if ok { + s.items[key] = was + } else { + delete(s.items, key) + } + return AttentionItem{}, false, err + } + return cloneAttention(next), !ok, nil +} + +// makeRoom keeps the store to maxAttentionItems before a new key: the +// longest-settled items go first, and with none settled the new one is +// refused. It holds s.mu. +func (s *AttentionStore) makeRoom() error { + if len(s.items) < maxAttentionItems { + return nil + } + var oldest *AttentionItem + for _, it := range s.items { + if it.Status != AttentionOpen && (oldest == nil || it.Settled.Before(oldest.Settled)) { + oldest = it + } + } + if oldest == nil { + return httpError{http.StatusTooManyRequests, errAttentionFull.Error()} + } + delete(s.items, oldest.Key) + return nil +} + +// Sweep settles the open items whose session is not in live, and forgets +// items past their time. It returns the items it settled. +func (s *AttentionStore) Sweep(live map[string]bool) []AttentionItem { + s.mu.Lock() + defer s.mu.Unlock() + s.load() + now := s.now() + var settled []AttentionItem + changed := false + for key, it := range s.items { + if it.Status == AttentionOpen && it.Session != "" && !live[it.Session] { + it.Status, it.ClosedReason, it.Settled = AttentionClosed, attentionSessionEnded, now + it.Revision++ + it.By, it.Updated = "berth", now + settled = append(settled, cloneAttention(*it)) + changed = true + } + if expired(*it, now) { + delete(s.items, key) + changed = true + } + } + if changed { + _ = s.save() + } + sort.Slice(settled, func(i, j int) bool { return settled[i].Key < settled[j].Key }) + return settled +} + +func cloneAttention(it AttentionItem) AttentionItem { + if it.Options != nil { + it.Options = append([]AttentionOption(nil), it.Options...) + } + return it +} + +func checkAttentionKey(key string) error { + if key == "" || len(key) > maxAttentionKey || strings.TrimSpace(key) != key || strings.IndexFunc(key, unicode.IsControl) >= 0 || !utf8.ValidString(key) { + return badRequest("a key is 1 to %d characters, with no control characters or surrounding spaces", maxAttentionKey) + } + return nil +} + +func checkAttentionPost(p AttentionPost) error { + if p.Key == "" { + return badRequest("a post needs a key or a session") + } + if err := checkAttentionKey(p.Key); err != nil { + return err + } + if !attentionWord.MatchString(p.Kind) || p.Kind == attentionMark { + return badRequest("kind is a short word (decision, action, review, blocked, …): lowercase letters, digits, - and _; mark is the box's own") + } + if strings.TrimSpace(p.Title) == "" || utf8.RuneCountInString(p.Title) > maxAttentionTitle { + return badRequest("a post needs a title of at most %d characters", maxAttentionTitle) + } + if len(p.Details) > maxAttentionText { + return badRequest("details may be %d KB", maxAttentionText>>10) + } + if len(p.Session) > maxAttentionSession { + return badRequest("that isn't a session name") + } + if len(p.Options) > maxAttentionOptions { + return badRequest("an ask offers at most %d options", maxAttentionOptions) + } + seen := map[string]bool{} + for _, o := range p.Options { + if strings.TrimSpace(o.Label) == "" || utf8.RuneCountInString(o.Label) > maxAttentionLabel || utf8.RuneCountInString(o.Description) > maxAttentionOptDesc { + return badRequest("each option needs a label of at most %d characters, and a description of at most %d", maxAttentionLabel, maxAttentionOptDesc) + } + if seen[o.Label] { + return badRequest("two options are labelled %q", o.Label) + } + seen[o.Label] = true + } + if p.Recommended != "" && !seen[p.Recommended] { + return badRequest("recommended must be one of the options' labels") + } + return nil +} diff --git a/internal/box/attention_test.go b/internal/box/attention_test.go new file mode 100644 index 000000000..1481a3e65 --- /dev/null +++ b/internal/box/attention_test.go @@ -0,0 +1,263 @@ +package box + +import ( + "context" + "net/http" + "net/url" + "os" + "path/filepath" + "strings" + "testing" + "time" + + "github.com/cosscom/shipyard/internal/events" + "github.com/cosscom/shipyard/internal/hooks" +) + +type attentionClock struct{ t time.Time } + +func (c *attentionClock) now() time.Time { return c.t } +func (c *attentionClock) add(d time.Duration) { c.t = c.t.Add(d) } +func newAttention(t *testing.T) (*AttentionStore, *attentionClock) { + clock := &attentionClock{time.Date(2026, 11, 6, 9, 0, 0, 0, time.UTC)} + return &AttentionStore{Path: filepath.Join(t.TempDir(), "attention.json"), Now: clock.now}, clock +} + +func TestAnAgentReplacesItsOwnAskByKey(t *testing.T) { + s, clock := newAttention(t) + first, err := s.Post(AttentionPost{Session: "shop-claude", Kind: "decision", Title: "Which queue?", Options: []AttentionOption{{Label: "SQS"}, {Label: "Redis"}}, Recommended: "SQS"}, "claude") + if err != nil { + t.Fatal(err) + } + if first.Key != "session:shop-claude" || first.Revision != 1 || first.Status != AttentionOpen || first.By != "claude" { + t.Fatalf("first post = %+v", first) + } + // Someone snoozes it, then the agent asks again: the same item, a new + // revision, open and no longer snoozed. + if _, _, err := s.Update(first.Key, AttentionUpdate{SnoozedUntil: strp("2026-11-06T12:00:00Z")}, "app"); err != nil { + t.Fatal(err) + } + clock.add(time.Minute) + second, err := s.Post(AttentionPost{Session: "shop-claude", Kind: "decision", Title: "Which queue, now that Redis is out?"}, "claude") + if err != nil { + t.Fatal(err) + } + if second.Revision != 3 || second.Title != "Which queue, now that Redis is out?" || !second.SnoozedUntil.IsZero() || len(second.Options) != 0 || !second.Created.Equal(first.Created) || !second.Updated.After(first.Updated) { + t.Fatalf("replaced = %+v", second) + } + if all := s.List(AttentionFilter{}); len(all) != 1 { + t.Fatalf("a replaced ask is still one item: %+v", all) + } + // A post or an update naming a revision that has gone by is refused, + // so a client never overwrites what it hasn't read. + if _, err := s.Post(AttentionPost{Session: "shop-claude", Kind: "decision", Title: "stale", Revision: 2}, "claude"); statusFor(err) != http.StatusConflict { + t.Fatalf("stale post: %v", err) + } + if _, _, err := s.Update(second.Key, AttentionUpdate{Revision: 1, Status: strp(AttentionClosed)}, "app"); statusFor(err) != http.StatusConflict { + t.Fatalf("stale update: %v", err) + } + answered, _, err := s.Update(second.Key, AttentionUpdate{Revision: 3, Answer: strp("SQS")}, "phone") + if err != nil { + t.Fatal(err) + } + if answered.Status != AttentionAnswered || answered.Answer != "SQS" || answered.Revision != 4 || !answered.Settled.Equal(clock.t) || answered.By != "phone" { + t.Fatalf("answered = %+v", answered) + } + if open := s.List(AttentionFilter{Open: true}); len(open) != 0 { + t.Fatalf("an answered ask is settled: %+v", open) + } +} + +func TestAMarkOnAnItemTheBoxDoesntHold(t *testing.T) { + s, _ := newAttention(t) + key := "run/r_12/review" // a plugin's key for a run's gate, slashes and all + m, made, err := s.Update(key, AttentionUpdate{SnoozedUntil: strp("2026-11-07T08:00:00+01:00")}, "app") + if err != nil { + t.Fatal(err) + } + if !made || m.Kind != "mark" || m.Status != AttentionOpen || m.Revision != 1 || !m.SnoozedUntil.Equal(time.Date(2026, 11, 7, 7, 0, 0, 0, time.UTC)) { + t.Fatalf("mark = %+v made=%v", m, made) + } + r, made, err := s.Update(key, AttentionUpdate{Status: strp(AttentionClosed)}, "app") + if err != nil { + t.Fatal(err) + } + if made || r.Status != AttentionClosed || r.ClosedReason != "resolved" || r.Revision != 2 { + t.Fatalf("resolved = %+v", r) + } + // Reopening clears what settled it; "" ends the snooze. + o, _, _ := s.Update(key, AttentionUpdate{Status: strp(AttentionOpen), SnoozedUntil: strp("")}, "app") + if o.Status != AttentionOpen || o.ClosedReason != "" || !o.Settled.IsZero() || !o.SnoozedUntil.IsZero() { + t.Fatalf("reopened = %+v", o) + } + // A post may not pretend to be a mark, and words stay bounded. + for _, p := range []AttentionPost{ + {Key: "x", Kind: "mark", Title: "t"}, + {Key: "x", Kind: "decision"}, + {Kind: "decision", Title: "no key"}, + {Key: " x", Kind: "decision", Title: "t"}, + {Key: "x", Kind: "decision", Title: "t", Recommended: "Maybe", Options: []AttentionOption{{Label: "Yes"}}}, + {Key: "x", Kind: "decision", Title: "t", Details: strings.Repeat("a", maxAttentionText+1)}, + } { + if _, err := s.Post(p, "x"); statusFor(err) != http.StatusBadRequest { + t.Fatalf("%+v: %v", p, err) + } + } + if _, _, err := s.Update(key, AttentionUpdate{SnoozedUntil: strp("tomorrow")}, "app"); statusFor(err) != http.StatusBadRequest { + t.Fatalf("a bad time: %v", err) + } +} + +func TestItemsSettleWhenTheirSessionEndsAndAgeOut(t *testing.T) { + s, clock := newAttention(t) + s.Post(AttentionPost{Session: "a", Kind: "decision", Title: "A?"}, "claude") + s.Post(AttentionPost{Session: "b", Kind: "action", Title: "Run the migration"}, "claude") + s.Update("standalone", AttentionUpdate{SnoozedUntil: strp("2026-11-30T00:00:00Z")}, "app") + s.Update("session:b-done", AttentionUpdate{Session: "b", Status: strp(AttentionClosed)}, "app") + + // a ends; b still runs. + settled := s.Sweep(map[string]bool{"b": true}) + if len(settled) != 1 || settled[0].Key != "session:a" || settled[0].Status != AttentionClosed || settled[0].ClosedReason != "session_ended" || settled[0].By != "berth" || settled[0].Revision != 2 { + t.Fatalf("settled = %+v", settled) + } + // Sweeping again settles nothing new; an item settled already isn't + // settled twice. + if again := s.Sweep(map[string]bool{}); len(again) != 1 || again[0].Key != "session:b" { + t.Fatalf("second sweep = %+v", again) + } + if again := s.Sweep(map[string]bool{}); len(again) != 0 { + t.Fatalf("third sweep = %+v", again) + } + if n := len(s.List(AttentionFilter{})); n != 4 { + t.Fatalf("settled items are kept for a while: %d", n) + } + // 14 days on, the settled ones are gone. The snoozed mark stays until + // its snooze has run out as well. + clock.add(AttentionKeep + time.Hour) + s.Sweep(nil) + if all := s.List(AttentionFilter{}); len(all) != 1 || all[0].Key != "standalone" { + t.Fatalf("after 14 days: %+v", all) + } + clock.add(30 * 24 * time.Hour) + s.Sweep(nil) + if all := s.List(AttentionFilter{}); len(all) != 0 { + t.Fatalf("an old mark whose snooze ran out: %+v", all) + } +} + +func TestAttentionSurvivesARestart(t *testing.T) { + s, clock := newAttention(t) + s.Post(AttentionPost{Key: "review/shop#12", Kind: "review", Title: "PR #12 is green", Details: "Ready to merge.", Options: []AttentionOption{{Label: "Merge", Description: "squash"}}}, "claude") + s.Update("session:gone", AttentionUpdate{Status: strp(AttentionClosed)}, "app") + s.Update("session:old", AttentionUpdate{Status: strp(AttentionClosed)}, "app") + + // A new store on the same file, as berthd has after a restart. + again := &AttentionStore{Path: s.Path, Now: clock.now} + it, ok := again.Get("review/shop#12") + if !ok || it.Title != "PR #12 is green" || it.Details != "Ready to merge." || it.Options[0].Description != "squash" || it.Revision != 1 { + t.Fatalf("after restart: %+v %v", it, ok) + } + if n := len(again.List(AttentionFilter{})); n != 3 { + t.Fatalf("%d items after restart", n) + } + // One read 15 days on leaves the settled ones behind. + clock.add(15 * 24 * time.Hour) + later := &AttentionStore{Path: s.Path, Now: clock.now} + if all := later.List(AttentionFilter{}); len(all) != 1 || all[0].Key != "review/shop#12" { + t.Fatalf("read 15 days on: %+v", all) + } +} + +func TestAttentionKeepsToItsSize(t *testing.T) { + s, _ := newAttention(t) + s.Update("settled", AttentionUpdate{Status: strp(AttentionClosed)}, "app") + for i := 1; i < maxAttentionItems; i++ { + s.items["k"+itoa(i)] = &AttentionItem{Key: "k" + itoa(i), Kind: "decision", Status: AttentionOpen} + } + // Full: the settled one makes room, then there is none. + if _, err := s.Post(AttentionPost{Key: "new", Kind: "decision", Title: "t"}, "x"); err != nil { + t.Fatal(err) + } + if _, ok := s.Get("settled"); ok { + t.Fatal("the settled item should have made room") + } + if _, err := s.Post(AttentionPost{Key: "one-more", Kind: "decision", Title: "t"}, "x"); statusFor(err) != http.StatusTooManyRequests { + t.Fatalf("full: %v", err) + } + // Replacing an item that's there still works. + if _, err := s.Post(AttentionPost{Key: "new", Kind: "decision", Title: "t2"}, "x"); err != nil { + t.Fatal(err) + } +} + +func TestAttentionRoutes(t *testing.T) { + dir := t.TempDir() + cfg := filepath.Join(dir, "hooks.json") + os.WriteFile(cfg, []byte(`{"hooks":[{"on":"before:attention.update","run":"grep -q never-snooze && { echo no snoozing that; exit 1; }; exit 0"}]}`), 0o600) + bus := &events.Bus{Sequence: true} + bx := &Box{Name: "devbox", Events: bus, Hooks: &hooks.Runner{Path: cfg}} + bx.Attention = &AttentionStore{Path: filepath.Join(dir, "attention.json")} + c := localBox(t, bx) + ctx := context.Background() + evs, stop := bus.Subscribe() + defer stop() + if !contains(bx.Capabilities(), "attention") { + t.Fatal("no attention capability") + } + + var it AttentionItem + if err := c.Call(ctx, "POST", "/v1/attention", AttentionPost{Key: "acme/shop/ask", Kind: "decision", Title: "Ship on Friday?", Details: "secret plans"}, &it); err != nil { + t.Fatal(err) + } + e := waitEventType(t, evs, "attention.posted") + if e.Data["key"] != "acme/shop/ask" || e.Data["revision"] != 1 || e.Data["status"] != "open" { + t.Fatalf("posted event %+v", e) + } + if _, ok := e.Data["title"]; ok { + t.Fatalf("an event never carries an item's words: %+v", e.Data) + } + // The key goes path-escaped, slashes and all. + path := "/v1/attention/" + url.PathEscape("acme/shop/ask") + var snoozed AttentionItem + if err := c.Call(ctx, "PATCH", path, map[string]any{"revision": 1, "snoozed_until": "2026-11-07T09:00:00Z"}, &snoozed); err != nil { + t.Fatal(err) + } + if snoozed.Revision != 2 || snoozed.SnoozedUntil.IsZero() { + t.Fatalf("snoozed = %+v", snoozed) + } + if e := waitEventType(t, evs, "attention.updated"); e.Data["snoozed_until"] == nil { + t.Fatalf("updated event %+v", e) + } + var got AttentionItem + if err := c.Call(ctx, "GET", path, nil, &got); err != nil || got.Revision != 2 || got.Title != "Ship on Friday?" { + t.Fatalf("get = %+v %v", got, err) + } + // A turn's "#" too. + if err := c.Call(ctx, "PATCH", "/v1/attention/"+url.PathEscape("devl/shop#3"), map[string]any{"status": "closed"}, &got); err != nil || got.Key != "devl/shop#3" || got.Kind != "mark" { + t.Fatalf("mark with #: %+v %v", got, err) + } + var list []AttentionItem + if err := c.Call(ctx, "GET", "/v1/attention?open=1", nil, &list); err != nil || len(list) != 1 { + t.Fatalf("list = %+v %v", list, err) + } + // Refusals: a stale revision, an unknown key to read, a gate. + if err := c.Call(ctx, "PATCH", path, map[string]any{"revision": 1, "status": "closed"}, nil); err == nil || !strings.Contains(err.Error(), "changed since") { + t.Fatalf("stale: %v", err) + } + if err := c.Call(ctx, "GET", "/v1/attention/nope", nil, nil); err == nil || !strings.Contains(err.Error(), "no attention item") { + t.Fatalf("unknown: %v", err) + } + if err := c.Call(ctx, "PATCH", "/v1/attention/never-snooze", map[string]any{"snoozed_until": "2026-11-07T09:00:00Z"}, nil); err == nil || !strings.Contains(err.Error(), "no snoozing that") { + t.Fatalf("gate: %v", err) + } + if _, ok := bx.Attention.Get("never-snooze"); ok { + t.Fatal("a refused update made a mark") + } + + // A session that ends settles its items, announced as the box's own. + bx.Attention.Post(AttentionPost{Session: "shop-claude", Kind: "action", Title: "Log in to Fly"}, "claude") + bx.sweepAttention([]Session{{Name: "shop-claude", Exited: true}}) + if e := waitEventType(t, evs, "attention.settled"); e.Data["key"] != "session:shop-claude" || e.Data["closed_reason"] != "session_ended" || e.Origin != "berth" { + t.Fatalf("settled event %+v", e) + } +} diff --git a/internal/box/attentionapi.go b/internal/box/attentionapi.go new file mode 100644 index 000000000..a5b6d30d4 --- /dev/null +++ b/internal/box/attentionapi.go @@ -0,0 +1,156 @@ +package box + +import ( + "context" + "errors" + "net/http" + + "github.com/cosscom/shipyard/internal/events" +) + +// The attention API (attention.go): +// +// GET /v1/attention?session=S&open=1 the items, the most recently changed first +// POST /v1/attention post an ask, or replace the one at its key +// GET /v1/attention/{key} one item +// PATCH /v1/attention/{key} answer, close, reopen or snooze one; a key +// with no item gets a mark +// +// A key is path-escaped (encodeURIComponent), so it may hold slashes. +// Events say what changed and never an item's words: attention.posted, +// attention.updated and attention.settled carry key, revision, kind, +// status, session and, when set, closed_reason and snoozed_until. + +func (b *Box) mountAttention(route func(string, func(http.ResponseWriter, *http.Request) error)) { + route("GET /v1/attention", b.listAttention) + route("POST /v1/attention", b.postAttention) + route("GET /v1/attention/{key...}", b.getAttention) + route("PATCH /v1/attention/{key...}", b.updateAttention) +} + +var errNoAttention = httpError{http.StatusNotImplemented, "this box keeps no attention items"} + +func (b *Box) listAttention(w http.ResponseWriter, r *http.Request) error { + if b.Attention == nil { + return errNoAttention + } + q := r.URL.Query() + writeJSON(w, b.Attention.List(AttentionFilter{Session: q.Get("session"), Open: q.Get("open") == "1"})) + return nil +} + +func (b *Box) getAttention(w http.ResponseWriter, r *http.Request) error { + if b.Attention == nil { + return errNoAttention + } + it, ok := b.Attention.Get(r.PathValue("key")) + if !ok { + return httpError{http.StatusNotFound, "no attention item with that key"} + } + writeJSON(w, it) + return nil +} + +func (b *Box) postAttention(w http.ResponseWriter, r *http.Request) error { + if b.Attention == nil { + return errNoAttention + } + var p AttentionPost + if err := decodeLimit(r, &p, 4*maxAttentionText); err != nil { + return err + } + // An ask about a session settles when it ends, so it must be one that + // runs. + if p.Session != "" && b.Sessions != nil { + if _, err := b.Sessions.Get(r.Context(), p.Session); err != nil { + if errors.Is(err, ErrUnknownSession) { + return httpError{http.StatusNotFound, "no session named " + p.Session + " on this box"} + } + return err + } + } + key := p.Key + if key == "" && p.Session != "" { + key = SessionKey(p.Session) + } + if err := b.before(r, "attention.post", map[string]any{"key": key, "kind": p.Kind, "session": p.Session}); err != nil { + return err + } + it, err := b.Attention.Post(p, origin(r)) + if err != nil { + return err + } + b.publish(r, "attention.posted", attentionEventData(it)) + writeJSON(w, it) + return nil +} + +func (b *Box) updateAttention(w http.ResponseWriter, r *http.Request) error { + if b.Attention == nil { + return errNoAttention + } + key := r.PathValue("key") + var u AttentionUpdate + if err := decodeLimit(r, &u, 2*maxAttentionText); err != nil { + return err + } + gate := map[string]any{"key": key} + if u.Status != nil { + gate["status"] = *u.Status + } + if u.SnoozedUntil != nil { + gate["snoozed_until"] = *u.SnoozedUntil + } + if err := b.before(r, "attention.update", gate); err != nil { + return err + } + it, _, err := b.Attention.Update(key, u, origin(r)) + if err != nil { + return err + } + b.publish(r, "attention.updated", attentionEventData(it)) + writeJSON(w, it) + return nil +} + +// sweepAttention settles the items of sessions that ended, given the +// sessions there are, and forgets items past their time. +func (b *Box) sweepAttention(all []Session) { + if b.Attention == nil { + return + } + live := make(map[string]bool, len(all)) + for _, s := range all { + if !s.Exited { + live[s.Name] = true + } + } + for _, it := range b.Attention.Sweep(live) { + b.Events.Publish(events.Event{Type: "attention.settled", Box: b.Name, Origin: "berth", Data: attentionEventData(it)}) + } +} + +// attentionSessionsEnded sweeps at once after sessions were stopped, rather +// than at the next refresh. +func (b *Box) attentionSessionsEnded(ctx context.Context) { + if b.Attention == nil || b.Sessions == nil { + return + } + if all, err := b.Sessions.List(ctx); err == nil { + b.sweepAttention(all) + } +} + +func attentionEventData(it AttentionItem) map[string]any { + d := map[string]any{"key": it.Key, "revision": it.Revision, "kind": it.Kind, "status": it.Status} + if it.Session != "" { + d["session"] = it.Session + } + if it.ClosedReason != "" { + d["closed_reason"] = it.ClosedReason + } + if !it.SnoozedUntil.IsZero() { + d["snoozed_until"] = it.SnoozedUntil + } + return d +} diff --git a/internal/box/turns.go b/internal/box/turns.go index 6278999cf..375b9ae9d 100644 --- a/internal/box/turns.go +++ b/internal/box/turns.go @@ -1707,6 +1707,7 @@ func (b *Box) refreshTurns(ctx context.Context) { b.enrich(ctx, all) b.Turns.Prune(all) b.Turns.Expire(time.Now()) + b.sweepAttention(all) } // PendingExpiry is how long a sent prompt may stay pending, its agent idle, diff --git a/internal/box/upgrade.go b/internal/box/upgrade.go index e1f93c03a..ebe2a1664 100644 --- a/internal/box/upgrade.go +++ b/internal/box/upgrade.go @@ -229,6 +229,11 @@ func (b *Box) Capabilities() []string { // (/v1/helpers, helpers.go). caps = append(caps, "helpers") } + if b.Attention != nil { + // attention: GET/POST /v1/attention and GET/PATCH + // /v1/attention/{key}, what needs a person (attention.go). + caps = append(caps, "attention") + } return caps } diff --git a/packages/plugin-sdk/src/index.ts b/packages/plugin-sdk/src/index.ts index 798ee82ad..2ac5c31e2 100644 --- a/packages/plugin-sdk/src/index.ts +++ b/packages/plugin-sdk/src/index.ts @@ -1,6 +1,9 @@ import type { ComponentType } from "react"; import type { + AttentionItem, + AttentionPost, + AttentionUpdate, BerthEvent, BoxInfo, BoxStatus, @@ -517,6 +520,29 @@ export interface BerthHelper { attachEnv>(box: string, where: { location: string; worktree: string; env?: Record }): Promise>; } +// What needs a person on a box (see AttentionItem), the same for every app +// paired with it. A plugin that shows asks and snoozes them keeps them here +// rather than in its storage, which is this computer's alone. Every call +// rejects with code "unsupported" (status 501) or "not_found" (404) on a +// box whose berthd is from before this: check that info().capabilities has +// "attention". Absent in a Shipyard from before it: check before calling. +export interface BerthAttention { + // The items, the most recently changed first: one session's, or only the + // open ones. Settled items are kept for 14 days. + list(box: string, opts?: { session?: string; open?: boolean }): Promise; + // One item, or undefined when there is none at that key. + get(box: string, key: string): Promise; + // Posts an ask, replacing the one at its key: it opens again with a new + // revision, its answer and snooze gone. A session named must be running + // (404 otherwise). + post(box: string, item: AttentionPost): Promise; + // Answers, closes, reopens or snoozes an item; at a key with no item it + // makes a mark. Rejects with status 409 when revision is given and the + // item has changed since, and 403 "refused" when a before:attention.update + // hook says no. + update(box: string, key: string, change: AttentionUpdate): Promise; +} + // What a box found in a worktree's .env files for a helper (attachEnv). export interface HelperEnvLook> { plugin: string; @@ -556,6 +582,8 @@ export interface BerthPluginContext { // plugin's own id; one without "box.helper" in its manifest has none to // install. readonly helper: BerthHelper; + // What needs a person, kept on each box (see BerthAttention). + readonly attention: BerthAttention; addSidebarItem(item: SidebarItem): Dispose; addScreen(screen: Screen): Dispose; addWorktreePanel(panel: WorktreePanel): Dispose; diff --git a/packages/plugin-sdk/src/types.ts b/packages/plugin-sdk/src/types.ts index 88d7b5d40..3c2ccb21d 100644 --- a/packages/plugin-sdk/src/types.ts +++ b/packages/plugin-sdk/src/types.ts @@ -833,3 +833,70 @@ export interface RunRequest { group?: string; budget?: Run["budget"]; } + +// What needs a person, kept on the box (GET /v1/attention): an ask an agent +// posted, or a person's mark (kind "mark") on an item a plugin builds from +// the box's own signals. Every app paired with the box reads the same +// items, so a snooze on one computer holds on the others. +export type AttentionStatus = "open" | "answered" | "closed"; + +export interface AttentionOption { + label: string; + description?: string; +} + +export interface AttentionItem { + key: string; + // Counts every change from 1. + revision: number; + // The poster's word for it ("decision", "action", "review", "blocked"), + // or "mark". + kind: string; + title?: string; + details?: string; + options?: AttentionOption[]; + // The label of the option the poster recommends. + recommended?: string; + // answered and closed are settled. + status: AttentionStatus; + answer?: string; + // Why it closed: "resolved" unless someone said, "session_ended" when its + // session ended while it was open. + closed_reason?: string; + snoozed_until?: string; + // The session it is about: it settles when that session ends. + session?: string; + // The origin of the last change: "app", "phone", an agent's tool, or + // "berth" for the box itself. + by?: string; + created: string; + updated: string; + settled?: string; +} + +// A new ask, or one that replaces the ask at its key. Without a key it is +// the session's own ("session:"). +export interface AttentionPost { + key?: string; + kind: string; + title: string; + details?: string; + options?: AttentionOption[]; + recommended?: string; + session?: string; + // The revision last read: refused (409) if the item changed since. + revision?: number; +} + +// A change to an item; what it leaves out stays. snoozed_until is an ISO +// time, or "" to end a snooze. An answer alone makes it answered, a +// closed_reason alone closes it, and status "open" reopens it. +export interface AttentionUpdate { + revision?: number; + status?: AttentionStatus; + answer?: string; + closed_reason?: string; + snoozed_until?: string; + // The session a new mark is about. + session?: string; +} From f79aa9a04919d5c7ea70b4f154bb46e4d4b682d4 Mon Sep 17 00:00:00 2001 From: Stener1 Date: Fri, 6 Nov 2026 12:51:10 +0000 Subject: [PATCH 2/4] refactor(box): settle a stopped session's attention items from its event, and simplify the store Co-Authored-By: Claude Opus 5.5 --- .claude/skills/verify-berthd/SKILL.md | 2 +- .../skills/verify-berthd/features/README.md | 2 +- .../skills/verify-berthd/features/plugins.md | 2 + .../verify-berthd/scripts/attention-live.sh | 21 ++- .../scenarios/attention-two-laptops-live.mjs | 18 +- app/src/lib/attention.ts | 39 ++-- cmd/berthd/main.go | 2 + internal/box/api.go | 2 - internal/box/attention.go | 170 ++++++++++-------- internal/box/attention_test.go | 50 +++--- internal/box/attentionapi.go | 50 +++--- 11 files changed, 202 insertions(+), 156 deletions(-) mode change 100644 => 100755 .claude/skills/verify-berthd/scripts/attention-live.sh diff --git a/.claude/skills/verify-berthd/SKILL.md b/.claude/skills/verify-berthd/SKILL.md index d484f15e8..e6fd82235 100644 --- a/.claude/skills/verify-berthd/SKILL.md +++ b/.claude/skills/verify-berthd/SKILL.md @@ -52,7 +52,7 @@ export const plugins = [{ id: "counts", name: "Counts", source: `export default export const allowErrors = [/count boom/]; export const init = (layout) => { /* runs in the page before the app, with initArg */ }; export const initArg = { BracketLeft: "å" }; -export async function run({ page, shot, step, expect }) { … } +export async function run({ page, shot, step, expect, call, out }) { … } ``` `init`, when there is one, runs in the page before the app's own scripts, with `initArg` (Playwright's `addInitScript`): it stands in for what the browser would say, such as a keyboard layout (`keyboard-layout`). `expect` is Playwright's, so assertions wait as they do in the e2e suite. Use stable hooks: `data-testid` (`nav-`, `nav-more`, `nav-badge`, `home-grid`, the list at the top of `app/e2e/fixtures.ts`), roles and accessible names. Never coordinates. diff --git a/.claude/skills/verify-berthd/features/README.md b/.claude/skills/verify-berthd/features/README.md index 11a07f637..e649aab46 100644 --- a/.claude/skills/verify-berthd/features/README.md +++ b/.claude/skills/verify-berthd/features/README.md @@ -16,7 +16,7 @@ The maintained list of what a person can do in the app, and how to prove each on | Back and forward: the window strip's arrows, ⌘[ and ⌘], the mouse's buttons; archiving or removing the worktree in front goes back, not Home; the dashboard keeps its scroll | [navigation-history.md](navigation-history.md) | `navigation-history`; `app/e2e/back-forward.spec.ts` asserts it | | Shortcuts in the keyboard's layout (⌘Å and ⌘¨ for Back and Forward on Norwegian) in tooltips, the Keyboard shortcuts sheet, Settings and ⌘K; the sheet's descriptions fit their lines | [navigation-history.md](navigation-history.md) (`history-layout`) | `keyboard-layout`, `keyboard-layout-us`; `app/e2e/back-forward.spec.ts` asserts it | | Stopping agents from the Agent Dashboard: one, the picked ones, a column, Clean up; archiving their worktrees where it's safe | [dashboard-stop.md](dashboard-stop.md) | `dashboard-stop-archive`; `app/e2e/dashboard-stop-archive.spec.ts` asserts it | -| Plugins: built-in and your own, what they add (screens, sidebar items, widgets, panels) | [plugins.md](plugins.md) | `plugin-nav-badges` (sidebar items); `plugin-attach-live` (`api.attach`, a real box, `--agent`) | +| Plugins: built-in and your own, what they add (screens, sidebar items, widgets, panels) | [plugins.md](plugins.md) | `plugin-nav-badges` (sidebar items); `plugin-attach-live` (`api.attach`, a real box, `--agent`); `attention-two-laptops-live` (`berth.attention`, one box and two laptop agents, `../scripts/attention-live.sh`) | | Home's widget grid | [home-widgets.md](home-widgets.md) | none yet; `app/e2e/home-widgets.spec.ts` asserts it | | Questions in chat: an agent's form of questions as a card, answered on a real box; kept with Retry and Copy when it can't be | [chat-questions.md](chat-questions.md) | `chat-questions-live` (a real box, `--agent`); `app/e2e/chat.spec.ts` on the mock | | Chat's foot and reply box: a live turn's spinner, timer, Stop and Queue; a turn that ended while its work runs on shows an hourglass and Send | [chat-turn-state.md](chat-turn-state.md) | `chat-turn-state`; `chat-turn-state-live` (a real box, `--agent`); `app/e2e/chat.spec.ts` asserts it | diff --git a/.claude/skills/verify-berthd/features/plugins.md b/.claude/skills/verify-berthd/features/plugins.md index 8c6838cfe..683cd5ecc 100644 --- a/.claude/skills/verify-berthd/features/plugins.md +++ b/.claude/skills/verify-berthd/features/plugins.md @@ -8,6 +8,7 @@ Plugins add screens, sidebar items, Home widgets, worktree panels and sections, - `plugin-sidebar`: `addSidebarItem` puts a place under More; `useCount` gives it a count (see [sidebar-nav.md](sidebar-nav.md) `nav-counts`). - `plugin-widget`: `addHomeWidget` adds to Home's picker (see [home-widgets.md](home-widgets.md)). - `plugin-attach`: `berth.api.attach` uploads a file the way the composer does, into a running session's worktree or a worktree an agent is about to start in (`.berth/attachments`, which git ignores); a pasted `image.png` goes as `pasted-HHMMSS.png`, the box puts a timestamp before every name, and a file over 20 MB is refused before it is sent. `pasted: true` shrinks a big screenshot to WebP first. A path from `berth.api.localPaths(text)` goes up through the laptop agent, except to a box paired at 127.0.0.1, where it resolves as it is. Needs a real box: `../scripts/scenarios/plugin-attach-live.mjs` on `../scripts/dev-box.sh`, run twice: as `up` starts it, and with `VERIFY_BOX_ADDR=$(tailscale ip -4):7544` so the box counts as another computer (`node drive.mjs plugin-attach-live --port $PORT --agent http://127.0.0.1:9378`). The routes, the refusal, paths and `localPaths` are unit-tested in `app/src/lib/attachments.test.ts`. +- `plugin-attention`: `berth.attention` lists, posts and updates the box's attention items, so a plugin's snooze or "resolved" on one computer is what every other computer paired with the box reads (a plugin's `storage` is per computer). A post replaces the item at its key and counts its revision on; an update naming a revision that has gone by is refused (409); an open item about a session closes as `session_ended` when the session is stopped or its program ends. Needs a real box and two laptops: `../scripts/attention-live.sh up` starts box `devq` on 7547 with laptop A (UI 9472) and laptop B (UI 9474), each paired with its own code, away from `dev-box.sh`'s ports; then `eval "$(attention-live.sh env)"` and `node drive.mjs attention-two-laptops-live --port $PORT --agent http://127.0.0.1:9472`. It drives laptop B in a second browser of its own and puts B's screenshots in the same folder. Run it once per `up`: it posts fixed keys, so a second pass sees later revisions; `down` then `up` between passes. The store's rules are `go test ./internal/box -run 'Attention|AMark|ItemsSettle|AnAgentReplaces'`, the app's calls `app/src/lib/attention.test.ts`. - `plugin-errors`: a plugin that throws while loading shows as failed in Settings → Plugins; one that throws while rendering shows a note in its own place; a sidebar count that throws shows no count, and one that throws while its plugin turns off (an effect's cleanup) leaves the other counts and the app's dialogs working (`app/e2e/plugin-nav-badges.spec.ts`). The rest of the app carries on. ## How to get to it (user POV) @@ -19,6 +20,7 @@ Plugins add screens, sidebar items, Home widgets, worktree panels and sections, - A scenario's `plugins` array is loaded as extra built-ins: `{ id, name, source }`, where `source` is an ES module whose default export gets the plugin context (`berth`). Bare imports of `react` and `@berth/plugin` work as in a real plugin (the app rewrites them to its shims). See `../scripts/scenarios/plugin-nav-badges.mjs`. - This is the way to prove a plugin-facing API: write the smallest plugin that uses it, and drive what it adds. +- `run` also gets `out`, the evidence folder, for a scenario that drives a second browser (a second laptop) and shoots it too (`attention-two-laptops-live`). ## Gotchas diff --git a/.claude/skills/verify-berthd/scripts/attention-live.sh b/.claude/skills/verify-berthd/scripts/attention-live.sh old mode 100644 new mode 100755 index 3202daef3..5d772bf13 --- a/.claude/skills/verify-berthd/scripts/attention-live.sh +++ b/.claude/skills/verify-berthd/scripts/attention-live.sh @@ -13,11 +13,14 @@ root=$(git -C "$(dirname "$0")" rev-parse --show-toplevel) dir=${VERIFY_BOX_DIR:-/tmp/vb-attention} addr=127.0.0.1:7547 box_env=(BERTH_HOME="$dir/box-home" BERTH_USER_DIR="$dir/box-user" TMUX_TMPDIR="$dir/tmux") -lap_env() { echo BERTH_HOME="$dir/lap-$1-home" BERTH_USER_DIR="$dir/lap-$1-user" BERTH_PROXY_ADDR="127.0.0.1:$2" BERTH_UI_ADDR="127.0.0.1:$3"; } box() { env "${box_env[@]}" "$root/bin/berthd" "$@"; } -# shellcheck disable=SC2046 -lap() { local who=$1 p=$2 u=$3; shift 3; env $(lap_env "$who" "$p" "$u") "$root/bin/berth" "$@"; } -tm() { env TMUX_TMPDIR="$dir/tmux" tmux -L berth "$@"; } +# lap a|b …: laptop A (proxy 9471, UI 9472) or B (9473, 9474). +lap() { + local who=$1 p=9471 + shift + [[ $who == b ]] && p=9473 + env BERTH_HOME="$dir/lap-$who-home" BERTH_USER_DIR="$dir/lap-$who-user" BERTH_PROXY_ADDR="127.0.0.1:$p" BERTH_UI_ADDR="127.0.0.1:$((p + 1))" "$root/bin/berth" "$@" +} case ${1:-} in up) @@ -34,8 +37,8 @@ up) exit 1 fi # Each laptop pairs with a code of its own, as two computers do. - lap a 9471 9472 pair "$(box pair --address "$addr" --json | jq -r .link)" --name devq >/dev/null - lap b 9473 9474 pair "$(box pair --address "$addr" --json | jq -r .link)" --name devq >/dev/null + lap a pair "$(box pair --address "$addr" --json | jq -r .link)" --name devq >/dev/null + lap b pair "$(box pair --address "$addr" --json | jq -r .link)" --name devq >/dev/null repo=$dir/acme-shop if [[ ! -d $repo/.git ]]; then git init -q -b main "$repo" && echo "# acme shop" >"$repo/README.md" @@ -43,7 +46,7 @@ up) fi box location add acme-shop "$repo" >/dev/null # The CLI starts each agent; wait until both answer. - lap a 9471 9472 ui-token >/dev/null && lap b 9473 9474 ui-token >/dev/null + lap a ui-token >/dev/null && lap b ui-token >/dev/null for u in 9472 9474; do until ss -ltn | grep -q "127.0.0.1:$u "; do sleep 0.5; done done @@ -51,11 +54,11 @@ up) ;; env) # The tokens are secrets: export them, don't print them anywhere else. - echo "export VERIFY_BERTH_TOKEN=$(lap a 9471 9472 ui-token | jq -r .token) VERIFY_BERTH_TOKEN_B=$(lap b 9473 9474 ui-token | jq -r .token) VERIFY_AGENT_B=http://127.0.0.1:9474 TMUX_TMPDIR=$dir/tmux" + echo "export VERIFY_BERTH_TOKEN=$(lap a ui-token | jq -r .token) VERIFY_BERTH_TOKEN_B=$(lap b ui-token | jq -r .token) VERIFY_AGENT_B=http://127.0.0.1:9474" ;; down) [[ -f $dir/berthd.pid ]] && kill "$(cat "$dir/berthd.pid")" 2>/dev/null || true - tm kill-server 2>/dev/null || true + env TMUX_TMPDIR="$dir/tmux" tmux -L berth kill-server 2>/dev/null || true # The agents the CLI started: the ones on 9472/9474 with this state, no other. for port in 9472 9474; do for pid in $(ss -ltnpH "sport = :$port" 2>/dev/null | grep -o 'pid=[0-9]*' | cut -d= -f2 | sort -u); do diff --git a/.claude/skills/verify-berthd/scripts/scenarios/attention-two-laptops-live.mjs b/.claude/skills/verify-berthd/scripts/scenarios/attention-two-laptops-live.mjs index 7421f962f..6a89c5b80 100644 --- a/.claude/skills/verify-berthd/scripts/scenarios/attention-two-laptops-live.mjs +++ b/.claude/skills/verify-berthd/scripts/scenarios/attention-two-laptops-live.mjs @@ -109,6 +109,8 @@ export async function run({ page, shot, step, expect, call, out }) { step("laptop B: another browser, its own laptop agent paired with the same box, its own plugin storage"); const browser = page.context().browser(); const ctxB = await browser.newContext({ viewport: { width: 1440, height: 900 }, colorScheme: "dark" }); + // As drive.mjs readies its own page: the machine's agent refused, the + // scenario's plugin served as a built-in. await ctxB.route(/^https?:\/\/(?:[^/]*\.)?(?:localhost|127\.0\.0\.1|\[::1\]):(?:1377|1378|1379)(?:\/|$)/, (r) => r.abort()); const pageB = await ctxB.newPage(); const errorsB = []; @@ -123,6 +125,8 @@ export async function run({ page, shot, step, expect, call, out }) { const origin = new URL(page.url()).origin; await pageB.goto(`${origin}/?${new URLSearchParams({ token: TOKEN_B, agent: AGENT_B, ...params })}`); await expect(pageB.getByTestId("nav-home")).toBeVisible({ timeout: 20_000 }); + // Laptop B's own API, as call() is laptop A's. + const callB = (path, init = {}) => fetch(`${AGENT_B}/v1/boxes/${BOX}/api${path}`, { ...init, headers: { Authorization: `Bearer ${TOKEN_B}`, "Content-Type": "application/json" } }); // B's screenshots go beside the driver's own. const shotB = (label) => pageB.screenshot({ path: `${out}/${label}.png` }); try { @@ -138,8 +142,8 @@ export async function run({ page, shot, step, expect, call, out }) { await expect(row(page, ASK)).toHaveText(`${ASK} · decision · closed (resolved) · snoozed until ${UNTIL} · rev 3 · by app`); await shot("4-a-sees-b-resolve"); - step(`a stale write is refused: A snoozes ${ASK} again from revision 2`); - const stale = await fetch(`${AGENT_B}/v1/boxes/${BOX}/api/attention/${encodeURIComponent(ASK)}`, { method: "PATCH", headers: { Authorization: `Bearer ${TOKEN_B}`, "Content-Type": "application/json" }, body: JSON.stringify({ revision: 2, snoozed_until: "" }) }); + step(`a stale write is refused: ending ${ASK}'s snooze from revision 2 answers 409`); + const stale = await callB(`/attention/${encodeURIComponent(ASK)}`, { method: "PATCH", body: JSON.stringify({ revision: 2, snoozed_until: "" }) }); if (stale.status !== 409) throw new Error(`a stale revision answered ${stale.status}, not 409`); step(`end session ${SESSION}: its open ask ${DEPLOY} settles on both laptops, as the box's own change`); @@ -149,8 +153,16 @@ export async function run({ page, shot, step, expect, call, out }) { await shot("5-a-session-ended"); await shotB("6-b-session-ended"); + step("start a session whose program ends by itself after 4 s, with an ask: the box's next look at its sessions settles it"); + await box("/sessions", { method: "POST", body: JSON.stringify({ location: "acme-shop", name: "att-brief", command: "sleep 4" }) }); + await box("/attention", { method: "POST", body: JSON.stringify({ session: "att-brief", kind: "blocked", title: "The acme build is stuck" }) }); + await expect(row(page, "session:att-brief")).toHaveText("session:att-brief · blocked · open · not snoozed · rev 1 · by app"); + await expect(row(pageB, "session:att-brief")).toHaveText("session:att-brief · blocked · closed (session_ended) · not snoozed · rev 2 · by berth", { timeout: 30_000 }); + await expect(row(page, "session:att-brief")).toHaveText("session:att-brief · blocked · closed (session_ended) · not snoozed · rev 2 · by berth"); + await box("/sessions/att-brief", { method: "DELETE" }).catch(() => {}); + step("second view: the box's API through laptop B says the same"); - const res = await fetch(`${AGENT_B}/v1/boxes/${BOX}/api/attention`, { headers: { Authorization: `Bearer ${TOKEN_B}` } }); + const res = await callB("/attention"); const items = await res.json(); const ask = items.find((i) => i.key === ASK); const deploy = items.find((i) => i.key === DEPLOY); diff --git a/app/src/lib/attention.ts b/app/src/lib/attention.ts index 70ed6d5cd..7b6732a3e 100644 --- a/app/src/lib/attention.ts +++ b/app/src/lib/attention.ts @@ -1,4 +1,4 @@ -import type { AttentionItem, AttentionPost, AttentionUpdate, BerthAttention } from "@berth/plugin"; +import type { AttentionItem, BerthAttention } from "@berth/plugin"; import type { Client } from "@/lib/api"; @@ -14,29 +14,22 @@ const missing = (err: unknown) => { return e?.status === 404 && e.code === "not_found"; }; -export const attentionApi = { - list: async (c: Client, box: string, opts: { session?: string; open?: boolean } = {}) => { - const q = new URLSearchParams(); - if (opts.session) q.set("session", opts.session); - if (opts.open) q.set("open", "1"); - const qs = q.toString(); - return (await c.box(box, "GET", `attention${qs ? `?${qs}` : ""}`)) ?? []; - }, - get: (c: Client, box: string, key: string) => - c.box(box, "GET", itemPath(key)).catch((err: unknown) => { - if (missing(err)) return undefined; - throw err; - }), - post: (c: Client, box: string, item: AttentionPost) => c.box(box, "POST", "attention", item), - update: (c: Client, box: string, key: string, change: AttentionUpdate) => c.box(box, "PATCH", itemPath(key), change), -}; - // makeAttentionApi is the store as plugins see it (berth.attention). -export function makeAttentionApi(client: Client): BerthAttention { +export function makeAttentionApi(c: Client): BerthAttention { return { - list: (box, opts) => attentionApi.list(client, box, opts), - get: (box, key) => attentionApi.get(client, box, key), - post: (box, item) => attentionApi.post(client, box, item), - update: (box, key, change) => attentionApi.update(client, box, key, change), + list: async (box, opts = {}) => { + const q = new URLSearchParams(); + if (opts.session) q.set("session", opts.session); + if (opts.open) q.set("open", "1"); + const qs = q.toString(); + return (await c.box(box, "GET", `attention${qs ? `?${qs}` : ""}`)) ?? []; + }, + get: (box, key) => + c.box(box, "GET", itemPath(key)).catch((err: unknown) => { + if (missing(err)) return undefined; + throw err; + }), + post: (box, item) => c.box(box, "POST", "attention", item), + update: (box, key, change) => c.box(box, "PATCH", itemPath(key), change), }; } diff --git a/cmd/berthd/main.go b/cmd/berthd/main.go index e90f17bcd..e570fecb8 100644 --- a/cmd/berthd/main.go +++ b/cmd/berthd/main.go @@ -410,6 +410,8 @@ func serve(b boxHome, args []string) error { defer bx.Helpers.Watch(bus)() bx.AutoFix = &box.AutoFixStore{Path: filepath.Join(b.dir, "autofix.json")} bx.Attention = &box.AttentionStore{Path: filepath.Join(b.dir, "attention.json")} + // A stopped session's asks settle at once, however it was stopped. + defer bx.WatchAttention()() // How agents start when what starts them names no driver // (agentdefaults.go): the app keeps it to the person's Labs setting. bx.AgentDefaults = &box.AgentDefaultsStore{Path: filepath.Join(b.dir, "agent-defaults.json")} diff --git a/internal/box/api.go b/internal/box/api.go index 763f5d42d..b7a682f67 100644 --- a/internal/box/api.go +++ b/internal/box/api.go @@ -672,7 +672,6 @@ func (b *Box) stopSessionsIn(from, dir string) { b.Events.Publish(events.Event{Type: "session.stopped", Box: b.Name, Origin: from, Data: map[string]any{"name": s.Name, "path": s.Dir}}) } } - b.attentionSessionsEnded(ctx) } func (b *Box) listSessions(w http.ResponseWriter, r *http.Request) error { @@ -929,7 +928,6 @@ func (b *Box) removeSession(w http.ResponseWriter, r *http.Request) error { return err } b.publish(r, "session.stopped", map[string]any{"name": name}) - b.attentionSessionsEnded(r.Context()) writeJSON(w, map[string]string{"removed": name}) return nil } diff --git a/internal/box/attention.go b/internal/box/attention.go index 887878b70..1cf5e5bc2 100644 --- a/internal/box/attention.go +++ b/internal/box/attention.go @@ -2,11 +2,11 @@ package box import ( "encoding/json" - "errors" "fmt" "net/http" "os" "regexp" + "slices" "sort" "strings" "sync" @@ -124,8 +124,8 @@ const ( ) var ( - errAttentionRevision = errors.New("the item changed since that revision; read it again") - errAttentionFull = fmt.Errorf("the box already holds %d open attention items; settle some first", maxAttentionItems) + errAttentionRevision = httpError{http.StatusConflict, "the item changed since that revision; read it again"} + errAttentionFull = httpError{http.StatusTooManyRequests, fmt.Sprintf("the box already holds %d open attention items; settle some first", maxAttentionItems)} attentionWord = regexp.MustCompile(`^[a-z][a-z0-9_-]{0,31}$`) ) @@ -240,12 +240,47 @@ func (s *AttentionStore) Get(key string) (AttentionItem, bool) { // SessionKey is the key of a session's own ask. func SessionKey(session string) string { return "session:" + session } +// KeyOrSession is the post's key: the one it names, or its session's. +func (p AttentionPost) KeyOrSession() string { + if p.Key == "" && p.Session != "" { + return SessionKey(p.Session) + } + return p.Key +} + +// slot finds the item at key for a change, refusing one made from a +// revision that has gone by, and makes room when there is no item yet. It +// holds s.mu. +func (s *AttentionStore) slot(key string, revision int) (*AttentionItem, error) { + was := s.items[key] + if revision != 0 && (was == nil || was.Revision != revision) { + return nil, errAttentionRevision + } + if was == nil { + return nil, s.makeRoom() + } + return was, nil +} + +// commit puts it at its key and saves, putting back what was there (nil: +// nothing) when the save fails. It holds s.mu. +func (s *AttentionStore) commit(it, was *AttentionItem) error { + s.items[it.Key] = it + if err := s.save(); err != nil { + if was != nil { + s.items[it.Key] = was + } else { + delete(s.items, it.Key) + } + return err + } + return nil +} + // Post puts an ask at its key, replacing the one there: it opens again with // a new revision, and its answer and snooze are gone. func (s *AttentionStore) Post(p AttentionPost, by string) (AttentionItem, error) { - if p.Key == "" && p.Session != "" { - p.Key = SessionKey(p.Session) - } + p.Key = p.KeyOrSession() if err := checkAttentionPost(p); err != nil { return AttentionItem{}, err } @@ -253,76 +288,66 @@ func (s *AttentionStore) Post(p AttentionPost, by string) (AttentionItem, error) defer s.mu.Unlock() s.load() now := s.now() - was, ok := s.items[p.Key] - if p.Revision != 0 && (!ok || was.Revision != p.Revision) { - return AttentionItem{}, httpError{http.StatusConflict, errAttentionRevision.Error()} - } - if !ok { - if err := s.makeRoom(); err != nil { - return AttentionItem{}, err - } + was, err := s.slot(p.Key, p.Revision) + if err != nil { + return AttentionItem{}, err } it := &AttentionItem{ Key: p.Key, Revision: 1, Kind: p.Kind, Title: p.Title, Details: p.Details, Options: p.Options, Recommended: p.Recommended, Status: AttentionOpen, Session: p.Session, By: by, Created: now, Updated: now, } - if ok { + if was != nil { it.Revision, it.Created = was.Revision+1, was.Created } - s.items[p.Key] = it - if err := s.save(); err != nil { - if ok { - s.items[p.Key] = was - } else { - delete(s.items, p.Key) - } + if err := s.commit(it, was); err != nil { return AttentionItem{}, err } return cloneAttention(*it), nil } // Update changes the item at key, or makes a mark there when there is -// none. made says it made one. -func (s *AttentionStore) Update(key string, u AttentionUpdate, by string) (it AttentionItem, made bool, err error) { +// none. +func (s *AttentionStore) Update(key string, u AttentionUpdate, by string) (AttentionItem, error) { if err := checkAttentionKey(key); err != nil { - return AttentionItem{}, false, err + return AttentionItem{}, err } var until time.Time if u.SnoozedUntil != nil && *u.SnoozedUntil != "" { t, err := time.Parse(time.RFC3339, *u.SnoozedUntil) if err != nil { - return AttentionItem{}, false, badRequest("snoozed_until must be an RFC 3339 time, or \"\" to end the snooze") + return AttentionItem{}, badRequest("snoozed_until must be an RFC 3339 time, or \"\" to end the snooze") } until = t.UTC() } if u.Status != nil && *u.Status != AttentionOpen && *u.Status != AttentionAnswered && *u.Status != AttentionClosed { - return AttentionItem{}, false, badRequest("status must be open, answered or closed") + return AttentionItem{}, badRequest("status must be open, answered or closed") } if u.Answer != nil && len(*u.Answer) > maxAttentionText { - return AttentionItem{}, false, badRequest("an answer may be %d KB", maxAttentionText>>10) + return AttentionItem{}, badRequest("an answer may be %d KB", maxAttentionText>>10) } if u.ClosedReason != nil && *u.ClosedReason != "" && !attentionWord.MatchString(*u.ClosedReason) { - return AttentionItem{}, false, badRequest("closed_reason is a short word: lowercase letters, digits, - and _") + return AttentionItem{}, badRequest("closed_reason is a short word: lowercase letters, digits, - and _") } - if len(u.Session) > maxAttentionSession { - return AttentionItem{}, false, badRequest("that isn't a session name") + if err := checkAttentionSession(u.Session); err != nil { + return AttentionItem{}, err } s.mu.Lock() defer s.mu.Unlock() s.load() now := s.now() - was, ok := s.items[key] - if u.Revision != 0 && (!ok || was.Revision != u.Revision) { - return AttentionItem{}, false, httpError{http.StatusConflict, errAttentionRevision.Error()} + was, err := s.slot(key, u.Revision) + if err != nil { + return AttentionItem{}, err } - var next AttentionItem - if ok { + next := AttentionItem{Key: key, Kind: attentionMark, Status: AttentionOpen, Session: u.Session, Created: now} + if was != nil { next = cloneAttention(*was) - } else { - if err := s.makeRoom(); err != nil { - return AttentionItem{}, false, err - } - next = AttentionItem{Key: key, Kind: attentionMark, Status: AttentionOpen, Session: u.Session, Created: now} + } + if u.Answer != nil { + next.Answer = *u.Answer + } + if u.ClosedReason != nil { + next.ClosedReason = *u.ClosedReason } status := next.Status switch { @@ -334,21 +359,12 @@ func (s *AttentionStore) Update(key string, u AttentionUpdate, by string) (it At status = AttentionClosed } if status != next.Status { - next.Status = status - next.Settled = time.Time{} - if status != AttentionOpen { - next.Settled = now - } + next.Settled = now } - if u.Answer != nil { - next.Answer = *u.Answer - } - if u.ClosedReason != nil { - next.ClosedReason = *u.ClosedReason - } - switch next.Status { + next.Status = status + switch status { case AttentionOpen: - next.Answer, next.ClosedReason = "", "" + next.Answer, next.ClosedReason, next.Settled = "", "", time.Time{} case AttentionAnswered: next.ClosedReason = "" case AttentionClosed: @@ -361,16 +377,10 @@ func (s *AttentionStore) Update(key string, u AttentionUpdate, by string) (it At } next.Revision++ next.By, next.Updated = by, now - s.items[key] = &next - if err := s.save(); err != nil { - if ok { - s.items[key] = was - } else { - delete(s.items, key) - } - return AttentionItem{}, false, err + if err := s.commit(&next, was); err != nil { + return AttentionItem{}, err } - return cloneAttention(next), !ok, nil + return cloneAttention(next), nil } // makeRoom keeps the store to maxAttentionItems before a new key: the @@ -387,15 +397,26 @@ func (s *AttentionStore) makeRoom() error { } } if oldest == nil { - return httpError{http.StatusTooManyRequests, errAttentionFull.Error()} + return errAttentionFull } delete(s.items, oldest.Key) return nil } // Sweep settles the open items whose session is not in live, and forgets -// items past their time. It returns the items it settled. +// items past their time. It returns the items it settled. It is the +// backstop for sessions that end without saying so (a program that exits); +// SettleSession is the prompt way for one that was stopped. func (s *AttentionStore) Sweep(live map[string]bool) []AttentionItem { + return s.settle(func(it *AttentionItem) bool { return !live[it.Session] }, true) +} + +// SettleSession settles the open items about a session that ended. +func (s *AttentionStore) SettleSession(name string) []AttentionItem { + return s.settle(func(it *AttentionItem) bool { return it.Session == name }, false) +} + +func (s *AttentionStore) settle(ended func(*AttentionItem) bool, forget bool) []AttentionItem { s.mu.Lock() defer s.mu.Unlock() s.load() @@ -403,14 +424,14 @@ func (s *AttentionStore) Sweep(live map[string]bool) []AttentionItem { var settled []AttentionItem changed := false for key, it := range s.items { - if it.Status == AttentionOpen && it.Session != "" && !live[it.Session] { + if it.Status == AttentionOpen && it.Session != "" && ended(it) { it.Status, it.ClosedReason, it.Settled = AttentionClosed, attentionSessionEnded, now it.Revision++ it.By, it.Updated = "berth", now settled = append(settled, cloneAttention(*it)) changed = true } - if expired(*it, now) { + if forget && expired(*it, now) { delete(s.items, key) changed = true } @@ -423,12 +444,17 @@ func (s *AttentionStore) Sweep(live map[string]bool) []AttentionItem { } func cloneAttention(it AttentionItem) AttentionItem { - if it.Options != nil { - it.Options = append([]AttentionOption(nil), it.Options...) - } + it.Options = slices.Clone(it.Options) return it } +func checkAttentionSession(name string) error { + if len(name) > maxAttentionSession { + return badRequest("that isn't a session name") + } + return nil +} + func checkAttentionKey(key string) error { if key == "" || len(key) > maxAttentionKey || strings.TrimSpace(key) != key || strings.IndexFunc(key, unicode.IsControl) >= 0 || !utf8.ValidString(key) { return badRequest("a key is 1 to %d characters, with no control characters or surrounding spaces", maxAttentionKey) @@ -452,8 +478,8 @@ func checkAttentionPost(p AttentionPost) error { if len(p.Details) > maxAttentionText { return badRequest("details may be %d KB", maxAttentionText>>10) } - if len(p.Session) > maxAttentionSession { - return badRequest("that isn't a session name") + if err := checkAttentionSession(p.Session); err != nil { + return err } if len(p.Options) > maxAttentionOptions { return badRequest("an ask offers at most %d options", maxAttentionOptions) diff --git a/internal/box/attention_test.go b/internal/box/attention_test.go index 1481a3e65..e26145923 100644 --- a/internal/box/attention_test.go +++ b/internal/box/attention_test.go @@ -14,12 +14,8 @@ import ( "github.com/cosscom/shipyard/internal/hooks" ) -type attentionClock struct{ t time.Time } - -func (c *attentionClock) now() time.Time { return c.t } -func (c *attentionClock) add(d time.Duration) { c.t = c.t.Add(d) } -func newAttention(t *testing.T) (*AttentionStore, *attentionClock) { - clock := &attentionClock{time.Date(2026, 11, 6, 9, 0, 0, 0, time.UTC)} +func newAttention(t *testing.T) (*AttentionStore, *clock) { + clock := &clock{time.Date(2026, 11, 6, 9, 0, 0, 0, time.UTC)} return &AttentionStore{Path: filepath.Join(t.TempDir(), "attention.json"), Now: clock.now}, clock } @@ -34,10 +30,10 @@ func TestAnAgentReplacesItsOwnAskByKey(t *testing.T) { } // Someone snoozes it, then the agent asks again: the same item, a new // revision, open and no longer snoozed. - if _, _, err := s.Update(first.Key, AttentionUpdate{SnoozedUntil: strp("2026-11-06T12:00:00Z")}, "app"); err != nil { + if _, err := s.Update(first.Key, AttentionUpdate{SnoozedUntil: strp("2026-11-06T12:00:00Z")}, "app"); err != nil { t.Fatal(err) } - clock.add(time.Minute) + clock.advance(time.Minute) second, err := s.Post(AttentionPost{Session: "shop-claude", Kind: "decision", Title: "Which queue, now that Redis is out?"}, "claude") if err != nil { t.Fatal(err) @@ -53,10 +49,10 @@ func TestAnAgentReplacesItsOwnAskByKey(t *testing.T) { if _, err := s.Post(AttentionPost{Session: "shop-claude", Kind: "decision", Title: "stale", Revision: 2}, "claude"); statusFor(err) != http.StatusConflict { t.Fatalf("stale post: %v", err) } - if _, _, err := s.Update(second.Key, AttentionUpdate{Revision: 1, Status: strp(AttentionClosed)}, "app"); statusFor(err) != http.StatusConflict { + if _, err := s.Update(second.Key, AttentionUpdate{Revision: 1, Status: strp(AttentionClosed)}, "app"); statusFor(err) != http.StatusConflict { t.Fatalf("stale update: %v", err) } - answered, _, err := s.Update(second.Key, AttentionUpdate{Revision: 3, Answer: strp("SQS")}, "phone") + answered, err := s.Update(second.Key, AttentionUpdate{Revision: 3, Answer: strp("SQS")}, "phone") if err != nil { t.Fatal(err) } @@ -71,22 +67,22 @@ func TestAnAgentReplacesItsOwnAskByKey(t *testing.T) { func TestAMarkOnAnItemTheBoxDoesntHold(t *testing.T) { s, _ := newAttention(t) key := "run/r_12/review" // a plugin's key for a run's gate, slashes and all - m, made, err := s.Update(key, AttentionUpdate{SnoozedUntil: strp("2026-11-07T08:00:00+01:00")}, "app") + m, err := s.Update(key, AttentionUpdate{SnoozedUntil: strp("2026-11-07T08:00:00+01:00")}, "app") if err != nil { t.Fatal(err) } - if !made || m.Kind != "mark" || m.Status != AttentionOpen || m.Revision != 1 || !m.SnoozedUntil.Equal(time.Date(2026, 11, 7, 7, 0, 0, 0, time.UTC)) { - t.Fatalf("mark = %+v made=%v", m, made) + if m.Kind != "mark" || m.Status != AttentionOpen || m.Revision != 1 || !m.SnoozedUntil.Equal(time.Date(2026, 11, 7, 7, 0, 0, 0, time.UTC)) { + t.Fatalf("mark = %+v", m) } - r, made, err := s.Update(key, AttentionUpdate{Status: strp(AttentionClosed)}, "app") + r, err := s.Update(key, AttentionUpdate{Status: strp(AttentionClosed)}, "app") if err != nil { t.Fatal(err) } - if made || r.Status != AttentionClosed || r.ClosedReason != "resolved" || r.Revision != 2 { + if r.Kind != "mark" || r.Status != AttentionClosed || r.ClosedReason != "resolved" || r.Revision != 2 { t.Fatalf("resolved = %+v", r) } // Reopening clears what settled it; "" ends the snooze. - o, _, _ := s.Update(key, AttentionUpdate{Status: strp(AttentionOpen), SnoozedUntil: strp("")}, "app") + o, _ := s.Update(key, AttentionUpdate{Status: strp(AttentionOpen), SnoozedUntil: strp("")}, "app") if o.Status != AttentionOpen || o.ClosedReason != "" || !o.Settled.IsZero() || !o.SnoozedUntil.IsZero() { t.Fatalf("reopened = %+v", o) } @@ -103,7 +99,7 @@ func TestAMarkOnAnItemTheBoxDoesntHold(t *testing.T) { t.Fatalf("%+v: %v", p, err) } } - if _, _, err := s.Update(key, AttentionUpdate{SnoozedUntil: strp("tomorrow")}, "app"); statusFor(err) != http.StatusBadRequest { + if _, err := s.Update(key, AttentionUpdate{SnoozedUntil: strp("tomorrow")}, "app"); statusFor(err) != http.StatusBadRequest { t.Fatalf("a bad time: %v", err) } } @@ -133,12 +129,12 @@ func TestItemsSettleWhenTheirSessionEndsAndAgeOut(t *testing.T) { } // 14 days on, the settled ones are gone. The snoozed mark stays until // its snooze has run out as well. - clock.add(AttentionKeep + time.Hour) + clock.advance(AttentionKeep + time.Hour) s.Sweep(nil) if all := s.List(AttentionFilter{}); len(all) != 1 || all[0].Key != "standalone" { t.Fatalf("after 14 days: %+v", all) } - clock.add(30 * 24 * time.Hour) + clock.advance(30 * 24 * time.Hour) s.Sweep(nil) if all := s.List(AttentionFilter{}); len(all) != 0 { t.Fatalf("an old mark whose snooze ran out: %+v", all) @@ -161,7 +157,7 @@ func TestAttentionSurvivesARestart(t *testing.T) { t.Fatalf("%d items after restart", n) } // One read 15 days on leaves the settled ones behind. - clock.add(15 * 24 * time.Hour) + clock.advance(15 * 24 * time.Hour) later := &AttentionStore{Path: s.Path, Now: clock.now} if all := later.List(AttentionFilter{}); len(all) != 1 || all[0].Key != "review/shop#12" { t.Fatalf("read 15 days on: %+v", all) @@ -254,10 +250,20 @@ func TestAttentionRoutes(t *testing.T) { t.Fatal("a refused update made a mark") } - // A session that ends settles its items, announced as the box's own. + // A session that is stopped settles its items at once, announced as + // the box's own; one whose program ended settles at the next sweep. + defer bx.WatchAttention()() bx.Attention.Post(AttentionPost{Session: "shop-claude", Kind: "action", Title: "Log in to Fly"}, "claude") - bx.sweepAttention([]Session{{Name: "shop-claude", Exited: true}}) + bx.Attention.Post(AttentionPost{Session: "shop-codex", Kind: "action", Title: "Rotate the acme key"}, "codex") + bus.Publish(events.Event{Type: "session.stopped", Data: map[string]any{"name": "shop-claude"}}) if e := waitEventType(t, evs, "attention.settled"); e.Data["key"] != "session:shop-claude" || e.Data["closed_reason"] != "session_ended" || e.Origin != "berth" { t.Fatalf("settled event %+v", e) } + if it, _ := bx.Attention.Get("session:shop-codex"); it.Status != AttentionOpen { + t.Fatalf("another session's ask settled: %+v", it) + } + bx.sweepAttention([]Session{{Name: "shop-codex", Exited: true}}) + if e := waitEventType(t, evs, "attention.settled"); e.Data["key"] != "session:shop-codex" { + t.Fatalf("swept event %+v", e) + } } diff --git a/internal/box/attentionapi.go b/internal/box/attentionapi.go index a5b6d30d4..c33ee281c 100644 --- a/internal/box/attentionapi.go +++ b/internal/box/attentionapi.go @@ -1,8 +1,6 @@ package box import ( - "context" - "errors" "net/http" "github.com/cosscom/shipyard/internal/events" @@ -63,17 +61,11 @@ func (b *Box) postAttention(w http.ResponseWriter, r *http.Request) error { // runs. if p.Session != "" && b.Sessions != nil { if _, err := b.Sessions.Get(r.Context(), p.Session); err != nil { - if errors.Is(err, ErrUnknownSession) { - return httpError{http.StatusNotFound, "no session named " + p.Session + " on this box"} - } return err } } - key := p.Key - if key == "" && p.Session != "" { - key = SessionKey(p.Session) - } - if err := b.before(r, "attention.post", map[string]any{"key": key, "kind": p.Kind, "session": p.Session}); err != nil { + p.Key = p.KeyOrSession() + if err := b.before(r, "attention.post", map[string]any{"key": p.Key, "kind": p.Kind, "session": p.Session}); err != nil { return err } it, err := b.Attention.Post(p, origin(r)) @@ -104,7 +96,7 @@ func (b *Box) updateAttention(w http.ResponseWriter, r *http.Request) error { if err := b.before(r, "attention.update", gate); err != nil { return err } - it, _, err := b.Attention.Update(key, u, origin(r)) + it, err := b.Attention.Update(key, u, origin(r)) if err != nil { return err } @@ -114,7 +106,8 @@ func (b *Box) updateAttention(w http.ResponseWriter, r *http.Request) error { } // sweepAttention settles the items of sessions that ended, given the -// sessions there are, and forgets items past their time. +// sessions there are, and forgets items past their time. refreshTurns runs +// it as the backstop for a session whose program ended by itself. func (b *Box) sweepAttention(all []Session) { if b.Attention == nil { return @@ -125,19 +118,30 @@ func (b *Box) sweepAttention(all []Session) { live[s.Name] = true } } - for _, it := range b.Attention.Sweep(live) { - b.Events.Publish(events.Event{Type: "attention.settled", Box: b.Name, Origin: "berth", Data: attentionEventData(it)}) - } + b.publishSettled(b.Attention.Sweep(live)) } -// attentionSessionsEnded sweeps at once after sessions were stopped, rather -// than at the next refresh. -func (b *Box) attentionSessionsEnded(ctx context.Context) { - if b.Attention == nil || b.Sessions == nil { - return - } - if all, err := b.Sessions.List(ctx); err == nil { - b.sweepAttention(all) +// WatchAttention settles a session's items as soon as it is stopped, +// however it was: every stop ends in a session.stopped event, which is what +// this listens for. The returned func stops it. +func (b *Box) WatchAttention() (stop func()) { + if b.Attention == nil || b.Events == nil { + return func() {} + } + return b.Events.Observe(func(e events.Event) { + if e.Type != "session.stopped" { + return + } + if name, _ := e.Data["name"].(string); name != "" { + // Not here: an observer must be quick and must not publish. + go b.publishSettled(b.Attention.SettleSession(name)) + } + }) +} + +func (b *Box) publishSettled(settled []AttentionItem) { + for _, it := range settled { + b.Events.Publish(events.Event{Type: "attention.settled", Box: b.Name, Origin: "berth", Data: attentionEventData(it)}) } } From 5b1b834db72b6747944a7f57ee614e0c759f852e Mon Sep 17 00:00:00 2001 From: Stener1 Date: Fri, 6 Nov 2026 12:55:10 +0000 Subject: [PATCH 3/4] fix(box): settle attention items only on listings that could see them, and never lose items on a failed read or write Co-Authored-By: Claude Opus 5.5 --- docs/reference/plugin-sdk.mdx | 7 +- internal/box/attention.go | 153 +++++++++++++++++++--------- internal/box/attention_test.go | 166 +++++++++++++++++++++++++++---- internal/box/attentionapi.go | 27 +++-- internal/box/turns.go | 3 +- packages/plugin-sdk/src/index.ts | 4 +- 6 files changed, 277 insertions(+), 83 deletions(-) diff --git a/docs/reference/plugin-sdk.mdx b/docs/reference/plugin-sdk.mdx index 356854f41..d9d2e6f57 100644 --- a/docs/reference/plugin-sdk.mdx +++ b/docs/reference/plugin-sdk.mdx @@ -320,14 +320,15 @@ counts its `revision` on, so an agent replaces its own ask rather than adding another; a post with a `session` and no key is that session's own, `session:`. An item is `open`, `answered` or `closed`; the last two are settled. An open item about a session closes by itself, with -`closed_reason: "session_ended"`, when the session ends, and a settled item -is forgotten 14 days later. +`closed_reason: "session_ended"`, when the session ends: at once when it is +stopped, and within about 20 seconds when its program ends by itself. A +settled item is forgotten 14 days later. | Call | What it does | | --- | --- | | `list(box, { session?, open? })` | `AttentionItem[]`, the most recently changed first: `{ key, revision, kind, title?, details?, options?, recommended?, status, answer?, closed_reason?, snoozed_until?, session?, by, created, updated, settled? }`. `by` is where the last change came from (`app`, `phone`, an agent's tool, `berth`) | | `get(box, key)` | One item, or `undefined` when there is none at that key | -| `post(box, { key?, kind, title, details?, options?, recommended?, session?, revision? })` | Posts an ask, or replaces the one at its key: it opens again, its answer and snooze gone. `kind` is a short lowercase word of the poster's (`mark` is the box's own), `options` up to 12 `{ label, description? }` and `recommended` one of their labels. A `session` must be running (404 otherwise) | +| `post(box, { key?, kind, title, details?, options?, recommended?, session?, revision? })` | Posts an ask, or replaces the one at its key: it opens again, its answer and snooze gone. `kind` is a short lowercase word of the poster's (`mark` is the box's own), `options` up to 12 `{ label, description? }` and `recommended` one of their labels. A `session` must be on the box (404 otherwise) with its program running (409 `session_exited` otherwise) | | `update(box, key, { revision?, status?, answer?, closed_reason?, snoozed_until?, session? })` | Answers, closes, reopens or snoozes an item; what it leaves out stays. An `answer` alone makes it `answered`, a `closed_reason` alone closes it (closed without one is `resolved`), and `status: "open"` reopens it. `snoozed_until` is an ISO time, or `""` to end a snooze. At a key with no item it makes a mark (`kind: "mark"`), about `session` when given one | With `revision`, a post or update is refused (409) if the item has changed diff --git a/internal/box/attention.go b/internal/box/attention.go index 1cf5e5bc2..b881de388 100644 --- a/internal/box/attention.go +++ b/internal/box/attention.go @@ -2,6 +2,7 @@ package box import ( "encoding/json" + "errors" "fmt" "net/http" "os" @@ -138,6 +139,10 @@ type AttentionStore struct { mu sync.Mutex items map[string]*AttentionItem loaded bool + // missing names the sessions of open items that the last sweep didn't + // find. A session is taken to have ended only when two sweeps in a row + // miss it, so one listing that came back short settles nothing. + missing map[string]bool } func (s *AttentionStore) now() time.Time { @@ -147,30 +152,41 @@ func (s *AttentionStore) now() time.Time { return time.Now().UTC() } -// load reads the file once. It holds s.mu. An item kept past its time is -// left behind as it is read. -func (s *AttentionStore) load() { +// load reads the file once. It holds s.mu. A file it can't read now fails +// the call, and the next call tries again: starting empty would have the +// next change write over every item. A file that isn't JSON is set aside +// as attention.json.corrupt-