Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions changelog.d/1308-rate-limit-retry.fixed.md
Original file line number Diff line number Diff line change
@@ -0,0 +1 @@
A Bot turn the provider refused with a requests-per-window `429` (`Limit type: requests`, `Remaining: 0`) is now asked again after a backoff — 2 s, 8 s and 30 s — instead of spending the turn, and the group round or sender keeps waiting for the same exchange. A refusal that outlasts every attempt ends the turn once, in the visible failure line the chat already draws, naming the provider text. A budget or quota refusal is not asked again ([#1308](https://github.com/ThinkFlowLab/FlyBot/issues/1308)).
14 changes: 8 additions & 6 deletions core/src/host/delivery.js
Original file line number Diff line number Diff line change
Expand Up @@ -70,10 +70,11 @@ export function install(rt) {
return resolved.agent
}

const headerFor = (role, fromName, roomName, audience, note = 'new message', deferred = false, { resumed = false, retried = false, answeredIn } = {}) => {
const headerFor = (role, fromName, roomName, audience, note = 'new message', deferred = false, { resumed = false, retried = false, rateLimited = false, answeredIn } = {}) => {
if (answeredIn !== undefined) return `[The user answered your question card in the group "${answeredIn}" · your reply goes back to ${fromName} there]`
if (deferred) return `[Message from ${fromName} · you read it during your last task; answer it now]`
if (retried && role !== 'room') return '[Your last reply wrote a tool call as text, so nothing ran]'
if (rateLimited && role !== 'room') return '[Your last turn was cut off by the provider\'s rate limit · the same request again]'
if (resumed && role === 'resume') return '[DeepSeek Harness restarted while you were working]'
if (resumed && role !== 'room') return `[Message from ${fromName} · DeepSeek Harness restarted while you were working on it]`
if (role === 'reply') return `[Reply from ${fromName}]`
Expand Down Expand Up @@ -165,7 +166,7 @@ export function install(rt) {
// group turn (groups.js). `fromRoomId`: the group whose Session the sender wrote
// from; the reply goes back to that Session (turns.js). `answeredIn`: the group where the user
// answered the Bot's question card about work handed over from there (tools.js, #550).
async function deliver({ fromId, toId, sessionId, text, role, hop = 0, replyTo, roomId, audience, note, exchangeId = randomUUID(), steer = false, deferred = false, resumed = false, retried = false, resent = false, askedBy, fromRoomId, answeredIn, fromName: givenName, untrusted = false, remote, attachments, origin: givenOrigin, taskDrive = false }) {
async function deliver({ fromId, toId, sessionId, text, role, hop = 0, replyTo, roomId, audience, note, exchangeId = randomUUID(), steer = false, deferred = false, resumed = false, retried = false, rateLimited = false, resent = false, askedBy, fromRoomId, answeredIn, fromName: givenName, untrusted = false, remote, attachments, origin: givenOrigin, taskDrive = false }) {
const from = rt.botOf(fromId)
const to = rt.botOf(toId)
if (to === undefined) throw new Error(`Unknown Bot ${toId}`)
Expand All @@ -176,12 +177,12 @@ export function install(rt) {
const targetSessionId = sessionId ?? rt.chatOf(toId)
// Only a fresh Bot-to-Bot line is judged: deferred / resumed / retried / resent carry on a
// message this Session already has, so dropping one would lose it instead of a duplicate.
if (fromId !== undefined && !remote?.handle && !deferred && !resumed && !retried && !resent && duplicateDelivery(fromId, targetSessionId, text)) {
if (fromId !== undefined && !remote?.handle && !deferred && !resumed && !retried && !rateLimited && !resent && duplicateDelivery(fromId, targetSessionId, text)) {
rt.warn?.('ds-bot: delivery: dedupe %s -> %s', fromId, targetSessionId)
return exchangeId
}
const origin = originOf({ from, to, fromId, remote, given: givenOrigin })
const firstContact = remote?.handle && !deferred && !resumed && !retried && !resent ? await noteFriendLine(to, sessionId, roomId, remote.handle) : false
const firstContact = remote?.handle && !deferred && !resumed && !retried && !rateLimited && !resent ? await noteFriendLine(to, sessionId, roomId, remote.handle) : false
// Session events reject `undefined` members, so optional fields are omitted instead.
const source = {
kind: SOURCE_KIND,
Expand All @@ -199,6 +200,7 @@ export function install(rt) {
...(deferred ? { deferred: true } : {}),
...(resumed ? { resumed: true } : {}),
...(retried ? { retried: true } : {}),
...(rateLimited ? { rateLimited: true } : {}),
...(askedBy ? { askedBy } : {}),
// A board wake delivered as a room turn (#1182). A round's own room turn does not set this.
...(taskDrive ? { taskDrive: true } : {}),
Expand All @@ -210,7 +212,7 @@ export function install(rt) {
// A remote member's files, as references the chat shows as chips (fetched on demand).
...(remote?.handle && Array.isArray(attachments) && attachments.length ? { attachments: attachments.map(file => ({ id: String(file.id ?? ''), name: String(file.name), size: Number(file.size), mime: String(file.mime ?? '') })) } : {}),
}
if (EXCHANGE_ROLES.has(role) && !deferred && !resumed && !retried && !resent) {
if (EXCHANGE_ROLES.has(role) && !deferred && !resumed && !retried && !rateLimited && !resent) {
const exchange = { id: exchangeId, from: fromId, to: toId, text, role, time: Date.now(), ...(replyTo ? { replyTo } : {}), ...(audience === 'user' ? { audience } : {}) }
await recordOrWarn(`the ${role} exchange`, () => rt.mutate(() => { state.exchanges.push(exchange) }))
}
Expand All @@ -220,7 +222,7 @@ export function install(rt) {
// queues for the next turn (see turns.js for how each is routed).
const body = fromId ? stripSessionChips(text) : text
const message = createUserMessage({
content: [{ type: 'text', text: `${headerFor(role, fromName, room?.name, audience, note, deferred, { resumed, retried, answeredIn })}\n${untrusted ? fenceUntrusted(fromName, body) : body}` }],
content: [{ type: 'text', text: `${headerFor(role, fromName, room?.name, audience, note, deferred, { resumed, retried, rateLimited, answeredIn })}\n${untrusted ? fenceUntrusted(fromName, body) : body}` }],
source,
})
if (steer) agent.steer(message)
Expand Down
78 changes: 75 additions & 3 deletions core/src/host/turns.js
Original file line number Diff line number Diff line change
Expand Up @@ -101,6 +101,53 @@ export function install(rt) {
})
}

// A provider rate limit that clears on its own — the requests-per-window 429 (#1308) — is
// asked again instead of spending the turn: the wait grows with each attempt and the
// provider's truncated `reset` is never parsed. A budget or quota refusal is not this:
// the same key gets the same refusal. A limit that outlasts every attempt ends the turn
// once, in the visible failure line `noteEnded` already records (#1044).
const RATE_LIMIT_TEXT = 'The provider rate limited the last call (requests per window), so it did not run. Do the same work again now.'
const RATE_LIMIT_BACKOFF_MS = Array.isArray(rt.config?.rateLimitBackoffMs) ? rt.config.rateLimitBackoffMs : [2_000, 8_000, 30_000]
const RATE_LIMIT_CODES = new Set(['RATE_LIMIT', 'rate_limited', 'throttling_error'])
const QUOTA_TEXT = /\bbudget\b|\bquota\b|budget_exceeded|insufficient_?quota/i
const rateLimitFailure = (error) => {
const code = String(error?.code ?? '')
const status = Number(error?.status ?? error?.statusCode)
const message = String(error?.message ?? '')
if (status === 402 || code === 'QUOTA' || code === 'ACCOUNT_QUOTA' || QUOTA_TEXT.test(message)) return false
return status === 429 || RATE_LIMIT_CODES.has(code)
}
// Attempts already spent on the turn running now, per Session. A turn that ends without a
// rate limit clears it.
const rateLimitRetries = new Map()
const forgetRateLimit = (id) => {
const state = rateLimitRetries.get(id)
if (state?.timer !== undefined) clearTimeout(state.timer)
rateLimitRetries.delete(id)
}
// The same trigger, asked again: a group round keeps its waiter, a sender keeps its
// exchange, and anything else is a nudge in the Bot's own chat, as the written-tool-call
// retry does.
const retryRateLimited = (id, info, origin) => {
const keeps = info !== undefined && (info.role === 'room' || expectsReply(info))
return rt.deliver({
origin,
...(keeps && info.senderSessionId ? { fromId: info.senderSessionId } : {}),
toId: rt.selfOf(id),
sessionId: id,
role: keeps ? info.role : 'retry',
...(keeps && info.roomId ? { roomId: info.roomId } : {}),
hop: keeps ? info.hop ?? 0 : 0,
...(keeps && info.exchangeId ? { exchangeId: info.exchangeId } : {}),
...(keeps && info.audience ? { audience: info.audience } : {}),
...(keeps && info.askedBy ? { askedBy: info.askedBy } : {}),
...(keeps && info.senderRoomId ? { fromRoomId: info.senderRoomId } : {}),
note: 'the provider rate limited the last call, so it is repeated',
rateLimited: true,
text: RATE_LIMIT_TEXT,
})
}

/** @type {Map<string, {turn?:number, pending?:any, info?:any, origin?:string, next?:string, texts:string[], opened?:boolean, joined?:{info:any, from:number}[]}>} */
const live = new Map()
/** @type {Map<string, {sessionId:string, resolve:(text:string|null)=>void, fail:(failure:any)=>void, touch:()=>void, approval:()=>void}>} member turn waiters keyed by exchange id */
Expand Down Expand Up @@ -332,7 +379,7 @@ export function install(rt) {
// A turn that died before writing a line leaves the chat looking like a Bot that chose to stay
// quiet (#1044), so a failure with no text is marked `silent` and carries the streak of them at the
// same place: one blip and a provider that has been refusing for hours read differently.
const noteEnded = (id, turn, info, reason, failure, calls, texts = 0) => {
const noteEnded = (id, turn, info, reason, failure, calls, texts = 0, rateLimited = undefined) => {
if (failure === undefined && reason?.kind !== 'completed') return
const owner = rt.groupOwners.get(id)
const bot = rt.botOf(owner?.botId ?? rt.selfOf(id))
Expand All @@ -352,6 +399,7 @@ export function install(rt) {
at,
...(calls > 0 ? { calls } : {}),
...(failure ? { failed: (failure.message || failure.code).slice(0, 160) } : {}),
...(rateLimited !== undefined ? { rateLimited } : {}),
...(silent ? { silent: true, streak } : {}),
}], at)
}
Expand Down Expand Up @@ -565,16 +613,40 @@ export function install(rt) {
const reason = event.data?.reason
const failure = reason?.kind === 'error' && !switched ? turnFailure(reason.error) : undefined
if (failure !== undefined) noteTurnFailure(id, failure)
rt.scheduleRunEnded?.(id, { failed: failure === undefined ? undefined : failureLine(failure), aborted: reason?.kind === 'aborted' })
if (switched) break
// A rate limit the provider lifts on its own is not this turn's answer (#1308): the same
// trigger is asked again after a backoff, and only an attempt that outlasts them ends the
// turn. A stopped round turn is not asked again.
const spent = rateLimitRetries.get(id)?.spent ?? 0
const rateLimited = failure !== undefined && rateLimitFailure(reason.error)
&& !(info?.role === 'room' && stoppedRoomTurns.has(info.exchangeId))
if (!rateLimited) forgetRateLimit(id)
if (rateLimited && spent < RATE_LIMIT_BACKOFF_MS.length) {
const wait = RATE_LIMIT_BACKOFF_MS[spent]
rt.warn('ds-bot: the provider rate limited %s (attempt %d of %d), asking again in %d ms: %s', id, spent + 1, RATE_LIMIT_BACKOFF_MS.length + 1, wait, failureLine(failure).slice(0, 200))
const timer = setTimeout(() => {
rateLimitRetries.set(id, { spent: spent + 1 })
retryRateLimited(id, info, origin).catch((error) => {
rt.warn('ds-bot: the rate-limited turn for %s could not be asked again: %s', id, error)
forgetRateLimit(id)
noteEnded(id, answeredTurn, info, reason, failure, toolCalls, texts.length, spent + 1)
})
}, wait)
timer.unref?.()
rateLimitRetries.set(id, { spent, timer })
break
}
const exhausted = rateLimited ? spent + 1 : undefined
if (exhausted !== undefined) forgetRateLimit(id)
rt.scheduleRunEnded?.(id, { failed: failure === undefined ? undefined : failureLine(failure), aborted: reason?.kind === 'aborted' })
// A model that writes its tool call as text ends the turn as if it were done,
// though nothing ran. It gets one more turn on the same trigger, so a waiting
// round or sender gets the real answer; a second time, the text goes on without
// the call.
const wroteCall = failure === undefined && reason?.kind === 'completed' && lastCalled !== true && writesToolCall(lastText)
const retry = wroteCall && info?.retried !== true && !(info?.role === 'room' && stoppedRoomTurns.has(info.exchangeId))
if (retry) void retryWrittenCall(id, info, origin).catch(error => rt.warn('ds-bot: asking %s to run its written tool call failed: %s', id, error))
if (!retry) noteEnded(id, answeredTurn, info, reason, failure, toolCalls, texts.length)
if (!retry) noteEnded(id, answeredTurn, info, reason, failure, toolCalls, texts.length, exhausted)
const replyText = () => {
const reply = (wroteCall ? texts.map(stripWrittenCalls) : texts).join('\n\n').trim()
return wroteCall && reply === '' ? 'Tried to run a tool but wrote the call as text, so it did not run.' : reply
Expand Down
6 changes: 3 additions & 3 deletions core/test/group-status.test.js
Original file line number Diff line number Diff line change
Expand Up @@ -237,14 +237,14 @@ test('a member that fails after the round moved on shows the failure in the grou
const [approval] = (await host.call('state', {})).value.work.approvals
await host.call('approval-answer', { id: approval.id, choice: 'allow' })
await outcome
host.emit('gb', 'turn/end', { turn: 1, reason: { kind: 'error', error: { code: 'RATE_LIMIT', status: 429, message: 'Too many requests' } } })
host.emit('gb', 'turn/end', { turn: 1, reason: { kind: 'error', error: { code: 'INTERNAL', status: 500, message: 'boom' } } })
await settle(() => host.agentOf('r1').calls.length === 1)
const post = host.agentOf('r1').calls[0]
assert.deepEqual([post.source.late, post.source.quiet, post.source.failed], [true, true, { code: 'RATE_LIMIT', status: 429, message: 'Too many requests' }])
assert.deepEqual([post.source.late, post.source.quiet, post.source.failed], [true, true, { code: 'INTERNAL', status: 500, message: 'boom' }])

host.emit('r1', 'user/message', { content: post.content, source: post.source })
const round = JSON.parse(await host.tool('group_relay', {}, 'r1', { callId: 'c2', signal: new AbortController().signal }))
assert.deepEqual(round.replies.map(reply => [reply.kind, reply.name, reply.code, reply.status]), [['failed', 'Writer', 'RATE_LIMIT', 429]])
assert.deepEqual(round.replies.map(reply => [reply.kind, reply.name, reply.code, reply.status]), [['failed', 'Writer', 'INTERNAL', 500]])
assert.deepEqual(round.failed, ['Writer'])
assert.equal(host.agentOf('gb').calls.length, 1)
})
Expand Down
Loading
Loading