From 8e5d2805c1c2840b520b073cb18ca34fa0c1c7de Mon Sep 17 00:00:00 2001 From: Mohamed Boudra Date: Sat, 14 Feb 2026 08:46:38 +0700 Subject: [PATCH] refactor(server): centralize correlated RPC response matching in daemon client What changed: - Added a private sendCorrelatedRequest helper in daemon-client to centralize requestId-correlated response selection. - Migrated a large contiguous set of RPC methods to the helper: checkout/worktree/branch operations, file explorer/token/icon, provider+speech+commands RPCs, wait_for_finish, and terminal list/create/subscribe/kill/attach/detach. - Preserved existing behavior including special-case filtering for checkout subscriptionId. Reasoning: - daemon-client had repeated type+requestId matching blocks across many methods, increasing maintenance cost and decision-point sprawl. - This refactor keeps transport behavior unchanged while reducing duplicated control flow and making future RPC additions less error-prone. Verification: - npm run -w packages/server test -- src/client/daemon-client.test.ts - npm run -w packages/server typecheck - npm run typecheck Notes for next agent: - There are still many non-migrated sendRequest callsites in daemon-client (notably agent lifecycle/interaction and some status-typed selectors) that can be moved to the same helper in follow-up refactors. - I hit TS generic narrowing limits while extracting the helper and solved it via a constrained correlated message type + explicit payload narrowing cast; behavior remains covered by existing daemon-client tests. --- packages/server/src/client/daemon-client.ts | 372 ++++++-------------- 1 file changed, 102 insertions(+), 270 deletions(-) diff --git a/packages/server/src/client/daemon-client.ts b/packages/server/src/client/daemon-client.ts index b18f69240..50a92142e 100644 --- a/packages/server/src/client/daemon-client.ts +++ b/packages/server/src/client/daemon-client.ts @@ -302,6 +302,15 @@ type WaitHandle = { type RpcWaitResult = | { kind: "ok"; value: T } | { kind: "error"; error: DaemonRpcError }; +type CorrelatedResponseMessage = Extract< + SessionOutboundMessage, + { payload: { requestId: string } } +>; +type CorrelatedResponseType = CorrelatedResponseMessage["type"]; +type CorrelatedResponsePayload = Extract< + CorrelatedResponseMessage, + { type: TType } +>["payload"]; class DaemonRpcError extends Error { readonly requestId: string; @@ -853,6 +862,42 @@ export class DaemonClient { return result.value; } + private async sendCorrelatedRequest< + TResponseType extends CorrelatedResponseType, + TResult = CorrelatedResponsePayload, + >( + params: { + requestId: string; + message: SessionInboundMessage; + timeout: number; + responseType: TResponseType; + options?: { skipQueue?: boolean }; + selectPayload?: (payload: CorrelatedResponsePayload) => TResult | null; + } + ): Promise { + return this.sendRequest({ + requestId: params.requestId, + message: params.message, + timeout: params.timeout, + options: params.options, + select: (msg) => { + const correlated = msg as CorrelatedResponseMessage; + if (correlated.type !== params.responseType) { + return null; + } + const payload = + correlated.payload as unknown as CorrelatedResponsePayload; + if (payload.requestId !== params.requestId) { + return null; + } + if (!params.selectPayload) { + return payload as TResult; + } + return params.selectPayload(payload); + }, + }); + } + private sendSessionMessageStrict(message: SessionInboundMessage): void { if (!this.transport || this.connectionState.status !== "connected") { throw new Error("Transport not connected"); @@ -1844,22 +1889,17 @@ export class DaemonClient { }); try { - return await this.sendRequest({ + return await this.sendCorrelatedRequest({ requestId: resolvedRequestId, message, + responseType: "subscribe_checkout_diff_response", timeout: 60000, options: { skipQueue: true }, - select: (msg) => { - if (msg.type !== "subscribe_checkout_diff_response") { + selectPayload: (payload) => { + if (payload.subscriptionId !== subscriptionId) { return null; } - if (msg.payload.requestId !== resolvedRequestId) { - return null; - } - if (msg.payload.subscriptionId !== subscriptionId) { - return null; - } - return msg.payload; + return payload; }, }); } catch (error) { @@ -1893,20 +1933,12 @@ export class DaemonClient { addAll: input.addAll, requestId: resolvedRequestId, }); - return this.sendRequest({ + return this.sendCorrelatedRequest({ requestId: resolvedRequestId, message, + responseType: "checkout_commit_response", timeout: 60000, options: { skipQueue: true }, - select: (msg) => { - if (msg.type !== "checkout_commit_response") { - return null; - } - if (msg.payload.requestId !== resolvedRequestId) { - return null; - } - return msg.payload; - }, }); } @@ -1924,20 +1956,12 @@ export class DaemonClient { requireCleanTarget: input.requireCleanTarget, requestId: resolvedRequestId, }); - return this.sendRequest({ + return this.sendCorrelatedRequest({ requestId: resolvedRequestId, message, + responseType: "checkout_merge_response", timeout: 60000, options: { skipQueue: true }, - select: (msg) => { - if (msg.type !== "checkout_merge_response") { - return null; - } - if (msg.payload.requestId !== resolvedRequestId) { - return null; - } - return msg.payload; - }, }); } @@ -1954,20 +1978,12 @@ export class DaemonClient { requireCleanTarget: input.requireCleanTarget, requestId: resolvedRequestId, }); - return this.sendRequest({ + return this.sendCorrelatedRequest({ requestId: resolvedRequestId, message, + responseType: "checkout_merge_from_base_response", timeout: 60000, options: { skipQueue: true }, - select: (msg) => { - if (msg.type !== "checkout_merge_from_base_response") { - return null; - } - if (msg.payload.requestId !== resolvedRequestId) { - return null; - } - return msg.payload; - }, }); } @@ -1978,20 +1994,12 @@ export class DaemonClient { cwd, requestId: resolvedRequestId, }); - return this.sendRequest({ + return this.sendCorrelatedRequest({ requestId: resolvedRequestId, message, + responseType: "checkout_push_response", timeout: 60000, options: { skipQueue: true }, - select: (msg) => { - if (msg.type !== "checkout_push_response") { - return null; - } - if (msg.payload.requestId !== resolvedRequestId) { - return null; - } - return msg.payload; - }, }); } @@ -2009,20 +2017,12 @@ export class DaemonClient { baseRef: input.baseRef, requestId: resolvedRequestId, }); - return this.sendRequest({ + return this.sendCorrelatedRequest({ requestId: resolvedRequestId, message, + responseType: "checkout_pr_create_response", timeout: 60000, options: { skipQueue: true }, - select: (msg) => { - if (msg.type !== "checkout_pr_create_response") { - return null; - } - if (msg.payload.requestId !== resolvedRequestId) { - return null; - } - return msg.payload; - }, }); } @@ -2036,20 +2036,12 @@ export class DaemonClient { cwd, requestId: resolvedRequestId, }); - return this.sendRequest({ + return this.sendCorrelatedRequest({ requestId: resolvedRequestId, message, + responseType: "checkout_pr_status_response", timeout: 60000, options: { skipQueue: true }, - select: (msg) => { - if (msg.type !== "checkout_pr_status_response") { - return null; - } - if (msg.payload.requestId !== resolvedRequestId) { - return null; - } - return msg.payload; - }, }); } @@ -2064,20 +2056,12 @@ export class DaemonClient { repoRoot: input.repoRoot, requestId: resolvedRequestId, }); - return this.sendRequest({ + return this.sendCorrelatedRequest({ requestId: resolvedRequestId, message, + responseType: "paseo_worktree_list_response", timeout: 60000, options: { skipQueue: true }, - select: (msg) => { - if (msg.type !== "paseo_worktree_list_response") { - return null; - } - if (msg.payload.requestId !== resolvedRequestId) { - return null; - } - return msg.payload; - }, }); } @@ -2093,20 +2077,12 @@ export class DaemonClient { branchName: input.branchName, requestId: resolvedRequestId, }); - return this.sendRequest({ + return this.sendCorrelatedRequest({ requestId: resolvedRequestId, message, + responseType: "paseo_worktree_archive_response", timeout: 20000, options: { skipQueue: true }, - select: (msg) => { - if (msg.type !== "paseo_worktree_archive_response") { - return null; - } - if (msg.payload.requestId !== resolvedRequestId) { - return null; - } - return msg.payload; - }, }); } @@ -2121,20 +2097,12 @@ export class DaemonClient { branchName: options.branchName, requestId: resolvedRequestId, }); - return this.sendRequest({ + return this.sendCorrelatedRequest({ requestId: resolvedRequestId, message, + responseType: "validate_branch_response", timeout: 10000, options: { skipQueue: true }, - select: (msg) => { - if (msg.type !== "validate_branch_response") { - return null; - } - if (msg.payload.requestId !== resolvedRequestId) { - return null; - } - return msg.payload; - }, }); } @@ -2150,20 +2118,12 @@ export class DaemonClient { limit: options.limit, requestId: resolvedRequestId, }); - return this.sendRequest({ + return this.sendCorrelatedRequest({ requestId: resolvedRequestId, message, + responseType: "branch_suggestions_response", timeout: 10000, options: { skipQueue: true }, - select: (msg) => { - if (msg.type !== "branch_suggestions_response") { - return null; - } - if (msg.payload.requestId !== resolvedRequestId) { - return null; - } - return msg.payload; - }, }); } @@ -2185,20 +2145,12 @@ export class DaemonClient { mode, requestId: resolvedRequestId, }); - return this.sendRequest({ + return this.sendCorrelatedRequest({ requestId: resolvedRequestId, message, + responseType: "file_explorer_response", timeout: 10000, options: { skipQueue: true }, - select: (msg) => { - if (msg.type !== "file_explorer_response") { - return null; - } - if (msg.payload.requestId !== resolvedRequestId) { - return null; - } - return msg.payload; - }, }); } @@ -2214,20 +2166,12 @@ export class DaemonClient { path, requestId: resolvedRequestId, }); - return this.sendRequest({ + return this.sendCorrelatedRequest({ requestId: resolvedRequestId, message, + responseType: "file_download_token_response", timeout: 10000, options: { skipQueue: true }, - select: (msg) => { - if (msg.type !== "file_download_token_response") { - return null; - } - if (msg.payload.requestId !== resolvedRequestId) { - return null; - } - return msg.payload; - }, }); } @@ -2241,20 +2185,12 @@ export class DaemonClient { cwd, requestId: resolvedRequestId, }); - return this.sendRequest({ + return this.sendCorrelatedRequest({ requestId: resolvedRequestId, message, + responseType: "project_icon_response", timeout: 10000, options: { skipQueue: true }, - select: (msg) => { - if (msg.type !== "project_icon_response") { - return null; - } - if (msg.payload.requestId !== resolvedRequestId) { - return null; - } - return msg.payload; - }, }); } @@ -2273,20 +2209,12 @@ export class DaemonClient { cwd: options?.cwd, requestId: resolvedRequestId, }); - return this.sendRequest({ + return this.sendCorrelatedRequest({ requestId: resolvedRequestId, message, + responseType: "list_provider_models_response", timeout: 30000, options: { skipQueue: true }, - select: (msg) => { - if (msg.type !== "list_provider_models_response") { - return null; - } - if (msg.payload.requestId !== resolvedRequestId) { - return null; - } - return msg.payload; - }, }); } @@ -2298,20 +2226,12 @@ export class DaemonClient { type: "list_available_providers_request", requestId: resolvedRequestId, }); - return this.sendRequest({ + return this.sendCorrelatedRequest({ requestId: resolvedRequestId, message, + responseType: "list_available_providers_response", timeout: 30000, options: { skipQueue: true }, - select: (msg) => { - if (msg.type !== "list_available_providers_response") { - return null; - } - if (msg.payload.requestId !== resolvedRequestId) { - return null; - } - return msg.payload; - }, }); } @@ -2321,20 +2241,12 @@ export class DaemonClient { type: "speech_models_list_request", requestId: resolvedRequestId, }); - return this.sendRequest({ + return this.sendCorrelatedRequest({ requestId: resolvedRequestId, message, + responseType: "speech_models_list_response", timeout: 30000, options: { skipQueue: true }, - select: (msg) => { - if (msg.type !== "speech_models_list_response") { - return null; - } - if (msg.payload.requestId !== resolvedRequestId) { - return null; - } - return msg.payload; - }, }); } @@ -2347,20 +2259,12 @@ export class DaemonClient { modelIds: options?.modelIds, requestId: resolvedRequestId, }); - return this.sendRequest({ + return this.sendCorrelatedRequest({ requestId: resolvedRequestId, message, + responseType: "speech_models_download_response", timeout: 30 * 60 * 1000, options: { skipQueue: true }, - select: (msg) => { - if (msg.type !== "speech_models_download_response") { - return null; - } - if (msg.payload.requestId !== resolvedRequestId) { - return null; - } - return msg.payload; - }, }); } @@ -2374,20 +2278,12 @@ export class DaemonClient { agentId, requestId: resolvedRequestId, }); - return this.sendRequest({ + return this.sendCorrelatedRequest({ requestId: resolvedRequestId, message, + responseType: "list_commands_response", timeout: 30000, options: { skipQueue: true }, - select: (msg) => { - if (msg.type !== "list_commands_response") { - return null; - } - if (msg.payload.requestId !== resolvedRequestId) { - return null; - } - return msg.payload; - }, }); } @@ -2405,20 +2301,12 @@ export class DaemonClient { args, requestId: resolvedRequestId, }); - return this.sendRequest({ + return this.sendCorrelatedRequest({ requestId: resolvedRequestId, message, + responseType: "execute_command_response", timeout: 30000, options: { skipQueue: true }, - select: (msg) => { - if (msg.type !== "execute_command_response") { - return null; - } - if (msg.payload.requestId !== resolvedRequestId) { - return null; - } - return msg.payload; - }, }); } @@ -2502,20 +2390,12 @@ export class DaemonClient { agentId, timeoutMs: timeout, }); - const payload = await this.sendRequest({ + const payload = await this.sendCorrelatedRequest({ requestId, message, + responseType: "wait_for_finish_response", timeout: timeout + 5000, options: { skipQueue: true }, - select: (msg) => { - if (msg.type !== "wait_for_finish_response") { - return null; - } - if (msg.payload.requestId !== requestId) { - return null; - } - return msg.payload; - }, }); return { status: payload.status, @@ -2539,20 +2419,12 @@ export class DaemonClient { cwd, requestId: resolvedRequestId, }); - return this.sendRequest({ + return this.sendCorrelatedRequest({ requestId: resolvedRequestId, message, + responseType: "list_terminals_response", timeout: 10000, options: { skipQueue: true }, - select: (msg) => { - if (msg.type !== "list_terminals_response") { - return null; - } - if (msg.payload.requestId !== resolvedRequestId) { - return null; - } - return msg.payload; - }, }); } @@ -2568,20 +2440,12 @@ export class DaemonClient { name, requestId: resolvedRequestId, }); - return this.sendRequest({ + return this.sendCorrelatedRequest({ requestId: resolvedRequestId, message, + responseType: "create_terminal_response", timeout: 10000, options: { skipQueue: true }, - select: (msg) => { - if (msg.type !== "create_terminal_response") { - return null; - } - if (msg.payload.requestId !== resolvedRequestId) { - return null; - } - return msg.payload; - }, }); } @@ -2595,20 +2459,12 @@ export class DaemonClient { terminalId, requestId: resolvedRequestId, }); - return this.sendRequest({ + return this.sendCorrelatedRequest({ requestId: resolvedRequestId, message, + responseType: "subscribe_terminal_response", timeout: 10000, options: { skipQueue: true }, - select: (msg) => { - if (msg.type !== "subscribe_terminal_response") { - return null; - } - if (msg.payload.requestId !== resolvedRequestId) { - return null; - } - return msg.payload; - }, }); } @@ -2640,20 +2496,12 @@ export class DaemonClient { terminalId, requestId: resolvedRequestId, }); - return this.sendRequest({ + return this.sendCorrelatedRequest({ requestId: resolvedRequestId, message, + responseType: "kill_terminal_response", timeout: 10000, options: { skipQueue: true }, - select: (msg) => { - if (msg.type !== "kill_terminal_response") { - return null; - } - if (msg.payload.requestId !== resolvedRequestId) { - return null; - } - return msg.payload; - }, }); } @@ -2675,20 +2523,12 @@ export class DaemonClient { ...(options?.rows !== undefined ? { rows: options.rows } : {}), ...(options?.cols !== undefined ? { cols: options.cols } : {}), }); - return this.sendRequest({ + return this.sendCorrelatedRequest({ requestId: resolvedRequestId, message, + responseType: "attach_terminal_stream_response", timeout: 10000, options: { skipQueue: true }, - select: (msg) => { - if (msg.type !== "attach_terminal_stream_response") { - return null; - } - if (msg.payload.requestId !== resolvedRequestId) { - return null; - } - return msg.payload; - }, }); } @@ -2702,20 +2542,12 @@ export class DaemonClient { streamId, requestId: resolvedRequestId, }); - const payload = await this.sendRequest({ + const payload = await this.sendCorrelatedRequest({ requestId: resolvedRequestId, message, + responseType: "detach_terminal_stream_response", timeout: 10000, options: { skipQueue: true }, - select: (msg) => { - if (msg.type !== "detach_terminal_stream_response") { - return null; - } - if (msg.payload.requestId !== resolvedRequestId) { - return null; - } - return msg.payload; - }, }); this.terminalStreamHandlers.delete(streamId); this.bufferedTerminalStreamChunks.delete(streamId);