fix(host): give the pre-step compaction a deadline (#1285)

changelog.d/1285-pre-step-compaction-deadline.fixed.md

@@ -0,0 +1 @@
+A turn no longer stays open because making room before its next step never returns: the pre-step compaction now carries a deadline (`COMPACT_DEADLINE_MS`, `config.compactDeadlineMs`, five minutes) beside the turn's own signal, and a summary request still retrying when it passes ends that turn with a warning instead of parking it — `turn/end` runs, the queued messages stay for the next turn, and `stop_bot` is no longer the only way out ([#1285](https://github.com/ThinkFlowLab/FlyBot/issues/1285)).

core/src/host/parts.js

@@ -160,6 +160,11 @@ export function install(rt) {
   const STEWARD_TIMEOUT_MS = 120_000
   const STEWARD_LINES = 40
   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,
+  // no turn end, `load: 1` until a human stops it (#1285). Its own deadline ends
+  // that park; the next turn tries again.
+  const COMPACT_DEADLINE_MS = config.compactDeadlineMs ?? 300_000
 
   // A broken provider adapter must not crash every turn; estimate from the
   // derived messages instead, with one warning.
@@ -556,25 +561,50 @@ export function install(rt) {
     return 'switch'
   }
 
+  // Puts a step's claimed input back in the queue. A turn that ends in an error keeps
+  // nothing: DSH claimed these messages durably before the pre-step hook ran and does
+  // not return them, so the batch goes back as ordinary queued input. The loop starts
+  // the next turn while input is pending, so that message is taken up again.
+  function keepClaimed(agent, messages) {
+    for (const message of messages ?? []) {
+      try {
+        agent.inbox.append('next-turn', message)
+      } catch (error) {
+        rt.warn('ds-bot: input for %s could not go back in the queue: %s', agent.id, error)
+      }
+    }
+  }
+
   // Condenses the part through its own backend. `inTurn`: before a step of a running
   // turn, through the backend's overflow path, which condenses all but the latest
   // node; otherwise as an idle maintenance pass. True when the part ended up below
   // the check line, false when it did not (it then moves on), undefined when a turn
   // got in first (the part is checked again when that turn ends).
-  async function condense(agent, lines, { inTurn, signal }) {
+  async function condense(agent, lines, { inTurn, signal, keep }) {
     const engine = engineOf(agent)
     const before = tokensOf(agent.session)
     const sessionId = agent.session.id
+    // Before a step, the compaction carries the turn's signal and now its own deadline.
+    const deadline = inTurn ? AbortSignal.timeout(COMPACT_DEADLINE_MS) : undefined
     condensing.add(sessionId)
     try {
       // Through the service itself, past DS Bot's own wrapper: only a call on the
       // service gives the engine its own context, with the services it injects.
       const result = inTurn
-        ? await engine.compactIfNeeded(agent, 'context-overflow', signal)
+        ? await engine.compactIfNeeded(agent, 'context-overflow', AbortSignal.any([signal, deadline].filter(Boolean)))
         : await engine.compactNow(agent, AbortSignal.timeout(NOTE_TIMEOUT_MS))
       if (result === null) return false
     } catch (error) {
       if (!inTurn && (error?.code === 'busy' || error?.code === 'cancelled')) return undefined
+      // Out of time before a step: end this turn here, where `turn/end` still runs,
+      // rather than wait for a summary that may never come. DSH claimed this step's
+      // input before the hook ran and a turn that ends in an error does not get it
+      // back, so the batch is queued again before the throw (keepClaimed).
+      if (deadline?.aborted && !signal?.aborted) {
+        rt.warn('ds-bot: making room for %s ran out of %d ms before a step; the turn ends here and the next one tries again: %s', agent.id, COMPACT_DEADLINE_MS, error)
+        keepClaimed(agent, keep)
+        throw error
+      }
       rt.warn('ds-bot: condensing %s failed: %s', agent.id, error)
       return false
     } finally {
@@ -955,7 +985,7 @@ 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 })) 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)
   }, { prepend: true })
 

core/test/pre-step-compaction.test.js

@@ -0,0 +1,209 @@
+import assert from 'node:assert/strict'
+import { setTimeout as delay } from 'node:timers/promises'
+import { test } from 'node:test'
+import { fakeContext } from './fake-ctx.js'
+import { install } from '../src/host/parts.js'
+
+// #1285: the pre-step that makes room for a step hands the engine the turn's own
+// signal and nothing else. A compaction whose summary request keeps retrying never
+// returns, so the turn never takes another step, the session keeps reading as
+// working, and only a human `stop_bot` ends it.
+function setup({ compaction, deadlineMs = 50, tokens = () => 30_000 } = {}) {
+  const ctx = fakeContext()
+  const warned = []
+  const engine = {
+    summarize: async function* () {},
+    compactIfNeeded(target, trigger, signal) {
+      return compaction({ target, trigger, signal })
+    },
+    compactNow: async () => null,
+  }
+  const session = {
+    id: 'bot',
+    header: { cwd: '/work' },
+    requestHeader: () => ({ config: { provider: 'deepseek', model: 'flash', maxTokens: 0 } }),
+    deriveMessages: () => [],
+  }
+  const inbox = stubInbox()
+  const agent = { id: 'bot', session, inbox }
+  const rt = {
+    ctx,
+    config: { timeZone: 'UTC', compactDeadlineMs: deadlineMs },
+    state: { bots: {}, rooms: {}, exchanges: [], questions: {}, revision: 0 },
+    chatOwners: new Map(),
+    groupOwners: new Map(),
+    chatOf: () => undefined,
+    roomOf: () => undefined,
+    botOf: () => ({ id: 'bot', name: 'Writer', sessionId: 'bot', createdAt: 1 }),
+    selfOf: () => ({}),
+    pastOwners: new Map(),
+    readOnly: () => undefined,
+    tagUsage: (messages) => messages,
+    renewMemory() {},
+    async load() {},
+    isBotSession: () => true,
+    async *eventsBackwards() {},
+    async createSession() { throw new Error('a switch is not part of this test') },
+    liveOf: () => ({ switched: false, info: undefined }),
+    warn: (format, ...args) => warned.push(String(format).replace(/%[sd]/g, () => String(args.shift() ?? ''))),
+    async save() {},
+  }
+  Object.assign(ctx, {
+    llm: {
+      listProviders: () => [{ id: 'deepseek', name: 'DeepSeek' }],
+      listModels: async () => [],
+      resolveModelInfo: async () => ({ context: { contextWindow: 20_000 }, defaultMaxTokens: 0 }),
+      stream: () => {
+        return (async function* () {
+          yield { type: 'text-delta', text: '{"action":"compact","reason":"still working"}' }
+          yield { type: 'finish', reason: { kind: 'completed' } }
+        })()
+      },
+    },
+  })
+  ctx.set('compaction', engine)
+  ctx.set('tokenMeter', { measure: () => ({ totalTokens: tokens() }) })
+  install(rt)
+  const hook = ctx.listeners.get('agent/pre-step').at(-1)
+  const step = (payload = {}) => hook({
+    agent, turn: 1, step: 3, messages: [],
+    signal: new AbortController().signal,
+    ...payload,
+  }, async () => ({ kind: 'enter', messages: [] }))
+  return { ctx, agent, engine, warned, step, inbox }
+}
+
+// The two pending lists DSH's turn loop claims from, with the same rules the host
+// applies: an identity may sit in only one of them, and a turn takes every next-step
+// message plus one queued next-turn message before the pre-step hook runs.
+function stubInbox() {
+  const pending = { 'next-turn': [], 'next-step': [] }
+  return {
+    append(target, message) {
+      const other = target === 'next-turn' ? 'next-step' : 'next-turn'
+      if ([...pending[target], ...pending[other]].some(held => held.id === message.id)) throw new Error(`message "${message.id}" is already pending`)
+      pending[target].push(message)
+    },
+    claim() {
+      const claimed = [...pending['next-step'], ...pending['next-turn'].slice(0, 1)]
+      pending['next-step'] = []
+      pending['next-turn'] = pending['next-turn'].slice(1)
+      return claimed
+    },
+    waiting: target => pending[target].slice(),
+    hasPending: () => pending['next-turn'].length > 0 || pending['next-step'].length > 0,
+  }
+}
+
+// A compaction that answers a summary request only when its signal says to stop:
+// what a request stuck in retries looks like from the outside.
+function stubborn() {
+  return ({ signal }) => new Promise((_resolve, reject) => {
+    const stop = () => reject(Object.assign(new Error('aborted'), { name: 'AbortError', code: 'ABORTED' }))
+    if (signal?.aborted) stop()
+    else signal?.addEventListener('abort', stop, { once: true })
+  })
+}
+
+test('#1285: a pre-step compaction that never returns stops the turn instead of parking it', async () => {
+  const { warned, step } = setup({ compaction: stubborn() })
+  const outcome = await Promise.race([
+    step().then(() => 'returned', (error) => `rejected: ${error.message}`),
+    delay(2_000).then(() => 'hung'),
+  ])
+  assert.notEqual(outcome, 'hung', 'the pre-step must end instead of waiting for a compaction that never returns')
+  assert.match(warned.join('\n'), /ran out of 50 ms/)
+})
+
+test('#1285: a compaction that answers inside the deadline leaves the step alone', async () => {
+  let compacted = false
+  const { step, warned } = setup({
+    compaction: async () => { compacted = true; return { compact: true } },
+    tokens: () => (compacted ? 10_000 : 30_000),
+  })
+  assert.deepEqual(await step(), { kind: 'enter', messages: [] })
+  assert.deepEqual(warned, [])
+})
+
+test('#1285: the turn the deadline ends puts the step input back for the next turn', async () => {
+  const { step, inbox, warned } = setup({ compaction: stubborn() })
+  const batch = [
+    { id: 'm1', content: [{ type: 'text', text: 'first' }] },
+    { id: 'm2', content: [{ type: 'text', text: 'second' }] },
+  ]
+  // The turn loop claimed the batch before the hook ran, as DSH does.
+  inbox.append('next-step', batch[0])
+  inbox.append('next-turn', batch[1])
+  assert.deepEqual(inbox.claim(), batch, 'the batch is claimed before the hook runs')
+  assert.deepEqual(inbox.waiting('next-turn'), [])
+
+  const outcome = await Promise.race([
+    step({ messages: batch }).then(() => 'returned', (error) => `rejected: ${error.message}`),
+    delay(2_000).then(() => 'hung'),
+  ])
+  assert.notEqual(outcome, 'hung', 'the pre-step must end instead of waiting for a compaction that never returns')
+  assert.match(warned.join('\n'), /ran out of 50 ms/)
+
+  // Back in the queue, not only in a call: the next turn claims them the ordinary way.
+  assert.equal(inbox.hasPending(), true)
+  assert.deepEqual(inbox.waiting('next-turn'), batch)
+  assert.deepEqual(inbox.claim(), [batch[0]])
+  assert.deepEqual(inbox.claim(), [batch[1]])
+  assert.equal(inbox.hasPending(), false)
+})
+
+// A compaction that ignores the signal it was given and answers only to the turn:
+// the deadline has already fired when the user stops the turn, so both are aborted.
+function deafToDeadline(turn) {
+  return () => new Promise((_resolve, reject) => {
+    const stop = () => reject(Object.assign(new Error('stopped'), { name: 'AbortError', code: 'ABORTED' }))
+    if (turn.signal.aborted) stop()
+    else turn.signal.addEventListener('abort', stop, { once: true })
+  })
+}
+
+test('#1285: a turn the user stopped during the wait takes the old path', async () => {
+  const { step, warned } = setup({ compaction: stubborn() })
+  const turn = new AbortController()
+  const pending = step({ signal: turn.signal })
+  await delay(0)
+  turn.abort(new Error('stopped'))
+  assert.deepEqual(await pending, { kind: 'enter', messages: [] }, 'the step goes on as it did before the deadline')
+  assert.match(warned.join('\n'), /condensing bot failed/)
+  assert.doesNotMatch(warned.join('\n'), /ran out of/)
+})
+
+test('#1285: a deadline already reached does not claim a turn the user stopped', async () => {
+  const turn = new AbortController()
+  const { step, warned } = setup({ compaction: deafToDeadline(turn), deadlineMs: 20 })
+  const pending = step({ signal: turn.signal })
+  await delay(60)
+  turn.abort(new Error('stopped'))
+  assert.deepEqual(await pending, { kind: 'enter', messages: [] }, "the turn is the user's, not the deadline's")
+  assert.match(warned.join('\n'), /condensing bot failed/)
+  assert.doesNotMatch(warned.join('\n'), /ran out of/)
+})
+
+test('#1285: putting the same input back twice adds no second copy', async () => {
+  const { step, inbox, warned } = setup({ compaction: stubborn() })
+  const batch = [
+    { id: 'm1', content: [{ type: 'text', text: 'first' }] },
+    { id: 'm2', content: [{ type: 'text', text: 'second' }] },
+  ]
+  inbox.append('next-step', batch[0])
+  inbox.append('next-turn', batch[1])
+  inbox.claim()
+  const first = await Promise.race([
+    step({ messages: batch }).then(() => 'returned', error => `rejected: ${error.message}`),
+    delay(2_000).then(() => 'hung'),
+  ])
+  assert.notEqual(first, 'hung', 'the deadline ends the turn instead of parking it')
+  assert.match(first, /rejected: aborted/)
+  assert.deepEqual(inbox.waiting('next-turn'), batch)
+
+  // The same batch ends a second turn while it is already queued: the inbox refuses an
+  // identity it holds, and the put-back keeps going instead of throwing over it.
+  await assert.rejects(step({ messages: batch }), /aborted/)
+  assert.deepEqual(inbox.waiting('next-turn'), batch)
+  assert.match(warned.join('\n'), /could not go back in the queue/)
+})