diff --git a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts index 84c78b91453c2..0da177a31f7ed 100644 --- a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts @@ -6018,6 +6018,148 @@ describe("CodexAdapterV2 post-settle continuation", () => { ), ); + // regression: #27773 a 429 on a run's first request still reports the reset time + it.effect( + "recovers resetAt from a fresh rate-limits read when the 429 never primed the snapshot", + () => + Effect.scoped( + Effect.gen(function* () { + const nativeThreadId = "native-limit-cold-429"; + const nativeTurnId = "turn-limit-cold-429"; + const message = "Provider stopped this request."; + const resetAt = "2033-05-19T07:20:00.000Z"; + const transcript = makeCodexReplayTranscript({ + scenario: "codex-limit-cold-429", + entries: [ + ...codexReplayPreamble({ nativeThreadId, nativeTurnId, prompt: "Continue." }), + { + type: "emit_inbound", + label: "error", + frame: { + method: "error", + params: { + threadId: nativeThreadId, + turnId: nativeTurnId, + willRetry: false, + error: { message, codexErrorInfo: "usageLimitExceeded" }, + }, + }, + }, + // The 429's own notification is an empty non-codex snapshot the adapter must ignore. + { + type: "emit_inbound", + label: "account/rateLimits/updated", + frame: { + method: "account/rateLimits/updated", + params: { + rateLimits: { limitId: "premium", primary: null, secondary: null }, + }, + }, + }, + { + type: "emit_inbound", + label: "turn/completed", + frame: { + method: "turn/completed", + params: { + threadId: nativeThreadId, + turn: { + ...makeCodexReplayTurn({ id: nativeTurnId, status: "failed" }), + error: { message, codexErrorInfo: "usageLimitExceeded" }, + }, + }, + }, + }, + { + type: "expect_outbound", + label: "account/rateLimits/read", + frame: { id: 4, method: "account/rateLimits/read", params: null }, + }, + { + type: "emit_inbound", + label: "account/rateLimits/read", + frame: { + id: 4, + result: { + rateLimits: { limitId: "premium", primary: null, secondary: null }, + rateLimitsByLimitId: { + codex: { + limitId: "codex", + primary: { + usedPercent: 100, + resetsAt: 2000100000, + windowDurationMins: 300, + }, + }, + }, + }, + }, + }, + ], + }); + const resetReceipt = yield* Deferred.make(); + const readRequests: Array = []; + const harness = yield* makeCodexReplayHarness( + transcript, + (event) => + event.type === "turn_item.updated" && + event.turnItem.type === "error" && + event.turnItem.failure.resetAt === resetAt + ? Deferred.succeed(resetReceipt, undefined) + : Effect.void, + (method) => + Effect.sync(() => { + readRequests.push(method); + }), + ); + yield* harness.runtime.startTurn( + makeCodexTestTurnInput({ + threadId: harness.threadId, + providerThread: harness.providerThread, + now: yield* DateTime.now, + text: "Continue.", + attemptId: RunAttemptId.make("attempt-limit-cold-429"), + }), + ); + yield* harness.firstTerminal; + const terminal = harness.terminalEvents()[0]; + assert.equal(terminal?.status, "failed"); + if (terminal?.status !== "failed") return; + assert.equal(terminal.failure.class, "usage_limit"); + yield* Deferred.await(resetReceipt); + assert.include(readRequests, "account/rateLimits/read"); + const item = harness.events.find( + (event) => + event.type === "turn_item.updated" && + event.turnItem.type === "error" && + event.turnItem.failure.resetAt === resetAt, + ); + assert.isDefined(item); + assert.equal(terminal.failure.resetAt, null); + assert.isBelow(harness.events.indexOf(terminal), harness.events.indexOf(item!)); + assert.lengthOf( + harness.events.filter( + (event) => + event.type === "turn_item.updated" && event.usageLimitResetAtEnrichment === true, + ), + 1, + ); + assert.isTrue( + item?.type === "turn_item.updated" && item.usageLimitResetAtEnrichment === true, + ); + }).pipe( + Effect.provide( + Layer.mergeAll( + IdAllocator.layer, + McpProviderSessions.layer, + NodeServices.layer, + NodeCrypto.layer, + ), + ), + ), + ), + ); + // regression: #28032 an unlisted codexErrorInfo dropped the failed turn/completed it.effect("fails the turn when Codex reports an error info the schema does not list", () => Effect.scoped( diff --git a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts index 07c4e4e72eca7..73b36f73a0096 100644 --- a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts @@ -21,6 +21,7 @@ import { buildCodexInitializeParams, codexAppServerEnvironment, codexDefaultReasoningEffortOverride, + RATE_LIMITS_PROBE_TIMEOUT_MS, } from "../../provider/CodexProvider.ts"; import { codexRateLimitsToUpdate, @@ -86,6 +87,7 @@ import * as Context from "effect/Context"; import * as Digest from "effect/crypto/Digest"; import * as DateTime from "effect/DateTime"; import * as Deferred from "effect/Deferred"; +import * as Duration from "effect/Duration"; import * as Effect from "effect/Effect"; import * as Fiber from "effect/Fiber"; import * as FileSystem from "effect/FileSystem"; @@ -4491,29 +4493,58 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( }).pipe(Effect.orDie), ); + const applyRateLimitSnapshot = Effect.fnUntraced(function* ( + update: CodexRateLimitSnapshot, + ) { + yield* Ref.update(rateLimitSnapshot, (previous) => + mergeCodexRateLimits(previous, update), + ); + const snapshot = yield* Ref.get(rateLimitSnapshot); + for (const item of (yield* Ref.get(limitedTurnItems)).values()) { + const resetAt = failureResetAt(item.failure, snapshot); + if (resetAt === null || item.failure.resetAt != null) continue; + const updated = { + ...item, + updatedAt: yield* DateTime.now, + failure: { ...item.failure, resetAt }, + }; + yield* Ref.update(limitedTurnItems, (current) => + new Map(current).set(item.providerThreadId!, updated), + ); + yield* emitProviderEvent({ + type: "turn_item.updated", + driver: CODEX_PROVIDER, + turnItem: updated, + usageLimitResetAtEnrichment: true, + }); + } + }); + + // A 429 that kills a run's first request never delivers a codex-family + // notification, so a usage-limit failure without a reset time reads + // fresh limits straight from the app-server. + const backfillRateLimitSnapshot = Effect.fn("CodexAdapterV2.backfillRateLimitSnapshot")( + function* (): Effect.fn.Return { + const response = yield* client.request("account/rateLimits/read", null).pipe( + Effect.timeoutOption(Duration.millis(RATE_LIMITS_PROBE_TIMEOUT_MS)), + Effect.map(Option.getOrUndefined), + Effect.catch((error) => + Effect.logWarning("Codex rate-limit read failed.", { cause: error }).pipe( + Effect.as(undefined), + ), + ), + ); + if (response === undefined) return; + yield* applyRateLimitSnapshot( + response.rateLimitsByLimitId?.codex ?? response.rateLimits, + ); + }, + ); + yield* client.handleServerNotification("account/rateLimits/updated", (payload) => Effect.gen(function* () { - const previous = yield* Ref.getAndUpdate(rateLimitSnapshot, (previous) => - mergeCodexRateLimits(previous, payload.rateLimits), - ); - const snapshot = yield* Ref.get(rateLimitSnapshot); - for (const item of (yield* Ref.get(limitedTurnItems)).values()) { - const resetAt = failureResetAt(item.failure, snapshot); - if (resetAt === null || item.failure.resetAt != null) continue; - const updated = { - ...item, - updatedAt: yield* DateTime.now, - failure: { ...item.failure, resetAt }, - }; - yield* Ref.update(limitedTurnItems, (current) => - new Map(current).set(item.providerThreadId!, updated), - ); - yield* emitProviderEvent({ - type: "turn_item.updated", - driver: CODEX_PROVIDER, - turnItem: updated, - }); - } + const previous = yield* Ref.get(rateLimitSnapshot); + yield* applyRateLimitSnapshot(payload.rateLimits); const update = codexRateLimitsToUpdate(payload.rateLimits, previous); if (update && adapterOptions.onUsageLimits) { const checkedAt = DateTime.formatIso(yield* DateTime.now); @@ -5920,6 +5951,11 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( yield* Ref.update(limitedTurnItems, (items) => new Map(items).set(context.providerThread.id, item), ); + // Forked because the read's own response arrives on the protocol + // reader fiber this terminal handler runs on. + if (current.failure.resetAt == null) { + yield* backfillRateLimitSnapshot().pipe(Effect.forkIn(scope)); + } } }); diff --git a/apps/server/src/orchestration-v2/EventSink.ts b/apps/server/src/orchestration-v2/EventSink.ts index b210d3c9840ae..5fc1c884ce291 100644 --- a/apps/server/src/orchestration-v2/EventSink.ts +++ b/apps/server/src/orchestration-v2/EventSink.ts @@ -346,7 +346,38 @@ const layerBase: Layer.Layer< event.payload, event.payload.runId === null ? undefined : runOrdinals.get(event.payload.runId), ) - .pipe(Effect.map((payload) => ({ ...event, payload }))) + .pipe( + Effect.flatMap((payload) => + Effect.gen(function* () { + if ( + event.driver === "codex" && + payload.type === "error" && + payload.failure.class === "usage_limit" && + payload.failure.resetAt == null + ) { + // The direct enrichment lane can commit before the queued terminal. + const existing = yield* projectionStore.getTurnItem({ + threadId: payload.threadId, + itemId: payload.id, + }); + if ( + existing?.type === "error" && + existing.failure.class === "usage_limit" && + existing.failure.resetAt != null + ) { + return { + ...event, + payload: { + ...payload, + failure: { ...payload.failure, resetAt: existing.failure.resetAt }, + }, + }; + } + } + return { ...event, payload }; + }), + ), + ) : Effect.succeed(event), { concurrency: 1 }, ); diff --git a/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts b/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts index b1e05577bfbae..04b29560f92ec 100644 --- a/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts +++ b/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts @@ -29,6 +29,7 @@ import * as Layer from "effect/Layer"; import * as Option from "effect/Option"; import * as Queue from "effect/Queue"; import * as Ref from "effect/Ref"; +import * as Schema from "effect/Schema"; import * as Stream from "effect/Stream"; import { TestClock } from "effect/testing"; import { HttpServer } from "effect/http"; @@ -53,8 +54,11 @@ import * as GitVcsDriver from "../vcs/GitVcsDriver.ts"; import * as ThreadCommandExecutor from "./ThreadCommandExecutor.ts"; import * as ProviderSessionManager from "./ProviderSessionManager.ts"; import * as CommandOutputHub from "./CommandOutputHub.ts"; +import { makeProviderFailureTurnItem } from "@t3tools/provider-core/server/failure"; import * as ProviderAdapter from "@t3tools/provider-core/server/ProviderAdapter"; +const decodeProviderEvent = Schema.decodeUnknownSync(ProviderAdapter.ProviderAdapterV2Event); + const layerTestDatabase = SqlitePersistence.layerMemory; const layerTestStores = Layer.merge(EventStore.layer, ProjectionStore.layer).pipe( Layer.provide(layerTestDatabase), @@ -426,6 +430,7 @@ const makeProviderAdapter = Effect.fnUntraced(function* ( readonly mcpConfigs?: Ref.Ref< ReadonlyArray >; + readonly beforeEvent?: (event: ProviderAdapter.ProviderAdapterV2Event) => Effect.Effect; readonly beforeOpen?: (input: { readonly providerSessionId: ProviderSessionId; readonly initialProviderItemIdentityVersion?: 2; @@ -511,7 +516,10 @@ const makeProviderAdapter = Effect.fnUntraced(function* ( cause: "process exited", }), ) - : Stream.fromQueue(events), + : Stream.fromQueue(events).pipe( + Stream.rechunk(1), + Stream.tap((event) => options.beforeEvent?.(event) ?? Effect.void), + ), ...(options.backgroundTasksEndWithProcess === false ? {} : { backgroundTasksEndWithProcess: true as const }), @@ -566,6 +574,7 @@ function layerTest(input: { readonly mcpConfigs?: Ref.Ref< ReadonlyArray >; + readonly beforeEvent?: (event: ProviderAdapter.ProviderAdapterV2Event) => Effect.Effect; readonly beforeOpen?: (input: { readonly providerSessionId: ProviderSessionId; readonly initialProviderItemIdentityVersion?: 2; @@ -603,6 +612,7 @@ function layerTest(input: { const layerRegistry = Layer.unwrap( makeProviderAdapter(input.state, { failEventStream: input.failEventStream ?? false, + ...(input.beforeEvent === undefined ? {} : { beforeEvent: input.beforeEvent }), ...(input.capabilities === undefined ? {} : { capabilities: input.capabilities }), ...(input.mcpConfigs === undefined ? {} : { mcpConfigs: input.mcpConfigs }), ...(input.beforeOpen === undefined ? {} : { beforeOpen: input.beforeOpen }), @@ -648,6 +658,7 @@ function layerTest(input: { ), ); return Layer.mergeAll( + layerProviderEventIngestorTest, layerTestStores, layerConfiguredEventSink, IdAllocator.layer, @@ -3260,6 +3271,224 @@ it.effect.each(["manual_shutdown", "server_shutdown"] as const)( }), ); +it.effect.each([false, true])( + "ProviderSessionManagerV2 persists reset enrichment with subscriber present: %s", + (subscribed) => + Effect.gen(function* () { + const state = yield* Ref.make(emptyState); + const consumed = yield* Deferred.make(); + yield* Effect.gen(function* () { + const eventSink = yield* EventSink.EventSinkV2; + const idAllocator = yield* IdAllocator.IdAllocatorV2; + const manager = yield* ProviderSessionManager.ProviderSessionManagerV2; + const ingestor = yield* ProviderEventIngestor.ProviderEventIngestorV2; + const projectionStore = yield* ProjectionStore.ProjectionStoreV2; + const now = yield* DateTime.now; + const projectId = yield* idAllocator.allocate.project({ fixtureName: "reset-enrichment" }); + const threadId = yield* idAllocator.allocate.thread({ + fixtureName: "reset-enrichment", + projectId, + }); + const otherThreadId = yield* idAllocator.allocate.thread({ + fixtureName: "reset-enrichment-other", + projectId, + }); + const providerSessionId = yield* idAllocator.allocate.providerSession({ + providerInstanceId: modelSelection.instanceId, + threadId, + }); + const providerThread = makeProviderThread({ + idAllocator, + threadId, + providerSessionId, + now, + }); + const runId = idAllocator.derive.run({ threadId, ordinal: 1 }); + const nodeId = idAllocator.derive.rootNode({ runId }); + const providerTurnId = idAllocator.derive.providerTurn({ + driver: CODEX_DRIVER, + nativeTurnId: "limited-turn", + }); + yield* eventSink.write({ + events: [ + yield* makeThreadCreatedEvent({ idAllocator, threadId, now }), + yield* makeThreadCreatedEvent({ idAllocator, threadId: otherThreadId, now }), + ], + }); + const runtime = yield* manager.open({ + threadId, + providerSessionId, + modelSelection, + runtimePolicy, + }); + assert.isDefined(runtime.subscribeEvents); + const subscription = yield* runtime.subscribeEvents!; + const receivedEvents = yield* Queue.unbounded(); + yield* subscription.events.pipe( + Stream.runForEach((event) => Queue.offer(receivedEvents, event)), + Effect.forkScoped, + ); + const queue = (yield* Ref.get(state)).eventQueues.get(String(providerSessionId))!; + const failure = { + class: "usage_limit", + message: "Usage limit reached", + code: null, + retryable: null, + resetAt: null, + } as const; + const terminal = { + type: "turn.terminal", + driver: CODEX_DRIVER, + providerThreadId: providerThread.id, + providerTurnId, + runOrdinal: 1, + status: "failed", + failure, + failureItemOrdinal: 1, + threadDisposition: "reusable", + } as const; + const initialItem = makeProviderFailureTurnItem({ + driver: CODEX_DRIVER, + threadId, + runId, + nodeId, + providerThreadId: providerThread.id, + providerTurnId, + itemOrdinal: 1, + failure: { ...failure, resetAt: "2026-11-02T12:00:00.000Z" }, + occurredAt: now, + }); + const beforeInitial = yield* eventSink.latestSequence({ threadId }); + for (const driver of [CODEX_DRIVER, ProviderDriverKind.make("muse")]) { + const initial = { type: "turn_item.updated", driver, turnItem: initialItem } as const; + yield* Queue.offer(queue, initial); + assert.deepEqual(yield* Queue.take(receivedEvents), initial); + assert.equal(yield* eventSink.latestSequence({ threadId }), beforeInitial); + } + yield* Queue.offer(queue, terminal); + assert.deepEqual(yield* Queue.take(receivedEvents), terminal); + yield* ingestor.ingestNormalized({ + providerSessionId, + providerInstanceId: modelSelection.instanceId, + threadId, + runId, + nodeId, + event: terminal, + }); + const initialProjection = yield* projectionStore.getThreadProjection(threadId); + assert.deepInclude(initialProjection.turnItems[0], { failure }); + if (!subscribed) yield* subscription.close; + const resetAt = "2026-11-02T12:00:00.000Z"; + const item = makeProviderFailureTurnItem({ + driver: CODEX_DRIVER, + threadId, + runId, + nodeId, + providerThreadId: providerThread.id, + providerTurnId, + itemOrdinal: 1, + failure: { ...failure, resetAt }, + occurredAt: now, + }); + const newerRunId = idAllocator.derive.run({ threadId, ordinal: 2 }); + const newerItem = makeProviderFailureTurnItem({ + driver: CODEX_DRIVER, + threadId, + runId: newerRunId, + nodeId: idAllocator.derive.rootNode({ runId: newerRunId }), + providerThreadId: providerThread.id, + providerTurnId: idAllocator.derive.providerTurn({ + driver: CODEX_DRIVER, + nativeTurnId: "newer-turn", + }), + itemOrdinal: 2, + failure: { ...failure, resetAt: "2026-11-03T12:00:00.000Z" }, + occurredAt: now, + }); + yield* ingestor.ingestNormalized({ + providerSessionId, + providerInstanceId: modelSelection.instanceId, + threadId, + event: { type: "turn_item.updated", driver: CODEX_DRIVER, turnItem: newerItem }, + }); + const afterSequence = yield* eventSink.latestSequence({ threadId }); + yield* Queue.offerAll(queue, [ + decodeProviderEvent({ + type: "turn_item.updated", + driver: CODEX_DRIVER, + turnItem: item, + usageLimitResetAtEnrichment: true, + }), + { + type: "provider_session.updated", + driver: CODEX_DRIVER, + providerSession: runtime.providerSession, + }, + ]); + yield* Deferred.await(consumed); + const projection = yield* projectionStore.getThreadProjection(threadId); + assert.deepEqual( + projection.turnItems.find((candidate) => candidate.id === item.id), + item, + ); + assert.deepEqual( + projection.turnItems.find((candidate) => candidate.id === newerItem.id), + newerItem, + ); + const updates = yield* eventSink + .stream({ threadId, afterSequence }) + .pipe(Stream.take(1), Stream.runCollect); + assert.equal(updates.length, 1); + assert.deepInclude(updates[0]!.event, { + type: "turn-item.updated", + threadId, + runId, + nodeId, + payload: item, + }); + assert.equal(yield* eventSink.latestSequence({ threadId }), afterSequence + 1); + yield* ingestor.ingestNormalized({ + providerSessionId, + providerInstanceId: modelSelection.instanceId, + threadId, + runId, + nodeId, + event: terminal, + }); + const afterTerminal = yield* projectionStore.getThreadProjection(threadId); + const persisted = afterTerminal.turnItems.find((candidate) => candidate.id === item.id); + assert.deepInclude(persisted, { failure: { ...failure, resetAt } }); + assert.deepEqual( + afterTerminal.turnItems.find((candidate) => candidate.id === newerItem.id), + newerItem, + ); + const terminalUpdates = yield* eventSink + .stream({ threadId, afterSequence: afterSequence + 1 }) + .pipe(Stream.take(1), Stream.runCollect); + assert.deepInclude(terminalUpdates[0]!.event.payload, { + failure: { ...failure, resetAt }, + }); + assert.deepEqual((yield* projectionStore.getThreadProjection(otherThreadId)).turnItems, []); + if (subscribed) { + const next = yield* Queue.take(receivedEvents); + assert.equal(next.type, "provider_session.updated"); + yield* subscription.close; + } + }).pipe( + Effect.provide( + layerTest({ + state, + idleTimeoutMs: 1000, + beforeEvent: (event) => + event.type === "provider_session.updated" + ? Deferred.succeed(consumed, undefined).pipe(Effect.asVoid) + : Effect.void, + }), + ), + ); + }), +); + it.effect("ProviderSessionManagerV2 persists session-scoped runtime requests without a run", () => 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 f06011676c667..3ed9ea211add3 100644 --- a/apps/server/src/orchestration-v2/ProviderSessionManager.ts +++ b/apps/server/src/orchestration-v2/ProviderSessionManager.ts @@ -2004,7 +2004,15 @@ export const layerWithOptions = ( // (project trust, login, or session-switch hooks). Persist // their runless request artifacts directly so the normal T3 // request UI can answer them and unblock session setup. - const threadId = sessionScopedRuntimeRequestThreadId(event); + const threadId = + event.type === "turn_item.updated" && + event.driver === "codex" && + event.usageLimitResetAtEnrichment === true && + event.turnItem.type === "error" && + event.turnItem.failure.class === "usage_limit" && + event.turnItem.failure.resetAt != null + ? event.turnItem.threadId + : sessionScopedRuntimeRequestThreadId(event); if (threadId !== undefined) { yield* Effect.gen(function* () { const current = (yield* Ref.get(sessions)).get( diff --git a/apps/server/src/provider/CodexProvider.ts b/apps/server/src/provider/CodexProvider.ts index 2d546b36a9b15..1069599b6105e 100644 --- a/apps/server/src/provider/CodexProvider.ts +++ b/apps/server/src/provider/CodexProvider.ts @@ -55,7 +55,7 @@ import { import * as HostProcess from "@t3tools/shared/HostProcess"; import packageJson from "../../package.json" with { type: "json" }; const isCodexAppServerSpawnError = Schema.is(CodexErrors.CodexAppServerSpawnError); -const RATE_LIMITS_PROBE_TIMEOUT_MS = 3_000; +export const RATE_LIMITS_PROBE_TIMEOUT_MS = 3_000; type CodexRateLimitsProbe = | { diff --git a/packages/provider-core/src/server/ProviderAdapter.ts b/packages/provider-core/src/server/ProviderAdapter.ts index 9ea7597e6d087..062eae7d9e289 100644 --- a/packages/provider-core/src/server/ProviderAdapter.ts +++ b/packages/provider-core/src/server/ProviderAdapter.ts @@ -125,6 +125,8 @@ export const ProviderAdapterV2Event = Schema.Union([ type: Schema.Literal("turn_item.updated"), driver: ProviderDriverKind, turnItem: OrchestrationV2TurnItem, + /** Codex reset enrichment outlives the failed run's event subscription. */ + usageLimitResetAtEnrichment: Schema.optional(Schema.Literal(true)), }), // Start before the root terminal; clear only after the final covered projection. // Consumer lifetime cannot depend on a producer probe that runs ahead of its queue.