From 8ba50b1d45a768e36c346845c843b78868ea574f Mon Sep 17 00:00:00 2001 From: Danyang Chen Date: Thu, 22 Oct 2026 12:46:56 +0800 Subject: [PATCH 1/3] Hold a burst of the user's lines and open one turn A burst typed in a row opened one turn per line, so the Bot started on the first line and then took the next ones as separate work (#642). In a Bot's own chat the host now holds the user's lines for 1500 ms and opens one turn with them in arrival order, each on its own line. A line that arrives while a turn runs still lands as its own queued item, a Bot-to-Bot delivery never waits behind the window, and a group room keeps its round rules. --- changelog.d/642-settle-window.changed.md | 1 + core/src/host/channel.js | 52 ++++++++ core/test/install.test.js | 1 + core/test/settle.test.js | 158 +++++++++++++++++++++++ docs/subsystems/core-host.md | 6 + patches/ds-bot/0001-core-fixes.patch | 5 +- patches/ds-bot/0005-social-seams.patch | 83 +++++++++--- patches/ds-bot/files/test/settle.test.js | 158 +++++++++++++++++++++++ 8 files changed, 447 insertions(+), 17 deletions(-) create mode 100644 changelog.d/642-settle-window.changed.md create mode 100644 core/test/settle.test.js create mode 100644 patches/ds-bot/files/test/settle.test.js diff --git a/changelog.d/642-settle-window.changed.md b/changelog.d/642-settle-window.changed.md new file mode 100644 index 000000000..c519c8fbe --- /dev/null +++ b/changelog.d/642-settle-window.changed.md @@ -0,0 +1 @@ +Typing two or three short lines in a row in a Bot's own chat now opens one turn instead of one turn per line: the host holds the lines for 1.5 seconds and hands them over together, in the order they were typed, each on its own line. A line sent while the Bot is already answering still lands on its own, and a group chat is unchanged ([#642](https://github.com/ThinkFlowLab/FlyBot/issues/642)). diff --git a/core/src/host/channel.js b/core/src/host/channel.js index c045cfbe4..acdf9831c 100644 --- a/core/src/host/channel.js +++ b/core/src/host/channel.js @@ -9,6 +9,8 @@ import { unwrapUntrusted } from './delivery.js' const NOTIFY_KINDS = ['approval', 'finished', 'failed'] const LANGUAGES = { zh: 'Chinese (简体中文)', en: 'English' } +// How long a burst of the user's lines is collected before it opens one turn (#642). +const SETTLE_MS = 1_500 const invalid = message => Object.assign(new Error(message), { code: 'bot/invalid' }) // Programming and file system errors are DS Bot's own faults, logged with their stack; @@ -33,6 +35,53 @@ export function install(rt) { rt.watched = () => Date.now() - watchedAt < WATCH_MS rt.uiLanguage = () => (uiLocale === undefined ? 'English' : LANGUAGES[uiLocale.split('-')[0].toLowerCase()] ?? uiLocale) + // A burst of the user's own lines (#642): with no turn running, hold them for a moment and open + // one turn with every line in arrival order. A line that arrives while a turn runs is not held — + // it lands as its own queued item, so a burst never reorders work that is already under way. + const settling = new Map() + const turnRunning = sessionId => rt.live?.get(sessionId)?.turn !== undefined + const openSettledTurn = async (sessionId, lines) => { + try { + await rt.ensureCapabilities(sessionId) + await ctx.sessionController.prompt({ + requestId: randomUUID(), + sessionId, + mode: 'queue', + content: [{ type: 'text', text: lines.join('\n') }], + }, AbortSignal.timeout(30_000)) + } catch (error) { + rt.warn?.('ds-bot: held user lines were not delivered: %s', error) + } + } + const flushSettled = sessionId => { + const held = settling.get(sessionId) + if (held === undefined) return Promise.resolve() + settling.delete(sessionId) + clearTimeout(held.timer) + return openSettledTurn(sessionId, held.lines) + } + // True when the line was held for the window; false means the caller opens its own turn. + const settleLine = async (sessionId, text) => { + if (turnRunning(sessionId)) { + // A turn began inside the window: the held lines queue ahead of this one, in order. + await flushSettled(sessionId) + return false + } + const held = settling.get(sessionId) + if (held !== undefined) { + held.lines.push(text) + return true + } + const waiting = { lines: [text], timer: setTimeout(() => void flushSettled(sessionId), SETTLE_MS) } + waiting.timer?.unref?.() + settling.set(sessionId, waiting) + return true + } + ctx.effect(() => () => { + for (const held of settling.values()) clearTimeout(held.timer) + settling.clear() + }, 'ds-bot: settle window') + const ok = value => ({ ok: true, value }) const fail = (message, code = 'bot/error', details = {}) => ({ ok: false, error: { code, message, details } }) // Models come from DSH, not state.json, so a provider added in Harness settings @@ -467,6 +516,9 @@ export function install(rt) { await ctx.sessions.flush(agent.session) return ok(true) } + // The Bot's own chat holds a burst into one turn; a group room already joins a mid-round + // line to the round it belongs to (#642), so its line must not wait behind a timer. + if (room === undefined && await settleLine(sessionId, text)) return ok(true) await rt.ensureCapabilities(sessionId) await ctx.sessionController.prompt({ requestId: randomUUID(), diff --git a/core/test/install.test.js b/core/test/install.test.js index add152242..80f70848f 100644 --- a/core/test/install.test.js +++ b/core/test/install.test.js @@ -107,6 +107,7 @@ test('apply registers listeners, tools, and effects in a fixed order', (t) => { 'on agent/created', 'on session/event', 'on session/event', + 'effect ds-bot: settle window', 'effect ds-bot: api route', 'route POST /api/bot', 'effect ds-bot: load state', diff --git a/core/test/settle.test.js b/core/test/settle.test.js new file mode 100644 index 000000000..a44c88336 --- /dev/null +++ b/core/test/settle.test.js @@ -0,0 +1,158 @@ +import assert from 'node:assert/strict' +import { randomUUID } from 'node:crypto' +import { mkdtempSync, rmSync } from 'node:fs' +import { tmpdir } from 'node:os' +import { join } from 'node:path' +import { test } from 'node:test' +import { createRuntime } from '../index.js' +import { fakeContext } from './fake-ctx.js' + +// The user's own chat: every `send` reaches the host separately, the way the composer +// posts one card per line. Prompts are recorded in the order they open a turn. +function start() { + const home = mkdtempSync(join(tmpdir(), 'ds-bot-settle-')) + const ctx = fakeContext() + const prompts = [] + const sent = [] + const workspaces = new Map() + let sessions = 0 + Object.assign(ctx, { + llm: { + listProviders: () => [{ id: 'deepseek', name: 'DeepSeek' }], + listModels: async () => [{ id: 'deepseek-v4-flash' }], + resolveModelInfo: async () => ({}), + }, + workspaceRegistry: { + get: id => workspaces.get(id), + resolveByPath: async () => undefined, + create: async (path) => { + const workspace = { id: `w${workspaces.size + 1}`, path } + workspaces.set(workspace.id, workspace) + return workspace + }, + archiveSession: async () => {}, + }, + sessionController: { + create: async () => ({ sessionId: `s${++sessions}` }), + resolveAgent: async sessionId => ({ agent: { session: { id: sessionId }, followup: message => sent.push({ sessionId, message }) } }), + selectModel: async () => {}, + prompt: async (request) => { + prompts.push({ sessionId: request.sessionId, text: request.content.map(part => part.text).join('') }) + return { accepted: true } + }, + page: async () => ({ events: [] }), + }, + sessionTitle: { rename: async () => {} }, + sessions: { flush: async () => {} }, + }) + let route + const register = ctx.connection.fetch.register + ctx.connection.fetch.register = (options) => { + route = options + return register(options) + } + const rt = createRuntime(ctx, { home }) + const call = async (endpoint, payload) => { + const response = await route.fetch(new Request('http://localhost/api/bot', { method: 'POST', body: JSON.stringify({ endpoint, payload }) })) + return response.json() + } + const stop = async () => { + await ctx.dispose() + rmSync(home, { recursive: true, force: true, maxRetries: 5 }) + } + return { call, prompts, sent, rt, ctx, stop } +} + +// A tick hands the timer's callback its turn; the promises it starts settle on the +// microtask queue, which setImmediate drains without touching the mocked clock. +const drain = () => new Promise(resolve => setImmediate(resolve)) + +test('#642: three lines inside the window open one turn, in arrival order', async (t) => { + const { call, prompts, rt, stop } = start() + t.after(stop) + t.mock.timers.enable({ apis: ['setTimeout'] }) + await rt.load() + const chief = (await call('state', {})).value.mainBotId + + await call('send', { sessionId: chief, text: 'change that one' }) + t.mock.timers.tick(300) + await call('send', { sessionId: chief, text: 'and that one too' }) + t.mock.timers.tick(300) + await call('send', { sessionId: chief, text: 'and read the log while you are there' }) + assert.deepEqual(prompts, [], 'no turn opens while the burst is still inside the window') + + t.mock.timers.tick(1_500) + await drain() + assert.equal(prompts.length, 1, 'the burst opens one turn, not three') + assert.equal(prompts[0].text, 'change that one\nand that one too\nand read the log while you are there') +}) + +test('#642: a line after the window opens its own turn', async (t) => { + const { call, prompts, rt, stop } = start() + t.after(stop) + t.mock.timers.enable({ apis: ['setTimeout'] }) + await rt.load() + const chief = (await call('state', {})).value.mainBotId + + await call('send', { sessionId: chief, text: 'first' }) + t.mock.timers.tick(1_500) + await drain() + assert.deepEqual(prompts.map(entry => entry.text), ['first']) + + t.mock.timers.tick(500) + await call('send', { sessionId: chief, text: 'first + 2 s' }) + t.mock.timers.tick(1_500) + await drain() + assert.deepEqual(prompts.map(entry => entry.text), ['first', 'first + 2 s']) +}) + +test('#642: one line alone keeps its own turn, after the window', async (t) => { + const { call, prompts, rt, stop } = start() + t.after(stop) + t.mock.timers.enable({ apis: ['setTimeout'] }) + await rt.load() + const chief = (await call('state', {})).value.mainBotId + + await call('send', { sessionId: chief, text: 'only one line' }) + assert.deepEqual(prompts, []) + t.mock.timers.tick(1_500) + await drain() + assert.deepEqual(prompts.map(entry => entry.text), ['only one line']) +}) + +test('#642: a Bot-to-Bot line in the middle of a burst is not held by the timer', async (t) => { + const { call, prompts, sent, rt, stop } = start() + t.after(stop) + t.mock.timers.enable({ apis: ['setTimeout'] }) + await rt.load() + const chief = (await call('state', {})).value.mainBotId + const writer = (await call('create-bot', { name: 'Writer' })).value + + await call('send', { sessionId: chief, text: 'the user is typing' }) + await rt.deliver({ fromId: writer.id, toId: chief, text: 'a Bot line', role: 'request' }) + assert.equal(sent.length, 1, 'the Bot line reaches its Session at once, behind no user timer') + assert.deepEqual(prompts, []) + + t.mock.timers.tick(1_500) + await drain() + assert.deepEqual(prompts.map(entry => entry.text), ['the user is typing']) +}) + +test('#642: a line arriving while a turn runs is not held, and the held lines stay ahead of it', async (t) => { + const { call, prompts, rt, stop } = start() + t.after(stop) + t.mock.timers.enable({ apis: ['setTimeout'] }) + await rt.load() + const chief = (await call('state', {})).value.mainBotId + + await call('send', { sessionId: chief, text: 'held one' }) + // The turn starts inside the window (a schedule, or another Bot's line). + rt.liveOf(chief).turn = 'turn-1' + await call('send', { sessionId: chief, text: 'busy one' }) + await drain() + assert.deepEqual(prompts.map(entry => entry.text), ['held one', 'busy one'], 'arrival order, each as its own item') + + t.mock.timers.tick(1_500) + await drain() + assert.equal(prompts.length, 2, 'nothing is left over for the closed window') +}) diff --git a/docs/subsystems/core-host.md b/docs/subsystems/core-host.md index bc2ba098e..7555d3f9f 100644 --- a/docs/subsystems/core-host.md +++ b/docs/subsystems/core-host.md @@ -181,6 +181,12 @@ Work a member hands a Bot with `message_bot` from its group Session runs in a hi - `turns.js` keeps `endedView()` (`trimEnded`): the last 20 turns, plus, past those, each place's (Bot + group) latest turn while it is a failure younger than 7 days (at most 50, for the Agent inbox's failed turns, #471), of the user's own Bots that finished (`reason.kind === 'completed'`) or failed (`failed`: the failure's message, else its code, at most 160 characters; an AUTH message stays out as in `turnFailure`), as `{ id: '::', botId, sessionId | roomId, at, calls? }` (`calls` is that turn's tool-call count, #471; left out for 0). Process-local, never saved. Not listed: a turn stopped, switched to a new part or retried for a tool call written as text; a friend's Bot (`bot.remote`); a turn a friend started (`fromRemote`, a remote-chat line); a hidden turn another Bot asked for (`exchangeTurns`); a group turn in a room with an `origin` (face-to-face, inbound-share). It rides `live().work.ended`; `rt.ownedLive` keeps the actor's own Bots' entries. `set-prefs` takes `notifications: false` (kept) or `true` / `null` (cleared: on by default), and `notify: { approval | finished | failed: boolean }` (#469: merged into `prefs.notify`, only the `false` ones kept, removed when empty). The client's use is in [core-client](core-client.md#system-notifications-353). +### A burst of the user's lines (#642) + +- Channel `send` in the Bot's **own chat** (no room on that Session) collects the user's lines: the first line that arrives with no turn running starts a 1500 ms `SETTLE_MS` timer on that Session (`channel.js` `settleLine`), further lines inside the window join its text, and the timer opens **one** turn carrying them in arrival order, each on its own line. No sender labels, no "read N messages as one" notice, no new setting; the order is arrival order, never sorted. +- A line arriving while a turn runs is not held: the window flushes what it holds first (so the queued items stay in arrival order) and the line lands as its own queued item, as before. A Bot-to-Bot `deliver` never waits behind the window — it has its own queue and dedupe (#641). +- A group room is not collected: its line must reach the Session at once so a running round can drain it (`group_relay`, #299), so `settleLine` is only reached when `rt.roomOf(sessionId)` is undefined. The pending timer is cleared on dispose. Covered by `core/test/settle.test.js`. + ### Diagnostics - Host diagnostics never depend on `ctx.logger`: DSH may export no logger, and a test or plugin-local host may carry one without `warn`. `rt.warn` still writes stderr and the diagnostics log in that case (`src/host/diagnostics.js`). Plugins reach the same log through `flybot.warn` (#68); see [plugin-runtime](plugin-runtime.md). diff --git a/patches/ds-bot/0001-core-fixes.patch b/patches/ds-bot/0001-core-fixes.patch index b67777335..64bf4dffb 100644 --- a/patches/ds-bot/0001-core-fixes.patch +++ b/patches/ds-bot/0001-core-fixes.patch @@ -5507,7 +5507,7 @@ diff --git a/test/home-lock.test.js b/test/home-lock.test.js diff --git a/test/install.test.js b/test/install.test.js --- a/test/install.test.js +++ b/test/install.test.js -@@ -91,12 +91,22 @@ test('apply registers listeners, tools, and effects in a fixed order', (t) => { +@@ -91,12 +91,23 @@ 'on llm/stream', 'effect ds-bot: load secrets', 'effect ds-bot: connectors', @@ -5527,10 +5527,11 @@ diff --git a/test/install.test.js b/test/install.test.js + 'on agent/created', + 'on session/event', + 'on session/event', ++ 'effect ds-bot: settle window', 'effect ds-bot: api route', 'route POST /api/bot', 'effect ds-bot: load state', -@@ -137,6 +137,17 @@ test('the client channel answers from the saved state', async (t) => { +@@ -137,6 +137,17 @@ assert.equal((await call('set-prefs', { motion: 'lively' })).value.motion, 'lively') assert.equal('motion' in (await call('set-prefs', { motion: 'normal' })).value, false) assert.deepEqual((await call('set-prefs', { motion: 'wild' })).error, { code: 'bot/invalid', message: 'Motion must be quiet, normal or lively', details: {} }) diff --git a/patches/ds-bot/0005-social-seams.patch b/patches/ds-bot/0005-social-seams.patch index 44110ecc6..bf2ce4271 100644 --- a/patches/ds-bot/0005-social-seams.patch +++ b/patches/ds-bot/0005-social-seams.patch @@ -221,7 +221,7 @@ diff --git a/src/client/mark.js b/src/client/mark.js diff --git a/src/host/channel.js b/src/host/channel.js --- a/src/host/channel.js +++ b/src/host/channel.js -@@ -3,7 +3,11 @@ import { randomUUID } from 'node:crypto' +@@ -3,8 +3,14 @@ import { createUserMessage } from '@deepseek-ai/dsh-llm' import { questionReply } from './parts.js' import { COLORS, EXTRA_COLORS, ROUTE } from './constants.js' @@ -231,9 +231,12 @@ diff --git a/src/host/channel.js b/src/host/channel.js +const NOTIFY_KINDS = ['approval', 'finished', 'failed'] const LANGUAGES = { zh: 'Chinese (简体中文)', en: 'English' } ++// How long a burst of the user's lines is collected before it opens one turn (#642). ++const SETTLE_MS = 1_500 const invalid = message => Object.assign(new Error(message), { code: 'bot/invalid' }) -@@ -21,14 +21,21 @@ export function install(rt) { + // Programming and file system errors are DS Bot's own faults, logged with their stack; +@@ -21,14 +21,68 @@ // The DSH interface language of the last browser that asked for the team. let uiLocale @@ -245,6 +248,53 @@ diff --git a/src/host/channel.js b/src/host/channel.js + let watchedAt = 0 + rt.watched = () => Date.now() - watchedAt < WATCH_MS + rt.uiLanguage = () => (uiLocale === undefined ? 'English' : LANGUAGES[uiLocale.split('-')[0].toLowerCase()] ?? uiLocale) ++ ++ // A burst of the user's own lines (#642): with no turn running, hold them for a moment and open ++ // one turn with every line in arrival order. A line that arrives while a turn runs is not held — ++ // it lands as its own queued item, so a burst never reorders work that is already under way. ++ const settling = new Map() ++ const turnRunning = sessionId => rt.live?.get(sessionId)?.turn !== undefined ++ const openSettledTurn = async (sessionId, lines) => { ++ try { ++ await rt.ensureCapabilities(sessionId) ++ await ctx.sessionController.prompt({ ++ requestId: randomUUID(), ++ sessionId, ++ mode: 'queue', ++ content: [{ type: 'text', text: lines.join('\n') }], ++ }, AbortSignal.timeout(30_000)) ++ } catch (error) { ++ rt.warn?.('ds-bot: held user lines were not delivered: %s', error) ++ } ++ } ++ const flushSettled = sessionId => { ++ const held = settling.get(sessionId) ++ if (held === undefined) return Promise.resolve() ++ settling.delete(sessionId) ++ clearTimeout(held.timer) ++ return openSettledTurn(sessionId, held.lines) ++ } ++ // True when the line was held for the window; false means the caller opens its own turn. ++ const settleLine = async (sessionId, text) => { ++ if (turnRunning(sessionId)) { ++ // A turn began inside the window: the held lines queue ahead of this one, in order. ++ await flushSettled(sessionId) ++ return false ++ } ++ const held = settling.get(sessionId) ++ if (held !== undefined) { ++ held.lines.push(text) ++ return true ++ } ++ const waiting = { lines: [text], timer: setTimeout(() => void flushSettled(sessionId), SETTLE_MS) } ++ waiting.timer?.unref?.() ++ settling.set(sessionId, waiting) ++ return true ++ } ++ ctx.effect(() => () => { ++ for (const held of settling.values()) clearTimeout(held.timer) ++ settling.clear() ++ }, 'ds-bot: settle window') const ok = value => ({ ok: true, value }) - const fail = (message, code = 'bot/error') => ({ ok: false, error: { code, message, details: {} } }) @@ -258,7 +308,7 @@ diff --git a/src/host/channel.js b/src/host/channel.js revision: state.revision, workspaceId: state.workspaceId ?? null, // `mainBotId` is the first Main Bot, which the sidebar shows as its tile. -@@ -36,17 +36,52 @@ export function install(rt) { +@@ -36,17 +36,52 @@ mainBotIds: state.mainBotIds, // `sessionId` is the Bot's current own-chat part; `parts` its earlier ones. // `waiting`: the Bot holds its greeting until the user picks its model. @@ -316,7 +366,7 @@ diff --git a/src/host/channel.js b/src/host/channel.js // The team's DSH Agent (at most one): plain dsh Sessions, not Bots. agents: Object.values(state.agents), // Key names and who may use them; values never leave the host. -@@ -54,7 +54,8 @@ export function install(rt) { +@@ -54,7 +54,8 @@ capabilities: rt.capabilityLibraryView(), secretRequests: state.secretRequests ?? {}, colors: [...COLORS, ...EXTRA_COLORS], @@ -326,7 +376,7 @@ diff --git a/src/host/channel.js b/src/host/channel.js defaultModel: rt.cachedDefault() ?? null, permissions: rt.permissionCatalog?.() ?? null, modelsKey: modelsKey(), -@@ -65,22 +65,49 @@ export function install(rt) { +@@ -65,22 +65,49 @@ memory: rt.memoryEnabled !== false, // Set while another dsh host holds the team (store.js): why this one cannot change it. readOnly: rt.readOnly() ?? null, @@ -387,7 +437,7 @@ diff --git a/src/host/channel.js b/src/host/channel.js function setPrefs(body) { if (body.theme !== undefined) { -@@ -97,28 +97,57 @@ export function install(rt) { +@@ -97,28 +97,57 @@ else if (body.motion === 'quiet' || body.motion === 'lively') state.prefs.motion = body.motion else throw invalid('Motion must be quiet, normal or lively') } @@ -449,7 +499,7 @@ diff --git a/src/host/channel.js b/src/host/channel.js case 'duplicate-bot': await rt.load() return ok(await rt.duplicateBot(body.id)) -@@ -141,7 +141,7 @@ export function install(rt) { +@@ -141,7 +141,7 @@ return ok([...rt.roomProgress.entries()].filter(([, entry]) => entry.roomId === body.roomId).map(([callId, entry]) => ({ callId, replies: entry.replies, speaking: entry.speaking ?? null }))) case 'create-bot': { await rt.bootstrap() @@ -458,7 +508,7 @@ diff --git a/src/host/channel.js b/src/host/channel.js return ok(bot) } case 'update-bot': -@@ -150,17 +150,23 @@ export function install(rt) { +@@ -150,17 +150,23 @@ case 'delete-bot': await rt.load() await rt.deleteBot(body.id) @@ -485,7 +535,7 @@ diff --git a/src/host/channel.js b/src/host/channel.js if (body.greet === true) await rt.greetRoom(room.id).catch(error => rt.warn('ds-bot: greeting in %s failed: %s', room.name, error)) return ok(room) } -@@ -175,8 +175,26 @@ export function install(rt) { +@@ -175,8 +175,26 @@ add: Array.isArray(body.add) ? body.add : undefined, remove: Array.isArray(body.remove) ? body.remove : undefined, workDir: body.workDir, @@ -514,7 +564,7 @@ diff --git a/src/host/channel.js b/src/host/channel.js } case 'folders': case 'folder-pick': -@@ -203,17 +203,19 @@ export function install(rt) { +@@ -203,17 +203,19 @@ case 'remove-agent': await rt.load() await rt.removeAgent(body.agentId) @@ -537,7 +587,7 @@ diff --git a/src/host/channel.js b/src/host/channel.js case 'stop-member-work': return ok(await rt.stopMemberWork(String(body.roomId ?? ''), String(body.botId ?? ''))) case 'dismiss-question': -@@ -231,6 +231,9 @@ export function install(rt) { +@@ -231,6 +231,9 @@ await rt.load() return ok(await rt.secretEndpoint(endpoint, body)) case 'connector-connect': @@ -547,7 +597,7 @@ diff --git a/src/host/channel.js b/src/host/channel.js case 'connector-disconnect': case 'connector-retry': await rt.load() -@@ -241,11 +241,22 @@ export function install(rt) { +@@ -241,11 +241,22 @@ case 'bot-capability': await rt.load() return ok(await rt.capabilityEndpoint(endpoint, body)) @@ -571,7 +621,7 @@ diff --git a/src/host/channel.js b/src/host/channel.js case 'memory-view': case 'memory-ask': case 'memory-forget': -@@ -258,8 +258,17 @@ export function install(rt) { +@@ -258,8 +258,17 @@ const sessionId = rt.botOf(body.sessionId) ? rt.chatOf(body.sessionId) : past && past.roomId === undefined ? rt.chatOf(past.botId) : body.sessionId const question = state.questions[sessionId] if (question === undefined || question.id !== body.questionId) return fail('This question is no longer open', 'bot/stale') @@ -589,7 +639,7 @@ diff --git a/src/host/channel.js b/src/host/channel.js // Claim the question before queueing so a second answer is refused. delete state.questions[sessionId] try { -@@ -278,9 +278,43 @@ export function install(rt) { +@@ -278,9 +278,46 @@ const text = String(body.text ?? '').trim() if (text === '' || typeof body.sessionId !== 'string') return fail('Nothing to send', 'bot/invalid') await rt.load() @@ -630,10 +680,13 @@ diff --git a/src/host/channel.js b/src/host/channel.js + await ctx.sessions.flush(agent.session) + return ok(true) + } ++ // The Bot's own chat holds a burst into one turn; a group room already joins a mid-round ++ // line to the round it belongs to (#642), so its line must not wait behind a timer. ++ if (room === undefined && await settleLine(sessionId, text)) return ok(true) await rt.ensureCapabilities(sessionId) await ctx.sessionController.prompt({ requestId: randomUUID(), -@@ -319,32 +319,112 @@ export function install(rt) { +@@ -319,32 +319,112 @@ case 'diagnostics-clear': await rt.diagnostics.clear() return ok(true) diff --git a/patches/ds-bot/files/test/settle.test.js b/patches/ds-bot/files/test/settle.test.js new file mode 100644 index 000000000..a44c88336 --- /dev/null +++ b/patches/ds-bot/files/test/settle.test.js @@ -0,0 +1,158 @@ +import assert from 'node:assert/strict' +import { randomUUID } from 'node:crypto' +import { mkdtempSync, rmSync } from 'node:fs' +import { tmpdir } from 'node:os' +import { join } from 'node:path' +import { test } from 'node:test' +import { createRuntime } from '../index.js' +import { fakeContext } from './fake-ctx.js' + +// The user's own chat: every `send` reaches the host separately, the way the composer +// posts one card per line. Prompts are recorded in the order they open a turn. +function start() { + const home = mkdtempSync(join(tmpdir(), 'ds-bot-settle-')) + const ctx = fakeContext() + const prompts = [] + const sent = [] + const workspaces = new Map() + let sessions = 0 + Object.assign(ctx, { + llm: { + listProviders: () => [{ id: 'deepseek', name: 'DeepSeek' }], + listModels: async () => [{ id: 'deepseek-v4-flash' }], + resolveModelInfo: async () => ({}), + }, + workspaceRegistry: { + get: id => workspaces.get(id), + resolveByPath: async () => undefined, + create: async (path) => { + const workspace = { id: `w${workspaces.size + 1}`, path } + workspaces.set(workspace.id, workspace) + return workspace + }, + archiveSession: async () => {}, + }, + sessionController: { + create: async () => ({ sessionId: `s${++sessions}` }), + resolveAgent: async sessionId => ({ agent: { session: { id: sessionId }, followup: message => sent.push({ sessionId, message }) } }), + selectModel: async () => {}, + prompt: async (request) => { + prompts.push({ sessionId: request.sessionId, text: request.content.map(part => part.text).join('') }) + return { accepted: true } + }, + page: async () => ({ events: [] }), + }, + sessionTitle: { rename: async () => {} }, + sessions: { flush: async () => {} }, + }) + let route + const register = ctx.connection.fetch.register + ctx.connection.fetch.register = (options) => { + route = options + return register(options) + } + const rt = createRuntime(ctx, { home }) + const call = async (endpoint, payload) => { + const response = await route.fetch(new Request('http://localhost/api/bot', { method: 'POST', body: JSON.stringify({ endpoint, payload }) })) + return response.json() + } + const stop = async () => { + await ctx.dispose() + rmSync(home, { recursive: true, force: true, maxRetries: 5 }) + } + return { call, prompts, sent, rt, ctx, stop } +} + +// A tick hands the timer's callback its turn; the promises it starts settle on the +// microtask queue, which setImmediate drains without touching the mocked clock. +const drain = () => new Promise(resolve => setImmediate(resolve)) + +test('#642: three lines inside the window open one turn, in arrival order', async (t) => { + const { call, prompts, rt, stop } = start() + t.after(stop) + t.mock.timers.enable({ apis: ['setTimeout'] }) + await rt.load() + const chief = (await call('state', {})).value.mainBotId + + await call('send', { sessionId: chief, text: 'change that one' }) + t.mock.timers.tick(300) + await call('send', { sessionId: chief, text: 'and that one too' }) + t.mock.timers.tick(300) + await call('send', { sessionId: chief, text: 'and read the log while you are there' }) + assert.deepEqual(prompts, [], 'no turn opens while the burst is still inside the window') + + t.mock.timers.tick(1_500) + await drain() + assert.equal(prompts.length, 1, 'the burst opens one turn, not three') + assert.equal(prompts[0].text, 'change that one\nand that one too\nand read the log while you are there') +}) + +test('#642: a line after the window opens its own turn', async (t) => { + const { call, prompts, rt, stop } = start() + t.after(stop) + t.mock.timers.enable({ apis: ['setTimeout'] }) + await rt.load() + const chief = (await call('state', {})).value.mainBotId + + await call('send', { sessionId: chief, text: 'first' }) + t.mock.timers.tick(1_500) + await drain() + assert.deepEqual(prompts.map(entry => entry.text), ['first']) + + t.mock.timers.tick(500) + await call('send', { sessionId: chief, text: 'first + 2 s' }) + t.mock.timers.tick(1_500) + await drain() + assert.deepEqual(prompts.map(entry => entry.text), ['first', 'first + 2 s']) +}) + +test('#642: one line alone keeps its own turn, after the window', async (t) => { + const { call, prompts, rt, stop } = start() + t.after(stop) + t.mock.timers.enable({ apis: ['setTimeout'] }) + await rt.load() + const chief = (await call('state', {})).value.mainBotId + + await call('send', { sessionId: chief, text: 'only one line' }) + assert.deepEqual(prompts, []) + t.mock.timers.tick(1_500) + await drain() + assert.deepEqual(prompts.map(entry => entry.text), ['only one line']) +}) + +test('#642: a Bot-to-Bot line in the middle of a burst is not held by the timer', async (t) => { + const { call, prompts, sent, rt, stop } = start() + t.after(stop) + t.mock.timers.enable({ apis: ['setTimeout'] }) + await rt.load() + const chief = (await call('state', {})).value.mainBotId + const writer = (await call('create-bot', { name: 'Writer' })).value + + await call('send', { sessionId: chief, text: 'the user is typing' }) + await rt.deliver({ fromId: writer.id, toId: chief, text: 'a Bot line', role: 'request' }) + assert.equal(sent.length, 1, 'the Bot line reaches its Session at once, behind no user timer') + assert.deepEqual(prompts, []) + + t.mock.timers.tick(1_500) + await drain() + assert.deepEqual(prompts.map(entry => entry.text), ['the user is typing']) +}) + +test('#642: a line arriving while a turn runs is not held, and the held lines stay ahead of it', async (t) => { + const { call, prompts, rt, stop } = start() + t.after(stop) + t.mock.timers.enable({ apis: ['setTimeout'] }) + await rt.load() + const chief = (await call('state', {})).value.mainBotId + + await call('send', { sessionId: chief, text: 'held one' }) + // The turn starts inside the window (a schedule, or another Bot's line). + rt.liveOf(chief).turn = 'turn-1' + await call('send', { sessionId: chief, text: 'busy one' }) + await drain() + assert.deepEqual(prompts.map(entry => entry.text), ['held one', 'busy one'], 'arrival order, each as its own item') + + t.mock.timers.tick(1_500) + await drain() + assert.equal(prompts.length, 2, 'nothing is left over for the closed window') +}) From eff9f5abb1533ef0aeb7a2d68df3d8ab68fff5db Mon Sep 17 00:00:00 2001 From: Danyang Chen Date: Thu, 22 Oct 2026 12:55:08 +0800 Subject: [PATCH 2/3] Drop the unused randomUUID import in the settle test Lint: Identifier 'randomUUID' is imported but never used. --- core/test/settle.test.js | 1 - patches/ds-bot/files/test/settle.test.js | 1 - 2 files changed, 2 deletions(-) diff --git a/core/test/settle.test.js b/core/test/settle.test.js index a44c88336..395d848fe 100644 --- a/core/test/settle.test.js +++ b/core/test/settle.test.js @@ -1,5 +1,4 @@ import assert from 'node:assert/strict' -import { randomUUID } from 'node:crypto' import { mkdtempSync, rmSync } from 'node:fs' import { tmpdir } from 'node:os' import { join } from 'node:path' diff --git a/patches/ds-bot/files/test/settle.test.js b/patches/ds-bot/files/test/settle.test.js index a44c88336..395d848fe 100644 --- a/patches/ds-bot/files/test/settle.test.js +++ b/patches/ds-bot/files/test/settle.test.js @@ -1,5 +1,4 @@ import assert from 'node:assert/strict' -import { randomUUID } from 'node:crypto' import { mkdtempSync, rmSync } from 'node:fs' import { tmpdir } from 'node:os' import { join } from 'node:path' From a3cbbcdc848afa1b21abfce0d9c15061669b205a Mon Sep 17 00:00:00 2001 From: Danyang Chen Date: Thu, 22 Oct 2026 13:47:44 +0800 Subject: [PATCH 3/3] Report a held line's failure and keep it The settle window answered ok(true) and swallowed a prompt error, so a line the host could not deliver disappeared while the chat showed it as sent, and the held lines were dropped before the prompt ran. flushSettled() now keeps them, re-arms the window a few times, and rethrows so the send path answers the caller the error instead of ok; the lines go out with the next line after the retries (#682). --- core/src/host/channel.js | 65 +++++++++++++++------- core/test/settle.test.js | 51 ++++++++++++++++++ patches/ds-bot/0005-social-seams.patch | 69 +++++++++++++++++------- patches/ds-bot/files/test/settle.test.js | 51 ++++++++++++++++++ 4 files changed, 198 insertions(+), 38 deletions(-) diff --git a/core/src/host/channel.js b/core/src/host/channel.js index acdf9831c..a59ee364b 100644 --- a/core/src/host/channel.js +++ b/core/src/host/channel.js @@ -11,6 +11,8 @@ const NOTIFY_KINDS = ['approval', 'finished', 'failed'] const LANGUAGES = { zh: 'Chinese (简体中文)', en: 'English' } // How long a burst of the user's lines is collected before it opens one turn (#642). const SETTLE_MS = 1_500 +// How many windows a held burst is retried on its own before it waits for the user's next line. +const SETTLE_RETRIES = 3 const invalid = message => Object.assign(new Error(message), { code: 'bot/invalid' }) // Programming and file system errors are DS Bot's own faults, logged with their stack; @@ -40,40 +42,67 @@ export function install(rt) { // it lands as its own queued item, so a burst never reorders work that is already under way. const settling = new Map() const turnRunning = sessionId => rt.live?.get(sessionId)?.turn !== undefined + const armSettled = (sessionId, held) => { + clearTimeout(held.timer) + held.timer = setTimeout(() => { + void flushSettled(sessionId).catch(error => rt.warn?.('ds-bot: held user lines were not delivered; they are kept: %s', error)) + }, SETTLE_MS) + held.timer?.unref?.() + } const openSettledTurn = async (sessionId, lines) => { - try { - await rt.ensureCapabilities(sessionId) - await ctx.sessionController.prompt({ - requestId: randomUUID(), - sessionId, - mode: 'queue', - content: [{ type: 'text', text: lines.join('\n') }], - }, AbortSignal.timeout(30_000)) - } catch (error) { - rt.warn?.('ds-bot: held user lines were not delivered: %s', error) - } + await rt.ensureCapabilities(sessionId) + await ctx.sessionController.prompt({ + requestId: randomUUID(), + sessionId, + mode: 'queue', + content: [{ type: 'text', text: lines.join('\n') }], + }, AbortSignal.timeout(30_000)) } - const flushSettled = sessionId => { + // Throws when the turn could not be opened (#682): the caller that is still waiting for an + // answer gets the error, and the lines stay held — a held line is not dropped because a + // prompt failed, and `send` never answers ok for a line that then disappears. + const flushSettled = async (sessionId) => { const held = settling.get(sessionId) - if (held === undefined) return Promise.resolve() - settling.delete(sessionId) + if (held === undefined) return clearTimeout(held.timer) - return openSettledTurn(sessionId, held.lines) + held.timer = undefined + const lines = held.lines.splice(0) + try { + await openSettledTurn(sessionId, lines) + } catch (error) { + held.lines.unshift(...lines) + held.failures += 1 + // Retried on the next window a few times; past that the lines wait for the user's next + // line instead of retrying on their own forever. + if (held.failures < SETTLE_RETRIES) armSettled(sessionId, held) + else rt.warn?.('ds-bot: %d held user lines for %s could not be delivered after %d tries; they go out with the next line', lines.length, sessionId, held.failures) + throw error + } + held.failures = 0 + // A line that arrived while the prompt was in flight stays in order behind the ones just sent. + if (held.lines.length > 0) armSettled(sessionId, held) + else settling.delete(sessionId) } // True when the line was held for the window; false means the caller opens its own turn. const settleLine = async (sessionId, text) => { if (turnRunning(sessionId)) { - // A turn began inside the window: the held lines queue ahead of this one, in order. + // A turn began inside the window: the held lines queue ahead of this one, in order, and a + // failure there is the caller's to report. await flushSettled(sessionId) return false } const held = settling.get(sessionId) if (held !== undefined) { held.lines.push(text) + // A failed flush past its retries leaves the entry without a timer: this line re-arms it. + if (held.timer === undefined) { + held.failures = 0 + armSettled(sessionId, held) + } return true } - const waiting = { lines: [text], timer: setTimeout(() => void flushSettled(sessionId), SETTLE_MS) } - waiting.timer?.unref?.() + const waiting = { lines: [text], failures: 0, timer: undefined } + armSettled(sessionId, waiting) settling.set(sessionId, waiting) return true } diff --git a/core/test/settle.test.js b/core/test/settle.test.js index 395d848fe..0c8be70d8 100644 --- a/core/test/settle.test.js +++ b/core/test/settle.test.js @@ -155,3 +155,54 @@ test('#642: a line arriving while a turn runs is not held, and the held lines st await drain() assert.equal(prompts.length, 2, 'nothing is left over for the closed window') }) + +test('#682: a failed flush keeps the held line and retries on the next window', async (t) => { + const { call, prompts, ctx, rt, stop } = start() + t.after(stop) + t.mock.timers.enable({ apis: ['setTimeout'] }) + await rt.load() + const chief = (await call('state', {})).value.mainBotId + const orig = ctx.sessionController.prompt + let fail = true + ctx.sessionController.prompt = async (request) => { + if (fail) throw Object.assign(new Error('ds-bot: the Session is gone'), { code: 'bot/refused' }) + return orig(request) + } + + assert.equal((await call('send', { sessionId: chief, text: 'keep me' })).ok, true) + t.mock.timers.tick(1_500) + await drain() + assert.deepEqual(prompts, [], 'the failed prompt delivered nothing') + + fail = false + t.mock.timers.tick(1_500) + await drain() + assert.deepEqual(prompts.map(entry => entry.text), ['keep me'], 'the held line went out on the retry, in order') +}) + +test('#682: a failed flush on the send path answers the sender the error, not ok', async (t) => { + const { call, prompts, ctx, rt, stop } = start() + t.after(stop) + t.mock.timers.enable({ apis: ['setTimeout'] }) + await rt.load() + const chief = (await call('state', {})).value.mainBotId + const orig = ctx.sessionController.prompt + let fail = true + ctx.sessionController.prompt = async (request) => { + if (fail) throw Object.assign(new Error('ds-bot: the Session is gone'), { code: 'bot/refused' }) + return orig(request) + } + + await call('send', { sessionId: chief, text: 'held one' }) + // The turn starts inside the window, so the next line must not be held behind the timer. + rt.liveOf(chief).turn = 'turn-1' + const refused = await call('send', { sessionId: chief, text: 'busy one' }) + assert.equal(refused.ok, false, 'a line the host could not deliver is not reported as sent') + assert.equal(refused.error.code, 'bot/refused') + assert.deepEqual(prompts, []) + + fail = false + const retried = await call('send', { sessionId: chief, text: 'busy again' }) + assert.equal(retried.ok, true) + assert.deepEqual(prompts.map(entry => entry.text), ['held one', 'busy again'], 'the kept line still goes first') +}) diff --git a/patches/ds-bot/0005-social-seams.patch b/patches/ds-bot/0005-social-seams.patch index bf2ce4271..99ec5df70 100644 --- a/patches/ds-bot/0005-social-seams.patch +++ b/patches/ds-bot/0005-social-seams.patch @@ -221,7 +221,7 @@ diff --git a/src/client/mark.js b/src/client/mark.js diff --git a/src/host/channel.js b/src/host/channel.js --- a/src/host/channel.js +++ b/src/host/channel.js -@@ -3,8 +3,14 @@ +@@ -3,8 +3,16 @@ import { createUserMessage } from '@deepseek-ai/dsh-llm' import { questionReply } from './parts.js' import { COLORS, EXTRA_COLORS, ROUTE } from './constants.js' @@ -233,10 +233,12 @@ diff --git a/src/host/channel.js b/src/host/channel.js const LANGUAGES = { zh: 'Chinese (简体中文)', en: 'English' } +// How long a burst of the user's lines is collected before it opens one turn (#642). +const SETTLE_MS = 1_500 ++// How many windows a held burst is retried on its own before it waits for the user's next line. ++const SETTLE_RETRIES = 3 const invalid = message => Object.assign(new Error(message), { code: 'bot/invalid' }) // Programming and file system errors are DS Bot's own faults, logged with their stack; -@@ -21,14 +21,68 @@ +@@ -21,14 +21,95 @@ // The DSH interface language of the last browser that asked for the team. let uiLocale @@ -254,40 +256,67 @@ diff --git a/src/host/channel.js b/src/host/channel.js + // it lands as its own queued item, so a burst never reorders work that is already under way. + const settling = new Map() + const turnRunning = sessionId => rt.live?.get(sessionId)?.turn !== undefined ++ const armSettled = (sessionId, held) => { ++ clearTimeout(held.timer) ++ held.timer = setTimeout(() => { ++ void flushSettled(sessionId).catch(error => rt.warn?.('ds-bot: held user lines were not delivered; they are kept: %s', error)) ++ }, SETTLE_MS) ++ held.timer?.unref?.() ++ } + const openSettledTurn = async (sessionId, lines) => { -+ try { -+ await rt.ensureCapabilities(sessionId) -+ await ctx.sessionController.prompt({ -+ requestId: randomUUID(), -+ sessionId, -+ mode: 'queue', -+ content: [{ type: 'text', text: lines.join('\n') }], -+ }, AbortSignal.timeout(30_000)) -+ } catch (error) { -+ rt.warn?.('ds-bot: held user lines were not delivered: %s', error) -+ } ++ await rt.ensureCapabilities(sessionId) ++ await ctx.sessionController.prompt({ ++ requestId: randomUUID(), ++ sessionId, ++ mode: 'queue', ++ content: [{ type: 'text', text: lines.join('\n') }], ++ }, AbortSignal.timeout(30_000)) + } -+ const flushSettled = sessionId => { ++ // Throws when the turn could not be opened (#682): the caller that is still waiting for an ++ // answer gets the error, and the lines stay held — a held line is not dropped because a ++ // prompt failed, and `send` never answers ok for a line that then disappears. ++ const flushSettled = async (sessionId) => { + const held = settling.get(sessionId) -+ if (held === undefined) return Promise.resolve() -+ settling.delete(sessionId) ++ if (held === undefined) return + clearTimeout(held.timer) -+ return openSettledTurn(sessionId, held.lines) ++ held.timer = undefined ++ const lines = held.lines.splice(0) ++ try { ++ await openSettledTurn(sessionId, lines) ++ } catch (error) { ++ held.lines.unshift(...lines) ++ held.failures += 1 ++ // Retried on the next window a few times; past that the lines wait for the user's next ++ // line instead of retrying on their own forever. ++ if (held.failures < SETTLE_RETRIES) armSettled(sessionId, held) ++ else rt.warn?.('ds-bot: %d held user lines for %s could not be delivered after %d tries; they go out with the next line', lines.length, sessionId, held.failures) ++ throw error ++ } ++ held.failures = 0 ++ // A line that arrived while the prompt was in flight stays in order behind the ones just sent. ++ if (held.lines.length > 0) armSettled(sessionId, held) ++ else settling.delete(sessionId) + } + // True when the line was held for the window; false means the caller opens its own turn. + const settleLine = async (sessionId, text) => { + if (turnRunning(sessionId)) { -+ // A turn began inside the window: the held lines queue ahead of this one, in order. ++ // A turn began inside the window: the held lines queue ahead of this one, in order, and a ++ // failure there is the caller's to report. + await flushSettled(sessionId) + return false + } + const held = settling.get(sessionId) + if (held !== undefined) { + held.lines.push(text) ++ // A failed flush past its retries leaves the entry without a timer: this line re-arms it. ++ if (held.timer === undefined) { ++ held.failures = 0 ++ armSettled(sessionId, held) ++ } + return true + } -+ const waiting = { lines: [text], timer: setTimeout(() => void flushSettled(sessionId), SETTLE_MS) } -+ waiting.timer?.unref?.() ++ const waiting = { lines: [text], failures: 0, timer: undefined } ++ armSettled(sessionId, waiting) + settling.set(sessionId, waiting) + return true + } diff --git a/patches/ds-bot/files/test/settle.test.js b/patches/ds-bot/files/test/settle.test.js index 395d848fe..0c8be70d8 100644 --- a/patches/ds-bot/files/test/settle.test.js +++ b/patches/ds-bot/files/test/settle.test.js @@ -155,3 +155,54 @@ test('#642: a line arriving while a turn runs is not held, and the held lines st await drain() assert.equal(prompts.length, 2, 'nothing is left over for the closed window') }) + +test('#682: a failed flush keeps the held line and retries on the next window', async (t) => { + const { call, prompts, ctx, rt, stop } = start() + t.after(stop) + t.mock.timers.enable({ apis: ['setTimeout'] }) + await rt.load() + const chief = (await call('state', {})).value.mainBotId + const orig = ctx.sessionController.prompt + let fail = true + ctx.sessionController.prompt = async (request) => { + if (fail) throw Object.assign(new Error('ds-bot: the Session is gone'), { code: 'bot/refused' }) + return orig(request) + } + + assert.equal((await call('send', { sessionId: chief, text: 'keep me' })).ok, true) + t.mock.timers.tick(1_500) + await drain() + assert.deepEqual(prompts, [], 'the failed prompt delivered nothing') + + fail = false + t.mock.timers.tick(1_500) + await drain() + assert.deepEqual(prompts.map(entry => entry.text), ['keep me'], 'the held line went out on the retry, in order') +}) + +test('#682: a failed flush on the send path answers the sender the error, not ok', async (t) => { + const { call, prompts, ctx, rt, stop } = start() + t.after(stop) + t.mock.timers.enable({ apis: ['setTimeout'] }) + await rt.load() + const chief = (await call('state', {})).value.mainBotId + const orig = ctx.sessionController.prompt + let fail = true + ctx.sessionController.prompt = async (request) => { + if (fail) throw Object.assign(new Error('ds-bot: the Session is gone'), { code: 'bot/refused' }) + return orig(request) + } + + await call('send', { sessionId: chief, text: 'held one' }) + // The turn starts inside the window, so the next line must not be held behind the timer. + rt.liveOf(chief).turn = 'turn-1' + const refused = await call('send', { sessionId: chief, text: 'busy one' }) + assert.equal(refused.ok, false, 'a line the host could not deliver is not reported as sent') + assert.equal(refused.error.code, 'bot/refused') + assert.deepEqual(prompts, []) + + fail = false + const retried = await call('send', { sessionId: chief, text: 'busy again' }) + assert.equal(retried.ok, true) + assert.deepEqual(prompts.map(entry => entry.text), ['held one', 'busy again'], 'the kept line still goes first') +})