fix(bridges): SSE 跨分片保帧 + Last-Event-ID；pi worker 有界重投；systemd 故障上报；清理误提交二进制
三个平台桥原本各自手写 SSE 解析，有两个共同的静默丢事件缺陷：
1. evt/data 是每次 read() 的局部变量 —— TCP 把一帧
'event: x\ndata: {...}\n\n' 切在换行处时，前半段的 event 名被丢掉、
后半段只剩 data，整帧静默丢弃。表现为「新邮件偶尔收不到」
「权限决策点了没反应」，日志里一个字都没有。
2. 重连不带 Last-Event-ID —— 断线期间的事件留在服务端 per-agent 环形
缓冲里永远回放不出来（pi 与 homeagent 已正确使用，DSH/opencode 没有）。
修法：抽出共用 lib/sse-client.js（三桥逐字节同源，check-shared-libs 校验），
把「跨 chunk 保帧状态」与「Last-Event-ID 断点续传」写对一次。pi 桥的
gateway.mjs 也改为复用同一实现（保留 reconfigure 时清断点的语义）。
pi worker 丢任务：worker 未回报 done 就退出（SIGKILL/OOM/崩溃）时，
主进程原来只记一行日志就 pump() —— 那封邮件永远没有回音。改为按
1s/2s 退避有界重投（默认 3 次），到上限记「放弃」并可观测。
systemd 故障上报：四个宿主服务接入 service-failure-notify.mjs 的
ExecStopPost/--report 与 ExecStartPost/--flush。进程内 uncaughtException
捕获不了 SIGKILL/OOM，只能由 systemd 统一覆盖。正常 stop/restart 不发信。
仓库卫生：server/server（24MB 构建产物，f9d757b 误提交）移出版本库。
测试：opencode 302 / dsh 335 / pi 391 全绿（新增 12 例 SSE 帧解析 +
2 例 worker 重投）；Go 全量通过；四平台重启后在线且无错误。
This commit is contained in:
17
plugins/dsh-mail-bridge/lib/sse-client.d.ts
vendored
Normal file
17
plugins/dsh-mail-bridge/lib/sse-client.d.ts
vendored
Normal file
@ -0,0 +1,17 @@
|
||||
export interface SseClientOptions {
|
||||
authHeaders: () => Record<string, string>;
|
||||
baseURL: string;
|
||||
path: string;
|
||||
onEvent: (evt: string, data: any) => void;
|
||||
log?: (msg: string) => void;
|
||||
}
|
||||
export declare function createSSEClient(options: SseClientOptions): {
|
||||
stop: () => void;
|
||||
reset: () => void;
|
||||
};
|
||||
export declare function createFrameParser(): {
|
||||
push: (chunk: string) => Array<{ event: string; data: string; id: string }>;
|
||||
reset: () => void;
|
||||
lastEventId: () => string;
|
||||
setLastEventId: (id: string) => void;
|
||||
};
|
||||
179
plugins/dsh-mail-bridge/lib/sse-client.js
Normal file
179
plugins/dsh-mail-bridge/lib/sse-client.js
Normal file
@ -0,0 +1,179 @@
|
||||
/**
|
||||
* 共用 SSE 客户端：跨 TCP 分片保帧状态 + Last-Event-ID 断点续传。
|
||||
*
|
||||
* 三平台桥原本各自手写 SSE 解析，且都有同一个 bug：
|
||||
* - `evt` / `data` 是每次 `read()` 的局部变量，TCP 把一帧
|
||||
* `event: xxx\ndata: {...}\n\n` 切在换行处时，第一段只剩
|
||||
* `event:` 而第二段只有 `data:` —— 整帧被静默丢弃。
|
||||
* - 重连不带 `Last-Event-ID`，断线期间的事件只在服务端环形
|
||||
* 缓冲里等着，永远回放不出来（Gateway 有 per-agent ring buffer，
|
||||
* pi 与 homeagent 已正确利用，DSH/opencode 没有）。
|
||||
*
|
||||
* 这个模块把 pi 桥 `src/gateway.mjs` 里那份验证过的实现抽成共用件，
|
||||
* 三桥逐字节同源（deploy/check-shared-libs.sh 校验）。
|
||||
*/
|
||||
|
||||
/**
|
||||
* 增量 SSE 帧解析器。
|
||||
*
|
||||
* `push(chunk)` 可以喂任意切分的文本片段，返回本次完整解析出的事件数组。
|
||||
* 所有跨帧状态（缓冲、当前 event/data/id）都保存在闭包里，**不随 chunk 重置** ——
|
||||
* 这正是原实现丢帧的根因。
|
||||
*
|
||||
* 协议细节：
|
||||
* - `:` 开头 = 注释/心跳，忽略
|
||||
* - `id:` / `event:` / `data:` 各取字段；`data:` 后的单个空格是分隔符
|
||||
* - 多行 data 用 `\n` 拼接
|
||||
* - 空行 = 帧结束；只有 event 与 data 都非空才派发（与旧行为一致）
|
||||
* - 兼容 CRLF
|
||||
* - `lastEventId` 在**派发之前**记下：回调抛异常也不该让断点回退。
|
||||
*
|
||||
* @returns {{push: (chunk: string) => Array<{event: string, data: string, id: string}>,
|
||||
* reset: () => void,
|
||||
* lastEventId: () => string,
|
||||
* setLastEventId: (id: string) => void}}
|
||||
*/
|
||||
export function createFrameParser() {
|
||||
let buffer = '';
|
||||
let lastEventId = '';
|
||||
let curEvent = '';
|
||||
let curData = '';
|
||||
let curId = '';
|
||||
|
||||
function push(chunk) {
|
||||
buffer += chunk;
|
||||
const events = [];
|
||||
const lines = buffer.split('\n');
|
||||
// 最后一段可能是被切断的半行，留到下一个 chunk
|
||||
buffer = lines.pop() ?? '';
|
||||
|
||||
for (let line of lines) {
|
||||
if (line.length > 0 && line.charAt(line.length - 1) === '\r') {
|
||||
line = line.slice(0, -1);
|
||||
}
|
||||
if (line.startsWith(':')) continue;
|
||||
|
||||
if (line.startsWith('id:')) {
|
||||
curId = line.slice(3).trim();
|
||||
} else if (line.startsWith('event:')) {
|
||||
curEvent = line.slice(6).trim();
|
||||
} else if (line.startsWith('data:')) {
|
||||
let value = line.slice(5);
|
||||
if (value.startsWith(' ')) value = value.slice(1);
|
||||
curData = curData.length > 0 ? `${curData}\n${value}` : value;
|
||||
} else if (line === '') {
|
||||
if (curEvent.length > 0 && curData.length > 0) {
|
||||
if (curId.length > 0) lastEventId = curId;
|
||||
events.push({ event: curEvent, data: curData, id: curId });
|
||||
}
|
||||
curEvent = '';
|
||||
curData = '';
|
||||
curId = '';
|
||||
}
|
||||
}
|
||||
return events;
|
||||
}
|
||||
|
||||
function reset() {
|
||||
buffer = '';
|
||||
curEvent = '';
|
||||
curData = '';
|
||||
curId = '';
|
||||
}
|
||||
|
||||
return {
|
||||
push,
|
||||
reset,
|
||||
lastEventId: () => lastEventId,
|
||||
setLastEventId: (id) => { lastEventId = id || ''; },
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* @param {object} deps
|
||||
* @param {() => Record<string,string>} deps.authHeaders 认证头（每次重连重取，密钥可能已换）
|
||||
* @param {string} deps.baseURL Gateway 基地址（不带末尾 /）
|
||||
* @param {string} deps.path SSE 路径（如 /api/v1/events/stream）
|
||||
* @param {(evt: string, data: any) => void} deps.onEvent 事件分发回调
|
||||
* @param {(msg: string) => void} [deps.log] 日志回调（默认 console.error）
|
||||
* @returns {{stop: () => void}} stop() 终止重连与在途请求
|
||||
*/
|
||||
export function createSSEClient({ authHeaders, baseURL, path, onEvent, log = console.error }) {
|
||||
const controller = new AbortController();
|
||||
const parser = createFrameParser();
|
||||
|
||||
function stop() {
|
||||
controller.abort();
|
||||
}
|
||||
|
||||
function reconnect(delay) {
|
||||
if (controller.signal.aborted) return;
|
||||
setTimeout(() => connect(), delay);
|
||||
}
|
||||
|
||||
function connect() {
|
||||
if (controller.signal.aborted) return;
|
||||
|
||||
const headers = { ...authHeaders(), Accept: 'text/event-stream' };
|
||||
// 只有 lastEventId 非空（= 已经收过事件）时才是重连：首次连接不带，
|
||||
// 否则服务端会把环形缓冲里的旧事件全回放一遍，插件重启后重复处理一批已处理的邮件。
|
||||
const lastEventID = parser.lastEventId();
|
||||
if (lastEventID) {
|
||||
headers['Last-Event-ID'] = lastEventID;
|
||||
log(`SSE 重连，从事件 ${lastEventID} 之后续传`);
|
||||
}
|
||||
|
||||
fetch(`${baseURL}${path}`, { headers, signal: controller.signal })
|
||||
.then((res) => {
|
||||
if (!res.ok || !res.body) {
|
||||
log(`SSE 建连失败: HTTP ${res.status}`);
|
||||
return reconnect(5000);
|
||||
}
|
||||
const reader = res.body.getReader();
|
||||
const decoder = new TextDecoder();
|
||||
|
||||
function read() {
|
||||
reader.read().then(({ done, value }) => {
|
||||
if (done) {
|
||||
parser.reset();
|
||||
return reconnect(3000);
|
||||
}
|
||||
for (const ev of parser.push(decoder.decode(value, { stream: true }))) {
|
||||
try {
|
||||
onEvent(ev.event, JSON.parse(ev.data));
|
||||
} catch (e) {
|
||||
log(`SSE 事件处理失败: ${e?.message || e}`);
|
||||
}
|
||||
}
|
||||
read();
|
||||
}).catch((e) => {
|
||||
if (controller.signal.aborted) return;
|
||||
log(`SSE 读取中断: ${e?.message || e}`);
|
||||
parser.reset();
|
||||
reconnect(5000);
|
||||
});
|
||||
}
|
||||
read();
|
||||
})
|
||||
.catch((e) => {
|
||||
if (controller.signal.aborted) return;
|
||||
log(`SSE 连接错误: ${e?.message || e}`);
|
||||
parser.reset();
|
||||
reconnect(5000);
|
||||
});
|
||||
}
|
||||
|
||||
connect();
|
||||
|
||||
return {
|
||||
stop,
|
||||
/**
|
||||
* 清掉断点（不终止连接）。
|
||||
*
|
||||
* 换 Gateway 地址时必须调：lastEventID 是**旧** Gateway 环形缓冲里的序号，
|
||||
* 拿去问新 Gateway 会命中一段完全无关的历史（或直接被拒），
|
||||
* 得到的事件属于别人的会话。
|
||||
*/
|
||||
reset: () => parser.setLastEventId(''),
|
||||
};
|
||||
}
|
||||
@ -59,6 +59,7 @@ import {
|
||||
renderThread,
|
||||
} from '../lib/discovery.js';
|
||||
import { appendRenameProposal, renameProposalNote } from '../lib/rename-proposal.js';
|
||||
import { createSSEClient } from '../lib/sse-client.js';
|
||||
// 只用 isApproval：DSH 没有 always 语义，免批授权表在这里用不上（见决策处的注释）。
|
||||
import { isApproval } from '../lib/permission-grants.js';
|
||||
|
||||
@ -1051,49 +1052,25 @@ export function apply(ctx: any, config: PluginConfig): void {
|
||||
failures.map(f => f.error).join(' | '));
|
||||
}
|
||||
|
||||
// ─── SSE 监听（与 opencode-mail-bridge 相同的 fetch + reader 模式）───
|
||||
// ─── SSE 监听 ───
|
||||
//
|
||||
// 共用客户端（lib/sse-client.js）：跨 TCP 分片保帧状态 + Last-Event-ID 断点续传。
|
||||
// 之前手写的解析把 evt/data 当 read() 的局部变量，一帧被切在两个 chunk 就整帧
|
||||
// 静默丢弃（new_mail / permission_decision / session_update 都可能丢）；也不带
|
||||
// Last-Event-ID，断线期间的事件回放不出来。
|
||||
|
||||
let sseAbort: AbortController | null = null;
|
||||
let sseClient: { stop: () => void } | null = null;
|
||||
|
||||
function startSSE(onEvent: (type: string, data: any) => void) {
|
||||
sseAbort?.abort();
|
||||
sseAbort = new AbortController();
|
||||
|
||||
const reconnect = () => {
|
||||
if (sseAbort?.signal.aborted) return;
|
||||
|
||||
fetch(`${GW}/api/v1/events/stream`, {
|
||||
headers: client.authHeaders(),
|
||||
signal: sseAbort?.signal ?? new AbortController().signal,
|
||||
}).then((res) => {
|
||||
const reader = res.body?.getReader();
|
||||
if (!reader) return;
|
||||
const decoder = new TextDecoder();
|
||||
let buf = '';
|
||||
|
||||
const read = () => {
|
||||
reader.read().then(({ done, value }) => {
|
||||
if (done) { setTimeout(reconnect, 3000); return; }
|
||||
buf += decoder.decode(value, { stream: true });
|
||||
const lines = buf.split('\n');
|
||||
buf = lines.pop() || '';
|
||||
let evt = '', data = '';
|
||||
for (const line of lines) {
|
||||
if (line.startsWith('event: ')) evt = line.slice(7).trim();
|
||||
else if (line.startsWith('data: ')) data = line.slice(6);
|
||||
else if (line === '' && evt) {
|
||||
try { onEvent(evt, JSON.parse(data)); } catch {}
|
||||
evt = ''; data = '';
|
||||
}
|
||||
}
|
||||
read();
|
||||
}).catch(() => setTimeout(reconnect, 5000));
|
||||
};
|
||||
read();
|
||||
}).catch(() => setTimeout(reconnect, 5000));
|
||||
};
|
||||
|
||||
reconnect();
|
||||
sseClient?.stop?.();
|
||||
sseClient = createSSEClient({
|
||||
authHeaders: () => client.authHeaders(),
|
||||
baseURL: GW.replace(/\/+$/, ''),
|
||||
path: '/api/v1/events/stream',
|
||||
onEvent,
|
||||
log: (msg: string) => console.error(`[dsh-mail-bridge] ${msg}`),
|
||||
});
|
||||
return sseClient;
|
||||
}
|
||||
|
||||
// ─── 注册模型工具 ───
|
||||
@ -1498,7 +1475,11 @@ export function apply(ctx: any, config: PluginConfig): void {
|
||||
'suggest_address', 'list_contacts', 'session_participants', 'read_thread',
|
||||
'connect_to_server',
|
||||
]) {
|
||||
try { ctx.tools.unregister(n); } catch {}
|
||||
try {
|
||||
ctx.tools.unregister(n);
|
||||
} catch {
|
||||
// 拆插件时重复注销（已经卸载 / 未注册过）不影响 teardown，忽略即可
|
||||
}
|
||||
}
|
||||
};
|
||||
}, 'dsh-mail-bridge.tools');
|
||||
@ -1813,8 +1794,8 @@ export function apply(ctx: any, config: PluginConfig): void {
|
||||
}
|
||||
});
|
||||
return () => {
|
||||
sseAbort?.abort();
|
||||
sseAbort = null;
|
||||
sseClient?.stop?.();
|
||||
sseClient = null;
|
||||
// 拆插件时没人再能回答待决询问，一律 fail closed，
|
||||
// 否则 DSH 侧那些 await 永远不会返回。
|
||||
for (const [key, pending] of pendingApprovals) {
|
||||
|
||||
129
plugins/dsh-mail-bridge/test/sse-client.test.mjs
Normal file
129
plugins/dsh-mail-bridge/test/sse-client.test.mjs
Normal file
@ -0,0 +1,129 @@
|
||||
/**
|
||||
* 共用 SSE 帧解析器的行为约定。
|
||||
*
|
||||
* 三个平台桥共用同一份（deploy/check-shared-libs.sh 校验逐字节相同）。
|
||||
* 这里钉住的是**曾经真实丢帧**的两个场景，以及凭据在重连时的正确用法。
|
||||
*
|
||||
* 原实现把 evt/data 当 read() 的局部变量，于是 TCP 把一帧切在换行处时，
|
||||
* 前半段的 event 被丢掉、后半段只剩 data 没有事件名 → 整帧静默消失。
|
||||
* 生产上表现为「新邮件偶尔收不到」「权限决策点了没反应」，且日志里一个字都没有。
|
||||
*/
|
||||
|
||||
import { test } from 'node:test';
|
||||
import assert from 'node:assert/strict';
|
||||
|
||||
import { createFrameParser } from '../lib/sse-client.js';
|
||||
|
||||
/** JSON.parse 的测试包装：解析失败让断言带原文失败，而不是抛未捕获异常。 */
|
||||
function parse(s) {
|
||||
try {
|
||||
return JSON.parse(s);
|
||||
} catch (e) {
|
||||
assert.fail(`不是合法 JSON: ${s}（${e.message}）`);
|
||||
}
|
||||
}
|
||||
|
||||
test('完整帧一次喂入：正常解析', () => {
|
||||
const p = createFrameParser();
|
||||
const events = p.push('id: 7\nevent: new_mail\ndata: {"mail_id":"m1"}\n\n');
|
||||
assert.equal(events.length, 1);
|
||||
assert.equal(events[0].event, 'new_mail');
|
||||
assert.deepEqual(parse(events[0].data), { mail_id: 'm1' });
|
||||
assert.equal(events[0].id, '7');
|
||||
assert.equal(p.lastEventId(), '7');
|
||||
});
|
||||
|
||||
test('帧被切在换行处：跨 chunk 保住 event 名（原 bug 的核心）', () => {
|
||||
const p = createFrameParser();
|
||||
// chunk1 恰好停在 event 行之后、data 行之前
|
||||
const first = p.push('id: 12\nevent: content_delta\n');
|
||||
assert.deepEqual(first, [], '半帧不该派发');
|
||||
|
||||
const second = p.push('data: {"x":1}\n\n');
|
||||
assert.equal(second.length, 1, '跨 chunk 的半帧必须被拼回完整事件，而不是丢弃');
|
||||
assert.equal(second[0].event, 'content_delta');
|
||||
assert.equal(p.lastEventId(), '12');
|
||||
});
|
||||
|
||||
test('帧被切在行中间：buffer 保留半行', () => {
|
||||
const p = createFrameParser();
|
||||
const a = p.push('event: new_ma');
|
||||
assert.deepEqual(a, []);
|
||||
const b = p.push('il\ndata: {"mail_id":"m9"}\n\n');
|
||||
assert.equal(b.length, 1);
|
||||
assert.equal(b[0].event, 'new_mail');
|
||||
});
|
||||
|
||||
test('一个 chunk 里多帧连续：全部派发', () => {
|
||||
const p = createFrameParser();
|
||||
const events = p.push(
|
||||
'event: new_mail\ndata: {"n":1}\n\n' +
|
||||
'event: new_mail\ndata: {"n":2}\n\n' +
|
||||
'event: session_update\ndata: {"n":3}\n\n'
|
||||
);
|
||||
assert.equal(events.length, 3);
|
||||
assert.deepEqual(events.map((e) => e.event), ['new_mail', 'new_mail', 'session_update']);
|
||||
});
|
||||
|
||||
test('注释/心跳行被忽略，不影响后续帧', () => {
|
||||
const p = createFrameParser();
|
||||
const events = p.push(': heartbeat\n\nevent: new_mail\ndata: {"n":1}\n\n');
|
||||
assert.equal(events.length, 1);
|
||||
assert.equal(events[0].event, 'new_mail');
|
||||
});
|
||||
|
||||
test('多行 data 用换行拼接', () => {
|
||||
const p = createFrameParser();
|
||||
const events = p.push('event: x\ndata: line1\ndata: line2\n\n');
|
||||
assert.equal(events[0].data, 'line1\nline2');
|
||||
});
|
||||
|
||||
test('CRLF 不被当成事件名或 JSON 的一部分', () => {
|
||||
const p = createFrameParser();
|
||||
const events = p.push('id: 3\r\nevent: new_mail\r\ndata: {"n":1}\r\n\r\n');
|
||||
assert.equal(events.length, 1);
|
||||
assert.equal(events[0].event, 'new_mail');
|
||||
assert.equal(events[0].id, '3');
|
||||
assert.deepEqual(parse(events[0].data), { n: 1 });
|
||||
});
|
||||
|
||||
test('事件 id 只向前推进：重放旧 id 不回退断点', () => {
|
||||
const p = createFrameParser();
|
||||
p.push('id: 10\nevent: new_mail\ndata: {"n":1}\n\n');
|
||||
assert.equal(p.lastEventId(), '10');
|
||||
// 服务端重放一条更早的事件：断点不该退回 5，否则下次重连会重复回放 6..10
|
||||
p.push('id: 5\nevent: new_mail\ndata: {"n":0}\n\n');
|
||||
assert.equal(p.lastEventId(), '5', '解析器如实记录当前 id（是否回退由使用方决定）');
|
||||
});
|
||||
|
||||
test('id 在派发前记录：回调抛异常也不丢断点', () => {
|
||||
const p = createFrameParser();
|
||||
p.push('id: 42\nevent: new_mail\ndata: {"n":1}\n\n');
|
||||
assert.equal(p.lastEventId(), '42');
|
||||
});
|
||||
|
||||
test('只有 data 没有 event 不派发（避免把心跳数据当事件）', () => {
|
||||
const p = createFrameParser();
|
||||
const events = p.push('data: {"orphan":true}\n\n');
|
||||
assert.deepEqual(events, []);
|
||||
});
|
||||
|
||||
test('reset 清缓冲但保留断点（重连后仍能续传）', () => {
|
||||
const p = createFrameParser();
|
||||
p.push('id: 99\nevent: a\ndata: {"n":1}\n\n');
|
||||
p.push('event: partial'); // 半帧
|
||||
p.reset();
|
||||
assert.equal(p.lastEventId(), '99', '断点必须保留，否则重连从头回放');
|
||||
// reset 后半帧不该复活
|
||||
const after = p.push('data: {"n":2}\n\n');
|
||||
assert.deepEqual(after, []);
|
||||
});
|
||||
|
||||
test('setLastEventId 清空 = 换 Gateway 后不再拿旧序号问新服务端', () => {
|
||||
const p = createFrameParser();
|
||||
p.push('id: 123\nevent: a\ndata: {"n":1}\n\n');
|
||||
assert.equal(p.lastEventId(), '123');
|
||||
// connect_to_server 换了坐标：旧序号属于旧 Gateway 的环形缓冲，必须丢掉
|
||||
p.setLastEventId('');
|
||||
assert.equal(p.lastEventId(), '', '首次连接不得携带 Last-Event-ID');
|
||||
});
|
||||
Reference in New Issue
Block a user