fix(gateway): Read the model from any body the gateway accepts

CHANGELOG.md

@@ -8,6 +8,7 @@ This project uses selective package publishing. Each release entry lists the pub
 
 ### Fixed
 
+- CLI gateway: `@preset/<slug>` now applies to every accepted request body, however large, and an alias that cannot be resolved is refused instead of forwarded. The gateway only read a body's model from the first 4 MiB, so a larger one (a chat with a few MiB image attached, which Open WebUI inlines as base64) reached the buyer with the literal `@preset/<slug>` as its model: the buyer answered `502 model_not_found` ("No policy-allowed peer currently serves model \"@preset/…\"") even though sellers within the caps existed, and the request log recorded no model. The preset's price caps were never applied to such requests, so the allow list and the accounting were wrong in the same way. The gateway now reads the model exactly as the buyer does (`service` before `model`, the whole body, multipart forms included), so the two cannot disagree about which model is bought.
 - CLI buyer: when a seller advertises the same model under several aliases, routing sends one candidate for that seller and uses the cheapest alias price instead of submitting duplicate destinations to the router.
 - Gateway console: video jobs no longer fill Logs with "not recorded" and "Free" rows. A video is charged once, on the retrieve that downloads the finished file, but the gateway expected spend on every create and status poll that named a model. After 10 minutes those rows read "not recorded" (as if the buyer had lost spend), and creates read "Free" because video is priced per second, not per token. Now a create shows its job's charge once the video is delivered ("on delivery" until then), a status poll shows "—", and only the delivery can be pending or not recorded. `GET /console/api/requests` adds `video` (`action`: `create`, `poll` or `delivery`, `jobId`, and a create's `charged`) to video requests. Totals are unchanged; they already counted each job once.
 - CLI gateway: a request the client cancels before the buyer answers (for example a client-side timeout) is now logged as `499 client_closed_request` instead of `502 buyer_unavailable`, so the console no longer reports a buyer outage when the buyer was still working on the request.

apps/cli/src/gateway/gateway-policy.test.ts

@@ -279,6 +279,74 @@ test('@preset/<slug> rewrites the request and narrows the policy', async () => {
   }
 })
 
+test('a body too large to be sniffed still runs its preset and keeps its caps', async () => {
+  const ctx = await setup()
+  try {
+    // Caps are only worth anything if the preset is applied: an unresolved
+    // `@preset/` alias must never reach the buyer as a model name.
+    const { secret, key } = ctx.store.createKey({ label: 'k', limits: NO_LIMITS, expiresAt: null, routingPolicy: { allowedModels: ['fast-model'] } })
+    ctx.store.createPreset({
+      slug: 'cap', name: 'Cap', workspaceId: key.workspaceId, model: 'fast-model',
+      routingPolicy: { maxInputUsdPerMillion: 5_000, maxOutputUsdPerMillion: 10_000 }, systemPrompt: null, params: {},
+    })
+    const large = { model: '@preset/cap', messages: [{ role: 'user', content: 'x'.repeat(5 * 1024 * 1024) }] }
+    const response = await ctx.send(secret, '/v1/chat/completions', large)
+    assert.equal(response.status, 200)
+    const forwarded = ctx.buyer.captured[0]!
+    assert.equal(JSON.parse(forwarded.body).model, 'fast-model')
+    assert.ok(!forwarded.body.includes('@preset/'), 'the alias never reaches the buyer')
+    const policy = decodePolicyHeader(forwarded.headers[ROUTING_POLICY_HEADER] as string)
+    assert.ok(policy, 'the preset cap reaches the buyer')
+    assert.equal(policy.maxInputUsdPerMillion, 5_000)
+    assert.equal(policy.maxOutputUsdPerMillion, 10_000)
+    assert.deepEqual(policy.allowedModels, ['fast-model'])
+    assert.equal(ctx.store.listRequests({ keyId: key.id })[0]!.model, 'fast-model', 'the ledger records the model that is bought')
+
+    const unknown = await ctx.send(secret, '/v1/chat/completions', { model: '@preset/nope', messages: [{ role: 'user', content: 'x'.repeat(5 * 1024 * 1024) }] })
+    assert.equal(unknown.status, 400)
+    assert.equal(unknown.body.error.code, 'preset_not_found')
+    assert.equal(ctx.buyer.captured.length, 1, 'nothing is forwarded for an unknown preset')
+  } finally {
+    await ctx.close()
+  }
+})
+
+test('a large body never forwards an unresolved @preset/ alias to the buyer', async () => {
+  const ctx = await setup()
+  try {
+    // The production symptom: with nothing else to stop it, an alias the
+    // gateway failed to resolve went out as a model name and the buyer
+    // answered `model_not_found` — the caps looked broken but were never read.
+    const { secret } = ctx.store.createKey({ label: 'open', limits: NO_LIMITS, expiresAt: null })
+    ctx.store.createPreset({ slug: 'cap', name: 'Cap', workspaceId: null, model: 'fast-model', routingPolicy: null, systemPrompt: null, params: {} })
+    const large = { model: '@preset/cap', messages: [{ role: 'user', content: 'x'.repeat(5 * 1024 * 1024) }] }
+    assert.equal((await ctx.send(secret, '/v1/chat/completions', large)).status, 200)
+    const forwarded = ctx.buyer.captured[0]!
+    assert.equal(JSON.parse(forwarded.body).model, 'fast-model')
+    assert.ok(!forwarded.body.includes('@preset/'), 'the alias never reaches the buyer as a model')
+
+    const unroutable = await ctx.send(secret, '/v1/chat/completions', { model: '@preset/nope', messages: [{ role: 'user', content: 'x'.repeat(5 * 1024 * 1024) }] })
+    assert.equal(unroutable.status, 400, 'an unresolvable alias is refused, never forwarded')
+    assert.equal(ctx.buyer.captured.length, 1)
+  } finally {
+    await ctx.close()
+  }
+})
+
+test('a body too large to be sniffed is still allow-listed and accounted by its model', async () => {
+  const ctx = await setup()
+  try {
+    const { secret, key } = ctx.store.createKey({ label: 'k', limits: NO_LIMITS, expiresAt: null, routingPolicy: { allowedModels: ['good-model'] } })
+    const withContent = (body: Record<string, unknown>) => ({ ...body, messages: [{ role: 'user', content: 'x'.repeat(5 * 1024 * 1024) }] })
+    assert.equal((await ctx.send(secret, '/v1/chat/completions', withContent({ model: 'good-model' }))).status, 200)
+    assert.equal((await ctx.send(secret, '/v1/chat/completions', withContent({ model: 'other-model' }))).status, 403, 'a large body cannot smuggle a disallowed model past the allow list')
+    assert.equal((await ctx.send(secret, '/v1/chat/completions', withContent({ model: 'good-model', service: 'other-model' }))).status, 403, 'nor past it through a service')
+    assert.equal(ctx.store.listRequests({ keyId: key.id }).map((row) => row.model).filter(Boolean).join(','), 'good-model')
+  } finally {
+    await ctx.close()
+  }
+})
+
 test('a request is admitted only while key, member and workspace budgets all have room', async () => {
   const ctx = await setup({ costUsdc: 400_000 })
   try {

apps/cli/src/gateway/policy-resolver.test.ts

@@ -8,7 +8,7 @@ import { GATEWAY_CONTROL_HEADER, GATEWAY_CONTROL_SECRET_ENV, ROUTING_POLICY_HEAD
 import { buyerFetch, loadOrCreateControlSecret, recordGatewayBuyerPort, runningGatewayBuyerPort } from './buyer-control.js'
 import { OtlpExporter, tracesUrl } from './observability.js'
 import { GATEWAY_ROUTING_SETTING, checkPolicyInput, isEmptyPolicy, policyInputProblem, resolvePolicy } from './policy-resolver.js'
-import { applyPreset, effectiveRequestModel, extractEndUser, sniffMultipartModel } from './request-shaping.js'
+import { applyPreset, effectiveRequestModel, extractEndUser } from './request-shaping.js'
 import { DEFAULT_WORKSPACE_ID, type PresetRecord } from './store.js'
 import { tempDataDir } from './console-api/test-support.js'
 
@@ -90,15 +90,24 @@ test('end user and multipart model are read from the request', () => {
   assert.equal(extractEndUser({}, undefined), null)
   const boundary = 'XyZ'
   const form = Buffer.from(`--${boundary}\r\nContent-Disposition: form-data; name="prompt"\r\n\r\nhello\r\n--${boundary}\r\nContent-Disposition: form-data; name="model"\r\n\r\nimage-model\r\n--${boundary}--\r\n`)
-  assert.equal(sniffMultipartModel(form, `multipart/form-data; boundary=${boundary}`), 'image-model')
-  assert.equal(sniffMultipartModel(form, 'application/json'), null)
   const multipart = `multipart/form-data; boundary=${boundary}`
   assert.equal(effectiveRequestModel(null, form, multipart), 'image-model')
+  assert.equal(effectiveRequestModel(null, form, 'application/json'), null, 'a JSON content type is not read as a form')
   const withService = Buffer.from(`--${boundary}\r\nContent-Disposition: form-data; name="service"\r\n\r\nother-image\r\n${form.toString('latin1')}`)
   assert.equal(effectiveRequestModel(null, withService, multipart), 'other-image', 'service wins, as the buyer reads it')
   assert.equal(effectiveRequestModel({ model: 'm', service: 's' }, Buffer.alloc(0), 'application/json'), 's')
   assert.equal(effectiveRequestModel({ model: ' m ' }, Buffer.alloc(0), 'application/json'), 'm')
   assert.equal(effectiveRequestModel({ model: 'm', service: 7 }, Buffer.alloc(0), 'application/json'), null)
+
+  // The buyer's parser reads the whole form and skips file parts, so the model
+  // is found wherever it sits: the gateway must not read the form differently.
+  const file = Buffer.alloc(5 * 1024 * 1024, 0x41)
+  const late = Buffer.concat([
+    Buffer.from(`--${boundary}\r\nContent-Disposition: form-data; name="image"; filename="big.png"\r\nContent-Type: image/png\r\n\r\n`),
+    file,
+    Buffer.from(`\r\n--${boundary}\r\nContent-Disposition: form-data; name="model"\r\n\r\nlate-model\r\n--${boundary}--\r\n`),
+  ])
+  assert.equal(effectiveRequestModel(null, late, multipart), 'late-model', 'a model past the first bytes of a large upload is still read')
 })
 
 test('a preset drops a client service so the buyer routes by the preset\'s model', () => {

apps/cli/src/gateway/request-shaping.ts

@@ -1,3 +1,4 @@
+import { parseMultipartFormFields } from '@antseed/api-adapter'
 import type { PresetRecord } from './store.js'
 
 export const PRESET_MODEL_PREFIX = '@preset/'
@@ -42,12 +43,15 @@ export function extractEndUser(body: Record<string, unknown> | null, header: str
   return value ? value.slice(0, MAX_END_USER_LENGTH) : null
 }
 
-/** A text field of a multipart form (image edits), without a full parser; `model` by default. */
-export function sniffMultipartModel(body: Buffer, contentType: string | undefined, field: 'model' | 'service' = 'model'): string | null {
-  if (!contentType?.toLowerCase().startsWith('multipart/form-data')) return null
-  const text = body.subarray(0, 4 * 1024 * 1024).toString('latin1')
-  const match = new RegExp(`content-disposition:\\s*form-data;\\s*name="${field}"[^\\r\\n]*\\r\\n(?:[^\\r\\n]+\\r\\n)*\\r\\n([^\\r\\n]*)\\r\\n`, 'i').exec(text)
-  return match ? Buffer.from(match[1]!, 'latin1').toString('utf8').slice(0, 200) : null
+/**
+ * A multipart form's text fields, read with the buyer's own parser
+ * (`parseMultipartFormFields`): the same `model`/`service` values, and the
+ * whole body, not just its start. Null unless the content type is multipart.
+ */
+function multipartFields(body: Buffer, contentType: string | undefined): Record<string, string> | null {
+  return contentType?.toLowerCase().startsWith('multipart/form-data')
+    ? parseMultipartFormFields(body, contentType)
+    : null
 }
 
 /**
@@ -56,10 +60,13 @@ export function sniffMultipartModel(body: Buffer, contentType: string | undefine
  * proxy/request-utils.ts): a `service` field wins over `model`, and one that
  * is not a usable name leaves the request without a model. The gateway's
  * model allow list and its accounting use this, not `model` alone.
+ *
+ * The body is read whole, however large: the buyer reads it whole too, so a
+ * body the gateway cannot read is one whose preset, allow list and accounting
+ * it would get wrong while the buyer still routes it.
  */
 export function effectiveRequestModel(parsed: Record<string, unknown> | null, body: Buffer, contentType: string | undefined): string | null {
-  const raw = parsed
-    ? parsed['service'] ?? parsed['model']
-    : sniffMultipartModel(body, contentType, 'service') ?? sniffMultipartModel(body, contentType)
+  const fields: Record<string, unknown> | null = parsed ?? multipartFields(body, contentType)
+  const raw = fields?.['service'] ?? fields?.['model']
   return typeof raw === 'string' && raw.trim() ? raw.trim() : null
 }

apps/cli/src/gateway/server.ts

@@ -52,9 +52,12 @@ const HOP_BY_HOP = new Set([
   'upgrade',
 ])
 
-/** Bodies are buffered to read the model for metrics; the buyer buffers them too. */
-const MAX_MODEL_SNIFF_BYTES = 4 * 1024 * 1024
-/** Largest request body the gateway accepts, matching the buyer's default upload limit. */
+/**
+ * Largest request body the gateway accepts, matching the buyer's default
+ * upload limit. Every accepted body is parsed to read its model, the way the
+ * buyer reads it: that model decides the preset, the allow list and the
+ * accounting, so the two must never disagree about it.
+ */
 const MAX_BODY_BYTES = 64 * 1024 * 1024
 /** How much of a failed response is read for its error code and message. */
 const MAX_ERROR_BODY_BYTES = 16 * 1024
@@ -836,9 +839,9 @@ function sendBodyTooLarge(res: http.ServerResponse): void {
   sendError(res, 413, 'invalid_request_error', 'request_too_large', `Request bodies are limited to ${MAX_BODY_BYTES / 1024 / 1024} MiB`)
 }
 
-/** A JSON object body, or null (empty, too large to inspect, not JSON, not an object). */
+/** A JSON object body, or null (empty, not JSON, not an object). */
 function parseJsonBody(body: Buffer, contentType: string | undefined): Record<string, unknown> | null {
-  if (body.length === 0 || body.length > MAX_MODEL_SNIFF_BYTES) return null
+  if (body.length === 0) return null
   if (contentType && !contentType.includes('json')) return null
   try {
     return jsonObject(JSON.parse(body.toString('utf8')))