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

@@ -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 () => {

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<T> {
   resolve(value: T): void;
   reject(error: Error): void;
   timeoutHandle: ReturnType<typeof setTimeout> | null;
+  requestId?: string;
 }
 
 interface WaitHandle<T> {
@@ -796,6 +834,16 @@ interface WaitHandle<T> {
   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<T>(
     predicate: (msg: SessionOutboundMessage) => T | null,
     timeout = 30000,
-    _options?: { skipQueue?: boolean },
+    options?: WaitOptions,
   ): WaitHandle<T> {
     // 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);
     });