/** * dream —— 经验汇流([29-DREAM](../../docs/todo/system-design/29-dream-design.md))。 * * 定时把**所有 agent 在所有公司**的原始记录提炼成不含公司特定信息的经验, * 沉淀到本体上,让经验跨公司汇流而不泄漏任何一家的具体信息。 * * ── 四条拍板(2026-08-20,§三)────────────────────────────────────── * A 去敏 = 提炼 agent 按写作要求产出 + **独立的判定 agent** 过闸门。 * 失效方向是「少沉淀」不是「泄漏」—— 两种错代价不对称时判据偏向贵的那侧。 * B 落点 = 本体根 `memory/`(跨公司那一份)与 `SKILL.md` 都写; * `soul.md` / `agent.md` 直接落盘,但写前必须过 definition-write-gate。 * C 触发 = 定时每日一次。 * D 顺序 = 上游记忆写入与 dream 一起做。 * * ── 两条贯穿全程的判据 ────────────────────────────────────────────── * **① 「有哪些公司」只认 `company_agents` 表,不扫目录。** * `system/agents/{aid}/{cid}/` 是约定不是校验,实查 96 个二级目录里混着 * `train/` `workspace/` 和带时间戳的测试残留(`tmp-w7-*-1786986291739-*`)。 * 拿目录枚举公司 = 把它们当原料。目录只用来**定位文件**,不用来**枚举公司**。 * * **② 空转不许起 agent 调用。** * 实查 75 份 memory.md 里只有 7 份有非模板内容 —— 每日跑起来绝大多数 agent 没有新原料。 * 不拦的话每天几十次「读完发现没东西」的 LLM 调用在烧钱,日志里全是噪声。 * 判据是水位线(`dream.json` 的 `lastRunAt`)与原料的 mtime 比对。 */ const fs = require('fs'); const path = require('path'); const locator = require('./agent-locator'); const { canOverwriteDefinition } = require('./definition-write-gate'); const RAW_SECTION = '## 原始记录(未提炼)'; /** 本体根上的跨公司经验落点(决策 B) */ const CROSS_COMPANY_MEMORY = 'memory'; const STATE_FILE = 'dream.json'; // ── 注入缝:两次 agent 调用都可替换(写法照抄 graph-engine.setAgentExecutor)── let distiller = null; let screener = null; /** 替换「提炼」实现(传 null 回默认实现) */ function setDistiller(fn) { distiller = fn; } /** 替换「去敏判定」实现(传 null 回默认实现) */ function setScreener(fn) { screener = fn; } /** * 这个本体在哪些公司有过运行态?**答案来自 `company_agents` 表**(判据①)。 * * `_global` 单独补上:它是保留段,全局 agent 的运行态落在那里, * 不在 company_agents 里(那张表记的是「公司雇了谁」,全局 agent 没人雇)。 * 算不算「一家公司」在设计里是 ❓,这里的处理是**算原料、但在提炼时标明它是全局态**。 */ function companiesOf(agentId) { const out = []; try { const { listCompanyAgents } = require('./company-service'); for (const row of listCompanyAgents()) { if (row.system_agent_id === agentId || row.role_name === agentId) out.push(row.company_id); } } catch (err) { // 库还没建好不该让 dream 整个挂掉,但**必须让调用方知道这一轮的原料是不全的** console.warn(`[Dream] ⚠️ 读 company_agents 失败,本轮只看 _global: ${err.message}`); } if (fs.existsSync(path.join(locator.bodyDir(agentId), '_global'))) out.push('_global'); return [...new Set(out)]; } /** 一条原料:正文 + 它从哪来(决策 D 的「可追溯」起步) */ function readRawEntries(agentId, companyId) { const md = path.join(locator.companyDir(agentId, companyId), 'memory', 'memory.md'); let text; try { text = fs.readFileSync(md, 'utf-8'); } catch { return []; } const idx = text.indexOf(RAW_SECTION); if (idx < 0) return []; const after = text.slice(idx + RAW_SECTION.length); const section = after.split(/\n## /)[0]; return section .split('\n') .map((l) => l.trim()) .filter((l) => l.startsWith('- ')) .map((l, i) => ({ agentId, companyId, source: `${companyId}:memory.md#${i + 1}`, text: l.slice(2).trim() })); } /** 本体根的 dream 状态(水位线 + 上一轮产出) */ function statePath(agentId) { return path.join(locator.bodyDir(agentId), CROSS_COMPANY_MEMORY, STATE_FILE); } function readState(agentId) { try { return JSON.parse(fs.readFileSync(statePath(agentId), 'utf-8')); } catch { return {}; } } function writeState(agentId, state) { const p = statePath(agentId); fs.mkdirSync(path.dirname(p), { recursive: true }); fs.writeFileSync(p, `${JSON.stringify(state, null, 2)}\n`, 'utf-8'); } /** * 采集这个本体的全部新原料。 * * **水位线按「条数」而不是 mtime**:mtime 会被任何一次无关写入(例如另一段代码 * 重写整份 memory.md)推动,而条数只在真的多出记录时才增长。 * 拿 mtime 当判据的话,空转保护会在「文件被碰过但内容没变」时失效。 */ function collectMaterial(agentId) { const state = readState(agentId); const seen = state.seenCounts || {}; const fresh = []; const counts = {}; for (const cid of companiesOf(agentId)) { const entries = readRawEntries(agentId, cid); counts[cid] = entries.length; const from = seen[cid] || 0; if (entries.length > from) fresh.push(...entries.slice(from)); } return { fresh, counts }; } /** * 跑一轮 dream。 * * @param {string} agentId 本体 id * @param {{dryRun?: boolean}} opts * @returns {Promise} 这一轮做了什么(含 skipped 的原因,别让调用方猜) */ async function dreamOnce(agentId, { dryRun = false } = {}) { const { fresh, counts } = collectMaterial(agentId); // 判据②:没有新原料就**一次 agent 调用都不起** if (fresh.length === 0) { return { agentId, skipped: 'no-new-material', counts, distilled: 0, accepted: 0 }; } const candidates = await (distiller || defaultDistiller)({ agentId, entries: fresh }); const accepted = []; const rejected = []; // 决策 D 的可做部分:**「结论对不对」机器判不了,但「引用是不是编的」可以。** // 提炼产物的 sources 必须都出自这一轮真喂进去的原料 —— 编一个出处是 // 机器可检的谎,而它恰恰是「无法验证」这件事最危险的形态: // 一条编了出处的经验,人抽查时会顺着一个不存在的指针放弃,然后默认它是真的。 const knownSources = new Set(fresh.map((e) => e.source)); for (const c of candidates || []) { const cited = Array.isArray(c.sources) ? c.sources : []; const dangling = cited.filter((x) => !knownSources.has(x)); if (!cited.length || dangling.length) { rejected.push({ ...c, reason: cited.length ? `引用了本轮不存在的出处:${dangling.join('、')}(可追溯是唯一的质量抓手,编出处等于把它拆掉)` : '没有给出 sources —— 无法回查,按拒绝处理', }); continue; } // 决策 A:**独立**的判定,不与提炼共享上下文;判不准 = 拒绝(fail-closed) let verdict; try { verdict = await (screener || defaultScreener)({ agentId, candidate: c }); } catch (err) { rejected.push({ ...c, reason: `判定失败,按拒绝处理:${err.message}` }); continue; } if (verdict && verdict.shareable === true) accepted.push(c); else rejected.push({ ...c, reason: (verdict && verdict.reason) || '判定未通过' }); } let definitions = []; if (accepted.length) { if (!dryRun) writeDistilled(agentId, accepted); // 决策 B:三个 git 跟踪的定义文件也写,但每一份都要过闸门(见 writeDefinitionFiles) definitions = writeDefinitionFiles(agentId, accepted, { dryRun }); } if (!dryRun) { writeState(agentId, { ...readState(agentId), seenCounts: counts, lastRunAt: new Date().toISOString(), lastAccepted: accepted.length, lastRejected: rejected.length, }); } return { agentId, skipped: null, counts, distilled: (candidates || []).length, accepted: accepted.length, rejected, definitions, // 逐个文件的「写了没有、没写是为什么」—— 不写不该是静默的 }; } /** * 抽查用:把已沉淀的每条经验和它引用的**原料原文**并排列出来。 * * 决策 D(提炼质量怎么评)目前没有自动判据,也不该假装有 —— 「这条结论对不对」 * 需要理解内容。能自动做的只有两件事,两件都做了: * · 写入时校验 `sources` 都出自本轮真喂进去的原料(编出处直接拒,见 dreamOnce); * · 这里把结论与原文并排摆出来,让人抽查的成本降到几秒。 * * **不要把这个函数的输出当成「质量分」**。它只是把回查这件事变便宜了。 */ function auditDistilled(agentId) { const file = path.join(locator.bodyDir(agentId), CROSS_COMPANY_MEMORY, 'experience.md'); let text; try { text = fs.readFileSync(file, 'utf-8'); } catch { return []; } // 原料按 source 建索引(跨公司都要,因为一条经验可能引了不同公司的记录) const index = new Map(); for (const cid of companiesOf(agentId)) { for (const e of readRawEntries(agentId, cid)) index.set(e.source, e.text); } const out = []; for (const block of text.split(/\n### /).slice(1)) { const title = block.split('\n')[0].trim(); const m = /原料:(.+)$/m.exec(block); const cited = m ? m[1].split(/[、,]/).map((x) => x.trim()).filter(Boolean) : []; out.push({ title, sources: cited.map((src) => ({ source: src, text: index.has(src) ? index.get(src) : null, // null = 指针悬空(原料被清理过 / 出处是编的) })), }); } return out; } /** * 落盘(决策 B)。 * * 只写**本体根的 `memory/`**:那是纯经验积累,删掉即可,无副作用。 * `SKILL.md` / `soul.md` / `agent.md` 是 git 跟踪的**定义**, * 每一份都必须先过 `definition-write-gate` —— 覆盖一份尚未进入 git 历史的内容 * 等于让上一版永久消失,且不报错。这里只做**闸门检查 + 记录**, * 真正改定义留给后续步骤(本轮不写定义文件,见 §五 实现记录)。 * * **每条沉淀都带 sources**(决策 D):没有它,dream 的产出无法验证对错。 */ function writeDistilled(agentId, accepted) { const dir = path.join(locator.bodyDir(agentId), CROSS_COMPANY_MEMORY); fs.mkdirSync(dir, { recursive: true }); const file = path.join(dir, 'experience.md'); const now = new Date().toISOString().substring(0, 10); const blocks = accepted.map((c) => [ `### ${c.title || '(未命名)'}`, '', c.body || '', '', `> 提炼于 ${now};原料:${(c.sources || []).join('、') || '(未记录)'}`, '', ].join('\n')); const head = fs.existsSync(file) ? fs.readFileSync(file, 'utf-8').replace(/\s*$/, '') : '# 跨公司经验(dream 提炼)'; fs.writeFileSync(file, `${head}\n\n${blocks.join('\n')}`, 'utf-8'); return file; } /** * 把提炼产物写进**定义文件**(决策 B:`SKILL.md` / `soul.md` / `agent.md` 直接落盘)。 * * ── 形状:一个**受管区块**,不是自由追加 ────────────────────────────── * dream 只动一对标记之间的内容: * * … * * 区块外的一个字节都不碰。理由是这三个文件是**人和 agent 共同维护**的定义, * 自由追加会让「哪些是人写的、哪些是 dream 写的」在几轮之后再也分不开, * 而分不开就没法分别回滚(决策 B.1 的四条硬约束之一)。 * 区块整体重写还带来一个额外好处:**幂等** —— 同一批经验重复跑不会越写越长。 * * ── 三道闸门,缺一不写 ──────────────────────────────────────────────── * ① `definition-write-gate`:这份文件的当前内容必须已经在 git 历史里 * (拍板要求:别导致直接修改而没有 git 历史)。 * ② **`agent.md` 的前 2000 字符必须仍然自足**:给规划 agent 看的「角色定义摘录」 * 只取前 2000 字符(`graph-engine.js:915`)。dream 的区块因此**只准放在文件末尾**, * 且写完要复核前 2000 字符没被顶掉。 * ③ 写完之后毕业闸门仍要过(`missingBodyDefinition` 为空)—— * 把 `agent.md` 写坏会让本体退回「定义没填满」,**雇不进任何公司**。 * * 任何一道不过:**不写,并把原因返回**。少沉淀一条经验 vs 改坏一个所有公司共用的本体, * 两者代价不对称。 */ const DREAM_BEGIN = ''; const DREAM_END = ''; const AGENT_MD_PREFIX_BUDGET = 2000; function renderDreamBlock(accepted, { heading }) { const now = new Date().toISOString().substring(0, 10); return [ DREAM_BEGIN, `## ${heading}`, '', `_由 dream 于 ${now} 提炼自跨公司经验;本区块由机器维护,区块外的内容不会被改动。_`, '', ...accepted.map((c) => `- **${c.title}** —— ${String(c.body || '').replace(/\n+/g, ' ')}`), '', DREAM_END, ].join('\n'); } /** 把区块塞进(或替换掉)正文;区块外一个字节不动 */ function upsertBlock(text, block) { const b = text.indexOf(DREAM_BEGIN); const e = text.indexOf(DREAM_END); if (b >= 0 && e > b) return `${text.slice(0, b)}${block}${text.slice(e + DREAM_END.length)}`; return `${text.replace(/\s*$/, '')}\n\n${block}\n`; } /** * @param {string} agentId * @param {Array} accepted 已过去敏闸门的产物 * @param {{files?: string[], dryRun?: boolean}} opts * @returns {Array<{file, written, reason}>} 逐个文件报「写了没有、没写是为什么」 */ function writeDefinitionFiles(agentId, accepted, { files = ['SKILL.md', 'soul.md', 'agent.md'], dryRun = false } = {}) { const body = locator.bodyDir(agentId); const HEADINGS = { 'SKILL.md': '跨公司经验(dream)', 'soul.md': '从经验里长出来的倾向(dream)', 'agent.md': '跨公司经验提醒(dream)', }; const out = []; if (!accepted.length) return files.map((f) => ({ file: f, written: false, reason: '本轮没有通过闸门的产物' })); for (const f of files) { const abs = path.join(body, f); const gate = canOverwriteDefinition(abs); if (!gate.allowed) { out.push({ file: f, written: false, reason: gate.reason }); continue; } let text; try { text = fs.readFileSync(abs, 'utf-8'); } catch (err) { out.push({ file: f, written: false, reason: `读不到:${err.message}` }); continue; } const next = upsertBlock(text, renderDreamBlock(accepted, { heading: HEADINGS[f] || '跨公司经验(dream)' })); // 闸门②:agent.md 的前 2000 字符不许被顶掉 if (f === 'agent.md' && next.slice(0, AGENT_MD_PREFIX_BUDGET) !== text.slice(0, AGENT_MD_PREFIX_BUDGET)) { out.push({ file: f, written: false, reason: `写入会改变 agent.md 的前 ${AGENT_MD_PREFIX_BUDGET} 字符 —— ` + '规划 agent 只看这一段(graph-engine 的「角色定义摘录」),改了等于悄悄换掉它读到的画像', }); continue; } if (!dryRun) { fs.writeFileSync(abs, next, 'utf-8'); // 闸门③:写完复核毕业闸门,不过就**立刻还原** const missing = locator.missingBodyDefinition(agentId); if (missing.length) { fs.writeFileSync(abs, text, 'utf-8'); out.push({ file: f, written: false, reason: `写完毕业闸门不过(缺 ${missing.join(' / ')}),已还原` }); continue; } } out.push({ file: f, written: true, reason: dryRun ? '(dry-run)' : `已更新 ${gate.rel} 的 dream 区块` }); } return out; } /** 定义文件能不能写 —— 给后续步骤用,也给控制台显示「为什么没改」 */ function definitionWritability(agentId) { const body = locator.bodyDir(agentId); return ['agent.md', 'soul.md', 'SKILL.md'].map((f) => ({ file: f, ...canOverwriteDefinition(path.join(body, f)), })); } // ── 默认实现:两次**互相不共享上下文**的真 agent 调用(决策 A)────────── const DISTILLER_AGENT = 'dream-distiller'; const SCREENER_AGENT = 'dream-screener'; /** * 从 agent 回复里取 JSON。 * * 写法与 `graph-engine.extractPlannedGraph` 同源:```json 围栏块优先, * 其次从第一个 `{`/`[` 起做括号配平。**取不出来一律返回 null,不猜** —— * 猜错的代价是把一段自然语言当成结构化产物落库。 */ function extractJson(text, expect) { const src = String(text || ''); const tryParse = (raw) => { try { const v = JSON.parse(raw); if (expect === 'array' && !Array.isArray(v)) return null; if (expect === 'object' && (Array.isArray(v) || typeof v !== 'object' || v === null)) return null; return v; } catch { return null; } }; for (const m of src.matchAll(/```(?:json|jsonc)?\s*([\s\S]*?)```/g)) { const v = tryParse(m[1]); if (v !== null) return v; } const open = expect === 'array' ? '[' : '{'; const close = expect === 'array' ? ']' : '}'; let depth = 0; let start = -1; for (let i = 0; i < src.length; i++) { if (src[i] === open) { if (depth === 0) start = i; depth++; } else if (src[i] === close) { depth--; if (depth === 0 && start >= 0) { const v = tryParse(src.slice(start, i + 1)); if (v !== null) return v; start = -1; } } } return null; } /** * 起一次 agent 调用。dream 的两个 agent 都是全局的,用裸 id(落 `_global`)。 * * 单独抽出来并留一个替换点,是为了让**默认实现本身**能被测到: * 不留的话,测试只能在自己那边重抄一遍解析与失效分支 —— 那是在验一份复制品, * 代码改了它照样绿。这条缝换掉的只有「真起 CLI」那一层,解析、判空、失效方向全走真代码。 */ let agentCaller = null; /** 替换 dream 的 agent 调用层(传 null 回真实现)。只给测试用。 */ function setDreamAgentCaller(fn) { agentCaller = fn; } async function callDreamAgent(agentName, prompt) { if (agentCaller) return agentCaller(agentName, prompt); return realCallDreamAgent(agentName, prompt); } async function realCallDreamAgent(agentName, prompt) { const AgentManager = require('../src/core/agent-manager'); const manager = new AgentManager({}, path.resolve(__dirname, '../..')); const res = await manager.callAgent(agentName, prompt, { metadata: { layer: 'dream', agent: agentName }, }); return (res && res.output) || ''; } /** * 提炼:一次调用吃掉这一轮的**全部**新原料。 * * 为什么不逐条提炼:一条经验常常要**跨多条记录**才看得出来 * (「三次都是因为验收标准自证才返工」)。逐条喂等于把这种归纳能力关掉。 */ async function defaultDistiller({ agentId, entries }) { const lines = entries.map((e) => `[出处 ${e.source}] ${e.text}`).join('\n'); const prompt = [ `下面是本体 \`${agentId}\` 在各家公司留下的原始记录,共 ${entries.length} 条。`, '按你 agent.md 里的产出契约提炼成可复用经验,交一个 ```json 数组。', '没有值得留下的就交 []。', '', lines, ].join('\n'); const out = await callDreamAgent(DISTILLER_AGENT, prompt); const arr = extractJson(out, 'array'); if (arr === null) { // **不静默返回空**:解析不出来和「没提炼出东西」是两件事,混为一谈会让 // 「产出格式坏了」这种故障以「今天没经验可沉淀」的形式永远藏下去。 throw new Error(`[Dream] ${DISTILLER_AGENT} 的回复里取不出 JSON 数组(前 200 字:${out.slice(0, 200)})`); } // 形状不对的条目直接丢,但**必须留下痕迹** —— 无声丢弃会让人以为 agent 没产出 return arr.filter((c) => { const ok = c && typeof c.title === 'string' && typeof c.body === 'string'; if (!ok) console.warn(`[Dream] ⚠️ 丢弃形状不对的提炼产物: ${JSON.stringify(c).slice(0, 120)}`); return ok; }); } /** * 去敏判定:**一条一次调用**,且只喂 title + body。 * * 逐条、且不给原料,是决策 A 的核心 —— 判定方的处境必须与 * 「一个完全不相干的第三方读到这段文字」完全一样。 * 把整批一起喂过去,它会从别的条目里获得上下文,那就不是独立判定了。 */ async function defaultScreener({ candidate }) { const prompt = [ '判断下面这一条提炼产物能不能给别家公司看。按你 agent.md 的契约交一个 ```json 对象。', '', `title: ${candidate.title}`, `body: ${candidate.body}`, ].join('\n'); const out = await callDreamAgent(SCREENER_AGENT, prompt); const obj = extractJson(out, 'object'); if (obj === null || typeof obj.shareable !== 'boolean') { // 判不出来 = 不放行(fail-closed)。这里**返回**而不是抛: // 抛也会被 dreamOnce 记成拒绝,但返回能把「为什么」原样带进 rejected[]。 return { shareable: false, reason: `判定回复无法解析成 {shareable:boolean}(前 200 字:${out.slice(0, 200)})` }; } return { shareable: obj.shareable === true, reason: obj.reason || '' }; } // ── 定时(决策 C:每日一次)──────────────────────────────────────────── const DAY_MS = 24 * 60 * 60 * 1000; let sweepTimer = null; /** * 有哪些本体值得跑一轮 dream? * * 判据是**本体目录存在 + 不是保留 id**,和 `heartbeat-service.start()` 扫的同一套。 * 不查 `system_agents` 表:那张表里可能有磁盘上已经没有的悬空行, * 而 dream 要读的是磁盘上的记忆文件 —— **以「读得到原料的那一侧」为准**。 */ function listDreamableAgents() { try { return fs.readdirSync(locator.AGENTS_ROOT, { withFileTypes: true }) .filter((d) => d.isDirectory() && !d.name.startsWith('_') && !d.name.startsWith('.')) .map((d) => d.name); } catch { return []; } } /** * 扫一遍所有本体,各跑一次 `dreamOnce`。 * * **串行**,不是并发:每一轮都可能起两次 LLM 调用,几十个本体并发等于同时拉起几十个 * claude 子进程 —— 本仓库为这个付过代价(2h43m 漏出 82 个进程 / 17.5 GB, * 记在 28-followups §五 第五条)。慢一点没有代价,dream 不赶时间。 * * 空转保护在 `dreamOnce` 里:没有新原料的本体连一次 agent 调用都不起, * 所以「76 个本体全扫一遍」的实际成本是 76 次目录读。 */ async function dreamSweep({ onResult, agents = null } = {}) { const results = []; for (const agentId of agents || listDreamableAgents()) { try { const r = await dreamOnce(agentId); results.push(r); if (onResult) onResult(r); } catch (err) { // 一个本体炸了不该拖垮整轮 —— 但**必须留下是谁炸的**, // 否则现场只看到「今天没沉淀」,查不出是没原料还是一直在抛 console.error(`[Dream] ⚠️ ${agentId} 本轮失败: ${err.message}`); results.push({ agentId, error: err.message }); } } const ran = results.filter((r) => !r.skipped && !r.error); console.log(`[Dream] 🌙 扫描完成:${results.length} 个本体,${ran.length} 个有新原料,` + `沉淀 ${ran.reduce((n, r) => n + (r.accepted || 0), 0)} 条`); return results; } /** * 挂上每日定时(决策 C)。 * * `unref()`:**绝不拖着进程不退出**。与 task-scheduler 的兜底定时器同一处理 —— * 少了它,`npm test` 之类的短命进程会挂在这个 24 小时的 timer 上不肯结束。 */ function startDaily({ intervalMs = DAY_MS } = {}) { if (sweepTimer) return sweepTimer; sweepTimer = setInterval(() => { dreamSweep().catch((err) => console.error(`[Dream] 定时扫描失败: ${err.message}`)); }, intervalMs); if (sweepTimer.unref) sweepTimer.unref(); console.log(`[Dream] 🌙 已挂载每日经验汇流(间隔 ${Math.round(intervalMs / 3600_000)} 小时)`); return sweepTimer; } function stopDaily() { if (sweepTimer) clearInterval(sweepTimer); sweepTimer = null; } module.exports = { dreamOnce, dreamSweep, startDaily, stopDaily, listDreamableAgents, collectMaterial, companiesOf, readRawEntries, defaultDistiller, defaultScreener, extractJson, DISTILLER_AGENT, SCREENER_AGENT, writeDistilled, writeDefinitionFiles, definitionWritability, auditDistilled, readState, writeState, DREAM_BEGIN, DREAM_END, setDistiller, setScreener, setDreamAgentCaller, RAW_SECTION, CROSS_COMPANY_MEMORY, };