diff --git a/sdk/ARCHITECTURE.md b/sdk/ARCHITECTURE.md index 0a057eb486..73b71d1a8d 100644 --- a/sdk/ARCHITECTURE.md +++ b/sdk/ARCHITECTURE.md @@ -209,6 +209,19 @@ workspace-wide operations on busy sessions (for example the desktop's checkpoint-restore gate) depend on this: a defaulted `running` with no owning turn leaves such gates blocked with nothing to clear them. +Starting a session that already has a run in flight joins it rather than +replacing it: `LocalRuntimeHost.startSession` returns the live session, and a +start that also carries a prompt is rejected so the caller queues it instead. +The running agent keeps the client-owned capabilities (hooks, tool executors, +the mistake-limit prompt) it was built with, but the hub records every client +that registers each capability for a session, joiners included, and +`requestCapability` sends each request to the original owner while it is +connected, otherwise to the first registered client that is still connected +and attached to the session. A request with no such client fails immediately +instead of waiting, so a run whose app has gone away ends with an error the +user can continue from. Detaching or unregistering also cancels that client's +pending requests. + Command progress follows the same runtime event boundary as other agent output. Shell executors emit structured stdout/stderr chunks through `AgentToolContext.emitUpdate`; the agent runtime projects them as tool diff --git a/sdk/packages/core/src/hub/server/boundary.test.ts b/sdk/packages/core/src/hub/server/boundary.test.ts index cd08f77a6f..a414a37488 100644 --- a/sdk/packages/core/src/hub/server/boundary.test.ts +++ b/sdk/packages/core/src/hub/server/boundary.test.ts @@ -16,6 +16,10 @@ import { handleApprovalRespond, requestToolApproval, } from "./handlers/approval-handlers"; +import { + recordCapabilityClients, + requestCapability, +} from "./handlers/capability-handlers"; import { ensureSessionParticipant, ensureSessionState, @@ -1039,6 +1043,12 @@ describe("HubServerTransport boundaries", () => { const events: HubEventEnvelope[] = []; transport.subscribe("client-1", (event) => events.push(event)); + await transport.handleCommand({ + version: "v1", + requestId: "req-register", + command: "client.register", + clientId: "client-1", + }); const reply = await transport.handleCommand({ version: "v1", requestId: "req-create", @@ -1171,6 +1181,12 @@ describe("HubServerTransport boundaries", () => { const events: HubEventEnvelope[] = []; transport.subscribe("owner-client", (event) => events.push(event)); + await transport.handleCommand({ + version: "v1", + requestId: "req-register", + command: "client.register", + clientId: "owner-client", + }); await transport.handleCommand({ version: "v1", requestId: "req-create", @@ -1733,6 +1749,120 @@ describe("HubServerTransport boundaries", () => { ); }); + it("fails capability requests to a client that already unregistered", async () => { + const transport = createTransport(); + const ctx = getContext(transport); + const register = (clientId: string) => + transport.handleCommand({ + version: "v1", + requestId: `req-register-${clientId}`, + command: "client.register", + clientId, + payload: { clientId, clientType: "desktop" }, + }); + await register("owner-client"); + await transport.handleCommand({ + version: "v1", + requestId: "req-unregister", + command: "client.unregister", + clientId: "owner-client", + }); + + // A run still in flight after its client restarted raises its next hook: + // it must fail instead of waiting on a client that can never answer. + await expect( + requestCapability( + ctx, + "session-1", + "hook.beforeTool", + {}, + "owner-client", + ), + ).rejects.toThrow(/disconnected before it could respond/); + expect(ctx.pendingCapabilityRequests.size).toBe(0); + + await register("live-client"); + void requestCapability( + ctx, + "session-1", + "hook.beforeTool", + {}, + "live-client", + ); + expect(ctx.pendingCapabilityRequests.size).toBe(1); + }); + + it("sends a capability request to a client that joined the task when the owner is gone", async () => { + const transport = createTransport(); + const ctx = getContext(transport); + // Registered in this order; "left-client" detaches from the task later. + for (const clientId of ["owner-client", "left-client", "joined-client"]) { + await transport.handleCommand({ + version: "v1", + requestId: `req-register-${clientId}`, + command: "client.register", + clientId, + payload: { clientId, clientType: "desktop" }, + }); + ensureSessionParticipant(ctx, "session-1", clientId, "observer"); + recordCapabilityClients(ctx, "session-1", clientId, [ + { capabilityName: "hook.beforeTool" }, + ]); + } + await transport.handleCommand({ + version: "v1", + requestId: "req-detach", + command: "session.detach", + clientId: "left-client", + sessionId: "session-1", + }); + const joinedEvents: HubEventEnvelope[] = []; + transport.subscribe("joined-client", (event) => joinedEvents.push(event)); + + // While the owner is connected it keeps the request; unregistering + // cancels it. + const ownerRequest = requestCapability( + ctx, + "session-1", + "hook.beforeTool", + {}, + "owner-client", + ); + const ownerRequestCancelled = expect(ownerRequest).rejects.toThrow( + /disconnected before it could respond/, + ); + expect([...ctx.pendingCapabilityRequests.values()][0]?.targetClientId).toBe( + "owner-client", + ); + await transport.handleCommand({ + version: "v1", + requestId: "req-unregister", + command: "client.unregister", + clientId: "owner-client", + }); + await ownerRequestCancelled; + + // The client that detached is skipped even though it is still connected. + void requestCapability( + ctx, + "session-1", + "hook.beforeTool", + {}, + "owner-client", + ); + expect( + [...ctx.pendingCapabilityRequests.values()].map((r) => r.targetClientId), + ).toEqual(["joined-client"]); + expect(joinedEvents).toEqual( + expect.arrayContaining([ + expect.objectContaining({ + event: "capability.requested", + payload: expect.objectContaining({ targetClientId: "joined-client" }), + }), + ]), + ); + }); + it("cancels pending capability requests when a run is aborted", async () => { const abort = vi.fn().mockResolvedValue(undefined); const transport = createTransport({ diff --git a/sdk/packages/core/src/hub/server/handlers/capability-handlers.ts b/sdk/packages/core/src/hub/server/handlers/capability-handlers.ts index 1ba71b1939..5fa2c58304 100644 --- a/sdk/packages/core/src/hub/server/handlers/capability-handlers.ts +++ b/sdk/packages/core/src/hub/server/handlers/capability-handlers.ts @@ -3,14 +3,66 @@ import { createSessionId } from "@cline/shared"; import { logHubMessage } from "../hub-server-logging"; import { errorReply, type HubTransportContext, okReply } from "./context"; +/** + * Shown to the user when a run fails because the client that owned one of its + * capabilities (hooks, client tools) went away mid-request. Client ids stay in + * the hub log (targetClientId), not in this text. + */ +export const CAPABILITY_OWNER_GONE_MESSAGE = + "The app running this task disconnected before it could respond. Send a message to continue."; + +export function recordCapabilityClients( + ctx: HubTransportContext, + sessionId: string, + clientId: string, + contributions: readonly { capabilityName: string }[], +): void { + if (!ctx.capabilityClientsBySession || contributions.length === 0) return; + // ponytail: entries live as long as the hub process; prune on session delete if it grows. + let byCapability = ctx.capabilityClientsBySession.get(sessionId); + if (!byCapability) { + byCapability = new Map(); + ctx.capabilityClientsBySession.set(sessionId, byCapability); + } + for (const { capabilityName } of contributions) { + const clients = byCapability.get(capabilityName) ?? []; + if (!clients.includes(clientId)) clients.push(clientId); + byCapability.set(capabilityName, clients); + } +} + +/** + * The requested owner if connected, else another client that registered the + * capability and is still connected and attached to the session (a client + * that detached no longer receives the session's requests). + */ +function resolveCapabilityOwner( + ctx: HubTransportContext, + sessionId: string, + capabilityName: string, + targetClientId: string, +): string | undefined { + if (ctx.clients.has(targetClientId)) return targetClientId; + const participants = ctx.sessionState.get(sessionId)?.participants; + return ctx.capabilityClientsBySession + ?.get(sessionId) + ?.get(capabilityName) + ?.find( + (clientId) => ctx.clients.has(clientId) && participants?.has(clientId), + ); +} + export async function requestCapability( ctx: HubTransportContext, sessionId: string, capabilityName: string, payload: Record, - targetClientId: string, + requestedClientId: string, onProgress?: (payload: Record) => void, ): Promise | undefined> { + const targetClientId = + resolveCapabilityOwner(ctx, sessionId, capabilityName, requestedClientId) ?? + requestedClientId; const requestId = createSessionId("capreq_"); const startedAt = performance.now(); logHubMessage("info", "capability.request.start", { @@ -19,6 +71,21 @@ export async function requestCapability( capabilityName, targetClientId, }); + // Unregistering cancels the owner's pending requests, but a request raised + // after that (a run still in flight when its client restarted) would wait + // forever: nobody can answer it and nothing times it out. + if (!ctx.clients.has(targetClientId)) { + logHubMessage("warn", "capability.request.end", { + requestId, + sessionId, + capabilityName, + targetClientId, + ok: false, + error: "target_client_not_connected", + durationMs: 0, + }); + throw new Error(CAPABILITY_OWNER_GONE_MESSAGE); + } return await new Promise((resolve, reject) => { ctx.pendingCapabilityRequests.set(requestId, { sessionId, diff --git a/sdk/packages/core/src/hub/server/handlers/context.ts b/sdk/packages/core/src/hub/server/handlers/context.ts index fadee2acd9..8d49f2bff8 100644 --- a/sdk/packages/core/src/hub/server/handlers/context.ts +++ b/sdk/packages/core/src/hub/server/handlers/context.ts @@ -61,6 +61,13 @@ export interface HubTransportContext { readonly sessionState: Map; readonly pendingApprovals: Map; readonly pendingCapabilityRequests: Map; + /** + * Per session and capability, every client that registered it, in order. + * A running task keeps the callbacks it was built with; this lets a request + * fall over to another connected client (one that joined the task) when the + * original owner is gone. + */ + readonly capabilityClientsBySession?: Map>; readonly suppressNextTerminalEventBySession: Map; /** * Count of RPC-driven turns (`run.start` / session input commands) diff --git a/sdk/packages/core/src/hub/server/handlers/session-handlers.ts b/sdk/packages/core/src/hub/server/handlers/session-handlers.ts index 3d7898c123..fbaefc1264 100644 --- a/sdk/packages/core/src/hub/server/handlers/session-handlers.ts +++ b/sdk/packages/core/src/hub/server/handlers/session-handlers.ts @@ -34,7 +34,11 @@ import { } from "../hub-client-contributions"; import { logHubMessage } from "../hub-server-logging"; import { toHubSessionRecord } from "../hub-session-records"; -import { cancelPendingCapabilityRequests } from "./capability-handlers"; +import { + CAPABILITY_OWNER_GONE_MESSAGE, + cancelPendingCapabilityRequests, + recordCapabilityClients, +} from "./capability-handlers"; import { asPlainRecord, ensureSessionParticipant, @@ -368,6 +372,7 @@ export async function handleSessionCreate( sessionId, configExtensionCount: configExtensions?.length ?? 0, }); + recordCapabilityClients(ctx, sessionId, clientId, clientContributions); const clientContributionRuntime = createHubClientContributionRuntime({ sessionId, targetClientId: clientId, @@ -675,6 +680,7 @@ export async function handleSessionRestore( const configExtensions = parseRuntimeConfigExtensions( runtimeOptions.configExtensions, ); + recordCapabilityClients(ctx, sessionId, clientId, clientContributions); const clientContributionRuntime = createHubClientContributionRuntime({ sessionId, targetClientId: clientId, @@ -956,7 +962,7 @@ export async function handleSessionDetach( ctx, (request) => request.sessionId === sessionId && request.targetClientId === clientId, - `Capability owner client ${clientId} detached before request was resolved.`, + CAPABILITY_OWNER_GONE_MESSAGE, ); const [session, snapshot] = await Promise.all([ readHubSessionRecord(ctx, sessionId), diff --git a/sdk/packages/core/src/hub/server/hub-server-transport.ts b/sdk/packages/core/src/hub/server/hub-server-transport.ts index c7457112ae..7ac7e408ec 100644 --- a/sdk/packages/core/src/hub/server/hub-server-transport.ts +++ b/sdk/packages/core/src/hub/server/hub-server-transport.ts @@ -58,6 +58,7 @@ import { resolvePendingApproval, } from "./handlers/approval-handlers"; import { + CAPABILITY_OWNER_GONE_MESSAGE, cancelPendingCapabilityRequests, handleCapabilityProgress, handleCapabilityRequest, @@ -333,6 +334,7 @@ export class HubServerTransport implements NativeHubTransport { sessionState: this.sessionState, pendingApprovals: this.pendingApprovals, pendingCapabilityRequests: this.pendingCapabilityRequests, + capabilityClientsBySession: new Map(), suppressNextTerminalEventBySession: this.suppressNextTerminalEventBySession, activeRpcTurnCountBySession: this.activeRpcTurnCountBySession, @@ -1307,7 +1309,7 @@ export class HubServerTransport implements NativeHubTransport { cancelPendingCapabilityRequests( this.ctx, (request) => request.targetClientId === clientId, - `Capability owner client ${clientId} disconnected before request was resolved.`, + CAPABILITY_OWNER_GONE_MESSAGE, ); } diff --git a/sdk/packages/core/src/runtime/host/local-runtime-host.test.ts b/sdk/packages/core/src/runtime/host/local-runtime-host.test.ts index cdbd5f8242..4ee51e413b 100644 --- a/sdk/packages/core/src/runtime/host/local-runtime-host.test.ts +++ b/sdk/packages/core/src/runtime/host/local-runtime-host.test.ts @@ -3517,6 +3517,109 @@ describe("LocalRuntimeHost", () => { }); }); + it("reattaches instead of replacing a session that is mid-run", async () => { + const sessionId = "sess-live-takeover"; + const manifest = createManifest(sessionId); + const sessionService = { + ensureSessionsDir: vi.fn().mockReturnValue("/tmp/sessions"), + createRootSessionWithArtifacts: vi.fn().mockResolvedValue({ + manifestPath: "/tmp/manifest.json", + messagesPath: "/tmp/messages.json", + manifest, + }), + persistSessionMessages: vi.fn(), + updateSessionStatus: vi.fn().mockResolvedValue({ + updated: true, + endedAt: "2026-01-01T00:00:05.000Z", + }), + writeSessionManifest: vi.fn(), + listSessions: vi.fn().mockResolvedValue([]), + deleteSession: vi.fn().mockResolvedValue({ deleted: true }), + }; + const runtimeBuilder = { + build: vi.fn().mockReturnValue({ tools: [], shutdown: vi.fn() }), + }; + const createAgentStub = (onRun: () => Promise) => { + let activeRun = false; + let rejectRun: ((error: Error) => void) | undefined; + return { + run: vi.fn(() => { + activeRun = true; + return new Promise((_resolve, reject) => { + rejectRun = (error) => { + activeRun = false; + reject(error); + }; + void onRun(); + }); + }), + continue: vi.fn(), + abort: vi.fn(() => rejectRun?.(new Error("aborted by user"))), + subscribeEvents: vi.fn().mockReturnValue(() => {}), + getAgentId: vi.fn().mockReturnValue("agent-root-1"), + getConversationId: vi.fn().mockReturnValue("conv-root-1"), + shutdown: vi.fn().mockResolvedValue(undefined), + getMessages: vi.fn().mockReturnValue([]), + canStartRun: vi.fn(() => !activeRun), + }; + }; + let markRunStarted: (() => void) | undefined; + const runStarted = new Promise((resolve) => { + markRunStarted = resolve; + }); + const agents = [ + createAgentStub(async () => markRunStarted?.()), + createAgentStub(async () => {}), + ]; + const createAgent = vi.fn(() => agents[createAgent.mock.calls.length - 1]); + const manager = new RuntimeHostUnderTest({ + distinctId, + sessionService: sessionService as never, + runtimeBuilder, + createAgent: createAgent as never, + }); + await manager.startSession( + normalizeStartInput({ + config: createConfig({ sessionId }), + interactive: true, + }), + ); + const firstTurn = manager.runTurn({ sessionId, prompt: "slow" }); + await runStarted; + + // A second client (desktop reload, CLI resume) starts the same session + // with its on-disk snapshot while the run is still in flight. + const reattached = await manager.startSession( + normalizeStartInput({ + config: createConfig({ sessionId }), + interactive: true, + initialMessages: [ + { role: "user", content: [{ type: "text", text: "slow" }] }, + ] as never, + }), + ); + expect(reattached).toMatchObject({ + sessionId, + manifestPath: "/tmp/manifest.json", + messagesPath: "/tmp/messages.json", + }); + expect(createAgent).toHaveBeenCalledTimes(1); + await expect( + manager.startSession( + normalizeStartInput({ + config: createConfig({ sessionId }), + interactive: true, + prompt: "hello", + }), + ), + ).rejects.toThrow(/already has a run in progress/); + + // Stop still reaches the original run instead of an idle replacement. + await manager.abort(sessionId); + expect(agents[0].abort).toHaveBeenCalled(); + await expect(firstTurn).resolves.toMatchObject({ finishReason: "aborted" }); + }); + it("clears the remaining queue when a queue-initiated turn is aborted", async () => { const sessionId = "sess-second-abort-clears-queue"; const manifest = createManifest(sessionId); diff --git a/sdk/packages/core/src/runtime/host/local-runtime-host.ts b/sdk/packages/core/src/runtime/host/local-runtime-host.ts index af29cbfca4..e1ce57ed2d 100644 --- a/sdk/packages/core/src/runtime/host/local-runtime-host.ts +++ b/sdk/packages/core/src/runtime/host/local-runtime-host.ts @@ -491,6 +491,18 @@ export class LocalRuntimeHost implements RuntimeHost { async startSession(input: StartSessionInput): Promise { const requestedSessionId = input.config.sessionId?.trim() ?? ""; const sessionId = requestedSessionId || createSessionId(); + // Starting a session that is mid-run must not replace it: the replaced + // run keeps going unreachable (no abort, no queue) and its transcript + // writes resolve to the replacement's stale snapshot. Reattach instead. + const live = this.sessions.get(sessionId); + if (live && !live.agent.canStartRun()) { + if (input.prompt?.trim()) { + throw new Error( + `Session ${sessionId} already has a run in progress; queue the prompt instead.`, + ); + } + return await this.liveSessionStartResult(live); + } const isReadOnlyResumeStart = requestedSessionId.length > 0 && (input.initialMessages?.length ?? 0) > 0 && @@ -520,6 +532,28 @@ export class LocalRuntimeHost implements RuntimeHost { ); } + private async liveSessionStartResult( + session: ActiveSession, + ): Promise { + const artifacts = session.artifacts; + if (!artifacts) { + throw new Error( + `Session ${session.sessionId} is running but was never persisted`, + ); + } + const manifest = + (await this.invokeOptionalValue( + "readSessionManifest", + session.sessionId, + )) ?? artifacts.manifest; + return { + sessionId: session.sessionId, + manifest, + manifestPath: artifacts.manifestPath, + messagesPath: artifacts.messagesPath, + }; + } + private async startResolvedSession( input: ResolvedStartSessionInput, sessionId: string,