fix(groups): end a stop the child ignores, and let the room see why it died (#1164)

Three defects the merge review of #1153 left as follow-ups, all in the external room driver.

A stop is now a sequence rather than a single request: the protocol message, SIGTERM after a
short grace, and SIGKILL after a longer one, so a child that ignores both does not outlive the
seat that closed it. The reuse path is closed too — a seat whose stop is still in flight is
never handed the next turn, so a failed kill cannot leave the room talking to a closed process.

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 instead of only in a buffer nobody reads. The report
moved to the child's `close` event, which is the point where its last stderr lines have been
read. The timeout state reads as "timed out" rather than "process timeout".

Author: Birfy

changelog.d/1164-external-driver-hardening.fixed.md

@@ -0,0 +1 @@
+The external room driver now ends a stop it is ignored: after `{ type: 'stop' }` and a `SIGTERM` grace it escalates to `SIGKILL`, and a seat whose stop is still in flight is never handed the next turn, so a child cannot outlive its membership. A crash line carries the last ~200 characters of the child's `stderr` (flattened and capped) instead of only `exit N`, and the room-turn timeout reads as "timed out" rather than "process timeout" (#1164).

core/src/host/room-participant-external.js

@@ -17,6 +17,15 @@ const INTERPRETERS = new Set(['sh', 'bash', 'dash', 'zsh', 'ksh', 'mksh', 'cmd',
 const MAX_COMMAND = 32
 const MAX_ENV = 32
 const MAX_OUT = 4096
+// A stop is a request, a signal is a demand, a SIGKILL is the end: a child that ignores the
+// first two would otherwise outlive the seat that closed it (#1164).
+const TERM_GRACE_MS = 50
+const KILL_GRACE_MS = 500
+// 'close' is not guaranteed: a grandchild can hold the inherited stdio open, so a crash is
+// reported from 'exit' after this much stderr has arrived (#1164).
+const DRAIN_MS = 100
+// Only the last words of a crashing child are worth a room line (#1164).
+const STDERR_TAIL = 200
 
 const oneLine = value => String(value ?? '').replace(/[\r\n]+/g, ' ').trim().slice(0, 200)
 
@@ -63,6 +72,15 @@ export function createExternalRoomParticipant(rt) {
   const keyOf = (room, member) => `${room.id}\u0000${member.id}`
   const sessionIdOf = (room, member) => `external:${room.id}:${member.id}`
 
+  // 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.`
+    return `${member.name}'s external process ${state === 'timeout' ? 'timed out' : state}${detail ? ` (${detail})` : ''}.`
+  }
+
+  // A crash line carries the child's last words, flattened and capped, so `exit 1` is diagnosable
+  // from the room itself (#1164).
+  const tailOf = (entry) => String(entry.stderr ?? '').slice(-STDERR_TAIL).replace(/[\r\n]+/g, ' ').trim().slice(0, STDERR_TAIL)
   // The room sees what happened to its participant, not just a turn that never came back (#198).
   async function report(room, member, state, detail) {
     rt.warn?.('ds-bot: external Bot %s in %s: %s%s', member.name, room.name, state, detail ? ` (${detail})` : '')
@@ -73,9 +91,7 @@ export function createExternalRoomParticipant(rt) {
         authorId: member.id,
         authorKind: 'bot',
         kind: 'external',
-        text: state === 'crashed'
-          ? `${member.name}'s external process crashed${detail ? ` (${detail})` : ''}. Its turns in this room have no answer until it starts again.`
-          : `${member.name}'s external process ${state}${detail ? ` (${detail})` : ''}.`,
+        text: stateLine(member, state, detail),
       })
     } catch (error) {
       rt.warn?.('ds-bot: the state of %s in %s was not posted: %s', member.name, room.name, error)
@@ -96,11 +112,14 @@ export function createExternalRoomParticipant(rt) {
     const key = keyOf(room, member)
     const known = processes.get(key)
     if (known !== undefined) {
+      // A live process answers every turn, unless its seat has already been stopped: a stop that
+      // was ignored must never hand the next turn to the process the room just closed (#1164).
+      if (known.stopping === true) processes.delete(key)
       // Still running: the same process answers every turn in this room.
-      if (known.child !== undefined || known.turn !== undefined) return known
+      else if (known.child !== undefined || known.turn !== undefined) return known
       // It died (crash, clean exit, or the user stopped it). The next turn starts a fresh
       // one, which is what the room's line promises (#198).
-      processes.delete(key)
+      else processes.delete(key)
     }
     const record = member.external
     const entry = { state: 'starting', detail: undefined, turn: undefined, child: undefined, buffer: '' }
@@ -135,19 +154,40 @@ export function createExternalRoomParticipant(rt) {
     child.stderr.on('data', chunk => { entry.stderr = `${entry.stderr ?? ''}${chunk}`.slice(-MAX_OUT) })
     child.on('error', (error) => {
       entry.child = undefined
+      entry.reported = true
       settle(entry, 'crashed', oneLine(error?.message ?? error))
       void report(room, member, 'crashed', entry.detail)
     })
-    child.on('exit', (code, signal) => {
-      entry.child = undefined
-      // A stop the child obeyed — it read the message and left on its own — is the same exit the
-      // signal branch below skips: it stopped when asked, so the room must not hear 'crashed' (#198).
+    const reportExit = (code, signal) => {
+      // One line either way: whichever of 'exit' and 'close' reports first wins (#1164).
+      if (entry.reported === true) return
+      entry.reported = true
+      // A stop the child obeyed — it read the message and left on its own — is not an ending the
+      // room should hear about: it stopped when asked, so no 'crashed' line (#198).
       if (entry.state === 'stopped' || entry.stopping === true) return
-      const clean = code === 0 && entry.stopping !== true
+      const clean = code === 0
       const detail = signal !== null ? `signal ${signal}` : `exit ${code}`
-      settle(entry, clean ? 'exited' : 'crashed', detail)
-      void report(room, member, entry.state, detail)
+      const tail = clean ? '' : tailOf(entry)
+      const said = tail === '' ? detail : `${detail}: ${tail}`
+      settle(entry, clean ? 'exited' : 'crashed', said)
+      void report(room, member, entry.state, said)
       entry.stopping = false
+    }
+    // 'exit' must free the seat at once — otherwise the next turn writes into a dead child and the
+    // room only sees a missed turn. It also arms one short drain, so the child's last stderr lines
+    // can still arrive before the crash line is posted (#1164).
+    let drain
+    child.on('exit', (code, signal) => {
+      entry.child = undefined
+      drain = setTimeout(() => reportExit(code, signal), DRAIN_MS)
+      drain.unref?.()
+    })
+    // 'close' carries the full tail, but it may never come: a grandchild holding the inherited
+    // stdio postpones it for as long as it lives, so it is the confirmation, not the only path.
+    child.on('close', (code, signal) => {
+      clearTimeout(drain)
+      entry.child = undefined
+      reportExit(code, signal)
     })
     return entry
   }
@@ -160,10 +200,13 @@ export function createExternalRoomParticipant(rt) {
     // The protocol message is a courtesy the child may act on; the signal follows so a child that
     // never reads still goes (#198).
     try { child.stdin.write(`${JSON.stringify({ type: 'stop' })}\n`) } catch { /* the pipe may already be closed */ }
-    const kill = () => { try { child.kill('SIGTERM') } catch { /* the process may already be gone */ } }
-    const grace = setTimeout(kill, 50)
-    grace.unref?.()
-    child.once?.('exit', () => clearTimeout(grace))
+    const sendSignal = (name) => { try { child.kill(name) } catch { /* the process may already be gone */ } }
+    // SIGTERM first, then SIGKILL: a stop that is ignored is not a stop (#1164).
+    const term = setTimeout(() => sendSignal('SIGTERM'), TERM_GRACE_MS)
+    const hard = setTimeout(() => sendSignal('SIGKILL'), TERM_GRACE_MS + KILL_GRACE_MS)
+    term.unref?.()
+    hard.unref?.()
+    child.once?.('exit', () => { clearTimeout(term); clearTimeout(hard) })
   }
 
   // A member who loses the seat, or a Bot that is deleted, takes its process with it: a child

core/test/room-external.test.js

@@ -129,6 +129,91 @@ test('a turn that times out comes back as no answer, not as a model failure (#19
   assert.equal(entry.botId, 'slow')
 })
 
+test('a child that ignores the stop is killed after the grace (#1164)', async (t) => {
+  // SIGTERM is handled and deliberately ignored: only an escalation can take this child.
+  const STUBBORN = "const fs=require('node:fs');fs.appendFileSync(process.env.SEEN,'start '+process.pid+'\\n');process.on('SIGTERM',()=>{});process.stdin.resume()"
+  const dir = mkdtempSync(join(tmpdir(), 'flybot-external-'))
+  const seen = join(dir, 'pids.txt')
+  writeFileSync(seen, '')
+  const { driver, run } = fixture(t)
+  const member = { id: 'stubborn', name: 'Stubborn', external: { driver: 'subprocess', command: [process.execPath, '-e', STUBBORN], env: { SEEN: seen } } }
+  const turn = run(member, 'work')
+  await waitFor(() => /start \d+/.test(readFileSync(seen, 'utf8')))
+  const pid = Number(readFileSync(seen, 'utf8').match(/start (\d+)/)[1])
+  assert.equal(driver.closeParticipant('ops', 'stubborn'), true)
+  assert.equal(await turn, null)
+  await waitFor(() => { try { process.kill(pid, 0); return false } catch { return true } }, 4000)
+})
+
+test('a seat that was closed is never handed the next turn (#1164)', async (t) => {
+  // The child records each turn with its own pid and never answers. A stopped turn leaves the
+  // entry in place while the child may still be running: that is the reuse path — the next turn
+  // must reach a fresh process, not the one the user just stopped.
+  const PID_LOG = "const fs=require('node:fs');fs.appendFileSync(process.env.SEEN,'start '+process.pid+'\\n');process.on('SIGTERM',()=>{});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);if(JSON.parse(l).type==='turn')fs.appendFileSync(process.env.SEEN,'turn '+process.pid+'\\n')}})"
+  const dir = mkdtempSync(join(tmpdir(), 'flybot-external-'))
+  const seen = join(dir, 'pids.txt')
+  writeFileSync(seen, '')
+  const { driver, run } = fixture(t)
+  const member = { id: 'seat2', name: 'Seat2', external: { driver: 'subprocess', command: [process.execPath, '-e', PID_LOG], env: { SEEN: seen } } }
+  const turns = () => readFileSync(seen, 'utf8').split('\n').filter(row => row.startsWith('turn ')).map(row => row.slice(5))
+  const stopped = run(member, 'one')
+  await waitFor(() => turns().length === 1)
+  const first = turns()[0]
+  assert.equal(await driver.cancel({ sessionId: 'external:ops:seat2' }), true)
+  assert.equal(await stopped, null)
+  const next = run(member, 'two')
+  await waitFor(() => turns().length === 2)
+  const second = turns()[1]
+  assert.notEqual(second, first, 'the closed process must not answer the next turn')
+  await driver.cancel({ sessionId: 'external:ops:seat2' })
+  assert.equal(await next, null)
+})
+test('the timeout state line says timed out (#1164)', async (t) => {
+  const { rt, lines, run } = fixture(t)
+  rt.roomReplyTimeoutMs = 200
+  const member = { id: 'late', name: 'Late', external: { driver: 'subprocess', command: [process.execPath, '-e', 'process.stdin.resume()'] } }
+  assert.deepEqual(await run(member, 'work'), { missed: true })
+  const entry = await waitForLine(lines, 'late')
+  assert.match(entry.text, /external process timed out/)
+  assert.doesNotMatch(entry.text, /process timeout/)
+})
+
+test('a crash line carries the last words from stderr (#1164)', async (t) => {
+  const BOOM = "process.stderr.write('boom: missing config\\n',()=>process.exit(3))"
+  const { lines, run } = fixture(t)
+  const member = { id: 'boom', name: 'Boom', external: { driver: 'subprocess', command: [process.execPath, '-e', BOOM] } }
+  assert.equal((await run(member))?.failure?.reason, 'crashed')
+  const entry = await waitForLine(lines, 'boom')
+  assert.match(entry.text, /crashed \(exit 3: boom: missing config\)/)
+  // The tail is capped: 400 characters of stderr keep only the last 200.
+  const NOISE = "process.stderr.write('x'.repeat(400),()=>process.exit(3))"
+  const noisy = { id: 'noise', name: 'Noise', external: { driver: 'subprocess', command: [process.execPath, '-e', NOISE] } }
+  assert.equal((await run(noisy))?.failure?.reason, 'crashed')
+  const noise = await waitForLine(lines, 'noise')
+  assert.match(noise.text, /x{200}/)
+  assert.doesNotMatch(noise.text, /x{201}/)
+})
+test('a crash is reported when a grandchild holds the stdio open (#1164)', async (t) => {
+  // On the first launch the child leaves a grandchild on its inherited stdio and exits 1, so
+  // 'close' is postponed for as long as the grandchild lives (measured here: exit at ~40 ms,
+  // close at ~1280 ms). The crash line must not wait for 'close', and the seat must be free at
+  // once. The marker makes the second launch a normal responder.
+  const HELD = "const fs=require('node:fs');const cp=require('node:child_process');if(!fs.existsSync(process.env.MARK)){fs.writeFileSync(process.env.MARK,'1');fs.writeSync(2,'boom: helper holds the pipe\\n');cp.spawn(process.execPath,['-e','setTimeout(()=>{},1200)'],{stdio:'inherit'});process.exit(1)}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:'again: '+m.text})+'\\n')}})"
+  const dir = mkdtempSync(join(tmpdir(), 'flybot-external-'))
+  const mark = join(dir, 'mark')
+  const { lines, run } = fixture(t)
+  const member = { id: 'held', name: 'Held', external: { driver: 'subprocess', command: [process.execPath, '-e', HELD], env: { MARK: mark } } }
+  const started = Date.now()
+  const first = run(member)
+  // 900 ms is before the grandchild's 1.2 s: only reporting from 'exit' can meet this.
+  const entry = await waitForLine(lines, 'held', 900)
+  const elapsed = Date.now() - started
+  assert.match(entry.text, /crashed \(exit 1: boom: helper holds the pipe\)/)
+  assert.ok(elapsed < 900, `the crash line waited for close (${elapsed} ms)`)
+  assert.equal((await first)?.failure?.reason, 'crashed')
+  // The seat was cleared on 'exit', so the next turn reaches a fresh process, not a dead pipe.
+  assert.equal(await run(member, 'ok'), 'again: ok')
+})
 test('a record may not carry a command line, only arguments (#198)', () => {
   assert.throws(() => cleanExternal({ command: 'node agent.js' }), /array of arguments/)
   assert.throws(() => cleanExternal({ command: [] }), /1 to 32/)
@@ -140,6 +225,10 @@ test('a record may not carry a command line, only arguments (#198)', () => {
     { driver: 'subprocess', command: ['node', 'agent.js'], cwd: '/tmp', env: { HOME: '/tmp' } })
 })
 
+const waitForLine = async (lines, botId, ms = 4000) => {
+  await waitFor(() => lines.some(line => line.botId === botId), ms)
+  return lines.find(line => line.botId === botId)
+}
 const waitFor = async (done, ms = 4000) => {
   const started = Date.now()
   while (!done()) {