From 226b2d066a90b9ca7ef09aed8ccbc874b5404486 Mon Sep 17 00:00:00 2001 From: HeavyGee <133152184+heavygee@users.noreply.github.com> Date: Mon, 27 Jul 2026 12:52:54 +0100 Subject: [PATCH] feat(hub): native companion (FCM) push channel + device registry + pairing QR (#803) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * feat(hub): native companion (FCM) push channel + device registry Adds opt-in FCM HTTP v1 notification delivery so a companion mobile/wearable app can receive permission, ready, and task notifications end-to-end. The channel is gated entirely on FCM_SERVICE_ACCOUNT_PATH + FCM_PROJECT_ID being set; operators not running a companion see zero behavior change. What lands: - POST/DELETE /api/devices/register — JWT-authed FCM token registry, upsert on (namespace, deviceId, platform), platforms `phone` | `wear`. - Sqlite v9 → v10 migration adds `fcm_devices` (idx on namespace + token). - FcmService — minimal HTTP v1 client, RS256 service-account JWT via jose (dep already in tree), 5-minute access-token cache, 401 retry. - FcmNotificationChannel — implements NotificationChannel, sends data-only FCM (so companion can route to phone+watch surfaces). Body composition parses an optional trailing `AGENT_NOTIFY_SUMMARY {json}` line for richer ready summaries; truncates plain assistant text to 280 chars otherwise. Tags each payload with `severity` (info/warning/success/error) so clients can color/categorise the notification. - PushNotificationChannel gains a NativeFallbackProbe — when a namespace has at least one registered FCM device, web-push and SSE in-page toast are skipped so the operator does not double-notify on phone+browser. Probe is no-op when no FCM device is registered; PWA-only setups unchanged. Branch trace gated on HAPI_NOTIFY_DEBUG=1. - shared/src/messages.ts — `extractAssistantPlainText` (codex + Claude SDK shapes) and `extractNotifySummary` (strict end-anchored line parser). - hub/src/notifications/toolArgs.ts — tool-arg formatters lifted out of telegram/sessionView (kept duplicated there in this PR; refactor of Telegram is a follow-up). - docs/api/native-companion-contract.md — payload + endpoints + env vars, versioned at contract v1. Test coverage: - 260 hub tests pass (incl. 23 new across FCM channel, push dedup, v10 migration, devices route). - 60 shared tests pass (messages parsers). Notes for reviewers: - Reference companion implementation lives in a separate Android repo (Kotlin, phone APK + Wear OS APK) — this PR is hub-side only. - No new runtime deps (`jose` and `zod` already declared in hub). Co-authored-by: Cursor * docs(contract): clarify scope - companion is remote-hub client, not hub-on-phone Adds a Scope section to the native-companion contract so anyone implementing it knows the audience: operators running the hub on a server who want phone/watch as a notification surface, not users expecting a Termux-bundled hub. Mirrors the framing now in heavygee/hapi-companion README. Co-authored-by: Cursor * docs(contract): correct Scope section - hub topology is unchanged Removes the prior framing that referenced a non-existent 'Termux hub-on-phone' alternative. This contract describes a native client to the same hub the PWA talks to; it does not change where the hub runs. Co-authored-by: Cursor * feat(web): companion app pairing QR in Settings Companion section in Settings renders a QR code encoding the deeplink hapicompanion://bind?hub=&code=. Scanning it from the HAPI companion app (Android phone or Wear OS) auto-fills the bind form and authenticates against this hub - no manual URL/token paste. QR is gated behind a Show button so the access token doesn't sit visible on screen by default; a Copy link affordance and the textual deeplink are also exposed for manual onboarding. Adds qrcode + @types/qrcode to web/ (already a hub dep, no new resolved package - just a workspace declaration). Co-authored-by: Cursor * feat(hub): terminal QR for companion app pairing alongside PWA QR After the existing PWA access QR is rendered on tunnel start, also print the hapicompanion://bind?hub=...&code=... deeplink and a matching QR. Same tunnel + token, different scheme: phones with the companion app installed pick up the deeplink via the manifest intent filter; phones without it ignore it and fall back to the PWA QR above. QR rendering failure is non-fatal in both cases - the textual deeplink above the QR is sufficient for manual paste. Co-authored-by: Cursor * fix(fcm): address HAPI Bot review on PR #803 Two bugs surfaced by the upstream review bot: 1) Web Push silently dropped when FCM is not actually configured. The native-fallback probe only checked the device registry; it did not check whether resolveFcmConfig() actually succeeded. So an operator who previously enabled FCM, registered a phone, then later started the hub WITHOUT FCM_SERVICE_ACCOUNT_PATH would see the probe return true (devices still in DB) -> Web Push suppressed -> no FCM channel registered -> notifications go to /dev/null. Fix: extracted the probe construction into buildNativeFallbackProbe() which short-circuits to () => false when fcmConfig is missing. Probe never even consults the device store in the no-config branch, so stale rows can never matter. 2) Transient FCM failures permanently unregistered devices. sendToToken() returned a single boolean and sendToNamespace() removed any device whose send returned false. A 429 (rate limit), 503 (server error), 401 (auth glitch), or even an ECONNREFUSED would delete the device row, after which the user would need to re-pair to get notifications again. The bot caught it; the fix is the obvious one. Fix: sendToToken() now returns 'sent' | 'invalid' | 'failed'. - 'invalid' is reserved for the responses that genuinely indicate a dead token: HTTP 404 with UNREGISTERED/NOT_FOUND, and HTTP 400 with INVALID_ARGUMENT explicitly referencing the token field. - Everything else (429, 5xx, 401, 403, network errors) is 'failed' and counts toward the failed tally without removing the device. sendToNamespace() only calls removeDeviceByToken() on 'invalid'. Tests: 11 new tests across two new files. fcmService.test.ts covers all six branches (200, 404 unregistered, 429, 503, 401, network error) plus a mixed-batch case that proves invalid tokens get removed in the same call where transient-failure tokens survive. nativeFallbackProbe .test.ts covers both no-config and configured branches plus the explicit "no-config never touches the store" guarantee. Hub test count: 273 -> 284 (all passing). Co-authored-by: Cursor * docs(contract): correct FCM visibility rule and remove unsupported event type HAPI Bot review on PR #803 caught two contract-doc accuracy gaps: 1) Visibility rule was wrong. Doc said "FCM fires when Web Push would fire AND client not visible via SSE", but FcmNotificationChannel ALWAYS fires regardless of PWA visibility (deliberately - native companion is the canonical wrist-first surface, and there is a passing test asserting this). Companion app implementers reading the contract would have built foreground-suppression logic and then dropped notifications when the PWA tab was open. 2) Documented `session-completed` event doesn't exist. NotificationHub never calls into a 'session-completed' channel method on FcmNotificationChannel; the type would never reach a native client. Removed from the documented enum, leaving only the three actual events: ready, permission-request, task-notification. Co-authored-by: Cursor * docs(contract): drop trailing whitespace, use blank line for paragraph break Co-authored-by: Cursor * fix(web): persist CLI access token after Telegram bind so pairing QR works The Settings -> Companion pairing QR reads the original CLI access token from localStorage (hapi_access_token::) so it can be encoded into the hapicompanion://bind deeplink. For browser/CLI logins useAuthSource already persists the token via setAccessToken, but the Telegram Mini App bind path went through useAuth.bind() which exchanged the typed CLI token for a JWT and never persisted it. Telegram users therefore always saw the "signed in via Telegram..." fallback and got no usable QR. After a successful client.bind() we now mirror useAuthSource's behavior and write the same accessToken to the same localStorage key, restoring parity between the two auth paths. No change for browser/CLI users. Co-authored-by: Cursor * fix(fcm): gate native-fallback probe on rolling FCM health The native-fallback probe previously returned true whenever FCM was configured AND devices were registered, which suppressed web-push for the namespace. The HAPI Bot correctly pointed out the gap: if the FCM pipeline silently breaks (expired service-account key, sustained 5xx, OAuth token-fetch failure, network blackhole) the operator gets nothing on either channel until they manually intervene. Approach (deliberate, not the bot's exact suggested fix): - FcmService now keeps a small rolling window (last 8 outcomes) of send attempts and exposes `isHealthy()`. The threshold is 5+/8 failures = unhealthy; the buffer starts empty so a freshly-booted hub is optimistic ("innocent until proven guilty") and does not double-fire on event #1. - Token-fetch failure (`getFcmAccessToken` throws) now records exactly one health-failure (not one per device), short-circuits the send loop, and returns a result so `sendToNamespace` no longer leaks the exception. - `invalid` token responses are explicitly excluded from the health buffer because they are per-device facts (rotated/uninstalled token), not pipeline failures - FCM was reachable, it just rejected one stale token. - `buildNativeFallbackProbe` now optionally accepts the FcmService and short-circuits to "let web-push fire" when health is bad, before it even queries the device registry. The single-arg call shape is still supported for back-compat. Why not the bot's exact suggestion ("invert: call FCM first, fall back on result.sent === 0"): - Couples PushNotificationChannel to FcmService and FcmSendPayload, reversing the clean parallel-channel architecture established earlier in this PR. - Treats every transient single-event failure as fallback-worthy, which re-opens the duplicate-notification race that the suppression logic was added to close (FCM HTTP timeout that delivers later + the web push we sent in the meantime = two pings). - A rolling health window only flips on sustained breakage, which is the actual operational scenario the bot is worried about. The wrist-first design intent ("FCM fires unconditionally, web-push is suppressed for the same namespace") documented in docs/api/native-companion-contract.md is preserved on the happy path. The probe only re-enables web-push when there is concrete evidence the native pipeline is not delivering. Tests: - New FcmService.isHealthy suite covers empty-buffer, threshold flip, recovery as failures age out of the window, invalid-token exclusion, and network-error path. - nativeFallbackProbe gains coverage for the unhealthy-but-registered, healthy-and-registered, and absent-fcmService (back-compat) cases. - All 292 hub tests still pass; typecheck clean. Co-authored-by: Cursor * refactor(telegram): drop duplicate tool-args formatter, use shared module The Telegram session view had its own copy of formatToolArgumentsDetailed identical to the one in hub/src/notifications/toolArgs.ts (already used by the FCM channel). Replace the local copy with an import. Removes ~70 lines of duplication, plus the now-unused MAX_TOOL_ARGS_LENGTH constant and `truncate` import. The shared signature accepts an optional opts arg whose default maxArgLength is 150 - matching the prior constant - so the call site is unchanged. Two benign upgrades come along for the ride from the shared module: ?? instead of || on field fallbacks (no real-world difference; permission arguments never carry empty-string fields), and String(...) wrapping plus a typeof object guard that makes non-string values render gracefully instead of throwing into the catch block. Hub tests: 311 pass / 0 fail. Telegram subset: 5 pass / 0 fail. typecheck green. Cold-reviewed by an out-of-context Claude Opus peer before push. Co-authored-by: Cursor * fix(fcm): require positive evidence in health window before suppressing web-push Addresses HAPI Bot Major review on PR #803. The previous health gate treated an empty outcome buffer as healthy ("innocent until proven guilty"). That created a silent-blackhole window on cold start with broken FCM credentials: the push channel suppressed SSE/Web Push for the first ~5 events while the FCM channel attempted each delivery and recorded failures, until enough stacked to flip the threshold. Every notification in that gap was silently lost. New invariant: isHealthy() requires at least one successful FCM send in the recent window (HEALTH_WINDOW=8) AND failures below threshold (HEALTH_FAILURE_THRESHOLD=5). Both conditions are necessary; either alone is insufficient evidence to safely suppress web-push fallback. Trade-off: one duplicated notification per hub restart per namespace. On the first event after restart, web-push fires alongside FCM (because the gate has no positive evidence yet). Once FCM records that first success, the gate engages and subsequent events are FCM-only. Worth it for guaranteed delivery during cold-start outages. Tests reworked to match new semantics: - "starts UNHEALTHY with empty buffer" (was: healthy) - "flips to healthy after first successful send" (new) - "stays unhealthy across failures-only run" (new, exercises the exact blackhole scenario the bot flagged) - "flips back to unhealthy after threshold breach with prior successes" (renamed, establishes successes first) - "invalid tokens don't count against health" (reworked: send a mixed batch first to establish health, then verify invalids don't flip it) - "network errors count as failures" (reworked: establish health first) Hub tests: 313 pass / 0 fail. typecheck green. Co-authored-by: Cursor * fix(hub): bump FCM migration to V10→V11 after upstream service_tier V9→V10 Upstream/main landed sessions.service_tier at schema v10. The companion FCM device registry now migrates at v11 so both changes compose cleanly after the courtesy rebase onto current upstream/main. Co-authored-by: Cursor * fix(hub): per-dispatch native gate instead of stale FCM probe FCM runs before web-push; PushNotificationChannel skips web/SSE only when the same notify() dispatch already delivered via FCM. Removes the isHealthy()+device-row probe that could suppress web-push after warm FCM outages. Co-authored-by: Cursor * fix(hub,web): cap notifySummary for FCM limits; fix PWA test cast Rebase follow-up: truncate AGENT_NOTIFY_SUMMARY summary/action before FCM data payload (bot Major). Fix usePwaUpdate.test.ts setTimeout mock cast so bun typecheck passes on current main. Co-authored-by: Cursor * fix(hub): cap all FCM notifySummary fields and task bodies Whitelist and truncate AGENT_NOTIFY_SUMMARY auxiliary fields before JSON serialization; cap task-notification summaries to glance limit. Co-authored-by: Cursor * fix(hub): FCM fetch timeouts and cap Grep/Glob permission args 10s AbortSignal.timeout on OAuth + FCM send so sequential web-push fallback is not blocked on hung Google endpoints; truncate Grep/Glob pattern in permission detail formatter. Co-authored-by: Cursor * fix(hub): bind FCM token to one namespace on re-pair Delete stale fcm_devices rows sharing the same token when a native install registers under a different namespace. Co-authored-by: Cursor * fix(web): localize Companion settings and pairing copy Add en/zh-CN keys for the Companion section title and CompanionPairing strings; matches locale-driven Settings pattern (bot Minor on #803). Co-authored-by: Cursor * fix(hub): tighten FCM token-invalid detection and truncation edge cases Parse FCM error JSON: only UNREGISTERED or token-field INVALID_ARGUMENT unregister devices; generic NOT_FOUND stays transient. Guard limit<=3 in truncateReadyText so tiny action budgets cannot blow the glance cap. Co-authored-by: Cursor * fix(hub): parse FcmError details.errorCode for UNREGISTERED tokens FCM v1 often returns HTTP 404 with root NOT_FOUND plus details[].errorCode UNREGISTERED; prune those tokens while keeping generic project/resource NOT_FOUND transient. Co-authored-by: Cursor * fix(web): mock AppContext for About Companion pairing in settings tests Settings About now mounts CompanionPairing via useAppContext after the #1027 hub redesign rebase; wrap the About route test with AppContext and Companion mocks so the suite stays green. Co-authored-by: Cursor * docs(contract): point companion auth at POST /api/auth, not /api/bind Pairing QR carries the CLI access token as `code`. /api/bind requires Telegram initData; native companions must use /api/auth with accessToken. Co-authored-by: Cursor * fix(web): mount Companion pairing under Settings General About is version/links only after the settings hub redesign; pairing is setup, so keep Companion with language prefs and update the route tests. Co-authored-by: Cursor --------- Co-authored-by: Cursor Co-authored-by: Debian --- bun.lock | 2 + docs/api/native-companion-contract.md | 105 ++++ hub/src/fcm/fcmAuth.ts | 72 +++ hub/src/fcm/fcmConfig.ts | 26 + hub/src/fcm/fcmNotificationChannel.test.ts | 595 ++++++++++++++++++ hub/src/fcm/fcmNotificationChannel.ts | 308 +++++++++ hub/src/fcm/fcmService.test.ts | 440 +++++++++++++ hub/src/fcm/fcmService.ts | 277 ++++++++ hub/src/notifications/notificationHub.ts | 10 +- .../notifications/notificationSendContext.ts | 14 + hub/src/notifications/notificationTypes.ts | 7 +- hub/src/notifications/toolArgs.ts | 157 +++++ hub/src/push/pushNotificationChannel.test.ts | 116 ++++ hub/src/push/pushNotificationChannel.ts | 83 +-- hub/src/push/pushService.ts | 2 + hub/src/startHub.ts | 55 +- hub/src/store/fcmDevices.test.ts | 16 + hub/src/store/fcmDevices.ts | 75 +++ hub/src/store/fcmStore.ts | 23 + hub/src/store/index.ts | 41 +- hub/src/store/migration-v10.test.ts | 131 ++++ hub/src/store/types.ts | 10 + hub/src/telegram/sessionView.ts | 78 +-- hub/src/web/routes/devices.test.ts | 56 ++ hub/src/web/routes/devices.ts | 44 ++ hub/src/web/server.ts | 2 + shared/src/messages.test.ts | 221 +++++++ shared/src/messages.ts | 103 +++ web/package.json | 2 + .../components/settings/CompanionPairing.tsx | 126 ++++ web/src/hooks/useAuth.ts | 14 + web/src/hooks/usePwaUpdate.test.ts | 2 +- web/src/lib/locales/en.ts | 10 +- web/src/lib/locales/zh-CN.ts | 10 +- web/src/routes/settings/general.tsx | 8 + web/src/routes/settings/index.test.tsx | 14 + 36 files changed, 3124 insertions(+), 131 deletions(-) create mode 100644 docs/api/native-companion-contract.md create mode 100644 hub/src/fcm/fcmAuth.ts create mode 100644 hub/src/fcm/fcmConfig.ts create mode 100644 hub/src/fcm/fcmNotificationChannel.test.ts create mode 100644 hub/src/fcm/fcmNotificationChannel.ts create mode 100644 hub/src/fcm/fcmService.test.ts create mode 100644 hub/src/fcm/fcmService.ts create mode 100644 hub/src/notifications/notificationSendContext.ts create mode 100644 hub/src/notifications/toolArgs.ts create mode 100644 hub/src/store/fcmDevices.test.ts create mode 100644 hub/src/store/fcmDevices.ts create mode 100644 hub/src/store/fcmStore.ts create mode 100644 hub/src/store/migration-v10.test.ts create mode 100644 hub/src/web/routes/devices.test.ts create mode 100644 hub/src/web/routes/devices.ts create mode 100644 shared/src/messages.test.ts create mode 100644 web/src/components/settings/CompanionPairing.tsx diff --git a/bun.lock b/bun.lock index 4fe8c67e..6bc7ef30 100644 --- a/bun.lock +++ b/bun.lock @@ -118,6 +118,7 @@ "html2canvas-pro": "^2.0.4", "katex": "^0.16.45", "mermaid": "^11.12.0", + "qrcode": "^1.5.4", "react": "^19.2.3", "react-dom": "^19.2.3", "rehype-katex": "^7.0.1", @@ -138,6 +139,7 @@ "@tailwindcss/postcss": "^4.1.18", "@testing-library/jest-dom": "^6.6.3", "@testing-library/react": "^16.3.0", + "@types/qrcode": "^1.5.6", "@types/react": "^19.2.7", "@types/react-dom": "^19.2.3", "@vitejs/plugin-react": "^5.1.2", diff --git a/docs/api/native-companion-contract.md b/docs/api/native-companion-contract.md new file mode 100644 index 00000000..09238cf8 --- /dev/null +++ b/docs/api/native-companion-contract.md @@ -0,0 +1,105 @@ +# Native companion API contract (phone + Wear) + +**Audience:** Implementers of native companion apps (Android phone + Wear OS, iOS, etc.) that pair with a hapi hub via FCM. + +**Auth:** Exchange the pairing `code` / CLI access token with `POST /api/auth`: +`{ "accessToken": "" }`. Use the returned JWT as `Authorization: Bearer ` +for device registration and session actions. `POST /api/bind` is only for Telegram Mini App +binding (requires Telegram `initData`). + +## Scope + +A companion implementing this contract is a **native client to the same hub the PWA talks to**, surfacing notifications and reply / approve actions on a phone or wearable. Hub topology is unchanged - the hub still runs on the operator's dev machine. + +--- + +## Device registration (FCM) + +### Register + +`POST /api/devices/register` + +```json +{ + "token": "", + "platform": "phone", + "deviceId": "" +} +``` + +`platform`: `"phone"` | `"wear"` + +**Response:** `{ "ok": true }` + +Upsert on `(namespace, deviceId, platform)` - same device re-registering replaces the FCM token. + +### Unregister + +`DELETE /api/devices/register` + +```json +{ + "token": "" +} +``` + +--- + +## Outbound push (hub → device) + +Hub sends FCM HTTP v1 whenever a notification event is emitted for a +namespace with registered native devices and FCM is configured. The native +companion is treated as the canonical wrist-first surface, so FCM fires +**unconditionally** (independent of whether a PWA tab happens to be +foreground / visible via SSE) - that's deliberate, see +`FcmNotificationChannel.deliver()`. Web Push is suppressed for the same +namespace to avoid duplicate OS notifications. + +### Data payload (all platforms) + +| Key | Example | Purpose | +|-----|---------|---------| +| `type` | `ready` | `ready`, `permission-request`, `task-notification` | +| `sessionId` | uuid | Target session | +| `sessionName` | string | Display name (`agent - project`) | +| `url` | `/sessions/{id}` | Deep link path | +| `requestId` | uuid | Permission only - approve/deny | +| `title` | string | Notification title | +| `body` | string | Notification body | +| `severity` | `info` | `info` (ready), `warning` (permission), `success` / `error` (task) | +| `notifySummary` | JSON string | Optional: parsed `AGENT_NOTIFY_SUMMARY` line from agent text | + +Native apps **must** handle `data` for Wear; notification block is for display. + +### Client actions (native - not hub) + +| User action | Hub API | +|-------------|---------| +| Send text | `POST /api/sessions/:id/messages` `{ "text": "...", "localId": "..." }` | +| Allow | `POST /api/sessions/:id/permissions/:requestId/approve` | +| Deny | `POST /api/sessions/:id/permissions/:requestId/deny` | + +`sentFrom` extension (optional future): `android-phone`, `android-wear`. + +--- + +## Environment (hub operator) + +```bash +FCM_SERVICE_ACCOUNT_PATH=/path/to/service-account.json +FCM_PROJECT_ID=your-firebase-project-id +``` + +When unset, hub skips FCM channel (Web Push / Telegram unchanged). + +The native push channel is **opt-in**: operators who don't run a companion +app see no behavior change. When at least one device is registered for a +namespace, the existing Web Push channel suppresses its fallback for that +namespace to avoid double-notifying (one in the native app, one from the +PWA service worker). PWA-only operators are unaffected. + +--- + +## Versioning + +Contract version **1**. Breaking changes require `data.contractVersion` in FCM payload and doc update. diff --git a/hub/src/fcm/fcmAuth.ts b/hub/src/fcm/fcmAuth.ts new file mode 100644 index 00000000..9049bb20 --- /dev/null +++ b/hub/src/fcm/fcmAuth.ts @@ -0,0 +1,72 @@ +import { readFileSync } from 'node:fs' +import * as jose from 'jose' + +export type ServiceAccount = { + client_email: string + private_key: string + project_id?: string +} + +const FCM_SCOPE = 'https://www.googleapis.com/auth/firebase.messaging' +export const FCM_REQUEST_TIMEOUT_MS = 10_000 + +let cachedToken: { accessToken: string; expiresAtMs: number } | null = null + +export function loadServiceAccount(path: string): ServiceAccount { + const raw = readFileSync(path, 'utf8') + const parsed = JSON.parse(raw) as ServiceAccount + if (!parsed.client_email || !parsed.private_key) { + throw new Error('FCM service account JSON missing client_email or private_key') + } + return parsed +} + +export async function getFcmAccessToken(serviceAccount: ServiceAccount): Promise { + const nowMs = Date.now() + if (cachedToken && cachedToken.expiresAtMs > nowMs + 60_000) { + return cachedToken.accessToken + } + + const nowSec = Math.floor(nowMs / 1000) + const key = await jose.importPKCS8(serviceAccount.private_key, 'RS256') + const assertion = await new jose.SignJWT({ scope: FCM_SCOPE }) + .setProtectedHeader({ alg: 'RS256', typ: 'JWT' }) + .setIssuer(serviceAccount.client_email) + .setSubject(serviceAccount.client_email) + .setAudience('https://oauth2.googleapis.com/token') + .setIssuedAt(nowSec) + .setExpirationTime(nowSec + 3600) + .sign(key) + + const response = await fetch('https://oauth2.googleapis.com/token', { + method: 'POST', + headers: { 'content-type': 'application/x-www-form-urlencoded' }, + body: new URLSearchParams({ + grant_type: 'urn:ietf:params:oauth:grant-type:jwt-bearer', + assertion + }), + signal: AbortSignal.timeout(FCM_REQUEST_TIMEOUT_MS) + }) + + if (!response.ok) { + const body = await response.text().catch(() => '') + throw new Error(`FCM OAuth token exchange failed: ${response.status} ${body}`) + } + + const json = await response.json() as { access_token?: string; expires_in?: number } + if (!json.access_token) { + throw new Error('FCM OAuth response missing access_token') + } + + const expiresInSec = json.expires_in ?? 3600 + cachedToken = { + accessToken: json.access_token, + expiresAtMs: nowMs + expiresInSec * 1000 + } + return cachedToken.accessToken +} + +/** Test helper */ +export function clearFcmAccessTokenCache(): void { + cachedToken = null +} diff --git a/hub/src/fcm/fcmConfig.ts b/hub/src/fcm/fcmConfig.ts new file mode 100644 index 00000000..626c016b --- /dev/null +++ b/hub/src/fcm/fcmConfig.ts @@ -0,0 +1,26 @@ +import { existsSync } from 'node:fs' +import { loadServiceAccount } from './fcmAuth' + +export type FcmConfig = { + projectId: string + serviceAccountPath: string + serviceAccount: ReturnType +} + +export function resolveFcmConfig(): FcmConfig | null { + const serviceAccountPath = process.env.FCM_SERVICE_ACCOUNT_PATH?.trim() + if (!serviceAccountPath || !existsSync(serviceAccountPath)) { + return null + } + + const serviceAccount = loadServiceAccount(serviceAccountPath) + const projectId = process.env.FCM_PROJECT_ID?.trim() + || serviceAccount.project_id + || null + if (!projectId) { + console.warn('[Fcm] FCM_PROJECT_ID unset and service account JSON has no project_id') + return null + } + + return { projectId, serviceAccountPath, serviceAccount } +} diff --git a/hub/src/fcm/fcmNotificationChannel.test.ts b/hub/src/fcm/fcmNotificationChannel.test.ts new file mode 100644 index 00000000..50c00e54 --- /dev/null +++ b/hub/src/fcm/fcmNotificationChannel.test.ts @@ -0,0 +1,595 @@ +import { describe, expect, it } from 'bun:test' +import type { Session } from '../sync/syncEngine' +import { FcmNotificationChannel } from './fcmNotificationChannel' +import type { FcmSendPayload } from './fcmService' + +function createSession(overrides: Partial = {}): Session { + return { + id: 'session-ready', + namespace: 'default', + name: 'Demo', + active: true, + metadata: { flavor: 'codex', name: 'Demo' }, + ...overrides + } as Session +} + +describe('FcmNotificationChannel', () => { + it('always fires FCM regardless of PWA visibility (wrist-first)', async () => { + const sent: FcmSendPayload[] = [] + const toasts: unknown[] = [] + const channel = new FcmNotificationChannel( + { + sendToNamespace: async (_namespace: string, payload: FcmSendPayload) => { + sent.push(payload) + } + } as never, + { + sendToast: async (_namespace: string, event: unknown) => { + toasts.push(event) + return 1 + } + } as never, + { + hasVisibleConnection: () => true + } as never + ) + + await channel.sendReady(createSession()) + + // The watch is the canonical surface when a native companion is + // registered. The previous behaviour silently swallowed FCM when + // the PWA was foreground - that broke the wrist-first UX. We now + // fire FCM unconditionally and let the PWA's own SyncEngine event + // stream handle in-page toasts (or not, per UX preference). + expect(sent).toHaveLength(1) + expect(toasts).toHaveLength(0) + }) + + it('includes requestId on permission-request payloads', async () => { + const sent: FcmSendPayload[] = [] + const channel = new FcmNotificationChannel( + { + sendToNamespace: async (_namespace: string, payload: FcmSendPayload) => { + sent.push(payload) + } + } as never, + { + sendToast: async () => 0 + } as never, + { + hasVisibleConnection: () => false + } as never + ) + + await channel.sendPermissionRequest(createSession({ + agentState: { + requests: { + 'req-42': { tool: 'Bash', arguments: {} } + } + } + })) + + expect(sent).toHaveLength(1) + expect(sent[0].data.type).toBe('permission-request') + expect(sent[0].data.requestId).toBe('req-42') + expect(sent[0].data.contractVersion).toBe('1') + }) + + it('enriches permission-request body with tool args (Edit)', async () => { + const sent: FcmSendPayload[] = [] + const channel = new FcmNotificationChannel( + { + sendToNamespace: async (_namespace: string, payload: FcmSendPayload) => { + sent.push(payload) + } + } as never, + { + sendToast: async () => 0 + } as never, + { + hasVisibleConnection: () => false + } as never + ) + + await channel.sendPermissionRequest(createSession({ + agentState: { + requests: { + 'req-99': { + tool: 'Edit', + arguments: { + file_path: '/home/u/proj/hub/src/server.ts', + old_string: 'foo', + new_string: 'bar' + } + } + } + } + })) + + expect(sent).toHaveLength(1) + const body = sent[0].body ?? '' + const dataBody = sent[0].data.body ?? '' + // Glance line: agent + tool + compact arg (last two path segments). + expect(body).toContain('Edit') + expect(body).toContain('hub/src/server.ts') + // Detail: full file path on its own line, plus old/new previews - + // visible when the watch operator taps to expand. + expect(body).toContain('File: /home/u/proj/hub/src/server.ts') + expect(body).toContain('Old: "foo"') + expect(body).toContain('New: "bar"') + // data.body must mirror notification body so the watch sees the same + // text the FCM `notification` field would. + expect(dataBody).toBe(body) + }) + + it('truncates long Grep permission patterns for FCM data limits', async () => { + const sent: FcmSendPayload[] = [] + const channel = new FcmNotificationChannel( + { + sendToNamespace: async (_namespace: string, payload: FcmSendPayload) => { + sent.push(payload) + } + } as never, + { sendToast: async () => 0 } as never, + { hasVisibleConnection: () => false } as never + ) + + await channel.sendPermissionRequest(createSession({ + agentState: { + requests: { + 'req-grep': { + tool: 'Grep', + arguments: { pattern: 'P'.repeat(500), path: '/tmp' } + } + } + } + })) + + expect(sent[0].body).toContain('Pattern:') + expect(sent[0].body).toContain('...') + expect(sent[0].data.body.length).toBeLessThan(600) + }) + + it('falls back gracefully when no tool args are present', async () => { + const sent: FcmSendPayload[] = [] + const channel = new FcmNotificationChannel( + { + sendToNamespace: async (_namespace: string, payload: FcmSendPayload) => { + sent.push(payload) + } + } as never, + { + sendToast: async () => 0 + } as never, + { + hasVisibleConnection: () => false + } as never + ) + + await channel.sendPermissionRequest(createSession({ + agentState: { + requests: { + 'req-1': { tool: 'NewExperimentalTool', arguments: { foo: 'bar' } } + } + } + })) + + const body = sent[0].body ?? '' + // Compact returns '' for tools not in its switch table, so the glance + // line collapses to bare " " - confirming we never emit + // " : " with a dangling colon. + expect(body.split('\n')[0]).toMatch(/NewExperimentalTool$/) + expect(body).not.toContain(': \n') + }) + + function makeStoreWithMessages(messages: Array<{ content: unknown }>) { + // Minimal store stub: only the bits FcmNotificationChannel touches. + // Mirrors the real `getMessages` contract: callers receive the last N + // rows in ASCENDING seq order (oldest first, latest last). + return { + messages: { + getMessages: (_sessionId: string, _limit: number) => messages.map((m, i) => ({ + id: `m-${i}`, + sessionId: 'session-ready', + content: m.content, + createdAt: i, + seq: i + 1, + localId: null, + invokedAt: null, + scheduledAt: null + })) + } + } as never + } + + it('sendReady prefers AGENT_NOTIFY_SUMMARY when last assistant message has one', async () => { + const sent: FcmSendPayload[] = [] + const store = makeStoreWithMessages([ + // DESC order: index 0 = latest. Latest assistant text contains a summary. + { + content: { + role: 'agent', + content: { + type: 'codex', + data: { + type: 'message', + message: 'Did the work.\n\nAGENT_NOTIFY_SUMMARY {"version":1,"summary":"Tokens revoked","action":"Upload preview","status":"done"}' + } + } + } + } + ]) + const channel = new FcmNotificationChannel( + { + sendToNamespace: async (_namespace: string, payload: FcmSendPayload) => { + sent.push(payload) + } + } as never, + { sendToast: async () => 0 } as never, + { hasVisibleConnection: () => false } as never, + store + ) + + await channel.sendReady(createSession()) + + expect(sent).toHaveLength(1) + const p = sent[0] + expect(p.title).toBe('Codex - Demo') + expect(p.body).toBe('Tokens revoked\n-> Upload preview') + expect(p.data.notifySummary).toBeDefined() + const parsed = JSON.parse(p.data.notifySummary as string) + expect(parsed.summary).toBe('Tokens revoked') + expect(parsed.action).toBe('Upload preview') + }) + + it('sendReady caps long AGENT_NOTIFY_SUMMARY text for FCM data limits', async () => { + const sent: FcmSendPayload[] = [] + const longSummary = 'S'.repeat(400) + const longAction = 'A'.repeat(400) + const store = makeStoreWithMessages([ + { + content: { + role: 'agent', + content: { + type: 'codex', + data: { + type: 'message', + message: `Done.\n\nAGENT_NOTIFY_SUMMARY {"version":1,"summary":"${longSummary}","action":"${longAction}","status":"done"}` + } + } + } + } + ]) + const channel = new FcmNotificationChannel( + { + sendToNamespace: async (_namespace: string, payload: FcmSendPayload) => { + sent.push(payload) + } + } as never, + { sendToast: async () => 0 } as never, + { hasVisibleConnection: () => false } as never, + store + ) + + await channel.sendReady(createSession()) + + expect(sent).toHaveLength(1) + expect(sent[0].body.length).toBeLessThanOrEqual(280) + const parsed = JSON.parse(sent[0].data.notifySummary as string) + expect(parsed.summary.length).toBeLessThanOrEqual(280) + expect(parsed.action.length).toBeLessThanOrEqual(280) + }) + + it('sendReady respects tiny action budget when summary fills the glance limit', async () => { + const sent: FcmSendPayload[] = [] + const summary278 = 'S'.repeat(278) + const store = makeStoreWithMessages([ + { + content: { + role: 'agent', + content: { + type: 'codex', + data: { + type: 'message', + message: `Done.\n\nAGENT_NOTIFY_SUMMARY {"version":1,"summary":"${summary278}","action":"ACTION","status":"done"}` + } + } + } + } + ]) + const channel = new FcmNotificationChannel( + { + sendToNamespace: async (_namespace: string, payload: FcmSendPayload) => { + sent.push(payload) + } + } as never, + { sendToast: async () => 0 } as never, + { hasVisibleConnection: () => false } as never, + store + ) + + await channel.sendReady(createSession()) + + expect(sent[0].body.length).toBeLessThanOrEqual(280) + expect(sent[0].body).not.toContain('ACTION') + }) + + it('sendReady caps auxiliary notifySummary fields for FCM data limits', async () => { + const sent: FcmSendPayload[] = [] + const longAgent = 'G'.repeat(200) + const longProject = 'P'.repeat(200) + const store = makeStoreWithMessages([ + { + content: { + role: 'agent', + content: { + type: 'codex', + data: { + type: 'message', + message: `Done.\n\nAGENT_NOTIFY_SUMMARY {"version":1,"summary":"ok","action":"go","status":"${'x'.repeat(80)}","agent":"${longAgent}","project":"${longProject}"}` + } + } + } + } + ]) + const channel = new FcmNotificationChannel( + { + sendToNamespace: async (_namespace: string, payload: FcmSendPayload) => { + sent.push(payload) + } + } as never, + { sendToast: async () => 0 } as never, + { hasVisibleConnection: () => false } as never, + store + ) + + await channel.sendReady(createSession()) + + const parsed = JSON.parse(sent[0].data.notifySummary as string) + expect(parsed.status.length).toBeLessThanOrEqual(32) + expect(parsed.agent.length).toBeLessThanOrEqual(80) + expect(parsed.project.length).toBeLessThanOrEqual(80) + }) + + it('sendReady truncates last assistant text when no summary is present', async () => { + const sent: FcmSendPayload[] = [] + const longText = 'A'.repeat(500) + const store = makeStoreWithMessages([ + { + content: { + role: 'agent', + content: { type: 'codex', data: { type: 'message', message: longText } } + } + } + ]) + const channel = new FcmNotificationChannel( + { + sendToNamespace: async (_namespace: string, payload: FcmSendPayload) => { + sent.push(payload) + } + } as never, + { sendToast: async () => 0 } as never, + { hasVisibleConnection: () => false } as never, + store + ) + + await channel.sendReady(createSession()) + + const body = sent[0].body + expect(body.length).toBeLessThanOrEqual(280) + expect(body.endsWith('...')).toBe(true) + expect(sent[0].data.notifySummary).toBeUndefined() + }) + + it('sendReady skips tool-call messages and uses the last assistant TEXT message', async () => { + const sent: FcmSendPayload[] = [] + // Real getMessages returns ASC (oldest first, newest last). The newest + // here is a tool-call-result; the channel must walk back past two + // tool-call frames to find the actual assistant text. + const store = makeStoreWithMessages([ + { + content: { + role: 'agent', + content: { type: 'codex', data: { type: 'message', message: 'The actual reply.' } } + } + }, + { + content: { + role: 'agent', + content: { type: 'codex', data: { type: 'tool-call', name: 'Bash', callId: 'x', input: {} } } + } + }, + { + content: { + role: 'agent', + content: { type: 'codex', data: { type: 'tool-call-result', output: {} } } + } + } + ]) + const channel = new FcmNotificationChannel( + { + sendToNamespace: async (_namespace: string, payload: FcmSendPayload) => { + sent.push(payload) + } + } as never, + { sendToast: async () => 0 } as never, + { hasVisibleConnection: () => false } as never, + store + ) + + await channel.sendReady(createSession()) + + expect(sent[0].body).toBe('The actual reply.') + }) + + it('sendReady picks the LATEST assistant text when multiple text messages exist (ASC ordering regression guard)', async () => { + const sent: FcmSendPayload[] = [] + // Two text messages, oldest first - the channel must return the + // last one ("Latest reply.") not the first ("Older reply."). + // This guards against a real bug where we walked the array + // assuming DESC ordering and picked the oldest. + const store = makeStoreWithMessages([ + { + content: { + role: 'agent', + content: { type: 'codex', data: { type: 'message', message: 'Older reply.' } } + } + }, + { + content: { + role: 'agent', + content: { type: 'codex', data: { type: 'message', message: 'Latest reply.' } } + } + } + ]) + const channel = new FcmNotificationChannel( + { + sendToNamespace: async (_namespace: string, payload: FcmSendPayload) => { + sent.push(payload) + } + } as never, + { sendToast: async () => 0 } as never, + { hasVisibleConnection: () => false } as never, + store + ) + + await channel.sendReady(createSession()) + + expect(sent[0].body).toBe('Latest reply.') + }) + + it('sendReady falls back to "is waiting" line when no agent text exists', async () => { + const sent: FcmSendPayload[] = [] + const store = makeStoreWithMessages([]) + const channel = new FcmNotificationChannel( + { + sendToNamespace: async (_namespace: string, payload: FcmSendPayload) => { + sent.push(payload) + } + } as never, + { sendToast: async () => 0 } as never, + { hasVisibleConnection: () => false } as never, + store + ) + + await channel.sendReady(createSession()) + + expect(sent[0].title).toBe('Ready for input') + expect(sent[0].body).toBe('Codex is waiting in Demo') + }) + + it('sendReady falls back when the channel has no store (test/legacy wiring)', async () => { + const sent: FcmSendPayload[] = [] + const channel = new FcmNotificationChannel( + { + sendToNamespace: async (_namespace: string, payload: FcmSendPayload) => { + sent.push(payload) + } + } as never, + { sendToast: async () => 0 } as never, + { hasVisibleConnection: () => false } as never + // store omitted on purpose + ) + + await channel.sendReady(createSession()) + + expect(sent[0].title).toBe('Ready for input') + expect(sent[0].body).toBe('Codex is waiting in Demo') + }) + + it('sets severity=info on ready notifications', async () => { + const sent: FcmSendPayload[] = [] + const channel = new FcmNotificationChannel( + { sendToNamespace: async (_n: string, p: FcmSendPayload) => { sent.push(p) } } as never, + { sendToast: async () => 0 } as never, + { hasVisibleConnection: () => false } as never + ) + await channel.sendReady(createSession()) + expect(sent[0].data.severity).toBe('info') + }) + + it('sets nativeGate.sent when FCM delivers at least one message', async () => { + const gate = { sent: false } + const channel = new FcmNotificationChannel( + { + sendToNamespace: async () => ({ sent: 1, failed: 0, invalidTokens: [] }) + } as never, + { sendToast: async () => 0 } as never, + { hasVisibleConnection: () => false } as never + ) + + await channel.sendReady(createSession(), { nativeGate: gate }) + + expect(gate.sent).toBe(true) + }) + + it('leaves nativeGate.sent false when FCM sends zero messages', async () => { + const gate = { sent: false } + const channel = new FcmNotificationChannel( + { + sendToNamespace: async () => ({ sent: 0, failed: 1, invalidTokens: [] }) + } as never, + { sendToast: async () => 0 } as never, + { hasVisibleConnection: () => false } as never + ) + + await channel.sendReady(createSession(), { nativeGate: gate }) + + expect(gate.sent).toBe(false) + }) + + it('sets severity=warning on permission-request notifications', async () => { + const sent: FcmSendPayload[] = [] + const channel = new FcmNotificationChannel( + { sendToNamespace: async (_n: string, p: FcmSendPayload) => { sent.push(p) } } as never, + { sendToast: async () => 0 } as never, + { hasVisibleConnection: () => false } as never + ) + await channel.sendPermissionRequest(createSession({ + agentState: { requests: { 'r-1': { tool: 'Bash', arguments: {} } } } + })) + expect(sent[0].data.severity).toBe('warning') + }) + + it('sets severity=success on completed task notifications', async () => { + const sent: FcmSendPayload[] = [] + const channel = new FcmNotificationChannel( + { sendToNamespace: async (_n: string, p: FcmSendPayload) => { sent.push(p) } } as never, + { sendToast: async () => 0 } as never, + { hasVisibleConnection: () => false } as never + ) + await channel.sendTaskNotification(createSession(), { status: 'completed', summary: 'Tests passed' }) + expect(sent[0].data.severity).toBe('success') + }) + + it('sets severity=error on failed task notifications', async () => { + const sent: FcmSendPayload[] = [] + const channel = new FcmNotificationChannel( + { sendToNamespace: async (_n: string, p: FcmSendPayload) => { sent.push(p) } } as never, + { sendToast: async () => 0 } as never, + { hasVisibleConnection: () => false } as never + ) + // The hub's failure detection catches 'failed' / 'error' / 'killed' / 'aborted'. + for (const status of ['failed', 'error', 'killed', 'aborted']) { + sent.length = 0 + await channel.sendTaskNotification(createSession(), { status, summary: 'oh no' }) + expect(sent[0].data.severity).toBe('error') + } + }) + + it('sendTaskNotification caps long task summaries for FCM data limits', async () => { + const sent: FcmSendPayload[] = [] + const channel = new FcmNotificationChannel( + { sendToNamespace: async (_n: string, p: FcmSendPayload) => { sent.push(p) } } as never, + { sendToast: async () => 0 } as never, + { hasVisibleConnection: () => false } as never + ) + await channel.sendTaskNotification(createSession(), { + status: 'completed', + summary: 'T'.repeat(500) + }) + expect(sent[0].body).toContain('...') + expect(sent[0].body.length).toBeLessThan(350) + }) +}) diff --git a/hub/src/fcm/fcmNotificationChannel.ts b/hub/src/fcm/fcmNotificationChannel.ts new file mode 100644 index 00000000..ba004270 --- /dev/null +++ b/hub/src/fcm/fcmNotificationChannel.ts @@ -0,0 +1,308 @@ +import type { Session } from '../sync/syncEngine' +import type { NotificationChannel, TaskNotification } from '../notifications/notificationTypes' +import type { NotificationSendContext } from '../notifications/notificationSendContext' +import { getAgentName, getSessionName } from '../notifications/sessionInfo' +import { formatToolArgumentsCompact, formatToolArgumentsDetailed } from '../notifications/toolArgs' +import { extractAssistantPlainText, extractNotifySummary, unwrapRoleWrappedRecordEnvelope } from '@hapi/protocol/messages' +import type { Store } from '../store' +import type { SSEManager } from '../sse/sseManager' +import type { VisibilityTracker } from '../visibility/visibilityTracker' +import type { FcmSendPayload, FcmService } from './fcmService' + +const CONTRACT_VERSION = '1' + +/** + * `ready` body content limit for the wrist-glance line on the watch. + * BigTextStyle still expands when the operator taps, so this cap only + * affects the collapsed glance. ~280 chars matches the watch's three-line + * collapsed render at default font scale. + */ +const READY_BODY_GLANCE_LIMIT = 280 + +export class FcmNotificationChannel implements NotificationChannel { + constructor( + private readonly fcmService: FcmService, + private readonly sseManager: SSEManager, + private readonly visibilityTracker: VisibilityTracker, + private readonly store?: Store + ) {} + + async sendPermissionRequest(session: Session, ctx?: NotificationSendContext): Promise { + if (!session.active) { + return + } + + const name = getSessionName(session) + const agentName = getAgentName(session) + const requests = session.agentState?.requests ?? null + const requestEntries = requests ? Object.entries(requests) : [] + const [requestId, request] = requestEntries[0] ?? [undefined, null] + + // Glance line: keep brutally short so the wrist-collapsed + // notification still shows the first ~40 chars without truncation. + // Format: " : " e.g. "Claude Edit: .../hub/server.ts" + // Fallback when args aren't useful: " " e.g. "Cursor Bash" + const toolName = request?.tool ?? '' + const compact = request ? formatToolArgumentsCompact(request.tool, request.arguments) : '' + const glance = toolName + ? (compact ? `${agentName} ${toolName}: ${compact}` : `${agentName} ${toolName}`) + : `${agentName} - ${name}` + + // Detailed body: rendered when the operator taps the notification on + // Wear OS (BigTextStyle on the watch side). Lines after the first + // are hidden in the collapsed glance, so we can be generous here. + const detailed = request + ? formatToolArgumentsDetailed(request.tool, request.arguments, { maxArgLength: 120 }) + : '' + const bodyLines = [glance] + if (name && name !== glance) { + bodyLines.push(`Session: ${name}`) + } + if (detailed) { + bodyLines.push(detailed) + } + + const path = this.buildSessionPath(session.id) + + const payload = this.buildPayload({ + title: 'Permission Request', + body: bodyLines.join('\n'), + tag: `permission-${session.id}`, + type: 'permission-request', + sessionId: session.id, + sessionName: name, + url: path, + requestId, + severity: 'warning' + }) + + await this.deliver(session, payload, ctx) + } + + async sendReady(session: Session, ctx?: NotificationSendContext): Promise { + if (!session.active) { + return + } + + const agentName = getAgentName(session) + const name = getSessionName(session) + const path = this.buildSessionPath(session.id) + + const composed = this.composeReadyBody(session, agentName, name) + + const payload = this.buildPayload({ + title: composed.title, + body: composed.body, + tag: `ready-${session.id}`, + type: 'ready', + sessionId: session.id, + sessionName: name, + url: path, + severity: 'info' + }) + + if (composed.notifySummary) { + payload.data.notifySummary = JSON.stringify(composed.notifySummary) + } + + await this.deliver(session, payload, ctx) + } + + /** + * Build the title/body for a `ready` notification. + * + * Strategy: + * 1. If the operator's `AGENTS.md` has the agent emit + * `AGENT_NOTIFY_SUMMARY {...json...}` as the trailing line, parse it + * and use `summary` (+ `action` on a second line) for the body. + * Title becomes ` - ` so the summary text owns the + * body. + * 2. Otherwise fall back to the first ~280 chars of the most recent + * assistant text. Same title pattern. + * 3. If no assistant text can be found at all (cold start, store + * unavailable, all recent messages are tool calls), fall back to + * the previous " is waiting in " content so we + * never regress to a worse notification than today. + */ + private composeReadyBody( + session: Session, + agentName: string, + sessionName: string + ): { title: string; body: string; notifySummary?: Record } { + const fallback = { + title: 'Ready for input', + body: `${agentName} is waiting in ${sessionName}` + } + + if (!this.store) return fallback + + const lastText = this.findLastAssistantPlainText(session.id) + if (!lastText) return fallback + + const summary = extractNotifySummary(lastText) + const headerTitle = `${agentName} - ${sessionName}` + + if (summary?.summary) { + const summaryLine = this.truncateReadyText(summary.summary, READY_BODY_GLANCE_LIMIT) + const actionLine = summary.action && summary.action !== summary.summary + ? this.truncateReadyText( + `-> ${summary.action}`, + Math.max(0, READY_BODY_GLANCE_LIMIT - summaryLine.length - 1) + ) + : '' + const body = [summaryLine, actionLine].filter(Boolean).join('\n') + const notifySummary = { + ...(typeof summary.version === 'number' ? { version: summary.version } : {}), + summary: summaryLine, + ...(summary.action + ? { action: this.truncateReadyText(summary.action, READY_BODY_GLANCE_LIMIT) } + : {}), + ...(summary.status ? { status: this.truncateReadyText(summary.status, 32) } : {}), + ...(summary.agent ? { agent: this.truncateReadyText(summary.agent, 80) } : {}), + ...(summary.project ? { project: this.truncateReadyText(summary.project, 80) } : {}) + } + return { + title: headerTitle, + body, + notifySummary + } + } + + const trimmed = lastText.trim() + if (trimmed.length === 0) return fallback + + // -3 leaves room for the '...' suffix so the body still fits the + // glance limit. Without this it would tip over by 2 characters. + const body = trimmed.length > READY_BODY_GLANCE_LIMIT + ? trimmed.slice(0, READY_BODY_GLANCE_LIMIT - 3).trimEnd() + '...' + : trimmed + + return { title: headerTitle, body } + } + + private truncateReadyText(text: string, limit: number): string { + const trimmed = text.trim() + if (limit <= 0 || trimmed.length === 0) { + return '' + } + if (trimmed.length <= limit) { + return trimmed + } + if (limit <= 3) { + return '.'.repeat(limit) + } + return trimmed.slice(0, limit - 3).trimEnd() + '...' + } + + /** + * Walk the most recent stored messages and return the plain text of + * the latest assistant message that *has* text content. Tool calls, + * tool results, and reasoning blocks are skipped (they have no + * text body, they would just show as `null` and force the fallback). + */ + private findLastAssistantPlainText(sessionId: string): string | null { + if (!this.store) return null + + let messages + try { + // 20 is generous: most ready events fire 1-3 messages after + // the latest assistant text, and we cap to 20 to avoid + // pathological scans on long sessions. + messages = this.store.messages.getMessages(sessionId, 20) + } catch { + return null + } + + // getMessages returns the LAST `limit` rows in ASCENDING seq order + // (it queries DESC then reverses for caller convenience), so the + // freshest message lives at the END of the array. Walk backwards + // so we hit the latest assistant text first. + for (let i = messages.length - 1; i >= 0; i -= 1) { + const msg = messages[i] + const record = unwrapRoleWrappedRecordEnvelope(msg.content) + if (record?.role !== 'agent') continue + const text = extractAssistantPlainText(record.content) + if (text && text.trim().length > 0) { + return text + } + } + return null + } + + async sendTaskNotification(session: Session, notification: TaskNotification, ctx?: NotificationSendContext): Promise { + if (!session.active) { + return + } + + const agentName = getAgentName(session) + const name = getSessionName(session) + const normalizedStatus = notification.status?.trim().toLowerCase() + const isFailure = normalizedStatus === 'failed' + || normalizedStatus === 'error' + || normalizedStatus === 'killed' + || normalizedStatus === 'aborted' + const path = this.buildSessionPath(session.id) + const taskSummary = this.truncateReadyText(notification.summary, READY_BODY_GLANCE_LIMIT) + + const payload = this.buildPayload({ + title: isFailure ? 'Task failed' : 'Task completed', + body: `${agentName} · ${name} · ${taskSummary}`, + type: 'task-notification', + sessionId: session.id, + sessionName: name, + url: path, + severity: isFailure ? 'error' : 'success' + }) + + await this.deliver(session, payload, ctx) + } + + private buildPayload(input: { + title: string + body: string + tag?: string + type: string + sessionId: string + sessionName: string + url: string + requestId?: string + severity?: 'info' | 'success' | 'warning' | 'error' + }): FcmSendPayload { + return { + title: input.title, + body: input.body, + tag: input.tag, + data: { + type: input.type, + sessionId: input.sessionId, + sessionName: input.sessionName, + url: input.url, + requestId: input.requestId, + title: input.title, + body: input.body, + contractVersion: CONTRACT_VERSION, + severity: input.severity + } + } + } + + private async deliver(session: Session, payload: FcmSendPayload, ctx?: NotificationSendContext): Promise { + // Native companion is the canonical surface: always fire FCM when the + // hub asks us to. The previous SSE-toast shortcut here meant that + // when the operator had the PWA open in foreground, the watch got + // NOTHING - the in-page React toast was the only signal. That broke + // the wrist-first UX (the whole point of installing a watch app) + // and confused the operator about whether the agent was making + // progress. SSE in-page toasts are still emitted by the PWA's own + // SyncEngine event stream for users who want them; this channel's + // job is to reach the wrist, period. + const result = await this.fcmService.sendToNamespace(session.namespace, payload) + if ((result?.sent ?? 0) > 0 && ctx?.nativeGate) { + ctx.nativeGate.sent = true + } + } + + private buildSessionPath(sessionId: string): string { + return `/sessions/${sessionId}` + } +} diff --git a/hub/src/fcm/fcmService.test.ts b/hub/src/fcm/fcmService.test.ts new file mode 100644 index 00000000..d2526a47 --- /dev/null +++ b/hub/src/fcm/fcmService.test.ts @@ -0,0 +1,440 @@ +import { afterEach, beforeEach, describe, expect, it, mock } from 'bun:test' +import { FcmService, type FcmSendPayload } from './fcmService' + +mock.module('./fcmAuth', () => ({ + getFcmAccessToken: async () => 'test-access-token', + loadServiceAccount: () => ({ client_email: 'x', private_key: 'y' }) +})) + +type FakeStore = { + fcm: { + getDevicesByNamespace: ReturnType + removeDeviceByToken: ReturnType + } +} + +function makeStore(devices: Array<{ token: string; platform: 'phone' | 'wear'; deviceId: string; namespace: string }>): FakeStore { + return { + fcm: { + getDevicesByNamespace: mock((ns: string) => + devices + .filter(d => d.namespace === ns) + .map(d => ({ + id: 0, + namespace: d.namespace, + token: d.token, + platform: d.platform, + deviceId: d.deviceId, + createdAt: 0, + updatedAt: 0 + })) + ), + removeDeviceByToken: mock(() => {}) + } + } +} + +function makePayload(overrides: Partial = {}): FcmSendPayload { + return { + title: 'T', + body: 'B', + data: { + type: 'ready', + sessionId: 'sess-1', + sessionName: 'Demo', + url: 'https://hapi.example.com/sessions/sess-1', + title: 'T', + body: 'B', + contractVersion: '1', + ...overrides + } + } +} + +describe('FcmService.sendToNamespace', () => { + let originalFetch: typeof globalThis.fetch + beforeEach(() => { + originalFetch = globalThis.fetch + }) + afterEach(() => { + globalThis.fetch = originalFetch + }) + + it('removes the device row when FCM returns 404 UNREGISTERED (token rotated)', async () => { + const store = makeStore([ + { namespace: 'default', token: 'rotated-token', platform: 'phone', deviceId: 'p1' } + ]) + globalThis.fetch = mock(async () => + new Response('{"error":{"status":"UNREGISTERED"}}', { status: 404 }) + ) as unknown as typeof fetch + + const svc = new FcmService('proj-id', { client_email: 'x', private_key: 'y' }, store as never) + const result = await svc.sendToNamespace('default', makePayload()) + + expect(result.sent).toBe(0) + expect(result.failed).toBe(1) + expect(result.invalidTokens).toEqual(['rotated-token']) + expect(store.fcm.removeDeviceByToken).toHaveBeenCalledWith('default', 'rotated-token') + }) + + it('keeps the device row on generic 404 NOT_FOUND (bad project/resource config)', async () => { + const store = makeStore([ + { namespace: 'default', token: 'live-token', platform: 'phone', deviceId: 'p1' } + ]) + globalThis.fetch = mock(async () => + new Response('{"error":{"status":"NOT_FOUND","message":"Requested entity was not found."}}', { status: 404 }) + ) as unknown as typeof fetch + + const svc = new FcmService('proj-id', { client_email: 'x', private_key: 'y' }, store as never) + const result = await svc.sendToNamespace('default', makePayload()) + + expect(result.failed).toBe(1) + expect(result.invalidTokens).toEqual([]) + expect(store.fcm.removeDeviceByToken).not.toHaveBeenCalled() + }) + + it('removes the device row on canonical 404 NOT_FOUND + FcmError UNREGISTERED', async () => { + const store = makeStore([ + { namespace: 'default', token: 'dead-token', platform: 'phone', deviceId: 'p1' } + ]) + globalThis.fetch = mock(async () => + new Response(JSON.stringify({ + error: { + status: 'NOT_FOUND', + details: [{ + '@type': 'type.googleapis.com/google.firebase.fcm.v1.FcmError', + errorCode: 'UNREGISTERED' + }] + } + }), { status: 404 }) + ) as unknown as typeof fetch + + const svc = new FcmService('proj-id', { client_email: 'x', private_key: 'y' }, store as never) + const result = await svc.sendToNamespace('default', makePayload()) + + expect(result.invalidTokens).toEqual(['dead-token']) + expect(store.fcm.removeDeviceByToken).toHaveBeenCalledWith('default', 'dead-token') + }) + + it('keeps the device row on 400 INVALID_ARGUMENT without token field violation', async () => { + const store = makeStore([ + { namespace: 'default', token: 'live-token', platform: 'phone', deviceId: 'p1' } + ]) + globalThis.fetch = mock(async () => + new Response(JSON.stringify({ + error: { + status: 'INVALID_ARGUMENT', + details: [{ + fieldViolations: [{ field: 'message.data.body', description: 'too long' }] + }] + } + }), { status: 400 }) + ) as unknown as typeof fetch + + const svc = new FcmService('proj-id', { client_email: 'x', private_key: 'y' }, store as never) + const result = await svc.sendToNamespace('default', makePayload()) + + expect(result.failed).toBe(1) + expect(store.fcm.removeDeviceByToken).not.toHaveBeenCalled() + }) + + it('keeps the device row on transient 429 (rate limit) - regression for HAPI Bot finding', async () => { + const store = makeStore([ + { namespace: 'default', token: 'rate-limited-token', platform: 'phone', deviceId: 'p1' } + ]) + globalThis.fetch = mock(async () => + new Response('{"error":{"status":"RESOURCE_EXHAUSTED"}}', { status: 429 }) + ) as unknown as typeof fetch + + const svc = new FcmService('proj-id', { client_email: 'x', private_key: 'y' }, store as never) + const result = await svc.sendToNamespace('default', makePayload()) + + expect(result.sent).toBe(0) + expect(result.failed).toBe(1) + expect(result.invalidTokens).toEqual([]) + // Critical: must NOT remove the device on a transient failure. + expect(store.fcm.removeDeviceByToken).not.toHaveBeenCalled() + }) + + it('keeps the device row on transient 503 (server error)', async () => { + const store = makeStore([ + { namespace: 'default', token: 'live-token', platform: 'wear', deviceId: 'w1' } + ]) + globalThis.fetch = mock(async () => + new Response('Service Unavailable', { status: 503 }) + ) as unknown as typeof fetch + + const svc = new FcmService('proj-id', { client_email: 'x', private_key: 'y' }, store as never) + const result = await svc.sendToNamespace('default', makePayload()) + + expect(result.failed).toBe(1) + expect(result.invalidTokens).toEqual([]) + expect(store.fcm.removeDeviceByToken).not.toHaveBeenCalled() + }) + + it('keeps the device row on 401 auth glitch (our problem, not the device\'s)', async () => { + const store = makeStore([ + { namespace: 'default', token: 'live-token', platform: 'phone', deviceId: 'p1' } + ]) + globalThis.fetch = mock(async () => + new Response('{"error":{"status":"UNAUTHENTICATED"}}', { status: 401 }) + ) as unknown as typeof fetch + + const svc = new FcmService('proj-id', { client_email: 'x', private_key: 'y' }, store as never) + const result = await svc.sendToNamespace('default', makePayload()) + + expect(result.failed).toBe(1) + expect(store.fcm.removeDeviceByToken).not.toHaveBeenCalled() + }) + + it('keeps the device row when fetch itself throws (network error)', async () => { + const store = makeStore([ + { namespace: 'default', token: 'live-token', platform: 'phone', deviceId: 'p1' } + ]) + globalThis.fetch = mock(async () => { + throw new Error('ECONNREFUSED') + }) as unknown as typeof fetch + + const svc = new FcmService('proj-id', { client_email: 'x', private_key: 'y' }, store as never) + const result = await svc.sendToNamespace('default', makePayload()) + + expect(result.failed).toBe(1) + expect(store.fcm.removeDeviceByToken).not.toHaveBeenCalled() + }) + + it('treats timed-out FCM send as transient failure', async () => { + const store = makeStore([ + { namespace: 'default', token: 'live-token', platform: 'phone', deviceId: 'p1' } + ]) + globalThis.fetch = mock(async (_url, init) => { + expect(init?.signal).toBeDefined() + throw new DOMException('The operation was aborted.', 'AbortError') + }) as unknown as typeof fetch + + const svc = new FcmService('proj-id', { client_email: 'x', private_key: 'y' }, store as never) + const result = await svc.sendToNamespace('default', makePayload()) + + expect(result.failed).toBe(1) + expect(store.fcm.removeDeviceByToken).not.toHaveBeenCalled() + }) + + it('counts a 200 response as sent', async () => { + const store = makeStore([ + { namespace: 'default', token: 'live-token', platform: 'phone', deviceId: 'p1' } + ]) + globalThis.fetch = mock(async () => + new Response('{"name":"projects/proj-id/messages/0:1234567890"}', { status: 200 }) + ) as unknown as typeof fetch + + const svc = new FcmService('proj-id', { client_email: 'x', private_key: 'y' }, store as never) + const result = await svc.sendToNamespace('default', makePayload()) + + expect(result.sent).toBe(1) + expect(result.failed).toBe(0) + expect(store.fcm.removeDeviceByToken).not.toHaveBeenCalled() + }) + + it('mixed batch: removes invalid token, keeps device with transient failure, counts good send', async () => { + const store = makeStore([ + { namespace: 'default', token: 'good-token', platform: 'phone', deviceId: 'p1' }, + { namespace: 'default', token: 'rotated-token', platform: 'phone', deviceId: 'p2' }, + { namespace: 'default', token: 'rate-limited-token', platform: 'wear', deviceId: 'w1' } + ]) + + const responseFor: Record Response> = { + 'good-token': () => new Response('{"name":"ok"}', { status: 200 }), + 'rotated-token': () => new Response('{"error":{"status":"UNREGISTERED"}}', { status: 404 }), + 'rate-limited-token': () => new Response('{"error":{"status":"RESOURCE_EXHAUSTED"}}', { status: 429 }) + } + globalThis.fetch = mock(async (_url: unknown, init?: RequestInit) => { + const body = JSON.parse((init?.body as string) ?? '{}') as { message?: { token?: string } } + const token = body.message?.token ?? '' + const fn = responseFor[token] + return fn ? fn() : new Response('unknown', { status: 500 }) + }) as unknown as typeof fetch + + const svc = new FcmService('proj-id', { client_email: 'x', private_key: 'y' }, store as never) + const result = await svc.sendToNamespace('default', makePayload()) + + expect(result.sent).toBe(1) + expect(result.failed).toBe(2) + expect(result.invalidTokens).toEqual(['rotated-token']) + // Only the truly-rotated token gets unregistered. The rate-limited + // device must survive to be retried on the next notification. + expect(store.fcm.removeDeviceByToken).toHaveBeenCalledTimes(1) + expect(store.fcm.removeDeviceByToken).toHaveBeenCalledWith('default', 'rotated-token') + }) + + it('returns zero counts when namespace has no devices', async () => { + const store = makeStore([]) + globalThis.fetch = mock(async () => new Response('should-not-be-called', { status: 200 })) as unknown as typeof fetch + + const svc = new FcmService('proj-id', { client_email: 'x', private_key: 'y' }, store as never) + const result = await svc.sendToNamespace('empty-ns', makePayload()) + + expect(result).toEqual({ sent: 0, failed: 0, invalidTokens: [] }) + expect(globalThis.fetch).not.toHaveBeenCalled() + }) +}) + +describe('FcmService.isHealthy (rolling outcome window)', () => { + let originalFetch: typeof globalThis.fetch + beforeEach(() => { + originalFetch = globalThis.fetch + }) + afterEach(() => { + globalThis.fetch = originalFetch + }) + + it('starts UNHEALTHY with an empty outcome buffer (no positive evidence yet)', () => { + // Cold-start invariant (HAPI Bot Major fix on PR #803): the gate + // requires at least one observed success before suppressing + // web-push. Otherwise a hub started with broken FCM credentials + // silently drops the first N notifications while waiting for the + // failure threshold to trip. + const store = makeStore([]) + const svc = new FcmService('proj-id', { client_email: 'x', private_key: 'y' }, store as never) + expect(svc.isHealthy()).toBe(false) + }) + + it('flips to healthy after the first successful send', async () => { + const store = makeStore([ + { namespace: 'default', token: 't1', platform: 'phone', deviceId: 'p1' } + ]) + globalThis.fetch = mock(async () => + new Response('{"name":"ok"}', { status: 200 }) + ) as unknown as typeof fetch + + const svc = new FcmService('proj-id', { client_email: 'x', private_key: 'y' }, store as never) + expect(svc.isHealthy()).toBe(false) + await svc.sendToNamespace('default', makePayload()) + expect(svc.isHealthy()).toBe(true) + }) + + it('stays unhealthy across a run of failures with no successes (broken-FCM cold start)', async () => { + const store = makeStore([ + { namespace: 'default', token: 't1', platform: 'phone', deviceId: 'p1' } + ]) + globalThis.fetch = mock(async () => + new Response('Service Unavailable', { status: 503 }) + ) as unknown as typeof fetch + + const svc = new FcmService('proj-id', { client_email: 'x', private_key: 'y' }, store as never) + + // Without any prior success the gate must stay unhealthy regardless + // of where we are in the failure-threshold count. This is the exact + // silent-blackhole window the bot flagged. + for (let i = 0; i < 5; i += 1) { + await svc.sendToNamespace('default', makePayload()) + expect(svc.isHealthy()).toBe(false) + } + }) + + it('flips back to unhealthy when failures stack past threshold after prior successes', async () => { + const store = makeStore([ + { namespace: 'default', token: 't1', platform: 'phone', deviceId: 'p1' } + ]) + let callCount = 0 + globalThis.fetch = mock(async () => { + callCount += 1 + // First 3 succeed, then 503s + if (callCount <= 3) return new Response('{"name":"ok"}', { status: 200 }) + return new Response('Service Unavailable', { status: 503 }) + }) as unknown as typeof fetch + + const svc = new FcmService('proj-id', { client_email: 'x', private_key: 'y' }, store as never) + + // 3 successes establish health + for (let i = 0; i < 3; i += 1) await svc.sendToNamespace('default', makePayload()) + expect(svc.isHealthy()).toBe(true) + + // 4 failures: window is [S,S,S,F,F,F,F] - 4 < 5 -> still healthy + for (let i = 0; i < 4; i += 1) await svc.sendToNamespace('default', makePayload()) + expect(svc.isHealthy()).toBe(true) + + // 5th failure: [S,S,S,F,F,F,F,F] - 5 >= 5 -> unhealthy + await svc.sendToNamespace('default', makePayload()) + expect(svc.isHealthy()).toBe(false) + }) + + it('recovers to healthy as recent successes age out the failure tail', async () => { + const store = makeStore([ + { namespace: 'default', token: 't1', platform: 'phone', deviceId: 'p1' } + ]) + let callCount = 0 + globalThis.fetch = mock(async () => { + callCount += 1 + // First 5 calls fail (503), rest succeed + if (callCount <= 5) { + return new Response('Service Unavailable', { status: 503 }) + } + return new Response('{"name":"ok"}', { status: 200 }) + }) as unknown as typeof fetch + + const svc = new FcmService('proj-id', { client_email: 'x', private_key: 'y' }, store as never) + + for (let i = 0; i < 5; i += 1) { + await svc.sendToNamespace('default', makePayload()) + } + expect(svc.isHealthy()).toBe(false) + + // 4 successes after 5 failures: window is [F,F,F,F,F,S,S,S,S] -> trim + // to last 8: [F,F,F,F,S,S,S,S] -> 4 failures, threshold 5 -> healthy. + for (let i = 0; i < 4; i += 1) { + await svc.sendToNamespace('default', makePayload()) + } + expect(svc.isHealthy()).toBe(true) + }) + + it('does NOT count invalid-token responses against health (per-device fact, not pipeline failure)', async () => { + const store = makeStore([ + { namespace: 'default', token: 'good', platform: 'phone', deviceId: 'p1' }, + { namespace: 'default', token: 'rotated', platform: 'phone', deviceId: 'p2' } + ]) + globalThis.fetch = mock(async (url: unknown, init?: unknown) => { + // Different responses per device token. We use the request + // body to discriminate - both calls go to the same URL. + const body = JSON.parse(((init as { body?: string })?.body) ?? '{}') + const token = body?.message?.token + if (token === 'good') return new Response('{"name":"ok"}', { status: 200 }) + return new Response('{"error":{"status":"UNREGISTERED"}}', { status: 404 }) + }) as unknown as typeof fetch + + const svc = new FcmService('proj-id', { client_email: 'x', private_key: 'y' }, store as never) + + // First send produces 1 sent + 1 invalid. After this the rotated + // token is removed from the store, leaving only the good one. + await svc.sendToNamespace('default', makePayload()) + expect(svc.isHealthy()).toBe(true) + + // Subsequent successful sends do not record additional outcomes + // for the (now-pruned) invalid token. Health stays true. + for (let i = 0; i < 10; i += 1) { + await svc.sendToNamespace('default', makePayload()) + } + expect(svc.isHealthy()).toBe(true) + }) + + it('counts fetch-throw (network error) as a health failure', async () => { + const store = makeStore([ + { namespace: 'default', token: 't1', platform: 'phone', deviceId: 'p1' } + ]) + let callCount = 0 + globalThis.fetch = mock(async () => { + callCount += 1 + // First few succeed (establish health), rest throw network error + if (callCount <= 3) return new Response('{"name":"ok"}', { status: 200 }) + throw new Error('ECONNREFUSED') + }) as unknown as typeof fetch + + const svc = new FcmService('proj-id', { client_email: 'x', private_key: 'y' }, store as never) + + // Establish health with 3 successes + for (let i = 0; i < 3; i += 1) await svc.sendToNamespace('default', makePayload()) + expect(svc.isHealthy()).toBe(true) + + // 5 network errors stack past threshold and flip health + for (let i = 0; i < 5; i += 1) await svc.sendToNamespace('default', makePayload()) + expect(svc.isHealthy()).toBe(false) + }) +}) diff --git a/hub/src/fcm/fcmService.ts b/hub/src/fcm/fcmService.ts new file mode 100644 index 00000000..fb0cf925 --- /dev/null +++ b/hub/src/fcm/fcmService.ts @@ -0,0 +1,277 @@ +import type { Store } from '../store' +import { getFcmAccessToken, FCM_REQUEST_TIMEOUT_MS, type ServiceAccount } from './fcmAuth' + +export type FcmDataPayload = { + type: string + sessionId: string + sessionName: string + url: string + requestId?: string + title: string + body: string + contractVersion: string + /** + * Visual urgency hint for the client. Drives the notification accent + * color on Wear OS / phone, and may be used for sound channel routing + * later. Independent of `type` because `task-notification` is one + * type that splits across success and failure outcomes. + * + * - `info` ready / ambient ('no action needed') -> blue + * - `success` task completed -> green + * - `warning` permission request -> amber + * - `error` task failed / aborted -> red + */ + severity?: 'info' | 'success' | 'warning' | 'error' + /** + * JSON-stringified `AGENT_NOTIFY_SUMMARY` object when the agent emitted + * one as the trailing line of its last message. The companion app may + * use this for richer rendering or future event-bus routing; absent + * when the agent did not emit a summary. + */ + notifySummary?: string +} + +export type FcmSendPayload = { + title: string + body: string + tag?: string + data: FcmDataPayload +} + +type FcmSendResult = { + sent: number + failed: number + invalidTokens: string[] +} + +/** + * Outcome of a single FCM send. We split `failed` into: + * - `invalid`: token is dead and will never succeed (uninstall, rotation, + * malformed). Safe to remove from the device registry. + * - `failed`: transient or out-of-band error (rate limit, 5xx, auth + * glitch). MUST NOT be treated as token death - we'd silently + * unregister live devices on every Google blip. + */ +type FcmTokenSendResult = 'sent' | 'invalid' | 'failed' + +export class FcmService { + /** + * Rolling window of the last N send outcomes. Drives `isHealthy()`, + * which the native-fallback probe consults to decide whether suppressing + * web-push for this namespace is still safe. We deliberately do NOT + * count `invalid` here - an invalid token is a per-device fact, not an + * FCM-pipeline-broken signal (FCM was reachable, it just rejected one + * stale token). Only `sent` and `failed` populate the buffer. + */ + private recentOutcomes: Array<'sent' | 'failed'> = [] + private static readonly HEALTH_WINDOW = 8 + private static readonly HEALTH_FAILURE_THRESHOLD = 5 + + constructor( + private readonly projectId: string, + private readonly serviceAccount: ServiceAccount, + private readonly store: Store + ) {} + + /** + * Health gate for the native-fallback probe. Returns true only when the + * recent-outcome window contains at least one positive datapoint AND + * failures have not stacked past the threshold. When unhealthy, the + * probe lets web-push fire as a last-resort surface for this namespace. + * + * "Needs positive evidence" semantics intentionally: an empty buffer + * (cold-start) and a buffer dominated by failures-only both render + * unhealthy. This closes the silent-blackhole window where a hub with + * broken Firebase credentials would suppress web-push for the first N + * events while waiting for failures to accumulate past the threshold. + * + * Trade-off: one duplicated notification per hub restart per namespace + * (web-push + FCM both fire on event #1; FCM success records `sent` and + * the gate engages from event #2 onward). Worth it for guaranteed + * delivery on cold start. + * + * Addresses HAPI Bot Major review on PR #803. + */ + isHealthy(): boolean { + const successes = this.recentOutcomes.filter((o) => o === 'sent').length + if (successes === 0) return false + const failures = this.recentOutcomes.filter((o) => o === 'failed').length + return failures < FcmService.HEALTH_FAILURE_THRESHOLD + } + + private recordOutcome(outcome: 'sent' | 'failed'): void { + this.recentOutcomes.push(outcome) + if (this.recentOutcomes.length > FcmService.HEALTH_WINDOW) { + this.recentOutcomes.shift() + } + } + + async sendToNamespace(namespace: string, payload: FcmSendPayload): Promise { + const devices = this.store.fcm.getDevicesByNamespace(namespace) + if (devices.length === 0) { + return { sent: 0, failed: 0, invalidTokens: [] } + } + + let accessToken: string + try { + accessToken = await getFcmAccessToken(this.serviceAccount) + } catch (e) { + // Token-fetch failure (expired service account key, OAuth + // outage, network) - count one health-failure (not one per + // device, that would over-weight the buffer) and return. + console.error('[FcmService] Token fetch failed:', e instanceof Error ? e.message : e) + this.recordOutcome('failed') + return { sent: 0, failed: devices.length, invalidTokens: [] } + } + + const invalidTokens: string[] = [] + let sent = 0 + let failed = 0 + + await Promise.all(devices.map(async (device) => { + const result = await this.sendToToken(accessToken, device.token, payload, device.platform) + // `invalid` is a per-device fact, not a pipeline signal - + // exclude it from the health buffer (see field doc above). + if (result === 'sent') { + this.recordOutcome('sent') + } else if (result === 'failed') { + this.recordOutcome('failed') + } + if (result === 'sent') { + sent += 1 + return + } + failed += 1 + if (result === 'invalid') { + invalidTokens.push(device.token) + this.store.fcm.removeDeviceByToken(namespace, device.token) + } + })) + + return { sent, failed, invalidTokens } + } + + private async sendToToken( + accessToken: string, + token: string, + payload: FcmSendPayload, + platform: 'phone' | 'wear' + ): Promise { + const url = `https://fcm.googleapis.com/v1/projects/${this.projectId}/messages:send` + const dataRecord: Record = { + type: payload.data.type, + sessionId: payload.data.sessionId, + sessionName: payload.data.sessionName, + url: payload.data.url, + title: payload.data.title, + body: payload.data.body, + contractVersion: payload.data.contractVersion + } + if (payload.data.requestId) { + dataRecord.requestId = payload.data.requestId + } + if (payload.data.severity) { + dataRecord.severity = payload.data.severity + } + if (payload.data.notifySummary) { + dataRecord.notifySummary = payload.data.notifySummary + } + + // Data-only: if we also send `notification`, Android does not call + // onMessageReceived while backgrounded — Wear relay never runs. + const message: Record = { + token, + data: dataRecord, + android: { + priority: 'HIGH' + } + } + + if (platform === 'wear') { + message.android = { + ...(message.android as Record), + direct_boot_ok: true + } + } + + let response: Response + try { + response = await fetch(url, { + method: 'POST', + headers: { + authorization: `Bearer ${accessToken}`, + 'content-type': 'application/json' + }, + body: JSON.stringify({ message }), + signal: AbortSignal.timeout(FCM_REQUEST_TIMEOUT_MS) + }) + } catch (e) { + // Network error (DNS, TCP, TLS) - transient, never a token-death signal. + console.error('[FcmService] Send threw:', e instanceof Error ? e.message : e) + return 'failed' + } + + if (response.ok) { + return 'sent' + } + + const body = await response.text().catch(() => '') + const invalid = this.isInvalidFcmTokenResponse(response.status, body) + if (!invalid) { + console.error('[FcmService] Send failed (transient):', response.status, body.slice(0, 200)) + } + return invalid ? 'invalid' : 'failed' + } + + /** + * Parse FCM v1 error JSON and decide whether the token itself is dead. + * Generic 404/NOT_FOUND (bad project id, missing resource) must not + * unregister live devices — only explicit UNREGISTERED or token-field + * INVALID_ARGUMENT qualifies. + */ + private isInvalidFcmTokenResponse(status: number, body: string): boolean { + type FcmErrorDetail = { + '@type'?: string + errorCode?: string + fieldViolations?: Array<{ field?: string }> + } + + const parsedError = ((): { + error?: { + status?: string + details?: FcmErrorDetail[] + } + } | null => { + try { + return JSON.parse(body) as { + error?: { + status?: string + details?: FcmErrorDetail[] + } + } + } catch { + return null + } + })() + + const errorStatus = parsedError?.error?.status ?? '' + const details = parsedError?.error?.details ?? [] + const fcmErrorCode = details.find((detail) => + detail['@type'] === 'type.googleapis.com/google.firebase.fcm.v1.FcmError' + )?.errorCode ?? '' + const tokenFieldViolation = details.some((detail) => + detail.fieldViolations?.some((violation) => + /message\.token|token/i.test(violation.field ?? '') + ) + ) + + const isUnregistered = status === 404 && ( + fcmErrorCode === 'UNREGISTERED' || errorStatus === 'UNREGISTERED' + ) + const isMalformedToken = status === 400 + && errorStatus === 'INVALID_ARGUMENT' + && (fcmErrorCode === 'INVALID_ARGUMENT' || tokenFieldViolation) + + return isUnregistered || isMalformedToken + } +} diff --git a/hub/src/notifications/notificationHub.ts b/hub/src/notifications/notificationHub.ts index bfe109ab..f9d627e1 100644 --- a/hub/src/notifications/notificationHub.ts +++ b/hub/src/notifications/notificationHub.ts @@ -1,6 +1,7 @@ import type { Session, SyncEngine, SyncEvent } from '../sync/syncEngine' import type { SessionEndReason } from '@hapi/protocol' import type { NotificationChannel, NotificationHubOptions, TaskNotification } from './notificationTypes' +import type { NotificationSendContext } from './notificationSendContext' import { extractMessageEventType, extractTaskNotification } from './eventParsing' export class NotificationHub { @@ -182,9 +183,10 @@ export class NotificationHub { } private async notifyReady(session: Session): Promise { + const ctx: NotificationSendContext = { nativeGate: { sent: false } } for (const channel of this.channels) { try { - await channel.sendReady(session) + await channel.sendReady(session, ctx) } catch (error) { console.error('[NotificationHub] Failed to send ready notification:', error) } @@ -192,9 +194,10 @@ export class NotificationHub { } private async notifyPermission(session: Session): Promise { + const ctx: NotificationSendContext = { nativeGate: { sent: false } } for (const channel of this.channels) { try { - await channel.sendPermissionRequest(session) + await channel.sendPermissionRequest(session, ctx) } catch (error) { console.error('[NotificationHub] Failed to send permission notification:', error) } @@ -202,9 +205,10 @@ export class NotificationHub { } private async notifyTask(session: Session, notification: TaskNotification): Promise { + const ctx: NotificationSendContext = { nativeGate: { sent: false } } for (const channel of this.channels) { try { - await channel.sendTaskNotification(session, notification) + await channel.sendTaskNotification(session, notification, ctx) } catch (error) { console.error('[NotificationHub] Failed to send task notification:', error) } diff --git a/hub/src/notifications/notificationSendContext.ts b/hub/src/notifications/notificationSendContext.ts new file mode 100644 index 00000000..baef669a --- /dev/null +++ b/hub/src/notifications/notificationSendContext.ts @@ -0,0 +1,14 @@ +/** + * Per-notification dispatch context shared across channels in one + * NotificationHub notify* call. FcmNotificationChannel runs first and + * sets `nativeGate.sent` when FCM actually delivers; PushNotificationChannel + * consults the same gate before suppressing web-push/SSE (never on stale + * registration/health probes alone). + */ +export type NativeDeliveryGate = { + sent: boolean +} + +export type NotificationSendContext = { + nativeGate?: NativeDeliveryGate +} diff --git a/hub/src/notifications/notificationTypes.ts b/hub/src/notifications/notificationTypes.ts index 07e2b642..6130cf30 100644 --- a/hub/src/notifications/notificationTypes.ts +++ b/hub/src/notifications/notificationTypes.ts @@ -1,5 +1,6 @@ import type { Session } from '../sync/syncEngine' import type { SessionEndReason } from '@hapi/protocol' +import type { NotificationSendContext } from './notificationSendContext' export type TaskNotification = { summary: string @@ -7,9 +8,9 @@ export type TaskNotification = { } export type NotificationChannel = { - sendReady: (session: Session) => Promise - sendPermissionRequest: (session: Session) => Promise - sendTaskNotification: (session: Session, notification: TaskNotification) => Promise + sendReady: (session: Session, ctx?: NotificationSendContext) => Promise + sendPermissionRequest: (session: Session, ctx?: NotificationSendContext) => Promise + sendTaskNotification: (session: Session, notification: TaskNotification, ctx?: NotificationSendContext) => Promise sendSessionCompletion?: (session: Session, reason: SessionEndReason) => Promise } diff --git a/hub/src/notifications/toolArgs.ts b/hub/src/notifications/toolArgs.ts new file mode 100644 index 00000000..a8092dc6 --- /dev/null +++ b/hub/src/notifications/toolArgs.ts @@ -0,0 +1,157 @@ +/** + * Tool argument formatters shared across notification channels. + * + * Originally lived inside hub/src/telegram/sessionView.ts. Lifted into + * notifications/ so the FCM (Wear OS) channel can reuse the same + * tool-aware extraction without forking the switch table - keeping + * Telegram and Wear notifications in sync as we add tools. + * + * Two surfaces are exposed: + * + * - `formatToolArgumentsDetailed` - multi-line, Telegram-grade detail. + * Renders inside Telegram bot messages where vertical space is cheap. + * Also rendered by Wear OS when the operator taps the notification + * to expand (BigTextStyle). + * + * - `formatToolArgumentsCompact` - single-line, glance-friendly. + * Squeezed into the wrist's collapsed notification line (~40 chars + * before truncation). The detailed form is the source of truth; the + * compact form is a deliberately-brutal summary of just enough to + * know which file/cmd/url is at stake. + */ + +const DEFAULT_DETAIL_MAX_ARG_LENGTH = 150 + +function truncate(text: string, maxLen: number): string { + if (!text) return '' + if (text.length <= maxLen) return text + return text.slice(0, Math.max(0, maxLen - 3)) + '...' +} + +function shortPath(file: string): string { + if (!file) return '' + const segs = file.split('/') + if (segs.length <= 2) return file + return `.../${segs.slice(-2).join('/')}` +} + +export function formatToolArgumentsDetailed( + tool: string, + args: unknown, + opts: { maxArgLength?: number } = {} +): string { + if (!args || typeof args !== 'object') return '' + const maxLen = opts.maxArgLength ?? DEFAULT_DETAIL_MAX_ARG_LENGTH + const a = args as Record + + try { + switch (tool) { + case 'Edit': { + const file = (a.file_path as string | undefined) ?? (a.path as string | undefined) ?? 'unknown' + const oldStr = a.old_string ? truncate(String(a.old_string), 50) : '' + const newStr = a.new_string ? truncate(String(a.new_string), 50) : '' + let result = `File: ${truncate(file, maxLen)}` + if (oldStr) result += `\nOld: "${oldStr}"` + if (newStr) result += `\nNew: "${newStr}"` + return result + } + case 'Write': { + const file = (a.file_path as string | undefined) ?? (a.path as string | undefined) ?? 'unknown' + const content = a.content ? `${String(a.content).length} chars` : '' + return `File: ${truncate(file, maxLen)}${content ? ` (${content})` : ''}` + } + case 'Read': { + const file = (a.file_path as string | undefined) ?? (a.path as string | undefined) ?? 'unknown' + return `File: ${truncate(file, maxLen)}` + } + case 'Bash': { + const cmd = (a.command as string | undefined) ?? '' + return `Command: ${truncate(cmd, maxLen)}` + } + case 'Agent': + case 'Task': { + const desc = (a.description as string | undefined) ?? (a.prompt as string | undefined) ?? '' + return `Task: ${truncate(desc, maxLen)}` + } + case 'Grep': + case 'Glob': { + const pattern = (a.pattern as string | undefined) ?? '' + const path = (a.path as string | undefined) ?? '' + let result = `Pattern: ${truncate(pattern, maxLen)}` + if (path) result += `\nPath: ${truncate(path, 80)}` + return result + } + case 'WebFetch': { + const url = (a.url as string | undefined) ?? '' + return `URL: ${truncate(url, maxLen)}` + } + case 'TodoWrite': { + const todos = a.todos as unknown[] | undefined + const count = todos?.length ?? 0 + return `Updating ${count} todo items` + } + default: { + const argStr = JSON.stringify(args) + if (argStr && argStr.length > 10) { + return `Args: ${truncate(argStr, maxLen)}` + } + return '' + } + } + } catch { + return '' + } +} + +/** + * Single-line summary tuned for the Wear OS collapsed notification line + * (~40 chars displayable before the system truncates). Always returns + * a one-liner with no embedded newlines. Empty string means we have no + * useful summary to show beyond the tool name itself. + */ +export function formatToolArgumentsCompact(tool: string, args: unknown): string { + if (!args || typeof args !== 'object') return '' + const a = args as Record + + try { + switch (tool) { + case 'Edit': + case 'Write': + case 'Read': { + const file = (a.file_path as string | undefined) ?? (a.path as string | undefined) + if (!file) return '' + return shortPath(file) + } + case 'Bash': { + const cmd = (a.command as string | undefined) ?? '' + return truncate(cmd, 60) + } + case 'Agent': + case 'Task': { + const desc = (a.description as string | undefined) ?? (a.prompt as string | undefined) ?? '' + return truncate(desc, 60) + } + case 'Grep': + case 'Glob': { + const pattern = (a.pattern as string | undefined) ?? '' + return truncate(pattern, 60) + } + case 'WebFetch': { + const url = (a.url as string | undefined) ?? '' + try { + if (url) return new URL(url).host + } catch { /* fall through */ } + return truncate(url, 60) + } + case 'TodoWrite': { + const todos = a.todos as unknown[] | undefined + const count = todos?.length ?? 0 + return `${count} items` + } + default: + return '' + } + } catch { + return '' + } +} diff --git a/hub/src/push/pushNotificationChannel.test.ts b/hub/src/push/pushNotificationChannel.test.ts index 8ba04c7f..bd0d0c33 100644 --- a/hub/src/push/pushNotificationChannel.test.ts +++ b/hub/src/push/pushNotificationChannel.test.ts @@ -75,4 +75,120 @@ describe('PushNotificationChannel', () => { expect(pushed[0].payload.tag).toBeUndefined() expect(pushed[1].payload.tag).toBeUndefined() }) + + it('skips web-push when native FCM delivered in the same dispatch', async () => { + const pushed: Array<{ namespace: string; payload: PushPayload }> = [] + const channel = new PushNotificationChannel( + { + sendToNamespace: async (namespace: string, payload: PushPayload) => { + pushed.push({ namespace, payload }) + } + } as never, + { + sendToast: async () => 0 + } as never, + { + hasVisibleConnection: () => false + } as never, + '' + ) + + const ctx = { nativeGate: { sent: true } } + + await channel.sendPermissionRequest(createSession({ + agentState: { + requests: { 'req-1': { tool: 'Bash', arguments: {} } } + } + }), ctx) + await channel.sendReady(createSession(), ctx) + await channel.sendTaskNotification(createSession(), { + status: 'completed', + summary: 'Done' + }, ctx) + + expect(pushed).toHaveLength(0) + }) + + it('falls back to web-push when native gate is unset (FCM failed or absent)', async () => { + const pushed: Array<{ namespace: string; payload: PushPayload }> = [] + const channel = new PushNotificationChannel( + { + sendToNamespace: async (namespace: string, payload: PushPayload) => { + pushed.push({ namespace, payload }) + } + } as never, + { + sendToast: async () => 0 + } as never, + { + hasVisibleConnection: () => false + } as never, + '' + ) + + await channel.sendReady(createSession(), { nativeGate: { sent: false } }) + + expect(pushed).toHaveLength(1) + }) + + it('still sends web-push when no native gate is provided', async () => { + const pushed: Array<{ namespace: string; payload: PushPayload }> = [] + const channel = new PushNotificationChannel( + { + sendToNamespace: async (namespace: string, payload: PushPayload) => { + pushed.push({ namespace, payload }) + } + } as never, + { + sendToast: async () => 0 + } as never, + { + hasVisibleConnection: () => false + } as never, + '' + ) + + await channel.sendReady(createSession()) + + expect(pushed).toHaveLength(1) + }) + + it('also skips SSE in-page toast when native gate reports delivery', async () => { + const pushed: Array<{ namespace: string; payload: PushPayload }> = [] + const toasts: unknown[] = [] + const channel = new PushNotificationChannel( + { + sendToNamespace: async (namespace: string, payload: PushPayload) => { + pushed.push({ namespace, payload }) + } + } as never, + { + sendToast: async (_namespace: string, event: unknown) => { + toasts.push(event) + return 99 + } + } as never, + { + hasVisibleConnection: () => true + } as never, + '' + ) + + const ctx = { nativeGate: { sent: true } } + + await channel.sendReady(createSession(), ctx) + await channel.sendPermissionRequest(createSession({ + agentState: { requests: { 'r-1': { tool: 'Bash', arguments: {} } } } + }), ctx) + await channel.sendTaskNotification(createSession(), { + status: 'completed', + summary: 'Done' + }, ctx) + + // Even when the PWA is foreground/visible, the operator asked to mute + // it - the in-page React toast and the OS web-push are both dropped + // when an FCM companion is on the wrist. + expect(toasts).toHaveLength(0) + expect(pushed).toHaveLength(0) + }) }) diff --git a/hub/src/push/pushNotificationChannel.ts b/hub/src/push/pushNotificationChannel.ts index de3dbe88..bfe751f5 100644 --- a/hub/src/push/pushNotificationChannel.ts +++ b/hub/src/push/pushNotificationChannel.ts @@ -1,5 +1,6 @@ import type { Session } from '../sync/syncEngine' import type { NotificationChannel, TaskNotification } from '../notifications/notificationTypes' +import type { NotificationSendContext } from '../notifications/notificationSendContext' import { getAgentName, getSessionName } from '../notifications/sessionInfo' import type { SSEManager } from '../sse/sseManager' import type { VisibilityTracker } from '../visibility/visibilityTracker' @@ -13,15 +14,27 @@ export class PushNotificationChannel implements NotificationChannel { _appUrl: string ) {} - async sendPermissionRequest(session: Session): Promise { + /** + * Debug observability: gated on `HAPI_NOTIFY_DEBUG=1`. Lets the operator + * see which branch each notification took so we can root-cause "still + * getting PWA notifications" reports without committing permanent log + * spam to the hub journal. + */ + private logBranch(method: string, namespace: string, branch: string, extra: string = ''): void { + if (process.env.HAPI_NOTIFY_DEBUG !== '1') return + const note = extra ? ` ${extra}` : '' + console.log(`[Push.${method}] ns=${namespace} ${branch}${note}`) + } + + async sendPermissionRequest(session: Session, ctx?: NotificationSendContext): Promise { if (!session.active) { return } const name = getSessionName(session) - const request = session.agentState?.requests - ? Object.values(session.agentState.requests)[0] - : null + const requests = session.agentState?.requests ?? null + const requestEntries = requests ? Object.entries(requests) : [] + const [requestId, request] = requestEntries[0] ?? [undefined, null] const toolName = request?.tool ? ` (${request.tool})` : '' const payload: PushPayload = { @@ -31,30 +44,15 @@ export class PushNotificationChannel implements NotificationChannel { data: { type: 'permission-request', sessionId: session.id, - url: this.buildSessionPath(session.id) + url: this.buildSessionPath(session.id), + requestId } } - const url = payload.data?.url ?? this.buildSessionPath(session.id) - if (this.visibilityTracker.hasVisibleConnection(session.namespace)) { - const delivered = await this.sseManager.sendToast(session.namespace, { - type: 'toast', - data: { - title: payload.title, - body: payload.body, - sessionId: session.id, - url - } - }) - if (delivered > 0) { - return - } - } - - await this.pushService.sendToNamespace(session.namespace, payload) + await this.deliverWebOrToast(session, payload, ctx, 'permission') } - async sendReady(session: Session): Promise { + async sendReady(session: Session, ctx?: NotificationSendContext): Promise { if (!session.active) { return } @@ -73,26 +71,10 @@ export class PushNotificationChannel implements NotificationChannel { } } - const url = payload.data?.url ?? this.buildSessionPath(session.id) - if (this.visibilityTracker.hasVisibleConnection(session.namespace)) { - const delivered = await this.sseManager.sendToast(session.namespace, { - type: 'toast', - data: { - title: payload.title, - body: payload.body, - sessionId: session.id, - url - } - }) - if (delivered > 0) { - return - } - } - - await this.pushService.sendToNamespace(session.namespace, payload) + await this.deliverWebOrToast(session, payload, ctx, 'ready') } - async sendTaskNotification(session: Session, notification: TaskNotification): Promise { + async sendTaskNotification(session: Session, notification: TaskNotification, ctx?: NotificationSendContext): Promise { if (!session.active) { return } @@ -115,6 +97,20 @@ export class PushNotificationChannel implements NotificationChannel { } } + await this.deliverWebOrToast(session, payload, ctx, 'task') + } + + private async deliverWebOrToast( + session: Session, + payload: PushPayload, + ctx: NotificationSendContext | undefined, + method: 'permission' | 'ready' | 'task' + ): Promise { + if (ctx?.nativeGate?.sent) { + this.logBranch(method, session.namespace, 'defer-to-native', 'fcm-delivered-this-dispatch') + return + } + const url = payload.data?.url ?? this.buildSessionPath(session.id) if (this.visibilityTracker.hasVisibleConnection(session.namespace)) { const delivered = await this.sseManager.sendToast(session.namespace, { @@ -127,10 +123,15 @@ export class PushNotificationChannel implements NotificationChannel { } }) if (delivered > 0) { + this.logBranch(method, session.namespace, 'sse-toast-delivered', `count=${delivered}`) return } + this.logBranch(method, session.namespace, 'sse-toast-zero', 'visible but delivered=0') + } else { + this.logBranch(method, session.namespace, 'not-visible') } + this.logBranch(method, session.namespace, 'web-push-fired') await this.pushService.sendToNamespace(session.namespace, payload) } diff --git a/hub/src/push/pushService.ts b/hub/src/push/pushService.ts index 3a02d1d9..e44a8fd9 100644 --- a/hub/src/push/pushService.ts +++ b/hub/src/push/pushService.ts @@ -10,6 +10,8 @@ export type PushPayload = { type: string sessionId: string url: string + /** First pending permission request id (permission-request pushes only). */ + requestId?: string } } diff --git a/hub/src/startHub.ts b/hub/src/startHub.ts index 58a47344..a5df16db 100644 --- a/hub/src/startHub.ts +++ b/hub/src/startHub.ts @@ -11,6 +11,9 @@ import { SSEManager } from './sse/sseManager' import { getOrCreateVapidKeys } from './config/vapidKeys' import { PushService } from './push/pushService' import { PushNotificationChannel } from './push/pushNotificationChannel' +import { FcmService } from './fcm/fcmService' +import { FcmNotificationChannel } from './fcm/fcmNotificationChannel' +import { resolveFcmConfig } from './fcm/fcmConfig' import { VisibilityTracker } from './visibility/visibilityTracker' import { TunnelManager } from './tunnel' import { waitForTunnelTlsReady } from './tunnel/tlsGate' @@ -196,9 +199,33 @@ export async function startHub(options: StartHubOptions = {}): Promise { + it('moves a token to a new namespace and removes the old namespace row', () => { + const store = new Store(':memory:') + const device = { token: 'shared-token', platform: 'phone' as const, deviceId: 'pixel-1' } + + store.fcm.upsertDevice('namespace-a', device) + store.fcm.upsertDevice('namespace-b', device) + + expect(store.fcm.getDevicesByNamespace('namespace-a')).toHaveLength(0) + expect(store.fcm.getDevicesByNamespace('namespace-b')).toHaveLength(1) + expect(store.fcm.getDevicesByNamespace('namespace-b')[0].token).toBe('shared-token') + }) +}) diff --git a/hub/src/store/fcmDevices.ts b/hub/src/store/fcmDevices.ts new file mode 100644 index 00000000..a451bb00 --- /dev/null +++ b/hub/src/store/fcmDevices.ts @@ -0,0 +1,75 @@ +import type { Database } from 'bun:sqlite' + +import type { StoredFcmDevice } from './types' + +type DbFcmDeviceRow = { + id: number + namespace: string + token: string + platform: string + device_id: string + created_at: number + updated_at: number +} + +function toStoredFcmDevice(row: DbFcmDeviceRow): StoredFcmDevice { + return { + id: row.id, + namespace: row.namespace, + token: row.token, + platform: row.platform as StoredFcmDevice['platform'], + deviceId: row.device_id, + createdAt: row.created_at, + updatedAt: row.updated_at + } +} + +export function upsertFcmDevice( + db: Database, + namespace: string, + device: { token: string; platform: 'phone' | 'wear'; deviceId: string } +): void { + const now = Date.now() + const params = { + namespace, + token: device.token, + platform: device.platform, + device_id: device.deviceId, + created_at: now, + updated_at: now + } + + db.transaction(() => { + // One FCM token must not deliver across namespaces. Re-pairing the + // same native install under a new namespace drops stale rows that + // still reference this token elsewhere. + db.prepare(` + DELETE FROM fcm_devices + WHERE token = @token + AND (namespace != @namespace OR device_id != @device_id OR platform != @platform) + `).run(params) + + db.prepare(` + INSERT INTO fcm_devices ( + namespace, token, platform, device_id, created_at, updated_at + ) VALUES ( + @namespace, @token, @platform, @device_id, @created_at, @updated_at + ) + ON CONFLICT(namespace, device_id, platform) + DO UPDATE SET + token = excluded.token, + updated_at = excluded.updated_at + `).run(params) + })() +} + +export function removeFcmDeviceByToken(db: Database, namespace: string, token: string): void { + db.prepare('DELETE FROM fcm_devices WHERE namespace = ? AND token = ?').run(namespace, token) +} + +export function getFcmDevicesByNamespace(db: Database, namespace: string): StoredFcmDevice[] { + const rows = db.prepare( + 'SELECT * FROM fcm_devices WHERE namespace = ? ORDER BY updated_at DESC' + ).all(namespace) as DbFcmDeviceRow[] + return rows.map(toStoredFcmDevice) +} diff --git a/hub/src/store/fcmStore.ts b/hub/src/store/fcmStore.ts new file mode 100644 index 00000000..b90ca1bb --- /dev/null +++ b/hub/src/store/fcmStore.ts @@ -0,0 +1,23 @@ +import type { Database } from 'bun:sqlite' + +import type { StoredFcmDevice } from './types' +import { getFcmDevicesByNamespace, removeFcmDeviceByToken, upsertFcmDevice } from './fcmDevices' + +export class FcmStore { + constructor(private readonly db: Database) {} + + upsertDevice( + namespace: string, + device: { token: string; platform: 'phone' | 'wear'; deviceId: string } + ): void { + upsertFcmDevice(this.db, namespace, device) + } + + removeDeviceByToken(namespace: string, token: string): void { + removeFcmDeviceByToken(this.db, namespace, token) + } + + getDevicesByNamespace(namespace: string): StoredFcmDevice[] { + return getFcmDevicesByNamespace(this.db, namespace) + } +} diff --git a/hub/src/store/index.ts b/hub/src/store/index.ts index b0b2c6b0..c4c0e2f6 100644 --- a/hub/src/store/index.ts +++ b/hub/src/store/index.ts @@ -5,6 +5,7 @@ import { dirname } from 'node:path' import { MachineStore } from './machineStore' import { MessageStore } from './messageStore' import { PushStore } from './pushStore' +import { FcmStore } from './fcmStore' import { SessionStore } from './sessionStore' import { UserStore } from './userStore' @@ -12,6 +13,7 @@ export type { StoredMachine, StoredMessage, StoredPushSubscription, + StoredFcmDevice, StoredSession, StoredUser, VersionedUpdateResult @@ -20,16 +22,18 @@ export type { CancelQueuedMessageResult, LookupQueuedMessageResult } from './mes export { MachineStore } from './machineStore' export { MessageStore } from './messageStore' export { PushStore } from './pushStore' +export { FcmStore } from './fcmStore' export { SessionStore } from './sessionStore' export { UserStore } from './userStore' -const SCHEMA_VERSION: number = 10 +const SCHEMA_VERSION: number = 11 const REQUIRED_TABLES = [ 'sessions', 'machines', 'messages', 'users', - 'push_subscriptions' + 'push_subscriptions', + 'fcm_devices' ] as const export class Store { @@ -42,6 +46,7 @@ export class Store { readonly messages: MessageStore readonly users: UserStore readonly push: PushStore + readonly fcm: FcmStore /** * Filesystem path of the underlying SQLite database, or ':memory:' for @@ -92,6 +97,7 @@ export class Store { this.messages = new MessageStore(this.db) this.users = new UserStore(this.db) this.push = new PushStore(this.db) + this.fcm = new FcmStore(this.db) } close(): void { @@ -124,6 +130,7 @@ export class Store { 7: () => this.migrateFromV7ToV8(), 8: () => this.migrateFromV8ToV9(), 9: () => this.migrateFromV9ToV10(), + 10: () => this.migrateFromV10ToV11(), }) if (currentVersion === 0) { @@ -252,6 +259,19 @@ export class Store { UNIQUE(namespace, endpoint) ); CREATE INDEX IF NOT EXISTS idx_push_subscriptions_namespace ON push_subscriptions(namespace); + + CREATE TABLE IF NOT EXISTS fcm_devices ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + namespace TEXT NOT NULL, + token TEXT NOT NULL, + platform TEXT NOT NULL, + device_id TEXT NOT NULL, + created_at INTEGER NOT NULL, + updated_at INTEGER NOT NULL, + UNIQUE(namespace, device_id, platform) + ); + CREATE INDEX IF NOT EXISTS idx_fcm_devices_namespace ON fcm_devices(namespace); + CREATE INDEX IF NOT EXISTS idx_fcm_devices_token ON fcm_devices(token); `) } @@ -435,6 +455,23 @@ export class Store { } } + private migrateFromV10ToV11(): void { + this.db.exec(` + CREATE TABLE IF NOT EXISTS fcm_devices ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + namespace TEXT NOT NULL, + token TEXT NOT NULL, + platform TEXT NOT NULL, + device_id TEXT NOT NULL, + created_at INTEGER NOT NULL, + updated_at INTEGER NOT NULL, + UNIQUE(namespace, device_id, platform) + ); + CREATE INDEX IF NOT EXISTS idx_fcm_devices_namespace ON fcm_devices(namespace); + CREATE INDEX IF NOT EXISTS idx_fcm_devices_token ON fcm_devices(token); + `) + } + private getSessionColumnNames(): Set { const rows = this.db.prepare('PRAGMA table_info(sessions)').all() as Array<{ name: string }> return new Set(rows.map((row) => row.name)) diff --git a/hub/src/store/migration-v10.test.ts b/hub/src/store/migration-v10.test.ts new file mode 100644 index 00000000..710f21e0 --- /dev/null +++ b/hub/src/store/migration-v10.test.ts @@ -0,0 +1,131 @@ +import { describe, expect, it } from 'bun:test' +import { Database } from 'bun:sqlite' +import { mkdtempSync, rmSync } from 'node:fs' +import { join } from 'node:path' +import { tmpdir } from 'node:os' +import { Store } from './index' + +describe('Store V10→V11 migration: fcm_devices', () => { + it('fresh DB has fcm_devices table', () => { + const store = new Store(':memory:') + expect(tableExists(store, 'fcm_devices')).toBe(true) + }) + + it('V10 DB migrates to V11: fcm_devices created', () => { + const dir = mkdtempSync(join(tmpdir(), 'hapi-migration-v11-test-')) + const dbPath = join(dir, 'test.db') + let store: Store | undefined + try { + const db = new Database(dbPath, { create: true, readwrite: true, strict: true }) + db.exec('PRAGMA journal_mode = WAL') + db.exec('PRAGMA foreign_keys = ON') + createV10Schema(db) + db.exec('PRAGMA user_version = 10') + db.close() + + store = new Store(dbPath) + expect(tableExists(store, 'fcm_devices')).toBe(true) + } finally { + store?.close() + rmSync(dir, { recursive: true, force: true }) + } + }) + + it('upsert replaces token for same namespace+deviceId+platform', () => { + const store = new Store(':memory:') + store.fcm.upsertDevice('default', { + token: 'tok-a', + platform: 'phone', + deviceId: 'pixel-1' + }) + store.fcm.upsertDevice('default', { + token: 'tok-b', + platform: 'phone', + deviceId: 'pixel-1' + }) + const devices = store.fcm.getDevicesByNamespace('default') + expect(devices).toHaveLength(1) + expect(devices[0].token).toBe('tok-b') + }) +}) + +function tableExists(store: Store, name: string): boolean { + const db: Database = (store as unknown as { db: Database }).db + const row = db.prepare( + "SELECT name FROM sqlite_master WHERE type='table' AND name=?" + ).get(name) as { name: string } | null + return row !== null +} + +function createV10Schema(db: Database): void { + db.exec(` + CREATE TABLE IF NOT EXISTS sessions ( + id TEXT PRIMARY KEY, + tag TEXT, + namespace TEXT NOT NULL DEFAULT 'default', + machine_id TEXT, + created_at INTEGER NOT NULL, + updated_at INTEGER NOT NULL, + metadata TEXT, + metadata_version INTEGER DEFAULT 1, + agent_state TEXT, + agent_state_version INTEGER DEFAULT 1, + model TEXT, + model_reasoning_effort TEXT, + effort TEXT, + service_tier TEXT, + todos TEXT, + todos_updated_at INTEGER, + team_state TEXT, + team_state_updated_at INTEGER, + active INTEGER DEFAULT 0, + active_at INTEGER, + seq INTEGER DEFAULT 0 + ); + + CREATE TABLE IF NOT EXISTS machines ( + id TEXT PRIMARY KEY, + namespace TEXT NOT NULL DEFAULT 'default', + created_at INTEGER NOT NULL, + updated_at INTEGER NOT NULL, + metadata TEXT, + metadata_version INTEGER DEFAULT 1, + runner_state TEXT, + runner_state_version INTEGER DEFAULT 1, + active INTEGER DEFAULT 0, + active_at INTEGER, + seq INTEGER DEFAULT 0 + ); + + CREATE TABLE IF NOT EXISTS messages ( + id TEXT PRIMARY KEY, + session_id TEXT NOT NULL, + content TEXT NOT NULL, + created_at INTEGER NOT NULL, + seq INTEGER NOT NULL, + local_id TEXT, + invoked_at INTEGER, + scheduled_at INTEGER, + FOREIGN KEY (session_id) REFERENCES sessions(id) ON DELETE CASCADE + ); + + CREATE TABLE IF NOT EXISTS users ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + platform TEXT NOT NULL, + platform_user_id TEXT NOT NULL, + namespace TEXT NOT NULL DEFAULT 'default', + created_at INTEGER NOT NULL, + UNIQUE(platform, platform_user_id) + ); + + CREATE TABLE IF NOT EXISTS push_subscriptions ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + namespace TEXT NOT NULL, + endpoint TEXT NOT NULL, + p256dh TEXT NOT NULL, + auth TEXT NOT NULL, + created_at INTEGER NOT NULL, + UNIQUE(namespace, endpoint) + ); + `) +} diff --git a/hub/src/store/types.ts b/hub/src/store/types.ts index ae3f6482..6917c19a 100644 --- a/hub/src/store/types.ts +++ b/hub/src/store/types.ts @@ -64,6 +64,16 @@ export type StoredPushSubscription = { createdAt: number } +export type StoredFcmDevice = { + id: number + namespace: string + token: string + platform: 'phone' | 'wear' + deviceId: string + createdAt: number + updatedAt: number +} + export type VersionedUpdateResult = | { result: 'success'; version: number; value: T } | { result: 'version-mismatch'; version: number; value: T } diff --git a/hub/src/telegram/sessionView.ts b/hub/src/telegram/sessionView.ts index f61d5b6d..0f64303c 100644 --- a/hub/src/telegram/sessionView.ts +++ b/hub/src/telegram/sessionView.ts @@ -8,10 +8,9 @@ import { InlineKeyboard } from 'grammy' import type { Machine, Session } from '../sync/syncEngine' import { ACTIONS } from './callbacks' -import { createCallbackData, truncate, getSessionName } from './renderer' +import { createCallbackData, getSessionName } from './renderer' import { getAgentName } from '../notifications/sessionInfo' - -const MAX_TOOL_ARGS_LENGTH = 150 +import { formatToolArgumentsDetailed } from '../notifications/toolArgs' type NotificationContext = { hasContext: boolean @@ -157,79 +156,6 @@ export function createNotificationKeyboard(session: Session, publicUrl: string): return keyboard } -/** - * Format detailed tool arguments for notification display - */ -function formatToolArgumentsDetailed(tool: string, args: any): string { - if (!args) return '' - - try { - switch (tool) { - case 'Edit': { - const file = args.file_path || args.path || 'unknown' - const oldStr = args.old_string ? truncate(args.old_string, 50) : '' - const newStr = args.new_string ? truncate(args.new_string, 50) : '' - let result = `File: ${truncate(file, MAX_TOOL_ARGS_LENGTH)}` - if (oldStr) result += `\nOld: "${oldStr}"` - if (newStr) result += `\nNew: "${newStr}"` - return result - } - - case 'Write': { - const file = args.file_path || args.path || 'unknown' - const content = args.content ? `${args.content.length} chars` : '' - return `File: ${truncate(file, MAX_TOOL_ARGS_LENGTH)}${content ? ` (${content})` : ''}` - } - - case 'Read': { - const file = args.file_path || args.path || 'unknown' - return `File: ${truncate(file, MAX_TOOL_ARGS_LENGTH)}` - } - - case 'Bash': { - const cmd = args.command || '' - return `Command: ${truncate(cmd, MAX_TOOL_ARGS_LENGTH)}` - } - - case 'Agent': - case 'Task': { - const desc = args.description || args.prompt || '' - return `Task: ${truncate(desc, MAX_TOOL_ARGS_LENGTH)}` - } - - case 'Grep': - case 'Glob': { - const pattern = args.pattern || '' - const path = args.path || '' - let result = `Pattern: ${pattern}` - if (path) result += `\nPath: ${truncate(path, 80)}` - return result - } - - case 'WebFetch': { - const url = args.url || '' - return `URL: ${truncate(url, MAX_TOOL_ARGS_LENGTH)}` - } - - case 'TodoWrite': { - const count = args.todos?.length || 0 - return `Updating ${count} todo items` - } - - default: { - // Generic args display for unknown tools - const argStr = JSON.stringify(args) - if (argStr.length > 10) { - return `Args: ${truncate(argStr, MAX_TOOL_ARGS_LENGTH)}` - } - return '' - } - } - } catch { - return '' - } -} - function buildMiniAppDeepLink(baseUrl: string, startParam: string): string { try { const url = new URL(baseUrl) diff --git a/hub/src/web/routes/devices.test.ts b/hub/src/web/routes/devices.test.ts new file mode 100644 index 00000000..07819e2d --- /dev/null +++ b/hub/src/web/routes/devices.test.ts @@ -0,0 +1,56 @@ +import { describe, expect, it } from 'bun:test' +import { Hono } from 'hono' +import { SignJWT } from 'jose' +import type { WebAppEnv } from '../middleware/auth' +import { createAuthMiddleware } from '../middleware/auth' +import { Store } from '../../store' +import { createDevicesRoutes } from './devices' + +const JWT_SECRET = new TextEncoder().encode('test-secret') + +async function authHeaders() { + const token = await new SignJWT({ uid: 1, ns: 'default' }) + .setProtectedHeader({ alg: 'HS256' }) + .setIssuedAt() + .setExpirationTime('1h') + .sign(JWT_SECRET) + return { authorization: `Bearer ${token}` } +} + +function createApp(store: Store) { + const app = new Hono() + app.use('*', createAuthMiddleware(JWT_SECRET)) + app.route('/api', createDevicesRoutes(store)) + return app +} + +describe('devices routes', () => { + it('registers and unregisters FCM devices for namespace', async () => { + const store = new Store(':memory:') + const app = createApp(store) + const headers = await authHeaders() + + const register = await app.request('/api/devices/register', { + method: 'POST', + headers: { ...headers, 'content-type': 'application/json' }, + body: JSON.stringify({ + token: 'fcm-tok-1', + platform: 'wear', + deviceId: 'watch-1' + }) + }) + expect(register.status).toBe(200) + + const devices = store.fcm.getDevicesByNamespace('default') + expect(devices).toHaveLength(1) + expect(devices[0].platform).toBe('wear') + + const unregister = await app.request('/api/devices/register', { + method: 'DELETE', + headers: { ...headers, 'content-type': 'application/json' }, + body: JSON.stringify({ token: 'fcm-tok-1' }) + }) + expect(unregister.status).toBe(200) + expect(store.fcm.getDevicesByNamespace('default')).toHaveLength(0) + }) +}) diff --git a/hub/src/web/routes/devices.ts b/hub/src/web/routes/devices.ts new file mode 100644 index 00000000..e42fb6ca --- /dev/null +++ b/hub/src/web/routes/devices.ts @@ -0,0 +1,44 @@ +import { Hono } from 'hono' +import { z } from 'zod' +import type { Store } from '../../store' +import type { WebAppEnv } from '../middleware/auth' + +const registerSchema = z.object({ + token: z.string().min(1), + platform: z.enum(['phone', 'wear']), + deviceId: z.string().min(1).max(128) +}) + +const unregisterSchema = z.object({ + token: z.string().min(1) +}) + +export function createDevicesRoutes(store: Store): Hono { + const app = new Hono() + + app.post('/devices/register', async (c) => { + const json = await c.req.json().catch(() => null) + const parsed = registerSchema.safeParse(json) + if (!parsed.success) { + return c.json({ error: 'Invalid body', issues: parsed.error.flatten() }, 400) + } + + const namespace = c.get('namespace') + store.fcm.upsertDevice(namespace, parsed.data) + return c.json({ ok: true }) + }) + + app.delete('/devices/register', async (c) => { + const json = await c.req.json().catch(() => null) + const parsed = unregisterSchema.safeParse(json) + if (!parsed.success) { + return c.json({ error: 'Invalid body', issues: parsed.error.flatten() }, 400) + } + + const namespace = c.get('namespace') + store.fcm.removeDeviceByToken(namespace, parsed.data.token) + return c.json({ ok: true }) + }) + + return app +} diff --git a/hub/src/web/server.ts b/hub/src/web/server.ts index e3c1a96a..aab64425 100644 --- a/hub/src/web/server.ts +++ b/hub/src/web/server.ts @@ -22,6 +22,7 @@ import { createGitRoutes } from './routes/git' import { createCliRoutes } from './routes/cli' import { createCodexDesktopRoutes } from './routes/codexDesktop' import { createPushRoutes } from './routes/push' +import { createDevicesRoutes } from './routes/devices' import { createVoiceRoutes } from './routes/voice' import type { SSEManager } from '../sse/sseManager' import type { VisibilityTracker } from '../visibility/visibilityTracker' @@ -254,6 +255,7 @@ function createWebApp(options: { getSyncEngine: options.getSyncEngine })) app.route('/api', createPushRoutes(options.store, options.vapidPublicKey)) + app.route('/api', createDevicesRoutes(options.store)) app.route('/api', createVoiceRoutes()) // Skip static serving in relay mode, show helpful message on root diff --git a/shared/src/messages.test.ts b/shared/src/messages.test.ts new file mode 100644 index 00000000..b6f4de14 --- /dev/null +++ b/shared/src/messages.test.ts @@ -0,0 +1,221 @@ +import { describe, expect, test } from 'bun:test' +import { + extractAssistantPlainText, + extractNotifySummary, + isRedundantGoalStatusEventContent, + type NotifySummary +} from './messages' + +describe('extractAssistantPlainText', () => { + test('returns null for non-objects', () => { + expect(extractAssistantPlainText(null)).toBeNull() + expect(extractAssistantPlainText(undefined)).toBeNull() + expect(extractAssistantPlainText('string')).toBeNull() + expect(extractAssistantPlainText(42)).toBeNull() + }) + + test('extracts codex/message text', () => { + const content = { + type: 'codex', + data: { + type: 'message', + message: 'Hello there.' + } + } + expect(extractAssistantPlainText(content)).toBe('Hello there.') + }) + + test('returns null for codex/tool-call (no text)', () => { + const content = { + type: 'codex', + data: { + type: 'tool-call', + name: 'Edit', + callId: 'x', + input: {} + } + } + expect(extractAssistantPlainText(content)).toBeNull() + }) + + test('returns null for codex/tool-call-result (no text)', () => { + const content = { + type: 'codex', + data: { + type: 'tool-call-result', + output: {} + } + } + expect(extractAssistantPlainText(content)).toBeNull() + }) + + test('returns null when codex/message string is empty', () => { + const content = { type: 'codex', data: { type: 'message', message: '' } } + expect(extractAssistantPlainText(content)).toBeNull() + }) + + test('extracts output/assistant text from claude SDK content array', () => { + const content = { + type: 'output', + data: { + type: 'assistant', + message: { + content: [ + { type: 'text', text: 'Line one.' }, + { type: 'tool_use', name: 'Edit' }, + { type: 'text', text: 'Line two.' } + ] + } + } + } + expect(extractAssistantPlainText(content)).toBe('Line one.\nLine two.') + }) + + test('returns null for output/assistant with no text blocks', () => { + const content = { + type: 'output', + data: { + type: 'assistant', + message: { content: [{ type: 'tool_use', name: 'Edit' }] } + } + } + expect(extractAssistantPlainText(content)).toBeNull() + }) + + test('returns null for output/user (not assistant)', () => { + const content = { type: 'output', data: { type: 'user', message: { content: [] } } } + expect(extractAssistantPlainText(content)).toBeNull() + }) + + test('returns null for unknown content shapes', () => { + expect(extractAssistantPlainText({ type: 'event', data: {} })).toBeNull() + expect(extractAssistantPlainText({ type: 'text' })).toBeNull() + }) +}) + +describe('extractNotifySummary', () => { + const FULL_LINE = 'AGENT_NOTIFY_SUMMARY {"version":1,"agent":"hapi-monitor agent","project":"hapi-monitor","status":"done","action":"Revoke tokens","summary":"Published v0.1.0"}' + + test('returns null on non-string input', () => { + expect(extractNotifySummary(null)).toBeNull() + expect(extractNotifySummary(undefined)).toBeNull() + expect(extractNotifySummary({})).toBeNull() + expect(extractNotifySummary(42)).toBeNull() + expect(extractNotifySummary('')).toBeNull() + }) + + test('parses a summary on its own line at the very end', () => { + const result = extractNotifySummary(FULL_LINE) + expect(result).not.toBeNull() + const r = result as NotifySummary + expect(r.version).toBe(1) + expect(r.agent).toBe('hapi-monitor agent') + expect(r.project).toBe('hapi-monitor') + expect(r.status).toBe('done') + expect(r.action).toBe('Revoke tokens') + expect(r.summary).toBe('Published v0.1.0') + }) + + test('parses summary as last non-empty line after preceding prose', () => { + const text = `Here is what I did.\n\nThings worked.\n\n${FULL_LINE}` + const r = extractNotifySummary(text) + expect(r?.summary).toBe('Published v0.1.0') + }) + + test('tolerates trailing whitespace and blank lines', () => { + const r = extractNotifySummary(`prose\n\n${FULL_LINE}\n\n \n`) + expect(r?.summary).toBe('Published v0.1.0') + }) + + test('returns null when summary is not on the LAST non-empty line', () => { + // Operator wrote prose AFTER the line - non-compliant. + const text = `${FULL_LINE}\nOh, one more thing.` + expect(extractNotifySummary(text)).toBeNull() + }) + + test('returns null when prefix is missing', () => { + expect(extractNotifySummary('NOTIFY_SUMMARY {"summary":"x"}')).toBeNull() + expect(extractNotifySummary('agent_notify_summary {"summary":"x"}')).toBeNull() + }) + + test('returns null when JSON is malformed', () => { + expect(extractNotifySummary('AGENT_NOTIFY_SUMMARY {bogus}')).toBeNull() + expect(extractNotifySummary('AGENT_NOTIFY_SUMMARY {"summary":')).toBeNull() + expect(extractNotifySummary('AGENT_NOTIFY_SUMMARY not-json')).toBeNull() + }) + + test('drops fields with wrong types but keeps valid ones', () => { + const text = 'AGENT_NOTIFY_SUMMARY {"version":"oops","summary":"x","action":42,"status":"done"}' + const r = extractNotifySummary(text) + expect(r?.summary).toBe('x') + expect(r?.status).toBe('done') + expect(r?.version).toBeUndefined() + expect(r?.action).toBeUndefined() + }) + + test('ignores in-message quotes of the line - only the LAST line is parsed', () => { + // This very test message contains the literal prefix in a quoted explanation, + // but the trailing line is plain prose, so we return null. + const text = `Earlier I described the format as 'AGENT_NOTIFY_SUMMARY {...}', but here is plain text.` + expect(extractNotifySummary(text)).toBeNull() + }) + + test('returns null for whitespace-only input', () => { + expect(extractNotifySummary(' \n\n ')).toBeNull() + }) + + test('handles JSON with internal braces (escaped within strings)', () => { + const text = 'AGENT_NOTIFY_SUMMARY {"summary":"thing {nested} thing","status":"done"}' + const r = extractNotifySummary(text) + expect(r?.summary).toBe('thing {nested} thing') + expect(r?.status).toBe('done') + }) +}) + +describe('extractNotifySummary + extractAssistantPlainText (integration)', () => { + test('codex assistant text containing a trailing summary line', () => { + const content = { + type: 'codex', + data: { + type: 'message', + message: 'Did the work.\n\nAGENT_NOTIFY_SUMMARY {"summary":"Done","status":"done"}' + } + } + const text = extractAssistantPlainText(content) + expect(text).not.toBeNull() + const r = extractNotifySummary(text!) + expect(r?.summary).toBe('Done') + }) + + test('claude SDK output with summary in the last text block', () => { + const content = { + type: 'output', + data: { + type: 'assistant', + message: { + content: [ + { type: 'text', text: 'Quick update.' }, + { type: 'text', text: 'AGENT_NOTIFY_SUMMARY {"summary":"All checks green","status":"done","action":"Merge PR"}' } + ] + } + } + } + const text = extractAssistantPlainText(content) + const r = extractNotifySummary(text!) + expect(r?.summary).toBe('All checks green') + expect(r?.action).toBe('Merge PR') + }) +}) + +describe('isRedundantGoalStatusEventContent (regression-guard for messages.ts edits)', () => { + test('still detects goal-active events', () => { + const value = { + role: 'agent', + content: { + type: 'event', + data: { type: 'message', message: 'Goal active · build the thing' } + } + } + expect(isRedundantGoalStatusEventContent(value)).toBe(true) + }) +}) diff --git a/shared/src/messages.ts b/shared/src/messages.ts index 44bd0053..2a33ca31 100644 --- a/shared/src/messages.ts +++ b/shared/src/messages.ts @@ -76,4 +76,107 @@ export function isRedundantGoalStatusEventContent(value: unknown): boolean { return isRedundantGoalStatusMessageText(data.message) } +/** + * Best-effort plain-text extraction from a stored agent message's `content`. + * + * Two structural shapes are common in this fork: + * + * 1. `codex` flavor: content.type = 'codex', content.data.type = 'message' + * -> assistant text at `content.data.message` (string). + * + * 2. `output` flavor (Claude SDK passthrough): content.type = 'output', + * content.data.type = 'assistant' -> text at + * `content.data.message.content[i].text` (array of `{type:'text', text}`). + * + * Returns `null` when the content does not look like assistant *text* + * (tool calls, tool results, reasoning, token counts, etc.) so callers can + * skip those messages and fall back to the previous one. + */ +export function extractAssistantPlainText(content: unknown): string | null { + if (!isObject(content)) return null + + if (content.type === 'codex') { + const data = isObject(content.data) ? content.data : null + if (!data || data.type !== 'message') return null + return typeof data.message === 'string' && data.message.length > 0 + ? data.message + : null + } + + if (content.type === 'output') { + const data = isObject(content.data) ? content.data : null + if (!data || data.type !== 'assistant') return null + const message = isObject(data.message) ? data.message : null + const blocks = Array.isArray(message?.content) ? message.content : null + if (!blocks) return null + const textParts: string[] = [] + for (const block of blocks) { + if (!isObject(block)) continue + if (block.type === 'text' && typeof block.text === 'string') { + textParts.push(block.text) + } + } + if (textParts.length === 0) return null + return textParts.join('\n') + } + + return null +} + +const NOTIFY_SUMMARY_PREFIX = 'AGENT_NOTIFY_SUMMARY ' + +export type NotifySummary = { + version?: number + agent?: string + project?: string + status?: string + action?: string + summary?: string +} + +/** + * Look for an `AGENT_NOTIFY_SUMMARY {...json...}` line as the **last + * non-empty line** of an agent's plain-text message. + * + * Strict end-anchor: anything below the JSON line (even whitespace) is + * fine, but if the agent wrote prose AFTER the line we treat it as + * non-compliant and return null. This also makes false positives from + * `AGENT_NOTIFY_SUMMARY` quoted inside an earlier paragraph harmless, + * because such a quote is never the last line. + * + * Returns the parsed object on success, `null` on any deviation. The + * shape is intentionally loose - we only trust `summary`, `action`, and + * `status` for notification rendering, but the full object is forwarded + * onto the meta-event bus when Phase 2 lands. + */ +export function extractNotifySummary(text: unknown): NotifySummary | null { + if (typeof text !== 'string' || text.length === 0) return null + + const lines = text.split('\n') + let lastIdx = lines.length - 1 + while (lastIdx >= 0 && lines[lastIdx].trim() === '') lastIdx -= 1 + if (lastIdx < 0) return null + + const lastLine = lines[lastIdx].trim() + if (!lastLine.startsWith(NOTIFY_SUMMARY_PREFIX)) return null + + const jsonPart = lastLine.slice(NOTIFY_SUMMARY_PREFIX.length).trim() + if (!jsonPart.startsWith('{') || !jsonPart.endsWith('}')) return null + + try { + const parsed: unknown = JSON.parse(jsonPart) + if (!isObject(parsed)) return null + const result: NotifySummary = {} + if (typeof parsed.version === 'number') result.version = parsed.version + if (typeof parsed.agent === 'string') result.agent = parsed.agent + if (typeof parsed.project === 'string') result.project = parsed.project + if (typeof parsed.status === 'string') result.status = parsed.status + if (typeof parsed.action === 'string') result.action = parsed.action + if (typeof parsed.summary === 'string') result.summary = parsed.summary + return result + } catch { + return null + } +} + export type { RoleWrappedRecord } diff --git a/web/package.json b/web/package.json index b1102676..436ff722 100644 --- a/web/package.json +++ b/web/package.json @@ -40,6 +40,7 @@ "mermaid": "^11.12.0", "react": "^19.2.3", "react-dom": "^19.2.3", + "qrcode": "^1.5.4", "rehype-katex": "^7.0.1", "remark-breaks": "^4.0.0", "remark-gfm": "^4.0.1", @@ -58,6 +59,7 @@ "@testing-library/react": "^16.3.0", "@types/react": "^19.2.7", "@types/react-dom": "^19.2.3", + "@types/qrcode": "^1.5.6", "@tailwindcss/postcss": "^4.1.18", "@vitejs/plugin-react": "^5.1.2", "autoprefixer": "^10.4.23", diff --git a/web/src/components/settings/CompanionPairing.tsx b/web/src/components/settings/CompanionPairing.tsx new file mode 100644 index 00000000..1c1db9c6 --- /dev/null +++ b/web/src/components/settings/CompanionPairing.tsx @@ -0,0 +1,126 @@ +import { useEffect, useMemo, useRef, useState } from 'react' +import QRCode from 'qrcode' +import { Button } from '@/components/ui/button' +import { useTranslation } from '@/lib/use-translation' + +type CompanionPairingProps = { + baseUrl: string +} + +const COMPANION_DEEPLINK_SCHEME = 'hapicompanion://bind' +const ACCESS_TOKEN_PREFIX = 'hapi_access_token::' + +function readAccessToken(baseUrl: string): string { + if (typeof window === 'undefined') return '' + try { + return window.localStorage.getItem(`${ACCESS_TOKEN_PREFIX}${baseUrl}`) ?? '' + } catch { + return '' + } +} + +function buildDeeplink(hub: string, code: string): string { + const params = new URLSearchParams({ hub, code }) + return `${COMPANION_DEEPLINK_SCHEME}?${params.toString()}` +} + +export function CompanionPairing({ baseUrl }: CompanionPairingProps) { + const { t } = useTranslation() + const [revealed, setRevealed] = useState(false) + const [copied, setCopied] = useState(false) + const canvasRef = useRef(null) + + const accessToken = useMemo(() => readAccessToken(baseUrl), [baseUrl]) + + const deeplink = useMemo(() => { + const hub = (baseUrl || '').trim() + const code = (accessToken || '').trim() + if (!hub || !code) return '' + return buildDeeplink(hub, code) + }, [baseUrl, accessToken]) + + useEffect(() => { + if (!revealed || !deeplink || !canvasRef.current) return + let cancelled = false + QRCode.toCanvas(canvasRef.current, deeplink, { + errorCorrectionLevel: 'M', + margin: 2, + scale: 6, + color: { + dark: '#000000', + light: '#ffffff' + } + }).catch(() => { + // QR rendering failures are non-fatal; the textual link below still works. + }) + return () => { + cancelled = true + void cancelled + } + }, [deeplink, revealed]) + + const handleCopy = async () => { + if (!deeplink) return + try { + await navigator.clipboard.writeText(deeplink) + setCopied(true) + setTimeout(() => setCopied(false), 2000) + } catch { + // Clipboard may be unavailable (e.g. insecure context); user can long-press the link instead. + } + } + + if (!deeplink) { + return ( +

+ {t('settings.companion.noToken')} +

+ ) + } + + return ( +
+

+ {t('settings.companion.description')} +

+ + {!revealed ? ( + + ) : ( +
+ +
+ + +
+

+ {deeplink} +

+
+ )} +
+ ) +} diff --git a/web/src/hooks/useAuth.ts b/web/src/hooks/useAuth.ts index 8a6c38b2..48f04fa1 100644 --- a/web/src/hooks/useAuth.ts +++ b/web/src/hooks/useAuth.ts @@ -37,6 +37,16 @@ function isNotBoundError(error: unknown): boolean { return error instanceof ApiError && error.status === 401 && error.code === 'not_bound' } +const ACCESS_TOKEN_PREFIX = 'hapi_access_token::' + +function rememberAccessToken(baseUrl: string, accessToken: string): void { + try { + localStorage.setItem(`${ACCESS_TOKEN_PREFIX}${baseUrl}`, accessToken) + } catch { + // Ignore storage errors (private mode, full quota, etc.) + } +} + export function useAuth(authSource: AuthSource | null, baseUrl: string): { token: string | null user: AuthResponse['user'] | null @@ -149,6 +159,10 @@ export function useAuth(authSource: AuthSource | null, baseUrl: string): { setToken(auth.token) setUser(auth.user) setNeedsBinding(false) + // Persist the CLI access token so Settings → Companion pairing QR + // can encode the same long-lived token in the deeplink. The PWA + // already does this for browser/CLI logins via useAuthSource. + rememberAccessToken(baseUrl, accessToken) } catch (error) { setError(error instanceof Error ? error.message : 'Binding failed') throw error diff --git a/web/src/hooks/usePwaUpdate.test.ts b/web/src/hooks/usePwaUpdate.test.ts index d5733b69..1880b718 100644 --- a/web/src/hooks/usePwaUpdate.test.ts +++ b/web/src/hooks/usePwaUpdate.test.ts @@ -130,7 +130,7 @@ describe('requestPwaUpdateReload', () => { setTimeoutFn: vi.fn((callback, delay) => { expect(delay).toBe(PWA_UPDATE_RELOAD_FALLBACK_MS) return setTimeout(callback, delay) - }) as typeof setTimeout, + }) as unknown as typeof setTimeout, }) await pending diff --git a/web/src/lib/locales/en.ts b/web/src/lib/locales/en.ts index bc33d8db..9ec9222f 100644 --- a/web/src/lib/locales/en.ts +++ b/web/src/lib/locales/en.ts @@ -611,7 +611,7 @@ export default { 'settings.hub.description': 'Choose a category to adjust HAPI to your workflow.', 'settings.hub.voice.summary': 'Voice, language, and behavior', 'settings.general.title': 'General', - 'settings.general.description': 'Language and general application preferences.', + 'settings.general.description': 'Language, companion pairing, and general application preferences.', 'settings.language.title': 'Language', 'settings.language.label': 'Language', 'settings.display.title': 'Display', @@ -800,6 +800,14 @@ export default { 'settings.voice.session.label': 'Session behavior', 'settings.voice.proactive': 'Start voice session with summary', 'settings.voice.proactive.description': 'When on, starting a voice session opens with a spoken summary of current agent activity. When off, the assistant greets you and waits for you to speak.', + 'settings.companion.title': 'Companion', + 'settings.companion.noToken': 'Pairing requires the original access token (CLI_API_TOKEN). It looks like you signed in via Telegram or another flow that did not store one — paste the token manually in the companion app instead.', + 'settings.companion.description': 'Scan from the HAPI companion app on Android (phone or Wear OS) to bind it to this hub. The pairing code is your access token — treat it like a password.', + 'settings.companion.showQr': 'Show pairing QR', + 'settings.companion.qrAriaLabel': 'Companion pairing QR code', + 'settings.companion.copyLink': 'Copy link', + 'settings.companion.copied': 'Copied!', + 'settings.companion.hide': 'Hide', 'settings.about.title': 'About', 'settings.about.description': 'HAPI links and version information.', 'settings.about.website': 'Website', diff --git a/web/src/lib/locales/zh-CN.ts b/web/src/lib/locales/zh-CN.ts index 4a8aabd0..68aaafbf 100644 --- a/web/src/lib/locales/zh-CN.ts +++ b/web/src/lib/locales/zh-CN.ts @@ -615,7 +615,7 @@ export default { 'settings.hub.description': '选择一个分类，按你的工作方式调整 HAPI。', 'settings.hub.voice.summary': '声音、语言和行为', 'settings.general.title': '通用', - 'settings.general.description': '语言和通用应用偏好。', + 'settings.general.description': '语言、伴侣应用配对和通用应用偏好。', 'settings.language.title': '语言', 'settings.language.label': '语言', 'settings.display.title': '显示', @@ -804,6 +804,14 @@ export default { 'settings.voice.session.label': '会话行为', 'settings.voice.proactive': '以摘要开始语音会话', 'settings.voice.proactive.description': '开启后，启动语音会话时将朗读当前代理活动的摘要。关闭后，助手向您打招呼并等待您先开口。', + 'settings.companion.title': '伴侣应用', + 'settings.companion.noToken': '配对需要原始访问令牌（CLI_API_TOKEN）。您似乎通过 Telegram 或其他未保存令牌的流程登录 — 请在伴侣应用中手动粘贴令牌。', + 'settings.companion.description': '在 Android 手机或 Wear OS 上的 HAPI 伴侣应用中扫描，以绑定此 Hub。配对码即您的访问令牌 — 请像密码一样妥善保管。', + 'settings.companion.showQr': '显示配对二维码', + 'settings.companion.qrAriaLabel': '伴侣应用配对二维码', + 'settings.companion.copyLink': '复制链接', + 'settings.companion.copied': '已复制！', + 'settings.companion.hide': '隐藏', 'settings.about.title': '关于', 'settings.about.description': 'HAPI 链接和版本信息。', 'settings.about.website': '官方网站', diff --git a/web/src/routes/settings/general.tsx b/web/src/routes/settings/general.tsx index a0c4dabd..8210cd51 100644 --- a/web/src/routes/settings/general.tsx +++ b/web/src/routes/settings/general.tsx @@ -1,4 +1,6 @@ import { useTranslation, type Locale } from '@/lib/use-translation' +import { useAppContext } from '@/lib/app-context' +import { CompanionPairing } from '@/components/settings/CompanionPairing' import { SettingsChoiceGroup, SettingsPageContent, SettingsSection } from '@/components/settings/SettingsPrimitives' const locales: ReadonlyArray<{ value: Locale; label: string }> = [ @@ -8,11 +10,17 @@ const locales: ReadonlyArray<{ value: Locale; label: string }> = [ export default function SettingsGeneralPage() { const { t, locale, setLocale } = useTranslation() + const { baseUrl } = useAppContext() return ( + +
+ +
+ ) } diff --git a/web/src/routes/settings/index.test.tsx b/web/src/routes/settings/index.test.tsx index ff66266f..aa8726c5 100644 --- a/web/src/routes/settings/index.test.tsx +++ b/web/src/routes/settings/index.test.tsx @@ -126,6 +126,17 @@ vi.mock('@/hooks/useChatSurfaceColors', () => ({ toCustomChatSurfaceColorPreference: (value: string) => `custom:${value}`, })) +vi.mock('@/lib/app-context', () => ({ + useAppContext: () => ({ + api: {}, + baseUrl: 'http://127.0.0.1:3006', + }), +})) + +vi.mock('@/components/settings/CompanionPairing', () => ({ + CompanionPairing: () =>
Companion pairing
, +})) + vi.mock('@/components/settings/VoiceAdvancedControls', () => ({ VoiceRespondsControls: () =>
Response length controls
, VoiceSoundsControls: () =>
Sound controls
, @@ -177,6 +188,8 @@ describe('responsive settings pages', () => { it('changes the application language inline', () => { renderPage() + expect(screen.getByText('Companion')).toBeInTheDocument() + expect(screen.getByText('Companion pairing')).toBeInTheDocument() fireEvent.click(screen.getByRole('radio', { name: '简体中文' })) expect(localStorage.getItem('hapi-lang')).toBe('zh-CN') }) @@ -200,6 +213,7 @@ describe('responsive settings pages', () => { it('renders About metadata on its own route page', () => { renderPage() + expect(screen.queryByText('Companion')).not.toBeInTheDocument() expect(screen.getByText('App Version')).toBeInTheDocument() expect(screen.getByText(String(__APP_VERSION__))).toBeInTheDocument() expect(screen.getByText('Protocol Version')).toBeInTheDocument()