diff options
| author | Adam Malczewski <[email protected]> | 2026-06-01 09:28:21 +0900 |
|---|---|---|
| committer | Adam Malczewski <[email protected]> | 2026-06-01 09:28:21 +0900 |
| commit | 5e72191cac9469c2ade91aaba1e62f69fa1ad94a (patch) | |
| tree | bfa21aa2841277b8861bd05a0e2472211bed0aa9 /packages/core/src | |
| parent | 97c1b40ead19cdfe54b9a7aeb2c0fdcc1c9653b1 (diff) | |
| download | dispatch-5e72191cac9469c2ade91aaba1e62f69fa1ad94a.tar.gz dispatch-5e72191cac9469c2ade91aaba1e62f69fa1ad94a.zip | |
feat(core): ntfy.sh notification dispatcher module
Adds a transport-agnostic NotificationDispatcher and a fire-and-forget
ntfy.sh transport (no SDK; just fetch). Configuration is persisted as a
single global JSON blob under the 'ntfy_config' settings key.
Event taxonomy (per-event toggles):
- turn-completed — assistant turn finished cleanly
- turn-error — final turn error (after all fallbacks)
- permission-required — new permission prompt was created
- agent-spawned — top-level user-agent tab spawned via 'summon'
Design:
- Single internal notify(event) interface so a future transport (email,
webhook) plugs in without changing call sites.
- attachToAgentManager + attachToPermissionManager subscribe to the
existing event streams via narrow listener interfaces (no @dispatch/api
dependency back into core).
- 5s in-memory dedupe window on dedupeKey suppresses permission re-emits.
- 10s per-request abort timeout so a hung ntfy server can't pin a worker.
- All sends are fire-and-forget: void Promise.resolve(...).catch(warn).
Tests (39 new):
- ntfy transport: URL/headers/body/auth/click, header sanitization,
per-event-type defaults, error paths.
- config: defaults, normalization tolerance, round-trip, redaction.
- dispatcher: master switch, per-event toggle, dedupe, agent/permission
hookups, top-level-only filtering for agent-spawned, dispose.
Diffstat (limited to 'packages/core/src')
| -rw-r--r-- | packages/core/src/index.ts | 2 | ||||
| -rw-r--r-- | packages/core/src/notifications/config.ts | 74 | ||||
| -rw-r--r-- | packages/core/src/notifications/dispatcher.ts | 238 | ||||
| -rw-r--r-- | packages/core/src/notifications/index.ts | 29 | ||||
| -rw-r--r-- | packages/core/src/notifications/ntfy.ts | 136 | ||||
| -rw-r--r-- | packages/core/src/notifications/types.ts | 98 |
6 files changed, 577 insertions, 0 deletions
diff --git a/packages/core/src/index.ts b/packages/core/src/index.ts index b1b17cc..327b0a5 100644 --- a/packages/core/src/index.ts +++ b/packages/core/src/index.ts @@ -68,6 +68,8 @@ export { export { createProvider } from "./llm/provider.js"; // Models export { ModelRegistry } from "./models/index.js"; +// Notifications (ntfy.sh) +export * from "./notifications/index.js"; export * from "./permission/index.js"; // Skills export { diff --git a/packages/core/src/notifications/config.ts b/packages/core/src/notifications/config.ts new file mode 100644 index 0000000..310c606 --- /dev/null +++ b/packages/core/src/notifications/config.ts @@ -0,0 +1,74 @@ +// Persisted ntfy config — single global JSON blob under one settings key. +// +// One global config (no per-user split): the rest of Dispatch's settings +// table is global today (cf. `title_model_*`, `perm_*`), so notification +// config follows the same pattern. + +import { deleteSetting, getSetting, setSetting } from "../db/settings.js"; +import type { NotificationEventType, NtfyConfig } from "./types.js"; +import { NTFY_DEFAULT_EVENTS, NTFY_EVENT_TYPES } from "./types.js"; + +export const NTFY_CONFIG_KEY = "ntfy_config"; + +/** Defaults returned when nothing is persisted yet. */ +export function defaultNtfyConfig(): NtfyConfig { + return { + enabled: false, + topicUrl: "", + authToken: "", + events: { ...NTFY_DEFAULT_EVENTS }, + }; +} + +/** + * Normalize an arbitrary parsed JSON value into a complete `NtfyConfig`. + * Tolerant of missing / unexpected fields so a config from an older build + * never throws — missing event toggles fall back to defaults. + */ +export function normalizeNtfyConfig(raw: unknown): NtfyConfig { + const base = defaultNtfyConfig(); + if (!raw || typeof raw !== "object") return base; + const obj = raw as Record<string, unknown>; + const out: NtfyConfig = { + enabled: typeof obj.enabled === "boolean" ? obj.enabled : base.enabled, + topicUrl: typeof obj.topicUrl === "string" ? obj.topicUrl : base.topicUrl, + authToken: typeof obj.authToken === "string" ? obj.authToken : base.authToken, + events: { ...base.events }, + }; + const rawEvents = obj.events; + if (rawEvents && typeof rawEvents === "object") { + const evObj = rawEvents as Record<string, unknown>; + for (const key of NTFY_EVENT_TYPES) { + const v = evObj[key]; + if (typeof v === "boolean") out.events[key as NotificationEventType] = v; + } + } + return out; +} + +/** Load the persisted config (or defaults if none/corrupt). */ +export function loadNtfyConfig(): NtfyConfig { + const raw = getSetting(NTFY_CONFIG_KEY); + if (!raw) return defaultNtfyConfig(); + try { + return normalizeNtfyConfig(JSON.parse(raw)); + } catch { + return defaultNtfyConfig(); + } +} + +/** Persist a complete config (after server-side normalization). */ +export function saveNtfyConfig(config: NtfyConfig): void { + const normalized = normalizeNtfyConfig(config); + setSetting(NTFY_CONFIG_KEY, JSON.stringify(normalized)); +} + +/** Wipe the persisted config (revert to defaults on next load). */ +export function clearNtfyConfig(): void { + deleteSetting(NTFY_CONFIG_KEY); +} + +/** Strip the auth token from a config before returning it over the API. */ +export function redactNtfyConfig(config: NtfyConfig): NtfyConfig & { hasAuthToken: boolean } { + return { ...config, authToken: "", hasAuthToken: config.authToken.trim().length > 0 }; +} diff --git a/packages/core/src/notifications/dispatcher.ts b/packages/core/src/notifications/dispatcher.ts new file mode 100644 index 0000000..4f4fc79 --- /dev/null +++ b/packages/core/src/notifications/dispatcher.ts @@ -0,0 +1,238 @@ +// NotificationDispatcher — turns high-level Dispatch events into +// `sendNtfy(...)` calls, gated by the persisted user config. +// +// The dispatcher is transport-agnostic at the `notify(event)` interface +// boundary: only `sendNtfy` is wired today, but adding another transport +// (email, webhook, etc.) means changing this one file, not the call sites. +// +// All sends are non-blocking (fire-and-forget). A 10-second timeout in +// `sendNtfy` bounds the worst case; the dispatcher additionally guards +// every send in a try/catch so a transport bug can never propagate into +// the agent loop. + +import { loadNtfyConfig } from "./config.js"; +import { type FetchLike, sendNtfy } from "./ntfy.js"; +import type { NotificationEvent, NtfyConfig } from "./types.js"; + +/** Minimal shape of an `AgentManager`-style event stream we hook into. */ +export interface AgentEventSource { + onEvent( + listener: (event: { type: string; tabId: string; [key: string]: unknown }) => void, + ): () => void; +} + +/** Minimal shape of a `PermissionManager`-style prompt source. */ +export interface PermissionPromptSource { + onPromptAdded( + listener: (prompt: { id: string; permission: string; description: string }) => void, + ): () => void; +} + +/** Look up a human-readable tab title for nicer notification text. */ +export type TabTitleLookup = (tabId: string) => string | null; + +export interface DispatcherOptions { + /** Override the config loader (tests). Defaults to `loadNtfyConfig`. */ + loadConfig?: () => NtfyConfig; + /** Override the transport (tests). Defaults to the real `sendNtfy`. */ + send?: (config: NtfyConfig, event: NotificationEvent) => Promise<unknown>; + /** Optional fetch override (forwarded to `sendNtfy` when `send` not set). */ + fetchImpl?: FetchLike; + /** Look up a tab title for richer titles. */ + getTabTitle?: TabTitleLookup; + /** + * How long (ms) a dedupeKey is suppressed for. Permission prompts re-emit + * the whole pending list on every change, so dedupe is essential. + */ + dedupeWindowMs?: number; +} + +export class NotificationDispatcher { + private loadConfig: () => NtfyConfig; + private send: (config: NtfyConfig, event: NotificationEvent) => Promise<unknown>; + private getTabTitle: TabTitleLookup | undefined; + private dedupeWindowMs: number; + /** Recently-sent dedupeKey → expiresAt epoch ms. */ + private recentlySent = new Map<string, number>(); + private unsubs: Array<() => void> = []; + + constructor(opts: DispatcherOptions = {}) { + this.loadConfig = opts.loadConfig ?? loadNtfyConfig; + this.send = + opts.send ?? ((config, event) => sendNtfy(config, event, opts.fetchImpl ?? undefined)); + this.getTabTitle = opts.getTabTitle; + this.dedupeWindowMs = opts.dedupeWindowMs ?? 5_000; + } + + /** + * Single internal entry point — every public hook funnels through here. + * Public so a future caller can synthesize an arbitrary notification + * (e.g. a CLI `dispatch notify` command); kept narrow. + */ + notify(event: NotificationEvent): void { + const config = this.loadConfig(); + if (!config.enabled) return; + if (!config.events[event.type]) return; + if (event.dedupeKey && this.isDuplicate(event.dedupeKey)) return; + if (event.dedupeKey) this.markSent(event.dedupeKey); + + // Fire-and-forget: never await, never throw. + try { + void Promise.resolve(this.send(config, event)).catch((err) => { + console.warn( + `[ntfy] send failed for ${event.type}: ${err instanceof Error ? err.message : String(err)}`, + ); + }); + } catch (err) { + // Guard the synchronous portion of `send` too. + console.warn( + `[ntfy] dispatch threw for ${event.type}: ${err instanceof Error ? err.message : String(err)}`, + ); + } + } + + /** + * Hook into an `AgentManager`-style event stream. + * + * Maps: + * - `done` → `turn-completed` + * - `error` → `turn-error` + * - `tab-created` → `agent-spawned` (only top-level user-agent tabs) + * + * `status` events are ignored — they fire on every transition and we'd + * either spam or duplicate the `done`/`error` notifications. + */ + attachToAgentManager(source: AgentEventSource): () => void { + const unsub = source.onEvent((event) => { + if (event.type === "done") { + this.notify(this.buildTurnCompleted(event)); + } else if (event.type === "error") { + this.notify(this.buildTurnError(event)); + } else if (event.type === "tab-created") { + const ev = event as unknown as { + tabId: string; + id: string; + title: string; + parentTabId: string | null; + agentSlug?: string | null; + }; + // Only notify for top-level user-agent tabs spawned via `summon`. + // Filtering on `agentSlug` skips "blank" new tabs the user opened + // manually, which would be noisy. + if (ev.parentTabId === null && ev.agentSlug) { + this.notify(this.buildAgentSpawned(ev)); + } + } + }); + this.unsubs.push(unsub); + return unsub; + } + + /** Hook into a `PermissionManager`-style prompt source. */ + attachToPermissionManager(source: PermissionPromptSource): () => void { + const unsub = source.onPromptAdded((prompt) => { + this.notify(this.buildPermissionRequired(prompt)); + }); + this.unsubs.push(unsub); + return unsub; + } + + /** Release all hooks acquired via `attachTo*`. */ + dispose(): void { + for (const u of this.unsubs) { + try { + u(); + } catch { + // best-effort + } + } + this.unsubs = []; + this.recentlySent.clear(); + } + + // ─── Event builders (internal) ──────────────────────────────── + + private buildTurnCompleted(event: { tabId: string }): NotificationEvent { + const tabLabel = this.tabLabel(event.tabId); + return { + type: "turn-completed", + title: `Turn complete — ${tabLabel}`, + message: `Assistant finished a turn in ${tabLabel}.`, + tabId: event.tabId, + }; + } + + private buildTurnError(event: { + tabId: string; + error?: unknown; + statusCode?: unknown; + }): NotificationEvent { + const tabLabel = this.tabLabel(event.tabId); + const errText = typeof event.error === "string" ? event.error : "Unknown error"; + const statusText = typeof event.statusCode === "number" ? ` (status ${event.statusCode})` : ""; + return { + type: "turn-error", + title: `Turn failed — ${tabLabel}`, + message: `${errText}${statusText}`, + tabId: event.tabId, + }; + } + + private buildPermissionRequired(prompt: { + id: string; + permission: string; + description: string; + }): NotificationEvent { + return { + type: "permission-required", + title: `Permission required: ${prompt.permission}`, + message: prompt.description || `Agent is requesting ${prompt.permission} permission.`, + // Permission prompts can re-emit (e.g. another prompt arrives while + // this one is still pending) — dedupe on the prompt id. + dedupeKey: `permission:${prompt.id}`, + }; + } + + private buildAgentSpawned(ev: { + tabId: string; + id: string; + title: string; + agentSlug?: string | null; + }): NotificationEvent { + return { + type: "agent-spawned", + title: `User agent spawned — ${ev.agentSlug ?? "agent"}`, + message: ev.title, + tabId: ev.tabId ?? ev.id, + }; + } + + private tabLabel(tabId: string): string { + const title = this.getTabTitle?.(tabId); + if (title?.trim()) return title.trim(); + return `tab ${tabId.slice(0, 8)}`; + } + + // ─── Dedupe helpers ─────────────────────────────────────────── + + private isDuplicate(key: string): boolean { + const expires = this.recentlySent.get(key); + if (expires === undefined) return false; + if (expires <= Date.now()) { + this.recentlySent.delete(key); + return false; + } + return true; + } + + private markSent(key: string): void { + // Lazy-evict expired entries when the map gets large. + if (this.recentlySent.size > 256) { + const now = Date.now(); + for (const [k, exp] of this.recentlySent) { + if (exp <= now) this.recentlySent.delete(k); + } + } + this.recentlySent.set(key, Date.now() + this.dedupeWindowMs); + } +} diff --git a/packages/core/src/notifications/index.ts b/packages/core/src/notifications/index.ts new file mode 100644 index 0000000..ea99a58 --- /dev/null +++ b/packages/core/src/notifications/index.ts @@ -0,0 +1,29 @@ +// @dispatch/core — ntfy.sh push notifications + +export { + clearNtfyConfig, + defaultNtfyConfig, + loadNtfyConfig, + NTFY_CONFIG_KEY, + normalizeNtfyConfig, + redactNtfyConfig, + saveNtfyConfig, +} from "./config.js"; +export { + type AgentEventSource, + type DispatcherOptions, + NotificationDispatcher, + type PermissionPromptSource, + type TabTitleLookup, +} from "./dispatcher.js"; +export { type FetchLike, type NtfySendResult, sendNtfy, validateTopicUrl } from "./ntfy.js"; +export { + type NotificationEvent, + type NotificationEventType, + NTFY_DEFAULT_EVENTS, + NTFY_DEFAULT_PRIORITIES, + NTFY_DEFAULT_TAGS, + NTFY_EVENT_TYPES, + type NtfyConfig, + type NtfyPriority, +} from "./types.js"; diff --git a/packages/core/src/notifications/ntfy.ts b/packages/core/src/notifications/ntfy.ts new file mode 100644 index 0000000..07ce33b --- /dev/null +++ b/packages/core/src/notifications/ntfy.ts @@ -0,0 +1,136 @@ +// ntfy.sh HTTP transport. +// +// ntfy's API is a simple POST to `https://<server>/<topic>` with the body +// as the message and metadata passed via HTTP headers: +// Title: notification title +// Priority: 1..5 (3 = default) +// Tags: comma-separated emoji shortcodes +// Click: URL opened when the notification is tapped +// +// We intentionally use `fetch` directly — no SDK, no extra deps. + +import type { NotificationEvent, NtfyConfig } from "./types.js"; +import { NTFY_DEFAULT_PRIORITIES, NTFY_DEFAULT_TAGS } from "./types.js"; + +export interface NtfySendResult { + ok: boolean; + status?: number; + error?: string; +} + +/** + * Lightweight fetch shape so callers (and tests) can inject a mock without + * pulling in the DOM `fetch` type from a `Headers` instance. + */ +export type FetchLike = ( + input: string, + init: { method: string; headers: Record<string, string>; body: string; signal?: AbortSignal }, +) => Promise<{ ok: boolean; status: number; statusText?: string; text(): Promise<string> }>; + +/** + * Validate a ntfy topic URL. Accepts only `http(s)://host/topic` with a + * non-empty topic path. Returns `null` on success, a human-readable error + * string on failure. + */ +export function validateTopicUrl(topicUrl: string): string | null { + const trimmed = topicUrl.trim(); + if (!trimmed) return "Topic URL is required"; + let url: URL; + try { + url = new URL(trimmed); + } catch { + return "Topic URL is not a valid URL"; + } + if (url.protocol !== "http:" && url.protocol !== "https:") { + return "Topic URL must use http:// or https://"; + } + // Path must be a non-empty topic (more than just "/") + const topic = url.pathname.replace(/^\/+|\/+$/g, ""); + if (!topic) return "Topic URL must include a topic name (e.g. https://ntfy.sh/my-topic)"; + return null; +} + +/** + * Send a single notification to the configured ntfy topic. + * + * Fire-and-forget at call sites: the dispatcher uses + * `void sendNtfy(...).catch(...)` so a slow/broken ntfy server never blocks + * a turn. We still return a structured result so the explicit + * `POST /notifications/test` route can surface failures back to the UI. + * + * Pure with respect to `config` / `event` — no DB, no module state. + */ +export async function sendNtfy( + config: NtfyConfig, + event: NotificationEvent, + fetchImpl: FetchLike = globalThis.fetch as unknown as FetchLike, + timeoutMs = 10_000, +): Promise<NtfySendResult> { + if (!config.enabled) return { ok: false, error: "Notifications are disabled" }; + const topicErr = validateTopicUrl(config.topicUrl); + if (topicErr) return { ok: false, error: topicErr }; + + const priority = event.priority ?? NTFY_DEFAULT_PRIORITIES[event.type] ?? 3; + const baseTags = event.tags ?? NTFY_DEFAULT_TAGS[event.type] ?? []; + const tags = [...baseTags]; + if (event.tabId) { + // Short, ASCII-only tag so ntfy's comma-separated header parser is happy. + tags.push(`tab-${event.tabId.slice(0, 8)}`); + } + + const headers: Record<string, string> = { + // ntfy treats the title/priority/tags/click headers as ASCII-only. Strip + // control chars from the title; the body is sent UTF-8 verbatim. + Title: sanitizeHeader(event.title), + Priority: String(priority), + "Content-Type": "text/plain; charset=utf-8", + }; + if (tags.length > 0) headers.Tags = tags.map(sanitizeHeader).join(","); + if (event.clickUrl) headers.Click = event.clickUrl; + if (config.authToken && config.authToken.trim() !== "") { + headers.Authorization = `Bearer ${config.authToken.trim()}`; + } + + // Per-request abort so a hung server doesn't pin a Bun worker forever. + const controller = new AbortController(); + const timer = setTimeout(() => controller.abort(), timeoutMs); + + try { + const res = await fetchImpl(config.topicUrl.trim(), { + method: "POST", + headers, + body: event.message, + signal: controller.signal, + }); + if (!res.ok) { + const text = await safeReadText(res); + return { + ok: false, + status: res.status, + error: `ntfy responded ${res.status} ${res.statusText ?? ""}: ${text}`.trim(), + }; + } + return { ok: true, status: res.status }; + } catch (err) { + const msg = err instanceof Error ? err.message : String(err); + return { ok: false, error: msg }; + } finally { + clearTimeout(timer); + } +} + +function sanitizeHeader(value: string): string { + // Strip CR/LF (header injection guard) and trim. ntfy is tolerant of + // non-ASCII in titles, but we still drop control chars. + // biome-ignore lint/suspicious/noControlCharactersInRegex: intentional + return value.replace(/[\r\n\u0000-\u001f]+/g, " ").trim(); +} + +async function safeReadText(res: { text(): Promise<string> }): Promise<string> { + try { + const t = await res.text(); + return t.length > 200 ? `${t.slice(0, 200)}…` : t; + } catch { + return ""; + } +} diff --git a/packages/core/src/notifications/types.ts b/packages/core/src/notifications/types.ts new file mode 100644 index 0000000..f6baa27 --- /dev/null +++ b/packages/core/src/notifications/types.ts @@ -0,0 +1,98 @@ +// ntfy.sh push notifications — types + +/** + * Catalog of notification-worthy events. + * + * Kept intentionally small and stable: each entry is something a human + * actually wants pushed to their phone. New event types should be added + * with a sensible default (`NTFY_DEFAULT_EVENTS`) and a mapping in the + * dispatcher. + */ +export type NotificationEventType = + | "turn-completed" + | "turn-error" + | "permission-required" + | "agent-spawned"; + +/** ntfy priority levels (1=min … 5=max). */ +export type NtfyPriority = 1 | 2 | 3 | 4 | 5; + +/** + * A single notification request. Synthesised by the dispatcher from a + * higher-level event source (AgentManager / PermissionManager); fed to + * the ntfy transport. + * + * `dedupeKey` lets the dispatcher suppress duplicate sends (e.g. the + * permission system re-emits the pending list on every change). + */ +export interface NotificationEvent { + type: NotificationEventType; + /** Notification title (short). */ + title: string; + /** Notification body. */ + message: string; + /** Optional ntfy tags (emoji shortcodes — e.g. `["white_check_mark"]`). */ + tags?: string[]; + /** Optional priority override. Defaults are per-event-type. */ + priority?: NtfyPriority; + /** Optional URL the notification deep-links to when tapped. */ + clickUrl?: string; + /** Origin tab id (informational; included in tags as `tab:<short>`). */ + tabId?: string; + /** + * Stable key for suppressing duplicates. Same key + same type within a + * short window ⇒ dropped silently. + */ + dedupeKey?: string; +} + +/** + * Persisted ntfy configuration. Lives in the `settings` table under a + * single key (`ntfy_config`) — one global config, matching the codebase's + * existing single-user assumption (cf. `title_model_*`, `perm_*`). + * + * - `enabled` — master switch. Off ⇒ dispatcher never sends. + * - `topicUrl` — full URL, e.g. `https://ntfy.sh/my-secret-topic`. Missing + * ⇒ dispatcher never sends. + * - `authToken` — optional bearer token for private ntfy servers. + * - `events` — per-event-type enable map. Missing entries default to OFF + * so a newly-added event type doesn't silently start firing. + */ +export interface NtfyConfig { + enabled: boolean; + topicUrl: string; + authToken: string; + events: Record<NotificationEventType, boolean>; +} + +/** All event types this build knows about (the source of truth for UI). */ +export const NTFY_EVENT_TYPES: NotificationEventType[] = [ + "turn-completed", + "turn-error", + "permission-required", + "agent-spawned", +]; + +/** Default per-event-type toggles. */ +export const NTFY_DEFAULT_EVENTS: Record<NotificationEventType, boolean> = { + "turn-completed": true, + "turn-error": true, + "permission-required": true, + "agent-spawned": false, +}; + +/** Default priority per event type (when the event itself doesn't override). */ +export const NTFY_DEFAULT_PRIORITIES: Record<NotificationEventType, NtfyPriority> = { + "turn-completed": 3, + "turn-error": 4, + "permission-required": 4, + "agent-spawned": 2, +}; + +/** Default tag (emoji) per event type. */ +export const NTFY_DEFAULT_TAGS: Record<NotificationEventType, string[]> = { + "turn-completed": ["white_check_mark"], + "turn-error": ["rotating_light"], + "permission-required": ["lock"], + "agent-spawned": ["sparkles"], +}; |
