From d42ab91971a92bf2e2a981848989c7ca6536a18e Mon Sep 17 00:00:00 2001 From: Mohamed Boudra Date: Thu, 16 Jul 2026 16:44:01 +0200 Subject: [PATCH] Show agent history errors without a one-minute wait (#2124) * fix(client): fail invalid RPC responses immediately Correlated responses that failed schema validation were discarded before request matching, leaving callers blocked until the RPC timeout. Preserve only the raw correlation identity at the boundary so the matching request fails with a protocol error. * test(client): preserve invalid response diagnostics * fix(client): ignore invalid correlated progress events --- packages/client/src/daemon-client.test.ts | 66 +++++++++++++++-- packages/client/src/daemon-client.ts | 90 ++++++++++++++++++++++- 2 files changed, 145 insertions(+), 11 deletions(-) diff --git a/packages/client/src/daemon-client.test.ts b/packages/client/src/daemon-client.test.ts index 252ed1457..3b3183862 100644 --- a/packages/client/src/daemon-client.test.ts +++ b/packages/client/src/daemon-client.test.ts @@ -4947,9 +4947,13 @@ test("parses canonical fetch_agent_timeline_response payloads without crashing", expect(logger.warn).not.toHaveBeenCalled(); }); -test("drops invalid fetch_agent_timeline_response tool_call payloads and logs validation warning", async () => { - const logger = createMockLogger(); +test("rejects and logs a correlated response that violates the protocol schema", async () => { const mock = createMockTransport(); + const warnings: string[] = []; + const logger: Logger = { + ...noopLogger, + warn: (_fields, message) => warnings.push(message ?? ""), + }; const client = new DaemonClient({ url: "ws://test", @@ -4964,9 +4968,9 @@ test("drops invalid fetch_agent_timeline_response tool_call payloads and logs va mock.triggerOpen(); await connectPromise; - const received: unknown[] = []; - const unsubscribe = client.on("fetch_agent_timeline_response", (msg) => { - received.push(msg); + const response = client.fetchAgentTimeline("agent_cli", { + requestId: "req-invalid", + timeout: 1, }); mock.triggerMessage( @@ -5013,10 +5017,56 @@ test("drops invalid fetch_agent_timeline_response tool_call payloads and logs va }), ); - unsubscribe(); + await expect(response).rejects.toMatchObject({ + requestId: "req-invalid", + message: expect.stringMatching(/validation/i), + }); + expect(warnings).toEqual(["Message validation failed"]); +}); - expect(received).toHaveLength(0); - expect(logger.warn).toHaveBeenCalled(); +test("does not reject a request for an invalid correlated progress event", async () => { + const mock = createMockTransport(); + const client = new DaemonClient({ + url: "ws://test", + clientId: "clsk_unit_test", + logger: noopLogger, + reconnect: { enabled: false }, + transportFactory: () => mock.transport, + }); + clients.push(client); + + const connectPromise = client.connect(); + mock.triggerOpen(); + await connectPromise; + + const response = client.updateDaemon("req-update"); + + mock.triggerMessage( + wrapSessionMessage({ + type: "daemon.update.progress", + payload: { + requestId: "req-update", + phase: "verifying", + }, + }), + ); + mock.triggerMessage( + wrapSessionMessage({ + type: "daemon.update.response", + payload: { + requestId: "req-update", + success: true, + error: null, + previousVersion: "0.1.106", + newVersion: "0.1.107", + }, + }), + ); + + await expect(response).resolves.toMatchObject({ + requestId: "req-update", + success: true, + }); }); test("sends subscribe/unsubscribe terminals messages", async () => { diff --git a/packages/client/src/daemon-client.ts b/packages/client/src/daemon-client.ts index daf510e8b..cdd23680b 100644 --- a/packages/client/src/daemon-client.ts +++ b/packages/client/src/daemon-client.ts @@ -176,6 +176,43 @@ function normalizePassword(value: string | undefined): string | null { return value.length > 0 ? value : null; } +function extractCorrelatedResponseIdentity(input: unknown): CorrelatedResponseIdentity | null { + if (!input || typeof input !== "object") { + return null; + } + + const envelope = input as { type?: unknown; message?: unknown }; + if (envelope.type !== "session" || !envelope.message || typeof envelope.message !== "object") { + return null; + } + + const message = envelope.message as { type?: unknown; payload?: unknown }; + if ( + typeof message.type !== "string" || + !( + message.type === "rpc_error" || + message.type.endsWith("_response") || + message.type.endsWith(".response") || + message.type.endsWith("/response") + ) + ) { + return null; + } + if (!message.payload || typeof message.payload !== "object") { + return null; + } + + const payload = message.payload as { requestId?: unknown }; + if (typeof payload.requestId !== "string") { + return null; + } + + return { + requestId: payload.requestId, + responseType: message.type, + }; +} + export type { DaemonTransport, DaemonTransportFactory, @@ -789,6 +826,7 @@ interface Waiter { resolve(value: T): void; reject(error: Error): void; timeoutHandle: ReturnType | null; + requestId?: string; } interface WaitHandle { @@ -796,6 +834,16 @@ interface WaitHandle { cancel: (error: Error) => void; } +interface WaitOptions { + skipQueue?: boolean; + requestId?: string; +} + +interface CorrelatedResponseIdentity { + requestId: string; + responseType?: string; +} + interface PendingBinaryFileRead { cwd: string; path: string; @@ -848,6 +896,20 @@ class DaemonRpcError extends Error { } } +class DaemonProtocolError extends Error { + readonly requestId: string; + readonly responseType?: string; + readonly code = "invalid_response"; + + constructor(identity: CorrelatedResponseIdentity) { + const responseLabel = identity.responseType ?? "unknown response"; + super(`Response validation failed for ${responseLabel}`); + this.name = "DaemonProtocolError"; + this.requestId = identity.requestId; + this.responseType = identity.responseType; + } +} + class PingTimeoutError extends Error { constructor(readonly timeoutMs: number) { super(`Ping timed out (${timeoutMs}ms)`); @@ -1548,7 +1610,7 @@ export class DaemonClient { return { kind: "ok", value }; }, timeout, - params.options, + { ...params.options, requestId: params.requestId }, ); try { @@ -4934,14 +4996,22 @@ export class DaemonClient { const parsed = validateWSOutboundMessage(parsedJson); if (!parsed.success) { - const msgType = + const responseIdentity = extractCorrelatedResponseIdentity(parsedJson); + const envelopeType = parsedJson != null && typeof parsedJson === "object" && "type" in parsedJson && typeof parsedJson.type === "string" ? parsedJson.type : "unknown"; + const msgType = responseIdentity?.responseType ?? envelopeType; this.logger.warn({ msgType, error: parsed.error.message }, "Message validation failed"); + if (responseIdentity) { + this.rejectWaitersForRequestId( + responseIdentity.requestId, + new DaemonProtocolError(responseIdentity), + ); + } return; } @@ -5262,6 +5332,19 @@ export class DaemonClient { } } + private rejectWaitersForRequestId(requestId: string, error: Error): void { + for (const waiter of Array.from(this.waiters)) { + if (waiter.requestId !== requestId) { + continue; + } + this.waiters.delete(waiter); + if (waiter.timeoutHandle) { + clearTimeout(waiter.timeoutHandle); + } + waiter.reject(error); + } + } + private clearWaiters(error: Error): void { for (const waiter of Array.from(this.waiters)) { if (waiter.timeoutHandle) { @@ -5331,7 +5414,7 @@ export class DaemonClient { private waitForWithCancel( predicate: (msg: SessionOutboundMessage) => T | null, timeout = 30000, - _options?: { skipQueue?: boolean }, + options?: WaitOptions, ): WaitHandle { // Capture stack trace at call site, not inside setTimeout const timeoutError = new Error(`Timeout waiting for message (${timeout}ms)`); @@ -5368,6 +5451,7 @@ export class DaemonClient { resolve: wrappedResolve, reject: wrappedReject, timeoutHandle, + requestId: options?.requestId, }; this.waiters.add(waiter); });