From d3e9ad9b0f6b8896f54b46d1cd3ec4f0dd8f7a39 Mon Sep 17 00:00:00 2001 From: Theo Browne Date: Sat, 8 Aug 2026 19:19:47 -0700 Subject: [PATCH 1/3] fix(server): settle no longer leaves monitors and dev servers running Settling a thread now stops its idle provider session the same way archive does, so PR watch loops and background dev servers die when you mark the thread done instead of running on for hours and waking the thread back up. Co-Authored-By: Claude Fable 5 --- apps/server/src/server.test.ts | 119 +++++++++++++++++++++++++++++++++ apps/server/src/ws.ts | 74 ++++++++++++-------- 2 files changed, 164 insertions(+), 29 deletions(-) diff --git a/apps/server/src/server.test.ts b/apps/server/src/server.test.ts index d982c2e192c..f95e77c459a 100644 --- a/apps/server/src/server.test.ts +++ b/apps/server/src/server.test.ts @@ -7020,6 +7020,125 @@ it.layer(NodeServices.layer)("server router seam", (it) => { }).pipe(Effect.provide(NodeHttpServer.layerTest)), ); + it.effect("stops the provider session after settle without closing terminals", () => + Effect.gen(function* () { + const threadId = ThreadId.make("thread-settle"); + const effects: string[] = []; + const dispatchedCommands: Array = []; + const now = "2026-01-01T00:00:00.000Z"; + + yield* buildAppUnderTest({ + layers: { + terminalManager: { + close: (input) => + Effect.sync(() => { + effects.push(`terminal.close:${input.threadId}`); + }), + }, + orchestrationEngine: { + dispatch: (command) => + Effect.sync(() => { + dispatchedCommands.push(command); + effects.push(`dispatch:${command.type}`); + return { sequence: dispatchedCommands.length }; + }), + }, + projectionSnapshotQuery: { + getThreadShellById: () => + Effect.succeed( + Option.some( + makeDefaultOrchestrationThreadShell({ + id: threadId, + updatedAt: now, + session: { + threadId, + status: "ready", + providerName: "claudeAgent", + runtimeMode: "full-access", + activeTurnId: null, + lastError: null, + updatedAt: now, + }, + }), + ), + ), + }, + }, + }); + + const wsUrl = yield* getWsServerUrl("/ws"); + const dispatchResult = yield* Effect.scoped( + withWsRpcClient(wsUrl, (client) => + client[ORCHESTRATION_WS_METHODS.dispatchCommand]({ + type: "thread.settle", + commandId: CommandId.make("cmd-thread-settle"), + threadId, + }), + ), + ); + + assert.equal(dispatchResult.sequence, 1); + assert.deepEqual(effects, ["dispatch:thread.settle", "dispatch:thread.session.stop"]); + const sessionStopCommand = dispatchedCommands[1]; + assert.equal(sessionStopCommand?.type, "thread.session.stop"); + if (sessionStopCommand?.type === "thread.session.stop") { + assert.equal(sessionStopCommand.threadId, threadId); + assert.equal(sessionStopCommand.commandId, "session-stop-for-settle:cmd-thread-settle"); + } + }).pipe(Effect.provide(NodeHttpServer.layerTest)), + ); + + it.effect("settles without dispatching session stop when the thread has no session", () => + Effect.gen(function* () { + const threadId = ThreadId.make("thread-settle-no-session"); + const effects: string[] = []; + const dispatchedCommands: Array = []; + + yield* buildAppUnderTest({ + layers: { + terminalManager: { + close: (input) => + Effect.sync(() => { + effects.push(`terminal.close:${input.threadId}`); + }), + }, + orchestrationEngine: { + dispatch: (command) => + Effect.sync(() => { + dispatchedCommands.push(command); + effects.push(`dispatch:${command.type}`); + return { sequence: dispatchedCommands.length }; + }), + }, + projectionSnapshotQuery: { + getThreadShellById: () => + Effect.succeed( + Option.some(makeDefaultOrchestrationThreadShell({ id: threadId, session: null })), + ), + }, + }, + }); + + const wsUrl = yield* getWsServerUrl("/ws"); + const dispatchResult = yield* Effect.scoped( + withWsRpcClient(wsUrl, (client) => + client[ORCHESTRATION_WS_METHODS.dispatchCommand]({ + type: "thread.settle", + commandId: CommandId.make("cmd-thread-settle-no-session"), + threadId, + }), + ), + ); + + assert.equal(dispatchResult.sequence, 1); + assert.deepEqual(effects, ["dispatch:thread.settle"]); + assert.deepEqual( + dispatchedCommands.map((command) => command.type), + ["thread.settle"], + ); + }).pipe(Effect.provide(NodeHttpServer.layerTest)), + ); + it.effect("archives and still closes terminals when session stop fails", () => Effect.gen(function* () { const threadId = ThreadId.make("thread-archive-stop-failure"); diff --git a/apps/server/src/ws.ts b/apps/server/src/ws.ts index 6d518fe16cf..8783e5949de 100644 --- a/apps/server/src/ws.ts +++ b/apps/server/src/ws.ts @@ -1036,53 +1036,69 @@ const makeWsRpcLayer = ( ORCHESTRATION_WS_METHODS.dispatchCommand, Effect.gen(function* () { const normalizedCommand = yield* normalizeDispatchCommand(command); - const shouldStopSessionAfterArchive = - normalizedCommand.type === "thread.archive" - ? yield* projectionSnapshotQuery - .getThreadShellById(normalizedCommand.threadId) - .pipe( - Effect.map( - Option.match({ - onNone: () => false, - onSome: (thread) => - thread.session !== null && thread.session.status !== "stopped", - }), - ), - Effect.orElseSucceed(() => false), - ) - : false; + // Archive and settle both mean "done with this thread", so a + // live provider session must not keep running background work + // (PR monitors, dev servers, subagent fleets) after either + // lands. The decider rejects settling a starting/running + // session, so for settle this only ever stops an idle one; a + // stopped session-set does not count as activity, so the stop + // cannot un-settle the thread it follows. + const parkingCommand = + normalizedCommand.type === "thread.archive" || + normalizedCommand.type === "thread.settle" + ? normalizedCommand + : undefined; + const shouldStopSessionAfterCommand = parkingCommand + ? yield* projectionSnapshotQuery.getThreadShellById(parkingCommand.threadId).pipe( + Effect.map( + Option.match({ + onNone: () => false, + onSome: (thread) => + thread.session !== null && thread.session.status !== "stopped", + }), + ), + Effect.orElseSucceed(() => false), + ) + : false; const result = yield* dispatchNormalizedCommand(normalizedCommand); - if (normalizedCommand.type === "thread.archive") { - if (shouldStopSessionAfterArchive) { + if (parkingCommand) { + const parkingKind = parkingCommand.type === "thread.archive" ? "archive" : "settle"; + if (shouldStopSessionAfterCommand) { yield* Effect.gen(function* () { const stopCommand = yield* normalizeDispatchCommand({ type: "thread.session.stop", commandId: CommandId.make( - `session-stop-for-archive:${normalizedCommand.commandId}`, + `session-stop-for-${parkingKind}:${parkingCommand.commandId}`, ), - threadId: normalizedCommand.threadId, + threadId: parkingCommand.threadId, createdAt: yield* nowIso, }); yield* dispatchNormalizedCommand(stopCommand); }).pipe( Effect.catchCause((cause) => - Effect.logWarning("failed to stop provider session during archive", { - threadId: normalizedCommand.threadId, + Effect.logWarning(`failed to stop provider session during ${parkingKind}`, { + threadId: parkingCommand.threadId, cause, }), ), ); } - yield* terminalManager.close({ threadId: normalizedCommand.threadId }).pipe( - Effect.catch((error) => - Effect.logWarning("failed to close thread terminals after archive", { - threadId: normalizedCommand.threadId, - error: error.message, - }), - ), - ); + // Terminals are user-opened panes, not thread background + // work: archive removes the thread from view so they close + // with it, but a settled thread stays reachable and may be + // un-settled, so its terminals stay up. + if (parkingCommand.type === "thread.archive") { + yield* terminalManager.close({ threadId: parkingCommand.threadId }).pipe( + Effect.catch((error) => + Effect.logWarning("failed to close thread terminals after archive", { + threadId: parkingCommand.threadId, + error: error.message, + }), + ), + ); + } } return result; }).pipe( From 40fbae2681f66344b4c700be49671e05edf9843e Mon Sep 17 00:00:00 2001 From: Theo Browne Date: Sat, 8 Aug 2026 19:24:23 -0700 Subject: [PATCH 2/3] fix(server): log the session-state read failure instead of swallowing it Review follow-up: the pre-stop projection read stays best-effort, but a failed read now leaves a warning instead of silently skipping the stop. Co-Authored-By: Claude Fable 5 --- apps/server/src/ws.ts | 10 +++++++++- 1 file changed, 9 insertions(+), 1 deletion(-) diff --git a/apps/server/src/ws.ts b/apps/server/src/ws.ts index 8783e5949de..4fb351f2a4b 100644 --- a/apps/server/src/ws.ts +++ b/apps/server/src/ws.ts @@ -1048,6 +1048,9 @@ const makeWsRpcLayer = ( normalizedCommand.type === "thread.settle" ? normalizedCommand : undefined; + // Best-effort on purpose: the user's archive/settle must not + // fail because this cleanup read blipped, so a failed read + // logs and skips the stop instead of propagating. const shouldStopSessionAfterCommand = parkingCommand ? yield* projectionSnapshotQuery.getThreadShellById(parkingCommand.threadId).pipe( Effect.map( @@ -1057,7 +1060,12 @@ const makeWsRpcLayer = ( thread.session !== null && thread.session.status !== "stopped", }), ), - Effect.orElseSucceed(() => false), + Effect.catchCause((cause) => + Effect.logWarning( + "failed to read thread session state before session-stop check", + { threadId: parkingCommand.threadId, cause }, + ).pipe(Effect.as(false)), + ), ) : false; const result = yield* dispatchNormalizedCommand(normalizedCommand); From 4a614a995c83e5919f617c591996beea56fe7103 Mon Sep 17 00:00:00 2001 From: Theo Browne Date: Sat, 8 Aug 2026 19:30:38 -0700 Subject: [PATCH 3/3] fix(server): settle cleanup cannot kill a session started mid-settle Review follow-up: the settle-originated session stop is now conditional (onlyIfSettled) and decided in the decider against the serialized read model, so a turn started between the settle and the stop survives. Archive stops stay unconditional. Co-Authored-By: Claude Fable 5 --- .../src/orchestration/decider.settled.test.ts | 51 +++++++++++++++++++ apps/server/src/orchestration/decider.ts | 23 ++++++++- apps/server/src/server.test.ts | 1 + apps/server/src/ws.ts | 6 +++ packages/contracts/src/orchestration.ts | 6 +++ 5 files changed, 86 insertions(+), 1 deletion(-) diff --git a/apps/server/src/orchestration/decider.settled.test.ts b/apps/server/src/orchestration/decider.settled.test.ts index 73f1cbf9127..e057764683e 100644 --- a/apps/server/src/orchestration/decider.settled.test.ts +++ b/apps/server/src/orchestration/decider.settled.test.ts @@ -518,4 +518,55 @@ it.layer(NodeServices.layer)("settled thread decider", (it) => { expect(routineEvents.map((event) => event.type)).toEqual(["thread.activity-appended"]); }), ); + + it.effect("drops an onlyIfSettled session stop when the thread was re-engaged", () => + Effect.gen(function* () { + const stopCommand = (commandId: string) => + ({ + type: "thread.session.stop", + commandId: CommandId.make(commandId), + threadId: ThreadId.make("thread-1"), + createdAt: NOW, + onlyIfSettled: true, + }) as const; + + // Still settled with an idle session: the cleanup stop goes through. + const stopped = yield* decideOrchestrationCommand({ + command: stopCommand("cmd-stop-settled-idle"), + readModel: makeReadModel("settled", null, makeSession("ready")), + }); + const stoppedEvents = Array.isArray(stopped) ? stopped : [stopped]; + expect(stoppedEvents.map((event) => event.type)).toEqual(["thread.session-stop-requested"]); + + // Re-engaged before the stop was decided (a turn start unsettles the + // thread): the stale cleanup stop must not kill the new session. + const unsettledError = yield* decideOrchestrationCommand({ + command: stopCommand("cmd-stop-unsettled"), + readModel: makeReadModel(null, null, makeSession("starting")), + }).pipe(Effect.flip); + expect(unsettledError._tag).toBe("OrchestrationCommandInvariantError"); + + // Still settled but the session is already coming alive: same drop. + const aliveError = yield* decideOrchestrationCommand({ + command: stopCommand("cmd-stop-session-alive"), + readModel: makeReadModel("settled", null, makeSession("starting")), + }).pipe(Effect.flip); + expect(aliveError._tag).toBe("OrchestrationCommandInvariantError"); + + // Without the flag the stop stays unconditional (archive, stop button). + const unconditional = yield* decideOrchestrationCommand({ + command: { + type: "thread.session.stop", + commandId: CommandId.make("cmd-stop-unconditional"), + threadId: ThreadId.make("thread-1"), + createdAt: NOW, + }, + readModel: makeReadModel(null, null, makeSession("starting")), + }); + const unconditionalEvents = Array.isArray(unconditional) ? unconditional : [unconditional]; + expect(unconditionalEvents.map((event) => event.type)).toEqual([ + "thread.session-stop-requested", + ]); + }), + ); }); diff --git a/apps/server/src/orchestration/decider.ts b/apps/server/src/orchestration/decider.ts index 3de2592c884..6e87b5bd5f1 100644 --- a/apps/server/src/orchestration/decider.ts +++ b/apps/server/src/orchestration/decider.ts @@ -1116,11 +1116,32 @@ export const decideOrchestrationCommand = Effect.fn("decideOrchestrationCommand" } case "thread.session.stop": { - yield* requireThread({ + const thread = yield* requireThread({ readModel, command, threadId: command.threadId, }); + // Settle-cleanup stops are conditional: between the settle landing and + // this command, another client may have re-engaged the thread (a turn + // start unsettles it and brings the session alive). Commands are + // decided serially against this read model, so checking here — not in + // the dispatcher's pre-settle snapshot — closes that race. + if (command.onlyIfSettled === true) { + const sessionComingAlive = + thread.session?.status === "starting" || thread.session?.status === "running"; + if ( + thread.settledOverride !== "settled" || + sessionComingAlive || + threadHasQueuedTurnStart(thread, command.createdAt) + ) { + return yield* Effect.fail( + new OrchestrationCommandInvariantError({ + commandType: command.type, + detail: `thread ${command.threadId} was re-engaged after settle; skipping session stop`, + }), + ); + } + } return { ...(yield* withEventBase({ aggregateKind: "thread", diff --git a/apps/server/src/server.test.ts b/apps/server/src/server.test.ts index f95e77c459a..cb64c6a4802 100644 --- a/apps/server/src/server.test.ts +++ b/apps/server/src/server.test.ts @@ -7084,6 +7084,7 @@ it.layer(NodeServices.layer)("server router seam", (it) => { if (sessionStopCommand?.type === "thread.session.stop") { assert.equal(sessionStopCommand.threadId, threadId); assert.equal(sessionStopCommand.commandId, "session-stop-for-settle:cmd-thread-settle"); + assert.equal(sessionStopCommand.onlyIfSettled, true); } }).pipe(Effect.provide(NodeHttpServer.layerTest)), ); diff --git a/apps/server/src/ws.ts b/apps/server/src/ws.ts index 4fb351f2a4b..789f39b6397 100644 --- a/apps/server/src/ws.ts +++ b/apps/server/src/ws.ts @@ -1080,6 +1080,12 @@ const makeWsRpcLayer = ( ), threadId: parkingCommand.threadId, createdAt: yield* nowIso, + // A settled thread can be re-engaged before this stop is + // decided; the decider then drops the stop instead of + // killing the new session. Archive stops stay + // unconditional: turn starts on archived threads are + // rejected, so there is no new session to protect. + ...(parkingKind === "settle" ? { onlyIfSettled: true } : {}), }); yield* dispatchNormalizedCommand(stopCommand); diff --git a/packages/contracts/src/orchestration.ts b/packages/contracts/src/orchestration.ts index 87270d98c1f..96d20c0a392 100644 --- a/packages/contracts/src/orchestration.ts +++ b/packages/contracts/src/orchestration.ts @@ -870,6 +870,12 @@ const ThreadSessionStopCommand = Schema.Struct({ commandId: CommandId, threadId: ThreadId, createdAt: IsoDateTime, + // Settle-cleanup stops are conditional: the decider drops the stop if the + // thread was re-engaged (unsettled, session starting/running, or a queued + // turn start) between the settle and this command. Guarding in the decider + // closes the race a post-settle snapshot read cannot: commands are decided + // serially against the authoritative read model. + onlyIfSettled: Schema.optional(Schema.Boolean), }); const DispatchableClientOrchestrationCommand = Schema.Union([