fix(parts): ask the context steward again before moving a part on

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)).

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)
     }