Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
17 commits
Select commit Hold shift + click to select a range
e7d82f8
test(server): settling a thread releases its devices (reproduces #32511)
juliusmarminge Nov 7, 2026
7120eec
fix(server): settling a thread releases its devices
juliusmarminge Nov 7, 2026
d630ade
fix(server): keep a re-engaged thread's devices and skip the device l…
juliusmarminge Nov 7, 2026
98619b3
test(server): a settled thread without devices leaves the device serv…
juliusmarminge Nov 7, 2026
2a6434c
fix(server): release each settled device separately
juliusmarminge Nov 7, 2026
7c6f7d4
fix(server): stop releasing devices once a settled thread re-engages
juliusmarminge Nov 7, 2026
120c6e8
Merge remote-tracking branch 'origin/main' into pr-32524
juliusmarminge Nov 10, 2026
2e051a5
fix(server): device releases no longer stall other threads' settle cl…
juliusmarminge Nov 10, 2026
45db815
fix(server): a thread re-engaged while its device release waits keeps…
juliusmarminge Nov 10, 2026
0394c18
Merge remote-tracking branch 'origin/main' into pr-32524
juliusmarminge Nov 10, 2026
d2bf0a9
docs: settling a thread releases its devices
juliusmarminge Nov 10, 2026
a45db96
test(server): check the close guard after a real lifecycle lock wait
juliusmarminge Nov 10, 2026
f890b60
Merge remote-tracking branch 'origin/main' into fix/issue-32511
juliusmarminge Nov 10, 2026
5d758c4
test(server): drop the deviceless settle case the empty loop already …
juliusmarminge Nov 10, 2026
0d00de5
Merge origin/main into fix/issue-32511
juliusmarminge Nov 13, 2026
5e1bd83
Merge origin/main into fix/issue-32511
juliusmarminge Nov 14, 2026
de7040e
test(server): cut settle device tests to the cases a regression would…
juliusmarminge Nov 14, 2026
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
20 changes: 20 additions & 0 deletions apps/server/src/device/DeviceService.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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<void>();
const unlock = yield* Deferred.make<void>();
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(
Expand Down
13 changes: 11 additions & 2 deletions apps/server/src/device/DeviceService.ts
Original file line number Diff line number Diff line change
Expand Up @@ -130,6 +130,11 @@ export interface DeviceReadiness extends DeviceHost.DeviceHostReady {
readonly hostId: DeviceHostId;
}

export interface DeviceCloseOptions<E> {
/** Runs under the lifecycle lock; false releases nothing. */
readonly when?: Effect.Effect<boolean, E>;
}

export interface DeviceAgentReadiness extends DeviceReadiness {
/** Null when each device has its own daemon; see `DeviceOperations.agentEndpoint`. */
readonly agentDevice: DeviceHost.AgentDeviceEndpoint | null;
Expand Down Expand Up @@ -210,7 +215,10 @@ export class DeviceService extends Context.Service<
input: DeviceOpenInput,
context?: DeviceHost.DeviceOpenContext,
) => Effect.Effect<DeviceSession, DeviceError>;
readonly close: (input: DeviceCloseInput) => Effect.Effect<void, DeviceError>;
readonly close: <E = never>(
input: DeviceCloseInput,
options?: DeviceCloseOptions<E>,
) => Effect.Effect<void, DeviceError | E>;
readonly shutdown: (input: DeviceShutdownInput) => Effect.Effect<void, DeviceError>;
readonly detail: (input: DeviceDetailInput) => Effect.Effect<DeviceDetail, DeviceError>;
readonly action: (input: DeviceActionInput) => Effect.Effect<DeviceDetail, DeviceError>;
Expand Down Expand Up @@ -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) =>
Expand Down
118 changes: 111 additions & 7 deletions apps/server/src/orchestration-v2/ThreadSettlementService.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -8,6 +8,7 @@ import {
ProviderInstanceId,
MessageId,
RuntimeRequestId,
DeviceHostUnavailableError,
RunId,
type OrchestrationV2Run,
ThreadId,
Expand Down Expand Up @@ -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";
Expand Down Expand Up @@ -588,6 +590,9 @@ interface HarnessOptions {
input: ProjectSetupScriptRunner.ProjectSetupScriptRunnerInput,
) => Effect.Effect<void>;
readonly onCloseIdle?: () => Effect.Effect<void>;
readonly onDeviceClose?: () => Effect.Effect<void>;
readonly deviceCloseError?: DeviceHostUnavailableError;
readonly deviceSessions?: ReadonlyArray<{ readonly hostId: string; readonly deviceId: string }>;
readonly currentThreads?: ReadonlyArray<OrchestrationV2AppThread>;
readonly projections?: ProjectionStore.ProjectionStoreV2Shape;
}
Expand Down Expand Up @@ -615,6 +620,9 @@ const makeHarness = Effect.fn("makeThreadSettlementHarness")(function* (options:
>([]);
const summaryRecovery = yield* Ref.make<ReadonlyArray<boolean | undefined>>([]);
const invalidatedCwds = yield* Ref.make<ReadonlyArray<string>>([]);
const deviceCloses = yield* Ref.make<
ReadonlyArray<Parameters<DeviceService.DeviceService["Service"]["close"]>[0]>
>([]);
const domainEvents = yield* PubSub.unbounded<OrchestrationV2DomainEvent>();
const observedEvents = yield* Queue.unbounded<EventId>();
const closedIdle = yield* Queue.unbounded<{ readonly threadId: string }>();
Expand Down Expand Up @@ -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(
Expand All @@ -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(
Expand Down Expand Up @@ -772,6 +803,7 @@ const makeHarness = Effect.fn("makeThreadSettlementHarness")(function* (options:
invalidatedCwds,
closedIdle,
scriptRuns,
deviceCloses,
observedEvents,
publishEvent: (event: OrchestrationV2DomainEvent) => PubSub.publish(domainEvents, event),
updateSettings,
Expand Down Expand Up @@ -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<void>();
const lockFreed = yield* Deferred.make<void>();
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");
Expand Down Expand Up @@ -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));
}),
);
Expand Down
50 changes: 49 additions & 1 deletion apps/server/src/orchestration-v2/ThreadSettlementService.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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";
Expand Down Expand Up @@ -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<ThreadId>();
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<ThreadId, number>();
Expand Down Expand Up @@ -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<void>();

// 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.
Expand Down Expand 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"];
});
2 changes: 2 additions & 0 deletions docs/user/thread-sidebar.md
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
Loading