diff --git a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts index bd7b6f4208763..104112d0d1b4d 100644 --- a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts @@ -915,6 +915,7 @@ describe("CodexAdapterV2 post-settle continuation", () => { Effect.void, onRequest: (method: string, params: unknown) => Effect.Effect = () => Effect.void, readChildMetadata?: Parameters[2], + resolveRuntime?: CodexAdapterV2.CodexAdapterV2Options["resolveRuntime"], ) => Effect.gen(function* () { const fileSystem = yield* FileSystem.FileSystem; @@ -958,6 +959,7 @@ describe("CodexAdapterV2 post-settle continuation", () => { instanceId: CodexAdapterV2.CODEX_DEFAULT_INSTANCE_ID, settings: DEFAULT_CODEX_SETTINGS, environment: {}, + ...(resolveRuntime === undefined ? {} : { resolveRuntime }), clientFactory, crypto: yield* Crypto.Crypto, fileSystem, @@ -1021,6 +1023,7 @@ describe("CodexAdapterV2 post-settle continuation", () => { > => event.type === "subagent.updated", ); return { + adapter, runtime, providerThread, threadId, @@ -1146,6 +1149,146 @@ describe("CodexAdapterV2 post-settle continuation", () => { ), ); + it.effect.each( + [ + { + code: "unauthorized", + message: + "Your access token could not be refreshed because you have since logged out or signed in to another account. Please sign in again.", + stale: true, + }, + { + code: "unauthorized", + message: + "Error running remote compact task: Your access token could not be refreshed because you have since logged out or signed in to another account. Please sign in again.", + stale: true, + }, + { + code: "unauthorized", + message: + "Your access token could not be refreshed because your refresh token has expired. Please log out and sign in again.", + stale: false, + }, + ].flatMap((failure) => + [false, true].flatMap((managed) => + [false, true].map((notifyError) => ({ ...failure, managed, notifyError })), + ), + ), + )( + "handles auth failure with managed=$managed, notifyError=$notifyError: $message", + ({ code, message, stale, managed, notifyError }) => + Effect.scoped( + Effect.gen(function* () { + const nativeThreadId = "auth-change-thread"; + const nativeTurnId = "auth-change-turn"; + const error = { message, codexErrorInfo: code, additionalDetails: null }; + const harness = yield* makeCodexReplayHarness( + makeCodexReplayTranscript({ + scenario: "external-auth-change", + entries: [ + ...codexReplayPreamble({ nativeThreadId, nativeTurnId, prompt: "hello" }), + ...(notifyError + ? [ + { + type: "emit_inbound" as const, + label: "error", + frame: { + method: "error", + params: { + threadId: nativeThreadId, + turnId: nativeTurnId, + error, + willRetry: false, + }, + }, + }, + ] + : []), + { + type: "emit_inbound", + label: "failed", + frame: { + method: "turn/completed", + params: { + threadId: nativeThreadId, + turn: { + ...makeCodexReplayTurn({ id: nativeTurnId, status: "failed" }), + error, + }, + }, + }, + }, + ], + }), + undefined, + undefined, + undefined, + managed + ? Effect.succeed({ + config: DEFAULT_CODEX_SETTINGS, + environment: {}, + revision: "managed-test", + }) + : undefined, + ); + yield* harness.runtime.startTurn( + makeCodexTestTurnInput({ + threadId: harness.threadId, + providerThread: harness.providerThread, + now: yield* DateTime.now, + attemptId: RunAttemptId.make("auth-change-attempt"), + text: "hello", + }), + ); + yield* harness.firstTerminal; + assert.equal( + yield* harness.runtime.hasStaleCredentials ?? Effect.succeed(false), + stale && !managed, + ); + assert.equal(harness.terminalEvents()[0]?.status, "failed"); + assert.include(harness.terminalEvents()[0]?.failure?.message, message); + if (managed) { + assert.isUndefined(harness.runtime.hasStaleCredentials); + assert.equal(harness.terminalEvents()[0]?.failure?.message, message); + } + if (stale && !managed) { + const replacement = yield* harness.adapter.openSession({ + threadId: harness.threadId, + providerSessionId: ProviderSessionId.make("auth-change-replacement"), + modelSelection: CODEX_TEST_MODEL_SELECTION, + runtimePolicy: CODEX_TEST_RUNTIME_POLICY, + }); + const replacementThread = yield* replacement.ensureThread({ + threadId: harness.threadId, + modelSelection: CODEX_TEST_MODEL_SELECTION, + runtimePolicy: CODEX_TEST_RUNTIME_POLICY, + }); + const failedAgain = yield* replacement.events.pipe( + Stream.filter((event) => event.type === "turn.terminal"), + Stream.runHead, + Effect.forkChild({ startImmediately: true }), + ); + yield* replacement.startTurn( + makeCodexTestTurnInput({ + threadId: harness.threadId, + providerThread: replacementThread, + now: yield* DateTime.now, + attemptId: RunAttemptId.make("auth-change-replacement-attempt"), + text: "hello", + }), + ); + yield* Fiber.join(failedAgain); + assert.isFalse(yield* replacement.hasStaleCredentials ?? Effect.succeed(false)); + assert.include(harness.terminalEvents()[0]?.failure?.message, "run codex login"); + } + }), + ).pipe( + Effect.provide( + Layer.mergeAll(IdAllocator.layer, McpProviderSessions.layer, NodeServices.layer), + ), + ), + ); + it.effect("adds the session MCP server without replacing launch-argument servers", () => Effect.gen(function* () { const scenario = "launch-mcp-servers"; diff --git a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts index 21c9943b41191..399fdcaeaff8c 100644 --- a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts @@ -335,6 +335,26 @@ function toProtocolError( }); } +// Codex's guarded refresh refuses a different on-disk account; account/read cannot reload it. +function isCodexAccountMismatch(message: string, code: string | null | undefined): boolean { + return ( + code === "unauthorized" && + message.includes( + "Your access token could not be refreshed because you have since logged out or signed in to another account.", + ) + ); +} + +function codexAuthRecoveryMessage( + message: string | undefined, + code: string | null | undefined, + allowAuthRecovery: boolean, +) { + return allowAuthRecovery && message !== undefined && isCodexAccountMismatch(message, code) + ? `${message} Let other Codex turns and background work finish, then send your message again. If it still fails, run codex login in this provider's home and restart T3 when Codex is idle.` + : message; +} + function normalizeCodexCause(error: unknown): unknown { return error; } @@ -1703,6 +1723,8 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( threadId === null ? Effect.succeed(undefined) : mcpSessions.read(threadId); const { clientFactory, crypto, fileSystem, idAllocator, serverConfig } = adapterOptions; const continuationRequests = adapterOptions.continuationRequests; + // A replacement must complete a turn before another auth restart; broken saved auth must not loop. + const authRecoveryAvailable = yield* Ref.make(true); return ProviderAdapter.ProviderAdapterV2.of({ instanceId: adapterOptions.instanceId, @@ -1725,6 +1747,19 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( }), ), ); + const staleCredentials = yield* Ref.make(false); + const recordAuthFailure = Effect.fn("CodexAdapterV2.recordAuthFailure")(function* ( + message: string, + code: string | null, + ) { + if ( + resolvedRuntime === undefined && + isCodexAccountMismatch(message, code) && + (yield* Ref.getAndSet(authRecoveryAvailable, false)) + ) { + yield* Ref.set(staleCredentials, true); + } + }); const client = yield* clientFactory.open({ instanceId: adapterOptions.instanceId, threadId: input.threadId, @@ -4516,6 +4551,12 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( yield* client.handleServerNotification("error", (payload) => Effect.gen(function* () { + if (!payload.willRetry) { + yield* recordAuthFailure( + payload.error.message, + codexErrorInfoCode(payload.error.codexErrorInfo), + ); + } const context = yield* awaitActiveTurn(payload.turnId); if (context === undefined) { return; @@ -4527,7 +4568,11 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( nativeCode: notificationCode, failure: withCodexManagedFailure( makeProviderFailure({ - message: payload.error.additionalDetails?.trim() || payload.error.message, + message: codexAuthRecoveryMessage( + payload.error.additionalDetails?.trim() || payload.error.message, + notificationCode, + resolvedRuntime === undefined, + ), code: notificationCode, class: codexUsageLimitError(notificationCode, payload.error.codexErrorInfo) ? "usage_limit" @@ -5629,7 +5674,11 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( ? previousFailure.failure : withCodexManagedFailure( makeProviderFailure({ - message: input.failureMessage, + message: codexAuthRecoveryMessage( + input.failureMessage, + failureCode, + resolvedRuntime === undefined, + ), code: failureCode, class: failureClass, }), @@ -6122,6 +6171,15 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( yield* client.handleServerNotification("turn/completed", (payload) => Effect.gen(function* () { + if (payload.turn.error != null) { + yield* recordAuthFailure( + payload.turn.error.message, + codexErrorInfoCode(payload.turn.error.codexErrorInfo), + ); + } + if (payload.turn.status === "completed" && !(yield* Ref.get(staleCredentials))) { + yield* Ref.set(authRecoveryAvailable, true); + } const context = (yield* Ref.get(activeTurns)).get(payload.turn.id); if (context === undefined) { return; @@ -6552,6 +6610,9 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( providerSessionId: input.providerSessionId, providerSession: session, events: Stream.fromEffectRepeat(Queue.take(events)), + ...(resolvedRuntime === undefined + ? { hasStaleCredentials: Ref.get(staleCredentials) } + : {}), canReuseContextUsage: canReuseCodexContextUsage, // Idle release can beat delayed subagent resumes: Codex reports them completed between turns with no resume-expected signal. hasPendingBackgroundWork: Effect.gen(function* () { diff --git a/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts b/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts index 010f11742933e..8c7ff04dda5f4 100644 --- a/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts +++ b/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts @@ -328,11 +328,14 @@ const makeProviderAdapter = Effect.fnUntraced(function* ( readonly providerSessionId: ProviderSessionId; readonly initialProviderItemIdentityVersion?: 2; }) => Effect.Effect; + readonly hasStaleCredentials?: Effect.Effect; + readonly beforeResume?: Effect.Effect; readonly hasPendingBackgroundWork?: Effect.Effect; readonly hasPendingBackgroundWorkForThread?: Effect.Effect; readonly backgroundTasksEndWithProcess?: boolean; readonly hangSessionScopeClose?: boolean; readonly startTurn?: Effect.Effect; + readonly compactThread?: Effect.Effect; readonly beforeUnload?: Effect.Effect; readonly spawnBeforeOpen?: boolean; readonly scopeCloseReached?: Deferred.Deferred; @@ -420,13 +423,22 @@ const makeProviderAdapter = Effect.fnUntraced(function* ( : { hasPendingBackgroundWorkForThread: () => options.hasPendingBackgroundWorkForThread!, }), + ...(options.hasStaleCredentials === undefined + ? {} + : { hasStaleCredentials: options.hasStaleCredentials }), ensureThread: () => unimplemented("ensureThread unused in test"), resumeThread: (threadInput) => Ref.update(state, (current) => ({ ...current, resumeCount: current.resumeCount + 1, - })).pipe(Effect.as(threadInput.providerThread)), + })).pipe( + Effect.andThen(options.beforeResume ?? Effect.void), + Effect.as(threadInput.providerThread), + ), startTurn: () => options.startTurn ?? Effect.void, + ...(options.compactThread === undefined + ? {} + : { compactThread: () => options.compactThread! }), steerTurn: () => Effect.void, interruptTurn: () => Ref.update(state, (current) => ({ @@ -475,11 +487,14 @@ function layerTest(input: { readonly armed: Ref.Ref; readonly paused: Deferred.Deferred; }; + readonly hasStaleCredentials?: Effect.Effect; + readonly beforeResume?: Effect.Effect; readonly hasPendingBackgroundWork?: Effect.Effect; readonly hasPendingBackgroundWorkForThread?: Effect.Effect; readonly backgroundTasksEndWithProcess?: boolean; readonly hangSessionScopeClose?: boolean; readonly startTurn?: Effect.Effect; + readonly compactThread?: Effect.Effect; readonly beforeUnload?: Effect.Effect; readonly spawnBeforeOpen?: boolean; readonly scopeCloseReached?: Deferred.Deferred; @@ -499,6 +514,10 @@ function layerTest(input: { const layerRegistry = Layer.unwrap( makeProviderAdapter(input.state, { failEventStream: input.failEventStream ?? false, + ...(input.hasStaleCredentials === undefined + ? {} + : { hasStaleCredentials: input.hasStaleCredentials }), + ...(input.beforeResume === undefined ? {} : { beforeResume: input.beforeResume }), ...(input.capabilities === undefined ? {} : { capabilities: input.capabilities }), ...(input.mcpConfigs === undefined ? {} : { mcpConfigs: input.mcpConfigs }), ...(input.beforeOpen === undefined ? {} : { beforeOpen: input.beforeOpen }), @@ -515,6 +534,7 @@ function layerTest(input: { ? {} : { hangSessionScopeClose: input.hangSessionScopeClose }), ...(input.startTurn === undefined ? {} : { startTurn: input.startTurn }), + ...(input.compactThread === undefined ? {} : { compactThread: input.compactThread }), ...(input.beforeUnload === undefined ? {} : { beforeUnload: input.beforeUnload }), ...(input.spawnBeforeOpen === undefined ? {} : { spawnBeforeOpen: input.spawnBeforeOpen }), ...(input.scopeCloseReached === undefined @@ -3542,3 +3562,183 @@ it.effect( assert.isFalse(denied?.capabilities?.has("device")); }), ); + +it.effect.each(["startTurn", "compactThread"] as const)( + "ProviderSessionManagerV2 replaces stale credentials only after shared work settles and resumes native history with %s", + (operation) => + Effect.gen(function* () { + const state = yield* Ref.make(emptyState); + const stale = yield* Ref.make(false); + const background = yield* Ref.make(false); + const resumeEntered = yield* Deferred.make(); + const resumeContinue = yield* Deferred.make(); + const pauseResume = yield* Ref.make(false); + const pauseAttachWrite = { + armed: yield* Ref.make(false), + paused: yield* Deferred.make(), + }; + const pauseProbe = yield* Ref.make(false); + const probeEntered = yield* Deferred.make(); + const probeContinue = yield* Deferred.make(); + yield* Effect.gen(function* () { + const manager = yield* ProviderSessionManager.ProviderSessionManagerV2; + const sink = yield* EventSink.EventSinkV2; + const projections = yield* ProjectionStore.ProjectionStoreV2; + const ids = yield* IdAllocator.IdAllocatorV2; + const now = yield* DateTime.now; + const a = ThreadId.make("auth-recovery-a"); + const b = ThreadId.make("auth-recovery-b"); + const providerSessionId = ids.derive.providerSession({ + providerInstanceId: modelSelection.instanceId, + }); + yield* sink.write({ + events: [ + yield* makeThreadCreatedEvent({ idAllocator: ids, threadId: a, now }), + yield* makeThreadCreatedEvent({ idAllocator: ids, threadId: b, now }), + ], + }); + const open = (threadId: ThreadId) => + manager.open({ threadId, providerSessionId, modelSelection, runtimePolicy }); + const first = yield* open(a); + const providerThread = makeProviderThread({ + idAllocator: ids, + threadId: a, + providerSessionId, + now, + nativeThreadId: "persisted-native-history", + }); + yield* sink.write({ + events: [ + { + id: yield* ids.allocate.event({ threadId: a }), + type: "provider-thread.updated", + threadId: a, + driver: CODEX_DRIVER, + occurredAt: now, + payload: providerThread, + }, + ], + }); + const runId = ids.derive.run({ threadId: a, ordinal: 1 }); + const startInput = { + appThread: (yield* projections.getThreadProjection(a)).thread, + threadId: a, + runId, + runOrdinal: 1, + providerTurnOrdinal: 1, + attemptId: ids.derive.runAttempt({ runId, attemptOrdinal: 1 }), + rootNodeId: ids.derive.rootNode({ runId }), + providerThread, + message: { + createdBy: "user", + creationSource: "web", + messageId: yield* ids.allocate.message({ threadId: a, ordinal: 1 }), + text: "hello", + attachments: [], + }, + modelSelection, + runtimePolicy, + } as const; + yield* first[operation]!(startInput); + yield* Ref.set(stale, true); + assert.strictEqual(yield* open(b), first); + assert.equal((yield* Ref.get(state)).closeCount, 0); + const subscription = yield* first.subscribeEvents!; + const terminal: ProviderAdapter.ProviderAdapterV2Event = { + type: "turn.terminal", + driver: CODEX_DRIVER, + providerThreadId: providerThread.id, + providerTurnId: ids.derive.providerTurn({ + driver: CODEX_DRIVER, + nativeTurnId: "finished", + }), + runOrdinal: 1, + status: "completed", + failure: null, + threadDisposition: "reusable", + }; + yield* Queue.offer( + (yield* Ref.get(state)).eventQueues.get(String(providerSessionId))!, + terminal, + ); + assert.deepEqual(yield* subscription.events.pipe(Stream.runHead), Option.some(terminal)); + yield* Ref.set(background, true); + assert.strictEqual(yield* open(b), first); + yield* Ref.set(background, false); + yield* Ref.set(pauseResume, true); + const pending = yield* first + .resumeThread({ threadId: a, providerThread, modelSelection, runtimePolicy }) + .pipe(Effect.forkChild); + yield* Deferred.await(resumeEntered); + assert.strictEqual(yield* open(b), first); + yield* Deferred.succeed(resumeContinue, undefined); + yield* Fiber.join(pending); + yield* manager.detach({ providerSessionId, threadId: a }); + yield* Ref.set(pauseAttachWrite.armed, true); + const pendingStart = yield* first[operation]!(startInput).pipe(Effect.forkChild); + yield* Deferred.await(pauseAttachWrite.paused); + assert.strictEqual(yield* open(b), first); + yield* Fiber.interrupt(pendingStart); + yield* Ref.set(pauseProbe, true); + const reopening = yield* open(b).pipe(Effect.forkChild); + yield* Deferred.await(probeEntered); + yield* first[operation]!({ ...startInput, runOrdinal: 2 }); + yield* Deferred.succeed(probeContinue, undefined); + assert.strictEqual(yield* Fiber.join(reopening), first); + const finished = yield* first.subscribeEvents!; + const secondTerminal = { ...terminal, runOrdinal: 2 }; + yield* Queue.offer( + (yield* Ref.get(state)).eventQueues.get(String(providerSessionId))!, + secondTerminal, + ); + assert.deepEqual(yield* finished.events.pipe(Stream.runHead), Option.some(secondTerminal)); + const replacement = yield* open(b); + assert.notStrictEqual(replacement, first); + assert.equal((yield* Ref.get(state)).closeCount, 1); + const saved = (yield* projections.getThreadProjection(a)).providerThreads.find( + (thread) => thread.id === providerThread.id, + )!; + assert.equal(saved.nativeThreadRef?.nativeId, "persisted-native-history"); + yield* replacement.resumeThread({ + threadId: a, + providerThread: saved, + modelSelection, + runtimePolicy, + }); + assert.equal((yield* Ref.get(state)).resumeCount, 2); + assert.equal((yield* first[operation]!(startInput).pipe(Effect.result))._tag, "Failure"); + assert.strictEqual(yield* open(a), replacement); + assert.equal((yield* Ref.get(state)).openCount, 2); + }).pipe( + Effect.provide( + layerTest({ + state, + pauseAttachWrite, + compactThread: Effect.void, + idleTimeoutMs: 60_000, + beforeOpen: () => Ref.set(stale, false), + hasStaleCredentials: Ref.get(stale), + hasPendingBackgroundWork: Ref.getAndSet(pauseProbe, false).pipe( + Effect.flatMap((pause) => + pause + ? Deferred.succeed(probeEntered, undefined).pipe( + Effect.andThen(Deferred.await(probeContinue)), + ) + : Effect.void, + ), + Effect.andThen(Ref.get(background)), + ), + beforeResume: Ref.getAndSet(pauseResume, false).pipe( + Effect.flatMap((pause) => + pause + ? Deferred.succeed(resumeEntered, undefined).pipe( + Effect.andThen(Deferred.await(resumeContinue)), + ) + : Effect.void, + ), + ), + }), + ), + ); + }), +); diff --git a/apps/server/src/orchestration-v2/ProviderSessionManager.ts b/apps/server/src/orchestration-v2/ProviderSessionManager.ts index 00b42d510bd91..574017db6a25f 100644 --- a/apps/server/src/orchestration-v2/ProviderSessionManager.ts +++ b/apps/server/src/orchestration-v2/ProviderSessionManager.ts @@ -202,6 +202,7 @@ interface LiveSessionEntry { readonly scope: Scope.Closeable; readonly idleGeneration: number; readonly busyTurns: ReadonlySet; + readonly pendingOperations: number; readonly lastActivityAtMs: number; readonly idleFiber: Fiber.Fiber | null; readonly pinnedSinceMs: number | null; @@ -893,6 +894,7 @@ export const layerWithOptions = ( function* (input: { readonly providerSessionId: ProviderSessionId; readonly onlyIfIdleGeneration?: number; + readonly onlyIfIdleRuntime?: ProviderAdapter.ProviderAdapterV2SessionRuntime; }): Effect.fn.Return, DateTime.Utc]> { const key = sessionKey(input.providerSessionId); const candidate = (yield* Ref.get(sessions)).get(key); @@ -905,9 +907,16 @@ export const layerWithOptions = ( if (existing !== candidate) { return [existing === undefined ? "gone" : "changed", current] as const; } + if ( + input.onlyIfIdleRuntime !== undefined && + existing.runtime !== input.onlyIfIdleRuntime + ) { + return ["kept", current] as const; + } if ( input.onlyIfIdleGeneration !== undefined && (existing.busyTurns.size > 0 || + existing.pendingOperations > 0 || existing.idleGeneration !== input.onlyIfIdleGeneration) ) { return ["kept", current] as const; @@ -970,6 +979,7 @@ export const layerWithOptions = ( readonly detachedThreadId?: ThreadId; readonly cancelIdleFiber?: boolean; readonly onlyIfIdleGeneration?: number; + readonly onlyIfIdleRuntime?: ProviderAdapter.ProviderAdapterV2SessionRuntime; readonly gracefulSubscribers?: boolean; }) => Effect.acquireUseRelease( @@ -1145,6 +1155,36 @@ export const layerWithOptions = ( }, ); + const releaseIfStaleCredentials = Effect.fn( + "ProviderSessionManager.releaseIfStaleCredentials", + )( + function* (entry: LiveSessionEntry) { + const runtime = entry.runtime; + if ( + entry.busyTurns.size > 0 || + entry.pendingOperations > 0 || + runtime.hasStaleCredentials === undefined + ) + return; + if (!(yield* runtime.hasStaleCredentials)) return; + if ( + runtime.hasPendingBackgroundWork !== undefined && + (yield* runtime.hasPendingBackgroundWork.pipe( + Effect.catchCause(() => Effect.succeed(true)), + )) + ) + return; + yield* releaseEntry({ + providerSessionId: runtime.providerSessionId, + reason: "manual_shutdown", + detail: "Codex sign-in changed. Send the message again to use the current sign-in.", + onlyIfIdleGeneration: entry.idleGeneration, + onlyIfIdleRuntime: runtime, + }); + }, + (effect, entry) => effect.pipe(logReleaseFailure(entry.runtime.providerSessionId)), + ); + const withActivityError = ( providerSessionId: ProviderSessionId, effect: Effect.Effect, @@ -1681,7 +1721,46 @@ export const layerWithOptions = ( modelFamily: normalizeModelMetricLabel(model), }, }); - return { + const withTurnActivity = + (start: typeof runtime.startTurn, operation = "send"): typeof runtime.startTurn => + (input) => + observeActivity( + providerSessionId, + ensureThreadAttached({ + providerSessionId, + threadId: input.threadId, + providerInstanceId: runtime.instanceId, + }), + ).pipe( + // Failed or stopped starts may omit turn.terminal; clear their turns so sessions can idle. + Effect.andThen( + holdThreadLoaded({ + providerSessionId, + threadId: input.threadId, + providerThread: input.providerThread, + }), + ), + Effect.andThen( + Effect.acquireUseRelease( + observeActivity( + providerSessionId, + markBusy( + providerSessionId, + busyTurnKey(input.providerThread.id, input.runOrdinal), + ), + ), + () => start(input).pipe(turnMetrics(operation, input.modelSelection.model)), + (_, exit) => + Exit.isFailure(exit) + ? observeActivity( + providerSessionId, + markIdle(providerSessionId, input.providerThread.id, input.runOrdinal), + ) + : Effect.void, + ), + ), + ); + const exposed: ProviderAdapter.ProviderAdapterV2SessionRuntime = { ...runtime, subscribeEvents, events: Stream.unwrap( @@ -1785,44 +1864,12 @@ export const layerWithOptions = ( }), ), ), - startTurn: (input) => - observeActivity( - providerSessionId, - ensureThreadAttached({ - providerSessionId, - threadId: input.threadId, - providerInstanceId: runtime.instanceId, + startTurn: withTurnActivity(runtime.startTurn), + ...(runtime.compactThread === undefined + ? {} + : { + compactThread: withTurnActivity(runtime.compactThread, "compact"), }), - ).pipe( - // Failed or stopped starts may omit turn.terminal; clear their turns so sessions can idle. - Effect.andThen( - holdThreadLoaded({ - providerSessionId, - threadId: input.threadId, - providerThread: input.providerThread, - }), - ), - Effect.andThen( - Effect.acquireUseRelease( - observeActivity( - providerSessionId, - markBusy( - providerSessionId, - busyTurnKey(input.providerThread.id, input.runOrdinal), - ), - ), - () => - runtime.startTurn(input).pipe(turnMetrics("send", input.modelSelection.model)), - (_, exit) => - Exit.isFailure(exit) - ? observeActivity( - providerSessionId, - markIdle(providerSessionId, input.providerThread.id, input.runOrdinal), - ) - : Effect.void, - ), - ), - ), steerTurn: (input) => observeActivity(providerSessionId, touchActivity(providerSessionId)).pipe( Effect.andThen(runtime.steerTurn(input).pipe(turnMetrics("steer"))), @@ -1840,6 +1887,61 @@ export const layerWithOptions = ( ), ), }; + // Reserve before any async attachment/start work; an old handle must never send after replacement. + const withRuntimeUse = (effect: Effect.Effect) => + runtime.hasStaleCredentials === undefined + ? effect + : Effect.acquireUseRelease( + Ref.modify(sessions, (current) => { + const key = sessionKey(providerSessionId); + const entry = current.get(key); + if (entry?.runtime !== runtime) return [false, current] as const; + return [ + true, + new Map(current).set(key, { + ...entry, + pendingOperations: entry.pendingOperations + 1, + idleGeneration: entry.idleGeneration + 1, + }), + ] as const; + }).pipe( + Effect.flatMap((reserved) => + reserved + ? Effect.void + : Effect.fail( + new ProviderAdapter.ProviderAdapterProtocolError({ + driver: runtime.driver, + detail: + "Provider session changed before this request started. Send the message again.", + }), + ), + ), + ), + () => effect, + () => + Ref.update(sessions, (current) => { + const key = sessionKey(providerSessionId); + const entry = current.get(key); + return entry?.runtime !== runtime + ? current + : new Map(current).set(key, { + ...entry, + pendingOperations: entry.pendingOperations - 1, + }); + }).pipe(Effect.andThen(scheduleIdleReleaseInternal(providerSessionId))), + ); + return { + ...exposed, + ensureThread: (input) => withRuntimeUse(exposed.ensureThread(input)), + resumeThread: (input) => withRuntimeUse(exposed.resumeThread(input)), + forkThread: (input) => withRuntimeUse(exposed.forkThread(input)), + startTurn: (input) => withRuntimeUse(exposed.startTurn(input)), + ...(exposed.compactThread === undefined + ? {} + : { + compactThread: (input) => withRuntimeUse(exposed.compactThread!(input)), + }), + }; }; const persistProviderSessionUpdate = Effect.fn( @@ -2065,6 +2167,8 @@ export const layerWithOptions = ( } } const key = sessionKey(input.providerSessionId); + const live = (yield* Ref.get(sessions)).get(key); + if (live !== undefined) yield* releaseIfStaleCredentials(live); const existing = (yield* Ref.get(sessions)).get(key); if (existing !== undefined) { const attached = existing.attachedThreadIds.has(input.threadId); @@ -2181,6 +2285,7 @@ export const layerWithOptions = ( scope: sessionScope, idleGeneration: 0, busyTurns: new Set(), + pendingOperations: 0, lastActivityAtMs: now, idleFiber: null, pinnedSinceMs: null, diff --git a/packages/provider-core/src/server/ProviderAdapter.ts b/packages/provider-core/src/server/ProviderAdapter.ts index 7c77b7bb0ac6f..75ce2fa16b348 100644 --- a/packages/provider-core/src/server/ProviderAdapter.ts +++ b/packages/provider-core/src/server/ProviderAdapter.ts @@ -548,6 +548,8 @@ export interface ProviderAdapterV2SessionRuntime { readonly events: Stream.Stream; /** Manager runtimes broadcast independent subscriptions; adapters may expose only a single-consumer stream. */ readonly subscribeEvents?: Effect.Effect; + /** Requests replacement on the next idle open; never retries an accepted turn. */ + readonly hasStaleCredentials?: Effect.Effect; readonly hasPendingBackgroundWork?: Effect.Effect; /** Overrides the session-wide probe so sibling threads cannot keep this root subscription open. */ readonly hasPendingBackgroundWorkForThread?: (