/** pi-ntfy runtime event wiring. * * Event wiring (see DESIGN.md for the reasoning): * * agent_start → remember when the run started * agent_end → remember the terminal stopReason of this low-level run * agent_settled → the single send point after retries and compaction * tool_call → pi is asking the user something * session_shutdown → the session is going away */ import type { ExtensionAPI } from "@earendil-works/pi-coding-agent"; import { invalidateConfigCache, loadConfig } from "./config.ts"; import { filterConfigToChannel, handlePushCommand } from "./command.ts"; import { createNotifier, type PayloadInput } from "./notifier.ts"; import { summarizeError } from "./render.ts"; import type { ChannelResult, NotifyConfig } from "./types.ts"; interface RunState { startedAt: number; stopReason?: string; errorMessage?: string; task?: string; summary?: string; tools: Set; } const ERROR_REASON_CHARS = 120; const CONTEXT_CHARS = 180; function compactText(value: string | undefined, max = CONTEXT_CHARS): string { if (!value) return ""; const text = value .replace(/```[\s\S]*?```/g, "[代码]") .replace(/\s+/g, " ") .trim(); return text.length > max ? `${text.slice(0, max - 1)}…` : text; } function assistantText(message: { content?: unknown }): string { if (!Array.isArray(message.content)) return ""; return message.content .filter((part): part is { type: "text"; text: string } => Boolean( part && typeof part === "object" && (part as { type?: unknown }).type === "text" && typeof (part as { text?: unknown }).text === "string", ), ) .map((part) => part.text) .join(" "); } function describeResults(results: ChannelResult[]): string { if (results.length === 0) return "没有匹配的渠道"; return results .map((r) => r.ok ? `${r.channel} ok(${r.status}, ${r.ms}ms)` : `${r.channel} 失败(${r.error ?? "unknown"})`, ) .join("; "); } export default function piNtfy(pi: ExtensionAPI) { const notifier = createNotifier(); let run: RunState | undefined; let runtimeEnabled = true; let currentTask = ""; /** Config for the current context, or undefined when this run must stay quiet. */ function activeConfig(mode: string): NotifyConfig | undefined { if (!runtimeEnabled) return undefined; const config = loadConfig(); if (!config.enabled || !config.exists) return undefined; // All modes push unless listed in excludedModes (e.g. subagents on json). if (config.excludedModes.includes(mode)) return undefined; return config; } function baseInput( ctx: { cwd: string; mode?: string; sessionManager: { getSessionName(): string | undefined; getSessionFile(): string | undefined; getSessionId?(): string; }; model: { id: string; provider: string } | undefined; thinkingLevel?: string; }, state?: RunState, ): Pick< PayloadInput, | "cwd" | "session" | "sessionId" | "model" | "mode" | "thinking" | "task" | "summary" | "tools" > { const file = ctx.sessionManager.getSessionFile(); const session = ctx.sessionManager.getSessionName() ?? (file ? file .split("/") .pop() ?.replace(/\.jsonl?$/, "") : undefined); return { cwd: ctx.cwd, session: session ?? "", sessionId: ctx.sessionManager.getSessionId?.() ?? "", model: ctx.model ? `${ctx.model.provider}/${ctx.model.id}` : "", mode: ctx.mode ?? "", thinking: ctx.thinkingLevel ?? "", task: compactText(state?.task), summary: compactText(state?.summary), tools: state ? [...state.tools].join(", ") : "", }; } pi.on("session_start", async () => { run = undefined; currentTask = ""; invalidateConfigCache(); }); pi.on("input", async (event) => { const text = event.text.trim(); if (!text || text.startsWith("/")) return; currentTask = compactText(text); if (run) run.task = currentTask; }); pi.on("agent_start", async () => { run = { startedAt: Date.now(), task: currentTask, tools: new Set() }; }); pi.on("agent_end", async (event) => { if (!run) run = { startedAt: Date.now(), task: currentTask, tools: new Set() }; for (let i = event.messages.length - 1; i >= 0; i--) { const message = event.messages[i]; if (message.role !== "assistant") continue; const assistant = message as { stopReason?: string; errorMessage?: string; content?: unknown; }; run.stopReason = assistant.stopReason; run.errorMessage = assistant.errorMessage; run.summary = compactText(assistantText(assistant)); return; } }); pi.on("tool_execution_start", async (event) => { if (!run) return; run.tools.add(event.toolName); }); pi.on("agent_settled", async (_event, ctx) => { const state = run; run = undefined; const config = activeConfig(ctx.mode); if (!config) return; const durationMs = state ? Date.now() - state.startedAt : undefined; const stopReason = state?.stopReason; if (stopReason === "aborted" || stopReason === "error") { const reason = stopReason === "aborted" ? "本轮被取消" : `运行出错:${summarizeError(state?.errorMessage, ERROR_REASON_CHARS) || "未知错误"}`; notifier.fire(config, { event: "interrupted", status: "已中断", reason, durationMs, ...baseInput(ctx, state), }); return; } // Short runs are usually the user still sitting at the keyboard. if (durationMs !== undefined && durationMs < config.minDurationSec * 1000) return; notifier.fire(config, { event: "idle", status: "已就绪", reason: "输出结束,等待输入", durationMs, ...baseInput(ctx, state), }); }); pi.on("tool_call", async (event, ctx) => { const config = activeConfig(ctx.mode); if (!config) return; if (!config.needInputTools.includes(event.toolName)) return; notifier.fire(config, { event: "needInput", status: "需要确认", reason: "pi 正在等你回答问题", durationMs: run ? Date.now() - run.startedAt : undefined, ...baseInput(ctx, run), }); }); pi.on("session_shutdown", async (event, ctx) => { const config = activeConfig(ctx.mode); if (!config) { await notifier.flush(1000); return; } if (event.reason === "quit") { // pi awaits this handler with no timeout of its own, so the cap is ours. await notifier.fireBlocking( config, { event: "exit", status: "已退出", reason: "会话结束", ...baseInput(ctx, run), }, config.shutdownTimeoutMs, ); } // Drain notifications that were fired moments before the exit. await notifier.flush(config.shutdownTimeoutMs); }); pi.registerCommand("ntfy", { description: "pi-ntfy: status|list|get|set|enable|disable|test|events|on|off|help", handler: async (args, ctx) => { let result: ReturnType; try { result = handlePushCommand(args, { runtimeEnabled, setRuntimeEnabled: (v) => { runtimeEnabled = v; }, last: (() => { const last = notifier.lastResults(); return last ? { at: last.at, event: last.event, results: last.results } : undefined; })(), describeResults, }); } catch (error) { ctx.ui.notify( error instanceof Error ? error.message : String(error), "error", ); return; } if (result.runTest) { invalidateConfigCache(); let config = loadConfig(); if (!config.exists && config.channels.length === 0) { ctx.ui.notify(`未找到配置文件:${config.path}`, "error"); return; } try { if (result.testFilter) { config = filterConfigToChannel(config, result.testFilter); } } catch (error) { ctx.ui.notify( error instanceof Error ? error.message : String(error), "error", ); return; } const results = await notifier.sendTest( { ...config, enabled: true }, { event: "test", status: "测试", reason: result.testFilter ? `pi-ntfy 测试 (${result.testFilter})` : "pi-ntfy 通道测试", durationMs: 0, ...baseInput(ctx, run), }, ); const ok = results.length > 0 && results.every((r) => r.ok); ctx.ui.notify( `pi-ntfy 测试:${describeResults(results)}`, ok ? "info" : "error", ); return; } ctx.ui.notify(result.message, result.ok ? "info" : "error"); }, }); }