commit 7dc17e1a9f901c025121fb2e9c8e317e5436f409 Author: 杨豪 Date: Thu Sep 17 09:50:46 2026 +0800 feat: add pi-ntfy NTFY notification plugin 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; +}