diff --git a/packages/app/src/components/agent-list.tsx b/packages/app/src/components/agent-list.tsx index 8acf855b6..7fe680f65 100644 --- a/packages/app/src/components/agent-list.tsx +++ b/packages/app/src/components/agent-list.tsx @@ -172,6 +172,8 @@ export function AgentList({ queryKey, queryFn: async () => await client.getCheckoutStatus(agent.cwd), staleTime: CHECKOUT_STATUS_STALE_TIME, + }).catch((error) => { + console.warn("[checkout_status] prefetch failed", error); }); } } diff --git a/packages/app/src/components/agent-stream-view.tsx b/packages/app/src/components/agent-stream-view.tsx index 2c3be4a9d..e4940382b 100644 --- a/packages/app/src/components/agent-stream-view.tsx +++ b/packages/app/src/components/agent-stream-view.tsx @@ -45,7 +45,7 @@ import type { AgentPermissionResponse } from "@server/server/agent/agent-sdk-typ import type { Agent } from "@/contexts/session-context"; import { useSessionStore } from "@/stores/session-store"; import { useFileExplorerActions } from "@/hooks/use-file-explorer-actions"; -import type { DaemonClientV2 } from "@server/client/daemon-client-v2"; +import type { DaemonClient } from "@server/client/daemon-client"; import { parseToolCallDisplay } from "@/utils/tool-call-parsers"; import { ToolCallDetailsContent } from "./tool-call-details"; import { ToolCallSheetProvider } from "./tool-call-sheet"; @@ -760,7 +760,7 @@ function PermissionRequestCard({ client, }: { permission: PendingPermission; - client: DaemonClientV2 | null; + client: DaemonClient | null; }) { const { theme } = useUnistyles(); const isMobile = diff --git a/packages/app/src/components/grouped-agent-list.tsx b/packages/app/src/components/grouped-agent-list.tsx index 9ac14235e..bd0e95e05 100644 --- a/packages/app/src/components/grouped-agent-list.tsx +++ b/packages/app/src/components/grouped-agent-list.tsx @@ -302,6 +302,8 @@ export function GroupedAgentList({ queryKey, queryFn: async () => await client.getCheckoutStatus(agent.cwd), staleTime: CHECKOUT_STATUS_STALE_TIME, + }).catch((error) => { + console.warn("[checkout_status] prefetch failed", error); }); } }, [agents, queryClient]); @@ -325,7 +327,9 @@ export function GroupedAgentList({ const session = useSessionStore.getState().sessions[agent.serverId]; const client = session?.client ?? null; if (client) { - void client.archiveAgent(agent.id); + void client.archiveAgent(agent.id).catch((error) => { + console.warn("[archive_agent] failed", error); + }); } }, [] diff --git a/packages/app/src/components/message-input.tsx b/packages/app/src/components/message-input.tsx index 4b19e290e..2bb290bd2 100644 --- a/packages/app/src/components/message-input.tsx +++ b/packages/app/src/components/message-input.tsx @@ -26,7 +26,7 @@ import Animated, { } from "react-native-reanimated"; import { useDictation } from "@/hooks/use-dictation"; import { DictationOverlay } from "./dictation-controls"; -import type { DaemonClientV2 } from "@server/client/daemon-client-v2"; +import type { DaemonClient } from "@server/client/daemon-client"; import { usePanelStore } from "@/stores/panel-store"; import { useVoiceOptional } from "@/contexts/voice-context"; @@ -51,7 +51,7 @@ export interface MessageInputProps { images?: ImageAttachment[]; onPickImages?: () => void; onRemoveImage?: (index: number) => void; - client: DaemonClientV2 | null; + client: DaemonClient | null; placeholder?: string; autoFocus?: boolean; disabled?: boolean; diff --git a/packages/app/src/contexts/session-context.tsx b/packages/app/src/contexts/session-context.tsx index ae31a7de2..981f5246a 100644 --- a/packages/app/src/contexts/session-context.tsx +++ b/packages/app/src/contexts/session-context.tsx @@ -20,7 +20,7 @@ import type { } from "@server/shared/messages"; import type { AgentLifecycleStatus } from "@server/shared/agent-lifecycle"; import type { AgentPermissionRequest } from "@server/server/agent/agent-sdk-types"; -import type { DaemonClientV2, ConnectionState } from "@server/client/daemon-client-v2"; +import type { DaemonClient, ConnectionState } from "@server/client/daemon-client"; import { File } from "expo-file-system"; import { useDaemonConnections } from "./daemon-connections-context"; import { diff --git a/packages/app/src/dictation/dictation-stream-sender.ts b/packages/app/src/dictation/dictation-stream-sender.ts index 6819bd6a4..82d11ee42 100644 --- a/packages/app/src/dictation/dictation-stream-sender.ts +++ b/packages/app/src/dictation/dictation-stream-sender.ts @@ -1,8 +1,8 @@ import { generateMessageId } from "@/types/stream"; -import type { DaemonClientV2 } from "@server/client/daemon-client-v2"; +import type { DaemonClient } from "@server/client/daemon-client"; export type DictationStreamSenderParams = { - client: DaemonClientV2 | null; + client: DaemonClient | null; format: string; createDictationId?: () => string; }; @@ -22,7 +22,7 @@ type DictationFinishResult = { dictationId: string; text: string }; * so enqueues can't "miss" a flush due to in-flight await/coalescing bugs. */ export class DictationStreamSender { - private client: DaemonClientV2 | null; + private client: DaemonClient | null; private readonly format: string; private readonly createDictationId: () => string; @@ -40,7 +40,7 @@ export class DictationStreamSender { this.createDictationId = params.createDictationId ?? generateMessageId; } - setClient(client: DaemonClientV2 | null): void { + setClient(client: DaemonClient | null): void { this.client = client; } diff --git a/packages/app/src/hooks/use-client-activity.ts b/packages/app/src/hooks/use-client-activity.ts index ad2099a14..600457c89 100644 --- a/packages/app/src/hooks/use-client-activity.ts +++ b/packages/app/src/hooks/use-client-activity.ts @@ -1,12 +1,12 @@ import { useEffect, useRef, useCallback } from "react"; import { AppState, Platform } from "react-native"; -import type { DaemonClientV2 } from "@server/client/daemon-client-v2"; +import type { DaemonClient } from "@server/client/daemon-client"; const HEARTBEAT_INTERVAL_MS = 15_000; const ACTIVITY_HEARTBEAT_THROTTLE_MS = 5_000; interface ClientActivityOptions { - client: DaemonClientV2; + client: DaemonClient; focusedAgentId: string | null; } diff --git a/packages/app/src/hooks/use-daemon-client.ts b/packages/app/src/hooks/use-daemon-client.ts index 9bd6e78eb..fac963f69 100644 --- a/packages/app/src/hooks/use-daemon-client.ts +++ b/packages/app/src/hooks/use-daemon-client.ts @@ -1,6 +1,6 @@ import { useEffect, useMemo } from "react"; import { AppState } from "react-native"; -import { DaemonClientV2 } from "@server/client/daemon-client-v2"; +import { DaemonClient } from "@server/client/daemon-client"; import { createTauriWebSocketTransportFactory } from "@/utils/tauri-daemon-transport"; function runDaemonRequest(label: string, promise: Promise): void { @@ -9,11 +9,11 @@ function runDaemonRequest(label: string, promise: Promise): void { }); } -export function useDaemonClient(url: string): DaemonClientV2 { +export function useDaemonClient(url: string): DaemonClient { const client = useMemo( () => { const tauriTransportFactory = createTauriWebSocketTransportFactory(); - return new DaemonClientV2({ + return new DaemonClient({ url, suppressSendErrors: true, ...(tauriTransportFactory diff --git a/packages/app/src/hooks/use-dictation.shared.ts b/packages/app/src/hooks/use-dictation.shared.ts index e5a047749..bd74c836b 100644 --- a/packages/app/src/hooks/use-dictation.shared.ts +++ b/packages/app/src/hooks/use-dictation.shared.ts @@ -1,7 +1,7 @@ export type DictationStatus = "idle" | "recording" | "uploading" | "failed"; export type UseDictationOptions = { - client: import("@server/client/daemon-client-v2").DaemonClientV2 | null; + client: import("@server/client/daemon-client").DaemonClient | null; onTranscript: (text: string, meta: { requestId: string }) => void; onPartialTranscript?: (text: string, meta: { requestId: string }) => void; onError?: (error: Error) => void; diff --git a/packages/app/src/hooks/use-push-token-registration.ts b/packages/app/src/hooks/use-push-token-registration.ts index 9248fe413..73337e5af 100644 --- a/packages/app/src/hooks/use-push-token-registration.ts +++ b/packages/app/src/hooks/use-push-token-registration.ts @@ -3,7 +3,7 @@ import { Platform } from "react-native"; import AsyncStorage from "@react-native-async-storage/async-storage"; import * as Notifications from "expo-notifications"; import Constants from "expo-constants"; -import type { DaemonClientV2 } from "@server/client/daemon-client-v2"; +import type { DaemonClient } from "@server/client/daemon-client"; const STORAGE_PREFIX = "@paseo:expo-push-token:"; @@ -26,7 +26,7 @@ async function ensurePushPermission(): Promise { } export function usePushTokenRegistration(params: { - client: DaemonClientV2; + client: DaemonClient; serverId: string; }): void { const { client, serverId } = params; diff --git a/packages/app/src/stores/session-store.ts b/packages/app/src/stores/session-store.ts index 37552812d..7a006e60f 100644 --- a/packages/app/src/stores/session-store.ts +++ b/packages/app/src/stores/session-store.ts @@ -1,6 +1,6 @@ import { create } from "zustand"; import { subscribeWithSelector } from "zustand/middleware"; -import type { DaemonClientV2 } from "@server/client/daemon-client-v2"; +import type { DaemonClient } from "@server/client/daemon-client"; import type { useAudioPlayer } from "@/hooks/use-audio-player"; import type { AgentDirectoryEntry } from "@/types/agent-directory"; import type { StreamItem } from "@/types/stream"; @@ -150,7 +150,7 @@ export interface SessionState { serverId: string; // Daemon client (immutable reference) - client: DaemonClientV2 | null; + client: DaemonClient | null; // Connection snapshot (mutable) connection: DaemonConnectionSnapshot; @@ -205,10 +205,10 @@ interface SessionStoreState { // Action types interface SessionStoreActions { // Session management - initializeSession: (serverId: string, client: DaemonClientV2, audioPlayer: ReturnType) => void; + initializeSession: (serverId: string, client: DaemonClient, audioPlayer: ReturnType) => void; clearSession: (serverId: string) => void; getSession: (serverId: string) => SessionState | undefined; - updateSessionClient: (serverId: string, client: DaemonClientV2) => void; + updateSessionClient: (serverId: string, client: DaemonClient) => void; updateSessionConnection: (serverId: string, connection: DaemonConnectionSnapshot) => void; // Audio state @@ -280,7 +280,7 @@ function logSessionStoreUpdate( } -function createDefaultConnectionSnapshot(client?: DaemonClientV2 | null): DaemonConnectionSnapshot { +function createDefaultConnectionSnapshot(client?: DaemonClient | null): DaemonConnectionSnapshot { if (!client) { return { isConnected: false, isConnecting: false, lastError: null }; } @@ -293,7 +293,7 @@ function createDefaultConnectionSnapshot(client?: DaemonClientV2 | null): Daemon } // Helper to create initial session state -function createInitialSessionState(serverId: string, client: DaemonClientV2, audioPlayer: ReturnType): SessionState { +function createInitialSessionState(serverId: string, client: DaemonClient, audioPlayer: ReturnType): SessionState { return { serverId, client, diff --git a/packages/app/src/utils/tauri-daemon-transport.ts b/packages/app/src/utils/tauri-daemon-transport.ts index 615baca28..2e90bcb14 100644 --- a/packages/app/src/utils/tauri-daemon-transport.ts +++ b/packages/app/src/utils/tauri-daemon-transport.ts @@ -1,4 +1,4 @@ -import type { DaemonTransport, DaemonTransportFactory } from "@server/client/daemon-client-v2"; +import type { DaemonTransport, DaemonTransportFactory } from "@server/client/daemon-client"; type TauriWebSocketMessage = | { type: "Text"; data: string } diff --git a/packages/app/src/utils/test-daemon-connection.ts b/packages/app/src/utils/test-daemon-connection.ts index a487e1073..a6ccbb190 100644 --- a/packages/app/src/utils/test-daemon-connection.ts +++ b/packages/app/src/utils/test-daemon-connection.ts @@ -1,5 +1,5 @@ -import { DaemonClientV2 } from "@server/client/daemon-client-v2"; -import type { ConnectionState } from "@server/client/daemon-client-v2"; +import { DaemonClient } from "@server/client/daemon-client"; +import type { ConnectionState } from "@server/client/daemon-client"; import { buildDaemonWebSocketUrl } from "./daemon-endpoints"; function normalizeNonEmptyString(value: unknown): string | null { @@ -45,7 +45,7 @@ export async function testDaemonEndpointConnection( const timeoutMs = options?.timeoutMs ?? 6000; const url = buildDaemonWebSocketUrl(endpoint); - const client = new DaemonClientV2({ + const client = new DaemonClient({ url, suppressSendErrors: true, }); diff --git a/packages/cli/docs/type-audit.md b/packages/cli/docs/type-audit.md index 0bb1b528b..6a1f86b1d 100644 --- a/packages/cli/docs/type-audit.md +++ b/packages/cli/docs/type-audit.md @@ -11,7 +11,7 @@ - `loadConfig`, `resolvePaseoHome` - `createRootLogger`, `LogLevel`, `LogFormat` - `loadPersistedConfig`, `PersistedConfig` -- `DaemonClientV2`, `DaemonClientV2Config`, `ConnectionState`, `DaemonEvent` +- `DaemonClient`, `DaemonClientConfig`, `ConnectionState`, `DaemonEvent` No agent snapshot/timeline/permission/message types are exported. diff --git a/packages/cli/src/commands/agent/attach.ts b/packages/cli/src/commands/agent/attach.ts index 0a062d214..f4cf8bede 100644 --- a/packages/cli/src/commands/agent/attach.ts +++ b/packages/cli/src/commands/agent/attach.ts @@ -1,7 +1,7 @@ import type { Command } from 'commander' import { connectToDaemon, getDaemonHost } from '../../utils/client.js' import type { - DaemonClientV2, + DaemonClient, AgentStreamMessage, AgentStreamSnapshotMessage, AgentStreamEventPayload, @@ -107,7 +107,7 @@ export async function runAttachCommand( process.exit(1) } - let client: DaemonClientV2 + let client: DaemonClient try { client = await connectToDaemon({ host: options.host as string | undefined }) } catch (err) { diff --git a/packages/cli/src/commands/agent/logs.ts b/packages/cli/src/commands/agent/logs.ts index 56aed5249..9660883ff 100644 --- a/packages/cli/src/commands/agent/logs.ts +++ b/packages/cli/src/commands/agent/logs.ts @@ -2,7 +2,7 @@ import type { Command } from 'commander' import { connectToDaemon, getDaemonHost } from '../../utils/client.js' import type { CommandOptions } from '../../output/index.js' import type { - DaemonClientV2, + DaemonClient, AgentStreamMessage, AgentStreamSnapshotMessage, AgentTimelineItem, @@ -80,7 +80,7 @@ export async function runLogsCommand( process.exit(1) } - let client: DaemonClientV2 + let client: DaemonClient try { client = await connectToDaemon({ host: options.host as string | undefined }) } catch (err) { @@ -172,7 +172,7 @@ export async function runLogsCommand( * Follow mode: stream logs in real-time until interrupted */ async function runFollowMode( - client: DaemonClientV2, + client: DaemonClient, agentId: string, options: AgentLogsOptions ): Promise { diff --git a/packages/cli/src/utils/client.ts b/packages/cli/src/utils/client.ts index 6f049151d..c0deea3ed 100644 --- a/packages/cli/src/utils/client.ts +++ b/packages/cli/src/utils/client.ts @@ -1,4 +1,4 @@ -import { DaemonClientV2 } from '@paseo/server' +import { DaemonClient } from '@paseo/server' import WebSocket from 'ws' export interface ConnectOptions { @@ -35,12 +35,12 @@ function createNodeWebSocketFactory() { * Create and connect a daemon client * Returns the connected client or throws if connection fails */ -export async function connectToDaemon(options?: ConnectOptions): Promise { +export async function connectToDaemon(options?: ConnectOptions): Promise { const host = getDaemonHost(options) const timeout = options?.timeout ?? DEFAULT_TIMEOUT const url = `ws://${host}/ws` - const client = new DaemonClientV2({ + const client = new DaemonClient({ url, webSocketFactory: createNodeWebSocketFactory(), reconnect: { enabled: false }, @@ -76,7 +76,7 @@ export async function connectToDaemon(options?: ConnectOptions): Promise { +export async function tryConnectToDaemon(options?: ConnectOptions): Promise { try { return await connectToDaemon(options) } catch { diff --git a/packages/server/src/client/daemon-client-v2.test.ts b/packages/server/src/client/daemon-client.test.ts similarity index 77% rename from packages/server/src/client/daemon-client-v2.test.ts rename to packages/server/src/client/daemon-client.test.ts index ec7cdc243..c98c16ac1 100644 --- a/packages/server/src/client/daemon-client-v2.test.ts +++ b/packages/server/src/client/daemon-client.test.ts @@ -1,5 +1,5 @@ import { afterEach, describe, expect, test, vi } from "vitest"; -import { DaemonClientV2, type DaemonTransport } from "./daemon-client-v2"; +import { DaemonClient, type DaemonTransport } from "./daemon-client"; function createMockLogger() { return { @@ -49,8 +49,8 @@ function createMockTransport() { }; } -describe("DaemonClientV2", () => { - const clients: DaemonClientV2[] = []; +describe("DaemonClient", () => { + const clients: DaemonClient[] = []; afterEach(async () => { for (const client of clients) { @@ -63,7 +63,7 @@ describe("DaemonClientV2", () => { const logger = createMockLogger(); const mock = createMockTransport(); - const client = new DaemonClientV2({ + const client = new DaemonClient({ url: "ws://test", logger, reconnect: { enabled: false }, @@ -137,4 +137,38 @@ describe("DaemonClientV2", () => { isGit: false, }); }); + + test("cancels waiters when send fails (no leaked timeouts)", async () => { + vi.useFakeTimers(); + const logger = createMockLogger(); + const mock = createMockTransport(); + + const transportFactory = () => ({ + ...mock.transport, + send: () => { + throw new Error("boom"); + }, + }); + + const client = new DaemonClient({ + url: "ws://test", + logger, + reconnect: { enabled: false }, + transportFactory, + }); + clients.push(client); + + const connectPromise = client.connect(); + mock.triggerOpen(); + await connectPromise; + + const promise = client.getCheckoutStatus("/tmp/project"); + await expect(promise).rejects.toThrow("boom"); + + // Ensure we didn't leave a waiter behind that will reject later. + expect((client as any).waiters.size).toBe(0); + + vi.runOnlyPendingTimers(); + vi.useRealTimers(); + }); }); diff --git a/packages/server/src/client/daemon-client-v2.ts b/packages/server/src/client/daemon-client.ts similarity index 84% rename from packages/server/src/client/daemon-client-v2.ts rename to packages/server/src/client/daemon-client.ts index bb77fe7a7..652bd27bd 100644 --- a/packages/server/src/client/daemon-client-v2.ts +++ b/packages/server/src/client/daemon-client.ts @@ -139,7 +139,7 @@ export type DaemonEvent = export type DaemonEventHandler = (event: DaemonEvent) => void; -export type DaemonClientV2Config = { +export type DaemonClientConfig = { url: string; authHeader?: string; suppressSendErrors?: boolean; @@ -220,6 +220,29 @@ type Waiter = { timeoutHandle: ReturnType | null; }; +type WaitHandle = { + promise: Promise; + cancel: (error: Error) => void; +}; + +type RpcWaitResult = + | { kind: "ok"; value: T } + | { kind: "error"; error: DaemonRpcError }; + +class DaemonRpcError extends Error { + readonly requestId: string; + readonly requestType?: string; + readonly code?: string; + + constructor(params: { requestId: string; error: string; requestType?: string; code?: string }) { + super(params.error); + this.name = "DaemonRpcError"; + this.requestId = params.requestId; + this.requestType = params.requestType; + this.code = params.code; + } +} + const DEFAULT_RECONNECT_BASE_DELAY_MS = 1500; const DEFAULT_RECONNECT_MAX_DELAY_MS = 30000; @@ -233,7 +256,7 @@ interface PendingSend { timeoutHandle: ReturnType; } -export class DaemonClientV2 { +export class DaemonClient { private transport: DaemonTransport | null = null; private transportCleanup: Array<() => void> = []; private rawMessageListeners: Set<(message: SessionOutboundMessage) => void> = new Set(); @@ -262,7 +285,7 @@ export class DaemonClientV2 { private logger: Logger; private pendingSendQueue: PendingSend[] = []; - constructor(private config: DaemonClientV2Config) { + constructor(private config: DaemonClientConfig) { this.logger = config.logger ?? consoleLogger; } @@ -616,6 +639,54 @@ export class DaemonClientV2 { } } + private async sendRequest( + params: { + requestId: string; + message: SessionInboundMessage; + timeout: number; + select: (msg: SessionOutboundMessage) => T | null; + options?: { skipQueue?: boolean }; + } + ): Promise { + const { promise, cancel } = this.waitForWithCancel>( + (msg) => { + if (msg.type === "rpc_error" && msg.payload.requestId === params.requestId) { + return { + kind: "error", + error: new DaemonRpcError({ + requestId: msg.payload.requestId, + error: msg.payload.error, + requestType: msg.payload.requestType, + code: msg.payload.code, + }), + }; + } + const value = params.select(msg); + if (value === null) { + return null; + } + return { kind: "ok", value }; + }, + params.timeout, + params.options + ); + + try { + await this.sendSessionMessageOrThrow(params.message); + } catch (error) { + const err = error instanceof Error ? error : new Error(String(error)); + cancel(err); + void promise.catch(() => undefined); + throw err; + } + + const result = await promise; + if (result.kind === "error") { + throw result.error; + } + return result.value; + } + private sendSessionMessageStrict(message: SessionInboundMessage): void { if (!this.transport || this.connectionState.status !== "connected") { throw new Error("Transport not connected"); @@ -672,9 +743,12 @@ export class DaemonClientV2 { requestId: resolvedRequestId, ...(options?.filter ? { filter: options.filter } : {}), }); - - const response = this.waitFor( - (msg) => { + return this.sendRequest({ + requestId: resolvedRequestId, + message, + timeout: 10000, + options: { skipQueue: true }, + select: (msg) => { if (msg.type !== "fetch_agents_response") { return null; } @@ -683,11 +757,7 @@ export class DaemonClientV2 { } return msg.payload.agents; }, - 10000, - { skipQueue: true } - ); - await this.sendSessionMessageOrThrow(message); - return response; + }); } async fetchAgent(agentId: string, requestId?: string): Promise { @@ -697,9 +767,12 @@ export class DaemonClientV2 { requestId: resolvedRequestId, agentId, }); - - const response = this.waitFor( - (msg) => { + const payload = await this.sendRequest({ + requestId: resolvedRequestId, + message, + timeout: 10000, + options: { skipQueue: true }, + select: (msg) => { if (msg.type !== "fetch_agent_response") { return null; } @@ -708,11 +781,7 @@ export class DaemonClientV2 { } return msg.payload; }, - 10000, - { skipQueue: true } - ); - await this.sendSessionMessageOrThrow(message); - const payload = await response; + }); if (payload.error) { throw new Error(payload.error); } @@ -771,8 +840,12 @@ export class DaemonClientV2 { voiceConversationId, requestId: resolvedRequestId, }); - const response = this.waitFor( - (msg) => { + return this.sendRequest({ + requestId: resolvedRequestId, + message, + timeout: 10000, + options: { skipQueue: true }, + select: (msg) => { if (msg.type !== "voice_conversation_loaded") { return null; } @@ -781,11 +854,7 @@ export class DaemonClientV2 { } return msg.payload; }, - 10000, - { skipQueue: true } - ); - await this.sendSessionMessageOrThrow(message); - return response; + }); } async listVoiceConversations(requestId?: string): Promise { @@ -794,8 +863,12 @@ export class DaemonClientV2 { type: "list_voice_conversations_request", requestId: resolvedRequestId, }); - const response = this.waitFor( - (msg) => { + return this.sendRequest({ + requestId: resolvedRequestId, + message, + timeout: 10000, + options: { skipQueue: true }, + select: (msg) => { if (msg.type !== "list_voice_conversations_response") { return null; } @@ -804,11 +877,7 @@ export class DaemonClientV2 { } return msg.payload; }, - 10000, - { skipQueue: true } - ); - await this.sendSessionMessageOrThrow(message); - return response; + }); } async deleteVoiceConversation( @@ -821,8 +890,12 @@ export class DaemonClientV2 { voiceConversationId, requestId: resolvedRequestId, }); - const response = this.waitFor( - (msg) => { + return this.sendRequest({ + requestId: resolvedRequestId, + message, + timeout: 10000, + options: { skipQueue: true }, + select: (msg) => { if (msg.type !== "delete_voice_conversation_response") { return null; } @@ -831,11 +904,7 @@ export class DaemonClientV2 { } return msg.payload; }, - 10000, - { skipQueue: true } - ); - await this.sendSessionMessageOrThrow(message); - return response; + }); } // ============================================================================ @@ -861,8 +930,12 @@ export class DaemonClientV2 { : {}), }); - const statusPromise = this.waitFor( - (msg) => { + const status = await this.sendRequest({ + requestId, + message, + timeout: 15000, + options: { skipQueue: true }, + select: (msg) => { if (msg.type !== "status") { return null; } @@ -876,12 +949,7 @@ export class DaemonClientV2 { } return null; }, - 15000, - { skipQueue: true } - ); - - await this.sendSessionMessageOrThrow(message); - const status = await statusPromise; + }); if (status.status === "agent_create_failed") { throw new Error(status.error); } @@ -896,8 +964,12 @@ export class DaemonClientV2 { agentId, requestId, }); - const response = this.waitFor( - (msg) => { + await this.sendRequest({ + requestId, + message, + timeout: 10000, + options: { skipQueue: true }, + select: (msg) => { if (msg.type !== "agent_deleted") { return null; } @@ -906,11 +978,7 @@ export class DaemonClientV2 { } return msg.payload; }, - 10000, - { skipQueue: true } - ); - await this.sendSessionMessageOrThrow(message); - await response; + }); } async archiveAgent(agentId: string): Promise<{ archivedAt: string }> { @@ -920,8 +988,12 @@ export class DaemonClientV2 { agentId, requestId, }); - const response = this.waitFor( - (msg) => { + const result = await this.sendRequest({ + requestId, + message, + timeout: 10000, + options: { skipQueue: true }, + select: (msg) => { if (msg.type !== "agent_archived") { return null; } @@ -930,11 +1002,7 @@ export class DaemonClientV2 { } return msg.payload; }, - 10000, - { skipQueue: true } - ); - await this.sendSessionMessageOrThrow(message); - const result = await response; + }); return { archivedAt: result.archivedAt }; } @@ -950,8 +1018,12 @@ export class DaemonClientV2 { ...(overrides ? { overrides } : {}), }); - const statusPromise = this.waitFor( - (msg) => { + const status = await this.sendRequest({ + requestId, + message, + timeout: 15000, + options: { skipQueue: true }, + select: (msg) => { if (msg.type !== "status") { return null; } @@ -961,12 +1033,7 @@ export class DaemonClientV2 { } return null; }, - 15000, - { skipQueue: true } - ); - - await this.sendSessionMessageOrThrow(message); - const status = await statusPromise; + }); return status.agent; } @@ -981,8 +1048,12 @@ export class DaemonClientV2 { agentId, requestId: resolvedRequestId, }); - const response = this.waitFor( - (msg) => { + return this.sendRequest({ + requestId: resolvedRequestId, + message, + timeout: 15000, + options: { skipQueue: true }, + select: (msg) => { if (msg.type !== "status") { return null; } @@ -992,11 +1063,7 @@ export class DaemonClientV2 { } return null; }, - 15000, - { skipQueue: true } - ); - await this.sendSessionMessageOrThrow(message); - return response; + }); } async initializeAgent( @@ -1009,8 +1076,12 @@ export class DaemonClientV2 { agentId, requestId: resolvedRequestId, }); - const response = this.waitFor( - (msg) => { + const payload = await this.sendRequest({ + requestId: resolvedRequestId, + message, + timeout: 10000, + options: { skipQueue: true }, + select: (msg) => { if (msg.type !== "initialize_agent_request") { return null; } @@ -1019,11 +1090,7 @@ export class DaemonClientV2 { } return msg.payload; }, - 10000, - { skipQueue: true } - ); - await this.sendSessionMessageOrThrow(message); - const payload = await response; + }); if (payload.error) { throw new Error(payload.error); } @@ -1053,8 +1120,12 @@ export class DaemonClientV2 { ...(messageId ? { messageId } : {}), ...(options?.images ? { images: options.images } : {}), }); - const response = this.waitFor( - (msg) => { + const payload = await this.sendRequest({ + requestId, + message, + timeout: 15000, + options: { skipQueue: true }, + select: (msg) => { if (msg.type !== "send_agent_message_response") { return null; } @@ -1063,11 +1134,7 @@ export class DaemonClientV2 { } return msg.payload; }, - 15000, - { skipQueue: true } - ); - await this.sendSessionMessageOrThrow(message); - const payload = await response; + }); if (!payload.accepted) { throw new Error(payload.error ?? "sendAgentMessage rejected"); } @@ -1093,8 +1160,12 @@ export class DaemonClientV2 { modeId, requestId, }); - const response = this.waitFor( - (msg) => { + const payload = await this.sendRequest({ + requestId, + message, + timeout: 15000, + options: { skipQueue: true }, + select: (msg) => { if (msg.type !== "set_agent_mode_response") { return null; } @@ -1103,11 +1174,7 @@ export class DaemonClientV2 { } return msg.payload; }, - 15000, - { skipQueue: true } - ); - await this.sendSessionMessageOrThrow(message); - const payload = await response; + }); if (!payload.accepted) { throw new Error(payload.error ?? "setAgentMode rejected"); } @@ -1121,8 +1188,12 @@ export class DaemonClientV2 { modelId, requestId, }); - const response = this.waitFor( - (msg) => { + const payload = await this.sendRequest({ + requestId, + message, + timeout: 15000, + options: { skipQueue: true }, + select: (msg) => { if (msg.type !== "set_agent_model_response") { return null; } @@ -1131,11 +1202,7 @@ export class DaemonClientV2 { } return msg.payload; }, - 15000, - { skipQueue: true } - ); - await this.sendSessionMessageOrThrow(message); - const payload = await response; + }); if (!payload.accepted) { throw new Error(payload.error ?? "setAgentModel rejected"); } @@ -1152,8 +1219,12 @@ export class DaemonClientV2 { thinkingOptionId, requestId, }); - const response = this.waitFor( - (msg) => { + const payload = await this.sendRequest({ + requestId, + message, + timeout: 15000, + options: { skipQueue: true }, + select: (msg) => { if (msg.type !== "set_agent_thinking_response") { return null; } @@ -1162,11 +1233,7 @@ export class DaemonClientV2 { } return msg.payload; }, - 15000, - { skipQueue: true } - ); - await this.sendSessionMessageOrThrow(message); - const payload = await response; + }); if (!payload.accepted) { throw new Error(payload.error ?? "setAgentThinkingOption rejected"); } @@ -1182,9 +1249,12 @@ export class DaemonClientV2 { ...(reason && reason.trim().length > 0 ? { reason } : {}), requestId: resolvedRequestId, }); - - const response = this.waitFor( - (msg) => { + return this.sendRequest({ + requestId: resolvedRequestId, + message, + timeout: 10000, + options: { skipQueue: true }, + select: (msg) => { if (msg.type !== "status") { return null; } @@ -1199,12 +1269,7 @@ export class DaemonClientV2 { } return restarted.data; }, - 10000, - { skipQueue: true } - ); - - await this.sendSessionMessageOrThrow(message); - return response; + }); } // ============================================================================ @@ -1224,7 +1289,7 @@ export class DaemonClientV2 { } startDictationStream(dictationId: string, format: string): Promise { - const ackPromise = this.waitFor( + const ack = this.waitForWithCancel( (msg) => { if (msg.type !== "dictation_stream_ack") { return null; @@ -1239,9 +1304,10 @@ export class DaemonClientV2 { }, 30000, { skipQueue: true } - ).then(() => undefined); + ); + const ackPromise = ack.promise.then(() => undefined); - const errorPromise = this.waitFor( + const streamError = this.waitForWithCancel( (msg) => { if (msg.type !== "dictation_stream_error") { return null; @@ -1253,11 +1319,21 @@ export class DaemonClientV2 { }, 30000, { skipQueue: true } - ).then((payload) => { + ); + const errorPromise = streamError.promise.then((payload) => { throw new Error(payload.error); }); - this.sendSessionMessageStrict({ type: "dictation_stream_start", dictationId, format }); + try { + this.sendSessionMessageStrict({ type: "dictation_stream_start", dictationId, format }); + } catch (error) { + const err = error instanceof Error ? error : new Error(String(error)); + ack.cancel(err); + streamError.cancel(err); + void ackPromise.catch(() => undefined); + void errorPromise.catch(() => undefined); + throw err; + } return Promise.race([ackPromise, errorPromise]); } @@ -1266,7 +1342,7 @@ export class DaemonClientV2 { } finishDictationStream(dictationId: string, finalSeq: number): Promise<{ dictationId: string; text: string }> { - const finalPromise = this.waitFor( + const final = this.waitForWithCancel( (msg) => { if (msg.type !== "dictation_stream_final") { return null; @@ -1280,7 +1356,7 @@ export class DaemonClientV2 { { skipQueue: true } ); - const errorPromise = this.waitFor( + const streamError = this.waitForWithCancel( (msg) => { if (msg.type !== "dictation_stream_error") { return null; @@ -1292,11 +1368,23 @@ export class DaemonClientV2 { }, 30000, { skipQueue: true } - ).then((payload) => { + ); + + const finalPromise = final.promise; + const errorPromise = streamError.promise.then((payload) => { throw new Error(payload.error); }); - this.sendSessionMessageStrict({ type: "dictation_stream_finish", dictationId, finalSeq }); + try { + this.sendSessionMessageStrict({ type: "dictation_stream_finish", dictationId, finalSeq }); + } catch (error) { + const err = error instanceof Error ? error : new Error(String(error)); + final.cancel(err); + streamError.cancel(err); + void finalPromise.catch(() => undefined); + void errorPromise.catch(() => undefined); + throw err; + } return Promise.race([finalPromise, errorPromise]); } @@ -1336,31 +1424,31 @@ export class DaemonClientV2 { requestId: resolvedRequestId, }); - const responsePromise = (async () => { - const response = this.waitFor( - (msg) => { - if (msg.type !== "checkout_status_response") { - return null; - } - if (msg.payload.requestId !== resolvedRequestId) { - return null; - } - return msg.payload; - }, - 60000, - { skipQueue: true } - ); - await this.sendSessionMessageOrThrow(message); - return response; - })(); + const responsePromise = this.sendRequest({ + requestId: resolvedRequestId, + message, + timeout: 60000, + options: { skipQueue: true }, + select: (msg) => { + if (msg.type !== "checkout_status_response") { + return null; + } + if (msg.payload.requestId !== resolvedRequestId) { + return null; + } + return msg.payload; + }, + }); if (!requestId) { this.checkoutStatusInFlight.set(cwd, responsePromise); - responsePromise.finally(() => { - if (this.checkoutStatusInFlight.get(cwd) === responsePromise) { - this.checkoutStatusInFlight.delete(cwd); - } - }); + void responsePromise + .finally(() => { + if (this.checkoutStatusInFlight.get(cwd) === responsePromise) { + this.checkoutStatusInFlight.delete(cwd); + } + }) + .catch(() => undefined); } return responsePromise; @@ -1378,8 +1466,12 @@ export class DaemonClientV2 { compare, requestId: resolvedRequestId, }); - const response = this.waitFor( - (msg) => { + return this.sendRequest({ + requestId: resolvedRequestId, + message, + timeout: 60000, + options: { skipQueue: true }, + select: (msg) => { if (msg.type !== "checkout_diff_response") { return null; } @@ -1388,11 +1480,7 @@ export class DaemonClientV2 { } return msg.payload; }, - 60000, - { skipQueue: true } - ); - await this.sendSessionMessageOrThrow(message); - return response; + }); } async checkoutCommit( @@ -1408,8 +1496,12 @@ export class DaemonClientV2 { addAll: input.addAll, requestId: resolvedRequestId, }); - const response = this.waitFor( - (msg) => { + return this.sendRequest({ + requestId: resolvedRequestId, + message, + timeout: 60000, + options: { skipQueue: true }, + select: (msg) => { if (msg.type !== "checkout_commit_response") { return null; } @@ -1418,11 +1510,7 @@ export class DaemonClientV2 { } return msg.payload; }, - 60000, - { skipQueue: true } - ); - await this.sendSessionMessageOrThrow(message); - return response; + }); } async checkoutMerge( @@ -1439,8 +1527,12 @@ export class DaemonClientV2 { requireCleanTarget: input.requireCleanTarget, requestId: resolvedRequestId, }); - const response = this.waitFor( - (msg) => { + return this.sendRequest({ + requestId: resolvedRequestId, + message, + timeout: 60000, + options: { skipQueue: true }, + select: (msg) => { if (msg.type !== "checkout_merge_response") { return null; } @@ -1449,11 +1541,7 @@ export class DaemonClientV2 { } return msg.payload; }, - 60000, - { skipQueue: true } - ); - await this.sendSessionMessageOrThrow(message); - return response; + }); } async checkoutMergeFromBase( @@ -1469,8 +1557,12 @@ export class DaemonClientV2 { requireCleanTarget: input.requireCleanTarget, requestId: resolvedRequestId, }); - const response = this.waitFor( - (msg) => { + return this.sendRequest({ + requestId: resolvedRequestId, + message, + timeout: 60000, + options: { skipQueue: true }, + select: (msg) => { if (msg.type !== "checkout_merge_from_base_response") { return null; } @@ -1479,11 +1571,7 @@ export class DaemonClientV2 { } return msg.payload; }, - 60000, - { skipQueue: true } - ); - await this.sendSessionMessageOrThrow(message); - return response; + }); } async checkoutPush(cwd: string, requestId?: string): Promise { @@ -1493,8 +1581,12 @@ export class DaemonClientV2 { cwd, requestId: resolvedRequestId, }); - const response = this.waitFor( - (msg) => { + return this.sendRequest({ + requestId: resolvedRequestId, + message, + timeout: 60000, + options: { skipQueue: true }, + select: (msg) => { if (msg.type !== "checkout_push_response") { return null; } @@ -1503,11 +1595,7 @@ export class DaemonClientV2 { } return msg.payload; }, - 60000, - { skipQueue: true } - ); - await this.sendSessionMessageOrThrow(message); - return response; + }); } async checkoutPrCreate( @@ -1524,8 +1612,12 @@ export class DaemonClientV2 { baseRef: input.baseRef, requestId: resolvedRequestId, }); - const response = this.waitFor( - (msg) => { + return this.sendRequest({ + requestId: resolvedRequestId, + message, + timeout: 60000, + options: { skipQueue: true }, + select: (msg) => { if (msg.type !== "checkout_pr_create_response") { return null; } @@ -1534,11 +1626,7 @@ export class DaemonClientV2 { } return msg.payload; }, - 60000, - { skipQueue: true } - ); - await this.sendSessionMessageOrThrow(message); - return response; + }); } async checkoutPrStatus( @@ -1551,8 +1639,12 @@ export class DaemonClientV2 { cwd, requestId: resolvedRequestId, }); - const response = this.waitFor( - (msg) => { + return this.sendRequest({ + requestId: resolvedRequestId, + message, + timeout: 60000, + options: { skipQueue: true }, + select: (msg) => { if (msg.type !== "checkout_pr_status_response") { return null; } @@ -1561,11 +1653,7 @@ export class DaemonClientV2 { } return msg.payload; }, - 60000, - { skipQueue: true } - ); - await this.sendSessionMessageOrThrow(message); - return response; + }); } async getPaseoWorktreeList( @@ -1579,8 +1667,12 @@ export class DaemonClientV2 { repoRoot: input.repoRoot, requestId: resolvedRequestId, }); - const response = this.waitFor( - (msg) => { + return this.sendRequest({ + requestId: resolvedRequestId, + message, + timeout: 60000, + options: { skipQueue: true }, + select: (msg) => { if (msg.type !== "paseo_worktree_list_response") { return null; } @@ -1589,11 +1681,7 @@ export class DaemonClientV2 { } return msg.payload; }, - 60000, - { skipQueue: true } - ); - await this.sendSessionMessageOrThrow(message); - return response; + }); } async archivePaseoWorktree( @@ -1608,8 +1696,12 @@ export class DaemonClientV2 { branchName: input.branchName, requestId: resolvedRequestId, }); - const response = this.waitFor( - (msg) => { + return this.sendRequest({ + requestId: resolvedRequestId, + message, + timeout: 20000, + options: { skipQueue: true }, + select: (msg) => { if (msg.type !== "paseo_worktree_archive_response") { return null; } @@ -1618,11 +1710,7 @@ export class DaemonClientV2 { } return msg.payload; }, - 20000, - { skipQueue: true } - ); - await this.sendSessionMessageOrThrow(message); - return response; + }); } async getGitDiff( @@ -1635,8 +1723,12 @@ export class DaemonClientV2 { agentId, requestId: resolvedRequestId, }); - const response = this.waitFor( - (msg) => { + return this.sendRequest({ + requestId: resolvedRequestId, + message, + timeout: 10000, + options: { skipQueue: true }, + select: (msg) => { if (msg.type !== "git_diff_response") { return null; } @@ -1645,11 +1737,7 @@ export class DaemonClientV2 { } return msg.payload; }, - 10000, - { skipQueue: true } - ); - await this.sendSessionMessageOrThrow(message); - return response; + }); } async getHighlightedDiff( @@ -1662,8 +1750,12 @@ export class DaemonClientV2 { agentId, requestId: resolvedRequestId, }); - const response = this.waitFor( - (msg) => { + return this.sendRequest({ + requestId: resolvedRequestId, + message, + timeout: 10000, + options: { skipQueue: true }, + select: (msg) => { if (msg.type !== "highlighted_diff_response") { return null; } @@ -1672,11 +1764,7 @@ export class DaemonClientV2 { } return msg.payload; }, - 10000, - { skipQueue: true } - ); - await this.sendSessionMessageOrThrow(message); - return response; + }); } async validateBranch( @@ -1690,8 +1778,12 @@ export class DaemonClientV2 { branchName: options.branchName, requestId: resolvedRequestId, }); - const response = this.waitFor( - (msg) => { + return this.sendRequest({ + requestId: resolvedRequestId, + message, + timeout: 10000, + options: { skipQueue: true }, + select: (msg) => { if (msg.type !== "validate_branch_response") { return null; } @@ -1700,11 +1792,7 @@ export class DaemonClientV2 { } return msg.payload; }, - 10000, - { skipQueue: true } - ); - await this.sendSessionMessageOrThrow(message); - return response; + }); } // ============================================================================ @@ -1725,8 +1813,12 @@ export class DaemonClientV2 { mode, requestId: resolvedRequestId, }); - const response = this.waitFor( - (msg) => { + return this.sendRequest({ + requestId: resolvedRequestId, + message, + timeout: 10000, + options: { skipQueue: true }, + select: (msg) => { if (msg.type !== "file_explorer_response") { return null; } @@ -1735,11 +1827,7 @@ export class DaemonClientV2 { } return msg.payload; }, - 10000, - { skipQueue: true } - ); - await this.sendSessionMessageOrThrow(message); - return response; + }); } async requestDownloadToken( @@ -1754,8 +1842,12 @@ export class DaemonClientV2 { path, requestId: resolvedRequestId, }); - const response = this.waitFor( - (msg) => { + return this.sendRequest({ + requestId: resolvedRequestId, + message, + timeout: 10000, + options: { skipQueue: true }, + select: (msg) => { if (msg.type !== "file_download_token_response") { return null; } @@ -1764,11 +1856,7 @@ export class DaemonClientV2 { } return msg.payload; }, - 10000, - { skipQueue: true } - ); - await this.sendSessionMessageOrThrow(message); - return response; + }); } async requestProjectIcon( @@ -1781,8 +1869,12 @@ export class DaemonClientV2 { cwd, requestId: resolvedRequestId, }); - const response = this.waitFor( - (msg) => { + return this.sendRequest({ + requestId: resolvedRequestId, + message, + timeout: 10000, + options: { skipQueue: true }, + select: (msg) => { if (msg.type !== "project_icon_response") { return null; } @@ -1791,11 +1883,7 @@ export class DaemonClientV2 { } return msg.payload; }, - 10000, - { skipQueue: true } - ); - await this.sendSessionMessageOrThrow(message); - return response; + }); } // ============================================================================ @@ -1813,8 +1901,12 @@ export class DaemonClientV2 { cwd: options?.cwd, requestId: resolvedRequestId, }); - const response = this.waitFor( - (msg) => { + return this.sendRequest({ + requestId: resolvedRequestId, + message, + timeout: 30000, + options: { skipQueue: true }, + select: (msg) => { if (msg.type !== "list_provider_models_response") { return null; } @@ -1823,11 +1915,7 @@ export class DaemonClientV2 { } return msg.payload; }, - 30000, - { skipQueue: true } - ); - await this.sendSessionMessageOrThrow(message); - return response; + }); } async listCommands( @@ -1840,8 +1928,12 @@ export class DaemonClientV2 { agentId, requestId: resolvedRequestId, }); - const response = this.waitFor( - (msg) => { + return this.sendRequest({ + requestId: resolvedRequestId, + message, + timeout: 30000, + options: { skipQueue: true }, + select: (msg) => { if (msg.type !== "list_commands_response") { return null; } @@ -1850,11 +1942,7 @@ export class DaemonClientV2 { } return msg.payload; }, - 30000, - { skipQueue: true } - ); - await this.sendSessionMessageOrThrow(message); - return response; + }); } async executeCommand( @@ -1871,8 +1959,12 @@ export class DaemonClientV2 { args, requestId: resolvedRequestId, }); - const response = this.waitFor( - (msg) => { + return this.sendRequest({ + requestId: resolvedRequestId, + message, + timeout: 30000, + options: { skipQueue: true }, + select: (msg) => { if (msg.type !== "execute_command_response") { return null; } @@ -1881,11 +1973,7 @@ export class DaemonClientV2 { } return msg.payload; }, - 30000, - { skipQueue: true } - ); - await this.sendSessionMessageOrThrow(message); - return response; + }); } // ============================================================================ @@ -1917,8 +2005,12 @@ export class DaemonClientV2 { requestId, response, }); - const resolved = this.waitFor( - (msg) => { + return this.sendRequest({ + requestId, + message, + timeout, + options: { skipQueue: true }, + select: (msg) => { if (msg.type !== "agent_permission_resolved") { return null; } @@ -1930,11 +2022,7 @@ export class DaemonClientV2 { } return msg.payload; }, - timeout, - { skipQueue: true } - ); - await this.sendSessionMessageOrThrow(message); - return resolved; + }); } // ============================================================================ @@ -1968,8 +2056,12 @@ export class DaemonClientV2 { agentId, timeoutMs: timeout, }); - const response = this.waitFor( - (msg) => { + const payload = await this.sendRequest({ + requestId, + message, + timeout: timeout + 5000, + options: { skipQueue: true }, + select: (msg) => { if (msg.type !== "wait_for_finish_response") { return null; } @@ -1978,11 +2070,7 @@ export class DaemonClientV2 { } return msg.payload; }, - timeout + 5000, - { skipQueue: true } - ); - await this.sendSessionMessageOrThrow(message); - const payload = await response; + }); return { status: payload.status, final: payload.final, @@ -2004,8 +2092,12 @@ export class DaemonClientV2 { cwd, requestId: resolvedRequestId, }); - const response = this.waitFor( - (msg) => { + return this.sendRequest({ + requestId: resolvedRequestId, + message, + timeout: 10000, + options: { skipQueue: true }, + select: (msg) => { if (msg.type !== "list_terminals_response") { return null; } @@ -2014,11 +2106,7 @@ export class DaemonClientV2 { } return msg.payload; }, - 10000, - { skipQueue: true } - ); - await this.sendSessionMessageOrThrow(message); - return response; + }); } async createTerminal( @@ -2033,8 +2121,12 @@ export class DaemonClientV2 { name, requestId: resolvedRequestId, }); - const response = this.waitFor( - (msg) => { + return this.sendRequest({ + requestId: resolvedRequestId, + message, + timeout: 10000, + options: { skipQueue: true }, + select: (msg) => { if (msg.type !== "create_terminal_response") { return null; } @@ -2043,11 +2135,7 @@ export class DaemonClientV2 { } return msg.payload; }, - 10000, - { skipQueue: true } - ); - await this.sendSessionMessageOrThrow(message); - return response; + }); } async subscribeTerminal( @@ -2060,8 +2148,12 @@ export class DaemonClientV2 { terminalId, requestId: resolvedRequestId, }); - const response = this.waitFor( - (msg) => { + return this.sendRequest({ + requestId: resolvedRequestId, + message, + timeout: 10000, + options: { skipQueue: true }, + select: (msg) => { if (msg.type !== "subscribe_terminal_response") { return null; } @@ -2070,11 +2162,7 @@ export class DaemonClientV2 { } return msg.payload; }, - 10000, - { skipQueue: true } - ); - await this.sendSessionMessageOrThrow(message); - return response; + }); } unsubscribeTerminal(terminalId: string): void { @@ -2105,8 +2193,12 @@ export class DaemonClientV2 { terminalId, requestId: resolvedRequestId, }); - const response = this.waitFor( - (msg) => { + return this.sendRequest({ + requestId: resolvedRequestId, + message, + timeout: 10000, + options: { skipQueue: true }, + select: (msg) => { if (msg.type !== "kill_terminal_response") { return null; } @@ -2115,11 +2207,7 @@ export class DaemonClientV2 { } return msg.payload; }, - 10000, - { skipQueue: true } - ); - await this.sendSessionMessageOrThrow(message); - return response; + }); } async waitForTerminalOutput( @@ -2349,26 +2437,79 @@ export class DaemonClientV2 { timeout = 30000, _options?: { skipQueue?: boolean } ): Promise { + return this.waitForWithCancel(predicate, timeout, _options).promise; + } + + private waitForWithCancel( + predicate: (msg: SessionOutboundMessage) => T | null, + timeout = 30000, + _options?: { skipQueue?: boolean } + ): WaitHandle { // Capture stack trace at call site, not inside setTimeout const timeoutError = new Error(`Timeout waiting for message (${timeout}ms)`); - return new Promise((resolve, reject) => { + let waiter: Waiter | null = null; + let settled = false; + let rejectFn: ((error: Error) => void) | null = null; + + const promise = new Promise((resolve, reject) => { + const wrappedResolve = (value: T) => { + if (settled) return; + settled = true; + resolve(value); + }; + const wrappedReject = (error: Error) => { + if (settled) return; + settled = true; + reject(error); + }; + rejectFn = wrappedReject; + const timeoutHandle = timeout > 0 ? setTimeout(() => { - this.waiters.delete(waiter); - reject(timeoutError); + if (waiter) { + this.waiters.delete(waiter); + } + wrappedReject(timeoutError); }, timeout) : null; - const waiter: Waiter = { + waiter = { predicate, - resolve, - reject, + resolve: wrappedResolve, + reject: wrappedReject, timeoutHandle, }; this.waiters.add(waiter); }); + + const cancel = (error: Error) => { + if (settled) { + return; + } + + if (waiter) { + this.waiters.delete(waiter); + if (waiter.timeoutHandle) { + clearTimeout(waiter.timeoutHandle); + } + } + + if (rejectFn) { + rejectFn(error); + return; + } + + // Extremely unlikely: cancel called before the Promise executor ran. + queueMicrotask(() => { + if (!settled && rejectFn) { + rejectFn(error); + } + }); + }; + + return { promise, cancel }; } } diff --git a/packages/server/src/server/agent/providers/codex-app-server-agent.test.ts b/packages/server/src/server/agent/providers/codex-app-server-agent.test.ts index 2483bb4a6..6f71e4f43 100644 --- a/packages/server/src/server/agent/providers/codex-app-server-agent.test.ts +++ b/packages/server/src/server/agent/providers/codex-app-server-agent.test.ts @@ -326,7 +326,7 @@ describe("Codex app-server provider (integration)", () => { "*** Add File: patch.txt", "+patched", "*** End Patch", - ].join("\\n"); + ].join("\n"); const patchEvents = session.stream( [ "Use the apply_patch tool and nothing else.", @@ -335,7 +335,7 @@ describe("Codex app-server provider (integration)", () => { "Apply the following patch exactly:", patch, "After it completes, reply PATCH_DONE.", - ].join("\\n") + ].join("\n") ); for await (const event of patchEvents) { @@ -612,7 +612,7 @@ describe("Codex app-server provider (integration)", () => { "*** Add File: approval-test.txt", "+ok", "*** End Patch", - ].join("\\n"); + ].join("\n"); const events = session.stream( [ "Use the apply_patch tool and nothing else.", @@ -621,7 +621,7 @@ describe("Codex app-server provider (integration)", () => { "Apply the following patch exactly:", patch, "After approval, reply FILE_DONE.", - ].join("\\n") + ].join("\n") ); let failure: string | null = null; diff --git a/packages/server/src/server/daemon-client-v2.e2e.test.ts b/packages/server/src/server/daemon-client.e2e.test.ts similarity index 99% rename from packages/server/src/server/daemon-client-v2.e2e.test.ts rename to packages/server/src/server/daemon-client.e2e.test.ts index 82314baa1..d8d93550a 100644 --- a/packages/server/src/server/daemon-client-v2.e2e.test.ts +++ b/packages/server/src/server/daemon-client.e2e.test.ts @@ -18,7 +18,7 @@ import { } from "./test-utils/dictation-e2e.js"; function tmpCwd(): string { - return mkdtempSync(path.join(tmpdir(), "daemon-client-v2-")); + return mkdtempSync(path.join(tmpdir(), "daemon-client-")); } function waitForSignal( @@ -52,7 +52,7 @@ function waitForSignal( }); } -describe("daemon client v2 E2E", () => { +describe("daemon client E2E", () => { let ctx: DaemonTestContext; beforeAll(async () => { diff --git a/packages/server/src/server/daemon-e2e/checkout-debug.ts b/packages/server/src/server/daemon-e2e/checkout-debug.ts index 60cb3d9e6..1fd0230fa 100644 --- a/packages/server/src/server/daemon-e2e/checkout-debug.ts +++ b/packages/server/src/server/daemon-e2e/checkout-debug.ts @@ -10,7 +10,7 @@ */ import { WebSocket } from "ws"; -import { DaemonClientV2 } from "../../client/daemon-client-v2.js"; +import { DaemonClient } from "../../client/daemon-client.js"; // Patch WebSocket to log all messages const OriginalWebSocket = WebSocket; @@ -36,7 +36,7 @@ async function testMultiAgentSequence() { console.log("\n=== Testing multi-agent checkout sequence ==="); console.log(`Daemon URL: ${DAEMON_URL}`); - const client = new DaemonClientV2({ + const client = new DaemonClient({ url: DAEMON_URL, webSocketFactory: (url) => new LoggingWebSocket(url) as any, reconnect: { enabled: false }, diff --git a/packages/server/src/server/exports.ts b/packages/server/src/server/exports.ts index 02b9c533b..674525f92 100644 --- a/packages/server/src/server/exports.ts +++ b/packages/server/src/server/exports.ts @@ -4,7 +4,7 @@ export { loadConfig, type CliConfigOverrides } from "./config.js"; export { resolvePaseoHome } from "./paseo-home.js"; export { createRootLogger, type LogLevel, type LogFormat } from "./logger.js"; export { loadPersistedConfig, type PersistedConfig } from "./persisted-config.js"; -export { DaemonClientV2, type DaemonClientV2Config, type ConnectionState, type DaemonEvent } from "../client/daemon-client-v2.js"; +export { DaemonClient, type DaemonClientConfig, type ConnectionState, type DaemonEvent } from "../client/daemon-client.js"; // Agent SDK types for CLI commands export type { diff --git a/packages/server/src/server/session.ts b/packages/server/src/server/session.ts index 4474912aa..90bc8aa60 100644 --- a/packages/server/src/server/session.ts +++ b/packages/server/src/server/session.ts @@ -1023,17 +1023,36 @@ export class Session { break; } } catch (error: any) { + const err = error instanceof Error ? error : new Error(String(error)); this.sessionLogger.error( - { err: error }, + { err }, "Error handling message" ); + + const requestId = (msg as { requestId?: unknown }).requestId; + if (typeof requestId === "string") { + try { + this.emit({ + type: "rpc_error", + payload: { + requestId, + requestType: msg.type, + error: "Request failed", + code: "handler_error", + }, + }); + } catch (emitError) { + this.sessionLogger.error({ err: emitError }, "Failed to emit rpc_error"); + } + } + this.emit({ type: "activity_log", payload: { id: uuidv4(), timestamp: new Date(), type: "error", - content: `Error: ${error.message}`, + content: `Error: ${err.message}`, }, }); } @@ -1976,7 +1995,11 @@ export class Session { } private async generateCommitMessage(cwd: string): Promise { - const diff = await getCheckoutDiff(cwd, { mode: "uncommitted" }, { paseoHome: this.paseoHome }); + const diff = await getCheckoutDiff( + cwd, + { mode: "uncommitted", includeStructured: true }, + { paseoHome: this.paseoHome } + ); const schema = z.object({ message: z .string() @@ -1984,11 +2007,29 @@ export class Session { .max(72) .describe("Concise git commit message, imperative mood, no trailing period."), }); + const fileList = + diff.structured && diff.structured.length > 0 + ? [ + "Files changed:", + ...diff.structured.map((file) => { + const changeType = file.isNew ? "A" : file.isDeleted ? "D" : "M"; + const status = file.status && file.status !== "ok" ? ` [${file.status}]` : ""; + return `${changeType}\t${file.path}\t(+${file.additions} -${file.deletions})${status}`; + }), + ].join("\n") + : "Files changed: (unknown)"; + const maxPatchChars = 120_000; + const patch = + diff.diff.length > maxPatchChars + ? `${diff.diff.slice(0, maxPatchChars)}\n\n... (diff truncated to ${maxPatchChars} chars)\n` + : diff.diff; const prompt = [ "Write a concise git commit message for the changes below.", "Return JSON only with a single field 'message'.", "", - diff.diff.length > 0 ? diff.diff : "(No diff available)", + fileList, + "", + patch.length > 0 ? patch : "(No diff available)", ].join("\n"); try { const result = await generateStructuredAgentResponse({ @@ -2023,6 +2064,7 @@ export class Session { { mode: "base", baseRef, + includeStructured: true, }, { paseoHome: this.paseoHome } ); @@ -2030,11 +2072,29 @@ export class Session { title: z.string().min(1).max(72), body: z.string().min(1), }); + const fileList = + diff.structured && diff.structured.length > 0 + ? [ + "Files changed:", + ...diff.structured.map((file) => { + const changeType = file.isNew ? "A" : file.isDeleted ? "D" : "M"; + const status = file.status && file.status !== "ok" ? ` [${file.status}]` : ""; + return `${changeType}\t${file.path}\t(+${file.additions} -${file.deletions})${status}`; + }), + ].join("\n") + : "Files changed: (unknown)"; + const maxPatchChars = 200_000; + const patch = + diff.diff.length > maxPatchChars + ? `${diff.diff.slice(0, maxPatchChars)}\n\n... (diff truncated to ${maxPatchChars} chars)\n` + : diff.diff; const prompt = [ "Write a pull request title and body for the changes below.", "Return JSON only with fields 'title' and 'body'.", "", - diff.diff.length > 0 ? diff.diff : "(No diff available)", + fileList, + "", + patch.length > 0 ? patch : "(No diff available)", ].join("\n"); try { return await generateStructuredAgentResponse({ diff --git a/packages/server/src/server/test-utils/daemon-client.ts b/packages/server/src/server/test-utils/daemon-client.ts index 7270ea3e0..6aa3613aa 100644 --- a/packages/server/src/server/test-utils/daemon-client.ts +++ b/packages/server/src/server/test-utils/daemon-client.ts @@ -1,13 +1,13 @@ import WebSocket from "ws"; import { - DaemonClientV2 as SharedDaemonClient, - type DaemonClientV2Config as SharedDaemonClientConfig, + DaemonClient as SharedDaemonClient, + type DaemonClientConfig as SharedDaemonClientConfig, type CreateAgentRequestOptions, type DaemonEvent, type DaemonEventHandler, type SendMessageOptions, type WebSocketLike, -} from "../../client/daemon-client-v2.js"; +} from "../../client/daemon-client.js"; export type DaemonClientConfig = Omit< SharedDaemonClientConfig, diff --git a/packages/server/src/server/test-utils/message-collector.ts b/packages/server/src/server/test-utils/message-collector.ts index 2b039df97..0b3a2ef8b 100644 --- a/packages/server/src/server/test-utils/message-collector.ts +++ b/packages/server/src/server/test-utils/message-collector.ts @@ -1,4 +1,4 @@ -import type { DaemonClientV2 } from "../../client/daemon-client-v2.js"; +import type { DaemonClient } from "../../client/daemon-client.js"; import type { SessionOutboundMessage } from "../../shared/messages.js"; export interface MessageCollector { @@ -7,7 +7,7 @@ export interface MessageCollector { unsubscribe: () => void; } -export function createMessageCollector(client: DaemonClientV2): MessageCollector { +export function createMessageCollector(client: DaemonClient): MessageCollector { const messages: SessionOutboundMessage[] = []; const unsubscribe = client.subscribeRawMessages((message) => { messages.push(message); @@ -20,4 +20,3 @@ export function createMessageCollector(client: DaemonClientV2): MessageCollector unsubscribe, }; } - diff --git a/packages/server/src/server/websocket-session-bridge.ts b/packages/server/src/server/websocket-session-bridge.ts index 0219311b3..f9eae2489 100644 --- a/packages/server/src/server/websocket-session-bridge.ts +++ b/packages/server/src/server/websocket-session-bridge.ts @@ -157,7 +157,56 @@ export class WebSocketSessionBridge { private async handleRawMessage(ws: WebSocket, data: Buffer | ArrayBuffer | Buffer[]): Promise { try { const parsed = JSON.parse(data.toString()); - const message = WSInboundMessageSchema.parse(parsed); + const parsedMessage = WSInboundMessageSchema.safeParse(parsed); + if (!parsedMessage.success) { + const requestInfo = extractRequestInfoFromUnknownWsInbound(parsed); + const isUnknownSchema = + requestInfo?.requestId != null && + typeof parsed === "object" && + parsed != null && + "type" in parsed && + (parsed as { type?: unknown }).type === "session"; + + this.logger.warn( + { + requestId: requestInfo?.requestId, + requestType: requestInfo?.requestType, + error: parsedMessage.error.message, + }, + "WS inbound message validation failed" + ); + + if (requestInfo) { + this.sendToClient( + ws, + wrapSessionMessage({ + type: "rpc_error", + payload: { + requestId: requestInfo.requestId, + requestType: requestInfo.requestType, + error: isUnknownSchema ? "Unknown request schema" : "Invalid message", + code: isUnknownSchema ? "unknown_schema" : "invalid_message", + }, + }) + ); + return; + } + + const errorMessage = `Invalid message: ${parsedMessage.error.message}`; + this.sendToClient( + ws, + wrapSessionMessage({ + type: "status", + payload: { + status: "error", + message: errorMessage, + }, + }) + ); + return; + } + + const message = parsedMessage.data; const messageSummary = { type: message.type, @@ -234,6 +283,23 @@ export class WebSocketSessionBridge { "Failed to parse/handle message" ); + const requestInfo = extractRequestInfoFromUnknownWsInbound(parsedPayload); + if (requestInfo) { + this.sendToClient( + ws, + wrapSessionMessage({ + type: "rpc_error", + payload: { + requestId: requestInfo.requestId, + requestType: requestInfo.requestType, + error: "Invalid message", + code: "invalid_message", + }, + }) + ); + return; + } + this.sendToClient( ws, wrapSessionMessage({ @@ -466,3 +532,38 @@ export class WebSocketSessionBridge { } } } + +function extractRequestInfoFromUnknownWsInbound( + payload: unknown +): { requestId: string; requestType?: string } | null { + if (!payload || typeof payload !== "object") { + return null; + } + + const record = payload as { + type?: unknown; + requestId?: unknown; + message?: unknown; + }; + + // Session-wrapped messages + if (record.type === "session" && record.message && typeof record.message === "object") { + const msg = record.message as { requestId?: unknown; type?: unknown }; + if (typeof msg.requestId === "string") { + return { + requestId: msg.requestId, + ...(typeof msg.type === "string" ? { requestType: msg.type } : {}), + }; + } + } + + // Non-session messages (future-proof) + if (typeof record.requestId === "string") { + return { + requestId: record.requestId, + ...(typeof record.type === "string" ? { requestType: record.type } : {}), + }; + } + + return null; +} diff --git a/packages/server/src/shared/messages.ts b/packages/server/src/shared/messages.ts index 22118fa55..e3006d31b 100644 --- a/packages/server/src/shared/messages.ts +++ b/packages/server/src/shared/messages.ts @@ -1024,6 +1024,16 @@ export const StatusMessageSchema = z.object({ .passthrough(), // Allow additional fields }); +export const RpcErrorMessageSchema = z.object({ + type: z.literal("rpc_error"), + payload: z.object({ + requestId: z.string(), + requestType: z.string().optional(), + error: z.string(), + code: z.string().optional(), + }), +}); + const AgentStatusWithRequestSchema = z.object({ agentId: z.string(), requestId: z.string(), @@ -1607,6 +1617,7 @@ export const SessionOutboundMessageSchema = z.discriminatedUnion("type", [ DictationStreamFinalMessageSchema, DictationStreamErrorMessageSchema, StatusMessageSchema, + RpcErrorMessageSchema, InitializeAgentResponseMessageSchema, ArtifactMessageSchema, VoiceConversationLoadedMessageSchema, @@ -1663,6 +1674,7 @@ export type AssistantChunkMessage = z.infer; export type AudioOutputMessage = z.infer; export type TranscriptionResultMessage = z.infer; export type StatusMessage = z.infer; +export type RpcErrorMessage = z.infer; export type ArtifactMessage = z.infer; export type VoiceConversationLoadedMessage = z.infer< typeof VoiceConversationLoadedMessageSchema diff --git a/packages/server/src/utils/checkout-git.test.ts b/packages/server/src/utils/checkout-git.test.ts index 7d30a074e..e5e82dbbd 100644 --- a/packages/server/src/utils/checkout-git.test.ts +++ b/packages/server/src/utils/checkout-git.test.ts @@ -102,6 +102,36 @@ describe("checkout git utilities", () => { expect(logMessage).toBe(message); }); + it("diffs base mode against merge-base (no base-only deletions)", async () => { + execSync("git checkout -b feature", { cwd: repoDir }); + + // Advance base branch after feature splits off. + execSync("git checkout main", { cwd: repoDir }); + writeFileSync(join(repoDir, "base-only.txt"), "base\n"); + execSync("git add base-only.txt", { cwd: repoDir }); + execSync("git -c commit.gpgsign=false commit -m 'base only'", { cwd: repoDir }); + + // Make a feature change. + execSync("git checkout feature", { cwd: repoDir }); + writeFileSync(join(repoDir, "feature.txt"), "feature\n"); + execSync("git add feature.txt", { cwd: repoDir }); + execSync("git -c commit.gpgsign=false commit -m 'feature commit'", { cwd: repoDir }); + + const diff = await getCheckoutDiff(repoDir, { mode: "base", baseRef: "main" }); + expect(diff.diff).toContain("feature.txt"); + expect(diff.diff).not.toContain("base-only.txt"); + }); + + it("does not throw on large diffs (marks file as too_large)", async () => { + const large = Array.from({ length: 200_000 }, (_, i) => `line ${i}`).join("\n") + "\n"; + writeFileSync(join(repoDir, "file.txt"), large); + + const diff = await getCheckoutDiff(repoDir, { mode: "uncommitted", includeStructured: true }); + expect(diff.structured?.some((f) => f.path === "file.txt" && f.status === "too_large")).toBe( + true + ); + }); + it("handles status/diff/commit in a .paseo worktree", async () => { const result = await createWorktree({ branchName: "main", diff --git a/packages/server/src/utils/checkout-git.ts b/packages/server/src/utils/checkout-git.ts index 4488b0c95..f40cc1086 100644 --- a/packages/server/src/utils/checkout-git.ts +++ b/packages/server/src/utils/checkout-git.ts @@ -1,4 +1,4 @@ -import { exec, execFile } from "child_process"; +import { exec, execFile, spawn } from "child_process"; import { promisify } from "util"; import { resolve, dirname, basename } from "path"; import { realpathSync } from "fs"; @@ -14,6 +14,203 @@ const READ_ONLY_GIT_ENV: NodeJS.ProcessEnv = { GIT_OPTIONAL_LOCKS: "0", }; +const SMALL_OUTPUT_MAX_BUFFER = 20 * 1024 * 1024; // 20MB + +async function execGit(command: string, options: { cwd: string; env?: NodeJS.ProcessEnv }): Promise<{ stdout: string; stderr: string }> { + return execAsync(command, { ...options, maxBuffer: SMALL_OUTPUT_MAX_BUFFER }); +} + +async function execGitFile( + args: string[], + options: { cwd: string; env?: NodeJS.ProcessEnv } +): Promise<{ stdout: string; stderr: string }> { + return execFileAsync("git", args, { ...options, maxBuffer: SMALL_OUTPUT_MAX_BUFFER }); +} + +type LimitedTextResult = { + text: string; + truncated: boolean; + exitCode: number | null; + signal: NodeJS.Signals | null; +}; + +async function spawnLimitedText(params: { + cmd: string; + args: string[]; + cwd: string; + env?: NodeJS.ProcessEnv; + maxBytes: number; + acceptExitCodes?: number[]; +}): Promise { + const accept = new Set(params.acceptExitCodes ?? [0]); + + return new Promise((resolvePromise, rejectPromise) => { + const child = spawn(params.cmd, params.args, { + cwd: params.cwd, + env: params.env, + stdio: ["ignore", "pipe", "pipe"], + }); + + const stdoutChunks: Buffer[] = []; + let stdoutBytes = 0; + let truncated = false; + + const stop = () => { + if (child.killed) return; + try { + child.kill("SIGKILL"); + } catch { + // ignore + } + }; + + child.stdout.on("data", (chunk: Buffer) => { + if (truncated) return; + stdoutBytes += chunk.length; + if (stdoutBytes > params.maxBytes) { + truncated = true; + stop(); + return; + } + stdoutChunks.push(chunk); + }); + + // We don't buffer stderr (it can be large too). Keep it minimal for debugging. + let stderrPreview = ""; + child.stderr.on("data", (chunk: Buffer) => { + if (stderrPreview.length > 2048) return; + stderrPreview += chunk.toString("utf8"); + }); + + child.on("error", (error) => { + rejectPromise(error); + }); + + child.on("close", (code, signal) => { + if (code !== null && !accept.has(code) && !truncated) { + rejectPromise(new Error(`Command failed: ${params.cmd} ${params.args.join(" ")} (code ${code})\n${stderrPreview}`)); + return; + } + resolvePromise({ + text: Buffer.concat(stdoutChunks).toString("utf8"), + truncated, + exitCode: code, + signal, + }); + }); + }); +} + +type CheckoutFileChange = { + path: string; + oldPath?: string; + status: string; + isNew: boolean; + isDeleted: boolean; + isUntracked?: boolean; +}; + +async function listCheckoutFileChanges(cwd: string, ref: string): Promise { + const changes: CheckoutFileChange[] = []; + + const { stdout: nameStatusOut } = await execGit(`git diff --name-status ${ref}`, { + cwd, + env: READ_ONLY_GIT_ENV, + }); + for (const line of nameStatusOut.split("\n").map((l) => l.trim()).filter(Boolean)) { + // `--name-status` uses TAB separators, which preserves filenames with spaces. + const tabParts = line.split("\t"); + const rawStatus = (tabParts[0] ?? "").trim(); + if (!rawStatus) continue; + + if (rawStatus.startsWith("R") || rawStatus.startsWith("C")) { + const oldPath = tabParts[1]; + const newPath = tabParts[2]; + if (newPath) { + changes.push({ + path: newPath, + ...(oldPath ? { oldPath } : {}), + status: rawStatus, + isNew: false, + isDeleted: false, + }); + } + continue; + } + + const path = tabParts[1]; + if (!path) continue; + const code = rawStatus[0]; + changes.push({ + path, + status: rawStatus, + isNew: code === "A", + isDeleted: code === "D", + }); + } + + const { stdout: untrackedOut } = await execGit("git ls-files --others --exclude-standard", { + cwd, + env: READ_ONLY_GIT_ENV, + }); + for (const file of untrackedOut.split("\n").map((l) => l.trim()).filter(Boolean)) { + changes.push({ + path: file, + status: "U", + isNew: true, + isDeleted: false, + isUntracked: true, + }); + } + + // Deduplicate by path (prefer tracked status over untracked marker if both appear). + const byPath = new Map(); + for (const change of changes) { + const existing = byPath.get(change.path); + if (!existing) { + byPath.set(change.path, change); + continue; + } + if (existing.isUntracked && !change.isUntracked) { + byPath.set(change.path, change); + } + } + return Array.from(byPath.values()); +} + +async function tryResolveMergeBase(cwd: string, baseRef: string): Promise { + try { + const { stdout } = await execGit(`git merge-base ${baseRef} HEAD`, { cwd, env: READ_ONLY_GIT_ENV }); + const sha = stdout.trim(); + return sha.length > 0 ? sha : null; + } catch { + return null; + } +} + +type FileStat = { additions: number; deletions: number; isBinary: boolean } | null; + +async function tryGetNumstat(cwd: string, args: string[]): Promise { + try { + const { stdout } = await execGitFile(args, { cwd, env: READ_ONLY_GIT_ENV }); + const line = stdout.trim().split("\n").map((l) => l.trim()).filter(Boolean)[0] ?? ""; + if (!line) return null; + const [aRaw, dRaw] = line.split(/\s+/); + if (!aRaw || !dRaw) return null; + if (aRaw === "-" || dRaw === "-") { + return { additions: 0, deletions: 0, isBinary: true }; + } + const additions = Number.parseInt(aRaw, 10); + const deletions = Number.parseInt(dRaw, 10); + if (Number.isNaN(additions) || Number.isNaN(deletions)) { + return null; + } + return { additions, deletions, isBinary: false }; + } catch { + return null; + } +} + export class NotGitRepoError extends Error { readonly cwd: string; readonly code = "NOT_GIT_REPO"; @@ -435,32 +632,58 @@ async function getAheadOfOrigin(cwd: string, currentBranch: string): Promise { - let untrackedDiff = ""; - try { - const { stdout: untrackedFiles } = await execAsync( - "git ls-files --others --exclude-standard", - { cwd, env: READ_ONLY_GIT_ENV } - ); - const newFiles = untrackedFiles.trim().split("\n").filter(Boolean); +const PER_FILE_DIFF_MAX_BYTES = 1024 * 1024; // 1MB +const TOTAL_DIFF_MAX_BYTES = 2 * 1024 * 1024; // 2MB - for (const file of newFiles) { - try { - const { stdout: fileDiff } = await execAsync( - `git diff --no-index /dev/null "${file}" || true`, - { cwd, env: READ_ONLY_GIT_ENV } - ); - if (fileDiff) { - untrackedDiff += fileDiff; - } - } catch { - // Ignore errors for individual files - } - } - } catch { - // Ignore errors getting untracked files +function buildPlaceholderParsedDiffFile( + change: CheckoutFileChange, + options: { status: "too_large" | "binary"; stat?: FileStat } +): ParsedDiffFile { + return { + path: change.path, + isNew: change.isNew, + isDeleted: change.isDeleted, + additions: options.stat?.additions ?? 0, + deletions: options.stat?.deletions ?? 0, + hunks: [], + status: options.status, + }; +} + +async function getPerFileDiffText( + cwd: string, + ref: string, + change: CheckoutFileChange +): Promise<{ text: string; truncated: boolean; stat: FileStat }> { + const stat: FileStat = + change.isUntracked + ? null + : await tryGetNumstat(cwd, ["diff", "--numstat", ref, "--", change.path]); + + if (stat?.isBinary) { + return { text: "", truncated: false, stat }; } - return untrackedDiff; + + if (change.isUntracked) { + const result = await spawnLimitedText({ + cmd: "git", + args: ["diff", "--no-index", "/dev/null", "--", change.path], + cwd, + env: READ_ONLY_GIT_ENV, + maxBytes: PER_FILE_DIFF_MAX_BYTES, + acceptExitCodes: [0, 1], + }); + return { text: result.text, truncated: result.truncated, stat }; + } + + const result = await spawnLimitedText({ + cmd: "git", + args: ["diff", ref, "--", change.path], + cwd, + env: READ_ONLY_GIT_ENV, + maxBytes: PER_FILE_DIFF_MAX_BYTES, + }); + return { text: result.text, truncated: result.truncated, stat }; } export async function getCheckoutStatus( @@ -530,43 +753,102 @@ export async function getCheckoutDiff( ): Promise { await requireGitRepo(cwd); - let diff = ""; + let refForDiff: string; + if (compare.mode === "uncommitted") { - const { stdout: trackedDiff } = await execAsync("git diff HEAD", { - cwd, - env: READ_ONLY_GIT_ENV, - }); - const untrackedDiff = await getUntrackedDiff(cwd); - diff = trackedDiff + untrackedDiff; + refForDiff = "HEAD"; } else { const configured = await getConfiguredBaseRefForCwd(cwd, context); const baseRef = configured.baseRef ?? compare.baseRef ?? (await resolveBaseRef(cwd)); if (!baseRef) { - diff = ""; - } else if (configured.isPaseoOwnedWorktree && compare.baseRef && compare.baseRef !== baseRef) { - throw new Error(`Base ref mismatch: expected ${baseRef}, got ${compare.baseRef}`); - } else { - const normalizedBaseRef = normalizeLocalBranchRefName(baseRef); - // Find the merge-base (common ancestor) to diff only changes on this branch - const { stdout: mergeBaseOut } = await execAsync( - `git merge-base ${normalizedBaseRef} HEAD`, - { cwd, env: READ_ONLY_GIT_ENV } - ); - const mergeBase = mergeBaseOut.trim(); - // Diff from merge-base to working tree (includes uncommitted changes) - const { stdout: trackedDiff } = await execAsync(`git diff ${mergeBase}`, { - cwd, - env: READ_ONLY_GIT_ENV, - }); - const untrackedDiff = await getUntrackedDiff(cwd); - diff = trackedDiff + untrackedDiff; + return { diff: "" }; } + if (configured.isPaseoOwnedWorktree && compare.baseRef && compare.baseRef !== baseRef) { + throw new Error(`Base ref mismatch: expected ${baseRef}, got ${compare.baseRef}`); + } + + const normalizedBaseRef = normalizeLocalBranchRefName(baseRef); + const bestBaseRef = await resolveBestBaseRefForMerge(cwd, normalizedBaseRef); + refForDiff = (await tryResolveMergeBase(cwd, bestBaseRef)) ?? bestBaseRef; + } + + const changes = await listCheckoutFileChanges(cwd, refForDiff); + changes.sort((a, b) => a.path.localeCompare(b.path)); + + const structured: ParsedDiffFile[] = []; + let diffText = ""; + let diffBytes = 0; + const appendDiff = (text: string) => { + if (!text) return; + if (diffBytes >= TOTAL_DIFF_MAX_BYTES) return; + const buf = Buffer.from(text, "utf8"); + if (diffBytes + buf.length <= TOTAL_DIFF_MAX_BYTES) { + diffText += text; + diffBytes += buf.length; + return; + } + const remaining = TOTAL_DIFF_MAX_BYTES - diffBytes; + if (remaining > 0) { + diffText += buf.subarray(0, remaining).toString("utf8"); + diffBytes = TOTAL_DIFF_MAX_BYTES; + } + }; + + for (const change of changes) { + const { text, truncated, stat } = await getPerFileDiffText(cwd, refForDiff, change); + + if (!compare.includeStructured) { + if (stat?.isBinary) { + appendDiff(`# ${change.path}: binary diff omitted\n`); + } else if (truncated) { + appendDiff(`# ${change.path}: diff too large omitted\n`); + } else { + appendDiff(text); + } + if (diffBytes >= TOTAL_DIFF_MAX_BYTES) { + break; + } + continue; + } + + if (stat?.isBinary) { + structured.push(buildPlaceholderParsedDiffFile(change, { status: "binary", stat })); + appendDiff(`# ${change.path}: binary diff omitted\n`); + continue; + } + + if (truncated) { + structured.push(buildPlaceholderParsedDiffFile(change, { status: "too_large", stat })); + appendDiff(`# ${change.path}: diff too large omitted\n`); + continue; + } + + appendDiff(text); + const parsed = await parseAndHighlightDiff(text, cwd); + const parsedFile = + parsed[0] ?? + ({ + path: change.path, + isNew: change.isNew, + isDeleted: change.isDeleted, + additions: stat?.additions ?? 0, + deletions: stat?.deletions ?? 0, + hunks: [], + } satisfies ParsedDiffFile); + + structured.push({ + ...parsedFile, + path: change.path, + isNew: change.isNew, + isDeleted: change.isDeleted, + status: "ok", + }); } if (compare.includeStructured) { - return { diff, structured: await parseAndHighlightDiff(diff, cwd) }; + return { diff: diffText, structured }; } - return { diff }; + return { diff: diffText }; } export async function commitChanges(