feat: 工作区归属修复 + 平台会话同步 + 对话树整树展开 + DSH 插件
四个各自独立的生产缺陷，共同的根源都是「本该属于会话的属性没有存在会话上」。 ## 1. dsh 指定工作目录完全失效（所有会话落进「未分组」） 插件建会话时用的 cwd 是自己拼的 `~/.dsh/mail-sessions/mail-<uuid>` —— 每封邮件一个全新的空目录。DSH 与 opencode 都按 cwd 给会话分组，于是所有 邮件会话既不属于任何项目、彼此也不同组。 而 Gateway 从来没把地址里的 path 位发给插件：`notifyRecipients` 的 payload 只有 mail_id/session_id/from_name/subject，`to_workspace` 虽然入库了却不在 SSE 事件里，插件即使想用也拿不到。 - SSE `new_mail` 事件加 `to_workspace`。**每个收件方拿到自己那个地址的 path**， 不是主收件人的 —— 抄送给 opencode@/a 与主发给 dsh@/b 是两个工作区 - 两个插件的 cwd 都改为取寻址的 path 位；不存在的目录**不创建**而是回退到 兜底目录（一个笔误不该在磁盘上落下真目录，Agent 会在里面一无所获地干活） - 拒绝相对路径：cwd 的相对基准是 harness 进程的启动目录，systemd 下通常是 `/` ## 2. 会话别名列不出工作区下的历史会话（无法选择） workspace 只存在于 `mails.to_workspace` 上，「这个工作区下有哪些会话」必须 JOIN mails 再从收发双方的 workspace 里猜。而 Agent 回信时 from_workspace 填的是 **Agent 名**而不是路径，旧条件 `to_workspace = $p OR from_workspace = $p` 在只剩 Agent 回信可匹配时两边都对不上。 - `sessions.workspace` 新列，`CreateSession` 从地址的 path 位带入 - `SuggestSessionCandidates` 取代 `SuggestSessionsFor`：以会话自己的 workspace 为权威，历史会话（该列为空）回退到 mails 反推 —— 升级后老会话不该消失 - `FindOrCreateDefaultSession` 同步改用会话的 workspace ## 3. 平台侧会话在补全里根本不存在 人直接在 opencode/DSH 界面上开的会话，Gateway 一无所知。 新增 `agent_platform_sessions` 镜像表，插件在心跳里上报快照。 **上报而非 Gateway 反向拉取**：当前架构是单向的（Agent 持密钥主动连 Gateway， Gateway 从不外呼），反向拉取需要它保存各平台的地址与凭证，那是另一套信任模型。 - 与 sessions 表分开存：镜像里是别人家的会话，id 属于平台的 id 空间，没有 本侧的 owner/预算/邮件。混进 sessions 会让每一处「按会话鉴权」都要先判断 这条到底是不是真的本侧会话 - **整表替换而非增量合并**：平台侧删掉的会话必须从候选里消失 —— session 位是 三态语义，指向不存在的会话直接 404 - **nil 与空数组语义不同**：插件拉不到列表时省略该字段（保留镜像）， 而不是传空数组把镜像抹掉 - **subagent 子会话不上报**：实测 DSH 的 list 里混着 49 条子会话，标题就是 派活的提示词前缀（九条都叫 "You are auditing ONE file"），slug 全撞名； 它们是父 agent 内部的工作单元，人往里发邮件毫无意义 - **slug 撞名只留最近那条**：服务端只能取其中一条，上报同名项只会让补全里 出现几个点哪个都不确定的候选 - DSH 插件此前**完全没有心跳** —— Gateway 靠 last_seen 判在线，一直靠注册撑着 补全候选带标题与来源：`suggestions` 保留纯字符串数组（不打破已部署的前端与 第三方客户端），新增同序的 `candidates`。过滤时标题也参与匹配 —— 人记得的是 「缓存选型」而不是 brisk-harbor 这种随机短名。 ## 4. 对话树看不见抄送与转发产生的分支 旧实现从锚点分「祖先链 + 子树」两路展开，而**兄弟节点既不是锚点的祖先也不是 它的子孙**：一封抄送给两个 Agent 的邮件收到两个回复，从其中一个看树永远看不到 另一个；挂在原件上的转发分支同理。 改为先 `ThreadRootOf` 上溯到线索根，再从根整树 BFS。只剩一个加载方向， 因此不再需要滚动位置补偿。前端补上抄送人列表与转发标记 —— 树上两个兄弟节点 为什么并列，唯一的解释就是父邮件抄送给了两个人。 ## 5. DSH 插件（Phase 7.7） 卡了一下午的 `Cannot read properties of undefined (reading 'kind')` 根因是 `followup()` 的参数形状：DSH 要完整的 UserMessage（content + source）， 而我照抄了 opencode 的 parts 数组。错误抛在 agent-loop 内部，不指向调用点。 - `agent/status` → idle 时自动转发最后一条 assistant 消息（对应 opencode 的 session.idle），复用 relay-dedup 让位于模型的主动回信，走免配额通道 - `approval/request` 权限询问转邮件问人。与 opencode 的差异：那边的 permission.ask 是同步钩子只能立即返回 ask，DSH 这边是异步 waterfall， 可以真的等人 —— 拆插件时未决询问一律 fail closed，否则 await 永不返回 - 会话别名由模型标题派生（保留中文，去掉 `.` `@` `/` 等寻址分隔符 —— 留在别名里会让它自己被解析器切开） - 逻辑放 lib/ 下的纯函数并加测试：三类约定都是「错了不当场报错、只在深处 炸一个无关错误」 ## 其他 - `deploy/reset-demo.sh`：清空演示邮件数据，保留账号与密钥。备份用 `.backup` 而非 cp（WAL 下 cp 拿到的是缺尾巴的库）；手工按依赖顺序删（SQLite 的 foreign_keys 默认关，声明了 REFERENCES 也不级联）；只在目标是默认库时才碰 systemd（演练时误停过一次生产服务） - 插件 dist/ 不进版本库，install.sh 负责构建 - `permission_decision` 事件补 session_id：插件重启丢了待决映射时要靠它定位会话
This commit is contained in:
696
plugins/dsh-mail-bridge/src/index.ts
Normal file
696
plugins/dsh-mail-bridge/src/index.ts
Normal file
@ -0,0 +1,696 @@
|
||||
/**
|
||||
* dsh-mail-bridge — DeepSeek Harness ↔ AgentMail 桥接插件
|
||||
*
|
||||
* 与 opencode-mail-bridge 共享同一套 Gateway API。
|
||||
* DSH 用 Cordis 插件框架（@deepseek-ai/cordis），不是 opencode 的 @opencode-ai/plugin。
|
||||
*
|
||||
* 关键差异：
|
||||
* - opencode: client.session.create() + client.session.promptAsync()
|
||||
* - DSH: ctx.agents.create() + agent.followup()
|
||||
* - opencode: tool 用 zod schema
|
||||
* - DSH: tool 用 defineTool() + 参数 spec 格式
|
||||
*/
|
||||
|
||||
import { readFileSync, writeFileSync, existsSync, mkdirSync } from 'node:fs';
|
||||
import { readFile, writeFile } from 'node:fs/promises';
|
||||
import { randomBytes } from 'node:crypto';
|
||||
import { homedir } from 'node:os';
|
||||
import { join, dirname } from 'node:path';
|
||||
import {
|
||||
explicitSends,
|
||||
noteExplicitSend,
|
||||
shouldSkipAutoRelay,
|
||||
} from '../lib/relay-dedup.js';
|
||||
import {
|
||||
userMessage,
|
||||
replySubject,
|
||||
lastAssistantText,
|
||||
modelTitle,
|
||||
} from '../lib/message.js';
|
||||
import { snapshotDshSessions, slugFromTitle } from '../lib/session-snapshot.js';
|
||||
import { resolveWorkspaceCwd, ensureCwd } from '../lib/workspace.js';
|
||||
|
||||
// ─── 凭证管理 ───
|
||||
|
||||
const CONFIG_DIR = process.env.AGENTMAIL_CONFIG_DIR || join(homedir(), '.agentmail');
|
||||
const KEY_FILE = join(CONFIG_DIR, 'agent.key');
|
||||
|
||||
function readLocalKey(): string | null {
|
||||
try {
|
||||
if (!existsSync(KEY_FILE)) return null;
|
||||
const raw = JSON.parse(readFileSync(KEY_FILE, 'utf8'));
|
||||
return typeof raw.key === 'string' ? raw.key : null;
|
||||
} catch { return null; }
|
||||
}
|
||||
|
||||
function saveLocalKey(key: string) {
|
||||
try {
|
||||
mkdirSync(dirname(KEY_FILE), { recursive: true });
|
||||
writeFileSync(KEY_FILE, JSON.stringify({ key }, null, 2), 'utf8');
|
||||
} catch { /* 忽略 */ }
|
||||
}
|
||||
|
||||
function generateLocalKey(): string {
|
||||
const key = 'ak_' + randomBytes(24).toString('hex');
|
||||
saveLocalKey(key);
|
||||
return key;
|
||||
}
|
||||
|
||||
// ─── Gateway HTTP 客户端 ───
|
||||
|
||||
class GatewayClient {
|
||||
baseURL: string;
|
||||
agentName: string;
|
||||
agentKey: string;
|
||||
agentSecret: string;
|
||||
|
||||
constructor(gatewayURL: string, agentName: string, agentKey: string, agentSecret: string) {
|
||||
this.baseURL = gatewayURL.replace(/\/+$/, '');
|
||||
this.agentName = agentName;
|
||||
this.agentKey = agentKey;
|
||||
this.agentSecret = agentSecret;
|
||||
}
|
||||
|
||||
/** 与 opencode-mail-bridge 的 authHeaders() 相同逻辑 */
|
||||
authHeaders(): Record<string, string> {
|
||||
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 post(path: string, body: Record<string, unknown>): Promise<any> {
|
||||
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() as any;
|
||||
if (!res.ok) throw new Error(data?.error || `POST ${path} failed: ${res.status}`);
|
||||
return data;
|
||||
}
|
||||
|
||||
async get(path: string): Promise<any> {
|
||||
const res = await fetch(`${this.baseURL}/api/v1${path}`, { headers: this.authHeaders() });
|
||||
const data = await res.json() as any;
|
||||
if (!res.ok) throw new Error(data?.error || `GET ${path} failed: ${res.status}`);
|
||||
return data;
|
||||
}
|
||||
|
||||
async register(): Promise<void> {
|
||||
await this.post('/agent/register', {
|
||||
name: this.agentName,
|
||||
secret: this.agentSecret || '',
|
||||
workspaces: [],
|
||||
platform: 'dsh',
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
// ─── 会话映射（与 opencode-mail-bridge 相同结构）───
|
||||
|
||||
const sessionMap = new Map<string, { dshSessionId: string; directory: string }>();
|
||||
const reverseMap = new Map<string, string>();
|
||||
const mailDrivenSessions = new Set<string>();
|
||||
const mailContexts = new Map<string, { replyTo: string; subject: string; mailID: string }>();
|
||||
const relayedSummaries = new Map<string, string>();
|
||||
const syncedTitles = new Map<string, string>();
|
||||
|
||||
// 权限询问：DSH 的 approval/request 是 waterfall 钩子，插件把它转成邮件问人，
|
||||
// 人类决策通过 SSE 回来后再 resolve 这个 promise，让 DSH 自己恢复执行。
|
||||
// relay_key 用 `${sessionId}:${toolName}:${callId}` —— DSH 不给询问发 id，
|
||||
// 而同一个 callId 的同一个工具只会问一次。
|
||||
interface PendingApproval {
|
||||
resolve: (outcome: string) => void;
|
||||
sessionId: string;
|
||||
}
|
||||
const pendingApprovals = new Map<string, PendingApproval>();
|
||||
|
||||
// ─── 运行时导入 DSH 内部函数 ───
|
||||
|
||||
let _defineTool: any;
|
||||
function defineTool(opts: any): any {
|
||||
if (!_defineTool) {
|
||||
try {
|
||||
const dshToolsPath = require.resolve('@deepseek-ai/dsh-tools');
|
||||
_defineTool = require(dshToolsPath).defineTool;
|
||||
} catch { return opts; }
|
||||
}
|
||||
return _defineTool(opts);
|
||||
}
|
||||
|
||||
// ─── Cordis 插件入口 ───
|
||||
|
||||
// Cordis 要求插件声明依赖的服务：没有 inject，ctx.tools / ctx.agents 根本不存在
|
||||
// （报 `cannot get property "tools" without inject`）。
|
||||
//
|
||||
// sessionQuery 不列在这里而用 ctx.get('sessionQuery') 取：inject 是硬依赖，
|
||||
// 列进去的服务没挂载时整个插件不会启动 —— 而会话上报只是补全体验，
|
||||
// 不应该能把邮件投递整体拘死。
|
||||
export const inject = ['agents', 'tools'];
|
||||
export const name = 'dsh-mail-bridge';
|
||||
|
||||
interface PluginConfig {
|
||||
gateway: { url: string; agentName: string; agentKey: string; agentSecret: string };
|
||||
reply: { provider?: string; model?: string };
|
||||
}
|
||||
|
||||
export function apply(ctx: any, config: PluginConfig): void {
|
||||
const GW = config.gateway.url || 'http://127.0.0.1:8180';
|
||||
const AGENT_NAME = config.gateway.agentName || 'dsh';
|
||||
let AGENT_KEY = config.gateway.agentKey || '';
|
||||
const AGENT_SECRET = config.gateway.agentSecret || '';
|
||||
const REPLY_PROVIDER = config.reply.provider || '';
|
||||
const REPLY_MODEL = config.reply.model || 'AUTO';
|
||||
|
||||
if (!AGENT_KEY && !AGENT_SECRET) {
|
||||
AGENT_KEY = readLocalKey() || generateLocalKey();
|
||||
}
|
||||
|
||||
const client = new GatewayClient(GW, AGENT_NAME, AGENT_KEY, AGENT_SECRET);
|
||||
|
||||
// 注册 Agent
|
||||
(async () => {
|
||||
try {
|
||||
await client.register();
|
||||
ctx.logger.info(`[dsh-mail-bridge] 已接入 ${GW}，身份 ${AGENT_NAME}`);
|
||||
} catch (e: any) {
|
||||
ctx.logger.error(`[dsh-mail-bridge] 注册失败: ${e?.message || e}`);
|
||||
}
|
||||
})();
|
||||
|
||||
// ─── 心跳 + 平台会话上报 ───
|
||||
//
|
||||
// 心跳两个职责：
|
||||
// 1. 保活 —— Gateway 靠 last_seen 判在线，不发心跳就会被当成离线（之前真的没发）
|
||||
// 2. 上报平台侧会话快照 —— 写信时想续谈某条会话，得先知道那个工作区下
|
||||
// 有哪些会话可续；Gateway 只看得见邮件驱动的那部分。
|
||||
//
|
||||
// 上报而不是让 Gateway 反向拉取：当前架构是单向的（Agent 持密钥主动连
|
||||
// Gateway，Gateway 从不外呼），反向拉取需要它保存各平台的地址与凭证。
|
||||
|
||||
/**
|
||||
* 收集本机 DSH 的会话快照。
|
||||
*
|
||||
* 优先用 ctx.sessionQuery（包含已落盘的历史会话），它不可用时退到
|
||||
* ctx.agents.list()（只有当前活着的）。base bundle 里 session-query-sqlite 的
|
||||
* openAt 是 'never'，但 listSessions/readTitle 这些精确读不依赖 SQLite
|
||||
* —— 只有全文搜索会报 SESSION_QUERY_SEARCH_DISABLED。
|
||||
*
|
||||
* 返回 undefined 表示「本次拿不到列表」，调用方应当省略字段而不是传空数组：
|
||||
* 空数组的语义是「平台侧确实一条会话都没有」，会把服务端的镜像抹掉。
|
||||
*/
|
||||
async function collectSessions(): Promise<any[] | undefined> {
|
||||
const q = ctx.get('sessionQuery');
|
||||
if (q?.listSessions) {
|
||||
try {
|
||||
const records = await q.listSessions();
|
||||
const ids = records
|
||||
.map((r: any) => r?.header?.id)
|
||||
.filter((id: any): id is string => typeof id === 'string');
|
||||
// 标题在日志里，需要单独 fold。批量读而不是逐个读：
|
||||
// readTitleSnapshots 共用一次 corpus 观测，而逐个 readTitle 会重复加载日志。
|
||||
const titles = new Map<string, string>();
|
||||
if (q.readTitleSnapshots && ids.length > 0) {
|
||||
const results = await q.readTitleSnapshots(ids);
|
||||
results.forEach((res: any, i: number) => {
|
||||
// 单个会话读失败不该拘到其他会话（接口本身就是逐条隔离的）
|
||||
if (res?.status === 'rejected') return;
|
||||
const title = res?.value?.title?.title ?? res?.title?.title;
|
||||
if (typeof title === 'string' && title) titles.set(ids[i], title);
|
||||
});
|
||||
}
|
||||
return records.map((r: any) => ({
|
||||
id: r?.header?.id,
|
||||
cwd: r?.header?.cwd ?? '',
|
||||
title: titles.get(r?.header?.id) ?? '',
|
||||
updatedAt: r?.header?.createdAt,
|
||||
// subagent 子会话要被过滤掉，判据在 header 上
|
||||
origin: r?.header?.origin,
|
||||
delegationDepth: r?.header?.delegationDepth,
|
||||
}));
|
||||
} catch (e: any) {
|
||||
ctx.logger.warn(`[dsh-mail-bridge] sessionQuery 不可用，退到活会话列表: ${e?.message || e}`);
|
||||
}
|
||||
}
|
||||
|
||||
// 退路：只报当前活着的 agent。比什么都不报强 ——
|
||||
// 它们恰好是正在进行的任务，也是最可能被续谈的那批。
|
||||
try {
|
||||
const live = ctx.agents?.list?.() ?? [];
|
||||
return live.map((a: any) => ({
|
||||
id: String(a?.id ?? ''),
|
||||
cwd: a?.session?.header?.cwd ?? '',
|
||||
title: modelTitle(a?.session?.events ?? []),
|
||||
updatedAt: a?.session?.header?.createdAt,
|
||||
origin: a?.session?.header?.origin,
|
||||
delegationDepth: a?.session?.header?.delegationDepth,
|
||||
}));
|
||||
} catch {
|
||||
return undefined;
|
||||
}
|
||||
}
|
||||
|
||||
async function beat(): Promise<void> {
|
||||
let body: Record<string, unknown> = {};
|
||||
const entries = await collectSessions();
|
||||
if (entries) {
|
||||
body = {
|
||||
platform_sessions: snapshotDshSessions(entries, (id) => mailDrivenSessions.has(id)),
|
||||
};
|
||||
}
|
||||
try {
|
||||
await client.post('/agent/heartbeat', body);
|
||||
} catch {
|
||||
// 心跳失败不报错：网络抖动很常见，下一轮会补上。
|
||||
// 真的持续连不上时 Gateway 会把它判成离线，那才是可见的信号。
|
||||
}
|
||||
}
|
||||
|
||||
ctx.effect(() => {
|
||||
void beat();
|
||||
const timer = setInterval(() => { void beat(); }, 30_000);
|
||||
return () => clearInterval(timer);
|
||||
}, 'dsh-mail-bridge.heartbeat');
|
||||
|
||||
// ─── 获取默认模型 ───
|
||||
|
||||
function modelSelection(): { provider: string; model: string } | undefined {
|
||||
if (REPLY_PROVIDER && REPLY_MODEL) {
|
||||
return { provider: REPLY_PROVIDER, model: REPLY_MODEL };
|
||||
}
|
||||
const defaults = ctx.get('agentDefaultModel');
|
||||
const sel = defaults?.currentSelection?.();
|
||||
if (sel?.provider && sel.model) {
|
||||
return { provider: sel.provider, model: sel.model };
|
||||
}
|
||||
return undefined;
|
||||
}
|
||||
|
||||
// ─── 投递邮件到 DSH 会话 ───
|
||||
|
||||
async function deliverMail(data: any, kind: string): Promise<{ sessionID: string; reused: boolean }> {
|
||||
const mailSessionID = data.session_id;
|
||||
const existing = mailSessionID ? sessionMap.get(mailSessionID) : undefined;
|
||||
|
||||
if (existing) {
|
||||
const live = ctx.agents.get(existing.dshSessionId);
|
||||
if (live) {
|
||||
const promptText = kind === 'permission'
|
||||
? `你之前发起的权限请求已有结论：${data.decision}（决策人：${data.decided_by || '用户'}）。请据此继续后续工作。`
|
||||
: [
|
||||
`本会话收到一封新邮件（AgentMail 续谈）。`,
|
||||
``,
|
||||
`发件人：${data.from_name || 'unknown'}`,
|
||||
`主题：${data.subject || '(无主题)'}`,
|
||||
`邮件 ID：${data.mail_id || 'unknown'}`,
|
||||
``,
|
||||
`请先调用 read_inbox 读取完整正文，然后处理其中的请求。`,
|
||||
`回信不用你自己发：把这一轮做完、把结论说出来就行。`,
|
||||
].join('\n');
|
||||
live.followup(userMessage(promptText));
|
||||
return { sessionID: existing.dshSessionId, reused: true };
|
||||
}
|
||||
}
|
||||
|
||||
// 新开会话
|
||||
const sessionId = `mail-${mailSessionID || Date.now()}`;
|
||||
// cwd 取寻址里的 path 位（to_workspace）。
|
||||
//
|
||||
// 之前这里硬拼 `~/.dsh/mail-sessions/mail-<uuid>` —— 每封邮件一个全新的空目录。
|
||||
// DSH 按 cwd 给会话分组，于是所有邮件会话既不属于任何项目、彼此也不同组，
|
||||
// 界面上全落进「未分组」。path 位本来就是「希望它在哪儿干活」。
|
||||
const { cwd, grouped } = resolveWorkspaceCwd(data.to_workspace, sessionId);
|
||||
ensureCwd(cwd, grouped);
|
||||
if (!grouped && data.to_workspace) {
|
||||
ctx.logger.warn(
|
||||
`[dsh-mail-bridge] 工作目录 ${data.to_workspace} 不可用，回退到 ${cwd}`);
|
||||
}
|
||||
|
||||
const selection = modelSelection();
|
||||
const agentOpts = selection
|
||||
? { provider: selection.provider, model: selection.model }
|
||||
: {};
|
||||
|
||||
const promptText = kind === 'permission'
|
||||
? `你之前发起的权限请求已有结论：${data.decision}（决策人：${data.decided_by || '用户'}）。请据此继续。`
|
||||
: [
|
||||
`你收到一封新邮件（AgentMail）。`,
|
||||
``,
|
||||
`发件人：${data.from_name || 'unknown'}`,
|
||||
`主题：${data.subject || '(无主题)'}`,
|
||||
`邮件 ID：${data.mail_id || 'unknown'}`,
|
||||
`身份：你是 ${AGENT_NAME}`,
|
||||
``,
|
||||
`请先调用 read_inbox 读取完整正文（附带附件清单，如有附件可用 download_attachment 取回），然后处理其中的请求。`,
|
||||
``,
|
||||
`**回信不用你自己发**：你把本轮工作做完、把结论正常说出来就行，`,
|
||||
`插件会在这一轮结束时自动把你最后那段话作为回信发回给 ${data.from_name || '发件人'}（不消耗你的发信配额）。`,
|
||||
`只有在需要主动联系其他人、或要带附件时才调用 send_mail。`,
|
||||
].join('\n');
|
||||
|
||||
const handle = await ctx.agents.create({
|
||||
sessionId,
|
||||
meta: { cwd },
|
||||
agentOptions: agentOpts,
|
||||
// setup 留空：DSH 的 base bundle 已经注册了 agent-loop、llm、tools 等服务。
|
||||
// modelSelection 通过 agentOptions 传入即可 —— 挂载 preset 或
|
||||
// installModelSelection 反而会让 turn 崩溃（实测）。
|
||||
setup: undefined,
|
||||
});
|
||||
|
||||
if (mailSessionID) {
|
||||
sessionMap.set(mailSessionID, { dshSessionId: sessionId, directory: cwd });
|
||||
reverseMap.set(sessionId, mailSessionID);
|
||||
mailDrivenSessions.add(sessionId);
|
||||
mailContexts.set(mailSessionID, {
|
||||
replyTo: data.from_name || '',
|
||||
subject: data.subject || '',
|
||||
mailID: data.mail_id || '',
|
||||
});
|
||||
}
|
||||
|
||||
handle.agent.followup(userMessage(promptText));
|
||||
return { sessionID: sessionId, reused: false };
|
||||
}
|
||||
|
||||
// ─── SSE 监听（与 opencode-mail-bridge 相同的 fetch + reader 模式）───
|
||||
|
||||
let sseAbort: AbortController | 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();
|
||||
}
|
||||
|
||||
// ─── 注册模型工具 ───
|
||||
|
||||
ctx.effect(() => {
|
||||
// send_mail
|
||||
ctx.tools.register(defineTool({
|
||||
name: 'send_mail',
|
||||
description: '发送邮件。三维地址 name@path.session：省略 session 投递到默认会话，.new 强制新建，.具体别名 必须已存在。回复来信请传 reply_to。',
|
||||
parameters: {
|
||||
to: { type: 'string', required: true, description: '收件人三维地址' },
|
||||
subject: { type: 'string', required: true, description: '邮件主题' },
|
||||
body: { type: 'string', required: true, description: '邮件正文（Markdown）' },
|
||||
cc: { type: 'string', description: '抄送，逗号分隔多个三维地址' },
|
||||
reply_to: { type: 'string', description: '回复某封邮件时传其 mail_id' },
|
||||
session_alias: { type: 'string', description: '给新会话命名' },
|
||||
attachment_ids: { type: 'array', items: { type: 'string' }, description: '附件 ID 列表' },
|
||||
},
|
||||
output: {
|
||||
schema: { type: 'string' },
|
||||
render: (_args: any, value: string) => [{ type: 'text', text: value }],
|
||||
},
|
||||
async execute(args: any, toolCtx: any): Promise<string> {
|
||||
const result = await client.post('/mail/send', {
|
||||
to: args.to, subject: args.subject, body: args.body,
|
||||
cc: args.cc || '', reply_to: args.reply_to || '',
|
||||
session_alias: args.session_alias || '',
|
||||
attachment_ids: args.attachment_ids || [],
|
||||
});
|
||||
noteExplicitSend(toolCtx?.sessionID, args.to, args.reply_to);
|
||||
const budget = typeof result.budget_remaining === 'number'
|
||||
? ` 本任务剩余 ${result.budget_remaining}/${result.budget_max} 个来回。` : '';
|
||||
return `邮件已发送（ID: ${result.mail_id}）${budget}`;
|
||||
},
|
||||
}));
|
||||
|
||||
// read_inbox
|
||||
ctx.tools.register(defineTool({
|
||||
name: 'read_inbox',
|
||||
description: '读取收件箱邮件列表。返回最新的邮件，每封含 mail_id、发件人、主题、正文、附件清单。',
|
||||
parameters: {
|
||||
status: { type: 'string', description: '过滤状态（all/unread/read）' },
|
||||
limit: { type: 'number', description: '返回数量上限' },
|
||||
},
|
||||
output: {
|
||||
schema: { type: 'string' },
|
||||
render: (_args: any, value: string) => [{ type: 'text', text: value }],
|
||||
},
|
||||
async execute(args: any): Promise<string> {
|
||||
const { mails } = await client.get(
|
||||
`/mail/inbox?status=${args.status || 'all'}&limit=${args.limit || 20}`
|
||||
);
|
||||
if (!mails?.length) return '收件箱为空。';
|
||||
return mails.map((m: any) => {
|
||||
const att = m.attachments?.length
|
||||
? ` [附件: ${m.attachments.map((a: any) => a.filename).join(', ')}]` : '';
|
||||
return `- ID: ${m.mail_id} | ${m.from_name} | ${m.subject}${att}\n ${m.body.slice(0, 200)}`;
|
||||
}).join('\n');
|
||||
},
|
||||
}));
|
||||
|
||||
// upload_attachment
|
||||
ctx.tools.register(defineTool({
|
||||
name: 'upload_attachment',
|
||||
description: '上传本地文件作为邮件附件，返回 attachment_id。',
|
||||
parameters: {
|
||||
file_path: { type: 'string', required: true, description: '本地文件路径' },
|
||||
},
|
||||
output: {
|
||||
schema: { type: 'string' },
|
||||
render: (_args: any, value: string) => [{ type: 'text', text: value }],
|
||||
},
|
||||
async execute(args: any): Promise<string> {
|
||||
const data = await readFile(args.file_path);
|
||||
const filename = args.file_path.split('/').pop() || 'file';
|
||||
const res = await fetch(`${client.baseURL}/api/v1/attachments`, {
|
||||
method: 'POST',
|
||||
headers: { ...client.authHeaders(), 'Content-Type': 'application/octet-stream', 'X-Filename': filename },
|
||||
body: data,
|
||||
});
|
||||
const json = await res.json() as any;
|
||||
if (!res.ok) throw new Error(json?.error || `HTTP ${res.status}`);
|
||||
const a = json.attachment;
|
||||
return `已上传 ${a.filename}（${a.size_bytes} 字节）。attachment_id: ${a.attachment_id}`;
|
||||
},
|
||||
}));
|
||||
|
||||
// download_attachment
|
||||
ctx.tools.register(defineTool({
|
||||
name: 'download_attachment',
|
||||
description: '下载邮件附件到本地文件。',
|
||||
parameters: {
|
||||
attachment_id: { type: 'string', required: true, description: '附件 ID' },
|
||||
save_path: { type: 'string', required: true, description: '保存路径' },
|
||||
},
|
||||
output: {
|
||||
schema: { type: 'string' },
|
||||
render: (_args: any, value: string) => [{ type: 'text', text: value }],
|
||||
},
|
||||
async execute(args: any): Promise<string> {
|
||||
const res = await fetch(`${client.baseURL}/api/v1/attachments/${args.attachment_id}`, {
|
||||
headers: client.authHeaders(),
|
||||
});
|
||||
if (!res.ok) throw new Error(`下载失败: HTTP ${res.status}`);
|
||||
const buf = Buffer.from(await res.arrayBuffer());
|
||||
await writeFile(args.save_path, buf);
|
||||
return `已保存到 ${args.save_path}（${buf.length} 字节）`;
|
||||
},
|
||||
}));
|
||||
|
||||
return () => {
|
||||
for (const n of ['send_mail', 'read_inbox', 'upload_attachment', 'download_attachment']) {
|
||||
try { ctx.tools.unregister(n); } catch {}
|
||||
}
|
||||
};
|
||||
}, 'dsh-mail-bridge.tools');
|
||||
|
||||
// ─── turn 完成后自动转发回复 ───
|
||||
//
|
||||
// 与 opencode-mail-bridge 的 session.idle 同一职责：模型把话说完了，
|
||||
// 插件把它最后那段话搬到邮件里 —— 不该让模型自己记得调 send_mail。
|
||||
ctx.on('agent/status', async (payload: any) => {
|
||||
if (payload?.status !== 'idle') return;
|
||||
const agent = payload.agent;
|
||||
if (!agent?.id) return;
|
||||
const mailSessionID = reverseMap.get(String(agent.id));
|
||||
if (!mailSessionID) return;
|
||||
const mctx = mailContexts.get(mailSessionID);
|
||||
if (!mctx?.replyTo) return;
|
||||
|
||||
// 取最后一条 assistant 消息的文本
|
||||
const events = agent.session?.events ?? [];
|
||||
const lastText = lastAssistantText(events);
|
||||
if (!lastText) return;
|
||||
|
||||
// 本轮模型已亲手回过这条线索 → 不再自动转发（与 opencode 侧同一取舍）
|
||||
if (shouldSkipAutoRelay(explicitSends.get(String(agent.id)), mctx.replyTo, mctx.mailID)) {
|
||||
explicitSends.delete(String(agent.id));
|
||||
return;
|
||||
}
|
||||
// 同一条消息只转一次
|
||||
if (relayedSummaries.get(String(agent.id)) === lastText) return;
|
||||
|
||||
try {
|
||||
await client.post('/mail/send', {
|
||||
to: mctx.replyTo,
|
||||
subject: replySubject(mctx.subject),
|
||||
body: lastText,
|
||||
reply_to: mctx.mailID || '',
|
||||
// relay + relay_key：走免配额通道（harness 的搬运不该收费）
|
||||
relay: 'summary',
|
||||
relay_key: `${agent.id}:${events.length}`,
|
||||
});
|
||||
relayedSummaries.set(String(agent.id), lastText);
|
||||
explicitSends.delete(String(agent.id));
|
||||
ctx.logger.info(`[dsh-mail-bridge] 总结已回信 ${mctx.replyTo}（不计配额）`);
|
||||
} catch (e: any) {
|
||||
ctx.logger.error(`[dsh-mail-bridge] 转发回复失败: ${e?.message || e}`);
|
||||
}
|
||||
|
||||
// 会话命名回写：DSH 首轮结束后由模型生成摘要标题，把它同步回 AgentMail。
|
||||
// 与 opencode 侧同一决定：不另造一套命名，平台叫什么这边就叫什么。
|
||||
//
|
||||
// DSH 没有 opencode 那样的 slug，别名由标题派生（slugFromTitle 会去掉
|
||||
// `.` `@` `/` 这些寻址分隔符 —— 留在别名里会让它自己被解析器切开）。
|
||||
// 别名与标题一起发：服务端撞名时自动追 -2/-3 后缀，并尊重人工改过的别名。
|
||||
const title = modelTitle(events);
|
||||
if (title && syncedTitles.get(String(agent.id)) !== title) {
|
||||
syncedTitles.set(String(agent.id), title);
|
||||
const alias = slugFromTitle(title);
|
||||
try {
|
||||
await client.post(`/sessions/${mailSessionID}/sync`,
|
||||
alias ? { title, alias } : { title });
|
||||
ctx.logger.info(
|
||||
`[dsh-mail-bridge] 会话命名已同步: title=${title}${alias ? ` alias=${alias}` : ''}`);
|
||||
} catch (e: any) {
|
||||
ctx.logger.error(`[dsh-mail-bridge] 命名同步失败: ${e?.message || e}`);
|
||||
}
|
||||
}
|
||||
});
|
||||
|
||||
// ─── 平台原生权限询问 → 转成邮件问人 ───
|
||||
//
|
||||
// 这是 harness 的职责，不该让模型自己调一个 request_permission 工具：
|
||||
// 模型可能忘了调，也可能在不需要时乱调，而真正被 DSH 拦下的那次询问反而没人看见。
|
||||
//
|
||||
// 与 opencode 的差异：opencode 的 permission.ask 是同步钩子，卡在里面会把整个
|
||||
// 请求挂住，所以那边只能“转出去就返回 ask”；而 DSH 的 approval/request 是
|
||||
// **异步 waterfall**，返回 Promise<ApprovalOutcome> 就是它要的，因此可以真的等人。
|
||||
ctx.on('approval/request', async (req: any, next: () => Promise<string>) => {
|
||||
const agentId = String(req?.agent?.id ?? '');
|
||||
if (!mailDrivenSessions.has(agentId)) return next(); // 非邮件驱动的会话不接管
|
||||
const mailSessionID = reverseMap.get(agentId);
|
||||
if (!mailSessionID) return next();
|
||||
|
||||
// DSH 不给询问发 id，用 (会话, 工具, callId) 做幂等键。
|
||||
const relayKey = `${agentId}:${req.toolName}:${req.callId ?? 'nocall'}`;
|
||||
|
||||
try {
|
||||
await client.post('/permission/request', {
|
||||
question: `请求执行 ${req.toolName}`,
|
||||
options: ['同意', '拒绕'],
|
||||
context: [
|
||||
`工具：${req.toolName}`,
|
||||
req.callId ? `调用 ID：${req.callId}` : '',
|
||||
req.reason ? `理由：${req.reason}` : '',
|
||||
].filter(Boolean).join('\n'),
|
||||
session_id: mailSessionID,
|
||||
relay_key: relayKey,
|
||||
});
|
||||
} catch (e: any) {
|
||||
// 转不出去就别把 DSH 挂在那儿等：交给下一个 answerer（本地 UI）接管。
|
||||
ctx.logger.error(`[dsh-mail-bridge] 权限询问转发失败: ${e?.message || e}`);
|
||||
return next();
|
||||
}
|
||||
|
||||
ctx.logger.info(`[dsh-mail-bridge] 权限询问已转邮件 ${relayKey}`);
|
||||
|
||||
// 等人类决策；DSH 撤销询问（signal abort）时结算为 cancelled。
|
||||
return new Promise<string>((resolve) => {
|
||||
pendingApprovals.set(relayKey, { resolve, sessionId: agentId });
|
||||
req.signal?.addEventListener('abort', () => {
|
||||
if (pendingApprovals.delete(relayKey)) resolve('cancelled');
|
||||
}, { once: true });
|
||||
});
|
||||
});
|
||||
|
||||
/** 人类决策回来：先看是不是在等的那条 approval，否则当普通通知投给会话。 */
|
||||
function handlePermissionDecision(data: any): void {
|
||||
const relayKey = String(data?.relay_key ?? '');
|
||||
const pending = relayKey ? pendingApprovals.get(relayKey) : undefined;
|
||||
if (pending) {
|
||||
pendingApprovals.delete(relayKey);
|
||||
// AgentMail 的选项文本 → DSH 的 ApprovalOutcome。
|
||||
// 只有“同意”才放行，其余（包括认不出的选项）一律 fail closed。
|
||||
const decision = String(data?.decision ?? '');
|
||||
const outcome = /^(同意|allow|approve|yes)/i.test(decision) ? 'allowed-once' : 'rejected';
|
||||
pending.resolve(outcome);
|
||||
ctx.logger.info(`[dsh-mail-bridge] 权限决策 ${relayKey} -> ${outcome}`);
|
||||
return;
|
||||
}
|
||||
// 没在等（插件重启后丢了 pendingApprovals，或历史数据）→ 当一封通知投进会话。
|
||||
deliverMail(data, 'permission')
|
||||
.catch((e: any) => ctx.logger.error(`[dsh-mail-bridge] 权限决策投递失败: ${e?.message || e}`));
|
||||
}
|
||||
|
||||
// ─── 启动 SSE（与 opencode-mail-bridge 相同的事件处理）───
|
||||
|
||||
ctx.effect(() => {
|
||||
startSSE((type, data) => {
|
||||
switch (type) {
|
||||
case 'new_mail':
|
||||
deliverMail(data, 'mail')
|
||||
.then(({ sessionID, reused }) => {
|
||||
console.error(`[dsh-mail-bridge] ${type} -> ${reused ? '续谈' : '新会话'} ${sessionID}`);
|
||||
})
|
||||
.catch((e: any) => {
|
||||
ctx.logger.error(`[dsh-mail-bridge] ${type} 处理失败: ${e?.message || e}`);
|
||||
});
|
||||
break;
|
||||
case 'permission_decision':
|
||||
handlePermissionDecision(data);
|
||||
break;
|
||||
}
|
||||
});
|
||||
return () => {
|
||||
sseAbort?.abort();
|
||||
sseAbort = null;
|
||||
// 拆插件时没人再能回答待决询问，一律 fail closed，
|
||||
// 否则 DSH 侧那些 await 永远不会返回。
|
||||
for (const [key, pending] of pendingApprovals) {
|
||||
pending.resolve('unavailable');
|
||||
pendingApprovals.delete(key);
|
||||
}
|
||||
};
|
||||
}, 'dsh-mail-bridge.sse');
|
||||
}
|
||||
Reference in New Issue
Block a user