From 8ef5db8de2815389057f82a8aea42336f63098e0 Mon Sep 17 00:00:00 2001 From: Danyang Chen Date: Thu, 29 Oct 2026 10:49:01 +0800 Subject: [PATCH 1/2] Interrupt a turn when a human line arrives after a 60s tool wait Steer only admits a human line at the next step boundary, so a foreground tool or job wait holds the turn and the line waits for the wall clock. Watch the Session events for an open tool call, and on a human message in the Bot own chat past 60s cancel the turn the way Stop does with keepInbox, so the line survives and is claimed at once. The threshold is fixed; no settings key. Fixes #1509 docs/subsystems/core-host.md carries the behaviour and the 60s line. --- changelog.d/1509-steering-interrupt.fixed.md | 1 + core/index.js | 3 +- core/src/host/steering-interrupt.js | 93 ++++++++++++++++++ core/test/steering-interrupt.test.js | 97 +++++++++++++++++++ docs/subsystems/core-host.md | 6 +- patches/ds-bot/0002-flybot-api-host.patch | 5 +- .../files/src/host/steering-interrupt.js | 93 ++++++++++++++++++ .../files/test/steering-interrupt.test.js | 97 +++++++++++++++++++ 8 files changed, 391 insertions(+), 4 deletions(-) create mode 100644 changelog.d/1509-steering-interrupt.fixed.md create mode 100644 core/src/host/steering-interrupt.js create mode 100644 core/test/steering-interrupt.test.js create mode 100644 patches/ds-bot/files/src/host/steering-interrupt.js create mode 100644 patches/ds-bot/files/test/steering-interrupt.test.js diff --git a/changelog.d/1509-steering-interrupt.fixed.md b/changelog.d/1509-steering-interrupt.fixed.md new file mode 100644 index 000000000..8a843e2f1 --- /dev/null +++ b/changelog.d/1509-steering-interrupt.fixed.md @@ -0,0 +1 @@ +A human message sent while a foreground tool or job wait has been open for more than 60 seconds now stops that turn the way **Stop** does and keeps the message in the inbox, so the line is claimed at once instead of waiting for the command to finish. Shorter waits, finished waits, group rooms and idle Bots are untouched ([#1509](https://github.com/ThinkFlowLab/FlyBot/issues/1509)). diff --git a/core/index.js b/core/index.js index 6eaebca19..c7482f0b0 100644 --- a/core/index.js +++ b/core/index.js @@ -15,6 +15,7 @@ import * as models from './src/host/models.js' import * as permissions from './src/host/permissions.js' import * as turns from './src/host/turns.js' import * as busyAck from './src/host/busy-ack.js' +import * as steeringInterrupt from './src/host/steering-interrupt.js' import * as delivery from './src/host/delivery.js' import * as roster from './src/host/roster.js' import * as folders from './src/host/folders.js' @@ -54,7 +55,7 @@ export const inject = ['llm', 'tools', 'systemPrompt', 'sessionController', 'ses // the functions other modules call; they call them through `rt` at run time, never during // install. Listeners run and tools are listed in registration order, so this order holds: // usage meters model requests after every listener that rewrites them. -const MODULES = [diagnostics, store, harness, memberAuth, authz, models, permissions, turns, busyAck, delivery, roster, folders, agents, prompt, timeContext, tools, decisions, chatIndex, groups, roomMembership, reach, tasks, taskDrive, approvals, resume, memory, parts, images, secrets, connectors, connectorAccess, capabilities, schedules, automationDefault, browser, usage, update, dshUpdate, hooks, remoteChat, workflowStore, channel] +const MODULES = [diagnostics, store, harness, memberAuth, authz, models, permissions, turns, busyAck, steeringInterrupt, delivery, roster, folders, agents, prompt, timeContext, tools, decisions, chatIndex, groups, roomMembership, reach, tasks, taskDrive, approvals, resume, memory, parts, images, secrets, connectors, connectorAccess, capabilities, schedules, automationDefault, browser, usage, update, dshUpdate, hooks, remoteChat, workflowStore, channel] export function apply(ctx, config = {}) { const rt = createRuntime(ctx, config) diff --git a/core/src/host/steering-interrupt.js b/core/src/host/steering-interrupt.js new file mode 100644 index 000000000..7632b489e --- /dev/null +++ b/core/src/host/steering-interrupt.js @@ -0,0 +1,93 @@ +// Steering interrupt (#1509): a human message that arrives while a foreground tool/job +// wait has been open for more than 60s aborts that turn the way Stop does, keeping the +// inbox so the message is claimed at once instead of waiting for the blocking step. +export const INTERRUPT_AFTER_MS = 60_000 + +// Whether this human send should abort the running turn. `waitStartedAt` is the start of +// the oldest tool call still open in the Session — the observable foreground wait. +export function shouldInterruptOnHumanSend({ + sourceKind, + turnRunning, + ownChat, + waitStartedAt, + now, + thresholdMs = INTERRUPT_AFTER_MS, +}) { + return sourceKind === 'user' + && turnRunning === true + && ownChat === true + && typeof waitStartedAt === 'number' + && typeof now === 'number' + && now - waitStartedAt > thresholdMs +} + +export function install(rt) { + const { ctx } = rt + /** @type {Map>} tool calls still open per Session, by call id */ + const open = new Map() + + const callsOf = id => { + let calls = open.get(id) + if (calls === undefined) open.set(id, calls = new Map()) + return calls + } + const startedAtOf = id => { + const calls = open.get(id) + return calls === undefined || calls.size === 0 ? undefined : Math.min(...calls.values()) + } + const eventTime = event => (typeof event.time === 'number' ? event.time : Date.now()) + const now = () => (typeof rt.now === 'function' ? rt.now() : Date.now()) + const turnRunning = id => rt.live?.get(id)?.turn !== undefined + + const end = (id, key) => { + if (key === '') return + const calls = open.get(id) + if (calls === undefined) return + calls.delete(key) + if (calls.size === 0) open.delete(id) + } + + const interrupt = id => { + // Drop the evidence first: one abort per wait, however many lines arrive together. + open.delete(id) + void Promise.resolve(rt.resolveAgent?.(id)) + .then(agent => agent?.cancel?.({ kind: 'hook', reason: 'ds-bot/interrupt-long-wait' }, { keepInbox: true })) + .catch(() => {}) + } + + ctx.on('session/event', (session, event) => { + const id = session?.id + if (id === undefined || event?.type === undefined) return + switch (event.type) { + case 'tool/call': + callsOf(id).set(String(event.data?.callId ?? ''), eventTime(event)) + return + case 'tool/ptc-dispatch-start': + callsOf(id).set(String(event.data?.subCallId ?? ''), eventTime(event)) + return + case 'tool/result': + end(id, String(event.data?.callId ?? '')) + return + case 'tool/ptc-dispatch': + end(id, String(event.data?.subCallId ?? '')) + return + case 'turn/end': + open.delete(id) + return + case 'user/message': { + const source = event.data?.source + const waitStartedAt = startedAtOf(id) + if (!shouldInterruptOnHumanSend({ + sourceKind: source?.kind, + turnRunning: turnRunning(id), + ownChat: rt.chatOwners?.has(id) === true, + waitStartedAt, + now: now(), + })) return + interrupt(id) + return + } + default: + } + }) +} diff --git a/core/test/steering-interrupt.test.js b/core/test/steering-interrupt.test.js new file mode 100644 index 000000000..cfab265fc --- /dev/null +++ b/core/test/steering-interrupt.test.js @@ -0,0 +1,97 @@ +import assert from 'node:assert/strict' +import { test } from 'node:test' +import { + INTERRUPT_AFTER_MS, install, shouldInterruptOnHumanSend, +} from '../src/host/steering-interrupt.js' + +const T0 = 1_000_000 + +test('the line is 60s: 59s keeps the turn, past 60s interrupts it', () => { + const at = (elapsed) => shouldInterruptOnHumanSend({ + sourceKind: 'user', turnRunning: true, ownChat: true, waitStartedAt: T0, now: T0 + elapsed, + }) + assert.equal(INTERRUPT_AFTER_MS, 60_000) + assert.equal(at(0), false) + assert.equal(at(59_000), false) + assert.equal(at(60_000), false) + assert.equal(at(60_001), true) + assert.equal(at(10 * 60_000), true) +}) + +test('only a human line in an own chat with a turn on an open wait may interrupt', () => { + const base = { sourceKind: 'user', turnRunning: true, ownChat: true, waitStartedAt: T0, now: T0 + 61_000 } + assert.equal(shouldInterruptOnHumanSend(base), true) + assert.equal(shouldInterruptOnHumanSend({ ...base, sourceKind: 'bot' }), false) + assert.equal(shouldInterruptOnHumanSend({ ...base, sourceKind: 'schedule' }), false) + assert.equal(shouldInterruptOnHumanSend({ ...base, turnRunning: false }), false) + assert.equal(shouldInterruptOnHumanSend({ ...base, ownChat: false }), false) + assert.equal(shouldInterruptOnHumanSend({ ...base, waitStartedAt: undefined }), false) +}) + +function bench() { + const listeners = new Map() + const cancels = [] + let clock = T0 + const rt = { + live: new Map([['a', { turn: 1 }]]), + chatOwners: new Map([['a', 'a']]), + now: () => clock, + resolveAgent: async () => ({ cancel: (cause, options) => cancels.push({ id: 'a', cause, options }) }), + } + rt.ctx = { on: (name, fn) => (listeners.get(name) ?? listeners.set(name, []).get(name)).push(fn) } + install(rt) + const emit = (id, event) => { for (const fn of listeners.get('session/event') ?? []) fn({ id }, event) } + return { emit, cancels, advance: ms => { clock += ms }, rt } +} + +const toolCall = (callId, at) => ({ type: 'tool/call', time: at, data: { callId, name: 'bash' } }) +const toolResult = (callId, at) => ({ type: 'tool/result', time: at, data: { callId } }) +const human = (at) => ({ type: 'user/message', time: at, data: { source: { kind: 'user' }, content: [{ type: 'text', text: 'steer' }] } }) + +test('a human line 59s into a foreground wait does not abort the turn', async () => { + const b = bench() + b.emit('a', toolCall('c1', T0)) + b.advance(59_000) + b.emit('a', human(T0 + 59_000)) + await new Promise(resolve => setImmediate(resolve)) + assert.equal(b.cancels.length, 0) +}) + +test('a human line past 60s aborts exactly once, keeping the inbox, and the message is not lost', async () => { + const b = bench() + b.emit('a', toolCall('c1', T0)) + b.advance(61_000) + b.emit('a', human(T0 + 61_000)) + b.emit('a', human(T0 + 61_100)) + await new Promise(resolve => setImmediate(resolve)) + assert.equal(b.cancels.length, 1) + assert.deepEqual(b.cancels[0].options, { keepInbox: true }) + assert.equal(b.cancels[0].cause.kind, 'hook') +}) + +test('a finished wait, a group Session and an idle Session never interrupt', async () => { + const b = bench() + b.emit('a', toolCall('c1', T0)) + b.emit('a', toolResult('c1', T0 + 1_000)) + b.advance(120_000) + b.emit('a', human(T0 + 120_000)) + await new Promise(resolve => setImmediate(resolve)) + assert.equal(b.cancels.length, 0) + + // A group room the Bot speaks in is not the human's own chat. + b.rt.chatOwners.delete('a') + b.emit('a', toolCall('c2', T0 + 120_000)) + b.advance(61_000) + b.emit('a', human(T0 + 181_000)) + await new Promise(resolve => setImmediate(resolve)) + assert.equal(b.cancels.length, 0) + + // No turn in flight: the line starts a fresh turn by itself. + b.rt.chatOwners.set('a', 'a') + b.rt.live.delete('a') + b.emit('a', toolCall('c3', T0 + 181_000)) + b.advance(61_000) + b.emit('a', human(T0 + 242_000)) + await new Promise(resolve => setImmediate(resolve)) + assert.equal(b.cancels.length, 0) +}) diff --git a/docs/subsystems/core-host.md b/docs/subsystems/core-host.md index aaf399208..18afa62c5 100644 --- a/docs/subsystems/core-host.md +++ b/docs/subsystems/core-host.md @@ -405,7 +405,11 @@ Work a member hands a Bot with `message_bot` from its group Session runs in a hi - A friend's line in a Bot's own chat (#420): `deliver()` (`noteFriendLine`) stamps the receiving Bot with `friendLineAt` (published in the roster, so the save moves the revision) and keeps the senders seen per Bot in `state.remoteContacts[botId][handle]` (host only, capped at 200, dropped with the Bot). A sender's first line carries `source.remote.firstContact`. Group lines, and re-deliveries (`deferred`, `resumed`, `retried`, `resent`), stamp nothing. Covered by `core/test/flybot.test.js`. ## Busy-send acknowledgment (#1584) -When a human `user/message` lands in a Bot's own chat while that Session already has a live turn, the host records a short spoken acknowledgment on the live payload (`busyAcks`). The client draws it as one bot bubble (and also from a local pending queue/steer submission) so the chat is not silent while a blocking job wait holds the next step. Related but separate: [#1509](https://github.com/ThinkFlowLab/FlyBot/issues/1509) (Steer does not abort a tool wait). Host: `core/src/host/busy-ack.js`. Client: `core/src/client/busy-ack.js`. Tests: `core/test/busy-ack.test.js`, `core/test/busy-ack-client.test.js`. +When a human `user/message` lands in a Bot's own chat while that Session already has a live turn, the host records a short spoken acknowledgment on the live payload (`busyAcks`). The client draws it as one bot bubble (and also from a local pending queue/steer submission) so the chat is not silent while a blocking job wait holds the next step. Host: `core/src/host/busy-ack.js`. Client: `core/src/client/busy-ack.js`. Tests: `core/test/busy-ack.test.js`, `core/test/busy-ack-client.test.js`. + +## Steering interrupt: a wait over 60s yields to a human send (#1509) + +Steer alone admits a human line only at the next step boundary, so a foreground `bash`/job wait that blocks the step keeps the line waiting for the wall clock. `core/src/host/steering-interrupt.js` watches the Session events itself: a `tool/call` (or a program sub-call) is an open wait from its `event.time`, and `tool/result` / `turn/end` close it. When a `user/message` from source `user` lands in that Session while it is the Bot's own chat, a turn is live, and the oldest open wait is more than `INTERRUPT_AFTER_MS` (60 000 ms) old, the host cancels the turn the way Stop does — `agent.cancel({ kind: 'hook', reason: 'ds-bot/interrupt-long-wait' }, { keepInbox: true })` — exactly once per wait. `keepInbox` is what makes the send safe: the abort ends the turn, and the human line survives in the inbox for the turn that starts right after, so it is claimed at once and never lost. Waits of 60s and shorter are left alone, a finished wait is not a wait, and a group room or an idle Session is never touched. The threshold is fixed; no settings key. Tests: `core/test/steering-interrupt.test.js`. - **Pickup after a restart** (`resume.js`): every running Bot turn and every delivered-but-not-taken message is noted in `inflight.json`. At the next start a noted turn is picked up once only if a restart really cut it: the durable Session log's `turn/end` for that turn is `interrupted` (or an abort caused by `disposed`, or there is no end). A turn that finished while this plugin was not listening — `completed`, `blocked`, `max-tokens`, `forked`, or aborted by the user / hook / parent — is not repeated; a `turn/start` seen after the start means the noted turn ended too. Notes from a build before the record carried `turn` keep the old behaviour. A pickup is confirmed before its note is dropped: the notes read at start are seeded into the live `turns`/`queued` (so the file keeps everything still owed), a turn's note stays while its pickup runs and is dropped only once handed over (unless the resumed turn has written its own note first), and a queued message whose resend fails is kept for the next start, an exchange is noted once (a resend delivers the `exchangeId` the notes already hold, and noting it again left a copy behind after a turn took it up, so the start after that sent the message once more), and a turn taking up a message forgets every note of its exchange; a crash or a failed delivery during the pickup no longer loses the work — a pickup is at-least-once, not once: a crash between handing the turn over and the write that drops its note repeats that pickup once, because the `taken` set and the inbox that would tell the next start otherwise are process-local. That is the safe side of the trade, a repeated turn the user can see instead of work the user was told about going silently dead. Covered by `core/test/follow-through.test.js`. - 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`. diff --git a/patches/ds-bot/0002-flybot-api-host.patch b/patches/ds-bot/0002-flybot-api-host.patch index 74d5a4673..4291e28de 100644 --- a/patches/ds-bot/0002-flybot-api-host.patch +++ b/patches/ds-bot/0002-flybot-api-host.patch @@ -3,7 +3,7 @@ FlyBot patch 0002-flybot-api-host on Fei2411/DeepSeek-Bot@2b10c70 (ds-bot/). Gen diff --git a/index.js b/index.js --- a/index.js +++ b/index.js -@@ -4,9 +4,17 @@ +@@ -4,9 +4,18 @@ import { join } from 'node:path' import * as diagnostics from './src/host/diagnostics.js' import * as store from './src/host/store.js' @@ -18,6 +18,7 @@ diff --git a/index.js b/index.js import * as permissions from './src/host/permissions.js' import * as turns from './src/host/turns.js' +import * as busyAck from './src/host/busy-ack.js' ++import * as steeringInterrupt from './src/host/steering-interrupt.js' import * as delivery from './src/host/delivery.js' import * as roster from './src/host/roster.js' import * as folders from './src/host/folders.js' @@ -51,7 +52,7 @@ diff --git a/index.js b/index.js // install. Listeners run and tools are listed in registration order, so this order holds: // usage meters model requests after every listener that rewrites them. -const MODULES = [diagnostics, store, models, permissions, turns, delivery, roster, folders, agents, prompt, timeContext, tools, chatIndex, groups, approvals, resume, memory, parts, images, secrets, connectors, capabilities, schedules, browser, usage, update, dshUpdate, channel] -+const MODULES = [diagnostics, store, harness, memberAuth, authz, models, permissions, turns, busyAck, delivery, roster, folders, agents, prompt, timeContext, tools, decisions, chatIndex, groups, roomMembership, reach, tasks, taskDrive, approvals, resume, memory, parts, images, secrets, connectors, connectorAccess, capabilities, schedules, automationDefault, browser, usage, update, dshUpdate, hooks, remoteChat, workflowStore, channel] ++const MODULES = [diagnostics, store, harness, memberAuth, authz, models, permissions, turns, busyAck, steeringInterrupt, delivery, roster, folders, agents, prompt, timeContext, tools, decisions, chatIndex, groups, roomMembership, reach, tasks, taskDrive, approvals, resume, memory, parts, images, secrets, connectors, connectorAccess, capabilities, schedules, automationDefault, browser, usage, update, dshUpdate, hooks, remoteChat, workflowStore, channel] export function apply(ctx, config = {}) { - createRuntime(ctx, config) diff --git a/patches/ds-bot/files/src/host/steering-interrupt.js b/patches/ds-bot/files/src/host/steering-interrupt.js new file mode 100644 index 000000000..7632b489e --- /dev/null +++ b/patches/ds-bot/files/src/host/steering-interrupt.js @@ -0,0 +1,93 @@ +// Steering interrupt (#1509): a human message that arrives while a foreground tool/job +// wait has been open for more than 60s aborts that turn the way Stop does, keeping the +// inbox so the message is claimed at once instead of waiting for the blocking step. +export const INTERRUPT_AFTER_MS = 60_000 + +// Whether this human send should abort the running turn. `waitStartedAt` is the start of +// the oldest tool call still open in the Session — the observable foreground wait. +export function shouldInterruptOnHumanSend({ + sourceKind, + turnRunning, + ownChat, + waitStartedAt, + now, + thresholdMs = INTERRUPT_AFTER_MS, +}) { + return sourceKind === 'user' + && turnRunning === true + && ownChat === true + && typeof waitStartedAt === 'number' + && typeof now === 'number' + && now - waitStartedAt > thresholdMs +} + +export function install(rt) { + const { ctx } = rt + /** @type {Map>} tool calls still open per Session, by call id */ + const open = new Map() + + const callsOf = id => { + let calls = open.get(id) + if (calls === undefined) open.set(id, calls = new Map()) + return calls + } + const startedAtOf = id => { + const calls = open.get(id) + return calls === undefined || calls.size === 0 ? undefined : Math.min(...calls.values()) + } + const eventTime = event => (typeof event.time === 'number' ? event.time : Date.now()) + const now = () => (typeof rt.now === 'function' ? rt.now() : Date.now()) + const turnRunning = id => rt.live?.get(id)?.turn !== undefined + + const end = (id, key) => { + if (key === '') return + const calls = open.get(id) + if (calls === undefined) return + calls.delete(key) + if (calls.size === 0) open.delete(id) + } + + const interrupt = id => { + // Drop the evidence first: one abort per wait, however many lines arrive together. + open.delete(id) + void Promise.resolve(rt.resolveAgent?.(id)) + .then(agent => agent?.cancel?.({ kind: 'hook', reason: 'ds-bot/interrupt-long-wait' }, { keepInbox: true })) + .catch(() => {}) + } + + ctx.on('session/event', (session, event) => { + const id = session?.id + if (id === undefined || event?.type === undefined) return + switch (event.type) { + case 'tool/call': + callsOf(id).set(String(event.data?.callId ?? ''), eventTime(event)) + return + case 'tool/ptc-dispatch-start': + callsOf(id).set(String(event.data?.subCallId ?? ''), eventTime(event)) + return + case 'tool/result': + end(id, String(event.data?.callId ?? '')) + return + case 'tool/ptc-dispatch': + end(id, String(event.data?.subCallId ?? '')) + return + case 'turn/end': + open.delete(id) + return + case 'user/message': { + const source = event.data?.source + const waitStartedAt = startedAtOf(id) + if (!shouldInterruptOnHumanSend({ + sourceKind: source?.kind, + turnRunning: turnRunning(id), + ownChat: rt.chatOwners?.has(id) === true, + waitStartedAt, + now: now(), + })) return + interrupt(id) + return + } + default: + } + }) +} diff --git a/patches/ds-bot/files/test/steering-interrupt.test.js b/patches/ds-bot/files/test/steering-interrupt.test.js new file mode 100644 index 000000000..cfab265fc --- /dev/null +++ b/patches/ds-bot/files/test/steering-interrupt.test.js @@ -0,0 +1,97 @@ +import assert from 'node:assert/strict' +import { test } from 'node:test' +import { + INTERRUPT_AFTER_MS, install, shouldInterruptOnHumanSend, +} from '../src/host/steering-interrupt.js' + +const T0 = 1_000_000 + +test('the line is 60s: 59s keeps the turn, past 60s interrupts it', () => { + const at = (elapsed) => shouldInterruptOnHumanSend({ + sourceKind: 'user', turnRunning: true, ownChat: true, waitStartedAt: T0, now: T0 + elapsed, + }) + assert.equal(INTERRUPT_AFTER_MS, 60_000) + assert.equal(at(0), false) + assert.equal(at(59_000), false) + assert.equal(at(60_000), false) + assert.equal(at(60_001), true) + assert.equal(at(10 * 60_000), true) +}) + +test('only a human line in an own chat with a turn on an open wait may interrupt', () => { + const base = { sourceKind: 'user', turnRunning: true, ownChat: true, waitStartedAt: T0, now: T0 + 61_000 } + assert.equal(shouldInterruptOnHumanSend(base), true) + assert.equal(shouldInterruptOnHumanSend({ ...base, sourceKind: 'bot' }), false) + assert.equal(shouldInterruptOnHumanSend({ ...base, sourceKind: 'schedule' }), false) + assert.equal(shouldInterruptOnHumanSend({ ...base, turnRunning: false }), false) + assert.equal(shouldInterruptOnHumanSend({ ...base, ownChat: false }), false) + assert.equal(shouldInterruptOnHumanSend({ ...base, waitStartedAt: undefined }), false) +}) + +function bench() { + const listeners = new Map() + const cancels = [] + let clock = T0 + const rt = { + live: new Map([['a', { turn: 1 }]]), + chatOwners: new Map([['a', 'a']]), + now: () => clock, + resolveAgent: async () => ({ cancel: (cause, options) => cancels.push({ id: 'a', cause, options }) }), + } + rt.ctx = { on: (name, fn) => (listeners.get(name) ?? listeners.set(name, []).get(name)).push(fn) } + install(rt) + const emit = (id, event) => { for (const fn of listeners.get('session/event') ?? []) fn({ id }, event) } + return { emit, cancels, advance: ms => { clock += ms }, rt } +} + +const toolCall = (callId, at) => ({ type: 'tool/call', time: at, data: { callId, name: 'bash' } }) +const toolResult = (callId, at) => ({ type: 'tool/result', time: at, data: { callId } }) +const human = (at) => ({ type: 'user/message', time: at, data: { source: { kind: 'user' }, content: [{ type: 'text', text: 'steer' }] } }) + +test('a human line 59s into a foreground wait does not abort the turn', async () => { + const b = bench() + b.emit('a', toolCall('c1', T0)) + b.advance(59_000) + b.emit('a', human(T0 + 59_000)) + await new Promise(resolve => setImmediate(resolve)) + assert.equal(b.cancels.length, 0) +}) + +test('a human line past 60s aborts exactly once, keeping the inbox, and the message is not lost', async () => { + const b = bench() + b.emit('a', toolCall('c1', T0)) + b.advance(61_000) + b.emit('a', human(T0 + 61_000)) + b.emit('a', human(T0 + 61_100)) + await new Promise(resolve => setImmediate(resolve)) + assert.equal(b.cancels.length, 1) + assert.deepEqual(b.cancels[0].options, { keepInbox: true }) + assert.equal(b.cancels[0].cause.kind, 'hook') +}) + +test('a finished wait, a group Session and an idle Session never interrupt', async () => { + const b = bench() + b.emit('a', toolCall('c1', T0)) + b.emit('a', toolResult('c1', T0 + 1_000)) + b.advance(120_000) + b.emit('a', human(T0 + 120_000)) + await new Promise(resolve => setImmediate(resolve)) + assert.equal(b.cancels.length, 0) + + // A group room the Bot speaks in is not the human's own chat. + b.rt.chatOwners.delete('a') + b.emit('a', toolCall('c2', T0 + 120_000)) + b.advance(61_000) + b.emit('a', human(T0 + 181_000)) + await new Promise(resolve => setImmediate(resolve)) + assert.equal(b.cancels.length, 0) + + // No turn in flight: the line starts a fresh turn by itself. + b.rt.chatOwners.set('a', 'a') + b.rt.live.delete('a') + b.emit('a', toolCall('c3', T0 + 181_000)) + b.advance(61_000) + b.emit('a', human(T0 + 242_000)) + await new Promise(resolve => setImmediate(resolve)) + assert.equal(b.cancels.length, 0) +}) From a2eba6499154402a541881ab1a7a61f80b846f15 Mon Sep 17 00:00:00 2001 From: Danyang Chen Date: Thu, 29 Oct 2026 11:06:03 +0800 Subject: [PATCH 2/2] Keep the install order test in step with the new module index.js registers steering-interrupt after busy-ack, which adds one session/event listener to the fixed-order list the install test pins. Fixes #1509 --- core/test/install.test.js | 1 + patches/ds-bot/0001-core-fixes.patch | 3 ++- 2 files changed, 3 insertions(+), 1 deletion(-) diff --git a/core/test/install.test.js b/core/test/install.test.js index 62161c897..05b551576 100644 --- a/core/test/install.test.js +++ b/core/test/install.test.js @@ -39,6 +39,7 @@ test('apply registers listeners, tools, and effects in a fixed order', (t) => { 'on session/event', 'on session/event', 'on agent/status', + 'on session/event', 'effect ds-bot: agent tool restrict retry', 'on agent/created', 'effect ds-bot: identity section', diff --git a/patches/ds-bot/0001-core-fixes.patch b/patches/ds-bot/0001-core-fixes.patch index 7faaab1d7..0c1a6d750 100644 --- a/patches/ds-bot/0001-core-fixes.patch +++ b/patches/ds-bot/0001-core-fixes.patch @@ -8967,12 +8967,13 @@ diff --git a/test/images.test.js b/test/images.test.js diff --git a/test/install.test.js b/test/install.test.js --- a/test/install.test.js +++ b/test/install.test.js -@@ -37,6 +37,8 @@ test('apply registers listeners, tools, and effects in a fixed order', (t) => { +@@ -37,6 +37,9 @@ test('apply registers listeners, tools, and effects in a fixed order', (t) => { 'effect ds-bot: state lock', 'on llm/adapters-updated', 'on session/event', + 'on session/event', + 'on agent/status', ++ 'on session/event', 'effect ds-bot: agent tool restrict retry', 'on agent/created', 'effect ds-bot: identity section',