Skip to content
46 changes: 45 additions & 1 deletion apps/server/src/device/DeviceService.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -49,7 +49,7 @@ const fixture = Effect.fn("fixture")(function* (
| DeviceHost.AgentDeviceStartError,
inspectError = false,
installTool?: Parameters<typeof DeviceService.makeWithHosts>[3],
options: { readonly endsDevices?: boolean } = {},
options: { readonly endsDevices?: boolean; readonly onShutdown?: Effect.Effect<void> } = {},
) {
const settings = yield* Ref.make(DEFAULT_SERVER_SETTINGS);
const starts: string[] = [];
Expand All @@ -68,6 +68,7 @@ const fixture = Effect.fn("fixture")(function* (
return HttpClientResponse.fromWeb(request, new Response(new Uint8Array([137, 80, 78, 71])));
}
if (request.url.endsWith("/shutdown")) {
yield* options.onShutdown ?? Effect.void;
shutDown = true;
booted = false;
return HttpClientResponse.fromWeb(request, Response.json({ ok: true }));
Expand Down Expand Up @@ -586,6 +587,49 @@ it.effect("keeps shutdown successful when subsequent discovery fails", () =>
}).pipe(Effect.scoped),
);

it.effect("a device no thread has open is not turned off under an open racing it", () =>
Effect.gen(function* () {
const started = yield* Deferred.make<void>();
const finish = yield* Deferred.make<void>();
let pause = false;
const { service } = yield* fixture(Effect.void, undefined, false, undefined, false, undefined, {
onShutdown: Effect.suspend(() =>
pause
? Deferred.succeed(started, undefined).pipe(Effect.andThen(Deferred.await(finish)))
: Effect.void,
),
});
yield* service.configure({ enabled: true });
const first = ThreadId.make("race-first");
const other = ThreadId.make("race-other");
const { deviceId, platform } = yield* service.open({
threadId: first,
deviceId: "Pixel_API_35",
platform: "android",
});
yield* service.close({ threadId: first, deviceId });

// An open that registers first keeps the device on.
yield* service.open({ threadId: other, deviceId, platform });
expect(yield* service.shutdownUnlessOpen({ deviceId, platform })).toBe(false);
expect((yield* service.state).sessions.map((session) => session.threadId)).toEqual([other]);
yield* service.close({ threadId: other, deviceId });

// An open that lands while the shutdown runs fails instead of registering.
pause = true;
const shuttingDown = yield* service
.shutdownUnlessOpen({ deviceId, platform })
.pipe(Effect.forkChild);
yield* Deferred.await(started);
const opening = yield* Effect.exit(service.open({ threadId: other, deviceId, platform }));
yield* Deferred.succeed(finish, undefined);
expect(yield* Fiber.join(shuttingDown)).toBe(true);
expect(opening._tag).toBe("Failure");
expect(String(opening)).toContain("turned off while it was opening");
expect((yield* service.state).sessions).toEqual([]);
}).pipe(Effect.scoped),
);

it.effect.each(["shutdown", "close"] as const)(
"%s releases iOS capture so reopening uses a fresh session",
(operation) =>
Expand Down
52 changes: 52 additions & 0 deletions apps/server/src/device/DeviceService.ts
Original file line number Diff line number Diff line change
Expand Up @@ -139,6 +139,10 @@ export class DeviceService extends Context.Service<
) => Effect.Effect<DeviceSession, DeviceError>;
readonly close: (input: DeviceCloseInput) => Effect.Effect<void, DeviceError>;
readonly shutdown: (input: DeviceShutdownInput) => Effect.Effect<void, DeviceError>;
/** Turns a device off unless a thread has it open; false when one does. */
readonly shutdownUnlessOpen: (
input: DeviceShutdownInput,
) => Effect.Effect<boolean, DeviceError>;
readonly detail: (input: DeviceDetailInput) => Effect.Effect<DeviceDetail, DeviceError>;
readonly action: (input: DeviceActionInput) => Effect.Effect<DeviceDetail, DeviceError>;
readonly screenshot: (input: {
Expand Down Expand Up @@ -187,6 +191,12 @@ export const makeWithHosts = Effect.fn("DeviceService.makeWithHosts")(function*
const settings = yield* ServerSettings.ServerSettingsService;
const crypto = yield* Crypto.Crypto;
const lifecycleLock = yield* Semaphore.make(1);
// shutdownUnlessOpen claims a device under the lifecycle lock, so an open that
// overlaps the claim or the shutdown cannot register a session on it.
let shutdownEpoch = 0;
const shutdownClaims = new Map<string, number>();
const shutdownsInFlight = new Map<string, number>();
const deviceKey = (hostId: DeviceHostId, deviceId: DeviceId) => `${hostId}\u0000${deviceId}`;
const readDeviceSettings = settings.getSettings.pipe(
Effect.map((value) => ({
enabled: value.enableDeviceSupport,
Expand Down Expand Up @@ -585,6 +595,7 @@ export const makeWithHosts = Effect.fn("DeviceService.makeWithHosts")(function*

const open: DeviceService["Service"]["open"] = Effect.fn("DeviceService.open")(
function* (input, context) {
const openedEpoch = shutdownEpoch;
const host = yield* resolveHost(input.hostId);
yield* ensurePlatform(host, input.platform);
const ready = yield* readiness(host.id);
Expand Down Expand Up @@ -671,6 +682,13 @@ export const makeWithHosts = Effect.fn("DeviceService.makeWithHosts")(function*
hostLabel: host.label,
reason: "Device support was turned off while the device was opening.",
});
const key = deviceKey(host.id, device.id);
if (shutdownsInFlight.has(key) || (shutdownClaims.get(key) ?? 0) > openedEpoch)
return yield* new DeviceHostUnavailableError({
hostId: host.id,
hostLabel: host.label,
reason: "The device was turned off while it was opening. Retry the operation.",
});
yield* publish((current) => ({
...current,
sessions: [
Expand Down Expand Up @@ -754,6 +772,39 @@ export const makeWithHosts = Effect.fn("DeviceService.makeWithHosts")(function*
},
);

const shutdownUnlessOpen: DeviceService["Service"]["shutdownUnlessOpen"] = Effect.fn(
"DeviceService.shutdownUnlessOpen",
)(function* (input) {
const host = yield* resolveHost(input.hostId);
const key = deviceKey(host.id, input.deviceId);
const claimed = yield* lifecycleLock.withPermit(
Effect.gen(function* () {
const { state } = yield* SynchronizedRef.get(stateRef);
if (
state.sessions.some(
(session) => session.hostId === host.id && session.deviceId === input.deviceId,
)
)
return false;
shutdownEpoch += 1;
shutdownClaims.set(key, shutdownEpoch);
shutdownsInFlight.set(key, (shutdownsInFlight.get(key) ?? 0) + 1);
return true;
}),
);
if (!claimed) return false;
yield* shutdown({ ...input, hostId: host.id }).pipe(
Effect.ensuring(
Effect.sync(() => {
const remaining = (shutdownsInFlight.get(key) ?? 1) - 1;
if (remaining === 0) shutdownsInFlight.delete(key);
else shutdownsInFlight.set(key, remaining);
}),
),
);
Comment thread
coderabbitai[bot] marked this conversation as resolved.
Resolved
Comment thread
coderabbitai[bot] marked this conversation as resolved.
Resolved
return true;
});

const screenshot: DeviceService["Service"]["screenshot"] = Effect.fn("DeviceService.screenshot")(
function* (input) {
const { ready, device } = yield* resolveDevice(input.hostId, input.deviceId);
Expand Down Expand Up @@ -882,6 +933,7 @@ export const makeWithHosts = Effect.fn("DeviceService.makeWithHosts")(function*
open,
close,
shutdown,
shutdownUnlessOpen,
detail,
action,
screenshot,
Expand Down
86 changes: 86 additions & 0 deletions apps/server/src/mcp/McpDeviceToolkit.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,7 @@ import * as NodeServices from "@effect/platform-node/NodeServices";
import {
DeviceHostUnavailableError,
DeviceId,
type DeviceSession,
EnvironmentId,
ProviderInstanceId,
ThreadId,
Expand Down Expand Up @@ -235,3 +236,88 @@ it.effect("rejects unavailable agent access before booting or opening a device",
),
);
});

it.effect("device_close with shutdown turns off a device whose session a restart forgot", () => {
const shutdowns: Array<unknown> = [];
const closes: Array<unknown> = [];
const otherThread = ThreadId.make("thread-device-other");
const layerFor = (sessions: ReadonlyArray<DeviceSession>) =>
McpHttpServer.layerDeviceToolkit.pipe(
Layer.provideMerge(McpServer.McpServer.layer),
Layer.provideMerge(McpToolAccessTestkit.liveThreadsLayer),
Layer.provide(
Layer.mock(DeviceService.DeviceService)({
list: Effect.succeed({ ...state, sessions }),
sessionsForThread: (id) =>
Effect.succeed(sessions.filter((session) => session.threadId === id)),
shutdownUnlessOpen: (input) => Effect.sync(() => shutdowns.push(input) > 0),
close: (input) => Effect.sync(() => void closes.push(input)),
}),
),
Layer.provide(layerProjectsMock),
Layer.provide(
ServerConfig.layerTest(process.cwd(), { prefix: "t3-mcp-device-toolkit-test-" }),
),
Layer.provide(NodeServices.layer),
);
const call = (args: Record<string, unknown>, sessions: ReadonlyArray<DeviceSession> = []) =>
Effect.gen(function* () {
const server = yield* McpServer.McpServer;
return yield* server
.callTool({ name: "device_close", arguments: args })
.pipe(
Effect.provideService(McpInvocationContext.McpInvocationContext, invocation(["device"])),
Effect.provideService(McpSchema.McpServerClient, client),
);
}).pipe(Effect.scoped, Effect.provide(layerFor(sessions)));
const session = (owner: ThreadId, hostId = "local"): DeviceSession => ({
threadId: owner,
hostId,
deviceId: DeviceId.make("UDID-1"),
platform: "ios",
openedAt: "2026-09-08T00:00:00.000Z",
});

return Effect.gen(function* () {
const forgotten = yield* call({ deviceId: "UDID-1", shutdown: true });
expect(forgotten.isError).toBe(false);
expect(shutdowns).toEqual([{ hostId: "local", deviceId: "UDID-1", platform: "ios" }]);
expect(closes).toEqual([]);

const foreign = yield* call({ deviceId: "UDID-1", shutdown: true }, [session(otherThread)]);
expect(foreign.isError).toBe(true);
expect(foreign.content).toEqual([
expect.objectContaining({ text: expect.stringContaining("open in another thread") }),
]);
expect(shutdowns).toHaveLength(1);

const untargeted = yield* call({ shutdown: true });
expect(untargeted.isError).toBe(true);
expect(untargeted.content).toEqual([
expect.objectContaining({ text: expect.stringContaining("nothing was turned off") }),
]);
expect(shutdowns).toHaveLength(1);

const recorded = yield* call({ deviceId: "UDID-1", shutdown: true }, [session(threadId)]);
expect(recorded.isError).toBe(false);
expect(closes).toEqual([{ threadId, deviceId: "UDID-1", shutdown: true }]);
expect(shutdowns).toHaveLength(1);

// Without hostId the session on an SSH host is this thread's, so close turns it off there.
const remote = yield* call({ deviceId: "UDID-1", shutdown: true }, [
session(threadId, "ssh-mac"),
]);
expect(remote.isError).toBe(false);
expect(closes).toHaveLength(2);
expect(shutdowns).toHaveLength(1);

const remoteForeign = yield* call({ deviceId: "UDID-1", shutdown: true }, [
session(otherThread, "ssh-mac"),
]);
expect(remoteForeign.isError).toBe(true);
expect(remoteForeign.content).toEqual([
expect.objectContaining({ text: expect.stringContaining("open in another thread") }),
]);
expect(shutdowns).toHaveLength(1);
});
});
35 changes: 35 additions & 0 deletions apps/server/src/mcp/toolkits/device/handlers.ts
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,7 @@ import {
type DeviceHostId,
type DeviceId,
type DevicePlatform,
type DeviceSession,
type DeviceSummary,
DeviceToolUnavailableError,
LOCAL_DEVICE_HOST_ID,
Expand Down Expand Up @@ -251,6 +252,40 @@ const handlers = {
Effect.gen(function* () {
const scope = yield* requireDeviceAccess;
const devices = yield* DeviceService.DeviceService;
if (input.shutdown === true) {
// Like devices.close, an omitted hostId matches a session on any host.
const matches = (session: DeviceSession) =>
(input.hostId === undefined || session.hostId === input.hostId) &&
(input.deviceId === undefined || session.deviceId === input.deviceId);
const recorded = (yield* devices.sessionsForThread(scope.thread.threadId)).some(matches);
// Sessions live in memory, so a restart forgets them and close alone would turn nothing off.
if (!recorded) {
if (input.deviceId === undefined) {
return yield* new DeviceToolUnavailableError({
reason:
"No device is open in this thread, so nothing was turned off. Pass deviceId to turn a device off.",
});
}
const openElsewhere = new DeviceToolUnavailableError({
reason: `Device ${input.deviceId}${input.hostId === undefined ? "" : ` on host ${input.hostId}`} is open in another thread, so it was not turned off.`,
});
const state = yield* devices.list;
Comment thread
coderabbitai[bot] marked this conversation as resolved.
Resolved
if (state.sessions.some(matches)) return yield* openElsewhere;
const target = yield* pickDevice(state.devices, input);
// The service rechecks under its session lock, since another thread may open it meanwhile.
if (
target.booted &&
!(yield* devices.shutdownUnlessOpen({
hostId: target.hostId,
deviceId: target.id,
platform: target.platform,
}))
) {
return yield* openElsewhere;
}
return {};
}
}
yield* devices.close({
threadId: scope.thread.threadId,
...(input.hostId === undefined ? {} : { hostId: input.hostId }),
Expand Down
Loading