refactor(server): inject push notification sender (#872)

packages/server/src/server/bootstrap.ts

@@ -119,6 +119,7 @@ import { createConfiguredTerminalManager } from "../terminal/terminal-manager-fa
 import { createConnectionOfferV2, encodeOfferToFragmentUrl } from "./connection-offer.js";
 import { loadOrCreateDaemonKeyPair } from "./daemon-keypair.js";
 import { startRelayTransport, type RelayTransportController } from "./relay-transport.js";
+import type { PushNotificationSender } from "./push/notifications.js";
 import { getOrCreateServerId } from "./server-id.js";
 import { resolveDaemonVersion } from "./daemon-version.js";
 import type { AgentClient, AgentProvider } from "./agent/agent-sdk-types.js";
@@ -205,6 +206,7 @@ export interface PaseoDaemonConfig {
   providerOverrides?: Record<string, ProviderOverride>;
   log?: PersistedConfig["log"];
   onLifecycleIntent?: (intent: DaemonLifecycleIntent) => void;
+  pushNotificationSender?: PushNotificationSender;
 }
 
 export interface PaseoDaemon {
@@ -842,6 +844,7 @@ export async function createPaseoDaemon(
             (hostname) => scriptHealthMonitor.getHealthForHostname(hostname),
             workspaceGitService,
             github,
+            config.pushNotificationSender,
           );
 
           if (relayEnabled) {

packages/server/src/server/client-activity.e2e.test.ts

@@ -1,12 +1,19 @@
-import { afterEach, beforeEach, describe, expect, test, vi } from "vitest";
+import { afterEach, beforeEach, describe, expect, test } from "vitest";
 import { createTestPaseoDaemon, type TestPaseoDaemon } from "./test-utils/paseo-daemon.js";
 import { DaemonClient } from "./test-utils/daemon-client.js";
 import type { AgentStreamEventPayload } from "../shared/messages.js";
 import type { AgentSnapshotPayload } from "./messages.js";
-import { PushService } from "./push/push-service.js";
+import type { PushNotificationSender, PushPayload } from "./push/notifications.js";
-import { PushTokenStore } from "./push/token-store.js";
 import { PRESENCE_THRESHOLD_MS } from "./agent-attention-policy.js";
 
+class RecordingPushNotificationSender implements PushNotificationSender {
+  readonly sent: PushPayload[] = [];
+
+  async send(payload: PushPayload): Promise<void> {
+    this.sent.push(payload);
+  }
+}
+
 /**
  * Tests for client activity tracking and smart notifications.
  *
@@ -31,23 +38,17 @@ describe("client activity tracking", () => {
   let daemon: TestPaseoDaemon;
   let client1: DaemonClient;
   let client2: DaemonClient;
-  let sendPushSpy: ReturnType<typeof vi.spyOn>;
+  let pushNotifications: RecordingPushNotificationSender;
-  let getAllTokensSpy: ReturnType<typeof vi.spyOn>;
 
   beforeEach(async () => {
-    sendPushSpy = vi.spyOn(PushService.prototype, "sendPush").mockResolvedValue(undefined);
+    pushNotifications = new RecordingPushNotificationSender();
-    getAllTokensSpy = vi
+    daemon = await createTestPaseoDaemon({ pushNotificationSender: pushNotifications });
-      .spyOn(PushTokenStore.prototype, "getAllTokens")
-      .mockReturnValue(["ExponentPushToken[activity-test]"]);
-    daemon = await createTestPaseoDaemon();
   });
 
   afterEach(async () => {
     if (client1) await client1.close().catch(() => {});
     if (client2) await client2.close().catch(() => {});
     await daemon.close();
-    sendPushSpy.mockRestore();
-    getAllTokensSpy.mockRestore();
   }, 30000);
 
   async function createClient(): Promise<DaemonClient> {
@@ -205,7 +206,7 @@ describe("client activity tracking", () => {
 
       expect(attention.reason).toBe("finished");
       expect(attention.shouldNotify).toBe(false);
-      expect(sendPushSpy).toHaveBeenCalledTimes(1);
+      expect(pushNotifications.sent).toHaveLength(1);
     }, 120000);
 
     test("notification when no heartbeat received (legacy/new client)", async () => {
@@ -225,7 +226,7 @@ describe("client activity tracking", () => {
 
       expect(attention.reason).toBe("finished");
       expect(attention.shouldNotify).toBe(false);
-      expect(sendPushSpy).toHaveBeenCalledTimes(1);
+      expect(pushNotifications.sent).toHaveLength(1);
     }, 120000);
   });
 
@@ -309,7 +310,7 @@ describe("client activity tracking", () => {
       // No stale client is selected for in-app; push handles the fallback.
       expect(attention1.shouldNotify).toBe(false);
       expect(attention2.shouldNotify).toBe(false);
-      expect(sendPushSpy).toHaveBeenCalledTimes(1);
+      expect(pushNotifications.sent).toHaveLength(1);
     }, 120000);
 
     test("notifies only the present Electron-style web client when Firefox is stale", async () => {
@@ -345,7 +346,7 @@ describe("client activity tracking", () => {
 
       expect(attention1.shouldNotify).toBe(false);
       expect(attention2.shouldNotify).toBe(true);
-      expect(sendPushSpy).not.toHaveBeenCalled();
+      expect(pushNotifications.sent).toEqual([]);
     }, 120000);
   });
 
@@ -465,7 +466,7 @@ describe("client activity tracking", () => {
 
       expect(attention1.shouldNotify).toBe(false);
       expect(attention2.shouldNotify).toBe(true);
-      expect(sendPushSpy).not.toHaveBeenCalled();
+      expect(pushNotifications.sent).toEqual([]);
     }, 120000);
 
     test("notify web when user active on web but looking at different agent", async () => {
@@ -543,7 +544,7 @@ describe("client activity tracking", () => {
 
       expect(attention1.shouldNotify).toBe(false);
       expect(attention2.shouldNotify).toBe(false);
-      expect(sendPushSpy).toHaveBeenCalledTimes(1);
+      expect(pushNotifications.sent).toHaveLength(1);
     }, 120000);
   });
 
@@ -581,7 +582,7 @@ describe("client activity tracking", () => {
 
       expect(attention1.shouldNotify).toBe(false);
       expect(attention2.shouldNotify).toBe(false);
-      expect(sendPushSpy).toHaveBeenCalledTimes(1);
+      expect(pushNotifications.sent).toHaveLength(1);
     }, 120000);
 
     test("notification when app not visible but activity is recent", async () => {
@@ -647,7 +648,7 @@ describe("client activity tracking", () => {
 
       expect(attention1.shouldNotify).toBe(true);
       expect(attention2.shouldNotify).toBe(false);
-      expect(sendPushSpy).not.toHaveBeenCalled();
+      expect(pushNotifications.sent).toEqual([]);
     }, 120000);
   });
 });

packages/server/src/server/push/notifications.ts

@@ -0,0 +1,29 @@
+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);
+    },
+  };
+}

packages/server/src/server/push/push-service.ts

@@ -1,7 +1,7 @@
 import type { PushTokenStore } from "./token-store.js";
 import type pino from "pino";
 
-interface PushPayload {
+export interface PushPayload {
   title: string;
   body: string;
   data?: Record<string, unknown>;

packages/server/src/server/test-utils/paseo-daemon.ts

@@ -11,6 +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";
 
 interface TestPaseoDaemonOptions {
   downloadTokenTtlMs?: number;
@@ -30,6 +31,7 @@ interface TestPaseoDaemonOptions {
   voiceLlmModel?: string | null;
   dictationFinalTimeoutMs?: number;
   auth?: PaseoDaemonConfig["auth"];
+  pushNotificationSender?: PushNotificationSender;
 }
 
 export interface TestPaseoDaemon {
@@ -157,6 +159,7 @@ async function prepareTestDaemonConfig(
     relayEndpoint: options.relayEndpoint ?? "relay.paseo.sh:443",
     appBaseUrl: "https://app.paseo.sh",
     auth: options.auth,
+    pushNotificationSender: options.pushNotificationSender,
     openai: options.openai,
     speech: options.speech,
     voiceLlmProvider: options.voiceLlmProvider ?? null,

packages/server/src/server/websocket-server.notifications.test.ts

@@ -10,6 +10,7 @@ import type { LoopService } from "./loop-service.js";
 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 type { PushNotificationSender, PushPayload } from "./push/notifications.js";
 
 const wsModuleMock = vi.hoisted(() => {
   class MockWebSocketServer {
@@ -28,11 +29,6 @@ const wsModuleMock = vi.hoisted(() => {
   return { MockWebSocketServer };
 });
 
-const pushMocks = vi.hoisted(() => ({
-  getAllTokens: vi.fn(() => ["ExponentPushToken[token-1]"]),
-  sendPush: vi.fn(async () => {}),
-}));
-
 vi.mock("ws", () => ({
   WebSocketServer: wsModuleMock.MockWebSocketServer,
 }));
@@ -43,19 +39,6 @@ vi.mock("./session.js", () => ({
   },
 }));
 
-vi.mock("./push/token-store.js", () => ({
-  PushTokenStore: class {
-    getAllTokens = pushMocks.getAllTokens;
-    removeToken = vi.fn();
-  },
-}));
-
-vi.mock("./push/push-service.js", () => ({
-  PushService: class {
-    sendPush = pushMocks.sendPush;
-  },
-}));
-
 import { VoiceAssistantWebSocketServer } from "./websocket-server.js";
 
 interface WebSocketServerInternals {
@@ -81,7 +64,16 @@ function createLogger() {
   return logger;
 }
 
+class RecordingPushNotificationSender implements PushNotificationSender {
+  readonly sent: PushPayload[] = [];
+
+  async send(payload: PushPayload): Promise<void> {
+    this.sent.push(payload);
+  }
+}
+
 function createServer(agentManagerOverrides?: Record<string, unknown>) {
+  const pushNotifications = new RecordingPushNotificationSender();
   const agentManager = {
     setAgentAttentionCallback: vi.fn(),
     getAgent: vi.fn(() => null),
@@ -137,9 +129,18 @@ function createServer(agentManagerOverrides?: Record<string, unknown>) {
       })),
       dispose: vi.fn(),
     }),
+    undefined,
+    undefined,
+    undefined,
+    undefined,
+    undefined,
+    undefined,
+    undefined,
+    undefined,
+    pushNotifications,
   );
 
-  return { server, agentManager };
+  return { server, agentManager, pushNotifications };
 }
 
 function createOpenSocket() {
@@ -208,7 +209,7 @@ describe("VoiceAssistantWebSocketServer notification payloads", () => {
     const getLastAssistantMessage = vi.fn(
       async () => "**Done**. Updated `README.md` and [link](https://example.com).",
     );
-    const { server } = createServer({
+    const { server, pushNotifications } = createServer({
       getAgent: vi.fn(() => ({
         config: { title: null },
         cwd: "/tmp/worktree",
@@ -223,21 +224,23 @@ describe("VoiceAssistantWebSocketServer notification payloads", () => {
       reason: "finished",
     });
 
-    expect(pushMocks.sendPush).toHaveBeenCalledWith(["ExponentPushToken[token-1]"], {
+    expect(pushNotifications.sent).toEqual([
-      title: "Agent finished",
+      {
-      body: "Done. Updated README.md and link.",
+        title: "Agent finished",
-      data: {
+        body: "Done. Updated README.md and link.",
-        serverId: "srv-test",
+        data: {
-        agentId: "agent-1",
+          serverId: "srv-test",
-        reason: "finished",
+          agentId: "agent-1",
+          reason: "finished",
+        },
       },
-    });
+    ]);
     expect(getLastAssistantMessage).toHaveBeenCalledWith("agent-1");
   });
 
   it("sends push notifications regardless of UI label presence", async () => {
     const getLastAssistantMessage = vi.fn(async () => "Done.");
-    const { server } = createServer({
+    const { server, pushNotifications } = createServer({
       getAgent: vi.fn(() => ({
         config: { title: null },
         cwd: "/tmp/worktree",
@@ -253,12 +256,12 @@ describe("VoiceAssistantWebSocketServer notification payloads", () => {
       reason: "finished",
     });
 
-    expect(pushMocks.sendPush).toHaveBeenCalledTimes(1);
+    expect(pushNotifications.sent).toHaveLength(1);
     expect(getLastAssistantMessage).toHaveBeenCalledWith("agent-2");
   });
 
   it("routes a hidden stale focused browser tab's notification to the present Electron web client", async () => {
-    const { server } = createServer();
+    const { server, pushNotifications } = createServer();
     const nowMs = Date.now();
     const electronWs = connectClient(server, {
       deviceType: "web",
@@ -281,11 +284,11 @@ describe("VoiceAssistantWebSocketServer notification payloads", () => {
 
     expect(readAttentionRequiredMessage(electronWs).shouldNotify).toBe(true);
     expect(readAttentionRequiredMessage(firefoxWs).shouldNotify).toBe(false);
-    expect(pushMocks.sendPush).not.toHaveBeenCalled();
+    expect(pushNotifications.sent).toEqual([]);
   });
 
   it("pushes non-error attention when the only connected client has never sent a heartbeat", async () => {
-    const { server } = createServer();
+    const { server, pushNotifications } = createServer();
     const ws = connectClient(server, null);
 
     await asInternals<WebSocketServerInternals>(server).broadcastAgentAttention({
@@ -295,11 +298,11 @@ describe("VoiceAssistantWebSocketServer notification payloads", () => {
     });
 
     expect(readAttentionRequiredMessage(ws).shouldNotify).toBe(false);
-    expect(pushMocks.sendPush).toHaveBeenCalledTimes(1);
+    expect(pushNotifications.sent).toHaveLength(1);
   });
 
   it("does not push error attention when the only connected client has never sent a heartbeat", async () => {
-    const { server } = createServer();
+    const { server, pushNotifications } = createServer();
     const ws = connectClient(server, null);
 
     await asInternals<WebSocketServerInternals>(server).broadcastAgentAttention({
@@ -309,6 +312,6 @@ describe("VoiceAssistantWebSocketServer notification payloads", () => {
     });
 
     expect(readAttentionRequiredMessage(ws).shouldNotify).toBe(false);
-    expect(pushMocks.sendPush).not.toHaveBeenCalled();
+    expect(pushNotifications.sent).toEqual([]);
   });
 });

packages/server/src/server/websocket-server.ts

@@ -40,7 +40,7 @@ import { buildProviderRegistry, createClientsFromRegistry } from "./agent/provid
 import type { WorkspaceGitRuntimeSnapshot, WorkspaceGitService } from "./workspace-git-service.js";
 import { buildWorkspaceGitMetadataFromSnapshot } from "./workspace-git-metadata.js";
 import { PushTokenStore } from "./push/token-store.js";
-import { PushService } from "./push/push-service.js";
+import { createPushNotificationSender, type PushNotificationSender } from "./push/notifications.js";
 import type { ScriptHealthState } from "./script-health-monitor.js";
 import type { ScriptRouteStore } from "./script-proxy.js";
 import type { WorkspaceScriptRuntimeStore } from "./workspace-script-runtime-store.js";
@@ -358,7 +358,7 @@ export class VoiceAssistantWebSocketServer {
   private readonly paseoHome: string;
   private readonly daemonConfigStore: DaemonConfigStore;
   private readonly pushTokenStore: PushTokenStore;
-  private readonly pushService: PushService;
+  private readonly pushNotificationSender: PushNotificationSender;
   private readonly mcpBaseUrl: string | null;
   private speech!: SpeechService | null;
   private terminalManager!: TerminalManager | null;
@@ -453,6 +453,7 @@ export class VoiceAssistantWebSocketServer {
     resolveScriptHealth?: (hostname: string) => ScriptHealthState | null,
     workspaceGitService?: WorkspaceGitService,
     github?: GitHubService,
+    pushNotificationSender?: PushNotificationSender,
   ) {
     this.logger = logger.child({ module: "websocket-server" });
     this.serverId = serverId;
@@ -529,7 +530,8 @@ export class VoiceAssistantWebSocketServer {
 
     const pushLogger = this.logger.child({ module: "push" });
     this.pushTokenStore = new PushTokenStore(pushLogger, join(paseoHome, "push-tokens.json"));
-    this.pushService = new PushService(pushLogger, this.pushTokenStore);
+    this.pushNotificationSender =
+      pushNotificationSender ?? createPushNotificationSender(pushLogger, this.pushTokenStore);
 
     this.agentManager.setAgentAttentionCallback((params) => {
       void this.broadcastAgentAttention(params).catch((err) => {
@@ -1819,11 +1821,9 @@ export class VoiceAssistantWebSocketServer {
     });
 
     if (plan.shouldPush) {
-      const tokens = this.pushTokenStore.getAllTokens();
+      void this.pushNotificationSender.send(notification).catch((err) => {
-      this.logger.info({ tokenCount: tokens.length }, "Sending push notification");
+        this.logger.warn({ err, agentId: params.agentId }, "Failed to send push notification");
-      if (tokens.length > 0) {
+      });
-        void this.pushService.sendPush(tokens, notification);
-      }
     }
 
     for (const [clientIndex, { ws }] of clientEntries.entries()) {