From 8b32c2e0f02723fe9e9aae08d4f2dd336e37e8df Mon Sep 17 00:00:00 2001 From: Day Dreamer <87640074+HugoXK@users.noreply.github.com> Date: Tue, 27 Oct 2026 20:49:08 +0800 Subject: [PATCH] fix(parts): ask the context steward again before moving a part on A per-minute rate limit on the steward's request read as "no answer" and moved a part that was still working: a fresh Session, the latest messages re-sent, a note written through the same limited key, and a handoff that said the previous part had grown long. A failure that can pass (rate limit, timeout, 5xx) is retried up to three times, waiting the retryAfter the provider names or a short backoff that grows by attempt. Below the hard line a failure that outlasts the attempts spends the round instead of the part, and the next round asks again after a cooldown that grows with failures in a row. Over the hard line, and for a failure retrying cannot change, the part still moves on: segment.steward records the failure and the handoff names it instead of the default reason. The note's own retry policy (NOTE_ATTEMPTS, NOTE_COOLDOWN_MS) is unchanged. Fixes #1447 --- changelog.d/1447-steward-retry.fixed.md | 1 + core/src/host/parts.js | 114 +++++++-- core/test/steward-retry.test.js | 230 ++++++++++++++++++ docs/subsystems/core-host.md | 1 + patches/ds-bot/0014-flybot-branding.patch | 159 +++++++++++- .../ds-bot/files/test/steward-retry.test.js | 230 ++++++++++++++++++ patches/ds-bot/topics.json | 3 +- 7 files changed, 707 insertions(+), 31 deletions(-) create mode 100644 changelog.d/1447-steward-retry.fixed.md create mode 100644 core/test/steward-retry.test.js create mode 100644 patches/ds-bot/files/test/steward-retry.test.js diff --git a/changelog.d/1447-steward-retry.fixed.md b/changelog.d/1447-steward-retry.fixed.md new file mode 100644 index 00000000..86861822 --- /dev/null +++ b/changelog.d/1447-steward-retry.fixed.md @@ -0,0 +1 @@ +A single throttled context-steward call no longer moves the conversation on to a fresh part: a rate limit, a timeout or a 5xx on that one request is now retried (up to three times, honouring the provider's `retryAfter`), and if it still fails below the point where the part must move, the round is spent instead of the part and the next round asks again only after a cooldown. Over that point, and for a refusal that retrying cannot change, the part still moves on, and the handoff now records the real reason instead of the default "the previous part grew long" ([#1447](https://github.com/ThinkFlowLab/FlyBot/issues/1447)). diff --git a/core/src/host/parts.js b/core/src/host/parts.js index 1e718558..0ce8ff13 100644 --- a/core/src/host/parts.js +++ b/core/src/host/parts.js @@ -159,6 +159,16 @@ export function install(rt) { const STEWARD_MAX_TOKENS = 4_096 const STEWARD_TIMEOUT_MS = 120_000 const STEWARD_LINES = 40 + // The steward decides whether the part moves, not how a note reads, so its retry is + // its own policy and not NOTE_ATTEMPTS / NOTE_COOLDOWN_MS above: a rate limit is + // counted per minute and recovers in seconds, and asking again costs one request + // instead of a whole part (#1447). A failure that outlasts these attempts below the + // hard line spends the round instead of the part, and the next round asks again + // after a wait that grows with each failure in a row. + const STEWARD_ATTEMPTS = 3 + const STEWARD_RETRY_MS = 250 + const STEWARD_RETRY_MAX_MS = 5_000 + const STEWARD_COOLDOWN_MS = [60_000, 300_000] const FOLDER_RETRY_MS = 300_000 // Making room before a step takes the turn's own signal and nothing else, so a // summary request stuck in retries parks the turn before its first step: no step, @@ -547,17 +557,75 @@ export function install(rt) { return parseDecision(assembler.blocks().filter(block => block.type === 'text').map(block => block.text).join('\n')) } + // A failure worth asking again: a rate limit, a timeout, a 5xx. Anything else — a + // request the provider refused, an answer that does not parse — is asked of the same + // key again and comes back the same. + const transientFailure = (error) => { + const code = String(error?.code ?? '') + if (code === 'throttling_error' || code === 'rate_limited' || code === 'TIMEOUT' || code === 'REQUEST_TIMEOUT' || code === 'ABORT_ERR') return true + const status = Number(error?.status ?? error?.statusCode) + return status === 429 || (status >= 500 && status <= 599) + } + + // The wait the provider asked for, capped, or a short backoff that grows by attempt. + const stewardRetryDelay = (error, attempt) => { + const asked = Number(error?.retryAfter ?? error?.retry_after) + if (Number.isFinite(asked) && asked > 0) return Math.min(asked * 1_000, STEWARD_RETRY_MAX_MS) + return STEWARD_RETRY_MS * attempt + } + const sleep = (ms) => new Promise(resolve => setTimeout(resolve, ms)) + + // A steward that could not be asked, per Session: how many rounds in a row, when to + // ask again, and the failure that a forced move writes into its segment (#1447). + const stewardFailures = new Map() + const stewardBlame = (sessionId) => stewardFailures.get(sessionId)?.text + // What the steward decides; a part that cannot be condensed, or a steward that - // gives no answer, moves on as it always could. - async function decide(agent, { midTurn }) { + // gives no usable answer, moves on as it always could. A rate limit is not an + // answer: it is asked again, and below the hard line a failure that outlasts the + // attempts costs the round, not the part. + async function decide(agent, { midTurn, hard = false }) { if (engineOf(agent) === undefined) return 'switch' - try { - const action = await askSteward(agent, { midTurn }) - if (action !== undefined) return action - rt.warn('ds-bot: the context steward gave no usable answer for %s; moving on to a fresh part', agent.id) - } catch (error) { - rt.warn('ds-bot: the context steward failed for %s; moving on to a fresh part: %s', agent.id, error) + let failure + let transient = false + for (let attempt = 1; attempt <= STEWARD_ATTEMPTS; attempt += 1) { + try { + const action = await askSteward(agent, { midTurn }) + if (action !== undefined) { + stewardFailures.delete(agent.id) + return action + } + rt.warn('ds-bot: the context steward gave no usable answer for %s; moving on to a fresh part', agent.id) + failure = undefined + transient = false + break + } catch (error) { + failure = error + transient = transientFailure(error) + if (!transient || attempt === STEWARD_ATTEMPTS) break + rt.warn('ds-bot: the context steward did not answer for %s (attempt %d of %d), asking again in %d ms: %s', agent.id, attempt, STEWARD_ATTEMPTS, stewardRetryDelay(error, attempt), error) + await sleep(stewardRetryDelay(error, attempt)) + } + } + if (failure !== undefined) { + stewardFailures.set(agent.id, { + count: 0, + until: 0, + text: `${failure?.code ?? failure?.status ?? 'error'}: ${String(failure?.message ?? failure).slice(0, 160)}`, + }) + } else { + // An answer that does not parse is not a failure to report later: the move it + // forces must not carry an older failure it did not cause. + stewardFailures.delete(agent.id) + } + if (transient && !hard) { + const count = (stewardFailures.get(agent.id)?.count ?? 0) + 1 + const waitMs = STEWARD_COOLDOWN_MS[Math.min(count, STEWARD_COOLDOWN_MS.length) - 1] + stewardFailures.set(agent.id, { ...stewardFailures.get(agent.id), count, until: Date.now() + waitMs }) + rt.warn('ds-bot: the context steward could not be asked for %s (%d in a row, %d attempts each); the part stays and is asked again in %ds: %s', agent.id, count, STEWARD_ATTEMPTS, waitMs / 1000, failure) + return 'wait' } + rt.warn('ds-bot: the context steward failed for %s; moving on to a fresh part: %s', agent.id, failure ?? 'no usable answer') return 'switch' } @@ -635,14 +703,16 @@ export function install(rt) { : `${head} It is the note written at the end of part ${segment.noteFrom}, carried over unchanged; anything newer is in the messages below, and read_own_chat finds the rest.)\n\n${segment.note}` } - function handoffPackage({ room, segment, lines, number, resume, overflow, failed, bare, folder }) { + function handoffPackage({ room, segment, lines, number, resume, overflow, failed, bare, folder, steward }) { const where = room ? `your conversation in the group chat "${room.name}"` : 'your own chat with the user' const lookup = room ? 'read_group_chat shows the group\'s latest messages, and read_own_chat your own chat.' : 'read_own_chat searches every part, reads around an anchor such as [part 2 #1234 · …], and shows the note that closed any earlier part.' const reason = failed !== undefined ? `a request in the previous part kept failing (${failed})` : overflow ? 'the previous part overflowed the model\'s window' - : folder !== undefined ? `your working folder changed to ${folder}` : 'the previous part grew long' + : folder !== undefined ? `your working folder changed to ${folder}` + : steward !== undefined ? `the context steward could not be asked whether to keep it (${steward})` + : 'the previous part grew long' const { recent, earlier, unlisted, complete } = lines const omitted = [ ...(unlisted > 0 ? [`(${unlisted} older messages are not listed.)`] : []), @@ -745,10 +815,10 @@ export function install(rt) { // `hard`: past the hard line or the window, the part moves on even without a note // from the model. `folder`: the part moves on because its working folder changed. - function startSwitch(agent, { resume = false, carry = [], info, joined, overflow = false, failed, bare = false, folder = false, hard = overflow || folder } = {}) { + function startSwitch(agent, { resume = false, carry = [], info, joined, overflow = false, failed, bare = false, folder = false, steward, hard = overflow || folder } = {}) { const place = placeOf(agent.id) if (place === undefined || switches.has(agent.id) || readOnly()) return false - const job = { ...place, carry: [...carry], info, joined, resume, overflow, failed, bare, folder, hard, origin: settled(rt, agent.id) } + const job = { ...place, carry: [...carry], info, joined, resume, overflow, failed, bare, folder, steward, hard, origin: settled(rt, agent.id) } switches.set(agent.id, job) void runSwitch(agent, job) .catch(error => rt.warn('ds-bot: %s did not move on to a fresh part: %s', rt.botOf(job.botId)?.name ?? agent.id, error)) @@ -867,7 +937,7 @@ export function install(rt) { } await rt.applyPermission(bot, freshId) .catch(error => rt.warn('ds-bot: permission for the fresh part of %s not applied: %s', bot.name, error)) - const segment = { sessionId: agent.id, endedAt: Date.now(), ...(tokens !== undefined ? { tokens } : {}), ...noted } + const segment = { sessionId: agent.id, endedAt: Date.now(), ...(tokens !== undefined ? { tokens } : {}), ...noted, ...(job.steward === undefined ? {} : { steward: job.steward }) } const resume = job.resume || job.overflow // Commit the new part before its handoff turn can run: a handoff visible to a failure handler // before the segment is saved would let a second switch lose the previous part's note. A failed @@ -930,7 +1000,7 @@ export function install(rt) { folderRetry.set(freshId, Date.now() + FOLDER_RETRY_MS) rt.warn('ds-bot: the fresh part of %s did not start in %s; it works in %s', bot.name, intended, cwd) } - const text = handoffPackage({ room, segment, lines: carried, number: number + 1, resume, overflow: job.overflow, failed: job.failed, bare: job.bare, folder }) + const text = handoffPackage({ room, segment, lines: carried, number: number + 1, resume, overflow: job.overflow, failed: job.failed, bare: job.bare, folder, steward: job.steward }) fresh.followup(createUserMessage({ content: [{ type: 'text', text }], source: handoffSource(job, resume) })) await ctx.sessions.flush(fresh.session) freshParts.set(freshId, estimateTokens(text)) @@ -985,8 +1055,10 @@ export function install(rt) { const lines = await contextLines(agent.session).catch(() => undefined) const tokens = tokensOf(agent.session) if (lines === undefined || tokens === undefined || tokens < lines.hard || !grown(agent.id, tokens, lines) || readOnly()) return next() - if (await decide(agent, { midTurn: payload.step > 1 }) === 'compact' && await condense(agent, lines, { inTurn: true, signal: payload.signal, keep: payload.messages })) return next() - return switchBeforeStep(agent, payload, { hard: true }, next) + // Before a step the part is already past the hard line, so a steward that cannot + // be asked moves the part on here; the failure still names itself in its segment. + if (await decide(agent, { midTurn: payload.step > 1, hard: true }) === 'compact' && await condense(agent, lines, { inTurn: true, signal: payload.signal, keep: payload.messages })) return next() + return switchBeforeStep(agent, payload, { hard: true, steward: stewardBlame(agent.id) }, next) }, { prepend: true }) // `messages` left the inbox for this step, so the switch carries them along. @@ -1071,15 +1143,19 @@ export function install(rt) { const tokens = tokensOf(agent.session) if (lines === undefined || tokens === undefined || tokens < lines.check || !grown(agent.id, tokens, lines)) return const hard = tokens >= lines.hard - if (!hard && Date.now() < (noteFailures.get(agent.id)?.until ?? 0)) return + // Below the hard line a part that could not be written or decided waits out its + // cooldown instead of asking again at every idle (#1447). + if (!hard && (Date.now() < (noteFailures.get(agent.id)?.until ?? 0) || Date.now() < (stewardFailures.get(agent.id)?.until ?? 0))) return checking.add(agent.id) try { - if (await decide(agent, { midTurn: false }) === 'compact') { + const decision = await decide(agent, { midTurn: false, hard }) + if (decision === 'wait') return + if (decision === 'compact') { const condensed = await condense(agent, lines, { inTurn: false }) if (condensed !== false) return } if (switches.has(agent.id)) return - startSwitch(agent, { hard: (tokensOf(agent.session) ?? tokens) >= lines.hard }) + startSwitch(agent, { hard: (tokensOf(agent.session) ?? tokens) >= lines.hard, steward: stewardBlame(agent.id) }) } finally { checking.delete(agent.id) } diff --git a/core/test/steward-retry.test.js b/core/test/steward-retry.test.js new file mode 100644 index 00000000..80d21fe1 --- /dev/null +++ b/core/test/steward-retry.test.js @@ -0,0 +1,230 @@ +import assert from 'node:assert/strict' +import { mkdirSync, mkdtempSync, readFileSync, realpathSync, rmSync, writeFileSync } from 'node:fs' +import { tmpdir } from 'node:os' +import { join } from 'node:path' +import { test } from 'node:test' +import { apply } from '../index.js' +import { fakeContext } from './fake-ctx.js' + +// A Main Bot and a Bot, both idle in their own chat. +const TEAM = { + version: 1, + mainBotIds: ['a'], + workspaceId: 'w1', + bots: { + a: { id: 'a', name: 'Chief', role: 'Boss', sessionId: 'a', createdAt: 1 }, + b: { id: 'b', name: 'Writer', role: 'Write', sessionId: 'b', createdAt: 2 }, + }, + rooms: {}, + agents: {}, + exchanges: [], + exchangeTurns: {}, + questions: {}, + prefs: {}, + revision: 0, +} + +// The provider's stream, in the shape `BlockAssembler` reads: text deltas and a finish. +const answer = (text) => (async function* () { + yield { type: 'text-delta', index: 0, text } + yield { type: 'finish', reason: { kind: 'stop' } } +})() + +// A rate limit as the provider reports it: per-minute, recovered in seconds, and the +// `retryAfter` it asks for. +const throttled = (retryAfter = 0) => Object.assign(new Error('429: rate limit exceeded (requests)'), { + status: 429, + code: 'throttling_error', + retryAfter, +}) + +// A provider error that is not a rate limit: asking again would change nothing. +const broken = () => Object.assign(new Error('500: the provider failed'), { status: 500, code: 'PROVIDER_ERROR' }) + +// One model window, and a part whose size the test sets: `lines` for window 2000 with a +// 100-token reserve are hard 1805 / check 1520, so `between` sits past the check line and +// short of the hard one, and `over` is past both. +const WINDOW = { context: { contextWindow: 2_000 }, defaultMaxTokens: 100 } +const between = 'x'.repeat(4_800) +const over = 'x'.repeat(5_600) + +// The team's data in a temporary home, a project folder beside it, a compaction engine +// that makes room when the steward says so, and a provider stream the test drives. +async function start({ text = between, stream } = {}) { + const root = realpathSync.native(mkdtempSync(join(tmpdir(), 'ds-bot-steward-retry-'))) + const dsh = join(root, 'dsh') + const home = join(dsh, 'bot') + const team = join(home, 'workspace') + const project = join(root, 'project') + for (const dir of [team, project]) mkdirSync(dir, { recursive: true }) + writeFileSync(join(home, 'state.json'), JSON.stringify(TEAM)) + + const messages = [{ role: 'user', content: [{ type: 'text', text }] }] + const condensed = [] + const compaction = { + compactIfNeeded: async () => null, + compactNow: async () => { + condensed.push('compactNow') + // What a summary leaves behind: the part is below the check line afterwards. + messages.splice(0, messages.length, { role: 'user', content: [{ type: 'text', text: 'condensed' }] }) + return { summary: 'condensed' } + }, + compactRegion: async () => {}, + summarize: async () => '', + } + + const ctx = fakeContext({ services: { compaction } }) + const created = [] + const cwds = new Map() + const agents = new Map() + const agentOf = (sessionId) => { + if (!agents.has(sessionId)) { + const calls = [] + agents.set(sessionId, { + id: sessionId, + session: { + id: sessionId, + header: { cwd: cwds.get(sessionId) ?? team }, + requestHeader: () => ({ config: { provider: 'deepseek', model: 'deepseek-v4-flash', maxTokens: 100 } }), + deriveMessages: () => messages, + toolHistory: () => [], + }, + calls, + status: 'idle', + followup(message) { calls.push(['followup', message]) }, + steer(message) { calls.push(['steer', message]) }, + inject(message) { calls.push(['inject', message]) }, + cancel() {}, + async whenIdle() {}, + async runMaintenance(fn) { return fn(AbortSignal.timeout(60_000)) }, + inbox: { nextStep: [], nextTurn: [], remove() {} }, + }) + } + return agents.get(sessionId) + } + const archived = [] + const registry = new Map([['w1', { id: 'w1', path: team, title: 'DS Bot' }]]) + const calls = [] + Object.assign(ctx, { + llm: { + listProviders: () => [], + listModels: async () => [], + resolveModelInfo: async () => WINDOW, + stream: (request) => { + calls.push(request) + return stream(calls.length, request) + }, + }, + workspaceRegistry: { + get: id => registry.get(id), + list: () => [...registry.values()], + resolveByPath: async path => [...registry.values()].find(workspace => workspace.path === path), + create: async path => ({ id: 'w-new', path }), + async archiveSession(sessionId) { archived.push(sessionId) }, + }, + sessionController: { + create: async (options) => { + const sessionId = `s${created.length + 1}` + created.push(options) + cwds.set(sessionId, options.cwd ?? registry.get(options.workspaceId)?.path) + return { sessionId } + }, + resolveAgent: async sessionId => ({ agent: agentOf(sessionId) }), + selectModel: async () => {}, + page: async () => ({ records: [], hasMore: false }), + }, + sessionTitle: { rename: async () => {} }, + sessions: { flush: async () => {} }, + }) + let route + const register = ctx.connection.fetch.register + ctx.connection.fetch.register = (options) => { + route = options + return register(options) + } + apply(ctx, { home }) + const call = async (endpoint, payload) => (await route.fetch(new Request('http://localhost/api/bot', { + method: 'POST', body: JSON.stringify({ endpoint, payload }), + }))).json() + const tool = (name, args, sessionId) => ctx.registeredTools.get(name).execute(args, { agent: agentOf(sessionId), name, callId: 'c1' }) + const stop = () => { + ctx.dispose() + rmSync(root, { recursive: true, force: true, maxRetries: 5 }) + } + // Loading once fills the Session-owner indexes. + await tool('list_bots', {}, 'a') + // The engine is adopted where DSH announces a new agent; the idle check reads it there. + const b = agentOf('b') + ctx.emit('agent/created', { agent: b }) + const idle = () => ctx.emit('agent/status', { agent: b, status: 'idle' }) + return { ctx, created, calls, condensed, archived, agentOf, call, idle, state: () => JSON.parse(readFileSync(join(home, 'state.json'), 'utf8')), stop } +} + +const settle = () => new Promise(resolve => setTimeout(resolve, 20)) +const until = async (check, ms = 5_000) => { + const end = Date.now() + ms + while (!check()) { + if (Date.now() > end) throw new Error('until: the condition did not hold within 5 s') + await settle() + } +} +const forMs = async (ms) => { + const end = Date.now() + ms + while (Date.now() < end) await settle() +} + +test('#1447: a throttled steward is asked again instead of moving the part on', async (t) => { + let stewardCalls = 0 + const h = await start({ + stream: (count) => { + if (count === 1) throw throttled(0) + stewardCalls += 1 + return answer('{"action":"compact","reason":"still working"}') + }, + }) + t.after(() => h.stop()) + h.idle() + await until(() => h.calls.length >= 2) + await forMs(200) + assert.equal(h.calls.length, 2, 'the steward was asked twice: once refused, once answered') + assert.equal(stewardCalls, 1, 'the second answer is the one that counts') + assert.deepEqual(h.condensed, ['compactNow'], 'the answer was acted on: the part was condensed in place') + assert.equal(h.created.length, 0, 'no fresh part was started for one rate limit') + assert.deepEqual(h.archived, [], 'nothing was archived') +}) + +test('#1447: below the hard line a steward that keeps being turned down costs the round, not the part', async (t) => { + const h = await start({ stream: () => { throw throttled(0) } }) + t.after(() => h.stop()) + h.idle() + await until(() => h.calls.length >= 3, 10_000) + await forMs(200) + assert.equal(h.calls.length, 3, 'the steward was asked the three attempts and no more') + assert.equal(h.created.length, 0, 'the part stays: nothing was started for it') + assert.deepEqual(h.archived, [], 'nothing was archived') + // The next idle inside the cooldown does not ask again: the same key is still limited. + h.idle() + await forMs(300) + assert.equal(h.calls.length, 3, 'the cooldown holds: the part waits without another request') + assert.equal(h.created.length, 0, 'and it still stays') +}) + +test('#1447: a steward that cannot be asked over the hard line still moves the part on and says why', async (t) => { + const h = await start({ text: over, stream: () => { throw broken() } }) + t.after(() => h.stop()) + h.idle() + await until(() => h.created.length === 1, 15_000) + assert.equal(h.created.length, 1, 'the part past the hard line moved on') + // The switch starts the fresh part first and archives the finished one once the move is + // done, so the archive lands a step after `created`: wait for it instead of reading it + // the moment the new Session exists. The assertion itself stays as strict. + await until(() => h.archived.length === 1, 15_000) + assert.deepEqual(h.archived, ['b'], 'the finished part was archived') + const packageText = (h.agentOf('s1').calls.find(([kind]) => kind === 'followup') ?? [])[1] + ?.content?.map(block => block.text).join('') + assert.match(packageText ?? '', /the context steward could not be asked/) + assert.doesNotMatch(packageText ?? '', /the previous part grew long/) + const stored = h.state().bots.b.segments?.at(-1) + assert.match(String(stored?.steward ?? ''), /PROVIDER_ERROR/, 'the segment keeps the failure that forced the move') +}) + diff --git a/docs/subsystems/core-host.md b/docs/subsystems/core-host.md index a722fa5a..9af3e06b 100644 --- a/docs/subsystems/core-host.md +++ b/docs/subsystems/core-host.md @@ -406,6 +406,7 @@ Work a member hands a Bot with `message_bot` from its group Session runs in a hi - A repeated Bot-to-Bot line becomes one turn (#641): `deliver()` keys the last 60 s of Bot-to-Bot text by sender + recipient Session + the text with its whitespace collapsed (`DEDUPE_MS`, `duplicateDelivery`), drops a repeat before it is recorded as an exchange or handed to the Session, and journals the drop as `delivery: dedupe -> ` in the host warning log. The key includes the Session, so the same line to another Session (a group copy, another Bot) is its own delivery; `deferred` / `resumed` / `retried` / `resent` carry on a message the Session already has and are never judged, and a friend's line, the user's own message and a group round are not Bot-to-Bot. Routing (parent and direct children) and the existing unread-queue refusal are unchanged, and the map is process-local and capped at 500 entries. Covered by `core/test/repeated-delivery.test.js`. - **The check line is 45% of the usable window, not 80%** (`steward.js` `linesFor`): a Bot's conversation is condensed once its estimated size passes `DEFAULT_CHECK_RATIO` of the model's declared window less the room its reply needs, so the conversation is condensed earlier than upstream ds-bot does. The share is a floor on the prompt of a long turn, because every call re-sends the fixed block and the history is most of each prompt (#796); the trade is a checkpoint sooner and less detail kept (#1233). `checkRatio` in the profile still sets the share, and the hard line stays at 95% of the usable window, so a share at or above 0.95 can never lift the check line above it. Covered by `core/test/steward.test.js`. - **A checkpoint is judged by the prompt that is left** (`parts.js` `pressureLeft`): the token meter stays anchored on the provider usage from before the span was replaced, and its surface delta only subtracts a 4-characters-per-token estimate of the removed text. A denser report — the mock's 3 characters per token, or CJK on a real model — can therefore still read above the check line after the checkpoint has replaced the span, and the part would move on even though the steward said to condense in place. After a compaction, the part stays when the heuristic size of the remaining surface plus the tool schemas is under the check line. A surface that is still over it moves on, as before. The decision to cross the line in the first place still uses the anchored total (#917). +- **The steward is asked again before a part moves** (`parts.js` `decide`, #1447): its one answer decides whether the part moves, and a rate limit on that request — a per-minute count on the same key the conversation uses, recovered in seconds — used to read as "no answer" and moved a part that was still working: a fresh Session, the latest messages re-sent, a note written through the same limited key, and a handoff whose reason said "the previous part grew long". A failure that can pass (rate limit, timeout, 5xx) is now retried up to `STEWARD_ATTEMPTS` (3), waiting the `retryAfter` the provider names (capped at `STEWARD_RETRY_MAX_MS`, 5 s) or a short backoff that grows by attempt; a success at any attempt is the answer. Below the hard line a failure that outlasts the attempts spends the round instead of the part: the part stays, and the next round asks again only after `STEWARD_COOLDOWN_MS` (60 s, then 300 s, counting failures in a row). Over the hard line, and for a failure that retrying cannot change, the part still moves on; `segment.steward` then records the failure and the handoff says `the context steward could not be asked whether to keep it (…)` rather than the default reason. A note's own retry (`NOTE_ATTEMPTS`, `NOTE_COOLDOWN_MS`) is a separate policy and unchanged. Covered by `core/test/steward-retry.test.js`. - **Making room before a step has its own deadline** (`parts.js` `condense`, #1285): the pre-step overflow compaction used to carry only the turn's own signal, so a summary request stuck in retries — a 429 storm retries each request five times with its backoff — parked the turn before its first step: no `step/start`, no `turn/end`, `load: 1` and the roster's `stalled` hint until a human called `stop_bot`. `COMPACT_DEADLINE_MS` (`config.compactDeadlineMs`, 300 000 ms) now bounds that call through `AbortSignal.any([signal, deadline])`. When the deadline is what ended the wait, `condense` warns `ds-bot: making room for %s ran out of %d ms before a step; the turn ends here and the next one tries again: …` and rethrows, so DSH's own `finally` appends `turn/end` and the turn fails normally. DSH claimed that step's input before the pre-step hook ran and does not hand it back for a turn that ends in an error, so before the throw the deadline branch queues the batch again (`keepClaimed`, `agent.inbox.append('next-turn', …)`, one ordinary queued message each, a copy the inbox already holds refused): nothing the turn was woken for is lost, and the next turn — which the loop starts straight away while input is pending — reads it and tries to make room again. A provider that stays down therefore fails one turn per deadline, back to back, each ending normally instead of parking at `load: 1`; no backoff is added here, the queue is the thing that carries the work forward. A turn the user aborted keeps the old path (warn, `false`, move on), and the `!signal?.aborted` half of the guard keeps a turn the user stopped in charge even when the deadline was already reached. The idle pass is unchanged on `NOTE_TIMEOUT_MS` (600 000 ms, one shared signal over both note attempts). The turn-ending waits stay where they were: steward 120 s, idle condense and the handoff note each at that same 600 s budget. Covered by `core/test/pre-step-compaction.test.js`. - **A part is committed before its handoff runs** (`parts.js` `moveOn`): the Bot's `sessionId` / `room.sessions`, the new segment and the `rescued` mark are saved first, so the turn the handoff starts can never see an uncommitted part — a failure handler that did would start a second switch against it and lose the previous part's note, and the rescue would not know it may only happen once. The handoff package is written next; if that fails, the pointer is rolled back (best effort saved, the fresh Session archived, the rescued mark dropped), so the old part is current and the finalizer hands the input on there. Two windows stay open, and both are written down rather than discovered later: a crash after that save and before the handoff package is flushed leaves the fresh part current with no package — the stopped turn's input still goes to it, so no work is lost, but the note is, and the rollback never runs; and the rollback's own save is best effort (`rt.save().catch(() => {})`), so a failing rollback leaves the file showing the fresh part while memory holds the old one, and the next start behaves like the first window. Neither is a reason to clear the pointer before the save: a note lost this way is a part without a handoff note, while clearing first loses the handoff package itself. The move is marked done only after the stopped turn's input and the old inbox messages are forwarded; a switch that never starts (no place, another switch, a read-only host) lets the step continue instead of rejecting its messages, which DSH has already taken out of the inbox. An armed scheduled task keeps the earlier Session active and DSH refuses to archive it, so the switch keeps the `moveSchedules` promise and waits for the move before archiving the earlier part; when the move failed and the refusal reports `schedule` as the only activity, the plain archive is retried with `{ stopActivity: true }` (the way a deletion does, `roster.js:450`), so a finished part is never left active with a task nobody sees; a refusal caused by a job, turn or subagent is not retried, because `stopActivity` would stop that too (#896). Covered by `core/test/work-dir.test.js`, `core/test/part-archive.test.js` and `core/test/bot-messages.test.js`. diff --git a/patches/ds-bot/0014-flybot-branding.patch b/patches/ds-bot/0014-flybot-branding.patch index 8f96ec39..18d94457 100644 --- a/patches/ds-bot/0014-flybot-branding.patch +++ b/patches/ds-bot/0014-flybot-branding.patch @@ -482,9 +482,20 @@ diff --git a/src/host/parts.js b/src/host/parts.js export function install(rt) { const { ctx, config, state } = rt -@@ -145,25 +145,53 @@ export function install(rt) { +@@ -144,26 +144,64 @@ export function install(rt) { + const STEWARD_MAX_TOKENS = 4_096 const STEWARD_TIMEOUT_MS = 120_000 const STEWARD_LINES = 40 ++ // The steward decides whether the part moves, not how a note reads, so its retry is ++ // its own policy and not NOTE_ATTEMPTS / NOTE_COOLDOWN_MS above: a rate limit is ++ // counted per minute and recovers in seconds, and asking again costs one request ++ // instead of a whole part (#1447). A failure that outlasts these attempts below the ++ // hard line spends the round instead of the part, and the next round asks again ++ // after a wait that grows with each failure in a row. ++ const STEWARD_ATTEMPTS = 3 ++ const STEWARD_RETRY_MS = 250 ++ const STEWARD_RETRY_MAX_MS = 5_000 ++ const STEWARD_COOLDOWN_MS = [60_000, 300_000] const FOLDER_RETRY_MS = 300_000 + // Making room before a step takes the turn's own signal and nothing else, so a + // summary request stuck in retries parks the turn before its first step: no step, @@ -538,7 +549,87 @@ diff --git a/src/host/parts.js b/src/host/parts.js const clock = createClock(config.timeZone) // Lines for the model the Session last routed to; undefined while it has not -@@ -518,25 +518,50 @@ export function install(rt) { +@@ -504,39 +504,122 @@ export function install(rt) { + return parseDecision(assembler.blocks().filter(block => block.type === 'text').map(block => block.text).join('\n')) + } + ++ // A failure worth asking again: a rate limit, a timeout, a 5xx. Anything else — a ++ // request the provider refused, an answer that does not parse — is asked of the same ++ // key again and comes back the same. ++ const transientFailure = (error) => { ++ const code = String(error?.code ?? '') ++ if (code === 'throttling_error' || code === 'rate_limited' || code === 'TIMEOUT' || code === 'REQUEST_TIMEOUT' || code === 'ABORT_ERR') return true ++ const status = Number(error?.status ?? error?.statusCode) ++ return status === 429 || (status >= 500 && status <= 599) ++ } ++ ++ // The wait the provider asked for, capped, or a short backoff that grows by attempt. ++ const stewardRetryDelay = (error, attempt) => { ++ const asked = Number(error?.retryAfter ?? error?.retry_after) ++ if (Number.isFinite(asked) && asked > 0) return Math.min(asked * 1_000, STEWARD_RETRY_MAX_MS) ++ return STEWARD_RETRY_MS * attempt ++ } ++ const sleep = (ms) => new Promise(resolve => setTimeout(resolve, ms)) ++ ++ // A steward that could not be asked, per Session: how many rounds in a row, when to ++ // ask again, and the failure that a forced move writes into its segment (#1447). ++ const stewardFailures = new Map() ++ const stewardBlame = (sessionId) => stewardFailures.get(sessionId)?.text ++ + // What the steward decides; a part that cannot be condensed, or a steward that +- // gives no answer, moves on as it always could. +- async function decide(agent, { midTurn }) { ++ // gives no usable answer, moves on as it always could. A rate limit is not an ++ // answer: it is asked again, and below the hard line a failure that outlasts the ++ // attempts costs the round, not the part. ++ async function decide(agent, { midTurn, hard = false }) { + if (engineOf(agent) === undefined) return 'switch' +- try { +- const action = await askSteward(agent, { midTurn }) +- if (action !== undefined) return action +- rt.warn('ds-bot: the context steward gave no usable answer for %s; moving on to a fresh part', agent.id) +- } catch (error) { +- rt.warn('ds-bot: the context steward failed for %s; moving on to a fresh part: %s', agent.id, error) ++ let failure ++ let transient = false ++ for (let attempt = 1; attempt <= STEWARD_ATTEMPTS; attempt += 1) { ++ try { ++ const action = await askSteward(agent, { midTurn }) ++ if (action !== undefined) { ++ stewardFailures.delete(agent.id) ++ return action ++ } ++ rt.warn('ds-bot: the context steward gave no usable answer for %s; moving on to a fresh part', agent.id) ++ failure = undefined ++ transient = false ++ break ++ } catch (error) { ++ failure = error ++ transient = transientFailure(error) ++ if (!transient || attempt === STEWARD_ATTEMPTS) break ++ rt.warn('ds-bot: the context steward did not answer for %s (attempt %d of %d), asking again in %d ms: %s', agent.id, attempt, STEWARD_ATTEMPTS, stewardRetryDelay(error, attempt), error) ++ await sleep(stewardRetryDelay(error, attempt)) ++ } + } ++ if (failure !== undefined) { ++ stewardFailures.set(agent.id, { ++ count: 0, ++ until: 0, ++ text: `${failure?.code ?? failure?.status ?? 'error'}: ${String(failure?.message ?? failure).slice(0, 160)}`, ++ }) ++ } else { ++ // An answer that does not parse is not a failure to report later: the move it ++ // forces must not carry an older failure it did not cause. ++ stewardFailures.delete(agent.id) ++ } ++ if (transient && !hard) { ++ const count = (stewardFailures.get(agent.id)?.count ?? 0) + 1 ++ const waitMs = STEWARD_COOLDOWN_MS[Math.min(count, STEWARD_COOLDOWN_MS.length) - 1] ++ stewardFailures.set(agent.id, { ...stewardFailures.get(agent.id), count, until: Date.now() + waitMs }) ++ rt.warn('ds-bot: the context steward could not be asked for %s (%d in a row, %d attempts each); the part stays and is asked again in %ds: %s', agent.id, count, STEWARD_ATTEMPTS, waitMs / 1000, failure) ++ return 'wait' ++ } ++ rt.warn('ds-bot: the context steward failed for %s; moving on to a fresh part: %s', agent.id, failure ?? 'no usable answer') return 'switch' } @@ -600,7 +691,7 @@ diff --git a/src/host/parts.js b/src/host/parts.js partBase.set(agent.id, after) ctx.logger.info('ds-bot: %s condensed from %d to %d tokens (check line %d, hard line %d)', agent.id, before ?? -1, after ?? -1, lines.check, lines.hard) return after < lines.check -@@ -561,7 +561,7 @@ export function install(rt) { +@@ -561,20 +561,22 @@ export function install(rt) { const why = segment.fallback === 'overflow' ? 'the part overflowed the model\'s window' : segment.fallback === 'failure' ? 'a request in the previous part kept failing' : 'the model did not write a usable note' @@ -609,6 +700,23 @@ diff --git a/src/host/parts.js b/src/host/parts.js return segment.note === undefined ? `${head} There was no earlier note to carry over. Go by the messages below, and look up anything else.)` : `${head} It is the note written at the end of part ${segment.noteFrom}, carried over unchanged; anything newer is in the messages below, and read_own_chat finds the rest.)\n\n${segment.note}` + } + +- function handoffPackage({ room, segment, lines, number, resume, overflow, failed, bare, folder }) { ++ function handoffPackage({ room, segment, lines, number, resume, overflow, failed, bare, folder, steward }) { + const where = room ? `your conversation in the group chat "${room.name}"` : 'your own chat with the user' + const lookup = room + ? 'read_group_chat shows the group\'s latest messages, and read_own_chat your own chat.' + : 'read_own_chat searches every part, reads around an anchor such as [part 2 #1234 · …], and shows the note that closed any earlier part.' + const reason = failed !== undefined ? `a request in the previous part kept failing (${failed})` + : overflow ? 'the previous part overflowed the model\'s window' +- : folder !== undefined ? `your working folder changed to ${folder}` : 'the previous part grew long' ++ : folder !== undefined ? `your working folder changed to ${folder}` ++ : steward !== undefined ? `the context steward could not be asked whether to keep it (${steward})` ++ : 'the previous part grew long' + const { recent, earlier, unlisted, complete } = lines + const omitted = [ + ...(unlisted > 0 ? [`(${unlisted} older messages are not listed.)`] : []), @@ -582,7 +582,7 @@ export function install(rt) { ] return [ @@ -645,14 +753,17 @@ diff --git a/src/host/parts.js b/src/host/parts.js const switches = new Map() // Parts whose move to their working folder failed, and the time before which it is // not tried again. -@@ -672,8 +672,8 @@ export function install(rt) { +@@ -670,10 +670,10 @@ export function install(rt) { + + // `hard`: past the hard line or the window, the part moves on even without a note // from the model. `folder`: the part moves on because its working folder changed. - function startSwitch(agent, { resume = false, carry = [], info, joined, overflow = false, failed, bare = false, folder = false, hard = overflow || folder } = {}) { +- function startSwitch(agent, { resume = false, carry = [], info, joined, overflow = false, failed, bare = false, folder = false, hard = overflow || folder } = {}) { ++ function startSwitch(agent, { resume = false, carry = [], info, joined, overflow = false, failed, bare = false, folder = false, steward, hard = overflow || folder } = {}) { const place = placeOf(agent.id) - if (place === undefined || switches.has(agent.id) || readOnly()) return - const job = { ...place, carry: [...carry], info, joined, resume, overflow, failed, bare, folder, hard } + if (place === undefined || switches.has(agent.id) || readOnly()) return false -+ const job = { ...place, carry: [...carry], info, joined, resume, overflow, failed, bare, folder, hard, origin: settled(rt, agent.id) } ++ const job = { ...place, carry: [...carry], info, joined, resume, overflow, failed, bare, folder, steward, hard, origin: settled(rt, agent.id) } switches.set(agent.id, job) void runSwitch(agent, job) .catch(error => rt.warn('ds-bot: %s did not move on to a fresh part: %s', rt.botOf(job.botId)?.name ?? agent.id, error)) @@ -717,7 +828,8 @@ diff --git a/src/host/parts.js b/src/host/parts.js } await rt.applyPermission(bot, freshId) .catch(error => rt.warn('ds-bot: permission for the fresh part of %s not applied: %s', bot.name, error)) - const segment = { sessionId: agent.id, endedAt: Date.now(), ...(tokens !== undefined ? { tokens } : {}), ...noted } +- const segment = { sessionId: agent.id, endedAt: Date.now(), ...(tokens !== undefined ? { tokens } : {}), ...noted } ++ const segment = { sessionId: agent.id, endedAt: Date.now(), ...(tokens !== undefined ? { tokens } : {}), ...noted, ...(job.steward === undefined ? {} : { steward: job.steward }) } + const resume = job.resume || job.overflow + // Commit the new part before its handoff turn can run: a handoff visible to a failure handler + // before the segment is saved would let a second switch lose the previous part's note. A failed @@ -784,7 +896,7 @@ diff --git a/src/host/parts.js b/src/host/parts.js + folderRetry.set(freshId, Date.now() + FOLDER_RETRY_MS) + rt.warn('ds-bot: the fresh part of %s did not start in %s; it works in %s', bot.name, intended, cwd) + } -+ const text = handoffPackage({ room, segment, lines: carried, number: number + 1, resume, overflow: job.overflow, failed: job.failed, bare: job.bare, folder }) ++ const text = handoffPackage({ room, segment, lines: carried, number: number + 1, resume, overflow: job.overflow, failed: job.failed, bare: job.bare, folder, steward: job.steward }) + fresh.followup(createUserMessage({ content: [{ type: 'text', text }], source: handoffSource(job, resume) })) + await ctx.sessions.flush(fresh.session) + freshParts.set(freshId, estimateTokens(text)) @@ -816,7 +928,7 @@ diff --git a/src/host/parts.js b/src/host/parts.js ctx.logger.info('ds-bot: %s%s moved on to part %d at %d tokens (check line %d, hard line %d); %s; %d latest and %d earlier messages carried%s', bot.name, room ? ` in ${room.name}` : '', number + 1, tokens ?? -1, lines?.check ?? -1, lines?.hard ?? -1, written !== undefined ? `note written in ${written.attempts} attempt(s)` : `note put together by DS Bot (${noted.fallback}${noted.note ? `, carries part ${noted.noteFrom}'s` : ', nothing to carry'})`, carried.recent.length, carried.earlier.length, folder === undefined ? '' : `; now works in ${folder}`) -@@ -840,16 +840,16 @@ export function install(rt) { +@@ -840,16 +840,18 @@ export function install(rt) { freshParts.delete(agent.id) partBase.set(agent.id, (tokensOf(agent.session) ?? 0) + fresh) } @@ -827,8 +939,10 @@ diff --git a/src/host/parts.js b/src/host/parts.js if (lines === undefined || tokens === undefined || tokens < lines.hard || !grown(agent.id, tokens, lines) || readOnly()) return next() - if (await decide(agent, { midTurn: payload.step > 1 }) === 'compact' && await condense(agent, lines, { inTurn: true, signal: payload.signal })) return next() - return switchBeforeStep(agent, payload, { hard: true }) -+ if (await decide(agent, { midTurn: payload.step > 1 }) === 'compact' && await condense(agent, lines, { inTurn: true, signal: payload.signal, keep: payload.messages })) return next() -+ return switchBeforeStep(agent, payload, { hard: true }, next) ++ // Before a step the part is already past the hard line, so a steward that cannot ++ // be asked moves the part on here; the failure still names itself in its segment. ++ if (await decide(agent, { midTurn: payload.step > 1, hard: true }) === 'compact' && await condense(agent, lines, { inTurn: true, signal: payload.signal, keep: payload.messages })) return next() ++ return switchBeforeStep(agent, payload, { hard: true, steward: stewardBlame(agent.id) }, next) }, { prepend: true }) // `messages` left the inbox for this step, so the switch carries them along. @@ -850,6 +964,29 @@ diff --git a/src/host/parts.js b/src/host/parts.js return { kind: 'reject' } } +@@ -927,15 +927,19 @@ export function install(rt) { + const tokens = tokensOf(agent.session) + if (lines === undefined || tokens === undefined || tokens < lines.check || !grown(agent.id, tokens, lines)) return + const hard = tokens >= lines.hard +- if (!hard && Date.now() < (noteFailures.get(agent.id)?.until ?? 0)) return ++ // Below the hard line a part that could not be written or decided waits out its ++ // cooldown instead of asking again at every idle (#1447). ++ if (!hard && (Date.now() < (noteFailures.get(agent.id)?.until ?? 0) || Date.now() < (stewardFailures.get(agent.id)?.until ?? 0))) return + checking.add(agent.id) + try { +- if (await decide(agent, { midTurn: false }) === 'compact') { ++ const decision = await decide(agent, { midTurn: false, hard }) ++ if (decision === 'wait') return ++ if (decision === 'compact') { + const condensed = await condense(agent, lines, { inTurn: false }) + if (condensed !== false) return + } + if (switches.has(agent.id)) return +- startSwitch(agent, { hard: (tokensOf(agent.session) ?? tokens) >= lines.hard }) ++ startSwitch(agent, { hard: (tokensOf(agent.session) ?? tokens) >= lines.hard, steward: stewardBlame(agent.id) }) + } finally { + checking.delete(agent.id) + } diff --git a/src/host/secrets.js b/src/host/secrets.js --- a/src/host/secrets.js +++ b/src/host/secrets.js diff --git a/patches/ds-bot/files/test/steward-retry.test.js b/patches/ds-bot/files/test/steward-retry.test.js new file mode 100644 index 00000000..80d21fe1 --- /dev/null +++ b/patches/ds-bot/files/test/steward-retry.test.js @@ -0,0 +1,230 @@ +import assert from 'node:assert/strict' +import { mkdirSync, mkdtempSync, readFileSync, realpathSync, rmSync, writeFileSync } from 'node:fs' +import { tmpdir } from 'node:os' +import { join } from 'node:path' +import { test } from 'node:test' +import { apply } from '../index.js' +import { fakeContext } from './fake-ctx.js' + +// A Main Bot and a Bot, both idle in their own chat. +const TEAM = { + version: 1, + mainBotIds: ['a'], + workspaceId: 'w1', + bots: { + a: { id: 'a', name: 'Chief', role: 'Boss', sessionId: 'a', createdAt: 1 }, + b: { id: 'b', name: 'Writer', role: 'Write', sessionId: 'b', createdAt: 2 }, + }, + rooms: {}, + agents: {}, + exchanges: [], + exchangeTurns: {}, + questions: {}, + prefs: {}, + revision: 0, +} + +// The provider's stream, in the shape `BlockAssembler` reads: text deltas and a finish. +const answer = (text) => (async function* () { + yield { type: 'text-delta', index: 0, text } + yield { type: 'finish', reason: { kind: 'stop' } } +})() + +// A rate limit as the provider reports it: per-minute, recovered in seconds, and the +// `retryAfter` it asks for. +const throttled = (retryAfter = 0) => Object.assign(new Error('429: rate limit exceeded (requests)'), { + status: 429, + code: 'throttling_error', + retryAfter, +}) + +// A provider error that is not a rate limit: asking again would change nothing. +const broken = () => Object.assign(new Error('500: the provider failed'), { status: 500, code: 'PROVIDER_ERROR' }) + +// One model window, and a part whose size the test sets: `lines` for window 2000 with a +// 100-token reserve are hard 1805 / check 1520, so `between` sits past the check line and +// short of the hard one, and `over` is past both. +const WINDOW = { context: { contextWindow: 2_000 }, defaultMaxTokens: 100 } +const between = 'x'.repeat(4_800) +const over = 'x'.repeat(5_600) + +// The team's data in a temporary home, a project folder beside it, a compaction engine +// that makes room when the steward says so, and a provider stream the test drives. +async function start({ text = between, stream } = {}) { + const root = realpathSync.native(mkdtempSync(join(tmpdir(), 'ds-bot-steward-retry-'))) + const dsh = join(root, 'dsh') + const home = join(dsh, 'bot') + const team = join(home, 'workspace') + const project = join(root, 'project') + for (const dir of [team, project]) mkdirSync(dir, { recursive: true }) + writeFileSync(join(home, 'state.json'), JSON.stringify(TEAM)) + + const messages = [{ role: 'user', content: [{ type: 'text', text }] }] + const condensed = [] + const compaction = { + compactIfNeeded: async () => null, + compactNow: async () => { + condensed.push('compactNow') + // What a summary leaves behind: the part is below the check line afterwards. + messages.splice(0, messages.length, { role: 'user', content: [{ type: 'text', text: 'condensed' }] }) + return { summary: 'condensed' } + }, + compactRegion: async () => {}, + summarize: async () => '', + } + + const ctx = fakeContext({ services: { compaction } }) + const created = [] + const cwds = new Map() + const agents = new Map() + const agentOf = (sessionId) => { + if (!agents.has(sessionId)) { + const calls = [] + agents.set(sessionId, { + id: sessionId, + session: { + id: sessionId, + header: { cwd: cwds.get(sessionId) ?? team }, + requestHeader: () => ({ config: { provider: 'deepseek', model: 'deepseek-v4-flash', maxTokens: 100 } }), + deriveMessages: () => messages, + toolHistory: () => [], + }, + calls, + status: 'idle', + followup(message) { calls.push(['followup', message]) }, + steer(message) { calls.push(['steer', message]) }, + inject(message) { calls.push(['inject', message]) }, + cancel() {}, + async whenIdle() {}, + async runMaintenance(fn) { return fn(AbortSignal.timeout(60_000)) }, + inbox: { nextStep: [], nextTurn: [], remove() {} }, + }) + } + return agents.get(sessionId) + } + const archived = [] + const registry = new Map([['w1', { id: 'w1', path: team, title: 'DS Bot' }]]) + const calls = [] + Object.assign(ctx, { + llm: { + listProviders: () => [], + listModels: async () => [], + resolveModelInfo: async () => WINDOW, + stream: (request) => { + calls.push(request) + return stream(calls.length, request) + }, + }, + workspaceRegistry: { + get: id => registry.get(id), + list: () => [...registry.values()], + resolveByPath: async path => [...registry.values()].find(workspace => workspace.path === path), + create: async path => ({ id: 'w-new', path }), + async archiveSession(sessionId) { archived.push(sessionId) }, + }, + sessionController: { + create: async (options) => { + const sessionId = `s${created.length + 1}` + created.push(options) + cwds.set(sessionId, options.cwd ?? registry.get(options.workspaceId)?.path) + return { sessionId } + }, + resolveAgent: async sessionId => ({ agent: agentOf(sessionId) }), + selectModel: async () => {}, + page: async () => ({ records: [], hasMore: false }), + }, + sessionTitle: { rename: async () => {} }, + sessions: { flush: async () => {} }, + }) + let route + const register = ctx.connection.fetch.register + ctx.connection.fetch.register = (options) => { + route = options + return register(options) + } + apply(ctx, { home }) + const call = async (endpoint, payload) => (await route.fetch(new Request('http://localhost/api/bot', { + method: 'POST', body: JSON.stringify({ endpoint, payload }), + }))).json() + const tool = (name, args, sessionId) => ctx.registeredTools.get(name).execute(args, { agent: agentOf(sessionId), name, callId: 'c1' }) + const stop = () => { + ctx.dispose() + rmSync(root, { recursive: true, force: true, maxRetries: 5 }) + } + // Loading once fills the Session-owner indexes. + await tool('list_bots', {}, 'a') + // The engine is adopted where DSH announces a new agent; the idle check reads it there. + const b = agentOf('b') + ctx.emit('agent/created', { agent: b }) + const idle = () => ctx.emit('agent/status', { agent: b, status: 'idle' }) + return { ctx, created, calls, condensed, archived, agentOf, call, idle, state: () => JSON.parse(readFileSync(join(home, 'state.json'), 'utf8')), stop } +} + +const settle = () => new Promise(resolve => setTimeout(resolve, 20)) +const until = async (check, ms = 5_000) => { + const end = Date.now() + ms + while (!check()) { + if (Date.now() > end) throw new Error('until: the condition did not hold within 5 s') + await settle() + } +} +const forMs = async (ms) => { + const end = Date.now() + ms + while (Date.now() < end) await settle() +} + +test('#1447: a throttled steward is asked again instead of moving the part on', async (t) => { + let stewardCalls = 0 + const h = await start({ + stream: (count) => { + if (count === 1) throw throttled(0) + stewardCalls += 1 + return answer('{"action":"compact","reason":"still working"}') + }, + }) + t.after(() => h.stop()) + h.idle() + await until(() => h.calls.length >= 2) + await forMs(200) + assert.equal(h.calls.length, 2, 'the steward was asked twice: once refused, once answered') + assert.equal(stewardCalls, 1, 'the second answer is the one that counts') + assert.deepEqual(h.condensed, ['compactNow'], 'the answer was acted on: the part was condensed in place') + assert.equal(h.created.length, 0, 'no fresh part was started for one rate limit') + assert.deepEqual(h.archived, [], 'nothing was archived') +}) + +test('#1447: below the hard line a steward that keeps being turned down costs the round, not the part', async (t) => { + const h = await start({ stream: () => { throw throttled(0) } }) + t.after(() => h.stop()) + h.idle() + await until(() => h.calls.length >= 3, 10_000) + await forMs(200) + assert.equal(h.calls.length, 3, 'the steward was asked the three attempts and no more') + assert.equal(h.created.length, 0, 'the part stays: nothing was started for it') + assert.deepEqual(h.archived, [], 'nothing was archived') + // The next idle inside the cooldown does not ask again: the same key is still limited. + h.idle() + await forMs(300) + assert.equal(h.calls.length, 3, 'the cooldown holds: the part waits without another request') + assert.equal(h.created.length, 0, 'and it still stays') +}) + +test('#1447: a steward that cannot be asked over the hard line still moves the part on and says why', async (t) => { + const h = await start({ text: over, stream: () => { throw broken() } }) + t.after(() => h.stop()) + h.idle() + await until(() => h.created.length === 1, 15_000) + assert.equal(h.created.length, 1, 'the part past the hard line moved on') + // The switch starts the fresh part first and archives the finished one once the move is + // done, so the archive lands a step after `created`: wait for it instead of reading it + // the moment the new Session exists. The assertion itself stays as strict. + await until(() => h.archived.length === 1, 15_000) + assert.deepEqual(h.archived, ['b'], 'the finished part was archived') + const packageText = (h.agentOf('s1').calls.find(([kind]) => kind === 'followup') ?? [])[1] + ?.content?.map(block => block.text).join('') + assert.match(packageText ?? '', /the context steward could not be asked/) + assert.doesNotMatch(packageText ?? '', /the previous part grew long/) + const stored = h.state().bots.b.segments?.at(-1) + assert.match(String(stored?.steward ?? ''), /PROVIDER_ERROR/, 'the segment keeps the failure that forced the move') +}) + diff --git a/patches/ds-bot/topics.json b/patches/ds-bot/topics.json index 7c944fb4..a1d47044 100644 --- a/patches/ds-bot/topics.json +++ b/patches/ds-bot/topics.json @@ -53,7 +53,8 @@ "test/images.test.js", "test/session-identity.test.js", "src/host/steward.js", - "test/steward.test.js" + "test/steward.test.js", + "test/steward-retry.test.js" ], "0002-flybot-api-host": [ "index.js",