From 01ab81792366d3c08991940f75bb715b6d6dbf60 Mon Sep 17 00:00:00 2001 From: casquijo Date: Thu, 10 Sep 2026 09:37:10 -0300 Subject: [PATCH] fix(usage): enforce spend caps for room turns --- docs/verification/spend-cap.md | 21 +- server/channel-queue.test.ts | 8 +- server/channel-queue.ts | 7 +- server/index.ts | 188 +++++++++++++++--- server/routines.test.ts | 6 +- server/routines.ts | 2 + server/spend-cap-api.test.ts | 348 ++++++++++++++++++++++++++++++++- 7 files changed, 536 insertions(+), 44 deletions(-) diff --git a/docs/verification/spend-cap.md b/docs/verification/spend-cap.md index 0183a7e12f..74ac452ae4 100644 --- a/docs/verification/spend-cap.md +++ b/docs/verification/spend-cap.md @@ -3,10 +3,11 @@ ## Sub-features - Refuse every new turn once the month's reported cost reaches the workspace - cap: the message routes answer 409 with `code: "spend_cap"`, and a routine, - peer hop or webhook stops in `startTurn` with the same refusal. + cap: the message routes answer 409 with `code: "spend_cap"`; routines, peer + hops and webhooks stop in `startTurn`; and every room member, chained mention, + goal step, retry, queued send or calendar call checks again at dispatch. - Count a turn the moment it settles, not when the ledger's append lands or a - cache expires. + cache expires, including each provider-backed room or calendar member. - Price turns from the operator's list (`driver/model`, then model, then `default`) into a billable column in `/api/usage` and its CSV. - Do nothing at all without the `budgets` / `billing` entitlements. @@ -25,12 +26,14 @@ pnpm exec vitest run --no-file-parallelism server/spend-cap-api.test.ts The test writes a stand-in enterprise layer (the folder shape core loads, granting `budgets` and `billing`) and launches the `control-omb` fixture with -it through `launchVerificationServer(..., { dir, licenseKey })`. It sets a -$0.015 cap and a default price list, sends two turns that the fake engine -books at $0.01 each, and checks the third is refused with 409 `spend_cap`, -that `/api/usage` reports the cap exceeded, warned, and priced, that the CSV -carries `billable_usd`, and that raising the cap lets the next turn through. -It prints the fixture's server log path and removes its temporary homes. +it through `launchVerificationServer(..., { dir, licenseKey })`. It covers the +direct 409, pricing and raised-cap path, then proves that room and calendar +members are booked individually, the next room member is stopped at the +provider boundary, room-goal spend is attributed to its routine, each queued +user keeps their own origin, calendar operations remain owner-attributed when +user work queues behind them, and a capped scheduled goal is blocked rather +than retried as a provider failure. It prints each fixture's server log path +and removes its temporary homes. For the same by hand, launch a fixture with `OMB_ENTERPRISE_DIR` pointing at a folder whose `server/index.js` exports such a `register()`, and diff --git a/server/channel-queue.test.ts b/server/channel-queue.test.ts index 0817706a88..ec72fd3c41 100644 --- a/server/channel-queue.test.ts +++ b/server/channel-queue.test.ts @@ -16,10 +16,12 @@ describe("channel queue", () => { }); const first = queueChannelMessage("group-a", "thread-a", "first follow-up", { sendId: "send_first_123456", + usageTrigger: { kind: "user", label: "Alice device" }, }); queueChannelMessage("group-a", "thread-a", "second follow-up", { sendId: "send_second_123456", mode: "goal", + usageTrigger: { kind: "user", label: "Bob device" }, }); drainChannelMessages(() => working, run); @@ -35,6 +37,7 @@ describe("channel queue", () => { threadId: "thread-a", text: "first follow-up", mode: "chat", + usageTrigger: { kind: "user", label: "Alice device" }, })); expect(_queuedChannelCount("thread-a")).toBe(1); @@ -44,13 +47,14 @@ describe("channel queue", () => { expect(run).toHaveBeenLastCalledWith(expect.objectContaining({ text: "second follow-up", mode: "goal", + usageTrigger: { kind: "user", label: "Bob device" }, })); expect(_queuedChannelCount("thread-a")).toBe(0); }); it("cancels only the requested channel message", () => { - const keep = queueChannelMessage("group-b", "thread-b", "keep"); - const drop = queueChannelMessage("group-b", "thread-b", "drop"); + const keep = queueChannelMessage("group-b", "thread-b", "keep", { usageTrigger: { kind: "owner" } }); + const drop = queueChannelMessage("group-b", "thread-b", "drop", { usageTrigger: { kind: "owner" } }); expect(cancelChannelMessage("group-b", drop.id)).toBe(true); expect(cancelChannelMessage("group-b", drop.id)).toBe(false); diff --git a/server/channel-queue.ts b/server/channel-queue.ts index 3c4a1b65e4..4ef5f21214 100644 --- a/server/channel-queue.ts +++ b/server/channel-queue.ts @@ -7,6 +7,7 @@ // never saw. import { newId } from "./contracts.ts"; +import type { UsageTrigger } from "./usage-ledger.ts"; interface ChannelQueueItem { id: string; @@ -14,6 +15,8 @@ interface ChannelQueueItem { replyToId?: string; sendId?: string; mode: "chat" | "goal"; + /** Origin captured when the request entered the queue. */ + readonly usageTrigger: UsageTrigger; /** kept so the drain appends it with the same provenance it arrived with */ via?: "api"; } @@ -38,7 +41,8 @@ export function queueChannelMessage( sendId?: string; mode?: "chat" | "goal"; via?: "api"; - } = {}, + usageTrigger: UsageTrigger; + }, ): QueuedChannelMessage { const entry = queues.get(threadId) ?? { groupId, items: [] }; if (entry.groupId !== groupId) throw new Error("queued task belongs to another channel"); @@ -48,6 +52,7 @@ export function queueChannelMessage( replyToId: options.replyToId, sendId: options.sendId, mode: options.mode ?? "chat", + usageTrigger: { ...options.usageTrigger }, via: options.via, }; entry.items.push(item); diff --git a/server/index.ts b/server/index.ts index 34d0cc9abe..a0191d0f1d 100644 --- a/server/index.ts +++ b/server/index.ts @@ -465,19 +465,19 @@ const registry = new ProviderRegistry(BUILT_IN_DRIVERS); // any other copy on PATH. registerEnginesBinDir(); -// Who asked for the next turn on a thread, noted where a message comes in -// and read by the usage ledger when the turn settles. The last note stands -// until the next message: a resumed connector or a drained queued send -// still belongs to the person who wrote, and routine or peer turns are -// told apart before this is consulted. +// Who asked for work on a thread. Direct turns read this when they settle; +// room ingress copies it onto each operation/queue item so later messages +// cannot change the origin of work already accepted by the harness. const turnTriggers = new Map(); -function noteTurnTrigger(threadId: string, auth: RequestAuth): void { - turnTriggers.set( - threadId, - auth.kind === "session" - ? { kind: "user", ...(auth.session.email ? { email: auth.session.email } : {}), label: auth.session.label } - : { kind: "owner" }, - ); +function copyUsageTrigger(trigger: UsageTrigger): UsageTrigger { + return { ...trigger }; +} +function noteTurnTrigger(threadId: string, auth: RequestAuth): UsageTrigger { + const trigger: UsageTrigger = auth.kind === "session" + ? { kind: "user", ...(auth.session.email ? { email: auth.session.email } : {}), label: auth.session.label } + : { kind: "owner" }; + turnTriggers.set(threadId, trigger); + return copyUsageTrigger(trigger); } const providerAuthSessions = new ProviderAuthSessions(); await registry.load(instanceConfigs(cfg)); @@ -1809,6 +1809,8 @@ const publicBotQueuedMessages = () => queuedSteerSnapshot((botId, threadId) => B type GroupTurnOperation = { id: string; threadId: string; + /** Immutable origin of every provider turn dispatched for this operation. */ + readonly usageTrigger: UsageTrigger; botIds: Set; cancelled: boolean; cancellation: AbortController; @@ -2075,10 +2077,12 @@ function beginGroupTurnOperation( groupId: string, threadId: string, botIds: Iterable = [], + usageTrigger: UsageTrigger = turnTriggers.get(threadId) ?? { kind: "owner" }, ): GroupTurnOperation { const operation = { id: randomUUID(), threadId, + usageTrigger: copyUsageTrigger(usageTrigger), botIds: new Set(botIds), cancelled: false, cancellation: new AbortController(), @@ -2699,6 +2703,28 @@ function notify(notification: Notification | null) { // records the active member here before dispatching its turn. const groupSpeakers = new Map(); +// Usage attribution is fixed at the provider boundary. A queued message or +// calendar occurrence can update the thread's next trigger while this member +// is still running; the terminal event must keep the speaker, model and +// initiator that actually dispatched this generation. +type GroupTurnUsageSnapshot = Readonly<{ + generation: string; + speaker: { botId: string; name: string; color: string }; + botId: string; + botName: string; + instanceId: string; + driverKind: string; + model: string; + trigger: UsageTrigger; +}>; +const groupTurnUsageSnapshots = new Map(); + +function clearGroupTurnUsageSnapshot(threadId: string, generation: string): void { + if (groupTurnUsageSnapshots.get(threadId)?.generation === generation) { + groupTurnUsageSnapshots.delete(threadId); + } +} + // The latest running token totals for the turn in flight on each thread. // Providers report cumulative-within-turn numbers; the final value is folded // into the task's tally when the turn settles. @@ -3676,9 +3702,12 @@ bus.subscribe((event: RuntimeEvent) => { lastReply.delete(event.threadId); const lastReported = turnUsage.get(event.threadId); turnUsage.delete(event.threadId); - // group turns run on the room's thread — the speaking bot's task - // tally is not the right home for a shared room's spend, so only - // 1:1 task turns are tallied for now. + // The driver's own per-turn figure (turn.completed.usage) is + // authoritative; a driver that only streams the running indicator + // falls back to its last value. Retries therefore settle once. + const tokens = event.usage ?? lastReported; + // A room turn has no private bot task to tally, but it still belongs in + // the durable workspace ledger below. Keep the per-task counter 1:1. if (bot) { const resourceOwner = turnResourceOwners.get(event.threadId); const generation = directTurnGenerationByThread.get(event.threadId); @@ -3709,11 +3738,8 @@ bus.subscribe((event: RuntimeEvent) => { drainDelegationWakes(); } }; - // bank what this turn spent before the bot broadcast carries the - // task list to every window. The driver's own per-turn figure - // (turn.completed.usage) is authoritative; a driver that only - // streams the running indicator falls back to its last value. - const tokens = event.usage ?? lastReported; + // Bank what this turn spent before the bot broadcast carries the + // task list to every window. store.addTaskUsage(bot.id, event.threadId, { input: tokens?.input, output: tokens?.output, @@ -3788,13 +3814,40 @@ bus.subscribe((event: RuntimeEvent) => { settleDirectTurn(); } } - const speaker = groupSpeakers.get(event.threadId); + const groupUsage = groupTurnUsageSnapshots.get(event.threadId); + if ( + !bot && + speaker && + groupUsage?.speaker === speaker && + turnResourceOwners.get(event.threadId)?.generation === groupUsage.generation + ) { + // Consume only this generation. Its member-turn finally block may + // run after a replacement has already published a newer snapshot. + clearGroupTurnUsageSnapshot(event.threadId, groupUsage.generation); + appendUsage(DATA_DIR, { + botId: groupUsage.botId, + botName: groupUsage.botName, + threadId: event.threadId, + instanceId: groupUsage.instanceId, + driverKind: groupUsage.driverKind, + model: groupUsage.model, + input: tokens?.input ?? 0, + output: tokens?.output ?? 0, + ...(typeof tokens?.cachedInput === "number" ? { cachedInput: tokens.cachedInput } : {}), + costUsd: event.cost ?? null, + trigger: groupUsage.trigger, + }); + // This runs in the main subscriber before the member-turn waiter + // resumes, so the very next room dispatch observes the settled cost. + noteSpend(DATA_DIR, event.cost ?? null); + } + const settlingSpeaker = groupSpeakers.get(event.threadId); const group = store.groupByThread(event.threadId); - if (speaker && group?.busyBotId === speaker.botId) { + if (settlingSpeaker && group?.busyBotId === settlingSpeaker.botId) { releaseTurnResources(turnResourceOwners.get(event.threadId)); groupSpeakers.delete(event.threadId); store.patchGroup(group.id, { busyBotId: null, unread: true }); - const speakingBot = store.bot(speaker.botId); + const speakingBot = store.bot(settlingSpeaker.botId); if (speakingBot?.busy) { store.setActivity(speakingBot.id, "idle"); retryDelegationsWaitingOn(speakingBot.id); @@ -5538,11 +5591,12 @@ routines = new RoutineManager({ startTurn: (botId, threadId, prompt, runOn, triggerSource, onDispatchError) => startTurn(botId, prompt, { threadId, runOn, automationSource: triggerSource, onDispatchError }) .then(() => undefined), - startGoal: async (groupId, threadId, prompt, coordinatorBotId, runId, _onDispatchError) => { + startGoal: async (groupId, threadId, prompt, coordinatorBotId, runId, routine, _onDispatchError) => { startGroupTurn(groupId, prompt, undefined, undefined, "goal", undefined, { threadId, goalCoordinatorBotId: coordinatorBotId, goalRunId: runId, + routineTrigger: { kind: "routine", routineId: routine.routineId, label: routine.routineName }, }); }, interruptTurn: async (botId, threadId, runOn) => { @@ -5876,6 +5930,7 @@ type GroupMemberTurnOutcome = | "settled" | "provider_failed" | "dispatch_failed" + | "spend_cap" | "stalled" | "timed_out" | "cancelled" @@ -5976,6 +6031,27 @@ async function runGroupMemberTurn( ? group.threadId === threadId : Boolean(group && store.groupTaskByThread(group.id, threadId)); if (!group || !bot || !ownsThread) return false; + const spendCapReached = () => { + try { + assertWithinBudget(cfg, DATA_DIR); + return false; + } catch (error) { + const message = error instanceof Error ? error.message : String(error); + store.appendMessage(threadId, { + role: "bot", + kind: "activity", + from: { botId: bot.id, name: bot.name, color: bot.color }, + tool: { name: `error: ${message.slice(0, 140)}`, ok: false }, + }); + onDispatchError?.(message); + if (orchestration) { + orchestration.result.outcome = "spend_cap"; + orchestration.result.stopReason = message; + } + return true; + } + }; + if (spendCapReached()) return false; if (bot.approvalGrant) { onDispatchError?.(`${bot.name}'s approval level is still being confirmed — skipped this round`); return true; @@ -6371,6 +6447,23 @@ async function runGroupMemberTurn( } return false; } + // Setup above can yield while another bot settles and reaches the cap. + // Re-check at the actual provider boundary so room members, chained + // mentions, goal steps, retries, queued sends and calendar calls all obey + // the same workspace limit immediately before dispatch. + if (spendCapReached()) return false; + const dispatchSpeaker = roomSpeaker; + if (!dispatchSpeaker) return false; + groupTurnUsageSnapshots.set(threadId, { + generation: internalGeneration, + speaker: dispatchSpeaker, + botId: readyBot.id, + botName: readyBot.name, + instanceId: readyBot.modelSelection.instanceId, + driverKind: instance.driverKind, + model: readyBot.modelSelection.model, + trigger: copyUsageTrigger(operation?.usageTrigger ?? { kind: "owner" }), + }); let replyText = ""; let providerTurnId: string | undefined; let abandoned = false; @@ -6650,6 +6743,7 @@ async function runGroupMemberTurn( } finally { // Covers connector/setup failures, cancellation before dispatch, and all // other early returns that never produce a provider terminal event. + clearGroupTurnUsageSnapshot(threadId, internalGeneration); revokeInternalCapabilityGeneration(threadId, internalGeneration); if (!retainRoomVmLease) releaseRoomVmLease(); if (!providerDispatched && roomSpeaker && groupSpeakers.get(threadId) === roomSpeaker) { @@ -6738,6 +6832,7 @@ async function runGroupGoalStep(args: { if (coordinatorTurn && !coordinatorTurn.turnId) coordinatorTurn.turnId = turnId; }, }, + args.operation, ); if (result.outcome === "busy") continue; // One retry for a transient provider failure: a 13-turn goal must not @@ -6833,6 +6928,15 @@ async function runGroupGoalOperation(args: { }), }); if (args.operation.cancelled) return; + if (coordinatorResult.outcome === "spend_cap") { + finishGroupGoalRun( + args.groupId, + args.operation, + "blocked", + coordinatorResult.stopReason ?? "This workspace has reached its monthly spend limit.", + ); + return; + } if (coordinatorResult.outcome === "unavailable") { finishGroupGoalRun(args.groupId, args.operation, "blocked", `${args.coordinator.name} is not available.`); return; @@ -6918,6 +7022,15 @@ async function runGroupGoalOperation(args: { }), }); if (args.operation.cancelled) return; + if (workerResult.outcome === "spend_cap") { + finishGroupGoalRun( + args.groupId, + args.operation, + "blocked", + workerResult.stopReason ?? "This workspace has reached its monthly spend limit.", + ); + return; + } if (workerResult.outcome === "unavailable") { finishGroupGoalRun(args.groupId, args.operation, "blocked", `${workerBot.name} is not available.`); return; @@ -6983,6 +7096,10 @@ type StartGroupTurnOptions = { /** The message came through the HTTP API with nothing to say a person * sent it (see Message.via). */ via?: "api"; + /** Origin captured when this message entered the room harness. */ + usageTrigger?: UsageTrigger; + /** A scheduled room goal always overrides the ambient/user origin. */ + routineTrigger?: Extract; }; function startGroupTurn( @@ -7074,10 +7191,14 @@ function startGroupTurn( return message; } + const usageTrigger = copyUsageTrigger( + options.routineTrigger ?? options.usageTrigger ?? { kind: "owner" }, + ); const operation = beginGroupTurnOperation( groupId, threadId, goalCoordinator ? [] : responders.map((responder) => responder.id), + usageTrigger, ); if (goalCoordinator) { const runId = options.goalRunId?.trim() || `goal-${Date.now().toString(36)}-${randomUUID()}`; @@ -7172,14 +7293,14 @@ function drainQueuedChannelSends(): void { const group = store.group(groupId); return group ? groupIsWorking(group) : false; }, - ({ groupId, threadId, text, replyToId, sendId, mode, id, via }) => { + ({ groupId, threadId, text, replyToId, sendId, mode, id, via, usageTrigger }) => { const group = store.group(groupId); const ownsThread = group?.dm ? group.threadId === threadId : Boolean(group && store.groupTaskByThread(group.id, threadId)); if (!group || !ownsThread) return; try { - startGroupTurn(groupId, text, resolveReplyTarget(threadId, replyToId), sendId, mode, id, { via }); + startGroupTurn(groupId, text, resolveReplyTarget(threadId, replyToId), sendId, mode, id, { via, usageTrigger }); } catch (error) { store.appendMessage(threadId, { role: "bot", @@ -7232,7 +7353,9 @@ function deliverCalendarCall(call: CalendarCall, scheduledFor: number): void { const threadIds = new Set([group.threadId, ...(group.tasks ?? []).map((task) => task.threadId)]); const messages = [...threadIds].flatMap((threadId) => store.messagesFor(threadId)); if (messages.some((message) => message.sendId === sendId)) return; - startGroupTurn(group.id, text, undefined, sendId); + // Calendar calls are owner-scheduled work. Carry that origin on this exact + // operation even when paired-user messages are already waiting behind it. + startGroupTurn(group.id, text, undefined, sendId, "chat", undefined, { usageTrigger: { kind: "owner" } }); } function roomSetupPending(group: GroupRecord): boolean { @@ -7707,6 +7830,9 @@ function dispatchConnectorResume(entry: { botId: string; threadId: string; resum () => operation.cancelled, () => groupProviderHandshakeStarted(operation), () => groupProviderHandshakeSettled(operation), + undefined, + undefined, + operation, ); }); const tracked = next.finally(() => finishGroupTurnOperation(groupId, operation)); @@ -7852,6 +7978,9 @@ function dispatchSecretResume(entry: SecretResumeEntry) { () => operation.cancelled, () => groupProviderHandshakeStarted(operation), () => groupProviderHandshakeSettled(operation), + undefined, + undefined, + operation, ); }); const tracked = next.finally(() => finishGroupTurnOperation(groupId, operation)); @@ -11220,7 +11349,7 @@ const handleRequest = async (req: IncomingMessage, res: ServerResponse) => { return json(res, 400, { error: "threadId must be a task id" }); } const threadId = body.threadId ?? group.threadId; - noteTurnTrigger(threadId, auth); + const usageTrigger = noteTurnTrigger(threadId, auth); try { assertWithinBudget(cfg, DATA_DIR); } catch (error) { @@ -11285,10 +11414,11 @@ const handleRequest = async (req: IncomingMessage, res: ServerResponse) => { sendId, mode: channelMode, via, + usageTrigger, }); return { ok: true as const, queued: true as const, queueId: queued.id, threadId }; } - const message = startGroupTurn(current.id, text, replyTo, sendId, channelMode, undefined, { via }); + const message = startGroupTurn(current.id, text, replyTo, sendId, channelMode, undefined, { via, usageTrigger }); return { ok: true as const, threadId, message }; }, ); diff --git a/server/routines.test.ts b/server/routines.test.ts index c46ff756c6..6826f05ec2 100644 --- a/server/routines.test.ts +++ b/server/routines.test.ts @@ -33,6 +33,7 @@ function harness(start = new Date(2026, 7, 17, 8, 0, 0).getTime()) { prompt: string; coordinatorBotId: string; runId: string; + routine: { routineId: string; routineName: string }; onDispatchError: (message: string) => void; }> = []; const runOns: string[] = []; @@ -67,8 +68,8 @@ function harness(start = new Date(2026, 7, 17, 8, 0, 0).getTime()) { runOns.push(runOn); triggerSources.push(triggerSource); }, - startGoal: async (groupId, threadId, prompt, coordinatorBotId, runId, onDispatchError) => { - startedGoals.push({ groupId, threadId, prompt, coordinatorBotId, runId, onDispatchError }); + startGoal: async (groupId, threadId, prompt, coordinatorBotId, runId, routine, onDispatchError) => { + startedGoals.push({ groupId, threadId, prompt, coordinatorBotId, runId, routine, onDispatchError }); }, interruptTurn: async (botId, threadId, runOn) => { interruptedTurns.push({ botId, threadId, runOn }); @@ -1539,6 +1540,7 @@ describe("RoutineManager", () => { prompt: "Prepare and verify the launch", coordinatorBotId: "chief-1", runId: run.id, + routine: { routineId: routine.id, routineName: "Team launch" }, }); expect(run).toMatchObject({ status: "running", threadId: "goal-thread-1" }); expect(h.manager.listRoutines()[0]).toMatchObject({ target: "bot", groupId: undefined }); diff --git a/server/routines.ts b/server/routines.ts index 2486f762ed..0e12ea1933 100644 --- a/server/routines.ts +++ b/server/routines.ts @@ -246,6 +246,7 @@ export interface RoutineManagerOptions { prompt: string, coordinatorBotId: string, runId: string, + routine: Pick, onDispatchError: (message: string) => void, ) => Promise; interruptTurn?: (botId: string, threadId: string, runOn: RoutineRunOn) => Promise; @@ -1384,6 +1385,7 @@ export class RoutineManager { prompt, run.botId, run.id, + { routineId: run.routineId, routineName: run.routineName }, (message) => this.failThread(task.threadId, message), ); } else { diff --git a/server/spend-cap-api.test.ts b/server/spend-cap-api.test.ts index 1be9165028..2da5a7fe06 100644 --- a/server/spend-cap-api.test.ts +++ b/server/spend-cap-api.test.ts @@ -20,7 +20,13 @@ describe("spend cap and prices through real turns", () => { layerDir = mkdtempSync(join(tmpdir(), "omb-fake-layer-")); mkdirSync(join(layerDir, "server")); writeFileSync(join(layerDir, "server", "index.js"), 'export async function register() { return { customer: "Fixture Co", features: ["budgets", "billing"], expiresAt: "2099-01-01" }; }\n'); - session = await launchVerificationServer(process.env, undefined, undefined, undefined, { dir: layerDir, licenseKey: "fixture-key" }); + session = await launchVerificationServer( + { ...process.env, FAKE_CLAUDE_MODE: "slow" }, + undefined, + undefined, + undefined, + { dir: layerDir, licenseKey: "fixture-key" }, + ); }, 60_000); afterEach(async () => { @@ -32,7 +38,18 @@ describe("spend cap and prices through real turns", () => { const control = (args: string[]) => runControlOmb([...args, "--url", session.info.url]) as Promise; const api = (path: string, init: RequestInit = {}) => fetch(`${session.info.url}${path}`, init); const put = (body: unknown) => api("/api/config", { method: "PUT", headers: { "content-type": "application/json" }, body: JSON.stringify(body) }); + const post = (path: string, body: unknown, token?: string) => api(path, { + method: "POST", + headers: { "content-type": "application/json", ...(token ? { authorization: `Bearer ${token}` } : {}) }, + body: JSON.stringify(body), + }); const send = (botId: string, text: string) => api(`/api/bots/${encodeURIComponent(botId)}/messages`, { method: "POST", headers: { "content-type": "application/json" }, body: JSON.stringify({ text }) }); + const pair = async (label: string) => { + const opened = (await (await post("/api/auth/pairing", {})).json()) as { code: string }; + const paired = (await (await post("/api/auth/pair", { code: opened.code, label })).json()) as { token: string }; + expect(paired.token).toBeTypeOf("string"); + return paired.token; + }; it("refuses the next turn once the month reaches the cap, prices turns, and lets a raised cap through", async () => { const edition = (await (await api("/api/edition")).json()) as { edition: string; features: string[] }; @@ -72,4 +89,333 @@ describe("spend cap and prices through real turns", () => { expect(raised.budget).toMatchObject({ monthlyUsd: 1, exceeded: false }); expect(raised.total.turns).toBe(3); }, 150_000); + + it("books each room member and checks the cap again before the next member dispatch", async () => { + expect((await put({ budgets: { monthlyUsd: 0.005 } })).status).toBe(200); + const first = await control(["new-bot", "--name", "Room first"]); + const second = await control(["new-bot", "--name", "Room second"]); + const firstId = first.bot.id as string; + const secondId = second.bot.id as string; + const created = await post("/api/groups", { + name: "Budget room", + memberIds: [firstId, secondId], + setup: { bulletin: "", defaultResponder: { kind: "everyone" } }, + }); + expect(created.status).toBe(201); + const groupId = ((await created.json()) as any).group.id as string; + + expect((await post(`/api/groups/${groupId}/messages`, { text: "First bounded round" })).status).toBe(202); + await expect.poll(async () => { + const [usage, fleet] = await Promise.all([ + api("/api/usage").then((response) => response.json()) as Promise, + api("/api/bots?messages=50").then((response) => response.json()) as Promise, + ]); + const room = fleet.groups.find((candidate: { id: string }) => candidate.id === groupId); + const speakers = [...new Set(room?.messages + .filter((message: any) => message.role === "bot" && message.kind === "text" && message.from?.botId) + .map((message: any) => message.from.botId) ?? [])]; + const capErrors = room?.messages.filter((message: any) => + message.kind === "activity" && /monthly spend limit/i.test(message.tool?.name ?? "") + ).length ?? 0; + return { + working: room?.working, + turns: usage.total?.turns, + spentUsd: usage.budget?.spentUsd, + speakers, + capErrors, + }; + }, { timeout: 30_000 }).toEqual({ + working: false, + turns: 1, + spentUsd: 0.01, + speakers: [firstId], + capErrors: 1, + }); + + // A detached scheduled goal reaches the same dispatch guard. The cap is + // a blocked business condition, not a transient provider failure to retry. + const routine = await post("/api/routines", { + name: "Capped room goal", + prompt: "Coordinate work without exceeding the cap", + target: "room-goal", + groupId, + botId: firstId, + runOn: "maus", + schedule: { type: "once", at: Date.now() + 60_000 }, + durationMinutes: 30, + }); + expect(routine.status).toBe(201); + const routineId = ((await routine.json()) as any).routine.id as string; + const started = await post(`/api/routines/${routineId}/run`, {}); + expect(started.status).toBe(201); + const runId = ((await started.json()) as any).run.id as string; + await expect.poll(async () => { + const calendar = (await (await api("/api/routines")).json()) as any; + const run = calendar.runs.find((candidate: { id: string }) => candidate.id === runId); + return { status: run?.status, goalStatus: run?.goalStatus, error: run?.error }; + }, { timeout: 10_000 }).toEqual({ + status: "failed", + goalStatus: "blocked", + error: expect.stringMatching(/monthly spend limit/i), + }); + + // Raising the cap preserves normal room fan-out; both settled members are + // durable ledger rows attributed to their own bot. + expect((await put({ budgets: { monthlyUsd: 1 } })).status).toBe(200); + expect((await post(`/api/groups/${groupId}/messages`, { text: "Second complete round" })).status).toBe(202); + await expect.poll(async () => { + const usage = (await (await api("/api/usage?groupBy=bot")).json()) as any; + return { + turns: usage.total?.turns, + costUsd: usage.total?.costUsd, + byBot: usage.groups.map((entry: any) => [entry.key, entry.turns]), + }; + }, { timeout: 30_000 }).toEqual({ + turns: 3, + costUsd: 0.03, + byBot: expect.arrayContaining([ + [`bot:${firstId}`, 2], + [`bot:${secondId}`, 1], + ]), + }); + }, 150_000); + + it("attributes a room-goal member turn to its routine", async () => { + expect((await put({ budgets: { monthlyUsd: 1 } })).status).toBe(200); + const lead = await control(["new-bot", "--name", "Routine lead"]); + const leadId = lead.bot.id as string; + const created = await post("/api/groups", { + name: "Routine accounting room", + memberIds: [leadId], + setup: { bulletin: "", defaultResponder: { kind: "everyone" } }, + }); + expect(created.status).toBe(201); + const groupId = ((await created.json()) as any).group.id as string; + + const routine = await post("/api/routines", { + name: "Routine ledger attribution", + prompt: "Coordinate this bounded fixture goal", + target: "room-goal", + groupId, + botId: leadId, + runOn: "maus", + schedule: { type: "once", at: Date.now() + 60_000 }, + durationMinutes: 30, + }); + expect(routine.status).toBe(201); + const routineId = ((await routine.json()) as any).routine.id as string; + const started = await post(`/api/routines/${routineId}/run`, {}); + expect(started.status).toBe(201); + const runId = ((await started.json()) as any).run.id as string; + + // The generic fake reply is not a valid goal decision, so the run fails + // after one real coordinator turn. That settled turn still belongs to + // the routine even though its intermediate event is not a routine result. + await expect.poll(async () => { + const calendar = (await (await api("/api/routines")).json()) as any; + return calendar.runs.find((candidate: { id: string }) => candidate.id === runId)?.status; + }, { timeout: 30_000 }).toBe("failed"); + await expect.poll(async () => { + const usage = (await (await api("/api/usage?groupBy=user")).json()) as any; + return { turns: usage.total?.turns, groups: usage.groups }; + }, { timeout: 10_000 }).toEqual({ + turns: 1, + groups: [expect.objectContaining({ + key: `routine:${routineId}`, + label: "Routine: Routine ledger attribution", + turns: 1, + })], + }); + }, 150_000); + + it("keeps each user's attribution across two queued room turns", async () => { + expect((await put({ budgets: { monthlyUsd: 1 } })).status).toBe(200); + const alice = await pair("Alice device"); + const bob = await pair("Bob device"); + const carol = await pair("Carol device"); + const createdBot = await control(["new-bot", "--name", "Queue accountant"]); + const botId = createdBot.bot.id as string; + const created = await post("/api/groups", { + name: "Queued accounting room", + memberIds: [botId], + setup: { bulletin: "", defaultResponder: { kind: "everyone" } }, + }); + expect(created.status).toBe(201); + const groupId = ((await created.json()) as any).group.id as string; + + expect((await post(`/api/groups/${groupId}/messages`, { text: "Alice's turn" }, alice)).status).toBe(202); + // Slow mode emits an assistant item before its gated completion. Seeing + // it proves the first provider dispatch (and its usage snapshot) exists. + await expect.poll(async () => { + const fleet = (await (await api("/api/bots?messages=50")).json()) as any; + const room = fleet.groups.find((candidate: { id: string }) => candidate.id === groupId); + return { + working: room?.working, + hasInFlightReply: room?.messages.some((message: any) => message.role === "bot" && message.kind === "text"), + }; + }, { timeout: 30_000 }).toEqual({ working: true, hasInFlightReply: true }); + + const queued = await post(`/api/groups/${groupId}/messages`, { text: "Bob's queued turn" }, bob); + expect(queued.status).toBe(202); + expect(await queued.json()).toMatchObject({ queued: true }); + const queuedSecond = await post(`/api/groups/${groupId}/messages`, { text: "Carol's queued turn" }, carol); + expect(queuedSecond.status).toBe(202); + expect(await queuedSecond.json()).toMatchObject({ queued: true }); + + await expect.poll(async () => { + const [usage, fleet] = await Promise.all([ + api("/api/usage?groupBy=user").then((response) => response.json()) as Promise, + api("/api/bots?messages=50").then((response) => response.json()) as Promise, + ]); + const room = fleet.groups.find((candidate: { id: string }) => candidate.id === groupId); + return { + working: room?.working, + turns: usage.total?.turns, + byUser: usage.groups.map((entry: any) => [entry.key, entry.turns]), + }; + }, { timeout: 30_000 }).toEqual({ + working: false, + turns: 3, + byUser: expect.arrayContaining([ + ["user:alice device", 1], + ["user:bob device", 1], + ["user:carol device", 1], + ]), + }); + }, 150_000); + + it("keeps calendar turns owner-attributed when a paired user queues behind them", async () => { + expect((await put({ budgets: { monthlyUsd: 1 } })).status).toBe(200); + const user = await pair("Dana device"); + const first = await control(["new-bot", "--name", "Calendar queue first"]); + const second = await control(["new-bot", "--name", "Calendar queue second"]); + const firstId = first.bot.id as string; + const secondId = second.bot.id as string; + + const call = await post("/api/calendar-calls", { + name: "Calendar queue attribution", + description: "Keep this occurrence attributed to its owner", + botIds: [firstId, secondId], + schedule: { type: "once", at: Date.now() - 100 }, + durationMinutes: 5, + }); + expect(call.status).toBe(201); + const callId = ((await call.json()) as any).call.id as string; + let roomId: string | undefined; + // The first calendar member has dispatched but not completed. Queueing a + // paired send now must not retag the calendar operation's second member. + await expect.poll(async () => { + const fleet = (await (await api("/api/bots?messages=50")).json()) as any; + const room = fleet.groups.find((candidate: any) => candidate.messages?.some( + (message: any) => message.sendId?.startsWith(`calendar_${callId}_`), + )); + if (room) roomId = room.id; + return Boolean( + room?.working && + room.messages.some((message: any) => message.role === "bot" && message.kind === "text"), + ); + }, { timeout: 30_000 }).toBe(true); + expect(roomId).toBeTypeOf("string"); + + const queued = await post(`/api/groups/${roomId}/messages`, { text: "Dana's queued follow-up" }, user); + expect(queued.status).toBe(202); + expect(await queued.json()).toMatchObject({ queued: true }); + + await expect.poll(async () => { + const [usage, fleet] = await Promise.all([ + api("/api/usage?groupBy=user").then((response) => response.json()) as Promise, + api("/api/bots?messages=50").then((response) => response.json()) as Promise, + ]); + const room = fleet.groups.find((candidate: { id: string }) => candidate.id === roomId); + return { + working: room?.working, + turns: usage.total?.turns, + byUser: usage.groups.map((entry: any) => [entry.key, entry.turns]), + }; + }, { timeout: 30_000 }).toEqual({ + working: false, + turns: 4, + byUser: expect.arrayContaining([ + ["owner", 2], + ["user:dana device", 2], + ]), + }); + }, 150_000); + + it("does not let a due calendar call dispatch above the cap and books it normally after the cap is raised", async () => { + expect((await put({ budgets: { monthlyUsd: 0.005 } })).status).toBe(200); + const first = await control(["new-bot", "--name", "Calendar first"]); + const second = await control(["new-bot", "--name", "Calendar second"]); + const firstId = first.bot.id as string; + const secondId = second.bot.id as string; + + expect((await send(firstId, "Consume the initial allowance")).status).toBe(202); + expect(JSON.stringify(await control(["wait", "--bot", firstId, "--timeout", "30"]))).toContain("settled"); + await expect.poll(async () => ((await (await api("/api/usage")).json()) as any).total?.turns, { + timeout: 10_000, + }).toBe(1); + + const blockedCall = await post("/api/calendar-calls", { + name: "Blocked calendar call", + description: "Must not dispatch above the cap", + botIds: [firstId, secondId], + schedule: { type: "once", at: Date.now() - 100 }, + durationMinutes: 5, + }); + expect(blockedCall.status).toBe(201); + const blockedCallId = ((await blockedCall.json()) as any).call.id as string; + await expect.poll(async () => { + const [usage, fleet] = await Promise.all([ + api("/api/usage").then((response) => response.json()) as Promise, + api("/api/bots?messages=50").then((response) => response.json()) as Promise, + ]); + const room = fleet.groups.find((candidate: any) => candidate.messages?.some( + (message: any) => message.sendId?.startsWith(`calendar_${blockedCallId}_`), + )); + return { + roomCreated: Boolean(room), + working: room?.working, + turns: usage.total?.turns, + replies: room?.messages.filter((message: any) => message.role === "bot" && message.kind === "text").length ?? 0, + capErrors: room?.messages.filter((message: any) => + message.kind === "activity" && /monthly spend limit/i.test(message.tool?.name ?? "") + ).length ?? 0, + }; + }, { timeout: 30_000 }).toEqual({ + roomCreated: true, + working: false, + turns: 1, + replies: 0, + capErrors: 1, + }); + + expect((await put({ budgets: { monthlyUsd: 1 } })).status).toBe(200); + const allowedCall = await post("/api/calendar-calls", { + name: "Allowed calendar call", + description: "Both members should answer and be booked", + botIds: [firstId, secondId], + schedule: { type: "once", at: Date.now() - 100 }, + durationMinutes: 5, + }); + expect(allowedCall.status).toBe(201); + const allowedCallId = ((await allowedCall.json()) as any).call.id as string; + await expect.poll(async () => { + const [usage, fleet] = await Promise.all([ + api("/api/usage?groupBy=bot").then((response) => response.json()) as Promise, + api("/api/bots?messages=50").then((response) => response.json()) as Promise, + ]); + const room = fleet.groups.find((candidate: any) => candidate.messages?.some( + (message: any) => message.sendId?.startsWith(`calendar_${allowedCallId}_`), + )); + const speakers = [...new Set(room?.messages + .filter((message: any) => message.role === "bot" && message.kind === "text" && message.from?.botId) + .map((message: any) => message.from.botId) ?? [])].sort(); + return { turns: usage.total?.turns, costUsd: usage.total?.costUsd, working: room?.working, speakers }; + }, { timeout: 30_000 }).toEqual({ + turns: 3, + costUsd: 0.03, + working: false, + speakers: [firstId, secondId].sort(), + }); + }, 150_000); });