From d3cc5b8aae847a8fd75e4c94dc77008ae6357144 Mon Sep 17 00:00:00 2001 From: anishsharma21 Date: Fri, 11 Sep 2026 16:37:03 +1000 Subject: [PATCH 1/2] feat(server): add guarded idle session stops --- .../src/environment/ServerEnvironment.test.ts | 1 + .../src/environment/ServerEnvironment.ts | 1 + .../Layers/OrchestrationEngine.test.ts | 93 +++++++++++++ .../Layers/OrchestrationEngine.ts | 26 ++++ .../Layers/ProviderCommandReactor.test.ts | 83 ++++++++++-- .../Layers/ProviderCommandReactor.ts | 25 ++++ .../orchestration/SessionStopPolicy.test.ts | 51 ++++++++ .../src/orchestration/SessionStopPolicy.ts | 33 +++++ .../src/orchestration/decider.settled.test.ts | 123 ++++++++++++++++++ apps/server/src/orchestration/decider.ts | 27 ++++ packages/contracts/src/environment.ts | 2 + packages/contracts/src/orchestration.test.ts | 30 ++++- packages/contracts/src/orchestration.ts | 10 +- 13 files changed, 493 insertions(+), 12 deletions(-) create mode 100644 apps/server/src/orchestration/SessionStopPolicy.test.ts create mode 100644 apps/server/src/orchestration/SessionStopPolicy.ts diff --git a/apps/server/src/environment/ServerEnvironment.test.ts b/apps/server/src/environment/ServerEnvironment.test.ts index a12a8242b0d2..11938d6414e6 100644 --- a/apps/server/src/environment/ServerEnvironment.test.ts +++ b/apps/server/src/environment/ServerEnvironment.test.ts @@ -171,6 +171,7 @@ it.layer(NodeServices.layer)("ServerEnvironmentLive", (it) => { expect(second.capabilities.threadTitleRegeneration).toBe(true); expect(second.capabilities.threadPullRequests).toBe(true); expect(second.capabilities.threadPullRequestLinking).toBe(true); + expect(second.capabilities.guardedSessionStop).toBe(true); expect(second.capabilities.agentActivityPublishing).toBe(false); }), ); diff --git a/apps/server/src/environment/ServerEnvironment.ts b/apps/server/src/environment/ServerEnvironment.ts index 2aab17b27a76..88363372a87e 100644 --- a/apps/server/src/environment/ServerEnvironment.ts +++ b/apps/server/src/environment/ServerEnvironment.ts @@ -222,6 +222,7 @@ export const make = Effect.gen(function* () { threadSettlement: true, threadAutoSettlement: true, threadRestartContinuation: true, + guardedSessionStop: true, threadSnooze: true, environmentThemes: true, usageLimitSources: true, diff --git a/apps/server/src/orchestration/Layers/OrchestrationEngine.test.ts b/apps/server/src/orchestration/Layers/OrchestrationEngine.test.ts index 6d062ebcab4b..96c165841c0f 100644 --- a/apps/server/src/orchestration/Layers/OrchestrationEngine.test.ts +++ b/apps/server/src/orchestration/Layers/OrchestrationEngine.test.ts @@ -16,6 +16,7 @@ import { type OrchestrationCommand, type OrchestrationEvent, ProviderInstanceId, + ProviderDriverKind, } from "@t3tools/contracts"; import * as NodeServices from "@effect/platform-node/NodeServices"; import { it as effectIt } from "@effect/vitest"; @@ -697,6 +698,98 @@ describe("OrchestrationEngine", () => { }).pipe(Effect.provide(makeOrchestrationLayer())), ); + effectIt.effect("rejects a guarded session stop from a stale thread snapshot", () => + Effect.gen(function* () { + const engine = yield* OrchestrationEngineService; + const backgroundLiveness = yield* ThreadBackgroundLiveness.ThreadBackgroundLivenessService; + const projectId = ProjectId.make("project-guarded-stop"); + const threadId = ThreadId.make("thread-guarded-stop"); + yield* engine.dispatch({ + type: "project.create", + commandId: CommandId.make("cmd-guarded-stop-project"), + projectId, + title: "Project", + workspaceRoot: "/tmp/project-guarded-stop", + createdAt: now(), + }); + yield* engine.dispatch({ + type: "thread.create", + commandId: CommandId.make("cmd-guarded-stop-thread"), + threadId, + projectId, + title: "Thread", + modelSelection: { + instanceId: ProviderInstanceId.make("codex"), + model: "gpt-5-codex", + }, + interactionMode: DEFAULT_PROVIDER_INTERACTION_MODE, + runtimeMode: "full-access", + branch: null, + worktreePath: null, + createdAt: now(), + }); + yield* engine.dispatch({ + type: "thread.session.set", + commandId: CommandId.make("cmd-guarded-stop-session"), + threadId, + session: { + threadId, + status: "ready", + providerName: "codex", + providerInstanceId: ProviderInstanceId.make("codex"), + runtimeMode: "full-access", + activeTurnId: null, + lastError: null, + updatedAt: now(), + }, + createdAt: now(), + }); + const snapshotSequence = yield* engine.latestSequence; + yield* engine.dispatch({ + type: "thread.meta.update", + commandId: CommandId.make("cmd-guarded-stop-update"), + threadId, + title: "Changed", + }); + + const error = yield* engine + .dispatch({ + type: "thread.session.stop", + commandId: CommandId.make("cmd-guarded-stop-stale"), + threadId, + createdAt: now(), + onlyIfIdle: true, + snapshotSequence, + expectedProviderName: ProviderDriverKind.make("codex"), + }) + .pipe(Effect.flip); + + expect(error._tag).toBe("OrchestrationCommandInvariantError"); + + const freshSequence = yield* engine.latestSequence; + backgroundLiveness.recordTaskLiveness({ + threadId, + taskId: "guarded-stop-background-task", + taskType: "subagent", + status: undefined, + kind: "started", + }); + const backgroundError = yield* engine + .dispatch({ + type: "thread.session.stop", + commandId: CommandId.make("cmd-guarded-stop-background"), + threadId, + createdAt: now(), + onlyIfIdle: true, + snapshotSequence: freshSequence, + expectedProviderName: ProviderDriverKind.make("codex"), + }) + .pipe(Effect.flip); + expect(backgroundError._tag).toBe("OrchestrationCommandInvariantError"); + backgroundLiveness.clearThreadLiveness(threadId); + }).pipe(Effect.provide(makeOrchestrationLayer())), + ); + it("persists deterministic read models for repeated snapshot reads", async () => { const createdAt = now(); const system = await createOrchestrationSystem(); diff --git a/apps/server/src/orchestration/Layers/OrchestrationEngine.ts b/apps/server/src/orchestration/Layers/OrchestrationEngine.ts index fb2fadde5e63..05fc061c9f2b 100644 --- a/apps/server/src/orchestration/Layers/OrchestrationEngine.ts +++ b/apps/server/src/orchestration/Layers/OrchestrationEngine.ts @@ -185,6 +185,22 @@ const makeOrchestrationEngine = Effect.gen(function* () { }); } + if ( + envelope.command.type === "thread.session.stop" && + envelope.command.onlyIfIdle === true && + envelope.command.snapshotSequence !== undefined && + (yield* eventStore.hasEventAfter({ + aggregateKind: "thread", + aggregateId: envelope.command.threadId, + sequenceExclusive: envelope.command.snapshotSequence, + })) + ) { + return yield* new OrchestrationCommandInvariantError({ + commandType: envelope.command.type, + detail: `thread ${envelope.command.threadId} changed before guarded session stop`, + }); + } + // The decider compares the lookup inputs. Only recreation needs an // event check, since it can reset a thread to the same field values. if ( @@ -211,6 +227,16 @@ const makeOrchestrationEngine = Effect.gen(function* () { detail: `thread ${envelope.command.threadId} has live background work`, }); } + if ( + envelope.command.type === "thread.session.stop" && + envelope.command.onlyIfIdle === true && + threadBackgroundLiveness.getThreadBackgroundLiveness(envelope.command.threadId) !== null + ) { + return yield* new OrchestrationCommandInvariantError({ + commandType: envelope.command.type, + detail: `thread ${envelope.command.threadId} has live background work`, + }); + } // New and moved projects do not carry a resolved identity in the event-derived // command model. Legacy PR edits need it to identify the link they replace. diff --git a/apps/server/src/orchestration/Layers/ProviderCommandReactor.test.ts b/apps/server/src/orchestration/Layers/ProviderCommandReactor.test.ts index c927e72fe737..649e139d9678 100644 --- a/apps/server/src/orchestration/Layers/ProviderCommandReactor.test.ts +++ b/apps/server/src/orchestration/Layers/ProviderCommandReactor.test.ts @@ -173,6 +173,7 @@ describe("ProviderCommandReactor", () => { readonly unreadableHistory?: boolean; readonly titleRegenerationCompletionDispatchFailures?: number; readonly titleRegenerationBeforeStart?: "one" | "two"; + readonly startReactor?: boolean; readonly serverActivation?: Effect.Effect; readonly beforeReadySessionDispatch?: () => Effect.Effect; readonly beforeTurnStartDispatch?: () => Effect.Effect; @@ -191,6 +192,11 @@ describe("ProviderCommandReactor", () => { createdBaseDirs.add(baseDir); const { stateDir } = deriveServerPathsSync(baseDir, undefined); createdStateDirs.add(stateDir); + const backgroundLiveness = ThreadBackgroundLiveness.make(); + const backgroundLivenessLayer = Layer.succeed( + ThreadBackgroundLiveness.ThreadBackgroundLivenessService, + backgroundLiveness, + ); const runtimeEventPubSub = Effect.runSync(PubSub.unbounded()); const tryHandlePromptCommand = vi.fn( input?.tryHandlePromptCommandEffect ?? (() => Effect.succeed(false)), @@ -400,7 +406,7 @@ describe("ProviderCommandReactor", () => { const orchestrationLayer = OrchestrationEngineLive.pipe( Layer.provide(OrchestrationProjectionSnapshotQueryLive), - Layer.provide(ThreadBackgroundLiveness.layer), + Layer.provide(backgroundLivenessLayer), Layer.provide(ThreadPlanProgress.layer), Layer.provide(OrchestrationProjectionPipelineLive), Layer.provide(OrchestrationEventStoreLive), @@ -409,7 +415,7 @@ describe("ProviderCommandReactor", () => { Layer.provide(SqlitePersistenceMemory), ); const projectionSnapshotLayer = OrchestrationProjectionSnapshotQueryLive.pipe( - Layer.provide(ThreadBackgroundLiveness.layer), + Layer.provide(backgroundLivenessLayer), Layer.provide(ThreadPlanProgress.layer), Layer.provide(RepositoryIdentityResolver.layer), Layer.provide(SqlitePersistenceMemory), @@ -488,6 +494,7 @@ describe("ProviderCommandReactor", () => { }), ), Layer.provideMerge(ServerSettingsService.layerTest()), + Layer.provideMerge(backgroundLivenessLayer), Layer.provideMerge(SqlitePersistenceMemory), Layer.provideMerge(ServerConfig.layerTest(process.cwd(), baseDir)), Layer.provideMerge(NodeServices.layer), @@ -579,14 +586,18 @@ describe("ProviderCommandReactor", () => { } scope = await Effect.runPromise(Scope.make("sequential")); - await Effect.runPromise( - reactor - .start() - .pipe( - Scope.provide(scope), - Effect.provideService(ServerActivation, input?.serverActivation), - ), - ); + const start = () => + Effect.runPromise( + reactor + .start() + .pipe( + Scope.provide(scope!), + Effect.provideService(ServerActivation, input?.serverActivation), + ), + ); + if (input?.startReactor !== false) { + await start(); + } const drain = () => Effect.runPromise(reactor.drain); return { @@ -620,6 +631,8 @@ describe("ProviderCommandReactor", () => { generateThreadTitle, runtimeSessions, stateDir, + backgroundLiveness, + start, drain, runEffect, get titleRegenerationCompletionDispatchAttempts() { @@ -4130,6 +4143,56 @@ describe("ProviderCommandReactor", () => { }), ); + effectIt.effect("skips a guarded stop when background work appears before execution", () => + Effect.gen(function* () { + const harness = yield* Effect.promise(() => createHarness({ startReactor: false })); + const threadId = ThreadId.make("thread-1"); + const now = "2026-01-01T00:00:00.000Z"; + yield* harness.engine.dispatch({ + type: "thread.session.set", + commandId: CommandId.make("cmd-session-set-before-guarded-stop"), + threadId, + session: { + threadId, + status: "ready", + providerName: "codex", + providerInstanceId: ProviderInstanceId.make("codex"), + runtimeMode: "approval-required", + activeTurnId: null, + lastError: null, + updatedAt: now, + }, + createdAt: now, + }); + const snapshotSequence = yield* harness.engine.latestSequence; + yield* harness.engine.dispatch({ + type: "thread.session.stop", + commandId: CommandId.make("cmd-guarded-stop-before-background"), + threadId, + createdAt: now, + onlyIfIdle: true, + snapshotSequence, + expectedProviderName: ProviderDriverKind.make("codex"), + }); + harness.backgroundLiveness.recordTaskLiveness({ + threadId, + taskId: "guarded-stop-background-task", + taskType: "subagent", + status: undefined, + kind: "started", + }); + + yield* Effect.promise(() => harness.start()); + yield* Effect.promise(() => harness.drain()); + + expect(harness.stopSession).not.toHaveBeenCalled(); + const thread = yield* harness.snapshotQuery + .getThreadShellById(threadId) + .pipe(Effect.map(Option.getOrThrow)); + expect(thread.session?.status).toBe("ready"); + }), + ); + effectIt.effect("stops a ready provider session after automatic settlement", () => Effect.gen(function* () { const sessionStopped = yield* Deferred.make(); diff --git a/apps/server/src/orchestration/Layers/ProviderCommandReactor.ts b/apps/server/src/orchestration/Layers/ProviderCommandReactor.ts index 9b125922137c..e2b6118d1913 100644 --- a/apps/server/src/orchestration/Layers/ProviderCommandReactor.ts +++ b/apps/server/src/orchestration/Layers/ProviderCommandReactor.ts @@ -50,6 +50,8 @@ import { } from "../Services/ProviderCommandReactor.ts"; import { forkParked, ServerActivation } from "../../serverActivation.ts"; import { canReplaceThreadTitle, DEFAULT_THREAD_TITLE } from "../threadTitles.ts"; +import { canStopThreadSessionIfIdle } from "../SessionStopPolicy.ts"; +import { threadHasQueuedTurnStart } from "../ThreadSettlementPolicy.ts"; import { resolveSourceControlWriterModelSelection, ServerSettingsService, @@ -1757,6 +1759,29 @@ const make = Effect.gen(function* () { } const now = event.payload.createdAt; + if ( + event.payload.onlyIfIdle === true && + !canStopThreadSessionIfIdle({ + expectedProviderName: event.payload.expectedProviderName, + session: thread.session, + latestTurnState: thread.latestTurn?.state ?? null, + hasQueuedTurnStart: threadHasQueuedTurnStart( + thread, + DateTime.formatIso(yield* DateTime.now), + ), + hasPendingRequests: thread.hasPendingApprovals || thread.hasPendingUserInput, + backgroundLiveness: thread.backgroundLiveness ?? null, + }) + ) { + if (thread.session !== null) { + yield* setThreadSession({ + threadId: thread.id, + session: thread.session, + createdAt: DateTime.formatIso(yield* DateTime.now), + }); + } + return; + } const wasCompacting = compactingThreadIds.has(thread.id); stoppingThreadIds.add(thread.id); const clearStopping = Effect.sync(() => void stoppingThreadIds.delete(thread.id)); diff --git a/apps/server/src/orchestration/SessionStopPolicy.test.ts b/apps/server/src/orchestration/SessionStopPolicy.test.ts new file mode 100644 index 000000000000..4d370547ff94 --- /dev/null +++ b/apps/server/src/orchestration/SessionStopPolicy.test.ts @@ -0,0 +1,51 @@ +import { ProviderDriverKind, ProviderInstanceId, ThreadId, TurnId } from "@t3tools/contracts"; +import { describe, expect, it } from "vite-plus/test"; + +import { canStopThreadSessionIfIdle } from "./SessionStopPolicy.ts"; + +const idle = { + expectedProviderName: ProviderDriverKind.make("codex"), + session: { + threadId: ThreadId.make("thread-1"), + status: "ready" as const, + providerName: "codex", + providerInstanceId: ProviderInstanceId.make("codex"), + runtimeMode: "full-access" as const, + activeTurnId: null, + lastError: null, + updatedAt: "2026-09-11T00:00:00.000Z", + }, + latestTurnState: null, + hasQueuedTurnStart: false, + hasPendingRequests: false, + backgroundLiveness: null, +}; + +describe("guarded Codex session stop policy", () => { + it("allows the four non-working session states", () => { + for (const status of ["ready", "idle", "interrupted", "error"] as const) { + expect(canStopThreadSessionIfIdle({ ...idle, session: { ...idle.session, status } })).toBe( + true, + ); + } + }); + + it("rejects every signal that work is active or belongs to another provider", () => { + const unsafe = [ + { ...idle, expectedProviderName: ProviderDriverKind.make("claudeAgent") }, + { ...idle, session: { ...idle.session, providerName: "claudeAgent" } }, + { ...idle, session: { ...idle.session, status: "starting" as const } }, + { ...idle, session: { ...idle.session, status: "running" as const } }, + { ...idle, session: { ...idle.session, activeTurnId: TurnId.make("turn-1") } }, + { ...idle, latestTurnState: "running" as const }, + { ...idle, hasQueuedTurnStart: true }, + { ...idle, hasPendingRequests: true }, + { ...idle, backgroundLiveness: "working" as const }, + { ...idle, backgroundLiveness: "monitoring" as const }, + ]; + + for (const input of unsafe) { + expect(canStopThreadSessionIfIdle(input)).toBe(false); + } + }); +}); diff --git a/apps/server/src/orchestration/SessionStopPolicy.ts b/apps/server/src/orchestration/SessionStopPolicy.ts new file mode 100644 index 000000000000..b5bcd34502f6 --- /dev/null +++ b/apps/server/src/orchestration/SessionStopPolicy.ts @@ -0,0 +1,33 @@ +import type { + OrchestrationLatestTurnState, + OrchestrationSession, + ProviderDriverKind, +} from "@t3tools/contracts"; + +const idleSessionStatuses = new Set([ + "ready", + "idle", + "interrupted", + "error", +]); + +export function canStopThreadSessionIfIdle(input: { + readonly expectedProviderName: ProviderDriverKind | undefined; + readonly session: OrchestrationSession | null; + readonly latestTurnState: OrchestrationLatestTurnState | null; + readonly hasQueuedTurnStart: boolean; + readonly hasPendingRequests: boolean; + readonly backgroundLiveness: "working" | "monitoring" | null; +}): boolean { + return ( + input.expectedProviderName === "codex" && + input.session !== null && + input.session.providerName === "codex" && + idleSessionStatuses.has(input.session.status) && + input.session.activeTurnId === null && + input.latestTurnState !== "running" && + !input.hasQueuedTurnStart && + !input.hasPendingRequests && + input.backgroundLiveness === null + ); +} diff --git a/apps/server/src/orchestration/decider.settled.test.ts b/apps/server/src/orchestration/decider.settled.test.ts index bcb2408c21e1..423063c5aaa6 100644 --- a/apps/server/src/orchestration/decider.settled.test.ts +++ b/apps/server/src/orchestration/decider.settled.test.ts @@ -3,8 +3,10 @@ import { EventId, MessageId, ProjectId, + ProviderDriverKind, ProviderInstanceId, ThreadId, + TurnId, type OrchestrationEvent, type OrchestrationReadModel, type OrchestrationSession, @@ -850,4 +852,125 @@ it.layer(NodeServices.layer)("settled thread decider", (it) => { ]); }), ); + + it.effect("allows a guarded stop for an idle Codex session", () => + Effect.gen(function* () { + const stopped = yield* decideOrchestrationCommand({ + command: { + type: "thread.session.stop", + commandId: CommandId.make("cmd-stop-idle-codex"), + threadId: ThreadId.make("thread-1"), + createdAt: NOW, + onlyIfIdle: true, + snapshotSequence: 0, + expectedProviderName: ProviderDriverKind.make("codex"), + }, + readModel: makeReadModel("settled", null, { + ...makeSession("ready"), + providerName: "codex", + }), + }); + + const stoppedEvents = Array.isArray(stopped) ? stopped : [stopped]; + expect(stoppedEvents.map((event) => event.type)).toEqual(["thread.session-stop-requested"]); + }), + ); + + it.effect("rejects a guarded stop while the Codex session is running", () => + Effect.gen(function* () { + const error = yield* decideOrchestrationCommand({ + command: { + type: "thread.session.stop", + commandId: CommandId.make("cmd-stop-running-codex"), + threadId: ThreadId.make("thread-1"), + createdAt: NOW, + onlyIfIdle: true, + snapshotSequence: 0, + expectedProviderName: ProviderDriverKind.make("codex"), + }, + readModel: makeReadModel("settled", null, { + ...makeSession("running"), + providerName: "codex", + }), + }).pipe(Effect.flip); + + expect(error._tag).toBe("OrchestrationCommandInvariantError"); + }), + ); + + it.effect("rejects guarded stops for the wrong provider or an active turn", () => + Effect.gen(function* () { + const command = { + type: "thread.session.stop" as const, + commandId: CommandId.make("cmd-stop-unsafe-session"), + threadId: ThreadId.make("thread-1"), + createdAt: NOW, + onlyIfIdle: true, + snapshotSequence: 0, + expectedProviderName: ProviderDriverKind.make("codex"), + }; + const unsafeSessions: ReadonlyArray = [ + { ...makeSession("ready"), providerName: "claudeAgent" }, + { + ...makeSession("ready"), + providerName: "codex", + activeTurnId: TurnId.make("turn-1"), + }, + ]; + + for (const [index, session] of unsafeSessions.entries()) { + const error = yield* decideOrchestrationCommand({ + command: { ...command, commandId: CommandId.make(`cmd-stop-unsafe-${index}`) }, + readModel: makeReadModel("settled", null, session), + }).pipe(Effect.flip); + expect(error._tag).toBe("OrchestrationCommandInvariantError"); + } + }), + ); + + it.effect("does not expose guarded stops for non-Codex sessions", () => + Effect.gen(function* () { + const error = yield* decideOrchestrationCommand({ + command: { + type: "thread.session.stop", + commandId: CommandId.make("cmd-stop-guarded-claude"), + threadId: ThreadId.make("thread-1"), + createdAt: NOW, + onlyIfIdle: true, + snapshotSequence: 0, + expectedProviderName: ProviderDriverKind.make("claudeAgent"), + }, + readModel: makeReadModel("settled", null, { + ...makeSession("ready"), + providerName: "claudeAgent", + }), + }).pipe(Effect.flip); + + expect(error._tag).toBe("OrchestrationCommandInvariantError"); + }), + ); + + it.effect("allows every non-working Codex session state", () => + Effect.gen(function* () { + for (const status of ["ready", "idle", "interrupted", "error"] as const) { + const result = yield* decideOrchestrationCommand({ + command: { + type: "thread.session.stop", + commandId: CommandId.make(`cmd-stop-${status}`), + threadId: ThreadId.make("thread-1"), + createdAt: NOW, + onlyIfIdle: true, + snapshotSequence: 0, + expectedProviderName: ProviderDriverKind.make("codex"), + }, + readModel: makeReadModel("settled", null, { + ...makeSession(status), + providerName: "codex", + }), + }); + const events = Array.isArray(result) ? result : [result]; + expect(events.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 c0787a18d096..ee4c702d6deb 100644 --- a/apps/server/src/orchestration/decider.ts +++ b/apps/server/src/orchestration/decider.ts @@ -46,6 +46,7 @@ import { } from "./commandInvariants.ts"; import { projectEvent } from "./projector.ts"; import { threadHasQueuedTurnStart } from "./ThreadSettlementPolicy.ts"; +import { canStopThreadSessionIfIdle } from "./SessionStopPolicy.ts"; const isScriptRunCommand = Schema.is(SCRIPT_RUN_COMMAND_PATTERN); @@ -1689,6 +1690,26 @@ export const decideOrchestrationCommand = Effect.fn("decideOrchestrationCommand" ); } } + if (command.onlyIfIdle === true) { + if ( + command.snapshotSequence === undefined || + !canStopThreadSessionIfIdle({ + expectedProviderName: command.expectedProviderName, + session: thread.session, + latestTurnState: thread.latestTurn?.state ?? null, + hasQueuedTurnStart: hasQueuedTurnStartForThread(thread, command.createdAt), + hasPendingRequests: openRequests(thread).size > 0, + backgroundLiveness: null, + }) + ) { + return yield* Effect.fail( + new OrchestrationCommandInvariantError({ + commandType: command.type, + detail: `thread ${command.threadId} is not idle for a guarded session stop`, + }), + ); + } + } return { ...(yield* withEventBase({ aggregateKind: "thread", @@ -1700,6 +1721,12 @@ export const decideOrchestrationCommand = Effect.fn("decideOrchestrationCommand" payload: { threadId: command.threadId, createdAt: command.createdAt, + ...(command.onlyIfIdle === true + ? { + onlyIfIdle: true, + expectedProviderName: command.expectedProviderName, + } + : {}), }, }; } diff --git a/packages/contracts/src/environment.ts b/packages/contracts/src/environment.ts index 9dcc844e713a..e2556652e6c9 100644 --- a/packages/contracts/src/environment.ts +++ b/packages/contracts/src/environment.ts @@ -99,6 +99,8 @@ export const ExecutionEnvironmentCapabilities = Schema.Struct({ threadAutoSettlement: Schema.optionalKey(Schema.Boolean), /** Server persists the opt-in for continuing interrupted threads after restarts. */ threadRestartContinuation: Schema.optionalKey(Schema.Boolean), + /** Server can atomically reject session stops when a Codex thread is not idle. */ + guardedSessionStop: Schema.optionalKey(Schema.Boolean), /** Server understands thread.snooze / thread.unsnooze commands. Same version-skew contract as threadSettlement. */ threadSnooze: Schema.optionalKey(Schema.Boolean), diff --git a/packages/contracts/src/orchestration.test.ts b/packages/contracts/src/orchestration.test.ts index b8c8358813d4..d0c64301f6df 100644 --- a/packages/contracts/src/orchestration.test.ts +++ b/packages/contracts/src/orchestration.test.ts @@ -35,7 +35,7 @@ import { isProviderSendTurnSupportedImageMimeType, PROVIDER_SEND_TURN_MAX_FILE_BYTES, } from "./orchestration.ts"; -import { ProviderInstanceId } from "./providerInstance.ts"; +import { ProviderDriverKind, ProviderInstanceId } from "./providerInstance.ts"; const decodeTurnDiffInput = Schema.decodeUnknownEffect(OrchestrationGetTurnDiffInput); const decodeFullThreadDiffInput = Schema.decodeUnknownEffect(OrchestrationGetFullThreadDiffInput); @@ -70,6 +70,34 @@ const decodeThreadMetaUpdatedPayload = Schema.decodeUnknownEffect(ThreadMetaUpda const decodeDispatchCommandError = Schema.decodeUnknownEffect(OrchestrationDispatchCommandError); const decodeSnapShotAccessibility = Schema.decodeUnknownEffect(SnapShotAccessibility); +it.effect("decodes an idle-guarded Codex session stop without changing legacy stops", () => + Effect.gen(function* () { + const guarded = yield* decodeClientOrchestrationCommand({ + type: "thread.session.stop", + commandId: "cmd-guarded-stop", + threadId: "thread-1", + createdAt: "2026-09-11T00:00:00.000Z", + onlyIfIdle: true, + snapshotSequence: 42, + expectedProviderName: ProviderDriverKind.make("codex"), + }); + assert.strictEqual(guarded.type, "thread.session.stop"); + if (guarded.type === "thread.session.stop") { + assert.strictEqual(guarded.onlyIfIdle, true); + assert.strictEqual(guarded.snapshotSequence, 42); + assert.strictEqual(guarded.expectedProviderName, "codex"); + } + + const legacy = yield* decodeClientOrchestrationCommand({ + type: "thread.session.stop", + commandId: "cmd-legacy-stop", + threadId: "thread-1", + createdAt: "2026-09-11T00:00:00.000Z", + }); + assert.strictEqual(legacy.type, "thread.session.stop"); + }), +); + it.effect("decodes a dispatch error after its bootstrap thread was deleted", () => Effect.gen(function* () { const error = yield* decodeDispatchCommandError({ diff --git a/packages/contracts/src/orchestration.ts b/packages/contracts/src/orchestration.ts index 56a29b8f65f0..eba8cd0238e2 100644 --- a/packages/contracts/src/orchestration.ts +++ b/packages/contracts/src/orchestration.ts @@ -22,7 +22,7 @@ import { TrimmedString, TurnId, } from "./baseSchemas.ts"; -import { ProviderInstanceId } from "./providerInstance.ts"; +import { ProviderDriverKind, ProviderInstanceId } from "./providerInstance.ts"; import { PullRequestActor, PullRequestChecksState, @@ -1298,6 +1298,12 @@ const ThreadSessionStopCommand = Schema.Struct({ // closes the race a post-settle snapshot read cannot: commands are decided // serially against the authoritative read model. onlyIfSettled: Schema.optional(Schema.Boolean), + // External maintenance callers can request a compare-and-stop operation. + // The server accepts it only while the named provider is observably idle + // at the supplied snapshot sequence; ordinary client stops stay unchanged. + onlyIfIdle: Schema.optional(Schema.Boolean), + snapshotSequence: Schema.optional(NonNegativeInt), + expectedProviderName: Schema.optional(ProviderDriverKind), }); const DispatchableClientOrchestrationCommand = Schema.Union([ @@ -1773,6 +1779,8 @@ export const ThreadRevertedPayload = Schema.Struct({ export const ThreadSessionStopRequestedPayload = Schema.Struct({ threadId: ThreadId, createdAt: IsoDateTime, + onlyIfIdle: Schema.optional(Schema.Boolean), + expectedProviderName: Schema.optional(ProviderDriverKind), }); export const ThreadSessionSetPayload = Schema.Struct({ From f4802ad3d9568186a7cb3523dc553770984b747f Mon Sep 17 00:00:00 2001 From: anishsharma21 Date: Fri, 11 Sep 2026 17:10:25 +1000 Subject: [PATCH 2/2] fix(server): classify guarded stop rejections --- .../src/orchestration/decider.settled.test.ts | 22 +++++ apps/server/src/orchestration/decider.ts | 1 + apps/server/src/orchestration/http.ts | 27 ++++++- apps/server/src/server.test.ts | 80 +++++++++++++++++++ 4 files changed, 128 insertions(+), 2 deletions(-) diff --git a/apps/server/src/orchestration/decider.settled.test.ts b/apps/server/src/orchestration/decider.settled.test.ts index 423063c5aaa6..8eff0a281f8b 100644 --- a/apps/server/src/orchestration/decider.settled.test.ts +++ b/apps/server/src/orchestration/decider.settled.test.ts @@ -898,6 +898,28 @@ it.layer(NodeServices.layer)("settled thread decider", (it) => { }), ); + it.effect("rejects a guarded stop from a future snapshot", () => + Effect.gen(function* () { + const error = yield* decideOrchestrationCommand({ + command: { + type: "thread.session.stop", + commandId: CommandId.make("cmd-stop-future-snapshot"), + threadId: ThreadId.make("thread-1"), + createdAt: NOW, + onlyIfIdle: true, + snapshotSequence: 1, + expectedProviderName: ProviderDriverKind.make("codex"), + }, + readModel: makeReadModel("settled", null, { + ...makeSession("ready"), + providerName: "codex", + }), + }).pipe(Effect.flip); + + expect(error._tag).toBe("OrchestrationCommandInvariantError"); + }), + ); + it.effect("rejects guarded stops for the wrong provider or an active turn", () => Effect.gen(function* () { const command = { diff --git a/apps/server/src/orchestration/decider.ts b/apps/server/src/orchestration/decider.ts index ee4c702d6deb..ee64358dba93 100644 --- a/apps/server/src/orchestration/decider.ts +++ b/apps/server/src/orchestration/decider.ts @@ -1693,6 +1693,7 @@ export const decideOrchestrationCommand = Effect.fn("decideOrchestrationCommand" if (command.onlyIfIdle === true) { if ( command.snapshotSequence === undefined || + command.snapshotSequence > readModel.snapshotSequence || !canStopThreadSessionIfIdle({ expectedProviderName: command.expectedProviderName, session: thread.session, diff --git a/apps/server/src/orchestration/http.ts b/apps/server/src/orchestration/http.ts index f7147106c7a9..92003f28d6f1 100644 --- a/apps/server/src/orchestration/http.ts +++ b/apps/server/src/orchestration/http.ts @@ -1,10 +1,13 @@ import { AuthOrchestrationOperateScope, AuthOrchestrationReadScope, + EnvironmentInternalError, EnvironmentHttpApi, + EnvironmentRequestInvalidError, } from "@t3tools/contracts"; import * as Effect from "effect/Effect"; import * as Option from "effect/Option"; +import * as Schema from "effect/Schema"; import * as HttpApiBuilder from "effect/unstable/httpapi/HttpApiBuilder"; import { projectThreadDetailSnapshot } from "./ActivityPayloadProjection.ts"; @@ -18,6 +21,12 @@ import { } from "../auth/http.ts"; import { OrchestrationEngineService } from "./Services/OrchestrationEngine.ts"; import { ProjectionSnapshotQuery } from "./Services/ProjectionSnapshotQuery.ts"; +import { + isOrchestrationCommandRejection, + OrchestrationCommandPreviouslyRejectedError, +} from "./Errors.ts"; + +const isPreviouslyRejectedCommand = Schema.is(OrchestrationCommandPreviouslyRejectedError); export const orchestrationHttpApiLayer = HttpApiBuilder.group( EnvironmentHttpApi, @@ -100,8 +109,22 @@ export const orchestrationHttpApiLayer = HttpApiBuilder.group( Effect.tapError(() => cleanupFailedUploadedAttachments(args.payload, normalizedCommand), ), - Effect.catch((cause) => - failEnvironmentInternal("orchestration_dispatch_failed", cause), + Effect.catch( + ( + cause, + ): Effect.Effect< + never, + EnvironmentInternalError | EnvironmentRequestInvalidError + > => { + if ( + normalizedCommand.type === "thread.session.stop" && + normalizedCommand.onlyIfIdle === true && + (isOrchestrationCommandRejection(cause) || isPreviouslyRejectedCommand(cause)) + ) { + return failEnvironmentInvalidRequest("invalid_command"); + } + return failEnvironmentInternal("orchestration_dispatch_failed", cause); + }, ), ); }), diff --git a/apps/server/src/server.test.ts b/apps/server/src/server.test.ts index afb4adfef79c..1d95dd3cbe6e 100644 --- a/apps/server/src/server.test.ts +++ b/apps/server/src/server.test.ts @@ -114,6 +114,8 @@ import * as ExternalLauncher from "./process/externalLauncher.ts"; import * as RemoteOpenTargets from "./environment/RemoteOpenTargets.ts"; import * as OrchestrationEngine from "./orchestration/Services/OrchestrationEngine.ts"; import { + OrchestrationCommandInvariantError, + OrchestrationCommandPreviouslyRejectedError, OrchestrationListenerCallbackError, OrchestrationThreadSettleBlockedError, } from "./orchestration/Errors.ts"; @@ -2109,6 +2111,84 @@ it.layer(NodeServices.layer)("server router seam", (it) => { }).pipe(Effect.provide(NodeHttpServer.layerTest)), ); + it.effect("reports guarded command rejections as invalid requests", () => + Effect.gen(function* () { + yield* buildAppUnderTest({ + layers: { + orchestrationEngine: { + dispatch: (command) => + Effect.fail( + new OrchestrationCommandInvariantError({ + commandType: command.type, + detail: "thread is no longer idle", + }), + ), + }, + }, + }); + + const response = yield* fetchEffect(yield* getHttpServerUrl("/api/orchestration/dispatch"), { + method: "POST", + headers: { + cookie: yield* getAuthenticatedSessionCookieHeader(), + "content-type": "application/json", + }, + body: jsonRequestBody({ + type: "thread.session.stop", + commandId: "cmd-guarded-stop-rejected", + threadId: "thread-guarded-stop-rejected", + createdAt: "2026-09-11T10:01:00.000Z", + onlyIfIdle: true, + snapshotSequence: 7, + expectedProviderName: "codex", + }), + }); + const body = yield* responseJsonEffect<{ readonly reason: string }>(response); + + assert.equal(response.status, 400); + assert.equal(body.reason, "invalid_command"); + }).pipe(Effect.provide(NodeHttpServer.layerTest)), + ); + + it.effect("reports replayed guarded command rejections as invalid requests", () => + Effect.gen(function* () { + yield* buildAppUnderTest({ + layers: { + orchestrationEngine: { + dispatch: (command) => + Effect.fail( + new OrchestrationCommandPreviouslyRejectedError({ + commandId: command.commandId, + detail: "thread was no longer idle", + }), + ), + }, + }, + }); + + const response = yield* fetchEffect(yield* getHttpServerUrl("/api/orchestration/dispatch"), { + method: "POST", + headers: { + cookie: yield* getAuthenticatedSessionCookieHeader(), + "content-type": "application/json", + }, + body: jsonRequestBody({ + type: "thread.session.stop", + commandId: "cmd-guarded-stop-replayed", + threadId: "thread-guarded-stop-replayed", + createdAt: "2026-09-11T10:01:00.000Z", + onlyIfIdle: true, + snapshotSequence: 7, + expectedProviderName: "codex", + }), + }); + const body = yield* responseJsonEffect<{ readonly reason: string }>(response); + + assert.equal(response.status, 400); + assert.equal(body.reason, "invalid_command"); + }).pipe(Effect.provide(NodeHttpServer.layerTest)), + ); + it.effect("serves snapshots for MCP handoff thread IDs above the router default", () => Effect.gen(function* () { const threadId = ThreadId.make(