diff --git a/packages/server/src/client/daemon-client-v2.ts b/packages/server/src/client/daemon-client-v2.ts index 82cad8957..9469fc176 100644 --- a/packages/server/src/client/daemon-client-v2.ts +++ b/packages/server/src/client/daemon-client-v2.ts @@ -447,6 +447,17 @@ export class DaemonClientV2 { this.sendSessionMessage({ type: "clear_agent_attention", agentId }); } + sendHeartbeat(params: { + focusedAgentId: string | null; + lastActivityAt: string; + }): void { + this.sendSessionMessage({ + type: "client_heartbeat", + focusedAgentId: params.focusedAgentId, + lastActivityAt: params.lastActivityAt, + }); + } + // ============================================================================ // Conversation / Session RPC // ============================================================================ diff --git a/packages/server/src/server/agent/agent-manager.ts b/packages/server/src/server/agent/agent-manager.ts index 060a6a450..9c546ba9f 100644 --- a/packages/server/src/server/agent/agent-manager.ts +++ b/packages/server/src/server/agent/agent-manager.ts @@ -45,12 +45,19 @@ export type PersistedAgentQueryOptions = ListPersistedAgentsOptions & { provider?: AgentProvider; }; +export type AgentAttentionCallback = (params: { + agentId: string; + provider: AgentProvider; + reason: "finished" | "error" | "permission"; +}) => void; + export type AgentManagerOptions = { clients?: Partial>; maxTimelineItems?: number; idFactory?: () => string; registry?: AgentRegistry; agentControlMcp?: AgentControlMcpConfig; + onAgentAttention?: AgentAttentionCallback; }; export type WaitForAgentOptions = { @@ -176,6 +183,7 @@ export class AgentManager { private readonly registry?: AgentRegistry; private readonly previousStatuses = new Map(); private readonly agentControlMcp?: AgentControlMcpConfig; + private onAgentAttention?: AgentAttentionCallback; constructor(options?: AgentManagerOptions) { this.maxTimelineItems = @@ -183,6 +191,7 @@ export class AgentManager { this.idFactory = options?.idFactory ?? (() => randomUUID()); this.registry = options?.registry; this.agentControlMcp = options?.agentControlMcp; + this.onAgentAttention = options?.onAgentAttention; if (options?.clients) { for (const [provider, client] of Object.entries(options.clients)) { if (client) { @@ -196,6 +205,10 @@ export class AgentManager { this.clients.set(provider, client); } + setAgentAttentionCallback(callback: AgentAttentionCallback): void { + this.onAgentAttention = callback; + } + subscribe(callback: AgentSubscriber, options?: SubscribeOptions): () => void { const record: SubscriptionRecord = { callback, @@ -985,11 +998,10 @@ export class AgentManager { agent: ManagedAgent, reason: "finished" | "error" | "permission" ): void { - this.dispatchStream(agent.id, { - type: "attention_required", + this.onAgentAttention?.({ + agentId: agent.id, provider: agent.provider, reason, - timestamp: new Date().toISOString(), }); } diff --git a/packages/server/src/server/client-activity.e2e.test.ts b/packages/server/src/server/client-activity.e2e.test.ts new file mode 100644 index 000000000..72a391b39 --- /dev/null +++ b/packages/server/src/server/client-activity.e2e.test.ts @@ -0,0 +1,204 @@ +import { + describe, + test, + expect, + beforeEach, + afterEach, +} 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"; + +/** + * 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. + * + * Activity is determined by: + * - focusedAgentId: which agent the client is viewing + * - lastActivityAt: timestamp of last user interaction (must be within 60s) + */ +describe("client activity tracking", () => { + let daemon: TestPaseoDaemon; + let client1: DaemonClient; + let client2: DaemonClient; + + beforeEach(async () => { + daemon = await createTestPaseoDaemon(); + }); + + afterEach(async () => { + if (client1) await client1.close().catch(() => {}); + if (client2) await client2.close().catch(() => {}); + await daemon.close(); + }, 30000); + + async function createClient(): Promise { + const client = new DaemonClient({ + url: `ws://127.0.0.1:${daemon.port}/ws`, + authHeader: daemon.agentMcpAuthHeader, + messageQueueLimit: null, + }); + await client.connect(); + return client; + } + + function waitForAttentionRequired( + client: DaemonClient, + agentId: string, + timeout = 60000 + ): Promise> { + return new Promise((resolve, reject) => { + const timer = setTimeout(() => { + cleanup(); + reject(new Error(`Timeout waiting for attention_required (${timeout}ms)`)); + }, timeout); + + const cleanup = client.on("agent_stream", (msg) => { + if (msg.type !== "agent_stream") return; + if (msg.payload.agentId !== agentId) return; + if (msg.payload.event.type !== "attention_required") return; + + clearTimeout(timer); + cleanup(); + resolve(msg.payload.event); + }); + }); + } + + test("shouldNotify is false when client is active and focused on agent", async () => { + client1 = await createClient(); + + const agent = await client1.createAgent({ + provider: "codex", + cwd: "/tmp", + title: "Activity Test", + }); + + client1.sendHeartbeat({ + focusedAgentId: agent.id, + lastActivityAt: new Date().toISOString(), + }); + + 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; + + expect(attention.reason).toBe("finished"); + expect(attention.shouldNotify).toBe(false); + }, 120000); + + test("shouldNotify is true when client has stale activity", async () => { + client1 = await createClient(); + + const agent = await client1.createAgent({ + provider: "codex", + cwd: "/tmp", + title: "Stale Activity Test", + }); + + const staleTime = new Date(Date.now() - 120_000).toISOString(); + client1.sendHeartbeat({ + focusedAgentId: agent.id, + lastActivityAt: staleTime, + }); + + 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; + + expect(attention.reason).toBe("finished"); + expect(attention.shouldNotify).toBe(true); + }, 120000); + + test("shouldNotify is true when client is focused on different agent", async () => { + client1 = await createClient(); + + const agent1 = await client1.createAgent({ + provider: "codex", + cwd: "/tmp", + title: "Agent 1", + }); + + const agent2 = await client1.createAgent({ + provider: "codex", + cwd: "/tmp", + title: "Agent 2", + }); + + client1.sendHeartbeat({ + focusedAgentId: agent2.id, + lastActivityAt: new Date().toISOString(), + }); + + await new Promise((r) => setTimeout(r, 100)); + + const attentionPromise = waitForAttentionRequired(client1, agent1.id); + await client1.sendMessage(agent1.id, "Say 'hello' and nothing else"); + + const attention = await attentionPromise; + + expect(attention.reason).toBe("finished"); + expect(attention.shouldNotify).toBe(true); + }, 120000); + + test("shouldNotify is false when another client is active on agent", async () => { + client1 = await createClient(); + client2 = await createClient(); + + const agent = await client1.createAgent({ + provider: "codex", + cwd: "/tmp", + title: "Multi-client Test", + }); + + client1.sendHeartbeat({ + focusedAgentId: null, + lastActivityAt: new Date(Date.now() - 120_000).toISOString(), + }); + + client2.sendHeartbeat({ + focusedAgentId: agent.id, + lastActivityAt: new Date().toISOString(), + }); + + 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; + + expect(attention.reason).toBe("finished"); + expect(attention.shouldNotify).toBe(false); + }, 120000); + + test("shouldNotify is true when no heartbeat received", async () => { + client1 = await createClient(); + + const agent = await client1.createAgent({ + provider: "codex", + cwd: "/tmp", + title: "No Heartbeat Test", + }); + + const attentionPromise = waitForAttentionRequired(client1, agent.id); + await client1.sendMessage(agent.id, "Say 'hello' and nothing else"); + + const attention = await attentionPromise; + + expect(attention.reason).toBe("finished"); + expect(attention.shouldNotify).toBe(true); + }, 120000); +}); diff --git a/packages/server/src/server/session.ts b/packages/server/src/server/session.ts index 939cb0a62..84c81a902 100644 --- a/packages/server/src/server/session.ts +++ b/packages/server/src/server/session.ts @@ -232,6 +232,10 @@ export class Session { private readonly downloadTokenStore: DownloadTokenStore; private agentTitleCache: Map = new Map(); private unsubscribeAgentEvents: (() => void) | null = null; + private clientActivity: { + focusedAgentId: string | null; + lastActivityAt: Date; + } | null = null; constructor( clientId: string, @@ -282,6 +286,16 @@ export class Session { return this.conversationId; } + /** + * Get the client's current activity state + */ + public getClientActivity(): { + focusedAgentId: string | null; + lastActivityAt: Date; + } | null { + return this.clientActivity; + } + /** * Send initial state to client after connection */ @@ -795,6 +809,10 @@ export class Session { await this.handleClearAgentAttention(msg.agentId); break; + case "client_heartbeat": + this.handleClientHeartbeat(msg); + break; + case "list_commands_request": await this.handleListCommandsRequest(msg.agentId, msg.requestId); break; @@ -1843,6 +1861,19 @@ export class Session { } } + /** + * Handle client heartbeat for activity tracking + */ + private handleClientHeartbeat(msg: { + focusedAgentId: string | null; + lastActivityAt: string; + }): void { + this.clientActivity = { + focusedAgentId: msg.focusedAgentId, + lastActivityAt: new Date(msg.lastActivityAt), + }; + } + /** * Handle list commands request for an agent */ diff --git a/packages/server/src/server/websocket-server.ts b/packages/server/src/server/websocket-server.ts index 983154088..7d790a45c 100644 --- a/packages/server/src/server/websocket-server.ts +++ b/packages/server/src/server/websocket-server.ts @@ -12,6 +12,7 @@ import { Session } from "./session.js"; import { loadConversation } from "./persistence.js"; 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"; type AgentMcpClientConfig = { @@ -50,6 +51,10 @@ export class VoiceAssistantWebSocketServer { this.handleConnection(ws, request); }); + this.agentManager.setAgentAttentionCallback((params) => { + this.broadcastAgentAttention(params); + }); + console.log("✓ WebSocket server initialized on /ws"); } @@ -309,4 +314,54 @@ export class VoiceAssistantWebSocketServer { await Promise.all(cleanupPromises); 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; + + for (const [, session] of this.sessions) { + const activity = session.getClientActivity(); + if ( + activity !== null && + activity.focusedAgentId === agentId && + now - activity.lastActivityAt.getTime() < activityThresholdMs + ) { + return true; + } + } + return false; + } + + /** + * Broadcast an attention_required event to all clients with shouldNotify computed + */ + private broadcastAgentAttention(params: { + agentId: string; + provider: AgentProvider; + reason: "finished" | "error" | "permission"; + }): void { + const shouldNotify = !this.isAnyClientActiveOnAgent(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(), + }, + }); + + for (const [ws] of this.sessions) { + this.sendToClient(ws, message); + } + } } diff --git a/packages/server/src/shared/messages.ts b/packages/server/src/shared/messages.ts index ddaaf6abf..162a24571 100644 --- a/packages/server/src/shared/messages.ts +++ b/packages/server/src/shared/messages.ts @@ -202,6 +202,7 @@ export const AgentStreamEventPayloadSchema = z.discriminatedUnion("type", [ provider: AgentProviderSchema, reason: z.enum(["finished", "error", "permission"]), timestamp: z.string(), + shouldNotify: z.boolean(), }), ]); @@ -511,6 +512,12 @@ export const ClearAgentAttentionMessageSchema = z.object({ agentId: z.union([z.string(), z.array(z.string())]), }); +export const ClientHeartbeatMessageSchema = z.object({ + type: z.literal("client_heartbeat"), + focusedAgentId: z.string().nullable(), + lastActivityAt: z.string(), +}); + export const ListCommandsRequestSchema = z.object({ type: z.literal("list_commands_request"), agentId: z.string(), @@ -544,6 +551,7 @@ export const SessionInboundMessageSchema = z.discriminatedUnion("type", [ FileDownloadTokenRequestSchema, GitRepoInfoRequestMessageSchema, ClearAgentAttentionMessageSchema, + ClientHeartbeatMessageSchema, ListCommandsRequestSchema, ]); @@ -956,6 +964,7 @@ export type FileDownloadTokenResponse = z.infer; export type RestartServerRequestMessage = z.infer; export type ClearAgentAttentionMessage = z.infer; +export type ClientHeartbeatMessage = z.infer; export type ListCommandsRequest = z.infer; export type ListCommandsResponse = z.infer;