Coalesce a burst of the user's messages into one turn

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

core/src/host/channel.js

@@ -9,6 +9,10 @@ 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
+// 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;
@@ -33,6 +37,80 @@ 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 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) => {
+    await rt.ensureCapabilities(sessionId)
+    await ctx.sessionController.prompt({
+      requestId: randomUUID(),
+      sessionId,
+      mode: 'queue',
+      content: [{ type: 'text', text: lines.join('\n') }],
+    }, AbortSignal.timeout(30_000))
+  }
+  // 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
+    clearTimeout(held.timer)
+    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, 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], failures: 0, timer: undefined }
+    armSettled(sessionId, waiting)
+    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 +545,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(),

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',

core/test/settle.test.js

@@ -0,0 +1,208 @@
+import assert from 'node:assert/strict'
+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')
+})
+
+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')
+})

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: '<session>:<turn>:<at>', 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).

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