diff --git a/apps/server/src/orchestration/decider.settled.test.ts b/apps/server/src/orchestration/decider.settled.test.ts index 73f1cbf9127..17343da0e31 100644 --- a/apps/server/src/orchestration/decider.settled.test.ts +++ b/apps/server/src/orchestration/decider.settled.test.ts @@ -470,6 +470,28 @@ it.layer(NodeServices.layer)("settled thread decider", (it) => { }), ); + it.effect("rejects a conditional session write after the projected session changes", () => + Effect.gen(function* () { + const currentSession = makeSession("running"); + const error = yield* decideOrchestrationCommand({ + command: { + type: "thread.session.set", + commandId: CommandId.make("cmd-stale-session-write"), + threadId: ThreadId.make("thread-1"), + expectedSessionUpdatedAt: "2025-12-31T23:59:00.000Z", + session: { + ...currentSession, + status: "stopped", + }, + createdAt: NOW, + }, + readModel: makeReadModel(null, null, currentSession), + }).pipe(Effect.flip); + + expect(error._tag).toBe("OrchestrationCommandInvariantError"); + }), + ); + it.effect("unsettles for approval and user-input activities but not others", () => Effect.gen(function* () { const approvalResult = yield* decideOrchestrationCommand({ diff --git a/apps/server/src/orchestration/decider.ts b/apps/server/src/orchestration/decider.ts index 100369ae6e3..4172aa05230 100644 --- a/apps/server/src/orchestration/decider.ts +++ b/apps/server/src/orchestration/decider.ts @@ -946,6 +946,15 @@ export const decideOrchestrationCommand = Effect.fn("decideOrchestrationCommand" command, threadId: command.threadId, }); + if ( + command.expectedSessionUpdatedAt !== undefined && + thread.session?.updatedAt !== command.expectedSessionUpdatedAt + ) { + return yield* new OrchestrationCommandInvariantError({ + commandType: command.type, + detail: `Thread '${command.threadId}' session changed after the command was prepared.`, + }); + } const sessionSetEvent: Omit = { ...(yield* withEventBase({ aggregateKind: "thread", diff --git a/apps/server/src/provider/Layers/ProviderSessionReaper.test.ts b/apps/server/src/provider/Layers/ProviderSessionReaper.test.ts index 3843c8acbcd..16a1e5cbef4 100644 --- a/apps/server/src/provider/Layers/ProviderSessionReaper.test.ts +++ b/apps/server/src/provider/Layers/ProviderSessionReaper.test.ts @@ -17,6 +17,11 @@ import * as Scope from "effect/Scope"; import * as Stream from "effect/Stream"; import { afterEach, describe, expect, it, vi } from "vite-plus/test"; +import { + OrchestrationEngineService, + type OrchestrationEngineShape, +} from "../../orchestration/Services/OrchestrationEngine.ts"; +import { OrchestrationCommandInvariantError } from "../../orchestration/Errors.ts"; import { ProjectionSnapshotQuery } from "../../orchestration/Services/ProjectionSnapshotQuery.ts"; import { SqlitePersistenceMemory } from "../../persistence/Layers/Sqlite.ts"; import * as ProviderSessionRuntime from "../../persistence/ProviderSessionRuntime.ts"; @@ -137,6 +142,8 @@ describe("ProviderSessionReaper", () => { async function createHarness(input: { readonly readModel: ReturnType; + readonly dispatchImplementation?: OrchestrationEngineShape["dispatch"]; + readonly sweepIntervalMs?: number; readonly stopSessionImplementation?: (input: { readonly threadId: ThreadId; }) => ReturnType; @@ -176,6 +183,9 @@ describe("ProviderSessionReaper", () => { rollbackConversation: () => unsupported(), streamEvents: Stream.empty, }; + const dispatch = vi.fn( + input.dispatchImplementation ?? (() => Effect.succeed({ sequence: 1 })), + ); const runtimeRepositoryLayer = ProviderSessionRuntime.layer.pipe( Layer.provide(SqlitePersistenceMemory), @@ -185,11 +195,19 @@ describe("ProviderSessionReaper", () => { ); const layer = makeProviderSessionReaperLive({ inactivityThresholdMs: 1_000, - sweepIntervalMs: 60_000, + sweepIntervalMs: input.sweepIntervalMs ?? 60_000, }).pipe( Layer.provideMerge(providerSessionDirectoryLayer), Layer.provideMerge(runtimeRepositoryLayer), Layer.provideMerge(Layer.succeed(ProviderService, providerService)), + Layer.provideMerge( + Layer.succeed(OrchestrationEngineService, { + dispatch, + readEvents: () => Stream.empty, + streamDomainEvents: Stream.empty, + latestSequence: Effect.succeed(0), + }), + ), Layer.provideMerge( Layer.succeed(ProjectionSnapshotQuery, { getCommandReadModel: () => Effect.die("unused"), @@ -218,9 +236,189 @@ describe("ProviderSessionReaper", () => { ); runtime = ManagedRuntime.make(layer); - return { stopSession, stoppedThreadIds }; + return { dispatch, stopSession, stoppedThreadIds }; } + it("projects durable stopped bindings that were left working during shutdown", async () => { + const threadId = ThreadId.make("thread-reaper-stopped-during-shutdown"); + const projectedAt = "2026-04-14T00:00:00.000Z"; + const stoppedAt = "2026-04-14T00:01:00.000Z"; + const harness = await createHarness({ + readModel: makeReadModel([ + { + id: threadId, + session: { + threadId, + status: "starting", + providerName: "codex", + runtimeMode: "full-access", + activeTurnId: null, + lastError: null, + updatedAt: projectedAt, + }, + }, + ]), + }); + const repository = await runtime!.runPromise( + Effect.service(ProviderSessionRuntime.ProviderSessionRuntimeRepository), + ); + + await runtime!.runPromise( + repository.upsert({ + threadId, + providerName: "codex", + providerInstanceId: ProviderInstanceId.make("codex"), + adapterKey: "codex", + runtimeMode: "full-access", + status: "stopped", + lastSeenAt: stoppedAt, + resumeCursor: null, + runtimePayload: { + activeTurnId: null, + lastRuntimeEvent: "provider.stopAll", + lastRuntimeEventAt: stoppedAt, + }, + }), + ); + + const reaper = await runtime!.runPromise(Effect.service(ProviderSessionReaper)); + scope = await Effect.runPromise(Scope.make("sequential")); + await Effect.runPromise(reaper.start().pipe(Scope.provide(scope))); + + await waitFor(() => harness.dispatch.mock.calls.length === 1); + + expect(harness.stopSession).not.toHaveBeenCalled(); + expect(harness.dispatch.mock.calls[0]?.[0]).toMatchObject({ + type: "thread.session.set", + threadId, + expectedSessionUpdatedAt: projectedAt, + createdAt: stoppedAt, + session: { + threadId, + status: "stopped", + providerName: "codex", + providerInstanceId: "codex", + runtimeMode: "full-access", + activeTurnId: null, + lastError: null, + updatedAt: stoppedAt, + }, + }); + }); + + it("does not overwrite a projected session newer than the stopped binding", async () => { + const threadId = ThreadId.make("thread-reaper-stale-stopped-binding"); + const stoppedAt = "2026-04-14T00:00:00.000Z"; + const projectedAt = "2026-04-14T00:01:00.000Z"; + const harness = await createHarness({ + readModel: makeReadModel([ + { + id: threadId, + session: { + threadId, + status: "starting", + providerName: "codex", + runtimeMode: "full-access", + activeTurnId: null, + lastError: null, + updatedAt: projectedAt, + }, + }, + ]), + }); + const repository = await runtime!.runPromise( + Effect.service(ProviderSessionRuntime.ProviderSessionRuntimeRepository), + ); + + await runtime!.runPromise( + repository.upsert({ + threadId, + providerName: "codex", + providerInstanceId: ProviderInstanceId.make("codex"), + adapterKey: "codex", + runtimeMode: "full-access", + status: "stopped", + lastSeenAt: stoppedAt, + resumeCursor: null, + runtimePayload: null, + }), + ); + + const reaper = await runtime!.runPromise(Effect.service(ProviderSessionReaper)); + scope = await Effect.runPromise(Scope.make("sequential")); + await Effect.runPromise(reaper.start().pipe(Scope.provide(scope))); + await runtime!.runPromise(drainFibers); + + expect(harness.dispatch).not.toHaveBeenCalled(); + expect(harness.stopSession).not.toHaveBeenCalled(); + }); + + it("uses a fresh command ID when retrying a rejected reconciliation", async () => { + const threadId = ThreadId.make("thread-reaper-retry-reconciliation"); + const projectedAt = "2026-04-14T00:00:00.000Z"; + const stoppedAt = "2026-04-14T00:01:00.000Z"; + let attempt = 0; + const harness = await createHarness({ + readModel: makeReadModel([ + { + id: threadId, + session: { + threadId, + status: "starting", + providerName: "codex", + runtimeMode: "full-access", + activeTurnId: null, + lastError: null, + updatedAt: projectedAt, + }, + }, + ]), + dispatchImplementation: (command) => { + attempt += 1; + return attempt === 1 + ? Effect.fail( + new OrchestrationCommandInvariantError({ + commandType: command.type, + detail: "simulated concurrent session update", + }), + ) + : Effect.succeed({ sequence: 1 }); + }, + sweepIntervalMs: 1, + }); + const repository = await runtime!.runPromise( + Effect.service(ProviderSessionRuntime.ProviderSessionRuntimeRepository), + ); + + await runtime!.runPromise( + repository.upsert({ + threadId, + providerName: "codex", + providerInstanceId: ProviderInstanceId.make("codex"), + adapterKey: "codex", + runtimeMode: "full-access", + status: "stopped", + lastSeenAt: stoppedAt, + resumeCursor: null, + runtimePayload: null, + }), + ); + + const reaper = await runtime!.runPromise(Effect.service(ProviderSessionReaper)); + scope = await runtime!.runPromise(Scope.make("sequential")); + await runtime!.runPromise(reaper.start().pipe(Scope.provide(scope))); + + await waitFor(() => harness.dispatch.mock.calls.length >= 2); + await Effect.runPromise(Scope.close(scope, Exit.void)); + scope = null; + + const firstCommandId = harness.dispatch.mock.calls[0]?.[0].commandId; + const secondCommandId = harness.dispatch.mock.calls[1]?.[0].commandId; + expect(firstCommandId).toMatch(/^server:provider-session-reconcile:/); + expect(secondCommandId).toMatch(/^server:provider-session-reconcile:/); + expect(secondCommandId).not.toBe(firstCommandId); + }); + it("reaps stale persisted sessions without active turns", async () => { const threadId = ThreadId.make("thread-reaper-stale"); const now = "2026-01-01T00:00:00.000Z"; @@ -362,7 +560,7 @@ describe("ProviderSessionReaper", () => { const reaper = await runtime!.runPromise(Effect.service(ProviderSessionReaper)); scope = await Effect.runPromise(Scope.make("sequential")); await Effect.runPromise(reaper.start().pipe(Scope.provide(scope))); - await Effect.runPromise(drainFibers); + await runtime!.runPromise(drainFibers); expect(harness.stopSession).not.toHaveBeenCalled(); const remaining = await runtime!.runPromise(repository.getByThreadId({ threadId })); @@ -409,9 +607,9 @@ describe("ProviderSessionReaper", () => { ); const reaper = await runtime!.runPromise(Effect.service(ProviderSessionReaper)); - scope = await Effect.runPromise(Scope.make("sequential")); - await Effect.runPromise(reaper.start().pipe(Scope.provide(scope))); - await Effect.runPromise(drainFibers); + scope = await runtime!.runPromise(Scope.make("sequential")); + await runtime!.runPromise(reaper.start().pipe(Scope.provide(scope))); + await runtime!.runPromise(drainFibers); expect(harness.stopSession).not.toHaveBeenCalled(); const remaining = await runtime!.runPromise(repository.getByThreadId({ threadId })); @@ -495,8 +693,8 @@ describe("ProviderSessionReaper", () => { ); const reaper = await runtime!.runPromise(Effect.service(ProviderSessionReaper)); - scope = await Effect.runPromise(Scope.make("sequential")); - await Effect.runPromise(reaper.start().pipe(Scope.provide(scope))); + scope = await runtime!.runPromise(Scope.make("sequential")); + await runtime!.runPromise(reaper.start().pipe(Scope.provide(scope))); await waitFor(() => harness.stopSession.mock.calls.length === 2); @@ -578,8 +776,8 @@ describe("ProviderSessionReaper", () => { ); const reaper = await runtime!.runPromise(Effect.service(ProviderSessionReaper)); - scope = await Effect.runPromise(Scope.make("sequential")); - await Effect.runPromise(reaper.start().pipe(Scope.provide(scope))); + scope = await runtime!.runPromise(Scope.make("sequential")); + await runtime!.runPromise(reaper.start().pipe(Scope.provide(scope))); await waitFor(() => harness.stopSession.mock.calls.length === 2); diff --git a/apps/server/src/provider/Layers/ProviderSessionReaper.ts b/apps/server/src/provider/Layers/ProviderSessionReaper.ts index ca396b40596..16e928cf5ca 100644 --- a/apps/server/src/provider/Layers/ProviderSessionReaper.ts +++ b/apps/server/src/provider/Layers/ProviderSessionReaper.ts @@ -1,12 +1,18 @@ +import { CommandId } from "@t3tools/contracts"; import * as Clock from "effect/Clock"; +import * as Crypto from "effect/Crypto"; import * as Duration from "effect/Duration"; import * as Effect from "effect/Effect"; import * as Layer from "effect/Layer"; import * as Option from "effect/Option"; import * as Schedule from "effect/Schedule"; +import { OrchestrationEngineService } from "../../orchestration/Services/OrchestrationEngine.ts"; import { ProjectionSnapshotQuery } from "../../orchestration/Services/ProjectionSnapshotQuery.ts"; -import { ProviderSessionDirectory } from "../Services/ProviderSessionDirectory.ts"; +import { + ProviderSessionDirectory, + type ProviderRuntimeBindingWithMetadata, +} from "../Services/ProviderSessionDirectory.ts"; import { ProviderSessionReaper, type ProviderSessionReaperShape, @@ -25,7 +31,11 @@ const makeProviderSessionReaper = (options?: ProviderSessionReaperLiveOptions) = Effect.gen(function* () { const providerService = yield* ProviderService; const directory = yield* ProviderSessionDirectory; + const orchestrationEngine = yield* OrchestrationEngineService; const projectionSnapshotQuery = yield* ProjectionSnapshotQuery; + const crypto = yield* Crypto.Crypto; + const serverCommandId = (tag: string) => + crypto.randomUUIDv4.pipe(Effect.map((uuid) => CommandId.make(`server:${tag}:${uuid}`))); const inactivityThresholdMs = Math.max( 1, @@ -33,6 +43,88 @@ const makeProviderSessionReaper = (options?: ProviderSessionReaperLiveOptions) = ); const sweepIntervalMs = Math.max(1, options?.sweepIntervalMs ?? DEFAULT_SWEEP_INTERVAL_MS); + // Provider shutdown can persist "stopped" after runtime event consumers + // have closed, leaving the projection on its last transient status. + const reconcileStoppedBinding = Effect.fn("ProviderSessionReaper.reconcileStoppedBinding")( + function* (binding: ProviderRuntimeBindingWithMetadata) { + const thread = yield* projectionSnapshotQuery + .getThreadShellById(binding.threadId) + .pipe(Effect.map(Option.getOrUndefined)); + const session = thread?.session; + if (!session || session.status === "stopped") { + return; + } + + const stoppedAtMs = Date.parse(binding.lastSeenAt); + if (Number.isNaN(stoppedAtMs)) { + yield* Effect.logWarning("provider.session.reaper.invalid-last-seen", { + threadId: binding.threadId, + provider: binding.provider, + lastSeenAt: binding.lastSeenAt, + }); + return; + } + + const projectedAtMs = Date.parse(session.updatedAt); + if (!Number.isNaN(projectedAtMs) && stoppedAtMs < projectedAtMs) { + yield* Effect.logDebug("provider.session.reaper.skipped-stale-stopped-binding", { + threadId: binding.threadId, + provider: binding.provider, + bindingLastSeenAt: binding.lastSeenAt, + projectedSessionUpdatedAt: session.updatedAt, + }); + return; + } + + yield* orchestrationEngine + .dispatch({ + type: "thread.session.set", + commandId: yield* serverCommandId("provider-session-reconcile"), + threadId: binding.threadId, + expectedSessionUpdatedAt: session.updatedAt, + session: { + threadId: binding.threadId, + status: "stopped", + providerName: binding.provider, + ...(binding.providerInstanceId !== undefined + ? { providerInstanceId: binding.providerInstanceId } + : session.providerInstanceId !== undefined + ? { providerInstanceId: session.providerInstanceId } + : {}), + runtimeMode: binding.runtimeMode ?? session.runtimeMode, + activeTurnId: null, + lastError: session.lastError, + updatedAt: binding.lastSeenAt, + }, + createdAt: binding.lastSeenAt, + }) + .pipe( + Effect.tap(() => + Effect.logInfo("provider.session.reaper.reconciled-stopped-binding", { + threadId: binding.threadId, + provider: binding.provider, + stoppedAt: binding.lastSeenAt, + projectedStatus: session.status, + }), + ), + Effect.catchTag("OrchestrationCommandInvariantError", () => + Effect.logDebug("provider.session.reaper.skipped-concurrent-session-update", { + threadId: binding.threadId, + provider: binding.provider, + projectedSessionUpdatedAt: session.updatedAt, + }), + ), + Effect.catchCause((cause) => + Effect.logWarning("provider.session.reaper.reconcile-stopped-binding-failed", { + threadId: binding.threadId, + provider: binding.provider, + cause, + }), + ), + ); + }, + ); + const sweep = Effect.gen(function* () { const bindings = yield* directory.listBindings(); const now = yield* Clock.currentTimeMillis; @@ -40,6 +132,7 @@ const makeProviderSessionReaper = (options?: ProviderSessionReaperLiveOptions) = for (const binding of bindings) { if (binding.status === "stopped") { + yield* reconcileStoppedBinding(binding); continue; } diff --git a/packages/contracts/src/orchestration.ts b/packages/contracts/src/orchestration.ts index 84b7a8fa07f..2e0cc4bddc9 100644 --- a/packages/contracts/src/orchestration.ts +++ b/packages/contracts/src/orchestration.ts @@ -800,6 +800,7 @@ const ThreadSessionSetCommand = Schema.Struct({ commandId: CommandId, threadId: ThreadId, session: OrchestrationSession, + expectedSessionUpdatedAt: Schema.optional(IsoDateTime), createdAt: IsoDateTime, });