Skip to content

Commit 1b15169

Browse files
12ya12ya
committed
fix(server): only take a device over from a thread with no work in flight
Take over relied on the client's view of the holding thread. Clients omit older settled threads from their shells, and a thread can start a run between render and click, so a stale or missing shell could release a session a working agent was using. `device.close` now accepts `takeOver: { openedAt }`. Under the device lifecycle lock the server closes only the session opened at that time, and only when the holder has no active run or pending background task; otherwise it refuses with DeviceTakeoverRefusedError. A takeover never powers the device off. Web and mobile send the session they showed and no longer claim "isn't running" for a thread they cannot see.
1 parent 6a776e5 commit 1b15169

9 files changed

Lines changed: 179 additions & 10 deletions

File tree

‎apps/mobile/src/features/devices/DevicePreviewRouteScreen.tsx‎

Lines changed: 7 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -166,7 +166,12 @@ export function DevicePreviewScreen({
166166
const takeOver = async (holder: DeviceSession) => {
167167
const result = await close({
168168
environmentId,
169-
input: { threadId: holder.threadId, hostId: holder.hostId, deviceId: holder.deviceId },
169+
input: {
170+
threadId: holder.threadId,
171+
hostId: holder.hostId,
172+
deviceId: holder.deviceId,
173+
takeOver: { openedAt: holder.openedAt },
174+
},
170175
});
171176
if (result._tag === "Failure") {
172177
Dialog.alert("Could not take over device", String(Cause.squash(result.cause)));
@@ -434,7 +439,7 @@ function DeviceHolderBanner(props: {
434439
>
435440
<AppText numberOfLines={2} className="text-sm text-foreground-muted">
436441
Busy in <AppText className="font-t3-medium text-sm text-foreground">{title}</AppText>
437-
{running ? ", which is running." : ", which isn't running."}
442+
{running ? ", which is running." : shell ? ", which isn't running." : "."}
438443
</AppText>
439444
</Pressable>
440445
{running ? null : (

‎apps/server/src/device/DeviceService.test.ts‎

Lines changed: 75 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -40,6 +40,7 @@ import { makeDeviceHubOperations } from "./DeviceHubOperations.ts";
4040
import { NodeRuntimeUnavailableError } from "@t3tools/shared/nodeRuntime";
4141

4242
import * as DeviceService from "./DeviceService.ts";
43+
import * as DeviceThreadActivity from "./DeviceThreadActivity.ts";
4344
import * as SshDeviceHost from "./SshDeviceHost.ts";
4445
import { agentDeviceSession, releaseAgentDeviceSession } from "./AgentDeviceTarget.ts";
4546

@@ -1792,6 +1793,80 @@ describe("agent session release on close", () => {
17921793
}),
17931794
);
17941795

1796+
const withActivity = (active: ReadonlySet<string>) =>
1797+
Effect.provideService(
1798+
DeviceThreadActivity.DeviceThreadActivity,
1799+
DeviceThreadActivity.DeviceThreadActivity.of({
1800+
isActive: (threadId) => Effect.succeed(active.has(threadId)),
1801+
}),
1802+
);
1803+
1804+
it.effect("a takeover leaves a busy, unconfirmed, or replaced session claimed", () =>
1805+
Effect.gen(function* () {
1806+
const holder = ThreadId.make("holder");
1807+
const takeOver = (
1808+
service: DeviceService.DeviceService["Service"],
1809+
session: { readonly deviceId: string; readonly openedAt: string },
1810+
) =>
1811+
service.close({
1812+
threadId: holder,
1813+
deviceId: session.deviceId,
1814+
shutdown: true,
1815+
takeOver: { openedAt: session.openedAt },
1816+
});
1817+
1818+
const busy = yield* fixture().pipe(withActivity(new Set([holder])));
1819+
yield* busy.service.configure({ enabled: true, agentAccessEnabled: true });
1820+
const busySession = yield* openClaim(busy.service, holder);
1821+
expect(yield* takeOver(busy.service, busySession).pipe(Effect.flip)).toMatchObject({
1822+
_tag: "DeviceTakeoverRefusedError",
1823+
reason: "thread_active",
1824+
});
1825+
yield* takeOver(busy.service, { ...busySession, openedAt: "2000-01-01T00:00:00.000Z" });
1826+
1827+
// Without a thread activity source the server cannot vouch that the holder is idle.
1828+
const unknown = yield* fixture();
1829+
yield* unknown.service.configure({ enabled: true, agentAccessEnabled: true });
1830+
const unknownSession = yield* openClaim(unknown.service, holder);
1831+
expect(yield* takeOver(unknown.service, unknownSession).pipe(Effect.flip)).toMatchObject({
1832+
_tag: "DeviceTakeoverRefusedError",
1833+
reason: "activity_unknown",
1834+
});
1835+
1836+
for (const { service, closedSessions, requests } of [busy, unknown]) {
1837+
expect(closedSessions).toEqual([]);
1838+
expect((yield* service.state).sessions.map((entry) => entry.threadId)).toEqual([holder]);
1839+
expect(requests.some((url) => url.endsWith("/shutdown"))).toBe(false);
1840+
}
1841+
}),
1842+
);
1843+
1844+
it.effect("a takeover releases an idle holder's exact session without powering off", () =>
1845+
Effect.gen(function* () {
1846+
const { service, closedSessions, requests } = yield* fixture().pipe(withActivity(new Set()));
1847+
yield* service.configure({ enabled: true, agentAccessEnabled: true });
1848+
const holder = ThreadId.make("holder");
1849+
const taker = ThreadId.make("taker");
1850+
const session = yield* openClaim(service, holder);
1851+
yield* openClaim(service, taker, session.deviceId);
1852+
1853+
yield* service.close({
1854+
threadId: holder,
1855+
deviceId: session.deviceId,
1856+
shutdown: true,
1857+
takeOver: { openedAt: session.openedAt },
1858+
});
1859+
1860+
const name = yield* agentDeviceSession(holder, session.hostId, session.deviceId).pipe(
1861+
Effect.provide(NodeCrypto.layer),
1862+
);
1863+
expect(closedSessions).toMatchObject([{ params: { command: "close", session: name } }]);
1864+
const state = yield* service.state;
1865+
expect(state.sessions.map((entry) => entry.threadId)).toEqual([taker]);
1866+
expect(requests.some((url) => url.endsWith("/shutdown"))).toBe(false);
1867+
}),
1868+
);
1869+
17951870
it.effect.each(["reject", "http_error", "malformed"] as const)(
17961871
"retains retryable state on %s",
17971872
(failure) =>

‎apps/server/src/device/DeviceService.ts‎

Lines changed: 26 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -23,6 +23,7 @@ import {
2323
type DeviceSimlockIdentity,
2424
DeviceSimlockIdentityError,
2525
type DeviceSimlockIdentityInput,
26+
DeviceTakeoverRefusedError,
2627
type DeviceHostConfig,
2728
type EasDeviceHostConfig,
2829
type SimlockDeviceHostConfig,
@@ -72,6 +73,7 @@ import * as SecretResolver from "@t3tools/secret-provider-core/server/SecretReso
7273
import * as SecretResolverLayer from "../secrets/SecretResolver.ts";
7374
import { requireSettingsSecret } from "../secrets/settingsSecretValue.ts";
7475
import * as DeviceHost from "./DeviceHost.ts";
76+
import * as DeviceThreadActivity from "./DeviceThreadActivity.ts";
7577
import * as SshDeviceHost from "./SshDeviceHost.ts";
7678
import * as EasDeviceHost from "./EasDeviceHost.ts";
7779
import * as SimlockDeviceHost from "./SimlockDeviceHost.ts";
@@ -279,6 +281,7 @@ export const makeWithHosts = Effect.fn("DeviceService.makeWithHosts")(function*
279281
) {
280282
const settings = yield* ServerSettings.ServerSettingsService;
281283
const digest = yield* Digest.Digest;
284+
const threadActivity = yield* Effect.serviceOption(DeviceThreadActivity.DeviceThreadActivity);
282285
const lifecycleLock = yield* Semaphore.make(1);
283286
const readDeviceSettings = settings.getSettings.pipe(
284287
Effect.map((value) => ({
@@ -896,8 +899,22 @@ export const makeWithHosts = Effect.fn("DeviceService.makeWithHosts")(function*
896899
}));
897900
});
898901

902+
const requireIdleHolder = (threadId: ThreadId) => {
903+
const refuse = (reason: DeviceTakeoverRefusedError["reason"]) =>
904+
new DeviceTakeoverRefusedError({ threadId, reason });
905+
return Option.match(threadActivity, {
906+
onNone: () => Effect.fail(refuse("activity_unknown")),
907+
onSome: (activity) =>
908+
activity.isActive(threadId).pipe(
909+
Effect.mapError(() => refuse("activity_unknown")),
910+
Effect.flatMap((active) => (active ? Effect.fail(refuse("thread_active")) : Effect.void)),
911+
),
912+
});
913+
};
914+
899915
const close: DeviceService["Service"]["close"] = Effect.fn("DeviceService.close")(
900916
function* (input) {
917+
const poweringOff = input.shutdown === true && input.takeOver === undefined;
901918
// Open publishes under the same lock; an old close must not release a reopened claim.
902919
const closing = yield* lifecycleLock.withPermit(
903920
Effect.gen(function* () {
@@ -906,9 +923,13 @@ export const makeWithHosts = Effect.fn("DeviceService.makeWithHosts")(function*
906923
(session) =>
907924
session.threadId === input.threadId &&
908925
(input.hostId === undefined || session.hostId === input.hostId) &&
909-
(input.deviceId === undefined || session.deviceId === input.deviceId),
926+
(input.deviceId === undefined || session.deviceId === input.deviceId) &&
927+
(input.takeOver === undefined || session.openedAt === input.takeOver.openedAt),
910928
);
911-
const ending = input.shutdown
929+
if (input.takeOver !== undefined && closing.length > 0) {
930+
yield* requireIdleHolder(input.threadId);
931+
}
932+
const ending = poweringOff
912933
? state.sessions.filter((session) =>
913934
closing.some(
914935
(closed) =>
@@ -935,8 +956,8 @@ export const makeWithHosts = Effect.fn("DeviceService.makeWithHosts")(function*
935956
yield* ready.releaseAgentSession(name, session.deviceId);
936957
}
937958
}
938-
if (input.shutdown) yield* releaseLaunchers(ending);
939-
if (input.shutdown) {
959+
if (poweringOff) yield* releaseLaunchers(ending);
960+
if (poweringOff) {
940961
for (const session of closing) {
941962
const ready = yield* currentReadiness(session.hostId);
942963
if (!ready)
@@ -954,7 +975,7 @@ export const makeWithHosts = Effect.fn("DeviceService.makeWithHosts")(function*
954975
return closing;
955976
}),
956977
);
957-
if (input.shutdown)
978+
if (poweringOff)
958979
yield* Effect.forEach(
959980
closing,
960981
(session) =>

‎apps/server/src/device/DeviceThreadActivity.ts‎

Lines changed: 33 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,33 @@
1+
import type { ThreadId } from "@t3tools/contracts";
2+
import * as Context from "effect/Context";
3+
import * as Effect from "effect/Effect";
4+
import * as Layer from "effect/Layer";
5+
6+
import * as ProjectionStore from "../orchestration-v2/ProjectionStore.ts";
7+
8+
/** Whether a thread may still be driving its devices, so a takeover must leave them alone. */
9+
export class DeviceThreadActivity extends Context.Service<
10+
DeviceThreadActivity,
11+
{
12+
readonly isActive: (
13+
threadId: ThreadId,
14+
) => Effect.Effect<boolean, ProjectionStore.ProjectionStoreV2Error>;
15+
}
16+
>()("t3/device/DeviceThreadActivity") {}
17+
18+
export const layer = Layer.effect(
19+
DeviceThreadActivity,
20+
Effect.gen(function* () {
21+
const projections = yield* ProjectionStore.ProjectionStoreV2;
22+
return DeviceThreadActivity.of({
23+
// Background commands and subagents outlive their run and can still run agent-device.
24+
isActive: Effect.fn("DeviceThreadActivity.isActive")(function* (threadId: ThreadId) {
25+
const shell = yield* projections.getThreadShell(threadId);
26+
return (
27+
shell !== null &&
28+
(shell.activeRunId !== null || (shell.pendingBackgroundTasks?.length ?? 0) > 0)
29+
);
30+
}),
31+
});
32+
}),
33+
);

‎apps/server/src/server.ts‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -86,6 +86,7 @@ import * as PeerMcpClient from "./peer/PeerMcpClient.ts";
8686
import * as McpSessionRegistry from "./mcp/McpSessionRegistry.ts";
8787
import * as PreviewAutomationBroker from "./mcp/PreviewAutomationBroker.ts";
8888
import * as DeviceService from "./device/DeviceService.ts";
89+
import * as DeviceThreadActivity from "./device/DeviceThreadActivity.ts";
8990
import * as DeviceHubProxy from "./device/DeviceHubProxy.ts";
9091
import * as PreviewManager from "./preview/Manager.ts";
9192
import * as PortScanner from "./preview/PortScanner.ts";
@@ -501,6 +502,7 @@ const layerPreview = Layer.empty.pipe(
501502
);
502503

503504
const layerDevice = DeviceService.layer.pipe(
505+
Layer.provide(DeviceThreadActivity.layer.pipe(Layer.provide(ProjectionStoreV2.layer))),
504506
Layer.provide(layerServerSettings),
505507
Layer.provide(ProcessRunner.layer),
506508
Layer.provide(NetService.layer),

‎apps/web/src/components/device/DevicePanel.browser.test.tsx‎

Lines changed: 6 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -244,7 +244,12 @@ describe("DevicePanel takeover", () => {
244244
[
245245
{
246246
environmentId: threadA.environmentId,
247-
input: { threadId: holder, hostId: "local", deviceId: "phone" },
247+
input: {
248+
threadId: holder,
249+
hostId: "local",
250+
deviceId: "phone",
251+
takeOver: { openedAt: session.openedAt },
252+
},
248253
},
249254
],
250255
]);

‎apps/web/src/components/device/DevicePanel.tsx‎

Lines changed: 7 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -186,7 +186,12 @@ export function DevicePanel(props: {
186186
const settleOperation = startOperation(null);
187187
return close({
188188
environmentId,
189-
input: { threadId: holder.threadId, hostId: holder.hostId, deviceId: holder.deviceId },
189+
input: {
190+
threadId: holder.threadId,
191+
hostId: holder.hostId,
192+
deviceId: holder.deviceId,
193+
takeOver: { openedAt: holder.openedAt },
194+
},
190195
}).then((result) =>
191196
settleOperation(result._tag === "Failure" ? formatEnvironmentQueryError(result.cause) : null),
192197
);
@@ -434,7 +439,7 @@ function DeviceHolderNotice(props: {
434439
title: "another thread",
435440
}}
436441
/>
437-
{running ? ", which is running." : ", which isn't running."}
442+
{running ? ", which is running." : shell ? ", which isn't running." : "."}
438443
</span>
439444
{running ? null : (
440445
<Button

‎docs/internals/devices.md‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -122,6 +122,8 @@ checks that connection first: agent-device reports a sandboxed shell and a
122122
dropped tunnel alike as an unreachable daemon, so the launcher names which one it
123123
is. Ending a session (`device_close`, the Device panel, a shutdown) also closes
124124
its agent-device session, which quits the app that session opened and releases the CLI's claim on the device.
125+
Taking a device over from another thread is a close with `takeOver`; the server, not the client, checks that
126+
the holder has no active run or background task, because clients omit older settled threads from their shells.
125127

126128
Agents choose between these tools and ssh before they open anything, so
127129
sessions that hold `device` get a short block from

‎packages/contracts/src/device.ts‎

Lines changed: 21 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -358,6 +358,11 @@ export const DeviceCloseInput = Schema.Struct({
358358
deviceId: Schema.optional(DeviceId),
359359
/** Also shut the simulator or emulator down. Defaults to false. */
360360
shutdown: Schema.optional(Schema.Boolean),
361+
/**
362+
* Close another thread's session to take its device over: only the session opened
363+
* at `openedAt`, only while that thread is idle, and never powering the device off.
364+
*/
365+
takeOver: Schema.optional(Schema.Struct({ openedAt: Schema.String })),
361366
});
362367
export type DeviceCloseInput = typeof DeviceCloseInput.Type;
363368

@@ -620,13 +625,28 @@ export class DeviceActionUnavailableError extends Schema.TaggedError<DeviceActio
620625
}
621626
}
622627

628+
export class DeviceTakeoverRefusedError extends Schema.TaggedError<DeviceTakeoverRefusedError>()(
629+
"DeviceTakeoverRefusedError",
630+
{
631+
threadId: ThreadId,
632+
reason: Schema.Literals(["thread_active", "activity_unknown"]),
633+
},
634+
) {
635+
override get message(): string {
636+
return this.reason === "thread_active"
637+
? "That thread is working now, so the device stays with it. Take over after it finishes."
638+
: "Could not confirm that thread is idle, so the device stays with it.";
639+
}
640+
}
641+
623642
export const DeviceError = Schema.Union([
624643
DeviceHostUnavailableError,
625644
DevicePlatformUnavailableError,
626645
DeviceNotFoundError,
627646
DeviceBootError,
628647
DeviceOperationError,
629648
DeviceActionUnavailableError,
649+
DeviceTakeoverRefusedError,
630650
]);
631651
export type DeviceError = typeof DeviceError.Type;
632652

@@ -722,6 +742,7 @@ export const DeviceToolError = Schema.Union([
722742
DeviceBootError,
723743
DeviceOperationError,
724744
DeviceActionUnavailableError,
745+
DeviceTakeoverRefusedError,
725746
]);
726747
export type DeviceToolError = typeof DeviceToolError.Type;
727748

0 commit comments

Comments
 (0)