From 6a7d5972bd9faa0065ac23ec7b5164ea77c17811 Mon Sep 17 00:00:00 2001
From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com>
Date: Fri, 13 Nov 2026 10:28:29 +0000
Subject: [PATCH 1/4] Deliver bounded oversized completion refusals and
 advertise wire cap

---
 docs/server-reference.md                      |  19 +++
 .../cluster-info-envelope.schema.json         |   1 +
 ...low-task-completion-limits-v1-vectors.json |  53 ++++++++
 rust/src/runtime/payload_discovery.rs         |   1 +
 rust/src/runtime/payload_limits.rs            |  54 ++++++--
 rust/tests/execution/payload_discovery.rs     |   1 +
 rust/tests/execution/payload_limits.rs        | 128 ++++++++++++++++++
 .../server-parity/cluster-info-contract.mjs   |   9 +-
 .../server-parity/native-divergences.json     |   9 ++
 .../ServerParityClusterInfoContractTest.mjs   |   5 +-
 10 files changed, 270 insertions(+), 10 deletions(-)
 create mode 100644 resources/platform-protocol-specs/workflow-task-completion-limits-v1-vectors.json

diff --git a/docs/server-reference.md b/docs/server-reference.md
index 4a6be5b53..f67804688 100644
--- a/docs/server-reference.md
+++ b/docs/server-reference.md
@@ -645,6 +645,25 @@ workflow-task command payload.
 dedicated matching-role loop should skip a pass rather than duplicate work
 already covered by another matching-role process holding the repair throttle.
 
+The native completion transport correction for [#1136](https://github.com/durable-workflow/server/issues/1136)
+adds `limits.max_workflow_task_completion_bytes` to cluster discovery. It is the
+actual cap on the complete encoded JSON HTTP request body for workflow-task
+completion and activity continuation completion, including envelopes, metadata
+and commands. The default is 33,554,432 bytes. `limits.max_payload_bytes` remains
+the per-value and ordinary API request limit; a decoded argument sum does not
+measure completion wire size. SDKs use the explicit optional field for those
+two completion routes and preserve their conservative fallback when it is absent.
+
+A completion over that cap returns HTTP 413 with `reason: payload_too_large`,
+`limit_kind: workflow_task_completion_bytes` and the actual byte `limit`, before
+any workflow state changes. The server discards at most 64 MiB of remaining
+upload data for at most two seconds before closing the HTTP/1 connection. This
+lets ordinary within-budget uploads finish sending and receive the typed refusal;
+clients that exceed either discard bound can still encounter a transport failure.
+The [shared size vectors](../resources/platform-protocol-specs/workflow-task-completion-limits-v1-vectors.json)
+cover exact and over-cap bodies on both completion routes. Published SDK worker
+qualification remains required before claiming this correction delivered.
+
 ### Namespaces
 - `GET /api/namespaces` — List namespaces
 - `POST /api/namespaces` — Create namespace
diff --git a/resources/platform-protocol-specs/cluster-info-envelope.schema.json b/resources/platform-protocol-specs/cluster-info-envelope.schema.json
index baf08ce90..0022f45de 100644
--- a/resources/platform-protocol-specs/cluster-info-envelope.schema.json
+++ b/resources/platform-protocol-specs/cluster-info-envelope.schema.json
@@ -174,6 +174,7 @@
       "additionalProperties": true,
       "properties": {
         "max_payload_bytes": {"type": "integer", "minimum": 1},
+        "max_workflow_task_completion_bytes": {"type": "integer", "minimum": 0, "description": "Optional actual encoded JSON HTTP request-body cap for workflow-task completion and activity continuation completion. Per-value max_payload_bytes still applies. Missing means the producer has not advertised a separate completion budget; consumers must preserve their conservative fallback and must not infer a multiplier from server version."},
         "max_memo_bytes": {"type": "integer", "minimum": 1},
         "max_search_attributes": {"type": "integer", "minimum": 0},
         "max_search_attribute_key_length": {"type": "integer", "minimum": 1},
diff --git a/resources/platform-protocol-specs/workflow-task-completion-limits-v1-vectors.json b/resources/platform-protocol-specs/workflow-task-completion-limits-v1-vectors.json
new file mode 100644
index 000000000..406dc0817
--- /dev/null
+++ b/resources/platform-protocol-specs/workflow-task-completion-limits-v1-vectors.json
@@ -0,0 +1,53 @@
+{
+  "schema": "durable-workflow.workflow-task-completion-limits.v1",
+  "issue": "https://github.com/durable-workflow/server/issues/1136",
+  "measurement": "complete_encoded_json_http_request_body_bytes",
+  "default_limits": {
+    "max_payload_bytes": 2097152,
+    "max_workflow_task_completion_bytes": 33554432
+  },
+  "completion_routes": [
+    "/api/worker/workflow-tasks/fixture-task/complete",
+    "/api/worker/activity-tasks/fixture-task/continuation/complete"
+  ],
+  "ordinary_route": "/api/workflows",
+  "wire_cases": [
+    {
+      "id": "ordinary-exact",
+      "route_kind": "ordinary",
+      "wire_bytes": 2097152,
+      "size_admitted": true
+    },
+    {
+      "id": "ordinary-over",
+      "route_kind": "ordinary",
+      "wire_bytes": 2097153,
+      "size_admitted": false
+    },
+    {
+      "id": "completion-over-ordinary",
+      "route_kind": "completion",
+      "wire_bytes": 2097153,
+      "size_admitted": true
+    },
+    {
+      "id": "completion-exact",
+      "route_kind": "completion",
+      "wire_bytes": 33554432,
+      "size_admitted": true
+    },
+    {
+      "id": "completion-over",
+      "route_kind": "completion",
+      "wire_bytes": 33554433,
+      "size_admitted": false
+    }
+  ],
+  "oversized_response": {
+    "status": 413,
+    "reason": "payload_too_large",
+    "limit_kind": "workflow_task_completion_bytes",
+    "limit": 33554432
+  },
+  "scope": "Size admission alone. Authentication, lease fences, per-value Avro limits and command validation still apply; no durable work is admitted by a size refusal."
+}
diff --git a/rust/src/runtime/payload_discovery.rs b/rust/src/runtime/payload_discovery.rs
index 0a33468cb..849cc0220 100644
--- a/rust/src/runtime/payload_discovery.rs
+++ b/rust/src/runtime/payload_discovery.rs
@@ -104,6 +104,7 @@ pub(super) fn limits() -> Result<Value> {
     let attributes = super::search_attribute_upserts::Limits::from_env()?;
     Ok(json!({
         "max_payload_bytes": super::payload_limits::max_payload_bytes(),
+        "max_workflow_task_completion_bytes": super::payload_limits::max_completion_bytes(),
         "max_memo_bytes": super::schedules::max_memo_bytes(),
         "max_search_attributes": super::search_attributes::configured_limit()?,
         "max_search_attribute_key_length": attributes.key_length,
diff --git a/rust/src/runtime/payload_limits.rs b/rust/src/runtime/payload_limits.rs
index c99bb01aa..62313c724 100644
--- a/rust/src/runtime/payload_limits.rs
+++ b/rust/src/runtime/payload_limits.rs
@@ -4,13 +4,15 @@
 //! refused with PHP's JSON error, in the worker or control-plane envelope.
 //! The external payload upload keeps its own transport rules.
 use axum::{
-    body::{Body, to_bytes},
+    body::{Body, BodyDataStream},
     extract::Request,
     http::{Method, StatusCode, header},
     middleware::Next,
     response::Response,
 };
 use serde_json::{Value, json};
+use futures_util::{StreamExt, stream};
+use std::time::Duration;
 
 /// How many payloads' worth of bytes one completion that carries workflow
 /// commands may send (Server #1136, .github#228). Each command's arguments
@@ -18,6 +20,10 @@ use serde_json::{Value, json};
 /// request as a whole, so a fan-out whose commands are each within the limits
 /// is not refused as one oversized body.
 const COMPLETION_PAYLOADS: usize = 16;
+// Refusals must let ordinary clients finish their current upload before the
+// HTTP/1 connection closes. Bound both discarded traffic and slow senders.
+const REFUSAL_DRAIN_BYTES: usize = 64 * 1024 * 1024;
+const REFUSAL_DRAIN_TIMEOUT: Duration = Duration::from_secs(2);
 
 /// A request that completes a workflow task with commands.
 fn carries_commands(method: &Method, path: &str) -> bool {
@@ -37,13 +43,39 @@ pub(super) fn max_payload_bytes() -> usize {
         .unwrap_or(2 * 1024 * 1024)
 }
 
+pub(super) fn max_completion_bytes() -> usize {
+    max_payload_bytes().saturating_mul(COMPLETION_PAYLOADS)
+}
+
+fn oversized_response(response: Response, mut incoming: BodyDataStream) -> Response {
+    let (mut parts, body) = response.into_parts();
+    parts.headers.insert(header::CONNECTION, axum::http::HeaderValue::from_static("close"));
+    let response_stream = stream::once(async move {
+        let drain = async {
+            let mut consumed = 0_usize;
+            while let Some(Ok(chunk)) = incoming.next().await {
+                consumed = consumed.saturating_add(chunk.len());
+                if consumed >= REFUSAL_DRAIN_BYTES {
+                    break;
+                }
+            }
+        };
+        let _ = tokio::time::timeout(REFUSAL_DRAIN_TIMEOUT, drain).await;
+        // Keep the request body alive until the discard finishes. Dropping
+        // it with bytes in flight makes Hyper close its read side and can
+        // turn the typed 413 into a client write/read transport failure.
+        body.into_data_stream()
+    }).flatten();
+    Response::from_parts(parts, Body::from_stream(response_stream))
+}
+
 pub(super) async fn enforce(request: Request, next: Next) -> Response {
     let path = request.uri().path().to_owned();
     if request.method() == Method::POST && path == "/api/external-payloads/v1" {
         return next.run(request).await;
     }
     let completion = carries_commands(request.method(), &path);
-    let max = if completion { max_payload_bytes().saturating_mul(COMPLETION_PAYLOADS) } else { max_payload_bytes() };
+    let max = if completion { max_completion_bytes() } else { max_payload_bytes() };
     let worker = path == "/api/worker" || path.starts_with("/api/worker/");
     let declared = request
         .headers()
@@ -51,6 +83,7 @@ pub(super) async fn enforce(request: Request, next: Next) -> Response {
         .and_then(|value| value.to_str().ok())
         .and_then(|value| value.trim().parse::<u64>().ok());
     let (parts, body) = request.into_parts();
+    let mut incoming = body.into_data_stream();
     let refuse = |status: StatusCode, body: Value| {
         let parts = parts.clone();
         async move { super::http::admission_error(Request::from_parts(parts, Body::empty()), worker, status, body).await }
@@ -67,13 +100,18 @@ pub(super) async fn enforce(request: Request, next: Next) -> Response {
         body
     };
     if declared.is_some_and(|declared| declared > max as u64) {
-        return refuse(StatusCode::PAYLOAD_TOO_LARGE, too_large()).await;
+        return oversized_response(refuse(StatusCode::PAYLOAD_TOO_LARGE, too_large()).await, incoming);
     }
-    let Ok(bytes) = to_bytes(body, max.saturating_add(1)).await else {
-        return refuse(StatusCode::PAYLOAD_TOO_LARGE, too_large()).await;
-    };
-    if bytes.len() > max {
-        return refuse(StatusCode::PAYLOAD_TOO_LARGE, too_large()).await;
+    let mut bytes = Vec::new();
+    while let Some(chunk) = incoming.next().await {
+        let Ok(chunk) = chunk else {
+            return oversized_response(refuse(StatusCode::PAYLOAD_TOO_LARGE, too_large()).await, incoming);
+        };
+        if chunk.len() > max.saturating_sub(bytes.len()) {
+            drop(bytes);
+            return oversized_response(refuse(StatusCode::PAYLOAD_TOO_LARGE, too_large()).await, incoming);
+        }
+        bytes.extend_from_slice(&chunk);
     }
     let has_body = !bytes.is_empty() || declared.is_some_and(|declared| declared > 0);
     if matches!(parts.method, Method::POST | Method::PUT | Method::PATCH | Method::DELETE) && has_body {
diff --git a/rust/tests/execution/payload_discovery.rs b/rust/tests/execution/payload_discovery.rs
index 88419d968..0ef77fe1f 100644
--- a/rust/tests/execution/payload_discovery.rs
+++ b/rust/tests/execution/payload_discovery.rs
@@ -62,6 +62,7 @@ async fn cluster_info_discovers_runtime_payload_storage_like_php() {
         assert_eq!(storage["transport"], info["worker_protocol"]["server_capabilities"]["runtime_external_payload_transport"],
             "{name} namespace transport is the worker capability");
         assert_eq!(info["limits"]["max_payload_bytes"], 2_097_152);
+        assert_eq!(info["limits"]["max_workflow_task_completion_bytes"], 33_554_432);
         assert_eq!(info["limits"]["max_memo_bytes"], 262_144);
         assert_eq!(info["limits"]["max_search_attributes"], 100);
         assert_eq!(info["default_namespace"], "default");
diff --git a/rust/tests/execution/payload_limits.rs b/rust/tests/execution/payload_limits.rs
index 511f4d590..266ba25cc 100644
--- a/rust/tests/execution/payload_limits.rs
+++ b/rust/tests/execution/payload_limits.rs
@@ -1,4 +1,8 @@
 use super::*;
+use futures_util::{StreamExt, stream};
+use std::sync::{Arc, atomic::{AtomicUsize, Ordering}};
+use std::time::Duration;
+use tokio::io::{AsyncReadExt, AsyncWriteExt};
 
 /// Sends `body` raw, with an optional content type and without credentials
 /// unless `authorized`, and returns the status, headers and JSON answer.
@@ -65,6 +69,89 @@ async fn bodies_are_checked_before_authentication_like_php() {
     assert_eq!(status, StatusCode::UNAUTHORIZED, "{answer}");
 }
 
+// A router-only response test cannot see a TCP reset while the caller is
+// still writing. Both common request framings must deliver the actual 413.
+#[tokio::test]
+async fn oversized_completions_finish_the_upload_and_deliver_the_typed_tcp_refusal() {
+    let database = TestDatabase::new().await;
+    let runtime = database.open().await.unwrap();
+    let app = router(runtime.clone());
+    let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
+    let address = listener.local_addr().unwrap();
+    let (stop, stopped) = tokio::sync::oneshot::channel();
+    let server = tokio::spawn(async move {
+        axum::serve(listener, app).with_graceful_shutdown(async { let _ = stopped.await; }).await.unwrap();
+    });
+    let before = table_counts(&database).await;
+    for chunked in [false, true] {
+        let wire = tokio::time::timeout(Duration::from_secs(10), async {
+            let mut client = tokio::net::TcpStream::connect(address).await.unwrap();
+            let length = 36 * 1024 * 1024;
+            let framing = if chunked { "Transfer-Encoding: chunked\r\n".to_owned() }
+                else { format!("Content-Length: {length}\r\n") };
+            client.write_all(format!("POST /api/worker/workflow-tasks/fixture-task/complete HTTP/1.1\r\nHost: localhost\r\nContent-Type: application/json\r\n{framing}Connection: close\r\n\r\n").as_bytes()).await.unwrap();
+            let chunk = vec![b'x'; 64 * 1024];
+            for _ in 0..length/chunk.len() {
+                if chunked { client.write_all(format!("{:X}\r\n", chunk.len()).as_bytes()).await.unwrap(); }
+                client.write_all(&chunk).await.expect("the refusal must not reset an ordinary in-flight upload");
+                if chunked { client.write_all(b"\r\n").await.unwrap(); }
+            }
+            if chunked { client.write_all(b"0\r\n\r\n").await.unwrap(); }
+            let mut response = Vec::new();
+            client.take(1024 * 1024).read_to_end(&mut response).await.expect("the typed refusal must be readable");
+            String::from_utf8(response).unwrap()
+        }).await.expect("bounded upload/refusal round trip");
+        assert!(wire.starts_with("HTTP/1.1 413"), "{}", wire.chars().take(300).collect::<String>());
+        for field in ["\"reason\":\"payload_too_large\"", "\"limit_kind\":\"workflow_task_completion_bytes\"", "\"limit\":33554432"] {
+            assert!(wire.contains(field), "missing {field}");
+        }
+    }
+    assert_eq!(table_counts(&database).await, before, "size refusals admit no durable work");
+    stop.send(()).unwrap();
+    tokio::time::timeout(Duration::from_secs(3), server).await.unwrap().unwrap();
+    runtime.close().await;
+    database.remove().await;
+}
+
+#[tokio::test]
+async fn oversized_body_discard_has_a_byte_budget_and_does_not_collect_the_upload() {
+    let database = TestDatabase::new().await;
+    let runtime = database.open().await.unwrap();
+    let app = router(runtime.clone());
+    let observed = Arc::new(AtomicUsize::new(0));
+    let count = observed.clone();
+    let chunk = axum::body::Bytes::from(vec![b'x'; 1024 * 1024]);
+    let incoming = stream::repeat_with(move || {
+        count.fetch_add(1, Ordering::SeqCst);
+        Ok::<_, std::io::Error>(chunk.clone())
+    }).take(200);
+    let response = app.oneshot(Request::builder().method("POST").uri("/api/workflows")
+        .header("content-length", 200 * 1024 * 1024).body(Body::from_stream(incoming)).unwrap()).await.unwrap();
+    assert_eq!(response.status(), StatusCode::PAYLOAD_TOO_LARGE);
+    let answer = to_bytes(response.into_body(), 1024 * 1024).await.unwrap();
+    assert_eq!(serde_json::from_slice::<Value>(&answer).unwrap()["reason"], "payload_too_large");
+    let consumed = observed.load(Ordering::SeqCst);
+    assert!(consumed > 0 && consumed <= 65, "unbounded refused body consumption: {consumed} MiB");
+    runtime.close().await;
+    database.remove().await;
+}
+
+#[tokio::test]
+async fn a_stalled_oversized_body_cannot_hold_the_refusal_open_indefinitely() {
+    let database = TestDatabase::new().await;
+    let runtime = database.open().await.unwrap();
+    let app = router(runtime.clone());
+    let incoming = stream::pending::<Result<axum::body::Bytes, std::io::Error>>();
+    let response = app.oneshot(Request::builder().method("POST").uri("/api/workflows")
+        .header("content-length", 200 * 1024 * 1024).body(Body::from_stream(incoming)).unwrap()).await.unwrap();
+    assert_eq!(response.status(), StatusCode::PAYLOAD_TOO_LARGE);
+    let bytes = tokio::time::timeout(Duration::from_secs(5), to_bytes(response.into_body(), 1024 * 1024))
+        .await.expect("refusal discard deadline").unwrap();
+    assert_eq!(serde_json::from_slice::<Value>(&bytes).unwrap()["reason"], "payload_too_large");
+    runtime.close().await;
+    database.remove().await;
+}
+
 // Server #1136 (.github#228): a fan-out whose commands are each within the
 // payload limit completes even when the whole completion is over it; a
 // completion over the aggregate limit is refused with a typed 413.
@@ -90,3 +177,44 @@ async fn a_large_fan_out_completion_is_not_refused_as_one_body() {
     assert_eq!(answer["limit_kind"], "workflow_task_completion_bytes", "{answer}");
     assert_eq!(answer["limit"], 32 * 1024 * 1024, "{answer}");
 }
+
+// Portable wire-size vectors are shared with SDK planners. A valid JSON body
+// at the exact cap reaches authentication; the next byte is a typed refusal.
+#[tokio::test]
+async fn completion_wire_budget_matches_the_shared_discovery_vectors() {
+    let vectors: Value = serde_json::from_str(include_str!("../../../resources/platform-protocol-specs/workflow-task-completion-limits-v1-vectors.json")).unwrap();
+    let database = TestDatabase::new().await;
+    let runtime = database.open().await.unwrap();
+    let app = router(runtime.clone());
+    let (_, info) = request(&app, "GET", "/api/cluster/info", Value::Null).await;
+    for (field, expected) in vectors["default_limits"].as_object().unwrap() {
+        assert_eq!(&info["limits"][field], expected, "discovery {field}");
+    }
+    let before = table_counts(&database).await;
+    for case in vectors["wire_cases"].as_array().unwrap() {
+        let routes: Vec<&str> = if case["route_kind"] == "completion" {
+            vectors["completion_routes"].as_array().unwrap().iter().map(|route| route.as_str().unwrap()).collect()
+        } else { vec![vectors["ordinary_route"].as_str().unwrap()] };
+        let length = case["wire_bytes"].as_u64().unwrap() as usize;
+        for route in routes {
+            let mut body = vec![b'x'; length];
+            body[0] = b'"';
+            body[length - 1] = b'"';
+            let (status, _, response) = raw(&app, route, Some("application/json"), body, false).await;
+            if case["size_admitted"] == true {
+                assert_eq!(status, StatusCode::UNAUTHORIZED, "{} {route}: {response}", case["id"]);
+            } else {
+                assert_eq!(status, StatusCode::PAYLOAD_TOO_LARGE, "{} {route}: {response}", case["id"]);
+                assert_eq!(response["reason"], vectors["oversized_response"]["reason"]);
+                if case["route_kind"] == "completion" {
+                    for field in ["limit_kind", "limit"] {
+                        assert_eq!(response[field], vectors["oversized_response"][field], "{field}");
+                    }
+                }
+            }
+        }
+    }
+    assert_eq!(table_counts(&database).await, before);
+    runtime.close().await;
+    database.remove().await;
+}
diff --git a/scripts/conformance/server-parity/cluster-info-contract.mjs b/scripts/conformance/server-parity/cluster-info-contract.mjs
index 3d93f2d1f..a8a271b01 100644
--- a/scripts/conformance/server-parity/cluster-info-contract.mjs
+++ b/scripts/conformance/server-parity/cluster-info-contract.mjs
@@ -1,4 +1,8 @@
 import assert from 'node:assert/strict';
+import {readFileSync} from 'node:fs';
+import {nativeDivergence} from './native-divergences.mjs';
+
+const completionVectors = JSON.parse(readFileSync(new URL('../../../resources/platform-protocol-specs/workflow-task-completion-limits-v1-vectors.json', import.meta.url), 'utf8'));
 
 // The transport manifest a server advertises. Published PHP advertises the
 // fixture's manifest exactly. Native (`implementation: "rust"`) advertises it
@@ -45,7 +49,10 @@ export function checkClusterInfo(fixture, observation, workflowId) {
   assert.deepStrictEqual(Object.keys(state.discovery), Object.keys(definition.namespaces), 'every namespace read');
   for (const [key, discovery] of Object.entries(state.discovery)) {
     assert.equal(discovery.status, 200, `${key} cluster info`);
-    assert.deepStrictEqual(discovery.limits, definition.limits, `${key} request limits`);
+    const expectedLimits = nativeDivergence(implementation, 'completion-byte-limit-discovery')
+      ? {...definition.limits, max_workflow_task_completion_bytes: completionVectors.default_limits.max_workflow_task_completion_bytes}
+      : definition.limits;
+    assert.deepStrictEqual(discovery.limits, expectedLimits, `${key} request limits`);
     assert.equal(discovery.default_namespace, definition.default_namespace, `${key} default namespace`);
     assert.equal(discovery.server_id_is_text, true, `${key} server identity`);
     assert.deepStrictEqual(discovery.transport_capability, transport, `${key} transport capability`);
diff --git a/scripts/conformance/server-parity/native-divergences.json b/scripts/conformance/server-parity/native-divergences.json
index 1cebd2b73..1ec40c11b 100644
--- a/scripts/conformance/server-parity/native-divergences.json
+++ b/scripts/conformance/server-parity/native-divergences.json
@@ -1,6 +1,15 @@
 {
   "schema": "durable-workflow.server-parity-native-divergences/v1",
   "divergences": [
+    {
+      "id": "completion-byte-limit-discovery",
+      "issue": "https://github.com/durable-workflow/server/issues/1136",
+      "since": "native #1136 source correction",
+      "applies_to": ["cluster-info"],
+      "published_php": "The limits object does not advertise a separate workflow task completion request-body budget.",
+      "native": "limits.max_workflow_task_completion_bytes advertises the actual enforced wire-request cap for workflow and activity continuation completions, independently of per-value limits.",
+      "harness": "The native cluster-info comparator requires the exact field and shared-vector default value. PHP and embedded observations and frozen fixtures remain unchanged. Actual HTTP boundary and TCP refusal tests consume the shared completion-limit vectors."
+    },
     {
       "id": "request-id-reuse-conflict",
       "issue": "https://github.com/durable-workflow/server/issues/934",
diff --git a/tests/Unit/ServerParityClusterInfoContractTest.mjs b/tests/Unit/ServerParityClusterInfoContractTest.mjs
index c68e27aee..60378c995 100644
--- a/tests/Unit/ServerParityClusterInfoContractTest.mjs
+++ b/tests/Unit/ServerParityClusterInfoContractTest.mjs
@@ -15,7 +15,7 @@ const observation = (implementation = null) => {
   const policies = Object.fromEntries(Object.entries(definition.policies).map(([key, policy]) =>
     [key, {status: 200, response: {name: names[key], external_payload_storage: structuredClone(policy)}}]));
   const discovery = Object.fromEntries(Object.keys(definition.namespaces).map(key => [key, {
-    status: 200, implementation, limits: structuredClone(definition.limits),
+    status: 200, implementation, limits: {...structuredClone(definition.limits), ...(implementation === 'rust' ? {max_workflow_task_completion_bytes: 33554432} : {})},
     default_namespace: 'default', server_id_is_text: true, transport_capability: structuredClone(transport),
     namespace: {name: names[key], ...definition.rows[key],
       external_payload_storage: {...structuredClone(definition.storage_envelope), ...definition.storage[key], transport: structuredClone(transport)}},
@@ -71,6 +71,9 @@ for (const [label, implementation, mutate] of [
   ['a missing namespace reported as existing', 'rust', state => { read(state, 'missing').namespace.exists = true; }],
   ['another namespace read', 'rust', state => { read(state, 'available').namespace.name = 'default'; }],
   ['a namespace transport unlike the capability', 'rust', state => { read(state, 'available').namespace.external_payload_storage.transport.limits.max_payload_bytes = 1; }],
+  ['native missing its completion budget', 'rust', state => { delete read(state, 'available').limits.max_workflow_task_completion_bytes; }],
+  ['native advertising the per-value cap as its completion cap', 'rust', state => { read(state, 'available').limits.max_workflow_task_completion_bytes = 2097152; }],
+  ['PHP acquiring an unreviewed completion budget', null, state => { read(state, 'available').limits.max_workflow_task_completion_bytes = 33554432; }],
   ['another request limit', null, state => { read(state, 'missing').limits.max_payload_bytes = 1048576; }],
   ['an exposed provider', 'rust', state => { read(state, 'available').namespace.external_payload_storage.provider_details_exposed = true; }],
   ['a lost policy', 'rust', state => { state.policies.disabled.status = 422; }],

From cbb186fa17d62cd2eccacf4e1da1f90c73c82dd2 Mon Sep 17 00:00:00 2001
From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com>
Date: Fri, 13 Nov 2026 10:31:42 +0000
Subject: [PATCH 2/4] Record agreed SDK completion limit negotiation vectors

---
 docs/server-reference.md                      |  4 +
 ...low-task-completion-limits-v1-vectors.json | 94 ++++++++++++++++++-
 2 files changed, 97 insertions(+), 1 deletion(-)

diff --git a/docs/server-reference.md b/docs/server-reference.md
index f67804688..244354fa0 100644
--- a/docs/server-reference.md
+++ b/docs/server-reference.md
@@ -653,6 +653,10 @@ and commands. The default is 33,554,432 bytes. `limits.max_payload_bytes` remain
 the per-value and ordinary API request limit; a decoded argument sum does not
 measure completion wire size. SDKs use the explicit optional field for those
 two completion routes and preserve their conservative fallback when it is absent.
+A positive integer smaller than the ordinary cap is still authoritative for
+completion. Explicit malformed, null, zero, negative or non-integer values
+refuse completion planning before upload or submission; ordinary routes ignore
+this unused field.
 
 A completion over that cap returns HTTP 413 with `reason: payload_too_large`,
 `limit_kind: workflow_task_completion_bytes` and the actual byte `limit`, before
diff --git a/resources/platform-protocol-specs/workflow-task-completion-limits-v1-vectors.json b/resources/platform-protocol-specs/workflow-task-completion-limits-v1-vectors.json
index 406dc0817..5f766ab6b 100644
--- a/resources/platform-protocol-specs/workflow-task-completion-limits-v1-vectors.json
+++ b/resources/platform-protocol-specs/workflow-task-completion-limits-v1-vectors.json
@@ -49,5 +49,97 @@
     "limit_kind": "workflow_task_completion_bytes",
     "limit": 33554432
   },
-  "scope": "Size admission alone. Authentication, lease fences, per-value Avro limits and command validation still apply; no durable work is admitted by a size refusal."
+  "scope": "Size admission alone. Authentication, lease fences, per-value Avro limits and command validation still apply; no durable work is admitted by a size refusal.",
+  "negotiation_cases": [
+    {
+      "id": "absent",
+      "limits": {
+        "max_payload_bytes": 2097152
+      },
+      "expected": "ordinary_fallback"
+    },
+    {
+      "id": "positive",
+      "limits": {
+        "max_payload_bytes": 2097152,
+        "max_workflow_task_completion_bytes": 33554432
+      },
+      "expected": "completion_cap",
+      "completion_cap": 33554432
+    },
+    {
+      "id": "positive-smaller",
+      "limits": {
+        "max_payload_bytes": 2097152,
+        "max_workflow_task_completion_bytes": 1024
+      },
+      "expected": "completion_cap",
+      "completion_cap": 1024
+    },
+    {
+      "id": "null",
+      "limits": {
+        "max_payload_bytes": 2097152,
+        "max_workflow_task_completion_bytes": null
+      },
+      "expected": "refuse_completion_planning_before_upload_or_submission"
+    },
+    {
+      "id": "zero",
+      "limits": {
+        "max_payload_bytes": 2097152,
+        "max_workflow_task_completion_bytes": 0
+      },
+      "expected": "refuse_completion_planning_before_upload_or_submission"
+    },
+    {
+      "id": "negative",
+      "limits": {
+        "max_payload_bytes": 2097152,
+        "max_workflow_task_completion_bytes": -1
+      },
+      "expected": "refuse_completion_planning_before_upload_or_submission"
+    },
+    {
+      "id": "float",
+      "limits": {
+        "max_payload_bytes": 2097152,
+        "max_workflow_task_completion_bytes": 32.5
+      },
+      "expected": "refuse_completion_planning_before_upload_or_submission"
+    },
+    {
+      "id": "string",
+      "limits": {
+        "max_payload_bytes": 2097152,
+        "max_workflow_task_completion_bytes": "33554432"
+      },
+      "expected": "refuse_completion_planning_before_upload_or_submission"
+    },
+    {
+      "id": "bool",
+      "limits": {
+        "max_payload_bytes": 2097152,
+        "max_workflow_task_completion_bytes": true
+      },
+      "expected": "refuse_completion_planning_before_upload_or_submission"
+    },
+    {
+      "id": "object",
+      "limits": {
+        "max_payload_bytes": 2097152,
+        "max_workflow_task_completion_bytes": {}
+      },
+      "expected": "refuse_completion_planning_before_upload_or_submission"
+    },
+    {
+      "id": "array",
+      "limits": {
+        "max_payload_bytes": 2097152,
+        "max_workflow_task_completion_bytes": []
+      },
+      "expected": "refuse_completion_planning_before_upload_or_submission"
+    }
+  ],
+  "negotiation_scope": "Only workflow-task completion and activity-continuation completion consume this field. Ordinary routes ignore it. Explicit invalid values refuse planning without upload/submission; absence preserves the existing conservative fallback."
 }

From e69df7c37d675681ca301b9eedc2398bc1a3ac18 Mon Sep 17 00:00:00 2001
From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com>
Date: Fri, 13 Nov 2026 10:33:14 +0000
Subject: [PATCH 3/4] Select completion limit vectors for shared native
 qualification

---
 .github/workflows/server-parity.yml      | 1 +
 tests/Unit/ServerParityAggregateTest.mjs | 1 +
 2 files changed, 2 insertions(+)

diff --git a/.github/workflows/server-parity.yml b/.github/workflows/server-parity.yml
index 61099c1da..5e7a75b71 100644
--- a/.github/workflows/server-parity.yml
+++ b/.github/workflows/server-parity.yml
@@ -27,6 +27,7 @@ on:
       - tests/Unit/NativeDebugArtifactTest.mjs
       - config/dw-contract.php
       - resources/platform-protocol-specs/carried-poll-v1-vectors.json
+      - resources/platform-protocol-specs/workflow-task-completion-limits-v1-vectors.json
       - tests/Fixtures/ActivityMetadataProjection/**
       - tests/Fixtures/CodecRegression/**
       - tests/Fixtures/CommandIntent/**
diff --git a/tests/Unit/ServerParityAggregateTest.mjs b/tests/Unit/ServerParityAggregateTest.mjs
index b707ae2db..5fe64a8b1 100644
--- a/tests/Unit/ServerParityAggregateTest.mjs
+++ b/tests/Unit/ServerParityAggregateTest.mjs
@@ -51,6 +51,7 @@ test('pull requests run without a path filter and push keeps the one path list',
   // Inputs the fixture jobs read outside the server-parity directories.
   for (const input of [
     'tests/Fixtures/ActivityResultInspection/v1.json',
+    'resources/platform-protocol-specs/workflow-task-completion-limits-v1-vectors.json',
     'tests/Fixtures/ServerParityReaders/sdk-signals.mjs',
     'tests/Conformance/ServerParityDeadlinePollTest.mjs',
     'benchmarks/capacity/v1/bindings/php/capacity_adapter.php',

From 919b0f080ecf70c7d9b85da51fd32604fa9b2b2f Mon Sep 17 00:00:00 2001
From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com>
Date: Fri, 13 Nov 2026 10:37:37 +0000
Subject: [PATCH 4/4] Require positive advertised completion caps in the schema

---
 .../cluster-info-envelope.schema.json         |  2 +-
 .../WorkerProtocolOpenApiContractTest.php     | 26 +++++++++++++++++++
 2 files changed, 27 insertions(+), 1 deletion(-)

diff --git a/resources/platform-protocol-specs/cluster-info-envelope.schema.json b/resources/platform-protocol-specs/cluster-info-envelope.schema.json
index 0022f45de..341246e01 100644
--- a/resources/platform-protocol-specs/cluster-info-envelope.schema.json
+++ b/resources/platform-protocol-specs/cluster-info-envelope.schema.json
@@ -174,7 +174,7 @@
       "additionalProperties": true,
       "properties": {
         "max_payload_bytes": {"type": "integer", "minimum": 1},
-        "max_workflow_task_completion_bytes": {"type": "integer", "minimum": 0, "description": "Optional actual encoded JSON HTTP request-body cap for workflow-task completion and activity continuation completion. Per-value max_payload_bytes still applies. Missing means the producer has not advertised a separate completion budget; consumers must preserve their conservative fallback and must not infer a multiplier from server version."},
+        "max_workflow_task_completion_bytes": {"type": "integer", "minimum": 1, "description": "Optional actual encoded JSON HTTP request-body cap for workflow-task completion and activity continuation completion. Per-value max_payload_bytes still applies. Missing means the producer has not advertised a separate completion budget; consumers must preserve their conservative fallback and must not infer a multiplier from server version."},
         "max_memo_bytes": {"type": "integer", "minimum": 1},
         "max_search_attributes": {"type": "integer", "minimum": 0},
         "max_search_attribute_key_length": {"type": "integer", "minimum": 1},
diff --git a/tests/Unit/WorkerProtocolOpenApiContractTest.php b/tests/Unit/WorkerProtocolOpenApiContractTest.php
index 9267bdd17..0f0e2fe69 100644
--- a/tests/Unit/WorkerProtocolOpenApiContractTest.php
+++ b/tests/Unit/WorkerProtocolOpenApiContractTest.php
@@ -26,6 +26,32 @@ protected function setUp(): void
         $this->spec = $spec;
     }
 
+    public function test_completion_budget_schema_consumes_shared_positive_and_invalid_vectors(): void
+    {
+        $root = dirname(__DIR__, 2).'/resources/platform-protocol-specs/';
+        $document = json_decode(file_get_contents($root.'cluster-info-envelope.schema.json'), true, flags: JSON_THROW_ON_ERROR);
+        $vectors = json_decode(file_get_contents($root.'workflow-task-completion-limits-v1-vectors.json'), flags: JSON_THROW_ON_ERROR);
+        $schema = OpenApiSchema::fromFile($root.'cluster-info-envelope.schema.json');
+        $field = 'max_workflow_task_completion_bytes';
+        $this->assertNotContains($field, $document['properties']['limits']['required'], 'Older producers may omit the field.');
+        foreach ($vectors->negotiation_cases as $case) {
+            if (! property_exists($case->limits, $field)) {
+                continue;
+            }
+            $validate = fn () => $schema->assertReferenceMatches('#/properties/limits/properties/'.$field, $case->limits->{$field});
+            if ($case->expected === 'completion_cap') {
+                $validate();
+                continue;
+            }
+            try {
+                $validate();
+                $this->fail('Invalid completion budget accepted: '.$case->id);
+            } catch (AssertionFailedError $error) {
+                $this->assertStringNotContainsString('Invalid completion budget accepted:', $error->getMessage());
+            }
+        }
+    }
+
     public function test_workflow_claims_publish_an_additive_observation_clock(): void
     {
         $schema = OpenApiSchema::fromFile(dirname(__DIR__, 2).'/resources/platform-protocol-specs/worker-protocol-api.openapi.yaml');