feat: add smart notification routing with device-aware heartbeats and push support
This commit is contained in:
@@ -353,7 +353,6 @@ function AgentScreenContent({
|
||||
|
||||
// Get methods
|
||||
const methods = useSessionStore((state) => state.sessions[serverId]?.methods);
|
||||
const initializeAgent = methods?.initializeAgent;
|
||||
const refreshAgent = methods?.refreshAgent;
|
||||
const setFocusedAgentId = useCallback(
|
||||
(agentId: string | null) => {
|
||||
@@ -436,11 +435,11 @@ function AgentScreenContent({
|
||||
? isInitializingFromMap !== false
|
||||
: false;
|
||||
|
||||
// Track which agent we've already triggered initialization for
|
||||
const initializedAgentRef = useRef<string | null>(null);
|
||||
// Get ensureAgentIsInitialized from methods
|
||||
const ensureAgentIsInitialized = methods?.ensureAgentIsInitialized;
|
||||
|
||||
useEffect(() => {
|
||||
if (!resolvedAgentId || !initializeAgent) {
|
||||
if (!resolvedAgentId || !ensureAgentIsInitialized) {
|
||||
return;
|
||||
}
|
||||
|
||||
@@ -449,28 +448,15 @@ function AgentScreenContent({
|
||||
return;
|
||||
}
|
||||
|
||||
// Skip if already initializing
|
||||
if (isInitializingFromMap === true) {
|
||||
return;
|
||||
}
|
||||
|
||||
// Skip if we already triggered initialization for this agent
|
||||
if (initializedAgentRef.current === resolvedAgentId) {
|
||||
return;
|
||||
}
|
||||
|
||||
// Mark this agent as initialized
|
||||
initializedAgentRef.current = resolvedAgentId;
|
||||
|
||||
// Always fetch timeline when opening agent screen
|
||||
// Server returns full snapshot, client replaces cache
|
||||
// This ensures we always have complete history, even after reconnection
|
||||
console.log("[AgentScreen] Fetching agent timeline", {
|
||||
agentId: resolvedAgentId,
|
||||
// ensureAgentIsInitialized handles deduplication via module-level promises map
|
||||
// If already initialized or in-flight, returns resolved/pending promise immediately
|
||||
ensureAgentIsInitialized(resolvedAgentId).catch((error) => {
|
||||
console.warn("[AgentScreen] Agent initialization failed", {
|
||||
agentId: resolvedAgentId,
|
||||
error,
|
||||
});
|
||||
});
|
||||
|
||||
initializeAgent({ agentId: resolvedAgentId });
|
||||
}, [resolvedAgentId, initializeAgent, isInitializingFromMap, isConnected]);
|
||||
}, [resolvedAgentId, ensureAgentIsInitialized, isConnected]);
|
||||
|
||||
useEffect(() => {
|
||||
if (Platform.OS !== "web") {
|
||||
|
||||
@@ -12,6 +12,7 @@ import AsyncStorage from "@react-native-async-storage/async-storage";
|
||||
import { useMutation } from "@tanstack/react-query";
|
||||
import { useDaemonClient } from "@/hooks/use-daemon-client";
|
||||
import { useAudioPlayer } from "@/hooks/use-audio-player";
|
||||
import { useClientActivity } from "@/hooks/use-client-activity";
|
||||
import {
|
||||
applyStreamEvent,
|
||||
generateMessageId,
|
||||
@@ -194,6 +195,20 @@ type FileDownloadTokenPayload = Extract<
|
||||
|
||||
const SESSION_SNAPSHOT_STORAGE_PREFIX = "@paseo:session-snapshot:";
|
||||
|
||||
// Module-level map for agent initialization promises
|
||||
// Key: `${serverId}:${agentId}`, Value: { promise, resolve, reject }
|
||||
// This survives Fast Refresh because it's outside React component tree
|
||||
interface DeferredInit {
|
||||
promise: Promise<void>;
|
||||
resolve: () => void;
|
||||
reject: (error: Error) => void;
|
||||
}
|
||||
const agentInitializationPromises = new Map<string, DeferredInit>();
|
||||
|
||||
function getInitKey(serverId: string, agentId: string): string {
|
||||
return `${serverId}:${agentId}`;
|
||||
}
|
||||
|
||||
type PersistedSessionSnapshot = {
|
||||
agents: AgentSnapshotPayload[];
|
||||
savedAt: string;
|
||||
@@ -356,6 +371,7 @@ export interface SessionContextValue {
|
||||
requestId: string,
|
||||
response: any
|
||||
) => void;
|
||||
ensureAgentIsInitialized: (agentId: string) => Promise<void>;
|
||||
}
|
||||
|
||||
const SessionContext = createContext<SessionContextValue | null>(null);
|
||||
@@ -421,6 +437,14 @@ export function SessionProvider({
|
||||
(state) => state.updateSessionConnection
|
||||
);
|
||||
|
||||
// Track focused agent for heartbeat
|
||||
const focusedAgentId = useSessionStore(
|
||||
(state) => state.sessions[serverId]?.focusedAgentId ?? null
|
||||
);
|
||||
|
||||
// Client activity tracking (heartbeat, push token registration)
|
||||
useClientActivity({ client, focusedAgentId });
|
||||
|
||||
// State for voice detection flags (will be set by RealtimeContext)
|
||||
const isDetectingRef = useRef(false);
|
||||
const isSpeakingRef = useRef(false);
|
||||
@@ -1009,11 +1033,14 @@ export function SessionProvider({
|
||||
console.log("[Session] agent_stream", { agentId, event, timestamp });
|
||||
|
||||
if (event.type === "attention_required") {
|
||||
notifyAgentAttention({
|
||||
agentId,
|
||||
reason: event.reason,
|
||||
timestamp: event.timestamp,
|
||||
});
|
||||
console.log("[Session] attention_required", { agentId, shouldNotify: event.shouldNotify, reason: event.reason });
|
||||
if (event.shouldNotify) {
|
||||
notifyAgentAttention({
|
||||
agentId,
|
||||
reason: event.reason,
|
||||
timestamp: event.timestamp,
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
const session = useSessionStore.getState().sessions[serverId];
|
||||
@@ -1094,6 +1121,14 @@ export function SessionProvider({
|
||||
next.set(agentId, false);
|
||||
return next;
|
||||
});
|
||||
|
||||
// Resolve the initialization promise (even for empty history)
|
||||
const initKey = getInitKey(serverId, agentId);
|
||||
const deferred = agentInitializationPromises.get(initKey);
|
||||
if (deferred) {
|
||||
deferred.resolve();
|
||||
// Keep the promise in the map so subsequent calls return immediately
|
||||
}
|
||||
}
|
||||
);
|
||||
|
||||
@@ -1563,7 +1598,6 @@ export function SessionProvider({
|
||||
|
||||
const initializeAgent = useCallback(
|
||||
({ agentId, requestId }: { agentId: string; requestId?: string }) => {
|
||||
console.log("[Session] initializeAgent called", { agentId, requestId });
|
||||
setInitializingAgents(serverId, (prev) => {
|
||||
const next = new Map(prev);
|
||||
next.set(agentId, true);
|
||||
@@ -1635,6 +1669,94 @@ export function SessionProvider({
|
||||
]
|
||||
);
|
||||
|
||||
const INIT_TIMEOUT_MS = 10000;
|
||||
|
||||
const ensureAgentIsInitialized = useCallback(
|
||||
(agentId: string): Promise<void> => {
|
||||
const key = getInitKey(serverId, agentId);
|
||||
|
||||
// If we already have a promise (resolved or in-flight), return it
|
||||
const existing = agentInitializationPromises.get(key);
|
||||
if (existing) {
|
||||
return existing.promise;
|
||||
}
|
||||
|
||||
// Create a deferred promise
|
||||
let resolve: () => void;
|
||||
let reject: (error: Error) => void;
|
||||
const promise = new Promise<void>((res, rej) => {
|
||||
resolve = res;
|
||||
reject = rej;
|
||||
});
|
||||
|
||||
const deferred: DeferredInit = {
|
||||
promise,
|
||||
resolve: resolve!,
|
||||
reject: reject!,
|
||||
};
|
||||
agentInitializationPromises.set(key, deferred);
|
||||
|
||||
// Set up timeout
|
||||
const timeoutId = setTimeout(() => {
|
||||
const entry = agentInitializationPromises.get(key);
|
||||
if (entry === deferred) {
|
||||
agentInitializationPromises.delete(key);
|
||||
deferred.reject(new Error(`Agent initialization timed out after ${INIT_TIMEOUT_MS}ms`));
|
||||
}
|
||||
}, INIT_TIMEOUT_MS);
|
||||
|
||||
// Set UI loading state
|
||||
setInitializingAgents(serverId, (prev) => {
|
||||
const next = new Map(prev);
|
||||
next.set(agentId, true);
|
||||
return next;
|
||||
});
|
||||
|
||||
// Clear existing stream state
|
||||
setAgentStreamTail(serverId, (prev) => {
|
||||
const next = new Map(prev);
|
||||
next.set(agentId, []);
|
||||
return next;
|
||||
});
|
||||
clearAgentStreamHead(serverId, agentId);
|
||||
|
||||
if (!client) {
|
||||
console.warn("[Session] ensureAgentIsInitialized skipped: daemon unavailable");
|
||||
clearTimeout(timeoutId);
|
||||
agentInitializationPromises.delete(key);
|
||||
setInitializingAgents(serverId, (prev) => {
|
||||
const next = new Map(prev);
|
||||
next.set(agentId, false);
|
||||
return next;
|
||||
});
|
||||
deferred.reject(new Error("Daemon unavailable"));
|
||||
return promise;
|
||||
}
|
||||
|
||||
client
|
||||
.initializeAgent(agentId)
|
||||
.then(() => {
|
||||
// Note: We don't resolve here - we wait for agent_stream_snapshot
|
||||
// The snapshot handler will call deferred.resolve()
|
||||
clearTimeout(timeoutId);
|
||||
})
|
||||
.catch((error) => {
|
||||
console.warn("[Session] ensureAgentIsInitialized failed", { agentId, error });
|
||||
clearTimeout(timeoutId);
|
||||
agentInitializationPromises.delete(key);
|
||||
setInitializingAgents(serverId, (prev) => {
|
||||
const next = new Map(prev);
|
||||
next.set(agentId, false);
|
||||
return next;
|
||||
});
|
||||
deferred.reject(error instanceof Error ? error : new Error(String(error)));
|
||||
});
|
||||
|
||||
return promise;
|
||||
},
|
||||
[serverId, client, setAgentStreamTail, setInitializingAgents, clearAgentStreamHead]
|
||||
);
|
||||
|
||||
const requestProviderModels = useCallback(
|
||||
(provider: any, options?: { cwd?: string }) => {
|
||||
const requestId = generateMessageId();
|
||||
@@ -2231,6 +2353,7 @@ export function SessionProvider({
|
||||
createAgent,
|
||||
setAgentMode,
|
||||
respondToPermission,
|
||||
ensureAgentIsInitialized,
|
||||
}),
|
||||
[
|
||||
serverId,
|
||||
@@ -2254,6 +2377,7 @@ export function SessionProvider({
|
||||
createAgent,
|
||||
setAgentMode,
|
||||
respondToPermission,
|
||||
ensureAgentIsInitialized,
|
||||
]
|
||||
);
|
||||
|
||||
@@ -2280,6 +2404,7 @@ export function SessionProvider({
|
||||
createAgent,
|
||||
setAgentMode,
|
||||
respondToPermission,
|
||||
ensureAgentIsInitialized,
|
||||
}),
|
||||
[
|
||||
setVoiceDetectionFlags,
|
||||
@@ -2300,6 +2425,7 @@ export function SessionProvider({
|
||||
createAgent,
|
||||
setAgentMode,
|
||||
respondToPermission,
|
||||
ensureAgentIsInitialized,
|
||||
]
|
||||
);
|
||||
|
||||
|
||||
@@ -0,0 +1,112 @@
|
||||
import { useEffect, useRef, useCallback } from "react";
|
||||
import { AppState, Platform } from "react-native";
|
||||
import type { DaemonClientV2 } from "@server/client/daemon-client-v2";
|
||||
|
||||
const HEARTBEAT_INTERVAL_MS = 15_000;
|
||||
|
||||
interface ClientActivityOptions {
|
||||
client: DaemonClientV2;
|
||||
focusedAgentId: string | null;
|
||||
}
|
||||
|
||||
/**
|
||||
* Handles client activity reporting:
|
||||
* - Heartbeat sending every 15 seconds
|
||||
* - App visibility tracking (updates lastActivityAt on foreground)
|
||||
* - Sends heartbeat immediately when focused agent changes
|
||||
*/
|
||||
export function useClientActivity({ client, focusedAgentId }: ClientActivityOptions): void {
|
||||
const lastActivityAtRef = useRef<Date>(new Date());
|
||||
const appVisibleRef = useRef(AppState.currentState === "active");
|
||||
const heartbeatIntervalRef = useRef<ReturnType<typeof setInterval> | null>(null);
|
||||
const prevFocusedAgentIdRef = useRef<string | null>(focusedAgentId);
|
||||
|
||||
const deviceType = Platform.OS === "web" ? "web" : "mobile";
|
||||
|
||||
const recordUserActivity = useCallback(() => {
|
||||
lastActivityAtRef.current = new Date();
|
||||
}, []);
|
||||
|
||||
const sendHeartbeat = useCallback(() => {
|
||||
if (!client.isConnected) {
|
||||
console.log("[ClientActivity] sendHeartbeat skipped - not connected");
|
||||
return;
|
||||
}
|
||||
lastActivityAtRef.current = new Date();
|
||||
console.log("[ClientActivity] sendHeartbeat", {
|
||||
deviceType,
|
||||
focusedAgentId,
|
||||
lastActivityAt: lastActivityAtRef.current.toISOString(),
|
||||
appVisible: appVisibleRef.current,
|
||||
});
|
||||
client.sendHeartbeat({
|
||||
deviceType,
|
||||
focusedAgentId,
|
||||
lastActivityAt: lastActivityAtRef.current.toISOString(),
|
||||
appVisible: appVisibleRef.current,
|
||||
});
|
||||
}, [client, deviceType, focusedAgentId]);
|
||||
|
||||
// Track app visibility
|
||||
useEffect(() => {
|
||||
console.log("[ClientActivity] AppState effect mounted, current:", AppState.currentState);
|
||||
const subscription = AppState.addEventListener("change", (nextState) => {
|
||||
console.log("[ClientActivity] AppState changed:", nextState);
|
||||
appVisibleRef.current = nextState === "active";
|
||||
if (nextState === "active") {
|
||||
recordUserActivity();
|
||||
}
|
||||
});
|
||||
|
||||
return () => subscription.remove();
|
||||
}, [recordUserActivity]);
|
||||
|
||||
// Send heartbeat on focused agent change
|
||||
useEffect(() => {
|
||||
if (prevFocusedAgentIdRef.current !== focusedAgentId) {
|
||||
console.log("[ClientActivity] focusedAgentId changed:", prevFocusedAgentIdRef.current, "->", focusedAgentId);
|
||||
prevFocusedAgentIdRef.current = focusedAgentId;
|
||||
sendHeartbeat();
|
||||
}
|
||||
}, [focusedAgentId, sendHeartbeat]);
|
||||
|
||||
// Periodic heartbeat
|
||||
useEffect(() => {
|
||||
console.log("[ClientActivity] Heartbeat effect mounted, isConnected:", client.isConnected);
|
||||
|
||||
const startHeartbeat = () => {
|
||||
console.log("[ClientActivity] startHeartbeat called");
|
||||
if (heartbeatIntervalRef.current) {
|
||||
clearInterval(heartbeatIntervalRef.current);
|
||||
}
|
||||
sendHeartbeat();
|
||||
heartbeatIntervalRef.current = setInterval(sendHeartbeat, HEARTBEAT_INTERVAL_MS);
|
||||
};
|
||||
|
||||
const stopHeartbeat = () => {
|
||||
console.log("[ClientActivity] stopHeartbeat called");
|
||||
if (heartbeatIntervalRef.current) {
|
||||
clearInterval(heartbeatIntervalRef.current);
|
||||
heartbeatIntervalRef.current = null;
|
||||
}
|
||||
};
|
||||
|
||||
const unsubscribe = client.subscribeConnectionStatus((state) => {
|
||||
console.log("[ClientActivity] Connection status changed:", state.status);
|
||||
if (state.status === "connected") {
|
||||
startHeartbeat();
|
||||
} else {
|
||||
stopHeartbeat();
|
||||
}
|
||||
});
|
||||
|
||||
if (client.isConnected) {
|
||||
startHeartbeat();
|
||||
}
|
||||
|
||||
return () => {
|
||||
unsubscribe();
|
||||
stopHeartbeat();
|
||||
};
|
||||
}, [client, sendHeartbeat]);
|
||||
}
|
||||
@@ -204,6 +204,7 @@ export interface SessionState {
|
||||
}) => Promise<void>;
|
||||
setAgentMode: (agentId: string, modeId: string) => void;
|
||||
respondToPermission: (agentId: string, requestId: string, response: any) => void;
|
||||
ensureAgentIsInitialized: (agentId: string) => Promise<void>;
|
||||
} | null;
|
||||
|
||||
// Hydration status
|
||||
@@ -223,7 +224,7 @@ export interface SessionState {
|
||||
agentStreamTail: Map<string, StreamItem[]>;
|
||||
agentStreamHead: Map<string, StreamItem[]>;
|
||||
|
||||
// Initializing agents
|
||||
// Initializing agents (used for UI loading state)
|
||||
initializingAgents: Map<string, boolean>;
|
||||
|
||||
// Agents
|
||||
|
||||
@@ -12,6 +12,7 @@ import {
|
||||
/**
|
||||
* Simple hash function for deterministic ID generation
|
||||
*/
|
||||
|
||||
function simpleHash(str: string): string {
|
||||
let hash = 0;
|
||||
for (let i = 0; i < str.length; i += 1) {
|
||||
|
||||
@@ -448,13 +448,24 @@ export class DaemonClientV2 {
|
||||
}
|
||||
|
||||
sendHeartbeat(params: {
|
||||
deviceType: "web" | "mobile";
|
||||
focusedAgentId: string | null;
|
||||
lastActivityAt: string;
|
||||
appVisible: boolean;
|
||||
}): void {
|
||||
this.sendSessionMessage({
|
||||
type: "client_heartbeat",
|
||||
deviceType: params.deviceType,
|
||||
focusedAgentId: params.focusedAgentId,
|
||||
lastActivityAt: params.lastActivityAt,
|
||||
appVisible: params.appVisible,
|
||||
});
|
||||
}
|
||||
|
||||
registerPushToken(token: string): void {
|
||||
this.sendSessionMessage({
|
||||
type: "register_push_token",
|
||||
token,
|
||||
});
|
||||
}
|
||||
|
||||
|
||||
@@ -15,13 +15,21 @@ import type { AgentStreamEventPayload } from "../shared/messages.js";
|
||||
/**
|
||||
* Tests for client activity tracking and smart notifications.
|
||||
*
|
||||
* The server tracks client activity via heartbeats to determine whether
|
||||
* to notify users when agents need attention. If any client is actively
|
||||
* viewing an agent, shouldNotify is false. Otherwise, shouldNotify is true.
|
||||
* Core UX principle: The user is ONE person across all devices.
|
||||
* We want to notify them where they'll see it.
|
||||
*
|
||||
* Activity is determined by:
|
||||
* - focusedAgentId: which agent the client is viewing
|
||||
* - lastActivityAt: timestamp of last user interaction (must be within 60s)
|
||||
* Rules:
|
||||
* 1. If user is actively looking at the agent (focused + visible + recent activity) → no notification
|
||||
* 2. If user is on a device but looking elsewhere → notify on that device
|
||||
* 3. If web is stale (>2min no activity) but mobile is connected → notify mobile, not web
|
||||
* 4. Mobile is the fallback - always notify if connected and web is stale
|
||||
* 5. Switching tabs (appVisible=false) with recent activity → NO notification (user is still at computer)
|
||||
*
|
||||
* Heartbeat contains:
|
||||
* - deviceType: "web" | "mobile"
|
||||
* - focusedAgentId: which agent they're looking at (null if app not visible)
|
||||
* - lastActivityAt: timestamp of last user interaction
|
||||
* - appVisible: whether the app/tab is in foreground
|
||||
*/
|
||||
describe("client activity tracking", () => {
|
||||
let daemon: TestPaseoDaemon;
|
||||
@@ -71,134 +79,595 @@ describe("client activity tracking", () => {
|
||||
});
|
||||
}
|
||||
|
||||
test("shouldNotify is false when client is active and focused on agent", async () => {
|
||||
client1 = await createClient();
|
||||
// ===========================================================================
|
||||
// SINGLE CLIENT SCENARIOS
|
||||
// ===========================================================================
|
||||
|
||||
const agent = await client1.createAgent({
|
||||
provider: "codex",
|
||||
cwd: "/tmp",
|
||||
title: "Activity Test",
|
||||
});
|
||||
describe("single client - basic cases", () => {
|
||||
test("no notification when actively focused on agent", async () => {
|
||||
client1 = await createClient();
|
||||
|
||||
client1.sendHeartbeat({
|
||||
focusedAgentId: agent.id,
|
||||
lastActivityAt: new Date().toISOString(),
|
||||
});
|
||||
const agent = await client1.createAgent({
|
||||
provider: "claude",
|
||||
model: "claude-haiku-4-5",
|
||||
cwd: "/tmp",
|
||||
title: "Active Focus Test",
|
||||
});
|
||||
|
||||
await new Promise((r) => setTimeout(r, 100));
|
||||
client1.sendHeartbeat({
|
||||
deviceType: "web",
|
||||
focusedAgentId: agent.id,
|
||||
lastActivityAt: new Date().toISOString(),
|
||||
appVisible: true,
|
||||
});
|
||||
|
||||
const attentionPromise = waitForAttentionRequired(client1, agent.id);
|
||||
await client1.sendMessage(agent.id, "Say 'hello' and nothing else");
|
||||
await new Promise((r) => setTimeout(r, 100));
|
||||
|
||||
const attention = await attentionPromise;
|
||||
const attentionPromise = waitForAttentionRequired(client1, agent.id);
|
||||
await client1.sendMessage(agent.id, "Say 'hello' and nothing else");
|
||||
|
||||
expect(attention.reason).toBe("finished");
|
||||
expect(attention.shouldNotify).toBe(false);
|
||||
}, 120000);
|
||||
const attention = await attentionPromise;
|
||||
|
||||
test("shouldNotify is true when client has stale activity", async () => {
|
||||
client1 = await createClient();
|
||||
expect(attention.reason).toBe("finished");
|
||||
expect(attention.shouldNotify).toBe(false);
|
||||
}, 120000);
|
||||
|
||||
const agent = await client1.createAgent({
|
||||
provider: "codex",
|
||||
cwd: "/tmp",
|
||||
title: "Stale Activity Test",
|
||||
});
|
||||
test("notification when focused on different agent", async () => {
|
||||
client1 = await createClient();
|
||||
|
||||
const staleTime = new Date(Date.now() - 120_000).toISOString();
|
||||
client1.sendHeartbeat({
|
||||
focusedAgentId: agent.id,
|
||||
lastActivityAt: staleTime,
|
||||
});
|
||||
const agent1 = await client1.createAgent({
|
||||
provider: "claude",
|
||||
model: "claude-haiku-4-5",
|
||||
cwd: "/tmp",
|
||||
title: "Agent 1",
|
||||
});
|
||||
|
||||
await new Promise((r) => setTimeout(r, 100));
|
||||
const agent2 = await client1.createAgent({
|
||||
provider: "claude",
|
||||
model: "claude-haiku-4-5",
|
||||
cwd: "/tmp",
|
||||
title: "Agent 2",
|
||||
});
|
||||
|
||||
const attentionPromise = waitForAttentionRequired(client1, agent.id);
|
||||
await client1.sendMessage(agent.id, "Say 'hello' and nothing else");
|
||||
// User is looking at agent2, not agent1
|
||||
client1.sendHeartbeat({
|
||||
deviceType: "web",
|
||||
focusedAgentId: agent2.id,
|
||||
lastActivityAt: new Date().toISOString(),
|
||||
appVisible: true,
|
||||
});
|
||||
|
||||
const attention = await attentionPromise;
|
||||
await new Promise((r) => setTimeout(r, 100));
|
||||
|
||||
expect(attention.reason).toBe("finished");
|
||||
expect(attention.shouldNotify).toBe(true);
|
||||
}, 120000);
|
||||
const attentionPromise = waitForAttentionRequired(client1, agent1.id);
|
||||
await client1.sendMessage(agent1.id, "Say 'hello' and nothing else");
|
||||
|
||||
test("shouldNotify is true when client is focused on different agent", async () => {
|
||||
client1 = await createClient();
|
||||
const attention = await attentionPromise;
|
||||
|
||||
const agent1 = await client1.createAgent({
|
||||
provider: "codex",
|
||||
cwd: "/tmp",
|
||||
title: "Agent 1",
|
||||
});
|
||||
expect(attention.reason).toBe("finished");
|
||||
expect(attention.shouldNotify).toBe(true);
|
||||
}, 120000);
|
||||
|
||||
const agent2 = await client1.createAgent({
|
||||
provider: "codex",
|
||||
cwd: "/tmp",
|
||||
title: "Agent 2",
|
||||
});
|
||||
test("no notification when app is not visible but activity is recent (user just switched tabs)", async () => {
|
||||
client1 = await createClient();
|
||||
|
||||
client1.sendHeartbeat({
|
||||
focusedAgentId: agent2.id,
|
||||
lastActivityAt: new Date().toISOString(),
|
||||
});
|
||||
const agent = await client1.createAgent({
|
||||
provider: "claude",
|
||||
model: "claude-haiku-4-5",
|
||||
cwd: "/tmp",
|
||||
title: "App Hidden Test",
|
||||
});
|
||||
|
||||
await new Promise((r) => setTimeout(r, 100));
|
||||
// User switched away from the app but was active recently - they're still at the computer
|
||||
client1.sendHeartbeat({
|
||||
deviceType: "web",
|
||||
focusedAgentId: null, // null because app not visible
|
||||
lastActivityAt: new Date().toISOString(),
|
||||
appVisible: false,
|
||||
});
|
||||
|
||||
const attentionPromise = waitForAttentionRequired(client1, agent1.id);
|
||||
await client1.sendMessage(agent1.id, "Say 'hello' and nothing else");
|
||||
await new Promise((r) => setTimeout(r, 100));
|
||||
|
||||
const attention = await attentionPromise;
|
||||
const attentionPromise = waitForAttentionRequired(client1, agent.id);
|
||||
await client1.sendMessage(agent.id, "Say 'hello' and nothing else");
|
||||
|
||||
expect(attention.reason).toBe("finished");
|
||||
expect(attention.shouldNotify).toBe(true);
|
||||
}, 120000);
|
||||
const attention = await attentionPromise;
|
||||
|
||||
test("shouldNotify is false when another client is active on agent", async () => {
|
||||
client1 = await createClient();
|
||||
client2 = await createClient();
|
||||
expect(attention.reason).toBe("finished");
|
||||
expect(attention.shouldNotify).toBe(false);
|
||||
}, 120000);
|
||||
|
||||
const agent = await client1.createAgent({
|
||||
provider: "codex",
|
||||
cwd: "/tmp",
|
||||
title: "Multi-client Test",
|
||||
});
|
||||
test("notification when activity is stale (user walked away for 2+ minutes)", async () => {
|
||||
client1 = await createClient();
|
||||
|
||||
client1.sendHeartbeat({
|
||||
focusedAgentId: null,
|
||||
lastActivityAt: new Date(Date.now() - 120_000).toISOString(),
|
||||
});
|
||||
const agent = await client1.createAgent({
|
||||
provider: "claude",
|
||||
model: "claude-haiku-4-5",
|
||||
cwd: "/tmp",
|
||||
title: "Stale Activity Test",
|
||||
});
|
||||
|
||||
client2.sendHeartbeat({
|
||||
focusedAgentId: agent.id,
|
||||
lastActivityAt: new Date().toISOString(),
|
||||
});
|
||||
// User had agent focused but no activity for 2+ minutes (stale threshold)
|
||||
const staleTime = new Date(Date.now() - 125_000).toISOString(); // 2min 5sec ago
|
||||
client1.sendHeartbeat({
|
||||
deviceType: "web",
|
||||
focusedAgentId: agent.id,
|
||||
lastActivityAt: staleTime,
|
||||
appVisible: true,
|
||||
});
|
||||
|
||||
await new Promise((r) => setTimeout(r, 100));
|
||||
await new Promise((r) => setTimeout(r, 100));
|
||||
|
||||
const attentionPromise = waitForAttentionRequired(client1, agent.id);
|
||||
await client1.sendMessage(agent.id, "Say 'hello' and nothing else");
|
||||
const attentionPromise = waitForAttentionRequired(client1, agent.id);
|
||||
await client1.sendMessage(agent.id, "Say 'hello' and nothing else");
|
||||
|
||||
const attention = await attentionPromise;
|
||||
const attention = await attentionPromise;
|
||||
|
||||
expect(attention.reason).toBe("finished");
|
||||
expect(attention.shouldNotify).toBe(false);
|
||||
}, 120000);
|
||||
expect(attention.reason).toBe("finished");
|
||||
expect(attention.shouldNotify).toBe(true);
|
||||
}, 120000);
|
||||
|
||||
test("shouldNotify is true when no heartbeat received", async () => {
|
||||
client1 = await createClient();
|
||||
test("notification when no heartbeat received (legacy/new client)", async () => {
|
||||
client1 = await createClient();
|
||||
|
||||
const agent = await client1.createAgent({
|
||||
provider: "codex",
|
||||
cwd: "/tmp",
|
||||
title: "No Heartbeat Test",
|
||||
});
|
||||
const agent = await client1.createAgent({
|
||||
provider: "claude",
|
||||
model: "claude-haiku-4-5",
|
||||
cwd: "/tmp",
|
||||
title: "No Heartbeat Test",
|
||||
});
|
||||
|
||||
const attentionPromise = waitForAttentionRequired(client1, agent.id);
|
||||
await client1.sendMessage(agent.id, "Say 'hello' and nothing else");
|
||||
// Don't send any heartbeat - simulate legacy client
|
||||
|
||||
const attention = await attentionPromise;
|
||||
const attentionPromise = waitForAttentionRequired(client1, agent.id);
|
||||
await client1.sendMessage(agent.id, "Say 'hello' and nothing else");
|
||||
|
||||
expect(attention.reason).toBe("finished");
|
||||
expect(attention.shouldNotify).toBe(true);
|
||||
}, 120000);
|
||||
const attention = await attentionPromise;
|
||||
|
||||
expect(attention.reason).toBe("finished");
|
||||
expect(attention.shouldNotify).toBe(true);
|
||||
}, 120000);
|
||||
});
|
||||
|
||||
// ===========================================================================
|
||||
// TWO CLIENTS - SAME DEVICE TYPE (e.g., two browser tabs)
|
||||
// ===========================================================================
|
||||
|
||||
describe("two web clients", () => {
|
||||
test("no notification when other web client is active on agent", async () => {
|
||||
client1 = await createClient();
|
||||
client2 = await createClient();
|
||||
|
||||
const agent = await client1.createAgent({
|
||||
provider: "claude",
|
||||
model: "claude-haiku-4-5",
|
||||
cwd: "/tmp",
|
||||
title: "Two Tabs Test",
|
||||
});
|
||||
|
||||
// Client 1: not focused on agent
|
||||
client1.sendHeartbeat({
|
||||
deviceType: "web",
|
||||
focusedAgentId: null,
|
||||
lastActivityAt: new Date().toISOString(),
|
||||
appVisible: false,
|
||||
});
|
||||
|
||||
// Client 2: actively focused on agent
|
||||
client2.sendHeartbeat({
|
||||
deviceType: "web",
|
||||
focusedAgentId: agent.id,
|
||||
lastActivityAt: new Date().toISOString(),
|
||||
appVisible: true,
|
||||
});
|
||||
|
||||
await new Promise((r) => setTimeout(r, 100));
|
||||
|
||||
const attention1Promise = waitForAttentionRequired(client1, agent.id);
|
||||
const attention2Promise = waitForAttentionRequired(client2, agent.id);
|
||||
await client1.sendMessage(agent.id, "Say 'hello' and nothing else");
|
||||
|
||||
const [attention1, attention2] = await Promise.all([
|
||||
attention1Promise,
|
||||
attention2Promise,
|
||||
]);
|
||||
|
||||
// Neither should notify - user is actively watching on client2
|
||||
expect(attention1.shouldNotify).toBe(false);
|
||||
expect(attention2.shouldNotify).toBe(false);
|
||||
}, 120000);
|
||||
|
||||
test("both notify when both web clients are inactive", async () => {
|
||||
client1 = await createClient();
|
||||
client2 = await createClient();
|
||||
|
||||
const agent = await client1.createAgent({
|
||||
provider: "claude",
|
||||
model: "claude-haiku-4-5",
|
||||
cwd: "/tmp",
|
||||
title: "Both Inactive Test",
|
||||
});
|
||||
|
||||
const staleTime = new Date(Date.now() - 120_000).toISOString();
|
||||
|
||||
// Both clients stale
|
||||
client1.sendHeartbeat({
|
||||
deviceType: "web",
|
||||
focusedAgentId: null,
|
||||
lastActivityAt: staleTime,
|
||||
appVisible: false,
|
||||
});
|
||||
|
||||
client2.sendHeartbeat({
|
||||
deviceType: "web",
|
||||
focusedAgentId: null,
|
||||
lastActivityAt: staleTime,
|
||||
appVisible: false,
|
||||
});
|
||||
|
||||
await new Promise((r) => setTimeout(r, 100));
|
||||
|
||||
const attention1Promise = waitForAttentionRequired(client1, agent.id);
|
||||
const attention2Promise = waitForAttentionRequired(client2, agent.id);
|
||||
await client1.sendMessage(agent.id, "Say 'hello' and nothing else");
|
||||
|
||||
const [attention1, attention2] = await Promise.all([
|
||||
attention1Promise,
|
||||
attention2Promise,
|
||||
]);
|
||||
|
||||
// Both should notify - no one is watching
|
||||
expect(attention1.shouldNotify).toBe(true);
|
||||
expect(attention2.shouldNotify).toBe(true);
|
||||
}, 120000);
|
||||
});
|
||||
|
||||
// ===========================================================================
|
||||
// WEB + MOBILE - DEVICE PRIORITY SCENARIOS
|
||||
// ===========================================================================
|
||||
|
||||
describe("web and mobile clients", () => {
|
||||
test("no notification to either when user actively on agent (web)", async () => {
|
||||
client1 = await createClient(); // web
|
||||
client2 = await createClient(); // mobile
|
||||
|
||||
const agent = await client1.createAgent({
|
||||
provider: "claude",
|
||||
model: "claude-haiku-4-5",
|
||||
cwd: "/tmp",
|
||||
title: "Web Active Test",
|
||||
});
|
||||
|
||||
// Web: actively focused
|
||||
client1.sendHeartbeat({
|
||||
deviceType: "web",
|
||||
focusedAgentId: agent.id,
|
||||
lastActivityAt: new Date().toISOString(),
|
||||
appVisible: true,
|
||||
});
|
||||
|
||||
// Mobile: app in background
|
||||
client2.sendHeartbeat({
|
||||
deviceType: "mobile",
|
||||
focusedAgentId: null,
|
||||
lastActivityAt: new Date(Date.now() - 300_000).toISOString(), // 5 min ago
|
||||
appVisible: false,
|
||||
});
|
||||
|
||||
await new Promise((r) => setTimeout(r, 100));
|
||||
|
||||
const attention1Promise = waitForAttentionRequired(client1, agent.id);
|
||||
const attention2Promise = waitForAttentionRequired(client2, agent.id);
|
||||
await client1.sendMessage(agent.id, "Say 'hello' and nothing else");
|
||||
|
||||
const [attention1, attention2] = await Promise.all([
|
||||
attention1Promise,
|
||||
attention2Promise,
|
||||
]);
|
||||
|
||||
// Neither should notify - user sees it on web
|
||||
expect(attention1.shouldNotify).toBe(false);
|
||||
expect(attention2.shouldNotify).toBe(false);
|
||||
}, 120000);
|
||||
|
||||
test("no notification to either when user actively on agent (mobile)", async () => {
|
||||
client1 = await createClient(); // web
|
||||
client2 = await createClient(); // mobile
|
||||
|
||||
const agent = await client1.createAgent({
|
||||
provider: "claude",
|
||||
model: "claude-haiku-4-5",
|
||||
cwd: "/tmp",
|
||||
title: "Mobile Active Test",
|
||||
});
|
||||
|
||||
// Web: tab hidden, stale
|
||||
client1.sendHeartbeat({
|
||||
deviceType: "web",
|
||||
focusedAgentId: null,
|
||||
lastActivityAt: new Date(Date.now() - 120_000).toISOString(),
|
||||
appVisible: false,
|
||||
});
|
||||
|
||||
// Mobile: actively focused on agent
|
||||
client2.sendHeartbeat({
|
||||
deviceType: "mobile",
|
||||
focusedAgentId: agent.id,
|
||||
lastActivityAt: new Date().toISOString(),
|
||||
appVisible: true,
|
||||
});
|
||||
|
||||
await new Promise((r) => setTimeout(r, 100));
|
||||
|
||||
const attention1Promise = waitForAttentionRequired(client1, agent.id);
|
||||
const attention2Promise = waitForAttentionRequired(client2, agent.id);
|
||||
await client1.sendMessage(agent.id, "Say 'hello' and nothing else");
|
||||
|
||||
const [attention1, attention2] = await Promise.all([
|
||||
attention1Promise,
|
||||
attention2Promise,
|
||||
]);
|
||||
|
||||
// Neither should notify - user sees it on mobile
|
||||
expect(attention1.shouldNotify).toBe(false);
|
||||
expect(attention2.shouldNotify).toBe(false);
|
||||
}, 120000);
|
||||
|
||||
test("notify mobile only when web is stale", async () => {
|
||||
client1 = await createClient(); // web
|
||||
client2 = await createClient(); // mobile
|
||||
|
||||
const agent = await client1.createAgent({
|
||||
provider: "claude",
|
||||
model: "claude-haiku-4-5",
|
||||
cwd: "/tmp",
|
||||
title: "Web Stale Test",
|
||||
});
|
||||
|
||||
// Web: stale (user walked away from computer)
|
||||
client1.sendHeartbeat({
|
||||
deviceType: "web",
|
||||
focusedAgentId: agent.id, // was looking at agent
|
||||
lastActivityAt: new Date(Date.now() - 120_000).toISOString(), // but 2 min ago
|
||||
appVisible: true,
|
||||
});
|
||||
|
||||
// Mobile: connected but not active (phone in pocket)
|
||||
client2.sendHeartbeat({
|
||||
deviceType: "mobile",
|
||||
focusedAgentId: null,
|
||||
lastActivityAt: new Date(Date.now() - 300_000).toISOString(), // 5 min ago
|
||||
appVisible: false,
|
||||
});
|
||||
|
||||
await new Promise((r) => setTimeout(r, 100));
|
||||
|
||||
const attention1Promise = waitForAttentionRequired(client1, agent.id);
|
||||
const attention2Promise = waitForAttentionRequired(client2, agent.id);
|
||||
await client1.sendMessage(agent.id, "Say 'hello' and nothing else");
|
||||
|
||||
const [attention1, attention2] = await Promise.all([
|
||||
attention1Promise,
|
||||
attention2Promise,
|
||||
]);
|
||||
|
||||
// Web stale → don't notify web, notify mobile instead
|
||||
expect(attention1.shouldNotify).toBe(false);
|
||||
expect(attention2.shouldNotify).toBe(true);
|
||||
}, 120000);
|
||||
|
||||
test("notify web when user active on web but looking at different agent", async () => {
|
||||
client1 = await createClient(); // web
|
||||
client2 = await createClient(); // mobile
|
||||
|
||||
const agent1 = await client1.createAgent({
|
||||
provider: "claude",
|
||||
model: "claude-haiku-4-5",
|
||||
cwd: "/tmp",
|
||||
title: "Agent 1",
|
||||
});
|
||||
|
||||
const agent2 = await client1.createAgent({
|
||||
provider: "claude",
|
||||
model: "claude-haiku-4-5",
|
||||
cwd: "/tmp",
|
||||
title: "Agent 2",
|
||||
});
|
||||
|
||||
// Web: active but looking at agent2
|
||||
client1.sendHeartbeat({
|
||||
deviceType: "web",
|
||||
focusedAgentId: agent2.id,
|
||||
lastActivityAt: new Date().toISOString(),
|
||||
appVisible: true,
|
||||
});
|
||||
|
||||
// Mobile: not active
|
||||
client2.sendHeartbeat({
|
||||
deviceType: "mobile",
|
||||
focusedAgentId: null,
|
||||
lastActivityAt: new Date(Date.now() - 300_000).toISOString(),
|
||||
appVisible: false,
|
||||
});
|
||||
|
||||
await new Promise((r) => setTimeout(r, 100));
|
||||
|
||||
const attention1Promise = waitForAttentionRequired(client1, agent1.id);
|
||||
const attention2Promise = waitForAttentionRequired(client2, agent1.id);
|
||||
await client1.sendMessage(agent1.id, "Say 'hello' and nothing else");
|
||||
|
||||
const [attention1, attention2] = await Promise.all([
|
||||
attention1Promise,
|
||||
attention2Promise,
|
||||
]);
|
||||
|
||||
// User is active on web, notify them there (they'll see it)
|
||||
expect(attention1.shouldNotify).toBe(true);
|
||||
// Don't also notify mobile - user is at the computer
|
||||
expect(attention2.shouldNotify).toBe(false);
|
||||
}, 120000);
|
||||
|
||||
test("notify both when both devices inactive and no one watching agent", async () => {
|
||||
client1 = await createClient(); // web
|
||||
client2 = await createClient(); // mobile
|
||||
|
||||
const agent = await client1.createAgent({
|
||||
provider: "claude",
|
||||
model: "claude-haiku-4-5",
|
||||
cwd: "/tmp",
|
||||
title: "Both Inactive Test",
|
||||
});
|
||||
|
||||
const staleTime = new Date(Date.now() - 120_000).toISOString();
|
||||
|
||||
// Web: stale
|
||||
client1.sendHeartbeat({
|
||||
deviceType: "web",
|
||||
focusedAgentId: null,
|
||||
lastActivityAt: staleTime,
|
||||
appVisible: false,
|
||||
});
|
||||
|
||||
// Mobile: also stale (but mobile should still notify as fallback)
|
||||
client2.sendHeartbeat({
|
||||
deviceType: "mobile",
|
||||
focusedAgentId: null,
|
||||
lastActivityAt: staleTime,
|
||||
appVisible: false,
|
||||
});
|
||||
|
||||
await new Promise((r) => setTimeout(r, 100));
|
||||
|
||||
const attention1Promise = waitForAttentionRequired(client1, agent.id);
|
||||
const attention2Promise = waitForAttentionRequired(client2, agent.id);
|
||||
await client1.sendMessage(agent.id, "Say 'hello' and nothing else");
|
||||
|
||||
const [attention1, attention2] = await Promise.all([
|
||||
attention1Promise,
|
||||
attention2Promise,
|
||||
]);
|
||||
|
||||
// Mobile always notifies when no one is watching
|
||||
// Web is stale so don't bother notifying there
|
||||
expect(attention1.shouldNotify).toBe(false); // web stale
|
||||
expect(attention2.shouldNotify).toBe(true); // mobile fallback
|
||||
}, 120000);
|
||||
});
|
||||
|
||||
// ===========================================================================
|
||||
// EDGE CASES
|
||||
// ===========================================================================
|
||||
|
||||
describe("edge cases", () => {
|
||||
test("mobile notifies even with no heartbeat when web is stale", async () => {
|
||||
client1 = await createClient(); // web
|
||||
client2 = await createClient(); // mobile - no heartbeat
|
||||
|
||||
const agent = await client1.createAgent({
|
||||
provider: "claude",
|
||||
model: "claude-haiku-4-5",
|
||||
cwd: "/tmp",
|
||||
title: "Mobile No Heartbeat Test",
|
||||
});
|
||||
|
||||
// Web: stale
|
||||
client1.sendHeartbeat({
|
||||
deviceType: "web",
|
||||
focusedAgentId: agent.id,
|
||||
lastActivityAt: new Date(Date.now() - 120_000).toISOString(),
|
||||
appVisible: true,
|
||||
});
|
||||
|
||||
// Mobile: never sent heartbeat (new connection)
|
||||
// Should still receive notification as fallback
|
||||
|
||||
await new Promise((r) => setTimeout(r, 100));
|
||||
|
||||
const attention1Promise = waitForAttentionRequired(client1, agent.id);
|
||||
const attention2Promise = waitForAttentionRequired(client2, agent.id);
|
||||
await client1.sendMessage(agent.id, "Say 'hello' and nothing else");
|
||||
|
||||
const [attention1, attention2] = await Promise.all([
|
||||
attention1Promise,
|
||||
attention2Promise,
|
||||
]);
|
||||
|
||||
// Web stale → no notification
|
||||
// Mobile has no heartbeat → treat as should notify (we don't know device type)
|
||||
expect(attention1.shouldNotify).toBe(false);
|
||||
expect(attention2.shouldNotify).toBe(true);
|
||||
}, 120000);
|
||||
|
||||
test("no notification when app not visible but activity recent (switched tabs recently)", async () => {
|
||||
client1 = await createClient();
|
||||
|
||||
const agent = await client1.createAgent({
|
||||
provider: "claude",
|
||||
model: "claude-haiku-4-5",
|
||||
cwd: "/tmp",
|
||||
title: "Tab Switch Test",
|
||||
});
|
||||
|
||||
// User just switched tabs but was active 10 seconds ago - still at computer
|
||||
client1.sendHeartbeat({
|
||||
deviceType: "web",
|
||||
focusedAgentId: null, // not focused - tab hidden
|
||||
lastActivityAt: new Date(Date.now() - 10_000).toISOString(), // 10s ago
|
||||
appVisible: false,
|
||||
});
|
||||
|
||||
await new Promise((r) => setTimeout(r, 100));
|
||||
|
||||
const attentionPromise = waitForAttentionRequired(client1, agent.id);
|
||||
await client1.sendMessage(agent.id, "Say 'hello' and nothing else");
|
||||
|
||||
const attention = await attentionPromise;
|
||||
|
||||
// User is at the computer (recent activity) - no notification needed
|
||||
expect(attention.shouldNotify).toBe(false);
|
||||
}, 120000);
|
||||
|
||||
test("no notification to either when both have recent activity (user is present)", async () => {
|
||||
client1 = await createClient(); // web
|
||||
client2 = await createClient(); // mobile
|
||||
|
||||
const agent = await client1.createAgent({
|
||||
provider: "claude",
|
||||
model: "claude-haiku-4-5",
|
||||
cwd: "/tmp",
|
||||
title: "Both Recent Activity Test",
|
||||
});
|
||||
|
||||
// Web: tab hidden but recent activity (user at computer, switched tabs)
|
||||
client1.sendHeartbeat({
|
||||
deviceType: "web",
|
||||
focusedAgentId: null,
|
||||
lastActivityAt: new Date(Date.now() - 30_000).toISOString(), // 30s ago
|
||||
appVisible: false,
|
||||
});
|
||||
|
||||
// Mobile: connected, also recent activity
|
||||
client2.sendHeartbeat({
|
||||
deviceType: "mobile",
|
||||
focusedAgentId: null,
|
||||
lastActivityAt: new Date(Date.now() - 30_000).toISOString(),
|
||||
appVisible: false,
|
||||
});
|
||||
|
||||
await new Promise((r) => setTimeout(r, 100));
|
||||
|
||||
const attention1Promise = waitForAttentionRequired(client1, agent.id);
|
||||
const attention2Promise = waitForAttentionRequired(client2, agent.id);
|
||||
await client1.sendMessage(agent.id, "Say 'hello' and nothing else");
|
||||
|
||||
const [attention1, attention2] = await Promise.all([
|
||||
attention1Promise,
|
||||
attention2Promise,
|
||||
]);
|
||||
|
||||
// Both have recent activity - user is present, no notification needed
|
||||
expect(attention1.shouldNotify).toBe(false);
|
||||
expect(attention2.shouldNotify).toBe(false);
|
||||
}, 120000);
|
||||
});
|
||||
});
|
||||
|
||||
@@ -0,0 +1,111 @@
|
||||
import type { PushTokenStore } from "./token-store.js";
|
||||
|
||||
interface PushPayload {
|
||||
title: string;
|
||||
body: string;
|
||||
data?: Record<string, unknown>;
|
||||
}
|
||||
|
||||
interface ExpoPushMessage {
|
||||
to: string;
|
||||
title: string;
|
||||
body: string;
|
||||
data?: Record<string, unknown>;
|
||||
sound?: "default";
|
||||
}
|
||||
|
||||
interface ExpoPushTicket {
|
||||
status: "ok" | "error";
|
||||
id?: string;
|
||||
message?: string;
|
||||
details?: { error?: string };
|
||||
}
|
||||
|
||||
const EXPO_PUSH_URL = "https://exp.host/--/api/v2/push/send";
|
||||
const MAX_BATCH_SIZE = 100;
|
||||
|
||||
/**
|
||||
* Service for sending Expo push notifications.
|
||||
* Handles batching and invalid token removal.
|
||||
*/
|
||||
export class PushService {
|
||||
private tokenStore: PushTokenStore;
|
||||
|
||||
constructor(tokenStore: PushTokenStore) {
|
||||
this.tokenStore = tokenStore;
|
||||
}
|
||||
|
||||
async sendPush(tokens: string[], payload: PushPayload): Promise<void> {
|
||||
if (tokens.length === 0) {
|
||||
return;
|
||||
}
|
||||
|
||||
const messages: ExpoPushMessage[] = tokens.map((token) => ({
|
||||
to: token,
|
||||
title: payload.title,
|
||||
body: payload.body,
|
||||
data: payload.data,
|
||||
sound: "default",
|
||||
}));
|
||||
|
||||
// Batch tokens (max 100 per request per Expo limits)
|
||||
const batches: ExpoPushMessage[][] = [];
|
||||
for (let i = 0; i < messages.length; i += MAX_BATCH_SIZE) {
|
||||
batches.push(messages.slice(i, i + MAX_BATCH_SIZE));
|
||||
}
|
||||
|
||||
for (const batch of batches) {
|
||||
await this.sendBatch(batch);
|
||||
}
|
||||
}
|
||||
|
||||
private async sendBatch(messages: ExpoPushMessage[]): Promise<void> {
|
||||
try {
|
||||
const response = await fetch(EXPO_PUSH_URL, {
|
||||
method: "POST",
|
||||
headers: {
|
||||
"Content-Type": "application/json",
|
||||
Accept: "application/json",
|
||||
},
|
||||
body: JSON.stringify(messages),
|
||||
});
|
||||
|
||||
if (!response.ok) {
|
||||
console.error(
|
||||
`[PushService] Expo push API error: ${response.status} ${response.statusText}`
|
||||
);
|
||||
return;
|
||||
}
|
||||
|
||||
const result = (await response.json()) as { data: ExpoPushTicket[] };
|
||||
this.handleTickets(messages, result.data);
|
||||
} catch (error) {
|
||||
console.error("[PushService] Failed to send push notifications:", error);
|
||||
}
|
||||
}
|
||||
|
||||
private handleTickets(
|
||||
messages: ExpoPushMessage[],
|
||||
tickets: ExpoPushTicket[]
|
||||
): void {
|
||||
for (let i = 0; i < tickets.length; i++) {
|
||||
const ticket = tickets[i];
|
||||
const message = messages[i];
|
||||
|
||||
if (ticket.status === "error") {
|
||||
console.error(
|
||||
`[PushService] Push failed for token: ${ticket.message}`,
|
||||
ticket.details
|
||||
);
|
||||
|
||||
// Remove invalid tokens
|
||||
if (
|
||||
ticket.details?.error === "DeviceNotRegistered" ||
|
||||
ticket.details?.error === "InvalidCredentials"
|
||||
) {
|
||||
this.tokenStore.removeToken(message.to);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,23 @@
|
||||
/**
|
||||
* Simple in-memory store for Expo push tokens.
|
||||
* Tokens are used to send push notifications when all clients are stale.
|
||||
*/
|
||||
export class PushTokenStore {
|
||||
private tokens: Set<string> = new Set();
|
||||
|
||||
addToken(token: string): void {
|
||||
this.tokens.add(token);
|
||||
console.log(`[PushTokenStore] Added token (total: ${this.tokens.size})`);
|
||||
}
|
||||
|
||||
removeToken(token: string): void {
|
||||
const deleted = this.tokens.delete(token);
|
||||
if (deleted) {
|
||||
console.log(`[PushTokenStore] Removed token (total: ${this.tokens.size})`);
|
||||
}
|
||||
}
|
||||
|
||||
getAllTokens(): string[] {
|
||||
return Array.from(this.tokens);
|
||||
}
|
||||
}
|
||||
@@ -57,6 +57,7 @@ import {
|
||||
getDownloadableFileInfo,
|
||||
} from "./file-explorer/service.js";
|
||||
import { DownloadTokenStore } from "./file-download/token-store.js";
|
||||
import { PushTokenStore } from "./push/token-store.js";
|
||||
import {
|
||||
generateAgentTitle,
|
||||
isTitleGeneratorInitialized,
|
||||
@@ -230,17 +231,21 @@ export class Session {
|
||||
private readonly agentRegistry: AgentRegistry;
|
||||
private readonly agentMcpConfig: AgentMcpClientConfig;
|
||||
private readonly downloadTokenStore: DownloadTokenStore;
|
||||
private readonly pushTokenStore: PushTokenStore;
|
||||
private agentTitleCache: Map<string, string | null> = new Map();
|
||||
private unsubscribeAgentEvents: (() => void) | null = null;
|
||||
private clientActivity: {
|
||||
deviceType: "web" | "mobile";
|
||||
focusedAgentId: string | null;
|
||||
lastActivityAt: Date;
|
||||
appVisible: boolean;
|
||||
} | null = null;
|
||||
|
||||
constructor(
|
||||
clientId: string,
|
||||
onMessage: (msg: SessionOutboundMessage) => void,
|
||||
downloadTokenStore: DownloadTokenStore,
|
||||
pushTokenStore: PushTokenStore,
|
||||
agentManager: AgentManager,
|
||||
agentRegistry: AgentRegistry,
|
||||
agentMcpConfig: AgentMcpClientConfig,
|
||||
@@ -253,6 +258,7 @@ export class Session {
|
||||
this.conversationId = options?.conversationId || uuidv4();
|
||||
this.onMessage = onMessage;
|
||||
this.downloadTokenStore = downloadTokenStore;
|
||||
this.pushTokenStore = pushTokenStore;
|
||||
this.agentManager = agentManager;
|
||||
this.agentRegistry = agentRegistry;
|
||||
this.agentMcpConfig = agentMcpConfig;
|
||||
@@ -290,8 +296,10 @@ export class Session {
|
||||
* Get the client's current activity state
|
||||
*/
|
||||
public getClientActivity(): {
|
||||
deviceType: "web" | "mobile";
|
||||
focusedAgentId: string | null;
|
||||
lastActivityAt: Date;
|
||||
appVisible: boolean;
|
||||
} | null {
|
||||
return this.clientActivity;
|
||||
}
|
||||
@@ -816,6 +824,10 @@ export class Session {
|
||||
case "list_commands_request":
|
||||
await this.handleListCommandsRequest(msg.agentId, msg.requestId);
|
||||
break;
|
||||
|
||||
case "register_push_token":
|
||||
this.handleRegisterPushToken(msg.token);
|
||||
break;
|
||||
}
|
||||
} catch (error: any) {
|
||||
console.error(
|
||||
@@ -1865,15 +1877,28 @@ export class Session {
|
||||
* Handle client heartbeat for activity tracking
|
||||
*/
|
||||
private handleClientHeartbeat(msg: {
|
||||
deviceType: "web" | "mobile";
|
||||
focusedAgentId: string | null;
|
||||
lastActivityAt: string;
|
||||
appVisible: boolean;
|
||||
}): void {
|
||||
console.log(`[Session ${this.clientId}] handleClientHeartbeat`, msg);
|
||||
this.clientActivity = {
|
||||
deviceType: msg.deviceType,
|
||||
focusedAgentId: msg.focusedAgentId,
|
||||
lastActivityAt: new Date(msg.lastActivityAt),
|
||||
appVisible: msg.appVisible,
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* Handle push token registration
|
||||
*/
|
||||
private handleRegisterPushToken(token: string): void {
|
||||
this.pushTokenStore.addToken(token);
|
||||
console.log(`[Session ${this.clientId}] Registered push token`);
|
||||
}
|
||||
|
||||
/**
|
||||
* Handle list commands request for an agent
|
||||
*/
|
||||
|
||||
@@ -14,6 +14,8 @@ import { AgentManager } from "./agent/agent-manager.js";
|
||||
import { AgentRegistry } from "./agent/agent-registry.js";
|
||||
import type { AgentProvider } from "./agent/agent-sdk-types.js";
|
||||
import { DownloadTokenStore } from "./file-download/token-store.js";
|
||||
import { PushTokenStore } from "./push/token-store.js";
|
||||
import { PushService } from "./push/push-service.js";
|
||||
|
||||
type AgentMcpClientConfig = {
|
||||
agentMcpUrl: string;
|
||||
@@ -32,6 +34,8 @@ export class VoiceAssistantWebSocketServer {
|
||||
private agentManager: AgentManager;
|
||||
private agentRegistry: AgentRegistry;
|
||||
private downloadTokenStore: DownloadTokenStore;
|
||||
private pushTokenStore: PushTokenStore;
|
||||
private pushService: PushService;
|
||||
private readonly agentMcpConfig: AgentMcpClientConfig;
|
||||
|
||||
constructor(
|
||||
@@ -44,6 +48,8 @@ export class VoiceAssistantWebSocketServer {
|
||||
this.agentManager = agentManager;
|
||||
this.agentRegistry = agentRegistry;
|
||||
this.downloadTokenStore = downloadTokenStore;
|
||||
this.pushTokenStore = new PushTokenStore();
|
||||
this.pushService = new PushService(this.pushTokenStore);
|
||||
this.agentMcpConfig = agentMcpConfig;
|
||||
this.wss = new WebSocketServer({ server, path: "/ws" });
|
||||
|
||||
@@ -95,6 +101,7 @@ export class VoiceAssistantWebSocketServer {
|
||||
this.sendToClient(ws, wrapSessionMessage(msg));
|
||||
},
|
||||
this.downloadTokenStore,
|
||||
this.pushTokenStore,
|
||||
this.agentManager,
|
||||
this.agentRegistry,
|
||||
this.agentMcpConfig,
|
||||
@@ -315,52 +322,189 @@ export class VoiceAssistantWebSocketServer {
|
||||
this.wss.close();
|
||||
}
|
||||
|
||||
/**
|
||||
* Check if any connected client is actively viewing the specified agent
|
||||
*/
|
||||
private isAnyClientActiveOnAgent(agentId: string): boolean {
|
||||
const now = Date.now();
|
||||
const activityThresholdMs = 60_000;
|
||||
private readonly ACTIVITY_THRESHOLD_MS = 120_000; // 2 minutes
|
||||
|
||||
for (const [, session] of this.sessions) {
|
||||
const activity = session.getClientActivity();
|
||||
if (
|
||||
activity !== null &&
|
||||
activity.focusedAgentId === agentId &&
|
||||
now - activity.lastActivityAt.getTime() < activityThresholdMs
|
||||
) {
|
||||
return true;
|
||||
}
|
||||
/**
|
||||
* Get client activity state with computed staleness
|
||||
*/
|
||||
private getClientActivityState(session: Session): {
|
||||
deviceType: "web" | "mobile" | null;
|
||||
focusedAgentId: string | null;
|
||||
isStale: boolean;
|
||||
appVisible: boolean;
|
||||
} {
|
||||
const activity = session.getClientActivity();
|
||||
if (!activity) {
|
||||
console.log("[WS] getClientActivityState: no activity for session");
|
||||
return { deviceType: null, focusedAgentId: null, isStale: true, appVisible: false };
|
||||
}
|
||||
return false;
|
||||
const now = Date.now();
|
||||
const ageMs = now - activity.lastActivityAt.getTime();
|
||||
const isStale = ageMs >= this.ACTIVITY_THRESHOLD_MS;
|
||||
console.log("[WS] getClientActivityState", {
|
||||
deviceType: activity.deviceType,
|
||||
focusedAgentId: activity.focusedAgentId,
|
||||
lastActivityAt: activity.lastActivityAt.toISOString(),
|
||||
ageMs,
|
||||
isStale,
|
||||
appVisible: activity.appVisible,
|
||||
});
|
||||
return {
|
||||
deviceType: activity.deviceType,
|
||||
focusedAgentId: activity.focusedAgentId,
|
||||
isStale,
|
||||
appVisible: activity.appVisible,
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* Broadcast an attention_required event to all clients with shouldNotify computed
|
||||
* Compute shouldNotify for a specific client given all clients' states
|
||||
*
|
||||
* UX Rules:
|
||||
* 1. If ANY client is actively watching the agent (focused + visible + not stale) → no notifications
|
||||
* 2. If THIS client is not stale and focused elsewhere → notify THIS client (user is at computer)
|
||||
* 3. If THIS client is stale → only notify if mobile, or if no mobile available
|
||||
* 4. Don't notify non-stale clients that aren't focused on anything (just switched tabs)
|
||||
*/
|
||||
private computeShouldNotifyForClient(
|
||||
clientState: {
|
||||
deviceType: "web" | "mobile" | null;
|
||||
focusedAgentId: string | null;
|
||||
isStale: boolean;
|
||||
appVisible: boolean;
|
||||
},
|
||||
allClientStates: Array<{
|
||||
deviceType: "web" | "mobile" | null;
|
||||
focusedAgentId: string | null;
|
||||
isStale: boolean;
|
||||
appVisible: boolean;
|
||||
}>,
|
||||
agentId: string
|
||||
): boolean {
|
||||
// Rule 1: If any client is actively watching the agent, no one needs notification
|
||||
const isAnyoneActiveOnAgent = allClientStates.some(
|
||||
(state) =>
|
||||
state.focusedAgentId === agentId &&
|
||||
state.appVisible &&
|
||||
!state.isStale
|
||||
);
|
||||
if (isAnyoneActiveOnAgent) {
|
||||
return false;
|
||||
}
|
||||
|
||||
// No heartbeat (legacy client or just connected) → notify
|
||||
if (clientState.deviceType === null) {
|
||||
return true;
|
||||
}
|
||||
|
||||
// Rule 2: If THIS client is not stale and actively looking at a different agent → notify them
|
||||
if (!clientState.isStale && clientState.appVisible && clientState.focusedAgentId !== null) {
|
||||
return true;
|
||||
}
|
||||
|
||||
// Rule 3: If THIS client is not stale but just switched tabs (not focused on anything) → no notification
|
||||
// User is present at the computer, they'll come back
|
||||
if (!clientState.isStale) {
|
||||
return false;
|
||||
}
|
||||
|
||||
// Rule 4: THIS client is stale - check if another client will handle it
|
||||
const hasActiveWebClient = allClientStates.some(
|
||||
(state) => state.deviceType === "web" && !state.isStale
|
||||
);
|
||||
|
||||
if (clientState.deviceType === "mobile") {
|
||||
// Mobile only notifies if web is also stale (user truly away)
|
||||
// If web is active, they'll see it there
|
||||
return !hasActiveWebClient;
|
||||
}
|
||||
|
||||
if (clientState.deviceType === "web") {
|
||||
// Stale web: notify only if no other client can handle it
|
||||
// Other client = mobile or unknown (no heartbeat)
|
||||
const hasOtherClient = allClientStates.some(
|
||||
(state) => state !== clientState && (state.deviceType === "mobile" || state.deviceType === null)
|
||||
);
|
||||
return !hasOtherClient;
|
||||
}
|
||||
|
||||
// Fallback: notify
|
||||
return true;
|
||||
}
|
||||
|
||||
/**
|
||||
* Broadcast an attention_required event to all clients with per-client shouldNotify
|
||||
*/
|
||||
private broadcastAgentAttention(params: {
|
||||
agentId: string;
|
||||
provider: AgentProvider;
|
||||
reason: "finished" | "error" | "permission";
|
||||
}): void {
|
||||
const shouldNotify = !this.isAnyClientActiveOnAgent(params.agentId);
|
||||
// Collect all client states first
|
||||
const clientEntries: Array<{
|
||||
ws: WebSocket;
|
||||
state: {
|
||||
deviceType: "web" | "mobile" | null;
|
||||
focusedAgentId: string | null;
|
||||
isStale: boolean;
|
||||
appVisible: boolean;
|
||||
};
|
||||
}> = [];
|
||||
|
||||
const message = wrapSessionMessage({
|
||||
type: "agent_stream",
|
||||
payload: {
|
||||
agentId: params.agentId,
|
||||
event: {
|
||||
type: "attention_required",
|
||||
provider: params.provider,
|
||||
reason: params.reason,
|
||||
timestamp: new Date().toISOString(),
|
||||
shouldNotify,
|
||||
},
|
||||
timestamp: new Date().toISOString(),
|
||||
},
|
||||
for (const [ws, session] of this.sessions) {
|
||||
clientEntries.push({
|
||||
ws,
|
||||
state: this.getClientActivityState(session),
|
||||
});
|
||||
}
|
||||
|
||||
const allStates = clientEntries.map((e) => e.state);
|
||||
|
||||
console.log("[WS] broadcastAgentAttention", {
|
||||
agentId: params.agentId,
|
||||
reason: params.reason,
|
||||
clientCount: clientEntries.length,
|
||||
allStates,
|
||||
});
|
||||
|
||||
for (const [ws] of this.sessions) {
|
||||
// Check if all clients are stale - if so, send push notification
|
||||
const allClientsStale = allStates.every((state) => state.isStale);
|
||||
console.log("[WS] allClientsStale:", allClientsStale);
|
||||
if (allClientsStale) {
|
||||
const tokens = this.pushTokenStore.getAllTokens();
|
||||
console.log("[WS] Sending push notification, tokens:", tokens.length);
|
||||
if (tokens.length > 0) {
|
||||
void this.pushService.sendPush(tokens, {
|
||||
title: "Agent needs attention",
|
||||
body: `Reason: ${params.reason}`,
|
||||
data: { agentId: params.agentId },
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
// Send to each client with their specific shouldNotify value
|
||||
for (const { ws, state } of clientEntries) {
|
||||
const shouldNotify = this.computeShouldNotifyForClient(
|
||||
state,
|
||||
allStates,
|
||||
params.agentId
|
||||
);
|
||||
|
||||
const message = wrapSessionMessage({
|
||||
type: "agent_stream",
|
||||
payload: {
|
||||
agentId: params.agentId,
|
||||
event: {
|
||||
type: "attention_required",
|
||||
provider: params.provider,
|
||||
reason: params.reason,
|
||||
timestamp: new Date().toISOString(),
|
||||
shouldNotify,
|
||||
},
|
||||
timestamp: new Date().toISOString(),
|
||||
},
|
||||
});
|
||||
|
||||
this.sendToClient(ws, message);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -43,7 +43,11 @@ export async function generateAgentTitle(
|
||||
schema: z.object({
|
||||
title: z.string().describe("A concise 3-5 word title describing what the agent is working on"),
|
||||
}),
|
||||
prompt: `Generate a concise title for this agent session based on the activity. The title should describe what the agent is working on. Be specific but brief. Examples: "Fix Authentication Bug", "Build Dashboard Component", "Refactor API Routes".
|
||||
prompt: `Generate a concise title for this agent session. The title should describe what the user asked the agent to do.
|
||||
|
||||
IMPORTANT: Focus primarily on [User] messages to understand the task. User messages contain the actual request - use their words and intent to name the chat. Tool calls and assistant messages are just implementation details and should NOT drive the title.
|
||||
|
||||
Be specific but brief (3-5 words). Examples: "Fix Authentication Bug", "Build Dashboard Component", "Refactor API Routes".
|
||||
|
||||
Working directory: ${cwd}
|
||||
|
||||
|
||||
@@ -514,8 +514,10 @@ export const ClearAgentAttentionMessageSchema = z.object({
|
||||
|
||||
export const ClientHeartbeatMessageSchema = z.object({
|
||||
type: z.literal("client_heartbeat"),
|
||||
deviceType: z.enum(["web", "mobile"]),
|
||||
focusedAgentId: z.string().nullable(),
|
||||
lastActivityAt: z.string(),
|
||||
appVisible: z.boolean(),
|
||||
});
|
||||
|
||||
export const ListCommandsRequestSchema = z.object({
|
||||
@@ -524,6 +526,11 @@ export const ListCommandsRequestSchema = z.object({
|
||||
requestId: z.string(),
|
||||
});
|
||||
|
||||
export const RegisterPushTokenMessageSchema = z.object({
|
||||
type: z.literal("register_push_token"),
|
||||
token: z.string(),
|
||||
});
|
||||
|
||||
export const SessionInboundMessageSchema = z.discriminatedUnion("type", [
|
||||
UserTextMessageSchema,
|
||||
RealtimeAudioChunkMessageSchema,
|
||||
@@ -553,6 +560,7 @@ export const SessionInboundMessageSchema = z.discriminatedUnion("type", [
|
||||
ClearAgentAttentionMessageSchema,
|
||||
ClientHeartbeatMessageSchema,
|
||||
ListCommandsRequestSchema,
|
||||
RegisterPushTokenMessageSchema,
|
||||
]);
|
||||
|
||||
export type SessionInboundMessage = z.infer<typeof SessionInboundMessageSchema>;
|
||||
@@ -967,6 +975,7 @@ export type ClearAgentAttentionMessage = z.infer<typeof ClearAgentAttentionMessa
|
||||
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>;
|
||||
|
||||
// ============================================================================
|
||||
// WebSocket Level Messages (wraps session messages)
|
||||
|
||||
Reference in New Issue
Block a user