Add server-side client activity tracking for smart notifications

- Add client_heartbeat message type for clients to report activity
- Add shouldNotify field to attention_required events
- Track client activity (focusedAgentId, lastActivityAt) per session
- WebSocketServer checks all sessions to determine shouldNotify
- If any client is active on the agent (within 60s), shouldNotify=false
- Add onAgentAttention callback to AgentManager for clean separation
This commit is contained in:
Mohamed Boudra
2026-01-13 11:50:13 +07:00
parent 2fa7ec17cb
commit 7f7f5c53e6
6 changed files with 325 additions and 3 deletions

View File

@@ -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
// ============================================================================

View File

@@ -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<Record<AgentProvider, AgentClient>>;
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<string, AgentLifecycleStatus>();
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(),
});
}

View File

@@ -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<DaemonClient> {
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<Extract<AgentStreamEventPayload, { type: "attention_required" }>> {
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);
});

View File

@@ -232,6 +232,10 @@ export class Session {
private readonly downloadTokenStore: DownloadTokenStore;
private agentTitleCache: Map<string, string | null> = 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
*/

View File

@@ -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);
}
}
}

View File

@@ -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<typeof FileDownloadTokenResponse
export type GitRepoInfoResponse = z.infer<typeof GitRepoInfoResponseSchema>;
export type RestartServerRequestMessage = z.infer<typeof RestartServerRequestMessageSchema>;
export type ClearAgentAttentionMessage = z.infer<typeof ClearAgentAttentionMessageSchema>;
export type ClientHeartbeatMessage = z.infer<typeof ClientHeartbeatMessageSchema>;
export type ListCommandsRequest = z.infer<typeof ListCommandsRequestSchema>;
export type ListCommandsResponse = z.infer<typeof ListCommandsResponseSchema>;