feat: L0 线协议冻结 + 附件链路修复 + 人/Agent 区分
L0 核心： - 严格解码 Decode(DisallowUnknownFields) 全覆盖 29 个 DecodeBody 调用点 - DecodeLenient 心跳专用：容忍新字段但回报 unknown_fields - 400 消息列出本端点接受的全部字段（jsonFieldNames 反射 tag） - 日历 status 校验（create 补字段 + update 拦非法值） - 新增 strictdecode_test.go 10 例 + blob/list_test.go 6 例 A-4 附件挂载回滚：checkAttachable 在 CreateMail 前校验，失败按 解挂→释放 relay→删邮件→退预算回滚，幽灵邮件这条路堵住了 A-5 反向 GC：blob.Store.List() 枚举磁盘（跳 .upload-*）， SweepUnreferencedBlobs 按 attachments + calendar_attachments 反查， 48h 年龄下限兜上传窗口。已接进每小时 sweep 循环 C 人/Agent 区分：四个读路径 + threadCols 补 from_human / to_human （EXISTS users 判定），models.Mail 加 ToHuman。前端判据从 workspace 启发式改成显式布尔，mailCounterpart/sessionCounterpart 从 session_workspace 取 path（修 dsh@dsh 拼接 bug） 契约文档：SSE new_mail 补 4 字段（in_reply_to/from_human/ permission_mode/permission_enforcement），B-5 加 B-5.6 （Agent→Agent 不转发），B-3.4 MUST 改条件式，心跳补 mode_enforcement + unknown_fields，demo 死链修复 + from_human 检查 验收清单加 Agent→Agent 负向对照项
This commit is contained in:
208
plugins/pi-mail-bridge/lib/bounded.js
Normal file
208
plugins/pi-mail-bridge/lib/bounded.js
Normal file
@ -0,0 +1,208 @@
|
||||
/**
|
||||
* 有界容器 —— 给插件里那些「只增不减」的映射表兜底。
|
||||
*
|
||||
* # 为什么需要它
|
||||
*
|
||||
* 桥是**常驻进程**（pi 的守护进程能跑几十天，opencode/DSH 的插件跟着平台一起活）。
|
||||
* 里面每一张 `Map`/`Set` 都在回答「这条会话/这封邮件我处理过吗」，键来自外部
|
||||
* 事件流 —— 会话数与邮件数随时间单调增长，键却没有出口。
|
||||
*
|
||||
* 单条成本很小（uuid 键 + 短字符串值，几十到几百字节），所以它不是几小时内撑爆
|
||||
* 内存的那种故障。实际形态是：跑够久之后进程里躺着几十万个再也不会被查到的条目，
|
||||
* 且 **GC 回收不了**（还被强引用着）。这类问题不会在开发和测试里出现，
|
||||
* 只在生产上跑了几周后表现为「重启一下就好了」。
|
||||
*
|
||||
* # 淘汰策略：丢最久没被访问的
|
||||
*
|
||||
* JS 的 `Map`/`Set` 保证插入顺序，所以「删掉再插入」等价于「移到队尾」。
|
||||
* 读也算访问（`get`/`has` 会刷新顺序），于是长期活跃的会话不会因为条目老被丢掉 ——
|
||||
* 被淘汰的总是「很久没人问过」的那些。
|
||||
*
|
||||
* # 上限分表定义，因为丢一条的后果差别很大
|
||||
*
|
||||
* - `deliveredMails` 丢一条 → 那封邮件**理论上**可能被重复投递。但它防的两种
|
||||
* 重复（心跳与 SSE 建连之间的窗口、SSE 断线重放）都发生在秒到分钟级，
|
||||
* 几千封之前的 mail_id 不可能再来 —— 淘汰是安全的。
|
||||
*
|
||||
* - 会话级映射丢一条 → 那条会话下次来信时被当成新会话，平台侧上下文断掉。
|
||||
* 这是**真的行为退化**，所以上限给得大得多，并且优先靠 `session_archived`
|
||||
* 主动清理，让上限只当兜底。
|
||||
*
|
||||
* # 不要用它装「还在等结果的东西」
|
||||
*
|
||||
* 待决权限询问（opencode 的 `pendingPermissions`、DSH 的 `pendingApprovals`）
|
||||
* 里存的是 `resolve` 回调。静默淘汰一条会让对应的 `await` **永远不返回** ——
|
||||
* 平台侧那次工具调用就挂死了。那些表有确定的清理路径（决策到达 / 超时 / 拆插件
|
||||
* 时 fail closed），不该套上界。上界只适合「记录已经发生过的事实」的表。
|
||||
*
|
||||
* 三平台共用，必须逐字节相同（deploy/check-shared-libs.sh 校验）。
|
||||
*/
|
||||
|
||||
/**
|
||||
* 已投递邮件 id 的记忆上限。
|
||||
*
|
||||
* 2000 覆盖的是去重真正需要的时间窗：SSE 重放最多回放服务端环形缓冲的 500 条
|
||||
* 事件，一次补拉最多 5 封。留 2000 是三个数量级的余量，内存代价约 200KB。
|
||||
*/
|
||||
export const MAX_TRACKED_MAILS = 2000;
|
||||
|
||||
/**
|
||||
* 会话级映射的条目上限。
|
||||
*
|
||||
* 淘汰一条会让那条会话失去平台侧上下文，所以这个数字要远大于「同时在推进的
|
||||
* 任务数」。500 条 × 每条几百字节 ≈ 150KB —— 便宜到没有理由抠。
|
||||
*
|
||||
* 真正的清理来自 `session_archived`：会话归档后它的映射再无用处，那是确定性
|
||||
* 时机；上限兜的是「一直不归档」。
|
||||
*/
|
||||
export const MAX_TRACKED_SESSIONS = 500;
|
||||
|
||||
function normalizeLimit(limit) {
|
||||
const n = Number(limit);
|
||||
// 上限必须是正整数：0 会让每次 set 之后立刻把自己淘汰掉（表恒空，去重全部
|
||||
// 失效且不报错），NaN 会让 while 条件恒假（退化成无界）。两种都是静默的
|
||||
// 错误行为，不如当场拒绝。
|
||||
if (!Number.isFinite(n) || n < 1) {
|
||||
throw new RangeError(`有界容器的上限必须是 >= 1 的整数，收到 ${limit}`);
|
||||
}
|
||||
return Math.floor(n);
|
||||
}
|
||||
|
||||
/**
|
||||
* 有界 Map，超过上限时丢弃最久未访问的条目。
|
||||
*
|
||||
* 只实现桥里真正用到的那几个方法 —— 不做成 Map 的完整替身，那样会掩盖
|
||||
* 「这张表是有界的」这个必须被看见的事实。
|
||||
*/
|
||||
export class BoundedMap {
|
||||
/** @param {number} limit 条目上限 */
|
||||
constructor(limit) {
|
||||
this.limit = normalizeLimit(limit);
|
||||
/** @type {Map<any, any>} */
|
||||
this.map = new Map();
|
||||
/** 累计淘汰条数，观测用（日志里能看出上限是否设得太小）。 */
|
||||
this.evicted = 0;
|
||||
}
|
||||
|
||||
get size() {
|
||||
return this.map.size;
|
||||
}
|
||||
|
||||
has(key) {
|
||||
return this.map.has(key);
|
||||
}
|
||||
|
||||
/**
|
||||
* 取值并把该键移到队尾。
|
||||
*
|
||||
* 读也算访问：一条会话只要还在收信就会被反复 get，不刷新的话它会因为
|
||||
* 「插入得早」被淘汰 —— 那恰好淘汰了最该留的那些。
|
||||
*/
|
||||
get(key) {
|
||||
if (!this.map.has(key)) return undefined;
|
||||
const value = this.map.get(key);
|
||||
this.map.delete(key);
|
||||
this.map.set(key, value);
|
||||
return value;
|
||||
}
|
||||
|
||||
/** 取值但**不**刷新顺序。给「只是想看一眼」的场合。 */
|
||||
peek(key) {
|
||||
return this.map.get(key);
|
||||
}
|
||||
|
||||
set(key, value) {
|
||||
// 已存在时先删：Map 的 set 不改变已有键的位置，不删就刷不了活跃度。
|
||||
if (this.map.has(key)) this.map.delete(key);
|
||||
this.map.set(key, value);
|
||||
while (this.map.size > this.limit) {
|
||||
const oldest = this.map.keys().next().value;
|
||||
this.map.delete(oldest);
|
||||
this.evicted++;
|
||||
}
|
||||
return this;
|
||||
}
|
||||
|
||||
delete(key) {
|
||||
return this.map.delete(key);
|
||||
}
|
||||
|
||||
clear() {
|
||||
this.map.clear();
|
||||
}
|
||||
|
||||
keys() {
|
||||
return this.map.keys();
|
||||
}
|
||||
|
||||
values() {
|
||||
return this.map.values();
|
||||
}
|
||||
|
||||
entries() {
|
||||
return this.map.entries();
|
||||
}
|
||||
|
||||
[Symbol.iterator]() {
|
||||
return this.map[Symbol.iterator]();
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 有界 Set，超过上限时丢弃最久未访问的成员。
|
||||
*
|
||||
* `has` 也刷新顺序：与 `BoundedMap.get` 同理。对 `deliveredMails` 这意味着
|
||||
* 「刚被去重挡下的那封」会留得更久，正合语义。
|
||||
*/
|
||||
export class BoundedSet {
|
||||
/** @param {number} limit 成员上限 */
|
||||
constructor(limit) {
|
||||
this.limit = normalizeLimit(limit);
|
||||
/** @type {Set<any>} */
|
||||
this.set = new Set();
|
||||
this.evicted = 0;
|
||||
}
|
||||
|
||||
get size() {
|
||||
return this.set.size;
|
||||
}
|
||||
|
||||
has(value) {
|
||||
if (!this.set.has(value)) return false;
|
||||
this.set.delete(value);
|
||||
this.set.add(value);
|
||||
return true;
|
||||
}
|
||||
|
||||
/** 判断存在但**不**刷新顺序。 */
|
||||
peek(value) {
|
||||
return this.set.has(value);
|
||||
}
|
||||
|
||||
add(value) {
|
||||
if (this.set.has(value)) this.set.delete(value);
|
||||
this.set.add(value);
|
||||
while (this.set.size > this.limit) {
|
||||
const oldest = this.set.values().next().value;
|
||||
this.set.delete(oldest);
|
||||
this.evicted++;
|
||||
}
|
||||
return this;
|
||||
}
|
||||
|
||||
delete(value) {
|
||||
return this.set.delete(value);
|
||||
}
|
||||
|
||||
clear() {
|
||||
this.set.clear();
|
||||
}
|
||||
|
||||
values() {
|
||||
return this.set.values();
|
||||
}
|
||||
|
||||
[Symbol.iterator]() {
|
||||
return this.set[Symbol.iterator]();
|
||||
}
|
||||
}
|
||||
@ -7,6 +7,8 @@
|
||||
// 实测踩过 —— 插件静默不加载，邮件全都投不进去。
|
||||
// 因此入口文件只能 `export default`，其余东西一律搁在这里。
|
||||
|
||||
import { BoundedMap, MAX_TRACKED_SESSIONS } from './bounded.js';
|
||||
|
||||
/** 取三维地址的名字段：admin@root.alias -> admin */
|
||||
export function addrName(addr) {
|
||||
return String(addr || "").split("@")[0].trim();
|
||||
@ -25,8 +27,18 @@ export function addrName(addr) {
|
||||
*
|
||||
* 窗口是「一轮」：deliverMail 投递新邮件时清空（新一轮开始），
|
||||
* relaySummary 用完即清。
|
||||
*
|
||||
* # 为什么仍要有界
|
||||
*
|
||||
* 上面那些清理路径都要求「这个会话之后还有事发生」。一条只发过一次信、
|
||||
* 之后既没有新邮件也没有 idle 的会话（模型自己发完就没动静了，或者插件在
|
||||
* 那一轮之后重连），条目就永久留下。桥是常驻进程，这种残留会一直累积。
|
||||
*
|
||||
* 淘汰是安全的：条目的语义是「本轮已经亲手回过」，而「本轮」是分钟级的。
|
||||
* 丢掉一条很久以前的记录最坏结果是那条会话下一次 idle 时多转一封总结，
|
||||
* 而服务端的 relay_key 幂等还会兜一层。
|
||||
*/
|
||||
export const explicitSends = new Map(); // opencode session id -> { names:Set, replyTos:Set }
|
||||
export const explicitSends = new BoundedMap(MAX_TRACKED_SESSIONS); // opencode session id -> { names:Set, replyTos:Set }
|
||||
|
||||
/** 记下模型这一轮主动发了信，给谁、回的哪封。 */
|
||||
export function noteExplicitSend(sessionID, to, replyTo) {
|
||||
|
||||
321
plugins/pi-mail-bridge/src/session-scan.mjs
Normal file
321
plugins/pi-mail-bridge/src/session-scan.mjs
Normal file
@ -0,0 +1,321 @@
|
||||
/**
|
||||
* pi 会话目录的增量扫描器 —— 替换心跳路径上的 `SessionManager.listAll()`。
|
||||
*
|
||||
* # 为什么不能用 listAll
|
||||
*
|
||||
* 心跳每 30 秒要上报一次平台会话快照（`snapshotPiSessions`），而它只用到四个
|
||||
* 字段：`id` / `cwd` / `name` / `modified`。`SessionManager.listAll()` 为了拿到
|
||||
* 这四个字段，把 `~/.pi/agent/sessions` 下**每个 .jsonl 的每一行**都读进来并
|
||||
* `JSON.parse`，还顺手把所有消息正文拼成一个 `allMessagesText` 大字符串。
|
||||
*
|
||||
* 本机实测（115 个文件 / 145MB，其中单个会话文件 29MB、单行最长 2.6MB）：
|
||||
*
|
||||
* listAll() 1431ms RSS 41 → 323MB（heapUsed 141MB）
|
||||
* 仅读 header 3ms RSS 41 → 46MB
|
||||
* 本模块（冷启动） 516ms RSS 41 → 131MB
|
||||
* 本模块（稳态） 3ms 重扫 0 字节
|
||||
*
|
||||
* 那 282MB 每 30 秒分配一次、随即变成垃圾。GC 收得掉（所以 RSS 呈锯齿而不是
|
||||
* 单调上升），但代价是：常驻内存被垃圾撑到 300MB 上下，且每拍有 1.4 秒的同步
|
||||
* 解析跑在事件循环上 —— 那期间 SSE 读循环停着，新邮件事件在 TCP 缓冲区排队。
|
||||
*
|
||||
* # 三条省法
|
||||
*
|
||||
* 1. **id / cwd / created 只在首行**。header 是第一行，读 4KB 就够，不必读全文。
|
||||
* 2. **name 来自 `session_info` 行**，而那种行只有几百字节。按行扫描时长度超过
|
||||
* 上限的行**直接跳过、不materialize**，于是 2.6MB 的 message 行不进内存。
|
||||
* 3. **文件是 append-only 的**。缓存 `size`，下一拍只扫 `[上次 size, 现 size)`
|
||||
* 这段尾巴 —— 没有新消息的会话一个字节都不读。
|
||||
*
|
||||
* `modified` 改用 `stat.mtime`：listAll 是从最后一条消息的活动时间算的，
|
||||
* 而快照只拿它排序（「最近在谈的排前面」），mtime 表达的正是这件事，且免费。
|
||||
*
|
||||
* # 为什么这个模块不进 lib/（不与另两个平台共用）
|
||||
*
|
||||
* 它读的是 pi 自己的磁盘格式。opencode 的 `session.list()` 是进程内调用，
|
||||
* DSH 走 `sessionQuery` 的语料观测 —— 两者都没有「扫目录读文件」这一步，
|
||||
* 强行抽象成共用模块只会得到一个谁都不合身的接口。
|
||||
*/
|
||||
|
||||
import { createReadStream } from 'node:fs';
|
||||
import { readdir, stat, open } from 'node:fs/promises';
|
||||
import { join } from 'node:path';
|
||||
import { StringDecoder } from 'node:string_decoder';
|
||||
|
||||
/**
|
||||
* 单行长度上限（字符）。超过这个长度的行不参与解析。
|
||||
*
|
||||
* `session_info` 行的构成是固定的：type + 两个 8 字节 id + ISO 时间戳 + name，
|
||||
* 而 pi-web 的标题生成器把 name 截到 60 字符。几百字节封顶，留 4096 是十倍余量。
|
||||
*
|
||||
* 这个上限同时是**内存上界**：扫描时跨块累积的未完成行一旦超过它就被丢弃，
|
||||
* 因此无论会话里有多大的一条 message（实测见过 2.6MB），扫描峰值都不受影响。
|
||||
*/
|
||||
const MAX_LINE_CHARS = 4096;
|
||||
|
||||
/** header 只读这么多字节。第一行是 `{"type":"session",...}`，远不到 4KB。 */
|
||||
const HEADER_READ_BYTES = 4096;
|
||||
|
||||
/** `session_info` 行的判别串。先做子串命中再 JSON.parse，省掉绝大多数解析。 */
|
||||
const SESSION_INFO_NEEDLE = '"type":"session_info"';
|
||||
|
||||
/**
|
||||
* 读会话文件的 header（第一行）。
|
||||
*
|
||||
* @param {string} file
|
||||
* @returns {Promise<{id: string, cwd: string, created: Date} | null>}
|
||||
* 不是合法会话文件时返回 null（首行不是 session 类型、空文件、读不动）。
|
||||
*/
|
||||
async function readHeader(file) {
|
||||
let fh;
|
||||
try {
|
||||
fh = await open(file, 'r');
|
||||
const buf = Buffer.allocUnsafe(HEADER_READ_BYTES);
|
||||
const { bytesRead } = await fh.read(buf, 0, HEADER_READ_BYTES, 0);
|
||||
if (bytesRead === 0) return null;
|
||||
const text = buf.subarray(0, bytesRead).toString('utf8');
|
||||
const nl = text.indexOf('\n');
|
||||
// 没有换行说明首行比 4KB 还长 —— 那不是 header（header 是固定几个字段）。
|
||||
if (nl < 0) return null;
|
||||
const entry = JSON.parse(text.slice(0, nl));
|
||||
if (entry?.type !== 'session' || typeof entry.id !== 'string') return null;
|
||||
const ts = typeof entry.timestamp === 'string' ? new Date(entry.timestamp) : null;
|
||||
return {
|
||||
id: entry.id,
|
||||
// 老会话的 cwd 是空串（pi 的 SessionInfo 注释写明了），照实传下去 ——
|
||||
// snapshotPiSessions 会按空 workspace 上报，不该拿桥自己的 cwd 冒充。
|
||||
cwd: typeof entry.cwd === 'string' ? entry.cwd : '',
|
||||
created: ts && !Number.isNaN(ts.getTime()) ? ts : new Date(0),
|
||||
};
|
||||
} catch {
|
||||
// 读不动、JSON 坏了、文件正好被删 —— 一律当「不是会话」。
|
||||
// 单个坏文件不该让整份快照失败（listAll 也是这个策略）。
|
||||
return null;
|
||||
} finally {
|
||||
await fh?.close().catch(() => {});
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 扫一段字节区间，返回其中**最后一个** `session_info` 的 name。
|
||||
*
|
||||
* 「最后一个」而不是第一个：会话可以被改名多次，也可以显式清名
|
||||
* （`session_info` 不带 name = 清掉）。语义与 SDK 的 buildSessionInfo 一致 ——
|
||||
* 最新的那条生效。
|
||||
*
|
||||
* @param {string} file
|
||||
* @param {number} from 起始字节（含）。append-only 文件里它一定落在行首。
|
||||
* @param {number} to 结束字节（不含）
|
||||
* @returns {Promise<{name: string | undefined, found: boolean}>}
|
||||
* `found` 为假表示这段里没有任何 session_info —— 调用方应保留上一次的 name，
|
||||
* 而不是把它当成「被清空了」。
|
||||
*/
|
||||
async function scanRangeForName(file, from, to) {
|
||||
if (to <= from) return { name: undefined, found: false };
|
||||
|
||||
let name;
|
||||
let found = false;
|
||||
const decoder = new StringDecoder('utf8');
|
||||
let pending = '';
|
||||
// 当前这一行已经超过上限 → 丢弃它剩下的部分，直到下一个换行。
|
||||
// 这是内存上界的实现：巨大的 message 行永远不会被拼出来。
|
||||
let skipping = false;
|
||||
|
||||
const consider = (line) => {
|
||||
if (line.length > MAX_LINE_CHARS) return;
|
||||
if (!line.includes(SESSION_INFO_NEEDLE)) return;
|
||||
try {
|
||||
const entry = JSON.parse(line);
|
||||
if (entry?.type !== 'session_info') return;
|
||||
found = true;
|
||||
const n = typeof entry.name === 'string' ? entry.name.trim() : '';
|
||||
name = n || undefined;
|
||||
} catch {
|
||||
// 半截行（起点没对齐、文件正在被写）解析失败 —— 忽略即可，
|
||||
// 下一拍尾巴长出来之后会重新看到完整的那一行。
|
||||
}
|
||||
};
|
||||
|
||||
const stream = createReadStream(file, { start: from, end: to - 1 });
|
||||
for await (const chunk of stream) {
|
||||
const text = decoder.write(chunk);
|
||||
let start = 0;
|
||||
for (;;) {
|
||||
const nl = text.indexOf('\n', start);
|
||||
if (nl < 0) break;
|
||||
if (!skipping) consider(pending + text.slice(start, nl));
|
||||
pending = '';
|
||||
skipping = false;
|
||||
start = nl + 1;
|
||||
}
|
||||
if (skipping) continue;
|
||||
pending += text.slice(start);
|
||||
if (pending.length > MAX_LINE_CHARS) {
|
||||
pending = '';
|
||||
skipping = true;
|
||||
}
|
||||
}
|
||||
const tail = decoder.end();
|
||||
if (!skipping) {
|
||||
pending += tail;
|
||||
// 末行没有换行符时也要看一眼（正在被写入的会话就是这种状态）
|
||||
if (pending) consider(pending);
|
||||
}
|
||||
|
||||
return { name, found };
|
||||
}
|
||||
|
||||
/**
|
||||
* 建一个扫描器。缓存跨拍存活，因此要在插件启动时建一次、之后复用。
|
||||
*
|
||||
* @param {object} [opts]
|
||||
* @param {string} [opts.sessionsDir] 会话根目录。默认 `~/.pi/agent/sessions`
|
||||
* （由调用方传 `join(getAgentDir(), 'sessions')`，这里不 import SDK
|
||||
* —— 让这个模块可以脱离 SDK 单测）。
|
||||
* @returns {{scan: () => Promise<object[]>, stats: () => object}}
|
||||
*/
|
||||
export function createSessionScanner({ sessionsDir } = {}) {
|
||||
if (!sessionsDir) throw new Error('createSessionScanner 需要 sessionsDir');
|
||||
|
||||
/**
|
||||
* file -> { size, id, cwd, created, name, hasName }
|
||||
*
|
||||
* 这张表的键是磁盘上真实存在的文件，每次 scan 都会把消失的文件删掉 ——
|
||||
* 所以它不需要额外的上界：会话文件被删（pi 侧清理历史）时条目跟着走。
|
||||
*
|
||||
* `hasName` 与 `name === undefined` 不同：前者是「曾经见过 session_info」，
|
||||
* 后者可能是「见过但被清名了」。区分它们才能让增量扫描保留上一次的结论。
|
||||
*/
|
||||
const cache = new Map();
|
||||
let fullScans = 0;
|
||||
let tailScans = 0;
|
||||
let tailBytes = 0;
|
||||
|
||||
async function collectFiles() {
|
||||
const out = [];
|
||||
let dirs;
|
||||
try {
|
||||
dirs = await readdir(sessionsDir, { withFileTypes: true });
|
||||
} catch (e) {
|
||||
// 目录不存在（pi 从没跑过）= 确实一条会话都没有，空列表是正确答案。
|
||||
//
|
||||
// 其他错误（权限、I/O）必须**抛出去**：调用方据此省略 platform_sessions
|
||||
// 字段，保留服务端镜像。返回空数组的语义是「平台确实没有会话」，
|
||||
// 会把镜像抹掉（W-3 / N-7）—— 一次 EACCES 就能清空别人的补全候选。
|
||||
if (e?.code === 'ENOENT') return out;
|
||||
throw e;
|
||||
}
|
||||
for (const d of dirs) {
|
||||
if (!d.isDirectory() && !d.isSymbolicLink()) continue;
|
||||
const dir = join(sessionsDir, d.name);
|
||||
try {
|
||||
for (const f of await readdir(dir)) {
|
||||
if (f.endsWith('.jsonl')) out.push(join(dir, f));
|
||||
}
|
||||
} catch {
|
||||
// 单个子目录读不动（权限、正被删）不该拖垮整轮
|
||||
}
|
||||
}
|
||||
return out;
|
||||
}
|
||||
|
||||
/**
|
||||
* 扫一遍，返回与 `SessionManager.listAll()` 同形的条目
|
||||
* （`snapshotPiSessions` 用到的那四个字段 + created）。
|
||||
*
|
||||
* 只在这里做 I/O；调用方拿到的是纯数据。
|
||||
*/
|
||||
async function scan() {
|
||||
const files = await collectFiles();
|
||||
const alive = new Set(files);
|
||||
// 会话文件被删掉之后缓存里的条目必须走，否则这张表就是下一个泄露源。
|
||||
for (const key of [...cache.keys()]) {
|
||||
if (!alive.has(key)) cache.delete(key);
|
||||
}
|
||||
|
||||
const out = [];
|
||||
// 并发 16：这些都是小 I/O（稳态下多数只有一次 stat），并发高一点能盖住
|
||||
// 磁盘延迟；再高就只是给事件循环添堵。
|
||||
for (let i = 0; i < files.length; i += 16) {
|
||||
const batch = await Promise.all(
|
||||
files.slice(i, i + 16).map((f) => scanOne(f).catch(() => null)),
|
||||
);
|
||||
for (const item of batch) if (item) out.push(item);
|
||||
}
|
||||
return out;
|
||||
}
|
||||
|
||||
async function scanOne(file) {
|
||||
let st;
|
||||
try {
|
||||
st = await stat(file);
|
||||
} catch {
|
||||
cache.delete(file);
|
||||
return null;
|
||||
}
|
||||
|
||||
const cached = cache.get(file);
|
||||
|
||||
// 文件变小 = 被重写/截断（pi 只 append，所以这不该发生 —— 但真发生时
|
||||
// 缓存里的 size 会让我们从一个越界的偏移开始读）。整份重扫最保险。
|
||||
const shrank = cached && st.size < cached.size;
|
||||
|
||||
if (!cached || shrank) {
|
||||
const header = await readHeader(file);
|
||||
if (!header) {
|
||||
// 首行不是 header：不是会话文件。记一条空壳避免每拍都重读它。
|
||||
cache.set(file, { size: st.size, id: '', cwd: '', created: new Date(0), name: undefined, hasName: false });
|
||||
return null;
|
||||
}
|
||||
fullScans++;
|
||||
tailBytes += st.size;
|
||||
const { name, found } = await scanRangeForName(file, 0, st.size);
|
||||
const entry = { size: st.size, id: header.id, cwd: header.cwd, created: header.created, name, hasName: found };
|
||||
cache.set(file, entry);
|
||||
return toInfo(file, entry, st);
|
||||
}
|
||||
|
||||
// 空壳（已知不是会话文件）：文件长了也不用管，它不会突然变成会话。
|
||||
if (!cached.id) {
|
||||
cached.size = st.size;
|
||||
return null;
|
||||
}
|
||||
|
||||
if (st.size > cached.size) {
|
||||
tailScans++;
|
||||
tailBytes += st.size - cached.size;
|
||||
const { name, found } = await scanRangeForName(file, cached.size, st.size);
|
||||
cached.size = st.size;
|
||||
// 尾巴里没有 session_info 时**保留**旧 name。写成 `cached.name = name`
|
||||
// 会让每次有新消息的会话都丢掉名字 —— 而没有 name 的会话不上报
|
||||
// （S-1），于是活跃会话会从补全候选里消失。
|
||||
if (found) {
|
||||
cached.name = name;
|
||||
cached.hasName = true;
|
||||
}
|
||||
}
|
||||
return toInfo(file, cached, st);
|
||||
}
|
||||
|
||||
function toInfo(file, entry, st) {
|
||||
return {
|
||||
path: file,
|
||||
id: entry.id,
|
||||
cwd: entry.cwd,
|
||||
name: entry.name,
|
||||
created: entry.created,
|
||||
// 排序用「最近活跃」，mtime 就是它，且已经在手上（stat 已经做过了）
|
||||
modified: st.mtime,
|
||||
};
|
||||
}
|
||||
|
||||
/** 观测用：稳态下 fullScans 应当不再增长，tailBytes 每拍只涨一点。 */
|
||||
const stats = () => ({
|
||||
tracked: cache.size,
|
||||
fullScans,
|
||||
tailScans,
|
||||
tailBytes,
|
||||
});
|
||||
|
||||
return { scan, stats };
|
||||
}
|
||||
@ -107,6 +107,7 @@ export function createMailTools({ client, log, agentName = '', onReconnect }) {
|
||||
reply_to: params.reply_to || '',
|
||||
session_alias: params.session_alias || '',
|
||||
attachment_ids: params.attachment_ids || [],
|
||||
from_session_id: ctx?.sessionManager?.getSessionId?.() || '',
|
||||
// 这里**不带 relay**（N-5）：模型的自主发信要计配额，
|
||||
// 免配额通道只给插件代劳的转发（总结、权限询问、故障报告）。
|
||||
});
|
||||
|
||||
@ -54,6 +54,7 @@ import { explicitSends, shouldSkipAutoRelay } from '../lib/relay-dedup.js';
|
||||
import { autoRelayDecision } from '../lib/relay-policy.js';
|
||||
import { adoptedSessionID, adoptMissingMessage } from '../lib/adopt.js';
|
||||
import { isApproval, isAlwaysDecision } from '../lib/permission-grants.js';
|
||||
import { clampRelayKey, isPermanentFailure } from '../lib/relay-key.js';
|
||||
|
||||
// ─── 与主进程的通道 ───
|
||||
|
||||
@ -121,7 +122,12 @@ function permissionExtension() {
|
||||
if (grants.has(event.toolName)) return;
|
||||
|
||||
// relay_key 用 pi 给的 toolCallId（B-8.1）：服务端随决策事件回传它。
|
||||
const relayKey = `${sid || piSessionId}:${event.toolCallId}`;
|
||||
//
|
||||
// clampRelayKey 不是防御性冗余：启用 extended thinking 时 Bedrock 把
|
||||
// 思考签名拼进 toolCallId，实测长到 437 ~ 13601 字节，键直接超服务端
|
||||
// 160 字节列宽 → 400。同一条会话里长短两种形态混着出现，于是权限询问
|
||||
// 随机成功随机失败（生产日志 21:54:02 失败、21:54:24 同会话成功）。
|
||||
const relayKey = clampRelayKey(`${sid || piSessionId}:${event.toolCallId}`);
|
||||
|
||||
try {
|
||||
// 不传 `to`：决策人由服务端按 会话 owner → 线索里最近的人类 → 409
|
||||
@ -151,8 +157,31 @@ function permissionExtension() {
|
||||
log(`权限询问无人可投，当场拒绝 ${relayKey}：${b.error || ''}`);
|
||||
return { block: true, reason };
|
||||
}
|
||||
// 其余失败是暂时的 → 让位给 pi 本地决策（B-8.2）。
|
||||
log(`权限转发失败，让位给本地决策: ${describeError(e)}`);
|
||||
|
||||
// 其余 4xx（400 / 401 / 403 / 404 / 422…）同样永远不会因重试成功。
|
||||
//
|
||||
// 这里原来一律「让位给本地决策」，而邮件驱动的 worker 没有 TUI ——
|
||||
// 让位等于守卫消失。生产实测：relay_key 过长报 400 被当暂时失败，
|
||||
// 那次 bash 在无人批准的情况下执行了（21:54:02 让位，同会话
|
||||
// 21:54:24 另一次 key 正常，于是权限询问被随机吞掉）。
|
||||
//
|
||||
// fail closed：宁可让模型看到「权限系统坏了」并自己改道，
|
||||
// 也不能悄悄放行一条没人看过的命令。
|
||||
if (isPermanentFailure(e)) {
|
||||
const detail = describeError(e);
|
||||
log(`权限转发遇到永久失败（HTTP ${e?.status}），当场拒绝：${detail}`);
|
||||
return {
|
||||
block: true,
|
||||
reason: [
|
||||
`无法把 ${event.toolName} 的授权请求送达给人类：${detail}`,
|
||||
'这是一个不会因重试而改变的失败（请求本身被服务端拒绝）。',
|
||||
'请改用不需要授权的方式完成，或在回信里说明需要人工执行哪一步。',
|
||||
].join('\n'),
|
||||
};
|
||||
}
|
||||
|
||||
// 暂时失败（5xx / 408 / 429 / 网络层）→ 让位给 pi 本地决策（B-8.2）。
|
||||
log(`权限转发暂时失败，让位给本地决策: ${describeError(e)}`);
|
||||
return;
|
||||
}
|
||||
|
||||
@ -231,9 +260,18 @@ async function loadSession(mailTools) {
|
||||
|
||||
const adoptID = adoptedSessionID(data);
|
||||
if (adoptID) {
|
||||
const { SessionManager } = await import('@earendil-works/pi-coding-agent');
|
||||
// listAll 而不是 list(cwd)：worker 的进程 cwd 与会话 cwd 无关。
|
||||
const all = await SessionManager.listAll();
|
||||
// 用 sessionScanner 而不是 `SessionManager.listAll()`：这里只要 id → path，
|
||||
// 而那两个字段全在会话文件的**首行** header 里。listAll 为了拿它们会把
|
||||
// 每个 .jsonl 的每一行读进来并 JSON.parse，还把所有消息正文拼成一个大字符串
|
||||
// （本机 115 个文件 / 145MB 实测：1431ms、堆里瞬时 240MB）。
|
||||
//
|
||||
// worker 是短命进程，拿不到跨拍缓存的好处，但冷启动也依然便宜得多：
|
||||
// 没有任何一行巨大的 message 被 materialize（实测单行最长 2.63MB）。
|
||||
const { createSessionScanner } = await import('./session-scan.mjs');
|
||||
const { getAgentDir } = await import('@earendil-works/pi-coding-agent');
|
||||
const all = await createSessionScanner({
|
||||
sessionsDir: join(getAgentDir(), 'sessions'),
|
||||
}).scan();
|
||||
const info = all.find((e) => e?.id === adoptID);
|
||||
if (!info?.path) {
|
||||
// 镜像是快照，可以过期。**不能**退回「新建一条」—— 那会让人在 TUI 里
|
||||
|
||||
152
plugins/pi-mail-bridge/test/bounded.test.mjs
Normal file
152
plugins/pi-mail-bridge/test/bounded.test.mjs
Normal file
@ -0,0 +1,152 @@
|
||||
import assert from 'node:assert/strict';
|
||||
import test from 'node:test';
|
||||
|
||||
import { BoundedMap, BoundedSet, MAX_TRACKED_MAILS, MAX_TRACKED_SESSIONS } from '../lib/bounded.js';
|
||||
|
||||
// ─── 上限常量 ───
|
||||
|
||||
test('两个上限的相对大小编码了「丢一条的后果」', () => {
|
||||
// 会话级映射丢一条会让那条会话失去平台侧上下文（真的行为退化），
|
||||
// 而 deliveredMails 丢一条只是理论上可能重复投递一封几千封之前的邮件。
|
||||
// 所以邮件窗口可以给得比会话映射宽。
|
||||
assert.ok(MAX_TRACKED_MAILS >= MAX_TRACKED_SESSIONS,
|
||||
'已投递邮件的窗口应当比会话映射更宽（它的淘汰代价更小）');
|
||||
assert.ok(MAX_TRACKED_SESSIONS > 0);
|
||||
});
|
||||
|
||||
// ─── BoundedMap ───
|
||||
|
||||
test('BoundedMap 未达上限时与普通 Map 行为一致', () => {
|
||||
const m = new BoundedMap(10);
|
||||
m.set('a', 1).set('b', 2);
|
||||
assert.equal(m.size, 2);
|
||||
assert.equal(m.get('a'), 1);
|
||||
assert.equal(m.get('b'), 2);
|
||||
assert.equal(m.has('a'), true);
|
||||
assert.equal(m.has('zzz'), false);
|
||||
assert.equal(m.get('zzz'), undefined);
|
||||
assert.equal(m.evicted, 0);
|
||||
});
|
||||
|
||||
test('BoundedMap 超过上限时丢最老的，size 不再增长', () => {
|
||||
const m = new BoundedMap(3);
|
||||
m.set('a', 1).set('b', 2).set('c', 3).set('d', 4);
|
||||
assert.equal(m.size, 3, '上限之后 size 必须封顶 —— 这正是泄露的反面');
|
||||
assert.equal(m.has('a'), false, 'a 是最老的，应当被淘汰');
|
||||
assert.deepEqual([...m.keys()], ['b', 'c', 'd']);
|
||||
assert.equal(m.evicted, 1);
|
||||
});
|
||||
|
||||
test('BoundedMap 的 get 刷新活跃度，长期被读的键不会被淘汰', () => {
|
||||
const m = new BoundedMap(3);
|
||||
m.set('a', 1).set('b', 2).set('c', 3);
|
||||
m.get('a'); // a 变成最新
|
||||
m.set('d', 4); // 淘汰最老的 —— 现在是 b，不是 a
|
||||
assert.equal(m.has('a'), true, '读也算访问：还在收信的会话不该因为建得早被丢');
|
||||
assert.equal(m.has('b'), false);
|
||||
});
|
||||
|
||||
test('BoundedMap 的 peek 不刷新活跃度', () => {
|
||||
const m = new BoundedMap(3);
|
||||
m.set('a', 1).set('b', 2).set('c', 3);
|
||||
m.peek('a');
|
||||
m.set('d', 4);
|
||||
assert.equal(m.has('a'), false, 'peek 是「只看一眼」，不该改变淘汰顺序');
|
||||
});
|
||||
|
||||
test('BoundedMap 重复 set 同一个键只占一个位置且刷新顺序', () => {
|
||||
const m = new BoundedMap(2);
|
||||
m.set('a', 1).set('b', 2).set('a', 9);
|
||||
assert.equal(m.size, 2);
|
||||
assert.equal(m.get('a'), 9);
|
||||
m.set('c', 3);
|
||||
assert.equal(m.has('b'), false, 'a 被重新 set 过，b 才是最老的');
|
||||
assert.equal(m.has('a'), true);
|
||||
});
|
||||
|
||||
test('BoundedMap 支持 delete / clear / 迭代', () => {
|
||||
const m = new BoundedMap(5);
|
||||
m.set('a', 1).set('b', 2);
|
||||
assert.equal(m.delete('a'), true);
|
||||
assert.equal(m.delete('a'), false);
|
||||
assert.deepEqual([...m.entries()], [['b', 2]]);
|
||||
assert.deepEqual([...m.values()], [2]);
|
||||
assert.deepEqual([...m], [['b', 2]]);
|
||||
m.clear();
|
||||
assert.equal(m.size, 0);
|
||||
});
|
||||
|
||||
// ─── BoundedSet ───
|
||||
|
||||
test('BoundedSet 超过上限时丢最老的成员', () => {
|
||||
const s = new BoundedSet(3);
|
||||
s.add('m1').add('m2').add('m3').add('m4');
|
||||
assert.equal(s.size, 3);
|
||||
assert.equal(s.peek('m1'), false);
|
||||
assert.deepEqual([...s.values()], ['m2', 'm3', 'm4']);
|
||||
assert.equal(s.evicted, 1);
|
||||
});
|
||||
|
||||
test('BoundedSet 的 has 刷新活跃度', () => {
|
||||
const s = new BoundedSet(3);
|
||||
s.add('a').add('b').add('c');
|
||||
assert.equal(s.has('a'), true);
|
||||
s.add('d');
|
||||
assert.equal(s.peek('a'), true, '刚被去重挡下的那封应当留得更久');
|
||||
assert.equal(s.peek('b'), false);
|
||||
});
|
||||
|
||||
test('BoundedSet 重复 add 不占额外位置', () => {
|
||||
const s = new BoundedSet(2);
|
||||
s.add('a').add('a').add('a');
|
||||
assert.equal(s.size, 1);
|
||||
});
|
||||
|
||||
test('BoundedSet 支持 delete / clear / 迭代，且能喂给 new Set()', () => {
|
||||
const s = new BoundedSet(5);
|
||||
s.add('a').add('b');
|
||||
assert.equal(s.delete('a'), true);
|
||||
assert.deepEqual([...s], ['b']);
|
||||
// pool.mailDrivenIDs() 会 `new Set(retired)` —— 少了 Symbol.iterator 就炸
|
||||
assert.deepEqual([...new Set(s)], ['b']);
|
||||
s.clear();
|
||||
assert.equal(s.size, 0);
|
||||
});
|
||||
|
||||
// ─── 负向对照：非法上限必须当场报错 ───
|
||||
|
||||
test('上限为 0 时抛错，而不是静默变成一张永远空着的表', () => {
|
||||
// 0 的后果最隐蔽：每次 set 之后立刻把自己淘汰掉，于是去重全部失效，
|
||||
// 邮件被反复投递，而代码里一行错误都不打。
|
||||
assert.throws(() => new BoundedMap(0), RangeError);
|
||||
assert.throws(() => new BoundedSet(0), RangeError);
|
||||
});
|
||||
|
||||
test('上限为 NaN / 负数 / 非数字时抛错，而不是退化成无界', () => {
|
||||
for (const bad of [NaN, -1, 'abc', undefined, null]) {
|
||||
assert.throws(() => new BoundedMap(bad), RangeError, `BoundedMap(${String(bad)}) 应当抛错`);
|
||||
assert.throws(() => new BoundedSet(bad), RangeError, `BoundedSet(${String(bad)}) 应当抛错`);
|
||||
}
|
||||
});
|
||||
|
||||
test('小数上限向下取整', () => {
|
||||
const m = new BoundedMap(2.9);
|
||||
assert.equal(m.limit, 2);
|
||||
m.set('a', 1).set('b', 2).set('c', 3);
|
||||
assert.equal(m.size, 2);
|
||||
});
|
||||
|
||||
// ─── 压力：确认 size 真的封顶（这条是「不泄露」的直接断言）───
|
||||
|
||||
test('灌一万条之后 size 仍等于上限', () => {
|
||||
const s = new BoundedSet(100);
|
||||
for (let i = 0; i < 10_000; i++) s.add(`mail-${i}`);
|
||||
assert.equal(s.size, 100);
|
||||
assert.equal(s.evicted, 9900);
|
||||
assert.equal(s.peek('mail-9999'), true, '最新的必须还在');
|
||||
assert.equal(s.peek('mail-0'), false);
|
||||
|
||||
const m = new BoundedMap(100);
|
||||
for (let i = 0; i < 10_000; i++) m.set(`s-${i}`, { n: i });
|
||||
assert.equal(m.size, 100);
|
||||
});
|
||||
Reference in New Issue
Block a user