From 742fd9e2af01319c5b6aaf233d871817bfd27320 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Tue, 27 Oct 2026 06:17:56 +0000 Subject: [PATCH 1/5] test(server): reproduce #17470, closing a device leaves its agent-device session held DeviceService.close drops T3's session but never runs agent-device close for the --config/--session the agent was given, so the daemon's device claim stays live and other threads get DEVICE_IN_USE. Co-Authored-By: Claude Opus 5.5 --- apps/server/src/device/DeviceService.test.ts | 138 +++++++++++++++++++ 1 file changed, 138 insertions(+) diff --git a/apps/server/src/device/DeviceService.test.ts b/apps/server/src/device/DeviceService.test.ts index b4b06d56d7c33..430ff5b059137 100644 --- a/apps/server/src/device/DeviceService.test.ts +++ b/apps/server/src/device/DeviceService.test.ts @@ -19,6 +19,7 @@ import * as Deferred from "effect/Deferred"; import * as Fiber from "effect/Fiber"; import * as FileSystem from "effect/FileSystem"; import * as Layer from "effect/Layer"; +import * as Path from "effect/Path"; import * as Queue from "effect/Queue"; import * as Ref from "effect/Ref"; import * as Schema from "effect/Schema"; @@ -31,6 +32,7 @@ import * as ProcessRunner from "../processRunner.ts"; import * as ServerSettings from "../serverSettings.ts"; import * as DeviceHost from "./DeviceHost.ts"; import { makeDeviceHubOperations } from "./DeviceHubOperations.ts"; +import { AGENT_DEVICE_VERSION } from "./DeviceToolchain.ts"; import { NodeRuntimeUnavailableError } from "@t3tools/shared/nodeRuntime"; import * as DeviceService from "./DeviceService.ts"; @@ -586,6 +588,142 @@ it.effect("keeps shutdown successful when subsequent discovery fails", () => }).pipe(Effect.scoped), ); +it.effect("closing a thread's device ends the agent-device session its agent was given", () => + Effect.gen(function* () { + const config = yield* ServerConfig.ServerConfig; + const fs = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + // Installed already, so reaching the agent-device CLI runs no npm install. + const installDir = path.join(config.baseDir, "tools", "agent-device", AGENT_DEVICE_VERSION); + const entryPath = path.join( + installDir, + "node_modules", + "agent-device", + "bin", + "agent-device.mjs", + ); + yield* fs.makeDirectory(path.dirname(entryPath), { recursive: true }); + yield* fs.writeFileString(entryPath, ""); + yield* fs.writeFileString(path.join(installDir, ".install-complete"), AGENT_DEVICE_VERSION); + + const commands: Array> = []; + const http = HttpClient.make((request) => + Effect.succeed( + HttpClientResponse.fromWeb( + request, + new URL(request.url).pathname === "/api/devices" + ? Response.json({ + simulators: [], + emulators: [ + { + id: "emulator-5554", + name: "Pixel_API_35", + platform: "android", + version: "Android 15", + booted: true, + physical: false, + }, + ], + }) + : Response.json({ ok: true }), + ), + ), + ); + const run = () => Effect.succeed({ code: 0, stdout: " V....D libx264", stderr: "" }); + const ready: DeviceHost.DeviceHostReady = { + nodePath: process.execPath, + hub: { origin: "http://device.test" }, + devices: makeDeviceHubOperations({ + hostId: LOCAL_DEVICE_HOST_ID, + origin: "http://device.test", + http, + run, + androidAvailable: Effect.succeed(true), + }), + helpers: { serveSimAxSettings: null, serveSimCli: null }, + run, + }; + const host: DeviceHost.DeviceHost["Service"] = { + id: LOCAL_DEVICE_HOST_ID, + label: "This machine", + summary: Effect.succeed({ + id: LOCAL_DEVICE_HOST_ID, + kind: "local", + label: "Test server", + platforms: [{ platform: "android", available: true }], + hubInstalled: true, + agentDeviceInstalled: true, + }), + platformAvailability: (platform) => Effect.succeed({ platform, available: true }), + ensureReady: () => Effect.succeed(ready), + ensureAgentReady: () => + Effect.succeed({ + ...ready, + agentDevice: { baseUrl: "http://agent.test", token: "test", entryPath }, + }), + current: Effect.succeed(ready), + stopAgent: Effect.void, + stop: Effect.void, + }; + const service = yield* DeviceService.make.pipe( + Effect.provideService(DeviceHost.DeviceHost, host), + Effect.provideService(ProcessRunner.ProcessRunner, { + run: (input) => + Effect.sync(() => { + commands.push([input.command, ...input.args]); + return { + code: ChildProcessSpawner.ExitCode(0), + stdout: "", + stderr: "", + timedOut: false, + stdoutTruncated: false, + stderrTruncated: false, + stdoutInvalidUtf8: false, + stderrInvalidUtf8: false, + }; + }), + }), + Effect.provideService(HttpClient.HttpClient, http), + ); + + const threadId = ThreadId.make("thread-holding-device"); + const session = yield* service.open({ + threadId, + deviceId: "emulator-5554", + platform: "android", + }); + const agent = yield* service.agentTarget({ + threadId, + hostId: session.hostId, + deviceId: session.deviceId, + }); + const flag = (name: string) => agent.args[agent.args.indexOf(name) + 1]; + yield* service.close({ threadId }); + + expect((yield* service.state).sessions).toEqual([]); + // agent-device releases a claim held by T3's daemon only when its session is closed. + const closed = commands.filter((command) => { + const at = (name: string) => command[command.indexOf(name) + 1]; + return ( + command.includes("close") && + at("--config") === flag("--config") && + at("--session") === flag("--session") + ); + }); + expect(closed, `agent-device close for session ${flag("--session")}`).toHaveLength(1); + }).pipe( + Effect.provide( + Layer.mergeAll( + ServerSettings.layerTest({ enableDeviceSupport: true, enableAgentDeviceAccess: true }), + ServerConfig.layerTest(process.cwd(), { prefix: "t3-device-close-" }), + NodeCrypto.layer, + NetService.layer, + ).pipe(Layer.provideMerge(NodeServices.layer)), + ), + Effect.scoped, + ), +); + it.effect.each(["shutdown", "close"] as const)( "%s releases iOS capture so reopening uses a fresh session", (operation) => From f5a64cf9c5e5d5a8e0ec6a4e499377df03a0e622 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Tue, 27 Oct 2026 06:33:53 +0000 Subject: [PATCH 2/5] fix(server): closing, archiving or deleting a thread releases its device claims agent-device holds a device claim for each --config/--session T3 gives an agent, and releases it only when that session closes. T3 never closed it, so the claim stayed live while T3 ran and other threads got DEVICE_IN_USE. Dropping a thread's device session (close, with or without shutdown, and shutting a device down) now runs agent-device close with that session's flags through the local CLI, before any shutdown. It is best effort and starts, installs and boots nothing. Archiving and deleting a thread queue a device.cleanup effect next to terminal cleanup that closes its devices. Fixes pingdotgg/t3code#17470 Co-Authored-By: Claude Opus 5.5 --- apps/server/src/device/DeviceService.ts | 82 ++++++++++++++++++- apps/server/src/device/DeviceToolchain.ts | 10 +++ .../src/guest/GuestShareService.test.ts | 3 +- .../src/orchestration-v2/EffectOutbox.ts | 4 + .../src/orchestration-v2/EffectWorker.ts | 11 +++ .../src/orchestration-v2/Orchestrator.ts | 6 ++ .../ResourceCleanupService.ts | 17 +++- .../orchestration-v2/ThreadDeletion.test.ts | 1 + .../src/orchestration-v2/ThreadDeletion.ts | 6 ++ .../src/orchestration-v2/runtimeLayer.test.ts | 42 ++++++++++ .../project/ProjectService.deletion.test.ts | 9 +- .../src/provider/userInputAttachments.test.ts | 2 + 12 files changed, 189 insertions(+), 4 deletions(-) diff --git a/apps/server/src/device/DeviceService.ts b/apps/server/src/device/DeviceService.ts index d685dc46abf68..c8facd17d1a5c 100644 --- a/apps/server/src/device/DeviceService.ts +++ b/apps/server/src/device/DeviceService.ts @@ -32,7 +32,11 @@ import { SshCommandError } from "@t3tools/ssh/errors"; import * as FileSystem from "effect/FileSystem"; import { resolveNodeExecutable, nodeRuntimeUnavailableMessage } from "@t3tools/shared/nodeRuntime"; import * as Path from "effect/Path"; -import { ensureAgentDevice, ensureDeviceHub } from "./DeviceToolchain.ts"; +import { + ensureAgentDevice, + ensureDeviceHub, + installedAgentDeviceEntry, +} from "./DeviceToolchain.ts"; import * as ServerConfig from "../config.ts"; import { agentDeviceConfigPath, @@ -42,6 +46,7 @@ import { import * as Context from "effect/Context"; import * as Crypto from "effect/Crypto"; import * as DateTime from "effect/DateTime"; +import * as Duration from "effect/Duration"; import * as Effect from "effect/Effect"; import * as Layer from "effect/Layer"; import * as PubSub from "effect/PubSub"; @@ -183,6 +188,12 @@ export const makeWithHosts = Effect.fn("DeviceService.makeWithHosts")(function* }), ), installTool?: (tool: "hub" | "agent") => Effect.Effect, + endAgentSession: (target: { + readonly hostId: DeviceHostId; + /** Set for hosts with a daemon per device. */ + readonly deviceId?: DeviceId; + readonly session: string; + }) => Effect.Effect = () => Effect.void, ) { const settings = yield* ServerSettings.ServerSettingsService; const crypto = yield* Crypto.Crypto; @@ -691,12 +702,51 @@ export const makeWithHosts = Effect.fn("DeviceService.makeWithHosts")(function* }, ); + /** + * agent-device releases the device claim its daemon holds for a session only + * when that session closes. Best effort, and never starts a host or daemon. + */ + const releaseAgentSession = Effect.fn("DeviceService.releaseAgentSession")( + function* (session: DeviceSession) { + const deviceSettings = yield* readDeviceSettings; + if (!deviceSettings.enabled || !deviceSettings.agentAccessEnabled) return; + const ready = yield* hosts.get(session.hostId)?.current ?? Effect.succeed(null); + if (!ready) return; + // A device with its own daemon has nothing to release once it stops running. + if (ready.devices.agentEndpoint) yield* ready.devices.agentEndpoint(session.deviceId); + yield* endAgentSession({ + hostId: session.hostId, + ...(ready.devices.agentEndpoint ? { deviceId: session.deviceId } : {}), + session: yield* agentDeviceSession(session.threadId, session.hostId, session.deviceId).pipe( + Effect.provideService(Crypto.Crypto, crypto), + ), + }); + }, + (effect, session) => + effect.pipe( + Effect.catch((cause) => + Effect.logDebug("Could not end the agent-device session for a closed device", { + hostId: session.hostId, + deviceId: session.deviceId, + cause, + }), + ), + ), + ); + const shutdownDevice = Effect.fn("DeviceService.shutdownDevice")(function* ( hostId: DeviceHostId, deviceId: DeviceId, platform: DevicePlatform, ) { const ready = yield* readiness(hostId); + yield* Effect.forEach( + (yield* SynchronizedRef.get(stateRef)).state.sessions.filter( + (session) => session.hostId === ready.hostId && session.deviceId === deviceId, + ), + releaseAgentSession, + { discard: true }, + ); yield* ready.devices.shutdown({ id: deviceId, platform }); yield* publish((state) => ({ ...state, @@ -731,6 +781,7 @@ export const makeWithHosts = Effect.fn("DeviceService.makeWithHosts")(function* ...current, sessions: current.sessions.filter((session) => !closing.includes(session)), })); + yield* Effect.forEach(closing, releaseAgentSession, { discard: true }); if (input.shutdown) { yield* Effect.forEach( closing, @@ -991,6 +1042,35 @@ export const make = Effect.gen(function* () { }), ), ), + Effect.fn("DeviceService.endAgentSession")( + function* ({ hostId, deviceId, session }) { + const file = yield* configPath(hostId, deviceId); + // No config means no agent was ever given this host or device. + if (!(yield* fs.exists(file))) return; + const entryPath = yield* installedAgentDeviceEntry(config.baseDir); + if (entryPath === null) return; + const result = yield* runner.run({ + command: yield* resolveNodeExecutable("Device automation"), + args: [entryPath, "close", "--config", file, "--session", session], + env: { + ELECTRON_RUN_AS_NODE: "1", + AGENT_DEVICE_DAEMON_BASE_URL: undefined, + AGENT_DEVICE_DAEMON_AUTH_TOKEN: undefined, + AGENT_DEVICE_CONFIG: undefined, + }, + timeout: Duration.seconds(15), + }); + // A session its agent never used is not on the daemon, so close fails. + if (result.code !== 0) + yield* Effect.logDebug("agent-device close exited unsuccessfully", { + hostId, + code: result.code, + }); + }, + Effect.provideService(FileSystem.FileSystem, fs), + Effect.provideService(Path.Path, path), + Effect.catch((cause) => Effect.logDebug("agent-device close did not run", { cause })), + ), ); const hostContext = yield* Effect.context>>(); diff --git a/apps/server/src/device/DeviceToolchain.ts b/apps/server/src/device/DeviceToolchain.ts index 250e6cf14f3be..7b38bdf38f094 100644 --- a/apps/server/src/device/DeviceToolchain.ts +++ b/apps/server/src/device/DeviceToolchain.ts @@ -209,6 +209,16 @@ export const isDeviceHubInstalled = (baseDir: string) => export const isAgentDeviceInstalled = (baseDir: string) => isToolInstalled(baseDir, AGENT_DEVICE_SPEC, (paths) => paths.agentDevice); +/** The agent-device CLI entry when it is already installed; never installs it. */ +export const installedAgentDeviceEntry = Effect.fn("DeviceToolchain.installedAgentDeviceEntry")( + function* (baseDir: string) { + const path = yield* Path.Path; + return (yield* isAgentDeviceInstalled(baseDir)) + ? deviceToolchainPaths(path, baseDir).agentDevice.entryPath + : null; + }, +); + export const deviceToolVersions = Effect.fn("DeviceToolchain.versions")(function* ( baseDir: string, running: { hub?: string; agent?: string } = {}, diff --git a/apps/server/src/guest/GuestShareService.test.ts b/apps/server/src/guest/GuestShareService.test.ts index 2f0366d4d2d22..10ddeb2896c79 100644 --- a/apps/server/src/guest/GuestShareService.test.ts +++ b/apps/server/src/guest/GuestShareService.test.ts @@ -83,7 +83,7 @@ const layerProviders = Layer.mock(ProviderRegistry.ProviderRegistry)({ getProviders: Effect.succeed([]), }); const layerConfig = ServerConfig.layerTest(process.cwd(), { prefix: "t3-guest-share-" }); -// The real cleanup also closes terminals and previews; these tests only need the files gone. +// The real cleanup also closes terminals, previews and devices; these tests only need the files gone. const layerCleanup = Layer.effect( ResourceCleanupService.ResourceCleanupService, Effect.gen(function* () { @@ -93,6 +93,7 @@ const layerCleanup = Layer.effect( cleanupScheduledTasks: () => Effect.void, cleanupTerminals: () => Effect.void, cleanupPreviews: () => Effect.void, + cleanupDevices: () => Effect.void, cleanupAttachments: (attachmentIds) => Effect.forEach( attachmentIds, diff --git a/apps/server/src/orchestration-v2/EffectOutbox.ts b/apps/server/src/orchestration-v2/EffectOutbox.ts index 5a3ad681333cd..bf9f0b0eb601f 100644 --- a/apps/server/src/orchestration-v2/EffectOutbox.ts +++ b/apps/server/src/orchestration-v2/EffectOutbox.ts @@ -104,6 +104,9 @@ export const OrchestrationEffectRequestV2 = Schema.Union([ Schema.Struct({ type: Schema.Literal("preview.cleanup"), }), + Schema.Struct({ + type: Schema.Literal("device.cleanup"), + }), Schema.Struct({ type: Schema.Literal("attachment.cleanup"), attachmentIds: Schema.Array(Schema.String), @@ -130,6 +133,7 @@ export const REPLAY_SAFE_EFFECT_TYPES_AFTER_PROCESS_LOSS = [ "scheduled-tasks.cleanup", "terminal.cleanup", "preview.cleanup", + "device.cleanup", "attachment.cleanup", "thread-title.generate", "delegated-tasks.stop", diff --git a/apps/server/src/orchestration-v2/EffectWorker.ts b/apps/server/src/orchestration-v2/EffectWorker.ts index 6c0feea3526e8..54770d3dda2c0 100644 --- a/apps/server/src/orchestration-v2/EffectWorker.ts +++ b/apps/server/src/orchestration-v2/EffectWorker.ts @@ -511,6 +511,17 @@ export const layerExecutor: Layer.Layer< }), ), ); + case "device.cleanup": + return resourceCleanup.cleanupDevices(effect.threadId).pipe( + Effect.mapError( + (cause) => + new OrchestrationEffectExecutionError({ + effectId: effect.id, + effectType: effect.request.type, + cause, + }), + ), + ); case "attachment.cleanup": return resourceCleanup.cleanupAttachments(effect.request.attachmentIds).pipe( Effect.mapError( diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index ee8ab83b952aa..b3544a0bf755d 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -3934,6 +3934,12 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio threadId: command.threadId, request: { type: "terminal.cleanup" }, } satisfies PendingOrchestrationEffectV2, + { + id: `effect:${command.commandId}:device.cleanup`, + commandId: command.commandId, + threadId: command.threadId, + request: { type: "device.cleanup" }, + } satisfies PendingOrchestrationEffectV2, ]); } }); diff --git a/apps/server/src/orchestration-v2/ResourceCleanupService.ts b/apps/server/src/orchestration-v2/ResourceCleanupService.ts index ff56b42c492c1..2d96d91a93183 100644 --- a/apps/server/src/orchestration-v2/ResourceCleanupService.ts +++ b/apps/server/src/orchestration-v2/ResourceCleanupService.ts @@ -8,6 +8,7 @@ import * as Schema from "effect/Schema"; import { resolveAttachmentRelativePath } from "../attachmentPaths.ts"; import { parseAttachmentFileExtension, resolveAttachmentPathById } from "../attachmentStore.ts"; import * as ServerConfig from "../config.ts"; +import * as DeviceService from "../device/DeviceService.ts"; import * as PreviewManager from "../preview/Manager.ts"; import * as ScheduledTaskService from "../scheduledTasks/ScheduledTaskService.ts"; import * as TerminalManager from "../terminal/Manager.ts"; @@ -15,7 +16,7 @@ import * as TerminalManager from "../terminal/Manager.ts"; export class ResourceCleanupError extends Schema.TaggedError()( "ResourceCleanupError", { - operation: Schema.Literals(["terminal", "preview", "attachment", "scheduled-tasks"]), + operation: Schema.Literals(["terminal", "preview", "device", "attachment", "scheduled-tasks"]), threadId: Schema.optional(Schema.String), attachmentId: Schema.optional(Schema.String), cause: Schema.Defect(), @@ -26,6 +27,7 @@ export class ResourceCleanupService extends Context.Reference<{ readonly cleanupScheduledTasks: (threadId: ThreadId) => Effect.Effect; readonly cleanupTerminals: (threadId: string) => Effect.Effect; readonly cleanupPreviews: (threadId: string) => Effect.Effect; + readonly cleanupDevices: (threadId: ThreadId) => Effect.Effect; readonly cleanupAttachments: ( attachmentIds: ReadonlyArray, ) => Effect.Effect; @@ -34,6 +36,7 @@ export class ResourceCleanupService extends Context.Reference<{ cleanupScheduledTasks: () => Effect.void, cleanupTerminals: () => Effect.void, cleanupPreviews: () => Effect.void, + cleanupDevices: () => Effect.void, cleanupAttachments: () => Effect.void, }), }) {} @@ -43,6 +46,7 @@ export const layer = Layer.effect( Effect.gen(function* () { const terminals = yield* TerminalManager.TerminalManager; const previews = yield* PreviewManager.PreviewManager; + const devices = yield* DeviceService.DeviceService; const fileSystem = yield* FileSystem.FileSystem; const config = yield* ServerConfig.ServerConfig; const scheduledTasks = yield* ScheduledTaskService.ScheduledTaskService; @@ -81,6 +85,17 @@ export const layer = Layer.effect( ), ); }), + cleanupDevices: Effect.fn("ResourceCleanupService.cleanupDevices")(function ( + threadId: ThreadId, + ) { + return devices + .close({ threadId }) + .pipe( + Effect.mapError( + (cause) => new ResourceCleanupError({ operation: "device", threadId, cause }), + ), + ); + }), cleanupAttachments: Effect.fn("ResourceCleanupService.cleanupAttachments")(function ( attachmentIds: ReadonlyArray, ) { diff --git a/apps/server/src/orchestration-v2/ThreadDeletion.test.ts b/apps/server/src/orchestration-v2/ThreadDeletion.test.ts index ef7c23bf75120..25a4061eed7ad 100644 --- a/apps/server/src/orchestration-v2/ThreadDeletion.test.ts +++ b/apps/server/src/orchestration-v2/ThreadDeletion.test.ts @@ -293,6 +293,7 @@ it.effect("queues provider and resource cleanup and preserves an earlier deletio { type: "scheduled-tasks.cleanup" }, { type: "terminal.cleanup" }, { type: "preview.cleanup" }, + { type: "device.cleanup" }, { type: "attachment.cleanup", attachmentIds: ["shared_file"] }, ], ); diff --git a/apps/server/src/orchestration-v2/ThreadDeletion.ts b/apps/server/src/orchestration-v2/ThreadDeletion.ts index 67a27fcf76962..6fe47dc43221e 100644 --- a/apps/server/src/orchestration-v2/ThreadDeletion.ts +++ b/apps/server/src/orchestration-v2/ThreadDeletion.ts @@ -226,6 +226,12 @@ export const planThreadDeletion = Effect.fn("ThreadDeletion.planThreadDeletion") threadId: command.threadId, request: { type: "preview.cleanup" }, }); + effects.push({ + id: `effect:${command.commandId}:device.cleanup`, + commandId: command.commandId, + threadId: command.threadId, + request: { type: "device.cleanup" }, + }); const attachmentIds = Array.from(new Set(input.attachmentIds)); if (attachmentIds.length > 0) { effects.push({ diff --git a/apps/server/src/orchestration-v2/runtimeLayer.test.ts b/apps/server/src/orchestration-v2/runtimeLayer.test.ts index 025191f098823..8fa53f638330e 100644 --- a/apps/server/src/orchestration-v2/runtimeLayer.test.ts +++ b/apps/server/src/orchestration-v2/runtimeLayer.test.ts @@ -79,6 +79,7 @@ import * as ScheduledTaskService from "../scheduledTasks/ScheduledTaskService.ts import * as Scheduler from "../scheduling/Scheduler.ts"; import * as SecretRequests from "../secrets/SecretRequests.ts"; import * as PreviewManager from "../preview/Manager.ts"; +import * as DeviceService from "../device/DeviceService.ts"; import * as TerminalManager from "../terminal/Manager.ts"; import * as ResourceCleanupService from "./ResourceCleanupService.ts"; import * as ThreadLaunchService from "./ThreadLaunchService.ts"; @@ -317,12 +318,16 @@ const layerScheduledTasksTest = ScheduledTaskService.layer.pipe( ), ), ); +const closedDeviceThreads: Array = []; const layerResourceCleanupTest = ResourceCleanupService.layer.pipe( Layer.provideMerge(layerScheduledTasksTest), Layer.provide( Layer.mergeAll( Layer.mock(PreviewManager.PreviewManager)({ close: () => Effect.void }), Layer.mock(TerminalManager.TerminalManager)({ close: () => Effect.void }), + Layer.mock(DeviceService.DeviceService)({ + close: ({ threadId }) => Effect.sync(() => void closedDeviceThreads.push(threadId)), + }), ), ), ); @@ -491,6 +496,43 @@ it.layer(layerProjectDeletionTest)("project deletion during thread commands", (i }), ); + it.effect("archiving and deleting a thread close its devices through the effect worker", () => + Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const worker = yield* EffectWorker.OrchestrationEffectWorkerV2; + const threadId = ThreadId.make("device-cleanup-thread"); + const closes = () => closedDeviceThreads.filter((id) => id === threadId).length; + yield* orchestrator.dispatch({ + type: "thread.create", + commandId: CommandId.make("device-cleanup-create"), + createdBy: "user", + creationSource: "web", + threadId, + projectId: ProjectId.make("device-cleanup-project"), + title: "Device cleanup", + modelSelection, + runtimeMode: "full-access", + interactionMode: "default", + branch: null, + worktreePath: null, + }); + yield* orchestrator.dispatch({ + type: "thread.archive", + commandId: CommandId.make("device-cleanup-archive"), + threadId, + }); + yield* worker.drain(); + assert.equal(closes(), 1); + yield* orchestrator.dispatch({ + type: "thread.delete", + commandId: CommandId.make("device-cleanup-delete"), + threadId, + }); + yield* worker.drain(); + assert.equal(closes(), 2); + }), + ); + it.effect("waits for an in-flight thread update before planning deletion", () => Effect.gen(function* () { const projects = yield* ProjectService.ProjectService; diff --git a/apps/server/src/project/ProjectService.deletion.test.ts b/apps/server/src/project/ProjectService.deletion.test.ts index 41804842d5205..557772cf332d6 100644 --- a/apps/server/src/project/ProjectService.deletion.test.ts +++ b/apps/server/src/project/ProjectService.deletion.test.ts @@ -203,6 +203,7 @@ it.effect("retries a partial project deletion without repeating child events or assert.deepEqual( partialCleanup.map((effect) => [effect.thread_id, effect.effect_type]), [ + [firstThreadId, "device.cleanup"], [firstThreadId, "preview.cleanup"], [firstThreadId, "scheduled-tasks.cleanup"], [firstThreadId, "terminal.cleanup"], @@ -228,7 +229,7 @@ it.effect("retries a partial project deletion without repeating child events or assert.deepEqual(finalEvents[0], partialEvents[0]); assert.equal(finalEvents[2]?.command_id, commandId); const finalCleanup = yield* readCleanup; - assert.lengthOf(finalCleanup, 6); + assert.lengthOf(finalCleanup, 8); assert.deepEqual( finalCleanup.filter((effect) => effect.thread_id === firstThreadId), partialCleanup, @@ -238,6 +239,12 @@ it.effect("retries a partial project deletion without repeating child events or assert.deepEqual( finalCleanup.filter((effect) => effect.thread_id === threadId), [ + { + effect_id: `effect:${expectedCommandId}:device.cleanup`, + thread_id: threadId, + command_id: expectedCommandId, + effect_type: "device.cleanup", + }, { effect_id: `effect:${expectedCommandId}:preview.cleanup`, thread_id: threadId, diff --git a/apps/server/src/provider/userInputAttachments.test.ts b/apps/server/src/provider/userInputAttachments.test.ts index 81c167c038d87..f41817dba8ebe 100644 --- a/apps/server/src/provider/userInputAttachments.test.ts +++ b/apps/server/src/provider/userInputAttachments.test.ts @@ -18,6 +18,7 @@ import { extractDocumentText } from "./documentTextWorker.ts"; import * as Cleanup from "../orchestration-v2/ResourceCleanupService.ts"; import * as TerminalManager from "../terminal/Manager.ts"; import * as PreviewManager from "../preview/Manager.ts"; +import * as DeviceService from "../device/DeviceService.ts"; import * as ScheduledTaskService from "../scheduledTasks/ScheduledTaskService.ts"; import * as NodeZlib from "node:zlib"; import { appendUserInputAttachmentPaths } from "./userInputAttachments.ts"; @@ -327,6 +328,7 @@ describe("question answer paths", () => { Layer.mergeAll( Layer.mock(TerminalManager.TerminalManager)({}), Layer.mock(PreviewManager.PreviewManager)({}), + Layer.mock(DeviceService.DeviceService)({}), Layer.mock(ScheduledTaskService.ScheduledTaskService)({}), ), ), From 65747fe3e9ec0469c97627426d05063e2564731e Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Tue, 27 Oct 2026 07:18:36 +0000 Subject: [PATCH 3/5] fix(server): a reopened device waits for its closing agent-device session A device's agent-device session name is stable per thread and device, so an agentTarget during a pending release handed the agent a session the release then ended. Releases and session hand-out now share a lock. Releases also no longer skip when agent access is off: an EAS device's daemon keeps its claim then, and nothing on the release path starts anything. Co-Authored-By: Claude Opus 5.5 --- apps/server/src/device/DeviceService.test.ts | 108 +++++++++++++++++++ apps/server/src/device/DeviceService.ts | 8 +- 2 files changed, 113 insertions(+), 3 deletions(-) diff --git a/apps/server/src/device/DeviceService.test.ts b/apps/server/src/device/DeviceService.test.ts index 430ff5b059137..afd087287e3f9 100644 --- a/apps/server/src/device/DeviceService.test.ts +++ b/apps/server/src/device/DeviceService.test.ts @@ -724,6 +724,114 @@ it.effect("closing a thread's device ends the agent-device session its agent was ), ); +it.effect("reopening a device waits for its closing agent-device session to end", () => + Effect.gen(function* () { + const http = HttpClient.make((request) => + Effect.succeed( + HttpClientResponse.fromWeb( + request, + Response.json({ + simulators: [], + emulators: [ + { + id: "emulator-5554", + name: "Pixel", + version: "36", + platform: "android", + booted: true, + physical: false, + }, + ], + }), + ), + ), + ); + const run = () => Effect.succeed({ code: 0, stdout: " V....D libx264", stderr: "" }); + const ready: DeviceHost.DeviceHostAgentReady = { + nodePath: process.execPath, + hub: { origin: "http://device.test" }, + devices: makeDeviceHubOperations({ + hostId: LOCAL_DEVICE_HOST_ID, + origin: "http://device.test", + http, + run, + androidAvailable: Effect.succeed(true), + }), + agentDevice: { baseUrl: "http://agent.test", token: "test", entryPath: "/agent-device" }, + helpers: { serveSimAxSettings: null, serveSimCli: null }, + run, + }; + const host: DeviceHost.DeviceHost["Service"] = { + id: LOCAL_DEVICE_HOST_ID, + label: "This machine", + summary: Effect.succeed({ + id: LOCAL_DEVICE_HOST_ID, + kind: "local", + label: "Test server", + platforms: [{ platform: "android", available: true }], + hubInstalled: true, + agentDeviceInstalled: true, + }), + platformAvailability: (platform) => Effect.succeed({ platform, available: true }), + ensureReady: () => Effect.succeed(ready), + ensureAgentReady: () => Effect.succeed(ready), + // The daemon keeps running when agent access is turned off, as an EAS device's does. + current: Effect.succeed(ready), + stopAgent: Effect.void, + stop: Effect.void, + }; + const order: string[] = []; + const configured = yield* Queue.unbounded(); + const releaseStarted = yield* Deferred.make(); + const finishRelease = yield* Deferred.make(); + const service = yield* DeviceService.makeWithHosts( + new Map([[host.id, host]]), + undefined, + () => Queue.offer(configured, undefined).pipe(Effect.as("/agent-config.json")), + undefined, + ({ session }) => + Effect.gen(function* () { + yield* Deferred.succeed(releaseStarted, undefined); + yield* Deferred.await(finishRelease); + order.push(`released ${session}`); + }), + ).pipe(Effect.provide(NodeCrypto.layer)); + + const threadId = ThreadId.make("thread-reopening-device"); + const input = { + threadId, + hostId: LOCAL_DEVICE_HOST_ID, + deviceId: DeviceId.make("emulator-5554"), + platform: "android", + } as const; + yield* service.open(input); + yield* service.agentTarget(input); + yield* Queue.take(configured); + const closing = yield* service.close({ threadId }).pipe(Effect.forkChild); + yield* Deferred.await(releaseStarted); + yield* service.open(input); + const targeting = yield* service.agentTarget(input).pipe( + Effect.tap((target) => Effect.sync(() => order.push(`given ${target.args.at(-1)}`))), + Effect.forkChild, + ); + yield* Queue.take(configured); + yield* Deferred.succeed(finishRelease, undefined); + yield* Fiber.join(closing); + const target = yield* Fiber.join(targeting); + const session = target.args.at(-1); + expect(order).toEqual([`released ${session}`, `given ${session}`]); + + // Turning agent access off leaves the device's daemon holding the claim. + yield* service.configure({ agentAccessEnabled: false }); + yield* service.close({ threadId }); + expect(order).toEqual([`released ${session}`, `given ${session}`, `released ${session}`]); + }).pipe( + Effect.provide( + ServerSettings.layerTest({ enableDeviceSupport: true, enableAgentDeviceAccess: true }), + ), + ), +); + it.effect.each(["shutdown", "close"] as const)( "%s releases iOS capture so reopening uses a fresh session", (operation) => diff --git a/apps/server/src/device/DeviceService.ts b/apps/server/src/device/DeviceService.ts index c8facd17d1a5c..893dccc92d5ed 100644 --- a/apps/server/src/device/DeviceService.ts +++ b/apps/server/src/device/DeviceService.ts @@ -198,6 +198,9 @@ export const makeWithHosts = Effect.fn("DeviceService.makeWithHosts")(function* const settings = yield* ServerSettings.ServerSettingsService; const crypto = yield* Crypto.Crypto; const lifecycleLock = yield* Semaphore.make(1); + // Session names are stable per thread and device: a reopened device must not + // be handed its session while a close is still ending it. + const agentSessionLock = yield* Semaphore.make(1); const readDeviceSettings = settings.getSettings.pipe( Effect.map((value) => ({ enabled: value.enableDeviceSupport, @@ -708,8 +711,6 @@ export const makeWithHosts = Effect.fn("DeviceService.makeWithHosts")(function* */ const releaseAgentSession = Effect.fn("DeviceService.releaseAgentSession")( function* (session: DeviceSession) { - const deviceSettings = yield* readDeviceSettings; - if (!deviceSettings.enabled || !deviceSettings.agentAccessEnabled) return; const ready = yield* hosts.get(session.hostId)?.current ?? Effect.succeed(null); if (!ready) return; // A device with its own daemon has nothing to release once it stops running. @@ -724,6 +725,7 @@ export const makeWithHosts = Effect.fn("DeviceService.makeWithHosts")(function* }, (effect, session) => effect.pipe( + agentSessionLock.withPermit, Effect.catch((cause) => Effect.logDebug("Could not end the agent-device session for a closed device", { hostId: session.hostId, @@ -925,7 +927,7 @@ export const makeWithHosts = Effect.fn("DeviceService.makeWithHosts")(function* input.threadId, input.hostId, input.deviceId, - ).pipe(Effect.provideService(Crypto.Crypto, crypto)); + ).pipe(Effect.provideService(Crypto.Crypto, crypto), agentSessionLock.withPermit); return { agentDeviceId, args: ["--config", configPath, "--session", session] }; }), state: SynchronizedRef.get(stateRef).pipe(Effect.map(({ state }) => state)), From f8dd9bb4fa672617c98cccf5ab6defb157c7b970 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Tue, 27 Oct 2026 07:56:28 +0000 Subject: [PATCH 4/5] fix(server): a cancelled device close still releases its agent-device session Hold the agent-session lock across dropping the session and releasing it, and finish the release even if the close is interrupted, so a retry is never left without a session to release. Co-Authored-By: Claude Opus 5.5 --- apps/server/src/device/DeviceService.test.ts | 24 +++++++++++++++----- apps/server/src/device/DeviceService.ts | 16 +++++++------ 2 files changed, 27 insertions(+), 13 deletions(-) diff --git a/apps/server/src/device/DeviceService.test.ts b/apps/server/src/device/DeviceService.test.ts index afd087287e3f9..28bf83cb3bc10 100644 --- a/apps/server/src/device/DeviceService.test.ts +++ b/apps/server/src/device/DeviceService.test.ts @@ -782,8 +782,8 @@ it.effect("reopening a device waits for its closing agent-device session to end" }; const order: string[] = []; const configured = yield* Queue.unbounded(); - const releaseStarted = yield* Deferred.make(); - const finishRelease = yield* Deferred.make(); + const releaseStarted = yield* Queue.unbounded(); + const finishRelease = yield* Queue.unbounded(); const service = yield* DeviceService.makeWithHosts( new Map([[host.id, host]]), undefined, @@ -791,8 +791,8 @@ it.effect("reopening a device waits for its closing agent-device session to end" undefined, ({ session }) => Effect.gen(function* () { - yield* Deferred.succeed(releaseStarted, undefined); - yield* Deferred.await(finishRelease); + yield* Queue.offer(releaseStarted, undefined); + yield* Queue.take(finishRelease); order.push(`released ${session}`); }), ).pipe(Effect.provide(NodeCrypto.layer)); @@ -808,14 +808,14 @@ it.effect("reopening a device waits for its closing agent-device session to end" yield* service.agentTarget(input); yield* Queue.take(configured); const closing = yield* service.close({ threadId }).pipe(Effect.forkChild); - yield* Deferred.await(releaseStarted); + yield* Queue.take(releaseStarted); yield* service.open(input); const targeting = yield* service.agentTarget(input).pipe( Effect.tap((target) => Effect.sync(() => order.push(`given ${target.args.at(-1)}`))), Effect.forkChild, ); yield* Queue.take(configured); - yield* Deferred.succeed(finishRelease, undefined); + yield* Queue.offer(finishRelease, undefined); yield* Fiber.join(closing); const target = yield* Fiber.join(targeting); const session = target.args.at(-1); @@ -823,8 +823,20 @@ it.effect("reopening a device waits for its closing agent-device session to end" // Turning agent access off leaves the device's daemon holding the claim. yield* service.configure({ agentAccessEnabled: false }); + yield* Queue.offer(finishRelease, undefined); yield* service.close({ threadId }); + yield* Queue.take(releaseStarted); expect(order).toEqual([`released ${session}`, `given ${session}`, `released ${session}`]); + + // A cancelled close has already dropped the session, so nothing could retry its release. + yield* service.open(input); + const cancelled = yield* service.close({ threadId }).pipe(Effect.forkChild); + yield* Queue.take(releaseStarted); + cancelled.interruptUnsafe(); + yield* Queue.offer(finishRelease, undefined); + yield* Fiber.await(cancelled); + expect(order.at(-1)).toBe(`released ${session}`); + expect(order).toHaveLength(4); }).pipe( Effect.provide( ServerSettings.layerTest({ enableDeviceSupport: true, enableAgentDeviceAccess: true }), diff --git a/apps/server/src/device/DeviceService.ts b/apps/server/src/device/DeviceService.ts index 893dccc92d5ed..e93bba7db5b05 100644 --- a/apps/server/src/device/DeviceService.ts +++ b/apps/server/src/device/DeviceService.ts @@ -725,7 +725,6 @@ export const makeWithHosts = Effect.fn("DeviceService.makeWithHosts")(function* }, (effect, session) => effect.pipe( - agentSessionLock.withPermit, Effect.catch((cause) => Effect.logDebug("Could not end the agent-device session for a closed device", { hostId: session.hostId, @@ -748,7 +747,7 @@ export const makeWithHosts = Effect.fn("DeviceService.makeWithHosts")(function* ), releaseAgentSession, { discard: true }, - ); + ).pipe(agentSessionLock.withPermit); yield* ready.devices.shutdown({ id: deviceId, platform }); yield* publish((state) => ({ ...state, @@ -779,11 +778,14 @@ export const makeWithHosts = Effect.fn("DeviceService.makeWithHosts")(function* (input.deviceId === undefined || session.deviceId === input.deviceId), ); if (closing.length === 0) return; - yield* publish((current) => ({ - ...current, - sessions: current.sessions.filter((session) => !closing.includes(session)), - })); - yield* Effect.forEach(closing, releaseAgentSession, { discard: true }); + // Once a session leaves state nothing can retry its release, so finish it even if cancelled. + yield* Effect.gen(function* () { + yield* publish((current) => ({ + ...current, + sessions: current.sessions.filter((session) => !closing.includes(session)), + })); + yield* Effect.forEach(closing, releaseAgentSession, { discard: true }); + }).pipe(Effect.uninterruptible, agentSessionLock.withPermit); if (input.shutdown) { yield* Effect.forEach( closing, From 42f8f4b4bc954a78eb273b1cb045e48854641923 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Tue, 27 Oct 2026 08:23:04 +0000 Subject: [PATCH 5/5] fix(server): authenticate the agent-device close with the daemon endpoint agent-device 0.21.31 answers a close that carries only --config with 'Invalid token'; pass the config file's endpoint through the environment. Co-Authored-By: Claude Opus 5.5 --- apps/server/src/device/AgentDeviceTarget.ts | 8 ++++---- apps/server/src/device/DeviceService.test.ts | 7 +++++++ apps/server/src/device/DeviceService.ts | 7 +++++-- 3 files changed, 16 insertions(+), 6 deletions(-) diff --git a/apps/server/src/device/AgentDeviceTarget.ts b/apps/server/src/device/AgentDeviceTarget.ts index de5ff7ec713d4..8e7c296552651 100644 --- a/apps/server/src/device/AgentDeviceTarget.ts +++ b/apps/server/src/device/AgentDeviceTarget.ts @@ -7,11 +7,11 @@ import * as Path from "effect/Path"; import type { AgentDaemonEndpoint } from "./DeviceHost.ts"; -const encodeEndpoint = Schema.encodeEffect( - Schema.fromJsonString( - Schema.Struct({ daemonBaseUrl: Schema.String, daemonAuthToken: Schema.String }), - ), +const AgentDeviceConfigJson = Schema.fromJsonString( + Schema.Struct({ daemonBaseUrl: Schema.String, daemonAuthToken: Schema.String }), ); +const encodeEndpoint = Schema.encodeEffect(AgentDeviceConfigJson); +export const decodeAgentDeviceConfig = Schema.decodeUnknownEffect(AgentDeviceConfigJson); const key = Effect.fn("AgentDeviceTarget.key")(function* (value: string) { const crypto = yield* Crypto.Crypto; diff --git a/apps/server/src/device/DeviceService.test.ts b/apps/server/src/device/DeviceService.test.ts index 28bf83cb3bc10..afd7099acf3a5 100644 --- a/apps/server/src/device/DeviceService.test.ts +++ b/apps/server/src/device/DeviceService.test.ts @@ -607,6 +607,7 @@ it.effect("closing a thread's device ends the agent-device session its agent was yield* fs.writeFileString(path.join(installDir, ".install-complete"), AGENT_DEVICE_VERSION); const commands: Array> = []; + const closeEnv: Array> | undefined> = []; const http = HttpClient.make((request) => Effect.succeed( HttpClientResponse.fromWeb( @@ -671,6 +672,7 @@ it.effect("closing a thread's device ends the agent-device session its agent was run: (input) => Effect.sync(() => { commands.push([input.command, ...input.args]); + if (input.args.includes("close")) closeEnv.push(input.env); return { code: ChildProcessSpawner.ExitCode(0), stdout: "", @@ -711,6 +713,11 @@ it.effect("closing a thread's device ends the agent-device session its agent was ); }); expect(closed, `agent-device close for session ${flag("--session")}`).toHaveLength(1); + // agent-device authenticates the close with the token from the environment, not the config file. + expect(closeEnv[0]).toMatchObject({ + AGENT_DEVICE_DAEMON_BASE_URL: "http://agent.test", + AGENT_DEVICE_DAEMON_AUTH_TOKEN: "test", + }); }).pipe( Effect.provide( Layer.mergeAll( diff --git a/apps/server/src/device/DeviceService.ts b/apps/server/src/device/DeviceService.ts index e93bba7db5b05..b49fea080d661 100644 --- a/apps/server/src/device/DeviceService.ts +++ b/apps/server/src/device/DeviceService.ts @@ -40,6 +40,7 @@ import { import * as ServerConfig from "../config.ts"; import { agentDeviceConfigPath, + decodeAgentDeviceConfig, agentDeviceSession, writeAgentDeviceConfig, } from "./AgentDeviceTarget.ts"; @@ -1053,13 +1054,15 @@ export const make = Effect.gen(function* () { if (!(yield* fs.exists(file))) return; const entryPath = yield* installedAgentDeviceEntry(config.baseDir); if (entryPath === null) return; + const endpoint = yield* decodeAgentDeviceConfig(yield* fs.readFileString(file)); const result = yield* runner.run({ command: yield* resolveNodeExecutable("Device automation"), args: [entryPath, "close", "--config", file, "--session", session], + // agent-device 0.21.31 ignores the config file's token; the environment one authenticates. env: { ELECTRON_RUN_AS_NODE: "1", - AGENT_DEVICE_DAEMON_BASE_URL: undefined, - AGENT_DEVICE_DAEMON_AUTH_TOKEN: undefined, + AGENT_DEVICE_DAEMON_BASE_URL: endpoint.daemonBaseUrl, + AGENT_DEVICE_DAEMON_AUTH_TOKEN: endpoint.daemonAuthToken, AGENT_DEVICE_CONFIG: undefined, }, timeout: Duration.seconds(15),