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
This commit is contained in:
Mohamed Boudra
2026-08-10 23:33:31 +02:00
committed by GitHub
parent 29ae25d7aa
commit 6aa9f92156
33 changed files with 702 additions and 227 deletions
@@ -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: {
@@ -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<boolean> {
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<string | null>(null);
const lastSentTokenRef = useRef<string | null>(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]);
}
@@ -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<void> {
return revokeSubscription(input).catch((error) => {
console.warn("[PushNotifications] Failed to remove local push subscription", error);
});
}
@@ -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<void> {
// Push notifications are native-only.
}
@@ -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<void> {
// Push notifications are native-only.
}
@@ -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<boolean> {
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<string | null> {
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<void> {
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);
}
@@ -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;
}
@@ -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<void>();
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();
+14 -1
View File
@@ -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<void> | 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<void> {
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<void> {
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) => {
+53
View File
@@ -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();
+11
View File
@@ -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<void> {
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;
+19
View File
@@ -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<typeof ClientHeartbeatMessageSchema
export type ListCommandsRequest = z.infer<typeof ListCommandsRequestSchema>;
export type ListCommandsResponse = z.infer<typeof ListCommandsResponseSchema>;
export type RegisterPushTokenMessage = z.infer<typeof RegisterPushTokenMessageSchema>;
export type PushUnregisterRequest = z.infer<typeof PushUnregisterRequestSchema>;
export type PushUnregisterResponse = z.infer<typeof PushUnregisterResponseSchema>;
// Terminal message types
export type ListTerminalsRequest = z.infer<typeof ListTerminalsRequestSchema>;
+1 -1
View File
@@ -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";
@@ -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 {
@@ -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([]);
});
});
+45
View File
@@ -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<void>;
}
export type PushNotificationSender = Pick<PushNotifications, "send">;
export function createPushNotifications(options: {
logger: pino.Logger;
filePath: string;
now?: () => number;
deliver?: (tokens: string[], payload: PushPayload) => Promise<void>;
}): 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);
},
};
}
@@ -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<void>;
}
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);
},
};
}
@@ -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<void> {
@@ -102,7 +101,7 @@ export class PushService {
ticket.details?.error === "DeviceNotRegistered" ||
ticket.details?.error === "InvalidCredentials"
) {
this.tokenStore.removeToken(message.to);
this.revokeToken(message.to);
}
}
}
@@ -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: [] });
});
});
+91 -23
View File
@@ -10,36 +10,69 @@ import { ensurePrivateFile, writePrivateFileAtomicSync } from "../private-files.
*/
export class PushTokenStore {
private readonly logger: pino.Logger;
private tokens: Set<string> = new Set();
private subscriptions = new Map<string, number>();
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");
}
getAllTokens(): string[] {
return Array.from(this.tokens);
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);
}
}
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<string, number>();
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<string, number>): 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;
}
}
}
+85 -2
View File
@@ -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[] = [];
+21 -6
View File
@@ -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<void>;
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<void> {
@@ -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");
}
@@ -195,7 +195,7 @@ function createSessionForWorkspaceGitWatchTests(options?: {
onMessage: (message) => emitted.push(message as { type: string; payload: unknown }),
logger: createStub<pino.Logger>(logger),
downloadTokenStore: createStub<SessionOptions["downloadTokenStore"]>({}),
pushTokenStore: createStub<SessionOptions["pushTokenStore"]>({}),
pushNotifications: createStub<SessionOptions["pushNotifications"]>({}),
paseoHome: "/tmp/paseo-test",
agentManager: createStub<SessionOptions["agentManager"]>({
subscribe: () => () => {},
@@ -93,7 +93,7 @@ function createHarness(input: {
onMessage: (m) => emitted.push(m),
logger: createStub<SessionOptions["logger"]>(logger),
downloadTokenStore: createStub<SessionOptions["downloadTokenStore"]>({}),
pushTokenStore: createStub<SessionOptions["pushTokenStore"]>({}),
pushNotifications: createStub<SessionOptions["pushNotifications"]>({}),
paseoHome: mkdtempSync(path.join(tmpdir(), "paseo-invariant-test-")),
agentManager: createStub<SessionOptions["agentManager"]>({
subscribe: () => () => {},
@@ -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: () => () => {},
@@ -103,7 +103,7 @@ describe("snapshot mutation ownership boundary", () => {
onMessage,
logger: createStub<SessionOptions["logger"]>(logger),
downloadTokenStore: createStub<SessionOptions["downloadTokenStore"]>({}),
pushTokenStore: createStub<SessionOptions["pushTokenStore"]>({}),
pushNotifications: createStub<SessionOptions["pushNotifications"]>({}),
paseoHome: "/tmp/paseo-test",
agentManager: createStub<SessionOptions["agentManager"]>({
subscribe: () => () => {},
@@ -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;
@@ -53,8 +53,12 @@ export function asDownloadTokenStore(): SessionOptions["downloadTokenStore"] {
return createStub<SessionOptions["downloadTokenStore"]>({});
}
export function asPushTokenStore(): SessionOptions["pushTokenStore"] {
return createStub<SessionOptions["pushTokenStore"]>({});
export function asPushNotifications(
stub: {
[K in keyof SessionOptions["pushNotifications"]]?: unknown;
} = {},
): SessionOptions["pushNotifications"] {
return createStub<SessionOptions["pushNotifications"]>(stub);
}
export function asScheduleService(): SessionOptions["scheduleService"] {
@@ -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";
@@ -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<void> {
// no-op
}
},
vi.mock("./push/index.js", () => ({
createPushNotifications: () => ({
renew: () => undefined,
revoke: () => undefined,
send: async () => undefined,
}),
}));
import { z } from "zod";
@@ -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(() => {
+14 -7
View File
@@ -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.
@@ -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,