diff --git a/apps/server/src/device/DeviceService.test.ts b/apps/server/src/device/DeviceService.test.ts index e196b4cb7c3b4..0865ef0e838d8 100644 --- a/apps/server/src/device/DeviceService.test.ts +++ b/apps/server/src/device/DeviceService.test.ts @@ -1814,6 +1814,26 @@ describe("agent session release on close", () => { const other = ThreadId.make("other"); const session = yield* openClaim(service, threadId); yield* openClaim(service, other, session.deviceId); + 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, + ]); 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 b365a6b9fe5c1..575484e24fff6 100644 --- a/apps/server/src/device/DeviceService.ts +++ b/apps/server/src/device/DeviceService.ts @@ -130,6 +130,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; @@ -210,7 +215,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; @@ -904,10 +912,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 76243589a1e67..cd4849cef5b9c 100644 --- a/apps/server/src/orchestration-v2/ThreadSettlementService.test.ts +++ b/apps/server/src/orchestration-v2/ThreadSettlementService.test.ts @@ -8,6 +8,7 @@ import { ProviderInstanceId, MessageId, RuntimeRequestId, + DeviceHostUnavailableError, RunId, type OrchestrationV2Run, ThreadId, @@ -36,6 +37,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 SqlitePersistence from "../persistence/Sqlite.ts"; import * as ProjectSetupScriptRunner from "../project/ProjectSetupScriptRunner.ts"; @@ -588,6 +590,9 @@ interface HarnessOptions { input: ProjectSetupScriptRunner.ProjectSetupScriptRunnerInput, ) => Effect.Effect; readonly onCloseIdle?: () => Effect.Effect; + readonly onDeviceClose?: () => Effect.Effect; + readonly deviceCloseError?: DeviceHostUnavailableError; + readonly deviceSessions?: ReadonlyArray<{ readonly hostId: string; readonly deviceId: string }>; readonly currentThreads?: ReadonlyArray; readonly projections?: ProjectionStore.ProjectionStoreV2Shape; } @@ -615,6 +620,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 observedEvents = yield* Queue.unbounded(); const closedIdle = yield* Queue.unbounded<{ readonly threadId: string }>(); @@ -703,13 +711,14 @@ const makeHarness = Effect.fn("makeThreadSettlementHarness")(function* (options: ), ), ), - getThread: (threadId) => { - if (options.projections) return options.projections.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(() => { + if (options.projections) return options.projections.getThread(threadId); + 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).pipe( @@ -721,6 +730,28 @@ const makeHarness = Effect.fn("makeThreadSettlementHarness")(function* (options: closeOnSettle: (input) => Queue.offer(closedIdle, input).pipe(Effect.andThen(options.onCloseIdle?.() ?? Effect.void)), }), + Layer.mock(DeviceService.DeviceService)({ + sessionsForThread: (threadId) => + Effect.succeed( + (options.deviceSessions ?? [{ hostId: "local", deviceId: "emulator-5554" }]).map( + (session) => ({ + ...session, + threadId, + platform: "android" as const, + openedAt: "2026-01-01T00:00:00.000Z", + }), + ), + ), + // onDeviceClose stands in for the lifecycle lock wait; `when` runs once the lock is held. + close: (input, closeOptions) => + Effect.gen(function* () { + yield* options.onDeviceClose?.() ?? 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) => Queue.offer(scriptRuns, input).pipe( @@ -772,6 +803,7 @@ const makeHarness = Effect.fn("makeThreadSettlementHarness")(function* (options: invalidatedCwds, closedIdle, scriptRuns, + deviceCloses, observedEvents, publishEvent: (event: OrchestrationV2DomainEvent) => PubSub.publish(domainEvents, event), updateSettings, @@ -1883,6 +1915,73 @@ describe("ThreadSettlementServiceV2 terminals", () => { }), ); + it.effect("closes another thread's terminals while a device release waits on the lock", () => + 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 the other devices and runs the settle script when one release fails", () => + 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)); + }), + ); + it.effect("runs the settle script in the thread's own worktree", () => Effect.gen(function* () { const shared = appThread("shared-checkout-thread", "settled"); @@ -1929,6 +2028,11 @@ 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), + [], + ); }).pipe(Effect.provide(fixture.layer)); }), ); diff --git a/apps/server/src/orchestration-v2/ThreadSettlementService.ts b/apps/server/src/orchestration-v2/ThreadSettlementService.ts index cfcf1b1e057c0..ce1fca2fd502b 100644 --- a/apps/server/src/orchestration-v2/ThreadSettlementService.ts +++ b/apps/server/src/orchestration-v2/ThreadSettlementService.ts @@ -26,6 +26,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"; @@ -315,6 +316,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(); @@ -634,12 +636,55 @@ 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, 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 }) => + 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 }, + ); + }, + (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. @@ -772,6 +817,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"]; }); diff --git a/docs/user/thread-sidebar.md b/docs/user/thread-sidebar.md index e5fe950e140a2..02cf80855f032 100644 --- a/docs/user/thread-sidebar.md +++ b/docs/user/thread-sidebar.md @@ -173,6 +173,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