feat(routing): soonest-reset credential selector (MASTRA-765)

config.example.yaml

@@ -126,7 +126,11 @@ access:
 
 # Routing strategy for selecting credentials when multiple match.
 routing:
-  strategy: "round-robin" # round-robin (default), weighted-round-robin, fill-first
+  strategy: "round-robin" # round-robin (default), weighted-round-robin, fill-first, soonest-reset
+  # soonest-reset prefers the credential whose weekly (7-day) quota window resets soonest, so
+  # quota is used before it expires. Reset times are learned from Claude/Codex response headers.
+  # A credential with an exhausted window is skipped until that window resets. An unknown Claude
+  # reset is tried first; an unknown Codex reset last. Other providers behave like fill-first.
   # weighted-round-robin uses each credential's integer weight (default 1, maximum 1,000,000).
   # Non-positive weights exclude the credential while this strategy is active.
   # For OAuth/file credentials, add a top-level numeric "weight" field to the auth JSON.
@@ -182,7 +186,10 @@ routing:
     disable-cooling: false
 
     # When true, persist per-auth cooldown status as .cds files next to auth files.
-    # Default is false; when false, cooldown status is kept in memory only.
+    # The latest observed quota (rate-limit headers from Claude/Codex responses) is saved
+    # with it on every observation, so quota-aware routing such as soonest-reset and the
+    # management quota view keep their data across restarts.
+    # Default is false; when false, cooldown status and observed quota are kept in memory only.
     save-cooldown-status: false
 
     # Cooldown duration in seconds for transient upstream errors (408/500/502/503/504/520-526).

e2e/hermetic/mock-upstream.mjs

@@ -1,9 +1,12 @@
 // Mock Codex and Claude upstream. A credential whose key ends in `-limited` answers Codex
 // 429 usage_limit_reached; every other request gets a complete SSE stream (or a Claude JSON
 // message when the request does not stream). GET /_mock/requests returns what it received.
+// POST /_mock/script takes `{ [key]: { headers?, limited? } }`: `headers` replaces that
+// credential's quota headers and `limited: true` makes its Codex requests answer 429.
 import { createServer } from 'node:http';
 
 const requests = [];
+const scripts = new Map();
 const resetAt = () => Math.floor(Date.now() / 1000) + 3600;
 
 const sse = (res, headers, events) => {
@@ -13,7 +16,7 @@ const sse = (res, headers, events) => {
 };
 
 const codex = (res, key, body) => {
-  if (key.endsWith('-limited')) {
+  if (key.endsWith('-limited') || scripts.get(key)?.limited) {
     res.writeHead(429, { 'content-type': 'application/json' });
     const error = { type: 'usage_limit_reached', message: 'You have hit your usage limit.' };
     res.end(JSON.stringify({ error: { ...error, resets_at: resetAt(), resets_in_seconds: 3600 } }));
@@ -30,7 +33,7 @@ const codex = (res, key, body) => {
   const usage = { input_tokens: 5, output_tokens: 3, total_tokens: 8 };
   sse(
     res,
-    {
+    scripts.get(key)?.headers ?? {
       'x-codex-primary-used-percent': '10',
       'x-codex-primary-window-minutes': '300',
       'x-codex-primary-reset-at': String(resetAt()),
@@ -53,8 +56,8 @@ const codex = (res, key, body) => {
   );
 };
 
-const claude = (res, body) => {
-  const headers = {
+const claude = (res, key, body) => {
+  const headers = scripts.get(key)?.headers ?? {
     'anthropic-ratelimit-unified-status': 'allowed',
     'anthropic-ratelimit-unified-5h-status': 'allowed',
     'anthropic-ratelimit-unified-5h-utilization': '0.1',
@@ -104,9 +107,14 @@ export const startMock = () =>
         'Bearer ',
         '',
       );
+      if (req.url === '/_mock/script') {
+        for (const [scripted, script] of Object.entries(body)) scripts.set(scripted, script);
+        res.writeHead(204).end();
+        return;
+      }
       requests.push({ path: req.url, headers: req.headers, body, key });
       if (req.url === '/v1/responses') codex(res, key, body);
-      else claude(res, body);
+      else claude(res, key, body);
     });
     server.listen(0, '127.0.0.1', () => resolve(`http://127.0.0.1:${server.address().port}`));
   });

e2e/hermetic/start.mjs

@@ -12,28 +12,42 @@ writeFileSync('.e2e/mock.json', JSON.stringify({ url }));
 
 const dir = mkdtempSync(join(tmpdir(), 'cliproxy-e2e-'));
 mkdirSync(join(dir, 'auths'));
-const codex = (key, model, priority = 0) => ({
+const models = (names) => [names].flat().map((name) => ({ name }));
+const codex = (key, names, priority = 0) => ({
   'api-key': key,
   'base-url': `${url}/v1`,
   priority,
-  models: [{ name: model }],
+  models: models(names),
 });
-// YAML accepts JSON. The failover and cooldown tests own their models and credentials;
-// failover-a has the higher priority, so the proxy tries it first.
+const claude = (key, names) => ({ 'api-key': key, 'base-url': url, models: models(names) });
+// A selector scenario's credentials share its model; `<model>-<credential>` reaches one of them.
+const scenario = (make, model, credentials) =>
+  credentials.map((name) => make(`${model}-${name}`, [model, `${model}-${name}`]));
+// YAML accepts JSON. The failover, cooldown and selector tests own their models and
+// credentials; failover-a has the higher priority, so the proxy tries it first.
 const config = {
   host: '127.0.0.1',
   port: Number(process.env.PORT),
   'auth-dir': join(dir, 'auths'),
   'api-keys': ['e2e-client-key'],
+  routing: { strategy: 'soonest-reset', 'session-affinity': true },
+  'save-cooldown-status': true,
   'codex-api-key': [
     codex('codex-main', 'gpt-5.6-sol'),
     codex('failover-a-limited', 'e2e-failover', 1),
     codex('failover-b', 'e2e-failover'),
     codex('cooled-a-limited', 'e2e-cooled'),
     codex('cooled-b-limited', 'e2e-cooled'),
+    ...scenario(codex, 'e2e-s1', ['a', 'b', 'c']),
+    ...scenario(codex, 'e2e-s2', ['a', 'b']),
+    ...scenario(codex, 'e2e-s5', ['known', 'unknown']),
+    ...scenario(codex, 'e2e-s7', ['a', 'b']),
+    ...scenario(codex, 'e2e-s8', ['a', 'b']),
   ],
   'claude-api-key': [
-    { 'api-key': 'claude-main', 'base-url': url, models: [{ name: 'claude-sonnet-4-6' }] },
+    claude('claude-main', 'claude-sonnet-4-6'),
+    ...scenario(claude, 'e2e-s3', ['a', 'b']),
+    ...scenario(claude, 'e2e-s5c', ['x', 'y', 'z']),
   ],
 };
 writeFileSync(join(dir, 'config.yaml'), JSON.stringify(config, null, 2));

e2e/lib/proxy.ts

@@ -30,3 +30,14 @@ export async function upstreamRequests(marker: string): Promise<UpstreamRequest[
   const all: UpstreamRequest[] = await (await fetch(`${mockUrl()}/_mock/requests`)).json();
   return all.filter((request) => JSON.stringify(request.body).includes(marker));
 }
+
+export type Script = { headers?: Record<string, string>; limited?: boolean };
+
+/** Sets what the mock answers for each credential key (hermetic/mock-upstream.mjs). */
+export async function script(scripts: Record<string, Script>): Promise<void> {
+  const response = await fetch(`${mockUrl()}/_mock/script`, {
+    method: 'POST',
+    body: JSON.stringify(scripts),
+  });
+  if (!response.ok) throw new Error(`mock script failed: ${response.status}`);
+}

e2e/tests/failover.e2e.ts

@@ -41,7 +41,6 @@ test(
     expect((await response.json()).error).toMatchObject({
       code: 'model_cooldown',
       model: 'e2e-cooled',
-      provider: 'codex',
     });
     expect(await upstreamRequests(marker)).toEqual([]);
   },

e2e/tests/models.e2e.ts

@@ -17,7 +17,9 @@ test(
     });
     expect(response.status).toBe(200);
     const { data } = await response.json();
-    expect(data.map((model: { id: string }) => model.id).sort()).toEqual([
+    const ids: string[] = data.map((model: { id: string }) => model.id);
+    // Selector scenarios (tests/selector.e2e.ts) add their own e2e-s* models.
+    expect(ids.filter((id) => !id.startsWith('e2e-s')).sort()).toEqual([
       'claude-sonnet-4-6',
       'e2e-cooled',
       'e2e-failover',

e2e/tests/selector.e2e.ts

@@ -0,0 +1,169 @@
+// Soonest-reset selector scenarios (MASTRA-765, MASTRA-768). Each scenario owns its model and
+// credentials (hermetic/start.mjs): credential `<model>-<name>` serves `<model>` and its own
+// `<model>-<name>`, which seeds that credential's quota observation. Exhaustion is seeded with
+// 200 responses carrying exhausted headers, so the selector's header rule decides, not cooldown.
+// Several new sessions must land on one credential, which round-robin cannot do by chance.
+import { randomUUID } from 'node:crypto';
+import { expect, test } from 'e2e';
+import { clientKey, input, post, script, upstreamRequests } from '../lib/proxy.ts';
+
+const hermetic = { platforms: ['hermetic'] };
+const hour = 3600;
+const day = 24 * hour;
+const at = (seconds: number) => String(Math.floor(Date.now() / 1000) + seconds);
+
+// Codex Pro reports its weekly window as the primary one.
+const codexWeekly = (usedPercent: number, resetIn: number) => ({
+  'x-codex-primary-used-percent': String(usedPercent),
+  'x-codex-primary-window-minutes': '10080',
+  'x-codex-primary-reset-at': at(resetIn),
+});
+const claudeWindows = (
+  fiveHourUtilization: number,
+  fiveHourResetIn: number,
+  weekResetIn: number,
+) => ({
+  'anthropic-ratelimit-unified-status': 'allowed',
+  'anthropic-ratelimit-unified-5h-utilization': String(fiveHourUtilization),
+  'anthropic-ratelimit-unified-5h-reset': at(fiveHourResetIn),
+  'anthropic-ratelimit-unified-7d-utilization': '0.5',
+  'anthropic-ratelimit-unified-7d-reset': at(weekResetIn),
+});
+
+/** Sends one Responses request in `session`; returns the credential keys the mock saw. */
+async function responses(baseUrl: string | undefined, model: string, session = randomUUID()) {
+  const marker = randomUUID();
+  const response = await post(
+    baseUrl,
+    '/v1/responses',
+    { model, input: input(marker) },
+    { Session_id: session },
+  );
+  expect(response.status).toBe(200);
+  await response.text();
+  return (await upstreamRequests(marker)).map((request) => request.key);
+}
+
+/** Sends one Messages request in `session`; returns the credential keys the mock saw. */
+async function messages(baseUrl: string | undefined, model: string, session = randomUUID()) {
+  const marker = randomUUID();
+  const response = await post(
+    baseUrl,
+    '/v1/messages',
+    { model, max_tokens: 16, messages: [{ role: 'user', content: marker }] },
+    {
+      'x-api-key': clientKey,
+      'anthropic-version': '2023-06-01',
+      'X-Claude-Code-Session-Id': session,
+    },
+  );
+  expect(response.status).toBe(200);
+  await response.text();
+  return (await upstreamRequests(marker)).map((request) => request.key);
+}
+
+test('S1: new Codex sessions go to the soonest weekly reset', hermetic, async ({ app }) => {
+  await script({
+    'e2e-s1-a': { headers: codexWeekly(50, 6 * day) },
+    'e2e-s1-b': { headers: codexWeekly(50, 1 * day) },
+    'e2e-s1-c': { headers: codexWeekly(50, 3 * day) },
+  });
+  for (const name of ['a', 'b', 'c']) await responses(app.baseUrl, `e2e-s1-${name}`);
+  for (let i = 0; i < 3; i++) expect(await responses(app.baseUrl, 'e2e-s1')).toEqual(['e2e-s1-b']);
+});
+
+test(
+  'S2: a small remainder resetting soon goes before a fresh late window, then fails over',
+  hermetic,
+  async ({ app }) => {
+    await script({
+      'e2e-s2-a': { headers: codexWeekly(98, 6 * hour) },
+      'e2e-s2-b': { headers: codexWeekly(0, 5 * day) },
+    });
+    for (const name of ['a', 'b']) await responses(app.baseUrl, `e2e-s2-${name}`);
+    for (let i = 0; i < 2; i++) {
+      expect(await responses(app.baseUrl, 'e2e-s2')).toEqual(['e2e-s2-a']);
+    }
+    await script({ 'e2e-s2-a': { limited: true } });
+    expect(await responses(app.baseUrl, 'e2e-s2')).toEqual(['e2e-s2-a', 'e2e-s2-b']);
+  },
+);
+
+test(
+  'S3: an exhausted Claude 5-hour window is skipped until it resets',
+  hermetic,
+  async ({ app }) => {
+    const fiveHourReset = 4;
+    await script({
+      'e2e-s3-a': { headers: claudeWindows(1, fiveHourReset, 1 * day) },
+      'e2e-s3-b': { headers: claudeWindows(0.1, 4 * hour, 3 * day) },
+    });
+    for (const name of ['a', 'b']) await messages(app.baseUrl, `e2e-s3-${name}`);
+    for (let i = 0; i < 2; i++) expect(await messages(app.baseUrl, 'e2e-s3')).toEqual(['e2e-s3-b']);
+    await new Promise((resolve) => setTimeout(resolve, (fiveHourReset + 1) * 1000));
+    expect(await messages(app.baseUrl, 'e2e-s3')).toEqual(['e2e-s3-a']);
+  },
+);
+
+test(
+  'S5: an unknown Codex reset sorts last and an unknown Claude reset sorts first',
+  hermetic,
+  async ({ app }) => {
+    await script({
+      'e2e-s5-known': { headers: codexWeekly(50, 6 * day) },
+      'e2e-s5-unknown': { headers: {} },
+      'e2e-s5c-x': { headers: claudeWindows(0.1, 4 * hour, 6 * day) },
+      'e2e-s5c-y': { headers: claudeWindows(0.1, 4 * hour, 1 * day) },
+      'e2e-s5c-z': { headers: claudeWindows(0.1, 4 * hour, 3 * day) },
+    });
+    await responses(app.baseUrl, 'e2e-s5-known');
+    for (let i = 0; i < 2; i++) {
+      expect(await responses(app.baseUrl, 'e2e-s5')).toEqual(['e2e-s5-known']);
+    }
+
+    for (const name of ['y', 'z']) await messages(app.baseUrl, `e2e-s5c-${name}`);
+    expect(await messages(app.baseUrl, 'e2e-s5c')).toEqual(['e2e-s5c-x']);
+    // x's first response taught its late reset, so y (soonest) now leads.
+    for (let i = 0; i < 2; i++)
+      expect(await messages(app.baseUrl, 'e2e-s5c')).toEqual(['e2e-s5c-y']);
+  },
+);
+
+test(
+  'S7 (regression): a bound session stays on its credential when another becomes better',
+  hermetic,
+  async ({ app }) => {
+    await script({
+      'e2e-s7-a': { headers: codexWeekly(50, 3 * day) },
+      'e2e-s7-b': { headers: codexWeekly(50, 3 * day) },
+    });
+    const session = randomUUID();
+    const [bound] = await responses(app.baseUrl, 'e2e-s7', session);
+    const other = bound === 'e2e-s7-a' ? 'e2e-s7-b' : 'e2e-s7-a';
+    await script({
+      [bound]: { headers: codexWeekly(50, 6 * day) },
+      [other]: { headers: codexWeekly(50, 1 * hour) },
+    });
+    await responses(app.baseUrl, other);
+    for (let i = 0; i < 3; i++) {
+      expect(await responses(app.baseUrl, 'e2e-s7', session)).toEqual([bound]);
+    }
+  },
+);
+
+test(
+  'S8 (regression): a burst of new sessions at exhaustion completes without client errors',
+  hermetic,
+  async ({ app }) => {
+    await script({
+      'e2e-s8-a': { headers: codexWeekly(99, 1 * hour) },
+      'e2e-s8-b': { headers: codexWeekly(0, 5 * day) },
+    });
+    for (const name of ['a', 'b']) await responses(app.baseUrl, `e2e-s8-${name}`);
+    await script({ 'e2e-s8-a': { limited: true } });
+    const served = await Promise.all(
+      Array.from({ length: 6 }, () => responses(app.baseUrl, 'e2e-s8')),
+    );
+    for (const keys of served) expect(keys.at(-1)).toBe('e2e-s8-b');
+  },
+);

internal/clienterror/client_error.go

@@ -25,6 +25,10 @@ var requestFaultCodes = map[string]struct{}{
 	"unsupported_value":           {},
 	"invalid_request_error":       {},
 	"previous_response_not_found": {},
+	// Anthropic: server-side thread state for previous_message_id is missing on
+	// this account. Another credential cannot have it either; the client recovers
+	// by replaying the conversation as a new thread.
+	"thread_not_found": {},
 }
 
 var requestFaultTypes = map[string]struct{}{
@@ -170,7 +174,7 @@ func hasRequestFaultBody(err error) bool {
 	if body == "" || !json.Valid([]byte(body)) {
 		return false
 	}
-	for _, path := range []string{"error.code", "code", "response.error.code", "body.error.code"} {
+	for _, path := range []string{"error.code", "code", "response.error.code", "body.error.code", "error.details.error_code"} {
 		code := strings.ToLower(strings.TrimSpace(gjson.Get(body, path).String()))
 		if _, ok := requestFaultCodes[code]; ok {
 			return true

internal/clienterror/client_error_test.go

@@ -140,6 +140,18 @@ func TestIsRequestFaultStructuredIdentifiers(t *testing.T) {
 	}
 }
 
+func TestIsRequestFaultClaudeThreadNotFound(t *testing.T) {
+	err := errors.New(`{"type":"error","error":{"type":"not_found_error","message":"No thread state was found for the requested ` + "`previous_message_id`" + `.","details":{"error_code":"thread_not_found"}}}`)
+	if !IsRequestFault(http.StatusNotFound, err) {
+		t.Fatal("thread_not_found was not classified as a request fault")
+	}
+
+	other := errors.New(`{"type":"error","error":{"type":"not_found_error","message":"missing","details":{"error_code":"something_else"}}}`)
+	if IsRequestFault(http.StatusNotFound, other) {
+		t.Fatal("unrelated not_found_error was classified as a request fault")
+	}
+}
+
 func TestIsRequestFault(t *testing.T) {
 	tests := []struct {
 		name   string

internal/config/config.go

@@ -72,7 +72,8 @@ type Config struct {
 	// DisableCooling disables auth/model cooldown scheduling when true unless a credential or provider overrides it.
 	DisableCooling bool `yaml:"disable-cooling" json:"disable-cooling"`
 
-	// SaveCooldownStatus persists runtime cooldown status next to auth files when true.
+	// SaveCooldownStatus persists runtime cooldown status, and the latest observed quota
+	// snapshot, next to auth files when true.
 	SaveCooldownStatus bool `yaml:"save-cooldown-status" json:"save-cooldown-status"`
 
 	// TransientErrorCooldownSeconds controls cooldowns for transient upstream errors (408/500/502/503/504/520-526).

internal/config/config_types.go

@@ -351,7 +351,7 @@ type QuotaExceeded struct {
 // RoutingConfig configures how credentials are selected for requests.
 type RoutingConfig struct {
 	// Strategy selects the credential selection strategy.
-	// Supported values: "round-robin" (default), "weighted-round-robin", "fill-first".
+	// Supported values: "round-robin" (default), "weighted-round-robin", "fill-first", "soonest-reset".
 	Strategy string `yaml:"strategy,omitempty" json:"strategy,omitempty"`
 
 	// SessionAffinity enables universal session-sticky routing for all clients.