Report an over-long room answer instead of dropping it as a timeout

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

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, overflowTurn: undefined }
     processes.set(key, entry)
     let child
     try {
@@ -140,14 +155,35 @@ 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
+        // 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 (entry.turn !== undefined && entry.turn === owner && settleOverflow(entry)) {
+          void report(room, member, 'overflow', `answer exceeded ${MAX_OUT} bytes`)
+        }
+      }
       entry.buffer += chunk
       let index
       while ((index = entry.buffer.indexOf('\n')) !== -1) {
         const line = entry.buffer.slice(0, index).trim()
         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.overflowTurn = entry.turn
+        entry.buffer = ''
+      }
     })
     child.stdin.on('error', () => {})
     child.stderr.setEncoding('utf8')

core/test/room-external.test.js

@@ -280,6 +280,162 @@ 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 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');
+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)

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