From afa1f03c0ebbffef72310089c5310d70f908ba0b Mon Sep 17 00:00:00 2001 From: Theo Browne Date: Wed, 14 Oct 2026 02:18:25 -0700 Subject: [PATCH 1/4] fix(server): settled and idle-released sessions no longer leave threads Waiting When a provider session is released (Settle or archive detaching it, or idle release), its process is gone, but the background-task roster it kept on its provider threads stayed. The thread kept showing Waiting with a Stop button until the user pressed Stop. Release now records the task ids on the released session's provider threads as soon as the session leaves the live map, including the thread whose detach released it, and removes those ids once the process has closed. Tasks a same-id replacement session reports meanwhile stay. Server shutdown keeps rosters for startup recovery. Fixes #15239 Co-Authored-By: Claude Opus 5.5 --- .../ProviderSessionManager.test.ts | 163 ++++++++++++++++++ .../ProviderSessionManager.ts | 110 +++++++++++- 2 files changed, 270 insertions(+), 3 deletions(-) diff --git a/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts b/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts index 39bf2e2a8766..119e4d179a22 100644 --- a/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts +++ b/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts @@ -2769,6 +2769,169 @@ it.effect("ProviderSessionManagerV2 persists release when session scope close ha }), ); +// Seeds a thread with a background command on a provider thread of the +// session it then opens, and one on a provider thread of another session. +const openSessionWithBackgroundCommand = (fixtureName: string) => + Effect.gen(function* () { + const eventSink = yield* EventSink.EventSinkV2; + const idAllocator = yield* IdAllocator.IdAllocatorV2; + const manager = yield* ProviderSessionManager.ProviderSessionManagerV2; + const now = yield* DateTime.now; + const threadId = yield* idAllocator.allocate.thread({ + fixtureName, + projectId: yield* idAllocator.allocate.project({ fixtureName }), + }); + const providerSessionId = yield* idAllocator.allocate.providerSession({ + providerInstanceId: modelSelection.instanceId, + threadId, + }); + const otherProviderSessionId = yield* idAllocator.allocate.providerSession({ + providerInstanceId: modelSelection.instanceId, + threadId, + }); + const pendingBackgroundTasks = [ + { taskId: "dev-server", kind: "command", description: "python3 -m http.server" }, + ] as const; + const providerThreads = [ + { + ...makeProviderThread({ idAllocator, threadId, providerSessionId, now }), + pendingBackgroundTasks, + }, + { + ...makeProviderThread({ + idAllocator, + threadId, + providerSessionId: otherProviderSessionId, + now, + nativeThreadId: "native-thread-other-session", + }), + pendingBackgroundTasks, + }, + ]; + const writeProviderThread = (payload: OrchestrationV2ProviderThread) => + Effect.gen(function* () { + yield* eventSink.write({ + events: [ + { + id: yield* idAllocator.allocate.event({ threadId }), + type: "provider-thread.updated", + threadId, + driver: CODEX_DRIVER, + occurredAt: now, + payload, + }, + ], + }); + }); + yield* eventSink.write({ + events: [yield* makeThreadCreatedEvent({ idAllocator, threadId, now })], + }); + yield* Effect.forEach(providerThreads, writeProviderThread); + yield* manager.open({ threadId, providerSessionId, modelSelection, runtimePolicy }); + // The task ids each seeded provider thread lists now. + const rosters = Effect.gen(function* () { + const projectionStore = yield* ProjectionStore.ProjectionStoreV2; + const projection = yield* projectionStore.getThreadProjection(threadId); + return providerThreads.map((seeded) => + ( + projection.providerThreads.find((thread) => thread.id === seeded.id) + ?.pendingBackgroundTasks ?? [] + ).map((task) => task.taskId), + ); + }); + return { threadId, providerSessionId, providerThreads, writeProviderThread, rosters }; + }); + +it.effect( + "ProviderSessionManagerV2 clears the background roster of a session a detach releases", + () => + Effect.gen(function* () { + const state = yield* Ref.make(emptyState); + const effect = Effect.gen(function* () { + const manager = yield* ProviderSessionManager.ProviderSessionManagerV2; + const { threadId, providerSessionId, rosters } = yield* openSessionWithBackgroundCommand( + "provider-session-manager-detach-roster", + ); + assert.deepStrictEqual(yield* rosters, [["dev-server"], ["dev-server"]]); + + // Settle and archive detach the thread, which releases an exclusive + // (Claude-like) session. + yield* manager.detach({ providerSessionId, threadId, detail: "Thread settled." }); + + assert.isTrue(Option.isNone(yield* manager.get(providerSessionId))); + assert.deepStrictEqual(yield* rosters, [[], ["dev-server"]]); + }); + + yield* effect.pipe( + Effect.provide( + layerTest({ state, idleTimeoutMs: 60_000, capabilities: ExclusiveCapabilities }), + ), + ); + }), +); + +it.effect("ProviderSessionManagerV2 clears the background roster of an idle-released session", () => + Effect.gen(function* () { + const state = yield* Ref.make(emptyState); + const effect = Effect.gen(function* () { + const manager = yield* ProviderSessionManager.ProviderSessionManagerV2; + const { providerSessionId, rosters } = yield* openSessionWithBackgroundCommand( + "provider-session-manager-idle-roster", + ); + assert.deepStrictEqual(yield* rosters, [["dev-server"], ["dev-server"]]); + + yield* TestClock.adjust("1 second"); + yield* Effect.yieldNow; + + assert.isTrue(Option.isNone(yield* manager.get(providerSessionId))); + assert.deepStrictEqual(yield* rosters, [[], ["dev-server"]]); + }); + + yield* effect.pipe(Effect.provide(layerTest({ state, idleTimeoutMs: 1000 }))); + }), +); + +it.effect( + "ProviderSessionManagerV2 keeps background tasks a replacement reports during release", + () => + Effect.gen(function* () { + const state = yield* Ref.make(emptyState); + const scopeCloseReached = yield* Deferred.make(); + const effect = Effect.gen(function* () { + const manager = yield* ProviderSessionManager.ProviderSessionManagerV2; + const { providerSessionId, providerThreads, writeProviderThread, rosters } = + yield* openSessionWithBackgroundCommand("provider-session-manager-replacement-roster"); + const release = yield* manager.close(providerSessionId).pipe(Effect.forkChild); + + // While the old process closes, a replacement session with the same id + // reports a task of its own next to the dead one. + yield* Deferred.await(scopeCloseReached); + yield* writeProviderThread({ + ...providerThreads[0]!, + pendingBackgroundTasks: [ + ...providerThreads[0]!.pendingBackgroundTasks, + { taskId: "replacement-task", kind: "command" }, + ], + }); + yield* TestClock.adjust("30 seconds"); + yield* Fiber.join(release); + + assert.deepStrictEqual(yield* rosters, [["replacement-task"], ["dev-server"]]); + }); + + yield* effect.pipe( + Effect.provide( + layerTest({ + state, + idleTimeoutMs: 60_000, + hangSessionScopeClose: true, + scopeCloseReached, + }), + ), + ); + }), +); + it.effect("ProviderSessionManagerV2 defers idle release while background work is pending", () => Effect.gen(function* () { const state = yield* Ref.make(emptyState); diff --git a/apps/server/src/orchestration-v2/ProviderSessionManager.ts b/apps/server/src/orchestration-v2/ProviderSessionManager.ts index d7d1e12c10e8..2b6f74f58aaf 100644 --- a/apps/server/src/orchestration-v2/ProviderSessionManager.ts +++ b/apps/server/src/orchestration-v2/ProviderSessionManager.ts @@ -774,14 +774,106 @@ export const layerWithOptions = ( } }); + // Reads the background tasks a released session's process left on its + // provider threads. Read as soon as the session leaves the live map: + // after that, a replacement session with the same id can open and report + // tasks of its own on the same provider threads. Server shutdown keeps + // rosters, since startup recovery clears them and records the cancelled + // work. + const readReleasedBackgroundTasks = (input: { + readonly entry: LiveSessionEntry; + readonly reason: ProviderSessionReleaseReason; + readonly detachedThreadId?: ThreadId; + }) => + Effect.gen(function* () { + if (input.reason === "server_shutdown") return []; + const providerSessionId = input.entry.runtime.providerSessionId; + // A detach that released the session already removed its thread + // from the attached set. + const threadIds = new Set(input.entry.attachedThreadIds); + if (input.detachedThreadId !== undefined) threadIds.add(input.detachedThreadId); + const released = []; + for (const threadId of threadIds) { + const projection = yield* projectionStore.getThreadRecords(threadId, [ + "providerThreads", + ]); + for (const providerThread of projection.providerThreads) { + const tasks = providerThread.pendingBackgroundTasks ?? []; + if (providerThread.providerSessionId !== providerSessionId || tasks.length === 0) { + continue; + } + released.push({ + threadId, + providerSessionId, + providerThreadId: providerThread.id, + taskIds: new Set(tasks.map((task) => task.taskId)), + }); + } + } + return released; + }).pipe( + Effect.catchCause((cause) => + Effect.logWarning("orchestration-v2.provider-session-release-roster-read-failed", { + providerSessionId: input.entry.runtime.providerSessionId, + cause, + }).pipe(Effect.as([])), + ), + ); + type ReleasedBackgroundTasks = Effect.Success< + ReturnType + >[number]; + + // Removes the tasks `readReleasedBackgroundTasks` found from their + // rosters. Their process is gone, so nothing will report them ending, + // and a leftover roster keeps the thread Waiting (#15239). Tasks a + // replacement session reported since are left in place. + const writeReleasedBackgroundWorkEvents = ( + released: ReadonlyArray, + ) => + Effect.forEach( + released, + ({ threadId, providerSessionId, providerThreadId, taskIds }) => + Effect.gen(function* () { + const projection = yield* projectionStore.getThreadRecords(threadId, [ + "providerThreads", + ]); + const providerThread = projection.providerThreads.find( + (candidate) => candidate.id === providerThreadId, + ); + const tasks = providerThread?.pendingBackgroundTasks ?? []; + const remaining = tasks.filter((task) => !taskIds.has(task.taskId)); + if (providerThread === undefined || remaining.length === tasks.length) return; + const now = yield* DateTime.now; + yield* eventSink.write({ + events: [ + { + id: yield* idAllocator.allocate.event({ threadId, providerSessionId }), + type: "provider-thread.updated", + threadId, + driver: providerThread.driver, + providerInstanceId: providerThread.providerInstanceId, + occurredAt: now, + payload: { + ...providerThread, + pendingBackgroundTasks: remaining, + updatedAt: now, + }, + }, + ], + }); + }), + { discard: true }, + ); + // Records a released session as stopped and resolves the live runtime - // requests it left. Each write runs even if the other fails. Once a - // replacement session opens with the same id, it owns the session status, - // so only the requests are settled. + // requests and background-task rosters it left. Each write runs even if + // another fails. Once a replacement session opens with the same id, it + // owns the session status, so only the requests are settled. const writeReleaseRecords = (input: { readonly entry: LiveSessionEntry; readonly reason: ProviderSessionReleaseReason; readonly detail?: string; + readonly releasedBackgroundTasks: ReadonlyArray; readonly releasedAt: DateTime.Utc; readonly replaced: boolean; }) => @@ -795,6 +887,7 @@ export const layerWithOptions = ( input.entry.requestEventPermit.withPermits(1), ), ), + Effect.exit(writeReleasedBackgroundWorkEvents(input.releasedBackgroundTasks)), ], { concurrency: 1 }, ).pipe(Effect.flatMap(Exit.asVoidAll)); @@ -937,6 +1030,8 @@ export const layerWithOptions = ( readonly providerSessionId: ProviderSessionId; readonly reason: ProviderSessionReleaseReason; readonly detail?: string; + /** The thread whose detach released the session; it is no longer attached. */ + readonly detachedThreadId?: ThreadId; readonly cancelIdleFiber?: boolean; readonly onlyIfIdleGeneration?: number; readonly gracefulSubscribers?: boolean; @@ -948,6 +1043,13 @@ export const layerWithOptions = ( onNone: () => Effect.void, onSome: (entry) => Effect.gen(function* () { + const releasedBackgroundTasks = yield* readReleasedBackgroundTasks({ + entry, + reason: input.reason, + ...(input.detachedThreadId === undefined + ? {} + : { detachedThreadId: input.detachedThreadId }), + }); if (input.cancelIdleFiber !== false) { yield* cancelIdleFiber(entry.idleFiber); } @@ -969,6 +1071,7 @@ export const layerWithOptions = ( entry, reason: input.reason, ...(input.detail === undefined ? {} : { detail: input.detail }), + releasedBackgroundTasks, releasedAt, }; const recorded = yield* Effect.exit( @@ -2406,6 +2509,7 @@ export const layerWithOptions = ( providerSessionId: input.providerSessionId, reason: "manual_shutdown", ...(input.detail === undefined ? {} : { detail: input.detail }), + detachedThreadId: input.threadId, }); return; } From 7d5c543c808de6f873905be6f1ddb357eab3a8d6 Mon Sep 17 00:00:00 2001 From: Theo Browne Date: Wed, 14 Oct 2026 02:57:38 -0700 Subject: [PATCH 2/4] fix(server): release cleanup no longer overwrites a provider update The roster cleanup read the provider thread and wrote it back in two steps, so a provider update landing in between was overwritten. It now commits through a compare-and-set on the row's updatedAt and retries. Co-Authored-By: Claude Opus 5.5 --- apps/server/src/orchestration-v2/EventSink.ts | 54 ++++++++++++++++ .../ProviderSessionManager.test.ts | 59 +++++++++++++++++- .../ProviderSessionManager.ts | 61 +++++++++++-------- 3 files changed, 144 insertions(+), 30 deletions(-) diff --git a/apps/server/src/orchestration-v2/EventSink.ts b/apps/server/src/orchestration-v2/EventSink.ts index 039d1872da6f..3fa362228fdc 100644 --- a/apps/server/src/orchestration-v2/EventSink.ts +++ b/apps/server/src/orchestration-v2/EventSink.ts @@ -119,6 +119,22 @@ export interface EventSinkV2Shape { }, EventSinkV2Error >; + /** + * Atomically commit only while the provider thread row still has the + * `updatedAt` its writer read, so a read-modify-write of the row cannot + * overwrite an update a provider published in between. + */ + readonly writeIfProviderThreadUnchanged: (input: { + readonly providerThreadId: ProviderThreadId; + readonly expectedUpdatedAt: DateTime.Utc; + readonly events: ReadonlyArray; + }) => Effect.Effect< + { + readonly committed: boolean; + readonly storedEvents: ReadonlyArray; + }, + EventSinkV2Error + >; readonly commitCommand: (input: { readonly commandId: CommandId; readonly threadId: ThreadId; @@ -509,6 +525,38 @@ const layerBase: Layer.Layer< ); }); + const writeIfProviderThreadUnchangedEffect = Effect.fn( + "orchestrationV2.EventSink.writeIfProviderThreadUnchanged", + )(function* (input: Parameters[0]) { + yield* Effect.annotateCurrentSpan({ + "orchestration_v2.event_count": input.events.length, + "orchestration_v2.provider_thread_id": input.providerThreadId, + }); + + return yield* commitThenPublish( + Effect.gen(function* () { + const rows = yield* sql<{ readonly updated_at: string | null }>` + SELECT updated_at + FROM orchestration_v2_projection_provider_threads + WHERE provider_thread_id = ${input.providerThreadId} + LIMIT 1 + `; + if (rows[0]?.updated_at !== DateTime.formatIso(input.expectedUpdatedAt)) { + return { + committed: false as const, + storedEvents: [] as ReadonlyArray, + }; + } + + const normalized = yield* normalizeEvents(input.events); + const storedEvents = yield* eventStore.append({ events: normalized }); + yield* applyStoredEvents(storedEvents); + return { committed: true as const, storedEvents }; + }), + (result) => (result.committed ? publishStoredEvents(result.storedEvents) : Effect.void), + ); + }); + const existingCommandResult = (commandId: CommandId) => Effect.gen(function* () { const existing = yield* commandReceipts.getByCommandId(commandId); @@ -804,6 +852,12 @@ const layerBase: Layer.Layer< }), ), ), + writeIfProviderThreadUnchanged: (input) => + writeIfProviderThreadUnchangedEffect(input).pipe( + Effect.mapError( + (cause) => new EventSinkWriteError({ eventCount: input.events.length, cause }), + ), + ), commitCommand: (input) => commitCommandEffect(input).pipe( Effect.mapError( diff --git a/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts b/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts index 119e4d179a22..e01404e75a83 100644 --- a/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts +++ b/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts @@ -147,6 +147,22 @@ const layerPausingAttachEventSink = (pause: { }), ).pipe(Layer.provide(layerTestEventSink)); +const layerBeforeUnchangedWriteEventSink = (before: Ref.Ref>) => + Layer.effect( + EventSink.EventSinkV2, + Effect.gen(function* () { + const delegate = yield* EventSink.EventSinkV2; + return EventSink.EventSinkV2.of({ + ...delegate, + writeIfProviderThreadUnchanged: (input) => + Ref.getAndSet(before, Effect.void).pipe( + Effect.flatten, + Effect.andThen(delegate.writeIfProviderThreadUnchanged(input)), + ), + }); + }), + ).pipe(Layer.provide(layerTestEventSink)); + const CodexCapabilities: OrchestrationV2ProviderCapabilities = CodexProviderCapabilitiesV2; const ExclusiveCapabilities: OrchestrationV2ProviderCapabilities = { ...CodexCapabilities, @@ -458,6 +474,8 @@ function layerTest(input: { readonly failReleaseEventWrites?: boolean; readonly flakyReleaseWrites?: FlakyReleaseWrites; readonly pauseAttachWrite?: Parameters[0]; + /** Runs once, right before the next provider-thread compare-and-set write. */ + readonly beforeUnchangedWrite?: Ref.Ref>; /** Once armed, holds the next credential lookup until it is interrupted. */ readonly pauseResolve?: { readonly armed: Ref.Ref; @@ -478,9 +496,11 @@ function layerTest(input: { ? layerFlakyReleaseEventSink(input.flakyReleaseWrites) : input.pauseAttachWrite !== undefined ? layerPausingAttachEventSink(input.pauseAttachWrite) - : input.failReleaseEventWrites - ? layerFailingReleaseEventSink - : layerTestEventSink; + : input.beforeUnchangedWrite !== undefined + ? layerBeforeUnchangedWriteEventSink(input.beforeUnchangedWrite) + : input.failReleaseEventWrites + ? layerFailingReleaseEventSink + : layerTestEventSink; const layerRegistry = Layer.unwrap( makeProviderAdapter(input.state, { failEventStream: input.failEventStream ?? false, @@ -2932,6 +2952,39 @@ it.effect( }), ); +it.effect("ProviderSessionManagerV2 keeps a provider update that lands while release clears", () => + Effect.gen(function* () { + const state = yield* Ref.make(emptyState); + const beforeUnchangedWrite = yield* Ref.make>(Effect.void); + const effect = Effect.gen(function* () { + const manager = yield* ProviderSessionManager.ProviderSessionManagerV2; + const { providerSessionId, providerThreads, writeProviderThread, rosters } = + yield* openSessionWithBackgroundCommand("provider-session-manager-raced-roster"); + // A replacement reports a task after cleanup read the row, before it writes. + const seeded = providerThreads[0]!; + yield* Ref.set( + beforeUnchangedWrite, + writeProviderThread({ + ...seeded, + pendingBackgroundTasks: [ + ...seeded.pendingBackgroundTasks, + { taskId: "replacement-task", kind: "command" }, + ], + updatedAt: DateTime.add(seeded.updatedAt, { seconds: 1 }), + }).pipe(Effect.orDie), + ); + + yield* manager.close(providerSessionId); + + assert.deepStrictEqual(yield* rosters, [["replacement-task"], ["dev-server"]]); + }); + + yield* effect.pipe( + Effect.provide(layerTest({ state, idleTimeoutMs: 60_000, beforeUnchangedWrite })), + ); + }), +); + it.effect("ProviderSessionManagerV2 defers idle release while background work is pending", () => Effect.gen(function* () { const state = yield* Ref.make(emptyState); diff --git a/apps/server/src/orchestration-v2/ProviderSessionManager.ts b/apps/server/src/orchestration-v2/ProviderSessionManager.ts index 2b6f74f58aaf..fa78d4c97a0a 100644 --- a/apps/server/src/orchestration-v2/ProviderSessionManager.ts +++ b/apps/server/src/orchestration-v2/ProviderSessionManager.ts @@ -826,7 +826,9 @@ export const layerWithOptions = ( // Removes the tasks `readReleasedBackgroundTasks` found from their // rosters. Their process is gone, so nothing will report them ending, // and a leftover roster keeps the thread Waiting (#15239). Tasks a - // replacement session reported since are left in place. + // replacement session reported since are left in place. The write + // commits only if the row did not change after it was read, and is + // retried a few times when a provider update lands in between. const writeReleasedBackgroundWorkEvents = ( released: ReadonlyArray, ) => @@ -834,33 +836,38 @@ export const layerWithOptions = ( released, ({ threadId, providerSessionId, providerThreadId, taskIds }) => Effect.gen(function* () { - const projection = yield* projectionStore.getThreadRecords(threadId, [ - "providerThreads", - ]); - const providerThread = projection.providerThreads.find( - (candidate) => candidate.id === providerThreadId, - ); - const tasks = providerThread?.pendingBackgroundTasks ?? []; - const remaining = tasks.filter((task) => !taskIds.has(task.taskId)); - if (providerThread === undefined || remaining.length === tasks.length) return; - const now = yield* DateTime.now; - yield* eventSink.write({ - events: [ - { - id: yield* idAllocator.allocate.event({ threadId, providerSessionId }), - type: "provider-thread.updated", - threadId, - driver: providerThread.driver, - providerInstanceId: providerThread.providerInstanceId, - occurredAt: now, - payload: { - ...providerThread, - pendingBackgroundTasks: remaining, - updatedAt: now, + for (let attempt = 0; attempt < 3; attempt++) { + const projection = yield* projectionStore.getThreadRecords(threadId, [ + "providerThreads", + ]); + const providerThread = projection.providerThreads.find( + (candidate) => candidate.id === providerThreadId, + ); + const tasks = providerThread?.pendingBackgroundTasks ?? []; + const remaining = tasks.filter((task) => !taskIds.has(task.taskId)); + if (providerThread === undefined || remaining.length === tasks.length) return; + const now = yield* DateTime.now; + const { committed } = yield* eventSink.writeIfProviderThreadUnchanged({ + providerThreadId, + expectedUpdatedAt: providerThread.updatedAt, + events: [ + { + id: yield* idAllocator.allocate.event({ threadId, providerSessionId }), + type: "provider-thread.updated", + threadId, + driver: providerThread.driver, + providerInstanceId: providerThread.providerInstanceId, + occurredAt: now, + payload: { + ...providerThread, + pendingBackgroundTasks: remaining, + updatedAt: now, + }, }, - }, - ], - }); + ], + }); + if (committed) return; + } }), { discard: true }, ); From cfde4ef3507ffb16e43e9418ebaf4077541d61c3 Mon Sep 17 00:00:00 2001 From: Theo Browne Date: Wed, 14 Oct 2026 03:15:27 -0700 Subject: [PATCH 3/4] fix(server): release cleanup keeps rosters that can outlive the session Only runtimes whose background tasks end with their process (Claude, Muse) opt in; OpenCode's background subagents run on a shared server. The guard now compares the whole stored row, since not every provider-thread writer moves updatedAt. Co-Authored-By: Claude Opus 5.5 --- .../Adapters/ClaudeAdapterV2.ts | 2 ++ apps/server/src/orchestration-v2/EventSink.ts | 22 ++++++------ .../src/orchestration-v2/ProjectionStore.ts | 3 +- .../ProviderSessionManager.test.ts | 36 +++++++++++++++++-- .../ProviderSessionManager.ts | 16 ++++++--- .../src/server/ProviderAdapter.ts | 8 +++++ packages/provider-muse/src/server/adapter.ts | 2 ++ 7 files changed, 71 insertions(+), 18 deletions(-) diff --git a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts index 6c3c290096ba..9a871c49d400 100644 --- a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts @@ -9065,6 +9065,8 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( getModelContextWindow: (selection) => resolveClaudeCatalogContextWindowTokens(BUNDLED_CLAUDE_MODEL_CATALOG, selection), events: Stream.fromEffectRepeat(Queue.take(events)), + // Background shells, agents, monitors, and workflows end with the CLI. + backgroundTasksEndWithProcess: true, hasPendingBackgroundWork: Effect.gen(function* () { // Session capability: any native thread with pending work pins idle. for (const roster of (yield* Ref.get(pendingBackgroundTasksByNativeThread)).values()) { diff --git a/apps/server/src/orchestration-v2/EventSink.ts b/apps/server/src/orchestration-v2/EventSink.ts index 3fa362228fdc..9ff7a3b24966 100644 --- a/apps/server/src/orchestration-v2/EventSink.ts +++ b/apps/server/src/orchestration-v2/EventSink.ts @@ -2,6 +2,7 @@ import { CommandId, type OrchestrationV2Run, OrchestrationV2DomainEvent, + type OrchestrationV2ProviderThread, OrchestrationV2StoredEvent, ProviderThreadId, RunAttemptId, @@ -120,13 +121,13 @@ export interface EventSinkV2Shape { EventSinkV2Error >; /** - * Atomically commit only while the provider thread row still has the - * `updatedAt` its writer read, so a read-modify-write of the row cannot - * overwrite an update a provider published in between. + * Atomically commit only while the stored provider thread still equals + * `expected`, the row its writer read, so a read-modify-write of the row + * cannot overwrite an update published in between. The whole row is + * compared because not every writer moves `updatedAt`. */ readonly writeIfProviderThreadUnchanged: (input: { - readonly providerThreadId: ProviderThreadId; - readonly expectedUpdatedAt: DateTime.Utc; + readonly expected: OrchestrationV2ProviderThread; readonly events: ReadonlyArray; }) => Effect.Effect< { @@ -530,18 +531,19 @@ const layerBase: Layer.Layer< )(function* (input: Parameters[0]) { yield* Effect.annotateCurrentSpan({ "orchestration_v2.event_count": input.events.length, - "orchestration_v2.provider_thread_id": input.providerThreadId, + "orchestration_v2.provider_thread_id": input.expected.id, }); + const expectedJson = yield* ProjectionStore.encodeProviderThreadPayload(input.expected); return yield* commitThenPublish( Effect.gen(function* () { - const rows = yield* sql<{ readonly updated_at: string | null }>` - SELECT updated_at + const rows = yield* sql<{ readonly payload_json: string }>` + SELECT payload_json FROM orchestration_v2_projection_provider_threads - WHERE provider_thread_id = ${input.providerThreadId} + WHERE provider_thread_id = ${input.expected.id} LIMIT 1 `; - if (rows[0]?.updated_at !== DateTime.formatIso(input.expectedUpdatedAt)) { + if (rows[0]?.payload_json !== expectedJson) { return { committed: false as const, storedEvents: [] as ReadonlyArray, diff --git a/apps/server/src/orchestration-v2/ProjectionStore.ts b/apps/server/src/orchestration-v2/ProjectionStore.ts index 7c156ad71361..03ddc771a595 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.ts @@ -1056,7 +1056,8 @@ const encodeSubagentPayload = Schema.encodeEffect( const encodeProviderSessionPayload = Schema.encodeEffect( Schema.fromJsonString(OrchestrationV2ProviderSessionJsonSchema), ); -const encodeProviderThreadPayload = Schema.encodeEffect( +/** Encodes a provider thread exactly as its projection row stores `payload_json`. */ +export const encodeProviderThreadPayload = Schema.encodeEffect( Schema.fromJsonString(OrchestrationV2ProviderThreadJsonSchema), ); const encodeProviderTurnPayload = Schema.encodeEffect( diff --git a/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts b/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts index e01404e75a83..36fa2ced0d55 100644 --- a/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts +++ b/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts @@ -334,6 +334,8 @@ const makeProviderAdapter = Effect.fnUntraced(function* ( }) => Effect.Effect; readonly hasPendingBackgroundWork?: Effect.Effect; readonly hasPendingBackgroundWorkForThread?: Effect.Effect; + /** False models a runtime whose background tasks can outlive its process. */ + readonly backgroundTasksEndWithProcess?: boolean; readonly hangSessionScopeClose?: boolean; readonly startTurn?: Effect.Effect; readonly beforeUnload?: Effect.Effect; @@ -416,6 +418,9 @@ const makeProviderAdapter = Effect.fnUntraced(function* ( }), ) : Stream.fromQueue(events), + ...(options.backgroundTasksEndWithProcess === false + ? {} + : { backgroundTasksEndWithProcess: true as const }), ...(options.hasPendingBackgroundWork === undefined ? {} : { hasPendingBackgroundWork: options.hasPendingBackgroundWork }), @@ -483,6 +488,7 @@ function layerTest(input: { }; readonly hasPendingBackgroundWork?: Effect.Effect; readonly hasPendingBackgroundWorkForThread?: Effect.Effect; + readonly backgroundTasksEndWithProcess?: boolean; readonly hangSessionScopeClose?: boolean; readonly startTurn?: Effect.Effect; readonly beforeUnload?: Effect.Effect; @@ -513,6 +519,9 @@ function layerTest(input: { ...(input.hasPendingBackgroundWorkForThread === undefined ? {} : { hasPendingBackgroundWorkForThread: input.hasPendingBackgroundWorkForThread }), + ...(input.backgroundTasksEndWithProcess === undefined + ? {} + : { backgroundTasksEndWithProcess: input.backgroundTasksEndWithProcess }), ...(input.hangSessionScopeClose === undefined ? {} : { hangSessionScopeClose: input.hangSessionScopeClose }), @@ -2911,6 +2920,29 @@ it.effect("ProviderSessionManagerV2 clears the background roster of an idle-rele }), ); +it.effect("ProviderSessionManagerV2 keeps the roster of a runtime whose tasks can outlive it", () => + Effect.gen(function* () { + const state = yield* Ref.make(emptyState); + const effect = Effect.gen(function* () { + const manager = yield* ProviderSessionManager.ProviderSessionManagerV2; + const { providerSessionId, rosters } = yield* openSessionWithBackgroundCommand( + "provider-session-manager-outliving-roster", + ); + + // Like OpenCode, whose background subagents run on a shared server. + yield* manager.close(providerSessionId); + + assert.deepStrictEqual(yield* rosters, [["dev-server"], ["dev-server"]]); + }); + + yield* effect.pipe( + Effect.provide( + layerTest({ state, idleTimeoutMs: 60_000, backgroundTasksEndWithProcess: false }), + ), + ); + }), +); + it.effect( "ProviderSessionManagerV2 keeps background tasks a replacement reports during release", () => @@ -2960,7 +2992,8 @@ it.effect("ProviderSessionManagerV2 keeps a provider update that lands while rel const manager = yield* ProviderSessionManager.ProviderSessionManagerV2; const { providerSessionId, providerThreads, writeProviderThread, rosters } = yield* openSessionWithBackgroundCommand("provider-session-manager-raced-roster"); - // A replacement reports a task after cleanup read the row, before it writes. + // A replacement reports a task after cleanup read the row, before it + // writes, without moving updatedAt (not every writer does). const seeded = providerThreads[0]!; yield* Ref.set( beforeUnchangedWrite, @@ -2970,7 +3003,6 @@ it.effect("ProviderSessionManagerV2 keeps a provider update that lands while rel ...seeded.pendingBackgroundTasks, { taskId: "replacement-task", kind: "command" }, ], - updatedAt: DateTime.add(seeded.updatedAt, { seconds: 1 }), }).pipe(Effect.orDie), ); diff --git a/apps/server/src/orchestration-v2/ProviderSessionManager.ts b/apps/server/src/orchestration-v2/ProviderSessionManager.ts index fa78d4c97a0a..4d00a0c1e92a 100644 --- a/apps/server/src/orchestration-v2/ProviderSessionManager.ts +++ b/apps/server/src/orchestration-v2/ProviderSessionManager.ts @@ -779,14 +779,21 @@ export const layerWithOptions = ( // after that, a replacement session with the same id can open and report // tasks of its own on the same provider threads. Server shutdown keeps // rosters, since startup recovery clears them and records the cancelled - // work. + // work. Only runtimes whose tasks end with their process opt in (Claude, + // Muse). OpenCode's background subagents run on a server shared by its + // sessions and can outlive this one, so their roster stays. const readReleasedBackgroundTasks = (input: { readonly entry: LiveSessionEntry; readonly reason: ProviderSessionReleaseReason; readonly detachedThreadId?: ThreadId; }) => Effect.gen(function* () { - if (input.reason === "server_shutdown") return []; + if ( + input.reason === "server_shutdown" || + input.entry.runtime.backgroundTasksEndWithProcess !== true + ) { + return []; + } const providerSessionId = input.entry.runtime.providerSessionId; // A detach that released the session already removed its thread // from the attached set. @@ -827,7 +834,7 @@ export const layerWithOptions = ( // rosters. Their process is gone, so nothing will report them ending, // and a leftover roster keeps the thread Waiting (#15239). Tasks a // replacement session reported since are left in place. The write - // commits only if the row did not change after it was read, and is + // commits only if the stored row still equals what was read, and is // retried a few times when a provider update lands in between. const writeReleasedBackgroundWorkEvents = ( released: ReadonlyArray, @@ -848,8 +855,7 @@ export const layerWithOptions = ( if (providerThread === undefined || remaining.length === tasks.length) return; const now = yield* DateTime.now; const { committed } = yield* eventSink.writeIfProviderThreadUnchanged({ - providerThreadId, - expectedUpdatedAt: providerThread.updatedAt, + expected: providerThread, events: [ { id: yield* idAllocator.allocate.event({ threadId, providerSessionId }), diff --git a/packages/provider-core/src/server/ProviderAdapter.ts b/packages/provider-core/src/server/ProviderAdapter.ts index 7c316ff6c284..48d569137e7a 100644 --- a/packages/provider-core/src/server/ProviderAdapter.ts +++ b/packages/provider-core/src/server/ProviderAdapter.ts @@ -565,6 +565,14 @@ export interface ProviderAdapterV2SessionRuntime { readonly hasPendingBackgroundWorkForThread?: ( providerThread: OrchestrationV2ProviderThread, ) => Effect.Effect; + /** + * Set when every background task this session lists on its provider threads + * runs inside the session's own process and ends when it closes. The session + * manager then clears those tasks when it releases the session (#15239). + * Leave unset when a task can outlive the session, for example on a shared + * server, so its thread keeps showing Waiting. + */ + readonly backgroundTasksEndWithProcess?: true; /** * Capacity for the requested model/options, independent of native thread usage. * `cwd` is the thread's working directory, for providers whose project config diff --git a/packages/provider-muse/src/server/adapter.ts b/packages/provider-muse/src/server/adapter.ts index 2694b5990784..75e807400159 100644 --- a/packages/provider-muse/src/server/adapter.ts +++ b/packages/provider-muse/src/server/adapter.ts @@ -2107,6 +2107,8 @@ export const makeMuseAdapterV2 = Effect.fn("makeMuseAdapterV2")(function* ( providerSessionId: input.providerSessionId, providerSession: session, events: Stream.fromQueue(events), + // Workflows and subagents run in this session's own `muse` host. + backgroundTasksEndWithProcess: true, // A running workflow or a held Muse turn keeps the host: idle release must wait. hasPendingBackgroundWork: Effect.sync( () => observedChildren.size > 0 || wake !== undefined, From f2248534a197b12d1aa1485c9d49a333abe3215c Mon Sep 17 00:00:00 2001 From: Theo Browne Date: Wed, 14 Oct 2026 03:27:50 -0700 Subject: [PATCH 4/4] fix(server): log when release cleanup gives up on a changing roster Co-Authored-By: Claude Opus 5.5 --- apps/server/src/orchestration-v2/ProviderSessionManager.ts | 7 +++++++ 1 file changed, 7 insertions(+) diff --git a/apps/server/src/orchestration-v2/ProviderSessionManager.ts b/apps/server/src/orchestration-v2/ProviderSessionManager.ts index 4d00a0c1e92a..a9c6e2fc9f7f 100644 --- a/apps/server/src/orchestration-v2/ProviderSessionManager.ts +++ b/apps/server/src/orchestration-v2/ProviderSessionManager.ts @@ -874,6 +874,13 @@ export const layerWithOptions = ( }); if (committed) return; } + yield* Effect.logWarning( + "orchestration-v2.provider-session-release-roster-conflict", + { + providerSessionId, + providerThreadId, + }, + ); }), { discard: true }, );