From 07623f8b35f6ccd7c5715dbadaabd9f75419b075 Mon Sep 17 00:00:00 2001 From: imluckii Date: Fri, 30 Oct 2026 19:24:50 +0000 Subject: [PATCH 1/3] fix(server): codex usage-limit failures keep their reset time --- .../Adapters/CodexAdapterV2.test.ts | 125 ++++++++++++++++++ .../Adapters/CodexAdapterV2.ts | 75 ++++++++--- apps/server/src/provider/CodexProvider.ts | 2 +- 3 files changed, 181 insertions(+), 21 deletions(-) diff --git a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts index 9dfe62bc4d36b..d01a40aae708f 100644 --- a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts @@ -5083,6 +5083,131 @@ 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); + }).pipe( + Effect.provide( + Layer.mergeAll(IdAllocator.layer, McpProviderSessions.layer, NodeServices.layer), + ), + ), + ), + ); + const FAILED_SCENARIO = "codex-failed-mid-command"; const FAILED_NATIVE_THREAD = "native-codex-failed-thread"; const FAILED_NATIVE_TURN = "native-codex-failed-turn"; diff --git a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts index 59c26bd40b956..e2280f95c8b60 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, @@ -84,6 +85,7 @@ import * as Context from "effect/Context"; import * as Crypto from "effect/Crypto"; 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"; @@ -4361,28 +4363,56 @@ 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 resetAt = codexUsageLimitResetAt(yield* Ref.get(rateLimitSnapshot)); + for (const item of (yield* Ref.get(limitedTurnItems)).values()) { + 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, + }); + } + }); + + // 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 resetAt = codexUsageLimitResetAt(yield* Ref.get(rateLimitSnapshot)); - for (const item of (yield* Ref.get(limitedTurnItems)).values()) { - 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); @@ -5763,6 +5793,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/provider/CodexProvider.ts b/apps/server/src/provider/CodexProvider.ts index 2b2d5501881b4..dc5e813f42360 100644 --- a/apps/server/src/provider/CodexProvider.ts +++ b/apps/server/src/provider/CodexProvider.ts @@ -53,7 +53,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 = | { From e0a6ff61b1ba7541a663df19994ece50bdf2cf2d Mon Sep 17 00:00:00 2001 From: imluckii Date: Mon, 2 Nov 2026 09:34:12 +0000 Subject: [PATCH 2/3] fix(server): persist Codex reset enrichment after run settlement --- .../Adapters/CodexAdapterV2.test.ts | 12 + .../Adapters/CodexAdapterV2.ts | 1 + .../ProviderSessionManager.test.ts | 208 +++++++++++++++++- .../ProviderSessionManager.ts | 10 +- .../src/server/ProviderAdapter.ts | 2 + 5 files changed, 231 insertions(+), 2 deletions(-) diff --git a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts index d01a40aae708f..658e312ca037c 100644 --- a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts @@ -5200,6 +5200,18 @@ describe("CodexAdapterV2 post-settle continuation", () => { 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), diff --git a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts index e2280f95c8b60..5ef23d824e0e1 100644 --- a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts @@ -4384,6 +4384,7 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( type: "turn_item.updated", driver: CODEX_PROVIDER, turnItem: updated, + usageLimitResetAtEnrichment: true, }); } }); diff --git a/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts b/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts index 5bbb3926e2b66..8608d1eb2fb23 100644 --- a/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts +++ b/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts @@ -28,6 +28,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"; @@ -52,8 +53,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), @@ -329,6 +333,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; @@ -414,7 +419,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 }), @@ -469,6 +477,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; @@ -506,6 +515,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 }), @@ -552,6 +562,7 @@ function layerTest(input: { ), ); return Layer.mergeAll( + layerProviderEventIngestorTest, layerTestStores, layerConfiguredEventSink, IdAllocator.layer, @@ -3164,6 +3175,201 @@ 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, + }); + 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); + 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 03a7278774a64..d005921bc3e5f 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/packages/provider-core/src/server/ProviderAdapter.ts b/packages/provider-core/src/server/ProviderAdapter.ts index cadf4173a60b1..91159c6cc7593 100644 --- a/packages/provider-core/src/server/ProviderAdapter.ts +++ b/packages/provider-core/src/server/ProviderAdapter.ts @@ -118,6 +118,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. From dc900cef7b037987db929e520f6d921ad876e831 Mon Sep 17 00:00:00 2001 From: imluckii Date: Tue, 3 Nov 2026 11:44:40 +0000 Subject: [PATCH 3/3] fix(server): preserve enriched Codex usage-limit reset times --- apps/server/src/orchestration-v2/EventSink.ts | 33 ++++++++++++++++++- .../ProviderSessionManager.test.ts | 23 +++++++++++++ 2 files changed, 55 insertions(+), 1 deletion(-) 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 86a341bf6ff73..74e98141b3053 100644 --- a/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts +++ b/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts @@ -3376,6 +3376,8 @@ it.effect.each([false, true])( 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({ @@ -3446,6 +3448,27 @@ it.effect.each([false, true])( 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);