fix(obs): log the requests refused before dispatch

crates/aisix-proxy/AGENTS.md

@@ -28,6 +28,29 @@ interval) so a model that is slow to its first token doesn't look like an
 abandoned connection to a proxy in front. Only for SSE: the same wrapper on an
 opaque binary passthrough (audio, images) corrupts it.
 
+## Every terminal path emits the access log — including the ones that give up early
+
+The access log and `record_request` are emitted **by the handler**, at the end of
+dispatch, because that is the only place that knows the provider, model and token
+counts. A path that returns before reaching that tail therefore logs nothing, and
+nothing errors: the caller gets a correct status while the gateway keeps no record
+of the request, which is indistinguishable from the request never arriving.
+
+Two shapes give up early, and both must answer through
+`reject::reject_before_dispatch` (it renders the envelope *and* emits the
+telemetry, so the two can't drift apart):
+
+- **Middleware short-circuits** — anything that returns instead of calling
+  `next.run(request)` (see `enforce_request_body_limit`). These run ahead of
+  authentication, so they pass `api_key_id: None`.
+- **Extractor rejections a handler unwraps at its top** — the
+  `Result<Json<T>, JsonRejection>` / `Result<Bytes, BytesRejection>` parameters.
+  Auth already ran here, so pass the key id.
+
+A handler that instead wraps its whole dispatch and logs the wrapper's status
+(`/mcp`, `/a2a`, `/passthrough`, `/v1/videos`, `/v1/files`) is already covered —
+don't add a second emit to those, or the request logs twice.
+
 ## A per-model gate must say whether it binds the requested entry or each target
 
 `resolve_attempt_models` expands a routing model into targets, so `model_entry` /

crates/aisix-proxy/src/audio.rs

@@ -286,17 +286,23 @@ pub async fn speech(
     // envelope — see completions.rs.
     body: Result<Json<Value>, axum::extract::rejection::JsonRejection>,
 ) -> Response {
+    let started = Instant::now();
     let Json(body) = match body {
         Ok(json) => json,
+        // Answer through `reject` — see completions.rs.
         Err(rej) => {
-            return crate::error::proxy_error_from_json_rejection(
-                rej,
-                state.request_body_limit_bytes,
-            )
-            .into_response();
+            return crate::reject::reject_before_dispatch(
+                &state,
+                "POST",
+                "/v1/audio/speech",
+                &client.request_id,
+                Some(&auth.entry.id),
+                started,
+                crate::reject::Envelope::OpenAi,
+                crate::error::proxy_error_from_json_rejection(rej, state.request_body_limit_bytes),
+            );
         }
     };
-    let started = Instant::now();
     let request_id = client.request_id.clone();
     let api_key_id = auth.entry.id.clone();
     let model_name = body

crates/aisix-proxy/src/chat.rs

@@ -94,30 +94,21 @@ pub async fn chat_completions(
     let path = "/v1/chat/completions";
     let mut req = match body {
         Ok(Json(r)) => r,
+        // Classify the body-extractor failure (malformed JSON vs 413 cap
+        // vs transport read error) via the shared helper, then answer
+        // through `reject` so the rejection lands in the access log and
+        // the request metrics like every other terminal path here.
         Err(rej) => {
-            use axum::extract::rejection::JsonRejection;
-            use axum::http::StatusCode;
-            // BytesRejection → distinguish 413 (PAYLOAD_TOO_LARGE,
-            // real per-extractor cap exceeded) from 400 (transport-
-            // side read failure). `JsonRejection` is `#[non_exhaustive]`
-            // so the fallback `_` arm catches today's JsonDataError
-            // (the #324 case) / JsonSyntaxError / MissingJsonContentType
-            // AND any future variant axum adds, defaulting to 400
-            // until each new variant gets an explicit policy decision.
-            return match rej {
-                JsonRejection::BytesRejection(inner)
-                    if inner.status() == StatusCode::PAYLOAD_TOO_LARGE =>
-                {
-                    ProxyError::RequestTooLarge {
-                        limit_bytes: state.request_body_limit_bytes,
-                    }
-                }
-                JsonRejection::BytesRejection(_) => {
-                    ProxyError::InvalidRequest("failed to read request body".into())
-                }
-                _ => ProxyError::InvalidRequest("invalid JSON request body".into()),
-            }
-            .into_response();
+            return crate::reject::reject_before_dispatch(
+                &state,
+                method,
+                path,
+                &client.request_id,
+                Some(&auth.entry.id),
+                started,
+                crate::reject::Envelope::OpenAi,
+                crate::error::proxy_error_from_json_rejection(rej, state.request_body_limit_bytes),
+            );
         }
     };
     let request_id = client.request_id.clone();

crates/aisix-proxy/src/completions.rs

@@ -92,17 +92,25 @@ pub async fn completions(
     // chat.rs / messages.rs.
     body: Result<Json<Value>, axum::extract::rejection::JsonRejection>,
 ) -> Response {
+    let started = Instant::now();
     let Json(body) = match body {
         Ok(json) => json,
+        // Answer through `reject` so the refusal still produces the access
+        // log line + request metrics the handler tail emits for a served
+        // request — the tail it never reaches.
         Err(rej) => {
-            return crate::error::proxy_error_from_json_rejection(
-                rej,
-                state.request_body_limit_bytes,
-            )
-            .into_response();
+            return crate::reject::reject_before_dispatch(
+                &state,
+                "POST",
+                "/v1/completions",
+                &client.request_id,
+                Some(&auth.entry.id),
+                started,
+                crate::reject::Envelope::OpenAi,
+                crate::error::proxy_error_from_json_rejection(rej, state.request_body_limit_bytes),
+            );
         }
     };
-    let started = Instant::now();
     let request_id = client.request_id.clone();
     let api_key_id = auth.entry.id.clone();
     let model_name = body

crates/aisix-proxy/src/count_tokens.rs

@@ -69,18 +69,24 @@ pub async fn count_tokens(
         Ok(a) => a,
         Err(e) => return e.into_anthropic_response(),
     };
+    let started = Instant::now();
     let Json(body) = match body {
         Ok(j) => j,
+        // Answer through `reject` — see messages.rs.
         Err(rej) => {
-            return crate::error::proxy_error_from_json_rejection(
-                rej,
-                state.request_body_limit_bytes,
-            )
-            .into_anthropic_response();
+            return crate::reject::reject_before_dispatch(
+                &state,
+                "POST",
+                "/v1/messages/count_tokens",
+                &client.request_id,
+                Some(&auth.entry.id),
+                started,
+                crate::reject::Envelope::Anthropic,
+                crate::error::proxy_error_from_json_rejection(rej, state.request_body_limit_bytes),
+            );
         }
     };
 
-    let started = Instant::now();
     let request_id = client.request_id.clone();
     let api_key_id = auth.entry.id.clone();
 

crates/aisix-proxy/src/embeddings.rs

@@ -98,29 +98,20 @@ pub async fn embeddings(
     let api_key_id = auth.entry.id.clone();
     let body = match body {
         Ok(Json(b)) => b,
+        // Classification stays in the shared helper so this route can't
+        // drift from its siblings on the 413-vs-400 rules; `reject` gives
+        // the refusal the same access log + metrics a served request gets.
         Err(rej) => {
-            use axum::extract::rejection::JsonRejection;
-            // BytesRejection → distinguish 413 (PAYLOAD_TOO_LARGE,
-            // real per-extractor cap exceeded) from 400 (transport-
-            // side read failure). `JsonRejection` is `#[non_exhaustive]`
-            // so the fallback `_` arm catches today's JsonDataError
-            // (the #401 case) / JsonSyntaxError / MissingJsonContentType
-            // AND any future variant axum adds, defaulting to 400
-            // until each new variant gets an explicit policy decision.
-            return match rej {
-                JsonRejection::BytesRejection(inner)
-                    if inner.status() == StatusCode::PAYLOAD_TOO_LARGE =>
-                {
-                    ProxyError::RequestTooLarge {
-                        limit_bytes: state.request_body_limit_bytes,
-                    }
-                }
-                JsonRejection::BytesRejection(_) => {
-                    ProxyError::InvalidRequest("failed to read request body".into())
-                }
-                _ => ProxyError::InvalidRequest("invalid JSON request body".into()),
-            }
-            .into_response();
+            return crate::reject::reject_before_dispatch(
+                &state,
+                "POST",
+                "/v1/embeddings",
+                &request_id,
+                Some(&api_key_id),
+                started,
+                crate::reject::Envelope::OpenAi,
+                crate::error::proxy_error_from_json_rejection(rej, state.request_body_limit_bytes),
+            );
         }
     };
     let model_name = body.model.clone();

crates/aisix-proxy/src/images.rs

@@ -70,17 +70,23 @@ pub async fn image_generations(
     // envelope — see completions.rs.
     body: Result<Json<Value>, axum::extract::rejection::JsonRejection>,
 ) -> Response {
+    let started = Instant::now();
     let Json(body) = match body {
         Ok(json) => json,
+        // Answer through `reject` — see completions.rs.
         Err(rej) => {
-            return crate::error::proxy_error_from_json_rejection(
-                rej,
-                state.request_body_limit_bytes,
-            )
-            .into_response();
+            return crate::reject::reject_before_dispatch(
+                &state,
+                "POST",
+                "/v1/images/generations",
+                &client.request_id,
+                Some(&auth.entry.id),
+                started,
+                crate::reject::Envelope::OpenAi,
+                crate::error::proxy_error_from_json_rejection(rej, state.request_body_limit_bytes),
+            );
         }
     };
-    let started = Instant::now();
     let request_id = client.request_id.clone();
     let api_key_id = auth.entry.id.clone();
     let model_name = body

crates/aisix-proxy/src/jobs.rs

@@ -1027,17 +1027,23 @@ pub(crate) async fn create_batch(
     // text/plain rejection — see completions.rs.
     body: Result<Bytes, axum::extract::rejection::BytesRejection>,
 ) -> Response {
+    let started = Instant::now();
     let body = match body {
         Ok(bytes) => bytes,
+        // Answer through `reject` — see completions.rs.
         Err(rej) => {
-            return crate::error::proxy_error_from_bytes_rejection(
-                rej,
-                state.request_body_limit_bytes,
-            )
-            .into_response();
+            return crate::reject::reject_before_dispatch(
+                &state,
+                "POST",
+                "/v1/batches",
+                &client.request_id,
+                Some(&auth.entry.id),
+                started,
+                crate::reject::Envelope::OpenAi,
+                crate::error::proxy_error_from_bytes_rejection(rej, state.request_body_limit_bytes),
+            );
         }
     };
-    let started = Instant::now();
     let request_id = client.request_id.clone();
     let mut monitor_hits: Vec<aisix_core::GuardrailMonitorHit> = Vec::new();
 
@@ -1267,17 +1273,23 @@ pub(crate) async fn create_ft_job(
     // text/plain rejection — see completions.rs.
     body: Result<Bytes, axum::extract::rejection::BytesRejection>,
 ) -> Response {
+    let started = Instant::now();
     let body = match body {
         Ok(bytes) => bytes,
+        // Answer through `reject` — see completions.rs.
         Err(rej) => {
-            return crate::error::proxy_error_from_bytes_rejection(
-                rej,
-                state.request_body_limit_bytes,
-            )
-            .into_response();
+            return crate::reject::reject_before_dispatch(
+                &state,
+                "POST",
+                "/v1/fine_tuning/jobs",
+                &client.request_id,
+                Some(&auth.entry.id),
+                started,
+                crate::reject::Envelope::OpenAi,
+                crate::error::proxy_error_from_bytes_rejection(rej, state.request_body_limit_bytes),
+            );
         }
     };
-    let started = Instant::now();
     let request_id = client.request_id.clone();
     let mut monitor_hits: Vec<aisix_core::GuardrailMonitorHit> = Vec::new();