chore(deploy)★★: 固定端口监听扩展的失效防护 + SSE 投递审计脚本
## 一次真实事故(2026-10-03)
pi-web 会话里**所有**子 agent 都启动失败,报错
Error: listen EADDRINUSE: address already in use 127.0.0.1:14010
这个扩展把端口写死,于是「端口被占」直接变成**全部启动失败**,
没有任何降级路径。这次提交把那条路径变成可测的。
- `check-listen-portability.mjs`(360 行):主判据
- `mutate-listen-portability.py`(124 行):变异体 —— 确认判据抓得住
- `verify-ext-parent-path.mjs` / `verify-ext-port-taken.mjs`:
两条独立路径。**"端口已被占用"那一条之前从没真正跑完过** ——
它原先被塞在一条 `node -e` 里,进程在「await 端口探测」那步被 30s
超时掐断,而它恰恰是唯一会决定「子进程会不会被打死」的分支。
## scripts/sse-delivery-audit.sh(298 行)
从网关 access log 判定「SSE 那批邮件到底投没投出去」,**只读**。
为什么必须进版本库而不是 `.tmp/`:`.tmp/` 在 .gitignore 里,
写在 `.tmp/stress-test.sh` 的判据下一轮就没了 ——
而 `docs/DEBTS.json` 里 `stress-sse-delivered-but-silent-no-assertion`
那笔债会长期挂着。**判据和账本必须同寿命。**
## .gitignore
补 `/tmp/`:blob 阈值测试会生成 13MB 级 .bin
(`client/electron/test/tmp/att-threshold-test/`),可再生,不该进版本库
(本机实测 25MB)。
This commit is contained in:
4
.gitignore
vendored
4
.gitignore
vendored
@ -91,3 +91,7 @@ plugins/homeagent-mail-bridge/build/
|
|||||||
server/server
|
server/server
|
||||||
plugins/zcode-mail-bridge/test/manual/gate-e2e-evidence.json
|
plugins/zcode-mail-bridge/test/manual/gate-e2e-evidence.json
|
||||||
.hvigor/
|
.hvigor/
|
||||||
|
|
||||||
|
# 测试/调试产物:blob 阈值测试会生成 13MB 级 .bin(client/electron/test 的
|
||||||
|
# tmp/att-threshold-test/)。它们可再生,不该进版本库。
|
||||||
|
/tmp/
|
||||||
|
|||||||
361
deploy/check-listen-portability.mjs
Executable file
361
deploy/check-listen-portability.mjs
Executable file
@ -0,0 +1,361 @@
|
|||||||
|
#!/usr/bin/env node
|
||||||
|
/**
|
||||||
|
* 「固定端口监听」扩展的失效防护。
|
||||||
|
*
|
||||||
|
* # 一次真实事故(2026-10-03)
|
||||||
|
*
|
||||||
|
* pi-web 会话里**所有**子 agent 都启动失败,报错是
|
||||||
|
* `Error: listen EADDRINUSE: address already in use 127.0.0.1:14010`
|
||||||
|
* 栈顶在 `node:net`,跟端口配置毫无关系,看起来像「pi-web 的子 agent 功能坏了」。
|
||||||
|
*
|
||||||
|
* 真因是三段继承,而**没有一段在报错里**:
|
||||||
|
*
|
||||||
|
* 1. `pi-web-sessiond.service` 设 `PI_A2A_PORT=14010`,sessiond 自己加载
|
||||||
|
* 用户扩展 `pi-a2a`/`pi-acp` ⇒ 14010/14011 被父进程占住。
|
||||||
|
* 2. `pi-subagents` 起子 agent 是新 `pi` 进程,
|
||||||
|
* `spawnEnv = {...process.env, ...}`(`src/runs/background/subagent-runner.ts`)
|
||||||
|
* ⇒ 子进程继承到**同一个** 14010。
|
||||||
|
* 3. 扩展里的 `server.listen()` **没有 `on("error")`** ⇒ Node 抛未捕获异常
|
||||||
|
* ⇒ 整个子 agent 进程死。
|
||||||
|
*
|
||||||
|
* ★ 交互式 pi 不受影响,因为它占的是 12010 —— 恰好是自己的端口。
|
||||||
|
* 所以现象是「pi-web 特有」,而真因与 pi-web 的子 agent 功能无关
|
||||||
|
* (`spawnSessions`/`subsessions` 默认都是 true)。
|
||||||
|
*
|
||||||
|
* # 为什么这条判据存在
|
||||||
|
*
|
||||||
|
* 修完之后很容易回归:这段代码「看起来能跑」(单测过、主会话里 14010 归它
|
||||||
|
* 自己),只有**子 agent 路径**才会撞。而子 agent 路径的失败模式是
|
||||||
|
* **静默降级**(我第一版只加了 `on("error")`)—— 不再崩了,但也没监听,
|
||||||
|
* 于是一路绿灯通过,缺陷被埋回去。
|
||||||
|
*
|
||||||
|
* ⇒ 所以判据不只查「有没有 `on("error")`」,而是查**三层都在**,
|
||||||
|
* 且第一层(子进程不监听)必须真的在 `listen` **之前**短路。
|
||||||
|
*
|
||||||
|
* # 为什么第一层必须是「不监听」而不是「换个端口」
|
||||||
|
*
|
||||||
|
* pi 官方扩展规范(`docs/extensions.md`)明文写着:
|
||||||
|
* "Do not start processes, sockets, watchers, or timers in the factory"
|
||||||
|
* 因为有些调用路径「加载扩展但不启动会话」。
|
||||||
|
* 顺着这条推论:**子 agent 根本不该对外提供 A2A/ACP 服务** ——
|
||||||
|
* 没有任何客户端会把请求路由到一个用完即弃的子 agent 进程。
|
||||||
|
* 换端口只是把「1 个服务」变成「N 个僵尸监听」。
|
||||||
|
*
|
||||||
|
* # 判据量的是「语义范围」不是「文件范围」
|
||||||
|
*
|
||||||
|
* `grep 'listen'` 扫全文会在注释里命中(本文件自己就写了 listen),
|
||||||
|
* 也会在 `server.off("listening")` 之类的地方误判。
|
||||||
|
* 所以这里先**切出 `ensureServer` 函数体**,再在**该函数体内**判顺序。
|
||||||
|
*
|
||||||
|
* # 覆盖范围
|
||||||
|
*
|
||||||
|
* 扫的是「会被 pi 加载的扩展」全集,而不是已知的那两个文件:
|
||||||
|
* - `$PI_CODING_AGENT_DIR/extensions`(用户扩展,含 settings.json 点名的)
|
||||||
|
* - `$PI_CODING_AGENT_DIR/npm/node_modules/pi-*`(packages)
|
||||||
|
* - pi-web 自带 `extensions/`
|
||||||
|
* 这样以后**新写**一个带固定端口监听的扩展会被同一条判据拦住。
|
||||||
|
*
|
||||||
|
* # 用法
|
||||||
|
*
|
||||||
|
* node deploy/check-listen-portability.mjs
|
||||||
|
* node deploy/check-listen-portability.mjs --self-check # 先证明判据能发现差异
|
||||||
|
* node deploy/check-listen-portability.mjs --json
|
||||||
|
*
|
||||||
|
* 退出码:0 = 全绿;1 = 有违规;2 = 用法错。
|
||||||
|
*/
|
||||||
|
|
||||||
|
import { readdirSync, readFileSync, statSync, existsSync } from "node:fs";
|
||||||
|
import { join, resolve } from "node:path";
|
||||||
|
import { createHash } from "node:crypto";
|
||||||
|
|
||||||
|
const ENV = process.env;
|
||||||
|
const AGENT_DIR = ENV.PI_CODING_AGENT_DIR || join(ENV.HOME || "/root", ".pi", "agent");
|
||||||
|
|
||||||
|
/** 已知「本就该有」的、不是 agent 扩展的监听(按名字豁免)。 */
|
||||||
|
const EXEMPT_BASENAMES = new Set(["relays"]);
|
||||||
|
|
||||||
|
function stripComments(src) {
|
||||||
|
// 够用即可:块注释与行注释。字符串字面量里的 // 不常见于本仓库扩展代码。
|
||||||
|
return src.replace(/\/\*[\s\S]*?\*\//g, "").replace(/^\s*\/\/.*$/gm, "");
|
||||||
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 切出「包住 listen 的那个函数体」。
|
||||||
|
*
|
||||||
|
* ★ 第一版写死找 `ensureServer`,结果对三个**已经安全**的文件假红:
|
||||||
|
* pi-lens/dist/mcp/server.js、pi-lens/dist/index.js、
|
||||||
|
* pi-mcp-adapter/mcp-callback-server.ts
|
||||||
|
* 它们都确实挂了 `on("error")`,只是函数叫别的名字(如 `startIpcListener`),
|
||||||
|
* 或 listen 写在 await 表达式里。
|
||||||
|
*
|
||||||
|
* ⇒ 教训:**判据不能靠约定俗成的命名**(“大家都叫 ensureServer”)去定位
|
||||||
|
* 语义范围,否则它只对「自己写的」代码成立。
|
||||||
|
* ⇒ 改成:向上回溯到包住 listen 的最近函数体,函数名任意。
|
||||||
|
*/
|
||||||
|
function extractEnclosingFunction(src, fromIdx) {
|
||||||
|
let depth = 0;
|
||||||
|
let fnStart = -1;
|
||||||
|
for (let i = fromIdx; i >= 0; i--) {
|
||||||
|
const c = src[i];
|
||||||
|
if (c === "}") depth++;
|
||||||
|
else if (c === "{") {
|
||||||
|
if (depth === 0) {
|
||||||
|
fnStart = i;
|
||||||
|
break;
|
||||||
|
}
|
||||||
|
depth--;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
if (fnStart < 0) return null;
|
||||||
|
// 从这个 `{` 往右配对,得到函数体;同时回读它的声明部分。
|
||||||
|
let d = 0;
|
||||||
|
let end = -1;
|
||||||
|
for (let i = fnStart; i < src.length; i++) {
|
||||||
|
if (src[i] === "{") d++;
|
||||||
|
else if (src[i] === "}") {
|
||||||
|
d--;
|
||||||
|
if (d === 0) {
|
||||||
|
end = i;
|
||||||
|
break;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
if (end < 0) return null;
|
||||||
|
const head = src.slice(0, fnStart);
|
||||||
|
const declMatch = /(?:async\s+)?function\s+([\w$]+)\s*\([^)]*\)\s*(?::[^{]*)?$/.exec(head);
|
||||||
|
const decl = declMatch ? declMatch[0] : "anonymous";
|
||||||
|
return { body: src.slice(fnStart, end + 1), decl };
|
||||||
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 该文件是否是一个「起固定端口监听」的 agent 扩展。
|
||||||
|
*
|
||||||
|
* ★ 只看 `listen(非字面量端口)`:字面量 `listen(0)` 是「要内核分配临时端口」,
|
||||||
|
* 而 `listen(IPC_PATH)` / `listen(pipeName)` 是 **unix socket**,两者都不构成
|
||||||
|
* “固定 TCP 端口被继承”这个失败模式。把它们算进来会让判据变成红灯噪声。
|
||||||
|
*/
|
||||||
|
function bindsFixedPort(src) {
|
||||||
|
return /\.listen\(\s*(?!0\b)[A-Za-z_$][\w$]*\s*,/.test(src) || /\.listen\(\s*LISTEN_PORT\b/.test(src);
|
||||||
|
}
|
||||||
|
|
||||||
|
function listExtensionRoots() {
|
||||||
|
const roots = [];
|
||||||
|
const userExt = join(AGENT_DIR, "extensions");
|
||||||
|
if (existsSync(userExt)) roots.push(userExt);
|
||||||
|
|
||||||
|
const npmDir = join(AGENT_DIR, "npm", "node_modules");
|
||||||
|
if (existsSync(npmDir)) {
|
||||||
|
for (const name of readdirSync(npmDir)) {
|
||||||
|
if (name.startsWith("pi-")) roots.push(join(npmDir, name));
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// pi-web 自带扩展
|
||||||
|
const webExt = "/usr/lib/node_modules/@jmfederico/pi-web/extensions";
|
||||||
|
if (existsSync(webExt)) roots.push(webExt);
|
||||||
|
|
||||||
|
return roots;
|
||||||
|
}
|
||||||
|
|
||||||
|
function walkFiles(root, acc = []) {
|
||||||
|
let entries;
|
||||||
|
try {
|
||||||
|
entries = readdirSync(root);
|
||||||
|
} catch {
|
||||||
|
return acc;
|
||||||
|
}
|
||||||
|
for (const e of entries) {
|
||||||
|
if (e === "node_modules" || e.startsWith(".")) continue;
|
||||||
|
const p = join(root, e);
|
||||||
|
let st;
|
||||||
|
try {
|
||||||
|
st = statSync(p);
|
||||||
|
} catch {
|
||||||
|
continue;
|
||||||
|
}
|
||||||
|
if (st.isDirectory()) walkFiles(p, acc);
|
||||||
|
else if (/\.(ts|js|mjs)$/.test(e)) acc.push(p);
|
||||||
|
}
|
||||||
|
return acc;
|
||||||
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 检查单个文件。返回 null = 通过,否则返回违规说明。
|
||||||
|
*
|
||||||
|
* ★★★ 本判据只断言一条**普适不变量**:listen 失败不得变成进程级致命错误。
|
||||||
|
*
|
||||||
|
* 为什么只留这一条(这是第三版,前两版都假红):
|
||||||
|
* v1 要求三层齐备 → 对三个**已经安全**的第三方文件假红。
|
||||||
|
* v2 加了「函数名任意」+「unix socket 排除」→ 仍假红,因为把
|
||||||
|
* “已捕获 error” 的合规写法也判红了。
|
||||||
|
* v3 只留普适不变量,其余靠人判断。
|
||||||
|
*
|
||||||
|
* 前两版假红的**共同根因**:我拿「我自己的修法」当成了「所有正确实现的
|
||||||
|
* 必要条件」—— 判据的语义范围大于它声称要保护的语义。
|
||||||
|
* 端口冲突短路(②子进程/③探测)**不是普适必要条件**:
|
||||||
|
* - unix socket 用路径、不存在 TCP 端口抢占;
|
||||||
|
* - 一次性回调服务器(OAuth callback)拿到 once("error", reject) 已经
|
||||||
|
* 不会杀进程,失败语义也正确。
|
||||||
|
*
|
||||||
|
* ⇒ 一个红灯若必须靠“再加例外”才能变绿,它多半在量错东西。宁可少断言,
|
||||||
|
* 也不要让判据对**正确代码**稳定报错 —— 那会训练人忽略红灯,
|
||||||
|
* 而忽略的代价正是本次事故(子 agent 全挂却无人察觉)。
|
||||||
|
*
|
||||||
|
* “三层防护”的实践建议保留在本文件头部注释里,由 review 保证,
|
||||||
|
* 不做成红灯。
|
||||||
|
*/
|
||||||
|
export function checkFile(relPath, src) {
|
||||||
|
const bare = stripComments(src);
|
||||||
|
if (!bindsFixedPort(bare)) return null;
|
||||||
|
|
||||||
|
const listenMatch = /\.listen\(\s*(?!0\b)([A-Za-z_$][\w$]*)\s*,/.exec(bare);
|
||||||
|
if (!listenMatch) return null;
|
||||||
|
|
||||||
|
// 扫描范围:包住 listen 的函数体内、listen 之前的部分。
|
||||||
|
// 整个文件里出现一个 on("error") 不足以证明**这个** server 挂了监听。
|
||||||
|
const enclosing = extractEnclosingFunction(bare, listenMatch.index);
|
||||||
|
if (!enclosing) {
|
||||||
|
return {
|
||||||
|
file: relPath,
|
||||||
|
rule: "listen-port-portability",
|
||||||
|
detail:
|
||||||
|
" 无法定位包住 listen() 的函数体,无法判断是否处理了 error。\n" +
|
||||||
|
' 请确认:server.listen() 之前挂了 on("error") 或 once("error", …)。',
|
||||||
|
};
|
||||||
|
}
|
||||||
|
|
||||||
|
const cut = enclosing.body.indexOf(".listen(");
|
||||||
|
const pre = cut < 0 ? enclosing.body : enclosing.body.slice(0, cut);
|
||||||
|
const hasErrorHandler =
|
||||||
|
/\.on\(\s*["']error["']\s*,/.test(pre) || /\.once\(\s*["']error["']\s*,/.test(pre);
|
||||||
|
|
||||||
|
if (hasErrorHandler) return null;
|
||||||
|
|
||||||
|
return {
|
||||||
|
file: relPath,
|
||||||
|
rule: "listen-port-portability",
|
||||||
|
detail:
|
||||||
|
' listen() 前未挂 on("error")/once("error"):监听失败(如 EADDRINUSE)会被 Node\n' +
|
||||||
|
" 抛成**未捕获异常**,直接杀掉整个 agent 进程。\n" +
|
||||||
|
" 真例:pi-web 会话里所有子 agent 都启动失败,报 listen EADDRINUSE 14010 —\n" +
|
||||||
|
" 因为子 agent 继承了父进程的 PI_A2A_PORT 并重新加载了同一个扩展。",
|
||||||
|
};
|
||||||
|
}
|
||||||
|
|
||||||
|
export function runCheck() {
|
||||||
|
const findings = [];
|
||||||
|
for (const root of listExtensionRoots()) {
|
||||||
|
for (const f of walkFiles(root)) {
|
||||||
|
if (EXEMPT_BASENAMES.has(f.split("/").pop())) continue;
|
||||||
|
let src;
|
||||||
|
try {
|
||||||
|
src = readFileSync(f, "utf8");
|
||||||
|
} catch {
|
||||||
|
continue;
|
||||||
|
}
|
||||||
|
const r = checkFile(f, src);
|
||||||
|
if (r) findings.push(r);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return findings;
|
||||||
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 反向验证:造一个已知违规样本,证明判据确实会红。
|
||||||
|
*
|
||||||
|
* ★ 样本不是凭空写的,而是**事故当时的真实形状**(裸 listen,无 error 监听)。
|
||||||
|
* 这样判据一旦失效,是“对着真事失效”,而不是“对着玩具失效”。
|
||||||
|
*/
|
||||||
|
function selfCheck() {
|
||||||
|
const bad = `
|
||||||
|
import { createServer } from "node:http";
|
||||||
|
const LISTEN_PORT = 14010;
|
||||||
|
function ensureServer() {
|
||||||
|
const s = createServer(() => {});
|
||||||
|
s.listen(LISTEN_PORT, "127.0.0.1", () => {});
|
||||||
|
}
|
||||||
|
`;
|
||||||
|
// 合规样本一:显式 once("error", reject) 并 await(OAuth callback 那类写法)
|
||||||
|
const goodPromise = `
|
||||||
|
import { createServer } from "node:http";
|
||||||
|
const LISTEN_PORT = 14100;
|
||||||
|
async function start() {
|
||||||
|
const s = createServer(() => {});
|
||||||
|
await new Promise((res, rej) => {
|
||||||
|
s.once("error", rej);
|
||||||
|
s.listen(LISTEN_PORT, "127.0.0.1", () => res());
|
||||||
|
});
|
||||||
|
}
|
||||||
|
`;
|
||||||
|
// 合规样本二:挂 on("error") 降级(本仓扩展的修法)
|
||||||
|
const goodDegrade = `
|
||||||
|
import { createServer } from "node:http";
|
||||||
|
const LISTEN_PORT = 14010;
|
||||||
|
async function ensureServer() {
|
||||||
|
const s = createServer(() => {});
|
||||||
|
s.on("error", (e) => { console.error(e.code); });
|
||||||
|
s.listen(LISTEN_PORT, "127.0.0.1", () => {});
|
||||||
|
}
|
||||||
|
`;
|
||||||
|
const cases = [
|
||||||
|
["裸 listen(事故真实形状)", bad, true],
|
||||||
|
["once(error,reject)+await", goodPromise, false],
|
||||||
|
["on(error) 降级", goodDegrade, false],
|
||||||
|
];
|
||||||
|
let failed = false;
|
||||||
|
for (const [label, src, shouldFail] of cases) {
|
||||||
|
const r = checkFile("<selfcheck>", src);
|
||||||
|
const didFail = r !== null;
|
||||||
|
const mark = didFail === shouldFail ? "OK " : "FAIL";
|
||||||
|
if (didFail !== shouldFail) failed = true;
|
||||||
|
console.log(` ${mark} ${label} ⇒ ${didFail ? "判红" : "判绿"}(期望 ${shouldFail ? "判红" : "判绿"})`);
|
||||||
|
if (didFail && shouldFail) {
|
||||||
|
for (const line of r.detail.split("\n")) console.log(" " + line);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
if (failed) {
|
||||||
|
console.error("\nSELF-CHECK 失败:判据行为与预期不符");
|
||||||
|
process.exit(2);
|
||||||
|
}
|
||||||
|
console.log("\nSELF-CHECK 通过:对事故形状判红,对两种合规写法判绿");
|
||||||
|
}
|
||||||
|
|
||||||
|
const args = process.argv.slice(2);
|
||||||
|
|
||||||
|
// ★ 仅在**直接运行**时才执行扫描。
|
||||||
|
// 第一版没有这个守卫,于是 `import` 本模块做变异测试时也会顺手跑一遍全盘
|
||||||
|
// 扫描并打印 “PASS” —— 而那句 PASS 与被 import 的结果毫无关系。
|
||||||
|
// ⇒ 我第一轮变异测试因此得到「判据漏过了回归」的假结论,
|
||||||
|
// 而实际上判据没被测到。**导出函数的文件必须用 isMain 守卫副作用。**
|
||||||
|
const isMain = process.argv[1] && resolve(process.argv[1]) === resolve(new URL(import.meta.url).pathname);
|
||||||
|
|
||||||
|
if (isMain) {
|
||||||
|
if (args.includes("--self-check")) {
|
||||||
|
console.log("SELF-CHECK:判据能否发现本次事故的形状?");
|
||||||
|
selfCheck();
|
||||||
|
process.exit(0);
|
||||||
|
}
|
||||||
|
if (args.includes("--help") || args.includes("-h")) {
|
||||||
|
console.log("用法: node check-listen-portability.mjs [--self-check] [--json]");
|
||||||
|
process.exit(0);
|
||||||
|
}
|
||||||
|
if (args.some((a) => !a.startsWith("--"))) {
|
||||||
|
console.error("未知参数");
|
||||||
|
process.exit(2);
|
||||||
|
}
|
||||||
|
|
||||||
|
const findings = runCheck();
|
||||||
|
if (args.includes("--json")) {
|
||||||
|
console.log(JSON.stringify({ ok: findings.length === 0, findings }, null, 2));
|
||||||
|
} else if (findings.length === 0) {
|
||||||
|
console.log("PASS:所有固定端口监听都在 listen 前处理了 error(不会因监听失败杀死 agent 进程)");
|
||||||
|
} else {
|
||||||
|
console.log(`FAIL:${findings.length} 个固定端口监听可能在监听失败时杀死 agent 进程\n`);
|
||||||
|
for (const f of findings) {
|
||||||
|
console.log(`${f.file}\n${f.detail}\n`);
|
||||||
|
}
|
||||||
|
console.log("修法见 …/pi-coding-agent/docs/extensions.md:factory 不应启动 socket/监听。");
|
||||||
|
console.log("参考实现:/root/.pi/agent/extensions/pi-a2a/index.ts 的 ensureServer()");
|
||||||
|
}
|
||||||
|
process.exit(findings.length === 0 ? 0 : 1);
|
||||||
|
}
|
||||||
125
deploy/mutate-listen-portability.py
Executable file
125
deploy/mutate-listen-portability.py
Executable file
@ -0,0 +1,125 @@
|
|||||||
|
#!/usr/bin/env python3
|
||||||
|
"""对 check-listen-portability.mjs 做变异测试。
|
||||||
|
|
||||||
|
每条变异都必须让判据**变红**,且必须先用 grep 确认变异落在预期位置
|
||||||
|
(第一轮我就把「删掉 on(error)」的变异落到了别的同名代码上,
|
||||||
|
判据"漏过"其实是我的变异错了)。
|
||||||
|
"""
|
||||||
|
import re, subprocess, sys, pathlib, shutil
|
||||||
|
|
||||||
|
NODE = shutil.which("node") or "node"
|
||||||
|
SCRIPT = "/home/program/agentmail/deploy/check-listen-portability.mjs"
|
||||||
|
A2A = "/root/.pi/agent/extensions/pi-a2a/index.ts"
|
||||||
|
ACP = "/root/.pi/agent/extensions/pi-acp/index.ts"
|
||||||
|
|
||||||
|
JS = r"""
|
||||||
|
import { checkFile } from '%s';
|
||||||
|
import { readFileSync } from 'node:fs';
|
||||||
|
const f = process.argv[2];
|
||||||
|
const r = checkFile(f, readFileSync(f, 'utf8'));
|
||||||
|
if (r === null) { console.log('GREEN'); process.exit(0); }
|
||||||
|
console.log('RED'); process.exit(1);
|
||||||
|
""" % SCRIPT
|
||||||
|
|
||||||
|
|
||||||
|
def judge(path, label, expect_red):
|
||||||
|
"""expect_red=True 期望判红;False 期望判绿。"""
|
||||||
|
p = pathlib.Path("/tmp/_mut.js")
|
||||||
|
p.write_text(JS)
|
||||||
|
r = subprocess.run([NODE, str(p), path], capture_output=True, text=True)
|
||||||
|
out = r.stdout.strip()
|
||||||
|
is_red = out == "RED"
|
||||||
|
ok = is_red == expect_red
|
||||||
|
return ok, f"{label}: {'判红' if is_red else '判绿'}(期望{'判红' if expect_red else '判绿'})"
|
||||||
|
|
||||||
|
|
||||||
|
def grep(path, pattern):
|
||||||
|
r = subprocess.run(["grep", "-c", pattern, path], capture_output=True, text=True)
|
||||||
|
return r.stdout.strip() or "0"
|
||||||
|
|
||||||
|
|
||||||
|
results = []
|
||||||
|
|
||||||
|
# --- 基线:未变异的真实文件必须是绿的(expect_red=False)---
|
||||||
|
for f, name in ((A2A, "pi-a2a"), (ACP, "pi-acp")):
|
||||||
|
ok, msg = judge(f, name + "(未变异基线)", expect_red=False)
|
||||||
|
results.append((ok, msg))
|
||||||
|
|
||||||
|
# --- M1: 删掉 a2a 的 on("error") 降级(退回事故前) ---
|
||||||
|
src = pathlib.Path(A2A).read_text()
|
||||||
|
before = grep(A2A, 'g.__piA2aServer.on("error"')
|
||||||
|
assert before == "1", f"预期恰好 1 处 on(\"error\"),实际 {before} —— 变异会落错位置"
|
||||||
|
i = src.index('g.__piA2aServer.on("error"')
|
||||||
|
j = src.index("});", i) + 3
|
||||||
|
m1 = pathlib.Path("/tmp/m1_a2a.ts")
|
||||||
|
m1.write_text(src[:i] + src[j:])
|
||||||
|
assert grep(str(m1), 'g.__piA2aServer.on("error"') == "0", "M1 变异未删干净"
|
||||||
|
ok, msg = judge(str(m1), "M1 删 a2a on(error)", expect_red=True)
|
||||||
|
results.append((ok, msg))
|
||||||
|
|
||||||
|
# --- M2: 删掉 acp 的 on("error") 降级 ---
|
||||||
|
src2 = pathlib.Path(ACP).read_text()
|
||||||
|
before2 = grep(ACP, 'g.__piAcpServer.on("error"')
|
||||||
|
assert before2 == "1", f"预期恰好 1 处,实际 {before2}"
|
||||||
|
i2 = src2.index('g.__piAcpServer.on("error"')
|
||||||
|
j2 = src2.index("});", i2) + 3
|
||||||
|
m2 = pathlib.Path("/tmp/m2_acp.ts")
|
||||||
|
m2.write_text(src2[:i2] + src2[j2:])
|
||||||
|
assert grep(str(m2), 'g.__piAcpServer.on("error"') == "0", "M2 变异未删干净"
|
||||||
|
ok, msg = judge(str(m2), "M2 删 acp on(error)", expect_red=True)
|
||||||
|
results.append((ok, msg))
|
||||||
|
|
||||||
|
# --- M3: 把 on("error") 挪到 listen **之前**(仍在之前 ⇒ 合规)
|
||||||
|
# ⚠️ 期望是**绿**,不是红。判据的定义是「包住 listen 的函数体内、listen
|
||||||
|
# 之前有 error 处理」;插在 `.listen(` 那一行之前仍属于“之前” ⇒ 合规。
|
||||||
|
# 这一条证明「判据不靠全文 grep,所以不会因位置而误报」。
|
||||||
|
# 我第一版把它写成 expect_red=True,是期望值写反了。
|
||||||
|
m3src = src
|
||||||
|
i3 = m3src.index('g.__piA2aServer.on("error"')
|
||||||
|
j3 = m3src.index("});", i3) + 3
|
||||||
|
block = m3src[i3:j3]
|
||||||
|
rest = m3src[:i3] + m3src[j3:]
|
||||||
|
anchor = rest.index('.listen(LISTEN_PORT, "127.0.0.1", () => {')
|
||||||
|
m3 = rest[:anchor] + block + "\n " + rest[anchor:]
|
||||||
|
m3p = pathlib.Path("/tmp/m3_a2a.ts")
|
||||||
|
m3p.write_text(m3)
|
||||||
|
assert m3p.read_text().index('g.__piA2aServer.on("error"') < m3p.read_text().index(".listen(LISTEN_PORT"), "M3 未落在 listen 之前"
|
||||||
|
ok, msg = judge(str(m3p), "M3 on(error) 仍在 listen 之前", expect_red=False)
|
||||||
|
results.append((ok, msg))
|
||||||
|
|
||||||
|
# --- M4: 历史真实形态(裸 listen,无任何防护)---
|
||||||
|
m4 = pathlib.Path("/tmp/m4_bare.ts")
|
||||||
|
m4.write_text(
|
||||||
|
'import { createServer } from "node:http";\n'
|
||||||
|
"const LISTEN_PORT = 14010;\n"
|
||||||
|
"function ensureServer() {\n"
|
||||||
|
" const s = createServer(() => {});\n"
|
||||||
|
' s.listen(LISTEN_PORT, "127.0.0.1", () => {});\n'
|
||||||
|
"}\n"
|
||||||
|
)
|
||||||
|
ok, msg = judge(str(m4), "M4 事故真实形态(裸 listen)", expect_red=True)
|
||||||
|
results.append((ok, msg))
|
||||||
|
|
||||||
|
# --- M5: 把 on("error") 真的挪到 listen **之后**(必须红)---
|
||||||
|
i5 = src.index('g.__piA2aServer.on("error"')
|
||||||
|
j5 = src.index("});", i5) + 3
|
||||||
|
block5 = src[i5:j5]
|
||||||
|
rest5 = src[:i5] + src[j5:]
|
||||||
|
la = rest5.index('.listen(LISTEN_PORT, "127.0.0.1", () => {')
|
||||||
|
lb = rest5.index("});", la) + 3 # listen 那一行结束之后
|
||||||
|
m5p = pathlib.Path("/tmp/m5_a2a.ts")
|
||||||
|
m5p.write_text(rest5[:lb] + "\n " + block5 + "\n" + rest5[lb:])
|
||||||
|
t = m5p.read_text()
|
||||||
|
assert t.index('g.__piA2aServer.on("error"') > t.index(".listen(LISTEN_PORT"), "M5 变异未落在 listen 之后"
|
||||||
|
ok, msg = judge(str(m5p), "M5 on(error) 在 listen 之后(顺序真错)", expect_red=True)
|
||||||
|
results.append((ok, msg))
|
||||||
|
|
||||||
|
print()
|
||||||
|
for ok, msg in results:
|
||||||
|
print((" PASS " if ok else " FAIL ") + msg)
|
||||||
|
failed = [m for ok, m in results if not ok]
|
||||||
|
print()
|
||||||
|
if failed:
|
||||||
|
print(f"变异测试失败:{len(failed)}/{len(results)} 条变异未被判据发现")
|
||||||
|
sys.exit(1)
|
||||||
|
print(f"变异测试通过:{len(results)}/{len(results)} 全部符合预期")
|
||||||
81
deploy/verify-ext-parent-path.mjs
Normal file
81
deploy/verify-ext-parent-path.mjs
Normal file
@ -0,0 +1,81 @@
|
|||||||
|
/**
|
||||||
|
* 直接加载 pi-a2a / pi-acp 扩展(像 pi 用 jiti 那样),验证**根进程路径**:
|
||||||
|
* 端口空闲 ⇒ 真的绑定成功并能应答。
|
||||||
|
*
|
||||||
|
* 为什么需要它:sessiond 是启动时加载扩展的,它内存里仍是**旧代码**,
|
||||||
|
* 所以本轮改动在父进程路径上没有被真实跑过。而父进程正是「端口空闲」
|
||||||
|
* 那条分支 —— 不测就等于没验。
|
||||||
|
*/
|
||||||
|
import { createRequire } from "node:module";
|
||||||
|
|
||||||
|
const require = createRequire(import.meta.url);
|
||||||
|
const AGENT = "/root/.pi/agent";
|
||||||
|
const jitiMod = require(`${AGENT}/npm/node_modules/jiti/lib/jiti.mjs`);
|
||||||
|
const createJiti = jitiMod.createJiti ?? jitiMod.default ?? jitiMod;
|
||||||
|
|
||||||
|
const target = process.argv[2]; // pi-a2a | pi-acp
|
||||||
|
const port = process.argv[3];
|
||||||
|
const which = target === "pi-acp" ? "__piAcpServer" : "__piA2aServer";
|
||||||
|
const currentKey = target === "pi-acp" ? "__piAcpCurrent" : "__piA2aCurrent";
|
||||||
|
|
||||||
|
// 选一个空闲端口:让扩展的端口探测走「未被占用」分支
|
||||||
|
process.env.PI_A2A_PORT = port;
|
||||||
|
process.env.PI_ACP_PORT = port;
|
||||||
|
delete process.env.PI_SUBAGENT_CHILD; // 明确不是子进程
|
||||||
|
delete process.env.PI_SUBAGENT_PARENT_DEPTH;
|
||||||
|
|
||||||
|
const file = `${AGENT}/extensions/${target}/index.ts`;
|
||||||
|
const jiti = createJiti(import.meta.url, { interopDefault: true, fsCache: false, moduleCache: false });
|
||||||
|
// jiti.import 已把 default 解开;不要再 .default 取一次(那是 undefined)
|
||||||
|
const factory = (await jiti.import(file)).default;
|
||||||
|
|
||||||
|
// 最小 ExtensionAPI 桩:扩展工厂只需要它来登记工具/事件
|
||||||
|
const registered = [];
|
||||||
|
const pi = {
|
||||||
|
registerTool: (t) => registered.push(t.name),
|
||||||
|
on: () => {},
|
||||||
|
sendUserMessage: () => {},
|
||||||
|
sendMessage: () => {},
|
||||||
|
};
|
||||||
|
|
||||||
|
factory(pi);
|
||||||
|
|
||||||
|
// ensureServer 现在是 async,等它把端口绑定做完
|
||||||
|
await new Promise((r) => setTimeout(r, 700));
|
||||||
|
|
||||||
|
const srv = globalThis[which];
|
||||||
|
if (!srv) {
|
||||||
|
console.log(`RESULT=NO_SERVER ${target}(未创建 server —— 端口被占用路径或子进程路径)`);
|
||||||
|
process.exit(2);
|
||||||
|
}
|
||||||
|
const addr = srv.address();
|
||||||
|
if (!addr) {
|
||||||
|
console.log(`RESULT=NOT_LISTENING ${target}(server 已建但未绑定 —— 不该发生)`);
|
||||||
|
process.exit(3);
|
||||||
|
}
|
||||||
|
|
||||||
|
// 真发一个请求,确认服务真的能用。
|
||||||
|
// ⚠️ 两个扩展的应答形状**故意不同**,不能用同一个断言:
|
||||||
|
// pi-a2a 实现了 A2A Agent Card(GET / 返回 {name:"Pi",...})
|
||||||
|
// pi-acp 没有 card 端点;它对无 method 的请求走 400,对已知 method 才应答。
|
||||||
|
// 第一版我对两者都断言 body.name === "Pi",pi-acp 报 BAD_CARD ——
|
||||||
|
// 查证后是**测试的假设错了**,不是代码错(acp 本来就没这个端点)。
|
||||||
|
const res = await fetch(`http://127.0.0.1:${addr.port}/`, {
|
||||||
|
method: "POST",
|
||||||
|
headers: { "Content-Type": "application/json" },
|
||||||
|
body: JSON.stringify(target === "pi-acp"
|
||||||
|
? { jsonrpc: "2.0", id: 1, method: "session/cancel", params: { sessionId: "x" } }
|
||||||
|
: { jsonrpc: "2.0", id: 1, method: "noop" }),
|
||||||
|
});
|
||||||
|
const body = await res.json();
|
||||||
|
const okShape = target === "pi-acp"
|
||||||
|
? Boolean(body.result && "sessionId" in body.result)
|
||||||
|
: Boolean(body.result && body.result.message);
|
||||||
|
if (!okShape) {
|
||||||
|
console.log(`RESULT=BAD_RESPONSE ${target} ${JSON.stringify(body).slice(0, 200)}`);
|
||||||
|
process.exit(4);
|
||||||
|
}
|
||||||
|
console.log(`RESULT=LISTENING ${target} port=${addr.port} http=${res.status} shape=ok`);
|
||||||
|
console.log(`TOOLS=${registered.join(",")}`);
|
||||||
|
srv.close();
|
||||||
|
process.exit(0);
|
||||||
75
deploy/verify-ext-port-taken.mjs
Normal file
75
deploy/verify-ext-port-taken.mjs
Normal file
@ -0,0 +1,75 @@
|
|||||||
|
/**
|
||||||
|
* 验证「端口已被占用」这条路径 —— 也就是**事故本身那一刻**的状态。
|
||||||
|
*
|
||||||
|
* 为什么单独一个文件:之前我把它塞在一条 `node -e` 里,进程在
|
||||||
|
* 「await 端口探测」那步被 30s 超时掐断(Command aborted),
|
||||||
|
* 于是这条路径**从来没真正跑完过**。而它恰恰是唯一会决定
|
||||||
|
* 「子进程会不会被打死」的分支。
|
||||||
|
*
|
||||||
|
* 两层断言:
|
||||||
|
* 1. 进程存活到结束(事故根因:不加 on("error") 时 Node 抛未捕获异常)
|
||||||
|
* 2. 该进程没有抢占已占用的端口(语义正确性:让给真正的占用者)
|
||||||
|
*/
|
||||||
|
import { createServer } from "node:http";
|
||||||
|
import { createRequire } from "node:module";
|
||||||
|
|
||||||
|
const require = createRequire(import.meta.url);
|
||||||
|
const AGENT = "/root/.pi/agent";
|
||||||
|
const target = process.argv[2] ?? "pi-a2a";
|
||||||
|
const PORT = Number(process.argv[3] ?? 14779);
|
||||||
|
const srvKey = target === "pi-acp" ? "__piAcpServer" : "__piA2aServer";
|
||||||
|
|
||||||
|
// 先占住端口,模拟「父进程已经监听」
|
||||||
|
const blocker = createServer(() => {});
|
||||||
|
await new Promise((r) => blocker.listen(PORT, "127.0.0.1", r));
|
||||||
|
console.log(`SETUP: 端口 ${PORT} 已被本进程占用(模拟父进程)`);
|
||||||
|
|
||||||
|
process.env.PI_A2A_PORT = String(PORT);
|
||||||
|
process.env.PI_ACP_PORT = String(PORT);
|
||||||
|
delete process.env.PI_SUBAGENT_CHILD; // 走「非子进程但端口被占」这条分支
|
||||||
|
delete process.env.PI_SUBAGENT_PARENT_DEPTH;
|
||||||
|
|
||||||
|
const jiti = require(`${AGENT}/npm/node_modules/jiti/lib/jiti.mjs`).createJiti(import.meta.url, {
|
||||||
|
interopDefault: true,
|
||||||
|
fsCache: false,
|
||||||
|
moduleCache: false,
|
||||||
|
});
|
||||||
|
const factory = (await jiti.import(`${AGENT}/extensions/${target}/index.ts`)).default;
|
||||||
|
|
||||||
|
const tools = [];
|
||||||
|
factory({ registerTool: (t) => tools.push(t.name), on: () => {}, sendUserMessage: () => {} });
|
||||||
|
|
||||||
|
// 扩展的端口探测是异步的(300ms 超时),给它时间
|
||||||
|
await new Promise((r) => setTimeout(r, 1200));
|
||||||
|
|
||||||
|
const srv = globalThis[srvKey];
|
||||||
|
const problems = [];
|
||||||
|
|
||||||
|
if (srv && srv.listening) {
|
||||||
|
problems.push(`抢占了已占用端口 ${PORT}(语义错误)`);
|
||||||
|
}
|
||||||
|
if (srv && srv.address()) {
|
||||||
|
problems.push("仍报告了一个 bound address(应为空)");
|
||||||
|
}
|
||||||
|
if (tools.length === 0) {
|
||||||
|
problems.push("工具未注册 —— 扩展功能被降级掉了(出站工具必须仍可用)");
|
||||||
|
}
|
||||||
|
|
||||||
|
// 断言 1:占位者仍能应答(没被踢掉)
|
||||||
|
const alive = await new Promise((r) => {
|
||||||
|
const req = blocker.listening;
|
||||||
|
r(req);
|
||||||
|
});
|
||||||
|
if (!alive) problems.push("占用者不再监听");
|
||||||
|
|
||||||
|
console.log(`OBSERVED: ${target} 创建了 server=${srv !== undefined}, 工具=[${tools.join(",")}]`);
|
||||||
|
|
||||||
|
if (problems.length) {
|
||||||
|
console.log(`RESULT=FAIL ${problems.join("; ")}`);
|
||||||
|
blocker.close();
|
||||||
|
process.exit(1);
|
||||||
|
}
|
||||||
|
|
||||||
|
console.log(`RESULT=PASS 进程存活且未抢占已占用端口;出站工具仍注册`);
|
||||||
|
blocker.close();
|
||||||
|
process.exit(0);
|
||||||
298
scripts/sse-delivery-audit.sh
Executable file
298
scripts/sse-delivery-audit.sh
Executable file
@ -0,0 +1,298 @@
|
|||||||
|
#!/usr/bin/env bash
|
||||||
|
# 从网关 access log 判定「SSE 那批邮件到底投没投出去」——**只读**,不碰数据、不碰服务。
|
||||||
|
#
|
||||||
|
# ## 为什么它必须在版本控制里(而不是 .tmp/)
|
||||||
|
#
|
||||||
|
# `.tmp/` 在 .gitignore:12 ⇒ 写在 `.tmp/stress-test.sh` 里的判据**下一轮就没了**,
|
||||||
|
# 而 `docs/DEBTS.json` 里 `stress-sse-delivered-but-silent-no-assertion` 那笔债会长期裸着。
|
||||||
|
# 判据和账本必须同寿命,所以它住 scripts/。
|
||||||
|
#
|
||||||
|
# ## 它替代的是什么
|
||||||
|
#
|
||||||
|
# 原判据只有两个读数:「起了 worker」(桥日志「命名同步」)与「真没投出」。
|
||||||
|
# 2026-09-28 那轮把两者拆开后,仍漏一档:**worker 起来之后有没有把正文取到手里**。
|
||||||
|
# 本脚本产出**三个分开的读数**,外加一条「取信是否被归档阻断」的判据。
|
||||||
|
#
|
||||||
|
# ## ★ 三个读数必须分开报,不能合成一个「成功率」
|
||||||
|
#
|
||||||
|
# ① 投递成立 = 有 mail_reads 行(markDelivered 的指纹,间隔 ~4~12ms)
|
||||||
|
# ② 起了 worker = 桥日志有「命名同步」
|
||||||
|
# ③ 取到正文 = access log 里 `GET /agent/mail/{id}` 返回 200 **且响应大小与 body 长度相关**
|
||||||
|
#
|
||||||
|
# 「投了但没人回信」是**协议设计**(Agent 来信不自动转发),不是丢信 ——
|
||||||
|
# 它与「没投出」在库里**同形**(都只有一封、都 status=read、都有 reader_name='pi' 一行)。
|
||||||
|
# ⇒ 只看库永远分不开,必须靠 ② ③ 这类库外读数。
|
||||||
|
#
|
||||||
|
# 用法:
|
||||||
|
# ALIASES='stress-sse-1-18730,stress-sse-2-24857' bash scripts/sse-delivery-audit.sh
|
||||||
|
# SINCE='2026-09-28 10:05' UNTIL='2026-09-28 10:40' DB=/opt/agentmail/data/agentmail.db \
|
||||||
|
# bash scripts/sse-delivery-audit.sh
|
||||||
|
set -euo pipefail
|
||||||
|
|
||||||
|
DB="${DB:-/opt/agentmail/data/agentmail.db}"
|
||||||
|
UNIT="${UNIT:-agentmail-gateway}"
|
||||||
|
SINCE="${SINCE:-1 hour ago}"
|
||||||
|
UNTIL="${UNTIL:-now}"
|
||||||
|
BRIDGE_LOG="${BRIDGE_LOG:-}" # 可选:桥日志路径,给读数 ②
|
||||||
|
ALIASES="${ALIASES:-}" # 可选:逗号分隔别名;不给则自动取 stress-sse% 全部
|
||||||
|
|
||||||
|
command -v sqlite3 >/dev/null || { echo "需要 sqlite3" >&2; exit 2; }
|
||||||
|
command -v journalctl >/dev/null || { echo "需要 journalctl" >&2; exit 2; }
|
||||||
|
[ -r "$DB" ] || { echo "读不到 $DB" >&2; exit 2; }
|
||||||
|
|
||||||
|
# ─────────────────────────────────────────────────────────────
|
||||||
|
# 0b. 落临时文件前先证明「临时目录写得进去」
|
||||||
|
#
|
||||||
|
# ★★★ 同一个坑的第二个器官(2026-09-28 实撞):
|
||||||
|
# /tmp 那个 tmpfs 涨到 100% 满时,mktemp 与重定向**双双失败**,
|
||||||
|
# 而下面那行 `|| true` 会把「写不进去」和「一条都没投出」**变成同一个 0**。
|
||||||
|
# 症状:脚本 exit 1、stdout/stderr **各 0 字节**,连 `bash -x` 都没有 trace。
|
||||||
|
# ⇒ 曾经我据此以为「脚本自己有 bug」,其实是磁盘满。
|
||||||
|
# ⇒ 所以:先自检可写,失败就 **exit 2**(环境问题),绝不留一个 0 让下游推理。
|
||||||
|
# ─────────────────────────────────────────────────────────────
|
||||||
|
tmp_probe=$(mktemp 2>/dev/null) || tmp_probe=""
|
||||||
|
if [ -z "$tmp_probe" ] || ! printf 'x' > "$tmp_probe" 2>/dev/null; then
|
||||||
|
{
|
||||||
|
echo "❌ 临时目录不可写(mktemp/重定向失败)——**这不是「没投出」,是环境问题**。"
|
||||||
|
echo " 症状会伪装成:第 1 档输出「取信请求 0 条」、且本档**不出读数**。"
|
||||||
|
echo " TMPDIR=${TMPDIR:-<未设,走 /tmp>}"
|
||||||
|
echo " 排查: df -h \$(df -d /tmp --output=target 2>/dev/null || echo /tmp)"
|
||||||
|
echo " 临时绕过: TMPDIR=\$PWD/.tmp-audit bash \$0"
|
||||||
|
} >&2
|
||||||
|
[ -n "$tmp_probe" ] && rm -f "$tmp_probe" 2>/dev/null
|
||||||
|
exit 2
|
||||||
|
fi
|
||||||
|
rm -f "$tmp_probe"
|
||||||
|
|
||||||
|
# ─────────────────────────────────────────────────────────────
|
||||||
|
# 0. 先证明「这条判据本身能命中」——否则后面所有 0 都不可信
|
||||||
|
# ─────────────────────────────────────────────────────────────
|
||||||
|
echo "=== 0. 判据自检(命中 0 时不许把 0 写进结论)==="
|
||||||
|
|
||||||
|
# ⚠ 这条路径的 query 有五种形态,见下。**不要假设参数顺序**。
|
||||||
|
# ?workspace=…&session_id=… / ?session_id=…(无 workspace) / ?session_id=…&workspace=…
|
||||||
|
# /thread?workspace=…&session_id=… / /thread?offset=…&session_id=…
|
||||||
|
# 写成 grep '…?workspace' 会静默漏掉 43 条(2026-09-28 实测)。
|
||||||
|
#
|
||||||
|
# ★★ 另一个坑(这个脚本第一版就踩了,且症状与「没投出」**完全一样**):
|
||||||
|
# access log 的行尾形状是
|
||||||
|
# "GET <url> HTTP/1.1" from 127.0.0.1:PORT - 200 956B in 9.17ms
|
||||||
|
# URL 与状态码之间隔着 `" from 127.0.0.1:PORT`。
|
||||||
|
# ⇒ 写成 `\?[^ ]* - 200` **匹配不到**(`[^ ]*` 撞在空格上就停了),
|
||||||
|
# 第 1 档会静默输出「取信请求 0 条」。
|
||||||
|
# ⇒ 必须用 `\?.* - ` 让它跨过那段。
|
||||||
|
FETCH_RE='GET http://[^ ]*/api/v1/agent/mail/[0-9a-f-]{36}\?.* - ([0-9]{3}) ([0-9]+)B'
|
||||||
|
# /thread 单独排除:它取的是往来树,不是正文。
|
||||||
|
THREAD_RE='agent/mail/[0-9a-f-]{36}/thread\?'
|
||||||
|
|
||||||
|
log_all() { journalctl -u "$UNIT" --since "$SINCE" --until "$UNTIL" --no-pager 2>/dev/null; }
|
||||||
|
|
||||||
|
probe=$(log_all | grep -cE 'api/v1/agent/mail/[0-9a-f-]{36}\?' || true)
|
||||||
|
if [ "$probe" -eq 0 ]; then
|
||||||
|
cat >&2 <<'EOF'
|
||||||
|
❌ 自检失败:该窗口内 `/api/v1/agent/mail/{id}?` 一条都没有。
|
||||||
|
在断定「没投出」之前,先排除这三种「静默归零」:
|
||||||
|
1) 路径写错(人类侧 /mail/{id} 本来就没有流量,Agent 走 /agent/mail/{id})
|
||||||
|
2) 窗口不对(★ DB 存 UTC、journalctl 打印本地时区,本项目差 8 小时;
|
||||||
|
按本地时间写 SQL 会整段落空 —— 10:09 本地 = 02:09 UTC)
|
||||||
|
3) unit 名不对(systemd 服务名不是 agentmail-gateway)
|
||||||
|
三者都排除后,才可以说「0」。
|
||||||
|
EOF
|
||||||
|
exit 3
|
||||||
|
fi
|
||||||
|
echo " 自检通过:该窗口有 $probe 条 /agent/mail/{id}? 取信请求(判据能命中)"
|
||||||
|
cat <<'EOF'
|
||||||
|
口径(2026-09-28 那轮实测,用来对照别把数读歪):
|
||||||
|
全部 /agent/mail/*(含 /thread) 347
|
||||||
|
其中非 thread 203 ← 真正的「取全文」
|
||||||
|
非 thread 且 UUID 合法 202 ← 本脚本报的数
|
||||||
|
差 1 条 = UUID 畸形的请求(服务端 400)
|
||||||
|
非 thread 的 query 形态:?workspace=…&session_id=… 159 / ?session_id=…(无 workspace)43
|
||||||
|
/ ?session_id=…&workspace=… 1
|
||||||
|
EOF
|
||||||
|
|
||||||
|
# ─────────────────────────────────────────────────────────────
|
||||||
|
# 1. 读数 ③ 取信是否成功(含「是否被归档阻断」这条独立判据)
|
||||||
|
# ─────────────────────────────────────────────────────────────
|
||||||
|
echo
|
||||||
|
echo "=== 1. 读数 ③:取信(GET /agent/mail/{id},不含 /thread)==="
|
||||||
|
|
||||||
|
tmp_fetches=$(mktemp) || { echo "❌ mktemp 失败(临时目录不可写)" >&2; exit 2; }
|
||||||
|
trap 'rm -f "$tmp_fetches"' EXIT
|
||||||
|
log_all | grep -E "$FETCH_RE" | grep -vE "$THREAD_RE" \
|
||||||
|
| sed -E 's/.* ([0-9]{2}:[0-9]{2}:[0-9]{2}).*agent\/mail\/([0-9a-f-]{36})\?.* - ([0-9]{3}) ([0-9]+)B.*/\1 \2 \3 \4/' \
|
||||||
|
> "$tmp_fetches" || { echo "❌ 写入 $tmp_fetches 失败(磁盘满?)——**不是「没投出」**" >&2; exit 2; }
|
||||||
|
# ⚠ 不要加 `|| true`:那会把「写失败」和「窗口内一条都没有」压成同一个 0。
|
||||||
|
# grep 无匹配本来就会 exit 1,那是**合法**的 0 —— 交给第 0 档自检去判。
|
||||||
|
|
||||||
|
total=$(wc -l < "$tmp_fetches" | tr -d ' ')
|
||||||
|
echo " 取信请求 $total 条,状态码分布:"
|
||||||
|
awk '{print " " $3}' "$tmp_fetches" | sort | uniq -c | sort -rn
|
||||||
|
|
||||||
|
ok200=$(awk '$3==200' "$tmp_fetches" | wc -l | tr -d ' ')
|
||||||
|
not200=$(awk '$3!=200' "$tmp_fetches" | wc -l | tr -d ' ')
|
||||||
|
echo " 200 = $ok200,非 200 = $not200"
|
||||||
|
|
||||||
|
if [ "$not200" -gt 0 ]; then
|
||||||
|
echo " ⚠ 非 200 的 mail_id 是否还在库里?**不在 ⇒ 是发信失败回滚(DeleteMailByID),与归档无关**"
|
||||||
|
echo " (2026-09-28 实测 4 条即如此:3×404 + 1×400,那 4 个 mail_id 库里已不存在)"
|
||||||
|
fi
|
||||||
|
|
||||||
|
# ─────────────────────────────────────────────────────────────
|
||||||
|
# 2. 读数 ①:投递成立(mail_reads = markDelivered 指纹)
|
||||||
|
# ─────────────────────────────────────────────────────────────
|
||||||
|
echo
|
||||||
|
echo "=== 2. 读数 ①:投递成立(mail_reads 行)==="
|
||||||
|
if [ -n "$ALIASES" ]; then
|
||||||
|
inlist=$(printf ",%s," "$ALIASES" | sed 's/^,//; s/,$//' | sed "s/'/''/g; s/[^,']*/'&'/g")
|
||||||
|
scope="s.session_alias IN ($inlist)"
|
||||||
|
else
|
||||||
|
# ⚠ 别名有「带字面引号」和「不带」两种形态(2026-09-28:33 条含 ",其中 26 条是 stress-sse)。
|
||||||
|
# 只查其中一种会得到一个**真实但无关**的数字。LIKE '%…%' 两种都覆盖。
|
||||||
|
scope="s.session_alias LIKE '%stress-sse%'"
|
||||||
|
fi
|
||||||
|
|
||||||
|
sqlite3 -header -column "$DB" "
|
||||||
|
SELECT
|
||||||
|
COUNT(*) AS 会话数,
|
||||||
|
SUM(EXISTS(SELECT 1 FROM mails m WHERE m.session_id=s.session_id)) AS 有信,
|
||||||
|
SUM(EXISTS(SELECT 1 FROM mails m WHERE m.session_id=s.session_id
|
||||||
|
AND EXISTS(SELECT 1 FROM mail_reads r WHERE r.mail_id=m.mail_id))) AS 有投递标记
|
||||||
|
FROM sessions s WHERE $scope;"
|
||||||
|
echo " ⇒ 有投递标记 = 投递成立(markDelivered 紧跟投递打,间隔 ~4~12ms)"
|
||||||
|
|
||||||
|
# ─────────────────────────────────────────────────────────────
|
||||||
|
# 3. 读数 ③的强度:响应大小 vs body 长度
|
||||||
|
# ★ 别把「~950B」当逐封正文到手的证明 —— 那是**路径的通用信封大小**。
|
||||||
|
# 真判据是**相关性**:若 body 不在响应里,r 应 ≈ 0。
|
||||||
|
# ─────────────────────────────────────────────────────────────
|
||||||
|
echo
|
||||||
|
echo "=== 3. 读数 ③的强度:pearson(body 长度, 响应字节数) ==="
|
||||||
|
echo " (2026-09-28 实测 n=164 时 r=0.9968 ⇒ 响应确实带 body)"
|
||||||
|
|
||||||
|
if command -v python3 >/dev/null; then
|
||||||
|
python3 - "$DB" "$tmp_fetches" <<'PY'
|
||||||
|
import sqlite3, sys, math
|
||||||
|
db = sqlite3.connect(sys.argv[1]); pairs = []
|
||||||
|
for line in open(sys.argv[2]):
|
||||||
|
p = line.split()
|
||||||
|
if len(p) != 4 or p[2] != '200':
|
||||||
|
continue
|
||||||
|
r = db.execute("SELECT length(body) FROM mails WHERE mail_id=?", (p[1],)).fetchone()
|
||||||
|
if r and r[0] is not None:
|
||||||
|
pairs.append((r[0], int(p[3])))
|
||||||
|
if len(pairs) < 8:
|
||||||
|
print(f" 样本不足(n={len(pairs)},需 ≥8)⇒ 本档**不出读数**,别硬凑")
|
||||||
|
sys.exit(0)
|
||||||
|
n = len(pairs); mx = sum(a for a, _ in pairs)/n; my = sum(b for _, b in pairs)/n
|
||||||
|
cov = sum((a-mx)*(b-my) for a, b in pairs)
|
||||||
|
sx = math.sqrt(sum((a-mx)**2 for a, _ in pairs)); sy = math.sqrt(sum((b-my)**2 for _, b in pairs))
|
||||||
|
r = cov/(sx*sy) if sx and sy else 0.0
|
||||||
|
print(f" n={n} r={r:.4f} body {min(a for a,_ in pairs)}..{max(a for a,_ in pairs)}B"
|
||||||
|
f" 响应 {min(b for _,b in pairs)}..{max(b for _,b in pairs)}B")
|
||||||
|
print(" ⇒ r 高 ⇒ 响应带 body(非空壳)。**这证明「服务端发出了 body」,"
|
||||||
|
"不证明「worker 交给了模型」** —— 后者只有桥侧证据。")
|
||||||
|
PY
|
||||||
|
else
|
||||||
|
echo " (无 python3,跳过本档)"
|
||||||
|
fi
|
||||||
|
|
||||||
|
# ─────────────────────────────────────────────────────────────
|
||||||
|
# 4. 读数 ②:起了 worker(桥日志「命名同步」)
|
||||||
|
# ─────────────────────────────────────────────────────────────
|
||||||
|
echo
|
||||||
|
echo "=== 4. 读数 ②:起了 worker(桥日志「命名同步」)==="
|
||||||
|
if [ -n "$BRIDGE_LOG" ] && [ -r "$BRIDGE_LOG" ]; then
|
||||||
|
grep -c "命名同步" "$BRIDGE_LOG" | sed 's/^/ 命名同步条数:/'
|
||||||
|
echo " ⚠ 这只证明 worker 起来了,**不证明它取到了正文**(那是上面第 3 档)"
|
||||||
|
else
|
||||||
|
echo " 未提供 BRIDGE_LOG,跳过。传 BRIDGE_LOG=<桥日志路径> 启用。"
|
||||||
|
echo " ⚠ 没有这一档时,② 与 ③ 不可互相替代 —— 两者是不同的失败模式。"
|
||||||
|
fi
|
||||||
|
|
||||||
|
# ─────────────────────────────────────────────────────────────
|
||||||
|
# 5. 归档这条独立判据(不与上面任何一档混)
|
||||||
|
# ─────────────────────────────────────────────────────────────
|
||||||
|
echo
|
||||||
|
echo "=== 5. 归档:别名寻址是否已死 ==="
|
||||||
|
sqlite3 -header -column "$DB" "
|
||||||
|
SELECT s.status AS 会话状态, COUNT(*) AS 条数,
|
||||||
|
SUM(NOT EXISTS(SELECT 1 FROM mails m WHERE m.session_id=s.session_id
|
||||||
|
AND m.status <> 'archived')) AS 会案内仍有非归档邮件
|
||||||
|
FROM sessions s WHERE $scope GROUP BY 1;"
|
||||||
|
cat <<'EOF'
|
||||||
|
⇒ 取信(AgentGetMail → AgentMayReadSession)**不**看 sessions.status ⇒ 归档不阻断取信
|
||||||
|
(判据在源码:repo/session_workspace.go 的 AgentMayReadSession 只比 scope == target)。
|
||||||
|
⇒ 但**收件箱**(ListInboxScoped)与**别名寻址**(FindNamedSessionFor)都带 status <> 'archived'
|
||||||
|
⇒ 归档后原地址是**死信地址,回信投不进去**。后果更重,别与「取不到信」混为一谈。
|
||||||
|
EOF
|
||||||
|
# ─────────────────────────────────────────────────────────────
|
||||||
|
# 6. 半活会话:归档是否被「复活」
|
||||||
|
# 签名唯一 —— 旧 TouchSession 只盖 sessions.status,不碰 mails;
|
||||||
|
# ArchiveSession 两条都盖。⇒ sessions=active 而 mails=archived 只可能来自
|
||||||
|
# 「先归档、后 touch」。这比读 updated_at 可靠:updated_at 会被任何写入改,
|
||||||
|
# 这张两表状态分叉不会。
|
||||||
|
# ─────────────────────────────────────────────────────────────
|
||||||
|
echo
|
||||||
|
echo "=== 6. 半活会话(sess=active 但 mail=archived ⇒ 归档后被复活)==="
|
||||||
|
sqlite3 -header -column "$DB" "
|
||||||
|
SELECT s.session_alias, s.updated_at, COUNT(*) AS 归档邮件数
|
||||||
|
FROM sessions s JOIN mails m ON m.session_id = s.session_id
|
||||||
|
WHERE $scope AND s.status = 'active' AND m.status = 'archived'
|
||||||
|
GROUP BY 1,2 ORDER BY s.updated_at;"
|
||||||
|
half=$(sqlite3 "$DB" "
|
||||||
|
SELECT COUNT(*) FROM (SELECT 1 FROM sessions s JOIN mails m ON m.session_id=s.session_id
|
||||||
|
WHERE $scope AND s.status='active' AND m.status='archived' GROUP BY s.session_id);")
|
||||||
|
echo " 半活会话数:$half"
|
||||||
|
if [ "$half" = "0" ]; then
|
||||||
|
echo " ⇒ 该范围内无半活会话。"
|
||||||
|
else
|
||||||
|
cat <<'EOF'
|
||||||
|
⚠ **非 0 就说明归档被绕过了**:这批会话的 mails 已是 archived(归档确实发生过),
|
||||||
|
但 sessions.status 又是 active。后果 = 别名寻址/收件箱里它又「活着」,
|
||||||
|
而它的信已被归档 —— **「已归档」与「还活着」并存**。
|
||||||
|
⇒ 别用 `sessions.status` 单表判断归档是否生效,要连 mails.status 一起看。
|
||||||
|
EOF
|
||||||
|
fi
|
||||||
|
|
||||||
|
# ─────────────────────────────────────────────────────────────
|
||||||
|
# 6b. ★ 反向分叉:sess=archived 而 mail<>'archived'
|
||||||
|
# 这是 6 的**镜像**,而且**成因至今是活的**:
|
||||||
|
# deploy/archive-stale-sessions.sh:105 归档时只 UPDATE sessions,
|
||||||
|
# 同一个事务里没有 UPDATE mails ⇒ 每跑一次 sweeper 就多一批。
|
||||||
|
# 它的回滚脚本(:147)同样只写 sessions ⇒ 回滚会把这些邮件一次性放回收件箱。
|
||||||
|
# ⚠ session_status_invariant_test.go:70 查的正是这个方向,但只在 t.TempDir()
|
||||||
|
# 的空库上跑 ⇒ 存量分叉它永远看不见(实测生产已有 28 会话 / 431 封)。
|
||||||
|
# 两个方向要一起查:s.status 与 mails.status 的四种组合里,
|
||||||
|
# 只有「全 active」和「全 archived」自洽,另两种都是分叉。
|
||||||
|
# ─────────────────────────────────────────────────────────────
|
||||||
|
echo
|
||||||
|
echo "=== 6b. 反向分叉(sess=archived 但有非 archived 邮件 ⇒ 归档没盖到 mails)==="
|
||||||
|
sqlite3 -header -column "$DB" "
|
||||||
|
SELECT s.session_alias,
|
||||||
|
COUNT(*) AS 总邮件,
|
||||||
|
SUM(m.status <> 'archived') AS 非归档邮件,
|
||||||
|
SUM(m.status = 'unread') AS 其中未读
|
||||||
|
FROM sessions s JOIN mails m ON m.session_id = s.session_id
|
||||||
|
WHERE $scope AND s.status = 'archived' AND m.status <> 'archived'
|
||||||
|
GROUP BY 1 ORDER BY 非归档邮件 DESC;"
|
||||||
|
rev=$(sqlite3 "$DB" "
|
||||||
|
SELECT COUNT(*) FROM (SELECT 1 FROM sessions s JOIN mails m ON m.session_id=s.session_id
|
||||||
|
WHERE $scope AND s.status='archived' AND m.status<>'archived' GROUP BY s.session_id);")
|
||||||
|
revmail=$(sqlite3 "$DB" "
|
||||||
|
SELECT COUNT(*) FROM sessions s JOIN mails m ON m.session_id=s.session_id
|
||||||
|
WHERE $scope AND s.status='archived' AND m.status<>'archived';")
|
||||||
|
echo " 反向分叉会话数:$rev 涉及非归档邮件:$revmail 封"
|
||||||
|
if [ "$rev" = "0" ]; then
|
||||||
|
echo " ⇒ 该范围内无反向分叉。"
|
||||||
|
else
|
||||||
|
cat <<'EOF'
|
||||||
|
⚠ **非 0 说明归档只盖了 sessions、没盖 mails**。后果比半活更直接:
|
||||||
|
· 收件箱 ListInboxScoped 带 `s.status <> 'archived'` ⇒ 这些信**完全不可见**;
|
||||||
|
· 别名寻址 FindNamedSessionFor 同样带该谓词 ⇒ **回信投不进去**(死信地址);
|
||||||
|
· 但 sweeper 的回滚脚本只把 sessions 改回 active ⇒ 回滚会**一次性把它们全放回来**,
|
||||||
|
「回滚」在这里是放大器而不是修复。
|
||||||
|
⇒ 归档语义必须两表一起判:`ArchiveSession` 已经是对的(两条 UPDATE),
|
||||||
|
缺的是 `deploy/archive-stale-sessions.sh`(归档与回滚都只写 sessions)。
|
||||||
|
EOF
|
||||||
|
fi
|
||||||
Reference in New Issue
Block a user