From e2afacd6343a01cc3f88d70cd92662be8aeace46 Mon Sep 17 00:00:00 2001 From: Danyang Chen Date: Mon, 9 Nov 2026 04:37:33 +0800 Subject: [PATCH] fix(resume): write inflight.json atomically MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The notes of a running turn were written in place: the writer truncated the file and then wrote the new bytes, so a crash in that window left it empty or half written, and the next start scheduled no pickup at all — every turn the crash cut off was dropped silently (#2705). Stage a temp name beside the file and rename it over the old one, so a crash at any point leaves either the previous complete notes or the new complete notes. A file that still cannot be parsed is named once with its path and the reason instead of passing for "nothing was in flight", and the first write of a host sweeps the temp names a dead predecessor left behind. Tests: core/test/resume-atomic.test.js runs the real install() in child processes and kills it with kill -9 inside the write; the previous writer left the target truncated and scheduled no pickup. --- changelog.d/2705-inflight-atomic.fixed.md | 1 + core/src/host/resume.js | 66 ++++-- core/test/resume-atomic.test.js | 200 ++++++++++++++++++ docs/subsystems/core-host.md | 4 +- patches/ds-bot/0001-core-fixes.patch | 96 +++++++-- .../ds-bot/files/test/resume-atomic.test.js | 200 ++++++++++++++++++ 6 files changed, 540 insertions(+), 27 deletions(-) create mode 100644 changelog.d/2705-inflight-atomic.fixed.md create mode 100644 core/test/resume-atomic.test.js create mode 100644 patches/ds-bot/files/test/resume-atomic.test.js diff --git a/changelog.d/2705-inflight-atomic.fixed.md b/changelog.d/2705-inflight-atomic.fixed.md new file mode 100644 index 000000000..8a144f2b4 --- /dev/null +++ b/changelog.d/2705-inflight-atomic.fixed.md @@ -0,0 +1 @@ +`inflight.json` — the record that lets an interrupted turn resume — is now staged in a temp file and renamed over the old one, so a crash in the middle of the write leaves the previous complete notes instead of an empty or half-written file. Every turn and queued message the crash cut off is still picked up at the next start, where an empty file used to drop them all silently. A file that still cannot be parsed (a hand-edit, a disk error) is now reported once with its path and the reason instead of passing for "nothing was in flight", and a temp name a crashed write left behind is swept by the next host that writes. See [#2705](https://github.com/ThinkFlowLab/FlyBot/issues/2705). diff --git a/core/src/host/resume.js b/core/src/host/resume.js index 85397a19f..2b74676c1 100644 --- a/core/src/host/resume.js +++ b/core/src/host/resume.js @@ -5,10 +5,12 @@ // running Bot turn and each message it delivers until a turn takes it up, in // inflight.json; what is still noted at the next start is picked up once. import { randomUUID } from 'node:crypto' -import { readFileSync } from 'node:fs' -import { join } from 'node:path' +import { readFileSync, readdirSync, unlinkSync } from 'node:fs' +import { unlink } from 'node:fs/promises' +import { basename, join } from 'node:path' import { SOURCE_KIND } from './constants.js' -import { ensurePrivateDir, writePrivateFile } from './private-files.js' +import { ensurePrivateDir, replacePrivateFile } from './private-files.js' +import { tempWritable } from './real-dir.js' import { OTHER, settled } from './turn-origin.js' import { retiredBot } from './store.js' @@ -36,12 +38,25 @@ const triggerOf = info => ({ ...(info.senderRoomId ? { senderRoomId: info.senderRoomId } : {}), }) -const readNotes = (file) => { +// A file that is not the notes cannot be resumed safely, but that is not the same answer as +// "nothing was in flight": every turn it named is dropped, so it is named once — path and reason — +// instead of passing for a clean start (#2705). +const readNotes = (file, warn) => { + let text try { - const parsed = JSON.parse(readFileSync(file, 'utf8')) + text = readFileSync(file, 'utf8') + } catch (error) { + // A missing file is the ordinary case: nothing was noted, or the last write cleared it. + if (error?.code !== 'ENOENT') warn('ds-bot: reading %s failed (%s); nothing is picked up from it', file, error?.message ?? error) + return { turns: {}, queued: {} } + } + try { + const parsed = JSON.parse(text) + if (parsed === null || typeof parsed !== 'object' || Array.isArray(parsed)) throw new Error('not the object of turns and queued messages') const object = value => (value !== null && typeof value === 'object' && !Array.isArray(value) ? value : {}) - return { turns: object(parsed?.turns), queued: object(parsed?.queued) } - } catch { + return { turns: object(parsed.turns), queued: object(parsed.queued) } + } catch (error) { + warn('ds-bot: %s could not be read as the notes of a running turn (%s); the turns and messages it named are not picked up', file, error?.message ?? error) return { turns: {}, queued: {} } } } @@ -51,7 +66,7 @@ export function install(rt) { const file = join(home, 'inflight.json') // What the last run left; read before this run notes anything. - let interrupted = readNotes(file) + let interrupted = readNotes(file, rt.warn) /** @type {Record} */ let turns = {} @@ -61,6 +76,20 @@ export function install(rt) { let stopped = false let latest let due = false + // A write this host never finished left its temp name behind. It holds no state — the target it + // was going to replace is still the last complete notes — so the first write of the next host + // sweeps it. Only a writing host does: a read-only window leaves the home alone. + let swept = false + const sweepTemps = () => { + swept = true + const prefix = `${basename(file)}.` + let names = [] + try { names = readdirSync(home) } catch { return } + for (const name of names) { + if (!name.startsWith(prefix) || !name.endsWith('.tmp')) continue + try { unlinkSync(join(home, name)) } catch { /* another host may be writing it */ } + } + } const persist = () => { // Only the host that saves the team runs its work again. if (stopped || rt.readOnly() !== undefined) return @@ -69,15 +98,26 @@ export function install(rt) { // write the file once per change. if (due) return due = true - // Written in place: a rename over the file fails on Windows while anything reads - // it, and a torn write only loses what one restart would pick up. The notes name the - // Session, the room and the message a turn was handed, so the file is private like the - // team file: created 0600 in a 0700 home, and an existing file keeps the mode it has (#2697). + // Renamed into place, never written through: the file is the only record of the turns a crash + // cut off, and a crash inside an in-place write left it empty or half written — every one of + // them dropped on the next start (#2705). The notes name the Session, the room and the message + // a turn was handed, so the temp name is private like the team file: created 0600 in a 0700 + // home, at the mode an existing file already has (#2697). writing = writing.then(async () => { due = false const snapshot = latest await ensurePrivateDir(home) - await writePrivateFile(file, snapshot) + if (!swept) sweepTemps() + const temp = `${file}.${process.pid}.${randomUUID()}.tmp` + // `writeFile` follows a link at the temp name, which would take the notes outside the home + // and the rename after them would put the link where the notes belong (temp-name sweep). + if (!tempWritable(temp)) throw new Error(`${temp} is not a plain file of its own, so the notes were not written there`) + try { + await replacePrivateFile(temp, file, snapshot) + } catch (error) { + await unlink(temp).catch(() => { /* nothing was staged, or the rename already took it */ }) + throw error + } }).catch(error => rt.warn('ds-bot: noting running turns failed: %s', error)) } const originNote = (id) => { diff --git a/core/test/resume-atomic.test.js b/core/test/resume-atomic.test.js new file mode 100644 index 000000000..8218324a7 --- /dev/null +++ b/core/test/resume-atomic.test.js @@ -0,0 +1,200 @@ +// `inflight.json` is the one record of the turns a crash cut off, so the write that keeps it must +// never leave the file empty or half written: the writer stages a temp name beside it and renames +// it over, so a crash at any point leaves either the previous complete notes or the new complete +// notes (#2705). These cases run real child processes and a real `kill -9` in a temp home, so the +// window is the one the shipped code has, and a file nothing can parse is still reported instead of +// passing for "nothing was in flight". +import assert from 'node:assert/strict' +import { spawn } from 'node:child_process' +import { existsSync, mkdtempSync, readdirSync, readFileSync, rmSync, statSync, writeFileSync } from 'node:fs' +import { tmpdir } from 'node:os' +import { join } from 'node:path' +import { test } from 'node:test' +import { fileURLToPath } from 'node:url' + +const RESUME = fileURLToPath(new URL('../src/host/resume.js', import.meta.url)) +// The core modules resolve `@deepseek-ai/*` through this hook; the child processes need it too. +const HARNESS = fileURLToPath(new URL('./harness-resolve.mjs', import.meta.url)) + +const NOTE = { + turns: { 'sess-cut-off': { botId: 'bot-1', since: 1, turn: 3, trigger: { role: 'request', exchangeId: 'ex-1', senderSessionId: 'sess-peer' } } }, + queued: { 'sess-peer': [{ at: 2, args: { toId: 'bot-1', sessionId: 'sess-cut-off', exchangeId: 'ex-2', text: 'answer me' } }] }, +} +const NOTE_TEXT = JSON.stringify(NOTE, null, 2) + +const children = [] +const start = (code, args = []) => { + const child = spawn(process.execPath, ['--import', HARNESS, '--input-type=module', '-e', code, ...args], { stdio: ['ignore', 'pipe', 'pipe'] }) + children.push(child) + let output = '' + let errors = '' + child.stdout.setEncoding('utf8') + child.stderr.setEncoding('utf8') + child.stdout.on('data', chunk => { output += chunk }) + child.stderr.on('data', chunk => { errors += chunk }) + const exited = new Promise(resolve => child.on('exit', (code_, signal) => resolve({ code: code_, signal }))) + return { child, exited, get output() { return output }, get errors() { return errors } } +} +// Waits for the child's last printed line, which carries its answer as JSON. +const answer = async (run, tag, timeoutMs = 30_000) => { + const deadline = Date.now() + timeoutMs + while (!new RegExp(`^@@${tag} `, 'm').test(run.output) && Date.now() < deadline) await new Promise(resolve => setTimeout(resolve, 5)) + const matched = run.output.match(new RegExp(`^@@${tag} (.*)$`, 'm')) + assert.ok(matched, `the child never answered ${tag}\nstdout: ${run.output}\nstderr: ${run.errors}`) + return JSON.parse(matched[1]) +} +// A real install of the module under test: the pickup timer is captured, not run, so the case can +// read what the next start would schedule without the pickup itself racing the assertions. +const STUB = ` +const home = process.argv[1] +const timers = [] +const warns = [] +globalThis.setTimeout = (fn, ms) => { timers.push(ms); return { unref() {} } } +const effects = [] +const ctx = { on() {}, effect(fn) { effects.push(fn) }, sessionController: undefined } +const rt = { + ctx, home, load: async () => {}, readOnly: () => undefined, deliver: async () => {}, + chatOf: id => id, botOf: () => undefined, isBotSession: () => false, + warn: (...args) => warns.push(args.map(value => String(value)).join(' ')), +} +` +const READER = ` +import { install as installResume } from ${JSON.stringify(RESUME)} +${STUB} +installResume(rt) +for (const fn of effects) fn() +process.stdout.write('@@read ' + JSON.stringify({ timers, warns }) + '\\n') +` +const readNextStart = async (home) => { + const run = start(READER, [home]) + const seen = await answer(run, 'read') + await run.exited + return seen +} +// A real writer: it notes one delivery and then sits there, so the parent can kill it inside the +// window a real crash would land in. +const WRITER = ` +import { install as installResume } from ${JSON.stringify(RESUME)} +${STUB} +installResume(rt) +for (const fn of effects) fn() +process.stdout.write('@@ready ' + JSON.stringify({}) + '\\n') +// A real handle: the stubbed timer above would otherwise let the loop empty and the child exit +// before the write it is here for. +setInterval(() => {}, 1000) +await rt.deliver({ toId: 'bot-1', text: 'x'.repeat(40 * 1024 * 1024) }) +await new Promise(() => {}) +` +const tempNames = home => readdirSync(home).filter(name => name.startsWith('inflight.json.') && name.endsWith('.tmp')) +// The window a crash has to land in: the write is under way, so the in-place writer has already +// shortened the target and the atomic one has a temp name growing beside it. +const cutWindow = (home, previous) => { + for (const name of tempNames(home)) { + const size = statSync(join(home, name)).size + if (size > 0) return { temp: name, size } + } + const target = join(home, 'inflight.json') + if (!existsSync(target)) return { target: 'missing' } + const size = statSync(target).size + return size < previous ? { target: size } : undefined +} +const scratch = (t) => { + const home = mkdtempSync(join(tmpdir(), 'ds-bot-resume-atomic-')) + t.after(() => rmSync(home, { recursive: true, force: true, maxRetries: 5 })) + return home +} + +test('a write cut off by kill -9 keeps the complete notes on disk, and the next start picks them up', { timeout: 60_000 }, async (t) => { + const home = scratch(t) + writeFileSync(join(home, 'inflight.json'), NOTE_TEXT, { mode: 0o600 }) + + const run = start(WRITER, [home]) + await answer(run, 'ready') + let window + const deadline = Date.now() + 30_000 + while (window === undefined && Date.now() < deadline) { + window = cutWindow(home, NOTE_TEXT.length) + if (window === undefined) await new Promise(resolve => setTimeout(resolve, 1)) + } + assert.ok(window, 'the writer never showed a write in flight') + run.child.kill('SIGKILL') + const { signal } = await run.exited + assert.equal(signal, 'SIGKILL') + t.diagnostic(`cut the write inside the window: ${JSON.stringify(window)}`) + + // Whatever the crash landed on, the target itself is a complete record: a torn write in a temp + // name that was never renamed over it cannot shorten it. + const bytes = readFileSync(join(home, 'inflight.json'), 'utf8') + assert.equal(bytes.length, NOTE_TEXT.length, 'the killed write left the previous notes, not a half-written snapshot') + let onDisk + assert.doesNotThrow(() => { onDisk = JSON.parse(bytes) }, 'the killed write left a complete inflight.json behind') + assert.deepEqual(onDisk, NOTE, 'the notes on disk are the last complete ones') + const next = await readNextStart(home) + assert.deepEqual(next.timers, [3000], 'the next start schedules the pickup of the cut-off turn') + assert.deepEqual(next.warns, [], 'a complete file says nothing') +}) + +test('a file nothing can parse is reported once instead of passing for "nothing was in flight"', { timeout: 30_000 }, async (t) => { + const home = scratch(t) + const file = join(home, 'inflight.json') + + writeFileSync(file, '', { mode: 0o600 }) + const empty = await readNextStart(home) + assert.deepEqual(empty.timers, [], 'nothing is picked up from a file with no notes in it') + assert.equal(empty.warns.length, 1, 'the empty file is named once') + assert.match(empty.warns[0], /inflight\.json/) + assert.match(empty.warns[0], /not picked up|nothing is picked up|could not be read/i) + + writeFileSync(file, NOTE_TEXT.slice(0, Math.floor(NOTE_TEXT.length / 2)), { mode: 0o600 }) + const half = await readNextStart(home) + assert.deepEqual(half.timers, []) + assert.equal(half.warns.length, 1, 'the half-written file is named once') + + // A file that parses but is not the notes shape cannot be resumed either, and says so. + writeFileSync(file, '[]', { mode: 0o600 }) + const wrong = await readNextStart(home) + assert.deepEqual(wrong.timers, []) + assert.equal(wrong.warns.length, 1) + + // A missing file is the ordinary case: no pickup, and nothing to report. + rmSync(file) + const absent = await readNextStart(home) + assert.deepEqual(absent, { timers: [], warns: [] }) +}) + +test('a temp name a crash left behind is not read as state, and the next write clears it', { timeout: 30_000 }, async (t) => { + const home = scratch(t) + const stale = join(home, 'inflight.json.999999.deadbeef.tmp') + writeFileSync(stale, '{"turns":{},"queued":{}}', { mode: 0o600 }) + + const run = start(` + import { existsSync, readFileSync } from 'node:fs' + import { install as installResume } from ${JSON.stringify(RESUME)} + const home = process.argv[1] + const effects = [] + const ctx = { on() {}, effect(fn) { effects.push(fn) }, sessionController: undefined } + const rt = { + ctx, home, load: async () => {}, readOnly: () => undefined, deliver: async () => {}, + chatOf: id => id, botOf: () => undefined, isBotSession: () => false, + warn: message => process.stderr.write('warn: ' + String(message) + '\\n'), + } + installResume(rt) + for (const fn of effects) fn() + await rt.deliver({ toId: 'bot-1', text: 'after the crash' }) + const notes = home + '/inflight.json' + for (let i = 0; i < 2000; i += 1) { + await new Promise(resolve => setImmediate(resolve)) + try { if (JSON.parse(readFileSync(notes, 'utf8')).queued !== undefined) break } catch {} + } + process.stdout.write('@@wrote ' + JSON.stringify({ exists: existsSync(notes) }) + '\\n') + `, [home]) + const wrote = await answer(run, 'wrote') + await run.exited + assert.equal(wrote.exists, true) + assert.equal(existsSync(stale), false, 'the crashed write\'s temp name is not left in the home') + assert.match(readFileSync(join(home, 'inflight.json'), 'utf8'), /after the crash/) +}) + +test.after(() => { + for (const child of children) if (child.exitCode === null && child.signalCode === null) child.kill('SIGKILL') +}) diff --git a/docs/subsystems/core-host.md b/docs/subsystems/core-host.md index 59541d4d9..e09906d0a 100644 --- a/docs/subsystems/core-host.md +++ b/docs/subsystems/core-host.md @@ -230,7 +230,7 @@ An admin's `@` used to be the room's control plane: every `@Dev #1234 rebase and ### Host files are private (#2697) -The host's own files carry the user's content — a Bot's memory and the memory section each Session froze, the saved team (group messages included), the full-text index of every conversation part, the decision log, the OCR cache of the user's images and `inflight.json` (the Session, room and message a running turn was handed) — so each is created `0600` inside a `0700` folder, independent of the process umask. `src/host/private-files.js` is the one place those numbers live: `ensurePrivateDir` (`mkdir` with the mode, then `chmod`, because `mkdir` leaves an existing folder alone and the umask filters the mode it does set) and `stagePrivateFile`/`replacePrivateFile` (write the temp name, then `chmod` it to the mode the file being replaced already has, then `rename`). A rewrite therefore never widens: an owner's `chmod 0400` survives the next `remember`, and a file that does not exist yet is created private; `writePrivateFile` is the in-place variant for `inflight.json`, whose mode is set when the name is created and kept after that. SQLite creates the chat index itself with the ambient mode, so `chat-index.js` chmods the index and its `-wal`/`-shm` names right after SQLite makes them, and the home's own mode covers a WAL file that appears between two checks. Content that is public or computed (a build output, an exported static file) may keep the ambient mode, but the choice is written where it is made, never left to the default. Covered by `core/test/private-files.test.js`, which sets `umask 022` inside the test process and restores it. +The host's own files carry the user's content — a Bot's memory and the memory section each Session froze, the saved team (group messages included), the full-text index of every conversation part, the decision log, the OCR cache of the user's images and `inflight.json` (the Session, room and message a running turn was handed) — so each is created `0600` inside a `0700` folder, independent of the process umask. `src/host/private-files.js` is the one place those numbers live: `ensurePrivateDir` (`mkdir` with the mode, then `chmod`, because `mkdir` leaves an existing folder alone and the umask filters the mode it does set) and `stagePrivateFile`/`replacePrivateFile` (write the temp name, then `chmod` it to the mode the file being replaced already has, then `rename`). A rewrite therefore never widens: an owner's `chmod 0400` survives the next `remember`, and a file that does not exist yet is created private; `inflight.json` takes the same `stagePrivateFile` + `rename` shape with a `pid` and a random suffix in the temp name, so a crash inside the write leaves the previous complete notes rather than an empty file, and the first write of a host sweeps the temp names its dead predecessor left in the home. SQLite creates the chat index itself with the ambient mode, so `chat-index.js` chmods the index and its `-wal`/`-shm` names right after SQLite makes them, and the home's own mode covers a WAL file that appears between two checks. Content that is public or computed (a build output, an exported static file) may keep the ambient mode, but the choice is written where it is made, never left to the default. Covered by `core/test/private-files.test.js`, which sets `umask 022` inside the test process and restores it. ### Team state and access @@ -472,7 +472,7 @@ Steer alone admits a human line only at the next step boundary, so a foreground A group room takes the same 60s line, through the room's own path (#1650). A person typing in a room lands on the room's own Session (`room.id`), not in the member's, so the own-chat gate never saw it and the member's turn held the round for the whole wait. Now, when such a line (the splice, or the `user/message` that follows it) lands on a room's Session, the member turn the round is waiting on is live, and that member's oldest open wait is more than `INTERRUPT_AFTER_MS` old, the host lets the turn go through `stopRoomTurn(sessionId, exchangeId)` — the very cancel Stop uses, `{ kind: 'hook', reason: 'ds-bot/stopped' }` with no `keepInbox`, so nothing the member had begun reaches the room, exactly as after a Stop. The line is safe because it waits in the room's queued inbox, and the next round answers it. A room id never takes `reason: 'ds-bot/interrupt-long-wait'`; that reason stays the own-chat one. Only the member the round is waiting on is let go — another member running in the same room is untouched — and waits of 60s or less, a finished wait, and an idle Session are all left alone. -- **Pickup after a restart** (`resume.js`): every running Bot turn and every delivered-but-not-taken message is noted in `inflight.json`. At the next start a noted turn is picked up once only if a restart really cut it: the durable Session log's `turn/end` for that turn is `interrupted` (or an abort caused by `disposed`, or there is no end). A turn that finished while this plugin was not listening — `completed`, `blocked`, `max-tokens`, `forked`, or aborted by the user / hook / parent — is not repeated; a `turn/start` seen after the start means the noted turn ended too. Notes from a build before the record carried `turn` keep the old behaviour. A pickup is confirmed before its note is dropped: the notes read at start are seeded into the live `turns`/`queued` (so the file keeps everything still owed), a turn's note stays while its pickup runs and is dropped only once handed over (unless the resumed turn has written its own note first), and a queued message whose resend fails is kept for the next start, an exchange is noted once (a resend delivers the `exchangeId` the notes already hold, and noting it again left a copy behind after a turn took it up, so the start after that sent the message once more), and a turn taking up a message forgets every note of its exchange; a crash or a failed delivery during the pickup no longer loses the work — a pickup is at-least-once, not once: a crash between handing the turn over and the write that drops its note repeats that pickup once, because the `taken` set and the inbox that would tell the next start otherwise are process-local. That is the safe side of the trade, a repeated turn the user can see instead of work the user was told about going silently dead. Covered by `core/test/follow-through.test.js`. +- **Pickup after a restart** (`resume.js`): every running Bot turn and every delivered-but-not-taken message is noted in `inflight.json`. At the next start a noted turn is picked up once only if a restart really cut it: the durable Session log's `turn/end` for that turn is `interrupted` (or an abort caused by `disposed`, or there is no end). A turn that finished while this plugin was not listening — `completed`, `blocked`, `max-tokens`, `forked`, or aborted by the user / hook / parent — is not repeated; a `turn/start` seen after the start means the noted turn ended too. Notes from a build before the record carried `turn` keep the old behaviour. A pickup is confirmed before its note is dropped: the notes read at start are seeded into the live `turns`/`queued` (so the file keeps everything still owed), a turn's note stays while its pickup runs and is dropped only once handed over (unless the resumed turn has written its own note first), and a queued message whose resend fails is kept for the next start, an exchange is noted once (a resend delivers the `exchangeId` the notes already hold, and noting it again left a copy behind after a turn took it up, so the start after that sent the message once more), and a turn taking up a message forgets every note of its exchange; a crash or a failed delivery during the pickup no longer loses the work — a pickup is at-least-once, not once: a crash between handing the turn over and the write that drops its note repeats that pickup once, because the `taken` set and the inbox that would tell the next start otherwise are process-local. That is the safe side of the trade, a repeated turn the user can see instead of work the user was told about going silently dead. The file itself is staged as a temp name and renamed over the old one, so a crash inside the write leaves the previous complete notes instead of the empty file the reader would take for "nothing was in flight" (#2705); a file that still cannot be parsed (a hand-edit, a disk error) is reported once with its path and the reason rather than silently dropping every turn it named. Covered by `core/test/follow-through.test.js` and `core/test/resume-atomic.test.js`. - A repeated Bot-to-Bot line becomes one turn (#641): `deliver()` keys the last 60 s of Bot-to-Bot text by sender + recipient Session + the text with its whitespace collapsed (`DEDUPE_MS`, `duplicateDelivery`), drops a repeat before it is recorded as an exchange or handed to the Session, and journals the drop as `delivery: dedupe -> ` in the host warning log. The key includes the Session, so the same line to another Session (a group copy, another Bot) is its own delivery; `deferred` / `resumed` / `retried` / `resent` carry on a message the Session already has and are never judged, and a friend's line, the user's own message and a group round are not Bot-to-Bot. Routing (parent and direct children) and the existing unread-queue refusal are unchanged, and the map is process-local and capped at 500 entries. Covered by `core/test/repeated-delivery.test.js`. - **The check line is 45% of the usable window, not 80%** (`steward.js` `linesFor`): a Bot's conversation is condensed once its estimated size passes `DEFAULT_CHECK_RATIO` of the model's declared window less the room its reply needs, so the conversation is condensed earlier than upstream ds-bot does. The share is a floor on the prompt of a long turn, because every call re-sends the fixed block and the history is most of each prompt (#796); the trade is a checkpoint sooner and less detail kept (#1233). `checkRatio` in the profile still sets the share, and the hard line stays at 95% of the usable window, so a share at or above 0.95 can never lift the check line above it. Covered by `core/test/steward.test.js`. - **A checkpoint is judged by the prompt that is left** (`parts.js` `pressureLeft`): the token meter stays anchored on the provider usage from before the span was replaced, and its surface delta only subtracts a 4-characters-per-token estimate of the removed text. A denser report — the mock's 3 characters per token, or CJK on a real model — can therefore still read above the check line after the checkpoint has replaced the span, and the part would move on even though the steward said to condense in place. After a compaction, the part stays when the heuristic size of the remaining surface plus the tool schemas is under the check line. A surface that is still over it moves on, as before. The decision to cross the line in the first place still uses the anchored total (#917). diff --git a/patches/ds-bot/0001-core-fixes.patch b/patches/ds-bot/0001-core-fixes.patch index 16459e958..849569a95 100644 --- a/patches/ds-bot/0001-core-fixes.patch +++ b/patches/ds-bot/0001-core-fixes.patch @@ -6981,43 +6981,115 @@ diff --git a/src/host/prompt.js b/src/host/prompt.js diff --git a/src/host/resume.js b/src/host/resume.js --- a/src/host/resume.js +++ b/src/host/resume.js -@@ -6,9 +6,11 @@ +@@ -5,10 +5,14 @@ + // running Bot turn and each message it delivers until a turn takes it up, in // inflight.json; what is still noted at the next start is picked up once. import { randomUUID } from 'node:crypto' - import { readFileSync } from 'node:fs' +-import { readFileSync } from 'node:fs' -import { mkdir, writeFile } from 'node:fs/promises' - import { join } from 'node:path' +-import { join } from 'node:path' ++import { readFileSync, readdirSync, unlinkSync } from 'node:fs' ++import { unlink } from 'node:fs/promises' ++import { basename, join } from 'node:path' import { SOURCE_KIND } from './constants.js' -+import { ensurePrivateDir, writePrivateFile } from './private-files.js' ++import { ensurePrivateDir, replacePrivateFile } from './private-files.js' ++import { tempWritable } from './real-dir.js' +import { OTHER, settled } from './turn-origin.js' +import { retiredBot } from './store.js' // Long enough for DSH to finish opening its workspaces after a start. const RESUME_DELAY_MS = 3000 -@@ -51,7 +51,7 @@ export function install(rt) { +@@ -34,12 +34,25 @@ const triggerOf = info => ({ + ...(info.senderRoomId ? { senderRoomId: info.senderRoomId } : {}), + }) + +-const readNotes = (file) => { ++// A file that is not the notes cannot be resumed safely, but that is not the same answer as ++// "nothing was in flight": every turn it named is dropped, so it is named once — path and reason — ++// instead of passing for a clean start (#2705). ++const readNotes = (file, warn) => { ++ let text + try { +- const parsed = JSON.parse(readFileSync(file, 'utf8')) ++ text = readFileSync(file, 'utf8') ++ } catch (error) { ++ // A missing file is the ordinary case: nothing was noted, or the last write cleared it. ++ if (error?.code !== 'ENOENT') warn('ds-bot: reading %s failed (%s); nothing is picked up from it', file, error?.message ?? error) ++ return { turns: {}, queued: {} } ++ } ++ try { ++ const parsed = JSON.parse(text) ++ if (parsed === null || typeof parsed !== 'object' || Array.isArray(parsed)) throw new Error('not the object of turns and queued messages') + const object = value => (value !== null && typeof value === 'object' && !Array.isArray(value) ? value : {}) +- return { turns: object(parsed?.turns), queued: object(parsed?.queued) } +- } catch { ++ return { turns: object(parsed.turns), queued: object(parsed.queued) } ++ } catch (error) { ++ warn('ds-bot: %s could not be read as the notes of a running turn (%s); the turns and messages it named are not picked up', file, error?.message ?? error) + return { turns: {}, queued: {} } + } + } +@@ -49,9 +49,9 @@ export function install(rt) { + const file = join(home, 'inflight.json') + // What the last run left; read before this run notes anything. - let interrupted = readNotes(file) +- let interrupted = readNotes(file) ++ let interrupted = readNotes(file, rt.warn) - /** @type {Record} */ + /** @type {Record} */ let turns = {} /** @type {Record} deliveries no turn has taken up, by Session */ let queued = {} -@@ -68,14 +68,20 @@ export function install(rt) { +@@ -59,6 +59,20 @@ export function install(rt) { + let stopped = false + let latest + let due = false ++ // A write this host never finished left its temp name behind. It holds no state — the target it ++ // was going to replace is still the last complete notes — so the first write of the next host ++ // sweeps it. Only a writing host does: a read-only window leaves the home alone. ++ let swept = false ++ const sweepTemps = () => { ++ swept = true ++ const prefix = `${basename(file)}.` ++ let names = [] ++ try { names = readdirSync(home) } catch { return } ++ for (const name of names) { ++ if (!name.startsWith(prefix) || !name.endsWith('.tmp')) continue ++ try { unlinkSync(join(home, name)) } catch { /* another host may be writing it */ } ++ } ++ } + const persist = () => { + // Only the host that saves the team runs its work again. + if (stopped || rt.readOnly() !== undefined) return +@@ -67,15 +67,32 @@ export function install(rt) { + // write the file once per change. if (due) return due = true - // Written in place: a rename over the file fails on Windows while anything reads +- // Written in place: a rename over the file fails on Windows while anything reads - // it, and a torn write only loses what one restart would pick up. -+ // it, and a torn write only loses what one restart would pick up. The notes name the -+ // Session, the room and the message a turn was handed, so the file is private like the -+ // team file: created 0600 in a 0700 home, and an existing file keeps the mode it has (#2697). ++ // Renamed into place, never written through: the file is the only record of the turns a crash ++ // cut off, and a crash inside an in-place write left it empty or half written — every one of ++ // them dropped on the next start (#2705). The notes name the Session, the room and the message ++ // a turn was handed, so the temp name is private like the team file: created 0600 in a 0700 ++ // home, at the mode an existing file already has (#2697). writing = writing.then(async () => { due = false const snapshot = latest - await mkdir(home, { recursive: true }) - await writeFile(file, snapshot, 'utf8') + await ensurePrivateDir(home) -+ await writePrivateFile(file, snapshot) ++ if (!swept) sweepTemps() ++ const temp = `${file}.${process.pid}.${randomUUID()}.tmp` ++ // `writeFile` follows a link at the temp name, which would take the notes outside the home ++ // and the rename after them would put the link where the notes belong (temp-name sweep). ++ if (!tempWritable(temp)) throw new Error(`${temp} is not a plain file of its own, so the notes were not written there`) ++ try { ++ await replacePrivateFile(temp, file, snapshot) ++ } catch (error) { ++ await unlink(temp).catch(() => { /* nothing was staged, or the rename already took it */ }) ++ throw error ++ } }).catch(error => rt.warn('ds-bot: noting running turns failed: %s', error)) } + const originNote = (id) => { diff --git a/patches/ds-bot/files/test/resume-atomic.test.js b/patches/ds-bot/files/test/resume-atomic.test.js new file mode 100644 index 000000000..8218324a7 --- /dev/null +++ b/patches/ds-bot/files/test/resume-atomic.test.js @@ -0,0 +1,200 @@ +// `inflight.json` is the one record of the turns a crash cut off, so the write that keeps it must +// never leave the file empty or half written: the writer stages a temp name beside it and renames +// it over, so a crash at any point leaves either the previous complete notes or the new complete +// notes (#2705). These cases run real child processes and a real `kill -9` in a temp home, so the +// window is the one the shipped code has, and a file nothing can parse is still reported instead of +// passing for "nothing was in flight". +import assert from 'node:assert/strict' +import { spawn } from 'node:child_process' +import { existsSync, mkdtempSync, readdirSync, readFileSync, rmSync, statSync, writeFileSync } from 'node:fs' +import { tmpdir } from 'node:os' +import { join } from 'node:path' +import { test } from 'node:test' +import { fileURLToPath } from 'node:url' + +const RESUME = fileURLToPath(new URL('../src/host/resume.js', import.meta.url)) +// The core modules resolve `@deepseek-ai/*` through this hook; the child processes need it too. +const HARNESS = fileURLToPath(new URL('./harness-resolve.mjs', import.meta.url)) + +const NOTE = { + turns: { 'sess-cut-off': { botId: 'bot-1', since: 1, turn: 3, trigger: { role: 'request', exchangeId: 'ex-1', senderSessionId: 'sess-peer' } } }, + queued: { 'sess-peer': [{ at: 2, args: { toId: 'bot-1', sessionId: 'sess-cut-off', exchangeId: 'ex-2', text: 'answer me' } }] }, +} +const NOTE_TEXT = JSON.stringify(NOTE, null, 2) + +const children = [] +const start = (code, args = []) => { + const child = spawn(process.execPath, ['--import', HARNESS, '--input-type=module', '-e', code, ...args], { stdio: ['ignore', 'pipe', 'pipe'] }) + children.push(child) + let output = '' + let errors = '' + child.stdout.setEncoding('utf8') + child.stderr.setEncoding('utf8') + child.stdout.on('data', chunk => { output += chunk }) + child.stderr.on('data', chunk => { errors += chunk }) + const exited = new Promise(resolve => child.on('exit', (code_, signal) => resolve({ code: code_, signal }))) + return { child, exited, get output() { return output }, get errors() { return errors } } +} +// Waits for the child's last printed line, which carries its answer as JSON. +const answer = async (run, tag, timeoutMs = 30_000) => { + const deadline = Date.now() + timeoutMs + while (!new RegExp(`^@@${tag} `, 'm').test(run.output) && Date.now() < deadline) await new Promise(resolve => setTimeout(resolve, 5)) + const matched = run.output.match(new RegExp(`^@@${tag} (.*)$`, 'm')) + assert.ok(matched, `the child never answered ${tag}\nstdout: ${run.output}\nstderr: ${run.errors}`) + return JSON.parse(matched[1]) +} +// A real install of the module under test: the pickup timer is captured, not run, so the case can +// read what the next start would schedule without the pickup itself racing the assertions. +const STUB = ` +const home = process.argv[1] +const timers = [] +const warns = [] +globalThis.setTimeout = (fn, ms) => { timers.push(ms); return { unref() {} } } +const effects = [] +const ctx = { on() {}, effect(fn) { effects.push(fn) }, sessionController: undefined } +const rt = { + ctx, home, load: async () => {}, readOnly: () => undefined, deliver: async () => {}, + chatOf: id => id, botOf: () => undefined, isBotSession: () => false, + warn: (...args) => warns.push(args.map(value => String(value)).join(' ')), +} +` +const READER = ` +import { install as installResume } from ${JSON.stringify(RESUME)} +${STUB} +installResume(rt) +for (const fn of effects) fn() +process.stdout.write('@@read ' + JSON.stringify({ timers, warns }) + '\\n') +` +const readNextStart = async (home) => { + const run = start(READER, [home]) + const seen = await answer(run, 'read') + await run.exited + return seen +} +// A real writer: it notes one delivery and then sits there, so the parent can kill it inside the +// window a real crash would land in. +const WRITER = ` +import { install as installResume } from ${JSON.stringify(RESUME)} +${STUB} +installResume(rt) +for (const fn of effects) fn() +process.stdout.write('@@ready ' + JSON.stringify({}) + '\\n') +// A real handle: the stubbed timer above would otherwise let the loop empty and the child exit +// before the write it is here for. +setInterval(() => {}, 1000) +await rt.deliver({ toId: 'bot-1', text: 'x'.repeat(40 * 1024 * 1024) }) +await new Promise(() => {}) +` +const tempNames = home => readdirSync(home).filter(name => name.startsWith('inflight.json.') && name.endsWith('.tmp')) +// The window a crash has to land in: the write is under way, so the in-place writer has already +// shortened the target and the atomic one has a temp name growing beside it. +const cutWindow = (home, previous) => { + for (const name of tempNames(home)) { + const size = statSync(join(home, name)).size + if (size > 0) return { temp: name, size } + } + const target = join(home, 'inflight.json') + if (!existsSync(target)) return { target: 'missing' } + const size = statSync(target).size + return size < previous ? { target: size } : undefined +} +const scratch = (t) => { + const home = mkdtempSync(join(tmpdir(), 'ds-bot-resume-atomic-')) + t.after(() => rmSync(home, { recursive: true, force: true, maxRetries: 5 })) + return home +} + +test('a write cut off by kill -9 keeps the complete notes on disk, and the next start picks them up', { timeout: 60_000 }, async (t) => { + const home = scratch(t) + writeFileSync(join(home, 'inflight.json'), NOTE_TEXT, { mode: 0o600 }) + + const run = start(WRITER, [home]) + await answer(run, 'ready') + let window + const deadline = Date.now() + 30_000 + while (window === undefined && Date.now() < deadline) { + window = cutWindow(home, NOTE_TEXT.length) + if (window === undefined) await new Promise(resolve => setTimeout(resolve, 1)) + } + assert.ok(window, 'the writer never showed a write in flight') + run.child.kill('SIGKILL') + const { signal } = await run.exited + assert.equal(signal, 'SIGKILL') + t.diagnostic(`cut the write inside the window: ${JSON.stringify(window)}`) + + // Whatever the crash landed on, the target itself is a complete record: a torn write in a temp + // name that was never renamed over it cannot shorten it. + const bytes = readFileSync(join(home, 'inflight.json'), 'utf8') + assert.equal(bytes.length, NOTE_TEXT.length, 'the killed write left the previous notes, not a half-written snapshot') + let onDisk + assert.doesNotThrow(() => { onDisk = JSON.parse(bytes) }, 'the killed write left a complete inflight.json behind') + assert.deepEqual(onDisk, NOTE, 'the notes on disk are the last complete ones') + const next = await readNextStart(home) + assert.deepEqual(next.timers, [3000], 'the next start schedules the pickup of the cut-off turn') + assert.deepEqual(next.warns, [], 'a complete file says nothing') +}) + +test('a file nothing can parse is reported once instead of passing for "nothing was in flight"', { timeout: 30_000 }, async (t) => { + const home = scratch(t) + const file = join(home, 'inflight.json') + + writeFileSync(file, '', { mode: 0o600 }) + const empty = await readNextStart(home) + assert.deepEqual(empty.timers, [], 'nothing is picked up from a file with no notes in it') + assert.equal(empty.warns.length, 1, 'the empty file is named once') + assert.match(empty.warns[0], /inflight\.json/) + assert.match(empty.warns[0], /not picked up|nothing is picked up|could not be read/i) + + writeFileSync(file, NOTE_TEXT.slice(0, Math.floor(NOTE_TEXT.length / 2)), { mode: 0o600 }) + const half = await readNextStart(home) + assert.deepEqual(half.timers, []) + assert.equal(half.warns.length, 1, 'the half-written file is named once') + + // A file that parses but is not the notes shape cannot be resumed either, and says so. + writeFileSync(file, '[]', { mode: 0o600 }) + const wrong = await readNextStart(home) + assert.deepEqual(wrong.timers, []) + assert.equal(wrong.warns.length, 1) + + // A missing file is the ordinary case: no pickup, and nothing to report. + rmSync(file) + const absent = await readNextStart(home) + assert.deepEqual(absent, { timers: [], warns: [] }) +}) + +test('a temp name a crash left behind is not read as state, and the next write clears it', { timeout: 30_000 }, async (t) => { + const home = scratch(t) + const stale = join(home, 'inflight.json.999999.deadbeef.tmp') + writeFileSync(stale, '{"turns":{},"queued":{}}', { mode: 0o600 }) + + const run = start(` + import { existsSync, readFileSync } from 'node:fs' + import { install as installResume } from ${JSON.stringify(RESUME)} + const home = process.argv[1] + const effects = [] + const ctx = { on() {}, effect(fn) { effects.push(fn) }, sessionController: undefined } + const rt = { + ctx, home, load: async () => {}, readOnly: () => undefined, deliver: async () => {}, + chatOf: id => id, botOf: () => undefined, isBotSession: () => false, + warn: message => process.stderr.write('warn: ' + String(message) + '\\n'), + } + installResume(rt) + for (const fn of effects) fn() + await rt.deliver({ toId: 'bot-1', text: 'after the crash' }) + const notes = home + '/inflight.json' + for (let i = 0; i < 2000; i += 1) { + await new Promise(resolve => setImmediate(resolve)) + try { if (JSON.parse(readFileSync(notes, 'utf8')).queued !== undefined) break } catch {} + } + process.stdout.write('@@wrote ' + JSON.stringify({ exists: existsSync(notes) }) + '\\n') + `, [home]) + const wrote = await answer(run, 'wrote') + await run.exited + assert.equal(wrote.exists, true) + assert.equal(existsSync(stale), false, 'the crashed write\'s temp name is not left in the home') + assert.match(readFileSync(join(home, 'inflight.json'), 'utf8'), /after the crash/) +}) + +test.after(() => { + for (const child of children) if (child.exitCode === null && child.signalCode === null) child.kill('SIGKILL') +})