diff --git a/runtime/binding-mqtt-kafka/src/main/java/io/aklivity/zilla/runtime/binding/mqtt/kafka/internal/stream/MqttKafkaSessionFactory.java b/runtime/binding-mqtt-kafka/src/main/java/io/aklivity/zilla/runtime/binding/mqtt/kafka/internal/stream/MqttKafkaSessionFactory.java
index c66902598e3..b4cac2be6e0 100644
--- a/runtime/binding-mqtt-kafka/src/main/java/io/aklivity/zilla/runtime/binding/mqtt/kafka/internal/stream/MqttKafkaSessionFactory.java
+++ b/runtime/binding-mqtt-kafka/src/main/java/io/aklivity/zilla/runtime/binding/mqtt/kafka/internal/stream/MqttKafkaSessionFactory.java
@@ -3019,13 +3019,15 @@ protected final void doKafkaData(
             int limit,
             Flyweight extension)
         {
+            if (kafka != null)
+            {
+                doData(kafka, originId, routedId, initialId, initialSeq, initialAck, initialMax,
+                    traceId, authorization, budgetId, flags, reserved, buffer, offset, limit, extension);
 
-            doData(kafka, originId, routedId, initialId, initialSeq, initialAck, initialMax,
-                traceId, authorization, budgetId, flags, reserved, buffer, offset, limit, extension);
-
-            initialSeq += reserved;
+                initialSeq += reserved;
 
-            assert initialSeq - padding <= initialAck + initialMax;
+                assert initialSeq - padding <= initialAck + initialMax;
+            }
         }
 
         protected final void doKafkaData(
@@ -3037,12 +3039,15 @@ protected final void doKafkaData(
             OctetsFW payload,
             Flyweight extension)
         {
-            doData(kafka, originId, routedId, initialId, initialSeq, initialAck, initialMax,
-                traceId, authorization, budgetId, flags, reserved, payload, extension);
+            if (kafka != null)
+            {
+                doData(kafka, originId, routedId, initialId, initialSeq, initialAck, initialMax,
+                    traceId, authorization, budgetId, flags, reserved, payload, extension);
 
-            initialSeq += reserved;
+                initialSeq += reserved;
 
-            assert initialSeq <= initialAck + initialMax;
+                assert initialSeq <= initialAck + initialMax;
+            }
         }
 
         protected void doKafkaData(
@@ -3055,17 +3060,20 @@ protected void doKafkaData(
             Flyweight payload,
             Flyweight extension)
         {
-            final DirectBufferEx buffer = payload.buffer();
-            final int offset = payload.offset();
-            final int limit = payload.limit();
-            final int length = limit - offset;
+            if (kafka != null)
+            {
+                final DirectBufferEx buffer = payload.buffer();
+                final int offset = payload.offset();
+                final int limit = payload.limit();
+                final int length = limit - offset;
 
-            doData(kafka, originId, routedId, initialId, initialSeq, initialAck, initialMax,
-                traceId, authorization, budgetId, flags, reserved, buffer, offset, length, extension);
+                doData(kafka, originId, routedId, initialId, initialSeq, initialAck, initialMax,
+                    traceId, authorization, budgetId, flags, reserved, buffer, offset, length, extension);
 
-            initialSeq += reserved;
+                initialSeq += reserved;
 
-            assert initialSeq - padding <= initialAck + initialMax;
+                assert initialSeq - padding <= initialAck + initialMax;
+            }
         }
 
         private void doKafkaFlush(
diff --git a/runtime/binding-mqtt-kafka/src/test/java/io/aklivity/zilla/runtime/binding/mqtt/kafka/internal/stream/MqttKafkaSessionProxyIT.java b/runtime/binding-mqtt-kafka/src/test/java/io/aklivity/zilla/runtime/binding/mqtt/kafka/internal/stream/MqttKafkaSessionProxyIT.java
index ffbe9521ea9..939f667c22b 100644
--- a/runtime/binding-mqtt-kafka/src/test/java/io/aklivity/zilla/runtime/binding/mqtt/kafka/internal/stream/MqttKafkaSessionProxyIT.java
+++ b/runtime/binding-mqtt-kafka/src/test/java/io/aklivity/zilla/runtime/binding/mqtt/kafka/internal/stream/MqttKafkaSessionProxyIT.java
@@ -275,6 +275,17 @@ public void shouldNotSendWillMessageOnWillIdMismatch() throws Exception
         k3po.finish();
     }
 
+    @Test
+    @Configuration("proxy.yaml")
+    @Configure(name = WILL_AVAILABLE_NAME, value = "false")
+    @Specification({
+        "${mqtt}/session.will.message.qos2.abort.before.session.established/client",
+        "${kafka}/session.will.message.qos2.abort.before.session.established/server"})
+    public void shouldSkipWillSignalOnQos2AbortBeforeSessionEstablished() throws Exception
+    {
+        k3po.finish();
+    }
+
     @Test
     @Configuration("proxy.yaml")
     @Configure(name = PUBLISH_MAX_QOS_NAME, value = "1")
diff --git a/runtime/binding-mqtt/src/main/java/io/aklivity/zilla/runtime/binding/mqtt/internal/stream/MqttServerFactory.java b/runtime/binding-mqtt/src/main/java/io/aklivity/zilla/runtime/binding/mqtt/internal/stream/MqttServerFactory.java
index 4fd2e98ef6d..ec74d626047 100644
--- a/runtime/binding-mqtt/src/main/java/io/aklivity/zilla/runtime/binding/mqtt/internal/stream/MqttServerFactory.java
+++ b/runtime/binding-mqtt/src/main/java/io/aklivity/zilla/runtime/binding/mqtt/internal/stream/MqttServerFactory.java
@@ -3561,6 +3561,7 @@ private int onDecodeConnectWillPayload(
             final int willQos = decodeWillQos(connectFlags);
             final boolean willFlagSet = isSetWillFlag(connectFlags);
 
+            decode:
             if (willFlagSet && MqttState.initialOpened(session.state))
             {
                 int publishedWillSize = 0;
@@ -3588,6 +3589,12 @@ private int onDecodeConnectWillPayload(
 
                     final MqttWillMessageFW will = willMessageBuilder.build();
                     final int headerSize = willMessageBuilder.sizeof();
+
+                    if (!session.hasSessionWindow(headerSize))
+                    {
+                        break decode;
+                    }
+
                     int payloadSize = Math.min(limit - offset, session.initialBudget() - headerSize);
 
                     final OctetsFW payload = payloadRO.wrap(buffer, offset, offset + payloadSize);
diff --git a/runtime/binding-mqtt/src/test/java/io/aklivity/zilla/runtime/binding/mqtt/internal/stream/server/v5/SessionIT.java b/runtime/binding-mqtt/src/test/java/io/aklivity/zilla/runtime/binding/mqtt/internal/stream/server/v5/SessionIT.java
index 0e822225360..7c82a3e1984 100644
--- a/runtime/binding-mqtt/src/test/java/io/aklivity/zilla/runtime/binding/mqtt/internal/stream/server/v5/SessionIT.java
+++ b/runtime/binding-mqtt/src/test/java/io/aklivity/zilla/runtime/binding/mqtt/internal/stream/server/v5/SessionIT.java
@@ -187,6 +187,16 @@ public void shouldSendWillMessage10k() throws Exception
         k3po.finish();
     }
 
+    @Test
+    @Configuration("server.yaml")
+    @Specification({
+        "${net}/session.will.message.header.exceeds.window/client",
+        "${app}/session.will.message.header.exceeds.window/server"})
+    public void shouldDeferWillMessageWhenHeaderExceedsSessionWindow() throws Exception
+    {
+        k3po.finish();
+    }
+
     @Test
     @Configuration("server.yaml")
     @Specification({
diff --git a/specs/binding-mqtt-kafka.spec/src/main/scripts/io/aklivity/zilla/specs/binding/mqtt/kafka/streams/kafka/session.will.message.qos2.abort.before.session.established/client.rpt b/specs/binding-mqtt-kafka.spec/src/main/scripts/io/aklivity/zilla/specs/binding/mqtt/kafka/streams/kafka/session.will.message.qos2.abort.before.session.established/client.rpt
new file mode 100644
index 00000000000..2198cb5b8cc
--- /dev/null
+++ b/specs/binding-mqtt-kafka.spec/src/main/scripts/io/aklivity/zilla/specs/binding/mqtt/kafka/streams/kafka/session.will.message.qos2.abort.before.session.established/client.rpt
@@ -0,0 +1,49 @@
+#
+# Copyright 2021-2026 Aklivity Inc
+#
+# Licensed under the Aklivity Community License (the "License"); you may not use
+# this file except in compliance with the License.  You may obtain a copy of the
+# License at
+#
+#   https://www.aklivity.io/aklivity-community-license/
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+# WARRANTIES OF ANY KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations under the License.
+#
+
+connect
+        "zilla://streams/kafka0"
+         option zilla:window 8192
+         option zilla:transmission "duplex"
+
+write zilla:begin.ext ${kafka:beginEx()
+                               .typeId(zilla:id("kafka"))
+                               .meta()
+                                   .topic("mqtt-messages")
+                                   .build()
+                               .build()}
+
+connected
+
+write await SESSION_META_STREAMS_OPENED
+write abort
+
+
+connect
+        "zilla://streams/kafka0"
+         option zilla:window 8192
+         option zilla:transmission "duplex"
+
+write zilla:begin.ext ${kafka:beginEx()
+                               .typeId(zilla:id("kafka"))
+                               .meta()
+                                   .topic("mqtt-retained")
+                                   .build()
+                               .build()}
+
+connected
+
+write await SESSION_META_STREAMS_OPENED
+write abort
diff --git a/specs/binding-mqtt-kafka.spec/src/main/scripts/io/aklivity/zilla/specs/binding/mqtt/kafka/streams/kafka/session.will.message.qos2.abort.before.session.established/server.rpt b/specs/binding-mqtt-kafka.spec/src/main/scripts/io/aklivity/zilla/specs/binding/mqtt/kafka/streams/kafka/session.will.message.qos2.abort.before.session.established/server.rpt
new file mode 100644
index 00000000000..82182278285
--- /dev/null
+++ b/specs/binding-mqtt-kafka.spec/src/main/scripts/io/aklivity/zilla/specs/binding/mqtt/kafka/streams/kafka/session.will.message.qos2.abort.before.session.established/server.rpt
@@ -0,0 +1,51 @@
+#
+# Copyright 2021-2026 Aklivity Inc
+#
+# Licensed under the Aklivity Community License (the "License"); you may not use
+# this file except in compliance with the License.  You may obtain a copy of the
+# License at
+#
+#   https://www.aklivity.io/aklivity-community-license/
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+# WARRANTIES OF ANY KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations under the License.
+#
+
+accept "zilla://streams/kafka0"
+    option zilla:window 8192
+    option zilla:transmission "duplex"
+
+
+accepted
+
+read zilla:begin.ext ${kafka:beginEx()
+                               .typeId(zilla:id("kafka"))
+                               .meta()
+                                   .topic("mqtt-messages")
+                                   .build()
+                               .build()}
+
+connected
+
+read aborted
+
+
+accepted
+
+read zilla:begin.ext ${kafka:beginEx()
+                               .typeId(zilla:id("kafka"))
+                               .meta()
+                                   .topic("mqtt-retained")
+                                   .build()
+                               .build()}
+
+connected
+
+# partition metadata is still outstanding, no session state stream exists yet
+write notify SESSION_META_STREAMS_OPENED
+
+# no will signal and no expiry signal are produced,
+# the session was never established
+read aborted
diff --git a/specs/binding-mqtt-kafka.spec/src/main/scripts/io/aklivity/zilla/specs/binding/mqtt/kafka/streams/mqtt/session.will.message.qos2.abort.before.session.established/client.rpt b/specs/binding-mqtt-kafka.spec/src/main/scripts/io/aklivity/zilla/specs/binding/mqtt/kafka/streams/mqtt/session.will.message.qos2.abort.before.session.established/client.rpt
new file mode 100644
index 00000000000..f6db14f64c9
--- /dev/null
+++ b/specs/binding-mqtt-kafka.spec/src/main/scripts/io/aklivity/zilla/specs/binding/mqtt/kafka/streams/mqtt/session.will.message.qos2.abort.before.session.established/client.rpt
@@ -0,0 +1,38 @@
+#
+# Copyright 2021-2026 Aklivity Inc
+#
+# Licensed under the Aklivity Community License (the "License"); you may not use
+# this file except in compliance with the License.  You may obtain a copy of the
+# License at
+#
+#   https://www.aklivity.io/aklivity-community-license/
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+# WARRANTIES OF ANY KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations under the License.
+#
+
+# session stays half-duplex, the session reply stream is only opened once the
+# qos2 session state stream has been established
+connect "zilla://streams/mqtt0"
+         option zilla:window 8192
+         option zilla:transmission "half-duplex"
+
+write zilla:begin.ext ${mqtt:beginEx()
+                           .typeId(zilla:id("mqtt"))
+                           .session()
+                              .flags("WILL", "CLEAN_START")
+                              .expiry(1)
+                              .publishQosMax(2)
+                              .capabilities("REDIRECT")
+                              .clientId("client-1")
+                              .build()
+                           .build()}
+
+connected
+
+# abort while the qos2 partition metadata is still being negotiated,
+# no session state stream has been established yet
+write await SESSION_META_STREAMS_OPENED
+write abort
diff --git a/specs/binding-mqtt-kafka.spec/src/main/scripts/io/aklivity/zilla/specs/binding/mqtt/kafka/streams/mqtt/session.will.message.qos2.abort.before.session.established/server.rpt b/specs/binding-mqtt-kafka.spec/src/main/scripts/io/aklivity/zilla/specs/binding/mqtt/kafka/streams/mqtt/session.will.message.qos2.abort.before.session.established/server.rpt
new file mode 100644
index 00000000000..8c0ca1a99de
--- /dev/null
+++ b/specs/binding-mqtt-kafka.spec/src/main/scripts/io/aklivity/zilla/specs/binding/mqtt/kafka/streams/mqtt/session.will.message.qos2.abort.before.session.established/server.rpt
@@ -0,0 +1,35 @@
+#
+# Copyright 2021-2026 Aklivity Inc
+#
+# Licensed under the Aklivity Community License (the "License"); you may not use
+# this file except in compliance with the License.  You may obtain a copy of the
+# License at
+#
+#   https://www.aklivity.io/aklivity-community-license/
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+# WARRANTIES OF ANY KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations under the License.
+#
+
+accept "zilla://streams/mqtt0"
+        option zilla:window 8192
+        option zilla:transmission "half-duplex"
+
+accepted
+
+read zilla:begin.ext ${mqtt:matchBeginEx()
+                             .typeId(zilla:id("mqtt"))
+                             .session()
+                                .flags("WILL", "CLEAN_START")
+                                .expiry(1)
+                                .publishQosMax(2)
+                                .capabilities("REDIRECT")
+                                .clientId("client-1")
+                                .build()
+                             .build()}
+
+connected
+
+read aborted
diff --git a/specs/binding-mqtt-kafka.spec/src/test/java/io/aklivity/zilla/specs/binding/mqtt/kafka/streams/KafkaIT.java b/specs/binding-mqtt-kafka.spec/src/test/java/io/aklivity/zilla/specs/binding/mqtt/kafka/streams/KafkaIT.java
index ddacf07e744..f914e7c09c1 100644
--- a/specs/binding-mqtt-kafka.spec/src/test/java/io/aklivity/zilla/specs/binding/mqtt/kafka/streams/KafkaIT.java
+++ b/specs/binding-mqtt-kafka.spec/src/test/java/io/aklivity/zilla/specs/binding/mqtt/kafka/streams/KafkaIT.java
@@ -747,6 +747,15 @@ public void shouldGenerateLifeTimeIdOnCleanStart() throws Exception
         k3po.finish();
     }
 
+    @Test
+    @Specification({
+        "${kafka}/session.will.message.qos2.abort.before.session.established/client",
+        "${kafka}/session.will.message.qos2.abort.before.session.established/server"})
+    public void shouldSkipWillSignalOnQos2AbortBeforeSessionEstablished() throws Exception
+    {
+        k3po.finish();
+    }
+
     @Test
     @Specification({
         "${kafka}/session.will.message.will.id.mismatch.skip.delivery/client",
diff --git a/specs/binding-mqtt-kafka.spec/src/test/java/io/aklivity/zilla/specs/binding/mqtt/kafka/streams/MqttIT.java b/specs/binding-mqtt-kafka.spec/src/test/java/io/aklivity/zilla/specs/binding/mqtt/kafka/streams/MqttIT.java
index ae4ee9d6e7f..932cf7fd3ca 100644
--- a/specs/binding-mqtt-kafka.spec/src/test/java/io/aklivity/zilla/specs/binding/mqtt/kafka/streams/MqttIT.java
+++ b/specs/binding-mqtt-kafka.spec/src/test/java/io/aklivity/zilla/specs/binding/mqtt/kafka/streams/MqttIT.java
@@ -630,6 +630,17 @@ public void shouldSaveWillMessage() throws Exception
         k3po.finish();
     }
 
+    @Test
+    @Specification({
+        "${mqtt}/session.will.message.qos2.abort.before.session.established/client",
+        "${mqtt}/session.will.message.qos2.abort.before.session.established/server"})
+    public void shouldSkipWillSignalOnQos2AbortBeforeSessionEstablished() throws Exception
+    {
+        k3po.start();
+        k3po.notifyBarrier("SESSION_META_STREAMS_OPENED");
+        k3po.finish();
+    }
+
     @Test
     @Specification({
         "${mqtt}/publish.qos1/client",
diff --git a/specs/binding-mqtt.spec/src/main/scripts/io/aklivity/zilla/specs/binding/mqtt/streams/application/session.will.message.header.exceeds.window/client.rpt b/specs/binding-mqtt.spec/src/main/scripts/io/aklivity/zilla/specs/binding/mqtt/streams/application/session.will.message.header.exceeds.window/client.rpt
new file mode 100644
index 00000000000..0d0462aa662
--- /dev/null
+++ b/specs/binding-mqtt.spec/src/main/scripts/io/aklivity/zilla/specs/binding/mqtt/streams/application/session.will.message.header.exceeds.window/client.rpt
@@ -0,0 +1,45 @@
+#
+# Copyright 2021-2026 Aklivity Inc.
+#
+# Aklivity licenses this file to you under the Apache License,
+# version 2.0 (the "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at:
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+# License for the specific language governing permissions and limitations
+# under the License.
+#
+
+connect "zilla://streams/app0"
+         option zilla:window 32
+         option zilla:transmission "duplex"
+
+write zilla:begin.ext ${mqtt:beginEx()
+                           .typeId(zilla:id("mqtt"))
+                           .session()
+                              .flags("WILL", "CLEAN_START")
+                              .clientId("one")
+                              .build()
+                           .build()}
+
+read zilla:begin.ext ${mqtt:matchBeginEx()
+                              .typeId(zilla:id("mqtt"))
+                              .session()
+                                .flags("WILL", "CLEAN_START")
+                                .subscribeQosMax(2)
+                                .publishQosMax(2)
+                                .packetSizeMax(33792)
+                                .capabilities("RETAIN", "WILDCARD", "SUBSCRIPTION_IDS", "SHARED_SUBSCRIPTIONS")
+                                .clientId("one")
+                                .build()
+                              .build()}
+
+connected
+
+read zilla:data.empty
+
+write abort
diff --git a/specs/binding-mqtt.spec/src/main/scripts/io/aklivity/zilla/specs/binding/mqtt/streams/application/session.will.message.header.exceeds.window/server.rpt b/specs/binding-mqtt.spec/src/main/scripts/io/aklivity/zilla/specs/binding/mqtt/streams/application/session.will.message.header.exceeds.window/server.rpt
new file mode 100644
index 00000000000..a32f839eec1
--- /dev/null
+++ b/specs/binding-mqtt.spec/src/main/scripts/io/aklivity/zilla/specs/binding/mqtt/streams/application/session.will.message.header.exceeds.window/server.rpt
@@ -0,0 +1,52 @@
+#
+# Copyright 2021-2026 Aklivity Inc.
+#
+# Aklivity licenses this file to you under the Apache License,
+# version 2.0 (the "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at:
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+# License for the specific language governing permissions and limitations
+# under the License.
+#
+
+# window is smaller than the encoded will message header (44 bytes for topic
+# "wills/one"), so the will message must remain deferred instead of being
+# written with a negative payload size
+
+accept "zilla://streams/app0"
+        option zilla:window 32
+        option zilla:transmission "duplex"
+
+accepted
+
+read zilla:begin.ext ${mqtt:matchBeginEx()
+                             .typeId(zilla:id("mqtt"))
+                             .session()
+                                .flags("WILL", "CLEAN_START")
+                                .clientId("one")
+                                .build()
+                             .build()}
+
+write zilla:begin.ext ${mqtt:beginEx()
+                              .typeId(zilla:id("mqtt"))
+                              .session()
+                                .flags("WILL", "CLEAN_START")
+                                .subscribeQosMax(2)
+                                .publishQosMax(2)
+                                .packetSizeMax(33792)
+                                .capabilities("RETAIN", "WILDCARD", "SUBSCRIPTION_IDS", "SHARED_SUBSCRIPTIONS")
+                                .clientId("one")
+                                .build()
+                              .build()}
+
+connected
+
+write zilla:data.empty
+write flush
+
+read aborted
diff --git a/specs/binding-mqtt.spec/src/main/scripts/io/aklivity/zilla/specs/binding/mqtt/streams/network/v5/session.will.message.header.exceeds.window/client.rpt b/specs/binding-mqtt.spec/src/main/scripts/io/aklivity/zilla/specs/binding/mqtt/streams/network/v5/session.will.message.header.exceeds.window/client.rpt
new file mode 100644
index 00000000000..090baa0ac9d
--- /dev/null
+++ b/specs/binding-mqtt.spec/src/main/scripts/io/aklivity/zilla/specs/binding/mqtt/streams/network/v5/session.will.message.header.exceeds.window/client.rpt
@@ -0,0 +1,41 @@
+#
+# Copyright 2021-2026 Aklivity Inc.
+#
+# Aklivity licenses this file to you under the Apache License,
+# version 2.0 (the "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at:
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+# License for the specific language governing permissions and limitations
+# under the License.
+#
+
+connect "zilla://streams/net0"
+  option zilla:window 8192
+  option zilla:transmission "duplex"
+  option zilla:byteorder "network"
+
+connected
+
+write [0x10 0x2f]                                      # CONNECT
+      [0x00 0x04] "MQTT"                               # protocol name
+      [0x05]                                           # protocol version
+      [0x06]                                           # flags = will flag, clean start
+      [0x00 0x0a]                                      # keep alive = 10s
+      [0x05]                                           # properties
+      [0x27] 33792                                     # maximum packet size = 33792
+      [0x00 0x03] "one"                                # client id
+      [0x00]                                           # will properties
+      [0x00 0x09] "wills/one"                          # will topic
+      [0x00 0x0c] "will payload"                       # will payload
+
+read  [0x20 0x03]                                      # CONNACK
+      [0x00]                                           # flags = none
+      [0x00]                                           # reason code
+      [0x00]                                           # properties = none
+
+write abort
diff --git a/specs/binding-mqtt.spec/src/main/scripts/io/aklivity/zilla/specs/binding/mqtt/streams/network/v5/session.will.message.header.exceeds.window/server.rpt b/specs/binding-mqtt.spec/src/main/scripts/io/aklivity/zilla/specs/binding/mqtt/streams/network/v5/session.will.message.header.exceeds.window/server.rpt
new file mode 100644
index 00000000000..acb1cbb3062
--- /dev/null
+++ b/specs/binding-mqtt.spec/src/main/scripts/io/aklivity/zilla/specs/binding/mqtt/streams/network/v5/session.will.message.header.exceeds.window/server.rpt
@@ -0,0 +1,42 @@
+#
+# Copyright 2021-2026 Aklivity Inc.
+#
+# Aklivity licenses this file to you under the Apache License,
+# version 2.0 (the "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at:
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+# License for the specific language governing permissions and limitations
+# under the License.
+#
+
+accept "zilla://streams/net0"
+  option zilla:window 8192
+  option zilla:transmission "duplex"
+  option zilla:byteorder "network"
+
+accepted
+connected
+
+read  [0x10 0x2f]                                      # CONNECT
+      [0x00 0x04] "MQTT"                               # protocol name
+      [0x05]                                           # protocol version
+      [0x06]                                           # flags = will flag, clean start
+      [0x00 0x0a]                                      # keep alive = 10s
+      [0x05]                                           # properties
+      [0x27] 33792                                     # maximum packet size = 33792
+      [0x00 0x03] "one"                                # client id
+      [0x00]                                           # will properties
+      [0x00 0x09] "wills/one"                          # will topic
+      [0x00 0x0c] "will payload"                       # will payload
+
+write [0x20 0x03]                                      # CONNACK
+      [0x00]                                           # flags = none
+      [0x00]                                           # reason code
+      [0x00]                                           # properties = none
+
+read aborted
diff --git a/specs/binding-mqtt.spec/src/test/java/io/aklivity/zilla/specs/binding/mqtt/streams/application/SessionIT.java b/specs/binding-mqtt.spec/src/test/java/io/aklivity/zilla/specs/binding/mqtt/streams/application/SessionIT.java
index 383ee934b7a..0e35d62c2f6 100644
--- a/specs/binding-mqtt.spec/src/test/java/io/aklivity/zilla/specs/binding/mqtt/streams/application/SessionIT.java
+++ b/specs/binding-mqtt.spec/src/test/java/io/aklivity/zilla/specs/binding/mqtt/streams/application/SessionIT.java
@@ -119,6 +119,15 @@ public void shouldSendWillMessage10k() throws Exception
         k3po.finish();
     }
 
+    @Test
+    @Specification({
+        "${app}/session.will.message.header.exceeds.window/client",
+        "${app}/session.will.message.header.exceeds.window/server"})
+    public void shouldDeferWillMessageWhenHeaderExceedsSessionWindow() throws Exception
+    {
+        k3po.finish();
+    }
+
     @Test
     @Specification({
         "${app}/session.subscribe/client",
diff --git a/specs/binding-mqtt.spec/src/test/java/io/aklivity/zilla/specs/binding/mqtt/streams/network/v5/SessionIT.java b/specs/binding-mqtt.spec/src/test/java/io/aklivity/zilla/specs/binding/mqtt/streams/network/v5/SessionIT.java
index 00368cb4a7b..10acb33ecb1 100644
--- a/specs/binding-mqtt.spec/src/test/java/io/aklivity/zilla/specs/binding/mqtt/streams/network/v5/SessionIT.java
+++ b/specs/binding-mqtt.spec/src/test/java/io/aklivity/zilla/specs/binding/mqtt/streams/network/v5/SessionIT.java
@@ -129,6 +129,15 @@ public void shouldConnectWithWillMessageThenNormalDisconnect() throws Exception
         k3po.finish();
     }
 
+    @Test
+    @Specification({
+        "${net}/session.will.message.header.exceeds.window/client",
+        "${net}/session.will.message.header.exceeds.window/server"})
+    public void shouldDeferWillMessageWhenHeaderExceedsSessionWindow() throws Exception
+    {
+        k3po.finish();
+    }
+
     // [MQTT-3.1.2-15]
     @Test
     @Specification({