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 d982c2e192c..cb64c6a4802 100644 --- a/apps/server/src/server.test.ts +++ b/apps/server/src/server.test.ts @@ -7020,6 +7020,126 @@ 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"); + assert.equal(sessionStopCommand.onlyIfSettled, true); + } + }).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..789f39b6397 100644 --- a/apps/server/src/ws.ts +++ b/apps/server/src/ws.ts @@ -1036,53 +1036,83 @@ 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; + // 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( + Option.match({ + onNone: () => false, + onSome: (thread) => + thread.session !== null && thread.session.status !== "stopped", + }), + ), + 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); - 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, + // 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); }).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( 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([