Public: anyone can read this; the crew can change it.
A conversation turn that hits a rate limit rests the account that served it, as a run does #8222
Check · Review · Land · Live done — Landed and live in production.
mason/forge:mason/8212-a-conversation-turn-that-hits-a-rate-lim → main 2 commits fork 3 threads · 3 unresolved
4 files
changed +132
−40
src/crew/conversation.ts +29 −3
| @@ -17,7 +17,7 @@ import type { App } from "../app.ts"; | ||
| 17 | 17 | import { dailyLeft } from "../core/account_ceiling.ts"; |
| 18 | 18 | import { appendTurn, openConversation, removeTurn, setClaudeSession, type Conversation } from "../core/conversations.ts"; |
| 19 | 19 | import { WORDS_MAX } from "../core/fence.ts"; |
| 20 | import { PLATFORM, pickSubscription } from "../core/subscriptions.ts"; | |
| 20 | import { PLATFORM, pickSubscription, type Pick } from "../core/subscriptions.ts"; | |
| 21 | 21 | import { getUserById } from "../core/users.ts"; |
| 22 | 22 | import { mintViaToken, VIA_TOKEN_MAX_MS } from "../core/via_token.ts"; |
| 23 | 23 | import { id } from "../kernel/id.ts"; |
| @@ -25,6 +25,7 @@ import { err, ok, type Result } from "../kernel/result.ts"; | ||
| 25 | 25 | import { appendRunLog } from "../runs/log.ts"; |
| 26 | 26 | import { runClaude, type ClaudeOutcome } from "./claude.ts"; |
| 27 | 27 | import { CONVERSATION_BRIEF } from "./conversation_brief.ts"; |
| 28 | import { isRateLimited, takeLimit, whenText } from "./limits.ts"; | |
| 28 | 29 | import { credentialEnv, listenForSteer, steerRun } from "./runner.ts"; |
| 29 | 30 | import { sandboxPolicy } from "./sandbox.ts"; |
| 30 | 31 | |
| @@ -148,6 +149,20 @@ export function resumeInterrupted(app: App, options: ConversationOptions): Promi | ||
| 148 | 149 | return done; |
| 149 | 150 | } |
| 150 | 151 | |
| 152 | /** | |
| 153 | * The provider turned the turn away: the account rests and its window is | |
| 154 | * noted as a run's is (`takeLimit`). Returns the one line the person sees, | |
| 155 | * never the provider's words. The forge's own token cannot rest, so the next | |
| 156 | * message is accepted; the line says the window is full, not that anything rests. | |
| 157 | */ | |
| 158 | function limitLine(app: App, picked: Pick, message: string): string { | |
| 159 | const account = picked.kind === "ready" ? picked.subscription : undefined; | |
| 160 | const { until } = takeLimit(app.db, account?.id, message, "", new Date()); | |
| 161 | return account === undefined | |
| 162 | ? `The forge's own model window is full until ${whenText(until)}; try again then.` | |
| 163 | : `${account.name} is resting until ${whenText(until)}.`; | |
| 164 | } | |
| 165 | ||
| 151 | 166 | function begin( |
| 152 | 167 | app: App, |
| 153 | 168 | options: ConversationOptions, |
| @@ -183,9 +198,13 @@ function begin( | ||
| 183 | 198 | .run(runId, conversation.id, userId, subscriptionId, prompt, resumes, resumeSession ?? "", now.toISOString()); |
| 184 | 199 | live.set(conversation.id, runId); |
| 185 | 200 | const put = (role: "agent" | "tool", text: string): void => { |
| 186 | appendTurn(app.db, userId, conversation.id, { role, text: text.slice(0, 20_000) }); | |
| 201 | const stored = appendTurn(app.db, userId, conversation.id, { role, text: text.slice(0, 20_000) }); | |
| 202 | if (stored.ok && role === "agent") streamed.push({ id: stored.value.id, text }); | |
| 187 | 203 | }; |
| 188 | 204 | let lastText = ""; |
| 205 | // What the turn said as it went, by turn id, so a provider's refusal | |
| 206 | // printed as assistant text can be taken back. | |
| 207 | const streamed: { id: string; text: string }[] = []; | |
| 189 | 208 | // The policy only shapes a shell's sandbox; an MCP-only turn has no shell |
| 190 | 209 | // and no file tools (`claudeSettings` with `mcp` omits it), so there is |
| 191 | 210 | // nothing to enforce and an unenforced policy cannot fail the turn. |
| @@ -243,7 +262,14 @@ function begin( | ||
| 243 | 262 | } |
| 244 | 263 | app.db.query("DELETE FROM token WHERE id = ?1").run(minted.value.id); |
| 245 | 264 | const answer = outcome?.result ?? ""; |
| 246 | if (answer.trim() !== "" && answer !== lastText) put("agent", answer); | |
| 265 | if (outcome !== undefined && isRateLimited(outcome)) { | |
| 266 | // The provider's words are not shown: the streamed copy of them goes. | |
| 267 | const first = outcome.result.split("\n").find((l) => l.trim() !== "")?.trim() ?? ""; | |
| 268 | for (const turn of streamed) | |
| 269 | if (turn.text === outcome.result || (first !== "" && turn.text.includes(first))) removeTurn(app.db, userId, conversation.id, turn.id); | |
outdated
Streamed |
||
| 270 | put("agent", limitLine(app, picked, outcome.result)); | |
| 271 | } | |
| 272 | else if (answer.trim() !== "" && answer !== lastText) put("agent", answer); | |
| 247 | 273 | else if (answer.trim() === "" && lastText === "") put("agent", "I could not answer that just now."); |
| 248 | 274 | app.db |
| 249 | 275 | .query( |
src/crew/limits.ts +31 −1
| @@ -5,7 +5,9 @@ | ||
| 5 | 5 | * it will listen; the person who asked never has to press anything. |
| 6 | 6 | */ |
| 7 | 7 | |
| 8 | import type { LimitWindow } from "../core/standing.ts"; | |
| 8 | import type { Database } from "bun:sqlite"; | |
| 9 | import { noteLimit, type LimitWindow } from "../core/standing.ts"; | |
| 10 | import { PLATFORM, restSubscription } from "../core/subscriptions.ts"; | |
| 9 | 11 | import { isNetworkError } from "./network.ts"; |
| 10 | 12 | import type { ClaudeOutcome } from "./claude.ts"; |
| 11 | 13 | |
| @@ -98,6 +100,34 @@ function resetOf(message: string, now: Date): Date | undefined { | ||
| 98 | 100 | return nextWallTime(now, hour, minute, zone!, date); |
| 99 | 101 | } |
| 100 | 102 | |
| 103 | /** | |
| 104 | * The provider's refusal, taken in: the window it names is noted on the | |
| 105 | * account's standing, and a sealed account (`subscriptionId` given) rests | |
| 106 | * until the provider's time. The forge's own token (`undefined`) is noted | |
| 107 | * but never rested — nothing else would take its place. One place for a run | |
| 108 | * and a conversation turn alike. | |
| 109 | */ | |
| 110 | export function takeLimit( | |
| 111 | db: Database, | |
| 112 | subscriptionId: string | undefined, | |
| 113 | message: string, | |
| 114 | fallback: string, | |
| 115 | now: Date, | |
| 116 | ): { until: Date; limit: ReturnType<typeof limitOf>; said: string } { | |
| 117 | const said = (message.split("\n").find((l) => l.trim() !== "") ?? fallback).slice(0, 160); | |
| 118 | const until = resumeAt(message, now); | |
| 119 | const limit = limitOf(message, now); | |
| 120 | if (limit !== undefined) | |
| 121 | noteLimit(db, subscriptionId ?? PLATFORM, { | |
| 122 | window: limit.window, | |
| 123 | resetsAt: limit.resetsAt.toISOString(), | |
| 124 | observedAt: now.toISOString(), | |
| 125 | said, | |
| 126 | }); | |
| 127 | if (subscriptionId !== undefined) restSubscription(db, subscriptionId, until.toISOString(), said); | |
| 128 | return { until, limit, said }; | |
| 129 | } | |
| 130 | ||
| 101 | 131 | /** `12:04 UTC` — how a resume time reads on a card. */ |
| 102 | 132 | export function whenText(at: Date): string { |
| 103 | 133 | return `${at.toISOString().slice(11, 16)} UTC`; |
src/crew/runner.ts +9 −35
| @@ -123,6 +123,7 @@ import { | ||
| 123 | 123 | limitOf, |
| 124 | 124 | MAX_PAUSES, |
| 125 | 125 | resumeAt, |
| 126 | takeLimit, | |
| 126 | 127 | whenText, |
| 127 | 128 | } from "./limits.ts"; |
| 128 | 129 | import { clearStaleLocks, lockedOut } from "./locks.ts"; |
| @@ -133,7 +134,6 @@ import { parkedTwice } from "./parks.ts"; | ||
| 133 | 134 | import { heldNow, stuckOf } from "./held.ts"; |
| 134 | 135 | import { reserveOf } from "./reserve.ts"; |
| 135 | 136 | import { owed } from "./team_share.ts"; |
| 136 | import { noteLimit } from "../core/standing.ts"; | |
| 137 | 137 | import { pauseReason } from "./pacing.ts"; |
| 138 | 138 | import { grantLeft } from "../core/model_grant.ts"; |
| 139 | 139 | import { dailyLeft, reserveCeiling } from "../core/account_ceiling.ts"; |
| @@ -143,7 +143,6 @@ import { | ||
| 143 | 143 | MIN_RUN_USD, |
| 144 | 144 | pickSubscription, |
| 145 | 145 | PLATFORM, |
| 146 | restSubscription, | |
| 147 | 146 | type Pick, |
| 148 | 147 | } from "../core/subscriptions.ts"; |
| 149 | 148 | import { |
| @@ -1523,30 +1522,17 @@ async function work( | ||
| 1523 | 1522 | // A limit that never lifts is not waited on forever. |
| 1524 | 1523 | let limitLifts: Date | undefined; |
| 1525 | 1524 | if (isRateLimited(result) && !signal.aborted) { |
| 1526 | const said = ( | |
| 1527 | result.result.split("\n").find((l) => l.trim() !== "") ?? result.reason | |
| 1528 | ).slice(0, 160); | |
| 1525 | const now = new Date(); | |
| 1529 | 1526 | // The account rests until the provider's time; another on the same |
| 1530 | 1527 | // grant, if one is awake, takes the run straight away. |
| 1528 | const { until, limit, said } = takeLimit( | |
| 1529 | app.db, | |
| 1530 | picked.kind === "ready" ? picked.subscription.id : undefined, | |
| 1531 | result.result, | |
| 1532 | result.reason, | |
| 1533 | now, | |
| 1534 | ); | |
| 1531 | 1535 | if (picked.kind === "ready") { |
| 1532 | const now = new Date(); | |
| 1533 | const until = resumeAt(result.result, now); | |
| 1534 | // Which window is full and when it resets, in the provider's own | |
| 1535 | // words, for the account's standing (`core/standing.ts`). | |
| 1536 | const limit = limitOf(result.result, now); | |
| 1537 | if (limit !== undefined) | |
| 1538 | noteLimit(app.db, picked.subscription.id, { | |
| 1539 | window: limit.window, | |
| 1540 | resetsAt: limit.resetsAt.toISOString(), | |
| 1541 | observedAt: now.toISOString(), | |
| 1542 | said, | |
| 1543 | }); | |
| 1544 | restSubscription( | |
| 1545 | app.db, | |
| 1546 | picked.subscription.id, | |
| 1547 | until.toISOString(), | |
| 1548 | said, | |
| 1549 | ); | |
| 1550 | 1536 | // Only a look at who could take over: no reservation is made here. |
| 1551 | 1537 | // The run pauses, and on resume it picks again with its reserve. |
| 1552 | 1538 | const next = pickSubscription( |
| @@ -1571,18 +1557,6 @@ async function work( | ||
| 1571 | 1557 | costUsd: result.costUsd, |
| 1572 | 1558 | }; |
| 1573 | 1559 | } |
| 1574 | } else { | |
| 1575 | // The platform's token is not rested — nothing else would take the | |
| 1576 | // run — but its window is noted, so its standing reads true. | |
| 1577 | const now = new Date(); | |
| 1578 | const limit = limitOf(result.result, now); | |
| 1579 | if (limit !== undefined) | |
| 1580 | noteLimit(app.db, PLATFORM, { | |
| 1581 | window: limit.window, | |
| 1582 | resetsAt: limit.resetsAt.toISOString(), | |
| 1583 | observedAt: now.toISOString(), | |
| 1584 | said, | |
| 1585 | }); | |
| 1586 | 1560 | } |
| 1587 | 1561 | if (run.attempt < MAX_PAUSES) { |
| 1588 | 1562 | const at = resumeAt(result.result, new Date()); |
test/conversation-turn.test.ts +63 −1
| @@ -14,7 +14,7 @@ import { createApp, type App } from "../src/app.ts"; | ||
| 14 | 14 | import { clearAccountCeiling, daySpent, setAccountCeiling } from "../src/core/account_ceiling.ts"; |
| 15 | 15 | import { listTurns, openConversation } from "../src/core/conversations.ts"; |
| 16 | 16 | import { monthSpent } from "../src/core/model_grant.ts"; |
| 17 | import { PLATFORM } from "../src/core/subscriptions.ts"; | |
| 17 | import { addSubscription, getSubscription, PLATFORM, setGrants, wakeSubscription } from "../src/core/subscriptions.ts"; | |
| 18 | 18 | import { verifyToken } from "../src/core/tokens.ts"; |
| 19 | 19 | import { createUser, type User } from "../src/core/users.ts"; |
| 20 | 20 | import { TOOLS } from "../src/crew/claude.ts"; |
| @@ -217,4 +217,66 @@ describe("a conversation turn", () => { | ||
| 217 | 217 | const last = app.db.query<{ state: string }, [string]>("SELECT state FROM conversation_run WHERE user_id = ?1 ORDER BY rowid DESC LIMIT 1").get(alice.id); |
| 218 | 218 | expect(last?.state).toBe("failed"); |
| 219 | 219 | }); |
| 220 | ||
| 221 | /** A `claude` the provider turns away: it answers a session limit and nothing else. */ | |
| 222 | function limitedStub(name: string): string { | |
| 223 | const path = join(dir, name); | |
| 224 | writeFileSync( | |
| 225 | path, | |
| 226 | `#!/bin/sh | |
| 227 | if [ "$1" = "--version" ]; then echo "2.1.0 (fake)"; exit 0; fi | |
| 228 | IFS= read -r first | |
| 229 | echo '{"type":"system","subtype":"init","session_id":"s-limit","model":"fake"}' | |
| 230 | echo '{"type":"assistant","session_id":"s-limit","message":{"id":"m1","content":[{"type":"text","text":"You have hit your session limit · resets 7:20am (UTC)"}],"usage":{"input_tokens":1,"output_tokens":1}}}' | |
| 231 | echo '{"type":"result","subtype":"success","is_error":true,"result":"You have hit your session limit · resets 7:20am (UTC)","total_cost_usd":0,"num_turns":1,"session_id":"s-limit","terminal_reason":"api_error","usage":{"input_tokens":1,"output_tokens":1}}' | |
| 232 | `, | |
| 233 | ); | |
| 234 | chmodSync(path, 0o755); | |
| 235 | return path; | |
| 236 | } | |
| 237 | ||
| 238 | test("a turn the provider rate-limits rests the account, says so once, and the next message is refused while it rests", async () => { | |
| 239 | const account = addSubscription(app.db, dir, { name: "home", token: "sk-ant-api-test-home" }); | |
| 240 | if (!account.ok) throw new Error("account"); | |
| 241 | setGrants(app.db, alice.id, "", [account.value.id]); | |
| 242 | try { | |
| 243 | const said = say(app, options(limitedStub("limited")), alice.id, "hello?"); | |
| 244 | expect(said.ok).toBe(true); | |
| 245 | if (said.ok) await said.value.done; | |
| 246 | const line = turns(alice).at(-1)!; | |
| 247 | expect(line).toMatch(/^agent:home is resting until \d\d:\d\d UTC\.$/); | |
| 248 | expect(turns(alice).filter((t) => t.includes("session limit"))).toEqual([]); | |
| 249 | const rested = getSubscription(app.db, account.value.id)!; | |
| 250 | expect(rested.restingUntil).not.toBeNull(); | |
| 251 | const noted = app.db.query<{ span: string }, [string]>("SELECT span FROM subscription_limit WHERE subscription_id = ?1").all(account.value.id); | |
| 252 | expect(noted.map((r) => r.span)).toEqual(["5h"]); | |
| 253 | ||
| 254 | const before = turns(alice).length; | |
| 255 | const refused = say(app, options(stub("while-resting", "s", "answer")), alice.id, "again?"); | |
| 256 | expect(refused.ok ? "started" : refused.error).toBe("resting"); | |
| 257 | expect(turns(alice)).toHaveLength(before); | |
| 258 | ||
| 259 | wakeSubscription(app.db, account.value.id); | |
| 260 | const woke = say(app, options(stub("after-wake", "s", "answer")), alice.id, "now?"); | |
| 261 | expect(woke.ok).toBe(true); | |
| 262 | if (woke.ok) await woke.value.done; | |
| 263 | expect(turns(alice).at(-1)).toBe("agent:All quiet."); | |
| 264 | } finally { | |
| 265 | setGrants(app.db, alice.id, "", []); | |
| 266 | } | |
| 267 | }); | |
| 268 | ||
| 269 | test("a turn on the forge's own token says its window is full, rests nothing, and notes the window", async () => { | |
| 270 | const said = say(app, options(limitedStub("platform-limited")), alice.id, "on the platform?"); | |
| 271 | expect(said.ok).toBe(true); | |
| 272 | if (said.ok) await said.value.done; | |
| 273 | expect(turns(alice).at(-1)).toMatch(/^agent:The forge's own model window is full until \d\d:\d\d UTC; try again then\.$/); | |
| 274 | expect(turns(alice).filter((t) => t.includes("session limit"))).toEqual([]); | |
| 275 | const noted = app.db.query<{ span: string }, [string]>("SELECT span FROM subscription_limit WHERE subscription_id = ?1").all(PLATFORM); | |
| 276 | expect(noted.map((r) => r.span)).toEqual(["5h"]); | |
| 277 | // Nothing rests, so the next message is accepted. | |
| 278 | const next = say(app, options(stub("platform-next", "s", "answer")), alice.id, "again?"); | |
| 279 | expect(next.ok).toBe(true); | |
| 280 | if (next.ok) await next.value.done; | |
| 281 | }); | |
| 220 | 282 | }); |
new above
offline
This duplicates the block in runner.ts (~1525-1545). Share one helper.