chore(format): 撤销误入提交的整体重排，并关闭本仓库的格式化器
# 发生了什么 pi-lens 内置「安全格式化」：它会自动安装 biome 并对**编辑过的文件**跑 `biome format --write`。本机原先没有任何 biome 配置，于是 biome 用它自己的 默认值 —— tab 缩进 + 双引号 —— 把文件整体重写。 我在19a3161那次提交里用了 `git add -A`，把这批与功能无关的重排一起扫了进去： 约 7000 行改动散落在 20 个文件上，使那次提交无法审查，还掩盖了 server/internal/handler/permission.go 的一处删行（实为文件末尾空行，无代码丢失）。 # 为什么是「关掉」而不是「配置成我们的风格」 试过把缩进/引号/lineWidth 全部对齐本仓库习惯（biome.json + space/2/single/ lineWidth 120）：`biome format --write` 仍然改动 17 个文件。原因是本仓库从未按 biome 的规则排版过 —— 注释按语义换行、数组与调用按可读性手工折行， 这些无法由格式化器还原。也就是说只要格式化器开着，每次编辑都会产生与内容无关的 大面积 diff，把真正的改动埋掉。 因此 biome.jsonc 里 formatter 与 linter 都关闭：本仓库的静态检查由 tsc / go vet / tree-sitter / ast-grep 与各自测试套件承担，不引入会改动无关行的 自动修复。 （pi-lens 这一版把 format 服务的 enabled 硬编码为 true，没有配置开关， 所以只能在仓库侧用 biome 配置让它不动文件；已验证 `biome format --write` 对这些文件零改动。） # 本提交内容 把19a3161里除「有意改动」外的 20 个文件还原到重排前的样子。19a3161中真正有意的改动是 deploy/install.sh 的扩展注册与 plugins/pi-mail-bridge/extension/index.ts 新文件，两者原样保留。 验证：Go 全量、三桥插件（320/362/409）、前端 196 全绿； `biome format --write` 对还原后的文件零改动。
This commit is contained in:
@ -5,17 +5,16 @@
|
||||
* 本文件只管认证头、密钥解析与坐标变更。
|
||||
*/
|
||||
|
||||
import { readFileSync, writeFileSync, mkdirSync, existsSync } from "node:fs";
|
||||
import { randomBytes } from "node:crypto";
|
||||
import { homedir } from "node:os";
|
||||
import { join } from "node:path";
|
||||
import { readFileSync, writeFileSync, mkdirSync, existsSync } from 'node:fs';
|
||||
import { randomBytes } from 'node:crypto';
|
||||
import { homedir } from 'node:os';
|
||||
import { join } from 'node:path';
|
||||
|
||||
import { createSSEClient } from "../lib/sse-client.js";
|
||||
import { createSSEClient } from '../lib/sse-client.js';
|
||||
|
||||
const CONFIG_DIR =
|
||||
process.env.AGENTMAIL_CONFIG_DIR || join(homedir(), ".agentmail");
|
||||
export const KEY_FILE = join(CONFIG_DIR, "agent.key");
|
||||
const CONFIG_FILE = join(CONFIG_DIR, "config.json");
|
||||
const CONFIG_DIR = process.env.AGENTMAIL_CONFIG_DIR || join(homedir(), '.agentmail');
|
||||
export const KEY_FILE = join(CONFIG_DIR, 'agent.key');
|
||||
const CONFIG_FILE = join(CONFIG_DIR, 'config.json');
|
||||
|
||||
/**
|
||||
* 把管理员给的密钥落盘（0600）。
|
||||
@ -24,29 +23,23 @@ const CONFIG_FILE = join(CONFIG_DIR, "config.json");
|
||||
* 否则重启后又回到无法连接的状态 —— 而那正是这个工具要解决的问题。
|
||||
*/
|
||||
export function saveLocalKey(token) {
|
||||
mkdirSync(CONFIG_DIR, { recursive: true, mode: 0o700 });
|
||||
writeFileSync(
|
||||
KEY_FILE,
|
||||
JSON.stringify(
|
||||
{ key_token: token, created_at: new Date().toISOString() },
|
||||
null,
|
||||
2,
|
||||
),
|
||||
{ mode: 0o600 },
|
||||
);
|
||||
mkdirSync(CONFIG_DIR, { recursive: true, mode: 0o700 });
|
||||
writeFileSync(
|
||||
KEY_FILE,
|
||||
JSON.stringify({ key_token: token, created_at: new Date().toISOString() }, null, 2),
|
||||
{ mode: 0o600 },
|
||||
);
|
||||
}
|
||||
|
||||
/** 读取本地密钥文件；不存在或损坏时返回 null。 */
|
||||
export function readLocalKey() {
|
||||
try {
|
||||
if (!existsSync(KEY_FILE)) return null;
|
||||
const raw = JSON.parse(readFileSync(KEY_FILE, "utf8"));
|
||||
return typeof raw?.key_token === "string" && raw.key_token
|
||||
? raw.key_token
|
||||
: null;
|
||||
} catch {
|
||||
return null;
|
||||
}
|
||||
try {
|
||||
if (!existsSync(KEY_FILE)) return null;
|
||||
const raw = JSON.parse(readFileSync(KEY_FILE, 'utf8'));
|
||||
return typeof raw?.key_token === 'string' && raw.key_token ? raw.key_token : null;
|
||||
} catch {
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
@ -56,187 +49,164 @@ export function readLocalKey() {
|
||||
* 走 console.error 而不是任何结构化日志 —— 它一定进 journalctl（契约 9.8）。
|
||||
*/
|
||||
export function generateLocalKey(log = console.error) {
|
||||
const token = randomBytes(32).toString("hex");
|
||||
mkdirSync(CONFIG_DIR, { recursive: true, mode: 0o700 });
|
||||
writeFileSync(
|
||||
KEY_FILE,
|
||||
JSON.stringify(
|
||||
{ key_token: token, created_at: new Date().toISOString() },
|
||||
null,
|
||||
2,
|
||||
),
|
||||
{ mode: 0o600 },
|
||||
);
|
||||
// 调用方传进来的 log 已经带 [pi-mail-bridge] 前缀，这里不再自己加
|
||||
log(`已在 ${KEY_FILE} 生成本地密钥。`);
|
||||
log(`该密钥需管理员在 AgentMail 后台登记后才能接入：`);
|
||||
log(` ${token}`);
|
||||
return token;
|
||||
const token = randomBytes(32).toString('hex');
|
||||
mkdirSync(CONFIG_DIR, { recursive: true, mode: 0o700 });
|
||||
writeFileSync(
|
||||
KEY_FILE,
|
||||
JSON.stringify({ key_token: token, created_at: new Date().toISOString() }, null, 2),
|
||||
{ mode: 0o600 },
|
||||
);
|
||||
// 调用方传进来的 log 已经带 [pi-mail-bridge] 前缀，这里不再自己加
|
||||
log(`已在 ${KEY_FILE} 生成本地密钥。`);
|
||||
log(`该密钥需管理员在 AgentMail 后台登记后才能接入：`);
|
||||
log(` ${token}`);
|
||||
return token;
|
||||
}
|
||||
|
||||
/** 把 gateway 地址与身份记到 config.json，便于换机时人工核对。 */
|
||||
export function saveConfig(extra) {
|
||||
try {
|
||||
mkdirSync(CONFIG_DIR, { recursive: true, mode: 0o700 });
|
||||
let cur = {};
|
||||
if (existsSync(CONFIG_FILE)) {
|
||||
try {
|
||||
cur = JSON.parse(readFileSync(CONFIG_FILE, "utf8"));
|
||||
} catch {
|
||||
/* 损坏就重写 */
|
||||
}
|
||||
}
|
||||
writeFileSync(CONFIG_FILE, JSON.stringify({ ...cur, ...extra }, null, 2), {
|
||||
mode: 0o600,
|
||||
});
|
||||
} catch (e) {
|
||||
console.error("[pi-mail-bridge] 写 config.json 失败:", e?.message || e);
|
||||
}
|
||||
try {
|
||||
mkdirSync(CONFIG_DIR, { recursive: true, mode: 0o700 });
|
||||
let cur = {};
|
||||
if (existsSync(CONFIG_FILE)) {
|
||||
try { cur = JSON.parse(readFileSync(CONFIG_FILE, 'utf8')); } catch { /* 损坏就重写 */ }
|
||||
}
|
||||
writeFileSync(CONFIG_FILE, JSON.stringify({ ...cur, ...extra }, null, 2), { mode: 0o600 });
|
||||
} catch (e) {
|
||||
console.error('[pi-mail-bridge] 写 config.json 失败:', e?.message || e);
|
||||
}
|
||||
}
|
||||
|
||||
export class GatewayClient {
|
||||
/**
|
||||
* @param {{url: string, agentName: string, agentKey: string, agentSecret: string}} opts
|
||||
*/
|
||||
constructor({ url, agentName, agentKey, agentSecret }) {
|
||||
this.baseURL = String(url || "http://127.0.0.1:8180").replace(/\/+$/, "");
|
||||
this.agentName = agentName;
|
||||
this.agentKey = agentKey || "";
|
||||
this.agentSecret = agentSecret || "";
|
||||
this.sseClient = null;
|
||||
}
|
||||
/**
|
||||
* @param {{url: string, agentName: string, agentKey: string, agentSecret: string}} opts
|
||||
*/
|
||||
constructor({ url, agentName, agentKey, agentSecret }) {
|
||||
this.baseURL = String(url || 'http://127.0.0.1:8180').replace(/\/+$/, '');
|
||||
this.agentName = agentName;
|
||||
this.agentKey = agentKey || '';
|
||||
this.agentSecret = agentSecret || '';
|
||||
this.sseClient = null;
|
||||
}
|
||||
|
||||
/** 认证头：有密钥走 Bearer，否则退回 name/secret。 */
|
||||
authHeaders() {
|
||||
if (this.agentKey) {
|
||||
return {
|
||||
Authorization: `Bearer ${this.agentKey}`,
|
||||
"X-Agent-Name": this.agentName,
|
||||
};
|
||||
}
|
||||
return {
|
||||
"X-Agent-Name": this.agentName,
|
||||
"X-Agent-Secret": this.agentSecret,
|
||||
};
|
||||
}
|
||||
/** 认证头：有密钥走 Bearer，否则退回 name/secret。 */
|
||||
authHeaders() {
|
||||
if (this.agentKey) {
|
||||
return { Authorization: `Bearer ${this.agentKey}`, 'X-Agent-Name': this.agentName };
|
||||
}
|
||||
return { 'X-Agent-Name': this.agentName, 'X-Agent-Secret': this.agentSecret };
|
||||
}
|
||||
|
||||
async get(path) {
|
||||
const res = await fetch(`${this.baseURL}/api/v1${path}`, {
|
||||
headers: this.authHeaders(),
|
||||
});
|
||||
if (!res.ok) throw new Error(`GET ${path} 失败: HTTP ${res.status}`);
|
||||
return res.json();
|
||||
}
|
||||
async get(path) {
|
||||
const res = await fetch(`${this.baseURL}/api/v1${path}`, { headers: this.authHeaders() });
|
||||
if (!res.ok) throw new Error(`GET ${path} 失败: HTTP ${res.status}`);
|
||||
return res.json();
|
||||
}
|
||||
|
||||
async post(path, body) {
|
||||
const res = await fetch(`${this.baseURL}/api/v1${path}`, {
|
||||
method: "POST",
|
||||
headers: { "Content-Type": "application/json", ...this.authHeaders() },
|
||||
body: JSON.stringify(body),
|
||||
});
|
||||
const data = await res.json().catch(() => ({}));
|
||||
if (!res.ok) {
|
||||
const err = new Error(
|
||||
data?.error || `POST ${path} 失败: HTTP ${res.status}`,
|
||||
);
|
||||
err.status = res.status;
|
||||
// 响应体也带上：服务端对 409 会给 detail/suggestion，
|
||||
// 那些文字要原文转给模型（它据此决定换什么做法）。
|
||||
err.body = data;
|
||||
throw err;
|
||||
}
|
||||
return data;
|
||||
}
|
||||
async post(path, body) {
|
||||
const res = await fetch(`${this.baseURL}/api/v1${path}`, {
|
||||
method: 'POST',
|
||||
headers: { 'Content-Type': 'application/json', ...this.authHeaders() },
|
||||
body: JSON.stringify(body),
|
||||
});
|
||||
const data = await res.json().catch(() => ({}));
|
||||
if (!res.ok) {
|
||||
const err = new Error(data?.error || `POST ${path} 失败: HTTP ${res.status}`);
|
||||
err.status = res.status;
|
||||
// 响应体也带上：服务端对 409 会给 detail/suggestion，
|
||||
// 那些文字要原文转给模型（它据此决定换什么做法）。
|
||||
err.body = data;
|
||||
throw err;
|
||||
}
|
||||
return data;
|
||||
}
|
||||
|
||||
/** 注册。workspaces 传 []（B-1.2）—— 工作目录由每封邮件的 to_workspace 决定。 */
|
||||
async register() {
|
||||
return this.post("/agent/register", {
|
||||
name: this.agentName,
|
||||
secret: this.agentSecret || "",
|
||||
workspaces: [],
|
||||
platform: "pi",
|
||||
});
|
||||
}
|
||||
/** 注册。workspaces 传 []（B-1.2）—— 工作目录由每封邮件的 to_workspace 决定。 */
|
||||
async register() {
|
||||
return this.post('/agent/register', {
|
||||
name: this.agentName,
|
||||
secret: this.agentSecret || '',
|
||||
workspaces: [],
|
||||
platform: 'pi',
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* 上传附件。
|
||||
*
|
||||
* 必须走 multipart 的 `file` 字段：服务端是 `r.FormFile("file")`，
|
||||
* 且**不认 `X-Filename` 头**（grep 过 handler/attachments.go，没有这个分支）。
|
||||
* 直接 POST 二进制体会得到 400「缺少 file 字段」。
|
||||
*
|
||||
* 不手动设 Content-Type：让 undici 按 FormData 自己生成 boundary。
|
||||
*/
|
||||
async uploadFile(buf, filename) {
|
||||
const form = new FormData();
|
||||
form.append("file", new Blob([buf]), filename);
|
||||
const res = await fetch(`${this.baseURL}/api/v1/attachments`, {
|
||||
method: "POST",
|
||||
headers: this.authHeaders(),
|
||||
body: form,
|
||||
});
|
||||
const data = await res.json().catch(() => ({}));
|
||||
if (!res.ok) throw new Error(data?.error || `上传失败: HTTP ${res.status}`);
|
||||
return data.attachment;
|
||||
}
|
||||
/**
|
||||
* 上传附件。
|
||||
*
|
||||
* 必须走 multipart 的 `file` 字段：服务端是 `r.FormFile("file")`，
|
||||
* 且**不认 `X-Filename` 头**（grep 过 handler/attachments.go，没有这个分支）。
|
||||
* 直接 POST 二进制体会得到 400「缺少 file 字段」。
|
||||
*
|
||||
* 不手动设 Content-Type：让 undici 按 FormData 自己生成 boundary。
|
||||
*/
|
||||
async uploadFile(buf, filename) {
|
||||
const form = new FormData();
|
||||
form.append('file', new Blob([buf]), filename);
|
||||
const res = await fetch(`${this.baseURL}/api/v1/attachments`, {
|
||||
method: 'POST',
|
||||
headers: this.authHeaders(),
|
||||
body: form,
|
||||
});
|
||||
const data = await res.json().catch(() => ({}));
|
||||
if (!res.ok) throw new Error(data?.error || `上传失败: HTTP ${res.status}`);
|
||||
return data.attachment;
|
||||
}
|
||||
|
||||
async downloadFile(attachmentID) {
|
||||
const res = await fetch(
|
||||
`${this.baseURL}/api/v1/attachments/${attachmentID}`,
|
||||
{
|
||||
headers: this.authHeaders(),
|
||||
},
|
||||
);
|
||||
if (!res.ok) throw new Error(`下载失败: HTTP ${res.status}`);
|
||||
return Buffer.from(await res.arrayBuffer());
|
||||
}
|
||||
async downloadFile(attachmentID) {
|
||||
const res = await fetch(`${this.baseURL}/api/v1/attachments/${attachmentID}`, {
|
||||
headers: this.authHeaders(),
|
||||
});
|
||||
if (!res.ok) throw new Error(`下载失败: HTTP ${res.status}`);
|
||||
return Buffer.from(await res.arrayBuffer());
|
||||
}
|
||||
|
||||
/**
|
||||
* 建立 SSE 长连并自动重连。
|
||||
*
|
||||
* 实现委托给共用模块 `lib/sse-client.js`（三桥逐字节同源，由
|
||||
* deploy/check-shared-libs.sh 校验）—— 那里把「跨 TCP 分片保帧状态」与
|
||||
* 「Last-Event-ID 断点续传」两件事写对了一次，不必每个平台各抄一遍。
|
||||
*
|
||||
* 断线重连带 `Last-Event-ID`（D-7.2）：服务端有 per-agent 环形缓冲，
|
||||
* 能把断连期间的事件回放出来 —— 否则那段时间的邮件只能等下次重启补拉。
|
||||
* 首次连接**不带**（N-11）：那会让服务端把缓冲区里的旧事件全回放一遍。
|
||||
*/
|
||||
startSSE(onEvent, log = console.error) {
|
||||
this.sseClient?.stop?.();
|
||||
this.sseClient = createSSEClient({
|
||||
authHeaders: () => this.authHeaders(),
|
||||
baseURL: this.baseURL,
|
||||
path: "/api/v1/events/stream",
|
||||
onEvent,
|
||||
log,
|
||||
});
|
||||
}
|
||||
/**
|
||||
* 建立 SSE 长连并自动重连。
|
||||
*
|
||||
* 实现委托给共用模块 `lib/sse-client.js`（三桥逐字节同源，由
|
||||
* deploy/check-shared-libs.sh 校验）—— 那里把「跨 TCP 分片保帧状态」与
|
||||
* 「Last-Event-ID 断点续传」两件事写对了一次，不必每个平台各抄一遍。
|
||||
*
|
||||
* 断线重连带 `Last-Event-ID`（D-7.2）：服务端有 per-agent 环形缓冲，
|
||||
* 能把断连期间的事件回放出来 —— 否则那段时间的邮件只能等下次重启补拉。
|
||||
* 首次连接**不带**（N-11）：那会让服务端把缓冲区里的旧事件全回放一遍。
|
||||
*/
|
||||
startSSE(onEvent, log = console.error) {
|
||||
this.sseClient?.stop?.();
|
||||
this.sseClient = createSSEClient({
|
||||
authHeaders: () => this.authHeaders(),
|
||||
baseURL: this.baseURL,
|
||||
path: '/api/v1/events/stream',
|
||||
onEvent,
|
||||
log,
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* 换 Gateway 地址或换密钥。
|
||||
*
|
||||
* 守护进程不能靠重启来应用新配置 —— connect_to_server 是模型在**运行中**
|
||||
* 调的，它期望调完就能收信。所以这里除了改字段还要重置断点：
|
||||
* `lastEventID` 是**旧** Gateway 环形缓冲里的序号，拿去问新 Gateway 会
|
||||
* 命中一段完全无关的历史（或直接被拒），得到的事件属于别人的会话。
|
||||
*
|
||||
* @returns {boolean} 是否真的变了（没变就不必重连 SSE，省一次断流）
|
||||
*/
|
||||
reconfigure({ url, agentKey }) {
|
||||
const nextURL = url ? String(url).replace(/\/+$/, "") : this.baseURL;
|
||||
const nextKey = agentKey || this.agentKey;
|
||||
const changed = nextURL !== this.baseURL || nextKey !== this.agentKey;
|
||||
if (!changed) return false;
|
||||
/**
|
||||
* 换 Gateway 地址或换密钥。
|
||||
*
|
||||
* 守护进程不能靠重启来应用新配置 —— connect_to_server 是模型在**运行中**
|
||||
* 调的，它期望调完就能收信。所以这里除了改字段还要重置断点：
|
||||
* `lastEventID` 是**旧** Gateway 环形缓冲里的序号，拿去问新 Gateway 会
|
||||
* 命中一段完全无关的历史（或直接被拒），得到的事件属于别人的会话。
|
||||
*
|
||||
* @returns {boolean} 是否真的变了（没变就不必重连 SSE，省一次断流）
|
||||
*/
|
||||
reconfigure({ url, agentKey }) {
|
||||
const nextURL = url ? String(url).replace(/\/+$/, '') : this.baseURL;
|
||||
const nextKey = agentKey || this.agentKey;
|
||||
const changed = nextURL !== this.baseURL || nextKey !== this.agentKey;
|
||||
if (!changed) return false;
|
||||
|
||||
if (nextURL !== this.baseURL) this.sseClient?.reset?.();
|
||||
this.baseURL = nextURL;
|
||||
this.agentKey = nextKey;
|
||||
return true;
|
||||
}
|
||||
if (nextURL !== this.baseURL) this.sseClient?.reset?.();
|
||||
this.baseURL = nextURL;
|
||||
this.agentKey = nextKey;
|
||||
return true;
|
||||
}
|
||||
|
||||
stopSSE() {
|
||||
this.sseClient?.stop?.();
|
||||
this.sseClient = null;
|
||||
}
|
||||
stopSSE() {
|
||||
this.sseClient?.stop?.();
|
||||
this.sseClient = null;
|
||||
}
|
||||
}
|
||||
|
||||
@ -56,16 +56,12 @@
|
||||
* 一处只增不减的结构。
|
||||
*/
|
||||
|
||||
import { fork } from "node:child_process";
|
||||
import { fileURLToPath } from "node:url";
|
||||
import { fork } from 'node:child_process';
|
||||
import { fileURLToPath } from 'node:url';
|
||||
|
||||
import {
|
||||
BoundedMap,
|
||||
BoundedSet,
|
||||
MAX_TRACKED_SESSIONS,
|
||||
} from "../lib/bounded.js";
|
||||
import { BoundedMap, BoundedSet, MAX_TRACKED_SESSIONS } from '../lib/bounded.js';
|
||||
|
||||
const WORKER_PATH = fileURLToPath(new URL("./worker.mjs", import.meta.url));
|
||||
const WORKER_PATH = fileURLToPath(new URL('./worker.mjs', import.meta.url));
|
||||
|
||||
/**
|
||||
* @param {object} deps
|
||||
@ -81,328 +77,277 @@ const WORKER_PATH = fileURLToPath(new URL("./worker.mjs", import.meta.url));
|
||||
* 让调度不变量（并发上限、同会话串行、硬超时）能在毫秒级验证。
|
||||
*/
|
||||
export function createWorkerPool({
|
||||
log,
|
||||
config,
|
||||
onReconfigure,
|
||||
maxWorkers = 3,
|
||||
workerMaxMs = 600_000,
|
||||
maxAttempts = 3,
|
||||
workerPath = WORKER_PATH,
|
||||
log, config, onReconfigure,
|
||||
maxWorkers = 3, workerMaxMs = 600_000, maxAttempts = 3, workerPath = WORKER_PATH,
|
||||
}) {
|
||||
/** 正在跑的 worker：mailSessionKey -> {child, mailID, startedAt, timer} */
|
||||
const running = new Map();
|
||||
/** 等着派的活，先进先出。 */
|
||||
const queue = [];
|
||||
/** relay_key -> mailSessionKey，把决策路由回发起询问的那个 worker。 */
|
||||
const permissionRoutes = new Map();
|
||||
/**
|
||||
* 跨 worker 存活的会话状态：mailSessionKey -> {sessionFile, cwd, piSessionId,
|
||||
* grants:Set, lastSyncedName}。
|
||||
*
|
||||
* 这是 worker 一封一进程之后仍需在主进程留存的全部东西 —— 下一封邮件靠
|
||||
* sessionFile 接着谈，靠 grants 不重复问已经「一直同意」过的工具。
|
||||
*/
|
||||
const sessionState = new BoundedMap(MAX_TRACKED_SESSIONS);
|
||||
/**
|
||||
* 被模型降级换掉的旧 pi 会话 id。
|
||||
*
|
||||
* 仍要计入 mail_driven：它们已经参与过邮件往来，而磁盘上的会话文件
|
||||
* 不会因为换模型而消失 —— 心跳快照仍会上报它们。
|
||||
*/
|
||||
const retired = new BoundedSet(MAX_TRACKED_SESSIONS);
|
||||
let stopped = false;
|
||||
/** 正在跑的 worker：mailSessionKey -> {child, mailID, startedAt, timer} */
|
||||
const running = new Map();
|
||||
/** 等着派的活，先进先出。 */
|
||||
const queue = [];
|
||||
/** relay_key -> mailSessionKey，把决策路由回发起询问的那个 worker。 */
|
||||
const permissionRoutes = new Map();
|
||||
/**
|
||||
* 跨 worker 存活的会话状态：mailSessionKey -> {sessionFile, cwd, piSessionId,
|
||||
* grants:Set, lastSyncedName}。
|
||||
*
|
||||
* 这是 worker 一封一进程之后仍需在主进程留存的全部东西 —— 下一封邮件靠
|
||||
* sessionFile 接着谈，靠 grants 不重复问已经「一直同意」过的工具。
|
||||
*/
|
||||
const sessionState = new BoundedMap(MAX_TRACKED_SESSIONS);
|
||||
/**
|
||||
* 被模型降级换掉的旧 pi 会话 id。
|
||||
*
|
||||
* 仍要计入 mail_driven：它们已经参与过邮件往来，而磁盘上的会话文件
|
||||
* 不会因为换模型而消失 —— 心跳快照仍会上报它们。
|
||||
*/
|
||||
const retired = new BoundedSet(MAX_TRACKED_SESSIONS);
|
||||
let stopped = false;
|
||||
|
||||
/**
|
||||
* 邮件会话 id 作为串行化的键。
|
||||
*
|
||||
* 没有 session_id 的事件（理论上不该有）退回 mail_id：那样每封各占一个
|
||||
* worker，不会串行 —— 但它们本来也不属于同一条会话。
|
||||
*/
|
||||
const keyOf = (data) =>
|
||||
data?.session_id || `mail:${data?.mail_id || Math.random()}`;
|
||||
/**
|
||||
* 邮件会话 id 作为串行化的键。
|
||||
*
|
||||
* 没有 session_id 的事件（理论上不该有）退回 mail_id：那样每封各占一个
|
||||
* worker，不会串行 —— 但它们本来也不属于同一条会话。
|
||||
*/
|
||||
const keyOf = (data) => data?.session_id || `mail:${data?.mail_id || Math.random()}`;
|
||||
|
||||
function submit(kind, data, attempt = 1) {
|
||||
if (stopped) return;
|
||||
queue.push({ kind, data, key: keyOf(data), attempt });
|
||||
pump();
|
||||
}
|
||||
function submit(kind, data, attempt = 1) {
|
||||
if (stopped) return;
|
||||
queue.push({ kind, data, key: keyOf(data), attempt });
|
||||
pump();
|
||||
}
|
||||
|
||||
function pump() {
|
||||
if (stopped) return;
|
||||
for (let i = 0; i < queue.length; i++) {
|
||||
const job = queue[i];
|
||||
// 同一会话已有 worker 在跑 → 跳过它，看后面有没有别的会话可以先跑。
|
||||
// 不能 break：那会让一条慢会话把所有别的会话都堵住（正是要修的病）。
|
||||
if (running.has(job.key)) continue;
|
||||
if (running.size >= maxWorkers) return;
|
||||
queue.splice(i, 1);
|
||||
i--;
|
||||
spawn(job);
|
||||
}
|
||||
}
|
||||
function pump() {
|
||||
if (stopped) return;
|
||||
for (let i = 0; i < queue.length; i++) {
|
||||
const job = queue[i];
|
||||
// 同一会话已有 worker 在跑 → 跳过它，看后面有没有别的会话可以先跑。
|
||||
// 不能 break：那会让一条慢会话把所有别的会话都堵住（正是要修的病）。
|
||||
if (running.has(job.key)) continue;
|
||||
if (running.size >= maxWorkers) return;
|
||||
queue.splice(i, 1);
|
||||
i--;
|
||||
spawn(job);
|
||||
}
|
||||
}
|
||||
|
||||
function spawn(job) {
|
||||
const state = sessionState.get(job.key) || {
|
||||
grants: new Set(),
|
||||
lastSyncedName: "",
|
||||
};
|
||||
const child = fork(workerPath, [], {
|
||||
// stdio 继承：worker 里 pi SDK 自己打的东西直接进 journalctl。
|
||||
// 'ipc' 必须显式列出，否则 process.send 不存在。
|
||||
stdio: ["ignore", "inherit", "inherit", "ipc"],
|
||||
});
|
||||
function spawn(job) {
|
||||
const state = sessionState.get(job.key) || { grants: new Set(), lastSyncedName: '' };
|
||||
const child = fork(workerPath, [], {
|
||||
// stdio 继承：worker 里 pi SDK 自己打的东西直接进 journalctl。
|
||||
// 'ipc' 必须显式列出，否则 process.send 不存在。
|
||||
stdio: ['ignore', 'inherit', 'inherit', 'ipc'],
|
||||
});
|
||||
|
||||
// 硬超时：worker 卡死（模型不返回、权限等不到决策而主进程也没收到事件）
|
||||
// 时必须能回收，否则那条会话的后续邮件永远排队。
|
||||
const timer = setTimeout(() => {
|
||||
log(
|
||||
`worker ${child.pid} 处理 ${job.data?.mail_id} 超过 ${workerMaxMs / 1000}s，强杀`,
|
||||
);
|
||||
try {
|
||||
child.kill("SIGKILL");
|
||||
} catch {
|
||||
/* 已经死了 */
|
||||
}
|
||||
}, workerMaxMs);
|
||||
if (typeof timer.unref === "function") timer.unref();
|
||||
// 硬超时：worker 卡死（模型不返回、权限等不到决策而主进程也没收到事件）
|
||||
// 时必须能回收，否则那条会话的后续邮件永远排队。
|
||||
const timer = setTimeout(() => {
|
||||
log(`worker ${child.pid} 处理 ${job.data?.mail_id} 超过 ${workerMaxMs / 1000}s，强杀`);
|
||||
try { child.kill('SIGKILL'); } catch { /* 已经死了 */ }
|
||||
}, workerMaxMs);
|
||||
if (typeof timer.unref === 'function') timer.unref();
|
||||
|
||||
const entry = {
|
||||
child,
|
||||
mailID: job.data?.mail_id || "",
|
||||
key: job.key,
|
||||
startedAt: Date.now(),
|
||||
timer,
|
||||
settled: false,
|
||||
};
|
||||
running.set(job.key, entry);
|
||||
const entry = {
|
||||
child, mailID: job.data?.mail_id || '', key: job.key,
|
||||
startedAt: Date.now(), timer, settled: false,
|
||||
};
|
||||
running.set(job.key, entry);
|
||||
|
||||
child.on("message", (msg) => onWorkerMessage(entry, msg));
|
||||
child.on('message', (msg) => onWorkerMessage(entry, msg));
|
||||
|
||||
child.on("exit", (code, signal) => {
|
||||
clearTimeout(timer);
|
||||
running.delete(job.key);
|
||||
for (const [rk, k] of permissionRoutes)
|
||||
if (k === job.key) permissionRoutes.delete(rk);
|
||||
child.on('exit', (code, signal) => {
|
||||
clearTimeout(timer);
|
||||
running.delete(job.key);
|
||||
for (const [rk, k] of permissionRoutes) if (k === job.key) permissionRoutes.delete(rk);
|
||||
|
||||
// 没收到 `done` 就退出 = 这封邮件**从未处理完**。
|
||||
//
|
||||
// 这是生产上真实存在的静默丢信路径：worker 被 SIGKILL（硬超时）、
|
||||
// OOM、或自己崩溃时，`done` 永远不会到达，主进程只看到 exit code。
|
||||
// 原来这里只记一行日志就 pump() —— 发件人看到信发出去了，
|
||||
// 而那条会话再也不会有人回。
|
||||
//
|
||||
// 重投而不是直接由主进程回信：worker 崩溃可能是内存/上游瞬时故障，
|
||||
// 重启一个进程真能跑通。有界（maxAttempts）是因为「必定失败」的邮件
|
||||
// 无界重投会变成永久活锁，而日志里只有一行看不出是同一封在原地打转。
|
||||
if (!entry.settled && !stopped) {
|
||||
const attempt = job.attempt || 1;
|
||||
if (attempt < maxAttempts) {
|
||||
const delay = attempt * 1000;
|
||||
log(
|
||||
`worker ${child.pid}（mail ${entry.mailID}）未回报 done 就退出` +
|
||||
`（code=${code} signal=${signal || "-"}），${delay / 1000}s 后` +
|
||||
`第 ${attempt + 1}/${maxAttempts} 次重投`,
|
||||
);
|
||||
const retry = setTimeout(() => {
|
||||
if (stopped) return;
|
||||
queue.push({ ...job, attempt: attempt + 1 });
|
||||
pump();
|
||||
}, delay);
|
||||
if (typeof retry.unref === "function") retry.unref();
|
||||
// 退避期间不 pump：否则同一会话会被立刻重投，退避形同虚设
|
||||
return;
|
||||
}
|
||||
log(
|
||||
`worker ${child.pid}（mail ${entry.mailID}）重投 ${maxAttempts} 次仍未完成，放弃` +
|
||||
`（code=${code} signal=${signal || "-"}）`,
|
||||
);
|
||||
} else if (code !== 0) {
|
||||
log(
|
||||
`worker ${child.pid}（mail ${entry.mailID}）异常退出 code=${code} signal=${signal || "-"}`,
|
||||
);
|
||||
}
|
||||
pump();
|
||||
});
|
||||
// 没收到 `done` 就退出 = 这封邮件**从未处理完**。
|
||||
//
|
||||
// 这是生产上真实存在的静默丢信路径：worker 被 SIGKILL（硬超时）、
|
||||
// OOM、或自己崩溃时，`done` 永远不会到达，主进程只看到 exit code。
|
||||
// 原来这里只记一行日志就 pump() —— 发件人看到信发出去了，
|
||||
// 而那条会话再也不会有人回。
|
||||
//
|
||||
// 重投而不是直接由主进程回信：worker 崩溃可能是内存/上游瞬时故障，
|
||||
// 重启一个进程真能跑通。有界（maxAttempts）是因为「必定失败」的邮件
|
||||
// 无界重投会变成永久活锁，而日志里只有一行看不出是同一封在原地打转。
|
||||
if (!entry.settled && !stopped) {
|
||||
const attempt = job.attempt || 1;
|
||||
if (attempt < maxAttempts) {
|
||||
const delay = attempt * 1000;
|
||||
log(`worker ${child.pid}（mail ${entry.mailID}）未回报 done 就退出`
|
||||
+ `（code=${code} signal=${signal || '-'}），${delay / 1000}s 后`
|
||||
+ `第 ${attempt + 1}/${maxAttempts} 次重投`);
|
||||
const retry = setTimeout(() => {
|
||||
if (stopped) return;
|
||||
queue.push({ ...job, attempt: attempt + 1 });
|
||||
pump();
|
||||
}, delay);
|
||||
if (typeof retry.unref === 'function') retry.unref();
|
||||
// 退避期间不 pump：否则同一会话会被立刻重投，退避形同虚设
|
||||
return;
|
||||
}
|
||||
log(`worker ${child.pid}（mail ${entry.mailID}）重投 ${maxAttempts} 次仍未完成，放弃`
|
||||
+ `（code=${code} signal=${signal || '-'}）`);
|
||||
} else if (code !== 0) {
|
||||
log(`worker ${child.pid}（mail ${entry.mailID}）异常退出 code=${code} signal=${signal || '-'}`);
|
||||
}
|
||||
pump();
|
||||
});
|
||||
|
||||
child.on("error", (e) =>
|
||||
log(`worker ${child.pid} 出错: ${e?.message || e}`),
|
||||
);
|
||||
child.on('error', (e) => log(`worker ${child.pid} 出错: ${e?.message || e}`));
|
||||
|
||||
// 等 worker 说 ready 再派活：fork 返回时子进程的 import 还没跑完，
|
||||
// 此时 send 的消息会排在 IPC 队列里（能收到，但 ready 让顺序确定）。
|
||||
child.once("message", function first(msg) {
|
||||
if (msg?.type !== "ready") return;
|
||||
child.send({
|
||||
type: "job",
|
||||
kind: job.kind,
|
||||
data: job.data,
|
||||
session: {
|
||||
sessionFile: state.sessionFile || "",
|
||||
cwd: state.cwd || "",
|
||||
},
|
||||
grants: [...state.grants],
|
||||
lastSyncedName: state.lastSyncedName || "",
|
||||
config: config(),
|
||||
});
|
||||
});
|
||||
}
|
||||
// 等 worker 说 ready 再派活：fork 返回时子进程的 import 还没跑完，
|
||||
// 此时 send 的消息会排在 IPC 队列里（能收到，但 ready 让顺序确定）。
|
||||
child.once('message', function first(msg) {
|
||||
if (msg?.type !== 'ready') return;
|
||||
child.send({
|
||||
type: 'job',
|
||||
kind: job.kind,
|
||||
data: job.data,
|
||||
session: {
|
||||
sessionFile: state.sessionFile || '',
|
||||
cwd: state.cwd || '',
|
||||
},
|
||||
grants: [...state.grants],
|
||||
lastSyncedName: state.lastSyncedName || '',
|
||||
config: config(),
|
||||
});
|
||||
});
|
||||
}
|
||||
|
||||
function onWorkerMessage(entry, msg) {
|
||||
const state = sessionState.get(entry.key) || {
|
||||
grants: new Set(),
|
||||
lastSyncedName: "",
|
||||
};
|
||||
switch (msg?.type) {
|
||||
case "log":
|
||||
log(`[w${entry.child.pid}] ${msg.line}`);
|
||||
return;
|
||||
case "session_opened":
|
||||
// 一条会话可能先后用过多个 pi 会话 id（模型降级会换会话）。
|
||||
// 旧 id 仍计入 mail_driven，理由见 retired 的注释。
|
||||
if (state.piSessionId && state.piSessionId !== msg.piSessionId) {
|
||||
retired.add(state.piSessionId);
|
||||
}
|
||||
state.piSessionId = msg.piSessionId;
|
||||
state.sessionFile = msg.sessionFile;
|
||||
state.cwd = msg.cwd;
|
||||
sessionState.set(entry.key, state);
|
||||
return;
|
||||
case "permission_pending":
|
||||
permissionRoutes.set(msg.relayKey, entry.key);
|
||||
return;
|
||||
case "permission_grant":
|
||||
// 「一直同意」必须跨 worker 活着：worker 一封一进程，不存的话下一封
|
||||
// 邮件又问一遍，那个选项就是在骗人。
|
||||
state.grants.add(msg.toolName);
|
||||
sessionState.set(entry.key, state);
|
||||
return;
|
||||
case "name_synced":
|
||||
state.lastSyncedName = msg.signature;
|
||||
sessionState.set(entry.key, state);
|
||||
return;
|
||||
case "reconfigure":
|
||||
onReconfigure?.(msg.url, msg.agentKey);
|
||||
return;
|
||||
case "done":
|
||||
// 标记「这封真的处理完了」：exit 处理器据此区分「正常收尾」
|
||||
// 与「未回报就崩溃」（后者要重投）。
|
||||
entry.settled = true;
|
||||
if (!msg.ok) log(`投递 ${entry.mailID} 失败: ${msg.error}`);
|
||||
return;
|
||||
default:
|
||||
return;
|
||||
}
|
||||
}
|
||||
function onWorkerMessage(entry, msg) {
|
||||
const state = sessionState.get(entry.key) || { grants: new Set(), lastSyncedName: '' };
|
||||
switch (msg?.type) {
|
||||
case 'log':
|
||||
log(`[w${entry.child.pid}] ${msg.line}`);
|
||||
return;
|
||||
case 'session_opened':
|
||||
// 一条会话可能先后用过多个 pi 会话 id（模型降级会换会话）。
|
||||
// 旧 id 仍计入 mail_driven，理由见 retired 的注释。
|
||||
if (state.piSessionId && state.piSessionId !== msg.piSessionId) {
|
||||
retired.add(state.piSessionId);
|
||||
}
|
||||
state.piSessionId = msg.piSessionId;
|
||||
state.sessionFile = msg.sessionFile;
|
||||
state.cwd = msg.cwd;
|
||||
sessionState.set(entry.key, state);
|
||||
return;
|
||||
case 'permission_pending':
|
||||
permissionRoutes.set(msg.relayKey, entry.key);
|
||||
return;
|
||||
case 'permission_grant':
|
||||
// 「一直同意」必须跨 worker 活着：worker 一封一进程，不存的话下一封
|
||||
// 邮件又问一遍，那个选项就是在骗人。
|
||||
state.grants.add(msg.toolName);
|
||||
sessionState.set(entry.key, state);
|
||||
return;
|
||||
case 'name_synced':
|
||||
state.lastSyncedName = msg.signature;
|
||||
sessionState.set(entry.key, state);
|
||||
return;
|
||||
case 'reconfigure':
|
||||
onReconfigure?.(msg.url, msg.agentKey);
|
||||
return;
|
||||
case 'done':
|
||||
// 标记「这封真的处理完了」：exit 处理器据此区分「正常收尾」
|
||||
// 与「未回报就崩溃」（后者要重投）。
|
||||
entry.settled = true;
|
||||
if (!msg.ok) log(`投递 ${entry.mailID} 失败: ${msg.error}`);
|
||||
return;
|
||||
default:
|
||||
return;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 把权限决策路由到发起询问的那个 worker。
|
||||
*
|
||||
* @returns {boolean} 有没有找到对应的 worker。找不到说明那个 worker 已经退了
|
||||
* （桥重启、硬超时被杀、或者处理已经结束）—— 调用方据此走 B-4.2 的
|
||||
* 降级路径（把决策当一封通知投进原会话）。
|
||||
*/
|
||||
function routePermission(relayKey, decision) {
|
||||
const key = permissionRoutes.get(relayKey);
|
||||
if (!key) return false;
|
||||
const entry = running.get(key);
|
||||
if (!entry) {
|
||||
permissionRoutes.delete(relayKey);
|
||||
return false;
|
||||
}
|
||||
permissionRoutes.delete(relayKey);
|
||||
entry.child.send({ type: "permission_decision", relayKey, decision });
|
||||
return true;
|
||||
}
|
||||
/**
|
||||
* 把权限决策路由到发起询问的那个 worker。
|
||||
*
|
||||
* @returns {boolean} 有没有找到对应的 worker。找不到说明那个 worker 已经退了
|
||||
* （桥重启、硬超时被杀、或者处理已经结束）—— 调用方据此走 B-4.2 的
|
||||
* 降级路径（把决策当一封通知投进原会话）。
|
||||
*/
|
||||
function routePermission(relayKey, decision) {
|
||||
const key = permissionRoutes.get(relayKey);
|
||||
if (!key) return false;
|
||||
const entry = running.get(key);
|
||||
if (!entry) {
|
||||
permissionRoutes.delete(relayKey);
|
||||
return false;
|
||||
}
|
||||
permissionRoutes.delete(relayKey);
|
||||
entry.child.send({ type: 'permission_decision', relayKey, decision });
|
||||
return true;
|
||||
}
|
||||
|
||||
/** 这条邮件会话有 worker 在跑吗（B-4.2 判断降级路径用）。 */
|
||||
const hasSession = (mailSessionID) => sessionState.has(mailSessionID);
|
||||
/** 这条邮件会话有 worker 在跑吗（B-4.2 判断降级路径用）。 */
|
||||
const hasSession = (mailSessionID) => sessionState.has(mailSessionID);
|
||||
|
||||
/**
|
||||
* 忘掉一条已归档会话的全部状态。
|
||||
*
|
||||
* 归档是个**确定性的终点**：归档后那条会话不可寻址（别名 404），也不会再有
|
||||
* 新邮件投进来。把它的 sessionState 留着只是占内存，而上限淘汰是「猜」——
|
||||
* 能确切知道该删的时候就不该依赖猜。
|
||||
*
|
||||
* 正在跑的 worker **不杀**：归档不是中止指令，模型可能正在写文件；它自己跑完
|
||||
* 就退，只是那一轮的回信会因为会话已归档而被服务端拦下。
|
||||
*
|
||||
* @param {string} mailSessionID
|
||||
* @returns {boolean} 是否真的删掉了东西
|
||||
*/
|
||||
function forget(mailSessionID) {
|
||||
if (!mailSessionID) return false;
|
||||
// peek 而不是 get：这是清理路径，不该把即将删掉的条目刷成「最近活跃」。
|
||||
const state = sessionState.peek(mailSessionID);
|
||||
// 已归档会话的 pi 会话 id 也不必再报 mail_driven：那个标记的用途是让人在
|
||||
// 补全里看到「这条在跑邮件」，而已归档的会话不在补全候选里。
|
||||
if (state?.piSessionId) retired.delete(state.piSessionId);
|
||||
return sessionState.delete(mailSessionID);
|
||||
}
|
||||
/**
|
||||
* 忘掉一条已归档会话的全部状态。
|
||||
*
|
||||
* 归档是个**确定性的终点**：归档后那条会话不可寻址（别名 404），也不会再有
|
||||
* 新邮件投进来。把它的 sessionState 留着只是占内存，而上限淘汰是「猜」——
|
||||
* 能确切知道该删的时候就不该依赖猜。
|
||||
*
|
||||
* 正在跑的 worker **不杀**：归档不是中止指令，模型可能正在写文件；它自己跑完
|
||||
* 就退，只是那一轮的回信会因为会话已归档而被服务端拦下。
|
||||
*
|
||||
* @param {string} mailSessionID
|
||||
* @returns {boolean} 是否真的删掉了东西
|
||||
*/
|
||||
function forget(mailSessionID) {
|
||||
if (!mailSessionID) return false;
|
||||
// peek 而不是 get：这是清理路径，不该把即将删掉的条目刷成「最近活跃」。
|
||||
const state = sessionState.peek(mailSessionID);
|
||||
// 已归档会话的 pi 会话 id 也不必再报 mail_driven：那个标记的用途是让人在
|
||||
// 补全里看到「这条在跑邮件」，而已归档的会话不在补全候选里。
|
||||
if (state?.piSessionId) retired.delete(state.piSessionId);
|
||||
return sessionState.delete(mailSessionID);
|
||||
}
|
||||
|
||||
/**
|
||||
* 邮件驱动过的 pi 会话 id，喂给心跳快照的 `mail_driven` 标记。
|
||||
*
|
||||
* 不随 worker 退出而清：worker 退了不代表那条会话不再参与邮件往来 ——
|
||||
* 下一封邮件还会接着谈，而人在补全里需要看到它带着这个标记。
|
||||
* 重启丢是已知取舍（契约第六节）；确定性的清理时机是归档（见 forget）。
|
||||
*
|
||||
* 返回普通 Set 而不是 BoundedSet：调用方只拿它做一轮 has 查询就丢，
|
||||
* 没有长期持有，不需要上界。
|
||||
*/
|
||||
const mailDrivenIDs = () => {
|
||||
const out = new Set(retired);
|
||||
for (const st of sessionState.values()) {
|
||||
if (st.piSessionId) out.add(st.piSessionId);
|
||||
}
|
||||
return out;
|
||||
};
|
||||
/**
|
||||
* 邮件驱动过的 pi 会话 id，喂给心跳快照的 `mail_driven` 标记。
|
||||
*
|
||||
* 不随 worker 退出而清：worker 退了不代表那条会话不再参与邮件往来 ——
|
||||
* 下一封邮件还会接着谈，而人在补全里需要看到它带着这个标记。
|
||||
* 重启丢是已知取舍（契约第六节）；确定性的清理时机是归档（见 forget）。
|
||||
*
|
||||
* 返回普通 Set 而不是 BoundedSet：调用方只拿它做一轮 has 查询就丢，
|
||||
* 没有长期持有，不需要上界。
|
||||
*/
|
||||
const mailDrivenIDs = () => {
|
||||
const out = new Set(retired);
|
||||
for (const st of sessionState.values()) {
|
||||
if (st.piSessionId) out.add(st.piSessionId);
|
||||
}
|
||||
return out;
|
||||
};
|
||||
|
||||
function stop() {
|
||||
stopped = true;
|
||||
queue.length = 0;
|
||||
for (const { child, timer } of running.values()) {
|
||||
clearTimeout(timer);
|
||||
// 先 shutdown 让 worker 把未决权限 fail closed（B-9.2），再给它一点
|
||||
// 时间自己退。不直接 SIGKILL：那样 pi 侧的 await 不会返回，而 worker
|
||||
// 里可能正握着会话文件。
|
||||
try {
|
||||
child.send({ type: "shutdown" });
|
||||
} catch {
|
||||
/* 通道已断 */
|
||||
}
|
||||
setTimeout(() => {
|
||||
try {
|
||||
child.kill("SIGKILL");
|
||||
} catch {
|
||||
/* 已经死了 */
|
||||
}
|
||||
}, 2000).unref?.();
|
||||
}
|
||||
}
|
||||
function stop() {
|
||||
stopped = true;
|
||||
queue.length = 0;
|
||||
for (const { child, timer } of running.values()) {
|
||||
clearTimeout(timer);
|
||||
// 先 shutdown 让 worker 把未决权限 fail closed（B-9.2），再给它一点
|
||||
// 时间自己退。不直接 SIGKILL：那样 pi 侧的 await 不会返回，而 worker
|
||||
// 里可能正握着会话文件。
|
||||
try { child.send({ type: 'shutdown' }); } catch { /* 通道已断 */ }
|
||||
setTimeout(() => { try { child.kill('SIGKILL'); } catch { /* 已经死了 */ } }, 2000).unref?.();
|
||||
}
|
||||
}
|
||||
|
||||
/** 观测用：现在跑着几个、排了几个。 */
|
||||
const stats = () => ({
|
||||
running: running.size,
|
||||
queued: queue.length,
|
||||
sessions: sessionState.size,
|
||||
// 淘汰计数持续增长说明上限设得太小 —— 那意味着会话上下文在被白白丢掉，
|
||||
// 而症状是「这条会话怎么突然不记得前面说过什么了」。
|
||||
evictedSessions: sessionState.evicted,
|
||||
workers: [...running.values()].map((e) => ({
|
||||
pid: e.child.pid,
|
||||
mailID: e.mailID,
|
||||
ageMs: Date.now() - e.startedAt,
|
||||
})),
|
||||
});
|
||||
/** 观测用：现在跑着几个、排了几个。 */
|
||||
const stats = () => ({
|
||||
running: running.size,
|
||||
queued: queue.length,
|
||||
sessions: sessionState.size,
|
||||
// 淘汰计数持续增长说明上限设得太小 —— 那意味着会话上下文在被白白丢掉，
|
||||
// 而症状是「这条会话怎么突然不记得前面说过什么了」。
|
||||
evictedSessions: sessionState.evicted,
|
||||
workers: [...running.values()].map((e) => ({
|
||||
pid: e.child.pid, mailID: e.mailID, ageMs: Date.now() - e.startedAt,
|
||||
})),
|
||||
});
|
||||
|
||||
return {
|
||||
submit,
|
||||
routePermission,
|
||||
hasSession,
|
||||
forget,
|
||||
mailDrivenIDs,
|
||||
stop,
|
||||
stats,
|
||||
};
|
||||
return { submit, routePermission, hasSession, forget, mailDrivenIDs, stop, stats };
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user