From d172d6bd1832202c84883f6a20f00cc9e25d87b1 Mon Sep 17 00:00:00 2001 From: Danyang Chen Date: Sun, 8 Nov 2026 21:07:48 +0800 Subject: [PATCH 1/2] Report an over-long room answer instead of dropping it (#2689) An external room participant settles a line only when its newline arrives, so an answer that outgrew the 4 KB in-flight buffer had its bytes erased; the remaining fragment was ignored, the turn hit the reply timeout, and the room was told the process timed out. The round then counted a member that did answer as silence, and the child was killed for a timeout that never happened. Remember that the in-flight line overflowed instead of clearing it, drop only its bytes up to the newline that ends it, and settle that turn as { failure: { reason: 'error', detail: 'answer exceeded 4096 bytes' } } with one room line saying why. The process is kept for the next turn: a timeout is a child that will not answer, an overflow is an answer too large to carry. Fixes #2689 --- .../2689-external-long-answer.fixed.md | 1 + core/src/host/room-participant-external.js | 35 ++++- core/test/room-external.test.js | 123 ++++++++++++++++++ docs/subsystems/core-host.md | 2 +- .../src/host/room-participant-external.js | 35 ++++- .../ds-bot/files/test/room-external.test.js | 123 ++++++++++++++++++ 6 files changed, 312 insertions(+), 7 deletions(-) create mode 100644 changelog.d/2689-external-long-answer.fixed.md diff --git a/changelog.d/2689-external-long-answer.fixed.md b/changelog.d/2689-external-long-answer.fixed.md new file mode 100644 index 000000000..9638cc1b7 --- /dev/null +++ b/changelog.d/2689-external-long-answer.fixed.md @@ -0,0 +1 @@ +An external room participant whose answer line grows past the 4 KB stdout buffer is now reported to the room as `answer exceeded 4096 bytes` and keeps its process for later turns, instead of the answer being dropped as if the process had never spoken ([#2689](https://github.com/ThinkFlowLab/FlyBot/issues/2689)). diff --git a/core/src/host/room-participant-external.js b/core/src/host/room-participant-external.js index e21bbddef..c8c46b9c0 100644 --- a/core/src/host/room-participant-external.js +++ b/core/src/host/room-participant-external.js @@ -6,7 +6,8 @@ // The protocol on the child's stdio is one JSON object per line. // in: { type: 'turn', text, note, from?, exchangeId } { type: 'stop' } // out: { text } or { error } -// Anything else on stdout is ignored, so a process may log. +// Anything else on stdout is ignored, so a process may log. A line that outgrows the in-flight +// buffer is that turn's reported failure, never silence and never a timeout (#2689). import { spawn } from 'node:child_process' import { ROOM_REPLY_TIMEOUT_MS } from './constants.js' @@ -75,6 +76,8 @@ export function createExternalRoomParticipant(rt) { // A state is a sentence the room reads, not a state name: `timeout` reads as `timed out` (#1164). function stateLine(member, state, detail) { if (state === 'crashed') return `${member.name}'s external process crashed${detail ? ` (${detail})` : ''}. Its turns in this room have no answer until it starts again.` + // An over-long line is not a dead or a slow process: the seat keeps it for the next turn. + if (state === 'overflow') return oneLine(`${member.name} answered more than ${MAX_OUT} bytes in one line, so that answer was dropped; the process is still running for the next turn.`) return `${member.name}'s external process ${state === 'timeout' ? 'timed out' : state}${detail ? ` (${detail})` : ''}.` } @@ -108,6 +111,18 @@ export function createExternalRoomParticipant(rt) { } } + // An over-long line is the turn's answer with too many bytes to carry, which is why it ends the + // turn as a failure instead of ending the child: a timeout means the child will not answer, an + // overflow means it answered too large, and only the first one is a reason to kill it (#2689). + function settleOverflow(entry) { + if (entry.turn === undefined) return false + const turn = entry.turn + entry.turn = undefined + entry.state = 'idle' + turn.finish({ failure: { reason: 'error', detail: `answer exceeded ${MAX_OUT} bytes` } }) + return true + } + function start(room, member) { const key = keyOf(room, member) const known = processes.get(key) @@ -122,7 +137,7 @@ export function createExternalRoomParticipant(rt) { else processes.delete(key) } const record = member.external - const entry = { state: 'starting', detail: undefined, turn: undefined, child: undefined, buffer: '' } + const entry = { state: 'starting', detail: undefined, turn: undefined, child: undefined, buffer: '', overflowed: false } processes.set(key, entry) let child try { @@ -140,6 +155,16 @@ export function createExternalRoomParticipant(rt) { entry.state = 'idle' child.stdout.setEncoding('utf8') child.stdout.on('data', (chunk) => { + // The line already overflowed: drop its bytes up to the newline that ends it, then settle + // the turn it belongs to. Only the tail of the line is kept — long enough to find that + // newline, never parsed as an answer (#2689). + if (entry.overflowed === true) { + const cut = chunk.indexOf('\n') + if (cut === -1) return + entry.overflowed = false + chunk = chunk.slice(cut + 1) + if (settleOverflow(entry)) void report(room, member, 'overflow', `answer exceeded ${MAX_OUT} bytes`) + } entry.buffer += chunk let index while ((index = entry.buffer.indexOf('\n')) !== -1) { @@ -147,7 +172,11 @@ export function createExternalRoomParticipant(rt) { entry.buffer = entry.buffer.slice(index + 1) if (line !== '') settleLine(entry, line) } - if (entry.buffer.length > MAX_OUT) entry.buffer = '' + // Bound the unterminated tail: remember that this line overflowed instead of erasing it. + if (entry.buffer.length > MAX_OUT) { + entry.overflowed = true + entry.buffer = '' + } }) child.stdin.on('error', () => {}) child.stderr.setEncoding('utf8') diff --git a/core/test/room-external.test.js b/core/test/room-external.test.js index 3fdaaf227..3852fd377 100644 --- a/core/test/room-external.test.js +++ b/core/test/room-external.test.js @@ -280,6 +280,129 @@ test('a record may not carry a command line, only arguments (#198)', () => { { driver: 'subprocess', command: ['node', 'agent.js'], cwd: '/tmp', env: { HOME: '/tmp' } }) }) +// ---- #2689: a line that outgrows the 4 KB in-flight buffer ------------------------------- +// A line is settled only when its newline arrives, so an answer split across the buffer cap used +// to be erased: the turn then hit the reply timeout, the room was told the process timed out, and +// an answer that did exist was counted as silence. The cap bounds the unparsed tail, so an +// over-long line is now a reported failure of the turn, and the process stays for the next one. +const LONG_ANSWER = (text, splitAt) => `const fs=require('node:fs'); +fs.appendFileSync(process.env.SEEN,'start '+process.pid+'\\n'); +let b=''; +process.stdin.setEncoding('utf8'); +process.stdin.on('data',c=>{ + b+=c; + let i; + while((i=b.indexOf('\\n'))!==-1){ + const l=b.slice(0,i); + b=b.slice(i+1); + const m=JSON.parse(l); + if(m.type!=='turn')continue; + if(m.text==='big'){ + const j=JSON.stringify({text:${JSON.stringify(text)}}); + process.stdout.write(j.slice(0,${splitAt})); + setTimeout(()=>process.stdout.write(j.slice(${splitAt})+'\\n'),60); + } else process.stdout.write(JSON.stringify({text:'ok: '+m.text})+'\\n'); + } +});` + +test('an answer line split past the 4 KB buffer is a failure, not a timeout, and keeps the process (#2689)', async (t) => { + const dir = mkdtempSync(join(tmpdir(), 'flybot-external-')) + const seen = join(dir, 'seen.txt') + writeFileSync(seen, '') + const { driver, rt, lines, run } = fixture(t) + rt.roomReplyTimeoutMs = 700 + const member = { id: 'long', name: 'Long', external: { driver: 'subprocess', command: [process.execPath, '-e', LONG_ANSWER('x'.repeat(5000), 4200)], env: { SEEN: seen } } } + const starts = () => readFileSync(seen, 'utf8').split('\n').filter(row => row.startsWith('start ')) + + const reply = await run(member, 'big') + assert.deepEqual(reply, { failure: { reason: 'error', detail: 'answer exceeded 4096 bytes' } }, + 'the answer existed; it is reported as too large, not as silence') + const line = await waitForLine(lines, 'long') + assert.doesNotMatch(line.text, /timed out/, 'an over-long answer is not an unanswered turn') + assert.match(line.text, /more than 4096 bytes/) + assert.equal(driver.snapshot('external:ops:long').state, 'idle') + + // The process was not killed for timing out: the same one answers the next turn. + assert.equal(await run(member, 'two'), 'ok: two') + assert.equal(starts().length, 1, 'a long answer cost the seat its process') +}) + +test('bytes after an over-long line are dropped, never parsed as the answer (#2689)', async (t) => { + // The tail after the overflow happens to be a well-formed answer object; it is the remainder of + // the line that already overflowed, so it must not settle the turn with its text. + const SPLIT_FRAGMENT = `let b=''; +process.stdin.setEncoding('utf8'); +process.stdin.on('data',c=>{ + b+=c; + let i; + while((i=b.indexOf('\\n'))!==-1){ + const l=b.slice(0,i); + b=b.slice(i+1); + const m=JSON.parse(l); + if(m.type!=='turn')continue; + process.stdout.write('{"text":"'+'x'.repeat(4200)); + setTimeout(()=>process.stdout.write('{"text":"sneaky"}\\n'),60); + } +});` + const { rt, run } = fixture(t) + rt.roomReplyTimeoutMs = 700 + const member = { id: 'frag', name: 'Frag', external: { driver: 'subprocess', command: [process.execPath, '-e', SPLIT_FRAGMENT] } } + const reply = await run(member, 'big') + assert.deepEqual(reply, { failure: { reason: 'error', detail: 'answer exceeded 4096 bytes' } }) +}) + +test('a long line logged while no turn is open is discarded to its newline, not buffered (#2689)', async (t) => { + // A child may log; a log line past the cap must not grow the buffer without bound, and it must + // not leave the entry in a state where the next real answer is mangled or the turn fails. + const NOISY = `const fs=require('node:fs'); +fs.appendFileSync(process.env.SEEN,'start '+process.pid+'\\n'); +setTimeout(()=>process.stdout.write('z'.repeat(20000)),50); +setTimeout(()=>process.stdout.write('\\n'),150); +let b=''; +process.stdin.setEncoding('utf8'); +process.stdin.on('data',c=>{ + b+=c; + let i; + while((i=b.indexOf('\\n'))!==-1){ + const l=b.slice(0,i); + b=b.slice(i+1); + const m=JSON.parse(l); + if(m.type==='turn')process.stdout.write(JSON.stringify({text:'ok: '+m.text})+'\\n'); + } +});` + const dir = mkdtempSync(join(tmpdir(), 'flybot-external-')) + const seen = join(dir, 'seen.txt') + writeFileSync(seen, '') + const { driver, room, lines, run } = fixture(t) + const member = { id: 'noisy', name: 'Noisy', external: { driver: 'subprocess', command: [process.execPath, '-e', NOISY], env: { SEEN: seen } } } + // `open` starts the process without a turn: the noise line is written while the seat is idle. + await driver.open({ room, member }) + await waitFor(() => readFileSync(seen, 'utf8').includes('start ')) + // The turn opens after the noise line is closed: the noise never becomes its answer or line. + await new Promise(resolve => setTimeout(resolve, 300)) + assert.equal(await run(member, 'hi'), 'ok: hi') + assert.deepEqual(lines.filter(line => line.botId === 'noisy'), []) +}) + +test('a log line and a second object next to the answer are still ignored (#2689)', async (t) => { + const THREE = `let b=''; +process.stdin.setEncoding('utf8'); +process.stdin.on('data',c=>{ + b+=c; + let i; + while((i=b.indexOf('\\n'))!==-1){ + const l=b.slice(0,i); + b=b.slice(i+1); + const m=JSON.parse(l); + if(m.type==='turn')process.stdout.write('starting up\\n'+JSON.stringify({text:'ok: '+m.text})+'\\n'+JSON.stringify({text:'not this turn'})+'\\n'); + } +});` + const { run } = fixture(t) + const member = { id: 'three', name: 'Three', external: { driver: 'subprocess', command: [process.execPath, '-e', THREE] } } + assert.equal(await run(member, 'one'), 'ok: one') + assert.equal(await run(member, 'two'), 'ok: two') +}) + const waitForLine = async (lines, botId, ms = 4000) => { await waitFor(() => lines.some(line => line.botId === botId), ms) return lines.find(line => line.botId === botId) diff --git a/docs/subsystems/core-host.md b/docs/subsystems/core-host.md index c7c1f37c5..917632191 100644 --- a/docs/subsystems/core-host.md +++ b/docs/subsystems/core-host.md @@ -191,7 +191,7 @@ context rollover for execution events. This internal DSH boundary adds no external backend or task-specific session mapping; see the [same-room participant contract](../interop/room-participants.md) for those gaps. -A member whose Bot record carries `external` is spoken for by a supervised subprocess instead of a DSH Session (#198). `room-participant.js` `createRoomParticipants` is the one boundary: it keeps the DSH driver for every other member and routes a member with that record — and later that member's own session id — to `room-participant-external.js`. The record is a typed block, `external: { driver: 'subprocess', command: [argv...], cwd?, env? }`, written through `roster.js` `cleanExternal`, which refuses a string command (a shared room setting that runs a shell line is an injection surface), any driver but `subprocess`, and an interpreter as `command[0]` (`sh`, `bash`, `pwsh`, `env` and the like): an argv array whose first word is a shell is the same shell line spelled differently. One process is supervised per (room, Bot) and talks one JSON object per line on stdio (`{ type: 'turn', text, note, from, exchangeId }` in, `{ text }` or `{ error }` out), so the same external Bot can be invited to a second room as a second process over one identity. `open`, `runTurn`, `snapshot` and `cancel` reach each driver through the router. Start, clean exit, crash, a user stop and the room-turn timeout are explicit states: a crash is reported as the Bot's own line in the room (`groups.js` `externalRoomLine`), and a turn that reaches a dead process ends as `{ failure }` instead of silence. A process that died is not a dead end: the next turn in that room starts a fresh one, which is what the room's line promises. A stop asks on the protocol first (`{ type: 'stop' }`), sends `SIGTERM` after a short grace and `SIGKILL` after a longer one, so a child that ignores the request and the signal does not outlive its seat; an entry whose stop is still in flight is never reused for the next turn, and a turn the user stopped ends as nothing to answer rather than a failure. A crash line carries the last ~200 characters of the child's `stderr`, flattened to one line, so `exit 1` is diagnosable from the room itself; a timeout reads as `timed out`. A timed-out turn ends the child the same way a stopped one does — the protocol stop line, then `SIGTERM`, then `SIGKILL` — because the protocol carries no turn id: left running, its late answer would settle whichever turn is open then, which is the next one, so that turn would read the previous turn's text ([#2634](https://github.com/ThinkFlowLab/FlyBot/issues/2634)). The next turn therefore starts a fresh process. The seat owns the process: `store.js` `dropGroupSession` closes the `(room, Bot)` process when a member is removed or the Bot is deleted, and `groups.js` `install` closes the rest on host teardown (`ctx.effect`). +A member whose Bot record carries `external` is spoken for by a supervised subprocess instead of a DSH Session (#198). `room-participant.js` `createRoomParticipants` is the one boundary: it keeps the DSH driver for every other member and routes a member with that record — and later that member's own session id — to `room-participant-external.js`. The record is a typed block, `external: { driver: 'subprocess', command: [argv...], cwd?, env? }`, written through `roster.js` `cleanExternal`, which refuses a string command (a shared room setting that runs a shell line is an injection surface), any driver but `subprocess`, and an interpreter as `command[0]` (`sh`, `bash`, `pwsh`, `env` and the like): an argv array whose first word is a shell is the same shell line spelled differently. One process is supervised per (room, Bot) and talks one JSON object per line on stdio (`{ type: 'turn', text, note, from, exchangeId }` in, `{ text }` or `{ error }` out), so the same external Bot can be invited to a second room as a second process over one identity. `open`, `runTurn`, `snapshot` and `cancel` reach each driver through the router. Start, clean exit, crash, a user stop and the room-turn timeout are explicit states: a crash is reported as the Bot's own line in the room (`groups.js` `externalRoomLine`), and a turn that reaches a dead process ends as `{ failure }` instead of silence. A process that died is not a dead end: the next turn in that room starts a fresh one, which is what the room's line promises. A stop asks on the protocol first (`{ type: 'stop' }`), sends `SIGTERM` after a short grace and `SIGKILL` after a longer one, so a child that ignores the request and the signal does not outlive its seat; an entry whose stop is still in flight is never reused for the next turn, and a turn the user stopped ends as nothing to answer rather than a failure. A crash line carries the last ~200 characters of the child's `stderr`, flattened to one line, so `exit 1` is diagnosable from the room itself; a timeout reads as `timed out`. A timed-out turn ends the child the same way a stopped one does — the protocol stop line, then `SIGTERM`, then `SIGKILL` — because the protocol carries no turn id: left running, its late answer would settle whichever turn is open then, which is the next one, so that turn would read the previous turn's text ([#2634](https://github.com/ThinkFlowLab/FlyBot/issues/2634)). The next turn therefore starts a fresh process. The seat owns the process: `store.js` `dropGroupSession` closes the `(room, Bot)` process when a member is removed or the Bot is deleted, and `groups.js` `install` closes the rest on host teardown (`ctx.effect`). stdout is read as lines, and only a newline-terminated line is a message: a line that outgrows the 4 KB in-flight buffer is that turn's failure (`answer exceeded 4096 bytes`), not a timeout: only its bytes are dropped up to the newline, the process stays for the next turn because the child did answer, and the room gets one line saying the answer was too large, while a long line written with no turn open is discarded the same way and never parsed ([#2689](https://github.com/ThinkFlowLab/FlyBot/issues/2689)). ### Group standing section (#894) diff --git a/patches/ds-bot/files/src/host/room-participant-external.js b/patches/ds-bot/files/src/host/room-participant-external.js index e21bbddef..c8c46b9c0 100644 --- a/patches/ds-bot/files/src/host/room-participant-external.js +++ b/patches/ds-bot/files/src/host/room-participant-external.js @@ -6,7 +6,8 @@ // The protocol on the child's stdio is one JSON object per line. // in: { type: 'turn', text, note, from?, exchangeId } { type: 'stop' } // out: { text } or { error } -// Anything else on stdout is ignored, so a process may log. +// Anything else on stdout is ignored, so a process may log. A line that outgrows the in-flight +// buffer is that turn's reported failure, never silence and never a timeout (#2689). import { spawn } from 'node:child_process' import { ROOM_REPLY_TIMEOUT_MS } from './constants.js' @@ -75,6 +76,8 @@ export function createExternalRoomParticipant(rt) { // A state is a sentence the room reads, not a state name: `timeout` reads as `timed out` (#1164). function stateLine(member, state, detail) { if (state === 'crashed') return `${member.name}'s external process crashed${detail ? ` (${detail})` : ''}. Its turns in this room have no answer until it starts again.` + // An over-long line is not a dead or a slow process: the seat keeps it for the next turn. + if (state === 'overflow') return oneLine(`${member.name} answered more than ${MAX_OUT} bytes in one line, so that answer was dropped; the process is still running for the next turn.`) return `${member.name}'s external process ${state === 'timeout' ? 'timed out' : state}${detail ? ` (${detail})` : ''}.` } @@ -108,6 +111,18 @@ export function createExternalRoomParticipant(rt) { } } + // An over-long line is the turn's answer with too many bytes to carry, which is why it ends the + // turn as a failure instead of ending the child: a timeout means the child will not answer, an + // overflow means it answered too large, and only the first one is a reason to kill it (#2689). + function settleOverflow(entry) { + if (entry.turn === undefined) return false + const turn = entry.turn + entry.turn = undefined + entry.state = 'idle' + turn.finish({ failure: { reason: 'error', detail: `answer exceeded ${MAX_OUT} bytes` } }) + return true + } + function start(room, member) { const key = keyOf(room, member) const known = processes.get(key) @@ -122,7 +137,7 @@ export function createExternalRoomParticipant(rt) { else processes.delete(key) } const record = member.external - const entry = { state: 'starting', detail: undefined, turn: undefined, child: undefined, buffer: '' } + const entry = { state: 'starting', detail: undefined, turn: undefined, child: undefined, buffer: '', overflowed: false } processes.set(key, entry) let child try { @@ -140,6 +155,16 @@ export function createExternalRoomParticipant(rt) { entry.state = 'idle' child.stdout.setEncoding('utf8') child.stdout.on('data', (chunk) => { + // The line already overflowed: drop its bytes up to the newline that ends it, then settle + // the turn it belongs to. Only the tail of the line is kept — long enough to find that + // newline, never parsed as an answer (#2689). + if (entry.overflowed === true) { + const cut = chunk.indexOf('\n') + if (cut === -1) return + entry.overflowed = false + chunk = chunk.slice(cut + 1) + if (settleOverflow(entry)) void report(room, member, 'overflow', `answer exceeded ${MAX_OUT} bytes`) + } entry.buffer += chunk let index while ((index = entry.buffer.indexOf('\n')) !== -1) { @@ -147,7 +172,11 @@ export function createExternalRoomParticipant(rt) { entry.buffer = entry.buffer.slice(index + 1) if (line !== '') settleLine(entry, line) } - if (entry.buffer.length > MAX_OUT) entry.buffer = '' + // Bound the unterminated tail: remember that this line overflowed instead of erasing it. + if (entry.buffer.length > MAX_OUT) { + entry.overflowed = true + entry.buffer = '' + } }) child.stdin.on('error', () => {}) child.stderr.setEncoding('utf8') diff --git a/patches/ds-bot/files/test/room-external.test.js b/patches/ds-bot/files/test/room-external.test.js index 3fdaaf227..3852fd377 100644 --- a/patches/ds-bot/files/test/room-external.test.js +++ b/patches/ds-bot/files/test/room-external.test.js @@ -280,6 +280,129 @@ test('a record may not carry a command line, only arguments (#198)', () => { { driver: 'subprocess', command: ['node', 'agent.js'], cwd: '/tmp', env: { HOME: '/tmp' } }) }) +// ---- #2689: a line that outgrows the 4 KB in-flight buffer ------------------------------- +// A line is settled only when its newline arrives, so an answer split across the buffer cap used +// to be erased: the turn then hit the reply timeout, the room was told the process timed out, and +// an answer that did exist was counted as silence. The cap bounds the unparsed tail, so an +// over-long line is now a reported failure of the turn, and the process stays for the next one. +const LONG_ANSWER = (text, splitAt) => `const fs=require('node:fs'); +fs.appendFileSync(process.env.SEEN,'start '+process.pid+'\\n'); +let b=''; +process.stdin.setEncoding('utf8'); +process.stdin.on('data',c=>{ + b+=c; + let i; + while((i=b.indexOf('\\n'))!==-1){ + const l=b.slice(0,i); + b=b.slice(i+1); + const m=JSON.parse(l); + if(m.type!=='turn')continue; + if(m.text==='big'){ + const j=JSON.stringify({text:${JSON.stringify(text)}}); + process.stdout.write(j.slice(0,${splitAt})); + setTimeout(()=>process.stdout.write(j.slice(${splitAt})+'\\n'),60); + } else process.stdout.write(JSON.stringify({text:'ok: '+m.text})+'\\n'); + } +});` + +test('an answer line split past the 4 KB buffer is a failure, not a timeout, and keeps the process (#2689)', async (t) => { + const dir = mkdtempSync(join(tmpdir(), 'flybot-external-')) + const seen = join(dir, 'seen.txt') + writeFileSync(seen, '') + const { driver, rt, lines, run } = fixture(t) + rt.roomReplyTimeoutMs = 700 + const member = { id: 'long', name: 'Long', external: { driver: 'subprocess', command: [process.execPath, '-e', LONG_ANSWER('x'.repeat(5000), 4200)], env: { SEEN: seen } } } + const starts = () => readFileSync(seen, 'utf8').split('\n').filter(row => row.startsWith('start ')) + + const reply = await run(member, 'big') + assert.deepEqual(reply, { failure: { reason: 'error', detail: 'answer exceeded 4096 bytes' } }, + 'the answer existed; it is reported as too large, not as silence') + const line = await waitForLine(lines, 'long') + assert.doesNotMatch(line.text, /timed out/, 'an over-long answer is not an unanswered turn') + assert.match(line.text, /more than 4096 bytes/) + assert.equal(driver.snapshot('external:ops:long').state, 'idle') + + // The process was not killed for timing out: the same one answers the next turn. + assert.equal(await run(member, 'two'), 'ok: two') + assert.equal(starts().length, 1, 'a long answer cost the seat its process') +}) + +test('bytes after an over-long line are dropped, never parsed as the answer (#2689)', async (t) => { + // The tail after the overflow happens to be a well-formed answer object; it is the remainder of + // the line that already overflowed, so it must not settle the turn with its text. + const SPLIT_FRAGMENT = `let b=''; +process.stdin.setEncoding('utf8'); +process.stdin.on('data',c=>{ + b+=c; + let i; + while((i=b.indexOf('\\n'))!==-1){ + const l=b.slice(0,i); + b=b.slice(i+1); + const m=JSON.parse(l); + if(m.type!=='turn')continue; + process.stdout.write('{"text":"'+'x'.repeat(4200)); + setTimeout(()=>process.stdout.write('{"text":"sneaky"}\\n'),60); + } +});` + const { rt, run } = fixture(t) + rt.roomReplyTimeoutMs = 700 + const member = { id: 'frag', name: 'Frag', external: { driver: 'subprocess', command: [process.execPath, '-e', SPLIT_FRAGMENT] } } + const reply = await run(member, 'big') + assert.deepEqual(reply, { failure: { reason: 'error', detail: 'answer exceeded 4096 bytes' } }) +}) + +test('a long line logged while no turn is open is discarded to its newline, not buffered (#2689)', async (t) => { + // A child may log; a log line past the cap must not grow the buffer without bound, and it must + // not leave the entry in a state where the next real answer is mangled or the turn fails. + const NOISY = `const fs=require('node:fs'); +fs.appendFileSync(process.env.SEEN,'start '+process.pid+'\\n'); +setTimeout(()=>process.stdout.write('z'.repeat(20000)),50); +setTimeout(()=>process.stdout.write('\\n'),150); +let b=''; +process.stdin.setEncoding('utf8'); +process.stdin.on('data',c=>{ + b+=c; + let i; + while((i=b.indexOf('\\n'))!==-1){ + const l=b.slice(0,i); + b=b.slice(i+1); + const m=JSON.parse(l); + if(m.type==='turn')process.stdout.write(JSON.stringify({text:'ok: '+m.text})+'\\n'); + } +});` + const dir = mkdtempSync(join(tmpdir(), 'flybot-external-')) + const seen = join(dir, 'seen.txt') + writeFileSync(seen, '') + const { driver, room, lines, run } = fixture(t) + const member = { id: 'noisy', name: 'Noisy', external: { driver: 'subprocess', command: [process.execPath, '-e', NOISY], env: { SEEN: seen } } } + // `open` starts the process without a turn: the noise line is written while the seat is idle. + await driver.open({ room, member }) + await waitFor(() => readFileSync(seen, 'utf8').includes('start ')) + // The turn opens after the noise line is closed: the noise never becomes its answer or line. + await new Promise(resolve => setTimeout(resolve, 300)) + assert.equal(await run(member, 'hi'), 'ok: hi') + assert.deepEqual(lines.filter(line => line.botId === 'noisy'), []) +}) + +test('a log line and a second object next to the answer are still ignored (#2689)', async (t) => { + const THREE = `let b=''; +process.stdin.setEncoding('utf8'); +process.stdin.on('data',c=>{ + b+=c; + let i; + while((i=b.indexOf('\\n'))!==-1){ + const l=b.slice(0,i); + b=b.slice(i+1); + const m=JSON.parse(l); + if(m.type==='turn')process.stdout.write('starting up\\n'+JSON.stringify({text:'ok: '+m.text})+'\\n'+JSON.stringify({text:'not this turn'})+'\\n'); + } +});` + const { run } = fixture(t) + const member = { id: 'three', name: 'Three', external: { driver: 'subprocess', command: [process.execPath, '-e', THREE] } } + assert.equal(await run(member, 'one'), 'ok: one') + assert.equal(await run(member, 'two'), 'ok: two') +}) + const waitForLine = async (lines, botId, ms = 4000) => { await waitFor(() => lines.some(line => line.botId === botId), ms) return lines.find(line => line.botId === botId) From 38c8f89ae43778b92f907440b11cee45b899863e Mon Sep 17 00:00:00 2001 From: Danyang Chen Date: Sun, 8 Nov 2026 21:26:25 +0800 Subject: [PATCH 2/2] Keep an over-long idle log line out of the next turn The overflow flag was set whatever line crossed the cap, and it settled whatever turn was open when that line's newline finally arrived, so a log line written while the seat was idle swallowed a turn opened during its tail (#2689). Remember the turn that owned the line - none when no turn was open - and settle only that one; a turn opened later waits for its own answer. Adds the crossing case to room-external.test.js and refreshes the core-host invariant sentence. --- core/src/host/room-participant-external.js | 11 +++++-- core/test/room-external.test.js | 33 +++++++++++++++++++ docs/subsystems/core-host.md | 2 +- .../src/host/room-participant-external.js | 11 +++++-- .../ds-bot/files/test/room-external.test.js | 33 +++++++++++++++++++ 5 files changed, 85 insertions(+), 5 deletions(-) diff --git a/core/src/host/room-participant-external.js b/core/src/host/room-participant-external.js index c8c46b9c0..5a83c87a4 100644 --- a/core/src/host/room-participant-external.js +++ b/core/src/host/room-participant-external.js @@ -137,7 +137,7 @@ export function createExternalRoomParticipant(rt) { else processes.delete(key) } const record = member.external - const entry = { state: 'starting', detail: undefined, turn: undefined, child: undefined, buffer: '', overflowed: false } + const entry = { state: 'starting', detail: undefined, turn: undefined, child: undefined, buffer: '', overflowed: false, overflowTurn: undefined } processes.set(key, entry) let child try { @@ -161,9 +161,15 @@ export function createExternalRoomParticipant(rt) { if (entry.overflowed === true) { const cut = chunk.indexOf('\n') if (cut === -1) return + // The line belongs to the turn that was open when it overflowed, or to none: a turn that + // opened while its tail was still in flight is not settled by someone else's log line. + const owner = entry.overflowTurn entry.overflowed = false + entry.overflowTurn = undefined chunk = chunk.slice(cut + 1) - if (settleOverflow(entry)) void report(room, member, 'overflow', `answer exceeded ${MAX_OUT} bytes`) + if (entry.turn !== undefined && entry.turn === owner && settleOverflow(entry)) { + void report(room, member, 'overflow', `answer exceeded ${MAX_OUT} bytes`) + } } entry.buffer += chunk let index @@ -175,6 +181,7 @@ export function createExternalRoomParticipant(rt) { // Bound the unterminated tail: remember that this line overflowed instead of erasing it. if (entry.buffer.length > MAX_OUT) { entry.overflowed = true + entry.overflowTurn = entry.turn entry.buffer = '' } }) diff --git a/core/test/room-external.test.js b/core/test/room-external.test.js index 3852fd377..4bd203597 100644 --- a/core/test/room-external.test.js +++ b/core/test/room-external.test.js @@ -384,6 +384,39 @@ process.stdin.on('data',c=>{ assert.deepEqual(lines.filter(line => line.botId === 'noisy'), []) }) +test('a log line that overflowed while idle does not settle a turn opened before its newline (#2689)', async (t) => { + // The line owned no turn when it overflowed, and the turn opens while its tail is still in + // flight: that line's newline is not its answer, so it must not settle the turn. + const CROSSING = `const fs=require('node:fs'); +fs.appendFileSync(process.env.SEEN,'start '+process.pid+'\\n'); +setTimeout(()=>process.stdout.write('z'.repeat(6000)),50); +setTimeout(()=>process.stdout.write('\\n'),400); +let b=''; +process.stdin.setEncoding('utf8'); +process.stdin.on('data',c=>{ + b+=c; + let i; + while((i=b.indexOf('\\n'))!==-1){ + const l=b.slice(0,i); + b=b.slice(i+1); + const m=JSON.parse(l); + if(m.type==='turn')setTimeout(()=>process.stdout.write(JSON.stringify({text:'ok: '+m.text})+'\\n'),200); + } +});` + const dir = mkdtempSync(join(tmpdir(), 'flybot-external-')) + const seen = join(dir, 'seen.txt') + writeFileSync(seen, '') + const { driver, room, lines, run } = fixture(t) + const member = { id: 'cross', name: 'Cross', external: { driver: 'subprocess', command: [process.execPath, '-e', CROSSING], env: { SEEN: seen } } } + await driver.open({ room, member }) + await waitFor(() => readFileSync(seen, 'utf8').includes('start ')) + // The 6000-byte log line has overflowed while the seat is idle; the turn opens before its + // newline lands at ~400 ms, and the child answers 200 ms after that. + await new Promise(resolve => setTimeout(resolve, 300)) + assert.equal(await run(member, 'hi'), 'ok: hi', 'the idle log line did not swallow the turn') + assert.deepEqual(lines.filter(line => line.botId === 'cross'), [], 'an idle log line is not an answer') +}); + test('a log line and a second object next to the answer are still ignored (#2689)', async (t) => { const THREE = `let b=''; process.stdin.setEncoding('utf8'); diff --git a/docs/subsystems/core-host.md b/docs/subsystems/core-host.md index 917632191..0aab9a8b0 100644 --- a/docs/subsystems/core-host.md +++ b/docs/subsystems/core-host.md @@ -191,7 +191,7 @@ context rollover for execution events. This internal DSH boundary adds no external backend or task-specific session mapping; see the [same-room participant contract](../interop/room-participants.md) for those gaps. -A member whose Bot record carries `external` is spoken for by a supervised subprocess instead of a DSH Session (#198). `room-participant.js` `createRoomParticipants` is the one boundary: it keeps the DSH driver for every other member and routes a member with that record — and later that member's own session id — to `room-participant-external.js`. The record is a typed block, `external: { driver: 'subprocess', command: [argv...], cwd?, env? }`, written through `roster.js` `cleanExternal`, which refuses a string command (a shared room setting that runs a shell line is an injection surface), any driver but `subprocess`, and an interpreter as `command[0]` (`sh`, `bash`, `pwsh`, `env` and the like): an argv array whose first word is a shell is the same shell line spelled differently. One process is supervised per (room, Bot) and talks one JSON object per line on stdio (`{ type: 'turn', text, note, from, exchangeId }` in, `{ text }` or `{ error }` out), so the same external Bot can be invited to a second room as a second process over one identity. `open`, `runTurn`, `snapshot` and `cancel` reach each driver through the router. Start, clean exit, crash, a user stop and the room-turn timeout are explicit states: a crash is reported as the Bot's own line in the room (`groups.js` `externalRoomLine`), and a turn that reaches a dead process ends as `{ failure }` instead of silence. A process that died is not a dead end: the next turn in that room starts a fresh one, which is what the room's line promises. A stop asks on the protocol first (`{ type: 'stop' }`), sends `SIGTERM` after a short grace and `SIGKILL` after a longer one, so a child that ignores the request and the signal does not outlive its seat; an entry whose stop is still in flight is never reused for the next turn, and a turn the user stopped ends as nothing to answer rather than a failure. A crash line carries the last ~200 characters of the child's `stderr`, flattened to one line, so `exit 1` is diagnosable from the room itself; a timeout reads as `timed out`. A timed-out turn ends the child the same way a stopped one does — the protocol stop line, then `SIGTERM`, then `SIGKILL` — because the protocol carries no turn id: left running, its late answer would settle whichever turn is open then, which is the next one, so that turn would read the previous turn's text ([#2634](https://github.com/ThinkFlowLab/FlyBot/issues/2634)). The next turn therefore starts a fresh process. The seat owns the process: `store.js` `dropGroupSession` closes the `(room, Bot)` process when a member is removed or the Bot is deleted, and `groups.js` `install` closes the rest on host teardown (`ctx.effect`). stdout is read as lines, and only a newline-terminated line is a message: a line that outgrows the 4 KB in-flight buffer is that turn's failure (`answer exceeded 4096 bytes`), not a timeout: only its bytes are dropped up to the newline, the process stays for the next turn because the child did answer, and the room gets one line saying the answer was too large, while a long line written with no turn open is discarded the same way and never parsed ([#2689](https://github.com/ThinkFlowLab/FlyBot/issues/2689)). +A member whose Bot record carries `external` is spoken for by a supervised subprocess instead of a DSH Session (#198). `room-participant.js` `createRoomParticipants` is the one boundary: it keeps the DSH driver for every other member and routes a member with that record — and later that member's own session id — to `room-participant-external.js`. The record is a typed block, `external: { driver: 'subprocess', command: [argv...], cwd?, env? }`, written through `roster.js` `cleanExternal`, which refuses a string command (a shared room setting that runs a shell line is an injection surface), any driver but `subprocess`, and an interpreter as `command[0]` (`sh`, `bash`, `pwsh`, `env` and the like): an argv array whose first word is a shell is the same shell line spelled differently. One process is supervised per (room, Bot) and talks one JSON object per line on stdio (`{ type: 'turn', text, note, from, exchangeId }` in, `{ text }` or `{ error }` out), so the same external Bot can be invited to a second room as a second process over one identity. `open`, `runTurn`, `snapshot` and `cancel` reach each driver through the router. Start, clean exit, crash, a user stop and the room-turn timeout are explicit states: a crash is reported as the Bot's own line in the room (`groups.js` `externalRoomLine`), and a turn that reaches a dead process ends as `{ failure }` instead of silence. A process that died is not a dead end: the next turn in that room starts a fresh one, which is what the room's line promises. A stop asks on the protocol first (`{ type: 'stop' }`), sends `SIGTERM` after a short grace and `SIGKILL` after a longer one, so a child that ignores the request and the signal does not outlive its seat; an entry whose stop is still in flight is never reused for the next turn, and a turn the user stopped ends as nothing to answer rather than a failure. A crash line carries the last ~200 characters of the child's `stderr`, flattened to one line, so `exit 1` is diagnosable from the room itself; a timeout reads as `timed out`. A timed-out turn ends the child the same way a stopped one does — the protocol stop line, then `SIGTERM`, then `SIGKILL` — because the protocol carries no turn id: left running, its late answer would settle whichever turn is open then, which is the next one, so that turn would read the previous turn's text ([#2634](https://github.com/ThinkFlowLab/FlyBot/issues/2634)). The next turn therefore starts a fresh process. The seat owns the process: `store.js` `dropGroupSession` closes the `(room, Bot)` process when a member is removed or the Bot is deleted, and `groups.js` `install` closes the rest on host teardown (`ctx.effect`). stdout is read as lines, and only a newline-terminated line is a message: a line that outgrows the 4 KB in-flight buffer is that turn's failure (`answer exceeded 4096 bytes`), not a timeout: only its bytes are dropped up to the newline, the process stays for the next turn because the child did answer, and the room gets one line saying the answer was too large, while a long line written with no turn open is discarded to its newline and never parsed, and a turn opened while such a log line's tail is still in flight waits for its own answer instead of being settled by it ([#2689](https://github.com/ThinkFlowLab/FlyBot/issues/2689)). ### Group standing section (#894) diff --git a/patches/ds-bot/files/src/host/room-participant-external.js b/patches/ds-bot/files/src/host/room-participant-external.js index c8c46b9c0..5a83c87a4 100644 --- a/patches/ds-bot/files/src/host/room-participant-external.js +++ b/patches/ds-bot/files/src/host/room-participant-external.js @@ -137,7 +137,7 @@ export function createExternalRoomParticipant(rt) { else processes.delete(key) } const record = member.external - const entry = { state: 'starting', detail: undefined, turn: undefined, child: undefined, buffer: '', overflowed: false } + const entry = { state: 'starting', detail: undefined, turn: undefined, child: undefined, buffer: '', overflowed: false, overflowTurn: undefined } processes.set(key, entry) let child try { @@ -161,9 +161,15 @@ export function createExternalRoomParticipant(rt) { if (entry.overflowed === true) { const cut = chunk.indexOf('\n') if (cut === -1) return + // The line belongs to the turn that was open when it overflowed, or to none: a turn that + // opened while its tail was still in flight is not settled by someone else's log line. + const owner = entry.overflowTurn entry.overflowed = false + entry.overflowTurn = undefined chunk = chunk.slice(cut + 1) - if (settleOverflow(entry)) void report(room, member, 'overflow', `answer exceeded ${MAX_OUT} bytes`) + if (entry.turn !== undefined && entry.turn === owner && settleOverflow(entry)) { + void report(room, member, 'overflow', `answer exceeded ${MAX_OUT} bytes`) + } } entry.buffer += chunk let index @@ -175,6 +181,7 @@ export function createExternalRoomParticipant(rt) { // Bound the unterminated tail: remember that this line overflowed instead of erasing it. if (entry.buffer.length > MAX_OUT) { entry.overflowed = true + entry.overflowTurn = entry.turn entry.buffer = '' } }) diff --git a/patches/ds-bot/files/test/room-external.test.js b/patches/ds-bot/files/test/room-external.test.js index 3852fd377..4bd203597 100644 --- a/patches/ds-bot/files/test/room-external.test.js +++ b/patches/ds-bot/files/test/room-external.test.js @@ -384,6 +384,39 @@ process.stdin.on('data',c=>{ assert.deepEqual(lines.filter(line => line.botId === 'noisy'), []) }) +test('a log line that overflowed while idle does not settle a turn opened before its newline (#2689)', async (t) => { + // The line owned no turn when it overflowed, and the turn opens while its tail is still in + // flight: that line's newline is not its answer, so it must not settle the turn. + const CROSSING = `const fs=require('node:fs'); +fs.appendFileSync(process.env.SEEN,'start '+process.pid+'\\n'); +setTimeout(()=>process.stdout.write('z'.repeat(6000)),50); +setTimeout(()=>process.stdout.write('\\n'),400); +let b=''; +process.stdin.setEncoding('utf8'); +process.stdin.on('data',c=>{ + b+=c; + let i; + while((i=b.indexOf('\\n'))!==-1){ + const l=b.slice(0,i); + b=b.slice(i+1); + const m=JSON.parse(l); + if(m.type==='turn')setTimeout(()=>process.stdout.write(JSON.stringify({text:'ok: '+m.text})+'\\n'),200); + } +});` + const dir = mkdtempSync(join(tmpdir(), 'flybot-external-')) + const seen = join(dir, 'seen.txt') + writeFileSync(seen, '') + const { driver, room, lines, run } = fixture(t) + const member = { id: 'cross', name: 'Cross', external: { driver: 'subprocess', command: [process.execPath, '-e', CROSSING], env: { SEEN: seen } } } + await driver.open({ room, member }) + await waitFor(() => readFileSync(seen, 'utf8').includes('start ')) + // The 6000-byte log line has overflowed while the seat is idle; the turn opens before its + // newline lands at ~400 ms, and the child answers 200 ms after that. + await new Promise(resolve => setTimeout(resolve, 300)) + assert.equal(await run(member, 'hi'), 'ok: hi', 'the idle log line did not swallow the turn') + assert.deepEqual(lines.filter(line => line.botId === 'cross'), [], 'an idle log line is not an answer') +}); + test('a log line and a second object next to the answer are still ignored (#2689)', async (t) => { const THREE = `let b=''; process.stdin.setEncoding('utf8');