authorgravatar for me@t3.ggTheo Browne <me@t3.gg> 2026-08-08 22:44:15-04:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2026-08-08 19:44:15-07:00
log5bb8c0366447b1bb94c0ac41f2d4cd06730ffb94
treee571869c2754bf67b3755c5c90d25aa019afa1f6
parent886195ec1e6335d46d34fb3819ab4f3440b42b6d
signature Signed by PGP key B5690EEEBB952194

fix(server): settle no longer leaves monitors and dev servers running (#5774)

Co-authored-by: Claude Fable 5 <noreply@anthropic.com>

5 files changed, 258 insertions(+), 30 deletions(-)

apps/server/src/orchestration/decider.settled.test.ts+51
......@@ -518,4 +518,55 @@ it.layer(NodeServices.layer)("settled thread decider", (it) => {
518518 expect(routineEvents.map((event) => event.type)).toEqual(["thread.activity-appended"]);
519519 }),
520520 );
521
522 it.effect("drops an onlyIfSettled session stop when the thread was re-engaged", () =>
523 Effect.gen(function* () {
524 const stopCommand = (commandId: string) =>
525 ({
526 type: "thread.session.stop",
527 commandId: CommandId.make(commandId),
528 threadId: ThreadId.make("thread-1"),
529 createdAt: NOW,
530 onlyIfSettled: true,
531 }) as const;
532
533 // Still settled with an idle session: the cleanup stop goes through.
534 const stopped = yield* decideOrchestrationCommand({
535 command: stopCommand("cmd-stop-settled-idle"),
536 readModel: makeReadModel("settled", null, makeSession("ready")),
537 });
538 const stoppedEvents = Array.isArray(stopped) ? stopped : [stopped];
539 expect(stoppedEvents.map((event) => event.type)).toEqual(["thread.session-stop-requested"]);
540
541 // Re-engaged before the stop was decided (a turn start unsettles the
542 // thread): the stale cleanup stop must not kill the new session.
543 const unsettledError = yield* decideOrchestrationCommand({
544 command: stopCommand("cmd-stop-unsettled"),
545 readModel: makeReadModel(null, null, makeSession("starting")),
546 }).pipe(Effect.flip);
547 expect(unsettledError._tag).toBe("OrchestrationCommandInvariantError");
548
549 // Still settled but the session is already coming alive: same drop.
550 const aliveError = yield* decideOrchestrationCommand({
551 command: stopCommand("cmd-stop-session-alive"),
552 readModel: makeReadModel("settled", null, makeSession("starting")),
553 }).pipe(Effect.flip);
554 expect(aliveError._tag).toBe("OrchestrationCommandInvariantError");
555
556 // Without the flag the stop stays unconditional (archive, stop button).
557 const unconditional = yield* decideOrchestrationCommand({
558 command: {
559 type: "thread.session.stop",
560 commandId: CommandId.make("cmd-stop-unconditional"),
561 threadId: ThreadId.make("thread-1"),
562 createdAt: NOW,
563 },
564 readModel: makeReadModel(null, null, makeSession("starting")),
565 });
566 const unconditionalEvents = Array.isArray(unconditional) ? unconditional : [unconditional];
567 expect(unconditionalEvents.map((event) => event.type)).toEqual([
568 "thread.session-stop-requested",
569 ]);
570 }),
571 );
521572});
apps/server/src/orchestration/decider.ts+22-1
......@@ -1116,11 +1116,32 @@ export const decideOrchestrationCommand = Effect.fn("decideOrchestrationCommand"
11161116 }
11171117
11181118 case "thread.session.stop": {
1119 yield* requireThread({
1119 const thread = yield* requireThread({
11201120 readModel,
11211121 command,
11221122 threadId: command.threadId,
11231123 });
1124 // Settle-cleanup stops are conditional: between the settle landing and
1125 // this command, another client may have re-engaged the thread (a turn
1126 // start unsettles it and brings the session alive). Commands are
1127 // decided serially against this read model, so checking here — not in
1128 // the dispatcher's pre-settle snapshot — closes that race.
1129 if (command.onlyIfSettled === true) {
1130 const sessionComingAlive =
1131 thread.session?.status === "starting" || thread.session?.status === "running";
1132 if (
1133 thread.settledOverride !== "settled" ||
1134 sessionComingAlive ||
1135 threadHasQueuedTurnStart(thread, command.createdAt)
1136 ) {
1137 return yield* Effect.fail(
1138 new OrchestrationCommandInvariantError({
1139 commandType: command.type,
1140 detail: `thread ${command.threadId} was re-engaged after settle; skipping session stop`,
1141 }),
1142 );
1143 }
1144 }
11241145 return {
11251146 ...(yield* withEventBase({
11261147 aggregateKind: "thread",
apps/server/src/server.test.ts+120
......@@ -7020,6 +7020,126 @@ it.layer(NodeServices.layer)("server router seam", (it) => {
70207020 }).pipe(Effect.provide(NodeHttpServer.layerTest)),
70217021 );
70227022
7023 it.effect("stops the provider session after settle without closing terminals", () =>
7024 Effect.gen(function* () {
7025 const threadId = ThreadId.make("thread-settle");
7026 const effects: string[] = [];
7027 const dispatchedCommands: Array<OrchestrationCommand> = [];
7028 const now = "2026-01-01T00:00:00.000Z";
7029
7030 yield* buildAppUnderTest({
7031 layers: {
7032 terminalManager: {
7033 close: (input) =>
7034 Effect.sync(() => {
7035 effects.push(`terminal.close:${input.threadId}`);
7036 }),
7037 },
7038 orchestrationEngine: {
7039 dispatch: (command) =>
7040 Effect.sync(() => {
7041 dispatchedCommands.push(command);
7042 effects.push(`dispatch:${command.type}`);
7043 return { sequence: dispatchedCommands.length };
7044 }),
7045 },
7046 projectionSnapshotQuery: {
7047 getThreadShellById: () =>
7048 Effect.succeed(
7049 Option.some(
7050 makeDefaultOrchestrationThreadShell({
7051 id: threadId,
7052 updatedAt: now,
7053 session: {
7054 threadId,
7055 status: "ready",
7056 providerName: "claudeAgent",
7057 runtimeMode: "full-access",
7058 activeTurnId: null,
7059 lastError: null,
7060 updatedAt: now,
7061 },
7062 }),
7063 ),
7064 ),
7065 },
7066 },
7067 });
7068
7069 const wsUrl = yield* getWsServerUrl("/ws");
7070 const dispatchResult = yield* Effect.scoped(
7071 withWsRpcClient(wsUrl, (client) =>
7072 client[ORCHESTRATION_WS_METHODS.dispatchCommand]({
7073 type: "thread.settle",
7074 commandId: CommandId.make("cmd-thread-settle"),
7075 threadId,
7076 }),
7077 ),
7078 );
7079
7080 assert.equal(dispatchResult.sequence, 1);
7081 assert.deepEqual(effects, ["dispatch:thread.settle", "dispatch:thread.session.stop"]);
7082 const sessionStopCommand = dispatchedCommands[1];
7083 assert.equal(sessionStopCommand?.type, "thread.session.stop");
7084 if (sessionStopCommand?.type === "thread.session.stop") {
7085 assert.equal(sessionStopCommand.threadId, threadId);
7086 assert.equal(sessionStopCommand.commandId, "session-stop-for-settle:cmd-thread-settle");
7087 assert.equal(sessionStopCommand.onlyIfSettled, true);
7088 }
7089 }).pipe(Effect.provide(NodeHttpServer.layerTest)),
7090 );
7091
7092 it.effect("settles without dispatching session stop when the thread has no session", () =>
7093 Effect.gen(function* () {
7094 const threadId = ThreadId.make("thread-settle-no-session");
7095 const effects: string[] = [];
7096 const dispatchedCommands: Array<OrchestrationCommand> = [];
7097
7098 yield* buildAppUnderTest({
7099 layers: {
7100 terminalManager: {
7101 close: (input) =>
7102 Effect.sync(() => {
7103 effects.push(`terminal.close:${input.threadId}`);
7104 }),
7105 },
7106 orchestrationEngine: {
7107 dispatch: (command) =>
7108 Effect.sync(() => {
7109 dispatchedCommands.push(command);
7110 effects.push(`dispatch:${command.type}`);
7111 return { sequence: dispatchedCommands.length };
7112 }),
7113 },
7114 projectionSnapshotQuery: {
7115 getThreadShellById: () =>
7116 Effect.succeed(
7117 Option.some(makeDefaultOrchestrationThreadShell({ id: threadId, session: null })),
7118 ),
7119 },
7120 },
7121 });
7122
7123 const wsUrl = yield* getWsServerUrl("/ws");
7124 const dispatchResult = yield* Effect.scoped(
7125 withWsRpcClient(wsUrl, (client) =>
7126 client[ORCHESTRATION_WS_METHODS.dispatchCommand]({
7127 type: "thread.settle",
7128 commandId: CommandId.make("cmd-thread-settle-no-session"),
7129 threadId,
7130 }),
7131 ),
7132 );
7133
7134 assert.equal(dispatchResult.sequence, 1);
7135 assert.deepEqual(effects, ["dispatch:thread.settle"]);
7136 assert.deepEqual(
7137 dispatchedCommands.map((command) => command.type),
7138 ["thread.settle"],
7139 );
7140 }).pipe(Effect.provide(NodeHttpServer.layerTest)),
7141 );
7142
70237143 it.effect("archives and still closes terminals when session stop fails", () =>
70247144 Effect.gen(function* () {
70257145 const threadId = ThreadId.make("thread-archive-stop-failure");
apps/server/src/ws.ts+59-29
......@@ -1036,53 +1036,83 @@ const makeWsRpcLayer = (
10361036 ORCHESTRATION_WS_METHODS.dispatchCommand,
10371037 Effect.gen(function* () {
10381038 const normalizedCommand = yield* normalizeDispatchCommand(command);
1039 const shouldStopSessionAfterArchive =
1040 normalizedCommand.type === "thread.archive"
1041 ? yield* projectionSnapshotQuery
1042 .getThreadShellById(normalizedCommand.threadId)
1043 .pipe(
1044 Effect.map(
1045 Option.match({
1046 onNone: () => false,
1047 onSome: (thread) =>
1048 thread.session !== null && thread.session.status !== "stopped",
1049 }),
1050 ),
1051 Effect.orElseSucceed(() => false),
1052 )
1053 : false;
1039 // Archive and settle both mean "done with this thread", so a
1040 // live provider session must not keep running background work
1041 // (PR monitors, dev servers, subagent fleets) after either
1042 // lands. The decider rejects settling a starting/running
1043 // session, so for settle this only ever stops an idle one; a
1044 // stopped session-set does not count as activity, so the stop
1045 // cannot un-settle the thread it follows.
1046 const parkingCommand =
1047 normalizedCommand.type === "thread.archive" ||
1048 normalizedCommand.type === "thread.settle"
1049 ? normalizedCommand
1050 : undefined;
1051 // Best-effort on purpose: the user's archive/settle must not
1052 // fail because this cleanup read blipped, so a failed read
1053 // logs and skips the stop instead of propagating.
1054 const shouldStopSessionAfterCommand = parkingCommand
1055 ? yield* projectionSnapshotQuery.getThreadShellById(parkingCommand.threadId).pipe(
1056 Effect.map(
1057 Option.match({
1058 onNone: () => false,
1059 onSome: (thread) =>
1060 thread.session !== null && thread.session.status !== "stopped",
1061 }),
1062 ),
1063 Effect.catchCause((cause) =>
1064 Effect.logWarning(
1065 "failed to read thread session state before session-stop check",
1066 { threadId: parkingCommand.threadId, cause },
1067 ).pipe(Effect.as(false)),
1068 ),
1069 )
1070 : false;
10541071 const result = yield* dispatchNormalizedCommand(normalizedCommand);
1055 if (normalizedCommand.type === "thread.archive") {
1056 if (shouldStopSessionAfterArchive) {
1072 if (parkingCommand) {
1073 const parkingKind = parkingCommand.type === "thread.archive" ? "archive" : "settle";
1074 if (shouldStopSessionAfterCommand) {
10571075 yield* Effect.gen(function* () {
10581076 const stopCommand = yield* normalizeDispatchCommand({
10591077 type: "thread.session.stop",
10601078 commandId: CommandId.make(
1061 `session-stop-for-archive:${normalizedCommand.commandId}`,
1079 `session-stop-for-${parkingKind}:${parkingCommand.commandId}`,
10621080 ),
1063 threadId: normalizedCommand.threadId,
1081 threadId: parkingCommand.threadId,
10641082 createdAt: yield* nowIso,
1083 // A settled thread can be re-engaged before this stop is
1084 // decided; the decider then drops the stop instead of
1085 // killing the new session. Archive stops stay
1086 // unconditional: turn starts on archived threads are
1087 // rejected, so there is no new session to protect.
1088 ...(parkingKind === "settle" ? { onlyIfSettled: true } : {}),
10651089 });
10661090
10671091 yield* dispatchNormalizedCommand(stopCommand);
10681092 }).pipe(
10691093 Effect.catchCause((cause) =>
1070 Effect.logWarning("failed to stop provider session during archive", {
1071 threadId: normalizedCommand.threadId,
1094 Effect.logWarning(`failed to stop provider session during ${parkingKind}`, {
1095 threadId: parkingCommand.threadId,
10721096 cause,
10731097 }),
10741098 ),
10751099 );
10761100 }
10771101
1078 yield* terminalManager.close({ threadId: normalizedCommand.threadId }).pipe(
1079 Effect.catch((error) =>
1080 Effect.logWarning("failed to close thread terminals after archive", {
1081 threadId: normalizedCommand.threadId,
1082 error: error.message,
1083 }),
1084 ),
1085 );
1102 // Terminals are user-opened panes, not thread background
1103 // work: archive removes the thread from view so they close
1104 // with it, but a settled thread stays reachable and may be
1105 // un-settled, so its terminals stay up.
1106 if (parkingCommand.type === "thread.archive") {
1107 yield* terminalManager.close({ threadId: parkingCommand.threadId }).pipe(
1108 Effect.catch((error) =>
1109 Effect.logWarning("failed to close thread terminals after archive", {
1110 threadId: parkingCommand.threadId,
1111 error: error.message,
1112 }),
1113 ),
1114 );
1115 }
10861116 }
10871117 return result;
10881118 }).pipe(
packages/contracts/src/orchestration.ts+6
......@@ -870,6 +870,12 @@ const ThreadSessionStopCommand = Schema.Struct({
870870 commandId: CommandId,
871871 threadId: ThreadId,
872872 createdAt: IsoDateTime,
873 // Settle-cleanup stops are conditional: the decider drops the stop if the
874 // thread was re-engaged (unsettled, session starting/running, or a queued
875 // turn start) between the settle and this command. Guarding in the decider
876 // closes the race a post-settle snapshot read cannot: commands are decided
877 // serially against the authoritative read model.
878 onlyIfSettled: Schema.optional(Schema.Boolean),
873879});
874880
875881const DispatchableClientOrchestrationCommand = Schema.Union([