From ec049dbb2622aa182c07d3358bedbdf525c62778 Mon Sep 17 00:00:00 2001 From: Day Dreamer <87640074+HugoXK@users.noreply.github.com> Date: Thu, 29 Oct 2026 15:55:05 +0800 Subject: [PATCH] fix(host): let a room member's 60s wait go the way Stop does (#1650) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A person's line in a group room lands on the room's Session, not in the member Session's chat, so the own-chat 60s gate never saw it and a member stuck in a foreground wait held the round — and the person — until the tool returned. The host now runs the same 60s line through the room's own path: when a person's line lands on a room whose awaited member turn is live and that member's oldest open wait is past INTERRUPT_AFTER_MS, the member turn is let go with stopRoomTurn(sessionId, exchangeId) — the cancel Stop uses, no keepInbox, so nothing it had begun reaches the room. The line waits in the room's queued inbox and the next round answers it. A second member running in the same room is untouched, and a room id never takes reason 'ds-bot/interrupt-long-wait'. Red on f5b4112e with only the test file applied: 12 tests / 9 pass / 3 fail. With the fix: 12/12. --- .../1650-room-steering-let-go.fixed.md | 1 + core/src/host/steering-interrupt.js | 69 +++++- core/test/steering-interrupt.test.js | 228 ++++++++++++++++-- docs/subsystems/core-host.md | 2 + .../files/src/host/steering-interrupt.js | 69 +++++- .../files/test/steering-interrupt.test.js | 228 ++++++++++++++++-- 6 files changed, 547 insertions(+), 50 deletions(-) create mode 100644 changelog.d/1650-room-steering-let-go.fixed.md diff --git a/changelog.d/1650-room-steering-let-go.fixed.md b/changelog.d/1650-room-steering-let-go.fixed.md new file mode 100644 index 000000000..0777f2990 --- /dev/null +++ b/changelog.d/1650-room-steering-let-go.fixed.md @@ -0,0 +1 @@ +A person's line in a group room now lets a member's 60-second wait go instead of being held until it returns: when the member turn the round is waiting on has had a tool running for over a minute, the host ends that turn the way Stop does and the next round answers the line, while the person's message waits in the room's own queue. ([#1650](https://github.com/ThinkFlowLab/FlyBot/issues/1650)) diff --git a/core/src/host/steering-interrupt.js b/core/src/host/steering-interrupt.js index cb25cdb84..50dab9996 100644 --- a/core/src/host/steering-interrupt.js +++ b/core/src/host/steering-interrupt.js @@ -3,6 +3,10 @@ // inbox so the message is claimed at once instead of waiting for the blocking step. export const INTERRUPT_AFTER_MS = 60_000 +const longWait = ({ waitStartedAt, now, thresholdMs }) => typeof waitStartedAt === 'number' + && typeof now === 'number' + && now - waitStartedAt > thresholdMs + // 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({ @@ -16,9 +20,22 @@ export function shouldInterruptOnHumanSend({ return sourceKind === 'user' && turnRunning === true && ownChat === true - && typeof waitStartedAt === 'number' - && typeof now === 'number' - && now - waitStartedAt > thresholdMs + && longWait({ waitStartedAt, now, thresholdMs }) +} + +// A group room's own Session is not the Bot's chat (#1650): the line lands there, but the +// wait belongs to the member turn that room's round is waiting on, so that turn is the one +// that lets go — through Stop's room path, never the own-chat cancel. +export function shouldYieldRoomTurn({ + sourceKind, + roomTurnRunning, + waitStartedAt, + now, + thresholdMs = INTERRUPT_AFTER_MS, +}) { + return sourceKind === 'user' + && roomTurnRunning === true + && longWait({ waitStartedAt, now, thresholdMs }) } export function install(rt) { @@ -55,6 +72,26 @@ export function install(rt) { .catch(() => {}) } + // The member turn the room's round is still waiting on: the waiter is the round itself, + // and the room index plus the live exchange have to agree with it. A member that went on + // working has no waiter, so its turn no longer holds the room. + const awaitedRoomTurn = (roomId) => { + for (const [exchangeId, waiter] of rt.roomWaiters?.entries() ?? []) { + if (waiter?.roomId !== roomId) continue + const owner = rt.groupOwners?.get(waiter.sessionId) + const info = rt.live?.get(waiter.sessionId)?.info + if (owner?.roomId !== roomId || info?.role !== 'room' || info.exchangeId !== exchangeId) continue + return { sessionId: waiter.sessionId, exchangeId } + } + return undefined + } + + const yieldRoomTurn = (turn) => { + // Same as the own-chat path: the evidence goes first, so one let-go per wait. + open.delete(turn.sessionId) + void Promise.resolve(rt.stopRoomTurn?.(turn.sessionId, turn.exchangeId)).catch(() => {}) + } + ctx.on('session/event', (session, event) => { const id = session?.id if (id === undefined || event?.type === undefined) return @@ -78,15 +115,29 @@ export function install(rt) { return case 'user/message': { const source = event.data?.source - const waitStartedAt = startedAtOf(id) - if (!shouldInterruptOnHumanSend({ + if (rt.chatOwners?.has(id) === true) { + const waitStartedAt = startedAtOf(id) + if (!shouldInterruptOnHumanSend({ + sourceKind: source?.kind, + turnRunning: turnRunning(id), + ownChat: true, + waitStartedAt, + now: now(), + })) return + interrupt(id) + return + } + const room = rt.roomOf?.(id) + if (room === undefined) return + const turn = awaitedRoomTurn(room.id) + if (turn === undefined) return + if (!shouldYieldRoomTurn({ sourceKind: source?.kind, - turnRunning: turnRunning(id), - ownChat: rt.chatOwners?.has(id) === true, - waitStartedAt, + roomTurnRunning: turnRunning(turn.sessionId), + waitStartedAt: startedAtOf(turn.sessionId), now: now(), })) return - interrupt(id) + yieldRoomTurn(turn) return } default: diff --git a/core/test/steering-interrupt.test.js b/core/test/steering-interrupt.test.js index 841847304..8e555fbaa 100644 --- a/core/test/steering-interrupt.test.js +++ b/core/test/steering-interrupt.test.js @@ -1,5 +1,10 @@ import assert from 'node:assert/strict' +import { mkdtempSync, rmSync, writeFileSync } 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' import { INTERRUPT_AFTER_MS, install, shouldInterruptOnHumanSend, } from '../src/host/steering-interrupt.js' @@ -31,17 +36,23 @@ test('only a human line in an own chat with a turn on an open wait may interrupt function bench() { const listeners = new Map() const cancels = [] + const yields = [] let clock = T0 const rt = { live: new Map([['a', { turn: 1 }]]), chatOwners: new Map([['a', 'a']]), + groupOwners: new Map(), + roomWaiters: new Map(), + rooms: {}, now: () => clock, resolveAgent: async () => ({ cancel: (cause, options) => cancels.push({ id: 'a', cause, options }) }), + stopRoomTurn: async (sessionId, exchangeId) => { yields.push([sessionId, exchangeId]) }, } + rt.roomOf = id => rt.rooms[id] 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 } + return { emit, cancels, yields, advance: ms => { clock += ms }, rt } } const toolCall = (callId, at) => ({ type: 'tool/call', time: at, data: { callId, name: 'bash' } }) @@ -50,13 +61,25 @@ const toolCall = (callId, at) => ({ type: 'tool/call', time: at, data: { callId, const toolResult = (callId, at) => ({ type: 'tool/result', time: at, data: { message: { role: 'tool', toolCallId: callId, isError: false, content: [] } } }) const toolResultTopLevelId = (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' }] } }) +const flush = () => new Promise(resolve => setImmediate(resolve)) + +// A member's group turn the room's round is waiting on (#1650): the member Session sits in +// `groupOwners`, the live entry is the room turn, and the round's waiter names its exchange. +function roomTurn(b, { roomId = 'r1', sessionId = 'a', exchangeId = 'x1' } = {}) { + const rt = b.rt + rt.chatOwners.delete(sessionId) + rt.rooms[roomId] = { id: roomId, name: 'Ops', sessions: { writer: sessionId } } + rt.groupOwners.set(sessionId, { botId: 'writer', roomId }) + rt.live.set(sessionId, { turn: 1, info: { role: 'room', exchangeId, roomId } }) + rt.roomWaiters.set(exchangeId, { sessionId, roomId, botId: 'writer', exchangeId }) +} 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)) + await flush() assert.equal(b.cancels.length, 0) }) @@ -66,39 +89,212 @@ test('a human line past 60s aborts exactly once, keeping the inbox, and the mess b.advance(61_000) b.emit('a', human(T0 + 61_000)) b.emit('a', human(T0 + 61_100)) - await new Promise(resolve => setImmediate(resolve)) + await flush() 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 () => { +test('a finished wait 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)) + await flush() 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.emit('a', toolCall('c3', T0 + 120_000)) b.advance(61_000) b.emit('a', human(T0 + 242_000)) - await new Promise(resolve => setImmediate(resolve)) + await flush() assert.equal(b.cancels.length, 0) }) +test('a group room yields through the room path, never the own-chat one (#1650)', async () => { + const b = bench() + roomTurn(b) + b.emit('a', toolCall('c1', T0)) + b.advance(61_000) + // The person typed in the room's own Session; the wait is the member's. + b.emit('r1', human(T0 + 61_000)) + await flush() + assert.deepEqual(b.yields, [['a', 'x1']]) + assert.equal(b.cancels.length, 0, 'no room id takes the own-chat reason') + + // A second line in the same wait is still one let-go. + b.emit('r1', human(T0 + 61_100)) + await flush() + assert.equal(b.yields.length, 1) +}) + +test('a room waits for the same 60s, and a closed wait or a short one is left alone', async () => { + const b = bench() + roomTurn(b) + b.emit('a', toolCall('c1', T0)) + b.advance(60_000) + b.emit('r1', human(T0 + 60_000)) + await flush() + assert.deepEqual(b.yields, []) + + b.emit('a', toolResult('c1', T0 + 60_000)) + b.advance(61_000) + b.emit('r1', human(T0 + 121_000)) + await flush() + assert.deepEqual(b.yields, [], 'a finished wait is not a wait') + + // The member's turn is gone: no waiter holds the room any more. + b.emit('a', toolCall('c2', T0 + 121_000)) + b.advance(61_000) + b.rt.live.delete('a') + b.emit('r1', human(T0 + 182_000)) + await flush() + assert.deepEqual(b.yields, []) +}) + +test('a second member running in the same room is not let go', async () => { + const b = bench() + roomTurn(b) + // Another member is in a turn of its own (its round moved on), with a wait just as open. + b.rt.groupOwners.set('c', { botId: 'other', roomId: 'r1' }) + b.rt.live.set('c', { turn: 1, info: { role: 'room', exchangeId: 'x2', roomId: 'r1' } }) + b.emit('c', toolCall('c9', T0)) + b.emit('a', toolCall('c1', T0)) + b.advance(61_000) + b.emit('r1', human(T0 + 61_000)) + await flush() + assert.deepEqual(b.yields, [['a', 'x1']]) +}) + +// ---- a real host: the room Session, its round and the member turn in one runtime ---- + +const seedRoom = () => ({ + version: 1, + mainBotIds: ['a'], + bots: { + a: { id: 'a', name: 'Chief', role: 'Boss', sessionId: 'a', createdAt: 1 }, + b: { id: 'b', name: 'Writer', role: 'Write', sessionId: 'b', createdAt: 2 }, + }, + rooms: { r1: { id: 'r1', name: 'Ops', members: ['a', 'b'], mode: 'admin', admin: 'b', named: true, sessions: { b: 'gb' }, log: [], seq: 0, seen: {} } }, + agents: {}, exchanges: [], exchangeTurns: {}, questions: {}, prefs: {}, revision: 0, +}) + +async function roomHost(t) { + const home = mkdtempSync(join(tmpdir(), 'ds-bot-steering-')) + writeFileSync(join(home, 'state.json'), JSON.stringify(seedRoom())) + const ctx = fakeContext() + const agents = new Map() + const agentOf = (sessionId) => { + if (agents.has(sessionId)) return agents.get(sessionId) + const calls = [] + const cancelled = [] + agents.set(sessionId, { + session: { id: sessionId }, + calls, + cancelled, + inbox: { + nextStep: [], + nextTurn: [], + remove(id) { + for (const list of ['nextStep', 'nextTurn']) { + const at = this[list].findIndex(message => message.id === id) + if (at === -1) continue + this[list] = this[list].toSpliced(at, 1) + return true + } + return false + }, + }, + followup(message) { calls.push(message); this.inbox.nextTurn.push(message) }, + steer(message) { this.inbox.nextStep.push(message) }, + cancel(cause, options) { cancelled.push({ cause, options }) }, + }) + return agents.get(sessionId) + } + Object.assign(ctx, { + llm: { listProviders: () => [], listModels: async () => [], resolveModelInfo: async () => ({}) }, + workspaceRegistry: { + get: () => undefined, resolveByPath: async () => undefined, + create: async path => ({ id: 'w1', path }), archiveSession: async () => {}, + }, + sessionController: { + create: async () => ({ sessionId: 's1' }), + resolveAgent: async sessionId => ({ agent: agentOf(sessionId) }), + selectModel: async () => {}, + page: async () => ({ events: [] }), + }, + sessionTitle: { rename: async () => {} }, + sessions: { flush: async () => {} }, + }) + const rt = createRuntime(ctx, { home }) + t.after(async () => { await ctx.dispose(); rmSync(home, { recursive: true, force: true, maxRetries: 5 }) }) + // Session events reach every listener in DSH order; awaiting each keeps that order. + const emit = async (sessionId, event) => { + for (const listener of ctx.listeners.get('session/event') ?? []) await listener({ id: sessionId }, event) + } + const relay = (callId) => { + const stop = new AbortController() + t.after(() => stop.abort()) + return { + stop, + done: ctx.registeredTools.get('group_relay').execute({}, { agent: { id: 'r1', inbox: agentOf('r1').inbox }, callId, signal: stop.signal }), + } + } + // The member's turn: DSH took the queued prompt into the turn it starts. + const takeTurn = async (sessionId, turn, message) => { + agentOf(sessionId).inbox.nextTurn.length = 0 + await emit(sessionId, { type: 'turn/start', data: { turn } }) + await emit(sessionId, { type: 'user/message', data: { content: message.content, source: message.source } }) + } + return { rt, agentOf, emit, relay, takeTurn } +} + +const settle = async (check) => { + const until = Date.now() + 3000 + while (!check() && Date.now() < until) await new Promise(resolve => setImmediate(resolve)) + assert.ok(check(), 'condition not reached') +} +const textOf = message => message.content.map(block => block.text ?? '').join('\n') + +test('a real room round lets the member\'s 60s wait go and answers the line next round (#1650)', async (t) => { + const host = await roomHost(t) + const line = 'What is the deploy status?' + + const round = host.relay('c1') + await settle(() => host.agentOf('gb').calls.length === 1) + await host.takeTurn('gb', 1, host.agentOf('gb').calls[0]) + + // The member is stuck in a foreground wait that has been open for over a minute. + await host.emit('gb', { type: 'tool/call', time: Date.now() - 61_000, data: { callId: 'k1', name: 'bash', arguments: '{"command":"npm test"}' } }) + // The person types into the room's own Session. + await host.emit('r1', { type: 'user/message', data: { source: { kind: 'user', authorKind: 'human', authorName: 'You' }, content: [{ type: 'text', text: line }] } }) + await settle(() => host.agentOf('gb').cancelled.length === 1) + assert.deepEqual(host.agentOf('gb').cancelled[0], { cause: { kind: 'hook', reason: 'ds-bot/stopped' }, options: undefined }) + // The line waits in the room's own queue, not in the member's inbox. + assert.deepEqual(host.rt.roomInbox.get('r1')?.map(item => item.text), [line]) + + // The member's turn ends the way a Stop ends it; nothing it began reaches the room. + await host.emit('gb', { type: 'turn/end', data: { turn: 1, reason: { kind: 'aborted', reason: { kind: 'hook', reason: 'ds-bot/stopped' } } } }) + const first = JSON.parse(await round.done) + assert.equal(first.replies.some(reply => reply.text), false) + assert.equal(host.agentOf('r1').calls.length, 0) + + // The next round answers the line the person typed. + const next = host.relay('c2') + await settle(() => host.agentOf('gb').calls.length === 1 + 1) + await host.takeTurn('gb', 2, host.agentOf('gb').calls[1]) + assert.match(textOf(host.agentOf('gb').calls[1]), new RegExp(line)) + await host.emit('gb', { type: 'assistant/message', data: { message: { content: [{ type: 'text', text: 'Deployed at noon.' }] } } }) + await host.emit('gb', { type: 'turn/end', data: { turn: 2, reason: { kind: 'completed' } } }) + const second = JSON.parse(await next.done) + assert.ok(second.replies.some(reply => reply.text === 'Deployed at noon.'), 'the next round answered the line') + // The room has the person's line in its log, and the member was asked with it in the prompt. + assert.ok(host.rt.state.rooms.r1.log.some(entry => entry.text === line)) + assert.equal(host.rt.roomInbox.get('r1'), undefined) +}) + test("a landed tool result closes its wait, so a long turn is not read as a long wait (#1644)", async () => { const b = bench() b.emit('a', toolCall('c1', T0)) diff --git a/docs/subsystems/core-host.md b/docs/subsystems/core-host.md index 160bde3ce..160e0fed0 100644 --- a/docs/subsystems/core-host.md +++ b/docs/subsystems/core-host.md @@ -411,6 +411,8 @@ When a human `user/message` lands in a Bot's own chat while that Session already 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 — a result names its call on `data.message.toolCallId`, the shape the host writes (#1644), and `data.callId` only serves older payloads, so reading it alone left every wait open and turned the 60s line into the turn's age. 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`. +A group room takes the same 60s line, through the room's own path (#1650). A person typing in a room lands on the room's own Session (`room.id`), not in the member's, so the own-chat gate never saw it and the member's turn held the round for the whole wait. Now, when a `user/message` from source `user` lands on a room's Session, the member turn the round is waiting on is live, and that member's oldest open wait is more than `INTERRUPT_AFTER_MS` old, the host lets the turn go through `stopRoomTurn(sessionId, exchangeId)` — the very cancel Stop uses, `{ kind: 'hook', reason: 'ds-bot/stopped' }` with no `keepInbox`, so nothing the member had begun reaches the room, exactly as after a Stop. The line is safe because it waits in the room's queued inbox, and the next round answers it. A room id never takes `reason: 'ds-bot/interrupt-long-wait'`; that reason stays the own-chat one. Only the member the round is waiting on is let go — another member running in the same room is untouched — and waits of 60s or less, a finished wait, and an idle Session are all left alone. + - **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`. - **The check line is 45% of the usable window, not 80%** (`steward.js` `linesFor`): a Bot's conversation is condensed once its estimated size passes `DEFAULT_CHECK_RATIO` of the model's declared window less the room its reply needs, so the conversation is condensed earlier than upstream ds-bot does. The share is a floor on the prompt of a long turn, because every call re-sends the fixed block and the history is most of each prompt (#796); the trade is a checkpoint sooner and less detail kept (#1233). `checkRatio` in the profile still sets the share, and the hard line stays at 95% of the usable window, so a share at or above 0.95 can never lift the check line above it. Covered by `core/test/steward.test.js`. diff --git a/patches/ds-bot/files/src/host/steering-interrupt.js b/patches/ds-bot/files/src/host/steering-interrupt.js index cb25cdb84..50dab9996 100644 --- a/patches/ds-bot/files/src/host/steering-interrupt.js +++ b/patches/ds-bot/files/src/host/steering-interrupt.js @@ -3,6 +3,10 @@ // inbox so the message is claimed at once instead of waiting for the blocking step. export const INTERRUPT_AFTER_MS = 60_000 +const longWait = ({ waitStartedAt, now, thresholdMs }) => typeof waitStartedAt === 'number' + && typeof now === 'number' + && now - waitStartedAt > thresholdMs + // 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({ @@ -16,9 +20,22 @@ export function shouldInterruptOnHumanSend({ return sourceKind === 'user' && turnRunning === true && ownChat === true - && typeof waitStartedAt === 'number' - && typeof now === 'number' - && now - waitStartedAt > thresholdMs + && longWait({ waitStartedAt, now, thresholdMs }) +} + +// A group room's own Session is not the Bot's chat (#1650): the line lands there, but the +// wait belongs to the member turn that room's round is waiting on, so that turn is the one +// that lets go — through Stop's room path, never the own-chat cancel. +export function shouldYieldRoomTurn({ + sourceKind, + roomTurnRunning, + waitStartedAt, + now, + thresholdMs = INTERRUPT_AFTER_MS, +}) { + return sourceKind === 'user' + && roomTurnRunning === true + && longWait({ waitStartedAt, now, thresholdMs }) } export function install(rt) { @@ -55,6 +72,26 @@ export function install(rt) { .catch(() => {}) } + // The member turn the room's round is still waiting on: the waiter is the round itself, + // and the room index plus the live exchange have to agree with it. A member that went on + // working has no waiter, so its turn no longer holds the room. + const awaitedRoomTurn = (roomId) => { + for (const [exchangeId, waiter] of rt.roomWaiters?.entries() ?? []) { + if (waiter?.roomId !== roomId) continue + const owner = rt.groupOwners?.get(waiter.sessionId) + const info = rt.live?.get(waiter.sessionId)?.info + if (owner?.roomId !== roomId || info?.role !== 'room' || info.exchangeId !== exchangeId) continue + return { sessionId: waiter.sessionId, exchangeId } + } + return undefined + } + + const yieldRoomTurn = (turn) => { + // Same as the own-chat path: the evidence goes first, so one let-go per wait. + open.delete(turn.sessionId) + void Promise.resolve(rt.stopRoomTurn?.(turn.sessionId, turn.exchangeId)).catch(() => {}) + } + ctx.on('session/event', (session, event) => { const id = session?.id if (id === undefined || event?.type === undefined) return @@ -78,15 +115,29 @@ export function install(rt) { return case 'user/message': { const source = event.data?.source - const waitStartedAt = startedAtOf(id) - if (!shouldInterruptOnHumanSend({ + if (rt.chatOwners?.has(id) === true) { + const waitStartedAt = startedAtOf(id) + if (!shouldInterruptOnHumanSend({ + sourceKind: source?.kind, + turnRunning: turnRunning(id), + ownChat: true, + waitStartedAt, + now: now(), + })) return + interrupt(id) + return + } + const room = rt.roomOf?.(id) + if (room === undefined) return + const turn = awaitedRoomTurn(room.id) + if (turn === undefined) return + if (!shouldYieldRoomTurn({ sourceKind: source?.kind, - turnRunning: turnRunning(id), - ownChat: rt.chatOwners?.has(id) === true, - waitStartedAt, + roomTurnRunning: turnRunning(turn.sessionId), + waitStartedAt: startedAtOf(turn.sessionId), now: now(), })) return - interrupt(id) + yieldRoomTurn(turn) return } default: diff --git a/patches/ds-bot/files/test/steering-interrupt.test.js b/patches/ds-bot/files/test/steering-interrupt.test.js index 841847304..8e555fbaa 100644 --- a/patches/ds-bot/files/test/steering-interrupt.test.js +++ b/patches/ds-bot/files/test/steering-interrupt.test.js @@ -1,5 +1,10 @@ import assert from 'node:assert/strict' +import { mkdtempSync, rmSync, writeFileSync } 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' import { INTERRUPT_AFTER_MS, install, shouldInterruptOnHumanSend, } from '../src/host/steering-interrupt.js' @@ -31,17 +36,23 @@ test('only a human line in an own chat with a turn on an open wait may interrupt function bench() { const listeners = new Map() const cancels = [] + const yields = [] let clock = T0 const rt = { live: new Map([['a', { turn: 1 }]]), chatOwners: new Map([['a', 'a']]), + groupOwners: new Map(), + roomWaiters: new Map(), + rooms: {}, now: () => clock, resolveAgent: async () => ({ cancel: (cause, options) => cancels.push({ id: 'a', cause, options }) }), + stopRoomTurn: async (sessionId, exchangeId) => { yields.push([sessionId, exchangeId]) }, } + rt.roomOf = id => rt.rooms[id] 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 } + return { emit, cancels, yields, advance: ms => { clock += ms }, rt } } const toolCall = (callId, at) => ({ type: 'tool/call', time: at, data: { callId, name: 'bash' } }) @@ -50,13 +61,25 @@ const toolCall = (callId, at) => ({ type: 'tool/call', time: at, data: { callId, const toolResult = (callId, at) => ({ type: 'tool/result', time: at, data: { message: { role: 'tool', toolCallId: callId, isError: false, content: [] } } }) const toolResultTopLevelId = (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' }] } }) +const flush = () => new Promise(resolve => setImmediate(resolve)) + +// A member's group turn the room's round is waiting on (#1650): the member Session sits in +// `groupOwners`, the live entry is the room turn, and the round's waiter names its exchange. +function roomTurn(b, { roomId = 'r1', sessionId = 'a', exchangeId = 'x1' } = {}) { + const rt = b.rt + rt.chatOwners.delete(sessionId) + rt.rooms[roomId] = { id: roomId, name: 'Ops', sessions: { writer: sessionId } } + rt.groupOwners.set(sessionId, { botId: 'writer', roomId }) + rt.live.set(sessionId, { turn: 1, info: { role: 'room', exchangeId, roomId } }) + rt.roomWaiters.set(exchangeId, { sessionId, roomId, botId: 'writer', exchangeId }) +} 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)) + await flush() assert.equal(b.cancels.length, 0) }) @@ -66,39 +89,212 @@ test('a human line past 60s aborts exactly once, keeping the inbox, and the mess b.advance(61_000) b.emit('a', human(T0 + 61_000)) b.emit('a', human(T0 + 61_100)) - await new Promise(resolve => setImmediate(resolve)) + await flush() 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 () => { +test('a finished wait 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)) + await flush() 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.emit('a', toolCall('c3', T0 + 120_000)) b.advance(61_000) b.emit('a', human(T0 + 242_000)) - await new Promise(resolve => setImmediate(resolve)) + await flush() assert.equal(b.cancels.length, 0) }) +test('a group room yields through the room path, never the own-chat one (#1650)', async () => { + const b = bench() + roomTurn(b) + b.emit('a', toolCall('c1', T0)) + b.advance(61_000) + // The person typed in the room's own Session; the wait is the member's. + b.emit('r1', human(T0 + 61_000)) + await flush() + assert.deepEqual(b.yields, [['a', 'x1']]) + assert.equal(b.cancels.length, 0, 'no room id takes the own-chat reason') + + // A second line in the same wait is still one let-go. + b.emit('r1', human(T0 + 61_100)) + await flush() + assert.equal(b.yields.length, 1) +}) + +test('a room waits for the same 60s, and a closed wait or a short one is left alone', async () => { + const b = bench() + roomTurn(b) + b.emit('a', toolCall('c1', T0)) + b.advance(60_000) + b.emit('r1', human(T0 + 60_000)) + await flush() + assert.deepEqual(b.yields, []) + + b.emit('a', toolResult('c1', T0 + 60_000)) + b.advance(61_000) + b.emit('r1', human(T0 + 121_000)) + await flush() + assert.deepEqual(b.yields, [], 'a finished wait is not a wait') + + // The member's turn is gone: no waiter holds the room any more. + b.emit('a', toolCall('c2', T0 + 121_000)) + b.advance(61_000) + b.rt.live.delete('a') + b.emit('r1', human(T0 + 182_000)) + await flush() + assert.deepEqual(b.yields, []) +}) + +test('a second member running in the same room is not let go', async () => { + const b = bench() + roomTurn(b) + // Another member is in a turn of its own (its round moved on), with a wait just as open. + b.rt.groupOwners.set('c', { botId: 'other', roomId: 'r1' }) + b.rt.live.set('c', { turn: 1, info: { role: 'room', exchangeId: 'x2', roomId: 'r1' } }) + b.emit('c', toolCall('c9', T0)) + b.emit('a', toolCall('c1', T0)) + b.advance(61_000) + b.emit('r1', human(T0 + 61_000)) + await flush() + assert.deepEqual(b.yields, [['a', 'x1']]) +}) + +// ---- a real host: the room Session, its round and the member turn in one runtime ---- + +const seedRoom = () => ({ + version: 1, + mainBotIds: ['a'], + bots: { + a: { id: 'a', name: 'Chief', role: 'Boss', sessionId: 'a', createdAt: 1 }, + b: { id: 'b', name: 'Writer', role: 'Write', sessionId: 'b', createdAt: 2 }, + }, + rooms: { r1: { id: 'r1', name: 'Ops', members: ['a', 'b'], mode: 'admin', admin: 'b', named: true, sessions: { b: 'gb' }, log: [], seq: 0, seen: {} } }, + agents: {}, exchanges: [], exchangeTurns: {}, questions: {}, prefs: {}, revision: 0, +}) + +async function roomHost(t) { + const home = mkdtempSync(join(tmpdir(), 'ds-bot-steering-')) + writeFileSync(join(home, 'state.json'), JSON.stringify(seedRoom())) + const ctx = fakeContext() + const agents = new Map() + const agentOf = (sessionId) => { + if (agents.has(sessionId)) return agents.get(sessionId) + const calls = [] + const cancelled = [] + agents.set(sessionId, { + session: { id: sessionId }, + calls, + cancelled, + inbox: { + nextStep: [], + nextTurn: [], + remove(id) { + for (const list of ['nextStep', 'nextTurn']) { + const at = this[list].findIndex(message => message.id === id) + if (at === -1) continue + this[list] = this[list].toSpliced(at, 1) + return true + } + return false + }, + }, + followup(message) { calls.push(message); this.inbox.nextTurn.push(message) }, + steer(message) { this.inbox.nextStep.push(message) }, + cancel(cause, options) { cancelled.push({ cause, options }) }, + }) + return agents.get(sessionId) + } + Object.assign(ctx, { + llm: { listProviders: () => [], listModels: async () => [], resolveModelInfo: async () => ({}) }, + workspaceRegistry: { + get: () => undefined, resolveByPath: async () => undefined, + create: async path => ({ id: 'w1', path }), archiveSession: async () => {}, + }, + sessionController: { + create: async () => ({ sessionId: 's1' }), + resolveAgent: async sessionId => ({ agent: agentOf(sessionId) }), + selectModel: async () => {}, + page: async () => ({ events: [] }), + }, + sessionTitle: { rename: async () => {} }, + sessions: { flush: async () => {} }, + }) + const rt = createRuntime(ctx, { home }) + t.after(async () => { await ctx.dispose(); rmSync(home, { recursive: true, force: true, maxRetries: 5 }) }) + // Session events reach every listener in DSH order; awaiting each keeps that order. + const emit = async (sessionId, event) => { + for (const listener of ctx.listeners.get('session/event') ?? []) await listener({ id: sessionId }, event) + } + const relay = (callId) => { + const stop = new AbortController() + t.after(() => stop.abort()) + return { + stop, + done: ctx.registeredTools.get('group_relay').execute({}, { agent: { id: 'r1', inbox: agentOf('r1').inbox }, callId, signal: stop.signal }), + } + } + // The member's turn: DSH took the queued prompt into the turn it starts. + const takeTurn = async (sessionId, turn, message) => { + agentOf(sessionId).inbox.nextTurn.length = 0 + await emit(sessionId, { type: 'turn/start', data: { turn } }) + await emit(sessionId, { type: 'user/message', data: { content: message.content, source: message.source } }) + } + return { rt, agentOf, emit, relay, takeTurn } +} + +const settle = async (check) => { + const until = Date.now() + 3000 + while (!check() && Date.now() < until) await new Promise(resolve => setImmediate(resolve)) + assert.ok(check(), 'condition not reached') +} +const textOf = message => message.content.map(block => block.text ?? '').join('\n') + +test('a real room round lets the member\'s 60s wait go and answers the line next round (#1650)', async (t) => { + const host = await roomHost(t) + const line = 'What is the deploy status?' + + const round = host.relay('c1') + await settle(() => host.agentOf('gb').calls.length === 1) + await host.takeTurn('gb', 1, host.agentOf('gb').calls[0]) + + // The member is stuck in a foreground wait that has been open for over a minute. + await host.emit('gb', { type: 'tool/call', time: Date.now() - 61_000, data: { callId: 'k1', name: 'bash', arguments: '{"command":"npm test"}' } }) + // The person types into the room's own Session. + await host.emit('r1', { type: 'user/message', data: { source: { kind: 'user', authorKind: 'human', authorName: 'You' }, content: [{ type: 'text', text: line }] } }) + await settle(() => host.agentOf('gb').cancelled.length === 1) + assert.deepEqual(host.agentOf('gb').cancelled[0], { cause: { kind: 'hook', reason: 'ds-bot/stopped' }, options: undefined }) + // The line waits in the room's own queue, not in the member's inbox. + assert.deepEqual(host.rt.roomInbox.get('r1')?.map(item => item.text), [line]) + + // The member's turn ends the way a Stop ends it; nothing it began reaches the room. + await host.emit('gb', { type: 'turn/end', data: { turn: 1, reason: { kind: 'aborted', reason: { kind: 'hook', reason: 'ds-bot/stopped' } } } }) + const first = JSON.parse(await round.done) + assert.equal(first.replies.some(reply => reply.text), false) + assert.equal(host.agentOf('r1').calls.length, 0) + + // The next round answers the line the person typed. + const next = host.relay('c2') + await settle(() => host.agentOf('gb').calls.length === 1 + 1) + await host.takeTurn('gb', 2, host.agentOf('gb').calls[1]) + assert.match(textOf(host.agentOf('gb').calls[1]), new RegExp(line)) + await host.emit('gb', { type: 'assistant/message', data: { message: { content: [{ type: 'text', text: 'Deployed at noon.' }] } } }) + await host.emit('gb', { type: 'turn/end', data: { turn: 2, reason: { kind: 'completed' } } }) + const second = JSON.parse(await next.done) + assert.ok(second.replies.some(reply => reply.text === 'Deployed at noon.'), 'the next round answered the line') + // The room has the person's line in its log, and the member was asked with it in the prompt. + assert.ok(host.rt.state.rooms.r1.log.some(entry => entry.text === line)) + assert.equal(host.rt.roomInbox.get('r1'), undefined) +}) + test("a landed tool result closes its wait, so a long turn is not read as a long wait (#1644)", async () => { const b = bench() b.emit('a', toolCall('c1', T0))