Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
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
51 changes: 51 additions & 0 deletions apps/server/src/orchestration/decider.settled.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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",
]);
}),
);
});
23 changes: 22 additions & 1 deletion apps/server/src/orchestration/decider.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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",
Expand Down
120 changes: 120 additions & 0 deletions apps/server/src/server.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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<OrchestrationCommand> = [];
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<OrchestrationCommand> = [];

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");
Expand Down
88 changes: 59 additions & 29 deletions apps/server/src/ws.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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
Comment thread
t3dotgg marked this conversation as resolved.
? 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);
Comment thread
macroscopeapp[bot] marked this conversation as resolved.
}).pipe(
Effect.catchCause((cause) =>
Comment thread
t3dotgg marked this conversation as resolved.
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(
Expand Down
6 changes: 6 additions & 0 deletions packages/contracts/src/orchestration.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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([
Expand Down
Loading