diff --git a/changelog.d/2634.fixed.md b/changelog.d/2634.fixed.md new file mode 100644 index 000000000..24d1b4c73 --- /dev/null +++ b/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)). diff --git a/core/src/host/room-participant-external.js b/core/src/host/room-participant-external.js index f09f7509c..e21bbddef 100644 --- a/core/src/host/room-participant-external.js +++ b/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' diff --git a/core/test/room-external.test.js b/core/test/room-external.test.js index 65b094def..3fdaaf227 100644 --- a/core/test/room-external.test.js +++ b/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 diff --git a/docs/subsystems/core-host.md b/docs/subsystems/core-host.md index d3ede1009..5cb041891 100644 --- a/docs/subsystems/core-host.md +++ b/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) 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 f09f7509c..e21bbddef 100644 --- a/patches/ds-bot/files/src/host/room-participant-external.js +++ b/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' diff --git a/patches/ds-bot/files/test/room-external.test.js b/patches/ds-bot/files/test/room-external.test.js index 65b094def..3fdaaf227 100644 --- a/patches/ds-bot/files/test/room-external.test.js +++ b/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