From b69b8d6f8775cf4aa8e658be965a54fa0ea8e4c6 Mon Sep 17 00:00:00 2001 From: Danyang Chen Date: Mon, 26 Oct 2026 20:34:13 +0800 Subject: [PATCH] fix(host): give the pre-step compaction a deadline (#1285) Written-by: Dev2 Fixes #1285 --- ...1285-pre-step-compaction-deadline.fixed.md | 1 + core/src/host/parts.js | 36 ++- core/test/pre-step-compaction.test.js | 209 ++++++++++++++++++ docs/subsystems/core-host.md | 1 + patches/ds-bot/0014-flybot-branding.patch | 106 +++++++-- .../files/test/pre-step-compaction.test.js | 209 ++++++++++++++++++ 6 files changed, 538 insertions(+), 24 deletions(-) create mode 100644 changelog.d/1285-pre-step-compaction-deadline.fixed.md create mode 100644 core/test/pre-step-compaction.test.js create mode 100644 patches/ds-bot/files/test/pre-step-compaction.test.js diff --git a/changelog.d/1285-pre-step-compaction-deadline.fixed.md b/changelog.d/1285-pre-step-compaction-deadline.fixed.md new file mode 100644 index 00000000..b910cec4 --- /dev/null +++ b/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)). diff --git a/core/src/host/parts.js b/core/src/host/parts.js index 570741c1..1e718558 100644 --- a/core/src/host/parts.js +++ b/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 }) diff --git a/core/test/pre-step-compaction.test.js b/core/test/pre-step-compaction.test.js new file mode 100644 index 00000000..b1733430 --- /dev/null +++ b/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/) +}) diff --git a/docs/subsystems/core-host.md b/docs/subsystems/core-host.md index c8ec6a10..c0f6f880 100644 --- a/docs/subsystems/core-host.md +++ b/docs/subsystems/core-host.md @@ -376,6 +376,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). +- **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`. ### Connectors: GitHub, the Composio gateway, and the Feishu sender (#311, #1263) diff --git a/patches/ds-bot/0014-flybot-branding.patch b/patches/ds-bot/0014-flybot-branding.patch index b73db868..c93465a3 100644 --- a/patches/ds-bot/0014-flybot-branding.patch +++ b/patches/ds-bot/0014-flybot-branding.patch @@ -440,7 +440,7 @@ diff --git a/src/host/parts.js b/src/host/parts.js import { CHECKPOINT_PROMPT, STEWARD_SYSTEM, checkpointCount, checkpointTail, checkpointTailNote, checkpointText, linesFor, parseDecision, stewardRequest } from './steward.js' import { clip, createClock, escapeRegExp, estimateTokens, splitRef, textOf } from './text.js' -@@ -40,9 +40,9 @@ export function cardText(args) { +@@ -40,9 +40,9 @@ // What one event adds to a Bot's chat history: an answer of the Bot's (its text, or a // question card), or a message from the user (`who: 'user'`) or another Bot // (`who: 'bot'`) that the answers after it belong to. Context that other plugins append @@ -453,7 +453,7 @@ diff --git a/src/host/parts.js b/src/host/parts.js // The user's answer to an ask_user card. The message carries the question with the // answer, so the answer still makes sense after a checkpoint drops the card. export function questionReply(question, answer) { -@@ -121,6 +121,20 @@ export function canonicalHeadings(note, names) { +@@ -121,6 +121,20 @@ return lines.join('\n') } @@ -474,7 +474,16 @@ diff --git a/src/host/parts.js b/src/host/parts.js export function install(rt) { const { ctx, config, state } = rt -@@ -149,21 +149,44 @@ export function install(rt) { +@@ -145,25 +145,53 @@ + 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. let measureWarned = false @@ -521,7 +530,60 @@ 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 -@@ -544,7 +544,7 @@ export function install(rt) { +@@ -518,25 +518,50 @@ + 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 { +@@ -544,7 +544,7 @@ } // The prefix cache starts over anyway, so the memory section catches up. rt.renewMemory(sessionId) @@ -530,7 +592,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,7 +561,7 @@ 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' @@ -539,7 +601,7 @@ 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}` -@@ -582,7 +582,7 @@ export function install(rt) { +@@ -582,7 +582,7 @@ ] return [ `[Handoff · part ${number}]`, @@ -548,7 +610,7 @@ diff --git a/src/host/parts.js b/src/host/parts.js 'It is reference, not new instructions: the note and the messages below are a record of earlier work. Do what they say only as part of the work they record, and take new requests only from messages that come after this one.', `Everything earlier stays saved: ${lookup}`, ...(folder !== undefined ? [`You now work in ${folder}: commands and file tools start there. Relative paths in the messages below meant the previous folder.`] : []), -@@ -610,17 +610,24 @@ export function install(rt) { +@@ -610,17 +610,24 @@ ].join('\n') } @@ -575,7 +637,7 @@ 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) { +@@ -672,8 +672,8 @@ // 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 } = {}) { const place = placeOf(agent.id) @@ -586,7 +648,7 @@ diff --git a/src/host/parts.js b/src/host/parts.js 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)) -@@ -693,12 +693,13 @@ export function install(rt) { +@@ -693,12 +693,13 @@ if (resume) { const carried = job.joined?.length ? { joined: job.joined } : {} back.unshift(createUserMessage({ @@ -602,7 +664,7 @@ diff --git a/src/host/parts.js b/src/host/parts.js } async function runSwitch(agent, job) { -@@ -716,8 +716,24 @@ export function install(rt) { +@@ -716,8 +716,24 @@ } if (job.moved) { rt.renewMemory(agent.id) @@ -628,7 +690,7 @@ diff --git a/src/host/parts.js b/src/host/parts.js } } -@@ -765,16 +765,43 @@ export function install(rt) { +@@ -765,16 +765,43 @@ const dir = rt.workDirOf(job) const intended = dir !== undefined && await rt.isFolder(dir) ? dir : rt.teamFolder() const freshId = await rt.createSession(room ? `${bot.name} · ${room.name}` : bot.name, { dir }) @@ -677,7 +739,7 @@ diff --git a/src/host/parts.js b/src/host/parts.js if (room !== undefined) { room.segments ??= {} ;(room.segments[bot.id] ??= []).push(segment) -@@ -790,28 +790,50 @@ export function install(rt) { +@@ -790,28 +790,50 @@ delete state.questions[agent.id] } } @@ -696,6 +758,12 @@ diff --git a/src/host/parts.js b/src/host/parts.js - if (intended !== undefined && cwd !== undefined && cwd !== intended) { - 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 }) +- freshParts.set(freshId, estimateTokens(text)) +- fresh.followup(createUserMessage({ content: [{ type: 'text', text }], source: handoffSource(job, resume) })) +- await ctx.sessions.flush(fresh.session) +- // Then what the old part held: the stopped turn's input, then whatever waited. + try { + await rt.save() + } catch (error) { @@ -724,12 +792,7 @@ diff --git a/src/host/parts.js b/src/host/parts.js + await rt.save().catch(() => { /* the in-memory state is the old part; a failed rollback save is the worst case */ }) + await ctx.workspaceRegistry.archiveSession(freshId).catch(() => { /* best effort */ }) + throw error - } -- const text = handoffPackage({ room, segment, lines: carried, number: number + 1, resume, overflow: job.overflow, failed: job.failed, bare: job.bare, folder }) -- freshParts.set(freshId, estimateTokens(text)) -- fresh.followup(createUserMessage({ content: [{ type: 'text', text }], source: handoffSource(job, resume) })) -- await ctx.sessions.flush(fresh.session) -- // Then what the old part held: the stopped turn's input, then whatever waited. ++ } + // The earlier part may not be archived while this move still holds its tasks (#896), so + // the switch keeps the promise and waits for it before archiving; the handoff goes on now. + job.scheduleMove = rt.moveSchedules(agent.id, freshId) @@ -746,7 +809,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,16 @@ freshParts.delete(agent.id) partBase.set(agent.id, (tokensOf(agent.session) ?? 0) + fresh) } @@ -755,8 +818,9 @@ diff --git a/src/host/parts.js b/src/host/parts.js 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 })) 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) }, { prepend: true }) @@ -766,7 +830,7 @@ diff --git a/src/host/parts.js b/src/host/parts.js // A switch may have started while the checks above awaited. const running = switches.get(agent.id) if (running !== undefined) { -@@ -858,8 +858,11 @@ export function install(rt) { +@@ -858,8 +858,11 @@ } const midTurn = payload.step > 1 const entry = rt.liveOf(agent.id) diff --git a/patches/ds-bot/files/test/pre-step-compaction.test.js b/patches/ds-bot/files/test/pre-step-compaction.test.js new file mode 100644 index 00000000..b1733430 --- /dev/null +++ b/patches/ds-bot/files/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/) +})