diff --git a/changelog.d/2807.fixed.md b/changelog.d/2807.fixed.md new file mode 100644 index 00000000..7f8e2467 --- /dev/null +++ b/changelog.d/2807.fixed.md @@ -0,0 +1 @@ +The memory journal (`/memory/journal.jsonl`) no longer grows without bound: it keeps the newest 500 changes, and at most 1 MiB, folding anything older into a single `truncated` record that says how many changes it replaced. Remembering and forgetting a fact used to leave one journal line behind forever; the memory itself was already at a steady state, so only this log kept growing ([#2807](https://github.com/ThinkFlowLab/FlyBot/issues/2807)). diff --git a/core/src/host/memory.js b/core/src/host/memory.js index 39f3b995..309719bc 100644 --- a/core/src/host/memory.js +++ b/core/src/host/memory.js @@ -20,9 +20,9 @@ // memory section the first time it renders, so its prompt stays byte-stable and the // provider's prefix cache holds; a fresh part, or a checkpoint (which breaks the cache // anyway), freezes it anew. -import { createHash } from 'node:crypto' +import { createHash, randomUUID } from 'node:crypto' import { existsSync, readFileSync, readdirSync, unlinkSync } from 'node:fs' -import { appendFile, rename, unlink } from 'node:fs/promises' +import { appendFile, readFile, rename, stat, unlink } from 'node:fs/promises' import { dirname, join, relative, sep } from 'node:path' import { BlockAssembler } from '@deepseek-ai/dsh-llm' import { PRIVATE_FILE_MODE, ensurePrivateDir, ensurePrivateDirSync, replacePrivateFile, replacePrivateFileSync } from './private-files.js' @@ -80,6 +80,13 @@ const REVIEW_MEMORY_MAX = 16_000 // its lines leave the group log (which keeps 60). const REVIEW_GROUP_LAG = 30 const NEWS_TEXT_MAX = 160 +// The memory journal is a side channel, not history to keep forever (#2807): it holds the newest +// changes under a line cap and a byte cap, and the changes past them are folded into one +// `{ op: 'truncated', dropped }` line. A trim keeps most of the room, so once the file reaches a +// cap it is rewritten every few changes rather than on every one. +export const JOURNAL_KEPT = 500 +export const JOURNAL_MAX_BYTES = 1_024 * 1_024 +const JOURNAL_KEEP_TO = 0.9 // --------------------------------------------------------------------------- // Entries and files, as text. @@ -569,6 +576,60 @@ export function install(rt) { } await replacePrivateFile(temp, path, text) } + // The journal keeps its newest changes and folds the rest away, the way the state journal does + // (#2807): at most `JOURNAL_KEPT` lines and `JOURNAL_MAX_BYTES` bytes. The fold is one + // `truncated` line carrying how many changes it replaced; it keeps an earlier fold's count, so + // the tally survives repeated trims. The rewrite goes through a temp name, so a crash never + // leaves half a line. + let journalLines = null + let journalTrimRefused = false + const countJournalLines = async (path) => { + try { + return (await readFile(path, 'utf8')).split('\n').filter(Boolean).length + } catch (error) { + if (error?.code === 'ENOENT') return 0 + throw error + } + } + async function trimJournal(path) { + let text + try { + text = await readFile(path, 'utf8') + } catch (error) { + if (error?.code === 'ENOENT') return 0 + throw error + } + const lines = text.split('\n').filter(Boolean) + let dropped = 0 + try { + const first = JSON.parse(lines[0]) + if (first?.op === 'truncated' && Number.isSafeInteger(first.dropped) && first.dropped >= 0) dropped = first.dropped + } catch { /* the oldest line is folded with the rest */ } + const keepLines = Math.floor(JOURNAL_KEPT * JOURNAL_KEEP_TO) + const keepBytes = Math.floor(JOURNAL_MAX_BYTES * JOURNAL_KEEP_TO) + let from = 0 + let bytes = Buffer.byteLength(text, 'utf8') + while (from < lines.length - 1 && (lines.length - from > keepLines || bytes > keepBytes)) { + bytes -= Buffer.byteLength(lines[from], 'utf8') + 1 + from += 1 + } + dropped += from + const next = [JSON.stringify({ time: new Date().toISOString(), op: 'truncated', dropped }), ...lines.slice(from)] + .map(line => `${line}\n`).join('') + if (next === text) return lines.length + // The temp name is this write's own, and it takes the mode of the journal it replaces, so a + // file the owner narrowed by hand stays narrowed (#2697). + const temp = `${path}.${process.pid}.${randomUUID()}.tmp` + if (!tempWritable(temp)) { + if (!journalTrimRefused) { + journalTrimRefused = true + rt.warn('ds-bot: the memory journal was not trimmed: %s is not a plain file of its own', temp) + } + return lines.length + } + await replacePrivateFile(temp, path, next) + return lines.length - from + 1 + } async function journal(record) { if (!writeFolder(root)) return if (fileKindSync(journalPath) === 'other') { @@ -577,6 +638,15 @@ export function install(rt) { } await ensurePrivateDir(root) await appendFile(journalPath, `${JSON.stringify({ time: new Date().toISOString(), ...record })}\n`, { encoding: 'utf8', mode: PRIVATE_FILE_MODE }) + journalLines = journalLines === null ? await countJournalLines(journalPath) : journalLines + 1 + const { size } = await stat(journalPath) + if (journalLines <= JOURNAL_KEPT && size <= JOURNAL_MAX_BYTES) return + try { + journalLines = await trimJournal(journalPath) + } catch (error) { + // The journal is a side channel: a failed trim leaves the memory write itself alone. + rt.warn('ds-bot: the memory journal was not trimmed: %s', error) + } } let writing = Promise.resolve() const queued = (job) => { diff --git a/core/test/memory-journal-cap.test.js b/core/test/memory-journal-cap.test.js new file mode 100644 index 00000000..8227ca58 --- /dev/null +++ b/core/test/memory-journal-cap.test.js @@ -0,0 +1,95 @@ +import assert from 'node:assert/strict' +import { existsSync, mkdtempSync, readFileSync, rmSync, statSync, writeFileSync } from 'node:fs' +import { tmpdir } from 'node:os' +import { join } from 'node:path' +import { test } from 'node:test' +import { apply } from '../index.js' +import { fakeContext } from './fake-ctx.js' + +// The memory journal is a side channel: it keeps the newest changes, not every change a home +// ever made. The caps below are the ones `src/host/memory.js` writes down (`JOURNAL_KEPT` and +// `JOURNAL_MAX_BYTES`); the test states them itself so a silent widening of either cap fails +// here, the way the state journal's own cap test does. +const KEPT = 500 +const MAX_BYTES = 1_024 * 1_024 + +const TEAM = { + version: 1, + mainBotIds: ['m1'], + revision: 1, + whaleStored: true, + bots: { + m1: { id: 'm1', sessionId: 'm1', name: 'Chief', role: 'Chief of Staff', model: 'deepseek/v4', instructions: '', createdAt: 1 }, + b2: { id: 'b2', sessionId: 'b2', name: 'Writer', role: '写作', model: 'deepseek/v4', instructions: '', createdAt: 2 }, + }, + rooms: {}, + exchanges: [], + exchangeTurns: {}, + questions: {}, +} + +const ownChat = [ + { event: { seq: 10, type: 'user/message', data: { source: { kind: 'user' }, content: [{ type: 'text', text: 'Please remember this' }] } } }, + { event: { seq: 11, type: 'assistant/message', data: { message: { content: [{ type: 'text', text: 'Saved.' }] } } } }, +] + +// A real host (`apply()`), with only the services the memory tools ask for. +function start({ home = mkdtempSync(join(tmpdir(), 'ds-bot-memory-journal-')) } = {}) { + if (!existsSync(join(home, 'state.json'))) writeFileSync(join(home, 'state.json'), JSON.stringify(TEAM)) + const ctx = fakeContext({}) + Object.assign(ctx, { + llm: { + listProviders: () => [{ id: 'deepseek', name: 'DeepSeek' }], + listModels: async () => [{ id: 'v4' }], + resolveModelInfo: async () => ({ reasoning: { efforts: [{ id: 'off' }, { id: 'high' }] } }), + async* stream() { + yield { type: 'text-delta', index: 0, text: '' } + yield { type: 'finish', reason: { kind: 'stop' } } + }, + }, + sessionController: { + resolveAgent: async sessionId => ({ agent: { session: { id: sessionId, seq: ownChat.at(-1).event.seq + 1 } } }), + selectModel: async () => {}, + page: async () => ({ records: ownChat, hasMore: false }), + }, + sessions: { messageProjections: [] }, + workspaceRegistry: { archiveSession: async () => {} }, + }) + apply(ctx, { home }) + const tool = (name, args, sessionId) => ctx.registeredTools.get(name).execute(args, { agent: { id: sessionId } }) + const journalPath = join(home, 'memory', 'journal.jsonl') + const read = () => readFileSync(journalPath, 'utf8') + const lines = () => read().split('\n').filter(Boolean) + const stop = async () => { + await ctx.dispose() + rmSync(home, { recursive: true, force: true, maxRetries: 5 }) + } + return { tool, journalPath, read, lines, stop } +} + +test('the memory journal keeps its newest lines, like the state journal', async (t) => { + const { tool, journalPath, read, lines, stop } = start() + t.after(() => stop()) + + // 300 turns of saving one fact and forgetting it again: the memory itself is back to its + // empty skeleton at the end, while the journal would be 600 lines if nothing ever pruned it. + for (let i = 0; i < 300; i++) { + await tool('remember', { text: `Note number ${i}` }, 'b2') + await tool('forget', { entry: `Note number ${i}` }, 'b2') + } + + const size = statSync(journalPath).size + const kept = lines() + assert.ok(size <= MAX_BYTES, `the journal stays at or under ${MAX_BYTES} bytes (it is ${size})`) + assert.ok(kept.length <= KEPT, `the journal keeps at most ${KEPT} lines (it keeps ${kept.length})`) + assert.equal(read().includes('Note number 0'), false, 'the oldest changes are folded away, not kept') + const first = JSON.parse(kept[0]) + assert.equal(first.op, 'truncated', 'the folded changes are named by one record at the top') + assert.ok(first.dropped > 0, `the folded record counts what it replaced (got ${first.dropped})`) + + // Trimming keeps the newest side: a change made after the prune is on disk. + await tool('remember', { text: 'Note number last' }, 'b2') + const after = lines() + assert.ok(after.length <= KEPT, `the journal still keeps at most ${KEPT} lines (it keeps ${after.length})`) + assert.match(after.at(-1), /Note number last/, 'the newest change is still recorded after a prune') +}) diff --git a/patches/ds-bot/0001-core-fixes.patch b/patches/ds-bot/0001-core-fixes.patch index fc939f53..00eff591 100644 --- a/patches/ds-bot/0001-core-fixes.patch +++ b/patches/ds-bot/0001-core-fixes.patch @@ -6090,15 +6090,17 @@ diff --git a/src/host/images.js b/src/host/images.js diff --git a/src/host/memory.js b/src/host/memory.js --- a/src/host/memory.js +++ b/src/host/memory.js -@@ -21,13 +21,20 @@ +@@ -20,14 +20,21 @@ + // memory section the first time it renders, so its prompt stays byte-stable and the // provider's prefix cache holds; a fresh part, or a checkpoint (which breaks the cache // anyway), freezes it anew. - import { createHash } from 'node:crypto' +-import { createHash } from 'node:crypto' -import { existsSync, mkdirSync, readFileSync, readdirSync, renameSync, unlinkSync, writeFileSync } from 'node:fs' -import { appendFile, mkdir, rename, unlink, writeFile } from 'node:fs/promises' -import { dirname, join } from 'node:path' ++import { createHash, randomUUID } from 'node:crypto' +import { existsSync, readFileSync, readdirSync, unlinkSync } from 'node:fs' -+import { appendFile, rename, unlink } from 'node:fs/promises' ++import { appendFile, readFile, rename, stat, unlink } from 'node:fs/promises' +import { dirname, join, relative, sep } from 'node:path' import { BlockAssembler } from '@deepseek-ai/dsh-llm' -import { defineTool } from '@deepseek-ai/dsh-tools' @@ -6126,6 +6128,20 @@ diff --git a/src/host/memory.js b/src/host/memory.js // A change whose entry is gone; any other failure keeps the review's place. const NO_MATCH = 'memory/no-match' const GONE = 'memory/bot-gone' +@@ -70,6 +70,13 @@ const REVIEW_MEMORY_MAX = 16_000 + // its lines leave the group log (which keeps 60). + const REVIEW_GROUP_LAG = 30 + const NEWS_TEXT_MAX = 160 ++// The memory journal is a side channel, not history to keep forever (#2807): it holds the newest ++// changes under a line cap and a byte cap, and the changes past them are folded into one ++// `{ op: 'truncated', dropped }` line. A trim keeps most of the room, so once the file reaches a ++// cap it is rewritten every few changes rather than on every one. ++export const JOURNAL_KEPT = 500 ++export const JOURNAL_MAX_BYTES = 1_024 * 1_024 ++const JOURNAL_KEEP_TO = 0.9 + + // --------------------------------------------------------------------------- + // Entries and files, as text. @@ -98,9 +98,11 @@ export function cleanEntry(raw) { return splitMeta(line).text } @@ -6257,7 +6273,7 @@ diff --git a/src/host/memory.js b/src/host/memory.js const readText = (path) => { try { return readFileSync(path, 'utf8') -@@ -433,34 +433,68 @@ export function install(rt) { +@@ -433,34 +433,131 @@ export function install(rt) { return '' } } @@ -6322,6 +6338,60 @@ diff --git a/src/host/memory.js b/src/host/memory.js + throw new Error(`Not saved: ${temp} is not a plain file of its own, so nothing is written there.`) + } + await replacePrivateFile(temp, path, text) ++ } ++ // The journal keeps its newest changes and folds the rest away, the way the state journal does ++ // (#2807): at most `JOURNAL_KEPT` lines and `JOURNAL_MAX_BYTES` bytes. The fold is one ++ // `truncated` line carrying how many changes it replaced; it keeps an earlier fold's count, so ++ // the tally survives repeated trims. The rewrite goes through a temp name, so a crash never ++ // leaves half a line. ++ let journalLines = null ++ let journalTrimRefused = false ++ const countJournalLines = async (path) => { ++ try { ++ return (await readFile(path, 'utf8')).split('\n').filter(Boolean).length ++ } catch (error) { ++ if (error?.code === 'ENOENT') return 0 ++ throw error ++ } ++ } ++ async function trimJournal(path) { ++ let text ++ try { ++ text = await readFile(path, 'utf8') ++ } catch (error) { ++ if (error?.code === 'ENOENT') return 0 ++ throw error ++ } ++ const lines = text.split('\n').filter(Boolean) ++ let dropped = 0 ++ try { ++ const first = JSON.parse(lines[0]) ++ if (first?.op === 'truncated' && Number.isSafeInteger(first.dropped) && first.dropped >= 0) dropped = first.dropped ++ } catch { /* the oldest line is folded with the rest */ } ++ const keepLines = Math.floor(JOURNAL_KEPT * JOURNAL_KEEP_TO) ++ const keepBytes = Math.floor(JOURNAL_MAX_BYTES * JOURNAL_KEEP_TO) ++ let from = 0 ++ let bytes = Buffer.byteLength(text, 'utf8') ++ while (from < lines.length - 1 && (lines.length - from > keepLines || bytes > keepBytes)) { ++ bytes -= Buffer.byteLength(lines[from], 'utf8') + 1 ++ from += 1 ++ } ++ dropped += from ++ const next = [JSON.stringify({ time: new Date().toISOString(), op: 'truncated', dropped }), ...lines.slice(from)] ++ .map(line => `${line}\n`).join('') ++ if (next === text) return lines.length ++ // The temp name is this write's own, and it takes the mode of the journal it replaces, so a ++ // file the owner narrowed by hand stays narrowed (#2697). ++ const temp = `${path}.${process.pid}.${randomUUID()}.tmp` ++ if (!tempWritable(temp)) { ++ if (!journalTrimRefused) { ++ journalTrimRefused = true ++ rt.warn('ds-bot: the memory journal was not trimmed: %s is not a plain file of its own', temp) ++ } ++ return lines.length ++ } ++ await replacePrivateFile(temp, path, next) ++ return lines.length - from + 1 } async function journal(record) { - await mkdir(root, { recursive: true }) @@ -6333,6 +6403,15 @@ diff --git a/src/host/memory.js b/src/host/memory.js + } + await ensurePrivateDir(root) + await appendFile(journalPath, `${JSON.stringify({ time: new Date().toISOString(), ...record })}\n`, { encoding: 'utf8', mode: PRIVATE_FILE_MODE }) ++ journalLines = journalLines === null ? await countJournalLines(journalPath) : journalLines + 1 ++ const { size } = await stat(journalPath) ++ if (journalLines <= JOURNAL_KEPT && size <= JOURNAL_MAX_BYTES) return ++ try { ++ journalLines = await trimJournal(journalPath) ++ } catch (error) { ++ // The journal is a side channel: a failed trim leaves the memory write itself alone. ++ rt.warn('ds-bot: the memory journal was not trimmed: %s', error) ++ } } let writing = Promise.resolve() const queued = (job) => { diff --git a/patches/ds-bot/files/test/memory-journal-cap.test.js b/patches/ds-bot/files/test/memory-journal-cap.test.js new file mode 100644 index 00000000..8227ca58 --- /dev/null +++ b/patches/ds-bot/files/test/memory-journal-cap.test.js @@ -0,0 +1,95 @@ +import assert from 'node:assert/strict' +import { existsSync, mkdtempSync, readFileSync, rmSync, statSync, writeFileSync } from 'node:fs' +import { tmpdir } from 'node:os' +import { join } from 'node:path' +import { test } from 'node:test' +import { apply } from '../index.js' +import { fakeContext } from './fake-ctx.js' + +// The memory journal is a side channel: it keeps the newest changes, not every change a home +// ever made. The caps below are the ones `src/host/memory.js` writes down (`JOURNAL_KEPT` and +// `JOURNAL_MAX_BYTES`); the test states them itself so a silent widening of either cap fails +// here, the way the state journal's own cap test does. +const KEPT = 500 +const MAX_BYTES = 1_024 * 1_024 + +const TEAM = { + version: 1, + mainBotIds: ['m1'], + revision: 1, + whaleStored: true, + bots: { + m1: { id: 'm1', sessionId: 'm1', name: 'Chief', role: 'Chief of Staff', model: 'deepseek/v4', instructions: '', createdAt: 1 }, + b2: { id: 'b2', sessionId: 'b2', name: 'Writer', role: '写作', model: 'deepseek/v4', instructions: '', createdAt: 2 }, + }, + rooms: {}, + exchanges: [], + exchangeTurns: {}, + questions: {}, +} + +const ownChat = [ + { event: { seq: 10, type: 'user/message', data: { source: { kind: 'user' }, content: [{ type: 'text', text: 'Please remember this' }] } } }, + { event: { seq: 11, type: 'assistant/message', data: { message: { content: [{ type: 'text', text: 'Saved.' }] } } } }, +] + +// A real host (`apply()`), with only the services the memory tools ask for. +function start({ home = mkdtempSync(join(tmpdir(), 'ds-bot-memory-journal-')) } = {}) { + if (!existsSync(join(home, 'state.json'))) writeFileSync(join(home, 'state.json'), JSON.stringify(TEAM)) + const ctx = fakeContext({}) + Object.assign(ctx, { + llm: { + listProviders: () => [{ id: 'deepseek', name: 'DeepSeek' }], + listModels: async () => [{ id: 'v4' }], + resolveModelInfo: async () => ({ reasoning: { efforts: [{ id: 'off' }, { id: 'high' }] } }), + async* stream() { + yield { type: 'text-delta', index: 0, text: '' } + yield { type: 'finish', reason: { kind: 'stop' } } + }, + }, + sessionController: { + resolveAgent: async sessionId => ({ agent: { session: { id: sessionId, seq: ownChat.at(-1).event.seq + 1 } } }), + selectModel: async () => {}, + page: async () => ({ records: ownChat, hasMore: false }), + }, + sessions: { messageProjections: [] }, + workspaceRegistry: { archiveSession: async () => {} }, + }) + apply(ctx, { home }) + const tool = (name, args, sessionId) => ctx.registeredTools.get(name).execute(args, { agent: { id: sessionId } }) + const journalPath = join(home, 'memory', 'journal.jsonl') + const read = () => readFileSync(journalPath, 'utf8') + const lines = () => read().split('\n').filter(Boolean) + const stop = async () => { + await ctx.dispose() + rmSync(home, { recursive: true, force: true, maxRetries: 5 }) + } + return { tool, journalPath, read, lines, stop } +} + +test('the memory journal keeps its newest lines, like the state journal', async (t) => { + const { tool, journalPath, read, lines, stop } = start() + t.after(() => stop()) + + // 300 turns of saving one fact and forgetting it again: the memory itself is back to its + // empty skeleton at the end, while the journal would be 600 lines if nothing ever pruned it. + for (let i = 0; i < 300; i++) { + await tool('remember', { text: `Note number ${i}` }, 'b2') + await tool('forget', { entry: `Note number ${i}` }, 'b2') + } + + const size = statSync(journalPath).size + const kept = lines() + assert.ok(size <= MAX_BYTES, `the journal stays at or under ${MAX_BYTES} bytes (it is ${size})`) + assert.ok(kept.length <= KEPT, `the journal keeps at most ${KEPT} lines (it keeps ${kept.length})`) + assert.equal(read().includes('Note number 0'), false, 'the oldest changes are folded away, not kept') + const first = JSON.parse(kept[0]) + assert.equal(first.op, 'truncated', 'the folded changes are named by one record at the top') + assert.ok(first.dropped > 0, `the folded record counts what it replaced (got ${first.dropped})`) + + // Trimming keeps the newest side: a change made after the prune is on disk. + await tool('remember', { text: 'Note number last' }, 'b2') + const after = lines() + assert.ok(after.length <= KEPT, `the journal still keeps at most ${KEPT} lines (it keeps ${after.length})`) + assert.match(after.at(-1), /Note number last/, 'the newest change is still recorded after a prune') +})