Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions changelog.d/2807.fixed.md
Original file line number Diff line number Diff line change
@@ -0,0 +1 @@
The memory journal (`<home>/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)).
74 changes: 72 additions & 2 deletions core/src/host/memory.js
Original file line number Diff line number Diff line change
Expand Up @@ -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'
Expand Down Expand Up @@ -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.
Expand Down Expand Up @@ -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') {
Expand All @@ -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) => {
Expand Down
95 changes: 95 additions & 0 deletions core/test/memory-journal-cap.test.js
Original file line number Diff line number Diff line change
@@ -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')
})
87 changes: 83 additions & 4 deletions patches/ds-bot/0001-core-fixes.patch
Original file line number Diff line number Diff line change
Expand Up @@ -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'
Expand Down Expand Up @@ -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
}
Expand Down Expand Up @@ -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 ''
}
}
Expand Down Expand Up @@ -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 })
Expand All @@ -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) => {
Expand Down
Loading
Loading