authorgravatar for me@t3.ggTheo Browne <me@t3.gg> 2026-10-14 03:33:34-07:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2026-10-14 03:33:34-07:00
log519d64f6cec0a0f9ed18fbca4a3f8695cac44506
treed03bfae80f5ede9118aeef0399142ce79eb9c380
parentadb04202b25c1c45255e76be9f03f0b346b9c5c4
signature Signed by PGP key B5690EEEBB952194

fix(server): settled and idle-released threads no longer stay Waiting (#19355)

Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>

7 files changed, 448 insertions(+), 7 deletions(-)

apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts+2
...@@ -9065,6 +9065,8 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* (...@@ -9065,6 +9065,8 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* (
9065 getModelContextWindow: (selection) =>9065 getModelContextWindow: (selection) =>
9066 resolveClaudeCatalogContextWindowTokens(BUNDLED_CLAUDE_MODEL_CATALOG, selection),9066 resolveClaudeCatalogContextWindowTokens(BUNDLED_CLAUDE_MODEL_CATALOG, selection),
9067 events: Stream.fromEffectRepeat(Queue.take(events)),9067 events: Stream.fromEffectRepeat(Queue.take(events)),
9068 // Background shells, agents, monitors, and workflows end with the CLI.
9069 backgroundTasksEndWithProcess: true,
9068 hasPendingBackgroundWork: Effect.gen(function* () {9070 hasPendingBackgroundWork: Effect.gen(function* () {
9069 // Session capability: any native thread with pending work pins idle.9071 // Session capability: any native thread with pending work pins idle.
9070 for (const roster of (yield* Ref.get(pendingBackgroundTasksByNativeThread)).values()) {9072 for (const roster of (yield* Ref.get(pendingBackgroundTasksByNativeThread)).values()) {
apps/server/src/orchestration-v2/EventSink.ts+56
...@@ -2,6 +2,7 @@ import {...@@ -2,6 +2,7 @@ import {
2 CommandId,2 CommandId,
3 type OrchestrationV2Run,3 type OrchestrationV2Run,
4 OrchestrationV2DomainEvent,4 OrchestrationV2DomainEvent,
5 type OrchestrationV2ProviderThread,
5 OrchestrationV2StoredEvent,6 OrchestrationV2StoredEvent,
6 ProviderThreadId,7 ProviderThreadId,
7 RunAttemptId,8 RunAttemptId,
...@@ -119,6 +120,22 @@ export interface EventSinkV2Shape {...@@ -119,6 +120,22 @@ export interface EventSinkV2Shape {
119 },120 },
120 EventSinkV2Error121 EventSinkV2Error
121 >;122 >;
123 /**
124 * Atomically commit only while the stored provider thread still equals
125 * `expected`, the row its writer read, so a read-modify-write of the row
126 * cannot overwrite an update published in between. The whole row is
127 * compared because not every writer moves `updatedAt`.
128 */
129 readonly writeIfProviderThreadUnchanged: (input: {
130 readonly expected: OrchestrationV2ProviderThread;
131 readonly events: ReadonlyArray<OrchestrationV2DomainEvent>;
132 }) => Effect.Effect<
133 {
134 readonly committed: boolean;
135 readonly storedEvents: ReadonlyArray<OrchestrationV2StoredEvent>;
136 },
137 EventSinkV2Error
138 >;
122 readonly commitCommand: (input: {139 readonly commitCommand: (input: {
123 readonly commandId: CommandId;140 readonly commandId: CommandId;
124 readonly threadId: ThreadId;141 readonly threadId: ThreadId;
...@@ -509,6 +526,39 @@ const layerBase: Layer.Layer<...@@ -509,6 +526,39 @@ const layerBase: Layer.Layer<
509 );526 );
510 });527 });
511528
529 const writeIfProviderThreadUnchangedEffect = Effect.fn(
530 "orchestrationV2.EventSink.writeIfProviderThreadUnchanged",
531 )(function* (input: Parameters<EventSinkV2Shape["writeIfProviderThreadUnchanged"]>[0]) {
532 yield* Effect.annotateCurrentSpan({
533 "orchestration_v2.event_count": input.events.length,
534 "orchestration_v2.provider_thread_id": input.expected.id,
535 });
536 const expectedJson = yield* ProjectionStore.encodeProviderThreadPayload(input.expected);
537
538 return yield* commitThenPublish(
539 Effect.gen(function* () {
540 const rows = yield* sql<{ readonly payload_json: string }>`
541 SELECT payload_json
542 FROM orchestration_v2_projection_provider_threads
543 WHERE provider_thread_id = ${input.expected.id}
544 LIMIT 1
545 `;
546 if (rows[0]?.payload_json !== expectedJson) {
547 return {
548 committed: false as const,
549 storedEvents: [] as ReadonlyArray<OrchestrationV2StoredEvent>,
550 };
551 }
552
553 const normalized = yield* normalizeEvents(input.events);
554 const storedEvents = yield* eventStore.append({ events: normalized });
555 yield* applyStoredEvents(storedEvents);
556 return { committed: true as const, storedEvents };
557 }),
558 (result) => (result.committed ? publishStoredEvents(result.storedEvents) : Effect.void),
559 );
560 });
561
512 const existingCommandResult = (commandId: CommandId) =>562 const existingCommandResult = (commandId: CommandId) =>
513 Effect.gen(function* () {563 Effect.gen(function* () {
514 const existing = yield* commandReceipts.getByCommandId(commandId);564 const existing = yield* commandReceipts.getByCommandId(commandId);
...@@ -804,6 +854,12 @@ const layerBase: Layer.Layer<...@@ -804,6 +854,12 @@ const layerBase: Layer.Layer<
804 }),854 }),
805 ),855 ),
806 ),856 ),
857 writeIfProviderThreadUnchanged: (input) =>
858 writeIfProviderThreadUnchangedEffect(input).pipe(
859 Effect.mapError(
860 (cause) => new EventSinkWriteError({ eventCount: input.events.length, cause }),
861 ),
862 ),
807 commitCommand: (input) =>863 commitCommand: (input) =>
808 commitCommandEffect(input).pipe(864 commitCommandEffect(input).pipe(
809 Effect.mapError(865 Effect.mapError(
apps/server/src/orchestration-v2/ProjectionStore.ts+2-1
...@@ -1056,7 +1056,8 @@ const encodeSubagentPayload = Schema.encodeEffect(...@@ -1056,7 +1056,8 @@ const encodeSubagentPayload = Schema.encodeEffect(
1056const encodeProviderSessionPayload = Schema.encodeEffect(1056const encodeProviderSessionPayload = Schema.encodeEffect(
1057 Schema.fromJsonString(OrchestrationV2ProviderSessionJsonSchema),1057 Schema.fromJsonString(OrchestrationV2ProviderSessionJsonSchema),
1058);1058);
1059const encodeProviderThreadPayload = Schema.encodeEffect(1059/** Encodes a provider thread exactly as its projection row stores `payload_json`. */
1060export const encodeProviderThreadPayload = Schema.encodeEffect(
1060 Schema.fromJsonString(OrchestrationV2ProviderThreadJsonSchema),1061 Schema.fromJsonString(OrchestrationV2ProviderThreadJsonSchema),
1061);1062);
1062const encodeProviderTurnPayload = Schema.encodeEffect(1063const encodeProviderTurnPayload = Schema.encodeEffect(
apps/server/src/orchestration-v2/ProviderSessionManager.test.ts+251-3
...@@ -147,6 +147,22 @@ const layerPausingAttachEventSink = (pause: {...@@ -147,6 +147,22 @@ const layerPausingAttachEventSink = (pause: {
147 }),147 }),
148 ).pipe(Layer.provide(layerTestEventSink));148 ).pipe(Layer.provide(layerTestEventSink));
149149
150const layerBeforeUnchangedWriteEventSink = (before: Ref.Ref<Effect.Effect<void>>) =>
151 Layer.effect(
152 EventSink.EventSinkV2,
153 Effect.gen(function* () {
154 const delegate = yield* EventSink.EventSinkV2;
155 return EventSink.EventSinkV2.of({
156 ...delegate,
157 writeIfProviderThreadUnchanged: (input) =>
158 Ref.getAndSet(before, Effect.void).pipe(
159 Effect.flatten,
160 Effect.andThen(delegate.writeIfProviderThreadUnchanged(input)),
161 ),
162 });
163 }),
164 ).pipe(Layer.provide(layerTestEventSink));
165
150const CodexCapabilities: OrchestrationV2ProviderCapabilities = CodexProviderCapabilitiesV2;166const CodexCapabilities: OrchestrationV2ProviderCapabilities = CodexProviderCapabilitiesV2;
151const ExclusiveCapabilities: OrchestrationV2ProviderCapabilities = {167const ExclusiveCapabilities: OrchestrationV2ProviderCapabilities = {
152 ...CodexCapabilities,168 ...CodexCapabilities,
...@@ -318,6 +334,8 @@ const makeProviderAdapter = Effect.fnUntraced(function* (...@@ -318,6 +334,8 @@ const makeProviderAdapter = Effect.fnUntraced(function* (
318 }) => Effect.Effect<void>;334 }) => Effect.Effect<void>;
319 readonly hasPendingBackgroundWork?: Effect.Effect<boolean>;335 readonly hasPendingBackgroundWork?: Effect.Effect<boolean>;
320 readonly hasPendingBackgroundWorkForThread?: Effect.Effect<boolean>;336 readonly hasPendingBackgroundWorkForThread?: Effect.Effect<boolean>;
337 /** False models a runtime whose background tasks can outlive its process. */
338 readonly backgroundTasksEndWithProcess?: boolean;
321 readonly hangSessionScopeClose?: boolean;339 readonly hangSessionScopeClose?: boolean;
322 readonly startTurn?: Effect.Effect<void>;340 readonly startTurn?: Effect.Effect<void>;
323 readonly beforeUnload?: Effect.Effect<void>;341 readonly beforeUnload?: Effect.Effect<void>;
...@@ -400,6 +418,9 @@ const makeProviderAdapter = Effect.fnUntraced(function* (...@@ -400,6 +418,9 @@ const makeProviderAdapter = Effect.fnUntraced(function* (
400 }),418 }),
401 )419 )
402 : Stream.fromQueue(events),420 : Stream.fromQueue(events),
421 ...(options.backgroundTasksEndWithProcess === false
422 ? {}
423 : { backgroundTasksEndWithProcess: true as const }),
403 ...(options.hasPendingBackgroundWork === undefined424 ...(options.hasPendingBackgroundWork === undefined
404 ? {}425 ? {}
405 : { hasPendingBackgroundWork: options.hasPendingBackgroundWork }),426 : { hasPendingBackgroundWork: options.hasPendingBackgroundWork }),
...@@ -458,6 +479,8 @@ function layerTest(input: {...@@ -458,6 +479,8 @@ function layerTest(input: {
458 readonly failReleaseEventWrites?: boolean;479 readonly failReleaseEventWrites?: boolean;
459 readonly flakyReleaseWrites?: FlakyReleaseWrites;480 readonly flakyReleaseWrites?: FlakyReleaseWrites;
460 readonly pauseAttachWrite?: Parameters<typeof layerPausingAttachEventSink>[0];481 readonly pauseAttachWrite?: Parameters<typeof layerPausingAttachEventSink>[0];
482 /** Runs once, right before the next provider-thread compare-and-set write. */
483 readonly beforeUnchangedWrite?: Ref.Ref<Effect.Effect<void>>;
461 /** Once armed, holds the next credential lookup until it is interrupted. */484 /** Once armed, holds the next credential lookup until it is interrupted. */
462 readonly pauseResolve?: {485 readonly pauseResolve?: {
463 readonly armed: Ref.Ref<boolean>;486 readonly armed: Ref.Ref<boolean>;
...@@ -465,6 +488,7 @@ function layerTest(input: {...@@ -465,6 +488,7 @@ function layerTest(input: {
465 };488 };
466 readonly hasPendingBackgroundWork?: Effect.Effect<boolean>;489 readonly hasPendingBackgroundWork?: Effect.Effect<boolean>;
467 readonly hasPendingBackgroundWorkForThread?: Effect.Effect<boolean>;490 readonly hasPendingBackgroundWorkForThread?: Effect.Effect<boolean>;
491 readonly backgroundTasksEndWithProcess?: boolean;
468 readonly hangSessionScopeClose?: boolean;492 readonly hangSessionScopeClose?: boolean;
469 readonly startTurn?: Effect.Effect<void>;493 readonly startTurn?: Effect.Effect<void>;
470 readonly beforeUnload?: Effect.Effect<void>;494 readonly beforeUnload?: Effect.Effect<void>;
...@@ -478,9 +502,11 @@ function layerTest(input: {...@@ -478,9 +502,11 @@ function layerTest(input: {
478 ? layerFlakyReleaseEventSink(input.flakyReleaseWrites)502 ? layerFlakyReleaseEventSink(input.flakyReleaseWrites)
479 : input.pauseAttachWrite !== undefined503 : input.pauseAttachWrite !== undefined
480 ? layerPausingAttachEventSink(input.pauseAttachWrite)504 ? layerPausingAttachEventSink(input.pauseAttachWrite)
481 : input.failReleaseEventWrites505 : input.beforeUnchangedWrite !== undefined
482 ? layerFailingReleaseEventSink506 ? layerBeforeUnchangedWriteEventSink(input.beforeUnchangedWrite)
483 : layerTestEventSink;507 : input.failReleaseEventWrites
508 ? layerFailingReleaseEventSink
509 : layerTestEventSink;
484 const layerRegistry = Layer.unwrap(510 const layerRegistry = Layer.unwrap(
485 makeProviderAdapter(input.state, {511 makeProviderAdapter(input.state, {
486 failEventStream: input.failEventStream ?? false,512 failEventStream: input.failEventStream ?? false,
...@@ -493,6 +519,9 @@ function layerTest(input: {...@@ -493,6 +519,9 @@ function layerTest(input: {
493 ...(input.hasPendingBackgroundWorkForThread === undefined519 ...(input.hasPendingBackgroundWorkForThread === undefined
494 ? {}520 ? {}
495 : { hasPendingBackgroundWorkForThread: input.hasPendingBackgroundWorkForThread }),521 : { hasPendingBackgroundWorkForThread: input.hasPendingBackgroundWorkForThread }),
522 ...(input.backgroundTasksEndWithProcess === undefined
523 ? {}
524 : { backgroundTasksEndWithProcess: input.backgroundTasksEndWithProcess }),
496 ...(input.hangSessionScopeClose === undefined525 ...(input.hangSessionScopeClose === undefined
497 ? {}526 ? {}
498 : { hangSessionScopeClose: input.hangSessionScopeClose }),527 : { hangSessionScopeClose: input.hangSessionScopeClose }),
...@@ -2769,6 +2798,225 @@ it.effect("ProviderSessionManagerV2 persists release when session scope close ha...@@ -2769,6 +2798,225 @@ it.effect("ProviderSessionManagerV2 persists release when session scope close ha
2769 }),2798 }),
2770);2799);
27712800
2801// Seeds a thread with a background command on a provider thread of the
2802// session it then opens, and one on a provider thread of another session.
2803const openSessionWithBackgroundCommand = (fixtureName: string) =>
2804 Effect.gen(function* () {
2805 const eventSink = yield* EventSink.EventSinkV2;
2806 const idAllocator = yield* IdAllocator.IdAllocatorV2;
2807 const manager = yield* ProviderSessionManager.ProviderSessionManagerV2;
2808 const now = yield* DateTime.now;
2809 const threadId = yield* idAllocator.allocate.thread({
2810 fixtureName,
2811 projectId: yield* idAllocator.allocate.project({ fixtureName }),
2812 });
2813 const providerSessionId = yield* idAllocator.allocate.providerSession({
2814 providerInstanceId: modelSelection.instanceId,
2815 threadId,
2816 });
2817 const otherProviderSessionId = yield* idAllocator.allocate.providerSession({
2818 providerInstanceId: modelSelection.instanceId,
2819 threadId,
2820 });
2821 const pendingBackgroundTasks = [
2822 { taskId: "dev-server", kind: "command", description: "python3 -m http.server" },
2823 ] as const;
2824 const providerThreads = [
2825 {
2826 ...makeProviderThread({ idAllocator, threadId, providerSessionId, now }),
2827 pendingBackgroundTasks,
2828 },
2829 {
2830 ...makeProviderThread({
2831 idAllocator,
2832 threadId,
2833 providerSessionId: otherProviderSessionId,
2834 now,
2835 nativeThreadId: "native-thread-other-session",
2836 }),
2837 pendingBackgroundTasks,
2838 },
2839 ];
2840 const writeProviderThread = (payload: OrchestrationV2ProviderThread) =>
2841 Effect.gen(function* () {
2842 yield* eventSink.write({
2843 events: [
2844 {
2845 id: yield* idAllocator.allocate.event({ threadId }),
2846 type: "provider-thread.updated",
2847 threadId,
2848 driver: CODEX_DRIVER,
2849 occurredAt: now,
2850 payload,
2851 },
2852 ],
2853 });
2854 });
2855 yield* eventSink.write({
2856 events: [yield* makeThreadCreatedEvent({ idAllocator, threadId, now })],
2857 });
2858 yield* Effect.forEach(providerThreads, writeProviderThread);
2859 yield* manager.open({ threadId, providerSessionId, modelSelection, runtimePolicy });
2860 // The task ids each seeded provider thread lists now.
2861 const rosters = Effect.gen(function* () {
2862 const projectionStore = yield* ProjectionStore.ProjectionStoreV2;
2863 const projection = yield* projectionStore.getThreadProjection(threadId);
2864 return providerThreads.map((seeded) =>
2865 (
2866 projection.providerThreads.find((thread) => thread.id === seeded.id)
2867 ?.pendingBackgroundTasks ?? []
2868 ).map((task) => task.taskId),
2869 );
2870 });
2871 return { threadId, providerSessionId, providerThreads, writeProviderThread, rosters };
2872 });
2873
2874it.effect(
2875 "ProviderSessionManagerV2 clears the background roster of a session a detach releases",
2876 () =>
2877 Effect.gen(function* () {
2878 const state = yield* Ref.make(emptyState);
2879 const effect = Effect.gen(function* () {
2880 const manager = yield* ProviderSessionManager.ProviderSessionManagerV2;
2881 const { threadId, providerSessionId, rosters } = yield* openSessionWithBackgroundCommand(
2882 "provider-session-manager-detach-roster",
2883 );
2884 assert.deepStrictEqual(yield* rosters, [["dev-server"], ["dev-server"]]);
2885
2886 // Settle and archive detach the thread, which releases an exclusive
2887 // (Claude-like) session.
2888 yield* manager.detach({ providerSessionId, threadId, detail: "Thread settled." });
2889
2890 assert.isTrue(Option.isNone(yield* manager.get(providerSessionId)));
2891 assert.deepStrictEqual(yield* rosters, [[], ["dev-server"]]);
2892 });
2893
2894 yield* effect.pipe(
2895 Effect.provide(
2896 layerTest({ state, idleTimeoutMs: 60_000, capabilities: ExclusiveCapabilities }),
2897 ),
2898 );
2899 }),
2900);
2901
2902it.effect("ProviderSessionManagerV2 clears the background roster of an idle-released session", () =>
2903 Effect.gen(function* () {
2904 const state = yield* Ref.make(emptyState);
2905 const effect = Effect.gen(function* () {
2906 const manager = yield* ProviderSessionManager.ProviderSessionManagerV2;
2907 const { providerSessionId, rosters } = yield* openSessionWithBackgroundCommand(
2908 "provider-session-manager-idle-roster",
2909 );
2910 assert.deepStrictEqual(yield* rosters, [["dev-server"], ["dev-server"]]);
2911
2912 yield* TestClock.adjust("1 second");
2913 yield* Effect.yieldNow;
2914
2915 assert.isTrue(Option.isNone(yield* manager.get(providerSessionId)));
2916 assert.deepStrictEqual(yield* rosters, [[], ["dev-server"]]);
2917 });
2918
2919 yield* effect.pipe(Effect.provide(layerTest({ state, idleTimeoutMs: 1000 })));
2920 }),
2921);
2922
2923it.effect("ProviderSessionManagerV2 keeps the roster of a runtime whose tasks can outlive it", () =>
2924 Effect.gen(function* () {
2925 const state = yield* Ref.make(emptyState);
2926 const effect = Effect.gen(function* () {
2927 const manager = yield* ProviderSessionManager.ProviderSessionManagerV2;
2928 const { providerSessionId, rosters } = yield* openSessionWithBackgroundCommand(
2929 "provider-session-manager-outliving-roster",
2930 );
2931
2932 // Like OpenCode, whose background subagents run on a shared server.
2933 yield* manager.close(providerSessionId);
2934
2935 assert.deepStrictEqual(yield* rosters, [["dev-server"], ["dev-server"]]);
2936 });
2937
2938 yield* effect.pipe(
2939 Effect.provide(
2940 layerTest({ state, idleTimeoutMs: 60_000, backgroundTasksEndWithProcess: false }),
2941 ),
2942 );
2943 }),
2944);
2945
2946it.effect(
2947 "ProviderSessionManagerV2 keeps background tasks a replacement reports during release",
2948 () =>
2949 Effect.gen(function* () {
2950 const state = yield* Ref.make(emptyState);
2951 const scopeCloseReached = yield* Deferred.make<void>();
2952 const effect = Effect.gen(function* () {
2953 const manager = yield* ProviderSessionManager.ProviderSessionManagerV2;
2954 const { providerSessionId, providerThreads, writeProviderThread, rosters } =
2955 yield* openSessionWithBackgroundCommand("provider-session-manager-replacement-roster");
2956 const release = yield* manager.close(providerSessionId).pipe(Effect.forkChild);
2957
2958 // While the old process closes, a replacement session with the same id
2959 // reports a task of its own next to the dead one.
2960 yield* Deferred.await(scopeCloseReached);
2961 yield* writeProviderThread({
2962 ...providerThreads[0]!,
2963 pendingBackgroundTasks: [
2964 ...providerThreads[0]!.pendingBackgroundTasks,
2965 { taskId: "replacement-task", kind: "command" },
2966 ],
2967 });
2968 yield* TestClock.adjust("30 seconds");
2969 yield* Fiber.join(release);
2970
2971 assert.deepStrictEqual(yield* rosters, [["replacement-task"], ["dev-server"]]);
2972 });
2973
2974 yield* effect.pipe(
2975 Effect.provide(
2976 layerTest({
2977 state,
2978 idleTimeoutMs: 60_000,
2979 hangSessionScopeClose: true,
2980 scopeCloseReached,
2981 }),
2982 ),
2983 );
2984 }),
2985);
2986
2987it.effect("ProviderSessionManagerV2 keeps a provider update that lands while release clears", () =>
2988 Effect.gen(function* () {
2989 const state = yield* Ref.make(emptyState);
2990 const beforeUnchangedWrite = yield* Ref.make<Effect.Effect<void>>(Effect.void);
2991 const effect = Effect.gen(function* () {
2992 const manager = yield* ProviderSessionManager.ProviderSessionManagerV2;
2993 const { providerSessionId, providerThreads, writeProviderThread, rosters } =
2994 yield* openSessionWithBackgroundCommand("provider-session-manager-raced-roster");
2995 // A replacement reports a task after cleanup read the row, before it
2996 // writes, without moving updatedAt (not every writer does).
2997 const seeded = providerThreads[0]!;
2998 yield* Ref.set(
2999 beforeUnchangedWrite,
3000 writeProviderThread({
3001 ...seeded,
3002 pendingBackgroundTasks: [
3003 ...seeded.pendingBackgroundTasks,
3004 { taskId: "replacement-task", kind: "command" },
3005 ],
3006 }).pipe(Effect.orDie),
3007 );
3008
3009 yield* manager.close(providerSessionId);
3010
3011 assert.deepStrictEqual(yield* rosters, [["replacement-task"], ["dev-server"]]);
3012 });
3013
3014 yield* effect.pipe(
3015 Effect.provide(layerTest({ state, idleTimeoutMs: 60_000, beforeUnchangedWrite })),
3016 );
3017 }),
3018);
3019
2772it.effect("ProviderSessionManagerV2 defers idle release while background work is pending", () =>3020it.effect("ProviderSessionManagerV2 defers idle release while background work is pending", () =>
2773 Effect.gen(function* () {3021 Effect.gen(function* () {
2774 const state = yield* Ref.make(emptyState);3022 const state = yield* Ref.make(emptyState);
apps/server/src/orchestration-v2/ProviderSessionManager.ts+127-3
...@@ -774,14 +774,126 @@ export const layerWithOptions = (...@@ -774,14 +774,126 @@ export const layerWithOptions = (
774 }774 }
775 });775 });
776776
777 // Reads the background tasks a released session's process left on its
778 // provider threads. Read as soon as the session leaves the live map:
779 // after that, a replacement session with the same id can open and report
780 // tasks of its own on the same provider threads. Server shutdown keeps
781 // rosters, since startup recovery clears them and records the cancelled
782 // work. Only runtimes whose tasks end with their process opt in (Claude,
783 // Muse). OpenCode's background subagents run on a server shared by its
784 // sessions and can outlive this one, so their roster stays.
785 const readReleasedBackgroundTasks = (input: {
786 readonly entry: LiveSessionEntry;
787 readonly reason: ProviderSessionReleaseReason;
788 readonly detachedThreadId?: ThreadId;
789 }) =>
790 Effect.gen(function* () {
791 if (
792 input.reason === "server_shutdown" ||
793 input.entry.runtime.backgroundTasksEndWithProcess !== true
794 ) {
795 return [];
796 }
797 const providerSessionId = input.entry.runtime.providerSessionId;
798 // A detach that released the session already removed its thread
799 // from the attached set.
800 const threadIds = new Set(input.entry.attachedThreadIds);
801 if (input.detachedThreadId !== undefined) threadIds.add(input.detachedThreadId);
802 const released = [];
803 for (const threadId of threadIds) {
804 const projection = yield* projectionStore.getThreadRecords(threadId, [
805 "providerThreads",
806 ]);
807 for (const providerThread of projection.providerThreads) {
808 const tasks = providerThread.pendingBackgroundTasks ?? [];
809 if (providerThread.providerSessionId !== providerSessionId || tasks.length === 0) {
810 continue;
811 }
812 released.push({
813 threadId,
814 providerSessionId,
815 providerThreadId: providerThread.id,
816 taskIds: new Set(tasks.map((task) => task.taskId)),
817 });
818 }
819 }
820 return released;
821 }).pipe(
822 Effect.catchCause((cause) =>
823 Effect.logWarning("orchestration-v2.provider-session-release-roster-read-failed", {
824 providerSessionId: input.entry.runtime.providerSessionId,
825 cause,
826 }).pipe(Effect.as([])),
827 ),
828 );
829 type ReleasedBackgroundTasks = Effect.Success<
830 ReturnType<typeof readReleasedBackgroundTasks>
831 >[number];
832
833 // Removes the tasks `readReleasedBackgroundTasks` found from their
834 // rosters. Their process is gone, so nothing will report them ending,
835 // and a leftover roster keeps the thread Waiting (#15239). Tasks a
836 // replacement session reported since are left in place. The write
837 // commits only if the stored row still equals what was read, and is
838 // retried a few times when a provider update lands in between.
839 const writeReleasedBackgroundWorkEvents = (
840 released: ReadonlyArray<ReleasedBackgroundTasks>,
841 ) =>
842 Effect.forEach(
843 released,
844 ({ threadId, providerSessionId, providerThreadId, taskIds }) =>
845 Effect.gen(function* () {
846 for (let attempt = 0; attempt < 3; attempt++) {
847 const projection = yield* projectionStore.getThreadRecords(threadId, [
848 "providerThreads",
849 ]);
850 const providerThread = projection.providerThreads.find(
851 (candidate) => candidate.id === providerThreadId,
852 );
853 const tasks = providerThread?.pendingBackgroundTasks ?? [];
854 const remaining = tasks.filter((task) => !taskIds.has(task.taskId));
855 if (providerThread === undefined || remaining.length === tasks.length) return;
856 const now = yield* DateTime.now;
857 const { committed } = yield* eventSink.writeIfProviderThreadUnchanged({
858 expected: providerThread,
859 events: [
860 {
861 id: yield* idAllocator.allocate.event({ threadId, providerSessionId }),
862 type: "provider-thread.updated",
863 threadId,
864 driver: providerThread.driver,
865 providerInstanceId: providerThread.providerInstanceId,
866 occurredAt: now,
867 payload: {
868 ...providerThread,
869 pendingBackgroundTasks: remaining,
870 updatedAt: now,
871 },
872 },
873 ],
874 });
875 if (committed) return;
876 }
877 yield* Effect.logWarning(
878 "orchestration-v2.provider-session-release-roster-conflict",
879 {
880 providerSessionId,
881 providerThreadId,
882 },
883 );
884 }),
885 { discard: true },
886 );
887
777 // Records a released session as stopped and resolves the live runtime888 // Records a released session as stopped and resolves the live runtime
778 // requests it left. Each write runs even if the other fails. Once a889 // requests and background-task rosters it left. Each write runs even if
779 // replacement session opens with the same id, it owns the session status,890 // another fails. Once a replacement session opens with the same id, it
780 // so only the requests are settled.891 // owns the session status, so only the requests are settled.
781 const writeReleaseRecords = (input: {892 const writeReleaseRecords = (input: {
782 readonly entry: LiveSessionEntry;893 readonly entry: LiveSessionEntry;
783 readonly reason: ProviderSessionReleaseReason;894 readonly reason: ProviderSessionReleaseReason;
784 readonly detail?: string;895 readonly detail?: string;
896 readonly releasedBackgroundTasks: ReadonlyArray<ReleasedBackgroundTasks>;
785 readonly releasedAt: DateTime.Utc;897 readonly releasedAt: DateTime.Utc;
786 readonly replaced: boolean;898 readonly replaced: boolean;
787 }) =>899 }) =>
...@@ -795,6 +907,7 @@ export const layerWithOptions = (...@@ -795,6 +907,7 @@ export const layerWithOptions = (
795 input.entry.requestEventPermit.withPermits(1),907 input.entry.requestEventPermit.withPermits(1),
796 ),908 ),
797 ),909 ),
910 Effect.exit(writeReleasedBackgroundWorkEvents(input.releasedBackgroundTasks)),
798 ],911 ],
799 { concurrency: 1 },912 { concurrency: 1 },
800 ).pipe(Effect.flatMap(Exit.asVoidAll));913 ).pipe(Effect.flatMap(Exit.asVoidAll));
...@@ -937,6 +1050,8 @@ export const layerWithOptions = (...@@ -937,6 +1050,8 @@ export const layerWithOptions = (
937 readonly providerSessionId: ProviderSessionId;1050 readonly providerSessionId: ProviderSessionId;
938 readonly reason: ProviderSessionReleaseReason;1051 readonly reason: ProviderSessionReleaseReason;
939 readonly detail?: string;1052 readonly detail?: string;
1053 /** The thread whose detach released the session; it is no longer attached. */
1054 readonly detachedThreadId?: ThreadId;
940 readonly cancelIdleFiber?: boolean;1055 readonly cancelIdleFiber?: boolean;
941 readonly onlyIfIdleGeneration?: number;1056 readonly onlyIfIdleGeneration?: number;
942 readonly gracefulSubscribers?: boolean;1057 readonly gracefulSubscribers?: boolean;
...@@ -948,6 +1063,13 @@ export const layerWithOptions = (...@@ -948,6 +1063,13 @@ export const layerWithOptions = (
948 onNone: () => Effect.void,1063 onNone: () => Effect.void,
949 onSome: (entry) =>1064 onSome: (entry) =>
950 Effect.gen(function* () {1065 Effect.gen(function* () {
1066 const releasedBackgroundTasks = yield* readReleasedBackgroundTasks({
1067 entry,
1068 reason: input.reason,
1069 ...(input.detachedThreadId === undefined
1070 ? {}
1071 : { detachedThreadId: input.detachedThreadId }),
1072 });
951 if (input.cancelIdleFiber !== false) {1073 if (input.cancelIdleFiber !== false) {
952 yield* cancelIdleFiber(entry.idleFiber);1074 yield* cancelIdleFiber(entry.idleFiber);
953 }1075 }
...@@ -969,6 +1091,7 @@ export const layerWithOptions = (...@@ -969,6 +1091,7 @@ export const layerWithOptions = (
969 entry,1091 entry,
970 reason: input.reason,1092 reason: input.reason,
971 ...(input.detail === undefined ? {} : { detail: input.detail }),1093 ...(input.detail === undefined ? {} : { detail: input.detail }),
1094 releasedBackgroundTasks,
972 releasedAt,1095 releasedAt,
973 };1096 };
974 const recorded = yield* Effect.exit(1097 const recorded = yield* Effect.exit(
...@@ -2406,6 +2529,7 @@ export const layerWithOptions = (...@@ -2406,6 +2529,7 @@ export const layerWithOptions = (
2406 providerSessionId: input.providerSessionId,2529 providerSessionId: input.providerSessionId,
2407 reason: "manual_shutdown",2530 reason: "manual_shutdown",
2408 ...(input.detail === undefined ? {} : { detail: input.detail }),2531 ...(input.detail === undefined ? {} : { detail: input.detail }),
2532 detachedThreadId: input.threadId,
2409 });2533 });
2410 return;2534 return;
2411 }2535 }
packages/provider-core/src/server/ProviderAdapter.ts+8
...@@ -565,6 +565,14 @@ export interface ProviderAdapterV2SessionRuntime {...@@ -565,6 +565,14 @@ export interface ProviderAdapterV2SessionRuntime {
565 readonly hasPendingBackgroundWorkForThread?: (565 readonly hasPendingBackgroundWorkForThread?: (
566 providerThread: OrchestrationV2ProviderThread,566 providerThread: OrchestrationV2ProviderThread,
567 ) => Effect.Effect<boolean>;567 ) => Effect.Effect<boolean>;
568 /**
569 * Set when every background task this session lists on its provider threads
570 * runs inside the session's own process and ends when it closes. The session
571 * manager then clears those tasks when it releases the session (#15239).
572 * Leave unset when a task can outlive the session, for example on a shared
573 * server, so its thread keeps showing Waiting.
574 */
575 readonly backgroundTasksEndWithProcess?: true;
568 /**576 /**
569 * Capacity for the requested model/options, independent of native thread usage.577 * Capacity for the requested model/options, independent of native thread usage.
570 * `cwd` is the thread's working directory, for providers whose project config578 * `cwd` is the thread's working directory, for providers whose project config
packages/provider-muse/src/server/adapter.ts+2
...@@ -2107,6 +2107,8 @@ export const makeMuseAdapterV2 = Effect.fn("makeMuseAdapterV2")(function* (...@@ -2107,6 +2107,8 @@ export const makeMuseAdapterV2 = Effect.fn("makeMuseAdapterV2")(function* (
2107 providerSessionId: input.providerSessionId,2107 providerSessionId: input.providerSessionId,
2108 providerSession: session,2108 providerSession: session,
2109 events: Stream.fromQueue(events),2109 events: Stream.fromQueue(events),
2110 // Workflows and subagents run in this session's own `muse` host.
2111 backgroundTasksEndWithProcess: true,
2110 // A running workflow or a held Muse turn keeps the host: idle release must wait.2112 // A running workflow or a held Muse turn keeps the host: idle release must wait.
2111 hasPendingBackgroundWork: Effect.sync(2113 hasPendingBackgroundWork: Effect.sync(
2112 () => observedChildren.size > 0 || wake !== undefined,2114 () => observedChildren.size > 0 || wake !== undefined,