fix(host): close a tool wait from the result the host writes

changelog.d/1644-tool-result-closes-wait.fixed.md

@@ -0,0 +1 @@
+A tool result now closes its foreground wait, so the 60-second steering line only aborts a turn that is genuinely waiting: a human line that arrives after a turn has merely been running for a while, with every tool already returned, no longer interrupts it. The Bot's "running tool" chip also clears as soon as the tool returns. ([#1644](https://github.com/ThinkFlowLab/FlyBot/issues/1644))

core/src/host/steering-interrupt.js

@@ -66,7 +66,9 @@ export function install(rt) {
         callsOf(id).set(String(event.data?.subCallId ?? ''), eventTime(event))
         return
       case 'tool/result':
-        end(id, String(event.data?.callId ?? ''))
+        // The host reports a result as `{ message: ToolResultMessage }`, so the call id is
+        // on the message; only older payloads carry it at the top of `data` (#1644).
+        end(id, String(event.data?.callId ?? event.data?.message?.toolCallId ?? ''))
         return
       case 'tool/ptc-dispatch':
         end(id, String(event.data?.subCallId ?? ''))

core/src/host/turns.js

@@ -402,7 +402,9 @@ export function install(rt) {
         endCall(id, entry, event.data?.subCallId)
         break
       case 'tool/result':
-        endCall(id, entry, event.data?.callId)
+        // A result carries its call id on `data.message`, the shape the host writes; the
+        // top-level read only serves older payloads (#1644).
+        endCall(id, entry, event.data?.callId ?? event.data?.message?.toolCallId)
         break
       case 'turn/start':
         setActivity(id, 'thinking')

core/test/steering-interrupt.test.js

@@ -45,7 +45,10 @@ function bench() {
 }
 
 const toolCall = (callId, at) => ({ type: 'tool/call', time: at, data: { callId, name: 'bash' } })
-const toolResult = (callId, at) => ({ type: 'tool/result', time: at, data: { callId } })
+// The host writes a result as `{ message: ToolResultMessage }`, so the call id arrives on
+// the message and never at the top level of `data` (DSH's `tool/result` event map).
+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' }] } })
 
 test('a human line 59s into a foreground wait does not abort the turn', async () => {
@@ -95,3 +98,39 @@ test('a finished wait, a group Session and an idle Session never interrupt', asy
   await new Promise(resolve => setImmediate(resolve))
   assert.equal(b.cancels.length, 0)
 })
+
+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))
+  b.emit('a', toolResult('c1', T0 + 5_000))
+  b.advance(61_000)
+  b.emit('a', human(T0 + 61_000))
+  await new Promise(resolve => setImmediate(resolve))
+  assert.equal(b.cancels.length, 0)
+})
+
+test('a closed wait does not age a later call in the same turn (#1644)', async () => {
+  const b = bench()
+  b.emit('a', toolCall('c1', T0))
+  b.emit('a', toolResult('c1', T0 + 5_000))
+  b.emit('a', toolCall('c2', T0 + 100_000))
+  b.advance(130_000)
+  b.emit('a', human(T0 + 130_000))
+  await new Promise(resolve => setImmediate(resolve))
+  assert.equal(b.cancels.length, 0)
+  // The same call is still a wait once it has run past the line.
+  b.advance(31_000)
+  b.emit('a', human(T0 + 161_000))
+  await new Promise(resolve => setImmediate(resolve))
+  assert.equal(b.cancels.length, 1)
+})
+
+test('a result that names its call id at the top of data still closes the wait', async () => {
+  const b = bench()
+  b.emit('a', toolCall('c1', T0))
+  b.emit('a', toolResultTopLevelId('c1', T0 + 5_000))
+  b.advance(61_000)
+  b.emit('a', human(T0 + 61_000))
+  await new Promise(resolve => setImmediate(resolve))
+  assert.equal(b.cancels.length, 0)
+})

core/test/turns-tool-result.test.js

@@ -0,0 +1,50 @@
+import assert from 'node:assert/strict'
+import { test } from 'node:test'
+import { install } from '../src/host/turns.js'
+
+// The running Bot's tool chip is cleared from the same `tool/result` read that closes the
+// steer line's wait (#1644), so the payload shapes are pinned here too.
+function turnsHost() {
+  let listener
+  const rt = {
+    ctx: { on: (name, fn) => { if (name === 'session/event') listener = fn } },
+    state: { exchangeTurns: {}, questions: {} },
+    isBotSession: () => true,
+    roomOf: () => undefined,
+    groupOwners: new Map(),
+    chatOwners: new Map([['b', 'b']]),
+    botOf: id => ({ id, name: id }),
+    selfOf: id => id,
+    save: async () => {},
+    deliver: async () => {},
+    noteFailure() {},
+    warn() {},
+    switchJobs: () => [],
+    flybotReplyRemote: async () => ({}),
+  }
+  install(rt)
+  let seq = 0
+  const emit = (type, data) => listener({ id: 'b' }, { type, seq: ++seq, data })
+  return { rt, emit }
+}
+
+const toolResult = (callId) => ({ message: { role: 'tool', toolCallId: callId, isError: false, content: [] } })
+
+test("a tool result in the host's shape clears the running tool (#1644)", () => {
+  const { rt, emit } = turnsHost()
+  emit('turn/start', { turn: 1 })
+  emit('tool/call', { callId: 'c1', name: 'bash' })
+  assert.equal(rt.workView().b.tool, 'bash')
+  emit('tool/result', toolResult('c1'))
+  assert.equal(rt.liveOf('b').calls?.size ?? 0, 0)
+  assert.equal(rt.workView().b.tool, undefined)
+})
+
+test('a result that names its call id at the top of data still clears the running tool', () => {
+  const { rt, emit } = turnsHost()
+  emit('turn/start', { turn: 1 })
+  emit('tool/call', { callId: 'c1', name: 'bash' })
+  emit('tool/result', { callId: 'c1' })
+  assert.equal(rt.liveOf('b').calls?.size ?? 0, 0)
+  assert.equal(rt.workView().b.tool, undefined)
+})

docs/subsystems/core-host.md

@@ -409,7 +409,7 @@ When a human `user/message` lands in a Bot's own chat while that Session already
 
 ## 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`.
+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`.
 
 - **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 <sender> -> <session>` 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`.

patches/ds-bot/0001-core-fixes.patch

@@ -5931,6 +5931,17 @@ diff --git a/src/host/turns.js b/src/host/turns.js
          // A program is as quick as the tools it runs, so its sub-calls decide.
          if (rt.groupOwners.has(id) && name !== PROGRAM_TOOL) rt.workRoomTurn?.(id, name)
          break
+@@ -280,7 +280,9 @@ export function install(rt) {
+         endCall(id, entry, event.data?.subCallId)
+         break
+       case 'tool/result':
+-        endCall(id, entry, event.data?.callId)
++        // A result carries its call id on `data.message`, the shape the host writes; the
++        // top-level read only serves older payloads (#1644).
++        endCall(id, entry, event.data?.callId ?? event.data?.message?.toolCallId)
+         break
+       case 'turn/start':
+         setActivity(id, 'thinking')
 @@ -288,11 +288,17 @@ export function install(rt) {
          entry.since = event.time ?? Date.now()
          entry.texts = []

patches/ds-bot/files/src/host/steering-interrupt.js

@@ -66,7 +66,9 @@ export function install(rt) {
         callsOf(id).set(String(event.data?.subCallId ?? ''), eventTime(event))
         return
       case 'tool/result':
-        end(id, String(event.data?.callId ?? ''))
+        // The host reports a result as `{ message: ToolResultMessage }`, so the call id is
+        // on the message; only older payloads carry it at the top of `data` (#1644).
+        end(id, String(event.data?.callId ?? event.data?.message?.toolCallId ?? ''))
         return
       case 'tool/ptc-dispatch':
         end(id, String(event.data?.subCallId ?? ''))

patches/ds-bot/files/test/steering-interrupt.test.js

@@ -45,7 +45,10 @@ function bench() {
 }
 
 const toolCall = (callId, at) => ({ type: 'tool/call', time: at, data: { callId, name: 'bash' } })
-const toolResult = (callId, at) => ({ type: 'tool/result', time: at, data: { callId } })
+// The host writes a result as `{ message: ToolResultMessage }`, so the call id arrives on
+// the message and never at the top level of `data` (DSH's `tool/result` event map).
+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' }] } })
 
 test('a human line 59s into a foreground wait does not abort the turn', async () => {
@@ -95,3 +98,39 @@ test('a finished wait, a group Session and an idle Session never interrupt', asy
   await new Promise(resolve => setImmediate(resolve))
   assert.equal(b.cancels.length, 0)
 })
+
+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))
+  b.emit('a', toolResult('c1', T0 + 5_000))
+  b.advance(61_000)
+  b.emit('a', human(T0 + 61_000))
+  await new Promise(resolve => setImmediate(resolve))
+  assert.equal(b.cancels.length, 0)
+})
+
+test('a closed wait does not age a later call in the same turn (#1644)', async () => {
+  const b = bench()
+  b.emit('a', toolCall('c1', T0))
+  b.emit('a', toolResult('c1', T0 + 5_000))
+  b.emit('a', toolCall('c2', T0 + 100_000))
+  b.advance(130_000)
+  b.emit('a', human(T0 + 130_000))
+  await new Promise(resolve => setImmediate(resolve))
+  assert.equal(b.cancels.length, 0)
+  // The same call is still a wait once it has run past the line.
+  b.advance(31_000)
+  b.emit('a', human(T0 + 161_000))
+  await new Promise(resolve => setImmediate(resolve))
+  assert.equal(b.cancels.length, 1)
+})
+
+test('a result that names its call id at the top of data still closes the wait', async () => {
+  const b = bench()
+  b.emit('a', toolCall('c1', T0))
+  b.emit('a', toolResultTopLevelId('c1', T0 + 5_000))
+  b.advance(61_000)
+  b.emit('a', human(T0 + 61_000))
+  await new Promise(resolve => setImmediate(resolve))
+  assert.equal(b.cancels.length, 0)
+})

patches/ds-bot/files/test/turns-tool-result.test.js

@@ -0,0 +1,50 @@
+import assert from 'node:assert/strict'
+import { test } from 'node:test'
+import { install } from '../src/host/turns.js'
+
+// The running Bot's tool chip is cleared from the same `tool/result` read that closes the
+// steer line's wait (#1644), so the payload shapes are pinned here too.
+function turnsHost() {
+  let listener
+  const rt = {
+    ctx: { on: (name, fn) => { if (name === 'session/event') listener = fn } },
+    state: { exchangeTurns: {}, questions: {} },
+    isBotSession: () => true,
+    roomOf: () => undefined,
+    groupOwners: new Map(),
+    chatOwners: new Map([['b', 'b']]),
+    botOf: id => ({ id, name: id }),
+    selfOf: id => id,
+    save: async () => {},
+    deliver: async () => {},
+    noteFailure() {},
+    warn() {},
+    switchJobs: () => [],
+    flybotReplyRemote: async () => ({}),
+  }
+  install(rt)
+  let seq = 0
+  const emit = (type, data) => listener({ id: 'b' }, { type, seq: ++seq, data })
+  return { rt, emit }
+}
+
+const toolResult = (callId) => ({ message: { role: 'tool', toolCallId: callId, isError: false, content: [] } })
+
+test("a tool result in the host's shape clears the running tool (#1644)", () => {
+  const { rt, emit } = turnsHost()
+  emit('turn/start', { turn: 1 })
+  emit('tool/call', { callId: 'c1', name: 'bash' })
+  assert.equal(rt.workView().b.tool, 'bash')
+  emit('tool/result', toolResult('c1'))
+  assert.equal(rt.liveOf('b').calls?.size ?? 0, 0)
+  assert.equal(rt.workView().b.tool, undefined)
+})
+
+test('a result that names its call id at the top of data still clears the running tool', () => {
+  const { rt, emit } = turnsHost()
+  emit('turn/start', { turn: 1 })
+  emit('tool/call', { callId: 'c1', name: 'bash' })
+  emit('tool/result', { callId: 'c1' })
+  assert.equal(rt.liveOf('b').calls?.size ?? 0, 0)
+  assert.equal(rt.workView().b.tool, undefined)
+})