From 7dc17e1a9f901c025121fb2e9c8e317e5436f409 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=9D=A8=E8=B1=AA?= Date: Thu, 17 Sep 2026 09:50:46 +0800 Subject: [PATCH] feat: add pi-ntfy NTFY notification plugin --- .codegraph/.gitignore | 5 + .gitea/workflows/publish-npm.yml | 31 +++ DESIGN.md | 44 ++++ LICENSE | 21 ++ README.md | 72 ++++++ channels/index.ts | 21 ++ channels/ntfy.ts | 110 +++++++++ command.ts | 236 ++++++++++++++++++ config.example.json | 33 +++ config.ts | 397 +++++++++++++++++++++++++++++++ http.ts | 105 ++++++++ index.ts | 311 ++++++++++++++++++++++++ notifier.ts | 275 +++++++++++++++++++++ package.json | 54 +++++ render.ts | 118 +++++++++ types.ts | 69 ++++++ 16 files changed, 1902 insertions(+) create mode 100644 .codegraph/.gitignore create mode 100644 .gitea/workflows/publish-npm.yml create mode 100644 DESIGN.md create mode 100644 LICENSE create mode 100644 README.md create mode 100644 channels/index.ts create mode 100644 channels/ntfy.ts create mode 100644 command.ts create mode 100644 config.example.json create mode 100644 config.ts create mode 100644 http.ts create mode 100644 index.ts create mode 100644 notifier.ts create mode 100644 package.json create mode 100644 render.ts create mode 100644 types.ts diff --git a/.codegraph/.gitignore b/.codegraph/.gitignore new file mode 100644 index 0000000..d20c0fe --- /dev/null +++ b/.codegraph/.gitignore @@ -0,0 +1,5 @@ +# CodeGraph data files — local to each machine, not for committing. +# Ignore everything in .codegraph/ except this file itself, so transient +# files (the database, daemon.pid, sockets, logs) never show up in git. +* +!.gitignore diff --git a/.gitea/workflows/publish-npm.yml b/.gitea/workflows/publish-npm.yml new file mode 100644 index 0000000..64afb32 --- /dev/null +++ b/.gitea/workflows/publish-npm.yml @@ -0,0 +1,31 @@ +name: Publish pi-ntfy to npm + +on: + workflow_dispatch: + push: + tags: + - "v*" + +jobs: + publish: + runs-on: ubuntu-latest + steps: + - name: Checkout + uses: actions/checkout@v4 + + - name: Setup Node.js + uses: actions/setup-node@v4 + with: + node-version: "22" + registry-url: "https://registry.npmjs.org" + + - name: Verify package + run: | + node --version + npm --version + npm pack --dry-run + + - name: Publish to npm + run: npm publish --access public + env: + NODE_AUTH_TOKEN: ${{ secrets.NPM_PACKAGE_TOKEN }} diff --git a/DESIGN.md b/DESIGN.md new file mode 100644 index 0000000..4d60c7f --- /dev/null +++ b/DESIGN.md @@ -0,0 +1,44 @@ +# pi-ntfy 设计说明 + +## 目标 + +只通过 NTFY 推送 pi Agent 的状态，并让并行开发时可以区分项目、Session 和任务。 + +## 事件 + +- `agent_start`：记录开始时间和当前任务 +- `agent_end`：记录最终 assistant 摘要和停止原因 +- `agent_settled`：唯一的 idle/interrupted 推送点 +- `tool_execution_start`：记录本轮使用过的工具 +- `tool_call`：检测需要用户回答的工具 +- `session_shutdown`：按配置发送退出通知 + +## 内容 + +标题默认包含： + +```text +[项目] 状态 · Session +``` + +正文默认每项独占一行： + +```text +项目 +会话 +任务 +状态 +原因 +摘要 +耗时 +模型 +主机 +时间 +工具 +``` + +## 通道 + +本插件只有一个通道：`ntfy`。其他推送通道和对应实现已移除。 + +NTFY 元数据使用 `Title`、`Tags`、`X-Project`、`X-Session`、`X-Event`、`X-Task` headers。含中文的字段会写入 Base64 版本的 headers，完整中文内容始终保留在正文中。 diff --git a/LICENSE b/LICENSE new file mode 100644 index 0000000..558b190 --- /dev/null +++ b/LICENSE @@ -0,0 +1,21 @@ +MIT License + +Copyright (c) 2026 kiim-wong + +Permission is hereby granted, free of charge, to any person obtaining a copy +of this software and associated documentation files (the "Software"), to deal +in the Software without restriction, including without limitation the rights +to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +copies of the Software, and to permit persons to whom the Software is +furnished to do so, subject to the following conditions: + +The above copyright notice and this permission notice shall be included in all +copies or substantial portions of the Software. + +THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE +SOFTWARE. diff --git a/README.md b/README.md new file mode 100644 index 0000000..02a2c51 --- /dev/null +++ b/README.md @@ -0,0 +1,72 @@ +# pi-ntfy + +Pi coding agent 的 NTFY 专用通知插件，适合并行开发时区分项目、Session 和任务。 + +## 特性 + +- 仅支持 NTFY，移除 Bark、飞书、企业微信、钉钉和通用 Webhook +- 插件名称：`pi-ntfy` +- 标题包含项目、状态和 Session 名称 +- 正文包含项目、Session、任务、摘要、状态、原因 +- 耗时、模型、主机、时间各自独占一行 +- NTFY 使用状态标签和项目/Session/任务元数据 headers +- 支持环境变量保存 NTFY token + +## 安装 + +```bash +pi remove /home/yanghao05/Workspace/pi-push +pi install /home/yanghao05/Workspace/pi-push +``` + +不要同时安装或启用旧版 `npm:pi-agent-push`。 + +## 配置 + +```bash +cp config.example.json config.json +``` + +默认配置： + +```json +{ + "titleTemplate": "[{{project}}] {{status}} · {{session}}", + "template": "项目: {{project}}\n会话: {{session}}\n任务: {{task}}\n状态: {{status}}\n原因: {{reason}}\n摘要: {{summary}}\n耗时: {{duration}}\n模型: {{model}}\n主机: {{host}}\n时间: {{date}} {{time}}\n工具: {{tools}}", + "channels": [ + { + "type": "ntfy", + "enabled": true, + "topic": "$NTFY_TOPIC", + "server": "$NTFY_HOST", + "token": "$NTFY_TOKEN" + } + ] +} +``` + +支持的占位符： + +```text +{{status}} {{reason}} {{text}} {{title}} {{event}} +{{cwd}} {{project}} {{duration}} {{session}} {{sessionId}} +{{model}} {{mode}} {{thinking}} {{task}} {{summary}} {{tools}} +{{host}} {{time}} {{date}} +``` + +## 命令 + +```text +/ntfy +/ntfy list +/ntfy get +/ntfy set topic=xxx enabled=true +/ntfy enable +/ntfy disable +/ntfy test +/ntfy events idle=on interrupted=on needInput=on exit=off +/ntfy on +/ntfy off +``` + +旧的 `/push` 命令不再使用。 diff --git a/channels/index.ts b/channels/index.ts new file mode 100644 index 0000000..cab66c1 --- /dev/null +++ b/channels/index.ts @@ -0,0 +1,21 @@ +/** pi-ntfy channel registry. */ + +import type { HttpResult } from "../http.ts"; +import { localFailure } from "../http.ts"; +import type { NotifyPayload, NtfyChannel } from "../types.ts"; +import { sendNtfy } from "./ntfy.ts"; + +export function channelName(channel: NtfyChannel, index: number): string { + return channel.name?.trim() || `ntfy#${index + 1}`; +} + +export function sendToChannel( + channel: NtfyChannel, + payload: NotifyPayload, + timeoutMs: number, +): Promise { + if (channel.type !== "ntfy") { + return Promise.resolve(localFailure(`未知渠道类型: ${channel.type}`)); + } + return sendNtfy(channel, payload, timeoutMs); +} diff --git a/channels/ntfy.ts b/channels/ntfy.ts new file mode 100644 index 0000000..ffc26a5 --- /dev/null +++ b/channels/ntfy.ts @@ -0,0 +1,110 @@ +/** pi-ntfy NTFY channel. + * + * Fetch/HTTP headers are ByteString-only. Keep headers ASCII-safe and leave the + * full UTF-8 project/session/task information in the message body. Encoded + * metadata headers are added for consumers that need machine-readable values. + */ + +import { httpRequest, localFailure, type HttpResult } from "../http.ts"; +import type { NotifyPayload, NtfyChannel } from "../types.ts"; + +const DEFAULT_SERVER = "https://ntfy.sh"; + +function eventTagFor(event: NotifyPayload["event"]): string { + switch (event) { + case "idle": + return "white_check_mark"; + case "interrupted": + return "warning"; + case "needInput": + return "question"; + case "exit": + return "door"; + default: + return "test_tube"; + } +} + +function isAscii(value: string): boolean { + return [...value].every((char) => char.charCodeAt(0) <= 255); +} + +function base64(value: string): string { + return Buffer.from(value, "utf8").toString("base64url"); +} + +function addErrorDetail(result: HttpResult): HttpResult { + if (result.ok) return result; + let detail = result.error ?? `HTTP ${result.status}`; + if (result.body) { + try { + const data = JSON.parse(result.body) as { error?: string; message?: string }; + detail = data.error || data.message || detail; + } catch { + detail = result.body.slice(0, 200) || detail; + } + } + return { ...result, error: `ntfy ${detail}` }; +} + +export async function sendNtfy( + channel: NtfyChannel, + payload: NotifyPayload, + timeoutMs: number, +): Promise { + const topic = (channel.topic ?? "").trim(); + if (!topic) return localFailure("ntfy: topic 为空"); + + const server = (channel.server ?? DEFAULT_SERVER).trim().replace(/\/+$/, ""); + const headers: Record = { + "Content-Type": "text/plain; charset=utf-8", + }; + const token = (channel.token ?? "").trim(); + if (token) { + headers.Authorization = token.toLowerCase().startsWith("bearer ") + ? token + : `Bearer ${token}`; + } + + const project = payload.vars.project ?? ""; + const session = payload.vars.session ?? ""; + const task = payload.vars.task ?? ""; + const asciiSession = isAscii(session) ? session : ""; + const asciiTitle = isAscii(payload.title) + ? payload.title + : `[${project || "pi"}] pi ${payload.event}${asciiSession ? ` · ${asciiSession}` : ""}`; + + // NTFY headers must be ASCII-safe under Node's fetch implementation. + headers.Title = asciiTitle; + headers["X-Event"] = payload.event; + if (isAscii(project)) headers["X-Project"] = project; + if (asciiSession) headers["X-Session"] = asciiSession; + if (isAscii(task) && task) headers["X-Task"] = task; + if (project) headers["X-Project-B64"] = base64(project); + if (session) headers["X-Session-B64"] = base64(session); + if (task) headers["X-Task-B64"] = base64(task); + + if (channel.priority != null && channel.priority !== "") { + headers.Priority = String(channel.priority); + } + const configuredTags = + channel.tags == null + ? "" + : Array.isArray(channel.tags) + ? channel.tags.join(",") + : String(channel.tags); + const tags = [configuredTags, eventTagFor(payload.event)] + .filter(Boolean) + .join(","); + if (tags) headers.Tags = tags; + if (channel.clickUrl) headers.Click = channel.clickUrl; + if (channel.icon) headers.Icon = channel.icon; + if (channel.email) headers.Email = channel.email; + + const result = await httpRequest( + `${server}/${encodeURIComponent(topic)}`, + { method: "POST", headers, body: payload.text }, + timeoutMs, + ); + return addErrorDetail(result); +} diff --git a/command.ts b/command.ts new file mode 100644 index 0000000..ca57387 --- /dev/null +++ b/command.ts @@ -0,0 +1,236 @@ +/** pi-ntfy command parser and config mutations. */ + +import { + CHANNEL_SETTABLE_FIELDS, + formatChannelLabel, + invalidateConfigCache, + loadConfig, + loadRawConfig, + maskSecret, + saveRawConfig, + type RawConfigFile, +} from "./config.ts"; +import type { ChannelResult, NotifyConfig } from "./types.ts"; + +export interface CommandResult { + ok: boolean; + message: string; + runTest?: boolean; + testFilter?: string; +} + +function usage(): string { + return [ + "用法:", + " /ntfy 状态", + " /ntfy list 查看 ntfy 配置", + " /ntfy get 查看配置（密钥脱敏）", + " /ntfy enable|disable 开关 ntfy", + " /ntfy set key=value ... 设置 topic/token/server 等", + " /ntfy test 发送测试通知", + " /ntfy on | off 当前会话总开关", + " /ntfy events [k=on|off…] 设置事件开关", + " /ntfy help 帮助", + "", + "示例:", + " /ntfy set topic=my-topic token=$NTFY_TOKEN enabled=true", + " /ntfy test", + ].join("\n"); +} + +export function tokenize(args: string): string[] { + const out: string[] = []; + const re = /"([^"\\]|\\.)*"|'([^'\\]|\\.)*'|\S+/g; + for (const match of args.matchAll(re)) { + let token = match[0]; + if ( + (token.startsWith('"') && token.endsWith('"')) || + (token.startsWith("'") && token.endsWith("'")) + ) { + token = token.slice(1, -1).replace(/\\([\\'"])/g, "$1"); + } + out.push(token); + } + return out; +} + +function parseBool(raw: string): boolean { + const value = raw.trim().toLowerCase(); + if (["1", "true", "on", "yes", "开", "启用"].includes(value)) return true; + if (["0", "false", "off", "no", "关", "禁用"].includes(value)) return false; + throw new Error(`无法解析布尔值: ${raw}（用 true/false 或 on/off）`); +} + +function coerceField(key: string, raw: string): string | number | boolean { + if (key === "enabled") return parseBool(raw); + if (key === "timeoutMs" || key === "priority") + return /^\d+$/.test(raw) ? Number(raw) : raw; + return raw; +} + +function channelData(data: RawConfigFile): Array> { + if (!Array.isArray(data.channels)) data.channels = []; + return data.channels; +} + +function ntfyIndex(data: RawConfigFile): number { + const channels = channelData(data); + const index = channels.findIndex( + (channel) => String(channel.type ?? "ntfy") === "ntfy", + ); + if (index >= 0) return index; + channels.push({ type: "ntfy", name: "ntfy", enabled: true }); + return channels.length - 1; +} + +function applyPairs( + channel: Record, + pairs: string[], +): string[] { + const allowed = new Set(CHANNEL_SETTABLE_FIELDS.ntfy); + const changed: string[] = []; + for (const pair of pairs) { + const separator = pair.indexOf("="); + if (separator <= 0) throw new Error(`参数应为 key=value，收到: ${pair}`); + const key = pair.slice(0, separator).trim(); + if (!allowed.has(key)) + throw new Error( + `ntfy 不支持字段 "${key}"。可设: ${[...allowed].join(", ")}`, + ); + channel[key] = coerceField(key, pair.slice(separator + 1)); + changed.push(key); + } + if (changed.length === 0) throw new Error("请至少提供一个 key=value"); + return changed; +} + +function describeConfig( + config: NotifyConfig, + runtimeEnabled: boolean, + last?: { at: number; event: string; results: ChannelResult[] }, + describeResults?: (results: ChannelResult[]) => string, +): string { + const channel = config.channels[0]; + const lines = [ + `pi-ntfy: ${runtimeEnabled && config.enabled ? "已启用" : "已停用"}${config.exists ? "" : "（无配置文件）"}`, + `配置: ${config.path}`, + `渠道: ${channel ? formatChannelLabel(channel, 0) : "ntfy（未配置）"}`, + `事件: ${(["idle", "interrupted", "needInput", "exit"] as const).map((key) => `${key}=${config.events[key] ? "on" : "off"}`).join(" ")}`, + `模式: ${config.modes.join(",")}`, + ]; + if (last && describeResults) + lines.push( + `上次推送: ${new Date(last.at).toTimeString().slice(0, 8)} ${last.event} → ${describeResults(last.results)}`, + ); + else lines.push("上次推送: (无)"); + if (config.warnings.length > 0) + lines.push(`警告: ${config.warnings.join("; ")}`); + return lines.join("\n"); +} + +export function handlePushCommand( + args: string, + options: { + runtimeEnabled: boolean; + setRuntimeEnabled: (enabled: boolean) => void; + last?: { at: number; event: string; results: ChannelResult[] }; + describeResults: (results: ChannelResult[]) => string; + }, +): CommandResult { + const tokens = tokenize(args.trim()); + const action = (tokens[0] ?? "").toLowerCase(); + if (!action || action === "status") { + invalidateConfigCache(); + const config = loadConfig(); + return { + ok: true, + message: describeConfig( + config, + options.runtimeEnabled, + options.last, + options.describeResults, + ), + }; + } + if (["help", "-h", "--help"].includes(action)) + return { ok: true, message: usage() }; + if (action === "on" || action === "off") { + options.setRuntimeEnabled(action === "on"); + return { + ok: true, + message: `pi-ntfy 已${action === "on" ? "开启" : "关闭"}（仅当前会话）`, + }; + } + if (action === "test") return { ok: true, message: "", runTest: true }; + if (action === "list" || action === "get") { + const raw = loadRawConfig(); + const channels = channelData(raw.data); + if (channels.length === 0) return { ok: true, message: "没有配置 ntfy" }; + const channel = channels[0]; + if (action === "list") + return { + ok: true, + message: `${formatChannelLabel(channel, 0)} topic=${maskSecret("topic", channel.topic)}`, + }; + return { + ok: true, + message: [ + formatChannelLabel(channel, 0), + ...CHANNEL_SETTABLE_FIELDS.ntfy + .filter((key) => key in channel) + .map((key) => `${key}=${maskSecret(key, channel[key])}`), + ].join("\n"), + }; + } + if (action === "enable" || action === "disable") { + const raw = loadRawConfig(); + const index = ntfyIndex(raw.data); + channelData(raw.data)[index].enabled = action === "enable"; + const path = saveRawConfig(raw.data); + return { + ok: true, + message: `ntfy 已${action === "enable" ? "开启" : "关闭"}\n已写入 ${path}`, + }; + } + if (action === "set") { + const raw = loadRawConfig(); + const index = ntfyIndex(raw.data); + const changed = applyPairs(channelData(raw.data)[index], tokens.slice(1)); + const path = saveRawConfig(raw.data); + return { + ok: true, + message: `已更新 ntfy: ${changed.map((key) => `${key}=${maskSecret(key, channelData(raw.data)[index][key])}`).join(" ")}\n已写入 ${path}`, + }; + } + if (action === "events") { + const raw = loadRawConfig(); + const events = (raw.data.events ??= { + idle: true, + interrupted: true, + needInput: true, + exit: false, + }); + for (const pair of tokens.slice(1)) { + const separator = pair.indexOf("="); + if (separator <= 0) throw new Error(`参数应为 event=on|off，收到: ${pair}`); + const key = pair.slice(0, separator); + if (!(key in events)) throw new Error(`未知事件 "${key}"`); + events[key] = parseBool(pair.slice(separator + 1)); + } + if (tokens.length > 1) saveRawConfig(raw.data); + return { + ok: true, + message: Object.entries(events) + .map(([key, value]) => `${key}=${value ? "on" : "off"}`) + .join(" "), + }; + } + return { ok: false, message: `未知子命令 "${action}"\n${usage()}` }; +} + +export function filterConfigToChannel( + config: NotifyConfig, + _selector: string, +): NotifyConfig { + return { ...config, channels: config.channels.slice(0, 1) }; +} diff --git a/config.example.json b/config.example.json new file mode 100644 index 0000000..8812eff --- /dev/null +++ b/config.example.json @@ -0,0 +1,33 @@ +{ + "_说明": "pi-ntfy 只支持 NTFY。复制为 config.json 后按需修改。", + "_占位符": "{{status}} {{reason}} {{text}} {{title}} {{event}} {{cwd}} {{project}} {{duration}} {{session}} {{sessionId}} {{model}} {{mode}} {{thinking}} {{task}} {{summary}} {{tools}} {{host}} {{time}} {{date}}", + "enabled": true, + "modes": ["tui"], + "timeoutMs": 5000, + "shutdownTimeoutMs": 2000, + "dedupeMs": 3000, + "minDurationSec": 0, + "maxTextChars": 2000, + "debug": false, + "titleTemplate": "[{{project}}] {{status}} · {{session}}", + "template": "项目: {{project}}\n会话: {{session}}\n任务: {{task}}\n状态: {{status}}\n原因: {{reason}}\n摘要: {{summary}}\n耗时: {{duration}}\n模型: {{model}}\n主机: {{host}}\n时间: {{date}} {{time}}\n工具: {{tools}}", + "events": { + "idle": true, + "interrupted": true, + "needInput": true, + "exit": false + }, + "needInputTools": ["ask_user_question", "plan_mode_question"], + "channels": [ + { + "type": "ntfy", + "name": "手机 ntfy", + "enabled": true, + "topic": "$NTFY_TOPIC", + "server": "$NTFY_HOST", + "token": "$NTFY_TOKEN", + "priority": "default", + "tags": "pi,computer" + } + ] +} diff --git a/config.ts b/config.ts new file mode 100644 index 0000000..2851e4c --- /dev/null +++ b/config.ts @@ -0,0 +1,397 @@ +/** pi-ntfy configuration loading. */ + +import { + existsSync, + mkdirSync, + readFileSync, + statSync, + writeFileSync, +} from "node:fs"; +import { homedir } from "node:os"; +import { dirname, join } from "node:path"; +import { fileURLToPath } from "node:url"; +import type { ChannelConfig, NotifyConfig, NtfyChannel } from "./types.ts"; + +function resolveExtensionDir(): string { + try { + return dirname(fileURLToPath(import.meta.url)); + } catch { + return join(homedir(), ".pi", "agent", "extensions", "pi-ntfy"); + } +} + +export const EXTENSION_DIR = resolveExtensionDir(); + +export function configPath(): string { + const override = + process.env.PI_NTFY_CONFIG?.trim() || + process.env.PI_AGENT_PUSH_CONFIG?.trim() || + process.env.PI_PUSH_CONFIG?.trim(); + return override ? override : join(EXTENSION_DIR, "config.json"); +} + +export function logPath(): string { + // Next to the config file, so PI_AGENT_PUSH_CONFIG keeps everything together. + return join(dirname(configPath()), "push.log"); +} + +export const KNOWN_TYPES = ["ntfy"] as const; +export type KnownChannelType = (typeof KNOWN_TYPES)[number]; +/** Whole-value reference: "$BARK_KEY" */ +const ENV_REF = /^\$([A-Za-z_][A-Za-z0-9_]*)$/; +/** Embedded reference: "Bearer ${MY_TOKEN}" */ +const ENV_INTERPOLATION = /\$\{([A-Za-z_][A-Za-z0-9_]*)\}/g; + +export type ConfigValue = + | string + | number + | boolean + | null + | ConfigValue[] + | { [key: string]: ConfigValue }; + +/** Resolve environment references in string values, recursively. */ +function resolveEnv(value: ConfigValue): ConfigValue { + if (typeof value === "string") { + const match = ENV_REF.exec(value); + if (match) return process.env[match[1]] ?? ""; + return value.replace( + ENV_INTERPOLATION, + (_full, name: string) => process.env[name] ?? "", + ); + } + if (Array.isArray(value)) return value.map(resolveEnv); + if (value && typeof value === "object") { + const out: { [key: string]: ConfigValue } = {}; + for (const [key, item] of Object.entries(value)) { + out[key] = resolveEnv(item); + } + return out; + } + return value; +} + +function bool(value: unknown, fallback: boolean): boolean { + return typeof value === "boolean" ? value : fallback; +} + +function num( + value: unknown, + fallback: number, + min: number, + max: number, +): number { + if (typeof value !== "number" || !Number.isFinite(value)) return fallback; + return Math.min(max, Math.max(min, value)); +} + +function str(value: unknown, fallback: string): string { + return typeof value === "string" && value.length > 0 ? value : fallback; +} + +function strArray(value: unknown, fallback: string[]): string[] { + if (!Array.isArray(value)) return fallback; + const items = value.filter( + (v): v is string => typeof v === "string" && v.length > 0, + ); + return items.length > 0 ? items : fallback; +} + +export function defaultConfig(path: string): NotifyConfig { + return { + enabled: true, + // Only the interactive TUI by default: pi subagents run `pi --mode json -p`, + // and pushing for every one of those would be pure noise. + modes: ["tui"], + timeoutMs: 5000, + shutdownTimeoutMs: 2000, + dedupeMs: 3000, + minDurationSec: 0, + maxTextChars: 2000, + titleTemplate: "[{{project}}] {{status}} · {{session}}", + template: + "项目: {{project}}\n会话: {{session}}\n任务: {{task}}\n状态: {{status}}\n原因: {{reason}}\n摘要: {{summary}}\n耗时: {{duration}}\n模型: {{model}}\n主机: {{host}}\n时间: {{date}} {{time}}\n工具: {{tools}}", + events: { idle: true, interrupted: true, needInput: true, exit: false }, + // Question-shaped tools block inside their own execute() waiting on the UI, + // so agent_settled never fires while they wait — they need their own event. + // ask_user_question: @juicesharp/rpiv-ask-user-question + // plan_mode_question: @narumitw/pi-plan-mode + needInputTools: ["ask_user_question", "plan_mode_question"], + debug: false, + channels: [], + warnings: [], + path, + exists: false, + }; +} + +function normalizeChannels(raw: unknown, warnings: string[]): ChannelConfig[] { + if (raw === undefined) return []; + if (!Array.isArray(raw)) { + warnings.push('"channels" 必须是数组，已忽略'); + return []; + } + const channels: ChannelConfig[] = []; + raw.forEach((item, index) => { + if (!item || typeof item !== "object") { + warnings.push(`channels[${index}] 不是对象，已忽略`); + return; + } + const channel = resolveEnv(item) as NtfyChannel; + if (channel.type !== "ntfy") { + warnings.push(`channels[${index}] 不是 ntfy 渠道，已忽略`); + return; + } + channels.push(channel); + }); + return channels; +} + +let cache: { key: string; config: NotifyConfig } | undefined; + +/** Load the config, reusing the previous parse while the file is unchanged. */ +export function loadConfig(): NotifyConfig { + const path = configPath(); + if (!existsSync(path)) { + cache = undefined; + return defaultConfig(path); + } + + let key: string; + try { + const stat = statSync(path); + key = `${path}:${stat.mtimeMs}:${stat.size}`; + } catch { + key = `${path}:unstatable`; + } + if (cache && cache.key === key) return cache.config; + + const config = defaultConfig(path); + config.exists = true; + + let parsed: Record; + try { + parsed = JSON.parse(readFileSync(path, "utf-8")) as Record; + } catch (error) { + config.enabled = false; + config.warnings.push( + `config.json 解析失败：${error instanceof Error ? error.message : String(error)}`, + ); + cache = { key, config }; + return config; + } + + config.enabled = bool(parsed.enabled, config.enabled); + config.modes = strArray(parsed.modes, config.modes); + config.timeoutMs = num(parsed.timeoutMs, config.timeoutMs, 200, 60_000); + config.shutdownTimeoutMs = num( + parsed.shutdownTimeoutMs, + config.shutdownTimeoutMs, + 200, + 10_000, + ); + config.dedupeMs = num(parsed.dedupeMs, config.dedupeMs, 0, 600_000); + config.minDurationSec = num( + parsed.minDurationSec, + config.minDurationSec, + 0, + 86_400, + ); + config.maxTextChars = num(parsed.maxTextChars, config.maxTextChars, 20, 4000); + config.titleTemplate = str(parsed.titleTemplate, config.titleTemplate); + config.template = str(parsed.template, config.template); + config.needInputTools = strArray(parsed.needInputTools, config.needInputTools); + config.debug = bool(parsed.debug, config.debug); + + if (parsed.events && typeof parsed.events === "object") { + const events = parsed.events as Record; + config.events = { + idle: bool(events.idle, config.events.idle), + interrupted: bool(events.interrupted, config.events.interrupted), + needInput: bool(events.needInput, config.events.needInput), + exit: bool(events.exit, config.events.exit), + }; + } + + config.channels = normalizeChannels(parsed.channels, config.warnings); + if (config.enabled && config.channels.length === 0) { + config.warnings.push("没有配置任何推送渠道"); + } else if ( + config.enabled && + config.channels.every((c) => c.enabled === false) + ) { + config.warnings.push("所有渠道都是 enabled:false，填好 key 后记得打开"); + } + + cache = { key, config }; + return config; +} + +/** Drop the mtime cache — used by `/push` so status output is always fresh. */ +export function invalidateConfigCache(): void { + cache = undefined; +} + +/** Sensitive keys — masked in `/push get` output. */ +export const SECRET_KEYS = new Set(["token"]); + +/** Per-type fields users can set via `/push set`. */ +export const CHANNEL_SETTABLE_FIELDS: Record = { + ntfy: [ + "enabled", + "name", + "topic", + "server", + "token", + "priority", + "tags", + "clickUrl", + "icon", + "email", + "timeoutMs", + ], +}; + +export interface RawConfigFile { + enabled?: boolean; + modes?: string[]; + timeoutMs?: number; + shutdownTimeoutMs?: number; + dedupeMs?: number; + minDurationSec?: number; + maxTextChars?: number; + debug?: boolean; + titleTemplate?: string; + template?: string; + events?: Record; + needInputTools?: string[]; + channels?: Array>; + [key: string]: unknown; +} + +/** Read config.json without env expansion — for in-place edits. */ +export function loadRawConfig(): { + path: string; + exists: boolean; + data: RawConfigFile; +} { + const path = configPath(); + if (!existsSync(path)) { + return { + path, + exists: false, + data: { + enabled: true, + modes: ["tui"], + events: { idle: true, interrupted: true, needInput: true, exit: false }, + channels: [], + }, + }; + } + try { + const data = JSON.parse(readFileSync(path, "utf-8")) as RawConfigFile; + if (!Array.isArray(data.channels)) data.channels = []; + return { path, exists: true, data }; + } catch (error) { + throw new Error( + `config.json 解析失败：${error instanceof Error ? error.message : String(error)}`, + ); + } +} + +/** Atomically-ish write config.json and drop the mtime cache. */ +export function saveRawConfig(data: RawConfigFile): string { + const path = configPath(); + const dir = dirname(path); + if (!existsSync(dir)) mkdirSync(dir, { recursive: true }); + const body = `${JSON.stringify(data, null, "\t")}\n`; + writeFileSync(path, body, "utf-8"); + invalidateConfigCache(); + return path; +} + +export function maskSecret(key: string, value: unknown): string { + if (value == null) return ""; + const text = String(value); + if ( + !SECRET_KEYS.has(key) && + !/token|secret|key|password|authorization/i.test(key) + ) { + return text; + } + if (text.length === 0) return "(空)"; + if (text.startsWith("$")) return text; // env ref — safe to show + if (text.length <= 4) return "****"; + return `${"*".repeat(Math.min(8, text.length - 4))}${text.slice(-4)}`; +} + +/** + * Resolve a channel selector to an index in raw.channels. + * Accepts: type (`ntfy`), name, `type#N` (1-based among that type), or 1-based index. + */ +export function resolveChannelIndex( + channels: Array>, + selector: string, +): number { + const sel = selector.trim(); + if (!sel) throw new Error("请指定渠道（type / name / type#N / 序号）"); + + // pure number → 1-based index + if (/^\d+$/.test(sel)) { + const idx = Number(sel) - 1; + if (idx < 0 || idx >= channels.length) { + throw new Error(`渠道序号 ${sel} 超出范围 1..${channels.length || 0}`); + } + return idx; + } + + // type#N + const hash = /^(?[A-Za-z][A-Za-z0-9_-]*)#(?\d+)$/.exec(sel); + if (hash?.groups) { + const type = hash.groups.type.toLowerCase(); + const n = Number(hash.groups.n); + const matches = channels + .map((c, i) => ({ c, i })) + .filter(({ c }) => String(c.type ?? "").toLowerCase() === type); + if (matches.length === 0) throw new Error(`没有 type=${type} 的渠道`); + if (n < 1 || n > matches.length) { + throw new Error(`${type} 只有 ${matches.length} 个，无法选 #${n}`); + } + return matches[n - 1].i; + } + + const lower = sel.toLowerCase(); + + // exact name (case-insensitive) + const byName = channels.findIndex( + (c) => typeof c.name === "string" && c.name.toLowerCase() === lower, + ); + if (byName >= 0) return byName; + + // unique type + const byType = channels + .map((c, i) => ({ c, i })) + .filter(({ c }) => String(c.type ?? "").toLowerCase() === lower); + if (byType.length === 1) return byType[0].i; + if (byType.length > 1) { + throw new Error( + `type=${sel} 有 ${byType.length} 个，请用 ${sel}#1 或名称区分`, + ); + } + + throw new Error(`找不到渠道 "${sel}"（可用 /push list 查看）`); +} + +export function formatChannelLabel( + channel: Record, + index: number, +): string { + const type = String(channel.type ?? "?"); + const name = + typeof channel.name === "string" && channel.name.trim() + ? channel.name.trim() + : ""; + const enabled = channel.enabled === false ? "关" : "开"; + const base = name ? `${name} (${type}#${index + 1})` : `${type}#${index + 1}`; + return `${base} [${enabled}]`; +} diff --git a/http.ts b/http.ts new file mode 100644 index 0000000..bc52e02 --- /dev/null +++ b/http.ts @@ -0,0 +1,105 @@ +/** + * pi-agent-push — HTTP transport. + * + * Every request carries its own `AbortSignal.timeout`. It must never use the + * agent's `ctx.signal`: on the "interrupted" path that signal is already + * aborted, so the notification about the cancellation would cancel itself. + */ + +import type { ChannelResult } from "./types.ts"; + +export interface HttpResult { + ok: boolean; + status: number; + body: string; + error?: string; + ms: number; +} + +export interface HttpRequestInit { + method?: string; + headers?: Record; + body?: string; +} + +const MAX_BODY_ECHO = 300; + +export async function httpRequest( + url: string, + init: HttpRequestInit, + timeoutMs: number, +): Promise { + const startedAt = Date.now(); + try { + const response = await fetch(url, { + method: init.method ?? "POST", + headers: init.headers, + body: init.body, + // Deliberately NOT ctx.signal — see file header. + signal: AbortSignal.timeout(timeoutMs), + }); + const text = await response.text(); + return { + ok: response.ok, + status: response.status, + body: text.slice(0, MAX_BODY_ECHO), + error: response.ok ? undefined : `HTTP ${response.status}`, + ms: Date.now() - startedAt, + }; + } catch (error) { + const name = error instanceof Error ? error.name : ""; + const message = + name === "TimeoutError" || name === "AbortError" + ? `timeout after ${timeoutMs}ms` + : error instanceof Error + ? error.message + : String(error); + return { + ok: false, + status: 0, + body: "", + error: message, + ms: Date.now() - startedAt, + }; + } +} + +/** A local failure (bad config) shaped like an HttpResult. */ +export function localFailure(message: string): HttpResult { + return { ok: false, status: 0, body: "", error: message, ms: 0 }; +} + +export function toChannelResult( + name: string, + result: HttpResult, +): ChannelResult { + return { + channel: name, + ok: result.ok, + status: result.status, + error: result.error, + ms: result.ms, + }; +} + +/** + * Group-bot endpoints answer HTTP 200 even when they reject the message + * (missing keyword, bad signature, rate limit). Fold that into `ok`. + */ +export function applyApiStatus( + result: HttpResult, + pick: ( + payload: Record, + ) => { ok: boolean; message?: string } | undefined, +): HttpResult { + if (!result.ok || !result.body) return result; + let payload: Record; + try { + payload = JSON.parse(result.body) as Record; + } catch { + return result; + } + const verdict = pick(payload); + if (!verdict || verdict.ok) return result; + return { ...result, ok: false, error: verdict.message ?? "API 返回失败" }; +} diff --git a/index.ts b/index.ts new file mode 100644 index 0000000..f871292 --- /dev/null +++ b/index.ts @@ -0,0 +1,311 @@ +/** 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; + // Subagents run `pi --mode json -p`; pushing for each of those is noise. + if (!config.modes.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"); + }, + }); +} diff --git a/notifier.ts b/notifier.ts new file mode 100644 index 0000000..0b6a050 --- /dev/null +++ b/notifier.ts @@ -0,0 +1,275 @@ +/** pi-ntfy payload building and dispatch. */ + +import { appendFileSync, renameSync, statSync } from "node:fs"; +import { hostname } from "node:os"; +import { basename } from "node:path"; +import { channelName, sendToChannel } from "./channels/index.ts"; +import { logPath } from "./config.ts"; +import { toChannelResult } from "./http.ts"; +import { formatDuration, render } from "./render.ts"; +import type { + ChannelConfig, + ChannelResult, + NotifyConfig, + NotifyEvent, + NotifyPayload, +} from "./types.ts"; + +const LOG_ROTATE_BYTES = 256 * 1024; + +export interface PayloadInput { + event: NotifyEvent; + status: string; + reason: string; + cwd?: string; + durationMs?: number; + session?: string; + sessionId?: string; + model?: string; + mode?: string; + thinking?: string; + task?: string; + summary?: string; + tools?: string; +} + +export function buildPayload( + config: NotifyConfig, + input: PayloadInput, +): NotifyPayload { + const now = new Date(); + const cwd = input.cwd ?? process.cwd(); + const vars: Record = { + event: input.event, + status: input.status, + reason: input.reason, + cwd, + project: basename(cwd) || cwd, + duration: + input.durationMs === undefined ? "" : formatDuration(input.durationMs), + session: input.session ?? "", + sessionId: input.sessionId ?? "", + model: input.model ?? "", + mode: input.mode ?? "", + thinking: input.thinking ?? "", + task: input.task ?? "", + summary: input.summary ?? "", + tools: input.tools ?? "", + host: hostname(), + time: now.toTimeString().slice(0, 8), + date: now.toISOString().slice(0, 10), + }; + + const title = render(config.titleTemplate, vars).trim() || "pi"; + let text = render(config.template, vars).trim() || title; + if (text.length > config.maxTextChars) + text = `${text.slice(0, config.maxTextChars - 1)}…`; + + // NTFY accepts plain text bodies; metadata is carried in headers. + vars.title = title; + vars.text = text; + + return { + event: input.event, + status: input.status, + reason: input.reason, + title, + text, + vars, + }; +} + +interface Target { + channel: ChannelConfig; + name: string; +} + +export function resolveTargets( + config: NotifyConfig, + event: NotifyEvent, +): Target[] { + const targets: Target[] = []; + config.channels.forEach((channel, index) => { + if (channel.enabled === false) return; + if (event !== "test") { + const override = channel.events?.[event]; + const allowed = override ?? config.events[event] ?? false; + if (!allowed) return; + } + targets.push({ channel, name: channelName(channel, index) }); + }); + return targets; +} + +function writeLog(config: NotifyConfig, line: string): void { + const path = logPath(); + try { + const stat = statSync(path, { throwIfNoEntry: false }); + if (stat && stat.size > LOG_ROTATE_BYTES) renameSync(path, `${path}.1`); + } catch { + // rotation is best-effort + } + try { + appendFileSync(path, `${new Date().toISOString()} ${line}\n`, "utf-8"); + } catch { + // never let logging break a notification + } + void config; +} + +function capPromise( + promise: Promise, + ms: number, + fallback: T, +): Promise { + return Promise.race([ + promise, + new Promise((resolve) => { + const timer = setTimeout(() => resolve(fallback), ms); + // Do not hold the event loop open while pi is exiting. + (timer as { unref?: () => void }).unref?.(); + }), + ]); +} + +export interface Notifier { + /** Non-blocking send; failures are logged, never thrown. */ + fire(config: NotifyConfig, input: PayloadInput): void; + /** Awaited send with a hard cap — used on the shutdown path. */ + fireBlocking( + config: NotifyConfig, + input: PayloadInput, + capMs: number, + ): Promise; + /** Ignores event switches and pushes to every enabled channel. */ + sendTest(config: NotifyConfig, input: PayloadInput): Promise; + /** Wait for in-flight sends (bounded). */ + flush(capMs: number): Promise; + lastResults(): + | { at: number; event: NotifyEvent; results: ChannelResult[] } + | undefined; +} + +export function createNotifier(): Notifier { + const inflight = new Set>(); + let lastKey = ""; + let lastAt = 0; + let last: + | { at: number; event: NotifyEvent; results: ChannelResult[] } + | undefined; + + async function dispatch( + config: NotifyConfig, + payload: NotifyPayload, + targets: Target[], + ): Promise { + const results = await Promise.all( + targets.map(async ({ channel, name }) => { + try { + const result = await sendToChannel( + channel, + payload, + channel.timeoutMs ?? config.timeoutMs, + ); + return toChannelResult(name, result); + } catch (error) { + return { + channel: name, + ok: false, + status: 0, + error: error instanceof Error ? error.message : String(error), + ms: 0, + } satisfies ChannelResult; + } + }), + ); + + last = { at: Date.now(), event: payload.event, results }; + for (const result of results) { + if (result.ok && !config.debug) continue; + const verdict = result.ok ? "ok" : "FAIL"; + const detail = result.ok + ? `status=${result.status}` + : (result.error ?? "unknown"); + writeLog( + config, + `[${payload.event}] ${result.channel} ${verdict} ${detail} (${result.ms}ms)`, + ); + } + return results; + } + + function prepare( + config: NotifyConfig, + input: PayloadInput, + ignoreSwitches: boolean, + ): { payload: NotifyPayload; targets: Target[] } | undefined { + if (!config.enabled) return undefined; + const targets = resolveTargets(config, ignoreSwitches ? "test" : input.event); + if (targets.length === 0) return undefined; + + const payload = buildPayload(config, input); + if (!ignoreSwitches && config.dedupeMs > 0) { + const key = `${payload.event}|${payload.text}`; + const now = Date.now(); + if (key === lastKey && now - lastAt < config.dedupeMs) return undefined; + lastKey = key; + lastAt = now; + } + return { payload, targets }; + } + + function track(promise: Promise): Promise { + const tracked = promise.finally(() => { + inflight.delete(tracked); + }); + inflight.add(tracked); + return tracked; + } + + return { + fire(config, input) { + const prepared = prepare(config, input, false); + if (!prepared) return; + void track( + dispatch(config, prepared.payload, prepared.targets).catch( + () => [] as ChannelResult[], + ), + ); + }, + + async fireBlocking(config, input, capMs) { + const prepared = prepare(config, input, false); + if (!prepared) return []; + const run = track( + dispatch(config, prepared.payload, prepared.targets).catch( + () => [] as ChannelResult[], + ), + ); + return capPromise(run, capMs, []); + }, + + async sendTest(config, input) { + const prepared = prepare(config, input, true); + if (!prepared) return []; + return track( + dispatch(config, prepared.payload, prepared.targets).catch( + () => [] as ChannelResult[], + ), + ); + }, + + async flush(capMs) { + if (inflight.size === 0) return; + await capPromise( + Promise.allSettled([...inflight]).then(() => undefined), + capMs, + undefined, + ); + }, + + lastResults() { + return last; + }, + }; +} diff --git a/package.json b/package.json new file mode 100644 index 0000000..117d873 --- /dev/null +++ b/package.json @@ -0,0 +1,54 @@ +{ + "name": "pi-ntfy", + "version": "1.0.0", + "description": "Project and session-aware NTFY notifications for the pi coding agent.", + "type": "module", + "license": "MIT", + "author": "local", + "repository": { + "type": "git", + "url": "file:///home/yanghao05/Workspace/pi-push" + }, + "bugs": {}, + "homepage": "file:///home/yanghao05/Workspace/pi-push", + "keywords": [ + "pi", + "pi-coding-agent", + "pi-package", + "pi-extension", + "ntfy", + "notify" + ], + "files": [ + "channels", + "index.ts", + "command.ts", + "config.ts", + "http.ts", + "notifier.ts", + "render.ts", + "types.ts", + "config.example.json", + "README.md", + "LICENSE", + "DESIGN.md" + ], + "scripts": { + "test": "node test/run.ts", + "prepublishOnly": "node test/run.ts" + }, + "peerDependencies": { + "@earendil-works/pi-coding-agent": "*" + }, + "engines": { + "node": ">=22" + }, + "publishConfig": { + "access": "public" + }, + "pi": { + "extensions": [ + "./index.ts" + ] + } +} diff --git a/render.ts b/render.ts new file mode 100644 index 0000000..51fd89f --- /dev/null +++ b/render.ts @@ -0,0 +1,118 @@ +/** + * pi-agent-push — `{{placeholder}}` substitution. + * + * Escaping matters: a raw string template that lands inside JSON must have its + * values JSON-escaped, otherwise a quote or newline in an error message + * produces an invalid request body. Object templates avoid the problem + * entirely (substitute first, stringify after) and are the recommended form. + */ + +const PLACEHOLDER = /\{\{\s*([a-zA-Z0-9_]+)\s*\}\}/g; + +export type EscapeMode = "none" | "json" | "url"; + +function escapeValue(value: string, mode: EscapeMode): string { + if (mode === "json") { + const encoded = JSON.stringify(value); + return encoded.slice(1, -1); + } + if (mode === "url") return encodeURIComponent(value); + return value; +} + +/** Replace `{{name}}` in `template`. Unknown names collapse to "". */ +export function render( + template: string, + vars: Record, + mode: EscapeMode = "none", +): string { + return template.replace(PLACEHOLDER, (_match, name: string) => + escapeValue(vars[name] ?? "", mode), + ); +} + +export type RenderValue = + | string + | number + | boolean + | null + | RenderValue[] + | { [key: string]: RenderValue }; + +/** Recursively render every string inside an object/array template. */ +export function renderDeep( + value: RenderValue, + vars: Record, +): RenderValue { + if (typeof value === "string") return render(value, vars); + if (Array.isArray(value)) return value.map((item) => renderDeep(item, vars)); + if (value && typeof value === "object") { + const out: { [key: string]: RenderValue } = {}; + for (const [key, item] of Object.entries(value)) { + out[render(key, vars)] = renderDeep(item, vars); + } + return out; + } + return value; +} + +/** Format a millisecond duration as `1h2m`, `3m12s` or `12s`. */ +export function formatDuration(ms: number): string { + const totalSeconds = Math.max(0, Math.round(ms / 1000)); + if (totalSeconds < 60) return `${totalSeconds}s`; + const minutes = Math.floor(totalSeconds / 60); + const seconds = totalSeconds % 60; + if (minutes < 60) + return seconds === 0 ? `${minutes}m` : `${minutes}m${seconds}s`; + const hours = Math.floor(minutes / 60); + const restMinutes = minutes % 60; + return restMinutes === 0 ? `${hours}h` : `${hours}h${restMinutes}m`; +} + +/** First non-empty line, trimmed and capped — used for error messages. */ +export function firstLine(text: string | undefined, max: number): string { + if (!text) return ""; + const line = + text + .split("\n") + .map((l) => l.trim()) + .find((l) => l.length > 0) ?? ""; + return line.length > max ? `${line.slice(0, max - 1)}…` : line; +} + +function pickString(source: unknown, key: string): string { + if (!source || typeof source !== "object") return ""; + const value = (source as Record)[key]; + return typeof value === "string" ? value : ""; +} + +/** + * Turn a provider error into something worth reading on a phone. + * + * pi surfaces raw upstream bodies, e.g. + * `401 {"type":"error","error":{"type":"authentication_error","message":"invalid x-api-key"}}` + * which becomes `401 authentication_error: invalid x-api-key`. + */ +export function summarizeError(raw: string | undefined, max: number): string { + if (!raw) return ""; + const text = raw.trim(); + const braceIndex = text.indexOf("{"); + if (braceIndex >= 0) { + try { + const parsed = JSON.parse(text.slice(braceIndex)) as Record; + const nested = (parsed.error ?? parsed) as Record; + const message = + pickString(nested, "message") || pickString(parsed, "message"); + const type = pickString(nested, "type") || pickString(nested, "code"); + const detail = [type, message].filter((part) => part.length > 0).join(": "); + if (detail) { + const prefix = text.slice(0, braceIndex).trim(); + const combined = prefix ? `${prefix} ${detail}` : detail; + return combined.length > max ? `${combined.slice(0, max - 1)}…` : combined; + } + } catch { + // not JSON after all — fall through to the plain first line + } + } + return firstLine(text, max); +} diff --git a/types.ts b/types.ts new file mode 100644 index 0000000..af54215 --- /dev/null +++ b/types.ts @@ -0,0 +1,69 @@ +/** pi-ntfy shared types. */ + +export type NotifyEvent = + | "idle" + | "interrupted" + | "needInput" + | "exit" + | "test"; + +export interface EventToggles { + idle?: boolean; + interrupted?: boolean; + needInput?: boolean; + exit?: boolean; +} + +export interface NtfyChannel { + type: "ntfy"; + name?: string; + enabled?: boolean; + events?: EventToggles; + timeoutMs?: number; + topic: string; + server?: string; + token?: string; + priority?: string | number; + tags?: string | string[]; + clickUrl?: string; + icon?: string; + email?: string; +} + +export type ChannelConfig = NtfyChannel; + +export interface NotifyConfig { + enabled: boolean; + modes: string[]; + timeoutMs: number; + shutdownTimeoutMs: number; + dedupeMs: number; + minDurationSec: number; + maxTextChars: number; + titleTemplate: string; + template: string; + events: Required; + needInputTools: string[]; + debug: boolean; + channels: NtfyChannel[]; + warnings: string[]; + path: string; + exists: boolean; +} + +export interface NotifyPayload { + event: NotifyEvent; + status: string; + reason: string; + title: string; + text: string; + vars: Record; +} + +export interface ChannelResult { + channel: string; + ok: boolean; + status: number; + error?: string; + ms: number; +}