Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions apps/mobile/src/features/threads/ThreadComposer.tsx
Original file line number Diff line number Diff line change
Expand Up @@ -531,6 +531,7 @@ export const ThreadComposer = function ThreadComposer(props: ThreadComposerProps
props.serverConfig?.providers ?? [],
props.serverConfig?.usageLimitSources ?? [],
Date.now(),
currentModelSelection.model,
);
onShowUsageLimits(report);
if (!report) {
Expand Down
1 change: 1 addition & 0 deletions apps/mobile/src/features/threads/ThreadDetailScreen.tsx
Original file line number Diff line number Diff line change
Expand Up @@ -981,6 +981,7 @@ function ThreadDetailBody(
props.serverConfig?.providers ?? [],
props.serverConfig?.usageLimitSources ?? [],
usageLimitsPanel.now,
props.selectedThread.modelSelection.model,
)
: null;
const showUsageLimits = (report: UsageLimitsReport | null) =>
Expand Down
16 changes: 16 additions & 0 deletions apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -5597,6 +5597,20 @@ describe("CodexAdapterV2 post-settle continuation", () => {
expectedClass: "usage_limit",
},
{ name: "retry", code: "usageLimitExceeded", notification: true, expectedClass: "usage_limit" },
// regression: #26574 a proxy strips the rate-limit snapshot, so the reset is only in the text
{
name: "message-reset",
code: "usageLimitExceeded",
message: "You’ve hit your usage limit. Try again at May 19th, 2033 7:20 AM.",
notification: false,
expectedClass: "usage_limit",
expectedResetAt: DateTime.formatIso(
DateTime.makeZonedUnsafe("2033-05-19T07:21:00", {
timeZone: DateTime.zoneMakeLocal(),
adjustForTimeZone: true,
}),
),
},
{
name: "exhausted-429",
code: "responseTooManyFailedAttempts",
Expand Down Expand Up @@ -5859,6 +5873,8 @@ describe("CodexAdapterV2 post-settle continuation", () => {
assert.equal(terminal.failure.class, scenario.expectedClass);
assert.equal(terminal.threadDisposition, "reusable");
if (scenario.name === "deferred-reset") assert.equal(terminal.failure.resetAt, resetAt);
if ("expectedResetAt" in scenario)
assert.equal(terminal.failure.resetAt, scenario.expectedResetAt);
if (scenario.name === "matching-details")
assert.equal(terminal.failure.message, "Detailed provider allowance explanation.");
if ("resetAfter" in scenario) {
Expand Down
28 changes: 24 additions & 4 deletions apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts
Original file line number Diff line number Diff line change
Expand Up @@ -25,6 +25,7 @@ import {
import {
codexRateLimitsToUpdate,
mergeCodexRateLimits,
codexMessageResetAt,
codexUsageLimitResetAt,
type CodexRateLimitSnapshot,
} from "../../provider/codexUsageLimits.ts";
Expand Down Expand Up @@ -1169,11 +1170,14 @@ function withCodexManagedFailure(
function failureResetAt(
failure: OrchestrationV2ProviderFailure,
snapshot: CodexRateLimitSnapshot | undefined,
now: DateTime.Utc,
codexTimeZone: string | undefined,
) {
// Sharing caps have no reset timestamp in the ordinary Codex allowance snapshot.
return classifyCodexManagedError(failure.code)?.usageLimit
? null
: codexUsageLimitResetAt(snapshot);
: (codexUsageLimitResetAt(snapshot) ??
codexMessageResetAt(failure.message, now, codexTimeZone));
}

interface ActiveCodexTurnContext {
Expand Down Expand Up @@ -1754,6 +1758,7 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* (
settings: resolvedRuntime?.config ?? adapterOptions.settings,
environment: resolvedRuntime?.environment ?? adapterOptions.environment,
});
const codexTimeZone = (resolvedRuntime?.environment ?? adapterOptions.environment).TZ;
const additionalContextByThread = yield* Ref.make(
new Map<
string,
Expand Down Expand Up @@ -4498,7 +4503,12 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* (
);
const snapshot = yield* Ref.get(rateLimitSnapshot);
for (const item of (yield* Ref.get(limitedTurnItems)).values()) {
const resetAt = failureResetAt(item.failure, snapshot);
const resetAt = failureResetAt(
item.failure,
snapshot,
yield* DateTime.now,
codexTimeZone,
);
if (resetAt === null || item.failure.resetAt != null) continue;
const updated = {
...item,
Expand Down Expand Up @@ -5843,7 +5853,12 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* (
failure.class === "usage_limit"
? {
...failure,
resetAt: failureResetAt(failure, yield* Ref.get(rateLimitSnapshot)),
resetAt: failureResetAt(
failure,
yield* Ref.get(rateLimitSnapshot),
yield* DateTime.now,
codexTimeZone,
),
}
: failure,
...(input.providerRetry === undefined
Expand Down Expand Up @@ -5883,7 +5898,12 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* (
...event.failure,
resetAt:
event.failure.resetAt ??
failureResetAt(event.failure, yield* Ref.get(rateLimitSnapshot)),
failureResetAt(
event.failure,
yield* Ref.get(rateLimitSnapshot),
yield* DateTime.now,
codexTimeZone,
),
},
}
: event;
Expand Down
98 changes: 92 additions & 6 deletions apps/server/src/orchestration-v2/ProviderEventIngestor.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -23,6 +23,7 @@ import {
RunAttemptId,
RunId,
RuntimeRequestId,
type ServerProvider,
TurnItemId,
} from "@t3tools/contracts";
import * as Cause from "effect/Cause";
Expand All @@ -45,6 +46,7 @@ import * as EventSink from "./EventSink.ts";
import * as EventStore from "./EventStore.ts";
import * as IdAllocator from "@t3tools/provider-core/server/IdAllocator";
import * as ProjectionStore from "./ProjectionStore.ts";
import { UsageLimitResetFallback } from "../usage/UsageLimitSources.ts";
import * as ProviderEventIngestor from "./ProviderEventIngestor.ts";
import * as ProviderUsageAnalytics from "../telemetry/ProviderUsageAnalytics.ts";
import * as ThreadCommandExecutor from "./ThreadCommandExecutor.ts";
Expand Down Expand Up @@ -262,14 +264,17 @@ it.effect("refreshes the failing instance snapshot after a committed usage_limit
},
snapshot: {
refreshUsage: Deferred.succeed(refreshed, undefined).pipe(Effect.asVoid),
getSnapshot: Effect.die("unused"),
getSnapshot: Effect.sync(() => own as ServerProvider),
refresh: Effect.die("unused"),
resolveMaintenance: () => Effect.die("unused"),
applyUsageLimits: () => Effect.die("unused"),
streamChanges: Stream.empty,
},
};
let own: Partial<ServerProvider> | undefined;
const lookups: ProviderInstanceId[] = [];
const hubLookups: string[] = [];
const hubResetAt = "2033-05-19T07:20:00.000Z";
const instances = Layer.mock(ProviderInstanceRegistry.ProviderInstanceRegistry)({
getInstance: (id) =>
Effect.sync(() => {
Expand All @@ -283,7 +288,36 @@ it.effect("refreshes the failing instance snapshot after a committed usage_limit
const ingestor = yield* ProviderEventIngestor.ProviderEventIngestorV2;
const idAllocator = yield* IdAllocator.IdAllocatorV2;
const threadEvent = yield* threadCreatedEvent(now);
yield* eventSink.write({ events: [threadEvent] });
// The run started on an earlier model than the thread's current selection.
const runId = RunId.make("run:usage-limit");
yield* eventSink.write({
events: [
threadEvent,
{
id: yield* idAllocator.allocate.event({ threadId: threadEvent.threadId }),
type: "run.created",
threadId: threadEvent.threadId,
occurredAt: now,
payload: {
id: runId,
threadId: threadEvent.threadId,
ordinal: 1,
providerInstanceId: modelSelection.instanceId,
modelSelection: { ...modelSelection, model: "gpt-run-model" },
providerThreadId: null,
userMessageId: MessageId.make("message:usage-limit"),
rootNodeId: null,
activeAttemptId: null,
status: "running",
requestedAt: now,
startedAt: now,
completedAt: null,
checkpointId: null,
contextHandoffId: null,
},
},
],
});
const input = {
providerSessionId: yield* idAllocator.allocate.providerSession({
providerInstanceId: instance.instanceId,
Expand All @@ -292,9 +326,40 @@ it.effect("refreshes the failing instance snapshot after a committed usage_limit
providerInstanceId: instance.instanceId,
threadId: threadEvent.threadId,
};
for (const failureClass of ["validation_error", "usage_limit"] as const) {
const unread = { checkedAt: now.toString(), windows: [] };
const cases = [
{ failureClass: "validation_error", provider: undefined, resetAt: undefined },
{
failureClass: "usage_limit",
provider: {
auth: { status: "authenticated" },
usageLimits: { ...unread, unavailable: { reason: "probeFailed" } },
},
resetAt: undefined,
},
{
failureClass: "usage_limit",
provider: {
auth: { status: "authenticated", type: "customEndpoint" },
usageLimits: { ...unread, unavailable: { reason: "probeFailed" } },
},
resetAt: hubResetAt,
runId,
},
{
failureClass: "usage_limit",
provider: {
auth: { status: "unknown" },
usageLimits: { ...unread, unavailable: { reason: "unsupported" } },
},
resetAt: hubResetAt,
},
] as const;
for (const [index, { failureClass, provider, resetAt, ...run }] of cases.entries()) {
own = provider;
const stored = yield* ingestor.ingestNormalized({
...input,
...run,
event: {
type: "turn.terminal",
driver: instance.driverKind,
Expand All @@ -304,7 +369,7 @@ it.effect("refreshes the failing instance snapshot after a committed usage_limit
}),
providerTurnId: idAllocator.derive.providerTurn({
driver: instance.driverKind,
nativeTurnId: failureClass,
nativeTurnId: String(index),
}),
runOrdinal: 1,
failureItemOrdinal: 1,
Expand All @@ -318,10 +383,31 @@ it.effect("refreshes the failing instance snapshot after a committed usage_limit
});
assert.equal(stored.length, 1);
if (failureClass === "validation_error") assert.deepEqual(lookups, []);
// regression: #26574 a proxied CLI drops the reset, so the hub's pool refill fills it
const item = stored[0]?.event;
assert.equal(
item?.type === "turn-item.updated" && item.payload.type === "error"
? item.payload.failure.resetAt
: undefined,
resetAt,
);
}
assert.deepEqual(lookups, [instance.instanceId]);
assert.deepEqual(lookups, Array(6).fill(instance.instanceId));
assert.deepEqual(hubLookups, [
`${instance.driverKind}:gpt-run-model`,
`${instance.driverKind}:null`,
]);
yield* Deferred.await(refreshed);
}).pipe(Effect.provide(makeLayerTest(layerTestEventSink, instances)));
}).pipe(
Effect.provide(makeLayerTest(layerTestEventSink, instances)),
Effect.provideService(UsageLimitResetFallback, {
resetAt: (driver, model) =>
Effect.sync(() => {
hubLookups.push(`${driver}:${model}`);
return hubResetAt;
}),
}),
);
}),
),
);
Expand Down
35 changes: 34 additions & 1 deletion apps/server/src/orchestration-v2/ProviderEventIngestor.ts
Original file line number Diff line number Diff line change
Expand Up @@ -27,12 +27,14 @@ import * as SqlError from "effect/sql/SqlError";
import { AgentScope } from "@t3tools/shared/AgentScope";
import { isTemporaryWorktreeBranch } from "@t3tools/shared/git";
import { modelSelectionsEqual } from "@t3tools/shared/model";
import { hubRoutedModel, isProxied } from "@t3tools/shared/usageLimits";
import {
ProviderTurnAnalytics,
type ProviderTurnAnalyticsContext,
} from "../telemetry/ProviderUsageAnalytics.ts";
import * as GitVcsDriver from "../vcs/GitVcsDriver.ts";
import * as ProviderInstanceRegistry from "../provider/ProviderInstanceRegistry.ts";
import { UsageLimitResetFallback } from "../usage/UsageLimitSources.ts";
import * as EventSink from "./EventSink.ts";
import * as ProjectionStore from "./ProjectionStore.ts";
import * as IdAllocator from "@t3tools/provider-core/server/IdAllocator";
Expand Down Expand Up @@ -255,6 +257,7 @@ export const layer: Layer.Layer<
const analytics = yield* ProviderTurnAnalytics;
const agentScope = yield* AgentScope;
const instances = yield* ProviderInstanceRegistry.ProviderInstanceRegistry;
const usageLimitResets = yield* UsageLimitResetFallback;
const scope = yield* Effect.scope;
const completedTurnAnalytics = new Set<string>();

Expand Down Expand Up @@ -627,9 +630,39 @@ export const layer: Layer.Layer<
return dismissed;
}
const occurredAt = yield* DateTime.now;
const failure = (yield* agentScope.oomKilled(input.threadId))
const reported = (yield* agentScope.oomKilled(input.threadId))
? { ...input.event.failure, message: "Killed: out of memory.", code: "oom_kill" }
: input.event.failure;
// A proxy's CLI can drop the reset time; its hub still knows when the pool refills.
const provider =
reported.class === "usage_limit" && reported.resetAt == null
? yield* Effect.flatMap(
instances.getInstance(input.providerInstanceId),
(instance) => instance?.snapshot.getSnapshot ?? Effect.undefined,
)
: undefined;
// A direct login whose own probe failed is not served by the hub.
const proxied = provider !== undefined && isProxied(provider);
// The failing run's model, which may differ from a selection changed mid-turn.
const model =
provider !== undefined && proxied
? yield* Effect.gen(function* () {
const { runs } =
input.runId === undefined
? { runs: [] }
: yield* projections.getThreadRecords(input.threadId, ["runs"], {
runIds: [input.runId],
});
const selection =
runs[0]?.modelSelection ??
(yield* projections.getThread(input.threadId)).modelSelection;
return hubRoutedModel(provider, selection.model);
}).pipe(Effect.orElseSucceed(() => null))
: null;
const hubResetAt = proxied
? yield* usageLimitResets.resetAt(input.event.driver, model)
: null;
const failure = hubResetAt === null ? reported : { ...reported, resetAt: hubResetAt };
return [
...dismissed,
yield* makeDomainEvent(input, {
Expand Down
6 changes: 4 additions & 2 deletions apps/server/src/provider/CodexProvider.ts
Original file line number Diff line number Diff line change
Expand Up @@ -573,7 +573,7 @@ const probeCodexAppServerProvider = Effect.fn("probeCodexAppServerProvider")(fun
),
// Usage is an enrichment: a failure or a slow answer degrades to "no
// usage this probe" rather than costing the account and models.
input.skipNativeUsage
input.skipNativeUsage || !accountResponse.account
? Effect.undefined
: withCodexDiscoveryDeadline(
client.request("account/rateLimits/read", null),
Expand Down Expand Up @@ -842,7 +842,9 @@ export const checkCodexProviderStatus = Effect.fn("checkCodexProviderStatus")(fu
? { status: "ready" as const, auth: managedAuth, message: undefined }
: accountProbeStatus(snapshot);
const usageLimits =
snapshot.account.account?.type === "apiKey"
// A custom model provider has no subscription for Codex to read.
snapshot.account.account?.type === "apiKey" ||
(!managedAuth && !snapshot.account.account && !snapshot.account.requiresOpenaiAuth)
? makeUnavailableUsageLimits({ checkedAt, reason: "unsupported" })
: snapshot.rateLimits === undefined || "failure" in snapshot.rateLimits
? makeUnavailableUsageLimits({
Expand Down
2 changes: 2 additions & 0 deletions apps/server/src/provider/ProviderRegistry.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -478,6 +478,8 @@ it.layer(

assert.strictEqual(status.status, "ready");
assert.strictEqual(status.auth.status, "unknown");
// regression: #11873 a custom provider has no subscription to read, which is not a failure
assert.deepStrictEqual(status.usageLimits?.unavailable, { reason: "unsupported" });
}),
);

Expand Down
Loading
Loading