End an external room child when its turn times out

changelog.d/2634.fixed.md

@@ -0,0 +1 @@
+An external participant whose room turn times out is now stopped instead of being left running, so a late answer to the timed-out turn can no longer become the next turn's answer — a slow process no longer replies one turn behind ([#2634](https://github.com/ThinkFlowLab/FlyBot/issues/2634)).

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

@@ -285,6 +285,9 @@ export function createExternalRoomParticipant(rt) {
         finish(null)
       }
       timer = setTimeout(() => {
+        // The child is ended, not left working: the protocol carries no turn id, so its late line
+        // would settle whichever turn is open then — the next one (#2634).
+        endChild(entry)
         // A slow process is 'no answer', exactly what the DSH driver reports for the same case:
         // the room's round maps { missed: true } to silence, not to a model failure (#198).
         entry.state = 'timeout'

core/test/room-external.test.js

@@ -145,6 +145,61 @@ test('a child that ignores the stop is killed after the grace (#1164)', async (t
   await waitFor(() => { try { process.kill(pid, 0); return false } catch { return true } }, 4000)
 })
 
+test('a timed-out turn never settles the next one with its late answer (#2634)', async (t) => {
+  // The child holds its answer to the first turn until the marker appears and handles turns one
+  // at a time, so the stale line lands while the second turn is in flight (#2634).
+  const HELD = `const fs=require('node:fs');
+const queue=[];
+let pumping=false;
+function pump(){
+  if(pumping||queue.length===0)return;
+  if(queue[0].text==='one'&&!fs.existsSync(process.env.MARK)){setTimeout(pump,5);return}
+  pumping=true;
+  const m=queue.shift();
+  process.stdout.write(JSON.stringify({text:'answer to '+m.text})+'\\n');
+  setTimeout(()=>{pumping=false;pump()},0);
+}
+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;
+    fs.appendFileSync(process.env.SEEN,'turn '+process.pid+' '+m.text+'\\n');
+    queue.push(m);
+    pump();
+  }
+});`
+  const dir = mkdtempSync(join(tmpdir(), 'flybot-external-'))
+  const seen = join(dir, 'seen.txt')
+  const mark = join(dir, 'mark')
+  writeFileSync(seen, '')
+  const { rt, run } = fixture(t)
+  rt.roomReplyTimeoutMs = 300
+  const member = { id: 'held2', name: 'Held2', external: { driver: 'subprocess', command: [process.execPath, '-e', HELD], env: { SEEN: seen, MARK: mark } } }
+  const rows = () => readFileSync(seen, 'utf8').split('\n').filter(Boolean)
+  const seenTurns = () => rows().filter(row => row.startsWith('turn '))
+  const starts = () => rows().filter(row => row.startsWith('start '))
+  const first = run(member, 'one')
+  await waitFor(() => seenTurns().length === 1)
+  assert.deepEqual(await first, { missed: true })
+  // The second turn goes to whatever the driver believes answers it; the marker then releases the
+  // first turn's answer, which must not become the second turn's answer.
+  const second = run(member, 'two')
+  await waitFor(() => seenTurns().length === 2)
+  writeFileSync(mark, '1')
+  const answer = await second
+  assert.notDeepEqual(answer, 'answer to one', "the next turn was settled by the timed-out turn's late answer")
+  if (typeof answer === 'string') assert.equal(answer, 'answer to two')
+  await waitFor(() => starts().length === 2)
+  assert.notEqual(starts()[1], starts()[0], 'a timed-out process must not answer the next turn')
+})
+
 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

docs/subsystems/core-host.md

@@ -190,7 +190,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`. 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`).
 
 ### Group standing section (#894)
 

patches/ds-bot/files/src/host/room-participant-external.js

@@ -285,6 +285,9 @@ export function createExternalRoomParticipant(rt) {
         finish(null)
       }
       timer = setTimeout(() => {
+        // The child is ended, not left working: the protocol carries no turn id, so its late line
+        // would settle whichever turn is open then — the next one (#2634).
+        endChild(entry)
         // A slow process is 'no answer', exactly what the DSH driver reports for the same case:
         // the room's round maps { missed: true } to silence, not to a model failure (#198).
         entry.state = 'timeout'

patches/ds-bot/files/test/room-external.test.js

@@ -145,6 +145,61 @@ test('a child that ignores the stop is killed after the grace (#1164)', async (t
   await waitFor(() => { try { process.kill(pid, 0); return false } catch { return true } }, 4000)
 })
 
+test('a timed-out turn never settles the next one with its late answer (#2634)', async (t) => {
+  // The child holds its answer to the first turn until the marker appears and handles turns one
+  // at a time, so the stale line lands while the second turn is in flight (#2634).
+  const HELD = `const fs=require('node:fs');
+const queue=[];
+let pumping=false;
+function pump(){
+  if(pumping||queue.length===0)return;
+  if(queue[0].text==='one'&&!fs.existsSync(process.env.MARK)){setTimeout(pump,5);return}
+  pumping=true;
+  const m=queue.shift();
+  process.stdout.write(JSON.stringify({text:'answer to '+m.text})+'\\n');
+  setTimeout(()=>{pumping=false;pump()},0);
+}
+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;
+    fs.appendFileSync(process.env.SEEN,'turn '+process.pid+' '+m.text+'\\n');
+    queue.push(m);
+    pump();
+  }
+});`
+  const dir = mkdtempSync(join(tmpdir(), 'flybot-external-'))
+  const seen = join(dir, 'seen.txt')
+  const mark = join(dir, 'mark')
+  writeFileSync(seen, '')
+  const { rt, run } = fixture(t)
+  rt.roomReplyTimeoutMs = 300
+  const member = { id: 'held2', name: 'Held2', external: { driver: 'subprocess', command: [process.execPath, '-e', HELD], env: { SEEN: seen, MARK: mark } } }
+  const rows = () => readFileSync(seen, 'utf8').split('\n').filter(Boolean)
+  const seenTurns = () => rows().filter(row => row.startsWith('turn '))
+  const starts = () => rows().filter(row => row.startsWith('start '))
+  const first = run(member, 'one')
+  await waitFor(() => seenTurns().length === 1)
+  assert.deepEqual(await first, { missed: true })
+  // The second turn goes to whatever the driver believes answers it; the marker then releases the
+  // first turn's answer, which must not become the second turn's answer.
+  const second = run(member, 'two')
+  await waitFor(() => seenTurns().length === 2)
+  writeFileSync(mark, '1')
+  const answer = await second
+  assert.notDeepEqual(answer, 'answer to one', "the next turn was settled by the timed-out turn's late answer")
+  if (typeof answer === 'string') assert.equal(answer, 'answer to two')
+  await waitFor(() => starts().length === 2)
+  assert.notEqual(starts()[1], starts()[0], 'a timed-out process must not answer the next turn')
+})
+
 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