From 6aa9f921560b0ee2333706a7756a39776be64042 Mon Sep 17 00:00:00 2001 From: Mohamed Boudra Date: Mon, 10 Aug 2026 23:33:31 +0200 Subject: [PATCH] Stop notifications after removing a host (#3176) * fix(notifications): stop pushes after host removal Push tokens were persisted indefinitely with no relationship to app pairing state. Give registrations a 48-hour renewable lease so offline unpairing converges, and revoke immediately when the host is reachable. * fix(notifications): make token revocation bounded and durable --- packages/app/src/contexts/session-context.tsx | 4 +- .../src/hooks/use-push-token-registration.ts | 111 ----------------- .../src/push-notifications/index.native.ts | 12 ++ packages/app/src/push-notifications/index.ts | 9 ++ .../app/src/push-notifications/index.web.ts | 9 ++ .../internal/subscriptions.ts | 103 ++++++++++++++++ .../src/push-notifications/internal/types.ts | 11 ++ packages/app/src/runtime/host-runtime.test.ts | 48 ++++++++ packages/app/src/runtime/host-runtime.ts | 15 ++- packages/client/src/daemon-client.test.ts | 53 ++++++++ packages/client/src/daemon-client.ts | 11 ++ packages/protocol/src/messages.ts | 19 +++ packages/server/src/server/bootstrap.ts | 2 +- .../src/server/client-activity.e2e.test.ts | 2 +- packages/server/src/server/push/index.test.ts | 67 ++++++++++ packages/server/src/server/push/index.ts | 45 +++++++ .../server/src/server/push/notifications.ts | 29 ----- .../server/src/server/push/push-service.ts | 9 +- .../src/server/push/token-store.test.ts | 57 ++++++++- .../server/src/server/push/token-store.ts | 116 ++++++++++++++---- packages/server/src/server/session.test.ts | 87 ++++++++++++- packages/server/src/server/session.ts | 27 +++- .../session.workspace-git-watch.test.ts | 2 +- ...on.workspace-resolution-invariants.test.ts | 2 +- .../src/server/session.workspaces.test.ts | 20 +-- .../snapshot-mutation-ownership.test.ts | 2 +- .../src/server/test-utils/paseo-daemon.ts | 2 +- .../src/server/test-utils/session-stubs.ts | 8 +- .../websocket-server.notifications.test.ts | 2 +- .../websocket-server.relay-reconnect.test.ts | 20 +-- ...cket-server.terminal-notifications.test.ts | 2 +- .../server/src/server/websocket-server.ts | 21 ++-- .../server/src/server/wire-compat.test.ts | 2 +- 33 files changed, 702 insertions(+), 227 deletions(-) delete mode 100644 packages/app/src/hooks/use-push-token-registration.ts create mode 100644 packages/app/src/push-notifications/index.native.ts create mode 100644 packages/app/src/push-notifications/index.ts create mode 100644 packages/app/src/push-notifications/index.web.ts create mode 100644 packages/app/src/push-notifications/internal/subscriptions.ts create mode 100644 packages/app/src/push-notifications/internal/types.ts create mode 100644 packages/server/src/server/push/index.test.ts create mode 100644 packages/server/src/server/push/index.ts delete mode 100644 packages/server/src/server/push/notifications.ts diff --git a/packages/app/src/contexts/session-context.tsx b/packages/app/src/contexts/session-context.tsx index 8a93e9626..76eb78521 100644 --- a/packages/app/src/contexts/session-context.tsx +++ b/packages/app/src/contexts/session-context.tsx @@ -5,7 +5,7 @@ import { useQueryClient } from "@tanstack/react-query"; import { useTranslation } from "react-i18next"; import { useClientActivity } from "@/hooks/use-client-activity"; import { useAppVisible } from "@/hooks/use-app-visible"; -import { usePushTokenRegistration } from "@/hooks/use-push-token-registration"; +import { startPushNotifications } from "@/push-notifications"; import { createSetAgentInitializing, refreshAgentInitializationTimeout, @@ -418,7 +418,7 @@ function SessionProviderInternal({ children, serverId, client }: SessionProvider // Client activity tracking (heartbeat, push token registration) useClientActivity({ client, focusedAgentId, focusedTerminalId, onAppResumed: handleAppResumed }); - usePushTokenRegistration({ client, serverId }); + useEffect(() => startPushNotifications({ client, serverId }), [client, serverId]); const notifyAgentAttention = useCallback( (params: { diff --git a/packages/app/src/hooks/use-push-token-registration.ts b/packages/app/src/hooks/use-push-token-registration.ts deleted file mode 100644 index 4dd1209f9..000000000 --- a/packages/app/src/hooks/use-push-token-registration.ts +++ /dev/null @@ -1,111 +0,0 @@ -import { useCallback, useEffect, useRef } from "react"; -import { Platform } from "react-native"; -import AsyncStorage from "@react-native-async-storage/async-storage"; -import * as Notifications from "expo-notifications"; -import Constants from "expo-constants"; -import type { DaemonClient } from "@getpaseo/client/internal/daemon-client"; -import { isWeb } from "@/constants/platform"; - -const STORAGE_PREFIX = "@paseo:expo-push-token:"; - -function getExpoProjectId(): string | null { - const constants = Constants as unknown as { - easConfig?: { projectId?: unknown }; - expoConfig?: { extra?: { eas?: { projectId?: unknown } } }; - }; - const fromEas = constants?.easConfig?.projectId; - if (typeof fromEas === "string" && fromEas.trim()) return fromEas.trim(); - - const fromExtra = constants?.expoConfig?.extra?.eas?.projectId; - if (typeof fromExtra === "string" && fromExtra.trim()) return fromExtra.trim(); - - return null; -} - -async function ensurePushPermission(): Promise { - const existing = await Notifications.getPermissionsAsync(); - if (existing.status === Notifications.PermissionStatus.GRANTED) return true; - if (!existing.canAskAgain) return false; - const requested = await Notifications.requestPermissionsAsync(); - return requested.status === Notifications.PermissionStatus.GRANTED; -} - -export function usePushTokenRegistration(params: { client: DaemonClient; serverId: string }): void { - const { client, serverId } = params; - const tokenRef = useRef(null); - const lastSentTokenRef = useRef(null); - - const registerIfPossible = useCallback(async () => { - if (isWeb) return; - if (!client.isConnected) return; - const token = tokenRef.current; - if (!token) return; - if (lastSentTokenRef.current === token) return; - lastSentTokenRef.current = token; - client.registerPushToken(token); - }, [client]); - - useEffect(() => { - if (isWeb) return; - - const storageKey = `${STORAGE_PREFIX}${serverId}`; - let cancelled = false; - - const run = async () => { - const cached = await AsyncStorage.getItem(storageKey); - if (cancelled) return; - if (cached && typeof cached === "string") { - tokenRef.current = cached; - } - - const granted = await ensurePushPermission(); - if (!granted || cancelled) return; - - if (Platform.OS === "android") { - await Notifications.setNotificationChannelAsync("default", { - name: "default", - importance: Notifications.AndroidImportance.DEFAULT, - }); - } - - const projectId = getExpoProjectId(); - if (!projectId) { - console.warn("[PushToken] Missing EAS projectId; cannot fetch Expo push token"); - return; - } - - const result = await Notifications.getExpoPushTokenAsync({ projectId }); - if (cancelled) return; - - const token = result.data; - if (typeof token !== "string" || !token.trim()) return; - - tokenRef.current = token; - await AsyncStorage.setItem(storageKey, token); - await registerIfPossible(); - }; - - void run().catch((error) => { - console.warn("[PushToken] Failed to register push token", error); - }); - - return () => { - cancelled = true; - }; - }, [registerIfPossible, serverId]); - - useEffect(() => { - const unsubscribe = client.subscribeConnectionStatus((state) => { - if (state.status === "connected") { - void registerIfPossible(); - } else { - // Re-register on the next successful connect. - lastSentTokenRef.current = null; - } - }); - if (client.isConnected) { - void registerIfPossible(); - } - return unsubscribe; - }, [client, registerIfPossible]); -} diff --git a/packages/app/src/push-notifications/index.native.ts b/packages/app/src/push-notifications/index.native.ts new file mode 100644 index 000000000..579c9a073 --- /dev/null +++ b/packages/app/src/push-notifications/index.native.ts @@ -0,0 +1,12 @@ +import { revokeSubscription, startSubscription } from "./internal/subscriptions"; +import type { RevokePushNotificationsInput, StartPushNotificationsInput } from "./internal/types"; + +export function startPushNotifications(input: StartPushNotificationsInput): () => void { + return startSubscription(input); +} + +export function revokePushNotifications(input: RevokePushNotificationsInput): Promise { + return revokeSubscription(input).catch((error) => { + console.warn("[PushNotifications] Failed to remove local push subscription", error); + }); +} diff --git a/packages/app/src/push-notifications/index.ts b/packages/app/src/push-notifications/index.ts new file mode 100644 index 000000000..9aebb187d --- /dev/null +++ b/packages/app/src/push-notifications/index.ts @@ -0,0 +1,9 @@ +import type { RevokePushNotificationsInput, StartPushNotificationsInput } from "./internal/types"; + +export function startPushNotifications(_input: StartPushNotificationsInput): () => void { + return () => undefined; +} + +export async function revokePushNotifications(_input: RevokePushNotificationsInput): Promise { + // Push notifications are native-only. +} diff --git a/packages/app/src/push-notifications/index.web.ts b/packages/app/src/push-notifications/index.web.ts new file mode 100644 index 000000000..9aebb187d --- /dev/null +++ b/packages/app/src/push-notifications/index.web.ts @@ -0,0 +1,9 @@ +import type { RevokePushNotificationsInput, StartPushNotificationsInput } from "./internal/types"; + +export function startPushNotifications(_input: StartPushNotificationsInput): () => void { + return () => undefined; +} + +export async function revokePushNotifications(_input: RevokePushNotificationsInput): Promise { + // Push notifications are native-only. +} diff --git a/packages/app/src/push-notifications/internal/subscriptions.ts b/packages/app/src/push-notifications/internal/subscriptions.ts new file mode 100644 index 000000000..7a2934767 --- /dev/null +++ b/packages/app/src/push-notifications/internal/subscriptions.ts @@ -0,0 +1,103 @@ +import AsyncStorage from "@react-native-async-storage/async-storage"; +import Constants from "expo-constants"; +import * as Notifications from "expo-notifications"; +import { Platform } from "react-native"; +import type { RevokePushNotificationsInput, StartPushNotificationsInput } from "./types"; + +const STORAGE_PREFIX = "@paseo:expo-push-token:"; + +function storageKey(serverId: string): string { + return `${STORAGE_PREFIX}${serverId}`; +} + +function getExpoProjectId(): string | null { + const constants = Constants as unknown as { + easConfig?: { projectId?: unknown }; + expoConfig?: { extra?: { eas?: { projectId?: unknown } } }; + }; + const fromEas = constants.easConfig?.projectId; + if (typeof fromEas === "string" && fromEas.trim()) return fromEas.trim(); + const fromExtra = constants.expoConfig?.extra?.eas?.projectId; + return typeof fromExtra === "string" && fromExtra.trim() ? fromExtra.trim() : null; +} + +async function ensurePushPermission(): Promise { + const existing = await Notifications.getPermissionsAsync(); + if (existing.status === Notifications.PermissionStatus.GRANTED) return true; + if (!existing.canAskAgain) return false; + const requested = await Notifications.requestPermissionsAsync(); + return requested.status === Notifications.PermissionStatus.GRANTED; +} + +async function resolveToken(serverId: string): Promise { + const key = storageKey(serverId); + const cached = await AsyncStorage.getItem(key); + if (!(await ensurePushPermission())) { + await AsyncStorage.removeItem(key); + return null; + } + + if (Platform.OS === "android") { + await Notifications.setNotificationChannelAsync("default", { + name: "default", + importance: Notifications.AndroidImportance.DEFAULT, + }); + } + + const projectId = getExpoProjectId(); + if (!projectId) { + console.warn("[PushNotifications] Missing EAS projectId; cannot fetch Expo push token"); + return cached; + } + + const result = await Notifications.getExpoPushTokenAsync({ projectId }); + const token = result.data.trim(); + if (!token) return cached; + await AsyncStorage.setItem(key, token); + return token; +} + +export function startSubscription(input: StartPushNotificationsInput): () => void { + let stopped = false; + let token: string | null = null; + const register = () => { + if (!stopped && token && input.client.isConnected) { + input.client.registerPushToken(token); + } + }; + + void resolveToken(input.serverId) + .then((resolved) => { + if (stopped) return undefined; + token = resolved; + register(); + return undefined; + }) + .catch((error) => console.warn("[PushNotifications] Failed to register push token", error)); + + const unsubscribe = input.client.subscribeConnectionStatus((state) => { + if (state.status === "connected") register(); + }); + + return () => { + stopped = true; + unsubscribe(); + }; +} + +export async function revokeSubscription(input: RevokePushNotificationsInput): Promise { + const key = storageKey(input.serverId); + const token = await AsyncStorage.getItem(key); + if ( + token && + input.client?.isConnected && + input.client.getLastServerInfoMessage()?.features?.pushTokenRevocation === true + ) { + try { + await input.client.unregisterPushToken(token); + } catch (error) { + console.warn("[PushNotifications] Failed to revoke push token", error); + } + } + await AsyncStorage.removeItem(key); +} diff --git a/packages/app/src/push-notifications/internal/types.ts b/packages/app/src/push-notifications/internal/types.ts new file mode 100644 index 000000000..6256c298a --- /dev/null +++ b/packages/app/src/push-notifications/internal/types.ts @@ -0,0 +1,11 @@ +import type { DaemonClient } from "@getpaseo/client/internal/daemon-client"; + +export interface StartPushNotificationsInput { + client: DaemonClient; + serverId: string; +} + +export interface RevokePushNotificationsInput { + client: DaemonClient | null; + serverId: string; +} diff --git a/packages/app/src/runtime/host-runtime.test.ts b/packages/app/src/runtime/host-runtime.test.ts index 697872f09..1fe16b72c 100644 --- a/packages/app/src/runtime/host-runtime.test.ts +++ b/packages/app/src/runtime/host-runtime.test.ts @@ -1384,6 +1384,54 @@ describe("HostRuntimeController", () => { }); describe("HostRuntimeStore", () => { + it("revokes push notifications before removing a host", async () => { + const host = makeHost({ connections: [makeHost().connections[0]!] }); + const revocation = createDeferred(); + const storage = createMemoryHostRuntimeStorage({ + "@paseo:daemon-registry": JSON.stringify([host]), + "@paseo:e2e": "1", + }); + const store = new HostRuntimeStore({ + storage, + deps: makeDeps({}, []), + revokePushNotifications: async ({ serverId }) => { + expect(serverId).toBe(host.serverId); + expect(store.getHosts().map((candidate) => candidate.serverId)).toEqual([host.serverId]); + await revocation.promise; + }, + }); + await store.boot(); + + const removal = store.removeHost(host.serverId); + expect(store.getHosts().map((candidate) => candidate.serverId)).toEqual([host.serverId]); + revocation.resolve(); + await removal; + + expect(store.getHosts()).toEqual([]); + }); + + it("revokes push notifications when removing a host's final connection", async () => { + const host = makeHost({ connections: [makeHost().connections[0]!] }); + const revokedServerIds: string[] = []; + const storage = createMemoryHostRuntimeStorage({ + "@paseo:daemon-registry": JSON.stringify([host]), + "@paseo:e2e": "1", + }); + const store = new HostRuntimeStore({ + storage, + deps: makeDeps({}, []), + revokePushNotifications: async ({ serverId }) => { + revokedServerIds.push(serverId); + }, + }); + await store.boot(); + + await store.removeConnection(host.serverId, host.connections[0]!.id); + + expect(revokedServerIds).toEqual([host.serverId]); + expect(store.getHosts()).toEqual([]); + }); + it("restores the display replica before declaring the host registry loaded", async () => { const host = makeHost(); const storage = createMemoryHostRuntimeStorage(); diff --git a/packages/app/src/runtime/host-runtime.ts b/packages/app/src/runtime/host-runtime.ts index 982d411e4..5bd205074 100644 --- a/packages/app/src/runtime/host-runtime.ts +++ b/packages/app/src/runtime/host-runtime.ts @@ -57,6 +57,7 @@ import { encodeImages } from "@/utils/encode-images"; import { DirectorySync, type RefreshAgentDirectoryResult } from "@/runtime/directory-sync"; import { ReplicaCache } from "@/runtime/replica-cache"; import { nativePerformanceTrace } from "@/performance/native-trace"; +import { revokePushNotifications } from "@/push-notifications"; import { createAppWebSocketFactory } from "./websocket-factory"; export type HostRuntimeConnectionStatus = "idle" | "connecting" | "online" | "offline" | "error"; @@ -1366,11 +1367,17 @@ export class HostRuntimeStore { private bootPromise: Promise | null = null; private storage: HostRuntimeStorage; private replicaCache: ReplicaCache; + private readonly revokePushNotifications: typeof revokePushNotifications; - constructor(input?: { deps?: HostRuntimeControllerDeps; storage?: HostRuntimeStorage }) { + constructor(input?: { + deps?: HostRuntimeControllerDeps; + storage?: HostRuntimeStorage; + revokePushNotifications?: typeof revokePushNotifications; + }) { this.deps = input?.deps ?? createDefaultDeps(); this.storage = input?.storage ?? AsyncStorage; this.replicaCache = new ReplicaCache(this.storage); + this.revokePushNotifications = input?.revokePushNotifications ?? revokePushNotifications; } // --- Host registry --- @@ -1849,12 +1856,18 @@ export class HostRuntimeStore { } async removeHost(serverId: string): Promise { + await this.revokePushNotifications({ client: this.getClient(serverId), serverId }); const remaining = this.hosts.filter((daemon) => daemon.serverId !== serverId); this.setHostsAndSync(remaining); await this.persistHosts(); } async removeConnection(serverId: string, connectionId: string): Promise { + const host = this.hosts.find((candidate) => candidate.serverId === serverId); + if (host?.connections.length === 1 && host.connections[0]?.id === connectionId) { + await this.removeHost(serverId); + return; + } const now = new Date().toISOString(); const next = this.hosts .map((daemon) => { diff --git a/packages/client/src/daemon-client.test.ts b/packages/client/src/daemon-client.test.ts index d07be4c15..9739e04a5 100644 --- a/packages/client/src/daemon-client.test.ts +++ b/packages/client/src/daemon-client.test.ts @@ -967,6 +967,59 @@ test("keeps the transport connected when a session RPC ping times out", async () expect(client.getConnectionState().status).toBe("connected"); }); +test("waits for the daemon to acknowledge push token revocation", async () => { + const mock = createMockTransport(); + const client = new DaemonClient({ + url: "ws://test", + clientId: "clsk_push_revocation", + reconnect: { enabled: false }, + transportFactory: () => mock.transport, + }); + clients.push(client); + const connectPromise = client.connect(); + mock.triggerOpen({ features: { pushTokenRevocation: true } }); + await connectPromise; + + const revocation = client.unregisterPushToken("ExponentPushToken[test-device]"); + const request = parseSentFrame(mock.sent.at(-1)); + expect(request).toMatchObject({ + type: "push.unregister.request", + token: "ExponentPushToken[test-device]", + }); + mock.triggerMessage( + wrapSessionMessage({ + type: "push.unregister.response", + payload: { requestId: request.requestId }, + }), + ); + + await revocation; +}); + +test("bounds the wait for push token revocation", async () => { + useHeartbeatClock(); + const mock = createMockTransport(); + const client = new DaemonClient({ + url: "ws://test", + clientId: "clsk_push_revocation_timeout", + reconnect: { enabled: false }, + transportFactory: () => mock.transport, + }); + clients.push(client); + const connectPromise = client.connect(); + mock.triggerOpen({ features: { pushTokenRevocation: true } }); + await connectPromise; + + const revocation = client.unregisterPushToken("ExponentPushToken[test-device]"); + const rejection = expect(revocation).rejects.toThrow("Timeout waiting for message (2000ms)"); + await vi.advanceTimersByTimeAsync(1_999); + expect(client.getConnectionState().status).toBe("connected"); + + await vi.advanceTimersByTimeAsync(1); + await rejection; + expect(client.getConnectionState().status).toBe("connected"); +}); + test("defaults session RPC waiters to sixty seconds", async () => { useHeartbeatClock(); const logger = createMockLogger(); diff --git a/packages/client/src/daemon-client.ts b/packages/client/src/daemon-client.ts index bcdf143fb..d078d8473 100644 --- a/packages/client/src/daemon-client.ts +++ b/packages/client/src/daemon-client.ts @@ -876,6 +876,7 @@ function toTimeoutError(error: unknown, label: string, timeoutMs: number): Error const DEFAULT_RECONNECT_BASE_DELAY_MS = 1500; const DEFAULT_RECONNECT_MAX_DELAY_MS = 30000; const DEFAULT_SESSION_RPC_TIMEOUT_MS = 60_000; +const PUSH_TOKEN_REVOCATION_TIMEOUT_MS = 2_000; const DEFAULT_CONNECT_TIMEOUT_MS = 15_000; const DEFAULT_LIVENESS_TIMEOUT_MS = 5000; const LIVENESS_HEARTBEAT_INTERVAL_MS = 10_000; @@ -1840,6 +1841,16 @@ export class DaemonClient { }); } + async unregisterPushToken(token: string): Promise { + const requestId = this.createRequestId(); + await this.sendCorrelatedSessionRequest({ + requestId, + message: { type: "push.unregister.request", token, requestId }, + responseType: "push.unregister.response", + timeout: PUSH_TOKEN_REVOCATION_TIMEOUT_MS, + }); + } + async ping(params?: { requestId?: string; timeoutMs?: number }): Promise<{ requestId: string; clientSentAt: number; diff --git a/packages/protocol/src/messages.ts b/packages/protocol/src/messages.ts index aa297cc94..5cf5f5099 100644 --- a/packages/protocol/src/messages.ts +++ b/packages/protocol/src/messages.ts @@ -2409,6 +2409,19 @@ export const RegisterPushTokenMessageSchema = z.object({ token: z.string(), }); +export const PushUnregisterRequestSchema = z.object({ + type: z.literal("push.unregister.request"), + token: z.string(), + requestId: z.string(), +}); + +export const PushUnregisterResponseSchema = z.object({ + type: z.literal("push.unregister.response"), + payload: z.object({ + requestId: z.string(), + }), +}); + // ============================================================================ // Terminal Messages // ============================================================================ @@ -2753,6 +2766,7 @@ export const SessionInboundMessageSchema = z.discriminatedUnion("type", [ PingMessageSchema, ListCommandsRequestSchema, RegisterPushTokenMessageSchema, + PushUnregisterRequestSchema, ListTerminalsRequestSchema, SubscribeTerminalsRequestSchema, UnsubscribeTerminalsRequestSchema, @@ -2971,6 +2985,8 @@ export const ServerInfoStatusPayloadSchema = z daemonStatusRpc: z.boolean().optional(), // COMPAT(relayConfig): added in v0.2.6, remove gate after 2027-01-31. relayConfig: z.boolean().optional(), + // COMPAT(pushTokenRevocation): added in v0.3.2, remove gate after 2027-02-10. + pushTokenRevocation: z.boolean().optional(), // COMPAT(terminalRestoreModes): added in v0.1.81, remove gate after 2026-11-23. "terminal-restore-modes": z.boolean().optional(), // COMPAT(terminalInputModeReplay): added in v0.2.6, remove gate after 2027-02-02. @@ -5575,6 +5591,7 @@ export const SessionOutboundMessageSchema = z.discriminatedUnion("type", [ DictationStreamErrorMessageSchema, StatusMessageSchema, PongMessageSchema, + PushUnregisterResponseSchema, RpcErrorMessageSchema, ArtifactMessageSchema, AgentUpdateMessageSchema, @@ -6143,6 +6160,8 @@ export type ClientHeartbeatMessage = z.infer; export type ListCommandsResponse = z.infer; export type RegisterPushTokenMessage = z.infer; +export type PushUnregisterRequest = z.infer; +export type PushUnregisterResponse = z.infer; // Terminal message types export type ListTerminalsRequest = z.infer; diff --git a/packages/server/src/server/bootstrap.ts b/packages/server/src/server/bootstrap.ts index 0321d3295..e5f12972c 100644 --- a/packages/server/src/server/bootstrap.ts +++ b/packages/server/src/server/bootstrap.ts @@ -164,7 +164,7 @@ import { createConfiguredTerminalManager } from "../terminal/terminal-manager-fa import { applyTerminalAgentHookSetting } from "../terminal/agent-hooks/terminal-agent-hook-setting.js"; import { loadOrCreateDaemonKeyPair } from "./daemon-keypair.js"; import { createRelayRuntime, type RelayRuntime } from "./relay-runtime.js"; -import type { PushNotificationSender } from "./push/notifications.js"; +import type { PushNotificationSender } from "./push/index.js"; import { getOrCreateServerId } from "./server-id.js"; import { resolveDaemonVersion } from "./daemon-version.js"; import type { AgentClient, AgentProvider } from "./agent/agent-sdk-types.js"; diff --git a/packages/server/src/server/client-activity.e2e.test.ts b/packages/server/src/server/client-activity.e2e.test.ts index eaf5ae054..5348ea44d 100644 --- a/packages/server/src/server/client-activity.e2e.test.ts +++ b/packages/server/src/server/client-activity.e2e.test.ts @@ -3,7 +3,7 @@ import { createTestPaseoDaemon, type TestPaseoDaemon } from "./test-utils/paseo- import { DaemonClient } from "./test-utils/daemon-client.js"; import type { AgentStreamEventPayload } from "@getpaseo/protocol/messages"; import type { AgentSnapshotPayload } from "./messages.js"; -import type { PushNotificationSender, PushPayload } from "./push/notifications.js"; +import type { PushNotificationSender, PushPayload } from "./push/index.js"; import { PRESENCE_THRESHOLD_MS } from "./agent-attention-policy.js"; class RecordingPushNotificationSender implements PushNotificationSender { diff --git a/packages/server/src/server/push/index.test.ts b/packages/server/src/server/push/index.test.ts new file mode 100644 index 000000000..1cd4c1baa --- /dev/null +++ b/packages/server/src/server/push/index.test.ts @@ -0,0 +1,67 @@ +import { mkdtempSync, readFileSync, rmSync } from "node:fs"; +import { tmpdir } from "node:os"; +import path from "node:path"; +import type pino from "pino"; +import { afterEach, describe, expect, test } from "vitest"; + +import { createPushNotifications } from "./index.js"; + +function createLogger(): pino.Logger { + const logger = { + child: () => logger, + debug: () => undefined, + info: () => undefined, + warn: () => undefined, + error: () => undefined, + }; + return logger as unknown as pino.Logger; +} + +describe("push notifications", () => { + const homes: string[] = []; + + afterEach(() => { + for (const home of homes.splice(0)) { + rmSync(home, { recursive: true, force: true }); + } + }); + + test("an offline device stops receiving notifications after 48 hours", async () => { + const home = mkdtempSync(path.join(tmpdir(), "paseo-push-notifications-")); + homes.push(home); + const filePath = path.join(home, "push-tokens.json"); + let now = Date.parse("2026-08-10T00:00:00.000Z"); + const deliveries: string[][] = []; + const pushNotifications = createPushNotifications({ + logger: createLogger(), + filePath, + now: () => now, + deliver: async (tokens) => deliveries.push(tokens), + }); + + pushNotifications.renew("ExponentPushToken[offline-device]"); + now += 48 * 60 * 60 * 1000; + await pushNotifications.send({ title: "Agent finished", body: "Done" }); + + expect(deliveries).toEqual([]); + expect(JSON.parse(readFileSync(filePath, "utf8"))).toEqual({ subscriptions: [] }); + }); + + test("online revocation stops notifications immediately", async () => { + const home = mkdtempSync(path.join(tmpdir(), "paseo-push-notifications-")); + homes.push(home); + const deliveries: string[][] = []; + const pushNotifications = createPushNotifications({ + logger: createLogger(), + filePath: path.join(home, "push-tokens.json"), + now: () => Date.parse("2026-08-10T00:00:00.000Z"), + deliver: async (tokens) => deliveries.push(tokens), + }); + + pushNotifications.renew("ExponentPushToken[online-device]"); + pushNotifications.revoke("ExponentPushToken[online-device]"); + await pushNotifications.send({ title: "Agent finished", body: "Done" }); + + expect(deliveries).toEqual([]); + }); +}); diff --git a/packages/server/src/server/push/index.ts b/packages/server/src/server/push/index.ts new file mode 100644 index 000000000..3f1a8e090 --- /dev/null +++ b/packages/server/src/server/push/index.ts @@ -0,0 +1,45 @@ +import type pino from "pino"; + +import { PushService, type PushPayload } from "./push-service.js"; +import { PushTokenStore } from "./token-store.js"; + +export type { PushPayload }; + +const PUSH_TOKEN_LEASE_MS = 48 * 60 * 60 * 1000; + +export interface PushNotifications { + renew(token: string): void; + revoke(token: string): void; + send(payload: PushPayload): Promise; +} + +export type PushNotificationSender = Pick; + +export function createPushNotifications(options: { + logger: pino.Logger; + filePath: string; + now?: () => number; + deliver?: (tokens: string[], payload: PushPayload) => Promise; +}): PushNotifications { + const now = options.now ?? Date.now; + const store = new PushTokenStore(options.logger, options.filePath, now, PUSH_TOKEN_LEASE_MS); + const service = new PushService(options.logger, (token) => store.revokeToken(token)); + const deliver = + options.deliver ?? + ((tokens: string[], payload: PushPayload) => service.sendPush(tokens, payload)); + + return { + renew(token) { + store.renewToken(token); + }, + revoke(token) { + store.revokeToken(token); + }, + async send(payload) { + const tokens = store.getActiveTokens(); + options.logger.info({ tokenCount: tokens.length }, "Sending push notification"); + if (tokens.length === 0) return; + await deliver(tokens, payload); + }, + }; +} diff --git a/packages/server/src/server/push/notifications.ts b/packages/server/src/server/push/notifications.ts deleted file mode 100644 index 07c3a5bcd..000000000 --- a/packages/server/src/server/push/notifications.ts +++ /dev/null @@ -1,29 +0,0 @@ -import type pino from "pino"; - -import { PushService, type PushPayload } from "./push-service.js"; -import type { PushTokenStore } from "./token-store.js"; - -export type { PushPayload }; - -export interface PushNotificationSender { - send(payload: PushPayload): Promise; -} - -export function createPushNotificationSender( - logger: pino.Logger, - tokenStore: PushTokenStore, -): PushNotificationSender { - const pushService = new PushService(logger, tokenStore); - - return { - async send(payload) { - const tokens = tokenStore.getAllTokens(); - logger.info({ tokenCount: tokens.length }, "Sending push notification"); - if (tokens.length === 0) { - return; - } - - await pushService.sendPush(tokens, payload); - }, - }; -} diff --git a/packages/server/src/server/push/push-service.ts b/packages/server/src/server/push/push-service.ts index dcfadc122..b91de5fbd 100644 --- a/packages/server/src/server/push/push-service.ts +++ b/packages/server/src/server/push/push-service.ts @@ -1,4 +1,3 @@ -import type { PushTokenStore } from "./token-store.js"; import type pino from "pino"; export interface PushPayload { @@ -31,11 +30,11 @@ const MAX_BATCH_SIZE = 100; */ export class PushService { private readonly logger: pino.Logger; - private readonly tokenStore: PushTokenStore; + private readonly revokeToken: (token: string) => void; - constructor(logger: pino.Logger, tokenStore: PushTokenStore) { + constructor(logger: pino.Logger, revokeToken: (token: string) => void) { this.logger = logger.child({ component: "push-service" }); - this.tokenStore = tokenStore; + this.revokeToken = revokeToken; } async sendPush(tokens: string[], payload: PushPayload): Promise { @@ -102,7 +101,7 @@ export class PushService { ticket.details?.error === "DeviceNotRegistered" || ticket.details?.error === "InvalidCredentials" ) { - this.tokenStore.removeToken(message.to); + this.revokeToken(message.to); } } } diff --git a/packages/server/src/server/push/token-store.test.ts b/packages/server/src/server/push/token-store.test.ts index b5c66e10c..afd7d601c 100644 --- a/packages/server/src/server/push/token-store.test.ts +++ b/packages/server/src/server/push/token-store.test.ts @@ -5,6 +5,7 @@ import type pino from "pino"; import { describe, expect, test } from "vitest"; import { PRIVATE_FILE_MODE } from "../private-files.js"; +import { createPushNotifications } from "./index.js"; import { PushTokenStore } from "./token-store.js"; const MODE_MASK = 0o777; @@ -16,6 +17,7 @@ function createLogger(): pino.Logger { debug: () => undefined, info: () => undefined, warn: () => undefined, + error: () => undefined, }; return logger as unknown as pino.Logger; } @@ -29,9 +31,12 @@ describe.skipIf(process.platform === "win32")("PushTokenStore file permissions", const home = mkdtempSync(path.join(tmpdir(), "paseo-push-tokens-")); const tokenPath = path.join(home, "push-tokens.json"); try { - const store = new PushTokenStore(createLogger(), tokenPath); + const pushNotifications = createPushNotifications({ + logger: createLogger(), + filePath: tokenPath, + }); - store.addToken("ExponentPushToken[test]"); + pushNotifications.renew("ExponentPushToken[test]"); expect(modeOf(tokenPath)).toBe(PRIVATE_FILE_MODE); } finally { @@ -39,19 +44,61 @@ describe.skipIf(process.platform === "win32")("PushTokenStore file permissions", } }); - test("repairs existing push token file permissions when loading", () => { + test("repairs existing push token file permissions when loading", async () => { const home = mkdtempSync(path.join(tmpdir(), "paseo-push-tokens-")); const tokenPath = path.join(home, "push-tokens.json"); try { writeFileSync(tokenPath, JSON.stringify({ tokens: ["ExponentPushToken[test]"] })); chmodSync(tokenPath, PERMISSIVE_FILE_MODE); - const store = new PushTokenStore(createLogger(), tokenPath); + const deliveries: string[][] = []; + const pushNotifications = createPushNotifications({ + logger: createLogger(), + filePath: tokenPath, + deliver: async (tokens) => deliveries.push(tokens), + }); + await pushNotifications.send({ title: "Agent finished", body: "Done" }); - expect(store.getAllTokens()).toEqual(["ExponentPushToken[test]"]); + expect(deliveries).toEqual([["ExponentPushToken[test]"]]); expect(modeOf(tokenPath)).toBe(PRIVATE_FILE_MODE); } finally { rmSync(home, { recursive: true, force: true }); } }); }); + +describe("PushTokenStore persistence", () => { + test("does not apply a revocation until the updated subscriptions are durable", () => { + let rejectWrites = false; + let persisted = ""; + const store = new PushTokenStore( + createLogger(), + path.join(tmpdir(), `missing-push-token-store-${process.pid}`, "push-tokens.json"), + () => Date.parse("2026-08-10T00:00:00.000Z"), + 48 * 60 * 60 * 1000, + (_filePath, payload) => { + if (rejectWrites) throw new Error("disk full"); + persisted = String(payload); + }, + ); + + store.renewToken("ExponentPushToken[test]"); + rejectWrites = true; + + expect(() => store.revokeToken("ExponentPushToken[test]")).toThrow("disk full"); + expect(store.getActiveTokens()).toEqual(["ExponentPushToken[test]"]); + expect(JSON.parse(persisted)).toEqual({ + subscriptions: [ + { + token: "ExponentPushToken[test]", + expiresAt: "2026-08-12T00:00:00.000Z", + }, + ], + }); + + rejectWrites = false; + store.revokeToken("ExponentPushToken[test]"); + expect(store.getActiveTokens()).toEqual([]); + expect(JSON.parse(persisted)).toEqual({ subscriptions: [] }); + }); +}); diff --git a/packages/server/src/server/push/token-store.ts b/packages/server/src/server/push/token-store.ts index 438c87af2..540f05313 100644 --- a/packages/server/src/server/push/token-store.ts +++ b/packages/server/src/server/push/token-store.ts @@ -10,36 +10,69 @@ import { ensurePrivateFile, writePrivateFileAtomicSync } from "../private-files. */ export class PushTokenStore { private readonly logger: pino.Logger; - private tokens: Set = new Set(); + private subscriptions = new Map(); private readonly filePath: string; + private readonly now: () => number; + private readonly leaseMs: number; + private readonly write: typeof writePrivateFileAtomicSync; - constructor(logger: pino.Logger, filePath: string) { + constructor( + logger: pino.Logger, + filePath: string, + now: () => number, + leaseMs: number, + write: typeof writePrivateFileAtomicSync = writePrivateFileAtomicSync, + ) { this.logger = logger.child({ component: "token-store" }); this.filePath = filePath; + this.now = now; + this.leaseMs = leaseMs; + this.write = write; this.loadFromDisk(); } - addToken(token: string): void { + renewToken(token: string): void { const normalized = token.trim(); if (!normalized) return; - if (this.tokens.has(normalized)) return; - this.tokens.add(normalized); - this.persist(); - this.logger.debug({ total: this.tokens.size }, "Added token"); + const now = this.now(); + const currentExpiry = this.subscriptions.get(normalized); + if (currentExpiry !== undefined && currentExpiry - now > this.leaseMs / 2) return; + const next = new Map(this.subscriptions); + next.set(normalized, now + this.leaseMs); + this.persist(next); + this.subscriptions = next; + this.logger.debug({ total: this.subscriptions.size }, "Renewed token"); } - removeToken(token: string): void { + revokeToken(token: string): void { const normalized = token.trim(); if (!normalized) return; - const deleted = this.tokens.delete(normalized); - if (deleted) { - this.persist(); - this.logger.debug({ total: this.tokens.size }, "Removed token"); + if (!this.subscriptions.has(normalized)) return; + const next = new Map(this.subscriptions); + next.delete(normalized); + this.persist(next); + this.subscriptions = next; + this.logger.debug({ total: this.subscriptions.size }, "Revoked token"); + } + + getActiveTokens(): string[] { + const now = this.now(); + const active = new Map(this.subscriptions); + for (const [token, expiresAt] of this.subscriptions) { + if (expiresAt <= now) { + active.delete(token); + } } - } - - getAllTokens(): string[] { - return Array.from(this.tokens); + if (active.size !== this.subscriptions.size) { + try { + this.persist(active); + this.subscriptions = active; + } catch { + // Keep the previous state so a later send retries pruning. Expired tokens + // are still excluded from this delivery. + } + } + return Array.from(active.keys()); } private loadFromDisk(): void { @@ -49,25 +82,60 @@ export class PushTokenStore { } ensurePrivateFile(this.filePath); const raw = readFileSync(this.filePath, "utf-8"); - const parsed = JSON.parse(raw) as { tokens?: unknown }; - const tokens = Array.isArray(parsed.tokens) - ? parsed.tokens.filter((t): t is string => typeof t === "string" && t.trim().length > 0) + const parsed = JSON.parse(raw) as { subscriptions?: unknown; tokens?: unknown }; + const loaded = new Map(); + const subscriptions = Array.isArray(parsed.subscriptions) ? parsed.subscriptions : []; + for (const value of subscriptions) { + if (!value || typeof value !== "object") continue; + const candidate = value as { token?: unknown; expiresAt?: unknown }; + if (typeof candidate.token !== "string" || typeof candidate.expiresAt !== "string") + continue; + const token = candidate.token.trim(); + const expiresAt = Date.parse(candidate.expiresAt); + if (token && Number.isFinite(expiresAt)) { + loaded.set(token, expiresAt); + } + } + this.subscriptions = loaded; + + const legacyTokens = Array.isArray(parsed.tokens) + ? parsed.tokens.filter((token): token is string => typeof token === "string") : []; - this.tokens = new Set(tokens.map((t) => t.trim())); - this.logger.info({ total: this.tokens.size }, "Loaded push tokens"); + if (legacyTokens.length > 0) { + const migrated = new Map(loaded); + const expiresAt = this.now() + this.leaseMs; + for (const token of legacyTokens) { + const normalized = token.trim(); + if (normalized) migrated.set(normalized, expiresAt); + } + this.persist(migrated); + this.subscriptions = migrated; + } + this.logger.info({ total: this.subscriptions.size }, "Loaded push tokens"); } catch (error) { const err = error instanceof Error ? error : new Error(String(error)); this.logger.warn({ err }, "Failed to load push tokens"); } } - private persist(): void { + private persist(subscriptions: ReadonlyMap): void { try { - const payload = JSON.stringify({ tokens: Array.from(this.tokens) }, null, 2) + "\n"; - writePrivateFileAtomicSync(this.filePath, payload); + const payload = + JSON.stringify( + { + subscriptions: Array.from(subscriptions, ([token, expiresAt]) => ({ + token, + expiresAt: new Date(expiresAt).toISOString(), + })), + }, + null, + 2, + ) + "\n"; + this.write(this.filePath, payload); } catch (error) { const err = error instanceof Error ? error : new Error(String(error)); this.logger.warn({ err }, "Failed to persist push tokens"); + throw err; } } } diff --git a/packages/server/src/server/session.test.ts b/packages/server/src/server/session.test.ts index a648bf1b7..81312a757 100644 --- a/packages/server/src/server/session.test.ts +++ b/packages/server/src/server/session.test.ts @@ -33,7 +33,7 @@ import { asAgentManager, asAgentStorage, asDownloadTokenStore, - asPushTokenStore, + asPushNotifications, asScheduleService, asCheckoutDiffManager, asGitHubService, @@ -316,6 +316,7 @@ interface SessionForTestOptions { daemonVersion?: SessionOptions["daemonVersion"]; daemonRuntimeConfig?: SessionOptions["daemonRuntimeConfig"]; downloadTokenStore?: SessionOptions["downloadTokenStore"]; + pushNotifications?: SessionOptions["pushNotifications"]; messages?: unknown[]; targetedMessages?: Array<{ source: object; message: SessionOutboundMessage }>; binaryMessages?: Uint8Array[]; @@ -365,7 +366,7 @@ function createSessionForTest(options: SessionForTestOptions = {}): Session { onBinaryMessage: createBinaryMessageHandler(options.binaryMessages), logger, downloadTokenStore: options.downloadTokenStore ?? asDownloadTokenStore(), - pushTokenStore: asPushTokenStore(), + pushNotifications: options.pushNotifications ?? asPushNotifications(), paseoHome: options.paseoHome ?? "/tmp/paseo-home", agentManager: asAgentManager({ listAgents: vi.fn(() => []), @@ -1443,6 +1444,88 @@ describe("project config RPC authorization", () => { }); }); +test("push token registration can be revoked by the connected client", async () => { + const renewed: string[] = []; + const revoked: string[] = []; + const messages: unknown[] = []; + const session = createSessionForTest({ + messages, + pushNotifications: asPushNotifications({ + renew: (token: string) => renewed.push(token), + revoke: (token: string) => revoked.push(token), + }), + }); + + await session.handleMessage({ + type: "register_push_token", + token: "ExponentPushToken[test-device]", + }); + await session.handleMessage({ + type: "push.unregister.request", + token: "ExponentPushToken[test-device]", + requestId: "revoke-1", + }); + await session.handleMessage({ + type: "client_heartbeat", + deviceType: "mobile", + focusedAgentId: null, + lastActivityAt: "2026-08-10T00:00:00.000Z", + appVisible: false, + }); + + expect(renewed).toEqual(["ExponentPushToken[test-device]"]); + expect(revoked).toEqual(["ExponentPushToken[test-device]"]); + expect(messages).toEqual([ + { + type: "push.unregister.response", + payload: { requestId: "revoke-1" }, + }, + ]); +}); + +test("push token revocation only acknowledges durable removal", async () => { + const renewed: string[] = []; + const messages: SessionOutboundMessage[] = []; + const session = createSessionForTest({ + messages, + pushNotifications: asPushNotifications({ + renew: (token: string) => renewed.push(token), + revoke: () => { + throw new Error("disk full"); + }, + }), + }); + + await session.handleMessage({ + type: "register_push_token", + token: "ExponentPushToken[test-device]", + }); + await session.handleMessage({ + type: "push.unregister.request", + token: "ExponentPushToken[test-device]", + requestId: "revoke-failed", + }); + await session.handleMessage({ + type: "client_heartbeat", + deviceType: "mobile", + focusedAgentId: null, + lastActivityAt: "2026-08-10T00:00:00.000Z", + appVisible: false, + }); + + expect(renewed).toEqual(["ExponentPushToken[test-device]", "ExponentPushToken[test-device]"]); + expect(messages.some((message) => message.type === "push.unregister.response")).toBe(false); + expect(messages).toContainEqual({ + type: "rpc_error", + payload: { + requestId: "revoke-failed", + requestType: "push.unregister.request", + error: "Request failed: disk full", + code: "handler_error", + }, + }); +}); + describe("daemon status + pairing RPC", () => { const tempDirs: string[] = []; diff --git a/packages/server/src/server/session.ts b/packages/server/src/server/session.ts index 17ef78add..3f47aeb3b 100644 --- a/packages/server/src/server/session.ts +++ b/packages/server/src/server/session.ts @@ -169,7 +169,7 @@ import type { HubRelationshipManagement } from "./hub/relationship-controller.js import { HubExecutionController } from "./hub/execution-controller.js"; import type { HubExecutionAgents } from "./hub/daemon-executions.js"; import { DownloadTokenStore } from "./file-download/token-store.js"; -import { PushTokenStore } from "./push/token-store.js"; +import type { PushNotifications } from "./push/index.js"; import { archivePersistedWorkspaceRecord, archiveWorkspaceContents, @@ -439,7 +439,7 @@ export interface SessionOptions { onWorkspaceRecovered?: (workspace: PersistedWorkspaceRecord) => Promise; logger: pino.Logger; downloadTokenStore: DownloadTokenStore; - pushTokenStore: PushTokenStore; + pushNotifications: PushNotifications; paseoHome: string; worktreesRoot?: string; agentManager: AgentManager; @@ -627,7 +627,7 @@ export class Session { private readonly workspaceProvisioning: WorkspaceProvisioningService; private readonly workspaceRecovery: WorkspaceRecoveryService; private readonly daemonConfigStore: DaemonConfigStore; - private readonly pushTokenStore: PushTokenStore; + private readonly pushNotifications: PushNotifications; private unsubscribeAgentEvents: (() => void) | null = null; private unsubscribeProjectMutations: (() => void) | null = null; private unsubscribeWorkspaceMutations: (() => void) | null = null; @@ -649,6 +649,7 @@ export class Session { appVisible: boolean; appVisibilityChangedAt: Date; } | null = null; + private registeredPushToken: string | null = null; private readonly terminalManager: TerminalManager | null; private readonly providerSnapshotManager: ProviderSnapshotManager; private readonly serviceProxy: ServiceProxySubsystem | null; @@ -691,7 +692,7 @@ export class Session { onWorkspaceRecovered, logger, downloadTokenStore, - pushTokenStore, + pushNotifications, paseoHome, worktreesRoot, agentManager, @@ -741,7 +742,7 @@ export class Session { this.getTransportBufferedAmount = getTransportBufferedAmount ?? (() => 0); this.onLifecycleIntent = onLifecycleIntent ?? null; this.onWorkspaceRecovered = onWorkspaceRecovered ?? null; - this.pushTokenStore = pushTokenStore; + this.pushNotifications = pushNotifications; this.paseoHome = paseoHome; this.worktreesRoot = worktreesRoot; this.sessionLogger = logger.child({ @@ -2277,6 +2278,16 @@ export class Session { case "register_push_token": this.handleRegisterPushToken(msg.token); return; + case "push.unregister.request": + this.pushNotifications.revoke(msg.token); + if (this.registeredPushToken?.trim() === msg.token.trim()) { + this.registeredPushToken = null; + } + this.emit({ + type: "push.unregister.response", + payload: { requestId: msg.requestId }, + }); + return; } } @@ -3722,6 +3733,9 @@ export class Session { if (msg.appVisible && focusedTerminalId) { void this.clearFocusedTerminalAttention(focusedTerminalId); } + if (this.registeredPushToken) { + this.pushNotifications.renew(this.registeredPushToken); + } } private async clearFocusedTerminalAttention(terminalId: string): Promise { @@ -3740,7 +3754,8 @@ export class Session { * Handle push token registration */ private handleRegisterPushToken(token: string): void { - this.pushTokenStore.addToken(token); + this.registeredPushToken = token; + this.pushNotifications.renew(token); this.sessionLogger.info("Registered push token"); } diff --git a/packages/server/src/server/session.workspace-git-watch.test.ts b/packages/server/src/server/session.workspace-git-watch.test.ts index c038559fc..f3ae292f3 100644 --- a/packages/server/src/server/session.workspace-git-watch.test.ts +++ b/packages/server/src/server/session.workspace-git-watch.test.ts @@ -195,7 +195,7 @@ function createSessionForWorkspaceGitWatchTests(options?: { onMessage: (message) => emitted.push(message as { type: string; payload: unknown }), logger: createStub(logger), downloadTokenStore: createStub({}), - pushTokenStore: createStub({}), + pushNotifications: createStub({}), paseoHome: "/tmp/paseo-test", agentManager: createStub({ subscribe: () => () => {}, diff --git a/packages/server/src/server/session.workspace-resolution-invariants.test.ts b/packages/server/src/server/session.workspace-resolution-invariants.test.ts index bdf71e9de..911890fdc 100644 --- a/packages/server/src/server/session.workspace-resolution-invariants.test.ts +++ b/packages/server/src/server/session.workspace-resolution-invariants.test.ts @@ -93,7 +93,7 @@ function createHarness(input: { onMessage: (m) => emitted.push(m), logger: createStub(logger), downloadTokenStore: createStub({}), - pushTokenStore: createStub({}), + pushNotifications: createStub({}), paseoHome: mkdtempSync(path.join(tmpdir(), "paseo-invariant-test-")), agentManager: createStub({ subscribe: () => () => {}, diff --git a/packages/server/src/server/session.workspaces.test.ts b/packages/server/src/server/session.workspaces.test.ts index 53597b442..572680f9a 100644 --- a/packages/server/src/server/session.workspaces.test.ts +++ b/packages/server/src/server/session.workspaces.test.ts @@ -53,7 +53,7 @@ import { asAgentManager, asAgentStorage, asDownloadTokenStore, - asPushTokenStore, + asPushNotifications, asScheduleService, asCheckoutDiffManager, asDaemonConfigStore, @@ -639,7 +639,7 @@ function createSessionForWorkspaceTests( onWorkspaceRecovered: options.onWorkspaceRecovered, logger: asSessionLogger(logger), downloadTokenStore: asDownloadTokenStore(), - pushTokenStore: asPushTokenStore(), + pushNotifications: asPushNotifications(), paseoHome: options.paseoHome ?? "/tmp/paseo-test", worktreesRoot: options.worktreesRoot, agentManager, @@ -946,7 +946,7 @@ test("create_agent_request keeps requested child cwd when grouped under an exist onMessage: (message) => emitted.push(message), logger: asSessionLogger(logger), downloadTokenStore: asDownloadTokenStore(), - pushTokenStore: asPushTokenStore(), + pushNotifications: asPushNotifications(), paseoHome: path.join(workdir, "paseo-home"), agentManager, agentStorage, @@ -1098,7 +1098,7 @@ test("create_agent_request launches from an exact subdirectory in a created work onMessage: (message) => emitted.push(message), logger: asSessionLogger(logger), downloadTokenStore: asDownloadTokenStore(), - pushTokenStore: asPushTokenStore(), + pushNotifications: asPushNotifications(), paseoHome: path.join(workdir, "paseo-home"), agentManager, agentStorage, @@ -1235,7 +1235,7 @@ test("create_agent_request does not title an existing workspace from the agent p onMessage: vi.fn(), logger: asSessionLogger(logger), downloadTokenStore: asDownloadTokenStore(), - pushTokenStore: asPushTokenStore(), + pushNotifications: asPushNotifications(), paseoHome: path.join(workdir, "paseo-home"), agentManager, agentStorage, @@ -1503,7 +1503,7 @@ test("archive emits an authoritative agent_update upsert for subscribed clients" onMessage: (message) => emitted.push(message), logger: asSessionLogger(logger), downloadTokenStore: asDownloadTokenStore(), - pushTokenStore: asPushTokenStore(), + pushNotifications: asPushNotifications(), paseoHome: "/tmp/paseo-test", agentManager: asAgentManager({ subscribe: () => () => {}, @@ -1868,7 +1868,7 @@ test("close_items_request archives agents and kills terminals in one batch", asy onMessage: (message) => emitted.push(message), logger: asSessionLogger(sessionLogger), downloadTokenStore: asDownloadTokenStore(), - pushTokenStore: asPushTokenStore(), + pushNotifications: asPushNotifications(), paseoHome: "/tmp/paseo-test", agentManager: asAgentManager({ subscribe: () => () => {}, @@ -2036,7 +2036,7 @@ test("close_items_request archives stored agents that are not currently loaded", onMessage: (message) => emitted.push(message), logger: asSessionLogger(sessionLogger), downloadTokenStore: asDownloadTokenStore(), - pushTokenStore: asPushTokenStore(), + pushNotifications: asPushNotifications(), paseoHome: "/tmp/paseo-test", agentManager: asAgentManager({ subscribe: () => () => {}, @@ -2195,7 +2195,7 @@ test("close_items_request continues after an archive failure", async () => { onMessage: (message) => emitted.push(message), logger: asSessionLogger(sessionLogger), downloadTokenStore: asDownloadTokenStore(), - pushTokenStore: asPushTokenStore(), + pushNotifications: asPushNotifications(), paseoHome: "/tmp/paseo-test", agentManager: asAgentManager({ subscribe: () => () => {}, @@ -3456,7 +3456,7 @@ test("workspace update stream keeps persisted workspace visible after agents sto onMessage: (message) => emitted.push(message), logger: asSessionLogger(logger), downloadTokenStore: asDownloadTokenStore(), - pushTokenStore: asPushTokenStore(), + pushNotifications: asPushNotifications(), paseoHome: "/tmp/paseo-test", agentManager: asAgentManager({ subscribe: () => () => {}, diff --git a/packages/server/src/server/snapshot-mutation-ownership.test.ts b/packages/server/src/server/snapshot-mutation-ownership.test.ts index 373364348..126f06678 100644 --- a/packages/server/src/server/snapshot-mutation-ownership.test.ts +++ b/packages/server/src/server/snapshot-mutation-ownership.test.ts @@ -103,7 +103,7 @@ describe("snapshot mutation ownership boundary", () => { onMessage, logger: createStub(logger), downloadTokenStore: createStub({}), - pushTokenStore: createStub({}), + pushNotifications: createStub({}), paseoHome: "/tmp/paseo-test", agentManager: createStub({ subscribe: () => () => {}, diff --git a/packages/server/src/server/test-utils/paseo-daemon.ts b/packages/server/src/server/test-utils/paseo-daemon.ts index ed4708ec7..8481c4876 100644 --- a/packages/server/src/server/test-utils/paseo-daemon.ts +++ b/packages/server/src/server/test-utils/paseo-daemon.ts @@ -11,7 +11,7 @@ import { } from "../bootstrap.js"; import type { AgentClient, AgentProvider } from "../agent/agent-sdk-types.js"; import { createTestAgentClients } from "./fake-agent-client.js"; -import type { PushNotificationSender } from "../push/notifications.js"; +import type { PushNotificationSender } from "../push/index.js"; interface TestPaseoDaemonOptions { daemonVersion?: string; diff --git a/packages/server/src/server/test-utils/session-stubs.ts b/packages/server/src/server/test-utils/session-stubs.ts index f22f35069..46db4e174 100644 --- a/packages/server/src/server/test-utils/session-stubs.ts +++ b/packages/server/src/server/test-utils/session-stubs.ts @@ -53,8 +53,12 @@ export function asDownloadTokenStore(): SessionOptions["downloadTokenStore"] { return createStub({}); } -export function asPushTokenStore(): SessionOptions["pushTokenStore"] { - return createStub({}); +export function asPushNotifications( + stub: { + [K in keyof SessionOptions["pushNotifications"]]?: unknown; + } = {}, +): SessionOptions["pushNotifications"] { + return createStub(stub); } export function asScheduleService(): SessionOptions["scheduleService"] { diff --git a/packages/server/src/server/websocket-server.notifications.test.ts b/packages/server/src/server/websocket-server.notifications.test.ts index 5ec7a0fdb..e2bec9c71 100644 --- a/packages/server/src/server/websocket-server.notifications.test.ts +++ b/packages/server/src/server/websocket-server.notifications.test.ts @@ -9,7 +9,7 @@ import type { ScheduleService } from "./schedule/service.js"; import type { CheckoutDiffManager } from "./checkout-diff-manager.js"; import { asInternals, createStub } from "./test-utils/class-mocks.js"; import { createProviderSnapshotManagerStub } from "./test-utils/session-stubs.js"; -import type { PushNotificationSender, PushPayload } from "./push/notifications.js"; +import type { PushNotificationSender, PushPayload } from "./push/index.js"; import type { WorkspaceAutoName } from "./workspace-auto-name.js"; const WORKSPACE_ID = "workspace-1"; diff --git a/packages/server/src/server/websocket-server.relay-reconnect.test.ts b/packages/server/src/server/websocket-server.relay-reconnect.test.ts index 5473fa2c4..b14e90c7a 100644 --- a/packages/server/src/server/websocket-server.relay-reconnect.test.ts +++ b/packages/server/src/server/websocket-server.relay-reconnect.test.ts @@ -86,20 +86,12 @@ vi.mock("./session.js", () => ({ Session: sessionMock.MockSession, })); -vi.mock("./push/token-store.js", () => ({ - PushTokenStore: class { - getAllTokens(): string[] { - return []; - } - }, -})); - -vi.mock("./push/push-service.js", () => ({ - PushService: class { - async sendPush(): Promise { - // no-op - } - }, +vi.mock("./push/index.js", () => ({ + createPushNotifications: () => ({ + renew: () => undefined, + revoke: () => undefined, + send: async () => undefined, + }), })); import { z } from "zod"; diff --git a/packages/server/src/server/websocket-server.terminal-notifications.test.ts b/packages/server/src/server/websocket-server.terminal-notifications.test.ts index 231aaa714..1d7160ed3 100644 --- a/packages/server/src/server/websocket-server.terminal-notifications.test.ts +++ b/packages/server/src/server/websocket-server.terminal-notifications.test.ts @@ -15,7 +15,7 @@ import type { import type { PersistedWorkspaceRecord, WorkspaceRegistry } from "./workspace-registry.js"; import { asInternals, createStub } from "./test-utils/class-mocks.js"; import { createProviderSnapshotManagerStub } from "./test-utils/session-stubs.js"; -import type { PushNotificationSender, PushPayload } from "./push/notifications.js"; +import type { PushNotificationSender, PushPayload } from "./push/index.js"; import type { WorkspaceAutoName } from "./workspace-auto-name.js"; const wsModuleMock = vi.hoisted(() => { diff --git a/packages/server/src/server/websocket-server.ts b/packages/server/src/server/websocket-server.ts index 3026b60a3..e3d643d48 100644 --- a/packages/server/src/server/websocket-server.ts +++ b/packages/server/src/server/websocket-server.ts @@ -45,8 +45,11 @@ import type { GitCommandRuntimeMetricsSnapshot } from "../utils/git-command-runt import { snapshotGitCommandRuntimeMetrics } from "../utils/run-git-command.js"; import type { WorkspaceAutoName } from "./workspace-auto-name.js"; import { deriveProjectSlug } from "./workspace-git-metadata.js"; -import { PushTokenStore } from "./push/token-store.js"; -import { createPushNotificationSender, type PushNotificationSender } from "./push/notifications.js"; +import { + createPushNotifications, + type PushNotifications, + type PushNotificationSender, +} from "./push/index.js"; import type { ScriptHealthState } from "./script-health-monitor.js"; import type { ServiceProxySubsystem } from "./service-proxy.js"; import type { WorkspaceScriptRuntimeStore } from "./workspace-script-runtime-store.js"; @@ -534,7 +537,7 @@ export class VoiceAssistantWebSocketServer { private readonly paseoHome: string; private readonly worktreesRoot: string | undefined; private readonly daemonConfigStore: DaemonConfigStore; - private readonly pushTokenStore: PushTokenStore; + private readonly pushNotifications: PushNotifications; private readonly pushNotificationSender: PushNotificationSender; private readonly mcpBaseUrl: string | null; private speech!: SpeechService | null; @@ -683,9 +686,11 @@ export class VoiceAssistantWebSocketServer { }); const pushLogger = this.logger.child({ module: "push" }); - this.pushTokenStore = new PushTokenStore(pushLogger, join(paseoHome, "push-tokens.json")); - this.pushNotificationSender = - pushNotificationSender ?? createPushNotificationSender(pushLogger, this.pushTokenStore); + this.pushNotifications = createPushNotifications({ + logger: pushLogger, + filePath: join(paseoHome, "push-tokens.json"), + }); + this.pushNotificationSender = pushNotificationSender ?? this.pushNotifications; this.agentManager.setAgentAttentionCallback((params) => { void this.broadcastAgentAttention(params).catch((err) => { @@ -1332,7 +1337,7 @@ export class VoiceAssistantWebSocketServer { ); }, downloadTokenStore: this.downloadTokenStore, - pushTokenStore: this.pushTokenStore, + pushNotifications: this.pushNotifications, paseoHome: this.paseoHome, worktreesRoot: this.worktreesRoot, agentManager: this.agentManager, @@ -1542,6 +1547,8 @@ export class VoiceAssistantWebSocketServer { ...(this.advertiseDaemonStatusRpc ? { daemonStatusRpc: true } : {}), // COMPAT(relayConfig): added in v0.2.6, remove gate after 2027-01-31. ...(this.advertiseRelayConfig ? { relayConfig: true } : {}), + // COMPAT(pushTokenRevocation): added in v0.3.2, remove gate after 2027-02-10. + pushTokenRevocation: true, // COMPAT(terminalRestoreModes): added in v0.1.81, remove gate after 2026-11-23. "terminal-restore-modes": true, // COMPAT(terminalInputModeReplay): added in v0.2.6, remove gate after 2027-02-02. diff --git a/packages/server/src/server/wire-compat.test.ts b/packages/server/src/server/wire-compat.test.ts index 503bfc6a1..623e2be2c 100644 --- a/packages/server/src/server/wire-compat.test.ts +++ b/packages/server/src/server/wire-compat.test.ts @@ -209,7 +209,7 @@ function createSessionForWireCompatTest(options?: { onMessage: (message) => messages.push(message), logger: pino({ level: "silent" }), downloadTokenStore: {} as SessionOptions["downloadTokenStore"], - pushTokenStore: {} as SessionOptions["pushTokenStore"], + pushNotifications: {} as SessionOptions["pushNotifications"], paseoHome: "/tmp/paseo-home", agentManager: new InMemoryAgentManager( options?.rows ?? rows,