From e7d82f8ae0264272485a440d29b00b3298522f6d Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Sat, 7 Nov 2026 15:49:38 +0000 Subject: [PATCH 01/12] test(server): settling a thread releases its devices (reproduces #32511) Settling a thread closes its terminals and runs the settle script, but leaves its agent-device sessions open, so the device stays claimed and other threads cannot use it. The test settles a thread and expects DeviceService.close({ threadId }) with no shutdown. It fails on main because nothing calls it on settle. AI-assisted; verified with the tests and tools listed above. --- .../ThreadSettlementService.test.ts | 35 +++++++++++++++++++ 1 file changed, 35 insertions(+) diff --git a/apps/server/src/orchestration-v2/ThreadSettlementService.test.ts b/apps/server/src/orchestration-v2/ThreadSettlementService.test.ts index f3c83eca70646..de3bb603d6ff0 100644 --- a/apps/server/src/orchestration-v2/ThreadSettlementService.test.ts +++ b/apps/server/src/orchestration-v2/ThreadSettlementService.test.ts @@ -34,6 +34,7 @@ import * as Ref from "effect/Ref"; import * as Stream from "effect/Stream"; import { TestClock } from "effect/testing"; +import * as DeviceService from "../device/DeviceService.ts"; import * as GitManager from "../git/GitManager.ts"; import * as ProjectSetupScriptRunner from "../project/ProjectSetupScriptRunner.ts"; import * as PullRequestService from "../pullRequest/PullRequestService.ts"; @@ -588,6 +589,9 @@ const makeHarness = Effect.fn("makeThreadSettlementHarness")(function* (options: >([]); const summaryRecovery = yield* Ref.make>([]); const invalidatedCwds = yield* Ref.make>([]); + const deviceCloses = yield* Ref.make< + ReadonlyArray[0]> + >([]); const domainEvents = yield* PubSub.unbounded(); const closedIdle = yield* Queue.unbounded<{ readonly threadId: string }>(); const scriptRuns = @@ -685,6 +689,9 @@ const makeHarness = Effect.fn("makeThreadSettlementHarness")(function* (options: closeOnSettle: (input) => Queue.offer(closedIdle, input).pipe(Effect.andThen(options.onCloseIdle?.() ?? Effect.void)), }), + Layer.mock(DeviceService.DeviceService)({ + close: (input) => Ref.update(deviceCloses, (closes) => [...closes, input]), + }), Layer.mock(ProjectSetupScriptRunner.ProjectSetupScriptRunner)({ runForThread: (input) => Queue.offer(scriptRuns, input).pipe( @@ -735,6 +742,7 @@ const makeHarness = Effect.fn("makeThreadSettlementHarness")(function* (options: invalidatedCwds, closedIdle, scriptRuns, + deviceCloses, publishEvent: (event: OrchestrationV2DomainEvent) => PubSub.publish(domainEvents, event), updateSettings, publishMerge: PubSub.publish(mergedPullRequests, { @@ -1478,6 +1486,33 @@ describe("ThreadSettlementServiceV2 terminals", () => { ), ); + it.effect("releases a settled thread's devices without shutting them down", () => + Effect.scoped( + Effect.gen(function* () { + const settled = appThread("device-thread", "settled"); + const marker = appThread("marker-thread", "settled"); + const fixture = yield* makeHarness({ + snapshot: makeSnapshot([]), + currentThreads: [settled, marker], + }); + + yield* Effect.gen(function* () { + const service = yield* ThreadSettlementService.ThreadSettlementServiceV2; + yield* startHarness(service, fixture.activation, fixture.snapshotReads); + yield* fixture.publishEvent(settledEvent(settled)); + yield* fixture.publishEvent(settledEvent(marker)); + assert.deepStrictEqual(yield* Queue.take(fixture.closedIdle), { threadId: settled.id }); + assert.deepStrictEqual(yield* Queue.take(fixture.closedIdle), { threadId: marker.id }); + const closes = yield* Ref.get(fixture.deviceCloses); + assert.deepStrictEqual( + closes.filter((close) => close.threadId === settled.id), + [{ threadId: settled.id }], + ); + }).pipe(Effect.provide(fixture.layer)); + }), + ), + ); + it.effect("runs the settle script in the thread's own worktree", () => Effect.scoped( Effect.gen(function* () { From 7120eec8baa348a457c44040db0929b03a96e529 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Sat, 7 Nov 2026 15:55:37 +0000 Subject: [PATCH 02/12] fix(server): settling a thread releases its devices Settling a thread closed its terminals and ran the settle script but kept its agent-device sessions and claims, so the device stayed held and other threads could not use it until device_close. cleanUpSettledThread now calls DeviceService.close({ threadId }) right after closing terminals. The device stays booted; only the thread's session and claim are released. Every settle path emits thread.settled, so this covers manual, automatic and MCP settles. A failed release (for example an SSH host that is reconnecting) is logged and the rest of the cleanup, including the settle script, still runs. Fixes pingdotgg/t3code#32511 AI-assisted; verified with the tests and tools listed above. --- .../ThreadSettlementService.test.ts | 34 ++++++++++++++++++- .../ThreadSettlementService.ts | 9 +++++ 2 files changed, 42 insertions(+), 1 deletion(-) diff --git a/apps/server/src/orchestration-v2/ThreadSettlementService.test.ts b/apps/server/src/orchestration-v2/ThreadSettlementService.test.ts index de3bb603d6ff0..96088fd82ef96 100644 --- a/apps/server/src/orchestration-v2/ThreadSettlementService.test.ts +++ b/apps/server/src/orchestration-v2/ThreadSettlementService.test.ts @@ -7,6 +7,7 @@ import { EventId, ProviderInstanceId, RuntimeRequestId, + DeviceHostUnavailableError, RunId, ThreadId, TurnItemId, @@ -563,6 +564,7 @@ interface HarnessOptions { input: ProjectSetupScriptRunner.ProjectSetupScriptRunnerInput, ) => Effect.Effect; readonly onCloseIdle?: () => Effect.Effect; + readonly deviceCloseError?: DeviceHostUnavailableError; readonly currentThreads?: ReadonlyArray; } @@ -690,7 +692,10 @@ const makeHarness = Effect.fn("makeThreadSettlementHarness")(function* (options: Queue.offer(closedIdle, input).pipe(Effect.andThen(options.onCloseIdle?.() ?? Effect.void)), }), Layer.mock(DeviceService.DeviceService)({ - close: (input) => Ref.update(deviceCloses, (closes) => [...closes, input]), + close: (input) => + Ref.update(deviceCloses, (closes) => [...closes, input]).pipe( + Effect.andThen(options.deviceCloseError ?? Effect.void), + ), }), Layer.mock(ProjectSetupScriptRunner.ProjectSetupScriptRunner)({ runForThread: (input) => @@ -1513,6 +1518,33 @@ describe("ThreadSettlementServiceV2 terminals", () => { ), ); + it.effect("runs the settle script when releasing the thread's devices fails", () => + Effect.scoped( + Effect.gen(function* () { + const settled = appThread("reconnecting-host-thread", "settled", "/worktrees/thread"); + const fixture = yield* makeHarness({ + snapshot: makeSnapshot([]), + currentThreads: [settled], + existingWorktreePaths: ["/worktrees/thread"], + deviceCloseError: new DeviceHostUnavailableError({ + hostId: "ssh-host", + reason: "Device host is reconnecting. Retry close after it reconnects.", + }), + }); + + yield* Effect.gen(function* () { + const service = yield* ThreadSettlementService.ThreadSettlementServiceV2; + yield* startHarness(service, fixture.activation, fixture.snapshotReads); + yield* fixture.publishEvent(settledEvent(settled)); + assert.deepStrictEqual(yield* Queue.take(fixture.closedIdle), { threadId: settled.id }); + const run = yield* Queue.take(fixture.scriptRuns); + assert.strictEqual(run.threadId, settled.id); + assert.deepStrictEqual(yield* Ref.get(fixture.deviceCloses), [{ threadId: settled.id }]); + }).pipe(Effect.provide(fixture.layer)); + }), + ), + ); + it.effect("runs the settle script in the thread's own worktree", () => Effect.scoped( Effect.gen(function* () { diff --git a/apps/server/src/orchestration-v2/ThreadSettlementService.ts b/apps/server/src/orchestration-v2/ThreadSettlementService.ts index 8c852426da323..caf7d1765f24e 100644 --- a/apps/server/src/orchestration-v2/ThreadSettlementService.ts +++ b/apps/server/src/orchestration-v2/ThreadSettlementService.ts @@ -25,6 +25,7 @@ import * as Semaphore from "effect/Semaphore"; import type * as Scope from "effect/Scope"; import * as Stream from "effect/Stream"; +import * as DeviceService from "../device/DeviceService.ts"; import * as GitManager from "../git/GitManager.ts"; import * as ProjectSetupScriptRunner from "../project/ProjectSetupScriptRunner.ts"; import * as PullRequestService from "../pullRequest/PullRequestService.ts"; @@ -311,6 +312,7 @@ export const make = Effect.gen(function* () { // a minute, so these threads' next lookup reads the host fresh. const shellClosedThreads = new Set(); const terminals = yield* TerminalManager.TerminalManager; + const devices = yield* DeviceService.DeviceService; const projectScripts = yield* ProjectSetupScriptRunner.ProjectSetupScriptRunner; // Repeated thread.settled retains settledAt; run the action only once per settlement. const settleActionRunAt = new Map(); @@ -602,6 +604,13 @@ export const make = Effect.gen(function* () { const settled = yield* projections.getThread(threadId); if (settled.settledOverride !== "settled") return; yield* terminals.closeOnSettle({ threadId }); + yield* devices + .close({ threadId }) + .pipe( + Effect.catch((error) => + Effect.logWarning("releasing a settled thread's devices failed", { threadId, error }), + ), + ); const worktreePath = settled.worktreePath; if (worktreePath === null || !(yield* fileSystem.exists(worktreePath))) return; // Recheck after closing and worktree I/O; re-engaged threads must not be cleaned up. From d630aded41286864f0e90664478eafadeb79732b Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Sat, 7 Nov 2026 16:03:13 +0000 Subject: [PATCH 03/12] fix(server): keep a re-engaged thread's devices and skip the device lock without sessions Settle cleanup rechecks that the thread is still settled after its terminals close, so a thread reopened in that window keeps its device. Threads with no device sessions no longer take the device lifecycle lock or publish device state on settle. AI-assisted; verified with the tests and tools listed above. --- .../ThreadSettlementService.test.ts | 14 ++++++++++++++ .../ThreadSettlementService.ts | 18 +++++++++++------- 2 files changed, 25 insertions(+), 7 deletions(-) diff --git a/apps/server/src/orchestration-v2/ThreadSettlementService.test.ts b/apps/server/src/orchestration-v2/ThreadSettlementService.test.ts index 96088fd82ef96..cc1c1f1495966 100644 --- a/apps/server/src/orchestration-v2/ThreadSettlementService.test.ts +++ b/apps/server/src/orchestration-v2/ThreadSettlementService.test.ts @@ -692,6 +692,16 @@ const makeHarness = Effect.fn("makeThreadSettlementHarness")(function* (options: Queue.offer(closedIdle, input).pipe(Effect.andThen(options.onCloseIdle?.() ?? Effect.void)), }), Layer.mock(DeviceService.DeviceService)({ + sessionsForThread: (threadId) => + Effect.succeed([ + { + threadId, + hostId: "local", + deviceId: "emulator-5554", + platform: "android" as const, + openedAt: "2026-01-01T00:00:00.000Z", + }, + ]), close: (input) => Ref.update(deviceCloses, (closes) => [...closes, input]).pipe( Effect.andThen(options.deviceCloseError ?? Effect.void), @@ -1594,6 +1604,10 @@ describe("ThreadSettlementServiceV2 terminals", () => { yield* fixture.publishEvent(settledEvent(settled)); yield* fixture.publishEvent(settledEvent(marker)); assert.strictEqual((yield* Queue.take(fixture.scriptRuns)).threadId, marker.id); + assert.deepStrictEqual( + (yield* Ref.get(fixture.deviceCloses)).filter((close) => close.threadId === settled.id), + [], + ); }).pipe(Effect.provide(fixture.layer)); }), ), diff --git a/apps/server/src/orchestration-v2/ThreadSettlementService.ts b/apps/server/src/orchestration-v2/ThreadSettlementService.ts index caf7d1765f24e..da15b48a0865f 100644 --- a/apps/server/src/orchestration-v2/ThreadSettlementService.ts +++ b/apps/server/src/orchestration-v2/ThreadSettlementService.ts @@ -604,13 +604,17 @@ export const make = Effect.gen(function* () { const settled = yield* projections.getThread(threadId); if (settled.settledOverride !== "settled") return; yield* terminals.closeOnSettle({ threadId }); - yield* devices - .close({ threadId }) - .pipe( - Effect.catch((error) => - Effect.logWarning("releasing a settled thread's devices failed", { threadId, error }), - ), - ); + if ((yield* projections.getThread(threadId)).settledOverride !== "settled") return; + // close takes the device lifecycle lock, which a starting host holds; skip it without sessions. + if ((yield* devices.sessionsForThread(threadId)).length > 0) { + yield* devices + .close({ threadId }) + .pipe( + Effect.catch((error) => + Effect.logWarning("releasing a settled thread's devices failed", { threadId, error }), + ), + ); + } const worktreePath = settled.worktreePath; if (worktreePath === null || !(yield* fileSystem.exists(worktreePath))) return; // Recheck after closing and worktree I/O; re-engaged threads must not be cleaned up. From 98619b3ddf66431cf20e849d1bf1db4c357c0e8f Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Sat, 7 Nov 2026 16:08:19 +0000 Subject: [PATCH 04/12] test(server): a settled thread without devices leaves the device service alone AI-assisted; verified with the tests and tools listed above. --- .../ThreadSettlementService.test.ts | 45 +++++++++++++++---- 1 file changed, 36 insertions(+), 9 deletions(-) diff --git a/apps/server/src/orchestration-v2/ThreadSettlementService.test.ts b/apps/server/src/orchestration-v2/ThreadSettlementService.test.ts index cc1c1f1495966..23f445af4d0b4 100644 --- a/apps/server/src/orchestration-v2/ThreadSettlementService.test.ts +++ b/apps/server/src/orchestration-v2/ThreadSettlementService.test.ts @@ -565,6 +565,7 @@ interface HarnessOptions { ) => Effect.Effect; readonly onCloseIdle?: () => Effect.Effect; readonly deviceCloseError?: DeviceHostUnavailableError; + readonly withoutDeviceSessions?: boolean; readonly currentThreads?: ReadonlyArray; } @@ -693,15 +694,19 @@ const makeHarness = Effect.fn("makeThreadSettlementHarness")(function* (options: }), Layer.mock(DeviceService.DeviceService)({ sessionsForThread: (threadId) => - Effect.succeed([ - { - threadId, - hostId: "local", - deviceId: "emulator-5554", - platform: "android" as const, - openedAt: "2026-01-01T00:00:00.000Z", - }, - ]), + Effect.succeed( + options.withoutDeviceSessions + ? [] + : [ + { + threadId, + hostId: "local", + deviceId: "emulator-5554", + platform: "android" as const, + openedAt: "2026-01-01T00:00:00.000Z", + }, + ], + ), close: (input) => Ref.update(deviceCloses, (closes) => [...closes, input]).pipe( Effect.andThen(options.deviceCloseError ?? Effect.void), @@ -1555,6 +1560,28 @@ describe("ThreadSettlementServiceV2 terminals", () => { ), ); + it.effect("leaves the device service alone for a settled thread without devices", () => + Effect.scoped( + Effect.gen(function* () { + const settled = appThread("deviceless-thread", "settled", "/worktrees/deviceless"); + const fixture = yield* makeHarness({ + snapshot: makeSnapshot([]), + currentThreads: [settled], + existingWorktreePaths: ["/worktrees/deviceless"], + withoutDeviceSessions: true, + }); + + yield* Effect.gen(function* () { + const service = yield* ThreadSettlementService.ThreadSettlementServiceV2; + yield* startHarness(service, fixture.activation, fixture.snapshotReads); + yield* fixture.publishEvent(settledEvent(settled)); + assert.strictEqual((yield* Queue.take(fixture.scriptRuns)).threadId, settled.id); + assert.deepStrictEqual(yield* Ref.get(fixture.deviceCloses), []); + }).pipe(Effect.provide(fixture.layer)); + }), + ), + ); + it.effect("runs the settle script in the thread's own worktree", () => Effect.scoped( Effect.gen(function* () { From 2a6434c47581b0648b78324ae39a00055bdfa4f8 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Sat, 7 Nov 2026 16:11:42 +0000 Subject: [PATCH 05/12] fix(server): release each settled device separately An unreachable SSH device host no longer keeps a settled thread's other devices claimed: each device is closed on its own and a failure is logged. AI-assisted; verified with the tests and tools listed above. --- .../ThreadSettlementService.test.ts | 42 +++++++++++-------- .../ThreadSettlementService.ts | 23 ++++++---- 2 files changed, 40 insertions(+), 25 deletions(-) diff --git a/apps/server/src/orchestration-v2/ThreadSettlementService.test.ts b/apps/server/src/orchestration-v2/ThreadSettlementService.test.ts index 23f445af4d0b4..2a6a8a82f001d 100644 --- a/apps/server/src/orchestration-v2/ThreadSettlementService.test.ts +++ b/apps/server/src/orchestration-v2/ThreadSettlementService.test.ts @@ -565,7 +565,7 @@ interface HarnessOptions { ) => Effect.Effect; readonly onCloseIdle?: () => Effect.Effect; readonly deviceCloseError?: DeviceHostUnavailableError; - readonly withoutDeviceSessions?: boolean; + readonly deviceSessions?: ReadonlyArray<{ readonly hostId: string; readonly deviceId: string }>; readonly currentThreads?: ReadonlyArray; } @@ -695,21 +695,22 @@ const makeHarness = Effect.fn("makeThreadSettlementHarness")(function* (options: Layer.mock(DeviceService.DeviceService)({ sessionsForThread: (threadId) => Effect.succeed( - options.withoutDeviceSessions - ? [] - : [ - { - threadId, - hostId: "local", - deviceId: "emulator-5554", - platform: "android" as const, - openedAt: "2026-01-01T00:00:00.000Z", - }, - ], + (options.deviceSessions ?? [{ hostId: "local", deviceId: "emulator-5554" }]).map( + (session) => ({ + ...session, + threadId, + platform: "android" as const, + openedAt: "2026-01-01T00:00:00.000Z", + }), + ), ), close: (input) => Ref.update(deviceCloses, (closes) => [...closes, input]).pipe( - Effect.andThen(options.deviceCloseError ?? Effect.void), + Effect.andThen( + options.deviceCloseError && options.deviceCloseError.hostId === input.hostId + ? Effect.fail(options.deviceCloseError) + : Effect.void, + ), ), }), Layer.mock(ProjectSetupScriptRunner.ProjectSetupScriptRunner)({ @@ -1526,14 +1527,14 @@ describe("ThreadSettlementServiceV2 terminals", () => { const closes = yield* Ref.get(fixture.deviceCloses); assert.deepStrictEqual( closes.filter((close) => close.threadId === settled.id), - [{ threadId: settled.id }], + [{ threadId: settled.id, hostId: "local", deviceId: "emulator-5554" }], ); }).pipe(Effect.provide(fixture.layer)); }), ), ); - it.effect("runs the settle script when releasing the thread's devices fails", () => + it.effect("releases the other devices and runs the settle script when one release fails", () => Effect.scoped( Effect.gen(function* () { const settled = appThread("reconnecting-host-thread", "settled", "/worktrees/thread"); @@ -1541,6 +1542,10 @@ describe("ThreadSettlementServiceV2 terminals", () => { snapshot: makeSnapshot([]), currentThreads: [settled], existingWorktreePaths: ["/worktrees/thread"], + deviceSessions: [ + { hostId: "ssh-host", deviceId: "remote-pixel" }, + { hostId: "local", deviceId: "emulator-5554" }, + ], deviceCloseError: new DeviceHostUnavailableError({ hostId: "ssh-host", reason: "Device host is reconnecting. Retry close after it reconnects.", @@ -1554,7 +1559,10 @@ describe("ThreadSettlementServiceV2 terminals", () => { assert.deepStrictEqual(yield* Queue.take(fixture.closedIdle), { threadId: settled.id }); const run = yield* Queue.take(fixture.scriptRuns); assert.strictEqual(run.threadId, settled.id); - assert.deepStrictEqual(yield* Ref.get(fixture.deviceCloses), [{ threadId: settled.id }]); + assert.deepStrictEqual(yield* Ref.get(fixture.deviceCloses), [ + { threadId: settled.id, hostId: "ssh-host", deviceId: "remote-pixel" }, + { threadId: settled.id, hostId: "local", deviceId: "emulator-5554" }, + ]); }).pipe(Effect.provide(fixture.layer)); }), ), @@ -1568,7 +1576,7 @@ describe("ThreadSettlementServiceV2 terminals", () => { snapshot: makeSnapshot([]), currentThreads: [settled], existingWorktreePaths: ["/worktrees/deviceless"], - withoutDeviceSessions: true, + deviceSessions: [], }); yield* Effect.gen(function* () { diff --git a/apps/server/src/orchestration-v2/ThreadSettlementService.ts b/apps/server/src/orchestration-v2/ThreadSettlementService.ts index da15b48a0865f..e9de64774dfab 100644 --- a/apps/server/src/orchestration-v2/ThreadSettlementService.ts +++ b/apps/server/src/orchestration-v2/ThreadSettlementService.ts @@ -605,16 +605,23 @@ export const make = Effect.gen(function* () { if (settled.settledOverride !== "settled") return; yield* terminals.closeOnSettle({ threadId }); if ((yield* projections.getThread(threadId)).settledOverride !== "settled") return; - // close takes the device lifecycle lock, which a starting host holds; skip it without sessions. - if ((yield* devices.sessionsForThread(threadId)).length > 0) { - yield* devices - .close({ threadId }) - .pipe( + // Per device: a thread without devices skips the lock a starting host holds, and one + // unreachable host leaves no other device claimed. + yield* Effect.forEach( + yield* devices.sessionsForThread(threadId), + ({ hostId, deviceId }) => + devices.close({ threadId, hostId, deviceId }).pipe( Effect.catch((error) => - Effect.logWarning("releasing a settled thread's devices failed", { threadId, error }), + Effect.logWarning("releasing a settled thread's device failed", { + threadId, + hostId, + deviceId, + error, + }), ), - ); - } + ), + { discard: true }, + ); const worktreePath = settled.worktreePath; if (worktreePath === null || !(yield* fileSystem.exists(worktreePath))) return; // Recheck after closing and worktree I/O; re-engaged threads must not be cleaned up. From 7c6f7d418eae381c3f4bf130f95b4cecf1ab270d Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Sat, 7 Nov 2026 16:19:36 +0000 Subject: [PATCH 06/12] fix(server): stop releasing devices once a settled thread re-engages AI-assisted; verified with the tests and tools listed above. --- .../ThreadSettlementService.test.ts | 44 +++++++++++++++++++ .../ThreadSettlementService.ts | 28 ++++++------ 2 files changed, 59 insertions(+), 13 deletions(-) diff --git a/apps/server/src/orchestration-v2/ThreadSettlementService.test.ts b/apps/server/src/orchestration-v2/ThreadSettlementService.test.ts index 2a6a8a82f001d..6ee806d5d1809 100644 --- a/apps/server/src/orchestration-v2/ThreadSettlementService.test.ts +++ b/apps/server/src/orchestration-v2/ThreadSettlementService.test.ts @@ -564,6 +564,7 @@ interface HarnessOptions { input: ProjectSetupScriptRunner.ProjectSetupScriptRunnerInput, ) => Effect.Effect; readonly onCloseIdle?: () => Effect.Effect; + readonly onDeviceClose?: (hostId: string | undefined) => Effect.Effect; readonly deviceCloseError?: DeviceHostUnavailableError; readonly deviceSessions?: ReadonlyArray<{ readonly hostId: string; readonly deviceId: string }>; readonly currentThreads?: ReadonlyArray; @@ -706,6 +707,7 @@ const makeHarness = Effect.fn("makeThreadSettlementHarness")(function* (options: ), close: (input) => Ref.update(deviceCloses, (closes) => [...closes, input]).pipe( + Effect.andThen(options.onDeviceClose?.(input.hostId) ?? Effect.void), Effect.andThen( options.deviceCloseError && options.deviceCloseError.hostId === input.hostId ? Effect.fail(options.deviceCloseError) @@ -1507,6 +1509,48 @@ describe("ThreadSettlementServiceV2 terminals", () => { ), ); + it.effect("keeps the remaining devices of a thread re-engaged during an earlier release", () => + Effect.scoped( + Effect.gen(function* () { + const firstStarted = yield* Deferred.make(); + const firstFinished = yield* Deferred.make(); + const settled = appThread("multi-device-thread", "settled"); + const marker = appThread("marker-thread", "settled"); + const currentThreads = [settled, marker]; + const fixture = yield* makeHarness({ + snapshot: makeSnapshot([]), + currentThreads, + deviceSessions: [ + { hostId: "slow-ssh", deviceId: "pixel" }, + { hostId: "local", deviceId: "emulator-5554" }, + ], + onDeviceClose: (hostId) => + hostId === "slow-ssh" + ? Deferred.succeed(firstStarted, undefined).pipe( + Effect.andThen(Deferred.await(firstFinished)), + ) + : Effect.void, + }); + + yield* Effect.gen(function* () { + const service = yield* ThreadSettlementService.ThreadSettlementServiceV2; + yield* startHarness(service, fixture.activation, fixture.snapshotReads); + yield* fixture.publishEvent(settledEvent(settled)); + yield* Deferred.await(firstStarted); + currentThreads[0] = { ...settled, settledOverride: "active" }; + yield* Deferred.succeed(firstFinished, undefined); + yield* fixture.publishEvent(settledEvent(marker)); + assert.deepStrictEqual(yield* Queue.take(fixture.closedIdle), { threadId: settled.id }); + assert.deepStrictEqual(yield* Queue.take(fixture.closedIdle), { threadId: marker.id }); + assert.deepStrictEqual( + (yield* Ref.get(fixture.deviceCloses)).filter((close) => close.threadId === settled.id), + [{ threadId: settled.id, hostId: "slow-ssh", deviceId: "pixel" }], + ); + }).pipe(Effect.provide(fixture.layer)); + }), + ), + ); + it.effect("releases a settled thread's devices without shutting them down", () => Effect.scoped( Effect.gen(function* () { diff --git a/apps/server/src/orchestration-v2/ThreadSettlementService.ts b/apps/server/src/orchestration-v2/ThreadSettlementService.ts index e9de64774dfab..96deae2b4f8d7 100644 --- a/apps/server/src/orchestration-v2/ThreadSettlementService.ts +++ b/apps/server/src/orchestration-v2/ThreadSettlementService.ts @@ -604,22 +604,24 @@ export const make = Effect.gen(function* () { const settled = yield* projections.getThread(threadId); if (settled.settledOverride !== "settled") return; yield* terminals.closeOnSettle({ threadId }); - if ((yield* projections.getThread(threadId)).settledOverride !== "settled") return; - // Per device: a thread without devices skips the lock a starting host holds, and one - // unreachable host leaves no other device claimed. + // Per device: a thread without devices skips the lock a starting host holds, one unreachable + // host leaves no other device claimed, and a thread re-engaged meanwhile keeps the rest. yield* Effect.forEach( yield* devices.sessionsForThread(threadId), ({ hostId, deviceId }) => - devices.close({ threadId, hostId, deviceId }).pipe( - Effect.catch((error) => - Effect.logWarning("releasing a settled thread's device failed", { - threadId, - hostId, - deviceId, - error, - }), - ), - ), + Effect.gen(function* () { + if ((yield* projections.getThread(threadId)).settledOverride !== "settled") return; + yield* devices.close({ threadId, hostId, deviceId }).pipe( + Effect.catch((error) => + Effect.logWarning("releasing a settled thread's device failed", { + threadId, + hostId, + deviceId, + error, + }), + ), + ); + }), { discard: true }, ); const worktreePath = settled.worktreePath; From 2e051a539b19b76ca746612fe716ae27c19d7b7a Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Tue, 10 Nov 2026 14:41:11 +0000 Subject: [PATCH 07/12] fix(server): device releases no longer stall other threads' settle cleanup Releasing a settled thread's devices waits on the device lifecycle lock, which a starting SSH host can hold for minutes. It ran inline in the serial domain-event loop, so every later thread.settled waited behind it. Releases now run in a scoped FiberSet beside the event loop, and drain awaits them. The settle script no longer waits on device releases. AI-assisted; verified with the tests and tools listed above. --- .../ThreadSettlementService.test.ts | 41 +++++++++++++++++++ .../ThreadSettlementService.ts | 41 +++++++++++++++---- 2 files changed, 74 insertions(+), 8 deletions(-) diff --git a/apps/server/src/orchestration-v2/ThreadSettlementService.test.ts b/apps/server/src/orchestration-v2/ThreadSettlementService.test.ts index 6ee806d5d1809..694808774413f 100644 --- a/apps/server/src/orchestration-v2/ThreadSettlementService.test.ts +++ b/apps/server/src/orchestration-v2/ThreadSettlementService.test.ts @@ -1542,6 +1542,7 @@ describe("ThreadSettlementServiceV2 terminals", () => { yield* fixture.publishEvent(settledEvent(marker)); assert.deepStrictEqual(yield* Queue.take(fixture.closedIdle), { threadId: settled.id }); assert.deepStrictEqual(yield* Queue.take(fixture.closedIdle), { threadId: marker.id }); + yield* service.drain; assert.deepStrictEqual( (yield* Ref.get(fixture.deviceCloses)).filter((close) => close.threadId === settled.id), [{ threadId: settled.id, hostId: "slow-ssh", deviceId: "pixel" }], @@ -1551,6 +1552,42 @@ describe("ThreadSettlementServiceV2 terminals", () => { ), ); + it.effect("closes another thread's terminals while a device release waits on the lock", () => + Effect.scoped( + Effect.gen(function* () { + const releaseStarted = yield* Deferred.make(); + const lockFreed = yield* Deferred.make(); + const waiting = appThread("waiting-device-thread", "settled"); + const other = appThread("other-thread", "settled"); + const fixture = yield* makeHarness({ + snapshot: makeSnapshot([]), + currentThreads: [waiting, other], + deviceSessions: [{ hostId: "starting-ssh", deviceId: "pixel" }], + onDeviceClose: () => + Deferred.succeed(releaseStarted, undefined).pipe( + Effect.andThen(Deferred.await(lockFreed)), + ), + }); + + yield* Effect.gen(function* () { + const service = yield* ThreadSettlementService.ThreadSettlementServiceV2; + yield* startHarness(service, fixture.activation, fixture.snapshotReads); + yield* fixture.publishEvent(settledEvent(waiting)); + assert.deepStrictEqual(yield* Queue.take(fixture.closedIdle), { threadId: waiting.id }); + yield* Deferred.await(releaseStarted); + yield* fixture.publishEvent(settledEvent(other)); + assert.deepStrictEqual(yield* Queue.take(fixture.closedIdle), { threadId: other.id }); + yield* Deferred.succeed(lockFreed, undefined); + yield* service.drain; + assert.deepStrictEqual( + (yield* Ref.get(fixture.deviceCloses)).map((close) => close.threadId), + [waiting.id, other.id], + ); + }).pipe(Effect.provide(fixture.layer)); + }), + ), + ); + it.effect("releases a settled thread's devices without shutting them down", () => Effect.scoped( Effect.gen(function* () { @@ -1568,6 +1605,7 @@ describe("ThreadSettlementServiceV2 terminals", () => { yield* fixture.publishEvent(settledEvent(marker)); assert.deepStrictEqual(yield* Queue.take(fixture.closedIdle), { threadId: settled.id }); assert.deepStrictEqual(yield* Queue.take(fixture.closedIdle), { threadId: marker.id }); + yield* service.drain; const closes = yield* Ref.get(fixture.deviceCloses); assert.deepStrictEqual( closes.filter((close) => close.threadId === settled.id), @@ -1603,6 +1641,7 @@ describe("ThreadSettlementServiceV2 terminals", () => { assert.deepStrictEqual(yield* Queue.take(fixture.closedIdle), { threadId: settled.id }); const run = yield* Queue.take(fixture.scriptRuns); assert.strictEqual(run.threadId, settled.id); + yield* service.drain; assert.deepStrictEqual(yield* Ref.get(fixture.deviceCloses), [ { threadId: settled.id, hostId: "ssh-host", deviceId: "remote-pixel" }, { threadId: settled.id, hostId: "local", deviceId: "emulator-5554" }, @@ -1628,6 +1667,7 @@ describe("ThreadSettlementServiceV2 terminals", () => { yield* startHarness(service, fixture.activation, fixture.snapshotReads); yield* fixture.publishEvent(settledEvent(settled)); assert.strictEqual((yield* Queue.take(fixture.scriptRuns)).threadId, settled.id); + yield* service.drain; assert.deepStrictEqual(yield* Ref.get(fixture.deviceCloses), []); }).pipe(Effect.provide(fixture.layer)); }), @@ -1683,6 +1723,7 @@ describe("ThreadSettlementServiceV2 terminals", () => { yield* fixture.publishEvent(settledEvent(settled)); yield* fixture.publishEvent(settledEvent(marker)); assert.strictEqual((yield* Queue.take(fixture.scriptRuns)).threadId, marker.id); + yield* service.drain; assert.deepStrictEqual( (yield* Ref.get(fixture.deviceCloses)).filter((close) => close.threadId === settled.id), [], diff --git a/apps/server/src/orchestration-v2/ThreadSettlementService.ts b/apps/server/src/orchestration-v2/ThreadSettlementService.ts index 96deae2b4f8d7..5441e0896792b 100644 --- a/apps/server/src/orchestration-v2/ThreadSettlementService.ts +++ b/apps/server/src/orchestration-v2/ThreadSettlementService.ts @@ -598,14 +598,12 @@ export const make = Effect.gen(function* () { return sweepThread(threadId).pipe(FiberSet.run(threadSweeps), Effect.asVoid); }); - // Skip settle scripts in shared checkouts where other threads may still work. - const cleanUpSettledThread = Effect.fn("ThreadSettlementServiceV2.cleanUpSettledThread")( + // Per device: a thread without devices skips the lock a starting host holds, one unreachable + // host leaves no other device claimed, and a thread re-engaged meanwhile keeps the rest. + const releaseSettledThreadDevices = Effect.fn( + "ThreadSettlementServiceV2.releaseSettledThreadDevices", + )( function* (threadId: ThreadId) { - const settled = yield* projections.getThread(threadId); - if (settled.settledOverride !== "settled") return; - yield* terminals.closeOnSettle({ threadId }); - // Per device: a thread without devices skips the lock a starting host holds, one unreachable - // host leaves no other device claimed, and a thread re-engaged meanwhile keeps the rest. yield* Effect.forEach( yield* devices.sessionsForThread(threadId), ({ hostId, deviceId }) => @@ -624,6 +622,30 @@ export const make = Effect.gen(function* () { }), { discard: true }, ); + }, + (effect, threadId) => + effect.pipe( + Effect.catchCauseIf( + (cause) => !Cause.hasInterruptsOnly(cause), + (cause) => + Effect.logWarning("releasing a settled thread's devices failed", { + threadId, + cause: Cause.pretty(cause), + }), + ), + ), + ); + // Releases wait on the device lock a starting host can hold for minutes, so they run beside + // the event loop instead of stalling every other thread's settle cleanup. + const deviceReleases = yield* FiberSet.make(); + + // Skip settle scripts in shared checkouts where other threads may still work. + const cleanUpSettledThread = Effect.fn("ThreadSettlementServiceV2.cleanUpSettledThread")( + function* (threadId: ThreadId) { + const settled = yield* projections.getThread(threadId); + if (settled.settledOverride !== "settled") return; + yield* terminals.closeOnSettle({ threadId }); + yield* releaseSettledThreadDevices(threadId).pipe(FiberSet.run(deviceReleases)); const worktreePath = settled.worktreePath; if (worktreePath === null || !(yield* fileSystem.exists(worktreePath))) return; // Recheck after closing and worktree I/O; re-engaged threads must not be cleaned up. @@ -740,6 +762,9 @@ export const make = Effect.gen(function* () { return { start, - drain: Effect.andThen(fullSweeps.drain, FiberSet.awaitEmpty(threadSweeps)), + drain: fullSweeps.drain.pipe( + Effect.andThen(FiberSet.awaitEmpty(threadSweeps)), + Effect.andThen(FiberSet.awaitEmpty(deviceReleases)), + ), } satisfies ThreadSettlementServiceV2["Service"]; }); From 45db81569d379dd042036095432c640f8e5ba46d Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Tue, 10 Nov 2026 14:59:02 +0000 Subject: [PATCH 08/12] fix(server): a thread re-engaged while its device release waits keeps the device The settled check for a device release ran before DeviceService.close, which then waited on the lifecycle lock a starting host can hold for minutes. A thread reopened during that wait still lost its device. DeviceService.close takes an optional server-internal `when` guard that runs under the lifecycle lock; false releases nothing. The settlement release passes its still-settled check as that guard. AI-assisted; verified with the tests and tools listed above. --- apps/server/src/device/DeviceService.test.ts | 8 +++++ apps/server/src/device/DeviceService.ts | 13 +++++-- .../ThreadSettlementService.test.ts | 35 ++++++++++--------- .../ThreadSettlementService.ts | 31 ++++++++-------- 4 files changed, 53 insertions(+), 34 deletions(-) diff --git a/apps/server/src/device/DeviceService.test.ts b/apps/server/src/device/DeviceService.test.ts index a34dbc534f59a..59faba44064d6 100644 --- a/apps/server/src/device/DeviceService.test.ts +++ b/apps/server/src/device/DeviceService.test.ts @@ -1776,6 +1776,14 @@ describe("agent session release on close", () => { const other = ThreadId.make("other"); const session = yield* openClaim(service, threadId); yield* openClaim(service, other, session.deviceId); + yield* service.close( + { threadId, deviceId: session.deviceId }, + { when: Effect.succeed(false) }, + ); + expect((yield* service.state).sessions.map((entry) => entry.threadId)).toEqual([ + threadId, + other, + ]); yield* service.close({ threadId, deviceId: session.deviceId }); yield* service.close({ threadId, deviceId: session.deviceId }); const name = yield* agentDeviceSession(threadId, session.hostId, session.deviceId).pipe( diff --git a/apps/server/src/device/DeviceService.ts b/apps/server/src/device/DeviceService.ts index 790daabe7a445..e85108b4273c4 100644 --- a/apps/server/src/device/DeviceService.ts +++ b/apps/server/src/device/DeviceService.ts @@ -117,6 +117,11 @@ export interface DeviceReadiness extends DeviceHost.DeviceHostReady { readonly hostId: DeviceHostId; } +export interface DeviceCloseOptions { + /** Runs under the lifecycle lock; false releases nothing. */ + readonly when?: Effect.Effect; +} + export interface DeviceAgentReadiness extends DeviceReadiness { /** Null when each device has its own daemon; see `DeviceOperations.agentEndpoint`. */ readonly agentDevice: DeviceHost.AgentDeviceEndpoint | null; @@ -197,7 +202,10 @@ export class DeviceService extends Context.Service< input: DeviceOpenInput, context?: DeviceHost.DeviceOpenContext, ) => Effect.Effect; - readonly close: (input: DeviceCloseInput) => Effect.Effect; + readonly close: ( + input: DeviceCloseInput, + options?: DeviceCloseOptions, + ) => Effect.Effect; readonly shutdown: (input: DeviceShutdownInput) => Effect.Effect; readonly detail: (input: DeviceDetailInput) => Effect.Effect; readonly action: (input: DeviceActionInput) => Effect.Effect; @@ -897,10 +905,11 @@ export const makeWithHosts = Effect.fn("DeviceService.makeWithHosts")(function* }); const close: DeviceService["Service"]["close"] = Effect.fn("DeviceService.close")( - function* (input) { + function* (input, options) { // Open publishes under the same lock; an old close must not release a reopened claim. const closing = yield* lifecycleLock.withPermit( Effect.gen(function* () { + if (options?.when && !(yield* options.when)) return []; const { state } = yield* SynchronizedRef.get(stateRef); const closing = state.sessions.filter( (session) => diff --git a/apps/server/src/orchestration-v2/ThreadSettlementService.test.ts b/apps/server/src/orchestration-v2/ThreadSettlementService.test.ts index 694808774413f..08aa8ab12d335 100644 --- a/apps/server/src/orchestration-v2/ThreadSettlementService.test.ts +++ b/apps/server/src/orchestration-v2/ThreadSettlementService.test.ts @@ -678,12 +678,13 @@ const makeHarness = Effect.fn("makeThreadSettlementHarness")(function* (options: : snapshot.threads.filter((thread) => thread.id === threadId), ), ), - getThread: (threadId) => { - const thread = options.currentThreads?.find((candidate) => candidate.id === threadId); - return thread - ? Effect.succeed(thread) - : Effect.die(new Error(`Unexpected thread read: ${threadId}`)); - }, + getThread: (threadId) => + Effect.suspend(() => { + const thread = options.currentThreads?.find((candidate) => candidate.id === threadId); + return thread + ? Effect.succeed(thread) + : Effect.die(new Error(`Unexpected thread read: ${threadId}`)); + }), }), Layer.mock(Orchestrator.OrchestratorV2)({ streamDomainEvents: Stream.fromPubSub(domainEvents), @@ -705,15 +706,15 @@ const makeHarness = Effect.fn("makeThreadSettlementHarness")(function* (options: }), ), ), - close: (input) => - Ref.update(deviceCloses, (closes) => [...closes, input]).pipe( - Effect.andThen(options.onDeviceClose?.(input.hostId) ?? Effect.void), - Effect.andThen( - options.deviceCloseError && options.deviceCloseError.hostId === input.hostId - ? Effect.fail(options.deviceCloseError) - : Effect.void, - ), - ), + // onDeviceClose stands in for the lifecycle lock wait; `when` runs once the lock is held. + close: (input, closeOptions) => + Effect.gen(function* () { + yield* options.onDeviceClose?.(input.hostId) ?? Effect.void; + if (closeOptions?.when && !(yield* closeOptions.when)) return; + yield* Ref.update(deviceCloses, (closes) => [...closes, input]); + const error = options.deviceCloseError; + if (error && error.hostId === input.hostId) return yield* error; + }), }), Layer.mock(ProjectSetupScriptRunner.ProjectSetupScriptRunner)({ runForThread: (input) => @@ -1509,7 +1510,7 @@ describe("ThreadSettlementServiceV2 terminals", () => { ), ); - it.effect("keeps the remaining devices of a thread re-engaged during an earlier release", () => + it.effect("keeps the devices of a thread re-engaged while a release waits on the lock", () => Effect.scoped( Effect.gen(function* () { const firstStarted = yield* Deferred.make(); @@ -1545,7 +1546,7 @@ describe("ThreadSettlementServiceV2 terminals", () => { yield* service.drain; assert.deepStrictEqual( (yield* Ref.get(fixture.deviceCloses)).filter((close) => close.threadId === settled.id), - [{ threadId: settled.id, hostId: "slow-ssh", deviceId: "pixel" }], + [], ); }).pipe(Effect.provide(fixture.layer)); }), diff --git a/apps/server/src/orchestration-v2/ThreadSettlementService.ts b/apps/server/src/orchestration-v2/ThreadSettlementService.ts index 5441e0896792b..18e793f19581c 100644 --- a/apps/server/src/orchestration-v2/ThreadSettlementService.ts +++ b/apps/server/src/orchestration-v2/ThreadSettlementService.ts @@ -598,28 +598,29 @@ export const make = Effect.gen(function* () { return sweepThread(threadId).pipe(FiberSet.run(threadSweeps), Effect.asVoid); }); - // Per device: a thread without devices skips the lock a starting host holds, one unreachable - // host leaves no other device claimed, and a thread re-engaged meanwhile keeps the rest. + // Per device: a thread without devices skips the lock a starting host holds, and one + // unreachable host leaves no other device claimed. The settled check runs under that lock, so a + // thread re-engaged while a release waits keeps its devices. const releaseSettledThreadDevices = Effect.fn( "ThreadSettlementServiceV2.releaseSettledThreadDevices", )( function* (threadId: ThreadId) { + const stillSettled = projections + .getThread(threadId) + .pipe(Effect.map((thread) => thread.settledOverride === "settled")); yield* Effect.forEach( yield* devices.sessionsForThread(threadId), ({ hostId, deviceId }) => - Effect.gen(function* () { - if ((yield* projections.getThread(threadId)).settledOverride !== "settled") return; - yield* devices.close({ threadId, hostId, deviceId }).pipe( - Effect.catch((error) => - Effect.logWarning("releasing a settled thread's device failed", { - threadId, - hostId, - deviceId, - error, - }), - ), - ); - }), + devices.close({ threadId, hostId, deviceId }, { when: stillSettled }).pipe( + Effect.catch((error) => + Effect.logWarning("releasing a settled thread's device failed", { + threadId, + hostId, + deviceId, + error, + }), + ), + ), { discard: true }, ); }, From d2bf0a98da5ad1cceac070f86c12c944067975f6 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Tue, 10 Nov 2026 15:35:23 +0000 Subject: [PATCH 09/12] docs: settling a thread releases its devices Co-Authored-By: Claude Opus 5.5 --- docs/user/thread-sidebar.md | 2 ++ 1 file changed, 2 insertions(+) diff --git a/docs/user/thread-sidebar.md b/docs/user/thread-sidebar.md index 09ce63e2a3e65..df437306bf1d2 100644 --- a/docs/user/thread-sidebar.md +++ b/docs/user/thread-sidebar.md @@ -172,6 +172,8 @@ when the thread settles** enabled in Settings > Projects > Actions. You can also `"stopOnSettle": true` on a `t3.json` action and import it there. Start the command through that action so T3 can stop its terminal and child processes. Commands started through an agent's own shell tools are managed by the provider. +Settling also releases the devices the thread's agent opened, so another thread can use +them. The devices stay booted. To reclaim disk space from settled work, turn on **Run in the thread's worktree when the thread settles** for one of the project's actions, or set `"runOnSettle": true` on a From a45db96e1f154c64d2fc070980c2f4833147c5c9 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Tue, 10 Nov 2026 15:50:40 +0000 Subject: [PATCH 10/12] test(server): check the close guard after a real lifecycle lock wait Co-Authored-By: Claude Opus 5.5 --- apps/server/src/device/DeviceService.test.ts | 20 ++++++++++++++++---- 1 file changed, 16 insertions(+), 4 deletions(-) diff --git a/apps/server/src/device/DeviceService.test.ts b/apps/server/src/device/DeviceService.test.ts index 59faba44064d6..08ee8e22af1ae 100644 --- a/apps/server/src/device/DeviceService.test.ts +++ b/apps/server/src/device/DeviceService.test.ts @@ -1776,10 +1776,22 @@ describe("agent session release on close", () => { const other = ThreadId.make("other"); const session = yield* openClaim(service, threadId); yield* openClaim(service, other, session.deviceId); - yield* service.close( - { threadId, deviceId: session.deviceId }, - { when: Effect.succeed(false) }, - ); + const locked = yield* Deferred.make(); + const unlock = yield* Deferred.make(); + const holder = yield* service + .withLifecycleLock( + Deferred.succeed(locked, undefined).pipe(Effect.andThen(Deferred.await(unlock))), + ) + .pipe(Effect.forkChild); + yield* Deferred.await(locked); + let reengaged = false; + const closing = yield* service + .close({ threadId, deviceId: session.deviceId }, { when: Effect.sync(() => !reengaged) }) + .pipe(Effect.forkChild({ startImmediately: true })); + reengaged = true; + yield* Deferred.succeed(unlock, undefined); + yield* Fiber.join(closing); + yield* Fiber.join(holder); expect((yield* service.state).sessions.map((entry) => entry.threadId)).toEqual([ threadId, other, From 5d758c463ddc9d9ebf43f745a73a584cb70823d8 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Tue, 10 Nov 2026 20:44:24 +0000 Subject: [PATCH 11/12] test(server): drop the deviceless settle case the empty loop already covers AI-assisted; verified with the tests and tools listed above. Co-Authored-By: Claude Opus 5.5 --- .../ThreadSettlementService.test.ts | 23 ------------------- 1 file changed, 23 deletions(-) diff --git a/apps/server/src/orchestration-v2/ThreadSettlementService.test.ts b/apps/server/src/orchestration-v2/ThreadSettlementService.test.ts index 08aa8ab12d335..961d1e39ae2e4 100644 --- a/apps/server/src/orchestration-v2/ThreadSettlementService.test.ts +++ b/apps/server/src/orchestration-v2/ThreadSettlementService.test.ts @@ -1652,29 +1652,6 @@ describe("ThreadSettlementServiceV2 terminals", () => { ), ); - it.effect("leaves the device service alone for a settled thread without devices", () => - Effect.scoped( - Effect.gen(function* () { - const settled = appThread("deviceless-thread", "settled", "/worktrees/deviceless"); - const fixture = yield* makeHarness({ - snapshot: makeSnapshot([]), - currentThreads: [settled], - existingWorktreePaths: ["/worktrees/deviceless"], - deviceSessions: [], - }); - - yield* Effect.gen(function* () { - const service = yield* ThreadSettlementService.ThreadSettlementServiceV2; - yield* startHarness(service, fixture.activation, fixture.snapshotReads); - yield* fixture.publishEvent(settledEvent(settled)); - assert.strictEqual((yield* Queue.take(fixture.scriptRuns)).threadId, settled.id); - yield* service.drain; - assert.deepStrictEqual(yield* Ref.get(fixture.deviceCloses), []); - }).pipe(Effect.provide(fixture.layer)); - }), - ), - ); - it.effect("runs the settle script in the thread's own worktree", () => Effect.scoped( Effect.gen(function* () { From de7040e4717c4871ec23647b3936c4059b6d2f47 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Sat, 14 Nov 2026 00:40:00 +0000 Subject: [PATCH 12/12] test(server): cut settle device tests to the cases a regression would fail The re-engaged case is covered by the existing re-engaged settle test's device assertion, and the plain release case by the per-device failure test's exact close inputs. Co-Authored-By: Claude Opus 5.5 --- .../ThreadSettlementService.test.ts | 197 ++++++------------ 1 file changed, 61 insertions(+), 136 deletions(-) diff --git a/apps/server/src/orchestration-v2/ThreadSettlementService.test.ts b/apps/server/src/orchestration-v2/ThreadSettlementService.test.ts index 55cf49cd660a8..cd4849cef5b9c 100644 --- a/apps/server/src/orchestration-v2/ThreadSettlementService.test.ts +++ b/apps/server/src/orchestration-v2/ThreadSettlementService.test.ts @@ -590,7 +590,7 @@ interface HarnessOptions { input: ProjectSetupScriptRunner.ProjectSetupScriptRunnerInput, ) => Effect.Effect; readonly onCloseIdle?: () => Effect.Effect; - readonly onDeviceClose?: (hostId: string | undefined) => Effect.Effect; + readonly onDeviceClose?: () => Effect.Effect; readonly deviceCloseError?: DeviceHostUnavailableError; readonly deviceSessions?: ReadonlyArray<{ readonly hostId: string; readonly deviceId: string }>; readonly currentThreads?: ReadonlyArray; @@ -745,7 +745,7 @@ const makeHarness = Effect.fn("makeThreadSettlementHarness")(function* (options: // onDeviceClose stands in for the lifecycle lock wait; `when` runs once the lock is held. close: (input, closeOptions) => Effect.gen(function* () { - yield* options.onDeviceClose?.(input.hostId) ?? Effect.void; + yield* options.onDeviceClose?.() ?? Effect.void; if (closeOptions?.when && !(yield* closeOptions.when)) return; yield* Ref.update(deviceCloses, (closes) => [...closes, input]); const error = options.deviceCloseError; @@ -1915,146 +1915,71 @@ describe("ThreadSettlementServiceV2 terminals", () => { }), ); - it.effect("keeps the devices of a thread re-engaged while a release waits on the lock", () => - Effect.scoped( - Effect.gen(function* () { - const firstStarted = yield* Deferred.make(); - const firstFinished = yield* Deferred.make(); - const settled = appThread("multi-device-thread", "settled"); - const marker = appThread("marker-thread", "settled"); - const currentThreads = [settled, marker]; - const fixture = yield* makeHarness({ - snapshot: makeSnapshot([]), - currentThreads, - deviceSessions: [ - { hostId: "slow-ssh", deviceId: "pixel" }, - { hostId: "local", deviceId: "emulator-5554" }, - ], - onDeviceClose: (hostId) => - hostId === "slow-ssh" - ? Deferred.succeed(firstStarted, undefined).pipe( - Effect.andThen(Deferred.await(firstFinished)), - ) - : Effect.void, - }); - - yield* Effect.gen(function* () { - const service = yield* ThreadSettlementService.ThreadSettlementServiceV2; - yield* startHarness(service, fixture.activation, fixture.snapshotReads); - yield* fixture.publishEvent(settledEvent(settled)); - yield* Deferred.await(firstStarted); - currentThreads[0] = { ...settled, settledOverride: "active" }; - yield* Deferred.succeed(firstFinished, undefined); - yield* fixture.publishEvent(settledEvent(marker)); - assert.deepStrictEqual(yield* Queue.take(fixture.closedIdle), { threadId: settled.id }); - assert.deepStrictEqual(yield* Queue.take(fixture.closedIdle), { threadId: marker.id }); - yield* service.drain; - assert.deepStrictEqual( - (yield* Ref.get(fixture.deviceCloses)).filter((close) => close.threadId === settled.id), - [], - ); - }).pipe(Effect.provide(fixture.layer)); - }), - ), - ); - it.effect("closes another thread's terminals while a device release waits on the lock", () => - Effect.scoped( - Effect.gen(function* () { - const releaseStarted = yield* Deferred.make(); - const lockFreed = yield* Deferred.make(); - const waiting = appThread("waiting-device-thread", "settled"); - const other = appThread("other-thread", "settled"); - const fixture = yield* makeHarness({ - snapshot: makeSnapshot([]), - currentThreads: [waiting, other], - deviceSessions: [{ hostId: "starting-ssh", deviceId: "pixel" }], - onDeviceClose: () => - Deferred.succeed(releaseStarted, undefined).pipe( - Effect.andThen(Deferred.await(lockFreed)), - ), - }); - - yield* Effect.gen(function* () { - const service = yield* ThreadSettlementService.ThreadSettlementServiceV2; - yield* startHarness(service, fixture.activation, fixture.snapshotReads); - yield* fixture.publishEvent(settledEvent(waiting)); - assert.deepStrictEqual(yield* Queue.take(fixture.closedIdle), { threadId: waiting.id }); - yield* Deferred.await(releaseStarted); - yield* fixture.publishEvent(settledEvent(other)); - assert.deepStrictEqual(yield* Queue.take(fixture.closedIdle), { threadId: other.id }); - yield* Deferred.succeed(lockFreed, undefined); - yield* service.drain; - assert.deepStrictEqual( - (yield* Ref.get(fixture.deviceCloses)).map((close) => close.threadId), - [waiting.id, other.id], - ); - }).pipe(Effect.provide(fixture.layer)); - }), - ), - ); - - it.effect("releases a settled thread's devices without shutting them down", () => - Effect.scoped( - Effect.gen(function* () { - const settled = appThread("device-thread", "settled"); - const marker = appThread("marker-thread", "settled"); - const fixture = yield* makeHarness({ - snapshot: makeSnapshot([]), - currentThreads: [settled, marker], - }); + Effect.gen(function* () { + const releaseStarted = yield* Deferred.make(); + const lockFreed = yield* Deferred.make(); + const waiting = appThread("waiting-device-thread", "settled"); + const other = appThread("other-thread", "settled"); + const fixture = yield* makeHarness({ + snapshot: makeSnapshot([]), + currentThreads: [waiting, other], + deviceSessions: [{ hostId: "starting-ssh", deviceId: "pixel" }], + onDeviceClose: () => + Deferred.succeed(releaseStarted, undefined).pipe( + Effect.andThen(Deferred.await(lockFreed)), + ), + }); - yield* Effect.gen(function* () { - const service = yield* ThreadSettlementService.ThreadSettlementServiceV2; - yield* startHarness(service, fixture.activation, fixture.snapshotReads); - yield* fixture.publishEvent(settledEvent(settled)); - yield* fixture.publishEvent(settledEvent(marker)); - assert.deepStrictEqual(yield* Queue.take(fixture.closedIdle), { threadId: settled.id }); - assert.deepStrictEqual(yield* Queue.take(fixture.closedIdle), { threadId: marker.id }); - yield* service.drain; - const closes = yield* Ref.get(fixture.deviceCloses); - assert.deepStrictEqual( - closes.filter((close) => close.threadId === settled.id), - [{ threadId: settled.id, hostId: "local", deviceId: "emulator-5554" }], - ); - }).pipe(Effect.provide(fixture.layer)); - }), - ), + yield* Effect.gen(function* () { + const service = yield* ThreadSettlementService.ThreadSettlementServiceV2; + yield* startHarness(service, fixture.activation, fixture.snapshotReads); + yield* fixture.publishEvent(settledEvent(waiting)); + assert.deepStrictEqual(yield* Queue.take(fixture.closedIdle), { threadId: waiting.id }); + yield* Deferred.await(releaseStarted); + yield* fixture.publishEvent(settledEvent(other)); + assert.deepStrictEqual(yield* Queue.take(fixture.closedIdle), { threadId: other.id }); + yield* Deferred.succeed(lockFreed, undefined); + yield* service.drain; + assert.deepStrictEqual( + (yield* Ref.get(fixture.deviceCloses)).map((close) => close.threadId), + [waiting.id, other.id], + ); + }).pipe(Effect.provide(fixture.layer)); + }), ); it.effect("releases the other devices and runs the settle script when one release fails", () => - Effect.scoped( - Effect.gen(function* () { - const settled = appThread("reconnecting-host-thread", "settled", "/worktrees/thread"); - const fixture = yield* makeHarness({ - snapshot: makeSnapshot([]), - currentThreads: [settled], - existingWorktreePaths: ["/worktrees/thread"], - deviceSessions: [ - { hostId: "ssh-host", deviceId: "remote-pixel" }, - { hostId: "local", deviceId: "emulator-5554" }, - ], - deviceCloseError: new DeviceHostUnavailableError({ - hostId: "ssh-host", - reason: "Device host is reconnecting. Retry close after it reconnects.", - }), - }); + Effect.gen(function* () { + const settled = appThread("reconnecting-host-thread", "settled", "/worktrees/thread"); + const fixture = yield* makeHarness({ + snapshot: makeSnapshot([]), + currentThreads: [settled], + existingWorktreePaths: ["/worktrees/thread"], + deviceSessions: [ + { hostId: "ssh-host", deviceId: "remote-pixel" }, + { hostId: "local", deviceId: "emulator-5554" }, + ], + deviceCloseError: new DeviceHostUnavailableError({ + hostId: "ssh-host", + reason: "Device host is reconnecting. Retry close after it reconnects.", + }), + }); - yield* Effect.gen(function* () { - const service = yield* ThreadSettlementService.ThreadSettlementServiceV2; - yield* startHarness(service, fixture.activation, fixture.snapshotReads); - yield* fixture.publishEvent(settledEvent(settled)); - assert.deepStrictEqual(yield* Queue.take(fixture.closedIdle), { threadId: settled.id }); - const run = yield* Queue.take(fixture.scriptRuns); - assert.strictEqual(run.threadId, settled.id); - yield* service.drain; - assert.deepStrictEqual(yield* Ref.get(fixture.deviceCloses), [ - { threadId: settled.id, hostId: "ssh-host", deviceId: "remote-pixel" }, - { threadId: settled.id, hostId: "local", deviceId: "emulator-5554" }, - ]); - }).pipe(Effect.provide(fixture.layer)); - }), - ), + yield* Effect.gen(function* () { + const service = yield* ThreadSettlementService.ThreadSettlementServiceV2; + yield* startHarness(service, fixture.activation, fixture.snapshotReads); + yield* fixture.publishEvent(settledEvent(settled)); + assert.deepStrictEqual(yield* Queue.take(fixture.closedIdle), { threadId: settled.id }); + const run = yield* Queue.take(fixture.scriptRuns); + assert.strictEqual(run.threadId, settled.id); + yield* service.drain; + assert.deepStrictEqual(yield* Ref.get(fixture.deviceCloses), [ + { threadId: settled.id, hostId: "ssh-host", deviceId: "remote-pixel" }, + { threadId: settled.id, hostId: "local", deviceId: "emulator-5554" }, + ]); + }).pipe(Effect.provide(fixture.layer)); + }), ); it.effect("runs the settle script in the thread's own worktree", () =>