Implement DaemonClient core for E2E testing (Phase 1)
Add WebSocket client infrastructure for testing the daemon without Playwright: - DaemonClient class with full lifecycle management: - connect/close for WebSocket connection - createAgent, deleteAgent, listAgents for agent lifecycle - sendMessage, cancelAgent, setAgentMode for agent interaction - waitForAgentIdle, waitForPermission for async waiting - respondToPermission for permission handling - Event subscription via on() method - Test context helper (createDaemonTestContext) that creates isolated daemon + connected client for each test - One working E2E test that creates a Codex agent, sends a message, and verifies the full turn lifecycle (turn_started, assistant_message, turn_completed events) Key implementation details: - Uses skipQueueBefore option in waitFor to ignore stale messages - waitForAgentIdle tracks "running" state to avoid false positives - All methods properly typed using existing Zod schemas 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
Mohamed Boudra
co-authored by
Claude Opus 4.5
Claude Opus 4.5
parent
cf82296be9
commit
99a5ac97fb
@@ -0,0 +1,74 @@
|
||||
import { describe, test, expect, beforeEach, afterEach } from "vitest";
|
||||
import {
|
||||
createDaemonTestContext,
|
||||
type DaemonTestContext,
|
||||
} from "./test-utils/index.js";
|
||||
|
||||
describe("daemon E2E", () => {
|
||||
let ctx: DaemonTestContext;
|
||||
|
||||
beforeEach(async () => {
|
||||
ctx = await createDaemonTestContext();
|
||||
});
|
||||
|
||||
afterEach(async () => {
|
||||
await ctx.cleanup();
|
||||
});
|
||||
|
||||
test("creates agent and receives response", async () => {
|
||||
// Create a Codex agent
|
||||
const agent = await ctx.client.createAgent({
|
||||
provider: "codex",
|
||||
cwd: "/tmp",
|
||||
title: "Test Agent",
|
||||
});
|
||||
|
||||
expect(agent.id).toBeTruthy();
|
||||
expect(agent.provider).toBe("codex");
|
||||
expect(agent.status).toBe("idle");
|
||||
// Title may or may not be set depending on timing
|
||||
expect(agent.cwd).toBe("/tmp");
|
||||
|
||||
// Send a simple message
|
||||
await ctx.client.sendMessage(agent.id, "Say 'hello world' and nothing else");
|
||||
|
||||
// Wait for the agent to complete
|
||||
const finalState = await ctx.client.waitForAgentIdle(agent.id, 120000);
|
||||
|
||||
// Verify agent completed without error
|
||||
expect(finalState.status).toBe("idle");
|
||||
expect(finalState.lastError).toBeUndefined();
|
||||
expect(finalState.id).toBe(agent.id);
|
||||
|
||||
// Verify we received some stream events
|
||||
const queue = ctx.client.getMessageQueue();
|
||||
const streamEvents = queue.filter(
|
||||
(m) => m.type === "agent_stream" && m.payload.agentId === agent.id
|
||||
);
|
||||
expect(streamEvents.length).toBeGreaterThan(0);
|
||||
|
||||
// Verify there was a turn_started event
|
||||
const hasTurnStarted = streamEvents.some(
|
||||
(m) =>
|
||||
m.type === "agent_stream" && m.payload.event.type === "turn_started"
|
||||
);
|
||||
expect(hasTurnStarted).toBe(true);
|
||||
|
||||
// Verify there was a turn_completed event
|
||||
const hasTurnCompleted = streamEvents.some(
|
||||
(m) =>
|
||||
m.type === "agent_stream" && m.payload.event.type === "turn_completed"
|
||||
);
|
||||
expect(hasTurnCompleted).toBe(true);
|
||||
|
||||
// Verify there was an assistant message in the timeline
|
||||
const hasAssistantMessage = streamEvents.some((m) => {
|
||||
if (m.type !== "agent_stream" || m.payload.event.type !== "timeline") {
|
||||
return false;
|
||||
}
|
||||
const item = m.payload.event.item;
|
||||
return item.type === "assistant_message" && item.text.length > 0;
|
||||
});
|
||||
expect(hasAssistantMessage).toBe(true);
|
||||
}, 180000); // 3 minute timeout for E2E test
|
||||
});
|
||||
@@ -0,0 +1,478 @@
|
||||
import WebSocket from "ws";
|
||||
import { nanoid } from "nanoid";
|
||||
import type {
|
||||
SessionInboundMessage,
|
||||
SessionOutboundMessage,
|
||||
AgentSnapshotPayload,
|
||||
AgentStreamEventPayload,
|
||||
PersistedAgentDescriptorPayload,
|
||||
} from "../messages.js";
|
||||
import type {
|
||||
AgentPermissionRequest,
|
||||
AgentPermissionResponse,
|
||||
AgentPersistenceHandle,
|
||||
AgentProvider,
|
||||
} from "../agent/agent-sdk-types.js";
|
||||
|
||||
// ============================================================================
|
||||
// Configuration
|
||||
// ============================================================================
|
||||
|
||||
export interface DaemonClientConfig {
|
||||
url: string;
|
||||
authHeader?: string;
|
||||
}
|
||||
|
||||
export interface CreateAgentOptions {
|
||||
provider: AgentProvider;
|
||||
cwd: string;
|
||||
title?: string;
|
||||
model?: string;
|
||||
modeId?: string;
|
||||
initialPrompt?: string;
|
||||
mcpServers?: Record<string, unknown>;
|
||||
}
|
||||
|
||||
export interface SendMessageOptions {
|
||||
messageId?: string;
|
||||
images?: Array<{ data: string; mimeType: string }>;
|
||||
}
|
||||
|
||||
// ============================================================================
|
||||
// Event Types
|
||||
// ============================================================================
|
||||
|
||||
export type DaemonEvent =
|
||||
| { type: "agent_state"; agentId: string; payload: AgentSnapshotPayload }
|
||||
| {
|
||||
type: "agent_stream";
|
||||
agentId: string;
|
||||
event: AgentStreamEventPayload;
|
||||
timestamp: string;
|
||||
}
|
||||
| { type: "session_state"; agents: AgentSnapshotPayload[] }
|
||||
| { type: "status"; payload: { status: string } }
|
||||
| { type: "agent_deleted"; agentId: string }
|
||||
| { type: "agent_permission_request"; agentId: string; request: AgentPermissionRequest }
|
||||
| {
|
||||
type: "agent_permission_resolved";
|
||||
agentId: string;
|
||||
requestId: string;
|
||||
resolution: AgentPermissionResponse;
|
||||
}
|
||||
| { type: "error"; message: string };
|
||||
|
||||
export type DaemonEventHandler = (event: DaemonEvent) => void;
|
||||
|
||||
// ============================================================================
|
||||
// DaemonClient
|
||||
// ============================================================================
|
||||
|
||||
export class DaemonClient {
|
||||
private ws: WebSocket | null = null;
|
||||
private messageQueue: SessionOutboundMessage[] = [];
|
||||
private eventListeners: Set<DaemonEventHandler> = new Set();
|
||||
private messageListeners: Set<() => void> = new Set();
|
||||
|
||||
constructor(private config: DaemonClientConfig) {}
|
||||
|
||||
// ============================================================================
|
||||
// Connection
|
||||
// ============================================================================
|
||||
|
||||
async connect(): Promise<void> {
|
||||
return new Promise((resolve, reject) => {
|
||||
const headers: Record<string, string> = {};
|
||||
if (this.config.authHeader) {
|
||||
headers["Authorization"] = this.config.authHeader;
|
||||
}
|
||||
|
||||
this.ws = new WebSocket(this.config.url, { headers });
|
||||
|
||||
const onOpen = (): void => {
|
||||
cleanup();
|
||||
resolve();
|
||||
};
|
||||
|
||||
const onError = (err: Error): void => {
|
||||
cleanup();
|
||||
reject(err);
|
||||
};
|
||||
|
||||
const onMessage = (data: WebSocket.RawData): void => {
|
||||
try {
|
||||
const parsed = JSON.parse(data.toString()) as {
|
||||
type: string;
|
||||
message?: SessionOutboundMessage;
|
||||
};
|
||||
if (parsed.type === "pong") return;
|
||||
if (parsed.type === "session" && parsed.message) {
|
||||
this.handleSessionMessage(parsed.message);
|
||||
}
|
||||
} catch {
|
||||
// Ignore parse errors
|
||||
}
|
||||
};
|
||||
|
||||
const cleanup = (): void => {
|
||||
this.ws?.off("open", onOpen);
|
||||
this.ws?.off("error", onError);
|
||||
};
|
||||
|
||||
this.ws.on("open", onOpen);
|
||||
this.ws.on("error", onError);
|
||||
this.ws.on("message", onMessage);
|
||||
});
|
||||
}
|
||||
|
||||
async close(): Promise<void> {
|
||||
if (this.ws) {
|
||||
this.ws.close();
|
||||
this.ws = null;
|
||||
}
|
||||
this.messageQueue = [];
|
||||
this.eventListeners.clear();
|
||||
this.messageListeners.clear();
|
||||
}
|
||||
|
||||
// ============================================================================
|
||||
// Agent Lifecycle
|
||||
// ============================================================================
|
||||
|
||||
async createAgent(options: CreateAgentOptions): Promise<AgentSnapshotPayload> {
|
||||
const requestId = nanoid();
|
||||
this.send({
|
||||
type: "create_agent_request",
|
||||
requestId,
|
||||
config: {
|
||||
provider: options.provider,
|
||||
cwd: options.cwd,
|
||||
title: options.title,
|
||||
model: options.model,
|
||||
modeId: options.modeId,
|
||||
mcpServers: options.mcpServers,
|
||||
},
|
||||
initialPrompt: options.initialPrompt,
|
||||
});
|
||||
|
||||
// First get the agent ID from the initial state
|
||||
let agentId: string | null = null;
|
||||
await this.waitFor((msg) => {
|
||||
if (msg.type === "agent_state") {
|
||||
agentId = msg.payload.id;
|
||||
return msg.payload;
|
||||
}
|
||||
return null;
|
||||
}, 10000);
|
||||
|
||||
if (!agentId) {
|
||||
throw new Error("Failed to get agent ID from create response");
|
||||
}
|
||||
|
||||
// Wait for the agent to be idle
|
||||
return this.waitFor((msg) => {
|
||||
if (
|
||||
msg.type === "agent_state" &&
|
||||
msg.payload.id === agentId &&
|
||||
msg.payload.status === "idle"
|
||||
) {
|
||||
return msg.payload;
|
||||
}
|
||||
return null;
|
||||
}, 60000); // 60 second timeout for initialization
|
||||
}
|
||||
|
||||
async deleteAgent(agentId: string): Promise<void> {
|
||||
this.send({ type: "delete_agent_request", agentId });
|
||||
await this.waitFor((msg) => {
|
||||
if (msg.type === "agent_deleted" && msg.payload.agentId === agentId) {
|
||||
return true;
|
||||
}
|
||||
return null;
|
||||
});
|
||||
}
|
||||
|
||||
async listAgents(): Promise<AgentSnapshotPayload[]> {
|
||||
// session_state is sent on connection, or we can wait for it
|
||||
return this.waitFor((msg) => {
|
||||
if (msg.type === "session_state") {
|
||||
return msg.payload.agents;
|
||||
}
|
||||
return null;
|
||||
});
|
||||
}
|
||||
|
||||
async listPersistedAgents(): Promise<PersistedAgentDescriptorPayload[]> {
|
||||
this.send({ type: "list_persisted_agents_request" });
|
||||
return this.waitFor((msg) => {
|
||||
if (msg.type === "list_persisted_agents_response") {
|
||||
return msg.payload.items;
|
||||
}
|
||||
return null;
|
||||
});
|
||||
}
|
||||
|
||||
async resumeAgent(
|
||||
handle: AgentPersistenceHandle,
|
||||
overrides?: Partial<CreateAgentOptions>
|
||||
): Promise<AgentSnapshotPayload> {
|
||||
const requestId = nanoid();
|
||||
this.send({
|
||||
type: "resume_agent_request",
|
||||
requestId,
|
||||
handle,
|
||||
overrides: overrides as Record<string, unknown>,
|
||||
});
|
||||
|
||||
return this.waitFor((msg) => {
|
||||
if (msg.type === "agent_state") {
|
||||
return msg.payload;
|
||||
}
|
||||
return null;
|
||||
});
|
||||
}
|
||||
|
||||
// ============================================================================
|
||||
// Agent Interaction
|
||||
// ============================================================================
|
||||
|
||||
async sendMessage(
|
||||
agentId: string,
|
||||
text: string,
|
||||
options?: SendMessageOptions
|
||||
): Promise<void> {
|
||||
this.send({
|
||||
type: "send_agent_message",
|
||||
agentId,
|
||||
text,
|
||||
messageId: options?.messageId,
|
||||
images: options?.images,
|
||||
});
|
||||
}
|
||||
|
||||
async cancelAgent(agentId: string): Promise<void> {
|
||||
this.send({ type: "cancel_agent_request", agentId });
|
||||
}
|
||||
|
||||
async setAgentMode(agentId: string, modeId: string): Promise<void> {
|
||||
this.send({ type: "set_agent_mode", agentId, modeId });
|
||||
}
|
||||
|
||||
// ============================================================================
|
||||
// Permissions
|
||||
// ============================================================================
|
||||
|
||||
async respondToPermission(
|
||||
agentId: string,
|
||||
requestId: string,
|
||||
response: AgentPermissionResponse
|
||||
): Promise<void> {
|
||||
this.send({
|
||||
type: "agent_permission_response",
|
||||
agentId,
|
||||
requestId,
|
||||
response,
|
||||
});
|
||||
}
|
||||
|
||||
// ============================================================================
|
||||
// Waiting / Streaming
|
||||
// ============================================================================
|
||||
|
||||
async waitForAgentIdle(
|
||||
agentId: string,
|
||||
timeout = 60000
|
||||
): Promise<AgentSnapshotPayload> {
|
||||
// Record the current queue position so we only check messages from NOW
|
||||
const startPosition = this.messageQueue.length;
|
||||
|
||||
// First, wait for the agent to go to "running" state (or already be running)
|
||||
// This ensures we don't return on an old "idle" state from before the message
|
||||
let sawRunning = false;
|
||||
|
||||
return this.waitFor(
|
||||
(msg) => {
|
||||
if (msg.type === "agent_state" && msg.payload.id === agentId) {
|
||||
const status = msg.payload.status;
|
||||
if (status === "running") {
|
||||
sawRunning = true;
|
||||
}
|
||||
// Only return idle/error AFTER we've seen running
|
||||
if (sawRunning && (status === "idle" || status === "error")) {
|
||||
return msg.payload;
|
||||
}
|
||||
}
|
||||
return null;
|
||||
},
|
||||
timeout,
|
||||
{ skipQueueBefore: startPosition }
|
||||
);
|
||||
}
|
||||
|
||||
async waitForPermission(
|
||||
agentId: string,
|
||||
timeout = 30000
|
||||
): Promise<AgentPermissionRequest> {
|
||||
return this.waitFor((msg) => {
|
||||
// Check direct permission request message
|
||||
if (
|
||||
msg.type === "agent_permission_request" &&
|
||||
msg.payload.agentId === agentId
|
||||
) {
|
||||
return msg.payload.request;
|
||||
}
|
||||
// Check stream event
|
||||
if (msg.type === "agent_stream" && msg.payload.agentId === agentId) {
|
||||
if (msg.payload.event.type === "permission_requested") {
|
||||
return msg.payload.event.request;
|
||||
}
|
||||
}
|
||||
return null;
|
||||
}, timeout);
|
||||
}
|
||||
|
||||
// ============================================================================
|
||||
// Event Subscription
|
||||
// ============================================================================
|
||||
|
||||
on(handler: DaemonEventHandler): () => void {
|
||||
this.eventListeners.add(handler);
|
||||
return () => this.eventListeners.delete(handler);
|
||||
}
|
||||
|
||||
// ============================================================================
|
||||
// Internals
|
||||
// ============================================================================
|
||||
|
||||
private send(message: SessionInboundMessage): void {
|
||||
if (!this.ws || this.ws.readyState !== WebSocket.OPEN) {
|
||||
throw new Error("WebSocket not connected");
|
||||
}
|
||||
this.ws.send(JSON.stringify({ type: "session", message }));
|
||||
}
|
||||
|
||||
private handleSessionMessage(msg: SessionOutboundMessage): void {
|
||||
this.messageQueue.push(msg);
|
||||
|
||||
// Notify message listeners (for waitFor) - just signal, they'll check the queue
|
||||
for (const listener of this.messageListeners) {
|
||||
listener();
|
||||
}
|
||||
|
||||
// Notify event listeners
|
||||
const event = this.toEvent(msg);
|
||||
if (event) {
|
||||
for (const handler of this.eventListeners) {
|
||||
handler(event);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private toEvent(msg: SessionOutboundMessage): DaemonEvent | null {
|
||||
switch (msg.type) {
|
||||
case "agent_state":
|
||||
return {
|
||||
type: "agent_state",
|
||||
agentId: msg.payload.id,
|
||||
payload: msg.payload,
|
||||
};
|
||||
case "agent_stream":
|
||||
return {
|
||||
type: "agent_stream",
|
||||
agentId: msg.payload.agentId,
|
||||
event: msg.payload.event,
|
||||
timestamp: msg.payload.timestamp,
|
||||
};
|
||||
case "session_state":
|
||||
return { type: "session_state", agents: msg.payload.agents };
|
||||
case "status":
|
||||
return { type: "status", payload: msg.payload };
|
||||
case "agent_deleted":
|
||||
return { type: "agent_deleted", agentId: msg.payload.agentId };
|
||||
case "agent_permission_request":
|
||||
return {
|
||||
type: "agent_permission_request",
|
||||
agentId: msg.payload.agentId,
|
||||
request: msg.payload.request,
|
||||
};
|
||||
case "agent_permission_resolved":
|
||||
return {
|
||||
type: "agent_permission_resolved",
|
||||
agentId: msg.payload.agentId,
|
||||
requestId: msg.payload.requestId,
|
||||
resolution: msg.payload.resolution,
|
||||
};
|
||||
default:
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
||||
private async waitFor<T>(
|
||||
predicate: (msg: SessionOutboundMessage) => T | null,
|
||||
timeout = 30000,
|
||||
options?: { skipQueue?: boolean; skipQueueBefore?: number }
|
||||
): Promise<T> {
|
||||
// Record the starting queue length so we can track new messages
|
||||
const startQueueLength = options?.skipQueueBefore ?? this.messageQueue.length;
|
||||
|
||||
// Check queued messages first (unless skipped or with position offset)
|
||||
if (!options?.skipQueue && options?.skipQueueBefore === undefined) {
|
||||
for (const msg of this.messageQueue) {
|
||||
const result = predicate(msg);
|
||||
if (result !== null) return result;
|
||||
}
|
||||
}
|
||||
|
||||
// Wait for new messages only
|
||||
return new Promise((resolve, reject) => {
|
||||
// Track which messages we've already checked
|
||||
let checkedCount = startQueueLength;
|
||||
|
||||
const checkNewMessages = (): boolean => {
|
||||
// Check any messages added since we last checked
|
||||
while (checkedCount < this.messageQueue.length) {
|
||||
const msg = this.messageQueue[checkedCount];
|
||||
checkedCount++;
|
||||
const result = predicate(msg);
|
||||
if (result !== null) {
|
||||
cleanup();
|
||||
resolve(result);
|
||||
return true;
|
||||
}
|
||||
}
|
||||
return false;
|
||||
};
|
||||
|
||||
const listener = (): void => {
|
||||
checkNewMessages();
|
||||
};
|
||||
|
||||
const timer = setTimeout(() => {
|
||||
cleanup();
|
||||
reject(new Error(`Timeout waiting for message (${timeout}ms)`));
|
||||
}, timeout);
|
||||
|
||||
const cleanup = (): void => {
|
||||
clearTimeout(timer);
|
||||
this.messageListeners.delete(listener);
|
||||
};
|
||||
|
||||
this.messageListeners.add(listener);
|
||||
|
||||
// Check any messages that arrived between startQueueLength and now
|
||||
checkNewMessages();
|
||||
});
|
||||
}
|
||||
|
||||
// ============================================================================
|
||||
// Debug / Utilities
|
||||
// ============================================================================
|
||||
|
||||
getMessageQueue(): readonly SessionOutboundMessage[] {
|
||||
return this.messageQueue;
|
||||
}
|
||||
|
||||
clearMessageQueue(): void {
|
||||
this.messageQueue = [];
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,50 @@
|
||||
import { createTestPaseoDaemon, type TestPaseoDaemon } from "./paseo-daemon.js";
|
||||
import { DaemonClient } from "./daemon-client.js";
|
||||
|
||||
export interface DaemonTestContext {
|
||||
daemon: TestPaseoDaemon;
|
||||
client: DaemonClient;
|
||||
cleanup: () => Promise<void>;
|
||||
}
|
||||
|
||||
/**
|
||||
* Create a test context with an isolated daemon and connected client.
|
||||
*
|
||||
* Usage:
|
||||
* ```typescript
|
||||
* let ctx: DaemonTestContext;
|
||||
*
|
||||
* beforeEach(async () => {
|
||||
* ctx = await createDaemonTestContext();
|
||||
* });
|
||||
*
|
||||
* afterEach(async () => {
|
||||
* await ctx.cleanup();
|
||||
* });
|
||||
*
|
||||
* test("creates agent", async () => {
|
||||
* const agent = await ctx.client.createAgent({
|
||||
* provider: "codex",
|
||||
* cwd: "/tmp",
|
||||
* });
|
||||
* expect(agent.id).toBeTruthy();
|
||||
* });
|
||||
* ```
|
||||
*/
|
||||
export async function createDaemonTestContext(): Promise<DaemonTestContext> {
|
||||
const daemon = await createTestPaseoDaemon();
|
||||
const client = new DaemonClient({
|
||||
url: `ws://127.0.0.1:${daemon.port}/ws`,
|
||||
authHeader: daemon.agentMcpAuthHeader,
|
||||
});
|
||||
await client.connect();
|
||||
|
||||
return {
|
||||
daemon,
|
||||
client,
|
||||
cleanup: async () => {
|
||||
await client.close();
|
||||
await daemon.close();
|
||||
},
|
||||
};
|
||||
}
|
||||
@@ -0,0 +1,13 @@
|
||||
export { createTestPaseoDaemon, type TestPaseoDaemon } from "./paseo-daemon.js";
|
||||
export {
|
||||
DaemonClient,
|
||||
type DaemonClientConfig,
|
||||
type CreateAgentOptions,
|
||||
type SendMessageOptions,
|
||||
type DaemonEvent,
|
||||
type DaemonEventHandler,
|
||||
} from "./daemon-client.js";
|
||||
export {
|
||||
createDaemonTestContext,
|
||||
type DaemonTestContext,
|
||||
} from "./daemon-test-context.js";
|
||||
Reference in New Issue
Block a user