diff --git a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts index 6c3c290096ba4368ea0c2db0399ba3bcac0fa83b..9a871c49d4006142abc4956db49004ee5ee16583 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 039d1872da6f1c6b31fcd5ccf5875bf5ef35961d..9ff7a3b24966b5c8d0f50e82fbf17cdcd4774a7d 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, @@ -119,6 +120,22 @@ export interface EventSinkV2Shape { }, EventSinkV2Error >; + /** + * 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 expected: OrchestrationV2ProviderThread; + readonly events: ReadonlyArray; + }) => Effect.Effect< + { + readonly committed: boolean; + readonly storedEvents: ReadonlyArray; + }, + EventSinkV2Error + >; readonly commitCommand: (input: { readonly commandId: CommandId; readonly threadId: ThreadId; @@ -509,6 +526,39 @@ 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.expected.id, + }); + const expectedJson = yield* ProjectionStore.encodeProviderThreadPayload(input.expected); + + return yield* commitThenPublish( + Effect.gen(function* () { + const rows = yield* sql<{ readonly payload_json: string }>` + SELECT payload_json + FROM orchestration_v2_projection_provider_threads + WHERE provider_thread_id = ${input.expected.id} + LIMIT 1 + `; + if (rows[0]?.payload_json !== expectedJson) { + 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 +854,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/ProjectionStore.ts b/apps/server/src/orchestration-v2/ProjectionStore.ts index 7c156ad71361df6579a78347bd4661f3a9c57768..03ddc771a59525c20596e0ec0a6a04481c7051e0 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 39bf2e2a8766a96837ba9ef726a1fda03f614fb4..36fa2ced0d551ac7140ab63bd2bd6cc1dd71a054 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, @@ -318,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; @@ -400,6 +418,9 @@ const makeProviderAdapter = Effect.fnUntraced(function* ( }), ) : Stream.fromQueue(events), + ...(options.backgroundTasksEndWithProcess === false + ? {} + : { backgroundTasksEndWithProcess: true as const }), ...(options.hasPendingBackgroundWork === undefined ? {} : { hasPendingBackgroundWork: options.hasPendingBackgroundWork }), @@ -458,6 +479,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; @@ -465,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; @@ -478,9 +502,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, @@ -493,6 +519,9 @@ function layerTest(input: { ...(input.hasPendingBackgroundWorkForThread === undefined ? {} : { hasPendingBackgroundWorkForThread: input.hasPendingBackgroundWorkForThread }), + ...(input.backgroundTasksEndWithProcess === undefined + ? {} + : { backgroundTasksEndWithProcess: input.backgroundTasksEndWithProcess }), ...(input.hangSessionScopeClose === undefined ? {} : { hangSessionScopeClose: input.hangSessionScopeClose }), @@ -2769,6 +2798,225 @@ 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 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", + () => + 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 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, without moving updatedAt (not every writer does). + const seeded = providerThreads[0]!; + yield* Ref.set( + beforeUnchangedWrite, + writeProviderThread({ + ...seeded, + pendingBackgroundTasks: [ + ...seeded.pendingBackgroundTasks, + { taskId: "replacement-task", kind: "command" }, + ], + }).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 d7d1e12c10e8cc3e001c662cc6d3c706fa6d68aa..a9c6e2fc9f7fe1e0ba5827d970f415d7186e488f 100644 --- a/apps/server/src/orchestration-v2/ProviderSessionManager.ts +++ b/apps/server/src/orchestration-v2/ProviderSessionManager.ts @@ -774,14 +774,126 @@ 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. 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" || + 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. + 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. The write + // 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, + ) => + Effect.forEach( + released, + ({ threadId, providerSessionId, providerThreadId, taskIds }) => + Effect.gen(function* () { + 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({ + expected: providerThread, + 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; + } + yield* Effect.logWarning( + "orchestration-v2.provider-session-release-roster-conflict", + { + providerSessionId, + providerThreadId, + }, + ); + }), + { 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 +907,7 @@ export const layerWithOptions = ( input.entry.requestEventPermit.withPermits(1), ), ), + Effect.exit(writeReleasedBackgroundWorkEvents(input.releasedBackgroundTasks)), ], { concurrency: 1 }, ).pipe(Effect.flatMap(Exit.asVoidAll)); @@ -937,6 +1050,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 +1063,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 +1091,7 @@ export const layerWithOptions = ( entry, reason: input.reason, ...(input.detail === undefined ? {} : { detail: input.detail }), + releasedBackgroundTasks, releasedAt, }; const recorded = yield* Effect.exit( @@ -2406,6 +2529,7 @@ export const layerWithOptions = ( providerSessionId: input.providerSessionId, reason: "manual_shutdown", ...(input.detail === undefined ? {} : { detail: input.detail }), + detachedThreadId: input.threadId, }); return; } diff --git a/packages/provider-core/src/server/ProviderAdapter.ts b/packages/provider-core/src/server/ProviderAdapter.ts index 7c316ff6c2847a0ab652c120655ebc0757648ef5..48d569137e7a781c5fdcf3c406f463a2d741c61e 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 2694b59907844e6eb4fd33aa33c5e22e99b82bb5..75e8074001598c5c6c5d11d8ddf0f83fcedf5abf 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,