fix(performance): close CPU, FPS and log collectors when their start is cancelled

docs/telemetry.md

@@ -202,7 +202,9 @@ reconnect abort signal; cancellation without an aborted operation and host
 Cancelled stream requests close sessions that finish opening after cancellation
 or before a discarded response. Unexpected cleanup failures retain their own
 `stream.cancel_cleanup` capture. Response delivery releases cancellation ownership;
-failed response delivery closes its undelivered session. Physical iOS mirror startup now also captures
+failed response delivery closes its undelivered session. CPU, Display FPS and log session
+starts follow the same ownership and report their cleanup failures as `stream.cancel_cleanup`;
+their tool failures after request cancellation count as `cancelled`. Physical iOS mirror startup now also captures
 unexpected server errors before returning the existing visible failure.
 
 Server discovery failures no longer create a second generic UI exception. The

packages/core/src/cpu/tools.ts

@@ -5,6 +5,8 @@ import type { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js";
 import { registerAppTool } from "@modelcontextprotocol/ext-apps/server";
 import type { CallToolResult } from "@modelcontextprotocol/sdk/types.js";
 import { captureServerError } from "../telemetry.ts";
+import { openRequestSession } from "../request-session.ts";
+import type { ServerRequestContext } from "../request-session.ts";
 import { z } from "zod";
 import { cpuTargetSchema } from "@mobile-dev/shared/cpu.ts";
 import type { CpuTarget } from "@mobile-dev/shared/cpu.ts";
@@ -23,11 +25,11 @@ const metadata = { ui: { visibility: ["app", "model"] as ("app" | "model")[] } }
 const read = { readOnlyHint: true, destructiveHint: false, openWorldHint: false };
 const write = { ...read, readOnlyHint: false };
 
-function safe<T>(handler: (input: T) => Promise<CallToolResult>) {
-  return async (input: T): Promise<CallToolResult> => {
-    try { return await handler(input); }
+function safe<T>(handler: (input: T, context: ServerRequestContext) => Promise<CallToolResult>) {
+  return async (input: T, context: ServerRequestContext): Promise<CallToolResult> => {
+    try { return await handler(input, context); }
     catch (error) {
-      captureServerError(error, "cpu.tool");
+      captureServerError(error, "cpu.tool", { signal: context.signal });
       return { isError: true, content: [{ type: "text", text: errorMessage(error) }] };
     }
   };
@@ -72,9 +74,9 @@ export function registerCpuTools(server: McpServer, cpu: CpuSessions, simulators
   registerAppTool(server, "mobile_cpu_session", {
     title: "Monitor native app CPU and memory", description: "Monitor process and per-thread CPU plus main-process memory once per second without an app SDK, app launch or restart. Returns sessionId and cpuUri in the tool result. Poll mobile_read_cpu with sessionId and the previous cursor as after; close with mobile_cpu_close when finished. Android uses a small external native /proc collector over ADB, without root or debugger attachment; the device must permit ADB shell to read app counters. iOS uses debugserver and requires a development-signed build with get-task-allow and no existing Xcode/LLDB attachment. Physical iOS 17.4+ also requires pairing, Developer Mode and a mounted developer disk image; pass target.kind: physical and the hardware UDID from mobile_list_ios_devices. Attach and detach briefly pause the app. 100% is one occupied device CPU core. Memory is RSS on Android and physical footprint on iOS, reported in bytes. Idle sessions expire after five minutes.",
     inputSchema: { target: cpuTargetSchema }, annotations: write, _meta: metadata,
-  }, safe(async ({ target }: { target: CpuTarget }) => {
+  }, safe(async ({ target }: { target: CpuTarget }, context) => {
     await validateDevice(target);
-    const id = cpu.open(target);
+    const id = await openRequestSession(context.signal, async () => cpu.open(target), id => cpu.closeSession(id));
     const data = { target, sessionId: id, cpuUri: `cpu://mobile-dev/${id}/batch?after=0` };
     const text = JSON.stringify(data);
     return { content: [{ type: "text", text }], structuredContent: data };

packages/core/src/fps/tools.ts

@@ -12,6 +12,8 @@ import { listIosDevices } from "../ios-devices.ts";
 import { listAndroidLogDevices } from "../native-logs.ts";
 import type { DisplayFpsSessions } from "./sessions.ts";
 import { captureServerError } from "../telemetry.ts";
+import { openRequestSession } from "../request-session.ts";
+import type { ServerRequestContext } from "../request-session.ts";
 
 const sessionString = z.string();
 const sessionId = sessionString.regex(/^[a-f0-9]{64}$/);
@@ -28,11 +30,11 @@ const visibility: ("app" | "model")[] = ["app", "model"];
 const metadata = { ui: { visibility } };
 const read = { readOnlyHint: true, destructiveHint: false, openWorldHint: false };
 const write = { ...read, readOnlyHint: false };
-function safe<T>(handler: (input: T) => Promise<CallToolResult>) {
-  return async (input: T): Promise<CallToolResult> => {
-    try { return await handler(input); }
+function safe<T>(handler: (input: T, context: ServerRequestContext) => Promise<CallToolResult>) {
+  return async (input: T, context: ServerRequestContext): Promise<CallToolResult> => {
+    try { return await handler(input, context); }
     catch (error) {
-      captureServerError(error, "display_fps.tool");
+      captureServerError(error, "display_fps.tool", { signal: context.signal });
       return { isError: true, content: [{ type: "text", text: errorMessage(error) }] };
     }
   };
@@ -61,9 +63,9 @@ export function registerDisplayFpsTools(server: McpServer, sessions: DisplayFpsS
     const text = JSON.stringify(batch);
     return { contents: [{ uri: uri.href, mimeType: "application/json", text }] };
   });
-  const startSession = safe(async ({ target }: { target: DisplayFpsTarget }) => {
+  const startSession = safe(async ({ target }: { target: DisplayFpsTarget }, context) => {
     await validate(target);
-    const id = sessions.open(target);
+    const id = await openRequestSession(context.signal, async () => sessions.open(target), id => sessions.closeSession(id));
     const timeOrigin = performance.now() / 1000;
     const data = { sessionId: id, timeOrigin, fpsUri: `display-fps://mobile-dev/${id}/batch?after=0` };
     const text = JSON.stringify(data);

packages/core/src/log-tools.ts

@@ -14,17 +14,19 @@ import { listIosDevices } from "./ios-devices.ts";
 import { physicalIosLogDevice } from "./physical-ios-logs.ts";
 import type { CallToolResult } from "@modelcontextprotocol/sdk/types.js";
 import { captureServerError } from "./telemetry.ts";
+import { openRequestSession } from "./request-session.ts";
+import type { ServerRequestContext } from "./request-session.ts";
 import { parseResourceInput } from "./resource-input.ts";
 
 const sessionId = z.string().regex(/^[a-f0-9]{64}$/);
 const sequence = z.coerce.number().int().nonnegative().max(Number.MAX_SAFE_INTEGER);
 const annotations = { readOnlyHint: true, destructiveHint: false, openWorldHint: false };
 const metadata = { ui: { visibility: ["app", "model"] as ("app" | "model")[] } };
-function safe<T>(handler: (input: T) => Promise<CallToolResult>) {
-  return async (input: T): Promise<CallToolResult> => {
-    try { return await handler(input); }
+function safe<T>(handler: (input: T, context: ServerRequestContext) => Promise<CallToolResult>) {
+  return async (input: T, context: ServerRequestContext): Promise<CallToolResult> => {
+    try { return await handler(input, context); }
     catch (error) {
-      captureServerError(error, "logs.tool");
+      captureServerError(error, "logs.tool", { signal: context.signal });
       return { isError: true, content: [{ type: "text", text: errorMessage(error) }] };
     }
   };
@@ -56,7 +58,7 @@ export function registerLogTools(server: McpServer, logs: LogSessions, simulator
   registerAppTool(server, "mobile_logs_session", {
     title: "Start app logs", description: "Stream iOS simulator or physical-device unified logs, Android logcat, and/or a selected Metro inspector's JS console and exceptions. For a physical iPhone, use kind physical and the hardware UDID from mobile_list_ios_devices, not its CoreDevice ID. The phone must be connected and paired. Use an iOS executable name for process, or pid for one process lifetime; choose only one. Use an Android packageName to follow the app across restarts. Leaving the app filter empty includes all device processes. iOS unified logs exclude ordinary print/printf output and may redact private values. Metro URL must be local HTTP and targetId must come from mobile_log_sources. Does not launch or rebuild the app.",
     inputSchema: { options: logOptionsSchema }, annotations, _meta: metadata,
-  }, safe(async ({ options }: { options: LogOptions }) => {
+  }, safe(async ({ options }: { options: LogOptions }, context) => {
     if (options.native?.platform === "ios") {
       if (options.native.kind === "physical") await physicalIosLogDevice(options.native.deviceId, discoverIosDevices);
       else await simulators.device(options.native.deviceId, true);
@@ -65,7 +67,7 @@ export function registerLogTools(server: McpServer, logs: LogSessions, simulator
       throw new ExpectedOperationError("device_unavailable", "The selected Android device is absent or unauthorized. Refresh log sources.");
     }
     if (options.metro) parseBaseUrl(options.metro.url, "Metro URL");
-    const id = logs.open(options);
+    const id = await openRequestSession(context.signal, async () => logs.open(options), id => logs.closeSession(id));
     const handles = { sessionId: id, logsUri: `logs://mobile-dev/${id}/batch?after=0` };
     // Models need the handle for subsequent reads; _meta is only for the app.
     return { content: [{ type: "text", text: JSON.stringify({ status: "started", ...handles }) }], structuredContent: { options, ...handles }, _meta: handles };

packages/core/test/collector-startup.test.ts

@@ -0,0 +1,130 @@
+import test from "node:test";
+import type { TestContext } from "node:test";
+import assert from "node:assert/strict";
+import { setImmediate } from "node:timers/promises";
+import { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js";
+import { Client } from "@modelcontextprotocol/sdk/client/index.js";
+import { InMemoryTransport } from "@modelcontextprotocol/sdk/inMemory.js";
+import { CpuSessions } from "../src/cpu/sessions.ts";
+import { registerCpuTools } from "../src/cpu/tools.ts";
+import { DisplayFpsSessions } from "../src/fps/sessions.ts";
+import { registerDisplayFpsTools } from "../src/fps/tools.ts";
+import { LogSessions } from "../src/log-sessions.ts";
+import { registerLogTools } from "../src/log-tools.ts";
+import { installTracePropagation, instrumentMcpServer } from "../src/telemetry.ts";
+import { fakeSimctl, UDID } from "../../test-support/fixtures.ts";
+
+/** Holds the first device validation until released; later validations pass straight through. */
+function validationGate() {
+  let enter = () => {};
+  let release = () => {};
+  const entered = new Promise<void>(resolve => { enter = resolve; });
+  const released = new Promise<void>(resolve => { release = resolve; });
+  return { entered, release, async wait() { enter(); await released; } };
+}
+
+async function connect(t: TestContext, register: (server: McpServer) => void) {
+  const server = new McpServer({ name: "collector-startup-test", version: "1" });
+  instrumentMcpServer(server);
+  register(server);
+  const [clientTransport, serverTransport] = InMemoryTransport.createLinkedPair();
+  installTracePropagation(serverTransport);
+  const client = new Client({ name: "collector-startup-client", version: "1" });
+  t.after(async () => { await client.close(); await server.close(); });
+  await server.connect(serverTransport);
+  await client.connect(clientTransport);
+  return client;
+}
+
+/** Cancels a start request while its device validation is pending, then lets validation finish. */
+async function cancelDuringValidation(client: Client, gate: ReturnType<typeof validationGate>, name: string, args: Record<string, unknown>) {
+  const abort = new AbortController();
+  const pending = client.callTool({ name, arguments: args }, undefined, { signal: abort.signal });
+  const rejection = assert.rejects(pending);
+  await gate.entered;
+  abort.abort();
+  await rejection;
+  gate.release();
+  await setImmediate();
+}
+
+/** Starts a session, then aborts the request after its response arrived. */
+async function startDelivered(client: Client, name: string, args: Record<string, unknown>) {
+  const abort = new AbortController();
+  const result = await client.callTool({ name, arguments: args }, undefined, { signal: abort.signal });
+  assert.equal(result.isError, undefined, JSON.stringify(result.content));
+  abort.abort();
+  await setImmediate();
+  return result.structuredContent as { sessionId: string };
+}
+
+test("a CPU start cancelled during validation leaves no monitor, and a delivered session stays open", async t => {
+  const gate = validationGate();
+  const simctl = fakeSimctl();
+  let opened = 0;
+  let stopped = 0;
+  const cpu = new CpuSessions({ apps: async () => [{ bundleId: "dev.test", pid: 123 }], monitor: async () => {
+    opened++;
+    return { closed: new Promise(() => {}), async stop() { stopped++; } };
+  } });
+  t.after(() => cpu.close());
+  const client = await connect(t, server => {
+    registerCpuTools(server, cpu, { async device(udid, booted) { await gate.wait(); return simctl.device(udid, booted); } });
+  });
+  const args = { target: { deviceId: UDID, bundleId: "dev.test" } };
+  await cancelDuringValidation(client, gate, "mobile_cpu_session", args);
+  assert.equal(opened, stopped);
+  const { sessionId } = await startDelivered(client, "mobile_cpu_session", args);
+  assert.equal(stopped, opened - 1);
+  const reading = await client.callTool({ name: "mobile_read_cpu", arguments: { sessionId } });
+  assert.equal(reading.isError, undefined);
+  await client.callTool({ name: "mobile_cpu_close", arguments: { sessionId } });
+  assert.equal(stopped, opened);
+});
+
+test("a Display FPS start cancelled during validation leaves no monitor, and a delivered session stays open", async t => {
+  const gate = validationGate();
+  let opened = 0;
+  let stopped = 0;
+  const sessions = new DisplayFpsSessions(async () => {
+    opened++;
+    return { closed: new Promise(() => {}), async stop() { stopped++; } };
+  });
+  t.after(() => sessions.close());
+  const client = await connect(t, server => {
+    registerDisplayFpsTools(server, sessions, { ios: async () => [], android: async () => {
+      await gate.wait();
+      return [{ id: "emulator-5554", name: "Pixel" }];
+    } });
+  });
+  const args = { target: { platform: "android", deviceId: "emulator-5554" } };
+  await cancelDuringValidation(client, gate, "mobile_display_fps_session", args);
+  assert.equal(opened, stopped);
+  const { sessionId } = await startDelivered(client, "mobile_display_fps_session", args);
+  assert.equal(stopped, opened - 1);
+  const reading = await client.callTool({ name: "mobile_read_display_fps", arguments: { sessionId } });
+  assert.equal(reading.isError, undefined);
+  await client.callTool({ name: "mobile_display_fps_close", arguments: { sessionId } });
+  assert.equal(stopped, opened);
+});
+
+test("a log start cancelled during validation leaves no reader, and a delivered session stays open", async t => {
+  const gate = validationGate();
+  const simctl = fakeSimctl();
+  let opened = 0;
+  let stopped = 0;
+  const logs = new LogSessions({ native: () => { opened++; return async () => { stopped++; }; }, metro: () => async () => {} });
+  t.after(() => logs.close());
+  const client = await connect(t, server => {
+    registerLogTools(server, logs, { async device(udid, booted) { await gate.wait(); return simctl.device(udid, booted); } });
+  });
+  const args = { options: { native: { platform: "ios", deviceId: UDID } } };
+  await cancelDuringValidation(client, gate, "mobile_logs_session", args);
+  assert.equal(opened, stopped);
+  const { sessionId } = await startDelivered(client, "mobile_logs_session", args);
+  assert.equal(stopped, opened - 1);
+  const reading = await client.callTool({ name: "mobile_read_logs", arguments: { sessionId } });
+  assert.equal(reading.isError, undefined);
+  await client.callTool({ name: "mobile_logs_close", arguments: { sessionId } });
+  assert.equal(stopped, opened);
+});