fix(relay): send a failed direct fallback only once

mutations.toml

@@ -1233,6 +1233,22 @@ broad_command = ["bun", "scripts/run-mutation-bun.ts", "--broad", "packages/core
 broad_report = "tmp/mutations/core.xml"
 broad_id = "{name}"
 
+[[control]]
+id = "relay-direct-fallback-sends-once"
+guards = "A direct fallback that fails after it started must not be sent a second time by sendViaRelay's outer catch: its request may already have reached Anthropic."
+file = "packages/core/src/relay.ts"
+old = "    directFallbackStarted = true"
+new = "    directFallbackStarted = false // NON-VACUITY BREAK: forget that the direct fallback started, so a failure is sent again"
+test_file = "packages/core/src/tests/relay-authorization.test.ts"
+runner = "command"
+command = ["bun", "scripts/run-mutation-bun.ts", "packages/core/src/tests/relay-authorization.test.ts", "{test}"]
+test_count_pattern = "CKDEV executed {count} tests"
+expect_red = ["a direct fallback after a relay 5xx that fails is not sent again", "a direct fallback after a relay-only 401 that fails is not sent again", "a failed WebSocket relay whose HTTP relay fallback falls back direct and fails is not sent again"]
+expect_message = "error: expect(received).toBe(expected)"
+broad_command = ["bun", "scripts/run-mutation-bun.ts", "--broad", "packages/core", "tmp/mutations/core.xml"]
+broad_report = "tmp/mutations/core.xml"
+broad_id = "{name}"
+
 [[control]]
 id = "quota-call-pause-default-waits"
 guards = "Without an injected pause, QuotaManager waits on a real timer so physical quota GETs start at least one second apart."

packages/core/src/relay.ts

@@ -1867,6 +1867,15 @@ export async function sendViaRelay(options: {
     responseMode ??= requestResponseMode(bodyText)
     return responseMode
   }
+  // The direct fallback runs at most once per request. Once it has started,
+  // its request may already have reached Anthropic (a connection can reset
+  // after the body was sent), so its failure goes to the caller instead of
+  // sending the same request again.
+  let directFallbackStarted = false
+  const directFallback = () => {
+    directFallbackStarted = true
+    return fallback()
+  }
   try {
     let result: RelaySendResult
     const sendStart = perfNowMs()
@@ -1902,7 +1911,7 @@ export async function sendViaRelay(options: {
           payload,
           bodyText,
           requestMode,
-          fallback,
+          fallback: directFallback,
           signal: init?.signal,
           authorizeAttempt,
         })
@@ -1913,7 +1922,7 @@ export async function sendViaRelay(options: {
         payload,
         bodyText,
         requestMode,
-        fallback,
+        fallback: directFallback,
         signal: init?.signal,
         authorizeAttempt,
       })
@@ -1983,6 +1992,12 @@ export async function sendViaRelay(options: {
     // The upstream already answered this request; a direct fallback would
     // send it a second time.
     if (error instanceof RelayResponseMetadataError) throw error
+    if (directFallbackStarted) {
+      relayLog(
+        `direct fallback failed session=${shortAffinity(affinity)}; not sending it again code=relay_transport_error`,
+      )
+      throw error
+    }
     if (!config.fallbackToDirect) {
       relayLog(
         `relay failed session=${shortAffinity(affinity)} and fallbackToDirect=false code=relay_transport_error`,

packages/core/src/tests/relay-authorization.test.ts

@@ -173,6 +173,108 @@ test('a relay transport 401 without upstream provenance falls back direct withou
   }
 })
 
+// A direct fallback can fail after its request already reached Anthropic,
+// for example when the connection resets after the body was sent. Sending it
+// again would bill the same request twice, so its error goes to the caller.
+for (const [label, status] of [
+  ['a relay 5xx', 503],
+  ['a relay-only 401', 401],
+] as const) {
+  test(`a direct fallback after ${label} that fails is not sent again`, async () => {
+    const server = Bun.serve({
+      hostname: '127.0.0.1',
+      port: 0,
+      fetch: () => new Response('relay unavailable', { status }),
+    })
+    let direct = 0
+    try {
+      await expect(
+        sendViaRelay({
+          config: config(server.url.href),
+          input: 'https://api.anthropic.com/v1/messages',
+          init: { method: 'POST' },
+          headers: new Headers(),
+          body: body(`failed-fallback-${status}`),
+          affinity: randomUUID(),
+          fallback: async () => {
+            direct++
+            throw new Error('connection reset after the request was sent')
+          },
+          authorizeAttempt: async () => ({
+            headers: new Headers({ authorization: 'Bearer test-access' }),
+          }),
+        }),
+      ).rejects.toThrow('connection reset after the request was sent')
+      expect(direct).toBe(1)
+    } finally {
+      await server.stop(true)
+    }
+  })
+}
+
+test('a failed WebSocket relay whose HTTP relay fallback falls back direct and fails is not sent again', async () => {
+  // Refusing the upgrade fails the WebSocket; the HTTP relay then answers 503.
+  const server = Bun.serve({
+    hostname: '127.0.0.1',
+    port: 0,
+    fetch: () => new Response('relay unavailable', { status: 503 }),
+  })
+  let direct = 0
+  try {
+    await expect(
+      sendViaRelay({
+        config: config(server.url.href, 'websocket'),
+        input: 'https://api.anthropic.com/v1/messages',
+        init: { method: 'POST' },
+        headers: new Headers(),
+        body: body('failed-websocket-fallback'),
+        affinity: randomUUID(),
+        fallback: async () => {
+          direct++
+          throw new Error('connection reset after the request was sent')
+        },
+        authorizeAttempt: async () => ({
+          headers: new Headers({ authorization: 'Bearer test-access' }),
+        }),
+      }),
+    ).rejects.toThrow('connection reset after the request was sent')
+    expect(direct).toBe(1)
+  } finally {
+    server.stop(true)
+  }
+})
+
+test('a relay that fails before any direct fallback still falls back direct once', async () => {
+  // A stopped server leaves a port that refuses connections, so the relay
+  // POST itself rejects before the relay could forward anything.
+  const server = Bun.serve({
+    hostname: '127.0.0.1',
+    port: 0,
+    fetch: () => new Response('unused'),
+  })
+  const url = server.url.href
+  await server.stop(true)
+  let direct = 0
+  const response = await sendViaRelay({
+    config: config(url),
+    input: 'https://api.anthropic.com/v1/messages',
+    init: { method: 'POST' },
+    headers: new Headers(),
+    body: body('relay-unreachable'),
+    affinity: randomUUID(),
+    fallback: async () => {
+      direct++
+      return new Response('direct-ok', { status: 200 })
+    },
+    authorizeAttempt: async () => ({
+      headers: new Headers({ authorization: 'Bearer test-access' }),
+    }),
+  })
+  expect(response.status).toBe(200)
+  expect(await response.text()).toBe('direct-ok')
+  expect(direct).toBe(1)
+})
+
 test('optimistic WebSocket reconnect uses a fresh receipt without reporting its synthetic 200', async () => {
   const sent: Array<{
     id: string

packages/opencode/src/tests/index.test.ts

@@ -19265,31 +19265,56 @@ describe('auth.loader', () => {
         }),
       )
       const records: LogTestRecord[] = []
+      const affinity = 'quota-relay-direct-fallback'
+      // A marker in this request's body tells its own sends apart from any
+      // other request that reaches the mock. When an earlier test overruns
+      // Bun's timeouts, its body and its teardown keep running into this test:
+      // its own model request can arrive here, and its teardown can re-enable
+      // profile hydration, which sends GET /api/oauth/profile. Counting those
+      // would report a duplicate send that this request never made.
+      const probe = 'relay-direct-fallback-probe'
       let relay503Calls = 0
       let directCalls = 0
       __setLogTestSink((record) => records.push(record))
       globalThis.fetch = mock(
-        withNativeAdmission((input: string | URL | Request) => {
-          const url = extractUrl(input)
-          if (url === 'https://relay.example.test/') {
-            relay503Calls += 1
-            return Promise.resolve(
-              new Response('relay unavailable', { status: 503 }),
-            )
-          }
-          directCalls += 1
-          return Promise.resolve(
-            new Response('direct', { headers: quotaHeaders }),
-          )
-        }),
+        withNativeAdmission(
+          (input: string | URL | Request, init?: RequestInit) => {
+            const url = extractUrl(input)
+            const body = typeof init?.body === 'string' ? init.body : ''
+            if (url === 'https://relay.example.test/') {
+              let relayAffinity: unknown
+              try {
+                relayAffinity = JSON.parse(body).affinity
+              } catch {}
+              if (relayAffinity === affinity) relay503Calls += 1
+              return Promise.resolve(
+                new Response('relay unavailable', { status: 503 }),
+              )
+            }
+            if (
+              url.startsWith(MESSAGES_URL) &&
+              init?.method === 'POST' &&
+              body.includes(probe)
+            ) {
+              directCalls += 1
+              return Promise.resolve(
+                new Response('direct', { headers: quotaHeaders }),
+              )
+            }
+            return Promise.resolve(new Response('unrelated', { status: 404 }))
+          },
+        ),
       ) as unknown as typeof fetch
       const result = await loadFetch()
       setLogLevel('debug')
 
       try {
         const response = await result.fetch(MESSAGES_URL, {
-          ...EMPTY_POST,
-          headers: { 'x-session-affinity': 'quota-relay-direct-fallback' },
+          method: 'POST',
+          body: JSON.stringify({
+            messages: [{ role: 'user', content: probe }],
+          }),
+          headers: { 'x-session-affinity': affinity },
         })
         expect(await response.text()).toBe('direct')
         expect(relay503Calls).toBe(1)