fix(core): fail fast instead of handing callbacks to another client

A run's hooks and client tools stay with the app that started it. If that
app disconnects or leaves the task, its pending requests are cancelled and
later ones fail immediately rather than going to whichever app joined
later, which may enforce different rules under the same name. Each case
emits task.capability_owner_gone so we can track how often it happens.

sdk/ARCHITECTURE.md

@@ -213,14 +213,12 @@ 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.
+the mistake-limit prompt) it was built with and never hands them to a client
+that joined later, since a same-named hook or tool may follow different rules.
+If the owning client unregisters or detaches, its pending requests are
+cancelled and later requests fail immediately instead of waiting, so the run
+ends with an error the user can continue from
+(`task.capability_owner_gone` records each case).
 
 Command progress follows the same runtime event boundary as other agent output.
 Shell executors emit structured stdout/stderr chunks through

sdk/packages/core/src/hub/server/boundary.test.ts

@@ -16,10 +16,7 @@ import {
 	handleApprovalRespond,
 	requestToolApproval,
 } from "./handlers/approval-handlers";
-import {
-	recordCapabilityClients,
-	requestCapability,
-} from "./handlers/capability-handlers";
+import { requestCapability } from "./handlers/capability-handlers";
 import {
 	ensureSessionParticipant,
 	ensureSessionState,
@@ -1792,11 +1789,11 @@ describe("HubServerTransport boundaries", () => {
 		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();
+	it("fails capability requests to an owner that left the task but stays connected", async () => {
+		const capture = vi.fn();
+		const transport = createTransport({ telemetry: { capture } as never });
 		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"]) {
+		for (const clientId of ["owner-client", "joined-client"]) {
 			await transport.handleCommand({
 				version: "v1",
 				requestId: `req-register-${clientId}`,
@@ -1805,44 +1802,51 @@ describe("HubServerTransport boundaries", () => {
 				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));
+		const detach = () =>
+			transport.handleCommand({
+				version: "v1",
+				requestId: "req-detach",
+				command: "session.detach",
+				clientId: "owner-client",
+				sessionId: "session-1",
+			});
 
-		// While the owner is connected it keeps the request; unregistering
-		// cancels it.
-		const ownerRequest = requestCapability(
+		// A request already waiting is cancelled when the owner leaves.
+		const pending = requestCapability(
 			ctx,
 			"session-1",
 			"hook.beforeTool",
 			{},
 			"owner-client",
 		);
-		const ownerRequestCancelled = expect(ownerRequest).rejects.toThrow(
+		const cancelled = expect(pending).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;
+		await detach();
+		await cancelled;
 
-		// The client that detached is skipped even though it is still connected.
+		// Later requests fail at once and are not handed to the client that
+		// joined: it may enforce a different hook under the same name.
+		await expect(
+			requestCapability(
+				ctx,
+				"session-1",
+				"hook.beforeTool",
+				{},
+				"owner-client",
+			),
+		).rejects.toThrow(/disconnected before it could respond/);
+		expect(ctx.pendingCapabilityRequests.size).toBe(0);
+		expect(
+			capture.mock.calls.map(([e]) => [e.event, e.properties.reason]),
+		).toEqual([
+			["task.capability_owner_gone", "cancelled_while_waiting"],
+			["task.capability_owner_gone", "detached"],
+		]);
+
+		// Attaching again makes it the owner again.
+		ensureSessionParticipant(ctx, "session-1", "owner-client", "observer");
 		void requestCapability(
 			ctx,
 			"session-1",
@@ -1852,15 +1856,7 @@ describe("HubServerTransport boundaries", () => {
 		);
 		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" }),
-				}),
-			]),
-		);
+		).toEqual(["owner-client"]);
 	});
 
 	it("cancels pending capability requests when a run is aborted", async () => {

sdk/packages/core/src/hub/server/handlers/capability-handlers.ts

@@ -1,5 +1,6 @@
 import type { HubCommandEnvelope, HubReplyEnvelope } from "@cline/shared";
 import { createSessionId } from "@cline/shared";
+import { captureCapabilityOwnerGone } from "../../../services/telemetry/core-events";
 import { logHubMessage } from "../hub-server-logging";
 import { errorReply, type HubTransportContext, okReply } from "./context";
 
@@ -11,58 +12,14 @@ import { errorReply, type HubTransportContext, okReply } from "./context";
 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<string, unknown>,
-	requestedClientId: string,
+	targetClientId: string,
 	onProgress?: (payload: Record<string, unknown>) => void,
 ): Promise<Record<string, unknown> | undefined> {
-	const targetClientId =
-		resolveCapabilityOwner(ctx, sessionId, capabilityName, requestedClientId) ??
-		requestedClientId;
 	const requestId = createSessionId("capreq_");
 	const startedAt = performance.now();
 	logHubMessage("info", "capability.request.start", {
@@ -71,19 +28,31 @@ 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)) {
+	// Unregistering or detaching 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 answers it and nothing times it
+	// out. Requests are never handed to another client: a client that joined
+	// later may enforce different hooks or tools under the same name.
+	const goneReason = !ctx.clients.has(targetClientId)
+		? "disconnected"
+		: ctx.detachedClientsBySession?.get(sessionId)?.has(targetClientId)
+			? "detached"
+			: undefined;
+	if (goneReason) {
 		logHubMessage("warn", "capability.request.end", {
 			requestId,
 			sessionId,
 			capabilityName,
 			targetClientId,
 			ok: false,
-			error: "target_client_not_connected",
+			error: `target_client_${goneReason}`,
 			durationMs: 0,
 		});
+		captureCapabilityOwnerGone(ctx.telemetry, {
+			sessionId,
+			capabilityName,
+			reason: goneReason,
+		});
 		throw new Error(CAPABILITY_OWNER_GONE_MESSAGE);
 	}
 	return await new Promise((resolve, reject) => {
@@ -103,6 +72,13 @@ export async function requestCapability(
 					durationMs: Math.round(performance.now() - startedAt),
 				});
 				if (!result.ok) {
+					if (result.error === CAPABILITY_OWNER_GONE_MESSAGE) {
+						captureCapabilityOwnerGone(ctx.telemetry, {
+							sessionId,
+							capabilityName,
+							reason: "cancelled_while_waiting",
+						});
+					}
 					reject(
 						new Error(
 							result.error ||

sdk/packages/core/src/hub/server/handlers/context.ts

@@ -62,12 +62,11 @@ export interface HubTransportContext {
 	readonly pendingApprovals: Map<string, PendingApproval>;
 	readonly pendingCapabilityRequests: Map<string, PendingCapabilityRequest>;
 	/**
-	 * 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.
+	 * Clients that explicitly detached from a session while staying connected.
+	 * Capability requests aimed at them fail fast instead of waiting; attaching
+	 * again clears the entry.
 	 */
-	readonly capabilityClientsBySession?: Map<string, Map<string, string[]>>;
+	readonly detachedClientsBySession?: Map<string, Set<string>>;
 	readonly suppressNextTerminalEventBySession: Map<string, string>;
 	/**
 	 * Count of RPC-driven turns (`run.start` / session input commands)
@@ -226,6 +225,7 @@ export function ensureSessionState(
 	role: SessionParticipant["role"],
 	options: { interactive?: boolean } = {},
 ): HubSessionState {
+	ctx.detachedClientsBySession?.get(sessionId)?.delete(clientId);
 	const existing = ctx.sessionState.get(sessionId);
 	if (existing) {
 		if (options.interactive !== undefined) {
@@ -268,6 +268,7 @@ export function ensureSessionParticipant(
 	role: SessionParticipant["role"],
 	options: { interactive?: boolean } = {},
 ): HubSessionState {
+	ctx.detachedClientsBySession?.get(sessionId)?.delete(clientId);
 	const existing = ctx.sessionState.get(sessionId);
 	if (existing) {
 		if (options.interactive !== undefined) {

sdk/packages/core/src/hub/server/handlers/session-handlers.ts

@@ -37,7 +37,6 @@ import { toHubSessionRecord } from "../hub-session-records";
 import {
 	CAPABILITY_OWNER_GONE_MESSAGE,
 	cancelPendingCapabilityRequests,
-	recordCapabilityClients,
 } from "./capability-handlers";
 import {
 	asPlainRecord,
@@ -372,7 +371,9 @@ export async function handleSessionCreate(
 		sessionId,
 		configExtensionCount: configExtensions?.length ?? 0,
 	});
-	recordCapabilityClients(ctx, sessionId, clientId, clientContributions);
+	// Re-creating after a detach makes this client the owner again, including
+	// for callbacks raised while the session starts.
+	ctx.detachedClientsBySession?.get(sessionId)?.delete(clientId);
 	const clientContributionRuntime = createHubClientContributionRuntime({
 		sessionId,
 		targetClientId: clientId,
@@ -680,7 +681,7 @@ export async function handleSessionRestore(
 		const configExtensions = parseRuntimeConfigExtensions(
 			runtimeOptions.configExtensions,
 		);
-		recordCapabilityClients(ctx, sessionId, clientId, clientContributions);
+		ctx.detachedClientsBySession?.get(sessionId)?.delete(clientId);
 		const clientContributionRuntime = createHubClientContributionRuntime({
 			sessionId,
 			targetClientId: clientId,
@@ -948,6 +949,13 @@ export async function handleSessionDetach(
 		);
 	}
 	const clientId = envelope.clientId?.trim() || "hub-client";
+	// ponytail: entries live as long as the hub process; prune on session delete if it grows.
+	let detached = ctx.detachedClientsBySession?.get(sessionId);
+	if (!detached && ctx.detachedClientsBySession) {
+		detached = new Set();
+		ctx.detachedClientsBySession.set(sessionId, detached);
+	}
+	detached?.add(clientId);
 	const state = ctx.sessionState.get(sessionId);
 	if (state) {
 		state.participants.delete(clientId);

sdk/packages/core/src/hub/server/hub-server-transport.ts

@@ -334,7 +334,7 @@ export class HubServerTransport implements NativeHubTransport {
 			sessionState: this.sessionState,
 			pendingApprovals: this.pendingApprovals,
 			pendingCapabilityRequests: this.pendingCapabilityRequests,
-			capabilityClientsBySession: new Map(),
+			detachedClientsBySession: new Map(),
 			suppressNextTerminalEventBySession:
 				this.suppressNextTerminalEventBySession,
 			activeRpcTurnCountBySession: this.activeRpcTurnCountBySession,

sdk/packages/core/src/services/telemetry/core-events.ts

@@ -79,6 +79,7 @@ export const CORE_TELEMETRY_EVENTS = {
 		DIFF_EDIT_FAILED: "task.diff_edit_failed",
 		PROVIDER_API_ERROR: "task.provider_api_error",
 		MISTAKE_LIMIT_REACHED: "task.mistake_limit_reached",
+		CAPABILITY_OWNER_GONE: "task.capability_owner_gone",
 		PROVIDER_REQUEST_STARTED: TASK_PROVIDER_REQUEST_STARTED_EVENT,
 		PROVIDER_STREAM_STARTED: TASK_PROVIDER_STREAM_STARTED_EVENT,
 		FIRST_CHUNK_RECEIVED: TASK_FIRST_CHUNK_RECEIVED_EVENT,
@@ -645,6 +646,22 @@ export function captureMistakeLimitReached(
 	});
 }
 
+/**
+ * Records when a run's hook, client tool, or other client callback fails
+ * because the app that owns it disconnected or left the task.
+ */
+export function captureCapabilityOwnerGone(
+	telemetry: ITelemetryService | undefined,
+	properties: {
+		sessionId: string;
+		capabilityName: string;
+		/** disconnected / detached: failed before asking; cancelled_while_waiting: the app left mid-request. */
+		reason: "disconnected" | "detached" | "cancelled_while_waiting";
+	},
+): void {
+	emit(telemetry, CORE_TELEMETRY_EVENTS.TASK.CAPABILITY_OWNER_GONE, properties);
+}
+
 export function captureRunCommandsTimeout(
 	telemetry: ITelemetryService | undefined,
 	properties: RunCommandsTimeoutTelemetryProperties,