fix(gateway): unify INVALID_INPUT to 422 on hand-rolled paths + sink deadline (GW-16, GW-17)

GW-16: empty body / malformed first line / missing header fields /
per-line cap / batch over-cap rejections now route through
call_error_to_http_response_with_identity, mapping INVALID_INPUT to
422 — same status as mid-stream chunk errors. One error class, one
status.

GW-17: invoke_sink wraps the registry sink invoke in the same 30 s
tokio::time::timeout the Once-op invoke uses; a hung sink handler
surfaces as a TIMEOUT (504, retryable) error envelope instead of
holding the HTTP connection forever. The sink wrapper bounds the
whole dispatch (chunk pacing included), matching http-server.md's
deadline contract.

Docs: http-server.md error table documents the 422 triggers and the
sink deadline; http-adapters.md batch cap status corrected.

to_openapi: gateway spec version 1.3.0 -> 1.4.0 (ADR-045 minor):
/publish framing faults and /batch cap reject documented at 422 (the
400 slots moved with the runtime); /publish 400 slot removed; 504
description covers the sink dispatch.

Verification: scripts/verify.sh OK (397 tests); cargo test
--all-features OK (513 tests); clippy --all-features --all-targets -D
warnings OK; cargo fmt --check OK.

docs/architecture/http-adapters.md

@@ -545,7 +545,9 @@ adapter contract from alkcall ADR-022 faithful on the error axis — no
 silent dropping of error contracts. The `/batch` endpoint documents no
 HTTP 500 (review-002 PRJ-21): every per-entry dispatch failure is an
 in-band `results[]` entry; the only HTTP error status is the
-request-level cap failure (400). `BatchResultEntry.error` references
+request-level cap failure (422; GW-16 unified it with the
+`INVALID_INPUT → 422` mapping — it was hand-rolled as 400 before
+review-002). `BatchResultEntry.error` references
 the `BatchError` component (review-002 PRJ-16b): the serialized
 `CallError` as a oneOf over the six protocol-code envelopes plus a
 generic arm carrying the operation-declared codes. See alkcall ADR-016.

docs/architecture/http-server.md

@@ -217,7 +217,7 @@ the response is `text/event-stream` (negotiated via
   check before dispatch. The two methods diverge only on the return shape
   (stream vs single envelope). Streaming invokes set `deadline: None` —
   subscriptions are unbounded by contract, unlike the 30 s gateway
-  deadline on Once-op invokes (see Error Mapping below).
+  deadline on Once-op and sink invokes (see Error Mapping below).
 - For each `ResponseEnvelope` the stream yields, writes an SSE `data:` frame:
   `Ok(value)` → `data:` frame with the output serialized as JSON; `Err` →
   SSE error event with the `CallError` serialized, then close (an `Err` is
@@ -358,7 +358,7 @@ Schemas") map to HTTP status codes:
 |-------------|-------------|-------|
 | `NOT_FOUND` (operation not registered, or Internal op) | `404` | |
 | `FORBIDDEN` (insufficient scopes, or unauthenticated) | `401` (no token) / `403` (token present) | |
-| `INVALID_INPUT` (schema mismatch) | `422` | |
+| `INVALID_INPUT` (input-data fault) | `422` | one status for every trigger: schema mismatch, `/publish` NDJSON framing faults (empty body, malformed first line, missing `operation`/`chunk`, per-line cap, body-read failure), and the `/batch` over-cap reject (`INVALID_INPUT` is an input-data fault wherever it fires — GW-16 unified the formerly hand-rolled 400s onto this row) |
 | `INVALID_OPERATION_TYPE` (wrong dispatch path for the op's type) | `422` (token present) / `401` (no token) | consistent across `/call`, `/batch`, `/publish` — a client fault, never a server fault |
 | `TIMEOUT` | `504` | `retryable: true` |
 | `INTERNAL` | `500` | |
@@ -379,11 +379,15 @@ ADR-016) and `from_openapi`-imported codes are prefixed `HTTP_<status>`
 to avoid collision with protocol codes.
 
 **Per-endpoint dispatch deadline.** Once-op invokes (`/call`, `/batch`
-entries, `/search`, `/schema`, and the `/publish` final envelope) are
+entries, `/search`, `/schema`) and sink invokes (the `/publish` final
-bounded by a 30 s gateway deadline (`GatewayDispatch::invoke` wraps the
+envelope) are bounded by the 30 s gateway deadline
-registry invoke in `tokio::time::timeout`); a hung handler surfaces as
+(`GatewayDispatch::invoke` and `GatewayDispatch::invoke_sink` wrap the
-a `TIMEOUT` error (`504`, `retryable: true`), not an indefinitely-held
+registry invoke in `tokio::time::timeout`, GW-17); a hung handler —
-HTTP request. Streaming invokes (`/subscribe`) are unbounded —
+Once or sink — surfaces as a `TIMEOUT` error (`504`, `retryable:
+true`), not an indefinitely-held HTTP request. The sink wrapper bounds
+the whole dispatch (chunk pacing included), so the final envelope
+always arrives, or the deadline trips, within the window. Streaming
+invokes (`/subscribe`) are unbounded —
 subscriptions are long-lived by contract (alkcall ADR-021 sets
 `deadline: None` for the streaming branch). The same time/bytes split
 governs the **outbound** half of an imported subscription:

src/adapters/to_openapi.rs

@@ -30,7 +30,15 @@
 //! corrections remove only documentation of statuses/codes the runtime
 //! never emitted (a strict client matching them observed nothing to
 //! break), and every added slot documents behavior the runtime already
-//! had — the wire contract is unchanged.
+//! had — the wire contract is unchanged. `1.4.0` is the GW-16 status
+//! unification: the runtime's hand-rolled `/publish` framing faults
+//! and the `/batch` cap rejection moved from 400 to the documented
+//! `INVALID_INPUT → 422` mapping (one error class, one status), so the
+//! doc's 400/422 slots moved with the runtime. Clients matching the old
+//! 400 slots observed behavior the runtime no longer emits; the 422
+//! slots are where `INVALID_INPUT` was already documented, so the
+//! client-visible consequence is additive on the documented mapping
+//! side and the status drift is gone.
 //!
 //! # Error fidelity
 //!
@@ -76,7 +84,7 @@ use alkcall::registry::spec::ErrorDefinition;
 use super::openapi_spec::OpenAPISpec;
 use crate::gateway::MAX_BATCH_OPERATIONS;
 
-const GATEWAY_VERSION: &str = "1.3.0";
+const GATEWAY_VERSION: &str = "1.4.0";
 const GATEWAY_TITLE: &str = "alk gateway";
 const OPENAPI_VERSION: &str = "3.0.0";
 
@@ -349,10 +357,6 @@ fn publish_responses(operation_errors: &BTreeMap<u16, Value>) -> Value {
     let mut responses = json!({
         "200": json_response(ref_schema("CallOk"),
             "The operation's final ResponseEnvelope: request_id, result=ok, output."),
-        "400": json_response(one_of_refs(&[
-            "CallErrorInvalidInput".to_string(),
-        ]),
-            "Malformed NDJSON stream framing: empty body, first line missing 'operation'/'chunk', a line exceeding the 2 MiB per-line cap, or a raw body-read failure (INVALID_INPUT). A later NDJSON line that is not valid JSON is a dispatch-path 422. The gateway's 2 MiB + 64 KiB body-limit layer answers an oversized whole-body upload with a plain-text 413 before the route runs. A non-Pub op without a resolved token is the 401 below (INVALID_OPERATION_TYPE), not this status (PRJ-18)."),
         "401": json_response(one_of_refs(&[
             "CallErrorForbidden".to_string(),
             "CallErrorInvalidOperationType".to_string(),
@@ -366,14 +370,14 @@ fn publish_responses(operation_errors: &BTreeMap<u16, Value>) -> Value {
             "CallErrorInvalidInput".to_string(),
             "CallErrorInvalidOperationType".to_string(),
         ]),
-            "Dispatch-path client fault: a chunk failed the operation's publish_schema validation (INVALID_INPUT with details.chunk), a later NDJSON line was not valid JSON, or the operation's type is not Pub (INVALID_OPERATION_TYPE, token present)."),
+            "Dispatch-path client fault: NDJSON stream framing faults — empty body, first line missing 'operation'/'chunk', a line exceeding the 2 MiB per-line cap, or a raw body-read failure (INVALID_INPUT, GW-16 unified with the mid-stream status) — plus a later NDJSON line that was not valid JSON, a chunk that failed the operation's publish_schema validation (INVALID_INPUT with details.chunk), or the operation's type is not Pub (INVALID_OPERATION_TYPE, token present). The gateway's 2 MiB + 64 KiB body-limit layer answers an oversized whole-body upload with a plain-text 413 before the route runs."),
         "500": json_response(one_of_refs(&[
             "CallErrorInternal".to_string(),
             "PublishFailure".to_string(),
         ]),
             "Dispatcher failure, or an operation-level error code without HTTP_ prefix or http_status (the runtime mapper is purely code-driven and such codes surface as 500)."),
         "504": json_response(ref_response("Timeout"),
-            "The Once-op final envelope dispatch exceeded the 30 s gateway deadline (retryable)."),
+            "The sink dispatch exceeded the 30 s gateway deadline (retryable) — the same deadline bounds the Once-op invoke (GW-17)."),
     });
     merge_operation_errors(&mut responses, operation_errors);
     responses
@@ -400,10 +404,10 @@ fn batch_path_item() -> Value {
             "responses": {
                 "200": json_response(ref_schema("BatchResponse"),
                     "results[] shares entries' order with the request; each entry is an envelope-shaped {request_id, result, output|error} object; entries for Internal ops carry a NOT_FOUND in-band error. Per-call dispatch failures surface only as these in-band entries — there is no HTTP error status for an individual call."),
-                "400": json_response(one_of_refs(&[
+                "422": json_response(one_of_refs(&[
                     "BatchCapExceeded".to_string(),
                 ]),
-                    "Request-level failure: the batch exceeds 100 operations (INVALID_INPUT, JSON). Malformed JSON bodies are rejected earlier by the extractors with a plain-text 400 (not this JSON shape); a JSON array body whose items fail deserialization is a plain-text extractor 422. Oversized uploads are pre-empted by the gateway's body-limit layer with a plain-text 413."),
+                    "Request-level failure: the batch exceeds 100 operations (INVALID_INPUT, JSON; GW-16 unified with the INVALID_INPUT → 422 mapping). Malformed JSON bodies are rejected earlier by the extractors with a plain-text 400 (not this JSON shape); a JSON array body whose items fail deserialization is a plain-text extractor 422. Oversized uploads are pre-empted by the gateway's body-limit layer with a plain-text 413."),
             }
         }
     })
@@ -1066,7 +1070,7 @@ mod tests {
     }
 
     #[test]
-    fn info_version_is_1_3_0_after_projection_truthfulness() {
+    fn info_version_is_1_4_0_after_gw16_status_unification() {
         let registry = OperationRegistry::new();
         let spec = to_openapi(&registry).unwrap();
         let version = spec
@@ -1076,8 +1080,8 @@ mod tests {
             .unwrap();
         assert_eq!(version, GATEWAY_VERSION);
         assert_eq!(
-            version, "1.3.0",
+            version, "1.4.0",
-            "minor bump: doc-contract corrections only — the wire contract is unchanged (ADR-045)"
+            "minor bump: status-mapping truthfulness corrections — the 422 slots are where INVALID_INPUT was already documented (ADR-045)"
         );
     }
 
@@ -1328,12 +1332,15 @@ mod tests {
     }
 
     #[test]
-    fn batch_documents_400_json_shape_and_422_free_call_entries() {
+    fn batch_documents_422_json_shape_for_the_cap_reject() {
         let registry = OperationRegistry::new();
         let spec = to_openapi(&registry).unwrap();
         let responses = responses(&spec, PATH_BATCH, "post");
-        assert!(responses.contains_key("400"));
+        assert!(
-        let schema = response_schema(responses.get("400").unwrap());
+            !responses.contains_key("400"),
+            "GW-16: the batch cap reject moved to the INVALID_INPUT → 422 mapping"
+        );
+        let schema = response_schema(responses.get("422").unwrap());
         let refs = one_of_refs_of(schema);
         assert!(refs.contains(&"#/components/schemas/BatchCapExceeded".to_string()));
     }
@@ -1729,12 +1736,16 @@ mod tests {
         let registry = OperationRegistry::new();
         let spec = to_openapi(&registry).unwrap();
         let responses = responses(&spec, PATH_PUBLISH, "post");
-        for status in ["200", "400", "401", "403", "404", "422", "500", "504"] {
+        for status in ["200", "401", "403", "404", "422", "500", "504"] {
             assert!(
                 responses.contains_key(status),
                 "/publish {status} documented"
             );
         }
+        assert!(
+            !responses.contains_key("400"),
+            "GW-16: /publish framing faults are INVALID_INPUT → 422; no 400 slot remains"
+        );
         assert!(
             !responses.contains_key("429") && !responses.contains_key("503"),
             "no operation-declared statuses on an empty registry"
@@ -1742,12 +1753,12 @@ mod tests {
     }
 
     #[test]
-    fn publish_400_and_401_carry_the_identity_split_codes() {
+    fn publish_401_and_422_carry_the_identity_split_codes() {
         let registry = OperationRegistry::new();
         let spec = to_openapi(&registry).unwrap();
         let responses = responses(&spec, PATH_PUBLISH, "post");
-        let refs_400 = one_of_refs_of(response_schema(&responses["400"]));
+        let refs_422 = one_of_refs_of(response_schema(&responses["422"]));
-        assert!(refs_400.contains(&"#/components/schemas/CallErrorInvalidInput".to_string()));
+        assert!(refs_422.contains(&"#/components/schemas/CallErrorInvalidInput".to_string()));
         let refs_401 = one_of_refs_of(response_schema(&responses["401"]));
         assert!(
             refs_401.contains(&"#/components/schemas/CallErrorInvalidOperationType".to_string())
@@ -1755,23 +1766,30 @@ mod tests {
     }
 
     #[test]
-    fn publish_400_documents_only_the_framing_contract() {
+    fn publish_422_documents_the_unified_framing_and_chunk_contract() {
         let registry = OperationRegistry::new();
         let spec = to_openapi(&registry).unwrap();
         let responses = responses(&spec, PATH_PUBLISH, "post");
-        let refs_400 = one_of_refs_of(response_schema(&responses["400"]));
+        let refs_422 = one_of_refs_of(response_schema(&responses["422"]));
         assert_eq!(
-            refs_400,
+            refs_422,
-            vec!["#/components/schemas/CallErrorInvalidInput".to_string()],
+            vec![
-            "PRJ-18: /publish 400 oneOf is exactly the INVALID_INPUT framing code"
+                "#/components/schemas/CallErrorInvalidInput".to_string(),
+                "#/components/schemas/CallErrorInvalidOperationType".to_string(),
+            ],
+            "GW-16: /publish 422 oneOf carries the framing INVALID_INPUT plus INVALID_OPERATION_TYPE"
         );
-        let description = responses["400"]
+        let description = responses["422"]
             .get("description")
             .and_then(Value::as_str)
             .unwrap();
+        assert!(
+            description.contains("first line missing"),
+            "GW-16: the 422 description names the framing faults now mapped 422: {description}"
+        );
         assert!(
             !description.contains("reported INVALID_OPERATION_TYPE"),
-            "PRJ-18: the 400 description must not claim the 401-reported condition as a 400 outcome: {description}"
+            "the 422 description must not claim the 401-reported condition as a 422 outcome: {description}"
         );
     }
 

src/gateway/dispatch.rs

@@ -21,14 +21,17 @@
 //!
 //! # Deadline
 //!
-//! Once-op invokes ([`GatewayDispatch::invoke`]) are bounded by
+//! Once-op invokes ([`GatewayDispatch::invoke`]) and sink invokes
+//! ([`GatewayDispatch::invoke_sink`]; GW-17) are bounded by
 //! `DEFAULT_TIMEOUT` (30 s): the registry invoke is wrapped in
 //! `tokio::time::timeout` and a hung handler surfaces as a `TIMEOUT`
 //! error envelope (`504` under the gateway's error mapping), not an
-//! indefinitely-held HTTP request. Streaming and sink dispatch set
+//! indefinitely-held HTTP request. The sink wrapper bounds the whole
-//! `deadline: None` (subscriptions are unbounded per alkcall ADR-021,
+//! dispatch — chunk upload pacing and the handler's final-completion
-//! and a `/publish` body is bounded by the client's upload, not a
+//! await alike — so the final envelope always arrives (or the deadline
-//! fixed window).
+//! trips) within the window http-server.md documents. Streaming
+//! dispatch sets `deadline: None` (subscriptions are unbounded per
+//! alkcall ADR-021).
 
 use std::collections::HashMap;
 use std::sync::atomic::{AtomicUsize, Ordering};
@@ -128,6 +131,11 @@ impl GatewayDispatch {
     /// `ResponseEnvelope` is the result. Pre-handler failures
     /// (not-found, forbidden, non-Pub op) surface as a single error
     /// envelope — the same envelope the `/publish` route maps to HTTP.
+    /// The whole dispatch — chunk streaming and the handler's final
+    /// await alike — is bounded by the same 30 s gateway deadline as
+    /// the Once-op invoke (GW-17): a hung sink handler surfaces as a
+    /// `TIMEOUT` error envelope (504 under the gateway's error
+    /// mapping), not an indefinitely-held HTTP request.
     pub async fn invoke_sink(
         &self,
         identity: Option<Identity>,
@@ -138,9 +146,21 @@ impl GatewayDispatch {
         let operation_name = strip_leading_slash(op).to_string();
         let request_id = uuid::Uuid::new_v4().to_string();
         let context = self.build_root_context_sink(&request_id, &operation_name, identity);
+        let result = tokio::time::timeout(
+            DEFAULT_TIMEOUT,
             self.registry
-            .invoke_sink(&operation_name, input, publish_stream, context)
+                .invoke_sink(&operation_name, input, publish_stream, context),
-            .await
+        )
+        .await;
+        match result {
+            Ok(envelope) => envelope,
+            Err(_elapsed) => ResponseEnvelope::error(
+                request_id,
+                CallError::timeout(format!(
+                    "operation did not complete within the {DEFAULT_TIMEOUT:?} gateway deadline"
+                )),
+            ),
+        }
     }
 
     fn build_root_context_sink(
@@ -216,7 +236,8 @@ fn strip_leading_slash(operation_id: &str) -> &str {
 mod tests {
     use super::*;
     use alkcall::registry::registration::{
-        make_handler, make_streaming_handler, HandlerKind, HandlerRegistration, OperationProvenance,
+        make_handler, make_sink_handler, make_streaming_handler, HandlerKind, HandlerRegistration,
+        OperationProvenance,
     };
     use alkcall::registry::spec::{AccessControl, OperationSpec, OperationType, Visibility};
     use futures::StreamExt;
@@ -346,6 +367,82 @@ mod tests {
         assert!(envelope.result.is_ok(), "a fast handler must not time out");
     }
 
+    #[tokio::test]
+    async fn invoke_sink_enforces_the_default_deadline_on_a_hung_sink_handler() {
+        use std::time::Duration;
+
+        let mut registry = OperationRegistry::new();
+        registry
+            .register(HandlerRegistration::new(
+                spec("hung/sink", Visibility::External, OperationType::Pub),
+                HandlerKind::Sink(make_sink_handler(|_input, ctx, _chunks| async move {
+                    tokio::time::sleep(Duration::from_secs(120)).await;
+                    ResponseEnvelope::ok(ctx.request_id, serde_json::json!({}))
+                })),
+                OperationProvenance::Local,
+                None,
+                None,
+                Capabilities::new(),
+            ))
+            .unwrap();
+        let dispatch = GatewayDispatch::new(Arc::new(registry));
+        let chunks: alkcall::registry::registration::PublishStream =
+            Box::pin(futures::stream::empty());
+        let started = std::time::Instant::now();
+        let envelope = dispatch
+            .invoke_sink(None, "/hung/sink", serde_json::json!({}), chunks)
+            .await;
+        let elapsed = started.elapsed();
+        assert!(
+            elapsed < Duration::from_secs(60),
+            "the 30 s deadline must fire well before the 120 s handler sleep, took {elapsed:?}"
+        );
+        match envelope.result {
+            Err(error) => {
+                assert_eq!(error.code, "TIMEOUT");
+                assert!(error.retryable, "the deadline error is retryable");
+            }
+            Ok(v) => panic!("expected a TIMEOUT error, got {v:?}"),
+        }
+    }
+
+    #[tokio::test]
+    async fn invoke_sink_completes_within_the_deadline_for_a_fast_sink_handler() {
+        let mut registry = OperationRegistry::new();
+        registry
+            .register(HandlerRegistration::new(
+                spec("fast/sink", Visibility::External, OperationType::Pub),
+                HandlerKind::Sink(make_sink_handler(|_input, ctx, mut chunks| async move {
+                    use futures::StreamExt;
+                    let mut count = 0u64;
+                    while let Some(chunk) = chunks.next().await {
+                        if chunk.is_ok() {
+                            count += 1;
+                        }
+                    }
+                    ResponseEnvelope::ok(ctx.request_id, serde_json::json!({ "count": count }))
+                })),
+                OperationProvenance::Local,
+                None,
+                None,
+                Capabilities::new(),
+            ))
+            .unwrap();
+        let dispatch = GatewayDispatch::new(Arc::new(registry));
+        let chunks: alkcall::registry::registration::PublishStream =
+            Box::pin(futures::stream::iter(vec![
+                Ok(serde_json::json!({ "n": 1 })),
+                Ok(serde_json::json!({ "n": 2 })),
+            ]));
+        let envelope = dispatch
+            .invoke_sink(None, "/fast/sink", serde_json::json!({}), chunks)
+            .await;
+        assert!(
+            envelope.result.is_ok(),
+            "a fast sink handler must not time out, got {envelope:?}"
+        );
+    }
+
     #[tokio::test]
     async fn streaming_sub_op_streams_envelopes() {
         let mut registry = OperationRegistry::new();

src/gateway/routes.rs

@@ -14,10 +14,11 @@
 //! an extension extractors read), so the gateway installs its own cap
 //! instead.
 //!
-//! Module status mapping note (GW-16 tracks the drift): the
+//! Status mapping note (GW-16): all `INVALID_INPUT` responses on the
-//! hand-rolled pre-dispatch rejections below use 400/`INVALID_INPUT`
+//! hand-rolled pre-dispatch paths (empty body, malformed first line,
-//! while mid-stream chunk errors map 422 through `gateway::error`;
+//! missing header fields, per-line cap, batch over cap) route through
-//! normalizing them is GW-16, not GW-15.
+//! the shared `gateway::error` mapper and map `422` — the same status
+//! as mid-stream chunk errors. One error class, one status.
 
 use std::collections::VecDeque;
 use std::convert::Infallible;
@@ -192,16 +193,12 @@ pub(crate) async fn batch_handler(
     Json(requests): Json<Vec<CallRequest>>,
 ) -> Response {
     if requests.len() > MAX_BATCH_OPERATIONS {
-        return (
+        return call_error_to_http_response_with_identity(
-            StatusCode::BAD_REQUEST,
+            &CallError::invalid_input(format!(
-            Json(json!({
-                "code": "INVALID_INPUT",
-                "message": format!(
                 "batch exceeds the maximum of {MAX_BATCH_OPERATIONS} operations"
-                ),
+            )),
-            })),
+            identity.as_ref(),
-        )
+        );
-            .into_response();
     }
     let dispatch = state.dispatch();
     let mut results: Vec<Value> = Vec::with_capacity(requests.len());
@@ -401,11 +398,7 @@ fn missing_header_field_response() -> Response {
 }
 
 fn invalid_input_response(message: &str) -> Response {
-    (
+    call_error_to_http_response_with_identity(&CallError::invalid_input(message), None)
-        StatusCode::BAD_REQUEST,
-        Json(json!({ "code": "INVALID_INPUT", "message": message })),
-    )
-        .into_response()
 }
 
 /// The explicit gateway request-body limit (GW-15). See
@@ -1461,14 +1454,14 @@ mod tests {
         );
     }
     #[tokio::test]
-    async fn batch_exceeding_operation_cap_returns_400_invalid_input() {
+    async fn batch_exceeding_operation_cap_returns_422_invalid_input() {
         let router = build_router(registry_with_echo(), unused_provider());
         let requests: Vec<Value> = (0..MAX_BATCH_OPERATIONS + 1)
             .map(|i| json!({ "operation": "echo/run", "input": { "n": i } }))
             .collect();
         let req = json_request("POST", "/batch", json!(requests));
         let (status, body) = send(router, req).await;
-        assert_eq!(status, StatusCode::BAD_REQUEST);
+        assert_eq!(status, StatusCode::UNPROCESSABLE_ENTITY);
         assert_eq!(body.get("code"), Some(&json!("INVALID_INPUT")));
     }
 
@@ -2270,31 +2263,31 @@ mod tests {
     }
 
     #[tokio::test]
-    async fn publish_missing_operation_in_first_line_returns_400() {
+    async fn publish_missing_operation_in_first_line_returns_422() {
         let router = build_router(publish_registry(), unused_provider());
         let body = ndjson(&[json!({ "chunk": {} })]);
         let req = raw_request("POST", "/publish", body);
         let (status, body) = send(router, req).await;
-        assert_eq!(status, StatusCode::BAD_REQUEST);
+        assert_eq!(status, StatusCode::UNPROCESSABLE_ENTITY);
         assert_eq!(body.get("code"), Some(&json!("INVALID_INPUT")));
     }
 
     #[tokio::test]
-    async fn publish_empty_body_returns_400() {
+    async fn publish_empty_body_returns_422() {
         let router = build_router(publish_registry(), unused_provider());
         let req = raw_request("POST", "/publish", Vec::new());
         let (status, body) = send(router, req).await;
-        assert_eq!(status, StatusCode::BAD_REQUEST);
+        assert_eq!(status, StatusCode::UNPROCESSABLE_ENTITY);
         assert_eq!(body.get("code"), Some(&json!("INVALID_INPUT")));
     }
 
     #[tokio::test]
-    async fn publish_first_line_missing_chunk_returns_400_invalid_input() {
+    async fn publish_first_line_missing_chunk_returns_422_invalid_input() {
         let router = build_router(publish_registry(), unused_provider());
         let body = ndjson(&[json!({ "operation": "ingest/push" })]);
         let req = raw_request("POST", "/publish", body);
         let (status, body) = send(router, req).await;
-        assert_eq!(status, StatusCode::BAD_REQUEST);
+        assert_eq!(status, StatusCode::UNPROCESSABLE_ENTITY);
         assert_eq!(body.get("code"), Some(&json!("INVALID_INPUT")));
     }
 
@@ -2453,7 +2446,7 @@ mod tests {
         body.extend_from_slice(&oversized);
         body.push(b'\n');
         let (status, resp) = send(router, raw_request("POST", "/publish", body)).await;
-        assert_eq!(status, StatusCode::BAD_REQUEST);
+        assert_eq!(status, StatusCode::UNPROCESSABLE_ENTITY);
         assert_eq!(resp.get("code"), Some(&json!("INVALID_INPUT")));
         assert!(
             resp.get("message")
@@ -2552,7 +2545,7 @@ mod tests {
         let (status, resp) = send(router, raw_request("POST", "/publish", body)).await;
         assert_eq!(
             status,
-            StatusCode::BAD_REQUEST,
+            StatusCode::UNPROCESSABLE_ENTITY,
             "EOF with an over-cap unterminated tail must not yield the buffer: {resp}"
         );
         assert_eq!(resp.get("code"), Some(&json!("INVALID_INPUT")));
@@ -2638,9 +2631,9 @@ mod tests {
         let (status, resp) = send(router, raw_request("POST", "/publish", body)).await;
         assert_eq!(
             status,
-            StatusCode::BAD_REQUEST,
+            StatusCode::UNPROCESSABLE_ENTITY,
             "an over-cap line batched with complete lines must still abort the request with the \
-             cap error (today's status; GW-16 normalizes): {resp}"
+             cap error (GW-16 unifies it with the mid-stream 422): {resp}"
         );
         assert_eq!(resp.get("code"), Some(&json!("INVALID_INPUT")));
     }

src/server/adapter.rs

@@ -892,7 +892,7 @@ mod tests {
         // The 6-endpoint gateway doc; the version tracks the projection
         // truthfulness pass.
         assert!(text.contains("\"/publish\""), "publish path in doc");
-        assert!(text.contains("1.3.0"), "info.version 1.3.0 in doc");
+        assert!(text.contains("1.4.0"), "info.version 1.4.0 in doc");
         assert!(text.contains("gatewayPublish"), "publish operationId");
 
         let _ = server_task.await;

tests/full_surface.rs

@@ -265,7 +265,7 @@ async fn full_surface_gateway_over_http() {
         .unwrap();
     assert_eq!(resp.status(), 200);
     let body: serde_json::Value = resp.json().await.unwrap();
-    assert_eq!(body["info"]["version"], "1.3.0");
+    assert_eq!(body["info"]["version"], "1.4.0");
     assert!(body["paths"].get("/publish").is_some());
     assert!(body["paths"].get("/call").is_some());
 }